tidyup

Author: leon-ape

controllers/parameters/config_annotation.go

@@ -123,7 +123,7 @@ func updateConfigPhaseWithResult(cli client.Client, ctx intctrlutil.RequestCtx,
 		config.ObjectMeta.Annotations[core.GenerateRevisionPhaseKey(revision)] = string(b)
 	}
 
-	if err := cli.Patch(ctx.Ctx, config, patch, inDataContextUnspecified()); err != nil {
+	if err := cli.Patch(ctx.Ctx, config, patch); err != nil {
 		return intctrlutil.RequeueWithError(err, ctx.Log, "")
 	}
 	if result.Retry {
@@ -184,7 +184,7 @@ func updateAppliedConfigs(cli client.Client, ctx intctrlutil.RequestCtx, config
 
 	// delete reconfigure-policy
 	delete(config.ObjectMeta.Annotations, constant.UpgradePolicyAnnotationKey)
-	if err := cli.Patch(ctx.Ctx, config, patch, inDataContextUnspecified()); err != nil {
+	if err := cli.Patch(ctx.Ctx, config, patch); err != nil {
 		return false, err
 	}
 

controllers/parameters/reconfigure_controller.go

@@ -94,8 +94,7 @@ func (r *ReconfigureReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 
 	config := &corev1.ConfigMap{}
-	// TODO(leon): data or universal?
-	if err := r.Client.Get(reqCtx.Ctx, reqCtx.Req.NamespacedName, config, inDataContextUnspecified()); err != nil {
+	if err := r.Client.Get(reqCtx.Ctx, reqCtx.Req.NamespacedName, config); err != nil {
 		return intctrlutil.CheckedRequeueWithError(err, reqCtx.Log, "")
 	}
 	if model.IsObjectDeleting(config) {
@@ -232,12 +231,12 @@ func (r *ReconfigureReconciler) sync(reqCtx intctrlutil.RequestCtx, configMap *c
 	return r.performUpgrade(rctx, tasks)
 }
 
-func (r *ReconfigureReconciler) genReconfigureActionTasks(templateSpec *appsv1.ComponentFileTemplate, rctx *ReconcileContext, patch *core.ConfigPatchInfo, restart bool) ([]ReloadAction, error) {
-	var tasks []ReloadAction
+func (r *ReconfigureReconciler) genReconfigureActionTasks(templateSpec *appsv1.ComponentFileTemplate, rctx *ReconcileContext, patch *core.ConfigPatchInfo, restart bool) ([]reconfigureTask, error) {
+	var tasks []reconfigureTask
 
 	// If the patch or ConfigRender is nil, return a single restart task.
 	if patch == nil || rctx.ConfigRender == nil {
-		return []ReloadAction{r.buildRestartTask(templateSpec, rctx)}, nil
+		return []reconfigureTask{r.buildRestartTask(templateSpec, rctx)}, nil
 	}
 
 	// needReloadAction determines if a reload action is needed based on the ParametersDefinition and ReloadPolicy.
@@ -268,7 +267,7 @@ func (r *ReconfigureReconciler) genReconfigureActionTasks(templateSpec *appsv1.C
 
 	// If no tasks were added, return a single restart task.
 	if len(tasks) == 0 {
-		return []ReloadAction{r.buildRestartTask(templateSpec, rctx)}, nil
+		return []reconfigureTask{r.buildRestartTask(templateSpec, rctx)}, nil
 	}
 
 	return tasks, nil
@@ -279,7 +278,7 @@ func (r *ReconfigureReconciler) buildReloadTask(policy parametersv1alpha1.Reload
 	rctx *ReconcileContext,
 	pd *parametersv1alpha1.ParametersDefinition,
 	configDescription *parametersv1alpha1.ComponentConfigDescription,
-	patch *core.ConfigPatchInfo) ReloadAction {
+	patch *core.ConfigPatchInfo) reconfigureTask {
 	reCtx := reconfigureContext{
 		RequestCtx:               rctx.RequestCtx,
 		Client:                   rctx.Client,
@@ -294,13 +293,12 @@ func (r *ReconfigureReconciler) buildReloadTask(policy parametersv1alpha1.Reload
 		ReconfigureClientFactory: getClientFactory(),
 		Patch:                    patch,
 	}
-
-	return reconfigureTask{ReloadPolicy: policy, taskCtx: reCtx}
+	return reconfigureTask{policy: policy, taskCtx: reCtx}
 }
 
-func (r *ReconfigureReconciler) buildRestartTask(configTemplate *appsv1.ComponentFileTemplate, rctx *ReconcileContext) ReloadAction {
+func (r *ReconfigureReconciler) buildRestartTask(configTemplate *appsv1.ComponentFileTemplate, rctx *ReconcileContext) reconfigureTask {
 	return reconfigureTask{
-		ReloadPolicy: parametersv1alpha1.RestartPolicy,
+		policy: parametersv1alpha1.RestartPolicy,
 		taskCtx: reconfigureContext{
 			RequestCtx:           rctx.RequestCtx,
 			Client:               rctx.Client,
@@ -350,19 +348,20 @@ func (r *ReconfigureReconciler) updateConfigCMStatus(reqCtx intctrlutil.RequestC
 	return intctrlutil.Reconciled()
 }
 
-func (r *ReconfigureReconciler) performUpgrade(rctx *ReconcileContext, reloadTasks []ReloadAction) (ctrl.Result, error) {
-	var err error
-	var returnedStatus returnedStatus
-	var reloadType string
-
-	for _, task := range reloadTasks {
-		reloadType = task.ReloadType()
-		returnedStatus, err = task.ExecReload()
-		if err != nil || returnedStatus.Status != ESNone {
-			return r.status(rctx, returnedStatus, reloadType, err)
+func (r *ReconfigureReconciler) performUpgrade(rctx *ReconcileContext, tasks []reconfigureTask) (ctrl.Result, error) {
+	var (
+		err    error
+		policy string
+		status returnedStatus
+	)
+	for _, task := range tasks {
+		policy = string(task.policy)
+		status, err = task.reconfigure()
+		if err != nil || status.Status != ESNone {
+			return r.status(rctx, status, policy, err)
 		}
 	}
-	return r.succeed(rctx, reloadType, returnedStatus)
+	return r.succeed(rctx, policy, status)
 }
 
 func (r *ReconfigureReconciler) status(rctx *ReconcileContext, returnedStatus returnedStatus, policy string, err error) (ctrl.Result, error) {
@@ -384,13 +383,12 @@ func (r *ReconfigureReconciler) status(rctx *ReconcileContext, returnedStatus re
 	}
 }
 
-func (r *ReconfigureReconciler) succeed(rctx *ReconcileContext, reloadType string, returnedStatus returnedStatus) (ctrl.Result, error) {
+func (r *ReconfigureReconciler) succeed(rctx *ReconcileContext, policy string, status returnedStatus) (ctrl.Result, error) {
 	rctx.Recorder.Eventf(rctx.ConfigMap,
 		corev1.EventTypeNormal,
 		appsv1alpha1.ReasonReconfigureSucceed,
 		"the reconfigure[%s] has been processed successfully",
-		reloadType)
-
-	result := reconciled(returnedStatus, reloadType, parametersv1alpha1.CFinishedPhase)
-	return r.updateConfigCMStatus(rctx.RequestCtx, rctx.ConfigMap, reloadType, &result)
+		policy)
+	result := reconciled(status, policy, parametersv1alpha1.CFinishedPhase)
+	return r.updateConfigCMStatus(rctx.RequestCtx, rctx.ConfigMap, policy, &result)
 }

controllers/parameters/reconfigure_policy.go

@@ -20,7 +20,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package parameters
 
 import (
-	"fmt"
 	"strings"
 
 	"google.golang.org/grpc"
@@ -172,15 +171,3 @@ func makeReturnedStatus(status ExecStatus, ops ...func(status *returnedStatus))
 	}
 	return ret
 }
-
-func (r reconfigureTask) ReloadType() string {
-	return string(r.ReloadPolicy)
-}
-
-func (r reconfigureTask) ExecReload() (returnedStatus, error) {
-	if executor, ok := upgradePolicyMap[r.ReloadPolicy]; ok {
-		return executor.Upgrade(r.taskCtx)
-	}
-
-	return returnedStatus{}, fmt.Errorf("not support reload action[%s]", r.ReloadPolicy)
-}

controllers/parameters/types.go

@@ -21,6 +21,7 @@ package parameters
 
 import (
 	"context"
+	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -54,12 +55,14 @@ func GetInstanceSetRollingUpgradeFuncs() RollingUpgradeFuncs {
 	}
 }
 
-type ReloadAction interface {
-	ExecReload() (returnedStatus, error)
-	ReloadType() string
-}
-
 type reconfigureTask struct {
-	parametersv1alpha1.ReloadPolicy
+	policy  parametersv1alpha1.ReloadPolicy
 	taskCtx reconfigureContext
 }
+
+func (r reconfigureTask) reconfigure() (returnedStatus, error) {
+	if executor, ok := upgradePolicyMap[r.policy]; ok {
+		return executor.Upgrade(r.taskCtx)
+	}
+	return returnedStatus{}, fmt.Errorf("not support reload action[%s]", r.policy)
+}

controllers/parameters/utils.go

@@ -33,18 +33,13 @@ import (
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
-	"github.com/apecloud/kubeblocks/pkg/controller/multicluster"
 	intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil"
 	"github.com/apecloud/kubeblocks/pkg/parameters"
 	"github.com/apecloud/kubeblocks/pkg/parameters/core"
 	cfgproto "github.com/apecloud/kubeblocks/pkg/parameters/proto"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
 
-func inDataContextUnspecified() *multicluster.ClientOption {
-	return multicluster.InDataContextUnspecified()
-}
-
 // GetComponentPods gets all pods of the component.
 func GetComponentPods(params reconfigureContext) ([]corev1.Pod, error) {
 	componentPods := make([]corev1.Pod, 0)
@@ -62,7 +57,7 @@ func GetComponentPods(params reconfigureContext) ([]corev1.Pod, error) {
 
 func getPodsForOnlineUpdate(params reconfigureContext) ([]corev1.Pod, error) {
 	if len(params.InstanceSetUnits) > 1 {
-		return nil, fmt.Errorf("component require only one InstanceSet, actual %d components", len(params.InstanceSetUnits))
+		return nil, fmt.Errorf("component require only one InstanceSet, actual %d", len(params.InstanceSetUnits))
 	}
 
 	if len(params.InstanceSetUnits) == 0 {
@@ -122,12 +117,12 @@ func resolveReloadServerGrpcURL(pod *corev1.Pod) (string, error) {
 	return generateGrpcURL(pod, podPort)
 }
 
-func generateGrpcURL(pod *corev1.Pod, portPort int) (string, error) {
+func generateGrpcURL(pod *corev1.Pod, podPort int) (string, error) {
 	ip, err := ipAddressFromPod(pod.Status)
 	if err != nil {
 		return "", err
 	}
-	return net.JoinHostPort(ip.String(), strconv.Itoa(portPort)), nil
+	return net.JoinHostPort(ip.String(), strconv.Itoa(podPort)), nil
 }
 
 func ipAddressFromPod(status corev1.PodStatus) (net.IP, error) {

pkg/parameters/configmanager/builder.go

@@ -37,7 +37,6 @@ import (
 	appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1"
 	parametersv1alpha1 "github.com/apecloud/kubeblocks/apis/parameters/v1alpha1"
 	"github.com/apecloud/kubeblocks/pkg/constant"
-	"github.com/apecloud/kubeblocks/pkg/controller/multicluster"
 	cfgutil "github.com/apecloud/kubeblocks/pkg/parameters/util"
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
@@ -116,12 +115,12 @@ func createOrUpdateConfigMap(configInfo []ConfigSpecInfo, manager *CfgManagerBui
 		if err := controllerutil.SetOwnerReference(manager.Cluster, cmObj, scheme); err != nil {
 			return err
 		}
-		return cli.Create(ctx, cmObj, inDataContext())
+		return cli.Create(ctx, cmObj)
 	}
 	updateConfigCM := func(cm *corev1.ConfigMap, newConfig string) error {
 		patch := client.MergeFrom(cm.DeepCopy())
 		cm.Data[configManagerConfig] = newConfig
-		return cli.Patch(ctx, cm, patch, inDataContext())
+		return cli.Patch(ctx, cm, patch)
 	}
 
 	config, err := cfgutil.ToYamlConfig(configInfo)
@@ -133,7 +132,7 @@ func createOrUpdateConfigMap(configInfo []ConfigSpecInfo, manager *CfgManagerBui
 		Namespace: manager.Cluster.GetNamespace(),
 		Name:      fmt.Sprintf("%s%s-%s-config-manager-config", configManagerCMPrefix, manager.Cluster.GetName(), manager.ComponentName),
 	}
-	err = cli.Get(ctx, cmKey, cmObj, inDataContext())
+	err = cli.Get(ctx, cmKey, cmObj)
 	switch {
 	default:
 		return err
@@ -279,7 +278,3 @@ func buildConfigManagerCommonArgs() []string {
 	args = append(args, "--log-level", viper.GetString(constant.ConfigManagerLogLevel))
 	return args
 }
-
-func inDataContext() *multicluster.ClientOption {
-	return multicluster.InDataContext()
-}