NETOBSERV-560: fix fragmentation of reported flows (#51)

Author: Mario Macias

Makefile

@@ -126,6 +126,7 @@ image-push: ## Push OCI image with the manager.
 .PHONY: tests-e2e
 .ONESHELL:
 tests-e2e: prereqs
+	go clean -testcache
 	# making the local agent image available to kind in two ways, so it will work in different
 	# environments: (1) as image tagged in the local repository (2) as image archive.
 	$(OCI_BIN) build . -t localhost/ebpf-agent:test

bpf/flows.c

@@ -171,6 +171,8 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
     id.if_index = skb->ifindex;
     id.direction = direction;
 
+    // TODO: we need to add spinlock here when we deprecate versions prior to 5.1, or provide
+    // a spinlocked alternative version and use it selectively https://lwn.net/Articles/779120/
     flow_metrics *aggregate_flow = bpf_map_lookup_elem(&aggregated_flows, &id);
     if (aggregate_flow != NULL) {
         aggregate_flow->packets += 1;
@@ -186,8 +188,10 @@ static inline int flow_monitor(struct __sk_buff *skb, u8 direction) {
             .start_mono_time_ts = current_time,
             .end_mono_time_ts = current_time,
         };
-        
-        if (bpf_map_update_elem(&aggregated_flows, &id, &new_flow, BPF_NOEXIST) != 0) {
+
+        // even if we know that the entry is new, another CPU might be concurrently inserting a flow
+        // so we need to specify BPF_ANY
+        if (bpf_map_update_elem(&aggregated_flows, &id, &new_flow, BPF_ANY) != 0) {
             /*
                 When the map is full, we directly send the flow entry to userspace via ringbuffer,
                 until space is available in the kernel-side maps

pkg/ebpf/bpf_bpfeb.o

@@ -46,6 +46,7 @@ type FlowTracer struct {
 	// manages the access to the eviction routines, avoiding two evictions happening at the same time
 	flowsEvictor   *sync.Cond
 	lastEvictionNs uint64
+	cacheMaxSize   int
 }
 
 // TODO: decouple flowtracer logic from eBPF maps access so we can inject mocks for testing
@@ -94,6 +95,7 @@ func NewFlowTracer(
 		interfaceNamer:  namer,
 		flowsEvictor:    sync.NewCond(&sync.Mutex{}),
 		lastEvictionNs:  uint64(monotime.Now()),
+		cacheMaxSize:    cacheMaxSize,
 	}, nil
 }
 
@@ -300,12 +302,9 @@ func (m *FlowTracer) evictFlows(tlog *logrus.Entry, forwardFlows chan<- []*flow.
 	monotonicTimeNow := monotime.Now()
 	currentTime := time.Now()
 
-	mapKeys, mapValues := m.lookupAndDeleteMapKeysValues()
-
 	var forwardingFlows []*flow.Record
 	laterFlowNs := uint64(0)
-	for nf, flowKey := range mapKeys {
-		flowMetrics := mapValues[nf]
+	for flowKey, flowMetrics := range m.lookupAndDeleteFlowsMap() {
 		aggregatedMetrics := m.aggregate(flowMetrics)
 		// we ignore metrics that haven't been aggregated (e.g. all the mapped values are ignored)
 		if aggregatedMetrics.EndMonoTimeNs == 0 {
@@ -380,12 +379,12 @@ func (m *FlowTracer) listenAndForwardRingBuffer(ctx context.Context, forwardFlow
 // Changing this method invaction by BatchLookupAndDelete could improve performance
 // TODO: detect whether BatchLookupAndDelete is supported (Kernel>=5.6) and use it selectively
 // Supported Lookup/Delete oprations by kernel: https://github.com/iovisor/bcc/blob/master/docs/kernel-versions.md
-func (m *FlowTracer) lookupAndDeleteMapKeysValues() ([]flow.RecordKey, [][]flow.RecordMetrics) {
+func (m *FlowTracer) lookupAndDeleteFlowsMap() map[flow.RecordKey][]flow.RecordMetrics {
 	flowMap := m.objects.AggregatedFlows
-	var flowKeys []flow.RecordKey
-	var flowsValues [][]flow.RecordMetrics
 
 	iterator := flowMap.Iterate()
+	flows := make(map[flow.RecordKey][]flow.RecordMetrics, m.cacheMaxSize)
+
 	id := flow.RecordKey{}
 	var metrics []flow.RecordMetrics
 	// Changing Iterate+Delete by LookupAndDelete would prevent some possible race conditions
@@ -395,8 +394,10 @@ func (m *FlowTracer) lookupAndDeleteMapKeysValues() ([]flow.RecordKey, [][]flow.
 			log.WithError(err).WithField("flowId", id).
 				Warnf("couldn't delete flow entry")
 		}
-		flowKeys = append(flowKeys, id)
-		flowsValues = append(flowsValues, metrics)
+		// We observed that eBFP PerCPU map might insert multiple times the same key in the map
+		// (probably due to race conditions) so we need to re-join metrics again at userspace
+		// TODO: instrument how many times the keys are is repeated in the same eviction
+		flows[id] = append(flows[id], metrics...)
 	}
-	return flowKeys, flowsValues
+	return flows
 }