[release-15.0] OnlineDDL: cleanup cancelled migration artifacts; support --retain-artifacts=<duration> DDL strategy flag (#14029)

go/test/endtoend/onlineddl/scheduler/onlineddl_scheduler_test.go

@@ -154,10 +154,8 @@ func TestMain(m *testing.M) {
 	if err != nil {
 		fmt.Printf("%v\n", err)
 		os.Exit(1)
-	} else {
-		os.Exit(exitcode)
 	}
-
+	os.Exit(exitcode)
 }
 
 func TestSchemaChange(t *testing.T) {

go/vt/schema/ddl_strategy.go

@@ -19,12 +19,15 @@ package schema
 import (
 	"fmt"
 	"regexp"
+	"strconv"
+	"time"
 
 	"github.com/google/shlex"
 )
 
 var (
-	strategyParserRegexp = regexp.MustCompile(`^([\S]+)\s+(.*)$`)
+	strategyParserRegexp      = regexp.MustCompile(`^([\S]+)\s+(.*)$`)
+	retainArtifactsFlagRegexp = regexp.MustCompile(fmt.Sprintf(`^[-]{1,2}%s=(.*?)$`, retainArtifactsFlag))
 )
 
 const (
@@ -38,6 +41,7 @@ const (
 	allowConcurrentFlag    = "allow-concurrent"
 	fastOverRevertibleFlag = "fast-over-revertible"
 	fastRangeRotationFlag  = "fast-range-rotation"
+	retainArtifactsFlag    = "retain-artifacts"
 	vreplicationTestSuite  = "vreplication-test-suite"
 )
 
@@ -98,6 +102,9 @@ func ParseDDLStrategy(strategyVariable string) (*DDLStrategySetting, error) {
 	default:
 		return nil, fmt.Errorf("Unknown online DDL strategy: '%v'", strategy)
 	}
+	if _, err := setting.RetainArtifactsDuration(); err != nil {
+		return nil, err
+	}
 	return setting, nil
 }
 
@@ -168,7 +175,34 @@ func (setting *DDLStrategySetting) IsFastRangeRotationFlag() bool {
 	return setting.hasFlag(fastRangeRotationFlag)
 }
 
-// IsVreplicationTestSuite checks if strategy options include -vreplicatoin-test-suite
+// isRetainArtifactsFlag returns true when given option denotes a `--retain-artifacts=[...]` flag
+func isRetainArtifactsFlag(opt string) (string, bool) {
+	submatch := retainArtifactsFlagRegexp.FindStringSubmatch(opt)
+	if len(submatch) == 0 {
+		return "", false
+	}
+	return submatch[1], true
+}
+
+// RetainArtifactsDuration returns a the duration indicated by --retain-artifacts
+func (setting *DDLStrategySetting) RetainArtifactsDuration() (d time.Duration, err error) {
+	// We do some ugly manual parsing of --retain-artifacts
+	opts, _ := shlex.Split(setting.Options)
+	for _, opt := range opts {
+		if val, isRetainArtifacts := isRetainArtifactsFlag(opt); isRetainArtifacts {
+			// value is possibly quoted
+			if s, err := strconv.Unquote(val); err == nil {
+				val = s
+			}
+			if val != "" {
+				d, err = time.ParseDuration(val)
+			}
+		}
+	}
+	return d, err
+}
+
+// IsVreplicationTestSuite checks if strategy options include --vreplicatoin-test-suite
 func (setting *DDLStrategySetting) IsVreplicationTestSuite() bool {
 	return setting.hasFlag(vreplicationTestSuite)
 }
@@ -178,6 +212,9 @@ func (setting *DDLStrategySetting) RuntimeOptions() []string {
 	opts, _ := shlex.Split(setting.Options)
 	validOpts := []string{}
 	for _, opt := range opts {
+		if _, ok := isRetainArtifactsFlag(opt); ok {
+			continue
+		}
 		switch {
 		case isFlag(opt, declarativeFlag):
 		case isFlag(opt, skipTopoFlag):

go/vt/schema/ddl_strategy_test.go

@@ -19,6 +19,7 @@ package schema
 import (
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 )
@@ -37,6 +38,69 @@ func TestIsDirect(t *testing.T) {
 	assert.True(t, DDLStrategy("something").IsDirect())
 }
 
+func TestIsExpireArtifactsFlag(t *testing.T) {
+	tt := []struct {
+		s      string
+		expect bool
+		val    string
+		d      time.Duration
+	}{
+		{
+			s: "something",
+		},
+		{
+			s: "-retain-artifacts",
+		},
+		{
+			s: "--retain-artifacts",
+		},
+		{
+			s:      "--retain-artifacts=",
+			expect: true,
+		},
+		{
+			s:      "--retain-artifacts=0",
+			expect: true,
+			val:    "0",
+			d:      0,
+		},
+		{
+			s:      "-retain-artifacts=0",
+			expect: true,
+			val:    "0",
+			d:      0,
+		},
+		{
+			s:      "--retain-artifacts=1m",
+			expect: true,
+			val:    "1m",
+			d:      time.Minute,
+		},
+		{
+			s:      `--retain-artifacts="1m"`,
+			expect: true,
+			val:    `"1m"`,
+			d:      time.Minute,
+		},
+	}
+	for _, ts := range tt {
+		t.Run(ts.s, func(t *testing.T) {
+			setting, err := ParseDDLStrategy("online " + ts.s)
+			assert.NoError(t, err)
+
+			val, isRetainArtifacts := isRetainArtifactsFlag(ts.s)
+			assert.Equal(t, ts.expect, isRetainArtifacts)
+			assert.Equal(t, ts.val, val)
+
+			if ts.expect {
+				d, err := setting.RetainArtifactsDuration()
+				assert.NoError(t, err)
+				assert.Equal(t, ts.d, d)
+			}
+		})
+	}
+}
+
 func TestParseDDLStrategy(t *testing.T) {
 	tt := []struct {
 		strategyVariable     string
@@ -49,6 +113,7 @@ func TestParseDDLStrategy(t *testing.T) {
 		isAllowConcurrent    bool
 		fastOverRevertible   bool
 		fastRangeRotation    bool
+		expireArtifacts      time.Duration
 		runtimeOptions       string
 		err                  error
 	}{
@@ -145,6 +210,13 @@ func TestParseDDLStrategy(t *testing.T) {
 			runtimeOptions:    "",
 			fastRangeRotation: true,
 		},
+		{
+			strategyVariable: "vitess --retain-artifacts=4m",
+			strategy:         DDLStrategyVitess,
+			options:          "--retain-artifacts=4m",
+			runtimeOptions:   "",
+			expireArtifacts:  4 * time.Minute,
+		},
 	}
 	for _, ts := range tt {
 		setting, err := ParseDDLStrategy(ts.strategyVariable)
@@ -166,4 +238,8 @@ func TestParseDDLStrategy(t *testing.T) {
 		_, err := ParseDDLStrategy("other")
 		assert.Error(t, err)
 	}
+	{
+		_, err := ParseDDLStrategy("online --retain-artifacts=3")
+		assert.Error(t, err)
+	}
 }

go/vt/vttablet/onlineddl/executor.go

@@ -4527,6 +4527,11 @@ func (e *Executor) SubmitMigration(
 
 	revertedUUID, _ := onlineDDL.GetRevertUUID() // Empty value if the migration is not actually a REVERT. Safe to ignore error.
 	retainArtifactsSeconds := int64((retainOnlineDDLTables).Seconds())
+	if retainArtifacts, _ := onlineDDL.StrategySetting().RetainArtifactsDuration(); retainArtifacts != 0 {
+		// Explicit retention indicated by `--retain-artifact` DDL strategy flag for this migration. Override!
+		retainArtifactsSeconds = int64((retainArtifacts).Seconds())
+	}
+
 	_, allowConcurrentMigration := e.allowConcurrentMigration(onlineDDL)
 	query, err := sqlparser.ParseAndBind(sqlInsertMigration,
 		sqltypes.StringBindVariable(onlineDDL.UUID),

go/vt/vttablet/onlineddl/schema.go

@@ -404,7 +404,7 @@ const (
 			log_path
 		FROM _vt.schema_migrations
 		WHERE
-			migration_status IN ('complete', 'failed')
+			migration_status IN ('complete', 'cancelled', 'failed')
 			AND cleanup_timestamp IS NULL
 			AND completed_timestamp <= IF(retain_artifacts_seconds=0,
 				NOW() - INTERVAL %a SECOND,