controller.go

package rollout

import (
	"context"
	"encoding/json"
	"fmt"
	"reflect"
	"time"

	smiclientset "github.com/servicemeshinterface/smi-sdk-go/pkg/gen/client/split/clientset/versioned"
	log "github.com/sirupsen/logrus"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/api/extensions/v1beta1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/util/validation/field"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/dynamic/dynamiclister"
	appsinformers "k8s.io/client-go/informers/apps/v1"
	coreinformers "k8s.io/client-go/informers/core/v1"
	extensionsinformers "k8s.io/client-go/informers/extensions/v1beta1"
	"k8s.io/client-go/kubernetes"
	appslisters "k8s.io/client-go/listers/apps/v1"
	v1 "k8s.io/client-go/listers/core/v1"
	extensionslisters "k8s.io/client-go/listers/extensions/v1beta1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/kubernetes/cmd/kubeadm/app/util"
	"k8s.io/kubernetes/pkg/controller"
	"k8s.io/utils/pointer"

	"github.com/argoproj/argo-rollouts/controller/metrics"
	register "github.com/argoproj/argo-rollouts/pkg/apis/rollouts"
	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/validation"
	clientset "github.com/argoproj/argo-rollouts/pkg/client/clientset/versioned"
	informers "github.com/argoproj/argo-rollouts/pkg/client/informers/externalversions/rollouts/v1alpha1"
	listers "github.com/argoproj/argo-rollouts/pkg/client/listers/rollouts/v1alpha1"
	"github.com/argoproj/argo-rollouts/rollout/trafficrouting/istio"
	analysisutil "github.com/argoproj/argo-rollouts/utils/analysis"
	"github.com/argoproj/argo-rollouts/utils/conditions"
	controllerutil "github.com/argoproj/argo-rollouts/utils/controller"
	"github.com/argoproj/argo-rollouts/utils/defaults"
	experimentutil "github.com/argoproj/argo-rollouts/utils/experiment"
	istioutil "github.com/argoproj/argo-rollouts/utils/istio"
	logutil "github.com/argoproj/argo-rollouts/utils/log"
	replicasetutil "github.com/argoproj/argo-rollouts/utils/replicaset"
	serviceutil "github.com/argoproj/argo-rollouts/utils/service"
	unstructuredutil "github.com/argoproj/argo-rollouts/utils/unstructured"
)

const (
	virtualServiceIndexName = "byVirtualService"
)

// Controller is the controller implementation for Rollout resources
type Controller struct {
	reconcilerBase

	// namespace which namespace(s) operates on
	namespace string
	// rsControl is used for adopting/releasing replica sets.
	replicaSetControl controller.RSControlInterface

	metricsServer *metrics.MetricsServer

	// workqueue is a rate limited work queue. This is used to queue work to be
	// processed instead of performing it as soon as a change happens. This
	// means we can ensure we only process a fixed amount of resources at a
	// time, and makes it easy to ensure we are never processing the same item
	// simultaneously in two different workers.
	rolloutWorkqueue workqueue.RateLimitingInterface
	serviceWorkqueue workqueue.RateLimitingInterface
	ingressWorkqueue workqueue.RateLimitingInterface
}

// ControllerConfig describes the data required to instantiate a new rollout controller
type ControllerConfig struct {
	Namespace                       string
	KubeClientSet                   kubernetes.Interface
	ArgoProjClientset               clientset.Interface
	DynamicClientSet                dynamic.Interface
	SmiClientSet                    smiclientset.Interface
	ExperimentInformer              informers.ExperimentInformer
	AnalysisRunInformer             informers.AnalysisRunInformer
	AnalysisTemplateInformer        informers.AnalysisTemplateInformer
	ClusterAnalysisTemplateInformer informers.ClusterAnalysisTemplateInformer
	ReplicaSetInformer              appsinformers.ReplicaSetInformer
	ServicesInformer                coreinformers.ServiceInformer
	IngressInformer                 extensionsinformers.IngressInformer
	RolloutsInformer                informers.RolloutInformer
	IstioVirtualServiceInformer     cache.SharedIndexInformer
	ResyncPeriod                    time.Duration
	RolloutWorkQueue                workqueue.RateLimitingInterface
	ServiceWorkQueue                workqueue.RateLimitingInterface
	IngressWorkQueue                workqueue.RateLimitingInterface
	MetricsServer                   *metrics.MetricsServer
	Recorder                        record.EventRecorder
	DefaultIstioVersion             string
	DefaultTrafficSplitVersion      string
}

