gsadmin/spinifex
1
0
Fork 0
mirror of https://github.com/mulgadc/spinifex.git synced 2026-10-21 11:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement the MariaDB guest engine behind the shared seam

Browse Source 
Adds mariadbEngine alongside the PostgreSQL one: a three-stage probe that
separates a live-but-unreachable server from an absent one, ALTER USER password
rotation on stdin behind a session logging guard, a dynamic SET GLOBAL apply
with rollback of the installed file, file-comparison pending-restart against a
serving copy, last-known-good restore, service stop and restart, and BACKUP
STAGE quiesce on a held session with a bounded lock_wait_timeout.

The parameter-file machinery, service actions, quiesce hold bookkeeping and
client session are extracted so both engines share them, and a contract suite
runs the same cases against both through the interfaces. MariaDB is not
registered in the agent's engine table yet, so construction still refuses.
This commit is contained in:
Julian Sommer
2026-08-11 14:42:17 +10:00
parent 32ad16a24d
commit f5370a5d37
14 changed files with 2404 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/rds-agent/config.go
+8
View File
@@ -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
}
cmd/rds-agent/config_test.go
+47 -2
View File
@@ -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")
cmd/rds-agent/engine.go
+55 -468
View File
@@ -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
cmd/rds-agent/engine_contract_test.go
+329
View File
@@ -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)
}
}
})
}
cmd/rds-agent/engine_mariadb.go
+588
View File
@@ -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,
}
}
cmd/rds-agent/engine_mariadb_test.go
+682
View File
@@ -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())
}
}
cmd/rds-agent/engine_postgres.go
+379
View File
@@ -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
}
cmd/rds-agent/engine_test.go
+2 -23
View File
@@ -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)
}
}
cmd/rds-agent/heartbeat_test.go
+2 -2
View File
@@ -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)
}
cmd/rds-agent/parameters.go
+213
View File
@@ -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)
}
}
cmd/rds-agent/paramrecovery_test.go
+15 -15
View File
@@ -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)
}
cmd/rds-agent/quiesce.go
+72 -91
View File
@@ -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 {
cmd/rds-agent/quiesce_test.go
-76
View File
@@ -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(),
scripts/images/rds-mariadb/setup.sh
+12
View File
@@ -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.