update canal to support batch later. (#93)

Author: siddontang

.gitignore

@@ -1,2 +1 @@
-var
 bin

cmd/go-mysql-elasticsearch/main.go

@@ -8,6 +8,7 @@ import (
 	"syscall"
 
 	"github.com/juju/errors"
+	"github.com/ngaut/log"
 	"github.com/siddontang/go-mysql-elasticsearch/river"
 )
 
@@ -20,11 +21,14 @@ var data_dir = flag.String("data_dir", "", "path for go-mysql-elasticsearch to s
 var server_id = flag.Int("server_id", 0, "MySQL server id, as a pseudo slave")
 var flavor = flag.String("flavor", "", "flavor: mysql or mariadb")
 var execution = flag.String("exec", "", "mysqldump execution path")
+var logLevel = flag.String("log_level", "info", "log level")
 
 func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
 	flag.Parse()
 
+	log.SetLevelByString(*logLevel)
+
 	sc := make(chan os.Signal, 1)
 	signal.Notify(sc,
 		os.Kill,

etc/river.toml

@@ -7,8 +7,10 @@ my_pass = ""
 # Elasticsearch address
 es_addr = "127.0.0.1:9200"
 
-# Path to store data, like master.info, and dump MySQL data 
-data_dir = "./var"
+# Path to store data, like master.info, if not set or empty,
+# we must use this to support breakpoint resume syncing. 
+# TODO: support other storage, like etcd. 
+data_dir = ""
 
 # Inner Http status address
 stat_addr = "127.0.0.1:12800"

glide.lock

@@ -7,7 +7,7 @@ import:
 - package: github.com/satori/go.uuid
   version: ^1.1.0
 - package: github.com/siddontang/go-mysql
-  version: 3ca161ffa3b5844340e534d4b3cdc03b60f32610
+  version: 1a46a5f32930348fc4f6050692c5ec571e53e8f1
   subpackages:
   - canal
   - client

glide.yaml

@@ -0,0 +1,99 @@
+package river
+
+import (
+	"bytes"
+	"os"
+	"path"
+	"sync"
+	"time"
+
+	"github.com/BurntSushi/toml"
+	"github.com/juju/errors"
+	"github.com/ngaut/log"
+	"github.com/siddontang/go-mysql/mysql"
+	"github.com/siddontang/go/ioutil2"
+)
+
+type masterInfo struct {
+	sync.RWMutex
+
+	Name string `toml:"bin_name"`
+	Pos  uint32 `toml:"bin_pos"`
+
+	filePath     string
+	lastSaveTime time.Time
+}
+
+func loadMasterInfo(dataDir string) (*masterInfo, error) {
+	var m masterInfo
+
+	if len(dataDir) == 0 {
+		return &m, nil
+	}
+
+	m.filePath = path.Join(dataDir, "master.info")
+	m.lastSaveTime = time.Now()
+
+	if err := os.MkdirAll(dataDir, 0755); err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	f, err := os.Open(m.filePath)
+	if err != nil && !os.IsNotExist(errors.Cause(err)) {
+		return nil, errors.Trace(err)
+	} else if os.IsNotExist(errors.Cause(err)) {
+		return &m, nil
+	}
+	defer f.Close()
+
+	_, err = toml.DecodeReader(f, &m)
+	return &m, errors.Trace(err)
+}
+
+func (m *masterInfo) Save(pos mysql.Position) error {
+	log.Infof("save position %s", pos)
+
+	m.Lock()
+	defer m.Unlock()
+
+	m.Name = pos.Name
+	m.Pos = pos.Pos
+
+	if len(m.filePath) == 0 {
+		return nil
+	}
+
+	n := time.Now()
+	if n.Sub(m.lastSaveTime) < time.Second {
+		return nil
+	}
+
+	m.lastSaveTime = n
+	var buf bytes.Buffer
+	e := toml.NewEncoder(&buf)
+
+	e.Encode(m)
+
+	var err error
+	if err = ioutil2.WriteFileAtomic(m.filePath, buf.Bytes(), 0644); err != nil {
+		log.Errorf("canal save master info to file %s err %v", m.filePath, err)
+	}
+
+	return errors.Trace(err)
+}
+
+func (m *masterInfo) Position() mysql.Position {
+	m.RLock()
+	defer m.RUnlock()
+
+	return mysql.Position{
+		m.Name,
+		m.Pos,
+	}
+}
+
+func (m *masterInfo) Close() error {
+	pos := m.Position()
+
+	return m.Save(pos)
+}

river/master.go

@@ -28,6 +28,8 @@ type River struct {
 	es *elastic.Client
 
 	st *stat
+
+	master *masterInfo
 }
 
 func NewRiver(c *Config) (*River, error) {
@@ -40,6 +42,10 @@ func NewRiver(c *Config) (*River, error) {
 	r.rules = make(map[string]*Rule)
 
 	var err error
+	if r.master, err = loadMasterInfo(c.DataDir); err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	if err = r.newCanal(); err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -71,7 +77,6 @@ func (r *River) newCanal() error {
 	cfg.User = r.c.MyUser
 	cfg.Password = r.c.MyPassword
 	cfg.Flavor = r.c.Flavor
-	cfg.DataDir = r.c.DataDir
 
 	cfg.ServerID = r.c.ServerID
 	cfg.Dump.ExecutionPath = r.c.DumpExec
@@ -105,7 +110,7 @@ func (r *River) prepareCanal() error {
 		r.canal.AddDumpDatabases(keys...)
 	}
 
-	r.canal.RegRowsEventHandler(&rowsEventHandler{r})
+	r.canal.SetEventHandler(&eventHandler{r})
 
 	return nil
 }
@@ -235,7 +240,8 @@ func ruleKey(schema string, table string) string {
 }
 
 func (r *River) Run() error {
-	if err := r.canal.Start(); err != nil {
+	pos := r.master.Position()
+	if err := r.canal.StartFrom(pos); err != nil {
 		log.Errorf("start canal err %v", err)
 		return errors.Trace(err)
 	}
@@ -249,5 +255,7 @@ func (r *River) Close() {
 
 	r.canal.Close()
 
+	r.master.Close()
+
 	r.wg.Wait()
 }

river/river.go

@@ -10,7 +10,10 @@ import (
 	"github.com/ngaut/log"
 	"github.com/siddontang/go-mysql-elasticsearch/elastic"
 	"github.com/siddontang/go-mysql/canal"
+	"github.com/siddontang/go-mysql/mysql"
+	"github.com/siddontang/go-mysql/replication"
 	"github.com/siddontang/go-mysql/schema"
+	"golang.org/x/net/context"
 )
 
 const (
@@ -23,11 +26,28 @@ const (
 	fieldTypeList = "list"
 )
 
-type rowsEventHandler struct {
+type eventHandler struct {
 	r *River
 }
 
-func (h *rowsEventHandler) Do(e *canal.RowsEvent) error {
+func (h *eventHandler) OnRotate(_ context.Context, e *replication.RotateEvent) error {
+	pos := mysql.Position{
+		string(e.NextLogName),
+		uint32(e.Position),
+	}
+
+	return h.r.master.Save(pos)
+}
+
+func (h *eventHandler) OnDDL(_ context.Context, nextPos mysql.Position, _ *replication.QueryEvent) error {
+	return h.r.master.Save(nextPos)
+}
+
+func (h *eventHandler) OnXID(_ context.Context, nextPos mysql.Position) error {
+	return h.r.master.Save(nextPos)
+}
+
+func (h *eventHandler) OnRow(_ context.Context, e *canal.RowsEvent) error {
 	rule, ok := h.r.rules[ruleKey(e.Table.Schema, e.Table.Name)]
 	if !ok {
 		return nil
@@ -50,16 +70,11 @@ func (h *rowsEventHandler) Do(e *canal.RowsEvent) error {
 		return errors.Errorf("make %s ES request err %v", e.Action, err)
 	}
 
-	if err := h.r.doBulk(reqs); err != nil {
-		log.Errorf("do ES bulks err %v, stop", err)
-		return canal.ErrHandleInterrupted
-	}
-
-	return nil
+	return h.r.doBulk(reqs)
 }
 
-func (h *rowsEventHandler) String() string {
-	return "ESRiverRowsEventHandler"
+func (h *eventHandler) String() string {
+	return "ESRiverEventHandler"
 }
 
 // for insert and delete