Add production polish checks and settings

.github/workflows/test.yml

@@ -21,6 +21,13 @@ jobs:
         run: |
           python -m pip install --upgrade pip
           python -m pip install -r requirements.txt
+          python -m pip install -r requirements-dev.txt
+
+      - name: Lint
+        run: python -m ruff check .
+
+      - name: Compile Python
+        run: python -m compileall -q gateway worker training bench scripts tests
 
       - name: Run tests
         run: python -m pytest -q

bench/replay_traffic.py

@@ -9,7 +9,6 @@
 
 import httpx
 
-
 ROOT = Path(__file__).resolve().parents[1]
 RESULT_PATH = ROOT / "bench" / "results" / "replay-latest.json"
 TEXTS = [
@@ -31,7 +30,14 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-async def control_worker(client: httpx.AsyncClient, base_url: str, index: int, *, delay_ms: int, fail_rate: float) -> bool:
+async def control_worker(
+    client: httpx.AsyncClient,
+    base_url: str,
+    index: int,
+    *,
+    delay_ms: int,
+    fail_rate: float,
+) -> bool:
     try:
         response = await client.post(
             f"{base_url}/workers/{index}/control",
@@ -63,7 +69,13 @@ async def snapshot(client: httpx.AsyncClient, base_url: str, stage: str) -> dict
     }
 
 
-async def send_stage(client: httpx.AsyncClient, base_url: str, stage: str, seconds: float, rps: float) -> dict[str, Any]:
+async def send_stage(
+    client: httpx.AsyncClient,
+    base_url: str,
+    stage: str,
+    seconds: float,
+    rps: float,
+) -> dict[str, Any]:
     deadline = time.perf_counter() + seconds
     interval = 1 / rps
     sent = 0
@@ -106,17 +118,25 @@ async def main() -> None:
 
     async with httpx.AsyncClient(timeout=5.0) as client:
         await reset_workers(client, base_url)
-        stages.append(await send_stage(client, base_url, "baseline", args.seconds_per_stage, args.rps))
+        stages.append(
+            await send_stage(client, base_url, "baseline", args.seconds_per_stage, args.rps)
+        )
 
         await control_worker(client, base_url, 1, delay_ms=350, fail_rate=0)
-        stages.append(await send_stage(client, base_url, "slow-worker", args.seconds_per_stage, args.rps))
+        stages.append(
+            await send_stage(client, base_url, "slow-worker", args.seconds_per_stage, args.rps)
+        )
 
         await reset_workers(client, base_url)
         await control_worker(client, base_url, 0, delay_ms=0, fail_rate=0.45)
-        stages.append(await send_stage(client, base_url, "flaky-worker", args.seconds_per_stage, args.rps))
+        stages.append(
+            await send_stage(client, base_url, "flaky-worker", args.seconds_per_stage, args.rps)
+        )
 
         await reset_workers(client, base_url)
-        stages.append(await send_stage(client, base_url, "recovery", args.seconds_per_stage, args.rps))
+        stages.append(
+            await send_stage(client, base_url, "recovery", args.seconds_per_stage, args.rps)
+        )
 
     payload = {
         "base_url": base_url,

bench/run_bench.py

@@ -8,7 +8,6 @@
 
 import httpx
 
-
 SAMPLES = [
     "fast stable model service",
     "worker timeout and broken response",
@@ -26,7 +25,13 @@ def parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-async def call(client: httpx.AsyncClient, url: str, index: int, *, unique_requests: bool) -> tuple[float, bool, bool]:
+async def call(
+    client: httpx.AsyncClient,
+    url: str,
+    index: int,
+    *,
+    unique_requests: bool,
+) -> tuple[float, bool, bool]:
     started = time.perf_counter()
     text = SAMPLES[index % len(SAMPLES)]
     if unique_requests:
@@ -55,19 +60,32 @@ async def limited(index: int) -> tuple[float, bool, bool]:
     latencies = [latency for latency, ok, _ in results if ok]
     failures = sum(1 for _, ok, _ in results if not ok)
     cache_hits = sum(1 for _, ok, cached in results if ok and cached)
-    p95 = statistics.quantiles(latencies, n=20)[18] if len(latencies) >= 20 else max(latencies, default=0)
-
-    print(json.dumps({
-        "requests": args.requests,
-        "successes": len(latencies),
-        "failures": failures,
-        "cache_hits": cache_hits,
-        "cache_hit_rate": round(cache_hits / len(latencies), 4) if latencies else 0,
-        "total_ms": round(total_ms, 2),
-        "requests_per_second": round(args.requests / (total_ms / 1000), 2),
-        "p50_latency_ms": round(statistics.median(latencies), 2) if latencies else 0,
-        "p95_latency_ms": round(p95, 2),
-    }, indent=2))
+    p95 = (
+        statistics.quantiles(latencies, n=20)[18]
+        if len(latencies) >= 20
+        else max(latencies, default=0)
+    )
+
+    print(
+        json.dumps(
+            {
+                "requests": args.requests,
+                "successes": len(latencies),
+                "failures": failures,
+                "cache_hits": cache_hits,
+                "cache_hit_rate": round(cache_hits / len(latencies), 4)
+                if latencies
+                else 0,
+                "total_ms": round(total_ms, 2),
+                "requests_per_second": round(args.requests / (total_ms / 1000), 2),
+                "p50_latency_ms": round(statistics.median(latencies), 2)
+                if latencies
+                else 0,
+                "p95_latency_ms": round(p95, 2),
+            },
+            indent=2,
+        )
+    )
 
 
 if __name__ == "__main__":

bench/run_scenarios.py

@@ -16,7 +16,6 @@
 
 import httpx
 
-
 ROOT = Path(__file__).resolve().parents[1]
 RESULTS_DIR = ROOT / "bench" / "results"
 BASE_PORT = 8300

gateway/app/dashboard.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-
 DASHBOARD_HTML = """
 <!doctype html>
 <html lang="en">

gateway/app/main.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-import os
 import time
+from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator
+from typing import Any
 
 import httpx
 from fastapi import FastAPI, HTTPException
@@ -14,37 +14,13 @@
 from .metrics import MetricsStore
 from .router import Worker, WorkerPool
 from .schemas import GatewayPrediction, PredictRequest, WorkerControlRequest
+from .settings import GatewaySettings
 
-DEFAULT_WORKER_URL = "http://localhost:8011"
 MAX_WORKER_ATTEMPTS = 2
 JsonObject = dict[str, Any]
 WorkerRow = dict[str, str | int | float | None]
 
 
-def env_flag(name: str, default: bool = True) -> bool:
-    value = os.getenv(name)
-    if value is None:
-        return default
-    return value.lower() not in {"0", "false", "no", "off"}
-
-
-def parse_worker_hostports(env: dict[str, str] | None = None) -> list[str]:
-    values = env or os.environ
-    urls = []
-    for key in sorted(values):
-        if key.startswith("WORKER_") and key.endswith("_HOSTPORT"):
-            hostport = values[key].strip()
-            if hostport:
-                urls.append(f"http://{hostport}")
-    return urls
-
-
-def parse_worker_urls(raw: str | None) -> list[str]:
-    if not raw:
-        return parse_worker_hostports() or [DEFAULT_WORKER_URL]
-    return [item.strip() for item in raw.split(",") if item.strip()]
-
-
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     app.state.client = httpx.AsyncClient()
@@ -53,16 +29,17 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
 
 app = FastAPI(title="ModelMesh Gateway", version="0.1.0", lifespan=lifespan)
+settings = GatewaySettings.from_env()
 router = WorkerPool(
-    parse_worker_urls(os.getenv("WORKER_URLS")),
-    strategy=os.getenv("ROUTER_STRATEGY", "load_aware"),
+    settings.worker_urls,
+    strategy=settings.router_strategy,
 )
 cache = PredictionCache(
-    os.getenv("REDIS_URL"),
-    enabled=env_flag("CACHE_ENABLED", default=True),
+    settings.redis_url,
+    enabled=settings.cache_enabled,
 )
 metrics = MetricsStore()
-request_timeout = float(os.getenv("REQUEST_TIMEOUT_SECONDS", "1.0"))
+request_timeout = settings.request_timeout_seconds
 
 
 def elapsed_ms(started: float) -> float:

gateway/app/metrics.py

@@ -23,7 +23,13 @@ class MetricsStore:
     worker_errors: int = 0
     latencies_ms: deque[float] = field(default_factory=lambda: deque(maxlen=500))
 
-    def record_request(self, latency_ms: float, *, cached: bool, worker_error: bool = False) -> None:
+    def record_request(
+        self,
+        latency_ms: float,
+        *,
+        cached: bool,
+        worker_error: bool = False,
+    ) -> None:
         self.requests += 1
         self.latencies_ms.append(latency_ms)
         if cached:

gateway/app/settings.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from os import environ
+
+DEFAULT_WORKER_URL = "http://localhost:8011"
+ROUTER_STRATEGIES = {"load_aware", "round_robin"}
+FALSE_VALUES = {"0", "false", "no", "off"}
+TRUE_VALUES = {"1", "true", "yes", "on"}
+
+
+def parse_bool(raw: str | None, *, default: bool) -> bool:
+    if raw is None:
+        return default
+
+    value = raw.strip().lower()
+    if value in TRUE_VALUES:
+        return True
+    if value in FALSE_VALUES:
+        return False
+    raise ValueError(f"expected a boolean value, got {raw!r}")
+
+
+def parse_worker_hostports(env: Mapping[str, str] | None = None) -> list[str]:
+    values = env or environ
+    urls = []
+    for key in sorted(values):
+        if key.startswith("WORKER_") and key.endswith("_HOSTPORT"):
+            hostport = values[key].strip()
+            if hostport:
+                urls.append(f"http://{hostport}")
+    return urls
+
+
+def parse_worker_urls(raw: str | None, env: Mapping[str, str] | None = None) -> list[str]:
+    if raw:
+        return [item.strip().rstrip("/") for item in raw.split(",") if item.strip()]
+    return parse_worker_hostports(env) or [DEFAULT_WORKER_URL]
+
+
+def parse_timeout(raw: str | None, *, default: float) -> float:
+    if raw is None:
+        return default
+
+    try:
+        value = float(raw)
+    except ValueError as exc:
+        raise ValueError(f"REQUEST_TIMEOUT_SECONDS must be a number, got {raw!r}") from exc
+
+    if value <= 0:
+        raise ValueError("REQUEST_TIMEOUT_SECONDS must be greater than zero")
+    return value
+
+
+@dataclass(frozen=True)
+class GatewaySettings:
+    worker_urls: list[str]
+    router_strategy: str = "load_aware"
+    redis_url: str | None = None
+    cache_enabled: bool = True
+    request_timeout_seconds: float = 1.0
+
+    @classmethod
+    def from_env(cls, env: Mapping[str, str] | None = None) -> GatewaySettings:
+        values = env or environ
+        router_strategy = values.get("ROUTER_STRATEGY", "load_aware")
+        if router_strategy not in ROUTER_STRATEGIES:
+            raise ValueError("ROUTER_STRATEGY must be load_aware or round_robin")
+
+        return cls(
+            worker_urls=parse_worker_urls(values.get("WORKER_URLS"), values),
+            router_strategy=router_strategy,
+            redis_url=values.get("REDIS_URL"),
+            cache_enabled=parse_bool(values.get("CACHE_ENABLED"), default=True),
+            request_timeout_seconds=parse_timeout(
+                values.get("REQUEST_TIMEOUT_SECONDS"),
+                default=1.0,
+            ),
+        )

pyproject.toml

@@ -0,0 +1,15 @@
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py312"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP"]
+
+[tool.ruff.lint.per-file-ignores]
+"gateway/app/dashboard.py" = ["E501"]
+
+[tool.ruff.format]
+quote-style = "double"

requirements-dev.txt

@@ -0,0 +1 @@
+ruff==0.8.6

scripts/smoke_check.py

@@ -20,7 +20,12 @@ def get_json(url: str, timeout: float = 2.0) -> dict:
 
 def post_json(url: str, payload: dict, timeout: float = 2.0) -> dict:
     data = json.dumps(payload).encode("utf-8")
-    req = request.Request(url, data=data, headers={"content-type": "application/json"}, method="POST")
+    req = request.Request(
+        url,
+        data=data,
+        headers={"content-type": "application/json"},
+        method="POST",
+    )
     with request.urlopen(req, timeout=timeout) as response:
         return json.loads(response.read().decode("utf-8"))
 
@@ -55,12 +60,17 @@ def main() -> None:
     if not workers:
         raise RuntimeError("gateway returned no workers")
 
-    print(json.dumps({
-        "health": health,
-        "prediction": prediction,
-        "metrics": metrics,
-        "workers": workers,
-    }, indent=2))
+    print(
+        json.dumps(
+            {
+                "health": health,
+                "prediction": prediction,
+                "metrics": metrics,
+                "workers": workers,
+            },
+            indent=2,
+        )
+    )
 
 
 if __name__ == "__main__":

tests/test_cache.py

@@ -1,5 +1,5 @@
 from gateway.app.cache import PredictionCache
-from gateway.app.main import parse_worker_hostports, parse_worker_urls
+from gateway.app.settings import parse_worker_hostports, parse_worker_urls
 
 
 def test_memory_cache_round_trip() -> None: