[client,management] add netflow support to client and update management (#3414)

adds NetFlow functionality to track and log network traffic information between peers, with features including:

- Flow logging for TCP, UDP, and ICMP traffic
- Integration with connection tracking system
- Resource ID tracking in NetFlow events
- DNS and exit node collection configuration
- Flow API and Redis cache in management
- Memory-based flow storage implementation
- Kernel conntrack counters and userspace counters
- TCP state machine improvements for more accurate tracking
- Migration from net.IP to netip.Addr in the userspace firewall

client/internal/netflow/conntrack/conntrack.go

@@ -0,0 +1,306 @@
+//go:build linux && !android
+
+package conntrack
+
+import (
+	"encoding/binary"
+	"fmt"
+	"net/netip"
+	"sync"
+
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+	nfct "github.com/ti-mo/conntrack"
+	"github.com/ti-mo/netfilter"
+
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+)
+
+const defaultChannelSize = 100
+
+// ConnTrack manages kernel-based conntrack events
+type ConnTrack struct {
+	flowLogger nftypes.FlowLogger
+	iface      nftypes.IFaceMapper
+
+	conn *nfct.Conn
+	mux  sync.Mutex
+
+	instanceID     uuid.UUID
+	started        bool
+	done           chan struct{}
+	sysctlModified bool
+}
+
+// New creates a new connection tracker that interfaces with the kernel's conntrack system
+func New(flowLogger nftypes.FlowLogger, iface nftypes.IFaceMapper) *ConnTrack {
+	return &ConnTrack{
+		flowLogger: flowLogger,
+		iface:      iface,
+		instanceID: uuid.New(),
+		started:    false,
+		done:       make(chan struct{}, 1),
+	}
+}
+
+// Start begins tracking connections by listening for conntrack events. This method is idempotent.
+func (c *ConnTrack) Start(enableCounters bool) error {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	if c.started {
+		return nil
+	}
+
+	log.Info("Starting conntrack event listening")
+
+	if enableCounters {
+		c.EnableAccounting()
+	}
+
+	conn, err := nfct.Dial(nil)
+	if err != nil {
+		return fmt.Errorf("dial conntrack: %w", err)
+	}
+	c.conn = conn
+
+	events := make(chan nfct.Event, defaultChannelSize)
+	errChan, err := conn.Listen(events, 1, []netfilter.NetlinkGroup{
+		netfilter.GroupCTNew,
+		netfilter.GroupCTDestroy,
+	})
+
+	if err != nil {
+		if err := c.conn.Close(); err != nil {
+			log.Errorf("Error closing conntrack connection: %v", err)
+		}
+		c.conn = nil
+		return fmt.Errorf("start conntrack listener: %w", err)
+	}
+
+	c.started = true
+
+	go c.receiverRoutine(events, errChan)
+
+	return nil
+}
+
+func (c *ConnTrack) receiverRoutine(events chan nfct.Event, errChan chan error) {
+	for {
+		select {
+		case event := <-events:
+			c.handleEvent(event)
+		case err := <-errChan:
+			log.Errorf("Error from conntrack event listener: %v", err)
+			if err := c.conn.Close(); err != nil {
+				log.Errorf("Error closing conntrack connection: %v", err)
+			}
+			return
+		case <-c.done:
+			return
+		}
+	}
+}
+
+// Stop stops the connection tracking. This method is idempotent.
+func (c *ConnTrack) Stop() {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	if !c.started {
+		return
+	}
+
+	log.Info("Stopping conntrack event listening")
+
+	select {
+	case c.done <- struct{}{}:
+	default:
+	}
+
+	if c.conn != nil {
+		if err := c.conn.Close(); err != nil {
+			log.Errorf("Error closing conntrack connection: %v", err)
+		}
+		c.conn = nil
+	}
+
+	c.started = false
+
+	c.RestoreAccounting()
+}
+
+// Close stops listening for events and cleans up resources
+func (c *ConnTrack) Close() error {
+	c.mux.Lock()
+	defer c.mux.Unlock()
+
+	if c.started {
+		select {
+		case c.done <- struct{}{}:
+		default:
+		}
+	}
+
+	if c.conn != nil {
+		err := c.conn.Close()
+		c.conn = nil
+		c.started = false
+
+		c.RestoreAccounting()
+
+		if err != nil {
+			return fmt.Errorf("close conntrack: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// handleEvent processes incoming conntrack events
+func (c *ConnTrack) handleEvent(event nfct.Event) {
+	if event.Flow == nil {
+		return
+	}
+
+	if event.Type != nfct.EventNew && event.Type != nfct.EventDestroy {
+		return
+	}
+
+	flow := *event.Flow
+
+	proto := nftypes.Protocol(flow.TupleOrig.Proto.Protocol)
+	if proto == nftypes.ProtocolUnknown {
+		return
+	}
+	srcIP := flow.TupleOrig.IP.SourceAddress
+	dstIP := flow.TupleOrig.IP.DestinationAddress
+
+	if !c.relevantFlow(srcIP, dstIP) {
+		return
+	}
+
+	var srcPort, dstPort uint16
+	var icmpType, icmpCode uint8
+
+	switch proto {
+	case nftypes.TCP, nftypes.UDP, nftypes.SCTP:
+		srcPort = flow.TupleOrig.Proto.SourcePort
+		dstPort = flow.TupleOrig.Proto.DestinationPort
+	case nftypes.ICMP:
+		icmpType = flow.TupleOrig.Proto.ICMPType
+		icmpCode = flow.TupleOrig.Proto.ICMPCode
+	}
+
+	flowID := c.getFlowID(flow.ID)
+	direction := c.inferDirection(srcIP, dstIP)
+
+	eventType := nftypes.TypeStart
+	eventStr := "New"
+
+	if event.Type == nfct.EventDestroy {
+		eventType = nftypes.TypeEnd
+		eventStr = "Ended"
+	}
+
+	log.Tracef("%s %s %s connection: %s:%d -> %s:%d", eventStr, direction, proto, srcIP, srcPort, dstIP, dstPort)
+
+	c.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:     flowID,
+		Type:       eventType,
+		Direction:  direction,
+		Protocol:   proto,
+		SourceIP:   srcIP,
+		DestIP:     dstIP,
+		SourcePort: srcPort,
+		DestPort:   dstPort,
+		ICMPType:   icmpType,
+		ICMPCode:   icmpCode,
+		RxPackets:  c.mapRxPackets(flow, direction),
+		TxPackets:  c.mapTxPackets(flow, direction),
+		RxBytes:    c.mapRxBytes(flow, direction),
+		TxBytes:    c.mapTxBytes(flow, direction),
+	})
+}
+
+// relevantFlow checks if the flow is related to the specified interface
+func (c *ConnTrack) relevantFlow(srcIP, dstIP netip.Addr) bool {
+	// TODO: filter traffic by interface
+
+	wgnet := c.iface.Address().Network
+	if !wgnet.Contains(srcIP.AsSlice()) && !wgnet.Contains(dstIP.AsSlice()) {
+		return false
+	}
+
+	return true
+}
+
+// mapRxPackets maps packet counts to RX based on flow direction
+func (c *ConnTrack) mapRxPackets(flow nfct.Flow, direction nftypes.Direction) uint64 {
+	// For Ingress: CountersOrig is from external to us (RX)
+	// For Egress: CountersReply is from external to us (RX)
+	if direction == nftypes.Ingress {
+		return flow.CountersOrig.Packets
+	}
+	return flow.CountersReply.Packets
+}
+
+// mapTxPackets maps packet counts to TX based on flow direction
+func (c *ConnTrack) mapTxPackets(flow nfct.Flow, direction nftypes.Direction) uint64 {
+	// For Ingress: CountersReply is from us to external (TX)
+	// For Egress: CountersOrig is from us to external (TX)
+	if direction == nftypes.Ingress {
+		return flow.CountersReply.Packets
+	}
+	return flow.CountersOrig.Packets
+}
+
+// mapRxBytes maps byte counts to RX based on flow direction
+func (c *ConnTrack) mapRxBytes(flow nfct.Flow, direction nftypes.Direction) uint64 {
+	// For Ingress: CountersOrig is from external to us (RX)
+	// For Egress: CountersReply is from external to us (RX)
+	if direction == nftypes.Ingress {
+		return flow.CountersOrig.Bytes
+	}
+	return flow.CountersReply.Bytes
+}
+
+// mapTxBytes maps byte counts to TX based on flow direction
+func (c *ConnTrack) mapTxBytes(flow nfct.Flow, direction nftypes.Direction) uint64 {
+	// For Ingress: CountersReply is from us to external (TX)
+	// For Egress: CountersOrig is from us to external (TX)
+	if direction == nftypes.Ingress {
+		return flow.CountersReply.Bytes
+	}
+	return flow.CountersOrig.Bytes
+}
+
+// getFlowID creates a unique UUID based on the conntrack ID and instance ID
+func (c *ConnTrack) getFlowID(conntrackID uint32) uuid.UUID {
+	var buf [4]byte
+	binary.BigEndian.PutUint32(buf[:], conntrackID)
+	return uuid.NewSHA1(c.instanceID, buf[:])
+}
+
+func (c *ConnTrack) inferDirection(srcIP, dstIP netip.Addr) nftypes.Direction {
+	wgaddr := c.iface.Address().IP
+	wgnetwork := c.iface.Address().Network
+	src, dst := srcIP.AsSlice(), dstIP.AsSlice()
+
+	switch {
+	case wgaddr.Equal(src):
+		return nftypes.Egress
+	case wgaddr.Equal(dst):
+		return nftypes.Ingress
+	case wgnetwork.Contains(src):
+		// netbird network -> resource network
+		return nftypes.Ingress
+	case wgnetwork.Contains(dst):
+		// resource network -> netbird network
+		return nftypes.Egress
+
+		// TODO: handle site2site traffic
+	}
+
+	return nftypes.DirectionUnknown
+}

client/internal/netflow/conntrack/conntrack_nonlinux.go

@@ -0,0 +1,9 @@
+//go:build !linux || android
+
+package conntrack
+
+import nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+
+func New(flowLogger nftypes.FlowLogger, iface nftypes.IFaceMapper) nftypes.ConnTracker {
+	return nil
+}

client/internal/netflow/conntrack/sysctl.go

@@ -0,0 +1,73 @@
+//go:build linux && !android
+
+package conntrack
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	// conntrackAcctPath is the sysctl path for conntrack accounting
+	conntrackAcctPath = "net.netfilter.nf_conntrack_acct"
+)
+
+// EnableAccounting ensures that connection tracking accounting is enabled  in the kernel.
+func (c *ConnTrack) EnableAccounting() {
+	// haven't restored yet
+	if c.sysctlModified {
+		return
+	}
+
+	modified, err := setSysctl(conntrackAcctPath, 1)
+	if err != nil {
+		log.Warnf("Failed to enable conntrack accounting: %v", err)
+		return
+	}
+	c.sysctlModified = modified
+}
+
+// RestoreAccounting restores the connection tracking accounting setting to its original value.
+func (c *ConnTrack) RestoreAccounting() {
+	if !c.sysctlModified {
+		return
+	}
+
+	if _, err := setSysctl(conntrackAcctPath, 0); err != nil {
+		log.Warnf("Failed to restore conntrack accounting: %v", err)
+		return
+	}
+
+	c.sysctlModified = false
+}
+
+// setSysctl sets a sysctl configuration and returns whether it was modified.
+func setSysctl(key string, desiredValue int) (bool, error) {
+	path := fmt.Sprintf("/proc/sys/%s", strings.ReplaceAll(key, ".", "/"))
+
+	currentValue, err := os.ReadFile(path)
+	if err != nil {
+		return false, fmt.Errorf("read sysctl %s: %w", key, err)
+	}
+
+	currentV, err := strconv.Atoi(strings.TrimSpace(string(currentValue)))
+	if err != nil && len(currentValue) > 0 {
+		return false, fmt.Errorf("convert current value to int: %w", err)
+	}
+
+	if currentV == desiredValue {
+		return false, nil
+	}
+
+	// nolint:gosec
+	if err := os.WriteFile(path, []byte(strconv.Itoa(desiredValue)), 0644); err != nil {
+		return false, fmt.Errorf("write sysctl %s: %w", key, err)
+	}
+
+	log.Debugf("Set sysctl %s from %d to %d", key, currentV, desiredValue)
+	return true, nil
+}

client/internal/netflow/logger/logger.go

@@ -0,0 +1,162 @@
+package logger
+
+import (
+	"context"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/dnsfwd"
+	"github.com/netbirdio/netbird/client/internal/netflow/store"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/client/internal/peer"
+)
+
+type rcvChan chan *types.EventFields
+type Logger struct {
+	mux                sync.Mutex
+	ctx                context.Context
+	cancel             context.CancelFunc
+	enabled            atomic.Bool
+	rcvChan            atomic.Pointer[rcvChan]
+	cancelReceiver     context.CancelFunc
+	statusRecorder     *peer.Status
+	wgIfaceIPNet       net.IPNet
+	dnsCollection      atomic.Bool
+	exitNodeCollection atomic.Bool
+	Store              types.Store
+}
+
+func New(ctx context.Context, statusRecorder *peer.Status, wgIfaceIPNet net.IPNet) *Logger {
+
+	ctx, cancel := context.WithCancel(ctx)
+	return &Logger{
+		ctx:            ctx,
+		cancel:         cancel,
+		statusRecorder: statusRecorder,
+		wgIfaceIPNet:   wgIfaceIPNet,
+		Store:          store.NewMemoryStore(),
+	}
+}
+
+func (l *Logger) StoreEvent(flowEvent types.EventFields) {
+	if !l.enabled.Load() {
+		return
+	}
+
+	c := l.rcvChan.Load()
+	if c == nil {
+		return
+	}
+
+	select {
+	case *c <- &flowEvent:
+	default:
+		// todo: we should collect or log on this
+	}
+}
+
+func (l *Logger) Enable() {
+	go l.startReceiver()
+}
+
+func (l *Logger) startReceiver() {
+	if l.enabled.Load() {
+		return
+	}
+
+	l.mux.Lock()
+	ctx, cancel := context.WithCancel(l.ctx)
+	l.cancelReceiver = cancel
+	l.mux.Unlock()
+
+	c := make(rcvChan, 100)
+	l.rcvChan.Store(&c)
+	l.enabled.Store(true)
+
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("flow Memory store receiver stopped")
+			return
+		case eventFields := <-c:
+			id := uuid.New()
+			event := types.Event{
+				ID:          id,
+				EventFields: *eventFields,
+				Timestamp:   time.Now(),
+			}
+
+			var isExitNode bool
+			if event.Direction == types.Ingress {
+				if !l.wgIfaceIPNet.Contains(net.IP(event.SourceIP.AsSlice())) {
+					event.SourceResourceID, isExitNode = l.statusRecorder.CheckRoutes(event.SourceIP)
+				}
+			} else if event.Direction == types.Egress {
+				if !l.wgIfaceIPNet.Contains(net.IP(event.DestIP.AsSlice())) {
+					event.DestResourceID, isExitNode = l.statusRecorder.CheckRoutes(event.DestIP)
+				}
+			}
+
+			if l.shouldStore(eventFields, isExitNode) {
+				l.Store.StoreEvent(&event)
+			}
+		}
+	}
+}
+
+func (l *Logger) Disable() {
+	l.stop()
+	l.Store.Close()
+}
+
+func (l *Logger) stop() {
+	if !l.enabled.Load() {
+		return
+	}
+
+	l.enabled.Store(false)
+	l.mux.Lock()
+	if l.cancelReceiver != nil {
+		l.cancelReceiver()
+		l.cancelReceiver = nil
+	}
+	l.rcvChan.Store(nil)
+	l.mux.Unlock()
+}
+
+func (l *Logger) GetEvents() []*types.Event {
+	return l.Store.GetEvents()
+}
+
+func (l *Logger) DeleteEvents(ids []uuid.UUID) {
+	l.Store.DeleteEvents(ids)
+}
+
+func (l *Logger) UpdateConfig(dnsCollection, exitNodeCollection bool) {
+	l.dnsCollection.Store(dnsCollection)
+	l.exitNodeCollection.Store(exitNodeCollection)
+}
+
+func (l *Logger) Close() {
+	l.stop()
+	l.cancel()
+}
+
+func (l *Logger) shouldStore(event *types.EventFields, isExitNode bool) bool {
+	// check dns collection
+	if !l.dnsCollection.Load() && event.Protocol == types.UDP && (event.DestPort == 53 || event.DestPort == dnsfwd.ListenPort) {
+		return false
+	}
+
+	// check exit node collection
+	if !l.exitNodeCollection.Load() && isExitNode {
+		return false
+	}
+
+	return true
+}

