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"
)

var (
	startPos    int
	endPos      int
	startTime   int64
	endTime     int64
	includeGtid string
	excludeGtid string
	filePath    []string
	outPath     string
	vbo         bool
	skipquery   bool
)

func init() {
	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")

}

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
		}
		ParseBinlog()
	},
}

func main() {
	cmd.Execute()
}

func ParseBinlog() {
	var err error
	var igtid, egtid *gtid.Gtid
	if includeGtid != "" {
		igtid, err = gtid.Parse(includeGtid)
		if err != nil {
			starlog.Errorln(err)
			return
		}
	}
	if excludeGtid != "" {
		egtid, err = gtid.Parse(excludeGtid)
		if err != nil {
			starlog.Errorln(err)
			return
		}
	}
	foundCount := 0
	var totalGtid *gtid.Gtid
	var owrt *xlsx.File
	var res *xlsx.Sheet
	if outPath != "" {
		owrt = xlsx.NewFile()
		res, err = owrt.AddSheet("结果")
		if err != nil {
			starlog.Errorln(err)
			return
		}
		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("单语句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("变更内容")
		res.SetColWidth(0, 0, 5)
		res.SetColWidth(1, 1, 40)
		res.SetColWidth(3, 6, 6)
		res.SetColWidth(7, 7, 5)
		res.SetColWidth(15, 15, 40)
		owrt.Save(outPath)
	}
	getParser := func(fpath string) {
		err = binlog.ParseBinlogFile(fpath, func(tx binlog.Transaction) {
			if startTime != 0 && tx.Timestamp < startTime {
				return
			}
			if endTime != 0 && tx.Timestamp > endTime {
				return
			}
			if tx.StartPos < startPos {
				return
			}
			if endPos > 0 && tx.EndPos > endPos {
				return
			}
			if igtid != nil {
				if c, _ := igtid.Contain(tx.GTID); !c {
					return
				}
			}
			if egtid != nil {
				if c, _ := egtid.Contain(tx.GTID); c {
					return
				}
			}

			foundCount++
			if totalGtid == nil {
				totalGtid, _ = gtid.Parse(tx.GTID)
			} else {
				totalGtid.Add(tx.GTID)
			}
			if !vbo {
				fmt.Printf("已找到%d个合规GTID\r", foundCount)
			} else {
				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)
			}
			if outPath != "" {
				for _, t := range tx.Txs {
					if skipquery && (strings.ToLower(t.Sql) == "begin" || strings.ToLower(t.Sql) == "commit") {
						continue
					}
					r := res.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)
					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)
					r.AddCell().SetValue(t.Rows)
				}
			}
		})
		if outPath != "" {
			err = owrt.Save(outPath)
			if err != nil {
				starlog.Errorln(err)
				return
			}
		}
		if err != nil {
			starlog.Errorln(err)
			return
		}
	}
	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(`^mysql-bin\.\d+$`)
			for _, v := range files {
				if !v.IsDir() && rgp.MatchString(v.Name()) {
					getParser(filepath.Join(fp, v.Name()))
				}
			}
		} else {
			getParser(fp)
		}
	}

	if outPath != "" {
		owrt.Save(outPath)
	}
	fmt.Println("")
	allGtid := ""
	if totalGtid != nil {
		allGtid = totalGtid.String()
	}
	fmt.Printf("Total Gtid:%v\nTotal Binlog Number:%v\n", allGtid, foundCount)
}