Fix rs conflict with fallback to patch replicaset refactor

CHANGELOG.md

@@ -1,3 +1,18 @@
+<a name="v1.7.2"></a>
+## [v1.7.2](https://github.com/argoproj/argo-rollouts/compare/v1.7.1...v1.7.2) (2024-08-12)
+
+### Fix
+
+* replicaSet not scaled down due to incorrect annotations ([#3762](https://github.com/argoproj/argo-rollouts/issues/3762)) ([#3784](https://github.com/argoproj/argo-rollouts/issues/3784))
+* add update verb to ClusterRole permissions for scaleDown feature. Fixes [#3672](https://github.com/argoproj/argo-rollouts/issues/3672) ([#3675](https://github.com/argoproj/argo-rollouts/issues/3675))
+* **analysis:** explicitly set datadog aggregator to last only on v2 ([#3730](https://github.com/argoproj/argo-rollouts/issues/3730))
+* **analysis:** Take RollbackWindow into account when Reconciling Analysis Runs. Fixes [#3669](https://github.com/argoproj/argo-rollouts/issues/3669) ([#3670](https://github.com/argoproj/argo-rollouts/issues/3670))
+* **controller:** Get the right resourceName for traefik.io.Fixes [#3615](https://github.com/argoproj/argo-rollouts/issues/3615) ([#3759](https://github.com/argoproj/argo-rollouts/issues/3759))
+* **controller:** use the stableRS from the rollout context rather tha… ([#3664](https://github.com/argoproj/argo-rollouts/issues/3664))
+* **dashboard:** Update pod status logic to support native sidecars. Fixes [#3366](https://github.com/argoproj/argo-rollouts/issues/3366) ([#3639](https://github.com/argoproj/argo-rollouts/issues/3639))
+* **metricprovider:** reuse http.Transport for http.Client ([#3780](https://github.com/argoproj/argo-rollouts/issues/3780))
+
+
 <a name="v1.7.1"></a>
 ## [v1.7.1](https://github.com/argoproj/argo-rollouts/compare/v1.7.0...v1.7.1) (2024-06-22)
 

cmd/rollouts-controller/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 	"strings"
 	"time"
@@ -79,6 +80,7 @@ func newCommand() *cobra.Command {
 		printVersion                   bool
 		selfServiceNotificationEnabled bool
 		controllersEnabled             []string
+		pprofAddress                   string
 	)
 	electOpts := controller.NewLeaderElectionOptions()
 	var command = cobra.Command{
@@ -204,6 +206,11 @@ func newCommand() *cobra.Command {
 			ingressWrapper, err := ingressutil.NewIngressWrapper(mode, kubeClient, kubeInformerFactory)
 			checkError(err)
 
+			if pprofAddress != "" {
+				mux := controller.NewPProfServer()
+				go func() { log.Println(http.ListenAndServe(pprofAddress, mux)) }()
+			}
+
 			var cm *controller.Manager
 
 			enabledControllers, err := getEnabledControllers(controllersEnabled)
@@ -310,6 +317,7 @@ func newCommand() *cobra.Command {
 	command.Flags().DurationVar(&electOpts.LeaderElectionRetryPeriod, "leader-election-retry-period", controller.DefaultLeaderElectionRetryPeriod, "The duration the clients should wait between attempting acquisition and renewal of a leadership. This is only applicable if leader election is enabled.")
 	command.Flags().BoolVar(&selfServiceNotificationEnabled, "self-service-notification-enabled", false, "Allows rollouts controller to pull notification config from the namespace that the rollout resource is in. This is useful for self-service notification.")
 	command.Flags().StringSliceVar(&controllersEnabled, "controllers", nil, "Explicitly specify the list of controllers to run, currently only supports 'analysis', eg. --controller=analysis. Default: all controllers are enabled")
+	command.Flags().StringVar(&pprofAddress, "enable-pprof-address", "", "Enable pprof profiling on controller by providing a server address.")
 	return &command
 }
 

controller/profiling.go

@@ -0,0 +1,24 @@
+package controller
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/pprof"
+)
+
+const (
+	ProfilingPath = "/debug/pprof"
+)
+
+// NewPProfServer returns a new pprof server to gather runtime profiling data
+func NewPProfServer() *http.ServeMux {
+	mux := http.NewServeMux()
+
+	mux.HandleFunc(ProfilingPath, pprof.Index)
+	mux.HandleFunc(fmt.Sprintf("%s/cmdline", ProfilingPath), pprof.Cmdline)
+	mux.HandleFunc(fmt.Sprintf("%s/profile", ProfilingPath), pprof.Profile)
+	mux.HandleFunc(fmt.Sprintf("%s/symbol", ProfilingPath), pprof.Symbol)
+	mux.HandleFunc(fmt.Sprintf("%s/trace", ProfilingPath), pprof.Trace)
+
+	return mux
+}

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/grpc-gateway v1.16.0
 	github.com/hashicorp/go-plugin v1.6.1
-	github.com/influxdata/influxdb-client-go/v2 v2.13.0
+	github.com/influxdata/influxdb-client-go/v2 v2.14.0
 	github.com/juju/ansiterm v1.0.0
 	github.com/machinebox/graphql v0.2.2
 	github.com/mitchellh/mapstructure v1.5.0

go.sum

@@ -435,8 +435,8 @@ github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
-github.com/influxdata/influxdb-client-go/v2 v2.13.0 h1:ioBbLmR5NMbAjP4UVA5r9b5xGjpABD7j65pI8kFphDM=
-github.com/influxdata/influxdb-client-go/v2 v2.13.0/go.mod h1:k+spCbt9hcvqvUiz0sr5D8LolXHqAAOfPw9v/RIRHl4=
+github.com/influxdata/influxdb-client-go/v2 v2.14.0 h1:AjbBfJuq+QoaXNcrova8smSjwJdUHnwvfjMF71M1iI4=
+github.com/influxdata/influxdb-client-go/v2 v2.14.0/go.mod h1:Ahpm3QXKMJslpXl3IftVLVezreAUtBOTZssDrjZEFHI=
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf h1:7JTmneyiNEwVBOHSjoMxiWAqB992atOeepeFYegn5RU=
 github.com/influxdata/line-protocol v0.0.0-20210922203350-b1ad95c89adf/go.mod h1:xaLFMmpvUxqXtVkUJfg9QmT88cDaCJ3ZKgdZ78oO8Qo=
 github.com/jaytaylor/html2text v0.0.0-20190408195923-01ec452cbe43/go.mod h1:CVKlgaMiht+LXvHG173ujK6JUhZXKb2u/BQtjPDIvyk=

rollout/analysis_test.go

@@ -406,7 +406,7 @@ func TestInvalidSpecMissingClusterTemplatesBackgroundAnalysis(t *testing.T) {
 	f.objects = append(f.objects, r)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	expectedPatchWithoutSub := `{
 		"status": {
@@ -961,7 +961,7 @@ func TestFailCreateStepAnalysisRunIfInvalidTemplateRef(t *testing.T) {
 	f.objects = append(f.objects, r, at)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	expectedPatchWithoutSub := `{
 		"status": {
@@ -1006,7 +1006,7 @@ func TestFailCreateBackgroundAnalysisRunIfInvalidTemplateRef(t *testing.T) {
 	f.objects = append(f.objects, r, at)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	expectedPatchWithoutSub := `{
 		"status": {
@@ -1055,7 +1055,7 @@ func TestFailCreateBackgroundAnalysisRunIfMetricRepeated(t *testing.T) {
 	f.objects = append(f.objects, r, at, at2)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	expectedPatchWithoutSub := `{
 		"status": {

rollout/bluegreen.go

@@ -21,7 +21,7 @@ func (c *rolloutContext) rolloutBlueGreen() error {
 	if err != nil {
 		return err
 	}
-	c.newRS, err = c.getAllReplicaSetsAndSyncRevision(true)
+	c.newRS, err = c.getAllReplicaSetsAndSyncRevision()
 	if err != nil {
 		return fmt.Errorf("failed to getAllReplicaSetsAndSyncRevision in rolloutBlueGreen create true: %w", err)
 	}

rollout/bluegreen_test.go

@@ -950,8 +950,10 @@ func TestBlueGreenRolloutStatusHPAStatusFieldsNoActiveSelector(t *testing.T) {
 	f := newFixture(t)
 	defer f.Close()
 	f.objects = append(f.objects, ro)
+	f.kubeobjects = append(f.kubeobjects, activeSvc)
 	f.rolloutLister = append(f.rolloutLister, ro)
 	f.replicaSetLister = append(f.replicaSetLister, rs)
+	f.serviceLister = append(f.serviceLister, activeSvc)
 
 	ctrl, _, _ := f.newController(noResyncPeriodFunc)
 	roCtx, err := ctrl.newRolloutContext(ro)

rollout/canary.go

@@ -20,14 +20,14 @@ import (
 func (c *rolloutContext) rolloutCanary() error {
 	var err error
 	if replicasetutil.PodTemplateOrStepsChanged(c.rollout, c.newRS) {
-		c.newRS, err = c.getAllReplicaSetsAndSyncRevision(false)
+		c.newRS, err = c.getAllReplicaSetsAndSyncRevision()
 		if err != nil {
 			return fmt.Errorf("failed to getAllReplicaSetsAndSyncRevision in rolloutCanary with PodTemplateOrStepsChanged: %w", err)
 		}
 		return c.syncRolloutStatusCanary()
 	}
 
-	c.newRS, err = c.getAllReplicaSetsAndSyncRevision(true)
+	c.newRS, err = c.getAllReplicaSetsAndSyncRevision()
 	if err != nil {
 		return fmt.Errorf("failed to getAllReplicaSetsAndSyncRevision in rolloutCanary create true: %w", err)
 	}
@@ -448,13 +448,6 @@ func (c *rolloutContext) reconcileCanaryReplicaSets() (bool, error) {
 		return true, nil
 	}
 
-	// If we have updated both the replica count and the pod template hash c.newRS will be nil we want to reconcile the newRS so we look at the
-	// rollout status to get the newRS to reconcile it.
-	if c.newRS == nil && c.rollout.Status.CurrentPodHash != c.rollout.Status.StableRS {
-		rs, _ := replicasetutil.GetReplicaSetByTemplateHash(c.allRSs, c.rollout.Status.CurrentPodHash)
-		c.newRS = rs
-	}
-
 	scaledNewRS, err := c.reconcileNewReplicaSet()
 	if err != nil {
 		return false, err

rollout/canary_test.go

@@ -1,7 +1,6 @@
 package rollout
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -421,8 +420,11 @@ func TestResetCurrentStepIndexOnStepChange(t *testing.T) {
 	f.rolloutLister = append(f.rolloutLister, r2)
 	f.objects = append(f.objects, r2)
 
+	f.expectUpdateRolloutStatusAction(r2)
 	patchIndex := f.expectPatchRolloutAction(r2)
+	createRSIndex := f.expectCreateReplicaSetAction(rs1)
 	f.run(getKey(r2, t))
+	createdRS := f.getCreatedReplicaSet(createRSIndex)
 
 	patch := f.getPatchedRollout(patchIndex)
 	expectedPatchWithoutPodHash := `{
@@ -433,7 +435,7 @@ func TestResetCurrentStepIndexOnStepChange(t *testing.T) {
 			"conditions": %s
 		}
 	}`
-	newConditions := generateConditionsPatch(true, conditions.ReplicaSetUpdatedReason, r2, false, "", false)
+	newConditions := generateConditionsPatch(true, conditions.ReplicaSetUpdatedReason, createdRS, false, "", false)
 	expectedPatch := fmt.Sprintf(expectedPatchWithoutPodHash, expectedCurrentPodHash, expectedCurrentStepHash, newConditions)
 	assert.JSONEq(t, calculatePatch(r2, expectedPatch), patch)
 }
@@ -462,18 +464,23 @@ func TestResetCurrentStepIndexOnPodSpecChange(t *testing.T) {
 	f.rolloutLister = append(f.rolloutLister, r2)
 	f.objects = append(f.objects, r2)
 
+	f.expectUpdateRolloutStatusAction(r2)
 	patchIndex := f.expectPatchRolloutAction(r2)
+	createdRSIndex := f.expectCreateReplicaSetAction(rs1)
+
 	f.run(getKey(r2, t))
 
 	patch := f.getPatchedRollout(patchIndex)
+	updatedRS := f.getUpdatedReplicaSet(createdRSIndex)
+
 	expectedPatchWithoutPodHash := `{
 		"status": {
 			"currentStepIndex":0,
 			"currentPodHash": "%s",
 			"conditions": %s
 		}
 	}`
-	newConditions := generateConditionsPatch(true, conditions.ReplicaSetUpdatedReason, r2, false, "", false)
+	newConditions := generateConditionsPatch(true, conditions.ReplicaSetUpdatedReason, updatedRS, false, "", false)
 
 	expectedPatch := fmt.Sprintf(expectedPatchWithoutPodHash, expectedCurrentPodHash, newConditions)
 	assert.JSONEq(t, calculatePatch(r2, expectedPatch), patch)
@@ -1564,7 +1571,7 @@ func TestCanaryRolloutWithInvalidCanaryServiceName(t *testing.T) {
 	f.kubeobjects = append(f.kubeobjects, rs)
 
 	patchIndex := f.expectPatchRolloutAction(rollout)
-	f.run(getKey(rollout, t))
+	f.runExpectError(getKey(rollout, t), true)
 
 	patch := make(map[string]any)
 	patchData := f.getPatchedRollout(patchIndex)
@@ -1616,7 +1623,7 @@ func TestCanaryRolloutWithInvalidStableServiceName(t *testing.T) {
 	f.kubeobjects = append(f.kubeobjects, rs)
 
 	patchIndex := f.expectPatchRolloutAction(rollout)
-	f.run(getKey(rollout, t))
+	f.runExpectError(getKey(rollout, t), true)
 
 	patch := make(map[string]any)
 	patchData := f.getPatchedRollout(patchIndex)
@@ -1665,7 +1672,7 @@ func TestCanaryRolloutWithInvalidPingServiceName(t *testing.T) {
 	f.objects = append(f.objects, r)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	patch := make(map[string]any)
 	patchData := f.getPatchedRollout(patchIndex)
@@ -1697,7 +1704,7 @@ func TestCanaryRolloutWithInvalidPongServiceName(t *testing.T) {
 	f.serviceLister = append(f.serviceLister, pingSvc)
 
 	patchIndex := f.expectPatchRolloutAction(r)
-	f.run(getKey(r, t))
+	f.runExpectError(getKey(r, t), true)
 
 	patch := make(map[string]any)
 	patchData := f.getPatchedRollout(patchIndex)
@@ -1896,8 +1903,14 @@ func TestHandleNilNewRSOnScaleAndImageChange(t *testing.T) {
 	f.rolloutLister = append(f.rolloutLister, r2)
 	f.objects = append(f.objects, r2)
 
-	f.expectUpdateReplicaSetAction(rs1)
+	f.expectUpdateRolloutStatusAction(r2)
+	f.expectPatchRolloutAction(r2)
 	patchIndex := f.expectPatchRolloutAction(r2)
+
+	f.expectCreateReplicaSetAction(rs1)
+	f.expectUpdateReplicaSetAction(rs1)
+	f.expectUpdateReplicaSetAction(rs1)
+
 	f.run(getKey(r2, t))
 	patch := f.getPatchedRollout(patchIndex)
 	assert.JSONEq(t, calculatePatch(r2, OnlyObservedGenerationPatch), patch)
@@ -2105,49 +2118,6 @@ func TestIsDynamicallyRollingBackToStable(t *testing.T) {
 	}
 }
 
-func TestCanaryReplicaAndSpecChangedTogether(t *testing.T) {
-	f := newFixture(t)
-	defer f.Close()
-
-	originReplicas := 3
-	r1 := newCanaryRollout("foo", originReplicas, nil, nil, nil, intstr.FromInt(1), intstr.FromInt(0))
-	canarySVCName := "canary"
-	stableSVCName := "stable"
-	r1.Spec.Strategy.Canary.CanaryService = canarySVCName
-	r1.Spec.Strategy.Canary.StableService = stableSVCName
-
-	stableRS := newReplicaSetWithStatus(r1, originReplicas, originReplicas)
-	stableSVC := newService(stableSVCName, 80,
-		map[string]string{v1alpha1.DefaultRolloutUniqueLabelKey: stableRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]}, r1)
-
-	r2 := bumpVersion(r1)
-	canaryRS := newReplicaSetWithStatus(r2, originReplicas, originReplicas)
-	canarySVC := newService(canarySVCName, 80,
-		map[string]string{v1alpha1.DefaultRolloutUniqueLabelKey: canaryRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]}, r2)
-
-	f.replicaSetLister = append(f.replicaSetLister, canaryRS, stableRS)
-	f.serviceLister = append(f.serviceLister, canarySVC, stableSVC)
-
-	r3 := bumpVersion(r2)
-	r3.Spec.Replicas = pointer.Int32(int32(originReplicas) + 5)
-	r3.Status.StableRS = stableRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
-	r3.Status.CurrentPodHash = canaryRS.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]
-
-	f.rolloutLister = append(f.rolloutLister, r3)
-	f.kubeobjects = append(f.kubeobjects, canaryRS, stableRS, canarySVC, stableSVC)
-	f.objects = append(f.objects, r3)
-
-	ctrl, _, _ := f.newController(noResyncPeriodFunc)
-	roCtx, err := ctrl.newRolloutContext(r3)
-	assert.NoError(t, err)
-	err = roCtx.reconcile()
-	assert.NoError(t, err)
-	updated, err := f.kubeclient.AppsV1().ReplicaSets(r3.Namespace).Get(context.Background(), canaryRS.Name, metav1.GetOptions{})
-	assert.NoError(t, err)
-	// 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")

rollout/context.go

@@ -1,14 +1,10 @@
 package rollout
 
 import (
-	"time"
-
-	log "github.com/sirupsen/logrus"
-	appsv1 "k8s.io/api/apps/v1"
-	"k8s.io/apimachinery/pkg/util/validation/field"
-
 	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	analysisutil "github.com/argoproj/argo-rollouts/utils/analysis"
+	log "github.com/sirupsen/logrus"
+	appsv1 "k8s.io/api/apps/v1"
 )
 
 type rolloutContext struct {
@@ -53,19 +49,7 @@ type rolloutContext struct {
 }
 
 func (c *rolloutContext) reconcile() error {
-	// Get Rollout Validation errors
-	err := c.getRolloutValidationErrors()
-	if err != nil {
-		if vErr, ok := err.(*field.Error); ok {
-			// We want to frequently requeue rollouts with InvalidSpec errors, because the error
-			// condition might be timing related (e.g. the Rollout was applied before the Service).
-			c.enqueueRolloutAfter(c.rollout, 20*time.Second)
-			return c.createInvalidRolloutCondition(vErr, c.rollout)
-		}
-		return err
-	}
-
-	err = c.checkPausedConditions()
+	err := c.checkPausedConditions()
 	if err != nil {
 		return err
 	}