From 8f17c1fdae58dceccfb5715c1b031f1eec5dcb8b Mon Sep 17 00:00:00 2001 From: suetin Date: Thu, 27 Aug 2026 19:50:07 +0300 Subject: [PATCH 1/4] cancel switchover on timeout if possible --- internal/app/app.go | 80 +++++++++++----- internal/app/data.go | 7 +- internal/app/switchover_timeout.go | 42 +++++++++ internal/app/switchover_timeout_test.go | 117 ++++++++++++++++++++++++ internal/config/config.go | 2 +- internal/config/config_test.go | 14 +++ 6 files changed, 235 insertions(+), 27 deletions(-) create mode 100644 internal/app/switchover_timeout.go create mode 100644 internal/app/switchover_timeout_test.go create mode 100644 internal/config/config_test.go diff --git a/internal/app/app.go b/internal/app/app.go index d4010c0e..e8a6dc9e 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -473,12 +473,12 @@ func (app *App) stateManager() appState { if lightMaintenance && switchover.MasterTransition == FailoverTransition { app.logger.Info().Msgf("failover suppressed by light maintenance mode") } else { - if !switchover.InitiatedAt.IsZero() && time.Since(switchover.InitiatedAt) > app.config.SwitchoverTimeout { - app.logger.Error().Msgf("switchover %s => %s timed out after %s", switchover.From, switchover.To, time.Since(switchover.InitiatedAt)) - app.logSwitchoverFailure(switchover) - err = app.FailSwitchover(switchover, fmt.Errorf("switchover timed out after %s", time.Since(switchover.InitiatedAt))) + abortDeadline := app.newSwitchoverAbortDeadline(switchover) + if timeoutErr := abortDeadline.exceeded(time.Now()); timeoutErr != nil { + app.logger.Error().Err(timeoutErr).Msgf("switchover %s => %s timed out before it started", switchover.From, switchover.To) + err = app.FinishSwitchover(switchover, timeoutErr) if err != nil { - app.logger.Error().Err(err).Msg("failed to report switchover timeout") + app.logger.Error().Err(err).Msg("failed to reject timed out switchover") } return stateManager } @@ -497,23 +497,19 @@ func (app *App) stateManager() appState { app.logger.Error().Err(err).Msg("failed to start switchover") return stateManager } - err = app.performSwitchover(clusterState, activeNodes, switchover, master) + err = app.performSwitchover(clusterState, activeNodes, switchover, master, abortDeadline) if errors.Is(app.GetCurrentSwitchover(new(Switchover)), dcs.ErrNotFound) { app.logger.Error().Msgf("switchover was aborted") } else { + if errors.Is(err, ErrSwitchoverTimeout) { + app.logger.Error().Err(err).Msgf("switchover %s => %s timed out at a safe abort point", switchover.From, switchover.To) + } + err = app.recordSwitchoverAttemptResult(switchover, err) if err != nil { - err = app.FailSwitchover(switchover, err) - if err != nil { - app.logger.Error().Err(err).Msg("failed to report switchover failure") - } - } else { - err = app.FinishSwitchover(switchover, nil) - if err != nil { - // we failed to update status in DCS, it's highly possible - // that current process lost DCS connection - // and another process will take managerLock - app.logger.Error().Err(err).Msg("failed to report switchover finish") - } + // we failed to update status in DCS, it's highly possible + // that current process lost DCS connection + // and another process will take managerLock + app.logger.Error().Err(err).Msg("failed to report switchover result") } } return stateManager @@ -1256,7 +1252,13 @@ func (app *App) disableSemiSyncIfNonNeeded(node *mysql.Node, state *nodestate.No } // nolint: gocyclo, funlen -func (app *App) performSwitchover(clusterState map[string]*nodestate.NodeState, activeNodes []string, switchover *Switchover, oldMaster string) error { +func (app *App) performSwitchover( + clusterState map[string]*nodestate.NodeState, + activeNodes []string, + switchover *Switchover, + oldMaster string, + abortDeadline *switchoverAbortDeadline, +) error { if switchover.To != "" { if !slices.Contains(activeNodes, switchover.To) { return errors.New("switchover: failed: replica is not active, can't switch to it") @@ -1444,7 +1446,15 @@ func (app *App) performSwitchover(clusterState map[string]*nodestate.NodeState, } else { app.logger.Info().Msgf("switchover: new master %s is the most recent host, waiting for all binlogs to be applied", newMaster) } - caught, err := app.waitForCatchUp(newMasterNode, mostRecentGtidSet, app.config.SlaveCatchUpTimeout, time.Second) + caught, err := app.waitForCatchUp(newMasterNode, mostRecentGtidSet, app.config.SlaveCatchUpTimeout, time.Second, abortDeadline) + if errors.Is(err, ErrSwitchoverTimeout) { + // Catch-up may take long enough for this process to lose the manager + // lock. Only the current manager is allowed to reject the switchover. + if !app.AcquireLock(pathManagerLock) { + return errors.New("manager lock lost during switchover, new manager should finish the process, leaving") + } + return err + } if err != nil || app.emulateError("catchup_master_status") { return fmt.Errorf("failed to get gtid executed from %s: %w", newMaster, err) } @@ -1466,6 +1476,9 @@ func (app *App) performSwitchover(clusterState map[string]*nodestate.NodeState, if dubious := getDubiousHAHosts(clusterState); len(dubious) > 0 { return fmt.Errorf("switchover: failed to ping hosts: %v with dubious errors", dubious) } + if err := abortDeadline.exceeded(time.Now()); err != nil { + return err + } // turn slaves to the new master app.logger.Info().Msg("switchover: phase 5: turn to the new master") @@ -2322,9 +2335,18 @@ func (app *App) getClusterStateFromDcs() (map[string]*nodestate.NodeState, error return getNodeStatesInParallel(hosts, getter, app.logger) } -func (app *App) waitForCatchUp(node *mysql.Node, gtidset gtids.GTIDSet, timeout time.Duration, sleep time.Duration) (bool, error) { +func (app *App) waitForCatchUp( + node *mysql.Node, + gtidset gtids.GTIDSet, + timeout time.Duration, + sleep time.Duration, + abortDeadline *switchoverAbortDeadline, +) (bool, error) { deadline := time.Now().Add(timeout) for { + if err := abortDeadline.exceeded(time.Now()); err != nil { + return false, err + } gtidExecuted, err := node.GTIDExecutedParsed() if err != nil { return false, err @@ -2340,10 +2362,22 @@ func (app *App) waitForCatchUp(node *mysql.Node, gtidset gtids.GTIDSet, timeout if app.CheckAsyncSwitchAllowed(node, switchover) { return true, nil } - time.Sleep(sleep) - if time.Now().After(deadline) { + now := time.Now() + if !now.Before(deadline) { break } + wait := sleep + if remaining := deadline.Sub(now); wait > remaining { + wait = remaining + } + if abortDeadline != nil { + if remaining := abortDeadline.at.Sub(now); wait > remaining { + wait = remaining + } + } + if wait > 0 { + time.Sleep(wait) + } } return false, nil } diff --git a/internal/app/data.go b/internal/app/data.go index 5008e336..78129d8f 100644 --- a/internal/app/data.go +++ b/internal/app/data.go @@ -17,9 +17,10 @@ const ( ) var ( - ErrNoMaster = errors.New("no alive master found") - ErrManyMasters = errors.New("more than one master found") - ErrNoActiveNodes = errors.New("no active nodes found") + ErrNoMaster = errors.New("no alive master found") + ErrManyMasters = errors.New("more than one master found") + ErrNoActiveNodes = errors.New("no active nodes found") + ErrSwitchoverTimeout = errors.New("switchover timed out") ) const ( diff --git a/internal/app/switchover_timeout.go b/internal/app/switchover_timeout.go new file mode 100644 index 00000000..9087df4c --- /dev/null +++ b/internal/app/switchover_timeout.go @@ -0,0 +1,42 @@ +package app + +import ( + "errors" + "fmt" + "time" +) + +// switchoverAbortDeadline is created only for a switchover that has never been +// attempted before. Without a persisted phase, a retry cannot be aborted +// safely: an earlier attempt might have already changed the replication +// topology in phase 5. +type switchoverAbortDeadline struct { + at time.Time + timeout time.Duration +} + +func (app *App) newSwitchoverAbortDeadline(switchover *Switchover) *switchoverAbortDeadline { + if switchover.InitiatedAt.IsZero() || !switchover.StartedAt.IsZero() || switchover.RunCount != 0 { + return nil + } + return &switchoverAbortDeadline{ + at: switchover.InitiatedAt.Add(app.config.SwitchoverTimeout), + timeout: app.config.SwitchoverTimeout, + } +} + +func (deadline *switchoverAbortDeadline) exceeded(now time.Time) error { + if deadline == nil || now.Before(deadline.at) { + return nil + } + return fmt.Errorf("%w after %s", ErrSwitchoverTimeout, deadline.timeout) +} + +// recordSwitchoverAttemptResult retries regular errors, but a timeout observed +// at a safe abort point is terminal and moves the switch to last_rejected. +func (app *App) recordSwitchoverAttemptResult(switchover *Switchover, switchErr error) error { + if switchErr == nil || errors.Is(switchErr, ErrSwitchoverTimeout) { + return app.FinishSwitchover(switchover, switchErr) + } + return app.FailSwitchover(switchover, switchErr) +} diff --git a/internal/app/switchover_timeout_test.go b/internal/app/switchover_timeout_test.go new file mode 100644 index 00000000..7c58802d --- /dev/null +++ b/internal/app/switchover_timeout_test.go @@ -0,0 +1,117 @@ +package app + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/golang/mock/gomock" + "github.com/stretchr/testify/require" +) + +func TestNewSwitchoverAbortDeadline(t *testing.T) { + initiatedAt := time.Date(2026, time.August, 27, 12, 0, 0, 0, time.UTC) + tests := []struct { + name string + switchover Switchover + want bool + }{ + { + name: "first attempt", + switchover: Switchover{InitiatedAt: initiatedAt}, + want: true, + }, + { + name: "started by a previous manager", + switchover: Switchover{ + InitiatedAt: initiatedAt, + StartedAt: initiatedAt.Add(time.Minute), + }, + }, + { + name: "retry", + switchover: Switchover{ + InitiatedAt: initiatedAt, + RunCount: 1, + }, + }, + { + name: "legacy record without initiation time", + switchover: Switchover{}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := minConfig() + cfg.SwitchoverTimeout = 10 * time.Minute + app := newTestApp(t, cfg, nil) + + deadline := app.newSwitchoverAbortDeadline(&tt.switchover) + if !tt.want { + require.Nil(t, deadline) + return + } + require.NotNil(t, deadline) + require.Equal(t, initiatedAt.Add(10*time.Minute), deadline.at) + require.NoError(t, deadline.exceeded(deadline.at.Add(-time.Nanosecond))) + err := deadline.exceeded(deadline.at) + require.ErrorIs(t, err, ErrSwitchoverTimeout) + require.EqualError(t, err, "switchover timed out after 10m0s") + }) + } +} + +func TestWaitForCatchUpHonorsSwitchoverDeadline(t *testing.T) { + app := newTestApp(t, minConfig(), nil) + deadline := &switchoverAbortDeadline{ + at: time.Now().Add(-time.Second), + timeout: 10 * time.Minute, + } + + caught, err := app.waitForCatchUp(nil, nil, time.Hour, time.Hour, deadline) + require.False(t, caught) + require.ErrorIs(t, err, ErrSwitchoverTimeout) +} + +func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().DeleteCurrentSwitchover().Return(nil) + mockDCS.EXPECT().SetLastRejectedSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.Equal(t, 0, switchover.RunCount) + require.NotNil(t, switchover.Result) + require.False(t, switchover.Result.Ok) + require.Equal(t, "switchover timed out after 10m0s", switchover.Result.Error) + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + switchover := &Switchover{MasterTransition: FailoverTransition} + err := app.recordSwitchoverAttemptResult( + switchover, + fmt.Errorf("%w after %s", ErrSwitchoverTimeout, 10*time.Minute), + ) + require.NoError(t, err) +} + +func TestRecordSwitchoverAttemptResultRegularErrorIsRetried(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().SetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.Equal(t, 1, switchover.RunCount) + require.NotNil(t, switchover.Result) + require.False(t, switchover.Result.Ok) + require.Equal(t, "temporary failure", switchover.Result.Error) + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + err := app.recordSwitchoverAttemptResult(&Switchover{}, errors.New("temporary failure")) + require.NoError(t, err) +} diff --git a/internal/config/config.go b/internal/config/config.go index a6013fe0..f8c4e1fe 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -214,7 +214,7 @@ func DefaultConfig() (Config, error) { ShowOnlyGTIDDiff: false, ManagerSwitchover: false, ForceSwitchover: false, - SwitchoverTimeout: 30 * time.Minute, + SwitchoverTimeout: 10 * time.Minute, SwitchoverMaxAttempts: 60, ReplicationConvergenceTimeoutSwitchover: 300 * time.Second, DSNSettings: "?autocommit=1&sql_log_off=1", diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 00000000..2b642a67 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,14 @@ +package config + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDefaultSwitchoverTimeout(t *testing.T) { + cfg, err := DefaultConfig() + require.NoError(t, err) + require.Equal(t, 10*time.Minute, cfg.SwitchoverTimeout) +} From 8a9041420bb93f0f0a2e920a157fe217454ec407 Mon Sep 17 00:00:00 2001 From: suetin Date: Fri, 28 Aug 2026 00:42:30 +0300 Subject: [PATCH 2/4] cancel switchover on timeout if possible --- internal/app/app.go | 19 +++++-- internal/app/app_dcs.go | 12 +++-- internal/app/app_dcs_impl.go | 24 +++++++-- internal/app/cli_switch.go | 1 + internal/app/data.go | 2 + internal/app/idcs.go | 1 + internal/app/mock_idcs_test.go | 14 +++++ internal/app/switchover_timeout.go | 24 +++++++-- internal/app/switchover_timeout_test.go | 69 +++++++++++++++++++++++-- internal/dcs/dcs.go | 5 ++ internal/dcs/zk.go | 50 ++++++++++++++++-- tests/features/switchover_to.feature | 64 +++++++++++++++++++++++ tests/images/docker-compose.yaml | 3 ++ tests/images/mysql/mysync.yaml | 1 + tests/mysync_test.go | 22 ++++++++ 15 files changed, 284 insertions(+), 27 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index e8a6dc9e..bab7500d 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -476,6 +476,10 @@ func (app *App) stateManager() appState { abortDeadline := app.newSwitchoverAbortDeadline(switchover) if timeoutErr := abortDeadline.exceeded(time.Now()); timeoutErr != nil { app.logger.Error().Err(timeoutErr).Msgf("switchover %s => %s timed out before it started", switchover.From, switchover.To) + if !app.AcquireLock(pathManagerLock) { + app.logger.Error().Msg("manager lock lost before rejecting timed out switchover") + return stateManager + } err = app.FinishSwitchover(switchover, timeoutErr) if err != nil { app.logger.Error().Err(err).Msg("failed to reject timed out switchover") @@ -503,6 +507,10 @@ func (app *App) stateManager() appState { } else { if errors.Is(err, ErrSwitchoverTimeout) { app.logger.Error().Err(err).Msgf("switchover %s => %s timed out at a safe abort point", switchover.From, switchover.To) + if !app.AcquireLock(pathManagerLock) { + app.logger.Error().Msg("manager lock lost before rejecting timed out switchover") + return stateManager + } } err = app.recordSwitchoverAttemptResult(switchover, err) if err != nil { @@ -1448,11 +1456,6 @@ func (app *App) performSwitchover( } caught, err := app.waitForCatchUp(newMasterNode, mostRecentGtidSet, app.config.SlaveCatchUpTimeout, time.Second, abortDeadline) if errors.Is(err, ErrSwitchoverTimeout) { - // Catch-up may take long enough for this process to lose the manager - // lock. Only the current manager is allowed to reject the switchover. - if !app.AcquireLock(pathManagerLock) { - return errors.New("manager lock lost during switchover, new manager should finish the process, leaving") - } return err } if err != nil || app.emulateError("catchup_master_status") { @@ -1479,6 +1482,12 @@ func (app *App) performSwitchover( if err := abortDeadline.exceeded(time.Now()); err != nil { return err } + if !app.AcquireLock(pathManagerLock) { + return errors.New("manager lock lost before changing replication topology, new manager should finish the process, leaving") + } + if err := app.markSwitchoverUnabortable(switchover); err != nil { + return err + } // turn slaves to the new master app.logger.Info().Msg("switchover: phase 5: turn to the new master") diff --git a/internal/app/app_dcs.go b/internal/app/app_dcs.go index b2e766a0..6c6acca5 100644 --- a/internal/app/app_dcs.go +++ b/internal/app/app_dcs.go @@ -132,7 +132,6 @@ func (app *App) FinishSwitchover(switchover *Switchover, switchErr error) error path = pathLastRejectedSwitch } - app.logger.Info().Msgf("switchover: %s => %s %s", switchover.From, switchover.To, action) switchover.Result = new(SwitchoverResult) switchover.Result.Ok = result switchover.Result.FinishedAt = time.Now() @@ -141,6 +140,12 @@ func (app *App) FinishSwitchover(switchover *Switchover, switchErr error) error switchover.Result.Error = switchErr.Error() } + err := app.appDCS.DeleteCurrentSwitchoverVersion(switchover.DCSVersion) + if err != nil { + return err + } + + app.logger.Info().Msgf("switchover: %s => %s %s", switchover.From, switchover.To, action) if switchErr != nil { app.logSwitchoverFailure(switchover) } else if switchover.MasterTransition != FailoverTransition { @@ -149,10 +154,6 @@ func (app *App) FinishSwitchover(switchover *Switchover, switchErr error) error app.stopTiming(timingFailover) } - err := app.appDCS.DeleteCurrentSwitchover() - if err != nil { - return err - } if path == pathLastSwitch { return app.appDCS.SetLastSwitchover(switchover) } @@ -217,6 +218,7 @@ func (app *App) IssueFailover(master string) error { InitiatedAt: time.Now(), Cause: CauseAuto, MasterTransition: FailoverTransition, + Abortable: true, } return app.appDCS.CreateCurrentSwitchover(&switchover) } diff --git a/internal/app/app_dcs_impl.go b/internal/app/app_dcs_impl.go index 1b3aa303..ad38930e 100644 --- a/internal/app/app_dcs_impl.go +++ b/internal/app/app_dcs_impl.go @@ -197,17 +197,29 @@ func (a *appDCS) GetLastSwitchover(switchover *Switchover) error { // GetCurrentSwitchover reads the current in-progress switchover from ZK. // Returns dcs.ErrNotFound if no switchover is in progress. func (a *appDCS) GetCurrentSwitchover(switchover *Switchover) error { - return a.dcs.Get(pathCurrentSwitch, switchover) + version, err := a.dcs.GetVersion(pathCurrentSwitch, switchover) + if err == nil { + switchover.DCSVersion = version + } + return err } // CreateCurrentSwitchover creates a new switchover record in ZK (fails if one already exists). func (a *appDCS) CreateCurrentSwitchover(switchover *Switchover) error { - return a.dcs.Create(pathCurrentSwitch, switchover) + err := a.dcs.Create(pathCurrentSwitch, switchover) + if err == nil { + switchover.DCSVersion = 0 + } + return err } // SetCurrentSwitchover writes the current in-progress switchover to ZK. func (a *appDCS) SetCurrentSwitchover(switchover *Switchover) error { - return a.dcs.Set(pathCurrentSwitch, switchover) + version, err := a.dcs.SetVersion(pathCurrentSwitch, switchover, switchover.DCSVersion) + if err == nil { + switchover.DCSVersion = version + } + return err } // DeleteCurrentSwitchover removes the current switchover node from ZK. @@ -215,6 +227,12 @@ func (a *appDCS) DeleteCurrentSwitchover() error { return a.dcs.Delete(pathCurrentSwitch) } +// DeleteCurrentSwitchoverVersion removes the current switchover only if it has +// not been updated by another manager since this process read it. +func (a *appDCS) DeleteCurrentSwitchoverVersion(version int32) error { + return a.dcs.DeleteVersion(pathCurrentSwitch, version) +} + // SetLastSwitchover writes the completed switchover result to ZK. func (a *appDCS) SetLastSwitchover(switchover *Switchover) error { return a.dcs.Set(pathLastSwitch, switchover) diff --git a/internal/app/cli_switch.go b/internal/app/cli_switch.go index d5e446bc..3b6e6d12 100644 --- a/internal/app/cli_switch.go +++ b/internal/app/cli_switch.go @@ -148,6 +148,7 @@ func (app *App) CliSwitch(switchFrom, switchTo string, waitTimeout time.Duration switchover.InitiatedBy = util.GuessWhoRunning() + "@" + app.config.Hostname switchover.InitiatedAt = time.Now() switchover.Cause = CauseManual + switchover.Abortable = true if failover { switchover.MasterTransition = FailoverTransition } else { diff --git a/internal/app/data.go b/internal/app/data.go index 78129d8f..f6675fb5 100644 --- a/internal/app/data.go +++ b/internal/app/data.go @@ -51,6 +51,8 @@ type Switchover struct { StartedAt time.Time `json:"started_at"` Result *SwitchoverResult `json:"result"` RunCount int `json:"run_count,omitempty"` + Abortable bool `json:"abortable,omitempty"` + DCSVersion int32 `json:"-"` } func (sw *Switchover) String() string { diff --git a/internal/app/idcs.go b/internal/app/idcs.go index 93c92a67..b55b908d 100644 --- a/internal/app/idcs.go +++ b/internal/app/idcs.go @@ -48,6 +48,7 @@ type IAppDCS interface { GetLastSwitchover(switchover *Switchover) error SetCurrentSwitchover(switchover *Switchover) error DeleteCurrentSwitchover() error + DeleteCurrentSwitchoverVersion(version int32) error SetLastSwitchover(switchover *Switchover) error SetLastRejectedSwitchover(switchover *Switchover) error GetLastRejectedSwitchover(switchover *Switchover) error diff --git a/internal/app/mock_idcs_test.go b/internal/app/mock_idcs_test.go index 0b27b121..ba34e03c 100644 --- a/internal/app/mock_idcs_test.go +++ b/internal/app/mock_idcs_test.go @@ -92,6 +92,20 @@ func (mr *MockIAppDCSMockRecorder) DeleteCurrentSwitchover() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteCurrentSwitchover", reflect.TypeOf((*MockIAppDCS)(nil).DeleteCurrentSwitchover)) } +// DeleteCurrentSwitchoverVersion mocks base method. +func (m *MockIAppDCS) DeleteCurrentSwitchoverVersion(version int32) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteCurrentSwitchoverVersion", version) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteCurrentSwitchoverVersion indicates an expected call of DeleteCurrentSwitchoverVersion. +func (mr *MockIAppDCSMockRecorder) DeleteCurrentSwitchoverVersion(version interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteCurrentSwitchoverVersion", reflect.TypeOf((*MockIAppDCS)(nil).DeleteCurrentSwitchoverVersion), version) +} + // DeleteMaintenance mocks base method. func (m *MockIAppDCS) DeleteMaintenance() error { m.ctrl.T.Helper() diff --git a/internal/app/switchover_timeout.go b/internal/app/switchover_timeout.go index 9087df4c..1fe8ef3c 100644 --- a/internal/app/switchover_timeout.go +++ b/internal/app/switchover_timeout.go @@ -6,17 +6,21 @@ import ( "time" ) -// switchoverAbortDeadline is created only for a switchover that has never been -// attempted before. Without a persisted phase, a retry cannot be aborted -// safely: an earlier attempt might have already changed the replication -// topology in phase 5. +// switchoverAbortDeadline exists while the persisted switchover is still at a +// safe abort point. Abortable remains true across retries and manager restarts +// and is cleared in DCS before phase 5 changes the replication topology. type switchoverAbortDeadline struct { at time.Time timeout time.Duration } func (app *App) newSwitchoverAbortDeadline(switchover *Switchover) *switchoverAbortDeadline { - if switchover.InitiatedAt.IsZero() || !switchover.StartedAt.IsZero() || switchover.RunCount != 0 { + // Records created by an older worker have no abortable field. It is safe to + // initialize it only before their first attempt has started. + if !switchover.Abortable && switchover.StartedAt.IsZero() && switchover.RunCount == 0 { + switchover.Abortable = true + } + if switchover.InitiatedAt.IsZero() || !switchover.Abortable { return nil } return &switchoverAbortDeadline{ @@ -25,6 +29,16 @@ func (app *App) newSwitchoverAbortDeadline(switchover *Switchover) *switchoverAb } } +func (app *App) markSwitchoverUnabortable(switchover *Switchover) error { + updated := *switchover + updated.Abortable = false + if err := app.appDCS.SetCurrentSwitchover(&updated); err != nil { + return fmt.Errorf("failed to persist switchover safe-abort boundary: %w", err) + } + *switchover = updated + return nil +} + func (deadline *switchoverAbortDeadline) exceeded(now time.Time) error { if deadline == nil || now.Before(deadline.at) { return nil diff --git a/internal/app/switchover_timeout_test.go b/internal/app/switchover_timeout_test.go index 7c58802d..f5457e73 100644 --- a/internal/app/switchover_timeout_test.go +++ b/internal/app/switchover_timeout_test.go @@ -8,6 +8,8 @@ import ( "github.com/golang/mock/gomock" "github.com/stretchr/testify/require" + + "github.com/yandex/mysync/internal/dcs" ) func TestNewSwitchoverAbortDeadline(t *testing.T) { @@ -23,18 +25,36 @@ func TestNewSwitchoverAbortDeadline(t *testing.T) { want: true, }, { - name: "started by a previous manager", + name: "legacy record started by a previous manager", + switchover: Switchover{ + InitiatedAt: initiatedAt, + StartedAt: initiatedAt.Add(time.Minute), + }, + }, + { + name: "legacy retry", + switchover: Switchover{ + InitiatedAt: initiatedAt, + RunCount: 1, + }, + }, + { + name: "started by a previous manager while still abortable", switchover: Switchover{ InitiatedAt: initiatedAt, StartedAt: initiatedAt.Add(time.Minute), + Abortable: true, }, + want: true, }, { - name: "retry", + name: "retry before topology change remains abortable", switchover: Switchover{ InitiatedAt: initiatedAt, RunCount: 1, + Abortable: true, }, + want: true, }, { name: "legacy record without initiation time", @@ -49,6 +69,9 @@ func TestNewSwitchoverAbortDeadline(t *testing.T) { app := newTestApp(t, cfg, nil) deadline := app.newSwitchoverAbortDeadline(&tt.switchover) + if tt.switchover.StartedAt.IsZero() && tt.switchover.RunCount == 0 && !tt.switchover.InitiatedAt.IsZero() { + require.True(t, tt.switchover.Abortable) + } if !tt.want { require.Nil(t, deadline) return @@ -63,6 +86,29 @@ func TestNewSwitchoverAbortDeadline(t *testing.T) { } } +func TestMarkSwitchoverUnabortablePersistsBoundary(t *testing.T) { + for _, abortable := range []bool{true, false} { + t.Run(fmt.Sprintf("abortable_%t", abortable), func(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().SetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.False(t, switchover.Abortable) + require.Equal(t, int32(7), switchover.DCSVersion) + switchover.DCSVersion = 8 + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + switchover := &Switchover{Abortable: abortable, DCSVersion: 7} + require.NoError(t, app.markSwitchoverUnabortable(switchover)) + require.False(t, switchover.Abortable) + require.Equal(t, int32(8), switchover.DCSVersion) + }) + } +} + func TestWaitForCatchUpHonorsSwitchoverDeadline(t *testing.T) { app := newTestApp(t, minConfig(), nil) deadline := &switchoverAbortDeadline{ @@ -80,7 +126,7 @@ func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { defer ctrl.Finish() mockDCS := NewMockIAppDCS(ctrl) - mockDCS.EXPECT().DeleteCurrentSwitchover().Return(nil) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(nil) mockDCS.EXPECT().SetLastRejectedSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { require.Equal(t, 0, switchover.RunCount) require.NotNil(t, switchover.Result) @@ -90,7 +136,7 @@ func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { }) app := newTestApp(t, minConfig(), mockDCS) - switchover := &Switchover{MasterTransition: FailoverTransition} + switchover := &Switchover{MasterTransition: FailoverTransition, DCSVersion: 7} err := app.recordSwitchoverAttemptResult( switchover, fmt.Errorf("%w after %s", ErrSwitchoverTimeout, 10*time.Minute), @@ -98,6 +144,21 @@ func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { require.NoError(t, err) } +func TestRecordSwitchoverAttemptResultDoesNotDeleteNewManagerState(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(dcs.ErrVersionMismatch) + + app := newTestApp(t, minConfig(), mockDCS) + err := app.recordSwitchoverAttemptResult( + &Switchover{MasterTransition: FailoverTransition, DCSVersion: 7}, + fmt.Errorf("%w after %s", ErrSwitchoverTimeout, 10*time.Minute), + ) + require.ErrorIs(t, err, dcs.ErrVersionMismatch) +} + func TestRecordSwitchoverAttemptResultRegularErrorIsRetried(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() diff --git a/internal/dcs/dcs.go b/internal/dcs/dcs.go index 9973de2c..fe158108 100644 --- a/internal/dcs/dcs.go +++ b/internal/dcs/dcs.go @@ -22,9 +22,12 @@ type DCS interface { Create(path string, value any) error CreateEphemeral(path string, value any) error Set(path string, value any) error + SetVersion(path string, value any, version int32) (int32, error) SetEphemeral(path string, value any) error Get(path string, dest any) error + GetVersion(path string, dest any) (int32, error) Delete(path string) error + DeleteVersion(path string, version int32) error GetTree(path string) (any, error) GetChildren(path string) ([]string, error) Close() @@ -37,6 +40,8 @@ var ( ErrNotFound = errors.New("key was not found in DCS") // ErrMalformed means that we failed to unmarshall received data ErrMalformed = errors.New("failed to parse DCS value, possibly data format changed") + // ErrVersionMismatch means the node changed after it was read. + ErrVersionMismatch = errors.New("node version mismatch") ) // sep is a path separator for most common DCS diff --git a/internal/dcs/zk.go b/internal/dcs/zk.go index 6fbab298..e203676f 100644 --- a/internal/dcs/zk.go +++ b/internal/dcs/zk.go @@ -465,6 +465,26 @@ func (z *zkDCS) Set(path string, val any) error { return z.set(path, val, 0) } +func (z *zkDCS) SetVersion(path string, val any, version int32) (int32, error) { + fullPath := z.buildFullPath(path) + data, err := json.Marshal(val) + if err != nil { + panic(fmt.Sprintf("failed to serialize to JSON %#v", val)) + } + stat, err := z.retrySet(fullPath, data, version) + if errors.Is(err, zk.ErrBadVersion) { + return 0, ErrVersionMismatch + } + if errors.Is(err, zk.ErrNoNode) { + return 0, ErrNotFound + } + if err != nil { + z.logger.Error().Err(err).Msgf("failed to set node %s at version %d to %+v", fullPath, version, val) + return 0, err + } + return stat.Version, nil +} + func (z *zkDCS) SetEphemeral(path string, val any) error { return z.set(path, val, zk.FlagEphemeral) } @@ -486,21 +506,41 @@ func (z *zkDCS) Delete(path string) error { return err } -func (z *zkDCS) Get(path string, dest any) error { +func (z *zkDCS) DeleteVersion(path string, version int32) error { fullPath := z.buildFullPath(path) - data, _, err := z.retryGet(fullPath) + err := z.retryDelete(fullPath, version) + if errors.Is(err, zk.ErrBadVersion) { + return ErrVersionMismatch + } if errors.Is(err, zk.ErrNoNode) { return ErrNotFound } + if err != nil { + z.logger.Error().Err(err).Msgf("failed to delete node %s at version %d", fullPath, version) + } + return err +} + +func (z *zkDCS) Get(path string, dest any) error { + _, err := z.GetVersion(path, dest) + return err +} + +func (z *zkDCS) GetVersion(path string, dest any) (int32, error) { + fullPath := z.buildFullPath(path) + data, stat, err := z.retryGet(fullPath) + if errors.Is(err, zk.ErrNoNode) { + return 0, ErrNotFound + } if err != nil { z.logger.Error().Err(err).Msgf("failed to get node %s", fullPath) - return err + return 0, err } if err = json.Unmarshal(data, dest); err != nil { z.logger.Error().Err(err).Msgf("malformed node data %s (%s)", fullPath, data) - return ErrMalformed + return 0, ErrMalformed } - return nil + return stat.Version, nil } func (z *zkDCS) GetTree(path string) (any, error) { diff --git a/tests/features/switchover_to.feature b/tests/features/switchover_to.feature index 16ea4d26..4a8cffdb 100644 --- a/tests/features/switchover_to.feature +++ b/tests/features/switchover_to.feature @@ -350,3 +350,67 @@ Feature: manual switchover to new master """ mysql2 is not active """ + + Scenario: timeout survives manager restart and restores the old master + Given cluster environment is + """ + MYSYNC_SEMISYNC=false + MYSYNC_SWITCHOVER_TIMEOUT=12s + OFFLINE_MODE_ENABLE_LAG=300s + """ + And cluster is up and running + Then mysql host "mysql1" should be master + And zookeeper node "/test/active_nodes" should match json_exactly within "30" seconds + """ + ["mysql1","mysql2","mysql3"] + """ + When I set replication delay on host "mysql2" to "60" seconds + And I run SQL on mysql host "mysql1" + """ + CREATE TABLE IF NOT EXISTS mysql.switchover_timeout_test (id INT PRIMARY KEY) + """ + And I run SQL on mysql host "mysql1" + """ + INSERT INTO mysql.switchover_timeout_test VALUES (1) + """ + And I wait for "3" seconds + And I get zookeeper node "/test/manager" + And I save zookeeper query result as "manager" + And I run command on host "mysql1" + """ + mysync switch --to mysql2 --wait=0s + """ + Then command return code should be "0" + And zookeeper node "/test/switch" should match json within "10" seconds + """ + { + "to": "mysql2", + "abortable": true, + "started_at": "REGEXP:^20[0-9]{2}-" + } + """ + When I run command on host "{{.manager.hostname}}" + """ + supervisorctl restart mysync + """ + Then command return code should be "0" + And zookeeper node "/test/last_rejected_switch" should match json within "30" seconds + """ + { + "to": "mysql2", + "abortable": true, + "result": { + "ok": false, + "error": "switchover timed out after 12s" + } + } + """ + And zookeeper node "/test/switch" should not exist + And mysql host "mysql1" should be master + And mysql host "mysql1" should become writable within "30" seconds + And mysql host "mysql2" should become replica of "mysql1" within "30" seconds + And mysql host "mysql3" should become replica of "mysql1" within "30" seconds + And mysql replication on host "mysql2" should run fine within "30" seconds + And mysql replication on host "mysql3" should run fine within "30" seconds + When I set replication delay on host "mysql2" to "0" seconds + Then mysql replication on host "mysql2" should run fine within "30" seconds diff --git a/tests/images/docker-compose.yaml b/tests/images/docker-compose.yaml index 2f6615ad..cbe0c073 100644 --- a/tests/images/docker-compose.yaml +++ b/tests/images/docker-compose.yaml @@ -109,6 +109,7 @@ services: LOW_REPLICATION_MARK: RESETUP_HOST_LAG: OFFLINE_MODE_MAX_OFFLINE_PCT: + MYSYNC_SWITCHOVER_TIMEOUT: MYSYNC_SWITCHOVER_MAX_ATTEMPTS: healthcheck: test: "mysql --user=admin --password=admin_pwd -e 'SELECT 1'" @@ -168,6 +169,7 @@ services: LOW_REPLICATION_MARK: RESETUP_HOST_LAG: OFFLINE_MODE_MAX_OFFLINE_PCT: + MYSYNC_SWITCHOVER_TIMEOUT: MYSYNC_SWITCHOVER_MAX_ATTEMPTS: depends_on: mysql1: @@ -221,6 +223,7 @@ services: LOW_REPLICATION_MARK: RESETUP_HOST_LAG: OFFLINE_MODE_MAX_OFFLINE_PCT: + MYSYNC_SWITCHOVER_TIMEOUT: MYSYNC_SWITCHOVER_MAX_ATTEMPTS: depends_on: mysql1: diff --git a/tests/images/mysql/mysync.yaml b/tests/images/mysql/mysync.yaml index f8ad1924..d1c3d488 100644 --- a/tests/images/mysql/mysync.yaml +++ b/tests/images/mysql/mysync.yaml @@ -73,6 +73,7 @@ repl_mon: ${REPL_MON:-false} force_switchover: ${FORCE_SWITCHOVER:-false} manager_switchover: ${MANAGER_SWITCHOVER:-true} replication_convergence_timeout_switchover: 300s +switchover_timeout: ${MYSYNC_SWITCHOVER_TIMEOUT:-10m} switchover_max_attempts: ${MYSYNC_SWITCHOVER_MAX_ATTEMPTS:-60} manager_election_delay_after_quorum_loss: ${MANAGER_ELECTION_DELAY_AFTER_QUORUM_LOSS:-15s} resetup_host_lag: ${RESETUP_HOST_LAG:-30000s} diff --git a/tests/mysync_test.go b/tests/mysync_test.go index 0cec9864..0a309407 100644 --- a/tests/mysync_test.go +++ b/tests/mysync_test.go @@ -1143,6 +1143,27 @@ func (tctx *testContext) stepBreakReplicationOnHostInARepairableWay(host string) return nil } +func (tctx *testContext) stepSetReplicationDelayOnHost(host string, delay int) error { + v, err := tctx.GetVersion(host) + if err != nil { + return err + } + stopQuery := fmt.Sprintf("STOP SLAVE FOR CHANNEL '%s'", replicationChannel) + changeQuery := fmt.Sprintf("CHANGE MASTER TO MASTER_DELAY = %d FOR CHANNEL '%s'", delay, replicationChannel) + startQuery := fmt.Sprintf("START SLAVE FOR CHANNEL '%s'", replicationChannel) + if v.CheckIfVersionReplicaStatus() { + stopQuery = fmt.Sprintf("STOP REPLICA FOR CHANNEL '%s'", replicationChannel) + changeQuery = fmt.Sprintf("CHANGE REPLICATION SOURCE TO SOURCE_DELAY = %d FOR CHANNEL '%s'", delay, replicationChannel) + startQuery = fmt.Sprintf("START REPLICA FOR CHANNEL '%s'", replicationChannel) + } + for _, query := range []string{stopQuery, changeQuery, startQuery} { + if _, err := tctx.queryMysql(host, query, nil); err != nil { + return err + } + } + return nil +} + func (tctx *testContext) stepIGetZookeeperNode(node string) error { data, _, err := tctx.zk.Get(node) if err != nil { @@ -1842,6 +1863,7 @@ func InitializeScenario(s *godog.ScenarioContext) { s.Step(`^external replication source on mysql host "([^"]*)" should remain "([^"]*)" for "(\d+)" seconds$`, tctx.stepExternalReplicationSourceShouldRemainFor) s.Step(`^I break replication on host "([^"]*)"$`, tctx.stepBreakReplicationOnHost) s.Step(`^I break replication on host "([^"]*)" in repairable way$`, tctx.stepBreakReplicationOnHostInARepairableWay) + s.Step(`^I set replication delay on host "([^"]*)" to "(\d+)" seconds$`, tctx.stepSetReplicationDelayOnHost) s.Step(`^I set used space on host "([^"]*)" to (\d+)%$`, tctx.stepSetUsedSpace) s.Step(`^I set readonly file system on host "([^"]*)" to "([^"]*)"$`, tctx.stepSetReadonlyStatus) s.Step(`^SQL result should match (\w+) witch I will save as "(\w+)"$`, tctx.stepSQLResultShouldMatchWithTextIWillSave) From 551ec856fd77fe57f2f79cf0caae00447aa726ec Mon Sep 17 00:00:00 2001 From: suetin Date: Fri, 28 Aug 2026 03:24:50 +0300 Subject: [PATCH 3/4] cancel switchover on timeout if possible --- README.md | 5 ++ cmd/mysync/safe_abort.go | 31 ++++++++++ internal/app/cli_switch.go | 44 ++++++++++++++ internal/app/cli_switch_test.go | 65 +++++++++++++++++++++ internal/app/data.go | 9 +-- internal/app/switchover_timeout_test.go | 18 ++++++ tests/features/9.7/cascade_replicas.feature | 2 +- tests/features/cascade_replicas.84.feature | 2 +- tests/features/cascade_replicas.feature | 2 +- tests/features/switchover_to.feature | 6 +- 10 files changed, 175 insertions(+), 9 deletions(-) create mode 100644 cmd/mysync/safe_abort.go create mode 100644 internal/app/cli_switch_test.go diff --git a/README.md b/README.md index 9b5db4f1..ac52d62b 100644 --- a/README.md +++ b/README.md @@ -131,8 +131,13 @@ mysync hosts add fqdn3.db.company.net mysync info -s mysync switch --to fqdn2 mysync switch --from fqdn2 +mysync safe-abort +mysync abort mysync maint on mysync maint off ``` +`mysync safe-abort` aborts a switchover only while it is marked as safely +abortable. `mysync abort` remains a force operation and may require manual +cluster repair. diff --git a/cmd/mysync/safe_abort.go b/cmd/mysync/safe_abort.go new file mode 100644 index 00000000..b8d238c8 --- /dev/null +++ b/cmd/mysync/safe_abort.go @@ -0,0 +1,31 @@ +package main + +import ( + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/yandex/mysync/internal/app" +) + +var safeAbortCmd = &cobra.Command{ + Use: "safe-abort", + Short: "Abort a switchover only before the safe-abort boundary", + Long: "Atomically removes the current switchover only while it is marked abortable. " + + "It never force-aborts a switchover that has started changing the replication topology.", + Run: func(cmd *cobra.Command, args []string) { + app, err := app.NewApp(configFile, logLevel, true) + if err != nil { + fmt.Println(err) + os.Exit(1) + } + code := app.CliSafeAbort() + app.CloseLogger() + os.Exit(code) + }, +} + +func init() { + rootCmd.AddCommand(safeAbortCmd) +} diff --git a/internal/app/cli_switch.go b/internal/app/cli_switch.go index 3b6e6d12..0b76aa74 100644 --- a/internal/app/cli_switch.go +++ b/internal/app/cli_switch.go @@ -240,3 +240,47 @@ func (app *App) CliAbort() int { fmt.Printf("switchover aborted\n") return 0 } + +// safeAbortSwitchover removes the current switchover only while it is still +// before the persisted safe-abort boundary. The versioned delete prevents a +// concurrent manager from crossing that boundary between the read and delete. +func (app *App) safeAbortSwitchover() error { + switchover := new(Switchover) + if err := app.GetCurrentSwitchover(switchover); err != nil { + return err + } + if !switchover.Abortable { + return ErrSwitchoverNotAbortable + } + return app.appDCS.DeleteCurrentSwitchoverVersion(switchover.DCSVersion) +} + +// CliSafeAbort safely cleans an abortable switchover node from DCS. +func (app *App) CliSafeAbort() int { + err := app.connectDCS() + if err != nil { + app.logger.Error().Err(err).Msg("") + return 1 + } + defer app.dcs.Close() + app.dcs.Initialize() + + err = app.safeAbortSwitchover() + switch { + case errors.Is(err, dcs.ErrNotFound): + fmt.Println("no active switchover") + return 0 + case errors.Is(err, ErrSwitchoverNotAbortable): + fmt.Println("switchover has crossed the safe-abort boundary; nothing was aborted") + return 1 + case errors.Is(err, dcs.ErrVersionMismatch): + fmt.Println("switchover changed while safe abort was attempted; nothing was aborted") + return 1 + case err != nil: + app.logger.Error().Err(err).Msg("") + return 1 + } + + fmt.Println("switchover safely aborted") + return 0 +} diff --git a/internal/app/cli_switch_test.go b/internal/app/cli_switch_test.go new file mode 100644 index 00000000..801099eb --- /dev/null +++ b/internal/app/cli_switch_test.go @@ -0,0 +1,65 @@ +package app + +import ( + "testing" + + "github.com/golang/mock/gomock" + "github.com/stretchr/testify/require" + + "github.com/yandex/mysync/internal/dcs" +) + +func TestSafeAbortSwitchoverDeletesAbortableVersion(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + *switchover = Switchover{Abortable: true, DCSVersion: 7} + return nil + }) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(nil) + + app := newTestApp(t, minConfig(), mockDCS) + require.NoError(t, app.safeAbortSwitchover()) +} + +func TestSafeAbortSwitchoverRejectsUnabortableSwitch(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + *switchover = Switchover{Abortable: false, DCSVersion: 7} + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + require.ErrorIs(t, app.safeAbortSwitchover(), ErrSwitchoverNotAbortable) +} + +func TestSafeAbortSwitchoverDoesNotDeleteChangedSwitch(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + *switchover = Switchover{Abortable: true, DCSVersion: 7} + return nil + }) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(dcs.ErrVersionMismatch) + + app := newTestApp(t, minConfig(), mockDCS) + require.ErrorIs(t, app.safeAbortSwitchover(), dcs.ErrVersionMismatch) +} + +func TestSafeAbortSwitchoverReturnsNotFound(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).Return(dcs.ErrNotFound) + + app := newTestApp(t, minConfig(), mockDCS) + require.ErrorIs(t, app.safeAbortSwitchover(), dcs.ErrNotFound) +} diff --git a/internal/app/data.go b/internal/app/data.go index f6675fb5..35024ba7 100644 --- a/internal/app/data.go +++ b/internal/app/data.go @@ -17,10 +17,11 @@ const ( ) var ( - ErrNoMaster = errors.New("no alive master found") - ErrManyMasters = errors.New("more than one master found") - ErrNoActiveNodes = errors.New("no active nodes found") - ErrSwitchoverTimeout = errors.New("switchover timed out") + ErrNoMaster = errors.New("no alive master found") + ErrManyMasters = errors.New("more than one master found") + ErrNoActiveNodes = errors.New("no active nodes found") + ErrSwitchoverTimeout = errors.New("switchover timed out") + ErrSwitchoverNotAbortable = errors.New("switchover is not safe to abort") ) const ( diff --git a/internal/app/switchover_timeout_test.go b/internal/app/switchover_timeout_test.go index f5457e73..f9fa8b65 100644 --- a/internal/app/switchover_timeout_test.go +++ b/internal/app/switchover_timeout_test.go @@ -86,6 +86,24 @@ func TestNewSwitchoverAbortDeadline(t *testing.T) { } } +func TestSwitchoverTimeoutDisabledAfterAbortBoundary(t *testing.T) { + now := time.Date(2026, time.August, 27, 12, 30, 0, 0, time.UTC) + cfg := minConfig() + cfg.SwitchoverTimeout = 10 * time.Minute + app := newTestApp(t, cfg, nil) + switchover := &Switchover{ + InitiatedAt: now.Add(-30 * time.Minute), + StartedAt: now.Add(-20 * time.Minute), + RunCount: 1, + Abortable: false, + } + + deadline := app.newSwitchoverAbortDeadline(switchover) + require.Nil(t, deadline) + require.NoError(t, deadline.exceeded(now)) + require.False(t, switchover.Abortable) +} + func TestMarkSwitchoverUnabortablePersistsBoundary(t *testing.T) { for _, abortable := range []bool{true, false} { t.Run(fmt.Sprintf("abortable_%t", abortable), func(t *testing.T) { diff --git a/tests/features/9.7/cascade_replicas.feature b/tests/features/9.7/cascade_replicas.feature index bcc6d5d3..f5fecfc4 100644 --- a/tests/features/9.7/cascade_replicas.feature +++ b/tests/features/9.7/cascade_replicas.feature @@ -194,7 +194,7 @@ Feature: cascade replicas """ ["mysql1","mysql2"] """ - And mysql host "mysql3" should be replica of "mysql2" + And mysql host "mysql3" should become replica of "mysql2" within "45" seconds When I run SQL on mysql host "mysql2" """ STOP REPLICA; CHANGE REPLICATION SOURCE TO SOURCE_DELAY = 10 FOR CHANNEL ''; START REPLICA diff --git a/tests/features/cascade_replicas.84.feature b/tests/features/cascade_replicas.84.feature index 15f79ea4..0a4e2e7c 100644 --- a/tests/features/cascade_replicas.84.feature +++ b/tests/features/cascade_replicas.84.feature @@ -194,7 +194,7 @@ Feature: cascade replicas """ ["mysql1","mysql2"] """ - And mysql host "mysql3" should be replica of "mysql2" + And mysql host "mysql3" should become replica of "mysql2" within "45" seconds When I run SQL on mysql host "mysql2" """ STOP REPLICA; CHANGE REPLICATION SOURCE TO SOURCE_DELAY = 10 FOR CHANNEL ''; START REPLICA diff --git a/tests/features/cascade_replicas.feature b/tests/features/cascade_replicas.feature index 120e0adb..ac9cad5e 100644 --- a/tests/features/cascade_replicas.feature +++ b/tests/features/cascade_replicas.feature @@ -194,7 +194,7 @@ Feature: cascade replicas """ ["mysql1","mysql2"] """ - And mysql host "mysql3" should be replica of "mysql2" + And mysql host "mysql3" should become replica of "mysql2" within "45" seconds When I run SQL on mysql host "mysql2" """ STOP SLAVE; CHANGE MASTER TO MASTER_DELAY = 10 FOR CHANNEL ''; START SLAVE diff --git a/tests/features/switchover_to.feature b/tests/features/switchover_to.feature index 4a8cffdb..d772b40b 100644 --- a/tests/features/switchover_to.feature +++ b/tests/features/switchover_to.feature @@ -389,9 +389,11 @@ Feature: manual switchover to new master "started_at": "REGEXP:^20[0-9]{2}-" } """ - When I run command on host "{{.manager.hostname}}" + # SIGKILL is intentional: a graceful restart can let the old manager finish + # the timed-out switchover before supervisor starts the replacement process. + When I run command on host "{{.manager.hostname}}" with timeout "30" seconds """ - supervisorctl restart mysync + supervisorctl signal KILL mysync """ Then command return code should be "0" And zookeeper node "/test/last_rejected_switch" should match json within "30" seconds From ed41f23e2a1dc0e70988f3ca39775df2972f2b5b Mon Sep 17 00:00:00 2001 From: suetin Date: Fri, 28 Aug 2026 05:33:55 +0300 Subject: [PATCH 4/4] cancel switchover on timeout if possible --- README.md | 6 +- cmd/mysync/safe_abort.go | 6 +- internal/app/app.go | 128 +++++++++++++++++------- internal/app/app_dcs.go | 2 +- internal/app/app_dcs_impl.go | 55 ++++++++-- internal/app/cli_switch.go | 24 +++-- internal/app/cli_switch_test.go | 44 +++++--- internal/app/data.go | 37 ++++++- internal/app/idcs.go | 2 +- internal/app/mock_idcs_test.go | 8 +- internal/app/replication.go | 7 +- internal/app/switchover_timeout.go | 58 ++++++++++- internal/app/switchover_timeout_test.go | 88 +++++++++++++++- tests/features/switchover_to.feature | 67 +++++++++++++ 14 files changed, 437 insertions(+), 95 deletions(-) diff --git a/README.md b/README.md index ac52d62b..7bf060c6 100644 --- a/README.md +++ b/README.md @@ -137,7 +137,7 @@ mysync maint on mysync maint off ``` -`mysync safe-abort` aborts a switchover only while it is marked as safely -abortable. `mysync abort` remains a force operation and may require manual +`mysync safe-abort` requests that the current manager reject a switchover while +it is still marked as safely abortable. The manager records the result and runs +normal cleanup. `mysync abort` remains a force operation and may require manual cluster repair. - diff --git a/cmd/mysync/safe_abort.go b/cmd/mysync/safe_abort.go index b8d238c8..767f443e 100644 --- a/cmd/mysync/safe_abort.go +++ b/cmd/mysync/safe_abort.go @@ -11,9 +11,9 @@ import ( var safeAbortCmd = &cobra.Command{ Use: "safe-abort", - Short: "Abort a switchover only before the safe-abort boundary", - Long: "Atomically removes the current switchover only while it is marked abortable. " + - "It never force-aborts a switchover that has started changing the replication topology.", + Short: "Request abort before the switchover safe-abort boundary", + Long: "Atomically marks the current switchover for abort only while it is still abortable. " + + "The current manager records the rejection and performs normal cleanup.", Run: func(cmd *cobra.Command, args []string) { app, err := app.NewApp(configFile, logLevel, true) if err != nil { diff --git a/internal/app/app.go b/internal/app/app.go index bab7500d..4bb305db 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -469,26 +469,30 @@ func (app *App) stateManager() appState { // check if switchover required or in progress switchover := new(Switchover) if err := app.GetCurrentSwitchover(switchover); err == nil { + abortDeadline := app.newSwitchoverAbortDeadline(switchover) + if abortErr := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), false); abortErr != nil { + app.logger.Error().Err(abortErr).Msgf("switchover %s => %s aborted before it started", switchover.From, switchover.To) + if !app.AcquireLock(pathManagerLock) { + app.logger.Error().Msg("manager lock lost before rejecting aborted switchover") + return stateManager + } + err = app.FinishSwitchover(switchover, abortErr) + if err != nil { + app.logger.Error().Err(err).Msg("failed to reject aborted switchover") + } + return stateManager + } // failover via DCS is suppressed during light maintenance (only manual switchover is allowed) if lightMaintenance && switchover.MasterTransition == FailoverTransition { app.logger.Info().Msgf("failover suppressed by light maintenance mode") } else { - abortDeadline := app.newSwitchoverAbortDeadline(switchover) - if timeoutErr := abortDeadline.exceeded(time.Now()); timeoutErr != nil { - app.logger.Error().Err(timeoutErr).Msgf("switchover %s => %s timed out before it started", switchover.From, switchover.To) - if !app.AcquireLock(pathManagerLock) { - app.logger.Error().Msg("manager lock lost before rejecting timed out switchover") - return stateManager - } - err = app.FinishSwitchover(switchover, timeoutErr) - if err != nil { - app.logger.Error().Err(err).Msg("failed to reject timed out switchover") - } - return stateManager - } err = app.approveSwitchover(switchover, activeNodes, clusterState) if err != nil { app.logger.Error().Err(err).Msg("cannot perform switchover") + if !app.AcquireLock(pathManagerLock) { + app.logger.Error().Msg("manager lock lost before rejecting switchover") + return stateManager + } err = app.FinishSwitchover(switchover, err) if err != nil { app.logger.Error().Err(err).Msg("failed to reject switchover") @@ -502,15 +506,30 @@ func (app *App) stateManager() appState { return stateManager } err = app.performSwitchover(clusterState, activeNodes, switchover, master, abortDeadline) - if errors.Is(app.GetCurrentSwitchover(new(Switchover)), dcs.ErrNotFound) { + if !app.AcquireLock(pathManagerLock) { + app.logger.Error().Msg("manager lock lost before reporting switchover result") + return stateManager + } + currentSwitchover := new(Switchover) + currentErr := app.GetCurrentSwitchover(currentSwitchover) + if errors.Is(currentErr, dcs.ErrNotFound) { app.logger.Error().Msgf("switchover was aborted") + } else if currentErr != nil { + app.logger.Error().Err(currentErr).Msg("failed to read switchover before reporting result") + } else if currentSwitchover.OperationID != switchover.OperationID { + app.logger.Error().Msgf("switchover %s was superseded by operation %s", switchover.OperationID, currentSwitchover.OperationID) + } else if currentSwitchover.Abortable && currentSwitchover.AbortRequested { + switchover = currentSwitchover + err = switchoverAbortRequestedError(switchover) + err = app.recordSwitchoverAttemptResult(switchover, err) + if err != nil { + app.logger.Error().Err(err).Msg("failed to report safe abort result") + } + } else if currentSwitchover.DCSVersion != switchover.DCSVersion { + app.logger.Error().Msgf("switchover %s changed before its result could be reported", switchover.OperationID) } else { if errors.Is(err, ErrSwitchoverTimeout) { app.logger.Error().Err(err).Msgf("switchover %s => %s timed out at a safe abort point", switchover.From, switchover.To) - if !app.AcquireLock(pathManagerLock) { - app.logger.Error().Msg("manager lock lost before rejecting timed out switchover") - return stateManager - } } err = app.recordSwitchoverAttemptResult(switchover, err) if err != nil { @@ -1295,6 +1314,9 @@ func (app *App) performSwitchover( return err } } + if err := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), true); err != nil { + return err + } if switchover.MasterTransition != FailoverTransition { app.startTiming(timingDowntime, time.Time{}) @@ -1339,12 +1361,7 @@ func (app *App) performSwitchover( if err, ok := errs[oldMaster]; ok && err != nil && switchover.MasterTransition != FailoverTransition { err = fmt.Errorf("switchover: failed to set old master %s read-only %w", oldMaster, err) app.logger.Info().Msg(err.Error()) - switchErr := app.FinishSwitchover(switchover, err) - if switchErr != nil { - return fmt.Errorf("switchover: failed to reject switchover %w", switchErr) - } - app.logger.Info().Msg("switchover: rejected") - return err + return newTerminalSwitchoverError(err) } app.logger.Info().Msg("switchover: phase 2: stop replication") @@ -1447,19 +1464,42 @@ func (app *App) performSwitchover( if err != nil || app.emulateError("catchup_set_most_recent_online") { return err } - err = app.performChangeMaster(newMaster, mostRecent) - if err != nil || app.emulateError("catchup_change_master") { - return err + if switchoverSourceAlreadyMatches(clusterState, newMaster, mostRecent) { + // Phase 2 stopped the IO thread. If the candidate already streams + // from the most recent host, resume it without changing topology so + // the catch-up remains safe to abort. + if err := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), true); err != nil { + return err + } + err = newMasterNode.StartSlaveIOThread() + if err != nil { + return fmt.Errorf("failed to resume replication IO thread on %s: %w", newMaster, err) + } + } else { + if err := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), true); err != nil { + return err + } + if !app.AcquireLock(pathManagerLock) { + return errors.New("manager lock lost before changing replication source, new manager should finish the process, leaving") + } + if err := app.markSwitchoverUnabortable(switchover); err != nil { + return err + } + abortDeadline = nil + err = app.performChangeMaster(newMaster, mostRecent) + if err != nil || app.emulateError("catchup_change_master") { + return err + } } } else { app.logger.Info().Msgf("switchover: new master %s is the most recent host, waiting for all binlogs to be applied", newMaster) } - caught, err := app.waitForCatchUp(newMasterNode, mostRecentGtidSet, app.config.SlaveCatchUpTimeout, time.Second, abortDeadline) - if errors.Is(err, ErrSwitchoverTimeout) { + caught, err := app.waitForCatchUp(newMasterNode, mostRecentGtidSet, app.config.SlaveCatchUpTimeout, time.Second, abortDeadline, switchover) + if err != nil { return err } - if err != nil || app.emulateError("catchup_master_status") { - return fmt.Errorf("failed to get gtid executed from %s: %w", newMaster, err) + if app.emulateError("catchup_master_status") { + return fmt.Errorf("failed to get gtid executed from %s", newMaster) } if !caught || app.emulateError("catchup_failed") { return fmt.Errorf("new master %s failed to catch up %s within %s", @@ -1479,7 +1519,7 @@ func (app *App) performSwitchover( if dubious := getDubiousHAHosts(clusterState); len(dubious) > 0 { return fmt.Errorf("switchover: failed to ping hosts: %v with dubious errors", dubious) } - if err := abortDeadline.exceeded(time.Now()); err != nil { + if err := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), true); err != nil { return err } if !app.AcquireLock(pathManagerLock) { @@ -2175,6 +2215,15 @@ func (app *App) performChangeMaster(host, master string) error { return nil } +func switchoverSourceAlreadyMatches( + clusterState map[string]*nodestate.NodeState, + host string, + source string, +) bool { + state := clusterState[host] + return state != nil && state.SlaveState != nil && state.SlaveState.MasterHost == source +} + func (app *App) getNodeState(host string) *nodestate.NodeState { var node *mysql.Node if app.cluster.Local().Host() == host { @@ -2350,25 +2399,30 @@ func (app *App) waitForCatchUp( timeout time.Duration, sleep time.Duration, abortDeadline *switchoverAbortDeadline, + switchover *Switchover, ) (bool, error) { deadline := time.Now().Add(timeout) for { - if err := abortDeadline.exceeded(time.Now()); err != nil { + if err := app.checkSwitchoverAbort(switchover, abortDeadline, time.Now(), true); err != nil { return false, err } gtidExecuted, err := node.GTIDExecutedParsed() if err != nil { - return false, err + return false, fmt.Errorf("failed to get gtid executed from %s: %w", node.Host(), err) } app.logger.Info().Msgf("catch up: node %s has gtid %s, waiting for %s", node.Host(), gtidExecuted.String(), gtidset.String()) if gtidExecuted.Contain(gtidset) { return true, nil } - switchover := new(Switchover) - if errors.Is(app.GetCurrentSwitchover(switchover), dcs.ErrNotFound) { + currentSwitchover := new(Switchover) + currentErr := app.GetCurrentSwitchover(currentSwitchover) + if errors.Is(currentErr, dcs.ErrNotFound) { return false, nil } - if app.CheckAsyncSwitchAllowed(node, switchover) { + if currentErr != nil { + return false, currentErr + } + if app.CheckAsyncSwitchAllowed(node, currentSwitchover) { return true, nil } now := time.Now() diff --git a/internal/app/app_dcs.go b/internal/app/app_dcs.go index 6c6acca5..51cfc1f1 100644 --- a/internal/app/app_dcs.go +++ b/internal/app/app_dcs.go @@ -140,7 +140,7 @@ func (app *App) FinishSwitchover(switchover *Switchover, switchErr error) error switchover.Result.Error = switchErr.Error() } - err := app.appDCS.DeleteCurrentSwitchoverVersion(switchover.DCSVersion) + err := app.appDCS.DeleteCurrentSwitchoverVersion(switchover) if err != nil { return err } diff --git a/internal/app/app_dcs_impl.go b/internal/app/app_dcs_impl.go index ad38930e..ecb7bc63 100644 --- a/internal/app/app_dcs_impl.go +++ b/internal/app/app_dcs_impl.go @@ -5,6 +5,8 @@ import ( "fmt" "time" + "github.com/google/uuid" + nodestate "github.com/yandex/mysync/internal/app/node_state" "github.com/yandex/mysync/internal/config" "github.com/yandex/mysync/internal/dcs" @@ -197,15 +199,39 @@ func (a *appDCS) GetLastSwitchover(switchover *Switchover) error { // GetCurrentSwitchover reads the current in-progress switchover from ZK. // Returns dcs.ErrNotFound if no switchover is in progress. func (a *appDCS) GetCurrentSwitchover(switchover *Switchover) error { - version, err := a.dcs.GetVersion(pathCurrentSwitch, switchover) - if err == nil { - switchover.DCSVersion = version + for attempt := 0; attempt < 3; attempt++ { + // json.Unmarshal leaves fields untouched when they are absent in legacy + // payloads, so always decode into a zeroed value. + *switchover = Switchover{} + version, err := a.dcs.GetVersion(pathCurrentSwitch, switchover) + if err != nil { + return err + } + if switchover.OperationID != "" { + switchover.DCSVersion = version + return nil + } + + // Atomically migrate records written by older mysync/worker versions. + // A random ID is persisted before the record is returned, so even legacy + // records without initiated_at get a stable identity for later CASes. + switchover.OperationID = uuid.NewString() + newVersion, err := a.dcs.SetVersion(pathCurrentSwitch, switchover, version) + if errors.Is(err, dcs.ErrVersionMismatch) { + continue + } + if err != nil { + return err + } + switchover.DCSVersion = newVersion + return nil } - return err + return dcs.ErrVersionMismatch } // CreateCurrentSwitchover creates a new switchover record in ZK (fails if one already exists). func (a *appDCS) CreateCurrentSwitchover(switchover *Switchover) error { + switchover.OperationID = uuid.NewString() err := a.dcs.Create(pathCurrentSwitch, switchover) if err == nil { switchover.DCSVersion = 0 @@ -215,6 +241,9 @@ func (a *appDCS) CreateCurrentSwitchover(switchover *Switchover) error { // SetCurrentSwitchover writes the current in-progress switchover to ZK. func (a *appDCS) SetCurrentSwitchover(switchover *Switchover) error { + if err := a.verifyCurrentSwitchover(switchover); err != nil { + return err + } version, err := a.dcs.SetVersion(pathCurrentSwitch, switchover, switchover.DCSVersion) if err == nil { switchover.DCSVersion = version @@ -229,8 +258,22 @@ func (a *appDCS) DeleteCurrentSwitchover() error { // DeleteCurrentSwitchoverVersion removes the current switchover only if it has // not been updated by another manager since this process read it. -func (a *appDCS) DeleteCurrentSwitchoverVersion(version int32) error { - return a.dcs.DeleteVersion(pathCurrentSwitch, version) +func (a *appDCS) DeleteCurrentSwitchoverVersion(switchover *Switchover) error { + if err := a.verifyCurrentSwitchover(switchover); err != nil { + return err + } + return a.dcs.DeleteVersion(pathCurrentSwitch, switchover.DCSVersion) +} + +func (a *appDCS) verifyCurrentSwitchover(expected *Switchover) error { + current := new(Switchover) + if err := a.GetCurrentSwitchover(current); err != nil { + return err + } + if current.OperationID != expected.OperationID || current.DCSVersion != expected.DCSVersion { + return dcs.ErrVersionMismatch + } + return nil } // SetLastSwitchover writes the completed switchover result to ZK. diff --git a/internal/app/cli_switch.go b/internal/app/cli_switch.go index 0b76aa74..9eba2b51 100644 --- a/internal/app/cli_switch.go +++ b/internal/app/cli_switch.go @@ -241,10 +241,10 @@ func (app *App) CliAbort() int { return 0 } -// safeAbortSwitchover removes the current switchover only while it is still -// before the persisted safe-abort boundary. The versioned delete prevents a -// concurrent manager from crossing that boundary between the read and delete. -func (app *App) safeAbortSwitchover() error { +// requestSafeAbort asks the current manager to finish an abortable switchover. +// Keeping /switch until the manager records the terminal result prevents a +// delete/recreate race with a stale manager and lets normal cleanup run. +func (app *App) requestSafeAbort(requestedBy string) error { switchover := new(Switchover) if err := app.GetCurrentSwitchover(switchover); err != nil { return err @@ -252,10 +252,17 @@ func (app *App) safeAbortSwitchover() error { if !switchover.Abortable { return ErrSwitchoverNotAbortable } - return app.appDCS.DeleteCurrentSwitchoverVersion(switchover.DCSVersion) + if switchover.AbortRequested { + return nil + } + now := time.Now() + switchover.AbortRequested = true + switchover.AbortRequestedBy = requestedBy + switchover.AbortRequestedAt = &now + return app.appDCS.SetCurrentSwitchover(switchover) } -// CliSafeAbort safely cleans an abortable switchover node from DCS. +// CliSafeAbort requests manager-owned cleanup of an abortable switchover. func (app *App) CliSafeAbort() int { err := app.connectDCS() if err != nil { @@ -265,7 +272,8 @@ func (app *App) CliSafeAbort() int { defer app.dcs.Close() app.dcs.Initialize() - err = app.safeAbortSwitchover() + requestedBy := util.GuessWhoRunning() + "@" + app.config.Hostname + err = app.requestSafeAbort(requestedBy) switch { case errors.Is(err, dcs.ErrNotFound): fmt.Println("no active switchover") @@ -281,6 +289,6 @@ func (app *App) CliSafeAbort() int { return 1 } - fmt.Println("switchover safely aborted") + fmt.Println("safe abort requested; the current manager will finish cleanup") return 0 } diff --git a/internal/app/cli_switch_test.go b/internal/app/cli_switch_test.go index 801099eb..e57d1933 100644 --- a/internal/app/cli_switch_test.go +++ b/internal/app/cli_switch_test.go @@ -9,22 +9,28 @@ import ( "github.com/yandex/mysync/internal/dcs" ) -func TestSafeAbortSwitchoverDeletesAbortableVersion(t *testing.T) { +func TestRequestSafeAbortPersistsRequest(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockDCS := NewMockIAppDCS(ctrl) mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { - *switchover = Switchover{Abortable: true, DCSVersion: 7} + *switchover = Switchover{OperationID: "op-a", Abortable: true, DCSVersion: 7} + return nil + }) + mockDCS.EXPECT().SetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.Equal(t, "op-a", switchover.OperationID) + require.True(t, switchover.AbortRequested) + require.Equal(t, "operator@test", switchover.AbortRequestedBy) + require.NotNil(t, switchover.AbortRequestedAt) return nil }) - mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(nil) app := newTestApp(t, minConfig(), mockDCS) - require.NoError(t, app.safeAbortSwitchover()) + require.NoError(t, app.requestSafeAbort("operator@test")) } -func TestSafeAbortSwitchoverRejectsUnabortableSwitch(t *testing.T) { +func TestRequestSafeAbortRejectsUnabortableSwitch(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -35,25 +41,25 @@ func TestSafeAbortSwitchoverRejectsUnabortableSwitch(t *testing.T) { }) app := newTestApp(t, minConfig(), mockDCS) - require.ErrorIs(t, app.safeAbortSwitchover(), ErrSwitchoverNotAbortable) + require.ErrorIs(t, app.requestSafeAbort("operator@test"), ErrSwitchoverNotAbortable) } -func TestSafeAbortSwitchoverDoesNotDeleteChangedSwitch(t *testing.T) { +func TestRequestSafeAbortDoesNotOverwriteChangedSwitch(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockDCS := NewMockIAppDCS(ctrl) mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { - *switchover = Switchover{Abortable: true, DCSVersion: 7} + *switchover = Switchover{OperationID: "op-a", Abortable: true, DCSVersion: 7} return nil }) - mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(dcs.ErrVersionMismatch) + mockDCS.EXPECT().SetCurrentSwitchover(gomock.Any()).Return(dcs.ErrVersionMismatch) app := newTestApp(t, minConfig(), mockDCS) - require.ErrorIs(t, app.safeAbortSwitchover(), dcs.ErrVersionMismatch) + require.ErrorIs(t, app.requestSafeAbort("operator@test"), dcs.ErrVersionMismatch) } -func TestSafeAbortSwitchoverReturnsNotFound(t *testing.T) { +func TestRequestSafeAbortReturnsNotFound(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -61,5 +67,19 @@ func TestSafeAbortSwitchoverReturnsNotFound(t *testing.T) { mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).Return(dcs.ErrNotFound) app := newTestApp(t, minConfig(), mockDCS) - require.ErrorIs(t, app.safeAbortSwitchover(), dcs.ErrNotFound) + require.ErrorIs(t, app.requestSafeAbort("operator@test"), dcs.ErrNotFound) +} + +func TestRequestSafeAbortIsIdempotent(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + *switchover = Switchover{OperationID: "op-a", Abortable: true, AbortRequested: true, DCSVersion: 8} + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + require.NoError(t, app.requestSafeAbort("operator@test")) } diff --git a/internal/app/data.go b/internal/app/data.go index 35024ba7..48302eba 100644 --- a/internal/app/data.go +++ b/internal/app/data.go @@ -17,11 +17,13 @@ const ( ) var ( - ErrNoMaster = errors.New("no alive master found") - ErrManyMasters = errors.New("more than one master found") - ErrNoActiveNodes = errors.New("no active nodes found") - ErrSwitchoverTimeout = errors.New("switchover timed out") - ErrSwitchoverNotAbortable = errors.New("switchover is not safe to abort") + ErrNoMaster = errors.New("no alive master found") + ErrManyMasters = errors.New("more than one master found") + ErrNoActiveNodes = errors.New("no active nodes found") + ErrSwitchoverTimeout = errors.New("switchover timed out") + ErrSwitchoverNotAbortable = errors.New("switchover is not safe to abort") + ErrSwitchoverAbortRequested = errors.New("switchover safe abort requested") + ErrSwitchoverTerminal = errors.New("switchover failure is terminal") ) const ( @@ -52,7 +54,12 @@ type Switchover struct { StartedAt time.Time `json:"started_at"` Result *SwitchoverResult `json:"result"` RunCount int `json:"run_count,omitempty"` + OperationID string `json:"operation_id,omitempty"` Abortable bool `json:"abortable,omitempty"` + TopologyChanged bool `json:"topology_changed,omitempty"` + AbortRequested bool `json:"abort_requested,omitempty"` + AbortRequestedBy string `json:"abort_requested_by,omitempty"` + AbortRequestedAt *time.Time `json:"abort_requested_at,omitempty"` DCSVersion int32 `json:"-"` } @@ -87,6 +94,26 @@ type SwitchoverResult struct { FinishedAt time.Time `json:"finished_at"` } +type terminalSwitchoverError struct { + err error +} + +func (err terminalSwitchoverError) Error() string { + return err.err.Error() +} + +func (err terminalSwitchoverError) Unwrap() error { + return err.err +} + +func (err terminalSwitchoverError) Is(target error) bool { + return target == ErrSwitchoverTerminal || errors.Is(err.err, target) +} + +func newTerminalSwitchoverError(err error) error { + return terminalSwitchoverError{err: err} +} + // Maintenance struct presence means that cluster under manual control // Light mode allows everything except failover and switchover type MaintenanceMode string diff --git a/internal/app/idcs.go b/internal/app/idcs.go index b55b908d..75eb14b1 100644 --- a/internal/app/idcs.go +++ b/internal/app/idcs.go @@ -48,7 +48,7 @@ type IAppDCS interface { GetLastSwitchover(switchover *Switchover) error SetCurrentSwitchover(switchover *Switchover) error DeleteCurrentSwitchover() error - DeleteCurrentSwitchoverVersion(version int32) error + DeleteCurrentSwitchoverVersion(switchover *Switchover) error SetLastSwitchover(switchover *Switchover) error SetLastRejectedSwitchover(switchover *Switchover) error GetLastRejectedSwitchover(switchover *Switchover) error diff --git a/internal/app/mock_idcs_test.go b/internal/app/mock_idcs_test.go index ba34e03c..7144dd82 100644 --- a/internal/app/mock_idcs_test.go +++ b/internal/app/mock_idcs_test.go @@ -93,17 +93,17 @@ func (mr *MockIAppDCSMockRecorder) DeleteCurrentSwitchover() *gomock.Call { } // DeleteCurrentSwitchoverVersion mocks base method. -func (m *MockIAppDCS) DeleteCurrentSwitchoverVersion(version int32) error { +func (m *MockIAppDCS) DeleteCurrentSwitchoverVersion(switchover *Switchover) error { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "DeleteCurrentSwitchoverVersion", version) + ret := m.ctrl.Call(m, "DeleteCurrentSwitchoverVersion", switchover) ret0, _ := ret[0].(error) return ret0 } // DeleteCurrentSwitchoverVersion indicates an expected call of DeleteCurrentSwitchoverVersion. -func (mr *MockIAppDCSMockRecorder) DeleteCurrentSwitchoverVersion(version interface{}) *gomock.Call { +func (mr *MockIAppDCSMockRecorder) DeleteCurrentSwitchoverVersion(switchover interface{}) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteCurrentSwitchoverVersion", reflect.TypeOf((*MockIAppDCS)(nil).DeleteCurrentSwitchoverVersion), version) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteCurrentSwitchoverVersion", reflect.TypeOf((*MockIAppDCS)(nil).DeleteCurrentSwitchoverVersion), switchover) } // DeleteMaintenance mocks base method. diff --git a/internal/app/replication.go b/internal/app/replication.go index e6af991f..532be98d 100644 --- a/internal/app/replication.go +++ b/internal/app/replication.go @@ -438,12 +438,7 @@ func (app *App) optimizationPhase( ) if err != nil && errors.Is(err, ErrOptimizationPhaseDeadlineExceeded) { app.logger.Info().Msgf("switchover: phase 0: turbo mode failed: %v", err) - switchErr := app.FinishSwitchover(switchover, fmt.Errorf("turbo mode exceeded deadline")) - if switchErr != nil { - return fmt.Errorf("switchover: failed to reject switchover %w", switchErr) - } - app.logger.Info().Msg("switchover: rejected") - return err + return newTerminalSwitchoverError(fmt.Errorf("turbo mode exceeded deadline")) } // Conceptually, we should only reject the switchover if we encounter a DeadlineExceeded error. diff --git a/internal/app/switchover_timeout.go b/internal/app/switchover_timeout.go index 1fe8ef3c..d1a93aea 100644 --- a/internal/app/switchover_timeout.go +++ b/internal/app/switchover_timeout.go @@ -4,11 +4,13 @@ import ( "errors" "fmt" "time" + + "github.com/yandex/mysync/internal/dcs" ) // switchoverAbortDeadline exists while the persisted switchover is still at a // safe abort point. Abortable remains true across retries and manager restarts -// and is cleared in DCS before phase 5 changes the replication topology. +// and is cleared in DCS before the first replication-topology change. type switchoverAbortDeadline struct { at time.Time timeout time.Duration @@ -32,6 +34,7 @@ func (app *App) newSwitchoverAbortDeadline(switchover *Switchover) *switchoverAb func (app *App) markSwitchoverUnabortable(switchover *Switchover) error { updated := *switchover updated.Abortable = false + updated.TopologyChanged = true if err := app.appDCS.SetCurrentSwitchover(&updated); err != nil { return fmt.Errorf("failed to persist switchover safe-abort boundary: %w", err) } @@ -46,10 +49,57 @@ func (deadline *switchoverAbortDeadline) exceeded(now time.Time) error { return fmt.Errorf("%w after %s", ErrSwitchoverTimeout, deadline.timeout) } -// recordSwitchoverAttemptResult retries regular errors, but a timeout observed -// at a safe abort point is terminal and moves the switch to last_rejected. +func switchoverAbortRequestedError(switchover *Switchover) error { + if switchover.AbortRequestedBy == "" { + return ErrSwitchoverAbortRequested + } + return fmt.Errorf("%w by %s", ErrSwitchoverAbortRequested, switchover.AbortRequestedBy) +} + +func (app *App) checkSwitchoverAbort( + switchover *Switchover, + deadline *switchoverAbortDeadline, + now time.Time, + refresh bool, +) error { + if err := deadline.exceeded(now); err != nil { + return err + } + if switchover != nil && switchover.Abortable && switchover.AbortRequested { + return switchoverAbortRequestedError(switchover) + } + if switchover != nil && !switchover.Abortable { + return nil + } + if !refresh || switchover == nil { + return nil + } + + current := new(Switchover) + if err := app.GetCurrentSwitchover(current); err != nil { + return err + } + if current.OperationID != switchover.OperationID { + return dcs.ErrVersionMismatch + } + if current.DCSVersion == switchover.DCSVersion { + return nil + } + if !current.Abortable || !current.AbortRequested { + return dcs.ErrVersionMismatch + } + + switchover.AbortRequested = current.AbortRequested + switchover.AbortRequestedBy = current.AbortRequestedBy + switchover.AbortRequestedAt = current.AbortRequestedAt + switchover.DCSVersion = current.DCSVersion + return switchoverAbortRequestedError(switchover) +} + +// recordSwitchoverAttemptResult retries regular errors. Safe timeouts, explicit +// abort requests, and failures classified as terminal move to last_rejected. func (app *App) recordSwitchoverAttemptResult(switchover *Switchover, switchErr error) error { - if switchErr == nil || errors.Is(switchErr, ErrSwitchoverTimeout) { + if switchErr == nil || errors.Is(switchErr, ErrSwitchoverTimeout) || errors.Is(switchErr, ErrSwitchoverAbortRequested) || errors.Is(switchErr, ErrSwitchoverTerminal) { return app.FinishSwitchover(switchover, switchErr) } return app.FailSwitchover(switchover, switchErr) diff --git a/internal/app/switchover_timeout_test.go b/internal/app/switchover_timeout_test.go index f9fa8b65..32fd42af 100644 --- a/internal/app/switchover_timeout_test.go +++ b/internal/app/switchover_timeout_test.go @@ -9,6 +9,7 @@ import ( "github.com/golang/mock/gomock" "github.com/stretchr/testify/require" + nodestate "github.com/yandex/mysync/internal/app/node_state" "github.com/yandex/mysync/internal/dcs" ) @@ -113,6 +114,7 @@ func TestMarkSwitchoverUnabortablePersistsBoundary(t *testing.T) { mockDCS := NewMockIAppDCS(ctrl) mockDCS.EXPECT().SetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { require.False(t, switchover.Abortable) + require.True(t, switchover.TopologyChanged) require.Equal(t, int32(7), switchover.DCSVersion) switchover.DCSVersion = 8 return nil @@ -122,6 +124,7 @@ func TestMarkSwitchoverUnabortablePersistsBoundary(t *testing.T) { switchover := &Switchover{Abortable: abortable, DCSVersion: 7} require.NoError(t, app.markSwitchoverUnabortable(switchover)) require.False(t, switchover.Abortable) + require.True(t, switchover.TopologyChanged) require.Equal(t, int32(8), switchover.DCSVersion) }) } @@ -134,17 +137,58 @@ func TestWaitForCatchUpHonorsSwitchoverDeadline(t *testing.T) { timeout: 10 * time.Minute, } - caught, err := app.waitForCatchUp(nil, nil, time.Hour, time.Hour, deadline) + caught, err := app.waitForCatchUp(nil, nil, time.Hour, time.Hour, deadline, nil) require.False(t, caught) require.ErrorIs(t, err, ErrSwitchoverTimeout) } +func TestSwitchoverSourceAlreadyMatches(t *testing.T) { + clusterState := map[string]*nodestate.NodeState{ + "replica": {SlaveState: &nodestate.SlaveState{MasterHost: "source"}}, + "master": {}, + } + + require.True(t, switchoverSourceAlreadyMatches(clusterState, "replica", "source")) + require.False(t, switchoverSourceAlreadyMatches(clusterState, "replica", "other")) + require.False(t, switchoverSourceAlreadyMatches(clusterState, "master", "source")) + require.False(t, switchoverSourceAlreadyMatches(clusterState, "missing", "source")) +} + +func TestCheckSwitchoverAbortRefreshesPersistedRequest(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().GetCurrentSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + *switchover = Switchover{ + OperationID: "op-a", + Abortable: true, + AbortRequested: true, + AbortRequestedBy: "operator@test", + DCSVersion: 8, + } + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + switchover := &Switchover{OperationID: "op-a", Abortable: true, DCSVersion: 7} + err := app.checkSwitchoverAbort(switchover, nil, time.Now(), true) + require.ErrorIs(t, err, ErrSwitchoverAbortRequested) + require.EqualError(t, err, "switchover safe abort requested by operator@test") + require.True(t, switchover.AbortRequested) + require.Equal(t, int32(8), switchover.DCSVersion) +} + func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockDCS := NewMockIAppDCS(ctrl) - mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(nil) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.Equal(t, "op-a", switchover.OperationID) + require.Equal(t, int32(7), switchover.DCSVersion) + return nil + }) mockDCS.EXPECT().SetLastRejectedSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { require.Equal(t, 0, switchover.RunCount) require.NotNil(t, switchover.Result) @@ -154,7 +198,7 @@ func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { }) app := newTestApp(t, minConfig(), mockDCS) - switchover := &Switchover{MasterTransition: FailoverTransition, DCSVersion: 7} + switchover := &Switchover{MasterTransition: FailoverTransition, OperationID: "op-a", DCSVersion: 7} err := app.recordSwitchoverAttemptResult( switchover, fmt.Errorf("%w after %s", ErrSwitchoverTimeout, 10*time.Minute), @@ -162,16 +206,50 @@ func TestRecordSwitchoverAttemptResultTimeoutIsTerminal(t *testing.T) { require.NoError(t, err) } +func TestRecordSwitchoverAttemptResultSafeAbortIsTerminalAndAudited(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + mockDCS := NewMockIAppDCS(ctrl) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(gomock.Any()).Return(nil) + mockDCS.EXPECT().SetLastRejectedSwitchover(gomock.Any()).DoAndReturn(func(switchover *Switchover) error { + require.True(t, switchover.AbortRequested) + require.Equal(t, "operator@test", switchover.AbortRequestedBy) + require.False(t, switchover.Result.Ok) + require.Equal(t, "switchover safe abort requested by operator@test", switchover.Result.Error) + return nil + }) + + app := newTestApp(t, minConfig(), mockDCS) + switchover := &Switchover{ + OperationID: "op-a", + MasterTransition: FailoverTransition, + Abortable: true, + AbortRequested: true, + AbortRequestedBy: "operator@test", + DCSVersion: 8, + } + err := app.recordSwitchoverAttemptResult(switchover, switchoverAbortRequestedError(switchover)) + require.NoError(t, err) +} + +func TestTerminalSwitchoverErrorPreservesOriginalMessage(t *testing.T) { + err := newTerminalSwitchoverError(errors.New("cannot freeze old master")) + + require.ErrorIs(t, err, ErrSwitchoverTerminal) + require.EqualError(t, err, "cannot freeze old master") +} + func TestRecordSwitchoverAttemptResultDoesNotDeleteNewManagerState(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockDCS := NewMockIAppDCS(ctrl) - mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(int32(7)).Return(dcs.ErrVersionMismatch) + mockDCS.EXPECT().DeleteCurrentSwitchoverVersion(gomock.Any()).Return(dcs.ErrVersionMismatch) app := newTestApp(t, minConfig(), mockDCS) err := app.recordSwitchoverAttemptResult( - &Switchover{MasterTransition: FailoverTransition, DCSVersion: 7}, + &Switchover{MasterTransition: FailoverTransition, OperationID: "op-a", DCSVersion: 7}, fmt.Errorf("%w after %s", ErrSwitchoverTimeout, 10*time.Minute), ) require.ErrorIs(t, err, dcs.ErrVersionMismatch) diff --git a/tests/features/switchover_to.feature b/tests/features/switchover_to.feature index d772b40b..b4f95b63 100644 --- a/tests/features/switchover_to.feature +++ b/tests/features/switchover_to.feature @@ -40,6 +40,7 @@ Feature: manual switchover to new master "from": "", "to": "mysql2", "master_transition": "switchover", + "topology_changed": true, "result": { "ok": true } @@ -416,3 +417,69 @@ Feature: manual switchover to new master And mysql replication on host "mysql3" should run fine within "30" seconds When I set replication delay on host "mysql2" to "0" seconds Then mysql replication on host "mysql2" should run fine within "30" seconds + + Scenario: safe abort is handled and audited by the current manager + Given cluster environment is + """ + MYSYNC_SEMISYNC=false + MYSYNC_SWITCHOVER_TIMEOUT=2m + OFFLINE_MODE_ENABLE_LAG=300s + """ + And cluster is up and running + Then mysql host "mysql1" should be master + And zookeeper node "/test/active_nodes" should match json_exactly within "30" seconds + """ + ["mysql1","mysql2","mysql3"] + """ + When I set replication delay on host "mysql2" to "60" seconds + And I run SQL on mysql host "mysql1" + """ + CREATE TABLE IF NOT EXISTS mysql.safe_abort_test (id INT PRIMARY KEY) + """ + And I run SQL on mysql host "mysql1" + """ + INSERT INTO mysql.safe_abort_test VALUES (1) + """ + And I run command on host "mysql1" + """ + mysync switch --to mysql2 --wait=0s + """ + Then command return code should be "0" + And zookeeper node "/test/switch" should match json within "10" seconds + """ + { + "to": "mysql2", + "abortable": true, + "started_at": "REGEXP:^20[0-9]{2}-" + } + """ + When I run command on host "mysql1" + """ + mysync safe-abort + """ + Then command return code should be "0" + And command output should match regexp + """ + safe abort requested; the current manager will finish cleanup + """ + And zookeeper node "/test/last_rejected_switch" should match json within "30" seconds + """ + { + "to": "mysql2", + "abortable": true, + "abort_requested": true, + "result": { + "ok": false, + "error": "REGEXP:switchover safe abort requested by .*@mysql1" + } + } + """ + And zookeeper node "/test/switch" should not exist + And mysql host "mysql1" should be master + And mysql host "mysql1" should become writable within "30" seconds + And mysql host "mysql2" should become replica of "mysql1" within "30" seconds + And mysql host "mysql3" should become replica of "mysql1" within "30" seconds + And mysql replication on host "mysql2" should run fine within "30" seconds + And mysql replication on host "mysql3" should run fine within "30" seconds + When I set replication delay on host "mysql2" to "0" seconds + Then mysql replication on host "mysql2" should run fine within "30" seconds