Skip to content

Commit c404985

Browse files
LukeAVanDrieLukeAVanDrie
committed
fix(metrics): bound fairness_id label cardinality and prune GC'd flow series
The fairness_id label on request and flow-control metrics is populated from a client request header (or agent-identity attribute), so its cardinality is not operator-bounded. Prometheus vectors never evict label combinations: every distinct fairness ID ever observed permanently allocated histogram series across both the deprecated and llm_d_epp metric families, growing pod RSS and scrape payloads without bound. Under flow-ID churn this was ~60% of remaining allocations in the full-path flow-control benchmark. Three changes, following the existing model-label precedent: - Bound fairness_id through a boundedLabel limiter (cap 1000, overflow to "other") in every record function that takes it, exactly as model names are already bounded via boundModels. - Bound model labels in RecordFlowControlRequestQueueDuration, which was missing the boundModels call its siblings have. - Prune a flow's flow-control series (both families) when the registry garbage-collects the flow, so the vectors track live flows. Pruning runs after cleanupFlowResources and outside the registry lock; DeletePartialMatch scans whole vectors and must not run under it. FullPath benchmark (registry/flow churn, M4 Pro, -benchtime=3s): 118k -> 213k d/s, 14.5KB -> 4.0KB per op, 264 -> 94 allocs per op. Signed-off-by: Luke Van Drie <lukevandrie@google.com>
1 parent 04cc371 commit c404985

4 files changed

Lines changed: 116 additions & 0 deletions

File tree

‎pkg/epp/flowcontrol/registry/registry.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"context"
2121
"fmt"
2222
"slices"
23+
"strconv"
2324
"sync"
2425
"sync/atomic"
2526
"time"
@@ -31,6 +32,7 @@ import (
3132
"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/contracts"
3233
"github.com/llm-d/llm-d-router/pkg/epp/flowcontrol/framework/plugins/queue"
3334
"github.com/llm-d/llm-d-router/pkg/epp/framework/interface/flowcontrol"
35+
"github.com/llm-d/llm-d-router/pkg/epp/metrics"
3436
)
3537

3638
// propagateStatsDeltaFunc defines the callback function used to propagate statistics changes (deltas) up the hierarchy
@@ -472,6 +474,14 @@ func (fr *FlowRegistry) gcFlows() {
472474
}
473475

474476
fr.cleanupFlowResources(keysToClean)
477+
478+
// Prune the flows' metric series. Fairness IDs come from client input, so without pruning the
479+
// per-flow metric vectors grow monotonically with every fairness ID ever observed. Done after
480+
// cleanupFlowResources and outside fr.mu: DeletePartialMatch scans whole metric vectors, which
481+
// must not run under the registry write lock.
482+
for _, key := range keysToClean {
483+
metrics.DeleteFlowControlFlowSeries(key.ID, strconv.Itoa(key.Priority))
484+
}
475485
}
476486
}
477487

