Initial intg with UDN interface mapping api

Signed-off-by: Mohamed Mahmoud <[email protected]>

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/mdlayher/ethernet v0.0.0-20220221185849-529eae5b6118
 	github.com/netobserv/flowlogs-pipeline v1.6.1-crc0.0.20240930010330-bc3d4ebb4a88
 	github.com/netobserv/gopipes v0.3.0
-	github.com/ovn-org/ovn-kubernetes/go-controller v0.0.0-20241126140656-c95491e46334
+	github.com/ovn-kubernetes/ovn-kubernetes/go-controller v0.0.0-20241216151556-893bab69f917
 	github.com/paulbellamy/ratecounter v0.2.0
 	github.com/prometheus/client_golang v1.20.5
 	github.com/segmentio/kafka-go v0.4.47
@@ -152,3 +152,6 @@ require (
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect
 )
+
+// HACK
+replace github.com/ovn-kuberentes/ovn-kubernetes/go-controller => github.com/npinaeva/ovn-kubernetes/go-controller v0.0.0-20241216151556-893bab69f917

pkg/agent/agent.go

@@ -23,7 +23,7 @@ import (
 
 	"github.com/cilium/ebpf/ringbuf"
 	"github.com/gavv/monotime"
-	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
+	ovnobserv "github.com/ovn-kubernetes/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/segmentio/kafka-go/compress"
 	"github.com/sirupsen/logrus"
@@ -178,7 +178,7 @@ func FlowsAgent(cfg *Config) (*Flows, error) {
 	m := metrics.NewMetrics(metricsSettings)
 
 	var s *ovnobserv.SampleDecoder
-	if cfg.EnableNetworkEventsMonitoring {
+	if cfg.EnableNetworkEventsMonitoring || cfg.EnableUDNMapping {
 		if !kernel.IsKernelOlderThan("5.14.0") {
 			if s, err = ovnobserv.NewSampleDecoderWithDefaultCollector(context.Background(), networkEventsDBPath,
 				networkEventsOwnerName, cfg.NetworkEventsMonitoringGroupID); err != nil {

pkg/agent/config.go

@@ -248,7 +248,8 @@ type Config struct {
 	EbpfProgramManagerMode bool `env:"EBPF_PROGRAM_MANAGER_MODE" envDefault:"false"`
 	// BpfManBpfFSPath user configurable ebpf manager mount path
 	BpfManBpfFSPath string `env:"BPFMAN_BPF_FS_PATH" envDefault:"/run/netobserv/maps"`
-
+	// EnableUDNMapping to allow mapping pod's interface to udn label
+	EnableUDNMapping bool `env:"ENABLE_UDNMAPPING" envDefault:"false"`
 	/* Deprecated configs are listed below this line
 	 * See manageDeprecatedConfigs function for details
 	 */

pkg/decode/decode_protobuf.go

@@ -95,6 +95,10 @@ func RecordToMap(fr *model.Record) config.GenericMap {
 		interfaces = append(interfaces, fr.Interface)
 		directions = append(directions, int(fr.ID.Direction))
 	}
+
+	if len(fr.UdnList) != 0 {
+		out["Udns"] = fr.UdnList
+	}
 	out["Interfaces"] = interfaces
 	out["IfDirections"] = directions
 

pkg/exporter/grpc_proto.go

@@ -9,7 +9,7 @@ import (
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/utils"
 
-	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
+	ovnobserv "github.com/ovn-kubernetes/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 )

pkg/exporter/kafka_proto.go

@@ -7,7 +7,7 @@ import (
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/model"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
 
-	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
+	ovnobserv "github.com/ovn-kubernetes/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
 	kafkago "github.com/segmentio/kafka-go"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"

pkg/model/record.go

@@ -63,6 +63,7 @@ type Record struct {
 	TimeFlowRtt            time.Duration
 	DupList                []map[string]uint8
 	NetworkMonitorEventsMD []config.GenericMap
+	UdnList                []string
 }
 
 func NewRecord(

pkg/pbflow/proto.go

@@ -7,8 +7,8 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/model"
-	ovnmodel "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/model"
-	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
+	ovnobserv "github.com/ovn-kuberentes/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
+	ovnmodel "github.com/ovn-kubernetes/ovn-kubernetes/go-controller/observability-lib/model"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -95,10 +95,19 @@ func FlowToPB(fr *model.Record, s *ovnobserv.SampleDecoder) *Record {
 		pbflowRecord.DupList = make([]*DupMapEntry, 0)
 		for _, m := range fr.DupList {
 			for key, value := range m {
-				pbflowRecord.DupList = append(pbflowRecord.DupList, &DupMapEntry{
+				entry := DupMapEntry{
 					Interface: key,
 					Direction: Direction(value),
-				})
+				}
+				/*
+					if s != nil {
+						m, err := s.GetInterfaceUDNs()
+						if err != nil {
+							entry.Udn = m[entry.Interface]
+						}
+					}
+				*/
+				pbflowRecord.DupList = append(pbflowRecord.DupList, &entry)
 			}
 		}
 	}
@@ -222,6 +231,7 @@ func PBToFlow(pb *Record) *model.Record {
 			intf := entry.Interface
 			dir := uint8(entry.Direction)
 			out.DupList = append(out.DupList, map[string]uint8{intf: dir})
+			out.UdnList = append(out.UdnList, entry.Udn)
 		}
 	}
 	if len(pb.GetNetworkEventsMetadata()) != 0 {

proto/flow.proto

@@ -21,6 +21,7 @@ message Records {
 message DupMapEntry {
   string interface = 1;
   Direction direction = 2;
+  string udn = 3;
 }
 
 message NetworkEvent {