From 38e08c231c89898c623a6860622747afd9eee63d Mon Sep 17 00:00:00 2001 From: Antonio Ojea Date: Sat, 29 Nov 2025 17:30:47 +0000 Subject: [PATCH] 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). --- pkg/proxy/conntrack/cleanup.go | 52 +----- pkg/proxy/conntrack/cleanup_test.go | 130 +++---------- pkg/proxy/conntrack/conntrack.go | 18 +- pkg/proxy/conntrack/conntrack_test.go | 181 ++++++++++++------ pkg/proxy/conntrack/fake.go | 61 +++--- pkg/proxy/conntrack/filter.go | 98 ++++------ pkg/proxy/conntrack/filter_test.go | 255 +++++++++++++------------- 7 files changed, 356 insertions(+), 439 deletions(-) diff --git a/pkg/proxy/conntrack/cleanup.go b/pkg/proxy/conntrack/cleanup.go index 95e82317d26..4953adf7f30 100644 --- 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, - }, - } -} diff --git a/pkg/proxy/conntrack/cleanup_test.go b/pkg/proxy/conntrack/cleanup_test.go index 369bc35a8af..93e6eca157a 100644 --- 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)) - }) - } -} diff --git a/pkg/proxy/conntrack/conntrack.go b/pkg/proxy/conntrack/conntrack.go index 53cf6ba57f3..bf41a050db5 100644 --- 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 }) diff --git a/pkg/proxy/conntrack/conntrack_test.go b/pkg/proxy/conntrack/conntrack_test.go index 98f005a186c..ff20a02da83 100644 --- 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) } }) } diff --git a/pkg/proxy/conntrack/fake.go b/pkg/proxy/conntrack/fake.go index 8902da7a6f2..4183aa5bb3a 100644 --- 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 } diff --git a/pkg/proxy/conntrack/filter.go b/pkg/proxy/conntrack/filter.go index 9bb20ffd5a8..95ddfb9c3dc 100644 --- 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 } diff --git a/pkg/proxy/conntrack/filter_test.go b/pkg/proxy/conntrack/filter_test.go index a8b8a4b3f87..400e8d56c9a 100644 --- 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) + } }) } }