From 42d5f1fda6d5ed322eca277fbebfd9414713492a Mon Sep 17 00:00:00 2001 From: Victor Lyuboslavsky <2685025+getvictor@users.noreply.github.com> Date: Tue, 17 Feb 2026 07:10:52 -0600 Subject: [PATCH] Improve error handling on AWS DB failover (#39841) **Related issue:** Resolves #39228 Manually tested by triggering a failover on loadtest. # Checklist for submitter If some of the following don't apply, delete the relevant line. - [x] Changes file added for user-visible changes in `changes/`, `orbit/changes/` or `ee/fleetd-chrome/changes`. ## Testing - [x] Added/updated automated tests - [x] QA'd all new/changed functionality manually ## Summary by CodeRabbit * **Bug Fixes** * Health checks now detect a primary DB becoming read-only and report failure so the service restarts and reconnects to a writable primary. * Write failures due to DB read-only state now trigger immediate fatal handling to prompt graceful shutdown and recovery. * Improved detection and handling of read-only DB conditions to increase stability during failovers. --- changes/39228-db-failover | 1 + cmd/fleet/serve.go | 15 ++- server/datastore/mysql/mysql.go | 13 ++- server/datastore/mysql/mysql_test.go | 19 ++++ server/datastore/mysql/sessions.go | 5 + server/platform/mysql/common.go | 9 +- server/platform/mysql/errors.go | 29 ++++++ server/platform/mysql/errors_test.go | 52 ++++++++++ server/platform/mysql/retry.go | 48 +++++++++ server/platform/mysql/retry_test.go | 143 +++++++++++++++++++++++++++ 10 files changed, 331 insertions(+), 3 deletions(-) create mode 100644 changes/39228-db-failover create mode 100644 server/platform/mysql/errors_test.go create mode 100644 server/platform/mysql/retry_test.go diff --git a/changes/39228-db-failover b/changes/39228-db-failover new file mode 100644 index 0000000000..41a7ccba50 --- /dev/null +++ b/changes/39228-db-failover @@ -0,0 +1 @@ +Improve error handling on AWS DB failover. Fleet will now fail health check if the primary DB is read-only, or trigger graceful shutdown when write operations encounter read-only errors. \ No newline at end of file diff --git a/cmd/fleet/serve.go b/cmd/fleet/serve.go index 3315f059e7..e8a8d7a02b 100644 --- a/cmd/fleet/serve.go +++ b/cmd/fleet/serve.go @@ -823,6 +823,16 @@ the way that the Fleet server works. ctx, cancelFunc := context.WithCancel(baseCtx) defer cancelFunc() + // Channel used to trigger graceful shutdown on fatal DB errors (e.g. Aurora failover). + dbFatalCh := make(chan error, 1) + common_mysql.SetFatalErrorHandler(func(err error) { + level.Error(logger).Log("msg", "fatal database error detected, initiating graceful shutdown", "err", err) + select { + case dbFatalCh <- err: + default: + } + }) + var conditionalAccessMicrosoftProxy *conditional_access_microsoft_proxy.Proxy if config.MicrosoftCompliancePartner.IsSet() { var err error @@ -1718,7 +1728,10 @@ the way that the Fleet server works. go func() { sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) - <-sig // block on signal + select { + case <-sig: + case <-dbFatalCh: + } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() errs <- func() error { diff --git a/server/datastore/mysql/mysql.go b/server/datastore/mysql/mysql.go index 5d4d94256d..ac6b25832b 100644 --- a/server/datastore/mysql/mysql.go +++ b/server/datastore/mysql/mysql.go @@ -731,9 +731,20 @@ func (ds *Datastore) HealthCheck() error { // NOTE: does not receive a context as argument here, because the HealthCheck // interface potentially affects more than the datastore layer, and I'm not // sure we can safely identify and change them all at this moment. - if _, err := ds.primary.ExecContext(context.Background(), "select 1"); err != nil { + + // Check that the primary is reachable and not in read-only mode. + // After an AWS Aurora failover the old writer is demoted to a reader; + // detecting this lets the health check fail so the orchestrator can restart Fleet. + var readOnly int + if err := ds.primary.QueryRowContext(context.Background(), "SELECT @@read_only").Scan(&readOnly); err != nil { return err } + if readOnly == 1 { + // Intentionally return an error so that the health check endpoint returns a 500, + // signaling the orchestrator (ECS, Kubernetes) to restart Fleet with fresh DB connections. + return errors.New("primary database is read-only, possible failover detected") + } + if ds.readReplicaConfig != nil { var dst int if err := sqlx.GetContext(context.Background(), ds.replica, &dst, "select 1"); err != nil { diff --git a/server/datastore/mysql/mysql_test.go b/server/datastore/mysql/mysql_test.go index 30b76f85a9..0f740de8b8 100644 --- a/server/datastore/mysql/mysql_test.go +++ b/server/datastore/mysql/mysql_test.go @@ -702,6 +702,25 @@ func TestWithTxWillRollbackWhenPanic(t *testing.T) { require.NoError(t, mock.ExpectationsWereMet()) } +func TestHealthCheckDetectsReadOnly(t *testing.T) { + mock, ds := mockDatastore(t) + defer ds.Close() + + // Healthy: primary is writable. + mock.ExpectQuery("SELECT @@read_only"). + WillReturnRows(sqlmock.NewRows([]string{"@@read_only"}).AddRow(0)) + require.NoError(t, ds.HealthCheck()) + + // Unhealthy: primary is read-only (failover scenario). + mock.ExpectQuery("SELECT @@read_only"). + WillReturnRows(sqlmock.NewRows([]string{"@@read_only"}).AddRow(1)) + err := ds.HealthCheck() + require.Error(t, err) + assert.Contains(t, err.Error(), "read-only") + + require.NoError(t, mock.ExpectationsWereMet()) +} + func TestNewReadsPasswordFromDisk(t *testing.T) { passwordFile, err := os.CreateTemp(t.TempDir(), "*.passwordtest") require.NoError(t, err) diff --git a/server/datastore/mysql/sessions.go b/server/datastore/mysql/sessions.go index 1e4f27d4b4..8df29080fa 100644 --- a/server/datastore/mysql/sessions.go +++ b/server/datastore/mysql/sessions.go @@ -4,9 +4,11 @@ import ( "context" "database/sql" "errors" + "github.com/fleetdm/fleet/v4/server" "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" "github.com/fleetdm/fleet/v4/server/fleet" + common_mysql "github.com/fleetdm/fleet/v4/server/platform/mysql" "github.com/jmoiron/sqlx" ) @@ -188,6 +190,9 @@ func (ds *Datastore) MarkSessionAccessed(ctx context.Context, session *fleet.Ses ` results, err := ds.writer(ctx).ExecContext(ctx, sqlStatement, ds.clock.Now(), session.ID) if err != nil { + if common_mysql.IsReadOnlyError(err) { + common_mysql.TriggerFatalError(err) + } return ctxerr.Wrap(ctx, err, "updating mark session as accessed") } rows, err := results.RowsAffected() diff --git a/server/platform/mysql/common.go b/server/platform/mysql/common.go index cb94d49993..e6a8edd8c6 100644 --- a/server/platform/mysql/common.go +++ b/server/platform/mysql/common.go @@ -180,11 +180,18 @@ func WithTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) error if rbErr != nil && rbErr != sql.ErrTxDone { return ctxerr.Wrapf(ctx, err, "got err '%s' rolling back after err", rbErr.Error()) } + if IsReadOnlyError(err) { + TriggerFatalError(err) + } return err } if err := tx.Commit(); err != nil { - return ctxerr.Wrap(ctx, err, "commit transaction") + err = ctxerr.Wrap(ctx, err, "commit transaction") + if IsReadOnlyError(err) { + TriggerFatalError(err) + } + return err } return nil diff --git a/server/platform/mysql/errors.go b/server/platform/mysql/errors.go index e75e00b852..2f6e16466a 100644 --- a/server/platform/mysql/errors.go +++ b/server/platform/mysql/errors.go @@ -2,9 +2,12 @@ package mysql import ( "database/sql" + "errors" "fmt" + "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" platform_http "github.com/fleetdm/fleet/v4/server/platform/http" + "github.com/go-sql-driver/mysql" ) type NotFoundError struct { @@ -66,3 +69,29 @@ func (e *NotFoundError) IsClientError() bool { func (e *NotFoundError) Is(other error) bool { return other == sql.ErrNoRows } + +// MySQL error numbers for read-only conditions. These are not included in the +// VividCortex/mysqlerr package, so we define them here. +const ( + // erReadOnlyTransaction is MySQL error 1792: Cannot execute statement in a READ ONLY transaction. + erReadOnlyTransaction = 1792 + // erOptionPreventsStatement is MySQL error 1290: The MySQL server is running with the --read-only option. + erOptionPreventsStatement = 1290 + // erReadOnlyMode is MySQL error 1836: Running in read-only mode. + erReadOnlyMode = 1836 +) + +// IsReadOnlyError returns true if the error is a MySQL error indicating that +// the server is in read-only mode. This typically happens after an Aurora +// failover when the primary has been demoted to a reader. +func IsReadOnlyError(err error) bool { + err = ctxerr.Cause(err) + var mySQLErr *mysql.MySQLError + if errors.As(err, &mySQLErr) { + switch mySQLErr.Number { + case erReadOnlyTransaction, erOptionPreventsStatement, erReadOnlyMode: + return true + } + } + return false +} diff --git a/server/platform/mysql/errors_test.go b/server/platform/mysql/errors_test.go new file mode 100644 index 0000000000..21cdd03ca8 --- /dev/null +++ b/server/platform/mysql/errors_test.go @@ -0,0 +1,52 @@ +package mysql + +import ( + "fmt" + "testing" + + "github.com/go-sql-driver/mysql" + "github.com/stretchr/testify/assert" +) + +func TestIsReadOnlyError(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + err error + want bool + }{ + { + name: "unrelated MySQL error", + err: &mysql.MySQLError{Number: 1045, Message: "Access denied"}, + want: false, + }, + { + name: "error 1792 read-only transaction", + err: &mysql.MySQLError{Number: 1792, Message: "Cannot execute statement in a READ ONLY transaction."}, + want: true, + }, + { + name: "error 1290 option prevents statement", + err: &mysql.MySQLError{Number: 1290, Message: "The MySQL server is running with the --read-only option"}, + want: true, + }, + { + name: "error 1836 read-only mode", + err: &mysql.MySQLError{Number: 1836, Message: "Running in read-only mode"}, + want: true, + }, + { + name: "wrapped read-only error", + err: fmt.Errorf("transaction failed: %w", &mysql.MySQLError{Number: 1792, Message: "read only"}), + want: true, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.want, IsReadOnlyError(tc.err)) + }) + } +} diff --git a/server/platform/mysql/retry.go b/server/platform/mysql/retry.go index 85534dae44..6e003261d7 100644 --- a/server/platform/mysql/retry.go +++ b/server/platform/mysql/retry.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "errors" + "fmt" + "sync" "time" "github.com/VividCortex/mysqlerr" @@ -14,6 +16,40 @@ import ( "github.com/jmoiron/sqlx" ) +var ( + fatalErrorMu sync.RWMutex + fatalErrorHandler func(error) + fatalErrorOnce sync.Once +) + +// SetFatalErrorHandler registers a function that will be called (at most once) +// when a fatal database error is detected, such as the primary becoming +// read-only during an Aurora failover. The handler should trigger a graceful +// process shutdown. +// +// If no handler is set, the default behavior is to panic. +func SetFatalErrorHandler(fn func(error)) { + fatalErrorMu.Lock() + defer fatalErrorMu.Unlock() + fatalErrorHandler = fn + fatalErrorOnce = sync.Once{} // reset so handler fires on next fatal error +} + +// TriggerFatalError calls the registered fatal error handler exactly once. +// If no handler is registered, it panics (legacy behavior). +func TriggerFatalError(err error) { + fatalErrorMu.RLock() + defer fatalErrorMu.RUnlock() + + if fatalErrorHandler == nil { + panic(fmt.Sprintf("database is read-only, possible failover detected: %v", err)) + } + + fatalErrorOnce.Do(func() { + fatalErrorHandler(err) + }) +} + var DoRetryErr = errors.New("fleet datastore retry") type TxFn func(tx sqlx.ExtContext) error @@ -45,6 +81,13 @@ func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) return backoff.Permanent(ctxerr.Wrapf(ctx, err, "got err '%s' rolling back after err", rbErr.Error())) } + // Read-only errors indicate a DB failover occurred (primary demoted to reader). + // Trigger graceful shutdown so the orchestrator restarts and reconnects to the new primary. + if IsReadOnlyError(err) { + TriggerFatalError(err) + return backoff.Permanent(err) + } + if retryableError(err) { return err } @@ -56,6 +99,11 @@ func WithRetryTxx(ctx context.Context, db *sqlx.DB, fn TxFn, logger log.Logger) if err := tx.Commit(); err != nil { err = ctxerr.Wrap(ctx, err, "commit transaction") + if IsReadOnlyError(err) { + TriggerFatalError(err) + return backoff.Permanent(err) + } + if retryableError(err) { return err } diff --git a/server/platform/mysql/retry_test.go b/server/platform/mysql/retry_test.go new file mode 100644 index 0000000000..d272663872 --- /dev/null +++ b/server/platform/mysql/retry_test.go @@ -0,0 +1,143 @@ +package mysql + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/go-kit/log" + gmysql "github.com/go-sql-driver/mysql" + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// readOnlyErr returns a MySQL error that simulates a read-only database (error 1792). +func readOnlyErr() error { + return &gmysql.MySQLError{Number: 1792, Message: "Cannot execute statement in a READ ONLY transaction."} +} + +func TestTriggerFatalErrorCallsHandler(t *testing.T) { + var called atomic.Bool + var capturedErr atomic.Value + SetFatalErrorHandler(func(err error) { + called.Store(true) + capturedErr.Store(err) + }) + t.Cleanup(func() { SetFatalErrorHandler(nil) }) + + testErr := errors.New("test read-only error") + TriggerFatalError(testErr) + + assert.True(t, called.Load()) + assert.Equal(t, testErr, capturedErr.Load()) +} + +func TestTriggerFatalErrorPanicsWithoutHandler(t *testing.T) { + SetFatalErrorHandler(nil) + + assert.Panics(t, func() { + TriggerFatalError(errors.New("read-only")) + }) +} + +func TestTriggerFatalErrorFiresOnce(t *testing.T) { + var callCount atomic.Int32 + SetFatalErrorHandler(func(_ error) { + callCount.Add(1) + }) + t.Cleanup(func() { SetFatalErrorHandler(nil) }) + + var wg sync.WaitGroup + for range 100 { + wg.Go(func() { + TriggerFatalError(errors.New("read-only")) + }) + } + wg.Wait() + + assert.Equal(t, int32(1), callCount.Load()) +} + +func TestTransactionReadOnlyTriggersFatalError(t *testing.T) { + cases := []struct { + name string + txFunc func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error + setupMock func(mock sqlmock.Sqlmock) + }{ + { + name: "WithRetryTxx read-only from fn", + setupMock: func(mock sqlmock.Sqlmock) { + mock.ExpectBegin() + mock.ExpectRollback() + }, + txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error { + return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error { + return readOnlyErr() + }, log.NewNopLogger()) + }, + }, + { + name: "WithRetryTxx read-only from commit", + setupMock: func(mock sqlmock.Sqlmock) { + mock.ExpectBegin() + mock.ExpectCommit().WillReturnError(readOnlyErr()) + }, + txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error { + return WithRetryTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error { + return nil + }, log.NewNopLogger()) + }, + }, + { + name: "WithTxx read-only from fn", + setupMock: func(mock sqlmock.Sqlmock) { + mock.ExpectBegin() + mock.ExpectRollback() + }, + txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error { + return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error { + return readOnlyErr() + }, log.NewNopLogger()) + }, + }, + { + name: "WithTxx read-only from commit", + setupMock: func(mock sqlmock.Sqlmock) { + mock.ExpectBegin() + mock.ExpectCommit().WillReturnError(readOnlyErr()) + }, + txFunc: func(ctx *testing.T, db *sqlx.DB, mock sqlmock.Sqlmock) error { + return WithTxx(ctx.Context(), db, func(tx sqlx.ExtContext) error { + return nil + }, log.NewNopLogger()) + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + var handlerCalled atomic.Bool + SetFatalErrorHandler(func(_ error) { + handlerCalled.Store(true) + }) + t.Cleanup(func() { SetFatalErrorHandler(nil) }) + + db, mock, err := sqlmock.New() + require.NoError(t, err) + defer db.Close() + sqlxDB := sqlx.NewDb(db, "sqlmock") + + tc.setupMock(mock) + + err = tc.txFunc(t, sqlxDB, mock) + + require.Error(t, err) + assert.True(t, IsReadOnlyError(err)) + assert.True(t, handlerCalled.Load()) + require.NoError(t, mock.ExpectationsWereMet()) + }) + } +}