diff --git a/README.md b/README.md index 4c497e1..0360a86 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,21 @@ 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; 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 Install controller using Helm-chart: 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: diff --git a/go.mod b/go.mod index 2be1fb2..120ccaf 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.4 require ( github.com/google/nftables v0.3.0 + github.com/vishvananda/netns v0.0.5 golang.org/x/sys v0.46.0 k8s.io/api v0.36.2 k8s.io/apimachinery v0.36.2 diff --git a/go.sum b/go.sum index 9564083..25087a5 100644 --- a/go.sum +++ b/go.sum @@ -120,8 +120,8 @@ github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8= -github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= +github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zdEY= +github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= 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..baccb3c 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" @@ -55,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() @@ -96,6 +113,380 @@ type ServicesController struct { Clientset *kubernetes.Clientset Services *ServiceMap Proxy nat.ProxyProcessor + + // 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 + + // 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 + + // 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 + + // 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{} + + // 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. +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, 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, 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, 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) + if !ok { + continue + } + c.retryOne(ns, name, key) + } + + if cleanup { + log.Info("retrying startup cleanup") + if err := c.cleanupRemovedServices(); err != nil { + // 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") + } + } +} + +// 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. +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 + 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. +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 hosts ep's backend, and may +// therefore program the ingress half of the datapath. +// +// 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 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 + } + node, ok := endpointNode(ep) + if !ok { + return true + } + return node == c.NodeName +} + +// 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. + // 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) + failed = true + } + + // The destination rewrite and the port filter stay on the hosting node. + if !c.servesEndpoint(ep) { + if err := c.withdrawIngressRules(svcIP, podIP, ctx+" (backend not on this node)"); err != nil { + failed = true + } + 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.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. +// 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) 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 +// 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. @@ -132,6 +523,8 @@ func (c *ServicesController) Start(ctx context.Context) error { }, ) + c.svcStore = serviceInformer.GetStore() + serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: c.addServiceFunc, DeleteFunc: c.deleteServiceFunc, @@ -174,6 +567,8 @@ func (c *ServicesController) Start(ctx context.Context) error { }, ) + c.epStore = endpointsInformer.GetStore() + endpointsInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: c.addEndpointFunc, DeleteFunc: c.deleteEndpointFunc, @@ -190,12 +585,21 @@ 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, queued for retry") + c.markCleanupPending() + } else { + log.Info("cleanup of removed services completed") } - 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") @@ -205,6 +609,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. @@ -231,16 +639,16 @@ 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") } } // 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 @@ -258,13 +666,16 @@ 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) } // 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 { @@ -278,8 +689,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) } @@ -292,8 +702,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) } @@ -319,9 +728,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}) @@ -330,10 +740,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}) @@ -341,6 +751,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 { @@ -360,15 +774,17 @@ 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") } } // 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 @@ -385,14 +801,17 @@ 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) } // 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 @@ -408,8 +827,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 @@ -420,10 +838,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) } @@ -511,64 +927,118 @@ 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 // 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 +} + +// 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 { - keepMap := make(map[string]string) - // Get a snapshot of all managed services. - allServices := c.Services.GetAll() + // 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 + // 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) + allServices := c.snapshotSource() for _, serviceEndpoints := range allServices { - if serviceEndpoints.Service != nil && serviceEndpoints.Endpoint != nil { - var serviceIP, endpointIP string + 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. @@ -582,7 +1052,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 +1073,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..0b734bd 100644 --- a/pkg/controllers/services_controller_test.go +++ b/pkg/controllers/services_controller_test.go @@ -1,12 +1,226 @@ package controllers import ( + "errors" + "fmt" + "sync" "testing" + "time" 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" ) +// 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 + mu sync.Mutex + 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 + // failDeleteEgress makes DeleteEgressSNAT fail. + failDeleteEgress bool + // failDeleteIngress makes DeleteIngressDNAT fail. + failDeleteIngress bool +} + +func (r *recordingProxy) EnsureEgressSNAT(svcIP, podIP string) error { + r.record("EnsureEgressSNAT") + if r.failEgress { + return errors.New("commit refused") + } + return nil +} + +func (r *recordingProxy) DeleteEgressSNAT(svcIP, podIP string) error { + r.record("DeleteEgressSNAT") + if r.failDeleteEgress { + return errors.New("commit refused") + } + return nil +} + +func (r *recordingProxy) EnsureIngressDNAT(svcIP, podIP string) error { + r.record("EnsureIngressDNAT") + return nil +} + +func (r *recordingProxy) DeleteIngressDNAT(svcIP, podIP string) error { + r.record("DeleteIngressDNAT") + if r.failDeleteIngress { + return errors.New("commit refused") + } + return nil +} + +func (r *recordingProxy) CleanupRules(keepEgress, keepIngress map[string]string) error { + r.record("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.record("EnsurePortFilter") + if r.failPortFilter { + return errors.New("commit refused") + } + return nil +} + +func (r *recordingProxy) DeletePortFilter(svcIP, podIP string) error { + 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 + } + } + 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) + } + }) + } +} + +// 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") + + 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("EnsureEgressSNAT") { + t.Errorf("non-owning node must still program the source rewrite, 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) + } +} + +// 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") + // 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{} + 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}} } @@ -78,3 +292,393 @@ 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) + } +} + +// 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) + } + } +} + +// 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() +} + +// 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) + } +} + +// 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) + } +} + +// 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") + } +} 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 afe72ac..e66e2f4 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. @@ -387,244 +403,370 @@ 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) +// 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 { + p.mu.Lock() + defer p.mu.Unlock() - 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) - } + 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) + } + // 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 // 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 { + p.mu.Lock() + defer p.mu.Unlock() - // 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) + log.Info("Deleting egress SNAT", "svcIP", svcIP, "podIP", 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 { + p.mu.Lock() + defer p.mu.Unlock() + + 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) + } + // 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 + } + + 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 { + p.mu.Lock() + defer p.mu.Unlock() + + 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) } + 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("Ingress DNAT deleted successfully", "svcIP", svcIP, "podIP", podIP) + return nil +} - // 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) - } +// 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 +// 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 { + return nil + } + if errors.Is(err, unix.ENOENT) { + log.Info("Ignoring ENOENT on flush — element already gone", "op", op) + return nil + } + 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("NAT mapping and raw set elements deleted successfully", "svcIP", svcIP, "podIP", podIP) + 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 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 { + p.mu.Lock() + defer p.mu.Unlock() - // --- Step 1: Clean up mapping sets --- + log.Info("Starting CleanupRules", "keepEgress", keepEgress, "keepIngress", keepIngress) - // Retrieve current mappings from the pod_svc map. - // Note: pod_svc maps pod IP → svc IP. - podSvcElems, err := p.conn.GetSetElements(p.podSvcMap) + // --- Step 1: collect what has to change in both maps --- + + // 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 { + return fmt.Errorf("failed to diff pod_svc: %v", err) + } + svcPodDel, svcPodAdd, err := p.diffMap(p.svcPodMap, keepIngress) 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 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) - } - log.Info("Inconsistent mappings removed from both maps") + // --- 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 err := p.deleteElementsTolerant(p.podSvcMap, podSvcDel, "CleanupRules"); err != nil { + log.Error(err, "Failed to delete stale mappings from pod_svc") + return 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 { - 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. + // 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) } - 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 @@ -632,13 +774,16 @@ func (p *NFTProxyProcessor) CleanupRules(keepMap map[string]string) error { // 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)) @@ -648,10 +793,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 { @@ -691,6 +842,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 { @@ -722,6 +883,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. @@ -793,16 +957,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) - } + // 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 } - 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) - } + 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 { @@ -815,7 +977,9 @@ func (p *NFTProxyProcessor) CleanupPortFilters(keep map[string]PortFilterEntry) } } - // 5. Single flush. + // 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) } @@ -865,6 +1029,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 { @@ -883,6 +1050,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 { @@ -909,6 +1079,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)) @@ -944,10 +1117,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) - } + 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 { @@ -955,6 +1126,7 @@ func (p *NFTProxyProcessor) CleanupICMPAllow(keep map[string]string) error { } } + // Additions committed strictly, see CleanupPortFilters. if err := p.conn.Flush(); err != nil { return fmt.Errorf("failed to flush CleanupICMPAllow: %v", err) } diff --git a/pkg/proxy/nft_datapath_test.go b/pkg/proxy/nft_datapath_test.go new file mode 100644 index 0000000..e51af61 --- /dev/null +++ b/pkg/proxy/nft_datapath_test.go @@ -0,0 +1,181 @@ +//go:build linux + +package proxy + +import ( + "net" + "os" + "runtime" + "testing" + + "github.com/google/nftables" + "github.com/vishvananda/netns" +) + +// These tests program a real kernel, in a scratch network namespace, so they +// exercise how nftables actually commits a batch rather than how the code +// assumes it does. They need root and skip otherwise. + +// scratchNetns creates a network namespace that lives as long as the test, so +// nothing here ever touches the host's ruleset. +func scratchNetns(t *testing.T) netns.NsHandle { + t.Helper() + if os.Geteuid() != 0 { + t.Skip("needs root to create a network namespace") + } + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + orig, err := netns.Get() + if err != nil { + t.Fatalf("netns.Get: %v", err) + } + defer func() { _ = orig.Close() }() + ns, err := netns.New() // also switches this thread into it + if err != nil { + t.Skipf("cannot create a network namespace: %v", err) + } + if err := netns.Set(orig); err != nil { + t.Fatalf("netns.Set: %v", err) + } + t.Cleanup(func() { _ = ns.Close() }) + return ns +} + +// initializedProcessor returns a processor whose table is programmed in a +// scratch namespace. +func initializedProcessor(t *testing.T) *NFTProxyProcessor { + t.Helper() + ns := scratchNetns(t) + conn, err := nftables.New(nftables.WithNetNSFd(int(ns))) + if err != nil { + t.Fatalf("nftables.New: %v", err) + } + p := &NFTProxyProcessor{conn: conn} + if err := p.InitRules(); err != nil { + t.Fatalf("InitRules: %v", err) + } + return p +} + +func ipElem(key, val string) nftables.SetElement { + return nftables.SetElement{Key: net.ParseIP(key).To4(), Val: net.ParseIP(val).To4()} +} + +func seedMap(t *testing.T, p *NFTProxyProcessor, m *nftables.Set, kv map[string]string) { + t.Helper() + var elems []nftables.SetElement + for k, v := range kv { + elems = append(elems, ipElem(k, v)) + } + if err := p.conn.SetAddElements(m, elems); err != nil { + t.Fatalf("queue %s: %v", m.Name, err) + } + if err := p.conn.Flush(); err != nil { + t.Fatalf("seed %s: %v", m.Name, err) + } +} + +func readMap(t *testing.T, p *NFTProxyProcessor, m *nftables.Set) map[string]string { + t.Helper() + elems, err := p.conn.GetSetElements(m) + if err != nil { + t.Fatalf("read %s: %v", m.Name, err) + } + got := make(map[string]string, len(elems)) + for _, el := range elems { + got[net.IP(el.Key).String()] = net.IP(el.Val).String() + } + return got +} + +func assertMap(t *testing.T, p *NFTProxyProcessor, m *nftables.Set, want map[string]string) { + t.Helper() + got := readMap(t, p, m) + if len(got) != len(want) { + t.Errorf("%s = %v, want %v", m.Name, got, want) + return + } + for k, v := range want { + if got[k] != v { + t.Errorf("%s = %v, want %v", m.Name, got, want) + return + } + } +} + +// TestCleanupRulesReconcilesEachMapAgainstItsOwnScope seeds the state a node +// inherits from a build that scoped both maps alike, and checks each map ends +// up holding exactly its own keep set: pod_svc every managed service, svc_pod +// only the backend hosted here. +func TestCleanupRulesReconcilesEachMapAgainstItsOwnScope(t *testing.T) { + p := initializedProcessor(t) + + const ( + svcLocal, podLocal = "198.51.100.1", "10.244.0.1" // backend on this node + svcRemote, podRemote = "198.51.100.2", "10.244.1.2" // backend elsewhere + svcGone, podGone = "198.51.100.3", "10.244.1.3" // service deleted + svcNew, podNew = "198.51.100.4", "10.244.0.4" // not programmed yet + ) + seedMap(t, p, p.podSvcMap, map[string]string{ + podLocal: svcLocal, podRemote: svcRemote, podGone: svcGone, + }) + seedMap(t, p, p.svcPodMap, map[string]string{ + svcLocal: podLocal, svcRemote: podRemote, svcGone: podGone, + }) + + keepEgress := map[string]string{svcLocal: podLocal, svcRemote: podRemote, svcNew: podNew} + keepIngress := map[string]string{svcLocal: podLocal, svcNew: podNew} + if err := p.CleanupRules(keepEgress, keepIngress); err != nil { + t.Fatalf("CleanupRules: %v", err) + } + + assertMap(t, p, p.podSvcMap, map[string]string{ + podLocal: svcLocal, podRemote: svcRemote, podNew: svcNew, + }) + assertMap(t, p, p.svcPodMap, map[string]string{ + svcLocal: podLocal, svcNew: podNew, + }) +} + +// TestDeleteElementsTolerantSurvivesAMissingElement queues a deletion batch +// in which one element is already gone. The kernel aborts the whole +// transaction on it, so the elements that are present must still be removed +// by the per-element retry, not reported as removed while left in place. +func TestDeleteElementsTolerantSurvivesAMissingElement(t *testing.T) { + p := initializedProcessor(t) + + seedMap(t, p, p.podSvcMap, map[string]string{ + "10.244.0.1": "198.51.100.1", + "10.244.0.2": "198.51.100.2", + "10.244.0.9": "198.51.100.9", + }) + + del := []nftables.SetElement{ + ipElem("10.244.0.1", "198.51.100.1"), + ipElem("10.244.0.5", "198.51.100.5"), // never there + ipElem("10.244.0.2", "198.51.100.2"), + } + if err := p.deleteElementsTolerant(p.podSvcMap, del, "test"); err != nil { + t.Fatalf("deleteElementsTolerant: %v", err) + } + + assertMap(t, p, p.podSvcMap, map[string]string{"10.244.0.9": "198.51.100.9"}) +} + +// TestDeleteElementsTolerantReportsOtherFailures checks that only ENOENT is +// tolerated: a batch the kernel rejects for another reason must surface. +func TestDeleteElementsTolerantReportsOtherFailures(t *testing.T) { + p := initializedProcessor(t) + + seedMap(t, p, p.podSvcMap, map[string]string{"10.244.0.1": "198.51.100.1"}) + + // A 2-byte key cannot belong to an ipv4_addr map. + del := []nftables.SetElement{{Key: []byte{10, 244}}} + if err := p.deleteElementsTolerant(p.podSvcMap, del, "test"); err == nil { + t.Fatal("deleteElementsTolerant succeeded on a malformed element") + } + + assertMap(t, p, p.podSvcMap, map[string]string{"10.244.0.1": "198.51.100.1"}) +}