‎pkg/epp/metrics/cardinality.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,20 @@ func (b *boundedLabel) pin(v string) {
9797

9898
var modelLabelLimiter = newBoundedLabel(maxModelLabelValues)
9999

100+
// Fairness IDs are populated from a client request header (or an agent-identity attribute), so
101+
// like model names their cardinality is not operator-bounded. They label per-request and
102+
// flow-control metrics; without a cap, every distinct fairness ID ever observed permanently
103+
// grows the time series set. maxFairnessLabelValues bounds the distinct fairness_id label
104+
// values; values beyond the cap collapse to overflowValue.
105+
const maxFairnessLabelValues = 1000
106+
107+
var fairnessLabelLimiter = newBoundedLabel(maxFairnessLabelValues)
108+
109+
// boundFairnessID caps the request-derived fairness_id label.
110+
func boundFairnessID(fairnessID string) string {
111+
return fairnessLabelLimiter.bound(fairnessID)
112+
}
113+
100114
// PreAdmitModelLabels pins the given model names so they always emit their real
101115
// label value on model-labeled metrics, regardless of how many unconfigured
102116
// names have been admitted. The datastore calls this when InferenceModelRewrite

‎pkg/epp/metrics/cardinality_test.go‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package metrics
1919
import (
2020
"fmt"
2121
"testing"
22+
"time"
2223

2324
promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
2425
"github.com/stretchr/testify/require"
@@ -98,3 +99,56 @@ func TestRecordRequestCounterBoundsModelCardinality(t *testing.T) {
9899
require.LessOrEqualf(t, count, testCap+1,
99100
"model_name cardinality must stay bounded by the cap, got %d series", count)
100101
}
102+
103+
// The fairness_id label is populated from a client request header, so the package-level limiter
104+
// must collapse an unbounded flood of distinct IDs into the overflow bucket instead of minting a
105+
// series per ID.
106+
func TestFairnessLabelFloodCollapsesToOverflow(t *testing.T) {
107+
const testCap = 5
108+
old := fairnessLabelLimiter
109+
fairnessLabelLimiter = newBoundedLabel(testCap)
110+
flowControlRequestEnqueueDuration.Reset()
111+
llmdFlowControlRequestEnqueueDuration.Reset()
112+
t.Cleanup(func() {
113+
fairnessLabelLimiter = old
114+
flowControlRequestEnqueueDuration.Reset()
115+
llmdFlowControlRequestEnqueueDuration.Reset()
116+
})
117+
118+
for i := 0; i < 1000; i++ {
119+
RecordFlowControlRequestEnqueueDuration(fmt.Sprintf("tenant-%d", i), "0", "Dispatched", time.Millisecond)
120+
}
121+
122+
// testCap admitted IDs + 1 overflow series, per family.
123+
require.Equal(t, testCap+1, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration),
124+
"1000 distinct fairness IDs must collapse to cap+overflow series, not one series each")
125+
require.Equal(t, testCap+1, promtestutil.CollectAndCount(llmdFlowControlRequestEnqueueDuration),
126+
"the llm_d_epp family must be bounded identically")
127+
}
128+
129+
// DeleteFlowControlFlowSeries backs the flow registry's GC hook: once a flow is collected, its
130+
// series must not linger for the lifetime of the process.
131+
func TestDeleteFlowControlFlowSeries(t *testing.T) {
132+
old := fairnessLabelLimiter
133+
fairnessLabelLimiter = newBoundedLabel(10)
134+
flowControlRequestEnqueueDuration.Reset()
135+
llmdFlowControlRequestEnqueueDuration.Reset()
136+
t.Cleanup(func() {
137+
fairnessLabelLimiter = old
138+
flowControlRequestEnqueueDuration.Reset()
139+
llmdFlowControlRequestEnqueueDuration.Reset()
140+
})
141+
142+
RecordFlowControlRequestEnqueueDuration("tenant-a", "0", "Dispatched", time.Millisecond)
143+
RecordFlowControlRequestEnqueueDuration("tenant-a", "0", "Rejected", time.Millisecond)
144+
RecordFlowControlRequestEnqueueDuration("tenant-b", "0", "Dispatched", time.Millisecond)
145+
require.Equal(t, 3, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration),
146+
"setup: expected one series per (fairness_id, outcome) pair")
147+
148+
DeleteFlowControlFlowSeries("tenant-a", "0")
149+
150+
require.Equal(t, 1, promtestutil.CollectAndCount(flowControlRequestEnqueueDuration),
151+
"all of tenant-a's series (every outcome) must be pruned; tenant-b's must survive")
152+
require.Equal(t, 1, promtestutil.CollectAndCount(llmdFlowControlRequestEnqueueDuration),
153+
"the llm_d_epp family must be pruned identically")
154+
}

