Drainer support plugin framework

drainer/config.go

@@ -78,6 +78,11 @@ type SyncerConfig struct {
 	EnableDispatch    bool               `toml:"enable-dispatch" json:"enable-dispatch"`
 	SafeMode          bool               `toml:"safe-mode" json:"safe-mode"`
 	EnableCausality   bool               `toml:"enable-detect" json:"enable-detect"`
+	PluginPath        string             `toml:"plugin-path" json:"plugin-path"`
+	PluginNames       []string           `toml:"plugin-names" json:"plugin-names"`
+	SupportPlugin     bool               `toml:"support-plugin" json:"support-plugin"`
+	MarkDBName        string             `toml:"mark-db-name" json:"mark-db-name"`
+	MarkTableName     string             `toml:"mark-table-name" json:"mark-table-name"`
 }
 
 // RelayConfig is the Relay log's configuration.
@@ -158,6 +163,12 @@ func NewConfig() *Config {
 	fs.IntVar(&cfg.SyncedCheckTime, "synced-check-time", defaultSyncedCheckTime, "if we can't detect new binlog after many minute, we think the all binlog is all synced")
 	fs.StringVar(new(string), "log-rotate", "", "DEPRECATED")
 
+	fs.StringVar(&cfg.SyncerCfg.PluginPath, "plugin-path", "", "The path of the plugins")
+	fs.Var(newSliceNames([]string{}, &cfg.SyncerCfg.PluginNames), "plugin-names", "The names of the plugins")
+	fs.BoolVar(&cfg.SyncerCfg.SupportPlugin, "support-plugin", false, "Whether plugin is supported,default: false")
+	fs.StringVar(&cfg.SyncerCfg.MarkDBName, "mark-db-name", "rel", "mark database's name")
+	fs.StringVar(&cfg.SyncerCfg.MarkTableName, "mark-table-name", "_drainer_repl_mark", "mark table's name")
+
 	return cfg
 }
 

drainer/config_test.go

@@ -293,3 +293,75 @@ func (t *testKafkaSuite) TestConfigDestDBTypeKafka(c *C) {
 	c.Assert(cfg.SyncerCfg.To.KafkaVersion, Equals, defaultKafkaVersion)
 	c.Assert(cfg.SyncerCfg.To.KafkaMaxMessages, Equals, 1024)
 }
+
+func (t *testKafkaSuite) TestConfigPlugin(c *C) {
+	args := []string{}
+
+	cfg := NewConfig()
+	err := cfg.Parse(args)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(cfg.SyncerCfg.PluginPath), Equals, 0)
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 0)
+	c.Assert(cfg.SyncerCfg.SupportPlugin, Equals, false)
+	c.Assert(cfg.SyncerCfg.MarkDBName, Equals, "rel")
+	c.Assert(cfg.SyncerCfg.MarkTableName, Equals, "_drainer_repl_mark")
+
+	args = []string{
+		"-plugin-path", "/tmp/drainer/plugin",
+		"-plugin-names", "demo1",
+		"-support-plugin",
+		"-mark-db-name", "db1",
+		"-mark-table-name", "tb1",
+	}
+
+	cfg = NewConfig()
+	err = cfg.Parse(args)
+	c.Assert(err, IsNil)
+
+	c.Assert(cfg.SyncerCfg.PluginPath, Equals, "/tmp/drainer/plugin")
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 1)
+	c.Assert(cfg.SyncerCfg.PluginNames[0], Equals, "demo1")
+	c.Assert(cfg.SyncerCfg.SupportPlugin, Equals, true)
+	c.Assert(cfg.SyncerCfg.MarkDBName, Equals, "db1")
+	c.Assert(cfg.SyncerCfg.MarkTableName, Equals, "tb1")
+
+	args = []string{
+		"-plugin-names", "demo1,demo2",
+		"-mark-db-name", "",
+		"-mark-table-name", "",
+	}
+
+	cfg = NewConfig()
+	err = cfg.Parse(args)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 2)
+	c.Assert(cfg.SyncerCfg.PluginNames[0], Equals, "demo1")
+	c.Assert(cfg.SyncerCfg.PluginNames[1], Equals, "demo2")
+	c.Assert(cfg.SyncerCfg.MarkDBName, Equals, "")
+	c.Assert(cfg.SyncerCfg.MarkTableName, Equals, "")
+
+	args = []string{
+		"-plugin-names", "",
+	}
+
+	cfg = NewConfig()
+	err = cfg.Parse(args)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 1)
+	c.Assert(cfg.SyncerCfg.PluginNames[0], Equals, "")
+
+	args = []string{
+		"-plugin-names", ",",
+	}
+
+	cfg = NewConfig()
+	err = cfg.Parse(args)
+	c.Assert(err, IsNil)
+
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 2)
+	c.Assert(cfg.SyncerCfg.PluginNames[0], Equals, "")
+	c.Assert(cfg.SyncerCfg.PluginNames[1], Equals, "")
+}

