Updates for binlog primary protocol

go/mysql/binlog_dump.go

@@ -78,7 +78,15 @@ func (c *Conn) parseComBinlogDumpGTID(data []byte) (logFile string, logPos uint6
 		if !ok {
 			return logFile, logPos, position, readPacketErr
 		}
-		if gtid := string(data[pos : pos+int(dataSize)]); gtid != "" {
+
+		// NOTE: In testing a replication connection with a MySQL 8.0 replica, the replica sends back an 8 byte
+		//       GTID will all zero valued bytes, but Vitess was unable to parse it, so we added the nilGtid
+		//       check below. We may be able to remove this check if we find the primary is not sending back
+		//       some missing metadata, but documentation seems to indicate that the replica is allowed to send
+		//       an empty GTID set when no GTID was specified in the start replica statement. Looking through
+		//       MySQL primary <-> MySQL replica wire logs could help prove this out.
+		gtidBytes := data[pos : pos+int(dataSize)]
+		if gtid := string(gtidBytes); gtid != "" && !isNilGtid(gtidBytes) {
 			position, err = DecodePosition(gtid)
 			if err != nil {
 				return logFile, logPos, position, err
@@ -88,3 +96,14 @@ func (c *Conn) parseComBinlogDumpGTID(data []byte) (logFile string, logPos uint6
 
 	return logFile, logPos, position, nil
 }
+
+// isNilGtid returns true if the specified |bytes| that represent a serialized GTID are all zero valued,
+// otherwise returns false.
+func isNilGtid(bytes []byte) bool {
+	for _, b := range bytes {
+		if uint(b) != 0 {
+			return false
+		}
+	}
+	return true
+}

go/mysql/conn.go

@@ -1367,10 +1367,11 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 		return nil
 
 	case ComRegisterReplica:
-		// TODO: Seems like we probably need this command implemented, too, but it hasn't been needed
-		//       yet in a simple Vitess <-> Vitess replication test, so skipping for now.
-		//return c.handleComRegisterReplica(handler, data)
-		return fmt.Errorf("ComRegisterReplica not implemented")
+		ok := c.handleComRegisterReplica(handler, data)
+		if !ok {
+			return fmt.Errorf("error handling ComRegisterReplica packet: %v", data)
+		}
+		return nil
 
 	default:
 		log.Errorf("Got unhandled packet (default) from %s, returning error: %v", c, data)
@@ -1384,6 +1385,31 @@ func (c *Conn) handleNextCommand(handler Handler) error {
 	return nil
 }
 
+func (c *Conn) handleComRegisterReplica(handler Handler, data []byte) (kontinue bool) {
+	binlogReplicaHandler, ok := handler.(BinlogReplicaHandler)
+	if !ok {
+		log.Warningf("received COM_REGISTER_REPLICA command, but handler does not implement BinlogReplicaHandler")
+		return true
+	}
+
+	c.recycleReadPacket()
+
+	replicaHost, replicaPort, replicaUser, replicaPassword, err := c.parseComRegisterReplica(data)
+	if err != nil {
+		log.Errorf("conn %v: parseComRegisterReplica failed: %v", c.ID(), err)
+		return false
+	}
+	if err := binlogReplicaHandler.ComRegisterReplica(c, replicaHost, replicaPort, replicaUser, replicaPassword); err != nil {
+		c.writeErrorPacketFromError(err)
+		return false
+	}
+
+	if err := c.writeOKPacket(0, 0, c.StatusFlags, 0); err != nil {
+		c.writeErrorPacketFromError(err)
+	}
+	return true
+}
+
 func (c *Conn) handleComBinlogDumpGTID(handler Handler, data []byte) (kontinue bool) {
 	binlogReplicaHandler, ok := handler.(BinlogReplicaHandler)
 	if !ok {

go/mysql/encoding.go

@@ -188,6 +188,11 @@ func readEOFString(data []byte, pos int) (string, int, bool) {
 	return string(data[pos:]), len(data) - pos, true
 }
 
+func readUint8(data []byte, pos int) (uint8, int, bool) {
+	b, pos, ok := readByte(data, pos)
+	return uint8(b), pos, ok
+}
+
 func readUint16(data []byte, pos int) (uint16, int, bool) {
 	if pos+1 >= len(data) {
 		return 0, 0, false

go/mysql/register-replica.go

@@ -0,0 +1,72 @@
+/*
+Copyright 2022 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package mysql
+
+import (
+	vtrpcpb "github.com/dolthub/vitess/go/vt/proto/vtrpc"
+	"github.com/dolthub/vitess/go/vt/vterrors"
+)
+
+var (
+	comRegisterReplicaPacketErr = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "error reading BinlogDumpGTID packet")
+)
+
+func (c *Conn) parseComRegisterReplica(data []byte) (
+	replicaHost string,
+	replicaPort uint16,
+	replicaUser string,
+	replicaPassword string,
+	err error,
+) {
+	pos := 1
+	pos += 4 // server-id
+
+	// hostname
+	hostnameLen, pos, ok := readUint8(data, pos)
+	if !ok {
+		return replicaHost, replicaPort, replicaUser, replicaPassword, comRegisterReplicaPacketErr
+	}
+	replicaHost = string(data[pos : pos+int(hostnameLen)])
+	pos += int(hostnameLen)
+
+	// username
+	usernameLen, pos, ok := readUint8(data, pos)
+	if !ok {
+		return replicaHost, replicaPort, replicaUser, replicaPassword, comRegisterReplicaPacketErr
+	}
+	replicaUser = string(data[pos : pos+int(usernameLen)])
+	pos += int(usernameLen)
+
+	// password
+	passwordLen, pos, ok := readUint8(data, pos)
+	if !ok {
+		return replicaHost, replicaPort, replicaUser, replicaPassword, comRegisterReplicaPacketErr
+	}
+	replicaPassword = string(data[pos : pos+int(passwordLen)])
+	pos += int(passwordLen)
+
+	// port
+	replicaPort, _, ok = readUint16(data, pos)
+	if !ok {
+		return replicaHost, replicaPort, replicaUser, replicaPassword, comRegisterReplicaPacketErr
+	}
+	// remaining: (commented because of ineffectual assignment)
+	// pos += 4 // replication rank
+	// pos += 4 // master-id
+
+	return replicaHost, replicaPort, replicaUser, replicaPassword, nil
+}