From f6efaadd87db2553cfb92722dd1d2855646a1b43 Mon Sep 17 00:00:00 2001 From: Sugu Sougoumarane Date: Mon, 23 Sep 2019 15:30:18 -0700 Subject: [PATCH 1/4] migrater: option to cancel Signed-off-by: Sugu Sougoumarane --- go/vt/vtctl/vtctl.go | 5 +- go/vt/wrangler/migrater.go | 15 ++++-- go/vt/wrangler/migrater_test.go | 71 ++++++++++++++++++++++---- go/vt/wrangler/stream_migrater.go | 14 ++--- go/vt/wrangler/stream_migrater_test.go | 24 ++++----- 5 files changed, 94 insertions(+), 35 deletions(-) diff --git a/go/vt/vtctl/vtctl.go b/go/vt/vtctl/vtctl.go index e484117eb6a..12dc6d1387c 100644 --- a/go/vt/vtctl/vtctl.go +++ b/go/vt/vtctl/vtctl.go @@ -327,7 +327,7 @@ var commands = []commandGroup{ "[-cells=c1,c2,...] [-reverse] -workflow=workflow ", "Migrate read traffic for the specified workflow."}, {"MigrateWrites", commandMigrateWrites, - "[-filtered_replication_wait_time=30s] [-reverse_replication=] -workflow=workflow ", + "[-filtered_replication_wait_time=30s] [-cancel] [-reverse_replication=false] -workflow=workflow ", "Migrate write traffic for the specified workflow."}, {"CancelResharding", commandCancelResharding, "", @@ -1920,6 +1920,7 @@ func commandMigrateReads(ctx context.Context, wr *wrangler.Wrangler, subFlags *f func commandMigrateWrites(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag.FlagSet, args []string) error { filteredReplicationWaitTime := subFlags.Duration("filtered_replication_wait_time", 30*time.Second, "Specifies the maximum time to wait, in seconds, for filtered replication to catch up on master migrations. The migration will be aborted on timeout.") reverseReplication := subFlags.Bool("reverse_replication", true, "Also reverse the replication") + cancelMigrate := subFlags.Bool("cancel", false, "Cancel the failed migration and serve from source") workflow := subFlags.String("workflow", "", "Specifies the workflow name") if err := subFlags.Parse(args); err != nil { return err @@ -1932,7 +1933,7 @@ func commandMigrateWrites(ctx context.Context, wr *wrangler.Wrangler, subFlags * if *workflow == "" { return fmt.Errorf("a -workflow=workflow argument is required") } - journalID, err := wr.MigrateWrites(ctx, keyspace, *workflow, *filteredReplicationWaitTime, *reverseReplication) + journalID, err := wr.MigrateWrites(ctx, keyspace, *workflow, *filteredReplicationWaitTime, *cancelMigrate, *reverseReplication) if err != nil { return err } diff --git a/go/vt/wrangler/migrater.go b/go/vt/wrangler/migrater.go index 9addbda942f..f756dcbc0c0 100644 --- a/go/vt/wrangler/migrater.go +++ b/go/vt/wrangler/migrater.go @@ -138,7 +138,7 @@ func (wr *Wrangler) MigrateReads(ctx context.Context, targetKeyspace, workflow s } // MigrateWrites is a generic way of migrating write traffic for a resharding workflow. -func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow string, filteredReplicationWaitTime time.Duration, reverseReplication bool) (journalID int64, err error) { +func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow string, filteredReplicationWaitTime time.Duration, cancelMigrate, reverseReplication bool) (journalID int64, err error) { mi, err := wr.buildMigrater(ctx, targetKeyspace, workflow) if err != nil { wr.Logger().Errorf("buildMigrater failed: %v", err) @@ -189,6 +189,11 @@ func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow mi.wr.Logger().Errorf("buildStreamMigrater failed: %v", err) return 0, err } + if cancelMigrate { + mi.wr.Logger().Infof("Cancel was requested.") + mi.cancelMigration(ctx, sm) + return 0, nil + } sourceWorkflows, err = sm.stopStreams(ctx) if err != nil { mi.wr.Logger().Errorf("stopStreams failed: %v", err) @@ -216,6 +221,11 @@ func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow return 0, err } } else { + if cancelMigrate { + err := fmt.Errorf("migration has reached the point of no return, cannot cancel") + mi.wr.Logger().Errorf("%v", err) + return 0, err + } mi.wr.Logger().Infof("Journals were found. Completing the left over steps.") // Need to gather positions in case all journals were not created. if err := mi.gatherPositions(ctx); err != nil { @@ -237,8 +247,7 @@ func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow mi.wr.Logger().Errorf("changeRouting failed: %v", err) return 0, err } - sm := &streamMigrater{mi: mi} - if err := sm.finalize(ctx, sourceWorkflows); err != nil { + if err := streamMigraterfinalize(ctx, mi, sourceWorkflows); err != nil { mi.wr.Logger().Errorf("finalize failed: %v", err) return 0, err } diff --git a/go/vt/wrangler/migrater_test.go b/go/vt/wrangler/migrater_test.go index 29d2c30625b..b27c2fed478 100644 --- a/go/vt/wrangler/migrater_test.go +++ b/go/vt/wrangler/migrater_test.go @@ -234,7 +234,7 @@ func TestTableMigrateMainflow(t *testing.T) { //------------------------------------------------------------------------------------------------------------------- // Can't migrate writes if REPLICA and RDONLY have not fully migrated yet. - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want = "missing tablet type specific routing, read-only traffic must be migrated before migrating writes" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -300,7 +300,7 @@ func TestTableMigrateMainflow(t *testing.T) { } cancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, false, true) want = "DeadlineExceeded" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites(0 timeout) err: %v, must contain %v", err, want) @@ -415,7 +415,7 @@ func TestTableMigrateMainflow(t *testing.T) { } deleteTargetVReplication() - journalID, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + journalID, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) if err != nil { t.Fatal(err) } @@ -548,7 +548,7 @@ func TestShardMigrateMainflow(t *testing.T) { //------------------------------------------------------------------------------------------------------------------- // Can't migrate writes if REPLICA and RDONLY have not fully migrated yet. - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want = "cannot migrate MASTER away" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -610,7 +610,7 @@ func TestShardMigrateMainflow(t *testing.T) { } cancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, false, true) want = "DeadlineExceeded" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites(0 timeout) err: %v, must contain %v", err, want) @@ -707,7 +707,7 @@ func TestShardMigrateMainflow(t *testing.T) { } deleteTargetVReplication() - journalID, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + journalID, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) if err != nil { t.Fatal(err) } @@ -808,7 +808,7 @@ func TestMigrateFailJournal(t *testing.T) { tme.dbSourceClients[0].addQueryRE("insert into _vt.resharding_journal", nil, errors.New("journaling intentionally failed")) tme.dbSourceClients[1].addQueryRE("insert into _vt.resharding_journal", nil, errors.New("journaling intentionally failed")) - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "journaling intentionally failed" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites(0 timeout) err: %v, must contain %v", err, want) @@ -875,7 +875,7 @@ func TestTableMigrateJournalExists(t *testing.T) { tme.dbTargetClients[1].addQuery("delete from _vt.vreplication where id in (1, 2)", &sqltypes.Result{}, nil) tme.dbTargetClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (1, 2)", &sqltypes.Result{}, nil) - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) if err != nil { t.Fatal(err) } @@ -945,7 +945,7 @@ func TestShardMigrateJournalExists(t *testing.T) { tme.dbTargetClients[1].addQuery("delete from _vt.vreplication where id in (2)", &sqltypes.Result{}, nil) tme.dbTargetClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (2)", &sqltypes.Result{}, nil) - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) if err != nil { t.Fatal(err) } @@ -963,6 +963,55 @@ func TestShardMigrateJournalExists(t *testing.T) { verifyQueries(t, tme.allDBClients) } +func TestTableMigrateCancel(t *testing.T) { + ctx := context.Background() + tme := newTestTableMigrater(ctx, t) + defer tme.stopTablets(t) + + err := tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_RDONLY, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + err = tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_REPLICA, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + + checkJournals := func() { + tme.dbSourceClients[0].addQuery("select val from _vt.resharding_journal where id=7672494164556733923", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQuery("select val from _vt.resharding_journal where id=7672494164556733923", &sqltypes.Result{}, nil) + } + checkJournals() + + deleteReverseReplicaion := func() { + tme.dbSourceClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks1' and workflow = 'test_reverse'", resultid34, nil) + tme.dbSourceClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks1' and workflow = 'test_reverse'", resultid34, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.vreplication where id in (3, 4)", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQuery("delete from _vt.vreplication where id in (3, 4)", &sqltypes.Result{}, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (3, 4)", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (3, 4)", &sqltypes.Result{}, nil) + } + cancelMigration := func() { + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid12, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid12, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set state = 'Running', message = '' where id in (1, 2)", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("update _vt.vreplication set state = 'Running', message = '' where id in (1, 2)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 1", runningResult(1), nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 2", runningResult(2), nil) + tme.dbTargetClients[1].addQuery("select * from _vt.vreplication where id = 1", runningResult(1), nil) + tme.dbTargetClients[1].addQuery("select * from _vt.vreplication where id = 2", runningResult(2), nil) + + deleteReverseReplicaion() + } + cancelMigration() + + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true, false) + if err != nil { + t.Fatal(err) + } + verifyQueries(t, tme.allDBClients) +} + func TestTableMigrateNoReverse(t *testing.T) { ctx := context.Background() tme := newTestTableMigrater(ctx, t) @@ -1061,7 +1110,7 @@ func TestTableMigrateNoReverse(t *testing.T) { } deleteTargetVReplication() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, false) if err != nil { t.Fatal(err) } @@ -1124,7 +1173,7 @@ func TestMigrateFrozen(t *testing.T) { } deleteTargetVReplication() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 0*time.Second, false, true) if err != nil { t.Fatal(err) } diff --git a/go/vt/wrangler/stream_migrater.go b/go/vt/wrangler/stream_migrater.go index f4737eedab0..9d6821b2b44 100644 --- a/go/vt/wrangler/stream_migrater.go +++ b/go/vt/wrangler/stream_migrater.go @@ -551,24 +551,24 @@ func (sm *streamMigrater) deleteTargetStreams(ctx context.Context) error { return err } -// finalize performs the final cleanup: start all the newly migrated target streams -// and delete them from the source. -func (sm *streamMigrater) finalize(ctx context.Context, workflows []string) error { +// streamMigraterFinalize finalizes the stream migration. +// It's a standalone function because it does not use the streamMigrater state. +func streamMigraterfinalize(ctx context.Context, mi *migrater, workflows []string) error { if len(workflows) == 0 { return nil } workflowList := stringListify(workflows) - err := sm.mi.forAllSources(func(source *miSource) error { + err := mi.forAllSources(func(source *miSource) error { query := fmt.Sprintf("delete from _vt.vreplication where db_name=%s and workflow in (%s)", encodeString(source.master.DbName()), workflowList) - _, err := sm.mi.wr.VReplicationExec(ctx, source.master.Alias, query) + _, err := mi.wr.VReplicationExec(ctx, source.master.Alias, query) return err }) if err != nil { return err } - err = sm.mi.forAllTargets(func(target *miTarget) error { + err = mi.forAllTargets(func(target *miTarget) error { query := fmt.Sprintf("update _vt.vreplication set state='Running' where db_name=%s and workflow in (%s)", encodeString(target.master.DbName()), workflowList) - _, err := sm.mi.wr.VReplicationExec(ctx, target.master.Alias, query) + _, err := mi.wr.VReplicationExec(ctx, target.master.Alias, query) return err }) return err diff --git a/go/vt/wrangler/stream_migrater_test.go b/go/vt/wrangler/stream_migrater_test.go index 7e97aac9170..c50b5b03bb6 100644 --- a/go/vt/wrangler/stream_migrater_test.go +++ b/go/vt/wrangler/stream_migrater_test.go @@ -163,7 +163,7 @@ func TestStreamMigrateMainflow(t *testing.T) { tme.expectStartReverseVReplication() tme.expectDeleteTargetVReplication() - if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true); err != nil { + if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true); err != nil { t.Fatal(err) } @@ -330,7 +330,7 @@ func TestStreamMigrateTwoStreams(t *testing.T) { tme.expectStartReverseVReplication() tme.expectDeleteTargetVReplication() - if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true); err != nil { + if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true); err != nil { t.Fatal(err) } @@ -462,7 +462,7 @@ func TestStreamMigrateOneToMany(t *testing.T) { tme.expectStartReverseVReplication() tme.expectDeleteTargetVReplication() - if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true); err != nil { + if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true); err != nil { t.Fatal(err) } @@ -596,7 +596,7 @@ func TestStreamMigrateManyToOne(t *testing.T) { tme.expectStartReverseVReplication() tme.expectDeleteTargetVReplication() - if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true); err != nil { + if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true); err != nil { t.Fatal(err) } @@ -784,7 +784,7 @@ func TestStreamMigrateSyncSuccess(t *testing.T) { tme.expectStartReverseVReplication() tme.expectDeleteTargetVReplication() - if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true); err != nil { + if _, err := tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true); err != nil { t.Fatal(err) } @@ -939,7 +939,7 @@ func TestStreamMigrateSyncFail(t *testing.T) { tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "does not match" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, want %s", err, want) @@ -1030,7 +1030,7 @@ func TestStreamMigrateCancel(t *testing.T) { } cancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "intentionally failed" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, want %s", err, want) @@ -1100,7 +1100,7 @@ func TestStreamMigrateStoppedStreams(t *testing.T) { stopStreams() tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "cannot migrate until all strems are running: 0" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -1161,7 +1161,7 @@ func TestStreamMigrateStillCopying(t *testing.T) { stopStreams() tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "cannot migrate while vreplication streams in source shards are still copying: 0" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -1221,7 +1221,7 @@ func TestStreamMigrateEmptyWorflow(t *testing.T) { stopStreams() tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "VReplication streams must have named workflows for migration: shard: ks:0, stream: 1" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -1281,7 +1281,7 @@ func TestStreamMigrateDupWorflow(t *testing.T) { stopStreams() tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "VReplication stream has the same workflow name as the resharding workflow: shard: ks:0, stream: 1" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) @@ -1352,7 +1352,7 @@ func TestStreamMigrateStreamsMismatch(t *testing.T) { stopStreams() tme.expectCancelMigration() - _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true) + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "streams are mismatched across source shards" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, must contain %v", err, want) From 662b24212bd07c9deb9f1b8d0613f9916ec6215a Mon Sep 17 00:00:00 2001 From: Sugu Sougoumarane Date: Mon, 23 Sep 2019 19:03:55 -0700 Subject: [PATCH 2/4] migrater: tests for unsharded tables Signed-off-by: Sugu Sougoumarane --- go/vt/wrangler/migrater.go | 2 + go/vt/wrangler/migrater_env_test.go | 47 +++++-- go/vt/wrangler/migrater_test.go | 168 +++++++++++++++++++++++++ go/vt/wrangler/stream_migrater.go | 10 +- go/vt/wrangler/stream_migrater_test.go | 28 ++--- 5 files changed, 228 insertions(+), 27 deletions(-) diff --git a/go/vt/wrangler/migrater.go b/go/vt/wrangler/migrater.go index f756dcbc0c0..584c78d51e3 100644 --- a/go/vt/wrangler/migrater.go +++ b/go/vt/wrangler/migrater.go @@ -633,9 +633,11 @@ func (mi *migrater) waitForCatchup(ctx context.Context, filteredReplicationWaitT return mi.forAllUids(func(target *miTarget, uid uint32) error { bls := target.sources[uid] source := mi.sources[bls.Shard] + mi.wr.Logger().Infof("waiting for keyspace:shard: %v:%v, position %v", mi.targetKeyspace, target.si.ShardName(), source.position) if err := mi.wr.tmc.VReplicationWaitForPos(ctx, target.master.Tablet, int(uid), source.position); err != nil { return err } + mi.wr.Logger().Infof("position for keyspace:shard: %v:%v reached", mi.targetKeyspace, target.si.ShardName()) if _, err := mi.wr.tmc.VReplicationExec(ctx, target.master.Tablet, binlogplayer.StopVReplication(uid, "stopped for cutover")); err != nil { return err } diff --git a/go/vt/wrangler/migrater_env_test.go b/go/vt/wrangler/migrater_env_test.go index 4829f41b1d7..f2880c228f9 100644 --- a/go/vt/wrangler/migrater_env_test.go +++ b/go/vt/wrangler/migrater_env_test.go @@ -49,6 +49,10 @@ type testMigraterEnv struct { dbTargetClients []*fakeDBClient allDBClients []*fakeDBClient targetKeyspace string + sourceShards []string + targetShards []string + sourceKeyRanges []*topodatapb.KeyRange + targetKeyRanges []*topodatapb.KeyRange } // testShardMigraterEnv has some convenience functions for adding expected queries. @@ -56,26 +60,45 @@ type testMigraterEnv struct { // Use explicit queries for testing the actual shard migration. type testShardMigraterEnv struct { testMigraterEnv - sourceShards, targetShards []string - sourceKeyRanges, targetKeyRanges []*topodatapb.KeyRange } func newTestTableMigrater(ctx context.Context, t *testing.T) *testMigraterEnv { + return newTestTableMigraterCustom(ctx, t, []string{"-40", "40-"}, []string{"-80", "80-"}) +} + +func newTestTableMigraterCustom(ctx context.Context, t *testing.T, sourceShards, targetShards []string) *testMigraterEnv { tme := &testMigraterEnv{} tme.ts = memorytopo.NewServer("cell1", "cell2") tme.wr = New(logutil.NewConsoleLogger(), tme.ts, tmclient.NewTabletManagerClient()) - - sourceShards := []string{"-40", "40-"} - targetShards := []string{"-80", "80-"} + tme.sourceShards = sourceShards + tme.targetShards = targetShards tabletID := 10 for _, shard := range sourceShards { tme.sourceMasters = append(tme.sourceMasters, newFakeTablet(t, tme.wr, "cell1", uint32(tabletID), topodatapb.TabletType_MASTER, nil, TabletKeyspaceShard(t, "ks1", shard))) tabletID += 10 + + _, sourceKeyRange, err := topo.ValidateShardName(shard) + if err != nil { + t.Fatal(err) + } + if sourceKeyRange == nil { + sourceKeyRange = &topodatapb.KeyRange{} + } + tme.sourceKeyRanges = append(tme.sourceKeyRanges, sourceKeyRange) } for _, shard := range targetShards { tme.targetMasters = append(tme.targetMasters, newFakeTablet(t, tme.wr, "cell1", uint32(tabletID), topodatapb.TabletType_MASTER, nil, TabletKeyspaceShard(t, "ks2", shard))) tabletID += 10 + + _, targetKeyRange, err := topo.ValidateShardName(shard) + if err != nil { + t.Fatal(err) + } + if targetKeyRange == nil { + targetKeyRange = &topodatapb.KeyRange{} + } + tme.targetKeyRanges = append(tme.targetKeyRanges, targetKeyRange) } vs := &vschemapb.Keyspace{ @@ -100,11 +123,15 @@ func newTestTableMigrater(ctx context.Context, t *testing.T) *testMigraterEnv { }, }, } - if err := tme.ts.SaveVSchema(ctx, "ks1", vs); err != nil { - t.Fatal(err) + if len(sourceShards) != 1 { + if err := tme.ts.SaveVSchema(ctx, "ks1", vs); err != nil { + t.Fatal(err) + } } - if err := tme.ts.SaveVSchema(ctx, "ks2", vs); err != nil { - t.Fatal(err) + if len(targetShards) != 1 { + if err := tme.ts.SaveVSchema(ctx, "ks2", vs); err != nil { + t.Fatal(err) + } } if err := tme.ts.RebuildSrvVSchema(ctx, nil); err != nil { t.Fatal(err) @@ -396,6 +423,8 @@ func (tme *testShardMigraterEnv) expectCreateJournals() { } func (tme *testShardMigraterEnv) expectStartReverseVReplication() { + // NOTE: this is not a faithful reproduction of what should happen. + // The ids returned are not accurate. for _, dbclient := range tme.dbSourceClients { dbclient.addQuery("select id from _vt.vreplication where db_name = 'vt_ks'", resultid34, nil) dbclient.addQuery("update _vt.vreplication set state = 'Running', message = '' where id in (3, 4)", &sqltypes.Result{}, nil) diff --git a/go/vt/wrangler/migrater_test.go b/go/vt/wrangler/migrater_test.go index b27c2fed478..5231d7eb64e 100644 --- a/go/vt/wrangler/migrater_test.go +++ b/go/vt/wrangler/migrater_test.go @@ -728,6 +728,174 @@ func TestShardMigrateMainflow(t *testing.T) { checkIsMasterServing(t, tme.ts, "ks:80-", true) } +func TestTableMigrateOneToMany(t *testing.T) { + ctx := context.Background() + tme := newTestTableMigraterCustom(ctx, t, []string{"0"}, []string{"-80", "80-"}) + defer tme.stopTablets(t) + + err := tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_RDONLY, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + err = tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_REPLICA, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + + // checkJournals + tme.dbSourceClients[0].addQueryRE("select val from _vt.resharding_journal.*", &sqltypes.Result{}, nil) + + waitForCatchup := func() { + // mi.waitForCatchup-> mi.wr.tmc.VReplicationWaitForPos + state := sqltypes.MakeTestResult(sqltypes.MakeTestFields( + "pos|state|message", + "varchar|varchar|varchar"), + "MariaDB/5-456-892|Running", + ) + tme.dbTargetClients[0].addQuery("select pos, state, message from _vt.vreplication where id=1", state, nil) + tme.dbTargetClients[1].addQuery("select pos, state, message from _vt.vreplication where id=1", state, nil) + + // mi.waitForCatchup-> mi.wr.tmc.VReplicationExec('Stopped') + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where id = 1", resultid1, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set state = 'Stopped', message = 'stopped for cutover' where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where id = 1", resultid1, nil) + tme.dbTargetClients[1].addQuery("update _vt.vreplication set state = 'Stopped', message = 'stopped for cutover' where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbTargetClients[1].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + } + waitForCatchup() + + deleteReverseReplicaion := func() { + tme.dbSourceClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks1' and workflow = 'test_reverse'", resultid34, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.vreplication where id in (3, 4)", &sqltypes.Result{}, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (3, 4)", &sqltypes.Result{}, nil) + } + + createReverseVReplication := func() { + deleteReverseReplicaion() + + tme.dbSourceClients[0].addQueryRE(`insert into _vt.vreplication.*test_reverse.*ks2.*-80.*t1.*from t1\\".*t2.*from t2\\"`, &sqltypes.Result{InsertID: 1}, nil) + tme.dbSourceClients[0].addQueryRE(`insert into _vt.vreplication.*test_reverse.*ks2.*80-.*t1.*from t1\\".*t2.*from t2\\"`, &sqltypes.Result{InsertID: 2}, nil) + tme.dbSourceClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbSourceClients[0].addQuery("select * from _vt.vreplication where id = 2", stoppedResult(2), nil) + } + createReverseVReplication() + + createJournals := func() { + journal1 := "insert into _vt.resharding_journal.*tables.*t1.*t2.*local_position.*MariaDB/5-456-892.*shard_gtids.*-80.*MariaDB/5-456-893.*80.*MariaDB/5-456-893.*participants.*0" + tme.dbSourceClients[0].addQueryRE(journal1, &sqltypes.Result{}, nil) + } + createJournals() + + deleteTargetVReplication := func() { + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid1, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid1, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set message = 'FROZEN' where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbTargetClients[1].addQuery("update _vt.vreplication set message = 'FROZEN' where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid1, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid1, nil) + tme.dbTargetClients[0].addQuery("delete from _vt.vreplication where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("delete from _vt.vreplication where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (1)", &sqltypes.Result{}, nil) + } + deleteTargetVReplication() + + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, false) + if err != nil { + t.Fatal(err) + } + verifyQueries(t, tme.allDBClients) +} + +func TestTableMigrateManyToOne(t *testing.T) { + ctx := context.Background() + tme := newTestTableMigraterCustom(ctx, t, []string{"-80", "80-"}, []string{"0"}) + defer tme.stopTablets(t) + + err := tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_RDONLY, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + err = tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_REPLICA, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + + checkJournals := func() { + tme.dbSourceClients[0].addQueryRE("select val from _vt.resharding_journal.*", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQueryRE("select val from _vt.resharding_journal.*", &sqltypes.Result{}, nil) + } + checkJournals() + + waitForCatchup := func() { + // mi.waitForCatchup-> mi.wr.tmc.VReplicationWaitForPos + state := sqltypes.MakeTestResult(sqltypes.MakeTestFields( + "pos|state|message", + "varchar|varchar|varchar"), + "MariaDB/5-456-892|Running", + ) + tme.dbTargetClients[0].addQuery("select pos, state, message from _vt.vreplication where id=1", state, nil) + tme.dbTargetClients[0].addQuery("select pos, state, message from _vt.vreplication where id=2", state, nil) + + // mi.waitForCatchup-> mi.wr.tmc.VReplicationExec('Stopped') + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where id = 1", resultid1, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set state = 'Stopped', message = 'stopped for cutover' where id in (1)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where id = 2", resultid2, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set state = 'Stopped', message = 'stopped for cutover' where id in (2)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 2", stoppedResult(2), nil) + } + waitForCatchup() + + deleteReverseReplicaion := func() { + tme.dbSourceClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks1' and workflow = 'test_reverse'", resultid3, nil) + tme.dbSourceClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks1' and workflow = 'test_reverse'", resultid3, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.vreplication where id in (3)", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQuery("delete from _vt.vreplication where id in (3)", &sqltypes.Result{}, nil) + tme.dbSourceClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (3)", &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (3)", &sqltypes.Result{}, nil) + } + + createReverseVReplication := func() { + deleteReverseReplicaion() + + tme.dbSourceClients[0].addQueryRE(`insert into _vt.vreplication.*test_reverse.*ks2.*0.*t1.*in_keyrange.*c1.*hash.*-80.*t2.*in_keyrange.*c1.*-80`, &sqltypes.Result{InsertID: 1}, nil) + tme.dbSourceClients[1].addQueryRE(`insert into _vt.vreplication.*test_reverse.*ks2.*0.*t1.*in_keyrange.*c1.*hash.*80-.*t2.*in_keyrange.*c1.*80-`, &sqltypes.Result{InsertID: 1}, nil) + tme.dbSourceClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbSourceClients[1].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + } + createReverseVReplication() + + createJournals := func() { + journal := "insert into _vt.resharding_journal.*shard_gtids.*ks2.*0.*participants.*ks1.*80.*participants.*80" + tme.dbSourceClients[0].addQueryRE(journal, &sqltypes.Result{}, nil) + tme.dbSourceClients[1].addQueryRE(journal, &sqltypes.Result{}, nil) + } + createJournals() + + deleteTargetVReplication := func() { + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid12, nil) + tme.dbTargetClients[0].addQuery("update _vt.vreplication set message = 'FROZEN' where id in (1, 2)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 1", stoppedResult(1), nil) + tme.dbTargetClients[0].addQuery("select * from _vt.vreplication where id = 2", stoppedResult(2), nil) + + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks2' and workflow = 'test'", resultid12, nil) + tme.dbTargetClients[0].addQuery("delete from _vt.vreplication where id in (1, 2)", &sqltypes.Result{}, nil) + tme.dbTargetClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (1, 2)", &sqltypes.Result{}, nil) + } + deleteTargetVReplication() + + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, false) + if err != nil { + t.Fatal(err) + } + verifyQueries(t, tme.allDBClients) +} + // TestMigrateFailJournal tests that cancel doesn't get called after point of no return. // No need to test this for shard migrate because code paths are the same. func TestMigrateFailJournal(t *testing.T) { diff --git a/go/vt/wrangler/stream_migrater.go b/go/vt/wrangler/stream_migrater.go index 9d6821b2b44..35b10a4c15f 100644 --- a/go/vt/wrangler/stream_migrater.go +++ b/go/vt/wrangler/stream_migrater.go @@ -258,7 +258,7 @@ func (sm *streamMigrater) syncSourceStreams(ctx context.Context) (map[string]mys } var wg sync.WaitGroup allErrors := &concurrency.AllErrorRecorder{} - for _, tabletStreams := range sm.streams { + for shard, tabletStreams := range sm.streams { for _, vrs := range tabletStreams { key := fmt.Sprintf("%s:%s", vrs.bls.Keyspace, vrs.bls.Shard) pos := stopPositions[key] @@ -266,9 +266,9 @@ func (sm *streamMigrater) syncSourceStreams(ctx context.Context) (map[string]mys continue } wg.Add(1) - go func(vrs *vrStream) { + go func(vrs *vrStream, shard string) { defer wg.Done() - si, err := sm.mi.wr.ts.GetShard(ctx, vrs.bls.Keyspace, vrs.bls.Shard) + si, err := sm.mi.wr.ts.GetShard(ctx, sm.mi.sourceKeyspace, vrs.bls.Shard) if err != nil { allErrors.RecordError(err) return @@ -283,11 +283,13 @@ func (sm *streamMigrater) syncSourceStreams(ctx context.Context) (map[string]mys allErrors.RecordError(err) return } + sm.mi.wr.Logger().Infof("waiting for keyspace:shard: %v:%v, position %v", sm.mi.sourceKeyspace, shard, pos) if err := sm.mi.wr.tmc.VReplicationWaitForPos(ctx, master.Tablet, int(vrs.id), mysql.EncodePosition(pos)); err != nil { allErrors.RecordError(err) return } - }(vrs) + sm.mi.wr.Logger().Infof("position for keyspace:shard: %v:%v reached", sm.mi.sourceKeyspace, shard) + }(vrs, shard) } } wg.Wait() diff --git a/go/vt/wrangler/stream_migrater_test.go b/go/vt/wrangler/stream_migrater_test.go index c50b5b03bb6..ae8748950b0 100644 --- a/go/vt/wrangler/stream_migrater_test.go +++ b/go/vt/wrangler/stream_migrater_test.go @@ -62,7 +62,7 @@ func TestStreamMigrateMainflow(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -208,7 +208,7 @@ func TestStreamMigrateTwoStreams(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -224,7 +224,7 @@ func TestStreamMigrateTwoStreams(t *testing.T) { } for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -374,7 +374,7 @@ func TestStreamMigrateOneToMany(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -506,7 +506,7 @@ func TestStreamMigrateManyToOne(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -638,7 +638,7 @@ func TestStreamMigrateSyncSuccess(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -671,7 +671,7 @@ func TestStreamMigrateSyncSuccess(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -828,7 +828,7 @@ func TestStreamMigrateSyncFail(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -974,7 +974,7 @@ func TestStreamMigrateCancel(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -1073,7 +1073,7 @@ func TestStreamMigrateStoppedStreams(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -1134,7 +1134,7 @@ func TestStreamMigrateStillCopying(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -1195,7 +1195,7 @@ func TestStreamMigrateEmptyWorflow(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -1255,7 +1255,7 @@ func TestStreamMigrateDupWorflow(t *testing.T) { var rows []string for j, sourceShard := range tme.sourceShards { bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ @@ -1321,7 +1321,7 @@ func TestStreamMigrateStreamsMismatch(t *testing.T) { continue } bls := &binlogdatapb.BinlogSource{ - Keyspace: "ks", + Keyspace: "ks1", Shard: sourceShard, Filter: &binlogdatapb.Filter{ Rules: []*binlogdatapb.Rule{{ From 0d2934fe595cb89e4fedf05c21be0e083f32abf0 Mon Sep 17 00:00:00 2001 From: Sugu Sougoumarane Date: Tue, 24 Sep 2019 13:25:18 -0700 Subject: [PATCH 3/4] migrater: cancel skips checking of stopped streams Signed-off-by: Sugu Sougoumarane --- go/vt/wrangler/migrater.go | 2 +- go/vt/wrangler/migrater_env_test.go | 1 - go/vt/wrangler/migrater_test.go | 2 +- go/vt/wrangler/stream_migrater.go | 42 +++++----- go/vt/wrangler/stream_migrater_test.go | 111 +++++++++++++++++-------- 5 files changed, 99 insertions(+), 59 deletions(-) diff --git a/go/vt/wrangler/migrater.go b/go/vt/wrangler/migrater.go index 584c78d51e3..f2439e82d10 100644 --- a/go/vt/wrangler/migrater.go +++ b/go/vt/wrangler/migrater.go @@ -184,7 +184,7 @@ func (wr *Wrangler) MigrateWrites(ctx context.Context, targetKeyspace, workflow } if !journalsExist { mi.wr.Logger().Infof("No previous journals were found. Proceeding normally.") - sm, err := buildStreamMigrater(ctx, mi) + sm, err := buildStreamMigrater(ctx, mi, cancelMigrate) if err != nil { mi.wr.Logger().Errorf("buildStreamMigrater failed: %v", err) return 0, err diff --git a/go/vt/wrangler/migrater_env_test.go b/go/vt/wrangler/migrater_env_test.go index f2880c228f9..7963500d662 100644 --- a/go/vt/wrangler/migrater_env_test.go +++ b/go/vt/wrangler/migrater_env_test.go @@ -453,7 +453,6 @@ func (tme *testShardMigraterEnv) expectCancelMigration() { dbclient.addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow = 'test'", &sqltypes.Result{}, nil) } for _, dbclient := range tme.dbSourceClients { - dbclient.addQuery("select id, workflow, source, pos from _vt.vreplication where db_name='vt_ks' and workflow != 'test_reverse'", &sqltypes.Result{}, nil) dbclient.addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow != 'test_reverse'", &sqltypes.Result{}, nil) } tme.expectDeleteReverseVReplication() diff --git a/go/vt/wrangler/migrater_test.go b/go/vt/wrangler/migrater_test.go index 5231d7eb64e..ea445306303 100644 --- a/go/vt/wrangler/migrater_test.go +++ b/go/vt/wrangler/migrater_test.go @@ -782,7 +782,7 @@ func TestTableMigrateOneToMany(t *testing.T) { createReverseVReplication() createJournals := func() { - journal1 := "insert into _vt.resharding_journal.*tables.*t1.*t2.*local_position.*MariaDB/5-456-892.*shard_gtids.*-80.*MariaDB/5-456-893.*80.*MariaDB/5-456-893.*participants.*0" + journal1 := "insert into _vt.resharding_journal.*tables.*t1.*t2.*local_position.*MariaDB/5-456-892.*shard_gtids.*80.*MariaDB/5-456-893.*80.*MariaDB/5-456-893.*participants.*0" tme.dbSourceClients[0].addQueryRE(journal1, &sqltypes.Result{}, nil) } createJournals() diff --git a/go/vt/wrangler/stream_migrater.go b/go/vt/wrangler/stream_migrater.go index 35b10a4c15f..c2f1b2c768d 100644 --- a/go/vt/wrangler/stream_migrater.go +++ b/go/vt/wrangler/stream_migrater.go @@ -54,13 +54,13 @@ type vrStream struct { pos mysql.Position } -func buildStreamMigrater(ctx context.Context, mi *migrater) (*streamMigrater, error) { +func buildStreamMigrater(ctx context.Context, mi *migrater, cancelMigrate bool) (*streamMigrater, error) { sm := &streamMigrater{mi: mi} if sm.mi.migrationType == binlogdatapb.MigrationType_TABLES { // Source streams should be stopped only for shard migrations. return sm, nil } - streams, err := sm.readSourceStreams(ctx) + streams, err := sm.readSourceStreams(ctx, cancelMigrate) if err != nil { return nil, err } @@ -78,27 +78,29 @@ func buildStreamMigrater(ctx context.Context, mi *migrater) (*streamMigrater, er return sm, nil } -func (sm *streamMigrater) readSourceStreams(ctx context.Context) (map[string][]*vrStream, error) { +func (sm *streamMigrater) readSourceStreams(ctx context.Context, cancelMigrate bool) (map[string][]*vrStream, error) { streams := make(map[string][]*vrStream) var mu sync.Mutex err := sm.mi.forAllSources(func(source *miSource) error { - // This flow protects us from the following scenario: When we create streams, - // we always do it in two phases. We start them off as Stopped, and then - // update them to Running. If such an operation fails, we may be left with - // lingering Stopped streams. They should actually be cleaned up by the user. - // In the current workflow, we stop streams and restart them. - // Once existing streams are stopped, there will be confusion about which of - // them can be restarted because they will be no different from the lingering streams. - // To prevent this confusion, we first check if there are any stopped streams. - // If so, we request the operator to clean them up, or restart them before going ahead. - // This allows us to assume that all stopped streams can be safely restarted - // if we cancel the operation. - stoppedStreams, err := sm.readTabletStreams(ctx, source.master, "state = 'Stopped'") - if err != nil { - return err - } - if len(stoppedStreams) != 0 { - return fmt.Errorf("cannot migrate until all strems are running: %s", source.si.ShardName()) + if !cancelMigrate { + // This flow protects us from the following scenario: When we create streams, + // we always do it in two phases. We start them off as Stopped, and then + // update them to Running. If such an operation fails, we may be left with + // lingering Stopped streams. They should actually be cleaned up by the user. + // In the current workflow, we stop streams and restart them. + // Once existing streams are stopped, there will be confusion about which of + // them can be restarted because they will be no different from the lingering streams. + // To prevent this confusion, we first check if there are any stopped streams. + // If so, we request the operator to clean them up, or restart them before going ahead. + // This allows us to assume that all stopped streams can be safely restarted + // if we cancel the operation. + stoppedStreams, err := sm.readTabletStreams(ctx, source.master, "state = 'Stopped'") + if err != nil { + return err + } + if len(stoppedStreams) != 0 { + return fmt.Errorf("cannot migrate until all streams are running: %s", source.si.ShardName()) + } } tabletStreams, err := sm.readTabletStreams(ctx, source.master, "") if err != nil { diff --git a/go/vt/wrangler/stream_migrater_test.go b/go/vt/wrangler/stream_migrater_test.go index ae8748950b0..3c6ef0b1d8c 100644 --- a/go/vt/wrangler/stream_migrater_test.go +++ b/go/vt/wrangler/stream_migrater_test.go @@ -885,13 +885,6 @@ func TestStreamMigrateSyncFail(t *testing.T) { "int64|varbinary|varchar|varbinary"), sourceRows[i]...), nil) - - // sm.cancelMigration->sm.readSourceStreamsForCancel: this is not actually stopStream, but we're reusing the bls here. - dbclient.addQuery("select id, workflow, source, pos from _vt.vreplication where db_name='vt_ks' and workflow != 'test_reverse'", sqltypes.MakeTestResult(sqltypes.MakeTestFields( - "id|workflow|source|pos", - "int64|varbinary|varchar|varbinary"), - sourceRows[i]...), - nil) } } stopStreams() @@ -914,28 +907,9 @@ func TestStreamMigrateSyncFail(t *testing.T) { } syncSourceStreams() - tme.expectWaitForCatchup() - - smCancelMigration := func() { - // sm.migrateStreams->sm.deleteTargetStreams - tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", resultid34, nil) - tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", resultid34, nil) - tme.dbTargetClients[0].addQuery("delete from _vt.vreplication where id in (3, 4)", &sqltypes.Result{}, nil) - tme.dbTargetClients[1].addQuery("delete from _vt.vreplication where id in (3, 4)", &sqltypes.Result{}, nil) - tme.dbTargetClients[0].addQuery("delete from _vt.copy_state where vrepl_id in (3, 4)", &sqltypes.Result{}, nil) - tme.dbTargetClients[1].addQuery("delete from _vt.copy_state where vrepl_id in (3, 4)", &sqltypes.Result{}, nil) - - // sm.migrateStreams->->restart source streams - tme.dbSourceClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", resultid12, nil) - tme.dbSourceClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", resultid12, nil) - tme.dbSourceClients[0].addQuery("update _vt.vreplication set state = 'Running', stop_pos = null, message = '' where id in (1, 2)", &sqltypes.Result{}, nil) - tme.dbSourceClients[1].addQuery("update _vt.vreplication set state = 'Running', stop_pos = null, message = '' where id in (1, 2)", &sqltypes.Result{}, nil) - tme.dbSourceClients[0].addQuery("select * from _vt.vreplication where id = 1", runningResult(1), nil) - tme.dbSourceClients[1].addQuery("select * from _vt.vreplication where id = 1", runningResult(1), nil) - tme.dbSourceClients[0].addQuery("select * from _vt.vreplication where id = 2", runningResult(2), nil) - tme.dbSourceClients[1].addQuery("select * from _vt.vreplication where id = 2", runningResult(2), nil) - } - smCancelMigration() + // sm.deleteTargetStreams (simplified to delete nothing) + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1')", &sqltypes.Result{}, nil) tme.expectCancelMigration() @@ -944,6 +918,7 @@ func TestStreamMigrateSyncFail(t *testing.T) { if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, want %s", err, want) } + verifyQueries(t, tme.allDBClients) } func TestStreamMigrateCancel(t *testing.T) { @@ -1094,17 +1069,81 @@ func TestStreamMigrateStoppedStreams(t *testing.T) { "int64|varbinary|varchar|varbinary"), sourceRows[i]...), nil) - dbclient.addQuery("select vrepl_id from _vt.copy_state where vrepl_id in (1)", &sqltypes.Result{}, nil) } } stopStreams() - tme.expectCancelMigration() _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) - want := "cannot migrate until all strems are running: 0" + want := "cannot migrate until all streams are running: 0" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) } + verifyQueries(t, tme.allDBClients) +} + +func TestStreamMigrateCancelWithStoppedStreams(t *testing.T) { + ctx := context.Background() + tme := newTestShardMigrater(ctx, t, []string{"-40", "40-"}, []string{"-80", "80-"}) + defer tme.stopTablets(t) + + // Migrate reads + err := tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_RDONLY, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + err = tme.wr.MigrateReads(ctx, tme.targetKeyspace, "test", topodatapb.TabletType_REPLICA, nil, DirectionForward) + if err != nil { + t.Fatal(err) + } + + tme.expectCheckJournals() + + stopStreams := func() { + var sourceRows [][]string + for _, sourceTargetShard := range tme.sourceShards { + var rows []string + for j, sourceShard := range tme.sourceShards { + bls := &binlogdatapb.BinlogSource{ + Keyspace: "ks1", + Shard: sourceShard, + Filter: &binlogdatapb.Filter{ + Rules: []*binlogdatapb.Rule{{ + Match: "t1", + Filter: fmt.Sprintf("select * from t1 where in_keyrange('%s')", sourceTargetShard), + }, { + Match: "t2", + Filter: fmt.Sprintf("select * from t2 where in_keyrange('%s')", sourceTargetShard), + }}, + }, + } + rows = append(rows, fmt.Sprintf("%d|t1t2|%v|MariaDB/5-456-888", j+1, bls)) + } + sourceRows = append(sourceRows, rows) + } + + for i, dbclient := range tme.dbSourceClients { + // sm.stopStreams->sm.readSourceStreams->readTabletStreams('') and VReplicationExec(_vt.copy_state) + dbclient.addQuery("select id, workflow, source, pos from _vt.vreplication where db_name='vt_ks' and workflow != 'test_reverse'", sqltypes.MakeTestResult(sqltypes.MakeTestFields( + "id|workflow|source|pos", + "int64|varbinary|varchar|varbinary"), + sourceRows[i]...), + nil) + dbclient.addQuery("select vrepl_id from _vt.copy_state where vrepl_id in (1, 2)", &sqltypes.Result{}, nil) + } + } + stopStreams() + + // sm.migrateStreams->->sm.deleteTargetStreams (no previously migrated streams) + tme.dbTargetClients[0].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1t2')", &sqltypes.Result{}, nil) + tme.dbTargetClients[1].addQuery("select id from _vt.vreplication where db_name = 'vt_ks' and workflow in ('t1t2')", &sqltypes.Result{}, nil) + + tme.expectCancelMigration() + + _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, true, false) + if err != nil { + t.Fatal(err) + } + verifyQueries(t, tme.allDBClients) } func TestStreamMigrateStillCopying(t *testing.T) { @@ -1159,13 +1198,13 @@ func TestStreamMigrateStillCopying(t *testing.T) { } } stopStreams() - tme.expectCancelMigration() _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "cannot migrate while vreplication streams in source shards are still copying: 0" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) } + verifyQueries(t, tme.allDBClients) } func TestStreamMigrateEmptyWorflow(t *testing.T) { @@ -1219,13 +1258,13 @@ func TestStreamMigrateEmptyWorflow(t *testing.T) { } } stopStreams() - tme.expectCancelMigration() _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "VReplication streams must have named workflows for migration: shard: ks:0, stream: 1" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) } + verifyQueries(t, tme.allDBClients) } func TestStreamMigrateDupWorflow(t *testing.T) { @@ -1279,13 +1318,13 @@ func TestStreamMigrateDupWorflow(t *testing.T) { } } stopStreams() - tme.expectCancelMigration() _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "VReplication stream has the same workflow name as the resharding workflow: shard: ks:0, stream: 1" if err == nil || err.Error() != want { t.Errorf("MigrateWrites err: %v, want %v", err, want) } + verifyQueries(t, tme.allDBClients) } func TestStreamMigrateStreamsMismatch(t *testing.T) { @@ -1350,13 +1389,13 @@ func TestStreamMigrateStreamsMismatch(t *testing.T) { } } stopStreams() - tme.expectCancelMigration() _, err = tme.wr.MigrateWrites(ctx, tme.targetKeyspace, "test", 1*time.Second, false, true) want := "streams are mismatched across source shards" if err == nil || !strings.Contains(err.Error(), want) { t.Errorf("MigrateWrites err: %v, must contain %v", err, want) } + verifyQueries(t, tme.allDBClients) } func TestTemplatize(t *testing.T) { From 7a6f6b7686e44fd0cd5629e9d1d242503e7e71aa Mon Sep 17 00:00:00 2001 From: Sugu Sougoumarane Date: Sun, 27 Oct 2019 17:46:16 -0700 Subject: [PATCH 4/4] migrater: address review comments Signed-off-by: Sugu Sougoumarane --- go/vt/wrangler/migrater.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/go/vt/wrangler/migrater.go b/go/vt/wrangler/migrater.go index f2439e82d10..6e3470cce7c 100644 --- a/go/vt/wrangler/migrater.go +++ b/go/vt/wrangler/migrater.go @@ -679,7 +679,7 @@ func (mi *migrater) cancelMigration(ctx context.Context, sm *streamMigrater) { err = mi.deleteReverseVReplication(ctx) if err != nil { - mi.wr.Logger().Errorf("Cancel migration failed: could not restart vreplication: %v", err) + mi.wr.Logger().Errorf("Cancel migration failed: could not delete revers vreplication entries: %v", err) } }