[client] introduce client-side event aggregation (#6627)

* added an implementation of aggregating memory store Signed-off-by: Dmitri <dmitri.external@netbird.io> * initial support for aggregation of events Signed-off-by: Dmitri <dmitri.external@netbird.io> * added tcp-aggregation test Signed-off-by: Dmitri <dmitri.external@netbird.io> * added manager integration test Signed-off-by: Dmitri <dmitri.external@netbird.io> * added tracking of the number of start-, drop, and end-events in an aggregation window Signed-off-by: Dmitri <dmitri.external@netbird.io> * fixes based on sonarcube checks Signed-off-by: Dmitri <dmitri.external@netbird.io> * regenerated proto files Signed-off-by: Dmitri <dmitri.external@netbird.io> * removed inadvertenly added google proto files Signed-off-by: Dmitri <dmitri.external@netbird.io> * pacifying linter Signed-off-by: Dmitri <dmitri.external@netbird.io> * update test to validate event aggregation over tcp, udp, icmp, and icmpv6 Signed-off-by: Dmitri <dmitri.external@netbird.io> * updated event aggregation test Signed-off-by: Dmitri <dmitri.external@netbird.io> * regenerate protobufs with expected versions of protoc and protoc-gen-go Signed-off-by: Dmitri <dmitri.external@netbird.io> * remove protoc/protoc-gen headers from flow_grpc.pb.go Signed-off-by: Dmitri <dmitri.external@netbird.io> * updated openapi spec Signed-off-by: Dmitri <dmitri.external@netbird.io> * updated openapi NetworkTrafficEvent spec, regenerated types Signed-off-by: Dmitri <dmitri.external@netbird.io> * respond to feedback Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * fixed an issue with how we track events that shouldn't be aggregated Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * fixed mapping of events to protobuf Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * icmp code values in aggregated events do not matter Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * regenerate openapi types Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * added a comment re: unbounded unacked events Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * reset aggregated event type to unknown Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * fix event aggregation test Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * used the source port of the earliest event Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * add tracking of window starts and ends Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * updated openapi spec Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * reverted changes to generate.sh Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * cleanup handling of not-aggregated events + test Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * responded to feedback + small fixes Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * small fix in a test Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * another test Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * force setting non-empty rule id on aggregated events Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * fixed a couple of issues flagged by coderabbit Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * fix spelling Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> * handle exhausted retry backoffs Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io> --------- Signed-off-by: Dmitri <dmitri.external@netbird.io> Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>
2026-10-07 22:19:08 +02:00 · 2026-07-06 16:04:26 +02:00
parent 91acb8147c
commit 47352e6e45
12 changed files with 1081 additions and 149 deletions
@@ -9,12 +9,14 @@ import (
 	"sync"
 	"time"

+	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/types/known/timestamppb"

 	"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
 	"github.com/netbirdio/netbird/client/internal/netflow/logger"
+	"github.com/netbirdio/netbird/client/internal/netflow/store"
 	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/flow/client"
@@ -23,14 +25,16 @@ import (

 // Manager handles netflow tracking and logging
 type Manager struct {
-	mux            sync.Mutex
-	shutdownWg     sync.WaitGroup
-	logger         nftypes.FlowLogger
-	flowConfig     *nftypes.FlowConfig
-	conntrack      nftypes.ConnTracker
-	receiverClient *client.GRPCClient
-	publicKey      []byte
-	cancel         context.CancelFunc
+	mux               sync.Mutex
+	shutdownWg        sync.WaitGroup
+	logger            nftypes.FlowLogger
+	flowConfig        *nftypes.FlowConfig
+	conntrack         nftypes.ConnTracker
+	receiverClient    *client.GRPCClient
+	eventsWithoutAcks nftypes.Store
+	publicKey         []byte
+	cancel            context.CancelFunc
+	retryInterval     time.Duration
 }

 // NewManager creates a new netflow manager
@@ -48,9 +52,11 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
 	}

 	return &Manager{
-		logger:    flowLogger,
-		conntrack: ct,
-		publicKey: publicKey,
+		logger:            flowLogger,
+		conntrack:         ct,
+		publicKey:         publicKey,
+		retryInterval:     time.Second,
+		eventsWithoutAcks: store.NewMemoryStore(),
 	}
 }

@@ -66,6 +72,7 @@ func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
 }

 // enableFlow starts components for flow tracking
+// must be called under m.mux lock
 func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	// first make sender ready so events don't pile up
 	if m.needsNewClient(previous) {
@@ -85,6 +92,7 @@ func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
 	return nil
 }

