Repository: pingcap/pd
Pull request #2212 (2020-03-09 09:45:23)

schedulers: add sanity check for random-merge
### What problem does this PR solve? <!--add the issue link with summary if it exists-->
Fix https://github.com/pingcap/pd/issues/2210

### What is changed and how it works?
- port merge checks from mergeChecker to random-merge-scheduler
- minor refactor: remove `mergeChecker.ruleManager` and get it from `cluster`

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->
 - Unit test

Related changes
 - Need to cherry-pick to the release branch


disksing: /merge
sre-bot: Your auto merge job has been accepted, waiting for 2214
sre-bot: /run-all-tests

Patch:
diff --git a/pkg/mock/mockcluster/mockcluster.go b/pkg/mock/mockcluster/mockcluster.go
index 2c84e7cd8..08739da94 100644
--- a/pkg/mock/mockcluster/mockcluster.go
+++ b/pkg/mock/mockcluster/mockcluster.go
@@ -130,6 +130,11 @@ func (mc *Cluster) FitRegion(region *core.RegionInfo) *placement.RegionFit {
 	return mc.RuleManager.FitRegion(mc.BasicCluster, region)
 }
 
+// GetRuleManager returns the ruleManager of the cluster.
+func (mc *Cluster) GetRuleManager() *placement.RuleManager {
+	return mc.RuleManager
+}
+
 // SetStoreUp sets store state to be up.
 func (mc *Cluster) SetStoreUp(storeID uint64) {
 	store := mc.GetStore(storeID)
diff --git a/server/schedule/checker/merge_checker.go b/server/schedule/checker/merge_checker.go
index f059db69b..d1f04b035 100644
--- a/server/schedule/checker/merge_checker.go
+++ b/server/schedule/checker/merge_checker.go
@@ -30,20 +30,18 @@ import (
 
 // MergeChecker ensures region to merge with adjacent region when size is small
 type MergeChecker struct {
-	cluster     opt.Cluster
-	splitCache  *cache.TTLUint64
-	ruleManager *placement.RuleManager
-	startTime   time.Time // it's used to judge whether server recently start.
+	cluster    opt.Cluster
+	splitCache *cache.TTLUint64
+	startTime  time.Time // it's used to judge whether server recently start.
 }
 
 // NewMergeChecker creates a merge checker.
-func NewMergeChecker(ctx context.Context, cluster opt.Cluster, ruleManager *placement.RuleManager) *MergeChecker {
+func NewMergeChecker(ctx context.Context, cluster opt.Cluster) *MergeChecker {
 	splitCache := cache.NewIDTTL(ctx, time.Minute, cluster.GetSplitMergeInterval())
 	return &MergeChecker{
-		cluster:     cluster,
-		splitCache:  splitCache,
-		ruleManager: ruleManager,
-		startTime:   time.Now(),
+		cluster:    cluster,
+		splitCache: splitCache,
+		startTime:  time.Now(),
 	}
 }
 
@@ -135,12 +133,12 @@ func (m *MergeChecker) Check(region *core.RegionInfo) []*operator.Operator {
 }
 
 func (m *MergeChecker) checkTarget(region, adjacent *core.RegionInfo) bool {
-	return adjacent != nil && !m.cluster.IsRegionHot(adjacent) && m.allowMerge(region, adjacent) &&
+	return adjacent != nil && !m.cluster.IsRegionHot(adjacent) && AllowMerge(m.cluster, region, adjacent) &&
 		opt.IsRegionHealthy(m.cluster, adjacent) && opt.IsRegionReplicated(m.cluster, adjacent)
 }
 
-// allowMerge returns true if two regions can be merged according to the key type.
-func (m *MergeChecker) allowMerge(region *core.RegionInfo, adjacent *core.RegionInfo) bool {
+// AllowMerge returns true if two regions can be merged according to the key type.
+func AllowMerge(cluster opt.Cluster, region *core.RegionInfo, adjacent *core.RegionInfo) bool {
 	var start, end []byte
 	if bytes.Equal(region.GetEndKey(), adjacent.GetStartKey()) && len(region.GetEndKey()) != 0 {
 		start, end = region.GetStartKey(), adjacent.GetEndKey()
@@ -149,13 +147,19 @@ func (m *MergeChecker) allowMerge(region *core.RegionInfo, adjacent *core.Region
 	} else {
 		return false
 	}
-	if m.cluster.IsPlacementRulesEnabled() && len(m.ruleManager.GetSplitKeys(start, end)) > 0 {
-		return false
+	if cluster.IsPlacementRulesEnabled() {
+		type withRuleManager interface {
+			GetRuleManager() *placement.RuleManager
+		}
+		cl, ok := cluster.(withRuleManager)
+		if !ok || len(cl.GetRuleManager().GetSplitKeys(start, end)) > 0 {
+			return false
+		}
 	}
-	policy := m.cluster.GetKeyType()
+	policy := cluster.GetKeyType()
 	switch policy {
 	case core.Table:
-		if m.cluster.IsCrossTableMergeEnabled() {
+		if cluster.IsCrossTableMergeEnabled() {
 			return true
 		}
 		return isTableIDSame(region, adjacent)
diff --git a/server/schedule/checker_controller.go b/server/schedule/checker_controller.go
index 2a3484817..b237487bf 100644
--- a/server/schedule/checker_controller.go
+++ b/server/schedule/checker_controller.go
@@ -42,7 +42,7 @@ func NewCheckerController(ctx context.Context, cluster opt.Cluster, ruleManager
 		learnerChecker: checker.NewLearnerChecker(cluster),
 		replicaChecker: checker.NewReplicaChecker(cluster),
 		ruleChecker:    checker.NewRuleChecker(cluster, ruleManager),
-		mergeChecker:   checker.NewMergeChecker(ctx, cluster, ruleManager),
+		mergeChecker:   checker.NewMergeChecker(ctx, cluster),
 	}
 }
 
diff --git a/server/schedulers/random_merge.go b/server/schedulers/random_merge.go
index e4b67b3fb..b73bf51a1 100644
--- a/server/schedulers/random_merge.go
+++ b/server/schedulers/random_merge.go
@@ -19,6 +19,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/pd/v4/server/core"
 	"github.com/pingcap/pd/v4/server/schedule"
+	"github.com/pingcap/pd/v4/server/schedule/checker"
 	"github.com/pingcap/pd/v4/server/schedule/filter"
 	"github.com/pingcap/pd/v4/server/schedule/operator"
 	"github.com/pingcap/pd/v4/server/schedule/opt"
@@ -124,6 +125,11 @@ func (s *randomMergeScheduler) Schedule(cluster opt.Cluster) []*operator.Operato
 		return nil
 	}
 
+	if !s.allowMerge(cluster, region, target) {
+		schedulerCounter.WithLabelValues(s.GetName(), "not-allowed").Inc()
+		return nil
+	}
+
 	ops, err := operator.CreateMergeRegionOperator(RandomMergeType, cluster, region, target, operator.OpAdmin)
 	if err != nil {
 		log.Debug("fail to create merge region operator", zap.Error(err))
@@ -132,3 +138,16 @@ func (s *randomMergeScheduler) Schedule(cluster opt.Cluster) []*operator.Operato
 	ops[0].Counters = append(ops[0].Counters, schedulerCounter.WithLabelValues(s.GetName(), "new-operator"))
 	return ops
 }
+
+func (s *randomMergeScheduler) allowMerge(cluster opt.Cluster, region, target *core.RegionInfo) bool {
+	if !opt.IsRegionHealthy(cluster, region) || !opt.IsRegionHealthy(cluster, target) {
+		return false
+	}
+	if !opt.IsRegionReplicated(cluster, region) || !opt.IsRegionReplicated(cluster, target) {
+		return false
+	}
+	if cluster.IsRegionHot(region) || cluster.IsRegionHot(target) {
+		return false
+	}
+	return checker.AllowMerge(cluster, region, target)
+}


Test patch:
diff --git a/server/schedule/checker/merge_checker_test.go b/server/schedule/checker/merge_checker_test.go
index 5689f5eac..6213b5aa4 100644
--- a/server/schedule/checker/merge_checker_test.go
+++ b/server/schedule/checker/merge_checker_test.go
@@ -129,7 +129,7 @@ func (s *testMergeCheckerSuite) SetUpTest(c *C) {
 		s.cluster.PutRegion(region)
 	}
 	s.ctx, s.cancel = context.WithCancel(context.Background())
-	s.mc = NewMergeChecker(s.ctx, s.cluster, s.cluster.RuleManager)
+	s.mc = NewMergeChecker(s.ctx, s.cluster)
 }
 
 func (s *testMergeCheckerSuite) TearDownTest(c *C) {
@@ -168,7 +168,7 @@ func (s *testMergeCheckerSuite) TestBasic(c *C) {
 
 	// merge cannot across rule key.
 	s.cluster.EnablePlacementRules = true
-	s.mc.ruleManager.SetRule(&placement.Rule{
+	s.cluster.RuleManager.SetRule(&placement.Rule{
 		GroupID:     "pd",
 		ID:          "test",
 		Index:       1,
@@ -183,7 +183,7 @@ func (s *testMergeCheckerSuite) TestBasic(c *C) {
 	c.Assert(ops, NotNil)
 	c.Assert(ops[0].RegionID(), Equals, s.regions[2].GetID())
 	c.Assert(ops[1].RegionID(), Equals, s.regions[1].GetID())
-	s.mc.ruleManager.DeleteRule("test", "test")
+	s.cluster.RuleManager.DeleteRule("test", "test")
 
 	// Skip recently split regions.
 	s.cluster.ScheduleOptions.SplitMergeInterval = time.Hour
@@ -463,7 +463,7 @@ func (s *testMergeCheckerSuite) TestCache(c *C) {
 		s.cluster.PutRegion(region)
 	}
 
-	s.mc = NewMergeChecker(s.ctx, s.cluster, s.cluster.RuleManager)
+	s.mc = NewMergeChecker(s.ctx, s.cluster)
 
 	ops := s.mc.Check(s.regions[1])
 	c.Assert(ops, IsNil)
diff --git a/server/schedule/operator_controller_test.go b/server/schedule/operator_controller_test.go
index 26bf09e16..c56ad54a0 100644
--- a/server/schedule/operator_controller_test.go
+++ b/server/schedule/operator_controller_test.go
@@ -522,7 +522,7 @@ func (t *testOperatorControllerSuite) TestStoreLimitWithMerge(c *C) {
 		tc.PutRegion(region)
 	}
 
-	mc := checker.NewMergeChecker(t.ctx, tc, tc.RuleManager)
+	mc := checker.NewMergeChecker(t.ctx, tc)
 	oc := NewOperatorController(t.ctx, tc, mockhbstream.NewHeartbeatStream())
 
 	cfg.StoreBalanceRate = 60
diff --git a/server/schedulers/balance_test.go b/server/schedulers/balance_test.go
index 4aeacff3a..a9290269b 100644
--- a/server/schedulers/balance_test.go
+++ b/server/schedulers/balance_test.go
@@ -1152,6 +1152,10 @@ func (s *testRandomMergeSchedulerSuite) TestMerge(c *C) {
 
 	c.Assert(mb.IsScheduleAllowed(tc), IsTrue)
 	ops := mb.Schedule(tc)
+	c.Assert(ops, HasLen, 0) // regions are not fully replicated
+
+	opt.MaxReplicas = 1
+	ops = mb.Schedule(tc)
 	c.Assert(ops, HasLen, 2)
 	c.Assert(ops[0].Kind()&operator.OpMerge, Not(Equals), 0)
 	c.Assert(ops[1].Kind()&operator.OpMerge, Not(Equals), 0)