Added key to kafka message for connection tracking consistency

pkg/exporter/kafka_proto.go

@@ -2,6 +2,7 @@ package exporter
 
 import (
 	"context"
+	"strconv"
 
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/flow"
 	kafkago "github.com/segmentio/kafka-go"
@@ -28,6 +29,26 @@ func (kp *KafkaProto) ExportFlows(input <-chan []*flow.Record) {
 	}
 }
 
+func getIPKey(ip [16]uint8) string {
+	key := ""
+	for _, v := range ip {
+		key += strconv.Itoa(int(v))
+	}
+	return key
+}
+
+func getFlowKey(record *flow.Record) []byte {
+	key := ""
+	key1 := getIPKey(record.Id.SrcIp)
+	key2 := getIPKey(record.Id.DstIp)
+	if key1 > key2 {
+		key = key1 + key2
+	} else {
+		key = key2 + key1
+	}
+	return []byte(key)
+}
+
 func (kp *KafkaProto) batchAndSubmit(records []*flow.Record) {
 	klog.Debugf("sending %d records", len(records))
 	msgs := make([]kafkago.Message, 0, len(records))
@@ -37,7 +58,7 @@ func (kp *KafkaProto) batchAndSubmit(records []*flow.Record) {
 			klog.WithError(err).Debug("can't encode protobuf message. Ignoring")
 			continue
 		}
-		msgs = append(msgs, kafkago.Message{Value: pbBytes})
+		msgs = append(msgs, kafkago.Message{Value: pbBytes, Key: getFlowKey(record)})
 	}
 
 	if err := kp.Writer.WriteMessages(context.TODO(), msgs...); err != nil {