//nolint:lll package entity import ( "database/sql/driver" "encoding/base32" "encoding/json" "strings" "time" "github.com/google/uuid" dm "github.com/pingcap/tiflow/dm/openapi" pchecker "github.com/pingcap/tiflow/dm/pkg/checker" ) // --- Constants & Enums --- type PremiumDMTaskStage string const ( PremiumDMTaskStageSourceCreating PremiumDMTaskStage = "sourceCreating" PremiumDMTaskStageSourceBounding PremiumDMTaskStage = "sourceBounding" PremiumDMTaskStageTaskCreating PremiumDMTaskStage = "taskCreating" PremiumDMTaskStageRunning PremiumDMTaskStage = "running" PremiumDMTaskStagePausing PremiumDMTaskStage = "pausing" PremiumDMTaskStageStopped PremiumDMTaskStage = "stopped" PremiumDMTaskStageResuming PremiumDMTaskStage = "resuming" PremiumDMTaskStageDeleting PremiumDMTaskStage = "deleting" PremiumDMTaskStageDeleted PremiumDMTaskStage = "deleted" PremiumDMTaskStageFailed PremiumDMTaskStage = "failed" ) type PremiumDMTaskMode string const ( PremiumDMTaskModeFull PremiumDMTaskMode = "full" PremiumDMTaskModeIncremental PremiumDMTaskMode = "incremental" PremiumDMTaskModeAll PremiumDMTaskMode = "all" ) type PremiumDMTaskImportMode string const ( PremiumDMTaskImportModeLogical PremiumDMTaskImportMode = "logical" PremiumDMTaskImportModePhysical PremiumDMTaskImportMode = "physical" ) type PremiumDMSourceStage string const ( PremiumDMSourceStageCreating PremiumDMSourceStage = "creating" PremiumDMSourceStageBounding PremiumDMSourceStage = "bounding" PremiumDMSourceStageBounded PremiumDMSourceStage = "bounded" PremiumDMSourceStageDeleting PremiumDMSourceStage = "deleting" PremiumDMSourceStageDeleted PremiumDMSourceStage = "deleted" ) type PremiumDBType string const ( PremiumDBTypeMySQL PremiumDBType = "mysql" PremiumDBTypeAliCloudRDSMySQL PremiumDBType = "alicloud-rds-mysql" PremiumDBTypeAzureFlexibleMySQL PremiumDBType = "azure-flexible-mysql" ) type PremiumConnType string const ( PremiumConnTypePublic PremiumConnType = "PUBLIC" PremiumConnTypePrivateLink PremiumConnType = "PRIVATE_LINK" ) var ( ActivePremiumDMTaskStages = []PremiumDMTaskStage{ PremiumDMTaskStageSourceCreating, PremiumDMTaskStageSourceBounding, PremiumDMTaskStageTaskCreating, PremiumDMTaskStageRunning, PremiumDMTaskStagePausing, PremiumDMTaskStageStopped, PremiumDMTaskStageResuming, PremiumDMTaskStageFailed, } PollingPremiumDMTaskStages = []PremiumDMTaskStage{ PremiumDMTaskStageRunning, PremiumDMTaskStageStopped, PremiumDMTaskStageFailed, } NonDeletedPremiumDMTaskStages = append(ActivePremiumDMTaskStages, PremiumDMTaskStageDeleting) ) type PremiumDMTask struct { ID string `gorm:"column:id;type:varchar(255);not null;primaryKey"` Name string `gorm:"column:name;type:varchar(64);not null;uniqueIndex:idx_dm_task_cluster_name"` CreatedAt time.Time `gorm:"column:created_at;type:TIMESTAMP DEFAULT CURRENT_TIMESTAMP;not null"` // create and read only UpdatedAt *time.Time `gorm:"column:updated_at;type:TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP"` DeletedTime *time.Time `gorm:"column:deleted_time;type:TIMESTAMP NULL DEFAULT NULL"` FailedTime *time.Time `gorm:"column:failed_time;type:TIMESTAMP NULL DEFAULT NULL"` Stage PremiumDMTaskStage `gorm:"column:stage;type:enum('sourceCreating','sourceBounding','taskCreating','running','stopped','deleting','deleted','failed','pausing','resuming');index:idx_dm_task_stage"` OperationDispatchedAt *time.Time `gorm:"column:operation_dispatched_at;type:TIMESTAMP NULL DEFAULT NULL"` ErrMsg string `gorm:"column:err_msg;type:text"` CreatorID uint64 `gorm:"column:creator_id;type:bigint(20) unsigned;not null;default:0"` Creator string `gorm:"column:creator;type:VARCHAR(255)"` TiDBID string `gorm:"column:tidb_id;type:varchar(255);not null;index;uniqueIndex:idx_dm_task_cluster_name"` ProjectID string `gorm:"column:project_id;type:varchar(255);not null;default:'0'"` TenantID string `gorm:"column:tenant_id;type:varchar(255);not null;index"` CloudProvider string `gorm:"column:cloud_provider;type:VARCHAR(255);not null"` Region string `gorm:"column:region;type:VARCHAR(255);not null"` TargetUser string `gorm:"column:target_user;type:VARCHAR(255)"` // target user for DM task, used for DM worker to connect to target cluster TargetEncryptedPassword string `gorm:"column:target_encrypted_password;type:VARCHAR(255)"` // encrypted password for target user Mode PremiumDMTaskMode `gorm:"column:mode;type:enum('full','incremental','all');not null"` ImportMode PremiumDMTaskImportMode `gorm:"column:import_mode;type:enum('logical','physical');not null;default:'logical'"` ClusterStatusID uint64 `gorm:"column:cluster_status_id;type:bigint(20) NULL DEFAULT NULL"` DMSources []PremiumDMSource `gorm:"foreignKey:DMTaskID"` } type PremiumDMSource struct { ID string `gorm:"column:id;type:varchar(255);not null;primaryKey"` CreatedAt time.Time `gorm:"column:created_at;type:TIMESTAMP DEFAULT CURRENT_TIMESTAMP;not null"` // create and read only UpdatedAt *time.Time `gorm:"column:updated_at;type:TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP"` DeletedTime *time.Time `gorm:"column:deleted_time;type:TIMESTAMP NULL DEFAULT NULL"` Stage PremiumDMSourceStage `gorm:"column:stage;type:enum('creating','bounding','bounded','deleting','deleted')"` Creator string `gorm:"column:creator;type:VARCHAR(255)"` TiDBID string `gorm:"column:tidb_id;type:varchar(255);not null"` TenantID string `gorm:"column:tenant_id;type:varchar(255);not null"` SourceType PremiumDBType `gorm:"column:source_type;type:enum('mysql','alicloud-rds-mysql','azure-flexible-mysql');not null;default:mysql"` BinlogName string `gorm:"column:binlog_name;type:VARCHAR(256)"` BinlogPos int32 `gorm:"column:binlog_pos;type:INT"` BinlogGTID string `gorm:"column:binlog_gtid;type:VARCHAR(255)"` ConnProfile PremiumConnProfile `gorm:"column:conn_profile;type:JSON"` BlockAllowRules PremiumBlockAllowRules `gorm:"column:block_allow_rules;type:JSON"` MigrationRules PremiumMigrationRules `gorm:"column:migration_rules;type:JSON"` SubTaskStatus *PremiumSubTaskStatus `gorm:"column:sub_task_status;type:JSON"` DMTaskID string `gorm:"column:dm_task_id;type:varchar(255);not null;index"` } type PremiumSubTaskStatus struct { DumpStatus *dm.DumpStatus `json:"dump_status,omitempty"` // error message when something wrong ErrorMsg *string `json:"error_msg,omitempty"` LoadStatus *dm.LoadStatus `json:"load_status,omitempty"` // task name Name string `json:"name"` // source name SourceName string `json:"source_name"` Stage dm.TaskStage `json:"stage"` SyncStatus *dm.SyncStatus `json:"sync_status,omitempty"` // task unit type Unit string `json:"unit"` UnresolvedDdlLockId *string `json:"unresolved_ddl_lock_id,omitempty"` // worker name WorkerName string `json:"worker_name"` } //type DumpStatus struct { // Bps int64 `json:"bps"` // CompletedTables float64 `json:"completed_tables"` // EstimateTotalRows float64 `json:"estimate_total_rows"` // FinishedBytes float64 `json:"finished_bytes"` // FinishedRows float64 `json:"finished_rows"` // Progress string `json:"progress"` // TotalTables int64 `json:"total_tables"` //} // //type LoadStatus struct { // Bps int64 `json:"bps"` // FinishedBytes int64 `json:"finished_bytes"` // MetaBinlog string `json:"meta_binlog"` // MetaBinlogGtid string `json:"meta_binlog_gtid"` // Progress string `json:"progress"` // TotalBytes int64 `json:"total_bytes"` //} // //type PremiumSubTaskStage string // //// List of TaskStage //const ( // PREMIUMSUBTASKSTAGE_STOPPED PremiumSubTaskStage = "Stopped" // PREMIUMSUBTASKSTAGE_RUNNING PremiumSubTaskStage = "Running" // PREMIUMSUBTASKSTAGE_FINISHED PremiumSubTaskStage = "Finished" // PREMIUMSUBTASKSTAGE_PAUSED PremiumSubTaskStage = "Paused" //) //type SyncStatus struct { // BinlogType string `json:"binlog_type"` // // sharding DDL which current is blocking // BlockingDdls []string `json:"blocking_ddls"` // DumpIoTotalBytes int64 `json:"dump_io_total_bytes"` // IoTotalBytes int64 `json:"io_total_bytes"` // MasterBinlog string `json:"master_binlog"` // MasterBinlogGtid string `json:"master_binlog_gtid"` // RecentTps int64 `json:"recent_tps"` // SecondsBehindMaster int64 `json:"seconds_behind_master"` // Synced bool `json:"synced"` // SyncerBinlog string `json:"syncer_binlog"` // SyncerBinlogGtid string `json:"syncer_binlog_gtid"` // TotalEvents int64 `json:"total_events"` // TotalTps int64 `json:"total_tps"` // // sharding groups which current are un-resolved // UnresolvedGroups []ShardingGroup `json:"unresolved_groups"` //} //type ShardingGroup struct { // DdlList []string `json:"ddl_list"` // FirstLocation string `json:"first_location"` // Synced []string `json:"synced"` // Target string `json:"target"` // Unsynced []string `json:"unsynced"` //} func (s PremiumSubTaskStatus) Value() (driver.Value, error) { return json.Marshal(s) } func (s *PremiumSubTaskStatus) Scan(input any) error { return json.Unmarshal(input.([]byte), s) } type PremiumConnProfile struct { Type PremiumConnType `json:"type"` Target string `json:"target"` SinkEndpointID uint64 `json:"sink_endpoint_id,string"` Port int32 `json:"port"` User string `json:"user"` Password string `json:"password"` Security *PremiumSecurity `json:"security"` } func (t PremiumConnProfile) Value() (driver.Value, error) { return json.Marshal(t) } func (t *PremiumConnProfile) Scan(input any) error { return json.Unmarshal(input.([]byte), t) } type PremiumSecurity struct { CertAllowedCn []string `json:"cert_allowed_cn"` SslCaContent []byte `json:"ssl_ca_content"` SslCertContent []byte `json:"ssl_cert_content"` SslKeyContent []byte `json:"ssl_key_content"` } func (t PremiumSecurity) Value() (driver.Value, error) { return json.Marshal(t) } func (t *PremiumSecurity) Scan(input any) error { return json.Unmarshal(input.([]byte), t) } type PremiumBlockAllowRules struct { DoDBs []string `json:"do_dbs"` DoTables []PremiumTable `json:"do_tables"` } func (t PremiumBlockAllowRules) Value() (driver.Value, error) { return json.Marshal(t) } func (t *PremiumBlockAllowRules) Scan(input any) error { return json.Unmarshal(input.([]byte), t) } type PremiumTable struct { Schema string `json:"schema"` Table string `json:"table"` } func (t PremiumTable) Value() (driver.Value, error) { return json.Marshal(t) } func (t *PremiumTable) Scan(input any) error { return json.Unmarshal(input.([]byte), t) } // PremiumMigrationRules represents an array of MigrationRule persisted as JSON type PremiumMigrationRules []PremiumMigrationRule // PremiumMigrationRule describes how to route from a source pattern to a target table type PremiumMigrationRule struct { Source PremiumMigrationRuleSource `json:"source"` Target PremiumMigrationRuleTarget `json:"target"` } // PremiumMigrationRuleSource defines the pattern of source schema/table type PremiumMigrationRuleSource struct { SchemaPattern string `json:"schema_pattern"` TablePattern string `json:"table_pattern"` } // PremiumMigrationRuleTarget defines the concrete target schema/table type PremiumMigrationRuleTarget struct { Schema string `json:"schema"` Table string `json:"table"` } func (r PremiumMigrationRules) Value() (driver.Value, error) { return json.Marshal(r) } func (r *PremiumMigrationRules) Scan(input any) error { return json.Unmarshal(input.([]byte), r) } type PremiumDMPrecheckStage string const ( PremiumDMPrecheckStagePending PremiumDMPrecheckStage = "pending" PremiumDMPrecheckStageRunning PremiumDMPrecheckStage = "running" PremiumDMPrecheckStageFinished PremiumDMPrecheckStage = "finished" PremiumDMPrecheckStageCanceled PremiumDMPrecheckStage = "canceled" PremiumDMPrecheckStageFailed PremiumDMPrecheckStage = "failed" ) // PremiumPrecheckItemType is the persisted identity of a premium DM precheck step. // It is defined by this service (not by tiflow or public API protos). Numeric values are // stable for storage; add new values only with new numbers — do not renumber. type PremiumPrecheckItemType int32 const ( PremiumPrecheckItemTypeUnspecified PremiumPrecheckItemType = 0 PremiumPrecheckItemTypeDumpPrivilege PremiumPrecheckItemType = 1 PremiumPrecheckItemTypeReplicationPrivilege PremiumPrecheckItemType = 2 PremiumPrecheckItemTypeVersion PremiumPrecheckItemType = 3 PremiumPrecheckItemTypeServerID PremiumPrecheckItemType = 4 PremiumPrecheckItemTypeBinlogEnable PremiumPrecheckItemType = 5 PremiumPrecheckItemTypeBinlogFormat PremiumPrecheckItemType = 6 PremiumPrecheckItemTypeBinlogRowImage PremiumPrecheckItemType = 7 PremiumPrecheckItemTypeTableSchema PremiumPrecheckItemType = 8 PremiumPrecheckItemTypeBinlogDB PremiumPrecheckItemType = 9 PremiumPrecheckItemTypeConnNumber PremiumPrecheckItemType = 10 PremiumPrecheckItemTypeTargetDBPrivilege PremiumPrecheckItemType = 11 PremiumPrecheckItemTypeMetaPosition PremiumPrecheckItemType = 12 PremiumPrecheckItemTypeLightningTableEmpty PremiumPrecheckItemType = 13 PremiumPrecheckItemTypePrimaryKey PremiumPrecheckItemType = 14 ) type PremiumDMPrecheck struct { ID string `gorm:"column:id;type:varchar(255);not null;primaryKey"` CreatedAt time.Time `gorm:"column:created_at;type:TIMESTAMP DEFAULT CURRENT_TIMESTAMP;not null"` // create and read only UpdatedAt *time.Time `gorm:"column:updated_at;type:TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP"` CanceledTime *time.Time `gorm:"column:canceled_time;type:TIMESTAMP NULL DEFAULT NULL"` Creator string `gorm:"column:creator;type:VARCHAR(255)"` TiDBID string `gorm:"column:tidb_id;type:varchar(255);not null;index"` ProjectID string `gorm:"column:project_id;type:varchar(255);not null"` TenantID string `gorm:"column:tenant_id;type:varchar(255);not null;index"` Request json.RawMessage `gorm:"column:request;type:JSON"` ErrMsg string `gorm:"column:err_msg;type:text"` Stage string `gorm:"column:stage;type:enum('pending','running','finished','failed','canceled');index"` TotalItems int32 `gorm:"column:total_items;type:INT"` DMPrecheckItems []PremiumDMPrecheckItem `gorm:"foreignKey:DMPrecheckID"` } type PremiumDMPrecheckItem struct { ID string `gorm:"column:id;type:varchar(255);not null;primaryKey"` CreatedAt time.Time `gorm:"column:created_at;type:TIMESTAMP DEFAULT CURRENT_TIMESTAMP;not null"` // create and read only UpdatedAt *time.Time `gorm:"column:updated_at;type:TIMESTAMP NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP"` CanceledTime *time.Time `gorm:"column:canceled_time;type:TIMESTAMP NULL DEFAULT NULL"` Stage string `gorm:"column:stage;type:enum('pending','running','finished','failed','canceled');index:idx_dm_precheck_item_stage"` Description string `gorm:"column:description;type:varchar(1024);not null"` Type PremiumPrecheckItemType `gorm:"column:type;type:INT;not null"` ErrMsg string `gorm:"column:err_msg;type:text"` Results PremiumResults `gorm:"column:results;type:JSON"` DMPrecheckID string `gorm:"column:dm_precheck_id;type:varchar(255);not null;index:idx_dm_precheck_item_precheck"` } type PremiumResults pchecker.Results func (r PremiumResults) Value() (driver.Value, error) { b, err := json.Marshal(r) return string(b), err } func (r *PremiumResults) Scan(input any) error { return json.Unmarshal(input.([]byte), r) } type PremiumDMTaskFilter struct { ClusterID string TenantID string Stages []PremiumDMTaskStage OrderBy string Offset *int Limit *int } type PremiumDMPrecheckFilter struct { ClusterID string TenantID string Stages []string OrderBy string Offset *int Limit *int } const ( PremiumDMTaskPrefix = "tsk" PremiumDMSourcePrefix = "src" ) func GenerateDMIDWithPrefix(prefix string) string { base32Encoder := base32.StdEncoding.WithPadding(base32.NoPadding) uuidV4 := uuid.New() return "dm" + prefix + strings.ToLower(base32Encoder.EncodeToString(uuidV4[:])) }