NETOBSERV-578: deduplicate flows at agent level (#56)

* NETOBSERV-578: deduplicate flows at agent level

* Addressed code comments

Author: Mario Macias

cmd/netobserv-ebpf-agent.go

@@ -33,6 +33,9 @@ func main() {
 				Error("PProf HTTP listener stopped working")
 		}()
 	}
+	if config.DeduperFCExpiry == 0 {
+		config.DeduperFCExpiry = 2 * config.CacheActiveTimeout
+	}
 
 	logrus.WithField("configuration", fmt.Sprintf("%#v", config)).Debugf("configuration loaded")
 

docs/config.md

@@ -21,6 +21,14 @@ The following environment variables are available to configure the NetObserv eBF
   cache. If the accounter reaches the max number of flows, it flushes them to the collector.
 * `CACHE_ACTIVE_TIMEOUT` (default: `5s`). Duration string that specifies the maximum duration
   that flows are kept in the accounting cache before being flushed to the collector.
+* `DEDUPER` (default: `none`, disabled). Accepted values are `none` (disabled) and `firstCome`.
+  When enabled, it will detect duplicate flows (flows that have been detected e.g. through
+  both the physical and a virtual interface).
+  `firstCome` will forward only flows from the first interface the flows are received from.
+* `DEDUPER_FC_EXPIRY` (default: `2 * CACHE_ACTIVE_TIMEOUT`). Specifies the expiry duration of the `firstCome`
+  deduplicator. After a flow hasn't been received for that expiry time, the deduplicator forgets it.
+  That means that a flow from a connection that has been inactive during that period could be
+  forwarded again from a different interface.
 * `LOG_LEVEL` (default: `info`). From more to less verbose: `trace`, `debug`, `info`, `warn`,
   `error`, `fatal`, `panic`.
 * `KAFKA_BROKERS` (required if `EXPORT` is `kafka`). Comma-separated list of tha addresses of the

pkg/agent/agent.go

@@ -225,7 +225,13 @@ func (f *Flows) processRecords(tracedRecords <-chan []*flow.Record) *node.Termin
 	alog.Debug("registering exporter")
 	export := node.AsTerminal(f.exporter)
 	alog.Debug("connecting graph")
-	tracersCollector.SendsTo(export)
+	if f.cfg.Deduper == DeduperFirstCome {
+		deduper := node.AsMiddle(flow.Dedupe(f.cfg.DeduperFCExpiry))
+		tracersCollector.SendsTo(deduper)
+		deduper.SendsTo(export)
+	} else {
+		tracersCollector.SendsTo(export)
+	}
 	alog.Debug("starting graph")
 	tracersCollector.Start()
 	return export

pkg/agent/config.go

