add online_schema_migrations metric (#4497)

emit metric for gh-ost MySQL migrations to track it's usage

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -50,6 +50,12 @@ const (
 	queryStatusVarSQLMode = 1
 )
 
+const (
+	OnlineSchemaMigrationToolGhOst = "gh-ost"
+	OnlineSchemaMigrationToolPtOsc = "pt-online-schema-change"
+	OnlineSchemaMigrationToolOther = "other"
+)
+
 // sqlModeFromStatusVars extracts the sql_mode bitmask from a QueryEvent's status vars.
 // Both MySQL and MariaDB guarantee status vars are written in increasing order,
 // so we only need to parse 0 and 1 to get to 1
@@ -586,12 +592,15 @@ func (c *MySqlConnector) PullRecords(
 				c.logger.Warn("processing QueryEvent with logged warnings", slog.Any("warns", warns))
 			}
 			for _, stmt := range stmts {
-				if alterTableStmt, ok := stmt.(*ast.AlterTableStmt); ok {
+				switch s := stmt.(type) {
+				case *ast.AlterTableStmt:
 					if err := c.processAlterTableQuery(
-						ctx, catalogPool, otelManager, req, alterTableStmt,
+						ctx, catalogPool, otelManager, req, s,
 						string(ev.Schema), binlogRowMetadataSupported, req.InternalVersion); err != nil {
 						return fmt.Errorf("failed to process ALTER TABLE query: %w", err)
 					}
+				case *ast.RenameTableStmt:
+					c.processRenameTableQuery(ctx, otelManager, req, s, string(ev.Schema))
 				}
 			}
 		case *replication.RowsEvent:
@@ -1025,6 +1034,72 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 	return nil
 }
 
+// processRenameTableQuery detects online schema-migration tools (gh-ost,
+// pt-online-schema-change, ...). These never issue an ALTER on the tracked
+// table; instead they build a shadow/ghost copy with the new schema and
+// atomically rename it into place, e.g.:
+//
+//	RENAME TABLE users TO _users_del, _users_gho TO users
+//
+// The resulting schema change surfaces to us only later, as row-event metadata
+// (see processTableMapEventSchema). Here we just meter how often a tracked table
+// is renamed-into, so we can gauge how prevalent these tools are
+func (c *MySqlConnector) processRenameTableQuery(
+	ctx context.Context,
+	otelManager *otel_metrics.OtelManager,
+	req *model.PullRecordsRequest[model.RecordItems],
+	stmt *ast.RenameTableStmt,
+	stmtSchema string,
+) {
+	for _, t2t := range stmt.TableToTables {
+		if t2t.NewTable == nil || t2t.OldTable == nil {
+			continue
+		}
+
+		// if the rename target has no schema, fall back to the one on the event
+		newSchemaName := t2t.NewTable.Schema.String()
+		if newSchemaName == "" {
+			newSchemaName = stmtSchema
+		}
+		newTableName := newSchemaName + "." + t2t.NewTable.Name.String()
+
+		// only care about renames that land on a table we are replicating
+		if _, tracked := req.TableNameMapping[newTableName]; !tracked {
+			continue
+		}
+
+		oldTableName := t2t.OldTable.Name.String()
+		tool := classifyOnlineSchemaMigrationTool(oldTableName, t2t.NewTable.Name.String())
+
+		c.logger.Info("table atomically renamed into a tracked table, likely an online schema migration",
+			slog.String("table", newTableName),
+			slog.String("renamedFrom", oldTableName),
+			slog.String("tool", tool))
+
+		otelManager.Metrics.OnlineSchemaMigrationsCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet(
+			attribute.String(otel_metrics.SourcePeerType, "mysql"),
+			attribute.String(otel_metrics.OnlineSchemaMigrationTool, tool),
+		)))
+	}
+}
+
+// classifyOnlineSchemaMigrationTool guesses which online schema-change tool
+// performed a rename, based on the shadow table's naming convention:
+//   - gh-ost ghost table:                _<table>_gho
+//   - pt-online-schema-change new table: _<table>_new
+//
+// Anything else that renames into a tracked table is bucketed as "other".
+func classifyOnlineSchemaMigrationTool(oldTable, newTable string) string {
+	switch oldTable {
+	case "_" + newTable + "_gho":
+		return OnlineSchemaMigrationToolGhOst
+	case "_" + newTable + "_new":
+		return OnlineSchemaMigrationToolPtOsc
+	default:
+		return OnlineSchemaMigrationToolOther
+	}
+}
+
 func posToOffsetText(pos mysql.Position) string {
 	return fmt.Sprintf("!f:%s,%x", pos.Name, pos.Pos)
 }

