migrater: option to cancel a migration

go/vt/vtctl/vtctl.go

@@ -327,7 +327,7 @@ var commands = []commandGroup{
 				"[-cells=c1,c2,...] [-reverse] -workflow=workflow <target keyspace> <tablet type>",
 				"Migrate read traffic for the specified workflow."},
 			{"MigrateWrites", commandMigrateWrites,
-				"[-filtered_replication_wait_time=30s] [-reverse_replication=<true/false>] -workflow=workflow <target keyspace>",
+				"[-filtered_replication_wait_time=30s] [-cancel] [-reverse_replication=false] -workflow=workflow <target keyspace>",
 				"Migrate write traffic for the specified workflow."},
 			{"CancelResharding", commandCancelResharding,
 				"<keyspace/shard>",
@@ -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
 	}

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)
@@ -184,11 +184,16 @@ 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
 		}
+		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
 	}
@@ -624,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
 		}

go/vt/wrangler/migrater_env_test.go

@@ -49,33 +49,56 @@ 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.
 // They are approximate and should be only used to test other features like stream migration.
 // 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)
@@ -424,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()