diff --git a/metrics/metrics.go b/metrics/metrics.go index a86d1049..634e02eb 100644 --- a/metrics/metrics.go +++ b/metrics/metrics.go @@ -491,6 +491,26 @@ type Snapshot struct { // mode it correctly reads ~0 because that path does no pipeline work. Zeroing either // would hide a true number rather than protect anyone. ObserveLLMNotice string `json:"observe_llm_notice,omitempty"` + + // ObserveQueue is the off-path worker pool's counter tuple, filled by the host at + // serve time (the pool lives in `modes`, which sits above this package). All five + // counters are exposed rather than just `queued` because `dropped` is the one that + // changes a reader's conclusion: a drop is an observation silently given up, so a + // rising `dropped` means the potential_*/projected_* figures UNDERSTATE what + // compaction would have saved. Reporting the queue depth while hiding the drops is + // exactly the gap noted in headroom's dashboard. Omitted when no pool is running, so + // a sync-only deployment shows no phantom queue. + ObserveQueue *QueueStats `json:"observe_queue,omitempty"` +} + +// QueueStats mirrors modes.Stats. Declared here as a plain struct rather than importing +// modes because the dependency runs the other way (modes uses metrics, not vice versa). +type QueueStats struct { + Queued int64 `json:"queued"` + Pending int64 `json:"pending"` + Processed int64 `json:"processed"` + Dropped int64 `json:"dropped"` + Errors int64 `json:"errors"` } // SelectorMiss is one output shape that matched no filter, with how often it appeared. diff --git a/proxy/modes_test.go b/proxy/modes_test.go index dc720a66..c2d642a6 100644 --- a/proxy/modes_test.go +++ b/proxy/modes_test.go @@ -541,3 +541,37 @@ func settleGoroutines() { time.Sleep(5 * time.Millisecond) } } + +// The observe-mode docs tell operators to watch `dropped` and `errors`, and make a point +// of contrasting that against a dashboard which reports only queue depth. The pool tracked +// all five counters correctly and NOTHING SERVED THEM: metrics.Snapshot had no field and +// the /stats handler never called Stats(), so the documented counter was unreachable. This +// asserts the wiring, not the pool. +func TestObserveQueueCountersReachStats(t *testing.T) { + up, _ := captureUpstream(t) + h, agg := modeHandler(t, modePipeline, up.URL, components.ModeObserve) + srv := httptest.NewServer(h.Mux()) + defer srv.Close() + + for i := 0; i < 3; i++ { + post(t, srv, dupBody()) + } + awaitSnapshot(t, agg, func(s metrics.Snapshot) bool { return s.ObserveRequests > 0 }) + + var got metrics.Snapshot + res, err := http.Get(srv.URL + "/stats") + if err != nil { + t.Fatalf("GET /stats: %v", err) + } + defer res.Body.Close() + if err := json.NewDecoder(res.Body).Decode(&got); err != nil { + t.Fatalf("decode /stats: %v", err) + } + if got.ObserveQueue == nil { + t.Fatal("observe_queue absent from /stats: the pool's counters are still unreachable, " + + "so the documented `dropped` cannot be read by any consumer") + } + if got.ObserveQueue.Processed == 0 && got.ObserveQueue.Queued == 0 && got.ObserveQueue.Pending == 0 { + t.Fatalf("observe_queue served but empty after 3 observed requests: %+v", got.ObserveQueue) + } +} diff --git a/proxy/proxy.go b/proxy/proxy.go index 8c454d92..41ca54db 100644 --- a/proxy/proxy.go +++ b/proxy/proxy.go @@ -654,6 +654,17 @@ func (h *Handler) stats(w http.ResponseWriter, _ *http.Request) { snap.FrozenDropped, snap.FrozenRepaired = fl.FrozenLossStats() snap.FrozenFlips = snap.FrozenDropped - snap.FrozenRepaired } + // Off-path pool counters, same layering: the pool lives in `modes`, so `metrics` cannot + // read it and the host merges it here. Without this the observe-mode docs describe a + // `dropped` counter no consumer can reach — the pool tracked it correctly and nothing + // served it. + if h.pool != nil { + q := h.pool.Stats() + snap.ObserveQueue = &metrics.QueueStats{ + Queued: q.Queued, Pending: q.Pending, + Processed: q.Processed, Dropped: q.Dropped, Errors: q.Errors, + } + } json.NewEncoder(w).Encode(snap) }