// Package dm add by blacknc at 2024/10/31 00:19:09 package dm import ( "context" "time" "github.com/pingcap/errors" "github.com/pingcap/tiflow/dm/openapi" "github.com/tidbcloud/dataflow-service/internal/db" "github.com/tidbcloud/dataflow-service/internal/db/entity" "github.com/tidbcloud/dataflow-service/internal/handler/agent" "github.com/tidbcloud/dataflow-service/pkg/log" "github.com/tidbcloud/eks-provider/api/v1alpha1" "go.uber.org/zap" "gorm.io/gorm" ) const ( DMJobAnnotationPaused = "dataflow.pingcap.com/paused" ) type StateSync struct { StateBase } func (s *StateSync) String() string { return "StateSync" } func (s *StateSync) Exec(c *TaskContext) (next TaskState, err error) { ctx := c.ctx dmJob := c.job.DMJob logger := log.GetLogger(ctx) m := c.model if !m.needSyncStage(dmJob) { logger.Info("skip sync phase because this dm job not required it") next = new(StateFinishing) return } dmJobCR, err := m.resourceAPI.GetDMJob(ctx, dmJob.ProjectID.Uint64(), dmJob.ID.Uint64()) if err != nil { // this should not happend logger.Error("failed to get dm job cr", zap.Error(err)) return } if dmJobCR.GetAnnotations()[DMJobAnnotationPaused] == "true" { logger.Info("dm job cr is paused, skip sync phase") return } jobSources, err := GetDMJobSources(dmJob.ID) if err != nil { logger.Error("failed get dm job source", zap.Error(err)) return nil, err } else if len(jobSources) == 0 { // should not happend logger.Error("dm job source is empty, this should not happend!") return nil, nil } jobSpec, err := GetJobSpecByID(dmJob.SpecID) if err != nil { logger.Error("failed to get dm job spec", zap.Error(err)) return nil, err } masterServerName := dmJobCR.Status.MasterServerName dmOpenapiClient := c.agentClient.GetDMOpenapiClient(dmJob.ClusterID.Uint64(), masterServerName) dataSourceName := generateDMDataSourceName(&jobSources[0]) // need to scale down dm worker instance for physical import mode // if no scale is required, or if the scale has already been completed, skip directly toSpec := dmJobCR.Spec.Worker.SpecCode if dmJob.ImportMode == entity.DMImportModePhysical { toSpec = jobSpec.Name } done, err := s.reconcileDMSpec(c, dmOpenapiClient, dmJobCR, toSpec, dataSourceName) if err != nil { if err != agent.ErrorDMResourceNotFound { logger.Error("failed to reconcile for dm sync task", zap.String("toSpec", toSpec), zap.String("dataSourceName", dataSourceName), zap.Error(err)) return nil, err } } else if !done { logger.Info("scaling worker for dm sync task, waiting for the next retry.", zap.String("toSpec", toSpec), zap.String("dataSourceName", dataSourceName)) return nil, nil } logger.Info("component version info", zap.String("dmJob.componentVersion", dmJob.ComponentVersion), zap.String("dmJobCR.componentVersion", dmJobCR.Spec.Worker.BaseImage)) if cerr := UpdateDMJobComponentVersionWithImage(dmJob, dmJobCR.Spec.Worker.BaseImage); cerr != nil { logger.Error("failed to update dm job component version", zap.Error(cerr)) } if dmJobCR.Status.Phase != v1alpha1.DMJobPhaseRunning { // this should not happend logger.Error("dm job cr not running", zap.Any("phase", dmJobCR.Status.Phase)) return } taskName := generateDMTaskName(dmJob.ID, entity.DMStageSync) logger = logger.With(zap.String("taskName", taskName)) ctx = log.WithLogger(ctx, logger) sourceNames := GetDatasourceNames(jobSources) // query sync task status, and create sync task if not found taskStatusResp, err := dmOpenapiClient.QueryTaskStatus(ctx, taskName, sourceNames) if err == agent.ErrorDMResourceNotFound { // create sync task if not exists _, err = s.createDMTask(c, dmJobCR, openapi.TaskTaskModeIncremental, taskName, dmOpenapiClient, jobSources) if err != nil { logger.Error("failed to create sync task", zap.Error(err)) err = errors.Annotate(err, "failed to create sync task") return } // start task if err = dmOpenapiClient.StartTask(ctx, taskName, &openapi.StartTaskRequest{ SourceNameList: &sourceNames, }); err != nil { logger.Error("failed to start sync task", zap.Error(err)) err = errors.Annotate(err, "failed to start sync task") return } return } else if err != nil { err = errors.Annotate(err, "failed to query sync task status") return } taskStatus := taskStatusResp.Data if len(taskStatus) == 0 { // should not happend err = errors.New("sync task status is empty, this should not happend!") return } for i := range taskStatus { if taskStatus[i].Unit != DMTaskUintSync { logger.Error("sync task status got unexpected unit", zap.Any("taskStatus", taskStatus), zap.Int("idx", i)) err = errors.Errorf("sync task status got unexpected unit:%s", taskStatus[i].Unit) return } } // save sync status if err = s.saveSyncStatus(ctx, taskStatus, jobSources); err != nil { logger.Warn("failed to save sync status", zap.Error(err)) } allFinished, err := s.checkSubTaskStatus(c, dmOpenapiClient, entity.DMStageSync, taskStatus) if err == nil && allFinished { // stop sync task to prevent it from running again // https://pingcap.feishu.cn/wiki/WTDUwRrHYiJAAikTzzYcTMA0nJd if err = dmOpenapiClient.StopTask(ctx, taskName, &openapi.StopTaskRequest{SourceNameList: &sourceNames}); err != nil { logger.Error("failed to stop sync task", zap.Error(err)) err = errors.Annotate(err, "failed to stop sync task") return } next = new(StateFinishing) } return } func (s *StateSync) saveSyncStatus(ctx context.Context, taskStatus []openapi.SubTaskStatus, jobSources []entity.DMJobSource) error { logger := log.GetLogger(ctx) statusMap := make(map[string]*openapi.SubTaskStatus, len(taskStatus)) for i := range taskStatus { statusMap[taskStatus[i].SourceName] = &taskStatus[i] } if err := db.Transaction(func(tx *gorm.DB) error { for _, jobSource := range jobSources { dataSourceName := generateDMDataSourceName(&jobSource) status := statusMap[dataSourceName] if status == nil { logger.Error("[saveSyncStatus]not found matched status for job source", zap.String("dataSourceName", dataSourceName), zap.Any("taskStatus", taskStatus)) continue } if jobSource.SyncStatus == nil { jobSource.SyncStatus = &entity.DMSyncStatus{BeginTime: time.Now().Unix()} } jobSource.SyncStatus.UpdateTime = time.Now().Unix() jobSource.SyncStatus.Stage = status.Stage if status.Stage == "" { jobSource.SyncStatus.Stage = openapi.TaskStageRunning } jobSource.SyncStatus.ErrorMsg = status.ErrorMsg if status.SyncStatus != nil && status.ErrorMsg != nil { jobSource.SyncStatus.Stage = status.Stage jobSource.SyncStatus.BinlogType = status.SyncStatus.BinlogType jobSource.SyncStatus.BlockingDdls = status.SyncStatus.BlockingDdls jobSource.SyncStatus.DumpIoTotalBytes = status.SyncStatus.DumpIoTotalBytes jobSource.SyncStatus.IoTotalBytes = status.SyncStatus.IoTotalBytes jobSource.SyncStatus.MasterBinlog = status.SyncStatus.MasterBinlog jobSource.SyncStatus.MasterBinlogGtid = status.SyncStatus.MasterBinlogGtid jobSource.SyncStatus.RecentTps = status.SyncStatus.RecentTps jobSource.SyncStatus.SecondsBehindMaster = status.SyncStatus.SecondsBehindMaster jobSource.SyncStatus.Synced = status.SyncStatus.Synced jobSource.SyncStatus.SyncerBinlog = status.SyncStatus.SyncerBinlog jobSource.SyncStatus.SyncerBinlogGtid = status.SyncStatus.SyncerBinlogGtid jobSource.SyncStatus.TotalEvents = status.SyncStatus.TotalEvents jobSource.SyncStatus.TotalTps = status.SyncStatus.TotalTps } if err := jobSource.Update(tx, entity.DMJobSourceDBSchema.SyncStatus); err != nil { logger.Error("[saveSyncStatus]failed to update job source sync status", zap.String("dataSourceName", dataSourceName), zap.Any("syncStatus", status.SyncStatus), zap.Error(err)) return err } } return nil }); err != nil { logger.Error("[saveSyncStatus]failed to update dm job source status", zap.Error(err)) return err } return nil }