fix: harden live-overview probes (Qodo review of #52 + colleague)

- _metrics.parse_metrics: skip non-finite (NaN/inf) values — int() would raise
  and break the best-effort contract (Qodo #1).
- _metrics.http_get_text: cap the body at 5 MiB (read max_bytes+1, treat overflow
  as unavailable) so a misbehaving backend can't stress memory (Qodo #2).
- gateway /status: drop base_url from the payload — it's internal-only routing
  detail and /status may be reached over a public tunnel; matches the documented
  schema (Qodo #3).
- gateway fleet_status_payload: probe backends in parallel (ThreadPoolExecutor)
  with a bounded 3s timeout, so /status can't hang for timeout × N on a slow
  backend (Qodo #4); and probe_backend short-circuits /metrics when /health fails
  (colleague review). Order preserved.

Tests for each; 345 pass; black/isort/flake8/bandit clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NnDfFkZkXz8C68hr3AA9Qa

Author: OriNachum

model_gear/_metrics.py

@@ -10,9 +10,14 @@
 from __future__ import annotations
 
 import json
+import math
 import urllib.error
 import urllib.request
 
+# Cap a single GET body so a misbehaving backend can't stress memory/latency. A
+# vLLM /metrics scrape is well under this; /health is tiny.
+_MAX_BODY_BYTES = 5 * 1024 * 1024
+
 # The handful of vLLM series the live view reports. "busy" = running/waiting now;
 # "usage" = cumulative tokens + finished requests by reason. Summed across the
 # engine/model labels vLLM attaches (a single backend may expose >1 engine).
@@ -54,6 +59,8 @@ def parse_metrics(text: str) -> dict:
             val = float(value)
         except ValueError:
             continue
+        if not math.isfinite(val):
+            continue  # NaN/inf would later make int() raise — skip (best-effort contract)
         brace = left.find("{")
         name = left[:brace] if brace >= 0 else left
         labels = left[brace:] if brace >= 0 else ""
@@ -83,17 +90,27 @@ def parse_metrics(text: str) -> dict:
     return out
 
 
-def http_get_text(url: str, *, timeout: float = 3.0) -> str | None:
-    """Best-effort GET → body text, or ``None`` if unreachable / non-2xx. Never raises."""
+def http_get_text(
+    url: str, *, timeout: float = 3.0, max_bytes: int = _MAX_BODY_BYTES
+) -> str | None:
+    """Best-effort GET → body text, or ``None`` if unreachable / non-2xx / oversized.
+
+    Reads at most ``max_bytes`` (+1 to detect overflow): an over-cap body is treated
+    as unavailable rather than buffered whole, so a misbehaving backend can't stress
+    memory. Never raises.
+    """
     try:
         with urllib.request.urlopen(
             url, timeout=timeout
         ) as r:  # nosec B310 - http(s) only, fixed scheme
-            if 200 <= r.status < 300:
-                return r.read().decode("utf-8", errors="replace")
+            if not (200 <= r.status < 300):
+                return None
+            data = r.read(max_bytes + 1)
+            if len(data) > max_bytes:
+                return None  # oversized → best-effort fail rather than buffer it whole
+            return data.decode("utf-8", errors="replace")
     except (urllib.error.URLError, OSError, ValueError):
         return None
-    return None
 
 
 def http_get_json(url: str, *, timeout: float = 3.0) -> dict | None:
@@ -120,9 +137,9 @@ def probe_backend(base_url: str, *, timeout: float = 3.0) -> dict:
     ``None`` when ``/metrics`` is unreachable (an engine can be loading or down).
     """
     base = base_url.rstrip("/")
-    healthy = health_ok(base, timeout=timeout)
+    if not health_ok(base, timeout=timeout):
+        # Short-circuit: a down backend has no useful /metrics, so skip the second
+        # request (halves the timeout cost for a dead backend).
+        return {"health": "unreachable", "metrics": None}
     raw = http_get_text(base + "/metrics", timeout=timeout)
-    return {
-        "health": "ok" if healthy else "unreachable",
-        "metrics": parse_metrics(raw) if raw is not None else None,
-    }
+    return {"health": "ok", "metrics": parse_metrics(raw) if raw is not None else None}

model_gear/gateway/server.py

@@ -18,6 +18,7 @@
 import http.client
 import json
 import sys
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Callable, Iterable
@@ -341,6 +342,10 @@ def handle_audio_post(
 # one JSON the host-side `model overview --live` renders. The prober is injected so
 # this is unit-testable without sockets.
 
+# Per-backend probe timeout for /status: bounded + probed in parallel (below) so a
+# slow/down backend can't make the whole /status call hang for connect_timeout × N.
+_STATUS_PROBE_TIMEOUT = 3.0
+
 
 def _endpoints_for(table: RoutingTable, audio: bool) -> list[str]:
     """OpenAI endpoints this gateway actually serves, by the task families present."""
@@ -365,11 +370,24 @@ def _endpoints_for(table: RoutingTable, audio: bool) -> list[str]:
 def fleet_status_payload(
     table: RoutingTable, cfg: ServerConfig, probe=_metrics.probe_backend
 ) -> dict:
-    """Live status for every backend + an aggregate busy count + the endpoint list."""
+    """Live status for every backend + an aggregate busy count + the endpoint list.
+
+    Backends are probed **in parallel** with a bounded timeout, so a slow/down
+    backend can't make ``/status`` hang for ``timeout × N``. ``base_url`` is
+    intentionally **not** in the payload — those are internal-only routing details
+    and ``/status`` may be reached over a public tunnel.
+    """
+    members = list(table.backends)
+    if members:
+        with ThreadPoolExecutor(max_workers=len(members)) as pool:
+            results = list(
+                pool.map(lambda b: probe(b.base_url, timeout=_STATUS_PROBE_TIMEOUT), members)
+            )
+    else:
+        results = []
     backends: list[dict] = []
     running = waiting = 0
-    for b in table.backends:
-        st = probe(b.base_url, timeout=cfg.connect_timeout)
+    for b, st in zip(members, results):
         metrics = st.get("metrics") or {}
         running += int(metrics.get("running", 0) or 0)
         waiting += int(metrics.get("waiting", 0) or 0)
@@ -378,7 +396,6 @@ def fleet_status_payload(
                 "name": b.name,
                 "task": b.task,
                 "served_name": b.served_name,
-                "base_url": b.base_url,
                 "health": st.get("health", "unreachable"),
                 "metrics": st.get("metrics"),
             }

tests/test_gateway_status.py

@@ -78,6 +78,21 @@ def fake_probe(base_url, *, timeout):
     assert embed["health"] == "unreachable" and embed["metrics"] is None
     primary = next(b for b in payload["backends"] if b["name"] == "primary")
     assert primary["task"] == "generate" and primary["served_name"] == "P"
+    # base_url is internal-only routing detail — must NOT leak in /status (Qodo).
+    assert all("base_url" not in b for b in payload["backends"])
+
+
+def test_fleet_status_probes_in_parallel_preserve_order() -> None:
+    # Parallel fan-out must still return backends in table order.
+    seen = []
+
+    def probe(base_url, *, timeout):
+        seen.append(base_url)
+        return {"health": "ok", "metrics": {"running": 1, "waiting": 0}}
+
+    payload = S.fleet_status_payload(_table(), _cfg(), probe=probe)
+    assert [b["name"] for b in payload["backends"]] == ["primary", "embed", "rerank"]
+    assert payload["busy"] == {"running": 3, "waiting": 0}  # 3 backends × running 1
 
 
 def test_fleet_status_payload_all_unreachable() -> None:

tests/test_overview_live.py

@@ -52,6 +52,48 @@ def test_parse_metrics_skips_malformed_lines() -> None:
     assert m["running"] == 4
 
 
+def test_parse_metrics_skips_non_finite() -> None:
+    # NaN/inf must be dropped, not crash the later int() (Qodo: best-effort contract).
+    text = (
+        'vllm:num_requests_running{e="0"} nan\n'
+        'vllm:num_requests_running{e="1"} inf\n'
+        'vllm:num_requests_running{e="2"} 2.0\n'
+    )
+    assert _metrics.parse_metrics(text)["running"] == 2
+
+
+# --- http_get_text body cap + probe short-circuit -------------------------
+
+
+class _FakeResp:
+    def __init__(self, data: bytes, status: int = 200) -> None:
+        self._data, self.status = data, status
+
+    def read(self, n: int = -1) -> bytes:
+        return self._data[:n] if n and n > 0 else self._data
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+
+def test_http_get_text_caps_oversized_body(monkeypatch) -> None:
+    monkeypatch.setattr("urllib.request.urlopen", lambda url, timeout=0: _FakeResp(b"x" * 1000))
+    assert _metrics.http_get_text("http://x/m", max_bytes=100) is None  # over cap → unavailable
+    assert _metrics.http_get_text("http://x/m", max_bytes=5000) == "x" * 1000
+
+
+def test_probe_backend_short_circuits_when_unhealthy(monkeypatch) -> None:
+    calls: list[str] = []
+    monkeypatch.setattr(_metrics, "health_ok", lambda base, **k: False)
+    monkeypatch.setattr(_metrics, "http_get_text", lambda url, **k: calls.append(url))
+    st = _metrics.probe_backend("http://dead:8000")
+    assert st == {"health": "unreachable", "metrics": None}
+    assert calls == []  # /metrics never fetched for a down backend
+
+
 # --- section builders (pure) ----------------------------------------------