imp: add stale flows cleanup

clenaup will be done in case the controller will fail to delete flows

cleanup logic will list all pods on a local node (using specific cache)
for each bridge will check external ids related to rail-cni -  pod-uuid=rail_pod_id
if the pod doesn't exit it means that there are stale flows
creating a cookie from the pod uuid
removing the flows
removing external-id from the bridge

Signed-off-by: Michael Filanov <[email protected]>

Author: filanov

cmd/flowscontroller/main.go

@@ -21,8 +21,10 @@ import (
 	"crypto/tls"
 	"flag"
 	"os"
+	"time"
 
 	"github.com/Mellanox/spectrum-x-operator/internal/controller"
+	"github.com/Mellanox/spectrum-x-operator/internal/staleflows"
 	"github.com/Mellanox/spectrum-x-operator/pkg/exec"
 	"github.com/Mellanox/spectrum-x-operator/pkg/filewatcher"
 	"github.com/Mellanox/spectrum-x-operator/pkg/lib/netlink"
@@ -160,6 +162,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	staleFlowsCleaner := &staleflows.Cleaner{
+		Client:          mgr.GetClient(),
+		Flows:           &controller.Flows{Exec: &exec.Exec{}, NetlinkLib: netlink.New()},
+		CleanupInterval: 5 * time.Minute,
+	}
+
+	staleFlowsCleaner.StartCleanupRoutine(context.Background())
+
 	//+kubebuilder:scaffold:builder
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {

internal/controller/flows.go

@@ -16,13 +16,15 @@ limitations under the License.
 package controller
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	"github.com/Mellanox/spectrum-x-operator/pkg/exec"
 	libnetlink "github.com/Mellanox/spectrum-x-operator/pkg/lib/netlink"
 
 	"go.uber.org/multierr"
+	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 const (
@@ -38,6 +40,7 @@ type FlowsAPI interface {
 	AddPodRailFlows(cookie uint64, vf, bridge, podIP, podMAC string) error
 	DeletePodRailFlows(cookie uint64, podID string) error
 	IsBridgeManagedByRailCNI(bridge, podID string) (bool, error)
+	CleanupStaleFlowsForBridges(ctx context.Context, existingPodUIDs map[string]bool) error
 }
 
 var _ FlowsAPI = &Flows{}
@@ -125,23 +128,29 @@ func (f *Flows) DeletePodRailFlows(cookie uint64, podID string) error {
 		}
 
 		if val == RailPodID {
-			flow := fmt.Sprintf(`ovs-ofctl del-flows %s cookie=0x%x/-1`, bridge, cookie)
-			out, err := f.Exec.Execute(flow)
-			if err != nil {
-				errs = multierr.Append(errs, fmt.Errorf("failed to delete flows for bridge %s: %v, output: %s", bridge, err, out))
-				continue
-			}
-			// clear external id
-			_, err = f.Exec.Execute(fmt.Sprintf(`ovs-vsctl br-set-external-id %s %s ""`, bridge, podID))
-			if err != nil {
-				errs = multierr.Append(errs, fmt.Errorf("failed to clear external id for bridge %s: %v", bridge, err))
+			if err := f.cleanBridgeFlows(bridge, podID, cookie); err != nil {
+				errs = multierr.Append(errs, err)
 				continue
 			}
 		}
 	}
 	return errs
 }
 
+func (f *Flows) cleanBridgeFlows(bridge, podID string, cookie uint64) error {
+	flow := fmt.Sprintf("ovs-ofctl del-flows %s cookie=0x%x/-1", bridge, cookie)
+	out, err := f.Exec.Execute(flow)
+	if err != nil {
+		return fmt.Errorf("failed to delete flows for bridge %s: %v, output: %s", bridge, err, out)
+	}
+	// clear external id
+	_, err = f.Exec.Execute(fmt.Sprintf("ovs-vsctl remove bridge %s external_ids %s", bridge, podID))
+	if err != nil {
+		return fmt.Errorf("failed to clear external id for bridge %s: %v", bridge, err)
+	}
+	return nil
+}
+
 func (f *Flows) IsBridgeManagedByRailCNI(bridge, podID string) (bool, error) {
 	out, err := f.Exec.Execute(fmt.Sprintf("ovs-vsctl br-get-external-id %s %s", bridge, podID))
 	if err != nil {
@@ -168,3 +177,68 @@ func (f *Flows) getTorMac(torIP string) (string, error) {
 
 	return reply, nil
 }
+
+func (f *Flows) CleanupStaleFlowsForBridges(ctx context.Context, existingPodUIDs map[string]bool) error {
+	logr := log.FromContext(ctx)
+
+	// List all bridges once
+	out, err := f.Exec.Execute("ovs-vsctl list-br")
+	if err != nil {
+		return fmt.Errorf("failed to list bridges: out: %s, err: %v", out, err)
+	}
+
+	bridges := strings.Split(out, "\n")
+	var errs error
+
+	for _, bridge := range bridges {
+		// Get all external IDs for this bridge once
+		externalIDsOut, err := f.Exec.Execute(fmt.Sprintf("ovs-vsctl br-get-external-id %s", bridge))
+		if err != nil {
+			errs = multierr.Append(errs,
+				fmt.Errorf("failed to get external ids for bridge %s: out: %s, err: %v", bridge, externalIDsOut, err))
+			continue
+		}
+
+		// Collect all stale pods for this bridge
+		var stalePods []struct {
+			podID  string
+			cookie uint64
+		}
+
+		// Parse external IDs to find pod IDs managed by rail CNI
+		lines := strings.Split(externalIDsOut, "\n")
+		for _, line := range lines {
+			if strings.Contains(line, "="+RailPodID) {
+				// Extract the key part (pod ID) before the =
+				parts := strings.Split(line, "=")
+				if len(parts) >= 2 {
+					podID := parts[0]
+
+					// Check if this pod still exists
+					if !existingPodUIDs[podID] {
+						// Pod doesn't exist, add to cleanup list
+						cookie := GenerateUint64FromString(podID)
+						stalePods = append(stalePods, struct {
+							podID  string
+							cookie uint64
+						}{podID, cookie})
+					}
+				}
+			}
+		}
+
+		if len(stalePods) == 0 {
+			continue
+		}
+
+		for _, stalePod := range stalePods {
+			logr.Info("Cleaning up stale flows for bridge", "bridge", bridge, "podID", stalePod.podID, "cookie", stalePod.cookie)
+			if err := f.cleanBridgeFlows(bridge, stalePod.podID, stalePod.cookie); err != nil {
+				errs = multierr.Append(errs, err)
+				continue
+			}
+		}
+	}
+
+	return errs
+}

internal/controller/flows_test.go

@@ -17,6 +17,7 @@
 package controller
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"strings"
@@ -230,8 +231,8 @@ var _ = Describe("Flows", func() {
 				Return("rail_pod_id", nil)
 			execMock.EXPECT().Execute("ovs-ofctl del-flows br-rail1 cookie=0x5/-1").Return("", nil)
 			execMock.EXPECT().Execute("ovs-ofctl del-flows br-rail2 cookie=0x5/-1").Return("", nil)
-			execMock.EXPECT().Execute(`ovs-vsctl br-set-external-id br-rail1 test-pod-uid ""`).Return("", nil)
-			execMock.EXPECT().Execute(`ovs-vsctl br-set-external-id br-rail2 test-pod-uid ""`).Return("", nil)
+			execMock.EXPECT().Execute("ovs-vsctl remove bridge br-rail1 external_ids test-pod-uid").Return("", nil)
+			execMock.EXPECT().Execute("ovs-vsctl remove bridge br-rail2 external_ids test-pod-uid").Return("", nil)
 			err := flows.DeletePodRailFlows(0x5, "test-pod-uid")
 			Expect(err).Should(Succeed())
 		})
@@ -273,7 +274,7 @@ var _ = Describe("Flows", func() {
 				Return("rail_pod_id", nil)
 			execMock.EXPECT().Execute("ovs-ofctl del-flows br-rail1 cookie=0x5/-1").Return("", nil)
 			execMock.EXPECT().
-				Execute(`ovs-vsctl br-set-external-id br-rail1 test-pod-uid ""`).
+				Execute("ovs-vsctl remove bridge br-rail1 external_ids test-pod-uid").
 				Return("", fmt.Errorf("failed to clear external id"))
 
 			err := flows.DeletePodRailFlows(0x5, "test-pod-uid")
@@ -321,4 +322,71 @@ var _ = Describe("Flows", func() {
 			Expect(isManaged).Should(BeFalse())
 		})
 	})
+
+	Context("CleanupStaleFlowsForBridges", func() {
+		It("should cleanup stale flows for bridges", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("br-rail1\nbr-rail2", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail1").
+				Return("test-pod-id=rail_pod_id", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail2").
+				Return("test-pod-id=rail_pod_id", nil)
+			execMock.EXPECT().
+				Execute(fmt.Sprintf("ovs-ofctl del-flows br-rail1 cookie=0x%x/-1", GenerateUint64FromString("test-pod-id"))).
+				Return("", nil)
+			execMock.EXPECT().
+				Execute(fmt.Sprintf("ovs-ofctl del-flows br-rail2 cookie=0x%x/-1", GenerateUint64FromString("test-pod-id"))).
+				Return("", nil)
+			execMock.EXPECT().Execute("ovs-vsctl remove bridge br-rail1 external_ids test-pod-id").Return("", nil)
+			execMock.EXPECT().Execute("ovs-vsctl remove bridge br-rail2 external_ids test-pod-id").Return("", nil)
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{})
+			Expect(err).Should(Succeed())
+		})
+
+		It("should return error if failed to list bridges", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("", fmt.Errorf("failed to list bridges"))
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{})
+			Expect(err).Should(HaveOccurred())
+		})
+
+		It("should return error if failed to get external id", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("br-rail1", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail1").
+				Return("", fmt.Errorf("failed to get external id"))
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{})
+			Expect(err).Should(HaveOccurred())
+		})
+
+		It("should return error if failed to delete flows", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("br-rail1", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail1").
+				Return("test-pod-id=rail_pod_id", nil)
+			execMock.EXPECT().
+				Execute(fmt.Sprintf("ovs-ofctl del-flows br-rail1 cookie=0x%x/-1", GenerateUint64FromString("test-pod-id"))).
+				Return("", fmt.Errorf("failed to delete flows"))
+
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{})
+			Expect(err).Should(HaveOccurred())
+		})
+
+		It("should return error if failed to clear external id", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("br-rail1", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail1").
+				Return("test-pod-id=rail_pod_id", nil)
+			execMock.EXPECT().
+				Execute(fmt.Sprintf("ovs-ofctl del-flows br-rail1 cookie=0x%x/-1", GenerateUint64FromString("test-pod-id"))).
+				Return("", nil)
+			execMock.EXPECT().Execute("ovs-vsctl remove bridge br-rail1 external_ids test-pod-id").
+				Return("", fmt.Errorf("failed to clear external id"))
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{})
+			Expect(err).Should(HaveOccurred())
+		})
+
+		It("should succeed if there are no stale pods", func() {
+			execMock.EXPECT().Execute("ovs-vsctl list-br").Return("br-rail1", nil)
+			execMock.EXPECT().Execute("ovs-vsctl br-get-external-id br-rail1").
+				Return("test-pod-id=rail_pod_id", nil)
+			err := flows.CleanupStaleFlowsForBridges(context.Background(), map[string]bool{"test-pod-id": true})
+			Expect(err).Should(Succeed())
+		})
+	})
 })