// reconcilerBase is a shared datastructure containing all clients and configuration necessary to
// reconcile a rollout. This is shared between the controller and the rolloutContext
type reconcilerBase struct {
	// kubeclientset is a standard kubernetes clientset
	kubeclientset kubernetes.Interface
	// argoprojclientset is a clientset for our own API group
	argoprojclientset clientset.Interface
	// dynamicclientset is a dynamic clientset for interacting with unstructured resources.
	// It is used to interact with TrafficRouting resources
	dynamicclientset           dynamic.Interface
	smiclientset               smiclientset.Interface
	defaultIstioVersion        string
	defaultTrafficSplitVersion string

	replicaSetLister              appslisters.ReplicaSetLister
	replicaSetSynced              cache.InformerSynced
	rolloutsInformer              cache.SharedIndexInformer
	rolloutsLister                listers.RolloutLister
	rolloutsSynced                cache.InformerSynced
	rolloutsIndexer               cache.Indexer
	servicesLister                v1.ServiceLister
	ingressesLister               extensionslisters.IngressLister
	experimentsLister             listers.ExperimentLister
	analysisRunLister             listers.AnalysisRunLister
	analysisTemplateLister        listers.AnalysisTemplateLister
	clusterAnalysisTemplateLister listers.ClusterAnalysisTemplateLister
	// Include istioVirtualServiceInformer in Controller struct. If Istio does not exist and is
	// later added, then controller will auto-detect change and start istioVirtualServiceInformer
	istioVirtualServiceInformer cache.SharedIndexInformer
	istioVirtualServiceLister   dynamiclister.Lister

	podRestarter RolloutPodRestarter

	// used for unit testing
	enqueueRollout              func(obj interface{})                                         //nolint:structcheck
	enqueueRolloutAfter         func(obj interface{}, duration time.Duration)                 //nolint:structcheck
	newTrafficRoutingReconciler func(roCtx *rolloutContext) (TrafficRoutingReconciler, error) //nolint:structcheck

	// recorder is an event recorder for recording Event resources to the Kubernetes API.
	recorder     record.EventRecorder
	resyncPeriod time.Duration
}

