mcs: fix rule sync when meet "no rule left" and concurrency

pkg/mcs/scheduling/server/apis/v1/api.go

@@ -1330,5 +1330,5 @@ func checkRegionsReplicated(c *gin.Context) {
 		c.String(http.StatusBadRequest, err.Error())
 		return
 	}
-	c.String(http.StatusOK, state)
+	c.IndentedJSON(http.StatusOK, state)
 }

pkg/mcs/scheduling/server/rule/watcher.go

@@ -25,6 +25,7 @@
 	"github.com/tikv/pd/pkg/schedule/placement"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
+	"github.com/tikv/pd/pkg/utils/syncutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.etcd.io/etcd/mvcc/mvccpb"
 	"go.uber.org/zap"
@@ -36,6 +37,10 @@
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
 
+	// ruleCommonPathPrefix:
+	//  - Key: /pd/{cluster_id}/rule
+	//  - Value: placement.Rule or placement.RuleGroup
+	ruleCommonPathPrefix string
 	// rulesPathPrefix:
 	//   - Key: /pd/{cluster_id}/rules/{group_id}-{rule_id}
 	//   - Value: placement.Rule
@@ -60,8 +65,18 @@
 	regionLabeler *labeler.RegionLabeler
 
 	ruleWatcher  *etcdutil.LoopWatcher
-	groupWatcher *etcdutil.LoopWatcher
 	labelWatcher *etcdutil.LoopWatcher
+
+	// pendingDeletion is a structure used to track the rules or rule groups that are marked for deletion.
+	// If a rule or rule group cannot be deleted immediately due to the absence of rules,
+	// it will be held here and removed later when a new rule or rule group put event allows for its deletion.
+	pendingDeletion struct {
+		syncutil.RWMutex
+		// key: path, value: [groupID, ruleID]
+		// The map 'kvs' holds the rules or rule groups that are pending deletion.
+		// If a rule group needs to be deleted, the ruleID will be an empty string.
+		kvs map[string][2]string
+	}
 }
 
 // NewWatcher creates a new watcher to watch the Placement Rule change from PD API server.
@@ -79,22 +94,25 @@
 		ctx:                   ctx,
 		cancel:                cancel,
 		rulesPathPrefix:       endpoint.RulesPathPrefix(clusterID),
+		ruleCommonPathPrefix:  endpoint.RuleCommonPathPrefix(clusterID),
 		ruleGroupPathPrefix:   endpoint.RuleGroupPathPrefix(clusterID),
 		regionLabelPathPrefix: endpoint.RegionLabelPathPrefix(clusterID),
 		etcdClient:            etcdClient,
 		ruleStorage:           ruleStorage,
 		checkerController:     checkerController,
 		ruleManager:           ruleManager,
 		regionLabeler:         regionLabeler,
+		pendingDeletion: struct {
+			syncutil.RWMutex
+			kvs map[string][2]string
+		}{
+			kvs: make(map[string][2]string),
+		},
 	}
 	err := rw.initializeRuleWatcher()
 	if err != nil {
 		return nil, err
 	}
