modify for code review

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

manager/database/mysql.go

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

manager/job/sync_peers.go

@@ -41,8 +41,8 @@ import (
 
 // SyncPeers is an interface for sync peers.
 type SyncPeers interface {
-	// AsyncSyncPeers execute action to sync peers, which is async.
-	AsyncSyncPeers(context.Context, SyncPeersArgs) error
+	// Run execute action to sync peers, which is async.
+	Run(context.Context, SyncPeersArgs) error
 
 	// Serve started sync peers server.
 	Serve()
@@ -70,15 +70,17 @@ type SyncPeersArgs struct {
 // newSyncPeers returns a new SyncPeers.
 func newSyncPeers(cfg *config.Config, job *internaljob.Job, gdb *gorm.DB) (SyncPeers, error) {
 	return &syncPeers{
-		config: cfg,
-		db:     gdb,
-		job:    job,
-		done:   make(chan struct{}),
+		config:     cfg,
+		db:         gdb,
+		job:        job,
+		done:       make(chan struct{}),
+		workChan:   make(chan SyncPeersArgs),
+		syncLocker: sync.Mutex{},
 	}, nil
 }
 
-// AsyncSyncPeers start to sync peers.
-func (s *syncPeers) AsyncSyncPeers(ctx context.Context, args SyncPeersArgs) error {
+// Run start to sync peers.
+func (s *syncPeers) Run(ctx context.Context, args SyncPeersArgs) error {
 	if len(args.CandidateSchedulerClusters) == 0 {
 		if err := s.db.WithContext(ctx).Find(&args.CandidateSchedulerClusters).Error; err != nil {
 			return fmt.Errorf("failed to get candidate scheduler clusters: %v", err)
@@ -95,10 +97,12 @@ func (s *syncPeers) Serve() {
 	for {
 		select {
 		case <-tick.C:
+			logger.Debugf("start to sync peerrs periodically")
 			if err := s.syncPeers(context.Background(), nil); err != nil {
 				logger.Errorf("sync peers failed periodically: %v", err)
 			}
 		case args := <-s.workChan:
+			logger.Debugf("start to sync peerrs for request")
 			err := s.syncPeers(context.Background(), args.CandidateSchedulerClusters)
 			if err != nil {
 				logger.Errorf("sync peers failed for request: %v", err)
@@ -110,7 +114,7 @@ func (s *syncPeers) Serve() {
 				if err == nil {
 					state = machineryv1tasks.StateSuccess
 				}
-				if updateErr := s.db.WithContext(context.Background()).First(&job, args.TaskID).Updates(models.Job{
+				if updateErr := s.db.WithContext(context.Background()).First(&job, "task_id = ?", args.TaskID).Updates(models.Job{
 					State: state,
 				}).Error; updateErr != nil {
 					logger.Errorf("update sync peers job result failed for request: %v", updateErr)

manager/service/job.go

@@ -36,6 +36,11 @@ import (
 )
 
 func (s *service) CreateSyncPeersJob(ctx context.Context, json types.CreateSyncPeersJobRequest) (*models.Job, error) {
+	args, err := structure.StructToMap(json)
+	if err != nil {
+		return nil, err
+	}
+
 	candidateSchedulers, err := s.findCandidateSchedulers(ctx, json.SchedulerClusterIDs, nil)
 	if err != nil {
 		return nil, err
@@ -48,7 +53,7 @@ func (s *service) CreateSyncPeersJob(ctx context.Context, json types.CreateSyncP
 
 	taskID := fmt.Sprintf("manager_%v", uuid.New().String())
 
-	if err = s.job.SyncPeers.AsyncSyncPeers(ctx, job.SyncPeersArgs{
+	if err = s.job.SyncPeers.Run(ctx, job.SyncPeersArgs{
 		CandidateSchedulerClusters: candidateClusters,
 		TaskID:                     taskID,
 	}); err != nil {
@@ -59,6 +64,7 @@ func (s *service) CreateSyncPeersJob(ctx context.Context, json types.CreateSyncP
 	job := models.Job{
 		TaskID:            taskID,
 		BIO:               json.BIO,
+		Args:              args,
 		Type:              json.Type,
 		State:             machineryv1tasks.StateStarted,
 		UserID:            json.UserID,