‎pkg/epp/metrics/metrics.go‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -554,6 +554,7 @@ func Reset() {
554554
// RecordRequestCounter records the number of requests.
555555
func RecordRequestCounter(modelName, targetModelName, fairnessID string, priority int) {
556556
modelName, targetModelName = boundModels(modelName, targetModelName)
557+
fairnessID = boundFairnessID(fairnessID)
557558
prioStr := strconv.Itoa(priority)
558559
requestCounter.WithLabelValues(modelName, targetModelName, prioStr).Inc()
559560
llmdRequestCounter.WithLabelValues(modelName, targetModelName, fairnessID, prioStr).Inc()
@@ -562,6 +563,7 @@ func RecordRequestCounter(modelName, targetModelName, fairnessID string, priorit
562563
// RecordRequestErrCounter records the number of error requests.
563564
func RecordRequestErrCounter(modelName, targetModelName, fairnessID, priority string, code string) {
564565
modelName, targetModelName = boundModels(modelName, targetModelName)
566+
fairnessID = boundFairnessID(fairnessID)
565567
if code != "" {
566568
requestErrCounter.WithLabelValues(modelName, targetModelName, code).Inc()
567569
llmdRequestErrCounter.WithLabelValues(modelName, targetModelName, fairnessID, priority, code).Inc()
@@ -571,13 +573,15 @@ func RecordRequestErrCounter(modelName, targetModelName, fairnessID, priority st
571573
// RecordRequestSizes records the request sizes.
572574
func RecordRequestSizes(modelName, targetModelName, fairnessID, priority string, reqSize int) {
573575
modelName, targetModelName = boundModels(modelName, targetModelName)
576+
fairnessID = boundFairnessID(fairnessID)
574577
requestSizes.WithLabelValues(modelName, targetModelName).Observe(float64(reqSize))
575578
llmdRequestSizes.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(reqSize))
576579
}
577580

578581
// RecordRequestLatencies records duration of request.
579582
func RecordRequestLatencies(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, complete time.Time) bool {
580583
modelName, targetModelName = boundModels(modelName, targetModelName)
584+
fairnessID = boundFairnessID(fairnessID)
581585
if !complete.After(received) {
582586
log.FromContext(ctx).V(logutil.DEFAULT).Error(nil, "Request latency values are invalid",
583587
"modelName", modelName, "targetModelName", targetModelName, "completeTime", complete, "receivedTime", received)
@@ -592,13 +596,15 @@ func RecordRequestLatencies(ctx context.Context, modelName, targetModelName, fai
592596
// RecordResponseSizes records the response sizes.
593597
func RecordResponseSizes(modelName, targetModelName, fairnessID, priority string, size int) {
594598
modelName, targetModelName = boundModels(modelName, targetModelName)
599+
fairnessID = boundFairnessID(fairnessID)
595600
responseSizes.WithLabelValues(modelName, targetModelName).Observe(float64(size))
596601
llmdResponseSizes.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size))
597602
}
598603

599604
// RecordInputTokens records input tokens count.
600605
func RecordInputTokens(modelName, targetModelName, fairnessID, priority string, size int) {
601606
modelName, targetModelName = boundModels(modelName, targetModelName)
607+
fairnessID = boundFairnessID(fairnessID)
602608
if size > 0 {
603609
inputTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size))
604610
llmdInputTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size))
@@ -608,6 +614,7 @@ func RecordInputTokens(modelName, targetModelName, fairnessID, priority string,
608614
// RecordOutputTokens records output tokens count.
609615
func RecordOutputTokens(modelName, targetModelName, fairnessID, priority string, size int) {
610616
modelName, targetModelName = boundModels(modelName, targetModelName)
617+
fairnessID = boundFairnessID(fairnessID)
611618
if size > 0 {
612619
outputTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size))
613620
llmdOutputTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size))
@@ -617,13 +624,15 @@ func RecordOutputTokens(modelName, targetModelName, fairnessID, priority string,
617624
// RecordPromptCachedTokens records prompt cached tokens count.
618625
func RecordPromptCachedTokens(modelName, targetModelName, fairnessID, priority string, size int) {
619626
modelName, targetModelName = boundModels(modelName, targetModelName)
627+
fairnessID = boundFairnessID(fairnessID)
620628
promptCachedTokens.WithLabelValues(modelName, targetModelName).Observe(float64(size))
621629
llmdPromptCachedTokens.WithLabelValues(modelName, targetModelName, fairnessID, priority).Observe(float64(size))
622630
}
623631

624632
// RecordNormalizedTimePerOutputToken (NTPOT) records the normalized time per output token.
625633
func RecordNormalizedTimePerOutputToken(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, complete time.Time, outputTokenCount int) bool {
626634
modelName, targetModelName = boundModels(modelName, targetModelName)
635+
fairnessID = boundFairnessID(fairnessID)
627636
if outputTokenCount <= 0 {
628637
return false
629638
}
@@ -645,6 +654,7 @@ func RecordNormalizedTimePerOutputToken(ctx context.Context, modelName, targetMo
645654
// RecordRequestTTFT records the time to first token.
646655
func RecordRequestTTFT(ctx context.Context, modelName, targetModelName, fairnessID, priority string, streaming bool, received time.Time, firstToken time.Time) bool {
647656
modelName, targetModelName = boundModels(modelName, targetModelName)
657+
fairnessID = boundFairnessID(fairnessID)
648658
if firstToken.IsZero() {
649659
return false
650660
}
@@ -666,6 +676,7 @@ func RecordRequestTTFT(ctx context.Context, modelName, targetModelName, fairness
666676
// RecordRequestTPOT records the average time per output token.
667677
func RecordRequestTPOT(ctx context.Context, modelName, targetModelName, fairnessID, priority string, received time.Time, firstToken time.Time, complete time.Time, outputTokenCount int) bool {
668678
modelName, targetModelName = boundModels(modelName, targetModelName)
679+
fairnessID = boundFairnessID(fairnessID)
669680
if firstToken.IsZero() || outputTokenCount <= 1 {
670681
return false
671682
}
@@ -686,6 +697,7 @@ func RecordRequestTPOT(ctx context.Context, modelName, targetModelName, fairness
686697
// RecordInterTokenLatency records the time between consecutive response body chunks for streaming requests.
687698
func RecordInterTokenLatency(ctx context.Context, modelName, targetModelName, fairnessID, priority string, itlSeconds float64) bool {
688699
modelName, targetModelName = boundModels(modelName, targetModelName)
700+
fairnessID = boundFairnessID(fairnessID)
689701
if itlSeconds < 0 {
690702
log.FromContext(ctx).Error(nil, "Inter-token latency value must be non-negative",
691703
"modelName", modelName, "targetModelName", targetModelName, "itlSeconds", itlSeconds)
@@ -698,6 +710,7 @@ func RecordInterTokenLatency(ctx context.Context, modelName, targetModelName, fa
698710
// IncRunningRequests increases the current running requests.
699711
func IncRunningRequests(modelName, targetModelName, fairnessID, priority string) {
700712
modelName, targetModelName = boundModels(modelName, targetModelName)
713+
fairnessID = boundFairnessID(fairnessID)
701714
if modelName != "" {
702715
runningRequests.WithLabelValues(modelName).Inc()
703716
llmdRunningRequests.WithLabelValues(modelName, targetModelName, fairnessID, priority).Inc()
@@ -707,6 +720,7 @@ func IncRunningRequests(modelName, targetModelName, fairnessID, priority string)
707720
// DecRunningRequests decreases the current running requests.
708721
func DecRunningRequests(modelName, targetModelName, fairnessID, priority string) {
709722
modelName, targetModelName = boundModels(modelName, targetModelName)
723+
fairnessID = boundFairnessID(fairnessID)
710724
if modelName != "" {
711725
runningRequests.WithLabelValues(modelName).Dec()
712726
llmdRunningRequests.WithLabelValues(modelName, targetModelName, fairnessID, priority).Dec()
@@ -802,6 +816,8 @@ func RecordFlowControlRequestQueueDuration(
802816
modelName, targetModelName string,
803817
duration time.Duration,
804818
) {
819+
fairnessID = boundFairnessID(fairnessID)
820+
modelName, targetModelName = boundModels(modelName, targetModelName)
805821
flowControlRequestQueueDuration.WithLabelValues(
806822
fairnessID, priority, outcome,
807823
inferencePool,
@@ -826,6 +842,7 @@ func RecordFlowControlRequestEnqueueDuration(
826842
fairnessID string, priority string, outcome string,
827843
duration time.Duration,
828844
) {
845+
fairnessID = boundFairnessID(fairnessID)
829846
flowControlRequestEnqueueDuration.WithLabelValues(
830847
fairnessID, priority, outcome,
831848
).Observe(duration.Seconds())
@@ -838,27 +855,31 @@ func RecordFlowControlRequestEnqueueDuration(
838855
// IncFlowControlQueueSize increments the Flow Control queue size gauge.
839856
func IncFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) {
840857
modelName, targetModelName = boundModels(modelName, targetModelName)
858+
fairnessID = boundFairnessID(fairnessID)
841859
flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc()
842860
llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Inc()
843861
}
844862

845863
// DecFlowControlQueueSize decrements the Flow Control queue size gauge.
846864
func DecFlowControlQueueSize(fairnessID, priority, inferencePool, modelName, targetModelName string) {
847865
modelName, targetModelName = boundModels(modelName, targetModelName)
866+
fairnessID = boundFairnessID(fairnessID)
848867
flowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec()
849868
llmdFlowControlQueueSize.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Dec()
850869
}
851870

852871
// AddFlowControlQueueBytes increments the Flow Control queue bytes gauge.
853872
func AddFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) {
854873
modelName, targetModelName = boundModels(modelName, targetModelName)
874+
fairnessID = boundFairnessID(fairnessID)
855875
flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes))
856876
llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Add(float64(bytes))
857877
}
858878

859879
// SubFlowControlQueueBytes decrements the Flow Control queue bytes gauge.
860880
func SubFlowControlQueueBytes(fairnessID, priority, inferencePool, modelName, targetModelName string, bytes uint64) {
861881
modelName, targetModelName = boundModels(modelName, targetModelName)
882+
fairnessID = boundFairnessID(fairnessID)
862883
flowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes))
863884
llmdFlowControlQueueBytes.WithLabelValues(fairnessID, priority, inferencePool, modelName, targetModelName).Sub(float64(bytes))
864885
}
@@ -874,6 +895,23 @@ func IncFlowControlRequestsTotal(outcome, priority, inferencePool string) {
874895
llmdFlowControlRequestsTotal.WithLabelValues(outcome, priority, inferencePool).Inc()
875896
}
876897

898+
// DeleteFlowControlFlowSeries removes every flow-control series labeled with the given fairness ID
899+
// and priority, across both the deprecated and the llm_d_epp metric families. The fairness ID is
900+
// derived from client input, so its cardinality is unbounded; the flow registry calls this when it
901+
// garbage-collects an idle flow so that the metric vectors track live flows instead of growing
902+
// monotonically with every fairness ID ever observed.
903+
func DeleteFlowControlFlowSeries(fairnessID, priority string) {
904+
labels := prometheus.Labels{"fairness_id": fairnessID, "priority": priority}
905+
flowControlRequestQueueDuration.DeletePartialMatch(labels)
906+
flowControlRequestEnqueueDuration.DeletePartialMatch(labels)
907+
flowControlQueueSize.DeletePartialMatch(labels)
908+
flowControlQueueBytes.DeletePartialMatch(labels)
909+
llmdFlowControlRequestQueueDuration.DeletePartialMatch(labels)
910+
llmdFlowControlRequestEnqueueDuration.DeletePartialMatch(labels)
911+
llmdFlowControlQueueSize.DeletePartialMatch(labels)
912+
llmdFlowControlQueueBytes.DeletePartialMatch(labels)
913+
}
914+
877915
// RecordInferenceModelRewriteDecision records the routing decision for InferenceModelRewrite.
878916
// The rewrite name and target come from configuration; only the source model name is
879917
// request-derived and needs bounding (a generic rule matches arbitrary model names).

0 commit comments

Comments
 (0)