Merge branch 'main' into 2169-registry-capacity-counters

Author: LukeAVanDrie

AGENTS.md

@@ -41,10 +41,26 @@ llm-d Router. Go service that routes inference requests to model-serving pods vi
 
 - Standard Go. `make format` and `make lint` are authoritative.
 - Comments are terse and only present when the WHY is non-obvious. Never paraphrase the code.
-- Docs and comments describe the current state on its own terms. No "previously", "now", "recently", "renamed from", "added to fix", or other temporal or conversational framing. A reader with no context for the change must still understand the text.
+- Docs and comments describe the current state on its own terms. No "previously", "now", "recently", "renamed from", "added to fix", "this PR", "see above", or other temporal, deictic, or conversational framing. A reader with no context for the change must still understand the text.
 - State each fact once, in its canonical location. Do not duplicate across struct docs, prose, tables, inline comments, and examples.
 - Do not use Unicode symbols or special characters in general, unless explicitly requested.
 
+### Constructions to delete
+
+Model-drafted prose drifts toward compressed, rhetorical phrasing. Before shipping any prose (comments, docs, commit bodies, PR descriptions), decompress it: rewrite each such sentence as a plain statement of the fact it carries, judged against what the change is actually for. The test: if a sentence sounds quotable, delete it.
+
+| Tic | Example | Fix |
+|---|---|---|
+| X, not Y | "the queue is not a buffer, it is a fairness mechanism" | say what it is, once |
+| Coined aphorism | "a shed request is not a failure, it is the contract" | delete the sentence |
+| Closing zinger | a paragraph that ends on a beat instead of a fact | end on information |
+| Triads for rhythm | "no locks to take, no channels to drain, no state to sync" | one clause |
+| Portent counters | "three things follow", "two points are worth noting" | just say them |
+| Filler intensifiers | genuinely, precisely, critically, crucially, "worth noting" | cut |
+| Editorial tails | "..., which is exactly what we want" | cut, or a new sentence |
+| Grandeur adjectives | comprehensive, robust, seamless, production-ready | name the verified behavior |
+| Dash pileup | more than one dash pair per paragraph | periods |
+
 ### Logging
 
 The codebase uses `go-logr` via controller-runtime. Verbosity constants are defined in `pkg/common/observability/logging` (`DEFAULT=2`, `VERBOSE=3`, `DEBUG=4`, `TRACE=5`).

RELEASE-NOTES.md

