Add a tool to parse and print relay log content

Makefile

@@ -1,5 +1,5 @@
 ### Makefile for tidb-binlog
-.PHONY: build test check update clean pump drainer fmt reparo integration_test arbiter binlogctl
+.PHONY: build test check update clean pump drainer fmt reparo integration_test arbiter binlogctl relayprinter
 
 PROJECT=tidb-binlog
 
@@ -54,6 +54,9 @@ reparo:
 binlogctl:
 	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/binlogctl cmd/binlogctl/main.go
 
+relayprinter:
+	$(GOBUILD) -ldflags '$(LDFLAGS)' -o bin/relayprinter cmd/relayprinter/main.go
+
 install:
 	go install ./...
 

cmd/relayprinter/main.go

@@ -0,0 +1,56 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package main
+
+import (
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb-binlog/pkg/version"
+	"github.com/pingcap/tidb-binlog/relayprinter"
+	"go.uber.org/zap/zapcore"
+)
+
+func main() {
+	log.SetLevel(zapcore.ErrorLevel) // increase error log level
+	cfg := relayprinter.NewConfig()
+	if err := cfg.Parse(os.Args[1:]); err != nil {
+		panic(fmt.Sprintf("verify flags failed, see 'relayprinter --help'. %v", err))
+	}
+
+	fmt.Println(version.GetRawVersionInfo())
+
+	sc := make(chan os.Signal, 1)
+	signal.Notify(sc,
+		syscall.SIGHUP,
+		syscall.SIGINT,
+		syscall.SIGTERM,
+		syscall.SIGQUIT)
+
+	p := relayprinter.NewPrinter(cfg)
+
+	go func() {
+		<-sc
+		p.Close()
+		os.Exit(0)
+	}()
+
+	if err := p.Process(); err != nil {
+		fmt.Println("relay-printer process failed", err)
+	}
+	p.Close()
+}

cmd/relayprinter/relayprinter.toml

@@ -0,0 +1,23 @@
+# directory to relay log files
+data-dir = "/dir/to/relay/log"
+
+# specific file to handle, empty for all files
+# file = "binlog-0000000000000067-20201111050014"
+
+# do-db priority over do-table if have same db name
+# and we support regular expression , start with '~' declare use regular expression.
+#
+# do-db = ["~^b.*","s1"]
+# [[do-table]]
+# db-name = "test"
+# tbl-name = "log"
+
+# [[do-table]]
+# db-name = "test"
+# tbl-name = "~^a.*"
+
+# ignore-db = ["~^c.*","s2"]
+# [[ignore-table]]
+# db-name = "test"
+# tbl-name = "~^a.*"
+

pkg/loader/executor.go

