TiKV upgrade bug fix

charts/tidb-operator/templates/controller-manager-deployment.yaml

@@ -40,6 +40,9 @@ spec:
           - -tikv-failover-period={{ .Values.controllerManager.tikvFailoverPeriod | default "5m" }}
           - -tidb-failover-period={{ .Values.controllerManager.tidbFailoverPeriod | default "5m" }}
           - -v={{ .Values.controllerManager.logLevel }}
+          {{- if .Values.testMode }}
+          - -test-mode={{ .Values.testMode }}
+          {{- end}}
         env:
           - name: NAMESPACE
             valueFrom:

cmd/controller-manager/main.go

@@ -62,6 +62,7 @@ func init() {
 	flag.DurationVar(&pdFailoverPeriod, "pd-failover-period", time.Duration(5*time.Minute), "PD failover period default(5m)")
 	flag.DurationVar(&tikvFailoverPeriod, "tikv-failover-period", time.Duration(5*time.Minute), "TiKV failover period default(5m)")
 	flag.DurationVar(&tidbFailoverPeriod, "tidb-failover-period", time.Duration(5*time.Minute), "TiDB failover period")
+	flag.BoolVar(&controller.TestMode, "test-mode", false, "whether tidb-operator run in test mode")
 
 	flag.Parse()
 }

pkg/controller/controller_utils.go

@@ -31,6 +31,8 @@ var (
 	DefaultStorageClassName string
 	// ClusterScoped controls whether operator should manage kubernetes cluster wide TiDB clusters
 	ClusterScoped bool
+	// TestMode defines whether tidb operator run in test mode, test mode is only open when test
+	TestMode bool
 )
 
 const (

pkg/manager/member/tikv_upgrader.go

@@ -122,6 +122,10 @@ func (tku *tikvUpgrader) upgradeTiKVPod(tc *v1alpha1.TidbCluster, ordinal int32,
 				return err
 			}
 			_, evicting := upgradePod.Annotations[EvictLeaderBeginTime]
+			if !evicting {
+				glog.Infof("start to evict leader:index:%d,upgradePodName:%s,storeID:%d", ordinal, upgradePodName, storeID)
+				return tku.beginEvictLeader(tc, storeID, upgradePod)
+			}
 
 			if tku.readyToUpgrade(upgradePod, store) {
 				err := tku.endEvictLeader(tc, ordinal)
@@ -132,9 +136,6 @@ func (tku *tikvUpgrader) upgradeTiKVPod(tc *v1alpha1.TidbCluster, ordinal int32,
 				return nil
 			}
 
-			if !evicting {
-				return tku.beginEvictLeader(tc, storeID, upgradePod)
-			}
 			return controller.RequeueErrorf("tidbcluster: [%s/%s]'s tikv pod: [%s] is evicting leader", ns, tcName, upgradePodName)
 		}
 	}
@@ -173,29 +174,23 @@ func (tku *tikvUpgrader) beginEvictLeader(tc *v1alpha1.TidbCluster, storeID uint
 }
 
 func (tku *tikvUpgrader) endEvictLeader(tc *v1alpha1.TidbCluster, ordinal int32) error {
+	// wait 5 second before delete evict scheduler，it is for auto test can catch these info
+	if controller.TestMode {
+		time.Sleep(5 * time.Second)
+	}
 	store := tku.getStoreByOrdinal(tc, ordinal)
 	storeID, err := strconv.ParseUint(store.ID, 10, 64)
 	if err != nil {
 		return err
 	}
-	upgradedPodName := tikvPodName(tc.GetName(), ordinal)
-	upgradedPod, err := tku.podLister.Pods(tc.GetNamespace()).Get(upgradedPodName)
-	if err != nil {
-		return err
-	}
-	_, evicting := upgradedPod.Annotations[EvictLeaderBeginTime]
-	if evicting {
-		delete(upgradedPod.Annotations, EvictLeaderBeginTime)
-		_, err = tku.podControl.UpdatePod(tc, upgradedPod)
-		if err != nil {
-			return err
-		}
-	}
-	err = controller.GetPDClient(tku.pdControl, tc).EndEvictLeader(storeID)
+
+	err = tku.pdControl.GetPDClient(pdapi.Namespace(tc.GetNamespace()), tc.GetName()).EndEvictLeader(storeID)
 	if err != nil {
 		return err
 	}
 
+	glog.Infof("successed to remove evict leader,ordinal:%d,storeID:%d", ordinal, storeID)
+
 	return nil
 }
 

