fix(proxy): address review findings on admission control (#64)
CI / CI (pull_request) Successful in 33m12s
CI / CI (pull_request) Successful in 33m12s
- tracing: responseWriter now forwards Flush and exposes Unwrap. Without it w.(http.Flusher) failed for every request passing the tracing middleware, so the parked-stream keep-alive never started: a parked streaming request wrote nothing for up to ADMISSION_MAX_PARK_SECONDS, which is exactly the client timeout the keep-alive was added to prevent. - proxy: once the SSE status line is spent, ModifyResponse frames an upstream response >= 400 as an event: error instead of letting a raw error body land inside the stream, where a streaming client reads it as an empty completion and reports success. net/http keeps the first status, so the upstream status cannot be forwarded after the park has written one. - engine_state: new Snapshot() returns the state and its tick channel under one lock. Get and Tick taken separately can miss the poll that lands between the two calls, parking a request for an extra interval. - engine_state: observer_degraded now tracks current parsability rather than being raised once on the first poll and never cleared, which pinned it at 1 after a single truncated response and never reported drift that appeared later. markDegraded only touches the exporter when the value changes. - deployment: ADMISSION_MAX_INFLIGHT is set explicitly to 4, documenting the coupling to the engine's --max-num-seqs, instead of silently defaulting to VLLM_MAX_NUM_SEQS.
This commit is contained in:
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user