Merge branch 'master' into fix_coordinator

Author: ti-chi-bot[bot]

metrics/grafana/pd.json

@@ -425,12 +425,75 @@ func (c *Cluster) runCoordinator() {
 	c.coordinator.RunUntilStop()
 }
 
+func (c *Cluster) runMetricsCollectionJob() {
+	defer logutil.LogPanic()
+	defer c.wg.Done()
+
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-c.ctx.Done():
+			log.Info("metrics are reset")
+			c.resetMetrics()
+			log.Info("metrics collection job has been stopped")
+			return
+		case <-ticker.C:
+			c.collectMetrics()
+		}
+	}
+}
+
+func (c *Cluster) collectMetrics() {
+	statsMap := statistics.NewStoreStatisticsMap(c.persistConfig)
+	stores := c.GetStores()
+	for _, s := range stores {
+		statsMap.Observe(s)
+		statsMap.ObserveHotStat(s, c.hotStat.StoresStats)
+	}
+	statsMap.Collect()
+
+	c.coordinator.GetSchedulersController().CollectSchedulerMetrics()
+	c.coordinator.CollectHotSpotMetrics()
+	c.collectClusterMetrics()
+}
+
+func (c *Cluster) collectClusterMetrics() {
+	if c.regionStats == nil {
+		return
+	}
+	c.regionStats.Collect()
+	c.labelStats.Collect()
+	// collect hot cache metrics
+	c.hotStat.CollectMetrics()
+}
+
+func (c *Cluster) resetMetrics() {
+	statistics.Reset()
+
+	c.coordinator.GetSchedulersController().ResetSchedulerMetrics()
+	c.coordinator.ResetHotSpotMetrics()
+	c.resetClusterMetrics()
+}
+
+func (c *Cluster) resetClusterMetrics() {
+	if c.regionStats == nil {
+		return
+	}
+	c.regionStats.Reset()
+	c.labelStats.Reset()
+	// reset hot cache metrics
+	c.hotStat.ResetMetrics()
+}
+
 // StartBackgroundJobs starts background jobs.
 func (c *Cluster) StartBackgroundJobs() {
-	c.wg.Add(3)
+	c.wg.Add(4)
 	go c.updateScheduler()
 	go c.runUpdateStoreStats()
 	go c.runCoordinator()
+	go c.runMetricsCollectionJob()
 	c.running.Store(true)
 }
 

pkg/mcs/scheduling/server/cluster.go

@@ -499,6 +499,11 @@ func (o *PersistConfig) IsSchedulingHalted() bool {
 	return o.GetScheduleConfig().HaltScheduling
 }
 
+// GetStoresLimit gets the stores' limit.
+func (o *PersistConfig) GetStoresLimit() map[uint64]sc.StoreLimitConfig {
+	return o.GetScheduleConfig().StoreLimit
+}
+
 // GetStoreLimitByType returns the limit of a store with a given type.
 func (o *PersistConfig) GetStoreLimitByType(storeID uint64, typ storelimit.Type) (returned float64) {
 	limit := o.GetStoreLimit(storeID)
@@ -620,11 +625,21 @@ func (o *PersistConfig) GetRegionMaxSize() uint64 {
 	return o.GetStoreConfig().GetRegionMaxSize()
 }
 
-// GetRegionMaxKeys returns the region split keys
+// GetRegionMaxKeys returns the max region keys
 func (o *PersistConfig) GetRegionMaxKeys() uint64 {
 	return o.GetStoreConfig().GetRegionMaxKeys()
 }
 
+// GetRegionSplitSize returns the region split size in MB
+func (o *PersistConfig) GetRegionSplitSize() uint64 {
+	return o.GetStoreConfig().GetRegionSplitSize()
+}
+
+// GetRegionSplitKeys returns the region split keys
+func (o *PersistConfig) GetRegionSplitKeys() uint64 {
+	return o.GetStoreConfig().GetRegionSplitKeys()
+}
+
 // IsEnableRegionBucket return true if the region bucket is enabled.
 func (o *PersistConfig) IsEnableRegionBucket() bool {
 	return o.GetStoreConfig().IsEnableRegionBucket()

pkg/mcs/scheduling/server/config/config.go

@@ -47,6 +47,7 @@ type SchedulerConfigProvider interface {
 	SharedConfigProvider
 
 	IsSchedulingHalted() bool
+	GetStoresLimit() map[uint64]StoreLimitConfig
 
 	IsSchedulerDisabled(string) bool
 	AddSchedulerCfg(string, []string)
@@ -137,6 +138,8 @@ type ConfProvider interface {
 type StoreConfigProvider interface {
 	GetRegionMaxSize() uint64
 	GetRegionMaxKeys() uint64
+	GetRegionSplitSize() uint64
+	GetRegionSplitKeys() uint64
 	CheckRegionSize(uint64, uint64) error
 	CheckRegionKeys(uint64, uint64) error
 	IsEnableRegionBucket() bool

pkg/schedule/config/config_provider.go

@@ -156,6 +156,7 @@ func (c *Coordinator) PatrolRegions() {
 			// Note: we reset the ticker here to support updating configuration dynamically.
 			ticker.Reset(c.cluster.GetCheckerConfig().GetPatrolRegionInterval())
 		case <-c.ctx.Done():
+			patrolCheckRegionsGauge.Set(0)
 			log.Info("patrol regions has been stopped")
 			return
 		}

pkg/schedule/coordinator.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,52 @@ 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 {
+		handler.rd.JSON(w, http.StatusInternalServerError, errors.New("invalid argument for 'recovery-duration'").Error())
+		return
+	}
+	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 +318,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 +375,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 +529,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 +549,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 +583,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.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/evict_slow_trend_test.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
 		}