canal: support ipv6 address (#943)

* feat(canal): support ipv6 address

* feat: use net.SplitHostPort to parse ip with ports

* chore: optimze validate error message

* feat(ci): complete test cases

* feat(dumper): use net.SplitHostPort to parse address

Author: sinomoe

.github/workflows/ci.yml

@@ -17,6 +17,11 @@ jobs:
           echo -n "mysqldump -V: " ; mysqldump -V
 
           echo -e '[mysqld]\nserver-id=1\nlog-bin=mysql\nbinlog-format=row\ngtid-mode=ON\nenforce_gtid_consistency=ON\n' | sudo tee /etc/mysql/conf.d/replication.cnf
+          
+          # bind to :: for dual-stack listening
+          sudo sed -i 's/bind-address.*= 127.0.0.1/bind-address = ::/' /etc/mysql/mysql.conf.d/mysqld.cnf
+          sudo sed -i 's/mysqlx-bind-address.*= 127.0.0.1/mysqlx-bind-address = ::/' /etc/mysql/mysql.conf.d/mysqld.cnf
+          
           sudo service mysql start
 
           # apply this for mysql5 & mysql8 compatibility
@@ -109,5 +114,6 @@ jobs:
         uses: actions/setup-go@v5
         with:
           go-version: "1.22"
+          
       - name: Build on ${{ matrix.os }}/${{ matrix.arch }}
         run: GOARCH=${{ matrix.arch }} GOOS=${{ matrix.os }} go build ./...

canal/canal.go

@@ -481,18 +481,17 @@ func (c *Canal) prepareSyncer() error {
 	if strings.Contains(c.cfg.Addr, "/") {
 		cfg.Host = c.cfg.Addr
 	} else {
-		seps := strings.Split(c.cfg.Addr, ":")
-		if len(seps) != 2 {
-			return errors.Errorf("invalid mysql addr format %s, must host:port", c.cfg.Addr)
+		host, port, err := net.SplitHostPort(c.cfg.Addr)
+		if err != nil {
+			return errors.Errorf("invalid MySQL address format %s, must host:port", c.cfg.Addr)
 		}
-
-		port, err := strconv.ParseUint(seps[1], 10, 16)
+		portNumber, err := strconv.ParseUint(port, 10, 16)
 		if err != nil {
 			return errors.Trace(err)
 		}
 
-		cfg.Host = seps[0]
-		cfg.Port = uint16(port)
+		cfg.Host = host
+		cfg.Port = uint16(portNumber)
 	}
 
 	c.syncer = replication.NewBinlogSyncer(cfg)

canal/canal_test.go

@@ -16,12 +16,30 @@ import (
 )
 
 type canalTestSuite struct {
+	addr string
 	suite.Suite
 	c *Canal
 }
 
+type canalTestSuiteOption func(c *canalTestSuite)
+
+func withAddr(addr string) canalTestSuiteOption {
+	return func(c *canalTestSuite) {
+		c.addr = addr
+	}
+}
+
+func newCanalTestSuite(opts ...canalTestSuiteOption) *canalTestSuite {
+	c := new(canalTestSuite)
+	for _, opt := range opts {
+		opt(c)
+	}
+	return c
+}
+
 func TestCanalSuite(t *testing.T) {
-	suite.Run(t, new(canalTestSuite))
+	suite.Run(t, newCanalTestSuite())
+	suite.Run(t, newCanalTestSuite(withAddr(mysql.DEFAULT_IPV6_ADDR)))
 }
 
 const (
@@ -37,6 +55,9 @@ const (
 func (s *canalTestSuite) SetupSuite() {
 	cfg := NewDefaultConfig()
 	cfg.Addr = fmt.Sprintf("%s:%s", *test_util.MysqlHost, *test_util.MysqlPort)
+	if s.addr != "" {
+		cfg.Addr = s.addr
+	}
 	cfg.User = "root"
 	cfg.HeartbeatPeriod = 200 * time.Millisecond
 	cfg.ReadTimeout = 300 * time.Millisecond

dump/dumper.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"os/exec"
 	"regexp"
@@ -212,10 +213,14 @@ func (d *Dumper) Dump(w io.Writer) error {
 	if strings.Contains(d.Addr, "/") {
 		args = append(args, fmt.Sprintf("--socket=%s", d.Addr))
 	} else {
-		seps := strings.SplitN(d.Addr, ":", 2)
-		args = append(args, fmt.Sprintf("--host=%s", seps[0]))
-		if len(seps) > 1 {
-			args = append(args, fmt.Sprintf("--port=%s", seps[1]))
+		host, port, err := net.SplitHostPort(d.Addr)
+		if err != nil {
+			host = d.Addr
+		}
+
+		args = append(args, fmt.Sprintf("--host=%s", host))
+		if port != "" {
+			args = append(args, fmt.Sprintf("--port=%s", port))
 		}
 	}
 

mysql/const.go

@@ -172,6 +172,7 @@ const (
 
 const (
 	DEFAULT_ADDR                  = "127.0.0.1:3306"
+	DEFAULT_IPV6_ADDR             = "[::1]:3306"
 	DEFAULT_USER                  = "root"
 	DEFAULT_PASSWORD              = ""
 	DEFAULT_FLAVOR                = "mysql"