drainer/isyncer.go

@@ -0,0 +1,11 @@
+package drainer
+
+import (
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/pkg/loader"
+)
+
+// SyncerFilter is the interface that for syncer-plugin
+type SyncerFilter interface {
+	FilterTxn(txn *loader.Txn, info *loopbacksync.LoopBackSync) (bool, error)
+}

drainer/loopbacksync/loopbacksync.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb-binlog/pkg/plugin"
 	"go.uber.org/zap"
 )
 
@@ -36,37 +37,58 @@ const (
 	ChannelInfo = "channel_info"
 )
 
-// CreateMarkTableDDL is the DDL to create the mark table.
-var CreateMarkTableDDL string = fmt.Sprintf("CREATE TABLE If Not Exists %s (%s bigint not null,%s bigint not null DEFAULT 0, %s bigint DEFAULT 0, %s varchar(64) ,PRIMARY KEY (%s,%s));", MarkTableName, ID, ChannelID, Val, ChannelInfo, ID, ChannelID)
-
-// CreateMarkDBDDL is DDL to create the database of mark table.
-var CreateMarkDBDDL = "create database IF NOT EXISTS retl;"
-
 //LoopBackSync loopback sync info
 type LoopBackSync struct {
 	ChannelID       int64
 	LoopbackControl bool
+	MarkDBName      string
+	MarkTableName   string
 	SyncDDL         bool
+	Index           int64
+	PluginPath      string
+	PluginNames     []string
+	Hooks           []*plugin.EventHooks
+	SupportPlugin   bool
+	RecordID        int
 }
 
 //NewLoopBackSyncInfo return LoopBackSyncInfo objec
