feat: support canarying using Istio DestinationRule subsets (#985)

Signed-off-by: Jesse Suen <[email protected]>

.github/workflows/e2e.yaml

@@ -31,7 +31,9 @@ jobs:
       - name: Download cache
         run: go mod download
       - name: Install CRDs
-        run: kubectl apply -k  manifests/crds
+        run: |
+          kubectl apply -k  manifests/crds
+          kubectl apply -f  test/e2e/crds
       - name: Start controller
         run: make start-e2e &
       - name: Run e2e tests

cmd/rollouts-controller/main.go

@@ -27,17 +27,18 @@ import (
 	clientset "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned"
 	"github.com/argoproj/argo-rollouts/pkg/signals"
 	"github.com/argoproj/argo-rollouts/rollout/trafficrouting/alb"
+	"github.com/argoproj/argo-rollouts/rollout/trafficrouting/smi"
 	controllerutil "github.com/argoproj/argo-rollouts/utils/controller"
+	"github.com/argoproj/argo-rollouts/utils/defaults"
+	"github.com/argoproj/argo-rollouts/utils/istio"
 	istioutil "github.com/argoproj/argo-rollouts/utils/istio"
 	kubeclientmetrics "github.com/argoproj/argo-rollouts/utils/kubeclientmetrics"
 	"github.com/argoproj/argo-rollouts/utils/tolerantinformer"
 )
 
 const (
 	// CLIName is the name of the CLI
-	cliName                    = "argo-rollouts"
-	defaultIstioVersion        = "v1alpha3"
-	defaultTrafficSplitVersion = "v1alpha1"
+	cliName = "argo-rollouts"
 )
 
 func newCommand() *cobra.Command {
@@ -75,6 +76,8 @@ func newCommand() *cobra.Command {
 			stopCh := signals.SetupSignalHandler()
 
 			alb.SetDefaultVerifyWeight(albVerifyWeight)
+			istio.SetIstioAPIVersion(istioVersion)
+			smi.SetSMIAPIVersion(trafficSplitVersion)
 
 			config, err := clientConfig.ClientConfig()
 			checkError(err)
@@ -111,7 +114,6 @@ func newCommand() *cobra.Command {
 				kubeinformers.WithTweakListOptions(func(options *metav1.ListOptions) {
 					options.LabelSelector = jobprovider.AnalysisRunUIDLabelKey
 				}))
-			istioGVR := istioutil.GetIstioGVR(istioVersion)
 			// We need three dynamic informer factories:
 			// 1. The first is the dynamic informer for rollouts, analysisruns, analysistemplates, experiments
 			dynamicInformerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, resyncDuration, namespace, instanceIDTweakListFunc)
@@ -120,9 +122,8 @@ func newCommand() *cobra.Command {
 			// is to support the mode when the rollout controller is started and only operating against
 			// a single namespace (i.e. rollouts-controller --namespace foo).
 			clusterDynamicInformerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, resyncDuration, metav1.NamespaceAll, instanceIDTweakListFunc)
-			// 3. We finally need an istio dynamic informer factory which uses different resync
-			// period and does not use a tweakListFunc.
-			istioDynamicInformerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, 0, namespace, nil)
+			// 3. We finally need an istio dynamic informer factory which does not use a tweakListFunc.
+			istioDynamicInformerFactory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, resyncDuration, namespace, nil)
 			cm := controller.NewManager(
 				namespace,
 				kubeClient,
@@ -138,13 +139,12 @@ func newCommand() *cobra.Command {
 				tolerantinformer.NewTolerantAnalysisRunInformer(dynamicInformerFactory),
 				tolerantinformer.NewTolerantAnalysisTemplateInformer(dynamicInformerFactory),
 				tolerantinformer.NewTolerantClusterAnalysisTemplateInformer(clusterDynamicInformerFactory),
-				istioDynamicInformerFactory.ForResource(istioGVR).Informer(),
+				istioDynamicInformerFactory.ForResource(istioutil.GetIstioVirtualServiceGVR()).Informer(),
+				istioDynamicInformerFactory.ForResource(istioutil.GetIstioDestinationRuleGVR()).Informer(),
 				resyncDuration,
 				instanceID,
 				metricsPort,
 				k8sRequestProvider,
-				istioVersion,
-				trafficSplitVersion,
 				nginxIngressClasses,
 				albIngressClasses)
 			// notice that there is no need to run Start methods in a separate goroutine. (i.e. go kubeInformerFactory.Start(stopCh)
@@ -157,7 +157,7 @@ func newCommand() *cobra.Command {
 			jobInformerFactory.Start(stopCh)
 
 			// Check if Istio installed on cluster before starting dynamicInformerFactory
-			if istioutil.DoesIstioExist(dynamicClient, namespace, istioVersion) {
+			if istioutil.DoesIstioExist(dynamicClient, namespace) {
 				istioDynamicInformerFactory.Start(stopCh)
 			}
 
@@ -183,8 +183,8 @@ func newCommand() *cobra.Command {
 	command.Flags().IntVar(&analysisThreads, "analysis-threads", controller.DefaultAnalysisThreads, "Set the number of worker threads for the Experiment controller")
 	command.Flags().IntVar(&serviceThreads, "service-threads", controller.DefaultServiceThreads, "Set the number of worker threads for the Service controller")
 	command.Flags().IntVar(&ingressThreads, "ingress-threads", controller.DefaultIngressThreads, "Set the number of worker threads for the Ingress controller")
-	command.Flags().StringVar(&istioVersion, "istio-api-version", defaultIstioVersion, "Set the default Istio apiVersion that controller should look when manipulating VirtualServices.")
-	command.Flags().StringVar(&trafficSplitVersion, "traffic-split-api-version", defaultTrafficSplitVersion, "Set the default TrafficSplit apiVersion that controller uses when creating TrafficSplits.")
+	command.Flags().StringVar(&istioVersion, "istio-api-version", defaults.DefaultIstioVersion, "Set the default Istio apiVersion that controller should look when manipulating VirtualServices.")
+	command.Flags().StringVar(&trafficSplitVersion, "traffic-split-api-version", defaults.DefaultSMITrafficSplitVersion, "Set the default TrafficSplit apiVersion that controller uses when creating TrafficSplits.")
 	command.Flags().StringArrayVar(&albIngressClasses, "alb-ingress-classes", defaultALBIngressClass, "Defines all the ingress class annotations that the alb ingress controller operates on. Defaults to alb")
 	command.Flags().StringArrayVar(&nginxIngressClasses, "nginx-ingress-classes", defaultNGINXIngressClass, "Defines all the ingress class annotations that the nginx ingress controller operates on. Defaults to nginx")
 	command.Flags().BoolVar(&albVerifyWeight, "alb-verify-weight", false, "Verify ALB target group weights before progressing through steps (requires AWS privileges)")

controller/controller.go

@@ -33,7 +33,6 @@ import (
 	informers "github.com/argoproj/argo-rollouts/pkg/client/informers/externalversions/rollouts/v1alpha1"
 	"github.com/argoproj/argo-rollouts/rollout"
 	"github.com/argoproj/argo-rollouts/service"
-	istioutil "github.com/argoproj/argo-rollouts/utils/istio"
 )
 
 const controllerAgentName = "rollouts-controller"
@@ -79,17 +78,13 @@ type Manager struct {
 	ingressSynced                 cache.InformerSynced
 	jobSynced                     cache.InformerSynced
 	replicasSetSynced             cache.InformerSynced
-	istioVirtualServiceSynced     cache.InformerSynced
 
 	rolloutWorkqueue     workqueue.RateLimitingInterface
 	serviceWorkqueue     workqueue.RateLimitingInterface
 	ingressWorkqueue     workqueue.RateLimitingInterface
 	experimentWorkqueue  workqueue.RateLimitingInterface
 	analysisRunWorkqueue workqueue.RateLimitingInterface
 
-	defaultIstioVersion        string
-	defaultTrafficSplitVersion string
-
 	dynamicClientSet dynamic.Interface
 
 	namespace string
@@ -112,12 +107,11 @@ func NewManager(
 	analysisTemplateInformer informers.AnalysisTemplateInformer,
 	clusterAnalysisTemplateInformer informers.ClusterAnalysisTemplateInformer,
 	istioVirtualServiceInformer cache.SharedIndexInformer,
+	istioDestinationRuleInformer cache.SharedIndexInformer,
 	resyncPeriod time.Duration,
 	instanceID string,
 	metricsPort int,
 	k8sRequestProvider *metrics.K8sRequestsCountProvider,
-	defaultIstioVersion string,
-	defaultTrafficSplitVersion string,
 	nginxIngressClasses []string,
 	albIngressClasses []string,
 ) *Manager {
@@ -158,6 +152,7 @@ func NewManager(
 		AnalysisTemplateInformer:        analysisTemplateInformer,
 		ClusterAnalysisTemplateInformer: clusterAnalysisTemplateInformer,
 		IstioVirtualServiceInformer:     istioVirtualServiceInformer,
+		IstioDestinationRuleInformer:    istioDestinationRuleInformer,
 		ReplicaSetInformer:              replicaSetInformer,
 		ServicesInformer:                servicesInformer,
 		IngressInformer:                 ingressesInformer,
@@ -168,8 +163,6 @@ func NewManager(
 		IngressWorkQueue:                ingressWorkqueue,
 		MetricsServer:                   metricsServer,
 		Recorder:                        recorder,
-		DefaultIstioVersion:             defaultIstioVersion,
-		DefaultTrafficSplitVersion:      defaultTrafficSplitVersion,
 	})
 
 	experimentController := experiments.NewController(experiments.ControllerConfig{
@@ -234,7 +227,6 @@ func NewManager(
 		analysisTemplateSynced:        analysisTemplateInformer.Informer().HasSynced,
 		clusterAnalysisTemplateSynced: clusterAnalysisTemplateInformer.Informer().HasSynced,
 		replicasSetSynced:             replicaSetInformer.Informer().HasSynced,
-		istioVirtualServiceSynced:     istioVirtualServiceInformer.HasSynced,
 		rolloutWorkqueue:              rolloutWorkqueue,
 		experimentWorkqueue:           experimentWorkqueue,
 		analysisRunWorkqueue:          analysisRunWorkqueue,
@@ -245,8 +237,6 @@ func NewManager(
 		ingressController:             ingressController,
 		experimentController:          experimentController,
 		analysisController:            analysisController,
-		defaultIstioVersion:           defaultIstioVersion,
-		defaultTrafficSplitVersion:    defaultTrafficSplitVersion,
 		dynamicClientSet:              dynamicclientset,
 		namespace:                     namespace,
 	}
@@ -276,13 +266,6 @@ func (c *Manager) Run(rolloutThreadiness, serviceThreadiness, ingressThreadiness
 			return fmt.Errorf("failed to wait for cluster-scoped caches to sync")
 		}
 	}
-	// Check if Istio exists
-	if istioutil.DoesIstioExist(c.dynamicClientSet, c.namespace, c.defaultIstioVersion) {
-		// Wait for Istio cache to sync before starting workers
-		if ok := cache.WaitForCacheSync(stopCh, c.istioVirtualServiceSynced); !ok {
-			return fmt.Errorf("failed to wait for istio virtualService cache to sync")
-		}
-	}
 
 	// Start the informer factories to begin populating the informer caches
 	log.Info("Starting Controllers")