diff --git a/api/v1alpha1/cachebackend_types.go b/api/v1alpha1/cachebackend_types.go index cb47dd4e..3b5a713d 100644 --- a/api/v1alpha1/cachebackend_types.go +++ b/api/v1alpha1/cachebackend_types.go @@ -643,9 +643,10 @@ type CacheBackendIntegrationSpec struct { // default — the field carries no meaningful "wait forever" or "fail // immediately" semantics. // - // The first SGLangHiCache implementation publishes no Ready condition, so - // this field is inert for that engine-local backend until its separate - // readiness contract is implemented. + // SGLangHiCache readiness is derived from engine Pod injection receipts, + // convergence with the current adapter configuration, and Kubernetes Pod + // readiness rather than KV events, so this field is inert for that + // engine-local backend. // // The value is a Go duration string (e.g. "90s", "5m", "1h"). The CRD // schema types it as a string; a malformed value is rejected when diff --git a/config/crd/bases/inferencecache.io_cachebackends.yaml b/config/crd/bases/inferencecache.io_cachebackends.yaml index 89ced3b0..9cd6d043 100644 --- a/config/crd/bases/inferencecache.io_cachebackends.yaml +++ b/config/crd/bases/inferencecache.io_cachebackends.yaml @@ -574,9 +574,10 @@ spec: default — the field carries no meaningful "wait forever" or "fail immediately" semantics. - The first SGLangHiCache implementation publishes no Ready condition, so - this field is inert for that engine-local backend until its separate - readiness contract is implemented. + SGLangHiCache readiness is derived from engine Pod injection receipts, + convergence with the current adapter configuration, and Kubernetes Pod + readiness rather than KV events, so this field is inert for that + engine-local backend. The value is a Go duration string (e.g. "90s", "5m", "1h"). The CRD schema types it as a string; a malformed value is rejected when diff --git a/config/samples/README.md b/config/samples/README.md index 350dae5a..508352f8 100644 --- a/config/samples/README.md +++ b/config/samples/README.md @@ -54,9 +54,12 @@ they go `Ready` as soon as admission accepts the endpoint. See the [quickstart](../../docs/quickstart.md). `SGLangHiCache` is endpoint-free and has no endpoint publication race. Its -first implementation intentionally publishes no `Ready` condition; the -matching Pod's injection annotations are the available wiring signal until the -separate HiCache readiness contract ships. +`Ready` condition is derived from matching engine Pods: all participating Pods +must carry the current CacheBackend name, UID, and generation receipt, already +contain the configuration that the current HiCache adapter would inject, and +be Kubernetes Ready. This confirms that the current configuration reached +serving Pods; it does not claim a successful HiCache host-tier write/read round +trip. `recipe-multi-tenant.yaml` spans two namespaces, so it carries a `# verify-samples: skip` marker — server-side dry-run can't create the diff --git a/docs/concepts/cachebackend-engine-binding.md b/docs/concepts/cachebackend-engine-binding.md index b2de6d87..edbb489a 100644 --- a/docs/concepts/cachebackend-engine-binding.md +++ b/docs/concepts/cachebackend-engine-binding.md @@ -48,14 +48,19 @@ The match is evaluated **once at pod CREATE** by the mutating webhook. The wirin > **Native SGLang HiCache exception.** `type: SGLangHiCache` is engine-local: > the controller creates no backend workload or endpoint, and the webhook does > not wait for `status.endpoint`. It injects the typed `spec.hiCache` launch -> flags directly into matching SGLang Pods. The LMCache lifecycle below remains -> the endpoint-bearing path. +> flags directly into matching SGLang Pods. Its readiness is engine-side: +> every participating Pod must carry the current CacheBackend name, UID, and +> generation receipt, already contain the current adapter configuration +> including `spec.integration.engineOverrides`, and be Kubernetes Ready. This +> reports configuration rollout and serving availability, not a proven HiCache +> host-tier read/write. The LMCache lifecycle below remains the endpoint-bearing +> path. ## Lifecycle 1. **Apply the CacheBackend.** `kubectl apply -f cachebackend.yaml`. The reconciler creates the managed lmcache-server Deployment + Service and publishes the resolved address in `status.endpoint`. 2. **Deploy the engine.** Apply an engine Deployment whose pod template labels include every key/value in `spec.engineSelector.matchLabels`. New pods from that Deployment hit the mutating webhook at admission time. -3. **The webhook claims matching pods (precondition: status.endpoint published).** When the matched CacheBackend has `status.endpoint` populated by the time admission runs, the webhook injects LMCache env vars, the `--kv-transfer-config` CLI arg, and stamps TWO annotations: `inferencecache.io/injected-by: /` (operator-readable identity, visible in `kubectl describe pod`) and `inferencecache.io/injected-by-uid: ` (the matched CR's `metadata.uid` — an apiserver-assigned identifier that the events controller cross-checks against the live CR's UID before emitting. The check catches casual copy-paste of an injected pod's annotations into a fresh template, but it isn't a security boundary: UIDs are readable metadata, so a pod creator with `get` RBAC on CacheBackends can stamp the pair correctly). When the controller is started with `--kvevent-subscriber-image` set (empty by default) AND the matched CacheBackend has a model id configured, the kvevent-subscriber sidecar is also appended; otherwise the engine is wired without the sidecar. A controller watching pods then validates the UID annotation against the live CR's UID and records a `Normal InjectedByCacheBackend` event on the now-persisted pod, visible in `kubectl describe pod`. (The event is emitted from a controller rather than the webhook because the apiserver does not assign `metadata.uid` until after mutating admission — an event recorded from the webhook would have `involvedObject.uid=""` and would not surface under describe. The UID annotation is what lets the controller distinguish a real webhook injection from a user-supplied `injected-by` annotation when the webhook is unreachable under `failurePolicy=Ignore`.) **If the CacheBackend's `status.endpoint` is empty at admission time** (the operator applied the engine Deployment before the reconciler had a chance to publish it), the webhook fail-opens — the pod is admitted unwired, no annotations are stamped, no Event is emitted. Because admission is CREATE-only, this is permanent for that pod; recovery is `kubectl rollout restart deploy/` so new pods re-enter admission. Note that `status.matchedEnginePods` still counts the unwired pod (the selector matches its labels), so `Matched > 0` does **not** guarantee the pod was injected — the per-pod annotations and Event are the authoritative wiring signals. +3. **The webhook claims matching pods (precondition: status.endpoint published).** When the matched CacheBackend has `status.endpoint` populated by the time admission runs, the webhook injects LMCache env vars, the `--kv-transfer-config` CLI arg, and stamps three annotations: `inferencecache.io/injected-by: /` (operator-readable identity), `inferencecache.io/injected-by-uid: ` (the apiserver-assigned identity of the matched CR), and `inferencecache.io/injected-generation: ` (the spec generation validated and rendered at admission time). The events controller cross-checks the UID against the live CR before emitting; the generation receipt lets lifecycle controllers distinguish current pods from pods carrying an older configuration. These annotations are operational evidence, not a security boundary: UID and generation are readable metadata. When the controller is started with `--kvevent-subscriber-image` set (empty by default) AND the matched CacheBackend has a model id configured, the kvevent-subscriber sidecar is also appended; otherwise the engine is wired without the sidecar. A controller watching pods then validates the UID annotation against the live CR's UID and records a `Normal InjectedByCacheBackend` event on the now-persisted pod, visible in `kubectl describe pod`. (The event is emitted from a controller rather than the webhook because the apiserver does not assign `metadata.uid` until after mutating admission — an event recorded from the webhook would have `involvedObject.uid=""` and would not surface under describe.) **If the CacheBackend's `status.endpoint` is empty at admission time** (the operator applied the engine Deployment before the reconciler had a chance to publish it), the webhook fail-opens — the pod is admitted unwired, no injection annotations are stamped, no Event is emitted. Because admission is CREATE-only, this is permanent for that pod; recovery is `kubectl rollout restart deploy/` so new pods re-enter admission. Note that `status.matchedEnginePods` still counts the unwired pod (the selector matches its labels), so `Matched > 0` does **not** guarantee the pod was injected — the per-pod annotations and Event are the authoritative wiring signals. 4. **KV events flow (when the sidecar is configured).** When the subscriber sidecar is present, it streams the engine's KV-cache events to the policy server's index, which surfaces them in `CacheBackend.status` (the index-participation status fields). Without the sidecar, the binding is still observable — `status.matchedEnginePods` and the per-pod annotation/Event still materialize from step 3 — but no KV events flow and the index-participation fields stay unset. 5. **Observe and debug.** `kubectl get cachebackend` shows the `Matched` column — the snapshot count of pods whose labels currently match. `kubectl describe pod ` shows which CacheBackend (if any) claimed it. diff --git a/docs/design/cachebackend-api.md b/docs/design/cachebackend-api.md index 417366d0..49cc28a9 100644 --- a/docs/design/cachebackend-api.md +++ b/docs/design/cachebackend-api.md @@ -394,10 +394,13 @@ args/env only. The `(sglang, SGLangHiCache)` pair configures the selected SGLang engine Pods directly. It does not create a cache-server Deployment, Service, HPA, or -endpoint. The first implementation intentionally publishes no `Ready` -condition: Kubernetes Pod readiness proves that SGLang is serving, but does not -prove a HiCache host-tier write/read round trip. A dedicated readiness contract -is a separate follow-up. +endpoint. Its `Ready` condition reports configuration rollout and serving +availability, not a HiCache data-plane probe: `Ready=True` means every +participating engine Pod carries the current CacheBackend name, UID, and +generation receipt, already contains the configuration that the current +HiCache adapter would inject, and has Kubernetes `Ready=True`. It does not +prove a host-tier write/read round trip or distinguish a GPU hit from a +host-tier read. The required integration shape is: @@ -441,6 +444,40 @@ SGLang workload to apply a new configuration or switch between LMCache and HiCache. The webhook does not inspect image tags: the chosen image must support these SGLang arguments. +The controller evaluates selector-matched Pods with this contract: + +| Pod set | `Ready` | `Progressing` | `Degraded` | Reason | +|---|---|---|---|---| +| No active matching Pods | `False` | `True` | `False` | `AwaitingEnginePods` | +| Every active matching Pod explicitly opts out with `inferencecache.io/skip-inject` | `True` | `False` | `False` | `AllEnginePodsSkipped` | +| Any participating Pod lacks a complete injection receipt | `False` | `False` | `True` | `EnginePodsNotInjected` | +| Any receipt names another CacheBackend identity or a future generation | `False` | `False` | `True` | `EnginePodsInjectionMismatch` | +| At least one receipt carries an older generation | `False` | `True` | `False` | `EnginePodsRolloutInProgress` | +| All receipts are current, but at least one Pod does not contain the engine configuration the HiCache adapter would inject | `False` | `False` | `True` | `EnginePodsNotInjected` | +| All receipts are current, but at least one Pod is not Kubernetes Ready | `False` | `False` | `True` | `EnginePodsUnavailable` | +| All participating Pods carry the current receipt, contain the current adapter configuration, and are Kubernetes Ready | `True` | `False` | `False` | `EnginePodsReady` | + +Terminating Pods and terminal `Succeeded`/`Failed` Pods are excluded. A Pod +with a truthy `inferencecache.io/skip-inject` explicitly opts out and does not +block the remaining participants; this is a public operator control, not a +webhook-authenticated decision. Readiness deliberately does not require the +`inferencecache.io/inject-skipped` audit marker because both annotations are +user-writable and the marker is not an authentication boundary. For every +participant with a current receipt, the controller runs the complete webhook +engine mutation pipeline — canonical HiCache adapter injection followed by +`spec.integration.engineOverrides` — against an in-memory PodSpec copy and +requires it to produce no change. The receipt remains operational metadata +rather than a security boundary: actual configuration convergence prevents a +forged current receipt from proving that absent or conflicting HiCache +configuration was injected. + +The controller does not restart user-owned engine workloads. After a +CacheBackend spec update, old-generation Pods keep the backend at +`EnginePodsRolloutInProgress` until the workload owner rolls them. The +controller polls every 5 seconds while not Ready and uses the existing +30-second matched-Pod cadence after convergence; it does not add a +cluster-wide Pod watch. + HiCache host memory is charged to the engine container's cgroup. The operator must size the engine's memory request/limit and node capacity accordingly. Inference-cache does not derive resource changes from `sizeGB` or `ratio`, and @@ -910,7 +947,7 @@ A separate mutating admission webhook on `corev1/v1.Pod` (`name: mpod.inferencec |---|---| | Selection | Lists `CacheBackend`s in the pod's namespace via the manager's **APIReader** (uncached live client; an informer-cache miss on a freshly-Ready backend would leave the pod permanently unwired since pod CREATE is a one-shot), then matches `pod.Labels` against each `Spec.EngineSelector.MatchLabels`. The first matching `CacheBackend` wins; one with a nil or empty `EngineSelector` is skipped (a "match-everything" selector would silently claim every pod in the namespace). | | Injection | Resolves the runtime adapter via `runtime.Registry.Select(runtimeID, cache)`, resolves `spec.EffectiveRemoteStorage()` independently, and constructs a structured provider `Binding{Protocol, Endpoint}`. Managed ownership uses `status.endpoint` from the live Service; External ownership uses the trimmed, provider-validated `spec.remoteStorage.endpoint` (or legacy `spec.endpoint`) with no fallback to stale status; omitted canonical `remoteStorage` produces a nil host-only binding. The webhook calls `runtime.InjectEngineConfigWithBinding`, so the adapter selects the LMCache, RESP, or Mooncake engine wire from the binding protocol instead of inferring storage from `spec.type`. A missing endpoint fails open only when the selected adapter and binding require one. Events-only skips engine injection because it wires no KV connector and appends only the kvevent-subscriber sidecar. Adapters preserve existing user args/env and make repeat injection idempotent. | -| Annotations | Stamps TWO annotations on every successfully mutated pod: `inferencecache.io/injected-by: /` (operator-readable identity, shows in `kubectl describe pod`) AND `inferencecache.io/injected-by-uid: ` (the matched CR's metadata.uid). Successful injection also clears any stale `inferencecache.io/inject-skipped` marker. Reads `inferencecache.io/skip-inject: ` as an opt-out: the webhook returns Allowed, skips engine wiring, clears any stale injected-by/injected-by-uid pair, and stamps `inferencecache.io/inject-skipped: skip-inject-annotation` so explicit operator opt-out is distinguishable from selector drift. On all other fail-open returns after the pod is decoded (list/no match/missing endpoint/adapter errors), the webhook strips stale injected-by/injected-by-uid and inject-skipped annotations so a user cannot trick the events controller by pre-stamping a pod template. Decode failures fail open before a Pod exists to patch, so stale annotations cannot be cleared on that path. | +| Annotations | Stamps three annotations on every successfully mutated pod: `inferencecache.io/injected-by: /` (operator-readable identity), `inferencecache.io/injected-by-uid: ` (the matched CR's metadata.uid), and `inferencecache.io/injected-generation: ` (the spec generation validated and rendered at admission time). Successful injection also clears any stale `inferencecache.io/inject-skipped` marker. Reads `inferencecache.io/skip-inject: ` as an opt-out: the webhook returns Allowed, skips engine wiring, clears the complete injection receipt, and stamps `inferencecache.io/inject-skipped: skip-inject-annotation` so explicit operator opt-out is distinguishable from selector drift. On all other fail-open returns after the pod is decoded (list/no match/missing endpoint/adapter errors), the webhook strips stale injection and skip-decision annotations so a user cannot trick downstream controllers by pre-stamping a pod template. Decode failures fail open before a Pod exists to patch, so stale annotations cannot be cleared on that path. | | Events | The webhook itself does NOT record events (the apiserver assigns `metadata.uid` after mutating admission, so a webhook-recorded event would carry `involvedObject.uid=""` and be invisible to `kubectl describe pod`). Instead, the pod-watching `engine-pod-events` controller reads the persisted decision annotations after CREATE. For injected pods, it validates `inferencecache.io/injected-by-uid` against the live CR's `metadata.uid` and records a `Normal InjectedByCacheBackend` event on the now-persisted pod. For explicitly skipped pods carrying both a truthy `inferencecache.io/skip-inject` and `inferencecache.io/inject-skipped: skip-inject-annotation`, it records a `Normal SkippedByOperator` event on that pod. The skip marker is not authenticated, and `skipInjection` treats a pre-existing correct marker as already converged; `SkippedByOperator` therefore means the persisted pod carries the explicit opt-out plus skipped marker, not proof that the webhook authored the marker. The UID match REDUCES — but does NOT eliminate — the failurePolicy=Ignore forgery surface for injected pods: a casual copy-paste of an injected pod's annotations into a fresh template won't match the live CR's UID, but `metadata.uid` is not secret, so a pod creator with `get` RBAC on CacheBackends can read it and stamp the pair correctly. The injected Event signals "the webhook claims this pod was injected and the claim is consistent with the live CR," not "the webhook was cryptographically authenticated." The controller skips the injected event when the CR is missing, the UID annotation is absent, or the UID does not match — see the controller godoc for the full skip table. controller-runtime's EventBroadcaster aggregates duplicates on the apiserver side, so a re-enqueue across controller restarts upserts the existing event rather than spamming. | | Idempotency | The handler calls the adapter unconditionally on every admission and trusts the adapter to converge the full injected contract. For LMCache this is env plus the engine-specific required surface — `--kv-transfer-config` for vLLM; for SGLang `--enable-lmcache` + `--lmcache-config-file` **plus** the MP-worker native sidecar and the shared config / `/dev/shm` volumes + mounts. Its merge primitives (`upsertEnv` / `upsertArgPair` / `upsertFlag`, and for SGLang `adoptContainer` / `adoptVolume` / `upsertMountByName`) converge on the desired value rather than appending a duplicate. The SGLang `adopt*` pair additionally distinguishes the adapter's own prior injection (converge) from an operator's object squatting a reserved name (reject → fail-open admit) — see [Names the MP wire reserves](#sglang-engine-support). Native HiCache validates all reserved arguments against the original pod before mutation, preserves one matching or well-formed operator-supplied value, appends each missing canonical argument once, and rejects conflicts, malformed values, or duplicates without partially changing the pod. Re-admission of a fully-injected pod therefore produces an empty JSON-patch set. Trusting the adapter rather than a handler-side env-presence shortcut avoids the trap where a partially-injected pod is admitted permanently missing the rest of the contract. | | Fail-open | Every error path (decode failure, list error, no matching backend, missing `status.endpoint`, no registered adapter, adapter rejection, re-encode failure) returns `admission.Allowed(...)` with a reason — webhook errors MUST NOT block engine admission. `MutatingWebhookConfiguration.failurePolicy` is also pinned to `Ignore` as a belt-and-suspenders second layer. | diff --git a/docs/reference-stack/scripts/default_install_smoke.sh b/docs/reference-stack/scripts/default_install_smoke.sh index 8dec0846..629c9058 100755 --- a/docs/reference-stack/scripts/default_install_smoke.sh +++ b/docs/reference-stack/scripts/default_install_smoke.sh @@ -95,10 +95,12 @@ # Also exercises the validating webhook's EventsOnly+External rejection. # 9c. Native SGLang HiCache end-to-end: applying the committed HiCache sample # is observed by the controller without rendering a Deployment, Service, or -# HPA; status.endpoint stays empty and no synthetic Ready condition is -# published. A matching engine Pod sent through real server-side dry-run -# admission receives the complete native --hicache-* argument contract, -# proving the endpoint-free Pod webhook path without starting SGLang. +# HPA; status.endpoint stays empty and the CR waits at +# Ready=False/AwaitingEnginePods. A matching lightweight Pod receives the +# complete native --hicache-* argument contract plus the current generation +# receipt, becomes Kubernetes Ready, and drives the CR to +# Ready=True/EnginePodsReady. This proves the endpoint-free admission and +# controller contract without claiming a real SGLang or KV data-plane test. # 10. The /snapshot endpoint rejects unauthenticated callers AT THE NETWORK # LAYER: a side curl pod outside the controller's SA identity (and outside # the NetworkPolicy allowlist) has its connection to :8081 DROPPED by the @@ -245,6 +247,7 @@ # EXTERNAL_INJECT_TIMEOUT, EVENTSONLY_BACKEND_TIMEOUT, # EVENTSONLY_SMOKE_NS, EVENTSONLY_SMOKE_CB_NAME, SAMPLE_APPLY_NS, # HICACHE_SMOKE_TIMEOUT, HICACHE_SMOKE_NS, HICACHE_SMOKE_CB_NAME, +# HICACHE_SMOKE_POD_NAME, # KERNEL_CHECK_SMOKE_NS, KERNEL_CHECK_POD_TIMEOUT, # KERNEL_CHECK_COND_TIMEOUT, MOONCAKE_SMOKE_NS, MOONCAKE_MASTER_IMAGE. @@ -367,10 +370,11 @@ EVENTSONLY_SMOKE_NS="${EVENTSONLY_SMOKE_NS:-ic-smoke-events-only}" EVENTSONLY_SMOKE_CB_NAME="${EVENTSONLY_SMOKE_CB_NAME:-cachebackend-events-only}" # Native SGLang HiCache fixture identifiers. This engine-local backend has no -# endpoint or controller-owned workload, so its dedicated namespace should -# contain only the persisted CacheBackend used by the smoke. +# endpoint or controller-owned workload; the smoke adds one user-owned +# lightweight engine fixture to drive Kubernetes readiness. HICACHE_SMOKE_NS="${HICACHE_SMOKE_NS:-ic-smoke-sglang-hicache}" HICACHE_SMOKE_CB_NAME="${HICACHE_SMOKE_CB_NAME:-sglang-hicache}" +HICACHE_SMOKE_POD_NAME="${HICACHE_SMOKE_POD_NAME:-sglang-hicache-engine}" KIND="${KIND:-$([ -x ./bin/kind ] && echo ./bin/kind || echo kind)}" pf_pid="" @@ -459,6 +463,8 @@ collect_diagnostics() { # if the smoke aborted before that section. kubectl get cb -n "$HICACHE_SMOKE_NS" "$HICACHE_SMOKE_CB_NAME" -o yaml \ >"$LOG_DIR/sglang-hicache-cb.yaml" 2>&1 || true + kubectl get pod -n "$HICACHE_SMOKE_NS" "$HICACHE_SMOKE_POD_NAME" -o yaml \ + >"$LOG_DIR/sglang-hicache-engine-pod.yaml" 2>&1 || true kubectl get deploy,svc,hpa -n "$HICACHE_SMOKE_NS" \ >"$LOG_DIR/sglang-hicache-ns-workloads.txt" 2>&1 || true # Kernel-check smoke artefacts. Best-effort — the objects may not @@ -2677,10 +2683,10 @@ kubectl delete namespace "$EVENTSONLY_SMOKE_NS" --ignore-not-found --wait=false # --- Native SGLang HiCache end-to-end -------------------------------------- # This phase drives the new operator-facing surface through the real installed # CRD, validating webhook, controller, and Pod mutating webhook. HiCache is -# engine-local, so the observable controller contract is deliberately negative: -# no cache-server workload, no endpoint, and no synthetic Ready condition. The -# Pod is server-side dry-run only; its admitted shape proves native argument -# injection without pulling or starting a real SGLang image. +# engine-local, so the controller observes user-owned engine Pods rather than a +# cache-server workload. A lightweight busybox Pod exercises admission + +# Kubernetes readiness without claiming SGLang runtime or KV data-plane +# coverage. log "exercising native SGLang HiCache end-to-end in namespace $HICACHE_SMOKE_NS" kubectl create namespace "$HICACHE_SMOKE_NS" --dry-run=client -o yaml | kubectl apply -f - >/dev/null @@ -2693,31 +2699,44 @@ sed "s|^ name: sglang-hicache\$| name: $HICACHE_SMOKE_CB_NAME|" \ kubectl -n "$HICACHE_SMOKE_NS" apply -f "$hc_sample_tmp" >/dev/null \ || fail "kubectl apply native SGLang HiCache sample failed" -# Wait until the controller has taken the engine-local path. observedGeneration -# is the positive acknowledgement; endpoint/Ready/workload assertions below pin -# what that path intentionally does not publish or provision. +# Wait until the controller has taken the engine-local path. With no engine Pod +# yet, the contract is Ready=False/AwaitingEnginePods and Progressing=True. hc_generation="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ -o jsonpath='{.metadata.generation}')" deadline=$(($(date +%s) + HICACHE_SMOKE_TIMEOUT)) hc_observed_generation="" -until [ "$hc_observed_generation" = "$hc_generation" ]; do +hc_ready_status="" +hc_ready_reason="" +until [ "$hc_observed_generation" = "$hc_generation" ] && \ + [ "$hc_ready_status" = "False" ] && \ + [ "$hc_ready_reason" = "AwaitingEnginePods" ]; do hc_observed_generation="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ -o jsonpath='{.status.observedGeneration}' 2>/dev/null || true)" + hc_ready_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}' 2>/dev/null || true)" + hc_ready_reason="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Ready")].reason}' 2>/dev/null || true)" if [ "$(date +%s)" -ge "$deadline" ]; then kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" -o yaml || true - fail "native HiCache CR was not observed within ${HICACHE_SMOKE_TIMEOUT}s: generation=$hc_generation observedGeneration=$hc_observed_generation" + fail "native HiCache CR did not reach False/AwaitingEnginePods within ${HICACHE_SMOKE_TIMEOUT}s: generation=$hc_generation observedGeneration=$hc_observed_generation Ready=$hc_ready_status/$hc_ready_reason" fi sleep 1 done -log "native HiCache CR observed at generation $hc_observed_generation" +hc_progressing_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Progressing")].status}' 2>/dev/null || true)" +hc_degraded_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Degraded")].status}' 2>/dev/null || true)" +if [ "$hc_progressing_status" != "True" ] || [ "$hc_degraded_status" != "False" ]; then + kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" -o yaml || true + fail "native HiCache no-Pod conditions Progressing=$hc_progressing_status Degraded=$hc_degraded_status, want True/False" +fi +log "native HiCache CR observed at generation $hc_observed_generation and awaits engine Pods" hc_endpoint="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ -o jsonpath='{.status.endpoint}' 2>/dev/null || true)" -hc_ready="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ - -o jsonpath='{.status.conditions[?(@.type=="Ready")].type}' 2>/dev/null || true)" -if [ -n "$hc_endpoint" ] || [ -n "$hc_ready" ]; then +if [ -n "$hc_endpoint" ]; then kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" -o yaml || true - fail "native HiCache published server-backed status (endpoint=$hc_endpoint Ready=$hc_ready, want both absent)" + fail "native HiCache published server-backed endpoint=$hc_endpoint, want empty" fi hc_dep_count="$(kubectl -n "$HICACHE_SMOKE_NS" get deploy -o name 2>/dev/null | wc -l | tr -d ' ')" @@ -2727,7 +2746,7 @@ if [ "$hc_dep_count" != "0" ] || [ "$hc_svc_count" != "0" ] || [ "$hc_hpa_count" kubectl -n "$HICACHE_SMOKE_NS" get deploy,svc,hpa || true fail "native HiCache rendered controller-owned workload (deploy=$hc_dep_count svc=$hc_svc_count hpa=$hc_hpa_count, want 0/0/0)" fi -log "native HiCache rendered no Deployment, Service, or HPA and published no endpoint/Ready condition" +log "native HiCache rendered no Deployment, Service, or HPA and published no endpoint" # Exercise the installed Pod mutating webhook with a matching, single-container # engine Pod. The dry-run response is the fully admitted Pod, including webhook @@ -2737,7 +2756,7 @@ cat > "$hc_engine_fixture" </dev/null)"; then fail "matching SGLang Pod did not pass server-side dry-run admission" fi -hc_expected_args=$'sleep\n3600\n--enable-hierarchical-cache\n--hicache-ratio\n2.0\n--hicache-write-policy\nwrite_through\n--hicache-io-backend\nkernel\n--hicache-mem-layout\nlayer_first' +hc_expected_args=$'sleep 3600\n--enable-hierarchical-cache\n--hicache-ratio\n2.0\n--hicache-write-policy\nwrite_through\n--hicache-io-backend\nkernel\n--hicache-mem-layout\nlayer_first' if [ "$hc_pod_args" != "$hc_expected_args" ]; then printf '[install-smoke] admitted SGLang args:\n%s\n' "$hc_pod_args" >&2 fail "native HiCache Pod mutation did not produce the expected complete argument contract" @@ -2768,7 +2789,57 @@ hc_injected_by="$(kubectl create --dry-run=server --request-timeout=30s \ if [ "$hc_injected_by" != "$HICACHE_SMOKE_NS/$HICACHE_SMOKE_CB_NAME" ]; then fail "native HiCache dry-run Pod injected-by=$hc_injected_by, want $HICACHE_SMOKE_NS/$HICACHE_SMOKE_CB_NAME" fi -log "native HiCache Pod webhook injected the complete CLI contract and backend identity" +hc_cb_uid="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.metadata.uid}')" +hc_injected_uid="$(kubectl create --dry-run=server --request-timeout=30s \ + -f "$hc_engine_fixture" \ + -o go-template='{{index .metadata.annotations "inferencecache.io/injected-by-uid"}}' 2>/dev/null)" \ + || fail "could not read native HiCache injected-by-uid annotation from dry-run Pod" +hc_injected_generation="$(kubectl create --dry-run=server --request-timeout=30s \ + -f "$hc_engine_fixture" \ + -o go-template='{{index .metadata.annotations "inferencecache.io/injected-generation"}}' 2>/dev/null)" \ + || fail "could not read native HiCache injected-generation annotation from dry-run Pod" +if [ "$hc_injected_uid" != "$hc_cb_uid" ] || [ "$hc_injected_generation" != "$hc_generation" ]; then + fail "native HiCache dry-run Pod receipt uid=$hc_injected_uid generation=$hc_injected_generation, want uid=$hc_cb_uid generation=$hc_generation" +fi +log "native HiCache Pod webhook injected the complete CLI contract and current CacheBackend receipt" + +# Persist the same lightweight Pod. /bin/sh -c consumes only the first arg as +# its script, so the appended SGLang flags become unused positional parameters +# and busybox can reach Kubernetes Ready without pretending to be SGLang. +kubectl apply -f "$hc_engine_fixture" >/dev/null \ + || fail "could not create native HiCache readiness fixture Pod" +kubectl -n "$HICACHE_SMOKE_NS" wait --for=condition=Ready "pod/$HICACHE_SMOKE_POD_NAME" \ + --timeout="${HICACHE_SMOKE_TIMEOUT}s" >/dev/null \ + || fail "native HiCache readiness fixture Pod did not become Kubernetes Ready" + +deadline=$(($(date +%s) + HICACHE_SMOKE_TIMEOUT)) +hc_ready_status="" +hc_ready_reason="" +until [ "$hc_ready_status" = "True" ] && [ "$hc_ready_reason" = "EnginePodsReady" ]; do + hc_ready_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}' 2>/dev/null || true)" + hc_ready_reason="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Ready")].reason}' 2>/dev/null || true)" + if [ "$(date +%s)" -ge "$deadline" ]; then + kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" -o yaml || true + kubectl -n "$HICACHE_SMOKE_NS" get pod "$HICACHE_SMOKE_POD_NAME" -o yaml || true + fail "native HiCache CR did not reach True/EnginePodsReady within ${HICACHE_SMOKE_TIMEOUT}s: Ready=$hc_ready_status/$hc_ready_reason" + fi + sleep 1 +done + +hc_progressing_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Progressing")].status}' 2>/dev/null || true)" +hc_degraded_status="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.conditions[?(@.type=="Degraded")].status}' 2>/dev/null || true)" +hc_matched="$(kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" \ + -o jsonpath='{.status.matchedEnginePods}' 2>/dev/null || true)" +if [ "$hc_progressing_status" != "False" ] || [ "$hc_degraded_status" != "False" ] || [ "$hc_matched" != "1" ]; then + kubectl -n "$HICACHE_SMOKE_NS" get cb "$HICACHE_SMOKE_CB_NAME" -o yaml || true + fail "native HiCache converged conditions Progressing=$hc_progressing_status Degraded=$hc_degraded_status Matched=$hc_matched, want False/False/1" +fi +log "native HiCache CR reached Ready=True/EnginePodsReady from one current-generation Kubernetes Ready engine Pod" kubectl delete cb -n "$HICACHE_SMOKE_NS" "$HICACHE_SMOKE_CB_NAME" --ignore-not-found --wait=false >/dev/null || true kubectl delete namespace "$HICACHE_SMOKE_NS" --ignore-not-found --wait=false >/dev/null || true diff --git a/internal/controller/cachebackend_controller.go b/internal/controller/cachebackend_controller.go index e5a6ebba..33afca5a 100644 --- a/internal/controller/cachebackend_controller.go +++ b/internal/controller/cachebackend_controller.go @@ -33,23 +33,29 @@ import ( sglangadapter "github.com/cachebox-project/inference-cache/pkg/adapters/runtime/sglang" ) -// Status condition types published on a managed CacheBackend. +// Status condition types published on a CacheBackend with an active lifecycle. // -// Ready reports whether the managed backend workload is currently serving -// (gated by the KV-event readiness gate — see evaluateKVEventReadiness). +// Ready reports whether the backend's serving contract is currently satisfied: +// a managed backend workload is serving (optionally gated by KV events), an +// External endpoint is accepted, or every participating SGLang HiCache engine +// Pod carries the current receipt, contains the adapter's current configuration, +// and is Ready. Other host-only combinations retain their serverless readiness +// contract. // Progressing reports whether the controller is still driving the live state // toward the desired state (template render, child apply, rollout in flight, -// awaiting first KV event). Degraded reports a terminal unhealthy state. +// awaiting engine Pods, awaiting first KV event). Degraded reports a terminal +// unhealthy state. // Ready + Progressing together tell a still-converging backend // (Ready=False, Progressing=True) apart from a stuck/degraded one // (Ready=False, Progressing=False); Degraded names the specific failure. const ( conditionTypeReady = "Ready" conditionTypeProgressing = "Progressing" - // Degraded is published alongside Ready. It is True only when the - // backend is in a genuinely degraded terminal state (rolled out but - // replicas unavailable, or the workload is Available but no KV events - // observed within firstEventTimeout). + // Degraded is published alongside Ready. It is True only when the backend + // is in a genuinely degraded terminal state (rolled out but replicas are + // unavailable, a SGLang HiCache injection receipt is invalid, or the + // workload is Available but no KV events were observed within + // firstEventTimeout). conditionTypeDegraded = "Degraded" ) @@ -505,10 +511,11 @@ func (r *CacheBackendReconciler) dispatch(ctx context.Context, logger logr.Logge "namespace", backend.Namespace, "name", backend.Name) return ctrl.Result{}, r.reconcileUnmanaged(ctx, backend) } - // Native HiCache remains endpoint-free and intentionally publishes no - // Ready condition until its separate readiness contract is implemented. + // Native HiCache is an engine-local host-only hierarchy. It has no + // provider workload or endpoint; readiness comes from selector-matched + // engine Pods carrying the current injected configuration. if backend.Spec.EffectiveCacheType() == cachev1alpha1.CacheBackendTypeSGLangHiCache { - return ctrl.Result{}, r.reconcileUnmanaged(ctx, backend) + return r.reconcileEngineLocal(ctx, backend, adapter) } return r.reconcileHostOnly(ctx, backend) } @@ -905,9 +912,10 @@ func (r *CacheBackendReconciler) reconcileUnmanaged(ctx context.Context, backend }) } -// reconcileManaged renders the cache-server PodSpec + Service via the runtime -// adapter, wraps them into a Deployment + Service owned by the CR, and -// publishes the resolved endpoint to status. +// reconcileManaged wraps the remote-storage provider's rendered cache-server +// PodSpec + Service into a Deployment + Service owned by the CR, and publishes +// the resolved endpoint to status. Provider selection and rendering happen in +// dispatch before this function is called. // // Apply drives desired state; status reflects observed state. The two must not // block each other: if a desired-state write fails (e.g. a transient API-server @@ -918,9 +926,8 @@ func (r *CacheBackendReconciler) reconcileUnmanaged(ctx context.Context, backend func (r *CacheBackendReconciler) reconcileManaged(ctx context.Context, logger logr.Logger, backend *cachev1alpha1.CacheBackend, rendered *backendadapter.RenderedStorage) (ctrl.Result, error) { podSpec, svcSpec := rendered.PodSpec, rendered.Service if podSpec == nil || svcSpec == nil { - // Engine-local adapters such as native SGLang HiCache intentionally - // render no cache-server. Reuse the unmanaged lifecycle to shed any - // previously owned workload and clear server-backed status. + // Engine-local adapters render no provider here; their lifecycle is + // dispatched before provider selection. A nil managed render is invalid. logger.V(1).Info("adapter rendered no cache-server; treating as unmanaged", "namespace", backend.Namespace, "name", backend.Name) return ctrl.Result{}, r.reconcileUnmanaged(ctx, backend) diff --git a/internal/controller/cachebackend_controller_test.go b/internal/controller/cachebackend_controller_test.go index 9f8aaa33..8d72521a 100644 --- a/internal/controller/cachebackend_controller_test.go +++ b/internal/controller/cachebackend_controller_test.go @@ -804,15 +804,31 @@ func TestReconcileSwitchToStatefulSetClearsObservedServerInstance(t *testing.T) func TestReconcileSwitchToSGLangHiCacheCleansManagedState(t *testing.T) { scheme := newScheme(t) managed := lmcacheBackend("cache", "ns1") + managed.UID = types.UID("cache-uid") managed.Spec.Autoscaling = &cachev1alpha1.CacheBackendAutoscalingSpec{ MinReplicas: ptrInt32(1), MaxReplicas: 3, } + injectedOwner := managed.DeepCopy() + injectedOwner.Generation = 2 + injectedOwner.Spec.Runtime = cachev1alpha1.CacheBackendRuntimeSGLang + injectedOwner.Spec.Type = cachev1alpha1.CacheBackendTypeSGLangHiCache + injectedOwner.Spec.Integration = &cachev1alpha1.CacheBackendIntegrationSpec{ + Mode: cachev1alpha1.CacheBackendIntegrationModeOffload, + Role: cachev1alpha1.CacheBackendIntegrationRoleReadWrite, + } + injectedOwner.Spec.Autoscaling = nil + injectedOwner.Spec.EngineSelector = &cachev1alpha1.CacheBackendEngineSelector{ + MatchLabels: map[string]string{"app": "sglang"}, + } + injectedOwner.Spec.HiCache = &cachev1alpha1.SGLangHiCacheSpec{Ratio: "2"} + pod0 := engineLocalPodFixture("sglang-0", injectedOwner, injectedOwner.Generation, true) + pod1 := engineLocalPodFixture("sglang-1", injectedOwner, injectedOwner.Generation, true) r := newReconciler( scheme, managed, - &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "sglang-0", Namespace: "ns1", Labels: map[string]string{"app": "sglang"}}}, - &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "sglang-1", Namespace: "ns1", Labels: map[string]string{"app": "sglang"}}}, + &pod0, + &pod1, ) reconcile(t, r, "cache", "ns1") var managedHPA autoscalingv2.HorizontalPodAutoscaler @@ -837,12 +853,12 @@ func TestReconcileSwitchToSGLangHiCacheCleansManagedState(t *testing.T) { switching := getBackend(t, r, "cache", "ns1") switching.Generation = 2 + switching.Spec.Runtime = cachev1alpha1.CacheBackendRuntimeSGLang switching.Spec.Type = cachev1alpha1.CacheBackendTypeSGLangHiCache switching.Spec.DeploymentKind = cachev1alpha1.CacheBackendDeploymentKindStatefulSet switching.Spec.Integration = &cachev1alpha1.CacheBackendIntegrationSpec{ - Engine: "sglang", - Mode: cachev1alpha1.CacheBackendIntegrationModeOffload, - Role: cachev1alpha1.CacheBackendIntegrationRoleReadWrite, + Mode: cachev1alpha1.CacheBackendIntegrationModeOffload, + Role: cachev1alpha1.CacheBackendIntegrationRoleReadWrite, } switching.Spec.EngineSelector = &cachev1alpha1.CacheBackendEngineSelector{ MatchLabels: map[string]string{"app": "sglang"}, @@ -869,8 +885,17 @@ func TestReconcileSwitchToSGLangHiCacheCleansManagedState(t *testing.T) { if got.Status.Endpoint != "" || got.Status.ObservedServerInstance != "" { t.Fatalf("stale managed endpoint/server instance survived: %+v", got.Status) } - if len(got.Status.Conditions) != 0 { - t.Fatalf("SGLangHiCache first commit must publish no conditions, got %v", got.Status.Conditions) + ready := meta.FindStatusCondition(got.Status.Conditions, conditionTypeReady) + if ready == nil || ready.Status != metav1.ConditionTrue || ready.Reason != reasonEnginePodsReady { + t.Fatalf("Ready = %+v, want True/%s", ready, reasonEnginePodsReady) + } + progressing := meta.FindStatusCondition(got.Status.Conditions, conditionTypeProgressing) + if progressing == nil || progressing.Status != metav1.ConditionFalse { + t.Fatalf("Progressing = %+v, want False", progressing) + } + degraded := meta.FindStatusCondition(got.Status.Conditions, conditionTypeDegraded) + if degraded == nil || degraded.Status != metav1.ConditionFalse { + t.Fatalf("Degraded = %+v, want False", degraded) } if got.Status.ObservedGeneration != got.Generation { t.Fatalf("observedGeneration = %d, want generation %d", got.Status.ObservedGeneration, got.Generation) diff --git a/internal/controller/cachebackend_engine_local_readiness.go b/internal/controller/cachebackend_engine_local_readiness.go new file mode 100644 index 00000000..be21cffb --- /dev/null +++ b/internal/controller/cachebackend_engine_local_readiness.go @@ -0,0 +1,291 @@ +package controller + +import ( + "context" + "fmt" + "reflect" + "sort" + "strconv" + "strings" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + + cachev1alpha1 "github.com/cachebox-project/inference-cache/api/v1alpha1" + podwebhook "github.com/cachebox-project/inference-cache/internal/webhook/pod" + adapterruntime "github.com/cachebox-project/inference-cache/pkg/adapters/runtime" +) + +const ( + reasonEnginePodsReady = "EnginePodsReady" + reasonAwaitingEnginePods = "AwaitingEnginePods" + reasonEnginePodsRolloutInProgress = "EnginePodsRolloutInProgress" + reasonEnginePodsUnavailable = "EnginePodsUnavailable" + reasonEnginePodsNotInjected = "EnginePodsNotInjected" + reasonEnginePodsInjectionMismatch = "EnginePodsInjectionMismatch" + reasonAllEnginePodsSkipped = "AllEnginePodsSkipped" +) + +type engineLocalReadiness struct { + readyStatus metav1.ConditionStatus + readyReason string + readyMessage string + progressingStatus metav1.ConditionStatus + progressingReason string + progressingMessage string + degradedStatus metav1.ConditionStatus + degradedReason string + degradedMessage string +} + +// reconcileEngineLocal reports readiness from user-owned engine Pods. Dispatch +// currently scopes this stronger contract to native SGLang HiCache; other +// host-only combinations retain reconcileHostOnly's serverless contract. +// It never creates, patches, or restarts an engine workload. CacheBackend spec +// changes therefore remain Progressing until the workload owner replaces the +// stale-generation Pods and they pass CREATE admission again. +func (r *CacheBackendReconciler) reconcileEngineLocal( + ctx context.Context, + backend *cachev1alpha1.CacheBackend, + adapter adapterruntime.KVCacheRuntimeAdapter, +) (ctrl.Result, error) { + if err := r.cleanupOwnedWorkload(ctx, backend); err != nil { + return ctrl.Result{}, err + } + + selector := labels.SelectorFromSet(backend.Spec.EngineSelector.MatchLabels) + var pods corev1.PodList + reader := client.Reader(r.APIReader) + if reader == nil { + reader = r.Client + } + if err := reader.List(ctx, &pods, + client.InNamespace(backend.Namespace), + client.MatchingLabelsSelector{Selector: selector}, + ); err != nil { + // Readiness is observation-only. Preserve the last published verdict + // when the live Pod set cannot be observed, and let controller-runtime + // retry with backoff. + return ctrl.Result{}, fmt.Errorf("list engine Pods for CacheBackend %s/%s: %w", + backend.Namespace, backend.Name, err) + } + + readiness := evaluateEngineLocalReadiness(backend, pods.Items, adapter) + r.clearServerInstanceLatchShadow(backend) + r.probeLimiter.forget(client.ObjectKeyFromObject(backend).String()) + + err := r.patchStatus(ctx, backend, func() { + backend.Status.Endpoint = "" + backend.Status.ObservedServerInstance = "" + backend.Status.ObservedGeneration = backend.Generation + backend.Status.FirstAvailableAt = nil + + meta.SetStatusCondition(&backend.Status.Conditions, metav1.Condition{ + Type: conditionTypeReady, + Status: readiness.readyStatus, + Reason: readiness.readyReason, + Message: readiness.readyMessage, + ObservedGeneration: backend.Generation, + }) + meta.SetStatusCondition(&backend.Status.Conditions, metav1.Condition{ + Type: conditionTypeProgressing, + Status: readiness.progressingStatus, + Reason: readiness.progressingReason, + Message: readiness.progressingMessage, + ObservedGeneration: backend.Generation, + }) + meta.SetStatusCondition(&backend.Status.Conditions, metav1.Condition{ + Type: conditionTypeDegraded, + Status: readiness.degradedStatus, + Reason: readiness.degradedReason, + Message: readiness.degradedMessage, + ObservedGeneration: backend.Generation, + }) + + // Server-backed health signals do not apply to native SGLang HiCache. + meta.RemoveStatusCondition(&backend.Status.Conditions, conditionTypeFunctionalProbeOK) + meta.RemoveStatusCondition(&backend.Status.Conditions, conditionTypeEngineKernelsHealthy) + meta.RemoveStatusCondition(&backend.Status.Conditions, conditionTypeT2Degraded) + meta.RemoveStatusCondition(&backend.Status.Conditions, conditionTypeEngineCompatibility) + }) + if err != nil { + return ctrl.Result{}, err + } + if readiness.readyStatus != metav1.ConditionTrue { + return ctrl.Result{RequeueAfter: r.matchedEnginePodsChurnRequeueInterval()}, nil + } + return ctrl.Result{}, nil +} + +// evaluateEngineLocalReadiness applies the native SGLang HiCache readiness +// contract to one live selector result. Terminating and terminal Pods are +// historical and ignored. Explicitly skipped Pods opt out. Every remaining Pod +// must carry the current CacheBackend name/UID/generation receipt, already +// contain the engine config the adapter would inject for that CacheBackend, and +// be Kubernetes Ready. +func evaluateEngineLocalReadiness( + backend *cachev1alpha1.CacheBackend, + pods []corev1.Pod, + adapter adapterruntime.KVCacheRuntimeAdapter, +) engineLocalReadiness { + var ( + activeCount int + skipped []string + participants []corev1.Pod + ) + for i := range pods { + pod := &pods[i] + if pod.DeletionTimestamp != nil || + pod.Status.Phase == corev1.PodSucceeded || + pod.Status.Phase == corev1.PodFailed { + continue + } + activeCount++ + if podwebhook.SkipAnnotationOptsOut(pod.Annotations[podwebhook.AnnotationSkip]) { + // AnnotationSkip is the authoritative public operator opt-out. + // AnnotationInjectSkipped is webhook-written audit metadata, not an + // authenticated receipt, so readiness deliberately does not require + // it here. Both annotations are user-writable; failurePolicy=Ignore + // means their presence cannot authenticate webhook execution. + skipped = append(skipped, pod.Name) + continue + } + participants = append(participants, *pod) + } + + switch { + case activeCount == 0: + return engineLocalProgressing(reasonAwaitingEnginePods, + "no active engine Pods match spec.engineSelector") + case len(participants) == 0: + return engineLocalReady(reasonAllEnginePodsSkipped, + fmt.Sprintf("all %d active matching engine Pods explicitly skipped cache injection", activeCount)) + } + + wantBy := backend.Namespace + "/" + backend.Name + wantUID := string(backend.UID) + var missing, mismatched, stale, unconfigured, unavailable []string + for i := range participants { + pod := &participants[i] + injectedBy := pod.Annotations[podwebhook.AnnotationInjectedBy] + injectedUID := pod.Annotations[podwebhook.AnnotationInjectedByUID] + injectedGeneration := pod.Annotations[podwebhook.AnnotationInjectedGeneration] + generation, generationErr := strconv.ParseInt(injectedGeneration, 10, 64) + + switch { + case injectedBy == "" || injectedUID == "" || injectedGeneration == "" || generationErr != nil || generation < 0: + missing = append(missing, pod.Name) + case injectedBy != wantBy || injectedUID != wantUID || generation > backend.Generation: + mismatched = append(mismatched, pod.Name) + case generation < backend.Generation: + stale = append(stale, pod.Name) + case !engineConfigConverged(adapter, pod, backend): + unconfigured = append(unconfigured, pod.Name) + case !podIsReady(pod): + unavailable = append(unavailable, pod.Name) + } + } + + total := len(participants) + switch { + case len(missing) > 0: + return engineLocalDegraded(reasonEnginePodsNotInjected, + podDiagnostic("%d/%d engine Pods are missing a valid CacheBackend injection receipt and must be recreated: %s", + missing, total)) + case len(mismatched) > 0: + return engineLocalDegraded(reasonEnginePodsInjectionMismatch, + podDiagnostic("%d/%d engine Pods carry an injection receipt for a different CacheBackend identity or future generation: %s", + mismatched, total)) + case len(stale) > 0: + return engineLocalProgressing(reasonEnginePodsRolloutInProgress, + podDiagnostic("%d/%d engine Pods carry an older CacheBackend generation and must be rolled out: %s", + stale, total)) + case len(unconfigured) > 0: + return engineLocalDegraded(reasonEnginePodsNotInjected, + podDiagnostic("%d/%d engine Pods do not contain the current CacheBackend engine configuration and must be recreated: %s", + unconfigured, total)) + case len(unavailable) > 0: + return engineLocalDegraded(reasonEnginePodsUnavailable, + podDiagnostic("%d/%d current-generation engine Pods are not Ready: %s", + unavailable, total)) + default: + message := fmt.Sprintf("%d/%d engine Pods carry CacheBackend generation %d and are Ready", + total, total, backend.Generation) + if len(skipped) > 0 { + message += fmt.Sprintf("; %d additional matching Pods explicitly skipped injection", len(skipped)) + } + return engineLocalReady(reasonEnginePodsReady, message) + } +} + +// engineConfigConverged uses the Pod webhook's complete idempotent engine +// mutation contract as a read-only verifier. A converged PodSpec is unchanged +// when the current adapter configuration and engineOverrides are applied to an +// in-memory copy. This verifies the actual engine configuration instead of +// trusting user-writable receipt annotations as proof that the mutating +// webhook ran. +func engineConfigConverged( + adapter adapterruntime.KVCacheRuntimeAdapter, + pod *corev1.Pod, + backend *cachev1alpha1.CacheBackend, +) bool { + if adapter == nil { + return false + } + want := pod.Spec.DeepCopy() + if err := podwebhook.ApplyEngineConfigWithOverrides(adapter, want, nil, backend); err != nil { + return false + } + return reflect.DeepEqual(*want, pod.Spec) +} + +func engineLocalReady(reason, message string) engineLocalReadiness { + return engineLocalReadiness{ + readyStatus: metav1.ConditionTrue, + readyReason: reason, + readyMessage: message, + progressingStatus: metav1.ConditionFalse, + progressingReason: "Synced", + progressingMessage: message, + degradedStatus: metav1.ConditionFalse, + degradedReason: reasonNotDegraded, + degradedMessage: "backend is not in a degraded state", + } +} + +func engineLocalProgressing(reason, message string) engineLocalReadiness { + return engineLocalReadiness{ + readyStatus: metav1.ConditionFalse, + readyReason: reason, + readyMessage: message, + progressingStatus: metav1.ConditionTrue, + progressingReason: reason, + progressingMessage: message, + degradedStatus: metav1.ConditionFalse, + degradedReason: reasonNotDegraded, + degradedMessage: "backend is not in a degraded state", + } +} + +func engineLocalDegraded(reason, message string) engineLocalReadiness { + return engineLocalReadiness{ + readyStatus: metav1.ConditionFalse, + readyReason: reason, + readyMessage: message, + progressingStatus: metav1.ConditionFalse, + progressingReason: "Degraded", + progressingMessage: message, + degradedStatus: metav1.ConditionTrue, + degradedReason: reason, + degradedMessage: message, + } +} + +func podDiagnostic(format string, podNames []string, total int) string { + sort.Strings(podNames) + return truncateMessage(fmt.Sprintf(format, len(podNames), total, strings.Join(podNames, ", "))) +} diff --git a/internal/controller/cachebackend_engine_local_readiness_test.go b/internal/controller/cachebackend_engine_local_readiness_test.go new file mode 100644 index 00000000..b684a999 --- /dev/null +++ b/internal/controller/cachebackend_engine_local_readiness_test.go @@ -0,0 +1,393 @@ +package controller + +import ( + "context" + "errors" + "strconv" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + cachev1alpha1 "github.com/cachebox-project/inference-cache/api/v1alpha1" + podwebhook "github.com/cachebox-project/inference-cache/internal/webhook/pod" + sglangadapter "github.com/cachebox-project/inference-cache/pkg/adapters/runtime/sglang" +) + +func TestEvaluateEngineLocalReadiness(t *testing.T) { + backend := engineLocalBackendFixture() + adapter := sglangadapter.NewHiCacheAdapter() + current := func(name string) corev1.Pod { + return engineLocalPodFixture(name, backend, backend.Generation, true) + } + + terminating := current("terminating") + now := metav1.NewTime(time.Now()) + terminating.DeletionTimestamp = &now + failed := current("failed") + failed.Status.Phase = corev1.PodFailed + + skipped := current("skipped") + skipped.Annotations = map[string]string{ + podwebhook.AnnotationSkip: "true", + podwebhook.AnnotationInjectSkipped: podwebhook.InjectSkippedReasonSkipAnnotation, + } + skippedWithoutReceipt := skipped.DeepCopy() + skippedWithoutReceipt.Name = "skipped-without-receipt" + delete(skippedWithoutReceipt.Annotations, podwebhook.AnnotationInjectSkipped) + + missingUID := current("missing-uid") + delete(missingUID.Annotations, podwebhook.AnnotationInjectedByUID) + missingGeneration := current("missing-generation") + delete(missingGeneration.Annotations, podwebhook.AnnotationInjectedGeneration) + malformedGeneration := current("malformed-generation") + malformedGeneration.Annotations[podwebhook.AnnotationInjectedGeneration] = "not-a-number" + wrongOwner := current("wrong-owner") + wrongOwner.Annotations[podwebhook.AnnotationInjectedBy] = backend.Namespace + "/other" + futureGeneration := current("future-generation") + futureGeneration.Annotations[podwebhook.AnnotationInjectedGeneration] = + strconv.FormatInt(backend.Generation+1, 10) + stale := engineLocalPodFixture("stale", backend, backend.Generation-1, true) + unavailable := current("unavailable") + unavailable.Status.Conditions[0].Status = corev1.ConditionFalse + forgedReceipt := current("forged-current-receipt") + forgedReceipt.Spec.Containers[0].Args = nil + + tests := []struct { + name string + pods []corev1.Pod + ready metav1.ConditionStatus + readyReason string + progressing metav1.ConditionStatus + degraded metav1.ConditionStatus + messageHas string + }{ + { + name: "no matching pods waits", + ready: metav1.ConditionFalse, + readyReason: reasonAwaitingEnginePods, + progressing: metav1.ConditionTrue, + degraded: metav1.ConditionFalse, + }, + { + name: "terminal and terminating pods are ignored", + pods: []corev1.Pod{terminating, failed}, + ready: metav1.ConditionFalse, + readyReason: reasonAwaitingEnginePods, + progressing: metav1.ConditionTrue, + degraded: metav1.ConditionFalse, + }, + { + name: "all skipped is ready", + pods: []corev1.Pod{skipped}, + ready: metav1.ConditionTrue, + readyReason: reasonAllEnginePodsSkipped, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionFalse, + }, + { + name: "explicit skip does not require a webhook-authored receipt", + pods: []corev1.Pod{*skippedWithoutReceipt}, + ready: metav1.ConditionTrue, + readyReason: reasonAllEnginePodsSkipped, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionFalse, + }, + { + name: "skipped pod does not block a ready participant", + pods: []corev1.Pod{skipped, current("ready")}, + ready: metav1.ConditionTrue, + readyReason: reasonEnginePodsReady, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionFalse, + messageHas: "1 additional matching Pods explicitly skipped", + }, + { + name: "missing receipt degrades", + pods: []corev1.Pod{missingUID}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "missing-uid", + }, + { + name: "malformed generation degrades as not injected", + pods: []corev1.Pod{malformedGeneration}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "malformed-generation", + }, + { + name: "missing generation degrades as not injected", + pods: []corev1.Pod{missingGeneration}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "missing-generation", + }, + { + name: "missing receipt takes precedence over mismatched receipt", + pods: []corev1.Pod{wrongOwner, missingUID}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "missing-uid", + }, + { + name: "wrong owner degrades", + pods: []corev1.Pod{wrongOwner}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsInjectionMismatch, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "wrong-owner", + }, + { + name: "future generation degrades", + pods: []corev1.Pod{futureGeneration}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsInjectionMismatch, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "future-generation", + }, + { + name: "forged current receipt without engine config degrades", + pods: []corev1.Pod{forgedReceipt}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "forged-current-receipt", + }, + { + name: "stale generation is progressing even while a current pod is unavailable", + pods: []corev1.Pod{stale, unavailable}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsRolloutInProgress, + progressing: metav1.ConditionTrue, + degraded: metav1.ConditionFalse, + messageHas: "stale", + }, + { + name: "current generation unavailable degrades", + pods: []corev1.Pod{unavailable}, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsUnavailable, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionTrue, + messageHas: "unavailable", + }, + { + name: "all current generation pods ready", + pods: []corev1.Pod{current("b"), current("a")}, + ready: metav1.ConditionTrue, + readyReason: reasonEnginePodsReady, + progressing: metav1.ConditionFalse, + degraded: metav1.ConditionFalse, + messageHas: "2/2 engine Pods", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := evaluateEngineLocalReadiness(backend, tc.pods, adapter) + if got.readyStatus != tc.ready || got.readyReason != tc.readyReason { + t.Fatalf("Ready = %s/%s, want %s/%s; verdict=%+v", + got.readyStatus, got.readyReason, tc.ready, tc.readyReason, got) + } + if got.progressingStatus != tc.progressing { + t.Fatalf("Progressing = %s/%s, want %s; verdict=%+v", + got.progressingStatus, got.progressingReason, tc.progressing, got) + } + if got.degradedStatus != tc.degraded { + t.Fatalf("Degraded = %s/%s, want %s; verdict=%+v", + got.degradedStatus, got.degradedReason, tc.degraded, got) + } + if tc.messageHas != "" && !strings.Contains(got.readyMessage, tc.messageHas) { + t.Fatalf("Ready message %q does not contain %q", got.readyMessage, tc.messageHas) + } + }) + } +} + +func TestEvaluateEngineLocalReadinessRequiresEngineOverrides(t *testing.T) { + backend := engineLocalBackendFixture() + backend.Spec.Integration.EngineOverrides = &cachev1alpha1.EngineInjectionOverrides{ + Args: []string{"--custom-hicache-mode=enabled"}, + Env: []corev1.EnvVar{{Name: "HICACHE_CUSTOM_MODE", Value: "enabled"}}, + } + adapter := sglangadapter.NewHiCacheAdapter() + + configured := engineLocalPodFixture("configured", backend, backend.Generation, true) + missingArg := configured.DeepCopy() + missingArg.Name = "missing-override-arg" + missingArg.Spec.Containers[0].Args = nil + for _, arg := range configured.Spec.Containers[0].Args { + if arg != "--custom-hicache-mode=enabled" { + missingArg.Spec.Containers[0].Args = append(missingArg.Spec.Containers[0].Args, arg) + } + } + missingEnv := configured.DeepCopy() + missingEnv.Name = "missing-override-env" + missingEnv.Spec.Containers[0].Env = nil + for _, env := range configured.Spec.Containers[0].Env { + if env.Name != "HICACHE_CUSTOM_MODE" { + missingEnv.Spec.Containers[0].Env = append(missingEnv.Spec.Containers[0].Env, env) + } + } + + tests := []struct { + name string + pod corev1.Pod + ready metav1.ConditionStatus + readyReason string + }{ + { + name: "complete overrides are ready", + pod: configured, + ready: metav1.ConditionTrue, + readyReason: reasonEnginePodsReady, + }, + { + name: "missing override arg is not injected", + pod: *missingArg, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + }, + { + name: "missing override env is not injected", + pod: *missingEnv, + ready: metav1.ConditionFalse, + readyReason: reasonEnginePodsNotInjected, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := evaluateEngineLocalReadiness(backend, []corev1.Pod{tc.pod}, adapter) + if got.readyStatus != tc.ready || got.readyReason != tc.readyReason { + t.Fatalf("Ready = %s/%s, want %s/%s; verdict=%+v", + got.readyStatus, got.readyReason, tc.ready, tc.readyReason, got) + } + }) + } +} + +func TestReconcileEngineLocalPreservesReadinessOnPodListError(t *testing.T) { + scheme := newScheme(t) + backend := engineLocalBackendFixture() + backend.Generation = 2 + backend.Status.ObservedGeneration = 1 + meta.SetStatusCondition(&backend.Status.Conditions, metav1.Condition{ + Type: conditionTypeReady, + Status: metav1.ConditionTrue, + Reason: reasonEnginePodsReady, + Message: "last successful observation", + ObservedGeneration: 1, + }) + + listErr := errors.New("synthetic engine Pod list failure") + funcs := interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*corev1.PodList); ok { + return listErr + } + return c.List(ctx, list, opts...) + }, + } + r := newReconcilerWithInterceptor(scheme, funcs, backend) + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: backend.Name, Namespace: backend.Namespace}, + }) + if err == nil || !strings.Contains(err.Error(), listErr.Error()) { + t.Fatalf("Reconcile error = %v, want wrapped Pod list error", err) + } + + got := getBackend(t, r, backend.Name, backend.Namespace) + if got.Status.ObservedGeneration != 1 { + t.Fatalf("observedGeneration = %d, want preserved 1", got.Status.ObservedGeneration) + } + ready := meta.FindStatusCondition(got.Status.Conditions, conditionTypeReady) + if ready == nil || ready.Status != metav1.ConditionTrue || + ready.Reason != reasonEnginePodsReady || + ready.Message != "last successful observation" { + t.Fatalf("Ready = %+v, want last successful observation preserved", ready) + } +} + +func engineLocalBackendFixture() *cachev1alpha1.CacheBackend { + return &cachev1alpha1.CacheBackend{ + ObjectMeta: metav1.ObjectMeta{ + Name: "hicache", + Namespace: "engines", + UID: types.UID("hicache-uid"), + Generation: 3, + }, + Spec: cachev1alpha1.CacheBackendSpec{ + Runtime: cachev1alpha1.CacheBackendRuntimeSGLang, + Type: cachev1alpha1.CacheBackendTypeSGLangHiCache, + Integration: &cachev1alpha1.CacheBackendIntegrationSpec{ + Mode: cachev1alpha1.CacheBackendIntegrationModeOffload, + Role: cachev1alpha1.CacheBackendIntegrationRoleReadWrite, + FailOpen: ptrBool(true), + }, + EngineSelector: &cachev1alpha1.CacheBackendEngineSelector{ + MatchLabels: map[string]string{"app": "sglang"}, + }, + HiCache: &cachev1alpha1.SGLangHiCacheSpec{Ratio: "2"}, + }, + } +} + +func engineLocalPodFixture( + name string, + backend *cachev1alpha1.CacheBackend, + generation int64, + ready bool, +) corev1.Pod { + readyStatus := corev1.ConditionFalse + if ready { + readyStatus = corev1.ConditionTrue + } + pod := corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: backend.Namespace, + Labels: map[string]string{"app": "sglang"}, + Annotations: map[string]string{ + podwebhook.AnnotationInjectedBy: backend.Namespace + "/" + backend.Name, + podwebhook.AnnotationInjectedByUID: string(backend.UID), + podwebhook.AnnotationInjectedGeneration: strconv.FormatInt(generation, 10), + }, + }, + Spec: corev1.PodSpec{ + Containers: []corev1.Container{{Name: "sglang"}}, + }, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReady, + Status: readyStatus, + }}, + }, + } + adapter := sglangadapter.NewHiCacheAdapter() + if err := podwebhook.ApplyEngineConfigWithOverrides(adapter, &pod.Spec, nil, backend); err != nil { + panic(err) + } + return pod +} diff --git a/internal/webhook/pod/envtest_integration_test.go b/internal/webhook/pod/envtest_integration_test.go index c3e859f6..39daa4b3 100644 --- a/internal/webhook/pod/envtest_integration_test.go +++ b/internal/webhook/pod/envtest_integration_test.go @@ -193,6 +193,10 @@ func TestWebhookOnEnvtest_EndToEnd(t *testing.T) { t.Fatalf("annotation %s: got %q want %q (live CacheBackend UID)", AnnotationInjectedByUID, got.Annotations[AnnotationInjectedByUID], string(cb.UID)) } + if got.Annotations[AnnotationInjectedGeneration] != fmt.Sprint(cb.Generation) { + t.Fatalf("annotation %s: got %q want %q (live CacheBackend generation)", + AnnotationInjectedGeneration, got.Annotations[AnnotationInjectedGeneration], fmt.Sprint(cb.Generation)) + } if !containsArgFlag(got.Spec.Containers[0].Args, "--kv-transfer-config") { t.Fatalf("--kv-transfer-config flag not injected; args = %v", got.Spec.Containers[0].Args) } @@ -228,6 +232,7 @@ func TestWebhookOnEnvtest_EndToEnd(t *testing.T) { // the apiserver would actually see it. delete(pod2.Annotations, AnnotationInjectedBy) delete(pod2.Annotations, AnnotationInjectedByUID) + delete(pod2.Annotations, AnnotationInjectedGeneration) if err := mgr.GetClient().Create(ctx, pod2); err != nil { t.Fatalf("create second Pod: %v", err) } diff --git a/internal/webhook/pod/overrides.go b/internal/webhook/pod/overrides.go index c851b26e..1240d655 100644 --- a/internal/webhook/pod/overrides.go +++ b/internal/webhook/pod/overrides.go @@ -7,8 +7,55 @@ import ( corev1 "k8s.io/api/core/v1" cachev1alpha1 "github.com/cachebox-project/inference-cache/api/v1alpha1" + backendadapter "github.com/cachebox-project/inference-cache/pkg/adapters/backend" + adapterruntime "github.com/cachebox-project/inference-cache/pkg/adapters/runtime" ) +// ApplyEngineConfigWithOverrides applies the complete engine-container +// mutation contract shared by Pod admission and readiness verification: +// canonical runtime-adapter injection followed by +// spec.integration.engineOverrides. Keeping the two stages together prevents +// a readiness check from declaring a Pod converged after validating only the +// adapter's canonical args/env. +// +// EventsOnly deliberately leaves the engine container untouched, including +// engineOverrides, because that mode wires observation only and no KV +// connector. +func ApplyEngineConfigWithOverrides( + adapter adapterruntime.KVCacheRuntimeAdapter, + pod *corev1.PodSpec, + binding *backendadapter.Binding, + cache *cachev1alpha1.CacheBackend, +) error { + if cache.Spec.IsEventsOnly() { + return nil + } + + overrides := engineOverridesFor(cache) + overrideIdx := -1 + var preArgs []string + var preEnv []corev1.EnvVar + if overrides != nil { + if idx, ok := overrideTargetIndex(pod.Containers, adapter.EngineContainerName()); ok { + overrideIdx = idx + preArgs = append([]string(nil), pod.Containers[idx].Args...) + preEnv = append([]corev1.EnvVar(nil), pod.Containers[idx].Env...) + } + } + + if err := adapterruntime.InjectEngineConfigWithBinding(adapter, pod, binding, cache); err != nil { + return err + } + if overrides != nil && overrideIdx >= 0 { + pod.Containers[overrideIdx].Args, pod.Containers[overrideIdx].Env = applyEngineInjectionOverrides( + preArgs, pod.Containers[overrideIdx].Args, + preEnv, pod.Containers[overrideIdx].Env, + overrides, + ) + } + return nil +} + // applyEngineInjectionOverrides amends the engine container's args/env // produced by the runtime adapter, scoped to the entries the adapter // itself injected. The webhook snapshots the container BEFORE diff --git a/internal/webhook/pod/podinjector.go b/internal/webhook/pod/podinjector.go index 88f903d4..84a7bcc3 100644 --- a/internal/webhook/pod/podinjector.go +++ b/internal/webhook/pod/podinjector.go @@ -67,6 +67,14 @@ const AnnotationInjectedBy = "inferencecache.io/injected-by" // the event. const AnnotationInjectedByUID = "inferencecache.io/injected-by-uid" +// AnnotationInjectedGeneration is stamped alongside [AnnotationInjectedBy] +// and [AnnotationInjectedByUID] on every successful injection. It records the +// CacheBackend metadata.generation whose spec the runtime adapter validated +// and rendered into the pod. Engine-local readiness uses the complete +// name/UID/generation receipt to distinguish current pods from pods that still +// carry an older CacheBackend configuration. +const AnnotationInjectedGeneration = "inferencecache.io/injected-generation" + // AnnotationInjectSkipped is stamped when the webhook intentionally skips // injection because the operator set [AnnotationSkip]. It lets a persisted pod // distinguish an explicit opt-out from selector drift or fail-open admission. @@ -243,65 +251,14 @@ func (h *EngineInjector) Handle(ctx context.Context, req admission.Request) admi // free at the apiserver. mutated := pod.DeepCopy() - // Snapshot the engine container's pre-injection args/env so the - // override merge below can scope itself to the adapter-owned set. The - // override surface mutates only what InjectEngineConfig contributes; - // user pod-template args/env that the adapter does not touch stay - // protected. We snapshot before the adapter call (rather than re-deriving - // the canonical set afterwards) so this works for any adapter without - // changing the [adapterruntime.KVCacheRuntimeAdapter] contract. - overrides := engineOverridesFor(cache) - overrideIdx := -1 - var preArgs []string - var preEnv []corev1.EnvVar - if overrides != nil { - if idx, ok := overrideTargetIndex(mutated.Spec.Containers, adapter.EngineContainerName()); ok { - overrideIdx = idx - preArgs = append([]string(nil), mutated.Spec.Containers[idx].Args...) - preEnv = append([]corev1.EnvVar(nil), mutated.Spec.Containers[idx].Env...) - } - } - - // Events-only (tier-1 routing) wires NO KV connector regardless of - // spec.type: the engine container is left untouched so a hybrid-attention - // model's KV-cache manager is not disabled by a connector it cannot load. - // Skip InjectEngineConfig here, at the webhook, rather than relying on each - // adapter's own no-op: the connector no-op currently lives ONLY in the - // vLLM+LMCache adapter, so an admission-bypassed spec.type=External + - // mode=EventsOnly object would otherwise select the External adapter and - // inject the LMCache connector — violating the events-only "no connector" - // contract. Gating here makes the no-connector guarantee adapter-independent. - // The observation-sidecar append and the wired/injected-by logic below stay - // as-is (events-only's only wiring is the subscriber sidecar). - if !cache.Spec.IsEventsOnly() { - if err := adapterruntime.InjectEngineConfigWithBinding(adapter, &mutated.Spec, binding, cache); err != nil { - log.V(1).Info("fail-open: adapter rejected pod", - "runtime", string(runtimeID), "error", err.Error()) - return failOpen(req, &pod, fmt.Sprintf("adapter rejected pod (fail-open): %v", err)) - } - } - - // Apply spec.integration.engineOverrides scoped to the adapter-owned - // args/env derived from the pre/post diff. Admission has already - // hard-rejected overrides that overlap the adapter's reserved - // declarations, so the entries surviving to this point are safe to - // merge. Adapters with no canonical engine container (the reference - // adapter) return EngineContainerName() == "" and overrideIdx stays - // -1, so the merge is skipped — the override surface is for production - // adapters that target a specific engine container. - // - // Skip the merge entirely for events-only: InjectEngineConfig injected - // nothing in this mode (it is a no-op), so the engine container is left - // untouched by contract. Running the override merge here would let the - // override surface append non-adapter-owned args/env to the engine - // container even though the canonical injection contributed none — - // contradicting "the engine container is left otherwise untouched". - if overrides != nil && overrideIdx >= 0 && !cache.Spec.IsEventsOnly() { - mutated.Spec.Containers[overrideIdx].Args, mutated.Spec.Containers[overrideIdx].Env = applyEngineInjectionOverrides( - preArgs, mutated.Spec.Containers[overrideIdx].Args, - preEnv, mutated.Spec.Containers[overrideIdx].Env, - overrides, - ) + // Apply the same complete engine-container mutation pipeline that the + // readiness controller later replays as an idempotence check. The helper + // keeps canonical adapter injection and engineOverrides inseparable and + // preserves the EventsOnly no-connector/no-override contract. + if err := ApplyEngineConfigWithOverrides(adapter, &mutated.Spec, binding, cache); err != nil { + log.V(1).Info("fail-open: adapter rejected pod", + "runtime", string(runtimeID), "error", err.Error()) + return failOpen(req, &pod, fmt.Sprintf("adapter rejected pod (fail-open): %v", err)) } // Inject the kernel-check init container (adapters that opt in via the @@ -393,7 +350,8 @@ func (h *EngineInjector) Handle(ctx context.Context, req admission.Request) admi // which is skipped when the subscriber image / backendConfig.model is unset // (nothing injected) OR when a same-named container already exists (operator- // authored, unverified). In those cases the webhook added/verified no wiring, - // so stamping injected-by/injected-by-uid would trip the downstream + // so stamping the injected-by/injected-by-uid/injected-generation receipt + // would trip the downstream // InjectedByCacheBackend event controller on a non-existent injection and // report "wired" while no usable events may flow. Route that case through the // fail-open no-injection path (which strips any forged injection annotations @@ -411,6 +369,7 @@ func (h *EngineInjector) Handle(ctx context.Context, req admission.Request) admi delete(mutated.Annotations, AnnotationInjectSkipped) mutated.Annotations[AnnotationInjectedBy] = cache.Namespace + "/" + cache.Name mutated.Annotations[AnnotationInjectedByUID] = string(cache.UID) + mutated.Annotations[AnnotationInjectedGeneration] = strconv.FormatInt(cache.Generation, 10) mutatedRaw, err := json.Marshal(mutated) if err != nil { @@ -508,31 +467,33 @@ func (h *EngineInjector) logger(ctx context.Context) logr.Logger { } // failOpen builds the admission response for any fail-open return path -// AFTER the pod has been decoded. The webhook's contract is that -// AnnotationInjectedBy and AnnotationInjectSkipped on the persisted pod mean -// "the webhook successfully made this decision" — those are what the -// engine-pod-events controller keys `InjectedByCacheBackend` and +// AFTER the pod has been decoded. The webhook's contract is that the complete +// name/UID/generation injection receipt, or AnnotationInjectSkipped, on the +// persisted pod means "the webhook successfully made this decision" — those +// are what the engine-pod-events controller keys `InjectedByCacheBackend` and // `SkippedByOperator` off of. The annotations are user-controllable (anyone // with pod-create RBAC can set them) and the webhook does not overwrite them // on fail-open paths, so a copy/paste from a mutated pod's metadata, or an // attacker forging the annotations, would otherwise trip the controller into // emitting an event for a pod the webhook never touched. // -// Fix: on every fail-open return, strip the annotation if it was -// preset. Steady-state cost stays at zero patches per pod for the common -// no-match case (the vast majority of pods cluster-wide), because the -// helper short-circuits to admission.Allowed when the annotation is -// absent. +// Fix: on every fail-open return, strip any injection-receipt or verified-skip +// annotations that were preset. Steady-state cost stays at zero patches per +// pod for the common no-match case (the vast majority of pods cluster-wide), +// because the helper short-circuits to admission.Allowed when the annotations +// are absent. func failOpen(req admission.Request, pod *corev1.Pod, reason string) admission.Response { hasInjectedBy := pod.Annotations[AnnotationInjectedBy] != "" hasInjectedByUID := pod.Annotations[AnnotationInjectedByUID] != "" + hasInjectedGeneration := pod.Annotations[AnnotationInjectedGeneration] != "" hasInjectSkipped := pod.Annotations[AnnotationInjectSkipped] != "" - if !hasInjectedBy && !hasInjectedByUID && !hasInjectSkipped { + if !hasInjectedBy && !hasInjectedByUID && !hasInjectedGeneration && !hasInjectSkipped { return admission.Allowed(reason) } cleared := pod.DeepCopy() delete(cleared.Annotations, AnnotationInjectedBy) delete(cleared.Annotations, AnnotationInjectedByUID) + delete(cleared.Annotations, AnnotationInjectedGeneration) delete(cleared.Annotations, AnnotationInjectSkipped) if len(cleared.Annotations) == 0 { // Avoid emitting an empty-map annotations field; absent is the @@ -558,11 +519,13 @@ func skipInjection(req admission.Request, pod *corev1.Pod) admission.Response { } delete(mutated.Annotations, AnnotationInjectedBy) delete(mutated.Annotations, AnnotationInjectedByUID) + delete(mutated.Annotations, AnnotationInjectedGeneration) mutated.Annotations[AnnotationInjectSkipped] = InjectSkippedReasonSkipAnnotation if pod.Annotations[AnnotationInjectSkipped] == InjectSkippedReasonSkipAnnotation && pod.Annotations[AnnotationInjectedBy] == "" && - pod.Annotations[AnnotationInjectedByUID] == "" { + pod.Annotations[AnnotationInjectedByUID] == "" && + pod.Annotations[AnnotationInjectedGeneration] == "" { return admission.Allowed("skipped via " + AnnotationSkip) } raw, err := json.Marshal(mutated) diff --git a/internal/webhook/pod/podinjector_test.go b/internal/webhook/pod/podinjector_test.go index f55999b8..995c0d1e 100644 --- a/internal/webhook/pod/podinjector_test.go +++ b/internal/webhook/pod/podinjector_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "reflect" + "strconv" "strings" "testing" @@ -155,9 +156,10 @@ func sglangEnginePod(name string, labels map[string]string) *corev1.Pod { func readyCacheBackend(name, namespace string, selector map[string]string) *cachev1alpha1.CacheBackend { return &cachev1alpha1.CacheBackend{ ObjectMeta: metav1.ObjectMeta{ - Name: name, - Namespace: namespace, - UID: types.UID("cb-" + namespace + "-" + name + "-uid"), + Name: name, + Namespace: namespace, + UID: types.UID("cb-" + namespace + "-" + name + "-uid"), + Generation: 7, }, Spec: cachev1alpha1.CacheBackendSpec{ Type: cachev1alpha1.CacheBackendTypeLMCache, @@ -232,6 +234,9 @@ func TestHandle_MatchAndInject(t *testing.T) { if got, want := mutated.Annotations[AnnotationInjectedByUID], string(cb.UID); got != want { t.Fatalf("annotation %s: got %q, want %q (matched CR UID)", AnnotationInjectedByUID, got, want) } + if got, want := mutated.Annotations[AnnotationInjectedGeneration], strconv.FormatInt(cb.Generation, 10); got != want { + t.Fatalf("annotation %s: got %q, want %q (matched CR generation)", AnnotationInjectedGeneration, got, want) + } mustHaveArgPair(t, mutated, "--model", "Qwen/Qwen2.5-0.5B-Instruct") mustHaveArgFlag(t, mutated, "--kv-transfer-config") } @@ -1997,9 +2002,10 @@ func TestHandle_SkipAnnotationStampsSkippedReasonAndClearsInjectedBy(t *testing. h := newHandler(t, cb) pod := vllmEnginePod("engine-a", map[string]string{"app": "vllm"}) pod.Annotations = map[string]string{ - AnnotationSkip: "true", - AnnotationInjectedBy: ns + "/" + cb.Name, - AnnotationInjectedByUID: string(cb.UID), + AnnotationSkip: "true", + AnnotationInjectedBy: ns + "/" + cb.Name, + AnnotationInjectedByUID: string(cb.UID), + AnnotationInjectedGeneration: strconv.FormatInt(cb.Generation, 10), } req := newRequest(t, pod, ns) @@ -2017,6 +2023,9 @@ func TestHandle_SkipAnnotationStampsSkippedReasonAndClearsInjectedBy(t *testing. if got := mutated.Annotations[AnnotationInjectedByUID]; got != "" { t.Fatalf("annotation %s = %q, want cleared on skip path", AnnotationInjectedByUID, got) } + if got := mutated.Annotations[AnnotationInjectedGeneration]; got != "" { + t.Fatalf("annotation %s = %q, want cleared on skip path", AnnotationInjectedGeneration, got) + } } func mustHaveEnv(t *testing.T, pod *corev1.Pod, name, value string) { @@ -2222,7 +2231,11 @@ func TestHandle_FailOpenClearsForgedInjectedByAnnotation(t *testing.T) { } pod := vllmEnginePod("forger", tc.labels) - pod.Annotations = map[string]string{AnnotationInjectedBy: ns + "/totally-not-a-real-cb"} + pod.Annotations = map[string]string{ + AnnotationInjectedBy: ns + "/totally-not-a-real-cb", + AnnotationInjectedByUID: "forged-uid", + AnnotationInjectedGeneration: "999", + } req := newRequest(t, pod, ns) resp := h.Handle(context.Background(), req) @@ -2237,6 +2250,12 @@ func TestHandle_FailOpenClearsForgedInjectedByAnnotation(t *testing.T) { if got := mutated.Annotations[AnnotationInjectedBy]; got != "" { t.Fatalf("forged %s annotation survived fail-open: got %q, want \"\"", AnnotationInjectedBy, got) } + if got := mutated.Annotations[AnnotationInjectedByUID]; got != "" { + t.Fatalf("forged %s annotation survived fail-open: got %q, want \"\"", AnnotationInjectedByUID, got) + } + if got := mutated.Annotations[AnnotationInjectedGeneration]; got != "" { + t.Fatalf("forged %s annotation survived fail-open: got %q, want \"\"", AnnotationInjectedGeneration, got) + } }) } }