From d049bd123391f4c1a86e493888560d8549a52dc2 Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Mon, 16 Feb 2026 15:15:47 -0800 Subject: [PATCH] Add processing latency metric for RealFIFO --- .../client-go/tools/cache/fifo_metrics.go | 14 ++ .../tools/cache/reflector_metrics.go | 6 + .../client-go/tools/cache/the_real_fifo.go | 7 +- .../prometheus/clientgo/fifo/metrics.go | 41 ++++- .../client/metrics/fifo_metrics_test.go | 154 ++++++++++++++---- 5 files changed, 184 insertions(+), 38 deletions(-) diff --git a/staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go b/staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go index 613dcb7beb9..9e8ee069b2a 100644 --- a/staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go +++ b/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{} +} diff --git a/staging/src/k8s.io/client-go/tools/cache/reflector_metrics.go b/staging/src/k8s.io/client-go/tools/cache/reflector_metrics.go index 5c00115f59b..65e8a38afbe 100644 --- a/staging/src/k8s.io/client-go/tools/cache/reflector_metrics.go +++ b/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() {} diff --git a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go b/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go index c07c4eef8c4..bd6a3791fe8 100644 --- a/staging/src/k8s.io/client-go/tools/cache/the_real_fifo.go +++ b/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 diff --git a/staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go b/staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go index 295d8540e84..9eb04b30f4a 100644 --- a/staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go +++ b/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) + } +} diff --git a/test/integration/client/metrics/fifo_metrics_test.go b/test/integration/client/metrics/fifo_metrics_test.go index 510f7b9f967..2bf76e1b5ef 100644 --- a/test/integration/client/metrics/fifo_metrics_test.go +++ b/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 + } +}