*:Rollback config in store when kv.persist failed

server/cluster_test.go

@@ -16,6 +16,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"github.com/pkg/errors"
 	"strings"
 	"sync"
 	"time"
@@ -44,6 +45,14 @@ type testClusterSuite struct {
 	baseCluster
 }
 
+type testErrorKV struct {
+	core.KVBase
+}
+
+func (kv *testErrorKV) Save(key, value string) error {
+	return errors.New("save failed")
+}
+
 func mustNewGrpcClient(c *C, addr string) pdpb.PDClient {
 	conn, err := grpc.Dial(strings.TrimPrefix(addr, "http://"), grpc.WithInsecure())
 
@@ -600,3 +609,84 @@ func (s *testGetStoresSuite) BenchmarkGetStores(c *C) {
 		s.cluster.core.Stores.GetStores()
 	}
 }
+
+func (s *testClusterSuite) TestSetScheduleOpt(c *C) {
+	var err error
+	var cleanup func()
+	_, s.svr, cleanup, err = NewTestServer(c)
+	c.Assert(err, IsNil)
+	mustWaitLeader(c, []*Server{s.svr})
+	s.grpcPDClient = mustNewGrpcClient(c, s.svr.GetAddr())
+	defer cleanup()
+	clusterID := s.svr.clusterID
+
+	storeAddr := "127.0.0.1:0"
+	_, err = s.svr.bootstrapCluster(s.newBootstrapRequest(c, clusterID, storeAddr))
+	c.Assert(err, IsNil)
+
+	_, opt, err := newTestScheduleConfig()
+	c.Assert(err, IsNil)
+
+	scheduleCfg := opt.load()
+	replicateCfg := s.svr.GetReplicationConfig()
+	pdServerCfg := s.svr.scheduleOpt.loadPDServerConfig()
+
+	//PUT GET DELETE successed
+	replicateCfg.MaxReplicas = 5
+	scheduleCfg.MaxSnapshotCount = 10
+	pdServerCfg.UseRegionStorage = true
+	typ, labelKey, labelValue := "testTyp", "testKey", "testValue"
+	nsConfig := NamespaceConfig{LeaderScheduleLimit: uint64(200)}
+
+	c.Assert(s.svr.SetScheduleConfig(*scheduleCfg), IsNil)
+	c.Assert(s.svr.SetPDServerConfig(*pdServerCfg), IsNil)
+	c.Assert(s.svr.SetLabelProperty(typ, labelKey, labelValue), IsNil)
+	c.Assert(s.svr.SetNamespaceConfig("testNS", nsConfig), IsNil)
+	c.Assert(s.svr.SetReplicationConfig(*replicateCfg), IsNil)
+
+	c.Assert(s.svr.GetReplicationConfig().MaxReplicas, Equals, uint64(5))
+	c.Assert(s.svr.scheduleOpt.GetMaxSnapshotCount(), Equals, uint64(10))
+	c.Assert(s.svr.scheduleOpt.loadPDServerConfig().UseRegionStorage, Equals, true)
+	c.Assert(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ][0].Key, Equals, "testKey")
+	c.Assert(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ][0].Value, Equals, "testValue")
+	c.Assert(s.svr.GetNamespaceConfig("testNS").LeaderScheduleLimit, Equals, uint64(200))
+
+	c.Assert(s.svr.DeleteNamespaceConfig("testNS"), IsNil)
+	c.Assert(s.svr.DeleteLabelProperty(typ, labelKey, labelValue), IsNil)
+
+	c.Assert(s.svr.GetNamespaceConfig("testNS").LeaderScheduleLimit, Equals, uint64(0))
+	c.Assert(len(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ]), Equals, 0)
+
+	//PUT GET failed
+	oldKV := s.svr.kv
+	s.svr.kv = core.NewKV(&testErrorKV{})
+	replicateCfg.MaxReplicas = 7
+	scheduleCfg.MaxSnapshotCount = 20
+	pdServerCfg.UseRegionStorage = false
+
+	c.Assert(s.svr.SetScheduleConfig(*scheduleCfg), NotNil)
+	c.Assert(s.svr.SetReplicationConfig(*replicateCfg), NotNil)
+	c.Assert(s.svr.SetPDServerConfig(*pdServerCfg), NotNil)
+	c.Assert(s.svr.SetLabelProperty(typ, labelKey, labelValue), NotNil)
+	c.Assert(s.svr.SetNamespaceConfig("testNS", nsConfig), NotNil)
+
+	c.Assert(s.svr.GetReplicationConfig().MaxReplicas, Equals, uint64(5))
+	c.Assert(s.svr.scheduleOpt.GetMaxSnapshotCount(), Equals, uint64(10))
+	c.Assert(s.svr.scheduleOpt.loadPDServerConfig().UseRegionStorage, Equals, true)
+	c.Assert(s.svr.GetNamespaceConfig("testNS").LeaderScheduleLimit, Equals, uint64(0))
+	c.Assert(len(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ]), Equals, 0)
+
+	//DELETE failed
+	s.svr.kv = oldKV
+	c.Assert(s.svr.SetNamespaceConfig("testNS", nsConfig), IsNil)
+	c.Assert(s.svr.SetReplicationConfig(*replicateCfg), IsNil)
+
+	s.svr.kv = core.NewKV(&testErrorKV{})
+	c.Assert(s.svr.DeleteLabelProperty(typ, labelKey, labelValue), NotNil)
+	c.Assert(s.svr.GetNamespaceConfig("testNS").LeaderScheduleLimit, Equals, uint64(200))
+	c.Assert(s.svr.DeleteNamespaceConfig("testNS"), NotNil)
+
+	c.Assert(s.svr.GetNamespaceConfig("testNS").LeaderScheduleLimit, Equals, uint64(200))
+	c.Assert(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ][0].Key, Equals, "testKey")
+	c.Assert(s.svr.scheduleOpt.loadLabelPropertyConfig()[typ][0].Value, Equals, "testValue")
+}

