From fd81afe040bae6e976f1dc843cdf3c96482471b3 Mon Sep 17 00:00:00 2001 From: Baf <26187677+Bafff@users.noreply.github.com> Date: Wed, 10 Jun 2026 20:36:33 +0100 Subject: [PATCH] kube-proxy: clear stale conntrack entries for UDP services with no endpoints The conntrack reconciler skips services without serving endpoints, so conntrack entries established while endpoints existed are never removed when a UDP service scales down to zero. The REJECT (iptables) / reject (nftables) rule installed for such services does not cover those flows: they are DNATed to the deleted endpoint IP before the rule, which matches on the service IP, can be evaluated. One-way UDP senders (e.g. statsd clients) refresh the 30s conntrack timeout with every packet, so the stale flows blackhole traffic to the deleted pod IP indefinitely; recovery only happens when the service gets an endpoint again. This was handled before the reconciler rewrite (kubernetes#127318): the event-based cleanup cleared entries for every deleted UDP endpoint regardless of how many endpoints remained. Process services with an empty endpoints set instead of skipping them, so every entry directed to their ClusterIP, LoadBalancer IP and ExternalIP frontends is treated as stale and deleted. NodePort cleanup is still skipped for services without serving endpoints: NodePort entries are matched on the destination port only, and with an empty endpoints set that would also remove UDP flows not owned by kube-proxy (e.g. traffic to an unrelated host on the same port). --- pkg/proxy/conntrack/cleanup.go | 22 ++++++++---- pkg/proxy/conntrack/cleanup_test.go | 53 +++++++++++++++++++++++++---- 2 files changed, 63 insertions(+), 12 deletions(-) diff --git a/pkg/proxy/conntrack/cleanup.go b/pkg/proxy/conntrack/cleanup.go index 8035fcbdb90..f04bba05938 100644 --- a/pkg/proxy/conntrack/cleanup.go +++ b/pkg/proxy/conntrack/cleanup.go @@ -86,10 +86,15 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily, } } - // a Service without endpoints does not require to clean the conntrack entries associated. - if endpoints.Len() == 0 { - continue - } + // Note: a Service without any serving endpoints is processed too, with an + // empty endpoints set, so that all the existing entries directed to its + // frontends are removed. The REJECT (iptables) / reject (nftables) rule + // installed for a Service with no endpoints does not cover the previously + // established flows: those are DNATed to the (deleted) endpoint IP before + // the reject rule, which matches on the Service IP, can be evaluated. + // One-way UDP flows (e.g. statsd) refresh the conntrack entry timeout with + // every packet, so without this cleanup they keep sending traffic to the + // deleted endpoint IP indefinitely. // we need to filter entries that are directed to a Service IP:Port frontend // that does not have an Endpoint IP:Port backend as part of the serving endpoints @@ -105,8 +110,13 @@ func CleanStaleEntries(ct Interface, ipFamily v1.IPFamily, serviceIPEndpoints[net.JoinHostPort(externalIP.String(), portStr)] = endpoints } // we need to filter entries that are directed to a *:NodePort that does not have - // an Endpoint IP:Port backend as part of the serving endpoints - if svc.NodePort() != 0 { + // an Endpoint IP:Port backend as part of the serving endpoints. + // NodePort entries are matched on the destination port only, so with an + // empty endpoints set every UDP flow towards that port number would be + // removed, including flows not owned by kube-proxy (e.g. traffic to an + // unrelated host on the same port). Skip NodePort cleanup for services + // without serving endpoints until the match can be restricted to node IPs. + if svc.NodePort() != 0 && endpoints.Len() > 0 { // *:NodePort serviceNodePortEndpoints[svc.NodePort()] = endpoints } diff --git a/pkg/proxy/conntrack/cleanup_test.go b/pkg/proxy/conntrack/cleanup_test.go index 98afa791064..1213dc25a2d 100644 --- a/pkg/proxy/conntrack/cleanup_test.go +++ b/pkg/proxy/conntrack/cleanup_test.go @@ -458,6 +458,7 @@ func TestServiceWithoutEndpoints(t *testing.T) { { Name: "test-udp", Port: testServicePort, + NodePort: testServiceNodePort, Protocol: v1.ProtocolUDP, }, }, @@ -508,10 +509,38 @@ func TestServiceWithoutEndpoints(t *testing.T) { _ = endpointsMap.Update(ect) flows := []*netlink.ConntrackFlow{} - // 1 valid entry - flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testServingEndpointIP, testEndpointPort, unix.IPPROTO_UDP)) - // 1 stale entry - flows = append(flows, generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP)) + var entriesAfterCleanup []*netlink.ConntrackFlow + + // All UDP entries directed to the frontends of a service without serving + // endpoints are stale and must be cleared, regardless of which (deleted) + // endpoint they are DNATed to: established flows bypass the REJECT rule + // installed for services with no endpoints and would otherwise keep + // blackholing traffic to the deleted endpoint IP indefinitely. + for _, origDest := range []string{testClusterIP, testLoadBalancerIP, testExternalIP} { + for _, dnatDest := range []string{testServingEndpointIP, testDeletedEndpointIP} { + flows = append(flows, generateConntrackEntry(origDest, testServicePort, dnatDest, testEndpointPort, unix.IPPROTO_UDP)) + } + } + + // UDP entries not directed to a service frontend are not owned by + // kube-proxy and must be preserved. + entry := generateConntrackEntry(testExternalIP, testNonServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP) + flows = append(flows, entry) + entriesAfterCleanup = append(entriesAfterCleanup, entry) + + // non-UDP entries must be preserved. + entry = generateConntrackEntry(testExternalIP, testServicePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_TCP) + flows = append(flows, entry) + entriesAfterCleanup = append(entriesAfterCleanup, entry) + + // *:NodePort entries are matched on the destination port only, which with an + // empty endpoints set would also remove flows not owned by kube-proxy, so + // NodePort cleanup is skipped for services without serving endpoints and + // these entries must be preserved. + entry = generateConntrackEntry("", testServiceNodePort, testDeletedEndpointIP, testEndpointPort, unix.IPPROTO_UDP) + flows = append(flows, entry) + entriesAfterCleanup = append(entriesAfterCleanup, entry) + ct := newConntracker( &fakeHandler{ entries: flows, @@ -520,7 +549,19 @@ func TestServiceWithoutEndpoints(t *testing.T) { 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) + + require.Len(t, actualEntries, len(entriesAfterCleanup)) + + // sort the actual flows before comparison + sort.Slice(actualEntries, func(i, j int) bool { + return actualEntries[i].String() < actualEntries[j].String() + }) + // sort the expected flows before comparison + sort.Slice(entriesAfterCleanup, func(i, j int) bool { + return entriesAfterCleanup[i].String() < entriesAfterCleanup[j].String() + }) + + if diff := cmp.Diff(entriesAfterCleanup, actualEntries); len(diff) > 0 { + t.Errorf("unexpected entries after cleanup: %s", diff) } }