Drainer support plugin framework

drainer/checkpoint/checkpoint.go

@@ -54,6 +54,12 @@ func NewCheckPoint(cfg *Config) (CheckPoint, error) {
 		cp, err = newMysql(cfg)
 	case "file":
 		cp, err = NewFile(cfg.InitialCommitTS, cfg.CheckPointFile)
+	case "plugin":
+		if cfg.Db != nil {
+			cp, err = newMysql(cfg)
+		} else {
+			cp, err = NewFile(cfg.InitialCommitTS, cfg.CheckPointFile)
+		}
 	default:
 		err = errors.Errorf("unsupported checkpoint type %s", cfg.CheckpointType)
 	}

drainer/config.go

@@ -85,10 +85,13 @@ type SyncerConfig struct {
 	// for backward compatibility.
 	// disable* is keep for backward compatibility.
 	// if both setted, the disable one take affect.
-	DisableCausalityFlag *bool `toml:"-" json:"disable-detect-flag"`
-	EnableCausalityFlag  *bool `toml:"-" json:"enable-detect-flag"`
-	DisableCausalityFile *bool `toml:"disable-detect" json:"disable-detect"`
-	EnableCausalityFile  *bool `toml:"enable-detect" json:"enable-detect"`
+	DisableCausalityFlag *bool  `toml:"-" json:"disable-detect-flag"`
+	EnableCausalityFlag  *bool  `toml:"-" json:"enable-detect-flag"`
+	DisableCausalityFile *bool  `toml:"disable-detect" json:"disable-detect"`
+	EnableCausalityFile  *bool  `toml:"enable-detect" json:"enable-detect"`
+	PluginPath           string `toml:"plugin-path" json:"plugin-path"`
+	PluginName           string `toml:"plugin-name" json:"plugin-name"`
+	PluginCfgFile        string `toml:"plugin-cfg-file" json:"plugin-cfg-file"`
 }
 
 // EnableDispatch return true if enable dispatch.
@@ -216,6 +219,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.PluginName, "plugin-name", "", "syncer plugin name")
+	fs.StringVar(&cfg.SyncerCfg.PluginPath, "plugin-path", "", "syncer plugin path")
+	fs.StringVar(&cfg.SyncerCfg.PluginCfgFile, "plugin-cfg-path", "", "syncer plugin's config file")
 
 	return cfg
 }

drainer/sync/plugin_demo.go

@@ -0,0 +1,56 @@
+package sync
+
+import (
+	"fmt"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/drainer/relay"
+	"github.com/pingcap/tidb-binlog/drainer/translator"
+	"github.com/prometheus/client_golang/prometheus"
+	"go.uber.org/zap"
+)
+
+//SyncerDemo is a syncer demo
+type SyncerDemo struct {
+	*baseSyncer
+}
+
+//Sync is the method that interface must implement
+func (sd *SyncerDemo) Sync(item *Item) error {
+	//demo
+	log.Info("item", zap.String("%s", fmt.Sprintf("%v", item)))
+	sd.success <- item
+	return nil
+}
+
+//Close is the method that interface must implement
+func (sd *SyncerDemo) Close() error {
+	return nil
+}
+
+//SetSafeMode is the method that interface must implement
+func (sd *SyncerDemo) SetSafeMode(mode bool) bool {
+	return false
+}
+
+//NewSyncerDemo is a syncer demo
+func NewSyncerDemo(
+	cfg *DBConfig,
+	file string,
+	tableInfoGetter translator.TableInfoGetter,
+	worker int,
+	batchSize int,
+	queryHistogramVec *prometheus.HistogramVec,
+	sqlMode *string,
+	destDBType string,
+	relayer relay.Relayer,
+	info *loopbacksync.LoopBackSync,
+	enableDispatch bool,
+	enableCausility bool,
+) (Syncer, error) {
+	log.Info("call NewSyncerDemo()")
+	executor := &SyncerDemo{}
+	executor.baseSyncer = newBaseSyncer(tableInfoGetter)
+	return executor, nil
+}

drainer/syncer.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/drainer/syncplg"
 	"github.com/pingcap/tidb-binlog/pkg/loader"
 
 	"github.com/pingcap/errors"
@@ -120,6 +121,26 @@ func createDSyncer(cfg *SyncerConfig, schema *Schema, info *loopbacksync.LoopBac
 		// only use for test
 	case "_intercept":
 		dsyncer = newInterceptSyncer()
+	case "plugin":
+		if len(cfg.PluginName) == 0 || len(cfg.PluginPath) == 0 {
+			return nil, errors.Errorf("plugin-name or plugin-path is incorrect")
+		}
+		newSyncer, err := syncplg.LoadPlugin(cfg.PluginPath, cfg.PluginName)
+		if err != nil {
+			return nil, errors.Annotate(err, "fail to load plugin dsyncer")
+		}
+
+		var relayer relay.Relayer
+		if cfg.Relay.IsEnabled() {
+			if relayer, err = relay.NewRelayer(cfg.Relay.LogDir, cfg.Relay.MaxFileSize, schema); err != nil {
+				return nil, errors.Annotate(err, "fail to create relayer")
+			}
+		}
+
+		dsyncer, err = newSyncer(cfg.To, cfg.PluginCfgFile, schema, cfg.WorkerCount, cfg.TxnBatch, queryHistogramVec, cfg.StrSQLMode, cfg.DestDBType, relayer, info, cfg.EnableDispatch(), cfg.EnableCausality())
+		if err != nil {
+			return nil, errors.Annotate(err, "fail to create plugin dsyncer")
+		}
 	default:
 		return nil, errors.Errorf("unknown DestDBType: %s", cfg.DestDBType)
 	}

