feat: Charset parameter for Mysql connection

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,15 +243,17 @@ func (c *Canal) run() error {
 }
 
 func (c *Canal) Close() {
-	log.Infof("closing canal")
+	log.Debugf("closing canal")
 	c.m.Lock()
 	defer c.m.Unlock()
 
 	c.cancel()
 	c.syncer.Close()
 	c.connLock.Lock()
-	c.conn.Close()
-	c.conn = nil
+	if c.conn != nil {
+		c.conn.Close()
+		c.conn = nil
+	}
 	c.connLock.Unlock()
 
 	_ = c.eventHandler.OnPosSynced(c.master.Position(), c.master.GTIDSet(), true)
@@ -413,20 +420,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)
@@ -141,10 +141,48 @@ func (s *canalTestSuite) TestCanal(c *C) {
 	s.execute(c, "ALTER TABLE test.canal_test ADD `age` INT(5) NOT NULL AFTER `name`")
 	s.execute(c, "INSERT INTO test.canal_test (name,age) VALUES (?,?)", "d", "18")
 
-	err := s.c.CatchMasterPos(10 * time.Second)
+	err := CatchMasterPos(s.c, 10*time.Second)
 	c.Assert(err, IsNil)
 }
 
+func CatchMasterPos(c *Canal, timeout time.Duration) error {
+	pos, err := c.GetMasterPos()
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	return WaitUntilPos(c, pos, timeout)
+}
+
+func FlushBinlog(c *Canal) error {
+	_, err := c.Execute("FLUSH BINARY LOGS")
+	return errors.Trace(err)
+}
+
+func WaitUntilPos(c *Canal, pos mysql.Position, timeout time.Duration) error {
+	timer := time.NewTimer(timeout)
+	for {
+		select {
+		case <-timer.C:
+			return errors.Errorf("wait position %v too long > %s", pos, timeout)
+		default:
+			err := FlushBinlog(c)
+			if err != nil {
+				return errors.Trace(err)
+			}
+			curPos := c.master.Position()
+			if curPos.Compare(pos) >= 0 {
+				return nil
+			} else {
+				log.Debugf("master pos is %v, wait catching %v", curPos, pos)
+				time.Sleep(100 * time.Millisecond)
+			}
+		}
+	}
+
+	return nil
+}
+
 func (s *canalTestSuite) TestCanalFilter(c *C) {
 	// included
 	sch, err := s.c.GetTable("test", "canal_test")

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

@@ -2,6 +2,7 @@ package canal
 
 import (
 	"fmt"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -22,15 +23,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 +66,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 +93,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 {
@@ -142,7 +143,15 @@ func (c *Canal) runSyncBinlog() error {
 		case *replication.QueryEvent:
 			stmts, _, err := c.parser.Parse(string(e.Query), "", "")
 			if err != nil {
-				log.Errorf("parse query(%s) err %v, will skip this event", e.Query, err)
+				msg := err.Error()
+				if strings.Contains(strings.ToLower(msg), strings.ToLower("procedure")) {
+					// Cut the first row from the message since it contain the procedure call and not the entire message
+					fl := strings.Split(msg, "\n")
+					log.Debugf("parse SP Error: (%s)", fl[0])
+				} else {
+					log.Debugf("parse query(%s) err %v", e.Query, err)
+				}
+				log.Debugln("will skip this event")
 				continue
 			}
 			for _, stmt := range stmts {
@@ -232,7 +241,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)
 	}
@@ -270,38 +279,10 @@ func (c *Canal) handleRowsEvent(e *replication.BinlogEvent) error {
 		return errors.Errorf("%s not supported now", e.Header.EventType)
 	}
 	events := newRowsEvent(t, action, ev.Rows, e.Header)
+	events.Header.Gtid = c.SyncedGTIDSet()
 	return c.eventHandler.OnRow(events)
 }
 
-func (c *Canal) FlushBinlog() error {
-	_, err := c.Execute("FLUSH BINARY LOGS")
-	return errors.Trace(err)
-}
-
-func (c *Canal) WaitUntilPos(pos mysql.Position, timeout time.Duration) error {
-	timer := time.NewTimer(timeout)
-	for {
-		select {
-		case <-timer.C:
-			return errors.Errorf("wait position %v too long > %s", pos, timeout)
-		default:
-			err := c.FlushBinlog()
-			if err != nil {
-				return errors.Trace(err)
-			}
-			curPos := c.master.Position()
-			if curPos.Compare(pos) >= 0 {
-				return nil
-			} else {
-				log.Debugf("master pos is %v, wait catching %v", curPos, pos)
-				time.Sleep(100 * time.Millisecond)
-			}
-		}
-	}
-
-	return nil
-}
-
 func (c *Canal) GetMasterPos() (mysql.Position, error) {
 	rr, err := c.Execute("SHOW MASTER STATUS")
 	if err != nil {
@@ -336,12 +317,3 @@ func (c *Canal) GetMasterGTIDSet() (mysql.GTIDSet, error) {
 	}
 	return gset, nil
 }
-
-func (c *Canal) CatchMasterPos(timeout time.Duration) error {
-	pos, err := c.GetMasterPos()
-	if err != nil {
-		return errors.Trace(err)
-	}
-
-	return c.WaitUntilPos(pos, timeout)
-}

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...)