// NewController returns a new rollout controller
func NewController(cfg ControllerConfig) *Controller {

	replicaSetControl := controller.RealRSControl{
		KubeClient: cfg.KubeClientSet,
		Recorder:   cfg.Recorder,
	}

	podRestarter := RolloutPodRestarter{
		client:       cfg.KubeClientSet,
		resyncPeriod: cfg.ResyncPeriod,
		enqueueAfter: func(obj interface{}, duration time.Duration) {
			controllerutil.EnqueueAfter(obj, duration, cfg.RolloutWorkQueue)
		},
	}

	base := reconcilerBase{
		kubeclientset:                 cfg.KubeClientSet,
		argoprojclientset:             cfg.ArgoProjClientset,
		dynamicclientset:              cfg.DynamicClientSet,
		smiclientset:                  cfg.SmiClientSet,
		defaultIstioVersion:           cfg.DefaultIstioVersion,
		defaultTrafficSplitVersion:    cfg.DefaultTrafficSplitVersion,
		replicaSetLister:              cfg.ReplicaSetInformer.Lister(),
		replicaSetSynced:              cfg.ReplicaSetInformer.Informer().HasSynced,
		rolloutsInformer:              cfg.RolloutsInformer.Informer(),
		rolloutsIndexer:               cfg.RolloutsInformer.Informer().GetIndexer(),
		rolloutsLister:                cfg.RolloutsInformer.Lister(),
		rolloutsSynced:                cfg.RolloutsInformer.Informer().HasSynced,
		servicesLister:                cfg.ServicesInformer.Lister(),
		ingressesLister:               cfg.IngressInformer.Lister(),
		experimentsLister:             cfg.ExperimentInformer.Lister(),
		analysisRunLister:             cfg.AnalysisRunInformer.Lister(),
		analysisTemplateLister:        cfg.AnalysisTemplateInformer.Lister(),
		clusterAnalysisTemplateLister: cfg.ClusterAnalysisTemplateInformer.Lister(),
		istioVirtualServiceLister:     dynamiclister.New(cfg.IstioVirtualServiceInformer.GetIndexer(), istioutil.GetIstioGVR(cfg.DefaultIstioVersion)),
		istioVirtualServiceInformer:   cfg.IstioVirtualServiceInformer,
		recorder:                      cfg.Recorder,
		resyncPeriod:                  cfg.ResyncPeriod,
		podRestarter:                  podRestarter,
	}

	controller := &Controller{
		reconcilerBase:    base,
		namespace:         cfg.Namespace,
		replicaSetControl: replicaSetControl,
		rolloutWorkqueue:  cfg.RolloutWorkQueue,
		serviceWorkqueue:  cfg.ServiceWorkQueue,
		ingressWorkqueue:  cfg.IngressWorkQueue,
		metricsServer:     cfg.MetricsServer,
	}
	controller.enqueueRollout = func(obj interface{}) {
		controllerutil.EnqueueRateLimited(obj, cfg.RolloutWorkQueue)
	}
	controller.enqueueRolloutAfter = func(obj interface{}, duration time.Duration) {
		controllerutil.EnqueueAfter(obj, duration, cfg.RolloutWorkQueue)
	}

	controller.newTrafficRoutingReconciler = controller.NewTrafficRoutingReconciler

	log.Info("Setting up event handlers")
	// Set up an event handler for when rollout resources change
	cfg.RolloutsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: controller.enqueueRollout,
		UpdateFunc: func(old, new interface{}) {
			oldRollout := unstructuredutil.ObjectToRollout(old)
			newRollout := unstructuredutil.ObjectToRollout(new)
			if old == nil || new == nil {
				return
			}
			for _, s := range serviceutil.GetRolloutServiceKeys(oldRollout) {
				controller.serviceWorkqueue.AddRateLimited(s)
			}
			controller.enqueueRollout(newRollout)
		},
		DeleteFunc: func(obj interface{}) {
			if ro := unstructuredutil.ObjectToRollout(obj); ro != nil {
				for _, s := range serviceutil.GetRolloutServiceKeys(ro) {
					controller.serviceWorkqueue.AddRateLimited(s)
				}
			}
		},
	})

	// Indexer to frequently check/enqueue Rollouts that reference virtualServices
	util.CheckErr(cfg.RolloutsInformer.Informer().AddIndexers(cache.Indexers{
		virtualServiceIndexName: func(obj interface{}) (strings []string, e error) {
			if ro := unstructuredutil.ObjectToRollout(obj); ro != nil {
				return istio.GetRolloutVirtualServiceKeys(ro), nil
			}
			return
		},
	}))

	cfg.ReplicaSetInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			controllerutil.EnqueueParentObject(obj, register.RolloutKind, controller.enqueueRollout)
		},
		UpdateFunc: func(old, new interface{}) {
			newRS := new.(*appsv1.ReplicaSet)
			oldRS := old.(*appsv1.ReplicaSet)
			if newRS.ResourceVersion == oldRS.ResourceVersion {
				// Periodic resync will send update events for all known replicas.
				// Two different versions of the same Replica will always have different RVs.
				return
			}
			controllerutil.EnqueueParentObject(new, register.RolloutKind, controller.enqueueRollout)
		},
		DeleteFunc: func(obj interface{}) {
			controllerutil.EnqueueParentObject(obj, register.RolloutKind, controller.enqueueRollout)
		},
	})

	cfg.AnalysisRunInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			controllerutil.EnqueueParentObject(obj, register.RolloutKind, controller.enqueueRollout)
		},
		UpdateFunc: func(old, new interface{}) {
			oldAR := unstructuredutil.ObjectToAnalysisRun(old)
			newAR := unstructuredutil.ObjectToAnalysisRun(new)
			if oldAR == nil || newAR == nil {
				return
			}
			if newAR.Status.Phase == oldAR.Status.Phase {
				// Only enqueue rollout if the status changed
				return
			}
			controllerutil.EnqueueParentObject(new, register.RolloutKind, controller.enqueueRollout)
		},
		DeleteFunc: func(obj interface{}) {
			controllerutil.EnqueueParentObject(obj, register.RolloutKind, controller.enqueueRollout)
		},
	})

	cfg.IstioVirtualServiceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj interface{}) {
			controller.EnqueueIstioVsvc(obj)
		},
		// TODO: DeepEquals on httpRoutes
		UpdateFunc: func(old, new interface{}) {
			controller.EnqueueIstioVsvc(new)
		},
		DeleteFunc: func(obj interface{}) {
			controller.EnqueueIstioVsvc(obj)
		},
	})

	return controller
}

