disabled rollout (#155)

Signed-off-by: Kuromesi <blackfacepan@163.com>

api/v1alpha1/rollout_types.go

@@ -72,6 +72,10 @@ type RolloutSpec struct {
 	// RolloutID should be changed before each workload revision publication.
 	// It is to distinguish consecutive multiple workload publications and rollout progress.
 	DeprecatedRolloutID string `json:"rolloutID,omitempty"`
+	// if a rollout disabled, then the rollout would not watch changes of workload
+	//+kubebuilder:validation:Optional
+	//+kubebuilder:default=false
+	Disabled bool `json:"disabled"`
 }
 
 type ObjectRef struct {
@@ -257,6 +261,10 @@ const (
 	RolloutPhaseProgressing RolloutPhase = "Progressing"
 	// RolloutPhaseTerminating indicates a rollout is terminated
 	RolloutPhaseTerminating RolloutPhase = "Terminating"
+	// RolloutPhaseDisabled indicates a rollout is disabled
+	RolloutPhaseDisabled RolloutPhase = "Disabled"
+	// RolloutPhaseDisabling indicates a rollout is disabling and releasing resources
+	RolloutPhaseDisabling RolloutPhase = "Disabling"
 )
 
 // +genclient

config/crd/bases/rollouts.kruise.io_rollouts.yaml

@@ -56,6 +56,11 @@ spec:
           spec:
             description: RolloutSpec defines the desired state of Rollout
             properties:
+              disabled:
+                default: false
+                description: if a rollout disabled, then the rollout would not watch
+                  changes of workload
+                type: boolean
               objectRef:
                 description: 'INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
                   Important: Run "make" to regenerate code after modifying this file

pkg/controller/rollout/rollout_controller.go

@@ -130,6 +130,7 @@ func (r *RolloutReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 	if err != nil {
 		return ctrl.Result{}, err
 	}
+
 	// sync rollout status
 	retry, newStatus, err := r.calculateRolloutStatus(rollout)
 	if err != nil {
@@ -139,11 +140,14 @@ func (r *RolloutReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{RequeueAfter: time.Until(recheckTime)}, nil
 	}
 	var recheckTime *time.Time
+
 	switch rollout.Status.Phase {
 	case v1alpha1.RolloutPhaseProgressing:
 		recheckTime, err = r.reconcileRolloutProgressing(rollout, newStatus)
 	case v1alpha1.RolloutPhaseTerminating:
 		recheckTime, err = r.reconcileRolloutTerminating(rollout, newStatus)
+	case v1alpha1.RolloutPhaseDisabling:
+		recheckTime, err = r.reconcileRolloutDisabling(rollout, newStatus)
 	}
 	if err != nil {
 		return ctrl.Result{}, err

pkg/controller/rollout/rollout_status.go

@@ -49,6 +49,18 @@ func (r *RolloutReconciler) calculateRolloutStatus(rollout *v1alpha1.Rollout) (r
 		}
 		return false, newStatus, nil
 	}
+
+	if rollout.Spec.Disabled && newStatus.Phase != v1alpha1.RolloutPhaseDisabled && newStatus.Phase != v1alpha1.RolloutPhaseDisabling {
+		// if rollout in progressing, indicates a working rollout is disabled, then the rollout should be finalized
+		if newStatus.Phase == v1alpha1.RolloutPhaseProgressing {
+			newStatus.Phase = v1alpha1.RolloutPhaseDisabling
+			newStatus.Message = "Disabling rollout, release resources"
+		} else {
+			newStatus.Phase = v1alpha1.RolloutPhaseDisabled
+			newStatus.Message = "Rollout is disabled"
+		}
+	}
+
 	if newStatus.Phase == "" {
 		newStatus.Phase = v1alpha1.RolloutPhaseInitial
 	}
@@ -58,12 +70,14 @@ func (r *RolloutReconciler) calculateRolloutStatus(rollout *v1alpha1.Rollout) (r
 		klog.Errorf("rollout(%s/%s) get workload failed: %s", rollout.Namespace, rollout.Name, err.Error())
 		return false, nil, err
 	} else if workload == nil {
-		newStatus = &v1alpha1.RolloutStatus{
-			ObservedGeneration: rollout.Generation,
-			Phase:              v1alpha1.RolloutPhaseInitial,
-			Message:            "Workload Not Found",
+		if !rollout.Spec.Disabled {
+			newStatus = &v1alpha1.RolloutStatus{
+				ObservedGeneration: rollout.Generation,
+				Phase:              v1alpha1.RolloutPhaseInitial,
+				Message:            "Workload Not Found",
+			}
+			klog.Infof("rollout(%s/%s) workload not found, and reset status be Initial", rollout.Namespace, rollout.Name)
 		}
-		klog.Infof("rollout(%s/%s) workload not found, and reset status be Initial", rollout.Namespace, rollout.Name)
 		return false, newStatus, nil
 	}
 	klog.V(5).Infof("rollout(%s/%s) workload(%s)", rollout.Namespace, rollout.Name, util.DumpJSON(workload))
@@ -122,6 +136,11 @@ func (r *RolloutReconciler) calculateRolloutStatus(rollout *v1alpha1.Rollout) (r
 			}
 			newStatus.Message = "workload deployment is completed"
 		}
+	case v1alpha1.RolloutPhaseDisabled:
+		if !rollout.Spec.Disabled {
+			newStatus.Phase = v1alpha1.RolloutPhaseHealthy
+			newStatus.Message = "rollout is healthy"
+		}
 	}
 	return false, newStatus, nil
 }
@@ -207,6 +226,29 @@ func (r *RolloutReconciler) reconcileRolloutTerminating(rollout *v1alpha1.Rollou
 	return c.RecheckTime, nil
 }
 
+func (r *RolloutReconciler) reconcileRolloutDisabling(rollout *v1alpha1.Rollout, newStatus *v1alpha1.RolloutStatus) (*time.Time, error) {
+	workload, err := r.finder.GetWorkloadForRef(rollout)
+	if err != nil {
+		klog.Errorf("rollout(%s/%s) get workload failed: %s", rollout.Namespace, rollout.Name, err.Error())
+		return nil, err
+	}
+	c := &RolloutContext{Rollout: rollout, NewStatus: newStatus, Workload: workload}
+	done, err := r.doFinalising(c)
+	if err != nil {
+		return nil, err
+	} else if done {
+		klog.Infof("rollout(%s/%s) is disabled", rollout.Namespace, rollout.Name)
+		newStatus.Phase = v1alpha1.RolloutPhaseDisabled
+		newStatus.Message = "Rollout is disabled"
+	} else {
+		// Incomplete, recheck
+		expectedTime := time.Now().Add(time.Duration(defaultGracePeriodSeconds) * time.Second)
+		c.RecheckTime = &expectedTime
+		klog.Infof("rollout(%s/%s) disabling is incomplete, and recheck(%s)", rollout.Namespace, rollout.Name, expectedTime.String())
+	}
+	return c.RecheckTime, nil
+}
+
 // handle adding and handle finalizer logic, it turns if we should continue to reconcile
 func (r *RolloutReconciler) handleFinalizer(rollout *v1alpha1.Rollout) error {
 	// delete rollout crd, remove finalizer

pkg/controller/rollout/rollout_status_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package rollout
 
 import (
+	"context"
 	"testing"
 
 	"github.com/openkruise/rollouts/api/v1alpha1"
@@ -104,3 +105,70 @@ func TestCalculateRolloutHash(t *testing.T) {
 		})
 	}
 }
+
+func TestCalculateRolloutStatus(t *testing.T) {
+	cases := []struct {
+		name        string
+		getRollout  func() *v1alpha1.Rollout
+		expectPhase v1alpha1.RolloutPhase
+	}{
+		{
+			name: "apply an enabled rollout",
+			getRollout: func() *v1alpha1.Rollout {
+				obj := rolloutDemo.DeepCopy()
+				obj.Name = "Rollout-demo1"
+				obj.Status = v1alpha1.RolloutStatus{}
+				obj.Spec.Disabled = false
+				return obj
+			},
+			expectPhase: v1alpha1.RolloutPhaseInitial,
+		},
+		{
+			name: "disable an working rollout",
+			getRollout: func() *v1alpha1.Rollout {
+				obj := rolloutDemo.DeepCopy()
+				obj.Name = "Rollout-demo1"
+				obj.Status = v1alpha1.RolloutStatus{}
+				obj.Spec.Disabled = true
+				return obj
+			},
+			expectPhase: v1alpha1.RolloutPhaseDisabled,
+		},
+		{
+			name: "enable an disabled rollout",
+			getRollout: func() *v1alpha1.Rollout {
+				obj := rolloutDemo.DeepCopy()
+				obj.Name = "Rollout-demo2"
+				obj.Status = v1alpha1.RolloutStatus{}
+				obj.Spec.Disabled = false
+				return obj
+			},
+			expectPhase: v1alpha1.RolloutPhaseInitial,
+		},
+	}
+
+	t.Run("RolloutStatus test", func(t *testing.T) {
+		fc := fake.NewClientBuilder().WithScheme(scheme).Build()
+		r := &RolloutReconciler{
+			Client:                fc,
+			Scheme:                scheme,
+			Recorder:              record.NewFakeRecorder(10),
+			finder:                util.NewControllerFinder(fc),
+			trafficRoutingManager: trafficrouting.NewTrafficRoutingManager(fc),
+		}
+		r.canaryManager = &canaryReleaseManager{
+			Client:                fc,
+			trafficRoutingManager: r.trafficRoutingManager,
+			recorder:              r.Recorder,
+		}
+		for _, cs := range cases {
+			rollout := cs.getRollout()
+			fc.Create(context.TODO(), rollout)
+			_, newStatus, _ := r.calculateRolloutStatus(rollout)
+			r.updateRolloutStatusInternal(rollout, *newStatus)
+			if cs.expectPhase != newStatus.Phase {
+				t.Fatalf("expect phase %s, get %s, for rollout %s", cs.expectPhase, newStatus.Phase, rollout.Name)
+			}
+		}
+	})
+}

pkg/webhook/rollout/validating/rollout_create_update_handler_test.go

@@ -333,6 +333,7 @@ func TestRolloutValidateCreate(t *testing.T) {
 				object3 := rollout.DeepCopy()
 				object3.Name = "object-3"
 				object3.Spec.ObjectRef.WorkloadRef.Kind = "another"
+
 				return []client.Object{
 					object, object1, object2, object3,
 				}

pkg/webhook/workload/mutating/workload_update_handler.go

@@ -409,6 +409,10 @@ func (h *WorkloadHandler) fetchMatchedRollout(obj client.Object) (*appsv1alpha1.
 		if !rollout.DeletionTimestamp.IsZero() || rollout.Spec.ObjectRef.WorkloadRef == nil {
 			continue
 		}
+		if rollout.Status.Phase == appsv1alpha1.RolloutPhaseDisabled {
+			klog.Infof("Disabled rollout(%s/%s) fetched when fetching matched rollout", rollout.Namespace, rollout.Name)
+			continue
+		}
 		ref := rollout.Spec.ObjectRef.WorkloadRef
 		gv, err := schema.ParseGroupVersion(ref.APIVersion)
 		if err != nil {

test/e2e/rollout_test.go

@@ -5485,6 +5485,88 @@ var _ = SIGDescribe("Rollout", func() {
 		})
 
 	})
+
+	KruiseDescribe("Disabled rollout tests", func() {
+		rollout := &v1alpha1.Rollout{}
+		Expect(ReadYamlToObject("./test_data/rollout/rollout_disabled.yaml", rollout)).ToNot(HaveOccurred())
+		It("Rollout status tests", func() {
+			By("Create an enabled rollout")
+			rollout1 := rollout.DeepCopy()
+			rollout1.Name = "rollout-demo1"
+			rollout1.Spec.Disabled = false
+			CreateObject(rollout1)
+			time.Sleep(1 * time.Second)
+
+			By("Create another enabled rollout")
+			rollout2 := rollout.DeepCopy()
+			rollout2.Name = "rollout-demo2"
+			rollout2.Spec.Disabled = false
+			rollout2.SetNamespace(namespace)
+			Expect(k8sClient.Create(context.TODO(), rollout2)).Should(HaveOccurred())
+
+			By("Creating a disabled rollout")
+			rollout3 := rollout.DeepCopy()
+			rollout3.Name = "rollout-demo3"
+			rollout3.Spec.Disabled = true
+			rollout2.SetNamespace(namespace)
+			Expect(k8sClient.Create(context.TODO(), rollout2)).Should(HaveOccurred())
+			// wait for reconciling
+			time.Sleep(3 * time.Second)
+			Expect(GetObject(rollout1.Name, rollout1)).NotTo(HaveOccurred())
+			Expect(rollout1.Status.Phase).Should(Equal(v1alpha1.RolloutPhaseInitial))
+
+			By("Create workload")
+			deploy := &apps.Deployment{}
+			Expect(ReadYamlToObject("./test_data/rollout/deployment_disabled.yaml", deploy)).ToNot(HaveOccurred())
+			CreateObject(deploy)
+			WaitDeploymentAllPodsReady(deploy)
+			Expect(GetObject(rollout1.Name, rollout1)).NotTo(HaveOccurred())
+			Expect(rollout1.Status.Phase).Should(Equal(v1alpha1.RolloutPhaseHealthy))
+
+			By("Updating deployment version-1 to version-2")
+			Expect(GetObject(deploy.Name, deploy)).NotTo(HaveOccurred())
+			newEnvs := mergeEnvVar(deploy.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "VERSION", Value: "version-2"})
+			deploy.Spec.Template.Spec.Containers[0].Env = newEnvs
+			UpdateDeployment(deploy)
+			WaitRolloutCanaryStepPaused(rollout1.Name, 1)
+			Expect(GetObject(rollout1.Name, rollout1)).NotTo(HaveOccurred())
+			Expect(rollout1.Status.CanaryStatus.CanaryReplicas).Should(BeNumerically("==", 2))
+			Expect(rollout1.Status.CanaryStatus.CanaryReadyReplicas).Should(BeNumerically("==", 2))
+			Expect(GetObject(deploy.Name, deploy)).NotTo(HaveOccurred())
+			Expect(deploy.Spec.Paused).Should(BeTrue())
+
+			By("Disable a rolling rollout")
+			rollout1.Spec.Disabled = true
+			UpdateRollout(rollout1)
+			time.Sleep(5 * time.Second)
+
+			By("Rolling should be resumed")
+			Expect(GetObject(deploy.Name, deploy)).NotTo(HaveOccurred())
+			Expect(deploy.Spec.Paused).Should(BeFalse())
+
+			By("Batchrelease should be deleted")
+			key := types.NamespacedName{Namespace: namespace, Name: rollout1.Name}
+			Expect(k8sClient.Get(context.TODO(), key, &v1alpha1.BatchRelease{})).Should(HaveOccurred())
+			Expect(GetObject(rollout1.Name, rollout1)).NotTo(HaveOccurred())
+			Expect(rollout1.Status.Phase).Should(Equal(v1alpha1.RolloutPhaseDisabled))
+
+			By("Updating deployment version-2 to version-3")
+			Expect(GetObject(deploy.Name, deploy)).NotTo(HaveOccurred())
+			newEnvs = mergeEnvVar(deploy.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "VERSION", Value: "version-3"})
+			deploy.Spec.Template.Spec.Containers[0].Env = newEnvs
+			UpdateDeployment(deploy)
+			time.Sleep(3 * time.Second)
+			Expect(GetObject(deploy.Name, deploy)).NotTo(HaveOccurred())
+			Expect(deploy.Spec.Paused).Should(BeFalse())
+
+			By("Enable a disabled rollout")
+			rollout1.Spec.Disabled = false
+			UpdateRollout(rollout1)
+			time.Sleep(3 * time.Second)
+			Expect(GetObject(rollout1.Name, rollout1)).NotTo(HaveOccurred())
+			Expect(rollout1.Status.Phase).Should(Equal(v1alpha1.RolloutPhaseHealthy))
+		})
+	})
 })
 
 func mergeEnvVar(original []v1.EnvVar, add v1.EnvVar) []v1.EnvVar {

test/e2e/test_data/rollout/deployment_disabled.yaml

@@ -0,0 +1,23 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: workload-demo
+  namespace: default
+spec:
+  replicas: 10
+  selector:
+    matchLabels:
+      app: demo
+  template:
+    metadata:
+      labels:
+        app: demo
+    spec:
+      containers:
+      - name: busybox
+        image: busybox:latest
+        imagePullPolicy: IfNotPresent
+        command: ["/bin/sh", "-c", "sleep 100d"]
+        env:
+        - name: VERSION
+          value: "version-1"

test/e2e/test_data/rollout/rollout_disabled.yaml

@@ -0,0 +1,19 @@
+apiVersion: rollouts.kruise.io/v1alpha1
+kind: Rollout
+metadata:
+  name: rollouts-demo
+  namespace: default
+  annotations:
+    rollouts.kruise.io/rolling-style: partition
+spec:
+  disabled: false
+  objectRef:
+    workloadRef:
+      apiVersion: apps/v1
+      kind: Deployment
+      name: workload-demo
+  strategy:
+    canary:
+      steps:
+      - replicas: 2
+      - replicas: 50%