Add a new RequeueError type

Gopkg.toml

@@ -69,3 +69,7 @@ required = [
 [[override]]
     name = "github.com/BurntSushi/toml"
     revision = "3012a1dbe2e4bd1391d42b32f0577cb7bbc7f005"
+
+[[constraint]]
+    name = "github.com/pingcap/errors"
+    revision = "f29524922bfee4bf03f9c426c716e639cac63ea9"

Makefile

@@ -40,6 +40,7 @@ e2e-build:
 	$(GOENV) ginkgo build tests/e2e
 
 test:
+	@echo "Run unit tests"
 	@$(GOTEST) ./pkg/... && echo "\nUnit tests run successfully!"
 
 check-all: lint check-static check-shadow check-gosec megacheck errcheck

pkg/controller/controller_utils.go

@@ -38,6 +38,26 @@ const (
 	defaultPushgatewayImage = "prom/pushgateway:v0.3.1"
 )
 
+// RequeueError is used to requeue the item, this error type should't be considered as a real error
+type RequeueError struct {
+	s string
+}
+
+func (re *RequeueError) Error() string {
+	return re.s
+}
+
+// RequeueErrorf returns a RequeueError
+func RequeueErrorf(format string, a ...interface{}) error {
+	return &RequeueError{fmt.Sprintf(format, a...)}
+}
+
+// IsRequeueError returns whether err is a RequeueError
+func IsRequeueError(err error) bool {
+	_, ok := err.(*RequeueError)
+	return ok
+}
+
 // GetOwnerRef returns TidbCluster's OwnerReference
 func GetOwnerRef(tc *v1alpha1.TidbCluster) metav1.OwnerReference {
 	controller := true

pkg/controller/tidbcluster/tidb_cluster_control.go

@@ -20,6 +20,7 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/manager"
 	"github.com/pingcap/tidb-operator/pkg/util"
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
+	errorutils "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/tools/record"
 )
 
@@ -69,22 +70,30 @@ type defaultTidbClusterControl struct {
 
 // UpdateStatefulSet executes the core logic loop for a tidbcluster.
 func (tcc *defaultTidbClusterControl) UpdateTidbCluster(tc *v1alpha1.TidbCluster) error {
+	// perform the main update function and get the status
+
 	oldStatus := tc.Status.DeepCopy()
 	oldPDReplicas := tc.Spec.PD.Replicas
+	oldTiKVReplicas := tc.Spec.TiKV.Replicas
+	oldTiDBReplicas := tc.Spec.TiDB.Replicas
 
+	var errs []error
 	err := tcc.updateTidbCluster(tc)
 	if err != nil {
-		return err
+		errs = append(errs, err)
 	}
 
-	if !apiequality.Semantic.DeepEqual(&tc.Status, oldStatus) || tc.Spec.PD.Replicas != oldPDReplicas {
-		tc, err = tcc.tcControl.UpdateTidbCluster(tc.DeepCopy())
+	replicasChanged := tc.Spec.PD.Replicas != oldPDReplicas ||
+		tc.Spec.TiKV.Replicas != oldTiKVReplicas ||
+		tc.Spec.TiDB.Replicas != oldTiDBReplicas
+	if !apiequality.Semantic.DeepEqual(&tc.Status, oldStatus) || replicasChanged {
+		_, err := tcc.tcControl.UpdateTidbCluster(tc.DeepCopy())
 		if err != nil {
-			return err
+			errs = append(errs, err)
 		}
 	}
 
-	return nil
+	return errorutils.NewAggregate(errs)
 }
 
 func (tcc *defaultTidbClusterControl) updateTidbCluster(tc *v1alpha1.TidbCluster) error {

pkg/controller/tidbcluster/tidb_cluster_controller.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	pingcaperrors "github.com/pingcap/errors"
 	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/client/clientset/versioned"
 	informers "github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
@@ -216,7 +217,13 @@ func (tcc *Controller) processNextWorkItem() bool {
 	}
 	defer tcc.queue.Done(key)
 	if err := tcc.sync(key.(string)); err != nil {
-		utilruntime.HandleError(fmt.Errorf("Error syncing TidbCluster %v, requeuing: %v", key.(string), err))
+		if reqErr := pingcaperrors.Find(err, func(e error) bool {
+			return controller.IsRequeueError(e)
+		}); reqErr != nil {
+			glog.Infof("TidbCluster: %v, still need sync: %v, requeuing", key.(string), err)
+		} else {
+			utilruntime.HandleError(fmt.Errorf("TidbCluster: %v, sync failed %v, requeuing", key.(string), err))
+		}
 		tcc.queue.AddRateLimited(key)
 	} else {
 		tcc.queue.Forget(key)

pkg/manager/member/failover_test.go

@@ -61,9 +61,8 @@ func TestPDFailoverFailover(t *testing.T) {
 		pdClient.AddReaction(controller.DeleteMemberActionType, func(action *controller.Action) (interface{}, error) {
 			if test.delMemberFailed {
 				return nil, fmt.Errorf("failed to delete member")
-			} else {
-				return nil, nil
 			}
+			return nil, nil
 		})
 
 		pvc := newPVCForPDFailover(tc, v1alpha1.PDMemberType, 1)

pkg/manager/member/tikv_scaler.go

@@ -86,7 +86,7 @@ func (tsd *tikvScaler) ScaleIn(tc *v1alpha1.TidbCluster, oldSet *apps.StatefulSe
 				}
 			}
 			resetReplicas(newSet, oldSet)
-			return fmt.Errorf("TiKV %s/%s store %d  still in cluster, state: %s", ns, podName, id, state)
+			return controller.RequeueErrorf("TiKV %s/%s store %d  still in cluster, state: %s", ns, podName, id, state)
 		}
 	}
 	for _, store := range tc.Status.TiKV.Stores.TombStoneStores {

pkg/manager/member/tikv_scaler_test.go

@@ -37,7 +37,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 		hasPVC        bool
 		hasDeferAnn   bool
 		pvcDeleteErr  bool
-		err           bool
+		errExpectFn   func(*GomegaWithT, error)
 		changed       bool
 	}
 
@@ -75,11 +75,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 		}
 
 		err := scaler.ScaleOut(tc, oldSet, newSet)
-		if test.err {
-			g.Expect(err).To(HaveOccurred())
-		} else {
-			g.Expect(err).NotTo(HaveOccurred())
-		}
+		test.errExpectFn(g, err)
 		if test.changed {
 			g.Expect(int(*newSet.Spec.Replicas)).To(Equal(6))
 		} else {
@@ -94,7 +90,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 			hasPVC:        true,
 			hasDeferAnn:   false,
 			pvcDeleteErr:  false,
-			err:           false,
+			errExpectFn:   errExpectNil,
 			changed:       true,
 		},
 		{
@@ -103,7 +99,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 			hasPVC:        true,
 			hasDeferAnn:   false,
 			pvcDeleteErr:  false,
-			err:           false,
+			errExpectFn:   errExpectNil,
 			changed:       false,
 		},
 		{
@@ -112,7 +108,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 			hasPVC:        false,
 			hasDeferAnn:   false,
 			pvcDeleteErr:  false,
-			err:           false,
+			errExpectFn:   errExpectNil,
 			changed:       true,
 		},
 		{
@@ -121,7 +117,7 @@ func TestTiKVScalerScaleOut(t *testing.T) {
 			hasPVC:        true,
 			hasDeferAnn:   true,
 			pvcDeleteErr:  true,
-			err:           true,
+			errExpectFn:   errExpectNotNil,
 			changed:       false,
 		},
 	}
@@ -140,7 +136,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 		delStoreErr   bool
 		hasPVC        bool
 		pvcUpdateErr  bool
-		err           bool
+		errExpectFn   func(*GomegaWithT, error)
 		changed       bool
 	}
 
@@ -177,11 +173,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 		}
 
 		err := scaler.ScaleIn(tc, oldSet, newSet)
