1
0
0
0
专栏/.../

为TiDB DM添加阿里云RDS/DMS Online DDL支持

 pp9fly  发表于  2021-10-18
转载实践案例

转载好友文章：
作者：LittleMagic
链接：https://www.jianshu.com/p/bb7dbebbc552

Online DDL即无锁表结构变更，能够避免对表（尤其是大表）进行更改时，长时间阻塞DML操作。我们当前采用TiDB的DM组件实现上游许多业务库的合库合表与数据汇聚，DM原生支持的Online DDL工具有pt-osc（Percona开源）与gh-ost（GitHub开源）两种。但是，我们的业务库绝大多数都是阿里云RDS MySQL，采用其DMS工具做无锁变更。这导致我们每次都需要手动执行DDL语句，再指定位点恢复DM任务，操作起来比较繁琐，所以还是自己动手丰衣足食比较好。

gh-ost/DMS Online DDL Procedure

通过观察DMS Online DDL产生的binlog，可以发现它的风格与gh-ost相同。简单复习一下gh-ost的处理流程：

image

  1. 在目标库上创建影子表（ghost table，后缀为 _gho ），它的结构与被变更的原表完全相同；
  2. 创建日志表（log table，后缀为 _ghc ）用于记录整个DDL执行过程中的状态；
  3. 在影子表上执行DDL语句；
  4. gh-ost将自身伪装成MySQL slave，创建binlog streamer，将原表的全量数据和增量binlog迁移到影子表（两个操作同时进行）；
  5. 迁移完毕后，将原表锁表，并重命名加上 _del 后缀，再将影子表重命名去掉 _gho 后缀，完成cut-over。两个重命名操作是原子性执行的；
  6. 删除 _ghc 和 _del 表，并关闭binlog streamer。

原子性cut-over的思想十分巧妙，基于MySQL内部锁表之后，执行RENAME操作的优先级比任何DML都要高这一简单的原理。具体的分析可参见国外大佬的这篇文章。

阿里云DMS的执行流程与上述一样，只是表的命名有所变化而已：

  • 影子表： tp_[id]_ogt_[table]
  • 日志表： tp_[id]_ogl_[table]
  • 删除表： tp_[id]_del_[table]

下面直接上代码。

Hacking to the Code

DM将Online DDL过程中的表分为3类，代码文件online_ddl.go中的定义是：

type TableType string

const (
    realTable  TableType = "real table"
    ghostTable TableType = "ghost table"
    trashTable TableType = "trash table" // means we should ignore these tables
)
  • real table：原始表；
  • ghost table：影子表；
  • trash table：日志表与删除表。

另外还会通过OnlineDDLStorage结构体来维护执行过程中的必要信息，如数据库连接、schema/table名称、DDL语句等。

type OnlineDDLStorage struct {
    sync.RWMutex

    cfg *config.SubTaskConfig

    db        *conn.BaseDB
    dbConn    *DBConn
    schema    string // schema name, set through task config
    tableName string // table name with schema, now it's task name
    id        string // the source ID of the upstream MySQL/MariaDB replica.

    // map ghost schema => [ghost table => ghost ddl info, ...]
    ddls map[string]map[string]*GhostDDLInfo

    logCtx *tcontext.Context
}

DM内部的Online DDL模块是插件化的，只需实现OnlinePlugin接口即可。该接口的定义如下，注释写得比较清楚，笔者就不多废话了。

// OnlinePlugin handles online ddl solutions like pt, gh-ost.
type OnlinePlugin interface {
    // Apply does:
    // * detect online ddl
    // * record changes
    // * apply online ddl on real table
    // returns sqls, replaced/self schema, replaced/self table, error
    Apply(tctx *tcontext.Context, tables []*filter.Table, statement string, stmt ast.StmtNode) ([]string, string, string, error)
    // Finish would delete online ddl from memory and storage
    Finish(tctx *tcontext.Context, schema, table string) error
    // TableType returns ghhost/real table
    TableType(table string) TableType
    // RealName returns real table name that removed ghost suffix and handled by table router
    RealName(schema, table string) (string, string)
    // ResetConn reset db connection
    ResetConn(tctx *tcontext.Context) error
    // Clear clears all online information
    // TODO: not used now, check if we could remove it later
    Clear(tctx *tcontext.Context) error
    // Close closes online ddl plugin
    Close()
}

我们需要重点实现的方法是Apply()、TableType()和RealName()。后两者的逻辑比较简单，代码如下。

func (r *AliRDS) TableType(table string) TableType {
    if len(table) > 8 && strings.HasPrefix(table, "tp_") {
        if strings.Contains(table, "_ogt_") {
            return ghostTable
        }

        if strings.Contains(table, "_ogl_") || strings.Contains(table, "_del_") {
            return trashTable
        }
    }

    return realTable
}

