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

pkg/apis/pingcap/v1alpha1/crd_kinds.go

@@ -26,7 +26,7 @@ const (
 	TiDBClusterKindKey = "tidbcluster"
 
 	DMClusterName    = "dmclusters"
-	DMCLusterKind    = "DMCluster"
+	DMClusterKind    = "DMCluster"
 	DMClusterKindKey = "dmcluster"
 
 	BackupName    = "backups"
@@ -89,7 +89,7 @@ type CrdKinds struct {
 var DefaultCrdKinds = CrdKinds{
 	KindsString:           "",
 	TiDBCluster:           CrdKind{Plural: TiDBClusterName, Kind: TiDBClusterKind, ShortNames: []string{"tc"}, SpecName: SpecPath + TiDBClusterKind},
-	DMCluster:             CrdKind{Plural: DMClusterName, Kind: DMCLusterKind, ShortNames: []string{"dc"}, SpecName: SpecPath + DMCLusterKind},
+	DMCluster:             CrdKind{Plural: DMClusterName, Kind: DMClusterKind, ShortNames: []string{"dc"}, SpecName: SpecPath + DMClusterKind},
 	Backup:                CrdKind{Plural: BackupName, Kind: BackupKind, ShortNames: []string{"bk"}, SpecName: SpecPath + BackupKind},
 	Restore:               CrdKind{Plural: RestoreName, Kind: RestoreKind, ShortNames: []string{"rt"}, SpecName: SpecPath + RestoreKind},
 	BackupSchedule:        CrdKind{Plural: BackupScheduleName, Kind: BackupScheduleKind, ShortNames: []string{"bks"}, SpecName: SpecPath + BackupScheduleKind},

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 {
@@ -33,6 +38,14 @@ func (dc *DMCluster) Timezone() string {
 	return tz
 }
 
+func (dc *DMCluster) IsPVReclaimEnabled() bool {
+	enabled := dc.Spec.EnablePVReclaim
+	if enabled == nil {
+		return defaultEnablePVReclaim
+	}
+	return *enabled
+}
+
 func (dc *DMCluster) IsTLSClusterEnabled() bool {
 	return dc.Spec.TLSCluster != nil && dc.Spec.TLSCluster.Enabled
 }
@@ -84,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
 }
@@ -130,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/controller/dmcluster/dm_cluster_control.go

@@ -41,41 +41,41 @@ func NewDefaultDMClusterControl(
 	dcControl controller.DMClusterControlInterface,
 	masterMemberManager manager.DMManager,
 	workerMemberManager manager.DMManager,
-	//reclaimPolicyManager manager.DMManager,
+	reclaimPolicyManager manager.DMManager,
 	//metaManager manager.DMManager,
-	//orphanPodsCleaner member.OrphanPodsCleaner,
+	orphanPodsCleaner member.OrphanPodsCleaner,
 	pvcCleaner member.PVCCleanerInterface,
 	pvcResizer member.PVCResizerInterface,
-	//podRestarter member.PodRestarter,
+	podRestarter member.PodRestarter,
 	conditionUpdater DMClusterConditionUpdater,
 	recorder record.EventRecorder) ControlInterface {
 	return &defaultDMClusterControl{
 		dcControl,
 		masterMemberManager,
 		workerMemberManager,
-		//reclaimPolicyManager,
+		reclaimPolicyManager,
 		//metaManager,
-		//orphanPodsCleaner,
+		orphanPodsCleaner,
 		pvcCleaner,
-		//podRestarter,
+		podRestarter,
 		pvcResizer,
 		conditionUpdater,
 		recorder,
 	}
 }
 
 type defaultDMClusterControl struct {
-	dcControl           controller.DMClusterControlInterface
-	masterMemberManager manager.DMManager
-	workerMemberManager manager.DMManager
-	//reclaimPolicyManager manager.DMManager
+	dcControl            controller.DMClusterControlInterface
+	masterMemberManager  manager.DMManager
+	workerMemberManager  manager.DMManager
+	reclaimPolicyManager manager.DMManager
 	//metaManager       manager.DMManager
-	//orphanPodsCleaner member.OrphanPodsCleaner
-	pvcCleaner member.PVCCleanerInterface
-	//podRestarter     member.PodRestarter
-	pvcResizer       member.PVCResizerInterface
-	conditionUpdater DMClusterConditionUpdater
-	recorder         record.EventRecorder
+	orphanPodsCleaner member.OrphanPodsCleaner
+	pvcCleaner        member.PVCCleanerInterface
+	podRestarter      member.PodRestarter
+	pvcResizer        member.PVCResizerInterface
+	conditionUpdater  DMClusterConditionUpdater
+	recorder          record.EventRecorder
 }
 
 // UpdateStatefulSet executes the core logic loop for a dmcluster.
@@ -123,29 +123,25 @@ func (dcc *defaultDMClusterControl) validate(dc *v1alpha1.DMCluster) bool {
 
 func (dcc *defaultDMClusterControl) updateDMCluster(dc *v1alpha1.DMCluster) error {
 	var errs []error
-	// TODO: implement reclaimPolicyManager
-	// syncing all PVs managed by operator's reclaim policy to Retain
-	// if err := dcc.reclaimPolicyManager.Sync(dc); err != nil {
-	//	return err
-	// }
+	if err := dcc.reclaimPolicyManager.SyncDM(dc); err != nil {
+		return err
+	}
 
-	// TODO: add orphanPodsCleaner for dm cluster
-	// cleaning all orphan pods(pd, tikv or tiflash which don't have a related PVC) managed by operator
-	// skipReasons, err := dcc.orphanPodsCleaner.Clean(dc)
-	// if err != nil {
-	// 	return err
-	// }
-	// if klog.V(10) {
-	// 	for podName, reason := range skipReasons {
-	// 		klog.Infof("pod %s of cluster %s/%s is skipped, reason %q", podName, dc.Namespace, dc.Name, reason)
-	// 	}
-	// }
+	// cleaning all orphan pods(dm-master or dm-worker which don't have a related PVC) managed by operator
+	skipReasons, err := dcc.orphanPodsCleaner.Clean(dc)
+	if err != nil {
+		return err
+	}
+	if klog.V(10) {
+		for podName, reason := range skipReasons {
+			klog.Infof("pod %s of cluster %s/%s is skipped, reason %q", podName, dc.Namespace, dc.Name, reason)
+		}
+	}
 
-	// TODO: restarted pods in dm cluster
 	// sync all the pods which need to be restarted
-	// if err := dcc.podRestarter.Sync(dc); err != nil {
-	// 	return err
-	// }
+	if err := dcc.podRestarter.Sync(dc); err != nil {
+		return err
+	}
 
 	// works that should do to making the dm-master cluster current state match the desired state:
 	//   - create or update the dm-master service
@@ -158,7 +154,7 @@ func (dcc *defaultDMClusterControl) updateDMCluster(dc *v1alpha1.DMCluster) erro
 	//   - upgrade the dm-master cluster
 	//   - scale out/in the dm-master cluster
 	//   - failover the dm-master cluster
-	if err := dcc.masterMemberManager.Sync(dc); err != nil {
+	if err := dcc.masterMemberManager.SyncDM(dc); err != nil {
 		errs = append(errs, err)
 	}
 
@@ -170,7 +166,7 @@ func (dcc *defaultDMClusterControl) updateDMCluster(dc *v1alpha1.DMCluster) erro
 	//   - upgrade the dm-worker cluster
 	//   - scale out/in the dm-worker cluster
 	//   - failover the dm-worker cluster
-	if err := dcc.workerMemberManager.Sync(dc); err != nil {
+	if err := dcc.workerMemberManager.SyncDM(dc); err != nil {
 		errs = append(errs, err)
 	}
 
@@ -182,16 +178,15 @@ func (dcc *defaultDMClusterControl) updateDMCluster(dc *v1alpha1.DMCluster) erro
 	// 	return err
 	// }
 
-	// TODO: clean pods cleaning the pod scheduling annotation for dm cluster
-	// pvcSkipReasons, err := dcc.pvcCleaner.Clean(dc)
-	// if err != nil {
-	// 	return err
-	// }
-	// if klog.V(10) {
-	// 	for pvcName, reason := range pvcSkipReasons {
-	// 		klog.Infof("pvc %s of cluster %s/%s is skipped, reason %q", pvcName, dc.Namespace, dc.Name, reason)
-	// 	}
-	// }
+	pvcSkipReasons, err := dcc.pvcCleaner.Clean(dc)
+	if err != nil {
+		return err
+	}
+	if klog.V(10) {
+		for pvcName, reason := range pvcSkipReasons {
+			klog.Infof("pvc %s of cluster %s/%s is skipped, reason %q", pvcName, dc.Namespace, dc.Name, reason)
+		}
+	}
 
 	// TODO: sync dm cluster attributes
 	// syncing the some tidbcluster status attributes

pkg/controller/dmcluster/dm_cluster_controller.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/controller"
 	"github.com/pingcap/tidb-operator/pkg/dmapi"
 	mm "github.com/pingcap/tidb-operator/pkg/manager/member"
+	"github.com/pingcap/tidb-operator/pkg/manager/meta"
 
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
@@ -97,9 +98,12 @@ func NewController(
 	svcControl := controller.NewRealServiceControl(kubeCli, svcInformer.Lister(), recorder)
 	pvControl := controller.NewRealPVControl(kubeCli, pvcInformer.Lister(), pvInformer.Lister(), recorder)
 	pvcControl := controller.NewRealPVCControl(kubeCli, recorder, pvcInformer.Lister())
-	//podControl := controller.NewRealPodControl(kubeCli, nil, podInformer.Lister(), recorder)
+	podControl := controller.NewRealPodControl(kubeCli, nil, podInformer.Lister(), recorder)
 	typedControl := controller.NewTypedControl(controller.NewRealGenericControl(genericCli, recorder))
+	masterScaler := mm.NewMasterScaler(masterControl, pvcInformer.Lister(), pvcControl)
 	masterUpgrader := mm.NewMasterUpgrader(masterControl, podInformer.Lister())
+	workerScaler := mm.NewWorkerScaler(pvcInformer.Lister(), pvcControl)
+	podRestarter := mm.NewPodRestarter(kubeCli, podInformer.Lister())
 
 	dcc := &Controller{
 		kubeClient: kubeCli,
@@ -116,6 +120,7 @@ func NewController(
 				podInformer.Lister(),
 				epsInformer.Lister(),
 				pvcInformer.Lister(),
+				masterScaler,
 				masterUpgrader,
 			),
 			mm.NewWorkerMemberManager(
@@ -126,12 +131,13 @@ func NewController(
 				setInformer.Lister(),
 				svcInformer.Lister(),
 				podInformer.Lister(),
+				workerScaler,
+			),
+			meta.NewReclaimPolicyDMManager(
+				pvcInformer.Lister(),
+				pvInformer.Lister(),
+				pvControl,
 			),
-			//meta.NewReclaimPolicyManager(
-			//	pvcInformer.Lister(),
-			//	pvInformer.Lister(),
-			//	pvControl,
-			//),
 			//meta.NewMetaManager(
 			//	pvcInformer.Lister(),
 			//	pvcControl,
@@ -140,12 +146,12 @@ func NewController(
 			//	podInformer.Lister(),
 			//	podControl,
 			//),
-			//mm.NewOrphanPodsCleaner(
-			//	podInformer.Lister(),
-			//	podControl,
-			//	pvcInformer.Lister(),
-			//	kubeCli,
-			//),
+			mm.NewOrphanPodsCleaner(
+				podInformer.Lister(),
+				podControl,
+				pvcInformer.Lister(),
+				kubeCli,
+			),
 			mm.NewRealPVCCleaner(
 				kubeCli,
 				podInformer.Lister(),
@@ -160,7 +166,7 @@ func NewController(
 				scInformer,
 			),
 			//mm.NewDMClusterStatusManager(kubeCli, cli, scalerInformer.Lister(), tikvGroupInformer.Lister()),
-			//podRestarter,
+			podRestarter,
 			&dmClusterConditionUpdater{},
 			recorder,
 		),

pkg/dmapi/dmapi.go

@@ -34,6 +34,8 @@ type MasterClient interface {
 	GetWorkers() ([]*WorkersInfo, error)
 	GetLeader() (MembersLeader, error)
 	EvictLeader() error
+	DeleteMaster(name string) error
+	DeleteWorker(name string) error
 }
 
 var (
@@ -195,6 +197,34 @@ func (mc *masterClient) EvictLeader() error {
 	return nil
 }
 
+func (mc *masterClient) deleteMember(query string) error {
+	apiURL := fmt.Sprintf("%s/%s%s", mc.url, membersPrefix, query)
+	body, err := httputil.DeleteBodyOK(mc.httpClient, apiURL)
+	if err != nil {
+		return err
+	}
+	deleteMemeberResp := &RespHeader{}
+	err = json.Unmarshal(body, deleteMemeberResp)
+	if err != nil {
+		return fmt.Errorf("unable to unmarshal delete member resp: %s, query: %s, err: %s", body, query, err)
+	}
+	if !deleteMemeberResp.Result {
+		return fmt.Errorf("unable to delete member, query: %s, err: %s", query, deleteMemeberResp.Msg)
+	}
+
+	return nil
+}
+
+func (mc *masterClient) DeleteMaster(name string) error {
+	query := "/master/" + name
+	return mc.deleteMember(query)
+}
+
+func (mc *masterClient) DeleteWorker(name string) error {
+	query := "/worker/" + name
+	return mc.deleteMember(query)
+}
+
 // NewMasterClient returns a new MasterClient
 func NewMasterClient(url string, timeout time.Duration, tlsConfig *tls.Config, disableKeepalive bool) MasterClient {
 	return &masterClient{