Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

schedule: fix split-merge-interval update #8405

Merged
merged 4 commits into from
Jul 17, 2024
Merged

schedule: fix split-merge-interval update #8405

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
473c85d
schedule: fix split-merge-interval update
lhy1024 Jul 16, 2024
b4a11a5
add unit test
lhy1024 Jul 16, 2024
d5d7756
address comments
lhy1024 Jul 17, 2024
49285b1
Merge branch 'master' into fix-config-update
ti-chi-bot[bot] Jul 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions pkg/schedule/checker/merge_checker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,8 @@ const (
mergeOptionValueDeny = "deny"
)

var gcInterval = time.Minute

// MergeChecker ensures region to merge with adjacent region when size is small
type MergeChecker struct {
PauseController
Expand All @@ -57,7 +59,7 @@ type MergeChecker struct {

// NewMergeChecker creates a merge checker.
func NewMergeChecker(ctx context.Context, cluster sche.CheckerCluster, conf config.CheckerConfigProvider) *MergeChecker {
splitCache := cache.NewIDTTL(ctx, time.Minute, conf.GetSplitMergeInterval())
splitCache := cache.NewIDTTL(ctx, gcInterval, conf.GetSplitMergeInterval())
return &MergeChecker{
cluster: cluster,
conf: conf,
Expand Down Expand Up @@ -88,13 +90,16 @@ func (m *MergeChecker) Check(region *core.RegionInfo) []*operator.Operator {
return nil
}

// update the split cache.
// It must be called before the following merge checker logic.
m.splitCache.UpdateTTL(m.conf.GetSplitMergeInterval())
Copy link
Member

@rleungx rleungx Jul 17, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BTW, will it have a risk? If the heartbeat is not sent in time?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It has the same behavior as before in this case.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

At the same time, we wait until 90% of the heartbeats are collected before checking.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There still be 10% and merge check may create lots of scheduling?

Copy link
Contributor Author

@lhy1024 lhy1024 Jul 17, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The behavior is similar in this case. For example, if split-merge-interval is set to 1s.
In the past, it would pass the recently started check and then run UpdateTTL
Now, it will run UpdateTTL first and then pass the recently started check.

Perhaps we can add a longer time before recently started check, such as expireTime := m.startTime.Add(m.conf.GetSplitMergeInterval()+5*time.Minute)

Or we can add a filter to check whether the region reports a heartbeat after PD started. Will L109 be helpful?


expireTime := m.startTime.Add(m.conf.GetSplitMergeInterval())
if time.Now().Before(expireTime) {
mergeCheckerRecentlyStartCounter.Inc()
return nil
}

m.splitCache.UpdateTTL(m.conf.GetSplitMergeInterval())
if m.splitCache.Exists(region.GetID()) {
mergeCheckerRecentlySplitCounter.Inc()
return nil
Expand Down
16 changes: 16 additions & 0 deletions pkg/schedule/checker/merge_checker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ func TestMergeCheckerTestSuite(t *testing.T) {

func (suite *mergeCheckerTestSuite) SetupTest() {
cfg := mockconfig.NewTestOptions()
gcInterval = 100 * time.Millisecond
lhy1024 marked this conversation as resolved.
Show resolved Hide resolved
suite.ctx, suite.cancel = context.WithCancel(context.Background())
suite.cluster = mockcluster.NewCluster(suite.ctx, cfg)
suite.cluster.SetMaxMergeRegionSize(2)
Expand Down Expand Up @@ -84,6 +85,7 @@ func (suite *mergeCheckerTestSuite) SetupTest() {
}

func (suite *mergeCheckerTestSuite) TearDownTest() {
gcInterval = time.Minute
suite.cancel()
}

Expand Down Expand Up @@ -234,6 +236,7 @@ func (suite *mergeCheckerTestSuite) TestBasic() {
ops = suite.mc.Check(suite.regions[3])
re.Nil(ops)

// issue #4616
suite.cluster.SetSplitMergeInterval(500 * time.Millisecond)
ops = suite.mc.Check(suite.regions[2])
re.Nil(ops)
Expand All @@ -245,6 +248,19 @@ func (suite *mergeCheckerTestSuite) TestBasic() {
re.NotNil(ops)
ops = suite.mc.Check(suite.regions[3])
re.NotNil(ops)

// issue #8405
suite.mc.startTime = time.Now()
suite.cluster.SetSplitMergeInterval(time.Second)
suite.cluster.SetSplitMergeInterval(time.Hour)
suite.mc.RecordRegionSplit([]uint64{suite.regions[2].GetID()})
suite.cluster.SetSplitMergeInterval(time.Second)
suite.mc.Check(suite.regions[2]) // trigger the config update
time.Sleep(time.Second) // wait for the cache to gc
ops = suite.mc.Check(suite.regions[2])
re.NotNil(ops)
ops = suite.mc.Check(suite.regions[3])
re.NotNil(ops)
}

func (suite *mergeCheckerTestSuite) TestMatchPeers() {
Expand Down