Phase 0: Dual-key router — no hardcoded keys, gemma sync, 262K context fix
- Removed all hardcoded API_KEYS from source (env var now REQUIRED) - Added dual-key transition support (old keys deprecated, logged) - Model sync: qwen3.5-9b-vlm → gemma-4-12b everywhere - Dense context: 131K → 262K (all 3 models at 262K) - Added /admin/keys endpoints for key lifecycle management - Added ADMIN_KEY env var for admin auth - Dashboard already synced to gemma-4-12b labels Deployment: set API_KEYS from phase0-dual-keys.json in docker-compose Transition: 7 days dual-key, then POST /admin/keys/revoke
This commit is contained in:
Abiba Bot
+475
-62
@@ -1,4 +1,4 @@
|
||||
import os, json, time, logging, traceback, threading, queue
|
||||
import os, json, time, logging, traceback, threading, queue, statistics, math
|
||||
import requests, redis
|
||||
from flask import Flask, request, jsonify, Response, stream_with_context
|
||||
|
||||
@@ -19,16 +19,16 @@ GPU_URLS = {
|
||||
}
|
||||
# Max concurrent requests per GPU (based on llama.cpp --parallel)
|
||||
GPU_MAX_CONCURRENT = {
|
||||
"qwen3.6-35B-A3B": 2, # 2 slots
|
||||
"qwen3.6-35B-A3B": 2, # 2 slots (cross-agent spread prevents overheating)
|
||||
"qwen3.6-27B-code": 2, # 2 slots
|
||||
"gemma-4-12b": 2, # 2 slots (7.1GB VRAM)
|
||||
"gemma-4-12b": 2, # 2 slots (12GB VRAM, 4GB headroom)
|
||||
}
|
||||
|
||||
# Context window sizes (tokens) — used for compaction signals
|
||||
GPU_CONTEXT = {
|
||||
"qwen3.6-35B-A3B": 131072,
|
||||
"qwen3.6-35B-A3B": 262144,
|
||||
"qwen3.6-27B-code": 262144,
|
||||
"gemma-4-12b": 262144, # 262K context (gemma-4-12b)
|
||||
"gemma-4-12b": 262144,
|
||||
}
|
||||
|
||||
TIER_MODELS = {
|
||||
@@ -36,18 +36,45 @@ TIER_MODELS = {
|
||||
"professional": ["qwen3.6-35B-A3B", "qwen3.6-27B-code", "gemma-4-12b"],
|
||||
"enterprise": ["qwen3.6-35B-A3B", "qwen3.6-27B-code", "gemma-4-12b"],
|
||||
}
|
||||
API_KEYS = {
|
||||
"sk-syslog-local-master-key": {"tier": "enterprise", "agent": "admin"},
|
||||
"sk-syslog-abiba": {"tier": "enterprise", "agent": "Abiba"},
|
||||
"sk-syslog-mumuni": {"tier": "enterprise", "agent": "Mumuni"},
|
||||
"sk-syslog-tanko": {"tier": "enterprise", "agent": "Tanko"},
|
||||
"sk-syslog-koby": {"tier": "enterprise", "agent": "Koby"},
|
||||
"sk-syslog-kagenz0": {"tier": "enterprise", "agent": "Kagenz0"},
|
||||
"sk-syslog-koonimo": {"tier": "enterprise", "agent": "Koonimo"},
|
||||
"sk-starter-abc123": {"tier": "starter", "agent": "test-starter"},
|
||||
"sk-professional-xyz789": {"tier": "professional", "agent": "test-pro"},
|
||||
# ── PHASE 0: Dual-Key API Key System ──
|
||||
# API_KEYS env var is REQUIRED (JSON string). No hardcoded fallback.
|
||||
# Format: {"sk-xxx": {"tier": "enterprise", "agent": "Name"}}
|
||||
# Deprecated keys get {"deprecated": true} — still accepted, logged with warning.
|
||||
_raw_keys = os.environ.get("API_KEYS")
|
||||
if not _raw_keys:
|
||||
raise RuntimeError("FATAL: API_KEYS environment variable is required. "
|
||||
"Set it in docker-compose.yml or .env file. "
|
||||
"No hardcoded keys fallback — this is a security feature.")
|
||||
API_KEYS = json.loads(_raw_keys)
|
||||
log.info("Loaded %d API keys from env var (%d deprecated)",
|
||||
len(API_KEYS), sum(1 for v in API_KEYS.values() if v.get("deprecated")))
|
||||
# Rate limits: requests per minute per API key tier
|
||||
RATE_LIMIT_RPM = {
|
||||
"enterprise": 120,
|
||||
"professional": 60,
|
||||
"starter": 20,
|
||||
}
|
||||
|
||||
def check_rate_limit(api_key, tier):
|
||||
"""Token bucket rate limiter using Redis. Returns (allowed, retry_after_or_remaining, reset_seconds)."""
|
||||
if not r:
|
||||
return True, 999, 60
|
||||
limit = RATE_LIMIT_RPM.get(tier, 30)
|
||||
key = f"ratelimit:{api_key}"
|
||||
current = int(r.get(key) or 0)
|
||||
if current >= limit:
|
||||
ttl = r.ttl(key)
|
||||
retry = max(ttl, 1) if ttl and ttl > 0 else 60
|
||||
return False, retry, 0
|
||||
pipe = r.pipeline()
|
||||
pipe.incr(key)
|
||||
pipe.expire(key, 60) # 1-minute sliding window
|
||||
pipe.execute()
|
||||
remaining = limit - (current + 1)
|
||||
reset_seconds = r.ttl(key) or 60
|
||||
return True, remaining, reset_seconds
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s [ROUTER] %(levelname)s %(message)s")
|
||||
log = logging.getLogger("router")
|
||||
try: r = redis.from_url(REDIS_URL, decode_responses=True); r.ping()
|
||||
@@ -94,24 +121,25 @@ def gpu_decr(model):
|
||||
if v and int(v) < 0:
|
||||
r.set("active:" + model, 0) # never go negative
|
||||
|
||||
def check_gpu_health(model):
|
||||
def check_gpu_health(model, sidecar_timeout=5, gpu_timeout=3):
|
||||
url = GPU_SIDECARS.get(model)
|
||||
if not url: return {"status": "unknown"}
|
||||
try:
|
||||
resp = requests.get(url, timeout=5)
|
||||
resp = requests.get(url, timeout=sidecar_timeout)
|
||||
if resp.status_code == 200:
|
||||
d = resp.json()
|
||||
pct = (d.get("vram_used_mb",0) / max(d.get("vram_total_mb",1), 1)) * 100
|
||||
status = "healthy" if pct < 90 else "saturated"
|
||||
status = "healthy" # VRAM usage != saturation; busy slots handled by is_gpu_busy()
|
||||
vram_warning = pct >= 95
|
||||
# Also check if llama.cpp endpoint is actually responding
|
||||
gpu_url = GPU_URLS.get(model, "")
|
||||
try:
|
||||
hr = requests.get(gpu_url.replace("/v1","") + "/health", headers={"Authorization": "Bearer not-needed"}, timeout=3)
|
||||
hr = requests.get(gpu_url.replace("/v1","") + "/health", headers={"Authorization": "Bearer not-needed"}, timeout=gpu_timeout)
|
||||
if hr.status_code != 200:
|
||||
status = "down"
|
||||
except Exception:
|
||||
status = "down"
|
||||
return {"status": status, "vram_used_mb": d.get("vram_used_mb"), "vram_total_mb": d.get("vram_total_mb"), "vram_pct": round(pct,1), "temp_c": d.get("temp_c"), "gpu_util_pct": d.get("gpu_util_pct"), "gpu_name": d.get("gpu_name"), "power_w": d.get("power_w"), "power_limit_w": d.get("power_limit_w")}
|
||||
return {"status": status, "vram_warning": vram_warning, "vram_used_mb": d.get("vram_used_mb"), "vram_total_mb": d.get("vram_total_mb"), "vram_pct": round(pct,1), "temp_c": d.get("temp_c"), "gpu_util_pct": d.get("gpu_util_pct"), "gpu_name": d.get("gpu_name"), "power_w": d.get("power_w"), "power_limit_w": d.get("power_limit_w")}
|
||||
except Exception: pass
|
||||
return {"status": "down"}
|
||||
|
||||
@@ -121,14 +149,65 @@ def estimate_tokens(msgs):
|
||||
"""Estimate token count from messages. Uses JSON length / 3.5 (closer to real tokenizer ratios for dense text)."""
|
||||
return len(json.dumps(msgs, default=str)) // 3.5
|
||||
|
||||
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."""
|
||||
if not r: return
|
||||
try:
|
||||
total_ms = queue_ms + inference_ms
|
||||
tps = completion_tokens / (inference_ms / 1000) if inference_ms > 0 and completion_tokens > 0 else 0
|
||||
rec = json.dumps({
|
||||
"ts": time.time(),
|
||||
"model": model, "agent": agent, "tier": tier, "reason": reason,
|
||||
"queue_ms": round(queue_ms, 1),
|
||||
"inference_ms": round(inference_ms, 1),
|
||||
"total_ms": round(total_ms, 1),
|
||||
"prompt_tokens": prompt_tokens,
|
||||
"completion_tokens": completion_tokens,
|
||||
"tokens_per_sec": round(tps, 1),
|
||||
"stream": stream
|
||||
})
|
||||
# Global recent list (last 500)
|
||||
r.lpush("perf:recent", rec)
|
||||
r.ltrim("perf:recent", 0, 499)
|
||||
# Per-model list (last 200)
|
||||
r.lpush("perf:model:" + model, rec)
|
||||
r.ltrim("perf:model:" + model, 0, 199)
|
||||
# Per-reason list (last 200)
|
||||
r.lpush("perf:reason:" + reason, rec)
|
||||
r.ltrim("perf:reason:" + reason, 0, 199)
|
||||
# Per-agent list (last 200)
|
||||
r.lpush("perf:agent:" + agent, rec)
|
||||
r.ltrim("perf:agent:" + agent, 0, 199)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def is_gpu_busy(model):
|
||||
"""Check if GPU is at or near max concurrent capacity."""
|
||||
active = gpu_active_count(model)
|
||||
max_c = GPU_MAX_CONCURRENT.get(model, 1)
|
||||
return active >= max_c
|
||||
|
||||
def select_best_gpu(candidates, reason):
|
||||
"""Pick the best GPU from candidates IN ORDER — first non-busy one wins."""
|
||||
def select_best_gpu(candidates, reason, agent=""):
|
||||
"""Pick best GPU, spreading agents across GPUs to prevent hotspots."""
|
||||
# Count how many distinct agents are on each GPU
|
||||
gpu_agent_counts = {}
|
||||
if r:
|
||||
for m in GPU_URLS:
|
||||
count = 0
|
||||
for ak in API_KEYS.values():
|
||||
if r.get("agent_gpu:" + ak["agent"] + ":" + m):
|
||||
count += 1
|
||||
gpu_agent_counts[m] = count
|
||||
# First pass: prefer GPUs with 0 other agents (fresh GPU for this agent)
|
||||
for m in candidates:
|
||||
if not is_gpu_busy(m) and gpu_agent_counts.get(m, 0) == 0:
|
||||
return {"model": m, "reason": reason}
|
||||
# Second pass: prefer GPU this agent is NOT already on (skip own GPU)
|
||||
if agent:
|
||||
for m in candidates:
|
||||
if not is_gpu_busy(m) and not r.get("agent_gpu:" + agent + ":" + m):
|
||||
return {"model": m, "reason": reason}
|
||||
# Third pass: any non-busy GPU
|
||||
for m in candidates:
|
||||
if not is_gpu_busy(m):
|
||||
return {"model": m, "reason": reason}
|
||||
@@ -144,7 +223,7 @@ def select_best_gpu(candidates, reason):
|
||||
return {"model": best, "reason": "load_balanced_" + reason}
|
||||
return None
|
||||
|
||||
def route(rd, tier):
|
||||
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")])
|
||||
@@ -163,53 +242,52 @@ def route(rd, tier):
|
||||
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")
|
||||
alt = select_best_gpu(alts, "explicit", agent)
|
||||
if alt: return alt
|
||||
return {"model": target, "reason": "explicit"}
|
||||
|
||||
if hints:
|
||||
if hints.get("priority")=="speed" and "gemma-4-12b" in avail:
|
||||
return select_best_gpu(["gemma-4-12b"], "hint_speed") or {"model":"gemma-4-12b","reason":"hint_speed"}
|
||||
if hints.get("priority")=="quality" and "qwen3.6-27B-code" in avail:
|
||||
return select_best_gpu(["qwen3.6-27B-code"], "hint_quality") or {"model":"qwen3.6-27B-code","reason":"hint_quality"}
|
||||
return select_best_gpu(["gemma-4-12b"], "hint_speed", agent) or {"model":"gemma-4-12b","reason":"hint_speed"}
|
||||
if hints.get("priority")=="quality" and "qwen3.6-35B-A3B" in avail:
|
||||
return select_best_gpu(["qwen3.6-35B-A3B"], "hint_quality", agent) or {"model":"qwen3.6-35B-A3B","reason":"hint_quality"}
|
||||
|
||||
first_msg = msgs[0].get("content","") if msgs else ""
|
||||
words = len(first_msg.split()) if isinstance(first_msg, str) else 99
|
||||
|
||||
# TIER 1: Lightweight — single-turn short queries → VLM first
|
||||
if not sys and turns <= 1 and words <= 100 and "gemma-4-12b" in avail:
|
||||
# TIER 1: Lightweight — single-turn short queries → VLM (fastest)
|
||||
if not sys and turns <= 1 and t <= 500 and words <= 100 and "gemma-4-12b" in avail:
|
||||
if not is_gpu_busy("gemma-4-12b"):
|
||||
return {"model":"gemma-4-12b","reason":"lightweight"}
|
||||
# VLM busy — fall back to Dense, then MoE
|
||||
fallback = [m for m in ["qwen3.6-35B-A3B","qwen3.6-27B-code"] if m in avail]
|
||||
result = select_best_gpu(fallback, "lightweight_fallback")
|
||||
# VLM busy — Dense is faster for short queries than MoE
|
||||
fallback = [m for m in ["qwen3.6-27B-code","qwen3.6-35B-A3B"] if m in avail]
|
||||
result = select_best_gpu(fallback, "lightweight_fallback", agent)
|
||||
if result: return result
|
||||
|
||||
# TIER 2: Simple conversations — short context, any prompt → VLM preferred
|
||||
if t <= 1000 and turns <= 4 and "gemma-4-12b" in avail:
|
||||
# TIER 2: Simple conversations — VLM primary (up to 15K tok), fastest for moderate chat
|
||||
if t <= 15000 and turns <= 12 and "gemma-4-12b" in avail:
|
||||
if not is_gpu_busy("gemma-4-12b"):
|
||||
return {"model":"gemma-4-12b","reason":"simple_conv"}
|
||||
# VLM busy — try Dense
|
||||
if "qwen3.6-27B-code" in avail and not is_gpu_busy("qwen3.6-27B-code"):
|
||||
return {"model":"qwen3.6-27B-code","reason":"simple_conv_fallback"}
|
||||
# VLM busy — fall back to Dense, then MoE
|
||||
fallback = [m for m in ["qwen3.6-27B-code","qwen3.6-35B-A3B"] if m in avail]
|
||||
result = select_best_gpu(fallback, "simple_conv_fallback", agent)
|
||||
if result: return result
|
||||
|
||||
# TIER 3: Heavy reasoning — extremely large context or very long conversations
|
||||
if t > 50000 or turns > 25:
|
||||
# MoE first (131K context handles heavy sessions), then Dense (98K reasoning), then Light (131K fallback)
|
||||
# TIER 3: Medium complexity — Dense primary, VLM fallback (quality + speed balance)
|
||||
if t <= 25000:
|
||||
candidates = [m for m in ["qwen3.6-27B-code","gemma-4-12b","qwen3.6-35B-A3B"] if m in avail]
|
||||
result = select_best_gpu(candidates, "medium", agent)
|
||||
if result: return result
|
||||
|
||||
# TIER 4: Heavy reasoning — MoE primary (workhorse), Dense fallback
|
||||
if t > 25000:
|
||||
candidates = [m for m in ["qwen3.6-35B-A3B","qwen3.6-27B-code","gemma-4-12b"] if m in avail]
|
||||
result = select_best_gpu(candidates, "heavy_reasoning")
|
||||
result = select_best_gpu(candidates, "heavy_reasoning", agent)
|
||||
if result: return result
|
||||
|
||||
# TIER 4: Default — MoE first, VLM helps, Dense last (slow)
|
||||
if t <= 50000:
|
||||
candidates = [m for m in ["qwen3.6-35B-A3B","gemma-4-12b","qwen3.6-27B-code"] if m in avail]
|
||||
result = select_best_gpu(candidates, "default")
|
||||
if result: return result
|
||||
|
||||
# Fallback — best available
|
||||
if "qwen3.6-35B-A3B" in avail and not is_gpu_busy("qwen3.6-35B-A3B"):
|
||||
return {"model":"qwen3.6-35B-A3B","reason":"default_moe"}
|
||||
result = select_best_gpu([m for m in avail], "fallback")
|
||||
# TIER 5: Default — Dense primary, MoE fallback
|
||||
candidates = [m for m in ["qwen3.6-27B-code","gemma-4-12b","qwen3.6-35B-A3B"] if m in avail]
|
||||
result = select_best_gpu(candidates, "default", agent)
|
||||
if result: return result
|
||||
return {"model":avail[0],"reason":"last_resort"}
|
||||
|
||||
@@ -266,6 +344,27 @@ def chat():
|
||||
return jsonify({"error": "Unauthorized — valid API key required"}), 401
|
||||
ki = API_KEYS[ak]
|
||||
tier, agent = ki["tier"], ki["agent"]
|
||||
# Phase 0: dual-key transition — log deprecated key usage
|
||||
if ki.get("deprecated"):
|
||||
new_key = next((k for k, v in API_KEYS.items()
|
||||
if v.get("agent") == agent and not v.get("deprecated")), None)
|
||||
log.warning("DEPRECATED_KEY: agent=%s using old key %s...%s — switch to %s...%s",
|
||||
agent, ak[:12], ak[-8:],
|
||||
new_key[:12] if new_key else "N/A",
|
||||
new_key[-8:] if new_key else "N/A")
|
||||
if r:
|
||||
r.incr("deprecated_usage:" + agent)
|
||||
|
||||
# Rate limit check
|
||||
allowed, rl_val, reset_sec = check_rate_limit(ak, tier)
|
||||
if not allowed:
|
||||
resp = jsonify({"error": "Rate limit exceeded", "retry_after_s": rl_val})
|
||||
resp.headers["Retry-After"] = str(rl_val)
|
||||
resp.headers["X-RateLimit-Limit"] = str(RATE_LIMIT_RPM.get(tier, 30))
|
||||
resp.headers["X-RateLimit-Remaining"] = "0"
|
||||
resp.headers["X-RateLimit-Reset"] = str(int(time.time() + rl_val))
|
||||
log.warning("RATE_LIMIT: %s (%s) exceeded limit", agent, ak[-8:])
|
||||
return resp, 429
|
||||
|
||||
# Allow agent to override queue timeout via header
|
||||
q_timeout = int(request.headers.get("X-Queue-Timeout", str(QUEUE_TIMEOUT)))
|
||||
@@ -281,7 +380,7 @@ def chat():
|
||||
r.set("session:" + session_id, session_tokens, ex=86400) # TTL 24h
|
||||
except Exception: pass
|
||||
|
||||
d = route(rd, tier)
|
||||
d = route(rd, tier, agent)
|
||||
queue_start = time.time()
|
||||
|
||||
# Queue loop: wait for a GPU slot instead of immediate 503
|
||||
@@ -293,22 +392,31 @@ def chat():
|
||||
log.warning("QUEUE_TIMEOUT: %s waited %.1fs, all GPUs saturated", agent, elapsed)
|
||||
return resp, 503
|
||||
time.sleep(0.5) # poll every 500ms
|
||||
d = route(rd, tier)
|
||||
d = route(rd, tier, agent)
|
||||
|
||||
waited = time.time() - queue_start
|
||||
if waited > 0.5:
|
||||
log.info("QUEUED: %s waited %.1fs before slot opened", agent, waited)
|
||||
queue_ms = (time.time() - queue_start) * 1000
|
||||
if queue_ms > 500:
|
||||
log.info("QUEUED: %s waited %.0fms before slot opened", agent, queue_ms)
|
||||
model, reason, url = d["model"], d["reason"], GPU_URLS[d["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:
|
||||
try: r.setex("agent_gpu:" + agent + ":" + model, 120, "1")
|
||||
except: pass
|
||||
if r:
|
||||
try:
|
||||
r.incr("routes:"+model); r.incr("routes:tier:"+tier); r.incr("routes:agent:"+agent)
|
||||
r.incr("ts:"+model+":"+time.strftime("%Y%m%d%H"))
|
||||
r.lpush("routes:recent", json.dumps({"ts":time.time(),"model":model,"reason":reason,"tier":tier,"agent":agent}))
|
||||
r.lpush("routes:recent", json.dumps({"ts":time.time(),"model":model,"reason":reason,"tier":tier,"agent":agent,"queue_ms": round(queue_ms,1)}))
|
||||
r.ltrim("routes:recent",0,999)
|
||||
except Exception: pass
|
||||
start = time.time()
|
||||
@@ -319,14 +427,47 @@ def chat():
|
||||
|
||||
if resp.status_code != 200: return jsonify({"error":"GPU error "+str(resp.status_code)}), 502
|
||||
if is_stream:
|
||||
# Buffer SSE chunks, handle split lines for large responses
|
||||
chunks = []
|
||||
stream_timings = {}
|
||||
buf = "" # accumulate partial lines
|
||||
for raw in resp.iter_content(chunk_size=None, decode_unicode=True):
|
||||
if raw:
|
||||
cleaned = clean_unicode(raw)
|
||||
chunks.append(cleaned)
|
||||
buf += cleaned
|
||||
# Process complete lines from buffer
|
||||
while "\n" in buf:
|
||||
line, buf = buf.split("\n", 1)
|
||||
line = line.strip()
|
||||
if line.startswith("data: ") and not stream_timings:
|
||||
js = line[6:].strip()
|
||||
if js.startswith("{") and "timings" in js and "predicted_n" in js:
|
||||
try:
|
||||
tj = json.loads(js).get("timings", {})
|
||||
if tj:
|
||||
stream_timings = tj
|
||||
except: pass
|
||||
# Store perf record with real token counts from stream
|
||||
if stream_timings:
|
||||
pt = stream_timings.get("prompt_n", 0)
|
||||
ct = stream_timings.get("predicted_n", 0)
|
||||
tps = stream_timings.get("predicted_per_second", 0)
|
||||
gen_ms = stream_timings.get("predicted_ms", lat)
|
||||
store_perf_record(model, agent, tier, reason, queue_ms, gen_ms, pt, ct, True)
|
||||
else:
|
||||
store_perf_record(model, agent, tier, reason, queue_ms, lat, estimate_tokens(rd.get("messages",[])), 0, True)
|
||||
# Yield all chunks to client
|
||||
def gen():
|
||||
for raw in resp.iter_content(chunk_size=None, decode_unicode=True):
|
||||
if raw: yield clean_unicode(raw)
|
||||
for c in chunks: yield c
|
||||
bcast()
|
||||
ctx_remaining = GPU_CONTEXT.get(model, 65536) - max(session_tokens, estimate_tokens(rd.get("messages",[])))
|
||||
ctx_pct = ctx_remaining / GPU_CONTEXT.get(model, 65536) * 100
|
||||
ctx_warning = "compact_urgent" if ctx_pct < 5 else ("compact_recommended" if ctx_pct < 15 else ("compact_soon" if ctx_pct < 30 else "ok"))
|
||||
sse_resp = Response(stream_with_context(gen()), mimetype="text/event-stream")
|
||||
sse_resp.headers["X-RateLimit-Limit"] = str(_rl_limit)
|
||||
sse_resp.headers["X-RateLimit-Remaining"] = str(max(0, _rl_remaining))
|
||||
sse_resp.headers["X-RateLimit-Reset"] = str(int(time.time() + _rl_reset))
|
||||
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
|
||||
@@ -336,11 +477,21 @@ def chat():
|
||||
msg = c.get("message",{})
|
||||
if not msg.get("content") and msg.get("reasoning_content"):
|
||||
msg["content"] = msg["reasoning_content"]
|
||||
# Extract performance data from llama.cpp response
|
||||
usage = data.get("usage", {})
|
||||
timings = data.get("timings", {})
|
||||
prompt_tokens = usage.get("prompt_tokens", 0)
|
||||
completion_tokens = usage.get("completion_tokens", 0)
|
||||
inference_ms = lat # total GPU round-trip
|
||||
store_perf_record(model, agent, tier, reason, queue_ms, inference_ms, prompt_tokens, completion_tokens, False)
|
||||
ctx_remaining = GPU_CONTEXT.get(model, 65536) - max(session_tokens, estimate_tokens(rd.get("messages",[])))
|
||||
ctx_pct = ctx_remaining / GPU_CONTEXT.get(model, 65536) * 100
|
||||
ctx_warning = "compact_urgent" if ctx_pct < 5 else ("compact_recommended" if ctx_pct < 15 else ("compact_soon" if ctx_pct < 30 else "ok"))
|
||||
data["routing"] = {"model":model,"reason":reason,"gpu":url,"tier":tier,"agent":agent,"latency_ms":lat,"active_gpu":gpu_active_count(model),"context_remaining": max(0, ctx_remaining),"context_pct": round(ctx_pct,1),"context_warning": ctx_warning}
|
||||
data["routing"] = {"model":model,"reason":reason,"gpu":url,"tier":tier,"agent":agent,"latency_ms":lat,"queue_ms": round(queue_ms,1),"active_gpu":gpu_active_count(model),"context_remaining": max(0, ctx_remaining),"context_pct": round(ctx_pct,1),"context_warning": ctx_warning}
|
||||
resp = jsonify(data)
|
||||
resp.headers["X-RateLimit-Limit"] = str(_rl_limit)
|
||||
resp.headers["X-RateLimit-Remaining"] = str(max(0, _rl_remaining))
|
||||
resp.headers["X-RateLimit-Reset"] = str(int(time.time() + _rl_reset))
|
||||
resp.headers["X-Context-Remaining"] = str(max(0, ctx_remaining))
|
||||
resp.headers["X-Context-Warning"] = ctx_warning
|
||||
resp.headers["X-Context-Model"] = model
|
||||
@@ -355,14 +506,163 @@ def chat():
|
||||
log.error("Error: %s\n%s", e, traceback.format_exc())
|
||||
return jsonify({"error":str(e)}), 500
|
||||
|
||||
@app.route("/metrics/performance")
|
||||
def performance():
|
||||
"""Per-request performance analytics with percentiles per model/reason/agent."""
|
||||
if not r: return jsonify({"error": "Redis unavailable"}), 503
|
||||
try:
|
||||
window_hours = int(request.args.get("window", "24"))
|
||||
model_filter = request.args.get("model", "all")
|
||||
|
||||
# Load recent records
|
||||
cutoff = time.time() - (window_hours * 3600)
|
||||
raw = r.lrange("perf:recent", 0, -1)
|
||||
records = []
|
||||
for x in raw:
|
||||
try:
|
||||
rec = json.loads(x)
|
||||
if rec["ts"] >= cutoff:
|
||||
records.append(rec)
|
||||
except: pass
|
||||
|
||||
# Filter by model if specified
|
||||
if model_filter != "all":
|
||||
records = [r for r in records if r["model"] == model_filter]
|
||||
|
||||
if not records:
|
||||
return jsonify({"models": [], "reasons": [], "agents": [], "summary": {"total_requests": 0}})
|
||||
|
||||
def pct(values, p):
|
||||
if len(values) < 2: return round(values[0], 1) if values else 0
|
||||
return round(statistics.quantiles(sorted(values), n=100, method='inclusive')[min(p-1, 98)], 1)
|
||||
|
||||
# Per-model stats
|
||||
model_groups = {}
|
||||
for rec in records:
|
||||
m = rec["model"]
|
||||
if m not in model_groups: model_groups[m] = []
|
||||
model_groups[m].append(rec)
|
||||
|
||||
models = []
|
||||
for m, recs in sorted(model_groups.items()):
|
||||
latencies = [r["total_ms"] for r in recs]
|
||||
tps_vals = [r["tokens_per_sec"] for r in recs if r["tokens_per_sec"] > 0]
|
||||
non_stream = [r for r in recs if not r["stream"]]
|
||||
queue_times = [r["queue_ms"] for r in non_stream]
|
||||
models.append({
|
||||
"model": m,
|
||||
"count": len(recs),
|
||||
"stream_pct": round(len([r for r in recs if r["stream"]]) / len(recs) * 100, 1),
|
||||
"latency": {
|
||||
"avg": round(statistics.mean(latencies), 1),
|
||||
"p50": pct(latencies, 50),
|
||||
"p95": pct(latencies, 95),
|
||||
"p99": pct(latencies, 99)
|
||||
},
|
||||
"throughput": {
|
||||
"avg_tokens_per_sec": round(statistics.mean(tps_vals), 1) if tps_vals else 0,
|
||||
"p50": pct(tps_vals, 50) if tps_vals else 0,
|
||||
"p95": pct(tps_vals, 95) if tps_vals else 0,
|
||||
},
|
||||
"queue": {
|
||||
"avg_ms": round(statistics.mean(queue_times), 1) if queue_times else 0,
|
||||
"p95_ms": pct(queue_times, 95) if queue_times else 0,
|
||||
} if queue_times else None
|
||||
})
|
||||
|
||||
# Per-reason stats
|
||||
reason_groups = {}
|
||||
for rec in records:
|
||||
rsn = rec["reason"]
|
||||
if rsn not in reason_groups: reason_groups[rsn] = []
|
||||
reason_groups[rsn].append(rec)
|
||||
|
||||
reasons = []
|
||||
for rsn, recs in sorted(reason_groups.items(), key=lambda x: -len(x[1])):
|
||||
latencies = [r["total_ms"] for r in recs]
|
||||
reasons.append({
|
||||
"reason": rsn,
|
||||
"count": len(recs),
|
||||
"avg_total_ms": round(statistics.mean(latencies), 1),
|
||||
"p95_total_ms": pct(latencies, 95)
|
||||
})
|
||||
|
||||
# Per-agent stats
|
||||
agent_groups = {}
|
||||
for rec in records:
|
||||
ag = rec["agent"]
|
||||
if ag not in agent_groups: agent_groups[ag] = []
|
||||
agent_groups[ag].append(rec)
|
||||
|
||||
agents = []
|
||||
for ag, recs in sorted(agent_groups.items(), key=lambda x: -len(x[1])):
|
||||
latencies = [r["total_ms"] for r in recs]
|
||||
tps_vals = [r["tokens_per_sec"] for r in recs if r["tokens_per_sec"] > 0]
|
||||
agents.append({
|
||||
"agent": ag,
|
||||
"count": len(recs),
|
||||
"avg_total_ms": round(statistics.mean(latencies), 1),
|
||||
"avg_tokens_per_sec": round(statistics.mean(tps_vals), 1) if tps_vals else 0
|
||||
})
|
||||
|
||||
all_lat = [r["total_ms"] for r in records]
|
||||
all_tps = [r["tokens_per_sec"] for r in records if r["tokens_per_sec"] > 0]
|
||||
summary = {
|
||||
"total_requests": len(records),
|
||||
"window_hours": window_hours,
|
||||
"latency": {
|
||||
"avg_ms": round(statistics.mean(all_lat), 1),
|
||||
"p50_ms": pct(all_lat, 50),
|
||||
"p95_ms": pct(all_lat, 95),
|
||||
"p99_ms": pct(all_lat, 99)
|
||||
},
|
||||
"throughput_avg_tps": round(statistics.mean(all_tps), 1) if all_tps else 0
|
||||
}
|
||||
|
||||
return jsonify({"models": models, "reasons": reasons, "agents": agents, "summary": summary})
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@app.route("/metrics/scatter")
|
||||
def scatter():
|
||||
"""Return individual data points for scatter plots (prompt_tokens vs latency)."""
|
||||
if not r: return jsonify({"error": "Redis unavailable"}), 503
|
||||
try:
|
||||
window_hours = int(request.args.get("window", "24"))
|
||||
model_filter = request.args.get("model", "all")
|
||||
cutoff = time.time() - (window_hours * 3600)
|
||||
raw = r.lrange("perf:recent", 0, -1)
|
||||
points = []
|
||||
for x in raw:
|
||||
try:
|
||||
rec = json.loads(x)
|
||||
if rec["ts"] >= cutoff:
|
||||
if model_filter == "all" or rec["model"] == model_filter:
|
||||
points.append({
|
||||
"model": rec["model"],
|
||||
"agent": rec["agent"],
|
||||
"reason": rec["reason"],
|
||||
"prompt_tokens": int(rec.get("prompt_tokens", 0)),
|
||||
"completion_tokens": rec.get("completion_tokens", 0),
|
||||
"inference_ms": round(rec["inference_ms"], 1),
|
||||
"tokens_per_sec": rec.get("tokens_per_sec", 0),
|
||||
"stream": rec.get("stream", False)
|
||||
})
|
||||
except: pass
|
||||
return jsonify({"points": points, "count": len(points)})
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
@app.route("/v1/models")
|
||||
def models(): return jsonify({"object":"list","data":[{"id":m,"object":"model","owned_by":"syslog","status":check_gpu_health(m).get("status"),"gpu":check_gpu_health(m).get("gpu_name")} for m in GPU_URLS]})
|
||||
def models():
|
||||
def _h(m): return check_gpu_health(m, sidecar_timeout=1.5, gpu_timeout=1)
|
||||
return jsonify({"object":"list","data":[{"id":m,"object":"model","owned_by":"syslog","status":_h(m).get("status"),"gpu":_h(m).get("gpu_name")} for m in GPU_URLS]})
|
||||
|
||||
@app.route("/health")
|
||||
def health():
|
||||
gpus = {}
|
||||
for m in GPU_URLS:
|
||||
h = check_gpu_health(m)
|
||||
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)
|
||||
gpus[m] = h
|
||||
@@ -413,6 +713,119 @@ def stream():
|
||||
return Response(stream_with_context(ev()), mimetype="text/event-stream",
|
||||
headers={"Cache-Control":"no-cache","X-Accel-Buffering":"no","Access-Control-Allow-Origin":"*"})
|
||||
|
||||
# ── Phase 0: Admin Key Management ──
|
||||
ADMIN_KEY = os.environ.get("ADMIN_KEY", "")
|
||||
|
||||
def _admin_auth():
|
||||
"""Require admin key for management endpoints."""
|
||||
if not ADMIN_KEY:
|
||||
return False, "ADMIN_KEY not configured on server"
|
||||
ak = request.headers.get("Authorization","").replace("Bearer ","")
|
||||
if ak != ADMIN_KEY:
|
||||
return False, "Admin key required"
|
||||
return True, None
|
||||
|
||||
@app.route("/admin/keys")
|
||||
def admin_keys():
|
||||
"""List all API keys (masked) with agent, tier, and deprecation status."""
|
||||
ok, err = _admin_auth()
|
||||
if not ok: return jsonify({"error": err}), 401
|
||||
keys = []
|
||||
for key, info in API_KEYS.items():
|
||||
masked = key[:8] + "..." + key[-8:]
|
||||
keys.append({
|
||||
"masked": masked,
|
||||
"prefix": key[:8],
|
||||
"agent": info["agent"],
|
||||
"tier": info["tier"],
|
||||
"deprecated": info.get("deprecated", False),
|
||||
"length": len(key)
|
||||
})
|
||||
return jsonify({
|
||||
"total": len(keys),
|
||||
"active": sum(1 for k in keys if not k["deprecated"]),
|
||||
"deprecated": sum(1 for k in keys if k["deprecated"]),
|
||||
"keys": sorted(keys, key=lambda k: (k["deprecated"], k["agent"]))
|
||||
})
|
||||
|
||||
@app.route("/admin/keys/deprecation-summary")
|
||||
def admin_deprecation_summary():
|
||||
"""Summary of deprecated key usage (from Redis logs, if available)."""
|
||||
ok, err = _admin_auth()
|
||||
if not ok: return jsonify({"error": err}), 401
|
||||
deprecated_agents = []
|
||||
for key, info in API_KEYS.items():
|
||||
if info.get("deprecated"):
|
||||
# Check Redis for usage count
|
||||
count = 0
|
||||
if r:
|
||||
count = int(r.get("deprecated_usage:" + info["agent"]) or 0)
|
||||
deprecated_agents.append({
|
||||
"agent": info["agent"],
|
||||
"deprecated_uses": count,
|
||||
"needs_migration": count > 0
|
||||
})
|
||||
return jsonify({
|
||||
"deprecated_agents": sorted(deprecated_agents, key=lambda d: -d["deprecated_uses"]),
|
||||
"recommendation": "Run POST /admin/keys/revoke to remove keys with 0 usage"
|
||||
})
|
||||
|
||||
@app.route("/admin/keys/generate", methods=["POST"])
|
||||
def admin_generate_key():
|
||||
"""Generate a new API key for an agent. Body: {"agent": "Name", "tier": "enterprise"}"""
|
||||
ok, err = _admin_auth()
|
||||
if not ok: return jsonify({"error": err}), 401
|
||||
body = request.get_json(force=True)
|
||||
agent = body.get("agent", "").strip()
|
||||
tier = body.get("tier", "enterprise")
|
||||
if not agent:
|
||||
return jsonify({"error": "agent field required"}), 400
|
||||
if tier not in ("starter", "professional", "enterprise"):
|
||||
return jsonify({"error": "tier must be starter/professional/enterprise"}), 400
|
||||
# Generate secure key
|
||||
import secrets, hashlib
|
||||
prefix = hashlib.sha256(secrets.token_bytes(12)).hexdigest()[:8]
|
||||
suffix = secrets.token_hex(20)
|
||||
new_key = f"sk-{prefix}-{suffix}"
|
||||
# Update in-memory dict (note: not persisted across restarts without env var update)
|
||||
API_KEYS[new_key] = {"tier": tier, "agent": agent}
|
||||
log.info("KEY_GENERATED: agent=%s tier=%s key=%s...%s", agent, tier, new_key[:8], new_key[-8:])
|
||||
return jsonify({
|
||||
"agent": agent,
|
||||
"tier": tier,
|
||||
"key": new_key,
|
||||
"masked": new_key[:8] + "..." + new_key[-8:],
|
||||
"warning": "This key exists in memory only. Update API_KEYS env var and redeploy to persist."
|
||||
}), 201
|
||||
|
||||
@app.route("/admin/keys/revoke", methods=["POST"])
|
||||
def admin_revoke_key():
|
||||
"""Revoke a deprecated key. Body: {"agent": "Name"} or {"key_prefix": "sk-xxxx"}"""
|
||||
ok, err = _admin_auth()
|
||||
if not ok: return jsonify({"error": err}), 401
|
||||
body = request.get_json(force=True)
|
||||
agent = body.get("agent", "")
|
||||
key_prefix = body.get("key_prefix", "")
|
||||
revoked = []
|
||||
keys_to_remove = []
|
||||
for key, info in API_KEYS.items():
|
||||
if not info.get("deprecated"):
|
||||
continue
|
||||
if agent and info["agent"] == agent:
|
||||
keys_to_remove.append(key)
|
||||
elif key_prefix and key.startswith(key_prefix):
|
||||
keys_to_remove.append(key)
|
||||
for key in keys_to_remove:
|
||||
info = API_KEYS.pop(key)
|
||||
revoked.append({"agent": info["agent"], "masked": key[:8] + "..." + key[-8:]})
|
||||
log.warning("KEY_REVOKED: agent=%s key=%s...%s", info["agent"], key[:8], key[-8:])
|
||||
return jsonify({
|
||||
"revoked": len(revoked),
|
||||
"keys": revoked,
|
||||
"remaining_total": len(API_KEYS),
|
||||
"warning": "Memory-only revoke. Update API_KEYS env var and redeploy to persist."
|
||||
})
|
||||
|
||||
if __name__ == "__main__":
|
||||
log.info("Router on :9000 (load-aware)")
|
||||
app.run(host="0.0.0.0", port=9000, debug=False)
|
||||
|
||||
Reference in New Issue
Block a user