NETOBSERV-1644: intg with libovn-observ golang pkg to enrich ovs events

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

go.mod

@@ -14,12 +14,13 @@ require (
 	github.com/mdlayher/ethernet v0.0.0-20220221185849-529eae5b6118
 	github.com/netobserv/flowlogs-pipeline v0.1.12-0.20240325100124-fd783b283c7c
 	github.com/netobserv/gopipes v0.3.0
+	github.com/ovn-org/ovn-kubernetes/go-controller v0.0.0-00010101000000-000000000000
 	github.com/paulbellamy/ratecounter v0.2.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/segmentio/kafka-go v0.4.47
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
-	github.com/vishvananda/netlink v1.1.0
+	github.com/vishvananda/netlink v1.2.1-beta.2.0.20231024175852-77df5d35f725
 	github.com/vishvananda/netns v0.0.4
 	github.com/vladimirvivien/gexe v0.2.0
 	github.com/vmware/go-ipfix v0.9.0
@@ -39,11 +40,13 @@ require (
 	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.2.1 // indirect
+	github.com/cenkalti/hub v1.0.1 // indirect
+	github.com/cenkalti/rpc2 v0.0.0-20210604223624-c1acbc6ec984 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
-	github.com/evanphx/json-patch/v5 v5.6.0 // indirect
+	github.com/evanphx/json-patch/v5 v5.8.0 // indirect
 	github.com/go-kit/kit v0.13.0 // indirect
 	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
@@ -83,6 +86,7 @@ require (
 	github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f // indirect
 	github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500 // indirect
 	github.com/netsampler/goflow2 v1.3.7 // indirect
+	github.com/ovn-org/libovsdb v0.6.1-0.20240125124854-03f787b1a892 // indirect
 	github.com/pierrec/lz4/v4 v4.1.17 // indirect
 	github.com/pion/dtls/v2 v2.2.4 // indirect
 	github.com/pion/logging v0.2.2 // indirect
@@ -121,16 +125,18 @@ require (
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240227224415-6ceb2ff114de // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240318140521-94a12d6c2237 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/klog/v2 v2.110.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
 	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	lukechampine.com/uint128 v1.2.0 // indirect
-	sigs.k8s.io/controller-runtime v0.15.1 // indirect
+	sigs.k8s.io/controller-runtime v0.17.0 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
-	sigs.k8s.io/yaml v1.3.0 // indirect
+	sigs.k8s.io/yaml v1.4.0 // indirect
 )
+
+replace github.com/ovn-org/ovn-kubernetes/go-controller => github.com/npinaeva/ovn-kubernetes/go-controller v0.0.0-20240514113629-6445cb329c05

pkg/decode/decode_protobuf.go

@@ -50,6 +50,7 @@ func PBFlowToMap(pb *pbflow.Record) config.GenericMap {
 }
 
 // RecordToMap converts the flow from Agent inner model into FLP GenericMap model
+// nolint:golint,cyclop
 func RecordToMap(fr *flow.Record) config.GenericMap {
 	if fr == nil {
 		return config.GenericMap{}
@@ -136,6 +137,12 @@ func RecordToMap(fr *flow.Record) config.GenericMap {
 	if fr.TimeFlowRtt != 0 {
 		out["TimeFlowRttNs"] = fr.TimeFlowRtt.Nanoseconds()
 	}
+
+	for idx, md := range fr.OvsMonitorMD {
+		if md != "" {
+			out[fmt.Sprintf("OvsMonitorMD[%d]", idx)] = md
+		}
+	}
 	return out
 }
 

pkg/ebpf/tracer.go

@@ -269,7 +269,7 @@ func (m *FlowFetcher) Register(iface ifaces.Interface) error {
 	if err != nil {
 		return fmt.Errorf("failed to create handle for netns (%s): %w", iface.NetNS.String(), err)
 	}
-	defer handle.Delete()
+	defer handle.Close()
 
 	// Load pre-compiled programs and maps into the kernel, and rewrites the configuration
 	ipvlan, err := handle.LinkByIndex(iface.Index)
@@ -803,7 +803,7 @@ func registerInterface(iface ifaces.Interface) (*netlink.GenericQdisc, netlink.L
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to create handle for netns (%s): %w", iface.NetNS.String(), err)
 	}
-	defer handle.Delete()
+	defer handle.Close()
 
 	// Load pre-compiled programs and maps into the kernel, and rewrites the configuration
 	ipvlan, err := handle.LinkByIndex(iface.Index)

pkg/flow/account_test.go

@@ -111,6 +111,7 @@ func TestEvict_MaxEntries(t *testing.T) {
 			TimeFlowStart: now.Add(-(1000 - 123) * time.Nanosecond),
 			TimeFlowEnd:   now.Add(-(1000 - 789) * time.Nanosecond),
 			DupList:       make([]map[string]uint8, 0),
+			OvsMonitorMD:  make([]string, 0),
 		},
 		k2: {
 			RawRecord: RawRecord{
@@ -122,6 +123,7 @@ func TestEvict_MaxEntries(t *testing.T) {
 			TimeFlowStart: now.Add(-(1000 - 456) * time.Nanosecond),
 			TimeFlowEnd:   now.Add(-(1000 - 456) * time.Nanosecond),
 			DupList:       make([]map[string]uint8, 0),
+			OvsMonitorMD:  make([]string, 0),
 		},
 	}, received)
 }
@@ -191,6 +193,7 @@ func TestEvict_Period(t *testing.T) {
 		TimeFlowStart: now.Add(-1000 + 123),
 		TimeFlowEnd:   now.Add(-1000 + 789),
 		DupList:       make([]map[string]uint8, 0),
+		OvsMonitorMD:  make([]string, 0),
 	}, *records[0])
 	records = receiveTimeout(t, evictor)
 	require.Len(t, records, 1)
@@ -208,6 +211,7 @@ func TestEvict_Period(t *testing.T) {
 		TimeFlowStart: now.Add(-1000 + 1123),
 		TimeFlowEnd:   now.Add(-1000 + 1456),
 		DupList:       make([]map[string]uint8, 0),
+		OvsMonitorMD:  make([]string, 0),
 	}, *records[0])
 
 	// no more flows are evicted

pkg/flow/record.go

@@ -52,8 +52,9 @@ type Record struct {
 	// AgentIP provides information about the source of the flow (the Agent that traced it)
 	AgentIP net.IP
 	// Calculated RTT which is set when record is created by calling NewRecord
-	TimeFlowRtt time.Duration
-	DupList     []map[string]uint8
+	TimeFlowRtt  time.Duration
+	DupList      []map[string]uint8
+	OvsMonitorMD []string
 }
 
 func NewRecord(
@@ -80,6 +81,7 @@ func NewRecord(
 		record.DNSLatency = time.Duration(metrics.DnsRecord.Latency)
 	}
 	record.DupList = make([]map[string]uint8, 0)
+	record.OvsMonitorMD = make([]string, 0)
 	return &record
 }
 

pkg/pbflow/proto.go

@@ -1,11 +1,13 @@
 package pbflow
 
 import (
+	"context"
 	"encoding/binary"
 	"net"
 
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/flow"
+	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -94,6 +96,14 @@ func FlowToPB(fr *flow.Record) *Record {
 		pbflowRecord.Network.SrcAddr = &IP{IpFamily: &IP_Ipv4{Ipv4: flow.IntEncodeV4(fr.Id.SrcIp)}}
 		pbflowRecord.Network.DstAddr = &IP{IpFamily: &IP_Ipv4{Ipv4: flow.IntEncodeV4(fr.Id.DstIp)}}
 	}
+
+	if s, err := ovnobserv.NewSampleDecoder(context.Background()); err == nil {
+		for _, ovsMetadata := range fr.Metrics.OvsDpKeys {
+			if md, err := s.DecodeCookie8Bytes(ovsMetadata); err == nil {
+				pbflowRecord.OvsMetadata = append(pbflowRecord.OvsMetadata, md)
+			}
+		}
+	}
 	return &pbflowRecord
 }
 
@@ -152,6 +162,9 @@ func PBToFlow(pb *Record) *flow.Record {
 			out.DupList = append(out.DupList, map[string]uint8{intf: dir})
 		}
 	}
+	if len(pb.GetOvsMetadata()) != 0 {
+		out.OvsMonitorMD = append(out.OvsMonitorMD, pb.GetOvsMetadata()...)
+	}
 	return &out
 }
 

proto/flow.proto

@@ -22,6 +22,7 @@ message DupMapEntry {
   string interface = 1;
   Direction direction = 2;
 }
+
 message Record {
   // protocol as defined by ETH_P_* in linux/if_ether.h
   // https://github.com/torvalds/linux/blob/master/include/uapi/linux/if_ether.h
@@ -59,6 +60,7 @@ message Record {
   google.protobuf.Duration time_flow_rtt = 24;
   uint32 dns_errno = 25;
   repeated DupMapEntry dup_list = 26;
+  repeated string ovs_metadata = 27;
 }
 
 message DataLink {