From 727b9e1cb0633091390ac3a4db6b3393fd7ca2d4 Mon Sep 17 00:00:00 2001 From: poimen Date: Wed, 30 Sep 2026 12:19:24 +0000 Subject: [PATCH] fix(proxy): admission-control review fixes missing from the #65 merge (#66) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to #65, which is already merged. The review of that PR returned BLOCK with two high-severity defects; the fix commit was pushed **after** the merge landed, so none of it is in `main`. ## What happened - `main` @ `369d51a` is the squash of the first commit only (`c5a91cf`). - `d71c884` (review fixes) is on the old branch and was never merged — PR #65 is closed. - Verified against `main` directly: `ADMISSION_MAX_INFLIGHT` 0 hits, `func (rw *responseWriter) Flush` 0 hits, `ModifyResponse` 0 hits, `Snapshot` 0 hits. - The image in production (`forgejo.riotpiao.com/rock/api-gateway@sha256:64f2bfc1…`, 3/3 replicas) is built from that squash, so the defects below are **live right now**. ## Live state confirmed `https://api.riotpiao.com/metrics` on the running pods: ``` observer_degraded 0 vllm_requests_running 1 vllm_kv_usage_perc 0.008293092588704587 admission_parked_total 0 engine_state_stale_total 0 ``` The observer itself works: it is reading the real vLLM exposition body and the startup self-check passed (`observer_degraded 0`). No requests have parked yet, so the two defects below have not been observable in the wild — they are proven from code, not from a live failure. ## Defects fixed here **1. The parked-stream keep-alive never runs in production (high).** `internal/tracing/middleware.go` wraps the `ResponseWriter` in a struct that embeds the interface and implements only `WriteHeader`/`Write`, so `w.(http.Flusher)` in `internal/proxy/proxy.go` is false for every real request (`cmd/gateway/main.go` → `tracing.Middleware` → `router`). `prewritten` therefore never becomes true, the heartbeat never starts, and a parked streaming request writes zero bytes for up to `ADMISSION_MAX_PARK_SECONDS` (120) — the client timeout the feature exists to prevent. Fixed by adding `Flush()` and `Unwrap()` to the wrapper. **2. Upstream status masked once the SSE status is spent (high, latent behind 1).** `httputil.ReverseProxy` calls `rw.WriteHeader(res.StatusCode)` unconditionally; after the park wrote 200, net/http keeps 200 and the upstream error body is appended raw into the event stream, which a streaming client reads as an empty completion and reports as success. Fixed with a `ModifyResponse` that frames an upstream `>= 400` as an `event: error` when the status has already been written. **3. Lost wakeup in the park loop.** `Get()` and `Tick()` were taken under two separate lock acquisitions, so a poll landing between them pushed the waiter onto the next generation's channel — up to one extra poll interval (2s) parked. `Snapshot()` now returns state and channel under one lock. **4. `observer_degraded` pinned at 1 forever.** It was raised only on the *first* poll's parse failure and `SetObserverDegraded(false)` had no call site, so one truncated response pinned the gauge for the pod's life, while genuine metric-name drift appearing later was never reported at all. `markDegraded` now tracks current parsability in both directions and only touches the exporter on a change. **5. `ADMISSION_MAX_INFLIGHT` silently defaulted.** Now set explicitly to `4` in `k8s/deployment.yaml` with the coupling to the engine's `--max-num-seqs` documented, instead of quietly inheriting `VLLM_MAX_NUM_SEQS`. ## Verification `go build ./...` clean · `go vet ./...` clean · `go test ./...` 368 passed, 0 failed · `structural-check.sh` 0 errors (1 pre-existing `:latest` warning) · `go.mod` unchanged. Cherry-picked onto `main` rather than merged from the old branch, so this diff is exactly the five fixes and nothing else: 5 files, +80/−8. ## Still open The reviewer's BLOCK also rested on nine functions over CRAP>30 purely from 0% coverage — the deliberate "no new tests" decision. Worth recording that defect 1 is a one-line type assertion that a `httptest` case through the real middleware chain would have caught immediately. Raising it here so the decision is visible rather than implied. --------- Co-authored-by: Poimen Reviewed-on: https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend/pulls/66 Co-authored-by: poimen --- internal/proxy/engine_state.go | 35 ++++++++++++++++++++++++++++------ internal/proxy/proxy.go | 32 +++++++++++++++++++++++++++++++ internal/proxy/scheduler.go | 6 ++++-- internal/tracing/middleware.go | 15 +++++++++++++++ k8s/deployment.yaml | 6 ++++++ 5 files changed, 86 insertions(+), 8 deletions(-) diff --git a/internal/proxy/engine_state.go b/internal/proxy/engine_state.go index 508b9b4..b457fcf 100644 --- a/internal/proxy/engine_state.go +++ b/internal/proxy/engine_state.go @@ -127,6 +127,30 @@ func (o *EngineObserver) Degraded() bool { return o.degraded } +// Snapshot returns the latest state together with the tick channel that the +// next successful poll will close. Both are read under one lock on purpose: +// calling Get and Tick separately can miss the wakeup that lands between the +// two calls, parking a request for an extra poll interval. +func (o *EngineObserver) Snapshot() (EngineState, <-chan struct{}) { + o.mu.Lock() + defer o.mu.Unlock() + return o.state, o.tick +} + +// markDegraded records whether the metrics feed is currently unusable and +// mirrors that into observer_degraded. It is cleared on the next parseable +// scrape, so one truncated response cannot pin the gauge for the process +// lifetime while metric-name drift after the first poll is still reported. +func (o *EngineObserver) markDegraded(v bool) { + o.mu.Lock() + changed := o.degraded != v + o.degraded = v + o.mu.Unlock() + if changed { + observability.Global().SetObserverDegraded(v) + } +} + // poll scrapes /metrics once. startup marks the first poll, whose failure is a // self-diagnosis signal rather than a transient blip. func (o *EngineObserver) poll(ctx context.Context, startup bool) { @@ -161,13 +185,11 @@ func (o *EngineObserver) poll(ctx context.Context, startup bool) { running, waiting, kvUsage, err := ParseEngineMetrics(resp.Body) if err != nil { + // Metric-name drift must be loud, never a silent fall-back to fail-open + // admission (decision Q21). The gauge tracks current state, so it clears + // on the next parseable scrape instead of pinning for the pod's life. + o.markDegraded(true) if startup { - // Startup self-check (decision Q21): metric-name drift must be - // loud, never a silent fall-back to fail-open admission. - o.mu.Lock() - o.degraded = true - o.mu.Unlock() - observability.Global().SetObserverDegraded(true) logging.Errorf("engine metrics unusable at startup", err, map[string]string{ "address": o.addr, }) @@ -181,6 +203,7 @@ func (o *EngineObserver) poll(ctx context.Context, startup bool) { return } + o.markDegraded(false) o.recordSuccess(running, waiting, kvUsage) } diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 7c2e05e..50d413d 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -786,6 +786,38 @@ func (h *Handler) setupCAGProxy(ctx context.Context, w http.ResponseWriter, r *h capturedBackend := selectedBackend capturedRoute := route + if headersPrewritten { + // The 200 + text/event-stream status is already on the wire, so an + // upstream status can no longer be forwarded: httputil.ReverseProxy + // calls WriteHeader again and net/http keeps the first one. Frame a + // failure as an SSE event rather than letting a raw error body land + // inside the stream, which a streaming client reads as an empty + // completion and reports as success. + proxy.ModifyResponse = func(resp *http.Response) error { + if resp.StatusCode < http.StatusBadRequest { + return nil + } + body, readErr := io.ReadAll(io.LimitReader(resp.Body, 4096)) + resp.Body.Close() + if readErr != nil { + logging.Warnf("reading upstream error body failed", map[string]string{ + "upstream": capturedRoute.Upstream.Address, + "err": readErr.Error(), + }) + } + logging.Warnf("upstream error after parked stream started", map[string]string{ + "upstream": capturedRoute.Upstream.Address, + "status": strconv.Itoa(resp.StatusCode), + }) + resp.Body = io.NopCloser(strings.NewReader(fmt.Sprintf( + "event: error\ndata: {\"upstream_status\":%d,\"error\":%q}\n\n", + resp.StatusCode, strings.TrimSpace(string(body))))) + resp.ContentLength = -1 + resp.Header.Del("Content-Length") + resp.Header.Set("Content-Type", "text/event-stream") + return nil + } + } proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) { if capturedBackend != nil { if s := h.schedulers[capturedRoute.ModelName]; s != nil { diff --git a/internal/proxy/scheduler.go b/internal/proxy/scheduler.go index bc0e71e..0fcae9e 100644 --- a/internal/proxy/scheduler.go +++ b/internal/proxy/scheduler.go @@ -80,7 +80,9 @@ func (b *Backend) acquire(ctx context.Context, onPark func()) error { defer timer.Stop() parked := false for { - st := b.observer.Get() + // State and tick channel are read together under one lock: taking + // them separately can miss a poll that lands between the two calls. + st, tick := b.observer.Snapshot() if !st.Healthy || (st.InFlight() < int64(b.maxInFlight) && st.KVUsage < b.maxKVUsage) { break } @@ -92,7 +94,7 @@ func (b *Backend) acquire(ctx context.Context, onPark func()) error { } } select { - case <-b.observer.Tick(): + case <-tick: case <-ctx.Done(): return ctx.Err() case <-timer.C: diff --git a/internal/tracing/middleware.go b/internal/tracing/middleware.go index af7c4a7..b96494a 100644 --- a/internal/tracing/middleware.go +++ b/internal/tracing/middleware.go @@ -36,6 +36,21 @@ func (rw *responseWriter) Write(b []byte) (int, error) { return n, err } +// Flush forwards to the wrapped writer so that streaming handlers can still +// detect http.Flusher through this wrapper. Without it a bare +// `w.(http.Flusher)` assertion fails for every request that passes through +// this middleware, which silently disables SSE flushing and keep-alives. +func (rw *responseWriter) Flush() { + if f, ok := rw.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +// Unwrap exposes the wrapped writer to http.ResponseController. +func (rw *responseWriter) Unwrap() http.ResponseWriter { + return rw.ResponseWriter +} + // Middleware returns an HTTP middleware that adds tracing to requests. func Middleware(next http.Handler) http.Handler { tracer := otel.Tracer(tracerName) diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 6a7f02d..1709c65 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -70,6 +70,12 @@ spec: value: "info" - name: ADMISSION_MAX_PARK_SECONDS value: "120" + # Must track the engine's --max-num-seqs (homelab: reasoning.yaml). + # The gauge this compares against is engine-wide, so this number is a + # ceiling on concurrent requests across all gateway replicas, not a + # per-pod slot count. Drift here is silent; startup logs the value. + - name: ADMISSION_MAX_INFLIGHT + value: "4" # Only backends that publish the vLLM scheduler gauges may be observed. # `reasoning` is the only vLLM engine; ornith/qwen-cpu are Ollama and # embeddings/rerank are TEI, so observing them would hold