feat: support pause and resume reconciliation of a cluster

controllers/apps/cluster_controller.go

@@ -129,6 +129,8 @@ func (r *ClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		AddTransformer(
 			// handle cluster deletion first
 			&clusterDeletionTransformer{},
+			// handle cluster pause and resume
+			&clusterPauseTransformer{Client: r.Client},
 			// check is recovering from halted cluster
 			&clusterHaltRecoveryTransformer{},
 			// update finalizer and cd&cv labels

controllers/apps/component_controller.go

@@ -150,6 +150,8 @@ func (r *ComponentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		AddTransformer(
 			// handle component deletion and pre-terminate
 			&componentDeletionTransformer{},
+			// handle component pause and resume
+			&componentPauseTransformer{},
 			// handle finalizers and referenced definition labels
 			&componentMetaTransformer{},
 			// validate referenced componentDefinition objects, and build synthesized component

controllers/apps/configuration/configuration_controller.go

@@ -36,6 +36,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	"github.com/apecloud/kubeblocks/controllers/extensions"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/component"
 	configctrl "github.com/apecloud/kubeblocks/pkg/controller/configuration"
@@ -113,6 +114,13 @@ func (r *ConfigurationReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		reqCtx.Log.Info("cluster is deleting, skip reconcile")
 		return intctrlutil.Reconciled()
 	}
+	// if cluster is paused, pausing reconfigure as well
+	if fetcherTask.ClusterObj != nil {
+		if val, ok := fetcherTask.ClusterObj.Annotations[extensions.ControllerPaused]; ok && val == "true" {
+			reqCtx.Log.Info(fmt.Sprintf("cluster is paused, skip reconcile"))
+			return intctrlutil.Reconciled()
+		}
+	}
 	if fetcherTask.ClusterComObj == nil || fetcherTask.ComponentObj == nil {
 		return r.failWithInvalidComponent(config, reqCtx)
 	}

controllers/apps/configuration/reconfigure_controller.go

@@ -39,6 +39,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	"github.com/apecloud/kubeblocks/controllers/extensions"
 	cfgcm "github.com/apecloud/kubeblocks/pkg/configuration/config_manager"
 	"github.com/apecloud/kubeblocks/pkg/configuration/core"
 	"github.com/apecloud/kubeblocks/pkg/constant"
@@ -200,6 +201,14 @@ func (r *ReconfigureReconciler) sync(reqCtx intctrlutil.RequestCtx, configMap *c
 		return intctrlutil.RequeueWithErrorAndRecordEvent(configMap, r.Recorder, err, reqCtx.Log)
 	}
 
+	// if cluster is paused, pausing reconfigure as well
+	if reconcileContext.ClusterObj != nil {
+		if val, ok := reconcileContext.ClusterObj.Annotations[extensions.ControllerPaused]; ok && val == "true" {
+			reqCtx.Log.Info(fmt.Sprintf("reconfigure is paused beacuse cluster %s is paused", resources.clusterName))
+			return intctrlutil.Reconciled()
+		}
+	}
+
 	// Assumption: It is required that the cluster must have a component.
 	if reconcileContext.ClusterComObj == nil {
 		reqCtx.Log.Info("not found component.")

controllers/apps/transformer_cluster_pause.go

@@ -0,0 +1,142 @@
+/*
+Copyright (C) 2022-2024 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package apps
+
+import (
+	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	"github.com/apecloud/kubeblocks/controllers/extensions"
+	"github.com/apecloud/kubeblocks/pkg/constant"
+	"github.com/apecloud/kubeblocks/pkg/controller/graph"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// clusterPauseTransformer handles cluster pause and resume
+type clusterPauseTransformer struct {
+	client.Client
+}
+
+var _ graph.Transformer = &clusterPauseTransformer{}
+
+func (t *clusterPauseTransformer) Transform(ctx graph.TransformContext, dag *graph.DAG) error {
+	transCtx, _ := ctx.(*clusterTransformContext)
+	cluster := transCtx.OrigCluster
+	graphCli, _ := transCtx.Client.(model.GraphClient)
+	if checkPaused(cluster) {
+		// set paused for all components
+		compList := &appsv1alpha1.ComponentList{}
+		labels := constant.GetClusterWellKnownLabels(cluster.Name)
+		if err := transCtx.Client.List(transCtx.Context, compList, client.InNamespace(cluster.Namespace), client.MatchingLabels(labels)); err != nil {
+			return err
+		}
+		notPaused := false
+		for _, comp := range compList.Items {
+			newComp := comp.DeepCopy()
+			if !checkPaused(newComp) {
+				annotations := comp.GetAnnotations()
+				if annotations == nil {
+					annotations = make(map[string]string)
+				}
+				annotations[extensions.ControllerPaused] = "true"
+				newComp.SetAnnotations(annotations)
+				graphCli.Update(dag, comp.DeepCopy(), newComp)
+				notPaused = true
+			}
+
+		}
+		if notPaused {
+			transCtx.EventRecorder.Eventf(cluster, corev1.EventTypeNormal, "Paused",
+				"cluster is paused")
+		}
+		return graph.ErrPrematureStop
+
+	} else {
+		// set resumed for all components
+		compList := &appsv1alpha1.ComponentList{}
+		labels := constant.GetClusterWellKnownLabels(cluster.Name)
+		if err := transCtx.Client.List(transCtx.Context, compList, client.InNamespace(cluster.Namespace), client.MatchingLabels(labels)); err != nil {
+			return err
+		}
+		hasPaused := false
+
+		// remove resumed annotation of configmaps
+		var cmList, err = t.listConfigMaps(transCtx, cluster)
+		if err != nil {
+			return err
+		}
+		for _, cm := range cmList.Items {
+			if val, ok := cm.GetAnnotations()[extensions.ControllerResumed]; ok && val == "true" {
+				newCm := cm.DeepCopy()
+				delete(newCm.Annotations, extensions.ControllerResumed)
+				graphCli.Update(dag, cm.DeepCopy(), newCm)
+			}
+		}
+
+		for _, comp := range compList.Items {
+			newComp := comp.DeepCopy()
+			if checkPaused(newComp) {
+				delete(newComp.Annotations, extensions.ControllerPaused)
+				hasPaused = true
+				graphCli.Update(dag, comp.DeepCopy(), newComp)
+			}
+		}
+
+		if hasPaused {
+			// add annotations to all configmaps for reconfiguring after resumed
+			var cmList, err = t.listConfigMaps(transCtx, cluster)
+			if err != nil {
+				return err
+			}
+			for _, cm := range cmList.Items {
+				newCm := cm.DeepCopy()
+				annotations := cm.GetAnnotations()
+				if annotations == nil {
+					annotations = make(map[string]string)
+				}
+				annotations[extensions.ControllerResumed] = "true"
+				newCm.SetAnnotations(annotations)
+				graphCli.Update(dag, cm.DeepCopy(), newCm)
+			}
+
+			// add event
+			transCtx.EventRecorder.Eventf(cluster, corev1.EventTypeNormal, "Resumed",
+				"cluster is resumed")
+		}
+		return nil
+	}
+}
+
+func (t *clusterPauseTransformer) listConfigMaps(transCtx *clusterTransformContext, cluster *appsv1alpha1.Cluster) (*corev1.ConfigMapList, error) {
+	cmList := &corev1.ConfigMapList{}
+	ml := map[string]string{
+		constant.AppManagedByLabelKey: constant.AppName,
+		constant.AppInstanceLabelKey:  cluster.Name,
+	}
+	listOpts := []client.ListOption{
+		client.InNamespace(cluster.Namespace),
+		client.MatchingLabels(ml),
+	}
+	err := t.Client.List(transCtx, cmList, listOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return cmList, nil
+}

controllers/apps/transformer_component_pause.go

@@ -0,0 +1,80 @@
+/*
+Copyright (C) 2022-2024 ApeCloud Co., Ltd
+
+This file is part of KubeBlocks project
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU Affero General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU Affero General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package apps
+
+import (
+	"fmt"
+	appsv1alpha1 "github.com/apecloud/kubeblocks/apis/apps/v1alpha1"
+	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
+	"github.com/apecloud/kubeblocks/pkg/controller/graph"
+	"github.com/apecloud/kubeblocks/pkg/controller/model"
+	"github.com/pkg/errors"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// componentDeletionTransformer handles component deletion
+type componentPauseTransformer struct {
+}
+
+var _ graph.Transformer = &componentDeletionTransformer{}
+
+func (t *componentPauseTransformer) Transform(ctx graph.TransformContext, dag *graph.DAG) error {
+	transCtx, _ := ctx.(*componentTransformContext)
+
+	graphCli, _ := transCtx.Client.(model.GraphClient)
+	comp := transCtx.Component
+	// if paused
+	if checkPaused(comp) {
+		// get instanceSet and set paused
+		oldInstanceSet, err := t.getInstanceSet(transCtx, comp)
+		newInstanceSet := oldInstanceSet.DeepCopy()
+		if err != nil {
+			return err
+		}
+		newInstanceSet.Spec.Paused = true
+		// update in dag
+		graphCli.Update(dag, oldInstanceSet, newInstanceSet)
+		return graph.ErrPrematureStop
+	} else {
+		// get instanceSet and cancel paused
+		oldInstanceSet, err := t.getInstanceSet(transCtx, comp)
+		if model.IsReconciliationPaused(oldInstanceSet) {
+			newInstanceSet := oldInstanceSet.DeepCopy()
+			if err != nil {
+				return err
+			}
+			newInstanceSet.Spec.Paused = false
+			// update in dag
+			graphCli.Update(dag, oldInstanceSet, newInstanceSet)
+			return nil
+		}
+		return nil
+	}
+}
+
+func (t *componentPauseTransformer) getInstanceSet(transCtx *componentTransformContext, comp *appsv1alpha1.Component) (*workloads.InstanceSet, error) {
+	instanceName := comp.Name
+	instanceSet := &workloads.InstanceSet{}
+	err := transCtx.Client.Get(transCtx.Context, types.NamespacedName{Name: instanceName, Namespace: comp.Namespace}, instanceSet)
+	if err != nil {
+		return nil, errors.New(fmt.Sprintf("failed to get instanceSet %s: %v", instanceName, err))
+	}
+	return instanceSet, nil
+}

controllers/apps/utils.go

@@ -26,6 +26,7 @@ import (
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/apecloud/kubeblocks/controllers/extensions"
 	"github.com/apecloud/kubeblocks/pkg/constant"
 	"github.com/apecloud/kubeblocks/pkg/controller/model"
 	"github.com/apecloud/kubeblocks/pkg/controller/multicluster"
@@ -84,3 +85,14 @@ func clientOption(v *model.ObjectVertex) *multicluster.ClientOption {
 	}
 	return multicluster.InControlContext()
 }
+
+func checkPaused(object client.Object) bool {
+	annotations := object.GetAnnotations()
+	if annotations == nil {
+		return false
+	}
+	if val, ok := annotations[extensions.ControllerPaused]; ok && val == "true" {
+		return true
+	}
+	return false
+}

controllers/extensions/const.go

@@ -25,6 +25,7 @@ const (
 
 	// annotation keys
 	ControllerPaused     = "controller.kubeblocks.io/controller-paused"
+	ControllerResumed    = "controller.kubeblocks.io/controller-resumed"
 	SkipInstallableCheck = "extensions.kubeblocks.io/skip-installable-check"
 	NoDeleteJobs         = "extensions.kubeblocks.io/no-delete-jobs"
 	AddonDefaultIsEmpty  = "addons.extensions.kubeblocks.io/default-is-empty"

pkg/controller/instanceset/reconciler_deletion.go

@@ -34,9 +34,7 @@ func (r *deletionReconciler) PreCondition(tree *kubebuilderx.ObjectTree) *kubebu
 	if tree.GetRoot() == nil || !model.IsObjectDeleting(tree.GetRoot()) {
 		return kubebuilderx.ResultUnsatisfied
 	}
-	if model.IsReconciliationPaused(tree.GetRoot()) {
-		return kubebuilderx.ResultUnsatisfied
-	}
+	// deletion need to be handled while paused
 	return kubebuilderx.ResultSatisfied
 }