VDiff External DB

go/cmd/vttablet/vttablet.go

@@ -130,6 +130,8 @@ func main() {
 		log.Exitf("NewActionAgent() failed: %v", err)
 	}
 
+	tabletserver.InitAPI(agent.VREngine)
+
 	servenv.OnClose(func() {
 		// stop the agent so that our topo entry gets pruned properly
 		agent.Close()

go/mysql/binlog_event_filepos.go

@@ -19,7 +19,6 @@ package mysql
 import (
 	"encoding/binary"
 	"fmt"
-	"strconv"
 )
 
 // filePosBinlogEvent wraps a raw packet buffer and provides methods to examine
@@ -228,7 +227,7 @@ func newFilePosGTIDEvent(file string, pos int, timestamp uint32) filePosGTIDEven
 		},
 		gtid: filePosGTID{
 			file: file,
-			pos:  strconv.Itoa(pos),
+			pos:  pos,
 		},
 	}
 }

go/mysql/filepos_gtid.go

@@ -18,6 +18,7 @@ package mysql
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
 )
 
@@ -31,9 +32,14 @@ func parseFilePosGTID(s string) (GTID, error) {
 		return nil, fmt.Errorf("invalid FilePos GTID (%v): expecting file:pos", s)
 	}
 
+	pos, err := strconv.Atoi(parts[1])
+	if err != nil {
+		return nil, fmt.Errorf("invalid FilePos GTID (%v): expecting pos to be an integer", s)
+	}
+
 	return filePosGTID{
 		file: parts[0],
-		pos:  parts[1],
+		pos:  pos,
 	}, nil
 }
 
@@ -48,12 +54,13 @@ func parseFilePosGTIDSet(s string) (GTIDSet, error) {
 
 // filePosGTID implements GTID.
 type filePosGTID struct {
-	file, pos string
+	file string
+	pos  int
 }
 
 // String implements GTID.String().
 func (gtid filePosGTID) String() string {
-	return gtid.file + ":" + gtid.pos
+	return fmt.Sprintf("%s:%d", gtid.file, gtid.pos)
 }
 
 // Flavor implements GTID.Flavor().

go/mysql/filepos_gtid_test.go

@@ -0,0 +1,102 @@
+/*
+Copyright 2020 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 (
+	"testing"
+)
+
+func Test_filePosGTID_String(t *testing.T) {
+	type fields struct {
+		file string
+		pos  int
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		want   string
+	}{
+		{
+			"formats gtid correctly",
+			fields{file: "mysql-bin.166031", pos: 192394},
+			"mysql-bin.166031:192394",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gtid := filePosGTID{
+				file: tt.fields.file,
+				pos:  tt.fields.pos,
+			}
+			if got := gtid.String(); got != tt.want {
+				t.Errorf("filePosGTID.String() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_filePosGTID_ContainsGTID(t *testing.T) {
+	type fields struct {
+		file string
+		pos  int
+	}
+	type args struct {
+		other GTID
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		args   args
+		want   bool
+	}{
+		{
+			"returns true when the position is equal",
+			fields{file: "testfile", pos: 1234},
+			args{other: filePosGTID{file: "testfile", pos: 1234}},
+			true,
+		},
+		{
+			"returns true when the position is less than equal",
+			fields{file: "testfile", pos: 1234},
+			args{other: filePosGTID{file: "testfile", pos: 1233}},
+			true,
+		},
+		{
+			"returns false when the position is less than equal",
+			fields{file: "testfile", pos: 1234},
+			args{other: filePosGTID{file: "testfile", pos: 1235}},
+			false,
+		},
+		{
+			"it uses integer value for comparison (it is not lexicographical order)",
+			fields{file: "testfile", pos: 99761227},
+			args{other: filePosGTID{file: "testfile", pos: 103939867}},
+			false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			gtid := filePosGTID{
+				file: tt.fields.file,
+				pos:  tt.fields.pos,
+			}
+			if got := gtid.ContainsGTID(tt.args.other); got != tt.want {
+				t.Errorf("filePosGTID.ContainsGTID() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

go/mysql/flavor_filepos.go

@@ -39,35 +39,25 @@ func newFilePosFlavor() flavor {
 
 // masterGTIDSet is part of the Flavor interface.
 func (flv *filePosFlavor) masterGTIDSet(c *Conn) (GTIDSet, error) {
-	qr, err := c.ExecuteFetch("SHOW SLAVE STATUS", 100, true /* wantfields */)
+	qr, err := c.ExecuteFetch("SHOW MASTER STATUS", 100, true /* wantfields */)
 	if err != nil {
 		return nil, err
 	}
 	if len(qr.Rows) == 0 {
-		qr, err = c.ExecuteFetch("SHOW MASTER STATUS", 100, true /* wantfields */)
-		if err != nil {
-			return nil, err
-		}
-		if len(qr.Rows) == 0 {
-			return nil, errors.New("no master or slave status")
-		}
-		resultMap, err := resultToMap(qr)
-		if err != nil {
-			return nil, err
-		}
-		return filePosGTID{
-			file: resultMap["File"],
-			pos:  resultMap["Position"],
-		}, nil
+		return nil, errors.New("no master status")
 	}
