Fix missing labels in watch events

Storage-generated watch events did not include pod/node labels because
the storage layer has no access to the pod/node listers. This caused
any watcher with a non-empty label selector to receive zero events,
since matchesFilter() checked labels.Set(m.Labels) which was always nil.

Fix: Add a LabelLookupFunc to metricsWatcher that enriches events with
labels from the authoritative source (pod/node listers) before filtering
and forwarding. The Watch helpers in the API layer create this function
from the pod/node listers available at construction time.

The enrichLabels() method is called in Send() before matchesFilter(),
so both filtering and the forwarded event contain correct labels.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: hunter32292

pkg/api/node.go

@@ -62,7 +62,14 @@ func newNodeMetrics(groupResource schema.GroupResource, metrics NodeMetricsGette
 
 	// Set up watch helper if the metrics getter supports watching
 	if watchable, ok := metrics.(WatchableNodeMetricsGetter); ok {
-		nm.watchHelper = NewNodeMetricsWatchHelper(watchable)
+		labelLookup := func(_, name string) map[string]string {
+			node, err := nodeLister.Get(name)
+			if err != nil {
+				return nil
+			}
+			return node.Labels
+		}
+		nm.watchHelper = NewNodeMetricsWatchHelper(watchable, labelLookup)
 	}
 
 	return nm

pkg/api/pod.go

@@ -61,7 +61,17 @@ func newPodMetrics(groupResource schema.GroupResource, metrics PodMetricsGetter,
 
 	// Set up watch helper if the metrics getter supports watching
 	if watchable, ok := metrics.(WatchablePodMetricsGetter); ok {
-		pm.watchHelper = NewPodMetricsWatchHelper(watchable)
+		labelLookup := func(namespace, name string) map[string]string {
+			obj, err := podLister.ByNamespace(namespace).Get(name)
+			if err != nil {
+				return nil
+			}
+			if pom, ok := obj.(*metav1.PartialObjectMetadata); ok {
+				return pom.Labels
+			}
+			return nil
+		}
+		pm.watchHelper = NewPodMetricsWatchHelper(watchable, labelLookup)
 	}
 
 	return pm

pkg/api/watcher.go

@@ -34,6 +34,10 @@ const (
 	resourceTypeNode = "node"
 )
 
+// LabelLookupFunc returns labels for a given resource identified by namespace and name.
+// For cluster-scoped resources (nodes), namespace is empty.
+type LabelLookupFunc func(namespace, name string) map[string]string
+
 // metricsWatcher implements watch.Interface for metrics resources.
 // It receives events from the storage layer and filters them based on
 // namespace and label selector before forwarding to clients.
@@ -48,13 +52,17 @@ type metricsWatcher struct {
 
 	// Resource type for correct bookmark generation
 	resourceType string // "pod" or "node"
+
+	// labelLookup enriches events with labels from the authoritative source (pod/node lister).
+	// Storage-generated events don't carry labels, so this is needed for label selector filtering.
+	labelLookup LabelLookupFunc
 }
 
 var _ watch.Interface = &metricsWatcher{}
 var _ MetricsWatcher = &metricsWatcher{}
 
 // newMetricsWatcher creates a new watcher with the given filter criteria.
-func newMetricsWatcher(namespace string, labelSelector labels.Selector, resourceType string) *metricsWatcher {
+func newMetricsWatcher(namespace string, labelSelector labels.Selector, resourceType string, labelLookup LabelLookupFunc) *metricsWatcher {
 	if labelSelector == nil {
 		labelSelector = labels.Everything()
 	}
@@ -64,6 +72,7 @@ func newMetricsWatcher(namespace string, labelSelector labels.Selector, resource
 		namespace:     namespace,
 		labelSelector: labelSelector,
 		resourceType:  resourceType,
+		labelLookup:   labelLookup,
 	}
 }
 
@@ -103,15 +112,18 @@ func (w *metricsWatcher) Send(event WatchEvent) bool {
 	default:
 	}
 
+	// Enrich object with labels from the authoritative source if available
+	obj := w.enrichLabels(event.Object)
+
 	// Filter the event
-	if !w.matchesFilter(event.Object) {
+	if !w.matchesFilter(obj) {
 		return true // Event filtered out, but watcher is still alive
 	}
 
 	// Convert to watch.Event
 	watchEvent := watch.Event{
 		Type:   event.Type,
-		Object: w.toRuntimeObject(event.Object),
+		Object: w.toRuntimeObject(obj),
 	}
 
 	// Try to send, close if buffer is full (slow consumer)
@@ -145,6 +157,29 @@ func (w *metricsWatcher) matchesFilter(obj interface{}) bool {
 	}
 }
 
+// enrichLabels looks up current labels for the given metrics object.
+// Storage-generated events don't carry labels, so this uses the label lookup
+// function (backed by pod/node listers) to add them.
+func (w *metricsWatcher) enrichLabels(obj interface{}) interface{} {
+	if w.labelLookup == nil {
+		return obj
+	}
+	switch m := obj.(type) {
+	case metrics.PodMetrics:
+		if m.Labels == nil {
+			m.Labels = w.labelLookup(m.Namespace, m.Name)
+		}
+		return m
+	case metrics.NodeMetrics:
+		if m.Labels == nil {
+			m.Labels = w.labelLookup("", m.Name)
+		}
+		return m
+	default:
+		return obj
+	}
+}
+
 // toRuntimeObject converts the metrics object to a runtime.Object
 func (w *metricsWatcher) toRuntimeObject(obj interface{}) runtime.Object {
 	switch m := obj.(type) {
@@ -232,19 +267,21 @@ func (w *metricsWatcher) sendBookmark(resourceVersion string) bool {
 
 // PodMetricsWatchHelper helps create watches for pod metrics
 type PodMetricsWatchHelper struct {
-	storage WatchablePodMetricsGetter
+	storage     WatchablePodMetricsGetter
+	labelLookup LabelLookupFunc
 }
 
-// NewPodMetricsWatchHelper creates a new watch helper for pod metrics
-func NewPodMetricsWatchHelper(storage WatchablePodMetricsGetter) *PodMetricsWatchHelper {
-	return &PodMetricsWatchHelper{storage: storage}
+// NewPodMetricsWatchHelper creates a new watch helper for pod metrics.
+// labelLookup provides labels for pod metrics objects (used for label selector filtering).
+func NewPodMetricsWatchHelper(storage WatchablePodMetricsGetter, labelLookup LabelLookupFunc) *PodMetricsWatchHelper {
+	return &PodMetricsWatchHelper{storage: storage, labelLookup: labelLookup}
 }
 
 // Watch creates a new watch for pod metrics with the given filters.
 // If sendInitialEvents is true, it sends all current metrics as ADDED events
 // followed by a BOOKMARK before streaming updates.
 func (h *PodMetricsWatchHelper) Watch(ctx context.Context, namespace string, labelSelector labels.Selector, sendInitialEvents bool) (watch.Interface, error) {
-	w := newMetricsWatcher(namespace, labelSelector, resourceTypePod)
+	w := newMetricsWatcher(namespace, labelSelector, resourceTypePod, h.labelLookup)
 
 	var watcherID uint64
 
@@ -287,19 +324,21 @@ func (h *PodMetricsWatchHelper) Watch(ctx context.Context, namespace string, lab
 
 // NodeMetricsWatchHelper helps create watches for node metrics
 type NodeMetricsWatchHelper struct {
-	storage WatchableNodeMetricsGetter
+	storage     WatchableNodeMetricsGetter
+	labelLookup LabelLookupFunc
 }
 
-// NewNodeMetricsWatchHelper creates a new watch helper for node metrics
-func NewNodeMetricsWatchHelper(storage WatchableNodeMetricsGetter) *NodeMetricsWatchHelper {
-	return &NodeMetricsWatchHelper{storage: storage}
+// NewNodeMetricsWatchHelper creates a new watch helper for node metrics.
+// labelLookup provides labels for node metrics objects (used for label selector filtering).
+func NewNodeMetricsWatchHelper(storage WatchableNodeMetricsGetter, labelLookup LabelLookupFunc) *NodeMetricsWatchHelper {
+	return &NodeMetricsWatchHelper{storage: storage, labelLookup: labelLookup}
 }
 
 // Watch creates a new watch for node metrics with the given filters.
 // If sendInitialEvents is true, it sends all current metrics as ADDED events
 // followed by a BOOKMARK before streaming updates.
 func (h *NodeMetricsWatchHelper) Watch(ctx context.Context, labelSelector labels.Selector, sendInitialEvents bool) (watch.Interface, error) {
-	w := newMetricsWatcher("", labelSelector, resourceTypeNode) // Nodes are cluster-scoped
+	w := newMetricsWatcher("", labelSelector, resourceTypeNode, h.labelLookup) // Nodes are cluster-scoped
 
 	var watcherID uint64
 

pkg/api/watcher_test.go

@@ -27,7 +27,7 @@ import (
 )
 
 func TestMetricsWatcher_BasicSendReceive(t *testing.T) {
-	w := newMetricsWatcher("", labels.Everything(), resourceTypePod)
+	w := newMetricsWatcher("", labels.Everything(), resourceTypePod, nil)
 	defer w.Stop()
 
 	// Send an event
@@ -51,7 +51,7 @@ func TestMetricsWatcher_BasicSendReceive(t *testing.T) {
 }
 
 func TestMetricsWatcher_StopClosesChannel(t *testing.T) {
-	w := newMetricsWatcher("", labels.Everything(), resourceTypePod)
+	w := newMetricsWatcher("", labels.Everything(), resourceTypePod, nil)
 
 	// Stop the watcher
 	w.Stop()
@@ -75,7 +75,7 @@ func TestMetricsWatcher_StopClosesChannel(t *testing.T) {
 }
 
 func TestMetricsWatcher_NamespaceFilter(t *testing.T) {
-	w := newMetricsWatcher("test-ns", labels.Everything(), resourceTypePod)
+	w := newMetricsWatcher("test-ns", labels.Everything(), resourceTypePod, nil)
 	defer w.Stop()
 
 	// Event in matching namespace
@@ -113,7 +113,7 @@ func TestMetricsWatcher_NamespaceFilter(t *testing.T) {
 
 func TestMetricsWatcher_LabelSelectorFilter(t *testing.T) {
 	selector, _ := labels.Parse("app=myapp")
-	w := newMetricsWatcher("", selector, resourceTypeNode)
+	w := newMetricsWatcher("", selector, resourceTypeNode, nil)
 	defer w.Stop()
 
 	// Event with matching labels
@@ -150,7 +150,7 @@ func TestMetricsWatcher_LabelSelectorFilter(t *testing.T) {
 }
 
 func TestMetricsWatcher_SendInitialEvents(t *testing.T) {
-	w := newMetricsWatcher("", labels.Everything(), resourceTypePod)
+	w := newMetricsWatcher("", labels.Everything(), resourceTypePod, nil)
 	defer w.Stop()
 
 	// Create some initial objects
@@ -202,7 +202,7 @@ func TestMetricsWatcher_SendInitialEvents(t *testing.T) {
 }
 
 func TestMetricsWatcher_SlowConsumer(t *testing.T) {
-	w := newMetricsWatcher("", labels.Everything(), resourceTypePod)
+	w := newMetricsWatcher("", labels.Everything(), resourceTypePod, nil)
 	defer w.Stop()
 
 	event := WatchEvent{
@@ -239,7 +239,7 @@ func TestMetricsWatcher_ContextCancellation(t *testing.T) {
 		allMetrics: []metrics.PodMetrics{},
 	}
 
-	helper := NewPodMetricsWatchHelper(storage)
+	helper := NewPodMetricsWatchHelper(storage, nil)
 
 	w, err := helper.Watch(ctx, "", labels.Everything(), false)
 	if err != nil {
@@ -281,7 +281,7 @@ func TestPodMetricsWatchHelper_WatchWithInitialEvents(t *testing.T) {
 		allMetrics: []metrics.PodMetrics{pod1, pod2},
 	}
 
-	helper := NewPodMetricsWatchHelper(storage)
+	helper := NewPodMetricsWatchHelper(storage, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -329,7 +329,7 @@ func TestNodeMetricsWatchHelper_WatchWithInitialEvents(t *testing.T) {
 		allMetrics: []metrics.NodeMetrics{node1, node2},
 	}
 
-	helper := NewNodeMetricsWatchHelper(storage)
+	helper := NewNodeMetricsWatchHelper(storage, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -368,7 +368,7 @@ func TestNodeMetricsWatchHelper_WatchWithInitialEvents(t *testing.T) {
 func TestMetricsWatcher_BookmarkType(t *testing.T) {
 	// Test that pod watchers get PodMetrics bookmarks
 	t.Run("pod watcher gets PodMetrics bookmark", func(t *testing.T) {
-		w := newMetricsWatcher("", labels.Everything(), resourceTypePod)
+		w := newMetricsWatcher("", labels.Everything(), resourceTypePod, nil)
 		defer w.Stop()
 
 		// Send bookmark directly - it goes to result channel
@@ -393,7 +393,7 @@ func TestMetricsWatcher_BookmarkType(t *testing.T) {
 
 	// Test that node watchers get NodeMetrics bookmarks
 	t.Run("node watcher gets NodeMetrics bookmark", func(t *testing.T) {
-		w := newMetricsWatcher("", labels.Everything(), resourceTypeNode)
+		w := newMetricsWatcher("", labels.Everything(), resourceTypeNode, nil)
 		defer w.Stop()
 
 		ok := w.sendBookmark("456")
@@ -416,6 +416,58 @@ func TestMetricsWatcher_BookmarkType(t *testing.T) {
 	})
 }
 
+func TestMetricsWatcher_LabelEnrichment(t *testing.T) {
+	// Simulate: storage events have no labels, but labelLookup provides them
+	lookup := func(namespace, name string) map[string]string {
+		if name == "pod1" {
+			return map[string]string{"app": "myapp"}
+		}
+		if name == "pod2" {
+			return map[string]string{"app": "other"}
+		}
+		return nil
+	}
+
+	selector, _ := labels.Parse("app=myapp")
+	w := newMetricsWatcher("", selector, resourceTypePod, lookup)
+	defer w.Stop()
+
+	// Send events WITHOUT labels (as storage does)
+	pod1 := metrics.PodMetrics{}
+	pod1.Name = "pod1"
+	pod1.Namespace = "default"
+	// Note: no Labels set
+
+	pod2 := metrics.PodMetrics{}
+	pod2.Name = "pod2"
+	pod2.Namespace = "default"
+
+	w.Send(WatchEvent{Type: watch.Modified, Object: pod1})
+	w.Send(WatchEvent{Type: watch.Modified, Object: pod2})
+
+	// Should receive pod1 (labels enriched from lookup → matches selector)
+	select {
+	case event := <-w.ResultChan():
+		pm := event.Object.(*metrics.PodMetrics)
+		if pm.Name != "pod1" {
+			t.Errorf("Expected pod1, got %s", pm.Name)
+		}
+		if pm.Labels["app"] != "myapp" {
+			t.Errorf("Expected enriched labels, got %v", pm.Labels)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("Timeout waiting for enriched event")
+	}
+
+	// pod2 should be filtered out (labels enriched → doesn't match selector)
+	select {
+	case event := <-w.ResultChan():
+		t.Errorf("Unexpected event: %v", event)
+	case <-time.After(100 * time.Millisecond):
+		// Expected
+	}
+}
+
 // fakeWatchablePodStorage implements WatchablePodMetricsGetter for testing
 type fakeWatchablePodStorage struct {
 	currentRV  string