From 049ee94981757b366e85de6c1f9690896c2163d0 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Fri, 7 Aug 2026 17:44:42 +0200 Subject: [PATCH 01/11] 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 02/11] 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", From 16057043f51112d639047a760ba2fb98048d3f3d Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 12:05:23 +0200 Subject: [PATCH 03/11] proxy: keep the egress SNAT map cluster-wide Scoping the datapath to the node hosting the backend was right for the destination rewrite and wrong for the source one. Both were scoped at once, which fixed a managed backend calling another managed backend and broke every other intra-cluster client. An intra-cluster client is source-NATed by kube-ovn to its own node address, which OVN knows how to reach. The backend's reply is therefore tunnelled straight to that node over Geneve and never traverses the hosting node's netfilter hooks, so egress_snat there never sees it. The client's node is the only place left where the pod IP can still be turned back into the service IP, and with pod_svc node-local it holds no entry for a backend it does not host. The reply arrives with saddr=podIP, matches nothing in the client's conntrack, and the client answers with a RST. A connection to a backend on the same node works, one to a backend on any other node hangs. A managed client escapes this because egress_snat gives it its own service IP as source, an address OVN does not know: its peer's reply leaves through the host stack and gets rewritten on the way out. That is why scoping both maps looked correct. Program pod_svc on every node for every managed service, and keep svc_pod, allowed_ports and icmp_allowed_pods on the hosting node only. EnsureRules/DeleteRules are split into the egress and ingress halves so each map resolves its own key conflicts, and CleanupRules now takes one keep set per map, so a node purges the entries of the other scope on upgrade in either direction. Verified on a three-node cluster: adding the missing pod_svc entry by hand on the client's node unblocks the cross-node flow and removing it breaks it again, while a client turned into a managed backend reaches the same peers with no such entry. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 14 + pkg/controllers/services_controller.go | 129 ++++--- pkg/controllers/services_controller_test.go | 121 +++++- pkg/proxy/dummy.go | 22 +- pkg/proxy/interface.go | 36 +- pkg/proxy/nft.go | 384 +++++++++++--------- 6 files changed, 450 insertions(+), 256 deletions(-) diff --git a/README.md b/README.md index 4c497e1..9c8705c 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,20 @@ The nftables ruleset placed in table `ip cozy_proxy` consists of: `allowICMP: "true"` annotation is set, the pod IP is added to `icmp_allowed_pods` and ICMP toward it is accepted before the drop rule. +### Rule scoping + +The two rewrites are deliberately not scoped alike. Every controller instance +watches every Service, but programs: + +| Object | Scope | Why | +|---|---|---| +| `svc_pod` (`ingress_dnat`), `allowed_ports`, `icmp_allowed_pods` | the node hosting the backend pod | A non-hosting node would rewrite the destination before the packet has even left it. The hosting node then records a conntrack tuple of `(client -> podIP)`, while the reply leaves the pod and gets `saddr` rewritten to the service IP by `egress_snat` at priority `raw`, before conntrack. `(svcIP -> client)` matches nothing, so the reply is not `established` and `port_filter` drops it. | +| `pod_svc` (`egress_snat`) | every node | An intra-cluster client is source-NATed by the CNI to its own node address, which the overlay knows how to reach directly. The backend's reply is then tunnelled straight to that node and never traverses the hosting node's netfilter hooks, so the client's node is the only place left where the pod IP can still be turned back into the service IP. Without the entry there, the reply arrives with the wrong source and the client answers with a RST — a cross-node connection that hangs while the same-node one works. | + +`NODE_NAME` carries the node identity. When it is unset the ingress scope +check is disabled and everything is programmed everywhere, so the binary still +runs under a chart that does not inject it. + ## Installation Install controller using Helm-chart: diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index a38c1ae..e030517 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -97,9 +97,10 @@ type ServicesController struct { 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 is the node this instance runs on. It scopes the ingress + // half of the datapath — the destination rewrite and the port filter — + // to backend pods hosted here. Empty disables the check and programs + // everything everywhere. NodeName string } @@ -115,17 +116,20 @@ func endpointNode(ep *v1.Endpoints) (string, bool) { return *node, true } -// servesEndpoint reports whether this node must program datapath rules for ep. +// servesEndpoint reports whether this node hosts ep's backend, and may +// therefore program the ingress half of the datapath. // -// 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. +// Only the hosting node may rewrite the destination. Doing it 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. +// +// This says nothing about the egress half: the source rewrite is programmed on +// every node regardless, see applyRules. // // 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. +// carrying no NodeName) the ingress 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 @@ -137,27 +141,55 @@ func (c *ServicesController) servesEndpoint(ep *v1.Endpoints) bool { 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. +// applyRules programs the datapath for a (service, endpoint) pair. The two +// halves have different scopes. 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 + // The source rewrite goes on every node, including the ones that do not + // host the backend. When the client is inside the cluster, kube-ovn SNATs + // it to its own node address, which OVN knows how to reach: the backend's + // reply is then handed straight to that node over the Geneve tunnel and + // never traverses the backend node's netfilter hooks. The client's node is + // the last place where the pod IP can still be turned back into the + // service IP the client's conntrack is waiting for. Without it the reply + // arrives with the wrong source and the client answers it with a RST. + if err := c.Proxy.EnsureEgressSNAT(svcIP, podIP); err != nil { + log.Error(err, "failed to ensure egress SNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + } + + // The destination rewrite and the port filter stay on the hosting node. if !c.servesEndpoint(ep) { - c.withdrawRules(svcIP, podIP, ctx+" (backend not on this node)") + c.withdrawIngressRules(svcIP, podIP, ctx+" (backend not on this node)") return } - c.Proxy.EnsureRules(svcIP, podIP) + if err := c.Proxy.EnsureIngressDNAT(svcIP, podIP); err != nil { + log.Error(err, "failed to ensure ingress DNAT "+ctx, "svcIP", svcIP, "podIP", 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) { +// withdrawIngressRules removes the ingress half only — destination rewrite and +// port filter — and leaves the source rewrite in place. This is what a node +// that does not host the backend must end up with. +func (c *ServicesController) withdrawIngressRules(svcIP, podIP, ctx string) { c.clearPortFilter(svcIP, podIP, ctx) - c.Proxy.DeleteRules(svcIP, podIP) + if err := c.Proxy.DeleteIngressDNAT(svcIP, podIP); err != nil { + log.Error(err, "failed to delete ingress DNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + } +} + +// withdrawRules removes every datapath entry for the pair, both halves. Used +// when the pair itself is going away: service deleted, endpoint gone, or a pod +// IP that has been replaced. Absent entries are not an error. +func (c *ServicesController) withdrawRules(svcIP, podIP, ctx string) { + c.withdrawIngressRules(svcIP, podIP, ctx) + if err := c.Proxy.DeleteEgressSNAT(svcIP, podIP); err != nil { + log.Error(err, "failed to delete egress SNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + } } // withdrawStaleEndpoint drops the rules of a previous endpoint whose pod IP no @@ -334,8 +366,7 @@ func (c *ServicesController) deleteServiceFunc(obj interface{}) { svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP podIP := se.Endpoint.Subsets[0].Addresses[0].IP - c.clearPortFilter(svcIP, podIP, "on svc deletion") - c.Proxy.DeleteRules(svcIP, podIP) + c.withdrawRules(svcIP, podIP, "on svc deletion") c.Services.Delete(svc.Namespace, svc.Name) } @@ -354,8 +385,7 @@ func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { if hasValidServiceIP(se.Service) && hasValidEndpointIP(se.Endpoint) { svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP podIP := se.Endpoint.Subsets[0].Addresses[0].IP - c.clearPortFilter(svcIP, podIP, "on annotation removal") - c.Proxy.DeleteRules(svcIP, podIP) + c.withdrawRules(svcIP, podIP, "on annotation removal") } c.Services.Delete(svc.Namespace, svc.Name) } @@ -368,8 +398,7 @@ func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { if hasValidServiceIP(se.Service) && hasValidEndpointIP(se.Endpoint) { svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP podIP := se.Endpoint.Subsets[0].Addresses[0].IP - c.clearPortFilter(svcIP, podIP, "on svc IP loss") - c.Proxy.DeleteRules(svcIP, podIP) + c.withdrawRules(svcIP, podIP, "on svc IP loss") } c.Services.Delete(svc.Namespace, svc.Name) } @@ -395,9 +424,10 @@ func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { if ep == nil || !hasValidEndpointIP(ep) { if se, exists := c.Services.Get(svc.Namespace, svc.Name); exists && hasValidServiceIP(se.Service) && hasValidEndpointIP(se.Endpoint) { - c.Proxy.DeleteRules( + c.withdrawRules( se.Service.Status.LoadBalancer.Ingress[0].IP, se.Endpoint.Subsets[0].Addresses[0].IP, + "on endpoint loss", ) } c.Services.Set(svc.Namespace, svc.Name, &ServiceEndpoints{Service: svc, Endpoint: nil}) @@ -459,8 +489,7 @@ func (c *ServicesController) deleteEndpointFunc(obj interface{}) { } svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP podIP := se.Endpoint.Subsets[0].Addresses[0].IP - c.clearPortFilter(svcIP, podIP, "on endpoint delete") - c.Proxy.DeleteRules(svcIP, podIP) + c.withdrawRules(svcIP, podIP, "on endpoint delete") // Set the endpoint to nil. c.Services.SetEndpoint(ep.Namespace, ep.Name, nil) } @@ -482,8 +511,7 @@ func (c *ServicesController) updateEndpointFunc(oldObj, newObj interface{}) { if hasValidServiceIP(se.Service) && hasValidEndpointIP(se.Endpoint) { svcIP := se.Service.Status.LoadBalancer.Ingress[0].IP oldPodIP := se.Endpoint.Subsets[0].Addresses[0].IP - c.clearPortFilter(svcIP, oldPodIP, "on endpoint invalidation") - c.Proxy.DeleteRules(svcIP, oldPodIP) + c.withdrawRules(svcIP, oldPodIP, "on endpoint invalidation") } c.Services.SetEndpoint(ep.Namespace, ep.Name, ep) return @@ -620,34 +648,33 @@ func (c *ServicesController) clearPortFilter(svcIP, podIP, ctx string) { // cleanupRemovedServices performs an initial cleanup for removed services. func (c *ServicesController) cleanupRemovedServices() error { - keepMap := make(map[string]string) + // keepEgress holds every managed pair, because the source rewrite is + // programmed cluster-wide. keepIngress holds only the pairs whose backend + // runs here, because the destination rewrite is node-local. The difference + // between the two is what purges entries inherited from a build that + // scoped both maps alike. + keepEgress := make(map[string]string) + keepIngress := make(map[string]string) // Get a snapshot of all managed services. allServices := c.Services.GetAll() for _, serviceEndpoints := range allServices { - 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 serviceEndpoints.Service == nil || serviceEndpoints.Endpoint == nil { + continue + } + if !hasValidServiceIP(serviceEndpoints.Service) || !hasValidEndpointIP(serviceEndpoints.Endpoint) { + continue + } - if len(serviceEndpoints.Service.Status.LoadBalancer.Ingress) > 0 { - serviceIP = serviceEndpoints.Service.Status.LoadBalancer.Ingress[0].IP - } - if len(serviceEndpoints.Endpoint.Subsets) > 0 && len(serviceEndpoints.Endpoint.Subsets[0].Addresses) > 0 { - endpointIP = serviceEndpoints.Endpoint.Subsets[0].Addresses[0].IP - } + serviceIP := serviceEndpoints.Service.Status.LoadBalancer.Ingress[0].IP + endpointIP := serviceEndpoints.Endpoint.Subsets[0].Addresses[0].IP - if serviceIP != "" && endpointIP != "" { - keepMap[serviceIP] = endpointIP - } + keepEgress[serviceIP] = endpointIP + if c.servesEndpoint(serviceEndpoints.Endpoint) { + keepIngress[serviceIP] = endpointIP } } // Call InitialCleanup with the snapshot. - if err := c.Proxy.CleanupRules(keepMap); err != nil { + if err := c.Proxy.CleanupRules(keepEgress, keepIngress); err != nil { return fmt.Errorf("failed to perform initial cleanup: %w", err) } // Build per-svc port filter snapshot for services in non-passthrough mode. diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 40aae84..46b2ff5 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -37,16 +37,35 @@ func lbService(svcIP string, annot map[string]string) *v1.Service { // recordingProxy captures which datapath calls a controller makes. type recordingProxy struct { nat.DummyProxyProcessor - calls []string + calls []string + keepEgress map[string]string + keepIngress map[string]string } -func (r *recordingProxy) EnsureRules(svcIP, podIP string) error { - r.calls = append(r.calls, "EnsureRules") +func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { + r.calls = append(r.calls, "EnsureEgressSNAT") return nil } -func (r *recordingProxy) DeleteRules(svcIP, podIP string) error { - r.calls = append(r.calls, "DeleteRules") +func (r *recordingProxy) DeleteEgressSNAT(svcIP, podIP string) error { + r.calls = append(r.calls, "DeleteEgressSNAT") + return nil +} + +func (r *recordingProxy) EnsureIngressDNAT(svcIP, podIP string) error { + r.calls = append(r.calls, "EnsureIngressDNAT") + return nil +} + +func (r *recordingProxy) DeleteIngressDNAT(svcIP, podIP string) error { + r.calls = append(r.calls, "DeleteIngressDNAT") + return nil +} + +func (r *recordingProxy) CleanupRules(keepEgress, keepIngress map[string]string) error { + r.calls = append(r.calls, "CleanupRules") + r.keepEgress = keepEgress + r.keepIngress = keepIngress return nil } @@ -92,26 +111,46 @@ func TestServesEndpoint(t *testing.T) { } } -// 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) { +// The owning node programs both halves of the datapath. +func TestApplyRulesProgramsBothHalvesOnOwningNode(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) + + for _, want := range []string{"EnsureEgressSNAT", "EnsureIngressDNAT", "EnsurePortFilter"} { + if !local.has(want) { + t.Errorf("owning node must call %s, got %v", want, local.calls) + } } +} + +// A non-owning node must not translate the destination — doing so +// desynchronises conntrack on the owning node and gets the reply dropped by +// port_filter — but it must still program the source rewrite. +// +// That rewrite is the only thing that repairs a reply reaching this node +// straight over the overlay, which is what happens whenever the client is +// inside the cluster and kube-ovn SNATs it to this node's own address. +func TestApplyRulesKeepsEgressSNATOnRemoteBackend(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) 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("EnsureEgressSNAT") { + t.Errorf("non-owning node must still program the source rewrite, got %v", remote.calls) } - if !remote.has("DeleteRules") { - t.Errorf("non-owning node must withdraw any inherited rules, got %v", remote.calls) + if remote.has("DeleteEgressSNAT") { + t.Errorf("non-owning node must not withdraw the source rewrite, got %v", remote.calls) + } + if remote.has("EnsureIngressDNAT") || remote.has("EnsurePortFilter") { + t.Errorf("non-owning node must not program the ingress half, got %v", remote.calls) + } + if !remote.has("DeleteIngressDNAT") { + t.Errorf("non-owning node must withdraw an inherited destination rewrite, got %v", remote.calls) } } @@ -122,8 +161,12 @@ func TestWithdrawStaleEndpoint(t *testing.T) { 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) + // A pod IP that is gone must leave nothing behind, in either half: every + // node carries its source rewrite, so every node has to drop it. + for _, want := range []string{"DeleteIngressDNAT", "DeleteEgressSNAT"} { + if !moved.has(want) { + t.Errorf("changed pod IP must call %s, got %v", want, moved.calls) + } } same := &recordingProxy{} @@ -205,3 +248,49 @@ func TestAllowICMP(t *testing.T) { }) } } + +// The startup snapshot must keep every managed pair in the egress map and only +// the locally hosted ones in the ingress map. Handing the same set to both is +// exactly what leaves an intra-cluster client without a usable reply. +func TestCleanupRemovedServicesScopesKeepMaps(t *testing.T) { + annot := map[string]string{"networking.cozystack.io/wholeIP": "false"} + + ctrl := &ServicesController{Proxy: &recordingProxy{}, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.Services.Set("ns", "local", &ServiceEndpoints{ + Service: lbService("192.0.2.10", annot), + Endpoint: epOnNode("10.0.0.1", "node-a"), + }) + ctrl.Services.Set("ns", "remote", &ServiceEndpoints{ + Service: lbService("192.0.2.11", annot), + Endpoint: epOnNode("10.0.0.2", "node-b"), + }) + // A service still waiting for its endpoint contributes to neither map. + ctrl.Services.Set("ns", "pending", &ServiceEndpoints{ + Service: lbService("192.0.2.12", annot), + Endpoint: nil, + }) + + if err := ctrl.cleanupRemovedServices(); err != nil { + t.Fatalf("cleanupRemovedServices: %v", err) + } + rec := ctrl.Proxy.(*recordingProxy) + + wantEgress := map[string]string{"192.0.2.10": "10.0.0.1", "192.0.2.11": "10.0.0.2"} + if len(rec.keepEgress) != len(wantEgress) { + t.Fatalf("keepEgress = %v, want %v", rec.keepEgress, wantEgress) + } + for svc, pod := range wantEgress { + if rec.keepEgress[svc] != pod { + t.Errorf("keepEgress[%s] = %q, want %q", svc, rec.keepEgress[svc], pod) + } + } + + wantIngress := map[string]string{"192.0.2.10": "10.0.0.1"} + if len(rec.keepIngress) != len(wantIngress) { + t.Fatalf("keepIngress = %v, want %v", rec.keepIngress, wantIngress) + } + if rec.keepIngress["192.0.2.10"] != "10.0.0.1" { + t.Errorf("keepIngress = %v, want %v", rec.keepIngress, wantIngress) + } +} diff --git a/pkg/proxy/dummy.go b/pkg/proxy/dummy.go index 3734c38..1c1aeea 100644 --- a/pkg/proxy/dummy.go +++ b/pkg/proxy/dummy.go @@ -13,18 +13,28 @@ func (d *DummyProxyProcessor) InitRules() error { return nil } -func (d *DummyProxyProcessor) EnsureRules(SvcIP, PodIP string) error { - fmt.Printf("EnsureRules called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) +func (d *DummyProxyProcessor) EnsureEgressSNAT(SvcIP, PodIP string) error { + fmt.Printf("EnsureEgressSNAT called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) return nil } -func (d *DummyProxyProcessor) DeleteRules(SvcIP, PodIP string) error { - fmt.Printf("DeleteRules called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) +func (d *DummyProxyProcessor) DeleteEgressSNAT(SvcIP, PodIP string) error { + fmt.Printf("DeleteEgressSNAT called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) return nil } -func (d *DummyProxyProcessor) CleanupRules(KeepMap map[string]string) error { - fmt.Println("CleanupRules called with KeepMap:", KeepMap) +func (d *DummyProxyProcessor) EnsureIngressDNAT(SvcIP, PodIP string) error { + fmt.Printf("EnsureIngressDNAT called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) + return nil +} + +func (d *DummyProxyProcessor) DeleteIngressDNAT(SvcIP, PodIP string) error { + fmt.Printf("DeleteIngressDNAT called with SvcIP: %s, PodIP: %s\n", SvcIP, PodIP) + return nil +} + +func (d *DummyProxyProcessor) CleanupRules(keepEgress, keepIngress map[string]string) error { + fmt.Printf("CleanupRules called with keepEgress: %v, keepIngress: %v\n", keepEgress, keepIngress) return nil } diff --git a/pkg/proxy/interface.go b/pkg/proxy/interface.go index 4786120..f7c0e2b 100644 --- a/pkg/proxy/interface.go +++ b/pkg/proxy/interface.go @@ -4,9 +4,39 @@ import corev1 "k8s.io/api/core/v1" type ProxyProcessor interface { InitRules() error - EnsureRules(SvcIP, PodIP string) error - DeleteRules(SvcIP, PodIP string) error - CleanupRules(KeepMap map[string]string) error + + // EnsureEgressSNAT programs the pod_svc entry (pod IP → service IP) read + // by the egress_snat chain, so traffic leaving the backend is seen with + // the service IP as its source. + // + // Every node programs it, for every managed service, whether or not it + // hosts the backend. A reply from the backend to an intra-cluster client + // is handed straight to the client's node over the overlay: it never + // traverses the backend node's netfilter hooks, and the client's node is + // then the only place left where the source can still be rewritten. + EnsureEgressSNAT(SvcIP, PodIP string) error + + // DeleteEgressSNAT removes the pod_svc entry for the pair. No-op if absent. + DeleteEgressSNAT(SvcIP, PodIP string) error + + // EnsureIngressDNAT programs the svc_pod entry (service IP → pod IP) read + // by the ingress_dnat chain, so traffic addressed to the service IP is + // delivered to the backend. + // + // Only the node hosting the backend may program it. On any other node the + // rewrite would happen before the packet even leaves, and the hosting node + // would then record a conntrack tuple the reply can no longer match. + EnsureIngressDNAT(SvcIP, PodIP string) error + + // DeleteIngressDNAT removes the svc_pod entry for the pair. No-op if absent. + DeleteIngressDNAT(SvcIP, PodIP string) error + + // CleanupRules reconciles both maps against the desired state. Both + // arguments map service IP → pod IP: keepEgress covers every managed + // service, keepIngress only the backends hosted on this node. Anything + // else is removed, so state inherited from a build with different scoping + // is purged at startup. + CleanupRules(keepEgress, keepIngress map[string]string) error // EnsurePortFilter installs (or replaces) ingress port-filtering rules // for the given pod IP. Only TCP/UDP traffic destined to one of the diff --git a/pkg/proxy/nft.go b/pkg/proxy/nft.go index 1caa33f..5933745 100644 --- a/pkg/proxy/nft.go +++ b/pkg/proxy/nft.go @@ -387,146 +387,158 @@ func (p *NFTProxyProcessor) InitRules() error { return nil } -// EnsureRules ensures that a one-to-one mapping exists between svcIP and podIP. -// If a mapping already exists for svcIP with a different podIP, -// the old mapping is removed (from svc_pod, pod_svc, and from the raw pod set) -// before the new mapping is added. -func (p *NFTProxyProcessor) EnsureRules(svcIP, podIP string) error { - log.Info("Ensuring NAT mapping", "svcIP", svcIP, "podIP", podIP) - - parsedSvcIP := net.ParseIP(svcIP).To4() - if parsedSvcIP == nil { - return fmt.Errorf("invalid svcIP: %s", svcIP) - } - parsedPodIP := net.ParseIP(podIP).To4() - if parsedPodIP == nil { - return fmt.Errorf("invalid podIP: %s", podIP) - } +// EnsureEgressSNAT adds the pod_svc entry (podIP → svcIP) consulted by the +// egress_snat chain. If podIP is already mapped to another service, that stale +// entry is dropped first: a pod IP can only stand for one service IP. +// +// Programmed on every node, see ProxyProcessor.EnsureEgressSNAT. +func (p *NFTProxyProcessor) EnsureEgressSNAT(svcIP, podIP string) error { + log.Info("Ensuring egress SNAT", "svcIP", svcIP, "podIP", podIP) - // --- Remove conflicting mapping for svcIP in svc_pod map --- - // If svcIP already maps to a different pod, remove that mapping and - // delete the old pod from the raw pod set. - svcPodElems, err := p.conn.GetSetElements(p.svcPodMap) + parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) if err != nil { - log.Error(err, "Failed to get svc_pod map elements") - return fmt.Errorf("failed to get svc_pod map elements: %v", err) - } - for _, el := range svcPodElems { - if bytes.Equal(el.Key, parsedSvcIP) { - // Found an existing mapping for svcIP. - if !bytes.Equal(el.Val, parsedPodIP) { - oldPodIP := el.Val - log.Info("Updating mapping for svc", "svcIP", svcIP, "oldPodIP", net.IP(oldPodIP).String(), "newPodIP", podIP) - // Remove the old mapping from svc_pod. - if err := p.conn.SetDeleteElements(p.svcPodMap, []nftables.SetElement{{Key: parsedSvcIP, Val: oldPodIP}}); err != nil { - log.Error(err, "Failed to delete old svc_pod mapping", "svcIP", svcIP, "oldPodIP", net.IP(oldPodIP).String()) - return fmt.Errorf("failed to delete old svc_pod mapping: %v", err) - } - // Remove the corresponding mapping from pod_svc. - if err := p.conn.SetDeleteElements(p.podSvcMap, []nftables.SetElement{{Key: oldPodIP, Val: parsedSvcIP}}); err != nil { - log.Error(err, "Failed to delete corresponding pod_svc mapping", "oldPodIP", net.IP(oldPodIP).String(), "svcIP", svcIP) - return fmt.Errorf("failed to delete corresponding pod_svc mapping: %v", err) - } - } - break // svcIP mapping handled; exit loop. - } + return err } - // --- Remove conflicting mapping for podIP in pod_svc map --- - // If podIP already maps to a different svc, remove that mapping and delete the podIP - // from the raw pod set (since the old mapping is no longer desired). podSvcElems, err := p.conn.GetSetElements(p.podSvcMap) if err != nil { log.Error(err, "Failed to get pod_svc map elements") return fmt.Errorf("failed to get pod_svc map elements: %v", err) } for _, el := range podSvcElems { - if bytes.Equal(el.Key, parsedPodIP) { - // Found an existing mapping for podIP. - if !bytes.Equal(el.Val, parsedSvcIP) { - log.Info("Updating mapping for pod", "podIP", podIP, "oldSvcIP", net.IP(el.Val).String(), "newSvcIP", svcIP) - // Remove the old mapping from pod_svc. - if err := p.conn.SetDeleteElements(p.podSvcMap, []nftables.SetElement{{Key: parsedPodIP, Val: el.Val}}); err != nil { - log.Error(err, "Failed to delete old pod_svc mapping", "podIP", podIP, "oldSvcIP", net.IP(el.Val).String()) - return fmt.Errorf("failed to delete old pod_svc mapping: %v", err) - } - // Remove the corresponding mapping from svc_pod. - if err := p.conn.SetDeleteElements(p.svcPodMap, []nftables.SetElement{{Key: el.Val, Val: parsedPodIP}}); err != nil { - log.Error(err, "Failed to delete corresponding svc_pod mapping", "oldSvcIP", net.IP(el.Val).String(), "podIP", podIP) - return fmt.Errorf("failed to delete corresponding svc_pod mapping: %v", err) - } + if !bytes.Equal(el.Key, parsedPodIP) { + continue + } + if !bytes.Equal(el.Val, parsedSvcIP) { + log.Info("Updating egress SNAT for pod", "podIP", podIP, + "oldSvcIP", net.IP(el.Val).String(), "newSvcIP", svcIP) + if err := p.conn.SetDeleteElements(p.podSvcMap, []nftables.SetElement{{Key: parsedPodIP, Val: el.Val}}); err != nil { + log.Error(err, "Failed to delete stale pod_svc mapping", "podIP", podIP) + return fmt.Errorf("failed to delete stale pod_svc mapping: %v", err) } - break // podIP mapping handled; exit loop. } + break } - // --- Add the new mapping to both maps --- if err := p.conn.SetAddElements(p.podSvcMap, []nftables.SetElement{{Key: parsedPodIP, Val: parsedSvcIP}}); err != nil { log.Error(err, "Failed to add mapping to pod_svc", "podIP", podIP, "svcIP", svcIP) return fmt.Errorf("failed to add mapping to pod_svc: %v", err) } - if err := p.conn.SetAddElements(p.svcPodMap, []nftables.SetElement{{Key: parsedSvcIP, Val: parsedPodIP}}); err != nil { - log.Error(err, "Failed to add mapping to svc_pod", "svcIP", svcIP, "podIP", podIP) - return fmt.Errorf("failed to add mapping to svc_pod: %v", err) - } - log.Info("Added mapping", "svcIP", svcIP, "podIP", podIP) - - // Commit all changes. if err := p.conn.Flush(); err != nil { - log.Error(err, "Failed to commit EnsureNAT changes") - return fmt.Errorf("failed to commit EnsureNAT changes: %v", err) + log.Error(err, "Failed to commit egress SNAT changes") + return fmt.Errorf("failed to commit egress SNAT changes: %v", err) } - log.Info("NAT mapping ensured successfully", "svcIP", svcIP, "podIP", podIP) + log.Info("Egress SNAT ensured successfully", "svcIP", svcIP, "podIP", podIP) return nil } -// DeleteRules removes the mapping for the given svcIP and podIP from both maps -// and commits the removal from NAT translation maps. -func (p *NFTProxyProcessor) DeleteRules(svcIP, podIP string) error { - log.Info("Deleting NAT mapping", "svcIP", svcIP, "podIP", podIP) +// DeleteEgressSNAT removes the pod_svc entry for the pair. An entry that is +// already gone is not an error. +func (p *NFTProxyProcessor) DeleteEgressSNAT(svcIP, podIP string) error { + log.Info("Deleting egress SNAT", "svcIP", svcIP, "podIP", podIP) - // Parse svcIP and podIP into IPv4 byte slices. - parsedSvcIP := net.ParseIP(svcIP).To4() - if parsedSvcIP == nil { - return fmt.Errorf("invalid svcIP: %s", svcIP) - } - parsedPodIP := net.ParseIP(podIP).To4() - if parsedPodIP == nil { - return fmt.Errorf("invalid podIP: %s", podIP) + parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) + if err != nil { + return err } - // Delete mapping from the "pod_svc" map. if err := p.conn.SetDeleteElements(p.podSvcMap, []nftables.SetElement{ {Key: parsedPodIP, Val: parsedSvcIP}, }); err != nil { log.Error(err, "Failed to delete mapping from pod_svc", "podIP", podIP, "svcIP", svcIP) return fmt.Errorf("failed to delete mapping from pod_svc: %v", err) } + if err := p.flushTolerateENOENT("DeleteEgressSNAT"); err != nil { + log.Error(err, "Failed to commit egress SNAT deletion") + return fmt.Errorf("failed to commit egress SNAT deletion: %v", err) + } + log.Info("Egress SNAT deleted successfully", "svcIP", svcIP, "podIP", podIP) + return nil +} + +// EnsureIngressDNAT adds the svc_pod entry (svcIP → podIP) consulted by the +// ingress_dnat chain. If svcIP is already mapped to another pod — a migrated +// VM, a replaced backend — that stale entry is dropped first. +// +// Programmed only by the node hosting the backend, see +// ProxyProcessor.EnsureIngressDNAT. +func (p *NFTProxyProcessor) EnsureIngressDNAT(svcIP, podIP string) error { + log.Info("Ensuring ingress DNAT", "svcIP", svcIP, "podIP", podIP) + + parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) + if err != nil { + return err + } + + svcPodElems, err := p.conn.GetSetElements(p.svcPodMap) + if err != nil { + log.Error(err, "Failed to get svc_pod map elements") + return fmt.Errorf("failed to get svc_pod map elements: %v", err) + } + for _, el := range svcPodElems { + if !bytes.Equal(el.Key, parsedSvcIP) { + continue + } + if !bytes.Equal(el.Val, parsedPodIP) { + log.Info("Updating ingress DNAT for svc", "svcIP", svcIP, + "oldPodIP", net.IP(el.Val).String(), "newPodIP", podIP) + if err := p.conn.SetDeleteElements(p.svcPodMap, []nftables.SetElement{{Key: parsedSvcIP, Val: el.Val}}); err != nil { + log.Error(err, "Failed to delete stale svc_pod mapping", "svcIP", svcIP) + return fmt.Errorf("failed to delete stale svc_pod mapping: %v", err) + } + } + break + } + + if err := p.conn.SetAddElements(p.svcPodMap, []nftables.SetElement{{Key: parsedSvcIP, Val: parsedPodIP}}); err != nil { + log.Error(err, "Failed to add mapping to svc_pod", "svcIP", svcIP, "podIP", podIP) + return fmt.Errorf("failed to add mapping to svc_pod: %v", err) + } + if err := p.conn.Flush(); err != nil { + log.Error(err, "Failed to commit ingress DNAT changes") + return fmt.Errorf("failed to commit ingress DNAT changes: %v", err) + } + log.Info("Ingress DNAT ensured successfully", "svcIP", svcIP, "podIP", podIP) + return nil +} + +// DeleteIngressDNAT removes the svc_pod entry for the pair. An entry that is +// already gone is not an error. +func (p *NFTProxyProcessor) DeleteIngressDNAT(svcIP, podIP string) error { + log.Info("Deleting ingress DNAT", "svcIP", svcIP, "podIP", podIP) + + parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) + if err != nil { + return err + } - // Delete mapping from the "svc_pod" map. if err := p.conn.SetDeleteElements(p.svcPodMap, []nftables.SetElement{ {Key: parsedSvcIP, Val: parsedPodIP}, }); err != nil { log.Error(err, "Failed to delete mapping from svc_pod", "svcIP", svcIP, "podIP", podIP) return fmt.Errorf("failed to delete mapping from svc_pod: %v", err) } - - // Commit all changes. - if err := p.conn.Flush(); err != nil { - // Check if the error is ENOENT (no such file or directory) and ignore it. - // This may happen if the elements or even the table were already removed. - if errors.Is(err, unix.ENOENT) { - log.Info("Ignoring ENOENT error during flush in DeleteRules", "error", err) - } else { - log.Error(err, "Failed to commit DeleteNAT changes") - return fmt.Errorf("failed to commit DeleteNAT changes: %v", err) - } + if err := p.flushTolerateENOENT("DeleteIngressDNAT"); err != nil { + log.Error(err, "Failed to commit ingress DNAT deletion") + return fmt.Errorf("failed to commit ingress DNAT deletion: %v", err) } - - log.Info("NAT mapping and raw set elements deleted successfully", "svcIP", svcIP, "podIP", podIP) + log.Info("Ingress DNAT deleted successfully", "svcIP", svcIP, "podIP", podIP) return nil } +// parsePair validates a (svcIP, podIP) pair and returns both as IPv4 bytes. +func parsePair(svcIP, podIP string) (net.IP, net.IP, error) { + parsedSvcIP := net.ParseIP(svcIP).To4() + if parsedSvcIP == nil { + return nil, nil, fmt.Errorf("invalid svcIP: %s", svcIP) + } + parsedPodIP := net.ParseIP(podIP).To4() + if parsedPodIP == nil { + return nil, nil, fmt.Errorf("invalid podIP: %s", podIP) + } + return parsedSvcIP, parsedPodIP, nil +} + // flushTolerateENOENT commits the pending batch and treats ENOENT as success. // // Deleting a set element that is already gone reports ENOENT, which fails the @@ -545,103 +557,77 @@ func (p *NFTProxyProcessor) flushTolerateENOENT(op string) error { 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. -// 2. Adding any missing mappings from keepMap into both maps. -// 3. Cleaning up the raw sets (pod and svc) so that only the desired IPs remain. -func (p *NFTProxyProcessor) CleanupRules(keepMap map[string]string) error { - log.Info("Starting CleanupRules", "keepMap", keepMap) +// CleanupRules reconciles both NAT maps against the desired state. +// +// keepEgress and keepIngress both map service IP → pod IP. They differ in +// scope, which is the whole point: pod_svc must carry every managed service so +// a reply reaching this node over the overlay still gets its source rewritten, +// while svc_pod must carry only the backends hosted here so no other node +// translates a destination before the packet has left. +// +// Anything present in a map but absent from its keep set is removed, which is +// what purges state inherited from a build that scoped the two maps alike. +func (p *NFTProxyProcessor) CleanupRules(keepEgress, keepIngress map[string]string) error { + log.Info("Starting CleanupRules", "keepEgress", keepEgress, "keepIngress", keepIngress) - // --- Step 1: Clean up mapping sets --- + // --- Step 1: collect what has to change in both maps --- - // Retrieve current mappings from the pod_svc map. - // Note: pod_svc maps pod IP → svc IP. - podSvcElems, err := p.conn.GetSetElements(p.podSvcMap) + // pod_svc is keyed by pod IP, so the desired state is keepEgress inverted. + desiredPodSvc := make(map[string]string, len(keepEgress)) // pod → svc + for svc, pod := range keepEgress { + desiredPodSvc[pod] = svc + } + + podSvcDel, podSvcAdd, err := p.diffMap(p.podSvcMap, desiredPodSvc) if err != nil { - log.Error(err, "Failed to get pod_svc elements") - return fmt.Errorf("failed to get pod_svc elements: %v", err) + return fmt.Errorf("failed to diff pod_svc: %v", err) + } + svcPodDel, svcPodAdd, err := p.diffMap(p.svcPodMap, keepIngress) + if err != nil { + return fmt.Errorf("failed to diff svc_pod: %v", err) } - // Build a current mapping in svc->pod direction (for easy comparison with keepMap) - currentMapping := make(map[string]string) // key: svc, value: pod - for _, el := range podSvcElems { - pod := net.IP(el.Key).String() - svc := net.IP(el.Val).String() - currentMapping[svc] = pod - } - - // Prepare slices for elements to delete from both maps. - var toDeletePodSvc []nftables.SetElement - var toDeleteSvcPod []nftables.SetElement - - // For each mapping found in the current configuration, if it does not match the desired state, mark it for deletion. - for svc, pod := range currentMapping { - if expectedPod, ok := keepMap[svc]; !ok || expectedPod != pod { - log.Info("Marking inconsistent mapping for deletion", "svcIP", svc, "podIP", pod) - // Prepare deletion elements. - // pod_svc: key = pod, val = svc. - toDeletePodSvc = append(toDeletePodSvc, nftables.SetElement{ - Key: net.ParseIP(pod).To4(), - Val: net.ParseIP(svc).To4(), - }) - // svc_pod: key = svc, val = pod. - toDeleteSvcPod = append(toDeleteSvcPod, nftables.SetElement{ - Key: net.ParseIP(svc).To4(), - Val: net.ParseIP(pod).To4(), - }) - } - } - - // Delete any inconsistent mappings. - if len(toDeletePodSvc) > 0 { - if err := p.conn.SetDeleteElements(p.podSvcMap, toDeletePodSvc); err != nil { - log.Error(err, "Failed to delete inconsistent mappings from pod_svc") - return fmt.Errorf("failed to delete inconsistent mappings from pod_svc: %v", err) - } - if err := p.conn.SetDeleteElements(p.svcPodMap, toDeleteSvcPod); err != nil { - 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. + // --- Step 2: commit the deletions on their own --- + // + // An element that is already gone fails the flush, and a shared batch + // would report that as a cleanup failure, aborting the controller at + // startup. + if len(podSvcDel) > 0 || len(svcPodDel) > 0 { + if len(podSvcDel) > 0 { + if err := p.conn.SetDeleteElements(p.podSvcMap, podSvcDel); err != nil { + log.Error(err, "Failed to delete stale mappings from pod_svc") + return fmt.Errorf("failed to delete stale mappings from pod_svc: %v", err) + } + } + if len(svcPodDel) > 0 { + if err := p.conn.SetDeleteElements(p.svcPodMap, svcPodDel); err != nil { + log.Error(err, "Failed to delete stale mappings from svc_pod") + return fmt.Errorf("failed to delete stale mappings from svc_pod: %v", err) + } + } 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") + log.Info("Stale mappings removed", "podSvc", len(podSvcDel), "svcPod", len(svcPodDel)) } else { - log.Info("No inconsistent mappings found in maps") + log.Info("No stale mappings found in maps") } - // --- Step 2: Add missing mappings from keepMap --- - - // For every desired mapping in keepMap, ensure it exists in both maps. - for svc, pod := range keepMap { - // Check if the current mapping for svc exists and matches. - if existingPod, ok := currentMapping[svc]; !ok || existingPod != pod { - parsedSvcIP := net.ParseIP(svc).To4() - parsedPodIP := net.ParseIP(pod).To4() - if parsedSvcIP == nil || parsedPodIP == nil { - log.Error(nil, "Invalid IP in keepMap", "svcIP", svc, "podIP", pod) - continue - } - // Add mapping to pod_svc (pod → svc) - if err := p.conn.SetAddElements(p.podSvcMap, []nftables.SetElement{{Key: parsedPodIP, Val: parsedSvcIP}}); err != nil { - log.Error(err, "Failed to add missing mapping to pod_svc", "podIP", pod, "svcIP", svc) - return fmt.Errorf("failed to add missing mapping to pod_svc: %v", err) - } - // Add mapping to svc_pod (svc → pod) - if err := p.conn.SetAddElements(p.svcPodMap, []nftables.SetElement{{Key: parsedSvcIP, Val: parsedPodIP}}); err != nil { - log.Error(err, "Failed to add missing mapping to svc_pod", "svcIP", svc, "podIP", pod) - return fmt.Errorf("failed to add missing mapping to svc_pod: %v", err) - } - log.Info("Added missing mapping", "svcIP", svc, "podIP", pod) + // --- Step 3: add whatever is missing --- + if len(podSvcAdd) > 0 { + if err := p.conn.SetAddElements(p.podSvcMap, podSvcAdd); err != nil { + log.Error(err, "Failed to add missing mappings to pod_svc") + return fmt.Errorf("failed to add missing mappings to pod_svc: %v", err) + } + } + if len(svcPodAdd) > 0 { + if err := p.conn.SetAddElements(p.svcPodMap, svcPodAdd); err != nil { + log.Error(err, "Failed to add missing mappings to svc_pod") + return fmt.Errorf("failed to add missing mappings to svc_pod: %v", err) } } - // --- Final commit --- // 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. @@ -649,10 +635,48 @@ func (p *NFTProxyProcessor) CleanupRules(keepMap map[string]string) error { log.Error(err, "Failed to commit cleanup changes") return fmt.Errorf("failed to commit cleanup changes: %v", err) } - log.Info("CleanupRules completed successfully") + log.Info("CleanupRules completed successfully", + "podSvcAdded", len(podSvcAdd), "svcPodAdded", len(svcPodAdd)) return nil } +// diffMap compares one nft map against its desired state and returns the +// elements to delete and to add. Both the map and desired are read in the +// map's own key → value direction. +func (p *NFTProxyProcessor) diffMap(m *nftables.Set, desired map[string]string) (del, add []nftables.SetElement, err error) { + elems, err := p.conn.GetSetElements(m) + if err != nil { + log.Error(err, "Failed to get map elements", "map", m.Name) + return nil, nil, fmt.Errorf("failed to get %s elements: %v", m.Name, err) + } + + current := make(map[string]string, len(elems)) + for _, el := range elems { + key := net.IP(el.Key).String() + val := net.IP(el.Val).String() + current[key] = val + if want, ok := desired[key]; !ok || want != val { + log.Info("Marking stale mapping for deletion", "map", m.Name, "key", key, "value", val) + del = append(del, nftables.SetElement{Key: el.Key, Val: el.Val}) + } + } + + for key, val := range desired { + if cur, ok := current[key]; ok && cur == val { + continue + } + parsedKey := net.ParseIP(key).To4() + parsedVal := net.ParseIP(val).To4() + if parsedKey == nil || parsedVal == nil { + log.Error(nil, "Invalid IP in desired state", "map", m.Name, "key", key, "value", val) + continue + } + log.Info("Adding missing mapping", "map", m.Name, "key", key, "value", val) + add = append(add, nftables.SetElement{Key: parsedKey, Val: parsedVal}) + } + return del, add, nil +} + // EnsurePortFilter installs ingress port filtering rules for the given // (svcIP, podIP) pair. The actual nft set keys are pod IPs, because the // port_filter chain runs at priority filter (0), after ingress_dnat has From 1110b24d0e06b5996c9668e64023662d948fc3ee Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 13:17:23 +0200 Subject: [PATCH 04/11] proxy: retry cleanup deletions per element on ENOENT A flush is a single nftables transaction, so one element that is already gone aborts every other deletion queued with it. The startup purges tolerated that ENOENT on the whole batch, which reported success while removing nothing. Observed on a three-node cluster while rolling the image back: the node had seven stale pod_svc entries, the purge queued all seven plus the mirrored svc_pod deletions, svc_pod was empty, and the resulting ENOENT cancelled the pod_svc deletions too. The log said the cleanup completed successfully and the map still held every entry. Retry element by element when the batch reports ENOENT, so only the elements that really are gone are skipped. Applies to all three startup purges: the NAT maps, the port filters and the ICMP allowlist. Tolerating ENOENT on a single-element deletion stays as it was, since there is nothing else in the batch to lose. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/proxy/nft.go | 98 +++++++++++++++++++++++++++++------------------- 1 file changed, 60 insertions(+), 38 deletions(-) diff --git a/pkg/proxy/nft.go b/pkg/proxy/nft.go index 5933745..ec87e89 100644 --- a/pkg/proxy/nft.go +++ b/pkg/proxy/nft.go @@ -557,6 +557,51 @@ func (p *NFTProxyProcessor) flushTolerateENOENT(op string) error { return err } +// deleteElementsTolerant removes elements from a set, falling back to one +// commit per element when a batched commit reports that one of them is gone. +// +// A flush is a single nftables transaction: one missing element aborts every +// other deletion queued with it. Tolerating that ENOENT on the batch would +// report success while removing nothing, leaving exactly the stale state the +// purge exists to clear — which is how a node can keep entries belonging to a +// scope it no longer programs. Retrying per element skips only the ones that +// really are gone. +func (p *NFTProxyProcessor) deleteElementsTolerant(m *nftables.Set, elems []nftables.SetElement, op string) error { + if len(elems) == 0 { + return nil + } + + if err := p.conn.SetDeleteElements(m, elems); err != nil { + return fmt.Errorf("failed to queue deletions for %s: %v", m.Name, err) + } + err := p.conn.Flush() + if err == nil { + return nil + } + if !errors.Is(err, unix.ENOENT) { + return fmt.Errorf("failed to commit deletions for %s: %v", m.Name, err) + } + + log.Info("Batched deletion hit ENOENT, retrying element by element", + "op", op, "set", m.Name, "elements", len(elems)) + removed := 0 + for _, el := range elems { + if err := p.conn.SetDeleteElements(m, []nftables.SetElement{el}); err != nil { + return fmt.Errorf("failed to queue deletion for %s: %v", m.Name, err) + } + if err := p.conn.Flush(); err != nil { + if errors.Is(err, unix.ENOENT) { + continue // already gone, nothing to do + } + return fmt.Errorf("failed to commit deletion for %s: %v", m.Name, err) + } + removed++ + } + log.Info("Element-by-element deletion completed", + "op", op, "set", m.Name, "removed", removed, "alreadyGone", len(elems)-removed) + return nil +} + // CleanupRules reconciles both NAT maps against the desired state. // // keepEgress and keepIngress both map service IP → pod IP. They differ in @@ -593,21 +638,13 @@ func (p *NFTProxyProcessor) CleanupRules(keepEgress, keepIngress map[string]stri // would report that as a cleanup failure, aborting the controller at // startup. if len(podSvcDel) > 0 || len(svcPodDel) > 0 { - if len(podSvcDel) > 0 { - if err := p.conn.SetDeleteElements(p.podSvcMap, podSvcDel); err != nil { - log.Error(err, "Failed to delete stale mappings from pod_svc") - return fmt.Errorf("failed to delete stale mappings from pod_svc: %v", err) - } + if err := p.deleteElementsTolerant(p.podSvcMap, podSvcDel, "CleanupRules"); err != nil { + log.Error(err, "Failed to delete stale mappings from pod_svc") + return err } - if len(svcPodDel) > 0 { - if err := p.conn.SetDeleteElements(p.svcPodMap, svcPodDel); err != nil { - log.Error(err, "Failed to delete stale mappings from svc_pod") - return fmt.Errorf("failed to delete stale mappings from svc_pod: %v", err) - } - } - 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) + if err := p.deleteElementsTolerant(p.svcPodMap, svcPodDel, "CleanupRules"); err != nil { + log.Error(err, "Failed to delete stale mappings from svc_pod") + return err } log.Info("Stale mappings removed", "podSvc", len(podSvcDel), "svcPod", len(svcPodDel)) } else { @@ -845,23 +882,14 @@ func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) } } - // 4. Batch ops. - if len(delPods) > 0 { - if err := p.conn.SetDeleteElements(p.filteredPods, delPods); err != nil { - return fmt.Errorf("failed to delete stale filtered_pods: %v", err) - } - } - if len(delPorts) > 0 { - if err := p.conn.SetDeleteElements(p.allowedPorts, delPorts); err != nil { - return fmt.Errorf("failed to delete stale allowed_ports: %v", err) - } + // 4. Batch ops. Deletions are committed first and on their own, so an + // element that is already gone cannot fail the batch carrying the + // additions below. + if err := p.deleteElementsTolerant(p.filteredPods, delPods, "CleanupPortFilters"); err != nil { + return 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 err := p.deleteElementsTolerant(p.allowedPorts, delPorts, "CleanupPortFilters"); err != nil { + return err } if len(addPods) > 0 { if err := p.conn.SetAddElements(p.filteredPods, addPods); err != nil { @@ -1003,14 +1031,8 @@ func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { } } - if len(delPods) > 0 { - 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 err := p.deleteElementsTolerant(p.icmpAllowedPods, delPods, "CleanupICMPAllow"); err != nil { + return err } if len(addPods) > 0 { if err := p.conn.SetAddElements(p.icmpAllowedPods, addPods); err != nil { From 0cdd98446fe2ba480c310644066a21283552575e Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 14:01:14 +0200 Subject: [PATCH 05/11] proxy: serialize datapath writes and isolate conflict removals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four nodes lost one service each during a rolling restart, with the same commit error: "conn.Receive: netlink receive: no such file or directory". One of them, a public IP whose ingress DNAT never landed, stayed unreachable for nine minutes until its pod was restarted. nftables.Conn guards its message list but not the interval between queueing messages and committing them, and this connection is shared by the service informer, the endpoint informer and the startup reconciliation. One caller's flush therefore carries another's half-queued messages. A batch is a single transaction, so a stale deletion in it aborts the other caller's addition — and nothing retries, because the controller only reacts to events. The service stays unprogrammed until the next one. Take a mutex across every queue-then-flush sequence, so a batch only ever holds one caller's messages. EnsurePortFilter reaches DeletePortFilter through an unlocked inner function, since sync.Mutex is not reentrant. Also commit each conflict removal on its own, ahead of the addition that replaces it. Serialization alone leaves the ordinary case where the element read a moment earlier is already gone: the removal fails, and with it the addition queued behind it. Splitting them keeps the addition independent of an ENOENT that no longer matters. This is what made EnsurePortFilter dangerous — its rebuild deletes the pod's allowed_ports before re-adding them, so losing the additions leaves the pod in filtered_pods with no open port at all. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/proxy/nft.go | 78 ++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 76 insertions(+), 2 deletions(-) diff --git a/pkg/proxy/nft.go b/pkg/proxy/nft.go index ec87e89..7cb296b 100644 --- a/pkg/proxy/nft.go +++ b/pkg/proxy/nft.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net" + "sync" "github.com/google/nftables" "github.com/google/nftables/binaryutil" @@ -18,6 +19,18 @@ var log = ctrl.Log.WithName("nft-proxy-processor") // NFTProxyProcessor implements a NATProcessor using nftables. type NFTProxyProcessor struct { + // mu serializes every queue-then-flush sequence below. + // + // nftables.Conn guards its own message list, but not the interval between + // queueing messages and committing them. Two informer goroutines — plus + // the startup reconciliation — share this connection, so without this + // lock one flush carries the other caller's half-queued messages. A batch + // is a single transaction: one stale deletion in it aborts the other + // caller's addition, which is then never retried because the controller + // only reacts to events. The service stays unprogrammed until the next + // one, which is how a public IP can go dark for minutes after a restart. + mu sync.Mutex + conn *nftables.Conn // Table "cozy_proxy" will contain all objects. @@ -37,6 +50,9 @@ type NFTProxyProcessor struct { // InitRules initializes the nftables configuration in a single table "cozy_proxy". // It flushes the entire ruleset, then re-creates the table with the desired sets, maps, and chains. func (p *NFTProxyProcessor) InitRules() error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Initializing nftables NAT configuration") // Create a new connection if needed. @@ -393,6 +409,9 @@ func (p *NFTProxyProcessor) InitRules() error { // // Programmed on every node, see ProxyProcessor.EnsureEgressSNAT. func (p *NFTProxyProcessor) EnsureEgressSNAT(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Ensuring egress SNAT", "svcIP", svcIP, "podIP", podIP) parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) @@ -416,6 +435,13 @@ func (p *NFTProxyProcessor) EnsureEgressSNAT(svcIP, podIP string) error { log.Error(err, "Failed to delete stale pod_svc mapping", "podIP", podIP) return fmt.Errorf("failed to delete stale pod_svc mapping: %v", err) } + // Commit the removal on its own. Sharing the batch with the + // addition below would let an element that is already gone abort + // the addition too, and nothing retries it. + if err := p.flushTolerateENOENT("EnsureEgressSNAT conflict removal"); err != nil { + log.Error(err, "Failed to commit stale pod_svc removal", "podIP", podIP) + return fmt.Errorf("failed to commit stale pod_svc removal: %v", err) + } } break } @@ -435,6 +461,9 @@ func (p *NFTProxyProcessor) EnsureEgressSNAT(svcIP, podIP string) error { // DeleteEgressSNAT removes the pod_svc entry for the pair. An entry that is // already gone is not an error. func (p *NFTProxyProcessor) DeleteEgressSNAT(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Deleting egress SNAT", "svcIP", svcIP, "podIP", podIP) parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) @@ -463,6 +492,9 @@ func (p *NFTProxyProcessor) DeleteEgressSNAT(svcIP, podIP string) error { // Programmed only by the node hosting the backend, see // ProxyProcessor.EnsureIngressDNAT. func (p *NFTProxyProcessor) EnsureIngressDNAT(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Ensuring ingress DNAT", "svcIP", svcIP, "podIP", podIP) parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) @@ -486,6 +518,11 @@ func (p *NFTProxyProcessor) EnsureIngressDNAT(svcIP, podIP string) error { log.Error(err, "Failed to delete stale svc_pod mapping", "svcIP", svcIP) return fmt.Errorf("failed to delete stale svc_pod mapping: %v", err) } + // Commit the removal on its own, see EnsureEgressSNAT. + if err := p.flushTolerateENOENT("EnsureIngressDNAT conflict removal"); err != nil { + log.Error(err, "Failed to commit stale svc_pod removal", "svcIP", svcIP) + return fmt.Errorf("failed to commit stale svc_pod removal: %v", err) + } } break } @@ -505,6 +542,9 @@ func (p *NFTProxyProcessor) EnsureIngressDNAT(svcIP, podIP string) error { // DeleteIngressDNAT removes the svc_pod entry for the pair. An entry that is // already gone is not an error. func (p *NFTProxyProcessor) DeleteIngressDNAT(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Deleting ingress DNAT", "svcIP", svcIP, "podIP", podIP) parsedSvcIP, parsedPodIP, err := parsePair(svcIP, podIP) @@ -613,6 +653,9 @@ func (p *NFTProxyProcessor) deleteElementsTolerant(m *nftables.Set, elems []nfta // Anything present in a map but absent from its keep set is removed, which is // what purges state inherited from a build that scoped the two maps alike. func (p *NFTProxyProcessor) CleanupRules(keepEgress, keepIngress map[string]string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Starting CleanupRules", "keepEgress", keepEgress, "keepIngress", keepIngress) // --- Step 1: collect what has to change in both maps --- @@ -721,13 +764,16 @@ func (p *NFTProxyProcessor) diffMap(m *nftables.Set, desired map[string]string) // permitted on the post-DNAT pod IP; all other traffic destined to that // pod IP is dropped. Idempotent. func (p *NFTProxyProcessor) EnsurePortFilter(svcIP, podIP string, ports []corev1.ServicePort) error { + p.mu.Lock() + defer p.mu.Unlock() + // Empty ports list is documented as equivalent to DeletePortFilter: // the caller wants to disable filtering for this pod entirely. Without // this short-circuit we would add the pod to filtered_pods with no // matching allowed_ports entries, which would drop every ingress packet // to the pod IP — the opposite of "no filter". if len(ports) == 0 { - return p.DeletePortFilter(svcIP, podIP) + return p.deletePortFilterLocked(svcIP, podIP) } log.Info("Ensuring port filter", "svcIP", svcIP, "podIP", podIP, "portCount", len(ports)) @@ -737,10 +783,16 @@ func (p *NFTProxyProcessor) EnsurePortFilter(svcIP, podIP string, ports []corev1 } // 1. Remove any pre-existing entries in allowed_ports for podIP so we can - // rebuild the tuple set cleanly (idempotent). + // rebuild the tuple set cleanly (idempotent). Committed on its own: an + // element that is already gone would otherwise abort the additions below + // in the same transaction, leaving the pod in filtered_pods with no + // allowed port — every packet to it dropped. if err := p.removeAllowedPortsForPod(parsedPodIP); err != nil { return err } + if err := p.flushTolerateENOENT("EnsurePortFilter port rebuild"); err != nil { + return fmt.Errorf("failed to commit allowed_ports rebuild for pod %s: %v", podIP, err) + } // 2. Add podIP to filtered_pods (idempotent — Add ignores duplicates). if err := p.conn.SetAddElements(p.filteredPods, []nftables.SetElement{{Key: parsedPodIP}}); err != nil { @@ -780,6 +832,16 @@ func (p *NFTProxyProcessor) EnsurePortFilter(svcIP, podIP string, ports []corev1 // port entries for it. svcIP is used only for logging context. Tolerates // ENOENT for clean idempotency. func (p *NFTProxyProcessor) DeletePortFilter(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + + return p.deletePortFilterLocked(svcIP, podIP) +} + +// deletePortFilterLocked is DeletePortFilter without the lock, for callers +// that already hold it. sync.Mutex is not reentrant, so EnsurePortFilter must +// come through here rather than calling the exported method. +func (p *NFTProxyProcessor) deletePortFilterLocked(svcIP, podIP string) error { log.Info("Deleting port filter", "svcIP", svcIP, "podIP", podIP) parsedPodIP := net.ParseIP(podIP).To4() if parsedPodIP == nil { @@ -811,6 +873,9 @@ func (p *NFTProxyProcessor) DeletePortFilter(svcIP, podIP string) error { // batches SetAddElements / SetDeleteElements per set, and Flushes exactly // once. func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Starting CleanupPortFilters", "keepCount", len(keep)) // 1. Build desired state in memory. @@ -952,6 +1017,9 @@ func (p *NFTProxyProcessor) removeAllowedPortsForPod(parsedPodIP net.IP) error { // traffic to that pod IP bypasses the port_filter drop rule. svcIP is used // only for logging context. Idempotent. func (p *NFTProxyProcessor) EnsureICMPAllow(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Ensuring ICMP allow", "svcIP", svcIP, "podIP", podIP) parsedPodIP := net.ParseIP(podIP).To4() if parsedPodIP == nil { @@ -970,6 +1038,9 @@ func (p *NFTProxyProcessor) EnsureICMPAllow(svcIP, podIP string) error { // DeleteICMPAllow removes podIP from icmp_allowed_pods. Tolerates ENOENT for // clean idempotency. svcIP is used only for logging context. func (p *NFTProxyProcessor) DeleteICMPAllow(svcIP, podIP string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Deleting ICMP allow", "svcIP", svcIP, "podIP", podIP) parsedPodIP := net.ParseIP(podIP).To4() if parsedPodIP == nil { @@ -996,6 +1067,9 @@ func (p *NFTProxyProcessor) DeleteICMPAllow(svcIP, podIP string) error { // must remain in the set. Single-pass diff with one Flush, mirroring // CleanupPortFilters. func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { + p.mu.Lock() + defer p.mu.Unlock() + log.Info("Starting CleanupICMPAllow", "keepCount", len(keep)) desired := make(map[string]bool, len(keep)) From 76f25488325f65dc8f6d062801edacd51449d02d Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 14:29:08 +0200 Subject: [PATCH 06/11] proxy: retry failed datapath writes, commit additions strictly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review feedback on two gaps that the production incident had already demonstrated, plus one it had not. A refused commit was logged and forgotten. The informers only deliver events, so a service whose programming failed stayed unprogrammed until the next one — which is how a public IP stayed dark for nine minutes. Queue the service instead and re-attempt every 30s until a pass is clean. Every datapath call is idempotent, so a re-attempt on something already correct costs nothing. reconcilePortFilter now returns an error so the caller can see it; a filtered pod with no allowed_ports drops every packet, which is the failure worth retrying hardest. A failed startup reconciliation was logged too, and would otherwise wait for the next event or the 12-hour informer resync. It goes through the same queue and stays non-fatal. Additions are now committed with a strict flush. flushTolerateENOENT was used after the addition batches in all three startup purges, where an ENOENT means the table or the set is missing, not that an element was already gone — so it reported a successful reconciliation with nothing installed. Only deletions may tolerate it. Also recorded why errors.Is(err, unix.ENOENT) does hold on the path the kernel actually takes: the errno arrives as a netlink.OpError wrapping a syscall.Errno, which unwraps. github.com/google/nftables v0.3.0 has one branch in receiveAckAware that formats a trailing error ack instead of wrapping it; v0.3.0 is the latest release, and an ENOENT arriving that way is surfaced as a plain error rather than silently swallowed, which is the safe direction. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/controllers/services_controller.go | 160 +++++++++++++++++++- pkg/controllers/services_controller_test.go | 142 +++++++++++++++++ pkg/proxy/nft.go | 21 ++- 3 files changed, 315 insertions(+), 8 deletions(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index e030517..c13d919 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -3,6 +3,7 @@ package controllers import ( "context" "fmt" + "strings" "sync" "time" @@ -102,6 +103,123 @@ type ServicesController struct { // to backend pods hosted here. Empty disables the check and programs // everything everywhere. NodeName string + + // RetryInterval is how often failed datapath writes are re-attempted. + // Zero selects defaultRetryInterval. + RetryInterval time.Duration + + // retryMu guards pendingServices and pendingCleanup. + retryMu sync.Mutex + + // pendingServices holds the keys of services whose datapath programming + // failed. The informers only deliver events, so without a re-attempt a + // transient nftables failure leaves the service unprogrammed until the + // next event — long enough for a public IP to stay dark for minutes. + pendingServices map[string]struct{} + + // pendingCleanup records that the startup reconciliation failed and has + // to run again. It stays non-fatal, but it no longer waits for the next + // event or the 12-hour informer resync. + pendingCleanup bool +} + +// defaultRetryInterval is short enough that a failed write is repaired well +// within a human noticing, and long enough that a permanent failure does not +// flood the log. +const defaultRetryInterval = 30 * time.Second + +// markPending queues a service for another programming attempt. +func (c *ServicesController) markPending(namespace, name string) { + c.retryMu.Lock() + defer c.retryMu.Unlock() + if c.pendingServices == nil { + c.pendingServices = make(map[string]struct{}) + } + c.pendingServices[makeKey(namespace, name)] = struct{}{} +} + +// clearPending drops a service from the retry set after a clean pass. +func (c *ServicesController) clearPending(namespace, name string) { + c.retryMu.Lock() + defer c.retryMu.Unlock() + delete(c.pendingServices, makeKey(namespace, name)) +} + +// markCleanupPending queues the startup reconciliation for another attempt. +func (c *ServicesController) markCleanupPending() { + c.retryMu.Lock() + defer c.retryMu.Unlock() + c.pendingCleanup = true +} + +// takePending returns the queued work and clears it. Anything that fails again +// is re-queued by the attempt itself. +func (c *ServicesController) takePending() (services []string, cleanup bool) { + c.retryMu.Lock() + defer c.retryMu.Unlock() + for k := range c.pendingServices { + services = append(services, k) + } + c.pendingServices = nil + cleanup = c.pendingCleanup + c.pendingCleanup = false + return services, cleanup +} + +// retryPending re-attempts everything that failed since the last pass. Every +// datapath call is idempotent, so a re-attempt on something already correct is +// a no-op. +func (c *ServicesController) retryPending() { + services, cleanup := c.takePending() + + for _, key := range services { + ns, name, ok := splitKey(key) + if !ok { + continue + } + se, exists := c.Services.Get(ns, name) + if !exists || !hasValidServiceIP(se.Service) || !hasValidEndpointIP(se.Endpoint) { + // The service went away or lost its endpoint; the delete paths + // have already withdrawn its rules. + continue + } + log.Info("retrying datapath programming", "service", key) + c.applyRules(se.Service, se.Endpoint, "on retry") + } + + if cleanup { + log.Info("retrying startup cleanup") + if err := c.cleanupRemovedServices(); err != nil { + log.Error(err, "cleanup retry failed, will try again") + } + } +} + +// runRetryLoop re-attempts failed datapath writes until the context ends. +func (c *ServicesController) runRetryLoop(ctx context.Context) { + interval := c.RetryInterval + if interval <= 0 { + interval = defaultRetryInterval + } + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.retryPending() + } + } +} + +// splitKey reverses makeKey. +func splitKey(key string) (namespace, name string, ok bool) { + i := strings.Index(key, "/") + if i <= 0 || i == len(key)-1 { + return "", "", false + } + return key[:i], key[i+1:], true } // endpointNode returns the node hosting the endpoint's first address. @@ -156,20 +274,37 @@ func (c *ServicesController) applyRules(svc *v1.Service, ep *v1.Endpoints, ctx s // the last place where the pod IP can still be turned back into the // service IP the client's conntrack is waiting for. Without it the reply // arrives with the wrong source and the client answers it with a RST. + failed := false if err := c.Proxy.EnsureEgressSNAT(svcIP, podIP); err != nil { log.Error(err, "failed to ensure egress SNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = true } // The destination rewrite and the port filter stay on the hosting node. if !c.servesEndpoint(ep) { c.withdrawIngressRules(svcIP, podIP, ctx+" (backend not on this node)") + c.recordOutcome(svc, failed) return } if err := c.Proxy.EnsureIngressDNAT(svcIP, podIP); err != nil { log.Error(err, "failed to ensure ingress DNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = true + } + if err := c.reconcilePortFilter(svc, svcIP, podIP, ctx); err != nil { + failed = true + } + c.recordOutcome(svc, failed) +} + +// recordOutcome queues the service for another attempt when any part of its +// datapath failed, and clears a previous failure once a pass is clean. +func (c *ServicesController) recordOutcome(svc *v1.Service, failed bool) { + if failed { + c.markPending(svc.Namespace, svc.Name) + return } - c.reconcilePortFilter(svc, svcIP, podIP, ctx) + c.clearPending(svc.Namespace, svc.Name) } // withdrawIngressRules removes the ingress half only — destination rewrite and @@ -304,11 +439,17 @@ func (c *ServicesController) Start(ctx context.Context) error { // half-programmed, whereas the informers below converge on the next event. log.Info("running cleanup for removed services") if err := c.cleanupRemovedServices(); err != nil { - log.Error(err, "cleanup of removed services failed, continuing with reconciliation") + log.Error(err, "cleanup of removed services failed, queued for retry") + c.markCleanupPending() } else { log.Info("cleanup of removed services completed") } + // Re-attempt whatever failed. The informers only deliver events, so + // without this a transient nftables failure leaves a service + // unprogrammed until the next one. + go c.runRetryLoop(ctx) + <-ctx.Done() log.Info("shutting down services-controller") @@ -611,28 +752,39 @@ func allowICMP(svc *v1.Service) bool { // reconcilePortFilter applies the port-filter and ICMP-allow state implied by // the service's annotations. Call sites pass the resolved svcIP/podIP and a // short context string that ends up in error logs. -func (c *ServicesController) reconcilePortFilter(svc *v1.Service, svcIP, podIP, ctx string) { +// It returns an error when any part of the port-filter state could not be +// applied, so the caller can queue the service for another attempt. A filtered +// pod with missing allowed_ports drops every packet, so a silent failure here +// is an outage. +func (c *ServicesController) reconcilePortFilter(svc *v1.Service, svcIP, podIP, ctx string) error { + var failed error if wholeIPPassthrough(svc) { if err := c.Proxy.DeletePortFilter(svcIP, podIP); err != nil { log.Error(err, "failed to delete port filter "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } if err := c.Proxy.DeleteICMPAllow(svcIP, podIP); err != nil { log.Error(err, "failed to delete ICMP allow "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } - return + return failed } if err := c.Proxy.EnsurePortFilter(svcIP, podIP, svc.Spec.Ports); err != nil { log.Error(err, "failed to ensure port filter "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } if allowICMP(svc) { if err := c.Proxy.EnsureICMPAllow(svcIP, podIP); err != nil { log.Error(err, "failed to ensure ICMP allow "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } } else { if err := c.Proxy.DeleteICMPAllow(svcIP, podIP); err != nil { log.Error(err, "failed to delete ICMP allow "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } } + return failed } // clearPortFilter unconditionally removes both port-filter and ICMP-allow diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 46b2ff5..8e46e60 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -1,6 +1,7 @@ package controllers import ( + "errors" "testing" v1 "k8s.io/api/core/v1" @@ -40,10 +41,21 @@ type recordingProxy struct { calls []string keepEgress map[string]string keepIngress map[string]string + + // failEgress makes EnsureEgressSNAT fail, standing in for a refused + // nftables commit. + failEgress bool + // failPortFilter makes EnsurePortFilter fail. + failPortFilter bool + // failCleanup makes CleanupRules fail. + failCleanup bool } func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { r.calls = append(r.calls, "EnsureEgressSNAT") + if r.failEgress { + return errors.New("commit refused") + } return nil } @@ -66,11 +78,17 @@ func (r *recordingProxy) CleanupRules(keepEgress, keepIngress map[string]string) r.calls = append(r.calls, "CleanupRules") r.keepEgress = keepEgress r.keepIngress = keepIngress + if r.failCleanup { + return errors.New("commit refused") + } return nil } func (r *recordingProxy) EnsurePortFilter(svcIP, podIP string, ports []v1.ServicePort) error { r.calls = append(r.calls, "EnsurePortFilter") + if r.failPortFilter { + return errors.New("commit refused") + } return nil } @@ -294,3 +312,127 @@ func TestCleanupRemovedServicesScopesKeepMaps(t *testing.T) { t.Errorf("keepIngress = %v, want %v", rec.keepIngress, wantIngress) } } + +// A refused commit must not be forgotten: the informers only deliver events, so +// a service left unprogrammed would stay that way until the next one. +func TestFailedProgrammingIsQueuedForRetry(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + svc.Namespace, svc.Name = "ns", "svc1" + ep := epOnNode("10.0.0.1", "node-a") + + px := &recordingProxy{failEgress: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.applyRules(svc, ep, "test") + + keys, _ := ctrl.takePending() + if len(keys) != 1 || keys[0] != "ns/svc1" { + t.Fatalf("failed write must be queued, got %v", keys) + } + + // A clean pass clears it again. + px.failEgress = false + ctrl.applyRules(svc, ep, "test") + keys, _ = ctrl.takePending() + if len(keys) != 0 { + t.Errorf("clean pass must clear the queue, got %v", keys) + } +} + +// A port filter that could not be applied is an outage — the pod is filtered +// with no open port — so it must be queued too. +func TestFailedPortFilterIsQueuedForRetry(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + svc.Namespace, svc.Name = "ns", "svc2" + + px := &recordingProxy{failPortFilter: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.applyRules(svc, epOnNode("10.0.0.1", "node-a"), "test") + + keys, _ := ctrl.takePending() + if len(keys) != 1 || keys[0] != "ns/svc2" { + t.Fatalf("failed port filter must be queued, got %v", keys) + } +} + +// The retry pass must re-apply from the stored state. +func TestRetryPendingReappliesDatapath(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + svc.Namespace, svc.Name = "ns", "svc3" + ep := epOnNode("10.0.0.1", "node-a") + + px := &recordingProxy{failEgress: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.Services.Set("ns", "svc3", &ServiceEndpoints{Service: svc, Endpoint: ep}) + ctrl.applyRules(svc, ep, "test") + + px.failEgress = false + px.calls = nil + ctrl.retryPending() + + for _, want := range []string{"EnsureEgressSNAT", "EnsureIngressDNAT", "EnsurePortFilter"} { + if !px.has(want) { + t.Errorf("retry must call %s, got %v", want, px.calls) + } + } + if keys, _ := ctrl.takePending(); len(keys) != 0 { + t.Errorf("successful retry must clear the queue, got %v", keys) + } +} + +// A service that disappeared between the failure and the retry must not be +// re-applied; its delete path has already withdrawn the rules. +func TestRetryPendingSkipsVanishedService(t *testing.T) { + px := &recordingProxy{} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.markPending("ns", "gone") + + ctrl.retryPending() + if len(px.calls) != 0 { + t.Errorf("vanished service must not be re-applied, got %v", px.calls) + } +} + +// A failed startup cleanup must be re-attempted rather than wait for the next +// event or the 12-hour informer resync. +func TestFailedCleanupIsQueuedAndRetried(t *testing.T) { + px := &recordingProxy{failCleanup: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + + if err := ctrl.cleanupRemovedServices(); err == nil { + t.Fatal("cleanupRemovedServices must surface the failure") + } + ctrl.markCleanupPending() + + px.failCleanup = false + px.calls = nil + ctrl.retryPending() + if !px.has("CleanupRules") { + t.Errorf("cleanup must be retried, got %v", px.calls) + } + if _, pending := ctrl.takePending(); pending { + t.Error("successful cleanup retry must clear the flag") + } +} + +func TestSplitKey(t *testing.T) { + cases := []struct { + key, ns, name string + ok bool + }{ + {"ns/name", "ns", "name", true}, + {"ns/sub/name", "ns", "sub/name", true}, + {"noslash", "", "", false}, + {"/name", "", "", false}, + {"ns/", "", "", false}, + {"", "", "", false}, + } + for _, c := range cases { + ns, name, ok := splitKey(c.key) + if ok != c.ok || ns != c.ns || name != c.name { + t.Errorf("splitKey(%q) = (%q,%q,%v), want (%q,%q,%v)", c.key, ns, name, ok, c.ns, c.name, c.ok) + } + } +} diff --git a/pkg/proxy/nft.go b/pkg/proxy/nft.go index 7cb296b..e66e2f4 100644 --- a/pkg/proxy/nft.go +++ b/pkg/proxy/nft.go @@ -585,6 +585,13 @@ func parsePair(svcIP, podIP string) (net.IP, net.IP, error) { // 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. +// Note on detection: the errno travels as an *netlink.OpError wrapping a +// syscall.Errno, which errors.Is unwraps, and that is the path taken when the +// kernel refuses a batch. github.com/google/nftables v0.3.0 has one branch in +// receiveAckAware that formats a trailing error ack with %v instead of wrapping +// it, and an ENOENT arriving that way reads as an ordinary failure. v0.3.0 is +// the latest release, so the caller is returned an error in that case rather +// than silently continuing — which is the safe direction. func (p *NFTProxyProcessor) flushTolerateENOENT(op string) error { err := p.conn.Flush() if err == nil { @@ -711,7 +718,10 @@ func (p *NFTProxyProcessor) CleanupRules(keepEgress, keepIngress map[string]stri // 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 { + // Additions are committed strictly: an ENOENT on an addition-only batch + // means the table or the set is gone, so tolerating it would report a + // successful reconciliation with nothing installed. + if err := p.conn.Flush(); err != nil { log.Error(err, "Failed to commit cleanup changes") return fmt.Errorf("failed to commit cleanup changes: %v", err) } @@ -967,8 +977,10 @@ func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) } } - // 5. Commit the additions. Startup cleanup must not abort the pod. - if err := p.flushTolerateENOENT("CleanupPortFilters additions"); err != nil { + // 5. Commit the additions strictly: an ENOENT here means the table or the + // set is missing, not that an element was already gone. The caller keeps + // startup non-fatal. + if err := p.conn.Flush(); err != nil { return fmt.Errorf("failed to flush CleanupPortFilters: %v", err) } log.Info("CleanupPortFilters completed", @@ -1114,7 +1126,8 @@ func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { } } - if err := p.flushTolerateENOENT("CleanupICMPAllow additions"); err != nil { + // Additions committed strictly, see CleanupPortFilters. + if err := p.conn.Flush(); err != nil { return fmt.Errorf("failed to flush CleanupICMPAllow: %v", err) } log.Info("CleanupICMPAllow completed", From 2eafd9a44ed8c1dfe95a3ef1bac2d25f31853b11 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 15:01:55 +0200 Subject: [PATCH 07/11] controllers: serialize whole reconciliations against the retry loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retry loop added in the previous commit is a third concurrent reconciler, and it read the stored pair the way the informer callbacks do: ServiceMap.Get hands back the pointer to the shared ServiceEndpoints, whose Endpoint field SetEndpoint rewrites under the lock. Reading that field once Get has returned is an unsynchronized read, and the race detector says so. The consequence is worse than the read itself. An informer callback withdraws the rules of a replaced endpoint and then applies the new one. The retry could slip between the two carrying the endpoint it had snapshotted, and its writes would land after the update — restoring the mapping of a pod that no longer exists and pointing the service IP at a dead backend. The proxy serializes individual operations, not a whole reconciliation, so nothing prevented that ordering. Add ServiceMap.Snapshot, which copies the pair out under the lock, and a reconcileMu held across each whole reconciliation: the six informer callbacks, the startup cleanup, and one retry item at a time. The retry takes and releases it per item rather than for the whole pass, so a long pass does not starve the callbacks. The lock is always taken before the proxy's own, never the other way round. The concurrency test drives endpoint replacement against the retry loop; reverting either half of this fix makes it report a data race. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/controllers/services_controller.go | 81 ++++++++++++++++++-- pkg/controllers/services_controller_test.go | 83 +++++++++++++++++++-- 2 files changed, 149 insertions(+), 15 deletions(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index c13d919..81a6bf9 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -56,6 +56,22 @@ func (sm *ServiceMap) Get(namespace, name string) (*ServiceEndpoints, bool) { return se, ok } +// Snapshot returns the stored Service and Endpoints under the lock. +// +// Get hands back the pointer to the shared ServiceEndpoints, whose Endpoint +// field SetEndpoint rewrites under the lock. Reading that field after Get has +// returned is an unsynchronized read, so any caller outside the informer +// callbacks has to come through here. +func (sm *ServiceMap) Snapshot(namespace, name string) (*v1.Service, *v1.Endpoints, bool) { + sm.mu.Lock() + defer sm.mu.Unlock() + se, ok := sm.serviceMapping[makeKey(namespace, name)] + if !ok || se == nil { + return nil, nil, false + } + return se.Service, se.Endpoint, true +} + // Set stores the ServiceEndpoints under the given namespace and name. func (sm *ServiceMap) Set(namespace, name string, se *ServiceEndpoints) { sm.mu.Lock() @@ -108,6 +124,16 @@ type ServicesController struct { // Zero selects defaultRetryInterval. RetryInterval time.Duration + // reconcileMu serializes whole reconciliations, not just the individual + // datapath calls the proxy already serializes. + // + // An informer callback withdraws the rules of a replaced endpoint and then + // applies the new one. Without this lock the retry goroutine can slip + // between the two with the endpoint it snapshotted a moment earlier, and + // its writes land after the update — restoring the mapping of a pod that + // is gone, which points the service IP at a dead backend. + reconcileMu sync.Mutex + // retryMu guards pendingServices and pendingCleanup. retryMu sync.Mutex @@ -177,14 +203,7 @@ func (c *ServicesController) retryPending() { if !ok { continue } - se, exists := c.Services.Get(ns, name) - if !exists || !hasValidServiceIP(se.Service) || !hasValidEndpointIP(se.Endpoint) { - // The service went away or lost its endpoint; the delete paths - // have already withdrawn its rules. - continue - } - log.Info("retrying datapath programming", "service", key) - c.applyRules(se.Service, se.Endpoint, "on retry") + c.retryOne(ns, name, key) } if cleanup { @@ -195,6 +214,24 @@ func (c *ServicesController) retryPending() { } } +// retryOne re-applies a single service, reading its state and programming the +// datapath under the reconciliation lock so an informer callback cannot +// interleave and have this attempt restore what it just withdrew. +func (c *ServicesController) retryOne(namespace, name, key string) { + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + + svc, ep, exists := c.Services.Snapshot(namespace, name) + if !exists || !hasValidServiceIP(svc) || !hasValidEndpointIP(ep) { + // The service went away or lost its endpoint; the delete paths have + // already withdrawn its rules. Drop it from the queue. + c.clearPending(namespace, name) + return + } + log.Info("retrying datapath programming", "service", key) + c.applyRules(svc, ep, "on retry") +} + // runRetryLoop re-attempts failed datapath writes until the context ends. func (c *ServicesController) runRetryLoop(ctx context.Context) { interval := c.RetryInterval @@ -458,6 +495,10 @@ func (c *ServicesController) Start(ctx context.Context) error { // addServiceFunc handles the addition of a service. func (c *ServicesController) addServiceFunc(obj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + svc, ok := obj.(*v1.Service) if !ok { // Object is not a Service. @@ -490,6 +531,10 @@ func (c *ServicesController) addServiceFunc(obj interface{}) { // deleteServiceFunc handles the deletion of a service. func (c *ServicesController) deleteServiceFunc(obj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + svc, ok := obj.(*v1.Service) if !ok { // object is not Service @@ -513,6 +558,10 @@ func (c *ServicesController) deleteServiceFunc(obj interface{}) { // updateServiceFunc handles service updates. func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + // Cast the object to a Service type. svc, ok := newObj.(*v1.Service) if !ok { @@ -588,6 +637,10 @@ func (c *ServicesController) updateServiceFunc(oldObj, newObj interface{}) { // addEndpointFunc handles the addition of endpoints. func (c *ServicesController) addEndpointFunc(obj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + // Cast the object to an Endpoints type. ep, ok := obj.(*v1.Endpoints) if !ok { @@ -614,6 +667,10 @@ func (c *ServicesController) addEndpointFunc(obj interface{}) { // deleteEndpointFunc handles endpoint deletions. func (c *ServicesController) deleteEndpointFunc(obj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + ep, ok := obj.(*v1.Endpoints) if !ok { // object is not Endpoints @@ -637,6 +694,10 @@ func (c *ServicesController) deleteEndpointFunc(obj interface{}) { // updateEndpointFunc handles updates to endpoints. func (c *ServicesController) updateEndpointFunc(oldObj, newObj interface{}) { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + ep, ok := newObj.(*v1.Endpoints) if !ok { // object is not Endpoints @@ -800,6 +861,10 @@ func (c *ServicesController) clearPortFilter(svcIP, podIP, ctx string) { // cleanupRemovedServices performs an initial cleanup for removed services. func (c *ServicesController) cleanupRemovedServices() error { + // Reconciliations are serialized end to end, see reconcileMu. + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + // keepEgress holds every managed pair, because the source rewrite is // programmed cluster-wide. keepIngress holds only the pairs whose backend // runs here, because the destination rewrite is node-local. The difference diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 8e46e60..03f388d 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -2,7 +2,10 @@ package controllers import ( "errors" + "fmt" + "sync" "testing" + "time" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -38,6 +41,7 @@ func lbService(svcIP string, annot map[string]string) *v1.Service { // recordingProxy captures which datapath calls a controller makes. type recordingProxy struct { nat.DummyProxyProcessor + mu sync.Mutex calls []string keepEgress map[string]string keepIngress map[string]string @@ -52,7 +56,7 @@ type recordingProxy struct { } func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { - r.calls = append(r.calls, "EnsureEgressSNAT") + r.record("EnsureEgressSNAT") if r.failEgress { return errors.New("commit refused") } @@ -60,22 +64,22 @@ func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { } func (r *recordingProxy) DeleteEgressSNAT(svcIP, podIP string) error { - r.calls = append(r.calls, "DeleteEgressSNAT") + r.record("DeleteEgressSNAT") return nil } func (r *recordingProxy) EnsureIngressDNAT(svcIP, podIP string) error { - r.calls = append(r.calls, "EnsureIngressDNAT") + r.record("EnsureIngressDNAT") return nil } func (r *recordingProxy) DeleteIngressDNAT(svcIP, podIP string) error { - r.calls = append(r.calls, "DeleteIngressDNAT") + r.record("DeleteIngressDNAT") return nil } func (r *recordingProxy) CleanupRules(keepEgress, keepIngress map[string]string) error { - r.calls = append(r.calls, "CleanupRules") + r.record("CleanupRules") r.keepEgress = keepEgress r.keepIngress = keepIngress if r.failCleanup { @@ -85,7 +89,7 @@ func (r *recordingProxy) CleanupRules(keepEgress, keepIngress map[string]string) } func (r *recordingProxy) EnsurePortFilter(svcIP, podIP string, ports []v1.ServicePort) error { - r.calls = append(r.calls, "EnsurePortFilter") + r.record("EnsurePortFilter") if r.failPortFilter { return errors.New("commit refused") } @@ -93,11 +97,21 @@ func (r *recordingProxy) EnsurePortFilter(svcIP, podIP string, ports []v1.Servic } func (r *recordingProxy) DeletePortFilter(svcIP, podIP string) error { - r.calls = append(r.calls, "DeletePortFilter") + r.record("DeletePortFilter") return nil } +// record appends under the lock: the retry loop and the informer callbacks +// both drive the proxy in the concurrency test. +func (r *recordingProxy) record(call string) { + r.mu.Lock() + defer r.mu.Unlock() + r.calls = append(r.calls, call) +} + func (r *recordingProxy) has(call string) bool { + r.mu.Lock() + defer r.mu.Unlock() for _, c := range r.calls { if c == call { return true @@ -436,3 +450,58 @@ func TestSplitKey(t *testing.T) { } } } + +// The retry goroutine and the informer callbacks reconcile the same stored +// pair. The retry must read it under the map lock and hold the reconciliation +// lock while it programs, or it applies an endpoint the update has already +// withdrawn — pointing the service IP at a pod that is gone. +// +// Run with -race: an unsynchronized read of the shared Endpoint field shows up +// here, which is what reading it through ServiceMap.Get used to do. +func TestRetryIsSerializedAgainstEndpointUpdates(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + svc.Namespace, svc.Name = "ns", "svc" + + ctrl := &ServicesController{Proxy: &recordingProxy{}, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.Services.Set("ns", "svc", &ServiceEndpoints{ + Service: svc, + Endpoint: epOnNode("10.0.0.1", "node-a"), + }) + + stop := make(chan struct{}) + var wg sync.WaitGroup + + // Informer side: keep replacing the endpoint, as a migrating VM does. + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + ctrl.Services.SetEndpoint("ns", "svc", epOnNode(fmt.Sprintf("10.0.0.%d", i%250+1), "node-a")) + } + }() + + // Retry side. + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + } + ctrl.markPending("ns", "svc") + ctrl.retryPending() + } + }() + + time.Sleep(150 * time.Millisecond) + close(stop) + wg.Wait() +} From 0b10c267672d811316f6276c9f9a9a03f55758ab Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 15:27:59 +0200 Subject: [PATCH 08/11] controllers: retry failed datapath withdrawals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The retry queue only covered programming. A failed removal was logged and dropped, and it could not go through the service queue either: by the time a withdrawal runs the service is usually gone from the map, so there is nothing left to re-derive the pair from. Leaving that state behind is not merely untidy. A stale svc_pod entry keeps translating a service IP towards a pod that no longer exists, and a stale pod_svc entry rewrites the source of whatever pod next receives that IP — on a shared /16 with pod IPs recycling, one tenant's egress leaving under another tenant's service IP. A stale filtered_pods entry applies someone else's port list to the new occupant. Queue the pair explicitly instead. clearPortFilter, withdrawIngressRules and withdrawRules now return an error, the two withdraw paths queue themselves on failure, and the retry pass re-attempts them. A full withdrawal supersedes an ingress-only one for the same pair, so a node that also has to drop the source rewrite does not lose that. Withdrawals are retried before the applies, and programming a pair clears any removal queued for it, so an endpoint that flaps back to the pod IP whose withdrawal failed does not have its freshly installed rules deleted by the retry. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/controllers/services_controller.go | 118 ++++++++++++++++++-- pkg/controllers/services_controller_test.go | 98 +++++++++++++++- 2 files changed, 202 insertions(+), 14 deletions(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index 81a6bf9..dba9508 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -143,12 +143,60 @@ type ServicesController struct { // next event — long enough for a public IP to stay dark for minutes. pendingServices map[string]struct{} + // pendingWithdrawals holds datapath state that could not be removed. + // + // A failed withdrawal cannot go through pendingServices: by then the + // service is usually gone from Services, so there is nothing to re-derive + // the pair from. It is carried explicitly instead. Leaving it behind is + // not merely untidy — a stale pod_svc entry rewrites the source of + // whatever pod next receives that IP, which on a shared /16 means one + // tenant's egress leaving under another tenant's service IP. + pendingWithdrawals map[string]withdrawal + // pendingCleanup records that the startup reconciliation failed and has // to run again. It stays non-fatal, but it no longer waits for the next // event or the 12-hour informer resync. pendingCleanup bool } +// withdrawal is datapath state waiting to be removed. +type withdrawal struct { + svcIP, podIP string + // egress also withdraws the source rewrite. False when only the ingress + // half has to go, which is the case on a node that stopped hosting the + // backend but still carries the cluster-wide source rewrite. + egress bool +} + +// withdrawalKey identifies a pair independently of which half is pending. +func withdrawalKey(svcIP, podIP string) string { return svcIP + "/" + podIP } + +// markWithdrawal queues datapath state for another removal attempt. +func (c *ServicesController) markWithdrawal(svcIP, podIP string, egress bool) { + c.retryMu.Lock() + defer c.retryMu.Unlock() + if c.pendingWithdrawals == nil { + c.pendingWithdrawals = make(map[string]withdrawal) + } + k := withdrawalKey(svcIP, podIP) + // A full withdrawal supersedes an ingress-only one for the same pair. + if prev, ok := c.pendingWithdrawals[k]; ok && prev.egress { + egress = true + } + c.pendingWithdrawals[k] = withdrawal{svcIP: svcIP, podIP: podIP, egress: egress} +} + +// clearWithdrawal drops a queued removal. +// +// Called when the same pair is programmed again — an endpoint that flapped back +// to the pod IP whose withdrawal failed. Without this the retry would delete +// the rules that were just reinstalled. +func (c *ServicesController) clearWithdrawal(svcIP, podIP string) { + c.retryMu.Lock() + defer c.retryMu.Unlock() + delete(c.pendingWithdrawals, withdrawalKey(svcIP, podIP)) +} + // defaultRetryInterval is short enough that a failed write is repaired well // within a human noticing, and long enough that a permanent failure does not // flood the log. @@ -180,23 +228,33 @@ func (c *ServicesController) markCleanupPending() { // takePending returns the queued work and clears it. Anything that fails again // is re-queued by the attempt itself. -func (c *ServicesController) takePending() (services []string, cleanup bool) { +func (c *ServicesController) takePending() (services []string, withdrawals []withdrawal, cleanup bool) { c.retryMu.Lock() defer c.retryMu.Unlock() for k := range c.pendingServices { services = append(services, k) } c.pendingServices = nil + for _, w := range c.pendingWithdrawals { + withdrawals = append(withdrawals, w) + } + c.pendingWithdrawals = nil cleanup = c.pendingCleanup c.pendingCleanup = false - return services, cleanup + return services, withdrawals, cleanup } // retryPending re-attempts everything that failed since the last pass. Every // datapath call is idempotent, so a re-attempt on something already correct is // a no-op. func (c *ServicesController) retryPending() { - services, cleanup := c.takePending() + services, withdrawals, cleanup := c.takePending() + + // Withdrawals first: an apply for the same pair clears its queued removal + // at the moment it succeeds, so it wins either way. + for _, w := range withdrawals { + c.retryWithdrawal(w) + } for _, key := range services { ns, name, ok := splitKey(key) @@ -214,6 +272,21 @@ func (c *ServicesController) retryPending() { } } +// retryWithdrawal re-attempts a removal that failed. The pair is carried +// explicitly because the service it belonged to is usually gone by now. +func (c *ServicesController) retryWithdrawal(w withdrawal) { + c.reconcileMu.Lock() + defer c.reconcileMu.Unlock() + + log.Info("retrying datapath withdrawal", "svcIP", w.svcIP, "podIP", w.podIP, "egress", w.egress) + if w.egress { + // Re-queues itself on failure. + _ = c.withdrawRules(w.svcIP, w.podIP, "on retry") + return + } + _ = c.withdrawIngressRules(w.svcIP, w.podIP, "on retry") +} + // retryOne re-applies a single service, reading its state and programming the // datapath under the reconciliation lock so an informer callback cannot // interleave and have this attempt restore what it just withdrew. @@ -311,6 +384,9 @@ func (c *ServicesController) applyRules(svc *v1.Service, ep *v1.Endpoints, ctx s // the last place where the pod IP can still be turned back into the // service IP the client's conntrack is waiting for. Without it the reply // arrives with the wrong source and the client answers it with a RST. + // This pair is wanted again; drop any removal still queued for it. + c.clearWithdrawalFor(svc, ep) + failed := false if err := c.Proxy.EnsureEgressSNAT(svcIP, podIP); err != nil { log.Error(err, "failed to ensure egress SNAT "+ctx, "svcIP", svcIP, "podIP", podIP) @@ -319,7 +395,9 @@ func (c *ServicesController) applyRules(svc *v1.Service, ep *v1.Endpoints, ctx s // The destination rewrite and the port filter stay on the hosting node. if !c.servesEndpoint(ep) { - c.withdrawIngressRules(svcIP, podIP, ctx+" (backend not on this node)") + if err := c.withdrawIngressRules(svcIP, podIP, ctx+" (backend not on this node)"); err != nil { + failed = true + } c.recordOutcome(svc, failed) return } @@ -344,24 +422,42 @@ func (c *ServicesController) recordOutcome(svc *v1.Service, failed bool) { c.clearPending(svc.Namespace, svc.Name) } +// clearWithdrawalFor drops a queued removal for a pair that has just been +// programmed again, so the retry does not delete what was reinstalled. +func (c *ServicesController) clearWithdrawalFor(svc *v1.Service, ep *v1.Endpoints) { + c.clearWithdrawal(svc.Status.LoadBalancer.Ingress[0].IP, ep.Subsets[0].Addresses[0].IP) +} + // withdrawIngressRules removes the ingress half only — destination rewrite and // port filter — and leaves the source rewrite in place. This is what a node // that does not host the backend must end up with. -func (c *ServicesController) withdrawIngressRules(svcIP, podIP, ctx string) { - c.clearPortFilter(svcIP, podIP, ctx) +// It returns an error when any part could not be removed, so the caller can +// queue the pair for another attempt. +func (c *ServicesController) withdrawIngressRules(svcIP, podIP, ctx string) error { + failed := c.clearPortFilter(svcIP, podIP, ctx) if err := c.Proxy.DeleteIngressDNAT(svcIP, podIP); err != nil { log.Error(err, "failed to delete ingress DNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err + } + if failed != nil { + c.markWithdrawal(svcIP, podIP, false) } + return failed } // withdrawRules removes every datapath entry for the pair, both halves. Used // when the pair itself is going away: service deleted, endpoint gone, or a pod // IP that has been replaced. Absent entries are not an error. -func (c *ServicesController) withdrawRules(svcIP, podIP, ctx string) { - c.withdrawIngressRules(svcIP, podIP, ctx) +func (c *ServicesController) withdrawRules(svcIP, podIP, ctx string) error { + failed := c.withdrawIngressRules(svcIP, podIP, ctx) if err := c.Proxy.DeleteEgressSNAT(svcIP, podIP); err != nil { log.Error(err, "failed to delete egress SNAT "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err + } + if failed != nil { + c.markWithdrawal(svcIP, podIP, true) } + return failed } // withdrawStaleEndpoint drops the rules of a previous endpoint whose pod IP no @@ -850,13 +946,17 @@ func (c *ServicesController) reconcilePortFilter(svc *v1.Service, svcIP, podIP, // clearPortFilter unconditionally removes both port-filter and ICMP-allow // state for (svcIP, podIP). Used by delete paths. -func (c *ServicesController) clearPortFilter(svcIP, podIP, ctx string) { +func (c *ServicesController) clearPortFilter(svcIP, podIP, ctx string) error { + var failed error if err := c.Proxy.DeletePortFilter(svcIP, podIP); err != nil { log.Error(err, "failed to delete port filter "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } if err := c.Proxy.DeleteICMPAllow(svcIP, podIP); err != nil { log.Error(err, "failed to delete ICMP allow "+ctx, "svcIP", svcIP, "podIP", podIP) + failed = err } + return failed } // cleanupRemovedServices performs an initial cleanup for removed services. diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 03f388d..6233c91 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -53,6 +53,10 @@ type recordingProxy struct { failPortFilter bool // failCleanup makes CleanupRules fail. failCleanup bool + // failDeleteEgress makes DeleteEgressSNAT fail. + failDeleteEgress bool + // failDeleteIngress makes DeleteIngressDNAT fail. + failDeleteIngress bool } func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { @@ -65,6 +69,9 @@ func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { func (r *recordingProxy) DeleteEgressSNAT(svcIP, podIP string) error { r.record("DeleteEgressSNAT") + if r.failDeleteEgress { + return errors.New("commit refused") + } return nil } @@ -75,6 +82,9 @@ func (r *recordingProxy) EnsureIngressDNAT(svcIP, podIP string) error { func (r *recordingProxy) DeleteIngressDNAT(svcIP, podIP string) error { r.record("DeleteIngressDNAT") + if r.failDeleteIngress { + return errors.New("commit refused") + } return nil } @@ -338,7 +348,7 @@ func TestFailedProgrammingIsQueuedForRetry(t *testing.T) { ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} ctrl.applyRules(svc, ep, "test") - keys, _ := ctrl.takePending() + keys, _, _ := ctrl.takePending() if len(keys) != 1 || keys[0] != "ns/svc1" { t.Fatalf("failed write must be queued, got %v", keys) } @@ -346,7 +356,7 @@ func TestFailedProgrammingIsQueuedForRetry(t *testing.T) { // A clean pass clears it again. px.failEgress = false ctrl.applyRules(svc, ep, "test") - keys, _ = ctrl.takePending() + keys, _, _ = ctrl.takePending() if len(keys) != 0 { t.Errorf("clean pass must clear the queue, got %v", keys) } @@ -362,7 +372,7 @@ func TestFailedPortFilterIsQueuedForRetry(t *testing.T) { ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} ctrl.applyRules(svc, epOnNode("10.0.0.1", "node-a"), "test") - keys, _ := ctrl.takePending() + keys, _, _ := ctrl.takePending() if len(keys) != 1 || keys[0] != "ns/svc2" { t.Fatalf("failed port filter must be queued, got %v", keys) } @@ -389,7 +399,7 @@ func TestRetryPendingReappliesDatapath(t *testing.T) { t.Errorf("retry must call %s, got %v", want, px.calls) } } - if keys, _ := ctrl.takePending(); len(keys) != 0 { + if keys, _, _ := ctrl.takePending(); len(keys) != 0 { t.Errorf("successful retry must clear the queue, got %v", keys) } } @@ -426,7 +436,7 @@ func TestFailedCleanupIsQueuedAndRetried(t *testing.T) { if !px.has("CleanupRules") { t.Errorf("cleanup must be retried, got %v", px.calls) } - if _, pending := ctrl.takePending(); pending { + if _, _, pending := ctrl.takePending(); pending { t.Error("successful cleanup retry must clear the flag") } } @@ -505,3 +515,81 @@ func TestRetryIsSerializedAgainstEndpointUpdates(t *testing.T) { close(stop) wg.Wait() } + +// A withdrawal that failed cannot go through the service queue: by then the +// service is gone from the map. Left behind, a stale pod_svc entry rewrites the +// source of whatever pod next receives that IP. +func TestFailedWithdrawalIsQueuedAndRetried(t *testing.T) { + px := &recordingProxy{failDeleteEgress: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + + if err := ctrl.withdrawRules("192.0.2.10", "10.0.0.1", "test"); err == nil { + t.Fatal("withdrawRules must surface the failure") + } + _, pending, _ := ctrl.takePending() + if len(pending) != 1 || pending[0].svcIP != "192.0.2.10" || !pending[0].egress { + t.Fatalf("failed withdrawal must be queued as a full one, got %+v", pending) + } + + // Re-queue it and let the retry succeed. + ctrl.markWithdrawal("192.0.2.10", "10.0.0.1", true) + px.failDeleteEgress = false + px.calls = nil + ctrl.retryPending() + + for _, want := range []string{"DeleteIngressDNAT", "DeleteEgressSNAT"} { + if !px.has(want) { + t.Errorf("retry must call %s, got %v", want, px.calls) + } + } + if _, pending, _ := ctrl.takePending(); len(pending) != 0 { + t.Errorf("successful retry must clear the queue, got %+v", pending) + } +} + +// An endpoint that flaps back to the pod IP whose withdrawal failed must not +// have its freshly installed rules deleted by the queued removal. +func TestReprogrammingClearsQueuedWithdrawal(t *testing.T) { + svc := lbService("192.0.2.10", map[string]string{"networking.cozystack.io/wholeIP": "false"}) + svc.Namespace, svc.Name = "ns", "svc" + ep := epOnNode("10.0.0.1", "node-a") + + px := &recordingProxy{} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.markWithdrawal("192.0.2.10", "10.0.0.1", true) + + ctrl.applyRules(svc, ep, "test") + + if _, pending, _ := ctrl.takePending(); len(pending) != 0 { + t.Errorf("programming the pair again must drop its queued removal, got %+v", pending) + } +} + +// A node that stops hosting the backend withdraws only the ingress half, and +// must keep the cluster-wide source rewrite. +func TestIngressOnlyWithdrawalIsQueuedWithoutEgress(t *testing.T) { + px := &recordingProxy{failDeleteIngress: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + + if err := ctrl.withdrawIngressRules("192.0.2.10", "10.0.0.1", "test"); err == nil { + t.Fatal("withdrawIngressRules must surface the failure") + } + _, pending, _ := ctrl.takePending() + if len(pending) != 1 || pending[0].egress { + t.Fatalf("ingress-only failure must not queue an egress withdrawal, got %+v", pending) + } +} + +// A full withdrawal supersedes an ingress-only one already queued for the pair. +func TestFullWithdrawalSupersedesIngressOnly(t *testing.T) { + ctrl := &ServicesController{Proxy: &recordingProxy{}} + ctrl.markWithdrawal("192.0.2.10", "10.0.0.1", true) + ctrl.markWithdrawal("192.0.2.10", "10.0.0.1", false) + + _, pending, _ := ctrl.takePending() + if len(pending) != 1 || !pending[0].egress { + t.Fatalf("full withdrawal must win, got %+v", pending) + } +} From c78640c707fb767cbe1fa6f6d2c7839d24edb77d Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 16:04:23 +0200 Subject: [PATCH 09/11] controllers: build the startup snapshot from the informer stores MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found on a lab restart, not by reading the code: the startup purge marked four live pod_svc mappings stale and deleted them, then reported podSvcAdded: 0. The datapath was only correct again because the endpoint callbacks happened to fire a second time afterwards. WaitForCacheSync returns once the store is populated, not once every initial callback has run. cleanupRemovedServices read Services, which those callbacks fill, so it computed its desired state from a half filled map and treated everything missing from it as stale — including the cluster-wide source rewrites belonging to backends on other nodes. Nothing guarantees a second callback: a service whose mapping was deleted that way waits for an event, or for the 12-hour resync. Read the informer stores instead, which are authoritative as soon as the sync returns. Services stays the fallback when no informer is attached, which is how the unit tests drive the controller. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/controllers/services_controller.go | 46 +++++++++++++- pkg/controllers/services_controller_test.go | 66 +++++++++++++++++++++ 2 files changed, 110 insertions(+), 2 deletions(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index dba9508..31a218b 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -120,6 +120,15 @@ type ServicesController struct { // everything everywhere. NodeName string + // svcStore and epStore are the informer stores. The startup snapshot is + // built from them rather than from Services, because WaitForCacheSync + // returns once the store is populated, not once every initial callback + // has run. Reading the half-filled map made the purge treat a live + // mapping as stale and delete it, leaving the node without it until an + // event happened to re-apply that service — or until the 12-hour resync. + svcStore cache.Store + epStore cache.Store + // RetryInterval is how often failed datapath writes are re-attempted. // Zero selects defaultRetryInterval. RetryInterval time.Duration @@ -509,6 +518,8 @@ func (c *ServicesController) Start(ctx context.Context) error { }, ) + c.svcStore = serviceInformer.GetStore() + serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: c.addServiceFunc, DeleteFunc: c.deleteServiceFunc, @@ -551,6 +562,8 @@ func (c *ServicesController) Start(ctx context.Context) error { }, ) + c.epStore = endpointsInformer.GetStore() + endpointsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: c.addEndpointFunc, DeleteFunc: c.deleteEndpointFunc, @@ -959,6 +972,36 @@ func (c *ServicesController) clearPortFilter(svcIP, podIP, ctx string) error { return failed } +// snapshotSource returns the pairs the datapath should hold. +// +// It reads the informer stores when they are available, because they are +// authoritative as soon as WaitForCacheSync returns, whereas Services is +// filled by callbacks that may not have run yet. Falling back to Services +// keeps the controller usable without informers, which is how the unit tests +// drive it. +func (c *ServicesController) snapshotSource() map[string]*ServiceEndpoints { + if c.svcStore == nil || c.epStore == nil { + return c.Services.GetAll() + } + out := make(map[string]*ServiceEndpoints) + for _, obj := range c.svcStore.List() { + svc, ok := obj.(*v1.Service) + if !ok || !isCozyProxyService(svc) || !hasValidServiceIP(svc) { + continue + } + epObj, exists, err := c.epStore.GetByKey(makeKey(svc.Namespace, svc.Name)) + if err != nil || !exists { + continue + } + ep, ok := epObj.(*v1.Endpoints) + if !ok || !hasValidEndpointIP(ep) { + continue + } + out[makeKey(svc.Namespace, svc.Name)] = &ServiceEndpoints{Service: svc, Endpoint: ep} + } + return out +} + // cleanupRemovedServices performs an initial cleanup for removed services. func (c *ServicesController) cleanupRemovedServices() error { // Reconciliations are serialized end to end, see reconcileMu. @@ -972,8 +1015,7 @@ func (c *ServicesController) cleanupRemovedServices() error { // scoped both maps alike. keepEgress := make(map[string]string) keepIngress := make(map[string]string) - // Get a snapshot of all managed services. - allServices := c.Services.GetAll() + allServices := c.snapshotSource() for _, serviceEndpoints := range allServices { if serviceEndpoints.Service == nil || serviceEndpoints.Endpoint == nil { continue diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 6233c91..4eaa39f 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -10,6 +10,8 @@ import ( v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/tools/cache" + nat "github.com/cozystack/cozy-proxy/pkg/proxy" ) @@ -593,3 +595,67 @@ func TestFullWithdrawalSupersedesIngressOnly(t *testing.T) { t.Fatalf("full withdrawal must win, got %+v", pending) } } + +// The startup snapshot must come from the informer stores, not from the map the +// callbacks fill. WaitForCacheSync returns once the store is populated, not +// once every initial callback has run, and reading the half-filled map made the +// purge delete a live mapping it then had to wait for an event to restore. +func TestSnapshotSourcePrefersInformerStores(t *testing.T) { + managed := lbService("192.0.2.10", nil) + managed.Namespace, managed.Name = "ns", "managed" + managed.Labels = map[string]string{"service.kubernetes.io/service-proxy-name": "cozy-proxy"} + + unmanaged := lbService("192.0.2.11", nil) + unmanaged.Namespace, unmanaged.Name = "ns", "unmanaged" + + noIP := &v1.Service{ObjectMeta: metav1.ObjectMeta{ + Namespace: "ns", Name: "noip", + Labels: map[string]string{"service.kubernetes.io/service-proxy-name": "cozy-proxy"}, + }} + + noEndpoint := lbService("192.0.2.12", nil) + noEndpoint.Namespace, noEndpoint.Name = "ns", "noep" + noEndpoint.Labels = map[string]string{"service.kubernetes.io/service-proxy-name": "cozy-proxy"} + + svcStore := cache.NewStore(cache.MetaNamespaceKeyFunc) + for _, s := range []*v1.Service{managed, unmanaged, noIP, noEndpoint} { + if err := svcStore.Add(s); err != nil { + t.Fatalf("seeding service store: %v", err) + } + } + + ep := epOnNode("10.0.0.1", "node-a") + ep.Namespace, ep.Name = "ns", "managed" + epStore := cache.NewStore(cache.MetaNamespaceKeyFunc) + if err := epStore.Add(ep); err != nil { + t.Fatalf("seeding endpoint store: %v", err) + } + + ctrl := &ServicesController{Proxy: &recordingProxy{}, svcStore: svcStore, epStore: epStore} + // Deliberately empty: this is the map the callbacks had not filled yet. + ctrl.Services = NewServiceMap() + + got := ctrl.snapshotSource() + if len(got) != 1 { + t.Fatalf("snapshot must hold only the managed service with a valid IP and endpoint, got %d: %v", len(got), got) + } + se, ok := got["ns/managed"] + if !ok || se.Service.Name != "managed" || se.Endpoint.Subsets[0].Addresses[0].IP != "10.0.0.1" { + t.Errorf("unexpected snapshot entry: %+v", got) + } +} + +// Without informers the controller still has to work, which is how the unit +// tests drive it. +func TestSnapshotSourceFallsBackToServiceMap(t *testing.T) { + svc := lbService("192.0.2.10", nil) + svc.Namespace, svc.Name = "ns", "svc" + + ctrl := &ServicesController{Proxy: &recordingProxy{}} + ctrl.Services = NewServiceMap() + ctrl.Services.Set("ns", "svc", &ServiceEndpoints{Service: svc, Endpoint: epOnNode("10.0.0.1", "node-a")}) + + if got := ctrl.snapshotSource(); len(got) != 1 { + t.Errorf("fallback must read the service map, got %v", got) + } +} From 494dbf7b322dcc7280c97752da3b4ba289511c22 Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Tue, 15 Sep 2026 16:35:53 +0200 Subject: [PATCH 10/11] controllers: requeue the startup cleanup when its retry fails MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit takePending clears the cleanup flag before the retry runs, and the error branch only logged — so the reconciliation was attempted exactly once and then forgotten, while the message claimed it would try again. The per-service and withdrawal paths do not have this gap: they re-queue themselves from inside the call that failed. Put the flag back, and say what actually happens. Signed-off-by: Mattia Eleuteri Co-Authored-By: Claude Opus 5 (1M context) --- pkg/controllers/services_controller.go | 7 ++++++- pkg/controllers/services_controller_test.go | 23 +++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/pkg/controllers/services_controller.go b/pkg/controllers/services_controller.go index 31a218b..baccb3c 100644 --- a/pkg/controllers/services_controller.go +++ b/pkg/controllers/services_controller.go @@ -276,7 +276,12 @@ func (c *ServicesController) retryPending() { if cleanup { log.Info("retrying startup cleanup") if err := c.cleanupRemovedServices(); err != nil { - log.Error(err, "cleanup retry failed, will try again") + // takePending cleared the flag, and unlike the per-service and + // withdrawal paths — which re-queue themselves from inside the + // call that failed — nothing here would put it back. Without this + // the reconciliation is retried exactly once. + c.markCleanupPending() + log.Error(err, "cleanup retry failed, queued again") } } } diff --git a/pkg/controllers/services_controller_test.go b/pkg/controllers/services_controller_test.go index 4eaa39f..0b734bd 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -659,3 +659,26 @@ func TestSnapshotSourceFallsBackToServiceMap(t *testing.T) { t.Errorf("fallback must read the service map, got %v", got) } } + +// takePending clears the cleanup flag, so a retry that fails again has to put +// it back — otherwise the reconciliation is attempted exactly once and the +// stale state it exists to remove stays for good. +func TestFailedCleanupRetryIsQueuedAgain(t *testing.T) { + px := &recordingProxy{failCleanup: true} + ctrl := &ServicesController{Proxy: px, NodeName: "node-a"} + ctrl.Services = NewServiceMap() + ctrl.markCleanupPending() + + ctrl.retryPending() + if _, _, pending := ctrl.takePending(); !pending { + t.Fatal("a cleanup retry that failed must be queued again") + } + + // And it stops being queued once it succeeds. + ctrl.markCleanupPending() + px.failCleanup = false + ctrl.retryPending() + if _, _, pending := ctrl.takePending(); pending { + t.Error("a successful cleanup retry must clear the flag") + } +} From 9f5a54670f543302d4fc3d16f7306377356863aa Mon Sep 17 00:00:00 2001 From: Mattia Eleuteri Date: Thu, 1 Oct 2026 18:33:44 +0200 Subject: [PATCH 11/11] chart: inject NODE_NAME into the DaemonSet The binary scopes ingress rules to the node hosting the backend, and learns that node from NODE_NAME, but the chart never set it. Every supported install therefore fell back to programming every service on every node, which leaves the cross-node hang this branch fixes still reachable. Set it from spec.nodeName through the downward API. This used to live in cozystack's vendored copy of the chart (cozystack/cozystack#3610), where the next chart update would have dropped it. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Mattia Eleuteri --- README.md | 7 ++++--- charts/cozy-proxy/templates/daemonset.yaml | 8 ++++++++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 9c8705c..0360a86 100644 --- a/README.md +++ b/README.md @@ -80,9 +80,10 @@ watches every Service, but programs: | `svc_pod` (`ingress_dnat`), `allowed_ports`, `icmp_allowed_pods` | the node hosting the backend pod | A non-hosting node would rewrite the destination before the packet has even left it. The hosting node then records a conntrack tuple of `(client -> podIP)`, while the reply leaves the pod and gets `saddr` rewritten to the service IP by `egress_snat` at priority `raw`, before conntrack. `(svcIP -> client)` matches nothing, so the reply is not `established` and `port_filter` drops it. | | `pod_svc` (`egress_snat`) | every node | An intra-cluster client is source-NATed by the CNI to its own node address, which the overlay knows how to reach directly. The backend's reply is then tunnelled straight to that node and never traverses the hosting node's netfilter hooks, so the client's node is the only place left where the pod IP can still be turned back into the service IP. Without the entry there, the reply arrives with the wrong source and the client answers with a RST — a cross-node connection that hangs while the same-node one works. | -`NODE_NAME` carries the node identity. When it is unset the ingress scope -check is disabled and everything is programmed everywhere, so the binary still -runs under a chart that does not inject it. +`NODE_NAME` carries the node identity; the chart injects it from +`spec.nodeName`. When it is unset the ingress scope check is disabled and +everything is programmed everywhere, so the binary still runs under a +deployment that does not inject it. ## Installation diff --git a/charts/cozy-proxy/templates/daemonset.yaml b/charts/cozy-proxy/templates/daemonset.yaml index df774bf..c06bdfa 100644 --- a/charts/cozy-proxy/templates/daemonset.yaml +++ b/charts/cozy-proxy/templates/daemonset.yaml @@ -21,6 +21,14 @@ spec: - name: cozy-proxy image: "{{ .Values.image.repository }}:{{ .Values.image.tag }}" imagePullPolicy: {{ .Values.image.pullPolicy }} + env: + # Scopes the ingress rules to the backends hosted on this node; see + # "Rule scoping" in the README for why a non-hosting node must not + # program them. + - name: NODE_NAME + valueFrom: + fieldRef: + fieldPath: spec.nodeName securityContext: privileged: true capabilities: