Merge pull request #835 from slackhq/backport-increment-migration-metrics

Backport vitessio#19297: increment migration metrics

Author: yushuqin

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

@@ -46,6 +46,11 @@ import (
 
 const (
 	anyErrorIndicator = "<any-error-of-any-kind>"
+
+	// Migration metric names as exposed via /debug/vars
+	metricStartedMigrations    = "StartedMigrations"
+	metricSuccessfulMigrations = "SuccessfulMigrations"
+	metricFailedMigrations     = "FailedMigrations"
 )
 
 type testOnlineDDLStatementParams struct {
@@ -3049,3 +3054,164 @@ func runInTransaction(t *testing.T, ctx context.Context, tablet *cluster.Vttable
 	}
 	return err
 }
+
+// TestMigrationMetrics tests that migration metrics are correctly incremented
+// when migrations are executed end-to-end. This verifies that the metrics
+// tracking works correctly in a real cluster environment.
+func TestMigrationMetrics(t *testing.T) {
+	throttler.EnableLagThrottlerAndWaitForStatus(t, clusterInstance)
+
+	shards = clusterInstance.Keyspaces[0].Shards
+	require.Equal(t, 1, len(shards))
+
+	// Helper function to get metric value from /debug/vars
+	getMetric := func(metricName string) int64 {
+		vars := primaryTablet.VttabletProcess.GetVars()
+		if val, ok := vars[metricName]; ok {
+			switch v := val.(type) {
+			case float64:
+				return int64(v)
+			case int64:
+				return v
+			case int:
+				return int64(v)
+			}
+		}
+		return 0
+	}
+
+	// Create the stress_test table for our tests
+	t.Run("setup table", func(t *testing.T) {
+		createStatement := `
+			CREATE TABLE IF NOT EXISTS stress_test (
+				id bigint(20) not null,
+				rand_val varchar(32) null default '',
+				hint_col varchar(64) not null default 'just-created',
+				created_timestamp timestamp not null default current_timestamp,
+				updates int unsigned not null default 0,
+				PRIMARY KEY (id),
+				key created_idx(created_timestamp),
+				key updates_idx(updates)
+			) ENGINE=InnoDB
+		`
+		uuid := testOnlineDDLStatement(t, &testOnlineDDLStatementParams{
+			ddlStatement:    createStatement,
+			ddlStrategy:     "direct",
+			executeStrategy: "vtgate",
+		})
+		// For direct strategy, uuid may be empty
+		_ = uuid
+	})
+
+	// Test 1: Successful migration increments metrics
+	t.Run("successful migration increments metrics", func(t *testing.T) {
+		// Get baseline metrics
+		startedBefore := getMetric(metricStartedMigrations)
+		successfulBefore := getMetric(metricSuccessfulMigrations)
+
+		// Execute a simple migration
+		uuid := testOnlineDDLStatement(t, &testOnlineDDLStatementParams{
+			ddlStatement:    "ALTER TABLE stress_test ENGINE=InnoDB",
+			ddlStrategy:     "vitess",
+			executeStrategy: "vtgate",
+		})
+		require.NotEmpty(t, uuid)
+
+		// Wait for migration to complete
+		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid,
+			normalWaitTime, schema.OnlineDDLStatusComplete)
+
+		// Check metrics incremented correctly using assert.Eventually
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			startedAfter := getMetric(metricStartedMigrations)
+			successfulAfter := getMetric(metricSuccessfulMigrations)
+
+			assert.Equal(c, startedBefore+1, startedAfter,
+				"StartedMigrations should increment by 1")
+			assert.Equal(c, successfulBefore+1, successfulAfter,
+				"SuccessfulMigrations should increment by 1")
+		}, 10*time.Second, time.Second, "metrics did not increment correctly")
+	})
+
+	// Test 2: Failed migration increments metrics
+	t.Run("failed migration increments metrics", func(t *testing.T) {
+		failedBefore := getMetric(metricFailedMigrations)
+		startedBefore := getMetric(metricStartedMigrations)
+
+		// Execute a migration that will fail (invalid column)
+		uuid := testOnlineDDLStatement(t, &testOnlineDDLStatementParams{
+			ddlStatement:    "ALTER TABLE stress_test DROP COLUMN nonexistent_column",
+			ddlStrategy:     "vitess",
+			executeStrategy: "vtgate",
+		})
+		require.NotEmpty(t, uuid)
+
+		// Wait for migration to fail
+		onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid,
+			normalWaitTime, schema.OnlineDDLStatusFailed)
+
+		// Verify metrics
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			failedAfter := getMetric(metricFailedMigrations)
+			startedAfter := getMetric(metricStartedMigrations)
+
+			assert.Equal(c, failedBefore+1, failedAfter,
+				"FailedMigrations should increment by 1")
+			assert.Equal(c, startedBefore+1, startedAfter,
+				"StartedMigrations should increment by 1 even for failed migrations")
+		}, 10*time.Second, time.Second, "metrics did not increment correctly")
+	})
+
+	// Test 3: Multiple migrations increment metrics correctly
+	t.Run("multiple migrations increment metrics correctly", func(t *testing.T) {
+		startedBefore := getMetric(metricStartedMigrations)
+		successfulBefore := getMetric(metricSuccessfulMigrations)
+
+		// Run 3 migrations
+		uuids := make([]string, 3)
+		for i := range 3 {
+			uuids[i] = testOnlineDDLStatement(t, &testOnlineDDLStatementParams{
+				ddlStatement:    "ALTER TABLE stress_test ENGINE=InnoDB",
+				ddlStrategy:     "vitess",
+				executeStrategy: "vtgate",
+			})
+			require.NotEmpty(t, uuids[i])
+		}
+
+		// Wait for all to complete
+		for _, uuid := range uuids {
+			onlineddl.WaitForMigrationStatus(t, &vtParams, shards, uuid,
+				extendedWaitTime, schema.OnlineDDLStatusComplete)
+		}
+
+		// Verify metrics
+		assert.EventuallyWithT(t, func(c *assert.CollectT) {
+			startedAfter := getMetric(metricStartedMigrations)
+			successfulAfter := getMetric(metricSuccessfulMigrations)
+
+			assert.Equal(c, startedBefore+3, startedAfter,
+				"StartedMigrations should increment by 3")
+			assert.Equal(c, successfulBefore+3, successfulAfter,
+				"SuccessfulMigrations should increment by 3")
+		}, 15*time.Second, time.Second, "metrics did not increment correctly")
+	})
+
+	// Test 4: Verify metrics are accessible via /debug/vars
+	t.Run("metrics are exposed via debug vars", func(t *testing.T) {
+		vars := primaryTablet.VttabletProcess.GetVars()
+		require.NotNil(t, vars)
+
+		// Verify all three metrics exist
+		assert.Contains(t, vars, metricStartedMigrations,
+			"StartedMigrations metric should be exposed")
+		assert.Contains(t, vars, metricSuccessfulMigrations,
+			"SuccessfulMigrations metric should be exposed")
+		assert.Contains(t, vars, metricFailedMigrations,
+			"FailedMigrations metric should be exposed")
+
+		// Verify metrics are non-negative
+		assert.GreaterOrEqual(t, getMetric(metricStartedMigrations), int64(0))
+		assert.GreaterOrEqual(t, getMetric(metricSuccessfulMigrations), int64(0))
+		assert.GreaterOrEqual(t, getMetric(metricFailedMigrations), int64(0))
+	})
+}

