Emit metrics related to binlog processing

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
+}

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())
+}

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 {

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) {

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)
+}