From a682e62d074685da4307c0399545744a5c6f6cfb Mon Sep 17 00:00:00 2001 From: CooooolFrog Date: Wed, 11 Mar 2026 20:08:35 +0800 Subject: [PATCH 1/2] fix(job): support config sync peer task batch size Signed-off-by: CooooolFrog --- internal/job/types.go | 7 ++++++ manager/config/config.go | 10 +++++--- manager/config/constants.go | 6 +++-- manager/job/sync_peers.go | 50 +++++++++++++++++++++++++++++-------- scheduler/job/job.go | 41 ++++++++++++++++++++++++++++-- 5 files changed, 96 insertions(+), 18 deletions(-) diff --git a/internal/job/types.go b/internal/job/types.go index 60b2021fec6..91a941c0306 100644 --- a/internal/job/types.go +++ b/internal/job/types.go @@ -149,3 +149,10 @@ type DeleteFailureTask struct { HostType string `json:"host_type"` Description string `json:"description"` } + +// SyncPeerRequest defines the request parameters for sync peer task. +type SyncPeerRequest struct { + GroupUUID string `json:"group_uuid"` + GroupTaskCount int `json:"group_task_count"` + TotalGroupTask int `json:"total_group_task"` +} diff --git a/manager/config/config.go b/manager/config/config.go index 556b6e4f8e9..bc868ad1796 100644 --- a/manager/config/config.go +++ b/manager/config/config.go @@ -354,6 +354,9 @@ type SyncPeersConfig struct { // BatchSize is the batch size when operating gorm database. BatchSize int `yaml:"batchSize" mapstructure:"batchSize"` + + // SyncBatchSize is the batch size when get host info from scheduler. + SyncBatchSize int `yaml:"syncBatchSize" mapstructure:"syncBatchSize"` } type PreheatTLSClientConfig struct { @@ -446,9 +449,10 @@ func New() *Config { TLS: PreheatTLSClientConfig{}, }, SyncPeers: SyncPeersConfig{ - Interval: DefaultJobSyncPeersInterval, - Timeout: DefaultJobSyncPeersTimeout, - BatchSize: DefaultJobSyncPeersBatchSize, + Interval: DefaultJobSyncPeersInterval, + Timeout: DefaultJobSyncPeersTimeout, + BatchSize: DefaultJobSyncPeersDatabaseBatchSize, + SyncBatchSize: DefaultJobSyncPeersBatchSize, }, }, Metrics: MetricsConfig{ diff --git a/manager/config/constants.go b/manager/config/constants.go index 87aa1a662bb..6aac10010ce 100644 --- a/manager/config/constants.go +++ b/manager/config/constants.go @@ -112,8 +112,10 @@ const ( // DefaultClusterJobRateLimit is default rate limit(requests per second) for job Open API by cluster. DefaultClusterJobRateLimit uint32 = 10 - // DefaultJobSyncPeersBatchSize is the default batch size for syncing all peers information from the scheduler and - // operating on the database. + // DefaultJobSyncPeersDatabaseBatchSize is the default batch size for operating on the database. + DefaultJobSyncPeersDatabaseBatchSize = 1 + + // DefaultJobSyncPeersBatchSize is the default batch size for syncing all peers information from the scheduler. DefaultJobSyncPeersBatchSize = 500 ) diff --git a/manager/job/sync_peers.go b/manager/job/sync_peers.go index b74f7bdd001..80fd7e46dee 100644 --- a/manager/job/sync_peers.go +++ b/manager/job/sync_peers.go @@ -22,6 +22,7 @@ import ( "context" "errors" "fmt" + "reflect" "sync" "time" @@ -161,25 +162,52 @@ func (s *syncPeers) createSyncPeers(ctx context.Context, scheduler models.Schedu return nil, err } - // Initialize task signature. - task := &machineryv1tasks.Signature{ - UUID: fmt.Sprintf("task_%s", uuid.New().String()), - Name: internaljob.SyncPeersJob, - RoutingKey: queue.String(), + var tasks []*machineryv1tasks.Signature + for i := 0; i < s.config.Job.SyncPeers.SyncBatchSize; i++ { + // Initialize task signature. + task := &machineryv1tasks.Signature{ + UUID: fmt.Sprintf("task_%s", uuid.New().String()), + Name: internaljob.SyncPeersJob, + RoutingKey: queue.String(), + } + tasks = append(tasks, task) + } + group, err := machineryv1tasks.NewGroup(tasks...) + if err != nil { + return nil, err + } + if len(tasks) <= 0 { + return []*resource.Host{}, nil + } + for i, signature := range group.Tasks { + // Set signature args. + args, err := internaljob.MarshalRequest(internaljob.SyncPeerRequest{ + GroupUUID: group.GroupUUID, + GroupTaskCount: i, + TotalGroupTask: len(group.Tasks), + }) + if err != nil { + logger.Errorf("[sync-peers] sync peer task marshal request: %v, error: %v", args, err) + return nil, err + } + signature.Args = args } - // Send sync peer task to worker. - logger.Infof("[sync-peers] create sync peers in queue %v, task: %#v", queue, task) - asyncResult, err := s.job.Server.SendTaskWithContext(ctx, task) + logger.Infof("[sync-peers] create sync peers in queue %v, tasksNumber %d, taskUUID %v", queue, len(tasks), tasks[0].UUID) + asyncResults, err := s.job.Server.SendGroupWithContext(ctx, group, 0) if err != nil { logger.Errorf("[sync-peers] create sync peers in queue %v failed", queue, err) return nil, err } // Get sync peer task result. - results, err := asyncResult.GetWithTimeout(s.config.Job.SyncPeers.Timeout, DefaultTaskPollingInterval) - if err != nil { - return nil, err + results := make([]reflect.Value, len(asyncResults)) + for _, asyncResult := range asyncResults { + result, err := asyncResult.GetWithTimeout(s.config.Job.SyncPeers.Timeout, DefaultTaskPollingInterval) + if err != nil { + return nil, err + } + results = append(results, result...) } // Unmarshal sync peer task result. diff --git a/scheduler/job/job.go b/scheduler/job/job.go index 54f7f392f28..cb26da6a412 100644 --- a/scheduler/job/job.go +++ b/scheduler/job/job.go @@ -25,6 +25,7 @@ import ( "io" "net" "slices" + "sort" "strconv" "sync" @@ -860,7 +861,18 @@ func (j *job) selectPeers(ips []string, count *uint32, percentage *uint32, log * } // syncPeers is a job to sync peers. -func (j *job) syncPeers() (string, error) { +func (j *job) syncPeers(_ context.Context, data string) (string, error) { + req := &internaljob.SyncPeerRequest{} + if err := internaljob.UnmarshalRequest(data, req); err != nil { + logger.Errorf("[sync-peers] unmarshal request err: %s, request body: %s", err.Error(), data) + return "", err + } + + if err := validator.New().Struct(req); err != nil { + logger.Errorf("[sync-peers] sync peer task %s validate failed: %s", req.GroupUUID, err.Error()) + return "", err + } + hosts := make([]*resource.Host, 0, j.resource.HostManager().Len()) j.resource.HostManager().Range(func(key, value any) bool { host, ok := value.(*resource.Host) @@ -873,7 +885,32 @@ func (j *job) syncPeers() (string, error) { return true }) - return internaljob.MarshalResponse(hosts) + // Sort hosts by ID. + sort.Slice(hosts, func(i, j int) bool { + return hosts[i].ID < hosts[j].ID + }) + + // We ignore changes in hosts data across multiple task requests within the same group. + // TODO: (Could this cause problems? Perhaps we should create a snapshot of the hosts for same group to ensure that the host information is completely accurate.) + // Split hosts by task count. + total := len(hosts) + groupSize := total / req.TotalGroupTask + remainder := total % req.TotalGroupTask + + var start int + if req.GroupTaskCount < remainder { + start = req.GroupTaskCount * (groupSize + 1) + } else { + start = remainder*(groupSize+1) + (req.GroupTaskCount-remainder)*groupSize + } + end := start + groupSize + if req.GroupTaskCount < remainder { + end++ + } + + taskHosts := hosts[start:end] + + return internaljob.MarshalResponse(taskHosts) } // getTask is a job to get task. From bcf97b2499e046273f0db5f38d1dcb94d211e31b Mon Sep 17 00:00:00 2001 From: CooooolFrog Date: Tue, 31 Mar 2026 15:56:36 +0800 Subject: [PATCH 2/2] fix results slice Signed-off-by: CooooolFrog --- manager/job/sync_peers.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/manager/job/sync_peers.go b/manager/job/sync_peers.go index 80fd7e46dee..49afd0d614b 100644 --- a/manager/job/sync_peers.go +++ b/manager/job/sync_peers.go @@ -172,13 +172,13 @@ func (s *syncPeers) createSyncPeers(ctx context.Context, scheduler models.Schedu } tasks = append(tasks, task) } + if len(tasks) <= 0 { + return []*resource.Host{}, nil + } group, err := machineryv1tasks.NewGroup(tasks...) if err != nil { return nil, err } - if len(tasks) <= 0 { - return []*resource.Host{}, nil - } for i, signature := range group.Tasks { // Set signature args. args, err := internaljob.MarshalRequest(internaljob.SyncPeerRequest{ @@ -201,7 +201,7 @@ func (s *syncPeers) createSyncPeers(ctx context.Context, scheduler models.Schedu } // Get sync peer task result. - results := make([]reflect.Value, len(asyncResults)) + results := make([]reflect.Value, 0, len(asyncResults)) for _, asyncResult := range asyncResults { result, err := asyncResult.GetWithTimeout(s.config.Job.SyncPeers.Timeout, DefaultTaskPollingInterval) if err != nil {