-	err = rw.initializeGroupWatcher()
-	if err != nil {
-		return nil, err
-	}
 	err = rw.initializeRegionLabelWatcher()
 	if err != nil {
 		return nil, err
@@ -103,85 +121,87 @@
 }
 
 func (rw *Watcher) initializeRuleWatcher() error {
-	prefixToTrim := rw.rulesPathPrefix + "/"
 	putFn := func(kv *mvccpb.KeyValue) error {
-		log.Info("update placement rule", zap.String("key", string(kv.Key)), zap.String("value", string(kv.Value)))
-		rule, err := placement.NewRuleFromJSON(kv.Value)
-		if err != nil {
-			return err
-		}
-		// Update the suspect key ranges in the checker.
-		rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
-		if oldRule := rw.ruleManager.GetRule(rule.GroupID, rule.ID); oldRule != nil {
-			rw.checkerController.AddSuspectKeyRange(oldRule.StartKey, oldRule.EndKey)
+		err := func() error {
+			if strings.HasPrefix(string(kv.Key), rw.rulesPathPrefix) {
+				log.Info("update placement rule", zap.String("key", string(kv.Key)), zap.String("value", string(kv.Value)))
+				rule, err := placement.NewRuleFromJSON(kv.Value)
+				if err != nil {
+					return err
+				}
+				// Update the suspect key ranges in the checker.
+				rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
+				if oldRule := rw.ruleManager.GetRule(rule.GroupID, rule.ID); oldRule != nil {
+					rw.checkerController.AddSuspectKeyRange(oldRule.StartKey, oldRule.EndKey)
+				}
+				return rw.ruleManager.SetRule(rule)
+			} else if strings.HasPrefix(string(kv.Key), rw.ruleGroupPathPrefix) {
+				log.Info("update placement rule group", zap.String("key", string(kv.Key)), zap.String("value", string(kv.Value)))
+				ruleGroup, err := placement.NewRuleGroupFromJSON(kv.Value)
+				if err != nil {
+					return err
+				}
+				// Add all rule key ranges within the group to the suspect key ranges.
+				for _, rule := range rw.ruleManager.GetRulesByGroup(ruleGroup.ID) {
+					rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
+				}
+				return rw.ruleManager.SetRuleGroup(ruleGroup)
+			} else {
+				log.Warn("unknown key when update placement rule", zap.String("key", string(kv.Key)))
+				return nil
+			}
+		}()
+		if err == nil && rw.hasPendingDeletion() {
+			rw.tryFinishPendingDeletion()
 		}
-		return rw.ruleManager.SetRule(rule)
+		return err
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {
 		key := string(kv.Key)
-		log.Info("delete placement rule", zap.String("key", key))
-		ruleJSON, err := rw.ruleStorage.LoadRule(strings.TrimPrefix(key, prefixToTrim))
-		if err != nil {
-			return err
-		}
-		rule, err := placement.NewRuleFromJSON([]byte(ruleJSON))
-		if err != nil {
-			return err
+		groupID, ruleID, err := func() (string, string, error) {
+			if strings.HasPrefix(string(kv.Key), rw.rulesPathPrefix) {
+				log.Info("delete placement rule", zap.String("key", key))
+				ruleJSON, err := rw.ruleStorage.LoadRule(strings.TrimPrefix(key, rw.rulesPathPrefix+"/"))
+				if err != nil {
+					return "", "", err
+				}
+				rule, err := placement.NewRuleFromJSON([]byte(ruleJSON))
+				if err != nil {
+					return "", "", err
+				}
+				rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
+				return rule.GroupID, rule.ID, rw.ruleManager.DeleteRule(rule.GroupID, rule.ID)
+			} else if strings.HasPrefix(string(kv.Key), rw.ruleGroupPathPrefix) {
+				log.Info("delete placement rule group", zap.String("key", key))
+				trimmedKey := strings.TrimPrefix(key, rw.ruleGroupPathPrefix+"/")
+				for _, rule := range rw.ruleManager.GetRulesByGroup(trimmedKey) {
+					rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
+				}
+				return trimmedKey, "", rw.ruleManager.DeleteRuleGroup(trimmedKey)
+			} else {
+				log.Warn("unknown key when delete placement rule", zap.String("key", string(kv.Key)))
+				return "", "", nil
+			}
+		}()
+		if err != nil && strings.Contains(err.Error(), "no rule left") && groupID != "" {
+			rw.addPendingDeletion(key, groupID, ruleID)
 		}
-		rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
-		return rw.ruleManager.DeleteRule(rule.GroupID, rule.ID)
+		return nil
 	}
 	postEventFn := func() error {
 		return nil
 	}
 	rw.ruleWatcher = etcdutil.NewLoopWatcher(
 		rw.ctx, &rw.wg,
 		rw.etcdClient,
-		"scheduling-rule-watcher", rw.rulesPathPrefix,
+		"scheduling-rule-watcher", rw.ruleCommonPathPrefix,
 		putFn, deleteFn, postEventFn,
 		clientv3.WithPrefix(),
 	)
 	rw.ruleWatcher.StartWatchLoop()
 	return rw.ruleWatcher.WaitLoad()
 }
 
-func (rw *Watcher) initializeGroupWatcher() error {
-	prefixToTrim := rw.ruleGroupPathPrefix + "/"
-	putFn := func(kv *mvccpb.KeyValue) error {
-		log.Info("update placement rule group", zap.String("key", string(kv.Key)), zap.String("value", string(kv.Value)))
-		ruleGroup, err := placement.NewRuleGroupFromJSON(kv.Value)
-		if err != nil {
-			return err
-		}
-		// Add all rule key ranges within the group to the suspect key ranges.
-		for _, rule := range rw.ruleManager.GetRulesByGroup(ruleGroup.ID) {
-			rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
-		}
-		return rw.ruleManager.SetRuleGroup(ruleGroup)
-	}
-	deleteFn := func(kv *mvccpb.KeyValue) error {
-		key := string(kv.Key)
-		log.Info("delete placement rule group", zap.String("key", key))
-		trimmedKey := strings.TrimPrefix(key, prefixToTrim)
-		for _, rule := range rw.ruleManager.GetRulesByGroup(trimmedKey) {
-			rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
-		}
-		return rw.ruleManager.DeleteRuleGroup(trimmedKey)
-	}
-	postEventFn := func() error {
-		return nil
-	}
-	rw.groupWatcher = etcdutil.NewLoopWatcher(
-		rw.ctx, &rw.wg,
-		rw.etcdClient,
-		"scheduling-rule-group-watcher", rw.ruleGroupPathPrefix,
-		putFn, deleteFn, postEventFn,
-		clientv3.WithPrefix(),
-	)
-	rw.groupWatcher.StartWatchLoop()
-	return rw.groupWatcher.WaitLoad()
-}
-
 func (rw *Watcher) initializeRegionLabelWatcher() error {
 	prefixToTrim := rw.regionLabelPathPrefix + "/"
 	putFn := func(kv *mvccpb.KeyValue) error {
@@ -216,3 +236,39 @@
 	rw.cancel()
 	rw.wg.Wait()
 }
+
+func (rw *Watcher) hasPendingDeletion() bool {
+	rw.pendingDeletion.RLock()
+	defer rw.pendingDeletion.RUnlock()
+	return len(rw.pendingDeletion.kvs) > 0
+}
+
+func (rw *Watcher) addPendingDeletion(path, groupID, ruleID string) {
+	rw.pendingDeletion.Lock()
+	defer rw.pendingDeletion.Unlock()
+	rw.pendingDeletion.kvs[path] = [2]string{groupID, ruleID}
+}
+
+func (rw *Watcher) tryFinishPendingDeletion() {
+	rw.pendingDeletion.Lock()
+	defer rw.pendingDeletion.Unlock()
+	originLen := len(rw.pendingDeletion.kvs)
+	for k, v := range rw.pendingDeletion.kvs {
+		groupID, ruleID := v[0], v[1]
+		var err error
+		if ruleID == "" {
+			err = rw.ruleManager.DeleteRuleGroup(groupID)
+		} else {
+			err = rw.ruleManager.DeleteRule(groupID, ruleID)
+		}
+		if err == nil {
+			delete(rw.pendingDeletion.kvs, k)
+		}
+	}
+	// If the length of the map is changed, it means that some rules or rule groups have been deleted.
+	// We need to force load the rules and rule groups to make sure sync with etcd.
+	if len(rw.pendingDeletion.kvs) != originLen {
+		rw.ruleWatcher.ForceLoad()
+		log.Info("force load rules", zap.Int("pending deletion", len(rw.pendingDeletion.kvs)), zap.Int("origin", originLen))
+	}
+}

pkg/schedule/checker/rule_checker.go

@@ -199,7 +199,7 @@ func (c *RuleChecker) fixRulePeer(region *core.RegionInfo, fit *placement.Region
 		if c.isDownPeer(region, peer) {
 			if c.isStoreDownTimeHitMaxDownTime(peer.GetStoreId()) {
 				ruleCheckerReplaceDownCounter.Inc()
-				return c.replaceUnexpectRulePeer(region, rf, fit, peer, downStatus)
+				return c.replaceUnexpectedRulePeer(region, rf, fit, peer, downStatus)
 			}
 			// When witness placement rule is enabled, promotes the witness to voter when region has down voter.
 			if c.isWitnessEnabled() && core.IsVoter(peer) {
@@ -211,7 +211,7 @@ func (c *RuleChecker) fixRulePeer(region *core.RegionInfo, fit *placement.Region
 		}
 		if c.isOfflinePeer(peer) {
 			ruleCheckerReplaceOfflineCounter.Inc()
-			return c.replaceUnexpectRulePeer(region, rf, fit, peer, offlineStatus)
+			return c.replaceUnexpectedRulePeer(region, rf, fit, peer, offlineStatus)
 		}
 	}
 	// fix loose matched peers.
@@ -246,7 +246,7 @@ func (c *RuleChecker) addRulePeer(region *core.RegionInfo, fit *placement.Region
 					continue
 				}
 				ruleCheckerNoStoreThenTryReplace.Inc()
-				op, err := c.replaceUnexpectRulePeer(region, oldPeerRuleFit, fit, p, "swap-fit")
+				op, err := c.replaceUnexpectedRulePeer(region, oldPeerRuleFit, fit, p, "swap-fit")
 				if err != nil {
 					return nil, err
 				}
@@ -267,7 +267,7 @@ func (c *RuleChecker) addRulePeer(region *core.RegionInfo, fit *placement.Region
 }
 
 // The peer's store may in Offline or Down, need to be replace.
-func (c *RuleChecker) replaceUnexpectRulePeer(region *core.RegionInfo, rf *placement.RuleFit, fit *placement.RegionFit, peer *metapb.Peer, status string) (*operator.Operator, error) {
+func (c *RuleChecker) replaceUnexpectedRulePeer(region *core.RegionInfo, rf *placement.RuleFit, fit *placement.RegionFit, peer *metapb.Peer, status string) (*operator.Operator, error) {
 	var fastFailover bool
 	// If the store to which the original peer belongs is TiFlash, the new peer cannot be set to witness, nor can it perform fast failover
 	if c.isWitnessEnabled() && !c.cluster.GetStore(peer.StoreId).IsTiFlash() {

pkg/storage/endpoint/key_path.go

@@ -31,6 +31,7 @@ const (
 	serviceMiddlewarePath     = "service_middleware"
 	schedulePath              = "schedule"
 	gcPath                    = "gc"
+	ruleCommonPath            = "rule"
 	rulesPath                 = "rules"
 	ruleGroupPath             = "rule_group"
 	regionLabelPath           = "region_label"
@@ -102,6 +103,11 @@ func RulesPathPrefix(clusterID uint64) string {
 	return path.Join(PDRootPath(clusterID), rulesPath)
 }
 
+// RuleCommonPathPrefix returns the path prefix to save the placement rule common config.
+func RuleCommonPathPrefix(clusterID uint64) string {
+	return path.Join(PDRootPath(clusterID), ruleCommonPath)
+}
+
 // RuleGroupPathPrefix returns the path prefix to save the placement rule groups.
 func RuleGroupPathPrefix(clusterID uint64) string {
 	return path.Join(PDRootPath(clusterID), ruleGroupPath)

tests/server/api/region_test.go

@@ -217,10 +217,11 @@ func (suite *regionTestSuite) TestCheckRegionsReplicated() {
 		func(conf *config.Config, serverName string) {
 			conf.Replication.EnablePlacementRules = true
 		})
-	env.RunTestInPDMode(suite.checkRegionsReplicated)
+	env.RunTestInTwoModes(suite.checkRegionsReplicated)
 }
 
 func (suite *regionTestSuite) checkRegionsReplicated(cluster *tests.TestCluster) {
+	suite.pauseRuleChecker(cluster)
 	leader := cluster.GetLeaderServer()
 	urlPrefix := leader.GetAddr() + "/pd/api/v1"
 	re := suite.Require()
@@ -271,6 +272,14 @@ func (suite *regionTestSuite) checkRegionsReplicated(cluster *tests.TestCluster)
 	err = tu.CheckPostJSON(testDialClient, urlPrefix+"/config/placement-rule", data, tu.StatusOK(re))
 	suite.NoError(err)
 
+	tu.Eventually(re, func() bool {
+		respBundle := make([]placement.GroupBundle, 0)
+		err = tu.CheckGetJSON(testDialClient, urlPrefix+"/config/placement-rule", nil,
+			tu.StatusOK(re), tu.ExtractJSON(re, &respBundle))
+		suite.NoError(err)
+		return len(respBundle) == 1 && respBundle[0].ID == "5"
+	})
+
 	tu.Eventually(re, func() bool {
 		err = tu.ReadGetJSON(re, testDialClient, url, &status)
 		suite.NoError(err)
@@ -314,9 +323,24 @@ func (suite *regionTestSuite) checkRegionsReplicated(cluster *tests.TestCluster)
 	err = tu.CheckPostJSON(testDialClient, urlPrefix+"/config/placement-rule", data, tu.StatusOK(re))
 	suite.NoError(err)
 
-	err = tu.ReadGetJSON(re, testDialClient, url, &status)
-	suite.NoError(err)
-	suite.Equal("INPROGRESS", status)
+	tu.Eventually(re, func() bool {
+		respBundle := make([]placement.GroupBundle, 0)
+		err = tu.CheckGetJSON(testDialClient, urlPrefix+"/config/placement-rule", nil,
+			tu.StatusOK(re), tu.ExtractJSON(re, &respBundle))
+		suite.NoError(err)
+		if len(respBundle) != 2 {
+			return false
+		}
+		s1 := respBundle[0].ID == "5" && respBundle[1].ID == "6"
+		s2 := respBundle[0].ID == "6" && respBundle[1].ID == "5"
+		return s1 || s2
+	})
+
+	tu.Eventually(re, func() bool {
+		err = tu.ReadGetJSON(re, testDialClient, url, &status)
+		suite.NoError(err)
+		return status == "INPROGRESS"
+	})
 
 	r1 = core.NewTestRegionInfo(2, 1, []byte("a"), []byte("b"))
 	r1.GetMeta().Peers = append(r1.GetMeta().Peers, &metapb.Peer{Id: 5, StoreId: 1}, &metapb.Peer{Id: 6, StoreId: 1}, &metapb.Peer{Id: 7, StoreId: 1})
@@ -338,3 +362,17 @@ func (suite *regionTestSuite) checkRegionCount(cluster *tests.TestCluster, count
 		})
 	}
 }
+
+// pauseRuleChecker will pause rule checker to avoid unexpected operator.
+func (suite *regionTestSuite) pauseRuleChecker(cluster *tests.TestCluster) {
+	re := suite.Require()
+	checkerName := "rule"
+	addr := cluster.GetLeaderServer().GetAddr()
+	resp := make(map[string]interface{})
+	url := fmt.Sprintf("%s/pd/api/v1/checker/%s", addr, checkerName)
+	err := tu.CheckPostJSON(testDialClient, url, []byte(`{"delay":1000}`), tu.StatusOK(re))
+	re.NoError(err)
+	err = tu.ReadGetJSON(re, testDialClient, url, &resp)
+	re.NoError(err)
+	re.True(resp["paused"].(bool))
+}