Skip to content

Commit

Permalink
This is an automated cherry-pick of tikv#7122
Browse files Browse the repository at this point in the history
close tikv#7121

Signed-off-by: ti-chi-bot <[email protected]>
  • Loading branch information
rleungx committed May 11, 2024
1 parent 21b47d2 commit 5e56ccc
Show file tree
Hide file tree
Showing 10 changed files with 63 additions and 19 deletions.
2 changes: 1 addition & 1 deletion pkg/mock/mockcluster/mockcluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -166,7 +166,7 @@ func (mc *Cluster) AllocPeer(storeID uint64) (*metapb.Peer, error) {
func (mc *Cluster) initRuleManager() {
if mc.RuleManager == nil {
mc.RuleManager = placement.NewRuleManager(core.NewStorage(kv.NewMemoryKV()), 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
Original file line number Diff line number Diff line change
Expand Up @@ -350,7 +350,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
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,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
4 changes: 2 additions & 2 deletions server/cluster/cluster_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -850,7 +850,7 @@ func (s *testClusterInfoSuite) TestOfflineAndMerge(c *C) {
storage := core.NewStorage(kv.NewMemoryKV())
cluster.ruleManager = placement.NewRuleManager(storage, 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 @@ -1146,7 +1146,7 @@ func newTestCluster(ctx context.Context, opt *config.PersistOptions) *testCluste
rc := newTestRaftCluster(ctx, mockid.NewIDAllocator(), opt, storage, core.NewBasicCluster())
rc.ruleManager = placement.NewRuleManager(storage, rc, rc.GetOpts())
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
Original file line number Diff line number Diff line change
Expand Up @@ -252,6 +252,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)
}

// SetStoreLimit sets a store limit for a given type and rate.
func (o *PersistOptions) SetStoreLimit(storeID uint64, typ storelimit.Type, ratePerMin float64) {
v := o.GetScheduleConfig().Clone()
Expand Down
3 changes: 2 additions & 1 deletion server/schedule/placement/rule_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ func NewRuleManager(storage *core.Storage, storeSetInformer core.StoreSetInforme

// 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 @@ -83,6 +83,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
7 changes: 4 additions & 3 deletions server/schedule/placement/rule_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package placement

import (
"encoding/hex"

. "github.com/pingcap/check"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/tikv/pd/pkg/codec"
Expand All @@ -34,7 +35,7 @@ func (s *testManagerSuite) SetUpTest(c *C) {
s.store = core.NewStorage(kv.NewMemoryKV())
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 @@ -111,7 +112,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 @@ -126,7 +127,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
Original file line number Diff line number Diff line change
Expand Up @@ -859,7 +859,7 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
}
if cfg.EnablePlacementRules {
// initialize rule manager.
if err := raftCluster.GetRuleManager().Initialize(int(cfg.MaxReplicas), cfg.LocationLabels); err != nil {
if err := raftCluster.GetRuleManager().Initialize(int(cfg.MaxReplicas), cfg.LocationLabels, cfg.IsolationLevel); err != nil {
return err
}
} else {
Expand All @@ -878,19 +878,19 @@ func (s *Server) SetReplicationConfig(cfg config.ReplicationConfig) error {
defaultRule := s.GetRaftCluster().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 @@ -901,6 +901,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
if err := s.GetRaftCluster().GetRuleManager().SetRule(rule); err != nil {
log.Error("failed to update rule count",
errs.ZapError(err))
Expand Down
2 changes: 1 addition & 1 deletion server/statistics/region_collection_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ func (t *testRegionStatisticsSuite) SetUpTest(c *C) {
t.store = core.NewStorage(kv.NewMemoryKV())
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
Original file line number Diff line number Diff line change
Expand Up @@ -620,7 +620,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 @@ -629,6 +629,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 @@ -647,6 +656,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 @@ -655,8 +673,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 @@ -669,11 +692,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 @@ -703,8 +733,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 5e56ccc

Please sign in to comment.