Merge pull request #137101 from richabanker/informer-metric-latency

Add processing latency metric for RealFIFO

staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go

@@ -44,11 +44,19 @@ type FIFOMetricsProvider interface {
 	// For DeltaFIFO: Represents len(f.items) - the number of unique keys with pending deltas
 	// For RealFIFO: Represents len(f.items) - the total number of individual delta events queued
 	NewQueuedItemMetric(id InformerNameAndResource) GaugeMetric
+
+	// NewProcessingLatencyMetric returns a histogram metric for tracking the time taken
+	// to process events (execute handlers) after they are popped from the queue.
+	// The latency is measured in seconds.
+	// The returned metric should check id.Reserved() before updating to support
+	// dynamic informers that may shut down while the process is still running.
+	NewProcessingLatencyMetric(id InformerNameAndResource) HistogramMetric
 }
 
 // fifoMetrics holds all metrics for a FIFO.
 type fifoMetrics struct {
 	numberOfQueuedItem GaugeMetric
+	processingLatency  HistogramMetric
 }
 
 // SetFIFOMetricsProvider sets the metrics provider for all subsequently created
@@ -65,10 +73,12 @@ func newFIFOMetrics(id InformerNameAndResource, metricsProvider FIFOMetricsProvi
 	}
 	metrics := &fifoMetrics{
 		numberOfQueuedItem: noopMetric{},
+		processingLatency:  noopMetric{},
 	}
 
 	if id.Reserved() {
 		metrics.numberOfQueuedItem = metricsProvider.NewQueuedItemMetric(id)
+		metrics.processingLatency = metricsProvider.NewProcessingLatencyMetric(id)
 	}
 
 	return metrics
@@ -77,3 +87,7 @@ func newFIFOMetrics(id InformerNameAndResource, metricsProvider FIFOMetricsProvi
 func (noopFIFOMetricsProvider) NewQueuedItemMetric(InformerNameAndResource) GaugeMetric {
 	return noopMetric{}
 }
+
+func (noopFIFOMetricsProvider) NewProcessingLatencyMetric(InformerNameAndResource) HistogramMetric {
+	return noopMetric{}
+}

staging/src/k8s.io/client-go/tools/cache/reflector_metrics.go

@@ -40,6 +40,12 @@ type SummaryMetric interface {
 	Observe(float64)
 }
 
+// HistogramMetric captures individual observations into configurable buckets.
+// It also provides a sum and count of observations.
+type HistogramMetric interface {
+	Observe(float64)
+}
+
 type noopMetric struct{}
 
 func (noopMetric) Inc()            {}

staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go

@@ -448,7 +448,7 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) {
 	return Deltas{item}, err
 }
 
-// whileProcessing_locked calls the `process` function.
+// whileProcessing_locked calls the `process` function and records processing latency.
 // The lock must be held before calling `whileProcessing_locked`, and is held when `whileProcessing_locked` returns.
 // whileProcessing_locked releases the lock during the call to `process` if f.unlockWhileProcessing is true and the f.items queue is not too long.
 func (f *RealFIFO) whileProcessing_locked(process func() error) error {
@@ -459,7 +459,10 @@ func (f *RealFIFO) whileProcessing_locked(process func() error) error {
 		f.lock.Unlock()
 		defer f.lock.Lock()
 	}
-	return process()
+	startTime := time.Now()
+	err := process()
+	f.metrics.processingLatency.Observe(time.Since(startTime).Seconds())
+	return err
 }
 
 // batchable stores the delta types that can be batched

staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go

@@ -25,15 +25,27 @@ import (
 )
 
 var (
+	subsystem = "informer"
+
 	fifoQueuedItems = k8smetrics.NewGaugeVec(
 		&k8smetrics.GaugeOpts{
-			Subsystem:      "informer",
+			Subsystem:      subsystem,
 			Name:           "queued_items",
 			Help:           "Number of items currently queued in the FIFO.",
 			StabilityLevel: k8smetrics.ALPHA,
 		},
 		[]string{"name", "group", "version", "resource"},
 	)
+	fifoProcessingLatency = k8smetrics.NewHistogramVec(
+		&k8smetrics.HistogramOpts{
+			Subsystem:      subsystem,
+			Name:           "processing_latency_seconds",
+			Help:           "Time taken to process events after popping from the queue.",
+			Buckets:        []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+			StabilityLevel: k8smetrics.ALPHA,
+		},
+		[]string{"name", "group", "version", "resource"},
+	)
 	registerOnce sync.Once
 )
 