-func NewLoopBackSyncInfo(ChannelID int64, LoopbackControl, SyncDDL bool) *LoopBackSync {
+func NewLoopBackSyncInfo(ChannelID int64, LoopbackControl, SyncDDL bool, path string, names []string, SupportPlug bool, mdbname, mtablename string) *LoopBackSync {
 	l := &LoopBackSync{
 		ChannelID:       ChannelID,
 		LoopbackControl: LoopbackControl,
 		SyncDDL:         SyncDDL,
+		Index:           0,
+		PluginPath:      path,
+		PluginNames:     names,
+		SupportPlugin:   SupportPlug,
+		MarkDBName:      strings.TrimSpace(mdbname),
+		MarkTableName:   strings.TrimSpace(mtablename),
+	}
+	if l.SupportPlugin {
+		l.Hooks = make([]*plugin.EventHooks, 4)
+		l.Hooks[plugin.SyncerFilter] = &plugin.EventHooks{}
+
+		l.Hooks[plugin.ExecutorExtend] = &plugin.EventHooks{}
+		l.Hooks[plugin.LoaderInit] = &plugin.EventHooks{}
+		l.Hooks[plugin.LoaderDestroy] = &plugin.EventHooks{}
+
 	}
 	return l
 }
 
 // CreateMarkTable create the db and table if need.
-func CreateMarkTable(db *sql.DB) error {
-	_, err := db.Exec(CreateMarkDBDDL)
+func CreateMarkTable(db *sql.DB, mdbname, mtablename string) error {
+	var err error
+	var createMarkDBDDL = fmt.Sprintf("create database IF NOT EXISTS %s;", mdbname)
+	_, err = db.Exec(createMarkDBDDL)
 	if err != nil {
 		return errors.Annotate(err, "failed to create mark db")
 	}
 
-	_, err = db.Exec(CreateMarkTableDDL)
+	// CreateMarkTableDDL is the DDL to create the mark table.
+	var createMarkTableDDL string = fmt.Sprintf("CREATE TABLE If Not Exists %s.%s (%s bigint not null,%s bigint not null DEFAULT 0, %s bigint DEFAULT 0, %s varchar(64) ,PRIMARY KEY (%s,%s));", mdbname, mtablename, ID, ChannelID, Val, ChannelInfo, ID, ChannelID)
+	_, err = db.Exec(createMarkTableDDL)
 	if err != nil {
 		return errors.Annotate(err, "failed to create mark table")
 	}

drainer/loopbacksync/loopbacksync_test.go

@@ -15,6 +15,7 @@ package loopbacksync
 
 import (
 	"database/sql/driver"
+	"fmt"
 	"regexp"
 	"testing"
 
@@ -32,25 +33,34 @@ func (s *loopbackSuite) TestNewLoopBackSyncInfo(c *check.C) {
 	var ChannelID int64 = 1
 	var LoopbackControl = true
 	var SyncDDL = false
-	l := NewLoopBackSyncInfo(ChannelID, LoopbackControl, SyncDDL)
+
+	l := NewLoopBackSyncInfo(ChannelID, LoopbackControl, SyncDDL, "", nil, false, "rel", "_drainer_repl_mark")
 
 	c.Assert(l, check.DeepEquals, &LoopBackSync{
 		ChannelID:       ChannelID,
 		LoopbackControl: LoopbackControl,
 		SyncDDL:         SyncDDL,
+		PluginPath:      "",
+		PluginNames:     nil,
+		SupportPlugin:   false,
+		MarkDBName:      "rel",
+		MarkTableName:   "_drainer_repl_mark",
 	})
 }
 
 func (s *loopbackSuite) TestCreateMarkTable(c *check.C) {
 	db, mk, err := sqlmock.New()
 	c.Assert(err, check.IsNil)
 
+	CreateMarkDBDDL := "create database IF NOT EXISTS rel;"
+	CreateMarkTableDDL := fmt.Sprintf("CREATE TABLE If Not Exists %s.%s (%s bigint not null,%s bigint not null DEFAULT 0, %s bigint DEFAULT 0, %s varchar(64) ,PRIMARY KEY (%s,%s));", "rel", "_drainer_repl_mark", ID, ChannelID, Val, ChannelInfo, ID, ChannelID)
+
 	mk.ExpectExec(regexp.QuoteMeta(CreateMarkDBDDL)).
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mk.ExpectExec(regexp.QuoteMeta(CreateMarkTableDDL)).
 		WillReturnResult(sqlmock.NewResult(0, 0))
 
-	err = CreateMarkTable(db)
+	err = CreateMarkTable(db, "rel", "_drainer_repl_mark")
 	c.Assert(err, check.IsNil)
 
 	err = mk.ExpectationsWereMet()

drainer/syncer.go

@@ -19,12 +19,12 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
-	"github.com/pingcap/tidb-binlog/pkg/loader"
-
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/parser/model"
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/pkg/loader"
+	"github.com/pingcap/tidb-binlog/pkg/plugin"
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tidb-binlog/drainer/checkpoint"
@@ -77,8 +77,36 @@ func NewSyncer(cp checkpoint.CheckPoint, cfg *SyncerConfig, jobs []*model.Job) (
 		ignoreDBs = strings.Split(cfg.IgnoreSchemas, ",")
 	}
 	syncer.filter = filter.NewFilter(ignoreDBs, cfg.IgnoreTables, cfg.DoDBs, cfg.DoTables)
-	syncer.loopbackSync = loopbacksync.NewLoopBackSyncInfo(cfg.ChannelID, cfg.LoopbackControl, cfg.SyncDDL)
+	syncer.loopbackSync = loopbacksync.NewLoopBackSyncInfo(cfg.ChannelID, cfg.LoopbackControl, cfg.SyncDDL, cfg.PluginPath,
+		cfg.PluginNames, cfg.SupportPlugin, cfg.MarkDBName, cfg.MarkTableName)
+	if syncer.loopbackSync.SupportPlugin {
+		log.Info("Begin to Load syncer-plugins.")
+		for _, name := range syncer.loopbackSync.PluginNames {
+			n := strings.TrimSpace(name)
+			sym, err := plugin.LoadPlugin(syncer.loopbackSync.Hooks[plugin.SyncerFilter],
+				syncer.loopbackSync.PluginPath, n)
+			if err != nil {
+				log.Error("Load plugin failed.", zap.String("plugin name", n),
+					zap.String("error", err.Error()))
+				continue
+			}
 
+			newPlugin, ok := sym.(func() interface{})
+			if !ok {
+				log.Error("The correct new-function is not provided.", zap.String("plugin name", n), zap.String("type", "syncer plugin"))
+				continue
+			}
+			plg := newPlugin()
+			_, ok = plg.(SyncerFilter)
+			if !ok {
+				log.Info("SyncerFilter interface is not implemented.", zap.String("plugin name", n))
+			} else {
+				plugin.RegisterPlugin(syncer.loopbackSync.Hooks[plugin.SyncerFilter],
+					n, plg)
+				log.Info("Load plugin success.", zap.String("plugin name", n), zap.String("interface", "SyncerFilter"))
+			}
+		}
+	}
 	var err error
 	// create schema
 	syncer.schema, err = NewSchema(jobs, false)
@@ -357,8 +385,31 @@ ForLoop:
 				err = errors.Annotate(err, "handlePreviousDDLJobIfNeed failed")
 				break ForLoop
 			}
+
 			var isFilterTransaction = false
 			var err1 error
+
+			if s.loopbackSync.SupportPlugin {
+				hook := s.loopbackSync.Hooks[plugin.SyncerFilter]
+				var txn *loader.Txn
+				txn, err1 = translator.TiBinlogToTxn(s.schema, "", "", binlog, preWrite, false)
+				hook.Range(func(k, val interface{}) bool {
+					c, ok := val.(SyncerFilter)
+					if !ok {
+						return true
+					}
+					isFilterTransaction, err1 = c.FilterTxn(txn, s.loopbackSync)
+					if isFilterTransaction || err1 != nil {
+						return false
+					}
+					return true
+				})
+				if err1 != nil {
+					log.Warn("FilterTxn return error", zap.String("error", err1.Error()))
+					break ForLoop
+				}
+			}
+
 			if s.loopbackSync != nil && s.loopbackSync.LoopbackControl {
 				isFilterTransaction, err1 = loopBackStatus(binlog, preWrite, s.schema, s.loopbackSync)
 				if err1 != nil {
@@ -414,6 +465,36 @@ ForLoop:
 				break ForLoop
 			}
 
+			if s.loopbackSync.SupportPlugin {
+				var isFilterTransaction = false
+				var err1 error
+				txn := new(loader.Txn)
+				txn.DDL = &loader.DDL{
+					Database: schema,
+					Table:    table,
+					SQL:      string(binlog.GetDdlQuery()),
+				}
+				hook := s.loopbackSync.Hooks[plugin.SyncerFilter]
+				hook.Range(func(k, val interface{}) bool {
+					c, ok := val.(SyncerFilter)
+					if !ok {
+						return true
+					}
+					isFilterTransaction, err1 = c.FilterTxn(txn, s.loopbackSync)
+					if isFilterTransaction || err1 != nil {
+						return false
+					}
+					return true
+				})
+				if err1 != nil {
+					log.Warn("FilterTxn return error", zap.String("error", err1.Error()))
+					break ForLoop
+				}
+				if isFilterTransaction {
+					continue
+				}
+			}
+
 			if s.filter.SkipSchemaAndTable(schema, table) {
 				log.Info("skip ddl by filter", zap.String("schema", schema), zap.String("table", table),
 					zap.String("sql", sql), zap.Int64("commit ts", commitTS))

drainer/util.go

@@ -20,6 +20,7 @@ import (
 	"os"
 	"path"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/Shopify/sarama"
@@ -191,3 +192,19 @@ func genDrainerID(listenAddr string) (string, error) {
 
 	return fmt.Sprintf("%s:%s", hostname, port), nil
 }
+
+type sliceNames []string
+
+func newSliceNames(vals []string, p *[]string) *sliceNames {
+	*p = vals
+	return (*sliceNames)(p)
+}
+
+func (s *sliceNames) Set(val string) error {
+	*s = sliceNames(strings.Split(val, ","))
+	return nil
+}
+
+func (s *sliceNames) Get() interface{} { return []string(*s) }
+
+func (s *sliceNames) String() string { return strings.Join([]string(*s), ",") }