go/vt/vttablet/onlineddl/executor.go

@@ -2228,6 +2228,7 @@ func (e *Executor) readFailedCancelledMigrationsInContextBeforeMigration(ctx con
 func (e *Executor) failMigration(ctx context.Context, onlineDDL *schema.OnlineDDL, withError error) error {
 	defer e.triggerNextCheckInterval()
 	_ = e.updateMigrationStatusFailedOrCancelled(ctx, onlineDDL.UUID)
+	failedMigrations.Add(1)
 	if withError != nil {
 		_ = e.updateMigrationMessage(ctx, onlineDDL.UUID, withError.Error())
 	}
@@ -3261,6 +3262,7 @@ func (e *Executor) reviewStaleMigrations(ctx context.Context) error {
 		if err := e.updateMigrationStatus(ctx, onlineDDL.UUID, schema.OnlineDDLStatusFailed); err != nil {
 			return err
 		}
+		failedMigrations.Add(1)
 		defer e.triggerNextCheckInterval()
 		_ = e.updateMigrationStartedTimestamp(ctx, uuid)
 		// Because the migration is stale, it may not update completed_timestamp. It is essential to set completed_timestamp
@@ -4500,17 +4502,20 @@ func (e *Executor) onSchemaMigrationStatus(ctx context.Context,
 		{
 			_ = e.updateMigrationStartedTimestamp(ctx, uuid)
 			err = e.updateMigrationTimestamp(ctx, "liveness_timestamp", uuid)
+			startedMigrations.Add(1)
 		}
 	case schema.OnlineDDLStatusComplete:
 		{
 			progressPct = progressPctFull
 			_ = e.updateMigrationStartedTimestamp(ctx, uuid)
 			err = e.updateMigrationTimestamp(ctx, "completed_timestamp", uuid)
+			successfulMigrations.Add(1)
 		}
 	case schema.OnlineDDLStatusFailed:
 		{
 			_ = e.updateMigrationStartedTimestamp(ctx, uuid)
 			err = e.updateMigrationTimestamp(ctx, "completed_timestamp", uuid)
+			failedMigrations.Add(1)
 		}
 	}
 	if err != nil {

go/vt/vttablet/onlineddl/executor_test.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/vt/schema"
 )
 
 func TestShouldCutOverAccordingToBackoff(t *testing.T) {
@@ -221,3 +223,109 @@ func TestSafeMigrationCutOverThreshold(t *testing.T) {
 		})
 	}
 }
+
+func TestMigrationMetricsIncrement(t *testing.T) {
+	tcases := []struct {
+		name     string
+		testFunc func()
+		verify   func(before int64, after int64) bool
+	}{
+		{
+			name: "startedMigrations increments correctly",
+			testFunc: func() {
+				startedMigrations.Add(1)
+			},
+			verify: func(before int64, after int64) bool {
+				return after == before+1
+			},
+		},
+		{
+			name: "successfulMigrations increments correctly",
+			testFunc: func() {
+				successfulMigrations.Add(1)
+			},
+			verify: func(before int64, after int64) bool {
+				return after == before+1
+			},
+		},
+		{
+			name: "failedMigrations increments correctly",
+			testFunc: func() {
+				failedMigrations.Add(1)
+			},
+			verify: func(before int64, after int64) bool {
+				return after == before+1
+			},
+		},
+	}
+
+	for _, tcase := range tcases {
+		t.Run(tcase.name, func(t *testing.T) {
+			var before, after int64
+
+			switch tcase.name {
+			case "startedMigrations increments correctly":
+				before = startedMigrations.Get()
+				tcase.testFunc()
+				after = startedMigrations.Get()
+			case "successfulMigrations increments correctly":
+				before = successfulMigrations.Get()
+				tcase.testFunc()
+				after = successfulMigrations.Get()
+			case "failedMigrations increments correctly":
+				before = failedMigrations.Get()
+				tcase.testFunc()
+				after = failedMigrations.Get()
+			}
+
+			assert.True(t, tcase.verify(before, after), "metric should increment correctly: before=%d, after=%d", before, after)
+		})
+	}
+}
+
+func TestMigrationStatusTransitionsUpdateMetrics(t *testing.T) {
+	tcases := []struct {
+		name          string
+		status        schema.OnlineDDLStatus
+		expectStarted int64
+		expectSuccess int64
+		expectFailed  int64
+	}{
+		{
+			name:          "running status updates started metric",
+			status:        schema.OnlineDDLStatusRunning,
+			expectStarted: 1,
+		},
+		{
+			name:          "complete status updates successful metric",
+			status:        schema.OnlineDDLStatusComplete,
+			expectSuccess: 1,
+		},
+		{
+			name:         "failed status updates failed metric",
+			status:       schema.OnlineDDLStatusFailed,
+			expectFailed: 1,
+		},
+	}
+
+	for _, tcase := range tcases {
+		t.Run(tcase.name, func(t *testing.T) {
+			startedBefore := startedMigrations.Get()
+			successBefore := successfulMigrations.Get()
+			failedBefore := failedMigrations.Get()
+
+			switch tcase.status {
+			case schema.OnlineDDLStatusRunning:
+				startedMigrations.Add(1)
+			case schema.OnlineDDLStatusComplete:
+				successfulMigrations.Add(1)
+			case schema.OnlineDDLStatusFailed:
+				failedMigrations.Add(1)
+			}
+
+			assert.Equal(t, startedBefore+tcase.expectStarted, startedMigrations.Get(), "startedMigrations")
+			assert.Equal(t, successBefore+tcase.expectSuccess, successfulMigrations.Get(), "successfulMigrations")
+			assert.Equal(t, failedBefore+tcase.expectFailed, failedMigrations.Get(), "failedMigrations")
+		})
+	}
+}