binlog-parser/main.go

package main

import (
	"b612.me/mysql/binlog"
	"b612.me/mysql/gtid"
	"b612.me/starlog"
	"b612.me/staros"
	"fmt"
	"github.com/spf13/cobra"
	"github.com/tealeg/xlsx"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"time"
)

var (
	bigThan     int
	smallThan   int
	startPos    int
	endPos      int
	startTime   int64
	endTime     int64
	includeGtid string
	excludeGtid string
	filePath    []string
	outPath     string
	vbo         bool
	skipquery   bool
	pos         int64
	prefix      string
	outasRow    bool
	counts      int
)

func init() {
	cmd.Flags().IntVarP(&counts, "count", "c", 0, "counts of binlog transaction")
	cmd.Flags().IntVarP(&endPos, "pos", "P", 0, "skipPos of binlog")
	cmd.Flags().IntVarP(&startPos, "start-pos", "S", 0, "startPos of binlog")
	cmd.Flags().IntVarP(&endPos, "end-pos", "E", 0, "endPos of binlog")
	cmd.Flags().StringVarP(&includeGtid, "include-gtid", "i", "", "include gtid")
	cmd.Flags().StringVarP(&excludeGtid, "exclude-gtid", "e", "", "exclude gtid")
	cmd.Flags().StringSliceVarP(&filePath, "path", "p", []string{}, "binlog file path")
	cmd.Flags().StringVarP(&outPath, "savepath", "o", "", "output excel path")
	cmd.Flags().Int64Var(&startTime, "starttime", 0, "start unix timestamp")
	cmd.Flags().Int64Var(&endTime, "endtime", 0, "end unix timestamp")
	cmd.Flags().BoolVarP(&vbo, "verbose", "v", false, "show the detail verbose")
	cmd.Flags().BoolVarP(&skipquery, "skip-query", "s", true, "skip query write to xlsx like BEGIN COMMIT")
	cmd.Flags().IntVar(&bigThan, "big", 0, "show tx big than x bytes")
	cmd.Flags().IntVar(&smallThan, "small", 0, "show tx small than x bytes")
	cmd.Flags().StringVar(&prefix, "prefix", "mysql-bin", "mysql binlog prefix")
	cmd.Flags().BoolVarP(&outasRow, "row-mode", "r", false, "output as row")
}

var cmd = &cobra.Command{
	Use:   "",
	Short: "binlog parser",
	Long:  "binlog parser",
	Run: func(cmd *cobra.Command, args []string) {
		if len(filePath) == 0 {
			starlog.Warningln("Please enter a binlog path or folder")
			return
		}
		now := time.Now()
		ParseBinlog()
		cost := time.Now().Sub(now).Seconds()
		fmt.Println("")
		fmt.Printf("Time Cost:%.2fs", cost)
	},
}

func main() {
	cmd.Execute()
}

