mcs: fix rule manager initialize

pkg/mcs/scheduling/server/cluster.go

@@ -113,6 +113,7 @@ func NewCluster(
 		logRunner:       ratelimit.NewConcurrentRunner(logTaskRunner, ratelimit.NewConcurrencyLimiter(uint64(runtime.NumCPU()*2)), time.Minute),
 	}
 	c.coordinator = schedule.NewCoordinator(ctx, c, hbStreams)
+	c.ruleManager.SetInMicroService(true)
 	err = c.ruleManager.Initialize(persistConfig.GetMaxReplicas(), persistConfig.GetLocationLabels(), persistConfig.GetIsolationLevel())
 	if err != nil {
 		cancel()

pkg/schedule/placement/rule_manager.go

@@ -35,6 +35,7 @@ import (
 	"github.com/tikv/pd/pkg/core/constant"
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/schedule/config"
+	"github.com/tikv/pd/pkg/schedule/rangelist"
 	"github.com/tikv/pd/pkg/slice"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
@@ -56,9 +57,10 @@ type RuleManager struct {
 	ctx     context.Context
 	storage endpoint.RuleStorage
 	syncutil.RWMutex
-	initialized bool
-	ruleConfig  *ruleConfig
-	ruleList    ruleList
+	initialized    bool
+	inMicroService bool
+	ruleConfig     *ruleConfig
+	ruleList       ruleList
 
 	// used for rule validation
 	keyType          string
@@ -87,6 +89,15 @@ func (m *RuleManager) Initialize(maxReplica int, locationLabels []string, isolat
 	if m.initialized {
 		return nil
 	}
+	// If RuleManager is initialized in micro service,
+	// it will load from etcd watcher and do not modify rule directly.
+	if m.inMicroService {
+		m.ruleList = ruleList{
+			rangeList: rangelist.List{},
+		}
+		m.initialized = true
+		return nil
+	}
 
 	if err := m.loadRules(); err != nil {
 		return err
@@ -831,6 +842,13 @@ func (m *RuleManager) SetKeyType(h string) *RuleManager {
 	return m
 }
 
+// SetInMicroService set whether rule manager is in micro service.
+func (m *RuleManager) SetInMicroService(v bool) {
+	m.Lock()
+	defer m.Unlock()
+	m.inMicroService = v
+}
+
 func getStoresByRegion(storeSet StoreSet, region *core.RegionInfo) []*core.StoreInfo {
 	r := make([]*core.StoreInfo, 0, len(region.GetPeers()))
 	for _, peer := range region.GetPeers() {

tests/integrations/mcs/scheduling/rule_test.go

@@ -16,11 +16,15 @@ package scheduling
 
 import (
 	"context"
+	"encoding/json"
+	"fmt"
 	"sort"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
 
+	"github.com/pingcap/failpoint"
+
 	"github.com/tikv/pd/pkg/schedule/labeler"
 	"github.com/tikv/pd/pkg/schedule/placement"
 	"github.com/tikv/pd/pkg/utils/testutil"
@@ -46,6 +50,7 @@ func TestRule(t *testing.T) {
 
 func (suite *ruleTestSuite) SetupSuite() {
 	re := suite.Require()
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/cluster/highFrequencyClusterJobs", `return(true)`))
 
 	var err error
 	suite.ctx, suite.cancel = context.WithCancel(context.Background())
@@ -63,12 +68,14 @@ func (suite *ruleTestSuite) SetupSuite() {
 func (suite *ruleTestSuite) TearDownSuite() {
 	suite.cancel()
 	suite.cluster.Destroy()
+	re := suite.Require()
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/cluster/highFrequencyClusterJobs"))
 }
 
 func (suite *ruleTestSuite) TestRuleWatch() {
 	re := suite.Require()
 
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoint)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 
@@ -205,3 +212,54 @@ func (suite *ruleTestSuite) TestRuleWatch() {
 	re.Equal(labelRule.Labels, labelRules[1].Labels)
 	re.Equal(labelRule.RuleType, labelRules[1].RuleType)
 }
+
+func (suite *ruleTestSuite) TestSchedulingSwitch() {
+	re := suite.Require()
+
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 2, suite.cluster)
+	re.NoError(err)
+	defer tc.Destroy()
+	tc.WaitForPrimaryServing(re)
+
+	// Add a new rule from "" to ""
+	url := fmt.Sprintf("%s/pd/api/v1/config/placement-rule", suite.pdLeaderServer.GetAddr())
+	respBundle := make([]placement.GroupBundle, 0)
+	testutil.Eventually(re, func() bool {
+		err = testutil.CheckGetJSON(tests.TestDialClient, url, nil,
+			testutil.StatusOK(re), testutil.ExtractJSON(re, &respBundle))
+		re.NoError(err)
+		return len(respBundle) == 1 && len(respBundle[0].Rules) == 1
+	})
+
+	b2 := placement.GroupBundle{
+		ID:    "pd",
+		Index: 1,
+		Rules: []*placement.Rule{
+			{GroupID: "pd", ID: "rule0", Index: 1, Role: placement.Voter, Count: 3},
+		},
+	}
+	data, err := json.Marshal(b2)
+	re.NoError(err)
+
+	err = testutil.CheckPostJSON(tests.TestDialClient, url+"/pd", data, testutil.StatusOK(re))
+	re.NoError(err)
+	testutil.Eventually(re, func() bool {
+		err = testutil.CheckGetJSON(tests.TestDialClient, url, nil,
+			testutil.StatusOK(re), testutil.ExtractJSON(re, &respBundle))
+		re.NoError(err)
+		return len(respBundle) == 1 && len(respBundle[0].Rules) == 1
+	})
+
+	// Switch another server
+	oldPrimary := tc.GetPrimaryServer()
+	oldPrimary.Close()
+	tc.WaitForPrimaryServing(re)
+	newPrimary := tc.GetPrimaryServer()
+	re.NotEqual(oldPrimary.GetAddr(), newPrimary.GetAddr())
+	testutil.Eventually(re, func() bool {
+		err = testutil.CheckGetJSON(tests.TestDialClient, url, nil,
+			testutil.StatusOK(re), testutil.ExtractJSON(re, &respBundle))
+		re.NoError(err)
+		return len(respBundle) == 1 && len(respBundle[0].Rules) == 1
+	})
+}

tests/integrations/mcs/scheduling/server_test.go

@@ -91,7 +91,7 @@ func (suite *serverTestSuite) TearDownSuite() {
 func (suite *serverTestSuite) TestAllocID() {
 	re := suite.Require()
 	re.NoError(failpoint.Enable("github.com/tikv/pd/pkg/mcs/scheduling/server/fastUpdateMember", `return(true)`))
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -110,7 +110,7 @@ func (suite *serverTestSuite) TestAllocIDAfterLeaderChange() {
 	err = pd2.Run()
 	re.NotEmpty(suite.cluster.WaitLeader())
 	re.NoError(err)
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -138,7 +138,7 @@ func (suite *serverTestSuite) TestAllocIDAfterLeaderChange() {
 
 func (suite *serverTestSuite) TestPrimaryChange() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 2, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 2, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -164,7 +164,7 @@ func (suite *serverTestSuite) TestPrimaryChange() {
 
 func (suite *serverTestSuite) TestForwardStoreHeartbeat() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -225,7 +225,7 @@ func (suite *serverTestSuite) TestSchedulingServiceFallback() {
 		return suite.pdLeader.GetServer().GetRaftCluster().IsSchedulingControllerRunning()
 	})
 
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -242,7 +242,7 @@ func (suite *serverTestSuite) TestSchedulingServiceFallback() {
 	testutil.Eventually(re, func() bool {
 		return suite.pdLeader.GetServer().GetRaftCluster().IsSchedulingControllerRunning()
 	})
-	tc1, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc1, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc1.Destroy()
 	tc1.WaitForPrimaryServing(re)
@@ -278,7 +278,7 @@ func (suite *serverTestSuite) TestDisableSchedulingServiceFallback() {
 		return suite.pdLeader.GetServer().GetRaftCluster().IsSchedulingControllerRunning()
 	})
 
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -302,7 +302,7 @@ func (suite *serverTestSuite) TestDisableSchedulingServiceFallback() {
 
 func (suite *serverTestSuite) TestSchedulerSync() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -422,7 +422,7 @@ func checkEvictLeaderStoreIDs(re *require.Assertions, sc *schedulers.Controller,
 
 func (suite *serverTestSuite) TestForwardRegionHeartbeat() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -499,7 +499,7 @@ func (suite *serverTestSuite) TestForwardRegionHeartbeat() {
 
 func (suite *serverTestSuite) TestStoreLimit() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -660,7 +660,7 @@ func (suite *multipleServerTestSuite) TestReElectLeader() {
 	defer func() {
 		re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/member/skipCampaignLeaderCheck"))
 	}()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)
@@ -692,7 +692,7 @@ func (suite *multipleServerTestSuite) TestReElectLeader() {
 
 func (suite *serverTestSuite) TestOnlineProgress() {
 	re := suite.Require()
-	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.backendEndpoints)
+	tc, err := tests.NewTestSchedulingCluster(suite.ctx, 1, suite.cluster)
 	re.NoError(err)
 	defer tc.Destroy()
 	tc.WaitForPrimaryServing(re)

tests/scheduling_cluster.go

@@ -22,6 +22,7 @@ import (
 
 	scheduling "github.com/tikv/pd/pkg/mcs/scheduling/server"
 	sc "github.com/tikv/pd/pkg/mcs/scheduling/server/config"
+	"github.com/tikv/pd/pkg/mcs/utils/constant"
 	"github.com/tikv/pd/pkg/schedule/schedulers"
 	"github.com/tikv/pd/pkg/utils/tempurl"
 	"github.com/tikv/pd/pkg/utils/testutil"
@@ -31,17 +32,19 @@ import (
 type TestSchedulingCluster struct {
 	ctx context.Context
 
+	pd               *TestCluster
 	backendEndpoints string
 	servers          map[string]*scheduling.Server
 	cleanupFuncs     map[string]testutil.CleanupFunc
 }
 
 // NewTestSchedulingCluster creates a new scheduling test cluster.
-func NewTestSchedulingCluster(ctx context.Context, initialServerCount int, backendEndpoints string) (tc *TestSchedulingCluster, err error) {
+func NewTestSchedulingCluster(ctx context.Context, initialServerCount int, pd *TestCluster) (tc *TestSchedulingCluster, err error) {
 	schedulers.Register()
 	tc = &TestSchedulingCluster{
 		ctx:              ctx,
-		backendEndpoints: backendEndpoints,
+		pd:               pd,
+		backendEndpoints: pd.GetLeaderServer().GetAddr(),
 		servers:          make(map[string]*scheduling.Server, initialServerCount),
 		cleanupFuncs:     make(map[string]testutil.CleanupFunc, initialServerCount),
 	}
@@ -115,7 +118,9 @@ func (tc *TestSchedulingCluster) WaitForPrimaryServing(re *require.Assertions) *
 		}
 		return false
 	}, testutil.WithWaitFor(10*time.Second), testutil.WithTickInterval(50*time.Millisecond))
-
+	testutil.Eventually(re, func() bool {
+		return tc.pd.GetLeaderServer().GetRaftCluster().IsServiceIndependent(constant.SchedulingServiceName)
+	})
 	return primary
 }
 

tests/testutil.go

@@ -389,7 +389,7 @@ func (s *SchedulingTestEnvironment) startCluster(m SchedulerMode) {
 		re.NoError(leaderServer.BootstrapCluster())
 		leaderServer.GetRaftCluster().SetPrepared()
 		// start scheduling cluster
-		tc, err := NewTestSchedulingCluster(ctx, 1, leaderServer.GetAddr())
+		tc, err := NewTestSchedulingCluster(ctx, 1, cluster)
 		re.NoError(err)
 		tc.WaitForPrimaryServing(re)
 		tc.GetPrimaryServer().GetCluster().SetPrepared()

tools/pd-ctl/tests/config/config_test.go

@@ -740,7 +740,9 @@ func (suite *configTestSuite) checkPlacementRuleBundle(cluster *pdTests.TestClus
 	output, err = tests.ExecuteCommand(cmd, "-u", pdAddr, "config", "placement-rules", "rule-bundle", "get", placement.DefaultGroupID)
 	re.NoError(err)
 	re.NoError(json.Unmarshal(output, &bundle))
-	re.Equal(placement.GroupBundle{ID: placement.DefaultGroupID, Index: 0, Override: false, Rules: []*placement.Rule{{GroupID: placement.DefaultGroupID, ID: placement.DefaultRuleID, Role: placement.Voter, Count: 3}}}, bundle)
+	expect := placement.GroupBundle{ID: placement.DefaultGroupID, Index: 0, Override: false, Rules: []*placement.Rule{{GroupID: placement.DefaultGroupID, ID: placement.DefaultRuleID, Role: placement.Voter, Count: 3}}}
+	expect.Rules[0].CreateTimestamp = bundle.Rules[0].CreateTimestamp // skip create timestamp in mcs
+	re.Equal(expect, bundle)
 
 	f, err := os.CreateTemp("", "pd_tests")
 	re.NoError(err)