kube-proxy: optimize conntrack cleanup with O(n) flow filter

Previously, we created a separate filter for each stale flow, resulting in O(n^2) complexity when deleting flows because the netlink llibrary iterates over all filters for each flow. This change introduces a new filter backed by a `sets.Set` for O(1) lookup per flow. This reduces the overall complexity of cleaning up stale entries to O(n).
2026-07-17 02:06:23 +00:00 · 2025-11-29 17:30:47 +00:00
parent 61c629cc57
commit 38e08c231c
7 changed files with 356 additions and 439 deletions
--- a/pkg/proxy/conntrack/cleanup.go
+++ b/pkg/proxy/conntrack/cleanup.go
@@ -113,7 +113,7 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 		}
 	}

-	var filters []netlink.CustomConntrackFilter
+	var flows []*netlink.ConntrackFlow
 	for _, entry := range entries {
 		// we only deal with UDP protocol entries
 		if entry.Forward.Protocol != unix.IPPROTO_UDP {
@@ -133,7 +133,8 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 		// represent a serving endpoint of the service, we clear the entry.
 		endpoints, ok := serviceIPEndpoints[net.JoinHostPort(origDst, origPortDstStr)]
 		if ok && !endpoints.Has(net.JoinHostPort(replySrc, replyPortSrcStr)) {
-			filters = append(filters, filterForIPPortNAT(entry.Forward.DstIP, entry.Forward.DstPort, entry.Reverse.SrcIP, entry.Reverse.SrcPort, v1.ProtocolUDP))
+			flows = append(flows, entry)
+			continue
 		}

 		// if the original destination port (--orig-port-dst) of the entry is service
@@ -141,12 +142,13 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily,
 		// does not represent a serving endpoint of the service, we clear the entry.
 		endpoints, ok = serviceNodePortEndpoints[origPortDst]
 		if ok && !endpoints.Has(net.JoinHostPort(replySrc, replyPortSrcStr)) {
-			filters = append(filters, filterForPortNAT(entry.Forward.DstPort, entry.Reverse.SrcIP, entry.Reverse.SrcPort, v1.ProtocolUDP))
+			flows = append(flows, entry)
+			continue
 		}
 	}

 	var n int
-	if n, err = ct.ClearEntries(ipFamilyMap[ipFamily], filters...); err != nil {
+	if n, err = ct.DeleteEntries(ipFamilyMap[ipFamily], flows); err != nil {
 		klog.ErrorS(err, "Failed to clear all conntrack entries", "ipFamily", ipFamily, "entriesDeleted", n, "took", time.Since(start))
 	} else {
 		klog.V(4).InfoS("Finished reconciling conntrack entries", "ipFamily", ipFamily, "entriesDeleted", n, "took", time.Since(start))
@@ -160,45 +162,3 @@ var ipFamilyMap = map[v1.IPFamily]uint8{
 	v1.IPv4Protocol: unix.AF_INET,
 	v1.IPv6Protocol: unix.AF_INET6,
 }
-
-// protocolMap maps v1.Protocol to the Assigned Internet Protocol Number.
-// https://www.iana.org/assignments/protocol-numbers/protocol-numbers.xhtml
-var protocolMap = map[v1.Protocol]uint8{
-	v1.ProtocolTCP:  unix.IPPROTO_TCP,
-	v1.ProtocolUDP:  unix.IPPROTO_UDP,
-	v1.ProtocolSCTP: unix.IPPROTO_SCTP,
-}
-
-// filterForIPPortNAT returns *conntrackFilter to delete the conntrack entries for connections
-// specified by the destination IP (original direction) and destination port (original direction)
-// and source IP (reply direction) and source port (reply direction).
-func filterForIPPortNAT(origDst net.IP, origPortDst uint16, replySrc net.IP, replyPortSrc uint16, protocol v1.Protocol) *conntrackFilter {
-	klog.V(6).InfoS("Adding conntrack filter for cleanup", "orig-dst", origDst.String(), "orig-port-dst", origPortDst, "reply-src", replySrc.String(), "reply-port-src", replyPortSrc, "protocol", protocol)
-	return &conntrackFilter{
-		protocol: protocolMap[protocol],
-		original: &connectionTuple{
-			dstIP:   origDst,
-			dstPort: origPortDst,
-		},
-		reply: &connectionTuple{
-			srcIP:   replySrc,
-			srcPort: replyPortSrc,
-		},
-	}
-}
-
-// filterForPortNAT returns *conntrackFilter to delete the conntrack entries for connections specified by the
-// destination Port (original direction) and source IP (reply direction) and source port (reply direction).
-func filterForPortNAT(origPortDst uint16, replySrc net.IP, replyPortSrc uint16, protocol v1.Protocol) *conntrackFilter {
-	klog.V(6).InfoS("Adding conntrack filter for cleanup", "orig-port-dst", origPortDst, "reply-src", replySrc.String(), "reply-port-src", replyPortSrc, "protocol", protocol)
-	return &conntrackFilter{
-		protocol: protocolMap[protocol],
-		original: &connectionTuple{
-			dstPort: origPortDst,
-		},
-		reply: &connectionTuple{
-			srcIP:   replySrc,
-			srcPort: replyPortSrc,
-		},
-	}
-}
--- a/pkg/proxy/conntrack/cleanup_test.go
+++ b/pkg/proxy/conntrack/cleanup_test.go
@@ -22,7 +22,6 @@ package conntrack
 import (
 	"fmt"
 	"math/rand"
-	"net"
 	"sort"
 	"testing"

@@ -318,12 +317,15 @@ func TestCleanStaleEntries(t *testing.T) {
 	}

 	t.Logf("entries before cleanup %d after cleanup %d", len(entriesBeforeCleanup), len(entriesAfterCleanup))
-	fake := NewFake()
-	fake.entries = entriesBeforeCleanup
+	ct := newConntracker(
+		&fakeHandler{
+			entries: entriesBeforeCleanup,
+		},
+	)

 	legacyregistry.MustRegister(metrics.ReconcileConntrackFlowsDeletedEntriesTotal)
-	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
-	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	CleanStaleEntries(ct, testIPFamily, svcPortMap, endpointsMap)
+	actualEntries, _ := ct.ListEntries(ipFamilyMap[testIPFamily])

 	metricCount, err := testutil.GetCounterMetricValue(metrics.ReconcileConntrackFlowsDeletedEntriesTotal.WithLabelValues(string(testIPFamily)))
 	require.NoError(t, err)
@@ -408,12 +410,12 @@ func TestPerformanceCleanStaleEntries(t *testing.T) {
 	endpointsMap := make(proxy.EndpointsMap)
 	_ = endpointsMap.Update(ect)

-	fake := NewFake()
+	flows := []*netlink.ConntrackFlow{}
 	// 1 valid entry
-	fake.entries = append(fake.entries, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
 	expectedEntries := 1
 	// 1 stale entry
-	fake.entries = append(fake.entries, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
 	expectedDeleted := 1
 	// 1 M to the Service IP with random ports
 	for i := 0; i < 1000*1000; i++ {
@@ -423,11 +425,14 @@ func TestPerformanceCleanStaleEntries(t *testing.T) {
 		} else {
 			expectedEntries++
 		}
-		fake.entries = append(fake.entries, generateConntrackEntry(testExternalIP, port, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+		flows = append(flows, generateConntrackEntry(testExternalIP, port, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
 	}

-	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
-	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	fake := &fakeHandler{entries: flows}
+	ct := newConntracker(fake)
+
+	CleanStaleEntries(ct, testIPFamily, svcPortMap, endpointsMap)
+	actualEntries, _ := ct.ListEntries(ipFamilyMap[testIPFamily])
 	if len(actualEntries) != expectedEntries {
 		t.Errorf("unexpected number of entries, got %d expected %d", len(actualEntries), expectedEntries)
 	}
@@ -503,103 +508,20 @@ func TestServiceWithoutEndpoints(t *testing.T) {
 	endpointsMap := make(proxy.EndpointsMap)
 	_ = endpointsMap.Update(ect)

-	fake := NewFake()
+	flows := []*netlink.ConntrackFlow{}
 	// 1 valid entry
-	fake.entries = append(fake.entries, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
 	// 1 stale entry
-	fake.entries = append(fake.entries, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP))
+	ct := newConntracker(
+		&fakeHandler{
+			entries: flows,
+		},
+	)

-	CleanStaleEntries(fake, testIPFamily, svcPortMap, endpointsMap)
-	actualEntries, _ := fake.ListEntries(ipFamilyMap[testIPFamily])
+	CleanStaleEntries(ct, testIPFamily, svcPortMap, endpointsMap)
+	actualEntries, _ := ct.ListEntries(ipFamilyMap[testIPFamily])
 	if len(actualEntries) != 2 {
 		t.Errorf("unexpected number of entries, got %d expected %d", len(actualEntries), 2)
 	}
 }
-
-func TestFilterForIPPortNAT(t *testing.T) {
-	testCases := []struct {
-		name           string
-		origDst        net.IP
-		origPortDst    uint16
-		replySrc       net.IP
-		replySrcPort   uint16
-		protocol       v1.Protocol
-		expectedFilter *conntrackFilter
-	}{
-		{
-			name:         "ipv4 + SCTP",
-			origDst:      netutils.ParseIPSloppy("10.96.0.10"),
-			origPortDst:  80,
-			replySrc:     netutils.ParseIPSloppy("10.244.0.3"),
-			replySrcPort: 3000,
-			protocol:     v1.ProtocolSCTP,
-			expectedFilter: &conntrackFilter{
-				protocol: 132,
-				original: &connectionTuple{dstIP: netutils.ParseIPSloppy("10.96.0.10"), dstPort: 80},
-				reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("10.244.0.3"), srcPort: 3000},
-			},
-		},
-		{
-			name:         "ipv6 + UDP",
-			origDst:      netutils.ParseIPSloppy("2001:db8:1::2"),
-			origPortDst:  443,
-			replySrc:     netutils.ParseIPSloppy("4001:ab8::2"),
-			replySrcPort: 5000,
-			protocol:     v1.ProtocolUDP,
-			expectedFilter: &conntrackFilter{
-				protocol: 17,
-				original: &connectionTuple{dstIP: netutils.ParseIPSloppy("2001:db8:1::2"), dstPort: 443},
-				reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("4001:ab8::2"), srcPort: 5000},
-			},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expectedFilter, filterForIPPortNAT(tc.origDst, tc.origPortDst, tc.replySrc, tc.replySrcPort, tc.protocol))
-		})
-	}
-}
-
-func TestFilterForPortNAT(t *testing.T) {
-	testCases := []struct {
-		name           string
-		origPortDst    uint16
-		replySrc       net.IP
-		replySrcPort   uint16
-		protocol       v1.Protocol
-		expectedFamily netlink.InetFamily
-		expectedFilter *conntrackFilter
-	}{
-		{
-			name:         "ipv4 + TCP",
-			origPortDst:  80,
-			replySrc:     netutils.ParseIPSloppy("10.96.0.10"),
-			replySrcPort: 3000,
-			protocol:     v1.ProtocolTCP,
-			expectedFilter: &conntrackFilter{
-				protocol: 6,
-				original: &connectionTuple{dstPort: 80},
-				reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("10.96.0.10"), srcPort: 3000},
-			},
-		},
-		{
-			name:         "ipv6 + UDP",
-			origPortDst:  8000,
-			replySrc:     netutils.ParseIPSloppy("2001:db8:1::2"),
-			replySrcPort: 5000,
-			protocol:     v1.ProtocolUDP,
-			expectedFilter: &conntrackFilter{
-				protocol: 17,
-				original: &connectionTuple{dstPort: 8000},
-				reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("2001:db8:1::2"), srcPort: 5000},
-			},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			require.Equal(t, tc.expectedFilter, filterForPortNAT(tc.origPortDst, tc.replySrc, tc.replySrcPort, tc.protocol))
-		})
-	}
-}
--- a/pkg/proxy/conntrack/conntrack.go
+++ b/pkg/proxy/conntrack/conntrack.go
@@ -25,16 +25,14 @@ import (
 	"github.com/vishvananda/netlink"

 	"k8s.io/client-go/util/retry"
-	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/proxy/util"
 )

 // Interface for dealing with conntrack
 type Interface interface {
 	ListEntries(ipFamily uint8) ([]*netlink.ConntrackFlow, error)
-	// ClearEntries deletes conntrack entries for connections of the given IP family,
-	// filtered by the given filters.
-	ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) (int, error)
+	// DeleteEntries deletes the given conntrack entries.
+	DeleteEntries(ipFamily uint8, flows []*netlink.ConntrackFlow) (int, error)
 }

 // netlinkHandler allows consuming real and mockable implementation for testing.
@@ -67,19 +65,19 @@ func (ct *conntracker) ListEntries(ipFamily uint8) (entries []*netlink.Conntrack
 	return entries, err
 }

-// ClearEntries deletes conntrack entries for connections of the given IP family,
-// filtered by the given filters.
-func (ct *conntracker) ClearEntries(ipFamily uint8, filters ...netlink.CustomConntrackFilter) (int, error) {
-	if len(filters) == 0 {
-		klog.V(7).InfoS("no conntrack filters provided")
+// DeleteEntries deletes the given conntrack entries on the specified IP family.
+// It returns the number of deleted entries and an error if any.
+func (ct *conntracker) DeleteEntries(ipFamily uint8, flows []*netlink.ConntrackFlow) (int, error) {
+	if len(flows) == 0 {
 		return 0, nil
 	}

+	filter := newFlowFilter(flows)
 	var n uint
 	var err error
 	err = retry.OnError(util.MaxAttemptsEINTR, util.ShouldRetryOnEINTR, func() error {
 		var count uint
-		count, err = ct.handler.ConntrackDeleteFilters(netlink.ConntrackTable, netlink.InetFamily(ipFamily), filters...)
+		count, err = ct.handler.ConntrackDeleteFilters(netlink.ConntrackTable, netlink.InetFamily(ipFamily), filter)
 		n += count
 		return err
 	})
--- a/pkg/proxy/conntrack/conntrack_test.go
+++ b/pkg/proxy/conntrack/conntrack_test.go
@@ -22,85 +22,146 @@ package conntrack
 import (
 	"testing"

-	"github.com/stretchr/testify/require"
 	"github.com/vishvananda/netlink"
 	"golang.org/x/sys/unix"

 	netutils "k8s.io/utils/net"
 )

-type fakeHandler struct {
-	tableType netlink.ConntrackTableType
-	ipFamily  netlink.InetFamily
-	filters   []*conntrackFilter
-}
-
-func (f *fakeHandler) ConntrackTableList(_ netlink.ConntrackTableType, _ netlink.InetFamily) ([]*netlink.ConntrackFlow, error) {
-	return nil, nil
-}
-
-func (f *fakeHandler) ConntrackDeleteFilters(tableType netlink.ConntrackTableType, family netlink.InetFamily, netlinkFilters ...netlink.CustomConntrackFilter) (uint, error) {
-	f.tableType = tableType
-	f.ipFamily = family
-	f.filters = make([]*conntrackFilter, 0, len(netlinkFilters))
-	for _, netlinkFilter := range netlinkFilters {
-		f.filters = append(f.filters, netlinkFilter.(*conntrackFilter))
+func TestConntracker_DeleteEntries(t *testing.T) {
+	// Define some common flows
+	flow1 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("1.1.1.1"),
+			DstIP:    netutils.ParseIPSloppy("2.2.2.2"),
+			SrcPort:  1000,
+			DstPort:  2000,
+			Protocol: 6,
+		},
+		Reverse: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("2.2.2.2"),
+			DstIP:    netutils.ParseIPSloppy("1.1.1.1"),
+			SrcPort:  2000,
+			DstPort:  1000,
+			Protocol: 6,
+		},
+	}
+	flow2 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("3.3.3.3"),
+			DstIP:    netutils.ParseIPSloppy("4.4.4.4"),
+			SrcPort:  3000,
+			DstPort:  4000,
+			Protocol: 17,
+		},
+		Reverse: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("4.4.4.4"),
+			DstIP:    netutils.ParseIPSloppy("3.3.3.3"),
+			SrcPort:  4000,
+			DstPort:  3000,
+			Protocol: 17,
+		},
+	}
+	flow3 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("5.5.5.5"),
+			DstIP:    netutils.ParseIPSloppy("6.6.6.6"),
+			SrcPort:  5000,
+			DstPort:  6000,
+			Protocol: 132,
+		},
+		Reverse: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("6.6.6.6"),
+			DstIP:    netutils.ParseIPSloppy("5.5.5.5"),
+			SrcPort:  6000,
+			DstPort:  5000,
+			Protocol: 132,
+		},
+	}
+	// Edge case: Partial flow (missing some fields)
+	flowEmpty := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			Protocol: 6,
+		},
 	}
