k8s,metrics: Remove double k8s event accounting

operator/watchers/k8s_service_sync.go

@@ -14,13 +14,11 @@ import (
 	"github.com/cilium/cilium/pkg/k8s/resource"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	"github.com/cilium/cilium/pkg/k8s/utils"
-	"github.com/cilium/cilium/pkg/k8s/watchers/resources"
 	"github.com/cilium/cilium/pkg/kvstore"
 	"github.com/cilium/cilium/pkg/kvstore/store"
 	"github.com/cilium/cilium/pkg/loadbalancer"
 	"github.com/cilium/cilium/pkg/lock"
 	"github.com/cilium/cilium/pkg/logging/logfields"
-	"github.com/cilium/cilium/pkg/metrics"
 	serviceStore "github.com/cilium/cilium/pkg/service/store"
 )
 
@@ -33,10 +31,6 @@ var (
 	kvs               store.SyncStore
 )
 
-func k8sEventMetric(scope, action string) {
-	metrics.EventTS.WithLabelValues(metrics.LabelEventSourceK8s, scope, action)
-}
-
 func k8sServiceHandler(ctx context.Context, clusterName string, shared bool, clusterID uint32) {
 	serviceHandler := func(event k8s.ServiceEvent) {
 		defer event.SWG.Done()
@@ -116,11 +110,6 @@ type ServiceSyncParameters struct {
 func StartSynchronizingServices(ctx context.Context, wg *sync.WaitGroup, cfg ServiceSyncParameters) {
 	kvstoreReady := make(chan struct{}, 0)
 
-	endpointMetric := resources.MetricEndpoint
-	if k8s.SupportsEndpointSlice() {
-		endpointMetric = resources.MetricEndpointSlice
-	}
-
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
@@ -196,10 +185,8 @@ func StartSynchronizingServices(ctx context.Context, wg *sync.WaitGroup, cfg Ser
 						onSync()
 					}
 				case resource.Upsert:
-					k8sEventMetric(resources.MetricService, resources.MetricUpdate)
 					K8sSvcCache.UpdateService(ev.Object, swg)
 				case resource.Delete:
-					k8sEventMetric(resources.MetricService, resources.MetricDelete)
 					K8sSvcCache.DeleteService(ev.Object, swg)
 				}
 				ev.Done(nil)
@@ -217,10 +204,8 @@ func StartSynchronizingServices(ctx context.Context, wg *sync.WaitGroup, cfg Ser
 						onSync()
 					}
 				case resource.Upsert:
-					k8sEventMetric(endpointMetric, resources.MetricUpdate)
 					K8sSvcCache.UpdateEndpoints(ev.Object, swg)
 				case resource.Delete:
-					k8sEventMetric(endpointMetric, resources.MetricDelete)
 					K8sSvcCache.DeleteEndpoints(ev.Object.EndpointSliceID, swg)
 				}
 				ev.Done(nil)

pkg/k8s/watchers/cilium_cidr_group.go

@@ -15,7 +15,6 @@ import (
 	"github.com/cilium/cilium/pkg/k8s/client"
 	"github.com/cilium/cilium/pkg/k8s/resource"
 	"github.com/cilium/cilium/pkg/k8s/types"
-	"github.com/cilium/cilium/pkg/k8s/watchers/resources"
 	"github.com/cilium/cilium/pkg/logging/logfields"
 	"github.com/cilium/cilium/pkg/metrics"
 	"github.com/cilium/cilium/pkg/policy/api"
@@ -28,33 +27,21 @@ func (k *K8sWatcher) onUpsertCIDRGroup(
 	cs client.Clientset,
 	apiGroup, metricLabel string,
 ) error {
-	var (
-		equal  bool
-		action string
-	)
 
-	// wrap k.K8sEventReceived call into a naked func() to capture equal in the closure
 	defer func() {
-		k.K8sEventReceived(apiGroup, metricLabel, action, true, equal)
+		k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 	}()
 
 	oldCidrGroup, ok := cidrGroupCache[cidrGroup.Name]
 	if ok && oldCidrGroup.Spec.DeepEqual(&cidrGroup.Spec) {
-		equal = true
 		return nil
-	} else if ok {
-		action = resources.MetricUpdate
-	} else {
-		action = resources.MetricCreate
 	}
 
 	cidrGroupCpy := cidrGroup.DeepCopy()
 	cidrGroupCache[cidrGroup.Name] = cidrGroupCpy
 
 	err := k.updateCIDRGroupRefPolicies(cidrGroup.Name, cidrGroupCache, cnpCache, cs)
 
-	k.K8sEventProcessed(metricLabel, action, err == nil)
-
 	return err
 }
 
@@ -69,8 +56,7 @@ func (k *K8sWatcher) onDeleteCIDRGroup(
 
 	err := k.updateCIDRGroupRefPolicies(cidrGroupName, cidrGroupCache, cnpCache, cs)
 
-	k.K8sEventProcessed(metricLabel, resources.MetricDelete, err == nil)
-	k.K8sEventReceived(apiGroup, metricLabel, resources.MetricDelete, true, true)
+	k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 
 	return err
 }

pkg/k8s/watchers/cilium_network_policy.go

@@ -227,23 +227,13 @@ func (k *K8sWatcher) onUpsert(
 ) error {
 	initialRecvTime := time.Now()
 
-	var (
-		equal  bool
-		action string
-	)
-
-	// wrap k.K8sEventReceived call into a naked func() to capture equal in the closure
 	defer func() {
-		k.K8sEventReceived(apiGroup, metricLabel, action, true, equal)
+		k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 	}()
 
 	oldCNP, ok := cnpCache[key]
-	if !ok {
-		action = resources.MetricCreate
-	} else {
-		action = resources.MetricUpdate
+	if ok {
 		if oldCNP.DeepEqual(cnp) {
-			equal = true
 			return nil
 		}
 	}
@@ -282,8 +272,6 @@ func (k *K8sWatcher) onUpsert(
 		cnpCache[key] = cnpCpy
 	}
 
-	k.K8sEventProcessed(metricLabel, action, err == nil)
-
 	return err
 }
 
@@ -302,8 +290,7 @@ func (k *K8sWatcher) onDelete(
 	delete(cidrGroupPolicies, key)
 	metrics.CIDRGroupPolicies.Set(float64(len(cidrGroupPolicies)))
 
-	k.K8sEventProcessed(metricLabel, resources.MetricDelete, err == nil)
-	k.K8sEventReceived(apiGroup, metricLabel, resources.MetricDelete, true, true)
+	k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 
 	return err
 }

pkg/k8s/watchers/endpoints.go

@@ -25,11 +25,6 @@ func (k *K8sWatcher) endpointsInit() {
 	// real resource kind since the codepath is the same for all.
 	apiGroup := resources.K8sAPIGroupEndpointSliceOrEndpoint
 
-	metric := resources.MetricEndpoint
-	if k8s.SupportsEndpointSlice() {
-		metric = resources.MetricEndpointSlice
-	}
-
 	var synced atomic.Bool
 	synced.Store(false)
 
@@ -56,13 +51,11 @@ func (k *K8sWatcher) endpointsInit() {
 				case resource.Sync:
 					synced.Store(true)
 				case resource.Upsert:
-					k.K8sEventReceived(apiGroup, metric, resources.MetricUpdate, true, false)
+					k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 					k.updateEndpoint(event.Object, swg)
-					k.K8sEventProcessed(metric, resources.MetricUpdate, true)
 				case resource.Delete:
-					k.K8sEventReceived(apiGroup, metric, resources.MetricDelete, true, false)
+					k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 					k.K8sSvcCache.DeleteEndpoints(event.Object.EndpointSliceID, swg)
-					k.K8sEventProcessed(metric, resources.MetricDelete, true)
 				}
 				event.Done(nil)
 			}

pkg/k8s/watchers/namespace.go

@@ -15,7 +15,6 @@ import (
 	"github.com/cilium/cilium/pkg/k8s/resource"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	slim_metav1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/apis/meta/v1"
-	"github.com/cilium/cilium/pkg/k8s/watchers/resources"
 	"github.com/cilium/cilium/pkg/labels"
 	"github.com/cilium/cilium/pkg/labelsfilter"
 	"github.com/cilium/cilium/pkg/logging/logfields"
@@ -59,7 +58,6 @@ func (k *K8sWatcher) namespacesInit() {
 					synced.Store(true)
 				case resource.Upsert:
 					err = nsUpdater.update(event.Object)
-					k.K8sEventProcessed(metricNS, resources.MetricUpdate, err == nil)
 				}
 				event.Done(err)
 			}

pkg/k8s/watchers/node.go

@@ -14,7 +14,6 @@ import (
 	"github.com/cilium/cilium/pkg/k8s/client"
 	"github.com/cilium/cilium/pkg/k8s/resource"
 	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
-	"github.com/cilium/cilium/pkg/k8s/watchers/resources"
 	"github.com/cilium/cilium/pkg/k8s/watchers/subscriber"
 	"github.com/cilium/cilium/pkg/lock"
 	nodeTypes "github.com/cilium/cilium/pkg/node/types"
@@ -79,15 +78,13 @@ func (k *K8sWatcher) nodeEventLoop(synced *atomic.Bool, swg *lock.StoppableWaitG
 			case resource.Upsert:
 				newNode := event.Object
 				if oldNode == nil {
-					k.K8sEventReceived(apiGroup, metricNode, resources.MetricCreate, true, false)
+					k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 					errs = k.NodeChain.OnAddNode(newNode, swg)
-					k.K8sEventProcessed(metricNode, resources.MetricCreate, errs == nil)
 				} else {
 					equal := nodeEventsAreEqual(oldNode, newNode)
-					k.K8sEventReceived(apiGroup, metricNode, resources.MetricUpdate, true, equal)
+					k.k8sResourceSynced.SetEventTimestamp(apiGroup)
 					if !equal {
 						errs = k.NodeChain.OnUpdateNode(oldNode, newNode, swg)
-						k.K8sEventProcessed(metricNode, resources.MetricUpdate, errs == nil)
 					}
 				}
 				oldNode = newNode

pkg/k8s/watchers/pod.go

@@ -76,7 +76,9 @@ func (k *K8sWatcher) createAllPodsController(slimClient slimclientset.Interface)
 		cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				if pod := k8s.ObjTov1Pod(obj); pod != nil {
-					k.addK8sPodV1(pod)
+					err := k.addK8sPodV1(pod)
+					k.K8sEventProcessed(metricPod, resources.MetricCreate, err == nil)
+					k.K8sEventReceived(podApiGroup, metricPod, resources.MetricCreate, true, false)
 				} else {
 					k.K8sEventReceived(podApiGroup, metricPod, resources.MetricCreate, false, false)
 				}
@@ -87,7 +89,9 @@ func (k *K8sWatcher) createAllPodsController(slimClient slimclientset.Interface)
 						if oldPod.DeepEqual(newPod) {
 							k.K8sEventReceived(podApiGroup, metricPod, resources.MetricUpdate, false, true)
 						} else {
-							k.updateK8sPodV1(oldPod, newPod)
+							err := k.updateK8sPodV1(oldPod, newPod)
+							k.K8sEventProcessed(metricPod, resources.MetricUpdate, err == nil)
+							k.K8sEventReceived(podApiGroup, metricPod, resources.MetricUpdate, true, false)
 						}
 					}
 				} else {
@@ -96,7 +100,9 @@ func (k *K8sWatcher) createAllPodsController(slimClient slimclientset.Interface)
 			},
 			DeleteFunc: func(obj interface{}) {
 				if pod := k8s.ObjTov1Pod(obj); pod != nil {
-					k.deleteK8sPodV1(pod)
+					err := k.deleteK8sPodV1(pod)
+					k.K8sEventProcessed(metricPod, resources.MetricDelete, err == nil)
+					k.K8sEventReceived(podApiGroup, metricPod, resources.MetricDelete, true, false)
 				} else {
 					k.K8sEventReceived(podApiGroup, metricPod, resources.MetricDelete, false, false)
 				}
@@ -136,9 +142,11 @@ func (k *K8sWatcher) podsInit(slimClient slimclientset.Interface, asyncControlle
 					} else {
 						k.updateK8sPodV1(oldPod, newPod)
 					}
+					k.k8sResourceSynced.SetEventTimestamp(podApiGroup)
 					pods[ev.Key] = newPod
 				case resource.Delete:
 					k.deleteK8sPodV1(ev.Object)
+					k.k8sResourceSynced.SetEventTimestamp(podApiGroup)
 					delete(pods, ev.Key)
 				}
 
@@ -213,12 +221,8 @@ func (k *K8sWatcher) podsInit(slimClient slimclientset.Interface, asyncControlle
 	}
 }
 
-func (k *K8sWatcher) addK8sPodV1(pod *slim_corev1.Pod) {
+func (k *K8sWatcher) addK8sPodV1(pod *slim_corev1.Pod) error {
 	var err error
-	defer func() {
-		k.K8sEventProcessed(metricPod, resources.MetricCreate, err == nil)
-		k.K8sEventReceived(podApiGroup, metricPod, resources.MetricCreate, true, false)
-	}()
 
 	logger := log.WithFields(logrus.Fields{
 		logfields.K8sPodName:   pod.ObjectMeta.Name,
@@ -255,12 +259,12 @@ func (k *K8sWatcher) addK8sPodV1(pod *slim_corev1.Pod) {
 	// is no longer in use.
 	if !k8sUtils.IsPodRunning(pod.Status) {
 		err = k.deleteK8sPodV1(pod)
-		return
+		return err
 	}
 
 	if pod.Spec.HostNetwork && !option.Config.EnableLocalRedirectPolicy {
 		logger.Debug("Skip pod event using host networking")
-		return
+		return err
 	}
 
 	podIPs := k8sUtils.ValidIPs(pod.Status)
@@ -278,17 +282,15 @@ func (k *K8sWatcher) addK8sPodV1(pod *slim_corev1.Pod) {
 		logger.WithError(err).Warning("Unable to update ipcache map entry on pod add")
 	}
 	logger.Debug("Updated ipcache map entry on pod add")
+
+	return err
 }
 
-func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) {
+func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) error {
 	var err error
-	defer func() {
-		k.K8sEventProcessed(metricPod, resources.MetricUpdate, err == nil)
-		k.K8sEventReceived(podApiGroup, metricPod, resources.MetricUpdate, true, false)
-	}()
 
 	if oldK8sPod == nil || newK8sPod == nil {
-		return
+		return err
 	}
 
 	logger := log.WithFields(logrus.Fields{
@@ -310,13 +312,13 @@ func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) {
 	// is no longer in use.
 	if !k8sUtils.IsPodRunning(newK8sPod.Status) {
 		err = k.deleteK8sPodV1(newK8sPod)
-		return
+		return err
 	}
 
 	if newK8sPod.Spec.HostNetwork && !option.Config.EnableLocalRedirectPolicy &&
 		!option.Config.EnableSocketLBTracing {
 		logger.Debug("Skip pod event using host networking")
-		return
+		return err
 	}
 
 	k.cgroupManager.OnUpdatePod(oldK8sPod, newK8sPod)
@@ -327,7 +329,7 @@ func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) {
 
 	if err != nil {
 		logger.WithError(err).Warning("Unable to update ipcache map entry on pod update")
-		return
+		return err
 	}
 
 	// Check annotation updates.
@@ -371,21 +373,21 @@ func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) {
 
 	// Nothing changed.
 	if !annotationsChanged && !labelsChanged {
-		return
+		return err
 	}
 
 	podNSName := k8sUtils.GetObjNamespaceName(&newK8sPod.ObjectMeta)
 
 	podEP := k.endpointManager.LookupPodName(podNSName)
 	if podEP == nil {
 		log.WithField("pod", podNSName).Debugf("Endpoint not found running for the given pod")
-		return
+		return err
 	}
 
 	if labelsChanged {
 		err := updateEndpointLabels(podEP, oldPodLabels, newPodLabels)
 		if err != nil {
-			return
+			return err
 		}
 
 		// Synchronize Pod labels with CiliumEndpoint labels if there is a change.
@@ -424,6 +426,8 @@ func (k *K8sWatcher) updateK8sPodV1(oldK8sPod, newK8sPod *slim_corev1.Pod) {
 		}
 		realizePodAnnotationUpdate(podEP)
 	}
+
+	return err
 }
 
 func realizePodAnnotationUpdate(podEP *endpoint.Endpoint) {
@@ -504,10 +508,6 @@ func updateEndpointLabels(ep *endpoint.Endpoint, oldLbls, newLbls map[string]str
 
 func (k *K8sWatcher) deleteK8sPodV1(pod *slim_corev1.Pod) error {
 	var err error
-	defer func() {
-		k.K8sEventProcessed(metricPod, resources.MetricDelete, err == nil)
-		k.K8sEventReceived(podApiGroup, metricPod, resources.MetricDelete, true, false)
-	}()
 
 	logger := log.WithFields(logrus.Fields{
 		logfields.K8sPodName:   pod.ObjectMeta.Name,