client/internal/netflow/logger/logger_test.go

@@ -0,0 +1,68 @@
+package logger_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/netbirdio/netbird/client/internal/netflow/logger"
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+)
+
+func TestStore(t *testing.T) {
+	logger := logger.New(context.Background(), nil, net.IPNet{})
+	logger.Enable()
+
+	event := types.EventFields{
+		FlowID:    uuid.New(),
+		Type:      types.TypeStart,
+		Direction: types.Ingress,
+		Protocol:  6,
+	}
+
+	wait := func() { time.Sleep(time.Millisecond) }
+	wait()
+	logger.StoreEvent(event)
+	wait()
+
+	allEvents := logger.GetEvents()
+	matched := false
+	for _, e := range allEvents {
+		if e.EventFields.FlowID == event.FlowID {
+			matched = true
+		}
+	}
+	if !matched {
+		t.Errorf("didn't match any event")
+	}
+
+	// test disable
+	logger.Disable()
+	wait()
+	logger.StoreEvent(event)
+	wait()
+	allEvents = logger.GetEvents()
+	if len(allEvents) != 0 {
+		t.Errorf("expected 0 events, got %d", len(allEvents))
+	}
+
+	// test re-enable
+	logger.Enable()
+	wait()
+	logger.StoreEvent(event)
+	wait()
+
+	allEvents = logger.GetEvents()
+	matched = false
+	for _, e := range allEvents {
+		if e.EventFields.FlowID == event.FlowID {
+			matched = true
+		}
+	}
+	if !matched {
+		t.Errorf("didn't match any event")
+	}
+}