-
 	resultMap, err := resultToMap(qr)
 	if err != nil {
 		return nil, err
 	}
+	pos, err := strconv.Atoi(resultMap["Position"])
+	if err != nil {
+		return nil, fmt.Errorf("invalid FilePos GTID (%v): expecting pos to be an integer", resultMap["Position"])
+	}
+
 	return filePosGTID{
-		file: resultMap["Relay_Master_Log_File"],
-		pos:  resultMap["Exec_Master_Log_Pos"],
+		file: resultMap["File"],
+		pos:  pos,
 	}, nil
 }
 
@@ -86,13 +76,8 @@ func (flv *filePosFlavor) sendBinlogDumpCommand(c *Conn, slaveID uint32, startPo
 		return fmt.Errorf("startPos.GTIDSet is wrong type - expected filePosGTID, got: %#v", startPos.GTIDSet)
 	}
 
-	pos, err := strconv.Atoi(rpos.pos)
-	if err != nil {
-		return fmt.Errorf("invalid position: %v", startPos.GTIDSet)
-	}
 	flv.file = rpos.file
-
-	return c.WriteComBinlogDump(slaveID, rpos.file, uint32(pos), 0)
+	return c.WriteComBinlogDump(slaveID, rpos.file, uint32(rpos.pos), 0)
 }
 
 // readBinlogEvent is part of the Flavor interface.
@@ -200,9 +185,14 @@ func (flv *filePosFlavor) status(c *Conn) (SlaveStatus, error) {
 	}
 
 	status := parseSlaveStatus(resultMap)
+	pos, err := strconv.Atoi(resultMap["Exec_Master_Log_Pos"])
+	if err != nil {
+		return SlaveStatus{}, fmt.Errorf("invalid FilePos GTID (%v): expecting pos to be an integer", resultMap["Exec_Master_Log_Pos"])
+	}
+
 	status.Position.GTIDSet = filePosGTID{
 		file: resultMap["Relay_Master_Log_File"],
-		pos:  resultMap["Exec_Master_Log_Pos"],
+		pos:  pos,
 	}
 	return status, nil
 }
@@ -219,10 +209,10 @@ func (flv *filePosFlavor) waitUntilPositionCommand(ctx context.Context, pos Posi
 		if timeout <= 0 {
 			return "", fmt.Errorf("timed out waiting for position %v", pos)
 		}
-		return fmt.Sprintf("SELECT MASTER_POS_WAIT('%s', %s, %.6f)", filePosPos.file, filePosPos.pos, timeout.Seconds()), nil
+		return fmt.Sprintf("SELECT MASTER_POS_WAIT('%s', %d, %.6f)", filePosPos.file, filePosPos.pos, timeout.Seconds()), nil
 	}
 