@@ -5,8 +5,10 @@ import (
 )
 
 const (
-	ListenPoll  = "poll"
-	ListenWatch = "watch"
+	ListenPoll       = "poll"
+	ListenWatch      = "watch"
+	DeduperNone      = "none"
+	DeduperFirstCome = "firstCome"
 )
 
 type Config struct {
@@ -36,6 +38,17 @@ type Config struct {
 	// CacheActiveTimeout specifies the maximum duration that flows are kept in the accounting
 	// cache before being flushed for its later export
 	CacheActiveTimeout time.Duration `env:"CACHE_ACTIVE_TIMEOUT" envDefault:"5s"`
+	// Deduper specifies the deduper type. Accepted values are "none" (disabled) and "firstCome".
+	// When enabled, it will detect duplicate flows (flows that have been detected e.g. through
+	// both the physical and a virtual interface).
+	// "firstCome" will forward only flows from the first interface the flows are received from.
+	Deduper string `env:"DEDUPER" envDefault:"none"`
+	// DeduperFCExpiry specifies the expiry duration of the flows "firstCome" deduplicator. After
+	// a flow hasn't been received for that expiry time, the deduplicator forgets it. That means
+	// that a flow from a connection that has been inactive during that period could be forwarded
+	// again from a different interface.
+	// If the value is not set, it will default to 2 * CacheActiveTimeout
+	DeduperFCExpiry time.Duration `env:"DEDUPER_FC_EXPIRY"`
 	// Logger level. From more to less verbose: trace, debug, info, warn, error, fatal, panic.
 	LogLevel string `env:"LOG_LEVEL" envDefault:"info"`
 	// Sampling holds the rate at which packets should be sampled and sent to the target collector.

pkg/flow/deduper.go

@@ -0,0 +1,100 @@
+package flow
+
+import (
+	"container/list"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+var dlog = logrus.WithField("component", "flow/Deduper")
+var timeNow = time.Now
+
+// deduperCache implement a LRU cache whose elements are evicted if they haven't been accessed
+// during the expire duration.
+// It is not safe for concurrent access.
+type deduperCache struct {
+	expire time.Duration
+	// key: RecordKey with the interface and MACs erased, to detect duplicates
+	// value: listElement pointing to a struct entry
+	ifaces map[RecordKey]*list.Element
+	// element: entry structs of the ifaces map ordered by expiry time
+	entries *list.List
+}
+
+type entry struct {
+	key        *RecordKey
+	ifIndex    uint32
+	expiryTime time.Time
+}
+
+// Dedupe receives flows and filters these belonging to duplicate interfaces. It will forward
+// the flows from the first interface coming to it, until that flow expires in the cache
+// (no activity for it during the expiration time)
+func Dedupe(expireTime time.Duration) func(in <-chan []*Record, out chan<- []*Record) {
+	cache := &deduperCache{
+		expire:  expireTime,
+		entries: list.New(),
+		ifaces:  map[RecordKey]*list.Element{},
+	}
+	return func(in <-chan []*Record, out chan<- []*Record) {
+		for records := range in {
+			cache.removeExpired()
+			fwd := make([]*Record, 0, len(records))
+			for _, record := range records {
+				if !cache.isDupe(&record.RecordKey) {
+					fwd = append(fwd, record)
+				}
+			}
+			if len(fwd) > 0 {
+				out <- fwd
+			}
+		}
+	}
+}
+
+// isDupe returns whether the passed record has been already checked for duplicate for
+// another interface
+func (c *deduperCache) isDupe(key *RecordKey) bool {
+	rk := *key
+	rk.IFIndex = 0
+	rk.DataLink = DataLink{}
+	// If a flow has been accounted previously, whatever its interface was,
+	// it updates the expiry time for that flow
+	if ele, ok := c.ifaces[rk]; ok {
+		fEntry := ele.Value.(*entry)
+		fEntry.expiryTime = timeNow().Add(c.expire)
+		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
+	}
+	// 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,
+		expiryTime: timeNow().Add(c.expire),
+	}
+	c.ifaces[rk] = c.entries.PushFront(&e)
+	return false
+}
+
+func (c *deduperCache) removeExpired() {
+	now := timeNow()
+	ele := c.entries.Back()
+	evicted := 0
+	for ele != nil && now.After(ele.Value.(*entry).expiryTime) {
+		evicted++
+		c.entries.Remove(ele)
+		delete(c.ifaces, *ele.Value.(*entry).key)
+		ele = c.entries.Back()
+	}
+	if evicted > 0 {
+		dlog.WithFields(logrus.Fields{
+			"current":    c.entries.Len(),
+			"evicted":    evicted,
+			"expiryTime": c.expire,
+		}).Debug("entries evicted from the deduper cache")
+	}
+}

pkg/flow/deduper_test.go

@@ -0,0 +1,106 @@
+package flow
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	// the same flow from 2 different interfaces
+	oneIf1 = &Record{RawRecord: RawRecord{RecordKey: RecordKey{
+		EthProtocol: 1, Direction: 1, Transport: Transport{SrcPort: 123, DstPort: 456},
+		DataLink: DataLink{DstMac: MacAddr{0x1}, SrcMac: MacAddr{0x1}}, IFIndex: 1,
+	}, RecordMetrics: RecordMetrics{
+		Packets: 2, Bytes: 456,
+	}}, Interface: "eth0"}
+	oneIf2 = &Record{RawRecord: RawRecord{RecordKey: RecordKey{
+		EthProtocol: 1, Direction: 1, Transport: Transport{SrcPort: 123, DstPort: 456},
+		DataLink: DataLink{DstMac: MacAddr{0x2}, SrcMac: MacAddr{0x2}}, IFIndex: 2,
+	}, RecordMetrics: RecordMetrics{
+		Packets: 2, Bytes: 456,
+	}}, Interface: "123456789"}
+	// another fow from 2 different interfaces
+	twoIf1 = &Record{RawRecord: RawRecord{RecordKey: RecordKey{
+		EthProtocol: 1, Direction: 1, Transport: Transport{SrcPort: 333, DstPort: 456},
+		DataLink: DataLink{DstMac: MacAddr{0x1}, SrcMac: MacAddr{0x1}}, IFIndex: 1,
+	}, RecordMetrics: RecordMetrics{
+		Packets: 2, Bytes: 456,
+	}}, Interface: "eth0"}
+	twoIf2 = &Record{RawRecord: RawRecord{RecordKey: RecordKey{
+		EthProtocol: 1, Direction: 1, Transport: Transport{SrcPort: 333, DstPort: 456},
+		DataLink: DataLink{DstMac: MacAddr{0x2}, SrcMac: MacAddr{0x2}}, IFIndex: 2,
+	}, RecordMetrics: RecordMetrics{
+		Packets: 2, Bytes: 456,
+	}}, Interface: "123456789"}
+)
+
+func TestDedupe(t *testing.T) {
+	input := make(chan []*Record, 100)
+	output := make(chan []*Record, 100)
+
+	go Dedupe(time.Minute)(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)
+	}
+	deduped := receiveTimeout(t, output)
+	assert.Equal(t, []*Record{oneIf2, twoIf1, oneIf2}, deduped)
+
+	// should still accept records with same key, same interface,
+	// and discard these with same key, different interface
+	input <- []*Record{oneIf1, oneIf2}
+	deduped = receiveTimeout(t, output)
+	assert.Equal(t, []*Record{oneIf2}, deduped)
+}
+
+func TestDedupe_EvictFlows(t *testing.T) {
+	tm := &timerMock{now: time.Now()}
+	timeNow = tm.Now
+	input := make(chan []*Record, 100)
+	output := make(chan []*Record, 100)
+
+	go Dedupe(15*time.Second)(input, output)
+
+	// Should only accept records 1 and 2, at interface 1
+	input <- []*Record{oneIf1, twoIf1, oneIf2}
+	assert.Equal(t, []*Record{oneIf1, twoIf1},
+		receiveTimeout(t, output))
+
+	tm.now = tm.now.Add(10 * time.Second)
+
+	// After 10 seconds, it still filters existing flows from different interfaces
+	input <- []*Record{oneIf2}
+	time.Sleep(100 * time.Millisecond)
+	requireNoEviction(t, output)
+
+	tm.now = tm.now.Add(10 * time.Second)
+
+	// Record 2 hasn't been accounted for >expiryTime, so it will accept the it again
+	// whatever the interface.
+	// Since record 1 was accessed 10 seconds ago (<expiry time) it will filter it
+	input <- []*Record{oneIf2, twoIf2, twoIf1}
+	assert.Equal(t, []*Record{twoIf2},
+		receiveTimeout(t, output))
+
+	tm.now = tm.now.Add(20 * time.Second)
+
+	// when all the records expire, the deduper is reset for that flow
+	input <- []*Record{oneIf2, twoIf2}
+	assert.Equal(t, []*Record{oneIf2, twoIf2},
+		receiveTimeout(t, output))
+}
+
+type timerMock struct {
+	now time.Time
+}
+
+func (tm *timerMock) Now() time.Time {
+	return tm.now
+}