Support pause rollout

Author: jiangpengcheng

controllers/function_controller.go

@@ -99,6 +99,13 @@ func (r *FunctionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		function.Status.Conditions = make(map[v1alpha1.Component]v1alpha1.ResourceCondition)
 	}
 
+	isNewGeneration := r.checkIfFunctionGenerationsIsIncreased(function)
+
+	// skip reconcile if pauseRollout is set to true and the generation is not increased
+	if spec.IsPauseRollout(function) && !isNewGeneration {
+		return ctrl.Result{}, nil
+	}
+
 	err = r.ObserveFunctionStatefulSet(ctx, function)
 	if err != nil {
 		return reconcile.Result{}, err
@@ -130,8 +137,6 @@ func (r *FunctionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, err
 	}
 
-	isNewGeneration := r.checkIfFunctionGenerationsIsIncreased(function)
-
 	err = r.ApplyFunctionStatefulSet(ctx, function, isNewGeneration)
 	if err != nil {
 		return reconcile.Result{}, err

controllers/sink_controller.go

@@ -98,6 +98,13 @@ func (r *SinkReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		sink.Status.Conditions = make(map[v1alpha1.Component]v1alpha1.ResourceCondition)
 	}
 
+	isNewGeneration := r.checkIfSinkGenerationsIsIncreased(sink)
+
+	// skip reconcile if pauseRollout is set to true and the generation is not increased
+	if spec.IsPauseRollout(sink) && !isNewGeneration {
+		return ctrl.Result{}, nil
+	}
+
 	err = r.ObserveSinkStatefulSet(ctx, sink)
 	if err != nil {
 		return reconcile.Result{}, err
@@ -129,8 +136,6 @@ func (r *SinkReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.
 		return ctrl.Result{}, err
 	}
 
-	isNewGeneration := r.checkIfSinkGenerationsIsIncreased(sink)
-
 	err = r.ApplySinkStatefulSet(ctx, sink, isNewGeneration)
 	if err != nil {
 		return reconcile.Result{}, err

controllers/source_controller.go

@@ -98,6 +98,13 @@ func (r *SourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		source.Status.Conditions = make(map[v1alpha1.Component]v1alpha1.ResourceCondition)
 	}
 
+	isNewGeneration := r.checkIfSourceGenerationsIsIncreased(source)
+
+	// skip reconcile if pauseRollout is set to true and the generation is not increased
+	if spec.IsPauseRollout(source) && !isNewGeneration {
+		return ctrl.Result{}, nil
+	}
+
 	err = r.ObserveSourceStatefulSet(ctx, source)
 	if err != nil {
 		return reconcile.Result{}, err
@@ -129,8 +136,6 @@ func (r *SourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		return ctrl.Result{}, err
 	}
 
-	isNewGeneration := r.checkIfSourceGenerationsIsIncreased(source)
-
 	err = r.ApplySourceStatefulSet(ctx, source, isNewGeneration)
 	if err != nil {
 		return reconcile.Result{}, err

controllers/spec/common.go

@@ -99,6 +99,7 @@ const (
 	AnnotationPrometheusScrape = "prometheus.io/scrape"
 	AnnotationPrometheusPort   = "prometheus.io/port"
 	AnnotationManaged          = "compute.functionmesh.io/managed"
+	AnnotationPauseRollout     = "compute.functionmesh.io/pause-rollout"
 	AnnotationNeedCleanup      = "compute.functionmesh.io/need-cleanup"
 
 	// if labels contains below, we think it comes from function-mesh-worker-service
@@ -171,6 +172,11 @@ func IsManaged(object metav1.Object) bool {
 	return !exists || managed != "false"
 }
 
+func IsPauseRollout(object metav1.Object) bool {
+	pauseRollout, exists := object.GetAnnotations()[AnnotationPauseRollout]
+	return exists && pauseRollout == "true"
+}
+
 func NeedCleanup(object metav1.Object) bool {
 	// don't cleanup if it's managed by function-mesh-worker-service
 	_, exists := object.GetLabels()[LabelPulsarCluster]