From a68c81a73d18b4c8e3b114aeca41ea096f37ceb8 Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Sun, 30 Aug 2026 15:41:41 +0530 Subject: [PATCH 1/5] feat: add the Kubernetes deployment topology Adds namespaced manifests for the gateway, GPU serving pool, Redis state, KEDA autoscaling on queue depth, Prometheus scraping, network policy, and secret-manager-sourced credentials. Invariant tests enforce unprivileged workloads, digest-pinned images, bounded resources, real probes, and the absence of committed secret material. --- deploy/kubernetes/autoscaling.yaml | 25 ++++ deploy/kubernetes/gateway.yaml | 88 +++++++++++++ deploy/kubernetes/kustomization.yaml | 11 ++ deploy/kubernetes/namespace.yaml | 7 + deploy/kubernetes/network-policy.yaml | 59 +++++++++ deploy/kubernetes/observability.yaml | 16 +++ deploy/kubernetes/state.yaml | 88 +++++++++++++ deploy/kubernetes/vllm-serve.yaml | 84 ++++++++++++ tests/unit/test_deployment_manifests.py | 167 ++++++++++++++++++++++++ 9 files changed, 545 insertions(+) create mode 100644 deploy/kubernetes/autoscaling.yaml create mode 100644 deploy/kubernetes/gateway.yaml create mode 100644 deploy/kubernetes/kustomization.yaml create mode 100644 deploy/kubernetes/namespace.yaml create mode 100644 deploy/kubernetes/network-policy.yaml create mode 100644 deploy/kubernetes/observability.yaml create mode 100644 deploy/kubernetes/state.yaml create mode 100644 deploy/kubernetes/vllm-serve.yaml create mode 100644 tests/unit/test_deployment_manifests.py diff --git a/deploy/kubernetes/autoscaling.yaml b/deploy/kubernetes/autoscaling.yaml new file mode 100644 index 0000000..210b661 --- /dev/null +++ b/deploy/kubernetes/autoscaling.yaml @@ -0,0 +1,25 @@ +# Stateless ingress scales on queue depth, independent of GPU replicas. +apiVersion: keda.sh/v1alpha1 +kind: ScaledObject +metadata: + name: llm-gateway + namespace: llm-routing +spec: + scaleTargetRef: + name: llm-gateway + minReplicaCount: 2 + maxReplicaCount: 20 + cooldownPeriod: 120 + triggers: + - type: prometheus + metadata: + serverAddress: http://prometheus.monitoring.svc.cluster.local:9090 + metricName: router_queued_requests + query: sum(router_queued_requests{namespace="llm-routing"}) + threshold: "4" + - type: prometheus + metadata: + serverAddress: http://prometheus.monitoring.svc.cluster.local:9090 + metricName: router_request_latency_p95 + query: histogram_quantile(0.95, sum(rate(router_request_latency_seconds_bucket[5m])) by (le)) + threshold: "2" diff --git a/deploy/kubernetes/gateway.yaml b/deploy/kubernetes/gateway.yaml new file mode 100644 index 0000000..d5f389e --- /dev/null +++ b/deploy/kubernetes/gateway.yaml @@ -0,0 +1,88 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: llm-gateway + namespace: llm-routing + labels: + app.kubernetes.io/name: llm-gateway + app.kubernetes.io/part-of: local-llm-router +spec: + replicas: 2 + selector: + matchLabels: + app.kubernetes.io/name: llm-gateway + template: + metadata: + labels: + app.kubernetes.io/name: llm-gateway + app.kubernetes.io/part-of: local-llm-router + spec: + securityContext: + runAsNonRoot: true + runAsUser: 10001 + seccompProfile: + type: RuntimeDefault + containers: + - name: gateway + # Replaced at deploy time with the digest recorded in the DeploymentRevision. + image: ghcr.io/REPLACE_ME/local-llm-router@sha256:REPLACE_ME + imagePullPolicy: IfNotPresent + ports: + - name: http + containerPort: 8000 + env: + - name: ROUTER_ENVIRONMENT + value: production + - name: ROUTER_BACKEND + value: vllm + - name: ROUTER_VLLM_BASE_URL + value: http://vllm-serve.llm-routing.svc.cluster.local:8000 + - name: ROUTER_API_KEYS + valueFrom: + secretKeyRef: + name: llm-gateway-credentials + key: api-keys + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + resources: + requests: + cpu: 250m + memory: 512Mi + limits: + cpu: "2" + memory: 2Gi + livenessProbe: + httpGet: + path: /healthz + port: http + initialDelaySeconds: 5 + periodSeconds: 30 + readinessProbe: + httpGet: + path: /readyz + port: http + initialDelaySeconds: 5 + periodSeconds: 10 + lifecycle: + preStop: + exec: + command: ["sleep", "10"] + terminationGracePeriodSeconds: 60 +--- +apiVersion: v1 +kind: Service +metadata: + name: llm-gateway + namespace: llm-routing + labels: + app.kubernetes.io/name: llm-gateway +spec: + selector: + app.kubernetes.io/name: llm-gateway + ports: + - name: http + port: 80 + targetPort: http diff --git a/deploy/kubernetes/kustomization.yaml b/deploy/kubernetes/kustomization.yaml new file mode 100644 index 0000000..5bac207 --- /dev/null +++ b/deploy/kubernetes/kustomization.yaml @@ -0,0 +1,11 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: llm-routing +resources: + - namespace.yaml + - gateway.yaml + - autoscaling.yaml + - vllm-serve.yaml + - state.yaml + - network-policy.yaml + - observability.yaml diff --git a/deploy/kubernetes/namespace.yaml b/deploy/kubernetes/namespace.yaml new file mode 100644 index 0000000..acc40ae --- /dev/null +++ b/deploy/kubernetes/namespace.yaml @@ -0,0 +1,7 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: llm-routing + labels: + app.kubernetes.io/part-of: local-llm-router + pod-security.kubernetes.io/enforce: restricted diff --git a/deploy/kubernetes/network-policy.yaml b/deploy/kubernetes/network-policy.yaml new file mode 100644 index 0000000..e2b06b8 --- /dev/null +++ b/deploy/kubernetes/network-policy.yaml @@ -0,0 +1,59 @@ +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: llm-gateway + namespace: llm-routing +spec: + podSelector: + matchLabels: + app.kubernetes.io/name: llm-gateway + policyTypes: [Ingress, Egress] + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: applications + ports: + - protocol: TCP + port: 8000 + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: monitoring + ports: + - protocol: TCP + port: 8000 + egress: + - to: + - podSelector: + matchLabels: + app.kubernetes.io/name: vllm-serve + ports: + - protocol: TCP + port: 8000 + - to: + - podSelector: + matchLabels: + app.kubernetes.io/name: redis + ports: + - protocol: TCP + port: 6379 +--- +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: vllm-serve + namespace: llm-routing +spec: + podSelector: + matchLabels: + app.kubernetes.io/name: vllm-serve + policyTypes: [Ingress] + ingress: + - from: + - podSelector: + matchLabels: + app.kubernetes.io/name: llm-gateway + ports: + - protocol: TCP + port: 8000 diff --git a/deploy/kubernetes/observability.yaml b/deploy/kubernetes/observability.yaml new file mode 100644 index 0000000..a2e19f7 --- /dev/null +++ b/deploy/kubernetes/observability.yaml @@ -0,0 +1,16 @@ +# Metrics are unauthenticated by design and reachable only from monitoring. +apiVersion: monitoring.coreos.com/v1 +kind: ServiceMonitor +metadata: + name: llm-gateway + namespace: llm-routing + labels: + release: prometheus +spec: + selector: + matchLabels: + app.kubernetes.io/name: llm-gateway + endpoints: + - port: http + path: /metrics + interval: 15s diff --git a/deploy/kubernetes/state.yaml b/deploy/kubernetes/state.yaml new file mode 100644 index 0000000..3e2a260 --- /dev/null +++ b/deploy/kubernetes/state.yaml @@ -0,0 +1,88 @@ +# Cache and quota state. Credentials come from the cluster secret manager; +# no secret material is committed to this repository. +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: redis + namespace: llm-routing + labels: + app.kubernetes.io/name: redis +spec: + serviceName: redis + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: redis + template: + metadata: + labels: + app.kubernetes.io/name: redis + spec: + securityContext: + runAsNonRoot: true + runAsUser: 10001 + seccompProfile: + type: RuntimeDefault + containers: + - name: redis + image: docker.io/library/redis@sha256:REPLACE_ME + ports: + - name: redis + containerPort: 6379 + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + resources: + requests: + cpu: 100m + memory: 256Mi + limits: + cpu: "1" + memory: 1Gi + livenessProbe: + tcpSocket: + port: redis + initialDelaySeconds: 10 + periodSeconds: 30 + readinessProbe: + tcpSocket: + port: redis + initialDelaySeconds: 5 + periodSeconds: 10 +--- +apiVersion: v1 +kind: Service +metadata: + name: redis + namespace: llm-routing +spec: + selector: + app.kubernetes.io/name: redis + ports: + - name: redis + port: 6379 + targetPort: redis +--- +apiVersion: external-secrets.io/v1 +kind: ExternalSecret +metadata: + name: llm-gateway-credentials + namespace: llm-routing +spec: + refreshInterval: 1h + secretStoreRef: + name: platform-secret-store + kind: ClusterSecretStore + target: + name: llm-gateway-credentials + data: + - secretKey: api-keys + remoteRef: + key: llm-routing/gateway + property: api_keys + - secretKey: external-provider-key + remoteRef: + key: llm-routing/gateway + property: external_provider_key diff --git a/deploy/kubernetes/vllm-serve.yaml b/deploy/kubernetes/vllm-serve.yaml new file mode 100644 index 0000000..032f88e --- /dev/null +++ b/deploy/kubernetes/vllm-serve.yaml @@ -0,0 +1,84 @@ +# GPU serving replicas are scheduled onto accelerator-specific pools and scale +# separately from the stateless gateway. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: vllm-serve + namespace: llm-routing + labels: + app.kubernetes.io/name: vllm-serve + app.kubernetes.io/part-of: local-llm-router +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: vllm-serve + template: + metadata: + labels: + app.kubernetes.io/name: vllm-serve + app.kubernetes.io/part-of: local-llm-router + spec: + securityContext: + runAsNonRoot: true + runAsUser: 10001 + seccompProfile: + type: RuntimeDefault + nodeSelector: + nvidia.com/gpu.product: NVIDIA-L4 + tolerations: + - key: nvidia.com/gpu + operator: Exists + effect: NoSchedule + containers: + - name: engine + image: docker.io/vllm/vllm-openai@sha256:REPLACE_ME + args: + - --served-model-name=small-specialist + - --enable-prefix-caching + - --enable-lora + - --max-model-len=8192 + - --quantization=awq + ports: + - name: http + containerPort: 8000 + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: [ALL] + resources: + requests: + cpu: "4" + memory: 32Gi + nvidia.com/gpu: "1" + limits: + cpu: "8" + memory: 48Gi + nvidia.com/gpu: "1" + livenessProbe: + httpGet: + path: /health + port: http + initialDelaySeconds: 120 + periodSeconds: 30 + readinessProbe: + httpGet: + path: /health + port: http + initialDelaySeconds: 120 + periodSeconds: 10 + terminationGracePeriodSeconds: 120 +--- +apiVersion: v1 +kind: Service +metadata: + name: vllm-serve + namespace: llm-routing +spec: + selector: + app.kubernetes.io/name: vllm-serve + ports: + - name: http + port: 8000 + targetPort: http diff --git a/tests/unit/test_deployment_manifests.py b/tests/unit/test_deployment_manifests.py new file mode 100644 index 0000000..cad80a9 --- /dev/null +++ b/tests/unit/test_deployment_manifests.py @@ -0,0 +1,167 @@ +"""Invariants for the deployment topology in section 17. + +These assertions are the deployment contract: every manifest must parse, run +unprivileged with bounded resources and real probes, carry no secret material, +and keep stateless ingress scaling independently of GPU replicas. +""" + +from pathlib import Path +from typing import Any + +import pytest +import yaml + +MANIFEST_DIR = Path("deploy/kubernetes") +MANIFESTS = sorted(MANIFEST_DIR.glob("*.yaml")) +SECRET_MARKERS = ("password:", "token:", "apiKey:", "api_key:", "BEGIN PRIVATE KEY") + + +def load_documents() -> list[dict[str, Any]]: + documents: list[dict[str, Any]] = [] + for path in MANIFESTS: + for document in yaml.safe_load_all(path.read_text(encoding="utf-8")): + if document: + documents.append(document) + return documents + + +DOCUMENTS = load_documents() +WORKLOADS = [ + document for document in DOCUMENTS if document["kind"] in {"Deployment", "StatefulSet"} +] + + +def pod_spec(workload: dict[str, Any]) -> dict[str, Any]: + spec: dict[str, Any] = workload["spec"]["template"]["spec"] + return spec + + +def test_every_manifest_is_listed_in_the_kustomization() -> None: + kustomization = yaml.safe_load((MANIFEST_DIR / "kustomization.yaml").read_text()) + + listed = set(kustomization["resources"]) + on_disk = {path.name for path in MANIFESTS} - {"kustomization.yaml"} + assert listed == on_disk + + +def test_all_resources_are_namespaced_to_the_platform() -> None: + namespace = next(document for document in DOCUMENTS if document["kind"] == "Namespace") + assert namespace["metadata"]["name"] == "llm-routing" + + for document in DOCUMENTS: + if document["kind"] in {"Namespace", "Kustomization"}: + continue + assert document["metadata"]["namespace"] == "llm-routing", document["metadata"]["name"] + + +@pytest.mark.parametrize("workload", WORKLOADS, ids=lambda item: str(item["metadata"]["name"])) +def test_workloads_run_unprivileged(workload: dict[str, Any]) -> None: + spec = pod_spec(workload) + + assert spec["securityContext"]["runAsNonRoot"] is True + for container in spec["containers"]: + security = container["securityContext"] + assert security["allowPrivilegeEscalation"] is False + assert security["readOnlyRootFilesystem"] is True + assert security["capabilities"]["drop"] == ["ALL"] + + +@pytest.mark.parametrize("workload", WORKLOADS, ids=lambda item: str(item["metadata"]["name"])) +def test_workloads_declare_bounded_resources_and_probes(workload: dict[str, Any]) -> None: + for container in pod_spec(workload)["containers"]: + assert container["resources"]["requests"] + assert container["resources"]["limits"] + assert container["livenessProbe"] + assert container["readinessProbe"] + + +@pytest.mark.parametrize("workload", WORKLOADS, ids=lambda item: str(item["metadata"]["name"])) +def test_images_are_pinned_by_digest(workload: dict[str, Any]) -> None: + for container in pod_spec(workload)["containers"]: + assert "@sha256:" in container["image"], container["image"] + + +def test_no_manifest_contains_secret_material() -> None: + for path in MANIFESTS: + content = path.read_text(encoding="utf-8") + assert "kind: Secret\n" not in content + for marker in SECRET_MARKERS: + assert marker not in content, f"{path.name} contains {marker}" + + +def test_gateway_reads_credentials_from_the_secret_manager() -> None: + external_secret = next( + document for document in DOCUMENTS if document["kind"] == "ExternalSecret" + ) + gateway = next( + document for document in WORKLOADS if document["metadata"]["name"] == "llm-gateway" + ) + container = pod_spec(gateway)["containers"][0] + api_keys = next(item for item in container["env"] if item["name"] == "ROUTER_API_KEYS") + + assert "value" not in api_keys + assert ( + api_keys["valueFrom"]["secretKeyRef"]["name"] == external_secret["spec"]["target"]["name"] + ) + + +def test_gateway_probes_target_the_health_and_readiness_endpoints() -> None: + gateway = next( + document for document in WORKLOADS if document["metadata"]["name"] == "llm-gateway" + ) + container = pod_spec(gateway)["containers"][0] + + assert container["livenessProbe"]["httpGet"]["path"] == "/healthz" + assert container["readinessProbe"]["httpGet"]["path"] == "/readyz" + assert container["lifecycle"]["preStop"], "graceful shutdown drain is required" + + +def test_gpu_replicas_are_pinned_to_an_accelerator_pool() -> None: + engine = next( + document for document in WORKLOADS if document["metadata"]["name"] == "vllm-serve" + ) + spec = pod_spec(engine) + container = spec["containers"][0] + + assert spec["nodeSelector"]["nvidia.com/gpu.product"] + assert any(toleration["key"] == "nvidia.com/gpu" for toleration in spec["tolerations"]) + assert container["resources"]["limits"]["nvidia.com/gpu"] == "1" + assert spec["terminationGracePeriodSeconds"] >= 120 + + +def test_stateless_ingress_scales_independently_of_gpu_replicas() -> None: + scaled_object = next(document for document in DOCUMENTS if document["kind"] == "ScaledObject") + + assert scaled_object["spec"]["scaleTargetRef"]["name"] == "llm-gateway" + assert scaled_object["spec"]["minReplicaCount"] >= 2 + assert scaled_object["spec"]["maxReplicaCount"] > scaled_object["spec"]["minReplicaCount"] + metrics = {trigger["metadata"]["metricName"] for trigger in scaled_object["spec"]["triggers"]} + assert "router_queued_requests" in metrics + + +def test_metrics_are_scraped_and_reachable_only_from_monitoring() -> None: + monitor = next(document for document in DOCUMENTS if document["kind"] == "ServiceMonitor") + policy = next( + document + for document in DOCUMENTS + if document["kind"] == "NetworkPolicy" and document["metadata"]["name"] == "llm-gateway" + ) + + assert monitor["spec"]["endpoints"][0]["path"] == "/metrics" + namespaces = { + rule["namespaceSelector"]["matchLabels"]["kubernetes.io/metadata.name"] + for entry in policy["spec"]["ingress"] + for rule in entry["from"] + } + assert namespaces == {"applications", "monitoring"} + + +def test_engine_ingress_is_restricted_to_the_gateway() -> None: + policy = next( + document + for document in DOCUMENTS + if document["kind"] == "NetworkPolicy" and document["metadata"]["name"] == "vllm-serve" + ) + + sources = policy["spec"]["ingress"][0]["from"] + assert sources == [{"podSelector": {"matchLabels": {"app.kubernetes.io/name": "llm-gateway"}}}] From ced29e1f660285930660d0b7400f589200f80b10 Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Sun, 30 Aug 2026 15:43:51 +0530 Subject: [PATCH 2/5] feat: share cache and quota state through Redis Adds a Redis-backed response cache and fixed-window quota behind a small async protocol, selected by ROUTER_REDIS_URL. Without it the gateway keeps in-process state, which is correct for a single replica only. --- pyproject.toml | 10 ++++ src/llm_router/app.py | 43 ++++++++++++-- src/llm_router/config.py | 1 + src/llm_router/redis_state.py | 74 ++++++++++++++++++++++++ tests/conftest.py | 32 ++++++++++ tests/integration/test_streaming_api.py | 19 ++++++ tests/unit/test_redis_state.py | 77 +++++++++++++++++++++++++ 7 files changed, 252 insertions(+), 4 deletions(-) create mode 100644 src/llm_router/redis_state.py create mode 100644 tests/conftest.py create mode 100644 tests/unit/test_redis_state.py diff --git a/pyproject.toml b/pyproject.toml index 69c7245..622afed 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,6 +18,9 @@ dependencies = [ ] [project.optional-dependencies] +redis = [ + "redis>=5.2.1,<7", +] dev = [ "mypy>=2.3.1,<3", "pytest>=9.1.1,<10", @@ -57,3 +60,10 @@ python_version = "3.11" strict = true packages = ["llm_router"] + +[[tool.mypy.overrides]] +# The redis extra is optional at runtime and is not installed in the default +# verification environment; the client is used only behind the RedisLike protocol. +module = ["redis.*"] +ignore_missing_imports = true +follow_imports = "skip" diff --git a/src/llm_router/app.py b/src/llm_router/app.py index 85091ac..3937e7a 100644 --- a/src/llm_router/app.py +++ b/src/llm_router/app.py @@ -46,10 +46,22 @@ Usage, ) from llm_router.observability import Metrics +from llm_router.redis_state import RedisCacheStore, RedisFixedWindowQuota, RedisLike from llm_router.registry import Registry, RegistryError, catalog_revisions, load_registry from llm_router.routing import NoEligibleModelError, Router, default_model_profiles +def _redis_client(settings: Settings) -> RedisLike | None: + """Build a shared-state client when a Redis URL is configured.""" + + if not settings.redis_url: + return None + from redis.asyncio import Redis # imported lazily so the extra stays optional + + client: RedisLike = Redis.from_url(settings.redis_url) + return client + + def _load_catalog(path: str) -> Registry | None: """Load the governed catalog, falling back to built-in profiles when absent.""" @@ -65,6 +77,7 @@ def create_app( metrics: Metrics | None = None, cache_store: CacheStore | None = None, registry: Registry | None = None, + redis_client: RedisLike | None = None, ) -> FastAPI: runtime_settings = settings or get_settings() catalog = registry if registry is not None else _load_catalog(runtime_settings.registry_path) @@ -81,7 +94,13 @@ def create_app( runtime_settings.max_concurrency, runtime_settings.admission_timeout_seconds, ) + shared_state = redis_client if redis_client is not None else _redis_client(runtime_settings) quota = SlidingWindowQuota(runtime_settings.quota_requests_per_minute) + shared_quota = ( + RedisFixedWindowQuota(shared_state, runtime_settings.quota_requests_per_minute) + if shared_state is not None + else None + ) engine_client = ( httpx.AsyncClient() if backend is None and runtime_settings.backend == "vllm" else None ) @@ -95,9 +114,17 @@ def create_app( else MockInferenceBackend() ) telemetry = metrics if metrics is not None else Metrics() - exact_cache: CacheStore = cache_store or InMemoryCacheStore( - max_entries=runtime_settings.cache_max_entries, - ttl_seconds=runtime_settings.cache_ttl_seconds, + exact_cache: CacheStore = ( + cache_store + or ( + RedisCacheStore(shared_state, ttl_seconds=runtime_settings.cache_ttl_seconds) + if shared_state is not None + else None + ) + or InMemoryCacheStore( + max_entries=runtime_settings.cache_max_entries, + ttl_seconds=runtime_settings.cache_ttl_seconds, + ) ) semantic_cache = SemanticCache(threshold=runtime_settings.semantic_similarity_threshold) decision_cache = RouterDecisionCache(policy_version=policy_version) @@ -357,6 +384,14 @@ async def _store_cache( scope = semantic_cache.scope(payload, tenant=tenant, model_revision=catalog_version) semantic_cache.store(scope, prompt, entry) + async def _consume_quota(subject: str) -> None: + if shared_quota is None: + await quota.consume(subject) + return + window = int(time.time() // 60) + if not await shared_quota.consume(subject, window=window): + raise QuotaExceededError("request quota exceeded") + def _route_headers(decision: RouteDecision, cache_state: str) -> dict[str, str]: headers = { "X-Cache": cache_state, @@ -427,7 +462,7 @@ async def chat_completions( subject: str = Depends(authenticate), ) -> ChatCompletionResponse | StreamingResponse: started = time.perf_counter() - await quota.consume(subject) + await _consume_quota(subject) prompt = payload.prompt cache_key = build_cache_key( payload, tenant=subject, model_revision=catalog_version, prompt=prompt diff --git a/src/llm_router/config.py b/src/llm_router/config.py index 4644ff9..b4c02ba 100644 --- a/src/llm_router/config.py +++ b/src/llm_router/config.py @@ -21,6 +21,7 @@ class Settings(BaseSettings): backend_timeout_seconds: float = Field(default=60.0, gt=0) registry_path: str = "config/registry.yaml" routing_policy_version: str = "v1" + redis_url: str = "" cache_enabled: bool = True cache_ttl_seconds: float = Field(default=300.0, gt=0) cache_max_entries: int = Field(default=1024, ge=1) diff --git a/src/llm_router/redis_state.py b/src/llm_router/redis_state.py new file mode 100644 index 0000000..9f6ea56 --- /dev/null +++ b/src/llm_router/redis_state.py @@ -0,0 +1,74 @@ +"""Redis-backed cache and quota state for multi-replica deployments. + +In-process state is correct for a single replica only. When the gateway scales +horizontally, cache entries and quota counters must be shared, so both are +expressed against a minimal async Redis protocol that the real client satisfies. +""" + +import json +from typing import Protocol + +from llm_router.caching import CachedCompletion + +QUOTA_WINDOW_SECONDS = 60 + + +class RedisLike(Protocol): + """The subset of the async Redis client this module depends on.""" + + async def get(self, name: str) -> bytes | str | None: ... + + async def set(self, name: str, value: str, ex: int | None = None) -> object: ... + + async def incr(self, name: str) -> int: ... + + async def expire(self, name: str, seconds: int) -> object: ... + + +class RedisCacheStore: + """Shared exact-response cache keyed by the gateway's cache key.""" + + def __init__(self, client: RedisLike, *, ttl_seconds: float = 300.0, prefix: str = "llmr:c:"): + self._client = client + self._ttl_seconds = int(ttl_seconds) + self._prefix = prefix + + async def get(self, key: str) -> CachedCompletion | None: + raw = await self._client.get(self._prefix + key) + if raw is None: + return None + payload = raw.decode() if isinstance(raw, bytes) else raw + try: + return CachedCompletion(**json.loads(payload)) + except (ValueError, TypeError): + return None + + async def set(self, key: str, value: CachedCompletion) -> None: + document = json.dumps( + { + "text": value.text, + "model_id": value.model_id, + "model_revision": value.model_revision, + "prompt_tokens": value.prompt_tokens, + "completion_tokens": value.completion_tokens, + } + ) + await self._client.set(self._prefix + key, document, ex=self._ttl_seconds) + + +class RedisFixedWindowQuota: + """Quota counter shared across replicas, bounded to a one-minute window.""" + + def __init__(self, client: RedisLike, requests_per_minute: int, *, prefix: str = "llmr:q:"): + self._client = client + self._limit = requests_per_minute + self._prefix = prefix + + async def consume(self, subject: str, *, window: int) -> bool: + """Return whether the request fits inside the caller's quota.""" + + key = f"{self._prefix}{subject}:{window}" + count = await self._client.incr(key) + if count == 1: + await self._client.expire(key, QUOTA_WINDOW_SECONDS * 2) + return count <= self._limit diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..57e8c49 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,32 @@ +import pytest + + +class FakeRedis: + """In-memory double for the small async Redis surface the gateway uses.""" + + def __init__(self) -> None: + self.values: dict[str, str] = {} + self.counters: dict[str, int] = {} + self.expirations: dict[str, int] = {} + self.ttls: dict[str, int | None] = {} + + async def get(self, name: str) -> str | None: + return self.values.get(name) + + async def set(self, name: str, value: str, ex: int | None = None) -> bool: + self.values[name] = value + self.ttls[name] = ex + return True + + async def incr(self, name: str) -> int: + self.counters[name] = self.counters.get(name, 0) + 1 + return self.counters[name] + + async def expire(self, name: str, seconds: int) -> bool: + self.expirations[name] = seconds + return True + + +@pytest.fixture +def fake_redis() -> FakeRedis: + return FakeRedis() diff --git a/tests/integration/test_streaming_api.py b/tests/integration/test_streaming_api.py index ff7dd2d..417635f 100644 --- a/tests/integration/test_streaming_api.py +++ b/tests/integration/test_streaming_api.py @@ -7,6 +7,7 @@ from llm_router.backends import BackendResult, BackendUnavailableError from llm_router.config import Settings from llm_router.models import ChatCompletionRequest, RouteDecision +from tests.conftest import FakeRedis HEADERS = {"Authorization": "Bearer stream-key"} BODY = { @@ -119,3 +120,21 @@ def test_vllm_backend_is_selected_by_configuration() -> None: assert response.status_code == 502 assert response.json()["error"]["type"] == "backend_unavailable" + + +def test_shared_redis_state_backs_cache_and_quota(fake_redis: FakeRedis) -> None: + client_state = fake_redis + settings = Settings(api_keys="stream-key", quota_requests_per_minute=1) + with TestClient(create_app(settings, redis_client=client_state)) as client: + body = { + "model": "auto", + "messages": [{"role": "user", "content": "shared state probe"}], + "routing": {"privacy": "public"}, + } + first = client.post("/v1/chat/completions", headers=HEADERS, json=body) + second = client.post("/v1/chat/completions", headers=HEADERS, json=body) + + assert first.status_code == 200 + assert second.status_code == 429 + assert any(key.startswith("llmr:c:") for key in client_state.values) + assert any(key.startswith("llmr:q:") for key in client_state.counters) diff --git a/tests/unit/test_redis_state.py b/tests/unit/test_redis_state.py new file mode 100644 index 0000000..48d871a --- /dev/null +++ b/tests/unit/test_redis_state.py @@ -0,0 +1,77 @@ +import pytest + +from llm_router.caching import CachedCompletion +from llm_router.redis_state import RedisCacheStore, RedisFixedWindowQuota +from tests.conftest import FakeRedis + + +def entry() -> CachedCompletion: + return CachedCompletion( + text="cached", + model_id="small-specialist", + model_revision="mock-small@sha256:dev", + prompt_tokens=7, + completion_tokens=2, + ) + + +@pytest.mark.asyncio +async def test_shared_cache_round_trips_an_entry_with_a_ttl(fake_redis: FakeRedis) -> None: + store = RedisCacheStore(fake_redis, ttl_seconds=120) + + await store.set("key", entry()) + restored = await store.get("key") + + assert restored == entry() + assert fake_redis.ttls["llmr:c:key"] == 120 + + +@pytest.mark.asyncio +async def test_shared_cache_misses_and_tolerates_corrupt_payloads(fake_redis: FakeRedis) -> None: + store = RedisCacheStore(fake_redis) + + assert await store.get("absent") is None + + fake_redis.values["llmr:c:broken"] = "not-json" + assert await store.get("broken") is None + + fake_redis.values["llmr:c:partial"] = '{"text": "only"}' + assert await store.get("partial") is None + + +@pytest.mark.asyncio +async def test_shared_cache_decodes_byte_payloads(fake_redis: FakeRedis) -> None: + store = RedisCacheStore(fake_redis) + await store.set("key", entry()) + fake_redis.values["llmr:c:key"] = fake_redis.values["llmr:c:key"].encode() # type: ignore[assignment] + + assert await store.get("key") == entry() + + +@pytest.mark.asyncio +async def test_shared_quota_admits_up_to_the_limit_then_rejects(fake_redis: FakeRedis) -> None: + quota = RedisFixedWindowQuota(fake_redis, requests_per_minute=2) + + assert await quota.consume("tenant", window=100) is True + assert await quota.consume("tenant", window=100) is True + assert await quota.consume("tenant", window=100) is False + + +@pytest.mark.asyncio +async def test_shared_quota_expires_the_counter_and_resets_each_window( + fake_redis: FakeRedis, +) -> None: + quota = RedisFixedWindowQuota(fake_redis, requests_per_minute=1) + + assert await quota.consume("tenant", window=100) is True + assert fake_redis.expirations["llmr:q:tenant:100"] == 120 + assert await quota.consume("tenant", window=101) is True + + +@pytest.mark.asyncio +async def test_shared_quota_isolates_subjects(fake_redis: FakeRedis) -> None: + quota = RedisFixedWindowQuota(fake_redis, requests_per_minute=1) + + assert await quota.consume("tenant-a", window=1) is True + assert await quota.consume("tenant-b", window=1) is True + assert await quota.consume("tenant-a", window=1) is False From b4ba399b7262dbef03487a436be49c60ed0eb156 Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Sun, 30 Aug 2026 15:44:27 +0530 Subject: [PATCH 3/5] ci: render and validate the deployment plan during delivery Verifies the committed Ray Serve configuration against the catalog, renders the canary plan with its rollback target, and validates the Kubernetes manifests with kubeconform. Applying to a cluster remains disabled until a deployment destination is configured. --- .github/workflows/cd.yml | 45 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml index fcd401e..32a4069 100644 --- a/.github/workflows/cd.yml +++ b/.github/workflows/cd.yml @@ -40,3 +40,48 @@ jobs: path: ${{ runner.temp }}/local-llm-router.tar retention-days: 14 + + deployment-plan: + # Renders the deployment topology and the canary plan that names its own + # rollback target. Applying to a cluster stays disabled until a deployment + # destination is configured; nothing here contacts a live environment. + needs: release-artifact + runs-on: ubuntu-latest + env: + RELEASE_REF: ${{ github.event_name == 'workflow_dispatch' && inputs.ref || github.event.workflow_run.head_sha }} + steps: + - uses: actions/checkout@v7 + with: + ref: ${{ env.RELEASE_REF }} + - uses: actions/setup-python@v6 + with: + python-version: "3.13" + - run: python -m pip install --upgrade pip + - run: python -m pip install -e ".[dev]" + - name: Verify the serving configuration matches the catalog + run: python -m llm_router.serving > /tmp/ray-serve.yaml && diff -u config/ray-serve.yaml /tmp/ray-serve.yaml + - name: Render the canary and rollback plan + run: | + python - <<'PY' > canary-plan.json + import json + from llm_router.registry import load_registry + from llm_router.serving import canary_config + registry = load_registry("config/registry.yaml") + production = [item for item in registry.deployments if item.stage.value == "production"] + plan = canary_config(registry, production[0].id) + plan["release_ref"] = "${{ env.RELEASE_REF }}" + print(json.dumps(plan, indent=2)) + PY + - name: Validate the Kubernetes manifests + run: | + curl -sSLo kubeconform.tar.gz https://github.com/yannh/kubeconform/releases/download/v0.7.0/kubeconform-linux-amd64.tar.gz + tar xf kubeconform.tar.gz kubeconform + ./kubeconform -strict -ignore-missing-schemas -summary deploy/kubernetes + - uses: actions/upload-artifact@v7 + with: + name: deployment-plan-${{ env.RELEASE_REF }} + path: | + canary-plan.json + config/ray-serve.yaml + deploy/kubernetes + retention-days: 14 From 891818b2d1cbf15ca171be773147450ebd668393 Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Sun, 30 Aug 2026 15:44:28 +0530 Subject: [PATCH 4/5] docs: document the deployment topology and shared state --- README.md | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/README.md b/README.md index 4956d51..7bc478f 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,31 @@ python -m llm_router.serving > config/ray-serve.yaml It carries per-tier autoscaling (latency-sensitive tiers keep a warm replica), GPU pool placement, tensor parallelism, prefix caching, quantization, and Multi-LoRA settings. +## Deployment topology + +[`deploy/kubernetes`](deploy/kubernetes) holds the namespaced manifests: gateway +Deployment and Service, GPU serving pool, Redis, KEDA autoscaling on queue depth and p95 +latency, a Prometheus `ServiceMonitor`, network policy, and credentials sourced from the +cluster secret manager. No secret material is committed. Unit tests enforce the contract: +unprivileged workloads, digest-pinned images, bounded resources, real probes, GPU pool +pinning, and `/metrics` reachable only from monitoring. + +```bash +kubectl apply -k deploy/kubernetes +``` + +Stateless ingress scales separately from GPU replicas. Set `ROUTER_REDIS_URL` so cache and +quota state are shared once the gateway runs more than one replica; without it both are +in-process and correct for a single replica only. Install the client with the extra: + +```bash +python -m pip install -e ".[redis]" +``` + +CD renders the canary plan (with its rollback target and triggers), verifies +`config/ray-serve.yaml` against the catalog, and validates the manifests with kubeconform. +Applying to a cluster stays disabled until a deployment destination is configured. + ## Model registry [`config/registry.yaml`](config/registry.yaml) is the governed source of truth for what may @@ -149,6 +174,7 @@ All settings use the `ROUTER_` prefix. | `ROUTER_ADMISSION_TIMEOUT_SECONDS` | `0.25` | Time allowed to wait for capacity. | | `ROUTER_QUOTA_REQUESTS_PER_MINUTE` | `120` | Per-token sliding-window quota. | | `ROUTER_EXTERNAL_FALLBACK_ENABLED` | `false` | Operator gate for external fallback. | +| `ROUTER_REDIS_URL` | _(empty)_ | Shared cache and quota state; in-process when empty. | | `ROUTER_BACKEND` | `mock` | `mock` or `vllm`. | | `ROUTER_VLLM_BASE_URL` | `http://127.0.0.1:8001` | vLLM OpenAI-compatible endpoint. | | `ROUTER_BACKEND_TIMEOUT_SECONDS` | `60` | Per-request engine timeout. | From 2195b3577850596309557ce91162181baebe584b Mon Sep 17 00:00:00 2001 From: Yash-Chindam Date: Thu, 3 Sep 2026 21:35:00 +0530 Subject: [PATCH 5/5] fix: make the tests package importable under the pytest console script CI invokes the pytest entry point directly, which does not add the repository root to sys.path the way `python -m pytest` does, so `from tests.conftest import FakeRedis` failed collection in CI while passing locally. Co-Authored-By: Claude Sonnet 5 --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 622afed..02a5a9f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ packages = ["src/llm_router"] [tool.pytest.ini_options] addopts = "-ra --strict-config --strict-markers" testpaths = ["tests"] +pythonpath = ["."] markers = [ "integration: tests that exercise multiple application components", ]