Merge pull request #125377 from wojtek-t/watchcache_resource_version_metric

Expose resource version metrics from watchcache

staging/src/k8s.io/apiserver/pkg/storage/cacher/metrics/metrics.go

@@ -106,6 +106,17 @@ var (
 		[]string{"resource"},
 	)
 
+	watchCacheResourceVersion = compbasemetrics.NewGaugeVec(
+		&compbasemetrics.GaugeOpts{
+			Namespace:      namespace,
+			Subsystem:      subsystem,
+			Name:           "resource_version",
+			Help:           "Current resource version of watch cache broken by resource type.",
+			StabilityLevel: compbasemetrics.ALPHA,
+		},
+		[]string{"resource"},
+	)
+
 	watchCacheCapacityIncreaseTotal = compbasemetrics.NewCounterVec(
 		&compbasemetrics.CounterOpts{
 			Subsystem:      subsystem,
@@ -171,6 +182,7 @@ func Register() {
 		legacyregistry.MustRegister(EventsReceivedCounter)
 		legacyregistry.MustRegister(EventsCounter)
 		legacyregistry.MustRegister(TerminatedWatchersCounter)
+		legacyregistry.MustRegister(watchCacheResourceVersion)
 		legacyregistry.MustRegister(watchCacheCapacityIncreaseTotal)
 		legacyregistry.MustRegister(watchCacheCapacityDecreaseTotal)
 		legacyregistry.MustRegister(WatchCacheCapacity)
@@ -186,6 +198,11 @@ func RecordListCacheMetrics(resourcePrefix, indexName string, numFetched, numRet
 	listCacheNumReturned.WithLabelValues(resourcePrefix).Add(float64(numReturned))
 }
 
+// RecordResourceVersion sets the current resource version for a given resource type.
+func RecordResourceVersion(resourcePrefix string, resourceVersion uint64) {
+	watchCacheResourceVersion.WithLabelValues(resourcePrefix).Set(float64(resourceVersion))
+}
+
 // RecordsWatchCacheCapacityChange record watchCache capacity resize(increase or decrease) operations.
 func RecordsWatchCacheCapacityChange(objType string, old, new int) {
 	WatchCacheCapacity.WithLabelValues(objType).Set(float64(new))

staging/src/k8s.io/apiserver/pkg/storage/cacher/watch_cache.go

@@ -348,6 +348,7 @@ func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, upd
 	if w.eventHandler != nil {
 		w.eventHandler(wcEvent)
 	}
+	metrics.RecordResourceVersion(w.groupResource.String(), resourceVersion)
 	return nil
 }
 
@@ -430,6 +431,7 @@ func (w *watchCache) UpdateResourceVersion(resourceVersion string) {
 		}
 		w.eventHandler(wcEvent)
 	}
+	metrics.RecordResourceVersion(w.groupResource.String(), rv)
 }
 
 // List returns list of pointers to <storeElement> objects.
@@ -629,7 +631,9 @@ func (w *watchCache) Replace(objs []interface{}, resourceVersion string) error {
 		w.onReplace()
 	}
 	w.cond.Broadcast()
-	klog.V(3).Infof("Replace watchCache (rev: %v) ", resourceVersion)
+
+	metrics.RecordResourceVersion(w.groupResource.String(), version)
+	klog.V(3).Infof("Replaced watchCache (rev: %v) ", resourceVersion)
 	return nil
 }