dm-operator/: support scaling a dm cluster with dm-masters and dm-workers

pkg/apis/pingcap/v1alpha1/dmcluster.go

@@ -14,8 +14,13 @@
 package v1alpha1
 
 import (
+	"encoding/json"
 	"fmt"
 	"strings"
+
+	"github.com/pingcap/advanced-statefulset/client/apis/apps/v1/helper"
+	"github.com/pingcap/tidb-operator/pkg/label"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 func (dc *DMCluster) Scheme() string {
@@ -92,6 +97,17 @@ func (dc *DMCluster) WorkerStsDesiredReplicas() int32 {
 	return dc.Spec.Worker.Replicas
 }
 
+func (dc *DMCluster) WorkerStsDesiredOrdinals(excludeFailover bool) sets.Int32 {
+	if dc.Spec.Worker == nil {
+		return sets.Int32{}
+	}
+	replicas := dc.Spec.Worker.Replicas
+	if !excludeFailover {
+		replicas = dc.WorkerStsDesiredReplicas()
+	}
+	return helper.GetPodOrdinalsFromReplicasAndDeleteSlots(replicas, dc.getDeleteSlots(label.DMWorkerLabelVal))
+}
+
 func (dc *DMCluster) GetInstanceName() string {
 	return dc.Name
 }
@@ -138,6 +154,33 @@ func (dc *DMCluster) MasterScaling() bool {
 	return dc.Status.Master.Phase == ScalePhase
 }
 
+func (dc *DMCluster) getDeleteSlots(component string) (deleteSlots sets.Int32) {
+	deleteSlots = sets.NewInt32()
+	annotations := dc.GetAnnotations()
+	if annotations == nil {
+		return deleteSlots
+	}
+	var key string
+	if component == label.DMMasterLabelVal {
+		key = label.AnnDMMasterDeleteSlots
+	} else if component == label.DMWorkerLabelVal {
+		key = label.AnnDMWorkerDeleteSlots
+	} else {
+		return
+	}
+	value, ok := annotations[key]
+	if !ok {
+		return
+	}
+	var slice []int32
+	err := json.Unmarshal([]byte(value), &slice)
+	if err != nil {
+		return
+	}
+	deleteSlots.Insert(slice...)
+	return
+}
+
 func (dc *DMCluster) MasterIsAvailable() bool {
 	lowerLimit := dc.Spec.Master.Replicas/2 + 1
 	if int32(len(dc.Status.Master.Members)) < lowerLimit {

pkg/manager/member/dm_worker_member_manager.go

@@ -269,10 +269,7 @@ func (wmm *workerMemberManager) syncDMClusterStatus(dc *v1alpha1.DMCluster, set
 
 		// offline the workers that already been scaled-in
 		if status.Stage == "offline" {
-			ordinal, err := util.GetOrdinalFromPodName(worker.Name)
-			if err != nil {
-				klog.Errorf("invalid worker name %s, can't offline this worker automatically, err: %s", worker.Name, err)
-			} else if ordinal >= dc.WorkerStsDesiredReplicas() {
+			if !isWorkerPodDesired(dc, name) {
 				err := dmClient.DeleteWorker(name)
 				if err != nil {
 					klog.Errorf("fail to remove worker %s, err: %s", worker.Name, err)
@@ -549,3 +546,13 @@ func getWorkerConfigMap(dc *v1alpha1.DMCluster) (*corev1.ConfigMap, error) {
 	}
 	return cm, nil
 }
+
+func isWorkerPodDesired(dc *v1alpha1.DMCluster, podName string) bool {
+	ordinals := dc.WorkerStsDesiredOrdinals(true)
+	ordinal, err := util.GetOrdinalFromPodName(podName)
+	if err != nil {
+		klog.Errorf("unexpected pod name %q: %v", podName, err)
+		return false
+	}
+	return ordinals.Has(ordinal)
+}

pkg/manager/member/dm_worker_scaler.go

@@ -77,7 +77,10 @@ func (wsd *workerScaler) ScaleOut(meta metav1.Object, oldSet *apps.StatefulSet,
 	return nil
 }
 
-// We need remove member from cluster before reducing statefulset replicas
+// Different from dm-master, we need remove member from cluster **after** reducing statefulset replicas
+// Now it will be removed in syncing worker status. For dm-worker we can't remove its register info from dm-master
+// when it's still alive. So we delete it later after its keepalive lease is outdated or revoked.
+// We can defer deleting dm-worker register info because dm-master will patch replication task through keepalive info.
 // only remove one member at a time when scale down
 func (wsd *workerScaler) ScaleIn(meta metav1.Object, oldSet *apps.StatefulSet, newSet *apps.StatefulSet) error {
 	dc, ok := meta.(*v1alpha1.DMCluster)