OnlineDDL: "cancel-all" command to cancel all pending migrations in keyspace

go/test/endtoend/cluster/vtctlclient_process.go

@@ -96,6 +96,15 @@ func (vtctlclient *VtctlClientProcess) OnlineDDLCancelMigration(Keyspace, uuid s
 	)
 }
 
+// OnlineDDLCancelMigration cancels a given migration uuid
+func (vtctlclient *VtctlClientProcess) OnlineDDLCancelAllMigrations(Keyspace string) (result string, err error) {
+	return vtctlclient.ExecuteCommandWithOutput(
+		"OnlineDDL",
+		Keyspace,
+		"cancel-all",
+	)
+}
+
 // OnlineDDLRetryMigration retries a given migration uuid
 func (vtctlclient *VtctlClientProcess) OnlineDDLRetryMigration(Keyspace, uuid string) (result string, err error) {
 	return vtctlclient.ExecuteCommandWithOutput(

go/test/endtoend/onlineddl/onlineddl_test.go

@@ -24,6 +24,7 @@ import (
 	"path"
 	"regexp"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -182,6 +183,26 @@ func TestSchemaChange(t *testing.T) {
 		checkRecentMigrations(t, uuid, schema.OnlineDDLStatusFailed)
 		checkCancelMigration(t, uuid, false)
 		checkRetryMigration(t, uuid, true)
+		// migration will fail again
+	}
+	{
+		// no migrations pending at this time
+		time.Sleep(10 * time.Second)
+		checkCancelAllMigrations(t, 0)
+	}
+	{
+		// spawn n migrations; cancel them via cancel-all
+		var wg sync.WaitGroup
+		count := 4
+		for i := 0; i < count; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				_ = testAlterTable(t, alterTableThrottlingStatement, "gh-ost --max-load=Threads_running=1", "vtgate", "ghost_col")
+			}()
+		}
+		wg.Wait()
+		checkCancelAllMigrations(t, count)
 	}
 }
 
@@ -279,6 +300,18 @@ func checkCancelMigration(t *testing.T, uuid string, expectCancelPossible bool)
 	assert.Equal(t, len(clusterInstance.Keyspaces[0].Shards), len(m))
 }
 
+// checkCancelAllMigrations all pending migrations
+func checkCancelAllMigrations(t *testing.T, expectCount int) {
+	result, err := clusterInstance.VtctlclientProcess.OnlineDDLCancelAllMigrations(keyspaceName)
+	fmt.Println("# 'vtctlclient OnlineDDL cancel-all' output (for debug purposes):")
+	fmt.Println(result)
+	assert.NoError(t, err)
+
+	r := fullWordRegexp(fmt.Sprintf("%d", expectCount))
+	m := r.FindAllString(result, -1)
+	assert.Equal(t, len(clusterInstance.Keyspaces[0].Shards), len(m))
+}
+
 // checkRetryMigration attempts to retry a migration, and expects rejection
 func checkRetryMigration(t *testing.T, uuid string, expectRetryPossible bool) {
 	result, err := clusterInstance.VtctlclientProcess.OnlineDDLRetryMigration(keyspaceName, uuid)

go/vt/vtctl/vtctl.go

@@ -2547,6 +2547,13 @@ func commandOnlineDDL(ctx context.Context, wr *wrangler.Wrangler, subFlags *flag
 			uuid = arg
 			query = fmt.Sprintf(`update _vt.schema_migrations set migration_status='cancel' where migration_uuid='%s'`, uuid)
 		}
+	case "cancel-all":
+		{
+			if arg != "" {
+				return fmt.Errorf("UUID not allowed in %s", command)
+			}
+			query = `update _vt.schema_migrations set migration_status='cancel-all'`
+		}
 	default:
 		return fmt.Errorf("Unknown OnlineDDL command: %s", command)
 	}

go/vt/vttablet/onlineddl/executor.go

