mcs: disable PD scheduling when enabling scheduling service (#7039)

ref #5839, close #7041 Signed-off-by: Ryan Leung <[email protected]>
tikv · Sep 6, 2023 · 745c942 · 745c942
1 parent 7d50755
commit 745c942
Show file tree

Hide file tree

Showing 8 changed files with 191 additions and 119 deletions.
diff --git a/pkg/mcs/scheduling/server/server.go b/pkg/mcs/scheduling/server/server.go
@@ -45,6 +45,7 @@ import (
 	"github.com/tikv/pd/pkg/member"
 	"github.com/tikv/pd/pkg/schedule"
 	"github.com/tikv/pd/pkg/schedule/hbstream"
+	"github.com/tikv/pd/pkg/schedule/schedulers"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/apiutil"
@@ -162,6 +163,7 @@ func (s *Server) updateAPIServerMemberLoop() {
 		members, err := s.GetClient().MemberList(ctx)
 		if err != nil {
 			log.Warn("failed to list members", errs.ZapError(err))
+			continue
 		}
 		for _, ep := range members.Members {
 			status, err := s.GetClient().Status(ctx, ep.ClientURLs[0])
@@ -438,6 +440,7 @@ func CreateServer(ctx context.Context, cfg *config.Config) *Server {
 
 // CreateServerWrapper encapsulates the configuration/log/metrics initialization and create the server
 func CreateServerWrapper(cmd *cobra.Command, args []string) {
+	schedulers.Register()
 	cmd.Flags().Parse(args)
 	cfg := config.NewConfig()
 	flagSet := cmd.Flags()

diff --git a/pkg/schedule/coordinator.go b/pkg/schedule/coordinator.go
@@ -379,7 +379,7 @@ func (c *Coordinator) Run() {
 		}
 	}
 	log.Info("Coordinator starts to run schedulers")
-	c.initSchedulers()
+	c.InitSchedulers(true)
 
 	c.wg.Add(4)
 	// Starts to patrol regions.
@@ -391,7 +391,8 @@ func (c *Coordinator) Run() {
 	go c.driveSlowNodeScheduler()
 }
 
-func (c *Coordinator) initSchedulers() {
+// InitSchedulers initializes schedulers.
+func (c *Coordinator) InitSchedulers(needRun bool) {
 	var (
 		scheduleNames []string
 		configs       []string
@@ -401,7 +402,7 @@ func (c *Coordinator) initSchedulers() {
 		scheduleNames, configs, err = c.cluster.GetStorage().LoadAllScheduleConfig()
 		select {
 		case <-c.ctx.Done():
-			log.Info("Coordinator stops running")
+			log.Info("init schedulers has been stopped")
 			return
 		default:
 		}
@@ -439,8 +440,10 @@ func (c *Coordinator) initSchedulers() {
 			continue
 		}
 		log.Info("create scheduler with independent configuration", zap.String("scheduler-name", s.GetName()))
-		if err = c.schedulers.AddScheduler(s); err != nil {
-			log.Error("can not add scheduler with independent configuration", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", cfg.Args), errs.ZapError(err))
+		if needRun {
+			if err = c.schedulers.AddScheduler(s); err != nil {
+				log.Error("can not add scheduler with independent configuration", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", cfg.Args), errs.ZapError(err))
+			}
 		}
 	}
 
@@ -461,12 +464,14 @@ func (c *Coordinator) initSchedulers() {
 		}
 
 		log.Info("create scheduler", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", schedulerCfg.Args))
-		if err = c.schedulers.AddScheduler(s, schedulerCfg.Args...); err != nil && !errors.ErrorEqual(err, errs.ErrSchedulerExisted.FastGenByArgs()) {
-			log.Error("can not add scheduler", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", schedulerCfg.Args), errs.ZapError(err))
-		} else {
-			// Only records the valid scheduler config.
-			scheduleCfg.Schedulers[k] = schedulerCfg
-			k++
+		if needRun {
+			if err = c.schedulers.AddScheduler(s, schedulerCfg.Args...); err != nil && !errors.ErrorEqual(err, errs.ErrSchedulerExisted.FastGenByArgs()) {
+				log.Error("can not add scheduler", zap.String("scheduler-name", s.GetName()), zap.Strings("scheduler-args", schedulerCfg.Args), errs.ZapError(err))
+			} else {
+				// Only records the valid scheduler config.
+				scheduleCfg.Schedulers[k] = schedulerCfg
+				k++
+			}
 		}
 	}
 

diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go
@@ -61,7 +61,7 @@ func newStoreStatistics(opt *config.PersistOptions) *storeStatistics {
 	}
 }
 
-func (s *storeStatistics) Observe(store *core.StoreInfo, stats *StoresStats) {
+func (s *storeStatistics) Observe(store *core.StoreInfo) {
 	for _, k := range s.opt.GetLocationLabels() {
 		v := store.GetLabelValue(k)
 		if v == "" {
@@ -146,8 +146,12 @@ func (s *storeStatistics) Observe(store *core.StoreInfo, stats *StoresStats) {
 		storeStatusGauge.WithLabelValues(storeAddress, id, "store_slow_trend_result_value").Set(slowTrend.ResultValue)
 		storeStatusGauge.WithLabelValues(storeAddress, id, "store_slow_trend_result_rate").Set(slowTrend.ResultRate)
 	}
+}
 
+func (s *storeStatistics) ObserveHotStat(store *core.StoreInfo, stats *StoresStats) {
 	// Store flows.
+	storeAddress := store.GetAddress()
+	id := strconv.FormatUint(store.GetID(), 10)
 	storeFlowStats := stats.GetRollingStoreStats(store.GetID())
 	if storeFlowStats == nil {
 		return
@@ -298,8 +302,12 @@ func NewStoreStatisticsMap(opt *config.PersistOptions) *storeStatisticsMap {
 	}
 }
 
-func (m *storeStatisticsMap) Observe(store *core.StoreInfo, stats *StoresStats) {
-	m.stats.Observe(store, stats)
+func (m *storeStatisticsMap) Observe(store *core.StoreInfo) {
+	m.stats.Observe(store)
+}
+
+func (m *storeStatisticsMap) ObserveHotStat(store *core.StoreInfo, stats *StoresStats) {
+	m.stats.ObserveHotStat(store, stats)
 }
 
 func (m *storeStatisticsMap) Collect() {

diff --git a/pkg/statistics/store_collection_test.go b/pkg/statistics/store_collection_test.go
@@ -67,7 +67,8 @@ func TestStoreStatistics(t *testing.T) {
 	stores[5] = store5
 	storeStats := NewStoreStatisticsMap(opt)
 	for _, store := range stores {
-		storeStats.Observe(store, storesStats)
+		storeStats.Observe(store)
+		storeStats.ObserveHotStat(store, storesStats)
 	}
 	stats := storeStats.stats