func (r *AliRDS) RealName(schema, table string) (string, string) {
    tp := r.TableType(table)
    idx := -1

    if tp == ghostTable {
        idx = strings.Index(table, "_ogt_")
    } else if tp == trashTable {
        idx = strings.Index(table, "_ogl_")
        if idx == -1 {
            idx = strings.Index(table, "_del_")
        }
    }

    if idx > 0 {
        table = table[idx+5:]
    }
    return schema, table
}

接下来思考如何将DMS的步骤转化成TiDB的处理方式：

  • 只执行原表上的DML操作；
  • 不创建日志表；
  • 不创建影子表，但是将要执行的DDL记录到DM元数据库（默认为 dm_meta ）以及DM-Worker的内存里；
  • 在cut-over阶段，将上述记录的DDL的影子表名替换成原表名，直接执行替换后的DDL。

然后就可以顺理成章地写出Apply()方法了。

func (r *AliRDS) Apply(tctx *tcontext.Context, tables []*filter.Table, statement string, stmt ast.StmtNode) ([]string, string, string, error) {
    if len(tables) < 1 {
        return nil, "", "", terror.ErrSyncerUnitAliRDSApplyEmptyTable.Generate()
    }

    schema, table := tables[0].Schema, tables[0].Name
    targetSchema, targetTable := r.RealName(schema, table)
    tp := r.TableType(table)

    switch tp {
    case realTable:
        switch stmt.(type) {
        case *ast.RenameTableStmt:
            if len(tables) != parserpkg.SingleRenameTableNameNum {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()
            }

            tp1 := r.TableType(tables[1].Name)
            if tp1 == trashTable {
                return nil, "", "", nil
            } else if tp1 == ghostTable {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameToGhostTable.Generate(statement)
            }
        }
        return []string{statement}, schema, table, nil

    case trashTable:
        switch stmt.(type) {
        case *ast.RenameTableStmt:
            if len(tables) != parserpkg.SingleRenameTableNameNum {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()
            }

            tp1 := r.TableType(tables[1].Name)
            if tp1 == ghostTable {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameGhostTblToOther.Generate(statement)
            }
        }

    case ghostTable:
        switch stmt.(type) {
        case *ast.CreateTableStmt:
            err := r.storage.Delete(tctx, schema, table)
            if err != nil {
                return nil, "", "", err
            }
        case *ast.DropTableStmt:
            err := r.storage.Delete(tctx, schema, table)
            if err != nil {
                return nil, "", "", err
            }
        case *ast.RenameTableStmt:
            if len(tables) != parserpkg.SingleRenameTableNameNum {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameTableNotValid.Generate()
            }

            tp1 := r.TableType(tables[1].Name)
            if tp1 == realTable {
                rdsInfo := r.storage.Get(schema, table)
                if rdsInfo != nil {
                    return rdsInfo.DDLs, tables[1].Schema, tables[1].Name, nil
                }
                return nil, "", "", terror.ErrSyncerUnitAliRDSOnlineDDLOnGhostTbl.Generate(schema, table)
            } else if tp1 == ghostTable {
                return nil, "", "", terror.ErrSyncerUnitAliRDSRenameGhostTblToOther.Generate(statement)
            }

            err := r.storage.Delete(tctx, schema, table)
            if err != nil {
                return nil, "", "", err
            }
        }

    default:
        err := r.storage.Save(tctx, schema, table, targetSchema, targetTable, statement)
        if err != nil {
            return nil, "", "", err
        }
    }

    return nil, schema, table, nil
}

最后不要忘了添加Online DDL配置项的定义，以及相关的错误信息等边角代码，不一一列举了。

var OnlineDDLSchemes = map[string]func(*tcontext.Context, *config.SubTaskConfig) (OnlinePlugin, error){
    config.PT:    NewPT,
    config.GHOST: NewGhost,
    config.ALIRDS: NewAliRDS,
}

const (
    GHOST  = "gh-ost"
    PT     = "pt"
    ALIRDS = "ali-rds"
)

重新编译dm-master与dm-worker二进制文件，并替换掉原本通过TiUP部署的文件（具体路径因人而异）。

CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make dm-master
CGO_ENABLED=0 GOOS=linux GOARCH=amd64 make dm-worker

在DM作业的配置文件中指定 online-ddl-scheme 参数。

online-ddl-scheme: "ali-rds"

使用DMS工具做无锁表变更操作，可以发现能够正常同步了。

The End

继续搬砖去了。

Enjoy~

1
0
0
0

声明：本文转载于

评论
暂无评论
gh-ost/DMS Online DDL ProcedureHacking to the CodeThe End