flow/connectors/mysql/cdc_test.go

@@ -3,6 +3,7 @@ package connmysql
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"slices"
 	"strings"
 	"testing"
@@ -14,9 +15,15 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/metric"
+	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.temporal.io/sdk/log"
 
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/model"
+	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
@@ -155,6 +162,145 @@ func TestParseSQLParsesTrailingNull(t *testing.T) {
 	}
 }
 
+func TestClassifyOnlineSchemaMigrationTool(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		oldTable string
+		newTable string
+		want     string
+	}{
+		{"gh-ost", "_users_gho", "users", OnlineSchemaMigrationToolGhOst},
+		{"pt-osc", "_users_new", "users", OnlineSchemaMigrationToolPtOsc},
+		{"plain rename", "users_backup", "users", OnlineSchemaMigrationToolOther},
+		{"gh-ost del table is not the target", "_users_del", "users", OnlineSchemaMigrationToolOther},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			require.Equal(t, tc.want, classifyOnlineSchemaMigrationTool(tc.oldTable, tc.newTable))
+		})
+	}
+}
+
+func TestParseSQLParsesGhOstRename(t *testing.T) {
+	mysqlParser := parser.New()
+	// the atomic swap gh-ost issues at the end of a migration
+	query := []byte("RENAME TABLE `mydb`.`users` TO `mydb`.`_users_del`, `mydb`.`_users_gho` TO `mydb`.`users`")
+	stmts, warns, err := parseSQL(mysqlParser, query)
+	require.NoError(t, err)
+	require.Empty(t, warns)
+	require.Len(t, stmts, 1)
+
+	rename, ok := stmts[0].(*ast.RenameTableStmt)
+	require.True(t, ok)
+	require.Len(t, rename.TableToTables, 2)
+
+	// the rename that lands on the tracked table is the ghost table swap
+	swap := rename.TableToTables[1]
+	require.Equal(t, "users", swap.NewTable.Name.String())
+	require.Equal(t, "_users_gho", swap.OldTable.Name.String())
+	require.Equal(t, OnlineSchemaMigrationToolGhOst,
+		classifyOnlineSchemaMigrationTool(swap.OldTable.Name.String(), swap.NewTable.Name.String()))
+}
+
+// newMetricsTestOtelManager builds an OtelManager backed by a ManualReader so tests
+// can collect and assert recorded metric values without a real exporter.
+func newMetricsTestOtelManager(t *testing.T) (*otel_metrics.OtelManager, *sdkmetric.ManualReader) {
+	t.Helper()
+	reader := sdkmetric.NewManualReader()
+	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
+	om := &otel_metrics.OtelManager{
+		MetricsProvider:    provider,
+		Meter:              provider.Meter("test"),
+		Int64CountersCache: make(map[string]metric.Int64Counter),
+	}
+	counter, err := om.GetOrInitInt64Counter(otel_metrics.BuildMetricName(otel_metrics.OnlineSchemaMigrationsName))
+	require.NoError(t, err)
+	om.Metrics.OnlineSchemaMigrationsCounter = counter
+	t.Cleanup(func() { _ = provider.Shutdown(context.Background()) })
+	return om, reader
+}
+
+// collectOnlineSchemaMigrationCounts collects the online_schema_migrations counter and
+// returns its value bucketed by the `tool` attribute.
+func collectOnlineSchemaMigrationCounts(t *testing.T, ctx context.Context, reader *sdkmetric.ManualReader) map[string]int64 {
+	t.Helper()
+	var rm metricdata.ResourceMetrics
+	require.NoError(t, reader.Collect(ctx, &rm))
+
+	wantName := otel_metrics.BuildMetricName(otel_metrics.OnlineSchemaMigrationsName)
+	counts := make(map[string]int64)
+	for _, sm := range rm.ScopeMetrics {
+		for _, m := range sm.Metrics {
+			if m.Name != wantName {
+				continue
+			}
+			sum, ok := m.Data.(metricdata.Sum[int64])
+			require.True(t, ok, "expected Int64 sum for %s", wantName)
+			for _, dp := range sum.DataPoints {
+				tool, _ := dp.Attributes.Value(otel_metrics.OnlineSchemaMigrationTool)
+				source, _ := dp.Attributes.Value(otel_metrics.SourcePeerType)
+				require.Equal(t, "mysql", source.AsString())
+				counts[tool.AsString()] += dp.Value
+			}
+		}
+	}
+	return counts
+}
+
+func TestProcessRenameTableQueryMetric(t *testing.T) {
+	ctx := t.Context()
+	c := &MySqlConnector{logger: log.NewStructuredLogger(slog.Default())}
+
+	parseRename := func(query string) *ast.RenameTableStmt {
+		t.Helper()
+		stmts, _, err := parseSQL(parser.New(), []byte(query))
+		require.NoError(t, err)
+		require.Len(t, stmts, 1)
+		rename, ok := stmts[0].(*ast.RenameTableStmt)
+		require.True(t, ok)
+		return rename
+	}
+
+	for _, tc := range []struct {
+		name  string
+		query string
+		want  map[string]int64
+	}{
+		{
+			name: "gh-ost cutover on tracked table increments once",
+			// full gh-ost atomic swap: only the `_users_gho -> users` pair lands on a tracked table
+			query: "RENAME TABLE `mydb`.`users` TO `mydb`.`_users_del`, `mydb`.`_users_gho` TO `mydb`.`users`",
+			want:  map[string]int64{OnlineSchemaMigrationToolGhOst: 1},
+		},
+		{
+			name:  "pt-online-schema-change cutover",
+			query: "RENAME TABLE `mydb`.`users` TO `mydb`.`_users_old`, `mydb`.`_users_new` TO `mydb`.`users`",
+			want:  map[string]int64{OnlineSchemaMigrationToolPtOsc: 1},
+		},
+		{
+			name: "schema falls back to event schema",
+			// no schema qualifier on the statement; resolved from the event's schema
+			query: "RENAME TABLE `users` TO `_users_del`, `_users_gho` TO `users`",
+			want:  map[string]int64{OnlineSchemaMigrationToolGhOst: 1},
+		},
+		{
+			name: "rename into untracked table is ignored",
+			// neither target is a table we replicate
+			query: "RENAME TABLE `mydb`.`orders` TO `mydb`.`orders_archive`",
+			want:  map[string]int64{},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			om, reader := newMetricsTestOtelManager(t)
+			req := &model.PullRecordsRequest[model.RecordItems]{
+				TableNameMapping: map[string]model.NameAndExclude{"mydb.users": {Name: "users_dst"}},
+			}
+			c.processRenameTableQuery(ctx, om, req, parseRename(tc.query), "mydb")
+
+			require.Equal(t, tc.want, collectOnlineSchemaMigrationCounts(t, ctx, reader))
+		})
+	}
+}
+
 func TestGetTableSchemaCaseSensitiveIdentifiers(t *testing.T) {
 	t.Parallel()
 	ctx := t.Context()

flow/otel_metrics/attributes.go

@@ -49,6 +49,7 @@ const (
 	TypeChangeFromKey          = "from"
 	TypeChangeToKey            = "to"
 	SourceEventTypeKey         = "sourceEventType"
+	OnlineSchemaMigrationTool  = "tool"
 )
 
 const (

flow/otel_metrics/otel_manager.go

@@ -82,6 +82,7 @@ const (
 	ServerWalEndLagGaugeName             = "wal_end_lag"
 	UsedMySQLCharsetsName                = "used_mysql_charsets"
 	ColumnTypeChangesName                = "column_type_changes"
+	OnlineSchemaMigrationsName           = "online_schema_migrations"
 )
 
 type Metrics struct {
@@ -138,6 +139,7 @@ type Metrics struct {
 	ServerWalEndLagGauge             metric.Int64Gauge
 	UsedMySQLCharsetsCounter         metric.Int64Counter
 	ColumnTypeChangesCounter         metric.Int64Counter
+	OnlineSchemaMigrationsCounter    metric.Int64Counter
 }
 
 type SlotMetricGauges struct {
@@ -592,6 +594,15 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error {
 		return err
 	}
 
+	if om.Metrics.OnlineSchemaMigrationsCounter, err = om.GetOrInitInt64Counter(BuildMetricName(OnlineSchemaMigrationsName),
+		metric.WithDescription(
+			"Counter of online schema migrations detected on the CDC path, i.e. a tracked table being atomically "+
+				"renamed into by a shadow/ghost table, with `source` label holding the source peer type and `tool` "+
+				"label holding the detected migration tool (gh-ost, pt-online-schema-change, other)"),
+	); err != nil {
+		return err
+	}
+
 	if CodeNotificationCounter, err = om.GetOrInitInt64Counter(BuildMetricName(CodeNotificationCounterName),
 		metric.WithDescription("One-off notifications with unique `message` attribute, triggers generic non-paging alert"),
 	); err != nil {