feat(datalayer): extend file-discovery to support cluster endpoints

Signed-off-by: Liav Weiss <lweiss@redhat.com>

Author: liavweiss

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

@@ -28,10 +28,12 @@ import (
 	"os"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 
 	"github.com/fsnotify/fsnotify"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/validation"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/yaml"
 
@@ -43,11 +45,12 @@ const PluginType = "file-discovery"
 
 // EndpointEntry is the YAML/JSON representation of a single endpoint.
 type EndpointEntry struct {
-	Name      string            `json:"name"                yaml:"name"`
-	Namespace string            `json:"namespace,omitempty" yaml:"namespace,omitempty"`
-	Address   string            `json:"address"             yaml:"address"`
-	Port      string            `json:"port"                yaml:"port"`
-	Labels    map[string]string `json:"labels,omitempty"    yaml:"labels,omitempty"`
+	Name        string            `json:"name"                    yaml:"name"`
+	Namespace   string            `json:"namespace,omitempty"     yaml:"namespace,omitempty"`
+	Address     string            `json:"address"                 yaml:"address"`
+	Port        string            `json:"port"                    yaml:"port"`
+	MetricsPort string            `json:"metricsPort,omitempty"   yaml:"metricsPort,omitempty"`
+	Labels      map[string]string `json:"labels,omitempty"        yaml:"labels,omitempty"`
 }
 
 // EndpointsFile is the top-level structure of the endpoints YAML/JSON file.
@@ -238,9 +241,12 @@ func (f *FileDiscovery) load(notifier fwkdl.DiscoveryNotifier) error {
 	incoming := make(map[types.NamespacedName]struct{}, len(ef.Endpoints))
 	var errs []error
 	for _, e := range ef.Endpoints {
-		if ip := net.ParseIP(e.Address); ip == nil || ip.To4() == nil {
-			errs = append(errs, fmt.Errorf("endpoint %q: invalid IPv4 address %q", e.Name, e.Address))
-			continue
+		ip := net.ParseIP(e.Address)
+		if ip == nil || ip.To4() == nil {
+			if len(validation.IsDNS1123Subdomain(strings.ToLower(e.Address))) > 0 {
+				errs = append(errs, fmt.Errorf("endpoint %q: address %q is not a valid IPv4 or hostname", e.Name, e.Address))
+				continue
+			}
 		}
 		if portNum, err := strconv.Atoi(e.Port); err != nil || portNum < 1 || portNum > 65535 {
 			errs = append(errs, fmt.Errorf("endpoint %q: invalid port %q", e.Name, e.Port))
@@ -250,12 +256,20 @@ func (f *FileDiscovery) load(notifier fwkdl.DiscoveryNotifier) error {
 		if ns == "" {
 			ns = "default"
 		}
+		podName := e.Name
+		if ip == nil {
+			podName = ""
+		}
+		mPort := e.Port
+		if e.MetricsPort != "" {
+			mPort = e.MetricsPort
+		}
 		meta := &fwkdl.EndpointMetadata{
 			NamespacedName: types.NamespacedName{Name: e.Name, Namespace: ns},
-			PodName:        e.Name,
+			PodName:        podName,
 			Address:        e.Address,
 			Port:           e.Port,
-			MetricsHost:    net.JoinHostPort(e.Address, e.Port),
+			MetricsHost:    net.JoinHostPort(e.Address, mPort),
 			Labels:         e.Labels,
 		}
 		incoming[meta.NamespacedName] = struct{}{}

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

@@ -189,15 +189,39 @@ func TestStart_MissingFile(t *testing.T) {
 	assert.ErrorContains(t, err, "initial load failed")
 }
 
-func TestStart_InvalidIP(t *testing.T) {
+func TestStart_EmptyAddress(t *testing.T) {
 	path := writeTemp(t, `
 endpoints:
   - name: ep1
-    address: "not-an-ip"
+    address: ""
     port: "8000"
 `)
 	err := newFD(path, false).Start(context.Background(), &recordingNotifier{})
-	assert.ErrorContains(t, err, "invalid IPv4 address")
+	assert.ErrorContains(t, err, "is not a valid IPv4 or hostname")
+}
+
+func TestStart_InvalidAddressFormat(t *testing.T) {
+	tests := []struct {
+		name    string
+		address string
+	}{
+		{"spaces", "not a valid host"},
+		{"special chars", "!!!garbage!!!"},
+		{"trailing dot label", "host-.example.com"},
+		{"leading hyphen", "-host.example.com"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			path := writeTemp(t, fmt.Sprintf(`
+endpoints:
+  - name: ep1
+    address: %q
+    port: "8000"
+`, tt.address))
+			err := newFD(path, false).Start(context.Background(), &recordingNotifier{})
+			assert.ErrorContains(t, err, "is not a valid IPv4 or hostname")
+		})
+	}
 }
 
 func TestStart_RejectsIPv6(t *testing.T) {
@@ -208,7 +232,125 @@ endpoints:
     port: "8000"
 `)
 	err := newFD(path, false).Start(context.Background(), &recordingNotifier{})
-	assert.ErrorContains(t, err, "invalid IPv4 address")
+	assert.ErrorContains(t, err, "is not a valid IPv4 or hostname")
+}
+
+func TestStart_HostnameAddress(t *testing.T) {
+	path := writeTemp(t, `
+endpoints:
+  - name: cluster-us-east
+    namespace: clusters
+    address: "spoke-us-east.example.com"
+    port: "443"
+    labels:
+      region: us-east
+      model: llama-7b
+  - name: cluster-eu-west
+    namespace: clusters
+    address: "spoke-eu-west.example.com"
+    port: "443"
+    labels:
+      region: eu-west
+      model: llama-7b
+`)
+	notifier := &recordingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	require.NoError(t, newFD(path, false).Start(ctx, notifier))
+	require.Len(t, notifier.upserted, 2)
+
+	for _, m := range notifier.upserted {
+		assert.Empty(t, m.PodName, "hostname address should produce empty PodName")
+	}
+	assert.Equal(t, "spoke-us-east.example.com:443", notifier.upserted[0].MetricsHost)
+	assert.Equal(t, "spoke-eu-west.example.com:443", notifier.upserted[1].MetricsHost)
+	assert.ElementsMatch(t, []string{"clusters/cluster-us-east", "clusters/cluster-eu-west"}, notifier.upsertedNames())
+}
+
+func TestStart_IPv4PodNameIsSet(t *testing.T) {
+	path := writeTemp(t, `
+endpoints:
+  - name: vllm-pod-1
+    address: "10.0.0.5"
+    port: "8000"
+`)
+	notifier := &recordingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	require.NoError(t, newFD(path, false).Start(ctx, notifier))
+	require.Len(t, notifier.upserted, 1)
+	assert.Equal(t, "vllm-pod-1", notifier.upserted[0].PodName, "IPv4 address should preserve PodName")
+	assert.Equal(t, "10.0.0.5:8000", notifier.upserted[0].MetricsHost)
+}
+
+func TestStart_MixedIPv4AndHostname(t *testing.T) {
+	path := writeTemp(t, `
+endpoints:
+  - name: vllm-pod-1
+    address: "10.0.0.5"
+    port: "8000"
+  - name: cluster-us-east
+    address: "spoke-us-east.example.com"
+    port: "443"
+`)
+	notifier := &recordingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	require.NoError(t, newFD(path, false).Start(ctx, notifier))
+	require.Len(t, notifier.upserted, 2)
+
+	var pod, cluster *fwkdl.EndpointMetadata
+	for _, m := range notifier.upserted {
+		if m.NamespacedName.Name == "vllm-pod-1" {
+			pod = m
+		} else {
+			cluster = m
+		}
+	}
+	require.NotNil(t, pod)
+	require.NotNil(t, cluster)
+	assert.Equal(t, "vllm-pod-1", pod.PodName, "IPv4 endpoint should have PodName set")
+	assert.Empty(t, cluster.PodName, "hostname endpoint should have empty PodName")
+}
+
+func TestStart_MetricsPortOverride(t *testing.T) {
+	path := writeTemp(t, `
+endpoints:
+  - name: cluster-us-east
+    address: "spoke-gateway.example.com"
+    port: "443"
+    metricsPort: "9090"
+`)
+	notifier := &recordingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	require.NoError(t, newFD(path, false).Start(ctx, notifier))
+	require.Len(t, notifier.upserted, 1)
+
+	m := notifier.upserted[0]
+	assert.Equal(t, "spoke-gateway.example.com", m.Address, "Address should be unchanged")
+	assert.Equal(t, "443", m.Port, "Port should be unchanged")
+	assert.Equal(t, "spoke-gateway.example.com:9090", m.MetricsHost, "MetricsHost should use address with metricsPort")
+}
+
+func TestStart_MetricsFieldsNotSetFallsBack(t *testing.T) {
+	path := writeTemp(t, `
+endpoints:
+  - name: ep1
+    address: "10.0.0.1"
+    port: "8000"
+`)
+	notifier := &recordingNotifier{}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	require.NoError(t, newFD(path, false).Start(ctx, notifier))
+	require.Len(t, notifier.upserted, 1)
+	assert.Equal(t, "10.0.0.1:8000", notifier.upserted[0].MetricsHost, "MetricsHost should fall back to Address:Port")
 }
 
 func TestStart_InvalidPort(t *testing.T) {