feat: add monitor for caching client

cmd/manager/main.go

@@ -403,7 +403,8 @@ func main() {
 	// observed. *multicluster.Client satisfies clientcache.Client, providing
 	// both the inner client.Client and informer access for eviction.
 	clientCacheConfig := conf.GetConfigOrDie[clientcache.RootConfig]()
-	cachingClient, err := clientcache.New(multiclusterClient, scheme, clientCacheConfig.ClientCache)
+	clientCacheMonitor := clientcache.NewMonitor("cortex_")
+	cachingClient, err := clientcache.New(multiclusterClient, scheme, clientCacheConfig.ClientCache, clientCacheMonitor)
 	if err != nil {
 		setupLog.Error(err, "unable to create client cache")
 		os.Exit(1)
@@ -419,6 +420,7 @@ func main() {
 	metrics.Registry = monitoring.WrapRegistry(metrics.Registry, metricsConfig)
 	metrics.Registry.MustRegister(&logMetricsMonitor)
 	metrics.Registry.MustRegister(multiclusterMonitor)
+	metrics.Registry.MustRegister(clientCacheMonitor)
 
 	// TODO: Remove me after scheduling pipeline steps don't require DB connections anymore.
 	metrics.Registry.MustRegister(&db.Monitor)

pkg/clientcache/client.go

@@ -152,10 +152,11 @@ const defaultTTL = 2 * time.Minute
 type CachingClient struct {
 	client.Client // inner client, used for delegation
 
-	inner  Client
-	scheme *runtime.Scheme
-	ttl    time.Duration
-	gvks   map[schema.GroupVersionKind]bool
+	inner   Client
+	scheme  *runtime.Scheme
+	ttl     time.Duration
+	gvks    map[schema.GroupVersionKind]bool
+	monitor Monitor
 
 	mu       sync.RWMutex
 	byGVK    map[schema.GroupVersionKind]map[objectKey]*entry
@@ -170,8 +171,8 @@ type CachingClient struct {
 // New builds a CachingClient wrapping inner. informers supplies the informers
 // used for eviction, scheme resolves object GVKs, and conf lists the GVKs to
 // overlay and the TTL. GVK strings are formatted as "<group>/<version>/<Kind>"
-// and are resolved against scheme.
-func New(inner Client, scheme *runtime.Scheme, conf Config) (*CachingClient, error) {
+// and are resolved against scheme. A nil mon disables metric recording.
+func New(inner Client, scheme *runtime.Scheme, conf Config, mon Monitor) (*CachingClient, error) {
 	gvks, err := resolveGVKs(scheme, conf.GVKs)
 	if err != nil {
 		return nil, err
@@ -186,6 +187,7 @@ func New(inner Client, scheme *runtime.Scheme, conf Config) (*CachingClient, err
 		scheme:     scheme,
 		ttl:        ttl,
 		gvks:       gvks,
+		monitor:    mon,
 		byGVK:      make(map[schema.GroupVersionKind]map[objectKey]*entry),
 		indexers:   make(map[schema.GroupVersionKind]map[string]client.IndexerFunc),
 		writeLocks: newKeyedMutex(),
@@ -256,6 +258,7 @@ func (c *CachingClient) upsert(gvk schema.GroupVersionKind, obj client.Object) {
 		deleted:         false,
 		expiresAt:       time.Now().Add(c.ttl),
 	}
+	c.recordSizeLocked(gvk)
 }
 
 // tombstone marks the object as deleted in the overlay so it is filtered out
@@ -271,6 +274,7 @@ func (c *CachingClient) tombstone(gvk schema.GroupVersionKind, obj client.Object
 		deleted:         true,
 		expiresAt:       time.Now().Add(c.ttl),
 	}
+	c.recordSizeLocked(gvk)
 }
 
 // evictIfSeen removes the overlay entry for obj if the informer-observed object
@@ -297,6 +301,7 @@ func (c *CachingClient) evictIfSeen(gvk schema.GroupVersionKind, obj client.Obje
 		return
 	}
 	delete(entries, key)
+	c.recordSizeLocked(gvk)
 }
 
 // getEntry returns the overlay entry for the key, if present.
@@ -315,12 +320,13 @@ func (c *CachingClient) getEntry(gvk schema.GroupVersionKind, key objectKey) (*e
 func (c *CachingClient) cleanupExpired(now time.Time) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	for _, entries := range c.byGVK {
+	for gvk, entries := range c.byGVK {
 		for key, e := range entries {
 			if now.After(e.expiresAt) {
 				delete(entries, key)
 			}
 		}
+		c.recordSizeLocked(gvk)
 	}
 }
 
@@ -342,6 +348,15 @@ func (c *CachingClient) ensureGVK(gvk schema.GroupVersionKind) {
 	}
 }
 
+// recordSizeLocked reports the current overlay size (live + tombstones) for the
+// GVK to the monitor, if one is configured. Callers must hold c.mu.
+func (c *CachingClient) recordSizeLocked(gvk schema.GroupVersionKind) {
+	if c.monitor == nil {
+		return
+	}
+	c.monitor.observe(gvk, len(c.byGVK[gvk]))
+}
+
 // overlayList merges the overlay entries for the GVK into the informer result,
 // deduplicating by objectKey (overlay wins), dropping tombstones, and filtering
 // overlay-only entries against the list options' label and field selectors.
@@ -533,6 +548,7 @@ func (c *CachingClient) DeleteAllOf(ctx context.Context, obj client.Object, opts
 			expiresAt:       time.Now().Add(c.ttl),
 		}
 	}
+	c.recordSizeLocked(gvk)
 	return nil
 }
 

pkg/clientcache/client_test.go

@@ -224,7 +224,7 @@ func reservationConfig() Config {
 
 func newCaching(t *testing.T, inner Client) *CachingClient {
 	t.Helper()
-	c, err := New(inner, testScheme(t), reservationConfig())
+	c, err := New(inner, testScheme(t), reservationConfig(), nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -261,7 +261,7 @@ func waitFor(t *testing.T, cond func() bool) {
 func TestNewUnknownGVKError(t *testing.T) {
 	_, err := New(newTestClient(t), testScheme(t), Config{
 		GVKs: []string{"cortex.cloud/v1alpha1/DoesNotExist"},
-	})
+	}, nil)
 	if err == nil {
 		t.Fatalf("expected error for unknown GVK, got nil")
 	}
@@ -270,7 +270,7 @@ func TestNewUnknownGVKError(t *testing.T) {
 func TestNewDefaultTTL(t *testing.T) {
 	c, err := New(newTestClient(t), testScheme(t), Config{
 		GVKs: []string{"cortex.cloud/v1alpha1/Reservation"},
-	})
+	}, nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -283,7 +283,7 @@ func TestNewExplicitTTL(t *testing.T) {
 	c, err := New(newTestClient(t), testScheme(t), Config{
 		GVKs: []string{"cortex.cloud/v1alpha1/Reservation"},
 		TTL:  metav1.Duration{Duration: 90 * time.Second},
-	})
+	}, nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -548,7 +548,7 @@ func TestFieldMatching(t *testing.T) {
 
 func TestNonCachedGVKPassthrough(t *testing.T) {
 	inner := newTestClient(t)
-	c, err := New(inner, testScheme(t), Config{})
+	c, err := New(inner, testScheme(t), Config{}, nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -754,7 +754,7 @@ func TestGetNotFoundWithNoOverlay(t *testing.T) {
 
 func TestGetNonCachedPropagatesError(t *testing.T) {
 	sentinel := errors.New("get boom")
-	c, err := New(&errClient{Client: newTestClient(t), getErr: sentinel}, testScheme(t), Config{})
+	c, err := New(&errClient{Client: newTestClient(t), getErr: sentinel}, testScheme(t), Config{}, nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}
@@ -802,7 +802,7 @@ func TestStatusCreateDelegates(t *testing.T) {
 func TestStatusUpdateNonCachedNoOverlay(t *testing.T) {
 	r := newReservation("res-sn", "az-1", "")
 	inner := newTestClient(t, r)
-	c, err := New(inner, testScheme(t), Config{})
+	c, err := New(inner, testScheme(t), Config{}, nil)
 	if err != nil {
 		t.Fatalf("New: %v", err)
 	}

pkg/clientcache/monitor.go

@@ -0,0 +1,103 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package clientcache
+
+import (
+	"sync"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+// Monitor is the metrics sink for the CachingClient. It is optional on the
+// client: a nil Monitor causes recording to be skipped entirely. It embeds
+// prometheus.Collector so a concrete implementation can be registered with a
+// Prometheus registry.
+type Monitor interface {
+	prometheus.Collector
+
+	// observe records the current overlay size (live + tombstones) for a GVK.
+	// It is called by the client while holding c.mu so size is consistent.
+	observe(gvk schema.GroupVersionKind, size int)
+}
+
+// monitor is the default Prometheus-backed Monitor implementation.
+//
+// Overlay entries are normally evicted within milliseconds once the informer
+// observes the change, so a plain gauge scraped by Prometheus would almost
+// always read 0 and give no signal about transient spikes. To make abnormal
+// growth visible even when the overlay is empty at scrape time, the monitor
+// tracks a per-GVK high-watermark (the max size seen since the last scrape)
+// which is reset on each Collect, plus a current-size gauge to confirm the
+// overlay drains correctly.
+type monitor struct {
+	mu sync.Mutex
+	// maxSinceScrape is the high-watermark of overlay size per GVK since the
+	// last scrape. It is reset to a fresh map on each Collect.
+	maxSinceScrape map[schema.GroupVersionKind]int
+	// current is the last observed overlay size per GVK. It is a snapshot, not
+	// reset on scrape.
+	current map[schema.GroupVersionKind]int
+
+	maxDesc     *prometheus.Desc
+	currentDesc *prometheus.Desc
+}
+
+// NewMonitor creates a new Prometheus-backed CachingClient monitor. The prefix
+// is prepended to every metric name (e.g. pass "cortex_" to produce
+// "cortex_clientcache_overlay_entries_max").
+func NewMonitor(prefix string) Monitor {
+	return &monitor{
+		maxSinceScrape: make(map[schema.GroupVersionKind]int),
+		current:        make(map[schema.GroupVersionKind]int),
+		maxDesc: prometheus.NewDesc(
+			prefix+"clientcache_overlay_entries_max",
+			"Maximum overlay entries (live + tombstones) per GVK since the last scrape",
+			[]string{"gvk"}, nil,
+		),
+		currentDesc: prometheus.NewDesc(
+			prefix+"clientcache_overlay_entries",
+			"Current overlay entries (live + tombstones) per GVK at scrape time",
+			[]string{"gvk"}, nil,
+		),
+	}
+}
+
+// observe records the current overlay size for a GVK, updating the
+// high-watermark if it grew.
+func (m *monitor) observe(gvk schema.GroupVersionKind, size int) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.current[gvk] = size
+	if size > m.maxSinceScrape[gvk] {
+		m.maxSinceScrape[gvk] = size
+	}
+}
+
+// Describe implements prometheus.Collector.
+func (m *monitor) Describe(ch chan<- *prometheus.Desc) {
+	ch <- m.maxDesc
+	ch <- m.currentDesc
+}
+
+// Collect implements prometheus.Collector. It emits the current size and the
+// high-watermark per GVK, then resets the high-watermark so the next scrape
+// captures a fresh maximum.
+func (m *monitor) Collect(ch chan<- prometheus.Metric) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for gvk, size := range m.current {
+		ch <- prometheus.MustNewConstMetric(
+			m.currentDesc, prometheus.GaugeValue, float64(size), gvk.String(),
+		)
+	}
+	for gvk, max := range m.maxSinceScrape {
+		ch <- prometheus.MustNewConstMetric(
+			m.maxDesc, prometheus.GaugeValue, float64(max), gvk.String(),
+		)
+	}
+	// Reset the high-watermark after emitting so the next scrape window starts
+	// fresh. current is intentionally left as-is (it is a snapshot).
+	m.maxSinceScrape = make(map[schema.GroupVersionKind]int)
+}