From 62ee968748c2b0231e19b20a2f0d53946cb34bcc Mon Sep 17 00:00:00 2001 From: chethanuk Date: Sat, 20 Jun 2026 23:27:18 +0400 Subject: [PATCH 1/2] Add EPP in-flight requests and tokens metrics (#1089) Implements #1089 (xref GAIE #2072): EPP in-flight observability, owned by the in-flight load producer plugin. - Per-endpoint llm_d_epp_inflight_requests / llm_d_epp_inflight_tokens gauges via a custom prometheus.Collector reading the InFlightLoadProducer snapshot. Registered through the plugin metrics recorder (handle.Metrics()) with producer_name as a constant descriptor label, so multiple producers register without collision. - Per-model llm_d_epp_request_inflight gauge owned by the producer: incremented in PreRequest, decremented exactly once via a per-request PluginState entry's OnEvicted, so it stays balanced across completion, error, abort, and reaper paths. Mirrors the metric-ownership pattern of the disagg handler (profilehandler/disagg/metrics.go). - Adds InferenceRequest.IncomingModel so the producer can label the gauge with the incoming model name. - docs/metrics.md documents all three metrics. Fixes #1089 Signed-off-by: ChethanUK --- docs/metrics.md | 42 +++ docs/operations.md | 4 +- .../framework/interface/scheduling/types.go | 2 + .../dataproducer/inflightload/metrics.go | 101 +++++++ .../dataproducer/inflightload/metrics_test.go | 273 ++++++++++++++++++ .../dataproducer/inflightload/producer.go | 67 ++++- pkg/epp/metrics/collectors/inflight_load.go | 94 ++++++ .../metrics/collectors/inflight_load_test.go | 84 ++++++ pkg/epp/requestcontrol/director.go | 1 + 9 files changed, 664 insertions(+), 4 deletions(-) create mode 100644 pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go create mode 100644 pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go create mode 100644 pkg/epp/metrics/collectors/inflight_load.go create mode 100644 pkg/epp/metrics/collectors/inflight_load_test.go diff --git a/docs/metrics.md b/docs/metrics.md index 5b968f8c65..462dd8581e 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -68,3 +68,45 @@ Three metrics covering ext_proc gRPC stream lifecycle. Disabled by default; enab * **Release Stage:** ALPHA * **Description:** Total ext_proc gRPC streams completed, by gRPC status code. * **Usage:** Rate of `code="OK"` is the healthy stream-completion rate. A rising rate of `code="Internal"` or `code="Unknown"` indicates handler errors. `code="Canceled"` is expected on Envoy restarts and rolling EPP updates. + +## In-flight load metrics + +In-flight load, emitted under the `llm_d_epp_` prefix. Present only when an `InFlightLoadProducer` is +configured: the producer owns these metrics and registers them through the plugin metrics recorder. The +per-endpoint gauges read the producer's live per-endpoint counters (the same source as the +`/debug/plugins/state` dump and the token-load scorer); the per-model `request_inflight` gauge is moved +by the producer as requests are admitted and completed. + +### `inflight_requests` + +* **Type:** Gauge +* **Labels:** + * `endpoint_name`: string — the target endpoint (pod) name. + * `namespace`: string — the endpoint's namespace. + * `producer_name`: string — the configured `InFlightLoadProducer` instance name, so multiple producers emit distinct series. +* **Release Stage:** ALPHA +* **Description:** Requests currently in flight on each endpoint (scheduled, not yet completed), as tracked by the in-flight load producer. +* **Usage:** Per-replica queue depth for load-aware routing and capacity analysis. Unlike the per-model `request_inflight` gauge (admitted-but-not-completed, aggregated by model), this is broken down by endpoint so it shows which replica is loaded. + +### `inflight_tokens` + +* **Type:** Gauge +* **Labels:** + * `endpoint_name`: string — the target endpoint (pod) name. + * `namespace`: string — the endpoint's namespace. + * `producer_name`: string — the configured `InFlightLoadProducer` instance name. +* **Release Stage:** ALPHA +* **Description:** Tokens currently in flight on each endpoint — uncached prompt tokens, optionally plus estimated output tokens when the producer's `addEstimatedOutputTokens` is set. +* **Usage:** Per-replica token pressure, a finer load signal than request count when request sizes vary widely. + +### `request_inflight` + +* **Type:** Gauge +* **Labels:** + * `model_name`: string — the model named in the request body. + * `target_model_name`: string — the target model after traffic split. + * `fairness_id`: string — the flow-control fairness queue identity. + * `priority`: string — the request priority. +* **Release Stage:** ALPHA +* **Description:** Requests admitted to the endpoint picker but not yet completed, aggregated by model. +* **Usage:** Picker-wide concurrency by model. Unlike the per-endpoint `inflight_requests` gauge, this is aggregated across endpoints, so it answers "how much is in flight for this model" rather than "which replica is loaded". diff --git a/docs/operations.md b/docs/operations.md index f6df65b080..08083f846e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -17,7 +17,7 @@ The EPP acts as the routing intelligence engine. Its resource usage scales prima - **Idle CPU Scaling**: Idle CPU usage of the EPP container scales with the number of model-serving pods in the cluster due to continuous metric scraping. For example, in a cluster with 100 model-serving pods, the idle CPU usage of the EPP container grows to approximately **7.5 cores**. #### Memory Allocation -- **Base Memory**: EPP memory usage is relatively low and stable with small output token requests, but scales with the number of concurrent inflight requests. +- **Base Memory**: EPP memory usage is relatively low and stable with small output token requests. - **Inflight Requests Impact**: Memory usage increases with the number of concurrent inflight requests and the output (decode) token length. - **Sizing Guidelines**: - For a request rate of 50 to 100 requests/second with 1k output tokens, EPP requires between **4 GiB and 6 GiB** of memory. @@ -45,7 +45,7 @@ The EPP's scaling behavior and effectiveness are highly dependent on the configu The following tables present empirical benchmark results for EPP running with llm-d-simulator simulating Qwen/Qwen3-8B. #### Throughput and Prefix Block Sizing -This table shows peak CPU and memory utilization for EPP under a 100k token workload (95k system prompt, 5k question prompt, and 1k output tokens) when using approximate prefix caching across 100 model-serving pods. +This table shows peak CPU and memory utilization for EPP under a 100k input token workload (95k system prompt and 5k question prompt) with 1k output tokens when using approximate prefix caching across 100 model-serving pods. | Configuration | Request Rate (Req/s) | maxPrefixBlocksToMatch | Peak CPU (Cores) | Peak Memory (GiB) | Scheduler P50 Latency (s) | | :--- | :--- | :--- | :--- | :--- | :--- | diff --git a/pkg/epp/framework/interface/scheduling/types.go b/pkg/epp/framework/interface/scheduling/types.go index 056c8e9029..14832060d2 100644 --- a/pkg/epp/framework/interface/scheduling/types.go +++ b/pkg/epp/framework/interface/scheduling/types.go @@ -45,6 +45,8 @@ type RequestObjectives struct { type InferenceRequest struct { // RequestID is the Envoy generated Id for the request being processed RequestID string + // IncomingModel is the model named in the request body, before any traffic split. + IncomingModel string // TargetModel is the final target model after traffic split. TargetModel string // Data contains the request-body fields that we parse out as user input. diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go new file mode 100644 index 0000000000..7fb9ebfe85 --- /dev/null +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go @@ -0,0 +1,101 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package inflightload + +import ( + "errors" + "fmt" + "strconv" + + "github.com/prometheus/client_golang/prometheus" + compbasemetrics "k8s.io/component-base/metrics" + + metricsutil "github.com/llm-d/llm-d-router/pkg/common/observability/metrics" + fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" + "github.com/llm-d/llm-d-router/pkg/epp/metadata" + eppmetrics "github.com/llm-d/llm-d-router/pkg/epp/metrics" +) + +// requestInflight tracks requests admitted to the endpoint picker but not yet +// completed, aggregated per model. The producer owns the gauge: it increments +// in PreRequest and decrements via the per-request state entry's OnEvicted, so +// the count stays balanced across completion, error, and reaper paths. +var requestInflight = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: eppmetrics.LLMDRouterEndpointPickerSubsystem, + Name: "request_inflight", + Help: metricsutil.HelpMsgWithStability("Current number of in-flight requests in the endpoint picker (admitted, not yet completed).", compbasemetrics.ALPHA), + }, + []string{"model_name", "target_model_name", "fairness_id", "priority"}, +) + +// registerMetrics registers the producer-owned metrics with the plugin's metrics +// recorder. The per-model gauge is shared across producers and the per-endpoint +// collector carries a producer-specific descriptor, so an already-registered +// equivalent collector is a benign no-op. +func registerMetrics(registerer prometheus.Registerer, collector prometheus.Collector) error { + if registerer == nil { + return errors.New("inflight load metrics registerer is required") + } + for _, c := range []prometheus.Collector{requestInflight, collector} { + if err := registerer.Register(c); err != nil { + var alreadyRegistered prometheus.AlreadyRegisteredError + if errors.As(err, &alreadyRegistered) { + continue + } + return fmt.Errorf("register inflight load metric: %w", err) + } + } + return nil +} + +// requestInflightLabels carries the gauge label values captured at admission so +// the decrement, which may run on a background reaper goroutine, uses the same +// series the increment did. +type requestInflightLabels struct { + modelName string + targetModelName string + fairnessID string + priority string +} + +func newRequestInflightLabels(request *fwksched.InferenceRequest) requestInflightLabels { + fairnessID := request.FairnessID + if fairnessID == "" { + fairnessID = metadata.DefaultFairnessID + } + return requestInflightLabels{ + modelName: request.IncomingModel, + targetModelName: request.TargetModel, + fairnessID: fairnessID, + priority: strconv.Itoa(request.Objectives.Priority), + } +} + +func incRequestInflight(l requestInflightLabels) { + if l.modelName == "" { + return + } + requestInflight.WithLabelValues(l.modelName, l.targetModelName, l.fairnessID, l.priority).Inc() +} + +func decRequestInflight(l requestInflightLabels) { + if l.modelName == "" { + return + } + requestInflight.WithLabelValues(l.modelName, l.targetModelName, l.fairnessID, l.priority).Dec() +} diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go new file mode 100644 index 0000000000..bb822dd42e --- /dev/null +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go @@ -0,0 +1,273 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package inflightload + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + + "github.com/prometheus/client_golang/prometheus" + promtestutil "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + + fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requestcontrol" + fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling" + fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" + "github.com/llm-d/llm-d-router/pkg/epp/metadata" + "github.com/llm-d/llm-d-router/pkg/epp/metrics/collectors" +) + +func makeInflightRequest(requestID, incomingModel, fairnessID string, priority int) *fwksched.InferenceRequest { + return &fwksched.InferenceRequest{ + RequestID: requestID, + IncomingModel: incomingModel, + TargetModel: "t1", + FairnessID: fairnessID, + Objectives: fwksched.RequestObjectives{Priority: priority}, + Body: &fwkrh.InferenceRequestBody{ + TokenizedPrompt: &fwkrh.TokenizedPrompt{PerPromptTokens: [][]uint32{make([]uint32, 4)}}, + }, + } +} + +// gaugeFor reads the gauge value for a request's labels via the production label logic. +func gaugeFor(req *fwksched.InferenceRequest) float64 { + l := newRequestInflightLabels(req) + return promtestutil.ToFloat64(requestInflight.WithLabelValues(l.modelName, l.targetModelName, l.fairnessID, l.priority)) +} + +// The producer increments the per-model gauge in PreRequest and exposes it under the documented +// name and label set. +func TestRequestInflight_Golden(t *testing.T) { + requestInflight.Reset() + producer := newTestProducer(t) + + req := makeInflightRequest("req1", "m1", "tenant-x", 10) + producer.PreRequest(context.Background(), req, makeSchedulingResult("ep1")) + + expected := ` + # HELP llm_d_epp_request_inflight [ALPHA] Current number of in-flight requests in the endpoint picker (admitted, not yet completed). + # TYPE llm_d_epp_request_inflight gauge + llm_d_epp_request_inflight{fairness_id="tenant-x",model_name="m1",priority="10",target_model_name="t1"} 1 + ` + require.NoError(t, promtestutil.CollectAndCompare(requestInflight, strings.NewReader(expected), "llm_d_epp_request_inflight")) +} + +// Every terminal path decrements the gauge exactly once. Unlike the upstream defer-Dec pattern, the +// producer splits admission (PreRequest) from completion (ResponseBody), so the decrement rides on +// PluginState eviction: end-of-stream Delete and the janitor reaper must each fire it once, with no +// leak on abort/disconnect and no double-decrement when both paths run. +func TestRequestInflight_CleanupPaths(t *testing.T) { + endOfStream := func(p *InFlightLoadProducer, req *fwksched.InferenceRequest, res *fwksched.SchedulingResult) { + req.SchedulingResult = res + p.ResponseBody(context.Background(), req, &requestcontrol.Response{EndOfStream: true}, nil) + } + reap := func(p *InFlightLoadProducer, req *fwksched.InferenceRequest, _ *fwksched.SchedulingResult) { + p.PluginState.Delete(req.RequestID) + } + + tests := []struct { + name string + trigger func(*InFlightLoadProducer, *fwksched.InferenceRequest, *fwksched.SchedulingResult) + want float64 + }{ + {"end of stream decrements", endOfStream, 0}, + {"reaper decrements (no end of stream)", reap, 0}, + { + name: "end of stream then reaper does not go negative", + trigger: func(p *InFlightLoadProducer, req *fwksched.InferenceRequest, res *fwksched.SchedulingResult) { + endOfStream(p, req, res) + reap(p, req, res) + }, + want: 0, + }, + { + name: "reaper then end of stream does not double-decrement", + trigger: func(p *InFlightLoadProducer, req *fwksched.InferenceRequest, res *fwksched.SchedulingResult) { + reap(p, req, res) + endOfStream(p, req, res) + }, + want: 0, + }, + { + name: "start of stream holds the request in flight", + trigger: func(p *InFlightLoadProducer, req *fwksched.InferenceRequest, res *fwksched.SchedulingResult) { + req.SchedulingResult = res + p.ResponseBody(context.Background(), req, &requestcontrol.Response{StartOfStream: true}, nil) + }, + want: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + requestInflight.Reset() + producer := newTestProducer(t) + req := makeInflightRequest("req-"+tt.name, "m1", "tenant-x", 10) + res := makeSchedulingResult("ep1") + + producer.PreRequest(context.Background(), req, res) + require.Equal(t, float64(1), gaugeFor(req), "PreRequest should increment once") + + tt.trigger(producer, req, res) + require.Equal(t, tt.want, gaugeFor(req)) + }) + } +} + +// Requests that are not admitted to an endpoint, or carry no model name, are not counted and write no +// gauge state entry. +func TestRequestInflight_NotCounted(t *testing.T) { + tests := []struct { + name string + req *fwksched.InferenceRequest + result *fwksched.SchedulingResult + }{ + {"empty incoming model", makeInflightRequest("req-empty-model", "", "tenant-x", 10), makeSchedulingResult("ep1")}, + {"nil scheduling result", makeInflightRequest("req-nil-result", "m1", "tenant-x", 10), nil}, + {"empty profile results", makeInflightRequest("req-empty-result", "m1", "tenant-x", 10), &fwksched.SchedulingResult{}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + requestInflight.Reset() + producer := newTestProducer(t) + + producer.PreRequest(context.Background(), tt.req, tt.result) + + require.Equal(t, float64(0), gaugeFor(tt.req)) + _, err := producer.PluginState.Read(tt.req.RequestID, requestInflightStateKey) + require.ErrorIs(t, err, fwkplugin.ErrNotFound, "no gauge state entry should be written") + }) + } +} + +// A request fanned out across multiple profiles/endpoints is one in-flight request: the gauge moves by +// one, not once per endpoint. +func TestRequestInflight_IncrementsOncePerRequest(t *testing.T) { + requestInflight.Reset() + producer := newTestProducer(t) + ctx := context.Background() + + req := makeInflightRequest("req-multi", "m1", "tenant-x", 10) + res := &fwksched.SchedulingResult{ + PrimaryProfileName: "prefill", + ProfileResults: map[string]*fwksched.ProfileRunResult{ + "prefill": {TargetEndpoints: []fwksched.Endpoint{newStubSchedulingEndpoint("pod-a")}}, + "decode": {TargetEndpoints: []fwksched.Endpoint{newStubSchedulingEndpoint("pod-b")}}, + }, + } + + producer.PreRequest(ctx, req, res) + require.Equal(t, float64(1), gaugeFor(req)) + + req.SchedulingResult = res + producer.ResponseBody(ctx, req, &requestcontrol.Response{EndOfStream: true}, nil) + require.Equal(t, float64(0), gaugeFor(req)) +} + +// Concurrent admit/complete cycles balance to zero (run under -race for the OnEvicted/atomic guard). +func TestRequestInflight_ConcurrentBalanced(t *testing.T) { + requestInflight.Reset() + producer := newTestProducer(t) + ctx := context.Background() + + const n = 50 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func(i int) { + defer wg.Done() + req := makeInflightRequest(fmt.Sprintf("req-%d", i), "m1", "tenant-x", 10) + res := makeSchedulingResult("ep1") + producer.PreRequest(ctx, req, res) + req.SchedulingResult = res + producer.ResponseBody(ctx, req, &requestcontrol.Response{EndOfStream: true}, nil) + }(i) + } + wg.Wait() + + require.Equal(t, float64(0), promtestutil.ToFloat64(requestInflight.WithLabelValues("m1", "t1", "tenant-x", "10"))) +} + +// newRequestInflightLabels maps a request onto the gauge label set, defaulting an empty fairness id and +// stringifying the priority. +func TestNewRequestInflightLabels(t *testing.T) { + tests := []struct { + name string + req *fwksched.InferenceRequest + want requestInflightLabels + }{ + { + name: "all fields set", + req: makeInflightRequest("r", "m1", "tenant-x", 10), + want: requestInflightLabels{modelName: "m1", targetModelName: "t1", fairnessID: "tenant-x", priority: "10"}, + }, + { + name: "empty fairness defaults", + req: makeInflightRequest("r", "m1", "", 0), + want: requestInflightLabels{modelName: "m1", targetModelName: "t1", fairnessID: metadata.DefaultFairnessID, priority: "0"}, + }, + { + name: "negative priority stringified", + req: makeInflightRequest("r", "m1", "tenant-x", -1), + want: requestInflightLabels{modelName: "m1", targetModelName: "t1", fairnessID: "tenant-x", priority: "-1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.want, newRequestInflightLabels(tt.req)) + }) + } +} + +// registerMetrics is idempotent for the shared gauge and registers a distinct per-producer collector, +// so two producers register into one registry and both emit per-endpoint series. +func TestRegisterMetrics_MultiProducer(t *testing.T) { + requestInflight.Reset() + reg := prometheus.NewRegistry() + + a := &fakeSnapshotter{requests: map[string]int64{"ns1/ep1": 1}} + b := &fakeSnapshotter{requests: map[string]int64{"ns2/ep2": 2}} + require.NoError(t, registerMetrics(reg, collectors.NewInFlightLoadCollector("a", a))) + require.NoError(t, registerMetrics(reg, collectors.NewInFlightLoadCollector("b", b))) + + expected := ` +# HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_requests gauge +llm_d_epp_inflight_requests{endpoint_name="ep1",namespace="ns1",producer_name="a"} 1 +llm_d_epp_inflight_requests{endpoint_name="ep2",namespace="ns2",producer_name="b"} 2 +` + require.NoError(t, promtestutil.GatherAndCompare(reg, strings.NewReader(expected), "llm_d_epp_inflight_requests")) +} + +func TestRegisterMetrics_NilRegisterer(t *testing.T) { + require.Error(t, registerMetrics(nil, nil)) +} + +type fakeSnapshotter struct { + requests map[string]int64 + tokens map[string]int64 +} + +func (f *fakeSnapshotter) InFlightRequestsSnapshot() map[string]int64 { return f.requests } +func (f *fakeSnapshotter) InFlightTokensSnapshot() map[string]int64 { return f.tokens } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go index 4d4e2399bf..bd29ebd325 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go @@ -37,12 +37,16 @@ import ( sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications" inflightloadconstants "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/constants" tokenproducer "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/tokenizer" + "github.com/llm-d/llm-d-router/pkg/epp/metrics/collectors" ) const ( InFlightLoadProducerType = inflightloadconstants.InFlightLoadProducerType profilePrefill = "prefill" maxDebugDumpEndpoints = 100 + // requestInflightStateKey holds the single per-request entry whose OnEvicted + // decrements the request_inflight gauge exactly once. + requestInflightStateKey fwkplugin.StateKey = "inflight-request-gauge" ) // Config controls optional behaviors of InFlightLoadProducer. @@ -77,7 +81,7 @@ func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkp } } - return &InFlightLoadProducer{ + producer := &InFlightLoadProducer{ typedName: fwkplugin.TypedName{Type: InFlightLoadProducerType, Name: name}, requestTracker: newConcurrencyTracker(), tokenTracker: newConcurrencyTracker(), @@ -87,7 +91,15 @@ func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkp prefixMatchInfoDK: attrprefix.PrefixCacheMatchInfoDataKey.WithNonEmptyProducerName(cfg.PrefixMatchInfoProducerName), uncachedRequestTokensDk: attrconcurrency.UncachedRequestTokensDataKey.WithNonEmptyProducerName(name), PluginState: fwkplugin.NewPluginState(ctx), - }, nil + } + + // Own the in-flight metrics: the per-model gauge and the per-endpoint + // collector are registered through the plugin's metrics recorder. + if err := registerMetrics(handle.Metrics(), collectors.NewInFlightLoadCollector(name, producer)); err != nil { + return nil, err + } + + return producer, nil } var ( @@ -175,6 +187,31 @@ func (e *addedTokensEntry) OnEvicted(_ string, _ fwkplugin.StateKey) { } } +// requestInflightEntry decrements the per-model request_inflight gauge once when +// the request leaves the picker. The decremented guard makes OnEvicted safe to +// fire from both the end-of-stream Delete and the janitor reaper. +type requestInflightEntry struct { + labels requestInflightLabels + decremented atomic.Bool +} + +var _ fwkplugin.EvictableStateData = (*requestInflightEntry)(nil) + +func (e *requestInflightEntry) Clone() fwkplugin.StateData { + if e == nil { + return nil + } + clone := &requestInflightEntry{labels: e.labels} + clone.decremented.Store(e.decremented.Load()) + return clone +} + +func (e *requestInflightEntry) OnEvicted(_ string, _ fwkplugin.StateKey) { + if e.decremented.CompareAndSwap(false, true) { + decRequestInflight(e.labels) + } +} + type inFlightLoadState struct { Endpoints []endpointInFlightLoadState `json:"endpoints"` TotalEndpoints int `json:"totalEndpoints"` @@ -350,6 +387,13 @@ func (p *InFlightLoadProducer) PreRequest(ctx context.Context, request *fwksched return } + // One per-model gauge increment per request; the state entry's OnEvicted + // decrements it exactly once on completion or reaper cleanup. + if labels := newRequestInflightLabels(request); labels.modelName != "" { + incRequestInflight(labels) + p.PluginState.Write(request.RequestID, requestInflightStateKey, &requestInflightEntry{labels: labels}) + } + inputTokens := p.tokenEstimator.EstimateInput(request) for profileName, profileResult := range result.ProfileResults { @@ -589,6 +633,25 @@ func (p *InFlightLoadProducer) GetRequests(eid string) int64 { return p.requestTracker.get(eid) } +// InFlightRequestsSnapshot returns a copy of the per-endpoint in-flight request +// counts, keyed by the endpoint's "namespace/name". For Prometheus collection. +func (p *InFlightLoadProducer) InFlightRequestsSnapshot() map[string]int64 { + if p.requestTracker == nil { + return nil + } + return p.requestTracker.snapshot() +} + +// InFlightTokensSnapshot returns a copy of the per-endpoint in-flight token +// counts (uncached prompt tokens, optionally plus estimated output), keyed by +// the endpoint's "namespace/name". For Prometheus collection. +func (p *InFlightLoadProducer) InFlightTokensSnapshot() map[string]int64 { + if p.tokenTracker == nil { + return nil + } + return p.tokenTracker.snapshot() +} + // concurrencyTracker manages thread-safe counters for inflight requests. type concurrencyTracker struct { mu sync.RWMutex diff --git a/pkg/epp/metrics/collectors/inflight_load.go b/pkg/epp/metrics/collectors/inflight_load.go new file mode 100644 index 0000000000..d30c69afe1 --- /dev/null +++ b/pkg/epp/metrics/collectors/inflight_load.go @@ -0,0 +1,94 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package collectors + +import ( + "strings" + + "github.com/prometheus/client_golang/prometheus" + compbasemetrics "k8s.io/component-base/metrics" + + metricsutil "github.com/llm-d/llm-d-router/pkg/common/observability/metrics" +) + +// InFlightLoadSnapshotter is the read side of an in-flight load producer the +// collector needs: per-endpoint in-flight request and token counts keyed by the +// endpoint's "namespace/name". +type InFlightLoadSnapshotter interface { + InFlightRequestsSnapshot() map[string]int64 + InFlightTokensSnapshot() map[string]int64 +} + +type inFlightLoadCollector struct { + requestsDesc *prometheus.Desc + tokensDesc *prometheus.Desc + snapshotter InFlightLoadSnapshotter +} + +var _ prometheus.Collector = &inFlightLoadCollector{} + +// NewInFlightLoadCollector returns a prometheus.Collector that emits per-endpoint +// in-flight request and token gauges from the given producer's live snapshot. +// producerName is a constant label baked into the descriptors, so each configured +// producer owns a distinct descriptor and multiple collectors register without +// collision. +func NewInFlightLoadCollector(producerName string, s InFlightLoadSnapshotter) prometheus.Collector { + constLabels := prometheus.Labels{"producer_name": producerName} + return &inFlightLoadCollector{ + requestsDesc: prometheus.NewDesc( + "llm_d_epp_inflight_requests", + metricsutil.HelpMsgWithStability("Current number of in-flight requests per endpoint, as tracked by the in-flight load producer.", compbasemetrics.ALPHA), + []string{"endpoint_name", "namespace"}, constLabels, + ), + tokensDesc: prometheus.NewDesc( + "llm_d_epp_inflight_tokens", + metricsutil.HelpMsgWithStability("Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer.", compbasemetrics.ALPHA), + []string{"endpoint_name", "namespace"}, constLabels, + ), + snapshotter: s, + } +} + +func (c *inFlightLoadCollector) Describe(ch chan<- *prometheus.Desc) { + ch <- c.requestsDesc + ch <- c.tokensDesc +} + +func (c *inFlightLoadCollector) Collect(ch chan<- prometheus.Metric) { + if c.snapshotter == nil { + return + } + c.emit(ch, c.requestsDesc, c.snapshotter.InFlightRequestsSnapshot()) + c.emit(ch, c.tokensDesc, c.snapshotter.InFlightTokensSnapshot()) +} + +func (c *inFlightLoadCollector) emit(ch chan<- prometheus.Metric, desc *prometheus.Desc, counts map[string]int64) { + for endpointID, v := range counts { + name, namespace := splitNamespacedName(endpointID) + ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(v), name, namespace) + } +} + +// splitNamespacedName splits a "namespace/name" key (the string form of a +// k8s types.NamespacedName) into its name and namespace. A key with no +// separator is treated as a bare name with an empty namespace. +func splitNamespacedName(id string) (name, namespace string) { + if i := strings.IndexByte(id, '/'); i >= 0 { + return id[i+1:], id[:i] + } + return id, "" +} diff --git a/pkg/epp/metrics/collectors/inflight_load_test.go b/pkg/epp/metrics/collectors/inflight_load_test.go new file mode 100644 index 0000000000..684df828d2 --- /dev/null +++ b/pkg/epp/metrics/collectors/inflight_load_test.go @@ -0,0 +1,84 @@ +/* +Copyright 2026 The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package collectors + +import ( + "strings" + "testing" + + promtestutil "github.com/prometheus/client_golang/prometheus/testutil" +) + +type fakeInFlightLoad struct { + requests map[string]int64 + tokens map[string]int64 +} + +func (f *fakeInFlightLoad) InFlightRequestsSnapshot() map[string]int64 { return f.requests } +func (f *fakeInFlightLoad) InFlightTokensSnapshot() map[string]int64 { return f.tokens } + +func TestInFlightLoadCollectorEmpty(t *testing.T) { + collector := NewInFlightLoadCollector("default", &fakeInFlightLoad{}) + if err := promtestutil.CollectAndCompare(collector, strings.NewReader("")); err != nil { + t.Fatal(err) + } +} + +func TestInFlightLoadCollectorPerEndpoint(t *testing.T) { + collector := NewInFlightLoadCollector("default", &fakeInFlightLoad{ + requests: map[string]int64{ + "ns1/ep1": 3, + "ns2/ep2": 5, + }, + tokens: map[string]int64{ + "ns1/ep1": 30, + "ns2/ep2": 50, + }, + }) + + expected := ` +# HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_requests gauge +llm_d_epp_inflight_requests{endpoint_name="ep1",namespace="ns1",producer_name="default"} 3 +llm_d_epp_inflight_requests{endpoint_name="ep2",namespace="ns2",producer_name="default"} 5 +# HELP llm_d_epp_inflight_tokens [ALPHA] Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_tokens gauge +llm_d_epp_inflight_tokens{endpoint_name="ep1",namespace="ns1",producer_name="default"} 30 +llm_d_epp_inflight_tokens{endpoint_name="ep2",namespace="ns2",producer_name="default"} 50 +` + + if err := promtestutil.CollectAndCompare(collector, strings.NewReader(expected), + "llm_d_epp_inflight_requests", "llm_d_epp_inflight_tokens"); err != nil { + t.Fatal(err) + } +} + +func TestSplitNamespacedName(t *testing.T) { + cases := []struct { + id, wantName, wantNS string + }{ + {"ns/ep", "ep", "ns"}, + {"bare", "bare", ""}, + {"ns/ep/extra", "ep/extra", "ns"}, + } + for _, c := range cases { + name, ns := splitNamespacedName(c.id) + if name != c.wantName || ns != c.wantNS { + t.Errorf("splitNamespacedName(%q) = (%q,%q), want (%q,%q)", c.id, name, ns, c.wantName, c.wantNS) + } + } +} diff --git a/pkg/epp/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index afe3ea7c6e..cbf1e4007a 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/pkg/epp/requestcontrol/director.go @@ -264,6 +264,7 @@ func (d *Director) HandleRequest(ctx context.Context, reqCtx *handlers.RequestCo // Prepare InferenceRequest (needed for both saturation detection and Scheduler) reqCtx.SchedulingRequest = &fwksched.InferenceRequest{ RequestID: reqCtx.Request.Headers[reqcommon.RequestIDHeaderKey], + IncomingModel: reqCtx.IncomingModelName, TargetModel: reqCtx.TargetModelName, Body: inferenceRequestBody, Headers: reqCtx.Request.Headers, From 8418f8c016dd00f69c0688dfdde09a7c0457963a Mon Sep 17 00:00:00 2001 From: ChethanUK Date: Fri, 17 Jul 2026 21:09:38 +0200 Subject: [PATCH 2/2] Replace in-flight load collector with plugin-owned gauges Signed-off-by: ChethanUK --- docs/metrics.md | 6 +- docs/operations.md | 4 +- .../dataproducer/inflightload/metrics.go | 46 +++++-- .../dataproducer/inflightload/metrics_test.go | 116 +++++++++++++++--- .../dataproducer/inflightload/producer.go | 86 ++++++++----- .../inflightload/producer_test.go | 8 +- pkg/epp/metrics/collectors/inflight_load.go | 94 -------------- .../metrics/collectors/inflight_load_test.go | 84 ------------- 8 files changed, 203 insertions(+), 241 deletions(-) delete mode 100644 pkg/epp/metrics/collectors/inflight_load.go delete mode 100644 pkg/epp/metrics/collectors/inflight_load_test.go diff --git a/docs/metrics.md b/docs/metrics.md index 45712a32c4..f3316fe9c3 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -168,9 +168,9 @@ Three metrics covering ext_proc gRPC stream lifecycle. Disabled by default; enab In-flight load, emitted under the `llm_d_epp_` prefix. Present only when an `InFlightLoadProducer` is configured: the producer owns these metrics and registers them through the plugin metrics recorder. The -per-endpoint gauges read the producer's live per-endpoint counters (the same source as the -`/debug/plugins/state` dump and the token-load scorer); the per-model `request_inflight` gauge is moved -by the producer as requests are admitted and completed. +per-endpoint gauges are updated from the producer's live per-endpoint counters as requests are admitted +and released (the same source as the `/debug/plugins/state` dump and the token-load scorer); the +per-model `request_inflight` gauge is moved by the producer as requests are admitted and completed. ### `inflight_requests` diff --git a/docs/operations.md b/docs/operations.md index 08083f846e..f6df65b080 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -17,7 +17,7 @@ The EPP acts as the routing intelligence engine. Its resource usage scales prima - **Idle CPU Scaling**: Idle CPU usage of the EPP container scales with the number of model-serving pods in the cluster due to continuous metric scraping. For example, in a cluster with 100 model-serving pods, the idle CPU usage of the EPP container grows to approximately **7.5 cores**. #### Memory Allocation -- **Base Memory**: EPP memory usage is relatively low and stable with small output token requests. +- **Base Memory**: EPP memory usage is relatively low and stable with small output token requests, but scales with the number of concurrent inflight requests. - **Inflight Requests Impact**: Memory usage increases with the number of concurrent inflight requests and the output (decode) token length. - **Sizing Guidelines**: - For a request rate of 50 to 100 requests/second with 1k output tokens, EPP requires between **4 GiB and 6 GiB** of memory. @@ -45,7 +45,7 @@ The EPP's scaling behavior and effectiveness are highly dependent on the configu The following tables present empirical benchmark results for EPP running with llm-d-simulator simulating Qwen/Qwen3-8B. #### Throughput and Prefix Block Sizing -This table shows peak CPU and memory utilization for EPP under a 100k input token workload (95k system prompt and 5k question prompt) with 1k output tokens when using approximate prefix caching across 100 model-serving pods. +This table shows peak CPU and memory utilization for EPP under a 100k token workload (95k system prompt, 5k question prompt, and 1k output tokens) when using approximate prefix caching across 100 model-serving pods. | Configuration | Request Rate (Req/s) | maxPrefixBlocksToMatch | Peak CPU (Cores) | Peak Memory (GiB) | Scheduler P50 Latency (s) | | :--- | :--- | :--- | :--- | :--- | :--- | diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go index 7fb9ebfe85..54b3edae99 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go @@ -20,6 +20,7 @@ import ( "errors" "fmt" "strconv" + "strings" "github.com/prometheus/client_golang/prometheus" compbasemetrics "k8s.io/component-base/metrics" @@ -43,18 +44,39 @@ var requestInflight = prometheus.NewGaugeVec( []string{"model_name", "target_model_name", "fairness_id", "priority"}, ) +// inflightRequests and inflightTokens track per-endpoint in-flight load. The +// producer's concurrency trackers set them from the live counter value on +// every counter update and delete the series when the endpoint is removed. +var inflightRequests = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: eppmetrics.LLMDRouterEndpointPickerSubsystem, + Name: "inflight_requests", + Help: metricsutil.HelpMsgWithStability("Current number of in-flight requests per endpoint, as tracked by the in-flight load producer.", compbasemetrics.ALPHA), + }, + []string{"endpoint_name", "namespace", "producer_name"}, +) + +var inflightTokens = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Subsystem: eppmetrics.LLMDRouterEndpointPickerSubsystem, + Name: "inflight_tokens", + Help: metricsutil.HelpMsgWithStability("Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer.", compbasemetrics.ALPHA), + }, + []string{"endpoint_name", "namespace", "producer_name"}, +) + // registerMetrics registers the producer-owned metrics with the plugin's metrics -// recorder. The per-model gauge is shared across producers and the per-endpoint -// collector carries a producer-specific descriptor, so an already-registered -// equivalent collector is a benign no-op. -func registerMetrics(registerer prometheus.Registerer, collector prometheus.Collector) error { +// recorder. All metrics are shared package-level vectors, so a second producer +// instance re-registering them is a benign no-op; the producer_name label keeps +// each instance's per-endpoint series distinct. +func registerMetrics(registerer prometheus.Registerer) error { if registerer == nil { return errors.New("inflight load metrics registerer is required") } - for _, c := range []prometheus.Collector{requestInflight, collector} { - if err := registerer.Register(c); err != nil { + for _, collector := range []prometheus.Collector{requestInflight, inflightRequests, inflightTokens} { + if err := registerer.Register(collector); err != nil { var alreadyRegistered prometheus.AlreadyRegisteredError - if errors.As(err, &alreadyRegistered) { + if errors.As(err, &alreadyRegistered) && alreadyRegistered.ExistingCollector == collector { continue } return fmt.Errorf("register inflight load metric: %w", err) @@ -63,6 +85,16 @@ func registerMetrics(registerer prometheus.Registerer, collector prometheus.Coll return nil } +// splitNamespacedName splits a "namespace/name" key (the string form of a +// k8s types.NamespacedName) into its name and namespace. A key with no +// separator is treated as a bare name with an empty namespace. +func splitNamespacedName(id string) (name, namespace string) { + if i := strings.IndexByte(id, '/'); i >= 0 { + return id[i+1:], id[:i] + } + return id, "" +} + // requestInflightLabels carries the gauge label values captured at admission so // the decrement, which may run on a background reaper goroutine, uses the same // series the increment did. diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go index bb822dd42e..fb6ad8b09b 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go @@ -32,7 +32,6 @@ import ( fwkrh "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/requesthandling" fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" "github.com/llm-d/llm-d-router/pkg/epp/metadata" - "github.com/llm-d/llm-d-router/pkg/epp/metrics/collectors" ) func makeInflightRequest(requestID, incomingModel, fairnessID string, priority int) *fwksched.InferenceRequest { @@ -240,16 +239,16 @@ func TestNewRequestInflightLabels(t *testing.T) { } } -// registerMetrics is idempotent for the shared gauge and registers a distinct per-producer collector, -// so two producers register into one registry and both emit per-endpoint series. +// Two producers share the vec-backed gauges; the producer_name label keeps their per-endpoint +// series distinct. func TestRegisterMetrics_MultiProducer(t *testing.T) { - requestInflight.Reset() - reg := prometheus.NewRegistry() + inflightRequests.Reset() + inflightTokens.Reset() - a := &fakeSnapshotter{requests: map[string]int64{"ns1/ep1": 1}} - b := &fakeSnapshotter{requests: map[string]int64{"ns2/ep2": 2}} - require.NoError(t, registerMetrics(reg, collectors.NewInFlightLoadCollector("a", a))) - require.NoError(t, registerMetrics(reg, collectors.NewInFlightLoadCollector("b", b))) + a := newConcurrencyTracker(inflightRequests, "a") + b := newConcurrencyTracker(inflightRequests, "b") + a.add("ns1/ep1", 1) + b.add("ns2/ep2", 2) expected := ` # HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. @@ -257,17 +256,102 @@ func TestRegisterMetrics_MultiProducer(t *testing.T) { llm_d_epp_inflight_requests{endpoint_name="ep1",namespace="ns1",producer_name="a"} 1 llm_d_epp_inflight_requests{endpoint_name="ep2",namespace="ns2",producer_name="b"} 2 ` - require.NoError(t, promtestutil.GatherAndCompare(reg, strings.NewReader(expected), "llm_d_epp_inflight_requests")) + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(expected), "llm_d_epp_inflight_requests")) +} + +// registerMetrics is idempotent: re-registering the shared package-level vectors is a benign no-op. +func TestRegisterMetrics_Idempotent(t *testing.T) { + reg := prometheus.NewRegistry() + require.NoError(t, registerMetrics(reg)) + require.NoError(t, registerMetrics(reg)) } func TestRegisterMetrics_NilRegisterer(t *testing.T) { - require.Error(t, registerMetrics(nil, nil)) + require.Error(t, registerMetrics(nil)) +} + +// TestInflightGauges_Lifecycle drives a request through admission, completion, and endpoint +// removal, asserting the per-endpoint gauge series mirror the live counters at each step. +func TestInflightGauges_Lifecycle(t *testing.T) { + inflightRequests.Reset() + inflightTokens.Reset() + producer := newTestProducer(t) + ctx := context.Background() + + expect := func(t *testing.T, requests, tokens int64) { + t.Helper() + expectedRequests := fmt.Sprintf(` +# HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_requests gauge +llm_d_epp_inflight_requests{endpoint_name="ep1",namespace="default",producer_name="inflight-load-producer"} %d +`, requests) + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(expectedRequests), "llm_d_epp_inflight_requests")) + expectedTokens := fmt.Sprintf(` +# HELP llm_d_epp_inflight_tokens [ALPHA] Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_tokens gauge +llm_d_epp_inflight_tokens{endpoint_name="ep1",namespace="default",producer_name="inflight-load-producer"} %d +`, tokens) + require.NoError(t, promtestutil.CollectAndCompare(inflightTokens, strings.NewReader(expectedTokens), "llm_d_epp_inflight_tokens")) + } + + // Admission: 4 input + 6 estimated output = 10 tokens, 1 request. + req := makeTokenRequest("req-gauge-lifecycle", 4) + res := makeSchedulingResult("ep1") + producer.PreRequest(ctx, req, res) + expect(t, 1, 10) + + // Completion: series stay present at zero. + req.SchedulingResult = res + producer.ResponseBody(ctx, req, &requestcontrol.Response{EndOfStream: true}, nil) + expect(t, 0, 0) + + // Endpoint removal deletes the series. + producer.DeleteEndpoint(fullEndpointName("ep1")) + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(""), "llm_d_epp_inflight_requests")) + require.NoError(t, promtestutil.CollectAndCompare(inflightTokens, strings.NewReader(""), "llm_d_epp_inflight_tokens")) } -type fakeSnapshotter struct { - requests map[string]int64 - tokens map[string]int64 +// TestInflightGauges_OrphanRelease covers releases that land on an orphaned counter after an +// endpoint flap: publishing after the orphan decrement leaves the live series unchanged, and +// publishing a fully deleted endpoint does not resurrect its series. +func TestInflightGauges_OrphanRelease(t *testing.T) { + inflightRequests.Reset() + + tracker := newConcurrencyTracker(inflightRequests, "p") + counter := tracker.add("ns/ep", 1) + + tracker.delete("ns/ep") + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(""), "llm_d_epp_inflight_requests")) + + tracker.add("ns/ep", 5) + + // A release routed to the orphaned counter must not move the live series. + decrementClamped(counter, 1) + tracker.publish("ns/ep") + expected := ` +# HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. +# TYPE llm_d_epp_inflight_requests gauge +llm_d_epp_inflight_requests{endpoint_name="ep",namespace="ns",producer_name="p"} 5 +` + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(expected), "llm_d_epp_inflight_requests")) + + tracker.delete("ns/ep") + tracker.publish("ns/ep") + require.NoError(t, promtestutil.CollectAndCompare(inflightRequests, strings.NewReader(""), "llm_d_epp_inflight_requests")) } -func (f *fakeSnapshotter) InFlightRequestsSnapshot() map[string]int64 { return f.requests } -func (f *fakeSnapshotter) InFlightTokensSnapshot() map[string]int64 { return f.tokens } +func TestSplitNamespacedName(t *testing.T) { + cases := []struct { + id, wantName, wantNS string + }{ + {"ns/ep", "ep", "ns"}, + {"bare", "bare", ""}, + {"ns/ep/extra", "ep/extra", "ns"}, + } + for _, c := range cases { + name, ns := splitNamespacedName(c.id) + if name != c.wantName || ns != c.wantNS { + t.Errorf("splitNamespacedName(%q) = (%q,%q), want (%q,%q)", c.id, name, ns, c.wantName, c.wantNS) + } + } +} diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go index 9ddcdbbdd6..6f8a394da4 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer.go @@ -25,6 +25,7 @@ import ( "sync" "sync/atomic" + "github.com/prometheus/client_golang/prometheus" "sigs.k8s.io/controller-runtime/pkg/log" logutil "github.com/llm-d/llm-d-router/pkg/common/observability/logging" @@ -37,7 +38,6 @@ import ( sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications" inflightloadconstants "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/constants" tokenproducer "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/tokenizer" - "github.com/llm-d/llm-d-router/pkg/epp/metrics/collectors" ) const ( @@ -103,8 +103,8 @@ func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkp producer := &InFlightLoadProducer{ typedName: fwkplugin.TypedName{Type: InFlightLoadProducerType, Name: name}, - requestTracker: newConcurrencyTracker(), - tokenTracker: newConcurrencyTracker(), + requestTracker: newConcurrencyTracker(inflightRequests, name), + tokenTracker: newConcurrencyTracker(inflightTokens, name), tokenEstimator: NewSimpleTokenEstimatorWithConfig(outputRatio, cfg.MaxEstimatedOutputTokens), addEstimatedOutputTokens: cfg.AddEstimatedOutputTokens, dk: attrconcurrency.InFlightLoadDataKey.WithNonEmptyProducerName(name), @@ -113,9 +113,7 @@ func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkp PluginState: fwkplugin.NewPluginState(ctx), } - // Own the in-flight metrics: the per-model gauge and the per-endpoint - // collector are registered through the plugin's metrics recorder. - if err := registerMetrics(handle.Metrics(), collectors.NewInFlightLoadCollector(name, producer)); err != nil { + if err := registerMetrics(handle.Metrics()); err != nil { return nil, err } @@ -161,6 +159,11 @@ type addedTokensEntry struct { tokenCounter *atomic.Int64 requestCounter *atomic.Int64 requests atomic.Int32 + // endpointID plus the tracker back-references let OnEvicted republish the + // live gauge series after decrementing the captured counter instances. + endpointID string + tokenTracker *concurrencyTracker + requestTracker *concurrencyTracker } var _ fwkplugin.EvictableStateData = (*addedTokensEntry)(nil) @@ -176,6 +179,9 @@ func (e *addedTokensEntry) Clone() fwkplugin.StateData { clone := &addedTokensEntry{ tokenCounter: e.tokenCounter, requestCounter: e.requestCounter, + endpointID: e.endpointID, + tokenTracker: e.tokenTracker, + requestTracker: e.requestTracker, } clone.tokens.Store(e.tokens.Load()) clone.requests.Store(e.requests.Load()) @@ -185,9 +191,11 @@ func (e *addedTokensEntry) Clone() fwkplugin.StateData { func (e *addedTokensEntry) OnEvicted(_ string, _ fwkplugin.StateKey) { if t := e.tokens.Swap(0); t != 0 { decrementClamped(e.tokenCounter, t) + e.tokenTracker.publish(e.endpointID) } if e.requests.Swap(0) != 0 { decrementClamped(e.requestCounter, 1) + e.requestTracker.publish(e.endpointID) } } @@ -421,8 +429,11 @@ func (p *InFlightLoadProducer) PreRequest(ctx context.Context, request *fwksched tokenCounter := p.tokenTracker.add(eid, tokens) entry := &addedTokensEntry{ + endpointID: eid, tokenCounter: tokenCounter, requestCounter: requestCounter, + tokenTracker: p.tokenTracker, + requestTracker: p.requestTracker, } entry.tokens.Store(tokens) entry.requests.Store(1) @@ -542,6 +553,7 @@ func (p *InFlightLoadProducer) releaseTokensEarly(endpoint fwksched.Endpoint, re if entry, err := fwkplugin.ReadPluginStateKey[*addedTokensEntry](p.PluginState, request.RequestID, key); err == nil { if t := entry.tokens.Swap(0); t != 0 { decrementClamped(entry.tokenCounter, t) + p.tokenTracker.publish(eid) } } } @@ -640,34 +652,40 @@ func (p *InFlightLoadProducer) GetRequests(eid string) int64 { return p.requestTracker.get(eid) } -// InFlightRequestsSnapshot returns a copy of the per-endpoint in-flight request -// counts, keyed by the endpoint's "namespace/name". For Prometheus collection. -func (p *InFlightLoadProducer) InFlightRequestsSnapshot() map[string]int64 { - if p.requestTracker == nil { - return nil - } - return p.requestTracker.snapshot() +// concurrencyTracker manages thread-safe counters for inflight requests and +// mirrors each endpoint's live counter value into its gauge series. +type concurrencyTracker struct { + mu sync.RWMutex + counts map[string]*atomic.Int64 + gauge *prometheus.GaugeVec + producerName string } -// InFlightTokensSnapshot returns a copy of the per-endpoint in-flight token -// counts (uncached prompt tokens, optionally plus estimated output), keyed by -// the endpoint's "namespace/name". For Prometheus collection. -func (p *InFlightLoadProducer) InFlightTokensSnapshot() map[string]int64 { - if p.tokenTracker == nil { - return nil +func newConcurrencyTracker(gauge *prometheus.GaugeVec, producerName string) *concurrencyTracker { + return &concurrencyTracker{ + counts: make(map[string]*atomic.Int64), + gauge: gauge, + producerName: producerName, } - return p.tokenTracker.snapshot() } -// concurrencyTracker manages thread-safe counters for inflight requests. -type concurrencyTracker struct { - mu sync.RWMutex - counts map[string]*atomic.Int64 +// setGaugeLocked writes the endpoint's gauge series. Callers must hold t.mu. +func (t *concurrencyTracker) setGaugeLocked(endpointID string, value int64) { + name, namespace := splitNamespacedName(endpointID) + t.gauge.WithLabelValues(name, namespace, t.producerName).Set(float64(value)) } -func newConcurrencyTracker() *concurrencyTracker { - return &concurrencyTracker{ - counts: make(map[string]*atomic.Int64), +// publish syncs the endpoint's gauge series to the live counter value. The +// write lock totally orders publishes, so the last one observes all prior +// counter updates and the gauge converges; a decrement on an orphaned counter +// (endpoint flap) republishes the live counter unchanged. A missing entry +// means the endpoint was deleted: skip the write so a late release cannot +// resurrect the deleted series. +func (t *concurrencyTracker) publish(endpointID string) { + t.mu.Lock() + defer t.mu.Unlock() + if counter, ok := t.counts[endpointID]; ok { + t.setGaugeLocked(endpointID, counter.Load()) } } @@ -697,10 +715,11 @@ func (t *concurrencyTracker) inc(endpointID string) *atomic.Int64 { return t.add(endpointID, 1) } -// add applies delta to the endpoint's counter, creating it if absent, and returns the exact -// *atomic.Int64 instance that was mutated. Callers retain the returned pointer so the matching -// decrement always lands on this same instance, even if the endpoint is later deleted (flap) and a -// new counter is created under the same ID. See addedTokensEntry. +// add applies delta to the endpoint's counter, creating it if absent, publishes the resulting +// value to the gauge series, and returns the exact *atomic.Int64 instance that was mutated. +// Callers retain the returned pointer so the matching decrement always lands on this same +// instance, even if the endpoint is later deleted (flap) and a new counter is created under the +// same ID. See addedTokensEntry. func (t *concurrencyTracker) add(endpointID string, delta int64) *atomic.Int64 { t.mu.RLock() counter, exists := t.counts[endpointID] @@ -708,6 +727,7 @@ func (t *concurrencyTracker) add(endpointID string, delta int64) *atomic.Int64 { if exists { counter.Add(delta) + t.publish(endpointID) return counter } @@ -716,12 +736,14 @@ func (t *concurrencyTracker) add(endpointID string, delta int64) *atomic.Int64 { if counter, exists = t.counts[endpointID]; exists { counter.Add(delta) + t.setGaugeLocked(endpointID, counter.Load()) return counter } counter = &atomic.Int64{} counter.Store(delta) t.counts[endpointID] = counter + t.setGaugeLocked(endpointID, counter.Load()) return counter } @@ -755,4 +777,6 @@ func (t *concurrencyTracker) delete(endpointID string) { t.mu.Lock() defer t.mu.Unlock() delete(t.counts, endpointID) + name, namespace := splitNamespacedName(endpointID) + t.gauge.DeleteLabelValues(name, namespace, t.producerName) } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer_test.go index ff16c93837..fffedb8412 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer_test.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/producer_test.go @@ -262,8 +262,8 @@ func TestInFlightLoadProducer_DumpState(t *testing.T) { t.Parallel() producer := &InFlightLoadProducer{ - requestTracker: newConcurrencyTracker(), - tokenTracker: newConcurrencyTracker(), + requestTracker: newConcurrencyTracker(inflightRequests, "test"), + tokenTracker: newConcurrencyTracker(inflightTokens, "test"), } podA := fullEndpointName("pod-a") podB := fullEndpointName("pod-b") @@ -291,8 +291,8 @@ func TestInFlightLoadProducer_DumpStateCapsEndpoints(t *testing.T) { t.Parallel() producer := &InFlightLoadProducer{ - requestTracker: newConcurrencyTracker(), - tokenTracker: newConcurrencyTracker(), + requestTracker: newConcurrencyTracker(inflightRequests, "test"), + tokenTracker: newConcurrencyTracker(inflightTokens, "test"), } for i := range maxDebugDumpEndpoints + 5 { endpointID := fullEndpointName(fmt.Sprintf("pod-%03d", i)) diff --git a/pkg/epp/metrics/collectors/inflight_load.go b/pkg/epp/metrics/collectors/inflight_load.go deleted file mode 100644 index d30c69afe1..0000000000 --- a/pkg/epp/metrics/collectors/inflight_load.go +++ /dev/null @@ -1,94 +0,0 @@ -/* -Copyright 2026 The Kubernetes Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package collectors - -import ( - "strings" - - "github.com/prometheus/client_golang/prometheus" - compbasemetrics "k8s.io/component-base/metrics" - - metricsutil "github.com/llm-d/llm-d-router/pkg/common/observability/metrics" -) - -// InFlightLoadSnapshotter is the read side of an in-flight load producer the -// collector needs: per-endpoint in-flight request and token counts keyed by the -// endpoint's "namespace/name". -type InFlightLoadSnapshotter interface { - InFlightRequestsSnapshot() map[string]int64 - InFlightTokensSnapshot() map[string]int64 -} - -type inFlightLoadCollector struct { - requestsDesc *prometheus.Desc - tokensDesc *prometheus.Desc - snapshotter InFlightLoadSnapshotter -} - -var _ prometheus.Collector = &inFlightLoadCollector{} - -// NewInFlightLoadCollector returns a prometheus.Collector that emits per-endpoint -// in-flight request and token gauges from the given producer's live snapshot. -// producerName is a constant label baked into the descriptors, so each configured -// producer owns a distinct descriptor and multiple collectors register without -// collision. -func NewInFlightLoadCollector(producerName string, s InFlightLoadSnapshotter) prometheus.Collector { - constLabels := prometheus.Labels{"producer_name": producerName} - return &inFlightLoadCollector{ - requestsDesc: prometheus.NewDesc( - "llm_d_epp_inflight_requests", - metricsutil.HelpMsgWithStability("Current number of in-flight requests per endpoint, as tracked by the in-flight load producer.", compbasemetrics.ALPHA), - []string{"endpoint_name", "namespace"}, constLabels, - ), - tokensDesc: prometheus.NewDesc( - "llm_d_epp_inflight_tokens", - metricsutil.HelpMsgWithStability("Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer.", compbasemetrics.ALPHA), - []string{"endpoint_name", "namespace"}, constLabels, - ), - snapshotter: s, - } -} - -func (c *inFlightLoadCollector) Describe(ch chan<- *prometheus.Desc) { - ch <- c.requestsDesc - ch <- c.tokensDesc -} - -func (c *inFlightLoadCollector) Collect(ch chan<- prometheus.Metric) { - if c.snapshotter == nil { - return - } - c.emit(ch, c.requestsDesc, c.snapshotter.InFlightRequestsSnapshot()) - c.emit(ch, c.tokensDesc, c.snapshotter.InFlightTokensSnapshot()) -} - -func (c *inFlightLoadCollector) emit(ch chan<- prometheus.Metric, desc *prometheus.Desc, counts map[string]int64) { - for endpointID, v := range counts { - name, namespace := splitNamespacedName(endpointID) - ch <- prometheus.MustNewConstMetric(desc, prometheus.GaugeValue, float64(v), name, namespace) - } -} - -// splitNamespacedName splits a "namespace/name" key (the string form of a -// k8s types.NamespacedName) into its name and namespace. A key with no -// separator is treated as a bare name with an empty namespace. -func splitNamespacedName(id string) (name, namespace string) { - if i := strings.IndexByte(id, '/'); i >= 0 { - return id[i+1:], id[:i] - } - return id, "" -} diff --git a/pkg/epp/metrics/collectors/inflight_load_test.go b/pkg/epp/metrics/collectors/inflight_load_test.go deleted file mode 100644 index 684df828d2..0000000000 --- a/pkg/epp/metrics/collectors/inflight_load_test.go +++ /dev/null @@ -1,84 +0,0 @@ -/* -Copyright 2026 The Kubernetes Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package collectors - -import ( - "strings" - "testing" - - promtestutil "github.com/prometheus/client_golang/prometheus/testutil" -) - -type fakeInFlightLoad struct { - requests map[string]int64 - tokens map[string]int64 -} - -func (f *fakeInFlightLoad) InFlightRequestsSnapshot() map[string]int64 { return f.requests } -func (f *fakeInFlightLoad) InFlightTokensSnapshot() map[string]int64 { return f.tokens } - -func TestInFlightLoadCollectorEmpty(t *testing.T) { - collector := NewInFlightLoadCollector("default", &fakeInFlightLoad{}) - if err := promtestutil.CollectAndCompare(collector, strings.NewReader("")); err != nil { - t.Fatal(err) - } -} - -func TestInFlightLoadCollectorPerEndpoint(t *testing.T) { - collector := NewInFlightLoadCollector("default", &fakeInFlightLoad{ - requests: map[string]int64{ - "ns1/ep1": 3, - "ns2/ep2": 5, - }, - tokens: map[string]int64{ - "ns1/ep1": 30, - "ns2/ep2": 50, - }, - }) - - expected := ` -# HELP llm_d_epp_inflight_requests [ALPHA] Current number of in-flight requests per endpoint, as tracked by the in-flight load producer. -# TYPE llm_d_epp_inflight_requests gauge -llm_d_epp_inflight_requests{endpoint_name="ep1",namespace="ns1",producer_name="default"} 3 -llm_d_epp_inflight_requests{endpoint_name="ep2",namespace="ns2",producer_name="default"} 5 -# HELP llm_d_epp_inflight_tokens [ALPHA] Current number of in-flight tokens per endpoint (uncached prompt tokens, optionally plus estimated output), as tracked by the in-flight load producer. -# TYPE llm_d_epp_inflight_tokens gauge -llm_d_epp_inflight_tokens{endpoint_name="ep1",namespace="ns1",producer_name="default"} 30 -llm_d_epp_inflight_tokens{endpoint_name="ep2",namespace="ns2",producer_name="default"} 50 -` - - if err := promtestutil.CollectAndCompare(collector, strings.NewReader(expected), - "llm_d_epp_inflight_requests", "llm_d_epp_inflight_tokens"); err != nil { - t.Fatal(err) - } -} - -func TestSplitNamespacedName(t *testing.T) { - cases := []struct { - id, wantName, wantNS string - }{ - {"ns/ep", "ep", "ns"}, - {"bare", "bare", ""}, - {"ns/ep/extra", "ep/extra", "ns"}, - } - for _, c := range cases { - name, ns := splitNamespacedName(c.id) - if name != c.wantName || ns != c.wantNS { - t.Errorf("splitNamespacedName(%q) = (%q,%q), want (%q,%q)", c.id, name, ns, c.wantName, c.wantNS) - } - } -}