feat(epp): extend topology-extractor with zone and region

Add zone and region fields to the Topology attribute and the params
struct. Each param names the pod label to read; defaults are the
standard Kubernetes topology labels (corev1.LabelHostname,
LabelTopologyZone, LabelTopologyRegion).

The hostname label falling back to spec.hostname is preserved: when
the hostname label is absent the endpoint is tracked and stamped once
the Pod notification fires. Zone and region have no fallback.

Zone and region values are not read from Node objects -- that would
require RBAC for Node GET/LIST. They are populated when the pod itself
carries the corresponding labels.

Signed-off-by: Etai Lev Ran <elevran@gmail.com>

Author: elevran

pkg/epp/framework/plugins/datalayer/attribute/topology/data_types.go

@@ -32,12 +32,15 @@ const (
 var TopologyAttributeKey = plugin.NewDataKey("Topology", TopologyExtractorType)
 
 // Topology carries the locality information for an endpoint.
-// The Hostname field is populated once when the endpoint is created and
-// does not change unless the endpoint is re-created.
+// Fields are populated from pod labels at endpoint creation time.
+// Hostname falls back to spec.hostname when the configured label is absent.
 type Topology struct {
-	// Hostname identifies the node or host on which the endpoint runs.
-	// Derived from the Pod's hostname field or from a user-configured label.
+	// Hostname identifies the node on which the endpoint runs.
 	Hostname string
+	// Zone identifies the failure domain zone of the endpoint.
+	Zone string
+	// Region identifies the geographic region of the endpoint.
+	Region string
 }
 
 // Clone returns an independent copy of the Topology.

pkg/epp/framework/plugins/datalayer/extractor/topology/extractor.go

@@ -15,16 +15,13 @@ limitations under the License.
 */
 
 // Package topology provides a datalayer plugin that stamps each endpoint with
-// a Topology attribute containing the endpoint's hostname.
+// a Topology attribute containing locality fields (hostname, zone, region).
 //
-// The plugin registers for both endpoint lifecycle events and Pod k8s notification
-// events. When hostnameLabel is configured, the label value is read from the
-// endpoint event's pod metadata and written as the Topology attribute; Pod
-// notifications are a no-op. When hostnameLabel is empty, the endpoint event
-// tracks live endpoints keyed by pod identity; the Pod notification (ready pods
-// only) reads spec.hostname, caches it, and stamps all current endpoints for
-// that pod. Whichever event fires first, the attribute is set once both have
-// been processed.
+// Label names for each field are resolved from plugin parameters, defaulting to
+// the standard Kubernetes topology labels. Labels are read from the endpoint's
+// pod metadata at endpoint event time. When the hostname label is absent from
+// the pod, the plugin falls back to spec.hostname from Pod notification events.
+// Zone and region have no fallback.
 package topology
 
 import (
@@ -33,6 +30,7 @@ import (
 	"slices"
 	"sync"
 
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
@@ -53,25 +51,43 @@ var (
 // podGVK is the core/v1 Pod GVK watched by the pod notification handler.
 var podGVK = schema.GroupVersionKind{Group: "", Version: "v1", Kind: "Pod"}
 
+const (
+	defaultHostnameLabel = corev1.LabelHostname
+	defaultZoneLabel     = corev1.LabelTopologyZone
+	defaultRegionLabel   = corev1.LabelTopologyRegion
+)
+
 // params holds the user-facing configuration for the topology extractor.
+// Each field names the pod label used to read the corresponding topology value.
+// When a field is empty, the corresponding default Kubernetes topology label is used.
+// When the hostname label is absent on a pod, spec.hostname is used as a fallback.
+// Zone and region have no fallback.
 type params struct {
-	// HostnameLabel is the pod label whose value is used as the topology hostname.
-	// When empty (default), spec.hostname from the Pod object is used instead.
-	// When set but absent on the pod, the attribute is not added.
-	HostnameLabel string `json:"hostnameLabel,omitempty"`
+	// Hostname is the pod label name whose value is used as the topology hostname.
+	// Defaults to kubernetes.io/hostname.
+	Hostname string `json:"hostname,omitempty"`
+	// Zone is the pod label name whose value is used as the topology zone.
+	// Defaults to topology.kubernetes.io/zone.
+	Zone string `json:"zone,omitempty"`
+	// Region is the pod label name whose value is used as the topology region.
+	// Defaults to topology.kubernetes.io/region.
+	Region string `json:"region,omitempty"`
 }
 
 // TopologyExtractor stamps each endpoint with a Topology attribute.
 // It registers for both endpoint lifecycle events and Pod k8s notifications.
 type TopologyExtractor struct {
 	typedName     fwkplugin.TypedName
 	hostnameLabel string
+	zoneLabel     string
+	regionLabel   string
 	dk            fwkplugin.DataKey
 
 	// mu guards endpoints and hostnames.
 	mu sync.Mutex
 	// endpoints maps pod identity to all live Endpoints for that pod.
 	// Keyed by {PodName, Namespace} — one pod may have N rank endpoints.
+	// Only endpoints whose hostname label was absent are tracked here.
 	endpoints map[types.NamespacedName][]fwkdl.Endpoint
 	// hostnames caches spec.hostname per pod (ready pods only).
 	// Populated by Pod notifications; cleared on pod delete.
@@ -86,12 +102,23 @@ func Factory(name string, parameters *json.Decoder, _ fwkplugin.Handle) (fwkplug
 			return nil, err
 		}
 	}
+	if p.Hostname == "" {
+		p.Hostname = defaultHostnameLabel
+	}
+	if p.Zone == "" {
+		p.Zone = defaultZoneLabel
+	}
+	if p.Region == "" {
+		p.Region = defaultRegionLabel
+	}
 	if name == "" {
 		name = attrtopology.TopologyExtractorType
 	}
 	return &TopologyExtractor{
 		typedName:     fwkplugin.TypedName{Type: attrtopology.TopologyExtractorType, Name: name},
-		hostnameLabel: p.HostnameLabel,
+		hostnameLabel: p.Hostname,
+		zoneLabel:     p.Zone,
+		regionLabel:   p.Region,
 		dk:            attrtopology.TopologyAttributeKey.WithNonEmptyProducerName(name),
 		endpoints:     make(map[types.NamespacedName][]fwkdl.Endpoint),
 		hostnames:     make(map[types.NamespacedName]string),
@@ -157,31 +184,31 @@ func (h *endpointHandler) Extract(_ context.Context, event fwkdl.EndpointEvent)
 	}
 
 	if event.Type == fwkdl.EventDelete {
-		if h.ext.hostnameLabel == "" {
-			h.removeEndpoint(meta, event.Endpoint)
-		}
+		h.removeEndpoint(meta, event.Endpoint)
 		return nil
 	}
 
-	if h.ext.hostnameLabel != "" {
-		val, ok := meta.Labels[h.ext.hostnameLabel]
-		if !ok || val == "" {
-			return nil
-		}
-		event.Endpoint.GetAttributes().Put(h.ext.dk.String(), &attrtopology.Topology{Hostname: val})
-		return nil
-	}
+	hn := meta.Labels[h.ext.hostnameLabel]
+	zone := meta.Labels[h.ext.zoneLabel]
+	region := meta.Labels[h.ext.regionLabel]
 
-	// No-label path: track endpoint; stamp immediately if hostname already cached.
-	key := podKey(meta)
-	h.ext.mu.Lock()
-	h.ext.endpoints[key] = append(h.ext.endpoints[key], event.Endpoint)
-	hn := h.ext.hostnames[key]
-	h.ext.mu.Unlock()
+	if hn == "" {
+		// Hostname label absent: track endpoint for spec.hostname fallback via pod notification.
+		key := podKey(meta)
+		h.ext.mu.Lock()
+		h.ext.endpoints[key] = append(h.ext.endpoints[key], event.Endpoint)
+		hn = h.ext.hostnames[key]
+		h.ext.mu.Unlock()
+	}
 
-	if hn != "" {
-		event.Endpoint.GetAttributes().Put(h.ext.dk.String(), &attrtopology.Topology{Hostname: hn})
+	if hn == "" && zone == "" && region == "" {
+		return nil
 	}
+	event.Endpoint.GetAttributes().Put(h.ext.dk.String(), &attrtopology.Topology{
+		Hostname: hn,
+		Zone:     zone,
+		Region:   region,
+	})
 	return nil
 }
 
@@ -224,10 +251,6 @@ func (h *podNotificationHandler) GVK() schema.GroupVersionKind {
 }
 
 func (h *podNotificationHandler) Extract(_ context.Context, event fwkdl.NotificationEvent) error {
-	if h.ext.hostnameLabel != "" {
-		return nil
-	}
-
 	obj := event.Object
 	key := types.NamespacedName{Name: obj.GetName(), Namespace: obj.GetNamespace()}
 
@@ -249,7 +272,15 @@ func (h *podNotificationHandler) Extract(_ context.Context, event fwkdl.Notifica
 	h.ext.mu.Unlock()
 
 	for _, ep := range eps {
-		ep.GetAttributes().Put(h.ext.dk.String(), &attrtopology.Topology{Hostname: hostname})
+		// Preserve zone/region already set from the endpoint event.
+		topo := &attrtopology.Topology{Hostname: hostname}
+		if raw, ok := ep.GetAttributes().Get(h.ext.dk.String()); ok {
+			if existing, ok := raw.(*attrtopology.Topology); ok {
+				topo.Zone = existing.Zone
+				topo.Region = existing.Region
+			}
+		}
+		ep.GetAttributes().Put(h.ext.dk.String(), topo)
 	}
 	return nil
 }

pkg/epp/framework/plugins/datalayer/extractor/topology/extractor_test.go

@@ -148,7 +148,7 @@ func podEvent(t fwkdl.EventType, pod *unstructured.Unstructured) fwkdl.Notificat
 // --- label-configured path ---
 
 func TestLabel_EndpointHandler_LabelPresent(t *testing.T) {
-	_, epH, _ := getHandlers(t, &params{HostnameLabel: "topology.hostname"})
+	_, epH, _ := getHandlers(t, &params{Hostname: "topology.hostname"})
 
 	ep := newRankEndpoint("worker-1", 0, map[string]string{"topology.hostname": "rack-42"})
 	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
@@ -165,7 +165,7 @@ func TestLabel_EndpointHandler_LabelPresent(t *testing.T) {
 }
 
 func TestLabel_EndpointHandler_LabelMissing(t *testing.T) {
-	_, epH, _ := getHandlers(t, &params{HostnameLabel: "topology.hostname"})
+	_, epH, _ := getHandlers(t, &params{Hostname: "topology.hostname"})
 
 	ep := newRankEndpoint("worker-2", 0, map[string]string{"other": "value"})
 	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
@@ -177,7 +177,7 @@ func TestLabel_EndpointHandler_LabelMissing(t *testing.T) {
 }
 
 func TestLabel_EndpointHandler_NilLabels(t *testing.T) {
-	_, epH, _ := getHandlers(t, &params{HostnameLabel: "topology.hostname"})
+	_, epH, _ := getHandlers(t, &params{Hostname: "topology.hostname"})
 
 	ep := newRankEndpoint("worker-3", 0, nil)
 	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
@@ -189,7 +189,7 @@ func TestLabel_EndpointHandler_NilLabels(t *testing.T) {
 }
 
 func TestLabel_EndpointHandler_DeleteIsNoop(t *testing.T) {
-	_, epH, _ := getHandlers(t, &params{HostnameLabel: "topology.hostname"})
+	_, epH, _ := getHandlers(t, &params{Hostname: "topology.hostname"})
 
 	ep := newRankEndpoint("worker-4", 0, map[string]string{"topology.hostname": "rack-1"})
 	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventDelete, ep)); err != nil {
@@ -201,7 +201,7 @@ func TestLabel_EndpointHandler_DeleteIsNoop(t *testing.T) {
 }
 
 func TestLabel_PodHandler_IsNoop(t *testing.T) {
-	_, _, podH := getHandlers(t, &params{HostnameLabel: "topology.hostname"})
+	_, _, podH := getHandlers(t, &params{Hostname: "topology.hostname"})
 
 	pod := makePod("worker-1", "actual-hostname", true)
 	if err := podH.Extract(context.Background(), podEvent(fwkdl.EventAddOrUpdate, pod)); err != nil {
@@ -455,6 +455,124 @@ func TestNoLabel_EndpointDelete_RemovesFromMap(t *testing.T) {
 	}
 }
 
+// --- zone and region label extraction ---
+
+func TestLabel_ZoneAndRegion(t *testing.T) {
+	_, epH, _ := getHandlers(t, &params{
+		Hostname: "my/hostname",
+		Zone:     "my/zone",
+		Region:   "my/region",
+	})
+
+	ep := newRankEndpoint("worker-z1", 0, map[string]string{
+		"my/hostname": "host-1",
+		"my/zone":     "us-east-1a",
+		"my/region":   "us-east-1",
+	})
+	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
+		t.Fatalf("Extract: %v", err)
+	}
+
+	got, ok := readTopology(ep)
+	if !ok {
+		t.Fatal("expected Topology attribute")
+	}
+	if got.Hostname != "host-1" {
+		t.Errorf("hostname = %q, want %q", got.Hostname, "host-1")
+	}
+	if got.Zone != "us-east-1a" {
+		t.Errorf("zone = %q, want %q", got.Zone, "us-east-1a")
+	}
+	if got.Region != "us-east-1" {
+		t.Errorf("region = %q, want %q", got.Region, "us-east-1")
+	}
+}
+
+func TestLabel_ZoneAndRegionDefault(t *testing.T) {
+	_, epH, _ := getHandlers(t, nil)
+
+	ep := newRankEndpoint("worker-z2", 0, map[string]string{
+		defaultHostnameLabel: "host-2",
+		defaultZoneLabel:     "eu-west-1b",
+		defaultRegionLabel:   "eu-west-1",
+	})
+	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
+		t.Fatalf("Extract: %v", err)
+	}
+
+	got, ok := readTopology(ep)
+	if !ok {
+		t.Fatal("expected Topology attribute")
+	}
+	if got.Hostname != "host-2" {
+		t.Errorf("hostname = %q, want %q", got.Hostname, "host-2")
+	}
+	if got.Zone != "eu-west-1b" {
+		t.Errorf("zone = %q, want %q", got.Zone, "eu-west-1b")
+	}
+	if got.Region != "eu-west-1" {
+		t.Errorf("region = %q, want %q", got.Region, "eu-west-1")
+	}
+}
+
+func TestLabel_ZoneOnlyNoHostname_FallsBackToSpecHostname(t *testing.T) {
+	_, epH, podH := getHandlers(t, &params{Zone: "my/zone"})
+
+	// Endpoint has zone label but not the hostname label (defaults to kubernetes.io/hostname).
+	ep := newRankEndpoint("worker-z3", 0, map[string]string{
+		"my/zone": "ap-south-1a",
+	})
+	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
+		t.Fatalf("epH.Extract: %v", err)
+	}
+
+	// Zone set but hostname still empty — pod notification provides the fallback.
+	pod := makePod("worker-z3", testNodeHostname, true)
+	if err := podH.Extract(context.Background(), podEvent(fwkdl.EventAddOrUpdate, pod)); err != nil {
+		t.Fatalf("podH.Extract: %v", err)
+	}
+
+	got, ok := readTopology(ep)
+	if !ok {
+		t.Fatal("expected Topology attribute")
+	}
+	if got.Hostname != testNodeHostname {
+		t.Errorf("hostname = %q, want %q", got.Hostname, testNodeHostname)
+	}
+	if got.Zone != "ap-south-1a" {
+		t.Errorf("zone = %q, want %q", got.Zone, "ap-south-1a")
+	}
+}
+
+func TestLabel_HostnameLabelAbsent_FallsBackToSpecHostname(t *testing.T) {
+	_, epH, podH := getHandlers(t, &params{Hostname: "custom/hostname"})
+
+	// Endpoint does not have the configured hostname label.
+	ep := newRankEndpoint("worker-z4", 0, map[string]string{"other": "val"})
+	if err := epH.Extract(context.Background(), epEvent(fwkdl.EventAddOrUpdate, ep)); err != nil {
+		t.Fatalf("epH.Extract: %v", err)
+	}
+
+	// No attribute yet.
+	if _, ok := readTopology(ep); ok {
+		t.Error("expected no Topology attribute before pod notification")
+	}
+
+	// Pod notification provides spec.hostname fallback.
+	pod := makePod("worker-z4", testNodeHostname, true)
+	if err := podH.Extract(context.Background(), podEvent(fwkdl.EventAddOrUpdate, pod)); err != nil {
+		t.Fatalf("podH.Extract: %v", err)
+	}
+
+	got, ok := readTopology(ep)
+	if !ok {
+		t.Fatal("expected Topology attribute after pod notification")
+	}
+	if got.Hostname != testNodeHostname {
+		t.Errorf("hostname = %q, want %q", got.Hostname, testNodeHostname)
+	}
+}
+
 // --- registration ---
 
 func TestRegisterDependencies_RegistersBothHandlers(t *testing.T) {