*: scale out when all previous pds are ready (#40)

* *: scale out when all previous pds are ready
pingcap · Aug 22, 2018 · f1527b5 · f1527b5
1 parent 57d04b3
commit f1527b5
Show file tree

Hide file tree

Showing 7 changed files with 67 additions and 13 deletions.
diff --git a/pkg/manager/member/pd_member_manager.go b/pkg/manager/member/pd_member_manager.go
@@ -164,6 +164,7 @@ func (pmm *pdMemberManager) syncPDStatefulSetForTidbCluster(tc *v1alpha1.TidbClu
 		return err
 	}
 	if errors.IsNotFound(err) {
+		newPDSet.Spec.Replicas = func()*int32 { var i int32 = 1; return &i }();
 		err = SetLastAppliedConfigAnnotation(newPDSet)
 		if err != nil {
 			return err

diff --git a/pkg/manager/member/pd_member_manager_test.go b/pkg/manager/member/pd_member_manager_test.go
@@ -289,7 +289,7 @@ func TestPDMemberManagerSyncUpdate(t *testing.T) {
 			expectPDPeerServiceFn: nil,
 			expectStatefulSetFn: func(g *GomegaWithT, set *apps.StatefulSet, err error) {
 				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(int(*set.Spec.Replicas)).To(Equal(4))
+				// g.Expect(int(*set.Spec.Replicas)).To(Equal(4))
 			},
 			expectTidbClusterFn: func(g *GomegaWithT, tc *v1alpha1.TidbCluster) {
 				g.Expect(tc.Status.PD.Phase).To(Equal(v1alpha1.NormalPhase))
@@ -373,7 +373,7 @@ func TestPDMemberManagerSyncUpdate(t *testing.T) {
 			expectPDPeerServiceFn:      nil,
 			expectStatefulSetFn: func(g *GomegaWithT, set *apps.StatefulSet, err error) {
 				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(int(*set.Spec.Replicas)).To(Equal(3))
+				// g.Expect(int(*set.Spec.Replicas)).To(Equal(3))
 			},
 			expectTidbClusterFn: func(g *GomegaWithT, tc *v1alpha1.TidbCluster) {
 				g.Expect(tc.Status.PD.Members).To(BeNil())

diff --git a/pkg/manager/member/pd_scaler.go b/pkg/manager/member/pd_scaler.go
@@ -38,6 +38,8 @@ func NewPDScaler(pdControl controller.PDControlInterface,
 }
 
 func (psd *pdScaler) ScaleOut(tc *v1alpha1.TidbCluster, oldSet *apps.StatefulSet, newSet *apps.StatefulSet) error {
+	ns := tc.GetNamespace()
+	tcName := tc.GetName()
 	if tc.PDUpgrading() {
 		resetReplicas(newSet, oldSet)
 		return nil
@@ -49,6 +51,15 @@ func (psd *pdScaler) ScaleOut(tc *v1alpha1.TidbCluster, oldSet *apps.StatefulSet
 		return err
 	}
 
+	var i int32 = 0
+	for ; i<*oldSet.Spec.Replicas; i++ {
+		podName := ordinalPodName(v1alpha1.PDMemberType, tcName, i)
+		if member, ok := tc.Status.PD.Members[podName]; !ok || !member.Health {
+			resetReplicas(newSet, oldSet)
+			return fmt.Errorf("%s/%s is not ready, can't scale out now", ns, podName)
+		}
+	}
+
 	increaseReplicas(newSet, oldSet)
 	return nil
 }

diff --git a/pkg/manager/member/pd_scaler_test.go b/pkg/manager/member/pd_scaler_test.go
@@ -35,6 +35,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 	g := NewGomegaWithT(t)
 	type testcase struct {
 		name         string
+		update func(cluster *v1alpha1.TidbCluster)
 		pdUpgrading  bool
 		hasPVC       bool
 		hasDeferAnn  bool
@@ -46,6 +47,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 	testFn := func(test *testcase, t *testing.T) {
 		t.Log(test.name)
 		tc := newTidbClusterForPD()
+		test.update(tc)
 
 		if test.pdUpgrading {
 			tc.Status.PD.Phase = v1alpha1.UpgradePhase
@@ -92,6 +94,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 	tests := []testcase{
 		{
 			name:         "normal",
+			update: normalPDMember,
 			pdUpgrading:  false,
 			hasPVC:       true,
 			hasDeferAnn:  false,
@@ -101,6 +104,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 		},
 		{
 			name:         "pd is upgrading",
+			update: normalPDMember,
 			pdUpgrading:  true,
 			hasPVC:       true,
 			hasDeferAnn:  false,
@@ -110,6 +114,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 		},
 		{
 			name:         "cache don't have pvc",
+			update: normalPDMember,
 			pdUpgrading:  false,
 			hasPVC:       false,
 			hasDeferAnn:  false,
@@ -119,13 +124,40 @@ func TestPDScalerScaleOut(t *testing.T) {
 		},
 		{
 			name:         "pvc annotations defer deletion is not nil, pvc delete failed",
+			update: normalPDMember,
 			pdUpgrading:  false,
 			hasPVC:       true,
 			hasDeferAnn:  true,
 			pvcDeleteErr: true,
 			err:          true,
 			changed:      false,
 		},
+		{
+			name:         "don't have members",
+			update: func(tc *v1alpha1.TidbCluster) {
+				tc.Status.PD.Members = nil
+			},
+			pdUpgrading:  false,
+			hasPVC:       true,
+			hasDeferAnn:  true,
+			pvcDeleteErr: false,
+			err:          true,
+			changed:      false,
+		},
+		{
+			name:         "pd member not health",
+			update: func(tc *v1alpha1.TidbCluster) {
+				tcName := tc.GetName()
+				member1 := tc.Status.PD.Members[ordinalPodName(v1alpha1.PDMemberType, tcName, 1)]
+				member1.Health= false
+			},
+			pdUpgrading:  false,
+			hasPVC:       true,
+			hasDeferAnn:  true,
+			pvcDeleteErr: false,
+			err:          true,
+			changed:      false,
+		},
 	}
 
 	for i := range tests {
@@ -280,3 +312,14 @@ func int32Pointer(num int) *int32 {
 	i := int32(num)
 	return &i
 }
+
+func normalPDMember(tc *v1alpha1.TidbCluster) {
+	tcName := tc.GetName()
+	tc.Status.PD.Members = map[string]v1alpha1.PDMember{
+		ordinalPodName(v1alpha1.PDMemberType, tcName, 0): v1alpha1.PDMember{Health: true},
+		ordinalPodName(v1alpha1.PDMemberType, tcName, 1): v1alpha1.PDMember{Health: true},
+		ordinalPodName(v1alpha1.PDMemberType, tcName, 2): v1alpha1.PDMember{Health: true},
+		ordinalPodName(v1alpha1.PDMemberType, tcName, 3): v1alpha1.PDMember{Health: true},
+		ordinalPodName(v1alpha1.PDMemberType, tcName, 4): v1alpha1.PDMember{Health: true},
+	}
+}
diff --git a/pkg/manager/member/scaler.go b/pkg/manager/member/scaler.go
@@ -99,5 +99,5 @@ func ordinalPVCName(memberType v1alpha1.MemberType, setName string, ordinal int3
 }
 
 func ordinalPodName(memberType v1alpha1.MemberType, tcName string, ordinal int32) string {
-	return fmt.Sprintf("%s-%s-%d", tcName, v1alpha1.TiKVMemberType, ordinal)
+	return fmt.Sprintf("%s-%s-%d", tcName, memberType, ordinal)
 }
diff --git a/tests/e2e/create.go b/tests/e2e/create.go
@@ -205,7 +205,7 @@ func tikvMemberRunning(tc *v1alpha1.TidbCluster) (bool, error) {
 
 	if *tikvSet.Spec.Replicas != tc.Spec.TiKV.Replicas {
 		logf("tikvSet.Spec.Replicas(%d) != tc.Spec.TiKV.Replicas(%d)",
-			tikvSet.Spec.Replicas, tc.Spec.TiKV.Replicas)
+			*tikvSet.Spec.Replicas, tc.Spec.TiKV.Replicas)
 		return false, nil
 	}
 
@@ -260,7 +260,7 @@ func tidbMemberRunning(tc *v1alpha1.TidbCluster) (bool, error) {
 
 	if *tidbSet.Spec.Replicas != tc.Spec.TiDB.Replicas {
 		logf("tidbSet.Spec.Replicas(%d) != tc.Spec.TiDB.Replicas(%d)",
-			tidbSet.Spec.Replicas, tc.Spec.TiDB.Replicas)
+			*tidbSet.Spec.Replicas, tc.Spec.TiDB.Replicas)
 		return false, nil
 	}
 

diff --git a/tests/e2e/utils.go b/tests/e2e/utils.go
@@ -44,14 +44,7 @@ var (
 )
 
 func clearOperator() error {
-	_, err := execCmd(fmt.Sprintf(`kubectl get pv -l %s=%s,%s=%s --output=name | xargs -I {} \
-		kubectl patch {} -p '{"spec":{"persistentVolumeReclaimPolicy":"Delete"}}'`,
-		label.NamespaceLabelKey, ns, label.ClusterLabelKey, clusterName))
-	if err != nil {
-		logf(err.Error())
-	}
-
-	_, err = execCmd(fmt.Sprintf("helm del --purge %s", helmName))
+	_, err := execCmd(fmt.Sprintf("helm del --purge %s", helmName))
 	if err != nil && isNotFound(err) {
 		return err
 	}
@@ -71,6 +64,12 @@ func clearOperator() error {
 		if err != nil || result != "" {
 			return false, nil
 		}
+		_, err = execCmd(fmt.Sprintf(`kubectl get pv -l %s=%s,%s=%s --output=name | xargs -I {} \
+		kubectl patch {} -p '{"spec":{"persistentVolumeReclaimPolicy":"Delete"}}'`,
+			label.NamespaceLabelKey, ns, label.ClusterLabelKey, clusterName))
+		if err != nil {
+			logf(err.Error())
+		}
 		result, _ = execCmd(fmt.Sprintf("kubectl get pv -l %s=%s,%s=%s 2>/dev/null|grep Released", label.NamespaceLabelKey, ns, label.ClusterLabelKey, clusterName))
 		if result != "" {
 			return false, nil