Emit map as an OpenTelemetry metric

.github/workflows/e2e-test.yaml

@@ -26,7 +26,6 @@ on:
         default: us-central1-docker.pkg.dev/main-383408/otterize
 
 
-
 env:
   REGISTRY: ${{ inputs.registry }}
 

src/mapper/cmd/main.go

@@ -3,6 +3,12 @@ package main
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
@@ -14,6 +20,7 @@ import (
 	"github.com/otterize/network-mapper/src/mapper/pkg/config"
 	"github.com/otterize/network-mapper/src/mapper/pkg/intentsstore"
 	"github.com/otterize/network-mapper/src/mapper/pkg/kubefinder"
+	"github.com/otterize/network-mapper/src/mapper/pkg/metricexporter"
 	"github.com/otterize/network-mapper/src/mapper/pkg/resolvers"
 	sharedconfig "github.com/otterize/network-mapper/src/shared/config"
 	"github.com/otterize/network-mapper/src/shared/kubeutils"
@@ -23,14 +30,9 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/metadata"
-	"net/http"
-	"os"
-	"os/signal"
 	clientconfig "sigs.k8s.io/controller-runtime/pkg/client/config"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
-	"syscall"
-	"time"
 )
 
 func getClusterDomainOrDefault() string {
@@ -119,13 +121,24 @@ func main() {
 	if err != nil {
 		logrus.WithError(err).Fatal("Failed to initialize cloud client")
 	}
+
+	cloudUploaderConfig := clouduploader.ConfigFromViper()
 	if cloudEnabled {
-		cloudUploaderConfig := clouduploader.ConfigFromViper()
 		cloudUploader := clouduploader.NewCloudUploader(intentsHolder, cloudUploaderConfig, cloudClient)
-		go cloudUploader.PeriodicIntentsUpload(cloudClientCtx)
+		intentsHolder.RegisterGetCallback(cloudUploader.GetIntentCallback)
 		go cloudUploader.PeriodicStatusReport(cloudClientCtx)
 	}
 
+	if viper.GetBool(config.OTelEnabledKey) {
+		otelExporter, err := metricexporter.NewMetricExporter(cloudClientCtx)
+		intentsHolder.RegisterGetCallback(otelExporter.GetIntentCallback)
+		if err != nil {
+			logrus.WithError(err).Fatal("Failed to initialize otel exporter")
+		}
+	}
+
+	go intentsHolder.PeriodicIntentsUpload(cloudClientCtx, cloudUploaderConfig.UploadInterval)
+
 	telemetrysender.SetGlobalVersion(version.Version())
 	telemetrysender.SendNetworkMapper(telemetriesgql.EventTypeStarted, 1)
 	logrus.Info("Starting api server")

src/mapper/pkg/clouduploader/cloud_upload.go

@@ -2,13 +2,14 @@ package clouduploader
 
 import (
 	"context"
+	"time"
+
 	"github.com/cenkalti/backoff/v4"
 	"github.com/otterize/network-mapper/src/mapper/pkg/cloudclient"
 	"github.com/otterize/network-mapper/src/mapper/pkg/graph/model"
 	"github.com/otterize/network-mapper/src/mapper/pkg/intentsstore"
 	"github.com/samber/lo"
 	"github.com/sirupsen/logrus"
-	"time"
 )
 
 type CloudUploader struct {
@@ -25,10 +26,10 @@ func NewCloudUploader(intentsHolder *intentsstore.IntentsHolder, config Config,
 	}
 }
 
-func (c *CloudUploader) uploadDiscoveredIntents(ctx context.Context) {
+func (c *CloudUploader) GetIntentCallback(ctx context.Context, intents []intentsstore.TimestampedIntent) {
 	logrus.Info("Search for intents")
 
-	discoveredIntents := lo.Map(c.intentsHolder.GetNewIntentsSinceLastGet(), func(intent intentsstore.TimestampedIntent, _ int) *cloudclient.DiscoveredIntentInput {
+	discoveredIntents := lo.Map(intents, func(intent intentsstore.TimestampedIntent, _ int) *cloudclient.DiscoveredIntentInput {
 		return &cloudclient.DiscoveredIntentInput{
 			DiscoveredAt: lo.ToPtr(intent.Timestamp),
 			Intent: &cloudclient.IntentInput{
@@ -90,20 +91,6 @@ func (c *CloudUploader) reportStatus(ctx context.Context) {
 	}
 }
 
-func (c *CloudUploader) PeriodicIntentsUpload(ctx context.Context) {
-	logrus.Info("Starting periodic intents upload")
-
-	for {
-		select {
-		case <-time.After(c.config.UploadInterval):
-			c.uploadDiscoveredIntents(ctx)
-
-		case <-ctx.Done():
-			return
-		}
-	}
-}
-
 func (c *CloudUploader) PeriodicStatusReport(ctx context.Context) {
 	logrus.Info("Starting status reporting")
 	c.reportStatus(ctx)