Drainer support plugin framework

drainer/config.go

@@ -78,6 +78,9 @@ 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"`
 }
 
 // RelayConfig is the Relay log's configuration.
@@ -115,6 +118,22 @@ type Config struct {
 	tls             *tls.Config
 }
 
+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), ",") }
+
 // NewConfig return an instance of configuration
 func NewConfig() *Config {
 	cfg := &Config{
@@ -157,6 +176,9 @@ func NewConfig() *Config {
 	fs.IntVar(&maxBinlogItemCount, "cache-binlog-count", defaultBinlogItemCount, "blurry count of binlogs in cache, limit cache size")
 	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")
 
 	return cfg
 }

drainer/config_test.go

@@ -55,6 +55,9 @@ func (t *testDrainerSuite) TestConfig(c *C) {
 		"-config", "../cmd/drainer/drainer.toml",
 		"-addr", "192.168.15.10:8257",
 		"-advertise-addr", "192.168.15.10:8257",
+		"-plugin-path", "/opt/drainer/plugin",
+		"-plugin-names", "p1, p2",
+		"-support-plugin",
 	}
 
 	cfg := NewConfig()
@@ -68,6 +71,9 @@ func (t *testDrainerSuite) TestConfig(c *C) {
 	var strSQLMode *string
 	c.Assert(cfg.SyncerCfg.StrSQLMode, Equals, strSQLMode)
 	c.Assert(cfg.SyncerCfg.SQLMode, Equals, mysql.SQLMode(0))
+	c.Assert(cfg.SyncerCfg.PluginPath, Equals, "/opt/drainer/plugin")
+	c.Assert(len(cfg.SyncerCfg.PluginNames), Equals, 2)
+	c.Assert(cfg.SyncerCfg.SupportPlugin, Equals, true)
 }
 
 func (t *testDrainerSuite) TestValidateFilter(c *C) {

drainer/filter_txn.go

@@ -0,0 +1,10 @@
+package drainer
+
+import (
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/pkg/loader"
+)
+
+type LoopBack 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"
 )
 
@@ -47,14 +48,24 @@ type LoopBackSync struct {
 	ChannelID       int64
 	LoopbackControl bool
 	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) *LoopBackSync {
 	l := &LoopBackSync{
 		ChannelID:       ChannelID,
 		LoopbackControl: LoopbackControl,
 		SyncDDL:         SyncDDL,
+		Index:           0,
+		PluginPath:      path,
+		PluginNames:     names,
+		SupportPlugin:   SupportPlug,
 	}
 	return l
 }

drainer/loopbacksync/loopbacksync_test.go

@@ -32,7 +32,8 @@ 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)
 
 	c.Assert(l, check.DeepEquals, &LoopBackSync{
 		ChannelID:       ChannelID,

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,7 +77,22 @@ 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)
+	if syncer.loopbackSync.SupportPlugin {
+		for _, name := range syncer.loopbackSync.PluginNames {
+			sym, err := plugin.LoadPlugin(syncer.loopbackSync.Hooks[plugin.SyncerPlugin],
+				syncer.loopbackSync.PluginPath, name)
+			if err != nil {
+				return nil, err
+			}
+			newPlugin, ok := sym.(func() LoopBack)
+			if !ok {
+				continue
+			}
+			plugin.RegisterPlugin(syncer.loopbackSync.Hooks[plugin.SyncerPlugin],
+				name, newPlugin())
+		}
+	}
 
 	var err error
 	// create schema
@@ -357,9 +372,31 @@ ForLoop:
 				err = errors.Annotate(err, "handlePreviousDDLJobIfNeed failed")
 				break ForLoop
 			}
+
 			var isFilterTransaction = false
 			var err1 error
-			if s.loopbackSync != nil && s.loopbackSync.LoopbackControl {
+
+			if s.loopbackSync.SupportPlugin {
+				hook := s.loopbackSync.Hooks[plugin.SyncerPlugin]
+				var txn *loader.Txn
+				txn, err1 = translator.TiBinlogToTxn(s.schema, "", "", binlog, preWrite, false)
+				hook.Range(func(k, val interface{}) bool {
+					c, ok := val.(LoopBack)
+					if !ok {
+						return true
+					}
+					isFilterTransaction, err1 = c.FilterTxn(txn, s.loopbackSync)
+					if isFilterTransaction || err1 != nil {
+						return false
+					}
+					return true
+				})
+				if err1 != nil {
+					break ForLoop
+				}
+			}
+
+			if s.loopbackSync != nil && s.loopbackSync.LoopbackControl && !s.loopbackSync.SupportPlugin {
 				isFilterTransaction, err1 = loopBackStatus(binlog, preWrite, s.schema, s.loopbackSync)
 				if err1 != nil {
 					err = errors.Annotate(err1, "analyze transaction failed")
@@ -414,6 +451,35 @@ 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.SyncerPlugin]
+				hook.Range(func(k, val interface{}) bool {
+					c, ok := val.(LoopBack)
+					if !ok {
+						return true
+					}
+					isFilterTransaction, err1 = c.FilterTxn(txn, s.loopbackSync)
+					if isFilterTransaction || err1 != nil {
+						return false
+					}
+					return true
+				})
+				if err1 != nil {
+					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))
@@ -422,7 +488,7 @@ ForLoop:
 
 			shouldSkip := false
 
-			if !s.cfg.SyncDDL {
+			if !s.cfg.SyncDDL && !s.loopbackSync.SupportPlugin {
 				log.Info("skip ddl by SyncDDL setting to false", zap.String("schema", schema), zap.String("table", table),
 					zap.String("sql", sql), zap.Int64("commit ts", commitTS))
 				// A empty sql force it to evict the downstream table info.