epp: generalize endpoint identity and add endpoint type

Rename EndpointMetadata.PodName to Name -- the endpoint model is not
Kubernetes-only (file-discovered endpoints have no pod) and NamespacedName is the
identity. Name is always set, so classifying an endpoint no longer requires
blanking its identity.

Add EndpointMetadata.Type (EndpointTypeEngine, EndpointTypeEPP): what an endpoint
is, independent of how it was discovered. Engines are discovered from Pod objects
or from a file and are still engines; an EPP endpoint publishes aggregates over
the servers it schedules rather than engine metrics. The Kubernetes pod path sets
engine, and file-discovery parses the entry's type field, defaulting to engine and
rejecting unknown values.

The type is an attribute set at creation rather than a label, and the metrics
extractor is unchanged: EPP metrics warrant their own extractor rather than
overloading engine type.

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

Author: hexfusion

docs/discovery.md

@@ -115,11 +115,20 @@ type DiscoveryNotifier interface {
 | Field | Type | Description |
 |---|---|---|
 | `NamespacedName` | `types.NamespacedName` | Unique identity of the endpoint. |
-| `PodName` | `string` | Logical name (used in metrics). |
+| `Name` | `string` | Name of the workload behind the endpoint, shared by all its ranks. `NamespacedName` is the identity. |
 | `Address` | `string` | IP address of the inference server. |
 | `Port` | `string` | Port as a string (e.g. `"8000"`). |
 | `MetricsHost` | `string` | `host:port` for metrics scraping. Defaults to `address:port` if empty. |
 | `Labels` | `map[string]string` | Arbitrary labels available to scheduler plugins. |
+| `Type` | `EndpointType` | What the endpoint is: `engine` or `epp`. Defaults to `engine` when omitted in the endpoints file. |
+
+### Endpoint types
+
+`engine` is an inference server. Its load signals are that server's own engine metrics,
+and the `llm-d.ai/engine-type` label selects which schema to read (`vllm`, `sglang`, ...).
+
+`epp` is an endpoint picker. Its load signals are aggregates over the servers behind it,
+so `llm-d.ai/engine-type` does not apply.
 
 ### Ordering contract
 
@@ -197,6 +206,7 @@ endpoints:
     port: <string>              # required -- integer 1-65535 as a string
     labels:                     # optional -- arbitrary key/value labels
       <key>: <value>
+    type: <engine|epp>          # optional -- defaults to engine
 ```
 
 Example with two vLLM instances:

pkg/epp/controller/pod_reconciler_test.go

@@ -219,7 +219,7 @@ func TestPodReconciler(t *testing.T) {
 				gotPods := make([]*corev1.Pod, len(podList))
 				for idx, pm := range podList {
 					gotPods[idx] = &corev1.Pod{
-						ObjectMeta: metav1.ObjectMeta{Name: pm.GetMetadata().PodName, Namespace: pm.GetMetadata().NamespacedName.Namespace},
+						ObjectMeta: metav1.ObjectMeta{Name: pm.GetMetadata().Name, Namespace: pm.GetMetadata().NamespacedName.Namespace},
 						Status:     corev1.PodStatus{PodIP: pm.GetMetadata().GetIPAddress()},
 					}
 				}

pkg/epp/datalayer/logger/logger_test.go

@@ -84,10 +84,10 @@ func TestLogger(t *testing.T) {
 
 	logOutput := b.read()
 	assert.Contains(t, logOutput, "Refreshing Prometheus Metrics	{\"ReadyPods\": 2}")
-	assert.Contains(t, logOutput, "Current Pods and metrics gathered	{\"Fresh metrics\": \"[Metadata: {NamespacedName:default/pod1 PodName: Address:1.2.3.4:5678")
+	assert.Contains(t, logOutput, "Current Pods and metrics gathered	{\"Fresh metrics\": \"[Metadata: {NamespacedName:default/pod1 Name: Address:1.2.3.4:5678")
 	assert.Contains(t, logOutput, "Metrics: {ActiveModels:map[modelA:1] WaitingModels:map[modelB:2] MaxActiveModels:5")
 	assert.Contains(t, logOutput, "RunningRequestsSize:3 WaitingQueueSize:7 KVCacheUsagePercent:42.5 KvCacheMaxTokenCapacity:2048")
-	assert.Contains(t, logOutput, "Metadata: {NamespacedName:default/pod2 PodName: Address:1.2.3.4:5679")
+	assert.Contains(t, logOutput, "Metadata: {NamespacedName:default/pod2 Name: Address:1.2.3.4:5679")
 	assert.Contains(t, logOutput, "\"Stale metrics\": \"[]\"")
 }
 

pkg/epp/datastore/datastore.go

@@ -336,12 +336,13 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P
 		pods = append(pods,
 			&fwkdl.EndpointMetadata{
 				NamespacedName: createEndpointNamespacedName(pod, idx),
-				PodName:        pod.Name,
+				Name:           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,
+				Type:           fwkdl.EndpointTypeEngine,
 				RankIndex:      idx,
 			})
 	}
@@ -394,7 +395,7 @@ func (ds *datastore) podUpdateOrAddIfNotExist(ctx context.Context, pod *corev1.P
 func (ds *datastore) PodDelete(podName string) {
 	ds.pods.Range(func(k, v any) bool {
 		ep := v.(fwkdl.Endpoint)
-		if ep.GetMetadata().PodName == podName {
+		if ep.GetMetadata().Name == podName {
 			ds.pods.Delete(k)
 			ds.epf.ReleaseEndpoint(ep)
 		}

pkg/epp/datastore/datastore_test.go

@@ -514,7 +514,7 @@ func TestPods(t *testing.T) {
 				gotPods := make([]*corev1.Pod, len(podList))
 				for idx, pm := range podList {
 					gotPods[idx] = &corev1.Pod{
-						ObjectMeta: metav1.ObjectMeta{Name: pm.GetMetadata().PodName, Namespace: pm.GetMetadata().NamespacedName.Namespace},
+						ObjectMeta: metav1.ObjectMeta{Name: pm.GetMetadata().Name, Namespace: pm.GetMetadata().NamespacedName.Namespace},
 						Status: corev1.PodStatus{
 							PodIP:  pm.GetMetadata().GetIPAddress(),
 							HostIP: pm.GetMetadata().GetNodeAddress(),
@@ -659,12 +659,13 @@ func TestEndpointMetadata(t *testing.T) {
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolTargetPort,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolTargetPort),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 				},
 			},
 			op: func(ctx context.Context, ds Datastore) {
@@ -682,25 +683,27 @@ func TestEndpointMetadata(t *testing.T) {
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort0,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 				},
 				{
 					NamespacedName: types.NamespacedName{
 						Name:      pod1.Name + "-rank-1",
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort1,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 					RankIndex:   1,
 				},
 			},
@@ -719,25 +722,27 @@ func TestEndpointMetadata(t *testing.T) {
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort0,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 				},
 				{
 					NamespacedName: types.NamespacedName{
 						Name:      pod1.Name + "-rank-1",
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort1,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 					RankIndex:   1,
 				},
 				{
@@ -746,25 +751,27 @@ func TestEndpointMetadata(t *testing.T) {
 						Namespace: pod2.Namespace,
 					},
 
-					PodName:     pod2.Name,
+					Name:        pod2.Name,
 					Address:     pod2.Status.PodIP,
 					NodeAddress: pod2.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort0,
 					MetricsHost: net.JoinHostPort(pod2.Status.PodIP, inferencePoolMultiTargetPort0),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 				},
 				{
 					NamespacedName: types.NamespacedName{
 						Name:      pod2.Name + "-rank-1",
 						Namespace: pod2.Namespace,
 					},
 
-					PodName:     pod2.Name,
+					Name:        pod2.Name,
 					Address:     pod2.Status.PodIP,
 					NodeAddress: pod2.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort1,
 					MetricsHost: net.JoinHostPort(pod2.Status.PodIP, inferencePoolMultiTargetPort1),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 					RankIndex:   1,
 				},
 			},
@@ -783,25 +790,27 @@ func TestEndpointMetadata(t *testing.T) {
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort0,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort0),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 				},
 				{
 					NamespacedName: types.NamespacedName{
 						Name:      pod1.Name + "-rank-1",
 						Namespace: pod1.Namespace,
 					},
 
-					PodName:     pod1.Name,
+					Name:        pod1.Name,
 					Address:     pod1.Status.PodIP,
 					NodeAddress: pod1.Status.HostIP,
 					Port:        inferencePoolMultiTargetPort1,
 					MetricsHost: net.JoinHostPort(pod1.Status.PodIP, inferencePoolMultiTargetPort1),
 					Labels:      map[string]string{},
+					Type:        fwkdl.EndpointTypeEngine,
 					RankIndex:   1,
 				},
 			},

pkg/epp/framework/interface/datalayer/endpoint_metadata.go

@@ -17,23 +17,48 @@ limitations under the License.
 package datalayer
 
 import (
+	"errors"
 	"fmt"
 	"maps"
 
 	"k8s.io/apimachinery/pkg/types"
 )
 
-// EndpointMetadata represents the relevant Kubernetes Pod state of an inference server.
+type EndpointType string
+
+const (
+	EndpointTypeEngine EndpointType = "engine" // an inference server
+	EndpointTypeEPP    EndpointType = "epp"    // an endpoint picker over its own servers
+)
+
+// ErrUnknownEndpointType is returned by ParseEndpointType for an unrecognized type.
+var ErrUnknownEndpointType = errors.New("unknown endpoint type")
+
+// ParseEndpointType returns the endpoint type named by s, defaulting to engine when
+// unset, or ErrUnknownEndpointType for any other value.
+func ParseEndpointType(s string) (EndpointType, error) {
+	switch t := EndpointType(s); t {
+	case "":
+		return EndpointTypeEngine, nil
+	case EndpointTypeEngine, EndpointTypeEPP:
+		return t, nil
+	default:
+		return "", fmt.Errorf("%w %q", ErrUnknownEndpointType, s)
+	}
+}
+
+// EndpointMetadata describes an inference endpoint.
 type EndpointMetadata struct {
 	NamespacedName types.NamespacedName
-	PodName        string
+	Name           string
 	Address        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
+	Type        EndpointType
 	// RankIndex is this endpoint's position in the pool's TargetPorts,
 	// identifying the pod-local rank in multi-port deployments.
 	RankIndex int
@@ -60,12 +85,13 @@ func (epm *EndpointMetadata) Clone() *EndpointMetadata {
 			Name:      epm.NamespacedName.Name,
 			Namespace: epm.NamespacedName.Namespace,
 		},
-		PodName:     epm.PodName,
+		Name:        epm.Name,
 		Address:     epm.Address,
 		NodeAddress: epm.NodeAddress,
 		Port:        epm.Port,
 		MetricsHost: epm.MetricsHost,
 		Labels:      clonedLabels,
+		Type:        epm.Type,
 		RankIndex:   epm.RankIndex,
 	}
 }
@@ -77,11 +103,12 @@ func (epm *EndpointMetadata) Equal(other *EndpointMetadata) bool {
 		return epm == other
 	}
 	return epm.NamespacedName == other.NamespacedName &&
-		epm.PodName == other.PodName &&
+		epm.Name == other.Name &&
 		epm.Address == other.Address &&
 		epm.NodeAddress == other.NodeAddress &&
 		epm.Port == other.Port &&
 		epm.MetricsHost == other.MetricsHost &&
+		epm.Type == other.Type &&
 		epm.RankIndex == other.RankIndex &&
 		maps.Equal(epm.Labels, other.Labels)
 }

pkg/epp/framework/interface/datalayer/endpoint_metadata_test.go

@@ -69,11 +69,12 @@ func TestEndpointMetadataClone(t *testing.T) {
 func TestEndpointMetadataEqual(t *testing.T) {
 	base := &EndpointMetadata{
 		NamespacedName: types.NamespacedName{Name: "pod-a-rank-0", Namespace: "default"},
-		PodName:        "pod-a",
+		Name:           "pod-a",
 		Address:        "10.0.0.1",
 		Port:           "8000",
 		MetricsHost:    "10.0.0.1:9000",
 		Labels:         map[string]string{"app": "vllm"},
+		Type:           EndpointTypeEngine,
 		RankIndex:      1,
 	}
 
@@ -99,9 +100,9 @@ func TestEndpointMetadataEqual(t *testing.T) {
 			},
 		},
 		{
-			name: "pod name",
+			name: "endpoint name",
 			mutate: func(meta *EndpointMetadata) {
-				meta.PodName = "pod-b"
+				meta.Name = "pod-b"
 			},
 		},
 		{
@@ -128,6 +129,12 @@ func TestEndpointMetadataEqual(t *testing.T) {
 				meta.Labels["app"] = "sglang"
 			},
 		},
+		{
+			name: "type",
+			mutate: func(meta *EndpointMetadata) {
+				meta.Type = EndpointTypeEPP
+			},
+		},
 		{
 			name: "rank index",
 			mutate: func(meta *EndpointMetadata) {
@@ -151,7 +158,7 @@ func TestEndpointMetadataString(t *testing.T) {
 			Name:      pod.Name,
 			Namespace: pod.Namespace,
 		},
-		PodName:     pod.Name,
+		Name:        pod.Name,
 		Address:     pod.Status.PodIP,
 		Port:        "8000",
 		MetricsHost: "127.0.0.1:8000",

pkg/epp/framework/interface/scheduling/types_test.go

@@ -33,7 +33,7 @@ func (s cloneableString) Clone() fwkdl.Cloneable { return s }
 func newTestMetadata(name string) *fwkdl.EndpointMetadata {
 	return &fwkdl.EndpointMetadata{
 		NamespacedName: types.NamespacedName{Namespace: "ns", Name: name},
-		PodName:        name,
+		Name:           name,
 		Address:        "10.0.0.1",
 		Port:           "8000",
 		MetricsHost:    "10.0.0.1:9100",

pkg/epp/framework/plugins/datalayer/discovery/file/plugin.go

@@ -48,6 +48,7 @@ type EndpointEntry struct {
 	Address   string            `json:"address"             yaml:"address"`
 	Port      string            `json:"port"                yaml:"port"`
 	Labels    map[string]string `json:"labels,omitempty"    yaml:"labels,omitempty"`
+	Type      string            `json:"type,omitempty"      yaml:"type,omitempty"`
 }
 
 // EndpointsFile is the top-level structure of the endpoints YAML/JSON file.
@@ -246,17 +247,23 @@ func (f *FileDiscovery) load(notifier fwkdl.DiscoveryNotifier) error {
 			errs = append(errs, fmt.Errorf("endpoint %q: invalid port %q", e.Name, e.Port))
 			continue
 		}
+		epType, err := fwkdl.ParseEndpointType(e.Type)
+		if err != nil {
+			errs = append(errs, fmt.Errorf("endpoint %q: %w", e.Name, err))
+			continue
+		}
 		ns := e.Namespace
 		if ns == "" {
 			ns = "default"
 		}
 		meta := &fwkdl.EndpointMetadata{
 			NamespacedName: types.NamespacedName{Name: e.Name, Namespace: ns},
-			PodName:        e.Name,
+			Name:           e.Name,
 			Address:        e.Address,
 			Port:           e.Port,
 			MetricsHost:    net.JoinHostPort(e.Address, e.Port),
 			Labels:         e.Labels,
+			Type:           epType,
 		}
 		incoming[meta.NamespacedName] = struct{}{}
 		notifier.Upsert(meta)