Add metrics for endpoint and endpoint slice state

pkg/neg/manager.go

@@ -236,6 +236,7 @@ func (manager *syncerManager) EnsureSyncers(namespace, name string, newPorts neg
 				portInfo.EpCalculatorMode,
 				manager.logger.WithValues("service", klog.KRef(syncerKey.Namespace, syncerKey.Name), "negName", syncerKey.NegName),
 				manager.enableDualStackNEG,
+				manager.syncerMetrics,
 			)
 			syncer = negsyncer.NewTransactionSyncer(
 				syncerKey,

pkg/neg/metrics/endpoint_metrics.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2023 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 (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	endpointState      = "endpoint_state"
+	endpointSliceState = "endpoint_slice_state"
+
+	endpointStateKey      = "neg_sync_endpoint_state"
+	endpointSliceStateKey = "neg_sync_endpoint_slice_state"
+)
+
+var (
+	// syncerEndpointState tracks the count of endpoints in different states
+	syncerEndpointState = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: negControllerSubsystem,
+			Name:      endpointStateKey,
+			Help:      "Current count of endpoints in each state",
+		},
+		[]string{endpointState},
+	)
+
+	// syncerEndpointSliceState tracks the count of endpoint slices in different states
+	syncerEndpointSliceState = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Subsystem: negControllerSubsystem,
+			Name:      endpointSliceStateKey,
+			Help:      "Current count of endpoint slices in each state",
+		},
+		[]string{endpointSliceState},
+	)
+)

pkg/neg/metrics/neg_metrics_collector.go

