diff --git a/.github/workflows/start-mysql.sh b/.github/workflows/start-mysql.sh index 89b423f7..56c8ef61 100755 --- a/.github/workflows/start-mysql.sh +++ b/.github/workflows/start-mysql.sh @@ -1,12 +1,15 @@ #!/bin/bash set -xe +# mysql-3 is the promoted writer used by the master-failover integration test. +# It is started alongside the source/target so that test can run; other tests +# ignore it. if [ "$MYSQL_VERSION" == "8.0" ]; then - docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2 + docker compose -f docker-compose_8.0.yml up -d mysql-1 mysql-2 mysql-3 elif [ "$MYSQL_VERSION" == "8.4" ]; then - docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2 + docker compose -f docker-compose_8.4.yml up -d mysql-1 mysql-2 mysql-3 else - docker compose up -d mysql-1 mysql-2 + docker compose up -d mysql-1 mysql-2 mysql-3 fi MAX_ATTEMPTS=60 @@ -38,6 +41,8 @@ wait_for_configuration () { wait_for_version "ghostferry-mysql-1-1" wait_for_version "ghostferry-mysql-2-1" +wait_for_version "ghostferry-mysql-3-1" wait_for_configuration 29291 wait_for_configuration 29292 +wait_for_configuration 29293 diff --git a/.gitignore b/.gitignore index 3e50b25c..1c572c52 100644 --- a/.gitignore +++ b/.gitignore @@ -21,3 +21,6 @@ build/ .idea .ipynb_checkpoints + +# Local test tmpdir (macOS TMPDIR workaround) +.tmp_ghostferry/ diff --git a/binlog_coordinate.go b/binlog_coordinate.go index 2a989fdb..63dcecf8 100644 --- a/binlog_coordinate.go +++ b/binlog_coordinate.go @@ -261,6 +261,67 @@ func intersectGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) { return result, nil } +// unionGTIDStringInto returns a new GTID set equal to base with the GTID(s) in +// add merged in. base may be nil (treated as empty). It never mutates base. It +// is used by failover recovery to fold an in-flight transaction's GTID into the +// already-applied set before validating a candidate master. +func unionGTIDStringInto(base mysql.GTIDSet, add string) (mysql.GTIDSet, error) { + var result *mysql.MysqlGTIDSet + if base == nil { + empty, err := mysql.ParseMysqlGTIDSet("") + if err != nil { + return nil, err + } + result = empty.(*mysql.MysqlGTIDSet) + } else { + clone, ok := base.Clone().(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires a MySQL GTID set, got %T", base) + } + result = clone + } + + if err := result.Update(add); err != nil { + return nil, fmt.Errorf("GTID union: updating with %q: %w", add, err) + } + return result, nil +} + +// unionGTIDSets returns a new GTID set equal to a with b merged in. Either may +// be nil (treated as empty). It never mutates its inputs. It is used by failover +// recovery to require a candidate master to contain both the applied set and the +// cutover stop target. +func unionGTIDSets(a, b mysql.GTIDSet) (mysql.GTIDSet, error) { + var result *mysql.MysqlGTIDSet + if a == nil { + empty, err := mysql.ParseMysqlGTIDSet("") + if err != nil { + return nil, err + } + result = empty.(*mysql.MysqlGTIDSet) + } else { + clone, ok := a.Clone().(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", a) + } + result = clone + } + + if b != nil { + bMysql, ok := b.(*mysql.MysqlGTIDSet) + if !ok { + return nil, fmt.Errorf("GTID union requires MySQL GTID sets, got %T", b) + } + for _, uuidSet := range bMysql.Sets { + // Clone so result never aliases b's internal UUIDSet (AddSet stores + // the pointer directly when the SID is not already present). + result.AddSet(uuidSet.Clone()) + } + } + + return result, nil +} + // serializedBinlogCoordinate is the on-disk / on-wire shape of a // BinlogCoordinate. It is deliberately explicit and self-describing so that a // future GTID variant can be added as additional fields without breaking diff --git a/binlog_streamer.go b/binlog_streamer.go index 2819f938..72a9f3ca 100644 --- a/binlog_streamer.go +++ b/binlog_streamer.go @@ -52,6 +52,16 @@ type BinlogStreamer struct { // backwards compatibility. BinlogCoordinateMode BinlogCoordinateType + // MasterFailoverRecovery, when set together with SourceReconnector, enables + // automatic reconnection to a new source master on connection loss. Honored + // only in GTID mode (GTID sets are server-independent). Nil disables it. + MasterFailoverRecovery *MasterFailoverRecoveryConfig + + // SourceReconnector reconnects the source after a lost connection. It is + // supplied by the Ferry, which owns the SourceRuntime and therefore how the + // whole run repoints at the promoted writer. Required for failover recovery. + SourceReconnector SourceReconnector + lastStreamedBinlogPosition mysql.Position lastResumableBinlogPosition mysql.Position stopAtBinlogPosition mysql.Position @@ -74,6 +84,17 @@ type BinlogStreamer struct { lastResumableGTIDSet mysql.GTIDSet stopAtGTIDSet mysql.GTIDSet + // inFlightGTID is the GTID of the transaction currently streamed but not yet + // committed (set at GTIDEvent, cleared at its XIDEvent). Its rows may already + // have been emitted downstream before lastStreamedGTIDSet advances, so + // failover validation folds it into the applied set. Only touched by the Run + // goroutine. + inFlightGTID string + + // connMu guards DB and DBConfig, swapped by the Run goroutine during + // failover recovery while FlushAndStop (another goroutine) reads them. + connMu sync.Mutex + lastProcessedEventTime time.Time lastLagMetricEmittedTime time.Time @@ -100,29 +121,42 @@ func (s *BinlogStreamer) ensureLogger() { func (s *BinlogStreamer) createBinlogSyncer() error { var err error - var tlsConfig *tls.Config - if s.DBConfig.TLS != nil { - tlsConfig, err = s.DBConfig.TLS.BuildConfig() + if s.MyServerId == 0 { + s.MyServerId, err = s.generateNewServerId() if err != nil { + s.logger.WithError(err).Error("could not generate unique server_id") return err } } - if s.MyServerId == 0 { - s.MyServerId, err = s.generateNewServerId() + syncer, err := s.newBinlogSyncerFor(s.DBConfig, s.MyServerId) + if err != nil { + return err + } + s.binlogSyncer = syncer + return nil +} + +// newBinlogSyncerFor builds a BinlogSyncer targeting the given DB config and +// server id without mutating streamer state, so failover recovery can construct +// a candidate syncer before committing to it. +func (s *BinlogStreamer) newBinlogSyncerFor(dbConf *DatabaseConfig, serverID uint32) (*replication.BinlogSyncer, error) { + var tlsConfig *tls.Config + if dbConf.TLS != nil { + var err error + tlsConfig, err = dbConf.TLS.BuildConfig() if err != nil { - s.logger.WithError(err).Error("could not generate unique server_id") - return err + return nil, err } } syncerConfig := replication.BinlogSyncerConfig{ - ServerID: s.MyServerId, - Host: s.DBConfig.Host, - Port: s.DBConfig.Port, - User: s.DBConfig.User, - Password: s.DBConfig.Pass, + ServerID: serverID, + Host: dbConf.Host, + Port: dbConf.Port, + User: dbConf.User, + Password: dbConf.Pass, TLSConfig: tlsConfig, UseDecimal: true, UseFloatWithTrailingZero: true, @@ -130,8 +164,17 @@ func (s *BinlogStreamer) createBinlogSyncer() error { Logger: NewSlogLogger(s.logger), } - s.binlogSyncer = replication.NewBinlogSyncer(syncerConfig) - return nil + // When master-failover recovery is enabled, bound go-mysql's internal + // reconnect-to-the-same-host retries. Otherwise the syncer would retry the + // dead host forever and GetEvent would only ever surface context timeouts, + // so our failover recovery (which reconnects to a *different* host) would + // never be triggered. With the default (0) behavior is unchanged: infinite + // retries against the configured host, matching pre-failover Ghostferry. + if s.failoverRecoveryEnabled() { + syncerConfig.MaxReconnectAttempts = s.MasterFailoverRecovery.syncerMaxReconnectAttempts() + } + + return replication.NewBinlogSyncer(syncerConfig), nil } func (s *BinlogStreamer) ConnectBinlogStreamerToMysql() (mysql.Position, error) { @@ -261,6 +304,184 @@ func (s *BinlogStreamer) connectBinlogStreamerFromGTID(startFrom BinlogCoordinat return NewGTIDCoordinate(s.lastStreamedGTIDString()), nil } +// failoverRecoveryEnabled reports whether automatic master-failover recovery is +// configured and usable. It requires GTID mode and a reconnector. +func (s *BinlogStreamer) failoverRecoveryEnabled() bool { + return s.MasterFailoverRecovery != nil && + s.SourceReconnector != nil && + s.coordinateMode() == BinlogCoordinateGTID +} + +// currentDB returns the active source DB handle under the connection lock. Used +// by goroutines other than Run (FlushAndStop) so they observe a consistent +// handle across a failover swap. +func (s *BinlogStreamer) currentDB() *sql.DB { + s.connMu.Lock() + defer s.connMu.Unlock() + return s.DB +} + +// currentDBConfig returns the active DBConfig under the connection lock. +func (s *BinlogStreamer) currentDBConfig() *DatabaseConfig { + s.connMu.Lock() + defer s.connMu.Unlock() + return s.DBConfig +} + +// appliedGTIDSet returns everything the streamer has emitted downstream: the +// committed streamed set plus any in-flight transaction's GTID (whose rows may +// already have reached listeners before the commit advanced the streamed set). +// Failover validation must ensure a candidate contains this whole set. +// +// It fails closed: if an in-flight GTID is present but cannot be folded in, it +// returns an error rather than an under-approximated set, so recovery refuses +// to validate a candidate against an incomplete applied set (which could let a +// diverging master through). +func (s *BinlogStreamer) appliedGTIDSet() (mysql.GTIDSet, error) { + // Snapshot the streamed set under gtidMu; it is mutated by the XID/DDL paths + // and read by Progress() concurrently. + applied := s.lastStreamedGTIDClone() + if s.inFlightGTID != "" { + merged, err := unionGTIDStringInto(applied, s.inFlightGTID) + if err != nil { + return nil, fmt.Errorf("folding in-flight GTID %q into applied set: %w", s.inFlightGTID, err) + } + applied = merged + } + return applied, nil +} + +// recoverFromMasterFailover reconnects the source after a lost connection and +// restarts streaming from the last resumable GTID set. It is only meaningful in +// GTID mode; the caller must guard with failoverRecoveryEnabled. +// +// Each attempt asks the SourceReconnector for a validated connection to the new +// writer (the Ferry implements this via SourceRuntime.Replace, so the whole run +// repoints in one place), then rebuilds the binlog syncer against it and +// restarts StartSyncGTID from the resumable set (replaying the interrupted +// transaction). Returns an error only when recovery is exhausted. +func (s *BinlogStreamer) recoverFromMasterFailover(cause error) error { + cfg := s.MasterFailoverRecovery + + resumeSet := s.resumableGTIDClone() + appliedSet, err := s.appliedGTIDSet() + if err != nil { + // Fail closed: without a trustworthy applied set we cannot safely + // validate a candidate, so refuse to recover. + return fmt.Errorf("master failover recovery: %w (original error: %v)", err, cause) + } + + // If a cutover stop target has been recorded, the promoted master must also + // contain it; otherwise streaming would wait forever for GTIDs the new + // writer will never produce. Fold it into the set the candidate must + // contain. + if stopSet := s.stopGTIDClone(); stopSet != nil { + merged, mergeErr := unionGTIDSets(appliedSet, stopSet) + if mergeErr != nil { + return fmt.Errorf("master failover recovery: folding stop target into required set: %w (original error: %v)", mergeErr, cause) + } + appliedSet = merged + } + + previous := s.currentDBConfig() + previousHost, previousPort := "", uint16(0) + if previous != nil { + previousHost, previousPort = previous.Host, previous.Port + } + + s.logger.WithFields(Fields{ + "error": cause.Error(), + "resume_set": gtidSetString(resumeSet), + "applied_set": gtidSetString(appliedSet), + "dead_host": previousHost, + "dead_port": previousPort, + "max_attempts": cfg.MaxAttempts, + }).Warn("source connection lost; attempting master failover recovery") + + for attempt := 1; cfg.MaxAttempts == 0 || attempt <= cfg.MaxAttempts; attempt++ { + newDB, newConfig, err := s.SourceReconnector.Reconnect(previous, appliedSet) + if err != nil { + s.logger.WithError(err).WithField("attempt", attempt).Warn("failover: source reconnect failed") + time.Sleep(cfg.retryWait()) + continue + } + + newSyncer, newStreamer, err := s.buildSyncerFromGTID(newConfig, newDB, resumeSet) + if err != nil { + s.logger.WithError(err).WithField("attempt", attempt).Warn("failover: could not restart streaming on new master") + time.Sleep(cfg.retryWait()) + continue + } + + oldSyncer := s.binlogSyncer + s.connMu.Lock() + s.DB = newDB + s.DBConfig = newConfig + s.binlogSyncer = newSyncer + s.binlogStreamer = newStreamer + s.connMu.Unlock() + // Re-seed the streamed/resumable GTID sets under gtidMu so a concurrent + // Progress() read never observes a torn set during the swap. + s.seedGTIDSets(cloneOrEmpty(resumeSet)) + s.inFlightGTID = "" + + if oldSyncer != nil { + oldSyncer.Close() + } + + s.logger.WithFields(Fields{ + "new_host": newConfig.Host, + "new_port": newConfig.Port, + "resume_set": gtidSetString(resumeSet), + "attempt": attempt, + }).Info("master failover recovery succeeded; streaming resumed on new master") + return nil + } + + return fmt.Errorf("master failover recovery exhausted after %d attempts: %w", cfg.MaxAttempts, cause) +} + +// buildSyncerFromGTID constructs (without installing) a fresh binlog syncer +// against the given DB config and starts a GTID stream from resumeSet. It has +// no side effects on installed state, so callers can discard the result on a +// later failure. The server id is generated against db (the new master). +func (s *BinlogStreamer) buildSyncerFromGTID(dbConf *DatabaseConfig, db *sql.DB, resumeSet mysql.GTIDSet) (*replication.BinlogSyncer, *replication.BinlogStreamer, error) { + serverID, err := generateNewServerIdOn(db, s.logger) + if err != nil { + return nil, nil, err + } + + syncer, err := s.newBinlogSyncerFor(dbConf, serverID) + if err != nil { + return nil, nil, err + } + + streamer, err := syncer.StartSyncGTID(cloneOrEmpty(resumeSet)) + if err != nil { + syncer.Close() + return nil, nil, err + } + return syncer, streamer, nil +} + +// gtidSetString renders a possibly-nil GTID set for logging. +func gtidSetString(set mysql.GTIDSet) string { + if set == nil { + return "" + } + return set.String() +} + +// cloneOrEmpty clones set, returning a fresh empty MySQL GTID set when set is +// nil (a valid starting point for a fresh source). +func cloneOrEmpty(set mysql.GTIDSet) mysql.GTIDSet { + if set == nil { + empty, _ := mysql.ParseMysqlGTIDSet("") + return empty + } + return set.Clone() +} + // the default event handler is called for replication binLogEvents that do not have a // separate event Handler registered. @@ -372,6 +593,18 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query // set that existed BEFORE this transaction, so that an // interruption replays the whole in-flight transaction. s.setResumableToStreamed() + // Record this transaction's GTID as in-flight: its rows may be + // emitted downstream before the closing XIDEvent advances + // lastStreamedGTIDSet, so failover validation must ensure a + // candidate master contains it. GTIDNext returns the single-GTID + // set (uuid:gno) for this transaction. inFlightGTID is only + // touched by this (streaming) goroutine, so it needs no lock. + if nextSet, uerr := tev.GTIDNext(); uerr == nil { + s.inFlightGTID = nextSet.String() + } else { + s.logger.WithError(uerr).Warn("could not decode in-flight GTID; failover validation may under-approximate applied set") + s.inFlightGTID = "" + } case *replication.XIDEvent: // End of a transaction. GSet is the committed GTID set through // this transaction. setLastStreamedGTIDSet clones under gtidMu to @@ -380,6 +613,8 @@ func (s *BinlogStreamer) defaultEventHandler(ev *replication.BinlogEvent, query if tev.GSet != nil { s.setLastStreamedGTIDSet(tev.GSet) } + // The transaction committed; nothing is in flight now. + s.inFlightGTID = "" } } @@ -453,11 +688,28 @@ func (s *BinlogStreamer) Run() { ev, err = s.binlogStreamer.GetEvent(ctx) if err == context.DeadlineExceeded { timedOut = true - } else if err != nil { - s.ErrorHandler.Fatal("binlog_streamer", err) } }() + if err != nil && err != context.DeadlineExceeded { + // A non-timeout GetEvent error means the source connection was lost. + // In GTID mode with failover recovery enabled, reconnect to the new + // master and continue; otherwise this is fatal. + if s.failoverRecoveryEnabled() { + if recoverErr := s.recoverFromMasterFailover(err); recoverErr != nil { + s.ErrorHandler.Fatal("binlog_streamer", recoverErr) + return + } + // Recovered: reset per-iteration state so no stale RowsQueryEvent + // annotation or filename crosses from the old stream. + es = BinlogEventState{nextFilename: s.lastStreamedBinlogPosition.Name} + query = nil + continue + } + s.ErrorHandler.Fatal("binlog_streamer", err) + return + } + if timedOut { s.lastProcessedEventTime = time.Now() continue @@ -595,6 +847,28 @@ func (s *BinlogStreamer) resumableGTIDClone() mysql.GTIDSet { return s.lastResumableGTIDSet.Clone() } +// lastStreamedGTIDClone returns a mutable clone of the streamed set under the +// lock, or nil when unset. +func (s *BinlogStreamer) lastStreamedGTIDClone() mysql.GTIDSet { + s.gtidMu.RLock() + defer s.gtidMu.RUnlock() + if s.lastStreamedGTIDSet == nil { + return nil + } + return s.lastStreamedGTIDSet.Clone() +} + +// stopGTIDClone returns a mutable clone of the stop set under the lock, or nil +// when unset. +func (s *BinlogStreamer) stopGTIDClone() mysql.GTIDSet { + s.gtidMu.RLock() + defer s.gtidMu.RUnlock() + if s.stopAtGTIDSet == nil { + return nil + } + return s.stopAtGTIDSet.Clone() +} + // streamedGTIDCoordinate and resumableGTIDCoordinate return GTID coordinates // snapshotted under gtidMu, for stamping onto DML events. They read the sets // while holding the lock so the snapshot never races the streaming goroutine's @@ -666,7 +940,7 @@ func (s *BinlogStreamer) FlushAndStop() { // passed the stop coordinate. if s.coordinateMode() == BinlogCoordinateGTID { err := WithRetries(100, 600*time.Millisecond, s.logger, "read current executed GTID set", func() error { - gtidSet, err := ReadExecutedGTIDSet(s.DB) + gtidSet, err := ReadExecutedGTIDSet(s.currentDB()) if err != nil { return err } @@ -691,7 +965,7 @@ func (s *BinlogStreamer) FlushAndStop() { err := WithRetries(100, 600*time.Millisecond, s.logger, "read current binlog position", func() error { var err error - s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.DB) + s.stopAtBinlogPosition, err = ShowMasterStatusBinlogPosition(s.currentDB()) return err }) @@ -822,12 +1096,19 @@ func (s *BinlogStreamer) handleRowsEvent(ev *replication.BinlogEvent, query []by } func (s *BinlogStreamer) generateNewServerId() (uint32, error) { + return generateNewServerIdOn(s.DB, s.logger) +} + +// generateNewServerIdOn generates a server id not already in use on the given +// server. It takes an explicit DB so failover recovery can target a server +// other than the currently-installed one. +func generateNewServerIdOn(db *sql.DB, logger Logger) (uint32, error) { var id uint32 for { id = randomServerId() - exists, err := idExistsOnServer(id, s.DB) + exists, err := idExistsOnServer(id, db) if err != nil { return 0, err } @@ -835,7 +1116,9 @@ func (s *BinlogStreamer) generateNewServerId() (uint32, error) { break } - s.logger.WithField("server_id", id).Warn("server_id was taken, retrying") + if logger != nil { + logger.WithField("server_id", id).Warn("server_id was taken, retrying") + } } return id, nil diff --git a/binlog_streamer_gtid_test.go b/binlog_streamer_gtid_test.go index bffd4a09..3c8ce27e 100644 --- a/binlog_streamer_gtid_test.go +++ b/binlog_streamer_gtid_test.go @@ -6,6 +6,7 @@ import ( "github.com/go-mysql-org/go-mysql/mysql" "github.com/go-mysql-org/go-mysql/replication" + "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -64,6 +65,35 @@ func TestQueryEventAdvancesStreamedGTID(t *testing.T) { assert.Equal(t, before, s.lastStreamedGTIDSet.String(), "BEGIN must not advance the streamed GTID set") } +// TestInFlightGTIDTracking verifies a GTIDEvent records the in-flight +// transaction's GTID and the closing XIDEvent clears it. Failover validation +// relies on this to reject a candidate missing an emitted-but-uncommitted txn. +func TestInFlightGTIDTracking(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + sid, err := uuid.Parse("3e11fa47-71ca-11e1-9e33-c80aa9429562") + require.NoError(t, err) + es := &BinlogEventState{} + + gtidEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 100}, + Event: &replication.GTIDEvent{SID: sid[:], GNO: 101}, + } + _, err = s.defaultEventHandler(gtidEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101", s.inFlightGTID) + + xidEvent := &replication.BinlogEvent{ + Header: &replication.EventHeader{LogPos: 200}, + Event: &replication.XIDEvent{GSet: mustParseGTID(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101")}, + } + _, err = s.defaultEventHandler(xidEvent, nil, es) + require.NoError(t, err) + assert.Equal(t, "", s.inFlightGTID, "XIDEvent must clear the in-flight GTID") + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", s.lastStreamedGTIDSet.String()) +} + func TestCoordinateModeDefaultsToFilePosition(t *testing.T) { s := &BinlogStreamer{} assert.Equal(t, BinlogCoordinateFilePosition, s.coordinateMode()) diff --git a/config.go b/config.go index 4cf9dce0..c6efb42f 100644 --- a/config.go +++ b/config.go @@ -707,6 +707,15 @@ type Config struct { // additional coordinate sources. BinlogCoordinateMode BinlogCoordinateType + // MasterFailoverRecovery, when set, enables automatic reconnection of the + // source to a new master writer when the source connection is lost (a + // topology failover). It requires BinlogCoordinateMode "gtid" (GTID + // coordinates are server-independent; file/position cannot be carried across + // a failover). The Resolver is supplied by the embedding application to + // discover the new writer. Nil keeps the previous behavior: a lost source + // connection is fatal. + MasterFailoverRecovery *MasterFailoverRecoveryConfig + // This config is necessary for inline verification for a special case of // Ghostferry: // @@ -885,6 +894,15 @@ func (c *Config) ValidateConfig() error { return fmt.Errorf("invalid BinlogCoordinateMode %q, must be %q or %q", c.BinlogCoordinateMode, BinlogCoordinateFilePosition, BinlogCoordinateGTID) } + if c.MasterFailoverRecovery != nil { + if c.BinlogCoordinateMode != BinlogCoordinateGTID { + return fmt.Errorf("MasterFailoverRecovery requires BinlogCoordinateMode %q, got %q", BinlogCoordinateGTID, c.BinlogCoordinateMode) + } + if c.MasterFailoverRecovery.Resolver == nil { + return fmt.Errorf("MasterFailoverRecovery requires a Resolver") + } + } + if c.VerifierType == VerifierTypeIterative { if err := c.IterativeVerifierConfig.Validate(); err != nil { return fmt.Errorf("IterativeVerifierConfig invalid: %v", err) diff --git a/ferry.go b/ferry.go index 6c8c3a74..4f963e24 100644 --- a/ferry.go +++ b/ferry.go @@ -141,7 +141,14 @@ func (f *Ferry) NewDataIteratorWithoutStateTracker() *DataIterator { } func (f *Ferry) NewSourceBinlogStreamer() *BinlogStreamer { - return f.newBinlogStreamer(f.SourceDB, f.Config.Source, nil, nil, "source_binlog_streamer") + streamer := f.newBinlogStreamer(f.SourceDB, f.Config.Source, nil, nil, "source_binlog_streamer") + // Master-failover recovery only applies to the source stream: the target + // verifier streams from the (stable) target and does not fail over. The + // reconnector repoints the whole run via SourceRuntime when a failover + // occurs. + streamer.MasterFailoverRecovery = f.Config.MasterFailoverRecovery + streamer.SourceReconnector = f.newSourceReconnector() + return streamer } func (f *Ferry) NewTargetBinlogStreamer() (*BinlogStreamer, error) { diff --git a/ferry_failover.go b/ferry_failover.go new file mode 100644 index 00000000..33fcdd50 --- /dev/null +++ b/ferry_failover.go @@ -0,0 +1,69 @@ +package ghostferry + +import ( + sql "github.com/Shopify/ghostferry/sqlwrapper" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +// ferrySourceReconnector implements SourceReconnector for the Ferry. It is the +// single place a source master failover repoints the whole run: it resolves the +// promoted writer, opens one Ferry-owned connection to it via SourceRuntime +// (which atomically becomes the source for every consumer that starts new work +// afterwards — data iterator, verifiers, ...), validates it is a safe target, +// and hands the connection back to the binlog streamer. +// +// Because every source consumer already reads its connection from the +// SourceRuntime, there is nothing to hand-repoint here: the Replace is the swap. +type ferrySourceReconnector struct { + ferry *Ferry + resolver MasterWriterResolver +} + +// Reconnect satisfies SourceReconnector. previous is the config the streamer +// last used; appliedGTIDSet is everything already emitted downstream, which the +// promoted writer must contain. +func (r *ferrySourceReconnector) Reconnect(previous *DatabaseConfig, appliedGTIDSet mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) { + candidate, err := r.resolver.ResolveCurrentMaster(previous) + if err != nil { + return nil, nil, err + } + if candidate == nil { + return nil, nil, errNilCandidate + } + + // Install the candidate as the current source ONLY if it validates. Fail + // closed: the promoted writer must be a writable server with GTID mode on + // and must contain everything already applied downstream. A candidate that + // fails validation is never published to consumers. On success the old + // handle is retired (retained, not closed), since in-flight cursors / cached + // statements may still reference it. + newDB, err := r.ferry.sourceRuntime.ReplaceValidated( + candidate, + r.ferry.logger.WithField("dbname", "source_failover"), + func(db *sql.DB) error { + return validateFailoverTarget(db, candidate, appliedGTIDSet) + }, + ) + if err != nil { + return nil, nil, err + } + + r.ferry.logger.WithFields(Fields{ + "new_host": candidate.Host, + "new_port": candidate.Port, + }).Info("source repointed at promoted master after failover") + return newDB, candidate, nil +} + +// newSourceReconnector builds the reconnector wired to this Ferry, or nil when +// failover recovery is not configured. +func (f *Ferry) newSourceReconnector() SourceReconnector { + if f.Config.MasterFailoverRecovery == nil || f.Config.MasterFailoverRecovery.Resolver == nil { + return nil + } + return &ferrySourceReconnector{ + ferry: f, + resolver: f.Config.MasterFailoverRecovery.Resolver, + } +} diff --git a/ferry_failover_test.go b/ferry_failover_test.go new file mode 100644 index 00000000..272c69bb --- /dev/null +++ b/ferry_failover_test.go @@ -0,0 +1,70 @@ +package ghostferry + +import ( + "errors" + "testing" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func reconnectorTestConfig(host string) *DatabaseConfig { + return &DatabaseConfig{Host: host, Port: 3306, User: "root", Net: "tcp"} +} + +func TestNewSourceReconnectorNilWhenDisabled(t *testing.T) { + f := &Ferry{Config: &Config{}} + assert.Nil(t, f.newSourceReconnector()) + + f.Config.MasterFailoverRecovery = &MasterFailoverRecoveryConfig{} + assert.Nil(t, f.newSourceReconnector(), "nil resolver disables reconnector") +} + +func TestReconnectPropagatesResolverError(t *testing.T) { + wantErr := errors.New("no master found") + f := &Ferry{ + Config: &Config{Source: reconnectorTestConfig("old")}, + logger: LogWithField("tag", "test"), + sourceRuntime: NewSourceRuntime(&sql.DB{Marginalia: "old"}, reconnectorTestConfig("old")), + } + r := &ferrySourceReconnector{ + ferry: f, + resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, wantErr }), + } + + _, _, err := r.Reconnect(reconnectorTestConfig("old"), nil) + assert.ErrorIs(t, err, wantErr) +} + +func TestReconnectRejectsNilCandidate(t *testing.T) { + f := &Ferry{ + Config: &Config{Source: reconnectorTestConfig("old")}, + logger: LogWithField("tag", "test"), + sourceRuntime: NewSourceRuntime(&sql.DB{Marginalia: "old"}, reconnectorTestConfig("old")), + } + r := &ferrySourceReconnector{ + ferry: f, + resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, nil }), + } + + _, _, err := r.Reconnect(reconnectorTestConfig("old"), nil) + assert.ErrorIs(t, err, errNilCandidate) + + // The runtime must not have been swapped when the candidate is nil. + assert.Equal(t, "old", f.sourceRuntime.Config().Host) +} + +func TestNewSourceReconnectorWiredWhenConfigured(t *testing.T) { + f := &Ferry{ + Config: &Config{ + MasterFailoverRecovery: &MasterFailoverRecoveryConfig{ + Resolver: MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { return nil, nil }), + }, + }, + } + r := f.newSourceReconnector() + require.NotNil(t, r) + _, ok := r.(*ferrySourceReconnector) + assert.True(t, ok) +} diff --git a/go.mod b/go.mod index 0dbdbfd0..bd8d0b00 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( github.com/go-mysql-org/go-mysql v1.13.0 github.com/go-sql-driver/mysql v1.7.1 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db + github.com/google/uuid v1.3.0 github.com/gorilla/mux v1.6.1 github.com/rs/zerolog v1.35.0 github.com/shopspring/decimal v1.2.0 @@ -20,7 +21,6 @@ require ( github.com/Microsoft/go-winio v0.5.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/goccy/go-json v0.10.2 // indirect - github.com/google/uuid v1.3.0 // indirect github.com/gorilla/context v1.1.1 // indirect github.com/klauspost/compress v1.17.8 // indirect github.com/lann/builder v0.0.0-20180216234317-1b87b36280d0 // indirect diff --git a/master_failover.go b/master_failover.go new file mode 100644 index 00000000..74e7a89c --- /dev/null +++ b/master_failover.go @@ -0,0 +1,154 @@ +package ghostferry + +import ( + "errors" + "fmt" + "time" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + + "github.com/go-mysql-org/go-mysql/mysql" +) + +// errNilCandidate is returned when a resolver yields no candidate master. +var errNilCandidate = errors.New("failover: resolver returned a nil candidate master") + +// MasterWriterResolver resolves the current source master writer after a +// suspected failover. Implementations typically consult an external topology +// service (orchestrator, a service-discovery endpoint, etc.) to find the host +// that is now the writable primary. +// +// ResolveCurrentMaster receives the DatabaseConfig the source was last +// connected to (so the implementation can, for example, exclude the now-dead +// host) and returns the connection details of the host that should now be +// streamed from. Returning an error causes recovery to be retried after a +// backoff. +type MasterWriterResolver interface { + ResolveCurrentMaster(previous *DatabaseConfig) (*DatabaseConfig, error) +} + +// MasterWriterResolverFunc adapts a plain function to MasterWriterResolver. +type MasterWriterResolverFunc func(previous *DatabaseConfig) (*DatabaseConfig, error) + +func (f MasterWriterResolverFunc) ResolveCurrentMaster(previous *DatabaseConfig) (*DatabaseConfig, error) { + return f(previous) +} + +// MasterFailoverRecoveryConfig configures automatic reconnection to a new +// source master when the source connection is lost. +// +// Failover recovery is only supported in GTID binlog coordinate mode: GTID sets +// are server-independent, so a resume set valid on the old master is meaningful +// on the new one. File/position coordinates are per-host and cannot be carried +// across a failover, so recovery is refused in that mode. +type MasterFailoverRecoveryConfig struct { + // Resolver discovers the new master writer. Required; recovery is disabled + // when nil. + Resolver MasterWriterResolver + + // MaxAttempts bounds how many times recovery is attempted for a single + // disconnect before giving up and surfacing a fatal error. Zero means retry + // indefinitely. + MaxAttempts int + + // RetryWait is how long to wait between recovery attempts. Defaults to + // DefaultFailoverRetryWait when zero. + RetryWait time.Duration + + // SyncerMaxReconnectAttempts bounds how many times go-mysql's binlog syncer + // retries reconnecting to the SAME (now-dead) host before giving up and + // surfacing the error, which is what triggers our reconnect-to-a-new-host + // recovery. Without a bound the syncer retries the dead host forever and the + // failure never surfaces. Defaults to DefaultSyncerMaxReconnectAttempts. + SyncerMaxReconnectAttempts int +} + +// DefaultSyncerMaxReconnectAttempts bounds go-mysql's same-host reconnect +// retries when failover recovery is enabled, so a lost source surfaces as a +// GetEvent error (triggering failover) within a few seconds. +const DefaultSyncerMaxReconnectAttempts = 3 + +func (c *MasterFailoverRecoveryConfig) syncerMaxReconnectAttempts() int { + if c.SyncerMaxReconnectAttempts <= 0 { + return DefaultSyncerMaxReconnectAttempts + } + return c.SyncerMaxReconnectAttempts +} + +// DefaultFailoverRetryWait is the wait between failover recovery attempts when +// MasterFailoverRecoveryConfig.RetryWait is unset. +const DefaultFailoverRetryWait = 500 * time.Millisecond + +func (c *MasterFailoverRecoveryConfig) retryWait() time.Duration { + if c.RetryWait <= 0 { + return DefaultFailoverRetryWait + } + return c.RetryWait +} + +// SourceReconnector reconnects the source after the binlog streamer loses its +// connection. It is the single seam between the streamer (which only knows "I +// lost the source and here is the safe GTID set I must not fall behind") and +// the Ferry (which owns the SourceRuntime and therefore how the whole run +// repoints at a promoted writer). +// +// Reconnect is called on the streamer's Run goroutine when a non-timeout +// GetEvent error occurs. appliedGTIDSet is everything the streamer has already +// emitted downstream (committed set plus any in-flight transaction); the +// implementation must guarantee the returned source contains it, or fail +// closed, so recovery never resumes against a master that lost applied +// transactions (which would silently diverge the target). +// +// On success it returns the new source connection and the config it was opened +// from; the streamer rebuilds its binlog syncer against them. Returning an +// error means this attempt failed and the streamer will retry (subject to +// MaxAttempts). +type SourceReconnector interface { + Reconnect(previous *DatabaseConfig, appliedGTIDSet mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) +} + +// validateFailoverTarget verifies that db (already opened against candidate) is +// a safe source to resume streaming from: +// +// - it is a writer (@@read_only = OFF), not a demoted master or replica; +// - GTID mode is enabled (@@GLOBAL.gtid_mode = ON); and +// - its executed GTID set contains appliedSet, i.e. everything already +// emitted downstream. A candidate missing any applied transaction is +// rejected (fail closed) so recovery cannot silently diverge the target. +// +// It does not close db; the caller owns its lifecycle. +func validateFailoverTarget(db *sql.DB, candidate *DatabaseConfig, appliedSet mysql.GTIDSet) error { + if candidate == nil { + return fmt.Errorf("failover: nil candidate master config") + } + + isReadOnly, err := CheckDbIsAReplica(db) + if err != nil { + return fmt.Errorf("failover: checking candidate master %s:%d read_only: %w", candidate.Host, candidate.Port, err) + } + if isReadOnly { + return fmt.Errorf("failover: candidate master %s:%d is read_only (not a writer); rejecting", candidate.Host, candidate.Port) + } + + if err := CheckServerGTIDModeEnabled(db); err != nil { + return fmt.Errorf("failover: candidate master %s:%d rejected: %w", candidate.Host, candidate.Port, err) + } + + candidateSetStr, err := ReadExecutedGTIDSet(db) + if err != nil { + return fmt.Errorf("failover: reading candidate master executed GTID set: %w", err) + } + candidateSet, err := mysql.ParseMysqlGTIDSet(candidateSetStr) + if err != nil { + return fmt.Errorf("failover: parsing candidate master executed GTID set %q: %w", candidateSetStr, err) + } + + if appliedSet != nil && !candidateSet.Contain(appliedSet) { + return fmt.Errorf( + "failover: candidate master %s:%d executed set %q does not contain already-applied set %q; resuming there would diverge from the source", + candidate.Host, candidate.Port, candidateSet.String(), appliedSet.String(), + ) + } + + return nil +} diff --git a/master_failover_test.go b/master_failover_test.go new file mode 100644 index 00000000..f42a2832 --- /dev/null +++ b/master_failover_test.go @@ -0,0 +1,145 @@ +package ghostferry + +import ( + "errors" + "testing" + "time" + + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/go-mysql-org/go-mysql/mysql" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFailoverRecoveryEnabled(t *testing.T) { + reconnector := &fakeReconnector{} + resolver := MasterWriterResolverFunc(func(_ *DatabaseConfig) (*DatabaseConfig, error) { + return nil, nil + }) + _ = resolver + + // Disabled with no config. + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + assert.False(t, s.failoverRecoveryEnabled()) + + // Config but no reconnector. + s.MasterFailoverRecovery = &MasterFailoverRecoveryConfig{} + assert.False(t, s.failoverRecoveryEnabled()) + + // Config + reconnector but file/position mode. + s.BinlogCoordinateMode = BinlogCoordinateFilePosition + s.SourceReconnector = reconnector + assert.False(t, s.failoverRecoveryEnabled()) + + // GTID mode + config + reconnector: enabled. + s.BinlogCoordinateMode = BinlogCoordinateGTID + assert.True(t, s.failoverRecoveryEnabled()) +} + +func TestFailoverRetryWaitDefault(t *testing.T) { + c := &MasterFailoverRecoveryConfig{} + assert.Equal(t, DefaultFailoverRetryWait, c.retryWait()) + c.RetryWait = 2 * time.Second + assert.Equal(t, 2*time.Second, c.retryWait()) +} + +func TestMasterWriterResolverFunc(t *testing.T) { + want := &DatabaseConfig{Host: "new", Port: 3306} + prev := &DatabaseConfig{Host: "old", Port: 3306} + var got *DatabaseConfig + resolver := MasterWriterResolverFunc(func(p *DatabaseConfig) (*DatabaseConfig, error) { + got = p + return want, nil + }) + result, err := resolver.ResolveCurrentMaster(prev) + require.NoError(t, err) + assert.Same(t, want, result) + assert.Same(t, prev, got) +} + +func TestAppliedGTIDSetFoldsInFlight(t *testing.T) { + s := &BinlogStreamer{BinlogCoordinateMode: BinlogCoordinateGTID} + s.logger = LogWithField("tag", "test") + + // Committed set only. + s.lastStreamedGTIDSet = mustParseGTID(t, gtidSetTarget) // :1-100 + applied, err := s.appliedGTIDSet() + require.NoError(t, err) + assert.Equal(t, gtidSetTarget, applied.String()) + + // With an in-flight GTID, the applied set extends to include it. + s.inFlightGTID = "3e11fa47-71ca-11e1-9e33-c80aa9429562:101" + applied, err = s.appliedGTIDSet() + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", applied.String()) + + // A malformed in-flight GTID fails closed. + s.inFlightGTID = "not-a-gtid" + _, err = s.appliedGTIDSet() + assert.Error(t, err) +} + +func TestGtidSetStringNil(t *testing.T) { + assert.Equal(t, "", gtidSetString(nil)) + assert.Equal(t, gtidSetTarget, gtidSetString(mustParseGTID(t, gtidSetTarget))) +} + +func TestCloneOrEmpty(t *testing.T) { + empty := cloneOrEmpty(nil) + require.NotNil(t, empty) + assert.Equal(t, "", empty.String()) + + set := mustParseGTID(t, gtidSetTarget) + clone := cloneOrEmpty(set) + assert.Equal(t, gtidSetTarget, clone.String()) + require.NoError(t, clone.(*mysql.MysqlGTIDSet).Update("3e11fa47-71ca-11e1-9e33-c80aa9429562:101")) + assert.Equal(t, gtidSetTarget, set.String(), "clone must not alias source") +} + +func TestUnionGTIDStringInto(t *testing.T) { + res, err := unionGTIDStringInto(nil, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101") + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101", res.String()) + + base := mustParseGTID(t, gtidSetTarget) // :1-100 + merged, err := unionGTIDStringInto(base, "3e11fa47-71ca-11e1-9e33-c80aa9429562:101") + require.NoError(t, err) + assert.Equal(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101", merged.String()) + assert.Equal(t, gtidSetTarget, base.String(), "base must not be mutated") + + // The safety property: a candidate missing the in-flight GTID fails containment. + candidateMissing := mustParseGTID(t, gtidSetTarget) + assert.False(t, candidateMissing.Contain(merged)) + candidateHas := mustParseGTID(t, "3e11fa47-71ca-11e1-9e33-c80aa9429562:1-101") + assert.True(t, candidateHas.Contain(merged)) +} + +func TestUnionGTIDSets(t *testing.T) { + // nil inputs yield empty. + res, err := unionGTIDSets(nil, nil) + require.NoError(t, err) + assert.Equal(t, "", res.String()) + + applied := mustParseGTID(t, gtidSetTarget) // :1-100 + stop := mustParseGTID(t, gtidSetPast) // :1-150 + merged, err := unionGTIDSets(applied, stop) + require.NoError(t, err) + assert.Equal(t, gtidSetPast, merged.String()) + // Inputs unchanged. + assert.Equal(t, gtidSetTarget, applied.String()) + assert.Equal(t, gtidSetPast, stop.String()) + + // A candidate must contain the union of applied + stop target. + candidateShort := mustParseGTID(t, gtidSetTarget) // only :1-100 + assert.False(t, candidateShort.Contain(merged), "candidate missing stop-target GTIDs must fail") +} + +// fakeReconnector is a test double for SourceReconnector. +type fakeReconnector struct { + calls int +} + +func (f *fakeReconnector) Reconnect(_ *DatabaseConfig, _ mysql.GTIDSet) (*sql.DB, *DatabaseConfig, error) { + f.calls++ + return nil, nil, errors.New("not implemented") +} diff --git a/source_runtime.go b/source_runtime.go index ceeff36c..1a564482 100644 --- a/source_runtime.go +++ b/source_runtime.go @@ -71,11 +71,32 @@ func (r *SourceRuntime) Config() *DatabaseConfig { // If opening the new connection fails, the current source is left unchanged and // the error is returned. func (r *SourceRuntime) Replace(config *DatabaseConfig, logger Logger) (*sql.DB, error) { + return r.ReplaceValidated(config, logger, nil) +} + +// ReplaceValidated opens a new source connection from the given config, runs an +// optional validate hook against it, and only installs it as the current source +// if validation succeeds. This keeps the swap fail-closed: a candidate that +// fails validation is never published to consumers. +// +// If opening the connection or the validate hook fails, the newly opened handle +// is closed and the current source is left unchanged. On success the previous +// handle is retired (retained, not closed) as with Replace. +func (r *SourceRuntime) ReplaceValidated(config *DatabaseConfig, logger Logger, validate func(*sql.DB) error) (*sql.DB, error) { newDB, err := config.SqlDB(logger) if err != nil { return nil, err } + if validate != nil { + if err := validate(newDB); err != nil { + // The candidate is unsafe; do not publish it. Close the handle we + // opened for validation. + _ = newDB.Close() + return nil, err + } + } + r.mu.Lock() defer r.mu.Unlock() diff --git a/source_runtime_test.go b/source_runtime_test.go index cf421ca4..5cb14880 100644 --- a/source_runtime_test.go +++ b/source_runtime_test.go @@ -1,6 +1,7 @@ package ghostferry import ( + "errors" "testing" sql "github.com/Shopify/ghostferry/sqlwrapper" @@ -68,6 +69,41 @@ func TestSourceRuntimeCloseRetiredDrains(t *testing.T) { assert.Empty(t, rt.retired, "retired handles must be drained") } +func TestSourceRuntimeReplaceValidatedFailureDoesNotPublish(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + oldCfg := runtimeTestConfig("old-host") + rt := NewSourceRuntime(oldDB, oldCfg) + + validateErr := errors.New("candidate rejected") + _, err := rt.ReplaceValidated(runtimeTestConfig("bad-host"), nil, func(_ *sql.DB) error { + return validateErr + }) + assert.ErrorIs(t, err, validateErr) + + // The current source must be unchanged and nothing retired. + assert.Same(t, oldDB, rt.DB()) + assert.Same(t, oldCfg, rt.Config()) + assert.Empty(t, rt.retired) +} + +func TestSourceRuntimeReplaceValidatedSuccessPublishes(t *testing.T) { + oldDB := &sql.DB{Marginalia: "old"} + rt := NewSourceRuntime(oldDB, runtimeTestConfig("old-host")) + + validated := false + newCfg := runtimeTestConfig("new-host") + newDB, err := rt.ReplaceValidated(newCfg, nil, func(db *sql.DB) error { + validated = true + assert.NotNil(t, db) + return nil + }) + require.NoError(t, err) + assert.True(t, validated) + assert.Same(t, newDB, rt.DB()) + assert.Same(t, newCfg, rt.Config()) + assert.Contains(t, rt.retired, oldDB) +} + func TestSourceRuntimeNilInitial(t *testing.T) { rt := NewSourceRuntime(nil, nil) assert.Nil(t, rt.DB()) diff --git a/test/go/config_test.go b/test/go/config_test.go index 50b8d0f1..c015ba2f 100644 --- a/test/go/config_test.go +++ b/test/go/config_test.go @@ -220,6 +220,35 @@ func (this *ConfigTestSuite) TestBinlogCoordinateModeInvalidIsRejected() { this.Require().EqualError(err, `invalid BinlogCoordinateMode "banana", must be "file_position" or "gtid"`) } +func (this *ConfigTestSuite) TestMasterFailoverRecoveryRequiresGTIDMode() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateFilePosition + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return nil, nil + }), + } + err := this.config.ValidateConfig() + this.Require().EqualError(err, `MasterFailoverRecovery requires BinlogCoordinateMode "gtid", got "file_position"`) +} + +func (this *ConfigTestSuite) TestMasterFailoverRecoveryRequiresResolver() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{} + err := this.config.ValidateConfig() + this.Require().EqualError(err, "MasterFailoverRecovery requires a Resolver") +} + +func (this *ConfigTestSuite) TestMasterFailoverRecoveryValidInGTIDMode() { + this.config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + this.config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return nil, nil + }), + } + err := this.config.ValidateConfig() + this.Require().Nil(err) +} + func TestConfig(t *testing.T) { testhelpers.SetupTest() suite.Run(t, new(ConfigTestSuite)) diff --git a/test/go/master_failover_integration_test.go b/test/go/master_failover_integration_test.go new file mode 100644 index 00000000..b7175794 --- /dev/null +++ b/test/go/master_failover_integration_test.go @@ -0,0 +1,424 @@ +package test + +import ( + "fmt" + "os" + "os/exec" + "strings" + "sync" + "testing" + "time" + + sqlorig "database/sql" + + "github.com/Shopify/ghostferry" + sqlwrapper "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// The promoted-writer ("future master") MySQL. It exists in the docker-compose +// files as mysql-3 and is started by .github/workflows/start-mysql.sh. There is +// no N3_PORT helper, so hardcode it (matching the Ruby harness convention). +const futureMasterPort = 29293 + +// TestMasterFailoverRecoveryReconnectsToPromotedMaster drives a full source +// master failover and asserts Ghostferry reconnects to the promoted writer and +// finishes the move against it. +// +// It is GTID-only (failover recovery requires server-independent coordinates) +// and needs a container tool (docker/podman) to stop the source, so it skips +// when either precondition is absent. +func TestMasterFailoverRecoveryReconnectsToPromotedMaster(t *testing.T) { + if binlogModeFromEnv() != ghostferry.BinlogCoordinateGTID { + t.Skip("master failover recovery is only supported in gtid mode") + } + tool := containerTool(t) + if tool == "" { + t.Skip("master failover test requires docker or podman to stop the source container") + } + + sourcePort := testhelpers.TestSourcePort + targetPort := testhelpers.TestTargetPort + + sourceDB := openDB(t, sourcePort) + defer sourceDB.Close() + targetDB := openDB(t, targetPort) + defer targetDB.Close() + promotedDB := openDB(t, uint64(futureMasterPort)) + defer promotedDB.Close() + + // Clean slate on all three. + dropTestDBs(t, sourceDB) + dropTestDBs(t, targetDB) + dropTestDBs(t, promotedDB) + + fm := &futureMaster{t: t, db: promotedDB, sourceDB: sourceDB, containerTool: tool} + // Ensure the promoted server is restored even if the test fails. + defer fm.restoreSourceContainer() + defer fm.reset() + + // mysql-3 replicates the source BEFORE seeding, so it holds the real seed + // data and later writes; a failover to it then cannot lose applied + // transactions. + fm.setupAsReplicaOfSource() + + testhelpers.SeedInitialData(sourceDB, "gftest", "table1", 30) + testhelpers.SeedInitialData(targetDB, "gftest", "table1", 0) + + promoted := &ghostferry.DatabaseConfig{ + Host: "127.0.0.1", + Port: uint16(futureMasterPort), + Net: "tcp", + User: "root", + Pass: "", + Collation: "utf8mb4_unicode_ci", + Params: map[string]string{"charset": "utf8mb4"}, + } + + ferry := testhelpers.NewTestFerry() + ferry.Config.BinlogCoordinateMode = ghostferry.BinlogCoordinateGTID + ferry.Config.MasterFailoverRecovery = &ghostferry.MasterFailoverRecoveryConfig{ + Resolver: ghostferry.MasterWriterResolverFunc(func(_ *ghostferry.DatabaseConfig) (*ghostferry.DatabaseConfig, error) { + return promoted, nil + }), + MaxAttempts: 20, + RetryWait: 250 * time.Millisecond, + } + require.NoError(t, ferry.Config.ValidateConfig()) + + errHandler := &testhelpers.ErrorHandler{} + ferry.ErrorHandler = errHandler + + require.NoError(t, ferry.Initialize()) + require.NoError(t, ferry.Start()) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + ferry.Run() + }() + + // Wait until row copy is fully complete before touching the source, so the + // data iterator is no longer reading it when we take it down. (Failover + // during row copy is out of scope: in-flight cursors are bound to their + // connection for scan consistency.) + ferry.WaitUntilRowCopyIsComplete() + + // Now fail the source over while the binlog streamer is still live and has + // no stop target yet: + // 1. write extra rows to the source and wait for mysql-3 to replicate them, + // so they exist on the promoted master but have NOT been streamed yet; + // 2. promote mysql-3 and stop the old source container. + // The streamer then finds the old source gone and must recover onto mysql-3 + // to stream those rows before cutover completes. + insertRows(t, sourceDB, 50) + fm.waitUntilCaughtUpToSource() + fm.promote() + fm.stopSourceContainer() + + // Wait for the streamer to actually recover onto the promoted master before + // driving cutover, so cutover records its stop coordinate from the new + // writer. (In Go we control the flow directly, so we can synchronize here + // rather than racing the cutover path against recovery.) + waitUntilSourceRepointed(t, ferry, uint16(futureMasterPort), errHandler) + + ferry.FlushBinlogAndStopStreaming() + wg.Wait() + ferry.StopTargetVerifier() + + require.NoError(t, errHandler.LastError, "ghostferry should not have fataled") + + // The promoted master (mysql-3) is now the source of truth; the target must + // be identical to it. + assertTablesIdentical(t, promotedDB, targetDB, "gftest", "table1") + + // Recovery must actually have run and repointed the source. + assert.NotEqual(t, uint16(sourcePort), ferry.SourceRuntime().Config().Port, + "the ferry source should have been repointed away from the dead master") + assert.Equal(t, uint16(futureMasterPort), ferry.SourceRuntime().Config().Port, + "the ferry source should now be the promoted master") +} + +// waitUntilSourceRepointed blocks until the ferry's source runtime has been +// swapped to the expected (promoted) port, or the ferry fataled, or a timeout. +func waitUntilSourceRepointed(t *testing.T, ferry *testhelpers.TestFerry, wantPort uint16, errHandler *testhelpers.ErrorHandler) { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for time.Now().Before(deadline) { + if errHandler.LastError != nil { + t.Fatalf("ghostferry fataled during failover recovery: %v", errHandler.LastError) + } + if cfg := ferry.SourceRuntime().Config(); cfg != nil && cfg.Port == wantPort { + return + } + time.Sleep(200 * time.Millisecond) + } + t.Fatalf("source was not repointed to port %d within timeout", wantPort) +} + +// ---- helpers ---- + +func binlogModeFromEnv() ghostferry.BinlogCoordinateType { + mode := os.Getenv("GHOSTFERRY_BINLOG_COORDINATE_MODE") + if mode == "" { + return ghostferry.BinlogCoordinateFilePosition + } + return ghostferry.BinlogCoordinateType(mode) +} + +func openDB(t *testing.T, port uint64) *sqlwrapper.DB { + t.Helper() + dsn := fmt.Sprintf("root@tcp(127.0.0.1:%d)/?charset=utf8mb4&collation=utf8mb4_unicode_ci&interpolateParams=true", port) + db, err := sqlwrapper.Open("mysql", dsn, "") + require.NoError(t, err) + require.NoError(t, db.Ping()) + return db +} + +func dropTestDBs(t *testing.T, db *sqlwrapper.DB) { + t.Helper() + for _, name := range testhelpers.ApplicableTestDbs { + _, err := db.Exec(fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", name)) + require.NoError(t, err) + } +} + +func insertRows(t *testing.T, db *sqlwrapper.DB, count int) { + t.Helper() + for i := 0; i < count; i++ { + _, err := db.Exec("INSERT INTO gftest.table1 (id, data) VALUES (?, ?)", nil, testhelpers.RandData()) + require.NoError(t, err) + } +} + +func assertTablesIdentical(t *testing.T, a, b *sqlwrapper.DB, dbName, table string) { + t.Helper() + countA := tableRowCount(t, a, dbName, table) + countB := tableRowCount(t, b, dbName, table) + assert.Greater(t, countA, 0, "promoted master should have rows") + assert.Equal(t, countA, countB, "promoted master and target row counts differ") + + checksumA := tableChecksum(t, a, dbName, table) + checksumB := tableChecksum(t, b, dbName, table) + assert.Equal(t, checksumA, checksumB, "promoted master and target checksums differ") +} + +func tableRowCount(t *testing.T, db *sqlwrapper.DB, dbName, table string) int { + t.Helper() + var n int + require.NoError(t, db.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM `%s`.`%s`", dbName, table)).Scan(&n)) + return n +} + +func tableChecksum(t *testing.T, db *sqlwrapper.DB, dbName, table string) int64 { + t.Helper() + var name string + var checksum sqlorig.NullInt64 + require.NoError(t, db.QueryRow(fmt.Sprintf("CHECKSUM TABLE `%s`.`%s`", dbName, table)).Scan(&name, &checksum)) + return checksum.Int64 +} + +// containerTool returns "docker" or "podman" if one is on PATH and can see the +// source container, else "". +func containerTool(t *testing.T) string { + for _, tool := range []string{"docker", "podman"} { + if _, err := exec.LookPath(tool); err != nil { + continue + } + if sourceContainerName(tool) != "" { + return tool + } + } + return "" +} + +func sourceContainerName(tool string) string { + out, err := exec.Command(tool, "ps", "-a", "--format", "{{.Names}}").Output() + if err != nil { + return "" + } + names := strings.Split(string(out), "\n") + for _, candidate := range []string{"ghostferry_mysql-1_1", "ghostferry-mysql-1-1"} { + for _, n := range names { + if strings.TrimSpace(n) == candidate { + return candidate + } + } + } + return "" +} + +// futureMaster manages mysql-3 as a promotable replica of the source. +type futureMaster struct { + t *testing.T + db *sqlwrapper.DB + sourceDB *sqlwrapper.DB + containerTool string +} + +func (f *futureMaster) exec(db *sqlwrapper.DB, query string) { + f.t.Helper() + _, err := db.Exec(query) + require.NoError(f.t, err, "query: %s", query) +} + +func (f *futureMaster) execIgnore(db *sqlwrapper.DB, query string) { + _, _ = db.Exec(query) +} + +func (f *futureMaster) sourceExecutedGTIDSet() string { + f.t.Helper() + var gtid string + require.NoError(f.t, f.sourceDB.QueryRow("SELECT @@GLOBAL.gtid_executed").Scan(>id)) + return strings.Join(strings.Fields(gtid), "") +} + +func (f *futureMaster) resetBinaryLogsStatement() string { + var version string + require.NoError(f.t, f.db.QueryRow("SELECT VERSION()").Scan(&version)) + if versionAtLeast(version, 8, 4) { + return "RESET BINARY LOGS AND GTIDS" + } + return "RESET MASTER" +} + +func (f *futureMaster) setupAsReplicaOfSource() { + f.t.Helper() + f.execIgnore(f.db, "STOP REPLICA") + f.execIgnore(f.db, "RESET REPLICA ALL") + f.execIgnore(f.db, "SET GLOBAL read_only = OFF") + dropTestDBs(f.t, f.db) + + sourceGTID := f.sourceExecutedGTIDSet() + f.exec(f.db, f.resetBinaryLogsStatement()) + if sourceGTID != "" { + f.exec(f.db, fmt.Sprintf("SET GLOBAL gtid_purged = '%s'", sourceGTID)) + } + f.exec(f.db, "CHANGE REPLICATION SOURCE TO SOURCE_HOST='mysql-1', SOURCE_PORT=3306, SOURCE_USER='root', SOURCE_AUTO_POSITION=1") + f.exec(f.db, "START REPLICA") + f.waitUntilReplicaRunning() +} + +func (f *futureMaster) waitUntilReplicaRunning() { + f.t.Helper() + deadline := time.Now().Add(20 * time.Second) + for { + io, sql := f.replicaRunning() + if io && sql { + return + } + if time.Now().After(deadline) { + f.t.Fatalf("future master replica did not start in time") + } + time.Sleep(200 * time.Millisecond) + } +} + +func (f *futureMaster) replicaRunning() (ioRunning, sqlRunning bool) { + rows, err := f.db.Query("SHOW REPLICA STATUS") + require.NoError(f.t, err) + defer rows.Close() + + cols, err := rows.Columns() + require.NoError(f.t, err) + if !rows.Next() { + return false, false + } + vals := make([]interface{}, len(cols)) + ptrs := make([]interface{}, len(cols)) + for i := range vals { + ptrs[i] = &vals[i] + } + require.NoError(f.t, rows.Scan(ptrs...)) + for i, c := range cols { + s := asString(vals[i]) + if c == "Replica_IO_Running" { + ioRunning = s == "Yes" + } + if c == "Replica_SQL_Running" { + sqlRunning = s == "Yes" + } + } + return ioRunning, sqlRunning +} + +func (f *futureMaster) waitUntilCaughtUpToSource() { + f.t.Helper() + sourceGTID := f.sourceExecutedGTIDSet() + if sourceGTID == "" { + return + } + var result sqlorig.NullInt64 + require.NoError(f.t, f.db.QueryRow("SELECT WAIT_FOR_EXECUTED_GTID_SET(?, 20)", sourceGTID).Scan(&result)) + require.Equal(f.t, int64(0), result.Int64, "future master did not catch up to source in time") +} + +func (f *futureMaster) promote() { + f.t.Helper() + f.waitUntilCaughtUpToSource() + f.exec(f.db, "STOP REPLICA") + f.exec(f.db, "RESET REPLICA ALL") + f.exec(f.db, "SET GLOBAL read_only = OFF") + f.execIgnore(f.db, "SET GLOBAL super_read_only = OFF") +} + +func (f *futureMaster) stopSourceContainer() { + f.t.Helper() + name := sourceContainerName(f.containerTool) + require.NotEmpty(f.t, name, "source container not found") + require.NoError(f.t, exec.Command(f.containerTool, "stop", name).Run()) +} + +func (f *futureMaster) restoreSourceContainer() { + name := sourceContainerName(f.containerTool) + if name == "" { + return + } + _ = exec.Command(f.containerTool, "start", name).Run() + // Wait until the source accepts connections again for later tests. + deadline := time.Now().Add(180 * time.Second) + for time.Now().Before(deadline) { + db, err := sqlwrapper.Open("mysql", fmt.Sprintf("root@tcp(127.0.0.1:%d)/", testhelpers.TestSourcePort), "") + if err == nil { + pingErr := db.Ping() + db.Close() + if pingErr == nil { + return + } + } + time.Sleep(500 * time.Millisecond) + } +} + +func (f *futureMaster) reset() { + f.execIgnore(f.db, "STOP REPLICA") + f.execIgnore(f.db, "RESET REPLICA ALL") + f.execIgnore(f.db, "SET GLOBAL read_only = OFF") + for _, name := range testhelpers.ApplicableTestDbs { + f.execIgnore(f.db, fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", name)) + } +} + +func asString(v interface{}) string { + switch t := v.(type) { + case []byte: + return string(t) + case string: + return t + case nil: + return "" + default: + return fmt.Sprintf("%v", t) + } +} + +func versionAtLeast(version string, major, minor int) bool { + var maj, min int + if _, err := fmt.Sscanf(version, "%d.%d", &maj, &min); err != nil { + return false + } + return maj > major || (maj == major && min >= minor) +}