diff --git a/cmd/epp/runner/runner.go b/cmd/epp/runner/runner.go index c2fb54efbe..6d93f5e220 100644 --- a/cmd/epp/runner/runner.go +++ b/cmd/epp/runner/runner.go @@ -60,13 +60,16 @@ import ( fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" attrconcurrency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/concurrency" + attrgpu "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/gpu" attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency" attrmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/models" attrprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/prefix" attrsession "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/session" discoveryfile "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/discovery/file" + extdcgm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/dcgm" extractormetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/metrics" extmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/extractor/models" + srcdcgm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/dcgm" sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics" srcmodels "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/models" sourcenotifications "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/notifications" @@ -564,6 +567,10 @@ func (r *Runner) registerInTreePlugins() { fwkplugin.Register(srcmodels.ModelsDataSourceType, srcmodels.ModelDataSourceFactory) fwkplugin.Register(attrmodels.ModelsExtractorType, extmodels.ModelServerExtractorFactory) + // data layer DCGM source/extractor + fwkplugin.Register(srcdcgm.DCGMDataSourceType, srcdcgm.DCGMDataSourceFactory) + fwkplugin.Register(attrgpu.DCGMExtractorType, extdcgm.DCGMExtractorFactory) + fwkplugin.Register(prefix.PrefixCacheScorerPluginType, prefix.PrefixCachePluginFactory) fwkplugin.Register(maxscore.MaxScorePickerType, maxscore.MaxScorePickerFactory) fwkplugin.Register(random.RandomPickerType, random.RandomPickerFactory) diff --git a/pkg/epp/datastore/datastore.go b/pkg/epp/datastore/datastore.go index b7260474a4..8fce06d882 100644 --- a/pkg/epp/datastore/datastore.go +++ b/pkg/epp/datastore/datastore.go @@ -315,6 +315,7 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P NamespacedName: createEndpointNamespacedName(pod, idx), PodName: pod.Name, Address: pod.Status.PodIP, + NodeAddress: pod.Status.HostIP, Port: strconv.Itoa(port), MetricsHost: net.JoinHostPort(pod.Status.PodIP, strconv.Itoa(port)), Labels: labels, diff --git a/pkg/epp/datastore/datastore_test.go b/pkg/epp/datastore/datastore_test.go index 7492ff31a1..7b5bd05a90 100644 --- a/pkg/epp/datastore/datastore_test.go +++ b/pkg/epp/datastore/datastore_test.go @@ -305,6 +305,10 @@ var ( ObjectMeta: metav1.ObjectMeta{ Name: "pod1", }, + Status: corev1.PodStatus{ + PodIP: "10.0.0.1", + HostIP: "192.168.1.10", + }, } pod1Metrics = &fwkdl.Metrics{ WaitingQueueSize: 0, @@ -320,6 +324,10 @@ var ( ObjectMeta: metav1.ObjectMeta{ Name: "pod2", }, + Status: corev1.PodStatus{ + PodIP: "10.0.0.2", + HostIP: "192.168.1.10", + }, } pod2Metrics = &fwkdl.Metrics{ WaitingQueueSize: 1, @@ -506,7 +514,10 @@ func TestPods(t *testing.T) { for idx, pm := range podList { gotPods[idx] = &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{Name: pm.GetMetadata().PodName, Namespace: pm.GetMetadata().NamespacedName.Namespace}, - Status: corev1.PodStatus{PodIP: pm.GetMetadata().GetIPAddress()}, + Status: corev1.PodStatus{ + PodIP: pm.GetMetadata().GetIPAddress(), + HostIP: pm.GetMetadata().GetNodeAddress(), + }, } } if !cmp.Equal(gotPods, test.wantPods, cmpopts.SortSlices(func(a, b *corev1.Pod) bool { return a.Name < b.Name })) { @@ -649,6 +660,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolTargetPort, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolTargetPort), Labels: map[string]string{}, @@ -671,6 +683,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort0, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0), Labels: map[string]string{}, @@ -683,6 +696,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort1, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1), Labels: map[string]string{}, @@ -706,6 +720,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort0, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0), Labels: map[string]string{}, @@ -718,6 +733,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort1, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1), Labels: map[string]string{}, @@ -731,8 +747,9 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod2.Name, Address: pod2.Status.PodIP, + NodeAddress: pod2.Status.HostIP, Port: inferencePoolMultiTargetPort0, - MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0), + MetricsHost: net.JoinHostPort(pod2.Status.PodIP, inferencePoolMultiTargetPort0), Labels: map[string]string{}, }, { @@ -743,8 +760,9 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod2.Name, Address: pod2.Status.PodIP, + NodeAddress: pod2.Status.HostIP, Port: inferencePoolMultiTargetPort1, - MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1), + MetricsHost: net.JoinHostPort(pod2.Status.PodIP, inferencePoolMultiTargetPort1), Labels: map[string]string{}, RankIndex: 1, }, @@ -766,6 +784,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort0, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0), Labels: map[string]string{}, @@ -778,6 +797,7 @@ func TestEndpointMetadata(t *testing.T) { PodName: pod1.Name, Address: pod1.Status.PodIP, + NodeAddress: pod1.Status.HostIP, Port: inferencePoolMultiTargetPort1, MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1), Labels: map[string]string{}, diff --git a/pkg/epp/framework/interface/datalayer/endpoint_metadata.go b/pkg/epp/framework/interface/datalayer/endpoint_metadata.go index 6882c2bc9c..bd53aad2ca 100644 --- a/pkg/epp/framework/interface/datalayer/endpoint_metadata.go +++ b/pkg/epp/framework/interface/datalayer/endpoint_metadata.go @@ -28,9 +28,12 @@ type EndpointMetadata struct { NamespacedName types.NamespacedName PodName string Address string - Port string - MetricsHost string - Labels map[string]string + // NodeAddress is the node IP hosting this pod (pod.Status.HostIP). + // Empty for non-Kubernetes discovery sources (e.g. file discovery). + NodeAddress string + Port string + MetricsHost string + Labels map[string]string // RankIndex is this endpoint's position in the pool's TargetPorts, // identifying the pod-local rank in multi-port deployments. RankIndex int @@ -59,6 +62,7 @@ func (epm *EndpointMetadata) Clone() *EndpointMetadata { }, PodName: epm.PodName, Address: epm.Address, + NodeAddress: epm.NodeAddress, Port: epm.Port, MetricsHost: epm.MetricsHost, Labels: clonedLabels, @@ -75,6 +79,7 @@ func (epm *EndpointMetadata) Equal(other *EndpointMetadata) bool { return epm.NamespacedName == other.NamespacedName && epm.PodName == other.PodName && epm.Address == other.Address && + epm.NodeAddress == other.NodeAddress && epm.Port == other.Port && epm.MetricsHost == other.MetricsHost && epm.RankIndex == other.RankIndex && @@ -100,6 +105,14 @@ func (epm *EndpointMetadata) GetIPAddress() string { return epm.Address } +// GetNodeAddress returns the IP of the node hosting this endpoint. +func (epm *EndpointMetadata) GetNodeAddress() string { + if epm == nil { + return "" + } + return epm.NodeAddress +} + // GetPort returns the Endpoint's inference port. func (epm *EndpointMetadata) GetPort() string { return epm.Port diff --git a/pkg/epp/framework/plugins/datalayer/attribute/gpu/README.md b/pkg/epp/framework/plugins/datalayer/attribute/gpu/README.md new file mode 100644 index 0000000000..341191440a --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/attribute/gpu/README.md @@ -0,0 +1,15 @@ +# GPU Attributes + +This package defines the data structures for GPU hardware metrics collected from NVIDIA DCGM Exporter. + +## `GPUUtilization` + +A normalized GPU compute utilization value in [0.0, 1.0], derived from `DCGM_FI_DEV_GPU_UTIL` (which reports 0-100). For multi-GPU pods the extractor aggregates across visible devices using `max`. + +- **Key**: `GPUUtilizationDataKey` (`GPUUtilization/dcgm-extractor`) + +## Producers + +The following plugins produce this attribute: + +- **`dcgm-extractor`** (Data Layer): Extracts GPU utilization from the DCGM Exporter Prometheus endpoint. diff --git a/pkg/epp/framework/plugins/datalayer/attribute/gpu/data_types.go b/pkg/epp/framework/plugins/datalayer/attribute/gpu/data_types.go new file mode 100644 index 0000000000..e71af7b69a --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/attribute/gpu/data_types.go @@ -0,0 +1,45 @@ +/* +Copyright 2026 The llm-d 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 gpu + +import ( + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" +) + +const ( + // DCGMExtractorType is the plugin type for the DCGM metrics extractor. + DCGMExtractorType = "dcgm-extractor" +) + +// GPUUtilizationDataKey identifies per-endpoint GPU utilization published +// by the DCGM extractor and consumed by GPU-aware filters and scorers. +var GPUUtilizationDataKey = plugin.NewDataKey("GPUUtilization", DCGMExtractorType) + +// GPUUtilization is a normalized GPU compute utilization value in [0.0, 1.0], +// derived from DCGM_FI_DEV_GPU_UTIL (which reports 0-100). +// For multi-GPU pods the extractor aggregates across visible devices. +type GPUUtilization float64 + +func (v GPUUtilization) Clone() fwkdl.Cloneable { + return v +} + +// ReadGPUUtilization retrieves GPU utilization from an endpoint's AttributeMap. +func ReadGPUUtilization(attrs fwkdl.AttributeMap, key string) (GPUUtilization, bool) { + return fwkdl.ReadAttribute[GPUUtilization](attrs, key) +} diff --git a/pkg/epp/framework/plugins/datalayer/extractor/dcgm/README.md b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/README.md new file mode 100644 index 0000000000..1f109a215d --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/README.md @@ -0,0 +1,33 @@ +# DCGM Extractor + +**Type:** `dcgm-extractor` + +The DCGM Extractor converts the Prometheus metrics response from a +`dcgm-data-source` into a per-endpoint GPU utilization attribute consumed by +GPU-aware filters and scorers. + +## What it does + +1. Receives the parsed Prometheus metric families forwarded by `dcgm-data-source`. +2. Looks up the `DCGM_FI_DEV_GPU_UTIL` metric family. +3. Keeps samples that belong to this endpoint: + - If a sample has a `pod` label and the endpoint has a `PodName`, only + matching samples are kept (DaemonSet multi-pod payloads). + - If the sample has no `pod` label, it is kept (sidecar / unlabeled payloads). +4. Aggregates matching samples using `max` (highest-utilized GPU for the pod). +5. Normalizes the value from 0-100 to [0.0, 1.0]. +6. Stores the result as a `GPUUtilization` attribute on the corresponding endpoint. + +## Attributes produced + +- `GPUUtilization` stored at attribute key `GPUUtilizationDataKey` + (`GPUUtilization/dcgm-extractor`) on each endpoint. + +```go +key := attrgpu.GPUUtilizationDataKey.String() +util, ok := attrgpu.ReadGPUUtilization(endpoint.GetAttributes(), key) +``` + +## Configuration + +No configuration parameters. diff --git a/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor.go b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor.go new file mode 100644 index 0000000000..b9fd463526 --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor.go @@ -0,0 +1,143 @@ +/* +Copyright 2026 The llm-d 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 dcgm + +import ( + "context" + "encoding/json" + "fmt" + + dto "github.com/prometheus/client_model/go" + + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" + attrgpu "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/gpu" + sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics" +) + +const ( + gpuUtilMetricName = "DCGM_FI_DEV_GPU_UTIL" + podLabelName = "pod" +) + +var _ fwkplugin.ProducerPlugin = &Extractor{} + +var _ fwkdl.PollingExtractor[sourcemetrics.PrometheusMetricMap] = &Extractor{} + +// Extractor reads DCGM Exporter metrics and stores aggregated GPU +// utilization in the endpoint's AttributeMap. +type Extractor struct { + typedName fwkplugin.TypedName + dk fwkplugin.DataKey +} + +// NewDCGMExtractor returns a new DCGM metrics extractor. +func NewDCGMExtractor() *Extractor { + return &Extractor{ + typedName: fwkplugin.TypedName{ + Type: attrgpu.DCGMExtractorType, + Name: attrgpu.DCGMExtractorType, + }, + dk: attrgpu.GPUUtilizationDataKey, + } +} + +func (e *Extractor) TypedName() fwkplugin.TypedName { + return e.typedName +} + +// DCGMExtractorFactory instantiates a dcgm-extractor plugin from configuration. +func DCGMExtractorFactory(name string, _ *json.Decoder, _ fwkplugin.Handle) (fwkplugin.Plugin, error) { + ext := NewDCGMExtractor() + ext.typedName.Name = name + return ext, nil +} + +// Extract parses DCGM_FI_DEV_GPU_UTIL from the Prometheus families, +// keeps samples for this endpoint's pod (when the pod label is present), +// aggregates across matching GPUs (max), normalizes to [0.0, 1.0], and +// stores the result in the endpoint's AttributeMap. +func (e *Extractor) Extract(_ context.Context, in fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]) error { + family, ok := in.Payload[gpuUtilMetricName] + if !ok { + return fmt.Errorf("metric %q not found in DCGM response", gpuUtilMetricName) + } + + podName := "" + if meta := in.Endpoint.GetMetadata(); meta != nil { + podName = meta.PodName + } + + maxUtil := 0.0 + found := false + for _, m := range family.GetMetric() { + if !sampleBelongsToPod(m, podName) { + continue + } + val, err := gaugeValue(m) + if err != nil { + return err + } + if val > maxUtil { + maxUtil = val + } + found = true + } + if !found { + return fmt.Errorf("metric %q present but has no samples for pod %q", gpuUtilMetricName, podName) + } + + normalized := attrgpu.GPUUtilization(maxUtil / 100.0) + in.Endpoint.GetAttributes().Put(e.dk.String(), normalized) + return nil +} + +// Produces declares the data key this extractor publishes. +func (e *Extractor) Produces() map[fwkplugin.DataKey]any { + return map[fwkplugin.DataKey]any{e.dk: attrgpu.GPUUtilization(0)} +} + +// sampleBelongsToPod reports whether m should contribute to this endpoint's +// utilization. When podName is empty, or the sample has no pod label, the +// sample is kept (sidecar / single-tenant payloads). When both are set, only +// matching samples are kept (DaemonSet multi-pod payloads). +func sampleBelongsToPod(m *dto.Metric, podName string) bool { + if podName == "" { + return true + } + samplePod, ok := labelValue(m, podLabelName) + if !ok { + return true + } + return samplePod == podName +} + +func labelValue(m *dto.Metric, name string) (string, bool) { + for _, l := range m.GetLabel() { + if l.GetName() == name { + return l.GetValue(), true + } + } + return "", false +} + +func gaugeValue(m *dto.Metric) (float64, error) { + if g := m.GetGauge(); g != nil { + return g.GetValue(), nil + } + return 0, fmt.Errorf("expected gauge metric for %s", gpuUtilMetricName) +} diff --git a/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor_test.go b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor_test.go new file mode 100644 index 0000000000..c61003e939 --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/extractor/dcgm/extractor_test.go @@ -0,0 +1,253 @@ +/* +Copyright 2026 The llm-d 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 dcgm + +import ( + "context" + "math" + "strings" + "testing" + + "github.com/google/go-cmp/cmp" + dto "github.com/prometheus/client_model/go" + "google.golang.org/protobuf/proto" + + fwkdl "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/datalayer" + attrgpu "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/gpu" + sourcemetrics "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics" +) + +func makeGauge(values ...float64) *dto.MetricFamily { + gaugeType := dto.MetricType_GAUGE + fam := &dto.MetricFamily{ + Name: proto.String(gpuUtilMetricName), + Type: &gaugeType, + } + for _, v := range values { + fam.Metric = append(fam.Metric, &dto.Metric{ + Gauge: &dto.Gauge{Value: &v}, + }) + } + return fam +} + +type labeledSample struct { + pod string + value float64 +} + +func makeLabeledGauge(samples ...labeledSample) *dto.MetricFamily { + gaugeType := dto.MetricType_GAUGE + fam := &dto.MetricFamily{ + Name: proto.String(gpuUtilMetricName), + Type: &gaugeType, + } + for _, s := range samples { + m := &dto.Metric{ + Gauge: &dto.Gauge{Value: &s.value}, + } + if s.pod != "" { + m.Label = []*dto.LabelPair{{ + Name: proto.String(podLabelName), + Value: proto.String(s.pod), + }} + } + fam.Metric = append(fam.Metric, m) + } + return fam +} + +func TestExtractorExtract(t *testing.T) { + ctx := context.Background() + + extPlugin, err := DCGMExtractorFactory("test-extractor", nil, nil) + if err != nil { + t.Fatalf("failed to create extractor: %v", err) + } + extractor := extPlugin.(*Extractor) + + if exType := extPlugin.TypedName().Type; exType == "" { + t.Error("empty extractor type") + } + if exName := extPlugin.TypedName().Name; exName == "" { + t.Error("empty extractor name") + } + + key := attrgpu.GPUUtilizationDataKey.WithNonEmptyProducerName(attrgpu.DCGMExtractorType).String() + gaugeType := dto.MetricType_GAUGE + + tests := []struct { + name string + podName string + data sourcemetrics.PrometheusMetricMap + wantErr bool + errContains string + updated bool + wantUtil float64 + }{ + { + name: "missing metric", + data: sourcemetrics.PrometheusMetricMap{}, + wantErr: true, + errContains: gpuUtilMetricName, + }, + { + name: "empty samples", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: &dto.MetricFamily{ + Name: proto.String(gpuUtilMetricName), + Type: &gaugeType, + Metric: []*dto.Metric{}, + }, + }, + wantErr: true, + errContains: "no samples", + }, + { + name: "single GPU", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeGauge(73), + }, + updated: true, + wantUtil: 0.73, + }, + { + name: "multi-GPU takes max", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeGauge(73, 81, 65), + }, + updated: true, + wantUtil: 0.81, + }, + { + name: "DaemonSet payload filters by pod label", + podName: "vllm-pod-2", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeLabeledGauge( + labeledSample{"vllm-pod-1", 25}, + labeledSample{"vllm-pod-2", 80}, + labeledSample{"vllm-pod-3", 10}, + ), + }, + updated: true, + wantUtil: 0.80, + }, + { + name: "DaemonSet multi-GPU for same pod takes max", + podName: "vllm-pod-1", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeLabeledGauge( + labeledSample{"vllm-pod-1", 40}, + labeledSample{"vllm-pod-1", 70}, + labeledSample{"vllm-pod-2", 99}, + ), + }, + updated: true, + wantUtil: 0.70, + }, + { + name: "no matching pod label returns error", + podName: "missing-pod", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeLabeledGauge( + labeledSample{"vllm-pod-1", 50}, + ), + }, + wantErr: true, + errContains: "missing-pod", + }, + { + name: "unlabeled samples kept when podName set (sidecar compat)", + podName: "vllm-pod-1", + data: sourcemetrics.PrometheusMetricMap{ + gpuUtilMetricName: makeGauge(55), + }, + updated: true, + wantUtil: 0.55, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + defer func() { + if r := recover(); r != nil { + t.Errorf("Extract panicked: %v", r) + } + }() + + meta := &fwkdl.EndpointMetadata{PodName: tt.podName} + ep := fwkdl.NewEndpoint(meta, nil) + attr := ep.GetAttributes() + before, _ := attr.Get(key) + + err := extractor.Extract(ctx, fwkdl.PollInput[sourcemetrics.PrometheusMetricMap]{ + Payload: tt.data, Endpoint: ep, + }) + after, _ := attr.Get(key) + + if tt.wantErr && err == nil { + t.Error("expected error but got nil") + } + if !tt.wantErr && err != nil { + t.Errorf("unexpected error: %v", err) + } + if tt.wantErr && err != nil && tt.errContains != "" { + if !strings.Contains(err.Error(), tt.errContains) { + t.Errorf("error %q does not contain %q", err.Error(), tt.errContains) + } + } + + if tt.updated { + if diff := cmp.Diff(before, after); diff == "" { + t.Error("expected attribute to be updated, but no change detected") + } + val, ok := attrgpu.ReadGPUUtilization(attr, key) + if !ok { + t.Fatal("GPU utilization not found in attributes after extract") + } + if math.Abs(float64(val)-tt.wantUtil) > 0.001 { + t.Errorf("expected utilization %.3f, got %.3f", tt.wantUtil, float64(val)) + } + } else { + if diff := cmp.Diff(before, after); diff != "" { + t.Errorf("expected no attribute update, but got changes:\n%s", diff) + } + } + }) + } +} + +func TestFactory_SetsName(t *testing.T) { + p, err := DCGMExtractorFactory("my-dcgm", nil, nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := p.TypedName().Type; got != attrgpu.DCGMExtractorType { + t.Errorf("Type = %q, want %q", got, attrgpu.DCGMExtractorType) + } + if got := p.TypedName().Name; got != "my-dcgm" { + t.Errorf("Name = %q, want %q", got, "my-dcgm") + } +} + +func TestProduces_DeclaresGPUUtilization(t *testing.T) { + ext := NewDCGMExtractor() + produced := ext.Produces() + if _, ok := produced[attrgpu.GPUUtilizationDataKey]; !ok { + t.Error("Produces must declare GPUUtilizationDataKey") + } +} diff --git a/pkg/epp/framework/plugins/datalayer/source/dcgm/README.md b/pkg/epp/framework/plugins/datalayer/source/dcgm/README.md new file mode 100644 index 0000000000..d1f2dfccae --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/source/dcgm/README.md @@ -0,0 +1,71 @@ +# DCGM Data Source + +**Type:** `dcgm-data-source` + +The DCGM Data Source polls NVIDIA DCGM Exporter for GPU hardware metrics and +passes the response to a paired `dcgm-extractor`. + +## What it does + +1. Iterates over every ready endpoint associated with the `InferencePool`. +2. Issues a `GET ://:/` request to the DCGM Exporter. +3. Parses the Prometheus text response. +4. Returns the parsed metric families to the datalayer runtime, which forwards + them to any extractors wired to this source via `dataLayer: sources:`. + +The scrape host is either the pod IP (sidecar) or the node IP (DaemonSet), +controlled by `useNodeAddress`. In both cases `port` overrides the inference +server port via `HTTPDataSource` options. + +## Inputs consumed + +- Pod list from the `InferencePool` (polled individually on each scheduling cycle). +- When `useNodeAddress` is true, each endpoint's `NodeAddress` (`pod.Status.HostIP`). + +## Configuration + +- `scheme` (string, optional, default: `"http"`): Protocol scheme: `"http"` or `"https"`. +- `path` (string, optional, default: `"/metrics"`): URL path for the DCGM Exporter metrics endpoint. +- `port` (int, optional, default: `9400`): Port where the DCGM Exporter listens. +- `insecureSkipVerify` (bool, optional, default: `true`): Skip TLS certificate verification. +- `useNodeAddress` (bool, optional, default: `false`): When true, scrape + `NodeAddress:port` (DaemonSet). When false, scrape `PodIP:port` (sidecar). + +DaemonSet deployments require `DCGM_EXPORTER_KUBERNETES=true` on the exporter +so metrics include a `pod` label (GPU Operator sets this by default). + +```yaml +# Sidecar (default) +- type: dcgm-data-source + name: my-dcgm-source + parameters: + port: 9400 + +# DaemonSet +- type: dcgm-data-source + name: my-dcgm-source + parameters: + port: 9400 + useNodeAddress: true +``` + +## Complete Configuration Example + +```yaml +apiVersion: llm-d.ai/v1alpha1 +kind: EndpointPickerConfig +plugins: +- type: dcgm-data-source + name: dcgm-source + parameters: + port: 9400 + useNodeAddress: true +- type: dcgm-extractor + name: dcgm-extractor +# ... other plugins (filters, scorers, profile handler, picker) ... +dataLayer: + sources: + - pluginRef: dcgm-source + extractors: + - pluginRef: dcgm-extractor +``` diff --git a/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource.go b/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource.go new file mode 100644 index 0000000000..2197718fc8 --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource.go @@ -0,0 +1,95 @@ +/* +Copyright 2026 The llm-d 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 dcgm + +import ( + "encoding/json" + "fmt" + "io" + + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + + "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" + "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/http" + "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/metrics" +) + +const DCGMDataSourceType = "dcgm-data-source" + +const ( + defaultDCGMScheme = "http" + defaultDCGMPath = "/metrics" + defaultDCGMPort = 9400 + defaultDCGMInsecureSkipVerify = true +) + +type dcgmDatasourceParams struct { + Scheme string `json:"scheme"` + Path string `json:"path"` + Port int `json:"port"` + InsecureSkipVerify bool `json:"insecureSkipVerify"` + // UseNodeAddress scrapes the node IP (DaemonSet) instead of the pod IP (sidecar). + UseNodeAddress bool `json:"useNodeAddress"` +} + +// NewHTTPDCGMDataSource constructs a DCGMDataSource with the given parameters. +// Use this function directly in tests to bypass JSON parameter marshaling. +func NewHTTPDCGMDataSource(scheme, path string, port int, name string, opts ...http.Option) (*http.HTTPDataSource[metrics.PrometheusMetricMap], error) { + opts = append([]http.Option{http.WithPortOverride(port)}, opts...) + return http.NewHTTPDataSource(scheme, path, http.TLSOptions{SkipVerify: defaultDCGMInsecureSkipVerify}, + DCGMDataSourceType, name, parsePrometheus, opts...) +} + +// DCGMDataSourceFactory instantiates a dcgm-data-source plugin from configuration. +func DCGMDataSourceFactory(name string, parameters *json.Decoder, _ plugin.Handle) (plugin.Plugin, error) { + cfg := defaultDCGMConfigParams() + if parameters != nil { + if err := parameters.Decode(cfg); err != nil { + return nil, err + } + } + if cfg.Scheme != "http" && cfg.Scheme != "https" { + return nil, fmt.Errorf("unsupported scheme: %s", cfg.Scheme) + } + + opts := []http.Option{http.WithPortOverride(cfg.Port)} + if cfg.UseNodeAddress { + opts = append(opts, http.WithUseNodeAddress()) + } + + ds, err := http.NewHTTPDataSource(cfg.Scheme, cfg.Path, http.TLSOptions{SkipVerify: cfg.InsecureSkipVerify}, + DCGMDataSourceType, name, parsePrometheus, opts...) + if err != nil { + return nil, fmt.Errorf("failed to create DCGM data source: %w", err) + } + return ds, nil +} + +func defaultDCGMConfigParams() *dcgmDatasourceParams { + return &dcgmDatasourceParams{ + Scheme: defaultDCGMScheme, + Path: defaultDCGMPath, + Port: defaultDCGMPort, + InsecureSkipVerify: defaultDCGMInsecureSkipVerify, + } +} + +func parsePrometheus(data io.Reader) (metrics.PrometheusMetricMap, error) { + parser := expfmt.NewTextParser(model.LegacyValidation) + return parser.TextToMetricFamilies(data) +} diff --git a/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource_test.go b/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource_test.go new file mode 100644 index 0000000000..965d3e5b8d --- /dev/null +++ b/pkg/epp/framework/plugins/datalayer/source/dcgm/datasource_test.go @@ -0,0 +1,101 @@ +/* +Copyright 2026 The llm-d 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 dcgm + +import ( + "testing" + + fwkplugin "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/plugin" + "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/source/http" +) + +func TestDCGMDataSourceFactory_Defaults(t *testing.T) { + p, err := DCGMDataSourceFactory("test", fwkplugin.StrictDecoder(nil), nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if p == nil { + t.Fatal("expected non-nil plugin") + } + if got := p.TypedName().Type; got != DCGMDataSourceType { + t.Errorf("Type = %q, want %q", got, DCGMDataSourceType) + } + if got := p.TypedName().Name; got != "test" { + t.Errorf("Name = %q, want %q", got, "test") + } +} + +func TestDCGMDataSourceFactory_CustomParams(t *testing.T) { + raw := []byte(`{"scheme":"https","path":"/gpu","port":9500,"insecureSkipVerify":false,"useNodeAddress":true}`) + p, err := DCGMDataSourceFactory("custom", fwkplugin.StrictDecoder(raw), nil) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if p == nil { + t.Fatal("expected non-nil plugin") + } + if got := p.TypedName().Name; got != "custom" { + t.Errorf("Name = %q, want %q", got, "custom") + } +} + +func TestDCGMDataSourceFactory_InvalidScheme(t *testing.T) { + raw := []byte(`{"scheme":"grpc"}`) + _, err := DCGMDataSourceFactory("test", fwkplugin.StrictDecoder(raw), nil) + if err == nil { + t.Fatal("expected error for unsupported scheme") + } +} + +func TestDCGMDataSourceFactory_InvalidJSON(t *testing.T) { + raw := []byte(`{invalid`) + _, err := DCGMDataSourceFactory("test", fwkplugin.StrictDecoder(raw), nil) + if err == nil { + t.Fatal("expected error for malformed JSON") + } +} + +func TestDCGMDataSourceFactory_UnknownField(t *testing.T) { + raw := []byte(`{"scheme":"http","bogus":"value"}`) + _, err := DCGMDataSourceFactory("test", fwkplugin.StrictDecoder(raw), nil) + if err == nil { + t.Fatal("expected error for unknown field (strict decoding)") + } +} + +func TestNewHTTPDCGMDataSource(t *testing.T) { + ds, err := NewHTTPDCGMDataSource("http", "/metrics", 9400, "direct") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got := ds.TypedName().Type; got != DCGMDataSourceType { + t.Errorf("Type = %q, want %q", got, DCGMDataSourceType) + } + if got := ds.TypedName().Name; got != "direct" { + t.Errorf("Name = %q, want %q", got, "direct") + } +} + +func TestNewHTTPDCGMDataSource_WithUseNodeAddress(t *testing.T) { + ds, err := NewHTTPDCGMDataSource("http", "/metrics", 9400, "node", http.WithUseNodeAddress()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ds == nil { + t.Fatal("expected non-nil data source") + } +} diff --git a/pkg/epp/framework/plugins/datalayer/source/http/client.go b/pkg/epp/framework/plugins/datalayer/source/http/client.go index 8acd390d1d..82c4187c1d 100644 --- a/pkg/epp/framework/plugins/datalayer/source/http/client.go +++ b/pkg/epp/framework/plugins/datalayer/source/http/client.go @@ -35,6 +35,7 @@ type Client interface { // Addressable supports getting an IP address and a namespaced name. type Addressable interface { GetIPAddress() string + GetNodeAddress() string GetPort() string GetMetricsHost() string GetNamespacedName() types.NamespacedName diff --git a/pkg/epp/framework/plugins/datalayer/source/http/datasource.go b/pkg/epp/framework/plugins/datalayer/source/http/datasource.go index 70a47ea72b..8f1810771d 100644 --- a/pkg/epp/framework/plugins/datalayer/source/http/datasource.go +++ b/pkg/epp/framework/plugins/datalayer/source/http/datasource.go @@ -23,12 +23,14 @@ import ( "errors" "fmt" "io" + "net" "net/http" "net/url" "os" "reflect" "runtime/debug" "slices" + "strconv" "sync" "time" @@ -52,6 +54,14 @@ type HTTPDataSource[T any] struct { typedName fwkplugin.TypedName scheme string path string + // portOverride, when non-zero, replaces the port in the endpoint's + // MetricsHost with this value. This allows a source to target a + // different port on the same pod (e.g. DCGM Exporter on :9400) + // without changing the endpoint metadata set by the discovery layer. + portOverride int + // useNodeAddress, when true, scrapes NodeAddress:portOverride instead + // of the pod IP. Used for node-level exporters (e.g. DCGM DaemonSet). + useNodeAddress bool client Client // parser converts the response body to T. MUST NOT return (zero, nil) for nilable T; @@ -76,13 +86,45 @@ type TLSOptions struct { ClientKeyPath string } +// Option configures optional behaviour on an HTTPDataSource. +type Option func(*options) + +type options struct { + portOverride int + useNodeAddress bool +} + +// WithPortOverride makes the source scrape podIP:port instead of the +// endpoint's MetricsHost. Use this when a sidecar (e.g. DCGM Exporter) +// listens on a different port than the inference server. +func WithPortOverride(port int) Option { + return func(o *options) { o.portOverride = port } +} + +// WithUseNodeAddress makes the source scrape nodeIP:portOverride instead +// of podIP:portOverride. Requires a non-zero portOverride and a non-empty +// NodeAddress on the endpoint metadata. +func WithUseNodeAddress() Option { + return func(o *options) { o.useNodeAddress = true } +} + // NewHTTPDataSource constructs a typed polling dispatcher. For https, tlsOpts configures // server verification (CACertPath) and optional mTLS (ClientCertPath/ClientKeyPath). func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions, - pluginType, pluginName string, parser func(io.Reader) (T, error)) (*HTTPDataSource[T], error) { + pluginType, pluginName string, parser func(io.Reader) (T, error), + opts ...Option) (*HTTPDataSource[T], error) { if scheme != "http" && scheme != "https" { return nil, fmt.Errorf("unsupported scheme: %s", scheme) } + + var cfg options + for _, o := range opts { + o(&cfg) + } + if cfg.useNodeAddress && cfg.portOverride == 0 { + return nil, errors.New("WithUseNodeAddress requires a non-zero WithPortOverride") + } + cl := &client{ Client: http.Client{ Timeout: timeout, @@ -99,11 +141,13 @@ func NewHTTPDataSource[T any](scheme, path string, tlsOpts TLSOptions, cl.Transport = httpsTransport } return &HTTPDataSource[T]{ - typedName: fwkplugin.TypedName{Type: pluginType, Name: pluginName}, - scheme: scheme, - path: path, - client: cl, - parser: parser, + typedName: fwkplugin.TypedName{Type: pluginType, Name: pluginName}, + scheme: scheme, + path: path, + portOverride: cfg.portOverride, + useNodeAddress: cfg.useNodeAddress, + client: cl, + parser: parser, }, nil } @@ -231,5 +275,15 @@ func (s *HTTPDataSource[T]) AppendExtractor(ext fwkplugin.Plugin) error { } func (s *HTTPDataSource[T]) getEndpoint(ep Addressable) *url.URL { - return &url.URL{Scheme: s.scheme, Host: ep.GetMetricsHost(), Path: s.path} + host := ep.GetMetricsHost() + if s.portOverride > 0 { + ip := ep.GetIPAddress() + if s.useNodeAddress { + if nodeIP := ep.GetNodeAddress(); nodeIP != "" { + ip = nodeIP + } + } + host = net.JoinHostPort(ip, strconv.Itoa(s.portOverride)) + } + return &url.URL{Scheme: s.scheme, Host: host, Path: s.path} } diff --git a/pkg/epp/framework/plugins/datalayer/source/http/datasource_test.go b/pkg/epp/framework/plugins/datalayer/source/http/datasource_test.go index 236e84e943..bb84d52709 100644 --- a/pkg/epp/framework/plugins/datalayer/source/http/datasource_test.go +++ b/pkg/epp/framework/plugins/datalayer/source/http/datasource_test.go @@ -213,6 +213,71 @@ func (c *cancelOnExtract) Extract(ctx context.Context, in fwkdl.PollInput[int]) return err } +func TestGetEndpoint_PortOverride(t *testing.T) { + parser := func(r io.Reader) (int, error) { return 0, nil } + ep := &fwkdl.EndpointMetadata{ + Address: "10.0.0.1", + NodeAddress: "192.168.1.10", + MetricsHost: "10.0.0.1:8000", + } + + cases := []struct { + name string + opts []Option + wantHost string + }{ + { + name: "no option uses MetricsHost as-is", + opts: nil, + wantHost: "10.0.0.1:8000", + }, + { + name: "WithPortOverride overrides port with pod IP", + opts: []Option{WithPortOverride(9400)}, + wantHost: "10.0.0.1:9400", + }, + { + name: "WithUseNodeAddress scrapes node IP with port override", + opts: []Option{WithPortOverride(9400), WithUseNodeAddress()}, + wantHost: "192.168.1.10:9400", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + s, err := NewHTTPDataSource("http", "/metrics", TLSOptions{SkipVerify: true}, "test", "test", parser, tc.opts...) + assert.NoError(t, err) + got := s.getEndpoint(ep) + assert.Equal(t, tc.wantHost, got.Host) + assert.Equal(t, "http", got.Scheme) + assert.Equal(t, "/metrics", got.Path) + }) + } +} + +func TestNewHTTPDataSource_UseNodeAddressWithoutPortOverride(t *testing.T) { + parser := func(r io.Reader) (int, error) { return 0, nil } + _, err := NewHTTPDataSource("http", "/metrics", TLSOptions{SkipVerify: true}, "test", "test", parser, + WithUseNodeAddress()) + require.Error(t, err) + assert.Contains(t, err.Error(), "WithUseNodeAddress requires a non-zero WithPortOverride") +} + +func TestGetEndpoint_UseNodeAddress_FallsBackToPodIP(t *testing.T) { + parser := func(r io.Reader) (int, error) { return 0, nil } + ep := &fwkdl.EndpointMetadata{ + Address: "10.0.0.1", + NodeAddress: "", + MetricsHost: "10.0.0.1:8000", + } + s, err := NewHTTPDataSource("http", "/metrics", TLSOptions{SkipVerify: true}, "test", "test", parser, + WithPortOverride(9400), WithUseNodeAddress()) + require.NoError(t, err) + got := s.getEndpoint(ep) + assert.Equal(t, "10.0.0.1:9400", got.Host, + "empty NodeAddress must fall back to pod IP") +} + func TestAppendExtractor(t *testing.T) { cases := []struct { name string