feat(collector): use FlowSummary to reduce conntrack peak memory

cmd/agent/main.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/zerolog/log"
-	"github.com/ti-mo/conntrack"
+	"github.com/ClickHouse/conntrack"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 

go.mod

@@ -1,15 +1,16 @@
 module github.com/ClickHouse/kubenetmon
 
-go 1.23
+go 1.23.0
+
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.26.0
+	github.com/ClickHouse/conntrack v1.0.1-0.20260318115422-f142e39cec50
 	github.com/PraserX/ipconv v1.2.0
 	github.com/google/uuid v1.6.0
 	github.com/prometheus/client_golang v1.19.1
 	github.com/rs/zerolog v1.32.0
 	github.com/seancfoley/ipaddress-go v1.6.0
 	github.com/stretchr/testify v1.9.0
-	github.com/ti-mo/conntrack v0.5.0
 	go.uber.org/mock v0.4.0
 	google.golang.org/grpc v1.65.0
 	google.golang.org/protobuf v1.34.2
@@ -36,7 +37,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/josharian/native v1.1.0 // indirect
@@ -46,7 +47,7 @@ require (
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.19 // indirect
 	github.com/mdlayher/netlink v1.7.2 // indirect
-	github.com/mdlayher/socket v0.4.1 // indirect
+	github.com/mdlayher/socket v0.5.1 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
@@ -60,15 +61,15 @@ require (
 	github.com/seancfoley/bintree v1.3.1 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
 	github.com/shopspring/decimal v1.4.0 // indirect
-	github.com/ti-mo/netfilter v0.5.0 // indirect
+	github.com/ti-mo/netfilter v0.5.3 // indirect
 	go.opentelemetry.io/otel v1.26.0 // indirect
 	go.opentelemetry.io/otel/trace v1.26.0 // indirect
-	golang.org/x/net v0.36.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
-	golang.org/x/sync v0.11.0 // indirect
-	golang.org/x/sys v0.30.0 // indirect
-	golang.org/x/term v0.29.0 // indirect
-	golang.org/x/text v0.22.0 // indirect
+	golang.org/x/sync v0.14.0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	golang.org/x/term v0.31.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240730163845-b1a4ccb954bf // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect

go.sum

@@ -2,6 +2,8 @@ github.com/ClickHouse/ch-go v0.61.5 h1:zwR8QbYI0tsMiEcze/uIMK+Tz1D3XZXLdNrlaOpeE
 github.com/ClickHouse/ch-go v0.61.5/go.mod h1:s1LJW/F/LcFs5HJnuogFMta50kKDO0lf9zzfrbl0RQg=
 github.com/ClickHouse/clickhouse-go/v2 v2.26.0 h1:j4/y6NYaCcFkJwN/TU700ebW+nmsIy34RmUAAcZKy9w=
 github.com/ClickHouse/clickhouse-go/v2 v2.26.0/go.mod h1:iDTViXk2Fgvf1jn2dbJd1ys+fBkdD1UMRnXlwmhijhQ=
+github.com/ClickHouse/conntrack v1.0.1-0.20260318115422-f142e39cec50 h1:mKBWz03IBRCs0MwyD3Jzh1/luKVGfVgA2uSSLsnCSco=
+github.com/ClickHouse/conntrack v1.0.1-0.20260318115422-f142e39cec50/go.mod h1:h6dMz6YHTczWWCU164lkpw/6BNxeo6+YzyYR1uJ6qX8=
 github.com/PraserX/ipconv v1.2.0 h1:3bboP9EDfsuMF5C3qM25OmZA4+cCfk1Ahpx8zn5G2tM=
 github.com/PraserX/ipconv v1.2.0/go.mod h1:aBiLM1bDAjp1++Q0Sp3IrGbPd49b74IUgoRWod6EtPY=
 github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
@@ -45,8 +47,8 @@ github.com/google/gnostic-models v0.6.8/go.mod h1:5n7qKqH0f5wFt+aWF8CW6pZLLNOfYu
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
-github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
-github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -82,8 +84,8 @@ github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APP
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g=
 github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw=
-github.com/mdlayher/socket v0.4.1 h1:eM9y2/jlbs1M615oshPQOHZzj6R6wMT7bX5NPiQvn2U=
-github.com/mdlayher/socket v0.4.1/go.mod h1:cAqeGjoufqdxWkD7DkpyS+wcefOtmu5OQ8KuoJGIReA=
+github.com/mdlayher/socket v0.5.1 h1:VZaqt6RkGkt2OE9l3GcC6nZkqD3xKeQLyfleW/uBcos=
+github.com/mdlayher/socket v0.5.1/go.mod h1:TjPLHI1UgwEv5J1B5q0zTZq12A/6H7nKmtTanQE37IQ=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -138,10 +140,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-github.com/ti-mo/conntrack v0.5.0 h1:OWiWm18gx6IA0c8FvLuXpcvHUsR0Cyw6FIFIZtYJ2W4=
-github.com/ti-mo/conntrack v0.5.0/go.mod h1:xTW+s2bugPtNnx58p1yyz+UADwho2cZFom6SsK0UTw0=
-github.com/ti-mo/netfilter v0.5.0 h1:MZmsUw5bFRecOb0AeyjOPxTHg4UxYzyEs0Ek/6Lxoy8=
-github.com/ti-mo/netfilter v0.5.0/go.mod h1:nt+8B9hx/QpqHr7Hazq+2qMCCA8u2OTkyc/7+U9ARz8=
+github.com/ti-mo/netfilter v0.5.3 h1:ikzduvnaUMwre5bhbNwWOd6bjqLMVb33vv0XXbK0xGQ=
+github.com/ti-mo/netfilter v0.5.3/go.mod h1:08SyBCg6hu1qyQk4s3DjjJKNrm3RTb32nm6AzyT972E=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
 github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
@@ -169,16 +169,16 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.36.0 h1:vWF2fRbw4qslQsQzgFqZff+BItCvGFQqKzKIzx1rmoA=
-golang.org/x/net v0.36.0/go.mod h1:bFmbeoIPfrw4sMHNhb4J9f6+tPziuGjq7Jk/38fxi1I=
+golang.org/x/net v0.39.0 h1:ZCu7HMWDxpXpaiKdhzIfaltL9Lp31x/3fCP11bc6/fY=
+golang.org/x/net v0.39.0/go.mod h1:X7NRbYVEA+ewNkCNyJ513WmMdQ3BineSwVtN2zD/d+E=
 golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs=
 golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
-golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -188,17 +188,17 @@ golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
-golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
-golang.org/x/term v0.29.0 h1:L6pJp37ocefwRRtYPKSWOWzOtWSxVajvz2ldH/xi3iU=
-golang.org/x/term v0.29.0/go.mod h1:6bl4lRlvVuDgSf3179VpIxBF0o10JUpXWOnI7nErv7s=
+golang.org/x/term v0.31.0 h1:erwDkOK1Msy6offm1mOgvspSkslFnIGsFnxOKoufg3o=
+golang.org/x/term v0.31.0/go.mod h1:R4BeIy7D95HzImkxGkTW1UQTtP54tio2RyHz7PwK0aw=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM=
-golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY=
+golang.org/x/text v0.24.0 h1:dd5Bzh4yt5KYA8f9CJHCP4FB4D51c2c6JvN37xJJkJ0=
+golang.org/x/text v0.24.0/go.mod h1:L8rBsPeo2pSS+xqN0d5u2ikmjtmoJbDBT1b7nHvFCdU=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

pkg/collector/collector.go

@@ -15,7 +15,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/rs/zerolog/log"
-	"github.com/ti-mo/conntrack"
+	"github.com/ClickHouse/conntrack"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/status"
 )
@@ -131,7 +131,10 @@ func (collector *Collector) collectOnce() error {
 	}
 
 	now := collector.clock.Now()
-	flows, err := collector.conntrack.Dump(&conntrack.DumpOptions{ZeroCounters: true})
+	flows, err := collector.conntrack.DumpFlowSummaryFilter(
+		conntrack.NewExcludeUDPFilter(),
+		&conntrack.DumpOptions{ZeroCounters: true, Family: conntrack.ProtoIPv4},
+	)
 	if err != nil {
 		return fmt.Errorf("could not dump conntrack: %v", err)
 	}
@@ -185,7 +188,10 @@ func (collector *Collector) collectOnce() error {
 func (collector *Collector) conntrackCountsNonEmpty(uptimeWaitDuration time.Duration) (bool, error) {
 	for {
 		// Check conntrack counters.
-		flows, err := collector.conntrack.Dump(&conntrack.DumpOptions{ZeroCounters: false})
+		flows, err := collector.conntrack.DumpFlowSummaryFilter(
+			conntrack.NewExcludeUDPFilter(),
+			&conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4},
+		)
 		if err != nil {
 			return false, err
 		}
@@ -224,7 +230,7 @@ func (collector *Collector) conntrackCountsNonEmpty(uptimeWaitDuration time.Dura
 
 // shouldIgnoreFlow returns true if the flow is to be ignored and not reported
 // to the server for labelling, for example because it's a localhost flow.
-func (collector *Collector) shouldIgnoreFlow(flow *conntrack.Flow) bool {
+func (collector *Collector) shouldIgnoreFlow(flow *conntrack.FlowSummary) bool {
 	// Ignore flows with no data to report.
 	if flow.CountersOrig.Bytes == 0 && flow.CountersOrig.Packets == 0 && flow.CountersReply.Bytes == 0 && flow.CountersReply.Packets == 0 {
 		return true

pkg/collector/collector_test.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/ti-mo/conntrack"
+	"github.com/ClickHouse/conntrack"
 	"go.uber.org/mock/gomock"
 
 	mock_collector "github.com/ClickHouse/kubenetmon/pkg/collector/mock"
@@ -26,7 +26,7 @@ func TestNewCollector(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return(nil, errFake)
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return(nil, errFake)
 		collector, err := NewCollector(mockConntrack, nil, nil, 15*time.Second, "node", false, 1*time.Second)
 		assert.Error(t, err)
 		assert.Nil(t, collector)
@@ -36,7 +36,7 @@ func TestNewCollector(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return([]conntrack.Flow{
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{
 			{
 				CountersOrig: conntrack.Counter{
 					Packets: 0,
@@ -53,7 +53,7 @@ func TestNewCollector(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return([]conntrack.Flow{
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{
 			{
 				CountersOrig: conntrack.Counter{
 					Packets: 0,
@@ -81,7 +81,7 @@ func TestConntrackCountsNonEmpty(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return([]conntrack.Flow{
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{
 			{
 				CountersOrig: conntrack.Counter{
 					Packets: 0,
@@ -109,7 +109,7 @@ func TestConntrackCountsNonEmpty(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return([]conntrack.Flow{
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{
 			{
 				CountersOrig: conntrack.Counter{
 					Packets: 0,
@@ -137,7 +137,7 @@ func TestConntrackCountsNonEmpty(t *testing.T) {
 		t.Parallel()
 
 		mockConntrack := mock_collector.NewConntrack(ctrl)
-		mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return(nil, errFake)
+		mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return(nil, errFake)
 
 		collector := Collector{
 			conntrack: mockConntrack,
@@ -156,7 +156,7 @@ func TestShouldIgnoreFlow(t *testing.T) {
 		t.Parallel()
 
 		collector := Collector{}
-		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.Flow{
+		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.FlowSummary{
 			CountersOrig: conntrack.Counter{
 				Packets: 0,
 				Bytes:   0,
@@ -169,7 +169,7 @@ func TestShouldIgnoreFlow(t *testing.T) {
 		t.Parallel()
 
 		collector := Collector{}
-		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.Flow{
+		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.FlowSummary{
 			TupleOrig: conntrack.Tuple{
 				Proto: conntrack.ProtoTuple{
 					Protocol: 123,
@@ -187,7 +187,7 @@ func TestShouldIgnoreFlow(t *testing.T) {
 		t.Parallel()
 
 		collector := Collector{}
-		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.Flow{
+		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.FlowSummary{
 			TupleOrig: conntrack.Tuple{
 				Proto: conntrack.ProtoTuple{
 					Protocol: IP_PROTO_TCP,
@@ -209,7 +209,7 @@ func TestShouldIgnoreFlow(t *testing.T) {
 		t.Parallel()
 
 		collector := Collector{}
-		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.Flow{
+		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.FlowSummary{
 			TupleOrig: conntrack.Tuple{
 				Proto: conntrack.ProtoTuple{
 					Protocol: IP_PROTO_TCP,
@@ -231,7 +231,7 @@ func TestShouldIgnoreFlow(t *testing.T) {
 		t.Parallel()
 
 		collector := Collector{}
-		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.Flow{
+		shouldIgnore := collector.shouldIgnoreFlow(&conntrack.FlowSummary{
 			TupleOrig: conntrack.Tuple{
 				Proto: conntrack.ProtoTuple{
 					Protocol: IP_PROTO_TCP,
@@ -280,15 +280,15 @@ func TestCollect(t *testing.T) {
 	)
 
 	mockConntrack := mock_collector.NewConntrack(ctrl)
-	mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: false}).Return([]conntrack.Flow{
+	mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: false, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{
 		{
 			CountersReply: conntrack.Counter{
 				Packets: 0,
 				Bytes:   1,
 			},
 		},
 	}, nil)
-	mockConntrack.EXPECT().Dump(&conntrack.DumpOptions{ZeroCounters: true}).Return([]conntrack.Flow{{
+	mockConntrack.EXPECT().DumpFlowSummaryFilter(conntrack.NewExcludeUDPFilter(), &conntrack.DumpOptions{ZeroCounters: true, Family: conntrack.ProtoIPv4}).Return([]conntrack.FlowSummary{{
 		CountersOrig: conntrack.Counter{
 			Packets: origPackets,
 			Bytes:   origBytes,

pkg/collector/conntrack.go

@@ -1,9 +1,9 @@
 package collector
 
-import "github.com/ti-mo/conntrack"
+import "github.com/ClickHouse/conntrack"
 
 // ConntrackInterface is a dummy interface used to generate a mock "conntrack"
 // implementation used in unit tests.
 type ConntrackInterface interface {
-	Dump(opts *conntrack.DumpOptions) ([]conntrack.Flow, error)
+	DumpFlowSummaryFilter(filter conntrack.FlowSummaryFilter, opts *conntrack.DumpOptions) ([]conntrack.FlowSummary, error)
 }