@@ -29,18 +29,18 @@ import (
 type SyncerMetricsCollector interface {
 	// UpdateSyncerStatusInMetrics update the status of corresponding syncer based on the sync error
 	UpdateSyncerStatusInMetrics(key negtypes.NegSyncerKey, err error)
-
-	SetSyncerEPMetrics(key negtypes.NegSyncerKey, epState *negtypes.SyncerEPStat)
+	// UpdateSyncerEPMetrics update the endpoint and endpointSlice count for the given syncer
+	UpdateSyncerEPMetrics(key negtypes.NegSyncerKey, endpointCount, endpointSliceCount negtypes.StateCountMap)
 	SetLabelPropagationStats(key negtypes.NegSyncerKey, labelstatLabelPropagationStats LabelPropagationStats)
 }
 
 type SyncerMetrics struct {
 	// syncerStatusMap tracks the status of each syncer
 	syncerStatusMap map[negtypes.NegSyncerKey]negtypes.Reason
-	// syncerEndpointStateMap is a map between syncer and endpoint state counts
+	// syncerEndpointStateMap is a map between syncer and endpoint state counts.
 	syncerEndpointStateMap map[negtypes.NegSyncerKey]negtypes.StateCountMap
-	// syncerEPSStateMap is a map between syncer and endpoint slice state counts
-	syncerEPSStateMap map[negtypes.NegSyncerKey]negtypes.StateCountMap
+	// syncerEndpointSliceStateMap is a map between syncer and endpoint slice state counts.
+	syncerEndpointSliceStateMap map[negtypes.NegSyncerKey]negtypes.StateCountMap
 	// syncerLabelProagationStats is a map between syncer and label propagation stats.
 	syncerLabelProagationStats map[negtypes.NegSyncerKey]LabelPropagationStats
 	// mu avoid race conditions and ensure correctness of metrics
@@ -54,12 +54,12 @@ type SyncerMetrics struct {
 // NewNEGMetricsCollector initializes SyncerMetrics and starts a go routine to compute and export metrics periodically.
 func NewNegMetricsCollector(exportInterval time.Duration, logger klog.Logger) *SyncerMetrics {
 	return &SyncerMetrics{
-		syncerStatusMap:            make(map[negtypes.NegSyncerKey]negtypes.Reason),
-		syncerEndpointStateMap:     make(map[negtypes.NegSyncerKey]negtypes.StateCountMap),
-		syncerEPSStateMap:          make(map[negtypes.NegSyncerKey]negtypes.StateCountMap),
-		syncerLabelProagationStats: make(map[negtypes.NegSyncerKey]LabelPropagationStats),
-		metricsInterval:            exportInterval,
-		logger:                     logger.WithName("NegMetricsCollector"),
+		syncerStatusMap:             make(map[negtypes.NegSyncerKey]negtypes.Reason),
+		syncerEndpointStateMap:      make(map[negtypes.NegSyncerKey]negtypes.StateCountMap),
+		syncerEndpointSliceStateMap: make(map[negtypes.NegSyncerKey]negtypes.StateCountMap),
+		syncerLabelProagationStats:  make(map[negtypes.NegSyncerKey]LabelPropagationStats),
+		metricsInterval:             exportInterval,
+		logger:                      logger.WithName("NegMetricsCollector"),
 	}
 }
 
@@ -72,6 +72,8 @@ func FakeSyncerMetrics() *SyncerMetrics {
 func RegisterSyncerMetrics() {
 	prometheus.MustRegister(syncerSyncResult)
 	prometheus.MustRegister(syncerSyncerState)
+	prometheus.MustRegister(syncerEndpointState)
+	prometheus.MustRegister(syncerEndpointSliceState)
 }
 
 func (sm *SyncerMetrics) Run(stopCh <-chan struct{}) {
@@ -93,6 +95,14 @@ func (sm *SyncerMetrics) export() {
 	stateCount, syncerCount := sm.computeSyncerStateMetrics()
 	PublishSyncerStateMetrics(stateCount)
 
+	epStateCount, epsStateCount := sm.computeEndpointStateMetrics(false)
+	for state, count := range epStateCount {
+		syncerEndpointState.WithLabelValues(string(state)).Set(float64(count))
+	}
+	for state, count := range epsStateCount {
+		syncerEndpointSliceState.WithLabelValues(string(state)).Set(float64(count))
+	}
+
 	sm.logger.V(3).Info("Exporting syncer related metrics", "Syncer count", syncerCount, "Number of Endpoints", lpMetrics.NumberOfEndpoints)
 }
 
@@ -113,21 +123,21 @@ func (sm *SyncerMetrics) UpdateSyncerStatusInMetrics(key negtypes.NegSyncerKey,
 	sm.syncerStatusMap[key] = reason
 }
 
-// SetSyncerEPMetrics update the endpoint count based on the endpointStat
-func (sm *SyncerMetrics) SetSyncerEPMetrics(key negtypes.NegSyncerKey, endpointStat *negtypes.SyncerEPStat) {
+func (sm *SyncerMetrics) UpdateSyncerEPMetrics(key negtypes.NegSyncerKey, endpointCount, endpointSliceCount negtypes.StateCountMap) {
+	sm.logger.V(3).Info("Updating syncer endpoint", "syncerKey", key)
 	sm.mu.Lock()
 	defer sm.mu.Unlock()
 	if sm.syncerEndpointStateMap == nil {
 		sm.syncerEndpointStateMap = make(map[negtypes.NegSyncerKey]negtypes.StateCountMap)
-		sm.logger.V(3).Info("Syncer Metrics failed to initialize correctly, reinitializing syncerEPStateMap")
+		sm.logger.V(3).Info("Syncer Metrics failed to initialize correctly, reinitializing syncerEndpointStateMap")
 	}
-	sm.syncerEndpointStateMap[key] = endpointStat.EndpointStateCount
+	sm.syncerEndpointStateMap[key] = endpointCount
 
-	if sm.syncerEPSStateMap == nil {
-		sm.syncerEPSStateMap = make(map[negtypes.NegSyncerKey]negtypes.StateCountMap)
-		sm.logger.V(3).Info("Syncer Metrics failed to initialize correctly, reinitializing syncerEPSStateMap")
+	if sm.syncerEndpointSliceStateMap == nil {
+		sm.syncerEndpointSliceStateMap = make(map[negtypes.NegSyncerKey]negtypes.StateCountMap)
+		sm.logger.V(3).Info("Syncer Metrics failed to initialize correctly, reinitializing syncerEndpointSliceStateMap")
 	}
-	sm.syncerEPSStateMap[key] = endpointStat.EndpointSliceStateCount
+	sm.syncerEndpointSliceStateMap[key] = endpointSliceCount
 }
 
 func (sm *SyncerMetrics) SetLabelPropagationStats(key negtypes.NegSyncerKey, labelstatLabelPropagationStats LabelPropagationStats) {
@@ -145,7 +155,7 @@ func (sm *SyncerMetrics) DeleteSyncer(key negtypes.NegSyncerKey) {
 	defer sm.mu.Unlock()
 	delete(sm.syncerStatusMap, key)
 	delete(sm.syncerEndpointStateMap, key)
-	delete(sm.syncerEPSStateMap, key)
+	delete(sm.syncerEndpointSliceStateMap, key)
 	delete(sm.syncerLabelProagationStats, key)
 }
 
@@ -174,3 +184,24 @@ func (sm *SyncerMetrics) computeSyncerStateMetrics() (*syncerStateCount, int) {
 	}
 	return stateCount, syncerCount
 }
+
+// computeSyncerEndpointStateMetrics aggregates endpoint and endpoint slice counts from all syncers
+func (sm *SyncerMetrics) computeEndpointStateMetrics(forDegradedMode bool) (negtypes.StateCountMap, negtypes.StateCountMap) {
+	sm.mu.Lock()
+	defer sm.mu.Unlock()
+
+	epCounts := negtypes.StateCountMap{}
+	epsCounts := negtypes.StateCountMap{}
+	// collect count from each syncer
+	for _, epCount := range sm.syncerEndpointStateMap {
+		for _, state := range negtypes.StatesForEndpointMetrics() {
+			epCounts[state] += epCount[state]
+		}
+	}
+	for _, epsCount := range sm.syncerEndpointSliceStateMap {
+		for _, state := range negtypes.StatesForEndpointMetrics() {
+			epsCounts[state] += epsCount[state]
+		}
+	}
+	return epCounts, epsCounts
+}

pkg/neg/syncers/endpoints_calculator.go

@@ -24,7 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/ingress-gce/pkg/neg/metrics"
 	"k8s.io/ingress-gce/pkg/neg/types"
+	negtypes "k8s.io/ingress-gce/pkg/neg/types"
 	"k8s.io/ingress-gce/pkg/utils"
 	"k8s.io/klog/v2"
 )
@@ -185,26 +187,30 @@ func (l *ClusterL4ILBEndpointsCalculator) ValidateEndpoints(endpointData []types
 
 // L7EndpointsCalculator implements methods to calculate Network endpoints for VM_IP_PORT NEGs
 type L7EndpointsCalculator struct {
-	zoneGetter          types.ZoneGetter
-	servicePortName     string
-	podLister           cache.Indexer
-	nodeLister          cache.Indexer
-	serviceLister       cache.Indexer
-	networkEndpointType types.NetworkEndpointType
-	enableDualStackNEG  bool
-	logger              klog.Logger
+	zoneGetter           types.ZoneGetter
+	servicePortName      string
+	podLister            cache.Indexer
+	nodeLister           cache.Indexer
+	serviceLister        cache.Indexer
+	syncerKey            types.NegSyncerKey
+	networkEndpointType  types.NetworkEndpointType
+	enableDualStackNEG   bool
+	logger               klog.Logger
+	syncMetricsCollector *metrics.SyncerMetrics
 }
 
-func NewL7EndpointsCalculator(zoneGetter types.ZoneGetter, podLister, nodeLister, serviceLister cache.Indexer, svcPortName string, endpointType types.NetworkEndpointType, logger klog.Logger, enableDualStackNEG bool) *L7EndpointsCalculator {
+func NewL7EndpointsCalculator(zoneGetter types.ZoneGetter, podLister, nodeLister, serviceLister cache.Indexer, syncerKey types.NegSyncerKey, logger klog.Logger, enableDualStackNEG bool, syncMetricsCollector *metrics.SyncerMetrics) *L7EndpointsCalculator {
 	return &L7EndpointsCalculator{
-		zoneGetter:          zoneGetter,
-		servicePortName:     svcPortName,
-		podLister:           podLister,
-		nodeLister:          nodeLister,
-		serviceLister:       serviceLister,
-		networkEndpointType: endpointType,
-		enableDualStackNEG:  enableDualStackNEG,
-		logger:              logger.WithName("L7EndpointsCalculator"),
+		zoneGetter:           zoneGetter,
+		servicePortName:      syncerKey.PortTuple.Name,
+		podLister:            podLister,
+		nodeLister:           nodeLister,
+		serviceLister:        serviceLister,
+		syncerKey:            syncerKey,
+		networkEndpointType:  syncerKey.NegType,
+		enableDualStackNEG:   enableDualStackNEG,
+		logger:               logger.WithName("L7EndpointsCalculator"),
+		syncMetricsCollector: syncMetricsCollector,
 	}
 }
 
@@ -216,12 +222,16 @@ func (l *L7EndpointsCalculator) Mode() types.EndpointsCalculatorMode {
 // CalculateEndpoints determines the endpoints in the NEGs based on the current service endpoints and the current NEGs.
 func (l *L7EndpointsCalculator) CalculateEndpoints(eds []types.EndpointsData, _ map[string]types.NetworkEndpointSet) (map[string]types.NetworkEndpointSet, types.EndpointPodMap, int, error) {
 	result, err := toZoneNetworkEndpointMap(eds, l.zoneGetter, l.podLister, l.servicePortName, l.networkEndpointType, l.enableDualStackNEG)
-	return result.NetworkEndpointSet, result.EndpointPodMap, result.DupCount, err
+	if err != nil { // If current calculation ends up in error, we trigger and emit metrics in degraded mode.
+		l.syncMetricsCollector.UpdateSyncerEPMetrics(l.syncerKey, result.EPCount, result.EPSCount)
+	}
+	return result.NetworkEndpointSet, result.EndpointPodMap, result.EPCount[negtypes.Duplicate], err
 }
 
 // CalculateEndpoints determines the endpoints in the NEGs based on the current service endpoints and the current NEGs.
 func (l *L7EndpointsCalculator) CalculateEndpointsDegradedMode(eds []types.EndpointsData, _ map[string]types.NetworkEndpointSet) (map[string]types.NetworkEndpointSet, types.EndpointPodMap, error) {
 	result := toZoneNetworkEndpointMapDegradedMode(eds, l.zoneGetter, l.podLister, l.nodeLister, l.serviceLister, l.servicePortName, l.networkEndpointType, l.enableDualStackNEG)
+	l.syncMetricsCollector.UpdateSyncerEPMetrics(l.syncerKey, result.EPCount, result.EPSCount)
 	return result.NetworkEndpointSet, result.EndpointPodMap, nil
 }
 

pkg/neg/syncers/endpoints_calculator_test.go

@@ -28,6 +28,7 @@ import (
 	listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/cloud-provider-gcp/providers/gce"
+	"k8s.io/ingress-gce/pkg/neg/metrics"
 	negtypes "k8s.io/ingress-gce/pkg/neg/types"
 	"k8s.io/ingress-gce/pkg/utils"
 	"k8s.io/klog/v2"
@@ -228,13 +229,24 @@ func TestValidateEndpoints(t *testing.T) {
 	testPodName3 := "pod3"
 	testPodName4 := "pod4"
 	svcKey := fmt.Sprintf("%s/%s", testServiceName, testServiceNamespace)
+	svcPort := negtypes.NegSyncerKey{
+		Namespace: testServiceNamespace,
+		Name:      testServiceName,
+		NegType:   negtypes.VmIpPortEndpointType,
+		PortTuple: negtypes.SvcPortTuple{
+			Port:       80,
+			TargetPort: "8080",
+			Name:       testPortName,
+		},
+		NegName: testNegName,
+	}
 
 	zoneGetter := negtypes.NewFakeZoneGetter()
 	testContext := negtypes.NewTestContext()
 	podLister := testContext.PodInformer.GetIndexer()
 	nodeLister := testContext.NodeInformer.GetIndexer()
 	serviceLister := testContext.ServiceInformer.GetIndexer()
-	L7EndpointsCalculator := NewL7EndpointsCalculator(zoneGetter, podLister, nodeLister, serviceLister, testPortName, negtypes.VmIpPortEndpointType, klog.TODO(), testContext.EnableDualStackNEG)
+	L7EndpointsCalculator := NewL7EndpointsCalculator(zoneGetter, podLister, nodeLister, serviceLister, svcPort, klog.TODO(), testContext.EnableDualStackNEG, metrics.FakeSyncerMetrics())
 	L4LocalEndpointCalculator := NewLocalL4ILBEndpointsCalculator(listers.NewNodeLister(nodeLister), zoneGetter, svcKey, klog.TODO())
 	L4ClusterEndpointCalculator := NewClusterL4ILBEndpointsCalculator(listers.NewNodeLister(nodeLister), zoneGetter, svcKey, klog.TODO())
 

pkg/neg/syncers/transaction.go

@@ -183,7 +183,7 @@ func NewTransactionSyncer(
 	return syncer
 }
 
-func GetEndpointsCalculator(podLister, nodeLister, serviceLister cache.Indexer, zoneGetter negtypes.ZoneGetter, syncerKey negtypes.NegSyncerKey, mode negtypes.EndpointsCalculatorMode, logger klog.Logger, enableDualStackNEG bool) negtypes.NetworkEndpointsCalculator {
+func GetEndpointsCalculator(podLister, nodeLister, serviceLister cache.Indexer, zoneGetter negtypes.ZoneGetter, syncerKey negtypes.NegSyncerKey, mode negtypes.EndpointsCalculatorMode, logger klog.Logger, enableDualStackNEG bool, syncMetricsCollector *metrics.SyncerMetrics) negtypes.NetworkEndpointsCalculator {
 	serviceKey := strings.Join([]string{syncerKey.Name, syncerKey.Namespace}, "/")
 	if syncerKey.NegType == negtypes.VmIpEndpointType {
 		nodeLister := listers.NewNodeLister(nodeLister)
@@ -199,10 +199,10 @@ func GetEndpointsCalculator(podLister, nodeLister, serviceLister cache.Indexer,
 		podLister,
 		nodeLister,
 		serviceLister,
-		syncerKey.PortTuple.Name,
-		syncerKey.NegType,
+		syncerKey,
 		logger,
 		enableDualStackNEG,
+		syncMetricsCollector,
 	)
 }
 

pkg/neg/syncers/transaction_test.go

@@ -2092,7 +2092,7 @@ func TestCollectLabelStats(t *testing.T) {
 
 func newL4ILBTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, mode negtypes.EndpointsCalculatorMode) (negtypes.NegSyncer, *transactionSyncer) {
 	negsyncer, ts := newTestTransactionSyncer(fakeGCE, negtypes.VmIpEndpointType, false)
-	ts.endpointsCalculator = GetEndpointsCalculator(ts.podLister, ts.nodeLister, ts.serviceLister, ts.zoneGetter, ts.NegSyncerKey, mode, klog.TODO(), false)
+	ts.endpointsCalculator = GetEndpointsCalculator(ts.podLister, ts.nodeLister, ts.serviceLister, ts.zoneGetter, ts.NegSyncerKey, mode, klog.TODO(), false, nil)
 	return negsyncer, ts
 }
 
@@ -2134,7 +2134,7 @@ func newTestTransactionSyncer(fakeGCE negtypes.NetworkEndpointGroupCloud, negTyp
 		testContext.SvcNegInformer.GetIndexer(),
 		reflector,
 		GetEndpointsCalculator(testContext.PodInformer.GetIndexer(), testContext.NodeInformer.GetIndexer(), testContext.ServiceInformer.GetIndexer(),
-			fakeZoneGetter, svcPort, mode, klog.TODO(), testContext.EnableDualStackNEG),
+			fakeZoneGetter, svcPort, mode, klog.TODO(), testContext.EnableDualStackNEG, metrics.FakeSyncerMetrics()),
 		string(kubeSystemUID),
 		testContext.SvcNegClient,
 		metrics.FakeSyncerMetrics(),