From 7274f324e1108d0feaeed5fd7935b2489da7651c Mon Sep 17 00:00:00 2001 From: Tomas Hruby <49207409+tomastigera@users.noreply.github.com> Date: Fri, 11 Sep 2026 09:21:29 -0700 Subject: [PATCH] Merge pull request #13778 from tomastigera/tomas/bpf-svc-migration-fv [BPF] Keep pre-existing forwarded flows alive when switching to eBPF (cherry picked from commit d803bb953406d1fe42b3e4b6bce7726ec7555c1b) --- felix/dataplane/linux/int_dataplane.go | 12 ++ felix/fv/bpf_test.go | 172 +++++++++++++++++++++++++ 2 files changed, 184 insertions(+) diff --git a/felix/dataplane/linux/int_dataplane.go b/felix/dataplane/linux/int_dataplane.go index 4ef2e281944..10728d16017 100644 --- a/felix/dataplane/linux/int_dataplane.go +++ b/felix/dataplane/linux/int_dataplane.go @@ -2250,6 +2250,18 @@ func (d *InternalDataplane) setUpIptablesBPF() { Comment: []string{"From ", dataplanedefs.BPFOutDev, " device, mark verified, accept."}, }, ) + + // Forwarded between two host interfaces, so matched by none of the accepts + // above. Linux conntrack vetted it, the same signal INPUT trusts. + fwdRules = append(fwdRules, + generictables.Rule{ + Match: d.newMatch(). + MarkMatchesWithMask(tcdefs.MarkSeenFallThrough, tcdefs.MarkSeenFallThroughMask). + ConntrackState("ESTABLISHED,RELATED"), + Action: d.actions.Allow(), + Comment: []string{"Accept forwarded packets from flows that pre-date BPF."}, + }, + ) } t.InsertOrAppendRules("INPUT", inputRules) diff --git a/felix/fv/bpf_test.go b/felix/fv/bpf_test.go index b317d5ab994..5d11f13fc99 100644 --- a/felix/fv/bpf_test.go +++ b/felix/fv/bpf_test.go @@ -5574,6 +5574,178 @@ func describeBPFTests(opts ...bpfTestOpt) bool { It("should keep a connection up between hosts and local workloads when BPF is enabled", func() { verifyConnectivityWhileEnablingBPF(hostW[0], w[0][0]) }) + + // Connections through a service must survive the switch too. kube-proxy + // does not run in the FV, so we install the DNAT it would have written, + // in the chains Felix deletes on the switch. The connection has to + // outlive that deletion: the DNAT binding lives in the Linux conntrack + // entry, not in the rule. + // + // Pinned to one matrix point to keep the cost down; these flows pre-date + // BPF so they never take the DSR or tunnel path. The addresses below are + // v4, hence the explicit IPv6 exclusion. + if testOpts.dsr && !testOpts.ipv6 && testOpts.tunnel == "none" { + const ( + migrationSvcIP = "10.101.0.99" + migrationSvcPort = 8090 + ) + + installKubeProxyService := func(felix *infrastructure.Felix, backend *workload.Workload) { + target := net.JoinHostPort(backend.IP, backend.Ports) + if NFTMode() { + felix.Exec("nft", "add", "table", "ip", "kube-proxy") + felix.Exec("nft", "add", "chain", "ip", "kube-proxy", "services", + "{ type nat hook prerouting priority dstnat ; }") + felix.Exec("nft", "add", "rule", "ip", "kube-proxy", "services", + "ip", "daddr", migrationSvcIP, "tcp", "dport", fmt.Sprint(migrationSvcPort), + "dnat", "to", target) + return + } + felix.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", + "-N", "KUBE-SERVICES") + felix.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", + "-A", "KUBE-SERVICES", "-d", migrationSvcIP, "-p", "tcp", + "--dport", fmt.Sprint(migrationSvcPort), + "-j", "DNAT", "--to-destination", target) + felix.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", + "-I", "PREROUTING", "-j", "KUBE-SERVICES") + } + + // Marker that Felix's kube-proxy cleanup takes away, and the + // command that shows whether it is still there. + kubeProxyMarker := "KUBE-SERVICES" + dumpRulesCmd := []string{"iptables-save", "-t", "nat"} + if NFTMode() { + kubeProxyMarker = "kube-proxy" + dumpRulesCmd = []string{"nft", "list", "ruleset"} + } + verifySvcConnectivityWhileEnablingBPF := func(client, backend *workload.Workload) { + By("Creating the service") + // Must agree with the DNAT target below, which uses the same port. + backendPort, err := strconv.Atoi(backend.Ports) + Expect(err).NotTo(HaveOccurred()) + testSvc := k8sService("migration-svc", migrationSvcIP, backend, + migrationSvcPort, backendPort, 0, testOpts.protocol) + k8sClient := infra.(*infrastructure.K8sDatastoreInfra).K8sClient + _, err = k8sClient.CoreV1().Services(testSvc.Namespace).Create( + context.Background(), testSvc, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + Eventually(checkSvcEndpoints(k8sClient, testSvc), "10s").Should(Equal(1), + "Service endpoints didn't get created? Is controller-manager happy?") + + By("Installing the rules kube-proxy would have written") + installKubeProxyService(tc.Felixes[0], backend) + // Also proves the dump command works, so that the check for + // their removal below cannot pass vacuously. + Expect(tc.Felixes[0].ExecOutputFn(dumpRulesCmd...)()). + To(ContainSubstring(kubeProxyMarker)) + + By("Starting persistent connection via the service") + pc = client.StartPersistentConnection(migrationSvcIP, migrationSvcPort, + workload.PersistentConnectionOpts{ + MonitorConnectivity: true, + Timeout: 60 * time.Second, + }) + + By("having initial connectivity", expectPongs) + By("enabling BPF mode", enableBPF) // Waits for BPF programs to be installed + By("removing the kube-proxy rules", func() { + Eventually(tc.Felixes[0].ExecOutputFn(dumpRulesCmd...), "30s", "1s"). + ShouldNot(ContainSubstring(kubeProxyMarker)) + }) + By("still having connectivity on the existing connection", expectPongs) + + By("having connectivity on a new connection via the service", func() { + cc.ResetExpectations() + cc.Expect(Some, client, TargetIP(migrationSvcIP), + ExpectWithPorts(uint16(migrationSvcPort))) + cc.CheckConnectivity() + cc.ResetExpectations() + }) + } + + It("should keep a connection to a service with a local backend up when BPF is enabled", func() { + verifySvcConnectivityWhileEnablingBPF(w[0][1], w[0][0]) + }) + + It("should keep a connection to a service with a remote backend up when BPF is enabled", func() { + verifySvcConnectivityWhileEnablingBPF(w[0][1], w[1][0]) + }) + + // External client -> NodePort. With a backend on another node the + // forwarded legs enter and leave on the same host interface, which is + // the case that matched none of the BPF-mode FORWARD accepts. + verifyNodePortWhileEnablingBPF := func(backend *workload.Workload) { + const ( + npSvcIP = "10.101.0.98" + nodePort = 30333 + ) + backendPort, err := strconv.Atoi(backend.Ports) + Expect(err).NotTo(HaveOccurred()) + + By("Creating the NodePort service") + testSvc := k8sService("migration-np", npSvcIP, backend, + migrationSvcPort, backendPort, int32(nodePort), testOpts.protocol) + k8sClient := infra.(*infrastructure.K8sDatastoreInfra).K8sClient + _, err = k8sClient.CoreV1().Services(testSvc.Namespace).Create( + context.Background(), testSvc, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) + Eventually(checkSvcEndpoints(k8sClient, testSvc), "10s").Should(Equal(1)) + + // The masquerade mimics kube-proxy, which SNATs node port traffic so + // that replies come back via the ingress node. Only the iptables run + // reproduces the regression: nftables mode sets the FORWARD policy to + // ACCEPT (see infrastructure.Felix), so nothing drops the packet there. + By("Installing the NodePort rules kube-proxy would have written") + f := tc.Felixes[0] + target := net.JoinHostPort(backend.IP, backend.Ports) + if NFTMode() { + f.Exec("nft", "add", "table", "ip", "kube-proxy") + f.Exec("nft", "add", "chain", "ip", "kube-proxy", "nodeports", + "{ type nat hook prerouting priority dstnat ; }") + f.Exec("nft", "add", "rule", "ip", "kube-proxy", "nodeports", + "fib", "daddr", "type", "local", "tcp", "dport", fmt.Sprint(nodePort), + "dnat", "to", target) + f.Exec("nft", "add", "chain", "ip", "kube-proxy", "masq", + "{ type nat hook postrouting priority srcnat ; }") + f.Exec("nft", "add", "rule", "ip", "kube-proxy", "masq", + "ip", "daddr", backend.IP, "tcp", "dport", backend.Ports, "masquerade") + } else { + f.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", "-N", "KUBE-SERVICES") + f.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", "-A", "KUBE-SERVICES", + "-m", "addrtype", "--dst-type", "LOCAL", "-p", "tcp", + "--dport", fmt.Sprint(nodePort), "-j", "DNAT", "--to-destination", target) + f.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", + "-I", "PREROUTING", "-j", "KUBE-SERVICES") + f.Exec("iptables", "-w", "10", "-W", "100000", "-t", "nat", "-A", "POSTROUTING", + "-d", backend.IP, "-p", "tcp", "--dport", backend.Ports, "-j", "MASQUERADE") + } + + By("Starting persistent connection from the external client to the node port") + pc = &PersistentConnection{ + Runtime: externalClient, + RuntimeName: externalClient.Name, + IP: felixIP(0), + Port: nodePort, + Protocol: testOpts.protocol, + MonitorConnectivity: true, + Timeout: 60 * time.Second, + } + Expect(pc.Start()).NotTo(HaveOccurred()) + + By("having initial connectivity", expectPongs) + By("enabling BPF mode", enableBPF) + By("still having connectivity on the existing connection", expectPongs) + } + + It("should keep an external nodeport connection with a local backend up when BPF is enabled", func() { + verifyNodePortWhileEnablingBPF(w[0][0]) + }) + + It("should keep an external nodeport connection with a remote backend up when BPF is enabled", func() { + verifyNodePortWhileEnablingBPF(w[1][0]) + }) + } } })