-	return uint(len(f.filters)), nil
-}

-var _ netlinkHandler = (*fakeHandler)(nil)
-
-func TestConntracker_ClearEntries(t *testing.T) {
 	testCases := []struct {
-		name     string
-		ipFamily uint8
-		filters  []netlink.CustomConntrackFilter
+		name            string
+		initialEntries  []*netlink.ConntrackFlow
+		deleteFlows     []*netlink.ConntrackFlow
+		expectedDeleted int
+		expectedLeft    int
 	}{
 		{
-			name:     "single IPv6 filter",
-			ipFamily: unix.AF_INET6,
-			filters: []netlink.CustomConntrackFilter{
-				&conntrackFilter{
-					protocol: 17,
-					original: &connectionTuple{dstPort: 8000},
-					reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("2001:db8:1::2")},
-				},
-			},
+			name:            "Delete single flow",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2, flow3},
+			deleteFlows:     []*netlink.ConntrackFlow{flow1},
+			expectedDeleted: 1,
+			expectedLeft:    2,
 		},
 		{
-			name:     "multiple IPv4 filters",
-			ipFamily: unix.AF_INET,
-			filters: []netlink.CustomConntrackFilter{
-				&conntrackFilter{
-					protocol: 6,
-					original: &connectionTuple{dstPort: 3000},
-				},
-				&conntrackFilter{
-					protocol: 17,
-					original: &connectionTuple{dstPort: 5000},
-					reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("10.244.0.3")},
-				},
-				&conntrackFilter{
-					protocol: 132,
-					original: &connectionTuple{dstIP: netutils.ParseIPSloppy("10.96.0.10")},
-					reply:    &connectionTuple{srcIP: netutils.ParseIPSloppy("10.244.0.3")},
-				},
-			},
+			name:            "Delete multiple flows",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2, flow3},
+			deleteFlows:     []*netlink.ConntrackFlow{flow1, flow3},
+			expectedDeleted: 2,
+			expectedLeft:    1,
+		},
+		{
+			name:            "Delete non-existent flow",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2},
+			deleteFlows:     []*netlink.ConntrackFlow{flow3},
+			expectedDeleted: 0,
+			expectedLeft:    2,
+		},
+		{
+			name:            "Delete flow with empty fields (should match nothing)",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2, flow3},
+			deleteFlows:     []*netlink.ConntrackFlow{flowEmpty},
+			expectedDeleted: 0,
+			expectedLeft:    3,
+		},
+		{
+			name:            "Delete all flows",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2, flow3},
+			deleteFlows:     []*netlink.ConntrackFlow{flow1, flow2, flow3},
+			expectedDeleted: 3,
+			expectedLeft:    0,
+		},
+		{
+			name:            "Delete with empty list",
+			initialEntries:  []*netlink.ConntrackFlow{flow1, flow2},
+			deleteFlows:     []*netlink.ConntrackFlow{},
+			expectedDeleted: 0,
+			expectedLeft:    2,
 		},
 	}

 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			handler := &fakeHandler{}
