fix: rs conflict with fallback to patch

.codecov.yml

@@ -19,3 +19,5 @@ ignore:
   - 'pkg/client/.*'
   - 'vendor/.*'
   - '**/mocks/*'
+  - 'hack/gen-crd-spec/main.go'
+  - 'hack/gen-docs/main.go'

.github/workflows/e2e.yaml

@@ -95,4 +95,4 @@ jobs:
         with:
           name: e2e-controller-k8s-${{ matrix.kubernetes-minor-version }}.log
           path: /tmp/e2e-controller.log
-        if: ${{ failure() }}
+        if: ${{ always() }}

Makefile

@@ -22,7 +22,7 @@ DEV_IMAGE ?= false
 E2E_INSTANCE_ID ?= argo-rollouts-e2e
 E2E_TEST_OPTIONS ?=
 E2E_PARALLEL ?= 1
-E2E_WAIT_TIMEOUT ?= 120
+E2E_WAIT_TIMEOUT ?= 90
 GOPATH ?= $(shell go env GOPATH)
 
 # Global toolchain configuration

hack/gen-crd-spec/main.go

@@ -96,11 +96,12 @@ func NewCustomResourceDefinition() []*extensionsobj.CustomResourceDefinition {
 	// clean up stuff left by controller-gen
 	deleteFile("config/webhook/manifests.yaml")
 	deleteFile("config/webhook")
-	deleteFile("config/argoproj.io_analysisruns.yaml")
-	deleteFile("config/argoproj.io_analysistemplates.yaml")
-	deleteFile("config/argoproj.io_clusteranalysistemplates.yaml")
-	deleteFile("config/argoproj.io_experiments.yaml")
-	deleteFile("config/argoproj.io_rollouts.yaml")
+	deleteFile("config/crd/argoproj.io_analysisruns.yaml")
+	deleteFile("config/crd/argoproj.io_analysistemplates.yaml")
+	deleteFile("config/crd/argoproj.io_clusteranalysistemplates.yaml")
+	deleteFile("config/crd/argoproj.io_experiments.yaml")
+	deleteFile("config/crd/argoproj.io_rollouts.yaml")
+	deleteFile("config/crd")
 	deleteFile("config")
 
 	crds := []*extensionsobj.CustomResourceDefinition{}

rollout/canary.go

@@ -112,7 +112,7 @@
 	}
 	scaled, _, err := c.scaleReplicaSetAndRecordEvent(c.stableRS, desiredStableRSReplicaCount)
 	if err != nil {
-		return scaled, fmt.Errorf("failed to scaleReplicaSetAndRecordEvent in reconcileCanaryStableReplicaSet:L %w", err)
+		return scaled, fmt.Errorf("failed to scaleReplicaSetAndRecordEvent in reconcileCanaryStableReplicaSet: %w", err)
 	}
 	return scaled, err
 }

rollout/canary_test.go

@@ -4,10 +4,16 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
 	"strconv"
 	"testing"
 	"time"
 
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8stesting "k8s.io/client-go/testing"
+
 	"github.com/stretchr/testify/assert"
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/apps/v1"
@@ -2141,3 +2147,106 @@ func TestCanaryReplicaAndSpecChangedTogether(t *testing.T) {
 	// check the canary one is updated
 	assert.NotEqual(t, originReplicas, int(*updated.Spec.Replicas))
 }