// Run will set up the event handlers for types we are interested in, as well
// as syncing informer caches and starting workers. It will block until stopCh
// is closed, at which point it will shutdown the workqueue and wait for
// workers to finish processing their current work items.
func (c *Controller) Run(threadiness int, stopCh <-chan struct{}) error {
	log.Info("Starting Rollout workers")
	for i := 0; i < threadiness; i++ {
		go wait.Until(func() {
			controllerutil.RunWorker(c.rolloutWorkqueue, logutil.RolloutKey, c.syncHandler, c.metricsServer)
		}, time.Second, stopCh)
	}
	log.Info("Started Rollout workers")

	// Auto-detect whether user adds Istio Virtual Service to cluster
	// If it is added, then run the istioVirtualServiceInformer so that we can use istioVirtualServiceLister
	go c.runVirtualServiceInformer(stopCh)

	<-stopCh
	log.Info("Shutting down workers")

	return nil
}

func (c *Controller) runVirtualServiceInformer(stopCh <-chan struct{}) {
	ticker := time.NewTicker(10 * time.Minute)
	for !istioutil.DoesIstioExist(c.dynamicclientset, c.namespace, c.defaultIstioVersion) {
		// Should only execute if Istio is not installed on cluster
		select {
		case <-stopCh:
			ticker.Stop()
			return
		case <-ticker.C:
		}
	}
	ticker.Stop()
	if !c.istioVirtualServiceInformer.HasSynced() {
		// Should only execute if Istio was installed after Rollout Controller was started
		c.istioVirtualServiceInformer.Run(stopCh)
	}
}

func (c *Controller) EnqueueIstioVsvc(vsvc interface{}) {
	acc, err := meta.Accessor(vsvc)
	if err != nil {
		log.Errorf("Error processing istio vsvc from watch: %v: %v", err, vsvc)
		return
	}
	vsvcToEnqueue, err := c.rolloutsIndexer.ByIndex(virtualServiceIndexName, fmt.Sprintf("%s/%s", acc.GetNamespace(), acc.GetName()))
	if err != nil {
		log.Errorf("Cannot process indexer: %s", err.Error())
		return
	}
	for i := range vsvcToEnqueue {
		controllerutil.EnqueueParentObject(vsvcToEnqueue[i], register.RolloutKind, c.enqueueRollout)
	}
}

// syncHandler compares the actual state with the desired, and attempts to
// converge the two. It then updates the Phase block of the Rollout resource
// with the current status of the resource.
func (c *Controller) syncHandler(key string) error {
	ctx := context.TODO()
	startTime := time.Now()
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}
	rollout, err := c.rolloutsLister.Rollouts(namespace).Get(name)
	if k8serrors.IsNotFound(err) {
		return nil
	}
	if err != nil {
		return err
	}
	log.WithField(logutil.RolloutKey, name).WithField(logutil.NamespaceKey, namespace).Infof("Started syncing rollout at (%v)", startTime)

	// Remarshal the rollout to normalize all fields so that when we calculate hashes against the
	// rollout spec and pod template spec, the hash will be consistent. See issue #70
	// This also returns a copy of the rollout to prevent mutation of the informer cache.
	r := remarshalRollout(rollout)
	logCtx := logutil.WithRollout(r)

	// TODO(dthomson) remove in v0.9.0
	migrated := c.migrateCanaryStableRS(r)
	if migrated {
		logCtx.Info("Migrated stableRS field")
		return nil
	}

	if r.ObjectMeta.DeletionTimestamp != nil {
		logCtx.Info("No reconciliation as rollout marked for deletion")
		return nil
	}

	// In order to work with HPA, the rollout.Spec.Replica field cannot be nil. As a result, the controller will update
	// the rollout to have the replicas field set to the default value. see https://github.com/argoproj/argo-rollouts/issues/119
	if rollout.Spec.Replicas == nil {
		logCtx.Info("Setting .Spec.Replica to 1 from nil")
		r.Spec.Replicas = pointer.Int32Ptr(defaults.DefaultReplicas)
		_, err := c.argoprojclientset.ArgoprojV1alpha1().Rollouts(r.Namespace).Update(ctx, r, metav1.UpdateOptions{})
		return err
	}
	defer func() {
		duration := time.Since(startTime)
		c.metricsServer.IncRolloutReconcile(r, duration)
		logCtx.WithField("time_ms", duration.Seconds()*1e3).Info("Reconciliation completed")
	}()

	roCtx, err := c.newRolloutContext(r)
	if err != nil {
		return err
	}
	err = roCtx.reconcile()
	if roCtx.newRollout != nil {
		c.writeBackToInformer(roCtx.newRollout)
	}
	return err
}

