NETOBSERV-1240: prefer flows with DNS records when checking for dup

Signed-off-by: msherif1234 <[email protected]>

pkg/flow/deduper.go

@@ -26,6 +26,7 @@ type deduperCache struct {
 
 type entry struct {
 	key        *ebpf.BpfFlowId
+	record     *Record
 	ifIndex    uint32
 	expiryTime time.Time
 }
@@ -46,14 +47,7 @@ func Dedupe(expireTime time.Duration, justMark bool) func(in <-chan []*Record, o
 			cache.removeExpired()
 			fwd := make([]*Record, 0, len(records))
 			for _, record := range records {
-				if cache.isDupe((*ebpf.BpfFlowId)(&record.Id)) {
-					if justMark {
-						record.Duplicate = true
-					} else {
-						continue
-					}
-				}
-				fwd = append(fwd, record)
+				cache.checkDupe(record, justMark, &fwd)
 			}
 			if len(fwd) > 0 {
 				out <- fwd
@@ -62,10 +56,9 @@ func Dedupe(expireTime time.Duration, justMark bool) func(in <-chan []*Record, o
 	}
 }
 
-// isDupe returns whether the passed record has been already checked for duplicate for
-// another interface
-func (c *deduperCache) isDupe(key *ebpf.BpfFlowId) bool {
-	rk := *key
+// checkDupe check current record if its already available nad if not added to fwd records list
+func (c *deduperCache) checkDupe(r *Record, justMark bool, fwd *[]*Record) {
+	rk := r.Id
 	// zeroes fields from key that should be ignored from the flow comparison
 	rk.IfIndex = 0
 	rk.SrcMac = [MacLen]uint8{0, 0, 0, 0, 0, 0}
@@ -79,17 +72,36 @@ func (c *deduperCache) isDupe(key *ebpf.BpfFlowId) bool {
 		c.entries.MoveToFront(ele)
 		// The input flow is duplicate if its interface is different to the interface
 		// of the non-duplicate flow that was first registered in the cache
-		return fEntry.ifIndex != key.IfIndex
+		// except if the new flow has DNS enrichment in this case we will
+		// forward both flows
+		if r.DNSLatency != 0 && fEntry.record.DNSLatency == 0 && fEntry.ifIndex != r.Id.IfIndex {
+			if justMark {
+				fEntry.record.Duplicate = true
+			} else {
+				fwd = findAndDeleteRecord(fwd, fEntry.record)
+			}
+			*fwd = append(*fwd, r)
+			return
+		}
+		if fEntry.ifIndex != r.Id.IfIndex {
+			if justMark {
+				r.Duplicate = true
+			}
+			return
+		}
+		*fwd = append(*fwd, r)
+		return
 	}
 	// The flow has not been accounted previously (or was forgotten after expiration)
 	// so we register it for that concrete interface
 	e := entry{
 		key:        &rk,
-		ifIndex:    key.IfIndex,
+		record:     r,
+		ifIndex:    r.Id.IfIndex,
 		expiryTime: timeNow().Add(c.expire),
 	}
 	c.ifaces[rk] = c.entries.PushFront(&e)
-	return false
+	*fwd = append(*fwd, r)
 }
 
 func (c *deduperCache) removeExpired() {
@@ -110,3 +122,14 @@ func (c *deduperCache) removeExpired() {
 		}).Debug("entries evicted from the deduper cache")
 	}
 }
+
+func findAndDeleteRecord(fwd *[]*Record, r *Record) *[]*Record {
+	s := *fwd
+	for i, v := range s {
+		if v == r {
+			s = append(s[:i], s[i+1:]...)
+			break
+		}
+	}
+	return &s
+}

pkg/flow/deduper_test.go

@@ -23,7 +23,7 @@ var (
 	}, Metrics: ebpf.BpfFlowMetrics{
 		Packets: 2, Bytes: 456, Flags: 1,
 	}}, Interface: "123456789"}
-	// another fow from 2 different interfaces and directions
+	// another flow from 2 different interfaces and directions
 	twoIf1 = &Record{RawRecord: RawRecord{Id: ebpf.BpfFlowId{
 		EthProtocol: 1, Direction: 1, SrcPort: 333, DstPort: 456,
 		DstMac: MacAddr{0x1}, SrcMac: MacAddr{0x1}, IfIndex: 1,
@@ -36,6 +36,19 @@ var (
 	}, Metrics: ebpf.BpfFlowMetrics{
 		Packets: 2, Bytes: 456, Flags: 1,
 	}}, Interface: "123456789"}
+	// another flow from 2 different interfaces and directions with DNS latency set on the latest
+	threeIf1 = &Record{RawRecord: RawRecord{Id: ebpf.BpfFlowId{
+		EthProtocol: 1, Direction: 1, SrcPort: 433, DstPort: 456,
+		DstMac: MacAddr{0x1}, SrcMac: MacAddr{0x1}, IfIndex: 1,
+	}, Metrics: ebpf.BpfFlowMetrics{
+		Packets: 2, Bytes: 456, Flags: 1,
+	}}, Interface: "eth0"}
+	threeIf2 = &Record{RawRecord: RawRecord{Id: ebpf.BpfFlowId{
+		EthProtocol: 1, Direction: 0, SrcPort: 433, DstPort: 456,
+		DstMac: MacAddr{0x2}, SrcMac: MacAddr{0x2}, IfIndex: 2,
+	}, Metrics: ebpf.BpfFlowMetrics{
+		Packets: 2, Bytes: 456, Flags: 1,
+	}}, Interface: "123456789", DNSLatency: time.Millisecond}
 )
 
 func TestDedupe(t *testing.T) {
@@ -45,15 +58,17 @@ func TestDedupe(t *testing.T) {
 	go Dedupe(time.Minute, false)(input, output)
 
 	input <- []*Record{
-		oneIf2, // record 1 at interface 2: should be accepted
-		twoIf1, // record 2 at interface 1: should be accepted
-		oneIf1, // record 1 duplicate at interface 1: should NOT be accepted
-		oneIf1, //                                        (same record key, different interface)
-		twoIf2, // record 2 duplicate at interface 2: should NOT be accepted
-		oneIf2, // record 1 at interface 1: should be accepted (same record key, same interface)
+		oneIf2,   // record 1 at interface 2: should be accepted
+		twoIf1,   // record 2 at interface 1: should be accepted
+		oneIf1,   // record 1 duplicate at interface 1: should NOT be accepted
+		oneIf1,   //                                        (same record key, different interface)
+		twoIf2,   // record 2 duplicate at interface 2: should NOT be accepted
+		oneIf2,   // record 1 at interface 1: should be accepted (same record key, same interface)
+		threeIf1, // record 1 has no DNS but its duplicate should NOT be accepted
+		threeIf2, // record 2 should be accepted
 	}
 	deduped := receiveTimeout(t, output)
-	assert.Equal(t, []*Record{oneIf2, twoIf1, oneIf2}, deduped)
+	assert.Equal(t, []*Record{oneIf2, twoIf1, oneIf2, threeIf2}, deduped)
 
 	// should still accept records with same key, same interface,
 	// and discard these with same key, different interface