enhance: support syncpeers by service and optimize the merge logic

go.mod

@@ -62,6 +62,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.5
 	github.com/redis/go-redis/v9 v9.6.1
+	github.com/samber/lo v1.47.0
 	github.com/schollz/progressbar/v3 v3.17.0
 	github.com/shirou/gopsutil/v3 v3.24.5
 	github.com/soheilhy/cmux v0.1.5

go.sum

@@ -1384,6 +1384,8 @@ github.com/sagikazarmark/locafero v0.4.0 h1:HApY1R9zGo4DBgr7dqsTH/JJxLTTsOt7u6ke
 github.com/sagikazarmark/locafero v0.4.0/go.mod h1:Pe1W6UlPYUk/+wc/6KFhbORCfqzgYEpgQ3O5fPuL3H4=
 github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6gto+ugjYE=
 github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ=
+github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc=
+github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/samuel/go-zookeeper v0.0.0-20201211165307-7117e9ea2414 h1:AJNDS0kP60X8wwWFvbLPwDuojxubj9pbfK7pjHw0vKg=
 github.com/samuel/go-zookeeper v0.0.0-20201211165307-7117e9ea2414/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=

manager/config/config.go

@@ -338,6 +338,9 @@ type SyncPeersConfig struct {
 
 	// Timeout is the timeout for syncing peers information from the single scheduler.
 	Timeout time.Duration `yaml:"timeout" mapstructure:"timeout"`
+
+	// BatchSize is the batch size when operating gorm.
+	BatchSize int `yaml:"batchSize" mapstructure:"batchSize"`
 }
 
 type PreheatTLSClientConfig struct {
@@ -447,8 +450,9 @@ func New() *Config {
 				TLS:             PreheatTLSClientConfig{},
 			},
 			SyncPeers: SyncPeersConfig{
-				Interval: DefaultJobSyncPeersInterval,
-				Timeout:  DefaultJobSyncPeersTimeout,
+				Interval:  DefaultJobSyncPeersInterval,
+				Timeout:   DefaultJobSyncPeersTimeout,
+				BatchSize: DefaultJobSyncPeersBatchSize,
 			},
 		},
 		ObjectStorage: ObjectStorageConfig{

manager/config/constants.go

@@ -107,6 +107,9 @@ const (
 
 	// DefaultClusterJobRateLimit is default rate limit(requests per second) for job Open API by cluster.
 	DefaultClusterJobRateLimit = 10
+
+	// DefaultJobSyncPeersBatchSize is the default batch size for syncing all peers information from the scheduler.
+	DefaultJobSyncPeersBatchSize = 500
 )
 
 const (

manager/handlers/job.go

@@ -62,6 +62,20 @@ func (h *Handlers) CreateJob(ctx *gin.Context) {
 			return
 		}
 
+		ctx.JSON(http.StatusOK, job)
+	case job.SyncPeersJob:
+		var json types.CreateSyncPeersJobRequest
+		if err := ctx.ShouldBindBodyWith(&json, binding.JSON); err != nil {
+			ctx.JSON(http.StatusUnprocessableEntity, gin.H{"errors": err.Error()})
+			return
+		}
+
+		job, err := h.service.CreateSyncPeersJob(ctx.Request.Context(), json)
+		if err != nil {
+			ctx.Error(err) // nolint: errcheck
+			return
+		}
+
 		ctx.JSON(http.StatusOK, job)
 	case job.GetTaskJob:
 		var json types.CreateGetTaskJobRequest