Allow stop/start on specific shards

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

Author: rohit-nayak-ps

go/cmd/vtctldclient/command/vreplication/workflow/state.go

@@ -60,14 +60,17 @@ func commandUpdateState(cmd *cobra.Command, args []string) error {
 	cli.FinishedParsing(cmd)
 
 	var state binlogdatapb.VReplicationWorkflowState
+	var shards []string
 	switch strings.ToLower(cmd.Name()) {
 	case "start":
 		if err := common.CanRestartWorkflow(baseOptions.Keyspace, baseOptions.Workflow); err != nil {
 			return err
 		}
 		state = binlogdatapb.VReplicationWorkflowState_Running
+		shards = baseOptions.Shards
 	case "stop":
 		state = binlogdatapb.VReplicationWorkflowState_Stopped
+		shards = baseOptions.Shards
 	default:
 		return fmt.Errorf("invalid workflow state: %s", args[0])
 	}
@@ -80,6 +83,7 @@ func commandUpdateState(cmd *cobra.Command, args []string) error {
 			Cells:       textutil.SimulatedNullStringSlice,
 			TabletTypes: textutil.SimulatedNullTabletTypeSlice,
 			State:       &state,
+			Shards:      shards,
 		},
 	}
 

go/vt/proto/tabletmanagerdata/tabletmanagerdata.pb.go

@@ -1607,8 +1607,10 @@ func (s *Server) WorkflowUpdate(ctx context.Context, req *vtctldatapb.WorkflowUp
 	span.Annotate("on_ddl", req.TabletRequest.OnDdl)
 	span.Annotate("state", req.TabletRequest.State)
 	span.Annotate("config_overrides", req.TabletRequest.ConfigOverrides)
+	span.Annotate("shards", req.TabletRequest.Shards)
 
 	vx := vexec.NewVExec(req.Keyspace, req.TabletRequest.Workflow, s.ts, s.tmc, s.env.Parser())
+	vx.SetShardSubset(req.TabletRequest.Shards)
 	callback := func(ctx context.Context, tablet *topo.TabletInfo) (*querypb.QueryResult, error) {
 		res, err := s.tmc.UpdateVReplicationWorkflow(ctx, tablet.Tablet, req.TabletRequest)
 		if err != nil {

go/vt/proto/tabletmanagerdata/tabletmanagerdata_vtproto.pb.go

@@ -781,7 +781,7 @@ message UpdateVReplicationWorkflowRequest {
   optional TabletSelectionPreference tablet_selection_preference = 4;
   optional binlogdata.OnDDLAction on_ddl = 5;
   optional binlogdata.VReplicationWorkflowState state = 6;
-  reserved 7; // unused, was: repeated string shards
+  repeated string shards = 7;
   map<string, string> config_overrides = 8;
 }