Improve performance - prevent sending the same state to Otterize cloud more than once (#312)

Author: zohar7ch

src/mapper/pkg/metrics_collection_traffic/metrics_collection_traffic_handler.go

@@ -12,13 +12,15 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sync"
 )
 
 type MetricsCollectionTrafficHandler struct {
 	client.Client
 	serviceIdResolver *serviceidresolver.Resolver
 	otterizeCloud     cloudclient.CloudClient
 	cache             *MetricsCollectionTrafficCache
+	reportToCloudLock sync.Mutex
 }
 
 func NewMetricsCollectionTrafficHandler(client client.Client, serviceIdResolver *serviceidresolver.Resolver, otterizeCloud cloudclient.CloudClient) *MetricsCollectionTrafficHandler {
@@ -108,6 +110,9 @@ func (r *MetricsCollectionTrafficHandler) HandleAllServicesInNamespace(ctx conte
 }
 
 func (r *MetricsCollectionTrafficHandler) reportToCloud(ctx context.Context, namespace string, reason cloudclient.EligibleForMetricsCollectionReason, pods []cloudclient.K8sResourceEligibleForMetricsCollectionInput) error {
+	r.reportToCloudLock.Lock()
+	defer r.reportToCloudLock.Unlock()
+
 	// Remove duplicates - in case we have multiple pods that indicates on the same workload
 	pods = lo.UniqBy(pods, func(item cloudclient.K8sResourceEligibleForMetricsCollectionInput) string {
 		return item.Name

src/mapper/pkg/resourcevisibility/svc_reconciler.go

@@ -20,6 +20,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"slices"
+	"sync"
 	"time"
 )
 
@@ -29,6 +30,7 @@ type ServiceReconciler struct {
 	otterizeCloud                    cloudclient.CloudClient
 	kubeFinder                       KubeFinder
 	namespaceToReportedServicesCache *expirable.LRU[string, []byte]
+	reportToCloudLock                sync.Mutex
 }
 
 type KubeFinder interface {
@@ -80,6 +82,9 @@ func (r *ServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, errors.Wrap(err)
 	}
 
+	r.reportToCloudLock.Lock()
+	defer r.reportToCloudLock.Unlock()
+
 	hashSum, err := r.getCachedValue(servicesToReport)
 	if err != nil {
 		return ctrl.Result{}, errors.Wrap(err)

src/mapper/pkg/webhook_traffic/webhook_services_handler.go

@@ -14,6 +14,7 @@ import (
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sync"
 	"time"
 )
 
@@ -23,9 +24,10 @@ type KubeFinder interface {
 
 type WebhookServicesHandler struct {
 	client.Client
-	otterizeCloud cloudclient.CloudClient
-	kubeFinder    KubeFinder
-	cache         *WebhookServicesCache
+	otterizeCloud     cloudclient.CloudClient
+	kubeFinder        KubeFinder
+	cache             *WebhookServicesCache
+	reportToCloudLock sync.Mutex
 }
 
 func NewWebhookServicesHandler(client client.Client, otterizeCloud cloudclient.CloudClient, kubeFinder KubeFinder) *WebhookServicesHandler {
@@ -67,6 +69,9 @@ func (h *WebhookServicesHandler) HandleAll(ctx context.Context) error {
 }
 
 func (h *WebhookServicesHandler) reportToCloud(ctx context.Context, allWebhookServices []cloudclient.K8sWebhookServiceInput) error {
+	h.reportToCloudLock.Lock()
+	defer h.reportToCloudLock.Unlock()
+
 	// dedup
 	allWebhookServices = lo.UniqBy(allWebhookServices, K8sWebhookServiceInputKey)