added communication with cloud

src/node-agent/cmd/agent/main.go

@@ -4,6 +4,7 @@ import (
 	"github.com/otterize/network-mapper/src/node-agent/pkg/container"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/reconcilers"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/service"
+	"github.com/otterize/network-mapper/src/shared/cloudclient"
 	"github.com/sirupsen/logrus"
 	ctrl "sigs.k8s.io/controller-runtime"
 )
@@ -17,9 +18,18 @@ func main() {
 	criClient := service.CreateCRIClientOrDie()
 	containerManager := container.NewContainerManager(criClient)
 
+	cloudClient, cloudEnabled, err := cloudclient.NewClient(signalHandlerCtx)
+	if err != nil {
+		logrus.WithError(err).Panic("Failed to initialize cloud client")
+	}
+	if !cloudEnabled {
+		logrus.WithError(err).Panic("Cloud client is not enabled")
+	}
+
 	reconcilers.RegisterReconcilersOrDie(
 		mgr,
 		client,
+		cloudClient,
 		containerManager,
 	)
 

src/node-agent/pkg/ebpf/eventreader.go

@@ -14,6 +14,8 @@ import (
 	"github.com/otterize/network-mapper/src/node-agent/pkg/container"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/ebpf/types"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/eventparser"
+	"github.com/otterize/network-mapper/src/shared/cloudclient"
+	"github.com/samber/lo"
 	"github.com/sirupsen/logrus"
 	"io"
 	corev1 "k8s.io/api/core/v1"
@@ -24,10 +26,10 @@ import (
 )
 
 type EventReader struct {
-	client       client.Client
-	perfReader   *perf.Reader
-	containerMap map[uint32]container.ContainerInfo
-
+	client            client.Client
+	perfReader        *perf.Reader
+	cloudClient       cloudclient.CloudClient
+	containerMap      map[uint32]container.ContainerInfo
 	serviceIdResolver *serviceidresolver.Resolver
 }
 
@@ -37,7 +39,12 @@ func init() {
 	iamlivecore.ReadServiceFiles()
 }
 
-func NewEventReader(client client.Client, serviceIdResolver *serviceidresolver.Resolver, perfMap *ebpf.Map) (*EventReader, error) {
+func NewEventReader(
+	client client.Client,
+	cloudClient cloudclient.CloudClient,
+	serviceIdResolver *serviceidresolver.Resolver,
+	perfMap *ebpf.Map,
+) (*EventReader, error) {
 	perfReader, err := perf.NewReader(perfMap, os.Getpagesize()*64)
 	if err != nil {
 		return nil, errors.Wrap(err)
@@ -46,6 +53,7 @@ func NewEventReader(client client.Client, serviceIdResolver *serviceidresolver.R
 	return &EventReader{
 		client:            client,
 		perfReader:        perfReader,
+		cloudClient:       cloudClient,
 		containerMap:      make(map[uint32]container.ContainerInfo),
 		serviceIdResolver: serviceIdResolver,
 	}, nil
@@ -145,6 +153,26 @@ func (e *EventReader) updateWorkload(eventCtx types.EventContext) error {
 
 	logrus.Printf("SERVICE ID: %v", serviceId)
 
+	// Update the workload with the metadata
+	serviceMeta := cloudclient.ReportServiceMetadataInput{
+		Identity: cloudclient.ServiceIdentityInput{
+			Name:      serviceId.Name,
+			Namespace: serviceId.Namespace,
+			Kind:      serviceId.Kind,
+		},
+		Metadata: cloudclient.ServiceMetadataInput{
+			Tags: lo.MapToSlice(eventCtx.Metadata.Tags, func(key types.EventTag, _ bool) string {
+				return string(key)
+			}),
+		},
+	}
+
+	err = e.cloudClient.ReportServiceMeta(ctx, serviceMeta)
+	if err != nil {
+		return errors.Wrap(err)
+
+	}
+
 	return nil
 }
 

src/node-agent/pkg/eventparser/textualdata/handlers.go

@@ -33,7 +33,7 @@ func ContainsPaymentKeywords(ctx ebpftypes.EventContext, data string) error {
 	// Check if any of the keywords are present in the text
 	for _, keyword := range keywords {
 		if strings.Contains(data, keyword) {
-			// Set PCI tag
+			// Set PII tag
 			ctx.Metadata.AddTag(ebpftypes.EventTagPCI)
 		}
 	}
@@ -45,7 +45,7 @@ func ContainsAddress(ctx ebpftypes.EventContext, data string) error {
 	// Check if any of the keywords are present in the text
 	for _, keyword := range helpers.AddressKeywords {
 		if strings.Contains(data, keyword) {
-			// Set PCI tag
+			// Set PII tag
 			ctx.Metadata.AddTag(ebpftypes.EventTagPII)
 		}
 	}

src/node-agent/pkg/reconcilers/ebpfreconciler.go

@@ -9,6 +9,7 @@ import (
 	"github.com/otterize/network-mapper/src/node-agent/pkg/container"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/ebpf"
 	"github.com/otterize/network-mapper/src/node-agent/pkg/labels"
+	"github.com/otterize/network-mapper/src/shared/cloudclient"
 	"github.com/otterize/network-mapper/src/shared/kubeutils"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
@@ -29,10 +30,11 @@ type EBPFReconciler struct {
 
 func NewEBPFReconciler(
 	client client.Client,
+	cloudClient cloudclient.CloudClient,
 	containerManager *container.ContainerManager,
 ) (*EBPFReconciler, error) {
 	idResolver := serviceidresolver.NewResolver(client)
-	eventReader, err := ebpf.NewEventReader(client, idResolver, otrzebpf.Objs.SslEvents)
+	eventReader, err := ebpf.NewEventReader(client, cloudClient, idResolver, otrzebpf.Objs.SslEvents)
 	if err != nil {
 		return nil, errors.Wrap(err)
 	}

src/node-agent/pkg/reconcilers/reconcilers.go

@@ -2,6 +2,7 @@ package reconcilers
 
 import (
 	"github.com/otterize/network-mapper/src/node-agent/pkg/container"
+	"github.com/otterize/network-mapper/src/shared/cloudclient"
 	"github.com/sirupsen/logrus"
 	"reflect"
 	crtClient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -11,10 +12,12 @@ import (
 func RegisterReconcilersOrDie(
 	mgr manager.Manager,
 	client crtClient.Client,
+	cloudClient cloudclient.CloudClient,
 	containerManager *container.ContainerManager,
 ) {
 	ebpfReconciler, err := NewEBPFReconciler(
 		client,
+		cloudClient,
 		containerManager,
 	)
 

src/shared/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
+	ReportServiceMeta(ctx context.Context, serviceMeta ReportServiceMetadataInput) error
 }
 
 type CloudClientImpl struct {
@@ -97,3 +98,14 @@ func (c *CloudClientImpl) ReportK8sIngresses(ctx context.Context, namespace stri
 
 	return nil
 }
+
+func (c *CloudClientImpl) ReportServiceMeta(ctx context.Context, serviceMeta ReportServiceMetadataInput) error {
+	logrus.Debugf("Reporting service meta to cloud: %s", serviceMeta.Identity.Name)
+
+	_, err := ReportServiceMeta(ctx, c.client, serviceMeta)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	return nil
+}

src/shared/cloudclient/generate.go

@@ -3,6 +3,6 @@ package cloudclient
 import _ "github.com/suessflorian/gqlfetch"
 
 // The check for $CI makes sure we don't redownload the schema in CI.
-//go:generate sh -c "if [ -z $CI ]; then go run github.com/suessflorian/gqlfetch/gqlfetch --endpoint https://app.staging.otterize.com/api/graphql/v1beta > schema.graphql; fi"
+//go:generate sh -c "if [ -z $CI ]; then go run github.com/suessflorian/gqlfetch/gqlfetch --endpoint http://local.otterize.com:3000/api/graphql/v1beta > schema.graphql; fi"
 //go:generate go run github.com/Khan/genqlient ./genqlient.yaml
 //go:generate go run go.uber.org/mock/[email protected] -destination=./mocks/mocks.go  -package=cloudclientmocks -source=./cloud_client.go CloudClient

src/shared/cloudclient/genqlient.graphql

@@ -21,4 +21,8 @@ mutation ReportK8sServices($namespace: String!, $services: [K8sServiceInput!]!)
 
 mutation ReportK8sIngresses($namespace: String!, $ingresses: [K8sIngressInput!]!) {
     reportK8sIngresses(namespace: $namespace, ingresses: $ingresses)
+}
+
+mutation ReportServiceMeta($serviceMeta: ReportServiceMetadataInput!) {
+    reportServiceMeta(serviceMeta: $serviceMeta)
 }