Collect traffic levels data from node-agent and upload to cloud (#271)

Author: orishavit

build/kafka-watcher.Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.22.1-alpine as buildenv
+FROM --platform=$BUILDPLATFORM golang:1.22.1-alpine AS buildenv
 RUN apk add --no-cache ca-certificates git protoc
 RUN apk add build-base libpcap-dev
 WORKDIR /src
@@ -9,10 +9,10 @@ RUN go mod download
 
 COPY . .
 
-FROM buildenv as test
+FROM buildenv AS test
 RUN go test ./kafka-watcher/...
 
-FROM test as builder
+FROM test AS builder
 ARG TARGETOS
 ARG TARGETARCH
 RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -trimpath -o /main ./kafka-watcher/cmd

build/mapper.Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.22.1-alpine as buildenv
+FROM --platform=$BUILDPLATFORM golang:1.22.1-alpine AS buildenv
 RUN apk add --no-cache ca-certificates git protoc
 RUN apk add build-base libpcap-dev
 WORKDIR /src
@@ -9,15 +9,15 @@ RUN go mod download
 
 COPY . .
 
-FROM buildenv as test
+FROM buildenv AS test
 # install dependencies for "envtest" package
 RUN go install sigs.k8s.io/controller-runtime/tools/[email protected] && \
     source <(setup-envtest use -p env) && \
     mkdir -p /usr/local/kubebuilder && \
     ln -s "$KUBEBUILDER_ASSETS" /usr/local/kubebuilder/bin
 RUN go test ./mapper/...
 
-FROM test as builder
+FROM test AS builder
 ARG TARGETOS
 ARG TARGETARCH
 

build/sniffer.Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=linux/amd64 golang:1.22.1-alpine as buildenv
+FROM --platform=linux/amd64 golang:1.22.1-alpine AS buildenv
 RUN apk add --no-cache ca-certificates git protoc
 RUN apk add build-base libpcap-dev
 WORKDIR /src
@@ -9,12 +9,12 @@ RUN go mod download
 
 COPY . .
 
-FROM buildenv as test
+FROM buildenv AS test
 RUN go test ./sniffer/... && echo dep > /dep
 
 # We start from the base image again, only this time it's using the target arch instead of always amd64. This is done to make the build faster.
 # Unlike the mapper, it can't be amd64 throughout and use Go's cross-compilation, since the sniffer depends on libpcap (C library).
-FROM golang:1.22.1-alpine as builder
+FROM golang:1.22.1-alpine AS builder
 COPY --from=test /dep /dep
 RUN apk add --no-cache ca-certificates git protoc
 RUN apk add build-base libpcap-dev
@@ -29,7 +29,7 @@ RUN go build -trimpath -o /main ./sniffer/cmd
 ARG VERSION
 RUN echo -n $VERSION > /version
 
-FROM alpine as release
+FROM alpine AS release
 RUN apk add --no-cache ca-certificates libpcap
 WORKDIR /
 COPY --from=builder /main /main

src/mapper/cmd/main.go

@@ -14,6 +14,7 @@ import (
 	istiowatcher "github.com/otterize/network-mapper/src/istio-watcher/pkg/watcher"
 	"github.com/otterize/network-mapper/src/mapper/pkg/awsintentsholder"
 	"github.com/otterize/network-mapper/src/mapper/pkg/azureintentsholder"
+	"github.com/otterize/network-mapper/src/mapper/pkg/collectors/traffic"
 	"github.com/otterize/network-mapper/src/mapper/pkg/dnscache"
 	"github.com/otterize/network-mapper/src/mapper/pkg/dnsintentspublisher"
 	"github.com/otterize/network-mapper/src/mapper/pkg/externaltrafficholder"
@@ -166,6 +167,7 @@ func main() {
 	incomingTrafficIntentsHolder := incomingtrafficholder.NewIncomingTrafficIntentsHolder()
 	awsIntentsHolder := awsintentsholder.New()
 	azureIntentsHolder := azureintentsholder.New()
+	trafficCollector := traffic.NewCollector()
 
 	resolver := resolvers.NewResolver(
 		kubeFinder,
@@ -176,18 +178,14 @@ func main() {
 		azureIntentsHolder,
 		dnsCache,
 		incomingTrafficIntentsHolder,
+		trafficCollector,
 	)
 	resolver.Register(mapperServer)
 
 	metricsServer := echo.New()
 	metricsServer.HideBanner = true
 	metricsServer.GET("/metrics", echoprometheus.NewHandler())
 
-	cloudClient, cloudEnabled, err := cloudclient.NewClient(errGroupCtx)
-	if err != nil {
-		logrus.WithError(err).Panic("Failed to initialize cloud client")
-	}
-
 	if viper.GetBool(config.EnableIstioCollectionKey) {
 		istioWatcher, err := istiowatcher.NewWatcher(resolver.Mutation())
 		if err != nil {
@@ -201,6 +199,10 @@ func main() {
 	}
 
 	cloudUploaderConfig := clouduploader.ConfigFromViper()
+	cloudClient, cloudEnabled, err := cloudclient.NewClient(errGroupCtx)
+	if err != nil {
+		logrus.WithError(err).Panic("Failed to initialize cloud client")
+	}
 	if cloudEnabled {
 		cloudUploader := clouduploader.NewCloudUploader(intentsHolder, cloudUploaderConfig, cloudClient)
 
@@ -211,6 +213,7 @@ func main() {
 		}
 		awsIntentsHolder.RegisterNotifyIntents(cloudUploader.NotifyAWSIntents)
 		azureIntentsHolder.RegisterNotifyIntents(cloudUploader.NotifyAzureIntents)
+		trafficCollector.RegisterNotifyTraffic(cloudUploader.NotifyTrafficLevels)
 
 		go cloudUploader.PeriodicStatusReport(errGroupCtx)
 
@@ -271,6 +274,11 @@ func main() {
 		azureIntentsHolder.PeriodicIntentsUpload(errGroupCtx, cloudUploaderConfig.UploadInterval)
 		return nil
 	})
+	errgrp.Go(func() error {
+		defer errorreporter.AutoNotify()
+		trafficCollector.PeriodicUpload(errGroupCtx, cloudUploaderConfig.UploadInterval)
+		return nil
+	})
 
 	telemetrysender.SendNetworkMapper(telemetriesgql.EventTypeStarted, 1)
 	telemetrysender.NetworkMapperRunActiveReporter(errGroupCtx)

src/mapper/pkg/cloudclient/cloud_client.go

@@ -15,6 +15,7 @@ type CloudClient interface {
 	ReportIncomingTrafficDiscoveredIntents(ctx context.Context, intents []IncomingTrafficDiscoveredIntentInput) error
 	ReportK8sServices(ctx context.Context, namespace string, services []K8sServiceInput) error
 	ReportK8sIngresses(ctx context.Context, namespace string, ingresses []K8sIngressInput) error
+	ReportTrafficLevels(ctx context.Context, trafficLevels []TrafficLevelInput) error
 }
 
 type CloudClientImpl struct {
@@ -96,3 +97,21 @@ func (c *CloudClientImpl) ReportK8sIngresses(ctx context.Context, namespace stri
 
 	return nil
 }
+
+func (c *CloudClientImpl) ReportTrafficLevels(
+	ctx context.Context,
+	trafficLevels []TrafficLevelInput,
+) error {
+	logrus.Debug("Uploading traffic info to cloud")
+
+	_, err := ReportTrafficLevels(
+		ctx,
+		c.client,
+		trafficLevels,
+	)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	return nil
+}

src/mapper/pkg/cloudclient/generated.go

@@ -21,4 +21,10 @@ mutation ReportK8sServices($namespace: String!, $services: [K8sServiceInput!]!)
 
 mutation ReportK8sIngresses($namespace: String!, $ingresses: [K8sIngressInput!]!) {
     reportK8sIngresses(namespace: $namespace, ingresses: $ingresses)
+}
+
+mutation ReportTrafficLevels(
+    $trafficLevels: [TrafficLevelInput!]!
+) {
+    reportTrafficLevels(trafficLevels: $trafficLevels)
 }