package dm import ( "context" "encoding/base64" "encoding/json" "fmt" "math" "regexp" "strconv" "strings" "time" "github.com/BurntSushi/toml" "github.com/go-mysql-org/go-mysql/mysql" "github.com/pingcap/errcode" "github.com/pingcap/errors" lightningcfg "github.com/pingcap/tidb/pkg/lightning/config" "github.com/pingcap/tidb/pkg/util/filter" tfilter "github.com/pingcap/tidb/pkg/util/table-filter" router "github.com/pingcap/tidb/pkg/util/table-router" dmconfig "github.com/pingcap/tiflow/dm/config" dmdbconfig "github.com/pingcap/tiflow/dm/config/dbconfig" "github.com/pingcap/tiflow/dm/config/security" "github.com/pingcap/tiflow/dm/openapi" "github.com/pingcap/tiflow/dm/pkg/binlog" "github.com/pingcap/tiflow/dm/pkg/checker" engineerrors "github.com/pingcap/tiflow/pkg/errors" "github.com/tidbcloud/core-svc/idl/pbgen/proto/core" "github.com/tidbcloud/dataflow-service/app/models/config_manager" agentpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/agent" commonpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/common" dmpb "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dm" cluster_job "github.com/tidbcloud/dataflow-service/internal/cluster_job" "github.com/tidbcloud/dataflow-service/internal/common" commondm "github.com/tidbcloud/dataflow-service/internal/common/dm" "github.com/tidbcloud/dataflow-service/internal/config" "github.com/tidbcloud/dataflow-service/internal/db" "github.com/tidbcloud/dataflow-service/internal/db/entity" "github.com/tidbcloud/dataflow-service/internal/grpc/grpcli" "github.com/tidbcloud/dataflow-service/internal/handler/agent" "github.com/tidbcloud/dataflow-service/internal/metrics" "github.com/tidbcloud/dataflow-service/internal/model/base" "github.com/tidbcloud/dataflow-service/internal/model/connprofile" pimport "github.com/tidbcloud/dataflow-service/internal/model/import" "github.com/tidbcloud/dataflow-service/internal/utils" "github.com/tidbcloud/dataflow-service/pkg/account/accountdb" infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api" "github.com/tidbcloud/dataflow-service/pkg/log" "github.com/tidbcloud/eks-provider/api/v1alpha1" cloudv1 "github.com/tidbcloud/infra-api/api/v1" pkglog "github.com/tidbcloud/pkgv2/log" "github.com/tidbcloud/pkgv2/logid" "github.com/tidbcloud/pkgv2/secret/encrypt" "go.uber.org/zap" "gorm.io/gorm" "gorm.io/gorm/clause" ) type Model struct { base.Model resourceAPI ResourceAPI encryptor encrypt.Encryptor importModel *pimport.Model configCenter config_manager.ConfigCenterManager } var ( // todo: maybe maintain those doc urls in some configuration file tidbCloudBaseUrl = "https://docs.pingcap.com/tidbcloud/" dmDocUrl = tidbCloudBaseUrl + "migrate-from-mysql-using-data-migration" dmTroubleshootingDocUrl = tidbCloudBaseUrl + "tidb-cloud-dm-precheck-and-troubleshooting" precheckItemHelpUrls = map[string]string{ dmconfig.DumpPrivilegeChecking: dmDocUrl + "#grant-required-privileges-to-the-upstream-database", dmconfig.ReplicationPrivilegeChecking: dmDocUrl + "#grant-required-privileges-to-the-upstream-database", dmconfig.VersionChecking: dmDocUrl + "#make-sure-your-data-source-and-version-are-supported", dmconfig.ServerIDChecking: dmTroubleshootingDocUrl + "#error-message-check-whether-mysql-server_id-has-been-greater-than-0", dmconfig.BinlogEnableChecking: dmTroubleshootingDocUrl + "#error-message-check-whether-mysql-binlog-is-enabled", dmconfig.BinlogFormatChecking: dmTroubleshootingDocUrl + "#error-message-check-whether-mysql-binlog_format-is-row", dmconfig.BinlogRowImageChecking: dmTroubleshootingDocUrl + "#error-message-check-whether-mysql-binlog_row_image-is-full", dmconfig.TableSchemaChecking: dmDocUrl, // don't have doc in tidbcloud about this, give it a default url dmconfig.BinlogDBChecking: dmTroubleshootingDocUrl + "#error-message-check-whether-migrated-dbs-are-in-binlog_do_dbbinlog_ignore_db", dmconfig.ConnNumberChecking: dmTroubleshootingDocUrl + "#error-message-check-if-connection-concurrency-exceeds-databases-maximum-connection-limit", dmconfig.TargetDBPrivilegeChecking: dmDocUrl + "#grant-required-privileges-to-the-downstream-tidb-cloud-cluster", dmconfig.MetaPositionChecking: dmDocUrl, // TODO: add this later dmconfig.LightningTableEmptyChecking: dmDocUrl, // TODO: add this later } precheckItemTypes = map[string]dmpb.PrecheckItemType{ dmconfig.DumpPrivilegeChecking: dmpb.PrecheckItemType_DumpPrivilegeChecking, dmconfig.ReplicationPrivilegeChecking: dmpb.PrecheckItemType_ReplicationPrivilegeChecking, dmconfig.VersionChecking: dmpb.PrecheckItemType_VersionChecking, dmconfig.ServerIDChecking: dmpb.PrecheckItemType_ServerIDChecking, dmconfig.BinlogEnableChecking: dmpb.PrecheckItemType_BinlogEnableChecking, dmconfig.BinlogFormatChecking: dmpb.PrecheckItemType_BinlogFormatChecking, dmconfig.BinlogRowImageChecking: dmpb.PrecheckItemType_BinlogRowImageChecking, dmconfig.TableSchemaChecking: dmpb.PrecheckItemType_TableSchemaChecking, dmconfig.BinlogDBChecking: dmpb.PrecheckItemType_BinlogDBChecking, dmconfig.ConnNumberChecking: dmpb.PrecheckItemType_ConnNumberChecking, dmconfig.TargetDBPrivilegeChecking: dmpb.PrecheckItemType_TargetDBPrivilegeChecking, dmconfig.LightningTableEmptyChecking: dmpb.PrecheckItemType_LightningTableEmptyChecking, } // need to modify these check list if dm add new check allModeCheckList = []string{ dmconfig.DumpPrivilegeChecking, dmconfig.ReplicationPrivilegeChecking, dmconfig.VersionChecking, dmconfig.ServerIDChecking, dmconfig.BinlogEnableChecking, dmconfig.BinlogFormatChecking, dmconfig.BinlogRowImageChecking, dmconfig.TableSchemaChecking, dmconfig.BinlogDBChecking, dmconfig.ConnNumberChecking, dmconfig.TargetDBPrivilegeChecking, dmconfig.LightningTableEmptyChecking, } fullModeCheckList = []string{ dmconfig.DumpPrivilegeChecking, dmconfig.VersionChecking, dmconfig.TableSchemaChecking, dmconfig.ConnNumberChecking, dmconfig.TargetDBPrivilegeChecking, } incrModeCheckList = []string{ dmconfig.ReplicationPrivilegeChecking, dmconfig.VersionChecking, dmconfig.ServerIDChecking, dmconfig.BinlogEnableChecking, dmconfig.BinlogFormatChecking, dmconfig.BinlogRowImageChecking, dmconfig.TableSchemaChecking, dmconfig.BinlogDBChecking, dmconfig.TargetDBPrivilegeChecking, dmconfig.MetaPositionChecking, } physicalImportStoreBWLimit = map[int]int64{ 2: 52428800, // 50 MiB/s 4: 62914560, // 60 MiB/s 8: 78643200, // 75 MiB/s 16: 0, // no limit } ) func GetPrecheckItemHelpUrl(item string) string { return precheckItemHelpUrls[item] } func GetPrecheckItemType(item string) dmpb.PrecheckItemType { return precheckItemTypes[item] } func GetDmDocUrl() string { return dmDocUrl } func NewModel(log *zap.Logger, encryptor encrypt.Encryptor, importModel *pimport.Model, configCenter config_manager.ConfigCenterManager) Model { return Model{ Model: base.NewModel(log), encryptor: encryptor, importModel: importModel, configCenter: configCenter, } } // UserContext attaches a User to a Context. type UserContext struct { Model base.UserContext } func (c UserContext) GetPrecheckByID(id entity.ID) (*entity.DMPrecheck, error) { job := &entity.DMPrecheck{} err1 := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMPrecheckQuerySet(tx).IDEq(id).One(job) }) if err1 != nil { return nil, err1 } return job, nil } func (c UserContext) UpdatePrecheck(job *entity.DMPrecheck) error { return db.Transaction(func(tx *gorm.DB) error { return job.Update(tx, entity.DMPrecheckDBSchema.Status, entity.DMPrecheckDBSchema.ErrMsg, ) }) } func ignoreCheckingItems(items []string, ignoreItem string) []string { ignoredItems := make([]string, 0, len(items)) for _, item := range items { if item == ignoreItem { continue } ignoredItems = append(ignoredItems, item) } return ignoredItems } func getCheckItems(req *dmpb.CreateJobReq) []string { var items []string switch req.Mode { case dmconfig.ModeFull: items = fullModeCheckList case dmconfig.ModeIncrement: if len(req.Sources[0].GetBinlogGtid()) > 0 || len(req.Sources[0].GetBinlogName()) > 0 { items = incrModeCheckList } else { items = ignoreCheckingItems(incrModeCheckList, dmconfig.MetaPositionChecking) } case dmconfig.ModeAll: items = ignoreCheckingItems(allModeCheckList, dmconfig.LightningTableEmptyChecking) // TODO There is a bug in the empty table check item. It is temporarily disabled and will be reopened after being fixed later if entity.DMJobMode(req.Mode) == entity.DMJobModeAll && entity.DMImportMode(req.ImportMode) != entity.DMImportModeLogical { if config.Global().DMConfig.EnableLightningPrecheck { items = allModeCheckList } } default: // should not reach here return []string{} } return config.Global().DMConfig.PrecheckSkipConfigs.ApplyBlacklist(items, req.OrgId, req.ProjectId, req.ClusterId) } func (c UserContext) CreatePrecheck(req *dmpb.CreateJobReq) (entity.ID, error) { newPrecheckJob := &entity.DMPrecheck{ Status: entity.PrecheckStatusRunning, TenantID: entity.ID(req.OrgId), ProjectID: entity.ID(req.ProjectId), ClusterID: entity.ID(req.ClusterId), TotalItem: len(getCheckItems(req)), FullInstanceMigrate: req.FullInstanceMigrate, } err := db.Transaction(func(tx *gorm.DB) error { return newPrecheckJob.Create(tx) }) if err != nil { return 0, err } return newPrecheckJob.ID, nil } // this method changes req.Source in place to prepare binlog related param for increment mode job func (c UserContext) prepareBinlogParam(ctx context.Context, connProfiles []entity.DFConnProfile, req *dmpb.CreateJobReq) error { cluster, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, req.ClusterId) if err != nil { c.Log().Info("failed to get agent client", zap.Error(err)) return errcode.NewInternalErr(errors.Errorf("Failed to create job.")) } connProfileMap := make(map[uint64]*entity.DFConnProfile) for i := range connProfiles { p := &connProfiles[i] connProfileMap[p.ID.Uint64()] = p } for _, source := range req.Sources { if len(source.GetBinlogName()) > 0 || len(source.GetBinlogGtid()) > 0 { continue } // must exist, already checked in previous step connProfile := connProfileMap[source.ConnProfileId] host, err2 := GetHostFromConnProfile(connProfile.ConnMethod, connProfile.Host, connProfile.EndpointServiceName, cluster) if err2 != nil { c.Log().Error("get private link info failed", zap.Error(err2)) return errcode.NewInternalErr(errors.Errorf("Failed to create job.")) } decryptedPassword, err2 := c.decryptBase64(connProfile.EncryptedPassword) if err2 != nil { c.Log().Error("failed to decrypt cluster password", zap.Error(err2)) return errcode.NewInternalErr(errors.New("Failed to create job.")) } statusResp, err2 := agentClient.GetSourceBinlogStatus(ctx, &agentpb.CheckDBConnectivityReq{ Host: host, Port: uint32(connProfile.Port), User: connProfile.User, Password: decryptedPassword, SslCa: connProfile.SslCa, SslCert: connProfile.SslCert, SslKey: connProfile.SslKey, RequestId: logid.GetLogid(ctx), }) if err2 != nil { c.Log().Error("failed to get source binlog status", zap.Error(err2)) return errcode.NewInternalErr(errors.Errorf("Failed to create job.")) } c.Log().Info("got source binlog status", zap.Reflect("status", statusResp)) source.BinlogName = &statusResp.BinlogName source.BinlogPos = &statusResp.BinlogPos source.BinlogGtid = &statusResp.GtidStr } return nil } func (c UserContext) adjustBinlogParam(req *dmpb.CreateJobReq) { // we use mysql flavor here, it's actual the same for mariadb flavor zeroLoc := binlog.MustZeroLocation(mysql.MySQLFlavor) var ( defaultBinlogName = zeroLoc.Position.Name defaultBinlogPos = zeroLoc.Position.Pos defaultGtid = zeroLoc.GTIDSetStr() ) for _, s := range req.Sources { if len(s.GetBinlogName()) == 0 { s.BinlogName = &defaultBinlogName } if s.GetBinlogPos() == 0 { s.BinlogPos = &defaultBinlogPos } if len(s.GetBinlogGtid()) == 0 { s.BinlogGtid = &defaultGtid } } } func (c UserContext) Create(ctx context.Context, sourceConnProfiles []entity.DFConnProfile, req *dmpb.CreateJobReq, cluster *core.Cluster) (uint64, error) { encryptedPassword, err := c.encryptBase64(req.ClusterPassword) if err != nil { // don't want user know we failed to encrypt c.Log().Error("failed to encrypt cluster password", zap.Error(err)) return 0, errcode.NewInternalErr(errors.Errorf("failed to create job")) } if entity.DMJobMode(req.Mode) == entity.DMJobModeIncr { if err = c.prepareBinlogParam(ctx, sourceConnProfiles, req); err != nil { return 0, err } } // make sure all binlog_* param is not nil, so we can use them directly c.adjustBinlogParam(req) job := entity.DFJob{ TenantID: entity.ID(cluster.TenantId), ProjectID: entity.ID(cluster.ProjectId), ClusterID: entity.ID(cluster.Id), Name: req.Name, OneTime: entity.DMJobMode(req.Mode) == entity.DMJobModeFull, Phase: entity.DFJobPhaseCreating, ServiceCode: entity.ServiceCodeDM, CreatorUID: entity.ID(c.UserID()), Region: cluster.RegionName, Provider: cluster.InfraProviderType, WorkerVersion: entity.DMWorkerVersionV2, } dmJob := entity.DMJob{ TenantID: entity.ID(cluster.TenantId), ProjectID: entity.ID(cluster.ProjectId), ClusterID: entity.ID(req.ClusterId), ClusterUser: req.ClusterUser, EncryptedClusterPassword: encryptedPassword, Mode: entity.DMJobMode(req.Mode), ShardMode: entity.DMShardModeNone, // don't support online-ddl in first version OnlineDdl: false, CollationCompatible: entity.DMCollationModeLoose, SpecID: entity.ID(req.SpecId), PrecheckID: entity.ID(req.PrecheckId), ImportMode: entity.DMImportMode(req.ImportMode), } jobSources := []entity.DMJobSource{} for _, source := range req.Sources { jobSources = append(jobSources, dmpbSource2DMSource(req.Mode, source)) } err = db.Transaction(func(tx *gorm.DB) error { if err2 := job.Create(tx); err2 != nil { return errors.Trace(err2) } dmJob.ID = job.ID if err2 := dmJob.Create(tx); err2 != nil { return errors.Trace(err2) } for _, jobSource := range jobSources { jobSource.JobID = job.ID if err2 := jobSource.Create(tx); err2 != nil { return errors.Trace(err2) } } if err2 := cluster_job.RegisterDMPauseLock(ctx, job.ClusterID.Uint64(), job.ID.Uint64()); err2 != nil { c.Log().Error("register dm pause lock failed", zap.Error(err2), zap.Uint64("clusterID", job.ClusterID.Uint64()), zap.Uint64("jobID", job.ID.Uint64()), ) return errors.Annotate(err2, "register dm pause lock failed") } return nil }) if err != nil { // RegisterDMPauseLock is not part of the DB transaction. If transaction // commit fails after lock registration, compensate to avoid orphan pause // locks without corresponding DFJob rows. if job.ID > 0 { if unlockErr := cluster_job.DeregisterDMPauseLock(ctx, job.ClusterID.Uint64(), job.ID.Uint64()); unlockErr != nil { c.Log().Error("deregister dm pause lock failed after create transaction error", zap.Error(unlockErr), zap.Uint64("clusterID", job.ClusterID.Uint64()), zap.Uint64("jobID", job.ID.Uint64()), ) } } return 0, err } PublishEvent(ctx, &PublishEventReq{Job: &Job{DFJob: &job, DMJob: &dmJob}}) return uint64(job.ID), err } func (c UserContext) PrecheckWorker(ctx context.Context, req *dmpb.CreateJobReq, precheckID entity.ID, reqId string) { preCheckTimeout := config.Global().DMConfig.PreCheckTimeout logger := c.Log().With(zap.Uint64("precheck ID", uint64(precheckID)), zap.Duration("timeout", preCheckTimeout)) var ( isFailed = false errMsg = "" agentClient agent.AgentClient host, sourceHost string serviceName string connMethod entity.ConnMethod port int cluster *core.Cluster ) precheckJob, err := c.GetPrecheckByID(precheckID) if err != nil { logger.Error("fail to get precheck job", zap.Error(err)) } if precheckJob.Status == entity.PrecheckStatusCancel { return } defer func() { // update status of precheck job precheckJob, err1 := c.GetPrecheckByID(precheckID) if err1 != nil { logger.Error("fail to get precheck job", zap.Error(err1)) } if precheckJob.Status == entity.PrecheckStatusCancel { logger.Info("precheck job has been canceled, stop updating", zap.Uint64("precheck ID", uint64(precheckID))) return } if isFailed { precheckJob.Status = entity.PrecheckStatusFailed precheckJob.ErrMsg = errMsg } else { precheckJob.Status = entity.PrecheckStatusFinished } if err := c.UpdatePrecheck(precheckJob); err != nil { logger.Error("fail to update precheck job status", zap.Error(err)) } logger.Info("all precheck jobs finished", zap.Uint64("precheck ID", uint64(precheckID))) }() // for local test // agentClient = &agent.AgentClientImpl{ // BaseURL: "127.0.0.1:8080", // Doer: &http.Client{ // Timeout: time.Duration(40) * time.Second, // }, // ProviderType: "eks", // } // sourceHost = req.Sources[0].ConnProfile.Host // host = "10.2.5.9" // port = 4000 cluster, _, agentClient, err = common.GetClusterNetworkAgentById(ctx, req.ClusterId) if err != nil { logger.Error("failed to get agent", zap.Error(err)) errMsg = "failed to get agent: " + err.Error() isFailed = true } agentClient.SetHttpTimeout(preCheckTimeout) // assemble subtask config host, port = c.Model.GetClusterHostPort(cluster) // either the service name or the source host is set connMethod = entity.ConnMethod(req.Sources[0].ConnProfile.ConnMethod) if req.Sources[0].ConnProfile.EndpointServiceName != nil { serviceName = *req.Sources[0].ConnProfile.EndpointServiceName sourceHost, err = GetHostFromConnProfile(connMethod, "", serviceName, cluster) if err != nil { msg := "fail to get source host" logger.Error(msg, zap.Error(err)) isFailed = true errMsg = "failed to source host: " + err.Error() return } } if req.Sources[0].ConnProfile.Host != nil { sourceHost = *req.Sources[0].ConnProfile.Host } subTaskCfg := &dmconfig.SubTaskConfig{ Name: "precheck-task", Mode: req.Mode, From: dmdbconfig.DBConfig{ Host: sourceHost, Port: int(req.Sources[0].ConnProfile.Port), User: req.Sources[0].ConnProfile.User, }, To: dmdbconfig.DBConfig{ Host: host, Port: port, User: req.ClusterUser, }, SourceID: "test-precheck", LoaderConfig: dmconfig.DefaultLoaderConfig(), SyncerConfig: dmconfig.DefaultSyncerConfig(), MydumperConfig: dmconfig.DefaultMydumperConfig(), } subTaskCfg.LoaderConfig.PDAddr = fmt.Sprintf("db-pd.tidb%d.svc:2379", req.ClusterId) tlsConfig := config.Global().TlsConfig if tlsConfig.IsEnableTLS(cluster.TenantId, cluster.ProjectId) { tcNameGetter := infraapi.ClusterNameGetter{ InfraProvider: cluster.InfraProviderType, InfraAPIObjNamespace: infraapi.GetObjectNamespace(cluster.ProjectId), ClusterName: infraapi.ClusterName(cluster.Id), } tidbServerCertSecret, err := agentClient.GetTiDBServerSecret(ctx, &agentpb.GetTiDBServerSecretReq{ ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(), ClusterNamespace: tcNameGetter.ClusterNamespace(), ProviderType: cluster.InfraProviderType, }) if err != nil { logger.Error("failed to get cluster server secret", zap.Error(err)) errMsg = "failed to get cluster server secret: " + err.Error() isFailed = true return } subTaskCfg.To.Security = &security.Security{ SSLCABase64: base64.StdEncoding.EncodeToString([]byte(tidbServerCertSecret.SslCa)), } clientCertSecret, err := agentClient.GetClusterClientSecret(ctx, &agentpb.GetClusterClientSecretReq{ ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(), ClusterNamespace: tcNameGetter.ClusterNamespace(), ProviderType: cluster.InfraProviderType, }) if err != nil { logger.Error("failed to get cluster client secret", zap.Error(err)) errMsg = "failed to get cluster client secret: " + err.Error() isFailed = true return } subTaskCfg.LoaderConfig.Security = &security.Security{ SSLCABytes: []byte(clientCertSecret.SslCa), SSLCertBytes: []byte(clientCertSecret.SslCert), SSLKeyBytes: []byte(clientCertSecret.SslKey), } } if req.Sources[0].ConnProfile.SslCa != "" || req.Sources[0].ConnProfile.SslCert != "" || req.Sources[0].ConnProfile.SslKey != "" { subTaskCfg.From.Security = &security.Security{ SSLCABase64: base64.StdEncoding.EncodeToString([]byte(req.Sources[0].ConnProfile.SslCa)), SSLCertBase64: base64.StdEncoding.EncodeToString([]byte(req.Sources[0].ConnProfile.SslCert)), SSLKeyBase64: base64.StdEncoding.EncodeToString([]byte(req.Sources[0].ConnProfile.SslKey)), } } routeRules := RouteRules2Entity(req.Sources[0].RouteRules) for i := range routeRules { subTaskCfg.RouteRules = append(subTaskCfg.RouteRules, &routeRules[i]) } filterRule := BaRules2Entity(req.Sources[0].BaRules) subTaskCfg.BAList = &tfilter.MySQLReplicationRules{ DoDBs: filterRule.DoDBs, DoTables: filterRule.DoTables, IgnoreDBs: filterRule.IgnoreDBs, IgnoreTables: filterRule.IgnoreTables, } if len(req.Sources[0].GetBinlogGtid()) > 0 || len(req.Sources[0].GetBinlogName()) > 0 { subTaskCfg.Meta = &dmconfig.Meta{} if len(req.Sources[0].GetBinlogName()) > 0 { subTaskCfg.Meta.BinLogName = req.Sources[0].GetBinlogName() } if req.Sources[0].GetBinlogPos() > 0 { subTaskCfg.Meta.BinLogPos = req.Sources[0].GetBinlogPos() } if len(req.Sources[0].GetBinlogGtid()) > 0 { subTaskCfg.EnableGTID = true subTaskCfg.Meta.BinLogGTID = req.Sources[0].GetBinlogGtid() } } // only check for all mode now if entity.DMJobMode(req.Mode) == entity.DMJobModeAll && entity.DMImportMode(req.ImportMode) != entity.DMImportModeLogical { subTaskCfg.LoaderConfig.ImportMode = dmconfig.LoadModePhysical } checkItems := getCheckItems(req) var adminPassword string if utils.InArr(dmconfig.LightningTableEmptyChecking, checkItems) { adminPassword, err = c.GetCloudAdminPassword(ctx, logger, req.ClusterId) if err != nil { logger.Error("failed to get cloud admin password", zap.Error(err)) errMsg = "failed to get cloud admin password: " + err.Error() isFailed = true return } } for _, checkItem := range checkItems { // TODO: make it concurrent if the process is too slow // TODO: support multiple sources // the cluster password as well as the password is placed // in the request body, in order to be redacted. baseAgentReq := &agentpb.PrecheckDMJobReq{ ReqId: reqId, FullInstanceMigrate: req.FullInstanceMigrate, ClusterPassword: req.ClusterPassword, Password: req.Sources[0].ConnProfile.Password, } // dmconfig.LightningTableEmptyChecking need use cloud_admin account subTaskCfg := subTaskCfg if checkItem == dmconfig.LightningTableEmptyChecking { baseAgentReq.ClusterPassword = adminPassword newSubTaskCfg := *subTaskCfg newSubTaskCfg.To.User = "cloud_admin" subTaskCfg = &newSubTaskCfg } if err := c.doPrecheckDMJob(ctx, agentClient, precheckID, logger, subTaskCfg, checkItem, baseAgentReq); err != nil { isFailed = true } } } // doPrecheckDMJob only returns internal errors func (c UserContext) doPrecheckDMJob(ctx context.Context, agentClient agent.AgentClient, precheckID entity.ID, logger *zap.Logger, subTaskCfg *dmconfig.SubTaskConfig, checkItem string, agentReq *agentpb.PrecheckDMJobReq) error { var ( err error precheckItem = entity.DMPrecheckItem{ Desc: checkItem, Status: entity.PrecheckStatusRunning, PrecheckID: precheckID, } internalErrMsg = "internal error" skipUpdate = false ) logger = logger.With(zap.String("check item", checkItem)) markFailed := func() { precheckItem.ErrMsg = internalErrMsg precheckItem.Status = entity.PrecheckStatusFailed } defer func() { if rerr := recover(); rerr != nil { logger.Error("doPrecheckDMJob panic", zap.Any("err", rerr)) markFailed() if err == nil { err = errors.Errorf("doPrecheckDMJob panic:%v", rerr) } } if skipUpdate { // the job is possibly canceled, // no need to update the item return } if err1 := updateDMPrecheckItem(&precheckItem); err1 != nil { logger.Error("fail to update precheck item", zap.Reflect("item", precheckItem), zap.Error(err1)) } }() if err = createDMPrecheckItem(&precheckItem); err != nil { logger.Error("fail to create precheck item") skipUpdate = true return errors.Annotate(err, "fail to create precheck item") } subTaskCfg.IgnoreCheckingItems = filterAllExcept(checkItem) // precheck var strBuilder strings.Builder if err = toml.NewEncoder(&strBuilder).Encode(subTaskCfg); err != nil { msg := "failed to encode subtask config" logger.Error(msg, zap.Error(err)) markFailed() return errors.Annotate(err, msg) } agentReq.TaskConfig = []string{strBuilder.String()} agentResp, err := agentClient.PrecheckDMJob(ctx, agentReq) if err != nil || agentResp == nil { msg := "failed to precheck dm job" logger.Error(msg, zap.Error(err)) markFailed() return errors.Annotate(err, msg) } var newResult entity.DMPrecheckResult if err1 := json.Unmarshal(agentResp.CheckResults, &newResult); err1 != nil { msg := "failed to unmarshal check result" logger.Error(msg, zap.Error(err1)) markFailed() return errors.Annotate(err1, msg) } if err := overwritePrecheckResult(checkItem, &newResult); err != nil { msg := "fail to overwrite check result" logger.Error(msg, zap.Error(err)) markFailed() return errors.Annotate(err, msg) } precheckItem.Results = newResult precheckItem.Status = entity.PrecheckStatusFinished return nil } func (c UserContext) OperateDMJob(ctx context.Context, job *entity.DFJob, req *dmpb.OperateJobReq) error { logger := c.Log().With(zap.Uint64("dmJobID", job.ID.Uint64())).With(pkglog.FieldsFromContext(ctx)...) dmJob, err := GetDMJobByID(job.ID) if err != nil { logger.Error("failed to get dm job", zap.Error(err)) return errcode.NewInternalErr(errors.New("failed to get dm job")) } jobSources, err := GetDMJobSources(job.ID) if err != nil || len(jobSources) == 0 { logger.Error("failed to get dm job source", zap.Error(err), zap.Int("source-cnt", len(jobSources))) return errcode.NewInternalErr(errors.New("failed to get dm job sources")) } dmOpenapiClient, err := GetDMOpenapiClient(ctx, job) if err != nil { logger.Error("failed to init dm openapi client", zap.Error(err)) return errcode.NewInternalErr(errors.New("failed to init dm openapi client")) } stage := c.getDMStage(dmJob, &jobSources[0]) taskName := generateDMTaskName(dmJob.ID, stage) sourceNames := GetDatasourceNames(jobSources) taskStatus, err := dmOpenapiClient.QueryTaskStatus(ctx, taskName, sourceNames) if err != nil { logger.Error("failed to get dm job status", zap.String("taskName", taskName), zap.Error(err)) return errcode.NewInternalErr(errors.New("failed to get dm job status")) } else if len(taskStatus.Data) != len(jobSources) { logger.Error("dm job subtask status not match with job sources", zap.Int("subtaskStatusLen", len(taskStatus.Data)), zap.Int("jobSourcesLen", len(jobSources))) return errcode.NewInternalErr(errors.New("dm job subtask status not match with job sources")) } subTaskStatus := taskStatus.Data[0] switch req.Op { case dmpb.OperateJobReq_Pause: if subTaskStatus.Stage != openapi.TaskStagePaused { if subTaskStatus.Stage != openapi.TaskStageRunning { msg := fmt.Sprintf("dm job current stage %s, cannot do %s", subTaskStatus.Stage, strings.ToLower(req.Op.String())) logger.Error(msg, zap.String("taskName", taskName), zap.String("subTaskStage", string(subTaskStatus.Stage))) return errcode.NewInvalidInputErr(errors.New(msg)) } err = dmOpenapiClient.StopTask(ctx, taskName, &openapi.StopTaskRequest{SourceNameList: &sourceNames}) if err == nil { PublishEvent(ctx, &PublishEventReq{Job: &Job{DFJob: job, DMJob: dmJob}, Action: DMActionTaskPause}) } } case dmpb.OperateJobReq_Resume: if subTaskStatus.Stage != openapi.TaskStageRunning { if subTaskStatus.Stage != openapi.TaskStageStopped && subTaskStatus.Stage != openapi.TaskStagePaused { msg := fmt.Sprintf("dm job current stage %s, cannot do %s", subTaskStatus.Stage, strings.ToLower(req.Op.String())) logger.Error(msg, zap.String("taskName", taskName), zap.String("subTaskStage", string(subTaskStatus.Stage)), zap.Stringp("errorMsg", subTaskStatus.ErrorMsg)) return errcode.NewInvalidInputErr(errors.New(msg)) } err = dmOpenapiClient.StartTask(ctx, taskName, &openapi.StartTaskRequest{SourceNameList: &sourceNames}) if err == nil { PublishEvent(ctx, &PublishEventReq{Job: &Job{DFJob: job, DMJob: dmJob}, Action: DMActionTaskResume}) } } } if err != nil { logger.Error("failed to operate dm job", zap.String("op", req.Op.String()), zap.Error(err)) } return err } // UserContext returns a UserContext func (m Model) UserContext(ctx context.Context, user *accountdb.User) UserContext { return UserContext{ Model: m, UserContext: m.BaseUserContext(ctx, user, "dm"), } } func (m Model) RunDMJobs(ctx context.Context) error { logger := m.Log().With( zap.String("svc", string(entity.ServiceCodeDM)), zap.String("action", "RunDMJobs"), zap.Any("runID", logid.GenLogid()), ) logger.Info("dm create-job worker running") phases := []entity.DFJobPhase{ entity.DFJobPhaseCreating, entity.DFJobPhaseCreated, entity.DFJobPhaseModifying, entity.DFJobPhaseDeleting, entity.DFJobPhaseFinishing, } dfJobs, err := mgetDMJobsByPhase(ctx, logger, phases...) if err != nil { logger.Error("failed query creating jobs", zap.Error(err)) return err } for _, job := range dfJobs { // one job fail should not block other jobs runDMJob(ctx, logger, &job, m) } return nil } // param phase can be ignored actually, but to make this api matches deletePhaseMetric, we add it. func (m Model) addPhaseMetric(job *entity.DFJob, phase entity.DFJobPhase, lastTime time.Time) { metrics.DMJobDurationInPhase.WithLabelValues( fmt.Sprintf("%d", job.TenantID), fmt.Sprintf("%d", job.ClusterID), fmt.Sprintf("%d", job.ID), string(phase), ).Set(time.Since(lastTime).Seconds()) } func (m Model) deletePhaseMetric(logger *zap.Logger, phase entity.DFJobPhase, job *entity.DFJob) { if !metrics.DMJobDurationInPhase.DeleteLabelValues( fmt.Sprintf("%d", job.TenantID), fmt.Sprintf("%d", job.ClusterID), fmt.Sprintf("%d", job.ID), string(phase), ) { logger.Warn("failed to delete metric for job") } } func (m *Model) encryptBase64(in string) (string, error) { return encrypt.EncryptBase64(m.encryptor, encrypt.Secret(in)) } func (m *Model) decryptBase64(encrypted string) (string, error) { secret, err := encrypt.DecryptBase64(m.encryptor, encrypted) if err != nil { return "", err } return secret.String(), nil } func (m *Model) isJobNotFoundErr(err error) bool { return strings.Contains(err.Error(), string(engineerrors.ErrJobNotFound.ID())) } func (m *Model) needDumpStage(dmJob *entity.DMJob) bool { return dmJob.Mode == entity.DMJobModeAll || dmJob.Mode == entity.DMJobModeFull } func (m *Model) needLoadStage(dmJob *entity.DMJob) bool { return dmJob.Mode == entity.DMJobModeAll || dmJob.Mode == entity.DMJobModeFull } func (m *Model) needSyncStage(dmJob *entity.DMJob) bool { return dmJob.Mode == entity.DMJobModeAll || dmJob.Mode == entity.DMJobModeIncr } // SeparateDumpAndLoad returns true when we need to separate export and import in two different jobs func SeparateDumpAndLoad(mode entity.DMJobMode) bool { return mode == entity.DMJobModeAll } // UseImportAPI returns true when we need to use import API to do local import func UseImportAPI(importMode entity.DMImportMode) bool { return importMode != entity.DMImportModeLogical } func createImportUser(ctx context.Context, job *entity.DFJob) (*accountdb.User, error) { u, _ := accountdb.GetUserByID(job.CreatorUID.Uint64()) if u == nil { return nil, errors.New("failed to get user") } user, flag, er := accountdb.GetUserInOrg(u.Email, job.TenantID.Uint64(), false) if !flag || er != nil { return nil, errors.New("unknown user") } return user, nil } // GenerateImportTask generates import tasks for import API func GenerateImportTask(job *entity.DFJob, rcu int, jobSource entity.DMJobSource) *pimport.ImportTask { dataSourceType := entity.ImportDataSourceS3 if job.Provider == cloudv1.ProviderTypeGKE { dataSourceType = entity.ImportDataSourceGcs } imp := &pimport.ImportTask{ Import: entity.Import{ ProjectID: job.ProjectID, TenantID: job.TenantID, Name: "dm_" + strconv.FormatUint(job.ID.Uint64(), 10) + "_" + strconv.FormatUint(jobSource.ConnProfileID.Uint64(), 10), ClusterID: job.ClusterID, SourceURL: generateLoaderDir(job, jobSource.ConnProfileID.Uint64()), BucketRegion: job.Region, DataSourceType: dataSourceType, DataFormat: entity.ImportDataFormatSql, CreatedByDM: true, OptionalArgs: &entity.OptionalArgs{ StoreWriteBWLimit: lightningcfg.ByteSize(physicalImportStoreBWLimit[rcu]), }, }, } return imp } // ModifyJob can be reused if we need scale-in-out. func (m Model) ModifyJob(ctx context.Context, logger *zap.Logger, userCtx *UserContext, dfJob *entity.DFJob, jobID entity.ID, toSpecId uint64) error { // 1. check job phase if dfJob.Phase != entity.DFJobPhaseCreated { return errcode.NewInvalidInputErr(errors.New("dm job is not running")) } // 2. check current specID dmJob, err := GetDMJobByID(jobID) if err != nil { return errcode.NewInternalErr(err) } if toSpecId == uint64(dmJob.SpecID) { return errcode.NewInvalidInputErr(errors.New("Target job spec is the same as current job spec.")) } // 3. check toSpecID is valid newSpec, err := GetJobSpecByID(entity.ID(toSpecId)) if err != nil { if errors.Cause(err) == gorm.ErrRecordNotFound { return errcode.NewInvalidInputErr(errors.New("Target job spec is not found.")) } return errcode.NewInternalErr(err) } // 4. check full mode job if dmJob.Mode == entity.DMJobModeFull || dmJob.Mode == entity.DMJobModeAll { if SeparateDumpAndLoad(dmJob.Mode) && UseImportAPI(dmJob.ImportMode) { jobSources, err := GetDMJobSources(dfJob.ID) if err != nil { return err } currentPhase := dmpb.SubTask_Sync if !jobSources[0].DumpStatus.IsFinished() { currentPhase = dmpb.SubTask_Dump } else if !jobSources[0].LoadStatus.IsFinished() { currentPhase = dmpb.SubTask_Load } if currentPhase != dmpb.SubTask_Sync { return errcode.NewInternalErr(errors.Errorf("Job cannot scale up/down during %s phase for physical mode.", currentPhase)) } } jobSources, err := GetDMJobSources(dfJob.ID) if err != nil { return errors.Annotatef(err, "failed to get dm job sources:%d", dfJob.ID.Uint64()) } if err := m.isScaleSupported(ctx, logger, dfJob, dmJob, jobSources); err != nil { return err } } err = db.Transaction(func(tx *gorm.DB) error { dfJob.Phase = entity.DFJobPhaseModifying modifyJob := &entity.DMJobModifications{ JobID: jobID, Param: entity.DMJobModifyParam{ FromSpecID: dmJob.SpecID, ToSpecID: newSpec.ID, }, Type: entity.DMJobModifyTypeScaleUpDown, OperatorID: entity.ID(userCtx.UserID()), } err = updateDFJobPhaseWithExLock(tx, dfJob, entity.DFJobPhaseCreated, dfJob.Phase) if err != nil { if strings.Contains(err.Error(), "another modification is in progress") { return errcode.NewInvalidInputErr(err) } return errcode.NewInternalErr(err) } return modifyJob.Create(tx) }) if err != nil { return err } PublishEvent(ctx, &PublishEventReq{ Job: &Job{DFJob: dfJob, DMJob: dmJob}, FromSpecID: dmJob.SpecID, ToSpecID: newSpec.ID, }) return nil } func (m Model) getDMStage(dmJob *entity.DMJob, jobSource *entity.DMJobSource) entity.DMStageType { return GetDMStage(dmJob, jobSource) } func (m Model) isScaleSupported(ctx context.Context, logger *zap.Logger, dfJob *entity.DFJob, dmJob *entity.DMJob, jobSources []entity.DMJobSource) error { logger = logger.With(zap.Uint64("jobID", dfJob.ID.Uint64())) dmOpenapiClient, err := GetDMOpenapiClient(ctx, dfJob) if err != nil { logger.Error("failed to init dm openapi client", zap.Error(err)) return errcode.NewInternalErr(err) } for _, jobSource := range jobSources { stage := m.getDMStage(dmJob, &jobSource) if stage != entity.DMStageDump { continue } taskName := generateDMTaskName(dmJob.ID, stage) sourceName := generateDMDataSourceName(&jobSource) taskStatus, err := dmOpenapiClient.QueryTaskStatus(ctx, taskName, openapi.SourceNameList{sourceName}) if err != nil { logger.Error("failed to query dm job status", zap.String("taskName", taskName), zap.String("sourceName", sourceName), zap.Error(err)) return errcode.NewInternalErr(errors.New("failed to query dm job status")) } for _, subTaskStatus := range taskStatus.Data { if subTaskStatus.Stage != openapi.TaskStageFinished { return errcode.NewInvalidInputErr(errors.New("Job cannot scale up/down during dump phase.")) } } } return nil } func (m *Model) RunCleanupJobsForDeletedCluster(ctx context.Context) error { logger := m.Log().With(zap.String("svc", string(entity.ServiceCodeDM)), zap.String("action", "cleanup-jobs-for-deleted-cluster")) logger.Info("dm cleanup-job worker running") clusterIDs, err := getClusterIDsOfNonDeletedDMJob() if err != nil { logger.Error("failed to get cluster ids of non-deleted jobs", zap.Error(err)) return err } if len(clusterIDs) == 0 { return nil } listResp, err := grpcli.ListClusters(ctx, &core.ListClustersReq{ Filter: &core.ClusterFilter{ ClusterIds: clusterIDs, Statuses: grpcli.DeletedOrClearedClusterStatus, }, BaseReq: nil, }) if err != nil { logger.Error("failed to list clusters", zap.Error(err)) return err } for _, cluster := range listResp.Clusters { logger.Info("this cluster is deleted", zap.Uint64("cluster-id", cluster.Id)) m.cleanupJobsByCluster(logger, cluster) } return nil } func (m Model) cleanupJobsByCluster(inputLogger *zap.Logger, cluster *core.Cluster) { logger := inputLogger.With(zap.Uint64("cluster-id", cluster.Id)) _, dfJobs, err := ListDFJobsOfDM(entity.ID(cluster.Id), utils.Pager{Page: 1, PageSize: math.MaxInt32, Offset: 0}) if err != nil { logger.Error("failed to list df jobs", zap.Error(err)) return } for _, dfJob := range dfJobs { // we only mark it as deleting phase, and let job-delete worker do the real work if err = UpdateDFJobPhase(&dfJob, entity.DFJobPhaseDeleting); err != nil { logger.Error("failed to delete job", zap.Uint64("job-id", dfJob.ID.Uint64()), zap.Error(err)) // we don't return here to delete as many jobs as possible } else { logger.Info("job auto cleaned up", zap.Uint64("job-id", dfJob.ID.Uint64())) } } } func GetUnfinishedJobModification(jobID entity.ID) ([]entity.DMJobModifications, error) { var jobMods []entity.DMJobModifications err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMJobModificationsQuerySet(tx).FinishedAtIsNull().JobIDEq(jobID).All(&jobMods) }) return jobMods, err } func deleteDMJob(job *entity.DFJob) error { // save metadb data now := time.Now() job.Phase = entity.DFJobPhaseDeleted job.DeletedAt = &now // if the job is finishing, and user delete the dm job, we need set the finished time if job.Phase == entity.DFJobPhaseFinishing && job.FinishedAt == nil { job.FinishedAt = &now } return db.Transaction(func(tx *gorm.DB) error { if err := job.Update(tx, entity.DFJobDBSchema.Phase, entity.DFJobDBSchema.DeletedAt); err != nil { return err } if err := entity.NewDMJobQuerySet(tx).IDEq(job.ID).GetUpdater().SetDeletedAt(&now).Update(); err != nil { return err } return entity.NewDMJobSourceQuerySet(tx).JobIDEq(job.ID).GetUpdater().SetDeletedAt(&now).Update() }) } func mgetDMJobsByPhase(_ context.Context, logger *zap.Logger, phase ...entity.DFJobPhase) ([]entity.DFJob, error) { ret := []entity.DFJob{} if err := db.Transaction(func(tx *gorm.DB) error { txDB := tx.Where(getWhereClause(entity.NewDFJobQuerySet(db.GetDB()).PhaseIn(phase...).WorkerVersionEq(entity.DMWorkerVersionV2))) sqlStr := txDB.Statement.ToSQL(func(tx *gorm.DB) *gorm.DB { return tx.Find(&entity.DFJob{}) }) logger.Info("[mgetDMJobsByPhase]sql for select dm job", zap.Any("phase", phase), zap.String("sql", sqlStr)) return entity.NewDFJobQuerySet(txDB).All(&ret) }); err != nil { return nil, errors.Trace(err) } return ret, nil } type DBGetter interface { GetDB() *gorm.DB } func getWhereClause(querySet DBGetter) clause.Expression { return querySet.GetDB().Statement.Clauses["WHERE"].Expression } func getClusterIDsOfNonDeletedDMJob() ([]uint64, error) { ret := []uint64{} if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDFJobQuerySet(tx). PhaseNotIn(entity.DFJobPhaseDeleting, entity.DFJobPhaseDeleted). GetDB().Distinct().Pluck("cluster_id", &ret).Error }); err != nil { return nil, errors.Trace(err) } return ret, nil } func RouteRules2Entity(rules []*dmpb.RouteRule) entity.DMRouteRules { if len(rules) == 0 { return nil } res := entity.DMRouteRules{} for _, rule := range rules { res = append(res, router.TableRule{ SchemaPattern: rule.SchemaPattern, TablePattern: rule.TablePattern, TargetSchema: rule.TargetSchema, TargetTable: rule.TargetTable, }) } return res } func BaRules2Entity(rules *dmpb.BlockAllowRules) *entity.FilterRules { if rules == nil { // if user didn't specify ba rules, it's full instance migration, we filter all system database. return &entity.FilterRules{ IgnoreDBs: commondm.DefaultIgnoreDBs(), } } baTables2Inner := func(tables []*commonpb.Table) []*filter.Table { res := []*filter.Table{} for _, tbl := range tables { res = append(res, &filter.Table{ Schema: tbl.Schema, Name: tbl.Table, }) } return res } return &entity.FilterRules{ DoDBs: rules.DoDbs, IgnoreDBs: rules.IgnoreDbs, DoTables: baTables2Inner(rules.DoTables), IgnoreTables: baTables2Inner(rules.IgnoreTables), } } func GetDMJobByID(id entity.ID) (*entity.DMJob, error) { var dfJob entity.DMJob if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMJobQuerySet(tx).IDEq(id).One(&dfJob) }); err != nil { return nil, errors.Trace(err) } return &dfJob, nil } func GetDMJobSources(jobID entity.ID) ([]entity.DMJobSource, error) { ret := []entity.DMJobSource{} if err := db.Transaction(func(tx *gorm.DB) error { // we want a stable order when generating dm job config return entity.NewDMJobSourceQuerySet(tx).JobIDEq(jobID).PreloadConnProfile().OrderAscByConnProfileID().All(&ret) }); err != nil { return nil, errors.Trace(err) } return ret, nil } func GetDFJobByID(dfJobID entity.ID) (*entity.DFJob, error) { var dfJob entity.DFJob if err := db.Transaction(func(tx *gorm.DB) error { // deleting and deleted jobs are not visible to user return entity.NewDFJobQuerySet(tx).IDEq(dfJobID).PhaseNotIn(entity.DFJobPhaseDeleting, entity.DFJobPhaseDeleted).One(&dfJob) }); err != nil { return nil, errors.Trace(err) } return &dfJob, nil } func GetActiveDFJobByTenant(id entity.ID) ([]entity.DFJob, error) { var dfJobs []entity.DFJob if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDFJobQuerySet(tx).PhaseNotIn(entity.DFJobPhaseDeleted, entity.DFJobPhaseDeleting).TenantIDEq(id).All(&dfJobs) }); err != nil { return nil, errors.Trace(err) } return dfJobs, nil } func GetJobsForMetering(workerVersion string, startTs, endTs *time.Time) ([]entity.DFJob, error) { var dfJobs []entity.DFJob // job finished at [startTs, endTs] will be metered at [startTs+1H, endTs+1H]. // and thus, if a job failed to be metered at [startTs, endTs] (for [startTs-1H,endTs-1H]) // this time slot will be skipped. // it's fine to skip sometimes because it's rare to happen and mostly we tend to meter more than // less. if err := db.Transaction(func(tx *gorm.DB) error { // deleting job will be metered after being successfully deleted return tx.Where("((phase=? AND created_at < ?) OR (phase=? AND finished_at >= ? AND finished_at < ?) OR (phase=? AND deleted_at >= ? AND deleted_at < ?)) AND worker_version=?", entity.DFJobPhaseCreated, endTs, entity.DFJobPhaseFinished, startTs, endTs, entity.DFJobPhaseDeleted, startTs, endTs, workerVersion). Find(&dfJobs).Error }); err != nil { return nil, errors.Trace(err) } return dfJobs, nil } // UpdateDFJobPhase update job phase to any phase except deleted. for deleted phase, see method deleteDMJob func UpdateDFJobPhase(job *entity.DFJob, newPhase entity.DFJobPhase) error { // there's 2 part which may change job phase(x means any phase except deleted): // 1. delete job api: from x -> deleting // 2. dm workers: from creating -> created or from finishing -> finished or from x -> deleting // if the job is in deleting or deleted phase, we don't update to avoid overriding. return db.Transaction(func(tx *gorm.DB) error { updater := entity.NewDFJobQuerySet(tx).IDEq(job.ID).PhaseNotIn(entity.DFJobPhaseDeleting, entity.DFJobPhaseDeleted). GetUpdater() if newPhase == entity.DFJobPhaseFinished { curTs := time.Now() updater = updater.SetFinishedAt(&curTs) } return updater.SetPhase(newPhase).Update() }) } func UpdateDMJobComponentVersionWithImage(job *entity.DMJob, image string) error { var version string if pos := strings.LastIndex(image, ":"); pos == -1 { version = image } else { version = image[pos+1:] } return UpdateDMJobComponentVersion(job, version) } // UpdateDMJobComponentVersion update dm job component version func UpdateDMJobComponentVersion(job *entity.DMJob, componentVersion string) error { if job.ComponentVersion == componentVersion { return nil } return db.Transaction(func(tx *gorm.DB) error { updater := entity.NewDMJobQuerySet(tx).IDEq(job.ID).GetUpdater() return updater.SetComponentVersion(componentVersion).Update() }) } // UpdateDFJobErrorMessage update df job error message func UpdateDFJobErrorMessage(job *entity.DFJob, errorMessage string) error { if job.ErrorMessage == errorMessage { return nil } return db.Transaction(func(tx *gorm.DB) error { updater := entity.NewDFJobQuerySet(tx).IDEq(job.ID).GetUpdater() return updater.SetErrorMessage(errorMessage).Update() }) } // updateDFJobPhaseWithExLock will check whether the phase changes // conflict with other transactions. // expected conflict: // 1. created -> scaling vs created -> finishing // 2. scaling -> created vs scaling -> deleting func updateDFJobPhaseWithExLock(tx *gorm.DB, job *entity.DFJob, oldPhase, newPhase entity.DFJobPhase) error { result := tx.Model(&job).Where("phase = ?", oldPhase).Update("phase", newPhase) if result.Error != nil { return result.Error } if result.RowsAffected == 0 { return errors.New("Job might have finished or another modification is in progress.") } return nil } func UpdateDFJobName(dfJob *entity.DFJob, name string) error { dfJob.Name = name return UpdateDFJob(dfJob, entity.DFJobDBSchema.Name) } func UpdateDFJob(dfJob *entity.DFJob, fileds ...entity.DFJobDBSchemaField) error { return db.Transaction(func(tx *gorm.DB) error { return dfJob.Update(tx, fileds...) }) } func (m *Model) GetClusterHostPort(cluster *core.Cluster) (string, int) { tcNameGetter := infraapi.ClusterNameGetter{ InfraProvider: cluster.InfraProviderType, InfraAPIObjNamespace: infraapi.GetObjectNamespace(cluster.ProjectId), ClusterName: infraapi.ClusterName(cluster.Id), } // user can self-define port of tidb, but port of db-tidb(k8s service) is always 4000 return tcNameGetter.TiDBNamespacedServiceName(), 4000 } func filterAllExcept(checkItem string) []string { items := make([]string, 0) for item := range dmconfig.AllCheckingItems { if item != checkItem && item != dmconfig.AllChecking { items = append(items, item) } } return items } func GetHostFromConnProfile(connMethod entity.ConnMethod, host, serviceName string, cluster *core.Cluster) (string, error) { if connMethod == entity.ConnMethodPrivateLink && cluster != nil { privateLinkEndpoint, err := connprofile.GetAvailablePrivateLink( cluster.GetDataPlaneInfo().GetK8SNamespace(), cluster.GetDataPlaneInfo().GetNetworkName(), serviceName) if err != nil { return "", errors.Annotate(err, "failed to get private link endpoint") } return privateLinkEndpoint.RegionalDNS, nil } return host, nil } func createDMPrecheckItem(item *entity.DMPrecheckItem) error { return db.Transaction(func(tx *gorm.DB) error { var precheckJob entity.DMPrecheck if err := entity.NewDMPrecheckQuerySet(tx).IDEq(item.PrecheckID).One(&precheckJob); err != nil { return errors.Annotate(err, "fail to get precheck job") } if precheckJob.Status == entity.PrecheckStatusCancel { // canceled job return errors.New("precheck job canceled") } return tx.Create(item).Error }) } func updateDMPrecheckItem(item *entity.DMPrecheckItem) error { return db.Transaction(func(tx *gorm.DB) error { return item.Update(tx, entity.DMPrecheckItemDBSchema.Desc, entity.DMPrecheckItemDBSchema.ErrMsg, entity.DMPrecheckItemDBSchema.Results, entity.DMPrecheckItemDBSchema.Status, ) }) } func ListDFJobsOfDM(clusterID entity.ID, pager utils.Pager) (int64, []entity.DFJob, error) { var dfJobs []entity.DFJob var count int64 if err := db.Transaction(func(tx *gorm.DB) error { var err2 error count, err2 = entity.NewDFJobQuerySet(tx). ClusterIDEq(clusterID). PhaseNotIn(entity.DFJobPhaseDeleted, entity.DFJobPhaseDeleting). Count() if err2 != nil { return err2 } return entity.NewDFJobQuerySet(tx). ClusterIDEq(clusterID). PhaseNotIn(entity.DFJobPhaseDeleted, entity.DFJobPhaseDeleting). OrderDescByCreatedAt(). Limit(int(pager.PageSize)). Offset(int(pager.Offset)). All(&dfJobs) }); err != nil { return 0, nil, errors.Trace(err) } return count, dfJobs, nil } func GetJobSpecByID(id entity.ID) (*entity.DMJobSpec, error) { var ret entity.DMJobSpec if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMJobSpecQuerySet(tx).IDEq(id).One(&ret) }); err != nil { return nil, errors.Trace(err) } return &ret, nil } func ListJobSpecs() ([]entity.DMJobSpec, error) { ret := []entity.DMJobSpec{} if err := db.Transaction(func(tx *gorm.DB) error { // order by max incr phase speed to return a stable ordered list. return entity.NewDMJobSpecQuerySet(tx).OrderAscByMaxIncrPhaseSpeed().All(&ret) }); err != nil { return nil, errors.Trace(err) } return ret, nil } func CancelPrecheck(id entity.ID) error { return db.Transaction(func(tx *gorm.DB) error { return entity.NewDMPrecheckQuerySet(tx).IDEq(id).GetUpdater().SetStatus(entity.PrecheckStatusCancel).Update() }) } func GetPrecheckByID(id entity.ID) (*entity.DMPrecheck, error) { var ret entity.DMPrecheck if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMPrecheckQuerySet(tx).IDEq(id).One(&ret) }); err != nil { return nil, errors.Trace(err) } return &ret, nil } func GetPrecheckItemsByJobID(id entity.ID) ([]entity.DMPrecheckItem, error) { var ret []entity.DMPrecheckItem err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMPrecheckItemQuerySet(tx).PrecheckIDEq(id).All(&ret) }) if err != nil { return nil, err } return ret, nil } // ListPrechecksOfCluster returns precheck records of the given cluster in // reverse-chronological order, plus the total count without paging. It is the // backing query for dm.Service.ListPrechecks. func ListPrechecksOfCluster(clusterID entity.ID, pager utils.Pager) (int64, []entity.DMPrecheck, error) { var ( prechecks []entity.DMPrecheck total int64 ) if err := db.Transaction(func(tx *gorm.DB) error { var err2 error total, err2 = entity.NewDMPrecheckQuerySet(tx).ClusterIDEq(clusterID).Count() if err2 != nil { return err2 } return entity.NewDMPrecheckQuerySet(tx). ClusterIDEq(clusterID). OrderDescByID(). Limit(int(pager.PageSize)). Offset(int(pager.Offset)). All(&prechecks) }); err != nil { return 0, nil, errors.Trace(err) } return total, prechecks, nil } // ListPrecheckItemsByPrecheckIDs loads all precheck items of the given parent // precheck IDs in a single query, grouped by precheck ID. func ListPrecheckItemsByPrecheckIDs(ids []entity.ID) (map[entity.ID][]entity.DMPrecheckItem, error) { out := make(map[entity.ID][]entity.DMPrecheckItem, len(ids)) if len(ids) == 0 { return out, nil } var items []entity.DMPrecheckItem if err := db.Transaction(func(tx *gorm.DB) error { return entity.NewDMPrecheckItemQuerySet(tx).PrecheckIDIn(ids...).All(&items) }); err != nil { return nil, errors.Trace(err) } for i := range items { out[items[i].PrecheckID] = append(out[items[i].PrecheckID], items[i]) } return out, nil } func GetJobRelatedImports(ctx context.Context, logger *zap.Logger, job *entity.DFJob) (map[string]*pimport.ImportTask, error) { var err error pageSize := int32(30) totalImports := int64(pageSize) // task ConnProfileID -> import tasks dmImports := make(map[string]*pimport.ImportTask) var importTasks []*pimport.ImportTask for i := int32(1); i*pageSize <= int32(totalImports); i++ { totalImports, importTasks, err = pimport.ListImport(ctx, job.ClusterID, utils.AdjustedPager(&i, &pageSize), nil, true) if err != nil { logger.Error("failed to list import jobs", zap.Error(err)) return nil, err } for _, importTask := range importTasks { if sourceID := ExtractSourceIDFromImportName(job.ID, importTask.Name); len(sourceID) > 0 { dmImports[sourceID] = importTask } } } return dmImports, nil } func overwritePrecheckResult(checkItem string, pcResult *entity.DMPrecheckResult) error { // currently, almost all precheck items in cloud have one result, but 'conn_num' may has two results if len(pcResult.Results) < 1 { return errors.Errorf("unexpected check result:%+v", pcResult.Results) } if checkItem == dmconfig.TargetDBPrivilegeChecking { if pcResult.Results[0] == nil { return errors.Errorf("unexpected check result:%+v", pcResult.Results) } overwriteTargetPrivilegeResult(pcResult.Results[0]) } if checkItem == dmconfig.LightningTableEmptyChecking { if pcResult.Results[0] == nil { return errors.Errorf("unexpected check result:%+v", pcResult.Results) } overwriteTableEmptyResult(pcResult.Results[0]) } return nil } func overwriteTargetPrivilegeResult(result *checker.Result) { // Currently, dm checker will return 'warning' if target DB doesn't have some privileges on '*.*' since it can't recognizes // some specified schema/table-level privileges. // For cloud dm, we will return error simply and handle the error manually. if result.State == checker.StateWarning { result.State = checker.StateFailure } } func overwriteTableEmptyResult(result *checker.Result) { // For cloud dm, table empty check can be skipped through ignore, so we set this failure only to warning. if result.State == checker.StateFailure { result.State = checker.StateWarning } } func dmpbSource2DMSource(mode string, source *dmpb.Source) entity.DMJobSource { return entity.DMJobSource{ ConnProfileID: entity.ID(source.ConnProfileId), // the meta pos might be purged since job is created in async way, but it's quite rare since binlog // should keep at least for 24h in most cases. in this case, we better let dm query it when it's running. // right now, dataflow get this info and pass to DM, since it's static info we want to save to meta-db. Gtid: mode == dmconfig.ModeIncrement && len(source.GetBinlogGtid()) > 0, CaseSensitive: false, BinlogName: source.GetBinlogName(), BinlogPos: source.GetBinlogPos(), BinlogGtid: source.GetBinlogGtid(), RouteRules: RouteRules2Entity(source.RouteRules), BaRules: BaRules2Entity(source.BaRules), } } func impTask2LoadStatus(importTask *pimport.ImportTask, jobSource entity.DMJobSource) *entity.DMLoadStatus { loadStatus := &entity.DMLoadStatus{} if importTask.TotalSize != nil { loadStatus.TotalBytes = int64(*importTask.TotalSize) } if importTask.ProcessedDataSize != nil { loadStatus.FinishedBytes = int64(*importTask.ProcessedDataSize) } progress := float64(importTask.CompletedPercent+importTask.PostImportCompletedPercent) / 2.0 if importTask.Status == entity.ImportStatusCompleted { progress = 100 } loadStatus.Progress = fmt.Sprintf("%.2f", progress) loadStatus.MetaBinlog = fmt.Sprintf("(%s, %d)", jobSource.BinlogName, jobSource.BinlogPos) loadStatus.MetaBinlogGTID = jobSource.BinlogGtid finishedAt := time.Now() if importTask.FinishedAt != nil { finishedAt = *importTask.FinishedAt } startedAt := finishedAt if importTask.StartedAt != nil { startedAt = *importTask.StartedAt } loadStatus.Duration = int64(finishedAt.Sub(startedAt).Seconds()) if loadStatus.Duration > 0 { loadStatus.Bps = int64(float64(loadStatus.FinishedBytes) / float64(loadStatus.Duration)) } return loadStatus } // ExtractSourceIDFromImportName extracts job source ID from import name func ExtractSourceIDFromImportName(jobID entity.ID, name string) string { namePrefix := "dm_" + strconv.FormatUint(jobID.Uint64(), 10) + "_" if strings.HasPrefix(name, namePrefix) { return strings.TrimPrefix(name, namePrefix) } return "" } // GenerateEngineJobID generates job ID for separate import mode jobs // For all mode jobs created on EKS provider, they will be executed in several parts // physical mode: // // dump engine job -> import created by ImportAPI -> incremental engine job // // logical mode: // // dump engine job -> load&sync engine job // // Their names obey the following rules: // - task in dump mode: ${job.ID}-dump // - task in incremental/load&sync mode: ${job.ID}-sync // For other jobs, there engine job ids are always ${job.ID} func GenerateEngineJobID(dfJob *entity.DFJob, dmJob *entity.DMJob, finishedDump bool) string { id := strconv.FormatUint(dfJob.ID.Uint64(), 10) switch { case dmJob.Mode != entity.DMJobModeAll: return id case !finishedDump: return id + "-dump" default: return id + "-sync" } } func GenDMDataSourceName(jobSource *entity.DMJobSource) string { return generateDMDataSourceName(jobSource) } func generateDMDataSourceName(jobSource *entity.DMJobSource) string { return fmt.Sprintf("dm-%d-source-%d", jobSource.JobID.Uint64(), jobSource.ConnProfileID.Uint64()) } func ParseDMDataSourceName(name string) (dmJobID, connProfileID uint64, err error) { names := strings.Split(name, "-") if len(names) != 4 || names[0] != "dm" || names[2] != "source" { err = errors.Errorf("invalid dm data source name:%s", name) return } dmJobID, err = strconv.ParseUint(names[1], 10, 64) if err != nil { err = errors.Annotatef(err, "invalid dm data source name:%s, failed to parse dm job id", name) return } connProfileID, err = strconv.ParseUint(names[3], 10, 64) if err != nil { err = errors.Annotatef(err, "invalid dm data source name:%s, failed to parse conn profile id", name) return } return } func generateDMTaskName(dmJobID entity.ID, stage entity.DMStageType) string { return fmt.Sprintf("dm-%d-task-%s", dmJobID.Uint64(), stage) } func GenDMTaskName(dmJobID entity.ID, stage entity.DMStageType) string { return generateDMTaskName(dmJobID, stage) } func ConvertToStage(step dmpb.SubTask_Step) entity.DMStageType { switch step { case dmpb.SubTask_Dump: return entity.DMStageDump case dmpb.SubTask_Load: return entity.DMStageLoad case dmpb.SubTask_Sync: return entity.DMStageSync } return "" } func (m *Model) createDMDataSourceIfNotExists( ctx context.Context, dmOpenapiClient *agent.DMOpenapiClient, jobSource *entity.DMJobSource, cluster *core.Cluster, ) (*openapi.Source, error) { logger := log.GetLogger(ctx) dataSourceName := generateDMDataSourceName(jobSource) dataSource, err := dmOpenapiClient.GetDataSource(ctx, dataSourceName) if err == nil { return dataSource, nil } else if err != agent.ErrorDMResourceNotFound { return nil, err } // decrypt user db password connProfile := jobSource.ConnProfile decryptedSrcPassword, err := m.decryptBase64(connProfile.EncryptedPassword) if err != nil { logger.Error("failed to decrypt conn profile password", zap.Uint64("conn-profile-id", uint64(connProfile.ID)), zap.Error(err)) return nil, err } // get user db host sourceHost, err := GetHostFromConnProfile(connProfile.ConnMethod, connProfile.Host, connProfile.EndpointServiceName, cluster) if err != nil { logger.Error("failed to get user db host", zap.Uint64("conn-profile-id", uint64(connProfile.ID)), zap.Error(err)) return nil, err } dataSource, err = dmOpenapiClient.CreateDataSource(ctx, &openapi.CreateSourceRequest{ Source: openapi.Source{ Enable: true, EnableGtid: jobSource.Gtid, Host: sourceHost, Port: int(connProfile.Port), User: connProfile.User, Password: &decryptedSrcPassword, Security: &openapi.Security{ SslCaContent: connProfile.SslCa, SslCertContent: connProfile.SslCert, SslKeyContent: connProfile.SslKey, }, SourceName: dataSourceName, }, }) if err != nil { logger.Error("failed to create dm data source", zap.Error(err)) return nil, err } return dataSource, nil } func (m *Model) getTiDBClientCA(ctx context.Context, agentClient agent.AgentClient, dfJob *entity.DFJob) (string, string, string, error) { tlsConfig := config.Global().TlsConfig if !tlsConfig.IsEnableTLS(uint64(dfJob.TenantID), uint64(dfJob.ProjectID)) { return "", "", "", nil } logger := log.GetLogger(ctx) tcNameGetter := infraapi.ClusterNameGetter{ InfraProvider: dfJob.Provider, InfraAPIObjNamespace: infraapi.GetObjectNamespace(dfJob.ProjectID.Uint64()), ClusterName: infraapi.ClusterName(dfJob.ClusterID.Uint64()), } certSecret, err := agentClient.GetClusterClientSecret(ctx, &agentpb.GetClusterClientSecretReq{ ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(), ClusterNamespace: tcNameGetter.ClusterNamespace(), ProviderType: dfJob.Provider, RequestId: utils.UniqueID(), }) if err != nil { logger.Error("failed to get cluster client secret", zap.Error(err)) return "", "", "", errcode.NewInternalErr(errors.New("failed to get cluster")) } return certSecret.SslCa, certSecret.SslCert, certSecret.SslKey, nil } func (m *Model) getTiDBServerCA(ctx context.Context, agentClient agent.AgentClient, dfJob *entity.DFJob) (string, string, string, error) { tlsConfig := config.Global().TlsConfig if !tlsConfig.IsEnableTLS(uint64(dfJob.TenantID), uint64(dfJob.ProjectID)) { return "", "", "", nil } logger := log.GetLogger(ctx) tcNameGetter := infraapi.ClusterNameGetter{ InfraProvider: dfJob.Provider, InfraAPIObjNamespace: infraapi.GetObjectNamespace(dfJob.ProjectID.Uint64()), ClusterName: infraapi.ClusterName(dfJob.ClusterID.Uint64()), } tidbServerCertSecret, err := agentClient.GetTiDBServerSecret(ctx, &agentpb.GetTiDBServerSecretReq{ ClusterNamePrefix: tcNameGetter.ClusterServicePrefix(), ClusterNamespace: tcNameGetter.ClusterNamespace(), ProviderType: dfJob.Provider, }) if err != nil { logger.Error("failed to get cluster client secret", zap.Error(err)) return "", "", "", errcode.NewInternalErr(errors.New("failed to get cluster")) } return tidbServerCertSecret.SslCa, tidbServerCertSecret.SslCert, tidbServerCertSecret.SslKey, nil } func (m Model) scaleWorkerToSpec(ctx context.Context, dmOpenapiClient *agent.DMOpenapiClient, dmJob *entity.DMJob, dmJobCR *v1alpha1.DMJob, toSpec, dataSourceName string) (bool, error) { logger := log.GetLogger(ctx) switch dmJobCR.Status.Phase { case v1alpha1.DMJobPhaseRunning: if toSpec != dmJobCR.Spec.Worker.SpecCode { // has not committed modifications fromSpec := dmJobCR.Spec.Worker.SpecCode dmJobCR.Spec.Worker.SpecCode = toSpec if err := m.resourceAPI.UpdateDMJob(ctx, dmJobCR); err != nil { return false, err } logger.Info("update dm job cr worker spec code", zap.String("fromSpec", fromSpec), zap.String("topSpec", toSpec)) } else if dmJobCR.Spec.Worker.SpecCode == dmJobCR.Status.WorkerSpecCode { // 1. dmJobCR.Phase == running // 2. dmJobCR.Spec.Worker.SpecCode == dmJobCR.Status.WorkerSpecCode // (ensure the dmJobCR has been reconciled after // we change the dmJobCR.Spec.Worker.SpecCode) // 3. dmJobCR has job detail in dm master // (ensure the dmJobCR has already been scheduled on dm master) dataSource, err := dmOpenapiClient.GetDataSource(ctx, dataSourceName) if err != nil { return false, err } hasReady := dataSource.StatusList != nil && len(*dataSource.StatusList) > 0 if !hasReady { logger.Info("dm job data source is not ready when scaling, wait next times", zap.String("dataSourceName", dataSourceName)) } else { logger.Info("dm job data source is ready when scaling, scaling completed", zap.String("dataSourceName", dataSourceName)) } return hasReady, nil } case v1alpha1.DMJobPhaseScaling: logger.Info("the dm job is scaling.", zap.Any("dmJobID", dmJob.ID), zap.String("fromSpec", dmJobCR.Spec.Worker.SpecCode), zap.String("topSpec", dmJobCR.Status.WorkerSpecCode)) case v1alpha1.DMJobPhaseDeleting: // should not happen normally // The modification won't be applied and should be marked as finished. logger.Info("the dm job has been deleted in other routine", zap.Any("dmJobID", dmJob.ID)) } return false, nil } func (m Model) buildDMDataSource( ctx context.Context, dmOpenapiClient *agent.DMOpenapiClient, jobSources []entity.DMJobSource, cluster *core.Cluster, binlogFilterRuleName string, taskMode openapi.TaskTaskMode, dmJobMode entity.DMJobMode, ) ( taskSourceConf []openapi.TaskSourceConf, taskMigrateRules []openapi.TaskTableMigrateRule, err error, ) { logger := log.GetLogger(ctx) // create dm data source for each job source and prepare dump task parameters taskSourceConf = make([]openapi.TaskSourceConf, 0, len(jobSources)) taskMigrateRules = make([]openapi.TaskTableMigrateRule, 0, len(jobSources)) dataSources := make([]*openapi.Source, 0, len(jobSources)) for _, jobSource := range jobSources { var dataSource *openapi.Source dataSource, err = m.createDMDataSourceIfNotExists(ctx, dmOpenapiClient, &jobSource, cluster) if err != nil { logger.Error("failed to create dm data source", zap.Uint64("connProfileID", jobSource.ConnProfileID.Uint64()), zap.Error(err)) return } dataSources = append(dataSources, dataSource) binlogName := jobSource.BinlogName binlogPos := int(jobSource.BinlogPos) binlogGtid := jobSource.BinlogGtid // If the DM job is in all mode, the binlog position returned by the load phase // needs to be set in the request parameters of the sync phase. // Tips: sync and incremental are synonyms if dmJobMode == entity.DMJobModeAll && taskMode == openapi.TaskTaskModeIncremental && jobSource.LoadStatus.IsFinished() { binlogGtid = jobSource.LoadStatus.MetaBinlogGTID binlogName, binlogPos, err = parseLoadBinlog(jobSource.LoadStatus.MetaBinlog) if err != nil { logger.Error("failed to parse load task's meta binlog", zap.Error(err)) return } } taskSourceConf = append(taskSourceConf, openapi.TaskSourceConf{ SourceName: dataSource.SourceName, BinlogGtid: &binlogGtid, BinlogName: &binlogName, BinlogPos: &binlogPos, }) schemaExists := make(map[string]struct{}) for _, schema := range jobSource.BaRules.DoDBs { migrateRule := openapi.TaskTableMigrateRule{} migrateRule.Source.Schema = schema migrateRule.Source.SourceName = dataSource.SourceName migrateRule.Target = &openapi.TaskTableMigrateRuleTarget{ Schema: utils.ToPtr(schema), } taskMigrateRules = append(taskMigrateRules, migrateRule) schemaExists[schema] = struct{}{} } for _, table := range jobSource.BaRules.DoTables { migrateRule := openapi.TaskTableMigrateRule{} migrateRule.Source.SourceName = dataSource.SourceName migrateRule.Source.Schema = table.Schema migrateRule.Source.Table = table.Name if migrateRule.Source.Table == "~.*" { migrateRule.Source.Table = "*" } migrateRule.Target = &openapi.TaskTableMigrateRuleTarget{ Schema: utils.ToPtr(table.Schema), } if migrateRule.Source.Table != "*" { migrateRule.Target.Table = utils.ToPtr(migrateRule.Source.Table) } taskMigrateRules = append(taskMigrateRules, migrateRule) if _, ok := schemaExists[table.Schema]; !ok { migrateRule := openapi.TaskTableMigrateRule{} migrateRule.Source.Schema = table.Schema migrateRule.Source.SourceName = dataSource.SourceName migrateRule.Target = &openapi.TaskTableMigrateRuleTarget{ Schema: utils.ToPtr(table.Schema), } taskMigrateRules = append(taskMigrateRules, migrateRule) schemaExists[table.Schema] = struct{}{} } } // // At the db level rule, either all are added, or none are added. // if len(schemaExists) > 0 { // for _, table := range jobSource.BaRules.DoTables { // if _, ok := schemaExists[table.Schema]; ok { // continue // } // migrateRule := openapi.TaskTableMigrateRule{} // migrateRule.Source.Schema = table.Schema // migrateRule.Source.SourceName = dataSource.SourceName // migrateRule.Target = &struct { // Schema *string `json:"schema,omitempty"` // Table *string `json:"table,omitempty"` // }{ // Schema: utils.ToPtr(table.Schema), // } // taskMigrateRules = append(taskMigrateRules, migrateRule) // schemaExists[table.Schema] = struct{}{} // } // } // Background: The openAPI will combine the migrate rule and filter rule, // which is equivalent to ignoring the create database statement // related to the database before, and then it is equivalent to no filtering. // Now the above workaround is to filter the create database for all schemas // // The reason for adding the logic to filter the create database DDL is that DM // synchronizes unrelated databases downstream during the sync phase. // Currently, adding filtering rules is just a workaround, and ultimately DM // needs to fix this issue. At that time, this logic can be removed // refer to issue-001 for details: https://pingcap.feishu.cn/wiki/QPiQwPJVfiENc0kzVoRcJoqAnyg // // 2025-02-20 fangdong.jia When synchronizing all database tables, there is no need // to filter the create database DDL // if len(taskMigrateRules) > 0 && taskMode == openapi.TaskTaskModeIncremental && binlogFilterRuleName != "" { // filterMigrateRule := openapi.TaskTableMigrateRule{} // filterMigrateRule.Source.Schema = "*" // filterMigrateRule.Source.Table = "*" // filterMigrateRule.Source.SourceName = dataSource.SourceName // filterMigrateRule.BinlogFilterRule = &[]string{binlogFilterRuleName} // taskMigrateRules = append(taskMigrateRules, filterMigrateRule) // } } // wait data source bound with dm worker workListResp, err := dmOpenapiClient.ListWorks(ctx) if err != nil { logger.Error("failed to list dm works", zap.Error(err)) return } unboundSources := make([]string, 0, len(dataSources)) for _, dataSource := range dataSources { var hasBound bool for _, worker := range workListResp.Data { hasBound = hasBound || (dataSource.SourceName == worker.BoundSourceName && worker.BoundStage == "bound") } if !hasBound { unboundSources = append(unboundSources, dataSource.SourceName) } } if len(unboundSources) > 0 { logger.Error("some data sources have not yet bound workers, try again latter", zap.Strings("unboundSources", unboundSources)) err = errors.Errorf("some data sources have not yet bound workers, try again latter: %s", strings.Join(unboundSources, ",")) return } return } var binlogRegexp = regexp.MustCompile(`\(([^,]+),\s*(\d+)\)`) // parseLoadBinlog get binlog name and position from input string // example: // input: (mysql-bin.9999999, 1398) // output: binlogName=mysql-bin.9999999, binlogPos=1398 func parseLoadBinlog(binlog string) (binlogName string, binlogPos int, err error) { matches := binlogRegexp.FindStringSubmatch(binlog) if len(matches) < 3 { err = errors.Annotatef(err, "failed to parse load task binlog:%s", binlog) return } binlogName = matches[1] position, err := strconv.ParseInt(matches[2], 10, 64) if err != nil { err = errors.Annotatef(err, "invalid load task binlog:%s", binlog) } binlogPos = int(position) return } func (m Model) buildTargetDB( ctx context.Context, agentClient agent.AgentClient, dfJob *entity.DFJob, dmJob *entity.DMJob, taskMode openapi.TaskTaskMode, ) (targetDB *openapi.TaskTargetDataBase, err error) { logger := log.GetLogger(ctx) // prepare target db parameter tcNameGetter := infraapi.ClusterNameGetter{ InfraProvider: dfJob.Provider, InfraAPIObjNamespace: infraapi.GetObjectNamespace(dfJob.ProjectID.Uint64()), ClusterName: infraapi.ClusterName(dfJob.ClusterID.Uint64()), } // The cloud_admin account is required when loading data in physical mode. var dbUser, dbPassword string if taskMode == openapi.TaskTaskModeLoad && dmJob.ImportMode == entity.DMImportModePhysical { dbUser = "cloud_admin" dbPassword, err = m.GetCloudAdminPassword(ctx, logger, dmJob.ClusterID.Uint64()) if err != nil { logger.Error("failed to get cloud admin password", zap.Error(err)) return nil, err } } else { dbUser = dmJob.ClusterUser dbPassword, err = m.decryptBase64(dmJob.EncryptedClusterPassword) if err != nil { logger.Error("failed to decrypt user cluster password", zap.Error(err)) return nil, err } } sslCa, sslCert, sslKey, err := m.getTiDBServerCA(ctx, agentClient, dfJob) if err != nil { logger.Error("failed to get tidb server ca", zap.Error(err)) return nil, err } targetDB = &openapi.TaskTargetDataBase{ Host: tcNameGetter.TiDBNamespacedServiceName(), Port: 4000, User: dbUser, Password: dbPassword, Security: &openapi.Security{ SslCaContent: sslCa, SslCertContent: sslCert, SslKeyContent: sslKey, }, } return } func (m Model) recoverTask(ctx context.Context, dmOpenapiClient *agent.DMOpenapiClient, taskName string, sources openapi.SourceNameList) error { logger := log.GetLogger(ctx) if err := dmOpenapiClient.StartTask(ctx, taskName, &openapi.StartTaskRequest{SourceNameList: &sources}); err != nil { logger.Error("failed to recover(start) task", zap.String("taskName", taskName), zap.Strings("sources", sources), zap.Error(err)) return err } logger.Info("successfully recover(start) task", zap.String("taskName", taskName), zap.Strings("sources", sources)) return nil } func GetDMOpenapiClient(ctx context.Context, dfJob *entity.DFJob) (*agent.DMOpenapiClient, error) { _, _, agentClient, err := common.GetClusterNetworkAgentById(ctx, dfJob.ClusterID.Uint64()) if err != nil { return nil, err } resourceAPI, err := NewRegionalResourceAPI(ctx, dfJob.Provider, dfJob.Region, dfJob.TenantID.Uint64()) if err != nil { return nil, err } dmJobCR, err := resourceAPI.GetDMJob(ctx, dfJob.ProjectID.Uint64(), dfJob.ID.Uint64()) if err != nil { return nil, err } return agentClient.GetDMOpenapiClient(dfJob.ClusterID.Uint64(), dmJobCR.Status.MasterServerName), nil } func GetDatasourceNames(jobSources []entity.DMJobSource) openapi.SourceNameList { var result openapi.SourceNameList for _, jobSource := range jobSources { result = append(result, generateDMDataSourceName(&jobSource)) } return result } func (c *Model) GetCloudAdminPassword(ctx context.Context, logger *zap.Logger, clusterId uint64) (string, error) { var err error confidentialInfo, err := grpcli.GetClusterConfidentialInfoByID(ctx, clusterId) if err != nil { return "", errors.Annotate(err, "get cluster's confidential information by ID error") } encryptedPwd := confidentialInfo.GetEncryptedCloudAdminPasswd() if len(encryptedPwd) == 0 { logger.Error("the cloud admin password is empty") return "", errors.New("the cloud admin password is empty") } password, err := c.DecryptBase64(encryptedPwd) if err != nil { logger.Error("failed to decrypt cloud admin password", zap.Error(err)) return "", errors.Annotate(err, "decrypt cloud admin password error") } return password, nil } func (c *Model) DecryptBase64(password string) (string, error) { if len(password) == 0 { return "", nil } secret, err := encrypt.DecryptBase64(c.encryptor, password) return string(secret), err } func GetDMStage(dmJob *entity.DMJob, jobSource *entity.DMJobSource) entity.DMStageType { if dmJob.Mode == entity.DMJobModeIncr { return entity.DMStageSync } else if dmJob.Mode == entity.DMJobModeFull { if !jobSource.DumpStatus.IsFinished() { return entity.DMStageDump } else { return entity.DMStageLoad } } else { if !jobSource.DumpStatus.IsFinished() { return entity.DMStageDump } else if !jobSource.LoadStatus.IsFinished() { return entity.DMStageLoad } else { return entity.DMStageSync } } }