diff --git a/flow/alerting/classifier.go b/flow/alerting/classifier.go index 5ad5ba3d2..0b0dca9b5 100644 --- a/flow/alerting/classifier.go +++ b/flow/alerting/classifier.go @@ -145,6 +145,9 @@ var ( ErrorNotifyBinlogEventExceededMaxAllowedPacket = ErrorClass{ Class: "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET", action: NotifyUser, } + ErrorNotifyBinlogPartialRowEventUnsupported = ErrorClass{ + Class: "NOTIFY_BINLOG_PARTIAL_ROW_EVENT_UNSUPPORTED", action: NotifyUser, + } ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{ Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser, } @@ -1154,6 +1157,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) { } } + if _, ok := errors.AsType[*exceptions.MySQLUnsupportedPartialRowEventError](err); ok { + return ErrorNotifyBinlogPartialRowEventUnsupported, ErrorInfo{ + Source: ErrorSourceMySQL, + Code: "UNSUPPORTED_PARTIAL_ROW_EVENT", + } + } + if mysqlGeometryParseError, ok := errors.AsType[*exceptions.MySQLGeometryParseError](err); ok && strings.Contains(mysqlGeometryParseError.Error(), mysqlGeometryLinearRingNotClosedError) { return ErrorUnsupportedDatatype, ErrorInfo{ diff --git a/flow/alerting/classifier_test.go b/flow/alerting/classifier_test.go index 0c1dcb7f0..fa57f9b2f 100644 --- a/flow/alerting/classifier_test.go +++ b/flow/alerting/classifier_test.go @@ -1243,3 +1243,15 @@ func TestMySQLBinlogIncidentErrorShouldBeNotifyBinlogInvalid(t *testing.T) { Code: "BINLOG_INCIDENT", }, errInfo) } + +func TestMySQLUnsupportedPartialRowEventShouldBeNotifyPartialRowEventUnsupported(t *testing.T) { + t.Parallel() + + err := exceptions.NewMySQLUnsupportedPartialRowEventError(172, "e2e_test", "partial_rows") + errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("pulling records failed: %w", err)) + assert.Equal(t, ErrorNotifyBinlogPartialRowEventUnsupported, errorClass) + assert.Equal(t, ErrorInfo{ + Source: ErrorSourceMySQL, + Code: "UNSUPPORTED_PARTIAL_ROW_EVENT", + }, errInfo) +} diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 8a3132b7a..64ed9336d 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -12,6 +12,7 @@ import ( "log/slog" "math/rand/v2" "slices" + "strings" "sync/atomic" "time" @@ -45,6 +46,46 @@ const ( binlogStalenessMultiplier = 3 ) +const mariadbPartialRowDataEvent replication.EventType = 172 + +const ( + partialRowsHeaderLen = 4 + 4 + 1 + partialRowFlagOrigEventSize = 0x01 + binlogCommonHeaderLen = 19 + rowsEventTableIDSize = 6 +) + +// parsePartialRowEventTableID extracts the source table id from a MariaDB PARTIAL_ROW_DATA_EVENT +// +// [0:4] total_fragments (uint32 LE) +// [4:8] seq_no (uint32 LE, 1-based) +// [8] flags (bit 0 = FL_ORIG_EVENT_SIZE) +// [9:17] original size (present only when FL_ORIG_EVENT_SIZE is set, i.e. on the first fragment) +// then content (first fragment: embedded rows event = 19-byte header + post-header) +// +// The embedded rows-event post-header begins with a 6-byte little-endian table id. +func parsePartialRowEventTableID(data []byte) (uint64, uint32, bool) { + if len(data) < partialRowsHeaderLen { + return 0, 0, false + } + totalFragments := binary.LittleEndian.Uint32(data[0:4]) + seqNo := binary.LittleEndian.Uint32(data[4:8]) + flags := data[8] + if seqNo != 1 { + // continuation fragment: carries only raw row data, no embedded header, so no table id + return 0, totalFragments, false + } + contentStart := partialRowsHeaderLen + if flags&partialRowFlagOrigEventSize != 0 { + contentStart += 8 + } + tableIDStart := contentStart + binlogCommonHeaderLen + if len(data) < tableIDStart+rowsEventTableIDSize { + return 0, totalFragments, false + } + return mysql.FixedLengthInt(data[tableIDStart : tableIDStart+rowsEventTableIDSize]), totalFragments, true +} + const ( queryStatusVarFlags2 = 0 queryStatusVarSQLMode = 1 @@ -537,9 +578,22 @@ func (c *MySqlConnector) PullRecords( time.Now().UTC().Sub(time.UnixMicro(int64(commitTs))).Microseconds()) } } + recordUnsupportedEvent := func(ctx context.Context, event *replication.BinlogEvent, prefix string) { + otelManager.Metrics.UnsupportedBinlogEventCounter.Add(ctx, 1, metric.WithAttributeSet(attribute.NewSet( + attribute.String(otel_metrics.BinlogEventTypeKey, fmt.Sprintf("%s_%d", prefix, int(event.Header.EventType))), + ))) + if _, loaded := c.warnedUnsupportedEventTypes.LoadOrStore(event.Header.EventType, struct{}{}); !loaded { + c.logger.Warn("unsupported rows event", slog.Any("type", event.Header.EventType)) + } + } lastEventAt := time.Now() var mysqlParser *parser.Parser + // table id -> "schema.table", maintained from TABLE_MAP_EVENTs + tableIdToName := make(map[uint64]string) + // When we ignore an out-of-pipe fragmented rows event, its continuation fragments carry no table + // id; skip this many remaining PARTIAL_ROW_DATA_EVENTs before resolving table ids again. + var partialRowSkipFragments uint32 processEvent := func(event *replication.BinlogEvent) error { switch ev := event.Event.(type) { case *replication.GTIDEvent: @@ -566,12 +620,42 @@ func (c *MySqlConnector) PullRecords( c.logger.Info("rotate", slog.String("name", pos.Name), slog.Uint64("pos", uint64(pos.Pos))) } case *replication.GenericEvent: - // INCIDENT_EVENT (LOST_EVENTS) - fail and require resync - if event.Header.EventType == replication.INCIDENT_EVENT { + switch event.Header.EventType { + case replication.INCIDENT_EVENT: + // INCIDENT_EVENT (LOST_EVENTS) - fail and require resync incident, message := parseIncidentEvent(ev.Data) c.logger.Error("[mysql] received binlog incident event, resync required", slog.Uint64("incident", uint64(incident)), slog.String("message", message)) return exceptions.NewMySQLBinlogIncidentError(incident, message) + case mariadbPartialRowDataEvent: + if partialRowSkipFragments > 0 { + // continuation fragments of an out-of-pipe group we already decided to ignore + partialRowSkipFragments-- + break + } + tableID, totalFragments, ok := parsePartialRowEventTableID(ev.Data) + sourceTableName, known := tableIdToName[tableID] + if !ok || !known { + // couldn't recover/resolve the table, fail loudly + c.logger.Error("[mysql] received unresolvable MariaDB partial row data event, resync required", + slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent)), slog.Bool("parsed", ok)) + return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent), "", "") + } + if req.TableNameSchemaMapping[req.TableNameMapping[sourceTableName].Name] == nil { + // table not in the pipe - ignore this fragment group and its continuation fragments + if totalFragments > 1 { + partialRowSkipFragments = totalFragments - 1 + } + c.logger.Warn("[mysql] ignoring MariaDB partial row data event for table outside the pipe", + slog.String("table", sourceTableName), slog.Uint64("totalFragments", uint64(totalFragments))) + break + } + schemaName, tableName, _ := strings.Cut(sourceTableName, ".") + c.logger.Error("[mysql] received MariaDB partial row data event, resync required", + slog.Uint64("eventType", uint64(mariadbPartialRowDataEvent)), slog.String("table", sourceTableName)) + return exceptions.NewMySQLUnsupportedPartialRowEventError(byte(mariadbPartialRowDataEvent), schemaName, tableName) + default: + recordUnsupportedEvent(ctx, event, "Generic") } case *replication.QueryEvent: if !inTx && gset == nil && event.Header.LogPos > pos.Pos { @@ -603,6 +687,8 @@ func (c *MySqlConnector) PullRecords( c.processRenameTableQuery(ctx, otelManager, req, s, string(ev.Schema)) } } + case *replication.TableMapEvent: + tableIdToName[ev.TableID] = string(ev.Schema) + "." + string(ev.Table) case *replication.RowsEvent: sourceTableName := string(ev.Table.Schema) + "." + string(ev.Table.Table) // TODO this is fragile destinationTableName := req.TableNameMapping[sourceTableName].Name @@ -787,6 +873,8 @@ func (c *MySqlConnector) PullRecords( } case replication.WRITE_ROWS_EVENTv0, replication.UPDATE_ROWS_EVENTv0, replication.DELETE_ROWS_EVENTv0: return fmt.Errorf("mysql v0 replication protocol not supported") + default: + recordUnsupportedEvent(ctx, event, "Rows") } } if event.Header.Timestamp > 0 { @@ -795,6 +883,8 @@ func (c *MySqlConnector) PullRecords( int64(event.Header.Timestamp), ) } + default: + recordUnsupportedEvent(ctx, event, "Untyped") } return nil diff --git a/flow/connectors/mysql/cdc_test.go b/flow/connectors/mysql/cdc_test.go index 7ee420420..a163302bf 100644 --- a/flow/connectors/mysql/cdc_test.go +++ b/flow/connectors/mysql/cdc_test.go @@ -2,6 +2,7 @@ package connmysql import ( "context" + "encoding/hex" "fmt" "log/slog" "slices" @@ -438,3 +439,35 @@ func TestIntegrationGetTableSchemaPrimaryKeyVariants(t *testing.T) { }) } } + +func TestParsePartialRowEventTableID(t *testing.T) { + t.Parallel() + + // Real PARTIAL_ROW_DATA_EVENT payloads captured from MariaDB 12.3 + decode := func(s string) []byte { + b, err := hex.DecodeString(s) + require.NoError(t, err) + return b + } + // first fragment: total=9, seq=1, flags=FL_ORIG_EVENT_SIZE, then origSize + embedded rows event + firstFragment := decode("09000000010000000126200000000000002cc0476a" + + "17010000002a20000000000000000012000000000001000203fc0100000000200000") + // a continuation fragment: total=9, seq=2, no flags, raw row data + continuationFragment := decode("09000000020000000079797979797979797979") + + tableID, total, ok := parsePartialRowEventTableID(firstFragment) + require.True(t, ok, "first fragment table id should be recoverable") + require.Equal(t, uint64(18), tableID) + require.Equal(t, uint32(9), total) + + _, total, ok = parsePartialRowEventTableID(continuationFragment) + require.False(t, ok, "continuation fragment carries no table id") + require.Equal(t, uint32(9), total, "total_fragments is still readable on continuation fragments") + + // too short for even the post-header + _, _, ok = parsePartialRowEventTableID([]byte{0x09, 0x00, 0x00}) + require.False(t, ok) + // post-header present but content truncated before the embedded table id + _, _, ok = parsePartialRowEventTableID(firstFragment[:partialRowsHeaderLen+8+binlogCommonHeaderLen+2]) + require.False(t, ok) +} diff --git a/flow/connectors/mysql/mysql.go b/flow/connectors/mysql/mysql.go index 81f33060e..281e218ad 100644 --- a/flow/connectors/mysql/mysql.go +++ b/flow/connectors/mysql/mysql.go @@ -31,18 +31,19 @@ import ( type MySqlConnector struct { *metadataStore.PostgresMetadata - config *protos.MySqlConfig - ssh *utils.SSHTunnel - conn atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe - contexts atomic.Pointer[chan context.Context] - logger log.Logger - rdsAuth *utils.RDSAuth - serverVersion string - collationCharset atomic.Pointer[map[uint64]string] - warnedCharsets sync.Map - binlogHeartbeatPeriod time.Duration - totalBytesRead atomic.Int64 - deltaBytesRead atomic.Int64 + config *protos.MySqlConfig + ssh *utils.SSHTunnel + conn atomic.Pointer[client.Conn] // atomic used for internal concurrency, connector interface is not threadsafe + contexts atomic.Pointer[chan context.Context] + logger log.Logger + rdsAuth *utils.RDSAuth + serverVersion string + collationCharset atomic.Pointer[map[uint64]string] + warnedUnsupportedEventTypes sync.Map + warnedCharsets sync.Map + binlogHeartbeatPeriod time.Duration + totalBytesRead atomic.Int64 + deltaBytesRead atomic.Int64 } func NewMySqlConnector(ctx context.Context, config *protos.MySqlConfig) (*MySqlConnector, error) { diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 7d7a45bc8..7f76d67a9 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -13,15 +13,12 @@ import ( "github.com/go-mysql-org/go-mysql/mysql" "github.com/google/uuid" "github.com/stretchr/testify/require" - "github.com/testcontainers/testcontainers-go" - "github.com/testcontainers/testcontainers-go/wait" "github.com/PeerDB-io/peerdb/flow/connectors" connclickhouse "github.com/PeerDB-io/peerdb/flow/connectors/clickhouse" "github.com/PeerDB-io/peerdb/flow/generated/protos" "github.com/PeerDB-io/peerdb/flow/internal" "github.com/PeerDB-io/peerdb/flow/pkg/clickhouse" - "github.com/PeerDB-io/peerdb/flow/pkg/common" mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql" "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/types" @@ -1625,57 +1622,11 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { s.t.Skip("binlog incident injection requires a MySQL debug build; not available for MariaDB") } - req := testcontainers.ContainerRequest{ - Image: "ghcr.io/peerdb-io/mysql-debug:8.0.46", - Env: map[string]string{ - "MYSQL_ROOT_PASSWORD": internal.MySQLTestRootPasswordWithFallback("cipass"), - "MYSQL_ROOT_HOST": "%", - }, - // Keep the debug server small for the one-off testcontainers - Cmd: []string{ - "mysqld", - "--server-id=1", - "--log-bin=mysql-bin", - "--binlog-format=ROW", - "--innodb-buffer-pool-size=64M", - "--performance-schema=OFF", - "--mysqlx=0", - "--max-connections=20", - "--table-open-cache=64", - "--table-definition-cache=128", - "--innodb-log-buffer-size=8M", - }, - ExposedPorts: []string{"3306/tcp"}, - WaitingFor: wait.ForListeningPort("3306/tcp").WithStartupTimeout(3 * time.Minute), - } - - ctr, err := testcontainers.GenericContainer(s.t.Context(), testcontainers.GenericContainerRequest{ - ContainerRequest: req, - Started: true, - }) - testcontainers.CleanupContainer(s.t, ctr, testcontainers.StopTimeout(30*time.Second)) - require.NoError(s.t, err) - - mapped, err := ctr.MappedPort(s.t.Context(), "3306/tcp") - require.NoError(s.t, err) - port, err := strconv.Atoi(mapped.Port()) - require.NoError(s.t, err) - - suffix := "mydbginc_" + strings.ToLower(common.RandomString(8)) - config := &protos.MySqlConfig{ - // host.docker.internal resolves both from the test process (to the published port) and - // from the flow worker container (via host-gateway), unlike the container's own host. - Host: internal.MySQLTestHost(), - Port: uint32(port), - User: "root", - Password: internal.MySQLTestRootPasswordWithFallback("cipass"), - DisableTls: true, + src, suffix := SetupMySQLTestContainerSource(s.t, "mydbginc", MySQLTestContainerConfig{ + Image: "ghcr.io/peerdb-io/mysql-debug:8.0.46", Flavor: protos.MySqlFlavor_MYSQL_MYSQL, ReplicationMechanism: mySource.Config.ReplicationMechanism, - } - src, err := setupMyConnector(s.t, suffix, config, "mysql_debug_"+suffix) - require.NoError(s.t, err) - s.t.Cleanup(func() { src.Teardown(s.t, context.Background(), suffix) }) + }) srcTableName := "incident" srcFullName := fmt.Sprintf("e2e_test_%s.%s", suffix, srcTableName) @@ -1724,6 +1675,87 @@ func (s ClickHouseSuite) Test_MySQL_BinlogIncident() { RequireEnvCanceled(s.t, env) } +// Test_MariaDB_PartialRowEvent verifies how the mirror handles a MariaDB PARTIAL_ROW_DATA_EVENT +// (MariaDB 12.3+) - an oversized rows event fragmented across multiple binlog events. +func (s ClickHouseSuite) Test_MariaDB_PartialRowEvent() { + if s.cluster { + s.t.Skip("source-side partial row event coverage does not need to run against ClickHouse cluster") + } + mySource, ok := s.source.(*MySqlSource) + if !ok { + s.t.Skip("only applies to mysql") + } + if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MYSQL { + s.t.Skip("only applies to maria flavor") + } + + // binlog_row_event_fragment_threshold is pinned to its 1024-byte minimum so a modest row + // overflows a single rows event and MariaDB splits it into PARTIAL_ROW_DATA_EVENT fragments. + src, suffix := SetupMySQLTestContainerSource(s.t, "mdbpart", MySQLTestContainerConfig{ + Image: "mariadb:12.3", + Flavor: protos.MySqlFlavor_MYSQL_MARIA, + ReplicationMechanism: protos.MySqlReplicationMechanism_MYSQL_GTID, + ExtraServerFlags: []string{ + "--binlog-row-metadata=FULL", + "--binlog-row-event-fragment-threshold=1024", + }, + }) + + srcTableName := "partial_rows" + srcFullName := fmt.Sprintf("e2e_test_%s.%s", suffix, srcTableName) + dstTableName := "partial_rows_dst" + + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `CREATE TABLE %s (id INT PRIMARY KEY, payload LONGTEXT)`, srcFullName))) + + connectionGen := FlowConnectionGenerationConfig{ + FlowJobName: "test_mariadb_partial_row_" + suffix, + TableMappings: []*protos.TableMapping{{ + SourceTableIdentifier: srcFullName, + DestinationTableIdentifier: dstTableName, + ShardingKey: "id", + }}, + Destination: s.Peer().Name, + } + flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) + // The suite source is the shared CI MySQL; point the mirror at our MariaDB source instead. + flowConnConfig.SourceName = src.GeneratePeer(s.t).Name + + tc := NewTemporalClient(s.t) + env := ExecutePeerflow(s.t, tc, flowConnConfig) + SetupCDCFlowStatusQuery(s.t, env, flowConnConfig) + + catalogPool, err := internal.GetCatalogConnectionPoolFromEnv(s.t.Context()) + require.NoError(s.t, err) + + // A fragmented rows event on a table OUTSIDE the pipe must NOT fail the mirror + ignoredFullName := fmt.Sprintf("e2e_test_%s.partial_rows_ignored", suffix) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `CREATE TABLE %s (id INT PRIMARY KEY, payload LONGTEXT)`, ignoredFullName))) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (1, REPEAT('y', 8192))`, ignoredFullName))) + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (1, 'small')`, srcFullName))) + EnvWaitForCount(env, s, "waiting for in-pipe row past out-of-pipe partial row event", + dstTableName, "id,payload", 1) + + // An oversized row larger than the 1024-byte fragment threshold on the mirrored table itself is + // fragmented into PARTIAL_ROW_DATA_EVENTs, which we don't reassemble and must fail loudly on. + require.NoError(s.t, src.Exec(s.t.Context(), fmt.Sprintf( + `INSERT INTO %s (id, payload) VALUES (2, REPEAT('x', 8192))`, srcFullName))) + EnvWaitFor(s.t, env, 3*time.Minute, "waiting for partial row event error", func() bool { + count, err := GetLogCount(s.t.Context(), catalogPool, flowConnConfig.FlowJobName, "error", "fragmented oversized row events") + if err != nil { + s.t.Log("Error querying flow_errors:", err) + return false + } + return count > 0 + }) + + env.Cancel(s.t.Context()) + RequireEnvCanceled(s.t, env) +} + func (s ClickHouseSuite) Test_MySQL_Default_Partition_Key_Parallel_Snapshot() { if _, ok := s.source.(*MySqlSource); !ok { s.t.Skip("only applies to mysql") diff --git a/flow/e2e/mysql.go b/flow/e2e/mysql.go index b655c4134..752296410 100644 --- a/flow/e2e/mysql.go +++ b/flow/e2e/mysql.go @@ -3,7 +3,14 @@ package e2e import ( "context" "fmt" + "strconv" + "strings" "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/testcontainers/testcontainers-go" + "github.com/testcontainers/testcontainers-go/wait" "github.com/PeerDB-io/peerdb/flow/connectors" connmysql "github.com/PeerDB-io/peerdb/flow/connectors/mysql" @@ -34,6 +41,98 @@ func SetupMariaDB(t *testing.T, suffix string) (*MySqlSource, error) { return setupMyConnector(t, suffix, internal.GetMariaDBConfigFromEnv(flavor, replication), "mariadb") } +// MySQLTestContainerConfig parameterizes a throwaway MySQL/MariaDB testcontainer source. +type MySQLTestContainerConfig struct { + Image string + // ExtraServerFlags are appended to a common small-footprint server flag base (which already + // includes flavor-aware low-resource flags), e.g. "--binlog-row-event-fragment-threshold=1024". + ExtraServerFlags []string + Flavor protos.MySqlFlavor + ReplicationMechanism protos.MySqlReplicationMechanism +} + +// SetupMySQLTestContainerSource starts a throwaway MySQL/MariaDB server in a testcontainer and +// returns a MySqlSource pointed at it, plus the generated suffix used for its e2e database. It +// registers container and database cleanup on t. Use it to replace the shared CI source with an +// isolated server a test can reconfigure (custom image, extra server flags) - typically by assigning +// the returned source's peer to flowConnConfig.SourceName. +func SetupMySQLTestContainerSource( + t *testing.T, namePrefix string, cfg MySQLTestContainerConfig, +) (*MySqlSource, string) { + t.Helper() + + rootPassword := internal.MySQLTestRootPasswordWithFallback("cipass") + env := map[string]string{} + if cfg.Flavor == protos.MySqlFlavor_MYSQL_MARIA { + env["MARIADB_ROOT_PASSWORD"] = rootPassword + env["MARIADB_ROOT_HOST"] = "%" + } else { + env["MYSQL_ROOT_PASSWORD"] = rootPassword + env["MYSQL_ROOT_HOST"] = "%" + } + + // Common small-footprint server flags valid on both MySQL 8 and MariaDB. Both official + // entrypoints prepend the server binary when the first arg starts with "-", so none is needed. + // These throwaway servers run alongside every other container on the machine, so keep their + // resource usage low. + cmd := []string{ + "--server-id=1", + "--log-bin=mysql-bin", + "--binlog-format=ROW", + "--innodb-buffer-pool-size=64M", + "--innodb-log-buffer-size=8M", + "--performance-schema=OFF", + "--max-connections=20", + "--table-open-cache=64", + "--table-definition-cache=128", + } + // Flavor-specific low-resource flags for options that only exist on one server. + if cfg.Flavor == protos.MySqlFlavor_MYSQL_MARIA { + cmd = append(cmd, "--aria-pagecache-buffer-size=8M") + } else { + cmd = append(cmd, "--mysqlx=0") + } + cmd = append(cmd, cfg.ExtraServerFlags...) + + req := testcontainers.ContainerRequest{ + Image: cfg.Image, + Env: env, + Cmd: cmd, + ExposedPorts: []string{"3306/tcp"}, + WaitingFor: wait.ForListeningPort("3306/tcp").WithStartupTimeout(3 * time.Minute), + } + + ctr, err := testcontainers.GenericContainer(t.Context(), testcontainers.GenericContainerRequest{ + ContainerRequest: req, + Started: true, + }) + testcontainers.CleanupContainer(t, ctr, testcontainers.StopTimeout(30*time.Second)) + require.NoError(t, err) + + mapped, err := ctr.MappedPort(t.Context(), "3306/tcp") + require.NoError(t, err) + port, err := strconv.ParseUint(mapped.Port(), 10, 32) + require.NoError(t, err) + + suffix := namePrefix + "_" + strings.ToLower(common.RandomString(8)) + config := &protos.MySqlConfig{ + // host.docker.internal resolves both from the test process (to the published port) and from + // the flow worker container (via host-gateway), unlike the container's own hostname. + Host: internal.MySQLTestHost(), + Port: uint32(port), + User: "root", + Password: rootPassword, + DisableTls: true, + Flavor: cfg.Flavor, + ReplicationMechanism: cfg.ReplicationMechanism, + } + src, err := setupMyConnector(t, suffix, config, suffix) + require.NoError(t, err) + t.Cleanup(func() { src.Teardown(t, context.Background(), suffix) }) + + return src, suffix +} + func setupMyConnector(t *testing.T, suffix string, config *protos.MySqlConfig, peerName string) (*MySqlSource, error) { t.Helper() connector, err := connmysql.NewMySqlConnector(t.Context(), config) diff --git a/flow/otel_metrics/attributes.go b/flow/otel_metrics/attributes.go index 67af98133..46827ae52 100644 --- a/flow/otel_metrics/attributes.go +++ b/flow/otel_metrics/attributes.go @@ -50,6 +50,7 @@ const ( TypeChangeToKey = "to" SourceEventTypeKey = "sourceEventType" OnlineSchemaMigrationTool = "tool" + BinlogEventTypeKey = "eventType" ) const ( diff --git a/flow/otel_metrics/otel_manager.go b/flow/otel_metrics/otel_manager.go index 1753d0f36..6f14033fc 100644 --- a/flow/otel_metrics/otel_manager.go +++ b/flow/otel_metrics/otel_manager.go @@ -83,6 +83,7 @@ const ( UsedMySQLCharsetsName = "used_mysql_charsets" ColumnTypeChangesName = "column_type_changes" OnlineSchemaMigrationsName = "online_schema_migrations" + UnsupportedBinlogEventName = "unsupported_binlog_event" ) type Metrics struct { @@ -140,6 +141,7 @@ type Metrics struct { UsedMySQLCharsetsCounter metric.Int64Counter ColumnTypeChangesCounter metric.Int64Counter OnlineSchemaMigrationsCounter metric.Int64Counter + UnsupportedBinlogEventCounter metric.Int64Counter } type SlotMetricGauges struct { @@ -603,6 +605,14 @@ func (om *OtelManager) setupMetrics(ctx context.Context) error { return err } + if om.Metrics.UnsupportedBinlogEventCounter, err = om.GetOrInitInt64Counter(BuildMetricName(UnsupportedBinlogEventName), + metric.WithDescription( + "Counter of unsupported binlog generic events seen on the CDC path, with `eventType` label "+ + "holding the numeric binlog event type"), + ); 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 { diff --git a/flow/shared/exceptions/mysql.go b/flow/shared/exceptions/mysql.go index 907b6ee31..af3a38487 100644 --- a/flow/shared/exceptions/mysql.go +++ b/flow/shared/exceptions/mysql.go @@ -153,3 +153,25 @@ func (e *MySQLBinlogIncidentError) Error() string { } return fmt.Sprintf("MySQL binlog incident event received (incident=%d); a resync is required", e.Incident) } + +type MySQLUnsupportedPartialRowEventError struct { + Schema string + Table string + EventType byte +} + +func NewMySQLUnsupportedPartialRowEventError(eventType byte, schema string, table string) *MySQLUnsupportedPartialRowEventError { + return &MySQLUnsupportedPartialRowEventError{EventType: eventType, Schema: schema, Table: table} +} + +func (e *MySQLUnsupportedPartialRowEventError) Error() string { + if e.Schema != "" || e.Table != "" { + return fmt.Sprintf( + "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d) on table %s.%s: "+ + "fragmented oversized row events cannot be processed; a resync is required", + e.EventType, e.Schema, e.Table) + } + return fmt.Sprintf( + "unsupported MariaDB PARTIAL_ROW_DATA_EVENT (type %d): fragmented oversized row events cannot be processed; a resync is required", + e.EventType) +}