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 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. | 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/pyproject.toml b/pyproject.toml index 69c7245..02a5a9f 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", @@ -33,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", ] @@ -57,3 +61,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_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"}}}] 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