@@ -201,7 +201,7 @@ func (e *executor) bulkReplace(inserts []*DML) error {
 
 	var builder strings.Builder
 
-	cols := "(" + buildColumnList(info.columns) + ")"
+	cols := "(" + BuildColumnList(info.columns) + ")"
 	builder.WriteString("REPLACE INTO " + inserts[0].TableName() + cols + " VALUES ")
 
 	holder := fmt.Sprintf("(%s)", holderString(len(info.columns)))

pkg/loader/load.go

@@ -400,7 +400,7 @@ func (s *loaderImpl) execDDL(ddl *DDL) error {
 		}
 
 		if len(ddl.Database) > 0 && !isCreateDatabaseDDL(ddl.SQL) {
-			_, err = tx.Exec(fmt.Sprintf("use %s;", quoteName(ddl.Database)))
+			_, err = tx.Exec(fmt.Sprintf("use %s;", QuoteName(ddl.Database)))
 			if err != nil {
 				if rbErr := tx.Rollback(); rbErr != nil {
 					log.Error("Rollback failed", zap.Error(rbErr))

pkg/loader/model.go

@@ -170,12 +170,12 @@ func (dml *DML) updateSQL() (sql string, args []interface{}) {
 
 	fmt.Fprintf(builder, "UPDATE %s SET ", dml.TableName())
 
-	for _, name := range dml.columnNames() {
+	for _, name := range dml.ColumnNames() {
 		if len(args) > 0 {
 			builder.WriteByte(',')
 		}
 		arg := dml.Values[name]
-		fmt.Fprintf(builder, "%s = ?", quoteName(name))
+		fmt.Fprintf(builder, "%s = ?", QuoteName(name))
 		args = append(args, arg)
 	}
 
@@ -196,9 +196,9 @@ func (dml *DML) buildWhere(builder *strings.Builder) (args []interface{}) {
 			builder.WriteString(" AND ")
 		}
 		if wargs[i] == nil {
-			builder.WriteString(quoteName(wnames[i]) + " IS NULL")
+			builder.WriteString(QuoteName(wnames[i]) + " IS NULL")
 		} else {
-			builder.WriteString(quoteName(wnames[i]) + " = ?")
+			builder.WriteString(QuoteName(wnames[i]) + " = ?")
 			args = append(args, wargs[i])
 		}
 	}
@@ -235,7 +235,7 @@ func (dml *DML) whereSlice() (colNames []string, args []interface{}) {
 	}
 
 	// Fallback to use all columns
-	names := dml.columnNames()
+	names := dml.ColumnNames()
 	return names, dml.whereValues(names)
 }
 
@@ -250,7 +250,8 @@ func (dml *DML) deleteSQL() (sql string, args []interface{}) {
 	return
 }
 
-func (dml *DML) columnNames() []string {
+// ColumnNames returns sorted column names.
+func (dml *DML) ColumnNames() []string {
 	names := make([]string, 0, len(dml.Values))
 
 	for name := range dml.Values {
@@ -262,8 +263,8 @@ func (dml *DML) columnNames() []string {
 }
 
 func (dml *DML) replaceSQL() (sql string, args []interface{}) {
-	names := dml.columnNames()
-	sql = fmt.Sprintf("REPLACE INTO %s(%s) VALUES(%s)", dml.TableName(), buildColumnList(names), holderString(len(names)))
+	names := dml.ColumnNames()
+	sql = fmt.Sprintf("REPLACE INTO %s(%s) VALUES(%s)", dml.TableName(), BuildColumnList(names), holderString(len(names)))
 	for _, name := range names {
 		v := dml.Values[name]
 		args = append(args, v)

pkg/loader/util.go

@@ -173,7 +173,8 @@ func quoteSchema(schema string, table string) string {
 	return fmt.Sprintf("`%s`.`%s`", escapeName(schema), escapeName(table))
 }
 
-func quoteName(name string) string {
+// QuoteName quotes a database/table/column... name.
+func QuoteName(name string) string {
 	return "`" + escapeName(name) + "`"
 }
 
@@ -208,13 +209,14 @@ func splitDMLs(dmls []*DML, size int) (res [][]*DML) {
 	return
 }
 
-func buildColumnList(names []string) string {
+// BuildColumnList builds column names as "`col1`,`col2`,...".
+func BuildColumnList(names []string) string {
 	var b strings.Builder
 	for i, name := range names {
 		if i > 0 {
 			b.WriteString(",")
 		}
-		b.WriteString(quoteName(name))
+		b.WriteString(QuoteName(name))
 
 	}
 

relayprinter/config.go

@@ -0,0 +1,129 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package relayprinter
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb-binlog/pkg/filter"
+	"github.com/pingcap/tidb-binlog/pkg/util"
+	"github.com/pingcap/tidb-binlog/pkg/version"
+	"go.uber.org/zap"
+)
+
+// Config is the main configuration for the relay printer tool.
+type Config struct {
+	*flag.FlagSet `toml:"-" json:"-"`
+	Dir           string `toml:"data-dir" json:"data-dir"` // directory to relay log files
+	File          string `toml:"file" json:"file"`         // specific file to handle, empty for all files
+
+	DoTables []filter.TableName `toml:"do-table" json:"do-table"`
+	DoDBs    []string           `toml:"do-db" json:"do-db"`
+
+	IgnoreTables []filter.TableName `toml:"ignore-table" json:"ignore-table"`
+	IgnoreDBs    []string           `toml:"ignore-db" json:"ignore-db"`
+
+	configFile   string
+	printVersion bool
+}
+
+// NewConfig creates a Config instance.
+func NewConfig() *Config {
+	c := &Config{}
+	c.FlagSet = flag.NewFlagSet("relay-printer", flag.ContinueOnError)
+	fs := c.FlagSet
+	fs.Usage = func() {
+		fmt.Fprintln(os.Stderr, "Usage of relay-printer:")
+		fs.PrintDefaults()
+	}
+
+	fs.StringVar(&c.Dir, "data-dir", "", "directory to relay log files")
+	fs.StringVar(&c.File, "file", "", "specific file to handle, empty for all files")
+	fs.StringVar(&c.configFile, "config", "", "path to the configuration file")
+	fs.BoolVar(&c.printVersion, "V", false, "print relay-printer version info")
+
+	return c
+}
+
+func (c *Config) String() string {
+	cfgBytes, err := json.Marshal(c) //nolint:staticcheck
+	if err != nil {
+		log.Error("marshal config failed", zap.Error(err))
+	}
+
+	return string(cfgBytes)
+}
+
+// Parse parses keys/values from command line flags and toml configuration file.
+func (c *Config) Parse(args []string) error {
+	// Parse first to get the config file
+	perr := c.FlagSet.Parse(args)
+	switch perr {
+	case nil:
+	case flag.ErrHelp:
+		os.Exit(0)
+	default:
+		os.Exit(2)
+	}
+
+	if c.printVersion {
+		fmt.Println(version.GetRawVersionInfo())
+		os.Exit(0)
+	}
+
+	if c.configFile != "" {
+		// Load config file if specified
+		if err := c.configFromFile(c.configFile); err != nil {
+			return errors.Trace(err)
+		}
+	}
+
+	// Parse again to replace with command line options
+	if err := c.FlagSet.Parse(args); err != nil {
+		return errors.Trace(err)
+	}
+	if len(c.FlagSet.Args()) > 0 {
+		return errors.Errorf("'%s' is not a valid flag", c.FlagSet.Arg(0))
+	}
+	c.adjustDoDBAndTable()
+
+	return errors.Trace(c.validate())
+}
+
+func (c *Config) configFromFile(path string) error {
+	return util.StrictDecodeFile(path, "relay-printer", c)
+}
+
+func (c *Config) adjustDoDBAndTable() {
+	for i := 0; i < len(c.DoTables); i++ {
+		c.DoTables[i].Table = strings.ToLower(c.DoTables[i].Table)
+		c.DoTables[i].Schema = strings.ToLower(c.DoTables[i].Schema)
+	}
+	for i := 0; i < len(c.DoDBs); i++ {
+		c.DoDBs[i] = strings.ToLower(c.DoDBs[i])
+	}
+}
+
+func (c *Config) validate() error {
+	if c.Dir == "" {
+		return errors.New("data-dir is empty")
+	}
+	return nil
+}