// writeBackToInformer writes a just recently updated Rollout back into the informer cache.
// This prevents the situation where the controller operates on a stale rollout and repeats work
func (c *Controller) writeBackToInformer(ro *v1alpha1.Rollout) {
	logCtx := logutil.WithRollout(ro)
	obj, err := runtime.DefaultUnstructuredConverter.ToUnstructured(ro)
	if err != nil {
		logCtx.Errorf("failed to convert rollout to unstructured: %v", err)
		return
	}
	un := unstructured.Unstructured{Object: obj}
	err = c.rolloutsInformer.GetStore().Update(&un)
	if err != nil {
		logCtx.Errorf("failed to update informer store: %v", err)
		return
	}
}

func (c *Controller) newRolloutContext(rollout *v1alpha1.Rollout) (*rolloutContext, error) {
	rsList, err := c.getReplicaSetsForRollouts(rollout)
	if err != nil {
		return nil, err
	}

	newRS := replicasetutil.FindNewReplicaSet(rollout, rsList)
	olderRSs := replicasetutil.FindOldReplicaSets(rollout, rsList)
	stableRS := replicasetutil.GetStableRS(rollout, newRS, olderRSs)
	otherRSs := replicasetutil.GetOtherRSs(rollout, newRS, stableRS, olderRSs)

	exList, err := c.getExperimentsForRollout(rollout)
	if err != nil {
		return nil, err
	}
	currentEx := experimentutil.GetCurrentExperiment(rollout, exList)
	otherExs := experimentutil.GetOldExperiments(rollout, exList)

	arList, err := c.getAnalysisRunsForRollout(rollout)
	if err != nil {
		return nil, err
	}
	currentArs, otherArs := analysisutil.FilterCurrentRolloutAnalysisRuns(arList, rollout)

	logCtx := logutil.WithRollout(rollout)
	roCtx := rolloutContext{
		rollout:    rollout,
		log:        logCtx,
		newRS:      newRS,
		stableRS:   stableRS,
		olderRSs:   olderRSs,
		otherRSs:   otherRSs,
		allRSs:     rsList,
		currentArs: currentArs,
		otherArs:   otherArs,
		currentEx:  currentEx,
		otherExs:   otherExs,
		newStatus: v1alpha1.RolloutStatus{
			RestartedAt: rollout.Status.RestartedAt,
		},
		pauseContext: &pauseContext{
			rollout: rollout,
			log:     logCtx,
		},
		reconcilerBase: c.reconcilerBase,
	}
	return &roCtx, nil
}

func (c *rolloutContext) getRolloutValidationErrors() error {
	rolloutValidationErrors := validation.ValidateRollout(c.rollout)
	if len(rolloutValidationErrors) > 0 {
		return rolloutValidationErrors[0]
	}

	refResources, err := c.getRolloutReferencedResources()
	if err != nil {
		return err
	}

	rolloutValidationErrors = validation.ValidateRolloutReferencedResources(c.rollout, *refResources)
	if len(rolloutValidationErrors) > 0 {
		return rolloutValidationErrors[0]
	}
	return nil
}

