diff --git a/.github/workflows/misc_test_docker.yml b/.github/workflows/docker_test_1.yml similarity index 62% rename from .github/workflows/misc_test_docker.yml rename to .github/workflows/docker_test_1.yml index fd617e46b34..c963d1788b8 100644 --- a/.github/workflows/misc_test_docker.yml +++ b/.github/workflows/docker_test_1.yml @@ -1,10 +1,11 @@ -name: misc test +name: docker_test_1 on: [push, pull_request] jobs: build: - name: Misc Test + name: Docker Test 1 runs-on: ubuntu-latest + steps: - name: Set up Go @@ -15,6 +16,6 @@ jobs: - name: Check out code uses: actions/checkout@v2 - - name: Run Misc test which requires docker + - name: Run tests which requires docker 1 run: | - go run test.go -docker=true -shard 25 \ No newline at end of file + go run test.go -docker=true --follow -shard 10 \ No newline at end of file diff --git a/.github/workflows/cluster_vtctl_web.yml b/.github/workflows/docker_test_2.yml similarity index 58% rename from .github/workflows/cluster_vtctl_web.yml rename to .github/workflows/docker_test_2.yml index 82597559b0e..81433ea58a7 100644 --- a/.github/workflows/cluster_vtctl_web.yml +++ b/.github/workflows/docker_test_2.yml @@ -1,10 +1,11 @@ -name: cluster_vtctl_web +name: docker test 2 on: [push, pull_request] jobs: build: - name: cluster vtctl web + name: Docker Test 2 runs-on: ubuntu-latest + steps: - name: Set up Go @@ -15,8 +16,6 @@ jobs: - name: Check out code uses: actions/checkout@v2 - - name: Run vtctl web + - name: Run tests which requires docker - 2 run: | - # Running web test inside docker - go run test.go -docker=true -print-log -shard 10 - + go run test.go -docker=true --follow -shard 25 \ No newline at end of file diff --git a/build.env b/build.env index a09807bdf5a..c16ec1e2da8 100755 --- a/build.env +++ b/build.env @@ -43,3 +43,4 @@ mkdir -p .git/hooks ln -sf "$PWD/misc/git/pre-commit" .git/hooks/pre-commit ln -sf "$PWD/misc/git/commit-msg" .git/hooks/commit-msg git config core.hooksPath .git/hooks +export EXTRA_BIN=$PWD/test/bin \ No newline at end of file diff --git a/docker/test/run.sh b/docker/test/run.sh index 539a7fa57e0..842c4512b1b 100755 --- a/docker/test/run.sh +++ b/docker/test/run.sh @@ -139,6 +139,7 @@ args="$args -v /tmp/mavencache:/home/vitess/.m2" # Add in the vitess user args="$args --user vitess" +args="$args -v $PWD/test/bin:/tmp/bin" # Mount in host VTDATAROOT if one exists, since it might be a RAM disk or SSD. if [[ -n "$VTDATAROOT" ]]; then @@ -172,6 +173,7 @@ fi # Reset the environment if this was an old bootstrap. We can detect this from VTTOP presence. bashcmd=$(append_cmd "$bashcmd" "export VTROOT=/vt/src/vitess.io/vitess") bashcmd=$(append_cmd "$bashcmd" "export VTDATAROOT=/vt/vtdataroot") +bashcmd=$(append_cmd "$bashcmd" "export EXTRA_BIN=/tmp/bin") bashcmd=$(append_cmd "$bashcmd" "mkdir -p dist; mkdir -p bin; mkdir -p lib; mkdir -p vthook") bashcmd=$(append_cmd "$bashcmd" "rm -rf /vt/dist; ln -s /vt/src/vitess.io/vitess/dist /vt/dist") diff --git a/go/mysql/filepos_gtid.go b/go/mysql/filepos_gtid.go index 26d3ebdcfa1..8a276f0a018 100644 --- a/go/mysql/filepos_gtid.go +++ b/go/mysql/filepos_gtid.go @@ -147,6 +147,13 @@ func (gtid filePosGTID) Union(other GTIDSet) GTIDSet { return filePosOther } +// Last returns last filePosition +// For filePos based GTID we have only one position +// here we will just return the current filePos +func (gtid filePosGTID) Last() string { + return gtid.String() +} + func init() { gtidParsers[FilePosFlavorID] = parseFilePosGTID gtidSetParsers[FilePosFlavorID] = parseFilePosGTIDSet diff --git a/go/mysql/flavor_mysql.go b/go/mysql/flavor_mysql.go index 022ab4328d8..c4b963e3bd1 100644 --- a/go/mysql/flavor_mysql.go +++ b/go/mysql/flavor_mysql.go @@ -32,7 +32,8 @@ type mysqlFlavor struct{} // masterGTIDSet is part of the Flavor interface. func (mysqlFlavor) masterGTIDSet(c *Conn) (GTIDSet, error) { - qr, err := c.ExecuteFetch("SELECT @@GLOBAL.gtid_executed", 1, false) + // keep @@global as lowercase, as some servers like the Ripple binlog server only honors a lowercase `global` value + qr, err := c.ExecuteFetch("SELECT @@global.gtid_executed", 1, false) if err != nil { return nil, err } diff --git a/go/mysql/gtid_set.go b/go/mysql/gtid_set.go index b966e8bc2a5..812b7f33caf 100644 --- a/go/mysql/gtid_set.go +++ b/go/mysql/gtid_set.go @@ -49,6 +49,9 @@ type GTIDSet interface { // Union returns a union of the receiver GTIDSet and the supplied GTIDSet. Union(GTIDSet) GTIDSet + + // Union returns a union of the receiver GTIDSet and the supplied GTIDSet. + Last() string } // gtidSetParsers maps flavor names to parser functions. It is used by diff --git a/go/mysql/gtid_test.go b/go/mysql/gtid_test.go index bad7daabc85..67ac707224f 100644 --- a/go/mysql/gtid_test.go +++ b/go/mysql/gtid_test.go @@ -193,6 +193,7 @@ type fakeGTID struct { } func (f fakeGTID) String() string { return f.value } +func (f fakeGTID) Last() string { panic("not implemented") } func (f fakeGTID) Flavor() string { return f.flavor } func (fakeGTID) SourceServer() interface{} { return int(1) } func (fakeGTID) SequenceNumber() interface{} { return int(1) } diff --git a/go/mysql/mariadb_gtid.go b/go/mysql/mariadb_gtid.go index cf3856a0cdb..bb164e2f37c 100644 --- a/go/mysql/mariadb_gtid.go +++ b/go/mysql/mariadb_gtid.go @@ -233,6 +233,21 @@ func (gtidSet MariadbGTIDSet) Union(other GTIDSet) GTIDSet { return newSet } +//Last returns the last gtid +func (gtidSet MariadbGTIDSet) Last() string { + // Sort domains so the string format is deterministic. + domains := make([]uint32, 0, len(gtidSet)) + for domain := range gtidSet { + domains = append(domains, domain) + } + sort.Slice(domains, func(i, j int) bool { + return domains[i] < domains[j] + }) + + lastGTID := domains[len(gtidSet)-1] + return gtidSet[lastGTID].String() +} + // deepCopy returns a deep copy of the set. func (gtidSet MariadbGTIDSet) deepCopy() MariadbGTIDSet { newSet := make(MariadbGTIDSet, len(gtidSet)) diff --git a/go/mysql/mariadb_gtid_test.go b/go/mysql/mariadb_gtid_test.go index 0c03f967b3b..0f0f3896484 100644 --- a/go/mysql/mariadb_gtid_test.go +++ b/go/mysql/mariadb_gtid_test.go @@ -19,6 +19,9 @@ package mysql import ( "strings" "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestParseMariaGTID(t *testing.T) { @@ -640,3 +643,16 @@ func TestMariaGTIDSetUnionNewDomain(t *testing.T) { t.Error("Union result was not of type MariadbGTIDSet.") } } + +func TestMariaGTIDSetLast(t *testing.T) { + + testCases := map[string]string{ + "12-34-5678,11-22-3333,24-52-4523": "24-52-4523", + "12-34-5678": "12-34-5678", + } + for input, want := range testCases { + got, err := parseMariadbGTIDSet(input) + require.NoError(t, err) + assert.Equal(t, want, got.Last()) + } +} diff --git a/go/mysql/mysql56_gtid_set.go b/go/mysql/mysql56_gtid_set.go index a43a2959b30..408c3cac0d6 100644 --- a/go/mysql/mysql56_gtid_set.go +++ b/go/mysql/mysql56_gtid_set.go @@ -171,6 +171,26 @@ func (set Mysql56GTIDSet) String() string { return buf.String() } +// Last returns the last gtid as string +// For gtidset having multiple SIDs or multiple intervals +// it just returns the last SID with last interval +func (set Mysql56GTIDSet) Last() string { + buf := &bytes.Buffer{} + + if len(set.SIDs()) > 0 { + sid := set.SIDs()[len(set.SIDs())-1] + buf.WriteString(sid.String()) + sequences := set[sid] + if len(sequences) > 0 { + buf.WriteByte(':') + lastInterval := sequences[len(sequences)-1] + buf.WriteString(strconv.FormatInt(lastInterval.end, 10)) + } + } + + return buf.String() +} + // Flavor implements GTIDSet. func (Mysql56GTIDSet) Flavor() string { return Mysql56FlavorID } diff --git a/go/mysql/mysql56_gtid_set_test.go b/go/mysql/mysql56_gtid_set_test.go index d7140421760..fdb0569c1a2 100644 --- a/go/mysql/mysql56_gtid_set_test.go +++ b/go/mysql/mysql56_gtid_set_test.go @@ -21,6 +21,8 @@ import ( "sort" "strings" "testing" + + "github.com/stretchr/testify/assert" ) func TestSortSIDList(t *testing.T) { @@ -527,3 +529,36 @@ func TestMysql56GTIDSetSIDBlock(t *testing.T) { t.Errorf("NewMysql56GTIDSetFromSIDBlock(%#v) = %#v, want %#v", want, set, input) } } + +func TestMySQL56GTIDSetLast(t *testing.T) { + sid1 := SID{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15} + sid2 := SID{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 255} + + table := map[string]Mysql56GTIDSet{ + // Simple case + "00010203-0405-0607-0809-0a0b0c0d0e0f:5": { + sid1: []interval{{1, 5}}, + }, + "00010203-0405-0607-0809-0a0b0c0d0e0f:3": { + sid1: []interval{{end: 3}}, + }, + // Interval with same start and end + "00010203-0405-0607-0809-0a0b0c0d0e0f:12": { + sid1: []interval{{12, 12}}, + }, + // Multiple intervals + "00010203-0405-0607-0809-0a0b0c0d0e0f:20": { + sid1: []interval{{1, 5}, {10, 20}}, + }, + // Multiple SIDs + "00010203-0405-0607-0809-0a0b0c0d0eff:50": { + sid1: []interval{{1, 5}, {10, 20}}, + sid2: []interval{{1, 5}, {50, 50}}, + }, + } + + for want, input := range table { + got := strings.ToLower(input.Last()) + assert.Equal(t, want, got) + } +} diff --git a/go/test/endtoend/cluster/cluster_util.go b/go/test/endtoend/cluster/cluster_util.go index bce1cf73810..9f15d61a902 100644 --- a/go/test/endtoend/cluster/cluster_util.go +++ b/go/test/endtoend/cluster/cluster_util.go @@ -73,11 +73,11 @@ func GetMasterPosition(t *testing.T, vttablet Vttablet, hostname string) (string return pos, gtID } -// VerifyRowsInTablet Verify total number of rows in a tablet -func VerifyRowsInTablet(t *testing.T, vttablet *Vttablet, ksName string, expectedRows int) { +// VerifyRowsInTabletForTable Verify total number of rows in a table +func VerifyRowsInTabletForTable(t *testing.T, vttablet *Vttablet, ksName string, expectedRows int, tableName string) { timeout := time.Now().Add(10 * time.Second) for time.Now().Before(timeout) { - qr, err := vttablet.VttabletProcess.QueryTablet("select * from vt_insert_test", ksName, true) + qr, err := vttablet.VttabletProcess.QueryTablet("select * from "+tableName, ksName, true) require.Nil(t, err) if len(qr.Rows) == expectedRows { return @@ -87,6 +87,11 @@ func VerifyRowsInTablet(t *testing.T, vttablet *Vttablet, ksName string, expecte assert.Fail(t, "expected rows not found.") } +// VerifyRowsInTablet Verify total number of rows in a tablet +func VerifyRowsInTablet(t *testing.T, vttablet *Vttablet, ksName string, expectedRows int) { + VerifyRowsInTabletForTable(t, vttablet, ksName, expectedRows, "vt_insert_test") +} + // PanicHandler handles the panic in the testcase. func PanicHandler(t *testing.T) { err := recover() diff --git a/go/test/endtoend/recovery/pitr/binlog_server.go b/go/test/endtoend/recovery/pitr/binlog_server.go new file mode 100644 index 00000000000..efe04255fc7 --- /dev/null +++ b/go/test/endtoend/recovery/pitr/binlog_server.go @@ -0,0 +1,110 @@ +package pitr + +import ( + "fmt" + "os" + "os/exec" + "path" + "strings" + "syscall" + "time" + + "vitess.io/vitess/go/vt/log" +) + +const ( + binlogExecutableName = "rippled" + binlogDataDir = "binlog_dir" + binlogUser = "ripple" +) + +type binLogServer struct { + hostname string + port int + username string + dataDirectory string + executablePath string + + proc *exec.Cmd + exit chan error +} + +type mysqlMaster struct { + hostname string + port int + username string + password string +} + +// newBinlogServer returns an instance of binlog server +func newBinlogServer(hostname string, port int) (*binLogServer, error) { + dataDir := path.Join(os.Getenv("VTDATAROOT"), binlogDataDir) + fmt.Println(dataDir) + if _, err := os.Stat(dataDir); os.IsNotExist(err) { + err := os.Mkdir(dataDir, 0700) + if err != nil { + log.Error(err) + return nil, err + } + } + return &binLogServer{ + executablePath: path.Join(os.Getenv("EXTRA_BIN"), binlogExecutableName), + dataDirectory: dataDir, + username: binlogUser, + hostname: hostname, + port: port, + }, nil +} + +// start starts the binlog server points to running mysql port +func (bs *binLogServer) start(master mysqlMaster) error { + bs.proc = exec.Command( + bs.executablePath, + fmt.Sprintf("-ripple_datadir=%s", bs.dataDirectory), + fmt.Sprintf("-ripple_master_address=%s", master.hostname), + fmt.Sprintf("-ripple_master_port=%d", master.port), + fmt.Sprintf("-ripple_master_user=%s", master.username), + fmt.Sprintf("-ripple_server_ports=%d", bs.port), + ) + if master.password != "" { + bs.proc.Args = append(bs.proc.Args, fmt.Sprintf("-ripple_master_password=%s", master.password)) + } + + errFile, _ := os.Create(path.Join(bs.dataDirectory, "log.txt")) + bs.proc.Stderr = errFile + + bs.proc.Env = append(bs.proc.Env, os.Environ()...) + + log.Infof("Running binlog server with command: %v", strings.Join(bs.proc.Args, " ")) + + err := bs.proc.Start() + if err != nil { + return err + } + bs.exit = make(chan error) + go func() { + if bs.proc != nil { + bs.exit <- bs.proc.Wait() + } + }() + return nil +} + +func (bs *binLogServer) stop() error { + if bs.proc == nil || bs.exit == nil { + return nil + } + // Attempt graceful shutdown with SIGTERM first + bs.proc.Process.Signal(syscall.SIGTERM) + + select { + case err := <-bs.exit: + bs.proc = nil + return err + + case <-time.After(10 * time.Second): + bs.proc.Process.Kill() + bs.proc = nil + return <-bs.exit + } +} diff --git a/go/test/endtoend/recovery/pitr/main_test.go b/go/test/endtoend/recovery/pitr/main_test.go new file mode 100644 index 00000000000..2421e35a579 --- /dev/null +++ b/go/test/endtoend/recovery/pitr/main_test.go @@ -0,0 +1,134 @@ +package pitr + +import ( + "flag" + "fmt" + "os" + "os/exec" + "testing" + + "vitess.io/vitess/go/test/endtoend/cluster" + "vitess.io/vitess/go/vt/log" +) + +var ( + clusterInstance *cluster.LocalProcessCluster + masterTablet *cluster.Vttablet + replicaTablet *cluster.Vttablet + initDBFile string + + cell = "zone1" + hostname = "localhost" + binlogHost = "127.0.0.1" + keyspaceName = "ks" + partialRestoreKSName = "restoreks1" + fullRestoreKSName = "restoreks2" + dbName = "vt_ks" + shardName = "0" + shardKsName = "ks/0" + mysqlUserName = "vt_dba" +) + +func TestMain(m *testing.M) { + defer cluster.PanicHandler(nil) + flag.Parse() + + exitCode, err := func() (int, error) { + clusterInstance = cluster.NewCluster(cell, hostname) + defer clusterInstance.Teardown() + + // Start topo server + if err := clusterInstance.StartTopo(); err != nil { + log.Error(err) + return 1, err + } + + if err := clusterInstance.VtctlProcess.CreateKeyspace(keyspaceName); err != nil { + log.Error(err) + return 1, err + } + clusterInstance.Keyspaces = append(clusterInstance.Keyspaces, cluster.Keyspace{Name: keyspaceName}) + + masterTablet = clusterInstance.NewVttabletInstance("replica", 0, cell) + replicaTablet = clusterInstance.NewVttabletInstance("replica", 0, cell) + + err := startTablets([]*cluster.Vttablet{masterTablet, replicaTablet}) + if err != nil { + return 1, err + } + err = clusterInstance.VtctlclientProcess.InitShardMaster(keyspaceName, shardName, cell, masterTablet.TabletUID) + if err != nil { + return 1, err + } + return m.Run(), nil + }() + if err != nil { + log.Error(err) + os.Exit(1) + } + os.Exit(exitCode) + +} + +func startTablets(tablets []*cluster.Vttablet) error { + var mysqlProcesses []*exec.Cmd + shard := &cluster.Shard{ + Name: shardName, + } + for _, tablet := range tablets { + tablet.MysqlctlProcess = *cluster.MysqlCtlProcessInstance(tablet.TabletUID, tablet.MySQLPort, clusterInstance.TmpDirectory) + tablet.MysqlctlProcess.InitDBFile = initDBFile + if proc, err := tablet.MysqlctlProcess.StartProcess(); err != nil { + return err + } else { + // ignore golint warning, we need the else block to use proc + mysqlProcesses = append(mysqlProcesses, proc) //nolint + } + tablet.VttabletProcess = cluster.VttabletProcessInstance(tablet.HTTPPort, + tablet.GrpcPort, + tablet.TabletUID, + clusterInstance.Cell, + shardName, + keyspaceName, + clusterInstance.VtctldProcess.Port, + tablet.Type, + clusterInstance.TopoProcess.Port, + clusterInstance.Hostname, + clusterInstance.TmpDirectory, + clusterInstance.VtTabletExtraArgs, + clusterInstance.EnableSemiSync) + tablet.Alias = tablet.VttabletProcess.TabletPath + tablet.VttabletProcess.SupportsBackup = true + shard.Vttablets = append(shard.Vttablets, tablet) + shard.Vttablets = append(shard.Vttablets, tablet) + } + for _, proc := range mysqlProcesses { + proc.Wait() + } + queryCmds := []string{ + fmt.Sprintf("CREATE USER '%s'@'%%';", mysqlUserName), + fmt.Sprintf("GRANT ALL ON *.* TO '%s'@'%%';", mysqlUserName), + fmt.Sprintf("GRANT GRANT OPTION ON *.* TO '%s'@'%%';", mysqlUserName), + "FLUSH PRIVILEGES;", + fmt.Sprintf("create database %s;", dbName), + } + for _, tablet := range tablets { + for _, query := range queryCmds { + _, err := tablet.VttabletProcess.QueryTablet(query, keyspaceName, false) + if err != nil { + log.Error(err) + return err + } + } + } + for _, tablet := range tablets { + err := tablet.VttabletProcess.Setup() + if err != nil { + log.Error(err) + return err + } + } + + clusterInstance.Keyspaces[0].Shards = []cluster.Shard{*shard} + return nil +} diff --git a/go/test/endtoend/recovery/pitr/pitr_test.go b/go/test/endtoend/recovery/pitr/pitr_test.go new file mode 100644 index 00000000000..8c3bbec89e7 --- /dev/null +++ b/go/test/endtoend/recovery/pitr/pitr_test.go @@ -0,0 +1,172 @@ +package pitr + +import ( + "fmt" + "testing" + "time" + + "github.com/magiconair/properties/assert" + + "vitess.io/vitess/go/vt/log" + + "vitess.io/vitess/go/test/endtoend/cluster" + + "github.com/stretchr/testify/require" +) + +var ( + createTable = `create table product (id bigint(20) primary key, name char(10), created bigint(20));` + insertTable = `insert into product (id, name, created) values(%d, '%s', unix_timestamp());` + selectMaxID = `select max(id) from product` +) + +func TestPointInTimeRecovery(t *testing.T) { + // create table and insert 2 rows + _, err := masterTablet.VttabletProcess.QueryTablet(createTable, keyspaceName, true) + require.NoError(t, err) + insertRow(t, 1, "p1", false) + insertRow(t, 2, "p2", false) + + // wait till replica catchup + cluster.VerifyRowsInTabletForTable(t, replicaTablet, keyspaceName, 2, "product") + + //start the binlog server and point it to master + bs, err := newBinlogServer(hostname, clusterInstance.GetAndReservePort()) + defer bs.stop() + require.NoError(t, err) + + err = bs.start(mysqlMaster{ + hostname: binlogHost, + port: masterTablet.MysqlctlProcess.MySQLPort, + username: mysqlUserName, + }) + require.NoError(t, err) + + // take the backup (to simulate the regular backup) + err = clusterInstance.VtctlclientProcess.ExecuteCommand("Backup", replicaTablet.Alias) + require.NoError(t, err) + + backups, err := clusterInstance.ListBackups(shardKsName) + require.NoError(t, err) + require.Equal(t, len(backups), 1) + + // now insert some more data to simulate the changes after regular backup + // every insert has some time lag/difference to simulate the time gap between rows + // and when we recover to certain time, this time gap will be able to identify the exact eligible row + var timeToRecover string + for counter := 3; counter <= 7; counter++ { + if counter == 5 { // we want to recovery till this, so noting the time + tm := time.Now().Add(1 * time.Second).UTC() + timeToRecover = tm.Format(time.RFC3339) + } + insertRow(t, counter, fmt.Sprintf("prd-%d", counter), true) + } + currentTime := time.Now().UTC() + timeForCompleteBinlogRecover := currentTime.Format(time.RFC3339) + // create restoreSnapshot + createRestoreSnapshot(t, timeToRecover, partialRestoreKSName) + + // test the recovery with smaller binlog_lookup_timeout + recoveryTabletWithSmallTimeout := clusterInstance.NewVttabletInstance("replica", 0, cell) + launchRecoveryTablet(t, recoveryTabletWithSmallTimeout, bs, "1ms", partialRestoreKSName) + + // since we have smaller timeout, it will just get whatever available in the backup + sqlRes, err := recoveryTabletWithSmallTimeout.VttabletProcess.QueryTablet(selectMaxID, keyspaceName, true) + require.NoError(t, err) + assert.Equal(t, sqlRes.Rows[0][0].String(), "INT64(2)") + + // test the recovery with valid binlog_lookup_timeout and timeToRecover pointing to 5th row + recoveryTablet1 := clusterInstance.NewVttabletInstance("replica", 0, cell) + launchRecoveryTablet(t, recoveryTablet1, bs, "2m", partialRestoreKSName) + + sqlRes, err = recoveryTablet1.VttabletProcess.QueryTablet(selectMaxID, keyspaceName, true) + require.NoError(t, err) + assert.Equal(t, sqlRes.Rows[0][0].String(), "INT64(5)") + + // test the recovery with timetorecover > (timestmap of last binlog event in binlog server) + createRestoreSnapshot(t, timeForCompleteBinlogRecover, fullRestoreKSName) + + recoveryTablet2 := clusterInstance.NewVttabletInstance("replica", 0, cell) + launchRecoveryTablet(t, recoveryTablet2, bs, "2m", fullRestoreKSName) + + sqlRes, err = recoveryTablet2.VttabletProcess.QueryTablet(selectMaxID, keyspaceName, true) + require.NoError(t, err) + assert.Equal(t, sqlRes.Rows[0][0].String(), "INT64(7)") + + defer recoveryTablet1.MysqlctlProcess.Stop() + defer recoveryTablet1.VttabletProcess.TearDown() + + defer recoveryTablet2.MysqlctlProcess.Stop() + defer recoveryTablet2.VttabletProcess.TearDown() + + defer recoveryTabletWithSmallTimeout.MysqlctlProcess.Stop() + defer recoveryTabletWithSmallTimeout.VttabletProcess.TearDown() +} + +func insertRow(t *testing.T, id int, productName string, isSlow bool) { + _, err := masterTablet.VttabletProcess.QueryTablet(fmt.Sprintf(insertTable, id, productName), keyspaceName, true) + require.NoError(t, err) + if isSlow { + time.Sleep(1 * time.Second) + } +} + +func createRestoreSnapshot(t *testing.T, timeToRecover, restoreKSName string) { + output, err := clusterInstance.VtctlclientProcess.ExecuteCommandWithOutput("CreateKeyspace", + "-keyspace_type=SNAPSHOT", "-base_keyspace="+keyspaceName, + "-snapshot_time", timeToRecover, restoreKSName) + log.Info(output) + require.Nil(t, err) +} + +func launchRecoveryTablet(t *testing.T, tablet *cluster.Vttablet, binlogServer *binLogServer, lookupTimeout, restoreKSName string) { + tablet.MysqlctlProcess = *cluster.MysqlCtlProcessInstance(tablet.TabletUID, tablet.MySQLPort, clusterInstance.TmpDirectory) + tablet.MysqlctlProcess.InitDBFile = initDBFile + err := tablet.MysqlctlProcess.Start() + require.NoError(t, err) + + tablet.VttabletProcess = cluster.VttabletProcessInstance(tablet.HTTPPort, + tablet.GrpcPort, + tablet.TabletUID, + clusterInstance.Cell, + shardName, + keyspaceName, + clusterInstance.VtctldProcess.Port, + tablet.Type, + clusterInstance.TopoProcess.Port, + clusterInstance.Hostname, + clusterInstance.TmpDirectory, + clusterInstance.VtTabletExtraArgs, + clusterInstance.EnableSemiSync) + tablet.Alias = tablet.VttabletProcess.TabletPath + tablet.VttabletProcess.SupportsBackup = true + tablet.VttabletProcess.Keyspace = restoreKSName + tablet.VttabletProcess.EnableSemiSync = true + tablet.VttabletProcess.ExtraArgs = []string{ + "-disable_active_reparents", + "-enable_replication_reporter=false", + "-init_db_name_override", dbName, + "-init_tablet_type", "replica", + "-init_keyspace", restoreKSName, + "-init_shard", shardName, + "-binlog_host", binlogServer.hostname, + "-binlog_port", fmt.Sprintf("%d", binlogServer.port), + "-binlog_user", binlogServer.username, + "-pitr_gtid_lookup_timeout", lookupTimeout, + "-vreplication_healthcheck_topology_refresh", "1s", + "-vreplication_healthcheck_retry_delay", "1s", + "-vreplication_tablet_type", "replica", + "-vreplication_retry_delay", "1s", + "-degraded_threshold", "5s", + "-lock_tables_timeout", "5s", + "-watch_replication_stream", + "-serving_state_grace_period", "1s", + } + tablet.VttabletProcess.ServingStatus = "" + + err = tablet.VttabletProcess.Setup() + require.NoError(t, err) + + tablet.VttabletProcess.WaitForTabletTypesForTimeout([]string{"SERVING"}, 20*time.Second) + require.Nil(t, err) +} diff --git a/go/vt/dbconfigs/dbconfigs.go b/go/vt/dbconfigs/dbconfigs.go index 98dba288d78..efabbe9d5e8 100644 --- a/go/vt/dbconfigs/dbconfigs.go +++ b/go/vt/dbconfigs/dbconfigs.go @@ -400,6 +400,12 @@ func (dbcfgs *DBConfigs) getParams(userKey string, dbc *DBConfigs) (*UserConfig, return uc, cp } +// SetDbParams sets the dba and app params +func (dbcfgs *DBConfigs) SetDbParams(dbaParams, appParams mysql.ConnParams) { + dbcfgs.dbaParams = dbaParams + dbcfgs.appParams = appParams +} + // NewTestDBConfigs returns a DBConfigs meant for testing. func NewTestDBConfigs(genParams, appDebugParams mysql.ConnParams, dbname string) *DBConfigs { return &DBConfigs{ diff --git a/go/vt/vtctl/vtctl.go b/go/vt/vtctl/vtctl.go index eb103f4db9d..5667715e2b6 100644 --- a/go/vt/vtctl/vtctl.go +++ b/go/vt/vtctl/vtctl.go @@ -1631,6 +1631,9 @@ func commandCreateKeyspace(ctx context.Context, wr *wrangler.Wrangler, subFlags if err != nil { return err } + if timeTime.After(time.Now()) { + return vterrors.New(vtrpcpb.Code_INVALID_ARGUMENT, "snapshot_time can not be more than current time") + } snapshotTime = logutil.TimeToProto(timeTime) } ki := &topodatapb.Keyspace{ diff --git a/go/vt/vttablet/tabletmanager/restore.go b/go/vt/vttablet/tabletmanager/restore.go index f76ca5e5259..ca7cc1fc23e 100644 --- a/go/vt/vttablet/tabletmanager/restore.go +++ b/go/vt/vttablet/tabletmanager/restore.go @@ -21,6 +21,12 @@ import ( "fmt" "time" + "vitess.io/vitess/go/vt/proto/vttime" + + "vitess.io/vitess/go/vt/vttablet/tabletmanager/vreplication" + + "vitess.io/vitess/go/vt/dbconfigs" + "vitess.io/vitess/go/vt/topo" "vitess.io/vitess/go/vt/vterrors" "vitess.io/vitess/go/vt/vttablet/tmclient" @@ -33,6 +39,7 @@ import ( "vitess.io/vitess/go/vt/mysqlctl" "vitess.io/vitess/go/vt/topo/topoproto" + binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata" topodatapb "vitess.io/vitess/go/vt/proto/topodata" vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc" ) @@ -44,6 +51,13 @@ var ( restoreFromBackup = flag.Bool("restore_from_backup", false, "(init restore parameter) will check BackupStorage for a recent backup at startup and start there") restoreConcurrency = flag.Int("restore_concurrency", 4, "(init restore parameter) how many concurrent files to restore at once") waitForBackupInterval = flag.Duration("wait_for_backup_interval", 0, "(init restore parameter) if this is greater than 0, instead of starting up empty when no backups are found, keep checking at this interval for a backup to appear") + + // Flags for PITR + binlogHost = flag.String("binlog_host", "", "(PITR restore parameter) host name of binlog server.") + binlogPort = flag.Int("binlog_port", 0, "(PITR restore parameter) port of binlog server.") + binlogUser = flag.String("binlog_user", "", "(PITR restore parameter) username of binlog server.") + binlogPwd = flag.String("binlog_password", "", "(PITR restore parameter) password of binlog server.") + timeoutForGTIDLookup = flag.Duration("pitr_gtid_lookup_timeout", 60*time.Second, "(PITR restore parameter) timeout for fetching gtid from timestamp.") ) // RestoreData is the main entry point for backup restore. @@ -126,6 +140,14 @@ func (tm *TabletManager) restoreDataLocked(ctx context.Context, logger logutil.L if backupManifest != nil { pos = backupManifest.Position } + // If SnapshotTime is set , then apply the incremental change + if keyspaceInfo.SnapshotTime != nil { + err = tm.restoreToTimeFromBinlog(ctx, pos, keyspaceInfo.SnapshotTime) + if err != nil { + log.Errorf("unable to restore to the specified time %s, error : %v", keyspaceInfo.SnapshotTime.String(), err) + return nil + } + } switch err { case nil: // Starting from here we won't be able to recover if we get stopped by a cancelled @@ -166,6 +188,200 @@ func (tm *TabletManager) restoreDataLocked(ctx context.Context, logger logutil.L return nil } +// restoreToTimeFromBinlog restores to the snapshot time of the keyspace +// currently this works with mysql based database only (as it uses mysql specific queries for restoring) +func (tm *TabletManager) restoreToTimeFromBinlog(ctx context.Context, pos mysql.Position, restoreTime *vttime.Time) error { + // validate the dependent settings + if *binlogHost == "" || *binlogPort <= 0 || *binlogUser == "" { + log.Warning("invalid binlog server setting, restoring to last available backup.") + return nil + } + + timeoutCtx, cancelFnc := context.WithTimeout(ctx, *timeoutForGTIDLookup) + defer cancelFnc() + + afterGTIDPos, beforeGTIDPos, err := tm.getGTIDFromTimestamp(timeoutCtx, pos, restoreTime.Seconds) + if err != nil { + return err + } + + if afterGTIDPos == "" && beforeGTIDPos == "" { + return vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, fmt.Sprintf("unable to fetch the GTID for the specified time - %s", restoreTime.String())) + } else if afterGTIDPos == "" && beforeGTIDPos != "" { + log.Info("no afterGTIDPos found, which implies we reached the end of all GTID events") + } + + log.Infof("going to restore upto the GTID - %s", afterGTIDPos) + // when we don't have before GTID, we will take it as current backup pos's last GTID + // this is case where someone tries to restore just to the 1st event after backup + if beforeGTIDPos == "" { + beforeGTIDPos = pos.GTIDSet.Last() + } + err = tm.catchupToGTID(timeoutCtx, afterGTIDPos, beforeGTIDPos) + if err != nil { + return vterrors.Wrapf(err, "unable to replicate upto desired GTID : %s", afterGTIDPos) + } + + return nil +} + +// getGTIDFromTimestamp computes 2 GTIDs based on restoreTime +// afterPos is the GTID of the first event at or after restoreTime. +// beforePos is the GTID of the last event before restoreTime. This is the GTID upto which replication will be applied +// afterPos can be used directly in the query `START SLAVE UNTIL SQL_BEFORE_GTIDS = ''` +// beforePos will be used to check if replication was able to catch up from the binlog server +func (tm *TabletManager) getGTIDFromTimestamp(ctx context.Context, pos mysql.Position, restoreTime int64) (afterPos string, beforePos string, err error) { + connParams := &mysql.ConnParams{ + Host: *binlogHost, + Port: *binlogPort, + Uname: *binlogUser, + } + if binlogPwd != nil && *binlogPwd != "" { + connParams.Pass = *binlogPwd + } + dbCfgs := &dbconfigs.DBConfigs{ + Host: connParams.Host, + Port: connParams.Port, + } + dbCfgs.SetDbParams(*connParams, *connParams) + vsClient := vreplication.NewReplicaConnector(connParams) + + filter := &binlogdatapb.Filter{ + Rules: []*binlogdatapb.Rule{{ + Match: "/.*", + }}, + } + + // get current lastPos of binlog server, so that if we hit that in vstream, we'll return from there + binlogConn, err := mysql.Connect(ctx, connParams) + if err != nil { + return "", "", err + } + defer binlogConn.Close() + lastPos, err := binlogConn.MasterPosition() + if err != nil { + return "", "", err + } + + gtidsChan := make(chan []string) + + go func() { + err := vsClient.VStream(ctx, mysql.EncodePosition(pos), filter, func(events []*binlogdatapb.VEvent) error { + for _, event := range events { + if event.Gtid != "" { + // check if we reached the lastPos then return + eventPos, err := mysql.DecodePosition(event.Gtid) + if err != nil { + return err + } + + if eventPos.AtLeast(lastPos) { + gtidsChan <- []string{"", beforePos} + break + } + + if event.Timestamp >= restoreTime { + afterPos = event.Gtid + gtidsChan <- []string{event.Gtid, beforePos} + break + } + beforePos = event.Gtid + } + } + return nil + }) + if err != nil { + gtidsChan <- []string{"", ""} + } + }() + defer vsClient.Close(ctx) + select { + case val := <-gtidsChan: + return val[0], val[1], nil + case <-ctx.Done(): + log.Warningf("Can't find the GTID from restore time stamp, exiting.") + return "", beforePos, vterrors.New(vtrpcpb.Code_FAILED_PRECONDITION, "unable to find GTID from the snapshot time as context timed out") + } +} + +// catchupToGTID replicates upto specified GTID from binlog server +// +// copies the data from binlog server by pointing to as replica +// waits till all events to GTID replicated +// once done, it will reset the replication +func (tm *TabletManager) catchupToGTID(ctx context.Context, afterGTIDPos string, beforeGTIDPos string) error { + var afterGTIDStr string + if afterGTIDPos != "" { + afterGTIDParsed, err := mysql.DecodePosition(afterGTIDPos) + if err != nil { + return err + } + afterGTIDStr = afterGTIDParsed.GTIDSet.Last() + } + + beforeGTIDPosParsed, err := mysql.DecodePosition(beforeGTIDPos) + if err != nil { + return err + } + + // it uses mysql specific queries here + cmds := []string{ + "STOP SLAVE FOR CHANNEL '' ", + "STOP SLAVE IO_THREAD FOR CHANNEL ''", + fmt.Sprintf("CHANGE MASTER TO MASTER_HOST='%s',MASTER_PORT=%d, MASTER_USER='%s', MASTER_AUTO_POSITION = 1;", *binlogHost, *binlogPort, *binlogUser), + } + if afterGTIDPos == "" { // when the there is no afterPos, that means need to replicate completely + cmds = append(cmds, "START SLAVE") + } else { + cmds = append(cmds, fmt.Sprintf("START SLAVE UNTIL SQL_BEFORE_GTIDS = '%s'", afterGTIDStr)) + } + + if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil { + return vterrors.Wrap(err, fmt.Sprintf("failed to restart the replication until %s GTID", afterGTIDStr)) + } + log.Infof("Waiting for position to reach", beforeGTIDPosParsed.GTIDSet.Last()) + // Could not use `agent.MysqlDaemon.WaitMasterPos` as the SLAVE thread is stopped with `START SLAVE UNTIL SQL_BEFORE_GTIDS` + // this is as per https://dev.mysql.com/doc/refman/5.6/en/start-slave.html + // We need to wait till the slave catch upto the specified afterGTIDPos + chGTIDCaughtup := make(chan bool) + go func() { + timeToWait := time.Now().Add(*timeoutForGTIDLookup) + for time.Now().Before(timeToWait) { + pos, err := tm.MysqlDaemon.MasterPosition() + if err != nil { + chGTIDCaughtup <- false + } + + if pos.AtLeast(beforeGTIDPosParsed) { + chGTIDCaughtup <- true + } + select { + case <-ctx.Done(): + chGTIDCaughtup <- false + default: + time.Sleep(300 * time.Millisecond) + } + } + }() + select { + case resp := <-chGTIDCaughtup: + if resp { + cmds := []string{ + "STOP SLAVE", + "RESET SLAVE ALL", + } + if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil { + return vterrors.Wrap(err, "failed to stop replication") + } + return nil + } + return vterrors.Wrap(err, "error while fetching the current GTID position") + case <-ctx.Done(): + log.Warningf("Could not copy till GTID.") + return vterrors.Wrapf(err, "context timeout while restoring upto specified GTID - %s", beforeGTIDPos) + } +} + func (tm *TabletManager) startReplication(ctx context.Context, pos mysql.Position, tabletType topodatapb.TabletType) error { cmds := []string{ "STOP SLAVE", diff --git a/go/vt/vttablet/tabletmanager/vreplication/replica_connector.go b/go/vt/vttablet/tabletmanager/vreplication/replica_connector.go new file mode 100644 index 00000000000..438403d8595 --- /dev/null +++ b/go/vt/vttablet/tabletmanager/vreplication/replica_connector.go @@ -0,0 +1,100 @@ +/* +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 vreplication + +import ( + "vitess.io/vitess/go/mysql" + "vitess.io/vitess/go/vt/dbconfigs" + "vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv" + + "golang.org/x/net/context" + "vitess.io/vitess/go/sqltypes" + binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata" + querypb "vitess.io/vitess/go/vt/proto/query" + vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc" + "vitess.io/vitess/go/vt/vterrors" + "vitess.io/vitess/go/vt/vttablet/tabletserver/schema" + "vitess.io/vitess/go/vt/vttablet/tabletserver/vstreamer" +) + +// NewReplicaConnector returns replica connector +// +// This is used by binlog server to make vstream connection +// using the vstream connection, it will parse the events from binglog +// to fetch the corresponding GTID for required recovery time +func NewReplicaConnector(connParams *mysql.ConnParams) *replicaConnector { + + // Construct + config := tabletenv.NewDefaultConfig() + dbCfg := &dbconfigs.DBConfigs{ + Host: connParams.Host, + Port: connParams.Port, + } + dbCfg.SetDbParams(*connParams, *connParams) + config.DB = dbCfg + c := &replicaConnector{conn: connParams} + env := tabletenv.NewEnv(config, "source") + c.se = schema.NewEngine(env) + c.se.SkipMetaCheck = true + c.vstreamer = vstreamer.NewEngine(env, nil, c.se, "") + c.se.InitDBConfig(dbconfigs.New(connParams)) + + // Open + + c.vstreamer.Open() + + return c +} + +//----------------------------------------------------------- + +type replicaConnector struct { + conn *mysql.ConnParams + se *schema.Engine + vstreamer *vstreamer.Engine +} + +func (c *replicaConnector) shutdown() { + c.vstreamer.Close() + c.se.Close() +} + +func (c *replicaConnector) Open(ctx context.Context) error { + return nil +} + +func (c *replicaConnector) Close(ctx context.Context) error { + c.shutdown() + return nil +} + +func (c *replicaConnector) VStream(ctx context.Context, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error { + return c.vstreamer.Stream(ctx, startPos, nil, filter, send) +} + +// VStreamRows streams rows from query result +func (c *replicaConnector) VStreamRows(ctx context.Context, query string, lastpk *querypb.QueryResult, send func(*binlogdatapb.VStreamRowsResponse) error) error { + var row []sqltypes.Value + if lastpk != nil { + r := sqltypes.Proto3ToResult(lastpk) + if len(r.Rows) != 1 { + return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unexpected lastpk input: %v", lastpk) + } + row = r.Rows[0] + } + return c.vstreamer.StreamRows(ctx, query, row, send) +} diff --git a/go/vt/vttablet/tabletserver/schema/engine.go b/go/vt/vttablet/tabletserver/schema/engine.go index d8de480677c..5a36ac1eb46 100644 --- a/go/vt/vttablet/tabletserver/schema/engine.go +++ b/go/vt/vttablet/tabletserver/schema/engine.go @@ -64,6 +64,9 @@ type Engine struct { notifierMu sync.Mutex notifiers map[string]notifier + // SkipMetaCheck skips the metadata about the database and table information + SkipMetaCheck bool + historian *historian conns *connpool.Pool @@ -137,9 +140,12 @@ func (se *Engine) Open() error { if err := se.reload(ctx); err != nil { return err } - if err := se.historian.Open(); err != nil { - return err + if !se.SkipMetaCheck { + if err := se.historian.Open(); err != nil { + return err + } } + se.ticks.Start(func() { if err := se.Reload(ctx); err != nil { log.Errorf("periodic schema reload failed: %v", err) @@ -245,6 +251,10 @@ func (se *Engine) reload(ctx context.Context) error { if err != nil { return err } + // if this flag is set, then we don't need table meta information + if se.SkipMetaCheck { + return nil + } tableData, err := conn.Exec(ctx, mysql.BaseShowTables, maxTableCount, false) if err != nil { return err @@ -308,7 +318,8 @@ func (se *Engine) reload(ctx context.Context) error { } func (se *Engine) mysqlTime(ctx context.Context, conn *connpool.DBConn) (int64, error) { - tm, err := conn.Exec(ctx, "select unix_timestamp()", 1, false) + // Keep `SELECT UNIX_TIMESTAMP` is in uppercase because binlog server queries are case sensitive and expect it to be so. + tm, err := conn.Exec(ctx, "SELECT UNIX_TIMESTAMP()", 1, false) if err != nil { return 0, vterrors.Errorf(vtrpcpb.Code_UNKNOWN, "could not get MySQL time: %v", err) } diff --git a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go index 445eff89464..54adf427d05 100644 --- a/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go +++ b/go/vt/vttablet/tabletserver/vstreamer/vstreamer.go @@ -471,7 +471,7 @@ func (vs *vstreamer) parseEvent(ev mysql.BinlogEvent) ([]*binlogdatapb.VEvent, e if tm.Database == "_vt" && tm.Name == "resharding_journal" { // A journal is a special case that generates a JOURNAL event. return nil, vs.buildJournalPlan(id, tm) - } else if tm.Database == "_vt" && tm.Name == "schema_version" { + } else if tm.Database == "_vt" && tm.Name == "schema_version" && !vs.se.SkipMetaCheck { // Generates a Version event when it detects that a schema is stored in the schema_version table. return nil, vs.buildVersionPlan(id, tm) } diff --git a/test/bin/rippled b/test/bin/rippled new file mode 100755 index 00000000000..c7b6bea3b95 Binary files /dev/null and b/test/bin/rippled differ diff --git a/test/config.json b/test/config.json index f5fd2f50a56..a8099ef99ac 100644 --- a/test/config.json +++ b/test/config.json @@ -8,7 +8,7 @@ "java_test" ], "Manual": false, - "Shard": 25, + "Shard": 10, "RetryMax": 0, "Tags": [] }, @@ -266,6 +266,17 @@ "site_test" ] }, + "pitr": { + "File": "unused.go", + "Args": ["vitess.io/vitess/go/test/endtoend/recovery/pitr"], + "Command": [], + "Manual": false, + "Shard": 10, + "RetryMax": 0, + "Tags": [ + "site_test" + ] + }, "recovery": { "File": "unused.go", "Args": ["vitess.io/vitess/go/test/endtoend/recovery/unshardedrecovery"],