pkg/manager/member/tikv_upgrader_test.go

@@ -128,7 +128,13 @@ func TestTiKVUpgraderUpgrade(t *testing.T) {
 			changeOldSet: func(oldSet *apps.StatefulSet) {
 				SetLastAppliedConfigAnnotation(oldSet)
 			},
-			changePods:          nil,
+			changePods: func(pods []*corev1.Pod) {
+				for _, pod := range pods {
+					if pod.GetName() == tikvPodName(upgradeTcName, 2) {
+						pod.Annotations = map[string]string{EvictLeaderBeginTime: time.Now().Add(-1 * time.Minute).Format(time.RFC3339)}
+					}
+				}
+			},
 			beginEvictLeaderErr: false,
 			endEvictLeaderErr:   false,
 			updatePodErr:        false,
@@ -138,10 +144,6 @@ func TestTiKVUpgraderUpgrade(t *testing.T) {
 			expectFn: func(g *GomegaWithT, tc *v1alpha1.TidbCluster, newSet *apps.StatefulSet, pods map[string]*corev1.Pod) {
 				g.Expect(tc.Status.TiKV.Phase).To(Equal(v1alpha1.UpgradePhase))
 				g.Expect(*newSet.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(int32(2)))
-				if pods[tikvPodName(upgradeTcName, 2)].Annotations != nil {
-					_, exist := pods[tikvPodName(upgradeTcName, 2)].Annotations[EvictLeaderBeginTime]
-					g.Expect(exist).To(BeFalse())
-				}
 			},
 		},
 		{
@@ -163,7 +165,13 @@ func TestTiKVUpgraderUpgrade(t *testing.T) {
 				oldSet.Status.UpdatedReplicas = 1
 				oldSet.Spec.UpdateStrategy.RollingUpdate.Partition = func() *int32 { i := int32(2); return &i }()
 			},
-			changePods:          nil,
+			changePods: func(pods []*corev1.Pod) {
+				for _, pod := range pods {
+					if pod.GetName() == tikvPodName(upgradeTcName, 1) {
+						pod.Annotations = map[string]string{EvictLeaderBeginTime: time.Now().Add(-1 * time.Minute).Format(time.RFC3339)}
+					}
+				}
+			},
 			beginEvictLeaderErr: false,
 			endEvictLeaderErr:   false,
 			updatePodErr:        false,
@@ -172,10 +180,6 @@ func TestTiKVUpgraderUpgrade(t *testing.T) {
 			},
 			expectFn: func(g *GomegaWithT, tc *v1alpha1.TidbCluster, newSet *apps.StatefulSet, pods map[string]*corev1.Pod) {
 				g.Expect(*newSet.Spec.UpdateStrategy.RollingUpdate.Partition).To(Equal(int32(1)))
-				if pods[tikvPodName(upgradeTcName, 1)].Annotations != nil {
-					_, exist := pods[tikvPodName(upgradeTcName, 1)].Annotations[EvictLeaderBeginTime]
-					g.Expect(exist).To(BeFalse())
-				}
 			},
 		},
 		{

tests/actions.go

@@ -14,6 +14,7 @@
 package tests
 
 import (
+	"context"
 	"crypto/tls"
 	"database/sql"
 	"encoding/json"
@@ -128,6 +129,8 @@ type OperatorActions interface {
 	ScaleTidbClusterOrDie(info *TidbClusterConfig)
 	CheckScaleInSafely(info *TidbClusterConfig) error
 	CheckScaledCorrectly(info *TidbClusterConfig, podUIDsBeforeScale map[string]types.UID) error
+	CheckUpgradeOrDie(ctx context.Context, info *TidbClusterConfig)
+	CheckUpgrade(ctx context.Context, info *TidbClusterConfig) error
 	UpgradeTidbCluster(info *TidbClusterConfig) error
 	UpgradeTidbClusterOrDie(info *TidbClusterConfig)
 	DeployAdHocBackup(info *TidbClusterConfig) error
@@ -222,6 +225,7 @@ type OperatorConfig struct {
 	WebhookConfigName  string
 	Context            *apimachinery.CertContext
 	ImagePullPolicy    corev1.PullPolicy
+	TestMode           bool
 }
 
 type TidbClusterConfig struct {
@@ -345,6 +349,7 @@ func (oi *OperatorConfig) OperatorHelmSetString(m map[string]string) string {
 		"controllerManager.replicas":       "2",
 		"scheduler.replicas":               "2",
 		"imagePullPolicy":                  string(oi.ImagePullPolicy),
+		"testMode":                         strconv.FormatBool(oi.TestMode),
 	}
 	if oi.SchedulerTag != "" {
 		set["scheduler.kubeSchedulerImageTag"] = oi.SchedulerTag
@@ -672,7 +677,7 @@ func (oa *operatorActions) CheckTidbClusterStatus(info *TidbClusterConfig) error
 
 	ns := info.Namespace
 	tcName := info.ClusterName
-	if err := wait.Poll(oa.pollInterval, 30*time.Minute, func() (bool, error) {
+	if err := wait.Poll(oa.pollInterval, 120*time.Minute, func() (bool, error) {
 		var tc *v1alpha1.TidbCluster
 		var err error
 		if tc, err = oa.cli.PingcapV1alpha1().TidbClusters(ns).Get(tcName, metav1.GetOptions{}); err != nil {
@@ -737,7 +742,7 @@ func (oa *operatorActions) CheckTidbClusterStatus(info *TidbClusterConfig) error
 		return true, nil
 	}); err != nil {
 		glog.Errorf("check tidb cluster status failed: %s", err.Error())
-		return fmt.Errorf("failed to waiting for tidbcluster %s/%s ready in 30 minutes", ns, tcName)
+		return fmt.Errorf("failed to waiting for tidbcluster %s/%s ready in 120 minutes", ns, tcName)
 	}
 
 	return nil
@@ -891,11 +896,6 @@ func (oa *operatorActions) SetPartitionAnnotation(tcName string, nameSpace strin
 }
 
 func (oa *operatorActions) UpgradeTidbCluster(info *TidbClusterConfig) error {
-	// record tikv leader count in webhook first
-	err := webhook.GetAllKVLeaders(oa.cli, info.Namespace, info.ClusterName)
-	if err != nil {
-		return err
-	}
 	oa.EmitEvent(info, "UpgradeTidbCluster")
 
 	cmd := oa.getHelmUpgradeClusterCmd(info, nil)
@@ -913,6 +913,83 @@ func (oa *operatorActions) UpgradeTidbClusterOrDie(info *TidbClusterConfig) {
 	}
 }
 
+func (oa *operatorActions) CheckUpgrade(ctx context.Context, info *TidbClusterConfig) error {
+	ns := info.Namespace
+	tcName := info.ClusterName
+
+	findStoreFn := func(tc *v1alpha1.TidbCluster, podName string) string {
+		for storeID, store := range tc.Status.TiKV.Stores {
+			if store.PodName == podName {
+				return storeID
+			}
+		}
+
+		return ""
+	}
+
+	for {
+		tc, err := oa.cli.PingcapV1alpha1().TidbClusters(ns).Get(tcName, metav1.GetOptions{})
+		if err != nil {
+			glog.Errorf("failed to get tidbcluster: %s/%s, %v", ns, tcName, err)
+			continue
+		}
+		pdClient := pdapi.NewDefaultPDControl().GetPDClient(pdapi.Namespace(tc.GetNamespace()), tc.GetName())
+
+		replicas := tc.TiKVRealReplicas()
+		for i := replicas - 1; i >= 0; i-- {
+			if err := wait.PollImmediate(1*time.Second, 10*time.Minute, func() (done bool, err error) {
+				schedulers, err := pdClient.GetEvictLeaderSchedulers()
+				if err != nil {
+					glog.Errorf("failed to get evict leader schedulers, %v", err)
+					return false, nil
+				}
+				glog.V(4).Infof("index:%d,schedulers:%v,error:%v", i, schedulers, err)
+				if len(schedulers) > 1 {
+					return true, fmt.Errorf("there are too many evict leader schedulers: %v", schedulers)
+				}
+				if len(schedulers) == 0 {
+					return false, nil
+				}
+				podName := fmt.Sprintf("%s-tikv-%d", tcName, i)
+				scheduler := fmt.Sprintf("evict-leader-scheduler-%s", findStoreFn(tc, podName))
+				if schedulers[0] == scheduler {
+					glog.Infof("index: %d,the schedulers: %s = %s", i, schedulers[0], scheduler)
+					return true, nil
+				}
+				glog.Errorf("index: %d,the scheduler: %s != %s", i, schedulers[0], scheduler)
+				return false, nil
+			}); err != nil {
+				glog.Errorf("failed to check upgrade %s/%s, %v", ns, tcName, err)
+				return err
+			}
+		}
+		if err := wait.PollImmediate(1*time.Second, 6*time.Minute, func() (done bool, err error) {
+			schedulers, err := pdClient.GetEvictLeaderSchedulers()
+			if err != nil {
+				glog.Errorf("failed to get evict leader schedulers, %v", err)
+				return false, nil
+			}
+			if len(schedulers) == 0 {
+				return true, nil
+			}
+			glog.Errorf("schedulers: %v is not empty", schedulers)
+			return false, nil
+		}); err != nil {
+			glog.Errorf("failed to wait all schedulers deleted %s/%s, %v", ns, tcName, err)
+			return err
+		}
+		break
+	}
+
+	return nil
+}
+
+func (oa *operatorActions) CheckUpgradeOrDie(ctx context.Context, info *TidbClusterConfig) {
+	if err := oa.CheckUpgrade(ctx, info); err != nil {
+		slack.NotifyAndPanic(err)
+	}
+}
+
 func (oa *operatorActions) DeployMonitor(info *TidbClusterConfig) error { return nil }
 func (oa *operatorActions) CleanMonitor(info *TidbClusterConfig) error  { return nil }
 

tests/cmd/e2e/main.go

@@ -56,6 +56,7 @@ func main() {
 		WebhookSecretName:  "webhook-secret",
 		WebhookConfigName:  "webhook-config",
 		ImagePullPolicy:    v1.PullIfNotPresent,
+		TestMode:           true,
 	}
 
 	ns := os.Getenv("NAMESPACE")

tests/cmd/stability/main.go

@@ -14,6 +14,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	_ "net/http/pprof"
@@ -32,7 +33,7 @@ import (
 )
 
 var cfg *tests.Config
-var context *apimachinery.CertContext
+var certCtx *apimachinery.CertContext
 var upgradeVersions []string
 
 func main() {
@@ -46,11 +47,11 @@ func main() {
 	ns := os.Getenv("NAMESPACE")
 
 	var err error
-	context, err = apimachinery.SetupServerCert(ns, tests.WebhookServiceName)
+	certCtx, err = apimachinery.SetupServerCert(ns, tests.WebhookServiceName)
 	if err != nil {
 		panic(err)
 	}
-	go tests.StartValidatingAdmissionWebhookServerOrDie(context)
+	go tests.StartValidatingAdmissionWebhookServerOrDie(certCtx)
 
 	c := cron.New()
 	if err := c.AddFunc("0 0 10 * * *", func() {
@@ -158,17 +159,20 @@ func run() {
 		}
 
 		// upgrade
-		oa.RegisterWebHookAndServiceOrDie(context, ocfg)
+		oa.RegisterWebHookAndServiceOrDie(certCtx, ocfg)
+		ctx, cancel := context.WithCancel(context.Background())
 		for idx, cluster := range clusters {
 			assignedNodes := oa.GetTidbMemberAssignedNodesOrDie(cluster)
 			cluster.UpgradeAll(upgradeVersion)
 			oa.UpgradeTidbClusterOrDie(cluster)
+			oa.CheckUpgradeOrDie(ctx, cluster)
 			if idx == 0 {
 				oa.CheckManualPauseTiDBOrDie(cluster)
 			}
 			oa.CheckTidbClusterStatusOrDie(cluster)
 			oa.CheckTidbMemberAssignedNodesOrDie(cluster, assignedNodes)
 		}
+		cancel()
 
 		// configuration change
 		for _, cluster := range clusters {

tests/cmd/stability/stability.go

@@ -22,6 +22,7 @@ func newOperatorConfig() *tests.OperatorConfig {
 		WebhookSecretName:  "webhook-secret",
 		WebhookConfigName:  "webhook-config",
 		ImagePullPolicy:    v1.PullAlways,
+		TestMode:           true,
 	}
 }
 
@@ -70,6 +71,7 @@ func newTidbClusterConfig(ns, clusterName string) *tests.TidbClusterConfig {
 		TiKVGrpcConcurrency: 4,
 		TiDBTokenLimit:      1000,
 		PDLogLevel:          "info",
+		TopologyKey:         topologyKey,
 		SubValues:           tests.GetAffinityConfigOrDie(clusterName, ns, topologyKey, []string{topologyKey}),
 	}
 }

tests/config.go

@@ -78,7 +78,7 @@ func NewConfig() (*Config, error) {
 	flag.StringVar(&cfg.configFile, "config", "", "Config file")
 	flag.StringVar(&cfg.LogDir, "log-dir", "/logDir", "log directory")
 	flag.IntVar(&cfg.FaultTriggerPort, "fault-trigger-port", 23332, "the http port of fault trigger service")
-	flag.StringVar(&cfg.TidbVersions, "tidb-versions", "v3.0.0-rc.1,v3.0.0-rc.2", "tidb versions")
+	flag.StringVar(&cfg.TidbVersions, "tidb-versions", "v3.0.0-rc.1,v3.0.0-rc.2,v3.0.0", "tidb versions")
 	flag.StringVar(&cfg.OperatorTag, "operator-tag", "master", "operator tag used to choose charts")
 	flag.StringVar(&cfg.OperatorImage, "operator-image", "pingcap/tidb-operator:latest", "operator image")
 	flag.StringVar(&cfg.UpgradeOperatorTag, "upgrade-operator-tag", "", "upgrade operator tag used to choose charts")

tests/dt.go

@@ -181,6 +181,10 @@ func (oa *operatorActions) CheckDataRegionDisasterTolerance(cluster *TidbCluster
 		// regionRacks is map of rackName and the peerID
 		regionRacks := map[string]uint64{}
 		for _, peer := range region.Peers {
+			if len(region.Peers) != 3 {
+				glog.Infof("cluster[%s] region[%d]'s peers not equal 3,[%v]. May be the failover happened", cluster.ClusterName, region.ID, region.Peers)
+				continue
+			}
 			storeID := strconv.FormatUint(peer.StoreId, 10)
 			nodeName, err := oa.getNodeByStoreId(storeID, cluster)
 			if err != nil {
@@ -189,7 +193,7 @@ func (oa *operatorActions) CheckDataRegionDisasterTolerance(cluster *TidbCluster
 			rackName := rackNodeMap[nodeName]
 			// if the rack have more than one peer of the region, return error
 			if otherID, exist := regionRacks[rackName]; exist {
-				return fmt.Errorf("region[%d]'s peer: [%d]and[%d] are in same rack:[%s]", region.ID, otherID, peer.Id, rackName)
+				return fmt.Errorf("cluster[%s] region[%d]'s peer: [%d]and[%d] are in same rack:[%s]", cluster.ClusterName, region.ID, otherID, peer.Id, rackName)
 			}
 			// add a new pair of rack and peer
 			regionRacks[rackName] = peer.Id