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

examples/dm/README.md

@@ -8,7 +8,7 @@ The following steps will create a DM cluster.
 
 **Prerequisites**: 
 - Has TiDB operator `v1.2.0` or higher version installed. [Doc](https://pingcap.com/docs/stable/tidb-in-kubernetes/deploy/tidb-operator/)
-- Has default `StorageClass` configured, and there are enough PVs (by default, 6 PVs are required) of that storageClass:
+- Has default `StorageClass` configured, and there are enough PVs (by default, 2 PVs are required) of that storageClass:
 
   This could by verified by the following command:
 
@@ -24,7 +24,7 @@ The following steps will create a DM cluster.
   gold                 kubernetes.io/gce-pd      1d
   ```
 
-  Alternatively, you could specify the storageClass explicitly by modifying `tidb-cluster.yaml`.
+  Alternatively, you could specify the storageClass explicitly by modifying `dm-cluster.yaml`.
 
 ## Install
 
@@ -47,7 +47,7 @@ watch kubectl -n <namespace> get pod
 Explore the DM master interface:
 
 ```bash
-> kubectl -n <namespace> port-forward svc/basic-dm-master 8261:8261 &>/tmp/pf-dm.log &
+> kubectl -n <namespace> port-forward svc/basic-dm-master 8261:8261
 > dmctl --master-addr 127.0.0.1:8261 list-member
 ```
 

examples/dm/dm-cluster.yaml

@@ -10,15 +10,15 @@ spec:
     address: "http://basic-discovery.demo:10261"
   master:
     baseImage: pingcap/dm
-    replicas: 3
+    replicas: 1
     # if storageClassName is not set, the default Storage Class of the Kubernetes cluster will be used
     # storageClassName: local-storage
     storageSize: "1Gi"
     requests: {}
     config: {}
   worker:
     baseImage: pingcap/dm
-    replicas: 3
+    replicas: 1
     # if storageClassName is not set, the default Storage Class of the Kubernetes cluster will be used
     # storageClassName: local-storage
     storageSize: "1Gi"

pkg/apis/pingcap/v1alpha1/defaulting/dmcluster.go

@@ -16,7 +16,6 @@ package defaulting
 import (
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/utils/pointer"
 )
 
 const (
@@ -56,9 +55,6 @@ func setMasterSpecDefault(dc *v1alpha1.DMCluster) {
 			dc.Spec.Master.BaseImage = defaultMasterImage
 		}
 	}
-	if dc.Spec.Master.MaxFailoverCount == nil {
-		dc.Spec.Master.MaxFailoverCount = pointer.Int32Ptr(3)
-	}
 }
 
 func setWorkerSpecDefault(dc *v1alpha1.DMCluster) {
@@ -67,7 +63,4 @@ func setWorkerSpecDefault(dc *v1alpha1.DMCluster) {
 			dc.Spec.Worker.BaseImage = defaultWorkerImage
 		}
 	}
-	if dc.Spec.Worker.MaxFailoverCount == nil {
-		dc.Spec.Worker.MaxFailoverCount = pointer.Int32Ptr(3)
-	}
 }

pkg/apis/pingcap/v1alpha1/dmcluster.go

@@ -18,8 +18,6 @@ import (
 	"strings"
 )
 
-var latestImageVersion = "latest"
-
 func (dc *DMCluster) Scheme() string {
 	if dc.IsTLSClusterEnabled() {
 		return "https"
@@ -69,6 +67,7 @@ func (dc *DMCluster) MasterStsDesiredReplicas() int32 {
 	return dc.Spec.Master.Replicas + int32(len(dc.Status.Master.FailureMembers))
 }
 
+// TODO: support fail-over
 func (dc *DMCluster) WorkerStsDesiredReplicas() int32 {
 	if dc.Spec.Worker == nil {
 		return 0
@@ -82,35 +81,25 @@ func (dc *DMCluster) GetInstanceName() string {
 }
 
 func (dc *DMCluster) MasterImage() string {
-	image := dc.Spec.Master.Image
-	baseImage := dc.Spec.Master.BaseImage
-	// base image takes higher priority
-	if baseImage != "" {
-		version := dc.Spec.Master.Version
-		if version == nil {
-			version = &dc.Spec.Version
-		}
-		if version == nil {
-			version = &latestImageVersion
-		}
-		image = fmt.Sprintf("%s:%s", baseImage, *version)
+	image := dc.Spec.Master.BaseImage
+	version := dc.Spec.Master.Version
+	if version == nil {
+		version = &dc.Spec.Version
+	}
+	if *version != "" {
+		image = fmt.Sprintf("%s:%s", image, *version)
 	}
 	return image
 }
 
 func (dc *DMCluster) WorkerImage() string {
-	image := dc.Spec.Worker.Image
-	baseImage := dc.Spec.Worker.BaseImage
-	// base image takes higher priority
-	if baseImage != "" {
-		version := dc.Spec.Worker.Version
-		if version == nil {
-			version = &dc.Spec.Version
-		}
-		if version == nil {
-			version = &latestImageVersion
-		}
-		image = fmt.Sprintf("%s:%s", baseImage, *version)
+	image := dc.Spec.Worker.BaseImage
+	version := dc.Spec.Worker.Version
+	if version == nil {
+		version = &dc.Spec.Version
+	}
+	if *version != "" {
+		image = fmt.Sprintf("%s:%s", image, *version)
 	}
 	return image
 }
@@ -122,7 +111,7 @@ func (dc *DMCluster) MasterVersion() string {
 		return image[colonIdx+1:]
 	}
 
-	return latestImageVersion
+	return "latest"
 }
 
 func (dc *DMCluster) MasterUpgrading() bool {

pkg/controller/controller_utils.go

@@ -323,9 +323,14 @@ func DMMasterPeerMemberName(clusterName string) string {
 	return fmt.Sprintf("%s-dm-master-peer", clusterName)
 }
 
+// DMWorkerMemberName returns dm-worker member name
+func DMWorkerMemberName(clusterName string) string {
+	return fmt.Sprintf("%s-dm-worker", clusterName)
+}
+
 // DMWorkerPeerMemberName returns dm-worker peer service name
 func DMWorkerPeerMemberName(clusterName string) string {
-	return fmt.Sprintf("%s-dm-worker", clusterName)
+	return fmt.Sprintf("%s-dm-worker-peer", clusterName)
 }
 
 // AnnProm adds annotations for prometheus scraping metrics

pkg/controller/dmcluster/dm_cluster_control.go

@@ -45,6 +45,7 @@ func NewDefaultDMClusterControl(
 	//metaManager manager.DMManager,
 	//orphanPodsCleaner member.OrphanPodsCleaner,
 	pvcCleaner member.PVCCleanerInterface,
+	pvcResizer member.PVCResizerInterface,
 	//podRestarter member.PodRestarter,
 	conditionUpdater DMClusterConditionUpdater,
 	recorder record.EventRecorder) ControlInterface {
@@ -57,6 +58,7 @@ func NewDefaultDMClusterControl(
 		//orphanPodsCleaner,
 		pvcCleaner,
 		//podRestarter,
+		pvcResizer,
 		conditionUpdater,
 		recorder,
 	}
@@ -71,6 +73,7 @@ type defaultDMClusterControl struct {
 	//orphanPodsCleaner member.OrphanPodsCleaner
 	pvcCleaner member.PVCCleanerInterface
 	//podRestarter     member.PodRestarter
+	pvcResizer       member.PVCResizerInterface
 	conditionUpdater DMClusterConditionUpdater
 	recorder         record.EventRecorder
 }
@@ -194,5 +197,10 @@ func (dcc *defaultDMClusterControl) updateDMCluster(dc *v1alpha1.DMCluster) erro
 	// syncing the some tidbcluster status attributes
 	// 	- sync tidbmonitor reference
 	// return dcc.tidbClusterStatusManager.Sync(dc)
+
+	// resize PVC if necessary
+	if err := dcc.pvcResizer.ResizeDM(dc); err != nil {
+		errs = append(errs, err)
+	}
 	return errorutils.NewAggregate(errs)
 }

pkg/controller/dmcluster/dm_cluster_controller.go

@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/Masterminds/semver"
 	perrors "github.com/pingcap/errors"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
@@ -25,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"
+
 	apps "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -85,6 +87,7 @@ func NewController(
 	pvcInformer := kubeInformerFactory.Core().V1().PersistentVolumeClaims()
 	pvInformer := kubeInformerFactory.Core().V1().PersistentVolumes()
 	podInformer := kubeInformerFactory.Core().V1().Pods()
+	scInformer := kubeInformerFactory.Storage().V1().StorageClasses()
 	//nodeInformer := kubeInformerFactory.Core().V1().Nodes()
 	//secretInformer := kubeInformerFactory.Core().V1().Secrets()
 
@@ -150,6 +153,11 @@ func NewController(
 				pvInformer.Lister(),
 				pvControl,
 			),
+			mm.NewPVCResizer(
+				kubeCli,
+				pvcInformer,
+				scInformer,
+			),
 			//mm.NewDMClusterStatusManager(kubeCli, cli, scalerInformer.Lister(), tikvGroupInformer.Lister()),
 			//podRestarter,
 			&dmClusterConditionUpdater{},
@@ -245,6 +253,12 @@ func (dcc *Controller) sync(key string) error {
 	if err != nil {
 		return err
 	}
+	clusterVersionLT2, err := clusterVersionLessThan2(dc.MasterVersion())
+	if err != nil {
+		klog.V(4).Infof("cluster version: %s is not semantic versioning compatible", dc.MasterVersion())
+	} else if clusterVersionLT2 {
+		return fmt.Errorf("dm-operator only supports to deploy dm-2.0")
+	}
 
 	return dcc.syncDMCluster(dc.DeepCopy())
 }
@@ -362,3 +376,12 @@ func (dcc *Controller) resolveDMClusterFromSet(namespace string, set *apps.State
 	}
 	return dc
 }
+
+func clusterVersionLessThan2(version string) (bool, error) {
+	v, err := semver.NewVersion(version)
+	if err != nil {
+		return true, err
+	}
+
+	return v.Major() < 2, nil
+}

pkg/dmapi/master_control.go

@@ -19,6 +19,8 @@ import (
 	"net/http"
 	"sync"
 
+	"github.com/pingcap/tidb-operator/pkg/controller"
+
 	"github.com/pingcap/tidb-operator/pkg/pdapi"
 
 	"github.com/pingcap/tidb-operator/pkg/util"
@@ -82,7 +84,7 @@ func masterClientKey(scheme string, namespace Namespace, clusterName string) str
 
 // MasterClientURL builds the url of master client
 func MasterClientURL(namespace Namespace, clusterName string, scheme string) string {
-	return fmt.Sprintf("%s://%s-dm-master.%s:8261", scheme, clusterName, string(namespace))
+	return fmt.Sprintf("%s://%s.%s:8261", scheme, controller.DMMasterMemberName(clusterName), string(namespace))
 }
 
 // FakeMasterControl implements a fake version of MasterControlInterface.

pkg/manager/member/dm_master_member_manager.go

@@ -85,13 +85,6 @@ func NewMasterMemberManager(masterControl dmapi.MasterControlInterface,
 }
 
 func (mmm *masterMemberManager) Sync(dc *v1alpha1.DMCluster) error {
-	clusterVersionLT2, err := clusterVersionLessThan2(dc.MasterVersion())
-	if err != nil {
-		klog.V(4).Infof("cluster version: %s is not semantic versioning compatible", dc.MasterVersion())
-	} else if clusterVersionLT2 {
-		return fmt.Errorf("dm-operator only supports to deploy dm-2.0")
-	}
-
 	// Sync dm-master Service
 	if err := mmm.syncMasterServiceForDMCluster(dc); err != nil {
 		return err
@@ -324,7 +317,7 @@ func (mmm *masterMemberManager) syncDMClusterStatus(dc *v1alpha1.DMCluster, set
 		}
 		name := master.Name
 		if len(name) == 0 {
-			klog.Warningf("dm-master member: [%s] doesn't have a name, and can't get it from clientUrls: [%s], dm-master Info: [%v] in [%s/%s]",
+			klog.Warningf("dm-master member: [%s] doesn't have a name, clientUrls: [%s], dm-master Info: [%#v] in [%s/%s]",
 				id, master.ClientURLs, master, ns, dcName)
 			continue
 		}

pkg/manager/member/dm_worker_member_manager.go

@@ -172,9 +172,9 @@ func (wmm *workerMemberManager) syncWorkerStatefulSetForDMCluster(dc *v1alpha1.D
 	ns := dc.GetNamespace()
 	dcName := dc.GetName()
 
-	oldStsTmp, err := wmm.setLister.StatefulSets(ns).Get(controller.DMWorkerPeerMemberName(dcName))
+	oldStsTmp, err := wmm.setLister.StatefulSets(ns).Get(controller.DMWorkerMemberName(dcName))
 	if err != nil && !errors.IsNotFound(err) {
-		return fmt.Errorf("syncWorkerStatefulSetForDMCluster: failed to get sts %s for cluster %s/%s, error: %s", controller.DMWorkerPeerMemberName(dcName), ns, dcName, err)
+		return fmt.Errorf("syncWorkerStatefulSetForDMCluster: failed to get sts %s for cluster %s/%s, error: %s", controller.DMWorkerMemberName(dcName), ns, dcName, err)
 	}
 
 	stsNotExist := errors.IsNotFound(err)
@@ -211,11 +211,6 @@ func (wmm *workerMemberManager) syncWorkerStatefulSetForDMCluster(dc *v1alpha1.D
 		return nil
 	}
 
-	if dc.MasterUpgrading() {
-		klog.Warningf("dm-master is upgrading, skipping upgrade dm-worker")
-		return nil
-	}
-
 	return updateStatefulSet(wmm.setControl, dc, newSts, oldSts)
 }
 
@@ -384,7 +379,7 @@ func getNewWorkerSetForDMCluster(dc *v1alpha1.DMCluster, cm *corev1.ConfigMap) (
 	}
 
 	workerLabel := label.NewDM().Instance(instanceName).DMWorker()
-	setName := controller.DMWorkerPeerMemberName(dcName)
+	setName := controller.DMWorkerMemberName(dcName)
 	podAnnotations := CombineAnnotations(controller.AnnProm(8262), baseWorkerSpec.Annotations())
 	stsAnnotations := getStsAnnotations(dc.Annotations, label.DMWorkerLabelVal)
 
@@ -521,7 +516,7 @@ func getWorkerConfigMap(dc *v1alpha1.DMCluster) (*corev1.ConfigMap, error) {
 	workerLabel := label.NewDM().Instance(instanceName).DMWorker().Labels()
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:            controller.DMWorkerPeerMemberName(dc.Name),
+			Name:            controller.DMWorkerMemberName(dc.Name),
 			Namespace:       dc.Namespace,
 			Labels:          workerLabel,
 			OwnerReferences: []metav1.OwnerReference{controller.GetDMOwnerRef(dc)},