VtctldClient Reshard: add e2e tests to confirm CLI options and fix di…

…scovered issues. (#15353)

Signed-off-by: Rohit Nayak <rohit@planetscale.com>

go/cmd/vtctldclient/command/vreplication/reshard/create.go

@@ -60,20 +60,18 @@ func commandReshardCreate(cmd *cobra.Command, args []string) error {
  cli.FinishedParsing(cmd)
 
  req := &vtctldatapb.ReshardCreateRequest{
- Workflow: common.BaseOptions.Workflow,
- Keyspace: common.BaseOptions.TargetKeyspace,
-
+ Workflow: common.BaseOptions.Workflow,
+ Keyspace: common.BaseOptions.TargetKeyspace,
  TabletTypes: common.CreateOptions.TabletTypes,
  TabletSelectionPreference: tsp,
  Cells: common.CreateOptions.Cells,
  OnDdl: common.CreateOptions.OnDDL,
  DeferSecondaryKeys: common.CreateOptions.DeferSecondaryKeys,
  AutoStart: common.CreateOptions.AutoStart,
  StopAfterCopy: common.CreateOptions.StopAfterCopy,
-
- SourceShards: reshardCreateOptions.sourceShards,
- TargetShards: reshardCreateOptions.targetShards,
- SkipSchemaCopy: reshardCreateOptions.skipSchemaCopy,
+ SourceShards: reshardCreateOptions.sourceShards,
+ TargetShards: reshardCreateOptions.targetShards,
+ SkipSchemaCopy: reshardCreateOptions.skipSchemaCopy,
  }
  resp, err := common.GetClient().ReshardCreate(common.GetCommandCtx(), req)
  if err != nil {

go/test/endtoend/vreplication/vreplication_vtctldclient_cli_test.go

@@ -27,7 +27,6 @@ import (
  "google.golang.org/protobuf/encoding/protojson"
 
  "vitess.io/vitess/go/test/endtoend/cluster"
-
  binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
  tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
  topodatapb "vitess.io/vitess/go/vt/proto/topodata"
@@ -54,20 +53,33 @@ func TestVtctldclientCLI(t *testing.T) {
  require.NotNil(t, zone2)
  defer vc.TearDown()
 
- sourceKeyspace := "product"
- targetKeyspace := "customer"
+ sourceKeyspaceName := "product"
+ targetKeyspaceName := "customer"
  var mt iMoveTables
  workflowName := "wf1"
  targetTabs := setupMinimalCustomerKeyspace(t)
 
  t.Run("MoveTablesCreateFlags1", func(t *testing.T) {
- testMoveTablesFlags1(t, &mt, sourceKeyspace, targetKeyspace, workflowName, targetTabs)
+ testMoveTablesFlags1(t, &mt, sourceKeyspaceName, targetKeyspaceName, workflowName, targetTabs)
  })
  t.Run("MoveTablesCreateFlags2", func(t *testing.T) {
- testMoveTablesFlags2(t, &mt, sourceKeyspace, targetKeyspace, workflowName, targetTabs)
+ testMoveTablesFlags2(t, &mt, sourceKeyspaceName, targetKeyspaceName, workflowName, targetTabs)
+ })
+ t.Run("MoveTablesCompleteFlags3", func(t *testing.T) {
+ testMoveTablesFlags3(t, sourceKeyspaceName, targetKeyspaceName, targetTabs)
  })
- t.Run("MoveTablesCompleteFlags", func(t *testing.T) {
- testMoveTablesFlags3(t, sourceKeyspace, targetKeyspace, targetTabs)
+ t.Run("Reshard", func(t *testing.T) {
+ cell := vc.Cells["zone1"]
+ targetKeyspace := cell.Keyspaces[targetKeyspaceName]
+ sourceShard := "-80"
+ newShards := "-40,40-80"
+ require.NoError(t, vc.AddShards(t, []*Cell{cell}, targetKeyspace, newShards, 1, 0, 400, nil))
+ reshardWorkflowName := "reshard"
+ tablets := map[string]*cluster.VttabletProcess{
+ "-40": targetKeyspace.Shards["-40"].Tablets["zone1-400"].Vttablet,
+ "40-80": targetKeyspace.Shards["40-80"].Tablets["zone1-500"].Vttablet,
+ }
+ splitShard(t, targetKeyspaceName, reshardWorkflowName, sourceShard, newShards, tablets)
  })
 }
 
@@ -81,34 +93,31 @@ func testMoveTablesFlags1(t *testing.T, mt *iMoveTables, sourceKeyspace, targetK
  }
  completeFlags := []string{"--keep-routing-rules", "--keep-data"}
  switchFlags := []string{}
+ // Test one set of MoveTable flags.
  *mt = createMoveTables(t, sourceKeyspace, targetKeyspace, workflowName, tables, createFlags, completeFlags, switchFlags)
  (*mt).Show()
- moveTablesOutput := (*mt).GetLastOutput()
- // Test one set of MoveTable flags.
+ moveTablesResponse := getMoveTablesShowResponse(mt)
+ workflowResponse := getWorkflow(targetKeyspace, workflowName)
 
- workflowOutput, err := vc.VtctldClient.ExecuteCommandWithOutput("Workflow", "--keyspace", "customer", "show", "--workflow", "wf1")
- require.NoError(t, err)
- var moveTablesResponse vtctldatapb.GetWorkflowsResponse
- err = protojson.Unmarshal([]byte(moveTablesOutput), &moveTablesResponse)
- require.NoError(t, err)
-
- var workflowResponse vtctldatapb.GetWorkflowsResponse
- err = protojson.Unmarshal([]byte(workflowOutput), &workflowResponse)
- require.NoError(t, err)
-
- moveTablesResponse.Workflows[0].MaxVReplicationTransactionLag = 0
- moveTablesResponse.Workflows[0].MaxVReplicationLag = 0
- workflowResponse.Workflows[0].MaxVReplicationTransactionLag = 0
- workflowResponse.Workflows[0].MaxVReplicationLag = 0
  // also validates that MoveTables Show and Workflow Show return the same output.
- require.EqualValues(t, moveTablesResponse.CloneVT(), workflowResponse.CloneVT())
+ require.EqualValues(t, moveTablesResponse.CloneVT(), workflowResponse)
 
  // Validate that the flags are set correctly in the database.
- validateWorkflow1(t, workflowResponse.Workflows)
+ validateMoveTablesWorkflow(t, workflowResponse.Workflows)
  // Since we used --no-routing-rules, there should be no routing rules.
  confirmNoRoutingRules(t)
 }
 
+func getMoveTablesShowResponse(mt *iMoveTables) *vtctldatapb.GetWorkflowsResponse {
+ moveTablesOutput := (*mt).GetLastOutput()
+ var moveTablesResponse vtctldatapb.GetWorkflowsResponse
+ err := protojson.Unmarshal([]byte(moveTablesOutput), &moveTablesResponse)
+ require.NoError(vc.t, err)
+ moveTablesResponse.Workflows[0].MaxVReplicationTransactionLag = 0
+ moveTablesResponse.Workflows[0].MaxVReplicationLag = 0
+ return moveTablesResponse.CloneVT()
+}
+
 // Validates some of the flags created from the previous test.
 func testMoveTablesFlags2(t *testing.T, mt *iMoveTables, sourceKeyspace, targetKeyspace, workflowName string, targetTabs map[string]*cluster.VttabletProcess) {
  ksWorkflow := fmt.Sprintf("%s.%s", targetKeyspace, workflowName)
@@ -184,6 +193,135 @@ func createMoveTables(t *testing.T, sourceKeyspace, targetKeyspace, workflowName
  return mt
 }
 
+// reshard helpers
+
+func splitShard(t *testing.T, keyspace, workflowName, sourceShards, targetShards string, targetTabs map[string]*cluster.VttabletProcess) {
+ createFlags := []string{"--auto-start=false", "--defer-secondary-keys=false", "--stop-after-copy",
+ "--on-ddl", "STOP", "--tablet-types", "primary,rdonly", "--tablet-types-in-preference-order=true",
+ "--all-cells", "--format=json",
+ }
+ rs := newReshard(vc, &reshardWorkflow{
+ workflowInfo: &workflowInfo{
+ vc: vc,
+ workflowName: workflowName,
+ targetKeyspace: keyspace,
+ },
+ sourceShards: sourceShards,
+ targetShards: targetShards,
+ createFlags: createFlags,
+ }, workflowFlavorVtctld)
+
+ ksWorkflow := fmt.Sprintf("%s.%s", keyspace, workflowName)
+ rs.Create()
+ validateReshardResponse(rs)
+ workflowResponse := getWorkflow(keyspace, workflowName)
+ reshardShowResponse := getReshardShowResponse(&rs)
+ require.EqualValues(t, reshardShowResponse, workflowResponse)
+ validateReshardWorkflow(t, workflowResponse.Workflows)
+ waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", keyspace, workflowName), binlogdatapb.VReplicationWorkflowState_Stopped.String())
+ rs.Start()
+ waitForWorkflowState(t, vc, ksWorkflow, binlogdatapb.VReplicationWorkflowState_Stopped.String())
+ for _, tab := range targetTabs {
+ alias := fmt.Sprintf("zone1-%d", tab.TabletUID)
+ query := "update _vt.vreplication set source := replace(source, 'stop_after_copy:true', 'stop_after_copy:false') where db_name = 'vt_customer' and workflow = '" + workflowName + "'"
+ output, err := vc.VtctlClient.ExecuteCommandWithOutput("ExecuteFetchAsDba", alias, query)
+ require.NoError(t, err, output)
+ }
+ rs.Start()
+ waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", keyspace, workflowName), binlogdatapb.VReplicationWorkflowState_Running.String())
+ rs.Stop()
+ waitForWorkflowState(t, vc, ksWorkflow, binlogdatapb.VReplicationWorkflowState_Stopped.String())
+ rs.Start()
+ waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", keyspace, workflowName), binlogdatapb.VReplicationWorkflowState_Running.String())
+ for _, targetTab := range targetTabs {
+ catchup(t, targetTab, workflowName, "Reshard")
+ }
+ vdiff(t, keyspace, workflowName, "zone1", false, true, nil)
+
+ rs.SwitchReadsAndWrites()
+ waitForLowLag(t, keyspace, workflowName+"_reverse")
+ vdiff(t, keyspace, workflowName+"_reverse", "zone1", true, false, nil)
+
+ rs.ReverseReadsAndWrites()
+ waitForLowLag(t, keyspace, workflowName)
+ vdiff(t, keyspace, workflowName, "zone1", false, true, nil)
+ rs.SwitchReadsAndWrites()
+ rs.Complete()
+}
+
+func getReshardShowResponse(rs *iReshard) *vtctldatapb.GetWorkflowsResponse {
+ (*rs).Show()
+ reshardOutput := (*rs).GetLastOutput()
+ var reshardResponse vtctldatapb.GetWorkflowsResponse
+ err := protojson.Unmarshal([]byte(reshardOutput), &reshardResponse)
+ require.NoError(vc.t, err)
+ reshardResponse.Workflows[0].MaxVReplicationTransactionLag = 0
+ reshardResponse.Workflows[0].MaxVReplicationLag = 0
+ return reshardResponse.CloneVT()
+}
+
+func validateReshardResponse(rs iReshard) {
+ resp := getReshardResponse(rs)
+ require.NotNil(vc.t, resp)
+ require.NotNil(vc.t, resp.ShardStreams)
+ require.Equal(vc.t, len(resp.ShardStreams), 2)
+ keyspace := "customer"
+ for _, shard := range []string{"-40", "40-80"} {
+ streams := resp.ShardStreams[fmt.Sprintf("%s/%s", keyspace, shard)]
+ require.Equal(vc.t, 1, len(streams.Streams))
+ require.Equal(vc.t, binlogdatapb.VReplicationWorkflowState_Stopped.String(), streams.Streams[0].Status)
+ }
+}
+
+func validateReshardWorkflow(t *testing.T, workflows []*vtctldatapb.Workflow) {
+ require.Equal(t, 1, len(workflows))
+ wf := workflows[0]
+ require.Equal(t, "reshard", wf.Name)
+ require.Equal(t, binlogdatapb.VReplicationWorkflowType_Reshard.String(), wf.WorkflowType)
+ require.Equal(t, "None", wf.WorkflowSubType)
+ require.Equal(t, "customer", wf.Target.Keyspace)
+ require.Equal(t, 2, len(wf.Target.Shards))
+ require.Equal(t, "customer", wf.Source.Keyspace)
+ require.Equal(t, 1, len(wf.Source.Shards))
+ require.False(t, wf.DeferSecondaryKeys)
+
+ require.GreaterOrEqual(t, len(wf.ShardStreams), int(1))
+ oneStream := maps.Values(wf.ShardStreams)[0]
+ require.NotNil(t, oneStream)
+
+ stream := oneStream.Streams[0]
+ require.Equal(t, binlogdatapb.VReplicationWorkflowState_Stopped.String(), stream.State)
+ require.Equal(t, stream.TabletSelectionPreference, tabletmanagerdatapb.TabletSelectionPreference_INORDER)
+ require.True(t, slices.Equal([]topodatapb.TabletType{topodatapb.TabletType_PRIMARY, topodatapb.TabletType_RDONLY}, stream.TabletTypes))
+ require.True(t, slices.Equal([]string{"zone1", "zone2"}, stream.Cells))
+
+ bls := stream.BinlogSource
+ require.Equal(t, binlogdatapb.OnDDLAction_STOP, bls.OnDdl)
+ require.True(t, bls.StopAfterCopy)
+
+}
+
+func getReshardResponse(rs iReshard) *vtctldatapb.WorkflowStatusResponse {
+ reshardOutput := rs.GetLastOutput()
+ var reshardResponse vtctldatapb.WorkflowStatusResponse
+ err := protojson.Unmarshal([]byte(reshardOutput), &reshardResponse)
+ require.NoError(vc.t, err)
+ return reshardResponse.CloneVT()
+}
+
+// helper functions
+
+func getWorkflow(targetKeyspace, workflow string) *vtctldatapb.GetWorkflowsResponse {
+ workflowOutput, err := vc.VtctldClient.ExecuteCommandWithOutput("Workflow", "--keyspace", targetKeyspace, "show", "--workflow", workflow)
+ require.NoError(vc.t, err)
+ var workflowResponse vtctldatapb.GetWorkflowsResponse
+ err = protojson.Unmarshal([]byte(workflowOutput), &workflowResponse)
+ require.NoError(vc.t, err)
+ workflowResponse.Workflows[0].MaxVReplicationTransactionLag = 0
+ workflowResponse.Workflows[0].MaxVReplicationLag = 0
+ return workflowResponse.CloneVT()
+}
+
 func checkTablesExist(t *testing.T, tabletAlias string, tables []string) bool {
  tablesResponse, err := vc.VtctldClient.ExecuteCommandWithOutput("GetSchema", tabletAlias, "--tables", strings.Join(tables, ","), "--table-names-only")
  require.NoError(t, err)
@@ -211,6 +349,7 @@ func getRoutingRules(t *testing.T) *vschemapb.RoutingRules {
  require.NoError(t, err)
  return &routingRulesResponse
 }
+
 func confirmNoRoutingRules(t *testing.T) {
  routingRulesResponse := getRoutingRules(t)
  require.Zero(t, len(routingRulesResponse.Rules))
@@ -223,7 +362,7 @@ func confirmRoutingRulesExist(t *testing.T) {
 
 // We only want to validate non-standard attributes that are set by the CLI. The other end-to-end tests validate the rest.
 // We also check some of the standard attributes to make sure they are set correctly.
-func validateWorkflow1(t *testing.T, workflows []*vtctldatapb.Workflow) {
+func validateMoveTablesWorkflow(t *testing.T, workflows []*vtctldatapb.Workflow) {
  require.Equal(t, 1, len(workflows))
  wf := workflows[0]
  require.Equal(t, "wf1", wf.Name)

go/test/endtoend/vreplication/wrappers_test.go

@@ -74,9 +74,10 @@ type moveTablesWorkflow struct {
  tables string
  atomicCopy bool
  sourceShards string
- createFlags []string // currently only used by vtctld
 
+ // currently only used by vtctld
  lastOutput string
+ createFlags []string
  completeFlags []string
  switchFlags []string
 }
@@ -270,7 +271,12 @@ type reshardWorkflow struct {
  targetShards string
  skipSchemaCopy bool
 
- lastOutput string
+ // currently only used by vtctld
+ lastOutput string
+ createFlags []string
+ completeFlags []string
+ cancelFlags []string
+ switchFlags []string
 }
 
 type iReshard interface {
@@ -379,8 +385,9 @@ func (v VtctldReshard) Flavor() string {
 func (v VtctldReshard) exec(args ...string) {
  args2 := []string{"Reshard", "--workflow=" + v.workflowName, "--target-keyspace=" + v.targetKeyspace}
  args2 = append(args2, args...)
- if err := vc.VtctldClient.ExecuteCommand(args2...); err != nil {
- v.vc.t.Fatalf("failed to create Reshard workflow: %v", err)
+ var err error
+ if v.lastOutput, err = vc.VtctldClient.ExecuteCommandWithOutput(args2...); err != nil {
+ v.vc.t.Fatalf("failed to create Reshard workflow: %v: %s", err, v.lastOutput)
  }
 }
 
@@ -395,20 +402,22 @@ func (v VtctldReshard) Create() {
  if v.skipSchemaCopy {
  args = append(args, "--skip-schema-copy="+strconv.FormatBool(v.skipSchemaCopy))
  }
+ args = append(args, v.createFlags...)
  v.exec(args...)
 }
 
 func (v VtctldReshard) SwitchReadsAndWrites() {
- v.exec("SwitchTraffic")
+ args := []string{"SwitchTraffic"}
+ args = append(args, v.switchFlags...)
+ v.exec(args...)
 }
 
 func (v VtctldReshard) ReverseReadsAndWrites() {
  v.exec("ReverseTraffic")
 }
 
 func (v VtctldReshard) Show() {
- //TODO implement me
- panic("implement me")
+ v.exec("Show")
 }
 
 func (v VtctldReshard) SwitchReads() {
@@ -422,11 +431,15 @@ func (v VtctldReshard) SwitchWrites() {
 }
 
 func (v VtctldReshard) Cancel() {
- v.exec("Cancel")
+ args := []string{"Cancel"}
+ args = append(args, v.cancelFlags...)
+ v.exec(args...)
 }
 
 func (v VtctldReshard) Complete() {
- v.exec("Complete")
+ args := []string{"Complete"}
+ args = append(args, v.completeFlags...)
+ v.exec(args...)
 }
 
 func (v VtctldReshard) GetLastOutput() string {

go/vt/vtctl/workflow/server.go

@@ -1672,7 +1672,11 @@ func (s *Server) ReshardCreate(ctx context.Context, req *vtctldatapb.ReshardCrea
  log.Errorf("%w", err2)
  return nil, err
  }
- rs, err := s.buildResharder(ctx, keyspace, req.Workflow, req.SourceShards, req.TargetShards, strings.Join(cells, ","), "")
+ tabletTypesStr := topoproto.MakeStringTypeCSV(req.TabletTypes)
+ if req.TabletSelectionPreference == tabletmanagerdatapb.TabletSelectionPreference_INORDER {
+ tabletTypesStr = discovery.InOrderHint + tabletTypesStr
+ }
+ rs, err := s.buildResharder(ctx, keyspace, req.Workflow, req.SourceShards, req.TargetShards, strings.Join(cells, ","), tabletTypesStr)
  if err != nil {
  return nil, vterrors.Wrap(err, "buildResharder")
  }
@@ -1695,7 +1699,10 @@ func (s *Server) ReshardCreate(ctx context.Context, req *vtctldatapb.ReshardCrea
  } else {
  log.Warningf("Streams will not be started since --auto-start is set to false")
  }
- return nil, nil
+ return s.WorkflowStatus(ctx, &vtctldatapb.WorkflowStatusRequest{
+ Keyspace: keyspace,
+ Workflow: req.Workflow,
+ })
 }
 
 // VDiffCreate is part of the vtctlservicepb.VtctldServer interface.