client/internal/netflow/manager.go

@@ -0,0 +1,262 @@
+package netflow
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"runtime"
+	"sync"
+	"time"
+
+	"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"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+	"github.com/netbirdio/netbird/client/internal/peer"
+	"github.com/netbirdio/netbird/flow/client"
+	"github.com/netbirdio/netbird/flow/proto"
+)
+
+// Manager handles netflow tracking and logging
+type Manager struct {
+	mux            sync.Mutex
+	logger         nftypes.FlowLogger
+	flowConfig     *nftypes.FlowConfig
+	conntrack      nftypes.ConnTracker
+	ctx            context.Context
+	receiverClient *client.GRPCClient
+	publicKey      []byte
+}
+
+// NewManager creates a new netflow manager
+func NewManager(ctx context.Context, iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *peer.Status) *Manager {
+	var ipNet net.IPNet
+	if iface != nil {
+		ipNet = *iface.Address().Network
+	}
+	flowLogger := logger.New(ctx, statusRecorder, ipNet)
+
+	var ct nftypes.ConnTracker
+	if runtime.GOOS == "linux" && iface != nil && !iface.IsUserspaceBind() {
+		ct = conntrack.New(flowLogger, iface)
+	}
+
+	return &Manager{
+		logger:    flowLogger,
+		conntrack: ct,
+		ctx:       ctx,
+		publicKey: publicKey,
+	}
+}
+
+// Update applies new flow configuration settings
+// needsNewClient checks if a new client needs to be created
+func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
+	current := m.flowConfig
+	return previous == nil ||
+		!previous.Enabled ||
+		previous.TokenPayload != current.TokenPayload ||
+		previous.TokenSignature != current.TokenSignature ||
+		previous.URL != current.URL
+}
+
+// enableFlow starts components for flow tracking
+func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
+	// first make sender ready so events don't pile up
+	if m.needsNewClient(previous) {
+		if m.receiverClient != nil {
+			if err := m.receiverClient.Close(); err != nil {
+				log.Warnf("error closing previous flow client: %v", err)
+			}
+		}
+
+		flowClient, err := client.NewClient(m.flowConfig.URL, m.flowConfig.TokenPayload, m.flowConfig.TokenSignature, m.flowConfig.Interval)
+		if err != nil {
+			return fmt.Errorf("create client: %w", err)
+		}
+		log.Infof("flow client configured to connect to %s", m.flowConfig.URL)
+
+		m.receiverClient = flowClient
+		go m.receiveACKs(flowClient)
+		go m.startSender()
+	}
+
+	m.logger.Enable()
+
+	if m.conntrack != nil {
+		if err := m.conntrack.Start(m.flowConfig.Counters); err != nil {
+			return fmt.Errorf("start conntrack: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// disableFlow stops components for flow tracking
+func (m *Manager) disableFlow() error {
+	if m.conntrack != nil {
+		m.conntrack.Stop()
+	}
+
+	m.logger.Disable()
+
+	if m.receiverClient != nil {
+		return m.receiverClient.Close()
+	}
+	return nil
+}
+
+// Update applies new flow configuration settings
+func (m *Manager) Update(update *nftypes.FlowConfig) error {
+	if update == nil {
+		log.Debug("no update provided; skipping update")
+		return nil
+	}
+
+	log.Tracef("updating flow configuration with new settings: %+v", update)
+
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	previous := m.flowConfig
+	m.flowConfig = update
+
+	// Preserve TokenPayload and TokenSignature if they were set previously
+	if previous != nil && previous.TokenPayload != "" && m.flowConfig != nil && m.flowConfig.TokenPayload == "" {
+		m.flowConfig.TokenPayload = previous.TokenPayload
+		m.flowConfig.TokenSignature = previous.TokenSignature
+	}
+
+	m.logger.UpdateConfig(update.DNSCollection, update.ExitNodeCollection)
+
+	if update.Enabled {
+		log.Infof("netflow manager enabled; starting netflow manager")
+		return m.enableFlow(previous)
+	}
+
+	log.Infof("netflow manager disabled; stopping netflow manager")
+	err := m.disableFlow()
+	if err != nil {
+		log.Errorf("failed to disable netflow manager: %v", err)
+	}
+	return err
+}
+
+// Close cleans up all resources
+func (m *Manager) Close() {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	if m.conntrack != nil {
+		m.conntrack.Close()
+	}
+
+	if m.receiverClient != nil {
+		if err := m.receiverClient.Close(); err != nil {
+			log.Warnf("failed to close receiver client: %v", err)
+		}
+	}
+
+	m.logger.Close()
+}
+
+// GetLogger returns the flow logger
+func (m *Manager) GetLogger() nftypes.FlowLogger {
+	return m.logger
+}
+
+func (m *Manager) startSender() {
+	ticker := time.NewTicker(m.flowConfig.Interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-m.ctx.Done():
+			return
+		case <-ticker.C:
+			events := m.logger.GetEvents()
+			for _, event := range events {
+				if err := m.send(event); err != nil {
+					log.Errorf("failed to send flow event to server: %v", err)
+					continue
+				}
+				log.Tracef("sent flow event: %s", event.ID)
+			}
+		}
+	}
+}
+
+func (m *Manager) receiveACKs(client *client.GRPCClient) {
+	err := client.Receive(m.ctx, m.flowConfig.Interval, 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{uuid.UUID(ack.EventId)})
+		return nil
+	})
+
+	if err != nil && !errors.Is(err, context.Canceled) {
+		log.Errorf("failed to receive flow event ack: %v", err)
+	}
+}
+
+func (m *Manager) send(event *nftypes.Event) error {
+	m.mux.Lock()
+	client := m.receiverClient
+	m.mux.Unlock()
+
+	if client == nil {
+		return nil
+	}
+
+	return client.Send(toProtoEvent(m.publicKey, event))
+}
+
+func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
+	protoEvent := &proto.FlowEvent{
+		EventId:   event.ID[:],
+		Timestamp: timestamppb.New(event.Timestamp),
+		PublicKey: publicKey,
+		FlowFields: &proto.FlowFields{
+			FlowId:           event.FlowID[:],
+			RuleId:           event.RuleID,
+			Type:             proto.Type(event.Type),
+			Direction:        proto.Direction(event.Direction),
+			Protocol:         uint32(event.Protocol),
+			SourceIp:         event.SourceIP.AsSlice(),
+			DestIp:           event.DestIP.AsSlice(),
+			RxPackets:        event.RxPackets,
+			TxPackets:        event.TxPackets,
+			RxBytes:          event.RxBytes,
+			TxBytes:          event.TxBytes,
+			SourceResourceId: event.SourceResourceID,
+			DestResourceId:   event.DestResourceID,
+		},
+	}
+
+	if event.Protocol == nftypes.ICMP {
+		protoEvent.FlowFields.ConnectionInfo = &proto.FlowFields_IcmpInfo{
+			IcmpInfo: &proto.ICMPInfo{
+				IcmpType: uint32(event.ICMPType),
+				IcmpCode: uint32(event.ICMPCode),
+			},
+		}
+		return protoEvent
+	}
+
+	protoEvent.FlowFields.ConnectionInfo = &proto.FlowFields_PortInfo{
+		PortInfo: &proto.PortInfo{
+			SourcePort: uint32(event.SourcePort),
+			DestPort:   uint32(event.DestPort),
+		},
+	}
+
+	return protoEvent
+}

