feat: add the max traffic weight support for the traffic routing (ngi…

…nx/plugins). (#3215) * add support for the traffic weight > 100. Signed-off-by: Liming Liu <[email protected]> * unit test for the max weight ingress. Signed-off-by: Liming Liu <[email protected]> * unit test for the max weight in canary rollout. Signed-off-by: Liming Liu <[email protected]> * add docs Signed-off-by: Zach Aller <[email protected]> * add docs Signed-off-by: Zach Aller <[email protected]> --------- Signed-off-by: Liming Liu <[email protected]> Signed-off-by: Zach Aller <[email protected]> Co-authored-by: Zach Aller <[email protected]>
argoproj · Mar 6, 2024 · 017537b · 017537b
1 parent 9de465f
commit 017537b
Show file tree

Hide file tree

Showing 20 changed files with 850 additions and 555 deletions.
diff --git a/docs/features/specification.md b/docs/features/specification.md
@@ -386,6 +386,9 @@ spec:
       # will achieve traffic split via a weighted replica counts between
       # the canary and stable ReplicaSet.
       trafficRouting:
+        # Supports nginx and plugins only: This lets you control the denominator or total weight of traffic.
+        # The total weight of traffic. If unspecified, it defaults to 100
+        maxTrafficWeight: 1000
         # This is a list of routes that Argo Rollouts has the rights to manage it is currently only required for
         # setMirrorRoute and setHeaderRoute. The order of managedRoutes array also sets the precedence of the route
         # in the traffic router. Argo Rollouts will place these routes in the order specified above any routes already

diff --git a/manifests/crds/rollout-crd.yaml b/manifests/crds/rollout-crd.yaml
@@ -923,6 +923,9 @@ spec:
                               - name
                               type: object
                             type: array
+                          maxTrafficWeight:
+                            format: int32
+                            type: integer
                           nginx:
                             properties:
                               additionalIngressAnnotations:

diff --git a/manifests/install.yaml b/manifests/install.yaml
@@ -12843,6 +12843,9 @@ spec:
                               - name
                               type: object
                             type: array
+                          maxTrafficWeight:
+                            format: int32
+                            type: integer
                           nginx:
                             properties:
                               additionalIngressAnnotations:

diff --git a/pkg/apiclient/rollout/rollout.swagger.json b/pkg/apiclient/rollout/rollout.swagger.json
@@ -2303,6 +2303,11 @@
             "format": "byte"
           },
           "title": "+kubebuilder:validation:Schemaless\n+kubebuilder:pruning:PreserveUnknownFields\n+kubebuilder:validation:Type=object\nPlugins holds specific configuration that traffic router plugins can use for routing traffic"
+        },
+        "maxTrafficWeight": {
+          "type": "integer",
+          "format": "int32",
+          "title": "MaxTrafficWeight The total weight of traffic. If unspecified, it defaults to 100"
         }
       },
       "title": "RolloutTrafficRouting hosts all the different configuration for supported service meshes to enable more fine-grained traffic routing"

