Add FIFO queue depth metrics

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{}
+}

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

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)
+	}
+}

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)
+}