+			// avoid mutations of entries between tests
+			entries := make([]*netlink.ConntrackFlow, len(tc.initialEntries))
+			copy(entries, tc.initialEntries)
+
+			handler := &fakeHandler{
+				entries: entries,
+			}
 			ct := newConntracker(handler)
-			_, err := ct.ClearEntries(tc.ipFamily, tc.filters...)
-			require.NoError(t, err)
-			require.Equal(t, netlink.ConntrackTableType(netlink.ConntrackTable), handler.tableType)
-			require.Equal(t, netlink.InetFamily(tc.ipFamily), handler.ipFamily)
-			require.Equal(t, len(tc.filters), len(handler.filters))
-			for i := 0; i < len(tc.filters); i++ {
-				require.Equal(t, tc.filters[i], handler.filters[i])
+
+			n, err := ct.DeleteEntries(unix.AF_INET, tc.deleteFlows)
+			if err != nil {
+				t.Fatalf("DeleteEntries() error = %v, want nil", err)
+			}
+
+			if n != tc.expectedDeleted {
+				t.Errorf("DeleteEntries() = %d, want %d", n, tc.expectedDeleted)
+			}
+
+			left, err := ct.ListEntries(unix.AF_INET)
+			if err != nil {
+				t.Fatalf("ListEntries() error = %v", err)
+			}
+			if len(left) != tc.expectedLeft {
+				t.Errorf("ListEntries() left = %d, want %d", len(left), tc.expectedLeft)
 			}
 		})
 	}
