fix(trafficrouting): ensure DestinationRule is updated before SetWeight on rollback (#4612)

* fix(istio): ensure DestinationRule is updated before SetWeight on rollback

  - Return error from UpdateHash when delaying DR switch so SetWeight is not
    called, preventing VS weight update before DR subset switch (fixes rollback
    traffic hitting wrong subsets).
  - When shifting traffic to stable only (abort or dynamic rollback), only
    require stable RS availability so abort still completes when canary is
    failed (no regression of #4128).

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

* chore: add trafficrouting test for #4612

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

* refactor

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

* trigger build

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

* reduce cognitive complexity

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

* remove unused function

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>

---------

Signed-off-by: Andrew Brown <andrew.brown@wealthsimple.com>
Signed-off-by: Zach Aller <zachaller@users.noreply.github.com>
Co-authored-by: Zach Aller <zachaller@users.noreply.github.com>

Author: andrewjamesbrown

rollout/canary.go

@@ -259,28 +259,6 @@ func (c *rolloutContext) scaleDownOldReplicaSetsForCanary(oldRSs []*appsv1.Repli
 	return totalScaledDown, nil
 }
 
-// isDynamicallyRollingBackToStable returns true if we were in the middle of an canary update with
-// dynamic stable scaling, but was interrupted and are now rolling back to stable RS. This is similar
-// to, but different than aborting. With abort, desired hash != stable hash and so we know the
-// two hashes to balance traffic against. But with dynamically rolling back to stable, the
-// desired hash == stable hash, and so we must use the *previous* desired hash and balance traffic
-// between previous desired vs. stable hash, in order to safely shift traffic back to stable.
-// This function also returns the previous desired hash (where we are weighted to)
-func isDynamicallyRollingBackToStable(ro *v1alpha1.Rollout, desiredRS *appsv1.ReplicaSet) (bool, string) {
-	if rolloututil.IsFullyPromoted(ro) && ro.Spec.Strategy.Canary.TrafficRouting != nil && ro.Spec.Strategy.Canary.DynamicStableScale {
-		if ro.Status.Canary.Weights != nil {
-			currSelector := ro.Status.Canary.Weights.Canary.PodTemplateHash
-			desiredSelector := replicasetutil.GetPodTemplateHash(desiredRS)
-			if currSelector != desiredSelector {
-				if desiredRS.Status.AvailableReplicas < *ro.Spec.Replicas {
-					return true, currSelector
-				}
-			}
-		}
-	}
-	return false, ""
-}
-
 // canProceedWithScaleDownAnnotation returns whether or not it is safe to proceed with annotating
 // old replicasets with the scale-down-deadline in the traffic-routed canary strategy.
 // This method only matters with ALB canary + the target group verification feature.

rollout/canary_test.go

@@ -2107,7 +2107,7 @@ func TestIsDynamicallyRollingBackToStable(t *testing.T) {
 				desiredRS.Status.AvailableReplicas = *tc.rsAvailableReplicas
 			}
 
-			rbToStable, _ := isDynamicallyRollingBackToStable(ro, desiredRS)
+			rbToStable, _ := rolloututil.IsDynamicallyRollingBackToStable(ro, desiredRS)
 
 			assert.Equal(t, tc.expectedResult, rbToStable)
 		})

rollout/controller_test.go

