feat(router): Phase 1 - Circuit Breaker for GPU hosts
- Added is_circuit_tripped(), trip_circuit(), half_open_probe() functions - Filters out models with tripped circuits in route() function - Trips circuit on 502/504 errors in chat() function - Added /metrics/circuit-breaker endpoint for visibility (Abba suggestion) - Prevents hung GPU cascades (Node #480 scenario) Signed-off-by: Mumuni <mumuni@sysloggh.com>
This commit is contained in:
+49
-3
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user