Add Workflow Update Client Command

go/cmd/vtctldclient/command/workflows.go

@@ -18,11 +18,17 @@ package command
 
 import (
 	"fmt"
+	"sort"
+	"strings"
 
 	"github.com/spf13/cobra"
 
 	"vitess.io/vitess/go/cmd/vtctldclient/cli"
+	"vitess.io/vitess/go/textutil"
 
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
 )
 
@@ -35,6 +41,55 @@ var (
 		Args:                  cobra.ExactArgs(1),
 		RunE:                  commandGetWorkflows,
 	}
+
+	// Workflow is a parent command for Workflow* sub commands.
+	Workflow = &cobra.Command{
+		Use:                   "workflow",
+		Short:                 "Administer VReplication workflows (Reshard, MoveTables, etc) in the given keyspace",
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Workflow"},
+		Args:                  cobra.ExactArgs(1),
+		RunE:                  commandGetWorkflows,
+	}
+
+	// WorkflowUpdate makes a WorkflowUpdate gRPC call to a vtctld.
+	WorkflowUpdate = &cobra.Command{
+		Use:                   "update",
+		Short:                 "Update the configuration parameters for a VReplication workflow",
+		Example:               `vtctldclient --server=localhost:15999 workflow --keyspace=customer update --workflow=commerce2customer --cells "zone1" --cells "zone2" -c "zone3,zone4" -c "zone5"`,
+		DisableFlagsInUseLine: true,
+		Aliases:               []string{"Update"},
+		Args:                  cobra.NoArgs,
+		PreRunE: func(cmd *cobra.Command, args []string) error {
+			changes := false
+			if cmd.Flags().Lookup("cells").Changed { // No validation required
+				changes = true
+			} else {
+				workflowUpdateOptions.Cells = textutil.SimulatedNullStringSlice
+			}
+			if cmd.Flags().Lookup("tablet-types").Changed { // Validate the provided value(s)
+				changes = true
+				for _, tabletType := range workflowUpdateOptions.TabletTypes {
+					if _, ok := topodatapb.TabletType_value[strings.ToUpper(strings.TrimSpace(tabletType))]; !ok {
+						return fmt.Errorf("invalid tablet type: %s", tabletType)
+					}
+				}
+			} else {
+				workflowUpdateOptions.TabletTypes = textutil.SimulatedNullStringSlice
+			}
+			if cmd.Flags().Lookup("on-ddl").Changed { // Validate the provided value
+				changes = true
+				if _, ok := binlogdatapb.OnDDLAction_value[strings.ToUpper(workflowUpdateOptions.OnDDL)]; !ok {
+					return fmt.Errorf("invalid on-ddl value: %s", workflowUpdateOptions.OnDDL)
+				}
+			} // Simulated NULL will need to be handled in command
+			if !changes {
+				return fmt.Errorf("no configuration options specified to update")
+			}
+			return nil
+		},
+		RunE: commandWorkflowUpdate,
+	}
 )
 
 var getWorkflowsOptions = struct {
@@ -65,7 +120,70 @@ func commandGetWorkflows(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+var (
+	workflowOptions = struct {
+		Keyspace string
+	}{}
+	workflowUpdateOptions = struct {
+		Workflow    string
+		Cells       []string
+		TabletTypes []string
+		OnDDL       string
+	}{}
+)
+
+func commandWorkflowUpdate(cmd *cobra.Command, args []string) error {
+	cli.FinishedParsing(cmd)
+
+	// We've already validated any provided value, if one WAS provided.
+	// Now we need to do the mapping from the string representation to
+	// the enum value.
+	onddl := int32(textutil.SimulatedNullInt) // Simulated NULL when no value provided
+	if val, ok := binlogdatapb.OnDDLAction_value[strings.ToUpper(workflowUpdateOptions.OnDDL)]; ok {
+		onddl = val
+	}
+
+	req := &vtctldatapb.WorkflowUpdateRequest{
+		Keyspace: workflowOptions.Keyspace,
+		TabletRequest: &tabletmanagerdatapb.UpdateVRWorkflowRequest{
+			Workflow:    workflowUpdateOptions.Workflow,
+			Cells:       workflowUpdateOptions.Cells,
+			TabletTypes: workflowUpdateOptions.TabletTypes,
+			OnDdl:       binlogdatapb.OnDDLAction(onddl),
+		},
+	}
+
+	resp, err := client.WorkflowUpdate(commandCtx, req)
+	if err != nil {
+		return err
+	}
+
+	// Sort the inner TabletInfo slice for deterministic output.
+	sort.Slice(resp.Details, func(i, j int) bool {
+		return resp.Details[i].Tablet < resp.Details[j].Tablet
+	})
+
+	data, err := cli.MarshalJSON(resp)
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("%s\n", data)
+
+	return nil
+}
+
 func init() {
 	GetWorkflows.Flags().BoolVarP(&getWorkflowsOptions.ShowAll, "show-all", "a", false, "Show all workflows instead of just active workflows.")
 	Root.AddCommand(GetWorkflows)
+
+	Workflow.PersistentFlags().StringVarP(&workflowOptions.Keyspace, "keyspace", "k", "", "Keyspace context for the workflow (required)")
+	Workflow.MarkPersistentFlagRequired("keyspace")
+	Root.AddCommand(Workflow)
+	WorkflowUpdate.Flags().StringVarP(&workflowUpdateOptions.Workflow, "workflow", "w", "", "The workflow you want to update (required)")
+	WorkflowUpdate.MarkFlagRequired("workflow")
+	WorkflowUpdate.Flags().StringSliceVarP(&workflowUpdateOptions.Cells, "cells", "c", nil, "New Cell(s) or CellAlias(es) (comma-separated) to replicate from")
+	WorkflowUpdate.Flags().StringSliceVarP(&workflowUpdateOptions.TabletTypes, "tablet-types", "t", nil, "New source tablet types to replicate from (e.g. PRIMARY, REPLICA, RDONLY)")
+	WorkflowUpdate.Flags().StringVar(&workflowUpdateOptions.OnDDL, "on-ddl", "", "New instruction on what to do when DDL is encountered in the VReplication stream. Possible values are IGNORE, STOP, EXEC, and EXEC_IGNORE")
+	Workflow.AddCommand(WorkflowUpdate)
 }

go/flags/endtoend/vtctldclient.txt

@@ -89,6 +89,7 @@ Available Commands:
   ValidateVersionShard        Validates that the version on the primary matches all of the replicas.
   completion                  Generate the autocompletion script for the specified shell
   help                        Help about any command
+  workflow                    Administer VReplication workflows (Reshard, MoveTables, etc) in the given keyspace
 
 Flags:
       --action_timeout duration                timeout for the total command (default 1h0m0s)

go/test/endtoend/vreplication/resharding_workflows_v2_test.go

@@ -172,6 +172,27 @@ func tstWorkflowComplete(t *testing.T) error {
 	return tstWorkflowAction(t, workflowActionComplete, "", "")
 }
 
+// testWorkflowUpdate is a very simple test of the workflow update
+// vtctlclient/vtctldclient command.
+// It performs a non-behavior impacting update, setting tablet-types
+// to primary,replica,rdonly (the only applicable types in these tests).
+func testWorkflowUpdate(t *testing.T) {
+	tabletTypes := "primary,replica,rdonly"
+	// Test vtctlclient first
+	_, err := vc.VtctlClient.ExecuteCommandWithOutput("workflow", "--", "--tablet-types", tabletTypes, "no.exist", "update")
+	require.Error(t, err, err)
+	resp, err := vc.VtctlClient.ExecuteCommandWithOutput("workflow", "--", "--tablet-types", tabletTypes, ksWorkflow, "update")
+	require.NoError(t, err)
+	require.NotEqual(t, "", resp)
+
+	// Test vtctldclient last
+	_, err = vc.VtctldClient.ExecuteCommandWithOutput("workflow", "--keyspace", "noexist", "update", "--workflow", "noexist", "--tablet-types", tabletTypes)
+	require.Error(t, err)
+	resp, err = vc.VtctldClient.ExecuteCommandWithOutput("workflow", "--keyspace", targetKs, "update", "--workflow", workflowName, "--tablet-types", tabletTypes)
+	require.NoError(t, err, err)
+	require.NotEqual(t, "", resp)
+}
+
 func tstWorkflowCancel(t *testing.T) error {
 	return tstWorkflowAction(t, workflowActionCancel, "", "")
 }
@@ -391,6 +412,9 @@ func testReshardV2Workflow(t *testing.T) {
 	verifyNoInternalTables(t, vtgateConn, targetKs+"/-40")
 	verifyNoInternalTables(t, vtgateConn, targetKs+"/c0-")
 
+	// Confirm that updating Reshard workflows works.
+	testWorkflowUpdate(t)
+
 	testRestOfWorkflow(t)
 }
 
@@ -426,6 +450,9 @@ func testMoveTablesV2Workflow(t *testing.T) {
 	output, _ = vc.VtctlClient.ExecuteCommandWithOutput(listAllArgs...)
 	require.Contains(t, output, "Following workflow(s) found in keyspace customer: wf1")
 
+	// Confirm that updating MoveTable workflows works.
+	testWorkflowUpdate(t)
+
 	err := tstWorkflowCancel(t)
 	require.NoError(t, err)
 

go/textutil/strings.go

@@ -20,10 +20,16 @@ import (
 	"net/url"
 	"regexp"
 	"strings"
+
+	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/vt/proto/binlogdata"
 )
 
 var (
-	delimitedListRegexp = regexp.MustCompile(`[ ,;]+`)
+	delimitedListRegexp      = regexp.MustCompile(`[ ,;]+`)
+	SimulatedNullString      = sqltypes.NULL.String()
+	SimulatedNullStringSlice = []string{sqltypes.NULL.String()}
+	SimulatedNullInt         = -1
 )
 
 // SplitDelimitedList splits a given string by comma, semi-colon or space, and returns non-empty strings
@@ -73,3 +79,27 @@ func SingleWordCamel(w string) string {
 	}
 	return strings.ToUpper(w[0:1]) + strings.ToLower(w[1:])
 }
+
+// ValueIsSimulatedNull returns true if the value represents
+// a NULL or unknown/unspecified value. This is used to
+// distinguish between a zero value / default and a user
+// provided value that is equivalent (e.g. an empty string
+// or slice).
+func ValueIsSimulatedNull(val any) bool {
+	switch cval := val.(type) {
+	case string:
+		return cval == SimulatedNullString
+	case []string:
+		return len(cval) == 1 && cval[0] == sqltypes.NULL.String()
+	case binlogdata.OnDDLAction:
+		return int32(cval) == int32(SimulatedNullInt)
+	case int:
+		return cval == SimulatedNullInt
+	case int32:
+		return int32(cval) == int32(SimulatedNullInt)
+	case int64:
+		return int64(cval) == int64(SimulatedNullInt)
+	default:
+		return false
+	}
+}