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..a38c1ae 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. @@ -190,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") @@ -231,11 +311,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 +406,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 +436,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 +494,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 +627,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 +661,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 +682,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}} } 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",