+// must be called under m.mux lock
 func (m *Manager) resetClient() error {
 	if m.receiverClient != nil {
 		if err := m.receiverClient.Close(); err != nil {
@@ -107,14 +115,19 @@ func (m *Manager) resetClient() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	m.cancel = cancel

-	m.shutdownWg.Add(2)
+	m.shutdownWg.Add(3)
+	flowConfigInterval := m.flowConfig.Interval
 	go func() {
 		defer m.shutdownWg.Done()
-		m.receiveACKs(ctx, flowClient)
+		m.receiveACKs(ctx, flowClient, flowConfigInterval)
 	}()
 	go func() {
 		defer m.shutdownWg.Done()
-		m.startSender(ctx)
+		m.startSender(ctx, flowConfigInterval)
+	}()
+	go func() {
+		defer m.shutdownWg.Done()
+		m.startRetries(ctx, flowConfigInterval)
 	}()

 	return nil
@@ -198,8 +211,8 @@ func (m *Manager) GetLogger() nftypes.FlowLogger {
 	return m.logger
 }

-func (m *Manager) startSender(ctx context.Context) {
-	ticker := time.NewTicker(m.flowConfig.Interval)
+func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Duration) {
+	ticker := time.NewTicker(flowConfigInterval)
 	defer ticker.Stop()

 	for {
@@ -207,27 +220,29 @@ func (m *Manager) startSender(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			events := m.logger.GetEvents()
+			collectedEvents := m.logger.ResetAggregationWindow()
+			events := collectedEvents.GetAggregatedEvents()
 			for _, event := range events {
+				m.eventsWithoutAcks.StoreEvent(event)
 				if err := m.send(event); err != nil {
 					log.Errorf("failed to send flow event to server: %v", err)
-					continue
+				} else {
+					log.Tracef("sent flow event: %s", event.ID)
 				}
-				log.Tracef("sent flow event: %s", event.ID)
 			}
 		}
 	}
 }

-func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
-	err := client.Receive(ctx, m.flowConfig.Interval, func(ack *proto.FlowEventAck) error {
+func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, flowConfigInterval time.Duration) {
+	err := client.Receive(ctx, flowConfigInterval, func(ack *proto.FlowEventAck) error {
 		id, err := uuid.FromBytes(ack.EventId)
 		if err != nil {
 			log.Warnf("failed to convert ack event id to uuid: %v", err)
 			return nil
 		}
 		log.Tracef("received flow event ack: %s", id)
-		m.logger.DeleteEvents([]uuid.UUID{id})
+		m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
 		return nil
 	})

@@ -236,6 +251,51 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
 	}
 }

+// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
+// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
+func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Duration) {
+	timer := time.NewTimer(m.retryInterval)
+	retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
+		InitialInterval:     1 * time.Second,
+		RandomizationFactor: 0.5,
+		Multiplier:          1.7,
+		MaxInterval:         flowConfigInterval / 2,
+		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}, ctx)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			resetBackoff := true
+			for _, e := range m.eventsWithoutAcks.GetEvents() {
+				if e.Timestamp.Add(time.Second).After(time.Now()) {
+					// grace period on retries to avoid early retries
+					// do not retry if the event is less than 1 sec old
+					continue
+				}
+				if err := m.send(e); err != nil {
+					if nextBackoff := retryBackoff.NextBackOff(); nextBackoff != backoff.Stop {
+						timer = time.NewTimer(nextBackoff)
+						resetBackoff = false
+					} else {
+						resetBackoff = true // we exhausted retries, reset retry loop
+					}
+					break
+				}
+			}
+			if resetBackoff { // use regular retry interval in absence of network errors
+				retryBackoff.Reset()
+				timer = time.NewTimer(m.retryInterval)
+			}
+		}
+	}
+}
+
 func (m *Manager) send(event *nftypes.Event) error {
 	m.mux.Lock()
 	client := m.receiverClient
@@ -250,9 +310,11 @@ func (m *Manager) send(event *nftypes.Event) error {

 func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 	protoEvent := &proto.FlowEvent{
-		EventId:   event.ID[:],
-		Timestamp: timestamppb.New(event.Timestamp),
-		PublicKey: publicKey,
+		EventId:     event.ID[:],
+		Timestamp:   timestamppb.New(event.Timestamp),
+		PublicKey:   publicKey,
+		WindowStart: timestamppb.New(event.WindowStart),
+		WindowEnd:   timestamppb.New(event.WindowEnd),
 		FlowFields: &proto.FlowFields{
 			FlowId:           event.FlowID[:],
 			RuleId:           event.RuleID,
@@ -267,6 +329,9 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
 			TxBytes:          event.TxBytes,
 			SourceResourceId: event.SourceResourceID,
 			DestResourceId:   event.DestResourceID,
+			NumOfStarts:      event.NumOfStarts,
+			NumOfEnds:        event.NumOfEnds,
+			NumOfDrops:       event.NumOfDrops,
 		},
 	}