From 049ee94981757b366e85de6c1f9690896c2163d0 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Fri, 7 Aug 2026 17:44:42 +0200 Subject: [PATCH 1/2] proxy: scope datapath rules to the node hosting the backend Every node programmed the svc_pod/pod_svc maps and the port-filter sets for every service, so a node that did not host the backend still rewrote the destination of a packet leaving it. That premature DNAT breaks conntrack on the owning node. It records the flow as (srcSvcIP -> podIP) because the destination was already translated upstream, while the reply leaves the pod and gets its source rewritten to the service IP by egress_snat, which runs at prerouting priority raw, before conntrack. The tuple (dstSvcIP -> srcSvcIP) matches nothing, the reply is not established, and port_filter drops it since the initiator's ephemeral port is not in its own allowed_ports. The visible effect is that a PortList (wholeIP=false) backend can no longer open a connection to another cozy-proxy managed backend on a different node: the SYN arrives, the SYN-ACK is generated and dropped, and the caller hangs. Traffic from outside the cluster is unaffected, because it is not translated before reaching the owning node. Program the rules only where the backend pod runs, keyed on the endpoint's NodeName against NODE_NAME. Every node then sees a consistent conntrack view, and the maps only carry local entries. Rules for a pod that moved away are withdrawn, both on endpoint events and by the startup cleanup, so state inherited from a cluster-wide build is purged on upgrade. NODE_NAME is read from the environment; when it is absent the check is disabled and the previous cluster-wide behavior is kept, so the binary still runs under a chart that does not inject it yet. Signed-off-by: Mattia Eleuteri --- main.go | 11 ++ pkg/controllers/services_controller.go | 114 +++++++++++++++--- pkg/controllers/services_controller_test.go | 127 ++++++++++++++++++++ 3 files changed, 233 insertions(+), 19 deletions(-) diff --git a/main.go b/main.go index 5d6f9dd..e870daa 100644 --- a/main.go +++ b/main.go @@ -68,9 +68,20 @@ func main() { os.Exit(1) } + // Datapath rules are programmed only for backend pods hosted on this node. + // When NODE_NAME is absent the check is disabled and every service is + // programmed, which is the pre-node-local behavior: degraded, but it keeps + // a new binary working under a chart that does not inject the variable yet. + nodeName := os.Getenv("NODE_NAME") + if nodeName == "" { + log.Info("NODE_NAME is not set, falling back to programming rules for every node's backends; " + + "set it from spec.nodeName to scope rules to this node") + } + controller := &controllers.ServicesController{ Clientset: clientset, Proxy: &proxy.NFTProxyProcessor{}, + NodeName: nodeName, } if err := mgr.Add(controller); err != nil { diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index cb00a46..88e174a 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -96,6 +96,83 @@ type ServicesController struct { Clientset *kubernetes.Clientset Services *ServiceMap Proxy nat.ProxyProcessor + + // NodeName is the node this instance runs on. Datapath rules are only + // programmed for backend pods hosted here. Empty disables the check and + // restores the previous cluster-wide behavior. + NodeName string +} + +// endpointNode returns the node hosting the endpoint's first address. +func endpointNode(ep *v1.Endpoints) (string, bool) { + if !hasValidEndpointIP(ep) { + return "", false + } + node := ep.Subsets[0].Addresses[0].NodeName + if node == nil || *node == "" { + return "", false + } + return *node, true +} + +// servesEndpoint reports whether this node must program datapath rules for ep. +// +// The rules are node-local: only the node hosting the backend pod may rewrite +// the service IP. Programming them cluster-wide makes a non-owning node +// translate the destination before the packet even leaves it, so the owning +// node records a conntrack tuple the SNATed reply can no longer match, and +// port_filter drops that reply. +// +// When the node name is unknown (NODE_NAME not injected, or an endpoint +// carrying no NodeName) the rules are programmed anyway, so an older chart +// keeps the previous behavior instead of silently losing the datapath. +func (c *ServicesController) servesEndpoint(ep *v1.Endpoints) bool { + if c.NodeName == "" { + return true + } + node, ok := endpointNode(ep) + if !ok { + return true + } + return node == c.NodeName +} + +// applyRules programs the datapath for a (service, endpoint) pair when this +// node hosts the backend pod, and withdraws it otherwise. Both objects must +// already have been checked with hasValidServiceIP/hasValidEndpointIP. +func (c *ServicesController) applyRules(svc *v1.Service, ep *v1.Endpoints, ctx string) { + svcIP := svc.Status.LoadBalancer.Ingress[0].IP + podIP := ep.Subsets[0].Addresses[0].IP + + if !c.servesEndpoint(ep) { + c.withdrawRules(svcIP, podIP, ctx+" (backend not on this node)") + return + } + + c.Proxy.EnsureRules(svcIP, podIP) + c.reconcilePortFilter(svc, svcIP, podIP, ctx) +} + +// withdrawRules removes every datapath entry for the pair. Absent entries are +// not an error. +func (c *ServicesController) withdrawRules(svcIP, podIP, ctx string) { + c.clearPortFilter(svcIP, podIP, ctx) + c.Proxy.DeleteRules(svcIP, podIP) +} + +// withdrawStaleEndpoint drops the rules of a previous endpoint whose pod IP no +// longer matches the current one, which is what happens when a VM is migrated +// to another node. Without this the old node keeps a mapping for a pod it no +// longer hosts. +func (c *ServicesController) withdrawStaleEndpoint(svc *v1.Service, prev *v1.Endpoints, podIP, ctx string) { + if !hasValidServiceIP(svc) || !hasValidEndpointIP(prev) { + return + } + prevPodIP := prev.Subsets[0].Addresses[0].IP + if prevPodIP == podIP { + return + } + c.withdrawRules(svc.Status.LoadBalancer.Ingress[0].IP, prevPodIP, ctx+" (stale endpoint)") } // Start initializes the NAT, runs the service and endpoint informers, and cleans up removed services. @@ -231,11 +308,7 @@ func (c *ServicesController) addServiceFunc(obj interface{}) { if err == nil && ep != nil && hasValidEndpointIP(ep) && hasValidServiceIP(svc) { se.Endpoint = ep c.Services.Set(svc.Namespace, svc.Name, se) - svcIP := svc.Status.LoadBalancer.Ingress[0].IP - podIP := ep.Subsets[0].Addresses[0].IP - // Ensure NAT mapping rules are set. - c.Proxy.EnsureRules(svcIP, podIP) - c.reconcilePortFilter(svc, svcIP, podIP, "on svc add") + c.applyRules(svc, ep, "on svc add") } } @@ -330,10 +403,10 @@ func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { // At this point, both the Service and Endpoint have valid IPs. // Ensure NAT mapping is up-to-date. - svcIP := svc.Status.LoadBalancer.Ingress[0].IP - podIP := ep.Subsets[0].Addresses[0].IP - c.Proxy.EnsureRules(svcIP, podIP) - c.reconcilePortFilter(svc, svcIP, podIP, "on svc update") + if se, exists := c.Services.Get(svc.Namespace, svc.Name); exists { + c.withdrawStaleEndpoint(svc, se.Endpoint, ep.Subsets[0].Addresses[0].IP, "on svc update") + } + c.applyRules(svc, ep, "on svc update") // Update or add the service mapping with the new endpoint. c.Services.Set(svc.Namespace, svc.Name, &ServiceEndpoints{Service: svc, Endpoint: ep}) @@ -360,10 +433,8 @@ func (c *ServicesController) addEndpointFunc(obj interface{}) { // If both the Service and the Endpoint have valid IPs, ensure NAT mapping rules. if hasValidServiceIP(se.Service) && hasValidEndpointIP(ep) { - svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP - podIP := ep.Subsets[0].Addresses[0].IP - c.Proxy.EnsureRules(svcIP, podIP) - c.reconcilePortFilter(se.Service, svcIP, podIP, "on endpoint add") + c.withdrawStaleEndpoint(se.Service, se.Endpoint, ep.Subsets[0].Addresses[0].IP, "on endpoint add") + c.applyRules(se.Service, ep, "on endpoint add") } } @@ -420,10 +491,8 @@ func (c *ServicesController) updateEndpointFunc(oldObj, newObj interface{}) { if !hasValidEndpointIP(ep) { return } - svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP - podIP := ep.Subsets[0].Addresses[0].IP - c.Proxy.EnsureRules(svcIP, podIP) - c.reconcilePortFilter(se.Service, svcIP, podIP, "on endpoint update") + c.withdrawStaleEndpoint(se.Service, se.Endpoint, ep.Subsets[0].Addresses[0].IP, "on endpoint update") + c.applyRules(se.Service, ep, "on endpoint update") c.Services.SetEndpoint(ep.Namespace, ep.Name, ep) } @@ -555,6 +624,13 @@ func (c *ServicesController) cleanupRemovedServices() error { if serviceEndpoints.Service != nil && serviceEndpoints.Endpoint != nil { var serviceIP, endpointIP string + // Backends hosted elsewhere are not ours to program, so they must + // not be kept: this is what purges entries inherited from a build + // that programmed every service on every node. + if !c.servesEndpoint(serviceEndpoints.Endpoint) { + continue + } + if len(serviceEndpoints.Service.Status.LoadBalancer.Ingress) > 0 { serviceIP = serviceEndpoints.Service.Status.LoadBalancer.Ingress[0].IP } @@ -582,7 +658,7 @@ func (c *ServicesController) cleanupRemovedServices() error { if !hasValidServiceIP(se.Service) || !hasValidEndpointIP(se.Endpoint) { continue } - if wholeIPPassthrough(se.Service) { + if wholeIPPassthrough(se.Service) || !c.servesEndpoint(se.Endpoint) { continue } keepFilters[se.Service.Status.LoadBalancer.Ingress[0].IP] = nat.PortFilterEntry{ @@ -603,7 +679,7 @@ func (c *ServicesController) cleanupRemovedServices() error { if !hasValidServiceIP(se.Service) || !hasValidEndpointIP(se.Endpoint) { continue } - if wholeIPPassthrough(se.Service) || !allowICMP(se.Service) { + if wholeIPPassthrough(se.Service) || !allowICMP(se.Service) || !c.servesEndpoint(se.Endpoint) { continue } keepICMP[se.Service.Status.LoadBalancer.Ingress[0].IP] = se.Endpoint.Subsets[0].Addresses[0].IP diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 549e4b4..40aae84 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -5,8 +5,135 @@ import ( v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + nat "github.com/cozystack/cozy-proxy/pkg/proxy" ) +// epOnNode builds an Endpoints with a single address, optionally pinned to a +// node. Pass an empty node to omit NodeName, as an endpoint with no scheduling +// information would have. +func epOnNode(podIP, node string) *v1.Endpoints { + addr := v1.EndpointAddress{IP: podIP} + if node != "" { + addr.NodeName = &node + } + return &v1.Endpoints{ + Subsets: []v1.EndpointSubset{{Addresses: []v1.EndpointAddress{addr}}}, + } +} + +// lbService builds a LoadBalancer service already assigned the given IP. +func lbService(svcIP string, annot map[string]string) *v1.Service { + return &v1.Service{ + ObjectMeta: metav1.ObjectMeta{Annotations: annot}, + Status: v1.ServiceStatus{ + LoadBalancer: v1.LoadBalancerStatus{ + Ingress: []v1.LoadBalancerIngress{{IP: svcIP}}, + }, + }, + } +} + +// recordingProxy captures which datapath calls a controller makes. +type recordingProxy struct { + nat.DummyProxyProcessor + calls []string +} + +func (r *recordingProxy) EnsureRules(svcIP, podIP string) error { + r.calls = append(r.calls, "EnsureRules") + return nil +} + +func (r *recordingProxy) DeleteRules(svcIP, podIP string) error { + r.calls = append(r.calls, "DeleteRules") + return nil +} + +func (r *recordingProxy) EnsurePortFilter(svcIP, podIP string, ports []v1.ServicePort) error { + r.calls = append(r.calls, "EnsurePortFilter") + return nil +} + +func (r *recordingProxy) DeletePortFilter(svcIP, podIP string) error { + r.calls = append(r.calls, "DeletePortFilter") + return nil +} + +func (r *recordingProxy) has(call string) bool { + for _, c := range r.calls { + if c == call { + return true + } + } + return false +} + +func TestServesEndpoint(t *testing.T) { + cases := []struct { + name string + nodeName string + ep *v1.Endpoints + expect bool + }{ + {"backend on this node", "node-a", epOnNode("10.0.0.1", "node-a"), true}, + {"backend on another node", "node-a", epOnNode("10.0.0.1", "node-b"), false}, + {"endpoint without NodeName is programmed", "node-a", epOnNode("10.0.0.1", ""), true}, + {"NODE_NAME unset programs everything", "", epOnNode("10.0.0.1", "node-b"), true}, + {"invalid endpoint is programmed", "node-a", &v1.Endpoints{}, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + ctrl := &ServicesController{NodeName: c.nodeName} + if got := ctrl.servesEndpoint(c.ep); got != c.expect { + t.Errorf("servesEndpoint = %v, want %v", got, c.expect) + } + }) + } +} + +// A non-owning node must not translate the service IP: doing so desynchronises +// conntrack on the owning node and gets the reply dropped by port_filter. +func TestApplyRulesSkipsRemoteBackend(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + + local := &recordingProxy{} + localCtrl := &ServicesController{Proxy: local, NodeName: "node-a"} + localCtrl.applyRules(svc, epOnNode("10.0.0.1", "node-a"), "test") + if !local.has("EnsureRules") || !local.has("EnsurePortFilter") { + t.Errorf("owning node must program the datapath, got %v", local.calls) + } + + remote := &recordingProxy{} + remoteCtrl := &ServicesController{Proxy: remote, NodeName: "node-a"} + remoteCtrl.applyRules(svc, epOnNode("10.0.0.1", "node-b"), "test") + if remote.has("EnsureRules") || remote.has("EnsurePortFilter") { + t.Errorf("non-owning node must not program the datapath, got %v", remote.calls) + } + if !remote.has("DeleteRules") { + t.Errorf("non-owning node must withdraw any inherited rules, got %v", remote.calls) + } +} + +// A migrated VM leaves its old mapping behind on the node it came from. +func TestWithdrawStaleEndpoint(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + + moved := &recordingProxy{} + movedCtrl := &ServicesController{Proxy: moved, NodeName: "node-a"} + movedCtrl.withdrawStaleEndpoint(svc, epOnNode("10.0.0.1", "node-a"), "10.0.0.2", "test") + if !moved.has("DeleteRules") { + t.Errorf("changed pod IP must withdraw the previous mapping, got %v", moved.calls) + } + + same := &recordingProxy{} + sameCtrl := &ServicesController{Proxy: same, NodeName: "node-a"} + sameCtrl.withdrawStaleEndpoint(svc, epOnNode("10.0.0.1", "node-a"), "10.0.0.1", "test") + if len(same.calls) != 0 { + t.Errorf("unchanged pod IP must not touch the datapath, got %v", same.calls) + } +} + func svcWith(annot map[string]string) *v1.Service { return &v1.Service{ObjectMeta: metav1.ObjectMeta{Annotations: annot}} } From 8aee75fcd5761c22427106058f35936d3a61ee92 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Fri, 7 Aug 2026 18:11:06 +0200 Subject: [PATCH 2/2] proxy: do not abort the pod when startup cleanup hits ENOENT CleanupRules queued deletions and additions into a single batch and treated any flush error as fatal. Deleting a set element that is already gone reports ENOENT, which fails the whole flush, so the controller returned an error, the manager exited, and the DaemonSet pod entered CrashLoopBackOff with the node's datapath left half-programmed. Scoping the rules to the local node made this reliable rather than rare: the first startup after the change deletes every entry the node inherited for backends it does not host, which is most of them. Commit deletions separately from additions, tolerate ENOENT on the flush the way DeleteRules, DeletePortFilter and DeleteICMPAllow already do, and apply the same split to CleanupPortFilters and CleanupICMPAllow. A cleanup failure is now logged instead of aborting Start, since the informers converge on the next event anyway and staying up with stale entries beats exiting with a partial ruleset. Observed on a 3-node cluster carrying 15 managed services: the transition logs "Ignoring ENOENT on flush" for the cleanup deletions and completes with zero restarts, where it previously crash-looped. Signed-off-by: Mattia Eleuteri --- pkg/controllers/services_controller.go | 9 +++-- pkg/proxy/nft.go | 47 +++++++++++++++++++++++--- 2 files changed, 49 insertions(+), 7 deletions(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index 88e174a..a38c1ae 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -267,12 +267,15 @@ func (c *ServicesController) Start(ctx context.Context) error { } log.Info("endpoints synchronization completed") - // Run cleanup for removed services. + // Run cleanup for removed services. A failure here is logged but does not + // abort: exiting takes the pod down and leaves the node's datapath + // half-programmed, whereas the informers below converge on the next event. log.Info("running cleanup for removed services") if err := c.cleanupRemovedServices(); err != nil { - return fmt.Errorf("failed to cleanup removed services: %w", err) + log.Error(err, "cleanup of removed services failed, continuing with reconciliation") + } else { + log.Info("cleanup of removed services completed") } - log.Info("cleanup of removed services completed") <-ctx.Done() log.Info("shutting down services-controller") diff --git a/pkg/proxy/nft.go b/pkg/proxy/nft.go index afe72ac..1caa33f 100644 --- a/pkg/proxy/nft.go +++ b/pkg/proxy/nft.go @@ -527,6 +527,24 @@ func (p *NFTProxyProcessor) DeleteRules(svcIP, podIP string) error { return nil } +// flushTolerateENOENT commits the pending batch and treats ENOENT as success. +// +// Deleting a set element that is already gone reports ENOENT, which fails the +// whole flush. Deletions must therefore be committed on their own, so a stale +// element cannot mask a genuine failure among the additions that would +// otherwise share the batch. +func (p *NFTProxyProcessor) flushTolerateENOENT(op string) error { + err := p.conn.Flush() + if err == nil { + return nil + } + if errors.Is(err, unix.ENOENT) { + log.Info("Ignoring ENOENT on flush — element already gone", "op", op) + return nil + } + return err +} + // CleanupRules receives a keepMap (keys: svcIP, values: podIP) representing the desired state. // It recovers from an inconsistent state by: // 1. Removing any mappings in the pod_svc and svc_pod maps that do not match keepMap. @@ -585,6 +603,13 @@ func (p *NFTProxyProcessor) CleanupRules(keepMap map[string]string) error { log.Error(err, "Failed to delete inconsistent mappings from svc_pod") return fmt.Errorf("failed to delete inconsistent mappings from svc_pod: %v", err) } + // Commit the deletions before queueing the additions below: an element + // that is already gone fails the flush, and a shared batch would report + // that as a cleanup failure, which aborts the controller at startup. + if err := p.flushTolerateENOENT("CleanupRules deletions"); err != nil { + log.Error(err, "Failed to commit cleanup deletions") + return fmt.Errorf("failed to commit cleanup deletions: %v", err) + } log.Info("Inconsistent mappings removed from both maps") } else { log.Info("No inconsistent mappings found in maps") @@ -617,7 +642,10 @@ func (p *NFTProxyProcessor) CleanupRules(keepMap map[string]string) error { } // --- Final commit --- - if err := p.conn.Flush(); err != nil { + // Startup cleanup must not be fatal: aborting here takes the whole + // DaemonSet pod down and leaves the node's datapath half-programmed, while + // the reconcile loop would have converged on the next event anyway. + if err := p.flushTolerateENOENT("CleanupRules additions"); err != nil { log.Error(err, "Failed to commit cleanup changes") return fmt.Errorf("failed to commit cleanup changes: %v", err) } @@ -804,6 +832,13 @@ func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) return fmt.Errorf("failed to delete stale allowed_ports: %v", err) } } + // Commit deletions separately so an element that is already gone cannot + // fail the batch carrying the additions below. + if len(delPods) > 0 || len(delPorts) > 0 { + if err := p.flushTolerateENOENT("CleanupPortFilters deletions"); err != nil { + return fmt.Errorf("failed to flush CleanupPortFilters deletions: %v", err) + } + } if len(addPods) > 0 { if err := p.conn.SetAddElements(p.filteredPods, addPods); err != nil { return fmt.Errorf("failed to add filtered_pods: %v", err) @@ -815,8 +850,8 @@ func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) } } - // 5. Single flush. - if err := p.conn.Flush(); err != nil { + // 5. Commit the additions. Startup cleanup must not abort the pod. + if err := p.flushTolerateENOENT("CleanupPortFilters additions"); err != nil { return fmt.Errorf("failed to flush CleanupPortFilters: %v", err) } log.Info("CleanupPortFilters completed", @@ -948,6 +983,10 @@ func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { if err := p.conn.SetDeleteElements(p.icmpAllowedPods, delPods); err != nil { return fmt.Errorf("failed to delete stale icmp_allowed_pods: %v", err) } + // Commit deletions separately: see flushTolerateENOENT. + if err := p.flushTolerateENOENT("CleanupICMPAllow deletions"); err != nil { + return fmt.Errorf("failed to flush CleanupICMPAllow deletions: %v", err) + } } if len(addPods) > 0 { if err := p.conn.SetAddElements(p.icmpAllowedPods, addPods); err != nil { @@ -955,7 +994,7 @@ func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { } } - if err := p.conn.Flush(); err != nil { + if err := p.flushTolerateENOENT("CleanupICMPAllow additions"); err != nil { return fmt.Errorf("failed to flush CleanupICMPAllow: %v", err) } log.Info("CleanupICMPAllow completed",