server/option.go

@@ -15,6 +15,7 @@ package server
 
 import (
 	"reflect"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -28,7 +29,7 @@ import (
 type scheduleOption struct {
 	v              atomic.Value
 	rep            *Replication
-	ns             map[string]*namespaceOption
+	ns             sync.Map // concurrent map[string]*namespaceOption
 	labelProperty  atomic.Value
 	clusterVersion atomic.Value
 	pdServerConfig atomic.Value
@@ -37,10 +38,10 @@ type scheduleOption struct {
 func newScheduleOption(cfg *Config) *scheduleOption {
 	o := &scheduleOption{}
 	o.store(&cfg.Schedule)
-	o.ns = make(map[string]*namespaceOption)
+	o.ns = sync.Map{}
 	for name, nsCfg := range cfg.Namespace {
 		nsCfg := nsCfg
-		o.ns[name] = newNamespaceOption(&nsCfg)
+		o.ns.Store(name, newNamespaceOption(&nsCfg))
 	}
 	o.rep = newReplication(&cfg.Replication)
 	o.pdServerConfig.Store(&cfg.PDServerCfg)
@@ -61,8 +62,17 @@ func (o *scheduleOption) GetReplication() *Replication {
 	return o.rep
 }
 
+func (o *scheduleOption) getNS(name string) (*namespaceOption, bool) {
+	if n, ok := o.ns.Load(name); ok {
+		if n, ok := n.(*namespaceOption); ok {
+			return n, true
+		}
+	}
+	return nil, false
+}
+
 func (o *scheduleOption) GetMaxReplicas(name string) int {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetMaxReplicas()
 	}
 	return o.rep.GetMaxReplicas()
@@ -105,35 +115,35 @@ func (o *scheduleOption) GetMaxStoreDownTime() time.Duration {
 }
 
 func (o *scheduleOption) GetLeaderScheduleLimit(name string) uint64 {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetLeaderScheduleLimit()
 	}
 	return o.load().LeaderScheduleLimit
 }
 
 func (o *scheduleOption) GetRegionScheduleLimit(name string) uint64 {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetRegionScheduleLimit()
 	}
 	return o.load().RegionScheduleLimit
 }
 
 func (o *scheduleOption) GetReplicaScheduleLimit(name string) uint64 {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetReplicaScheduleLimit()
 	}
 	return o.load().ReplicaScheduleLimit
 }
 
 func (o *scheduleOption) GetMergeScheduleLimit(name string) uint64 {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetMergeScheduleLimit()
 	}
 	return o.load().MergeScheduleLimit
 }
 
 func (o *scheduleOption) GetHotRegionScheduleLimit(name string) uint64 {
-	if n, ok := o.ns[name]; ok {
+	if n, ok := o.getNS(name); ok {
 		return n.GetHotRegionScheduleLimit()
 	}
 	return o.load().HotRegionScheduleLimit
@@ -273,9 +283,21 @@ func (o *scheduleOption) loadPDServerConfig() *PDServerConfig {
 
 func (o *scheduleOption) persist(kv *core.KV) error {
 	namespaces := make(map[string]NamespaceConfig)
-	for name, ns := range o.ns {
-		namespaces[name] = *ns.load()
+
+	f := func(k, v interface{}) bool {
+		var kstr string
+		var ok bool
+		if kstr, ok = k.(string); !ok {
+			return false
+		}
+		if ns, ok := v.(*namespaceOption); ok {
+			namespaces[kstr] = *ns.load()
+			return true
+		}
+		return false
 	}
+	o.ns.Range(f)
+
 	cfg := &Config{
 		Schedule:       *o.load(),
 		Replication:    *o.rep.load(),
@@ -290,9 +312,21 @@ func (o *scheduleOption) persist(kv *core.KV) error {
 
 func (o *scheduleOption) reload(kv *core.KV) error {
 	namespaces := make(map[string]NamespaceConfig)
-	for name, ns := range o.ns {
-		namespaces[name] = *ns.load()
+
+	f := func(k, v interface{}) bool {
+		var kstr string
+		var ok bool
+		if kstr, ok = k.(string); !ok {
+			return false
+		}
+		if ns, ok := v.(*namespaceOption); ok {
+			namespaces[kstr] = *ns.load()
+			return true
+		}
+		return false
 	}
+	o.ns.Range(f)
+
 	cfg := &Config{
 		Schedule:       *o.load().clone(),
 		Replication:    *o.rep.load(),
@@ -311,7 +345,7 @@ func (o *scheduleOption) reload(kv *core.KV) error {
 		o.rep.store(&cfg.Replication)
 		for name, nsCfg := range cfg.Namespace {
 			nsCfg := nsCfg
-			o.ns[name] = newNamespaceOption(&nsCfg)
+			o.ns.Store(name, newNamespaceOption(&nsCfg))
 		}
 		o.labelProperty.Store(cfg.LabelProperty)
 		o.clusterVersion.Store(cfg.ClusterVersion)