From 760b566b1527e1db01308c58e6bc1626756dc0d0 Mon Sep 17 00:00:00 2001
From: sino <sino2322@gmail.com>
Date: Mon, 11 Nov 2024 16:01:22 +0800
Subject: [PATCH] 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
---
 .github/workflows/ci.yml |  6 ++++++
 canal/canal.go           | 13 ++++++-------
 canal/canal_test.go      | 23 ++++++++++++++++++++++-
 dump/dumper.go           | 13 +++++++++----
 mysql/const.go           |  1 +
 5 files changed, 44 insertions(+), 12 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index fe75cb5aa..6036731df 100644
--- a/.github/workflows/ci.yml
+++ b/.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 ./...
diff --git a/canal/canal.go b/canal/canal.go
index 1ba0c5165..64a1c343e 100644
--- a/canal/canal.go
+++ b/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)
diff --git a/canal/canal_test.go b/canal/canal_test.go
index 14a7056b9..444256076 100644
--- a/canal/canal_test.go
+++ b/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
diff --git a/dump/dumper.go b/dump/dumper.go
index c7ea56db7..b2b7e4377 100644
--- a/dump/dumper.go
+++ b/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))
 		}
 	}
 
diff --git a/mysql/const.go b/mysql/const.go
index b6da2e736..a11b05370 100644
--- a/mysql/const.go
+++ b/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"