Merge branch 'master' into update_pd_http_client_api

tikv · Nov 21, 2023 · c10956a · c10956a
2 parents cd5397a + 5a2a8d6
commit c10956a
Show file tree

Hide file tree

Showing 31 changed files with 201 additions and 122 deletions.
diff --git a/client/retry/backoff.go b/client/retry/backoff.go
@@ -34,9 +34,11 @@ func (bo *BackOffer) Exec(
 	fn func() error,
 ) error {
 	if err := fn(); err != nil {
+		after := time.NewTimer(bo.nextInterval())
+		defer after.Stop()
 		select {
 		case <-ctx.Done():
-		case <-time.After(bo.nextInterval()):
+		case <-after.C:
 			failpoint.Inject("backOffExecute", func() {
 				testBackOffExecuteFlag = true
 			})

diff --git a/pkg/core/store_option.go b/pkg/core/store_option.go
@@ -274,3 +274,23 @@ func SetLastAwakenTime(lastAwaken time.Time) StoreCreateOption {
 		store.lastAwakenTime = lastAwaken
 	}
 }
+
+// SetStoreMeta sets the meta for the store.
+func SetStoreMeta(newMeta *metapb.Store) StoreCreateOption {
+	return func(store *StoreInfo) {
+		meta := typeutil.DeepClone(store.meta, StoreFactory)
+		meta.Version = newMeta.GetVersion()
+		meta.GitHash = newMeta.GetGitHash()
+		meta.Address = newMeta.GetAddress()
+		meta.StatusAddress = newMeta.GetStatusAddress()
+		meta.PeerAddress = newMeta.GetPeerAddress()
+		meta.StartTimestamp = newMeta.GetStartTimestamp()
+		meta.DeployPath = newMeta.GetDeployPath()
+		meta.LastHeartbeat = newMeta.GetLastHeartbeat()
+		meta.State = newMeta.GetState()
+		meta.Labels = newMeta.GetLabels()
+		meta.NodeState = newMeta.GetNodeState()
+		meta.PhysicallyDestroyed = newMeta.GetPhysicallyDestroyed()
+		store.meta = meta
+	}
+}
diff --git a/pkg/mcs/scheduling/server/cluster.go b/pkg/mcs/scheduling/server/cluster.go
@@ -381,8 +381,7 @@ func (c *Cluster) HandleStoreHeartbeat(heartbeat *schedulingpb.StoreHeartbeatReq
 		return errors.Errorf("store %v not found", storeID)
 	}
 
-	nowTime := time.Now()
-	newStore := store.Clone(core.SetStoreStats(stats), core.SetLastHeartbeatTS(nowTime))
+	newStore := store.Clone(core.SetStoreStats(stats))
 
 	if store := c.GetStore(storeID); store != nil {
 		statistics.UpdateStoreHeartbeatMetrics(store)
@@ -486,10 +485,6 @@ func (c *Cluster) collectMetrics() {
 
 	c.coordinator.GetSchedulersController().CollectSchedulerMetrics()
 	c.coordinator.CollectHotSpotMetrics()
-	c.collectClusterMetrics()
-}
-
-func (c *Cluster) collectClusterMetrics() {
 	if c.regionStats == nil {
 		return
 	}
@@ -501,20 +496,8 @@ func (c *Cluster) collectClusterMetrics() {
 
 func (c *Cluster) resetMetrics() {
 	statistics.Reset()
-
 	schedulers.ResetSchedulerMetrics()
 	schedule.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.

diff --git a/pkg/mcs/scheduling/server/config/watcher.go b/pkg/mcs/scheduling/server/config/watcher.go
@@ -147,7 +147,8 @@ func (cw *Watcher) initializeSchedulerConfigWatcher() error {
 	prefixToTrim := cw.schedulerConfigPathPrefix + "/"
 	putFn := func(kv *mvccpb.KeyValue) error {
 		name := strings.TrimPrefix(string(kv.Key), prefixToTrim)
-		log.Info("update scheduler config", zap.String("name", string(kv.Value)))
+		log.Info("update scheduler config", zap.String("name", name),
+			zap.String("value", string(kv.Value)))
 		err := cw.storage.SaveSchedulerConfig(name, kv.Value)
 		if err != nil {
 			log.Warn("failed to save scheduler config",

diff --git a/pkg/mcs/scheduling/server/meta/watcher.go b/pkg/mcs/scheduling/server/meta/watcher.go
@@ -81,7 +81,7 @@ func (w *Watcher) initializeStoreWatcher() error {
 			w.basicCluster.PutStore(core.NewStoreInfo(store))
 			return nil
 		}
-		w.basicCluster.PutStore(origin.Clone(core.SetStoreState(store.GetState(), store.GetPhysicallyDestroyed())))
+		w.basicCluster.PutStore(origin.Clone(core.SetStoreMeta(store)))
 		return nil
 	}
 	deleteFn := func(kv *mvccpb.KeyValue) error {

diff --git a/pkg/schedule/operator/operator_test.go b/pkg/schedule/operator/operator_test.go
@@ -17,7 +17,6 @@ package operator
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -514,7 +513,7 @@ func (suite *operatorTestSuite) TestOpStepTimeout() {
 		},
 	}
 	for i, v := range testData {
-		fmt.Printf("case:%d\n", i)
+		suite.T().Logf("case: %d", i)
 		for _, step := range v.step {
 			suite.Equal(v.expect, step.Timeout(v.regionSize))
 		}

diff --git a/pkg/schedule/schedulers/base_scheduler.go b/pkg/schedule/schedulers/base_scheduler.go
@@ -92,8 +92,8 @@ func (s *BaseScheduler) GetNextInterval(interval time.Duration) time.Duration {
 	return intervalGrow(interval, MaxScheduleInterval, exponentialGrowth)
 }
 
-// Prepare does some prepare work
-func (s *BaseScheduler) Prepare(cluster sche.SchedulerCluster) error { return nil }
+// PrepareConfig does some prepare work about config.
+func (s *BaseScheduler) PrepareConfig(cluster sche.SchedulerCluster) error { return nil }
 
-// Cleanup does some cleanup work
-func (s *BaseScheduler) Cleanup(cluster sche.SchedulerCluster) {}
+// CleanConfig does some cleanup work about config.
+func (s *BaseScheduler) CleanConfig(cluster sche.SchedulerCluster) {}
diff --git a/pkg/schedule/schedulers/evict_leader.go b/pkg/schedule/schedulers/evict_leader.go
@@ -239,7 +239,7 @@ func pauseAndResumeLeaderTransfer(cluster *core.BasicCluster, old, new map[uint6
 	}
 }
 
-func (s *evictLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
+func (s *evictLeaderScheduler) PrepareConfig(cluster sche.SchedulerCluster) error {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	var res error
@@ -251,7 +251,7 @@ func (s *evictLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
 	return res
 }
 
-func (s *evictLeaderScheduler) Cleanup(cluster sche.SchedulerCluster) {
+func (s *evictLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	for id := range s.conf.StoreIDWithRanges {

diff --git a/pkg/schedule/schedulers/evict_slow_store.go b/pkg/schedule/schedulers/evict_slow_store.go
@@ -189,15 +189,15 @@ func (s *evictSlowStoreScheduler) EncodeConfig() ([]byte, error) {
 	return EncodeConfig(s.conf)
 }
 
-func (s *evictSlowStoreScheduler) Prepare(cluster sche.SchedulerCluster) error {
+func (s *evictSlowStoreScheduler) PrepareConfig(cluster sche.SchedulerCluster) error {
 	evictStore := s.conf.evictStore()
 	if evictStore != 0 {
 		return cluster.SlowStoreEvicted(evictStore)
 	}
 	return nil
 }
 
-func (s *evictSlowStoreScheduler) Cleanup(cluster sche.SchedulerCluster) {
+func (s *evictSlowStoreScheduler) CleanConfig(cluster sche.SchedulerCluster) {
 	s.cleanupEvictLeader(cluster)
 }
 

diff --git a/pkg/schedule/schedulers/evict_slow_store_test.go b/pkg/schedule/schedulers/evict_slow_store_test.go
@@ -123,13 +123,13 @@ func (suite *evictSlowStoreTestSuite) TestEvictSlowStorePrepare() {
 	suite.True(ok)
 	suite.Zero(es2.conf.evictStore())
 	// prepare with no evict store.
-	suite.es.Prepare(suite.tc)
+	suite.es.PrepareConfig(suite.tc)
 
 	es2.conf.setStoreAndPersist(1)
 	suite.Equal(uint64(1), es2.conf.evictStore())
 	suite.False(es2.conf.readyForRecovery())
 	// prepare with evict store.
-	suite.es.Prepare(suite.tc)
+	suite.es.PrepareConfig(suite.tc)
 }
 
 func (suite *evictSlowStoreTestSuite) TestEvictSlowStorePersistFail() {

diff --git a/pkg/schedule/schedulers/evict_slow_trend.go b/pkg/schedule/schedulers/evict_slow_trend.go
@@ -270,15 +270,15 @@ func (s *evictSlowTrendScheduler) EncodeConfig() ([]byte, error) {
 	return EncodeConfig(s.conf)
 }
 
-func (s *evictSlowTrendScheduler) Prepare(cluster sche.SchedulerCluster) error {
+func (s *evictSlowTrendScheduler) PrepareConfig(cluster sche.SchedulerCluster) error {
 	evictedStoreID := s.conf.evictedStore()
 	if evictedStoreID == 0 {
 		return nil
 	}
 	return cluster.SlowTrendEvicted(evictedStoreID)
 }
 
-func (s *evictSlowTrendScheduler) Cleanup(cluster sche.SchedulerCluster) {
+func (s *evictSlowTrendScheduler) CleanConfig(cluster sche.SchedulerCluster) {
 	s.cleanupEvictLeader(cluster)
 }
 

diff --git a/pkg/schedule/schedulers/evict_slow_trend_test.go b/pkg/schedule/schedulers/evict_slow_trend_test.go
@@ -255,10 +255,10 @@ func (suite *evictSlowTrendTestSuite) TestEvictSlowTrendPrepare() {
 	suite.True(ok)
 	suite.Zero(es2.conf.evictedStore())
 	// prepare with no evict store.
-	suite.es.Prepare(suite.tc)
+	suite.es.PrepareConfig(suite.tc)
 
 	es2.conf.setStoreAndPersist(1)
 	suite.Equal(uint64(1), es2.conf.evictedStore())
 	// prepare with evict store.
-	suite.es.Prepare(suite.tc)
+	suite.es.PrepareConfig(suite.tc)
 }
diff --git a/pkg/schedule/schedulers/grant_leader.go b/pkg/schedule/schedulers/grant_leader.go
@@ -197,7 +197,7 @@ func (s *grantLeaderScheduler) ReloadConfig() error {
 	return nil
 }
 
-func (s *grantLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
+func (s *grantLeaderScheduler) PrepareConfig(cluster sche.SchedulerCluster) error {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	var res error
@@ -209,7 +209,7 @@ func (s *grantLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
 	return res
 }
 
-func (s *grantLeaderScheduler) Cleanup(cluster sche.SchedulerCluster) {
+func (s *grantLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	for id := range s.conf.StoreIDWithRanges {

diff --git a/pkg/schedule/schedulers/scheduler.go b/pkg/schedule/schedulers/scheduler.go
@@ -42,8 +42,8 @@ type Scheduler interface {
 	ReloadConfig() error
 	GetMinInterval() time.Duration
 	GetNextInterval(interval time.Duration) time.Duration
-	Prepare(cluster sche.SchedulerCluster) error
-	Cleanup(cluster sche.SchedulerCluster)
+	PrepareConfig(cluster sche.SchedulerCluster) error
+	CleanConfig(cluster sche.SchedulerCluster)
 	Schedule(cluster sche.SchedulerCluster, dryRun bool) ([]*operator.Operator, []plan.Plan)
 	IsScheduleAllowed(cluster sche.SchedulerCluster) bool
 }

diff --git a/pkg/schedule/schedulers/scheduler_controller.go b/pkg/schedule/schedulers/scheduler_controller.go
@@ -156,7 +156,8 @@ func (c *Controller) AddSchedulerHandler(scheduler Scheduler, args ...string) er
 		return err
 	}
 	c.cluster.GetSchedulerConfig().AddSchedulerCfg(scheduler.GetType(), args)
-	return nil
+	err := scheduler.PrepareConfig(c.cluster)
+	return err
 }
 
 // RemoveSchedulerHandler removes the HTTP handler for a scheduler.
@@ -183,6 +184,7 @@ func (c *Controller) RemoveSchedulerHandler(name string) error {
 		return err
 	}
 
+	s.(Scheduler).CleanConfig(c.cluster)
 	delete(c.schedulerHandlers, name)
 
 	return nil
@@ -198,7 +200,7 @@ func (c *Controller) AddScheduler(scheduler Scheduler, args ...string) error {
 	}
 
 	s := NewScheduleController(c.ctx, c.cluster, c.opController, scheduler)
-	if err := s.Scheduler.Prepare(c.cluster); err != nil {
+	if err := s.Scheduler.PrepareConfig(c.cluster); err != nil {
 		return err
 	}
 
@@ -343,7 +345,7 @@ func (c *Controller) IsSchedulerExisted(name string) (bool, error) {
 func (c *Controller) runScheduler(s *ScheduleController) {
 	defer logutil.LogPanic()
 	defer c.wg.Done()
-	defer s.Scheduler.Cleanup(c.cluster)
+	defer s.Scheduler.CleanConfig(c.cluster)
 
 	ticker := time.NewTicker(s.GetInterval())
 	defer ticker.Stop()

diff --git a/pkg/statistics/hot_cache.go b/pkg/statistics/hot_cache.go
@@ -125,8 +125,8 @@ func (w *HotCache) CollectMetrics() {
 	w.CheckReadAsync(newCollectMetricsTask())
 }
 
-// ResetMetrics resets the hot cache metrics.
-func (w *HotCache) ResetMetrics() {
+// ResetHotCacheStatusMetrics resets the hot cache metrics.
+func ResetHotCacheStatusMetrics() {
 	hotCacheStatusGauge.Reset()
 }
 

diff --git a/pkg/statistics/region_collection.go b/pkg/statistics/region_collection.go
@@ -272,8 +272,8 @@ func (r *RegionStatistics) Collect() {
 	regionWitnessLeaderRegionCounter.Set(float64(len(r.stats[WitnessLeader])))
 }
 
-// Reset resets the metrics of the regions' status.
-func (r *RegionStatistics) Reset() {
+// ResetRegionStatsMetrics resets the metrics of the regions' status.
+func ResetRegionStatsMetrics() {
 	regionMissPeerRegionCounter.Set(0)
 	regionExtraPeerRegionCounter.Set(0)
 	regionDownPeerRegionCounter.Set(0)
@@ -326,8 +326,8 @@ func (l *LabelStatistics) Collect() {
 	}
 }
 
-// Reset resets the metrics of the label status.
-func (l *LabelStatistics) Reset() {
+// ResetLabelStatsMetrics resets the metrics of the label status.
+func ResetLabelStatsMetrics() {
 	regionLabelLevelGauge.Reset()
 }
 

diff --git a/pkg/statistics/store_collection.go b/pkg/statistics/store_collection.go
@@ -322,4 +322,7 @@ func Reset() {
 	storeStatusGauge.Reset()
 	clusterStatusGauge.Reset()
 	placementStatusGauge.Reset()
+	ResetRegionStatsMetrics()
+	ResetLabelStatsMetrics()
+	ResetHotCacheStatusMetrics()
 }
diff --git a/plugin/scheduler_example/evict_leader.go b/plugin/scheduler_example/evict_leader.go
@@ -186,7 +186,7 @@ func (s *evictLeaderScheduler) EncodeConfig() ([]byte, error) {
 	return schedulers.EncodeConfig(s.conf)
 }
 
-func (s *evictLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
+func (s *evictLeaderScheduler) PrepareConfig(cluster sche.SchedulerCluster) error {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	var res error
@@ -198,7 +198,7 @@ func (s *evictLeaderScheduler) Prepare(cluster sche.SchedulerCluster) error {
 	return res
 }
 
-func (s *evictLeaderScheduler) Cleanup(cluster sche.SchedulerCluster) {
+func (s *evictLeaderScheduler) CleanConfig(cluster sche.SchedulerCluster) {
 	s.conf.mu.RLock()
 	defer s.conf.mu.RUnlock()
 	for id := range s.conf.StoreIDWitRanges {

diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go
@@ -654,7 +654,7 @@ func (c *RaftCluster) runMetricsCollectionJob() {
 	ticker := time.NewTicker(metricsCollectionJobInterval)
 	failpoint.Inject("highFrequencyClusterJobs", func() {
 		ticker.Stop()
-		ticker = time.NewTicker(time.Microsecond)
+		ticker = time.NewTicker(time.Millisecond)
 	})
 	defer ticker.Stop()
 
@@ -734,10 +734,10 @@ func (c *RaftCluster) Stop() {
 		return
 	}
 	c.running = false
+	c.cancel()
 	if !c.IsServiceIndependent(mcsutils.SchedulingServiceName) {
 		c.stopSchedulingJobs()
 	}
-	c.cancel()
 	c.Unlock()
 
 	c.wg.Wait()