-	return fmt.Sprintf("SELECT MASTER_POS_WAIT('%s', %s)", filePosPos.file, filePosPos.pos), nil
+	return fmt.Sprintf("SELECT MASTER_POS_WAIT('%s', %d)", filePosPos.file, filePosPos.pos), nil
 }
 
 func (*filePosFlavor) startSlaveUntilAfter(pos Position) string {

go/vt/vttablet/tabletmanager/action_agent.go

@@ -311,7 +311,7 @@ func NewActionAgent(
 	vreplication.InitVStreamerClient(agent.DBConfigs)
 
 	// The db name is set by the Start function called above
-	agent.VREngine = vreplication.NewEngine(ts, tabletAlias.Cell, mysqld, func() binlogplayer.DBClient {
+	agent.VREngine = vreplication.NewEngine(ts, tabletAlias.GetCell(), agent.Tablet(), mysqld, func() binlogplayer.DBClient {
 		return binlogplayer.NewDBClient(agent.DBConfigs.FilteredWithDB())
 	},
 		agent.DBConfigs.FilteredWithDB().DbName,
@@ -382,7 +382,7 @@ func NewTestActionAgent(batchCtx context.Context, ts *topo.Server, tabletAlias *
 		Cnf:                 nil,
 		MysqlDaemon:         mysqlDaemon,
 		DBConfigs:           &dbconfigs.DBConfigs{},
-		VREngine:            vreplication.NewEngine(ts, tabletAlias.Cell, mysqlDaemon, binlogplayer.NewFakeDBClient, ti.DbName()),
+		VREngine:            vreplication.NewEngine(ts, tabletAlias.GetCell(), nil, mysqlDaemon, binlogplayer.NewFakeDBClient, ti.DbName()),
 		History:             history.New(historyLength),
 		_healthy:            fmt.Errorf("healthcheck not run yet"),
 	}
@@ -421,7 +421,7 @@ func NewComboActionAgent(batchCtx context.Context, ts *topo.Server, tabletAlias
 		Cnf:                 nil,
 		MysqlDaemon:         mysqlDaemon,
 		DBConfigs:           dbcfgs,
-		VREngine:            vreplication.NewEngine(nil, "", nil, nil, ""),
+		VREngine:            vreplication.NewEngine(nil, "", nil, nil, nil, ""),
 		gotMysqlPort:        true,
 		History:             history.New(historyLength),
 		_healthy:            fmt.Errorf("healthcheck not run yet"),

go/vt/vttablet/tabletmanager/init_tablet_test.go

@@ -194,7 +194,7 @@ func TestInitTablet(t *testing.T) {
 		TabletAlias: tabletAlias,
 		MysqlDaemon: mysqlDaemon,
 		DBConfigs:   &dbconfigs.DBConfigs{},
-		VREngine:    vreplication.NewEngine(nil, "", nil, nil, ""),
+		VREngine:    vreplication.NewEngine(nil, "", nil, nil, nil, ""),
 		batchCtx:    ctx,
 		History:     history.New(historyLength),
 		_healthy:    fmt.Errorf("healthcheck not run yet"),

go/vt/vttablet/tabletmanager/vreplication/controller.go

@@ -157,6 +157,53 @@ func (ct *controller) run(ctx context.Context) {
 	}
 }
 
+func (ct *controller) runVDiff(ctx context.Context) (err error) {
+	defer func() {
+		ct.sourceTablet.Set("")
+		if x := recover(); x != nil {
+			log.Errorf("stream %v: caught panic: %v\n%s", ct.id, x, tb.Stack(4))
+			err = fmt.Errorf("panic: %v", x)
+		}
+	}()
+
+	select {
+	case <-ctx.Done():
+		return nil
+	default:
+	}
+
+	dbClient := ct.dbClientFactory()
+	if err := dbClient.Connect(); err != nil {
+		return vterrors.Wrap(err, "can't connect to database")
+	}
+	defer dbClient.Close()
+
+	var tablet *topodatapb.Tablet
+	if ct.source.GetExternalMysql() == "" {
+		log.Infof("trying to find a tablet eligible for vreplication. stream id: %v", ct.id)
+		tablet, err = ct.tabletPicker.PickForStreaming(ctx)
+		if err != nil {
+			return err
+		}
+		log.Infof("found a tablet eligible for vreplication. stream id: %v  tablet: %s", ct.id, tablet.Alias.String())
+		ct.sourceTablet.Set(tablet.Alias.String())
+	}
+
+	switch {
+	case ct.source.Filter != nil:
+		var vsClient VStreamerClient
+		if ct.source.GetExternalMysql() == "" {
+			vsClient = NewTabletVStreamerClient(tablet, ct.mysqld)
+		} else {
+			vsClient = NewMySQLVStreamerClient()
+		}
+
+		vd := newVDiffer(ct.id, &ct.source, vsClient, ct.blpStats, dbClient, ct.vre, ct.workflow)
+		return vd.VDiff(ctx, 60*time.Second)
+	}
+	return fmt.Errorf("missing source")
+}
+
 func (ct *controller) runBlp(ctx context.Context) (err error) {
 	defer func() {
 		ct.sourceTablet.Set("")
@@ -222,7 +269,7 @@ func (ct *controller) runBlp(ctx context.Context) (err error) {
 
 		var vsClient VStreamerClient
 		if ct.source.GetExternalMysql() == "" {
-			vsClient = NewTabletVStreamerClient(tablet)
+			vsClient = NewTabletVStreamerClient(tablet, nil)
 		} else {
 			vsClient = NewMySQLVStreamerClient()
 		}