Fix tikv scale in failure in some cases (#726)

cmd/controller-manager/main.go

@@ -48,7 +48,6 @@ var (
 	leaseDuration      = 15 * time.Second
 	renewDuration      = 5 * time.Second
 	retryPeriod        = 3 * time.Second
-	resyncDuration     = 30 * time.Second
 	waitDuration       = 5 * time.Second
 )
 
@@ -62,6 +61,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.DurationVar(&controller.ResyncDuration, "resync-duration", time.Duration(30*time.Second), "Resync time of informer")
 	flag.BoolVar(&controller.TestMode, "test-mode", false, "whether tidb-operator run in test mode")
 
 	flag.Parse()
@@ -104,18 +104,18 @@ func main() {
 	var informerFactory informers.SharedInformerFactory
 	var kubeInformerFactory kubeinformers.SharedInformerFactory
 	if controller.ClusterScoped {
-		informerFactory = informers.NewSharedInformerFactory(cli, resyncDuration)
-		kubeInformerFactory = kubeinformers.NewSharedInformerFactory(kubeCli, resyncDuration)
+		informerFactory = informers.NewSharedInformerFactory(cli, controller.ResyncDuration)
+		kubeInformerFactory = kubeinformers.NewSharedInformerFactory(kubeCli, controller.ResyncDuration)
 	} else {
 		options := []informers.SharedInformerOption{
 			informers.WithNamespace(ns),
 		}
-		informerFactory = informers.NewSharedInformerFactoryWithOptions(cli, resyncDuration, options...)
+		informerFactory = informers.NewSharedInformerFactoryWithOptions(cli, controller.ResyncDuration, options...)
 
 		kubeoptions := []kubeinformers.SharedInformerOption{
 			kubeinformers.WithNamespace(ns),
 		}
-		kubeInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeCli, resyncDuration, kubeoptions...)
+		kubeInformerFactory = kubeinformers.NewSharedInformerFactoryWithOptions(kubeCli, controller.ResyncDuration, kubeoptions...)
 	}
 
 	rl := resourcelock.EndpointsLock{

pkg/controller/controller_utils.go

@@ -16,6 +16,7 @@ package controller
 import (
 	"fmt"
 	"math"
+	"time"
 
 	"github.com/golang/glog"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1alpha1"
@@ -33,6 +34,8 @@ var (
 	ClusterScoped bool
 	// TestMode defines whether tidb operator run in test mode, test mode is only open when test
 	TestMode bool
+	// ResyncDuration is the resync time of informer
+	ResyncDuration time.Duration
 )
 
 const (

pkg/controller/pvc_control.go

@@ -216,7 +216,7 @@ func (fpc *FakePVCControl) DeletePVC(_ *v1alpha1.TidbCluster, pvc *corev1.Persis
 	return fpc.PVCIndexer.Delete(pvc)
 }
 
-// Update updates the annotation, labels and spec of pvc
+// UpdatePVC updates the annotation, labels and spec of pvc
 func (fpc *FakePVCControl) UpdatePVC(_ *v1alpha1.TidbCluster, pvc *corev1.PersistentVolumeClaim) (*corev1.PersistentVolumeClaim, error) {
 	defer fpc.updatePVCTracker.inc()
 	if fpc.updatePVCTracker.errorReady() {

pkg/manager/member/pd_scaler_test.go

@@ -40,6 +40,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 		pdUpgrading      bool
 		hasPVC           bool
 		hasDeferAnn      bool
+		annoIsNil        bool
 		pvcDeleteErr     bool
 		statusSyncFailed bool
 		err              bool
@@ -61,19 +62,18 @@ func TestPDScalerScaleOut(t *testing.T) {
 
 		scaler, _, pvcIndexer, pvcControl := newFakePDScaler()
 
-		pvc1 := newPVCForStatefulSet(oldSet, v1alpha1.PDMemberType)
-		pvc2 := pvc1.DeepCopy()
-		pvc1.Name = ordinalPVCName(v1alpha1.PDMemberType, oldSet.GetName(), *oldSet.Spec.Replicas)
-		pvc2.Name = ordinalPVCName(v1alpha1.PDMemberType, oldSet.GetName(), *oldSet.Spec.Replicas+1)
+		pvc := newPVCForStatefulSet(oldSet, v1alpha1.PDMemberType)
+		pvc.Name = ordinalPVCName(v1alpha1.PDMemberType, oldSet.GetName(), *oldSet.Spec.Replicas)
+		if !test.annoIsNil {
+			pvc.Annotations = map[string]string{}
+		}
+
 		if test.hasDeferAnn {
-			pvc1.Annotations = map[string]string{}
-			pvc1.Annotations[label.AnnPVCDeferDeleting] = time.Now().Format(time.RFC3339)
-			pvc2.Annotations = map[string]string{}
-			pvc2.Annotations[label.AnnPVCDeferDeleting] = time.Now().Format(time.RFC3339)
+			pvc.Annotations = map[string]string{}
+			pvc.Annotations[label.AnnPVCDeferDeleting] = time.Now().Format(time.RFC3339)
 		}
 		if test.hasPVC {
-			pvcIndexer.Add(pvc1)
-			pvcIndexer.Add(pvc2)
+			pvcIndexer.Add(pvc)
 		}
 
 		if test.pvcDeleteErr {
@@ -102,6 +102,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      false,
+			annoIsNil:        true,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              false,
@@ -113,6 +114,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      true,
 			hasPVC:           true,
 			hasDeferAnn:      false,
+			annoIsNil:        true,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              false,
@@ -124,6 +126,19 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           false,
 			hasDeferAnn:      false,
+			annoIsNil:        true,
+			pvcDeleteErr:     false,
+			statusSyncFailed: false,
+			err:              false,
+			changed:          true,
+		},
+		{
+			name:             "pvc annotation is not nil but doesn't contain defer deletion annotation",
+			update:           normalPDMember,
+			pdUpgrading:      false,
+			hasPVC:           false,
+			hasDeferAnn:      false,
+			annoIsNil:        false,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              false,
@@ -135,6 +150,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      true,
+			annoIsNil:        false,
 			pvcDeleteErr:     true,
 			statusSyncFailed: false,
 			err:              true,
@@ -155,6 +171,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      true,
+			annoIsNil:        false,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              false,
@@ -168,6 +185,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      true,
+			annoIsNil:        false,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              true,
@@ -186,6 +204,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      false,
+			annoIsNil:        true,
 			pvcDeleteErr:     false,
 			statusSyncFailed: false,
 			err:              true,
@@ -197,6 +216,7 @@ func TestPDScalerScaleOut(t *testing.T) {
 			pdUpgrading:      false,
 			hasPVC:           true,
 			hasDeferAnn:      false,
+			annoIsNil:        true,
 			pvcDeleteErr:     false,
 			statusSyncFailed: true,
 			err:              true,

pkg/manager/member/tikv_scaler.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/pdapi"
 	apps "k8s.io/api/apps/v1beta1"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 )
 
 type tikvScaler struct {
@@ -78,6 +79,7 @@ func (tsd *tikvScaler) ScaleIn(tc *v1alpha1.TidbCluster, oldSet *apps.StatefulSe
 		resetReplicas(newSet, oldSet)
 		return err
 	}
+
 	for _, store := range tc.Status.TiKV.Stores {
 		if store.PodName == podName {
 			state := store.State
@@ -135,9 +137,48 @@ func (tsd *tikvScaler) ScaleIn(tc *v1alpha1.TidbCluster, oldSet *apps.StatefulSe
 		}
 	}
 
-	// store not found in TidbCluster status,
-	// this can happen when TiKV joins cluster but we haven't synced its status
-	// so return error to wait another round for safety
+	// When store not found in TidbCluster status, there are two situations as follows:
+	// 1. This can happen when TiKV joins cluster but we haven't synced its status.
+	//    In this situation return error to wait another round for safety.
+	//
+	// 2. This can happen when TiKV pod has not been successfully registered in the cluster, such as always pending.
+	//    In this situation we should delete this TiKV pod immediately to avoid blocking the subsequent operations.
+	if !podutil.IsPodReady(pod) {
+		pvcName := ordinalPVCName(v1alpha1.TiKVMemberType, setName, ordinal)
+		pvc, err := tsd.pvcLister.PersistentVolumeClaims(ns).Get(pvcName)
+		if err != nil {
+			resetReplicas(newSet, oldSet)
+			return err
+		}
+		safeTimeDeadline := pod.CreationTimestamp.Add(5 * controller.ResyncDuration)
+		if time.Now().Before(safeTimeDeadline) {
+			// Wait for 5 resync periods to ensure that the following situation does not occur:
+			//
+			// The tikv pod starts for a while, but has not synced its status, and then the pod becomes not ready.
+			// Here we wait for 5 resync periods to ensure that the status of this tikv pod has been synced.
+			// After this period of time, if there is still no information about this tikv in TidbCluster status,
+			// then we can be sure that this tikv has never been added to the tidb cluster.
+			// So we can scale in this tikv pod safely.
+			resetReplicas(newSet, oldSet)
+			return fmt.Errorf("TiKV %s/%s is not ready, wait for some resync periods to synced its status", ns, podName)
+		}
+		if pvc.Annotations == nil {
+			pvc.Annotations = map[string]string{}
+		}
+		now := time.Now().Format(time.RFC3339)
+		pvc.Annotations[label.AnnPVCDeferDeleting] = now
+		_, err = tsd.pvcControl.UpdatePVC(tc, pvc)
+		if err != nil {
+			glog.Errorf("pod %s not ready, tikv scale in: failed to set pvc %s/%s annotation: %s to %s",
+				podName, ns, pvcName, label.AnnPVCDeferDeleting, now)
+			resetReplicas(newSet, oldSet)
+			return err
+		}
+		glog.Infof("pod %s not ready, tikv scale in: set pvc %s/%s annotation: %s to %s",
+			podName, ns, pvcName, label.AnnPVCDeferDeleting, now)
+		decreaseReplicas(newSet, oldSet)
+		return nil
+	}
 	resetReplicas(newSet, oldSet)
 	return fmt.Errorf("TiKV %s/%s not found in cluster", ns, podName)
 }