@@ -119,6 +119,8 @@ type fixture struct {
 	// Objects from here preloaded into NewSimpleFake.
 	kubeobjects []runtime.Object
 	objects     []runtime.Object
+	// dynamicOnlyObjects: added to dynamic client only (not Argo client). Use for Istio VS/DR so listers don't see unstructured as rollout.
+	dynamicOnlyObjects []runtime.Object
 	// Acquire 'enqueuedObjectsLock' before accessing enqueuedObjects
 	enqueuedObjects     map[string]int
 	enqueuedObjectsLock sync.Mutex
@@ -127,6 +129,10 @@ type fixture struct {
 	// events holds all the K8s Event Reasons emitted during the run
 	events             []string
 	fakeTrafficRouting *mocks.TrafficRoutingReconciler
+	// reseedRolloutMutator, if set, is applied to the rollout when re-seeding between syncs (for multi-sync tests).
+	reseedRolloutMutator func(*v1alpha1.Rollout)
+	// allowErrorOnLastSync, if set, do not fail the test when the final sync returns an error (e.g. "delaying destination rule switch").
+	allowErrorOnLastSync bool
 }
 
 func newFixture(t *testing.T) *fixture {
@@ -562,6 +568,11 @@ func (f *fixture) newController(resync resyncFunc) (*Controller, informers.Share
 	f.client = fake.NewSimpleClientset(f.objects...)
 	f.kubeclient = k8sfake.NewSimpleClientset(f.kubeobjects...)
 
+	dynamicClientObjects := f.objects
+	if len(f.dynamicOnlyObjects) > 0 {
+		dynamicClientObjects = append(append([]runtime.Object{}, f.objects...), f.dynamicOnlyObjects...)
+	}
+
 	i := informers.NewSharedInformerFactory(f.client, resync())
 	k8sI := kubeinformers.NewSharedInformerFactory(f.kubeclient, resync())
 
@@ -581,7 +592,7 @@ func (f *fixture) newController(resync resyncFunc) (*Controller, informers.Share
 		destGVR: destGVR.Resource + "List",
 	}
 
-	dynamicClient := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(scheme, listMapping, f.objects...)
+	dynamicClient := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(scheme, listMapping, dynamicClientObjects...)
 	dynamicInformerFactory := dynamicinformer.NewDynamicSharedInformerFactory(dynamicClient, 0)
 	istioVirtualServiceInformer := dynamicInformerFactory.ForResource(istioutil.GetIstioVirtualServiceGVR()).Informer()
 	istioDestinationRuleInformer := dynamicInformerFactory.ForResource(istioutil.GetIstioDestinationRuleGVR()).Informer()
@@ -647,13 +658,15 @@ func (f *fixture) newController(resync resyncFunc) (*Controller, informers.Share
 	c.enqueueRolloutAfter = func(obj any, duration time.Duration) {
 		c.enqueueRollout(obj)
 	}
-	c.newTrafficRoutingReconciler = func(roCtx *rolloutContext) ([]trafficrouting.TrafficRoutingReconciler, error) {
-		if roCtx.rollout.Spec.Strategy.Canary == nil || roCtx.rollout.Spec.Strategy.Canary.TrafficRouting == nil {
-			return nil, nil
+	if f.fakeTrafficRouting != nil {
+		c.newTrafficRoutingReconciler = func(roCtx *rolloutContext) ([]trafficrouting.TrafficRoutingReconciler, error) {
+			if roCtx.rollout.Spec.Strategy.Canary == nil || roCtx.rollout.Spec.Strategy.Canary.TrafficRouting == nil {
+				return nil, nil
+			}
+			var reconcilers = []trafficrouting.TrafficRoutingReconciler{}
+			reconcilers = append(reconcilers, f.fakeTrafficRouting)
+			return reconcilers, nil
 		}
-		var reconcilers = []trafficrouting.TrafficRoutingReconciler{}
-		reconcilers = append(reconcilers, f.fakeTrafficRouting)
-		return reconcilers, nil
 	}
 
 	for _, r := range f.rolloutLister {
@@ -698,11 +711,68 @@ func (f *fixture) run(rolloutName string) {
 	f.runController(rolloutName, true, false, c, i, k8sI)
 }
 
+// runWithSyncs runs the controller syncHandler n times (e.g. 2 for two reconciliation loops) then verifies actions.
+func (f *fixture) runWithSyncs(rolloutName string, syncs int) {
+	c, i, k8sI := f.newController(noResyncPeriodFunc)
+	f.runControllerWithSyncs(rolloutName, syncs, true, false, c, i, k8sI)
+}
+
 func (f *fixture) runExpectError(rolloutName string, startInformers bool) {
 	c, i, k8sI := f.newController(noResyncPeriodFunc)
 	f.runController(rolloutName, startInformers, true, c, i, k8sI)
 }
 
+func (f *fixture) runControllerWithSyncs(rolloutName string, syncs int, startInformers bool, expectError bool, c *Controller, i informers.SharedInformerFactory, k8sI kubeinformers.SharedInformerFactory) *Controller {
+	if startInformers {
+		stopCh := make(chan struct{})
+		defer close(stopCh)
+		i.Start(stopCh)
+		k8sI.Start(stopCh)
+		assert.True(f.t, cache.WaitForCacheSync(stopCh, c.replicaSetSynced, c.rolloutsSynced))
+	}
+
+	for n := 0; n < syncs; n++ {
+		err := c.syncHandler(context.Background(), rolloutName)
+		allowErr := f.allowErrorOnLastSync && (n == syncs-1)
+		f.assertSyncHandlerResult(n, syncs, err, expectError, allowErr)
+		f.reseedRolloutInInformerIfNeeded(c, rolloutName, n, syncs)
+	}
+
+	return f.verifyActionsAndReturn(c)
+}
+
+func (f *fixture) assertSyncHandlerResult(n, syncs int, err error, expectError, allowErr bool) {
+	if !expectError && err != nil && !allowErr {
+		f.t.Errorf("error syncing rollout (sync %d/%d): %v", n+1, syncs, err)
+		return
+	}
+	if expectError && err == nil {
+		f.t.Error("expected error syncing rollout, got nil")
+	}
+}
+
+// reseedRolloutInInformer re-seeds the rollout in the informer so the next sync sees a typed Rollout
+// (controller writes Unstructured via persistRolloutToInformer). No-op when syncs <= 1 or on the last sync.
+func (f *fixture) reseedRolloutInInformerIfNeeded(c *Controller, rolloutName string, n, syncs int) {
+	if syncs <= 1 || n >= syncs-1 {
+		return
+	}
+	namespace, name, err := cache.SplitMetaNamespaceKey(rolloutName)
+	if err != nil {
+		f.t.Fatalf("re-seed rollout: split key: %v", err)
+	}
+	ro, err := f.client.ArgoprojV1alpha1().Rollouts(namespace).Get(context.Background(), name, metav1.GetOptions{})
+	if err != nil {
+		f.t.Fatalf("re-seed rollout: get: %v", err)
+	}
+	if f.reseedRolloutMutator != nil {
+		f.reseedRolloutMutator(ro)
+	}
+	if err := c.rolloutsIndexer.Update(ro); err != nil {
+		f.t.Fatalf("re-seed rollout: update indexer: %v", err)
+	}
+}
+
 func (f *fixture) runController(rolloutName string, startInformers bool, expectError bool, c *Controller, i informers.SharedInformerFactory, k8sI kubeinformers.SharedInformerFactory) *Controller {
 	if startInformers {
 		stopCh := make(chan struct{})
@@ -720,6 +790,10 @@ func (f *fixture) runController(rolloutName string, startInformers bool, expectE
 		f.t.Error("expected error syncing rollout, got nil")
 	}
 
+	return f.verifyActionsAndReturn(c)
+}
+
+func (f *fixture) verifyActionsAndReturn(c *Controller) *Controller {
 	actions := filterInformerActions(f.client.Actions())
 	for i, action := range actions {
 		if len(f.actions) < i+1 {
@@ -752,7 +826,6 @@ func (f *fixture) runController(rolloutName string, startInformers bool, expectE
 		f.t.Errorf("%d expected actions did not happen:%+v", len(f.kubeactions)-len(k8sActions), f.kubeactions[len(k8sActions):])
 	}
 	fakeRecorder := c.recorder.(*record.FakeEventRecorder)
-
 	f.events = fakeRecorder.Events()
 	return c
 }
@@ -850,6 +923,11 @@ func (f *fixture) expectUpdatePodAction(p *corev1.Pod) int {
 	return len
 }
 
+func (f *fixture) expectGetRolloutAction(rollout *v1alpha1.Rollout) int {
+	len := len(f.actions)
+	f.actions = append(f.actions, core.NewGetAction(v1alpha1.RolloutGVR, rollout.Namespace, rollout.Name))
+	return len
+}
 func (f *fixture) expectCreateExperimentAction(ex *v1alpha1.Experiment) int {
 	action := core.NewCreateAction(schema.GroupVersionResource{Resource: "experiments"}, ex.Namespace, ex)
 	len := len(f.actions)

rollout/service.go

@@ -269,7 +269,7 @@ func (c *rolloutContext) reconcileStableAndCanaryService() error {
 		return nil
 	}
 
-	if dynamicallyRollingBackToStable, currSelector := isDynamicallyRollingBackToStable(c.rollout, c.newRS); dynamicallyRollingBackToStable {
+	if dynamicallyRollingBackToStable, currSelector := rolloututils.IsDynamicallyRollingBackToStable(c.rollout, c.newRS); dynamicallyRollingBackToStable {
 		// User may have interrupted an update in order go back to stableRS, and is using dynamic
 		// stable scaling. If that is the case, the stableRS might be undersized and if we blindly
 		// switch service selector we could overwhelm stableRS pods.

rollout/trafficrouting.go

@@ -189,7 +189,7 @@ func (c *rolloutContext) reconcileTrafficRouting() error {
 			canaryHash = c.newRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
 		}
 
-		if dynamicallyRollingBackToStable, prevDesiredHash := isDynamicallyRollingBackToStable(c.rollout, c.newRS); dynamicallyRollingBackToStable {
+		if dynamicallyRollingBackToStable, prevDesiredHash := rolloututil.IsDynamicallyRollingBackToStable(c.rollout, c.newRS); dynamicallyRollingBackToStable {
 			desiredWeight = c.calculateDesiredWeightOnAbortOrStableRollback()
 			// Since stableRS == desiredRS, we must balance traffic between the
 			// *previous desired* vs. stable (as opposed to current desired vs. stable).

rollout/trafficrouting/istio/istio.go

@@ -27,6 +27,7 @@ import (
 	istioutil "github.com/argoproj/argo-rollouts/utils/istio"
 	logutil "github.com/argoproj/argo-rollouts/utils/log"
 	"github.com/argoproj/argo-rollouts/utils/record"
+	rolloututil "github.com/argoproj/argo-rollouts/utils/rollout"
 )
 
 const Http = "http"
@@ -36,12 +37,11 @@ const Type = "Istio"
 
 const SpecHttpNotFound = "spec.http not found"
 
-// NewReconciler returns a reconciler struct that brings the Virtual Service into the desired state
+// NewReconciler returns a reconciler struct that brings the Virtual Service into the desired state.
 func NewReconciler(r *v1alpha1.Rollout, client dynamic.Interface, recorder record.EventRecorder, virtualServiceLister, destinationRuleLister dynamiclister.Lister, replicaSets []*appsv1.ReplicaSet) *Reconciler {
 	return &Reconciler{
-		rollout: r,
-		log:     logutil.WithRollout(r),
-
+		rollout:               r,
+		log:                   logutil.WithRollout(r),
 		client:                client,
 		recorder:              recorder,
 		virtualServiceLister:  virtualServiceLister,
@@ -350,21 +350,36 @@ func (r *Reconciler) reconcileVirtualService(obj *unstructured.Unstructured, vsv
 	return newObj, len(patches) > 0, err
 }
 
-func (r *Reconciler) UpdateHash(canaryHash, stableHash string, additionalDestinations ...v1alpha1.WeightDestination) error {
-	// We need to check if the replicasets are ready here as well if we didn't define any services in the rollout
-	// See: https://github.com/argoproj/argo-rollouts/issues/2507
-	if r.rollout.Spec.Strategy.Canary.CanaryService == "" && r.rollout.Spec.Strategy.Canary.StableService == "" {
-
-		for _, rs := range r.replicaSets {
-			if *rs.Spec.Replicas > 0 {
-				rsHash := rs.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
-				// Only check availability for ReplicaSets that will receive traffic
-				if (rsHash == stableHash || rsHash == canaryHash) && rsHash != "" && !replicasetutil.IsReplicaSetAvailable(rs) {
-					r.log.Infof("delaying destination rule switch: ReplicaSet %s not fully available", rs.Name)
-					return nil
-				}
-			}
+// shouldDelayDestinationRuleUpdate returns true if updating the DestinationRule should be
+// delayed because a traffic-receiving ReplicaSet is not yet fully available.
+// See: https://github.com/argoproj/argo-rollouts/issues/2507
+func (r *Reconciler) shouldDelayDestinationRuleUpdate(canaryHash, stableHash string) (bool, string) {
+	if r.rollout.Spec.Strategy.Canary.CanaryService != "" || r.rollout.Spec.Strategy.Canary.StableService != "" {
+		return false, ""
+	}
+	abortOrDynamicRollbackToStable := rolloututil.AbortOrDynamicRollbackToStable(r.rollout, r.replicaSets, stableHash)
+	for _, rs := range r.replicaSets {
+		if *rs.Spec.Replicas == 0 {
+			continue
+		}
+		rsHash := rs.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
+		if rsHash == "" || (rsHash != stableHash && rsHash != canaryHash) {
+			continue
 		}
+		if abortOrDynamicRollbackToStable && rsHash == canaryHash {
+			continue
+		}
+		if !replicasetutil.IsReplicaSetAvailable(rs) {
+			return true, rs.Name
+		}
+	}
+	return false, ""
+}
+
+func (r *Reconciler) UpdateHash(canaryHash, stableHash string, additionalDestinations ...v1alpha1.WeightDestination) error {
+	if shouldDelay, rsName := r.shouldDelayDestinationRuleUpdate(canaryHash, stableHash); shouldDelay {
+		r.log.Infof("delaying destination rule switch: ReplicaSet %s not fully available", rsName)
+		return fmt.Errorf("delaying destination rule switch: ReplicaSet %s not fully available", rsName)
 	}
 
 	dRuleSpec := r.rollout.Spec.Strategy.Canary.TrafficRouting.Istio.DestinationRule