address comment

Signed-off-by: lhy1024 <[email protected]>

server/api/config_test.go

@@ -299,6 +299,7 @@ func (s *testConfigSuite) TestConfigPDServer(c *C) {
 	c.Assert(sc.MetricStorage, Equals, "")
 	c.Assert(sc.DashboardAddress, Equals, "auto")
 	c.Assert(sc.FlowRoundByDigit, Equals, int(3))
+	c.Assert(sc.SaveMinResolvedTSInterval, Equals, typeutil.NewDuration(0))
 	c.Assert(sc.MaxResetTSGap.Duration, Equals, 24*time.Hour)
 }
 

server/api/min_resolved_ts_test.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 
 	. "github.com/pingcap/check"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/tikv/pd/pkg/apiutil"
 	"github.com/tikv/pd/server"
@@ -32,6 +33,7 @@ type testMinResolvedTSSuite struct {
 }
 
 func (s *testMinResolvedTSSuite) SetUpSuite(c *C) {
+	c.Assert(failpoint.Enable("github.com/tikv/pd/server/highFrequencyClusterJobs", `return(true)`), IsNil)
 	s.svr, s.cleanup = mustNewServer(c)
 	mustWaitLeader(c, []*server.Server{s.svr})
 
@@ -51,13 +53,13 @@ func (s *testMinResolvedTSSuite) TestMinResolvedTS(c *C) {
 	storage := s.svr.GetStorage()
 	min := uint64(233)
 	storage.SaveMinResolvedTS(min)
-	result := &listMinResolvedTS{
+	result := &minResolvedTS{
 		MinResolvedTS: min,
 	}
 	res, err := testDialClient.Get(url)
 	c.Assert(err, IsNil)
 	defer res.Body.Close()
-	listResp := &listMinResolvedTS{}
+	listResp := &minResolvedTS{}
 	err = apiutil.ReadJSON(res.Body, listResp)
 	c.Assert(err, IsNil)
 	c.Assert(listResp, DeepEquals, result)

server/cluster/cluster.go

@@ -54,7 +54,6 @@ import (
 )
 
 var backgroundJobInterval = 10 * time.Second
-var saveMinResolvedTSInterval = 1 * time.Second
 
 const (
 	clientTimeout              = 3 * time.Second
@@ -252,6 +251,7 @@ func (c *RaftCluster) Start(s Server) error {
 	c.regionStats = statistics.NewRegionStatistics(c.opt, c.ruleManager)
 	c.limiter = NewStoreLimiter(s.GetPersistOptions())
 	c.unsafeRecoveryController = newUnsafeRecoveryController(cluster)
+	saveMinResolvedTSInterval := c.opt.GetSaveMinResolvedTSInterval()
 
 	c.wg.Add(6)
 	go c.runCoordinator()
@@ -1691,9 +1691,11 @@ func (c *RaftCluster) SetMinResolvedTS(storeID, minResolvedTS uint64) error {
 }
 
 func (c *RaftCluster) runMinResolvedTSJob(saveInterval time.Duration) {
-	defer logutil.LogPanic()
 	defer c.wg.Done()
-
+	if saveInterval == 0 {
+		return
+	}
+	defer logutil.LogPanic()
 	ticker := time.NewTicker(saveInterval)
 	defer ticker.Stop()
 	for {
@@ -1705,8 +1707,8 @@ func (c *RaftCluster) runMinResolvedTSJob(saveInterval time.Duration) {
 			minResolvedTS := c.GetMinResolvedTS()
 			if minResolvedTS != math.MaxUint64 {
 				c.Lock()
-				defer c.Unlock()
 				c.storage.SaveMinResolvedTS(minResolvedTS)
+				c.Unlock()
 			}
 		}
 	}

server/config/config.go

@@ -228,12 +228,13 @@ const (
 
 	defaultLeaderPriorityCheckInterval = time.Minute
 
-	defaultUseRegionStorage  = true
-	defaultTraceRegionFlow   = true
-	defaultFlowRoundByDigit  = 3 // KB
-	maxTraceFlowRoundByDigit = 5 // 0.1 MB
-	defaultMaxResetTSGap     = 24 * time.Hour
-	defaultKeyType           = "table"
+	defaultUseRegionStorage          = true
+	defaultTraceRegionFlow           = true
+	defaultFlowRoundByDigit          = 3 // KB
+	maxTraceFlowRoundByDigit         = 5 // 0.1 MB
+	defaultMaxResetTSGap             = 24 * time.Hour
+	defaultSaveMinResolvedTSInterval = 0
+	defaultKeyType                   = "table"
 
 	defaultStrictlyMatchLabel   = false
 	defaultEnablePlacementRules = true
@@ -1102,6 +1103,8 @@ type PDServerConfig struct {
 	TraceRegionFlow bool `toml:"trace-region-flow" json:"trace-region-flow,string,omitempty"`
 	// FlowRoundByDigit used to discretization processing flow information.
 	FlowRoundByDigit int `toml:"flow-round-by-digit" json:"flow-round-by-digit"`
+	// SaveMinResolvedTSInterval is the interval to save the min resolved ts.
+	SaveMinResolvedTSInterval typeutil.Duration `toml:"save-min-resolved-ts-interval" json:"save-min-resolved-ts-interval"`
 }
 
 func (c *PDServerConfig) adjust(meta *configMetaData) error {
@@ -1124,6 +1127,9 @@ func (c *PDServerConfig) adjust(meta *configMetaData) error {
 	if !meta.IsDefined("flow-round-by-digit") {
 		adjustInt(&c.FlowRoundByDigit, defaultFlowRoundByDigit)
 	}
+	if !meta.IsDefined("save-min-resolved-ts-interval") {
+		adjustDuration(&c.SaveMinResolvedTSInterval, defaultSaveMinResolvedTSInterval)
+	}
 	c.migrateConfigurationFromFile(meta)
 	return c.Validate()
 }

server/config/persist_options.go

@@ -653,6 +653,11 @@ func (o *PersistOptions) CheckLabelProperty(typ string, labels []*metapb.StoreLa
 	return false
 }
 
+// GetSaveMinResolvedTSInterval gets the interval for PD to save min resolved ts.
+func (o *PersistOptions) GetSaveMinResolvedTSInterval() time.Duration {
+	return o.GetPDServerConfig().SaveMinResolvedTSInterval.Duration
+}
+
 const ttlConfigPrefix = "/config/ttl"
 
 // SetTTLData set temporary configuration

server/storage/endpoint/min_resolved_ts.go

@@ -21,7 +21,6 @@ import (
 // MinResolvedTSPoint is the min resolved ts for a store
 // NOTE: This type is exported by HTTP API. Please pay more attention when modifying it.
 type MinResolvedTSPoint struct {
-	StoreID       uint64 `json:"store_id"`
 	MinResolvedTS uint64 `json:"min_resolved_ts"`
 }
 

tests/pdctl/config/config_test.go

@@ -186,6 +186,12 @@ func (s *configTestSuite) TestConfig(c *C) {
 	c.Assert(json.Unmarshal(output, &labelPropertyCfg), IsNil)
 	c.Assert(labelPropertyCfg, DeepEquals, svr.GetLabelProperty())
 
+	// config set save-min-resolved-ts-interval <value>
+	args = []string{"-u", pdAddr, "config", "set", "save-min-resolved-ts-interval", "1s"}
+	_, err = pdctl.ExecuteCommand(cmd, args...)
+	c.Assert(err, IsNil)
+	c.Assert(svr.GetPDServerConfig().SaveMinResolvedTSInterval, Equals, typeutil.NewDuration(time.Second))
+
 	// test config read and write
 	testItems := []testItem{
 		{"leader-schedule-limit", uint64(64), func(scheduleConfig *config.ScheduleConfig) interface{} {