Merge branch 'main' into fix/fc-config-defaults

Author: LukeAVanDrie

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 |

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))
+		}
 	}
 }
 

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")
+}

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

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")
+}