Fix issue where invalid traffic reports were reported to Otterize Cloud (#280)

Author: orishavit

src/mapper/pkg/clouduploader/cloud_upload.go

@@ -311,10 +311,10 @@ func (c *CloudUploader) NotifyTrafficLevels(ctx context.Context, trafficLevels t
 	var inputs []cloudclient.TrafficLevelInput
 	for trafficPair, trafficData := range trafficLevels {
 		inputs = append(inputs, cloudclient.TrafficLevelInput{
-			ClientName:          trafficPair.Source.Name,
-			ClientNamespace:     trafficPair.Source.Namespace,
-			ServerName:          trafficPair.Destination.Name,
-			ServerNamespace:     trafficPair.Destination.Namespace,
+			ClientName:          trafficPair.SourceName,
+			ClientNamespace:     trafficPair.SourceNamespace,
+			ServerName:          trafficPair.DestinationName,
+			ServerNamespace:     trafficPair.DestinationNamespace,
 			DataBytesPerSecond:  trafficData.Bytes,
 			FlowsCountPerSecond: trafficData.Flows,
 		})

src/mapper/pkg/collectors/traffic/collector.go

@@ -7,8 +7,10 @@ import (
 )
 
 type TrafficLevelKey struct {
-	Source      serviceidentity.ServiceIdentity
-	Destination serviceidentity.ServiceIdentity
+	SourceName           string
+	SourceNamespace      string
+	DestinationName      string
+	DestinationNamespace string
 }
 
 type TrafficLevelData struct {
@@ -34,8 +36,10 @@ func NewCollector() *Collector {
 
 func (c *Collector) Add(source, destination serviceidentity.ServiceIdentity, bytes, flows int) {
 	trafficKey := TrafficLevelKey{
-		Source:      source,
-		Destination: destination,
+		SourceName:           source.Name,
+		SourceNamespace:      source.Namespace,
+		DestinationName:      destination.Name,
+		DestinationNamespace: destination.Namespace,
 	}
 
 	c.trafficLevels[trafficKey] = append(c.trafficLevels[trafficKey], TrafficLevelData{

src/mapper/pkg/resolvers/schema.helpers.resolvers.go

@@ -382,7 +382,7 @@ func (r *Resolver) handleTrafficLevelReport(ctx context.Context, results model.T
 			logrus.
 				WithField("ip", report.SrcIP).
 				WithError(err).
-				Error("Could not resolve source IP to identity")
+				Debug("could not resolve source IP to identity")
 			continue
 		}
 
@@ -392,7 +392,19 @@ func (r *Resolver) handleTrafficLevelReport(ctx context.Context, results model.T
 			logrus.
 				WithField("ip", report.DstIP).
 				WithError(err).
-				Error("Could not resolve destination IP to identity")
+				Debug("could not resolve destination IP to identity")
+			continue
+		}
+
+		if sourceIdentity.Name == "" || destinationIdentity.Name == "" || sourceIdentity.Namespace == "" || destinationIdentity.Namespace == "" {
+			// catches a bug where the source or destination identity is not set, without any other errors
+			logrus.
+				WithContext(ctx).
+				WithField("sourceIP", report.SrcIP).
+				WithField("destinationIP", report.DstIP).
+				WithField("sourceIdentity", sourceIdentity.String()).
+				WithField("destinationIdentity", destinationIdentity.String()).
+				Error("invalid traffic level report")
 			continue
 		}
 
@@ -885,38 +897,38 @@ func (r *Resolver) resolveIPToIdentity(ctx context.Context, ip string) (servicei
 	isPod, err := r.kubeFinder.IsPodIp(ctx, ip)
 
 	if err != nil {
-		logrus.WithField("ip", ip).WithError(err).Error("could not determine if IP is a pod")
 		return serviceidentity.ServiceIdentity{}, errors.Wrap(err)
 	}
 
 	if isPod {
 		sourcePod, err := r.kubeFinder.ResolveIPToPod(ctx, ip)
 
 		if err != nil {
-			logrus.WithField("ip", ip).WithError(err).Error("could not resolve source pod")
 			return serviceidentity.ServiceIdentity{}, errors.Wrap(err)
 		}
 
 		identity, err = r.serviceIdResolver.ResolvePodToServiceIdentity(ctx, sourcePod)
 
 		if err != nil {
-			logrus.WithField("ip", ip).WithError(err).Error("could not resolve source identity")
 			return serviceidentity.ServiceIdentity{}, errors.Wrap(err)
 		}
 	} else {
 		sourceService, ok, err := r.kubeFinder.ResolveIPToService(ctx, ip)
-
-		if !ok || err != nil {
-			logrus.WithField("ip", ip).WithError(err).Error("could not resolve source service")
+		if !ok {
+			err = errors.New("no mapping between IP and service")
+		}
+		if err != nil {
 			return serviceidentity.ServiceIdentity{}, errors.Wrap(err)
 		}
 
 		otrSourceIdentity, ok, err := r.kubeFinder.ResolveOtterizeIdentityForService(ctx, sourceService, time.Now())
-
-		if !ok || err != nil {
-			logrus.WithField("ip", ip).WithError(err).Error("could not resolve source identity")
+		if !ok {
+			err = errors.New("no mapping between service and otterize identity")
+		}
+		if err != nil {
 			return serviceidentity.ServiceIdentity{}, errors.Wrap(err)
 		}
+
 		identity = serviceidentity.ServiceIdentity{
 			Name:      otrSourceIdentity.Name,
 			Namespace: otrSourceIdentity.Namespace,