From 2ab6151fd5737ba9671f72bb6cf71b2f57e7eea7 Mon Sep 17 00:00:00 2001
From: okJiang <819421878@qq.com>
Date: Wed, 14 Aug 2024 17:49:21 +0800
Subject: [PATCH] deprecate the `disable` in old scheduler config

Signed-off-by: okJiang <819421878@qq.com>
---
 pkg/mcs/scheduling/server/config/config.go    | 12 ----
 pkg/mcs/scheduling/server/config/watcher.go   |  1 +
 pkg/schedule/config/config_provider.go        |  1 -
 pkg/schedule/coordinator.go                   | 22 +++----
 pkg/schedule/diagnostic/diagnostic_manager.go |  4 +-
 pkg/schedule/handler/handler.go               | 12 +---
 pkg/schedule/schedulers/balance_leader.go     | 21 +++++--
 pkg/schedule/schedulers/balance_region.go     | 19 +++++-
 pkg/schedule/schedulers/balance_witness.go    |  4 +-
 pkg/schedule/schedulers/base_scheduler.go     | 19 +++++-
 pkg/schedule/schedulers/config.go             | 61 ++++++++++++++++++-
 pkg/schedule/schedulers/config_test.go        | 23 +++++++
 pkg/schedule/schedulers/evict_leader.go       |  7 +--
 pkg/schedule/schedulers/evict_slow_store.go   | 26 ++++++--
 pkg/schedule/schedulers/evict_slow_trend.go   |  7 +--
 pkg/schedule/schedulers/grant_hot_region.go   |  6 +-
 pkg/schedule/schedulers/grant_leader.go       | 10 +--
 pkg/schedule/schedulers/hot_region.go         | 25 ++++++--
 pkg/schedule/schedulers/hot_region_config.go  |  6 +-
 pkg/schedule/schedulers/init.go               | 54 ++++++++++++----
 pkg/schedule/schedulers/label.go              |  4 +-
 pkg/schedule/schedulers/random_merge.go       |  4 +-
 pkg/schedule/schedulers/scatter_range.go      |  4 +-
 pkg/schedule/schedulers/scheduler.go          | 10 ++-
 .../schedulers/scheduler_controller.go        | 21 +++----
 pkg/schedule/schedulers/shuffle_hot_region.go |  6 +-
 pkg/schedule/schedulers/shuffle_leader.go     |  4 +-
 pkg/schedule/schedulers/shuffle_region.go     |  2 +-
 .../schedulers/shuffle_region_config.go       |  2 -
 pkg/schedule/schedulers/split_bucket.go       |  4 +-
 .../schedulers/transfer_witness_leader.go     |  2 +-
 pkg/storage/kv/mem_kv.go                      |  2 +-
 pkg/utils/syncutil/mutex_sync.go              | 10 +++
 server/config/persist_options.go              | 14 +----
 .../mcs/scheduling/server_test.go             |  4 +-
 35 files changed, 294 insertions(+), 139 deletions(-)

diff --git a/pkg/mcs/scheduling/server/config/config.go b/pkg/mcs/scheduling/server/config/config.go
index 4b855d09899..93aa7b45322 100644
--- a/pkg/mcs/scheduling/server/config/config.go
+++ b/pkg/mcs/scheduling/server/config/config.go
@@ -647,18 +647,6 @@ func (o *PersistConfig) SetMaxReplicas(replicas int) {
 	o.SetReplicationConfig(v)
 }
 