client/internal/netflow/store/memory.go

@@ -0,0 +1,52 @@
+package store
+
+import (
+	"sync"
+
+	"golang.org/x/exp/maps"
+
+	"github.com/google/uuid"
+
+	"github.com/netbirdio/netbird/client/internal/netflow/types"
+)
+
+func NewMemoryStore() *Memory {
+	return &Memory{
+		events: make(map[uuid.UUID]*types.Event),
+	}
+}
+
+type Memory struct {
+	mux    sync.Mutex
+	events map[uuid.UUID]*types.Event
+}
+
+func (m *Memory) StoreEvent(event *types.Event) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	m.events[event.ID] = event
+}
+
+func (m *Memory) Close() {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	maps.Clear(m.events)
+}
+
+func (m *Memory) GetEvents() []*types.Event {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	events := make([]*types.Event, 0, len(m.events))
+	for _, event := range m.events {
+		events = append(events, event)
+	}
+	return events
+}
+
+func (m *Memory) DeleteEvents(ids []uuid.UUID) {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	for _, id := range ids {
+		delete(m.events, id)
+	}
+}

client/internal/netflow/types/types.go

@@ -0,0 +1,156 @@
+package types
+
+import (
+	"net/netip"
+	"strconv"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/netbirdio/netbird/client/iface/wgaddr"
+)
+
+type Protocol uint8
+
+const (
+	ProtocolUnknown = Protocol(0)
+	ICMP            = Protocol(1)
+	TCP             = Protocol(6)
+	UDP             = Protocol(17)
+	SCTP            = Protocol(132)
+)
+
+func (p Protocol) String() string {
+	switch p {
+	case 1:
+		return "ICMP"
+	case 6:
+		return "TCP"
+	case 17:
+		return "UDP"
+	case 132:
+		return "SCTP"
+	default:
+		return strconv.FormatUint(uint64(p), 10)
+	}
+}
+
+type Type int
+
+const (
+	TypeUnknown = Type(iota)
+	TypeStart
+	TypeEnd
+	TypeDrop
+)
+
+type Direction int
+
+func (d Direction) String() string {
+	switch d {
+	case Ingress:
+		return "ingress"
+	case Egress:
+		return "egress"
+	default:
+		return "unknown"
+	}
+}
+
+const (
+	DirectionUnknown = Direction(iota)
+	Ingress
+	Egress
+)
+
+type Event struct {
+	ID        uuid.UUID
+	Timestamp time.Time
+	EventFields
+}
+
+type EventFields struct {
+	FlowID           uuid.UUID
+	Type             Type
+	RuleID           []byte
+	Direction        Direction
+	Protocol         Protocol
+	SourceIP         netip.Addr
+	DestIP           netip.Addr
+	SourceResourceID []byte
+	DestResourceID   []byte
+	SourcePort       uint16
+	DestPort         uint16
+	ICMPType         uint8
+	ICMPCode         uint8
+	RxPackets        uint64
+	TxPackets        uint64
+	RxBytes          uint64
+	TxBytes          uint64
+}
+
+type FlowConfig struct {
+	URL                string
+	Interval           time.Duration
+	Enabled            bool
+	Counters           bool
+	TokenPayload       string
+	TokenSignature     string
+	DNSCollection      bool
+	ExitNodeCollection bool
+}
+
+type FlowManager interface {
+	// FlowConfig handles network map updates
+	Update(update *FlowConfig) error
+	// Close closes the manager
+	Close()
+	// GetLogger returns a flow logger
+	GetLogger() FlowLogger
+}
+
+type FlowLogger interface {
+	// StoreEvent stores a flow event
+	StoreEvent(flowEvent EventFields)
+	// GetEvents returns all stored events
+	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]uuid.UUID)
+	// Close closes the logger
+	Close()
+	// Enable enables the flow logger receiver
+	Enable()
+	// Disable disables the flow logger receiver
+	Disable()
+
+	// UpdateConfig updates the flow manager configuration
+	UpdateConfig(dnsCollection, exitNodeCollection bool)
+}
+
+type Store interface {
+	// StoreEvent stores a flow event
+	StoreEvent(event *Event)
+	// GetEvents returns all stored events
+	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]uuid.UUID)
+	// Close closes the store
+	Close()
+}
+
+// ConnTracker defines the interface for connection tracking functionality
+type ConnTracker interface {
+	// Start begins tracking connections by listening for conntrack events.
+	Start(bool) error
+	// Stop stops the connection tracking.
+	Stop()
+	// Close stops listening for events and cleans up resources
+	Close() error
+}
+
+// IFaceMapper provides interface to check if we're using userspace WireGuard
+type IFaceMapper interface {
+	IsUserspaceBind() bool
+	Name() string
+	Address() wgaddr.Address
+}