VReplication: Improve workflow cancel/delete

go/test/endtoend/cluster/vttablet_process.go

@@ -74,6 +74,7 @@ type VttabletProcess struct {
 	SupportsBackup              bool
 	ExplicitServingStatus       bool
 	ServingStatus               string
+	DbName                      string
 	DbPassword                  string
 	DbPort                      int
 	DbFlavor                    string
@@ -148,6 +149,8 @@ func (vttablet *VttabletProcess) Setup() (err error) {
 		return
 	}
 
+	vttablet.DbName = "vt_" + vttablet.Keyspace
+
 	vttablet.exit = make(chan error)
 	go func() {
 		if vttablet.proc != nil {
@@ -442,8 +445,11 @@ func (vttablet *VttabletProcess) TearDownWithTimeout(timeout time.Duration) erro
 
 // CreateDB creates the database for keyspace
 func (vttablet *VttabletProcess) CreateDB(keyspace string) error {
-	_, _ = vttablet.QueryTablet(fmt.Sprintf("drop database IF EXISTS vt_%s", keyspace), keyspace, false)
-	_, err := vttablet.QueryTablet(fmt.Sprintf("create database IF NOT EXISTS vt_%s", keyspace), keyspace, false)
+	if vttablet.DbName == "" {
+		vttablet.DbName = "vt_" + keyspace
+	}
+	_, _ = vttablet.QueryTablet(fmt.Sprintf("drop database IF EXISTS %s", vttablet.DbName), keyspace, false)
+	_, err := vttablet.QueryTablet(fmt.Sprintf("create database IF NOT EXISTS %s", vttablet.DbName), keyspace, false)
 	return err
 }
 

go/test/endtoend/vreplication/helper_test.go

@@ -92,7 +92,7 @@ func execQueryWithRetry(t *testing.T, conn *mysql.Conn, query string, timeout ti
 		select {
 		case <-ctx.Done():
 			require.FailNow(t, fmt.Sprintf("query %q did not succeed before the timeout of %s; last seen result: %v",
-				query, timeout, qr.Rows))
+				query, timeout, qr))
 		case <-ticker.C:
 			log.Infof("query %q failed with error %v, retrying in %ds", query, err, defaultTick)
 		}
@@ -147,19 +147,6 @@ func execVtgateQuery(t *testing.T, conn *mysql.Conn, database string, query stri
 	return qr
 }
 
-func execVtgateQueryWithRetry(t *testing.T, conn *mysql.Conn, database string, query string, timeout time.Duration) *sqltypes.Result {
-	if strings.TrimSpace(query) == "" {
-		return nil
-	}
-	if database != "" {
-		execQuery(t, conn, "use `"+database+"`;")
-	}
-	execQuery(t, conn, "begin")
-	qr := execQueryWithRetry(t, conn, query, timeout)
-	execQuery(t, conn, "commit")
-	return qr
-}
-
 func checkHealth(t *testing.T, url string) bool {
 	resp, err := http.Get(url)
 	require.NoError(t, err)
@@ -516,20 +503,24 @@ func validateDryRunResults(t *testing.T, output string, want []string) {
 	require.NotEmpty(t, output)
 	gotDryRun := strings.Split(output, "\n")
 	require.True(t, len(gotDryRun) > 3)
-	startRow := 3
-	if strings.Contains(gotDryRun[0], "deprecated") {
+	var startRow int
+	if strings.HasPrefix(gotDryRun[1], "Parameters:") { // vtctlclient
+		startRow = 3
+	} else if strings.Contains(gotDryRun[0], "deprecated") {
 		startRow = 4
+	} else {
+		startRow = 2
 	}
 	gotDryRun = gotDryRun[startRow : len(gotDryRun)-1]
 	if len(want) != len(gotDryRun) {
-		t.Fatalf("want and got: lengths don't match, \nwant\n%s\n\ngot\n%s", strings.Join(want, "\n"), strings.Join(gotDryRun, "\n"))
+		require.Fail(t, "invalid dry run results", "want and got: lengths don't match, \nwant\n%s\n\ngot\n%s", strings.Join(want, "\n"), strings.Join(gotDryRun, "\n"))
 	}
 	var match, fail bool
 	fail = false
 	for i, w := range want {
 		w = strings.TrimSpace(w)
 		g := strings.TrimSpace(gotDryRun[i])
-		if w[0] == '/' {
+		if len(w) > 0 && w[0] == '/' {
 			w = strings.TrimSpace(w[1:])
 			result := strings.HasPrefix(g, w)
 			match = result
@@ -538,11 +529,11 @@ func validateDryRunResults(t *testing.T, output string, want []string) {
 		}
 		if !match {
 			fail = true
-			t.Fatalf("want %s, got %s\n", w, gotDryRun[i])
+			require.Fail(t, "invlaid dry run results", "want %s, got %s\n", w, gotDryRun[i])
 		}
 	}
 	if fail {
-		t.Fatalf("Dry run results don't match, want %s, got %s", want, gotDryRun)
+		require.Fail(t, "invalid dry run results", "Dry run results don't match, want %s, got %s", want, gotDryRun)
 	}
 }
 
@@ -578,7 +569,7 @@ func isTableInDenyList(t *testing.T, vc *VitessCluster, ksShard string, table st
 	var err error
 	found := false
 	if output, err = vc.VtctlClient.ExecuteCommandWithOutput("GetShard", ksShard); err != nil {
-		t.Fatalf("%v %v", err, output)
+		require.Fail(t, "GetShard error", "%v %v", err, output)
 		return false, err
 	}
 	jsonparser.ArrayEach([]byte(output), func(value []byte, dataType jsonparser.ValueType, offset int, err error) {
@@ -602,8 +593,8 @@ func expectNumberOfStreams(t *testing.T, vtgateConn *mysql.Conn, name string, wo
 	waitForQueryResult(t, vtgateConn, database, query, fmt.Sprintf(`[[INT64(%d)]]`, want))
 }
 
-// confirmAllStreamsRunning confirms that all of the migrated streams are running
-// after a Reshard.
+// confirmAllStreamsRunning confirms that all of the workflow's streams are
+// in the running state.
 func confirmAllStreamsRunning(t *testing.T, vtgateConn *mysql.Conn, database string) {
 	query := sqlparser.BuildParsedQuery("select count(*) from %s.vreplication where state != '%s'",
 		sidecarDBIdentifier, binlogdatapb.VReplicationWorkflowState_Running.String()).Query
@@ -801,7 +792,7 @@ func isBinlogRowImageNoBlob(t *testing.T, tablet *cluster.VttabletProcess) bool
 
 func getRowCount(t *testing.T, vtgateConn *mysql.Conn, table string) int {
 	query := fmt.Sprintf("select count(*) from %s", table)
-	qr := execVtgateQuery(t, vtgateConn, "", query)
+	qr := execQuery(t, vtgateConn, query)
 	numRows, _ := qr.Rows[0][0].ToInt()
 	return numRows
 }

go/test/endtoend/vreplication/vreplication_test.go

@@ -367,6 +367,7 @@ func testVreplicationWorkflows(t *testing.T, limited bool, binlogRowImage string
 	expectNumberOfStreams(t, vtgateConn, "Customer3to2", "sales", "product:0", 3)
 	reshardCustomer3to1Merge(t)
 	confirmAllStreamsRunning(t, vtgateConn, "customer:0")
+
 	expectNumberOfStreams(t, vtgateConn, "Customer3to1", "sales", "product:0", 1)
 
 	t.Run("Verify CopyState Is Optimized Afterwards", func(t *testing.T) {
@@ -605,7 +606,7 @@ func TestCellAliasVreplicationWorkflow(t *testing.T) {
 	vc.AddKeyspace(t, []*Cell{cell1, cell2}, keyspace, shard, initialProductVSchema, initialProductSchema, defaultReplicas, defaultRdonly, 100, sourceKsOpts)
 
 	// Add cell alias containing only zone2
-	result, err := vc.VtctlClient.ExecuteCommandWithOutput("AddCellsAlias", "--", "--cells", "zone2", "alias")
+	result, err := vc.VtctldClient.ExecuteCommandWithOutput("AddCellsAlias", "--cells", "zone2", "alias")
 	require.NoError(t, err, "command failed with output: %v", result)
 
 	verifyClusterHealth(t, vc)
@@ -722,10 +723,13 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 		execVtgateQuery(t, vtgateConn, sourceKs, "update json_tbl set j1 = null, j2 = 'null', j3 = '\"null\"'")
 		execVtgateQuery(t, vtgateConn, sourceKs, "insert into json_tbl(id, j1, j2, j3) values (7, null, 'null', '\"null\"')")
 		waitForNoWorkflowLag(t, vc, targetKs, workflow)
-		for _, shard := range []string{"-80", "80-"} {
-			shardTarget := fmt.Sprintf("%s:%s", targetKs, shard)
-			if res := execVtgateQuery(t, vtgateConn, shardTarget, "select cid from customer"); len(res.Rows) > 0 {
-				waitForQueryResult(t, vtgateConn, shardTarget, "select distinct dec80 from customer", `[[DECIMAL(0)]]`)
+		for _, tablet := range []*cluster.VttabletProcess{customerTab1, customerTab2} {
+			// Query the tablet's mysqld directly as the targets will have denied table entries.
+			dbc, err := tablet.TabletConn(targetKs, true)
+			require.NoError(t, err)
+			defer dbc.Close()
+			if res := execQuery(t, dbc, "select cid from customer"); len(res.Rows) > 0 {
+				waitForQueryResult(t, dbc, tablet.DbName, "select distinct dec80 from customer", `[[DECIMAL(0)]]`)
 				dec80Replicated = true
 			}
 		}
@@ -833,7 +837,7 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 			printShardPositions(vc, ksShards)
 			switchWrites(t, workflowType, ksWorkflow, true)
 
-			output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
+			output, err := vc.VtctldClient.ExecuteCommandWithOutput("Workflow", "--keyspace", targetKs, "show", "--workflow", workflow)
 			require.NoError(t, err)
 			require.Contains(t, output, "'customer.reverse_bits'")
 			require.Contains(t, output, "'customer.bmd5'")
@@ -942,7 +946,7 @@ func reshardMerchant2to3SplitMerge(t *testing.T) {
 		var err error
 
 		for _, shard := range strings.Split("-80,80-", ",") {
-			output, err = vc.VtctlClient.ExecuteCommandWithOutput("GetShard", "merchant:"+shard)
+			output, err = vc.VtctldClient.ExecuteCommandWithOutput("GetShard", "merchant:"+shard)
 			if err == nil {
 				t.Fatal("GetShard merchant:-80 failed")
 			}
@@ -951,7 +955,7 @@ func reshardMerchant2to3SplitMerge(t *testing.T) {
 
 		for _, shard := range strings.Split("-40,40-c0,c0-", ",") {
 			ksShard := fmt.Sprintf("%s:%s", merchantKeyspace, shard)
-			output, err = vc.VtctlClient.ExecuteCommandWithOutput("GetShard", ksShard)
+			output, err = vc.VtctldClient.ExecuteCommandWithOutput("GetShard", ksShard)
 			if err != nil {
 				t.Fatalf("GetShard merchant failed for: %s: %v", shard, err)
 			}
@@ -1400,7 +1404,7 @@ func waitForLowLag(t *testing.T, keyspace, workflow string) {
 	waitDuration := 500 * time.Millisecond
 	duration := maxWait
 	for duration > 0 {
-		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", fmt.Sprintf("%s.%s", keyspace, workflow), "Show")
+		output, err := vc.VtctldClient.ExecuteCommandWithOutput("Workflow", "--keyspace", "show", "--workflow", workflow)
 		require.NoError(t, err)
 		lagSeconds, err = jsonparser.GetInt([]byte(output), "MaxVReplicationTransactionLag")
 
@@ -1483,7 +1487,7 @@ func reshardAction(t *testing.T, action, workflow, keyspaceName, sourceShards, t
 }
 
 func applyVSchema(t *testing.T, vschema, keyspace string) {
-	err := vc.VtctlClient.ExecuteCommand("ApplyVSchema", "--", "--vschema", vschema, keyspace)
+	err := vc.VtctldClient.ExecuteCommand("ApplyVSchema", "--vschema", vschema, keyspace)
 	require.NoError(t, err)
 }
 
@@ -1494,19 +1498,24 @@ func switchReadsDryRun(t *testing.T, workflowType, cells, ksWorkflow string, dry
 			"workflow type specified: %s", workflowType)
 	}
 	ensureCanSwitch(t, workflowType, cells, ksWorkflow)
-	output, err := vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--cells="+cells, "--tablet_types=rdonly,replica",
-		"--dry_run", "SwitchTraffic", ksWorkflow)
+	ks, wf, ok := strings.Cut(ksWorkflow, ".")
+	require.True(t, ok)
+	output, err := vc.VtctldClient.ExecuteCommandWithOutput(workflowType, "--workflow", wf, "--target-keyspace", ks, "SwitchTraffic",
+		"--cells="+cells, "--tablet-types=rdonly,replica", "--dry-run")
 	require.NoError(t, err, fmt.Sprintf("Switching Reads DryRun Error: %s: %s", err, output))
 	if dryRunResults != nil {
 		validateDryRunResults(t, output, dryRunResults)
 	}
 }
 
 func ensureCanSwitch(t *testing.T, workflowType, cells, ksWorkflow string) {
+	ks, wf, ok := strings.Cut(ksWorkflow, ".")
+	require.True(t, ok)
 	timer := time.NewTimer(defaultTimeout)
 	defer timer.Stop()
 	for {
-		_, err := vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--cells="+cells, "--dry_run", "SwitchTraffic", ksWorkflow)
+		_, err := vc.VtctldClient.ExecuteCommandWithOutput(workflowType, "--workflow", wf, "--target-keyspace", ks, "SwitchTraffic",
+			"--cells="+cells, "--dry-run")
 		if err == nil {
 			return
 		}
@@ -1532,11 +1541,13 @@ func switchReads(t *testing.T, workflowType, cells, ksWorkflow string, reverse b
 		command = "ReverseTraffic"
 	}
 	ensureCanSwitch(t, workflowType, cells, ksWorkflow)
-	output, err = vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--cells="+cells, "--tablet_types=rdonly",
-		command, ksWorkflow)
+	ks, wf, ok := strings.Cut(ksWorkflow, ".")
+	require.True(t, ok)
+	output, err = vc.VtctldClient.ExecuteCommandWithOutput(workflowType, "--workflow", wf, "--target-keyspace", ks, command,
+		"--cells="+cells, "--tablet-types=rdonly")
 	require.NoError(t, err, fmt.Sprintf("%s Error: %s: %s", command, err, output))
-	output, err = vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--cells="+cells, "--tablet_types=replica",
-		command, ksWorkflow)
+	output, err = vc.VtctldClient.ExecuteCommandWithOutput(workflowType, "--workflow", wf, "--target-keyspace", ks, command,
+		"--cells="+cells, "--tablet-types=replica")
 	require.NoError(t, err, fmt.Sprintf("%s Error: %s: %s", command, err, output))
 }
 
@@ -1575,8 +1586,10 @@ func switchWritesDryRun(t *testing.T, workflowType, ksWorkflow string, dryRunRes
 		require.FailNowf(t, "Invalid workflow type for SwitchTraffic, must be MoveTables or Reshard",
 			"workflow type specified: %s", workflowType)
 	}
-	output, err := vc.VtctlClient.ExecuteCommandWithOutput(workflowType, "--", "--tablet_types=primary", "--dry_run",
-		"SwitchTraffic", ksWorkflow)
+	ks, wf, ok := strings.Cut(ksWorkflow, ".")
+	require.True(t, ok)
+	output, err := vc.VtctldClient.ExecuteCommandWithOutput(workflowType, "--workflow", wf, "--target-keyspace", ks,
+		"SwitchTraffic", "--tablet-types=primary", "--dry-run")
 	require.NoError(t, err, fmt.Sprintf("Switch writes DryRun Error: %s: %s", err, output))
 	validateDryRunResults(t, output, dryRunResults)
 }

go/test/endtoend/vreplication/vreplication_test_env.go

@@ -19,31 +19,28 @@ package vreplication
 var dryRunResultsSwitchWritesCustomerShard = []string{
 	"Lock keyspace product",
 	"Lock keyspace customer",
-	"Stop writes on keyspace product, tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order]:",
-	"/       Keyspace product, Shard 0 at Position",
-	"Wait for VReplication on stopped streams to catchup for up to 30s",
-	"Create reverse replication workflow p2c_reverse",
+	"/Stop writes on keyspace product for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order]: [keyspace:product;shard:0;position:",
+	"Wait for vreplication on stopped streams to catchup for up to 30s",
+	"Create reverse vreplication workflow p2c_reverse",
 	"Create journal entries on source databases",
-	"Enable writes on keyspace customer tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order]",
+	"Enable writes on keyspace customer for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order]",
 	"Switch routing from keyspace product to keyspace customer",
 	"Routing rules for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order] will be updated",
-	"Switch writes completed, freeze and delete vreplication streams on:",
-	"       tablet 200 ",
-	"       tablet 300 ",
-	"Start reverse replication streams on:",
-	"       tablet 100 ",
-	"Mark vreplication streams frozen on:",
-	"       Keyspace customer, Shard -80, Tablet 200, Workflow p2c, DbName vt_customer",
-	"       Keyspace customer, Shard 80-, Tablet 300, Workflow p2c, DbName vt_customer",
+	"Switch writes completed, freeze and delete vreplication streams on: [tablet:200,tablet:300]",
+	"Start reverse vreplication streams on: [tablet:100]",
+	"Mark vreplication streams frozen on: [keyspace:customer;shard:-80;tablet:200;workflow:p2c;dbname:vt_customer,keyspace:customer;shard:80-;tablet:300;workflow:p2c;dbname:vt_customer]",
 	"Unlock keyspace customer",
 	"Unlock keyspace product",
+	"", // Additional empty newline in the output
 }
 
 var dryRunResultsReadCustomerShard = []string{
 	"Lock keyspace product",
 	"Switch reads for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order] to keyspace customer for tablet types [RDONLY,REPLICA]",
 	"Routing rules for tables [Lead,Lead-1,blüb_tbl,customer,db_order_test,geom_tbl,json_tbl,loadtest,reftable,vdiff_order] will be updated",
+	"Serving VSchema will be rebuilt for the customer keyspace",
 	"Unlock keyspace product",
+	"", // Additional empty newline in the output
 }
 
 var dryRunResultsSwitchWritesM2m3 = []string{