internal/controller/mock_flows.go

@@ -0,0 +1,88 @@
+/*
+ Copyright 2025, NVIDIA CORPORATION & AFFILIATES
+
+ Licensed under the Apache License, Version 2.0 (the "License");
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+     http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+*/
+
+package staleflows
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/Mellanox/spectrum-x-operator/internal/controller"
+
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+type Cleaner struct {
+	client.Client
+	CleanupInterval time.Duration
+	Flows           controller.FlowsAPI
+}
+
+// StartCleanupRoutine starts a background goroutine that periodically cleans up stale flows
+func (c *Cleaner) StartCleanupRoutine(ctx context.Context) {
+	go func() {
+		// Use configured interval or default to 5 minutes
+		interval := c.CleanupInterval
+		if interval == 0 {
+			interval = 5 * time.Minute
+		}
+
+		logr := log.FromContext(ctx).WithName("flow-cleanup")
+		logr.Info("Starting flow cleanup routine", "interval", interval)
+
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				logr.Info("Stopping flow cleanup routine")
+				return
+			case <-ticker.C:
+				if err := c.cleanupStaleFlows(ctx); err != nil {
+					logr.Error(err, "Failed to cleanup stale flows")
+				}
+			}
+		}
+	}()
+}
+
+// cleanupStaleFlows removes flows for pods that no longer exist
+func (c *Cleaner) cleanupStaleFlows(ctx context.Context) error {
+	pods := &corev1.PodList{}
+	// clinet cache is set to watch pods on the same node as the reconciler
+	if err := c.List(ctx, pods); err != nil {
+		return fmt.Errorf("failed to list pods: %w", err)
+	}
+
+	if len(pods.Items) == 0 {
+		return nil
+	}
+
+	existingPods := make(map[string]bool, len(pods.Items))
+	for _, pod := range pods.Items {
+		existingPods[string(pod.UID)] = true
+	}
+
+	if err := c.Flows.CleanupStaleFlowsForBridges(ctx, existingPods); err != nil {
+		return fmt.Errorf("failed to cleanup stale flows: %w", err)
+	}
+
+	return nil
+}