abiba-bot/inference-harness
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: 2-layer architecture foundation + agent migration

Browse Source 
Router v2:
- Atomic GPU slot booking (Redis Lua — closes TOCTOU race)
- Circuit breaker with failure threshold (3 failures/120s → 60s cooldown)
- GPU health scoring with configurable weights (VRAM 40% + temp 30% + load 30%)
- X-Usage-Tokens header for LiteLLM spend tracking
- /health/unified endpoint (aggregates all layers)
- Strict explicit model passthrough (no silent fallback)
- syslog-auto → auto routing fix

Infrastructure:
- Postgres 16 for LiteLLM state
- LiteLLM production config (router:9000, fallback chains, guardrails)
- Dual-path NGINX: /v1/→router, /litellm/v1/→LiteLLM, port 4000 UI
- DNS split-horizon (auth.sysloggh.net → 192.168.68.11)
- 6 LiteLLM virtual keys for agent cutover

Deployed to CT 116, all 6 containers healthy.
This commit is contained in:
Abiba
2026-06-17 22:40:33 +00:00
parent 776343f2ab
commit fd3c2a575a
4 changed files with 498 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
router/router.py
+167 -24
View File
@@ -14,6 +14,37 @@ redis.call('SET', key, max_val, 'EX', 86400)
return max_val
"""
# Phase 4: Atomic GPU slot booking (closes TOCTOU race between check and incr)
SLOT_BOOK_LUA = """
local key = KEYS[1]
local max_c = tonumber(ARGV[1])
local current = tonumber(redis.call('GET', key) or '0')
if current < max_c then
redis.call('INCR', key)
return 1
else
return 0
end
"""
SLOT_RELEASE_LUA = """
local key = KEYS[1]
local current = tonumber(redis.call('GET', key) or '0')
if current > 0 then
redis.call('DECR', key)
end
current = tonumber(redis.call('GET', key) or '0')
if current < 0 then
redis.call('SET', key, '0')
end
return redis.call('GET', key)
"""
# Phase 3b: Configurable health scoring weights (env-overridable)
HEALTH_WEIGHT_VRAM = float(os.environ.get("HEALTH_WEIGHT_VRAM", "0.40"))
HEALTH_WEIGHT_TEMP = float(os.environ.get("HEALTH_WEIGHT_TEMP", "0.30"))
HEALTH_WEIGHT_LOAD = float(os.environ.get("HEALTH_WEIGHT_LOAD", "0.30"))
HEALTH_TEMP_BASELINE = int(os.environ.get("HEALTH_TEMP_BASELINE", "30"))
REDIS_URL = os.environ.get("REDIS_URL", "redis://redis:6379")
GPU_MOE_URL = os.environ.get("GPU_MOE_URL", "http://192.168.68.15:8080/v1")
@@ -39,7 +70,7 @@ GPU_LABELS = {
}
GPU_MAX_CONCURRENT = {
"qwen3.6-35B-A3B": 2, # 2 slots (cross-agent spread prevents overheating)
"qwen3.6-35B-A3B": 1, # 2 slots (cross-agent spread prevents overheating)
"qwen3.6-27B-code": 2, # 2 slots (128K context frees VRAM)
"gemma-4-12b": 2, # 2 slots (7.1GB VRAM)
}
@@ -152,6 +183,36 @@ def gpu_decr(model):
v = rd.decr("active:" + model)
if v and int(v) < 0:
get_redis().set("active:" + model, 0) # never go negative
# Phase 4: Atomic GPU slot booking (Lua-based, closes TOCTOU race)
def gpu_book_slot(model):
"""Atomically book a GPU slot. Returns True if acquired, False if full."""
rd = get_redis()
if not rd:
return True # No Redis — allow everything (degraded mode)
try:
max_c = GPU_MAX_CONCURRENT.get(model, 1)
result = rd.eval(SLOT_BOOK_LUA, 1, "active:" + model, max_c)
return result == 1
except Exception:
# Lua not loaded — fall back to non-atomic
current = int(rd.get("active:" + model) or 0)
if current < GPU_MAX_CONCURRENT.get(model, 1):
rd.incr("active:" + model)
return True
return False
def gpu_release_slot(model):
"""Atomically release a GPU slot. Never goes negative."""
rd = get_redis()
if not rd:
return
try:
rd.eval(SLOT_RELEASE_LUA, 1, "active:" + model)
except Exception:
v = rd.decr("active:" + model)
if v and int(v) < 0:
rd.set("active:" + model, 0)
def check_gpu_health(model, sidecar_timeout=5, gpu_timeout=3):
url = GPU_SIDECARS.get(model)
if not url: return {"status": "unknown"}
@@ -222,18 +283,22 @@ def is_gpu_busy(model):
# Phase 3: Dynamic GPU Weighting (Health Score)
def gpu_health_score(model):
"""Score a GPU based on VRAM, temperature, and load. Lower = better."""
"""Score a GPU based on VRAM, temperature, power, and load. Lower = better.
Weights configurable via HEALTH_WEIGHT_VRAM/TEMP/LOAD env vars."""
h = check_gpu_health(model, sidecar_timeout=1.5, gpu_timeout=1)
if h.get("status") == "down":
return 999 # never pick down GPUs
if is_circuit_tripped(model):
return 998 # circuit open — skip but distinguishable from down
vram_pct = h.get("vram_pct") or 50
temp_c = h.get("temp_c") or 50
power_w = h.get("power_w") or 100
active = gpu_active_count(model)
max_c = GPU_MAX_CONCURRENT.get(model, 1)
load_pct = (active / max_c) * 100 if max_c > 0 else 0
# Score: lower = better (more headroom, cooler, less loaded)
score = (vram_pct or 0) * 0.4 + max((temp_c or 50) - 30, 0) * 0.3 + load_pct * 0.3
return score
temp_penalty = max(0, (temp_c or 50) - HEALTH_TEMP_BASELINE)
score = (vram_pct or 0) * HEALTH_WEIGHT_VRAM + temp_penalty * 0.5 * HEALTH_WEIGHT_TEMP + load_pct * HEALTH_WEIGHT_LOAD
return round(score, 1)
def select_best_gpu(candidates, reason, agent=""):
"""Pick best GPU, spreading agents across GPUs to prevent hotspots."""
@@ -278,20 +343,40 @@ def select_best_gpu(candidates, reason, agent=""):
# Phase 1: Circuit Breaker for GPU Hosts (Approved by Abiba)
CIRCUIT_FAIL_THRESHOLD = int(os.environ.get("CIRCUIT_FAIL_THRESHOLD", "3"))
CIRCUIT_FAIL_WINDOW = int(os.environ.get("CIRCUIT_FAIL_WINDOW", "120"))
CIRCUIT_COOLDOWN = int(os.environ.get("CIRCUIT_COOLDOWN", "60"))
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."""
def trip_circuit(model, duration=None):
"""Blacklist a GPU host for specified duration (default CIRCUIT_COOLDOWN).
Only trips after CIRCUIT_FAIL_THRESHOLD failures within CIRCUIT_FAIL_WINDOW."""
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)
return False
if duration is None:
duration = CIRCUIT_COOLDOWN
now = time.time()
fail_key = "circuit:" + model + ":failures"
pipe = r.pipeline()
pipe.lpush(fail_key, str(now))
pipe.ltrim(fail_key, 0, CIRCUIT_FAIL_THRESHOLD - 1)
pipe.lrange(fail_key, 0, -1)
results = pipe.execute()
failures = [float(f) for f in (results[-1] if results else [])]
recent = [f for f in failures if now - f <= CIRCUIT_FAIL_WINDOW]
if len(recent) >= CIRCUIT_FAIL_THRESHOLD:
key = "circuit:" + model + ":open"
r.set(key, 1, ex=duration)
r.incr("circuit:" + model + ":count")
log.warning("CIRCUIT_TRIPPED: %s — %d failures in %ds, cooldown %ds",
model, len(recent), CIRCUIT_FAIL_WINDOW, duration)
return True
return False
def half_open_probe(model):
"""Check if a GPU host can be un-blacklisted."""
@@ -329,13 +414,17 @@ def route(rd, tier, agent=""):
return {"model": allowed[0], "reason": "vision_unavailable"}
req = rd.get("model","auto")
# Map syslog-auto to auto for content-based routing
if req == "syslog-auto":
req = "auto"
if req != "auto":
# STRICT MODE: no silent fallback — LiteLLM handles failover chains.
# Returns saturated if explicit GPU is busy (keeps per-model metrics accurate).
target = req if req in avail else avail[0]
if is_gpu_busy(target) and req in allowed:
alts = [m for m in avail if m != target and m in allowed]
if alts:
alt = select_best_gpu(alts, "explicit", agent)
if alt: return alt
if req not in avail:
return {"model": req, "reason": "explicit_unavailable", "saturated": True}
if is_gpu_busy(target):
return {"model": target, "reason": "explicit_saturated", "saturated": True}
return {"model": target, "reason": "explicit"}
if hints:
@@ -503,14 +592,26 @@ def chat():
log.info("QUEUED: %s waited %.0fms before slot opened", agent, queue_ms)
model, reason, url = d["model"], d["reason"], GPU_URLS[d["model"]]
# Phase 4: Atomic slot booking (replaces non-atomic gpu_incr)
if not gpu_book_slot(model):
d = route(rd, tier, agent)
if d.get("saturated"):
resp = jsonify({"error": "All GPUs saturated", "retry_after_s": 3})
resp.headers["Retry-After"] = "3"
return resp, 503
model, reason = d["model"], d["reason"]
if not gpu_book_slot(model):
resp = jsonify({"error": "GPU slot race — retry", "retry_after_s": 1})
resp.headers["Retry-After"] = "1"
return resp, 503
url = GPU_URLS[model]
# Stash rate limit values for response headers
_rl_remaining = rl_val
_rl_limit = RATE_LIMIT_RPM.get(tier, 30)
_rl_reset = reset_sec
is_stream = rd.get("stream", False)
gpu_incr(model)
log.info("ROUTE: %s -> %s (%s) stream=%s active=%d/%d", agent, model, reason, is_stream, gpu_active_count(model), GPU_MAX_CONCURRENT.get(model,1))
# Track which GPU this agent is using (TTL 120s covers typical request)
if r and agent:
@@ -527,11 +628,11 @@ def chat():
resp = requests.post(url+"/chat/completions", json=rd,
headers={"Content-Type":"application/json","Authorization":"Bearer not-needed"}, timeout=300, stream=is_stream)
lat = int((time.time()-start)*1000)
gpu_decr(model)
gpu_release_slot(model)
if resp.status_code != 200:
if resp.status_code in (502, 504):
trip_circuit(model, 30)
trip_circuit(model)
return jsonify({"error":"GPU error "+str(resp.status_code)}), 502
if is_stream:
# Buffer SSE chunks, handle split lines for large responses
@@ -578,6 +679,12 @@ def chat():
sse_resp.headers["X-Context-Remaining"] = str(max(0, ctx_remaining))
sse_resp.headers["X-Context-Warning"] = ctx_warning
sse_resp.headers["X-Context-Model"] = model
# LiteLLM spend tracking: best-effort token counts from stream timings
pt = stream_timings.get("prompt_n", 0) if stream_timings else 0
ct = stream_timings.get("predicted_n", 0) if stream_timings else 0
sse_resp.headers["X-Usage-Tokens"] = json.dumps({
"prompt_tokens": pt, "completion_tokens": ct, "model": model
})
return sse_resp
data = clean_response(resp.json())
for c in data.get("choices",[]):
@@ -602,15 +709,19 @@ def chat():
resp.headers["X-Context-Remaining"] = str(max(0, ctx_remaining))
resp.headers["X-Context-Warning"] = ctx_warning
resp.headers["X-Context-Model"] = model
# LiteLLM spend tracking: return token counts for cost computation
resp.headers["X-Usage-Tokens"] = json.dumps({
"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "model": model
})
bcast()
return resp
except requests.Timeout:
gpu_decr(model)
trip_circuit(model, 30)
gpu_release_slot(model)
trip_circuit(model)
log.error("TIMEOUT: %s -> %s (Circuit tripped)", agent, model)
return jsonify({"error":"timeout"}), 504
except Exception as e:
gpu_decr(model)
gpu_release_slot(model)
log.error("Error: %s\n%s", e, traceback.format_exc())
return jsonify({"error":str(e)}), 500
@@ -871,6 +982,38 @@ def metrics_latency():
"requests_per_min": len(last_min),
"count": len(recent)
})
@app.route("/health/unified")
def health_unified():
"""Unified health aggregating all layers: Router + Redis + GPUs + Circuit Breaker + Scores."""
gpus = {}
for m in GPU_URLS:
h = check_gpu_health(m, sidecar_timeout=1.5, gpu_timeout=1)
h["active_requests"] = gpu_active_count(m)
h["max_concurrent"] = GPU_MAX_CONCURRENT.get(m, 1)
h["health_score"] = gpu_health_score(m)
h["circuit_open"] = is_circuit_tripped(m)
gpus[m] = h
circuit_state = {}
for m in GPU_URLS:
cooldown_until = r.ttl("circuit:" + m + ":open") if r else None
circuit_state[m] = {
"open": is_circuit_tripped(m),
"cooldown_remaining_s": max(0, cooldown_until) if cooldown_until and cooldown_until > 0 else 0,
"trip_count": int(r.get("circuit:" + m + ":count") or 0) if r else 0
}
overall = "healthy"
if not r:
overall = "degraded"
if all(circuit_state[m]["open"] for m in GPU_URLS):
overall = "down"
return jsonify({
"status": overall, "router": "healthy",
"redis": "connected" if r else "down",
"gpus": gpus, "circuit_breaker": circuit_state,
"scores": {m: gpu_health_score(m) for m in GPU_URLS},
"available_models": available_models(), "timestamp": time.time()
})
@app.route("/stream")
def stream():
def ev():