-// IsSchedulerDisabled returns if the scheduler is disabled.
-func (o *PersistConfig) IsSchedulerDisabled(tp types.CheckerSchedulerType) bool {
-	oldType := types.SchedulerTypeCompatibleMap[tp]
-	schedulers := o.GetScheduleConfig().Schedulers
-	for _, s := range schedulers {
-		if oldType == s.Type {
-			return s.Disable
-		}
-	}
-	return false
-}
-
 // SetPlacementRulesCacheEnabled sets if the placement rules cache is enabled.
 func (o *PersistConfig) SetPlacementRulesCacheEnabled(enabled bool) {
 	v := o.GetReplicationConfig().Clone()
diff --git a/pkg/mcs/scheduling/server/config/watcher.go b/pkg/mcs/scheduling/server/config/watcher.go
index d1ca99bd36d..4192b05399b 100644
--- a/pkg/mcs/scheduling/server/config/watcher.go
+++ b/pkg/mcs/scheduling/server/config/watcher.go
@@ -60,6 +60,7 @@ type Watcher struct {
 	*PersistConfig
 	// Some data, like the scheduler configs, should be loaded into the storage
 	// to make sure the coordinator could access them correctly.
+	// It is a memory storage.
 	storage storage.Storage
 	// schedulersController is used to trigger the scheduler's config reloading.
 	// Store as `*schedulers.Controller`.
diff --git a/pkg/schedule/config/config_provider.go b/pkg/schedule/config/config_provider.go
index d7bc38a7c03..1110ef357ea 100644
--- a/pkg/schedule/config/config_provider.go
+++ b/pkg/schedule/config/config_provider.go
@@ -50,7 +50,6 @@ type SchedulerConfigProvider interface {
 	SetSchedulingAllowanceStatus(bool, string)
 	GetStoresLimit() map[uint64]StoreLimitConfig
 
-	IsSchedulerDisabled(types.CheckerSchedulerType) bool
 	AddSchedulerCfg(types.CheckerSchedulerType, []string)
 	RemoveSchedulerCfg(types.CheckerSchedulerType)
 	Persist(endpoint.ConfigStorage) error
diff --git a/pkg/schedule/coordinator.go b/pkg/schedule/coordinator.go
index 89c99ac90b8..25c71fcbcf2 100644
--- a/pkg/schedule/coordinator.go
+++ b/pkg/schedule/coordinator.go
@@ -285,16 +285,16 @@ func (c *Coordinator) InitSchedulers(needRun bool) {
 			log.Error("the scheduler type not found", zap.String("scheduler-name", name), errs.ZapError(errs.ErrSchedulerNotFound))
 			continue
 		}
-		if cfg.Disable {
-			log.Info("skip create scheduler with independent configuration", zap.String("scheduler-name", name), zap.String("scheduler-type", cfg.Type), zap.Strings("scheduler-args", cfg.Args))
-			continue
-		}
 		s, err := schedulers.CreateScheduler(types.ConvertOldStrToType[cfg.Type], c.opController,
 			c.cluster.GetStorage(), schedulers.ConfigJSONDecoder([]byte(data)), c.schedulers.RemoveScheduler)
 		if err != nil {
 			log.Error("can not create scheduler with independent configuration", zap.String("scheduler-name", name), zap.Strings("scheduler-args", cfg.Args), errs.ZapError(err))
 			continue
 		}
+		if s.IsDisable() {
+			log.Info("skip create scheduler with independent configuration", zap.String("scheduler-name", name), zap.String("scheduler-type", cfg.Type), zap.Strings("scheduler-args", cfg.Args))
+			continue
+		}
 		if needRun {
 			log.Info("create scheduler with independent configuration", zap.String("scheduler-name", s.GetName()))
 			if err = c.schedulers.AddScheduler(s); err != nil {
@@ -313,13 +313,6 @@ func (c *Coordinator) InitSchedulers(needRun bool) {
 	// The old way to create the scheduler.
 	k := 0
 	for _, schedulerCfg := range scheduleCfg.Schedulers {
-		if schedulerCfg.Disable {
-			scheduleCfg.Schedulers[k] = schedulerCfg
-			k++
-			log.Info("skip create scheduler", zap.String("scheduler-type", schedulerCfg.Type), zap.Strings("scheduler-args", schedulerCfg.Args))
-			continue
-		}
-
 		tp := types.ConvertOldStrToType[schedulerCfg.Type]
 		s, err := schedulers.CreateScheduler(tp, c.opController,
 			c.cluster.GetStorage(), schedulers.ConfigSliceDecoder(tp, schedulerCfg.Args), c.schedulers.RemoveScheduler)
@@ -329,6 +322,13 @@ func (c *Coordinator) InitSchedulers(needRun bool) {
 			continue
 		}
 
+		if s.IsDisable() {
+			scheduleCfg.Schedulers[k] = schedulerCfg
+			k++
+			log.Info("skip create scheduler", zap.String("scheduler-type", schedulerCfg.Type), zap.Strings("scheduler-args", schedulerCfg.Args))
+			continue
+		}
+
 		if needRun {
 			log.Info("create scheduler", zap.String("scheduler-name", s.GetName()),
 				zap.Strings("scheduler-args", schedulerCfg.Args))
diff --git a/pkg/schedule/diagnostic/diagnostic_manager.go b/pkg/schedule/diagnostic/diagnostic_manager.go
index 5a56e849c3b..e150f74c1d3 100644
--- a/pkg/schedule/diagnostic/diagnostic_manager.go
+++ b/pkg/schedule/diagnostic/diagnostic_manager.go
@@ -48,8 +48,8 @@ func (d *Manager) GetDiagnosticResult(name string) (*schedulers.DiagnosticResult
 		res := &schedulers.DiagnosticResult{Name: name, Timestamp: ts, Status: schedulers.Disabled}
 		return res, nil
 	}
-	isDisabled := d.config.IsSchedulerDisabled(scheduler.Scheduler.GetType())
-	if isDisabled {
+
+	if scheduler.Scheduler.IsDisable() {
 		ts := uint64(time.Now().Unix())
 		res := &schedulers.DiagnosticResult{Name: name, Timestamp: ts, Status: schedulers.Disabled}
 		return res, nil
diff --git a/pkg/schedule/handler/handler.go b/pkg/schedule/handler/handler.go
index 748a17b87ef..b5e8629dd01 100644
--- a/pkg/schedule/handler/handler.go
+++ b/pkg/schedule/handler/handler.go
@@ -855,11 +855,7 @@ func (h *Handler) GetSchedulerByStatus(status string, needTS bool) (any, error)
 	case "disabled":
 		disabledSchedulers := make([]string, 0, len(schedulers))
 		for _, scheduler := range schedulers {
-			disabled, err := sc.IsSchedulerDisabled(scheduler)
-			if err != nil {
-				return nil, err
-			}
-			if disabled {
+			if sc.GetScheduler(scheduler).IsDisable() {
 				disabledSchedulers = append(disabledSchedulers, scheduler)
 			}
 		}
@@ -870,11 +866,7 @@ func (h *Handler) GetSchedulerByStatus(status string, needTS bool) (any, error)
 		// We should not return the disabled schedulers here.
 		enabledSchedulers := make([]string, 0, len(schedulers))
 		for _, scheduler := range schedulers {
-			disabled, err := sc.IsSchedulerDisabled(scheduler)
-			if err != nil {
-				return nil, err
-			}
-			if !disabled {
+			if !sc.GetScheduler(scheduler).IsDisable() {
 				enabledSchedulers = append(enabledSchedulers, scheduler)
 			}
 		}
diff --git a/pkg/schedule/schedulers/balance_leader.go b/pkg/schedule/schedulers/balance_leader.go
index ef254ee6950..3032f4557ff 100644
--- a/pkg/schedule/schedulers/balance_leader.go
+++ b/pkg/schedule/schedulers/balance_leader.go
@@ -33,7 +33,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/reflectutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
@@ -54,8 +53,7 @@ const (
 )
 
 type balanceLeaderSchedulerConfig struct {
-	syncutil.RWMutex
-	schedulerConfig
+	defaultSchedulerConfig
 
 	Ranges []core.KeyRange `json:"ranges"`
 	// Batch is used to generate multiple operators by one scheduling
@@ -166,7 +164,7 @@ type balanceLeaderScheduler struct {
 // each store balanced.
 func newBalanceLeaderScheduler(opController *operator.Controller, conf *balanceLeaderSchedulerConfig, options ...BalanceLeaderCreateOption) Scheduler {
 	s := &balanceLeaderScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.BalanceLeaderScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.BalanceLeaderScheduler, conf),
 		retryQuota:    newRetryQuota(),
 		conf:          conf,
 		handler:       newBalanceLeaderHandler(conf),
@@ -543,3 +541,18 @@ func (l *balanceLeaderScheduler) createOperator(solver *solver, collector *plan.
 	op.SetAdditionalInfo("targetScore", strconv.FormatFloat(solver.targetScore, 'f', 2, 64))
 	return op
 }
+
+// IsDiable implements the Scheduler interface.
+func (l *balanceLeaderScheduler) IsDisable() bool {
+	l.conf.RLock()
+	defer l.conf.RUnlock()
+	return l.conf.isDisable()
+}
+
+// SetDiable implements the Scheduler interface.
+func (l *balanceLeaderScheduler) SetDisable(disable bool) {
+	l.conf.RLock()
+	defer l.conf.RUnlock()
+	l.conf.setDisable(disable)
+	l.conf.save()
+}
diff --git a/pkg/schedule/schedulers/balance_region.go b/pkg/schedule/schedulers/balance_region.go
index 174b6af1c83..b5599b74916 100644
--- a/pkg/schedule/schedulers/balance_region.go
+++ b/pkg/schedule/schedulers/balance_region.go
@@ -36,6 +36,8 @@ const (
 )
 
 type balanceRegionSchedulerConfig struct {
+	defaultSchedulerConfig
+
 	Ranges []core.KeyRange `json:"ranges"`
 	// TODO: When we prepare to use Ranges, we will need to implement the ReloadConfig function for this scheduler.
 }
@@ -53,7 +55,7 @@ type balanceRegionScheduler struct {
 // each store balanced.
 func newBalanceRegionScheduler(opController *operator.Controller, conf *balanceRegionSchedulerConfig, opts ...BalanceRegionCreateOption) Scheduler {
 	scheduler := &balanceRegionScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.BalanceRegionScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.BalanceRegionScheduler, conf),
 		retryQuota:    newRetryQuota(),
 		name:          types.BalanceRegionScheduler.String(),
 		conf:          conf,
@@ -271,3 +273,18 @@ func (s *balanceRegionScheduler) transferPeer(solver *solver, collector *plan.Co
 	}
 	return nil
 }
+
+// IsDiable implements the Scheduler interface.
+func (s *balanceRegionScheduler) IsDisable() bool {
+	s.conf.Lock()
+	defer s.conf.Unlock()
+	return s.conf.isDisable()
+}
+
+// SetDiable implements the Scheduler interface.
+func (s *balanceRegionScheduler) SetDisable(disable bool) {
+	s.conf.Lock()
+	defer s.conf.Unlock()
+	s.conf.setDisable(disable)
+	s.conf.save()
+}
diff --git a/pkg/schedule/schedulers/balance_witness.go b/pkg/schedule/schedulers/balance_witness.go
index 450c43647cf..0377e026f52 100644
--- a/pkg/schedule/schedulers/balance_witness.go
+++ b/pkg/schedule/schedulers/balance_witness.go
@@ -34,7 +34,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/reflectutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -51,7 +50,6 @@ const (
 )
 
 type balanceWitnessSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	Ranges []core.KeyRange `json:"ranges"`
@@ -164,7 +162,7 @@ type balanceWitnessScheduler struct {
 // each store balanced.
 func newBalanceWitnessScheduler(opController *operator.Controller, conf *balanceWitnessSchedulerConfig, options ...BalanceWitnessCreateOption) Scheduler {
 	s := &balanceWitnessScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.BalanceWitnessScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.BalanceWitnessScheduler, conf),
 		retryQuota:    newRetryQuota(),
 		conf:          conf,
 		handler:       newBalanceWitnessHandler(conf),
diff --git a/pkg/schedule/schedulers/base_scheduler.go b/pkg/schedule/schedulers/base_scheduler.go
index b3dae9856e6..1a4857946e3 100644
--- a/pkg/schedule/schedulers/base_scheduler.go
+++ b/pkg/schedule/schedulers/base_scheduler.go
@@ -65,11 +65,16 @@ type BaseScheduler struct {
 
 	name string
 	tp   types.CheckerSchedulerType
+	conf schedulerConfig
 }
 
 // NewBaseScheduler returns a basic scheduler
-func NewBaseScheduler(opController *operator.Controller, tp types.CheckerSchedulerType) *BaseScheduler {
-	return &BaseScheduler{OpController: opController, tp: tp}
+func NewBaseScheduler(
+	opController *operator.Controller,
+	tp types.CheckerSchedulerType,
+	conf schedulerConfig,
+) *BaseScheduler {
+	return &BaseScheduler{OpController: opController, tp: tp, conf: conf}
 }
 
 func (*BaseScheduler) ServeHTTP(w http.ResponseWriter, _ *http.Request) {
@@ -100,7 +105,9 @@ func (*BaseScheduler) GetNextInterval(interval time.Duration) time.Duration {
 func (*BaseScheduler) PrepareConfig(sche.SchedulerCluster) error { return nil }
 
 // CleanConfig does some cleanup work about config.
-func (*BaseScheduler) CleanConfig(sche.SchedulerCluster) {}
+func (s *BaseScheduler) CleanConfig(sche.SchedulerCluster) error {
+	return s.conf.clean()
+}
 
 // GetName returns the name of the scheduler
 func (s *BaseScheduler) GetName() string {
@@ -114,3 +121,9 @@ func (s *BaseScheduler) GetName() string {
 func (s *BaseScheduler) GetType() types.CheckerSchedulerType {
 	return s.tp
 }
+
+// IsDiable implements the Scheduler interface.
+func (*BaseScheduler) IsDisable() bool { return false }
+
+// SetDisable implements the Scheduler interface.
+func (*BaseScheduler) SetDisable(bool) {}
diff --git a/pkg/schedule/schedulers/config.go b/pkg/schedule/schedulers/config.go
index 0c7caf686c3..c146261609e 100644
--- a/pkg/schedule/schedulers/config.go
+++ b/pkg/schedule/schedulers/config.go
@@ -18,18 +18,28 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/tikv/pd/pkg/storage/endpoint"
+	"github.com/tikv/pd/pkg/utils/syncutil"
 )
 
 type schedulerConfig interface {
+	syncutil.RWMutexInterface
+
+	init(name string, storage endpoint.ConfigStorage, data any)
 	save() error
 	load(any) error
-	init(name string, storage endpoint.ConfigStorage, data any)
+	clean() error
+	// setArgs([]string)
+	// getArgs() []string
 }
 
 type baseSchedulerConfig struct {
+	syncutil.RWMutex
+
 	name    string
 	storage endpoint.ConfigStorage
 
+	// Args is the input arguments of the scheduler.
+	// Args []string `json:"args"`
 	// data is the config of the scheduler.
 	data any
 }
@@ -58,3 +68,52 @@ func (b *baseSchedulerConfig) load(v any) error {
 	}
 	return DecodeConfig([]byte(data), v)
 }
+
+func (b *baseSchedulerConfig) clean() error {
+	return b.storage.RemoveSchedulerConfig(b.name)
+}
+
+// func (b *baseSchedulerConfig) setArgs(args []string) {
+// 	b.Args = args
+// }
+
+// func (b *baseSchedulerConfig) getArgs() []string {
+// 	return b.Args
+// }
+
+type defaultSchedulerConfig interface {
+	schedulerConfig
+
+	isDisable() bool
+	setDisable(bool)
+}
+
+type baseDefaultSchedulerConfig struct {
+	schedulerConfig
+
+	Disabled bool `json:"disabled"`
+}
+
+func newBaseDefaultSchedulerConfig() *baseDefaultSchedulerConfig {
+	return &baseDefaultSchedulerConfig{
+		schedulerConfig: &baseSchedulerConfig{},
+	}
+}
+
+func (b *baseDefaultSchedulerConfig) isDisable() bool {
+	b.load(b)
+	return b.Disabled
+}
+
+func (b *baseDefaultSchedulerConfig) setDisable(disabled bool) {
+	b.Lock()
+	defer b.Unlock()
+	b.Disabled = disabled
+}
+
+func (b *baseDefaultSchedulerConfig) clean() error {
+	b.Lock()
+	defer b.Unlock()
+	b.Disabled = false
+	return b.save()
+}
diff --git a/pkg/schedule/schedulers/config_test.go b/pkg/schedule/schedulers/config_test.go
index 31858bd7c10..0d2915f168b 100644
--- a/pkg/schedule/schedulers/config_test.go
+++ b/pkg/schedule/schedulers/config_test.go
@@ -48,3 +48,26 @@ func TestSchedulerConfig(t *testing.T) {
 	// report error because the config is empty and cannot be decoded
 	require.Error(t, cfg2.load(newTc))
 }
+
+func TestDefaultSchedulerConfig(t *testing.T) {
+	s := storage.NewStorageWithMemoryBackend()
+
+	type testConfig struct {
+		defaultSchedulerConfig
+		Value string `json:"value"`
+	}
+
+	cfg := &testConfig{
+		defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
+	}
+	cfg.init("test", s, cfg)
+	require.False(t, cfg.isDisable())
+	cfg.setDisable(true)
+	require.True(t, cfg.isDisable())
+
+	cfg2 := &testConfig{
+		defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
+	}
+	cfg.init("test", s, cfg2)
+	require.True(t, cfg.isDisable())
+}
diff --git a/pkg/schedule/schedulers/evict_leader.go b/pkg/schedule/schedulers/evict_leader.go
index 85f861f0082..a151ac6be7e 100644
--- a/pkg/schedule/schedulers/evict_leader.go
+++ b/pkg/schedule/schedulers/evict_leader.go
@@ -30,7 +30,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -45,7 +44,6 @@ const (
 )
 
 type evictLeaderSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	StoreIDWithRanges map[uint64][]core.KeyRange `json:"store-id-ranges"`
@@ -233,7 +231,7 @@ type evictLeaderScheduler struct {
 func newEvictLeaderScheduler(opController *operator.Controller, conf *evictLeaderSchedulerConfig) Scheduler {
 	handler := newEvictLeaderHandler(conf)
 	return &evictLeaderScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.EvictLeaderScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.EvictLeaderScheduler, conf),
 		conf:          conf,
 		handler:       handler,
 	}
@@ -265,8 +263,9 @@ func (s *evictLeaderScheduler) PrepareConfig(cluster sche.SchedulerCluster) erro
 }
 
 // CleanConfig implements the Scheduler interface.
-func (s *evictLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) {
+func (s *evictLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) error {
 	s.conf.resumeLeaderTransfer(cluster)
+	return s.BaseScheduler.CleanConfig(cluster)
 }
 
 // IsScheduleAllowed implements the Scheduler interface.
diff --git a/pkg/schedule/schedulers/evict_slow_store.go b/pkg/schedule/schedulers/evict_slow_store.go
index de581f597bb..3f9a11a7b66 100644
--- a/pkg/schedule/schedulers/evict_slow_store.go
+++ b/pkg/schedule/schedulers/evict_slow_store.go
@@ -28,7 +28,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -42,8 +41,7 @@ const (
 )
 
 type evictSlowStoreSchedulerConfig struct {
-	syncutil.RWMutex
-	schedulerConfig
+	defaultSchedulerConfig
 
 	cluster *core.BasicCluster
 	// Last timestamp of the chosen slow store for eviction.
@@ -55,7 +53,7 @@ type evictSlowStoreSchedulerConfig struct {
 
 func initEvictSlowStoreSchedulerConfig() *evictSlowStoreSchedulerConfig {
 	return &evictSlowStoreSchedulerConfig{
-		schedulerConfig:        &baseSchedulerConfig{},
+		defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
 		lastSlowStoreCaptureTS: time.Time{},
 		RecoveryDurationGap:    defaultRecoveryDurationGap,
 		EvictedStores:          make([]uint64, 0),
@@ -220,8 +218,9 @@ func (s *evictSlowStoreScheduler) PrepareConfig(cluster sche.SchedulerCluster) e
 }
 
 // CleanConfig implements the Scheduler interface.
-func (s *evictSlowStoreScheduler) CleanConfig(cluster sche.SchedulerCluster) {
+func (s *evictSlowStoreScheduler) CleanConfig(cluster sche.SchedulerCluster) error {
 	s.cleanupEvictLeader(cluster)
+	return s.BaseScheduler.CleanConfig(cluster)
 }
 
 func (s *evictSlowStoreScheduler) prepareEvictLeader(cluster sche.SchedulerCluster, storeID uint64) error {
@@ -313,11 +312,26 @@ func (s *evictSlowStoreScheduler) Schedule(cluster sche.SchedulerCluster, _ bool
 	return s.schedulerEvictLeader(cluster), nil
 }
 
+// IsDiable implements the Scheduler interface.
+func (s *evictSlowStoreScheduler) IsDisable() bool {
+	s.conf.Lock()
+	defer s.conf.Unlock()
+	return s.conf.isDisable()
+}
+
+// SetDisable implements the Scheduler interface.
+func (s *evictSlowStoreScheduler) SetDisable(disable bool) {
+	s.conf.Lock()
+	defer s.conf.Unlock()
+	s.conf.setDisable(disable)
+	s.conf.save()
+}
+
 // newEvictSlowStoreScheduler creates a scheduler that detects and evicts slow stores.
 func newEvictSlowStoreScheduler(opController *operator.Controller, conf *evictSlowStoreSchedulerConfig) Scheduler {
 	handler := newEvictSlowStoreHandler(conf)
 	return &evictSlowStoreScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.EvictSlowStoreScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.EvictSlowStoreScheduler, conf),
 		conf:          conf,
 		handler:       handler,
 	}
diff --git a/pkg/schedule/schedulers/evict_slow_trend.go b/pkg/schedule/schedulers/evict_slow_trend.go
index 427787016a2..e9bb171e853 100644
--- a/pkg/schedule/schedulers/evict_slow_trend.go
+++ b/pkg/schedule/schedulers/evict_slow_trend.go
@@ -29,7 +29,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -52,7 +51,6 @@ type slowCandidate struct {
 }
 
 type evictSlowTrendSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	cluster *core.BasicCluster
@@ -322,8 +320,9 @@ func (s *evictSlowTrendScheduler) PrepareConfig(cluster sche.SchedulerCluster) e
 }
 
 // CleanConfig implements the Scheduler interface.
-func (s *evictSlowTrendScheduler) CleanConfig(cluster sche.SchedulerCluster) {
+func (s *evictSlowTrendScheduler) CleanConfig(cluster sche.SchedulerCluster) error {
 	s.cleanupEvictLeader(cluster)
+	return s.BaseScheduler.CleanConfig(cluster)
 }
 
 func (s *evictSlowTrendScheduler) prepareEvictLeader(cluster sche.SchedulerCluster, storeID uint64) error {
@@ -440,7 +439,7 @@ func (s *evictSlowTrendScheduler) Schedule(cluster sche.SchedulerCluster, _ bool
 func newEvictSlowTrendScheduler(opController *operator.Controller, conf *evictSlowTrendSchedulerConfig) Scheduler {
 	handler := newEvictSlowTrendHandler(conf)
 	sche := &evictSlowTrendScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.EvictSlowTrendScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.EvictSlowTrendScheduler, conf),
 		conf:          conf,
 		handler:       handler,
 	}
diff --git a/pkg/schedule/schedulers/grant_hot_region.go b/pkg/schedule/schedulers/grant_hot_region.go
index a441f41062a..990011a5666 100644
--- a/pkg/schedule/schedulers/grant_hot_region.go
+++ b/pkg/schedule/schedulers/grant_hot_region.go
@@ -35,7 +35,6 @@ import (
 	"github.com/tikv/pd/pkg/statistics"
 	"github.com/tikv/pd/pkg/statistics/utils"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -46,7 +45,6 @@ const (
 )
 
 type grantHotRegionSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	cluster       *core.BasicCluster
@@ -121,8 +119,8 @@ type grantHotRegionScheduler struct {
 
 // newGrantHotRegionScheduler creates an admin scheduler that transfers hot region peer to fixed store and hot region leader to one store.
 func newGrantHotRegionScheduler(opController *operator.Controller, conf *grantHotRegionSchedulerConfig) *grantHotRegionScheduler {
-	base := newBaseHotScheduler(opController,
-		statistics.DefaultHistorySampleDuration, statistics.DefaultHistorySampleInterval)
+	base := newBaseHotScheduler(opController, statistics.DefaultHistorySampleDuration,
+		statistics.DefaultHistorySampleInterval, conf)
 	base.tp = types.GrantHotRegionScheduler
 	handler := newGrantHotRegionHandler(conf)
 	ret := &grantHotRegionScheduler{
diff --git a/pkg/schedule/schedulers/grant_leader.go b/pkg/schedule/schedulers/grant_leader.go
index d70c0b19d67..d48cfddbae0 100644
--- a/pkg/schedule/schedulers/grant_leader.go
+++ b/pkg/schedule/schedulers/grant_leader.go
@@ -30,7 +30,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -41,7 +40,6 @@ const (
 )
 
 type grantLeaderSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	StoreIDWithRanges map[uint64][]core.KeyRange `json:"store-id-ranges"`
@@ -149,7 +147,7 @@ type grantLeaderScheduler struct {
 // newGrantLeaderScheduler creates an admin scheduler that transfers all leaders
 // to a store.
 func newGrantLeaderScheduler(opController *operator.Controller, conf *grantLeaderSchedulerConfig) Scheduler {
-	base := NewBaseScheduler(opController, types.GrantLeaderScheduler)
+	base := NewBaseScheduler(opController, types.GrantLeaderScheduler, conf)
 	handler := newGrantLeaderHandler(conf)
 	return &grantLeaderScheduler{
 		BaseScheduler: base,
@@ -195,12 +193,14 @@ func (s *grantLeaderScheduler) PrepareConfig(cluster sche.SchedulerCluster) erro
 }
 
 // CleanConfig implements the Scheduler interface.
-func (s *grantLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) {
+func (s *grantLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) error {
 	s.conf.RLock()
-	defer s.conf.RUnlock()
 	for id := range s.conf.StoreIDWithRanges {
 		cluster.ResumeLeaderTransfer(id)
 	}
+	// BaseScheduler.CleanConfig will also lock the config, so we need to unlock it first.
+	s.conf.RUnlock()
+	return s.BaseScheduler.CleanConfig(cluster)
 }
 
 // IsScheduleAllowed implements the Scheduler interface.
diff --git a/pkg/schedule/schedulers/hot_region.go b/pkg/schedule/schedulers/hot_region.go
index ab595ec9058..30c124dbd84 100644
--- a/pkg/schedule/schedulers/hot_region.go
+++ b/pkg/schedule/schedulers/hot_region.go
@@ -90,8 +90,12 @@ type baseHotScheduler struct {
 	updateWriteTime time.Time
 }
 
-func newBaseHotScheduler(opController *operator.Controller, sampleDuration time.Duration, sampleInterval time.Duration) *baseHotScheduler {
-	base := NewBaseScheduler(opController, types.BalanceHotRegionScheduler)
+func newBaseHotScheduler(
+	opController *operator.Controller,
+	sampleDuration, sampleInterval time.Duration,
+	schedulerConfig schedulerConfig,
+) *baseHotScheduler {
+	base := NewBaseScheduler(opController, types.BalanceHotRegionScheduler, schedulerConfig)
 	ret := &baseHotScheduler{
 		BaseScheduler:  base,
 		regionPendings: make(map[uint64]*pendingInfluence),
@@ -200,8 +204,8 @@ type hotScheduler struct {
 }
 
 func newHotScheduler(opController *operator.Controller, conf *hotRegionSchedulerConfig) *hotScheduler {
-	base := newBaseHotScheduler(opController,
-		conf.getHistorySampleDuration(), conf.getHistorySampleInterval())
+	base := newBaseHotScheduler(opController, conf.getHistorySampleDuration(),
+		conf.getHistorySampleInterval(), conf)
 	ret := &hotScheduler{
 		name:             HotRegionName,
 		baseHotScheduler: base,
@@ -284,6 +288,19 @@ func (h *hotScheduler) Schedule(cluster sche.SchedulerCluster, _ bool) ([]*opera
 	return h.dispatch(typ, cluster), nil
 }
 
+// IsDisable implements the Scheduler interface.
+func (h *hotScheduler) IsDisable() bool {
+	return h.conf.isDisable()
+}
+
+// SetDisable implements the Scheduler interface.
+func (h *hotScheduler) SetDisable(disable bool) {
+	h.Lock()
+	defer h.Unlock()
+	h.conf.setDisable(disable)
+	h.conf.save()
+}
+
 func (h *hotScheduler) dispatch(typ resourceType, cluster sche.SchedulerCluster) []*operator.Operator {
 	h.Lock()
 	defer h.Unlock()
diff --git a/pkg/schedule/schedulers/hot_region_config.go b/pkg/schedule/schedulers/hot_region_config.go
index 0424a582bf4..2fb9a0993cb 100644
--- a/pkg/schedule/schedulers/hot_region_config.go
+++ b/pkg/schedule/schedulers/hot_region_config.go
@@ -29,7 +29,6 @@ import (
 	"github.com/tikv/pd/pkg/statistics"
 	"github.com/tikv/pd/pkg/statistics/utils"
 	"github.com/tikv/pd/pkg/utils/reflectutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"github.com/tikv/pd/pkg/versioninfo"
 	"github.com/unrolled/render"
@@ -58,7 +57,7 @@ var compatiblePrioritiesConfig = prioritiesConfig{
 // params about hot region.
 func initHotRegionScheduleConfig() *hotRegionSchedulerConfig {
 	cfg := &hotRegionSchedulerConfig{
-		schedulerConfig:        &baseSchedulerConfig{},
+		defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
 		MinHotByteRate:         100,
 		MinHotKeyRate:          10,
 		MinHotQueryRate:        10,
@@ -113,8 +112,7 @@ func (conf *hotRegionSchedulerConfig) getValidConf() *hotRegionSchedulerConfig {
 }
 
 type hotRegionSchedulerConfig struct {
-	syncutil.RWMutex
-	schedulerConfig
+	defaultSchedulerConfig
 
 	lastQuerySupported bool
 
diff --git a/pkg/schedule/schedulers/init.go b/pkg/schedule/schedulers/init.go
index 16f78284cf8..566be41e764 100644
--- a/pkg/schedule/schedulers/init.go
+++ b/pkg/schedule/schedulers/init.go
@@ -49,6 +49,7 @@ func schedulersRegister() {
 			}
 			conf.Ranges = ranges
 			conf.Batch = BalanceLeaderBatchSize
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -56,7 +57,7 @@ func schedulersRegister() {
 	RegisterScheduler(types.BalanceLeaderScheduler, func(opController *operator.Controller,
 		storage endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
 		conf := &balanceLeaderSchedulerConfig{
-			schedulerConfig: &baseSchedulerConfig{},
+			defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
 		}
 		if err := decoder(conf); err != nil {
 			return nil, err
@@ -81,17 +82,22 @@ func schedulersRegister() {
 				return err
 			}
 			conf.Ranges = ranges
+			// conf.setArgs(args)
 			return nil
 		}
 	})
 
 	RegisterScheduler(types.BalanceRegionScheduler, func(opController *operator.Controller,
-		_ endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
-		conf := &balanceRegionSchedulerConfig{}
+		storage endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
+		conf := &balanceRegionSchedulerConfig{
+			defaultSchedulerConfig: newBaseDefaultSchedulerConfig(),
+		}
 		if err := decoder(conf); err != nil {
 			return nil, err
 		}
-		return newBalanceRegionScheduler(opController, conf), nil
+		sche := newBalanceRegionScheduler(opController, conf)
+		conf.init(sche.GetName(), storage, conf)
+		return sche, nil
 	})
 
 	// balance witness
@@ -107,6 +113,7 @@ func schedulersRegister() {
 			}
 			conf.Ranges = ranges
 			conf.Batch = balanceWitnessBatchSize
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -149,6 +156,7 @@ func schedulersRegister() {
 			}
 			conf.StoreIDWithRanges[id] = ranges
 			conf.Batch = EvictLeaderBatchSize
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -215,6 +223,7 @@ func schedulersRegister() {
 			if !conf.setStore(leaderID, storeIDs) {
 				return errs.ErrSchedulerConfig
 			}
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -284,6 +293,7 @@ func schedulersRegister() {
 				return err
 			}
 			conf.StoreIDWithRanges[id] = ranges
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -316,17 +326,22 @@ func schedulersRegister() {
 				return err
 			}
 			conf.Ranges = ranges
+			// conf.setArgs(args)
 			return nil
 		}
 	})
 
 	RegisterScheduler(types.LabelScheduler, func(opController *operator.Controller,
-		_ endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
-		conf := &labelSchedulerConfig{}
+		storage endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
+		conf := &labelSchedulerConfig{
+			schedulerConfig: &baseSchedulerConfig{},
+		}
 		if err := decoder(conf); err != nil {
 			return nil, err
 		}
-		return newLabelScheduler(opController, conf), nil
+		sche := newLabelScheduler(opController, conf)
+		conf.init(sche.GetName(), storage, conf)
+		return sche, nil
 	})
 
 	// random merge
@@ -341,17 +356,22 @@ func schedulersRegister() {
 				return err
 			}
 			conf.Ranges = ranges
+			// conf.setArgs(args)
 			return nil
 		}
 	})
 
 	RegisterScheduler(types.RandomMergeScheduler, func(opController *operator.Controller,
-		_ endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
-		conf := &randomMergeSchedulerConfig{}
+		storage endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
+		conf := &randomMergeSchedulerConfig{
+			schedulerConfig: &baseSchedulerConfig{},
+		}
 		if err := decoder(conf); err != nil {
 			return nil, err
 		}
-		return newRandomMergeScheduler(opController, conf), nil
+		sche := newRandomMergeScheduler(opController, conf)
+		conf.init(sche.GetName(), storage, conf)
+		return sche, nil
 	})
 
 	// scatter range
@@ -371,6 +391,7 @@ func schedulersRegister() {
 			conf.StartKey = args[0]
 			conf.EndKey = args[1]
 			conf.RangeName = args[2]
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -407,6 +428,7 @@ func schedulersRegister() {
 				}
 				conf.Limit = limit
 			}
+			// conf.setArgs(args)
 			return nil
 		}
 	})
@@ -437,17 +459,22 @@ func schedulersRegister() {
 				return err
 			}
 			conf.Ranges = ranges
+			// conf.setArgs(args)
 			return nil
 		}
 	})
 
 	RegisterScheduler(types.ShuffleLeaderScheduler, func(opController *operator.Controller,
-		_ endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
-		conf := &shuffleLeaderSchedulerConfig{}
+		storage endpoint.ConfigStorage, decoder ConfigDecoder, _ ...func(string) error) (Scheduler, error) {
+		conf := &shuffleLeaderSchedulerConfig{
+			schedulerConfig: &baseSchedulerConfig{},
+		}
 		if err := decoder(conf); err != nil {
 			return nil, err
 		}
-		return newShuffleLeaderScheduler(opController, conf), nil
+		sche := newShuffleLeaderScheduler(opController, conf)
+		conf.init(sche.GetName(), storage, conf)
+		return sche, nil
 	})
 
 	// shuffle region
@@ -463,6 +490,7 @@ func schedulersRegister() {
 			}
 			conf.Ranges = ranges
 			conf.Roles = allRoles
+			// conf.setArgs(args)
 			return nil
 		}
 	})
diff --git a/pkg/schedule/schedulers/label.go b/pkg/schedule/schedulers/label.go
index 8d4f42262ac..6050d6d6105 100644
--- a/pkg/schedule/schedulers/label.go
+++ b/pkg/schedule/schedulers/label.go
@@ -34,6 +34,8 @@ const (
 )
 
 type labelSchedulerConfig struct {
+	schedulerConfig
+
 	Ranges []core.KeyRange `json:"ranges"`
 	// TODO: When we prepare to use Ranges, we will need to implement the ReloadConfig function for this scheduler.
 }
@@ -48,7 +50,7 @@ type labelScheduler struct {
 // the store with the specific label.
 func newLabelScheduler(opController *operator.Controller, conf *labelSchedulerConfig) Scheduler {
 	return &labelScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.LabelScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.LabelScheduler, conf),
 		conf:          conf,
 	}
 }
diff --git a/pkg/schedule/schedulers/random_merge.go b/pkg/schedule/schedulers/random_merge.go
index 676e5407e72..71d29ac5ae5 100644
--- a/pkg/schedule/schedulers/random_merge.go
+++ b/pkg/schedule/schedulers/random_merge.go
@@ -35,6 +35,8 @@ const (
 )
 
 type randomMergeSchedulerConfig struct {
+	schedulerConfig
+
 	Ranges []core.KeyRange `json:"ranges"`
 	// TODO: When we prepare to use Ranges, we will need to implement the ReloadConfig function for this scheduler.
 }
@@ -47,7 +49,7 @@ type randomMergeScheduler struct {
 // newRandomMergeScheduler creates an admin scheduler that randomly picks two adjacent regions
 // then merges them.
 func newRandomMergeScheduler(opController *operator.Controller, conf *randomMergeSchedulerConfig) Scheduler {
-	base := NewBaseScheduler(opController, types.RandomMergeScheduler)
+	base := NewBaseScheduler(opController, types.RandomMergeScheduler, conf)
 	return &randomMergeScheduler{
 		BaseScheduler: base,
 		conf:          conf,
diff --git a/pkg/schedule/schedulers/scatter_range.go b/pkg/schedule/schedulers/scatter_range.go
index e86785fcc19..5377f755ee0 100644
--- a/pkg/schedule/schedulers/scatter_range.go
+++ b/pkg/schedule/schedulers/scatter_range.go
@@ -27,7 +27,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/plan"
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 )
 
@@ -37,7 +36,6 @@ const (
 )
 
 type scatterRangeSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	RangeName string `json:"range-name"`
@@ -108,7 +106,7 @@ type scatterRangeScheduler struct {
 
 // newScatterRangeScheduler creates a scheduler that balances the distribution of leaders and regions that in the specified key range.
 func newScatterRangeScheduler(opController *operator.Controller, config *scatterRangeSchedulerConfig) Scheduler {
-	base := NewBaseScheduler(opController, types.ScatterRangeScheduler)
+	base := NewBaseScheduler(opController, types.ScatterRangeScheduler, config)
 
 	handler := newScatterRangeHandler(config)
 	scheduler := &scatterRangeScheduler{
diff --git a/pkg/schedule/schedulers/scheduler.go b/pkg/schedule/schedulers/scheduler.go
index 7fce5d9c46e..ec88d83b007 100644
--- a/pkg/schedule/schedulers/scheduler.go
+++ b/pkg/schedule/schedulers/scheduler.go
@@ -44,9 +44,17 @@ type Scheduler interface {
 	GetMinInterval() time.Duration
 	GetNextInterval(interval time.Duration) time.Duration
 	PrepareConfig(cluster sche.SchedulerCluster) error
-	CleanConfig(cluster sche.SchedulerCluster)
+	CleanConfig(cluster sche.SchedulerCluster) error
 	Schedule(cluster sche.SchedulerCluster, dryRun bool) ([]*operator.Operator, []plan.Plan)
 	IsScheduleAllowed(cluster sche.SchedulerCluster) bool
+	// IsDiable returns if the scheduler is disabled, it only works for default schedulers.
+	// - BalanceRegionScheduler
+	// - BalanceLeaderScheduler
+	// - BalanceHotRegionScheduler
+	// - EvictSlowStoreScheduler
+	IsDisable() bool
+	// SetDisable sets the scheduler's disable, it only works for default schedulers.
+	SetDisable(bool)
 }
 
 // EncodeConfig encode the custom config for each scheduler.
diff --git a/pkg/schedule/schedulers/scheduler_controller.go b/pkg/schedule/schedulers/scheduler_controller.go
index c29b75a371a..2b4172ffc06 100644
--- a/pkg/schedule/schedulers/scheduler_controller.go
+++ b/pkg/schedule/schedulers/scheduler_controller.go
@@ -143,8 +143,10 @@ func (c *Controller) AddSchedulerHandler(scheduler Scheduler, args ...string) er
 	defer c.Unlock()
 
 	name := scheduler.GetName()
-	if _, ok := c.schedulerHandlers[name]; ok {
+	if _, ok := c.schedulerHandlers[name]; ok && !scheduler.IsDisable() {
 		return errs.ErrSchedulerExisted.FastGenByArgs()
+	} else if ok && scheduler.IsDisable() {
+		scheduler.SetDisable(false)
 	}
 
 	c.schedulerHandlers[name] = scheduler
@@ -176,11 +178,6 @@ func (c *Controller) RemoveSchedulerHandler(name string) error {
 		return err
 	}
 
-	if err := c.storage.RemoveSchedulerConfig(name); err != nil {
-		log.Error("can not remove the scheduler config", errs.ZapError(err))
-		return err
-	}
-
 	s.(Scheduler).CleanConfig(c.cluster)
 	delete(c.schedulerHandlers, name)
 
@@ -192,8 +189,10 @@ func (c *Controller) AddScheduler(scheduler Scheduler, args ...string) error {
 	c.Lock()
 	defer c.Unlock()
 
-	if _, ok := c.schedulers[scheduler.GetName()]; ok {
+	if s, ok := c.schedulers[scheduler.GetName()]; ok && !s.IsDisable() {
 		return errs.ErrSchedulerExisted.FastGenByArgs()
+	} else if ok && scheduler.IsDisable() {
+		scheduler.SetDisable(false)
 	}
 
 	s := NewScheduleController(c.ctx, c.cluster, c.opController, scheduler)
@@ -231,11 +230,7 @@ func (c *Controller) RemoveScheduler(name string) error {
 		return err
 	}
 
-	if err := c.storage.RemoveSchedulerConfig(name); err != nil {
-		log.Error("can not remove the scheduler config", errs.ZapError(err))
-		return err
-	}
-
+	s.CleanConfig(c.cluster)
 	s.Stop()
 	schedulerStatusGauge.DeleteLabelValues(name, "allow")
 	delete(c.schedulers, name)
@@ -321,7 +316,7 @@ func (c *Controller) IsSchedulerDisabled(name string) (bool, error) {
 	if !ok {
 		return false, errs.ErrSchedulerNotFound.FastGenByArgs()
 	}
-	return c.cluster.GetSchedulerConfig().IsSchedulerDisabled(s.Scheduler.GetType()), nil
+	return s.IsDisable(), nil
 }
 
 // IsSchedulerExisted returns whether a scheduler is existed.
diff --git a/pkg/schedule/schedulers/shuffle_hot_region.go b/pkg/schedule/schedulers/shuffle_hot_region.go
index f8544fff48d..594f0910c9b 100644
--- a/pkg/schedule/schedulers/shuffle_hot_region.go
+++ b/pkg/schedule/schedulers/shuffle_hot_region.go
@@ -29,7 +29,6 @@ import (
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/statistics"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 	"go.uber.org/zap"
 )
@@ -40,7 +39,6 @@ const (
 )
 
 type shuffleHotRegionSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	Limit uint64 `json:"limit"`
@@ -72,8 +70,8 @@ type shuffleHotRegionScheduler struct {
 
 // newShuffleHotRegionScheduler creates an admin scheduler that random balance hot regions
 func newShuffleHotRegionScheduler(opController *operator.Controller, conf *shuffleHotRegionSchedulerConfig) Scheduler {
-	base := newBaseHotScheduler(opController,
-		statistics.DefaultHistorySampleDuration, statistics.DefaultHistorySampleInterval)
+	base := newBaseHotScheduler(opController, statistics.DefaultHistorySampleDuration,
+		statistics.DefaultHistorySampleInterval, conf)
 	base.tp = types.ShuffleHotRegionScheduler
 	handler := newShuffleHotRegionHandler(conf)
 	ret := &shuffleHotRegionScheduler{
diff --git a/pkg/schedule/schedulers/shuffle_leader.go b/pkg/schedule/schedulers/shuffle_leader.go
index 4270613667b..f283e996f8d 100644
--- a/pkg/schedule/schedulers/shuffle_leader.go
+++ b/pkg/schedule/schedulers/shuffle_leader.go
@@ -32,6 +32,8 @@ const (
 )
 
 type shuffleLeaderSchedulerConfig struct {
+	schedulerConfig
+
 	Ranges []core.KeyRange `json:"ranges"`
 	// TODO: When we prepare to use Ranges, we will need to implement the ReloadConfig function for this scheduler.
 }
@@ -45,7 +47,7 @@ type shuffleLeaderScheduler struct {
 // newShuffleLeaderScheduler creates an admin scheduler that shuffles leaders
 // between stores.
 func newShuffleLeaderScheduler(opController *operator.Controller, conf *shuffleLeaderSchedulerConfig) Scheduler {
-	base := NewBaseScheduler(opController, types.ShuffleLeaderScheduler)
+	base := NewBaseScheduler(opController, types.ShuffleLeaderScheduler, conf)
 	filters := []filter.Filter{
 		&filter.StoreStateFilter{ActionScope: base.GetName(), TransferLeader: true, OperatorLevel: constant.Low},
 		filter.NewSpecialUseFilter(base.GetName()),
diff --git a/pkg/schedule/schedulers/shuffle_region.go b/pkg/schedule/schedulers/shuffle_region.go
index 5d4c49e0fcc..7ab016f745a 100644
--- a/pkg/schedule/schedulers/shuffle_region.go
+++ b/pkg/schedule/schedulers/shuffle_region.go
@@ -45,7 +45,7 @@ func newShuffleRegionScheduler(opController *operator.Controller, conf *shuffleR
 		&filter.StoreStateFilter{ActionScope: ShuffleRegionName, MoveRegion: true, OperatorLevel: constant.Low},
 		filter.NewSpecialUseFilter(ShuffleRegionName),
 	}
-	base := NewBaseScheduler(opController, types.ShuffleRegionScheduler)
+	base := NewBaseScheduler(opController, types.ShuffleRegionScheduler, conf)
 	return &shuffleRegionScheduler{
 		BaseScheduler: base,
 		conf:          conf,
diff --git a/pkg/schedule/schedulers/shuffle_region_config.go b/pkg/schedule/schedulers/shuffle_region_config.go
index 2e3394a58df..ef341b7b770 100644
--- a/pkg/schedule/schedulers/shuffle_region_config.go
+++ b/pkg/schedule/schedulers/shuffle_region_config.go
@@ -22,7 +22,6 @@ import (
 	"github.com/tikv/pd/pkg/schedule/placement"
 	"github.com/tikv/pd/pkg/slice"
 	"github.com/tikv/pd/pkg/utils/apiutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 )
 
@@ -35,7 +34,6 @@ const (
 var allRoles = []string{roleLeader, roleFollower, roleLearner}
 
 type shuffleRegionSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 
 	Ranges []core.KeyRange `json:"ranges"`
diff --git a/pkg/schedule/schedulers/split_bucket.go b/pkg/schedule/schedulers/split_bucket.go
index 0d8fa614aef..f534caf516c 100644
--- a/pkg/schedule/schedulers/split_bucket.go
+++ b/pkg/schedule/schedulers/split_bucket.go
@@ -31,7 +31,6 @@ import (
 	types "github.com/tikv/pd/pkg/schedule/type"
 	"github.com/tikv/pd/pkg/statistics/buckets"
 	"github.com/tikv/pd/pkg/utils/reflectutil"
-	"github.com/tikv/pd/pkg/utils/syncutil"
 	"github.com/unrolled/render"
 )
 
@@ -52,7 +51,6 @@ func initSplitBucketConfig() *splitBucketSchedulerConfig {
 }
 
 type splitBucketSchedulerConfig struct {
-	syncutil.RWMutex
 	schedulerConfig
 	Degree     int    `json:"degree"`
 	SplitLimit uint64 `json:"split-limit"`
@@ -145,7 +143,7 @@ func newSplitBucketHandler(conf *splitBucketSchedulerConfig) http.Handler {
 }
 
 func newSplitBucketScheduler(opController *operator.Controller, conf *splitBucketSchedulerConfig) *splitBucketScheduler {
-	base := NewBaseScheduler(opController, types.SplitBucketScheduler)
+	base := NewBaseScheduler(opController, types.SplitBucketScheduler, conf)
 	handler := newSplitBucketHandler(conf)
 	ret := &splitBucketScheduler{
 		BaseScheduler: base,
diff --git a/pkg/schedule/schedulers/transfer_witness_leader.go b/pkg/schedule/schedulers/transfer_witness_leader.go
index 9e7bd9bccc1..f883bd7aee1 100644
--- a/pkg/schedule/schedulers/transfer_witness_leader.go
+++ b/pkg/schedule/schedulers/transfer_witness_leader.go
@@ -47,7 +47,7 @@ type transferWitnessLeaderScheduler struct {
 // newTransferWitnessLeaderScheduler creates an admin scheduler that transfers witness leader of a region.
 func newTransferWitnessLeaderScheduler(opController *operator.Controller) Scheduler {
 	return &transferWitnessLeaderScheduler{
-		BaseScheduler: NewBaseScheduler(opController, types.TransferWitnessLeaderScheduler),
+		BaseScheduler: NewBaseScheduler(opController, types.TransferWitnessLeaderScheduler, nil),
 		regions:       make(chan *core.RegionInfo, transferWitnessLeaderRecvMaxRegionSize),
 	}
 }
diff --git a/pkg/storage/kv/mem_kv.go b/pkg/storage/kv/mem_kv.go
index b97a3d6cfa1..dc24ab3e0f6 100644
--- a/pkg/storage/kv/mem_kv.go
+++ b/pkg/storage/kv/mem_kv.go
@@ -28,7 +28,7 @@ type memoryKV struct {
 	tree *btree.BTreeG[memoryKVItem]
 }
 
-// NewMemoryKV returns an in-memory kvBase for testing.
+// NewMemoryKV returns an in-memory kvBase.
 func NewMemoryKV() Base {
 	return &memoryKV{
 		tree: btree.NewG(2, func(i, j memoryKVItem) bool {
diff --git a/pkg/utils/syncutil/mutex_sync.go b/pkg/utils/syncutil/mutex_sync.go
index f6789b677c9..d5b5078643d 100644
--- a/pkg/utils/syncutil/mutex_sync.go
+++ b/pkg/utils/syncutil/mutex_sync.go
@@ -34,3 +34,13 @@ type Mutex struct {
 type RWMutex struct {
 	sync.RWMutex
 }
+
+type RWMutexInterface interface {
+	Lock()
+	RLock()
+	RLocker() sync.Locker
+	RUnlock()
+	TryLock() bool
+	TryRLock() bool
+	Unlock()
+}
diff --git a/server/config/persist_options.go b/server/config/persist_options.go
index b6963a6645a..a1cedcd77f1 100644
--- a/server/config/persist_options.go
+++ b/server/config/persist_options.go
@@ -669,18 +669,6 @@ func (o *PersistOptions) GetSchedulers() sc.SchedulerConfigs {
 	return o.GetScheduleConfig().Schedulers
 }
 
-// IsSchedulerDisabled returns if the scheduler is disabled.
-func (o *PersistOptions) IsSchedulerDisabled(tp types.CheckerSchedulerType) bool {
-	oldType := types.SchedulerTypeCompatibleMap[tp]
-	schedulers := o.GetScheduleConfig().Schedulers
-	for _, s := range schedulers {
-		if oldType == s.Type {
-			return s.Disable
-		}
-	}
-	return false
-}
-
 // GetHotRegionsWriteInterval gets interval for PD to store Hot Region information.
 func (o *PersistOptions) GetHotRegionsWriteInterval() time.Duration {
 	return o.GetScheduleConfig().HotRegionsWriteInterval.Duration
@@ -703,6 +691,7 @@ func (o *PersistOptions) AddSchedulerCfg(tp types.CheckerSchedulerType, args []s
 			return
 		}
 
+		// TODO: remove this after sc.SchedulerConfig deprecate
 		if reflect.DeepEqual(schedulerCfg, sc.SchedulerConfig{Type: oldType, Args: args, Disable: true}) {
 			schedulerCfg.Disable = false
 			v.Schedulers[i] = schedulerCfg
@@ -721,6 +710,7 @@ func (o *PersistOptions) RemoveSchedulerCfg(tp types.CheckerSchedulerType) {
 	for i, schedulerCfg := range v.Schedulers {
 		if oldType == schedulerCfg.Type {
 			if sc.IsDefaultScheduler(oldType) {
+				// TODO: remove this after sc.SchedulerConfig deprecate
 				schedulerCfg.Disable = true
 				v.Schedulers[i] = schedulerCfg
 			} else {
diff --git a/tests/integrations/mcs/scheduling/server_test.go b/tests/integrations/mcs/scheduling/server_test.go
index ab6bb93c60c..d44f684eb2c 100644
--- a/tests/integrations/mcs/scheduling/server_test.go
+++ b/tests/integrations/mcs/scheduling/server_test.go
@@ -375,9 +375,7 @@ func (suite *serverTestSuite) TestSchedulerSync() {
 	checkDisabled := func(name string, shouldDisabled bool) {
 		re.NotNil(schedulersController.GetScheduler(name), name)
 		testutil.Eventually(re, func() bool {
-			disabled, err := schedulersController.IsSchedulerDisabled(name)
-			re.NoError(err, name)
-			return disabled == shouldDisabled
+			return schedulersController.GetScheduler(name).IsDisable() == shouldDisabled
 		})
 	}
 	for _, name := range defaultSchedulerNames {