-		if test.err {
-			g.Expect(err).To(HaveOccurred())
-		} else {
-			g.Expect(err).NotTo(HaveOccurred())
-		}
+		test.errExpectFn(g, err)
 		if test.changed {
 			g.Expect(int(*newSet.Spec.Replicas)).To(Equal(4))
 		} else {
@@ -197,7 +189,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        true,
 			pvcUpdateErr:  false,
-			err:           true,
+			errExpectFn:   errExpectNotNil,
 			changed:       false,
 		},
 		{
@@ -207,7 +199,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        true,
 			pvcUpdateErr:  false,
-			err:           false,
+			errExpectFn:   errExpectNil,
 			changed:       false,
 		},
 		{
@@ -220,7 +212,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:  false,
 			hasPVC:       true,
 			pvcUpdateErr: false,
-			err:          true,
+			errExpectFn:  errExpectNotNil,
 			changed:      false,
 		},
 		{
@@ -235,7 +227,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:  false,
 			hasPVC:       true,
 			pvcUpdateErr: false,
-			err:          true,
+			errExpectFn:  errExpectNotNil,
 			changed:      false,
 		},
 		{
@@ -250,7 +242,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:  false,
 			hasPVC:       true,
 			pvcUpdateErr: false,
-			err:          true,
+			errExpectFn:  errExpectNotNil,
 			changed:      false,
 		},
 		{
@@ -265,7 +257,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:  false,
 			hasPVC:       true,
 			pvcUpdateErr: false,
-			err:          true,
+			errExpectFn:  errExpectNotNil,
 			changed:      false,
 		},
 		{
@@ -275,7 +267,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        true,
 			pvcUpdateErr:  false,
-			err:           true,
+			errExpectFn:   errExpectRequeue,
 			changed:       false,
 		},
 		{
@@ -285,7 +277,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        true,
 			pvcUpdateErr:  false,
-			err:           false,
+			errExpectFn:   errExpectNil,
 			changed:       true,
 		},
 		{
@@ -300,7 +292,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:  false,
 			hasPVC:       true,
 			pvcUpdateErr: false,
-			err:          true,
+			errExpectFn:  errExpectNotNil,
 			changed:      false,
 		},
 		{
@@ -310,7 +302,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        false,
 			pvcUpdateErr:  false,
-			err:           true,
+			errExpectFn:   errExpectNotNil,
 			changed:       false,
 		},
 		{
@@ -320,7 +312,7 @@ func TestTiKVScalerScaleIn(t *testing.T) {
 			delStoreErr:   false,
 			hasPVC:        true,
 			pvcUpdateErr:  true,
-			err:           true,
+			errExpectFn:   errExpectNotNil,
 			changed:       false,
 		},
 	}
@@ -365,3 +357,7 @@ func tombstoneStoreFun(tc *v1alpha1.TidbCluster) {
 		},
 	}
 }
+
+func errExpectRequeue(g *GomegaWithT, err error) {
+	g.Expect(controller.IsRequeueError(err)).To(Equal(true))
+}