Router: lazy Redis reconnect (survives Redis restarts/reboots)
This commit is contained in:
+30
-12
@@ -50,7 +50,7 @@ RATE_LIMIT_RPM = {
|
|||||||
|
|
||||||
def check_rate_limit(api_key, tier):
|
def check_rate_limit(api_key, tier):
|
||||||
"""Token bucket rate limiter using Redis. Returns (allowed, retry_after_or_remaining, reset_seconds)."""
|
"""Token bucket rate limiter using Redis. Returns (allowed, retry_after_or_remaining, reset_seconds)."""
|
||||||
if not r:
|
if not get_redis():
|
||||||
return True, 999, 60
|
return True, 999, 60
|
||||||
limit = RATE_LIMIT_RPM.get(tier, 30)
|
limit = RATE_LIMIT_RPM.get(tier, 30)
|
||||||
key = f"ratelimit:{api_key}"
|
key = f"ratelimit:{api_key}"
|
||||||
@@ -70,7 +70,25 @@ def check_rate_limit(api_key, tier):
|
|||||||
|
|
||||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [ROUTER] %(levelname)s %(message)s")
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s [ROUTER] %(levelname)s %(message)s")
|
||||||
log = logging.getLogger("router")
|
log = logging.getLogger("router")
|
||||||
try: r = redis.from_url(REDIS_URL, decode_responses=True); r.ping()
|
# Redis connection — initialized lazily, retries on first use
|
||||||
|
def get_redis():
|
||||||
|
global r
|
||||||
|
if r is not None:
|
||||||
|
try:
|
||||||
|
r.ping()
|
||||||
|
return r
|
||||||
|
except Exception:
|
||||||
|
r = None
|
||||||
|
try:
|
||||||
|
r = redis.from_url(REDIS_URL, decode_responses=True)
|
||||||
|
r.ping()
|
||||||
|
return r
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
|
||||||
|
r = None
|
||||||
|
try: get_redis()
|
||||||
|
except Exception: pass
|
||||||
except Exception: r = None
|
except Exception: r = None
|
||||||
|
|
||||||
|
|
||||||
@@ -78,7 +96,7 @@ def counter_audit_loop():
|
|||||||
"""Every 30s, check GPU slots and reset counters if all slots idle."""
|
"""Every 30s, check GPU slots and reset counters if all slots idle."""
|
||||||
while True:
|
while True:
|
||||||
time.sleep(30)
|
time.sleep(30)
|
||||||
if not r: continue
|
if not get_redis(): continue
|
||||||
for model, url in GPU_URLS.items():
|
for model, url in GPU_URLS.items():
|
||||||
try:
|
try:
|
||||||
resp = requests.get(url.replace("/v1","") + "/slots",
|
resp = requests.get(url.replace("/v1","") + "/slots",
|
||||||
@@ -89,7 +107,7 @@ def counter_audit_loop():
|
|||||||
if all_idle:
|
if all_idle:
|
||||||
current = int(r.get("active:" + model) or 0)
|
current = int(r.get("active:" + model) or 0)
|
||||||
if current > 0:
|
if current > 0:
|
||||||
r.set("active:" + model, 0)
|
rd.set("active:" + model, 0)
|
||||||
log.info("AUDIT: Reset stuck counter for %s (was %d)", model, current)
|
log.info("AUDIT: Reset stuck counter for %s (was %d)", model, current)
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
@@ -106,14 +124,14 @@ def gpu_active_count(model):
|
|||||||
return 0
|
return 0
|
||||||
|
|
||||||
def gpu_incr(model):
|
def gpu_incr(model):
|
||||||
if r: r.incr("active:" + model)
|
if get_redis(): get_redis().incr("active:" + model)
|
||||||
|
|
||||||
def gpu_decr(model):
|
def gpu_decr(model):
|
||||||
if r:
|
rd = get_redis()
|
||||||
v = r.decr("active:" + model)
|
if rd:
|
||||||
|
v = rd.decr("active:" + model)
|
||||||
if v and int(v) < 0:
|
if v and int(v) < 0:
|
||||||
r.set("active:" + model, 0) # never go negative
|
rd.set("active:" + model, 0) # never go negative
|
||||||
|
|
||||||
def check_gpu_health(model, sidecar_timeout=5, gpu_timeout=3):
|
def check_gpu_health(model, sidecar_timeout=5, gpu_timeout=3):
|
||||||
url = GPU_SIDECARS.get(model)
|
url = GPU_SIDECARS.get(model)
|
||||||
if not url: return {"status": "unknown"}
|
if not url: return {"status": "unknown"}
|
||||||
@@ -144,7 +162,7 @@ def estimate_tokens(msgs):
|
|||||||
|
|
||||||
def store_perf_record(model, agent, tier, reason, queue_ms, inference_ms, prompt_tokens, completion_tokens, stream):
|
def store_perf_record(model, agent, tier, reason, queue_ms, inference_ms, prompt_tokens, completion_tokens, stream):
|
||||||
"""Store detailed performance record in Redis for analytics."""
|
"""Store detailed performance record in Redis for analytics."""
|
||||||
if not r: return
|
if not get_redis(): return
|
||||||
try:
|
try:
|
||||||
total_ms = queue_ms + inference_ms
|
total_ms = queue_ms + inference_ms
|
||||||
tps = completion_tokens / (inference_ms / 1000) if inference_ms > 0 and completion_tokens > 0 else 0
|
tps = completion_tokens / (inference_ms / 1000) if inference_ms > 0 and completion_tokens > 0 else 0
|
||||||
@@ -518,7 +536,7 @@ def chat():
|
|||||||
@app.route("/metrics/performance")
|
@app.route("/metrics/performance")
|
||||||
def performance():
|
def performance():
|
||||||
"""Per-request performance analytics with percentiles per model/reason/agent."""
|
"""Per-request performance analytics with percentiles per model/reason/agent."""
|
||||||
if not r: return jsonify({"error": "Redis unavailable"}), 503
|
if not get_redis(): return jsonify({"error": "Redis unavailable"}), 503
|
||||||
try:
|
try:
|
||||||
window_hours = int(request.args.get("window", "24").replace("h",""))
|
window_hours = int(request.args.get("window", "24").replace("h",""))
|
||||||
model_filter = request.args.get("model", "all")
|
model_filter = request.args.get("model", "all")
|
||||||
@@ -635,7 +653,7 @@ def performance():
|
|||||||
@app.route("/metrics/scatter")
|
@app.route("/metrics/scatter")
|
||||||
def scatter():
|
def scatter():
|
||||||
"""Return individual data points for scatter plots (prompt_tokens vs latency)."""
|
"""Return individual data points for scatter plots (prompt_tokens vs latency)."""
|
||||||
if not r: return jsonify({"error": "Redis unavailable"}), 503
|
if not get_redis(): return jsonify({"error": "Redis unavailable"}), 503
|
||||||
try:
|
try:
|
||||||
window_hours = int(request.args.get("window", "24").replace("h",""))
|
window_hours = int(request.args.get("window", "24").replace("h",""))
|
||||||
model_filter = request.args.get("model", "all")
|
model_filter = request.args.get("model", "all")
|
||||||
|
|||||||
Reference in New Issue
Block a user