drainer/syncplg/plugin.go

@@ -0,0 +1,76 @@
+package syncplg
+
+import (
+	"errors"
+	"fmt"
+	"plugin"
+
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/drainer/relay"
+	"github.com/pingcap/tidb-binlog/drainer/sync"
+	"github.com/pingcap/tidb-binlog/drainer/translator"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	//NewPlugin is the name of exported function by syncer plugin
+	NewPlugin = "NewPluginFactory"
+)
+
+//FactoryInterface is interface of Factory
+type FactoryInterface interface {
+	NewSyncerPlugin(
+		cfg *sync.DBConfig,
+		file string,
+		tableInfoGetter translator.TableInfoGetter,
+		worker int,
+		batchSize int,
+		queryHistogramVec *prometheus.HistogramVec,
+		sqlMode *string,
+		destDBType string,
+		relayer relay.Relayer,
+		info *loopbacksync.LoopBackSync,
+		enableDispatch bool,
+		enableCausility bool,
+	) (sync.Syncer, error)
+}
+
+//NewSyncerFunc is a function type which syncer plugin must implement
+type NewSyncerFunc func(
+	cfg *sync.DBConfig,
+	file string,
+	tableInfoGetter translator.TableInfoGetter,
+	worker int,
+	batchSize int,
+	queryHistogramVec *prometheus.HistogramVec,
+	sqlMode *string,
+	destDBType string,
+	relayer relay.Relayer,
+	info *loopbacksync.LoopBackSync,
+	enableDispatch bool,
+	enableCausility bool,
+) (sync.Syncer, error)
+
+//LoadPlugin load syncer plugin
+func LoadPlugin(path, name string) (NewSyncerFunc, error) {
+	fp := path + "/" + name
+	p, err := plugin.Open(fp)
+	if err != nil {
+		return nil, fmt.Errorf("faile to Open %s . err: %s", fp, err.Error())
+	}
+
+	sym, err := p.Lookup(NewPlugin)
+	if err != nil {
+		return nil, err
+	}
+	newFactory, ok := sym.(func() interface{})
+	if !ok {
+		return nil, errors.New("function type is incorrect")
+	}
+	fac := newFactory()
+	plg, ok := fac.(FactoryInterface)
+	if !ok {
+		return nil, errors.New("not implement FactoryInterface")
+	}
+	return plg.NewSyncerPlugin, nil
+}

drainer/syncplg/syncerdemo/Makefile

@@ -0,0 +1,4 @@
+plugin:
+	go build -o syncerdemo.so -buildmode=plugin syncerdemo.go
+clean:
+	rm -rf syncerdemo.so

drainer/syncplg/syncerdemo/syncerdemo.go

@@ -0,0 +1,41 @@
+package main
+
+import (
+	"github.com/pingcap/log"
+	"github.com/pingcap/tidb-binlog/drainer/loopbacksync"
+	"github.com/pingcap/tidb-binlog/drainer/relay"
+	"github.com/pingcap/tidb-binlog/drainer/sync"
+	"github.com/pingcap/tidb-binlog/drainer/translator"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+//PluginFactory is the Factory struct
+type PluginFactory struct{}
+
+//NewPluginFactory is factory function of plugin
+func NewPluginFactory() interface{} {
+	log.Info("call NewPluginFactory")
+	return PluginFactory{}
+}
+
+//NewSyncerPlugin return A syncer instance which implemented interface of sync.Syncer
+func (pf PluginFactory) NewSyncerPlugin(
+	cfg *sync.DBConfig,
+	file string,
+	tableInfoGetter translator.TableInfoGetter,
+	worker int,
+	batchSize int,
+	queryHistogramVec *prometheus.HistogramVec,
+	sqlMode *string,
+	destDBType string,
+	relayer relay.Relayer,
+	info *loopbacksync.LoopBackSync,
+	enableDispatch bool,
+	enableCausility bool,
+) (sync.Syncer, error) {
+	return sync.NewSyncerDemo(cfg, file, tableInfoGetter, worker, batchSize, queryHistogramVec, sqlMode,
+		destDBType, relayer, info, enableDispatch, enableCausility)
+}
+
+var _ PluginFactory
+var _ = NewPluginFactory()

drainer/util.go

@@ -106,7 +106,7 @@ func GenCheckPointCfg(cfg *Config, id uint64) (*checkpoint.Config, error) {
 		}
 	case "":
 		switch cfg.SyncerCfg.DestDBType {
-		case "mysql", "tidb":
+		case "mysql", "tidb", "plugin":
 			checkpointCfg.CheckpointType = cfg.SyncerCfg.DestDBType
 			checkpointCfg.Db = &checkpoint.DBConfig{
 				Host:     cfg.SyncerCfg.To.Host,