[management] base manager

[client] UI Refactor Icon Paths (#3420)

.github/workflows/release.yml

@@ -71,7 +71,7 @@ jobs:
       - name: Install goversioninfo
         run: go install github.com/josephspurrier/goversioninfo/cmd/goversioninfo@233067e
       - name: Generate windows syso amd64
-        run: goversioninfo  -icon client/ui/netbird.ico -manifest client/manifest.xml -product-name ${{ env.PRODUCT_NAME }} -copyright "${{ env.COPYRIGHT }}" -ver-major ${{ steps.semver_parser.outputs.major }} -ver-minor ${{ steps.semver_parser.outputs.minor }} -ver-patch ${{ steps.semver_parser.outputs.patch }} -ver-build 0 -file-version ${{ steps.semver_parser.outputs.fullversion }}.0 -product-version ${{ steps.semver_parser.outputs.fullversion }}.0 -o client/resources_windows_amd64.syso
+        run: goversioninfo  -icon client/ui/assets/netbird.ico -manifest client/manifest.xml -product-name ${{ env.PRODUCT_NAME }} -copyright "${{ env.COPYRIGHT }}" -ver-major ${{ steps.semver_parser.outputs.major }} -ver-minor ${{ steps.semver_parser.outputs.minor }} -ver-patch ${{ steps.semver_parser.outputs.patch }} -ver-build 0 -file-version ${{ steps.semver_parser.outputs.fullversion }}.0 -product-version ${{ steps.semver_parser.outputs.fullversion }}.0 -o client/resources_windows_amd64.syso
       - name: Run GoReleaser
         uses: goreleaser/goreleaser-action@v4
         with:
@@ -150,7 +150,7 @@ jobs:
       - name: Install goversioninfo
         run: go install github.com/josephspurrier/goversioninfo/cmd/goversioninfo@233067e
       - name: Generate windows syso amd64
-        run: goversioninfo -64 -icon client/ui/netbird.ico -manifest client/ui/manifest.xml -product-name ${{ env.PRODUCT_NAME }}-"UI" -copyright "${{ env.COPYRIGHT }}" -ver-major ${{ steps.semver_parser.outputs.major }} -ver-minor ${{ steps.semver_parser.outputs.minor }} -ver-patch ${{ steps.semver_parser.outputs.patch }} -ver-build 0 -file-version ${{ steps.semver_parser.outputs.fullversion }}.0 -product-version ${{ steps.semver_parser.outputs.fullversion }}.0 -o client/ui/resources_windows_amd64.syso
+        run: goversioninfo -64 -icon client/ui/assets/netbird.ico -manifest client/ui/manifest.xml -product-name ${{ env.PRODUCT_NAME }}-"UI" -copyright "${{ env.COPYRIGHT }}" -ver-major ${{ steps.semver_parser.outputs.major }} -ver-minor ${{ steps.semver_parser.outputs.minor }} -ver-patch ${{ steps.semver_parser.outputs.patch }} -ver-build 0 -file-version ${{ steps.semver_parser.outputs.fullversion }}.0 -product-version ${{ steps.semver_parser.outputs.fullversion }}.0 -o client/ui/resources_windows_amd64.syso
 
       - name: Run GoReleaser
         uses: goreleaser/goreleaser-action@v4

.goreleaser_ui.yaml

@@ -53,9 +53,9 @@ nfpms:
     scripts:
       postinstall: "release_files/ui-post-install.sh"
     contents:
-      - src: client/ui/netbird.desktop
+      - src: client/ui/build/netbird.desktop
         dst: /usr/share/applications/netbird.desktop
-      - src: client/ui/netbird.png
+      - src: client/ui/assets/netbird.png
         dst: /usr/share/pixmaps/netbird.png
     dependencies:
       - netbird
@@ -72,9 +72,9 @@ nfpms:
     scripts:
       postinstall: "release_files/ui-post-install.sh"
     contents:
-      - src: client/ui/netbird.desktop
+      - src: client/ui/build/netbird.desktop
         dst: /usr/share/applications/netbird.desktop
-      - src: client/ui/netbird.png
+      - src: client/ui/assets/netbird.png
         dst: /usr/share/pixmaps/netbird.png
     dependencies:
       - netbird

client/firewall/create.go

@@ -10,17 +10,18 @@ import (
 
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
 	"github.com/netbirdio/netbird/client/firewall/uspfilter"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
 )
 
 // NewFirewall creates a firewall manager instance
-func NewFirewall(iface IFaceMapper, _ *statemanager.Manager, disableServerRoutes bool) (firewall.Manager, error) {
+func NewFirewall(iface IFaceMapper, _ *statemanager.Manager, flowLogger nftypes.FlowLogger, disableServerRoutes bool) (firewall.Manager, error) {
 	if !iface.IsUserspaceBind() {
 		return nil, fmt.Errorf("not implemented for this OS: %s", runtime.GOOS)
 	}
 
 	// use userspace packet filtering firewall
-	fm, err := uspfilter.Create(iface, disableServerRoutes)
+	fm, err := uspfilter.Create(iface, disableServerRoutes, flowLogger)
 	if err != nil {
 		return nil, err
 	}

client/firewall/create_linux.go

@@ -15,6 +15,7 @@ import (
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
 	nbnftables "github.com/netbirdio/netbird/client/firewall/nftables"
 	"github.com/netbirdio/netbird/client/firewall/uspfilter"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
 )
 
@@ -33,7 +34,7 @@ const SKIP_NFTABLES_ENV = "NB_SKIP_NFTABLES_CHECK"
 // FWType is the type for the firewall type
 type FWType int
 
-func NewFirewall(iface IFaceMapper, stateManager *statemanager.Manager, disableServerRoutes bool) (firewall.Manager, error) {
+func NewFirewall(iface IFaceMapper, stateManager *statemanager.Manager, flowLogger nftypes.FlowLogger, disableServerRoutes bool) (firewall.Manager, error) {
 	// on the linux system we try to user nftables or iptables
 	// in any case, because we need to allow netbird interface traffic
 	// so we use AllowNetbird traffic from these firewall managers
@@ -47,7 +48,7 @@ func NewFirewall(iface IFaceMapper, stateManager *statemanager.Manager, disableS
 	if err != nil {
 		log.Warnf("failed to create native firewall: %v. Proceeding with userspace", err)
 	}
-	return createUserspaceFirewall(iface, fm, disableServerRoutes)
+	return createUserspaceFirewall(iface, fm, disableServerRoutes, flowLogger)
 }
 
 func createNativeFirewall(iface IFaceMapper, stateManager *statemanager.Manager, routes bool) (firewall.Manager, error) {
@@ -77,12 +78,12 @@ func createFW(iface IFaceMapper) (firewall.Manager, error) {
 	}
 }
 
-func createUserspaceFirewall(iface IFaceMapper, fm firewall.Manager, disableServerRoutes bool) (firewall.Manager, error) {
+func createUserspaceFirewall(iface IFaceMapper, fm firewall.Manager, disableServerRoutes bool, flowLogger nftypes.FlowLogger) (firewall.Manager, error) {
 	var errUsp error
 	if fm != nil {
-		fm, errUsp = uspfilter.CreateWithNativeFirewall(iface, fm, disableServerRoutes)
+		fm, errUsp = uspfilter.CreateWithNativeFirewall(iface, fm, disableServerRoutes, flowLogger)
 	} else {
-		fm, errUsp = uspfilter.Create(iface, disableServerRoutes)
+		fm, errUsp = uspfilter.Create(iface, disableServerRoutes, flowLogger)
 	}
 
 	if errUsp != nil {

client/firewall/iptables/acl_linux.go

@@ -75,6 +75,7 @@ func (m *aclManager) init(stateManager *statemanager.Manager) error {
 }
 
 func (m *aclManager) AddPeerFiltering(
+	id []byte,
 	ip net.IP,
 	protocol firewall.Protocol,
 	sPort *firewall.Port,

client/firewall/iptables/manager_linux.go

@@ -96,21 +96,22 @@ func (m *Manager) Init(stateManager *statemanager.Manager) error {
 //
 // Comment will be ignored because some system this feature is not supported
 func (m *Manager) AddPeerFiltering(
+	id []byte,
 	ip net.IP,
-	protocol firewall.Protocol,
+	proto firewall.Protocol,
 	sPort *firewall.Port,
 	dPort *firewall.Port,
 	action firewall.Action,
 	ipsetName string,
-	_ string,
 ) ([]firewall.Rule, error) {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
-	return m.aclMgr.AddPeerFiltering(ip, protocol, sPort, dPort, action, ipsetName)
+	return m.aclMgr.AddPeerFiltering(id, ip, proto, sPort, dPort, action, ipsetName)
 }
 
 func (m *Manager) AddRouteFiltering(
+	id []byte,
 	sources []netip.Prefix,
 	destination netip.Prefix,
 	proto firewall.Protocol,
@@ -125,7 +126,7 @@ func (m *Manager) AddRouteFiltering(
 		return nil, fmt.Errorf("unsupported IP version: %s", destination.Addr().String())
 	}
 
-	return m.router.AddRouteFiltering(sources, destination, proto, sPort, dPort, action)
+	return m.router.AddRouteFiltering(id, sources, destination, proto, sPort, dPort, action)
 }
 
 // DeletePeerRule from the firewall by rule definition
@@ -166,7 +167,7 @@ func (m *Manager) SetLegacyManagement(isLegacy bool) error {
 }
 
 // Reset firewall to the default state
-func (m *Manager) Reset(stateManager *statemanager.Manager) error {
+func (m *Manager) Close(stateManager *statemanager.Manager) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
@@ -196,13 +197,13 @@ func (m *Manager) AllowNetbird() error {
 	}
 
 	_, err := m.AddPeerFiltering(
+		nil,
 		net.IP{0, 0, 0, 0},
 		"all",
 		nil,
 		nil,
 		firewall.ActionAccept,
 		"",
-		"",
 	)
 	if err != nil {
 		return fmt.Errorf("allow netbird interface traffic: %w", err)

client/firewall/iptables/manager_linux_test.go

@@ -62,7 +62,7 @@ func TestIptablesManager(t *testing.T) {
 	time.Sleep(time.Second)
 
 	defer func() {
-		err := manager.Reset(nil)
+		err := manager.Close(nil)
 		require.NoError(t, err, "clear the manager state")
 
 		time.Sleep(time.Second)
@@ -75,7 +75,7 @@ func TestIptablesManager(t *testing.T) {
 			IsRange: true,
 			Values:  []uint16{8043, 8046},
 		}
-		rule2, err = manager.AddPeerFiltering(ip, "tcp", port, nil, fw.ActionAccept, "", "accept HTTPS traffic from ports range")
+		rule2, err = manager.AddPeerFiltering(nil, ip, "tcp", port, nil, fw.ActionAccept, "")
 		require.NoError(t, err, "failed to add rule")
 
 		for _, r := range rule2 {
@@ -97,17 +97,17 @@ func TestIptablesManager(t *testing.T) {
 		// add second rule
 		ip := net.ParseIP("10.20.0.3")
 		port := &fw.Port{Values: []uint16{5353}}
-		_, err = manager.AddPeerFiltering(ip, "udp", nil, port, fw.ActionAccept, "", "accept Fake DNS traffic")
+		_, err = manager.AddPeerFiltering(nil, ip, "udp", nil, port, fw.ActionAccept, "")
 		require.NoError(t, err, "failed to add rule")
 
-		err = manager.Reset(nil)
+		err = manager.Close(nil)
 		require.NoError(t, err, "failed to reset")
 
 		ok, err := ipv4Client.ChainExists("filter", chainNameInputRules)
 		require.NoError(t, err, "failed check chain exists")
 
 		if ok {
-			require.NoErrorf(t, err, "chain '%v' still exists after Reset", chainNameInputRules)
+			require.NoErrorf(t, err, "chain '%v' still exists after Close", chainNameInputRules)
 		}
 	})
 }
@@ -136,7 +136,7 @@ func TestIptablesManagerIPSet(t *testing.T) {
 	time.Sleep(time.Second)
 
 	defer func() {
-		err := manager.Reset(nil)
+		err := manager.Close(nil)
 		require.NoError(t, err, "clear the manager state")
 
 		time.Sleep(time.Second)
@@ -148,7 +148,7 @@ func TestIptablesManagerIPSet(t *testing.T) {
 		port := &fw.Port{
 			Values: []uint16{443},
 		}
-		rule2, err = manager.AddPeerFiltering(ip, "tcp", port, nil, fw.ActionAccept, "default", "accept HTTPS traffic from ports range")
+		rule2, err = manager.AddPeerFiltering(nil, ip, "tcp", port, nil, fw.ActionAccept, "default")
 		for _, r := range rule2 {
 			require.NoError(t, err, "failed to add rule")
 			require.Equal(t, r.(*Rule).ipsetName, "default-sport", "ipset name must be set")
@@ -166,7 +166,7 @@ func TestIptablesManagerIPSet(t *testing.T) {
 	})
 
 	t.Run("reset check", func(t *testing.T) {
-		err = manager.Reset(nil)
+		err = manager.Close(nil)
 		require.NoError(t, err, "failed to reset")
 	})
 }
@@ -204,7 +204,7 @@ func TestIptablesCreatePerformance(t *testing.T) {
 			time.Sleep(time.Second)
 
 			defer func() {
-				err := manager.Reset(nil)
+				err := manager.Close(nil)
 				require.NoError(t, err, "clear the manager state")
 
 				time.Sleep(time.Second)
@@ -216,7 +216,7 @@ func TestIptablesCreatePerformance(t *testing.T) {
 			start := time.Now()
 			for i := 0; i < testMax; i++ {
 				port := &fw.Port{Values: []uint16{uint16(1000 + i)}}
-				_, err = manager.AddPeerFiltering(ip, "tcp", nil, port, fw.ActionAccept, "", "accept HTTP traffic")
+				_, err = manager.AddPeerFiltering(nil, ip, "tcp", nil, port, fw.ActionAccept, "")
 
 				require.NoError(t, err, "failed to add rule")
 			}

client/firewall/iptables/router_linux.go

@@ -15,7 +15,7 @@ import (
 
 	nberrors "github.com/netbirdio/netbird/client/errors"
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
-	"github.com/netbirdio/netbird/client/internal/acl/id"
+	nbid "github.com/netbirdio/netbird/client/internal/acl/id"
 	"github.com/netbirdio/netbird/client/internal/routemanager/ipfwdstate"
 	"github.com/netbirdio/netbird/client/internal/routemanager/refcounter"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
@@ -121,6 +121,7 @@ func (r *router) init(stateManager *statemanager.Manager) error {
 }
 
 func (r *router) AddRouteFiltering(
+	id []byte,
 	sources []netip.Prefix,
 	destination netip.Prefix,
 	proto firewall.Protocol,
@@ -128,7 +129,7 @@ func (r *router) AddRouteFiltering(
 	dPort *firewall.Port,
 	action firewall.Action,
 ) (firewall.Rule, error) {
-	ruleKey := id.GenerateRouteRuleKey(sources, destination, proto, sPort, dPort, action)
+	ruleKey := nbid.GenerateRouteRuleKey(sources, destination, proto, sPort, dPort, action)
 	if _, ok := r.rules[string(ruleKey)]; ok {
 		return ruleKey, nil
 	}

client/firewall/iptables/router_linux_test.go

@@ -330,7 +330,7 @@ func TestRouter_AddRouteFiltering(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ruleKey, err := r.AddRouteFiltering(tt.sources, tt.destination, tt.proto, tt.sPort, tt.dPort, tt.action)
+			ruleKey, err := r.AddRouteFiltering(nil, tt.sources, tt.destination, tt.proto, tt.sPort, tt.dPort, tt.action)
 			require.NoError(t, err, "AddRouteFiltering failed")
 
 			// Check if the rule is in the internal map

client/firewall/iptables/state_linux.go

@@ -62,7 +62,7 @@ func (s *ShutdownState) Cleanup() error {
 		ipt.aclMgr.ipsetStore = s.ACLIPsetStore
 	}
 
-	if err := ipt.Reset(nil); err != nil {
+	if err := ipt.Close(nil); err != nil {
 		return fmt.Errorf("reset iptables manager: %w", err)
 	}
 

client/firewall/manager/firewall.go

@@ -65,13 +65,13 @@ type Manager interface {
 	// If comment argument is empty firewall manager should set
 	// rule ID as comment for the rule
 	AddPeerFiltering(
+		id []byte,
 		ip net.IP,
 		proto Protocol,
 		sPort *Port,
 		dPort *Port,
 		action Action,
 		ipsetName string,
-		comment string,
 	) ([]Rule, error)
 
 	// DeletePeerRule from the firewall by rule definition
@@ -80,7 +80,15 @@ type Manager interface {
 	// IsServerRouteSupported returns true if the firewall supports server side routing operations
 	IsServerRouteSupported() bool
 
-	AddRouteFiltering(source []netip.Prefix, destination netip.Prefix, proto Protocol, sPort *Port, dPort *Port, action Action) (Rule, error)
+	AddRouteFiltering(
+		id []byte,
+		sources []netip.Prefix,
+		destination netip.Prefix,
+		proto Protocol,
+		sPort *Port,
+		dPort *Port,
+		action Action,
+	) (Rule, error)
 
 	// DeleteRouteRule deletes a routing rule
 	DeleteRouteRule(rule Rule) error
@@ -94,8 +102,8 @@ type Manager interface {
 	// SetLegacyManagement sets the legacy management mode
 	SetLegacyManagement(legacy bool) error
 
-	// Reset firewall to the default state
-	Reset(stateManager *statemanager.Manager) error
+	// Close closes the firewall manager
+	Close(stateManager *statemanager.Manager) error
 
 	// Flush the changes to firewall controller
 	Flush() error

client/firewall/nftables/acl_linux.go

@@ -84,13 +84,13 @@ func (m *AclManager) init(workTable *nftables.Table) error {
 // If comment argument is empty firewall manager should set
 // rule ID as comment for the rule
 func (m *AclManager) AddPeerFiltering(
+	id []byte,
 	ip net.IP,
 	proto firewall.Protocol,
 	sPort *firewall.Port,
 	dPort *firewall.Port,
 	action firewall.Action,
 	ipsetName string,
-	comment string,
 ) ([]firewall.Rule, error) {
 	var ipset *nftables.Set
 	if ipsetName != "" {
@@ -102,7 +102,7 @@ func (m *AclManager) AddPeerFiltering(
 	}
 
 	newRules := make([]firewall.Rule, 0, 2)
-	ioRule, err := m.addIOFiltering(ip, proto, sPort, dPort, action, ipset, comment)
+	ioRule, err := m.addIOFiltering(ip, proto, sPort, dPort, action, ipset)
 	if err != nil {
 		return nil, err
 	}
@@ -256,7 +256,6 @@ func (m *AclManager) addIOFiltering(
 	dPort *firewall.Port,
 	action firewall.Action,
 	ipset *nftables.Set,
-	comment string,
 ) (*Rule, error) {
 	ruleId := generatePeerRuleId(ip, sPort, dPort, action, ipset)
 	if r, ok := m.rules[ruleId]; ok {
@@ -338,7 +337,7 @@ func (m *AclManager) addIOFiltering(
 		mainExpressions = append(mainExpressions, &expr.Verdict{Kind: expr.VerdictDrop})
 	}
 
-	userData := []byte(strings.Join([]string{ruleId, comment}, " "))
+	userData := []byte(ruleId)
 
 	chain := m.chainInputRules
 	nftRule := m.rConn.AddRule(&nftables.Rule{

client/firewall/nftables/manager_linux.go

@@ -87,7 +87,7 @@ func (m *Manager) Init(stateManager *statemanager.Manager) error {
 	// We only need to record minimal interface state for potential recreation.
 	// Unlike iptables, which requires tracking individual rules, nftables maintains
 	// a known state (our netbird table plus a few static rules). This allows for easy
-	// cleanup using Reset() without needing to store specific rules.
+	// cleanup using Close() without needing to store specific rules.
 	if err := stateManager.UpdateState(&ShutdownState{
 		InterfaceState: &InterfaceState{
 			NameStr:       m.wgIface.Name(),
@@ -113,13 +113,13 @@ func (m *Manager) Init(stateManager *statemanager.Manager) error {
 // If comment argument is empty firewall manager should set
 // rule ID as comment for the rule
 func (m *Manager) AddPeerFiltering(
+	id []byte,
 	ip net.IP,
 	proto firewall.Protocol,
 	sPort *firewall.Port,
 	dPort *firewall.Port,
 	action firewall.Action,
 	ipsetName string,
-	comment string,
 ) ([]firewall.Rule, error) {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
@@ -129,10 +129,11 @@ func (m *Manager) AddPeerFiltering(
 		return nil, fmt.Errorf("unsupported IP version: %s", ip.String())
 	}
 
-	return m.aclManager.AddPeerFiltering(ip, proto, sPort, dPort, action, ipsetName, comment)
+	return m.aclManager.AddPeerFiltering(id, ip, proto, sPort, dPort, action, ipsetName)
 }
 
 func (m *Manager) AddRouteFiltering(
+	id []byte,
 	sources []netip.Prefix,
 	destination netip.Prefix,
 	proto firewall.Protocol,
@@ -147,7 +148,7 @@ func (m *Manager) AddRouteFiltering(
 		return nil, fmt.Errorf("unsupported IP version: %s", destination.Addr().String())
 	}
 
-	return m.router.AddRouteFiltering(sources, destination, proto, sPort, dPort, action)
+	return m.router.AddRouteFiltering(id, sources, destination, proto, sPort, dPort, action)
 }
 
 // DeletePeerRule from the firewall by rule definition
@@ -242,7 +243,7 @@ func (m *Manager) SetLegacyManagement(isLegacy bool) error {
 }
 
 // Reset firewall to the default state
-func (m *Manager) Reset(stateManager *statemanager.Manager) error {
+func (m *Manager) Close(stateManager *statemanager.Manager) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 

client/firewall/nftables/manager_linux_test.go

@@ -65,7 +65,7 @@ func TestNftablesManager(t *testing.T) {
 	time.Sleep(time.Second * 3)
 
 	defer func() {
-		err = manager.Reset(nil)
+		err = manager.Close(nil)
 		require.NoError(t, err, "failed to reset")
 		time.Sleep(time.Second)
 	}()
@@ -74,7 +74,7 @@ func TestNftablesManager(t *testing.T) {
 
 	testClient := &nftables.Conn{}
 
-	rule, err := manager.AddPeerFiltering(ip, fw.ProtocolTCP, nil, &fw.Port{Values: []uint16{53}}, fw.ActionDrop, "", "")
+	rule, err := manager.AddPeerFiltering(nil, ip, fw.ProtocolTCP, nil, &fw.Port{Values: []uint16{53}}, fw.ActionDrop, "")
 	require.NoError(t, err, "failed to add rule")
 
 	err = manager.Flush()
@@ -162,7 +162,7 @@ func TestNftablesManager(t *testing.T) {
 	// established rule remains
 	require.Len(t, rules, 1, "expected 1 rules after deletion")
 
-	err = manager.Reset(nil)
+	err = manager.Close(nil)
 	require.NoError(t, err, "failed to reset")
 }
 
@@ -191,7 +191,7 @@ func TestNFtablesCreatePerformance(t *testing.T) {
 			time.Sleep(time.Second * 3)
 
 			defer func() {
-				if err := manager.Reset(nil); err != nil {
+				if err := manager.Close(nil); err != nil {
 					t.Errorf("clear the manager state: %v", err)
 				}
 				time.Sleep(time.Second)
@@ -201,7 +201,7 @@ func TestNFtablesCreatePerformance(t *testing.T) {
 			start := time.Now()
 			for i := 0; i < testMax; i++ {
 				port := &fw.Port{Values: []uint16{uint16(1000 + i)}}
-				_, err = manager.AddPeerFiltering(ip, "tcp", nil, port, fw.ActionAccept, "", "accept HTTP traffic")
+				_, err = manager.AddPeerFiltering(nil, ip, "tcp", nil, port, fw.ActionAccept, "")
 				require.NoError(t, err, "failed to add rule")
 
 				if i%100 == 0 {
@@ -274,7 +274,7 @@ func TestNftablesManagerCompatibilityWithIptables(t *testing.T) {
 	require.NoError(t, manager.Init(nil))
 
 	t.Cleanup(func() {
-		err := manager.Reset(nil)
+		err := manager.Close(nil)
 		require.NoError(t, err, "failed to reset manager state")
 
 		// Verify iptables output after reset
@@ -283,10 +283,11 @@ func TestNftablesManagerCompatibilityWithIptables(t *testing.T) {
 	})
 
 	ip := net.ParseIP("100.96.0.1")
-	_, err = manager.AddPeerFiltering(ip, fw.ProtocolTCP, nil, &fw.Port{Values: []uint16{80}}, fw.ActionAccept, "", "test rule")
+	_, err = manager.AddPeerFiltering(nil, ip, fw.ProtocolTCP, nil, &fw.Port{Values: []uint16{80}}, fw.ActionAccept, "")
 	require.NoError(t, err, "failed to add peer filtering rule")
 
 	_, err = manager.AddRouteFiltering(
+		nil,
 		[]netip.Prefix{netip.MustParsePrefix("192.168.2.0/24")},
 		netip.MustParsePrefix("10.1.0.0/24"),
 		fw.ProtocolTCP,

client/firewall/nftables/router_linux.go

@@ -20,7 +20,7 @@ import (
 
 	nberrors "github.com/netbirdio/netbird/client/errors"
 	firewall "github.com/netbirdio/netbird/client/firewall/manager"
-	"github.com/netbirdio/netbird/client/internal/acl/id"
+	nbid "github.com/netbirdio/netbird/client/internal/acl/id"
 	"github.com/netbirdio/netbird/client/internal/routemanager/ipfwdstate"
 	"github.com/netbirdio/netbird/client/internal/routemanager/refcounter"
 	nbnet "github.com/netbirdio/netbird/util/net"
@@ -228,6 +228,7 @@ func (r *router) createContainers() error {
 
 // AddRouteFiltering appends a nftables rule to the routing chain
 func (r *router) AddRouteFiltering(
+	id []byte,
 	sources []netip.Prefix,
 	destination netip.Prefix,
 	proto firewall.Protocol,
@@ -236,7 +237,7 @@ func (r *router) AddRouteFiltering(
 	action firewall.Action,
 ) (firewall.Rule, error) {
 
-	ruleKey := id.GenerateRouteRuleKey(sources, destination, proto, sPort, dPort, action)
+	ruleKey := nbid.GenerateRouteRuleKey(sources, destination, proto, sPort, dPort, action)
 	if _, ok := r.rules[string(ruleKey)]; ok {
 		return ruleKey, nil
 	}

client/firewall/nftables/router_linux_test.go

@@ -38,7 +38,7 @@ func TestNftablesManager_AddNatRule(t *testing.T) {
 			// need fw manager to init both acl mgr and router for all chains to be present
 			manager, err := Create(ifaceMock)
 			t.Cleanup(func() {
-				require.NoError(t, manager.Reset(nil))
+				require.NoError(t, manager.Close(nil))
 			})
 			require.NoError(t, err)
 			require.NoError(t, manager.Init(nil))
@@ -127,7 +127,7 @@ func TestNftablesManager_RemoveNatRule(t *testing.T) {
 		t.Run(testCase.Name, func(t *testing.T) {
 			manager, err := Create(ifaceMock)
 			t.Cleanup(func() {
-				require.NoError(t, manager.Reset(nil))
+				require.NoError(t, manager.Close(nil))
 			})
 			require.NoError(t, err)
 			require.NoError(t, manager.Init(nil))
@@ -311,7 +311,7 @@ func TestRouter_AddRouteFiltering(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ruleKey, err := r.AddRouteFiltering(tt.sources, tt.destination, tt.proto, tt.sPort, tt.dPort, tt.action)
+			ruleKey, err := r.AddRouteFiltering(nil, tt.sources, tt.destination, tt.proto, tt.sPort, tt.dPort, tt.action)
 			require.NoError(t, err, "AddRouteFiltering failed")
 
 			t.Cleanup(func() {

client/firewall/nftables/state_linux.go

@@ -39,7 +39,7 @@ func (s *ShutdownState) Cleanup() error {
 		return fmt.Errorf("create nftables manager: %w", err)
 	}
 
-	if err := nft.Reset(nil); err != nil {
+	if err := nft.Close(nil); err != nil {
 		return fmt.Errorf("reset nftables manager: %w", err)
 	}
 

client/firewall/uspfilter/allow_netbird.go

@@ -13,7 +13,7 @@ import (
 )
 
 // Reset firewall to the default state
-func (m *Manager) Reset(stateManager *statemanager.Manager) error {
+func (m *Manager) Close(stateManager *statemanager.Manager) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
@@ -22,17 +22,17 @@ func (m *Manager) Reset(stateManager *statemanager.Manager) error {
 
 	if m.udpTracker != nil {
 		m.udpTracker.Close()
-		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger)
+		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.icmpTracker != nil {
 		m.icmpTracker.Close()
-		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger)
+		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.tcpTracker != nil {
 		m.tcpTracker.Close()
-		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger)
+		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.forwarder != nil {
@@ -48,7 +48,7 @@ func (m *Manager) Reset(stateManager *statemanager.Manager) error {
 	}
 
 	if m.nativeFirewall != nil {
-		return m.nativeFirewall.Reset(stateManager)
+		return m.nativeFirewall.Close(stateManager)
 	}
 	return nil
 }

client/firewall/uspfilter/allow_netbird_windows.go

@@ -21,8 +21,8 @@ const (
 	firewallRuleName        = "Netbird"
 )
 
-// Reset firewall to the default state
-func (m *Manager) Reset(*statemanager.Manager) error {
+// Close closes the firewall manager
+func (m *Manager) Close(*statemanager.Manager) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
@@ -31,17 +31,17 @@ func (m *Manager) Reset(*statemanager.Manager) error {
 
 	if m.udpTracker != nil {
 		m.udpTracker.Close()
-		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger)
+		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.icmpTracker != nil {
 		m.icmpTracker.Close()
-		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger)
+		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.tcpTracker != nil {
 		m.tcpTracker.Close()
-		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger)
+		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger, m.flowLogger)
 	}
 
 	if m.forwarder != nil {

client/firewall/uspfilter/conntrack/common.go

@@ -1,20 +1,26 @@
-// common.go
 package conntrack
 
 import (
+	"fmt"
 	"net"
-	"sync"
+	"net/netip"
 	"sync/atomic"
 	"time"
+
+	"github.com/google/uuid"
+
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 // BaseConnTrack provides common fields and locking for all connection types
 type BaseConnTrack struct {
-	SourceIP   net.IP
-	DestIP     net.IP
+	FlowId     uuid.UUID
+	Direction  nftypes.Direction
+	SourceIP   netip.Addr
+	DestIP     netip.Addr
 	SourcePort uint16
 	DestPort   uint16
-	lastSeen   atomic.Int64 // Unix nano for atomic access
+	lastSeen   atomic.Int64
 }
 
 // these small methods will be inlined by the compiler
@@ -35,92 +41,27 @@ func (b *BaseConnTrack) timeoutExceeded(timeout time.Duration) bool {
 	return time.Since(lastSeen) > timeout
 }
 
-// IPAddr is a fixed-size IP address to avoid allocations
-type IPAddr [16]byte
-
-// MakeIPAddr creates an IPAddr from net.IP
-func MakeIPAddr(ip net.IP) (addr IPAddr) {
-	// Optimization: check for v4 first as it's more common
-	if ip4 := ip.To4(); ip4 != nil {
-		copy(addr[12:], ip4)
-	} else {
-		copy(addr[:], ip.To16())
-	}
-	return addr
-}
-
 // ConnKey uniquely identifies a connection
 type ConnKey struct {
-	SrcIP   IPAddr
-	DstIP   IPAddr
+	SrcIP   netip.Addr
+	DstIP   netip.Addr
 	SrcPort uint16
 	DstPort uint16
 }
 
+func (c ConnKey) String() string {
+	return fmt.Sprintf("%s:%d -> %s:%d", c.SrcIP.Unmap(), c.SrcPort, c.DstIP.Unmap(), c.DstPort)
+}
+
 // makeConnKey creates a connection key
 func makeConnKey(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16) ConnKey {
+	srcAddr, _ := netip.AddrFromSlice(srcIP)
+	dstAddr, _ := netip.AddrFromSlice(dstIP)
+
 	return ConnKey{
-		SrcIP:   MakeIPAddr(srcIP),
-		DstIP:   MakeIPAddr(dstIP),
+		SrcIP:   srcAddr,
+		DstIP:   dstAddr,
 		SrcPort: srcPort,
 		DstPort: dstPort,
 	}
 }
-
-// ValidateIPs checks if IPs match without allocation
-func ValidateIPs(connIP IPAddr, pktIP net.IP) bool {
-	if ip4 := pktIP.To4(); ip4 != nil {
-		// Compare IPv4 addresses (last 4 bytes)
-		for i := 0; i < 4; i++ {
-			if connIP[12+i] != ip4[i] {
-				return false
-			}
-		}
-		return true
-	}
-	// Compare full IPv6 addresses
-	ip6 := pktIP.To16()
-	for i := 0; i < 16; i++ {
-		if connIP[i] != ip6[i] {
-			return false
-		}
-	}
-	return true
-}
-
-// PreallocatedIPs is a pool of IP byte slices to reduce allocations
-type PreallocatedIPs struct {
-	sync.Pool
-}
-
-// NewPreallocatedIPs creates a new IP pool
-func NewPreallocatedIPs() *PreallocatedIPs {
-	return &PreallocatedIPs{
-		Pool: sync.Pool{
-			New: func() interface{} {
-				ip := make(net.IP, 16)
-				return &ip
-			},
-		},
-	}
-}
-
-// Get retrieves an IP from the pool
-func (p *PreallocatedIPs) Get() net.IP {
-	return *p.Pool.Get().(*net.IP)
-}
-
-// Put returns an IP to the pool
-func (p *PreallocatedIPs) Put(ip net.IP) {
-	p.Pool.Put(&ip)
-}
-
-// copyIP copies an IP address efficiently
-func copyIP(dst, src net.IP) {
-	if len(src) == 16 {
-		copy(dst, src)
-	} else {
-		// Handle IPv4
-		copy(dst[12:], src.To4())
-	}
-}

client/firewall/uspfilter/conntrack/common_test.go

@@ -1,50 +1,23 @@
 package conntrack
 
 import (
+	"context"
 	"net"
 	"testing"
 
 	"github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	"github.com/netbirdio/netbird/client/internal/netflow"
 )
 
 var logger = log.NewFromLogrus(logrus.StandardLogger())
-
-func BenchmarkIPOperations(b *testing.B) {
-	b.Run("MakeIPAddr", func(b *testing.B) {
-		ip := net.ParseIP("192.168.1.1")
-		b.ResetTimer()
-		for i := 0; i < b.N; i++ {
-			_ = MakeIPAddr(ip)
-		}
-	})
-
-	b.Run("ValidateIPs", func(b *testing.B) {
-		ip1 := net.ParseIP("192.168.1.1")
-		ip2 := net.ParseIP("192.168.1.1")
-		addr := MakeIPAddr(ip1)
-		b.ResetTimer()
-		for i := 0; i < b.N; i++ {
-			_ = ValidateIPs(addr, ip2)
-		}
-	})
-
-	b.Run("IPPool", func(b *testing.B) {
-		pool := NewPreallocatedIPs()
-		b.ResetTimer()
-		for i := 0; i < b.N; i++ {
-			ip := pool.Get()
-			pool.Put(ip)
-		}
-	})
-
-}
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
 
 // Memory pressure tests
 func BenchmarkMemoryPressure(b *testing.B) {
 	b.Run("TCPHighLoad", func(b *testing.B) {
-		tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		// Generate different IPs
@@ -69,7 +42,7 @@ func BenchmarkMemoryPressure(b *testing.B) {
 	})
 
 	b.Run("UDPHighLoad", func(b *testing.B) {
-		tracker := NewUDPTracker(DefaultUDPTimeout, logger)
+		tracker := NewUDPTracker(DefaultUDPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		// Generate different IPs

client/firewall/uspfilter/conntrack/icmp.go

@@ -1,13 +1,17 @@
 package conntrack
 
 import (
+	"fmt"
 	"net"
+	"net/netip"
 	"sync"
 	"time"
 
 	"github.com/google/gopacket/layers"
+	"github.com/google/uuid"
 
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 const (
@@ -19,18 +23,21 @@ const (
 
 // ICMPConnKey uniquely identifies an ICMP connection
 type ICMPConnKey struct {
-	// Supports both IPv4 and IPv6
-	SrcIP    [16]byte
-	DstIP    [16]byte
-	Sequence uint16 // ICMP sequence number
-	ID       uint16 // ICMP identifier
+	SrcIP    netip.Addr
+	DstIP    netip.Addr
+	Sequence uint16
+	ID       uint16
+}
+
+func (i ICMPConnKey) String() string {
+	return fmt.Sprintf("%s -> %s (%d/%d)", i.SrcIP, i.DstIP, i.ID, i.Sequence)
 }
 
 // ICMPConnTrack represents an ICMP connection state
 type ICMPConnTrack struct {
 	BaseConnTrack
-	Sequence uint16
-	ID       uint16
+	ICMPType uint8
+	ICMPCode uint8
 }
 
 // ICMPTracker manages ICMP connection states
@@ -41,11 +48,11 @@ type ICMPTracker struct {
 	cleanupTicker *time.Ticker
 	mutex         sync.RWMutex
 	done          chan struct{}
-	ipPool        *PreallocatedIPs
+	flowLogger    nftypes.FlowLogger
 }
 
 // NewICMPTracker creates a new ICMP connection tracker
-func NewICMPTracker(timeout time.Duration, logger *nblog.Logger) *ICMPTracker {
+func NewICMPTracker(timeout time.Duration, logger *nblog.Logger, flowLogger nftypes.FlowLogger) *ICMPTracker {
 	if timeout == 0 {
 		timeout = DefaultICMPTimeout
 	}
@@ -56,41 +63,77 @@ func NewICMPTracker(timeout time.Duration, logger *nblog.Logger) *ICMPTracker {
 		timeout:       timeout,
 		cleanupTicker: time.NewTicker(ICMPCleanupInterval),
 		done:          make(chan struct{}),
-		ipPool:        NewPreallocatedIPs(),
+		flowLogger:    flowLogger,
 	}
 
 	go tracker.cleanupRoutine()
 	return tracker
 }
 
-// TrackOutbound records an outbound ICMP Echo Request
-func (t *ICMPTracker) TrackOutbound(srcIP net.IP, dstIP net.IP, id uint16, seq uint16) {
+func (t *ICMPTracker) updateIfExists(srcIP net.IP, dstIP net.IP, id uint16, seq uint16) (ICMPConnKey, bool) {
 	key := makeICMPKey(srcIP, dstIP, id, seq)
 
-	t.mutex.Lock()
+	t.mutex.RLock()
 	conn, exists := t.connections[key]
-	if !exists {
-		srcIPCopy := t.ipPool.Get()
-		dstIPCopy := t.ipPool.Get()
-		copyIP(srcIPCopy, srcIP)
-		copyIP(dstIPCopy, dstIP)
+	t.mutex.RUnlock()
 
-		conn = &ICMPConnTrack{
-			BaseConnTrack: BaseConnTrack{
-				SourceIP: srcIPCopy,
-				DestIP:   dstIPCopy,
-			},
-			ID:       id,
-			Sequence: seq,
-		}
+	if exists {
 		conn.UpdateLastSeen()
-		t.connections[key] = conn
 
-		t.logger.Trace("New ICMP connection %v", key)
+		return key, true
 	}
+
+	return key, false
+}
+
+// TrackOutbound records an outbound ICMP connection
+func (t *ICMPTracker) TrackOutbound(srcIP net.IP, dstIP net.IP, id uint16, seq uint16, typecode layers.ICMPv4TypeCode) {
+	if _, exists := t.updateIfExists(dstIP, srcIP, id, seq); !exists {
+		// if (inverted direction) conn is not tracked, track this direction
+		t.track(srcIP, dstIP, id, seq, typecode, nftypes.Egress)
+	}
+}
+
+// TrackInbound records an inbound ICMP Echo Request
+func (t *ICMPTracker) TrackInbound(srcIP net.IP, dstIP net.IP, id uint16, seq uint16, typecode layers.ICMPv4TypeCode) {
+	t.track(srcIP, dstIP, id, seq, typecode, nftypes.Ingress)
+}
+
+// track is the common implementation for tracking both inbound and outbound ICMP connections
+func (t *ICMPTracker) track(srcIP net.IP, dstIP net.IP, id uint16, seq uint16, typecode layers.ICMPv4TypeCode, direction nftypes.Direction) {
+	// TODO: icmp doesn't need to extend the timeout
+	key, exists := t.updateIfExists(srcIP, dstIP, id, seq)
+	if exists {
+		return
+	}
+
+	typ, code := typecode.Type(), typecode.Code()
+
+	// non echo requests don't need tracking
+	if typ != uint8(layers.ICMPv4TypeEchoRequest) {
+		t.logger.Trace("New %s ICMP connection %s type %d code %d", direction, key, typ, code)
+		t.sendStartEvent(direction, key, typ, code)
+		return
+	}
+
+	conn := &ICMPConnTrack{
+		BaseConnTrack: BaseConnTrack{
+			FlowId:    uuid.New(),
+			Direction: direction,
+			SourceIP:  key.SrcIP,
+			DestIP:    key.DstIP,
+		},
+		ICMPType: typ,
+		ICMPCode: code,
+	}
+	conn.UpdateLastSeen()
+
+	t.mutex.Lock()
+	t.connections[key] = conn
 	t.mutex.Unlock()
 
-	conn.UpdateLastSeen()
+	t.logger.Trace("New %s ICMP connection %s type %d code %d", direction, key, typ, code)
+	t.sendEvent(nftypes.TypeStart, key, conn)
 }
 
 // IsValidInbound checks if an inbound ICMP Echo Reply matches a tracked request
@@ -105,18 +148,13 @@ func (t *ICMPTracker) IsValidInbound(srcIP net.IP, dstIP net.IP, id uint16, seq
 	conn, exists := t.connections[key]
 	t.mutex.RUnlock()
 
-	if !exists {
+	if !exists || conn.timeoutExceeded(t.timeout) {
 		return false
 	}
 
-	if conn.timeoutExceeded(t.timeout) {
-		return false
-	}
+	conn.UpdateLastSeen()
 
-	return ValidateIPs(MakeIPAddr(srcIP), conn.DestIP) &&
-		ValidateIPs(MakeIPAddr(dstIP), conn.SourceIP) &&
-		conn.ID == id &&
-		conn.Sequence == seq
+	return true
 }
 
 func (t *ICMPTracker) cleanupRoutine() {
@@ -129,17 +167,17 @@ func (t *ICMPTracker) cleanupRoutine() {
 		}
 	}
 }
+
 func (t *ICMPTracker) cleanup() {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
 	for key, conn := range t.connections {
 		if conn.timeoutExceeded(t.timeout) {
-			t.ipPool.Put(conn.SourceIP)
-			t.ipPool.Put(conn.DestIP)
 			delete(t.connections, key)
 
-			t.logger.Debug("Removed ICMP connection %v (timeout)", key)
+			t.logger.Debug("Removed ICMP connection %s (timeout)", &key)
+			t.sendEvent(nftypes.TypeEnd, key, conn)
 		}
 	}
 }
@@ -150,19 +188,43 @@ func (t *ICMPTracker) Close() {
 	close(t.done)
 
 	t.mutex.Lock()
-	for _, conn := range t.connections {
-		t.ipPool.Put(conn.SourceIP)
-		t.ipPool.Put(conn.DestIP)
-	}
 	t.connections = nil
 	t.mutex.Unlock()
 }
 
+func (t *ICMPTracker) sendEvent(typ nftypes.Type, key ICMPConnKey, conn *ICMPConnTrack) {
+	t.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    conn.FlowId,
+		Type:      typ,
+		Direction: conn.Direction,
+		Protocol:  nftypes.ICMP, // TODO: adjust for IPv6/icmpv6
+		SourceIP:  key.SrcIP,
+		DestIP:    key.DstIP,
+		ICMPType:  conn.ICMPType,
+		ICMPCode:  conn.ICMPCode,
+	})
+}
+
+func (t *ICMPTracker) sendStartEvent(direction nftypes.Direction, key ICMPConnKey, typ, code uint8) {
+	t.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    uuid.New(),
+		Type:      nftypes.TypeStart,
+		Direction: direction,
+		Protocol:  nftypes.ICMP,
+		SourceIP:  key.SrcIP,
+		DestIP:    key.DstIP,
+		ICMPType:  typ,
+		ICMPCode:  code,
+	})
+}
+
 // makeICMPKey creates an ICMP connection key
 func makeICMPKey(srcIP net.IP, dstIP net.IP, id uint16, seq uint16) ICMPConnKey {
+	srcAddr, _ := netip.AddrFromSlice(srcIP)
+	dstAddr, _ := netip.AddrFromSlice(dstIP)
 	return ICMPConnKey{
-		SrcIP:    MakeIPAddr(srcIP),
-		DstIP:    MakeIPAddr(dstIP),
+		SrcIP:    srcAddr,
+		DstIP:    dstAddr,
 		ID:       id,
 		Sequence: seq,
 	}

client/firewall/uspfilter/conntrack/icmp_test.go

@@ -7,7 +7,7 @@ import (
 
 func BenchmarkICMPTracker(b *testing.B) {
 	b.Run("TrackOutbound", func(b *testing.B) {
-		tracker := NewICMPTracker(DefaultICMPTimeout, logger)
+		tracker := NewICMPTracker(DefaultICMPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -15,12 +15,12 @@ func BenchmarkICMPTracker(b *testing.B) {
 
 		b.ResetTimer()
 		for i := 0; i < b.N; i++ {
-			tracker.TrackOutbound(srcIP, dstIP, uint16(i%65535), uint16(i%65535))
+			tracker.TrackOutbound(srcIP, dstIP, uint16(i%65535), uint16(i%65535), 0)
 		}
 	})
 
 	b.Run("IsValidInbound", func(b *testing.B) {
-		tracker := NewICMPTracker(DefaultICMPTimeout, logger)
+		tracker := NewICMPTracker(DefaultICMPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -28,7 +28,7 @@ func BenchmarkICMPTracker(b *testing.B) {
 
 		// Pre-populate some connections
 		for i := 0; i < 1000; i++ {
-			tracker.TrackOutbound(srcIP, dstIP, uint16(i), uint16(i))
+			tracker.TrackOutbound(srcIP, dstIP, uint16(i), uint16(i), 0)
 		}
 
 		b.ResetTimer()

client/firewall/uspfilter/conntrack/tcp.go

@@ -8,7 +8,10 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/google/uuid"
+
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 const (
@@ -39,6 +42,35 @@ const (
 // TCPState represents the state of a TCP connection
 type TCPState int
 
+func (s TCPState) String() string {
+	switch s {
+	case TCPStateNew:
+		return "New"
+	case TCPStateSynSent:
+		return "SYN Sent"
+	case TCPStateSynReceived:
+		return "SYN Received"
+	case TCPStateEstablished:
+		return "Established"
+	case TCPStateFinWait1:
+		return "FIN Wait 1"
+	case TCPStateFinWait2:
+		return "FIN Wait 2"
+	case TCPStateClosing:
+		return "Closing"
+	case TCPStateTimeWait:
+		return "Time Wait"
+	case TCPStateCloseWait:
+		return "Close Wait"
+	case TCPStateLastAck:
+		return "Last ACK"
+	case TCPStateClosed:
+		return "Closed"
+	default:
+		return "Unknown"
+	}
+}
+
 const (
 	TCPStateNew TCPState = iota
 	TCPStateSynSent
@@ -53,19 +85,12 @@ const (
 	TCPStateClosed
 )
 
-// TCPConnKey uniquely identifies a TCP connection
-type TCPConnKey struct {
-	SrcIP   [16]byte
-	DstIP   [16]byte
-	SrcPort uint16
-	DstPort uint16
-}
-
 // TCPConnTrack represents a TCP connection state
 type TCPConnTrack struct {
 	BaseConnTrack
 	State       TCPState
 	established atomic.Bool
+	tombstone   atomic.Bool
 	sync.RWMutex
 }
 
@@ -79,6 +104,16 @@ func (t *TCPConnTrack) SetEstablished(state bool) {
 	t.established.Store(state)
 }
 
+// IsTombstone safely checks if the connection is marked for deletion
+func (t *TCPConnTrack) IsTombstone() bool {
+	return t.tombstone.Load()
+}
+
+// SetTombstone safely marks the connection for deletion
+func (t *TCPConnTrack) SetTombstone() {
+	t.tombstone.Store(true)
+}
+
 // TCPTracker manages TCP connection states
 type TCPTracker struct {
 	logger        *nblog.Logger
@@ -87,68 +122,94 @@ type TCPTracker struct {
 	cleanupTicker *time.Ticker
 	done          chan struct{}
 	timeout       time.Duration
-	ipPool        *PreallocatedIPs
+	flowLogger    nftypes.FlowLogger
 }
 
 // NewTCPTracker creates a new TCP connection tracker
-func NewTCPTracker(timeout time.Duration, logger *nblog.Logger) *TCPTracker {
+func NewTCPTracker(timeout time.Duration, logger *nblog.Logger, flowLogger nftypes.FlowLogger) *TCPTracker {
+	if timeout == 0 {
+		timeout = DefaultTCPTimeout
+	}
+
 	tracker := &TCPTracker{
 		logger:        logger,
 		connections:   make(map[ConnKey]*TCPConnTrack),
 		cleanupTicker: time.NewTicker(TCPCleanupInterval),
 		done:          make(chan struct{}),
 		timeout:       timeout,
-		ipPool:        NewPreallocatedIPs(),
+		flowLogger:    flowLogger,
 	}
 
 	go tracker.cleanupRoutine()
 	return tracker
 }
 
-// TrackOutbound processes an outbound TCP packet and updates connection state
-func (t *TCPTracker) TrackOutbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8) {
-	// Create key before lock
+func (t *TCPTracker) updateIfExists(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8) (ConnKey, bool) {
 	key := makeConnKey(srcIP, dstIP, srcPort, dstPort)
 
-	t.mutex.Lock()
+	t.mutex.RLock()
 	conn, exists := t.connections[key]
-	if !exists {
-		// Use preallocated IPs
-		srcIPCopy := t.ipPool.Get()
-		dstIPCopy := t.ipPool.Get()
-		copyIP(srcIPCopy, srcIP)
-		copyIP(dstIPCopy, dstIP)
+	t.mutex.RUnlock()
 
-		conn = &TCPConnTrack{
-			BaseConnTrack: BaseConnTrack{
-				SourceIP:   srcIPCopy,
-				DestIP:     dstIPCopy,
-				SourcePort: srcPort,
-				DestPort:   dstPort,
-			},
-			State: TCPStateNew,
-		}
+	if exists {
+		conn.Lock()
+		t.updateState(key, conn, flags, conn.Direction == nftypes.Egress)
 		conn.UpdateLastSeen()
-		conn.established.Store(false)
-		t.connections[key] = conn
+		conn.Unlock()
 
-		t.logger.Trace("New TCP connection: %s:%d -> %s:%d", srcIP, srcPort, dstIP, dstPort)
+		return key, true
 	}
+
+	return key, false
+}
+
+// TrackOutbound records an outbound TCP connection
+func (t *TCPTracker) TrackOutbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8) {
+	if _, exists := t.updateIfExists(dstIP, srcIP, dstPort, srcPort, flags); !exists {
+		// if (inverted direction) conn is not tracked, track this direction
+		t.track(srcIP, dstIP, srcPort, dstPort, flags, nftypes.Egress)
+	}
+}
+
+// TrackInbound processes an inbound TCP packet and updates connection state
+func (t *TCPTracker) TrackInbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8) {
+	t.track(srcIP, dstIP, srcPort, dstPort, flags, nftypes.Ingress)
+}
+
+// track is the common implementation for tracking both inbound and outbound connections
+func (t *TCPTracker) track(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8, direction nftypes.Direction) {
+	key, exists := t.updateIfExists(srcIP, dstIP, srcPort, dstPort, flags)
+	if exists {
+		return
+	}
+
+	conn := &TCPConnTrack{
+		BaseConnTrack: BaseConnTrack{
+			FlowId:     uuid.New(),
+			Direction:  direction,
+			SourceIP:   key.SrcIP,
+			DestIP:     key.DstIP,
+			SourcePort: srcPort,
+			DestPort:   dstPort,
+		},
+	}
+
+	conn.UpdateLastSeen()
+	conn.established.Store(false)
+	conn.tombstone.Store(false)
+
+	t.logger.Trace("New %s TCP connection: %s", direction, key)
+	t.updateState(key, conn, flags, direction == nftypes.Egress)
+
+	t.mutex.Lock()
+	t.connections[key] = conn
 	t.mutex.Unlock()
 
-	// Lock individual connection for state update
-	conn.Lock()
-	t.updateState(conn, flags, true)
-	conn.Unlock()
-	conn.UpdateLastSeen()
+	t.sendEvent(nftypes.TypeStart, key, conn)
 }
 
 // IsValidInbound checks if an inbound TCP packet matches a tracked connection
 func (t *TCPTracker) IsValidInbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, flags uint8) bool {
-	if !isValidFlagCombination(flags) {
-		return false
-	}
-
 	key := makeConnKey(dstIP, srcIP, dstPort, srcPort)
 
 	t.mutex.RLock()
@@ -159,21 +220,25 @@ func (t *TCPTracker) IsValidInbound(srcIP net.IP, dstIP net.IP, srcPort uint16,
 		return false
 	}
 
-	// Handle RST packets
+	// Handle RST flag specially - it always causes transition to closed
 	if flags&TCPRst != 0 {
-		conn.Lock()
-		if conn.IsEstablished() || conn.State == TCPStateSynSent || conn.State == TCPStateSynReceived {
-			conn.State = TCPStateClosed
-			conn.SetEstablished(false)
-			conn.Unlock()
+		if conn.IsTombstone() {
 			return true
 		}
+
+		conn.Lock()
+		conn.SetTombstone()
+		conn.State = TCPStateClosed
+		conn.SetEstablished(false)
 		conn.Unlock()
-		return false
+
+		t.logger.Trace("TCP connection reset: %s", key)
+		t.sendEvent(nftypes.TypeEnd, key, conn)
+		return true
 	}
 
 	conn.Lock()
-	t.updateState(conn, flags, false)
+	t.updateState(key, conn, flags, false)
 	conn.UpdateLastSeen()
 	isEstablished := conn.IsEstablished()
 	isValidState := t.isValidStateForFlags(conn.State, flags)
@@ -183,18 +248,15 @@ func (t *TCPTracker) IsValidInbound(srcIP net.IP, dstIP net.IP, srcPort uint16,
 }
 
 // updateState updates the TCP connection state based on flags
-func (t *TCPTracker) updateState(conn *TCPConnTrack, flags uint8, isOutbound bool) {
-	// Handle RST flag specially - it always causes transition to closed
-	if flags&TCPRst != 0 {
-		conn.State = TCPStateClosed
-		conn.SetEstablished(false)
+func (t *TCPTracker) updateState(key ConnKey, conn *TCPConnTrack, flags uint8, isOutbound bool) {
+	state := conn.State
+	defer func() {
+		if state != conn.State {
+			t.logger.Trace("TCP connection %s transitioned from %s to %s", key, state, conn.State)
+		}
+	}()
 
-		t.logger.Trace("TCP connection reset: %s:%d -> %s:%d",
-			conn.SourceIP, conn.SourcePort, conn.DestIP, conn.DestPort)
-		return
-	}
-
-	switch conn.State {
+	switch state {
 	case TCPStateNew:
 		if flags&TCPSyn != 0 && flags&TCPAck == 0 {
 			conn.State = TCPStateSynSent
@@ -203,11 +265,11 @@ func (t *TCPTracker) updateState(conn *TCPConnTrack, flags uint8, isOutbound boo
 	case TCPStateSynSent:
 		if flags&TCPSyn != 0 && flags&TCPAck != 0 {
 			if isOutbound {
-				conn.State = TCPStateSynReceived
-			} else {
-				// Simultaneous open
 				conn.State = TCPStateEstablished
 				conn.SetEstablished(true)
+			} else {
+				// Simultaneous open
+				conn.State = TCPStateSynReceived
 			}
 		}
 
@@ -241,6 +303,9 @@ func (t *TCPTracker) updateState(conn *TCPConnTrack, flags uint8, isOutbound boo
 	case TCPStateFinWait2:
 		if flags&TCPFin != 0 {
 			conn.State = TCPStateTimeWait
+
+			t.logger.Trace("TCP connection %s completed", key)
+			t.sendEvent(nftypes.TypeEnd, key, conn)
 		}
 
 	case TCPStateClosing:
@@ -248,8 +313,8 @@ func (t *TCPTracker) updateState(conn *TCPConnTrack, flags uint8, isOutbound boo
 			conn.State = TCPStateTimeWait
 			// Keep established = false from previous state
 
-			t.logger.Trace("TCP connection closed (simultaneous) - %s:%d -> %s:%d",
-				conn.SourceIP, conn.SourcePort, conn.DestIP, conn.DestPort)
+			t.logger.Trace("TCP connection %s closed (simultaneous)", key)
+			t.sendEvent(nftypes.TypeEnd, key, conn)
 		}
 
 	case TCPStateCloseWait:
@@ -260,17 +325,12 @@ func (t *TCPTracker) updateState(conn *TCPConnTrack, flags uint8, isOutbound boo
 	case TCPStateLastAck:
 		if flags&TCPAck != 0 {
 			conn.State = TCPStateClosed
+			conn.SetTombstone()
 
-			t.logger.Trace("TCP connection gracefully closed: %s:%d -> %s:%d",
-				conn.SourceIP, conn.SourcePort, conn.DestIP, conn.DestPort)
+			// Send close event for gracefully closed connections
+			t.sendEvent(nftypes.TypeEnd, key, conn)
+			t.logger.Trace("TCP connection %s closed gracefully", key)
 		}
-
-	case TCPStateTimeWait:
-		// Stay in TIME-WAIT for 2MSL before transitioning to closed
-		// This is handled by the cleanup routine
-
-		t.logger.Trace("TCP connection completed - %s:%d -> %s:%d",
-			conn.SourceIP, conn.SourcePort, conn.DestIP, conn.DestPort)
 	}
 }
 
@@ -331,6 +391,12 @@ func (t *TCPTracker) cleanup() {
 	defer t.mutex.Unlock()
 
 	for key, conn := range t.connections {
+		if conn.IsTombstone() {
+			// Clean up tombstoned connections without sending an event
+			delete(t.connections, key)
+			continue
+		}
+
 		var timeout time.Duration
 		switch {
 		case conn.State == TCPStateTimeWait:
@@ -341,14 +407,16 @@ func (t *TCPTracker) cleanup() {
 			timeout = TCPHandshakeTimeout
 		}
 
-		lastSeen := conn.GetLastSeen()
-		if time.Since(lastSeen) > timeout {
+		if conn.timeoutExceeded(timeout) {
 			// Return IPs to pool
-			t.ipPool.Put(conn.SourceIP)
-			t.ipPool.Put(conn.DestIP)
 			delete(t.connections, key)
 
-			t.logger.Trace("Cleaned up TCP connection: %s:%d -> %s:%d", conn.SourceIP, conn.SourcePort, conn.DestIP, conn.DestPort)
+			t.logger.Trace("Cleaned up timed-out TCP connection %s", &key)
+
+			// event already handled by state change
+			if conn.State != TCPStateTimeWait {
+				t.sendEvent(nftypes.TypeEnd, key, conn)
+			}
 		}
 	}
 }
@@ -360,10 +428,6 @@ func (t *TCPTracker) Close() {
 
 	// Clean up all remaining IPs
 	t.mutex.Lock()
-	for _, conn := range t.connections {
-		t.ipPool.Put(conn.SourceIP)
-		t.ipPool.Put(conn.DestIP)
-	}
 	t.connections = nil
 	t.mutex.Unlock()
 }
@@ -381,3 +445,16 @@ func isValidFlagCombination(flags uint8) bool {
 
 	return true
 }
+
+func (t *TCPTracker) sendEvent(typ nftypes.Type, key ConnKey, conn *TCPConnTrack) {
+	t.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:     conn.FlowId,
+		Type:       typ,
+		Direction:  conn.Direction,
+		Protocol:   nftypes.TCP,
+		SourceIP:   key.SrcIP,
+		DestIP:     key.DstIP,
+		SourcePort: key.SrcPort,
+		DestPort:   key.DstPort,
+	})
+}

client/firewall/uspfilter/conntrack/tcp_test.go

@@ -9,7 +9,7 @@ import (
 )
 
 func TestTCPStateMachine(t *testing.T) {
-	tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+	tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 	defer tracker.Close()
 
 	srcIP := net.ParseIP("100.64.0.1")
@@ -154,7 +154,7 @@ func TestTCPStateMachine(t *testing.T) {
 			t.Run(tt.name, func(t *testing.T) {
 				t.Helper()
 
-				tracker = NewTCPTracker(DefaultTCPTimeout, logger)
+				tracker = NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 				tt.test(t)
 			})
 		}
@@ -162,7 +162,7 @@ func TestTCPStateMachine(t *testing.T) {
 }
 
 func TestRSTHandling(t *testing.T) {
-	tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+	tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 	defer tracker.Close()
 
 	srcIP := net.ParseIP("100.64.0.1")
@@ -233,7 +233,7 @@ func establishConnection(t *testing.T, tracker *TCPTracker, srcIP, dstIP net.IP,
 
 func BenchmarkTCPTracker(b *testing.B) {
 	b.Run("TrackOutbound", func(b *testing.B) {
-		tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -246,7 +246,7 @@ func BenchmarkTCPTracker(b *testing.B) {
 	})
 
 	b.Run("IsValidInbound", func(b *testing.B) {
-		tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -264,7 +264,7 @@ func BenchmarkTCPTracker(b *testing.B) {
 	})
 
 	b.Run("ConcurrentAccess", func(b *testing.B) {
-		tracker := NewTCPTracker(DefaultTCPTimeout, logger)
+		tracker := NewTCPTracker(DefaultTCPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -287,7 +287,7 @@ func BenchmarkTCPTracker(b *testing.B) {
 // Benchmark connection cleanup
 func BenchmarkCleanup(b *testing.B) {
 	b.Run("TCPCleanup", func(b *testing.B) {
-		tracker := NewTCPTracker(100*time.Millisecond, logger) // Short timeout for testing
+		tracker := NewTCPTracker(100*time.Millisecond, logger, flowLogger) // Short timeout for testing
 		defer tracker.Close()
 
 		// Pre-populate with expired connections

client/firewall/uspfilter/conntrack/udp.go

@@ -5,7 +5,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/google/uuid"
+
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 const (
@@ -28,11 +31,11 @@ type UDPTracker struct {
 	cleanupTicker *time.Ticker
 	mutex         sync.RWMutex
 	done          chan struct{}
-	ipPool        *PreallocatedIPs
+	flowLogger    nftypes.FlowLogger
 }
 
 // NewUDPTracker creates a new UDP connection tracker
-func NewUDPTracker(timeout time.Duration, logger *nblog.Logger) *UDPTracker {
+func NewUDPTracker(timeout time.Duration, logger *nblog.Logger, flowLogger nftypes.FlowLogger) *UDPTracker {
 	if timeout == 0 {
 		timeout = DefaultUDPTimeout
 	}
@@ -43,7 +46,7 @@ func NewUDPTracker(timeout time.Duration, logger *nblog.Logger) *UDPTracker {
 		timeout:       timeout,
 		cleanupTicker: time.NewTicker(UDPCleanupInterval),
 		done:          make(chan struct{}),
-		ipPool:        NewPreallocatedIPs(),
+		flowLogger:    flowLogger,
 	}
 
 	go tracker.cleanupRoutine()
@@ -52,32 +55,57 @@ func NewUDPTracker(timeout time.Duration, logger *nblog.Logger) *UDPTracker {
 
 // TrackOutbound records an outbound UDP connection
 func (t *UDPTracker) TrackOutbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16) {
+	if _, exists := t.updateIfExists(dstIP, srcIP, dstPort, srcPort); !exists {
+		// if (inverted direction) conn is not tracked, track this direction
+		t.track(srcIP, dstIP, srcPort, dstPort, nftypes.Egress)
+	}
+}
+
+// TrackInbound records an inbound UDP connection
+func (t *UDPTracker) TrackInbound(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16) {
+	t.track(srcIP, dstIP, srcPort, dstPort, nftypes.Ingress)
+}
+
+func (t *UDPTracker) updateIfExists(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16) (ConnKey, bool) {
 	key := makeConnKey(srcIP, dstIP, srcPort, dstPort)
 
-	t.mutex.Lock()
+	t.mutex.RLock()
 	conn, exists := t.connections[key]
-	if !exists {
-		srcIPCopy := t.ipPool.Get()
-		dstIPCopy := t.ipPool.Get()
-		copyIP(srcIPCopy, srcIP)
-		copyIP(dstIPCopy, dstIP)
+	t.mutex.RUnlock()
 
-		conn = &UDPConnTrack{
-			BaseConnTrack: BaseConnTrack{
-				SourceIP:   srcIPCopy,
-				DestIP:     dstIPCopy,
-				SourcePort: srcPort,
-				DestPort:   dstPort,
-			},
-		}
+	if exists {
 		conn.UpdateLastSeen()
-		t.connections[key] = conn
-
-		t.logger.Trace("New UDP connection: %v", conn)
+		return key, true
 	}
+
+	return key, false
+}
+
+// track is the common implementation for tracking both inbound and outbound connections
+func (t *UDPTracker) track(srcIP net.IP, dstIP net.IP, srcPort uint16, dstPort uint16, direction nftypes.Direction) {
+	key, exists := t.updateIfExists(srcIP, dstIP, srcPort, dstPort)
+	if exists {
+		return
+	}
+
+	conn := &UDPConnTrack{
+		BaseConnTrack: BaseConnTrack{
+			FlowId:     uuid.New(),
+			Direction:  direction,
+			SourceIP:   key.SrcIP,
+			DestIP:     key.DstIP,
+			SourcePort: srcPort,
+			DestPort:   dstPort,
+		},
+	}
+	conn.UpdateLastSeen()
+
+	t.mutex.Lock()
+	t.connections[key] = conn
 	t.mutex.Unlock()
 
-	conn.UpdateLastSeen()
+	t.logger.Trace("New %s UDP connection: %s", direction, key)
+	t.sendEvent(nftypes.TypeStart, key, conn)
 }
 
 // IsValidInbound checks if an inbound packet matches a tracked connection
@@ -88,18 +116,13 @@ func (t *UDPTracker) IsValidInbound(srcIP net.IP, dstIP net.IP, srcPort uint16,
 	conn, exists := t.connections[key]
 	t.mutex.RUnlock()
 
-	if !exists {
+	if !exists || conn.timeoutExceeded(t.timeout) {
 		return false
 	}
 
-	if conn.timeoutExceeded(t.timeout) {
-		return false
-	}
+	conn.UpdateLastSeen()
 
-	return ValidateIPs(MakeIPAddr(srcIP), conn.DestIP) &&
-		ValidateIPs(MakeIPAddr(dstIP), conn.SourceIP) &&
-		conn.DestPort == srcPort &&
-		conn.SourcePort == dstPort
+	return true
 }
 
 // cleanupRoutine periodically removes stale connections
@@ -120,11 +143,10 @@ func (t *UDPTracker) cleanup() {
 
 	for key, conn := range t.connections {
 		if conn.timeoutExceeded(t.timeout) {
-			t.ipPool.Put(conn.SourceIP)
-			t.ipPool.Put(conn.DestIP)
 			delete(t.connections, key)
 
-			t.logger.Trace("Removed UDP connection %v (timeout)", conn)
+			t.logger.Trace("Removed UDP connection %s (timeout)", key)
+			t.sendEvent(nftypes.TypeEnd, key, conn)
 		}
 	}
 }
@@ -135,10 +157,6 @@ func (t *UDPTracker) Close() {
 	close(t.done)
 
 	t.mutex.Lock()
-	for _, conn := range t.connections {
-		t.ipPool.Put(conn.SourceIP)
-		t.ipPool.Put(conn.DestIP)
-	}
 	t.connections = nil
 	t.mutex.Unlock()
 }
@@ -150,14 +168,23 @@ func (t *UDPTracker) GetConnection(srcIP net.IP, srcPort uint16, dstIP net.IP, d
 
 	key := makeConnKey(srcIP, dstIP, srcPort, dstPort)
 	conn, exists := t.connections[key]
-	if !exists {
-		return nil, false
-	}
-
-	return conn, true
+	return conn, exists
 }
 
 // Timeout returns the configured timeout duration for the tracker
 func (t *UDPTracker) Timeout() time.Duration {
 	return t.timeout
 }
+
+func (t *UDPTracker) sendEvent(typ nftypes.Type, key ConnKey, conn *UDPConnTrack) {
+	t.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:     conn.FlowId,
+		Type:       typ,
+		Direction:  conn.Direction,
+		Protocol:   nftypes.UDP,
+		SourceIP:   key.SrcIP,
+		DestIP:     key.DstIP,
+		SourcePort: key.SrcPort,
+		DestPort:   key.DstPort,
+	})
+}

client/firewall/uspfilter/conntrack/udp_test.go

@@ -2,6 +2,7 @@ package conntrack
 
 import (
 	"net"
+	"net/netip"
 	"testing"
 	"time"
 
@@ -29,7 +30,7 @@ func TestNewUDPTracker(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			tracker := NewUDPTracker(tt.timeout, logger)
+			tracker := NewUDPTracker(tt.timeout, logger, flowLogger)
 			assert.NotNil(t, tracker)
 			assert.Equal(t, tt.wantTimeout, tracker.timeout)
 			assert.NotNil(t, tracker.connections)
@@ -40,29 +41,29 @@ func TestNewUDPTracker(t *testing.T) {
 }
 
 func TestUDPTracker_TrackOutbound(t *testing.T) {
-	tracker := NewUDPTracker(DefaultUDPTimeout, logger)
+	tracker := NewUDPTracker(DefaultUDPTimeout, logger, flowLogger)
 	defer tracker.Close()
 
-	srcIP := net.ParseIP("192.168.1.2")
-	dstIP := net.ParseIP("192.168.1.3")
+	srcIP := netip.MustParseAddr("192.168.1.2")
+	dstIP := netip.MustParseAddr("192.168.1.3")
 	srcPort := uint16(12345)
 	dstPort := uint16(53)
 
-	tracker.TrackOutbound(srcIP, dstIP, srcPort, dstPort)
+	tracker.TrackOutbound(srcIP.AsSlice(), dstIP.AsSlice(), srcPort, dstPort)
 
 	// Verify connection was tracked
-	key := makeConnKey(srcIP, dstIP, srcPort, dstPort)
+	key := makeConnKey(srcIP.AsSlice(), dstIP.AsSlice(), srcPort, dstPort)
 	conn, exists := tracker.connections[key]
 	require.True(t, exists)
-	assert.True(t, conn.SourceIP.Equal(srcIP))
-	assert.True(t, conn.DestIP.Equal(dstIP))
+	assert.True(t, conn.SourceIP.Compare(srcIP) == 0)
+	assert.True(t, conn.DestIP.Compare(dstIP) == 0)
 	assert.Equal(t, srcPort, conn.SourcePort)
 	assert.Equal(t, dstPort, conn.DestPort)
 	assert.WithinDuration(t, time.Now(), conn.GetLastSeen(), 1*time.Second)
 }
 
 func TestUDPTracker_IsValidInbound(t *testing.T) {
-	tracker := NewUDPTracker(1*time.Second, logger)
+	tracker := NewUDPTracker(1*time.Second, logger, flowLogger)
 	defer tracker.Close()
 
 	srcIP := net.ParseIP("192.168.1.2")
@@ -160,8 +161,8 @@ func TestUDPTracker_Cleanup(t *testing.T) {
 		timeout:       timeout,
 		cleanupTicker: time.NewTicker(cleanupInterval),
 		done:          make(chan struct{}),
-		ipPool:        NewPreallocatedIPs(),
 		logger:        logger,
+		flowLogger:    flowLogger,
 	}
 
 	// Start cleanup routine
@@ -211,7 +212,7 @@ func TestUDPTracker_Cleanup(t *testing.T) {
 
 func BenchmarkUDPTracker(b *testing.B) {
 	b.Run("TrackOutbound", func(b *testing.B) {
-		tracker := NewUDPTracker(DefaultUDPTimeout, logger)
+		tracker := NewUDPTracker(DefaultUDPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")
@@ -224,7 +225,7 @@ func BenchmarkUDPTracker(b *testing.B) {
 	})
 
 	b.Run("IsValidInbound", func(b *testing.B) {
-		tracker := NewUDPTracker(DefaultUDPTimeout, logger)
+		tracker := NewUDPTracker(DefaultUDPTimeout, logger, flowLogger)
 		defer tracker.Close()
 
 		srcIP := net.ParseIP("192.168.1.1")

client/firewall/uspfilter/forwarder/endpoint.go

@@ -1,6 +1,8 @@
 package forwarder
 
 import (
+	"fmt"
+
 	wgdevice "golang.zx2c4.com/wireguard/device"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
@@ -79,3 +81,10 @@ func (e *endpoint) AddHeader(*stack.PacketBuffer) {
 func (e *endpoint) ParseHeader(*stack.PacketBuffer) bool {
 	return true
 }
+
+type epID stack.TransportEndpointID
+
+func (i epID) String() string {
+	// src and remote is swapped
+	return fmt.Sprintf("%s:%d -> %s:%d", i.RemoteAddress, i.RemotePort, i.LocalAddress, i.LocalPort)
+}

client/firewall/uspfilter/forwarder/forwarder.go

@@ -18,6 +18,7 @@ import (
 
 	"github.com/netbirdio/netbird/client/firewall/uspfilter/common"
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 const (
@@ -29,6 +30,7 @@ const (
 
 type Forwarder struct {
 	logger       *nblog.Logger
+	flowLogger   nftypes.FlowLogger
 	stack        *stack.Stack
 	endpoint     *endpoint
 	udpForwarder *udpForwarder
@@ -38,7 +40,7 @@ type Forwarder struct {
 	netstack     bool
 }
 
-func New(iface common.IFaceMapper, logger *nblog.Logger, netstack bool) (*Forwarder, error) {
+func New(iface common.IFaceMapper, logger *nblog.Logger, flowLogger nftypes.FlowLogger, netstack bool) (*Forwarder, error) {
 	s := stack.New(stack.Options{
 		NetworkProtocols: []stack.NetworkProtocolFactory{ipv4.NewProtocol},
 		TransportProtocols: []stack.TransportProtocolFactory{
@@ -102,9 +104,10 @@ func New(iface common.IFaceMapper, logger *nblog.Logger, netstack bool) (*Forwar
 	ctx, cancel := context.WithCancel(context.Background())
 	f := &Forwarder{
 		logger:       logger,
+		flowLogger:   flowLogger,
 		stack:        s,
 		endpoint:     endpoint,
-		udpForwarder: newUDPForwarder(mtu, logger),
+		udpForwarder: newUDPForwarder(mtu, logger, flowLogger),
 		ctx:          ctx,
 		cancel:       cancel,
 		netstack:     netstack,

client/firewall/uspfilter/forwarder/icmp.go

@@ -3,14 +3,27 @@ package forwarder
 import (
 	"context"
 	"net"
+	"net/netip"
 	"time"
 
+	"github.com/google/uuid"
 	"gvisor.dev/gvisor/pkg/tcpip/header"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
+
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 // handleICMP handles ICMP packets from the network stack
 func (f *Forwarder) handleICMP(id stack.TransportEndpointID, pkt stack.PacketBufferPtr) bool {
+	flowID := uuid.New()
+
+	// Extract ICMP header to get type and code
+	icmpHdr := header.ICMPv4(pkt.TransportHeader().View().AsSlice())
+	icmpType := uint8(icmpHdr.Type())
+	icmpCode := uint8(icmpHdr.Code())
+
+	f.sendICMPEvent(nftypes.TypeStart, flowID, id, icmpType, icmpCode)
+
 	ctx, cancel := context.WithTimeout(f.ctx, 5*time.Second)
 	defer cancel()
 
@@ -18,7 +31,9 @@ func (f *Forwarder) handleICMP(id stack.TransportEndpointID, pkt stack.PacketBuf
 	// TODO: support non-root
 	conn, err := lc.ListenPacket(ctx, "ip4:icmp", "0.0.0.0")
 	if err != nil {
-		f.logger.Error("Failed to create ICMP socket for %v: %v", id, err)
+		f.logger.Error("Failed to create ICMP socket for %v: %v", epID(id), err)
+
+		f.sendICMPEvent(nftypes.TypeEnd, flowID, id, icmpType, icmpCode)
 
 		// This will make netstack reply on behalf of the original destination, that's ok for now
 		return false
@@ -27,6 +42,8 @@ func (f *Forwarder) handleICMP(id stack.TransportEndpointID, pkt stack.PacketBuf
 		if err := conn.Close(); err != nil {
 			f.logger.Debug("Failed to close ICMP socket: %v", err)
 		}
+
+		f.sendICMPEvent(nftypes.TypeEnd, flowID, id, icmpType, icmpCode)
 	}()
 
 	dstIP := f.determineDialAddr(id.LocalAddress)
@@ -36,12 +53,10 @@ func (f *Forwarder) handleICMP(id stack.TransportEndpointID, pkt stack.PacketBuf
 	fullPacket := stack.PayloadSince(pkt.TransportHeader())
 	payload := fullPacket.AsSlice()
 
-	icmpHdr := header.ICMPv4(pkt.TransportHeader().View().AsSlice())
-
 	// For Echo Requests, send and handle response
 	switch icmpHdr.Type() {
 	case header.ICMPv4Echo:
-		return f.handleEchoResponse(icmpHdr, payload, dst, conn, id)
+		return f.handleEchoResponse(icmpHdr, payload, dst, conn, id, flowID)
 	case header.ICMPv4EchoReply:
 		// dont process our own replies
 		return true
@@ -51,24 +66,24 @@ func (f *Forwarder) handleICMP(id stack.TransportEndpointID, pkt stack.PacketBuf
 	// For other ICMP types (Time Exceeded, Destination Unreachable, etc)
 	_, err = conn.WriteTo(payload, dst)
 	if err != nil {
-		f.logger.Error("Failed to write ICMP packet for %v: %v", id, err)
+		f.logger.Error("Failed to write ICMP packet for %v: %v", epID(id), err)
 		return true
 	}
 
-	f.logger.Trace("Forwarded ICMP packet %v type=%v code=%v",
-		id, icmpHdr.Type(), icmpHdr.Code())
+	f.logger.Trace("Forwarded ICMP packet %v type %v code %v",
+		epID(id), icmpHdr.Type(), icmpHdr.Code())
 
 	return true
 }
 
-func (f *Forwarder) handleEchoResponse(icmpHdr header.ICMPv4, payload []byte, dst *net.IPAddr, conn net.PacketConn, id stack.TransportEndpointID) bool {
+func (f *Forwarder) handleEchoResponse(icmpHdr header.ICMPv4, payload []byte, dst *net.IPAddr, conn net.PacketConn, id stack.TransportEndpointID, flowID uuid.UUID) bool {
 	if _, err := conn.WriteTo(payload, dst); err != nil {
-		f.logger.Error("Failed to write ICMP packet for %v: %v", id, err)
+		f.logger.Error("Failed to write ICMP packet for %v: %v", epID(id), err)
 		return true
 	}
 
-	f.logger.Trace("Forwarded ICMP packet %v type=%v code=%v",
-		id, icmpHdr.Type(), icmpHdr.Code())
+	f.logger.Trace("Forwarded ICMP packet %v type %v code %v",
+		epID(id), icmpHdr.Type(), icmpHdr.Code())
 
 	if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
 		f.logger.Error("Failed to set read deadline for ICMP response: %v", err)
@@ -101,9 +116,27 @@ func (f *Forwarder) handleEchoResponse(icmpHdr header.ICMPv4, payload []byte, ds
 
 	if err := f.InjectIncomingPacket(fullPacket); err != nil {
 		f.logger.Error("Failed to inject ICMP response: %v", err)
+
 		return true
 	}
 
-	f.logger.Trace("Forwarded ICMP echo reply for %v", id)
+	f.logger.Trace("Forwarded ICMP echo reply for %v type %v code %v",
+		epID(id), icmpHdr.Type(), icmpHdr.Code())
+
 	return true
 }
+
+// sendICMPEvent stores flow events for ICMP packets
+func (f *Forwarder) sendICMPEvent(typ nftypes.Type, flowID uuid.UUID, id stack.TransportEndpointID, icmpType, icmpCode uint8) {
+	f.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    flowID,
+		Type:      typ,
+		Direction: nftypes.Ingress,
+		Protocol:  nftypes.ICMP,
+		// TODO: handle ipv6
+		SourceIP: netip.AddrFrom4(id.LocalAddress.As4()),
+		DestIP:   netip.AddrFrom4(id.RemoteAddress.As4()),
+		ICMPType: icmpType,
+		ICMPCode: icmpCode,
+	})
+}

client/firewall/uspfilter/forwarder/tcp.go

@@ -5,24 +5,31 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"net/netip"
 
+	"github.com/google/uuid"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
 	"gvisor.dev/gvisor/pkg/tcpip/transport/tcp"
 	"gvisor.dev/gvisor/pkg/waiter"
+
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 // handleTCP is called by the TCP forwarder for new connections.
 func (f *Forwarder) handleTCP(r *tcp.ForwarderRequest) {
 	id := r.ID()
 
+	flowID := uuid.New()
+	f.sendTCPEvent(nftypes.TypeStart, flowID, id)
+
 	dialAddr := fmt.Sprintf("%s:%d", f.determineDialAddr(id.LocalAddress), id.LocalPort)
 
 	outConn, err := (&net.Dialer{}).DialContext(f.ctx, "tcp", dialAddr)
 	if err != nil {
 		r.Complete(true)
-		f.logger.Trace("forwarder: dial error for %v: %v", id, err)
+		f.logger.Trace("forwarder: dial error for %v: %v", epID(id), err)
 		return
 	}
 
@@ -44,12 +51,12 @@ func (f *Forwarder) handleTCP(r *tcp.ForwarderRequest) {
 
 	inConn := gonet.NewTCPConn(&wq, ep)
 
-	f.logger.Trace("forwarder: established TCP connection %v", id)
+	f.logger.Trace("forwarder: established TCP connection %v", epID(id))
 
-	go f.proxyTCP(id, inConn, outConn, ep)
+	go f.proxyTCP(id, inConn, outConn, ep, flowID)
 }
 
-func (f *Forwarder) proxyTCP(id stack.TransportEndpointID, inConn *gonet.TCPConn, outConn net.Conn, ep tcpip.Endpoint) {
+func (f *Forwarder) proxyTCP(id stack.TransportEndpointID, inConn *gonet.TCPConn, outConn net.Conn, ep tcpip.Endpoint, flowID uuid.UUID) {
 	defer func() {
 		if err := inConn.Close(); err != nil {
 			f.logger.Debug("forwarder: inConn close error: %v", err)
@@ -58,6 +65,8 @@ func (f *Forwarder) proxyTCP(id stack.TransportEndpointID, inConn *gonet.TCPConn
 			f.logger.Debug("forwarder: outConn close error: %v", err)
 		}
 		ep.Close()
+
+		f.sendTCPEvent(nftypes.TypeEnd, flowID, id)
 	}()
 
 	// Create context for managing the proxy goroutines
@@ -78,13 +87,28 @@ func (f *Forwarder) proxyTCP(id stack.TransportEndpointID, inConn *gonet.TCPConn
 
 	select {
 	case <-ctx.Done():
-		f.logger.Trace("forwarder: tearing down TCP connection %v due to context done", id)
+		f.logger.Trace("forwarder: tearing down TCP connection %v due to context done", epID(id))
 		return
 	case err := <-errChan:
 		if err != nil && !isClosedError(err) {
 			f.logger.Error("proxyTCP: copy error: %v", err)
 		}
-		f.logger.Trace("forwarder: tearing down TCP connection %v", id)
+		f.logger.Trace("forwarder: tearing down TCP connection %v", epID(id))
 		return
 	}
 }
+
+func (f *Forwarder) sendTCPEvent(typ nftypes.Type, flowID uuid.UUID, id stack.TransportEndpointID) {
+
+	f.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    flowID,
+		Type:      typ,
+		Direction: nftypes.Ingress,
+		Protocol:  6,
+		// TODO: handle ipv6
+		SourceIP:   netip.AddrFrom4(id.LocalAddress.As4()),
+		DestIP:     netip.AddrFrom4(id.RemoteAddress.As4()),
+		SourcePort: id.LocalPort,
+		DestPort:   id.RemotePort,
+	})
+}

client/firewall/uspfilter/forwarder/udp.go

@@ -5,10 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"net/netip"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/google/uuid"
 	"gvisor.dev/gvisor/pkg/tcpip"
 	"gvisor.dev/gvisor/pkg/tcpip/adapters/gonet"
 	"gvisor.dev/gvisor/pkg/tcpip/stack"
@@ -16,6 +18,7 @@ import (
 	"gvisor.dev/gvisor/pkg/waiter"
 
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 )
 
 const (
@@ -28,15 +31,17 @@ type udpPacketConn struct {
 	lastSeen atomic.Int64
 	cancel   context.CancelFunc
 	ep       tcpip.Endpoint
+	flowID   uuid.UUID
 }
 
 type udpForwarder struct {
 	sync.RWMutex
-	logger  *nblog.Logger
-	conns   map[stack.TransportEndpointID]*udpPacketConn
-	bufPool sync.Pool
-	ctx     context.Context
-	cancel  context.CancelFunc
+	logger     *nblog.Logger
+	flowLogger nftypes.FlowLogger
+	conns      map[stack.TransportEndpointID]*udpPacketConn
+	bufPool    sync.Pool
+	ctx        context.Context
+	cancel     context.CancelFunc
 }
 
 type idleConn struct {
@@ -44,13 +49,14 @@ type idleConn struct {
 	conn *udpPacketConn
 }
 
-func newUDPForwarder(mtu int, logger *nblog.Logger) *udpForwarder {
+func newUDPForwarder(mtu int, logger *nblog.Logger, flowLogger nftypes.FlowLogger) *udpForwarder {
 	ctx, cancel := context.WithCancel(context.Background())
 	f := &udpForwarder{
-		logger: logger,
-		conns:  make(map[stack.TransportEndpointID]*udpPacketConn),
-		ctx:    ctx,
-		cancel: cancel,
+		logger:     logger,
+		flowLogger: flowLogger,
+		conns:      make(map[stack.TransportEndpointID]*udpPacketConn),
+		ctx:        ctx,
+		cancel:     cancel,
 		bufPool: sync.Pool{
 			New: func() any {
 				b := make([]byte, mtu)
@@ -72,10 +78,10 @@ func (f *udpForwarder) Stop() {
 	for id, conn := range f.conns {
 		conn.cancel()
 		if err := conn.conn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP conn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP conn close error for %v: %v", epID(id), err)
 		}
 		if err := conn.outConn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", epID(id), err)
 		}
 
 		conn.ep.Close()
@@ -83,6 +89,21 @@ func (f *udpForwarder) Stop() {
 	}
 }
 
+// sendUDPEvent stores flow events for UDP connections
+func (f *udpForwarder) sendUDPEvent(typ nftypes.Type, flowID uuid.UUID, id stack.TransportEndpointID) {
+	f.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    flowID,
+		Type:      typ,
+		Direction: nftypes.Ingress,
+		Protocol:  17,
+		// TODO: handle ipv6
+		SourceIP:   netip.AddrFrom4(id.LocalAddress.As4()),
+		DestIP:     netip.AddrFrom4(id.RemoteAddress.As4()),
+		SourcePort: id.LocalPort,
+		DestPort:   id.RemotePort,
+	})
+}
+
 // cleanup periodically removes idle UDP connections
 func (f *udpForwarder) cleanup() {
 	ticker := time.NewTicker(time.Minute)
@@ -106,10 +127,10 @@ func (f *udpForwarder) cleanup() {
 			for _, idle := range idleConns {
 				idle.conn.cancel()
 				if err := idle.conn.conn.Close(); err != nil {
-					f.logger.Debug("forwarder: UDP conn close error for %v: %v", idle.id, err)
+					f.logger.Debug("forwarder: UDP conn close error for %v: %v", epID(idle.id), err)
 				}
 				if err := idle.conn.outConn.Close(); err != nil {
-					f.logger.Debug("forwarder: UDP outConn close error for %v: %v", idle.id, err)
+					f.logger.Debug("forwarder: UDP outConn close error for %v: %v", epID(idle.id), err)
 				}
 
 				idle.conn.ep.Close()
@@ -118,7 +139,9 @@ func (f *udpForwarder) cleanup() {
 				delete(f.conns, idle.id)
 				f.Unlock()
 
-				f.logger.Trace("forwarder: cleaned up idle UDP connection %v", idle.id)
+				f.logger.Trace("forwarder: cleaned up idle UDP connection %v", epID(idle.id))
+
+				f.sendUDPEvent(nftypes.TypeEnd, idle.conn.flowID, idle.id)
 			}
 		}
 	}
@@ -137,14 +160,18 @@ func (f *Forwarder) handleUDP(r *udp.ForwarderRequest) {
 	_, exists := f.udpForwarder.conns[id]
 	f.udpForwarder.RUnlock()
 	if exists {
-		f.logger.Trace("forwarder: existing UDP connection for %v", id)
+		f.logger.Trace("forwarder: existing UDP connection for %v", epID(id))
 		return
 	}
 
+	flowID := uuid.New()
+	f.sendUDPEvent(nftypes.TypeStart, flowID, id)
+
 	dstAddr := fmt.Sprintf("%s:%d", f.determineDialAddr(id.LocalAddress), id.LocalPort)
 	outConn, err := (&net.Dialer{}).DialContext(f.ctx, "udp", dstAddr)
 	if err != nil {
-		f.logger.Debug("forwarder: UDP dial error for %v: %v", id, err)
+		f.logger.Debug("forwarder: UDP dial error for %v: %v", epID(id), err)
+		f.sendUDPEvent(nftypes.TypeEnd, flowID, id)
 		// TODO: Send ICMP error message
 		return
 	}
@@ -155,8 +182,9 @@ func (f *Forwarder) handleUDP(r *udp.ForwarderRequest) {
 	if epErr != nil {
 		f.logger.Debug("forwarder: failed to create UDP endpoint: %v", epErr)
 		if err := outConn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", epID(id), err)
 		}
+		f.sendUDPEvent(nftypes.TypeEnd, flowID, id)
 		return
 	}
 
@@ -168,6 +196,7 @@ func (f *Forwarder) handleUDP(r *udp.ForwarderRequest) {
 		outConn: outConn,
 		cancel:  connCancel,
 		ep:      ep,
+		flowID:  flowID,
 	}
 	pConn.updateLastSeen()
 
@@ -177,17 +206,19 @@ func (f *Forwarder) handleUDP(r *udp.ForwarderRequest) {
 		f.udpForwarder.Unlock()
 		pConn.cancel()
 		if err := inConn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP inConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP inConn close error for %v: %v", epID(id), err)
 		}
 		if err := outConn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", epID(id), err)
 		}
+
+		f.sendUDPEvent(nftypes.TypeEnd, flowID, id)
 		return
 	}
 	f.udpForwarder.conns[id] = pConn
 	f.udpForwarder.Unlock()
 
-	f.logger.Trace("forwarder: established UDP connection to %v", id)
+	f.logger.Trace("forwarder: established UDP connection %v", epID(id))
 	go f.proxyUDP(connCtx, pConn, id, ep)
 }
 
@@ -195,10 +226,10 @@ func (f *Forwarder) proxyUDP(ctx context.Context, pConn *udpPacketConn, id stack
 	defer func() {
 		pConn.cancel()
 		if err := pConn.conn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP inConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP inConn close error for %v: %v", epID(id), err)
 		}
 		if err := pConn.outConn.Close(); err != nil {
-			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", id, err)
+			f.logger.Debug("forwarder: UDP outConn close error for %v: %v", epID(id), err)
 		}
 
 		ep.Close()
@@ -206,6 +237,8 @@ func (f *Forwarder) proxyUDP(ctx context.Context, pConn *udpPacketConn, id stack
 		f.udpForwarder.Lock()
 		delete(f.udpForwarder.conns, id)
 		f.udpForwarder.Unlock()
+
+		f.sendUDPEvent(nftypes.TypeEnd, pConn.flowID, id)
 	}()
 
 	errChan := make(chan error, 2)
@@ -220,17 +253,32 @@ func (f *Forwarder) proxyUDP(ctx context.Context, pConn *udpPacketConn, id stack
 
 	select {
 	case <-ctx.Done():
-		f.logger.Trace("forwarder: tearing down UDP connection %v due to context done", id)
+		f.logger.Trace("forwarder: tearing down UDP connection %v due to context done", epID(id))
 		return
 	case err := <-errChan:
 		if err != nil && !isClosedError(err) {
 			f.logger.Error("proxyUDP: copy error: %v", err)
 		}
-		f.logger.Trace("forwarder: tearing down UDP connection %v", id)
+		f.logger.Trace("forwarder: tearing down UDP connection %v", epID(id))
 		return
 	}
 }
 
+// sendUDPEvent stores flow events for UDP connections, mirrors the TCP version
+func (f *Forwarder) sendUDPEvent(typ nftypes.Type, flowID uuid.UUID, id stack.TransportEndpointID) {
+	f.flowLogger.StoreEvent(nftypes.EventFields{
+		FlowID:    flowID,
+		Type:      typ,
+		Direction: nftypes.Ingress,
+		Protocol:  17, // UDP protocol number
+		// TODO: handle ipv6
+		SourceIP:   netip.AddrFrom4(id.LocalAddress.As4()),
+		DestIP:     netip.AddrFrom4(id.RemoteAddress.As4()),
+		SourcePort: id.LocalPort,
+		DestPort:   id.RemotePort,
+	})
+}
+
 func (c *udpPacketConn) updateLastSeen() {
 	c.lastSeen.Store(time.Now().UnixNano())
 }

client/firewall/uspfilter/log/log.go

@@ -1,4 +1,4 @@
-// Package logger provides a high-performance, non-blocking logger for userspace networking
+// Package log provides a high-performance, non-blocking logger for userspace networking
 package log
 
 import (
@@ -13,13 +13,12 @@ import (
 )
 
 const (
-	maxBatchSize         = 1024 * 16  // 16KB max batch size
-	maxMessageSize       = 1024 * 2   // 2KB per message
-	bufferSize           = 1024 * 256 // 256KB ring buffer
+	maxBatchSize         = 1024 * 16
+	maxMessageSize       = 1024 * 2
 	defaultFlushInterval = 2 * time.Second
+	logChannelSize       = 1000
 )
 
-// Level represents log severity
 type Level uint32
 
 const (
@@ -42,32 +41,37 @@ var levelStrings = map[Level]string{
 	LevelTrace: "TRAC",
 }
 
-// Logger is a high-performance, non-blocking logger
-type Logger struct {
-	output    io.Writer
-	level     atomic.Uint32
-	buffer    *ringBuffer
-	shutdown  chan struct{}
-	closeOnce sync.Once
-	wg        sync.WaitGroup
-
-	// Reusable buffer pool for formatting messages
-	bufPool sync.Pool
+type logMessage struct {
+	level  Level
+	format string
+	args   []any
 }
 
+// Logger is a high-performance, non-blocking logger
+type Logger struct {
+	output     io.Writer
+	level      atomic.Uint32
+	msgChannel chan logMessage
+	shutdown   chan struct{}
+	closeOnce  sync.Once
+	wg         sync.WaitGroup
+	bufPool    sync.Pool
+}
+
+// NewFromLogrus creates a new Logger that writes to the same output as the given logrus logger
 func NewFromLogrus(logrusLogger *log.Logger) *Logger {
 	l := &Logger{
-		output:   logrusLogger.Out,
-		buffer:   newRingBuffer(bufferSize),
-		shutdown: make(chan struct{}),
+		output:     logrusLogger.Out,
+		msgChannel: make(chan logMessage, logChannelSize),
+		shutdown:   make(chan struct{}),
 		bufPool: sync.Pool{
-			New: func() interface{} {
-				// Pre-allocate buffer for message formatting
+			New: func() any {
 				b := make([]byte, 0, maxMessageSize)
 				return &b
 			},
 		},
 	}
+
 	logrusLevel := logrusLogger.GetLevel()
 	l.level.Store(uint32(logrusLevel))
 	level := levelStrings[Level(logrusLevel)]
@@ -79,97 +83,149 @@ func NewFromLogrus(logrusLogger *log.Logger) *Logger {
 	return l
 }
 
+// SetLevel sets the logging level
 func (l *Logger) SetLevel(level Level) {
 	l.level.Store(uint32(level))
-
 	log.Debugf("Set uspfilter logger loglevel to %v", levelStrings[level])
 }
 
-func (l *Logger) formatMessage(buf *[]byte, level Level, format string, args ...interface{}) {
-	*buf = (*buf)[:0]
-
-	// Timestamp
-	*buf = time.Now().AppendFormat(*buf, "2006-01-02T15:04:05-07:00")
-	*buf = append(*buf, ' ')
-
-	// Level
-	*buf = append(*buf, levelStrings[level]...)
-	*buf = append(*buf, ' ')
-
-	// Message
-	if len(args) > 0 {
-		*buf = append(*buf, fmt.Sprintf(format, args...)...)
-	} else {
-		*buf = append(*buf, format...)
+func (l *Logger) log(level Level, format string, args ...any) {
+	select {
+	case l.msgChannel <- logMessage{level: level, format: format, args: args}:
+	default:
 	}
-
-	*buf = append(*buf, '\n')
 }
 
-func (l *Logger) log(level Level, format string, args ...interface{}) {
-	bufp := l.bufPool.Get().(*[]byte)
-	l.formatMessage(bufp, level, format, args...)
-
-	if len(*bufp) > maxMessageSize {
-		*bufp = (*bufp)[:maxMessageSize]
-	}
-	_, _ = l.buffer.Write(*bufp)
-
-	l.bufPool.Put(bufp)
-}
-
-func (l *Logger) Error(format string, args ...interface{}) {
+// Error logs a message at error level
+func (l *Logger) Error(format string, args ...any) {
 	if l.level.Load() >= uint32(LevelError) {
 		l.log(LevelError, format, args...)
 	}
 }
 
-func (l *Logger) Warn(format string, args ...interface{}) {
+// Warn logs a message at warning level
+func (l *Logger) Warn(format string, args ...any) {
 	if l.level.Load() >= uint32(LevelWarn) {
 		l.log(LevelWarn, format, args...)
 	}
 }
 
-func (l *Logger) Info(format string, args ...interface{}) {
+// Info logs a message at info level
+func (l *Logger) Info(format string, args ...any) {
 	if l.level.Load() >= uint32(LevelInfo) {
 		l.log(LevelInfo, format, args...)
 	}
 }
 
-func (l *Logger) Debug(format string, args ...interface{}) {
+// Debug logs a message at debug level
+func (l *Logger) Debug(format string, args ...any) {
 	if l.level.Load() >= uint32(LevelDebug) {
 		l.log(LevelDebug, format, args...)
 	}
 }
 
-func (l *Logger) Trace(format string, args ...interface{}) {
+// Trace logs a message at trace level
+func (l *Logger) Trace(format string, args ...any) {
 	if l.level.Load() >= uint32(LevelTrace) {
 		l.log(LevelTrace, format, args...)
 	}
 }
 
-// worker periodically flushes the buffer
+func (l *Logger) formatMessage(buf *[]byte, level Level, format string, args ...any) {
+	*buf = (*buf)[:0]
+	*buf = time.Now().AppendFormat(*buf, "2006-01-02T15:04:05-07:00")
+	*buf = append(*buf, ' ')
+	*buf = append(*buf, levelStrings[level]...)
+	*buf = append(*buf, ' ')
+
+	var msg string
+	if len(args) > 0 {
+		msg = fmt.Sprintf(format, args...)
+	} else {
+		msg = format
+	}
+	*buf = append(*buf, msg...)
+	*buf = append(*buf, '\n')
+
+	if len(*buf) > maxMessageSize {
+		*buf = (*buf)[:maxMessageSize]
+	}
+}
+
+// processMessage handles a single log message and adds it to the buffer
+func (l *Logger) processMessage(msg logMessage, buffer *[]byte) {
+	bufp := l.bufPool.Get().(*[]byte)
+	defer l.bufPool.Put(bufp)
+
+	l.formatMessage(bufp, msg.level, msg.format, msg.args...)
+
+	if len(*buffer)+len(*bufp) > maxBatchSize {
+		_, _ = l.output.Write(*buffer)
+		*buffer = (*buffer)[:0]
+	}
+	*buffer = append(*buffer, *bufp...)
+}
+
+// flushBuffer writes the accumulated buffer to output
+func (l *Logger) flushBuffer(buffer *[]byte) {
+	if len(*buffer) > 0 {
+		_, _ = l.output.Write(*buffer)
+		*buffer = (*buffer)[:0]
+	}
+}
+
+// processBatch processes as many messages as possible without blocking
+func (l *Logger) processBatch(buffer *[]byte) {
+	for len(*buffer) < maxBatchSize {
+		select {
+		case msg := <-l.msgChannel:
+			l.processMessage(msg, buffer)
+		default:
+			return
+		}
+	}
+}
+
+// handleShutdown manages the graceful shutdown sequence with timeout
+func (l *Logger) handleShutdown(buffer *[]byte) {
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	for {
+		select {
+		case msg := <-l.msgChannel:
+			l.processMessage(msg, buffer)
+		case <-ctx.Done():
+			l.flushBuffer(buffer)
+			return
+		}
+
+		if len(l.msgChannel) == 0 {
+			l.flushBuffer(buffer)
+			return
+		}
+	}
+}
+
+// worker is the main goroutine that processes log messages
 func (l *Logger) worker() {
 	defer l.wg.Done()
 
 	ticker := time.NewTicker(defaultFlushInterval)
 	defer ticker.Stop()
 
-	buf := make([]byte, 0, maxBatchSize)
+	buffer := make([]byte, 0, maxBatchSize)
 
 	for {
 		select {
 		case <-l.shutdown:
+			l.handleShutdown(&buffer)
 			return
 		case <-ticker.C:
-			// Read accumulated messages
-			n, _ := l.buffer.Read(buf[:cap(buf)])
-			if n == 0 {
-				continue
-			}
-
-			// Write batch
-			_, _ = l.output.Write(buf[:n])
+			l.flushBuffer(&buffer)
+		case msg := <-l.msgChannel:
+			l.processMessage(msg, &buffer)
+			l.processBatch(&buffer)
 		}
 	}
 }

client/firewall/uspfilter/log/log_test.go

@@ -0,0 +1,121 @@
+package log_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/firewall/uspfilter/log"
+)
+
+type discard struct{}
+
+func (d *discard) Write(p []byte) (n int, err error) {
+	return len(p), nil
+}
+
+func BenchmarkLogger(b *testing.B) {
+	simpleMessage := "Connection established"
+
+	conntrackMessage := "TCP connection %s:%d -> %s:%d state changed to %d"
+	srcIP := "192.168.1.1"
+	srcPort := uint16(12345)
+	dstIP := "10.0.0.1"
+	dstPort := uint16(443)
+	state := 4 // TCPStateEstablished
+
+	complexMessage := "Packet inspection result: protocol=%s, direction=%s, flags=0x%x, sequence=%d, acknowledged=%d, payload_size=%d, fragmented=%v, connection_id=%s"
+	protocol := "TCP"
+	direction := "outbound"
+	flags := uint16(0x18) // ACK + PSH
+	sequence := uint32(123456789)
+	acknowledged := uint32(987654321)
+	payloadSize := 1460
+	fragmented := false
+	connID := "f7a12b3e-c456-7890-d123-456789abcdef"
+
+	b.Run("SimpleMessage", func(b *testing.B) {
+		logger := createTestLogger()
+		defer cleanupLogger(logger)
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			logger.Trace(simpleMessage)
+		}
+	})
+
+	b.Run("ConntrackMessage", func(b *testing.B) {
+		logger := createTestLogger()
+		defer cleanupLogger(logger)
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			logger.Trace(conntrackMessage, srcIP, srcPort, dstIP, dstPort, state)
+		}
+	})
+
+	b.Run("ComplexMessage", func(b *testing.B) {
+		logger := createTestLogger()
+		defer cleanupLogger(logger)
+
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			logger.Trace(complexMessage, protocol, direction, flags, sequence, acknowledged, payloadSize, fragmented, connID)
+		}
+	})
+}
+
+// BenchmarkLoggerParallel tests the logger under concurrent load
+func BenchmarkLoggerParallel(b *testing.B) {
+	logger := createTestLogger()
+	defer cleanupLogger(logger)
+
+	conntrackMessage := "TCP connection %s:%d -> %s:%d state changed to %d"
+	srcIP := "192.168.1.1"
+	srcPort := uint16(12345)
+	dstIP := "10.0.0.1"
+	dstPort := uint16(443)
+	state := 4
+
+	b.ResetTimer()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			logger.Trace(conntrackMessage, srcIP, srcPort, dstIP, dstPort, state)
+		}
+	})
+}
+
+// BenchmarkLoggerBurst tests how the logger handles bursts of messages
+func BenchmarkLoggerBurst(b *testing.B) {
+	logger := createTestLogger()
+	defer cleanupLogger(logger)
+
+	conntrackMessage := "TCP connection %s:%d -> %s:%d state changed to %d"
+	srcIP := "192.168.1.1"
+	srcPort := uint16(12345)
+	dstIP := "10.0.0.1"
+	dstPort := uint16(443)
+	state := 4
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < 100; j++ {
+			logger.Trace(conntrackMessage, srcIP, srcPort, dstIP, dstPort, state)
+		}
+	}
+}
+
+func createTestLogger() *log.Logger {
+	logrusLogger := logrus.New()
+	logrusLogger.SetOutput(&discard{})
+	logrusLogger.SetLevel(logrus.TraceLevel)
+	return log.NewFromLogrus(logrusLogger)
+}
+
+func cleanupLogger(logger *log.Logger) {
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	_ = logger.Stop(ctx)
+}

client/firewall/uspfilter/log/ringbuffer.go

@@ -1,85 +0,0 @@
-package log
-
-import "sync"
-
-// ringBuffer is a simple ring buffer implementation
-type ringBuffer struct {
-	buf  []byte
-	size int
-	r, w int64 // Read and write positions
-	mu   sync.Mutex
-}
-
-func newRingBuffer(size int) *ringBuffer {
-	return &ringBuffer{
-		buf:  make([]byte, size),
-		size: size,
-	}
-}
-
-func (r *ringBuffer) Write(p []byte) (n int, err error) {
-	if len(p) == 0 {
-		return 0, nil
-	}
-
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	if len(p) > r.size {
-		p = p[:r.size]
-	}
-
-	n = len(p)
-
-	// Write data, handling wrap-around
-	pos := int(r.w % int64(r.size))
-	writeLen := min(len(p), r.size-pos)
-	copy(r.buf[pos:], p[:writeLen])
-
-	// If we have more data and need to wrap around
-	if writeLen < len(p) {
-		copy(r.buf, p[writeLen:])
-	}
-
-	// Update write position
-	r.w += int64(n)
-
-	return n, nil
-}
-
-func (r *ringBuffer) Read(p []byte) (n int, err error) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
-
-	if r.w == r.r {
-		return 0, nil
-	}
-
-	// Calculate available data accounting for wraparound
-	available := int(r.w - r.r)
-	if available < 0 {
-		available += r.size
-	}
-	available = min(available, r.size)
-
-	// Limit read to buffer size
-	toRead := min(available, len(p))
-	if toRead == 0 {
-		return 0, nil
-	}
-
-	// Read data, handling wrap-around
-	pos := int(r.r % int64(r.size))
-	readLen := min(toRead, r.size-pos)
-	n = copy(p, r.buf[pos:pos+readLen])
-
-	// If we need more data and need to wrap around
-	if readLen < toRead {
-		n += copy(p[readLen:toRead], r.buf[:toRead-readLen])
-	}
-
-	// Update read position
-	r.r += int64(n)
-
-	return n, nil
-}

client/firewall/uspfilter/rule.go

@@ -12,6 +12,7 @@ import (
 // PeerRule to handle management of rules
 type PeerRule struct {
 	id         string
+	mgmtId     []byte
 	ip         net.IP
 	ipLayer    gopacket.LayerType
 	matchByIP  bool
@@ -19,7 +20,6 @@ type PeerRule struct {
 	sPort      *firewall.Port
 	dPort      *firewall.Port
 	drop       bool
-	comment    string
 
 	udpHook func([]byte) bool
 }
@@ -31,6 +31,7 @@ func (r *PeerRule) ID() string {
 
 type RouteRule struct {
 	id          string
+	mgmtId      []byte
 	sources     []netip.Prefix
 	destination netip.Prefix
 	proto       firewall.Protocol

client/firewall/uspfilter/tracer.go

@@ -317,12 +317,19 @@ func (m *Manager) handleLocalDelivery(trace *PacketTrace, packetData []byte, d *
 	}
 
 	trace.AddResult(StageRouting, "Packet destined for local delivery", true)
-	blocked := m.peerACLsBlock(srcIP, packetData, m.incomingRules, d)
 
-	msg := "Allowed by peer ACL rules"
-	if blocked {
-		msg = "Blocked by peer ACL rules"
+	ruleId, blocked := m.peerACLsBlock(srcIP, packetData, m.incomingRules, d)
+
+	strRuleId := "implicit"
+	if ruleId != nil {
+		strRuleId = string(ruleId)
 	}
+
+	msg := fmt.Sprintf("Allowed by peer ACL rules (%s)", strRuleId)
+	if blocked {
+		msg = fmt.Sprintf("Blocked by peer ACL rules (%s)", strRuleId)
+	}
+
 	trace.AddResult(StagePeerACL, msg, !blocked)
 
 	if m.netstack {
@@ -351,13 +358,18 @@ func (m *Manager) handleNativeRouter(trace *PacketTrace) *PacketTrace {
 }
 
 func (m *Manager) handleRouteACLs(trace *PacketTrace, d *decoder, srcIP, dstIP net.IP) *PacketTrace {
-	proto := getProtocolFromPacket(d)
+	proto, _ := getProtocolFromPacket(d)
 	srcPort, dstPort := getPortsFromPacket(d)
-	allowed := m.routeACLsPass(srcIP, dstIP, proto, srcPort, dstPort)
+	id, allowed := m.routeACLsPass(srcIP, dstIP, proto, srcPort, dstPort)
 
-	msg := "Allowed by route ACLs"
+	strId := string(id)
+	if id == nil {
+		strId = "implicit"
+	}
+
+	msg := fmt.Sprintf("Allowed by route ACLs (%s)", strId)
 	if !allowed {
-		msg = "Blocked by route ACLs"
+		msg = fmt.Sprintf("Blocked by route ACLs (%s)", strId)
 	}
 	trace.AddResult(StageRouteACL, msg, allowed)
 

client/firewall/uspfilter/uspfilter.go

@@ -22,6 +22,7 @@ import (
 	"github.com/netbirdio/netbird/client/firewall/uspfilter/forwarder"
 	nblog "github.com/netbirdio/netbird/client/firewall/uspfilter/log"
 	"github.com/netbirdio/netbird/client/iface/netstack"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
 )
 
@@ -96,6 +97,7 @@ type Manager struct {
 	tcpTracker  *conntrack.TCPTracker
 	forwarder   *forwarder.Forwarder
 	logger      *nblog.Logger
+	flowLogger  nftypes.FlowLogger
 }
 
 // decoder for packages
@@ -112,16 +114,16 @@ type decoder struct {
 }
 
 // Create userspace firewall manager constructor
-func Create(iface common.IFaceMapper, disableServerRoutes bool) (*Manager, error) {
-	return create(iface, nil, disableServerRoutes)
+func Create(iface common.IFaceMapper, disableServerRoutes bool, flowLogger nftypes.FlowLogger) (*Manager, error) {
+	return create(iface, nil, disableServerRoutes, flowLogger)
 }
 
-func CreateWithNativeFirewall(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableServerRoutes bool) (*Manager, error) {
+func CreateWithNativeFirewall(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableServerRoutes bool, flowLogger nftypes.FlowLogger) (*Manager, error) {
 	if nativeFirewall == nil {
 		return nil, errors.New("native firewall is nil")
 	}
 
-	mgr, err := create(iface, nativeFirewall, disableServerRoutes)
+	mgr, err := create(iface, nativeFirewall, disableServerRoutes, flowLogger)
 	if err != nil {
 		return nil, err
 	}
@@ -148,7 +150,7 @@ func parseCreateEnv() (bool, bool) {
 	return disableConntrack, enableLocalForwarding
 }
 
-func create(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableServerRoutes bool) (*Manager, error) {
+func create(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableServerRoutes bool, flowLogger nftypes.FlowLogger) (*Manager, error) {
 	disableConntrack, enableLocalForwarding := parseCreateEnv()
 
 	m := &Manager{
@@ -174,6 +176,7 @@ func create(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableSe
 		routingEnabled:      false,
 		stateful:            !disableConntrack,
 		logger:              nblog.NewFromLogrus(log.StandardLogger()),
+		flowLogger:          flowLogger,
 		netstack:            netstack.IsEnabled(),
 		localForwarding:     enableLocalForwarding,
 	}
@@ -185,9 +188,9 @@ func create(iface common.IFaceMapper, nativeFirewall firewall.Manager, disableSe
 	if disableConntrack {
 		log.Info("conntrack is disabled")
 	} else {
-		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger)
-		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger)
-		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger)
+		m.udpTracker = conntrack.NewUDPTracker(conntrack.DefaultUDPTimeout, m.logger, flowLogger)
+		m.icmpTracker = conntrack.NewICMPTracker(conntrack.DefaultICMPTimeout, m.logger, flowLogger)
+		m.tcpTracker = conntrack.NewTCPTracker(conntrack.DefaultTCPTimeout, m.logger, flowLogger)
 	}
 
 	// netstack needs the forwarder for local traffic
@@ -218,6 +221,7 @@ func (m *Manager) blockInvalidRouted(iface common.IFaceMapper) error {
 	log.Debugf("blocking invalid routed traffic for %s", wgPrefix)
 
 	if _, err := m.AddRouteFiltering(
+		nil,
 		[]netip.Prefix{netip.PrefixFrom(netip.IPv4Unspecified(), 0)},
 		wgPrefix,
 		firewall.ProtocolALL,
@@ -304,7 +308,7 @@ func (m *Manager) initForwarder() error {
 		return errors.New("forwarding not supported")
 	}
 
-	forwarder, err := forwarder.New(m.wgIface, m.logger, m.netstack)
+	forwarder, err := forwarder.New(m.wgIface, m.logger, m.flowLogger, m.netstack)
 	if err != nil {
 		m.routingEnabled = false
 		return fmt.Errorf("create forwarder: %w", err)
@@ -348,21 +352,21 @@ func (m *Manager) RemoveNatRule(pair firewall.RouterPair) error {
 // If comment argument is empty firewall manager should set
 // rule ID as comment for the rule
 func (m *Manager) AddPeerFiltering(
+	id []byte,
 	ip net.IP,
 	proto firewall.Protocol,
 	sPort *firewall.Port,
 	dPort *firewall.Port,
 	action firewall.Action,
 	_ string,
-	comment string,
 ) ([]firewall.Rule, error) {
 	r := PeerRule{
 		id:        uuid.New().String(),
+		mgmtId:    id,
 		ip:        ip,
 		ipLayer:   layers.LayerTypeIPv6,
 		matchByIP: true,
 		drop:      action == firewall.ActionDrop,
-		comment:   comment,
 	}
 	if ipNormalized := ip.To4(); ipNormalized != nil {
 		r.ipLayer = layers.LayerTypeIPv4
@@ -400,6 +404,7 @@ func (m *Manager) AddPeerFiltering(
 }
 
 func (m *Manager) AddRouteFiltering(
+	id []byte,
 	sources []netip.Prefix,
 	destination netip.Prefix,
 	proto firewall.Protocol,
@@ -408,7 +413,7 @@ func (m *Manager) AddRouteFiltering(
 	action firewall.Action,
 ) (firewall.Rule, error) {
 	if m.nativeRouter && m.nativeFirewall != nil {
-		return m.nativeFirewall.AddRouteFiltering(sources, destination, proto, sPort, dPort, action)
+		return m.nativeFirewall.AddRouteFiltering(id, sources, destination, proto, sPort, dPort, action)
 	}
 
 	m.mutex.Lock()
@@ -416,7 +421,9 @@ func (m *Manager) AddRouteFiltering(
 
 	ruleID := uuid.New().String()
 	rule := RouteRule{
+		// TODO: consolidate these IDs
 		id:          ruleID,
+		mgmtId:      id,
 		sources:     sources,
 		destination: destination,
 		proto:       proto,
@@ -533,14 +540,7 @@ func (m *Manager) processOutgoingHooks(packetData []byte) bool {
 
 	// Track all protocols if stateful mode is enabled
 	if m.stateful {
-		switch d.decoded[1] {
-		case layers.LayerTypeUDP:
-			m.trackUDPOutbound(d, srcIP, dstIP)
-		case layers.LayerTypeTCP:
-			m.trackTCPOutbound(d, srcIP, dstIP)
-		case layers.LayerTypeICMPv4:
-			m.trackICMPOutbound(d, srcIP, dstIP)
-		}
+		m.trackOutbound(d, srcIP, dstIP)
 	}
 
 	// Process UDP hooks even if stateful mode is disabled
@@ -562,17 +562,6 @@ func (m *Manager) extractIPs(d *decoder) (srcIP, dstIP net.IP) {
 	}
 }
 
-func (m *Manager) trackTCPOutbound(d *decoder, srcIP, dstIP net.IP) {
-	flags := getTCPFlags(&d.tcp)
-	m.tcpTracker.TrackOutbound(
-		srcIP,
-		dstIP,
-		uint16(d.tcp.SrcPort),
-		uint16(d.tcp.DstPort),
-		flags,
-	)
-}
-
 func getTCPFlags(tcp *layers.TCP) uint8 {
 	var flags uint8
 	if tcp.SYN {
@@ -596,13 +585,30 @@ func getTCPFlags(tcp *layers.TCP) uint8 {
 	return flags
 }
 
-func (m *Manager) trackUDPOutbound(d *decoder, srcIP, dstIP net.IP) {
-	m.udpTracker.TrackOutbound(
-		srcIP,
-		dstIP,
-		uint16(d.udp.SrcPort),
-		uint16(d.udp.DstPort),
-	)
+func (m *Manager) trackOutbound(d *decoder, srcIP, dstIP net.IP) {
+	transport := d.decoded[1]
+	switch transport {
+	case layers.LayerTypeUDP:
+		m.udpTracker.TrackOutbound(srcIP, dstIP, uint16(d.udp.SrcPort), uint16(d.udp.DstPort))
+	case layers.LayerTypeTCP:
+		flags := getTCPFlags(&d.tcp)
+		m.tcpTracker.TrackOutbound(srcIP, dstIP, uint16(d.tcp.SrcPort), uint16(d.tcp.DstPort), flags)
+	case layers.LayerTypeICMPv4:
+		m.icmpTracker.TrackOutbound(srcIP, dstIP, d.icmp4.Id, d.icmp4.Seq, d.icmp4.TypeCode)
+	}
+}
+
+func (m *Manager) trackInbound(d *decoder, srcIP, dstIP net.IP) {
+	transport := d.decoded[1]
+	switch transport {
+	case layers.LayerTypeUDP:
+		m.udpTracker.TrackInbound(srcIP, dstIP, uint16(d.udp.SrcPort), uint16(d.udp.DstPort))
+	case layers.LayerTypeTCP:
+		flags := getTCPFlags(&d.tcp)
+		m.tcpTracker.TrackInbound(srcIP, dstIP, uint16(d.tcp.SrcPort), uint16(d.tcp.DstPort), flags)
+	case layers.LayerTypeICMPv4:
+		m.icmpTracker.TrackInbound(srcIP, dstIP, d.icmp4.Id, d.icmp4.Seq, d.icmp4.TypeCode)
+	}
 }
 
 func (m *Manager) checkUDPHooks(d *decoder, dstIP net.IP, packetData []byte) bool {
@@ -618,17 +624,6 @@ func (m *Manager) checkUDPHooks(d *decoder, dstIP net.IP, packetData []byte) boo
 	return false
 }
 
-func (m *Manager) trackICMPOutbound(d *decoder, srcIP, dstIP net.IP) {
-	if d.icmp4.TypeCode.Type() == layers.ICMPv4TypeEchoRequest {
-		m.icmpTracker.TrackOutbound(
-			srcIP,
-			dstIP,
-			d.icmp4.Id,
-			d.icmp4.Seq,
-		)
-	}
-}
-
 // dropFilter implements filtering logic for incoming packets.
 // If it returns true, the packet should be dropped.
 func (m *Manager) dropFilter(packetData []byte) bool {
@@ -664,9 +659,27 @@ func (m *Manager) dropFilter(packetData []byte) bool {
 // handleLocalTraffic handles local traffic.
 // If it returns true, the packet should be dropped.
 func (m *Manager) handleLocalTraffic(d *decoder, srcIP, dstIP net.IP, packetData []byte) bool {
-	if m.peerACLsBlock(srcIP, packetData, m.incomingRules, d) {
-		m.logger.Trace("Dropping local packet (ACL denied): src=%s dst=%s",
-			srcIP, dstIP)
+	if ruleId, blocked := m.peerACLsBlock(srcIP, packetData, m.incomingRules, d); blocked {
+		srcAddr, _ := netip.AddrFromSlice(srcIP)
+		dstAddr, _ := netip.AddrFromSlice(dstIP)
+		_, pnum := getProtocolFromPacket(d)
+		srcPort, dstPort := getPortsFromPacket(d)
+
+		m.logger.Trace("Dropping local packet (ACL denied): rule_id=%s proto=%v src=%s:%d dst=%s:%d",
+			ruleId, pnum, srcAddr, srcPort, dstAddr, dstPort)
+
+		m.flowLogger.StoreEvent(nftypes.EventFields{
+			FlowID:     uuid.New(),
+			Type:       nftypes.TypeDrop,
+			RuleID:     ruleId,
+			Direction:  nftypes.Ingress,
+			Protocol:   pnum,
+			SourceIP:   srcAddr,
+			DestIP:     dstAddr,
+			SourcePort: srcPort,
+			DestPort:   dstPort,
+			// TODO: icmp type/code
+		})
 		return true
 	}
 
@@ -675,6 +688,9 @@ func (m *Manager) handleLocalTraffic(d *decoder, srcIP, dstIP net.IP, packetData
 		return m.handleNetstackLocalTraffic(packetData)
 	}
 
+	// track inbound packets to get the correct direction and session id for flows
+	m.trackInbound(d, srcIP, dstIP)
+
 	return false
 }
 
@@ -712,12 +728,28 @@ func (m *Manager) handleRoutedTraffic(d *decoder, srcIP, dstIP net.IP, packetDat
 		return false
 	}
 
-	proto := getProtocolFromPacket(d)
+	proto, pnum := getProtocolFromPacket(d)
 	srcPort, dstPort := getPortsFromPacket(d)
 
-	if !m.routeACLsPass(srcIP, dstIP, proto, srcPort, dstPort) {
-		m.logger.Trace("Dropping routed packet (ACL denied): src=%s:%d dst=%s:%d proto=%v",
-			srcIP, srcPort, dstIP, dstPort, proto)
+	if id, pass := m.routeACLsPass(srcIP, dstIP, proto, srcPort, dstPort); !pass {
+		srcAddr, _ := netip.AddrFromSlice(srcIP)
+		dstAddr, _ := netip.AddrFromSlice(dstIP)
+
+		m.logger.Trace("Dropping routed packet (ACL denied): rule_id=%s proto=%v src=%s:%d dst=%s:%d",
+			id, pnum, srcIP, srcPort, dstIP, dstPort)
+
+		m.flowLogger.StoreEvent(nftypes.EventFields{
+			FlowID:     uuid.New(),
+			Type:       nftypes.TypeDrop,
+			RuleID:     id,
+			Direction:  nftypes.Ingress,
+			Protocol:   pnum,
+			SourceIP:   srcAddr,
+			DestIP:     dstAddr,
+			SourcePort: srcPort,
+			DestPort:   dstPort,
+			// TODO: icmp type/code
+		})
 		return true
 	}
 
@@ -730,16 +762,16 @@ func (m *Manager) handleRoutedTraffic(d *decoder, srcIP, dstIP net.IP, packetDat
 	return true
 }
 
-func getProtocolFromPacket(d *decoder) firewall.Protocol {
+func getProtocolFromPacket(d *decoder) (firewall.Protocol, nftypes.Protocol) {
 	switch d.decoded[1] {
 	case layers.LayerTypeTCP:
-		return firewall.ProtocolTCP
+		return firewall.ProtocolTCP, nftypes.TCP
 	case layers.LayerTypeUDP:
-		return firewall.ProtocolUDP
+		return firewall.ProtocolUDP, nftypes.UDP
 	case layers.LayerTypeICMPv4, layers.LayerTypeICMPv6:
-		return firewall.ProtocolICMP
+		return firewall.ProtocolICMP, nftypes.ICMP
 	default:
-		return firewall.ProtocolALL
+		return firewall.ProtocolALL, nftypes.ProtocolUnknown
 	}
 }
 
@@ -812,25 +844,25 @@ func (m *Manager) isSpecialICMP(d *decoder) bool {
 		icmpType == layers.ICMPv4TypeTimeExceeded
 }
 
-func (m *Manager) peerACLsBlock(srcIP net.IP, packetData []byte, rules map[string]RuleSet, d *decoder) bool {
+func (m *Manager) peerACLsBlock(srcIP net.IP, packetData []byte, rules map[string]RuleSet, d *decoder) ([]byte, bool) {
 	if m.isSpecialICMP(d) {
-		return false
+		return nil, false
 	}
 
-	if filter, ok := validateRule(srcIP, packetData, rules[srcIP.String()], d); ok {
-		return filter
+	if mgmtId, filter, ok := validateRule(srcIP, packetData, rules[srcIP.String()], d); ok {
+		return mgmtId, filter
 	}
 
-	if filter, ok := validateRule(srcIP, packetData, rules["0.0.0.0"], d); ok {
-		return filter
+	if mgmtId, filter, ok := validateRule(srcIP, packetData, rules["0.0.0.0"], d); ok {
+		return mgmtId, filter
 	}
 
-	if filter, ok := validateRule(srcIP, packetData, rules["::"], d); ok {
-		return filter
+	if mgmtId, filter, ok := validateRule(srcIP, packetData, rules["::"], d); ok {
+		return mgmtId, filter
 	}
 
 	// Default policy: DROP ALL
-	return true
+	return nil, true
 }
 
 func portsMatch(rulePort *firewall.Port, packetPort uint16) bool {
@@ -850,7 +882,7 @@ func portsMatch(rulePort *firewall.Port, packetPort uint16) bool {
 	return false
 }
 
-func validateRule(ip net.IP, packetData []byte, rules map[string]PeerRule, d *decoder) (bool, bool) {
+func validateRule(ip net.IP, packetData []byte, rules map[string]PeerRule, d *decoder) ([]byte, bool, bool) {
 	payloadLayer := d.decoded[1]
 	for _, rule := range rules {
 		if rule.matchByIP && !ip.Equal(rule.ip) {
@@ -858,7 +890,7 @@ func validateRule(ip net.IP, packetData []byte, rules map[string]PeerRule, d *de
 		}
 
 		if rule.protoLayer == layerTypeAll {
-			return rule.drop, true
+			return rule.mgmtId, rule.drop, true
 		}
 
 		if payloadLayer != rule.protoLayer {
@@ -868,27 +900,27 @@ func validateRule(ip net.IP, packetData []byte, rules map[string]PeerRule, d *de
 		switch payloadLayer {
 		case layers.LayerTypeTCP:
 			if portsMatch(rule.sPort, uint16(d.tcp.SrcPort)) && portsMatch(rule.dPort, uint16(d.tcp.DstPort)) {
-				return rule.drop, true
+				return rule.mgmtId, rule.drop, true
 			}
 		case layers.LayerTypeUDP:
 			// if rule has UDP hook (and if we are here we match this rule)
 			// we ignore rule.drop and call this hook
 			if rule.udpHook != nil {
-				return rule.udpHook(packetData), true
+				return rule.mgmtId, rule.udpHook(packetData), true
 			}
 
 			if portsMatch(rule.sPort, uint16(d.udp.SrcPort)) && portsMatch(rule.dPort, uint16(d.udp.DstPort)) {
-				return rule.drop, true
+				return rule.mgmtId, rule.drop, true
 			}
 		case layers.LayerTypeICMPv4, layers.LayerTypeICMPv6:
-			return rule.drop, true
+			return rule.mgmtId, rule.drop, true
 		}
 	}
-	return false, false
+	return nil, false, false
 }
 
 // routeACLsPass returns treu if the packet is allowed by the route ACLs
-func (m *Manager) routeACLsPass(srcIP, dstIP net.IP, proto firewall.Protocol, srcPort, dstPort uint16) bool {
+func (m *Manager) routeACLsPass(srcIP, dstIP net.IP, proto firewall.Protocol, srcPort, dstPort uint16) ([]byte, bool) {
 	m.mutex.RLock()
 	defer m.mutex.RUnlock()
 
@@ -897,10 +929,10 @@ func (m *Manager) routeACLsPass(srcIP, dstIP net.IP, proto firewall.Protocol, sr
 
 	for _, rule := range m.routeRules {
 		if m.ruleMatches(rule, srcAddr, dstAddr, proto, srcPort, dstPort) {
-			return rule.action == firewall.ActionAccept
+			return rule.mgmtId, rule.action == firewall.ActionAccept
 		}
 	}
-	return false
+	return nil, false
 }
 
 func (m *Manager) ruleMatches(rule RouteRule, srcAddr, dstAddr netip.Addr, proto firewall.Protocol, srcPort, dstPort uint16) bool {
@@ -949,7 +981,6 @@ func (m *Manager) AddUDPPacketHook(
 		protoLayer: layers.LayerTypeUDP,
 		dPort:      &firewall.Port{Values: []uint16{dPort}},
 		ipLayer:    layers.LayerTypeIPv6,
-		comment:    fmt.Sprintf("UDP Hook direction: %v, ip:%v, dport:%d", in, ip, dPort),
 		udpHook:    hook,
 	}
 

client/firewall/uspfilter/uspfilter_bench_test.go

@@ -93,8 +93,7 @@ func BenchmarkCoreFiltering(b *testing.B) {
 			stateful: false,
 			setupFunc: func(m *Manager) {
 				// Single rule allowing all traffic
-				_, err := m.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolALL, nil, nil,
-					fw.ActionAccept, "", "allow all")
+				_, err := m.AddPeerFiltering(nil, net.ParseIP("0.0.0.0"), fw.ProtocolALL, nil, nil, fw.ActionAccept, "")
 				require.NoError(b, err)
 			},
 			desc: "Baseline: Single 'allow all' rule without connection tracking",
@@ -114,10 +113,15 @@ func BenchmarkCoreFiltering(b *testing.B) {
 				// Add explicit rules matching return traffic pattern
 				for i := 0; i < 1000; i++ { // Simulate realistic ruleset size
 					ip := generateRandomIPs(1)[0]
-					_, err := m.AddPeerFiltering(ip, fw.ProtocolTCP,
+					_, err := m.AddPeerFiltering(
+						nil,
+						ip,
+						fw.ProtocolTCP,
 						&fw.Port{Values: []uint16{uint16(1024 + i)}},
 						&fw.Port{Values: []uint16{80}},
-						fw.ActionAccept, "", "explicit return")
+						fw.ActionAccept,
+						"",
+					)
 					require.NoError(b, err)
 				}
 			},
@@ -128,8 +132,15 @@ func BenchmarkCoreFiltering(b *testing.B) {
 			stateful: true,
 			setupFunc: func(m *Manager) {
 				// Add some basic rules but rely on state for established connections
-				_, err := m.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolTCP, nil, nil,
-					fw.ActionDrop, "", "default drop")
+				_, err := m.AddPeerFiltering(
+					nil,
+					net.ParseIP("0.0.0.0"),
+					fw.ProtocolTCP,
+					nil,
+					nil,
+					fw.ActionDrop,
+					"",
+				)
 				require.NoError(b, err)
 			},
 			desc: "Connection tracking with established connections",
@@ -158,9 +169,9 @@ func BenchmarkCoreFiltering(b *testing.B) {
 				// Create manager and basic setup
 				manager, _ := Create(&IFaceMock{
 					SetFilterFunc: func(device.PacketFilter) error { return nil },
-				}, false)
+				}, false, flowLogger)
 				defer b.Cleanup(func() {
-					require.NoError(b, manager.Reset(nil))
+					require.NoError(b, manager.Close(nil))
 				})
 
 				manager.wgNetwork = &net.IPNet{
@@ -203,9 +214,9 @@ func BenchmarkStateScaling(b *testing.B) {
 		b.Run(fmt.Sprintf("conns_%d", count), func(b *testing.B) {
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.wgNetwork = &net.IPNet{
@@ -251,9 +262,9 @@ func BenchmarkEstablishmentOverhead(b *testing.B) {
 		b.Run(sc.name, func(b *testing.B) {
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.wgNetwork = &net.IPNet{
@@ -450,9 +461,9 @@ func BenchmarkRoutedNetworkReturn(b *testing.B) {
 		b.Run(sc.name, func(b *testing.B) {
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			// Setup scenario
@@ -577,9 +588,9 @@ func BenchmarkLongLivedConnections(b *testing.B) {
 
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			defer b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.SetNetwork(&net.IPNet{
@@ -590,10 +601,7 @@ func BenchmarkLongLivedConnections(b *testing.B) {
 			// Setup initial state based on scenario
 			if sc.rules {
 				// Single rule to allow all return traffic from port 80
-				_, err := manager.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolTCP,
-					&fw.Port{Values: []uint16{80}},
-					nil,
-					fw.ActionAccept, "", "return traffic")
+				_, err := manager.AddPeerFiltering(nil, net.ParseIP("0.0.0.0"), fw.ProtocolTCP, &fw.Port{Values: []uint16{80}}, nil, fw.ActionAccept, "")
 				require.NoError(b, err)
 			}
 
@@ -668,9 +676,9 @@ func BenchmarkShortLivedConnections(b *testing.B) {
 
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			defer b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.SetNetwork(&net.IPNet{
@@ -681,10 +689,7 @@ func BenchmarkShortLivedConnections(b *testing.B) {
 			// Setup initial state based on scenario
 			if sc.rules {
 				// Single rule to allow all return traffic from port 80
-				_, err := manager.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolTCP,
-					&fw.Port{Values: []uint16{80}},
-					nil,
-					fw.ActionAccept, "", "return traffic")
+				_, err := manager.AddPeerFiltering(nil, net.ParseIP("0.0.0.0"), fw.ProtocolTCP, &fw.Port{Values: []uint16{80}}, nil, fw.ActionAccept, "")
 				require.NoError(b, err)
 			}
 
@@ -787,9 +792,9 @@ func BenchmarkParallelLongLivedConnections(b *testing.B) {
 
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			defer b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.SetNetwork(&net.IPNet{
@@ -799,10 +804,7 @@ func BenchmarkParallelLongLivedConnections(b *testing.B) {
 
 			// Setup initial state based on scenario
 			if sc.rules {
-				_, err := manager.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolTCP,
-					&fw.Port{Values: []uint16{80}},
-					nil,
-					fw.ActionAccept, "", "return traffic")
+				_, err := manager.AddPeerFiltering(nil, net.ParseIP("0.0.0.0"), fw.ProtocolTCP, &fw.Port{Values: []uint16{80}}, nil, fw.ActionAccept, "")
 				require.NoError(b, err)
 			}
 
@@ -875,9 +877,9 @@ func BenchmarkParallelShortLivedConnections(b *testing.B) {
 
 			manager, _ := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			defer b.Cleanup(func() {
-				require.NoError(b, manager.Reset(nil))
+				require.NoError(b, manager.Close(nil))
 			})
 
 			manager.SetNetwork(&net.IPNet{
@@ -886,10 +888,7 @@ func BenchmarkParallelShortLivedConnections(b *testing.B) {
 			})
 
 			if sc.rules {
-				_, err := manager.AddPeerFiltering(net.ParseIP("0.0.0.0"), fw.ProtocolTCP,
-					&fw.Port{Values: []uint16{80}},
-					nil,
-					fw.ActionAccept, "", "return traffic")
+				_, err := manager.AddPeerFiltering(nil, net.ParseIP("0.0.0.0"), fw.ProtocolTCP, &fw.Port{Values: []uint16{80}}, nil, fw.ActionAccept, "")
 				require.NoError(b, err)
 			}
 
@@ -1033,14 +1032,7 @@ func BenchmarkRouteACLs(b *testing.B) {
 	}
 
 	for _, r := range rules {
-		_, err := manager.AddRouteFiltering(
-			r.sources,
-			r.dest,
-			r.proto,
-			nil,
-			r.port,
-			fw.ActionAccept,
-		)
+		_, err := manager.AddRouteFiltering(nil, r.sources, r.dest, r.proto, nil, r.port, fw.ActionAccept)
 		if err != nil {
 			b.Fatal(err)
 		}

client/firewall/uspfilter/uspfilter_filter_test.go

@@ -34,12 +34,12 @@ func TestPeerACLFiltering(t *testing.T) {
 		},
 	}
 
-	manager, err := Create(ifaceMock, false)
+	manager, err := Create(ifaceMock, false, flowLogger)
 	require.NoError(t, err)
 	require.NotNil(t, manager)
 
 	t.Cleanup(func() {
-		require.NoError(t, manager.Reset(nil))
+		require.NoError(t, manager.Close(nil))
 	})
 
 	manager.wgNetwork = wgNet
@@ -199,13 +199,13 @@ func TestPeerACLFiltering(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			rules, err := manager.AddPeerFiltering(
+				nil,
 				net.ParseIP(tc.ruleIP),
 				tc.ruleProto,
 				tc.ruleSrcPort,
 				tc.ruleDstPort,
 				tc.ruleAction,
 				"",
-				tc.name,
 			)
 			require.NoError(t, err)
 			require.NotEmpty(t, rules)
@@ -302,7 +302,7 @@ func setupRoutedManager(tb testing.TB, network string) *Manager {
 		},
 	}
 
-	manager, err := Create(ifaceMock, false)
+	manager, err := Create(ifaceMock, false, flowLogger)
 	require.NoError(tb, manager.EnableRouting())
 	require.NoError(tb, err)
 	require.NotNil(tb, manager)
@@ -310,7 +310,7 @@ func setupRoutedManager(tb testing.TB, network string) *Manager {
 	require.False(tb, manager.nativeRouter)
 
 	tb.Cleanup(func() {
-		require.NoError(tb, manager.Reset(nil))
+		require.NoError(tb, manager.Close(nil))
 	})
 
 	return manager
@@ -803,6 +803,7 @@ func TestRouteACLFiltering(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			rule, err := manager.AddRouteFiltering(
+				nil,
 				tc.rule.sources,
 				tc.rule.dest,
 				tc.rule.proto,
@@ -822,7 +823,7 @@ func TestRouteACLFiltering(t *testing.T) {
 
 			// testing routeACLsPass only and not DropIncoming, as routed packets are dropped after being passed
 			// to the forwarder
-			isAllowed := manager.routeACLsPass(srcIP, dstIP, tc.proto, tc.srcPort, tc.dstPort)
+			_, isAllowed := manager.routeACLsPass(srcIP, dstIP, tc.proto, tc.srcPort, tc.dstPort)
 			require.Equal(t, tc.shouldPass, isAllowed)
 		})
 	}
@@ -985,6 +986,7 @@ func TestRouteACLOrder(t *testing.T) {
 			var rules []fw.Rule
 			for _, r := range tc.rules {
 				rule, err := manager.AddRouteFiltering(
+					nil,
 					r.sources,
 					r.dest,
 					r.proto,
@@ -1007,7 +1009,7 @@ func TestRouteACLOrder(t *testing.T) {
 				srcIP := net.ParseIP(p.srcIP)
 				dstIP := net.ParseIP(p.dstIP)
 
-				isAllowed := manager.routeACLsPass(srcIP, dstIP, p.proto, p.srcPort, p.dstPort)
+				_, isAllowed := manager.routeACLsPass(srcIP, dstIP, p.proto, p.srcPort, p.dstPort)
 				require.Equal(t, p.shouldPass, isAllowed, "packet %d failed", i)
 			}
 		})

client/firewall/uspfilter/uspfilter_test.go

@@ -1,8 +1,10 @@
 package uspfilter
 
 import (
+	"context"
 	"fmt"
 	"net"
+	"net/netip"
 	"sync"
 	"testing"
 	"time"
@@ -18,9 +20,11 @@ import (
 	"github.com/netbirdio/netbird/client/firewall/uspfilter/log"
 	"github.com/netbirdio/netbird/client/iface"
 	"github.com/netbirdio/netbird/client/iface/device"
+	"github.com/netbirdio/netbird/client/internal/netflow"
 )
 
 var logger = log.NewFromLogrus(logrus.StandardLogger())
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
 
 type IFaceMock struct {
 	SetFilterFunc   func(device.PacketFilter) error
@@ -62,7 +66,7 @@ func TestManagerCreate(t *testing.T) {
 		SetFilterFunc: func(device.PacketFilter) error { return nil },
 	}
 
-	m, err := Create(ifaceMock, false)
+	m, err := Create(ifaceMock, false, flowLogger)
 	if err != nil {
 		t.Errorf("failed to create Manager: %v", err)
 		return
@@ -82,7 +86,7 @@ func TestManagerAddPeerFiltering(t *testing.T) {
 		},
 	}
 
-	m, err := Create(ifaceMock, false)
+	m, err := Create(ifaceMock, false, flowLogger)
 	if err != nil {
 		t.Errorf("failed to create Manager: %v", err)
 		return
@@ -92,9 +96,8 @@ func TestManagerAddPeerFiltering(t *testing.T) {
 	proto := fw.ProtocolTCP
 	port := &fw.Port{Values: []uint16{80}}
 	action := fw.ActionDrop
-	comment := "Test rule"
 
-	rule, err := m.AddPeerFiltering(ip, proto, nil, port, action, "", comment)
+	rule, err := m.AddPeerFiltering(nil, ip, proto, nil, port, action, "")
 	if err != nil {
 		t.Errorf("failed to add filtering: %v", err)
 		return
@@ -116,7 +119,7 @@ func TestManagerDeleteRule(t *testing.T) {
 		SetFilterFunc: func(device.PacketFilter) error { return nil },
 	}
 
-	m, err := Create(ifaceMock, false)
+	m, err := Create(ifaceMock, false, flowLogger)
 	if err != nil {
 		t.Errorf("failed to create Manager: %v", err)
 		return
@@ -126,9 +129,8 @@ func TestManagerDeleteRule(t *testing.T) {
 	proto := fw.ProtocolTCP
 	port := &fw.Port{Values: []uint16{80}}
 	action := fw.ActionDrop
-	comment := "Test rule 2"
 
-	rule2, err := m.AddPeerFiltering(ip, proto, nil, port, action, "", comment)
+	rule2, err := m.AddPeerFiltering(nil, ip, proto, nil, port, action, "")
 	if err != nil {
 		t.Errorf("failed to add filtering: %v", err)
 		return
@@ -187,7 +189,7 @@ func TestAddUDPPacketHook(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			manager, err := Create(&IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
-			}, false)
+			}, false, flowLogger)
 			require.NoError(t, err)
 
 			manager.AddUDPPacketHook(tt.in, tt.ip, tt.dPort, tt.hook)
@@ -236,7 +238,7 @@ func TestManagerReset(t *testing.T) {
 		SetFilterFunc: func(device.PacketFilter) error { return nil },
 	}
 
-	m, err := Create(ifaceMock, false)
+	m, err := Create(ifaceMock, false, flowLogger)
 	if err != nil {
 		t.Errorf("failed to create Manager: %v", err)
 		return
@@ -246,15 +248,14 @@ func TestManagerReset(t *testing.T) {
 	proto := fw.ProtocolTCP
 	port := &fw.Port{Values: []uint16{80}}
 	action := fw.ActionDrop
-	comment := "Test rule"
 
-	_, err = m.AddPeerFiltering(ip, proto, nil, port, action, "", comment)
+	_, err = m.AddPeerFiltering(nil, ip, proto, nil, port, action, "")
 	if err != nil {
 		t.Errorf("failed to add filtering: %v", err)
 		return
 	}
 
-	err = m.Reset(nil)
+	err = m.Close(nil)
 	if err != nil {
 		t.Errorf("failed to reset Manager: %v", err)
 		return
@@ -279,7 +280,7 @@ func TestNotMatchByIP(t *testing.T) {
 		},
 	}
 
-	m, err := Create(ifaceMock, false)
+	m, err := Create(ifaceMock, false, flowLogger)
 	if err != nil {
 		t.Errorf("failed to create Manager: %v", err)
 		return
@@ -292,9 +293,8 @@ func TestNotMatchByIP(t *testing.T) {
 	ip := net.ParseIP("0.0.0.0")
 	proto := fw.ProtocolUDP
 	action := fw.ActionAccept
-	comment := "Test rule"
 
-	_, err = m.AddPeerFiltering(ip, proto, nil, nil, action, "", comment)
+	_, err = m.AddPeerFiltering(nil, ip, proto, nil, nil, action, "")
 	if err != nil {
 		t.Errorf("failed to add filtering: %v", err)
 		return
@@ -333,7 +333,7 @@ func TestNotMatchByIP(t *testing.T) {
 		return
 	}
 
-	if err = m.Reset(nil); err != nil {
+	if err = m.Close(nil); err != nil {
 		t.Errorf("failed to reset Manager: %v", err)
 		return
 	}
@@ -347,12 +347,12 @@ func TestRemovePacketHook(t *testing.T) {
 	}
 
 	// creating manager instance
-	manager, err := Create(iface, false)
+	manager, err := Create(iface, false, flowLogger)
 	if err != nil {
 		t.Fatalf("Failed to create Manager: %s", err)
 	}
 	defer func() {
-		require.NoError(t, manager.Reset(nil))
+		require.NoError(t, manager.Close(nil))
 	}()
 
 	// Add a UDP packet hook
@@ -393,7 +393,7 @@ func TestRemovePacketHook(t *testing.T) {
 func TestProcessOutgoingHooks(t *testing.T) {
 	manager, err := Create(&IFaceMock{
 		SetFilterFunc: func(device.PacketFilter) error { return nil },
-	}, false)
+	}, false, flowLogger)
 	require.NoError(t, err)
 
 	manager.wgNetwork = &net.IPNet{
@@ -401,9 +401,9 @@ func TestProcessOutgoingHooks(t *testing.T) {
 		Mask: net.CIDRMask(16, 32),
 	}
 	manager.udpTracker.Close()
-	manager.udpTracker = conntrack.NewUDPTracker(100*time.Millisecond, logger)
+	manager.udpTracker = conntrack.NewUDPTracker(100*time.Millisecond, logger, flowLogger)
 	defer func() {
-		require.NoError(t, manager.Reset(nil))
+		require.NoError(t, manager.Close(nil))
 	}()
 
 	manager.decoders = sync.Pool{
@@ -479,12 +479,12 @@ func TestUSPFilterCreatePerformance(t *testing.T) {
 			ifaceMock := &IFaceMock{
 				SetFilterFunc: func(device.PacketFilter) error { return nil },
 			}
-			manager, err := Create(ifaceMock, false)
+			manager, err := Create(ifaceMock, false, flowLogger)
 			require.NoError(t, err)
 			time.Sleep(time.Second)
 
 			defer func() {
-				if err := manager.Reset(nil); err != nil {
+				if err := manager.Close(nil); err != nil {
 					t.Errorf("clear the manager state: %v", err)
 				}
 				time.Sleep(time.Second)
@@ -494,7 +494,7 @@ func TestUSPFilterCreatePerformance(t *testing.T) {
 			start := time.Now()
 			for i := 0; i < testMax; i++ {
 				port := &fw.Port{Values: []uint16{uint16(1000 + i)}}
-				_, err = manager.AddPeerFiltering(ip, "tcp", nil, port, fw.ActionAccept, "", "accept HTTP traffic")
+				_, err = manager.AddPeerFiltering(nil, ip, "tcp", nil, port, fw.ActionAccept, "")
 
 				require.NoError(t, err, "failed to add rule")
 			}
@@ -506,7 +506,7 @@ func TestUSPFilterCreatePerformance(t *testing.T) {
 func TestStatefulFirewall_UDPTracking(t *testing.T) {
 	manager, err := Create(&IFaceMock{
 		SetFilterFunc: func(device.PacketFilter) error { return nil },
-	}, false)
+	}, false, flowLogger)
 	require.NoError(t, err)
 
 	manager.wgNetwork = &net.IPNet{
@@ -515,7 +515,7 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 	}
 
 	manager.udpTracker.Close() // Close the existing tracker
-	manager.udpTracker = conntrack.NewUDPTracker(200*time.Millisecond, logger)
+	manager.udpTracker = conntrack.NewUDPTracker(200*time.Millisecond, logger, flowLogger)
 	manager.decoders = sync.Pool{
 		New: func() any {
 			d := &decoder{
@@ -530,12 +530,12 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 		},
 	}
 	defer func() {
-		require.NoError(t, manager.Reset(nil))
+		require.NoError(t, manager.Close(nil))
 	}()
 
 	// Set up packet parameters
-	srcIP := net.ParseIP("100.10.0.1")
-	dstIP := net.ParseIP("100.10.0.100")
+	srcIP := netip.MustParseAddr("100.10.0.1")
+	dstIP := netip.MustParseAddr("100.10.0.100")
 	srcPort := uint16(51334)
 	dstPort := uint16(53)
 
@@ -543,8 +543,8 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 	outboundIPv4 := &layers.IPv4{
 		TTL:      64,
 		Version:  4,
-		SrcIP:    srcIP,
-		DstIP:    dstIP,
+		SrcIP:    srcIP.AsSlice(),
+		DstIP:    dstIP.AsSlice(),
 		Protocol: layers.IPProtocolUDP,
 	}
 	outboundUDP := &layers.UDP{
@@ -573,11 +573,11 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 	require.False(t, drop, "Initial outbound packet should not be dropped")
 
 	// Verify connection was tracked
-	conn, exists := manager.udpTracker.GetConnection(srcIP, srcPort, dstIP, dstPort)
+	conn, exists := manager.udpTracker.GetConnection(srcIP.AsSlice(), srcPort, dstIP.AsSlice(), dstPort)
 
 	require.True(t, exists, "Connection should be tracked after outbound packet")
-	require.True(t, conntrack.ValidateIPs(conntrack.MakeIPAddr(srcIP), conn.SourceIP), "Source IP should match")
-	require.True(t, conntrack.ValidateIPs(conntrack.MakeIPAddr(dstIP), conn.DestIP), "Destination IP should match")
+	require.True(t, srcIP.Compare(conn.SourceIP) == 0, "Source IP should match")
+	require.True(t, dstIP.Compare(conn.DestIP) == 0, "Destination IP should match")
 	require.Equal(t, srcPort, conn.SourcePort, "Source port should match")
 	require.Equal(t, dstPort, conn.DestPort, "Destination port should match")
 
@@ -585,8 +585,8 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 	inboundIPv4 := &layers.IPv4{
 		TTL:      64,
 		Version:  4,
-		SrcIP:    dstIP, // Original destination is now source
-		DstIP:    srcIP, // Original source is now destination
+		SrcIP:    dstIP.AsSlice(), // Original destination is now source
+		DstIP:    srcIP.AsSlice(), // Original source is now destination
 		Protocol: layers.IPProtocolUDP,
 	}
 	inboundUDP := &layers.UDP{
@@ -641,7 +641,7 @@ func TestStatefulFirewall_UDPTracking(t *testing.T) {
 
 		// If the connection should still be valid, verify it exists
 		if cp.shouldAllow {
-			conn, exists := manager.udpTracker.GetConnection(srcIP, srcPort, dstIP, dstPort)
+			conn, exists := manager.udpTracker.GetConnection(srcIP.AsSlice(), srcPort, dstIP.AsSlice(), dstPort)
 			require.True(t, exists, "Connection should still exist during valid window")
 			require.True(t, time.Since(conn.GetLastSeen()) < manager.udpTracker.Timeout(),
 				"LastSeen should be updated for valid responses")

client/iface/netstack/tun.go

@@ -55,7 +55,7 @@ func (t *NetStackTun) Create() (tun.Device, *netstack.Net, error) {
 
 	skipProxy, err := strconv.ParseBool(os.Getenv(EnvSkipProxy))
 	if err != nil {
-		log.Errorf("failed to parse NB_ETSTACK_SKIP_PROXY: %s", err)
+		log.Errorf("failed to parse %s: %s", EnvSkipProxy, err)
 	}
 	if skipProxy {
 		return nsTunDev, tunNet, nil

client/installer.nsis

@@ -6,8 +6,8 @@
 !define DESCRIPTION "A WireGuard®-based mesh network that connects your devices into a single private network"
 !define INSTALLER_NAME "netbird-installer.exe"
 !define MAIN_APP_EXE "Netbird"
-!define ICON "ui\\netbird.ico"
-!define BANNER "ui\\banner.bmp"
+!define ICON "ui\\assets\\netbird.ico"
+!define BANNER "ui\\build\\banner.bmp"
 !define LICENSE_DATA "..\\LICENSE"
 
 !define INSTALL_DIR "$PROGRAMFILES64\${APP_NAME}"

client/internal/acl/manager.go

@@ -240,7 +240,7 @@ func (d *DefaultManager) applyRouteACL(rule *mgmProto.RouteFirewallRule) (id.Rul
 
 	dPorts := convertPortInfo(rule.PortInfo)
 
-	addedRule, err := d.firewall.AddRouteFiltering(sources, destination, protocol, nil, dPorts, action)
+	addedRule, err := d.firewall.AddRouteFiltering(rule.Id, sources, destination, protocol, nil, dPorts, action)
 	if err != nil {
 		return "", fmt.Errorf("add route rule: %w", err)
 	}
@@ -281,7 +281,7 @@ func (d *DefaultManager) protoRuleToFirewallRule(
 		}
 	}
 
-	ruleID := d.getPeerRuleID(ip, protocol, int(r.Direction), port, action, "")
+	ruleID := d.getPeerRuleID(ip, protocol, int(r.Direction), port, action)
 	if rulesPair, ok := d.peerRulesPairs[ruleID]; ok {
 		return ruleID, rulesPair, nil
 	}
@@ -289,11 +289,11 @@ func (d *DefaultManager) protoRuleToFirewallRule(
 	var rules []firewall.Rule
 	switch r.Direction {
 	case mgmProto.RuleDirection_IN:
-		rules, err = d.addInRules(ip, protocol, port, action, ipsetName, "")
+		rules, err = d.addInRules(r.Id, ip, protocol, port, action, ipsetName)
 	case mgmProto.RuleDirection_OUT:
 		// TODO: Remove this soon. Outbound rules are obsolete.
 		// We only maintain this for return traffic (inbound dir) which is now handled by the stateful firewall already
-		rules, err = d.addOutRules(ip, protocol, port, action, ipsetName, "")
+		rules, err = d.addOutRules(r.Id, ip, protocol, port, action, ipsetName)
 	default:
 		return "", nil, fmt.Errorf("invalid direction, skipping firewall rule")
 	}
@@ -322,14 +322,14 @@ func portInfoEmpty(portInfo *mgmProto.PortInfo) bool {
 }
 
 func (d *DefaultManager) addInRules(
+	id []byte,
 	ip net.IP,
 	protocol firewall.Protocol,
 	port *firewall.Port,
 	action firewall.Action,
 	ipsetName string,
-	comment string,
 ) ([]firewall.Rule, error) {
-	rule, err := d.firewall.AddPeerFiltering(ip, protocol, nil, port, action, ipsetName, comment)
+	rule, err := d.firewall.AddPeerFiltering(id, ip, protocol, nil, port, action, ipsetName)
 	if err != nil {
 		return nil, fmt.Errorf("add firewall rule: %w", err)
 	}
@@ -338,18 +338,18 @@ func (d *DefaultManager) addInRules(
 }
 
 func (d *DefaultManager) addOutRules(
+	id []byte,
 	ip net.IP,
 	protocol firewall.Protocol,
 	port *firewall.Port,
 	action firewall.Action,
 	ipsetName string,
-	comment string,
 ) ([]firewall.Rule, error) {
 	if shouldSkipInvertedRule(protocol, port) {
 		return nil, nil
 	}
 
-	rule, err := d.firewall.AddPeerFiltering(ip, protocol, port, nil, action, ipsetName, comment)
+	rule, err := d.firewall.AddPeerFiltering(id, ip, protocol, port, nil, action, ipsetName)
 	if err != nil {
 		return nil, fmt.Errorf("add firewall rule: %w", err)
 	}
@@ -364,9 +364,8 @@ func (d *DefaultManager) getPeerRuleID(
 	direction int,
 	port *firewall.Port,
 	action firewall.Action,
-	comment string,
 ) id.RuleID {
-	idStr := ip.String() + string(proto) + strconv.Itoa(direction) + strconv.Itoa(int(action)) + comment
+	idStr := ip.String() + string(proto) + strconv.Itoa(direction) + strconv.Itoa(int(action))
 	if port != nil {
 		idStr += port.String()
 	}

client/internal/acl/manager_test.go

@@ -1,6 +1,7 @@
 package acl
 
 import (
+	"context"
 	"net"
 	"testing"
 
@@ -10,9 +11,12 @@ import (
 	"github.com/netbirdio/netbird/client/firewall/manager"
 	"github.com/netbirdio/netbird/client/iface"
 	"github.com/netbirdio/netbird/client/internal/acl/mocks"
+	"github.com/netbirdio/netbird/client/internal/netflow"
 	mgmProto "github.com/netbirdio/netbird/management/proto"
 )
 
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+
 func TestDefaultManager(t *testing.T) {
 	networkMap := &mgmProto.NetworkMap{
 		FirewallRules: []*mgmProto.FirewallRule{
@@ -52,13 +56,13 @@ func TestDefaultManager(t *testing.T) {
 	ifaceMock.EXPECT().GetWGDevice().Return(nil).AnyTimes()
 
 	// we receive one rule from the management so for testing purposes ignore it
-	fw, err := firewall.NewFirewall(ifaceMock, nil, false)
+	fw, err := firewall.NewFirewall(ifaceMock, nil, flowLogger, false)
 	if err != nil {
 		t.Errorf("create firewall: %v", err)
 		return
 	}
 	defer func(fw manager.Manager) {
-		_ = fw.Reset(nil)
+		_ = fw.Close(nil)
 	}(fw)
 	acl := NewDefaultManager(fw)
 
@@ -346,13 +350,13 @@ func TestDefaultManagerEnableSSHRules(t *testing.T) {
 	ifaceMock.EXPECT().GetWGDevice().Return(nil).AnyTimes()
 
 	// we receive one rule from the management so for testing purposes ignore it
-	fw, err := firewall.NewFirewall(ifaceMock, nil, false)
+	fw, err := firewall.NewFirewall(ifaceMock, nil, flowLogger, false)
 	if err != nil {
 		t.Errorf("create firewall: %v", err)
 		return
 	}
 	defer func(fw manager.Manager) {
-		_ = fw.Reset(nil)
+		_ = fw.Close(nil)
 	}(fw)
 	acl := NewDefaultManager(fw)
 

client/internal/dns/server_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/netbirdio/netbird/client/iface/configurer"
 	"github.com/netbirdio/netbird/client/iface/device"
 	pfmock "github.com/netbirdio/netbird/client/iface/mocks"
+	"github.com/netbirdio/netbird/client/internal/netflow"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/statemanager"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
@@ -29,6 +30,8 @@ import (
 	"github.com/netbirdio/netbird/formatter"
 )
 
+var flowLogger = netflow.NewManager(context.Background(), nil, []byte{}).GetLogger()
+
 type mocWGIface struct {
 	filter device.PacketFilter
 }
@@ -916,7 +919,7 @@ func createWgInterfaceWithBind(t *testing.T) (*iface.WGIface, error) {
 		return nil, err
 	}
 
-	pf, err := uspfilter.Create(wgIface, false)
+	pf, err := uspfilter.Create(wgIface, false, flowLogger)
 	if err != nil {
 		t.Fatalf("failed to create uspfilter: %v", err)
 		return nil, err
@@ -1015,7 +1018,7 @@ func TestHandlerChain_DomainPriorities(t *testing.T) {
 				mh.AssertExpectations(t)
 			}
 
-			// Reset mocks
+			// Close mocks
 			if mh, ok := tc.expectedHandler.(*MockHandler); ok {
 				mh.ExpectedCalls = nil
 				mh.Calls = nil

client/internal/dnsfwd/manager.go

@@ -88,7 +88,7 @@ func (h *Manager) allowDNSFirewall() error {
 		return nil
 	}
 
-	dnsRules, err := h.firewall.AddPeerFiltering(net.IP{0, 0, 0, 0}, firewall.ProtocolUDP, nil, dport, firewall.ActionAccept, "", "")
+	dnsRules, err := h.firewall.AddPeerFiltering(nil, net.IP{0, 0, 0, 0}, firewall.ProtocolUDP, nil, dport, firewall.ActionAccept, "")
 	if err != nil {
 		log.Errorf("failed to add allow DNS router rules, err: %v", err)
 		return err

client/internal/engine.go

@@ -35,7 +35,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/dnsfwd"
 	"github.com/netbirdio/netbird/client/internal/ingressgw"
 	"github.com/netbirdio/netbird/client/internal/netflow"
-	"github.com/netbirdio/netbird/client/internal/netflow/types"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
 	"github.com/netbirdio/netbird/client/internal/networkmonitor"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/peer/guard"
@@ -191,7 +191,7 @@ type Engine struct {
 	persistNetworkMap bool
 	latestNetworkMap  *mgmProto.NetworkMap
 	connSemaphore     *semaphoregroup.SemaphoreGroup
-	flowManager       types.FlowManager
+	flowManager       nftypes.FlowManager
 }
 
 // Peer is an instance of the Connection Peer
@@ -234,7 +234,6 @@ func NewEngine(
 		statusRecorder: statusRecorder,
 		checks:         checks,
 		connSemaphore:  semaphoregroup.NewSemaphoreGroup(connInitLimit),
-		flowManager:    netflow.NewManager(clientCtx),
 	}
 	if runtime.GOOS == "ios" {
 		if !fileExists(mobileDep.StateFilePath) {
@@ -303,8 +302,6 @@ func (e *Engine) Stop() error {
 		return fmt.Errorf("failed to remove all peers: %s", err)
 	}
 
-	e.flowManager.Close()
-
 	if e.cancel != nil {
 		e.cancel()
 	}
@@ -314,6 +311,12 @@ func (e *Engine) Stop() error {
 	time.Sleep(500 * time.Millisecond)
 
 	e.close()
+
+	// stop flow manager after wg interface is gone
+	if e.flowManager != nil {
+		e.flowManager.Close()
+	}
+
 	log.Infof("stopped Netbird Engine")
 
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
@@ -348,6 +351,10 @@ func (e *Engine) Start() error {
 	}
 	e.wgInterface = wgIface
 
+	// start flow manager right after interface creation
+	publicKey := e.config.WgPrivateKey.PublicKey()
+	e.flowManager = netflow.NewManager(e.ctx, e.wgInterface, publicKey[:])
+
 	if e.config.RosenpassEnabled {
 		log.Infof("rosenpass is enabled")
 		if e.config.RosenpassPermissive {
@@ -454,7 +461,7 @@ func (e *Engine) createFirewall() error {
 	}
 
 	var err error
-	e.firewall, err = firewall.NewFirewall(e.wgInterface, e.stateManager, e.config.DisableServerRoutes)
+	e.firewall, err = firewall.NewFirewall(e.wgInterface, e.stateManager, e.flowManager.GetLogger(), e.config.DisableServerRoutes)
 	if err != nil || e.firewall == nil {
 		log.Errorf("failed creating firewall manager: %s", err)
 		return nil
@@ -488,13 +495,13 @@ func (e *Engine) initFirewall() error {
 
 	// this rule is static and will be torn down on engine down by the firewall manager
 	if _, err := e.firewall.AddPeerFiltering(
+		nil,
 		net.IP{0, 0, 0, 0},
 		firewallManager.ProtocolUDP,
 		nil,
 		&port,
 		firewallManager.ActionAccept,
 		"",
-		"",
 	); err != nil {
 		log.Errorf("failed to allow rosenpass interface traffic: %v", err)
 		return nil
@@ -518,6 +525,7 @@ func (e *Engine) blockLanAccess() {
 	v4 := netip.PrefixFrom(netip.IPv4Unspecified(), 0)
 	for _, network := range toBlock {
 		if _, err := e.firewall.AddRouteFiltering(
+			nil,
 			[]netip.Prefix{v4},
 			network,
 			firewallManager.ProtocolALL,
@@ -721,12 +729,13 @@ func (e *Engine) handleFlowUpdate(config *mgmProto.FlowConfig) error {
 	return e.flowManager.Update(flowConfig)
 }
 
-func toFlowLoggerConfig(config *mgmProto.FlowConfig) (*types.FlowConfig, error) {
+func toFlowLoggerConfig(config *mgmProto.FlowConfig) (*nftypes.FlowConfig, error) {
 	if config.GetInterval() == nil {
 		return nil, errors.New("flow interval is nil")
 	}
-	return &types.FlowConfig{
+	return &nftypes.FlowConfig{
 		Enabled:        config.GetEnabled(),
+		Counters:       config.GetCounters(),
 		URL:            config.GetUrl(),
 		TokenPayload:   config.GetTokenPayload(),
 		TokenSignature: config.GetTokenSignature(),
@@ -1419,7 +1428,7 @@ func (e *Engine) close() {
 	}
 
 	if e.firewall != nil {
-		err := e.firewall.Reset(e.stateManager)
+		err := e.firewall.Close(e.stateManager)
 		if err != nil {
 			log.Warnf("failed to reset firewall: %s", err)
 		}

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

@@ -107,6 +107,10 @@ func (l *Logger) GetEvents() []*types.Event {
 	return l.Store.GetEvents()
 }
 
+func (l *Logger) DeleteEvents(ids []string) {
+	l.Store.DeleteEvents(ids)
+}
+
 func (l *Logger) Close() {
 	l.stop()
 	l.cancel()

client/internal/netflow/manager.go

@@ -2,47 +2,183 @@ package netflow
 
 import (
 	"context"
+	"fmt"
+	"runtime"
 	"sync"
+	"time"
 
+	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/types"
+	nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
+	"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     types.FlowLogger
-	flowConfig *types.FlowConfig
+	mux            sync.Mutex
+	logger         nftypes.FlowLogger
+	flowConfig     *nftypes.FlowConfig
+	conntrack      nftypes.ConnTracker
+	ctx            context.Context
+	receiverClient *client.GRPCClient
+	publicKey      []byte
 }
 
-func NewManager(ctx context.Context) *Manager {
+// NewManager creates a new netflow manager
+func NewManager(ctx context.Context, iface nftypes.IFaceMapper, publicKey []byte) *Manager {
+	flowLogger := logger.New(ctx)
+
+	var ct nftypes.ConnTracker
+	if runtime.GOOS == "linux" && iface != nil && !iface.IsUserspaceBind() {
+		ct = conntrack.New(flowLogger, iface)
+	}
+
 	return &Manager{
-		logger: logger.New(ctx),
+		logger:    flowLogger,
+		conntrack: ct,
+		ctx:       ctx,
+		publicKey: publicKey,
 	}
 }
 
-func (m *Manager) Update(update *types.FlowConfig) error {
-	m.mux.Lock()
-	defer m.mux.Unlock()
+// Update applies new flow configuration settings
+func (m *Manager) Update(update *nftypes.FlowConfig) error {
 	if update == nil {
 		return nil
 	}
-
+	m.mux.Lock()
+	defer m.mux.Unlock()
+	previous := m.flowConfig
 	m.flowConfig = update
 
 	if update.Enabled {
+		if m.conntrack != nil {
+			if err := m.conntrack.Start(update.Counters); err != nil {
+				return fmt.Errorf("start conntrack: %w", err)
+			}
+		}
+
 		m.logger.Enable()
+		if previous == nil || !previous.Enabled {
+			flowClient, err := client.NewClient(m.ctx, m.flowConfig.URL, m.flowConfig.TokenPayload, m.flowConfig.TokenSignature)
+			if err != nil {
+				return err
+			}
+			log.Infof("flow client connected to %s", m.flowConfig.URL)
+			m.receiverClient = flowClient
+			go m.receiveACKs()
+			go m.startSender()
+		}
 		return nil
 	}
 
+	if m.conntrack != nil {
+		m.conntrack.Stop()
+	}
 	m.logger.Disable()
+	if previous != nil && previous.Enabled {
+		return m.receiverClient.Close()
+	}
 
 	return nil
 }
 
+// Close cleans up all resources
 func (m *Manager) Close() {
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	if m.conntrack != nil {
+		m.conntrack.Close()
+	}
 	m.logger.Close()
 }
 
-func (m *Manager) GetLogger() types.FlowLogger {
+// 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 {
+				log.Infof("send flow event to server: %s", event.ID)
+				err := m.send(event)
+				if err != nil {
+					log.Errorf("send flow event to server: %s", err)
+				}
+			}
+		}
+	}
+}
+
+func (m *Manager) receiveACKs() {
+	if m.receiverClient == nil {
+		return
+	}
+	err := m.receiverClient.Receive(m.ctx, func(ack *proto.FlowEventAck) error {
+		log.Infof("receive flow event ack: %s", ack.EventId)
+		m.logger.DeleteEvents([]string{ack.EventId})
+		return nil
+	})
+	if err != nil {
+		log.Errorf("receive flow event ack: %s", err)
+	}
+}
+
+func (m *Manager) send(event *nftypes.Event) error {
+	if m.receiverClient == nil {
+		return nil
+	}
+	return m.receiverClient.Send(m.ctx, 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,
+		},
+	}
+	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/types/types.go

@@ -5,19 +5,58 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
+	"github.com/netbirdio/netbird/client/iface/device"
 )
 
+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"
+	default:
+		return "unknown"
+	}
+}
+
 type Type int
 
 const (
-	TypeStart = iota
+	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 (
-	Ingress = iota
+	DirectionUnknown = Direction(iota)
+	Ingress
 	Egress
 )
 
@@ -30,20 +69,26 @@ type Event struct {
 type EventFields struct {
 	FlowID     uuid.UUID
 	Type       Type
+	RuleID     []byte
 	Direction  Direction
-	Protocol   uint8
+	Protocol   Protocol
 	SourceIP   netip.Addr
 	DestIP     netip.Addr
 	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
 }
@@ -62,6 +107,8 @@ type FlowLogger interface {
 	StoreEvent(flowEvent EventFields)
 	// GetEvents returns all stored events
 	GetEvents() []*Event
+	// DeleteEvents deletes events from the store
+	DeleteEvents([]string)
 	// Close closes the logger
 	Close()
 	// Enable enables the flow logger receiver
@@ -80,3 +127,20 @@ type Store interface {
 	// 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() device.WGAddress
+}

client/netbird.wxs

@@ -71,7 +71,7 @@
 		</InstallExecuteSequence>
 
 		<!-- Icons -->
-		<Icon Id="NetbirdIcon" SourceFile=".\client\ui\netbird.ico" />
+		<Icon Id="NetbirdIcon" SourceFile=".\client\ui\assets\netbird.ico" />
 		<Property Id="ARPPRODUCTICON" Value="NetbirdIcon" />
 
 	</Package>

client/resources.rc

@@ -5,5 +5,5 @@
 #define STRINGIZE(x) #x
 #define EXPAND(x) STRINGIZE(x)
 CREATEPROCESS_MANIFEST_RESOURCE_ID RT_MANIFEST manifest.xml
-7 ICON ui/netbird.ico
+7 ICON ui/assets/netbird.ico
 wintun.dll RCDATA wintun.dll

client/ui/client_ui.go

@@ -35,7 +35,9 @@ import (
 	"github.com/netbirdio/netbird/client/proto"
 	"github.com/netbirdio/netbird/client/ui/desktop"
 	"github.com/netbirdio/netbird/client/ui/event"
+	"github.com/netbirdio/netbird/client/ui/process"
 	"github.com/netbirdio/netbird/util"
+
 	"github.com/netbirdio/netbird/version"
 )
 
@@ -44,94 +46,125 @@ const (
 	failFastTimeout    = time.Second
 )
 
+const (
+	censoredPreSharedKey = "**********"
+)
+
 func main() {
-	var daemonAddr string
-
-	defaultDaemonAddr := "unix:///var/run/netbird.sock"
-	if runtime.GOOS == "windows" {
-		defaultDaemonAddr = "tcp://127.0.0.1:41731"
-	}
-
-	flag.StringVar(
-		&daemonAddr, "daemon-addr",
-		defaultDaemonAddr,
-		"Daemon service address to serve CLI requests [unix|tcp]://[path|host:port]")
-
-	var showSettings bool
-	flag.BoolVar(&showSettings, "settings", false, "run settings windows")
-	var showRoutes bool
-	flag.BoolVar(&showRoutes, "networks", false, "run networks windows")
-	var errorMSG string
-	flag.StringVar(&errorMSG, "error-msg", "", "displays a error message window")
-
-	tmpDir := "/tmp"
-	if runtime.GOOS == "windows" {
-		tmpDir = os.TempDir()
-	}
-
-	var saveLogsInFile bool
-	flag.BoolVar(&saveLogsInFile, "use-log-file", false, fmt.Sprintf("save logs in a file: %s/netbird-ui-PID.log", tmpDir))
-
-	flag.Parse()
+	daemonAddr, showSettings, showNetworks, errorMsg, saveLogsInFile := parseFlags()
 
+	// Initialize file logging if needed.
 	if saveLogsInFile {
-		logFile := path.Join(tmpDir, fmt.Sprintf("netbird-ui-%d.log", os.Getpid()))
-		err := util.InitLog("trace", logFile)
-		if err != nil {
+		if err := initLogFile(); err != nil {
 			log.Errorf("error while initializing log: %v", err)
 			return
 		}
 	}
 
+	// Create the Fyne application.
 	a := app.NewWithID("NetBird")
 	a.SetIcon(fyne.NewStaticResource("netbird", iconDisconnected))
 
-	if errorMSG != "" {
-		showErrorMSG(errorMSG)
+	// Show error message window if needed.
+	if errorMsg != "" {
+		showErrorMessage(errorMsg)
 		return
 	}
 
-	client := newServiceClient(daemonAddr, a, showSettings, showRoutes)
+	// Create the service client (this also builds the settings or networks UI if requested).
+	client := newServiceClient(daemonAddr, a, showSettings, showNetworks)
+
+	// Watch for theme/settings changes to update the icon.
+	go watchSettingsChanges(a, client)
+
+	// Run in window mode if any UI flag was set.
+	if showSettings || showNetworks {
+		a.Run()
+		return
+	}
+
+	// Check for another running process.
+	running, err := process.IsAnotherProcessRunning()
+	if err != nil {
+		log.Errorf("error while checking process: %v", err)
+		return
+	}
+	if running {
+		log.Warn("another process is running")
+		return
+	}
+
+	client.setDefaultFonts()
+	systray.Run(client.onTrayReady, client.onTrayExit)
+}
+
+// parseFlags reads and returns all needed command-line flags.
+func parseFlags() (daemonAddr string, showSettings, showNetworks bool, errorMsg string, saveLogsInFile bool) {
+	defaultDaemonAddr := "unix:///var/run/netbird.sock"
+	if runtime.GOOS == "windows" {
+		defaultDaemonAddr = "tcp://127.0.0.1:41731"
+	}
+	flag.StringVar(&daemonAddr, "daemon-addr", defaultDaemonAddr, "Daemon service address to serve CLI requests [unix|tcp]://[path|host:port]")
+	flag.BoolVar(&showSettings, "settings", false, "run settings window")
+	flag.BoolVar(&showNetworks, "networks", false, "run networks window")
+	flag.StringVar(&errorMsg, "error-msg", "", "displays an error message window")
+
+	tmpDir := "/tmp"
+	if runtime.GOOS == "windows" {
+		tmpDir = os.TempDir()
+	}
+	flag.BoolVar(&saveLogsInFile, "use-log-file", false, fmt.Sprintf("save logs in a file: %s/netbird-ui-PID.log", tmpDir))
+	flag.Parse()
+	return
+}
+
+// initLogFile initializes logging into a file.
+func initLogFile() error {
+	tmpDir := "/tmp"
+	if runtime.GOOS == "windows" {
+		tmpDir = os.TempDir()
+	}
+	logFile := path.Join(tmpDir, fmt.Sprintf("netbird-ui-%d.log", os.Getpid()))
+	return util.InitLog("trace", logFile)
+}
+
+// watchSettingsChanges listens for Fyne theme/settings changes and updates the client icon.
+func watchSettingsChanges(a fyne.App, client *serviceClient) {
 	settingsChangeChan := make(chan fyne.Settings)
 	a.Settings().AddChangeListener(settingsChangeChan)
-	go func() {
-		for range settingsChangeChan {
-			client.updateIcon()
-		}
-	}()
-
-	if showSettings || showRoutes {
-		a.Run()
-	} else {
-		running, err := isAnotherProcessRunning()
-		if err != nil {
-			log.Errorf("error while checking process: %v", err)
-		}
-		if running {
-			log.Warn("another process is running")
-			return
-		}
-		client.setDefaultFonts()
-		systray.Run(client.onTrayReady, client.onTrayExit)
+	for range settingsChangeChan {
+		client.updateIcon()
 	}
 }
 
-//go:embed netbird-systemtray-connected-macos.png
+// showErrorMessage displays an error message in a simple window.
+func showErrorMessage(msg string) {
+	a := app.New()
+	w := a.NewWindow("NetBird Error")
+	label := widget.NewLabel(msg)
+	label.Wrapping = fyne.TextWrapWord
+	w.SetContent(label)
+	w.Resize(fyne.NewSize(400, 100))
+	w.Show()
+	a.Run()
+}
+
+//go:embed assets/netbird-systemtray-connected-macos.png
 var iconConnectedMacOS []byte
 
-//go:embed netbird-systemtray-disconnected-macos.png
+//go:embed assets/netbird-systemtray-disconnected-macos.png
 var iconDisconnectedMacOS []byte
 
-//go:embed netbird-systemtray-update-disconnected-macos.png
+//go:embed assets/netbird-systemtray-update-disconnected-macos.png
 var iconUpdateDisconnectedMacOS []byte
 
-//go:embed netbird-systemtray-update-connected-macos.png
+//go:embed assets/netbird-systemtray-update-connected-macos.png
 var iconUpdateConnectedMacOS []byte
 
-//go:embed netbird-systemtray-connecting-macos.png
+//go:embed assets/netbird-systemtray-connecting-macos.png
 var iconConnectingMacOS []byte
 
-//go:embed netbird-systemtray-error-macos.png
+//go:embed assets/netbird-systemtray-error-macos.png
 var iconErrorMacOS []byte
 
 type serviceClient struct {
@@ -154,6 +187,7 @@ type serviceClient struct {
 	mAdminPanel        *systray.MenuItem
 	mSettings          *systray.MenuItem
 	mAbout             *systray.MenuItem
+	mGitHub            *systray.MenuItem
 	mVersionUI         *systray.MenuItem
 	mVersionDaemon     *systray.MenuItem
 	mUpdate            *systray.MenuItem
@@ -300,18 +334,6 @@ func (s *serviceClient) showSettingsUI() {
 	s.wSettings.Show()
 }
 
-// showErrorMSG opens a fyne app window to display the supplied message
-func showErrorMSG(msg string) {
-	app := app.New()
-	w := app.NewWindow("NetBird Error")
-	content := widget.NewLabel(msg)
-	content.Wrapping = fyne.TextWrapWord
-	w.SetContent(content)
-	w.Resize(fyne.NewSize(400, 100))
-	w.Show()
-	app.Run()
-}
-
 // getSettingsForm to embed it into settings window.
 func (s *serviceClient) getSettingsForm() *widget.Form {
 	return &widget.Form{
@@ -327,7 +349,7 @@ func (s *serviceClient) getSettingsForm() *widget.Form {
 		},
 		SubmitText: "Save",
 		OnSubmit: func() {
-			if s.iPreSharedKey.Text != "" && s.iPreSharedKey.Text != "**********" {
+			if s.iPreSharedKey.Text != "" && s.iPreSharedKey.Text != censoredPreSharedKey {
 				// validate preSharedKey if it added
 				if _, err := wgtypes.ParseKey(s.iPreSharedKey.Text); err != nil {
 					dialog.ShowError(fmt.Errorf("Invalid Pre-shared Key Value"), s.wSettings)
@@ -365,7 +387,7 @@ func (s *serviceClient) getSettingsForm() *widget.Form {
 					WireguardPort:        &port,
 				}
 
-				if s.iPreSharedKey.Text != "**********" {
+				if s.iPreSharedKey.Text != censoredPreSharedKey {
 					loginRequest.OptionalPreSharedKey = &s.iPreSharedKey.Text
 				}
 
@@ -587,26 +609,29 @@ func (s *serviceClient) onTrayReady() {
 	s.mAdminPanel = systray.AddMenuItem("Admin Panel", "Netbird Admin Panel")
 	systray.AddSeparator()
 
-	s.mSettings = systray.AddMenuItem("Settings", "Settings of the application")
-	s.mAllowSSH = s.mSettings.AddSubMenuItemCheckbox("Allow SSH", "Allow SSH connections", false)
-	s.mAutoConnect = s.mSettings.AddSubMenuItemCheckbox("Connect on Startup", "Connect automatically when the service starts", false)
-	s.mEnableRosenpass = s.mSettings.AddSubMenuItemCheckbox("Enable Quantum-Resistance", "Enable post-quantum security via Rosenpass", false)
-	s.mNotifications = s.mSettings.AddSubMenuItemCheckbox("Notifications", "Enable notifications", false)
-	s.mAdvancedSettings = s.mSettings.AddSubMenuItem("Advanced Settings", "Advanced settings of the application")
-	s.mCreateDebugBundle = s.mSettings.AddSubMenuItem("Create Debug Bundle", "Create and open debug information bundle")
+	s.mSettings = systray.AddMenuItem("Settings", settingsMenuDescr)
+	s.mAllowSSH = s.mSettings.AddSubMenuItemCheckbox("Allow SSH", allowSSHMenuDescr, false)
+	s.mAutoConnect = s.mSettings.AddSubMenuItemCheckbox("Connect on Startup", autoConnectMenuDescr, false)
+	s.mEnableRosenpass = s.mSettings.AddSubMenuItemCheckbox("Enable Quantum-Resistance", quantumResistanceMenuDescr, false)
+	s.mNotifications = s.mSettings.AddSubMenuItemCheckbox("Notifications", notificationsMenuDescr, false)
+	s.mAdvancedSettings = s.mSettings.AddSubMenuItem("Advanced Settings", advancedSettingsMenuDescr)
+	s.mCreateDebugBundle = s.mSettings.AddSubMenuItem("Create Debug Bundle", debugBundleMenuDescr)
 	s.loadSettings()
 
 	s.exitNodeMu.Lock()
-	s.mExitNode = systray.AddMenuItem("Exit Node", "Select exit node for routing traffic")
+	s.mExitNode = systray.AddMenuItem("Exit Node", exitNodeMenuDescr)
 	s.mExitNode.Disable()
 	s.exitNodeMu.Unlock()
 
-	s.mNetworks = systray.AddMenuItem("Networks", "Open the networks management window")
+	s.mNetworks = systray.AddMenuItem("Networks", networksMenuDescr)
 	s.mNetworks.Disable()
 	systray.AddSeparator()
 
 	s.mAbout = systray.AddMenuItem("About", "About")
 	s.mAbout.SetIcon(s.icAbout)
+	
+	s.mGitHub = s.mAbout.AddSubMenuItem("GitHub", "GitHub")
+
 	versionString := normalizedVersion(version.NetbirdVersion())
 	s.mVersionUI = s.mAbout.AddSubMenuItem(fmt.Sprintf("GUI: %s", versionString), fmt.Sprintf("GUI Version: %s", versionString))
 	s.mVersionUI.Disable()
@@ -615,11 +640,11 @@ func (s *serviceClient) onTrayReady() {
 	s.mVersionDaemon.Disable()
 	s.mVersionDaemon.Hide()
 
-	s.mUpdate = s.mAbout.AddSubMenuItem("Download latest version", "Download latest version")
+	s.mUpdate = s.mAbout.AddSubMenuItem("Download latest version", latestVersionMenuDescr)
 	s.mUpdate.Hide()
 
 	systray.AddSeparator()
-	s.mQuit = systray.AddMenuItem("Quit", "Quit the client app")
+	s.mQuit = systray.AddMenuItem("Quit", quitMenuDescr)
 
 	// update exit node menu in case service is already connected
 	go s.updateExitNodes()
@@ -717,6 +742,11 @@ func (s *serviceClient) onTrayReady() {
 			case <-s.mQuit.ClickedCh:
 				systray.Quit()
 				return
+			case <-s.mGitHub.ClickedCh:
+				err := openURL("https://github.com/netbirdio/netbird")
+				if err != nil {
+					log.Errorf("%s", err)
+				}
 			case <-s.mUpdate.ClickedCh:
 				err := openURL(version.DownloadUrl())
 				if err != nil {

client/ui/const.go

@@ -0,0 +1,15 @@
+package main
+
+const (
+	settingsMenuDescr          = "Settings of the application"
+	allowSSHMenuDescr          = "Allow SSH connections"
+	autoConnectMenuDescr       = "Connect automatically when the service starts"
+	quantumResistanceMenuDescr = "Enable post-quantum security via Rosenpass"
+	notificationsMenuDescr     = "Enable notifications"
+	advancedSettingsMenuDescr  = "Advanced settings of the application"
+	debugBundleMenuDescr       = "Create and open debug information bundle"
+	exitNodeMenuDescr          = "Select exit node for routing traffic"
+	networksMenuDescr          = "Open the networks management window"
+	latestVersionMenuDescr     = "Download latest version"
+	quitMenuDescr              = "Quit the client app"
+)

client/ui/icons.go

@@ -6,38 +6,38 @@ import (
 	_ "embed"
 )
 
-//go:embed netbird.png
+//go:embed assets/netbird.png
 var iconAbout []byte
 
-//go:embed netbird-systemtray-connected.png
+//go:embed assets/netbird-systemtray-connected.png
 var iconConnected []byte
 
-//go:embed netbird-systemtray-connected-dark.png
+//go:embed assets/netbird-systemtray-connected-dark.png
 var iconConnectedDark []byte
 
-//go:embed netbird-systemtray-disconnected.png
+//go:embed assets/netbird-systemtray-disconnected.png
 var iconDisconnected []byte
 
-//go:embed netbird-systemtray-update-disconnected.png
+//go:embed assets/netbird-systemtray-update-disconnected.png
 var iconUpdateDisconnected []byte
 
-//go:embed netbird-systemtray-update-disconnected-dark.png
+//go:embed assets/netbird-systemtray-update-disconnected-dark.png
 var iconUpdateDisconnectedDark []byte
 
-//go:embed netbird-systemtray-update-connected.png
+//go:embed assets/netbird-systemtray-update-connected.png
 var iconUpdateConnected []byte
 
-//go:embed netbird-systemtray-update-connected-dark.png
+//go:embed assets/netbird-systemtray-update-connected-dark.png
 var iconUpdateConnectedDark []byte
 
-//go:embed netbird-systemtray-connecting.png
+//go:embed assets/netbird-systemtray-connecting.png
 var iconConnecting []byte
 
-//go:embed netbird-systemtray-connecting-dark.png
+//go:embed assets/netbird-systemtray-connecting-dark.png
 var iconConnectingDark []byte
 
-//go:embed netbird-systemtray-error.png
+//go:embed assets/netbird-systemtray-error.png
 var iconError []byte
 
-//go:embed netbird-systemtray-error-dark.png
+//go:embed assets/netbird-systemtray-error-dark.png
 var iconErrorDark []byte

client/ui/icons_windows.go

@@ -1,41 +1,41 @@
 package main
 
 import (
- _ "embed"
+	_ "embed"
 )
 
-//go:embed netbird.ico
+//go:embed assets/netbird.ico
 var iconAbout []byte
 
-//go:embed netbird-systemtray-connected.ico
+//go:embed assets/netbird-systemtray-connected.ico
 var iconConnected []byte
 
-//go:embed netbird-systemtray-connected-dark.ico
+//go:embed assets/netbird-systemtray-connected-dark.ico
 var iconConnectedDark []byte
 
-//go:embed netbird-systemtray-disconnected.ico
+//go:embed assets/netbird-systemtray-disconnected.ico
 var iconDisconnected []byte
 
-//go:embed netbird-systemtray-update-disconnected.ico
+//go:embed assets/netbird-systemtray-update-disconnected.ico
 var iconUpdateDisconnected []byte
 
-//go:embed netbird-systemtray-update-disconnected-dark.ico
+//go:embed assets/netbird-systemtray-update-disconnected-dark.ico
 var iconUpdateDisconnectedDark []byte
 
-//go:embed netbird-systemtray-update-connected.ico
+//go:embed assets/netbird-systemtray-update-connected.ico
 var iconUpdateConnected []byte
 
-//go:embed netbird-systemtray-update-connected-dark.ico
+//go:embed assets/netbird-systemtray-update-connected-dark.ico
 var iconUpdateConnectedDark []byte
 
-//go:embed netbird-systemtray-connecting.ico
+//go:embed assets/netbird-systemtray-connecting.ico
 var iconConnecting []byte
 
-//go:embed netbird-systemtray-connecting-dark.ico
+//go:embed assets/netbird-systemtray-connecting-dark.ico
 var iconConnectingDark []byte
 
-//go:embed netbird-systemtray-error.ico
+//go:embed assets/netbird-systemtray-error.ico
 var iconError []byte
 
-//go:embed netbird-systemtray-error-dark.ico
+//go:embed assets/netbird-systemtray-error-dark.ico
 var iconErrorDark []byte

client/ui/network.go

@@ -363,7 +363,7 @@ func (s *serviceClient) recreateExitNodeMenu(exitNodes []*proto.Network) {
 
 	if runtime.GOOS == "linux" || runtime.GOOS == "freebsd" {
 		s.mExitNode.Remove()
-		s.mExitNode = systray.AddMenuItem("Exit Node", "Select exit node for routing traffic")
+		s.mExitNode = systray.AddMenuItem("Exit Node", exitNodeMenuDescr)
 	}
 
 	for _, node := range exitNodes {

client/ui/process/process.go

@@ -1,4 +1,4 @@
-package main
+package process
 
 import (
 	"os"
@@ -8,7 +8,7 @@ import (
 	"github.com/shirou/gopsutil/v3/process"
 )
 
-func isAnotherProcessRunning() (bool, error) {
+func IsAnotherProcessRunning() (bool, error) {
 	processes, err := process.Processes()
 	if err != nil {
 		return false, err

client/ui/process/process_nonwindows.go

@@ -1,6 +1,6 @@
 //go:build !windows
 
-package main
+package process
 
 import (
 	"os"

client/ui/process/process_windows.go

@@ -1,4 +1,4 @@
-package main
+package process
 
 import (
 	"os/user"

flow/client/auth.go

@@ -0,0 +1,32 @@
+package client
+
+import (
+	"context"
+	"fmt"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+var _ credentials.PerRPCCredentials = (*authToken)(nil)
+
+type authToken struct {
+	metaMap map[string]string
+}
+
+func (t authToken) GetRequestMetadata(context.Context, ...string) (map[string]string, error) {
+	return t.metaMap, nil
+}
+
+func (authToken) RequireTransportSecurity() bool {
+	return false // Set to true if you want to require a secure connection
+}
+
+// WithAuthToken returns a DialOption which sets the receiver flow credentials and places auth state on each outbound RPC
+func withAuthToken(payload, signature string) grpc.DialOption {
+	value := fmt.Sprintf("%s.%s", signature, payload)
+	authMap := map[string]string{
+		"authorization": "Bearer " + value,
+	}
+	return grpc.WithPerRPCCredentials(authToken{metaMap: authMap})
+}

flow/client/client.go

@@ -0,0 +1,158 @@
+package client
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/keepalive"
+
+	"github.com/netbirdio/netbird/flow/proto"
+	"github.com/netbirdio/netbird/util/embeddedroots"
+	nbgrpc "github.com/netbirdio/netbird/util/grpc"
+)
+
+type GRPCClient struct {
+	realClient proto.FlowServiceClient
+	clientConn *grpc.ClientConn
+	stream     proto.FlowService_EventsClient
+}
+
+func NewClient(ctx context.Context, addr, payload, signature string) (*GRPCClient, error) {
+
+	transportOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+	if strings.Contains(addr, "443") {
+
+		certPool, err := x509.SystemCertPool()
+		if err != nil || certPool == nil {
+			log.Debugf("System cert pool not available; falling back to embedded cert, error: %v", err)
+			certPool = embeddedroots.Get()
+		}
+
+		transportOption = grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{
+			RootCAs: certPool,
+		}))
+	}
+
+	connCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	conn, err := grpc.DialContext(
+		connCtx,
+		addr,
+		transportOption,
+		nbgrpc.WithCustomDialer(),
+		grpc.WithBlock(),
+		grpc.WithKeepaliveParams(keepalive.ClientParameters{
+			Time:    30 * time.Second,
+			Timeout: 10 * time.Second,
+		}),
+		withAuthToken(payload, signature),
+	)
+
+	if err != nil {
+		return nil, fmt.Errorf("dialing with context: %s", err)
+	}
+
+	client := &GRPCClient{
+		realClient: proto.NewFlowServiceClient(conn),
+		clientConn: conn,
+	}
+	return client, nil
+}
+
+func (c *GRPCClient) Close() error {
+	return c.clientConn.Close()
+}
+
+func (c *GRPCClient) Receive(ctx context.Context, msgHandler func(msg *proto.FlowEventAck) error) error {
+	backOff := defaultBackoff(ctx)
+	operation := func() error {
+		connState := c.clientConn.GetState()
+		if connState == connectivity.Shutdown {
+			return backoff.Permanent(fmt.Errorf("connection to signal has been shut down"))
+		}
+
+		stream, err := c.realClient.Events(ctx, grpc.WaitForReady(true))
+		if err != nil {
+			return err
+		}
+		c.stream = stream
+
+		err = checkHeader(stream)
+		if err != nil {
+			return err
+		}
+
+		return c.receive(stream, msgHandler)
+	}
+
+	err := backoff.Retry(operation, backOff)
+	if err != nil {
+		log.Errorf("exiting the flow receiver service connection retry loop due to the unrecoverable error: %v", err)
+		return err
+	}
+
+	return nil
+}
+
+func (c *GRPCClient) receive(stream proto.FlowService_EventsClient, msgHandler func(msg *proto.FlowEventAck) error) error {
+	for {
+		msg, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+
+		if err := msgHandler(msg); err != nil {
+			return err
+		}
+	}
+}
+
+func checkHeader(stream proto.FlowService_EventsClient) error {
+	header, err := stream.Header()
+	if err != nil {
+		log.Errorf("waiting for flow receiver header: %s", err)
+		return err
+	}
+
+	if len(header) == 0 {
+		log.Error("flow receiver sent no headers")
+		return fmt.Errorf("should have headers")
+	}
+	return nil
+}
+
+func defaultBackoff(ctx context.Context) backoff.BackOff {
+	return backoff.WithContext(&backoff.ExponentialBackOff{
+		InitialInterval:     800 * time.Millisecond,
+		RandomizationFactor: 1,
+		Multiplier:          1.7,
+		MaxInterval:         10 * time.Second,
+		MaxElapsedTime:      3 * 30 * 24 * time.Hour, // 3 months
+		Stop:                backoff.Stop,
+		Clock:               backoff.SystemClock,
+	}, ctx)
+}
+
+func (c *GRPCClient) Send(ctx context.Context, event *proto.FlowEvent) error {
+	if c.stream == nil {
+		return fmt.Errorf("stream not initialized")
+	}
+
+	err := c.stream.Send(event)
+	if err != nil {
+		return fmt.Errorf("sending flow event: %s", err)
+	}
+
+	return nil
+}

flow/proto/flow.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
 // 	protoc-gen-go v1.26.0
-// 	protoc        v3.21.9
+// 	protoc        v4.24.3
 // source: flow.proto
 
 package proto
@@ -28,6 +28,7 @@ const (
 	Type_TYPE_UNKNOWN Type = 0
 	Type_TYPE_START   Type = 1
 	Type_TYPE_END     Type = 2
+	Type_TYPE_DROP    Type = 3
 )
 
 // Enum value maps for Type.
@@ -36,11 +37,13 @@ var (
 		0: "TYPE_UNKNOWN",
 		1: "TYPE_START",
 		2: "TYPE_END",
+		3: "TYPE_DROP",
 	}
 	Type_value = map[string]int32{
 		"TYPE_UNKNOWN": 0,
 		"TYPE_START":   1,
 		"TYPE_END":     2,
+		"TYPE_DROP":    3,
 	}
 )
 
@@ -128,13 +131,11 @@ type FlowEvent struct {
 
 	// Unique client event identifier
 	EventId string `protobuf:"bytes,1,opt,name=event_id,json=eventId,proto3" json:"event_id,omitempty"`
-	// Unique client flow session identifier
-	FlowId string `protobuf:"bytes,2,opt,name=flow_id,json=flowId,proto3" json:"flow_id,omitempty"`
 	// When the event occurred
-	Timestamp *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+	Timestamp *timestamppb.Timestamp `protobuf:"bytes,2,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
 	// Public key of the sending peer
-	PublicKey   []byte       `protobuf:"bytes,4,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
-	EventFields *EventFields `protobuf:"bytes,5,opt,name=event_fields,json=eventFields,proto3" json:"event_fields,omitempty"`
+	PublicKey  []byte      `protobuf:"bytes,3,opt,name=public_key,json=publicKey,proto3" json:"public_key,omitempty"`
+	FlowFields *FlowFields `protobuf:"bytes,4,opt,name=flow_fields,json=flowFields,proto3" json:"flow_fields,omitempty"`
 }
 
 func (x *FlowEvent) Reset() {
@@ -176,13 +177,6 @@ func (x *FlowEvent) GetEventId() string {
 	return ""
 }
 
-func (x *FlowEvent) GetFlowId() string {
-	if x != nil {
-		return x.FlowId
-	}
-	return ""
-}
-
 func (x *FlowEvent) GetTimestamp() *timestamppb.Timestamp {
 	if x != nil {
 		return x.Timestamp
@@ -197,9 +191,9 @@ func (x *FlowEvent) GetPublicKey() []byte {
 	return nil
 }
 
-func (x *FlowEvent) GetEventFields() *EventFields {
+func (x *FlowEvent) GetFlowFields() *FlowFields {
 	if x != nil {
-		return x.EventFields
+		return x.FlowFields
 	}
 	return nil
 }
@@ -252,32 +246,42 @@ func (x *FlowEventAck) GetEventId() string {
 	return ""
 }
 
-type EventFields struct {
+type FlowFields struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// Event type
-	Type Type `protobuf:"varint,1,opt,name=type,proto3,enum=flow.Type" json:"type,omitempty"`
+	// Unique client flow session identifier
+	FlowId []byte `protobuf:"bytes,1,opt,name=flow_id,json=flowId,proto3" json:"flow_id,omitempty"`
+	// Flow type
+	Type Type `protobuf:"varint,2,opt,name=type,proto3,enum=flow.Type" json:"type,omitempty"`
+	// RuleId identifies the rule that allowed or denied the connection
+	RuleId []byte `protobuf:"bytes,3,opt,name=rule_id,json=ruleId,proto3" json:"rule_id,omitempty"`
 	// Initiating traffic direction
-	Direction Direction `protobuf:"varint,2,opt,name=direction,proto3,enum=flow.Direction" json:"direction,omitempty"`
+	Direction Direction `protobuf:"varint,4,opt,name=direction,proto3,enum=flow.Direction" json:"direction,omitempty"`
 	// IP protocol number
-	Protocol uint32 `protobuf:"varint,3,opt,name=protocol,proto3" json:"protocol,omitempty"`
+	Protocol uint32 `protobuf:"varint,5,opt,name=protocol,proto3" json:"protocol,omitempty"`
 	// Source IP address
-	SourceIp []byte `protobuf:"bytes,4,opt,name=source_ip,json=sourceIp,proto3" json:"source_ip,omitempty"`
+	SourceIp []byte `protobuf:"bytes,6,opt,name=source_ip,json=sourceIp,proto3" json:"source_ip,omitempty"`
 	// Destination IP address
-	DestIp []byte `protobuf:"bytes,5,opt,name=dest_ip,json=destIp,proto3" json:"dest_ip,omitempty"`
+	DestIp []byte `protobuf:"bytes,7,opt,name=dest_ip,json=destIp,proto3" json:"dest_ip,omitempty"`
 	// Layer 4 -specific information
 	//
 	// Types that are assignable to ConnectionInfo:
 	//
-	//	*EventFields_PortInfo
-	//	*EventFields_IcmpInfo
-	ConnectionInfo isEventFields_ConnectionInfo `protobuf_oneof:"connection_info"`
+	//	*FlowFields_PortInfo
+	//	*FlowFields_IcmpInfo
+	ConnectionInfo isFlowFields_ConnectionInfo `protobuf_oneof:"connection_info"`
+	// Number of packets
+	RxPackets uint64 `protobuf:"varint,10,opt,name=rx_packets,json=rxPackets,proto3" json:"rx_packets,omitempty"`
+	TxPackets uint64 `protobuf:"varint,11,opt,name=tx_packets,json=txPackets,proto3" json:"tx_packets,omitempty"`
+	// Number of bytes
+	RxBytes uint64 `protobuf:"varint,12,opt,name=rx_bytes,json=rxBytes,proto3" json:"rx_bytes,omitempty"`
+	TxBytes uint64 `protobuf:"varint,13,opt,name=tx_bytes,json=txBytes,proto3" json:"tx_bytes,omitempty"`
 }
 
-func (x *EventFields) Reset() {
-	*x = EventFields{}
+func (x *FlowFields) Reset() {
+	*x = FlowFields{}
 	if protoimpl.UnsafeEnabled {
 		mi := &file_flow_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -285,13 +289,13 @@ func (x *EventFields) Reset() {
 	}
 }
 
-func (x *EventFields) String() string {
+func (x *FlowFields) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*EventFields) ProtoMessage() {}
+func (*FlowFields) ProtoMessage() {}
 
-func (x *EventFields) ProtoReflect() protoreflect.Message {
+func (x *FlowFields) ProtoReflect() protoreflect.Message {
 	mi := &file_flow_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -303,84 +307,126 @@ func (x *EventFields) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use EventFields.ProtoReflect.Descriptor instead.
-func (*EventFields) Descriptor() ([]byte, []int) {
+// Deprecated: Use FlowFields.ProtoReflect.Descriptor instead.
+func (*FlowFields) Descriptor() ([]byte, []int) {
 	return file_flow_proto_rawDescGZIP(), []int{2}
 }
 
-func (x *EventFields) GetType() Type {
+func (x *FlowFields) GetFlowId() []byte {
+	if x != nil {
+		return x.FlowId
+	}
+	return nil
+}
+
+func (x *FlowFields) GetType() Type {
 	if x != nil {
 		return x.Type
 	}
 	return Type_TYPE_UNKNOWN
 }
 
-func (x *EventFields) GetDirection() Direction {
+func (x *FlowFields) GetRuleId() []byte {
+	if x != nil {
+		return x.RuleId
+	}
+	return nil
+}
+
+func (x *FlowFields) GetDirection() Direction {
 	if x != nil {
 		return x.Direction
 	}
 	return Direction_DIRECTION_UNKNOWN
 }
 
-func (x *EventFields) GetProtocol() uint32 {
+func (x *FlowFields) GetProtocol() uint32 {
 	if x != nil {
 		return x.Protocol
 	}
 	return 0
 }
 
-func (x *EventFields) GetSourceIp() []byte {
+func (x *FlowFields) GetSourceIp() []byte {
 	if x != nil {
 		return x.SourceIp
 	}
 	return nil
 }
 
-func (x *EventFields) GetDestIp() []byte {
+func (x *FlowFields) GetDestIp() []byte {
 	if x != nil {
 		return x.DestIp
 	}
 	return nil
 }
 
-func (m *EventFields) GetConnectionInfo() isEventFields_ConnectionInfo {
+func (m *FlowFields) GetConnectionInfo() isFlowFields_ConnectionInfo {
 	if m != nil {
 		return m.ConnectionInfo
 	}
 	return nil
 }
 
-func (x *EventFields) GetPortInfo() *PortInfo {
-	if x, ok := x.GetConnectionInfo().(*EventFields_PortInfo); ok {
+func (x *FlowFields) GetPortInfo() *PortInfo {
+	if x, ok := x.GetConnectionInfo().(*FlowFields_PortInfo); ok {
 		return x.PortInfo
 	}
 	return nil
 }
 
-func (x *EventFields) GetIcmpInfo() *ICMPInfo {
-	if x, ok := x.GetConnectionInfo().(*EventFields_IcmpInfo); ok {
+func (x *FlowFields) GetIcmpInfo() *ICMPInfo {
+	if x, ok := x.GetConnectionInfo().(*FlowFields_IcmpInfo); ok {
 		return x.IcmpInfo
 	}
 	return nil
 }
 
-type isEventFields_ConnectionInfo interface {
-	isEventFields_ConnectionInfo()
+func (x *FlowFields) GetRxPackets() uint64 {
+	if x != nil {
+		return x.RxPackets
+	}
+	return 0
 }
 
-type EventFields_PortInfo struct {
+func (x *FlowFields) GetTxPackets() uint64 {
+	if x != nil {
+		return x.TxPackets
+	}
+	return 0
+}
+
+func (x *FlowFields) GetRxBytes() uint64 {
+	if x != nil {
+		return x.RxBytes
+	}
+	return 0
+}
+
+func (x *FlowFields) GetTxBytes() uint64 {
+	if x != nil {
+		return x.TxBytes
+	}
+	return 0
+}
+
+type isFlowFields_ConnectionInfo interface {
+	isFlowFields_ConnectionInfo()
+}
+
+type FlowFields_PortInfo struct {
 	// TCP/UDP port information
-	PortInfo *PortInfo `protobuf:"bytes,6,opt,name=port_info,json=portInfo,proto3,oneof"`
+	PortInfo *PortInfo `protobuf:"bytes,8,opt,name=port_info,json=portInfo,proto3,oneof"`
 }
 
-type EventFields_IcmpInfo struct {
+type FlowFields_IcmpInfo struct {
 	// ICMP type and code
-	IcmpInfo *ICMPInfo `protobuf:"bytes,7,opt,name=icmp_info,json=icmpInfo,proto3,oneof"`
+	IcmpInfo *ICMPInfo `protobuf:"bytes,9,opt,name=icmp_info,json=icmpInfo,proto3,oneof"`
 }
 
-func (*EventFields_PortInfo) isEventFields_ConnectionInfo() {}
+func (*FlowFields_PortInfo) isFlowFields_ConnectionInfo() {}
 
-func (*EventFields_IcmpInfo) isEventFields_ConnectionInfo() {}
+func (*FlowFields_IcmpInfo) isFlowFields_ConnectionInfo() {}
 
 // TCP/UDP port information
 type PortInfo struct {
@@ -500,62 +546,72 @@ var file_flow_proto_rawDesc = []byte{
 	0x0a, 0x0a, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x04, 0x66, 0x6c,
 	0x6f, 0x77, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
 	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x22, 0xce, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
+	0x6f, 0x74, 0x6f, 0x22, 0xb2, 0x01, 0x0a, 0x09, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
 	0x74, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x17, 0x0a, 0x07,
-	0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x66,
-	0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
-	0x6d, 0x70, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c,
-	0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73,
-	0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12,
-	0x1d, 0x0a, 0x0a, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x04, 0x20,
-	0x01, 0x28, 0x0c, 0x52, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x4b, 0x65, 0x79, 0x12, 0x34,
-	0x0a, 0x0c, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x18, 0x05,
-	0x20, 0x01, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0b, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x46, 0x69,
-	0x65, 0x6c, 0x64, 0x73, 0x22, 0x29, 0x0a, 0x0c, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x22,
-	0x9f, 0x02, 0x0a, 0x0b, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x12,
-	0x1e, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e,
-	0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
-	0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01,
-	0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74,
-	0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1a,
-	0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0d,
-	0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x1b, 0x0a, 0x09, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x08, 0x73,
-	0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70, 0x12, 0x17, 0x0a, 0x07, 0x64, 0x65, 0x73, 0x74, 0x5f,
-	0x69, 0x70, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x64, 0x65, 0x73, 0x74, 0x49, 0x70,
-	0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x06, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x50, 0x6f, 0x72, 0x74, 0x49,
-	0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x70, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12,
-	0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x18, 0x07, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e,
-	0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x49, 0x6e, 0x66, 0x6f, 0x42, 0x11,
-	0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x66,
-	0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1f, 0x0a,
-	0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x1b,
-	0x0a, 0x09, 0x64, 0x65, 0x73, 0x74, 0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x0d, 0x52, 0x08, 0x64, 0x65, 0x73, 0x74, 0x50, 0x6f, 0x72, 0x74, 0x22, 0x44, 0x0a, 0x08, 0x49,
-	0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f,
-	0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70,
-	0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x64,
-	0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f, 0x64,
-	0x65, 0x2a, 0x36, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x0c, 0x54, 0x59, 0x50,
-	0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x54,
-	0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x41, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x54,
-	0x59, 0x50, 0x45, 0x5f, 0x45, 0x4e, 0x44, 0x10, 0x02, 0x2a, 0x3b, 0x0a, 0x09, 0x44, 0x69, 0x72,
-	0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15, 0x0a, 0x11, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54,
-	0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a,
-	0x07, 0x49, 0x4e, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x45, 0x47,
-	0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x32, 0x42, 0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x65,
-	0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x33, 0x0a, 0x06, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x12,
-	0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x1a, 0x12, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e,
-	0x74, 0x41, 0x63, 0x6b, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x01, 0x28, 0x09, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x38, 0x0a, 0x09,
+	0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75,
+	0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x09, 0x74, 0x69, 0x6d,
+	0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63,
+	0x5f, 0x6b, 0x65, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x70, 0x75, 0x62, 0x6c,
+	0x69, 0x63, 0x4b, 0x65, 0x79, 0x12, 0x31, 0x0a, 0x0b, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x66, 0x69,
+	0x65, 0x6c, 0x64, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x66, 0x6c, 0x6f,
+	0x77, 0x2e, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x52, 0x0a, 0x66, 0x6c,
+	0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x73, 0x22, 0x29, 0x0a, 0x0c, 0x46, 0x6c, 0x6f, 0x77,
+	0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x12, 0x19, 0x0a, 0x08, 0x65, 0x76, 0x65, 0x6e,
+	0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x65, 0x76, 0x65, 0x6e,
+	0x74, 0x49, 0x64, 0x22, 0xc4, 0x03, 0x0a, 0x0a, 0x46, 0x6c, 0x6f, 0x77, 0x46, 0x69, 0x65, 0x6c,
+	0x64, 0x73, 0x12, 0x17, 0x0a, 0x07, 0x66, 0x6c, 0x6f, 0x77, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0c, 0x52, 0x06, 0x66, 0x6c, 0x6f, 0x77, 0x49, 0x64, 0x12, 0x1e, 0x0a, 0x04, 0x74,
+	0x79, 0x70, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0a, 0x2e, 0x66, 0x6c, 0x6f, 0x77,
+	0x2e, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x17, 0x0a, 0x07, 0x72,
+	0x75, 0x6c, 0x65, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x72, 0x75,
+	0x6c, 0x65, 0x49, 0x64, 0x12, 0x2d, 0x0a, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f,
+	0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x44,
+	0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x09, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74,
+	0x69, 0x6f, 0x6e, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18,
+	0x05, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12,
+	0x1b, 0x0a, 0x09, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x69, 0x70, 0x18, 0x06, 0x20, 0x01,
+	0x28, 0x0c, 0x52, 0x08, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x49, 0x70, 0x12, 0x17, 0x0a, 0x07,
+	0x64, 0x65, 0x73, 0x74, 0x5f, 0x69, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x06, 0x64,
+	0x65, 0x73, 0x74, 0x49, 0x70, 0x12, 0x2d, 0x0a, 0x09, 0x70, 0x6f, 0x72, 0x74, 0x5f, 0x69, 0x6e,
+	0x66, 0x6f, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e,
+	0x50, 0x6f, 0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x70, 0x6f, 0x72, 0x74,
+	0x49, 0x6e, 0x66, 0x6f, 0x12, 0x2d, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x69, 0x6e, 0x66,
+	0x6f, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0e, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x49,
+	0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f, 0x48, 0x00, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x49,
+	0x6e, 0x66, 0x6f, 0x12, 0x1d, 0x0a, 0x0a, 0x72, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74,
+	0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x72, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65,
+	0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x74, 0x78, 0x5f, 0x70, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x73,
+	0x18, 0x0b, 0x20, 0x01, 0x28, 0x04, 0x52, 0x09, 0x74, 0x78, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74,
+	0x73, 0x12, 0x19, 0x0a, 0x08, 0x72, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0c, 0x20,
+	0x01, 0x28, 0x04, 0x52, 0x07, 0x72, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x12, 0x19, 0x0a, 0x08,
+	0x74, 0x78, 0x5f, 0x62, 0x79, 0x74, 0x65, 0x73, 0x18, 0x0d, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07,
+	0x74, 0x78, 0x42, 0x79, 0x74, 0x65, 0x73, 0x42, 0x11, 0x0a, 0x0f, 0x63, 0x6f, 0x6e, 0x6e, 0x65,
+	0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x66, 0x6f, 0x22, 0x48, 0x0a, 0x08, 0x50, 0x6f,
+	0x72, 0x74, 0x49, 0x6e, 0x66, 0x6f, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65,
+	0x5f, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0a, 0x73, 0x6f, 0x75,
+	0x72, 0x63, 0x65, 0x50, 0x6f, 0x72, 0x74, 0x12, 0x1b, 0x0a, 0x09, 0x64, 0x65, 0x73, 0x74, 0x5f,
+	0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x08, 0x64, 0x65, 0x73, 0x74,
+	0x50, 0x6f, 0x72, 0x74, 0x22, 0x44, 0x0a, 0x08, 0x49, 0x43, 0x4d, 0x50, 0x49, 0x6e, 0x66, 0x6f,
+	0x12, 0x1b, 0x0a, 0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0d, 0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1b, 0x0a,
+	0x09, 0x69, 0x63, 0x6d, 0x70, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d,
+	0x52, 0x08, 0x69, 0x63, 0x6d, 0x70, 0x43, 0x6f, 0x64, 0x65, 0x2a, 0x45, 0x0a, 0x04, 0x54, 0x79,
+	0x70, 0x65, 0x12, 0x10, 0x0a, 0x0c, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x55, 0x4e, 0x4b, 0x4e, 0x4f,
+	0x57, 0x4e, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x53, 0x54, 0x41,
+	0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x45, 0x4e, 0x44,
+	0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x54, 0x59, 0x50, 0x45, 0x5f, 0x44, 0x52, 0x4f, 0x50, 0x10,
+	0x03, 0x2a, 0x3b, 0x0a, 0x09, 0x44, 0x69, 0x72, 0x65, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x15,
+	0x0a, 0x11, 0x44, 0x49, 0x52, 0x45, 0x43, 0x54, 0x49, 0x4f, 0x4e, 0x5f, 0x55, 0x4e, 0x4b, 0x4e,
+	0x4f, 0x57, 0x4e, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x4e, 0x47, 0x52, 0x45, 0x53, 0x53,
+	0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x45, 0x47, 0x52, 0x45, 0x53, 0x53, 0x10, 0x02, 0x32, 0x42,
+	0x0a, 0x0b, 0x46, 0x6c, 0x6f, 0x77, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x33, 0x0a,
+	0x06, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x73, 0x12, 0x0f, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e, 0x46,
+	0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x1a, 0x12, 0x2e, 0x66, 0x6c, 0x6f, 0x77, 0x2e,
+	0x46, 0x6c, 0x6f, 0x77, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x41, 0x63, 0x6b, 0x22, 0x00, 0x28, 0x01,
+	0x30, 0x01, 0x42, 0x08, 0x5a, 0x06, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -577,18 +633,18 @@ var file_flow_proto_goTypes = []interface{}{
 	(Direction)(0),                // 1: flow.Direction
 	(*FlowEvent)(nil),             // 2: flow.FlowEvent
 	(*FlowEventAck)(nil),          // 3: flow.FlowEventAck
-	(*EventFields)(nil),           // 4: flow.EventFields
+	(*FlowFields)(nil),            // 4: flow.FlowFields
 	(*PortInfo)(nil),              // 5: flow.PortInfo
 	(*ICMPInfo)(nil),              // 6: flow.ICMPInfo
 	(*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp
 }
 var file_flow_proto_depIdxs = []int32{
 	7, // 0: flow.FlowEvent.timestamp:type_name -> google.protobuf.Timestamp
-	4, // 1: flow.FlowEvent.event_fields:type_name -> flow.EventFields
-	0, // 2: flow.EventFields.type:type_name -> flow.Type
-	1, // 3: flow.EventFields.direction:type_name -> flow.Direction
-	5, // 4: flow.EventFields.port_info:type_name -> flow.PortInfo
-	6, // 5: flow.EventFields.icmp_info:type_name -> flow.ICMPInfo
+	4, // 1: flow.FlowEvent.flow_fields:type_name -> flow.FlowFields
+	0, // 2: flow.FlowFields.type:type_name -> flow.Type
+	1, // 3: flow.FlowFields.direction:type_name -> flow.Direction
+	5, // 4: flow.FlowFields.port_info:type_name -> flow.PortInfo
+	6, // 5: flow.FlowFields.icmp_info:type_name -> flow.ICMPInfo
 	2, // 6: flow.FlowService.Events:input_type -> flow.FlowEvent
 	3, // 7: flow.FlowService.Events:output_type -> flow.FlowEventAck
 	7, // [7:8] is the sub-list for method output_type
@@ -629,7 +685,7 @@ func file_flow_proto_init() {
 			}
 		}
 		file_flow_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*EventFields); i {
+			switch v := v.(*FlowFields); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -666,8 +722,8 @@ func file_flow_proto_init() {
 		}
 	}
 	file_flow_proto_msgTypes[2].OneofWrappers = []interface{}{
-		(*EventFields_PortInfo)(nil),
-		(*EventFields_IcmpInfo)(nil),
+		(*FlowFields_PortInfo)(nil),
+		(*FlowFields_IcmpInfo)(nil),
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{