feat: add multi-cluster replication and remote cleanup support

Author: Amar-08

cmd/flagger/main.go

@@ -88,6 +88,9 @@ var (
 	kubeconfigServiceMesh    string
 	clusterName              string
 	noCrossNamespaceRefs     bool
+	istioMultiClusterEnabled bool
+	istioMultiClusterLabel   string
+	istioMultiClusterNs      string
 )
 
 func init() {
@@ -123,6 +126,9 @@ func init() {
 	flag.StringVar(&kubeconfigServiceMesh, "kubeconfig-service-mesh", "", "Path to a kubeconfig for the service mesh control plane cluster.")
 	flag.StringVar(&clusterName, "cluster-name", "", "Cluster name to be included in alert msgs.")
 	flag.BoolVar(&noCrossNamespaceRefs, "no-cross-namespace-refs", false, "When set to true, Flagger can only refer to resources in the same namespace.")
+	flag.BoolVar(&istioMultiClusterEnabled, "istio-multicluster-enabled", false, "Enable Istio multi-cluster support.")
+	flag.StringVar(&istioMultiClusterLabel, "istio-multicluster-secret-label", "istio/multiCluster=true", "Label on secrets for Istio multi-cluster discovery.")
+	flag.StringVar(&istioMultiClusterNs, "istio-multicluster-secret-namespace", "istio-system", "Namespace where Istio multi-cluster secrets are located.")
 }
 
 func main() {
@@ -228,7 +234,13 @@ func main() {
 		setOwnerRefs = false
 	}
 
-	routerFactory := router.NewFactory(cfg, kubeClient, flaggerClient, knativeClient, ingressAnnotationsPrefix, ingressClass, logger, meshClient, setOwnerRefs)
+	var clusterManager *router.ClusterManager
+	if istioMultiClusterEnabled {
+		clusterManager = router.NewClusterManager(kubeClient, meshClient, logger, istioMultiClusterLabel, istioMultiClusterNs)
+		clusterManager.Start(stopCh)
+	}
+
+	routerFactory := router.NewFactory(cfg, kubeClient, flaggerClient, knativeClient, ingressAnnotationsPrefix, ingressClass, logger, meshClient, setOwnerRefs, clusterManager)
 
 	var configTracker canary.Tracker
 	if enableConfigTracking {

pkg/controller/controller.go

@@ -159,9 +159,9 @@ func NewController(
 
 				ctrl.enqueue(new)
 			} else if !newCanary.DeletionTimestamp.IsZero() && hasFinalizer(&newCanary) ||
-				!hasFinalizer(&newCanary) && newCanary.Spec.RevertOnDeletion {
+				!hasFinalizer(&newCanary) && (newCanary.Spec.RevertOnDeletion || ctrl.routerFactory.IsMultiClusterEnabled()) {
 				// If this was marked for deletion and has finalizers enqueue for finalizing or
-				// if this canary doesn't have finalizers and RevertOnDeletion is true updated speck enqueue
+				// if this canary doesn't have finalizers and RevertOnDeletion is true (or multi-cluster is enabled) updated speck enqueue
 				ctrl.enqueue(new)
 			}
 
@@ -266,8 +266,8 @@ func (c *Controller) syncHandler(key string) error {
 		return fmt.Errorf("invalid canary spec: %s", err)
 	}
 
-	// Finalize if canary has been marked for deletion and revert is desired
-	if cd.Spec.RevertOnDeletion && cd.ObjectMeta.DeletionTimestamp != nil {
+	// Finalize if canary has been marked for deletion and revert is desired (or multi-cluster is enabled)
+	if (cd.Spec.RevertOnDeletion || c.routerFactory.IsMultiClusterEnabled()) && cd.ObjectMeta.DeletionTimestamp != nil {
 		// If finalizers have been previously removed proceed
 		if !hasFinalizer(cd) {
 			c.logger.Infof("Canary %s.%s has been finalized", cd.Name, cd.Namespace)
@@ -306,8 +306,8 @@ func (c *Controller) syncHandler(key string) error {
 
 	c.canaries.Store(fmt.Sprintf("%s.%s", cd.Name, cd.Namespace), cd)
 
-	// If opt in for revertOnDeletion add finalizer if not present
-	if cd.Spec.RevertOnDeletion && !hasFinalizer(cd) {
+	// If opt in for revertOnDeletion (or multi-cluster is enabled) add finalizer if not present
+	if (cd.Spec.RevertOnDeletion || c.routerFactory.IsMultiClusterEnabled()) && !hasFinalizer(cd) {
 		if err := c.addFinalizer(cd); err != nil {
 			return fmt.Errorf("unable to add finalizer to canary %s.%s: %w", cd.Name, cd.Namespace, err)
 		}

pkg/controller/finalizer.go

@@ -18,8 +18,8 @@ package controller
 
 import (
 	"context"
-	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -80,12 +80,12 @@ func (c *Controller) finalize(old interface{}) error {
 		Cap:      canary.GetAnalysisInterval(),
 		Steps:    4,
 	}
-	retry.OnError(backoff, func(err error) bool {
-		return err.Error() == "retriable error"
+	err = retry.OnError(backoff, func(err error) bool {
+		return strings.Contains(err.Error(), "retriable: ")
 	}, func() error {
 		retriable, err := canaryController.IsCanaryReady(canary)
 		if err != nil && retriable {
-			return errors.New("retriable error")
+			return fmt.Errorf("retriable: %w", err)
 		}
 		return err
 	})

pkg/controller/multi_cluster_controller_test.go

@@ -0,0 +1,68 @@
+package controller
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/fluxcd/flagger/pkg/router"
+)
+
+func TestController_MultiClusterFinalizer(t *testing.T) {
+	// 1. Setup Canary WITHOUT RevertOnDeletion
+	canary := newDeploymentTestCanary()
+	canary.Spec.RevertOnDeletion = false
+
+	// 2. Setup Fixture with Multi-Cluster ENABLED
+	// We'll mock the ClusterManager directly in the factory
+	mocks := newDeploymentFixture(canary)
+
+	// Inject ClusterManager to simulate multi-cluster enabled
+	cm := router.NewClusterManager(mocks.kubeClient, mocks.flaggerClient, mocks.logger, "istio/multiCluster=true", "istio-system")
+	rf := router.NewFactory(nil, mocks.kubeClient, mocks.flaggerClient, nil, "", "", mocks.logger, mocks.flaggerClient, true, cm)
+
+	mocks.ctrl.routerFactory = rf // Override with our multi-cluster factory
+
+	// 3. Sync the canary - this should ADD the finalizer because multi-cluster is enabled
+	key := fmt.Sprintf("%s/%s", canary.Namespace, canary.Name)
+	err := mocks.ctrl.syncHandler(key)
+	require.NoError(t, err)
+
+	// 4. Verify Finalizer was added
+	c, err := mocks.flaggerClient.FlaggerV1beta1().Canaries(canary.Namespace).Get(context.Background(), canary.Name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.True(t, hasFinalizer(c), "Finalizer should be added in multi-cluster mode even if RevertOnDeletion is false")
+
+	// Initialize canary (creates primary, set status etc)
+	mocks.ctrl.advanceCanary(canary.Name, canary.Namespace)
+	mocks.makePrimaryReady(t)
+	mocks.makeCanaryReady(t)
+	mocks.ctrl.advanceCanary(canary.Name, canary.Namespace)
+
+	// Update informer indexer
+	c, _ = mocks.flaggerClient.FlaggerV1beta1().Canaries(canary.Namespace).Get(context.Background(), canary.Name, metav1.GetOptions{})
+	err = mocks.ctrl.flaggerInformers.CanaryInformer.Informer().GetIndexer().Update(c)
+	require.NoError(t, err)
+
+	// 5. Mark for Deletion
+	now := metav1.Now()
+	c.DeletionTimestamp = &now
+	c, err = mocks.flaggerClient.FlaggerV1beta1().Canaries(canary.Namespace).Update(context.Background(), c, metav1.UpdateOptions{})
+	require.NoError(t, err)
+	// Update informer indexer
+	err = mocks.ctrl.flaggerInformers.CanaryInformer.Informer().GetIndexer().Update(c)
+	require.NoError(t, err)
+
+	// 6. Sync again - this should TRIGGER finalization
+	err = mocks.ctrl.syncHandler(key)
+	require.NoError(t, err)
+
+	// 7. Verify Finalizer was removed (after successful finalization)
+	c, err = mocks.flaggerClient.FlaggerV1beta1().Canaries(canary.Namespace).Get(context.Background(), canary.Name, metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.False(t, hasFinalizer(c), "Finalizer should be removed after successful finalization")
+}

pkg/controller/scheduler.go

@@ -550,9 +550,6 @@ func (c *Controller) runPromotionTrafficShift(canary *flaggerv1.Canary, canaryCo
 			}
 		}
 	}
-
-	return
-
 }
 
 func (c *Controller) runCanary(canary *flaggerv1.Canary, canaryController canary.Controller,

pkg/controller/scheduler_daemonset_fixture_test.go

@@ -92,7 +92,7 @@ func newDaemonSetFixture(c *flaggerv1.Canary) daemonSetFixture {
 	}
 
 	// init router
-	rf := router.NewFactory(nil, kubeClient, flaggerClient, nil, "annotationsPrefix", "", logger, flaggerClient, true)
+	rf := router.NewFactory(nil, kubeClient, flaggerClient, nil, "annotationsPrefix", "", logger, flaggerClient, true, nil)
 
 	// init observer
 	observerFactory, _ := observers.NewFactory(testMetricsServerURL)

pkg/controller/scheduler_deployment_fixture_test.go

@@ -121,7 +121,7 @@ func newDeploymentFixture(c *flaggerv1.Canary) fixture {
 	}
 
 	// init router
-	rf := router.NewFactory(nil, kubeClient, flaggerClient, nil, "annotationsPrefix", "", logger, flaggerClient, true)
+	rf := router.NewFactory(nil, kubeClient, flaggerClient, nil, "annotationsPrefix", "", logger, flaggerClient, true, nil)
 
 	// init observer
 	observerFactory, _ := observers.NewFactory(testMetricsServerURL)

pkg/router/factory.go

@@ -38,6 +38,7 @@ type Factory struct {
 	ingressClass             string
 	logger                   *zap.SugaredLogger
 	setOwnerRefs             bool
+	clusterManager           *ClusterManager
 }
 
 func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface,
@@ -47,7 +48,8 @@ func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface,
 	ingressClass string,
 	logger *zap.SugaredLogger,
 	meshClient clientset.Interface,
-	setOwnerRefs bool) *Factory {
+	setOwnerRefs bool,
+	clusterManager *ClusterManager) *Factory {
 	return &Factory{
 		kubeConfig:               kubeConfig,
 		meshClient:               meshClient,
@@ -58,6 +60,7 @@ func NewFactory(kubeConfig *restclient.Config, kubeClient kubernetes.Interface,
 		ingressClass:             ingressClass,
 		logger:                   logger,
 		setOwnerRefs:             setOwnerRefs,
+		clusterManager:           clusterManager,
 	}
 }
 
@@ -109,11 +112,12 @@ func (factory *Factory) MeshRouter(provider string, labelSelector string) Interf
 		}
 	case provider == flaggerv1.IstioProvider:
 		return &IstioRouter{
-			logger:        factory.logger,
-			flaggerClient: factory.flaggerClient,
-			kubeClient:    factory.kubeClient,
-			istioClient:   factory.meshClient,
-			setOwnerRefs:  factory.setOwnerRefs,
+			logger:         factory.logger,
+			flaggerClient:  factory.flaggerClient,
+			kubeClient:     factory.kubeClient,
+			istioClients:   factory.getMeshClients(),
+			setOwnerRefs:   factory.setOwnerRefs,
+			clusterManager: factory.clusterManager,
 		}
 	case strings.HasPrefix(provider, flaggerv1.SMIProvider+":v1alpha1"):
 		mesh := strings.TrimPrefix(provider, flaggerv1.SMIProvider+":v1alpha1:")
@@ -225,11 +229,24 @@ func (factory *Factory) MeshRouter(provider string, labelSelector string) Interf
 		return &NopRouter{}
 	default:
 		return &IstioRouter{
-			logger:        factory.logger,
-			flaggerClient: factory.flaggerClient,
-			kubeClient:    factory.kubeClient,
-			istioClient:   factory.meshClient,
-			setOwnerRefs:  factory.setOwnerRefs,
+			logger:         factory.logger,
+			flaggerClient:  factory.flaggerClient,
+			kubeClient:     factory.kubeClient,
+			istioClients:   factory.getMeshClients(),
+			setOwnerRefs:   factory.setOwnerRefs,
+			clusterManager: factory.clusterManager,
 		}
 	}
 }
+
+func (factory *Factory) getMeshClients() []clientset.Interface {
+	if factory.clusterManager != nil {
+		return factory.clusterManager.GetClients()
+	}
+	return []clientset.Interface{factory.meshClient}
+}
+
+// IsMultiClusterEnabled returns true if multi-cluster support is enabled
+func (factory *Factory) IsMultiClusterEnabled() bool {
+	return factory.clusterManager != nil
+}