Added waitTimeBetweenConnection check

README.md

@@ -132,7 +132,7 @@ type MyEventHandler struct {
 }
 
 func (h *MyEventHandler) OnRow(e *canal.RowsEvent) error {
-log.Infof("%s %v\n", e.Action, e.Rows)
+log.Debugf("%s %v\n", e.Action, e.Rows)
 return nil
 }
 

canal/canal.go

@@ -65,6 +65,10 @@ func NewCanal(cfg *Config) (*Canal, error) {
 
 c.ctx, c.cancel = context.WithCancel(context.Background())
 
+if cfg.WaitTimeBetweenConnectionSeconds > 0 {
+cfg.WaitTimeBetweenConnectionSeconds = time.Duration(5) * time.Second
+}
+
 c.dumpDoneCh = make(chan struct{})
 c.eventHandler = &DummyEventHandler{}
 c.parser = parser.New()
@@ -192,6 +196,7 @@ func (c *Canal) RunFrom(pos mysql.Position) error {
 return c.Run()
 }
 
+// Start from selected GTIDSet
 func (c *Canal) StartFromGTID(set mysql.GTIDSet) error {
 c.master.UpdateGTIDSet(set)
 
@@ -238,7 +243,7 @@ func (c *Canal) run() error {
 }
 
 func (c *Canal) Close() {
-log.Infof("closing canal")
+log.Debugf("closing canal")
 c.m.Lock()
 defer c.m.Unlock()
 
@@ -413,20 +418,21 @@ func (c *Canal) checkBinlogRowFormat() error {
 
 func (c *Canal) prepareSyncer() error {
 cfg := replication.BinlogSyncerConfig{
-ServerID: c.cfg.ServerID,
-Flavor: c.cfg.Flavor,
-User: c.cfg.User,
-Password: c.cfg.Password,
-Charset: c.cfg.Charset,
-HeartbeatPeriod: c.cfg.HeartbeatPeriod,
-ReadTimeout: c.cfg.ReadTimeout,
-UseDecimal: c.cfg.UseDecimal,
-ParseTime: c.cfg.ParseTime,
-SemiSyncEnabled: c.cfg.SemiSyncEnabled,
-MaxReconnectAttempts: c.cfg.MaxReconnectAttempts,
-DisableRetrySync: c.cfg.DisableRetrySync,
-TimestampStringLocation: c.cfg.TimestampStringLocation,
-TLSConfig: c.cfg.TLSConfig,
+ServerID: c.cfg.ServerID,
+Flavor: c.cfg.Flavor,
+User: c.cfg.User,
+Password: c.cfg.Password,
+Charset: c.cfg.Charset,
+HeartbeatPeriod: c.cfg.HeartbeatPeriod,
+ReadTimeout: c.cfg.ReadTimeout,
+UseDecimal: c.cfg.UseDecimal,
+ParseTime: c.cfg.ParseTime,
+SemiSyncEnabled: c.cfg.SemiSyncEnabled,
+MaxReconnectAttempts: c.cfg.MaxReconnectAttempts,
+DisableRetrySync: c.cfg.DisableRetrySync,
+TimestampStringLocation: c.cfg.TimestampStringLocation,
+TLSConfig: c.cfg.TLSConfig,
+WaitTimeBetweenConnectionSeconds: c.cfg.WaitTimeBetweenConnectionSeconds,
 }
 
 if strings.Contains(c.cfg.Addr, "/") {

canal/canal_test.go

@@ -111,7 +111,7 @@ type testEventHandler struct {
 }
 
 func (h *testEventHandler) OnRow(e *RowsEvent) error {
-log.Infof("OnRow %s %v\n", e.Action, e.Rows)
+log.Debugf("OnRow %s %v\n", e.Action, e.Rows)
 umi, ok := e.Rows[0][4].(uint32) // 4th col is umi. mysqldump gives uint64 instead of uint32
 if ok && (umi != umiA && umi != umiB && umi != umiC) {
 return fmt.Errorf("invalid unsigned medium int %d", umi)

canal/config.go

@@ -81,6 +81,9 @@ type Config struct {
 // this configuration will not work if DisableRetrySync is true
 MaxReconnectAttempts int `toml:"max_reconnect_attempts"`
 
+// WaitTimeBetweenConnectionSeconds is the time to wait before retrying to connect.
+WaitTimeBetweenConnectionSeconds time.Duration `toml:"max_wait_time_between_connection_seconds"`
+
 // whether disable re-sync for broken connection
 DisableRetrySync bool `toml:"disable_retry_sync"`
 

canal/dump.go

@@ -163,13 +163,13 @@ func (c *Canal) dump() error {
 if err != nil {
 return errors.Trace(err)
 }
-log.Infof("skip master data, get current binlog position %v", pos)
+log.Debugf("skip master data, get current binlog position %v", pos)
 h.name = pos.Name
 h.pos = uint64(pos.Pos)
 }
 
 start := time.Now()
-log.Info("try dump MySQL and parse")
+log.Debug("try dump MySQL and parse")
 if err := c.dumper.DumpAndParse(h); err != nil {
 return errors.Trace(err)
 }
@@ -185,7 +185,7 @@ func (c *Canal) dump() error {
 c.master.UpdateGTIDSet(h.gset)
 startPos = h.gset
 }
-log.Infof("dump MySQL and parse OK, use %0.2f seconds, start binlog replication at %s",
+log.Debugf("dump MySQL and parse OK, use %0.2f seconds, start binlog replication at %s",
 time.Now().Sub(start).Seconds(), startPos)
 return nil
 }
@@ -196,12 +196,12 @@ func (c *Canal) tryDump() error {
 if (len(pos.Name) > 0 && pos.Pos > 0) ||
 (gset != nil && gset.String() != "") {
 // we will sync with binlog name and position
-log.Infof("skip dump, use last binlog replication pos %s or GTID set %v", pos, gset)
+log.Debugf("skip dump, use last binlog replication pos %s or GTID set %v", pos, gset)
 return nil
 }
 
 if c.dumper == nil {
-log.Info("skip dump, no mysqldump")
+log.Debug("skip dump, no mysqldump")
 return nil
 }
 

canal/sync.go

@@ -22,15 +22,15 @@ func (c *Canal) startSyncer() (*replication.BinlogStreamer, error) {
 if err != nil {
 return nil, errors.Errorf("start sync replication at binlog %v error %v", pos, err)
 }
-log.Infof("start sync binlog at binlog file %v", pos)
+log.Debugf("start sync binlog at binlog file %v", pos)
 return s, nil
 } else {
 gsetClone := gset.Clone()
 s, err := c.syncer.StartSyncGTID(gset)
 if err != nil {
 return nil, errors.Errorf("start sync replication at GTID set %v error %v", gset, err)
 }
-log.Infof("start sync binlog at GTID set %v", gsetClone)
+log.Debugf("start sync binlog at GTID set %v", gsetClone)
 return s, nil
 }
 }
@@ -65,7 +65,7 @@ func (c *Canal) runSyncBinlog() error {
 switch e := ev.Event.(type) {
 case *replication.RotateEvent:
 fakeRotateLogName = string(e.NextLogName)
-log.Infof("received fake rotate event, next log name is %s", e.NextLogName)
+log.Debugf("received fake rotate event, next log name is %s", e.NextLogName)
 }
 
 continue
@@ -92,7 +92,7 @@ func (c *Canal) runSyncBinlog() error {
 case *replication.RotateEvent:
 pos.Name = string(e.NextLogName)
 pos.Pos = uint32(e.Position)
-log.Infof("rotate binlog to %s", pos)
+log.Debugf("rotate binlog to %s", pos)
 savePos = true
 force = true
 if err = c.eventHandler.OnRotate(e); err != nil {
@@ -232,7 +232,7 @@ func parseStmt(stmt ast.StmtNode) (ns []*node) {
 
 func (c *Canal) updateTable(db, table string) (err error) {
 c.ClearTableCache([]byte(db), []byte(table))
-log.Infof("table structure changed, clear table cache: %s.%s\n", db, table)
+log.Debugf("table structure changed, clear table cache: %s.%s\n", db, table)
 if err = c.eventHandler.OnTableChanged(db, table); err != nil && errors.Cause(err) != schema.ErrTableNotExist {
 return errors.Trace(err)
 }

dump/dumper.go

@@ -241,7 +241,7 @@ func (d *Dumper) Dump(w io.Writer) error {
 }
 
 args[passwordArgIndex] = "--password=******"
-log.Infof("exec mysqldump with %v", args)
+log.Debugf("exec mysqldump with %v", args)
 args[passwordArgIndex] = passwordArg
 cmd := exec.Command(d.ExecutionPath, args...)
 

replication/binlogsyncer.go

@@ -28,6 +28,9 @@ type BinlogSyncerConfig struct {
 // Flavor is "mysql" or "mariadb", if not set, use "mysql" default.
 Flavor string
 
+// MaxWaitTimeBetween Check connection
+WaitTimeBetweenConnectionSeconds time.Duration
+
 // Host is for MySQL server host.
 Host string
 // Port is for MySQL server port.
@@ -145,7 +148,7 @@ func NewBinlogSyncer(cfg BinlogSyncerConfig) *BinlogSyncer {
 // Clear the Password to avoid outputing it in log.
 pass := cfg.Password
 cfg.Password = ""
-log.Infof("create BinlogSyncer with config %v", cfg)
+log.Debugf("create BinlogSyncer with config %v", cfg)
 cfg.Password = pass
 
 b := new(BinlogSyncer)
@@ -177,7 +180,7 @@ func (b *BinlogSyncer) close() {
 return
 }
 
-log.Info("syncer is closing...")
+log.Debug("syncer is closing...")
 
 b.running = false
 b.cancel()
@@ -207,7 +210,7 @@ func (b *BinlogSyncer) close() {
 b.c.Close()
 }
 
-log.Info("syncer is closed")
+log.Debug("syncer is closed")
 }
 
 func (b *BinlogSyncer) isClosed() bool {
@@ -371,7 +374,7 @@ func (b *BinlogSyncer) GetNextPosition() Position {
 
 // StartSync starts syncing from the `pos` position.
 func (b *BinlogSyncer) StartSync(pos Position) (*BinlogStreamer, error) {
-log.Infof("begin to sync binlog from position %s", pos)
+log.Debugf("begin to sync binlog from position %s", pos)
 
 b.m.Lock()
 defer b.m.Unlock()
@@ -389,7 +392,7 @@ func (b *BinlogSyncer) StartSync(pos Position) (*BinlogStreamer, error) {
 
 // StartSyncGTID starts syncing from the `gset` GTIDSet.
 func (b *BinlogSyncer) StartSyncGTID(gset GTIDSet) (*BinlogStreamer, error) {
-log.Infof("begin to sync binlog from GTID set %s", gset)
+log.Debugf("begin to sync binlog from GTID set %s", gset)
 
 b.prevGset = gset
 
@@ -593,13 +596,13 @@ func (b *BinlogSyncer) retrySync() error {
 if b.currGset != nil {
 msg = fmt.Sprintf("%v (last read GTID=%v)", msg, b.currGset)
 }
-log.Infof(msg)
+log.Debugf(msg)
 
 if err := b.prepareSyncGTID(b.prevGset); err != nil {
 return errors.Trace(err)
 }
 } else {
-log.Infof("begin to re-sync from %s", b.nextPos)
+log.Debugf("begin to re-sync from %s", b.nextPos)
 if err := b.prepareSyncPos(b.nextPos); err != nil {
 return errors.Trace(err)
 }
@@ -684,6 +687,7 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {
 }
 
 for {
+// Read packet until CTX is done if not will continue after the loop
 select {
 case <-b.ctx.Done():
 s.close()
@@ -697,7 +701,8 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {
 return
 }
 
-log.Errorf("retry sync err: %v, wait 1s and retry again", err)
+log.Errorf("retry sync err: %v, wait %s s and retry again", err, b.cfg.WaitTimeBetweenConnectionSeconds)
+time.Sleep(b.cfg.WaitTimeBetweenConnectionSeconds)
 continue
 }
 }
@@ -731,7 +736,7 @@ func (b *BinlogSyncer) onStream(s *BinlogStreamer) {
 // refer to https://dev.mysql.com/doc/internals/en/com-binlog-dump.html#binlog-dump-non-block
 // when COM_BINLOG_DUMP command use BINLOG_DUMP_NON_BLOCK flag,
 // if there is no more event to send an EOF_Packet instead of blocking the connection
-log.Info("receive EOF packet, no more binlog event now.")
+log.Debug("receive EOF packet, no more binlog event now.")
 continue
 default:
 log.Errorf("invalid stream header %c", data[0])
@@ -787,7 +792,7 @@ func (b *BinlogSyncer) parseEvent(s *BinlogStreamer, data []byte) error {
 case *RotateEvent:
 b.nextPos.Name = string(event.NextLogName)
 b.nextPos.Pos = uint32(event.Position)
-log.Infof("rotate to %s", b.nextPos)
+log.Debugf("rotate to %s", b.nextPos)
 case *GTIDEvent:
 if b.prevGset == nil {
 break
@@ -860,5 +865,5 @@ func (b *BinlogSyncer) killConnection(conn *client.Conn, id uint32) {
 log.Error(errors.Trace(err))
 }
 }
-log.Infof("kill last connection id %d", id)
+log.Debugf("kill last connection id %d", id)
 }