func (c *rolloutContext) createInvalidRolloutCondition(validationError error, r *v1alpha1.Rollout) error {
	prevCond := conditions.GetRolloutCondition(r.Status, v1alpha1.InvalidSpec)
	invalidSpecCond := prevCond
	errorMessage := fmt.Sprintf("The Rollout \"%s\" is invalid: %s", r.Name, validationError.Error())
	if prevCond == nil || prevCond.Message != errorMessage {
		invalidSpecCond = conditions.NewRolloutCondition(v1alpha1.InvalidSpec, corev1.ConditionTrue, conditions.InvalidSpecReason, errorMessage)
	}
	c.log.Error(errorMessage)
	generation := conditions.ComputeGenerationHash(r.Spec)
	if r.Status.ObservedGeneration != generation || !reflect.DeepEqual(invalidSpecCond, prevCond) {
		newStatus := r.Status.DeepCopy()
		// SetRolloutCondition only updates the condition when the status and/or reason changes, but
		// the controller should update the invalidSpec if there is a change in why the spec is invalid
		if prevCond != nil && prevCond.Message != invalidSpecCond.Message {
			conditions.RemoveRolloutCondition(newStatus, v1alpha1.InvalidSpec)
		}
		err := c.patchCondition(r, newStatus, invalidSpecCond)
		if err != nil {
			return err
		}
	}
	return nil
}

func (c *rolloutContext) getRolloutReferencedResources() (*validation.ReferencedResources, error) {
	refResources := validation.ReferencedResources{}
	services, err := c.getReferencedServices()
	if err != nil {
		return nil, err
	}
	refResources.ServiceWithType = *services

	analysisTemplates, err := c.getReferencedRolloutAnalyses()
	if err != nil {
		return nil, err
	}
	refResources.AnalysisTemplateWithType = *analysisTemplates

	ingresses, err := c.getReferencedIngresses()
	if err != nil {
		return nil, err
	}
	refResources.Ingresses = *ingresses

	virtualServices, err := c.getReferencedVirtualServices()
	if err != nil {
		return nil, err
	}
	refResources.VirtualServices = *virtualServices

	return &refResources, nil
}

func (c *rolloutContext) getReferencedServices() (*[]validation.ServiceWithType, error) {
	services := []validation.ServiceWithType{}
	if c.rollout.Spec.Strategy.BlueGreen != nil {
		if c.rollout.Spec.Strategy.BlueGreen.ActiveService != "" {
			activeSvc, err := c.servicesLister.Services(c.rollout.Namespace).Get(c.rollout.Spec.Strategy.BlueGreen.ActiveService)
			if k8serrors.IsNotFound(err) {
				fldPath := validation.GetServiceWithTypeFieldPath(validation.ActiveService)
				return nil, field.Invalid(fldPath, c.rollout.Spec.Strategy.BlueGreen.ActiveService, err.Error())
			}
			if err != nil {
				return nil, err
			}
			services = append(services, validation.ServiceWithType{
				Service: activeSvc,
				Type:    validation.ActiveService,
			})
		}
		if c.rollout.Spec.Strategy.BlueGreen.PreviewService != "" {
			previewSvc, err := c.servicesLister.Services(c.rollout.Namespace).Get(c.rollout.Spec.Strategy.BlueGreen.PreviewService)
			if k8serrors.IsNotFound(err) {
				fldPath := validation.GetServiceWithTypeFieldPath(validation.PreviewService)
				return nil, field.Invalid(fldPath, c.rollout.Spec.Strategy.BlueGreen.PreviewService, err.Error())
			}
			if err != nil {
				return nil, err
			}
			services = append(services, validation.ServiceWithType{
				Service: previewSvc,
				Type:    validation.PreviewService,
			})
		}
	} else if c.rollout.Spec.Strategy.Canary != nil {
		if c.rollout.Spec.Strategy.Canary.StableService != "" {
			stableSvc, err := c.servicesLister.Services(c.rollout.Namespace).Get(c.rollout.Spec.Strategy.Canary.StableService)
			if k8serrors.IsNotFound(err) {
				fldPath := validation.GetServiceWithTypeFieldPath(validation.StableService)
				return nil, field.Invalid(fldPath, c.rollout.Spec.Strategy.Canary.StableService, err.Error())
			}
			if err != nil {
				return nil, err
			}
			services = append(services, validation.ServiceWithType{
				Service: stableSvc,
				Type:    validation.StableService,
			})
		}
		if c.rollout.Spec.Strategy.Canary.CanaryService != "" {
			canarySvc, err := c.servicesLister.Services(c.rollout.Namespace).Get(c.rollout.Spec.Strategy.Canary.CanaryService)
			if k8serrors.IsNotFound(err) {
				fldPath := validation.GetServiceWithTypeFieldPath(validation.CanaryService)
				return nil, field.Invalid(fldPath, c.rollout.Spec.Strategy.Canary.CanaryService, err.Error())
			}
			if err != nil {
				return nil, err
			}
			services = append(services, validation.ServiceWithType{
				Service: canarySvc,
				Type:    validation.CanaryService,
			})
		}
	}
	return &services, nil
}

