feat: optimize implement of the sync peers

Signed-off-by: Gaius <[email protected]>

.github/workflows/compatibility-e2e-v2.yml

@@ -31,19 +31,19 @@ jobs:
         include:
           - module: manager
             image: manager
-            image-tag: v2.1.60
+            image-tag: v2.1.65
             chart-name: manager
           - module: scheduler
             image: scheduler
-            image-tag: v2.1.60
+            image-tag: v2.1.65
             chart-name: scheduler
           - module: client
             image: client
-            image-tag: v0.1.115
+            image-tag: v0.1.119
             chart-name: client
           - module: seed-client
             image: client
-            image-tag: v0.1.115
+            image-tag: v0.1.119
             chart-name: seed-client
 
     steps:

api/manager/docs.go

@@ -4202,28 +4202,34 @@ const docTemplate = `{
             ],
             "properties": {
                 "args": {
+                    "description": "Args is the arguments of the job.",
                     "type": "object",
                     "additionalProperties": {}
                 },
                 "bio": {
+                    "description": "BIO is the description of the job.",
                     "type": "string"
                 },
                 "scheduler_cluster_ids": {
+                    "description": "SchedulerClusterIDs is the scheduler cluster ids of the job.",
                     "type": "array",
                     "items": {
                         "type": "integer"
                     }
                 },
                 "seed_peer_cluster_ids": {
+                    "description": "SeedPeerClusterIDs is the seed peer cluster ids of the job.",
                     "type": "array",
                     "items": {
                         "type": "integer"
                     }
                 },
                 "type": {
+                    "description": "Type is the type of the job.",
                     "type": "string"
                 },
                 "user_id": {
+                    "description": "UserID is the user id of the job.",
                     "type": "integer"
                 }
             }
@@ -4682,6 +4688,11 @@ const docTemplate = `{
                     "type": "integer",
                     "maximum": 1000,
                     "minimum": 10
+                },
+                "job_rate_limit": {
+                    "type": "integer",
+                    "maximum": 1000000,
+                    "minimum": 1
                 }
             }
         },