@@ -65,6 +65,7 @@ var (
 )
 
 var vexecUpdateTemplates = []string{
+	`update _vt.schema_migrations set migration_status='val' where mysql_schema='val'`,
 	`update _vt.schema_migrations set migration_status='val' where migration_uuid='val' and mysql_schema='val'`,
 	`update _vt.schema_migrations set migration_status='val' where migration_uuid='val' and mysql_schema='val' and shard='val'`,
 }
@@ -894,6 +895,32 @@ func (e *Executor) cancelMigrations(ctx context.Context, uuids []string) (err er
 	return nil
 }
 
+// cancelPendingMigrations cancels all pending migrations (that are expected to run or are running)
+// for this keyspace
+func (e *Executor) cancelPendingMigrations(ctx context.Context) (result *sqltypes.Result, err error) {
+	parsed := sqlparser.BuildParsedQuery(sqlSelectPendingMigrations, "_vt")
+	r, err := e.execQuery(ctx, parsed.Query)
+	if err != nil {
+		return result, err
+	}
+	var uuids []string
+	for _, row := range r.Named().Rows {
+		uuid := row["migration_uuid"].ToString()
+		uuids = append(uuids, uuid)
+	}
+
+	result = &sqltypes.Result{}
+	for _, uuid := range uuids {
+		log.Infof("cancelPendingMigrations: cancelling %s", uuid)
+		res, err := e.cancelMigration(ctx, uuid, true)
+		if err != nil {
+			return result, err
+		}
+		result.AppendResult(res)
+	}
+	return result, nil
+}
+
 // scheduleNextMigration attemps to schedule a single migration to run next.
 // possibly there's no migrations to run. Possibly there's a migration running right now,
 // in which cases nothing happens.
@@ -1443,7 +1470,7 @@ func (e *Executor) VExec(ctx context.Context, vx *vexec.TabletVExec) (qr *queryp
 			return nil, err
 		}
 		if !match {
-			return nil, fmt.Errorf("Query must match one of these templates: %s", strings.Join(vexecUpdateTemplates, "; "))
+			return nil, fmt.Errorf("Query must match one of these templates: %s; query=%s", strings.Join(vexecUpdateTemplates, "; "), vx.Query)
 		}
 		if shard, _ := vx.ColumnStringVal(vx.WhereCols, "shard"); shard != "" {
 			// shard is specified.
@@ -1464,7 +1491,16 @@ func (e *Executor) VExec(ctx context.Context, vx *vexec.TabletVExec) (qr *queryp
 			if err != nil {
 				return nil, err
 			}
+			if !schema.IsOnlineDDLUUID(uuid) {
+				return nil, fmt.Errorf("Not an Online DDL UUID: %s", uuid)
+			}
 			return response(e.cancelMigration(ctx, uuid, true))
+		case cancelAllMigrationHint:
+			uuid, _ := vx.ColumnStringVal(vx.WhereCols, "migration_uuid")
+			if uuid != "" {
+				return nil, fmt.Errorf("Unexpetced UUID: %s", uuid)
+			}
+			return response(e.cancelPendingMigrations(ctx))
 		default:
 			return nil, fmt.Errorf("Unexpected value for migration_status: %v. Supported values are: %s, %s",
 				statusVal, retryMigrationHint, cancelMigrationHint)

go/vt/vttablet/onlineddl/schema.go

@@ -143,6 +143,12 @@ const (
 			migration_status='running'
 			AND liveness_timestamp < NOW() - INTERVAL %a MINUTE
 	`
+	sqlSelectPendingMigrations = `SELECT
+			migration_uuid
+		FROM %s.schema_migrations
+		WHERE
+			migration_status IN ('queued', 'ready', 'running')
+	`
 	sqlSelectUncollectedArtifacts = `SELECT
 			migration_uuid,
 			artifacts
@@ -213,8 +219,9 @@ const (
 )
 
 const (
-	retryMigrationHint  = "retry"
-	cancelMigrationHint = "cancel"
+	retryMigrationHint     = "retry"
+	cancelMigrationHint    = "cancel"
+	cancelAllMigrationHint = "cancel-all"
 )
 
 var (