@@ -45,6 +57,7 @@ func init() {
 func Register() {
 	registerOnce.Do(func() {
 		legacyregistry.MustRegister(fifoQueuedItems)
+		legacyregistry.MustRegister(fifoProcessingLatency)
 	})
 	cache.SetFIFOMetricsProvider(fifoMetricsProvider{})
 }
@@ -63,6 +76,18 @@ func (fifoMetricsProvider) NewQueuedItemMetric(id cache.InformerNameAndResource)
 	}
 }
 
+func (fifoMetricsProvider) NewProcessingLatencyMetric(id cache.InformerNameAndResource) cache.HistogramMetric {
+	return &reservedHistogramMetric{
+		id: id,
+		histogram: fifoProcessingLatency.WithLabelValues(
+			id.Name(),
+			id.GroupVersionResource().Group,
+			id.GroupVersionResource().Version,
+			id.GroupVersionResource().Resource,
+		),
+	}
+}
+
 // reservedGaugeMetric wraps a gauge and only updates it if the identifier
 // is still reserved. This supports dynamic informers (e.g., GC, ResourceQuota)
 // that may shut down while the process is still running.
@@ -76,3 +101,17 @@ func (r *reservedGaugeMetric) Set(value float64) {
 		r.gauge.Set(value)
 	}
 }
+
+// reservedHistogramMetric wraps a histogram and only updates it if the identifier
+// is still reserved. This supports dynamic informers (e.g., GC, ResourceQuota)
+// that may shut down while the process is still running.
+type reservedHistogramMetric struct {
+	id        cache.InformerNameAndResource
+	histogram cache.HistogramMetric
+}
+
+func (r *reservedHistogramMetric) Observe(value float64) {
+	if r.id.Reserved() {
+		r.histogram.Observe(value)
+	}
+}

test/integration/client/metrics/fifo_metrics_test.go

@@ -31,21 +31,22 @@ var podsGVR = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "p
 
 func TestRealFIFO_Metrics(t *testing.T) {
 	tests := []struct {
-		name           string
-		actions        []func(f *cache.RealFIFO)
-		expectedMetric int
+		name                        string
+		actions                     []func(f *cache.RealFIFO)
+		expectedQueuedItems         int
+		expectedLatencyObservations uint64
 	}{
 		{
-			name:           "empty queue has zero metric",
-			actions:        []func(f *cache.RealFIFO){},
-			expectedMetric: 0,
+			name:                "empty queue has zero metric",
+			actions:             []func(f *cache.RealFIFO){},
+			expectedQueuedItems: 0,
 		},
 		{
 			name: "Add increases metric",
 			actions: []func(f *cache.RealFIFO){
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) },
 			},
-			expectedMetric: 1,
+			expectedQueuedItems: 1,
 		},
 		{
 			name: "multiple Adds increase metric",
@@ -54,7 +55,7 @@ func TestRealFIFO_Metrics(t *testing.T) {
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("bar", 2)) },
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("baz", 3)) },
 			},
-			expectedMetric: 3,
+			expectedQueuedItems: 3,
 		},
 		{
 			name: "Update increases metric",
@@ -62,7 +63,7 @@ func TestRealFIFO_Metrics(t *testing.T) {
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) },
 				func(f *cache.RealFIFO) { _ = f.Update(mkFifoObj("foo", 2)) },
 			},
-			expectedMetric: 2,
+			expectedQueuedItems: 2,
 		},
 		{
 			name: "Delete increases metric",
@@ -70,10 +71,10 @@ func TestRealFIFO_Metrics(t *testing.T) {
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) },
 				func(f *cache.RealFIFO) { _ = f.Delete(mkFifoObj("foo", 2)) },
 			},
-			expectedMetric: 2,
+			expectedQueuedItems: 2,
 		},
 		{
-			name: "Pop decreases metric",
+			name: "Pop decreases metric and records latency",
 			actions: []func(f *cache.RealFIFO){
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) },
 				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("bar", 2)) },
@@ -81,7 +82,24 @@ func TestRealFIFO_Metrics(t *testing.T) {
 					_, _ = f.Pop(func(obj interface{}, isInInitialList bool) error { return nil })
 				},
 			},
