From 864357774f24ee17b2bc7bde84eb5b87cc7ab95b Mon Sep 17 00:00:00 2001 From: Richa Banker Date: Thu, 5 Feb 2026 11:42:20 -0800 Subject: [PATCH] Add FIFO queue depth metrics --- .../client-go/tools/cache/fifo_metrics.go | 79 ++++ .../client-go/tools/cache/the_real_fifo.go | 55 ++- .../prometheus/clientgo/fifo/metrics.go | 78 ++++ .../client/metrics/fifo_metrics_test.go | 351 ++++++++++++++++++ 4 files changed, 554 insertions(+), 9 deletions(-) create mode 100644 staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go create mode 100644 staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go create mode 100644 test/integration/client/metrics/fifo_metrics_test.go 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 new file mode 100644 index 00000000000..613dcb7beb9 --- /dev/null +++ b/staging/src/k8s.io/client-go/tools/cache/fifo_metrics.go @@ -0,0 +1,79 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package cache is a client-side caching mechanism. It is useful for +// reducing the number of server calls you'd otherwise need to make. +// Reflector watches a server and updates a Store. Two stores are provided; +// one that simply caches objects (for example, to allow a scheduler to +// list currently available nodes), and one that additionally acts as +// a FIFO queue (for example, to allow a scheduler to process incoming +// pods). +package cache + +import ( + "sync" +) + +var ( + globalFIFOMetricsProvider FIFOMetricsProvider = noopFIFOMetricsProvider{} + setFIFOMetricsProviderOnce sync.Once +) + +type noopFIFOMetricsProvider struct{} + +// FIFOMetricsProvider defines an interface for creating metrics that track FIFO queue operations. +type FIFOMetricsProvider interface { + // NewQueuedItemMetric returns a gauge metric for tracking the total number of items + // currently queued and waiting to be processed. + // The returned metric should check id.Reserved() before updating to support + // dynamic informers that may shut down while the process is still running. + // + // 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 +} + +// fifoMetrics holds all metrics for a FIFO. +type fifoMetrics struct { + numberOfQueuedItem GaugeMetric +} + +// SetFIFOMetricsProvider sets the metrics provider for all subsequently created +// FIFOs. Only the first call has an effect. +func SetFIFOMetricsProvider(metricsProvider FIFOMetricsProvider) { + setFIFOMetricsProviderOnce.Do(func() { + globalFIFOMetricsProvider = metricsProvider + }) +} + +func newFIFOMetrics(id InformerNameAndResource, metricsProvider FIFOMetricsProvider) *fifoMetrics { + if metricsProvider == nil { + metricsProvider = globalFIFOMetricsProvider + } + metrics := &fifoMetrics{ + numberOfQueuedItem: noopMetric{}, + } + + if id.Reserved() { + metrics.numberOfQueuedItem = metricsProvider.NewQueuedItemMetric(id) + } + + return metrics +} + +func (noopFIFOMetricsProvider) NewQueuedItemMetric(InformerNameAndResource) GaugeMetric { + return noopMetric{} +} 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 96d282f2caa..5ac89c94522 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 @@ -56,6 +56,14 @@ type RealFIFOOptions struct { // while processing events to allow other goroutines to add items to the queue. // If UnlockWhileProcessing is true, AtomicEvents must be true as well. UnlockWhileProcessing bool + + // Identifier is used to identify this FIFO for metrics and logging purposes. + // Optional. If zero value, metrics will not be published and trace logs will not + // include Name or Resource fields. + Identifier InformerNameAndResource + + // MetricsProvider is used to create metrics for the FIFO. + MetricsProvider FIFOMetricsProvider } const ( @@ -113,6 +121,12 @@ type RealFIFO struct { // This may only be set if emitAtomicEvents is true. If unlockWhileProcessing is true, // Pop and PopBatch must be called from a single threaded consumer. unlockWhileProcessing bool + + // identifier is used to identify this FIFO for metrics and logging purposes. + identifier InformerNameAndResource + + // metrics holds all metrics for this FIFO. + metrics *fifoMetrics } // ReplacedAllInfo is the object associated with a Delta of type=ReplacedAll @@ -209,6 +223,7 @@ func (f *RealFIFO) addToItems_locked(deltaActionType DeltaType, skipTransform bo Object: obj, }) f.cond.Broadcast() + f.metrics.numberOfQueuedItem.Set(float64(len(f.items))) return nil } @@ -238,6 +253,7 @@ func (f *RealFIFO) addReplaceToItemsLocked(objs []interface{}, resourceVersion s Object: info, }) f.cond.Broadcast() + f.metrics.numberOfQueuedItem.Set(float64(len(f.items))) return nil } @@ -248,6 +264,7 @@ func (f *RealFIFO) addResyncToItemsLocked() error { Object: SyncAllInfo{}, }) f.cond.Broadcast() + f.metrics.numberOfQueuedItem.Set(float64(len(f.items))) return nil } @@ -340,12 +357,21 @@ func (f *RealFIFO) Pop(process PopProcessFunc) (interface{}, error) { // https://github.com/kubernetes/kubernetes/issues/103789 if len(f.items) > 10 { id, _ := f.keyOf(item) - trace := utiltrace.New("RealFIFO Pop Process", - utiltrace.Field{Key: "ID", Value: id}, - utiltrace.Field{Key: "Depth", Value: len(f.items)}, - utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"}) + fields := []utiltrace.Field{ + {Key: "ID", Value: id}, + {Key: "Depth", Value: len(f.items)}, + {Key: "Reason", Value: "slow event handlers blocking the queue"}, + } + if name := f.identifier.Name(); len(name) > 0 { + fields = append(fields, utiltrace.Field{Key: "Name", Value: name}) + } + if gvr := f.identifier.GroupVersionResource(); !gvr.Empty() { + fields = append(fields, utiltrace.Field{Key: "Resource", Value: gvr}) + } + trace := utiltrace.New("RealFIFO Pop Process", fields...) defer trace.LogIfLong(100 * time.Millisecond) } + f.metrics.numberOfQueuedItem.Set(float64(len(f.items))) // Process the item, this may unlock the lock, and allow other goroutines to add items to the queue. err := f.whileProcessing_locked(func() error { @@ -459,13 +485,22 @@ func (f *RealFIFO) PopBatch(processBatch ProcessBatchFunc, processSingle PopProc // https://github.com/kubernetes/kubernetes/issues/103789 if len(f.items) > 10 { id, _ := f.keyOf(deltas[0]) - trace := utiltrace.New("RealFIFO PopBatch Process", - utiltrace.Field{Key: "ID", Value: id}, - utiltrace.Field{Key: "Depth", Value: len(f.items)}, - utiltrace.Field{Key: "Reason", Value: "slow event handlers blocking the queue"}, - utiltrace.Field{Key: "BatchSize", Value: len(deltas)}) + fields := []utiltrace.Field{ + {Key: "ID", Value: id}, + {Key: "Depth", Value: len(f.items)}, + {Key: "Reason", Value: "slow event handlers blocking the queue"}, + {Key: "BatchSize", Value: len(deltas)}, + } + if name := f.identifier.Name(); len(name) > 0 { + fields = append(fields, utiltrace.Field{Key: "Name", Value: name}) + } + if gvr := f.identifier.GroupVersionResource(); !gvr.Empty() { + fields = append(fields, utiltrace.Field{Key: "Resource", Value: gvr}) + } + trace := utiltrace.New("RealFIFO PopBatch Process", fields...) defer trace.LogIfLong(min(100*time.Millisecond*time.Duration(len(deltas)), time.Second)) } + f.metrics.numberOfQueuedItem.Set(float64(len(f.items))) if len(deltas) == 1 { return f.whileProcessing_locked(func() error { @@ -711,6 +746,8 @@ func NewRealFIFOWithOptions(opts RealFIFOOptions) *RealFIFO { batchSize: defaultBatchSize, emitAtomicEvents: opts.AtomicEvents, unlockWhileProcessing: opts.UnlockWhileProcessing, + identifier: opts.Identifier, + metrics: newFIFOMetrics(opts.Identifier, opts.MetricsProvider), } f.cond.L = &f.lock 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 new file mode 100644 index 00000000000..295d8540e84 --- /dev/null +++ b/staging/src/k8s.io/component-base/metrics/prometheus/clientgo/fifo/metrics.go @@ -0,0 +1,78 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package fifo + +import ( + "sync" + + "k8s.io/client-go/tools/cache" + k8smetrics "k8s.io/component-base/metrics" + "k8s.io/component-base/metrics/legacyregistry" +) + +var ( + fifoQueuedItems = k8smetrics.NewGaugeVec( + &k8smetrics.GaugeOpts{ + Subsystem: "informer", + Name: "queued_items", + Help: "Number of items currently queued in the FIFO.", + StabilityLevel: k8smetrics.ALPHA, + }, + []string{"name", "group", "version", "resource"}, + ) + registerOnce sync.Once +) + +func init() { + Register() +} + +// Register registers FIFO metrics and sets the metrics provider. +func Register() { + registerOnce.Do(func() { + legacyregistry.MustRegister(fifoQueuedItems) + }) + cache.SetFIFOMetricsProvider(fifoMetricsProvider{}) +} + +type fifoMetricsProvider struct{} + +func (fifoMetricsProvider) NewQueuedItemMetric(id cache.InformerNameAndResource) cache.GaugeMetric { + return &reservedGaugeMetric{ + id: id, + gauge: fifoQueuedItems.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. +type reservedGaugeMetric struct { + id cache.InformerNameAndResource + gauge cache.GaugeMetric +} + +func (r *reservedGaugeMetric) Set(value float64) { + if r.id.Reserved() { + r.gauge.Set(value) + } +} diff --git a/test/integration/client/metrics/fifo_metrics_test.go b/test/integration/client/metrics/fifo_metrics_test.go new file mode 100644 index 00000000000..510f7b9f967 --- /dev/null +++ b/test/integration/client/metrics/fifo_metrics_test.go @@ -0,0 +1,351 @@ +/* +Copyright The Kubernetes Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package metrics + +import ( + "fmt" + "strings" + "testing" + + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/tools/cache" + "k8s.io/component-base/metrics" + "k8s.io/component-base/metrics/testutil" +) + +var podsGVR = schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"} + +func TestRealFIFO_Metrics(t *testing.T) { + tests := []struct { + name string + actions []func(f *cache.RealFIFO) + expectedMetric int + }{ + { + name: "empty queue has zero metric", + actions: []func(f *cache.RealFIFO){}, + expectedMetric: 0, + }, + { + name: "Add increases metric", + actions: []func(f *cache.RealFIFO){ + func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) }, + }, + expectedMetric: 1, + }, + { + name: "multiple Adds increase metric", + 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)) }, + }, + expectedMetric: 3, + }, + { + name: "Update increases metric", + actions: []func(f *cache.RealFIFO){ + func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) }, + func(f *cache.RealFIFO) { _ = f.Update(mkFifoObj("foo", 2)) }, + }, + expectedMetric: 2, + }, + { + name: "Delete increases metric", + actions: []func(f *cache.RealFIFO){ + func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("foo", 1)) }, + func(f *cache.RealFIFO) { _ = f.Delete(mkFifoObj("foo", 2)) }, + }, + expectedMetric: 2, + }, + { + name: "Pop decreases metric", + 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.Pop(func(obj interface{}, isInInitialList bool) error { return nil }) + }, + }, + expectedMetric: 1, + }, + { + name: "Replace sets metric to new count", + actions: []func(f *cache.RealFIFO){ + func(f *cache.RealFIFO) { _ = f.Add(mkFifoObj("old", 1)) }, + func(f *cache.RealFIFO) { + _ = f.Replace([]interface{}{ + mkFifoObj("foo", 1), + mkFifoObj("bar", 2), + }, "0") + }, + }, + // 1 (Add) + 1 (Delete for "old") + 2 (Replace items) = 4 + expectedMetric: 4, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + metricsProvider := newTestFIFOMetricsProvider() + informerName, err := cache.NewInformerName("test-fifo") + if err != nil { + t.Fatalf("NewInformerName() unexpected error: %v", err) + } + defer informerName.Release() + id := informerName.WithResource(podsGVR) + + f := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id, + MetricsProvider: metricsProvider, + }) + + for _, action := range tt.actions { + 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) + } + }) + } +} + +func TestRealFIFO_MetricsNotPublishedForUnnamedFIFO(t *testing.T) { + metricsProvider := newTestFIFOMetricsProvider() + + // No InformerName configured - should not publish metrics + var id cache.InformerNameAndResource + f := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id, + MetricsProvider: metricsProvider, + }) + + // Perform operations + _ = f.Add(mkFifoObj("foo", 1)) + _ = f.Add(mkFifoObj("bar", 2)) + + // No metrics should be created because there's no identifier configured + want := "" + if err := testutil.GatherAndCompare(metricsProvider.registry, strings.NewReader(want), "informer_queued_items"); err != nil { + t.Fatal(err) + } +} + +func TestRealFIFO_MetricsNotPublishedForDuplicateGVR(t *testing.T) { + metricsProvider := newTestFIFOMetricsProvider() + + // Create InformerName + informerName, err := cache.NewInformerName("duplicate-test") + if err != nil { + t.Fatalf("NewInformerName() unexpected error: %v", err) + } + defer informerName.Release() + // Create first FIFO with a GVR - this should be reserved + id1 := informerName.WithResource(podsGVR) + if !id1.Reserved() { + t.Fatal("Expected first identifier to be reserved") + } + f1 := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id1, + MetricsProvider: metricsProvider, + }) + + // Create second FIFO with the same GVR - this should NOT be reserved + id2 := informerName.WithResource(podsGVR) + if id2.Reserved() { + t.Fatal("Expected second identifier with same GVR to not be reserved") + } + f2 := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id2, + MetricsProvider: metricsProvider, + }) + + // Add items to both FIFOs + _ = f1.Add(mkFifoObj("foo", 1)) + _ = 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) + } +} + +func TestRealFIFO_MetricsTrackedIndependentlyForDifferentFIFOs(t *testing.T) { + metricsProvider := newTestFIFOMetricsProvider() + + // Create two InformerNames with different names - both should be unique + informerName1, err := cache.NewInformerName("fifo-1") + if err != nil { + t.Fatalf("NewInformerName() unexpected error: %v", err) + } + defer informerName1.Release() + + id1 := informerName1.WithResource(podsGVR) + f1 := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id1, + MetricsProvider: metricsProvider, + }) + + informerName2, err := cache.NewInformerName("fifo-2") + if err != nil { + t.Fatalf("NewInformerName() unexpected error: %v", err) + } + defer informerName2.Release() + + id2 := informerName2.WithResource(podsGVR) + f2 := cache.NewRealFIFOWithOptions(cache.RealFIFOOptions{ + KeyFunction: testFifoObjectKeyFunc, + KnownObjects: emptyKnownObjects(), + Identifier: id2, + MetricsProvider: metricsProvider, + }) + + // Add items to f1 + _ = f1.Add(mkFifoObj("foo", 1)) + _ = f1.Add(mkFifoObj("bar", 2)) + + // Add items to f2 + _ = f2.Add(mkFifoObj("baz", 3)) + + // Verify metrics are tracked independently + want := `# 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 { + 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. +# 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 { + t.Fatal(err) + } +} + +type testFifoObject struct { + name string + val interface{} +} + +func testFifoObjectKeyFunc(obj interface{}) (string, error) { + return obj.(testFifoObject).name, nil +} + +func mkFifoObj(name string, val interface{}) testFifoObject { + return testFifoObject{name: name, val: val} +} + +type literalListerGetter func() []testFifoObject + +func (l literalListerGetter) List() []interface{} { + if l == nil { + return nil + } + result := []interface{}{} + for _, item := range l() { + result = append(result, item) + } + return result +} + +func (l literalListerGetter) ListKeys() []string { + if l == nil { + return nil + } + result := []string{} + for _, item := range l() { + result = append(result, item.name) + } + return result +} + +func (l literalListerGetter) Get(key string) (interface{}, bool, error) { + for _, item := range l() { + if item.name == key { + return item, true, nil + } + } + return nil, false, nil +} + +func (l literalListerGetter) GetByKey(key string) (interface{}, bool, error) { + return l.Get(key) +} + +func emptyKnownObjects() cache.KeyListerGetter { + return literalListerGetter( + func() []testFifoObject { + return []testFifoObject{} + }, + ) +} + +// 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 +} + +func newTestFIFOMetricsProvider() *testFIFOMetricsProvider { + registry := metrics.NewKubeRegistry() + gauge := metrics.NewGaugeVec( + &metrics.GaugeOpts{ + Subsystem: "informer", + Name: "queued_items", + Help: "Number of items currently queued in the FIFO.", + StabilityLevel: metrics.ALPHA, + }, + []string{"name", "group", "version", "resource"}, + ) + registry.MustRegister(gauge) + return &testFIFOMetricsProvider{ + registry: registry, + gauge: gauge, + } +} + +func (p *testFIFOMetricsProvider) NewQueuedItemMetric(id cache.InformerNameAndResource) cache.GaugeMetric { + return p.gauge.WithLabelValues(id.Name(), id.GroupVersionResource().Group, id.GroupVersionResource().Version, id.GroupVersionResource().Resource) +}