enhance: support syncpeers by service and optimize the merge logic

Signed-off-by: cormick <cormick1080@gmail.com>

go.mod

@@ -60,6 +60,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

@@ -1304,6 +1304,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/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/schollz/progressbar/v3 v3.17.0 h1:Fv+vG6O6jnJwdjCelvfyYO7sF2jaUGQVmdH4CxcZdsQ=

manager/config/config.go

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

manager/config/constants.go

@@ -113,6 +113,9 @@ const (
 
 	// DefaultJobSyncPeersTimeout is the default timeout for syncing all peers information from the scheduler.
 	DefaultJobSyncPeersTimeout = 10 * time.Minute
+
+	// DefaultJobSyncPeersBatchSize is the default batch size for syncing all peers information from the scheduler.
+	DefaultJobSyncPeersBatchSize = 500
 )
 
 const (

manager/database/mysql.go

@@ -71,14 +71,14 @@ func newMysql(cfg *config.Config) (*gorm.DB, error) {
 		return nil, err
 	}
 
-	// Run migration.
+	// AsyncSyncPeers migration.
 	if mysqlCfg.Migrate {
 		if err := migrate(db); err != nil {
 			return nil, err
 		}
 	}
 
-	// Run seed.
+	// AsyncSyncPeers seed.
 	if err := seed(db); err != nil {
 		return nil, err
 	}

manager/database/postgres.go

@@ -49,14 +49,14 @@ func newPostgres(cfg *config.Config) (*gorm.DB, error) {
 		return nil, err
 	}
 
-	// Run migration.
+	// AsyncSyncPeers migration.
 	if postgresCfg.Migrate {
 		if err := migrate(db); err != nil {
 			return nil, err
 		}
 	}
 
-	// Run seed.
+	// AsyncSyncPeers seed.
 	if err := seed(db); err != nil {
 		return nil, err
 	}

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