-			expectedMetric: 1,
+			expectedQueuedItems:         1,
+			expectedLatencyObservations: 1,
+		},
+		{
+			name: "PopBatch decreases metric and records latency",
+			actions: []func(f *cache.RealFIFO){
+				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) },
+				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("bar", 2)) },
+				func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("baz", 3)) },
+				func(f *cache.RealFIFO) {
+					_ = f.PopBatch(
+						func(deltas []cache.Delta, isInInitialList bool) error { return nil },
+						func(obj interface{}, isInInitialList bool) error { return nil },
+					)
+				},
+			},
+			expectedQueuedItems:         0,
+			expectedLatencyObservations: 1,
 		},
 		{
 			name: "Replace sets metric to new count",
@@ -95,7 +113,7 @@ func TestRealFIFO_Metrics(t *testing.T) {
 				},
 			},
 			// 1 (Add) + 1 (Delete for "old") + 2 (Replace items) = 4
-			expectedMetric: 4,
+			expectedQueuedItems: 4,
 		},
 	}
 
@@ -120,13 +138,11 @@ func TestRealFIFO_Metrics(t *testing.T) {
 				action(f)
 			}
 
-			want := fmt.Sprintf(`# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
-# TYPE informer_queued_items gauge
-informer_queued_items{group="",name="test-fifo",resource="pods",version="v1"} %d
-`, tt.expectedMetric)
-			if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(want), "informer_queued_items"); err != nil {
-				t.Fatal(err)
-			}
+			// Verify queued items metric
+			verifyQueuedItems(t, metricsProvider, "test-fifo", podsGVR, tt.expectedQueuedItems)
+
+			// Verify processing latency observations
+			verifyLatencyObservations(t, metricsProvider, "test-fifo", podsGVR, tt.expectedLatencyObservations)
 		})
 	}
 }
@@ -192,13 +208,7 @@ func TestRealFIFO_MetricsNotPublishedForDuplicateGVR(t *testing.T) {
 	_ = f2.Add(mkFifoObj("bar", 2))
 
 	// Only f1's metric should be published, f2 uses noopMetric
-	want := `# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
-# TYPE informer_queued_items gauge
-informer_queued_items{group="",name="duplicate-test",resource="pods",version="v1"} 1
-`
-	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(want), "informer_queued_items"); err != nil {
-		t.Fatal(err)
-	}
+	verifyQueuedItems(t, metricsProvider, "duplicate-test", podsGVR, 1)
 }
 
 func TestRealFIFO_MetricsTrackedIndependentlyForDifferentFIFOs(t *testing.T) {
@@ -241,24 +251,35 @@ func TestRealFIFO_MetricsTrackedIndependentlyForDifferentFIFOs(t *testing.T) {
 	_ = f2.Add(mkFifoObj("baz", 3))
 
 	// Verify metrics are tracked independently
-	want := `# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
+	wantInformerQueuedItemsMetric := `# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
 # TYPE informer_queued_items gauge
 informer_queued_items{group="",name="fifo-1",resource="pods",version="v1"} 2
 informer_queued_items{group="",name="fifo-2",resource="pods",version="v1"} 1
 `
-	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(want), "informer_queued_items"); err != nil {
+	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(wantInformerQueuedItemsMetric), "informer_queued_items"); err != nil {
 		t.Fatal(err)
 	}
 
 	// Pop from f1 and verify its metric decreases while f2's stays the same
 	_, _ = f1.Pop(func(obj interface{}, isInInitialList bool) error { return nil })
 
-	wantAfterPop := `# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
+	wantInformerQueuedItemsMetricAfterPop := `# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
 # TYPE informer_queued_items gauge
 informer_queued_items{group="",name="fifo-1",resource="pods",version="v1"} 1
 informer_queued_items{group="",name="fifo-2",resource="pods",version="v1"} 1
 `
-	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(wantAfterPop), "informer_queued_items"); err != nil {
+	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(wantInformerQueuedItemsMetricAfterPop), "informer_queued_items"); err != nil {
+		t.Fatal(err)
+	}
+
+	// Verify that processing latency was recorded for f1 only after the Pop.
+	// fifo-2 has count=0 because no Pop was called for it.
+	wantLatencyMetricAfterPop := `# HELP informer_processing_latency_seconds [ALPHA] Time taken to process events after popping from the queue.
+# TYPE informer_processing_latency_seconds histogram
+informer_processing_latency_seconds_count{group="",name="fifo-1",resource="pods",version="v1"} 1
+informer_processing_latency_seconds_count{group="",name="fifo-2",resource="pods",version="v1"} 0
+`
+	if err := testutil.GatherAndCompare(metricsProvider.gatherWithoutDurations(), strings.NewReader(wantLatencyMetricAfterPop), "informer_processing_latency_seconds"); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -324,8 +345,9 @@ func emptyKnownObjects() cache.KeyListerGetter {
 // testFIFOMetricsProvider is a test implementation of cache.FIFOMetricsProvider
 // that uses real component-base metrics registered with a custom registry.
 type testFIFOMetricsProvider struct {
-	registry metrics.KubeRegistry
-	gauge    *metrics.GaugeVec
+	registry  metrics.KubeRegistry
+	gauge     *metrics.GaugeVec
+	histogram *metrics.HistogramVec
 }
 
 func newTestFIFOMetricsProvider() *testFIFOMetricsProvider {
@@ -339,13 +361,75 @@ func newTestFIFOMetricsProvider() *testFIFOMetricsProvider {
 		},
 		[]string{"name", "group", "version", "resource"},
 	)
+	histogram := metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      "informer",
+			Name:           "processing_latency_seconds",
+			Help:           "Time taken to process events after popping from the queue.",
+			Buckets:        []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10},
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"name", "group", "version", "resource"},
+	)
 	registry.MustRegister(gauge)
