disttask: maintain managed nodes separately #49623
Conversation
ti-chi-bot
bot
added
release-note-none
Codecov Report
Additional details and impacted files@@ Coverage Diff @@
## master #49623 +/- ##
================================================
+ Coverage 70.9801% 71.4163% +0.4362%
================================================
Files 1368 1427 +59
Lines 398761 423088 +24327
================================================
+ Hits 283041 302154 +19113
- Misses 95945 102010 +6065
+ Partials 19775 18924 -851
Flags with carried forward coverage won't be shown. Click here to find out more.
|
| return nil, err | ||
| } | ||
| logutil.Logger(s.logCtx).Debug("eligible instances", zap.Int("num", len(serverNodes))) | ||
| if len(serverNodes) == 0 { |
There was a problem hiding this comment.
Then import into can't scale out nodes during execution?
There was a problem hiding this comment.
this part is same as before
if task can only run on some nodes(len > 0), we use it, else we can use all managed nodes
There was a problem hiding this comment.
I get it, instance is for local mode.
| @@ -173,8 +172,7 @@ func TestBackfillingSchedulerGlobalSortMode(t *testing.T) { | |||
| taskID, err := mgr.CreateTask(ctx, task.Key, proto.Backfill, 1, task.Meta) | |||
| require.NoError(t, err) | |||
| task.ID = taskID | |||
| serverInfos, _, err := sch.GetEligibleInstances(context.Background(), task) | |||
| require.NoError(t, err) | |||
| serverInfos := []string{":4000"} | |||
There was a problem hiding this comment.
| serverInfos := []string{":4000"} | |
| execIDs := []string{":4000"} |
| // if returned instances is empty, it means all instances are eligible. | ||
| // TODO: run import from server disk using framework makes this logic complicated, | ||
| // the instance might not be managed by framework. | ||
| GetEligibleInstances(ctx context.Context, task *proto.Task) ([]string, error) |
There was a problem hiding this comment.
Maybe we can remove this interface?
There was a problem hiding this comment.
not now, server disk import requires this
| @@ -120,9 +121,18 @@ func (sm *Manager) Start() { | |||
| failpoint.Inject("disableSchedulerManager", func() { | |||
| failpoint.Return() | |||
| }) | |||
| // init cached managed nodes | |||
| sm.nodeMgr.refreshManagedNodes(sm.ctx, sm.taskMgr) | |||
There was a problem hiding this comment.
it's called periodicly per second, here just init it to make test pass
| return nil, err | ||
| } | ||
| logutil.Logger(s.logCtx).Debug("eligible instances", zap.Int("num", len(serverNodes))) | ||
| if len(serverNodes) == 0 { |
There was a problem hiding this comment.
I get it, instance is for local mode.
|
/cc @tangenta |
| type NodeManager struct { | ||
| // prevLiveNodes is used to record the live nodes in last checking. | ||
| prevLiveNodes map[string]struct{} | ||
| managedNodes atomic.Pointer[[]string] |
There was a problem hiding this comment.
add some comments for managedNodes.
Co-authored-by: EasonBall <[email protected]>
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: okJiang, ywqzzy The full list of commands accepted by this bot can be found here. The pull request process is described here
Needs approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
bot
added
approved
lgtm
and removed
needs-1-more-lgtm
What problem does this PR solve?
Issue Number: ref #49008
Problem Summary:
What changed and how does it work?
GetEligibleInstancesreturns target node that task can run, if no node returned, we use all.Check List
Tests
Side effects
Documentation
Release note
Please refer to Release Notes Language Style Guide to write a quality release note.