diff --git a/cmd/rds-agent/config.go b/cmd/rds-agent/config.go index 91d1b23da..bf1c34fa7 100644 --- a/cmd/rds-agent/config.go +++ b/cmd/rds-agent/config.go @@ -57,6 +57,10 @@ type config struct { EngineUser string RCService string EngineService string + // Where the engine records the pid its probe checks for liveness, for an + // engine whose own client cannot tell a server still recovering from one that + // is not running at all. + EnginePidFile string // Where the data volume is mounted, and the two kernel surfaces a storage // grow resolves its device from. Overridable so a test can point them at @@ -86,6 +90,7 @@ func loadConfig(envFile string) config { EngineUser: get("RDS_ENGINE_USER"), RCService: get("RDS_RC_SERVICE"), EngineService: get("RDS_ENGINE_SERVICE"), + EnginePidFile: get("RDS_ENGINE_PIDFILE"), DataMount: get("RDS_DATA_MOUNT"), PollWait: defaultPollWait, } @@ -148,6 +153,9 @@ func (c *config) applyLayout(layout engineLayout) { if c.EngineService == "" { c.EngineService = layout.service } + if c.EnginePidFile == "" { + c.EnginePidFile = layout.pidFile + } if c.DataMount == "" { c.DataMount = layout.dataMount } diff --git a/cmd/rds-agent/config_test.go b/cmd/rds-agent/config_test.go index 53a9a44a9..9a4627b18 100644 --- a/cmd/rds-agent/config_test.go +++ b/cmd/rds-agent/config_test.go @@ -6,6 +6,8 @@ import ( "strings" "testing" "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" ) // Stamps the engine a real image bakes and loads the configuration against it, @@ -74,17 +76,60 @@ func TestNewEngine_RefusesAnImageWithNoEngineStamp(t *testing.T) { } func TestNewEngine_RefusesAnEngineItDoesNotImplement(t *testing.T) { - cfg := testLoadConfig(t, "mariadb") + cfg := testLoadConfig(t, "oracle") _, err := newEngine(cfg, nil, nil, nil) if err == nil { t.Fatal("newEngine built an engine this agent does not implement") } - if !strings.Contains(err.Error(), "mariadb") { + if !strings.Contains(err.Error(), "oracle") { t.Errorf("error = %v, want it to name the engine the image bakes", err) } } +// The implementation is the agent's, but the definition it validates names and +// classifies parameters against is the control plane's. An agent whose control +// plane offers no definition for the engine its image bakes refuses rather than +// inventing one. +func TestNewEngine_FollowsWhetherTheControlPlaneOffersTheBakedEngine(t *testing.T) { + cfg := testLoadConfig(t, engineMariaDB) + probe, err := newProbe(cfg, staticProbe(0)) + if err != nil { + t.Fatalf("newProbe: %v", err) + } + + built, err := newEngine(cfg, nil, nil, probe) + if _, offered := handlers_rds.LookupEngine(engineMariaDB); offered != nil { + if err == nil { + t.Fatal("built a MariaDB engine while the control plane offers no definition for it") + } + if !strings.Contains(err.Error(), engineMariaDB) { + t.Errorf("error = %v, want it to name the engine the image bakes", err) + } + return + } + if err != nil { + t.Fatalf("newEngine: %v", err) + } + if _, ok := built.(*mariadbEngine); !ok { + t.Errorf("newEngine built %T for a MariaDB image", built) + } +} + +// The probe reads the pidfile Alpine's packaged service starts mariadbd with, +// because nothing in the generated configuration can move it: the service passes +// --pid-file on the command line, which beats an option file. +func TestLoadConfig_TakesTheMariaDBPidfileFromTheLayout(t *testing.T) { + cfg := testLoadConfig(t, engineMariaDB) + + if cfg.EnginePidFile != engineLayouts[engineMariaDB].pidFile { + t.Errorf("pidfile = %q, want the MariaDB layout's", cfg.EnginePidFile) + } + if cfg.EnginePidFile == "" { + t.Error("an engine whose probe needs a pidfile was given none") + } +} + func TestLoadConfig_ReadsEnvFile(t *testing.T) { t.Setenv("RDS_ENGINE_FILE", filepath.Join(t.TempDir(), "absent-engine")) path := filepath.Join(t.TempDir(), "agent.env") diff --git a/cmd/rds-agent/engine.go b/cmd/rds-agent/engine.go index 3889accb0..94b69fd4a 100644 --- a/cmd/rds-agent/engine.go +++ b/cmd/rds-agent/engine.go @@ -3,25 +3,21 @@ package main import ( "bytes" "context" - "errors" "fmt" - "os" "os/exec" "os/user" "path/filepath" "strconv" "strings" - "sync" - "sync/atomic" "syscall" "time" handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" ) -// The in-guest half of the engine seam. Everything PostgreSQL-shaped that a -// control-plane command needs lives behind this, so the command registry above -// stays engine-agnostic. +// The in-guest half of the engine seam. Everything engine-shaped that a +// control-plane command needs lives behind this, so the command registry stays +// engine-agnostic. type engineOps interface { // Rotates the master role's password live. Never persisted anywhere in the // guest: it exists only for the length of this call. @@ -49,9 +45,9 @@ type parameterRecovery interface { Restart(ctx context.Context) error } -// Records the parameter file only after the probe observes the postmaster -// serving it. This is separate from apply, where static values are installed -// but not adopted until a later restart. +// Records the parameter file only after the probe observes the engine serving +// it. This is separate from apply, where static values are installed but not +// adopted until a later restart. type servingParameterRecorder interface { RecordServingParameters(ctx context.Context) error } @@ -79,16 +75,24 @@ type engineLayout struct { service string dataMount string port int + // Where the engine records the pid the probe checks for liveness. Empty for + // an engine whose probe answers without one. + pidFile string // The engine's own liveness signal, which is the only part of engineProbe // that is not shared. newProbe func(cfg config, run probeRunner) *engineProbe - // The implementation every control-plane directive is served by. - newEngine func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) engine + // The implementation every control-plane directive is served by. It fails + // when the control plane this agent was built against does not offer the + // engine the image bakes, rather than serving one it has no definition for. + newEngine func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) (engine, error) } -// The name the control plane knows this implementation by, which is also what -// setup.sh stamps into the image. -const enginePostgres = "postgres" +// The names the control plane knows these implementations by, which are also +// what setup.sh stamps into each image. +const ( + enginePostgres = "postgres" + engineMariaDB = "mariadb" +) var engineLayouts = map[string]engineLayout{ enginePostgres: { @@ -100,10 +104,27 @@ var engineLayouts = map[string]engineLayout{ dataMount: "/var/lib/postgresql", port: 5432, newProbe: newPostgresProbe, - newEngine: func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) engine { - return newPostgresEngine(cfg, run, startSess, probe) + newEngine: func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) (engine, error) { + return newPostgresEngine(cfg, run, startSess, probe), nil }, }, + engineMariaDB: { + binDir: "/usr/bin", + dataDir: "/var/lib/mysql/data", + socketDir: "/run/mysqld", + osUser: "mysql", + service: "mariadb", + dataMount: "/var/lib/mysql", + port: 3306, + // Alpine's packaged service starts mysqld_safe with + // --pid-file=/run/mysqld/$RC_SVCNAME.pid, and a command-line option beats + // the configuration file, so this is the path rather than anything the + // platform drop-in could name. setup.sh asserts it against the packaged + // init script at build time. + pidFile: "/run/mysqld/mariadb.pid", + newProbe: newMariaDBProbe, + newEngine: newMariaDBEngineFromCatalog, + }, } // Builds the health probe the image's engine supplies. @@ -116,14 +137,14 @@ func newProbe(cfg config, run probeRunner) (*engineProbe, error) { } // Builds the implementation the image bakes. Nothing here consults the control -// plane: the agent runs the engine it is made of, and the delivered engine is -// only ever checked against it. +// plane for which engine to be: the agent runs the engine it is made of, and the +// delivered engine is only ever checked against it. func newEngine(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) (engine, error) { layout, err := layoutFor(cfg) if err != nil { return nil, err } - return layout.newEngine(cfg, run, startSess, probe), nil + return layout.newEngine(cfg, run, startSess, probe) } func layoutFor(cfg config) (engineLayout, error) { @@ -146,10 +167,14 @@ type command struct { Stdin string // The OS user to drop to before exec. Empty runs as the agent itself. User string + // The statement that makes this client print the session sentinel. Read only + // when the command is started as a held session, whose reader has to know + // when the engine has finished each script it was fed. + SentinelStatement string } -// Returns stdout. stderr is folded into the error, because psql reports the -// actual SQL failure there while exiting with a bare status. +// Returns stdout. stderr is folded into the error, because both clients report +// the actual SQL failure there while exiting with a bare status. type commandRunner func(ctx context.Context, c command) (string, error) func execCommandRunner(ctx context.Context, c command) (string, error) { @@ -194,428 +219,16 @@ func lookupCredential(username string) (*syscall.Credential, error) { return &syscall.Credential{Uid: uint32(uid), Gid: uint32(gid)}, nil } -// The engine is reached over its unix socket under peer authentication, so the -// agent drops to the postgres OS user rather than holding a password of its own. -type postgresEngine struct { - // The control plane's own metadata for this engine, resolved once at - // startup: the live password apply runs as the cluster superuser, so it - // re-checks the role name against the same reserved set the API validates. - meta handlers_rds.Engine - run commandRunner - startSess sessionRunner - psql string - rcService string - service string - pgData string - socketDir string - osUser string - probe *engineProbe - // Set from the bootstrap config, on a different goroutine than the commands - // that read it. - port atomic.Int64 - - // Serializes parameter installs, serving snapshots and rollback restores so - // none can copy or replace an intermediate configuration. - paramMu sync.Mutex - repairTimeout time.Duration - repairPoll time.Duration - - // The backup mode currently held, if any. Guarded because the expiry timer - // releases it on its own goroutine. - mu sync.Mutex - held *quiesceHold -} - -var _ engine = (*postgresEngine)(nil) - -const ( - parameterRepairTimeout = 90 * time.Second - parameterRepairPoll = time.Second -) - -// pg_isready is what gives PostgreSQL three states for free: exit 0 is serving, -// exit 1 is a postmaster up and rejecting connections, and anything else is an -// engine that is not there at all. Resolved on PATH, where the client package -// puts it. -const postgresProbeBinary = "pg_isready" - -func newPostgresProbe(cfg config, run probeRunner) *engineProbe { - return newEngineProbe(cfg.EnginePort, postgresProbeState(cfg.EngineHost, run)) -} - -func postgresProbeState(host string, run probeRunner) probeStateFn { - return func(ctx context.Context, port int64) (engineState, string) { - portArg := strconv.FormatInt(port, 10) - code, err := run(ctx, postgresProbeBinary, "-h", host, "-p", portArg, "-q") - switch { - case err != nil: - // A missing binary or broken image. Reporting healthy on the strength of - // nothing would hide it, so this reads as absent like an engine that did - // not answer. - return engineAbsent, fmt.Sprintf("engine probe could not run: %v", err) - case code == 0: - return engineServing, "" - case code == 1: - return engineRecovering, "engine is rejecting connections (startup or recovery)" - default: - return engineAbsent, fmt.Sprintf("engine did not respond on %s:%s", host, portArg) - } - } -} - -// Resolved once, under the name the control plane knows this implementation by. -// A lookup failure is a mismatch between the agent and the control plane it was -// built against, so it fails at startup rather than at the first rotation. -var postgresEngineMeta = mustLookupEngine("postgres") - -func mustLookupEngine(name string) handlers_rds.Engine { - engine, err := handlers_rds.LookupEngine(name) - if err != nil { - panic("rds-agent: " + err.Error()) - } - return engine -} - -func newPostgresEngine(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) *postgresEngine { - e := &postgresEngine{ - meta: postgresEngineMeta, - run: run, - startSess: startSess, - psql: filepath.Join(cfg.EngineBinDir, "psql"), - rcService: cfg.RCService, - service: cfg.EngineService, - pgData: cfg.EngineDataDir, - socketDir: cfg.SocketDir, - osUser: cfg.EngineUser, - probe: probe, - repairTimeout: parameterRepairTimeout, - repairPoll: parameterRepairPoll, - } - e.port.Store(int64(cfg.EnginePort)) - return e -} - -func (e *postgresEngine) setPort(port int) { - e.port.Store(int64(port)) -} - -// The role name and the password ride the environment and are re-quoted by -// psql, so neither reaches a shell word or an argv another process can read. -func (e *postgresEngine) SetPassword(ctx context.Context, username, password string) error { - if username == "" || password == "" { - return fmt.Errorf("set-password requires both %s and %s", - handlers_rds.CommandParamMasterUsername, handlers_rds.CommandParamMasterUserPassword) - } - // The statement below runs as the cluster superuser over the socket under - // peer auth, so a reserved name in the command payload would hand the - // customer the bootstrap superuser rather than rotate their own role. - if err := e.meta.ValidateUsernameNotReserved(username); err != nil { - return fmt.Errorf("refusing to set the password of a role the engine reserves: %w", err) - } - // psql interpolates the password into the ALTER ROLE before the server sees - // it, and these three are what would write it to the log — the last on any - // failure at its own default. SUSET, so the parameter group cannot win. - const sql = `SET log_statement = 'none'; -SET log_min_duration_statement = -1; -SET log_min_error_statement = 'panic'; -\getenv master RDS_MASTER_USERNAME -\getenv password RDS_MASTER_PASSWORD -ALTER ROLE :"master" WITH LOGIN PASSWORD :'password'; -` - _, err := e.psqlRun(ctx, sql, - "RDS_MASTER_USERNAME="+username, - "RDS_MASTER_PASSWORD="+password, - ) - if err != nil { - // psql echoes the failing statement, which here would carry the new - // password back to the control plane and into the event ring. - return fmt.Errorf("apply the master password: %s", redact(err.Error(), password)) - } - return nil -} - -// The include the resolved parameter set is rendered to, and the copy of the -// last one the engine accepted. Both live beside the data rather than in /etc, -// matching rds-init: a class change boots a fresh root volume, which would -// otherwise revert them when a fresh root volume is used. -const ( - parametersFileName = "10-rds-parameters.conf" - // Deliberately not a .conf name: include_dir globs *.conf, so the rollback - // copy must not be read as a second set of settings. - lastGoodFileName = "10-rds-parameters.last-good" -) - -// Installs the resolved set, validates it with the engine's own config parser -// before the engine ever adopts it, and reloads. A value the engine refuses is -// rolled back here rather than left on the data volume, where it would survive -// every VM replace and turn the next restart into a boot loop. -func (e *postgresEngine) ApplyParameters(ctx context.Context, params []handlers_rds.Parameter) ([]string, error) { - e.paramMu.Lock() - defer e.paramMu.Unlock() - - initialState, _ := e.probe.state(ctx) - if initialState == engineRecovering { - return nil, errors.New("apply parameters while the engine is still starting or recovering") - } - if initialState == engineServing { - // A command can arrive before the first heartbeat. Seed the configuration - // currently being served before replacing its file in that window. - if err := e.recordServingParametersLocked(ctx); err != nil { - return nil, fmt.Errorf("record the parameters serving before the apply: %w", err) - } - } - - // The check has to run against the file in place, because the engine parses - // the datadir's own include_dir. The window that leaves is closed from both - // ends: the rollback below, and the last-known-good restore the agent runs at - // boot when the engine does not come up. - _, restore, err := e.installParameters(params) - if err != nil { - return nil, err - } - if err := e.checkConfig(ctx); err != nil { - return nil, errors.Join(fmt.Errorf("the engine rejected the parameter set: %w", err), restore()) - } - - state, _ := e.probe.state(ctx) - switch state { - case engineAbsent: - return e.restartOnRepairSetLocked(ctx) - case engineRecovering: - return nil, errors.Join(errors.New("the engine entered startup or recovery during the parameter apply"), restore()) - } - - if _, err := e.psqlRun(ctx, "SELECT pg_reload_conf();\n"); err != nil { - // The engine may have gone down after the first probe. In that case start - // it on the new, parser-checked repair set rather than restoring the set - // that already left it unable to serve. - if state, _ := e.probe.state(ctx); state == engineAbsent { - return e.restartOnRepairSetLocked(ctx) - } - return nil, errors.Join(fmt.Errorf("reload the engine configuration: %w", err), restore()) - } - return e.pendingRestartParameters(ctx) -} - -func (e *postgresEngine) restartOnRepairSetLocked(ctx context.Context) ([]string, error) { - repairCtx, cancel := context.WithTimeout(ctx, e.repairTimeout) - defer cancel() - - if err := e.Restart(repairCtx); err != nil { - return nil, fmt.Errorf("start the engine on the repaired parameter set: %w", err) - } - - timer := time.NewTimer(0) - defer timer.Stop() - lastMessage := "engine did not respond" - for { - select { - case <-repairCtx.Done(): - if ctx.Err() != nil { - return nil, fmt.Errorf("wait for the engine on the repaired parameter set: %w", ctx.Err()) - } - return nil, fmt.Errorf("wait for the engine on the repaired parameter set: %s: %w", lastMessage, repairCtx.Err()) - case <-timer.C: - } - - state, message := e.probe.state(repairCtx) - if state == engineServing { - pending, err := e.pendingRestartParameters(repairCtx) - if err == nil { - return pending, nil - } - lastMessage = err.Error() - } else if message != "" { - lastMessage = message - } - timer.Reset(e.repairPoll) - } -} - -func (e *postgresEngine) pendingRestartParameters(ctx context.Context) ([]string, error) { - out, err := e.psqlRun(ctx, "SELECT name FROM pg_settings WHERE pending_restart ORDER BY name;\n") - if err != nil { - return nil, fmt.Errorf("read the settings pending a restart: %w", err) - } - var pending []string - for line := range strings.SplitSeq(out, "\n") { - if name := strings.TrimSpace(line); name != "" { - pending = append(pending, name) - } - } - return pending, nil -} - -func (e *postgresEngine) parametersPath() string { - return filepath.Join(e.pgData, "conf.d", parametersFileName) -} - -func (e *postgresEngine) lastGoodPath() string { - return filepath.Join(e.pgData, "conf.d", lastGoodFileName) -} - -// Written as root and handed to the engine's user, at the same path and mode -// rds-init installs it at, so a later boot overwrites rather than shadows it. -// Returns the content that was there and a func putting it back, so the caller -// can undo the install without re-rendering anything. -func (e *postgresEngine) installParameters(params []handlers_rds.Parameter) (previous []byte, restore func() error, err error) { - path := e.parametersPath() - // A missing file is the first-ever apply, and its rollback is a removal. - previous, readErr := os.ReadFile(path) - if readErr != nil && !os.IsNotExist(readErr) { - return nil, nil, fmt.Errorf("read the installed parameters at %s: %w", path, readErr) - } - existed := readErr == nil - - if err := e.writeEngineFile(path, []byte(renderParameters(params))); err != nil { - return nil, nil, err - } - return previous, func() error { - if !existed { - if err := os.Remove(path); err != nil && !os.IsNotExist(err) { - return fmt.Errorf("withdraw the rejected parameters at %s: %w", path, err) - } - return nil - } - return e.writeEngineFile(path, previous) - }, nil -} - -// Snapshots the include only when the running postmaster has no settings still -// pending a restart. The parameter mutex keeps an apply from replacing the file -// between that check and the copy. -func (e *postgresEngine) RecordServingParameters(ctx context.Context) error { - e.paramMu.Lock() - defer e.paramMu.Unlock() - return e.recordServingParametersLocked(ctx) -} - -func (e *postgresEngine) recordServingParametersLocked(ctx context.Context) error { - if _, err := os.Stat(e.parametersPath()); err != nil { - if os.IsNotExist(err) { - return nil - } - return fmt.Errorf("inspect the serving parameters: %w", err) - } - pending, err := e.pendingRestartParameters(ctx) - if err != nil { - return err - } - if len(pending) > 0 { - return nil - } - - content, err := os.ReadFile(e.parametersPath()) - if err != nil { - return fmt.Errorf("read the serving parameters: %w", err) - } - lastGood, err := os.ReadFile(e.lastGoodPath()) - if err == nil && bytes.Equal(content, lastGood) { - return nil - } - if err != nil && !os.IsNotExist(err) { - return fmt.Errorf("read the last known good parameters: %w", err) - } - return e.writeEngineFile(e.lastGoodPath(), content) -} - -// Puts the last set the engine accepted back in place, for a restart that failed -// after a parameter change. The probe is checked again under the parameter lock -// so a repair that just brought the engine back cannot be reversed. -func (e *postgresEngine) RestoreLastKnownGoodParameters(ctx context.Context) (bool, error) { - e.paramMu.Lock() - defer e.paramMu.Unlock() - - if state, _ := e.probe.state(ctx); state != engineAbsent { - return false, nil - } - lastGood, err := os.ReadFile(e.lastGoodPath()) - if err != nil { - if os.IsNotExist(err) { - return false, nil - } - return false, fmt.Errorf("read the last known good parameters: %w", err) - } - current, err := os.ReadFile(e.parametersPath()) - if err != nil && !os.IsNotExist(err) { - return false, fmt.Errorf("read the installed parameters: %w", err) - } - if bytes.Equal(current, lastGood) { - return false, nil - } - if err := e.writeEngineFile(e.parametersPath(), lastGood); err != nil { - return false, err - } - return true, nil -} - -// Atomic: a temp file in the same directory, renamed over the target. The temp -// name deliberately does not end in .conf, so a crash between write and rename -// cannot leave the engine reading a half-written include. -func (e *postgresEngine) writeEngineFile(path string, content []byte) error { - tmp := path + ".new" - if err := os.WriteFile(tmp, content, 0o600); err != nil { - return fmt.Errorf("write %s: %w", tmp, err) - } - if err := e.chownToEngine(tmp); err != nil { - os.Remove(tmp) - return err - } - if err := os.Rename(tmp, path); err != nil { - os.Remove(tmp) - return fmt.Errorf("install %s: %w", path, err) - } - return nil -} - -// The engine's own parser, run offline against the datadir. Reading one setting -// back is enough: postgres parses postgresql.conf and every include first, and -// exits non-zero naming the file and line of an unknown parameter or a value -// outside its range. -func (e *postgresEngine) checkConfig(ctx context.Context) error { - if _, err := e.run(ctx, command{ - Name: filepath.Join(filepath.Dir(e.psql), "postgres"), - Args: []string{"-D", e.pgData, "-C", "shared_buffers"}, - Env: []string{"PATH=" + defaultGuestPath}, - User: e.osUser, - }); err != nil { - return err - } - return nil -} - -// The engine reads its own config, so a root-owned file would be unreadable to -// it. A guest with no such user is a broken image, not something to work around. -func (e *postgresEngine) chownToEngine(path string) error { - credential, err := lookupCredential(e.osUser) - if err != nil { - return err - } - if err := os.Chown(path, int(credential.Uid), int(credential.Gid)); err != nil { - return fmt.Errorf("hand %s to %s: %w", path, e.osUser, err) - } - return nil -} - -// Through the service manager rather than pg_ctl, so the supervisor records the -// engine as stopped and does not restart it underneath a VM that is going down. -func (e *postgresEngine) Stop(ctx context.Context) error { - return e.serviceAction(ctx, "stop") -} - -// Only the parameter rollback calls this: a restart the control plane wants goes -// through RebootDBInstance, which cycles the VM. -func (e *postgresEngine) Restart(ctx context.Context) error { - return e.serviceAction(ctx, "restart") -} - -func (e *postgresEngine) serviceAction(ctx context.Context, action string) error { - if _, err := e.run(ctx, command{ - Name: e.rcService, - Args: []string{e.service, action}, +// Through the service manager rather than the engine's own control tool, so the +// supervisor records the state and does not restart the engine underneath a VM +// that is going down. +func serviceAction(ctx context.Context, run commandRunner, rcService, service, action string) error { + if _, err := run(ctx, command{ + Name: rcService, + Args: []string{service, action}, Env: []string{"PATH=" + defaultGuestPath}, }); err != nil { - return fmt.Errorf("%s the %s service: %w", action, e.service, err) + return fmt.Errorf("%s the %s service: %w", action, service, err) } return nil } @@ -623,32 +236,6 @@ func (e *postgresEngine) serviceAction(ctx context.Context, action string) error // A minimal environment for a child that is not carrying secrets. const defaultGuestPath = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" -// Feeds sql on stdin rather than as an argument, so a statement is never -// visible in the process table. -func (e *postgresEngine) psqlRun(ctx context.Context, sql string, env ...string) (string, error) { - return e.run(ctx, command{ - Name: e.psql, - Args: e.psqlArgs(), - Env: append([]string{"PATH=" + defaultGuestPath}, env...), - Stdin: sql, - User: e.osUser, - }) -} - -// Reading the script from stdin is what lets one invocation serve both a -// one-shot run and a session held open across several statements. -func (e *postgresEngine) psqlArgs() []string { - return []string{ - "--no-psqlrc", "--quiet", "--no-align", "--tuples-only", - "-v", "ON_ERROR_STOP=1", - "-h", e.socketDir, - "-p", strconv.FormatInt(e.port.Load(), 10), - "-U", e.osUser, - "-d", "postgres", - "-f", "-", - } -} - func redact(s, secret string) string { if secret == "" { return s diff --git a/cmd/rds-agent/engine_contract_test.go b/cmd/rds-agent/engine_contract_test.go new file mode 100644 index 000000000..c64d4e960 --- /dev/null +++ b/cmd/rds-agent/engine_contract_test.go @@ -0,0 +1,329 @@ +package main + +import ( + "context" + "errors" + "os" + "slices" + "strings" + "testing" + "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// A probe the test drives directly, in place of whichever binaries the engine's +// own state function would have run. Each implementation's state function is +// covered on its own; what these cases exercise is what the engine does with the +// answer. +func drivenProbe(state *engineState) *engineProbe { + return newEngineProbe(0, func(context.Context, int64) (engineState, string) { + return *state, "driven by the test" + }) +} + +// One engine behind the interfaces the heartbeat, the commander and the rollback +// guard hold, plus the few facts a shared case needs to say what it means. +type engineHarness struct { + engine engine + runner *recordingRunner + params parameterStore + service string + // The state the probe reports, moved by the test rather than by the engine. + state *engineState + // Points the next quiesce at a session the test can inspect, and returns the + // commands those sessions were opened with. + useSession func(*fakeSession) *[]command + // A setting this engine's catalog really carries, so a shared case installs + // something the implementation would genuinely be handed. + parameter handlers_rds.Parameter + // What the engine's own include directive globs. Neither the rollback copy + // nor the serving copy may end in it. + includeSuffix string +} + +func engineHarnesses() []struct { + name string + build func(*testing.T) *engineHarness +} { + return []struct { + name string + build func(*testing.T) *engineHarness + }{ + {name: enginePostgres, build: buildPostgresHarness}, + {name: engineMariaDB, build: buildMariaDBHarness}, + } +} + +func forEachEngine(t *testing.T, run func(*testing.T, *engineHarness)) { + t.Helper() + for _, fixture := range engineHarnesses() { + t.Run(fixture.name, func(t *testing.T) { run(t, fixture.build(t)) }) + } +} + +func buildPostgresHarness(t *testing.T) *engineHarness { + t.Helper() + runner := &recordingRunner{} + e := newTestEngine(t, runner.run) + state := engineServing + e.probe = drivenProbe(&state) + + return &engineHarness{ + engine: e, runner: runner, params: e.params, service: e.service, state: &state, + useSession: func(session *fakeSession) *[]command { + started := &[]command{} + e.startSess = func(_ context.Context, c command) (engineSession, error) { + *started = append(*started, c) + return session, nil + } + return started + }, + parameter: handlers_rds.Parameter{Name: "work_mem", Value: "4096"}, + includeSuffix: ".conf", + } +} + +func buildMariaDBHarness(t *testing.T) *engineHarness { + t.Helper() + runner := &recordingRunner{} + e := newTestMariaDBEngine(t, runner.run) + state := engineServing + e.probe = drivenProbe(&state) + + return &engineHarness{ + engine: e, runner: runner, params: e.params, service: e.service, state: &state, + useSession: func(session *fakeSession) *[]command { + started := &[]command{} + e.startSess = func(_ context.Context, c command) (engineSession, error) { + *started = append(*started, c) + return session, nil + } + return started + }, + parameter: handlers_rds.Parameter{Name: "max_connections", Value: "200"}, + includeSuffix: ".cnf", + } +} + +// The parameters land where rds-init installs them, so the next boot overwrites +// the file rather than shadowing it with a second copy. +func TestEngineContract_ApplyParametersInstallsWhereRdsInitDoes(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + if _, err := h.engine.ApplyParameters(t.Context(), []handlers_rds.Parameter{h.parameter}); err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + + installed, err := os.ReadFile(h.params.installedPath()) + if err != nil { + t.Fatalf("read the installed parameters: %v", err) + } + if !strings.Contains(string(installed), h.parameter.Name+" = '"+h.parameter.Value+"'") { + t.Errorf("installed parameters = %q, want the resolved value", installed) + } + if !strings.HasSuffix(h.params.installedPath(), h.includeSuffix) { + t.Errorf("the installed set at %s is not a file the engine includes", h.params.installedPath()) + } + // Comparison material, not a second set of settings: an include glob that + // read either of these would apply a withdrawn configuration. + copies := []string{h.params.lastGoodPath()} + if h.params.serving != "" { + copies = append(copies, h.params.servingPath()) + } + for _, path := range copies { + if strings.HasSuffix(path, h.includeSuffix) { + t.Errorf("the copy at %s would be included by the engine", path) + } + } + }) +} + +// An engine still starting or replaying has not finished adopting the set it is +// on, so replacing that set underneath it would be installing over a moving +// target. +func TestEngineContract_ApplyParametersRefusesWhileTheEngineIsRecovering(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + *h.state = engineRecovering + + if _, err := h.engine.ApplyParameters(t.Context(), []handlers_rds.Parameter{h.parameter}); err == nil { + t.Fatal("ApplyParameters succeeded against an engine that was still coming up") + } + if _, err := os.Stat(h.params.installedPath()); !os.IsNotExist(err) { + t.Errorf("the refused set was installed anyway (stat err = %v)", err) + } + }) +} + +// The boot-time half of the parameter safety net: the last set the engine +// accepted goes back only when the engine is genuinely down, and only once. +func TestEngineContract_RestoreLastKnownGoodOnlyRollsBackADownEngine(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + accepted := h.parameter + if _, err := h.engine.ApplyParameters(t.Context(), []handlers_rds.Parameter{accepted}); err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + if err := h.engine.RecordServingParameters(t.Context()); err != nil { + t.Fatalf("RecordServingParameters: %v", err) + } + if _, err := h.engine.ApplyParameters(t.Context(), + []handlers_rds.Parameter{{Name: accepted.Name, Value: accepted.Value + "0"}}); err != nil { + t.Fatalf("second ApplyParameters: %v", err) + } + + // A guard whose deadline expired while an apply brought the engine back must + // not reverse that repair. + restored, err := h.engine.RestoreLastKnownGoodParameters(t.Context()) + if err != nil { + t.Fatalf("RestoreLastKnownGoodParameters against a serving engine: %v", err) + } + if restored { + t.Fatal("rolled the parameters back while the engine was serving") + } + + *h.state = engineAbsent + restored, err = h.engine.RestoreLastKnownGoodParameters(t.Context()) + if err != nil { + t.Fatalf("RestoreLastKnownGoodParameters: %v", err) + } + if !restored { + t.Fatal("did not roll back to a set that differs from the last accepted one") + } + installed, err := os.ReadFile(h.params.installedPath()) + if err != nil { + t.Fatalf("read the installed parameters: %v", err) + } + if !strings.Contains(string(installed), accepted.Name+" = '"+accepted.Value+"'") { + t.Errorf("installed = %q, want the last accepted set", installed) + } + + // Repeating a rollback would only churn a cluster that is failing for + // another reason. + restored, err = h.engine.RestoreLastKnownGoodParameters(t.Context()) + if err != nil { + t.Fatalf("RestoreLastKnownGoodParameters: %v", err) + } + if restored { + t.Error("rolled back again when the installed set already matched") + } + }) +} + +// Through the service manager rather than the engine's own control tool, so the +// supervisor records the engine as stopped and does not restart it underneath a +// VM that is going down. +func TestEngineContract_StopGoesThroughTheServiceManager(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + if err := h.engine.Stop(t.Context()); err != nil { + t.Fatalf("Stop: %v", err) + } + if len(h.runner.calls) != 1 { + t.Fatalf("ran %d commands, want the service stop", len(h.runner.calls)) + } + call := h.runner.calls[0] + if !slices.Equal(call.Args, []string{h.service, "stop"}) { + t.Errorf("stop ran %s %v, want the service manager", call.Name, call.Args) + } + }) +} + +// One backup at a time, released by Unquiesce or by its own deadline, whichever +// comes first. The deadline is what keeps a control plane that dies mid-snapshot +// from leaving the engine held. +func TestEngineContract_QuiesceHoldsUntilReleased(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + session := &fakeSession{} + started := h.useSession(session) + + if err := h.engine.Quiesce(t.Context(), "orders-db-pre-upgrade", time.Minute); err != nil { + t.Fatalf("Quiesce: %v", err) + } + if len(*started) != 1 { + t.Fatalf("started %d sessions, want 1", len(*started)) + } + if session.closes() != 0 { + t.Fatal("the session was closed while the backup was meant to be held") + } + + err := h.engine.Quiesce(t.Context(), "second", time.Minute) + if err == nil { + t.Fatal("a second quiesce was accepted while the first was held") + } + if !strings.Contains(err.Error(), "orders-db-pre-upgrade") { + t.Errorf("error %q does not name the backup already holding the engine", err) + } + if len(*started) != 1 { + t.Errorf("started %d sessions, want the second refused before opening one", len(*started)) + } + + if err := h.engine.Unquiesce(t.Context()); err != nil { + t.Fatalf("Unquiesce: %v", err) + } + if session.closes() != 1 { + t.Errorf("closed %d times, want the session ended exactly once", session.closes()) + } + }) +} + +// A hold that expired mid-snapshot means the snapshot was not taken against a +// held checkpoint, so the release reports it rather than succeeding silently. +func TestEngineContract_QuiesceReleasesItselfOnItsDeadline(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + session := &fakeSession{} + h.useSession(session) + + if err := h.engine.Quiesce(t.Context(), "orders-db-pre-upgrade", 10*time.Millisecond); err != nil { + t.Fatalf("Quiesce: %v", err) + } + deadline := time.Now().Add(2 * time.Second) + for session.closes() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if session.closes() != 1 { + t.Fatal("the backup session outlived its deadline") + } + + err := h.engine.Unquiesce(t.Context()) + if err == nil { + t.Fatal("Unquiesce succeeded after the hold expired") + } + if !strings.Contains(err.Error(), "expired") { + t.Errorf("error %q does not say the hold had expired", err) + } + }) +} + +// A failed start leaves nothing held, so a retry is not refused by a hold that +// never took effect. +func TestEngineContract_QuiesceEndsTheSessionWhenTheBackupWillNotStart(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + session := &fakeSession{execErr: errors.New("the engine refused the backup")} + h.useSession(session) + + if err := h.engine.Quiesce(t.Context(), "orders-db-pre-upgrade", time.Minute); err == nil { + t.Fatal("Quiesce succeeded against an engine that refused the backup") + } + if session.closes() != 1 { + t.Errorf("closed %d times, want the failed session ended", session.closes()) + } + if err := h.engine.Unquiesce(t.Context()); err == nil { + t.Error("a failed quiesce left a hold behind") + } + }) +} + +// Every command the control plane can issue is served by both implementations, +// so a directive cannot be answered as unsupported on one engine and honoured on +// the other. +func TestEngineContract_ServesEveryRegisteredCommand(t *testing.T) { + forEachEngine(t, func(t *testing.T, h *engineHarness) { + registry := newCommandRegistry(h.engine, &fakeStorage{}) + for _, name := range []string{ + handlers_rds.CommandSetPassword, handlers_rds.CommandApplyParams, handlers_rds.CommandStopEngine, + handlers_rds.CommandGrowFilesystem, handlers_rds.CommandQuiesce, handlers_rds.CommandUnquiesce, + } { + if _, ok := registry[name]; !ok { + t.Errorf("command %q has no handler", name) + } + } + }) +} diff --git a/cmd/rds-agent/engine_mariadb.go b/cmd/rds-agent/engine_mariadb.go new file mode 100644 index 000000000..c2fef72f5 --- /dev/null +++ b/cmd/rds-agent/engine_mariadb.go @@ -0,0 +1,588 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "syscall" + "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// The control plane's own statements about this engine, resolved once at +// startup. Held as functions rather than as the Engine value so nothing here +// re-derives a rule the API already owns. +type controlPlaneRules struct { + // The whole master-username rule rather than only the reserved set. MariaDB's + // client has no identifier quoting, so the name is interpolated into SQL by + // this process and every part of the rule is load-bearing here. + validateUsername func(username string) error + // Whether a setting takes effect only at a restart. Already customer-facing + // and authoritative, since the API refuses ApplyMethod=immediate on one. + isStatic func(name string) bool +} + +func controlPlaneRulesFrom(meta handlers_rds.Engine) controlPlaneRules { + return controlPlaneRules{ + validateUsername: meta.ValidateMasterUsername, + isStatic: func(name string) bool { + spec, ok := meta.LookupParameter(name) + // A name the catalog does not carry cannot be shown to have been adopted + // without a restart, and is never issued as a live SET GLOBAL either. + return !ok || spec.ApplyType == handlers_rds.ApplyTypeStatic + }, + } +} + +// The engine is reached over its unix socket as root, which the datadir's +// unix_socket plugin authenticates from the connecting process' own uid — the +// direct analogue of PostgreSQL's peer auth, and the reason the agent holds no +// password of its own. +type mariadbEngine struct { + quiesceState + + rules controlPlaneRules + run commandRunner + startSess sessionRunner + client string + admin string + socket string + rcService string + service string + probe *engineProbe + + params parameterStore + // Serializes parameter installs, serving snapshots and rollback restores so + // none can copy or replace an intermediate configuration. + paramMu sync.Mutex + repairTimeout time.Duration + repairPoll time.Duration +} + +var _ engine = (*mariadbEngine)(nil) + +const ( + mariadbClientBinary = "mariadb" + mariadbAdminBinary = "mariadb-admin" + // The account mariadb-install-db creates for the unix_socket plugin, which + // rds-init keeps for the platform and never hands to the customer. + mariadbSuperuser = "root" + // Named by the platform drop-in rds-init writes, so both halves reach the + // same socket without either asserting it to the other. + mariadbSocketFile = "mysqld.sock" + // The customer reaches the instance over its customer ENI, so the master is + // created and rotated on the wildcard host rather than on localhost. + mariadbMasterHost = "%" +) + +// The generated drop-ins on the data volume. The rollback and serving copies +// deliberately do not end in .cnf: !includedir reads *.cnf, and neither is a +// second set of settings. +const ( + mariadbParametersFile = "10-rds-parameters.cnf" + mariadbLastGoodFile = "10-rds-parameters.last-good" + mariadbServingFile = "10-rds-parameters.serving" + // MariaDB treats a setting before any group as a fatal parsing error, so + // every generated file carries the header rds-init writes. + mariadbParametersHeader = "[mysqld]\n" +) + +// The layout's factory. The rules come from the control plane's own definition +// of this engine, so a build whose control plane does not offer MariaDB refuses +// to run this implementation rather than inventing a definition for it. +func newMariaDBEngineFromCatalog(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) (engine, error) { + meta, err := handlers_rds.LookupEngine(engineMariaDB) + if err != nil { + return nil, fmt.Errorf("this image bakes %s, which this build's control plane does not offer: %w", engineMariaDB, err) + } + return newMariaDBEngine(cfg, controlPlaneRulesFrom(meta), run, startSess, probe), nil +} + +func newMariaDBEngine(cfg config, rules controlPlaneRules, run commandRunner, startSess sessionRunner, probe *engineProbe) *mariadbEngine { + return &mariadbEngine{ + rules: rules, + run: run, + startSess: startSess, + client: filepath.Join(cfg.EngineBinDir, mariadbClientBinary), + admin: filepath.Join(cfg.EngineBinDir, mariadbAdminBinary), + socket: mariadbSocketPath(cfg), + rcService: cfg.RCService, + service: cfg.EngineService, + probe: probe, + params: parameterStore{ + // The mount point rather than the datadir one level inside it: the + // include directory has to outlive the sweep a failed bootstrap runs. + dir: filepath.Join(cfg.DataMount, "conf.d"), + installed: mariadbParametersFile, + lastGood: mariadbLastGoodFile, + serving: mariadbServingFile, + header: mariadbParametersHeader, + osUser: cfg.EngineUser, + }, + repairTimeout: parameterRepairTimeout, + repairPoll: parameterRepairPoll, + } +} + +func mariadbSocketPath(cfg config) string { + return filepath.Join(cfg.SocketDir, mariadbSocketFile) +} + +// Nothing this implementation runs needs the assigned port: every connection it +// makes is over the engine's unix socket, and the platform drop-in rds-init +// writes is what puts the server on the port. The probe follows it separately. +func (e *mariadbEngine) setPort(int) {} + +// Reports whether a pid is a live process. EPERM is a process this agent may not +// signal rather than one that is gone, which matters because a stale pidfile +// read as "alive" is far less dangerous than a live engine read as "absent". +type processLivenessFn func(pid int) bool + +func processAlive(pid int) bool { + if pid <= 0 { + return false + } + err := syscall.Kill(pid, 0) + return err == nil || errors.Is(err, syscall.EPERM) +} + +func newMariaDBProbe(cfg config, run probeRunner) *engineProbe { + return newEngineProbe(cfg.EnginePort, mariadbProbeState(cfg, run, processAlive)) +} + +// Three stages, because MariaDB has no single signal that separates an engine +// still coming up from one that is not there at all: during InnoDB crash +// recovery mariadbd opens neither its socket nor its port, so a ping fails +// exactly as it would against nothing. +// +// Losing that distinction would break the rollback guard, which treats a +// recovering engine as a reason to reset its deadline. An instance killed hard +// mid-write can spend minutes replaying its redo log; read as absent, the guard +// would roll the parameter file back and restart the server mid-recovery, +// discarding the work — a rollback that exists to break a boot loop would create +// one, on an instance whose parameters were never at fault. +// +// The port is not consulted. Both stages below reach the server over its unix +// socket, whose path does not move with the port. +func mariadbProbeState(cfg config, run probeRunner, alive processLivenessFn) probeStateFn { + pidFile, socket := cfg.EnginePidFile, mariadbSocketPath(cfg) + admin, client := filepath.Join(cfg.EngineBinDir, mariadbAdminBinary), filepath.Join(cfg.EngineBinDir, mariadbClientBinary) + connect := []string{"--no-defaults", "--protocol=socket", "--socket=" + socket, "--user=" + mariadbSuperuser} + + return func(ctx context.Context, _ int64) (engineState, string) { + pid, err := readPidFile(pidFile) + switch { + case os.IsNotExist(err): + return engineAbsent, fmt.Sprintf("the engine has written no pidfile at %s", pidFile) + case err != nil: + return engineAbsent, fmt.Sprintf("the engine pidfile %s could not be read: %v", pidFile, err) + case !alive(pid): + return engineAbsent, fmt.Sprintf("the engine pidfile %s names pid %d, which is not running", pidFile, pid) + } + + // The process is up from here on, so every remaining failure is an engine + // that is not serving yet rather than one that is gone — including a probe + // binary that will not run at all. Reporting absent against a live server + // would have the rollback guard restart one that is making progress. + switch code, err := run(ctx, admin, append(slices.Clone(connect), "ping")...); { + case err != nil: + return engineRecovering, fmt.Sprintf("engine probe could not run: %v", err) + case code != 0: + return engineRecovering, "engine is not answering on its socket yet (startup or crash recovery)" + } + + // ping answers successfully even on ER_ACCESS_DENIED, so on its own it + // certifies a server that may be unable to execute anything at all. The + // statement is a literal in the argv rather than on stdin because it + // carries nothing secret and a probe reads no result back. + query := append(slices.Clone(connect), "--batch", "--skip-column-names", "--execute=SELECT 1") + switch code, err := run(ctx, client, query...); { + case err != nil: + return engineRecovering, fmt.Sprintf("engine probe could not run: %v", err) + case code != 0: + return engineRecovering, "engine answered its socket but would not execute a statement" + } + return engineServing, "" + } +} + +func readPidFile(path string) (int, error) { + raw, err := os.ReadFile(path) + if err != nil { + return 0, err + } + pid, err := strconv.Atoi(strings.TrimSpace(string(raw))) + if err != nil { + return 0, fmt.Errorf("%s does not hold a pid: %w", path, err) + } + return pid, nil +} + +func (e *mariadbEngine) Stop(ctx context.Context) error { + return serviceAction(ctx, e.run, e.rcService, e.service, "stop") +} + +// Only the parameter rollback and a failed apply call this: a restart the +// control plane wants goes through RebootDBInstance, which cycles the VM. +// +// The serving copy is refreshed first, because the server is about to start on +// whatever is installed. That is exactly what rds-init does on every boot, and +// it is what keeps the two equal by construction whenever the engine starts. +func (e *mariadbEngine) Restart(ctx context.Context) error { + if err := e.recordServingCopy(); err != nil { + return err + } + return serviceAction(ctx, e.run, e.rcService, e.service, "restart") +} + +// The installed set copied to the name the pending-restart comparison reads. A +// live SET GLOBAL apply deliberately does not touch it: that value was adopted +// without a restart, which is what "not pending" means. +func (e *mariadbEngine) recordServingCopy() error { + content, err := os.ReadFile(e.params.installedPath()) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("read the installed parameters: %w", err) + } + return e.params.write(e.params.servingPath(), content) +} + +// A value safe to interpolate into a single-quoted SQL literal. The quote is +// doubled, which holds in every sql_mode, and the backslash is doubled, which is +// why every statement built here pins a mode with backslash escapes on. +func sqlLiteral(value string) string { + return strings.ReplaceAll(strings.ReplaceAll(value, `\`, `\\`), `'`, `''`) +} + +// Turns off the two logs that would otherwise copy a statement's text, and pins +// the mode the escaping above assumes. All three are session-scoped: a rotation +// must not leave the customer's own general log switched off behind it. +const mariadbSessionGuard = `SET SESSION sql_log_off = 1; +SET SESSION slow_query_log = 0; +SET SESSION sql_mode = 'NO_ENGINE_SUBSTITUTION'; +` + +// Rotates the master's password. The statement is built here rather than by the +// client, which has neither psql's identifier interpolation nor its parameter +// quoting, and it rides stdin so it is never visible in the process table. +func (e *mariadbEngine) SetPassword(ctx context.Context, username, password string) error { + if username == "" || password == "" { + return fmt.Errorf("set-password requires both %s and %s", + handlers_rds.CommandParamMasterUsername, handlers_rds.CommandParamMasterUserPassword) + } + // This runs as the server's own superuser over the socket, so a reserved name + // in the command payload would rotate the platform's account rather than the + // customer's. The rest of the rule matters just as much here, because the name + // is interpolated into the statement below rather than quoted by the client. + if err := e.rules.validateUsername(username); err != nil { + return fmt.Errorf("refusing to set the password of role %q: %w", username, err) + } + + escaped := sqlLiteral(password) + sql := fmt.Sprintf("%sALTER USER '%s'@'%s' IDENTIFIED BY '%s';\n", + mariadbSessionGuard, username, mariadbMasterHost, escaped) + if _, err := e.clientRun(ctx, sql); err != nil { + // The client echoes the failing statement, which carries the escaped form + // of the password rather than the raw one — so redacting only the raw one + // would leak exactly the passwords that needed escaping. + return fmt.Errorf("apply the master password: %s", redact(redact(err.Error(), password), escaped)) + } + return nil +} + +// Installs the resolved set and applies the half of it a running server can +// adopt. MariaDB re-reads no configuration file while it is up, so SET GLOBAL is +// the only way an immediate apply reaches it; the static half waits in the +// installed file for the next start, and is what comes back as pending. +// +// There is no offline parser to check the set with first — MariaDB ships no +// equivalent of postgres -C — so the safety net for a static value the server +// will not start on is the boot-time rollback to the last accepted set. +func (e *mariadbEngine) ApplyParameters(ctx context.Context, params []handlers_rds.Parameter) ([]string, error) { + e.paramMu.Lock() + defer e.paramMu.Unlock() + + initialState, _ := e.probe.state(ctx) + if initialState == engineRecovering { + return nil, errors.New("apply parameters while the engine is still starting or recovering") + } + if initialState == engineServing { + // A command can arrive before the first heartbeat. Seed the configuration + // currently being served before replacing its file in that window. + if err := e.recordServingParametersLocked(ctx); err != nil { + return nil, fmt.Errorf("record the parameters serving before the apply: %w", err) + } + } + + restore, err := e.params.install(params) + if err != nil { + return nil, err + } + + state, _ := e.probe.state(ctx) + switch state { + case engineAbsent: + return e.restartOnRepairSetLocked(ctx) + case engineRecovering: + return nil, errors.Join(errors.New("the engine entered startup or recovery during the parameter apply"), restore()) + } + + if err := e.applyDynamicParameters(ctx, params); err != nil { + // The engine may have gone down after the first probe. In that case start it + // on the newly installed set rather than restoring one that already left it + // unable to serve. + if state, _ := e.probe.state(ctx); state == engineAbsent { + return e.restartOnRepairSetLocked(ctx) + } + return nil, errors.Join(err, restore()) + } + return e.pendingRestartParameters(ctx) +} + +// The dynamic half of the set, in one invocation that stops at the first refusal +// so the server's own message names the setting it would not take. +// +// Only names the catalog carries are emitted, which is also what keeps the +// statement safe to build: a name is an identifier here, and the classification +// that selects it comes from the same table the API validated the request +// against. MariaDB cannot set several globals atomically, so a value set before +// a refusal stays live until the next restart, when the restored file wins. +func (e *mariadbEngine) applyDynamicParameters(ctx context.Context, params []handlers_rds.Parameter) error { + var b strings.Builder + b.WriteString("SET SESSION sql_mode = 'NO_ENGINE_SUBSTITUTION';\n") + applied := 0 + for _, p := range params { + if p.Name == "" || e.rules.isStatic(p.Name) { + continue + } + fmt.Fprintf(&b, "SET GLOBAL %s = '%s';\n", p.Name, sqlLiteral(p.Value)) + applied++ + } + if applied == 0 { + return nil + } + if _, err := e.clientRun(ctx, b.String()); err != nil { + return fmt.Errorf("apply the dynamic parameters: %w", err) + } + return nil +} + +func (e *mariadbEngine) restartOnRepairSetLocked(ctx context.Context) ([]string, error) { + return awaitRepairedEngine(ctx, e.probe, e.Restart, e.pendingRestartParameters, e.repairTimeout, e.repairPoll) +} + +// MariaDB has no pending_restart. Nothing in information_schema reports a +// setting the server has stored but not adopted, so the answer is computed from +// the files instead: the catalog-static keys whose value in the installed +// drop-in differs from the one the server actually started on. +// +// Comparing live values against the desired set was rejected. mysqld silently +// rewrites values at startup — rounding the buffer pool and log file sizes, and +// downgrading max_connections when it cannot obtain enough file descriptors — +// and each rewrite would read as a permanent pending restart: un-clearable +// pending-reboot state, permanent drift, and a last-known-good copy frozen for +// good, silently disabling a recovery mechanism only exercised during an +// incident. +func (e *mariadbEngine) pendingRestartParameters(context.Context) ([]string, error) { + installed, err := readOptionFile(e.params.installedPath()) + if err != nil { + if os.IsNotExist(err) { + // Nothing is installed, so nothing is waiting to be adopted. + return nil, nil + } + return nil, fmt.Errorf("read the installed parameters: %w", err) + } + serving, err := readOptionFile(e.params.servingPath()) + if err != nil && !os.IsNotExist(err) { + return nil, fmt.Errorf("read the parameters the engine started on: %w", err) + } + + // An absent serving copy is a set the engine has not started on: rds-init + // writes the two together on every boot, so the whole static half counts as + // pending rather than being promoted on the strength of a missing file. + var pending []string + for name, value := range installed { + if !e.rules.isStatic(name) { + continue + } + if served, ok := serving[name]; !ok || served != value { + pending = append(pending, name) + } + } + // A static setting the group stopped naming reverts to its default at the next + // start, which the running server has not adopted either. + for name := range serving { + if _, ok := installed[name]; !ok && e.rules.isStatic(name) { + pending = append(pending, name) + } + } + slices.Sort(pending) + return pending, nil +} + +// Snapshots the installed set as the rollback target, but only when nothing in +// it is still waiting for a restart. The parameter mutex keeps an apply from +// replacing the file between that check and the copy. +func (e *mariadbEngine) RecordServingParameters(ctx context.Context) error { + e.paramMu.Lock() + defer e.paramMu.Unlock() + return e.recordServingParametersLocked(ctx) +} + +func (e *mariadbEngine) recordServingParametersLocked(ctx context.Context) error { + return recordLastGood(ctx, e.params, e.pendingRestartParameters) +} + +// Puts the last set the engine accepted back in place, for a start that failed +// after a parameter change. +func (e *mariadbEngine) RestoreLastKnownGoodParameters(ctx context.Context) (bool, error) { + e.paramMu.Lock() + defer e.paramMu.Unlock() + return restoreLastGood(ctx, e.params, e.probe) +} + +// The subset of MariaDB's option-file syntax the generated drop-ins are written +// in: a group header, comments, and one `name = value` per line. Nothing else +// reaches these two files, because only rds-init and this agent write them. +func readOptionFile(path string) (map[string]string, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, err + } + values := make(map[string]string) + for line := range strings.SplitSeq(string(raw), "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") || strings.HasPrefix(line, ";") || strings.HasPrefix(line, "[") { + continue + } + // A bare option such as skip-log-bin carries no value; it is still a setting + // whose presence or absence differs. + name, value, _ := strings.Cut(line, "=") + values[normaliseOptionName(name)] = strings.Trim(strings.TrimSpace(value), `"'`) + } + return values, nil +} + +// MariaDB reads - and _ as the same character in an option name, so the two +// spellings of one setting must not compare as two settings. +func normaliseOptionName(name string) string { + return strings.ReplaceAll(strings.TrimSpace(name), "-", "_") +} + +// How long any one BACKUP STAGE waits for the metadata locks it needs. Well +// under the control plane's own quiesce timeout, so a stage that cannot take its +// lock fails and is reported rather than being abandoned mid-wait with the +// request still queued in front of live traffic. +const mariadbQuiesceLockWait = 20 * time.Second + +// MariaDB's own backup API rather than FLUSH TABLES WITH READ LOCK. FTWRL turns +// a snapshot into a write outage: the whole database is read-only until it is +// released, so a control plane that died after issuing quiesce would leave the +// customer read-only for the full hold, on a schedule and unattended. Its +// acquisition phase is unbounded too, with writes queueing behind a pending lock +// request. BACKUP STAGE blocks commits rather than all writes and its flush +// stage covers Aria and MyISAM, so it yields a better consistency point while +// being far less disruptive — and it is what mariadb-backup itself uses. +func (e *mariadbEngine) Quiesce(ctx context.Context, label string, hold time.Duration) error { + if err := validateQuiesceRequest(label, hold); err != nil { + return err + } + + // Held across the whole start, so a second quiesce waits and then finds the + // first one's hold rather than opening a concurrent backup alongside it. + e.mu.Lock() + defer e.mu.Unlock() + if e.held != nil { + return fmt.Errorf("the engine is already quiesced for backup %s", e.held.label) + } + + // Deliberately not the command's context: the session has to outlive the call + // that started it, and its own deadline is what bounds it instead. + session, err := e.startSess(context.WithoutCancel(ctx), command{ + Name: e.client, + Args: e.clientArgs(), + Env: []string{"PATH=" + defaultGuestPath}, + SentinelStatement: "SELECT '" + sessionSentinel + "';\n", + }) + if err != nil { + return fmt.Errorf("open a backup session: %w", err) + } + + // The stages run in order and on one connection: MariaDB releases the whole + // hold with the session that took it, which is what bounds a control plane + // that dies mid-snapshot. Fed one at a time so a stage that cannot take its + // locks is named rather than reported as a line number. + stages := []string{ + fmt.Sprintf("SET SESSION lock_wait_timeout = %d;\n", int(mariadbQuiesceLockWait.Seconds())), + "BACKUP STAGE START;\n", + "BACKUP STAGE FLUSH;\n", + "BACKUP STAGE BLOCK_DDL;\n", + "BACKUP STAGE BLOCK_COMMIT;\n", + } + for _, sql := range stages { + if err := session.Exec(ctx, sql); err != nil { + if closeErr := session.Close(); closeErr != nil { + slog.Warn("rds-agent: closing a failed backup session", "err", closeErr) + } + return fmt.Errorf("put the engine into backup mode at %q: %w", strings.TrimSpace(sql), err) + } + } + + e.beginHoldLocked(label, session, hold) + return nil +} + +// Ends the backup cleanly. A missing hold is an error rather than a silent +// success: it means the deadline fired first, so the snapshot the control plane +// just took was not taken against a held stage. +func (e *mariadbEngine) Unquiesce(ctx context.Context) error { + held := e.takeHold() + if held == nil { + return errors.New("the engine is not quiesced; the backup hold had already expired") + } + + // The release has to run on the session that took the stages, and the session + // is closed either way — the engine ends an unreleased backup with it. + execErr := held.session.Exec(ctx, "BACKUP STAGE END;\n") + closeErr := held.session.Close() + if execErr != nil { + return fmt.Errorf("take the engine out of backup mode: %w", errors.Join(execErr, closeErr)) + } + if closeErr != nil { + return fmt.Errorf("close the backup session: %w", closeErr) + } + slog.Info("rds-agent: engine released from backup mode", "label", held.label) + return nil +} + +// Feeds sql on stdin rather than as an argument, so a statement is never visible +// in the process table. +func (e *mariadbEngine) clientRun(ctx context.Context, sql string) (string, error) { + return e.run(ctx, command{ + Name: e.client, + Args: e.clientArgs(), + Env: []string{"PATH=" + defaultGuestPath}, + Stdin: sql, + }) +} + +// --no-defaults so no option file can move the connection this agent makes, and +// --unbuffered so a held session's output reaches the reader as each statement +// finishes rather than when a pipe buffer fills. No User: the agent is root, and +// root is what the datadir's unix_socket plugin authenticates. +func (e *mariadbEngine) clientArgs() []string { + return []string{ + "--no-defaults", "--batch", "--skip-column-names", "--unbuffered", + "--protocol=socket", "--socket=" + e.socket, "--user=" + mariadbSuperuser, + } +} diff --git a/cmd/rds-agent/engine_mariadb_test.go b/cmd/rds-agent/engine_mariadb_test.go new file mode 100644 index 000000000..22153c0a5 --- /dev/null +++ b/cmd/rds-agent/engine_mariadb_test.go @@ -0,0 +1,682 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "os/user" + "path/filepath" + "slices" + "strconv" + "strings" + "testing" + "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// Stands in for the control plane's own definition of the engine. It mirrors the +// production wiring exactly, including that a name the catalog does not carry +// counts as static: a rule the fake inverted would hide the failure it exists to +// prevent. +func testMariaDBRules() controlPlaneRules { + applyTypes := map[string]string{ + "innodb_buffer_pool_size": handlers_rds.ApplyTypeStatic, + "innodb_log_file_size": handlers_rds.ApplyTypeStatic, + "max_connections": handlers_rds.ApplyTypeDynamic, + "long_query_time": handlers_rds.ApplyTypeDynamic, + } + reserved := []string{"root", "mysql", "mariadb.sys", "rdsadmin", "public"} + + return controlPlaneRules{ + validateUsername: func(username string) error { + lower := strings.ToLower(username) + if slices.Contains(reserved, lower) || strings.HasPrefix(lower, "mysql.") || strings.HasPrefix(lower, "mariadb.") { + return fmt.Errorf("MasterUsername %q is reserved by mariadb", username) + } + for _, r := range username { + if !isTestIdentifierRune(r) { + return fmt.Errorf("MasterUsername may contain only letters, digits and underscores") + } + } + return nil + }, + isStatic: func(name string) bool { + applyType, ok := applyTypes[name] + return !ok || applyType == handlers_rds.ApplyTypeStatic + }, + } +} + +func isTestIdentifierRune(r rune) bool { + return (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' +} + +func newTestMariaDBEngine(t *testing.T, run commandRunner) *mariadbEngine { + t.Helper() + cfg := testLoadConfig(t, engineMariaDB) + cfg.DataMount = t.TempDir() + if err := os.MkdirAll(filepath.Join(cfg.DataMount, "conf.d"), 0o750); err != nil { + t.Fatalf("create conf.d: %v", err) + } + // The guest's mysql user does not exist here, and the chown of the installed + // parameter file has to resolve to something real. + current, err := user.Current() + if err != nil { + t.Fatalf("resolve the current user: %v", err) + } + cfg.EngineUser = current.Username + + state := engineServing + return newMariaDBEngine(cfg, testMariaDBRules(), run, nil, drivenProbe(&state)) +} + +// The wiring the guest takes its classification from, checked against a catalog +// that is registered: what a spec says about a parameter is what the guest must +// see, and an unlisted name must read as static rather than as dynamic. +func TestControlPlaneRulesFrom_TakesTheClassificationFromTheCatalog(t *testing.T) { + rules := controlPlaneRulesFrom(postgresEngineMeta) + + for name, want := range map[string]bool{"shared_buffers": true, "work_mem": false} { + spec, ok := postgresEngineMeta.LookupParameter(name) + if !ok { + t.Fatalf("the catalog no longer carries %s", name) + } + if (spec.ApplyType == handlers_rds.ApplyTypeStatic) != want { + t.Fatalf("%s is %s in the catalog, so this test is asserting the wrong way round", name, spec.ApplyType) + } + if got := rules.isStatic(name); got != want { + t.Errorf("isStatic(%s) = %v, want %v", name, got, want) + } + } + if !rules.isStatic("not_a_parameter_any_catalog_carries") { + t.Error("a name the catalog does not carry read as dynamic, so it would be issued as a live SET GLOBAL") + } + if rules.validateUsername("postgres") == nil { + t.Error("the username rule does not refuse a reserved role") + } +} + +// The client has no parameter quoting of its own, so the password is built into +// the statement here. It must reach the process on stdin and nowhere else. +func TestMariaDBEngine_SetPasswordKeepsTheSecretOutOfArgvAndEnv(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "mulgamaster", "n3w-pw"); err != nil { + t.Fatalf("SetPassword: %v", err) + } + if len(runner.calls) != 1 { + t.Fatalf("ran %d commands, want 1", len(runner.calls)) + } + call := runner.calls[0] + for _, arg := range call.Args { + if strings.Contains(arg, "n3w-pw") { + t.Fatalf("argv %v carries the password", call.Args) + } + } + for _, env := range call.Env { + if strings.Contains(env, "n3w-pw") { + t.Fatalf("env %v carries the password, which the client would not read anyway", call.Env) + } + } + if !strings.Contains(call.Stdin, "ALTER USER 'mulgamaster'@'%' IDENTIFIED BY 'n3w-pw';") { + t.Errorf("SQL %q does not rotate the master on the wildcard host", call.Stdin) + } +} + +// ValidateMasterUserPassword permits both of these, and the client offers no +// equivalent of psql's parameter quoting, so the escaping here is what keeps a +// password from ending the literal it is inside. +func TestMariaDBEngine_SetPasswordEscapesQuotesAndBackslashes(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "mulgamaster", `a'b\c`); err != nil { + t.Fatalf("SetPassword: %v", err) + } + sql := runner.calls[0].Stdin + if !strings.Contains(sql, `IDENTIFIED BY 'a''b\\c';`) { + t.Errorf("SQL %q does not escape both the quote and the backslash", sql) + } + // The escaping assumes a mode where backslash escapes are on, so the statement + // cannot inherit whatever sql_mode the customer's parameter group set. + if !strings.Contains(sql, "SET SESSION sql_mode = 'NO_ENGINE_SUBSTITUTION';") { + t.Errorf("SQL %q does not pin the mode its escaping assumes", sql) + } +} + +// The password is interpolated into the statement, so a parameter group that +// turned the general or slow log on would copy it into a log file every snapshot +// then carries. Both are silenced for this session only: a rotation must not +// leave the customer's own logging switched off behind it. +func TestMariaDBEngine_SetPasswordSilencesStatementLoggingForItsSessionOnly(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "mulgamaster", "n3w-pw"); err != nil { + t.Fatalf("SetPassword: %v", err) + } + sql := runner.calls[0].Stdin + alter := strings.Index(sql, "ALTER USER") + if alter < 0 { + t.Fatalf("SQL %q does not alter the user", sql) + } + for _, guard := range []string{"SET SESSION sql_log_off = 1;", "SET SESSION slow_query_log = 0;"} { + at := strings.Index(sql, guard) + if at < 0 { + t.Errorf("SQL %q is missing %q", sql, guard) + continue + } + if at > alter { + t.Errorf("SQL %q sets %q only after the statement has already been logged", sql, guard) + } + } + if strings.Contains(sql, "SET GLOBAL general_log") || strings.Contains(sql, "SET GLOBAL slow_query_log") { + t.Errorf("SQL %q turns logging off globally, which outlives the rotation", sql) + } +} + +// The client echoes the statement it failed on, which carries the escaped form +// of the password rather than the raw one — so redacting only the raw one would +// leak exactly the passwords that needed escaping. +func TestMariaDBEngine_SetPasswordRedactsBothFormsOfTheSecret(t *testing.T) { + secret := `a'b\c` + runner := &recordingRunner{err: errors.New(`ERROR 1064 near "IDENTIFIED BY 'a''b\\c'"`)} + engine := newTestMariaDBEngine(t, runner.run) + + err := engine.SetPassword(context.Background(), "mulgamaster", secret) + if err == nil { + t.Fatal("SetPassword succeeded against a failing client") + } + if strings.Contains(err.Error(), secret) || strings.Contains(err.Error(), sqlLiteral(secret)) { + t.Errorf("error %q leaks the password", err) + } + if !strings.Contains(err.Error(), "[REDACTED]") { + t.Errorf("error %q does not mark the redaction", err) + } +} + +// The apply runs as the server's own superuser over the socket, and the name is +// interpolated into the statement rather than quoted by the client, so both +// halves of the rule are refused before any SQL is built. +func TestMariaDBEngine_SetPasswordRefusesNamesTheEngineWillNotTake(t *testing.T) { + for _, username := range []string{"root", "mysql", "mariadb.sys", "rdsadmin", "PUBLIC", "mysql.infoschema", `ma'ster`} { + t.Run(username, func(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), username, "n3w-pw"); err == nil { + t.Fatalf("SetPassword accepted %q", username) + } + if len(runner.calls) != 0 { + t.Errorf("ran %d commands for a refused name, want 0", len(runner.calls)) + } + }) + } +} + +func TestMariaDBEngine_SetPasswordRejectsAnIncompleteCommand(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "mulgamaster", ""); err == nil { + t.Error("SetPassword accepted an empty password") + } + if len(runner.calls) != 0 { + t.Errorf("ran %d commands for an incomplete request, want 0", len(runner.calls)) + } +} + +// A running server re-reads no configuration file, so the dynamic half has to +// reach it as SET GLOBAL. The static half is installed and waits for a restart, +// which is what comes back as pending. +func TestMariaDBEngine_ApplyParametersSetsOnlyTheDynamicHalfLive(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + pending, err := engine.ApplyParameters(context.Background(), []handlers_rds.Parameter{ + {Name: "max_connections", Value: "200"}, + {Name: "innodb_buffer_pool_size", Value: "1073741824"}, + }) + if err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + + installed, err := os.ReadFile(engine.params.installedPath()) + if err != nil { + t.Fatalf("read the installed parameters: %v", err) + } + if !strings.HasPrefix(string(installed), "[mysqld]\n") { + t.Errorf("installed parameters = %q, want the group header rds-init writes", installed) + } + for _, want := range []string{"max_connections = '200'", "innodb_buffer_pool_size = '1073741824'"} { + if !strings.Contains(string(installed), want) { + t.Errorf("installed parameters = %q, want %q", installed, want) + } + } + + if len(runner.calls) != 1 { + t.Fatalf("ran %d commands, want the one live apply", len(runner.calls)) + } + sql := runner.calls[0].Stdin + if !strings.Contains(sql, "SET GLOBAL max_connections = '200';") { + t.Errorf("SQL %q does not set the dynamic value live", sql) + } + if strings.Contains(sql, "innodb_buffer_pool_size") { + t.Errorf("SQL %q sets a static value live, which the server would refuse", sql) + } + if !slices.Equal(pending, []string{"innodb_buffer_pool_size"}) { + t.Errorf("pending = %v, want the static setting the engine has not adopted", pending) + } +} + +// A set with nothing dynamic in it must not open a connection at all: the +// installed file is the whole of the change. +func TestMariaDBEngine_ApplyParametersRunsNoStatementForAStaticOnlySet(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if _, err := engine.ApplyParameters(context.Background(), + []handlers_rds.Parameter{{Name: "innodb_log_file_size", Value: "268435456"}}); err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + if len(runner.calls) != 0 { + t.Errorf("ran %d commands for a static-only set, want 0", len(runner.calls)) + } +} + +// A value the server refuses leaves the installed file back where it was, so it +// does not sit on the data volume turning the next restart into a boot loop. +func TestMariaDBEngine_ApplyParametersRollsBackAValueTheServerRefuses(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if _, err := engine.ApplyParameters(context.Background(), + []handlers_rds.Parameter{{Name: "max_connections", Value: "200"}}); err != nil { + t.Fatalf("first ApplyParameters: %v", err) + } + + runner.err = errors.New("ERROR 1231 (42000): Variable 'max_connections' can't be set to the value of '0'") + if _, err := engine.ApplyParameters(context.Background(), + []handlers_rds.Parameter{{Name: "max_connections", Value: "0"}}); err == nil { + t.Fatal("ApplyParameters succeeded against a value the server refused") + } + + installed, err := os.ReadFile(engine.params.installedPath()) + if err != nil { + t.Fatalf("read the installed parameters: %v", err) + } + if !strings.Contains(string(installed), "max_connections = '200'") { + t.Errorf("installed parameters = %q, want the previous accepted set restored", installed) + } +} + +// The whole of the pending-restart answer, since MariaDB has none of its own: +// the installed drop-in against the one the server actually started on, over the +// static keys only. +func TestMariaDBEngine_PendingRestartComparesTheInstalledSetAgainstTheServingCopy(t *testing.T) { + engine := newTestMariaDBEngine(t, (&recordingRunner{}).run) + write := func(path, body string) { + t.Helper() + if err := os.WriteFile(path, []byte("[mysqld]\n"+body), 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } + } + pending := func() []string { + t.Helper() + names, err := engine.pendingRestartParameters(t.Context()) + if err != nil { + t.Fatalf("pendingRestartParameters: %v", err) + } + return names + } + + // A fresh boot: rds-init wrote the two together, so nothing is waiting. + write(engine.params.installedPath(), "innodb_buffer_pool_size = '1073741824'\nmax_connections = '100'\n") + write(engine.params.servingPath(), "innodb_buffer_pool_size = '1073741824'\nmax_connections = '100'\n") + if names := pending(); len(names) != 0 { + t.Errorf("pending = %v, want nothing on a set the engine started on", names) + } + + // A dynamic value the server took live is not pending: it was adopted without + // a restart, which is exactly what the serving copy not moving records. + write(engine.params.installedPath(), "innodb_buffer_pool_size = '1073741824'\nmax_connections = '200'\n") + if names := pending(); len(names) != 0 { + t.Errorf("pending = %v, want nothing for a value adopted live", names) + } + + // A static value is. + write(engine.params.installedPath(), "innodb_buffer_pool_size = '2147483648'\nmax_connections = '200'\n") + if names := pending(); !slices.Equal(names, []string{"innodb_buffer_pool_size"}) { + t.Errorf("pending = %v, want the static setting", names) + } + + // So is a static setting the group stopped naming, which reverts to its + // default at the next start. + write(engine.params.installedPath(), "max_connections = '200'\n") + if names := pending(); !slices.Equal(names, []string{"innodb_buffer_pool_size"}) { + t.Errorf("pending = %v, want the withdrawn static setting", names) + } + + // MariaDB reads - and _ as the same character, so one setting spelled two ways + // is not two settings. + write(engine.params.installedPath(), "innodb-buffer-pool-size = '1073741824'\nmax_connections = '200'\n") + if names := pending(); len(names) != 0 { + t.Errorf("pending = %v, want the two spellings of one setting to compare equal", names) + } +} + +// rds-init writes the serving copy beside the installed one on every boot, so a +// set with no copy beside it is one the engine has not started on. Promoting it +// would point the rollback at a configuration that has never served. +func TestMariaDBEngine_PendingRestartTreatsAMissingServingCopyAsUnadopted(t *testing.T) { + engine := newTestMariaDBEngine(t, (&recordingRunner{}).run) + if err := os.WriteFile(engine.params.installedPath(), + []byte("[mysqld]\ninnodb_buffer_pool_size = '1073741824'\nmax_connections = '100'\n"), 0o600); err != nil { + t.Fatalf("write installed parameters: %v", err) + } + + names, err := engine.pendingRestartParameters(t.Context()) + if err != nil { + t.Fatalf("pendingRestartParameters: %v", err) + } + if !slices.Equal(names, []string{"innodb_buffer_pool_size"}) { + t.Errorf("pending = %v, want the static half of a set the engine never started on", names) + } +} + +// The restart the agent performs is the only thing that moves the serving copy, +// which is what makes a fresh start clear whatever was pending. +func TestMariaDBEngine_RestartRefreshesTheServingCopy(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + installed := []byte("[mysqld]\ninnodb_buffer_pool_size = '2147483648'\n") + if err := os.WriteFile(engine.params.installedPath(), installed, 0o600); err != nil { + t.Fatalf("write installed parameters: %v", err) + } + if err := os.WriteFile(engine.params.servingPath(), []byte("[mysqld]\ninnodb_buffer_pool_size = '1073741824'\n"), 0o600); err != nil { + t.Fatalf("write serving parameters: %v", err) + } + + if err := engine.Restart(context.Background()); err != nil { + t.Fatalf("Restart: %v", err) + } + + serving, err := os.ReadFile(engine.params.servingPath()) + if err != nil { + t.Fatalf("read the serving copy: %v", err) + } + if string(serving) != string(installed) { + t.Errorf("serving copy = %q, want the set the engine was just started on", serving) + } + names, err := engine.pendingRestartParameters(t.Context()) + if err != nil { + t.Fatalf("pendingRestartParameters: %v", err) + } + if len(names) != 0 { + t.Errorf("pending = %v, want a restart to have cleared it", names) + } + // The copy the comparison reads must never be parsed as a second set of + // settings by the engine's own include glob. + if strings.HasSuffix(engine.params.servingPath(), ".cnf") { + t.Errorf("the serving copy at %s would be included by the engine", engine.params.servingPath()) + } +} + +// The rollback target advances only once a restart has proved the installed +// static values can serve, not when a dynamic apply merely set them live. +func TestMariaDBEngine_LastKnownGoodWaitsForARestartToAdoptAStaticSet(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + + if _, err := engine.ApplyParameters(t.Context(), + []handlers_rds.Parameter{{Name: "innodb_buffer_pool_size", Value: "1073741824"}}); err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + if err := engine.RecordServingParameters(t.Context()); err != nil { + t.Fatalf("RecordServingParameters: %v", err) + } + if _, err := os.Stat(engine.params.lastGoodPath()); !os.IsNotExist(err) { + t.Errorf("a set the engine has not restarted onto became the rollback target (stat err = %v)", err) + } + + if err := engine.Restart(t.Context()); err != nil { + t.Fatalf("Restart: %v", err) + } + if err := engine.RecordServingParameters(t.Context()); err != nil { + t.Fatalf("RecordServingParameters after the restart: %v", err) + } + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) + if err != nil { + t.Fatalf("read the last known good parameters: %v", err) + } + if !strings.Contains(string(lastGood), "innodb_buffer_pool_size = '1073741824'") { + t.Errorf("last known good = %q, want the set the engine restarted onto", lastGood) + } +} + +// A dynamic apply is adopted without a restart, so it may advance the rollback +// target on the next healthy beat rather than waiting for one. +func TestMariaDBEngine_LastKnownGoodAdvancesOnADynamicApply(t *testing.T) { + runner := &recordingRunner{} + engine := newTestMariaDBEngine(t, runner.run) + if err := os.WriteFile(engine.params.servingPath(), []byte("[mysqld]\n"), 0o600); err != nil { + t.Fatalf("write serving parameters: %v", err) + } + + if _, err := engine.ApplyParameters(t.Context(), + []handlers_rds.Parameter{{Name: "max_connections", Value: "200"}}); err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + if err := engine.RecordServingParameters(t.Context()); err != nil { + t.Fatalf("RecordServingParameters: %v", err) + } + + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) + if err != nil { + t.Fatalf("read the last known good parameters: %v", err) + } + if !strings.Contains(string(lastGood), "max_connections = '200'") { + t.Errorf("last known good = %q, want the set the engine is already running", lastGood) + } +} + +// A probe that reports what the test hands it, in place of whichever binaries +// the engine's own state function would have run. +func mariadbTestProbeConfig(t *testing.T) config { + t.Helper() + cfg := testLoadConfig(t, engineMariaDB) + cfg.EnginePidFile = filepath.Join(t.TempDir(), "mariadb.pid") + return cfg +} + +func writePid(t *testing.T, path string, pid int) { + t.Helper() + if err := os.WriteFile(path, []byte(strconv.Itoa(pid)+"\n"), 0o644); err != nil { + t.Fatalf("write pidfile: %v", err) + } +} + +// The three stages, and what each one separates. During InnoDB crash recovery +// mariadbd opens neither its socket nor its port, so nothing the client can ask +// distinguishes a server replaying its redo log from no server at all. +func TestMariaDBProbe_SeparatesRecoveringFromAbsent(t *testing.T) { + tests := []struct { + name string + writePid bool + alive bool + pingCode int + pingErr error + execCode int + want engineState + }{ + {name: "no pidfile at all", want: engineAbsent}, + {name: "pidfile names a dead process", writePid: true, want: engineAbsent}, + {name: "process alive but socket silent", writePid: true, alive: true, pingCode: 1, want: engineRecovering}, + {name: "process alive but probe cannot run", writePid: true, alive: true, pingErr: errors.New("no such file"), want: engineRecovering}, + {name: "answers ping but executes nothing", writePid: true, alive: true, execCode: 1, want: engineRecovering}, + {name: "serving", writePid: true, alive: true, want: engineServing}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := mariadbTestProbeConfig(t) + if tt.writePid { + writePid(t, cfg.EnginePidFile, 4242) + } + run := func(_ context.Context, name string, args ...string) (int, error) { + if strings.HasSuffix(name, mariadbAdminBinary) { + return tt.pingCode, tt.pingErr + } + if !slices.Contains(args, "--execute=SELECT 1") { + t.Errorf("the second stage ran %s %v, want a statement the server has to execute", name, args) + } + return tt.execCode, nil + } + stateFn := mariadbProbeState(cfg, run, func(int) bool { return tt.alive }) + + got, message := stateFn(context.Background(), int64(cfg.EnginePort)) + if got != tt.want { + t.Errorf("state = %v, want %v (%s)", got, tt.want, message) + } + if got != engineServing && message == "" { + t.Error("a non-serving result carried no message explaining it") + } + }) + } +} + +// The failure the guard exists for still behaves correctly: a static value +// mariadbd will not accept makes it exit during startup, so the pidfile names a +// process that is gone and the deadline runs. +func TestMariaDBProbe_MapsToTheHealthTheControlPlaneReads(t *testing.T) { + cfg := mariadbTestProbeConfig(t) + writePid(t, cfg.EnginePidFile, 4242) + alive := true + serving := true + probe := newEngineProbe(cfg.EnginePort, mariadbProbeState(cfg, + func(context.Context, string, ...string) (int, error) { + if serving { + return 0, nil + } + return 1, nil + }, + func(int) bool { return alive })) + + if got, _ := probe.Check(context.Background()); got != handlers_rds.EngineHealthHealthy { + t.Fatalf("health = %q, want healthy", got) + } + serving = false + if got, _ := probe.Check(context.Background()); got != handlers_rds.EngineHealthStarting { + t.Errorf("health while recovering = %q, want starting", got) + } + alive = false + if got, _ := probe.Check(context.Background()); got != handlers_rds.EngineHealthUnhealthy { + t.Errorf("health after the engine went away = %q, want unhealthy", got) + } +} + +// An instance killed hard mid-write can spend minutes replaying its redo log. If +// that read as absent, the guard would roll the parameter file back and restart +// the server mid-recovery — a rollback that exists to break a boot loop would +// create one, on an instance whose parameters were never at fault. +func TestMariaDBProbe_LiveButUnreachableEngineResetsTheRollbackDeadline(t *testing.T) { + cfg := mariadbTestProbeConfig(t) + writePid(t, cfg.EnginePidFile, 4242) + probe := newEngineProbe(cfg.EnginePort, mariadbProbeState(cfg, + func(context.Context, string, ...string) (int, error) { return 1, nil }, + func(int) bool { return true })) + + recovery := &fakeRecovery{restored: true} + guard := newTestGuard(recovery, probe) + // Bounded so the deadline-resetting loop cannot run forever if it stops + // resetting; the assertion is that the rollback was never reached. + ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond) + defer cancel() + guard.Run(ctx) + + if recovery.restarts != 0 { + t.Errorf("restarts = %d, want none while the engine is still recovering", recovery.restarts) + } +} + +// MariaDB releases a BACKUP STAGE with the connection that took it, so the whole +// sequence has to run on one held session — which is also what bounds a control +// plane that dies mid-snapshot. +func TestMariaDBEngine_QuiesceTakesTheBackupStagesOnAHeldSession(t *testing.T) { + session := &fakeSession{} + engine := newTestMariaDBEngine(t, (&recordingRunner{}).run) + var started []command + engine.startSess = func(_ context.Context, c command) (engineSession, error) { + started = append(started, c) + return session, nil + } + + if err := engine.Quiesce(context.Background(), "orders-db-pre-upgrade", time.Minute); err != nil { + t.Fatalf("Quiesce: %v", err) + } + + if len(started) != 1 { + t.Fatalf("started %d sessions, want 1", len(started)) + } + if !slices.Contains(started[0].Args, "--unbuffered") { + t.Errorf("session argv %v does not flush after each statement, so the sentinel could sit in a pipe buffer", started[0].Args) + } + statements := session.statements() + want := []string{ + "SET SESSION lock_wait_timeout = 20;", + "BACKUP STAGE START;", + "BACKUP STAGE FLUSH;", + "BACKUP STAGE BLOCK_DDL;", + "BACKUP STAGE BLOCK_COMMIT;", + } + if len(statements) != len(want) { + t.Fatalf("statements = %q, want the bounded stage sequence %q", statements, want) + } + for i, sql := range want { + if strings.TrimSpace(statements[i]) != sql { + t.Errorf("statement %d = %q, want %q", i, statements[i], sql) + } + } + // FLUSH TABLES WITH READ LOCK would make the whole database read-only for the + // length of the hold rather than blocking commits. + if strings.Contains(strings.Join(statements, " "), "FLUSH TABLES") { + t.Error("the quiesce took a read lock over the whole database") + } + if session.closes() != 0 { + t.Error("the session was closed while the backup was meant to be held") + } +} + +// The stage bound has to leave the control plane's own quiesce timeout room for +// the whole sequence, or a stage abandoned mid-wait leaves its lock request +// queued in front of live traffic. +func TestMariaDBEngine_QuiesceLockWaitFitsInsideTheControlPlaneTimeout(t *testing.T) { + const stages = 4 + if stages*mariadbQuiesceLockWait >= 2*time.Minute { + t.Errorf("%d stages of %s can outlast the control plane's quiesce timeout", stages, mariadbQuiesceLockWait) + } +} + +func TestMariaDBEngine_UnquiesceEndsTheBackupAndTheSession(t *testing.T) { + session := &fakeSession{} + engine := newTestMariaDBEngine(t, (&recordingRunner{}).run) + engine.startSess = func(context.Context, command) (engineSession, error) { return session, nil } + if err := engine.Quiesce(context.Background(), "orders-db-pre-upgrade", time.Minute); err != nil { + t.Fatalf("Quiesce: %v", err) + } + + if err := engine.Unquiesce(context.Background()); err != nil { + t.Fatalf("Unquiesce: %v", err) + } + + statements := session.statements() + if last := strings.TrimSpace(statements[len(statements)-1]); last != "BACKUP STAGE END;" { + t.Errorf("last statement = %q, want the stage released", last) + } + if session.closes() != 1 { + t.Errorf("closed %d times, want the session ended exactly once", session.closes()) + } +} diff --git a/cmd/rds-agent/engine_postgres.go b/cmd/rds-agent/engine_postgres.go new file mode 100644 index 000000000..681d9d4b2 --- /dev/null +++ b/cmd/rds-agent/engine_postgres.go @@ -0,0 +1,379 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// The engine is reached over its unix socket under peer authentication, so the +// agent drops to the postgres OS user rather than holding a password of its own. +type postgresEngine struct { + quiesceState + + // The control plane's own metadata for this engine, resolved once at + // startup: the live password apply runs as the cluster superuser, so it + // re-checks the role name against the same reserved set the API validates. + meta handlers_rds.Engine + run commandRunner + startSess sessionRunner + psql string + rcService string + service string + pgData string + socketDir string + osUser string + probe *engineProbe + // Set from the bootstrap config, on a different goroutine than the commands + // that read it. + port atomic.Int64 + + params parameterStore + // Serializes parameter installs, serving snapshots and rollback restores so + // none can copy or replace an intermediate configuration. + paramMu sync.Mutex + repairTimeout time.Duration + repairPoll time.Duration +} + +var _ engine = (*postgresEngine)(nil) + +// pg_isready is what gives PostgreSQL three states for free: exit 0 is serving, +// exit 1 is a postmaster up and rejecting connections, and anything else is an +// engine that is not there at all. Resolved on PATH, where the client package +// puts it. +const postgresProbeBinary = "pg_isready" + +func newPostgresProbe(cfg config, run probeRunner) *engineProbe { + return newEngineProbe(cfg.EnginePort, postgresProbeState(cfg.EngineHost, run)) +} + +func postgresProbeState(host string, run probeRunner) probeStateFn { + return func(ctx context.Context, port int64) (engineState, string) { + portArg := strconv.FormatInt(port, 10) + code, err := run(ctx, postgresProbeBinary, "-h", host, "-p", portArg, "-q") + switch { + case err != nil: + // A missing binary or broken image. Reporting healthy on the strength of + // nothing would hide it, so this reads as absent like an engine that did + // not answer. + return engineAbsent, fmt.Sprintf("engine probe could not run: %v", err) + case code == 0: + return engineServing, "" + case code == 1: + return engineRecovering, "engine is rejecting connections (startup or recovery)" + default: + return engineAbsent, fmt.Sprintf("engine did not respond on %s:%s", host, portArg) + } + } +} + +// Resolved once, under the name the control plane knows this implementation by. +// A lookup failure is a mismatch between the agent and the control plane it was +// built against, so it fails at startup rather than at the first rotation. +var postgresEngineMeta = mustLookupEngine(enginePostgres) + +func mustLookupEngine(name string) handlers_rds.Engine { + engine, err := handlers_rds.LookupEngine(name) + if err != nil { + panic("rds-agent: " + err.Error()) + } + return engine +} + +// The include the resolved parameter set is rendered to, and the copy of the +// last one the engine accepted. Both live beside the data rather than in /etc, +// matching rds-init: a class change boots a fresh root volume, which would +// otherwise revert them. +const ( + postgresParametersFile = "10-rds-parameters.conf" + // Deliberately not a .conf name: include_dir globs *.conf, so the rollback + // copy must not be read as a second set of settings. + postgresLastGoodFile = "10-rds-parameters.last-good" +) + +func newPostgresEngine(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) *postgresEngine { + e := &postgresEngine{ + meta: postgresEngineMeta, + run: run, + startSess: startSess, + psql: filepath.Join(cfg.EngineBinDir, "psql"), + rcService: cfg.RCService, + service: cfg.EngineService, + pgData: cfg.EngineDataDir, + socketDir: cfg.SocketDir, + osUser: cfg.EngineUser, + probe: probe, + params: parameterStore{ + dir: filepath.Join(cfg.EngineDataDir, "conf.d"), + installed: postgresParametersFile, + lastGood: postgresLastGoodFile, + osUser: cfg.EngineUser, + }, + repairTimeout: parameterRepairTimeout, + repairPoll: parameterRepairPoll, + } + e.port.Store(int64(cfg.EnginePort)) + return e +} + +func (e *postgresEngine) setPort(port int) { + e.port.Store(int64(port)) +} + +// The role name and the password ride the environment and are re-quoted by +// psql, so neither reaches a shell word or an argv another process can read. +func (e *postgresEngine) SetPassword(ctx context.Context, username, password string) error { + if username == "" || password == "" { + return fmt.Errorf("set-password requires both %s and %s", + handlers_rds.CommandParamMasterUsername, handlers_rds.CommandParamMasterUserPassword) + } + // The statement below runs as the cluster superuser over the socket under + // peer auth, so a reserved name in the command payload would hand the + // customer the bootstrap superuser rather than rotate their own role. + if err := e.meta.ValidateUsernameNotReserved(username); err != nil { + return fmt.Errorf("refusing to set the password of a role the engine reserves: %w", err) + } + // psql interpolates the password into the ALTER ROLE before the server sees + // it, and these three are what would write it to the log — the last on any + // failure at its own default. SUSET, so the parameter group cannot win. + const sql = `SET log_statement = 'none'; +SET log_min_duration_statement = -1; +SET log_min_error_statement = 'panic'; +\getenv master RDS_MASTER_USERNAME +\getenv password RDS_MASTER_PASSWORD +ALTER ROLE :"master" WITH LOGIN PASSWORD :'password'; +` + _, err := e.psqlRun(ctx, sql, + "RDS_MASTER_USERNAME="+username, + "RDS_MASTER_PASSWORD="+password, + ) + if err != nil { + // psql echoes the failing statement, which here would carry the new + // password back to the control plane and into the event ring. + return fmt.Errorf("apply the master password: %s", redact(err.Error(), password)) + } + return nil +} + +// Installs the resolved set, validates it with the engine's own config parser +// before the engine ever adopts it, and reloads. A value the engine refuses is +// rolled back here rather than left on the data volume, where it would survive +// every VM replace and turn the next restart into a boot loop. +func (e *postgresEngine) ApplyParameters(ctx context.Context, params []handlers_rds.Parameter) ([]string, error) { + e.paramMu.Lock() + defer e.paramMu.Unlock() + + initialState, _ := e.probe.state(ctx) + if initialState == engineRecovering { + return nil, errors.New("apply parameters while the engine is still starting or recovering") + } + if initialState == engineServing { + // A command can arrive before the first heartbeat. Seed the configuration + // currently being served before replacing its file in that window. + if err := e.recordServingParametersLocked(ctx); err != nil { + return nil, fmt.Errorf("record the parameters serving before the apply: %w", err) + } + } + + // The check has to run against the file in place, because the engine parses + // the datadir's own include_dir. The window that leaves is closed from both + // ends: the rollback below, and the last-known-good restore the agent runs at + // boot when the engine does not come up. + restore, err := e.params.install(params) + if err != nil { + return nil, err + } + if err := e.checkConfig(ctx); err != nil { + return nil, errors.Join(fmt.Errorf("the engine rejected the parameter set: %w", err), restore()) + } + + state, _ := e.probe.state(ctx) + switch state { + case engineAbsent: + return e.restartOnRepairSetLocked(ctx) + case engineRecovering: + return nil, errors.Join(errors.New("the engine entered startup or recovery during the parameter apply"), restore()) + } + + if _, err := e.psqlRun(ctx, "SELECT pg_reload_conf();\n"); err != nil { + // The engine may have gone down after the first probe. In that case start + // it on the new, parser-checked repair set rather than restoring the set + // that already left it unable to serve. + if state, _ := e.probe.state(ctx); state == engineAbsent { + return e.restartOnRepairSetLocked(ctx) + } + return nil, errors.Join(fmt.Errorf("reload the engine configuration: %w", err), restore()) + } + return e.pendingRestartParameters(ctx) +} + +func (e *postgresEngine) restartOnRepairSetLocked(ctx context.Context) ([]string, error) { + return awaitRepairedEngine(ctx, e.probe, e.Restart, e.pendingRestartParameters, e.repairTimeout, e.repairPoll) +} + +// The postmaster's own answer: a static setting it has stored but not adopted is +// what it reports here, so nothing in the guest has to classify one. +func (e *postgresEngine) pendingRestartParameters(ctx context.Context) ([]string, error) { + out, err := e.psqlRun(ctx, "SELECT name FROM pg_settings WHERE pending_restart ORDER BY name;\n") + if err != nil { + return nil, fmt.Errorf("read the settings pending a restart: %w", err) + } + var pending []string + for line := range strings.SplitSeq(out, "\n") { + if name := strings.TrimSpace(line); name != "" { + pending = append(pending, name) + } + } + return pending, nil +} + +// Snapshots the include only when the running postmaster has no settings still +// pending a restart. The parameter mutex keeps an apply from replacing the file +// between that check and the copy. +func (e *postgresEngine) RecordServingParameters(ctx context.Context) error { + e.paramMu.Lock() + defer e.paramMu.Unlock() + return e.recordServingParametersLocked(ctx) +} + +func (e *postgresEngine) recordServingParametersLocked(ctx context.Context) error { + return recordLastGood(ctx, e.params, e.pendingRestartParameters) +} + +// Puts the last set the engine accepted back in place, for a restart that failed +// after a parameter change. +func (e *postgresEngine) RestoreLastKnownGoodParameters(ctx context.Context) (bool, error) { + e.paramMu.Lock() + defer e.paramMu.Unlock() + return restoreLastGood(ctx, e.params, e.probe) +} + +// The engine's own parser, run offline against the datadir. Reading one setting +// back is enough: postgres parses postgresql.conf and every include first, and +// exits non-zero naming the file and line of an unknown parameter or a value +// outside its range. +func (e *postgresEngine) checkConfig(ctx context.Context) error { + if _, err := e.run(ctx, command{ + Name: filepath.Join(filepath.Dir(e.psql), "postgres"), + Args: []string{"-D", e.pgData, "-C", "shared_buffers"}, + Env: []string{"PATH=" + defaultGuestPath}, + User: e.osUser, + }); err != nil { + return err + } + return nil +} + +func (e *postgresEngine) Stop(ctx context.Context) error { + return serviceAction(ctx, e.run, e.rcService, e.service, "stop") +} + +// Only the parameter rollback calls this: a restart the control plane wants goes +// through RebootDBInstance, which cycles the VM. +func (e *postgresEngine) Restart(ctx context.Context) error { + return serviceAction(ctx, e.run, e.rcService, e.service, "restart") +} + +// Feeds sql on stdin rather than as an argument, so a statement is never +// visible in the process table. +func (e *postgresEngine) psqlRun(ctx context.Context, sql string, env ...string) (string, error) { + return e.run(ctx, command{ + Name: e.psql, + Args: e.psqlArgs(), + Env: append([]string{"PATH=" + defaultGuestPath}, env...), + Stdin: sql, + User: e.osUser, + }) +} + +// Reading the script from stdin is what lets one invocation serve both a +// one-shot run and a session held open across several statements. +func (e *postgresEngine) psqlArgs() []string { + return []string{ + "--no-psqlrc", "--quiet", "--no-align", "--tuples-only", + "-v", "ON_ERROR_STOP=1", + "-h", e.socketDir, + "-p", strconv.FormatInt(e.port.Load(), 10), + "-U", e.osUser, + "-d", "postgres", + "-f", "-", + } +} + +// Puts the engine into backup mode: the datadir is checkpointed and the engine +// stops writing over the pages a snapshot is about to read. The hold is released +// by Unquiesce, or by its own deadline, whichever comes first. +func (e *postgresEngine) Quiesce(ctx context.Context, label string, hold time.Duration) error { + if err := validateQuiesceRequest(label, hold); err != nil { + return err + } + + // Held across the whole start, so a second quiesce waits and then finds the + // first one's hold rather than opening a concurrent backup alongside it. + e.mu.Lock() + defer e.mu.Unlock() + if e.held != nil { + return fmt.Errorf("the engine is already quiesced for backup %s", e.held.label) + } + + // Deliberately not the command's context: the session has to outlive the call + // that started it, and its own deadline is what bounds it instead. + session, err := e.startSess(context.WithoutCancel(ctx), command{ + Name: e.psql, + Args: e.psqlArgs(), + Env: []string{"PATH=" + defaultGuestPath, "RDS_BACKUP_LABEL=" + label}, + User: e.osUser, + SentinelStatement: `\echo ` + sessionSentinel + "\n", + }) + if err != nil { + return fmt.Errorf("open a backup session: %w", err) + } + + // fast forces an immediate checkpoint rather than spreading it over the + // checkpoint interval, which would hold the snapshot open for minutes. + const sql = `\getenv label RDS_BACKUP_LABEL +SELECT pg_backup_start(:'label', fast => true); +` + if err := session.Exec(ctx, sql); err != nil { + if closeErr := session.Close(); closeErr != nil { + slog.Warn("rds-agent: closing a failed backup session", "err", closeErr) + } + return fmt.Errorf("put the engine into backup mode: %w", err) + } + + e.beginHoldLocked(label, session, hold) + return nil +} + +// Ends the backup cleanly. A missing hold is an error rather than a silent +// success: it means the deadline fired first, so the snapshot the control plane +// just took was not taken against a held checkpoint. +func (e *postgresEngine) Unquiesce(ctx context.Context) error { + held := e.takeHold() + if held == nil { + return errors.New("the engine is not quiesced; the backup hold had already expired") + } + + // The stop has to run on the session that started the backup, and the session + // is closed either way — the engine aborts an unstopped backup with it. + execErr := held.session.Exec(ctx, "SELECT pg_backup_stop();\n") + closeErr := held.session.Close() + if execErr != nil { + return fmt.Errorf("take the engine out of backup mode: %w", errors.Join(execErr, closeErr)) + } + if closeErr != nil { + return fmt.Errorf("close the backup session: %w", closeErr) + } + slog.Info("rds-agent: engine released from backup mode", "label", held.label) + return nil +} diff --git a/cmd/rds-agent/engine_test.go b/cmd/rds-agent/engine_test.go index c2f89fb2b..7a2f93406 100644 --- a/cmd/rds-agent/engine_test.go +++ b/cmd/rds-agent/engine_test.go @@ -372,7 +372,7 @@ func TestPostgresEngine_ApplyParametersRestartsOnARepairSetWhileEngineIsDown(t * if !strings.Contains(runner.calls[2].Stdin, "pending_restart") { t.Errorf("third call = %+v, want the pending-restart read", runner.calls[2]) } - installed, err := os.ReadFile(engine.parametersPath()) + installed, err := os.ReadFile(engine.params.installedPath()) if err != nil { t.Fatalf("read installed parameters: %v", err) } @@ -392,7 +392,7 @@ func TestPostgresEngine_ApplyParametersKeepsRepairSetWhenRestartTimesOut(t *test if err == nil || !strings.Contains(err.Error(), "wait for the engine") { t.Fatalf("ApplyParameters error = %v, want the repair wait failure", err) } - installed, readErr := os.ReadFile(engine.parametersPath()) + installed, readErr := os.ReadFile(engine.params.installedPath()) if readErr != nil { t.Fatalf("read installed parameters: %v", readErr) } @@ -400,24 +400,3 @@ func TestPostgresEngine_ApplyParametersKeepsRepairSetWhenRestartTimesOut(t *test t.Errorf("installed parameters = %q, want the checked repair set retained", installed) } } - -// Through the service manager, so the supervisor records the engine as stopped -// and does not restart it underneath a VM that is going down. -func TestPostgresEngine_StopGoesThroughTheServiceManager(t *testing.T) { - runner := &recordingRunner{} - engine := newTestEngine(t, runner.run) - - if err := engine.Stop(context.Background()); err != nil { - t.Fatalf("Stop: %v", err) - } - if len(runner.calls) != 1 { - t.Fatalf("ran %d commands, want 1", len(runner.calls)) - } - call := runner.calls[0] - if filepath.Base(call.Name) != "rc-service" { - t.Errorf("ran %q, want rc-service", call.Name) - } - if len(call.Args) != 2 || call.Args[0] != "postgresql" || call.Args[1] != "stop" { - t.Errorf("args = %v, want [postgresql stop]", call.Args) - } -} diff --git a/cmd/rds-agent/heartbeat_test.go b/cmd/rds-agent/heartbeat_test.go index c9ce6dc7f..8781d2a03 100644 --- a/cmd/rds-agent/heartbeat_test.go +++ b/cmd/rds-agent/heartbeat_test.go @@ -17,7 +17,7 @@ func TestHeartbeater_FirstServingProbeSeedsLastKnownGood(t *testing.T) { runner := &recordingRunner{} engine := newTestEngine(t, runner.run) content := []byte("work_mem = '4096'\n") - if err := os.WriteFile(engine.parametersPath(), content, 0o600); err != nil { + if err := os.WriteFile(engine.params.installedPath(), content, 0o600); err != nil { t.Fatalf("write installed parameters: %v", err) } @@ -25,7 +25,7 @@ func TestHeartbeater_FirstServingProbeSeedsLastKnownGood(t *testing.T) { h := newHeartbeater(cp, engine.probe, engine, 0) h.beat(context.Background()) - lastGood, err := os.ReadFile(engine.lastGoodPath()) + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read last known good: %v", err) } diff --git a/cmd/rds-agent/parameters.go b/cmd/rds-agent/parameters.go new file mode 100644 index 000000000..e014a19f1 --- /dev/null +++ b/cmd/rds-agent/parameters.go @@ -0,0 +1,213 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "os" + "path/filepath" + "time" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// The generated parameter files and what it takes to install one. They live on +// the data volume rather than under /etc, at the same paths rds-init writes them +// to: a class change is a VM replacement, so a set under /etc would be discarded +// with the boot volume and the instance would silently revert to catalog +// defaults while the API still reported the customer's values. +type parameterStore struct { + dir string + // The include the resolved set is rendered to. + installed string + // The copy of the last set the engine accepted, which a boot loop is rolled + // back to. Deliberately not a name the engine's own include glob reads, so it + // is never parsed as a second set of settings. + lastGood string + // What the engine actually started on, for an engine that cannot be asked + // which of its settings are still pending a restart. Empty for one that can. + serving string + // Prepended to every file written here, for a configuration format that has + // no setting outside a group. + header string + // The engine parses its own configuration, so a root-owned file would be + // unreadable to it. + osUser string +} + +func (s parameterStore) installedPath() string { return filepath.Join(s.dir, s.installed) } +func (s parameterStore) lastGoodPath() string { return filepath.Join(s.dir, s.lastGood) } +func (s parameterStore) servingPath() string { return filepath.Join(s.dir, s.serving) } + +// Byte-for-byte what rds-init writes for the same set, so the copies compared +// against it are comparing values rather than formatting. +func (s parameterStore) render(params []handlers_rds.Parameter) []byte { + return []byte(s.header + renderParameters(params)) +} + +// Installs the resolved set at the path rds-init uses, so a later boot +// overwrites rather than shadows it. Returns a func putting back whatever was +// there, so the caller can undo the install without re-rendering anything. +func (s parameterStore) install(params []handlers_rds.Parameter) (restore func() error, err error) { + path := s.installedPath() + // A missing file is the first-ever apply, and its rollback is a removal. + previous, readErr := os.ReadFile(path) + if readErr != nil && !os.IsNotExist(readErr) { + return nil, fmt.Errorf("read the installed parameters at %s: %w", path, readErr) + } + existed := readErr == nil + + if err := s.write(path, s.render(params)); err != nil { + return nil, err + } + return func() error { + if !existed { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("withdraw the rejected parameters at %s: %w", path, err) + } + return nil + } + return s.write(path, previous) + }, nil +} + +// Atomic: a temp file in the same directory, renamed over the target. The temp +// name deliberately does not end in the engine's include suffix, so a crash +// between write and rename cannot leave the engine reading a half-written file. +func (s parameterStore) write(path string, content []byte) error { + tmp := path + ".new" + if err := os.WriteFile(tmp, content, 0o600); err != nil { + return fmt.Errorf("write %s: %w", tmp, err) + } + if err := s.chownToEngine(tmp); err != nil { + os.Remove(tmp) + return err + } + if err := os.Rename(tmp, path); err != nil { + os.Remove(tmp) + return fmt.Errorf("install %s: %w", path, err) + } + return nil +} + +// A guest with no engine user is a broken image, not something to work around. +func (s parameterStore) chownToEngine(path string) error { + credential, err := lookupCredential(s.osUser) + if err != nil { + return err + } + if err := os.Chown(path, int(credential.Uid), int(credential.Gid)); err != nil { + return fmt.Errorf("hand %s to %s: %w", path, s.osUser, err) + } + return nil +} + +// Snapshots the installed set as the rollback target, but only once the engine +// is running all of it. A set with a setting still pending a restart has not +// been shown to serve, and promoting it would leave the rollback pointing at the +// configuration that is about to fail. +func recordLastGood(ctx context.Context, store parameterStore, pending pendingRestartFn) error { + if _, err := os.Stat(store.installedPath()); err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("inspect the serving parameters: %w", err) + } + names, err := pending(ctx) + if err != nil { + return err + } + if len(names) > 0 { + return nil + } + + content, err := os.ReadFile(store.installedPath()) + if err != nil { + return fmt.Errorf("read the serving parameters: %w", err) + } + lastGood, err := os.ReadFile(store.lastGoodPath()) + if err == nil && bytes.Equal(content, lastGood) { + return nil + } + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("read the last known good parameters: %w", err) + } + return store.write(store.lastGoodPath(), content) +} + +// Puts the last set the engine accepted back in place, for a restart that failed +// after a parameter change. The probe is checked again under the caller's +// parameter lock, so a repair that just brought the engine back is not reversed. +func restoreLastGood(ctx context.Context, store parameterStore, probe *engineProbe) (bool, error) { + if state, _ := probe.state(ctx); state != engineAbsent { + return false, nil + } + lastGood, err := os.ReadFile(store.lastGoodPath()) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, fmt.Errorf("read the last known good parameters: %w", err) + } + current, err := os.ReadFile(store.installedPath()) + if err != nil && !os.IsNotExist(err) { + return false, fmt.Errorf("read the installed parameters: %w", err) + } + if bytes.Equal(current, lastGood) { + return false, nil + } + if err := store.write(store.installedPath(), lastGood); err != nil { + return false, err + } + return true, nil +} + +// The settings the engine has accepted but will not honour until it restarts. +// Each implementation answers it its own way, and both the apply and the serving +// snapshot are gated on the same answer. +type pendingRestartFn func(ctx context.Context) ([]string, error) + +const ( + parameterRepairTimeout = 90 * time.Second + parameterRepairPoll = time.Second +) + +// Starts a down engine on the set that was just installed and waits for it to +// serve. Reached when the engine went away during an apply: the installed set is +// the newer of the two, so starting on it beats restoring one that already left +// the engine unable to serve. +func awaitRepairedEngine(ctx context.Context, probe *engineProbe, restart func(context.Context) error, + pending pendingRestartFn, timeout, poll time.Duration) ([]string, error) { + repairCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + if err := restart(repairCtx); err != nil { + return nil, fmt.Errorf("start the engine on the repaired parameter set: %w", err) + } + + timer := time.NewTimer(0) + defer timer.Stop() + lastMessage := "engine did not respond" + for { + select { + case <-repairCtx.Done(): + if ctx.Err() != nil { + return nil, fmt.Errorf("wait for the engine on the repaired parameter set: %w", ctx.Err()) + } + return nil, fmt.Errorf("wait for the engine on the repaired parameter set: %s: %w", lastMessage, repairCtx.Err()) + case <-timer.C: + } + + state, message := probe.state(repairCtx) + if state == engineServing { + names, err := pending(repairCtx) + if err == nil { + return names, nil + } + lastMessage = err.Error() + } else if message != "" { + lastMessage = message + } + timer.Reset(poll) + } +} diff --git a/cmd/rds-agent/paramrecovery_test.go b/cmd/rds-agent/paramrecovery_test.go index b438358e7..046fe7821 100644 --- a/cmd/rds-agent/paramrecovery_test.go +++ b/cmd/rds-agent/paramrecovery_test.go @@ -49,7 +49,7 @@ func TestPostgresEngine_ApplyParametersRollsBackAValueTheEngineRejects(t *testin t.Fatal("ApplyParameters succeeded against a config the engine rejected") } - installed, err := os.ReadFile(engine.parametersPath()) + installed, err := os.ReadFile(engine.params.installedPath()) if err != nil { t.Fatalf("read the installed parameters: %v", err) } @@ -68,7 +68,7 @@ func TestPostgresEngine_ApplyParametersWithdrawsTheFirstRejectedSet(t *testing.T []handlers_rds.Parameter{{Name: "work_mem", Value: "0"}}); err == nil { t.Fatal("ApplyParameters succeeded against a config the engine rejected") } - if _, err := os.Stat(engine.parametersPath()); !os.IsNotExist(err) { + if _, err := os.Stat(engine.params.installedPath()); !os.IsNotExist(err) { t.Errorf("the rejected include is still installed (stat err = %v)", err) } } @@ -96,7 +96,7 @@ func TestPostgresEngine_ApplyBeforeFirstHeartbeatSeedsLastKnownGood(t *testing.T runner := &pendingAfterReloadRunner{} engine := newTestEngine(t, runner.run) first := []byte("shared_buffers = '32768'\n") - if err := os.WriteFile(engine.parametersPath(), first, 0o600); err != nil { + if err := os.WriteFile(engine.params.installedPath(), first, 0o600); err != nil { t.Fatalf("write serving parameter set: %v", err) } @@ -107,7 +107,7 @@ func TestPostgresEngine_ApplyBeforeFirstHeartbeatSeedsLastKnownGood(t *testing.T if !slices.Equal(pending, []string{"shared_buffers"}) { t.Errorf("pending = %v, want shared_buffers", pending) } - lastGood, err := os.ReadFile(engine.lastGoodPath()) + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read last known good: %v", err) } @@ -152,10 +152,10 @@ func TestPostgresEngine_ServingSnapshotWaitsForParameterApply(t *testing.T) { } engine := newTestEngine(t, runner.run) first := []byte("shared_buffers = '32768'\n") - if err := os.WriteFile(engine.parametersPath(), first, 0o600); err != nil { + if err := os.WriteFile(engine.params.installedPath(), first, 0o600); err != nil { t.Fatalf("write serving parameter set: %v", err) } - if err := os.WriteFile(engine.lastGoodPath(), first, 0o600); err != nil { + if err := os.WriteFile(engine.params.lastGoodPath(), first, 0o600); err != nil { t.Fatalf("write last known good: %v", err) } @@ -191,7 +191,7 @@ func TestPostgresEngine_ServingSnapshotWaitsForParameterApply(t *testing.T) { t.Error("serving snapshot completed while the parameter apply held its transaction") } - lastGood, err := os.ReadFile(engine.lastGoodPath()) + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read last known good: %v", err) } @@ -215,7 +215,7 @@ func TestPostgresEngine_LastKnownGoodTracksTheServingSet(t *testing.T) { } assertLastGood := func(value string) { t.Helper() - lastGood, err := os.ReadFile(engine.lastGoodPath()) + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read the last known good parameters: %v", err) } @@ -239,8 +239,8 @@ func TestPostgresEngine_LastKnownGoodTracksTheServingSet(t *testing.T) { apply("16384") assertLastGood("8192") - if strings.HasSuffix(engine.lastGoodPath(), ".conf") { - t.Errorf("the rollback copy at %s would be included by the engine", engine.lastGoodPath()) + if strings.HasSuffix(engine.params.lastGoodPath(), ".conf") { + t.Errorf("the rollback copy at %s would be included by the engine", engine.params.lastGoodPath()) } } @@ -248,7 +248,7 @@ func TestPostgresEngine_RecordServingParametersSkipsPendingRestart(t *testing.T) runner := &recordingRunner{} engine := newTestEngine(t, runner.run) first := []byte("shared_buffers = '32768'\n") - if err := os.WriteFile(engine.parametersPath(), first, 0o600); err != nil { + if err := os.WriteFile(engine.params.installedPath(), first, 0o600); err != nil { t.Fatalf("write first parameter set: %v", err) } if err := engine.RecordServingParameters(t.Context()); err != nil { @@ -256,14 +256,14 @@ func TestPostgresEngine_RecordServingParametersSkipsPendingRestart(t *testing.T) } second := []byte("shared_buffers = '65536'\n") - if err := os.WriteFile(engine.parametersPath(), second, 0o600); err != nil { + if err := os.WriteFile(engine.params.installedPath(), second, 0o600); err != nil { t.Fatalf("write pending parameter set: %v", err) } runner.out = "shared_buffers\n" if err := engine.RecordServingParameters(t.Context()); err != nil { t.Fatalf("record with a pending restart: %v", err) } - lastGood, err := os.ReadFile(engine.lastGoodPath()) + lastGood, err := os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read last known good: %v", err) } @@ -277,7 +277,7 @@ func TestPostgresEngine_RecordServingParametersSkipsPendingRestart(t *testing.T) if err := engine.RecordServingParameters(t.Context()); err != nil { t.Fatalf("record the restarted parameter set: %v", err) } - lastGood, err = os.ReadFile(engine.lastGoodPath()) + lastGood, err = os.ReadFile(engine.params.lastGoodPath()) if err != nil { t.Fatalf("read advanced last known good: %v", err) } @@ -329,7 +329,7 @@ func TestPostgresEngine_RestoreLastKnownGoodParameters(t *testing.T) { if !restored { t.Fatal("did not restore a set that differs from the last known good") } - installed, err := os.ReadFile(engine.parametersPath()) + installed, err := os.ReadFile(engine.params.installedPath()) if err != nil { t.Fatalf("read the installed parameters: %v", err) } diff --git a/cmd/rds-agent/quiesce.go b/cmd/rds-agent/quiesce.go index 4353455a4..a87c128a9 100644 --- a/cmd/rds-agent/quiesce.go +++ b/cmd/rds-agent/quiesce.go @@ -15,10 +15,11 @@ import ( "time" ) -// PostgreSQL's non-exclusive backup API ties the backup to the session that -// started it: when that session ends, for any reason, the backup is aborted. -// That is what makes the hold self-expiring — a control plane that dies mid -// snapshot cannot leave the engine in backup mode forever. +// Both engines tie a backup hold to the session that took it: PostgreSQL aborts +// a non-exclusive backup when its session ends, and MariaDB releases a BACKUP +// STAGE with the connection that started it. That is what makes the hold +// self-expiring — a control plane that dies mid snapshot cannot leave the engine +// in backup mode forever. type engineSession interface { // Runs sql and waits for the engine to finish it. Exec(ctx context.Context, sql string) error @@ -31,7 +32,8 @@ type engineSession interface { type sessionRunner func(ctx context.Context, c command) (engineSession, error) // Written after every statement so the reader knows the engine is done with it. -// The statements this carries return LSNs, so it cannot appear in a result. +// The statements this carries return LSNs and backup stages, so it cannot appear +// in a result. const sessionSentinel = "--rds-agent-ready--" // The backup mode currently held, and the timer that ends it regardless of what @@ -42,98 +44,65 @@ type quiesceHold struct { expiry *time.Timer } -// Puts the engine into backup mode: the datadir is checkpointed and the engine -// stops writing over the pages a snapshot is about to read. The hold is released -// by Unquiesce, or by its own deadline, whichever comes first. -func (e *postgresEngine) Quiesce(ctx context.Context, label string, hold time.Duration) error { +// The hold bookkeeping, which is the same for both engines: only the statements +// that take and release it differ, not the single-hold rule or the deadline that +// releases it whatever happens to the caller. +type quiesceState struct { + // Guarded because the expiry timer releases the hold on its own goroutine. + mu sync.Mutex + held *quiesceHold +} + +// The two things every quiesce needs, checked before a session is opened so a +// malformed command costs the engine nothing. +func validateQuiesceRequest(label string, hold time.Duration) error { if label == "" { return errors.New("quiesce requires a backup label") } if hold <= 0 { return errors.New("quiesce requires a positive hold deadline") } + return nil +} - // Held across the whole start, so a second quiesce waits and then finds the - // first one's hold rather than opening a concurrent backup alongside it. - e.mu.Lock() - defer e.mu.Unlock() - if e.held != nil { - return fmt.Errorf("the engine is already quiesced for backup %s", e.held.label) - } - - // Deliberately not the command's context: the session has to outlive the call - // that started it, and its own deadline is what bounds it instead. - session, err := e.startSess(context.WithoutCancel(ctx), command{ - Name: e.psql, - Args: e.psqlArgs(), - Env: []string{"PATH=" + defaultGuestPath, "RDS_BACKUP_LABEL=" + label}, - Stdin: "", - User: e.osUser, - }) - if err != nil { - return fmt.Errorf("open a backup session: %w", err) - } - - // fast forces an immediate checkpoint rather than spreading it over the - // checkpoint interval, which would hold the snapshot open for minutes. - const sql = `\getenv label RDS_BACKUP_LABEL -SELECT pg_backup_start(:'label', fast => true); -` - if err := session.Exec(ctx, sql); err != nil { - if closeErr := session.Close(); closeErr != nil { - slog.Warn("rds-agent: closing a failed backup session", "err", closeErr) - } - return fmt.Errorf("put the engine into backup mode: %w", err) - } - - e.held = &quiesceHold{ +// Records the hold and arms its deadline. Called with mu held, so a second +// quiesce waits and then finds this hold rather than opening a concurrent +// backup alongside it. +func (q *quiesceState) beginHoldLocked(label string, session engineSession, hold time.Duration) { + q.held = &quiesceHold{ label: label, session: session, - expiry: time.AfterFunc(hold, func() { e.expireQuiesce(label) }), + expiry: time.AfterFunc(hold, func() { q.expire(label) }), } slog.Info("rds-agent: engine quiesced for backup", "label", label, "hold", hold) - return nil } -// Ends the backup cleanly. A missing hold is an error rather than a silent -// success: it means the deadline fired first, so the snapshot the control plane -// just took was not taken against a held checkpoint. -func (e *postgresEngine) Unquiesce(ctx context.Context) error { - e.mu.Lock() - held := e.held - e.held = nil - e.mu.Unlock() +// Takes the hold off the engine, if one is still there, and disarms its +// deadline. The caller ends the session, because the statement that releases the +// backup has to run on it first. +func (q *quiesceState) takeHold() *quiesceHold { + q.mu.Lock() + defer q.mu.Unlock() - if held == nil { - return errors.New("the engine is not quiesced; the backup hold had already expired") + held := q.held + q.held = nil + if held != nil { + held.expiry.Stop() } - held.expiry.Stop() - - // The stop has to run on the session that started the backup, and the session - // is closed either way — the engine aborts an unstopped backup with it. - execErr := held.session.Exec(ctx, "SELECT pg_backup_stop();\n") - closeErr := held.session.Close() - if execErr != nil { - return fmt.Errorf("take the engine out of backup mode: %w", errors.Join(execErr, closeErr)) - } - if closeErr != nil { - return fmt.Errorf("close the backup session: %w", closeErr) - } - slog.Info("rds-agent: engine released from backup mode", "label", held.label) - return nil + return held } -// The deadline. Ending the session is enough — the engine aborts the backup it -// was holding — so this never has to reach the engine itself. -func (e *postgresEngine) expireQuiesce(label string) { - e.mu.Lock() - held := e.held +// The deadline. Ending the session is enough — both engines release the hold +// with it — so this never has to reach the engine itself. +func (q *quiesceState) expire(label string) { + q.mu.Lock() + held := q.held if held == nil || held.label != label { - e.mu.Unlock() + q.mu.Unlock() return } - e.held = nil - e.mu.Unlock() + q.held = nil + q.mu.Unlock() slog.Warn("rds-agent: backup hold expired; releasing the engine", "label", label) if err := held.session.Close(); err != nil { @@ -141,21 +110,27 @@ func (e *postgresEngine) expireQuiesce(label string) { } } -// A psql child with its stdin held open, so statements can be fed to it one at a -// time and the engine keeps seeing a single session. -type psqlSession struct { +// A client child with its stdin held open, so statements can be fed to it one at +// a time and the engine keeps seeing a single session. +type clientSession struct { cmd *exec.Cmd stdin io.WriteCloser stdout *bufio.Reader stderr *bytes.Buffer + // The statement that makes this client print the sentinel, which is the only + // part of a held session that differs between engines. + sentinel string mu sync.Mutex closed bool } -var _ engineSession = (*psqlSession)(nil) +var _ engineSession = (*clientSession)(nil) func execSessionRunner(ctx context.Context, c command) (engineSession, error) { + if c.SentinelStatement == "" { + return nil, fmt.Errorf("%s was started as a session with no sentinel statement", c.Name) + } cmd := exec.CommandContext(ctx, c.Name, c.Args...) cmd.Env = c.Env if c.User != "" { @@ -180,24 +155,30 @@ func execSessionRunner(ctx context.Context, c command) (engineSession, error) { if err := cmd.Start(); err != nil { return nil, fmt.Errorf("start %s: %w", c.Name, err) } - return &psqlSession{cmd: cmd, stdin: stdin, stdout: bufio.NewReader(stdout), stderr: &stderr}, nil + return &clientSession{ + cmd: cmd, + stdin: stdin, + stdout: bufio.NewReader(stdout), + stderr: &stderr, + sentinel: c.SentinelStatement, + }, nil } -// Feeds sql and waits for the sentinel that follows it. ON_ERROR_STOP makes psql -// exit on a failed statement, so a sentinel that never arrives is the failure -// itself and the engine's own message is on stderr. -func (s *psqlSession) Exec(ctx context.Context, sql string) error { +// Feeds sql and waits for the sentinel that follows it. Both clients are run in +// a mode that exits on a failed statement, so a sentinel that never arrives is +// the failure itself and the engine's own message is on stderr. +func (s *clientSession) Exec(ctx context.Context, sql string) error { s.mu.Lock() defer s.mu.Unlock() if s.closed { return errors.New("the backup session is closed") } - if _, err := io.WriteString(s.stdin, sql+"\\echo "+sessionSentinel+"\n"); err != nil { + if _, err := io.WriteString(s.stdin, sql+s.sentinel); err != nil { return fmt.Errorf("%w: %s", err, strings.TrimSpace(s.stderr.String())) } - // Read on its own goroutine so a psql that never answers costs the caller its - // deadline rather than blocking it forever. + // Read on its own goroutine so a client that never answers costs the caller + // its deadline rather than blocking it forever. done := make(chan error, 1) go func() { done <- s.awaitSentinel() }() select { @@ -208,7 +189,7 @@ func (s *psqlSession) Exec(ctx context.Context, sql string) error { } } -func (s *psqlSession) awaitSentinel() error { +func (s *clientSession) awaitSentinel() error { for { line, err := s.stdout.ReadString('\n') if strings.TrimSpace(line) == sessionSentinel { @@ -226,7 +207,7 @@ func (s *psqlSession) awaitSentinel() error { // Closing stdin ends the script, and the wait reaps the child. A session that // will not end is killed, because leaving it is what would keep the engine in // backup mode. -func (s *psqlSession) Close() error { +func (s *clientSession) Close() error { s.mu.Lock() defer s.mu.Unlock() if s.closed { diff --git a/cmd/rds-agent/quiesce_test.go b/cmd/rds-agent/quiesce_test.go index f6d45ab27..a92f14822 100644 --- a/cmd/rds-agent/quiesce_test.go +++ b/cmd/rds-agent/quiesce_test.go @@ -2,7 +2,6 @@ package main import ( "context" - "errors" "slices" "strings" "sync" @@ -113,81 +112,6 @@ func TestPostgresEngine_UnquiesceStopsTheBackupAndEndsTheSession(t *testing.T) { } } -// A hold that expired mid-snapshot means the snapshot was not taken against a -// held checkpoint, so the release reports it rather than succeeding silently. -func TestPostgresEngine_UnquiesceWithoutAHoldIsAnError(t *testing.T) { - engine, _ := newQuiesceEngine(t, &fakeSession{}) - - err := engine.Unquiesce(context.Background()) - if err == nil { - t.Fatal("Unquiesce succeeded with no hold to release") - } - if !strings.Contains(err.Error(), "expired") { - t.Errorf("error %q does not say the hold had expired", err) - } -} - -// The engine aborts the backup when the session ends, which is what makes the -// hold self-expiring: a control plane that dies cannot leave it held forever. -func TestPostgresEngine_QuiesceReleasesItselfOnItsDeadline(t *testing.T) { - session := &fakeSession{} - engine, _ := newQuiesceEngine(t, session) - - if err := engine.Quiesce(context.Background(), "orders-db-pre-upgrade", 10*time.Millisecond); err != nil { - t.Fatalf("Quiesce: %v", err) - } - deadline := time.Now().Add(2 * time.Second) - for session.closes() == 0 && time.Now().Before(deadline) { - time.Sleep(time.Millisecond) - } - - if session.closes() != 1 { - t.Fatal("the backup session outlived its deadline") - } - // The hold is gone with it, so a later release is honest about that. - if err := engine.Unquiesce(context.Background()); err == nil { - t.Error("Unquiesce succeeded after the hold expired") - } -} - -// One backup at a time: a second start would open a concurrent backup alongside -// the first rather than extending it. -func TestPostgresEngine_QuiesceRefusesASecondHold(t *testing.T) { - session := &fakeSession{} - engine, started := newQuiesceEngine(t, session) - if err := engine.Quiesce(context.Background(), "first", time.Minute); err != nil { - t.Fatalf("Quiesce: %v", err) - } - - err := engine.Quiesce(context.Background(), "second", time.Minute) - if err == nil { - t.Fatal("a second quiesce was accepted while the first was held") - } - if !strings.Contains(err.Error(), "first") { - t.Errorf("error %q does not name the backup already holding the engine", err) - } - if len(*started) != 1 { - t.Errorf("started %d sessions, want the second refused before opening one", len(*started)) - } -} - -// A failed start leaves nothing held, so a retry is not refused by a hold that -// never took effect. -func TestPostgresEngine_QuiesceEndsTheSessionWhenTheBackupWillNotStart(t *testing.T) { - session := &fakeSession{execErr: errors.New(`ERROR: a backup is already in progress`)} - engine, _ := newQuiesceEngine(t, session) - - if err := engine.Quiesce(context.Background(), "orders-db-pre-upgrade", time.Minute); err == nil { - t.Fatal("Quiesce succeeded against an engine that refused the backup") - } - if session.closes() != 1 { - t.Errorf("closed %d times, want the failed session ended", session.closes()) - } - if engine.held != nil { - t.Error("a failed quiesce left a hold behind") - } -} - func TestCommandRegistry_QuiesceCarriesTheLabelAndDeadline(t *testing.T) { engine := &fakeEngine{} reply := newCommander(nil, newCommandRegistry(engine, &fakeStorage{}), 0).execute(context.Background(), diff --git a/scripts/images/rds-mariadb/setup.sh b/scripts/images/rds-mariadb/setup.sh index 0152a72a7..c07c437dc 100644 --- a/scripts/images/rds-mariadb/setup.sh +++ b/scripts/images/rds-mariadb/setup.sh @@ -71,6 +71,18 @@ grep -q '^rc_need="rds-init"' /etc/conf.d/mariadb || { exit 1 } +# rds-agent's health probe reads this pid to tell a server replaying its redo log +# from one that is not running at all. Nothing generated can move the path: the +# packaged service passes --pid-file on mysqld_safe's command line, which beats +# any option file. A package that renamed it would leave every instance reporting +# an engine that is not there, and the rollback guard restarting a healthy one. +grep -q '^pidfile="/run/mysqld/\$RC_SVCNAME.pid"$' /etc/init.d/mariadb && + grep -q -- '--pid-file=\$pidfile' /etc/init.d/mariadb || { + echo "[rds-mariadb-setup] /etc/init.d/mariadb no longer starts the engine with its pidfile at /run/mysqld/mariadb.pid" + grep -nE 'pidfile|command_args' /etc/init.d/mariadb + exit 1 +} + # Bind /dev/console to the serial port so userspace boot output reaches ttyS0, # which the orchestrator captures host-side. Linux makes the last console= the # controlling one, and stock Alpine lists tty0 last — reorder so ttyS0 wins.