+	registry.MustRegister(histogram)
 	return &testFIFOMetricsProvider{
-		registry: registry,
-		gauge:    gauge,
+		registry:  registry,
+		gauge:     gauge,
+		histogram: histogram,
 	}
 }
 
 func (p *testFIFOMetricsProvider) NewQueuedItemMetric(id cache.InformerNameAndResource) cache.GaugeMetric {
 	return p.gauge.WithLabelValues(id.Name(), id.GroupVersionResource().Group, id.GroupVersionResource().Version, id.GroupVersionResource().Resource)
 }
+
+func (p *testFIFOMetricsProvider) NewProcessingLatencyMetric(id cache.InformerNameAndResource) cache.HistogramMetric {
+	return p.histogram.WithLabelValues(id.Name(), id.GroupVersionResource().Group, id.GroupVersionResource().Version, id.GroupVersionResource().Resource)
+}
+
+func verifyQueuedItems(t *testing.T, metricsProvider *testFIFOMetricsProvider, informerName string, gvr schema.GroupVersionResource, expected int) {
+	t.Helper()
+	want := fmt.Sprintf(`# HELP informer_queued_items [ALPHA] Number of items currently queued in the FIFO.
+# TYPE informer_queued_items gauge
+informer_queued_items{group="%s",name="%s",resource="%s",version="%s"} %d
+`, gvr.Group, informerName, gvr.Resource, gvr.Version, expected)
+	if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(want), "informer_queued_items"); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// verifyLatencyObservations checks the histogram observation count using a custom gatherer
+// that strips timing-dependent values (bucket counts and sum) from the comparison, so that
+// we only verify the number of observations without being affected by the duration values.
+func verifyLatencyObservations(t *testing.T, metricsProvider *testFIFOMetricsProvider, informerName string, gvr schema.GroupVersionResource, expected uint64) {
+	t.Helper()
+	if expected == 0 {
+		return
+	}
+
+	want := fmt.Sprintf(`# HELP informer_processing_latency_seconds [ALPHA] Time taken to process events after popping from the queue.
+# TYPE informer_processing_latency_seconds histogram
+informer_processing_latency_seconds_count{group="%s",name="%s",resource="%s",version="%s"} %d
+`, gvr.Group, informerName, gvr.Resource, gvr.Version, expected)
+	if err := testutil.GatherAndCompare(metricsProvider.gatherWithoutDurations(), strings.NewReader(want), "informer_processing_latency_seconds"); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (p *testFIFOMetricsProvider) gatherWithoutDurations() testutil.GathererFunc {
+	return func() ([]*testutil.MetricFamily, error) {
+		got, err := p.registry.Gather()
+		for _, mf := range got {
+			for _, m := range mf.Metric {
+				if m.Histogram == nil {
+					continue
+				}
+				// Remove everything from a histogram that depends on timing.
+				m.Histogram.SampleSum = nil
+				m.Histogram.Bucket = nil
+			}
+		}
+		return got, err
+	}
+}