func (c *rolloutContext) getReferencedRolloutAnalyses() (*[]validation.AnalysisTemplateWithType, error) {
	analysisTemplates := []validation.AnalysisTemplateWithType{}
	if c.rollout.Spec.Strategy.BlueGreen != nil {
		blueGreen := c.rollout.Spec.Strategy.BlueGreen
		if blueGreen.PrePromotionAnalysis != nil {
			// CanaryStepIndex will be ignored
			templates, err := c.getReferencedAnalysisTemplates(c.rollout, blueGreen.PrePromotionAnalysis, validation.PrePromotionAnalysis, 0)
			if err != nil {
				return nil, err
			}
			analysisTemplates = append(analysisTemplates, templates...)
		}

		if blueGreen.PostPromotionAnalysis != nil {
			// CanaryStepIndex will be ignored
			templates, err := c.getReferencedAnalysisTemplates(c.rollout, blueGreen.PostPromotionAnalysis, validation.PostPromotionAnalysis, 0)
			if err != nil {
				return nil, err
			}
			analysisTemplates = append(analysisTemplates, templates...)
		}
	} else if c.rollout.Spec.Strategy.Canary != nil {
		canary := c.rollout.Spec.Strategy.Canary
		for i, step := range canary.Steps {
			if step.Analysis != nil {
				templates, err := c.getReferencedAnalysisTemplates(c.rollout, step.Analysis, validation.InlineAnalysis, i)
				if err != nil {
					return nil, err
				}
				analysisTemplates = append(analysisTemplates, templates...)
			}
		}
		if canary.Analysis != nil {
			templates, err := c.getReferencedAnalysisTemplates(c.rollout, &canary.Analysis.RolloutAnalysis, validation.BackgroundAnalysis, 0)
			if err != nil {
				return nil, err
			}
			analysisTemplates = append(analysisTemplates, templates...)
		}
	}
	return &analysisTemplates, nil
}

func (c *rolloutContext) getReferencedAnalysisTemplates(rollout *v1alpha1.Rollout, rolloutAnalysis *v1alpha1.RolloutAnalysis, templateType validation.AnalysisTemplateType, canaryStepIndex int) ([]validation.AnalysisTemplateWithType, error) {
	analysisTemplates := []validation.AnalysisTemplateWithType{}
	if rolloutAnalysis.Templates != nil {
		for i, template := range rolloutAnalysis.Templates {
			analysisTemplate, err := c.getReferencedAnalysisTemplate(template, templateType, i, canaryStepIndex)
			if err != nil {
				return nil, err
			}
			if analysisTemplate != nil {
				analysisTemplates = append(analysisTemplates, *analysisTemplate)
			}
		}
	}
	return analysisTemplates, nil
}

func (c *rolloutContext) getReferencedAnalysisTemplate(template v1alpha1.RolloutAnalysisTemplate, templateType validation.AnalysisTemplateType, analysisIndex int, canaryStepIndex int) (*validation.AnalysisTemplateWithType, error) {
	fldPath := validation.GetAnalysisTemplateWithTypeFieldPath(templateType, analysisIndex, canaryStepIndex)
	if template.ClusterScope {
		clusterAnalysisTemplate, err := c.clusterAnalysisTemplateLister.Get(template.TemplateName)
		if k8serrors.IsNotFound(err) {
			return nil, field.Invalid(fldPath, template.TemplateName, err.Error())
		}
		if err != nil {
			return nil, err
		}
		return &validation.AnalysisTemplateWithType{
			ClusterAnalysisTemplate: clusterAnalysisTemplate,
			TemplateType:            templateType,
			AnalysisIndex:           analysisIndex,
		}, nil
	}
	analysisTemplate, err := c.analysisTemplateLister.AnalysisTemplates(c.rollout.Namespace).Get(template.TemplateName)
	if k8serrors.IsNotFound(err) {
		return nil, field.Invalid(fldPath, template.TemplateName, err.Error())
	}
	if err != nil {
		return nil, err
	}
	return &validation.AnalysisTemplateWithType{
		AnalysisTemplate: analysisTemplate,
		TemplateType:     templateType,
	}, nil
}

