diff --git a/docs/metrics.md b/docs/metrics.md index fa84aafadf..f3091a5a69 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -63,6 +63,13 @@ Names below omit the subsystem prefix. Unless a section states otherwise, the pr and the release stage is ALPHA. Request and latency metrics share the label set `{model_name, target_model_name, fairness_id, priority}`. +Client-derived label values are cardinality-bounded: `model_name` and `target_model_name` (from the +request body) share a cap of 1000 distinct values, and `fairness_id` (from the +`x-llm-d-inference-fairness-id` header) is capped at 1000 distinct values. Caps apply over the +lifetime of the process; once a cap is reached, new values are reported as `other`. Model names +configured through InferenceModelRewrite rules never fold to `other`. Flow control series for a +`fairness_id` are removed when its flow is garbage collected. + ### Request and latency | Name | Type | Notes | diff --git a/pkg/epp/flowcontrol/registry/registry.go b/pkg/epp/flowcontrol/registry/registry.go index 2e85999df4..b0399c9617 100644 --- a/pkg/epp/flowcontrol/registry/registry.go +++ b/pkg/epp/flowcontrol/registry/registry.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "slices" + "strconv" "sync" "sync/atomic" "time" @@ -31,6 +32,7 @@ import ( "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/framework/plugins/queue" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" + "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) // propagateStatsDeltaFunc defines the callback function used to propagate statistics changes (deltas) up the hierarchy @@ -472,6 +474,14 @@ func (fr *FlowRegistry) gcFlows() { } fr.cleanupFlowResources(keysToClean) + + // Prune the flows' metric series. Fairness IDs come from client input, so without pruning the + // per-flow metric vectors grow monotonically with every fairness ID ever observed. Done after + // cleanupFlowResources and outside fr.mu: DeletePartialMatch scans whole metric vectors, which + // must not run under the registry write lock. + for _, key := range keysToClean { + metrics.DeleteFlowControlFlowSeries(key.ID, strconv.Itoa(key.Priority)) + } } } diff --git a/pkg/epp/flowcontrol/registry/registry_test.go b/pkg/epp/flowcontrol/registry/registry_test.go index 0afd0a4ebe..8ea93db694 100644 --- a/pkg/epp/flowcontrol/registry/registry_test.go +++ b/pkg/epp/flowcontrol/registry/registry_test.go @@ -19,6 +19,7 @@ package registry import ( "context" "fmt" + "strconv" "sync" "sync/atomic" "testing" @@ -28,10 +29,12 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" testclock "k8s.io/utils/clock/testing" + crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics" "github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol" "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol/mocks" + eppmetrics "github.com/llm-d/llm-d-router/pkg/epp/metrics" ) // --- Test Harness --- @@ -1349,3 +1352,45 @@ func TestFlowRegistry_FlowErrorScoping(t *testing.T) { assert.Equal(t, int32(concurrency), errorCount.Load(), "All requests should fail flow provisioning") assert.Equal(t, int32(0), successCount.Load(), "No request should succeed if flow provisioning failed") } + +// countSeriesWithFairnessID gathers the global metrics registry and counts series carrying the +// given fairness_id label value, across all metric families. +func countSeriesWithFairnessID(t *testing.T, fairnessID string) int { + t.Helper() + families, err := crmetrics.Registry.Gather() + require.NoError(t, err, "gathering the metrics registry must succeed") + n := 0 + for _, mf := range families { + for _, m := range mf.GetMetric() { + for _, lp := range m.GetLabel() { + if lp.GetName() == "fairness_id" && lp.GetValue() == fairnessID { + n++ + } + } + } + } + return n +} + +// Metric series are labeled by the flow's client-derived fairness ID, so they must not outlive the +// flow: gcFlows prunes them via metrics.DeleteFlowControlFlowSeries once the flow is collected. +func TestFlowRegistry_GarbageCollection_PrunesMetricSeries(t *testing.T) { + // Not parallel: reads the process-global metrics registry. The unique fairness ID keeps the + // assertions isolated from series recorded by other tests. + eppmetrics.Register() + h := newRegistryTestHarness(t, harnessOptions{manualGC: true}) + const flowID = "gc-metric-prune-flow" + key := flowcontrol.FlowKey{ID: flowID, Priority: highPriority} + + h.openConnectionOnFlow(key) + eppmetrics.RecordFlowControlRequestEnqueueDuration( + flowID, strconv.Itoa(highPriority), "Dispatched", time.Millisecond) + require.Positive(t, countSeriesWithFairnessID(t, flowID), "Setup: series must exist before GC") + + h.fakeClock.Step(h.config.FlowGCTimeout + time.Second) + h.fr.ExecuteGCCycle() + + h.assertFlowDoesNotExist(key, "Setup: idle flow must have been collected") + assert.Zero(t, countSeriesWithFairnessID(t, flowID), + "GC must prune every metric series labeled with the collected flow's fairness ID") +} diff --git a/pkg/epp/metrics/cardinality.go b/pkg/epp/metrics/cardinality.go index 857244463a..3aa3b4ccb2 100644 --- a/pkg/epp/metrics/cardinality.go +++ b/pkg/epp/metrics/cardinality.go @@ -97,6 +97,20 @@ func (b *boundedLabel) pin(v string) { var modelLabelLimiter = newBoundedLabel(maxModelLabelValues) +// Fairness IDs are populated from a client request header (or an agent-identity attribute), so +// like model names their cardinality is not operator-bounded. They label per-request and +// flow-control metrics; without a cap, every distinct fairness ID ever observed permanently +// grows the time series set. maxFairnessLabelValues bounds the distinct fairness_id label +// values; values beyond the cap collapse to overflowValue. +const maxFairnessLabelValues = 1000 + +var fairnessLabelLimiter = newBoundedLabel(maxFairnessLabelValues) + +// boundFairnessID caps the request-derived fairness_id label. +func boundFairnessID(fairnessID string) string { + return fairnessLabelLimiter.bound(fairnessID) +} + // PreAdmitModelLabels pins the given model names so they always emit their real // label value on model-labeled metrics, regardless of how many unconfigured // names have been admitted. The datastore calls this when InferenceModelRewrite diff --git a/pkg/epp/metrics/cardinality_test.go b/pkg/epp/metrics/cardinality_test.go index e49f7eb000..d67f26a640 100644 --- a/pkg/epp/metrics/cardinality_test.go +++ b/pkg/epp/metrics/cardinality_test.go @@ -19,6 +19,7 @@ package metrics import ( "fmt" "testing" + "time" promtestutil "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" @@ -98,3 +99,139 @@ func TestRecordRequestCounterBoundsModelCardinality(t *testing.T) { require.LessOrEqualf(t, count, testCap+1, "model_name cardinality must stay bounded by the cap, got %d series", count) } + +// The fairness_id label is populated from a client request header, so the package-level limiter +// must collapse an unbounded flood of distinct IDs into the overflow bucket instead of minting a +// series per ID. +func TestFairnessLabelFloodCollapsesToOverflow(t *testing.T) { + const testCap = 5 + old := fairnessLabelLimiter + fairnessLabelLimiter = newBoundedLabel(testCap) + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + t.Cleanup(func() { + fairnessLabelLimiter = old + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + }) + + for i := 0; i < 1000; i++ { + RecordFlowControlRequestEnqueueDuration(fmt.Sprintf("tenant-%d", i), "0", "Dispatched", time.Millisecond) + } + + // testCap admitted IDs + 1 overflow series, per family. + require.Equal(t, testCap+1, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration), + "1000 distinct fairness IDs must collapse to cap+overflow series, not one series each") + require.Equal(t, testCap+1, promtestutil.CollectAndCount(llmdFlowControlRequestEnqueueDuration), + "the llm_d_epp family must be bounded identically") +} + +// DeleteFlowControlFlowSeries backs the flow registry's GC hook: once a flow is collected, its +// series must not linger for the lifetime of the process. +func TestDeleteFlowControlFlowSeries(t *testing.T) { + old := fairnessLabelLimiter + fairnessLabelLimiter = newBoundedLabel(10) + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + t.Cleanup(func() { + fairnessLabelLimiter = old + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + }) + + RecordFlowControlRequestEnqueueDuration("tenant-a", "0", "Dispatched", time.Millisecond) + RecordFlowControlRequestEnqueueDuration("tenant-a", "0", "Rejected", time.Millisecond) + RecordFlowControlRequestEnqueueDuration("tenant-b", "0", "Dispatched", time.Millisecond) + require.Equal(t, 3, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration), + "setup: expected one series per (fairness_id, outcome) pair") + + DeleteFlowControlFlowSeries("tenant-a", "0") + + require.Equal(t, 1, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration), + "all of tenant-a's series (every outcome) must be pruned; tenant-b's must survive") + require.Equal(t, 1, promtestutil.CollectAndCount(llmdFlowControlRequestEnqueueDuration), + "the llm_d_epp family must be pruned identically") +} + +// The bound is applied mechanically in every record function that takes a fairness ID; this guards +// the request-metric family (distinct label ordering from the flow control family) against the +// pattern regressing there. +func TestFairnessLabelBoundOnRequestMetrics(t *testing.T) { + const testCap = 3 + oldFairness := fairnessLabelLimiter + fairnessLabelLimiter = newBoundedLabel(testCap) + oldModels := modelLabelLimiter + modelLabelLimiter = newBoundedLabel(10) + requestCounter.Reset() + llmdRequestCounter.Reset() + t.Cleanup(func() { + fairnessLabelLimiter = oldFairness + modelLabelLimiter = oldModels + requestCounter.Reset() + llmdRequestCounter.Reset() + }) + + for i := 0; i < 100; i++ { + RecordRequestCounter("model-a", "model-a", fmt.Sprintf("tenant-%d", i), 0) + } + + require.Equal(t, testCap+1, promtestutil.CollectAndCount(llmdRequestCounter), + "100 distinct fairness IDs must collapse to cap+overflow series on the request family") + require.Equal(t, 1, promtestutil.CollectAndCount(requestCounter), + "the deprecated family has no fairness_id label and must stay a single series") +} + +// RecordFlowControlRequestQueueDuration takes request-body model names; they must flow through the +// model limiter like every sibling record function. +func TestQueueDurationBoundsModelLabels(t *testing.T) { + const testCap = 3 + oldModels := modelLabelLimiter + modelLabelLimiter = newBoundedLabel(testCap) + oldFairness := fairnessLabelLimiter + fairnessLabelLimiter = newBoundedLabel(10) + flowControlRequestQueueDuration.Reset() + llmdFlowControlRequestQueueDuration.Reset() + t.Cleanup(func() { + modelLabelLimiter = oldModels + fairnessLabelLimiter = oldFairness + flowControlRequestQueueDuration.Reset() + llmdFlowControlRequestQueueDuration.Reset() + }) + + for i := 0; i < 100; i++ { + m := fmt.Sprintf("model-%d", i) + RecordFlowControlRequestQueueDuration("tenant", "0", "Dispatched", "pool", m, m, time.Millisecond) + } + + require.Equal(t, testCap+1, promtestutil.CollectAndCount(flowControlRequestQueueDuration), + "100 distinct model names must collapse to cap+overflow series, not one series each") + require.Equal(t, testCap+1, promtestutil.CollectAndCount(llmdFlowControlRequestQueueDuration), + "the llm_d_epp family must be bounded identically") +} + +// A client can choose the overflow value itself as its fairness ID; GC of that flow must not +// delete the shared overflow series that aggregates every capped-out tenant. +func TestDeleteFlowControlFlowSeriesPreservesOverflowSeries(t *testing.T) { + old := fairnessLabelLimiter + fairnessLabelLimiter = newBoundedLabel(1) + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + t.Cleanup(func() { + fairnessLabelLimiter = old + flowControlRequestEnqueueDuration.Reset() + llmdFlowControlRequestEnqueueDuration.Reset() + }) + + // tenant-a fills the single cap slot; tenant-b folds to the overflow series. + RecordFlowControlRequestEnqueueDuration("tenant-a", "0", "Dispatched", time.Millisecond) + RecordFlowControlRequestEnqueueDuration("tenant-b", "0", "Dispatched", time.Millisecond) + require.Equal(t, 2, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration), + "setup: expected the admitted series plus the overflow series") + + DeleteFlowControlFlowSeries(overflowValue, "0") + + require.Equal(t, 2, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration), + "deleting the overflow value must be a no-op; the shared overflow series must survive") + require.Equal(t, 2, promtestutil.CollectAndCount(llmdFlowControlRequestEnqueueDuration), + "the llm_d_epp family must be preserved identically") +} diff --git a/pkg/epp/metrics/metrics.go b/pkg/epp/metrics/metrics.go index 044d557b88..8011acd5ff 100644 --- a/pkg/epp/metrics/metrics.go +++ b/pkg/epp/metrics/metrics.go @@ -554,6 +554,7 @@ func Reset() { // RecordRequestCounter records the number of requests. func RecordRequestCounter(modelName, targetModelName, fairnessID string, priority int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) prioStr := strconv.Itoa(priority) requestCounter.WithLabelValues(modelName, targetModelName, prioStr).Inc() llmdRequestCounter.WithLabelValues(modelName, targetModelName, fairnessID, prioStr).Inc() @@ -562,6 +563,7 @@ func RecordRequestCounter(modelName, targetModelName, fairnessID string, priorit // RecordRequestErrCounter records the number of error requests. func RecordRequestErrCounter(modelName, targetModelName, fairnessID, priority string, code string) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if code != "" { requestErrCounter.WithLabelValues(modelName, targetModelName, code).Inc() llmdRequestErrCounter.WithLabelValues(modelName, targetModelName, fairnessID, priority, code).Inc() @@ -571,6 +573,7 @@ func RecordRequestErrCounter(modelName, targetModelName, fairnessID, priority st // RecordRequestSizes records the request sizes. func RecordRequestSizes(modelName, targetModelName, fairnessID, priority string, reqSize int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) requestSizes.WithLabelValues(modelName, targetModelName).Observe(float64(reqSize)) llmdRequestSizes.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(reqSize)) } @@ -578,6 +581,7 @@ func RecordRequestSizes(modelName, targetModelName, fairnessID, priority string, // RecordRequestLatencies records duration of request. func RecordRequestLatencies(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, complete time.Time) bool { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if !complete.After(received) { log.FromContext(ctx).V(logutil.DEFAULT).Error(nil, "Request latency values are invalid", "modelName", modelName, "targetModelName", targetModelName, "completeTime", complete, "receivedTime", received) @@ -592,6 +596,7 @@ func RecordRequestLatencies(ctx context.Context, modelName, targetModelName, fai // RecordResponseSizes records the response sizes. func RecordResponseSizes(modelName, targetModelName, fairnessID, priority string, size int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) responseSizes.WithLabelValues(modelName, targetModelName).Observe(float64(size)) llmdResponseSizes.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size)) } @@ -599,6 +604,7 @@ func RecordResponseSizes(modelName, targetModelName, fairnessID, priority string // RecordInputTokens records input tokens count. func RecordInputTokens(modelName, targetModelName, fairnessID, priority string, size int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if size > 0 { inputTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size)) llmdInputTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size)) @@ -608,6 +614,7 @@ func RecordInputTokens(modelName, targetModelName, fairnessID, priority string, // RecordOutputTokens records output tokens count. func RecordOutputTokens(modelName, targetModelName, fairnessID, priority string, size int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if size > 0 { outputTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size)) llmdOutputTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size)) @@ -617,6 +624,7 @@ func RecordOutputTokens(modelName, targetModelName, fairnessID, priority string, // RecordPromptCachedTokens records prompt cached tokens count. func RecordPromptCachedTokens(modelName, targetModelName, fairnessID, priority string, size int) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) promptCachedTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size)) llmdPromptCachedTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size)) } @@ -624,6 +632,7 @@ func RecordPromptCachedTokens(modelName, targetModelName, fairnessID, priority s // RecordNormalizedTimePerOutputToken (NTPOT) records the normalized time per output token. func RecordNormalizedTimePerOutputToken(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, complete time.Time, outputTokenCount int) bool { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if outputTokenCount <= 0 { return false } @@ -645,6 +654,7 @@ func RecordNormalizedTimePerOutputToken(ctx context.Context, modelName, targetMo // RecordRequestTTFT records the time to first token. func RecordRequestTTFT(ctx context.Context, modelName, targetModelName, fairnessID, priority string, streaming bool, received time.Time, firstToken time.Time) bool { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if firstToken.IsZero() { return false } @@ -666,6 +676,7 @@ func RecordRequestTTFT(ctx context.Context, modelName, targetModelName, fairness // RecordRequestTPOT records the average time per output token. func RecordRequestTPOT(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, firstToken time.Time, complete time.Time, outputTokenCount int) bool { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if firstToken.IsZero() || outputTokenCount <= 1 { return false } @@ -686,6 +697,7 @@ func RecordRequestTPOT(ctx context.Context, modelName, targetModelName, fairness // RecordInterTokenLatency records the time between consecutive response body chunks for streaming requests. func RecordInterTokenLatency(ctx context.Context, modelName, targetModelName, fairnessID, priority string, itlSeconds float64) bool { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if itlSeconds < 0 { log.FromContext(ctx).Error(nil, "Inter-token latency value must be non-negative", "modelName", modelName, "targetModelName", targetModelName, "itlSeconds", itlSeconds) @@ -698,6 +710,7 @@ func RecordInterTokenLatency(ctx context.Context, modelName, targetModelName, fa // IncRunningRequests increases the current running requests. func IncRunningRequests(modelName, targetModelName, fairnessID, priority string) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if modelName != "" { runningRequests.WithLabelValues(modelName).Inc() llmdRunningRequests.WithLabelValues(modelName, targetModelName, fairnessID, priority).Inc() @@ -707,6 +720,7 @@ func IncRunningRequests(modelName, targetModelName, fairnessID, priority string) // DecRunningRequests decreases the current running requests. func DecRunningRequests(modelName, targetModelName, fairnessID, priority string) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) if modelName != "" { runningRequests.WithLabelValues(modelName).Dec() llmdRunningRequests.WithLabelValues(modelName, targetModelName, fairnessID, priority).Dec() @@ -802,6 +816,8 @@ func RecordFlowControlRequestQueueDuration( modelName, targetModelName string, duration time.Duration, ) { + fairnessID = boundFairnessID(fairnessID) + modelName, targetModelName = boundModels(modelName, targetModelName) flowControlRequestQueueDuration.WithLabelValues( fairnessID, priority, outcome, inferencePool, @@ -826,6 +842,7 @@ func RecordFlowControlRequestEnqueueDuration( fairnessID string, priority string, outcome string, duration time.Duration, ) { + fairnessID = boundFairnessID(fairnessID) flowControlRequestEnqueueDuration.WithLabelValues( fairnessID, priority, outcome, ).Observe(duration.Seconds()) @@ -838,6 +855,7 @@ func RecordFlowControlRequestEnqueueDuration( // IncFlowControlQueueSize increments the Flow Control queue size gauge. func IncFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc() llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc() } @@ -845,6 +863,7 @@ func IncFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, tar // DecFlowControlQueueSize decrements the Flow Control queue size gauge. func DecFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec() llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec() } @@ -852,6 +871,7 @@ func DecFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, tar // AddFlowControlQueueBytes increments the Flow Control queue bytes gauge. func AddFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes)) llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes)) } @@ -859,6 +879,7 @@ func AddFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, ta // SubFlowControlQueueBytes decrements the Flow Control queue bytes gauge. func SubFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) { modelName, targetModelName = boundModels(modelName, targetModelName) + fairnessID = boundFairnessID(fairnessID) flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes)) llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes)) } @@ -874,6 +895,32 @@ func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) { llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc() } +// DeleteFlowControlFlowSeries removes every flow-control series labeled with the given fairness ID +// and priority, across both the deprecated and the llm_d_epp metric families. The fairness ID is +// derived from client input, so its cardinality is unbounded; the flow registry calls this when it +// garbage-collects an idle flow so that the metric vectors track live flows instead of growing +// monotonically with every fairness ID ever observed. +// +// Pruning is not synchronized with recording: a request reviving the flow concurrently with GC can +// have a queue gauge increment deleted here while its paired decrement lands afterwards, leaving +// the queue size/bytes gauges negative until the flow's next collection deletes the series again. +func DeleteFlowControlFlowSeries(fairnessID, priority string) { + // The overflow value aggregates every capped-out fairness ID, so a flow whose client-chosen ID + // equals it must not delete the shared series. + if fairnessID == overflowValue { + return + } + labels := prometheus.Labels{"fairness_id": fairnessID, "priority": priority} + flowControlRequestQueueDuration.DeletePartialMatch(labels) + flowControlRequestEnqueueDuration.DeletePartialMatch(labels) + flowControlQueueSize.DeletePartialMatch(labels) + flowControlQueueBytes.DeletePartialMatch(labels) + llmdFlowControlRequestQueueDuration.DeletePartialMatch(labels) + llmdFlowControlRequestEnqueueDuration.DeletePartialMatch(labels) + llmdFlowControlQueueSize.DeletePartialMatch(labels) + llmdFlowControlQueueBytes.DeletePartialMatch(labels) +} + // RecordInferenceModelRewriteDecision records the routing decision for InferenceModelRewrite. // The rewrite name and target come from configuration; only the source model name is // request-derived and needs bounding (a generic rule matches arbitrary model names).