Add DCGM data source and extractor for GPU utilization

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)

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,

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{},

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

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.

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

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.