diff --git a/pkg/apis/rollouts/v1alpha1/generated.pb.go b/pkg/apis/rollouts/v1alpha1/generated.pb.go
diff --git a/pkg/apis/rollouts/v1alpha1/generated.proto b/pkg/apis/rollouts/v1alpha1/generated.proto
diff --git a/pkg/apis/rollouts/v1alpha1/openapi_generated.go b/pkg/apis/rollouts/v1alpha1/openapi_generated.go
diff --git a/pkg/apis/rollouts/v1alpha1/types.go b/pkg/apis/rollouts/v1alpha1/types.go
@@ -379,11 +379,15 @@ type RolloutTrafficRouting struct {
 	ManagedRoutes []MangedRoutes `json:"managedRoutes,omitempty" protobuf:"bytes,8,rep,name=managedRoutes"`
 	// Apisix holds specific configuration to use Apisix to route traffic
 	Apisix *ApisixTrafficRouting `json:"apisix,omitempty" protobuf:"bytes,9,opt,name=apisix"`
+
 	// +kubebuilder:validation:Schemaless
 	// +kubebuilder:pruning:PreserveUnknownFields
 	// +kubebuilder:validation:Type=object
 	// Plugins holds specific configuration that traffic router plugins can use for routing traffic
 	Plugins map[string]json.RawMessage `json:"plugins,omitempty" protobuf:"bytes,10,opt,name=plugins"`
+
+	// MaxTrafficWeight The total weight of traffic. If unspecified, it defaults to 100
+	MaxTrafficWeight *int32 `json:"maxTrafficWeight,omitempty" protobuf:"varint,11,opt,name=maxTrafficWeight"`
 }
 
 type MangedRoutes struct {

diff --git a/pkg/apis/rollouts/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/rollouts/v1alpha1/zz_generated.deepcopy.go
diff --git a/pkg/apis/rollouts/validation/validation.go b/pkg/apis/rollouts/validation/validation.go
@@ -20,15 +20,16 @@ import (
 
 	"github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
 	"github.com/argoproj/argo-rollouts/utils/defaults"
+	"github.com/argoproj/argo-rollouts/utils/weightutil"
 )
 
 const (
 	// Validate Spec constants
 
 	// MissingFieldMessage the message to indicate rollout is missing a field
 	MissingFieldMessage = "Rollout has missing field '%s'"
-	// InvalidSetWeightMessage indicates the setweight value needs to be between 0 and 100
-	InvalidSetWeightMessage = "SetWeight needs to be between 0 and 100"
+	// InvalidSetWeightMessage indicates the setweight value needs to be between 0 and max weight
+	InvalidSetWeightMessage = "SetWeight needs to be between 0 and %d"
 	// InvalidCanaryExperimentTemplateWeightWithoutTrafficRouting indicates experiment weight cannot be set without trafficRouting
 	InvalidCanaryExperimentTemplateWeightWithoutTrafficRouting = "Experiment template weight cannot be set unless TrafficRouting is enabled"
 	// InvalidSetCanaryScaleTrafficPolicy indicates that TrafficRouting, required for SetCanaryScale, is missing
@@ -72,6 +73,8 @@ const (
 	InvalidCanaryDynamicStableScale = "Canary dynamicStableScale can only be used with traffic routing"
 	// InvalidCanaryDynamicStableScaleWithScaleDownDelay indicates that canary.dynamicStableScale cannot be used with scaleDownDelaySeconds
 	InvalidCanaryDynamicStableScaleWithScaleDownDelay = "Canary dynamicStableScale cannot be used with scaleDownDelaySeconds"
+	// InvalidCanaryMaxWeightOnlySupportInNginxAndPlugins indicates that canary.maxTrafficWeight cannot be used
+	InvalidCanaryMaxWeightOnlySupportInNginxAndPlugins = "Canary maxTrafficWeight in traffic routing only supported in Nginx and Plugins"
 	// InvalidPingPongProvidedMessage indicates that both ping and pong service must be set to use Ping-Pong feature
 	InvalidPingPongProvidedMessage = "Ping service and Pong service must to be set to use Ping-Pong feature"
 	// DuplicatedPingPongServicesMessage indicates that the rollout uses the same service for the ping and pong services
@@ -295,6 +298,12 @@ func ValidateRolloutStrategyCanary(rollout *v1alpha1.Rollout, fldPath *field.Pat
 		if canary.ScaleDownDelaySeconds != nil && canary.DynamicStableScale {
 			allErrs = append(allErrs, field.Invalid(fldPath.Child("dynamicStableScale"), canary.DynamicStableScale, InvalidCanaryDynamicStableScaleWithScaleDownDelay))
 		}
+		// only the nginx and plugin have this support for now
+		if canary.TrafficRouting.MaxTrafficWeight != nil {
+			if canary.TrafficRouting.Nginx == nil && len(canary.TrafficRouting.Plugins) == 0 {
+				allErrs = append(allErrs, field.Invalid(fldPath.Child("trafficRouting").Child("maxTrafficWeight"), canary.TrafficRouting.MaxTrafficWeight, InvalidCanaryMaxWeightOnlySupportInNginxAndPlugins))
+			}
+		}
 	}
 
 	for i, step := range canary.Steps {
@@ -306,8 +315,11 @@ func ValidateRolloutStrategyCanary(rollout *v1alpha1.Rollout, fldPath *field.Pat
 				step.Experiment == nil, step.Pause == nil, step.SetWeight == nil, step.Analysis == nil, step.SetCanaryScale == nil, step.SetHeaderRoute == nil, step.SetMirrorRoute == nil)
 			allErrs = append(allErrs, field.Invalid(stepFldPath, errVal, InvalidStepMessage))
 		}
-		if step.SetWeight != nil && (*step.SetWeight < 0 || *step.SetWeight > 100) {
-			allErrs = append(allErrs, field.Invalid(stepFldPath.Child("setWeight"), *canary.Steps[i].SetWeight, InvalidSetWeightMessage))
+
+		maxTrafficWeight := weightutil.MaxTrafficWeight(rollout)
+
+		if step.SetWeight != nil && (*step.SetWeight < 0 || *step.SetWeight > maxTrafficWeight) {
+			allErrs = append(allErrs, field.Invalid(stepFldPath.Child("setWeight"), *canary.Steps[i].SetWeight, fmt.Sprintf(InvalidSetWeightMessage, maxTrafficWeight)))
 		}
 		if step.Pause != nil && step.Pause.DurationSeconds() < 0 {
 			allErrs = append(allErrs, field.Invalid(stepFldPath.Child("pause").Child("duration"), step.Pause.DurationSeconds(), InvalidDurationMessage))

diff --git a/pkg/apis/rollouts/validation/validation_test.go b/pkg/apis/rollouts/validation/validation_test.go
@@ -397,7 +397,59 @@ func TestValidateRolloutStrategyCanary(t *testing.T) {
 		invalidRo := ro.DeepCopy()
 		invalidRo.Spec.Strategy.Canary.Steps[0].SetWeight = &setWeight
 		allErrs := ValidateRolloutStrategyCanary(invalidRo, field.NewPath(""))
-		assert.Equal(t, InvalidSetWeightMessage, allErrs[0].Detail)
+		assert.Equal(t, fmt.Sprintf(InvalidSetWeightMessage, 100), allErrs[0].Detail)
+	})
+
+	t.Run("only nginx/plugins support max weight value", func(t *testing.T) {
+		anyWeight := int32(1)
+
+		type testCases struct {
+			trafficRouting *v1alpha1.RolloutTrafficRouting
+			expectError    bool
+			expectedError  string
+		}
+
+		testCasesList := []testCases{
+			{
+				trafficRouting: &v1alpha1.RolloutTrafficRouting{
+					ALB:              &v1alpha1.ALBTrafficRouting{RootService: "root-service"},
+					MaxTrafficWeight: &anyWeight,
+				},
+				expectError:   true,
+				expectedError: InvalidCanaryMaxWeightOnlySupportInNginxAndPlugins,
+			},
+			{
+				trafficRouting: &v1alpha1.RolloutTrafficRouting{
+					Nginx: &v1alpha1.NginxTrafficRouting{
+						StableIngress: "stable-ingress",
+					},
+					MaxTrafficWeight: &anyWeight,
+				},
+				expectError: false,
+			},
+			{
+				trafficRouting: &v1alpha1.RolloutTrafficRouting{
+					Plugins: map[string]json.RawMessage{
+						"anyplugin": []byte(`{"key": "value"}`),
+					},
+					MaxTrafficWeight: &anyWeight,
+				},
+				expectError: false,
+			},
+		}
+
+		for _, testCase := range testCasesList {
+			invalidRo := ro.DeepCopy()
+			invalidRo.Spec.Strategy.Canary.Steps[0].SetWeight = &anyWeight
+			invalidRo.Spec.Strategy.Canary.TrafficRouting = testCase.trafficRouting
+			allErrs := ValidateRolloutStrategyCanary(invalidRo, field.NewPath(""))
+			if !testCase.expectError {
+				assert.Empty(t, allErrs)
+				continue
+			}
+
+			assert.Equal(t, testCase.expectedError, allErrs[0].Detail)
+		}
 	})
 
 	t.Run("invalid duration set in paused step", func(t *testing.T) {

diff --git a/pkg/kubectl-argo-rollouts/info/rollout_info.go b/pkg/kubectl-argo-rollouts/info/rollout_info.go
@@ -14,6 +14,7 @@ import (
 	"github.com/argoproj/argo-rollouts/utils/defaults"
 	replicasetutil "github.com/argoproj/argo-rollouts/utils/replicaset"
 	rolloututil "github.com/argoproj/argo-rollouts/utils/rollout"
+	"github.com/argoproj/argo-rollouts/utils/weightutil"
 )
 
 func NewRolloutInfo(
@@ -58,12 +59,12 @@ func NewRolloutInfo(
 		currentStep, _ := replicasetutil.GetCurrentCanaryStep(ro)
 
 		if currentStep == nil {
-			roInfo.ActualWeight = "100"
+			roInfo.ActualWeight = fmt.Sprintf("%d", weightutil.MaxTrafficWeight(ro))
 		} else if ro.Status.AvailableReplicas > 0 {
 			if ro.Spec.Strategy.Canary.TrafficRouting == nil {
 				for _, rs := range roInfo.ReplicaSets {
 					if rs.Canary {
-						roInfo.ActualWeight = fmt.Sprintf("%d", (rs.Available*100)/ro.Status.AvailableReplicas)
+						roInfo.ActualWeight = fmt.Sprintf("%d", (rs.Available*weightutil.MaxTrafficWeight(ro))/ro.Status.AvailableReplicas)
 					}
 				}
 			} else {

diff --git a/rollout/canary_test.go b/rollout/canary_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/argoproj/argo-rollouts/utils/annotations"
 	"github.com/argoproj/argo-rollouts/utils/conditions"
 	"github.com/argoproj/argo-rollouts/utils/hash"
+	ingressutil "github.com/argoproj/argo-rollouts/utils/ingress"
 	logutil "github.com/argoproj/argo-rollouts/utils/log"
 	"github.com/argoproj/argo-rollouts/utils/record"
 	replicasetutil "github.com/argoproj/argo-rollouts/utils/replicaset"
@@ -550,6 +551,92 @@ func TestCanaryRolloutCreateFirstReplicasetWithSteps(t *testing.T) {
 	assert.JSONEq(t, calculatePatch(r, expectedPatch), patch)
 }
 
+func TestCanaryRolloutWithMaxWeightInTrafficRouting(t *testing.T) {
+	testCases := []struct {
+		name                    string
+		maxWeight               *int32
+		setWeight               int32
+		expectedCreatedReplicas int32
+		expectedUpdatedReplicas int32
+	}{
+		{
+			name:                    "max weight 100",
+			maxWeight:               int32Ptr(100),
+			setWeight:               10,
+			expectedCreatedReplicas: 0,
+			expectedUpdatedReplicas: 1,
+		},
+		{
+			name:                    "max weight 1000",
+			maxWeight:               int32Ptr(1000),
+			setWeight:               200,
+			expectedCreatedReplicas: 0,
+			expectedUpdatedReplicas: 2,
+		},
+	}
+
+	for _, tc := range testCases {
+		f := newFixture(t)
+		defer f.Close()
+		steps := []v1alpha1.CanaryStep{{
+			SetWeight: int32Ptr(tc.setWeight),
+		}}
+		r1 := newCanaryRollout("foo", 10, nil, steps, int32Ptr(0), intstr.FromInt(1), intstr.FromInt(0))
+
+		canarySVCName := "canary"
+		stableSVCName := "stable"
+
+		ingressName := "ingress"
+		r1.Spec.Strategy.Canary.TrafficRouting = &v1alpha1.RolloutTrafficRouting{
+			MaxTrafficWeight: tc.maxWeight,
+			Nginx: &v1alpha1.NginxTrafficRouting{
+				StableIngress: ingressName,
+			},
+		}
+		r1.Spec.Strategy.Canary.StableService = stableSVCName
+		r1.Spec.Strategy.Canary.CanaryService = canarySVCName
+		r1.Status.StableRS = "895c6c4f9"
+		r2 := bumpVersion(r1)
+
+		f.rolloutLister = append(f.rolloutLister, r2)
+		f.objects = append(f.objects, r2)
+
+		rs1 := newReplicaSetWithStatus(r1, 10, 10)
+		rs2 := newReplicaSetWithStatus(r2, 1, 0)
+
+		stableSvc := newService(stableSVCName, 80,
+			map[string]string{v1alpha1.DefaultRolloutUniqueLabelKey: rs1.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]}, r1)
+
+		canarySvc := newService(canarySVCName, 80,
+			map[string]string{v1alpha1.DefaultRolloutUniqueLabelKey: rs2.Labels[v1alpha1.DefaultRolloutUniqueLabelKey]}, r1)
+		f.replicaSetLister = append(f.replicaSetLister, rs1)
+
+		ing := newIngress(ingressName, canarySvc, stableSvc)
+		ing.Spec.Rules[0].HTTP.Paths[0].Backend.ServiceName = stableSVCName
+		f.kubeobjects = append(f.kubeobjects, rs1, canarySvc, stableSvc, ing)
+		f.serviceLister = append(f.serviceLister, canarySvc, stableSvc)
+		f.ingressLister = append(f.ingressLister, ingressutil.NewLegacyIngress(ing))
+
+		createdRSIndex := f.expectCreateReplicaSetAction(rs2)
+		updatedRSIndex := f.expectUpdateReplicaSetAction(rs2)
+		updatedRolloutIndex := f.expectUpdateRolloutStatusAction(r2)
+		f.expectPatchRolloutAction(r2)
+		f.run(getKey(r2, t))
+
+		createdRS := f.getCreatedReplicaSet(createdRSIndex)
+		assert.Equal(t, tc.expectedCreatedReplicas, *createdRS.Spec.Replicas)
+		updatedRS := f.getUpdatedReplicaSet(updatedRSIndex)
+		assert.Equal(t, tc.expectedUpdatedReplicas, *updatedRS.Spec.Replicas)
+
+		updatedRollout := f.getUpdatedRollout(updatedRolloutIndex)
+		progressingCondition := conditions.GetRolloutCondition(updatedRollout.Status, v1alpha1.RolloutProgressing)
+		assert.NotNil(t, progressingCondition)
+		assert.Equal(t, conditions.NewReplicaSetReason, progressingCondition.Reason)
+		assert.Equal(t, corev1.ConditionTrue, progressingCondition.Status)
+		assert.Equal(t, fmt.Sprintf(conditions.NewReplicaSetMessage, createdRS.Name), progressingCondition.Message)
+	}
+
+}
 func TestCanaryRolloutCreateNewReplicaWithCorrectWeight(t *testing.T) {
 	f := newFixture(t)
 	defer f.Close()

diff --git a/rollout/trafficrouting.go b/rollout/trafficrouting.go
@@ -26,6 +26,7 @@ import (
 	"github.com/argoproj/argo-rollouts/utils/record"
 	replicasetutil "github.com/argoproj/argo-rollouts/utils/replicaset"
 	rolloututil "github.com/argoproj/argo-rollouts/utils/rollout"
+	"github.com/argoproj/argo-rollouts/utils/weightutil"
 )
 
 // NewTrafficRoutingReconciler identifies return the TrafficRouting Plugin that the rollout wants to modify
@@ -132,6 +133,7 @@ func (c *Controller) NewTrafficRoutingReconciler(roCtx *rolloutContext) ([]traff
 	return nil, nil
 }
 
+// this currently only be used in the canary strategy
 func (c *rolloutContext) reconcileTrafficRouting() error {
 	reconcilers, err := c.newTrafficRoutingReconciler(c)
 	// a return here does ensure that all trafficReconcilers are healthy
@@ -201,7 +203,7 @@ func (c *rolloutContext) reconcileTrafficRouting() error {
 			// But we can only increase canary weight according to available replica counts of the canary.
 			// we will need to set the desiredWeight to 0 when the newRS is not available.
 			if c.rollout.Spec.Strategy.Canary.DynamicStableScale {
-				desiredWeight = (100 * c.newRS.Status.AvailableReplicas) / *c.rollout.Spec.Replicas
+				desiredWeight = (weightutil.MaxTrafficWeight(c.rollout) * c.newRS.Status.AvailableReplicas) / *c.rollout.Spec.Replicas
 			} else if c.rollout.Status.Canary.Weights != nil {
 				desiredWeight = c.rollout.Status.Canary.Weights.Canary.Weight
 			}
@@ -229,7 +231,7 @@ func (c *rolloutContext) reconcileTrafficRouting() error {
 				desiredWeight = replicasetutil.GetCurrentSetWeight(c.rollout)
 				weightDestinations = append(weightDestinations, c.calculateWeightDestinationsFromExperiment()...)
 			} else {
-				desiredWeight = 100
+				desiredWeight = weightutil.MaxTrafficWeight(c.rollout)
 			}
 		}
 		// We need to check for revision > 1 because when we first install the rollout we run step 0 this prevents that.
@@ -303,7 +305,7 @@ func (c *rolloutContext) calculateDesiredWeightOnAbortOrStableRollback() int32 {
 	}
 	// When using dynamic stable scaling, we must dynamically decreasing the weight to the canary
 	// according to the availability of the stable (whatever it can support).
-	desiredWeight := 100 - ((100 * c.stableRS.Status.AvailableReplicas) / *c.rollout.Spec.Replicas)
+	desiredWeight := weightutil.MaxTrafficWeight(c.rollout) - ((weightutil.MaxTrafficWeight(c.rollout) * c.stableRS.Status.AvailableReplicas) / *c.rollout.Spec.Replicas)
 	if c.rollout.Status.Canary.Weights != nil {
 		// This ensures that if we are already at a lower weight, then we will not
 		// increase the weight because stable availability is flapping (e.g. pod restarts)
@@ -336,7 +338,7 @@ func calculateWeightStatus(ro *v1alpha1.Rollout, canaryHash, stableHash string,
 			ServiceName:     ro.Spec.Strategy.Canary.CanaryService,
 		},
 	}
-	stableWeight := 100 - desiredWeight
+	stableWeight := weightutil.MaxTrafficWeight(ro) - desiredWeight
 	for _, weightDest := range weightDestinations {
 		weights.Additional = append(weights.Additional, weightDest)
 		stableWeight -= weightDest.Weight