+
+func TestSyncRolloutWithConflictInScaleReplicaSet(t *testing.T) {
+	os.Setenv("ARGO_ROLLOUTS_LOG_RS_DIFF_CONFLICT", "true")
+	defer os.Unsetenv("ARGO_ROLLOUTS_LOG_RS_DIFF_CONFLICT")
+
+	f := newFixture(t)
+	defer f.Close()
+
+	steps := []v1alpha1.CanaryStep{
+		{
+			SetWeight: int32Ptr(10),
+		}, {
+			Pause: &v1alpha1.RolloutPause{
+				Duration: v1alpha1.DurationFromInt(10),
+			},
+		},
+	}
+	r1 := newCanaryRollout("foo", 10, nil, steps, int32Ptr(1), intstr.FromInt(1), intstr.FromInt(0))
+	r1.Spec.Template.Labels["rollout.argoproj.io/foo"] = "bar"
+
+	rs1 := newReplicaSetWithStatus(r1, 10, 10)
+	r1.Spec.Replicas = pointer.Int32(2)
+	f.kubeobjects = append(f.kubeobjects, rs1)
+	f.replicaSetLister = append(f.replicaSetLister, rs1)
+
+	f.rolloutLister = append(f.rolloutLister, r1)
+	f.objects = append(f.objects, r1)
+
+	f.expectPatchRolloutAction(r1)
+	f.expectUpdateReplicaSetAction(rs1)              // attempt to scale replicaset but conflict
+	patchIndex := f.expectPatchReplicaSetAction(rs1) // instead of update patch replicaset
+
+	key := fmt.Sprintf("%s/%s", r1.Namespace, r1.Name)
+	c, i, k8sI := f.newController(func() time.Duration { return 30 * time.Minute })
+
+	f.kubeclient.PrependReactor("update", "replicasets", func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+		return true, action.(k8stesting.UpdateAction).GetObject(), errors.NewConflict(schema.GroupResource{
+			Group:    "Apps",
+			Resource: "ReplicaSet",
+		}, action.(k8stesting.UpdateAction).GetObject().(*appsv1.ReplicaSet).Name, fmt.Errorf("test error"))
+	})
+
+	f.runController(key, true, false, c, i, k8sI)
+
+	updatedRs := f.getPatchedReplicaSet(patchIndex) // minus one because update did not happen because conflict
+	assert.Equal(t, int32(2), *updatedRs.Spec.Replicas)
+}
+
+func TestSyncRolloutWithConflictInSyncReplicaSetRevision(t *testing.T) {
+	os.Setenv("ARGO_ROLLOUTS_LOG_RS_DIFF_CONFLICT", "true")
+	defer os.Unsetenv("ARGO_ROLLOUTS_LOG_RS_DIFF_CONFLICT")
+
+	f := newFixture(t)
+	defer f.Close()
+
+	steps := []v1alpha1.CanaryStep{
+		{
+			SetWeight: int32Ptr(10),
+		}, {
+			Pause: &v1alpha1.RolloutPause{
+				Duration: v1alpha1.DurationFromInt(10),
+			},
+		},
+	}
+	r1 := newCanaryRollout("foo", 3, nil, steps, int32Ptr(1), intstr.FromInt(1), intstr.FromInt(0))
+	r2 := bumpVersion(r1)
+
+	rs1 := newReplicaSetWithStatus(r1, 3, 3)
+	rs2 := newReplicaSetWithStatus(r2, 3, 3)
+	rs2.Annotations["rollout.argoproj.io/revision"] = "1"
+
+	f.kubeobjects = append(f.kubeobjects, rs1, rs2)
+	f.replicaSetLister = append(f.replicaSetLister, rs1, rs2)
+
+	f.rolloutLister = append(f.rolloutLister, r2)
+	f.objects = append(f.objects, r2)
+
+	key := fmt.Sprintf("%s/%s", r1.Namespace, r1.Name)
+	c, i, k8sI := f.newController(func() time.Duration { return 30 * time.Minute })
+
+	f.kubeclient.PrependReactor("update", "replicasets", func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+		return true, &appsv1.ReplicaSet{}, errors.NewConflict(schema.GroupResource{
+			Group:    "Apps",
+			Resource: "ReplicaSet",
+		}, action.(k8stesting.UpdateAction).GetObject().(*appsv1.ReplicaSet).Name, fmt.Errorf("test error"))
+	})
+
+	f.expectPatchRolloutAction(r2)
+	f.expectUpdateReplicaSetAction(rs1)               // attempt to update replicaset revision but conflict
+	patchIndex1 := f.expectPatchReplicaSetAction(rs1) // instead of update patch replicaset
+
+	f.expectUpdateReplicaSetAction(rs2)               // attempt to scale replicaset but conflict
+	patchIndex2 := f.expectPatchReplicaSetAction(rs2) // instead of update patch replicaset
+
+	f.runController(key, true, false, c, i, k8sI)
+
+	updatedRs1 := f.getPatchedReplicaSet(patchIndex1)
+	assert.Equal(t, "2", updatedRs1.Annotations["rollout.argoproj.io/revision"])
+	assert.Equal(t, int32(3), *updatedRs1.Spec.Replicas)
+
+	updatedRs2 := f.getPatchedReplicaSet(patchIndex2)
+	assert.Equal(t, int32(0), *updatedRs2.Spec.Replicas)
+}

