proxy: scope datapath rules to the node hosting the backend

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 {

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

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}}
 }