diff --git a/docs/metrics.md b/docs/metrics.md index ba7106d732..f3316fe9c3 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -163,3 +163,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 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` + +* **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/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..54b3edae99 --- /dev/null +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics.go @@ -0,0 +1,133 @@ +/* +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" + "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" + 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"}, +) + +// 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. 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 _, collector := range []prometheus.Collector{requestInflight, inflightRequests, inflightTokens} { + if err := registerer.Register(collector); err != nil { + var alreadyRegistered prometheus.AlreadyRegisteredError + if errors.As(err, &alreadyRegistered) && alreadyRegistered.ExistingCollector == collector { + continue + } + return fmt.Errorf("register inflight load metric: %w", err) + } + } + 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. +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..fb6ad8b09b --- /dev/null +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/inflightload/metrics_test.go @@ -0,0 +1,357 @@ +/* +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" +) + +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)) + }) + } +} + +// Two producers share the vec-backed gauges; the producer_name label keeps their per-endpoint +// series distinct. +func TestRegisterMetrics_MultiProducer(t *testing.T) { + inflightRequests.Reset() + inflightTokens.Reset() + + 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. +# 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.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)) +} + +// 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")) +} + +// 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 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 9b5b27dce5..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" @@ -43,6 +44,9 @@ 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. @@ -97,17 +101,23 @@ func InFlightLoadProducerFactory(name string, decoder *json.Decoder, handle fwkp return nil, fmt.Errorf("maxEstimatedOutputTokens must be non-negative, got %v", *cfg.MaxEstimatedOutputTokens) } - return &InFlightLoadProducer{ + 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), prefixMatchInfoDK: attrprefix.PrefixCacheMatchInfoDataKey.WithNonEmptyProducerName(cfg.PrefixMatchInfoProducerName), uncachedRequestTokensDk: attrconcurrency.UncachedRequestTokensDataKey.WithNonEmptyProducerName(name), PluginState: fwkplugin.NewPluginState(ctx), - }, nil + } + + if err := registerMetrics(handle.Metrics()); err != nil { + return nil, err + } + + return producer, nil } var ( @@ -149,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) @@ -164,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()) @@ -173,9 +191,36 @@ 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) + } +} + +// 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) } } @@ -354,6 +399,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 { @@ -377,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) @@ -498,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) } } } @@ -596,15 +652,40 @@ func (p *InFlightLoadProducer) GetRequests(eid string) int64 { return p.requestTracker.get(eid) } -// concurrencyTracker manages thread-safe counters for inflight requests. +// 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 + mu sync.RWMutex + counts map[string]*atomic.Int64 + gauge *prometheus.GaugeVec + producerName string } -func newConcurrencyTracker() *concurrencyTracker { +func newConcurrencyTracker(gauge *prometheus.GaugeVec, producerName string) *concurrencyTracker { return &concurrencyTracker{ - counts: make(map[string]*atomic.Int64), + counts: make(map[string]*atomic.Int64), + gauge: gauge, + producerName: producerName, + } +} + +// 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)) +} + +// 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()) } } @@ -634,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] @@ -645,6 +727,7 @@ func (t *concurrencyTracker) add(endpointID string, delta int64) *atomic.Int64 { if exists { counter.Add(delta) + t.publish(endpointID) return counter } @@ -653,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 } @@ -692,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/requestcontrol/director.go b/pkg/epp/requestcontrol/director.go index 4e33e980f0..a2e18b9d6a 100644 --- a/pkg/epp/requestcontrol/director.go +++ b/pkg/epp/requestcontrol/director.go @@ -267,6 +267,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,