Wait for at least last step duration

api/v1alpha1/temporalworker_webhook.go

@@ -7,15 +7,15 @@ package v1alpha1
 import (
 	"context"
 	"fmt"
+	"time"
+
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
-	"time"
 )
 
 const (
@@ -73,21 +73,59 @@ func (r *TemporalWorkerDeployment) validateForUpdateOrCreate(ctx context.Context
 		return nil, apierrors.NewBadRequest("expected a TemporalWorkerDeployment")
 	}
 
+	return validateForUpdateOrCreate(nil, dep)
+}
+
+func validateForUpdateOrCreate(old, new *TemporalWorkerDeployment) (admission.Warnings, error) {
 	var allErrs field.ErrorList
 
-	if len(dep.Name) > maxTemporalWorkerDeploymentNameLen {
+	if len(new.GetName()) > maxTemporalWorkerDeploymentNameLen {
 		allErrs = append(allErrs,
-			field.Invalid(field.NewPath("name"), dep.Name, fmt.Sprintf("cannot be more than %d characters", maxTemporalWorkerDeploymentNameLen)),
+			field.Invalid(field.NewPath("metadata.name"), new.GetName(), fmt.Sprintf("cannot be more than %d characters", maxTemporalWorkerDeploymentNameLen)),
 		)
 	}
 
+	allErrs = append(allErrs, validateRolloutStrategy(new.Spec.RolloutStrategy)...)
+
 	if len(allErrs) > 0 {
-		return nil, apierrors.NewInvalid(
-			schema.GroupKind{Group: "your.group", Kind: "TemporalWorkerDeployment"},
-			dep.Name,
-			allErrs,
-		)
+		return nil, newInvalidErr(new, allErrs)
 	}
 
 	return nil, nil
 }
+
+func validateRolloutStrategy(s RolloutStrategy) []*field.Error {
+	var allErrs []*field.Error
+
+	if s.Strategy == UpdateProgressive {
+		rolloutSteps := s.Steps
+		if len(rolloutSteps) == 0 {
+			allErrs = append(allErrs,
+				field.Invalid(field.NewPath("spec.cutover.steps"), rolloutSteps, "steps are required for Progressive cutover"),
+			)
+		}
+		var lastRamp float32
+		for i, s := range rolloutSteps {
+			// Check duration >= 30s
+			if s.PauseDuration.Duration < 30*time.Second {
+				allErrs = append(allErrs,
+					field.Invalid(field.NewPath(fmt.Sprintf("spec.cutover.steps[%d].pauseDuration", i)), s.PauseDuration.Duration.String(), "pause duration must be at least 30s"),
+				)
+			}
+
+			// Check ramp value greater than last
+			if s.RampPercentage <= lastRamp {
+				allErrs = append(allErrs,
+					field.Invalid(field.NewPath(fmt.Sprintf("spec.cutover.steps[%d].rampPercentage", i)), s.RampPercentage, "rampPercentage must increase between each step"),
+				)
+			}
+			lastRamp = s.RampPercentage
+		}
+	}
+
+	return allErrs
+}
+
+func newInvalidErr(dep *TemporalWorkerDeployment, errs field.ErrorList) *apierrors.StatusError {
+	return apierrors.NewInvalid(dep.GroupVersionKind().GroupKind(), dep.GetName(), errs)
+}

api/v1alpha1/temporalworker_webhook_test.go

@@ -2,102 +2,132 @@
 //
 // This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2024 Datadog, Inc.
 
-package v1alpha1
+package v1alpha1_test
 
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+
+	temporaliov1alpha1 "github.com/temporalio/temporal-worker-controller/api/v1alpha1"
+	"github.com/temporalio/temporal-worker-controller/internal/testhelpers"
 )
 
 func TestTemporalWorkerDeployment_ValidateCreate(t *testing.T) {
-	tests := []struct {
-		name        string
-		obj         runtime.Object
-		expectError bool
-		errorMsg    string
+	tests := map[string]struct {
+		obj      runtime.Object
+		errorMsg string
 	}{
-		{
-			name:        "valid temporal worker deployment",
-			obj:         MakeTWDWithName("valid-worker"),
-			expectError: false,
+		"valid temporal worker deployment": {
+			obj: testhelpers.MakeTWDWithName("valid-worker"),
 		},
-		{
-			name:        "temporal worker deployment with name too long",
-			obj:         MakeTWDWithName("this-is-a-very-long-temporal-worker-deployment-name-that-exceeds-the-maximum-allowed-length-of-sixty-three-characters"),
-			expectError: true,
-			errorMsg:    "cannot be more than 63 characters",
+		"temporal worker deployment with name too long": {
+			obj:      testhelpers.MakeTWDWithName("this-is-a-very-long-temporal-worker-deployment-name-that-exceeds-the-maximum-allowed-length-of-sixty-three-characters"),
+			errorMsg: "cannot be more than 63 characters",
 		},
-		{
-			name: "invalid object type",
+		"invalid object type": {
 			obj: &v1.Pod{
 				ObjectMeta: metav1.ObjectMeta{
 					Name: "test",
 				},
 			},
-			expectError: true,
-			errorMsg:    "expected a TemporalWorkerDeployment",
+			errorMsg: "expected a TemporalWorkerDeployment",
+		},
+		"missing rollout steps": {
+			obj: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("prog-rollout-missing-steps"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
+				obj.Spec.RolloutStrategy.Strategy = temporaliov1alpha1.UpdateProgressive
+				obj.Spec.RolloutStrategy.Steps = nil
+				return obj
+			}),
+			errorMsg: "spec.cutover.steps: Invalid value: []v1alpha1.RolloutStep(nil): steps are required for Progressive cutover",
+		},
+		"ramp value for step <= previous step": {
+			obj: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("prog-rollout-decreasing-ramps"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
+				obj.Spec.RolloutStrategy.Strategy = temporaliov1alpha1.UpdateProgressive
+				obj.Spec.RolloutStrategy.Steps = []temporaliov1alpha1.RolloutStep{
+					{5, metav1.Duration{Duration: time.Minute}},
+					{10, metav1.Duration{Duration: time.Minute}},
+					{9, metav1.Duration{Duration: time.Minute}},
+					{50, metav1.Duration{Duration: time.Minute}},
+					{50, metav1.Duration{Duration: time.Minute}},
+					{75, metav1.Duration{Duration: time.Minute}},
+				}
+				return obj
+			}),
+			errorMsg: "[spec.cutover.steps[2].rampPercentage: Invalid value: 9: rampPercentage must increase between each step, spec.cutover.steps[4].rampPercentage: Invalid value: 50: rampPercentage must increase between each step]",
+		},
+		"pause duration < 30s": {
+			obj: testhelpers.ModifyObj(testhelpers.MakeTWDWithName("prog-rollout-decreasing-ramps"), func(obj *temporaliov1alpha1.TemporalWorkerDeployment) *temporaliov1alpha1.TemporalWorkerDeployment {
+				obj.Spec.RolloutStrategy.Strategy = temporaliov1alpha1.UpdateProgressive
+				obj.Spec.RolloutStrategy.Steps = []temporaliov1alpha1.RolloutStep{
+					{10, metav1.Duration{Duration: time.Minute}},
+					{25, metav1.Duration{Duration: 10 * time.Second}},
+					{50, metav1.Duration{Duration: time.Minute}},
+				}
+				return obj
+			}),
+			errorMsg: `spec.cutover.steps[1].pauseDuration: Invalid value: "10s": pause duration must be at least 30s`,
 		},
 	}
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
 			ctx := context.Background()
-			webhook := &TemporalWorkerDeployment{}
-
-			warnings, err := webhook.ValidateCreate(ctx, tt.obj)
-
-			if tt.expectError {
-				require.Error(t, err)
-				assert.Contains(t, err.Error(), tt.errorMsg)
-			} else {
-				require.NoError(t, err)
+			webhook := &temporaliov1alpha1.TemporalWorkerDeployment{}
+
+			assertAdmission := func(warnings admission.Warnings, err error) {
+				if tc.errorMsg != "" {
+					require.Error(t, err)
+					assert.Contains(t, err.Error(), tc.errorMsg)
+				} else {
+					require.NoError(t, err)
+				}
+
+				// Warnings should always be nil for this implementation
+				assert.Nil(t, warnings)
 			}
 
-			// Warnings should always be nil for this implementation
-			assert.Nil(t, warnings)
+			// Verify that create and update enforce the same rules
+			assertAdmission(webhook.ValidateCreate(ctx, tc.obj))
+			assertAdmission(webhook.ValidateUpdate(ctx, nil, tc.obj))
 		})
 	}
 }
 
 func TestTemporalWorkerDeployment_ValidateUpdate(t *testing.T) {
-	tests := []struct {
-		name        string
-		oldObj      runtime.Object
-		newObj      runtime.Object
-		expectError bool
-		errorMsg    string
+	tests := map[string]struct {
+		oldObj   runtime.Object
+		newObj   runtime.Object
+		errorMsg string
 	}{
-		{
-			name:        "valid update",
-			oldObj:      nil,
-			newObj:      MakeTWDWithName("valid-worker"),
-			expectError: false,
+		"valid update": {
+			oldObj: nil,
+			newObj: testhelpers.MakeTWDWithName("valid-worker"),
 		},
-		{
-			name:        "update with name too long",
-			oldObj:      nil,
-			newObj:      MakeTWDWithName("this-is-a-very-long-temporal-worker-deployment-name-that-exceeds-the-maximum-allowed-length-of-sixty-three-characters"),
-			expectError: true,
-			errorMsg:    "cannot be more than 63 characters",
+		"update with name too long": {
+			oldObj:   nil,
+			newObj:   testhelpers.MakeTWDWithName("this-is-a-very-long-temporal-worker-deployment-name-that-exceeds-the-maximum-allowed-length-of-sixty-three-characters"),
+			errorMsg: "cannot be more than 63 characters",
 		},
 	}
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
 			ctx := context.Background()
-			webhook := &TemporalWorkerDeployment{}
+			webhook := &temporaliov1alpha1.TemporalWorkerDeployment{}
 
-			warnings, err := webhook.ValidateUpdate(ctx, tt.oldObj, tt.newObj)
+			warnings, err := webhook.ValidateUpdate(ctx, tc.oldObj, tc.newObj)
 
-			if tt.expectError {
+			if tc.errorMsg != "" {
 				require.Error(t, err)
-				assert.Contains(t, err.Error(), tt.errorMsg)
+				assert.Contains(t, err.Error(), tc.errorMsg)
 			} else {
 				require.NoError(t, err)
 			}
@@ -110,9 +140,9 @@ func TestTemporalWorkerDeployment_ValidateUpdate(t *testing.T) {
 
 func TestTemporalWorkerDeployment_ValidateDelete(t *testing.T) {
 	ctx := context.Background()
-	webhook := &TemporalWorkerDeployment{}
+	webhook := &temporaliov1alpha1.TemporalWorkerDeployment{}
 
-	obj := &TemporalWorkerDeployment{
+	obj := &temporaliov1alpha1.TemporalWorkerDeployment{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "worker",
 		},

api/v1alpha1/worker_types.go

@@ -215,6 +215,10 @@ type TargetWorkerDeploymentVersion struct {
 	// Only set when Status is VersionStatusRamping.
 	// +optional
 	RampingSince *metav1.Time `json:"rampingSince"`
+
+	// RampLastModifiedAt is the time when the ramp percentage was last changed for the target version.
+	// +optional
+	RampLastModifiedAt *metav1.Time `json:"rampLastModifiedAt,omitempty"`
 }
 
 // DeprecatedWorkerDeploymentVersion represents a worker deployment version that is no longer

helm/temporal-worker-controller/templates/crds/temporal.io_temporalworkerdeployments.yaml

@@ -3879,6 +3879,9 @@ spec:
                     type: string
                   managedBy:
                     type: string
+                  rampLastModifiedAt:
+                    format: date-time
+                    type: string
                   rampPercentage:
                     type: number
                   rampingSince:
@@ -3944,6 +3947,9 @@ spec:
                     type: string
                   managedBy:
                     type: string
+                  rampLastModifiedAt:
+                    format: date-time
+                    type: string
                   rampPercentage:
                     type: number
                   rampingSince:

internal/controller/state_mapper.go

@@ -41,6 +41,7 @@ func (m *stateMapper) mapToStatus(targetVersionID string) *v1alpha1.TemporalWork
 	status.TargetVersion = m.mapTargetWorkerDeploymentVersion(targetVersionID)
 	if status.TargetVersion != nil && m.temporalState.RampingVersionID == targetVersionID {
 		status.TargetVersion.RampingSince = m.temporalState.RampingSince
+		status.TargetVersion.RampLastModifiedAt = m.temporalState.RampLastModifiedAt
 		rampPercentage := m.temporalState.RampPercentage
 		status.TargetVersion.RampPercentage = &rampPercentage
 	}