Coverage for orchestrant/benchmark/gateway.py: 95%
133 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-18 00:48 +0000
1"""Reports taken THROUGH the llm-stack gateway: which lane served, on what."""
3import collections
4import hashlib
5import json
6import threading
7import urllib.parse
8import urllib.request
11# What /gateway/info may put in a report: the image reference and shas.
12INFO_KEYS = (
13 "image",
14 "config_sha256",
15 "restart_sha256",
16 "registry_sha256",
17 "boot_config_sha256",
18 "lua_sha256",
19 "prompts_sha256",
20)
22# Process-wide: a report is written at run end from what its requests were told.
23_SERVED = collections.Counter()
24# Counter += is a read then a write: concurrent replies would lose counts, more so free-threaded.
25_SERVED_LOCK = threading.Lock()
28def _origin(url):
29 """host:port of `url`, localhost spelled 127.0.0.1; None without both."""
30 url = url or ""
31 try:
32 parts = urllib.parse.urlsplit(url if "://" in url else f"http://{url}")
33 port = parts.port
34 except ValueError:
35 return None
36 host = "127.0.0.1" if parts.hostname == "localhost" else parts.hostname
37 return f"{host}:{port}" if host and port else None
40def registry_sha256(doc):
41 """The registry's sha256 as the gateway's renderer computes it: canonical JSON."""
42 text = json.dumps(doc, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
43 return hashlib.sha256(text.encode("ascii")).hexdigest()
46def _registry(path=None):
47 """(path, document) of the registry the lab resolves backends from."""
48 from orchestrant.benchmark import openai_api # lazy: it imports provenance
50 path = path or openai_api.BACKENDS_FILE
51 try:
52 with open(path, encoding="utf-8") as f:
53 doc = json.load(f)
54 except (OSError, ValueError):
55 return path, {}
56 return path, doc if isinstance(doc, dict) else {}
59def _lane_names(serving, alias):
60 """[(role, lane)] an alias reaches: a route's lane and overflow, or raw-<lane>."""
61 lanes = serving.get("lanes") or {}
62 route = (serving.get("routes") or {}).get(alias)
63 if isinstance(route, dict):
64 pairs = (
65 ("primary", route.get("lane")),
66 ("overflow", route.get("overflow_lane")),
67 )
68 return [(role, lane) for role, lane in pairs if lane in lanes]
69 raw = alias[4:] if isinstance(alias, str) and alias.startswith("raw-") else None
70 offered = serving["gateway"].get("raw_routes") is True
71 return [("primary", raw)] if offered and raw in lanes else []
74def _lane_rows(serving, backends, alias, gateway):
75 """One row per lane an alias reaches, never one pointing back at the gateway."""
76 rows = []
77 for role, name in _lane_names(serving, alias):
78 backend = serving["lanes"][name].get("backend")
79 entry = backends.get(backend) or {}
80 url = entry.get("base_url")
81 url = url.rstrip("/") if isinstance(url, str) else None
82 # The renderer refuses such a lane; resolving one would recurse.
83 if url and _origin(url) != gateway:
84 rows.append(
85 {
86 "role": role,
87 "lane": name,
88 "backend": backend,
89 "base_url": url,
90 "model": entry.get("model"),
91 }
92 )
93 return rows
96def route(base_url, model, path=None):
97 """The lanes a gateway alias reaches, or None when base_url is not the gateway."""
98 path, doc = _registry(path)
99 serving = doc.get("serving")
100 gw = serving.get("gateway") if isinstance(serving, dict) else None
101 listen = gw.get("listen") if isinstance(gw, dict) else None
102 gateway = _origin(listen) if isinstance(listen, str) else None
103 if gateway is None or _origin(base_url) != gateway:
104 return None
105 try:
106 lanes = _lane_rows(serving, doc.get("backends") or {}, model, gateway)
107 error = None if lanes else f"the serving block routes no alias {model!r}"
108 except (AttributeError, TypeError, KeyError) as e: # a hand-edited block
109 lanes, error = [], f"unreadable serving block: {type(e).__name__}: {e}"[:200]
110 return {
111 "listen": listen,
112 "alias": model,
113 "lanes": lanes,
114 "registry": path,
115 "registry_sha256": registry_sha256(doc),
116 "error": error,
117 }
120def lane_behind(base_url, model, path=None):
121 """(base_url, model) of the lane the gateway sends `model` to first, or None."""
122 lanes = (route(base_url, model, path) or {}).get("lanes") or []
123 lane = next((r for r in lanes if r["role"] == "primary"), None)
124 return (lane["base_url"], lane["model"]) if lane else None
127def _header(headers, name):
128 try:
129 value = headers.get(name) if headers is not None else None
130 except Exception: # a header object of another shape is no evidence
131 return None
132 return value if isinstance(value, str) else None
135def note_reply(url, body, headers):
136 """Count one reply by the lane the gateway names in its X-Gw-Lane header."""
137 lane = _header(headers, "X-Gw-Lane")
138 if lane is None:
139 return
140 alias = body.get("model") if isinstance(body, dict) else None
141 alias = alias if isinstance(alias, str) else None
142 key = (_origin(url), alias, lane, _header(headers, "X-Gw-Rerouted"))
143 with _SERVED_LOCK:
144 _SERVED[key] += 1
147def served(base_url, model=None):
148 """Every reply this process got from the gateway at base_url, by alias, lane, reroute."""
149 origin = _origin(base_url)
150 with _SERVED_LOCK:
151 tally = sorted(_SERVED.items(), key=str)
152 return [
153 {"alias": alias, "lane": lane, "rerouted": rerouted, "replies": n}
154 for (where, alias, lane, rerouted), n in tally
155 if where == origin and (model is None or alias == model)
156 ]
159def info(base_url, timeout=3):
160 """The running gateway's /gateway/info, cut to INFO_KEYS, or {"error": ...}."""
161 url = f"{base_url.rstrip('/')}/gateway/info"
162 try:
163 with urllib.request.urlopen(url, timeout=timeout) as r: # nosec B310
164 doc = json.load(r)
165 except Exception as e: # a gateway that is down is a gap, not a crash
166 return {"error": f"{type(e).__name__}: {e}"[:160]}
167 if not isinstance(doc, dict):
168 return {"error": "/gateway/info is not a JSON object"}
169 return {k: doc[k] for k in INFO_KEYS if k in doc}
172def gateway_block(base_url, model, runtime_of, path=None):
173 """provenance["gateway"]: None for a direct lane; for the gateway, what served."""
174 found = route(base_url, model, path)
175 if found is None:
176 return None
177 for lane in found["lanes"]:
178 if lane["role"] != "primary":
179 lane["runtime"] = runtime_of(lane["base_url"], lane["model"])
180 live = info(base_url)
181 rendered_from = live.get("registry_sha256")
182 found["info"] = live
183 found["registry_matches"] = (
184 None if rendered_from is None else rendered_from == found["registry_sha256"]
185 )
186 found["served"] = served(base_url, model)
187 return found
190_GATEWAY_SHAS = (
191 ("restart_sha256", "image, boot config or Lua"),
192 ("config_sha256", "routes"),
193)
196def _lanes_served(block):
197 rows = [r for r in block.get("served") or [] if isinstance(r, dict)]
198 return sorted({f"{r.get('lane')} ({r.get('rerouted')})" for r in rows})
201def gateway_notes(old, new):
202 """For compare(): one run through the gateway and one not, or two gateways apart."""
203 before, after = old.get("gateway"), new.get("gateway")
204 if not (before or after):
205 return []
206 if not (before and after):
207 side, block = ("old", before) if before else ("new", after)
208 return [
209 f"only the {side} run went through the gateway (alias "
210 f"{block.get('alias')!r}): it re-encodes every body, and an alias "
211 f"that is not raw-* also shapes it (tools prompt, T=0 on GGUF lanes, "
212 f"power_mode dropped); its /v1/models lists aliases, not models"
213 ]
214 notes = []
215 aliases = before.get("alias"), after.get("alias")
216 if aliases[0] != aliases[1]:
217 notes.append(f"gateway alias differs: {aliases[0]!r} vs {aliases[1]!r}")
218 for key, what in _GATEWAY_SHAS:
219 was, now = ((b.get("info") or {}).get(key) for b in (before, after))
220 if was and now and was != now:
221 notes.append(f"the gateway's {what} changed ({key} {was[:12]}→{now[:12]})")
222 lanes = [_lanes_served(b) for b in (before, after)]
223 if lanes[0] != lanes[1]:
224 notes.append(f"other lanes served the two runs: {lanes[0]} vs {lanes[1]}")
225 return notes