func (c *rolloutContext) getReferencedIngresses() (*[]v1beta1.Ingress, error) {
	ingresses := []v1beta1.Ingress{}
	canary := c.rollout.Spec.Strategy.Canary
	fldPath := field.NewPath("spec", "strategy", "canary", "trafficRouting")
	if canary != nil && canary.TrafficRouting != nil {
		if canary.TrafficRouting.ALB != nil {
			ingress, err := c.ingressesLister.Ingresses(c.rollout.Namespace).Get(canary.TrafficRouting.ALB.Ingress)
			if k8serrors.IsNotFound(err) {
				return nil, field.Invalid(fldPath.Child("alb", "ingress"), canary.TrafficRouting.ALB.Ingress, err.Error())
			}
			if err != nil {
				return nil, err
			}
			ingresses = append(ingresses, *ingress)
		} else if canary.TrafficRouting.Nginx != nil {
			ingress, err := c.ingressesLister.Ingresses(c.rollout.Namespace).Get(canary.TrafficRouting.Nginx.StableIngress)
			if k8serrors.IsNotFound(err) {
				return nil, field.Invalid(fldPath.Child("nginx", "stableIngress"), canary.TrafficRouting.Nginx.StableIngress, err.Error())
			}
			if err != nil {
				return nil, err
			}
			ingresses = append(ingresses, *ingress)
		}
	}
	return &ingresses, nil
}

func (c *rolloutContext) getReferencedVirtualServices() (*[]unstructured.Unstructured, error) {
	ctx := context.TODO()
	virtualServices := []unstructured.Unstructured{}
	fldPath := field.NewPath("spec", "strategy", "canary", "trafficRouting", "istio", "virtualService", "name")
	if c.rollout.Spec.Strategy.Canary != nil {
		canary := c.rollout.Spec.Strategy.Canary
		if canary.TrafficRouting != nil && canary.TrafficRouting.Istio != nil {
			var vsvc *unstructured.Unstructured
			var err error
			vsvcName := canary.TrafficRouting.Istio.VirtualService.Name
			if c.istioVirtualServiceInformer.HasSynced() {
				vsvc, err = c.istioVirtualServiceLister.Namespace(c.rollout.Namespace).Get(vsvcName)
			} else {
				vsvc, err = c.dynamicclientset.Resource(istioutil.GetIstioGVR(c.defaultIstioVersion)).Namespace(c.rollout.Namespace).Get(ctx, vsvcName, metav1.GetOptions{})
			}

			if k8serrors.IsNotFound(err) {
				return nil, field.Invalid(fldPath, vsvcName, err.Error())
			}
			if err != nil {
				return nil, err
			}
			virtualServices = append(virtualServices, *vsvc)
		}
	}
	return &virtualServices, nil
}

func (c *Controller) migrateCanaryStableRS(rollout *v1alpha1.Rollout) bool {
	ctx := context.TODO()
	if rollout.Spec.Strategy.Canary == nil {
		return false
	}
	if rollout.Status.StableRS == "" && rollout.Status.Canary.StableRS == "" {
		return false
	}
	if rollout.Status.StableRS != "" && rollout.Status.Canary.StableRS != "" {
		return false
	}
	stableRS := rollout.Status.StableRS
	if rollout.Status.Canary.StableRS != "" {
		stableRS = rollout.Status.Canary.StableRS
	}
	rollout.Status.Canary.StableRS = stableRS
	rollout.Status.StableRS = stableRS
	_, err := c.argoprojclientset.ArgoprojV1alpha1().Rollouts(rollout.Namespace).Update(ctx, rollout, metav1.UpdateOptions{})
	if err != nil {
		logutil.WithRollout(rollout).Errorf("Unable to migrate Rollout's status.canary.stableRS to status.stableRS: %s", err.Error())
	}
	return true
}

func remarshalRollout(r *v1alpha1.Rollout) *v1alpha1.Rollout {
	rolloutBytes, err := json.Marshal(r)
	if err != nil {
		panic(err)
	}
	var remarshalled v1alpha1.Rollout
	err = json.Unmarshal(rolloutBytes, &remarshalled)
	if err != nil {
		panic(err)
	}
	return &remarshalled
}