利用 VPC Traffic Mirroring 进行数据库流量复制

package main

// 注意: 本方案为 poc 测试代码, 作为方案思路讲解, 线上使用前请充分验证 

import (
    "bufio"
    "database/sql"
    "flag"
    "fmt"
    "io"
    "log"
    "os"
    "os/exec"
    "regexp"

    _ "github.com/go-sql-driver/mysql"
)

var (
    db         *sql.DB
    flgHelp    bool
    printsql   int
    selectonly int
    host       string
    port       int
    user       string
    password   string
    database   string
    ratio      int
    totalNum   int
    errorNum   int
    skipNum    int
)

func main() {

    // 参数解析
    parseCmdLineFlags()
    if flgHelp {
        flag.Usage()
        os.Exit(0)
    }
    flag.Args()
    fmt.Println("\nRunning with Arguments: ")
    fmt.Printf(" * selectonly: %d\n", selectonly)
    fmt.Printf(" * printsql: %d\n", printsql)
    fmt.Printf(" * ratio: %d\n", ratio)
    fmt.Printf(" * host: %s\n", host)
    fmt.Printf(" * port: %d\n", port)
    fmt.Printf(" * user: %s\n\n", user)

    // 初始化数据库
    err := InitDB()
    //defer db.Close()
    if err != nil {
        errMsg := fmt.Sprintf("Failed to connect to database: %s:%s@tcp(%s:%d)/%s?charset=utf8mb4", user, password, host, port, database)
        log.Fatal(errMsg)
        log.Fatal("Error: %v\n", err)
        return
    }
    ExecCommand()
    //RunCommand()
}

// 参数解析
func parseCmdLineFlags() {
    flag.BoolVar(&flgHelp, "help", false, "Show help, this tool is only for Aurora MySQL/RDS MySQL test")
    flag.IntVar(&printsql, "printsql", 0, "Set 1 to print all sqls")
    flag.IntVar(&selectonly, "selectonly", 1, "Set 0 to apply all type sqls including select/insert/update/delete")
    flag.StringVar(&host, "h", "127.0.0.1", "Target mysql host addr")
    flag.IntVar(&port, "P", 3306, "Target mysql port")
    flag.StringVar(&user, "u", "root", "Target mysql user")
    flag.StringVar(&password, "p", "", "Target mysql password")
    flag.StringVar(&database, "db", "", "Target mysql database")
    flag.IntVar(&ratio, "ratio", 1, "Ratio for target replay, default is 1")
    flag.Parse()
}

// 初始化数据库
func InitDB() (err error) {
    log.Println("Init database connection")
    dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4", user, password, host, port, database)
    db, err = sql.Open("mysql", dsn)
    if err != nil {
        return err
    }
    db.SetMaxOpenConns(128)
    db.SetMaxIdleConns(32)

    // 尝试与数据库建立连接
    err = db.Ping()
    if err != nil {
        return err
    } else {
        log.Println("Database is connected successfully")
    }
    return nil
}

// 执行 SQL 语句
func RunSQL(sqlStr string) (err error) {
    if printsql == 1 {
        log.Println("[ Run SQL ] ", sqlStr)
    }
    if totalNum%10000 == 0 {
        log.Println("[ Execute Summary ] ", totalNum, "sqls in total,", errorNum, "failed,", skipNum, "skipped")
    }
    _, err = db.Exec(sqlStr)
    if err != nil {
        errorNum += 1
        log.Println("Query Error, SQL:", err)
        log.Println("Query Error, Msg:", err)
        return
    }
    //defer db.Close()
    return
}

// 输出内容
func ExecCommand() {
    var contentArray = make([]string, 0, 5)
    contentArray = contentArray[0:0]
    cmd := exec.Command("sudo", "tshark", "-i", "capture0", "-T", "fields", "-e", "mysql.query", "-Y", "mysql.command==3", "-l")
    //cmd := exec.Command("sudo", "/home/ec2-user/demo-1/mysql-sniffer", "-i", "capture0")
    //显示运行的命令
    log.Println("Run command: ", cmd)
    stdout, err := cmd.StdoutPipe()
    if err != nil {
        fmt.Fprintln(os.Stderr, "error=>", err.Error())
        return
    }
    cmd.Start() // Start开始执行包含的命令，但并不会等待该命令完成即返回。Wait方法会返回命令的返回状态码并在命令返回后释放相关的资源。

    reader := bufio.NewReader(stdout)

    var index int
    //实时循环读取输出流中的一行内容
    for {
        line, err2 := reader.ReadString('\n')
        if err2 != nil || io.EOF == err2 {
            break
        }
        //fmt.Println(line)
        GetAndFilterOutput(line)
        index++
        contentArray = append(contentArray, line)
    }

    cmd.Wait()
    return
}

// 过滤内容
func GetAndFilterOutput(output string) {

    sqlStr := output
    /*
        // 适用于 mysql-sniffer 的输出结果分割
        fmt.Print("\n==> : ", output, "\n")
        r := regexp.MustCompile("[^\\s]+")
        sqlStr := r.FindAllString(output, -1)[7]
        fmt.Print("\n==> : ", sqlStr, "\n")
    */

    // 对内容进行过滤
    if sqlStr == "\n" {
        return
    }
    /*
        TODO: SQL split 当前仅适配 sysbench 语句, 对于复杂业务语句目前需要做适配性改造
    */

    pattern := "(?i)^SELECT|^INSERT|^UPDATE|^DELETE"
    if selectonly == 1 {
        pattern = "(?i)^SELECT"
    }
    matched, _ := regexp.MatchString(pattern, sqlStr)

    totalNum += 1
    if matched {
        //fmt.Print("Run: ", sqlStr)
        for i := 0; i < ratio; i++ {
            go RunSQL(sqlStr)
        }
    } else {
        skipNum += 1
        //fmt.Print("SKIP: ", sqlStr)
    }

    return
}