OnlineDDL: Add cut-over-threshold flag to ApplySchema client command

go/cmd/vtctldclient/command/schema.go

@@ -31,15 +31,16 @@ import (
 	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/topo/topoproto"
 	"vitess.io/vitess/go/vt/vtctl/grpcvtctldserver"
+	"vitess.io/vitess/go/vt/vttablet/onlineddl"
 
 	vtctldatapb "vitess.io/vitess/go/vt/proto/vtctldata"
-	"vitess.io/vitess/go/vt/proto/vtrpc"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 var (
 	// ApplySchema makes an ApplySchema gRPC call to a vtctld.
 	ApplySchema = &cobra.Command{
-		Use:   "ApplySchema [--ddl-strategy <strategy>] [--uuid <uuid> ...] [--migration-context <context>] [--wait-replicas-timeout <duration>] [--caller-id <caller_id>] {--sql-file <file> | --sql <sql>} <keyspace>",
+		Use:   "ApplySchema [--ddl-strategy <strategy>] [--cut-over-threshold <duration>] [--uuid <uuid> ...] [--migration-context <context>] [--wait-replicas-timeout <duration>] [--caller-id <caller_id>] {--sql-file <file> | --sql <sql>} <keyspace>",
 		Short: "Applies the schema change to the specified keyspace on every primary, running in parallel on all shards. The changes are then propagated to replicas via replication.",
 		Long: `Applies the schema change to the specified keyspace on every primary, running in parallel on all shards. The changes are then propagated to replicas via replication.
 
@@ -103,6 +104,7 @@ var applySchemaOptions = struct {
 	SkipPreflight           bool
 	CallerID                string
 	BatchSize               int64
+	CutOverThreshold        time.Duration
 }{}
 
 func commandApplySchema(cmd *cobra.Command, args []string) error {
@@ -129,9 +131,9 @@ func commandApplySchema(cmd *cobra.Command, args []string) error {
 
 	cli.FinishedParsing(cmd)
 
-	var cid *vtrpc.CallerID
+	var cid *vtrpcpb.CallerID
 	if applySchemaOptions.CallerID != "" {
-		cid = &vtrpc.CallerID{Principal: applySchemaOptions.CallerID}
+		cid = &vtrpcpb.CallerID{Principal: applySchemaOptions.CallerID}
 	}
 
 	ks := cmd.Flags().Arg(0)
@@ -143,6 +145,7 @@ func commandApplySchema(cmd *cobra.Command, args []string) error {
 		UuidList:            applySchemaOptions.UUIDList,
 		MigrationContext:    applySchemaOptions.MigrationContext,
 		WaitReplicasTimeout: protoutil.DurationToProto(applySchemaOptions.WaitReplicasTimeout),
+		CutOverThreshold:    protoutil.DurationToProto(applySchemaOptions.CutOverThreshold),
 		CallerId:            cid,
 		BatchSize:           applySchemaOptions.BatchSize,
 	})
@@ -293,6 +296,7 @@ func init() {
 	ApplySchema.Flags().StringArrayVar(&applySchemaOptions.SQL, "sql", nil, "Semicolon-delimited, repeatable SQL commands to apply. Exactly one of --sql|--sql-file is required.")
 	ApplySchema.Flags().StringVar(&applySchemaOptions.SQLFile, "sql-file", "", "Path to a file containing semicolon-delimited SQL commands to apply. Exactly one of --sql|--sql-file is required.")
 	ApplySchema.Flags().Int64Var(&applySchemaOptions.BatchSize, "batch-size", 0, "How many queries to batch together. Only applicable when all queries are CREATE TABLE|VIEW")
+	ApplySchema.Flags().DurationVar(&applySchemaOptions.CutOverThreshold, "cut-over-threshold", onlineddl.DefaultCutOverThreshold, "The maximum time to wait for an OnlineDDL migration cut-over to complete before failing the operation, to retry again later.")
 
 	Root.AddCommand(ApplySchema)
 

go/vt/vtctl/grpcvtctldserver/server.go

@@ -276,6 +276,9 @@ func (s *VtctldServer) ApplySchema(ctx context.Context, req *vtctldatapb.ApplySc
 
 	executor := schemamanager.NewTabletExecutor(migrationContext, s.ts, s.tmc, logger, waitReplicasTimeout, req.BatchSize, s.ws.SQLParser())
 
+	if cutOverThreshold, set, err := protoutil.DurationFromProto(req.CutOverThreshold); set && err == nil {
+		req.DdlStrategy += fmt.Sprintf(" --cut-over-threshold=%v", cutOverThreshold)
+	}
 	if err = executor.SetDDLStrategy(req.DdlStrategy); err != nil {
 		err = vterrors.Wrapf(err, "invalid DdlStrategy: %s", req.DdlStrategy)
 		return resp, err

go/vt/vttablet/onlineddl/executor.go

@@ -94,7 +94,7 @@ var (
 	ptOSCBinaryPath         = "/usr/bin/pt-online-schema-change"
 	migrationCheckInterval  = 1 * time.Minute
 	retainOnlineDDLTables   = 24 * time.Hour
-	defaultCutOverThreshold = 10 * time.Second
+	DefaultCutOverThreshold = 10 * time.Second
 	maxConcurrentOnlineDDLs = 256
 
 	migrationNextCheckIntervals = []time.Duration{1 * time.Second, 5 * time.Second, 10 * time.Second, 20 * time.Second}
@@ -205,7 +205,7 @@ func getMigrationCutOverThreshold(onlineDDL *schema.OnlineDDL) time.Duration {
 	if threshold, _ := onlineDDL.StrategySetting().CutOverThreshold(); threshold != 0 {
 		return threshold
 	}
-	return defaultCutOverThreshold
+	return DefaultCutOverThreshold
 }
 
 // NewExecutor creates a new gh-ost executor.

proto/vtctldata.proto

@@ -408,6 +408,9 @@ message ApplySchemaRequest {
   vtrpc.CallerID caller_id = 9;
   // BatchSize indicates how many queries to apply together
   int64 batch_size = 10;
+  // CutOverThreshold is a limit on how long we attempt cutover work before
+  // giving up and returning an error, to retry the cutover again later.
+  vttime.Duration cut_over_threshold = 11;
 }
 
 message ApplySchemaResponse {