scheduler: make evict-slow-trend scheduler can manually modify recovery duration.

pkg/schedule/schedulers/evict_slow_trend.go

@@ -15,9 +15,12 @@
 package schedulers
 
 import (
+ "net/http"
  "strconv"
+ "sync/atomic"
  "time"
 
+ "github.com/gorilla/mux"
  "github.com/pingcap/errors"
  "github.com/pingcap/failpoint"
  "github.com/pingcap/log"
@@ -26,6 +29,8 @@ import (
  "github.com/tikv/pd/pkg/schedule/operator"
  "github.com/tikv/pd/pkg/schedule/plan"
  "github.com/tikv/pd/pkg/storage/endpoint"
+ "github.com/tikv/pd/pkg/utils/apiutil"
+ "github.com/unrolled/render"
  "go.uber.org/zap"
 )
 
@@ -54,11 +59,28 @@ type evictSlowTrendSchedulerConfig struct {
  evictCandidate slowCandidate
  // Last chosen candidate for eviction.
  lastEvictCandidate slowCandidate
-
+ // Duration gap for recovering the candidate, unit: s.
+ RecoveryDurationGap uint64 `json:"recovery-duration"`
  // Only evict one store for now
  EvictedStores []uint64 `json:"evict-by-trend-stores"`
 }
 
+func initEvictSlowTrendSchedulerConfig(storage endpoint.ConfigStorage) *evictSlowTrendSchedulerConfig {
+ return &evictSlowTrendSchedulerConfig{
+ storage: storage,
+ evictCandidate: slowCandidate{},
+ lastEvictCandidate: slowCandidate{},
+ RecoveryDurationGap: defaultRecoveryDurationGap,
+ EvictedStores: make([]uint64, 0),
+ }
+}
+
+func (conf *evictSlowTrendSchedulerConfig) Clone() *evictSlowTrendSchedulerConfig {
+ return &evictSlowTrendSchedulerConfig{
+ RecoveryDurationGap: atomic.LoadUint64(&conf.RecoveryDurationGap),
+ }
+}
+
 func (conf *evictSlowTrendSchedulerConfig) Persist() error {
  name := conf.getSchedulerName()
  data, err := EncodeConfig(conf)
@@ -116,6 +138,15 @@ func (conf *evictSlowTrendSchedulerConfig) lastCandidateCapturedSecs() uint64 {
  return DurationSinceAsSecs(conf.lastEvictCandidate.captureTS)
 }
 
+// readyForRecovery checks whether the last cpatured candidate is ready for recovery.
+func (conf *evictSlowTrendSchedulerConfig) readyForRecovery() bool {
+ recoveryDurationGap := atomic.LoadUint64(&conf.RecoveryDurationGap)
+ failpoint.Inject("transientRecoveryGap", func() {
+ recoveryDurationGap = 0
+ })
+ return conf.lastCandidateCapturedSecs() >= recoveryDurationGap
+}
+
 func (conf *evictSlowTrendSchedulerConfig) captureCandidate(id uint64) {
  conf.evictCandidate = slowCandidate{
  storeID: id,
@@ -162,9 +193,50 @@ func (conf *evictSlowTrendSchedulerConfig) clearAndPersist(cluster sche.Schedule
  return oldID, conf.Persist()
 }
 
+type evictSlowTrendHandler struct {
+ rd *render.Render
+ config *evictSlowTrendSchedulerConfig
+}
+
+func newEvictSlowTrendHandler(config *evictSlowTrendSchedulerConfig) http.Handler {
+ h := &evictSlowTrendHandler{
+ config: config,
+ rd: render.New(render.Options{IndentJSON: true}),
+ }
+ router := mux.NewRouter()
+ router.HandleFunc("/config", h.UpdateConfig).Methods(http.MethodPost)
+ router.HandleFunc("/list", h.ListConfig).Methods(http.MethodGet)
+ return router
+}
+
+func (handler *evictSlowTrendHandler) UpdateConfig(w http.ResponseWriter, r *http.Request) {
+ var input map[string]interface{}
+ if err := apiutil.ReadJSONRespondError(handler.rd, w, r.Body, &input); err != nil {
+ return
+ }
+ recoveryDurationGapFloat, ok := input["recovery-duration"].(float64)
+ if ok {
+ recoveryDurationGap := (uint64)(recoveryDurationGapFloat)
+ prevRecoveryDurationGap := atomic.LoadUint64(&handler.config.RecoveryDurationGap)
+ atomic.StoreUint64(&handler.config.RecoveryDurationGap, recoveryDurationGap)
+ log.Info("evict-slow-trend-scheduler update 'recovery-duration' - unit: s", zap.Uint64("prev", prevRecoveryDurationGap), zap.Uint64("cur", recoveryDurationGap))
+ }
+ handler.rd.JSON(w, http.StatusOK, nil)
+}
+
+func (handler *evictSlowTrendHandler) ListConfig(w http.ResponseWriter, r *http.Request) {
+ conf := handler.config.Clone()
+ handler.rd.JSON(w, http.StatusOK, conf)
+}
+
 type evictSlowTrendScheduler struct {
  *BaseScheduler
- conf *evictSlowTrendSchedulerConfig
+ conf *evictSlowTrendSchedulerConfig
+ handler http.Handler
+}
+
+func (s *evictSlowTrendScheduler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ s.handler.ServeHTTP(w, r)
 }
 
 func (s *evictSlowTrendScheduler) GetName() string {
@@ -244,7 +316,7 @@ func (s *evictSlowTrendScheduler) Schedule(cluster sche.SchedulerCluster, dryRun
  // slow node next time.
  log.Info("store evicted by slow trend has been removed", zap.Uint64("store-id", store.GetID()))
  storeSlowTrendActionStatusGauge.WithLabelValues("evict", "stop_removed").Inc()
- } else if checkStoreCanRecover(cluster, store, s.conf.lastCandidateCapturedSecs()) {
+ } else if checkStoreCanRecover(cluster, store) && s.conf.readyForRecovery() {
  log.Info("store evicted by slow trend has been recovered", zap.Uint64("store-id", store.GetID()))
  storeSlowTrendActionStatusGauge.WithLabelValues("evict", "stop_recovered").Inc()
  } else {
@@ -301,9 +373,11 @@ func (s *evictSlowTrendScheduler) Schedule(cluster sche.SchedulerCluster, dryRun
 }
 
 func newEvictSlowTrendScheduler(opController *operator.Controller, conf *evictSlowTrendSchedulerConfig) Scheduler {
+ handler := newEvictSlowTrendHandler(conf)
  return &evictSlowTrendScheduler{
  BaseScheduler: NewBaseScheduler(opController),
  conf: conf,
+ handler: handler,
  }
 }
 
@@ -453,7 +527,7 @@ func checkStoreSlowerThanOthers(cluster sche.SchedulerCluster, target *core.Stor
  return slowerThanStoresNum >= expected
 }
 
-func checkStoreCanRecover(cluster sche.SchedulerCluster, target *core.StoreInfo, recoveryGap uint64) bool {
+func checkStoreCanRecover(cluster sche.SchedulerCluster, target *core.StoreInfo) bool {
  /*
  //
  // This might not be necessary,
@@ -473,7 +547,7 @@ func checkStoreCanRecover(cluster sche.SchedulerCluster, target *core.StoreInfo,
  storeSlowTrendActionStatusGauge.WithLabelValues("recover.judging:got-event").Inc()
  }
  */
- return checkStoreFasterThanOthers(cluster, target) && checkStoreReadyForRecover(target, recoveryGap)
+ return checkStoreFasterThanOthers(cluster, target)
 }
 
 func checkStoreFasterThanOthers(cluster sche.SchedulerCluster, target *core.StoreInfo) bool {
@@ -507,19 +581,6 @@ func checkStoreFasterThanOthers(cluster sche.SchedulerCluster, target *core.Stor
  return fasterThanStores >= expected
 }
 
-// checkStoreReadyForRecover checks whether the given target store is ready for recover.
-func checkStoreReadyForRecover(target *core.StoreInfo, recoveryGap uint64) bool {
- durationGap := uint64(defaultRecoveryDurationGap)
- failpoint.Inject("transientRecoveryGap", func() {
- durationGap = 0
- })
- if targetSlowTrend := target.GetSlowTrend(); targetSlowTrend != nil {
- // TODO: setting the recovery time in SlowTrend
- return recoveryGap >= durationGap
- }
- return true
-}
-
 // DurationSinceAsSecs returns the duration gap since the given startTS, unit: s.
 func DurationSinceAsSecs(startTS time.Time) uint64 {
  return uint64(time.Since(startTS).Seconds())

pkg/schedule/schedulers/evict_slow_trend_test.go

@@ -93,7 +93,7 @@ func (suite *evictSlowTrendTestSuite) TestEvictSlowTrendBasicFuncs() {
  suite.Equal(*lastCapturedCandidate, es2.conf.evictCandidate)
  suite.Equal(es2.conf.candidateCapturedSecs(), uint64(0))
  suite.Equal(es2.conf.lastCandidateCapturedSecs(), uint64(0))
- suite.False(checkStoreReadyForRecover(store, es2.conf.lastCandidateCapturedSecs()))
+ suite.False(es2.conf.readyForRecovery())
  recoverTS := lastCapturedCandidate.recoverTS
  suite.True(recoverTS.After(lastCapturedCandidate.captureTS))
  // Pop captured store 1 and mark it has recovered.

pkg/schedule/schedulers/init.go

@@ -466,7 +466,7 @@ func schedulersRegister() {
  })
 
  RegisterScheduler(EvictSlowTrendType, func(opController *operator.Controller, storage endpoint.ConfigStorage, decoder ConfigDecoder, removeSchedulerCb ...func(string) error) (Scheduler, error) {
- conf := &evictSlowTrendSchedulerConfig{storage: storage, EvictedStores: make([]uint64, 0), evictCandidate: slowCandidate{}, lastEvictCandidate: slowCandidate{}}
+ conf := initEvictSlowTrendSchedulerConfig(storage)
  if err := decoder(conf); err != nil {
  return nil, err
  }

tests/pdctl/scheduler/scheduler_test.go

@@ -407,6 +407,21 @@ func (suite *schedulerTestSuite) checkScheduler(cluster *tests.TestCluster) {
  echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "add", "balance-leader-scheduler"}, nil)
  re.Contains(echo, "Success!")
 
+ // test evict-slow-trend scheduler config
+ echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "add", "evict-slow-trend-scheduler"}, nil)
+ re.Contains(echo, "Success!")
+ echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "show"}, nil)
+ re.Contains(echo, "evict-slow-trend-scheduler")
+ echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "config", "evict-slow-trend-scheduler", "set", "recovery-duration", "100"}, nil)
+ re.Contains(echo, "Success!")
+ conf = make(map[string]interface{})
+ mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "config", "evict-slow-trend-scheduler", "show"}, &conf)
+ re.Equal(100., conf["recovery-duration"])
+ echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "remove", "evict-slow-trend-scheduler"}, nil)
+ re.Contains(echo, "Success!")
+ echo = mustExec(re, cmd, []string{"-u", pdAddr, "scheduler", "show"}, nil)
+ re.NotContains(echo, "evict-slow-trend-scheduler")
+
  // test show scheduler with paused and disabled status.
  checkSchedulerWithStatusCommand := func(status string, expected []string) {
  var schedulers []string

tools/pd-ctl/pdctl/command/scheduler.go

@@ -499,6 +499,7 @@ func NewConfigSchedulerCommand() *cobra.Command {
  newConfigGrantHotRegionCommand(),
  newConfigBalanceLeaderCommand(),
  newSplitBucketCommand(),
+ newConfigEvictSlowTrendCommand(),
  )
  return c
 }
@@ -775,6 +776,25 @@ func setShuffleRegionSchedulerRolesCommandFunc(cmd *cobra.Command, args []string
  cmd.Println("Success!")
 }
 
+func newConfigEvictSlowTrendCommand() *cobra.Command {
+ c := &cobra.Command{
+ Use: "evict-slow-trend-scheduler",
+ Short: "evict-slow-trend-scheduler config",
+ Run: listSchedulerConfigCommandFunc,
+ }
+
+ c.AddCommand(&cobra.Command{
+ Use: "show",
+ Short: "list the config item",
+ Run: listSchedulerConfigCommandFunc,
+ }, &cobra.Command{
+ Use: "set <key> <value>",
+ Short: "set the config item",
+ Run: func(cmd *cobra.Command, args []string) { postSchedulerConfigCommandFunc(cmd, c.Name(), args) },
+ })
+ return c
+}
+
 // NewDescribeSchedulerCommand returns command to describe the scheduler.
 func NewDescribeSchedulerCommand() *cobra.Command {
  c := &cobra.Command{