Skip to content

Commit

Permalink
*: fix sync isolation level to default placement rule (tikv#7122) (ti…
Browse files Browse the repository at this point in the history 
…kv#7125)

close tikv#7121

Signed-off-by: ti-chi-bot <[email protected]>
Signed-off-by: Ryan Leung <[email protected]>

Co-authored-by: Ryan Leung <[email protected]>
Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
  • Loading branch information
@ti-chi-bot @rleungx @ti-chi-bot
3 people authored Sep 10, 2024
1 parent e878b62 commit 5c8d3f3
Show file tree
Hide file tree
Showing 10 changed files with 65 additions and 22 deletions.
2 changes: 1 addition & 1 deletion pkg/mock/mockcluster/mockcluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ func (mc *Cluster) AllocPeer(storeID uint64) (*metapb.Peer, error) {
func (mc *Cluster) initRuleManager() {
if mc.RuleManager == nil {
mc.RuleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), mc, mc.GetOpts())
mc.RuleManager.Initialize(int(mc.GetReplicationConfig().MaxReplicas), mc.GetReplicationConfig().LocationLabels)
mc.RuleManager.Initialize(int(mc.GetReplicationConfig().MaxReplicas), mc.GetReplicationConfig().LocationLabels, mc.GetReplicationConfig().IsolationLevel)
}
}

Expand Down
4 changes: 3 additions & 1 deletion server/api/operator_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -364,7 +364,9 @@ func (s *testTransferRegionOperatorSuite) TestTransferRegionWithPlacementRule(c
if tc.placementRuleEnable {
err := s.svr.GetRaftCluster().GetRuleManager().Initialize(
s.svr.GetRaftCluster().GetOpts().GetMaxReplicas(),
s.svr.GetRaftCluster().GetOpts().GetLocationLabels())
s.svr.GetRaftCluster().GetOpts().GetLocationLabels(),
s.svr.GetRaftCluster().GetOpts().GetIsolationLevel(),
)
c.Assert(err, IsNil)
}
if len(tc.rules) > 0 {
Expand Down
2 changes: 1 addition & 1 deletion server/cluster/cluster.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -253,7 +253,7 @@ func (c *RaftCluster) Start(s Server) error {

c.ruleManager = placement.NewRuleManager(c.storage, c, c.GetOpts())
if c.opt.IsPlacementRulesEnabled() {
err = c.ruleManager.Initialize(c.opt.GetMaxReplicas(), c.opt.GetLocationLabels())
err = c.ruleManager.Initialize(c.opt.GetMaxReplicas(), c.opt.GetLocationLabels(), c.opt.GetIsolationLevel())
if err != nil {
return err
}
Expand Down
10 changes: 5 additions & 5 deletions server/cluster/cluster_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ func (s *testClusterInfoSuite) TestSetOfflineStore(c *C) {
cluster.coordinator = newCoordinator(s.ctx, cluster, nil)
cluster.ruleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), cluster, cluster.GetOpts())
if opt.IsPlacementRulesEnabled() {
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels())
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel())
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -393,7 +393,7 @@ func (s *testClusterInfoSuite) TestUpStore(c *C) {
cluster.coordinator = newCoordinator(s.ctx, cluster, nil)
cluster.ruleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), cluster, cluster.GetOpts())
if opt.IsPlacementRulesEnabled() {
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels())
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel())
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -488,7 +488,7 @@ func (s *testClusterInfoSuite) TestDeleteStoreUpdatesClusterVersion(c *C) {
cluster.coordinator = newCoordinator(s.ctx, cluster, nil)
cluster.ruleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), cluster, cluster.GetOpts())
if opt.IsPlacementRulesEnabled() {
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels())
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel())
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -1118,7 +1118,7 @@ func (s *testClusterInfoSuite) TestOfflineAndMerge(c *C) {
cluster.coordinator = newCoordinator(s.ctx, cluster, nil)
cluster.ruleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), cluster, cluster.GetOpts())
if opt.IsPlacementRulesEnabled() {
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels())
err := cluster.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel())
if err != nil {
panic(err)
}
Expand Down Expand Up @@ -1705,7 +1705,7 @@ func newTestRaftCluster(
rc.InitCluster(id, opt, s, basicCluster)
rc.ruleManager = placement.NewRuleManager(storage.NewStorageWithMemoryBackend(), rc, opt)
if opt.IsPlacementRulesEnabled() {
err := rc.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels())
err := rc.ruleManager.Initialize(opt.GetMaxReplicas(), opt.GetLocationLabels(), opt.GetIsolationLevel())
if err != nil {
panic(err)
}
Expand Down
7 changes: 7 additions & 0 deletions server/config/persist_options.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,13 @@ func (o *PersistOptions) SetSplitMergeInterval(splitMergeInterval time.Duration)
o.SetScheduleConfig(v)
}

