diff --git a/router/router.py b/router/router.py index 0919828..80a4c21 100644 --- a/router/router.py +++ b/router/router.py @@ -236,13 +236,39 @@ def select_best_gpu(candidates, reason, agent=""): +# Phase 1: Circuit Breaker for GPU Hosts +def is_circuit_tripped(model): + """Check if a GPU host is currently blacklisted.""" + if not get_redis(): + return False + return r.exists("circuit:" + model + ":open") + +def trip_circuit(model, duration=30): + """Blacklist a GPU host for a specified duration.""" + if not get_redis(): + return + key = "circuit:" + model + ":open" + r.set(key, 1, ex=duration) + r.incr("circuit:" + model + ":count") + log.warning("CIRCUIT_TRIPPED: %s blacklisted for %ds", model, duration) + +def half_open_probe(model): + """Check if a GPU host can be un-blacklisted.""" + if not get_redis(): + return True + key = "circuit:" + model + ":open" + if not r.exists(key): + return True # no circuit + return False # still open + def route(rd, tier, agent=""): msgs = rd.get("messages",[]); t = estimate_tokens(msgs) sys = any(m.get("role")=="system" for m in msgs) turns = len([m for m in msgs if m.get("role") in ("user","assistant")]) hints = rd.get("routing_hints",{}) allowed = TIER_MODELS.get(tier, ["gemma-4-12b"]) - avail = [m for m in available_models() if m in allowed] + # Phase 1: Filter out models with tripped circuit breakers + avail = [m for m in available_models() if m in allowed and not is_circuit_tripped(m)] if not avail: return {"model": allowed[0], "reason": "all_saturated", "saturated": True} if all(is_gpu_busy(m) for m in avail): return {"model": avail[0], "reason": "all_saturated", "saturated": True} @@ -463,7 +489,10 @@ def chat(): lat = int((time.time()-start)*1000) gpu_decr(model) - if resp.status_code != 200: return jsonify({"error":"GPU error "+str(resp.status_code)}), 502 + if resp.status_code != 200: + if resp.status_code == 502 or resp.status_code == 504: + trip_circuit(model, 30) + return jsonify({"error":"GPU error "+str(resp.status_code)}), 502 if is_stream: # Buffer SSE chunks, handle split lines for large responses chunks = [] @@ -537,7 +566,8 @@ def chat(): return resp except requests.Timeout: gpu_decr(model) - log.error("TIMEOUT: %s -> %s", agent, model) + trip_circuit(model, 30) + log.error("TIMEOUT: %s -> %s (Circuit tripped)", agent, model) return jsonify({"error":"timeout"}), 504 except Exception as e: gpu_decr(model) @@ -734,6 +764,22 @@ def metrics_timeseries(): data["models"][model] = counts return jsonify(data) +@app.route("/metrics/circuit-breaker") +def metrics_circuit_breaker(): + """Expose circuit breaker status per model. Phase 1.""" + result = {} + if r: + for model in GPU_URLS: + key = "circuit:" + model + ":open" + duration = r.ttl(key) + trip_count = int(r.get("circuit:" + model + ":count") or 0) + result[model] = { + "tripped": r.exists(key), + "remaining_ttl": duration, + "trip_count": trip_count + } + return jsonify(result) + @app.route("/stream") def stream(): def ev():