rollout/controller.go

@@ -4,23 +4,30 @@
 	"context"
 	"encoding/json"
 	"fmt"
+	"os"
 	"reflect"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/argoproj/argo-rollouts/utils/annotations"
+
+	"github.com/argoproj/argo-rollouts/utils/diff"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 
 	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts"
 	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/apimachinery/pkg/api/errors"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	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/types"
+	patchtypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
@@ -937,3 +944,94 @@
 	}
 	return &remarshalled
 }
+
+// updateReplicaSetWithPatch updates the replicaset using Update and on failure falls back to a patch this function only exists to make sure we always can update
+// replicasets and to not get into an conflict loop updating replicasets. We should really look into a complete refactor of how rollouts handles replicasets such
+// that we do not keep a fully replicaset on the rollout context under newRS and instead switch to a patch only based approach.
+func (c *rolloutContext) updateReplicaSetFallbackToPatch(ctx context.Context, rs *appsv1.ReplicaSet) (*appsv1.ReplicaSet, error) {
+	updatedRS, err := c.kubeclientset.AppsV1().ReplicaSets(rs.Namespace).Update(ctx, rs, metav1.UpdateOptions{})
+	if err != nil {
+		if errors.IsConflict(err) {
+			if os.Getenv("ARGO_ROLLOUTS_LOG_RS_DIFF_CONFLICT") == "true" {
+				rsGet, err := c.replicaSetLister.ReplicaSets(rs.Namespace).Get(rs.Name)
+				if err != nil {
+					return nil, fmt.Errorf("error getting replicaset in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+				}
+				rsGetJson, err := json.Marshal(rsGet)
+				if err != nil {
+					return nil, fmt.Errorf("error marshalling informer replicaset in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+				}
+				rsCopyJson, err := json.Marshal(rs)
+				if err != nil {
+					return nil, fmt.Errorf("error marshalling memory replicaset in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+				}
+				c.log.Infof("Informer RS: %s", rsGetJson)
+				c.log.Infof("Memory   RS: %s", rsCopyJson)
+			}
+
+			c.log.Infof("Conflict when updating replicaset %s, falling back to patch", rs.Name)
+
+			patchRS := appsv1.ReplicaSet{}
+			patchRS.Spec.Replicas = rs.Spec.Replicas
+			patchRS.Spec.Template.Labels = rs.Spec.Template.Labels
+			patchRS.Spec.Template.Annotations = rs.Spec.Template.Annotations
+
+			patchRS.Annotations = make(map[string]string)
+			patchRS.Labels = make(map[string]string)
+			patchRS.Spec.Selector = &metav1.LabelSelector{
+				MatchLabels: make(map[string]string),
+			}
+
+			if _, found := rs.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]; found {
+				patchRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey] = rs.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
+			}
+
+			if _, found := rs.Annotations[v1alpha1.DefaultReplicaSetScaleDownDeadlineAnnotationKey]; found {
+				patchRS.Annotations[v1alpha1.DefaultReplicaSetScaleDownDeadlineAnnotationKey] = rs.Labels[v1alpha1.DefaultReplicaSetScaleDownDeadlineAnnotationKey]
+			}
+
+			if _, found := rs.Spec.Selector.MatchLabels[v1alpha1.DefaultRolloutUniqueLabelKey]; found {
+				patchRS.Spec.Selector.MatchLabels[v1alpha1.DefaultRolloutUniqueLabelKey] = rs.Spec.Selector.MatchLabels[v1alpha1.DefaultRolloutUniqueLabelKey]
+			}
+
+			for key, value := range rs.Annotations {
+				if strings.HasPrefix(key, annotations.RolloutLabel) ||
+					strings.HasPrefix(key, "argo-rollouts.argoproj.io") ||
+					strings.HasPrefix(key, "experiment.argoproj.io") {
+					patchRS.Annotations[key] = value
+				}
+			}
+			for key, value := range rs.Labels {
+				if strings.HasPrefix(key, annotations.RolloutLabel) ||
+					strings.HasPrefix(key, "argo-rollouts.argoproj.io") ||
+					strings.HasPrefix(key, "experiment.argoproj.io") {
+					patchRS.Labels[key] = value
+				}
+			}
+
+			patch, changed, err := diff.CreateTwoWayMergePatch(appsv1.ReplicaSet{}, patchRS, appsv1.ReplicaSet{})
+			if err != nil {
+				return nil, fmt.Errorf("error creating patch for conflict log in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+			}
+
+			if changed {
+				c.log.Infof("Patching replicaset with patch: %s", string(patch))
+				updatedRS, err = c.kubeclientset.AppsV1().ReplicaSets(rs.Namespace).Patch(ctx, rs.Name, patchtypes.StrategicMergePatchType, patch, metav1.PatchOptions{})
+				if err != nil {
+					return nil, fmt.Errorf("error patching replicaset in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+				}
+			}
+
+			err = c.replicaSetInformer.GetIndexer().Update(updatedRS)
+			if err != nil {
+				return nil, fmt.Errorf("error updating replicaset informer in updateReplicaSetFallbackToPatch %s: %w", rs.Name, err)
+			}
+
+			return updatedRS, err
+		}
+	}
+	if updatedRS != nil {
+		updatedRS.DeepCopyInto(rs)
+	}
+	return rs, err
+}

rollout/controller_test.go

@@ -792,6 +792,12 @@ func (f *fixture) expectPatchServiceAction(s *corev1.Service, newLabel string) i
 	return len
 }
 
+func (f *fixture) expectGetReplicaSetAction(r *appsv1.ReplicaSet) int { //nolint:unused
+	len := len(f.kubeactions)
+	f.kubeactions = append(f.kubeactions, core.NewGetAction(schema.GroupVersionResource{Resource: "replicasets"}, r.Namespace, r.Name))
+	return len
+}
+
 func (f *fixture) expectCreateReplicaSetAction(r *appsv1.ReplicaSet) int {
 	len := len(f.kubeactions)
 	f.kubeactions = append(f.kubeactions, core.NewCreateAction(schema.GroupVersionResource{Resource: "replicasets"}, r.Namespace, r))
@@ -950,6 +956,21 @@ func (f *fixture) getUpdatedReplicaSet(index int) *appsv1.ReplicaSet {
 	return rs
 }
 
+func (f *fixture) getPatchedReplicaSet(index int) *appsv1.ReplicaSet {
+	action := filterInformerActions(f.kubeclient.Actions())[index]
+	patchAction, ok := action.(core.PatchAction)
+	if !ok {
+		f.t.Fatalf("Expected Patch action, not %s", action.GetVerb())
+	}
+
+	rs := appsv1.ReplicaSet{}
+	err := json.Unmarshal(patchAction.GetPatch(), &rs)
+	if err != nil {
+		panic(err)
+	}
+	return &rs
+}
+
 func (f *fixture) verifyPatchedReplicaSet(index int, scaleDownDelaySeconds int32) {
 	action := filterInformerActions(f.kubeclient.Actions())[index]
 	patchAction, ok := action.(core.PatchAction)

rollout/ephemeralmetadata.go

@@ -82,14 +82,12 @@
 	}
 
 	// 2. Update ReplicaSet so that any new pods it creates will have the metadata
-	rs, err = c.kubeclientset.AppsV1().ReplicaSets(modifiedRS.Namespace).Update(ctx, modifiedRS, metav1.UpdateOptions{})
+	rs, err = c.updateReplicaSetFallbackToPatch(ctx, modifiedRS)
 	if err != nil {
-		return fmt.Errorf("error updating replicaset in syncEphemeralMetadata: %w", err)
-	}
-	err = c.replicaSetInformer.GetIndexer().Update(rs)
-	if err != nil {
-		return fmt.Errorf("error updating replicaset informer in syncEphemeralMetadata: %w", err)
+		c.log.Infof("failed to sync ephemeral metadata %v to ReplicaSet %s: %v", podMetadata, rs.Name, err)
+		return fmt.Errorf("failed to sync ephemeral metadata: %w", err)
 	}
+
 	c.log.Infof("synced ephemeral metadata %v to ReplicaSet %s", podMetadata, rs.Name)
 	return nil
 }