func ParseBinlog() {
	var err error
	foundCount := 0
	var totalGtid *gtid.Gtid
	var owrt *xlsx.File
	if outPath != "" {
		owrt, err = prepareXlsx()
		if err != nil {
			return
		}
	}
	getParser := func(fpath string) string {
		var sTime, eTime time.Time
		if startTime != 0 {
			sTime = time.Unix(startTime, 0)
		}
		if endTime != 0 {
			eTime = time.Unix(endTime, 0)
		}
		onlyGtid := false
		if !vbo && outPath == "" {
			onlyGtid = true
		}
		var filter = binlog.BinlogFilter{
			IncludeGtid:  includeGtid,
			ExcludeGtid:  excludeGtid,
			StartPos:     startPos,
			EndPos:       endPos,
			StartDate:    sTime,
			EndDate:      eTime,
			BigThan:      bigThan,
			SmallThan:    smallThan,
			OnlyShowGtid: onlyGtid,
		}
		var cGtid *gtid.Gtid
		proc := make(chan string, 1000)
		if !vbo {
			go func() {
				var latest string
				var count = uint16(0)
				for {
					select {
					case tmp := <-proc:
						if tmp == "end" {
							fmt.Println(latest)
							return
						}
						latest = tmp
						if count%10 == 0 {
							fmt.Print(latest)
						}
						count++
					case <-time.After(time.Millisecond * 200):
						fmt.Print(latest)
					}
				}
			}()
		}
		err = binlog.ParseBinlogWithFilter(fpath, pos, filter, func(tx binlog.Transaction) bool {
			foundCount++
			if cGtid == nil {
				cGtid, _ = gtid.Parse(tx.GTID)
			} else {
				cGtid.Add(tx.GTID)
			}
			if totalGtid == nil {
				totalGtid, _ = gtid.Parse(tx.GTID)
			} else {
				totalGtid.Add(tx.GTID)
			}
			if !vbo {
				proc <- fmt.Sprintf("已找到%d个合法GTID\r", foundCount)
			} else {
				if !outasRow {
					fmt.Printf("GTID:%s Time:%s StartPos:%v EndPos:%v RowsCount:%v Size:%v Detail:%+v\n",
						tx.GTID, tx.Time, tx.StartPos, tx.EndPos, tx.RowsCount, tx.Size, tx.Txs)
				} else {
					fmt.Printf("-------------------------\nGTID:%s Time:%s StartPos:%v EndPos:%v RowsCount:%v Size:%v\n\n",
						tx.GTID, tx.Time, tx.StartPos, tx.EndPos, tx.RowsCount, tx.Size)
					for _, t := range tx.Txs {
						if skipquery && (strings.ToLower(t.Sql) == "begin" || strings.ToLower(t.Sql) == "commit") {
							continue
						}
						_, sql := generateRowSql(t)
						fmt.Printf("GTID:\t%s\nTime:\t%s\nStartPos:\t%v\nEndPos:\t%v\nRowsCount:\t%v\nSQLOrigin:\t%v\nSQL:\t%+v\n\n",
							tx.GTID, t.Time, t.StartPos, t.EndPos, t.RowCount, t.Sql, sql)
					}
				}
			}
			if outPath != "" {
				add2Xlsx(owrt, tx, foundCount)
			}
			if counts > 0 && foundCount >= counts {
				return false
			}
			return true
		})
		if !vbo {
			time.Sleep(time.Millisecond * 500)
		}
		var cGtidStr string
		if cGtid != nil {
			cGtidStr = cGtid.String()
		}
		if outPath != "" {
			err = owrt.Save(outPath)
			if err != nil {
				starlog.Errorln(err)
				return cGtidStr
			}
		}
		if err != nil {
			starlog.Errorln(err)
			return cGtidStr
		}
		return cGtidStr
	}
	var gtidRes [][]string
	for _, fp := range filePath {
		if staros.IsFolder(fp) {
			files, err := os.ReadDir(fp)
			if err != nil {
				starlog.Errorln("read folder failed:", err)
				return
			}
			rgp := regexp.MustCompile(`^` + prefix + `\.\d+$`)
			for _, v := range files {
				if !v.IsDir() && rgp.MatchString(v.Name()) {
					gtidRes = append(gtidRes, []string{v.Name(), getParser(filepath.Join(fp, v.Name()))})
				}
			}
		} else {
			getParser(fp)
		}
	}
	if outPath != "" {
		owrt.Save(outPath)
	}
	fmt.Println("")
	if len(gtidRes) != 0 {
		for _, v := range gtidRes {
			fmt.Printf("%s:%s\n", v[0], v[1])
		}
	}
	fmt.Println("")
	allGtid := ""
	if totalGtid != nil {
		allGtid = totalGtid.String()
	}
	fmt.Printf("Total Gtid:%v\nTotal SQL Number:%v\n", allGtid, foundCount)
}

func prepareXlsx() (*xlsx.File, error) {
	owrt := xlsx.NewFile()
	res, err := owrt.AddSheet("结果")
	if err != nil {
		starlog.Errorln(err)
		return owrt, err
	}
	title := res.AddRow()
	title.AddCell().SetValue("序号")
	title.AddCell().SetValue("GTID")
	title.AddCell().SetValue("时间")
	title.AddCell().SetValue("时间戳")
	title.AddCell().SetValue("StartPos")
	title.AddCell().SetValue("EndPos")
	title.AddCell().SetValue("事务大小")
	title.AddCell().SetValue("影响行数")
	title.AddCell().SetValue("压缩类型")
	title.AddCell().SetValue("单语句StartPos")
	title.AddCell().SetValue("单语句EndPos")
	title.AddCell().SetValue("单语句时间")
	title.AddCell().SetValue("单语句影响行数")
	title.AddCell().SetValue("单语句影响库")
	title.AddCell().SetValue("单语句影响表")
	title.AddCell().SetValue("SQL类型")
	title.AddCell().SetValue("具体SQL")
	title.AddCell().SetValue("从属事务编号")
	title.AddCell().SetValue("同事务行编号")
	title.AddCell().SetValue("行变更内容")
	res.SetColWidth(0, 0, 5)
	res.SetColWidth(1, 1, 40)
	res.SetColWidth(3, 6, 6)
	res.SetColWidth(7, 7, 5)
	res.SetColWidth(16, 16, 40)
	return owrt, owrt.Save(outPath)
}

