diff --git a/go/binlog/binlog_dml_event.go b/go/binlog/binlog_dml_event.go index 626e5759e..3a27f28a4 100644 --- a/go/binlog/binlog_dml_event.go +++ b/go/binlog/binlog_dml_event.go @@ -65,3 +65,17 @@ func NewBinlogDMLEvent(databaseName, tableName string, dml EventDML) *BinlogDMLE func (bde *BinlogDMLEvent) String() string { return fmt.Sprintf("[%+v on %s:%s]", bde.DML, bde.DatabaseName, bde.TableName) } + +// EventTypeTag returns a lowercase DML type for metrics tags (insert/update/delete). +func (dml EventDML) EventTypeTag() string { + return strings.ToLower(string(dml)) +} + +// RowsInRowsEvent returns the number of logical rows in a binlog rows event. +// Update events encode two physical rows per logical row (before/after image). +func RowsInRowsEvent(rowCount int, dml EventDML) int { + if dml == UpdateDML { + return rowCount / 2 + } + return rowCount +} diff --git a/go/binlog/binlog_dml_event_test.go b/go/binlog/binlog_dml_event_test.go new file mode 100644 index 000000000..c6b00ab20 --- /dev/null +++ b/go/binlog/binlog_dml_event_test.go @@ -0,0 +1,24 @@ +/* + Copyright 2026 GitHub Inc. + See https://github.com/github/gh-ost/blob/master/LICENSE +*/ + +package binlog + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRowsInRowsEvent(t *testing.T) { + require.Equal(t, 2, RowsInRowsEvent(2, InsertDML)) + require.Equal(t, 1, RowsInRowsEvent(2, UpdateDML)) + require.Equal(t, 3, RowsInRowsEvent(3, DeleteDML)) +} + +func TestEventTypeTag(t *testing.T) { + require.Equal(t, "insert", InsertDML.EventTypeTag()) + require.Equal(t, "update", UpdateDML.EventTypeTag()) + require.Equal(t, "delete", DeleteDML.EventTypeTag()) +} diff --git a/go/binlog/gomysql_reader.go b/go/binlog/gomysql_reader.go index 189a5f399..f1edb3b8a 100644 --- a/go/binlog/gomysql_reader.go +++ b/go/binlog/gomysql_reader.go @@ -7,9 +7,11 @@ package binlog import ( "fmt" + "strings" "sync" "github.com/github/gh-ost/go/base" + "github.com/github/gh-ost/go/metrics" "github.com/github/gh-ost/go/mysql" "github.com/github/gh-ost/go/sql" @@ -31,6 +33,11 @@ type GoMySQLReader struct { // LastTrxCoords are the coordinates of the last transaction completely read. // If using the file coordinates it is binlog position of the transaction's XID event. LastTrxCoords mysql.BinlogCoordinates + + // Per-transaction counters for relevant (consumed) row events, flushed at transaction boundaries. + transactionRowEventTotal int64 + transactionRowTotal int64 + previousLastCommitted int64 } func NewGoMySQLReader(migrationContext *base.MigrationContext) *GoMySQLReader { @@ -85,12 +92,63 @@ func (gmr *GoMySQLReader) GetCurrentBinlogCoordinates() mysql.BinlogCoordinates return gmr.currentCoordinates.Clone() } +func (gmr *GoMySQLReader) isRelevantTable(databaseName, tableName string) bool { + if !strings.EqualFold(databaseName, gmr.migrationContext.DatabaseName) { + return false + } + if strings.EqualFold(tableName, gmr.migrationContext.OriginalTableName) { + return true + } + return strings.EqualFold(tableName, gmr.migrationContext.GetChangelogTableName()) +} + +func (gmr *GoMySQLReader) flushTransactionMetrics() { + if gmr.transactionRowEventTotal == 0 { + return + } + emit := gmr.migrationContext.Metrics + metrics.RecordBinlogTransactionSize(emit, gmr.transactionRowEventTotal, gmr.transactionRowTotal) + gmr.transactionRowEventTotal = 0 + gmr.transactionRowTotal = 0 +} + +func (gmr *GoMySQLReader) onGTIDEvent(event *replication.GTIDEvent) { + gmr.flushTransactionMetrics() + + emit := gmr.migrationContext.Metrics + if event.TransactionLength > 0 { + metrics.RecordGTIDTransactionLengthBytes(emit, event.TransactionLength) + } + if event.LastCommitted != gmr.previousLastCommitted { + metrics.RecordUnfilteredCommitGroupSize(emit, float64(event.SequenceNumber-event.LastCommitted)) + } + gmr.previousLastCommitted = event.LastCommitted +} + func (gmr *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent *replication.RowsEvent, entriesChannel chan<- *BinlogEntry) error { currentCoords := gmr.GetCurrentBinlogCoordinates() dml := ToEventDML(ev.Header.EventType.String()) if dml == NotDML { return fmt.Errorf("unknown DML type: %s", ev.Header.EventType.String()) } + + databaseName := string(rowsEvent.Table.Schema) + tableName := string(rowsEvent.Table.Table) + rowCount := int64(RowsInRowsEvent(len(rowsEvent.Rows), dml)) + eventType := dml.EventTypeTag() + emit := gmr.migrationContext.Metrics + metrics.RecordBinlogRowsEventProcessed(emit, tableName, eventType, rowCount) + relevant := gmr.isRelevantTable(databaseName, tableName) + if !relevant { + metrics.RecordBinlogRowsEventFiltered(emit, tableName, eventType, rowCount) + } else { + metrics.RecordBinlogRowsEventConsumed(emit, tableName, eventType, rowCount) + metrics.RecordBinlogRowsInEvent(emit, tableName, rowCount) + gmr.transactionRowEventTotal++ + gmr.transactionRowTotal += rowCount + } + + beforeChannel := time.Now() for i, row := range rowsEvent.Rows { if dml == UpdateDML && i%2 == 1 { // An update has two rows (WHERE+SET) @@ -125,6 +183,9 @@ func (gmr *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEvent // In reality, reads will be synchronous entriesChannel <- binlogEntry } + if relevant { + metrics.RecordBinlogStreamerBlockedOnOutChannel(emit, time.Since(beforeChannel)) + } return nil } @@ -160,6 +221,7 @@ func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChan if err != nil { return err } + gmr.onGTIDEvent(event) gmr.currentCoordinatesMutex.Lock() if gmr.LastTrxCoords != nil { gmr.currentCoordinates = gmr.LastTrxCoords.Clone() @@ -178,6 +240,9 @@ func (gmr *GoMySQLReader) StreamEvents(canStopStreaming func() bool, entriesChan gmr.migrationContext.Log.Infof("rotate to next log from %s:%d to %s", coords.LogFile, int64(ev.Header.LogPos), event.NextLogName) gmr.currentCoordinatesMutex.Unlock() case *replication.XIDEvent: + if !gmr.migrationContext.UseGTIDs { + gmr.flushTransactionMetrics() + } if gmr.migrationContext.UseGTIDs { gmr.LastTrxCoords = &mysql.GTIDBinlogCoordinates{GTIDSet: event.GSet.(*gomysql.MysqlGTIDSet)} } else { diff --git a/go/metrics/emit.go b/go/metrics/emit.go index 91472be73..dd1582a44 100644 --- a/go/metrics/emit.go +++ b/go/metrics/emit.go @@ -184,6 +184,84 @@ func RecordQueryDuration(emit Emitter, side string, kind string, duration time.D emit.Histogram("query.duration_milliseconds", float64(duration.Milliseconds()), "side:"+side, "kind:"+kind, "outcome:"+outcome) } +// RecordBinlogRowsEventProcessed emits gh_ost.binlog_events and gh_ost.binlog_rows for a +// rows event read from the binlog stream. +func RecordBinlogRowsEventProcessed(emit Emitter, tableName, binlogEventType string, rowCount int64) { + if emit == nil || tableName == "" || binlogEventType == "" || rowCount < 0 { + return + } + tableTag := "table:" + tableName + eventTypeTag := "binlog_event_type:" + binlogEventType + emit.Count("binlog_events", 1, "type:processed", tableTag) + emit.Count("binlog_rows", rowCount, "type:processed", tableTag, eventTypeTag) +} + +// RecordBinlogRowsEventFiltered emits gh_ost.binlog_events and gh_ost.binlog_rows for a +// rows event on a table that is not being migrated. +func RecordBinlogRowsEventFiltered(emit Emitter, tableName, binlogEventType string, rowCount int64) { + if emit == nil || tableName == "" || binlogEventType == "" || rowCount < 0 { + return + } + tableTag := "table:" + tableName + eventTypeTag := "binlog_event_type:" + binlogEventType + emit.Count("binlog_events", 1, "type:filtered", tableTag) + emit.Count("binlog_rows", rowCount, "type:filtered", tableTag, eventTypeTag) +} + +// RecordBinlogRowsEventConsumed emits gh_ost.binlog_events and gh_ost.binlog_rows for a +// rows event forwarded for application on a migrated table. +func RecordBinlogRowsEventConsumed(emit Emitter, tableName, binlogEventType string, rowCount int64) { + if emit == nil || tableName == "" || binlogEventType == "" || rowCount < 0 { + return + } + tableTag := "table:" + tableName + eventTypeTag := "binlog_event_type:" + binlogEventType + emit.Count("binlog_events", 1, "type:consumed", tableTag, eventTypeTag) + emit.Count("binlog_rows", rowCount, "type:consumed", tableTag, eventTypeTag) +} + +// RecordBinlogRowsInEvent emits gh_ost.binlog_rows_in_event for the row count in a forwarded event. +func RecordBinlogRowsInEvent(emit Emitter, tableName string, rowCount int64) { + if emit == nil || tableName == "" || rowCount < 0 { + return + } + emit.Histogram("binlog_rows_in_event", float64(rowCount), "table:"+tableName) +} + +// RecordBinlogTransactionSize emits row-event and row counts for a completed transaction +// on relevant (consumed) tables. +func RecordBinlogTransactionSize(emit Emitter, numRowEvents, numRows int64) { + if emit == nil || numRowEvents <= 0 || numRows < 0 { + return + } + emit.Histogram("transaction_num_row_events", float64(numRowEvents)) + emit.Histogram("transaction_num_rows", float64(numRows)) +} + +// RecordGTIDTransactionLengthBytes emits the transaction length from a GTID event (MySQL 8.0.2+). +func RecordGTIDTransactionLengthBytes(emit Emitter, length uint64) { + if emit == nil || length == 0 { + return + } + emit.Histogram("gtid_event_transaction_length_bytes", float64(length)) +} + +// RecordUnfilteredCommitGroupSize emits the parallel replication commit group size from a GTID event. +func RecordUnfilteredCommitGroupSize(emit Emitter, size float64) { + if emit == nil || size < 0 { + return + } + emit.Gauge("unfiltered_commit_group_size", size) +} + +// RecordBinlogStreamerBlockedOnOutChannel emits time spent blocked sending rows to the events channel. +func RecordBinlogStreamerBlockedOnOutChannel(emit Emitter, d time.Duration) { + if emit == nil || d < 0 { + return + } + emit.Histogram("binlog_streamer_blocked_on_out_channel_ms", float64(d.Milliseconds())) +} + // RecordSleep emits per-stage sleep/wait metrics (namespace is applied by the client): // gh_ost.sleep.duration_milliseconds and gh_ost.sleep.total_milliseconds, both tagged by stage. func RecordSleep(emit Emitter, stage string, d time.Duration) { diff --git a/go/metrics/emit_test.go b/go/metrics/emit_test.go index f2a1b5aba..875b49d04 100644 --- a/go/metrics/emit_test.go +++ b/go/metrics/emit_test.go @@ -404,3 +404,92 @@ func TestRecordSleepNilSafe(t *testing.T) { RecordSleep(&histogramCountSpy{}, "", time.Second) RecordSleep(&histogramCountSpy{}, "retry_backoff", -time.Second) } + +func TestRecordBinlogRowsEventMetrics(t *testing.T) { + spy := &histogramCountSpy{} + + RecordBinlogRowsEventProcessed(spy, "orders", "insert", 3) + RecordBinlogRowsEventFiltered(spy, "other_table", "update", 2) + RecordBinlogRowsEventConsumed(spy, "orders", "delete", 1) + RecordBinlogRowsInEvent(spy, "orders", 1) + + wantCounts := []struct { + name string + value int64 + tags []string + }{ + {"binlog_events", 1, []string{"type:processed", "table:orders"}}, + {"binlog_rows", 3, []string{"type:processed", "table:orders", "binlog_event_type:insert"}}, + {"binlog_events", 1, []string{"type:filtered", "table:other_table"}}, + {"binlog_rows", 2, []string{"type:filtered", "table:other_table", "binlog_event_type:update"}}, + {"binlog_events", 1, []string{"type:consumed", "table:orders", "binlog_event_type:delete"}}, + {"binlog_rows", 1, []string{"type:consumed", "table:orders", "binlog_event_type:delete"}}, + } + if len(spy.countNames) != len(wantCounts) { + t.Fatalf("got %d counts, want %d", len(spy.countNames), len(wantCounts)) + } + for i, want := range wantCounts { + if spy.countNames[i] != want.name || spy.countValues[i] != want.value { + t.Fatalf("[%d] got %s=%d, want %s=%d", i, spy.countNames[i], spy.countValues[i], want.name, want.value) + } + if !slices.Equal(spy.countTags[i], want.tags) { + t.Fatalf("[%d] got tags %#v, want %#v", i, spy.countTags[i], want.tags) + } + } + if len(spy.histogramNames) != 1 || spy.histogramNames[0] != "binlog_rows_in_event" || spy.histogramValues[0] != 1 { + t.Fatalf("got histogram %#v values %#v", spy.histogramNames, spy.histogramValues) + } + if !slices.Equal(spy.histogramTags[0], []string{"table:orders"}) { + t.Fatalf("got histogram tags %#v", spy.histogramTags[0]) + } +} + +func TestRecordBinlogRowsEventMetricsNilSafe(t *testing.T) { + RecordBinlogRowsEventProcessed(nil, "orders", "insert", 1) + RecordBinlogRowsEventFiltered(&histogramCountSpy{}, "", "insert", 1) + RecordBinlogRowsEventConsumed(&histogramCountSpy{}, "orders", "", 1) + RecordBinlogRowsInEvent(&histogramCountSpy{}, "", 1) + RecordBinlogRowsInEvent(&histogramCountSpy{}, "orders", -1) +} + +func TestRecordBinlogTransactionMetrics(t *testing.T) { + spy := &histogramCountSpy{} + gaugeSpy := &gaugeSpy{} + + RecordBinlogTransactionSize(spy, 3, 10) + RecordGTIDTransactionLengthBytes(spy, 4096) + RecordUnfilteredCommitGroupSize(gaugeSpy, 5) + RecordBinlogStreamerBlockedOnOutChannel(spy, 25*time.Millisecond) + + if len(spy.histogramNames) != 4 { + t.Fatalf("got %d histograms, want 4: %#v", len(spy.histogramNames), spy.histogramNames) + } + wantHistograms := []struct { + name string + value float64 + }{ + {"transaction_num_row_events", 3}, + {"transaction_num_rows", 10}, + {"gtid_event_transaction_length_bytes", 4096}, + {"binlog_streamer_blocked_on_out_channel_ms", 25}, + } + for i, want := range wantHistograms { + if spy.histogramNames[i] != want.name || spy.histogramValues[i] != want.value { + t.Fatalf("[%d] got %s=%v, want %s=%v", i, spy.histogramNames[i], spy.histogramValues[i], want.name, want.value) + } + } + if len(gaugeSpy.names) != 1 || gaugeSpy.names[0] != "unfiltered_commit_group_size" || gaugeSpy.values[0] != 5 { + t.Fatalf("got gauge %#v values %#v", gaugeSpy.names, gaugeSpy.values) + } +} + +func TestRecordBinlogTransactionMetricsNilSafe(t *testing.T) { + RecordBinlogTransactionSize(nil, 1, 1) + RecordBinlogTransactionSize(&histogramCountSpy{}, 0, 1) + RecordGTIDTransactionLengthBytes(nil, 100) + RecordGTIDTransactionLengthBytes(&histogramCountSpy{}, 0) + RecordUnfilteredCommitGroupSize(nil, 1) + RecordUnfilteredCommitGroupSize(&gaugeSpy{}, -1) + RecordBinlogStreamerBlockedOnOutChannel(nil, time.Second) + RecordBinlogStreamerBlockedOnOutChannel(&histogramCountSpy{}, -time.Second) +}