Update Go and go-pipes version (#80)

Author: Mario Macias

go.mod

@@ -1,13 +1,13 @@
 module github.com/netobserv/netobserv-ebpf-agent
 
-go 1.17
+go 1.18
 
 require (
 	github.com/caarlos0/env/v6 v6.9.1
 	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.2.0
+	github.com/netobserv/gopipes v0.3.0
 	github.com/paulbellamy/ratecounter v0.2.0
 	github.com/segmentio/kafka-go v0.4.35
 	github.com/sirupsen/logrus v1.8.1

go.sum

@@ -64,7 +64,7 @@ type Flows struct {
 	mapTracer *flow.MapTracer
 	rbTracer  *flow.RingBufTracer
 	accounter *flow.Accounter
-	exporter  flowExporter
+	exporter  node.TerminalFunc[[]*flow.Record]
 
 	// elements used to decorate flows with extra information
 	interfaceNamer flow.InterfaceNamer
@@ -82,10 +82,6 @@ type ebpfFlowFetcher interface {
 	ReadRingBuf() (ringbuf.Record, error)
 }
 
-// flowExporter abstract the ExportFlows' method of exporter.GRPCProto to allow dependency injection
-// in tests
-type flowExporter func(in <-chan []*flow.Record)
-
 // FlowsAgent instantiates a new agent, given a configuration.
 func FlowsAgent(cfg *Config) (*Flows, error) {
 	alog.Info("initializing Flows agent")
@@ -138,7 +134,7 @@ func FlowsAgent(cfg *Config) (*Flows, error) {
 func flowsAgent(cfg *Config,
 	informer ifaces.Informer,
 	fetcher ebpfFlowFetcher,
-	exporter flowExporter,
+	exporter node.TerminalFunc[[]*flow.Record],
 	agentIP net.IP,
 ) (*Flows, error) {
 	// configure allow/deny interfaces filter
@@ -189,7 +185,7 @@ func flowDirections(cfg *Config) (ingress, egress bool) {
 	}
 }
 
-func buildFlowExporter(cfg *Config) (flowExporter, error) {
+func buildFlowExporter(cfg *Config) (node.TerminalFunc[[]*flow.Record], error) {
 	switch cfg.Export {
 	case "grpc":
 		if cfg.TargetHost == "" || cfg.TargetPort == 0 {
@@ -342,7 +338,7 @@ func (f *Flows) interfacesManager(ctx context.Context) error {
 
 // buildAndStartPipeline creates the ETL flow processing graph.
 // For a more visual view, check the docs/architecture.md document.
-func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal, error) {
+func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal[[]*flow.Record], error) {
 
 	alog.Debug("registering interfaces' listener in background")
 	err := f.interfacesManager(ctx)
@@ -351,8 +347,8 @@ func (f *Flows) buildAndStartPipeline(ctx context.Context) (*node.Terminal, erro
 	}
 
 	alog.Debug("connecting flows' processing graph")
-	mapTracer := node.AsInit(f.mapTracer.TraceLoop(ctx))
-	rbTracer := node.AsInit(f.rbTracer.TraceLoop(ctx))
+	mapTracer := node.AsStart(f.mapTracer.TraceLoop(ctx))
+	rbTracer := node.AsStart(f.rbTracer.TraceLoop(ctx))
 
 	accounter := node.AsMiddle(f.accounter.Account,
 		node.ChannelBufferLen(f.cfg.BuffersLength))

pkg/agent/agent.go

@@ -41,7 +41,7 @@ func (m *MapTracer) Flush() {
 	m.evictionCond.Broadcast()
 }
 
-func (m *MapTracer) TraceLoop(ctx context.Context) node.InitFunc {
+func (m *MapTracer) TraceLoop(ctx context.Context) node.StartFunc[[]*Record] {
 	return func(out chan<- []*Record) {
 		evictionTicker := time.NewTicker(m.evictionTimeout)
 		go m.evictionSynchronization(ctx, out)

pkg/flow/tracer_map.go

@@ -51,7 +51,7 @@ func NewRingBufTracer(
 	}
 }
 
-func (m *RingBufTracer) TraceLoop(ctx context.Context) node.InitFunc {
+func (m *RingBufTracer) TraceLoop(ctx context.Context) node.StartFunc[*RawRecord] {
 	return func(out chan<- *RawRecord) {
 		debugging := logrus.IsLevelEnabled(logrus.DebugLevel)
 		for {