@@ -108,6 +108,7 @@ import (
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/anthropic"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/openai"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/passthrough"
+	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/sglanghttp"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/vertexai"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/vllmgrpc"
 	"github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requesthandling/parsers/vllmhttp"
@@ -708,6 +709,7 @@ func (r *Runner) registerInTreePlugins() {
 	fwkplugin.Register(passthrough.PassthroughParserType, fwkplugin.StabilityBeta, passthrough.PassthroughParserPluginFactory)
 	fwkplugin.Register(anthropic.AnthropicParserType, fwkplugin.StabilityBeta, anthropic.AnthropicParserPluginFactory)
 	fwkplugin.Register(vllmhttp.VllmHTTPParserType, fwkplugin.StabilityBeta, vllmhttp.VllmHTTPParserPluginFactory)
+	fwkplugin.Register(sglanghttp.SGLangHTTPParserType, fwkplugin.StabilityBeta, sglanghttp.SGLangHTTPParserPluginFactory)
 	fwkplugin.Register(vertexai.VertexAIParserType, fwkplugin.StabilityBeta, vertexai.VertexAIParserPluginFactory)
 
 	// register saturation detector plugins

cmd/epp/runner/runner.go

@@ -279,6 +279,55 @@ Exposed when the `flowControl` feature gate is enabled.
 *   **Usage:** A nonzero value during a dispatch stall indicates a model-server metrics collection
     problem (scrape path, port, TLS, auth) rather than genuine overload.
 
+#### `flow_control_capacity_utilization_requests`
+
+*   **Type:** Gauge
+*   **Labels:** `priority`, `inference_pool`
+*   **Description:** Fraction of a priority band's **effective** request-count capacity currently
+    occupied (0.0-1.0), aggregated over every flow in the band. This is not a per-flow-queue metric:
+    `priority` identifies the band. A band that does not configure `maxRequests` falls back to a
+    default denominator, so every configured band reports a series. The all-bands rollup is a
+    separate metric, `flow_control_global_capacity_utilization_requests`.
+*   **Usage:** Lets operators alert on "the band is at N% of its request limit" without joining
+    configured `maxRequests` values into the query. Sustained values near 1.0 precede
+    `flow_control_requests_total{outcome="RejectedCapacity"}` rising. Note the denominator is always
+    the band's own capacity: when a global cap sits below the sum of the band caps, admission is
+    bounded by the global cap first, so the per-band ratio understates real pressure — read it
+    alongside `flow_control_global_capacity_utilization_requests`. (Making a band's denominator
+    `min(band, global)` would make one band's ratio depend on other bands' configuration, which is
+    worse.)
+
+#### `flow_control_capacity_utilization_bytes`
+
+*   **Type:** Gauge
+*   **Labels:** `priority`, `inference_pool`
+*   **Description:** Byte-size counterpart of `flow_control_capacity_utilization_requests`: the
+    fraction of a priority band's effective byte-size capacity currently occupied (0.0-1.0),
+    aggregated over every flow in the band, with the same default-denominator fallback.
+*   **Usage:** Memory-pressure equivalent of the request-count ratio; a band can hit its `maxBytes`
+    ceiling long before its `maxRequests` one when payloads are large. The same global-cap caveat
+    applies — compare against `flow_control_global_capacity_utilization_bytes`.
+
+#### `flow_control_global_capacity_utilization_requests`
+
+*   **Type:** Gauge
+*   **Labels:** `inference_pool`
+*   **Description:** Fraction of the global request-count capacity currently occupied across all
+    priority bands (0.0-1.0). Global capacity is optional and unset by default, so this series is
+    emitted only when it is configured — absent, not 0.
+*   **Usage:** The rollup companion to the per-band ratio. It lives in its own metric family so that
+    aggregations over the per-band family (`sum`, `max`, `topk`) do not double count or rank against
+    the rollup.
+
+#### `flow_control_global_capacity_utilization_bytes`
+
+*   **Type:** Gauge
+*   **Labels:** `inference_pool`
+*   **Description:** Byte-size counterpart of `flow_control_global_capacity_utilization_requests`,
+    with the same optional-and-omitted-when-unset behaviour.
+*   **Usage:** Shows whether the global byte ceiling, rather than any single band, is what is
+    bounding admission.
+
 #### `flow_control_requests_total`
 
 *   **Type:** Counter

docs/metrics.md

@@ -78,6 +78,37 @@ func TestBuildChunkedBodyResponses(t *testing.T) {
 	}
 }
 
+// TestBuildChunkedBodyResponses_PreservesBody confirms that the chunks reassemble into
+// exactly the input bytes whether or not the caller mutated the body upstream: chunking
+// itself is agnostic to mutation, so skipping a rewrite must not change what reaches Envoy.
+func TestBuildChunkedBodyResponses_PreservesBody(t *testing.T) {
+	tests := []struct {
+		name string
+		body []byte
+	}{
+		{
+			name: "unmutated body",
+			body: []byte(`{"id":"cmpl-123","model":"vllm-backend-01","choices":[]}`),
+		},
+		{
+			name: "mutated body",
+			body: []byte(`{"id":"cmpl-123","model":"gpt-4-proxy","choices":[]}`),
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			responses := BuildChunkedBodyResponses(test.body, true)
+			var got []byte
+			for _, response := range responses {
+				got = append(got, response.BodyMutation.GetStreamedResponse().GetBody()...)
+			}
+			if string(got) != string(test.body) {
+				t.Fatalf("reassembled chunks = %q, want %q", got, test.body)
+			}
+		})
+	}
+}
+
 func generateBytes(count int) []byte {
 	arr := make([]byte, count)
 	_, _ = rand.Read(arr)

pkg/common/envoy/chunking_test.go

@@ -391,6 +391,39 @@ func (p *Processor) hasCapacity(priority int, itemByteSize uint64) (bool, contra
 	return true, snapshot, nil
 }
 
+// recordCapacityUtilization emits occupancy/effective-capacity ratio gauges per priority band (aggregated over every
+// flow in the band, never per flow queue), plus the all-bands rollup in its own metric family when a global capacity
+// is configured. It reads a single Stats() snapshot; the data source is expected to move with the engine-merge
+// refactor, but the metric contract (names, labels, semantics) stays stable (#2102).
+//
+// Band capacities always resolve to a value (applyDefaults supplies a fallback), so every configured band reports.
+// Global capacity is optional, so its series is omitted when unset rather than reported as a misleading 0.
+func (p *Processor) recordCapacityUtilization() {
+	stats := p.registry.Stats()
+
+	for priority, band := range stats.PerPriorityBandStats {
+		priorityStr := strconv.Itoa(priority)
+		if band.CapacityRequests > 0 {
+			metrics.RecordFlowControlCapacityUtilizationRequests(priorityStr, p.poolName,
+				float64(band.Len)/float64(band.CapacityRequests))
+		}
+		if band.CapacityBytes > 0 {
+			metrics.RecordFlowControlCapacityUtilizationBytes(priorityStr, p.poolName,
+				float64(band.ByteSize)/float64(band.CapacityBytes))
+		}
+	}
+
+	// All-bands rollup, only when a global capacity is configured.
+	if stats.TotalCapacityRequests > 0 {
+		metrics.RecordFlowControlGlobalCapacityUtilizationRequests(p.poolName,
+			float64(stats.TotalLen)/float64(stats.TotalCapacityRequests))
+	}
+	if stats.TotalCapacityBytes > 0 {
+		metrics.RecordFlowControlGlobalCapacityUtilizationBytes(p.poolName,
+			float64(stats.TotalByteSize)/float64(stats.TotalCapacityBytes))
+	}
+}
+
 // dispatchCycle attempts to dispatch a single item by iterating through priority bands from highest to lowest.
 // It applies the configured policies for each band to select an item and then attempts to dispatch it.
 // It returns true if an item was successfully dispatched, and false otherwise.
@@ -418,6 +451,9 @@ func (p *Processor) dispatchCycle(ctx context.Context) bool {
 	// Record pool saturation metric
 	metrics.RecordFlowControlPoolSaturation(p.poolName, saturation)
 
+	// Record capacity utilization ratios (the demand-side twin of saturation) from the same periodic sample.
+	p.recordCapacityUtilization()
+
 	priorities := p.registry.AllOrderedPriorityLevels()
 	ceilings := p.ceilingsBuffer(len(priorities))
 	p.usageLimitPolicy.ComputeLimit(ctx, saturation, priorities, ceilings)

pkg/epp/flowcontrol/controller/internal/processor.go

@@ -19,6 +19,7 @@ package flowcontrol_test
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -1178,9 +1179,21 @@ func queueSizeGaugeSum(t *testing.T, fairnessID string) float64 {
 // deterministic.
 func TestFlowControlMetricsEmitted(t *testing.T) {
 	eppmetrics.Register()
+	// The capacity gauges are keyed only by (priority, inference_pool), which every harness in this
+	// package shares, so the assertions below need a clean slate.
+	eppmetrics.Reset()
 
 	detector := newBlockedDetector()
-	h := newHarness(t, harnessOpts{detector: detector})
+	// bandMaxRequests bounds the band and maxRequests/maxBytes bound the registry as a whole, so both
+	// the per-band ratios and both all-bands rollups are computable (occupancy/effective capacity).
+	// Band capacity always resolves to a value; the global ones are optional, which the omission test
+	// below covers.
+	h := newHarness(t, harnessOpts{
+		detector:        detector,
+		bandMaxRequests: 10,
+		maxRequests:     4,
+		maxBytes:        10_000,
+	})
 
 	key := flowcontrol.FlowKey{ID: "metrics-flow", Priority: 0}
 
@@ -1203,6 +1216,26 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	require.Greater(t, queueSizeGaugeSum(t, key.ID), 0.0,
 		"queue_size should be > 0 while a request is actively queued")
 
+	// Both dimensions are bounded (bandMaxRequests=10, maxRequests=4 above), so with exactly one
+	// request queued the ratios are occupancy/effective capacity: 1/10 for the band and 1/4 for the
+	// all-bands rollup. Unlike queue_size, these gauges are refreshed by the dispatch cycle rather
+	// than synchronously on enqueue, so they trail admission by up to one tick.
+	priorityStr := strconv.Itoa(key.Priority)
+	require.Eventually(t, func() bool {
+		return capacityUtilizationGauge(t, capacityUtilizationRequestsFamily, priorityStr) > 0
+	}, time.Second, time.Millisecond,
+		"band utilization ratio should be > 0 while a request is actively queued")
+	require.InDelta(t, 0.1, capacityUtilizationGauge(t, capacityUtilizationRequestsFamily, priorityStr), 1e-9,
+		"band ratio should equal occupancy/capacity (1 queued / bandMaxRequests=10)")
+	require.InDelta(t, 0.25, globalCapacityUtilizationGauge(t, globalCapacityUtilizationRequestsFamily), 1e-9,
+		"rollup ratio should equal occupancy/global capacity (1 queued / maxRequests=4)")
+
+	// The bytes dimension is driven by the same snapshot, so it must also be reporting a live ratio.
+	require.Greater(t, capacityUtilizationGauge(t, capacityUtilizationBytesFamily, priorityStr), 0.0,
+		"band byte-size utilization should be > 0 while a request is actively queued")
+	require.Greater(t, globalCapacityUtilizationGauge(t, globalCapacityUtilizationBytesFamily), 0.0,
+		"rollup byte-size utilization should be > 0 while a request is actively queued")
+
 	// Unblock the detector so the request finalizes deterministically via dispatch.
 	detector.Unblock(1)
 
@@ -1218,6 +1251,102 @@ func TestFlowControlMetricsEmitted(t *testing.T) {
 	// gauge is already back at 0 -- no polling needed.
 	require.Zero(t, queueSizeGaugeSum(t, key.ID),
 		"queue_size should return to 0 after the request finalizes")
+
+	// The utilization gauges are dispatch-cycle driven, so they trail the drain by up to one tick.
+	require.Eventually(t, func() bool {
+		return capacityUtilizationGauge(t, capacityUtilizationRequestsFamily, priorityStr) == 0 &&
+			globalCapacityUtilizationGauge(t, globalCapacityUtilizationRequestsFamily) == 0
+	}, time.Second, time.Millisecond,
+		"band and rollup utilization ratios should return to 0 after the queue drains")
+}
+
+// TestFlowControlCapacityUtilizationOmitsUnsetGlobal verifies that the all-bands rollup is absent
+// rather than reported as 0 when no global capacity is configured. Downstream alerts may key off
+// series absence, so a refactor that starts emitting 0 here would silently change their meaning.
+func TestFlowControlCapacityUtilizationOmitsUnsetGlobal(t *testing.T) {
+	eppmetrics.Register()
+	eppmetrics.Reset()
+
+	detector := newBlockedDetector()
+	// bandMaxRequests only: the band reports as usual, the global capacity stays unset.
+	h := newHarness(t, harnessOpts{detector: detector, bandMaxRequests: 10})
+
+	key := flowcontrol.FlowKey{ID: "metrics-flow-no-global", Priority: 0}
+	priorityStr := strconv.Itoa(key.Priority)
+
+	results := make(chan dispatchResult, 1)
+	go func() {
+		reqCtx, reqCancel := context.WithTimeout(h.ctx, 5*time.Second)
+		defer reqCancel()
+		req := &testRequest{id: key.ID, key: key, byteSize: 100, ttl: 5 * time.Second}
+		outcome, err := h.fc.EnqueueAndWait(reqCtx, req)
+		results <- dispatchResult{id: key.ID, outcome: outcome, err: err}
+	}()
+
+	// Wait until the band series exists, which means a dispatch cycle has published a sample.
+	require.Eventually(t, func() bool {
+		return capacityUtilizationGauge(t, capacityUtilizationRequestsFamily, priorityStr) > 0
+	}, time.Second, time.Millisecond, "band utilization should be reported for a bounded band")
+
+	require.Equal(t, -1.0, globalCapacityUtilizationGauge(t, globalCapacityUtilizationRequestsFamily),
+		"no global request capacity is configured, so the rollup series must be absent, not 0")
+	require.Equal(t, -1.0, globalCapacityUtilizationGauge(t, globalCapacityUtilizationBytesFamily),
+		"no global byte capacity is configured, so the rollup series must be absent, not 0")
+
+	detector.Unblock(1)
+	select {
+	case r := <-results:
+		require.NoError(t, r.err)
+		require.Equal(t, fcTypes.QueueOutcomeDispatched, r.outcome)
+	case <-time.After(5 * time.Second):
+		t.Fatal("request did not dispatch after detector was unblocked")
+	}
+}
+
+// Metric family names asserted by the capacity utilization tests.
+const (
+	capacityUtilizationRequestsFamily       = "llm_d_epp_flow_control_capacity_utilization_requests"
+	capacityUtilizationBytesFamily          = "llm_d_epp_flow_control_capacity_utilization_bytes"
+	globalCapacityUtilizationRequestsFamily = "llm_d_epp_flow_control_global_capacity_utilization_requests"
+	globalCapacityUtilizationBytesFamily    = "llm_d_epp_flow_control_global_capacity_utilization_bytes"
+)
+
+// capacityUtilizationGauge returns the per-band capacity utilization ratio recorded in the given
+// metric family for the given priority band, or -1 if no series exists for that band.
+func capacityUtilizationGauge(t *testing.T, family, priority string) float64 {
+	t.Helper()
+	families, err := ctrlmetrics.Registry.Gather()
+	require.NoError(t, err)
+	for _, f := range families {
+		if f.GetName() != family {
+			continue
+		}
+		for _, m := range f.GetMetric() {
+			for _, lp := range m.GetLabel() {
+				if lp.GetName() == "priority" && lp.GetValue() == priority {
+					return m.GetGauge().GetValue()
+				}
+			}
+		}
+	}
+	return -1
+}
+
+// globalCapacityUtilizationGauge returns the all-bands capacity utilization ratio recorded in the
+// given metric family, or -1 if the series is absent (no global capacity configured).
+func globalCapacityUtilizationGauge(t *testing.T, family string) float64 {
+	t.Helper()
+	families, err := ctrlmetrics.Registry.Gather()
+	require.NoError(t, err)
+	for _, f := range families {
+		if f.GetName() != family {
+			continue
+		}
+		for _, m := range f.GetMetric() {
+			return m.GetGauge().GetValue()
+		}
+	}
+	return -1
 }
 
 // ============================================================================

pkg/epp/flowcontrol/integration_test.go

@@ -63,6 +63,12 @@ type Response struct {
 	ReqMetadata map[string]any
 	// Token usage counts parsed from the response body.
 	Usage requesthandling.Usage
+	// StreamedEvents is the running count of stream data events observed so far. It is the only
+	// length signal this record carries for a truncated stream, since a stream that never
+	// completes carries no usage block. Consumers must not treat zero as evidence that nothing
+	// was generated; requesthandling.ParsedResponse documents which parsers count and how the
+	// count deviates from the token count.
+	StreamedEvents int
 	// TerminationCause labels how the stream ended.
 	TerminationCause TerminationCause
 	// DynamicMetadata is a map of metadata that can be passed to the Envoy. It is populated into the dynamic

pkg/epp/framework/interface/requestcontrol/types.go

@@ -83,4 +83,12 @@ type ParseResult struct {
 type ParsedResponse struct {
 	// Usage is only populate when the raw response has usage.
 	Usage *Usage
+	// StreamedEvents is how many stream data events this parse observed, zero for a non-streamed
+	// response or a parser that does not count. For servers that stream one content delta per
+	// token the count approximates the tokens in the parsed chunk, but protocol events (role
+	// deltas, finish reasons, usage-only events, Responses API lifecycle events) inflate it, a
+	// server that batches tokens into one event deflates it, and the scan recognizes only the
+	// "data: " framing this parser already assumes (no-space prefixes, multi-line data fields,
+	// and keep-alive data lines skew it further).
+	StreamedEvents int
 }