func add2Xlsx(owrt *xlsx.File, tx binlog.Transaction, foundCount int) {
	for k, t := range tx.Txs {
		if skipquery && (strings.ToLower(t.Sql) == "begin" || strings.ToLower(t.Sql) == "commit") {
			continue
		}
		addRow := func() *xlsx.Row {
			r := owrt.Sheets[0].AddRow()
			r.AddCell().SetValue(foundCount)
			r.AddCell().SetValue(tx.GTID)
			r.AddCell().SetValue(tx.Time.String())
			r.AddCell().SetValue(tx.Timestamp)
			r.AddCell().SetValue(tx.StartPos)
			r.AddCell().SetValue(tx.EndPos)
			r.AddCell().SetValue(tx.Size)
			r.AddCell().SetValue(tx.RowsCount)
			if t.CompressionType == "" {
				r.AddCell().SetValue("NONE")
			} else {
				r.AddCell().SetValue(t.CompressionType)
			}
			r.AddCell().SetValue(t.StartPos)
			r.AddCell().SetValue(t.EndPos)
			r.AddCell().SetValue(t.Time.String())
			r.AddCell().SetValue(t.RowCount)
			r.AddCell().SetValue(t.Db)
			r.AddCell().SetValue(t.Table)
			r.AddCell().SetValue(t.SqlType)
			r.AddCell().SetValue(t.Sql)
			return r
		}
		if !outasRow {
			r := addRow()
			r.AddCell().SetValue(k + 1)
			r.AddCell().SetValue(1)
			r.AddCell().SetValue(t.Rows)
			continue
		}
		idx, sql := generateRowSql(t)
		r := addRow()
		r.AddCell().SetValue(k + 1)
		r.AddCell().SetValue(idx + 1)
		r.AddCell().SetValue(sql)
	}
}

func generateRowSql(t binlog.TxDetail) (int, interface{}) {
	switch t.SqlType {
	case "insert":
		for idx, rows := range t.Rows {
			setence := ""
			for _, row := range rows {
				switch row.(type) {
				case uint, uint64, uint32, uint16, uint8, int, int64, int32, int16, int8, float64, float32:
					setence += fmt.Sprintf("%v, ", row)
				case string:
					setence += fmt.Sprintf("'%v', ", strings.ReplaceAll(row.(string), "'", "''"))
				case []byte:
					setence += fmt.Sprintf("%v, ", row)
				case nil:
					setence += fmt.Sprintf("%v, ", "NULL")
				default:
					setence += fmt.Sprintf("%v, ", row)
				}
			}
			if setence != "" && len(setence) > 2 {
				setence = setence[:len(setence)-2]
			}
			return idx + 1, fmt.Sprintf(`INSERT INTO %s.%s VALUES(%v)`, t.Db, t.Table, setence)
		}
	case "update":
		var sql string
		var where string
		for idxc, rows := range t.Rows {
			setence := ""
			spec := ", "
			if idxc%2 == 0 {
				spec = " AND "
			}
			for idxf, row := range rows {
				switch row.(type) {
				case uint, uint64, uint32, uint16, uint8, int, int64, int32, int16, int8, float64, float32:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row, spec)
				case string:
					setence += fmt.Sprintf("$%d='%v'%s", idxf, strings.ReplaceAll(row.(string), "'", "''"), spec)
				case []byte:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row, spec)
				case nil:
					setence += fmt.Sprintf("$%d=%v%s", idxf, "NULL", spec)
				default:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row, spec)
				}
			}
			if setence != "" && len(setence) > 2 {
				setence = setence[:len(setence)-len(spec)]
			}
			if idxc%2 == 0 {
				where = setence
				continue
			}
			sql = fmt.Sprintf("UPDATE %s.%s SET (%v) WHERE %v", t.Db, t.Table, setence, where)
			return (idxc + 1) / 2, sql
		}
	case "delete":
		for idx, rows := range t.Rows {
			setence := ""
			spec := " AND "
			for idxf, row := range rows {
				switch row.(type) {
				case uint, uint64, uint32, uint16, uint8, int, int64, int32, int16, int8, float64, float32:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row)
				case string:
					setence += fmt.Sprintf("$%d='%v'%s", idxf, strings.ReplaceAll(row.(string), "'", "''"))
				case []byte:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row)
				case nil:
					setence += fmt.Sprintf("$%d=%v%s", idxf, "NULL")
				default:
					setence += fmt.Sprintf("$%d=%v%s", idxf, row)
				}
			}
			if setence != "" && len(setence) > 2 {
				setence = setence[:len(setence)-len(spec)]
			}
			sql := fmt.Sprintf("DELETE FROM %s.%s WHERE %v", t.Db, t.Table, setence)
			return idx + 1, sql

		}
	default:
		return 1, t.Rows
	}
	return 0, ""
}