--- a/pkg/proxy/conntrack/fake.go
+++ b/pkg/proxy/conntrack/fake.go
@@ -20,55 +20,60 @@ limitations under the License.
 package conntrack

 import (
+	"sync"
+
 	"github.com/vishvananda/netlink"
 )

-// FakeInterface implements Interface by just recording entries that have been cleared.
-type FakeInterface struct {
+// NewFake creates a new FakeInterface
+func NewFake() Interface {
+	return newConntracker(
+		&fakeHandler{
+			entries: make([]*netlink.ConntrackFlow, 0),
+		},
+	)
+}
+
+var _ netlinkHandler = (*fakeHandler)(nil)
+
+type fakeHandler struct {
+	mu              sync.Mutex
 	entries         []*netlink.ConntrackFlow
 	netlinkRequests int // try to get the estimated number of netlink request
 }

-var _ Interface = &FakeInterface{}
-
-// NewFake creates a new FakeInterface
-func NewFake() *FakeInterface {
-	return &FakeInterface{entries: make([]*netlink.ConntrackFlow, 0)}
+func (f *fakeHandler) ConntrackTableList(_ netlink.ConntrackTableType, _ netlink.InetFamily) ([]*netlink.ConntrackFlow, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.netlinkRequests++
+	return f.entries, nil
 }

-// ListEntries is part of Interface
-func (fake *FakeInterface) ListEntries(_ uint8) ([]*netlink.ConntrackFlow, error) {
-	entries := make([]*netlink.ConntrackFlow, len(fake.entries))
-	copy(entries, fake.entries)
-	// 1 netlink request to dump the table
-	// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L93-L94
-	fake.netlinkRequests++
-	return entries, nil
-}
+func (f *fakeHandler) ConntrackDeleteFilters(tableType netlink.ConntrackTableType, family netlink.InetFamily, netlinkFilters ...netlink.CustomConntrackFilter) (uint, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()

-// ClearEntries is part of Interface
-func (fake *FakeInterface) ClearEntries(_ uint8, filters ...netlink.CustomConntrackFilter) (int, error) {
-	var flows []*netlink.ConntrackFlow
-	before := len(fake.entries)
 	// 1 netlink request to dump the table
 	// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L163
-	fake.netlinkRequests++
+	f.netlinkRequests++
+	var dataplaneFlows []*netlink.ConntrackFlow
+	before := len(f.entries)

-	for _, flow := range fake.entries {
+	for _, flow := range f.entries {
 		var matched bool
-		for _, filter := range filters {
+		for _, filter := range netlinkFilters {
 			matched = filter.MatchConntrackFlow(flow)
 			if matched {
 				// 1 netlink request to delete the flow
 				// https://github.com/vishvananda/netlink/blob/0af32151e72b990c271ef6268e8aadb7e015f2bd/conntrack_linux.go#L182
-				fake.netlinkRequests++
-				break
+				f.netlinkRequests++
 			}
 		}
+		// no filter matched, keep the flow
 		if !matched {
-			flows = append(flows, flow)
+			dataplaneFlows = append(dataplaneFlows, flow)
 		}
 	}
-	fake.entries = flows
-	return before - len(fake.entries), nil
+	f.entries = dataplaneFlows
+	return uint(before - len(f.entries)), nil
 }
--- a/pkg/proxy/conntrack/filter.go
+++ b/pkg/proxy/conntrack/filter.go
@@ -20,82 +20,50 @@ limitations under the License.
 package conntrack

 import (
-	"net"
+	"fmt"

 	"github.com/vishvananda/netlink"

+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/klog/v2"
 )

-type connectionTuple struct {
-	srcIP   net.IP
-	srcPort uint16
-	dstIP   net.IP
-	dstPort uint16
+type flowFilter struct {
+	flows sets.Set[string]
 }

-type conntrackFilter struct {
-	protocol uint8
-	original *connectionTuple
-	reply    *connectionTuple
+var _ netlink.CustomConntrackFilter = (*flowFilter)(nil)
+
+func newFlowFilter(flows []*netlink.ConntrackFlow) *flowFilter {
+	f := &flowFilter{
+		flows: sets.New[string](),
+	}
+	for _, flow := range flows {
+		f.flows.Insert(flowKey(flow))
+	}
+	return f
 }

-var _ netlink.CustomConntrackFilter = (*conntrackFilter)(nil)
+// flowKey returns a unique key for the given flow.
+func flowKey(flow *netlink.ConntrackFlow) string {
+	return fmt.Sprintf("%d#%s#%d#%s#%d#%s#%d#%s#%d",
+		flow.Forward.Protocol,
+		flow.Forward.SrcIP.String(),
+		flow.Forward.SrcPort,
+		flow.Forward.DstIP.String(),
+		flow.Forward.DstPort,
+		flow.Reverse.SrcIP.String(),
+		flow.Reverse.SrcPort,
+		flow.Reverse.DstIP.String(),
+		flow.Reverse.DstPort)
+}

 // MatchConntrackFlow applies the filter to the flow and returns true if the flow matches the filter
-// false otherwise.
-func (f *conntrackFilter) MatchConntrackFlow(flow *netlink.ConntrackFlow) bool {
-	// return false in case of empty filter
-	if f.protocol == 0 && f.original == nil && f.reply == nil {
-		return false
+func (f *flowFilter) MatchConntrackFlow(flow *netlink.ConntrackFlow) bool {
+	if f.flows.Has(flowKey(flow)) {
+		// for debugging conntrack issues and understanding what entries are deleted
+		klog.V(6).InfoS("Deleting conntrack entry", "flow", flow)
+		return true
 	}
-
-	// -p, --protonum proto [Layer 4 Protocol, eg. 'tcp']
-	if f.protocol != 0 && f.protocol != flow.Forward.Protocol {
-		return false
-	}
-
-	// filter on original direction
-	if f.original != nil {
-		// --orig-src ip  [Source address from original direction]
-		if f.original.srcIP != nil && !f.original.srcIP.Equal(flow.Forward.SrcIP) {
-			return false
-		}
-		// --orig-dst ip  [Destination address from original direction]
-		if f.original.dstIP != nil && !f.original.dstIP.Equal(flow.Forward.DstIP) {
-			return false
-		}
-		// --orig-port-src port [Source port from original direction]
-		if f.original.srcPort != 0 && f.original.srcPort != flow.Forward.SrcPort {
-			return false
-		}
-		// --orig-port-dst port	[Destination port from original direction]
-		if f.original.dstPort != 0 && f.original.dstPort != flow.Forward.DstPort {
-			return false
-		}
-	}
-
-	// filter on reply direction
-	if f.reply != nil {
-		// --reply-src ip  [Source NAT ip]
-		if f.reply.srcIP != nil && !f.reply.srcIP.Equal(flow.Reverse.SrcIP) {
-			return false
-		}
-		// --reply-dst ip [Destination NAT ip]
-		if f.reply.dstIP != nil && !f.reply.dstIP.Equal(flow.Reverse.DstIP) {
-			return false
-		}
-		// --reply-port-src port [Source port from reply direction]
-		if f.reply.srcPort != 0 && f.reply.srcPort != flow.Reverse.SrcPort {
-			return false
-		}
-		// --reply-port-dst port	[Destination port from reply direction]
-		if f.reply.dstPort != 0 && f.reply.dstPort != flow.Reverse.DstPort {
-			return false
-		}
-	}
-
-	// appending a new line to the flow makes klog print multiline log which is easier to debug and understand.
-	klog.V(5).InfoS("Deleting conntrack entry", "flow", flow.String()+"\n")
-	return true
+	return false
 }
--- a/pkg/proxy/conntrack/filter_test.go
+++ b/pkg/proxy/conntrack/filter_test.go
@@ -22,151 +22,154 @@ package conntrack
 import (
 	"testing"

-	"github.com/stretchr/testify/require"
 	"github.com/vishvananda/netlink"
-	"golang.org/x/sys/unix"

 	netutils "k8s.io/utils/net"
 )

-func applyFilter(flowList []netlink.ConntrackFlow, ipv4Filter *conntrackFilter, ipv6Filter *conntrackFilter) (ipv4Match, ipv6Match int) {
-	for _, flow := range flowList {
-		if ipv4Filter.MatchConntrackFlow(&flow) == true {
-			ipv4Match++
-		}
-		if ipv6Filter.MatchConntrackFlow(&flow) == true {
-			ipv6Match++
-		}
+func TestFlowFilter(t *testing.T) {
+	flow1 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("1.1.1.1"),
+			DstIP:    netutils.ParseIPSloppy("2.2.2.2"),
+			SrcPort:  1000,
+			DstPort:  2000,
+			Protocol: 6,
+		},
+	}
+	flow2 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("3.3.3.3"),
+			DstIP:    netutils.ParseIPSloppy("4.4.4.4"),
+			SrcPort:  3000,
+			DstPort:  4000,
+			Protocol: 17,
+		},
+	}
+	flowEmpty := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			Protocol: 6,
+		},
 	}
-	return ipv4Match, ipv6Match
-}
-
-func TestConntrackFilter(t *testing.T) {
-	var flowList []netlink.ConntrackFlow
-	flow1 := netlink.ConntrackFlow{}
-	flow1.FamilyType = unix.AF_INET
-	flow1.Forward.SrcIP = netutils.ParseIPSloppy("10.0.0.1")
-	flow1.Forward.DstIP = netutils.ParseIPSloppy("20.0.0.1")
-	flow1.Forward.SrcPort = 1000
-	flow1.Forward.DstPort = 2000
-	flow1.Forward.Protocol = 17
-	flow1.Reverse.SrcIP = netutils.ParseIPSloppy("20.0.0.1")
-	flow1.Reverse.DstIP = netutils.ParseIPSloppy("192.168.1.1")
-	flow1.Reverse.SrcPort = 2000
-	flow1.Reverse.DstPort = 1000
-	flow1.Reverse.Protocol = 17
-
-	flow2 := netlink.ConntrackFlow{}
-	flow2.FamilyType = unix.AF_INET
-	flow2.Forward.SrcIP = netutils.ParseIPSloppy("10.0.0.2")
-	flow2.Forward.DstIP = netutils.ParseIPSloppy("20.0.0.2")
-	flow2.Forward.SrcPort = 5000
-	flow2.Forward.DstPort = 6000
-	flow2.Forward.Protocol = 6
-	flow2.Reverse.SrcIP = netutils.ParseIPSloppy("20.0.0.2")
-	flow2.Reverse.DstIP = netutils.ParseIPSloppy("192.168.1.1")
-	flow2.Reverse.SrcPort = 6000
-	flow2.Reverse.DstPort = 5000
-	flow2.Reverse.Protocol = 6
-
-	flow3 := netlink.ConntrackFlow{}
-	flow3.FamilyType = unix.AF_INET6
-	flow3.Forward.SrcIP = netutils.ParseIPSloppy("eeee:eeee:eeee:eeee:eeee:eeee:eeee:eeee")
-	flow3.Forward.DstIP = netutils.ParseIPSloppy("dddd:dddd:dddd:dddd:dddd:dddd:dddd:dddd")
-	flow3.Forward.SrcPort = 1000
-	flow3.Forward.DstPort = 2000
-	flow3.Forward.Protocol = 132
-	flow3.Reverse.SrcIP = netutils.ParseIPSloppy("dddd:dddd:dddd:dddd:dddd:dddd:dddd:dddd")
-	flow3.Reverse.DstIP = netutils.ParseIPSloppy("eeee:eeee:eeee:eeee:eeee:eeee:eeee:eeee")
-	flow3.Reverse.SrcPort = 2000
-	flow3.Reverse.DstPort = 1000
-	flow3.Reverse.Protocol = 132
-	flowList = append(flowList, flow1, flow2, flow3)

 	testCases := []struct {
-		name              string
-		filterV4          *conntrackFilter
-		filterV6          *conntrackFilter
-		expectedV4Matches int
-		expectedV6Matches int
+		name          string
+		filterFlows   []*netlink.ConntrackFlow
+		matchFlow     *netlink.ConntrackFlow
+		expectedMatch bool
 	}{
 		{
-			name:              "Empty filter",
-			filterV4:          &conntrackFilter{},
-			filterV6:          &conntrackFilter{},
-			expectedV4Matches: 0,
-			expectedV6Matches: 0,
+			name:          "Match existing flow",
+			filterFlows:   []*netlink.ConntrackFlow{flow1, flow2},
+			matchFlow:     flow1,
+			expectedMatch: true,
 		},
 		{
-			name:              "Protocol filter",
-			filterV4:          &conntrackFilter{protocol: 6},
-			filterV6:          &conntrackFilter{protocol: 17},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
+			name:          "No match for non-existent flow",
+			filterFlows:   []*netlink.ConntrackFlow{flow1},
+			matchFlow:     flow2,
+			expectedMatch: false,
 		},
 		{
-			name:              "Original Source IP filter",
-			filterV4:          &conntrackFilter{original: &connectionTuple{srcIP: netutils.ParseIPSloppy("10.0.0.1")}},
-			filterV6:          &conntrackFilter{original: &connectionTuple{srcIP: netutils.ParseIPSloppy("eeee:eeee:eeee:eeee:eeee:eeee:eeee:eeee")}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Original Destination IP filter",
-			filterV4:          &conntrackFilter{original: &connectionTuple{dstIP: netutils.ParseIPSloppy("20.0.0.1")}},
-			filterV6:          &conntrackFilter{original: &connectionTuple{dstIP: netutils.ParseIPSloppy("dddd:dddd:dddd:dddd:dddd:dddd:dddd:dddd")}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Original Source Port Filter",
-			filterV4:          &conntrackFilter{protocol: 6, original: &connectionTuple{srcPort: 5000}},
-			filterV6:          &conntrackFilter{protocol: 132, original: &connectionTuple{srcPort: 1000}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Original Destination Port Filter",
-			filterV4:          &conntrackFilter{protocol: 6, original: &connectionTuple{dstPort: 6000}},
-			filterV6:          &conntrackFilter{protocol: 132, original: &connectionTuple{dstPort: 2000}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Reply Source IP filter",
-			filterV4:          &conntrackFilter{reply: &connectionTuple{srcIP: netutils.ParseIPSloppy("20.0.0.1")}},
-			filterV6:          &conntrackFilter{reply: &connectionTuple{srcIP: netutils.ParseIPSloppy("dddd:dddd:dddd:dddd:dddd:dddd:dddd:dddd")}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Reply Destination IP filter",
-			filterV4:          &conntrackFilter{reply: &connectionTuple{dstIP: netutils.ParseIPSloppy("192.168.1.1")}},
-			filterV6:          &conntrackFilter{reply: &connectionTuple{dstIP: netutils.ParseIPSloppy("dddd:dddd:dddd:dddd:dddd:dddd:dddd:dddd")}},
-			expectedV4Matches: 2,
-			expectedV6Matches: 0,
-		},
-		{
-			name:              "Reply Source Port filter",
-			filterV4:          &conntrackFilter{protocol: 17, reply: &connectionTuple{srcPort: 2000}},
-			filterV6:          &conntrackFilter{protocol: 132, reply: &connectionTuple{srcPort: 2000}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
-		},
-		{
-			name:              "Reply Destination Port filter",
-			filterV4:          &conntrackFilter{protocol: 6, reply: &connectionTuple{dstPort: 5000}},
-			filterV6:          &conntrackFilter{protocol: 132, reply: &connectionTuple{dstPort: 1000}},
-			expectedV4Matches: 1,
-			expectedV6Matches: 1,
+			name:          "No match for empty flow against specific filter",
+			filterFlows:   []*netlink.ConntrackFlow{flow1},
+			matchFlow:     flowEmpty,
+			expectedMatch: false,
 		},
 	}

 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			v4Matches, v6Matches := applyFilter(flowList, tc.filterV4, tc.filterV6)
-			require.Equal(t, tc.expectedV4Matches, v4Matches)
-			require.Equal(t, tc.expectedV6Matches, v6Matches)
+			f := newFlowFilter(tc.filterFlows)
+			match := f.MatchConntrackFlow(tc.matchFlow)
+			if match != tc.expectedMatch {
+				t.Errorf("MatchConntrackFlow() = %v, want %v", match, tc.expectedMatch)
+			}
+		})
+	}
+}
+
+func TestFlowFilter_Exhaustive(t *testing.T) {
+	flow1 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("1.1.1.1"),
+			DstIP:    netutils.ParseIPSloppy("2.2.2.2"),
+			SrcPort:  1000,
+			DstPort:  2000,
+			Protocol: 6,
+		},
+		Reverse: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("2.2.2.2"),
+			DstIP:    netutils.ParseIPSloppy("1.1.1.1"),
+			SrcPort:  2000,
+			DstPort:  1000,
+			Protocol: 6,
+		},
+	}
+	flow2 := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("3.3.3.3"),
+			DstIP:    netutils.ParseIPSloppy("4.4.4.4"),
+			SrcPort:  3000,
+			DstPort:  4000,
+			Protocol: 17,
+		},
+		Reverse: netlink.IPTuple{
+			SrcIP:    netutils.ParseIPSloppy("4.4.4.4"),
+			DstIP:    netutils.ParseIPSloppy("3.3.3.3"),
+			SrcPort:  4000,
+			DstPort:  3000,
+			Protocol: 17,
+		},
+	}
+	flowEmpty := &netlink.ConntrackFlow{
+		Forward: netlink.IPTuple{
+			Protocol: 6,
+		},
+		Reverse: netlink.IPTuple{
+			Protocol: 6,
+		},
+	}
+
+	testCases := []struct {
+		name          string
+		filterFlows   []*netlink.ConntrackFlow
+		matchFlow     *netlink.ConntrackFlow
+		expectedMatch bool
+	}{
+		{
+			name:          "Match existing flow",
+			filterFlows:   []*netlink.ConntrackFlow{flow1, flow2},
+			matchFlow:     flow1,
+			expectedMatch: true,
+		},
+		{
+			name:          "No match for non-existent flow",
+			filterFlows:   []*netlink.ConntrackFlow{flow1},
+			matchFlow:     flow2,
+			expectedMatch: false,
+		},
+		{
+			name:          "No match for empty flow against specific filter",
+			filterFlows:   []*netlink.ConntrackFlow{flow1},
+			matchFlow:     flowEmpty,
+			expectedMatch: false,
+		},
+		{
+			name:          "Match empty flow if filter has empty flow",
+			filterFlows:   []*netlink.ConntrackFlow{flowEmpty},
+			matchFlow:     flowEmpty,
+			expectedMatch: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			f := newFlowFilter(tc.filterFlows)
+			match := f.MatchConntrackFlow(tc.matchFlow)
+			if match != tc.expectedMatch {
+				t.Errorf("MatchConntrackFlow() = %v, want %v", match, tc.expectedMatch)
+			}
 		})
 	}
 }