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