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"
@@ -62,6 +63,17 @@
 	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.
@@ -86,6 +98,12 @@
 		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 {
@@ -115,7 +133,11 @@
 		if oldRule := rw.ruleManager.GetRule(rule.GroupID, rule.ID); oldRule != nil {
 			rw.checkerController.AddSuspectKeyRange(oldRule.StartKey, oldRule.EndKey)
 		}
-		return rw.ruleManager.SetRule(rule)
+		err = rw.ruleManager.SetRule(rule)
+		if err == nil && rw.hasPendingDeletion() {
+			rw.tryFinishPendingDeletion()
+		}
+		return err
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {
 		key := string(kv.Key)
@@ -129,7 +151,11 @@
 			return err
 		}
 		rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
-		return rw.ruleManager.DeleteRule(rule.GroupID, rule.ID)
+		err = rw.ruleManager.DeleteRule(rule.GroupID, rule.ID)
+		if err != nil && strings.Contains(err.Error(), "no rule left") {
+			rw.addPendingDeletion(key, rule.GroupID, rule.ID)
+		}
+		return err
 	}
 	postEventFn := func() error {
 		return nil
@@ -157,7 +183,11 @@
 		for _, rule := range rw.ruleManager.GetRulesByGroup(ruleGroup.ID) {
 			rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
 		}
-		return rw.ruleManager.SetRuleGroup(ruleGroup)
+		err = rw.ruleManager.SetRuleGroup(ruleGroup)
+		if err == nil && rw.hasPendingDeletion() {
+			rw.tryFinishPendingDeletion()
+		}
+		return err
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {
 		key := string(kv.Key)
@@ -166,7 +196,11 @@
 		for _, rule := range rw.ruleManager.GetRulesByGroup(trimmedKey) {
 			rw.checkerController.AddSuspectKeyRange(rule.StartKey, rule.EndKey)
 		}
-		return rw.ruleManager.DeleteRuleGroup(trimmedKey)
+		err := rw.ruleManager.DeleteRuleGroup(trimmedKey)
+		if err != nil && strings.Contains(err.Error(), "no rule left") {
+			rw.addPendingDeletion(key, trimmedKey, "")
+		}
+		return err
 	}
 	postEventFn := func() error {
 		return nil
@@ -216,3 +250,40 @@
 	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()
+		rw.groupWatcher.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() {

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))
+}