// SetMaxStoreDownTime to set the max store down time. It's only used to test.
func (o *PersistOptions) SetMaxStoreDownTime(time time.Duration) {
v := o.GetScheduleConfig().Clone()
v.MaxStoreDownTime = typeutil.NewDuration(time)
o.SetScheduleConfig(v)
}

// SetMaxMergeRegionSize sets the max merge region size.
func (o *PersistOptions) SetMaxMergeRegionSize(maxMergeRegionSize uint64) {
v := o.GetScheduleConfig().Clone()
Expand Down
3 changes: 2 additions & 1 deletion server/schedule/placement/rule_manager.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ func NewRuleManager(storage endpoint.RuleStorage, storeSetInformer core.StoreSet

// Initialize loads rules from storage. If Placement Rules feature is never enabled, it creates default rule that is
// compatible with previous configuration.
func (m *RuleManager) Initialize(maxReplica int, locationLabels []string) error {
func (m *RuleManager) Initialize(maxReplica int, locationLabels []string, isolationLevel string) error {
m.Lock()
defer m.Unlock()
if m.initialized {
Expand All @@ -84,6 +84,7 @@ func (m *RuleManager) Initialize(maxReplica int, locationLabels []string) error
Role: Voter,
Count: maxReplica,
LocationLabels: locationLabels,
IsolationLevel: isolationLevel,
}
if err := m.storage.SaveRule(defaultRule.StoreKey(), defaultRule); err != nil {
return err
Expand Down
6 changes: 3 additions & 3 deletions server/schedule/placement/rule_manager_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ func (s *testManagerSuite) SetUpTest(c *C) {
s.store = storage.NewStorageWithMemoryBackend()
var err error
s.manager = NewRuleManager(s.store, nil, nil)
err = s.manager.Initialize(3, []string{"zone", "rack", "host"})
err = s.manager.Initialize(3, []string{"zone", "rack", "host"}, "")
c.Assert(err, IsNil)
}

Expand Down Expand Up @@ -113,7 +113,7 @@ func (s *testManagerSuite) TestSaveLoad(c *C) {
}

m2 := NewRuleManager(s.store, nil, nil)
err := m2.Initialize(3, []string{"no", "labels"})
err := m2.Initialize(3, []string{"no", "labels"}, "")
c.Assert(err, IsNil)
c.Assert(m2.GetAllRules(), HasLen, 3)
c.Assert(m2.GetRule("pd", "default").String(), Equals, rules[0].String())
Expand All @@ -128,7 +128,7 @@ func (s *testManagerSuite) TestSetAfterGet(c *C) {
s.manager.SetRule(rule)

m2 := NewRuleManager(s.store, nil, nil)
err := m2.Initialize(100, []string{})
err := m2.Initialize(100, []string{}, "")
c.Assert(err, IsNil)
rule = m2.GetRule("pd", "default")
c.Assert(rule.Count, Equals, 1)
Expand Down
11 changes: 6 additions & 5 deletions server/server.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -897,7 +897,7 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
}
if cfg.EnablePlacementRules {
// initialize rule manager.
if err := rc.GetRuleManager().Initialize(int(cfg.MaxReplicas), cfg.LocationLabels); err != nil {
if err := rc.GetRuleManager().Initialize(int(cfg.MaxReplicas), cfg.LocationLabels, cfg.IsolationLevel); err != nil {
return err
}
} else {
Expand All @@ -920,19 +920,19 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
defaultRule := rc.GetRuleManager().GetRule("pd", "default")

CheckInDefaultRule := func() error {
// replication config won't work when placement rule is enabled and exceeds one default rule
// replication config won't work when placement rule is enabled and exceeds one default rule
if !(defaultRule != nil &&
len(defaultRule.StartKey) == 0 && len(defaultRule.EndKey) == 0) {
return errors.New("cannot update MaxReplicas or LocationLabels when placement rules feature is enabled and not only default rule exists, please update rule instead")
return errors.New("cannot update MaxReplicas, LocationLabels or IsolationLevel when placement rules feature is enabled and not only default rule exists, please update rule instead")
}
if !(defaultRule.Count == int(old.MaxReplicas) && typeutil.StringsEqual(defaultRule.LocationLabels, []string(old.LocationLabels))) {
if !(defaultRule.Count == int(old.MaxReplicas) && typeutil.StringsEqual(defaultRule.LocationLabels, []string(old.LocationLabels)) && defaultRule.IsolationLevel == old.IsolationLevel) {
return errors.New("cannot to update replication config, the default rules do not consistent with replication config, please update rule instead")
}

return nil
}

if !(cfg.MaxReplicas == old.MaxReplicas && typeutil.StringsEqual(cfg.LocationLabels, old.LocationLabels)) {
if !(cfg.MaxReplicas == old.MaxReplicas && typeutil.StringsEqual(cfg.LocationLabels, old.LocationLabels) && cfg.IsolationLevel == old.IsolationLevel) {
if err := CheckInDefaultRule(); err != nil {
return err
}
Expand All @@ -943,6 +943,7 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
if rule != nil {
rule.Count = int(cfg.MaxReplicas)
rule.LocationLabels = cfg.LocationLabels
rule.IsolationLevel = cfg.IsolationLevel
rc := s.GetRaftCluster()
if rc == nil {
return errs.ErrNotBootstrapped.GenWithStackByArgs()
Expand Down
2 changes: 1 addition & 1 deletion server/statistics/region_collection_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ func (t *testRegionStatisticsSuite) SetUpTest(c *C) {
t.store = storage.NewStorageWithMemoryBackend()
var err error
t.manager = placement.NewRuleManager(t.store, nil, nil)
err = t.manager.Initialize(3, []string{"zone", "rack", "host"})
err = t.manager.Initialize(3, []string{"zone", "rack", "host"}, "")
c.Assert(err, IsNil)
}

Expand Down
40 changes: 36 additions & 4 deletions tests/pdctl/config/config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -637,7 +637,7 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
c.Assert(replicationCfg.MaxReplicas, Equals, expect)
}

checkLocaltionLabels := func(expect int) {
checkLocationLabels := func(expect int) {
args := []string{"-u", pdAddr, "config", "show", "replication"}
output, err := pdctl.ExecuteCommand(cmd, args...)
c.Assert(err, IsNil)
Expand All @@ -646,6 +646,15 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
c.Assert(replicationCfg.LocationLabels, HasLen, expect)
}

checkIsolationLevel := func(expect string) {
args := []string{"-u", pdAddr, "config", "show", "replication"}
output, err := pdctl.ExecuteCommand(cmd, args...)
c.Assert(err, IsNil)
replicationCfg := config.ReplicationConfig{}
c.Assert(json.Unmarshal(output, &replicationCfg), IsNil)
c.Assert(replicationCfg.IsolationLevel, Equals, expect)
}

checkRuleCount := func(expect int) {
args := []string{"-u", pdAddr, "config", "placement-rules", "show", "--group", "pd", "--id", "default"}
output, err := pdctl.ExecuteCommand(cmd, args...)
Expand All @@ -664,6 +673,15 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
c.Assert(rule.LocationLabels, HasLen, expect)
}

checkRuleIsolationLevel := func(expect string) {
args := []string{"-u", pdAddr, "config", "placement-rules", "show", "--group", "pd", "--id", "default"}
output, err := pdctl.ExecuteCommand(cmd, args...)
c.Assert(err, IsNil)
rule := placement.Rule{}
c.Assert(json.Unmarshal(output, &rule), IsNil)
c.Assert(rule.IsolationLevel, Equals, expect)
}

// update successfully when placement rules is not enabled.
output, err := pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "set", "max-replicas", "2")
c.Assert(err, IsNil)
Expand All @@ -672,8 +690,13 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
output, err = pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "set", "location-labels", "zone,host")
c.Assert(err, IsNil)
c.Assert(strings.Contains(string(output), "Success!"), IsTrue)
checkLocaltionLabels(2)
output, err = pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "set", "isolation-level", "zone")
c.Assert(err, IsNil)
c.Assert(strings.Contains(string(output), "Success!"), IsTrue)
checkLocationLabels(2)
checkRuleLocationLabels(2)
checkIsolationLevel("zone")
checkRuleIsolationLevel("zone")

// update successfully when only one default rule exists.
output, err = pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "placement-rules", "enable")
Expand All @@ -686,11 +709,18 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
checkMaxReplicas(3)
checkRuleCount(3)

// We need to change isolation first because we will validate
// if the location label contains the isolation level when setting location labels.
output, err = pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "set", "isolation-level", "host")
c.Assert(err, IsNil)
c.Assert(strings.Contains(string(output), "Success!"), IsTrue)
output, err = pdctl.ExecuteCommand(cmd, "-u", pdAddr, "config", "set", "location-labels", "host")
c.Assert(err, IsNil)
c.Assert(strings.Contains(string(output), "Success!"), IsTrue)
checkLocaltionLabels(1)
checkLocationLabels(1)
checkRuleLocationLabels(1)
checkIsolationLevel("host")
checkRuleIsolationLevel("host")

// update unsuccessfully when many rule exists.
f, _ := os.CreateTemp("/tmp", "pd_tests")
Expand Down Expand Up @@ -720,8 +750,10 @@ func (s *configTestSuite) TestUpdateDefaultReplicaConfig(c *C) {
c.Assert(err, IsNil)
checkMaxReplicas(4)
checkRuleCount(4)
checkLocaltionLabels(1)
checkLocationLabels(1)
checkRuleLocationLabels(1)
checkIsolationLevel("host")
checkRuleIsolationLevel("host")
}

func (s *configTestSuite) TestPDServerConfig(c *C) {
Expand Down

0 comments on commit 5c8d3f3

Please sign in to comment.