release v1.6.1

api/v1/polardbx/config.go

@@ -30,7 +30,6 @@ type CNStaticConfig struct {
 	EnableJvmRemoteDebug bool                          `json:"EnableJvmRemoteDebug,omitempty"`
 	ServerProperties     map[string]intstr.IntOrString `json:"ServerProperties,omitempty"`
 
-	// +kubebuilder:default=1
 	// +kubebuilder:validation:Enum=1;2;"1";"2";""
 
 	RPCProtocolVersion intstr.IntOrString `json:"RPCProtocolVersion,omitempty"`

build/images/xstore-tools/Dockerfile

@@ -65,7 +65,7 @@ RUN CGO_ENABLED=0 GOOS=linux GO111MODULE=on go build -tags polardbx -o polardbx-
 
 
 # Build the image with scripts
-FROM polardbx/xstore-tools-base:ts20240417185353
+FROM polardbx/xstore-tools-base:ts20240430102300
 
 ARG VERSION=test
 

charts/polardbx-operator/crds/polardbx.aliyun.com_polardbxbackups.yaml

@@ -189,7 +189,6 @@ spec:
                                 anyOf:
                                 - type: integer
                                 - type: string
-                                default: 1
                                 enum:
                                 - 1
                                 - 2

charts/polardbx-operator/crds/polardbx.aliyun.com_polardbxclusters.yaml

@@ -151,7 +151,6 @@ spec:
                             anyOf:
                             - type: integer
                             - type: string
-                            default: 1
                             enum:
                             - 1
                             - 2
@@ -2410,7 +2409,6 @@ spec:
                                 anyOf:
                                 - type: integer
                                 - type: string
-                                default: 1
                                 enum:
                                 - 1
                                 - 2

charts/polardbx-operator/values.yaml

@@ -15,7 +15,7 @@ images:
   polardbxClinic: polardbx-clinic
 
 # Default image tag. Use app version if not specified or 'latest' if useLatestImage is true.
-imageTag: v1.6.0
+imageTag: v1.6.0-20240430
 
 # version will be written to pxc and xstore annotation 
 version: v1.6.0
@@ -28,7 +28,6 @@ imagePullPolicy: IfNotPresent
 
 # Image pull secrets of operator pods.
 imagePullSecrets:
-  - name: regcred
 
 # Default settings for PolarDB-X Cluster.
 clusterDefaults:

hack/copy_image.sh

@@ -16,7 +16,7 @@
 
 
 SOURCE_REPO=polarximages-registry.cn-zhangjiakou.cr.aliyuncs.com/daixingpolarximages
-SOURCE_VERSION=v1.6.0
+SOURCE_VERSION=v1.6.0-20240430
 DEST_REPO=polardbx
 DEST_VERSION=v1.6.0
 TARGETS="xstore-tools polardbx-operator probe-proxy polardbx-exporter polardbx-init polardbx-hpfs polardbx-job polardbx-clinic polardbx-logstash"

pkg/operator/v1/polardbx/controllers/polardbxcluster_controller.go

@@ -19,12 +19,11 @@ package controllers
 import (
 	"context"
 	"errors"
+	"github.com/alibaba/polardbx-operator/pkg/operator/hint"
 	"github.com/alibaba/polardbx-operator/pkg/operator/v1/polardbx/steps/instance/pitr"
 	polarxJson "github.com/alibaba/polardbx-operator/pkg/util/json"
 	"time"
 
-	"github.com/alibaba/polardbx-operator/pkg/operator/hint"
-
 	polardbxv1 "github.com/alibaba/polardbx-operator/api/v1"
 	polardbxv1polardbx "github.com/alibaba/polardbx-operator/api/v1/polardbx"
 	"github.com/alibaba/polardbx-operator/pkg/debug"
@@ -311,14 +310,14 @@ func (r *PolarDBXReconciler) newReconcileTask(rc *polardbxreconcile.Context, pol
 				control.When(readonly,
 					instancesteps.WaitUntilPrimaryCNDeploymentsRolledOut,
 				),
+				// Only add or update, never remove.
+				instancesteps.CreateOrReconcileDNs,
+				instancesteps.WaitUntilDNsReady,
 				instancesteps.CreateOrReconcileCNs,
 				instancesteps.CreateOrReconcileCDCs,
 				instancesteps.WaitUntilCNCDCPodsReady,
 				instancesteps.CreateOrReconcileColumnars,
-				// Only add or update, never remove.
-				instancesteps.CreateOrReconcileDNs,
 
-				instancesteps.WaitUntilDNsReady,
 				instancesteps.WaitUntilCNDeploymentsRolledOut,
 				instancesteps.WaitUntilCDCDeploymentsRolledOut,
 				instancesteps.WaitUntilColumnarDeploymentsRolledOut,

pkg/operator/v1/polardbx/steps/instance/object.go

@@ -18,6 +18,7 @@ package instance
 
 import (
 	"fmt"
+	"k8s.io/apimachinery/pkg/api/equality"
 	"sort"
 	"strconv"
 	"strings"
@@ -541,6 +542,24 @@ var RemoveTrailingDNs = polardbxv1reconcile.NewStepBinder("RemoveTrailingDNs",
 	},
 )
 
+func isDeployPodSpecChanged(oldDeploy *appsv1.Deployment, newDeploy *appsv1.Deployment) bool {
+	// engine container
+	oldPodSpec := oldDeploy.Spec.Template.Spec
+	oldContainer := rebuildFromContainer(k8shelper.GetContainerFromPodSpec(&oldPodSpec, convention.ContainerEngine))
+	newPodSpec := newDeploy.Spec.Template.Spec
+	newContainer := rebuildFromContainer(k8shelper.GetContainerFromPodSpec(&newPodSpec, convention.ContainerEngine))
+	return !equality.Semantic.DeepEqual(oldContainer, newContainer)
+}
+
+func rebuildFromContainer(container *corev1.Container) *corev1.Container {
+	return &corev1.Container{
+		Image:   container.Image,
+		Ports:   container.Ports,
+		EnvFrom: container.EnvFrom,
+		Env:     container.Env,
+	}
+}
+
 func reconcileGroupedDeployments(rc *polardbxv1reconcile.Context, flow control.Flow, role string) (reconcile.Result, error) {
 	polardbxmeta.AssertRoleIn(role, polardbxmeta.RoleCN, polardbxmeta.RoleCDC, polardbxmeta.RoleColumnar)
 
@@ -593,11 +612,19 @@ func reconcileGroupedDeployments(rc *polardbxv1reconcile.Context, flow control.F
 				}
 				anyChanged = true
 			} else {
-				newDeploymentLabelHash := newDeployment.Labels[polardbxmeta.LabelHash]
-				if newDeploymentLabelHash != observedDeployment.Labels[polardbxmeta.LabelHash] {
+				if *(newDeployment.Spec.Replicas) != *(observedDeployment.Spec.Replicas) {
+					observedDeployment.Spec.Replicas = newDeployment.Spec.Replicas
+					err := rc.Client().Update(rc.Context(), observedDeployment)
+					if err != nil {
+						return flow.Error(err, "Unable to update deployment.",
+							"deployment", observedDeployment.Name)
+					}
+
+				} else if isDeployPodSpecChanged(observedDeployment, &newDeployment) {
+					newDeployHash := newDeployment.Labels[polardbxmeta.LabelHash]
 					convention.CopyMetadataForUpdate(&newDeployment.ObjectMeta, &observedDeployment.ObjectMeta, observedGeneration)
 					newDeployment.SetLabels(k8shelper.PatchLabels(newDeployment.Labels, map[string]string{
-						polardbxmeta.LabelHash: newDeploymentLabelHash,
+						polardbxmeta.LabelHash: newDeployHash,
 					}))
 					err := rc.Client().Update(rc.Context(), &newDeployment)
 					if err != nil {

pkg/operator/v1/xstore/change/driver/exec/delete_executor.go

@@ -86,6 +86,7 @@ func (exec *DeleteExec) Execute(
 			rc.Context(),
 			pod,
 			client.PropagationPolicy(metav1.DeletePropagationBackground),
+			client.GracePeriodSeconds(xstoreconvention.PodGracePeriodSeconds),
 		); err != nil {
 			return flow.Error(err, "Failed to delete the pod", "pod", exec.Step().Target)
 		}

pkg/operator/v1/xstore/change/driver/exec/update_executor.go

@@ -73,7 +73,7 @@ func (exec *UpdateExec) Execute(rc *xstorev1reconcile.Context, flow control.Flow
 
 			leaderLocalInfo, err := xstoreinstance.ShowThis(rc, leaderPod, flow.Logger(), true)
 			if err != nil || !strings.EqualFold(leaderLocalInfo.Role, xstoremeta.RoleLeader) {
-				return flow.RetryAfter(time.Minute, "Failed to query local info on leader pod.", "pod", leaderPod.Name)
+				return flow.RetryAfter(5*time.Second, "Failed to query local info on leader pod.", "pod", leaderPod.Name)
 			}
 
 			globalInfoItems, err := xstoreinstance.ShowGlobalInfo(rc, leaderPod, flow.Logger())
@@ -106,6 +106,7 @@ func (exec *UpdateExec) Execute(rc *xstorev1reconcile.Context, flow control.Flow
 				rc.Context(),
 				pod,
 				client.PropagationPolicy(metav1.DeletePropagationBackground),
+				client.GracePeriodSeconds(xstoreconvention.PodGracePeriodSeconds),
 			); err != nil {
 				return flow.Error(err, "Unable to delete the pod", "pod", target)
 			}

pkg/operator/v1/xstore/change/driver/planner/planner.go

@@ -111,8 +111,9 @@ func (p *Planner) buildExpectedNodes() map[string]model.PaxosNode {
 func (p *Planner) createNodeRebuildConfig(podName string) map[string]interface{} {
 	rebuildConfig := make(map[string]interface{})
 	config := xstoremeta.RebuildConfig{
-		LogSeparation: strconv.FormatBool(p.xstore.Spec.Config.Dynamic.LogDataSeparation),
-		NodeName:      p.xstore.Status.BoundVolumes[podName].Host,
+		LogSeparation:      strconv.FormatBool(p.xstore.Spec.Config.Dynamic.LogDataSeparation),
+		NodeName:           p.xstore.Status.BoundVolumes[podName].Host,
+		RpcProtocolVersion: p.xstore.Spec.Config.Dynamic.RpcProtocolVersion.String(),
 	}
 	rebuildConfig[xstoremeta.LabelConfigHash] = config.ComputeHash()
 	return rebuildConfig

pkg/operator/v1/xstore/controllers/xstore_backup_controller.go

@@ -57,7 +57,7 @@ func (r *XStoreBackupReconciler) Reconcile(ctx context.Context, request reconcil
 	)
 	defer rc.Close()
 
-	// Verify the existence of the xstore.
+	// Verify the existence of the xstore backup.
 	xstoreBackup, err := rc.GetXStoreBackup()
 	if err != nil {
 		if apierrors.IsNotFound(err) {
@@ -69,9 +69,16 @@ func (r *XStoreBackupReconciler) Reconcile(ctx context.Context, request reconcil
 		}
 	}
 
+	// Check whether backup is dummy
+	if xstoreBackup.Annotations[meta.AnnotationDummyBackup] == "true" || xstoreBackup.Annotations[xstoremeta.AnnotationDummyBackup] == "true" {
+		log.Info("Dummy xstore backup, skip")
+		return reconcile.Result{}, nil
+	}
+
 	// Record the context of the corresponding xstore
 	xstoreRequest := request
-	if xstoreBackup.GetDeletionTimestamp().IsZero() { // If request to delete, no need to care about xstore
+	if xstoreBackup.GetDeletionTimestamp().IsZero() && xstoreBackup.Status.Phase != xstorev1.XStoreBackupFinished {
+		// If backup has not finished nor been deleted, reconciler needs to get xstore
 		xstore, err := rc.GetXStore()
 		if err != nil {
 			log.Error(err, "Unable to get corresponding xstore")
@@ -94,12 +101,6 @@ func (r *XStoreBackupReconciler) Reconcile(ctx context.Context, request reconcil
 		return reconcile.Result{}, nil
 	}
 
-	// check whether backup is dummy
-	if xstoreBackup.Annotations[meta.AnnotationDummyBackup] == "true" || xstoreBackup.Annotations[xstoremeta.AnnotationDummyBackup] == "true" {
-		log.Info("Dummy xstore backup, skip")
-		return reconcile.Result{}, nil
-	}
-
 	return reconciler.Reconcile(rc, log.WithValues("engine", engine), request)
 }
 

pkg/operator/v1/xstore/controllers/xstore_controller.go

@@ -38,8 +38,9 @@ import (
 	"github.com/alibaba/polardbx-operator/pkg/k8s/control"
 	"github.com/alibaba/polardbx-operator/pkg/operator/v1/config"
 	"github.com/alibaba/polardbx-operator/pkg/operator/v1/xstore/plugin"
-	_ "github.com/alibaba/polardbx-operator/pkg/operator/v1/xstore/plugin/engines"
 	xstorev1reconcile "github.com/alibaba/polardbx-operator/pkg/operator/v1/xstore/reconcile"
+
+	_ "github.com/alibaba/polardbx-operator/pkg/operator/v1/xstore/plugin/engines"
 )
 
 type XStoreReconciler struct {

pkg/operator/v1/xstore/convention/convention.go

@@ -45,6 +45,8 @@ const (
 	ServiceTypeStaticClusterIp ServiceType = "staticClusterIp"
 )
 
+const PodGracePeriodSeconds = 300
+
 func GetXStoreServiceName(xstore *polardbxv1.XStore) string {
 	if xstore.Spec.ServiceName == "" {
 		return xstore.Name

pkg/operator/v1/xstore/factory/pod.go

@@ -214,7 +214,7 @@ func NewPod(rc *reconcile.Context, xstore *polardbxv1.XStore, nodeSet *polardbxv
 			ImagePullSecrets:              template.Spec.ImagePullSecrets,
 			DNSPolicy:                     corev1.DNSClusterFirstWithHostNet,
 			RestartPolicy:                 corev1.RestartPolicyAlways,
-			TerminationGracePeriodSeconds: pointer.Int64Ptr(300),
+			TerminationGracePeriodSeconds: pointer.Int64Ptr(convention.PodGracePeriodSeconds),
 			HostNetwork:                   boolutil.IsTrue(template.Spec.HostNetwork),
 			ShareProcessNamespace:         pointer.BoolPtr(true),
 			Affinity:                      opts.NewAffinity(factoryCtx),

pkg/operator/v1/xstore/factory/pod_extra.go

@@ -325,6 +325,7 @@ func (f *DefaultExtraPodFactory) NewEnvs(ctx *PodFactoryContext) (map[string][]c
 				{Name: "LIMITS_CPU", Value: strconv.FormatInt(resources.Limits.Cpu().MilliValue(), 10)},
 				{Name: "LIMITS_MEM", Value: strconv.FormatInt(resources.Limits.Memory().Value(), 10)},
 				{Name: "LOG_DATA_SEPARATION", Value: strconv.FormatBool(ctx.xstore.Spec.Config.Dynamic.LogDataSeparation)},
+				{Name: "RPC_PROTOCOL_VERSION", Value: ctx.xstore.Spec.Config.Dynamic.RpcProtocolVersion.String()},
 			},
 			f.newEnvsForEnginePorts(ctx),
 			engineEnvs,

pkg/operator/v1/xstore/meta/rebuild_config.go

@@ -3,8 +3,9 @@ package meta
 import "github.com/alibaba/polardbx-operator/pkg/meta/core/gms/security"
 
 type RebuildConfig struct {
-	LogSeparation string
-	NodeName      string
+	LogSeparation      string
+	NodeName           string
+	RpcProtocolVersion string
 }
 
 func (r *RebuildConfig) ComputeHash() string {

pkg/operator/v1/xstore/plugin/galaxy/reconcilers/galaxy_backup_reconciler.go

@@ -35,7 +35,7 @@ func (r *GalaxyBackupReconciler) Reconcile(rc *xstorev1reconcile.BackupContext,
 
 	isStandard := true
 	var err error
-	if backup.GetDeletionTimestamp().IsZero() {
+	if backup.GetDeletionTimestamp().IsZero() && backup.Status.Phase != xstorev1.XStoreBackupFinished {
 		isStandard, err = rc.GetXStoreIsStandard()
 		if err != nil {
 			log.Error(err, "Unable to get corresponding xstore")
@@ -51,7 +51,6 @@ func (r *GalaxyBackupReconciler) Reconcile(rc *xstorev1reconcile.BackupContext,
 }
 
 func (r *GalaxyBackupReconciler) newReconcileTask(rc *xstorev1reconcile.BackupContext, xstoreBackup *xstorev1.XStoreBackup, log logr.Logger, isStandard bool) (*control.Task, error) {
-
 	task := control.NewTask()
 
 	defer backupsteps.PersistentStatusChanges(task, true)

pkg/pitr/workflow.go

@@ -142,8 +142,8 @@ func LoadAllBinlog(pCtx *Context) error {
 					dataFilepath := vals[0]
 					metaFilepath := vals[1]
 					if dataFilepath == "" || metaFilepath == "" {
-						err := fmt.Errorf("invalid filepath, dataFilepath = %s, metaFilepath = %s", dataFilepath, metaFilepath)
-						return err
+						pCtx.Logger.Info(fmt.Sprintf("invalid filepath, dataFilepath = %s, metaFilepath = %s", dataFilepath, metaFilepath))
+						continue
 					}
 					binlogSources = append(binlogSources, BinlogSource{
 						Filename: filename,

tools/xstore/core/engine/galaxy/engine.py

@@ -225,11 +225,18 @@ def _default_dynamic_config(self) -> configparser.ConfigParser:
             'loose_mysqlx_max_connections': dynamic_config["mysqlx_max_connections"],
             'loose_galaxy_max_connections': dynamic_config["loose_galaxy_max_connections"],
             'default_time_zone': '+08:00',
-            'loose_new_rpc': self.new_rpc_enabled,
+            'loose_new_rpc': "on" if self.new_rpc_enabled else "off",
         }
         MySQLConfigManager.canonical_options(config)
         return config
 
+    def _final_config(self):
+        config = configparser.ConfigParser(allow_no_value=True)
+        config['mysqld'] = {}
+        if self.new_rpc_enabled:
+            config['mysqld']['loose_new_rpc'] = 'on'
+        return config
+
     def update_config(self, **override):
         mgr = MySQLConfigManager(self.file_config)
         override_config = self._default_dynamic_config()
@@ -243,6 +250,7 @@ def update_config(self, **override):
         override_configs += [self.file_config_dynamic, self._system_config()]
         if os.path.exists(self.file_config_override):
             override_configs += [self.file_config_override]
+        override_configs += [self._final_config()]
 
         mgr.update(template_config_file, overrides=override_configs)