Code improvements (#60)

Author: Mario Macias

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/cilium/ebpf v0.8.1
 	github.com/gavv/monotime v0.0.0-20190418164738-30dba4353424
 	github.com/mariomac/guara v0.0.0-20220523124851-5fc279816f1f
-	github.com/netobserv/gopipes v0.1.1
+	github.com/netobserv/gopipes v0.2.0
 	github.com/paulbellamy/ratecounter v0.2.0
 	github.com/segmentio/kafka-go v0.4.32
 	github.com/sirupsen/logrus v1.8.1

go.sum

@@ -372,8 +372,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8m
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mxk/go-flowrate v0.0.0-20140419014527-cca7078d478f/go.mod h1:ZdcZmHo+o7JKHSa8/e818NopupXU1YMK5fe1lsApnBw=
-github.com/netobserv/gopipes v0.1.1 h1:f8zJsvnMgRFRa2B+1siwRtW0Y4dqeBROmkcI/HgT1gE=
-github.com/netobserv/gopipes v0.1.1/go.mod h1:eGoHZW1ON8Dx/zmDXUhsbVNqatPjtpdO0UZBmGZGmVI=
+github.com/netobserv/gopipes v0.2.0 h1:CnJQq32+xNuM85eVYy/HOf+StTJdh2K6RdaEg7NAJDg=
+github.com/netobserv/gopipes v0.2.0/go.mod h1:eGoHZW1ON8Dx/zmDXUhsbVNqatPjtpdO0UZBmGZGmVI=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=

pkg/agent/agent.go

@@ -168,11 +168,10 @@ func flowDirections(cfg *Config) (ingress, egress bool) {
 // until the passed context is canceled
 func (f *Flows) Run(ctx context.Context) error {
 	alog.Info("starting Flows agent")
-	tracedRecords, err := f.interfacesManager(ctx)
+	graph, err := f.processPipeline(ctx)
 	if err != nil {
-		return err
+		return fmt.Errorf("starting processing graph: %w", err)
 	}
-	graph := f.processRecords(tracedRecords)
 
 	alog.Info("Flows agent successfully started")
 	<-ctx.Done()
@@ -187,7 +186,7 @@ func (f *Flows) Run(ctx context.Context) error {
 
 // interfacesManager uses an informer to check new/deleted network interfaces. For each running
 // interface, it registers a flow tracer that will forward new flows to the returned channel
-func (f *Flows) interfacesManager(ctx context.Context) (<-chan []*flow.Record, error) {
+func (f *Flows) interfacesManager(ctx context.Context) (node.InitFunc, error) {
 	slog := alog.WithField("function", "interfacesManager")
 
 	slog.Debug("subscribing for network interface events")
@@ -196,19 +195,14 @@ func (f *Flows) interfacesManager(ctx context.Context) (<-chan []*flow.Record, e
 		return nil, fmt.Errorf("instantiating interfaces' informer: %w", err)
 	}
 
-	tracedRecords := make(chan []*flow.Record, f.cfg.BuffersLength)
-
 	tctx, cancelTracer := context.WithCancel(ctx)
-	go f.tracer.Trace(tctx, tracedRecords)
-
 	go func() {
 		for {
 			select {
 			case <-ctx.Done():
 				slog.Debug("canceling flow tracer")
 				cancelTracer()
 				slog.Debug("closing channel and exiting internal goroutine")
-				close(tracedRecords)
 				return
 			case event := <-ifaceEvents:
 				slog.WithField("event", event).Debug("received event")
@@ -225,33 +219,36 @@ func (f *Flows) interfacesManager(ctx context.Context) (<-chan []*flow.Record, e
 		}
 	}()
 
-	return tracedRecords, nil
+	return func(out chan<- []*flow.Record) {
+		f.tracer.Trace(tctx, out)
+	}, nil
 }
 
-// processRecords creates the tracers --> accounter --> forwarder Flow processing graph
-func (f *Flows) processRecords(tracedRecords <-chan []*flow.Record) *node.Terminal {
-	// The start node receives Records from the eBPF flow tracers. Currently it is just an external
-	// channel forwarder, as the Pipes library does not yet accept
-	// adding/removing nodes dynamically: https://github.com/mariomac/pipes/issues/5
+// processPipeline creates the tracers --> accounter --> forwarder Flow processing graph
+func (f *Flows) processPipeline(ctx context.Context) (*node.Terminal, error) {
+
 	alog.Debug("registering tracers' input")
-	tracersCollector := node.AsInit(func(out chan<- []*flow.Record) {
-		for i := range tracedRecords {
-			out <- i
-		}
-	})
+	tracedRecords, err := f.interfacesManager(ctx)
+	if err != nil {
+		return nil, err
+	}
+	tracersCollector := node.AsInit(tracedRecords)
+
 	alog.Debug("registering exporter")
-	export := node.AsTerminal(f.exporter)
+	export := node.AsTerminal(f.exporter,
+		node.ChannelBufferLen(f.cfg.BuffersLength))
 	alog.Debug("connecting graph")
 	if f.cfg.Deduper == DeduperFirstCome {
-		deduper := node.AsMiddle(flow.Dedupe(f.cfg.DeduperFCExpiry))
+		deduper := node.AsMiddle(flow.Dedupe(f.cfg.DeduperFCExpiry),
+			node.ChannelBufferLen(f.cfg.BuffersLength))
 		tracersCollector.SendsTo(deduper)
 		deduper.SendsTo(export)
 	} else {
 		tracersCollector.SendsTo(export)
 	}
 	alog.Debug("starting graph")
 	tracersCollector.Start()
-	return export
+	return export, nil
 }
 
 func (f *Flows) onInterfaceAdded(iface ifaces.Interface) {

pkg/ebpf/tracer.go

@@ -8,6 +8,7 @@ import (
 	"io/fs"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/cilium/ebpf/ringbuf"
@@ -296,10 +297,13 @@ func (m *FlowTracer) pollAndForwardAggregateFlows(ctx context.Context, forwardFl
 		<-ctx.Done()
 	}()
 	go func() {
+		// flow eviction loop. It just keeps waiting for eviction until someone triggers the
+		// flowsEvictor.Broadcast signal
 		for {
 			// make sure we only evict once at a time, even if there are multiple eviction signals
 			m.flowsEvictor.L.Lock()
 			m.flowsEvictor.Wait()
+			tlog.Debug("eviction signal received")
 			m.evictFlows(tlog, forwardFlows)
 			m.flowsEvictor.L.Unlock()
 
@@ -315,13 +319,13 @@ func (m *FlowTracer) pollAndForwardAggregateFlows(ctx context.Context, forwardFl
 	for {
 		select {
 		case <-ctx.Done():
-			tlog.Debug("evicting flows after context cancelation")
+			tlog.Debug("triggering flow eviction after context cancelation")
 			m.flowsEvictor.Broadcast()
 			ticker.Stop()
 			tlog.Debug("exiting monitor")
 			return
 		case <-ticker.C:
-			tlog.Debug("evicting flows on timer")
+			tlog.Debug("triggering flow eviction on timer")
 			m.flowsEvictor.Broadcast()
 		}
 	}
@@ -372,6 +376,8 @@ func (m *FlowTracer) Trace(ctx context.Context, forwardFlows chan<- []*flow.Reco
 func (m *FlowTracer) listenAndForwardRingBuffer(ctx context.Context, forwardFlows chan<- []*flow.Record) {
 	flowAccount := make(chan *flow.RawRecord, m.buffersLength)
 	go m.accounter.Account(flowAccount, forwardFlows)
+	isForwarding := int32(0)
+	forwardedFlows := int32(0)
 	for {
 		select {
 		case <-ctx.Done():
@@ -393,8 +399,10 @@ func (m *FlowTracer) listenAndForwardRingBuffer(ctx context.Context, forwardFlow
 				log.WithError(err).Warn("reading ringbuf event")
 				continue
 			}
-			log.WithField("direction", readFlow.Direction).
-				Debug("received flow from ringbuffer. Evicting in-memory maps to leave free space")
+			if logrus.IsLevelEnabled(logrus.DebugLevel) {
+				m.logRingBufferFlows(&forwardedFlows, &isForwarding)
+			}
+			// forces a flow's eviction to leave room for new flows in the ebpf cache
 			m.flowsEvictor.Broadcast()
 
 			// Will need to send it to accounter anyway to account regardless of complete/ongoing flow
@@ -403,12 +411,30 @@ func (m *FlowTracer) listenAndForwardRingBuffer(ctx context.Context, forwardFlow
 	}
 }
 
+// logRingBufferFlows avoids flooding logs on long series of evicted flows by grouping how
+// many flows are forwarded
+func (m *FlowTracer) logRingBufferFlows(forwardedFlows, isForwarding *int32) {
+	atomic.AddInt32(forwardedFlows, 1)
+	if atomic.CompareAndSwapInt32(isForwarding, 0, 1) {
+		go func() {
+			time.Sleep(m.evictionTimeout)
+			log.WithFields(logrus.Fields{
+				"flows":         atomic.LoadInt32(forwardedFlows),
+				"cacheMaxFlows": m.cacheMaxSize,
+			}).Debug("received flows via ringbuffer. You might want to increase the CACHE_MAX_FLOWS value")
+			atomic.StoreInt32(forwardedFlows, 0)
+			atomic.StoreInt32(isForwarding, 0)
+		}()
+	}
+}
+
 // For synchronization purposes, we get/delete a whole snapshot of the flows map.
 // This way we avoid missing packets that could be updated on the
 // ebpf side while we process/aggregate them here
 // Changing this method invocation by BatchLookupAndDelete could improve performance
 // TODO: detect whether BatchLookupAndDelete is supported (Kernel>=5.6) and use it selectively
 // Supported Lookup/Delete operations by kernel: https://github.com/iovisor/bcc/blob/master/docs/kernel-versions.md
+// Race conditions here causes that some flows are lost in high-load scenarios
 func (m *FlowTracer) lookupAndDeleteFlowsMap() map[flow.RecordKey][]flow.RecordMetrics {
 	flowMap := m.objects.AggregatedFlows
 

pkg/flow/account.go

@@ -52,6 +52,8 @@ func (c *Accounter) Account(in <-chan *RawRecord, out chan<- []*Record) {
 			}
 			evictingEntries := c.entries
 			c.entries = make(map[RecordKey]*RecordMetrics, c.maxEntries)
+			logrus.WithField("flows", len(evictingEntries)).
+				Debug("evicting flows from userspace accounter on timeout")
 			go c.evict(evictingEntries, out)
 		case record, ok := <-in:
 			if !ok {
@@ -69,12 +71,13 @@ func (c *Accounter) Account(in <-chan *RawRecord, out chan<- []*Record) {
 				if len(c.entries) >= c.maxEntries {
 					evictingEntries := c.entries
 					c.entries = make(map[RecordKey]*RecordMetrics, c.maxEntries)
+					logrus.WithField("flows", len(evictingEntries)).
+						Debug("evicting flows from userspace accounter after reaching cache max length")
 					go c.evict(evictingEntries, out)
 				}
 				c.entries[record.RecordKey] = &record.RecordMetrics
 			}
 		}
-
 	}
 }
 

vendor/github.com/netobserv/gopipes/pkg/node/node.go

@@ -106,7 +106,7 @@ github.com/modern-go/reflect2
 # github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
 ## explicit
 github.com/munnerz/goautoneg
-# github.com/netobserv/gopipes v0.1.1
+# github.com/netobserv/gopipes v0.2.0
 ## explicit; go 1.17
 github.com/netobserv/gopipes/pkg/internal/connect
 github.com/netobserv/gopipes/pkg/internal/refl