@@ -4881,9 +4892,11 @@ const docTemplate = `{
             "type": "object",
             "properties": {
                 "bio": {
+                    "description": "BIO is the description of the job.",
                     "type": "string"
                 },
                 "user_id": {
+                    "description": "UserID is the user id of the job.",
                     "type": "integer"
                 }
             }

api/manager/swagger.json

@@ -4196,28 +4196,34 @@
             ],
             "properties": {
                 "args": {
+                    "description": "Args is the arguments of the job.",
                     "type": "object",
                     "additionalProperties": {}
                 },
                 "bio": {
+                    "description": "BIO is the description of the job.",
                     "type": "string"
                 },
                 "scheduler_cluster_ids": {
+                    "description": "SchedulerClusterIDs is the scheduler cluster ids of the job.",
                     "type": "array",
                     "items": {
                         "type": "integer"
                     }
                 },
                 "seed_peer_cluster_ids": {
+                    "description": "SeedPeerClusterIDs is the seed peer cluster ids of the job.",
                     "type": "array",
                     "items": {
                         "type": "integer"
                     }
                 },
                 "type": {
+                    "description": "Type is the type of the job.",
                     "type": "string"
                 },
                 "user_id": {
+                    "description": "UserID is the user id of the job.",
                     "type": "integer"
                 }
             }
@@ -4676,6 +4682,11 @@
                     "type": "integer",
                     "maximum": 1000,
                     "minimum": 10
+                },
+                "job_rate_limit": {
+                    "type": "integer",
+                    "maximum": 1000000,
+                    "minimum": 1
                 }
             }
         },
@@ -4875,9 +4886,11 @@
             "type": "object",
             "properties": {
                 "bio": {
+                    "description": "BIO is the description of the job.",
                     "type": "string"
                 },
                 "user_id": {
+                    "description": "UserID is the user id of the job.",
                     "type": "integer"
                 }
             }

api/manager/swagger.yaml

@@ -461,20 +461,26 @@ definitions:
     properties:
       args:
         additionalProperties: {}
+        description: Args is the arguments of the job.
         type: object
       bio:
+        description: BIO is the description of the job.
         type: string
       scheduler_cluster_ids:
+        description: SchedulerClusterIDs is the scheduler cluster ids of the job.
         items:
           type: integer
         type: array
       seed_peer_cluster_ids:
+        description: SeedPeerClusterIDs is the seed peer cluster ids of the job.
         items:
           type: integer
         type: array
       type:
+        description: Type is the type of the job.
         type: string
       user_id:
+        description: UserID is the user id of the job.
         type: integer
     required:
     - type
@@ -792,6 +798,10 @@ definitions:
         maximum: 1000
         minimum: 10
         type: integer
+      job_rate_limit:
+        maximum: 1000000
+        minimum: 1
+        type: integer
     type: object
   d7y_io_dragonfly_v2_manager_types.SchedulerClusterScopes:
     properties:
@@ -925,8 +935,10 @@ definitions:
   d7y_io_dragonfly_v2_manager_types.UpdateJobRequest:
     properties:
       bio:
+        description: BIO is the description of the job.
         type: string
       user_id:
+        description: UserID is the user id of the job.
         type: integer
     type: object
   d7y_io_dragonfly_v2_manager_types.UpdateOauthRequest:

go.mod

@@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
 go 1.23.0
 
 require (
-	d7y.io/api/v2 v2.0.175
+	d7y.io/api/v2 v2.0.177
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/RichardKnop/machinery v1.10.8
 	github.com/Showmax/go-fqdn v1.0.0

go.sum

@@ -53,8 +53,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-d7y.io/api/v2 v2.0.175 h1:yE1FeYnBEK/geHmDJbqXB0pUXtPBtqk9E7xijIVh0AA=
-d7y.io/api/v2 v2.0.175/go.mod h1:+l4ErhthKmcIhcRU6F01Km8q+yDyICF7JImefg0t6HY=
+d7y.io/api/v2 v2.0.177 h1:iC+Jm4n7lKs3N1JIO25XOdtELbKSlis85LEoGbYlp98=
+d7y.io/api/v2 v2.0.177/go.mod h1:+l4ErhthKmcIhcRU6F01Km8q+yDyICF7JImefg0t6HY=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0/go.mod h1:uGG2W01BaETf0Ozp+QxxKJdMBNRWPdstHG0Fmdwn1/U=

manager/config/config.go

@@ -339,7 +339,7 @@ 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 is the batch size when operating gorm database.
 	BatchSize int `yaml:"batchSize" mapstructure:"batchSize"`
 }
 
@@ -641,6 +641,10 @@ func (cfg *Config) Validate() error {
 		return errors.New("syncPeers requires parameter timeout")
 	}
 
+	if cfg.Job.SyncPeers.BatchSize == 0 {
+		return errors.New("syncPeers requires parameter batchSize")
+	}
+
 	if cfg.ObjectStorage.Enable {
 		if cfg.ObjectStorage.Name == "" {
 			return errors.New("objectStorage requires parameter name")

manager/config/config_test.go

@@ -188,8 +188,9 @@ func TestConfig_Load(t *testing.T) {
 				},
 			},
 			SyncPeers: SyncPeersConfig{
-				Interval: 13 * time.Hour,
-				Timeout:  2 * time.Minute,
+				Interval:  13 * time.Hour,
+				Timeout:   2 * time.Minute,
+				BatchSize: 50,
 			},
 		},
 		ObjectStorage: ObjectStorageConfig{
@@ -809,6 +810,21 @@ func TestConfig_Validate(t *testing.T) {
 				assert.EqualError(err, "syncPeers requires parameter timeout")
 			},
 		},
+		{
+			name:   "syncPeers requires parameter batchSize",
+			config: New(),
+			mock: func(cfg *Config) {
+				cfg.Auth.JWT = mockJWTConfig
+				cfg.Database.Type = DatabaseTypeMysql
+				cfg.Database.Mysql = mockMysqlConfig
+				cfg.Database.Redis = mockRedisConfig
+				cfg.Job.SyncPeers.BatchSize = 0
+			},
+			expect: func(t *testing.T, err error) {
+				assert := assert.New(t)
+				assert.EqualError(err, "syncPeers requires parameter batchSize")
+			},
+		},
 		{
 			name:   "objectStorage requires parameter name",
 			config: New(),

manager/config/constants.go

@@ -108,7 +108,8 @@ 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 is the default batch size for syncing all peers information from the scheduler and
+	// operating on the database.
 	DefaultJobSyncPeersBatchSize = 500
 )
 

manager/config/testdata/manager.yaml

@@ -79,6 +79,7 @@ job:
   syncPeers:
     interval: 13h
     timeout: 2m
+    batchSize: 50
 
 objectStorage:
   enable: true

manager/handlers/job.go

@@ -70,13 +70,15 @@ func (h *Handlers) CreateJob(ctx *gin.Context) {
 			return
 		}
 
-		job, err := h.service.CreateSyncPeersJob(ctx.Request.Context(), json)
-		if err != nil {
+		// CreateSyncPeersJob is a sync operation, so don't need to return the job id,
+		// and not record the job information in the database. If return success, need to
+		// query the peers table to get the latest data.
+		if err := h.service.CreateSyncPeersJob(ctx.Request.Context(), json); err != nil {
 			ctx.Error(err) // nolint: errcheck
 			return
 		}
 
-		ctx.JSON(http.StatusOK, job)
+		ctx.JSON(http.StatusOK, http.StatusText(http.StatusOK))
 	case job.GetTaskJob:
 		var json types.CreateGetTaskJobRequest
 		if err := ctx.ShouldBindBodyWith(&json, binding.JSON); err != nil {

manager/job/job.go

@@ -30,7 +30,7 @@ import (
 )
 
 // DefaultTaskPollingInterval is the default interval for polling task.
-const DefaultTaskPollingInterval = 5 * time.Second
+const DefaultTaskPollingInterval = 10 * time.Second
 
 // tracer is a global tracer for job.
 var tracer = otel.Tracer("manager")