From fc6e055ca09390bb1e1a52ec72a5ecf4c17e66a4 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Wed, 29 Jul 2026 14:01:02 +1000 Subject: [PATCH] =?UTF-8?q?feat:=20RDS=20lifecycle=20core=20=E2=80=94=20re?= =?UTF-8?q?boot,=20start,=20stop,=20delete,=20events=20(#596)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/rds-agent/agent.go | 4 +- cmd/rds-agent/bootstrap.go | 1 + cmd/rds-agent/command.go | 38 +- cmd/rds-agent/command_test.go | 4 +- cmd/rds-agent/config.go | 43 ++ cmd/rds-agent/engine.go | 249 +++++++++++ cmd/rds-agent/engine_test.go | 268 +++++++++++ spinifex/awserrors/awserrors.go | 4 + spinifex/daemon/daemon.go | 5 + spinifex/daemon/rds_deps.go | 2 + spinifex/gateway/rds/handler.go | 10 +- spinifex/gateway/rds/handler_test.go | 7 + spinifex/gateway/rds/instance.go | 22 + spinifex/handlers/rds/commands.go | 181 ++++++++ spinifex/handlers/rds/commands_test.go | 77 ++++ spinifex/handlers/rds/create.go | 1 + spinifex/handlers/rds/create_test.go | 12 +- spinifex/handlers/rds/delete.go | 395 ++++++++++++++++ spinifex/handlers/rds/delete_test.go | 312 +++++++++++++ spinifex/handlers/rds/describe.go | 1 + spinifex/handlers/rds/events.go | 314 +++++++++++++ spinifex/handlers/rds/events_test.go | 178 ++++++++ spinifex/handlers/rds/launch_test.go | 7 + spinifex/handlers/rds/lifecycle.go | 359 +++++++++++++++ spinifex/handlers/rds/lifecycle_test.go | 547 +++++++++++++++++++++++ spinifex/handlers/rds/reconciler.go | 116 ++++- spinifex/handlers/rds/reconciler_test.go | 10 +- spinifex/handlers/rds/records.go | 82 ++++ spinifex/handlers/rds/service.go | 5 + spinifex/handlers/rds/service_nats.go | 33 ++ spinifex/handlers/rds/store.go | 15 + spinifex/handlers/rds/subjects.go | 9 + spinifex/handlers/rds/validate.go | 5 +- 33 files changed, 3285 insertions(+), 31 deletions(-) create mode 100644 cmd/rds-agent/engine.go create mode 100644 cmd/rds-agent/engine_test.go create mode 100644 spinifex/handlers/rds/commands.go create mode 100644 spinifex/handlers/rds/commands_test.go create mode 100644 spinifex/handlers/rds/delete.go create mode 100644 spinifex/handlers/rds/delete_test.go create mode 100644 spinifex/handlers/rds/events.go create mode 100644 spinifex/handlers/rds/events_test.go create mode 100644 spinifex/handlers/rds/lifecycle.go create mode 100644 spinifex/handlers/rds/lifecycle_test.go diff --git a/cmd/rds-agent/agent.go b/cmd/rds-agent/agent.go index bac0708cc..bdce27cc1 100644 --- a/cmd/rds-agent/agent.go +++ b/cmd/rds-agent/agent.go @@ -152,6 +152,7 @@ type Agent struct { id identity cp controlPlane probe *engineProbe + engine *postgresEngine handoffWriter func(string, *handlers_rds.GetDBBootstrapConfigOutput) error hb *heartbeater @@ -167,8 +168,9 @@ func newAgent(cfg config, cp controlPlane, probe *engineProbe) *Agent { EngineVersion: cfg.EngineVersion, } a := &Agent{cfg: cfg, id: id, cp: cp, probe: probe, handoffWriter: writeHandoff} + a.engine = newPostgresEngine(cfg, execCommandRunner) a.hb = newHeartbeater(cp, probe, handlers_rds.HeartbeatInterval) - a.cmd = newCommander(cp, newCommandRegistry(), cfg.PollWait) + a.cmd = newCommander(cp, newCommandRegistry(a.engine), cfg.PollWait) return a } diff --git a/cmd/rds-agent/bootstrap.go b/cmd/rds-agent/bootstrap.go index 510e9e9c1..5e9a81638 100644 --- a/cmd/rds-agent/bootstrap.go +++ b/cmd/rds-agent/bootstrap.go @@ -40,6 +40,7 @@ func (a *Agent) bootstrap(ctx context.Context) error { } if cfg.Port > 0 { a.probe.setPort(int(cfg.Port)) + a.engine.setPort(int(cfg.Port)) } // Mode is logged but never branched on: rds-init decides whether to initdb // from the state of the datadir. diff --git a/cmd/rds-agent/command.go b/cmd/rds-agent/command.go index c3a8a2c45..10ba37a45 100644 --- a/cmd/rds-agent/command.go +++ b/cmd/rds-agent/command.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "strings" "time" handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" @@ -17,10 +18,39 @@ type commandHandler func(ctx context.Context, cmd handlers_rds.Command) (string, // apply, parameter reload, storage grow, snapshot quiesce — add an entry each. type commandRegistry map[string]commandHandler -// Currently empty; an unregistered type is replied to as unsupported, so a -// control plane ahead of the guest gets a clear answer rather than a timeout. -func newCommandRegistry() commandRegistry { - return commandRegistry{} +// An unregistered type is replied to as unsupported, so a control plane ahead +// of the guest gets a clear answer rather than a timeout. +func newCommandRegistry(engine engineOps) commandRegistry { + return commandRegistry{ + handlers_rds.CommandSetPassword: func(ctx context.Context, cmd handlers_rds.Command) (string, error) { + params := commandParams(cmd) + return "", engine.SetPassword(ctx, + params[handlers_rds.CommandParamMasterUsername], + params[handlers_rds.CommandParamMasterUserPassword]) + }, + handlers_rds.CommandApplyParams: func(ctx context.Context, cmd handlers_rds.Command) (string, error) { + pending, err := engine.ApplyParameters(ctx, cmd.Parameters) + if err != nil { + return "", err + } + // The reply carries no structured payload, so the settings still + // awaiting a restart come back as the message itself. + return strings.Join(pending, ","), nil + }, + handlers_rds.CommandStopEngine: func(ctx context.Context, cmd handlers_rds.Command) (string, error) { + return "", engine.Stop(ctx) + }, + } +} + +// A later value wins, so a malformed command carrying a parameter twice cannot +// leave the handler acting on the first of two conflicting values. +func commandParams(cmd handlers_rds.Command) map[string]string { + params := make(map[string]string, len(cmd.Parameters)) + for _, p := range cmd.Parameters { + params[p.Name] = p.Value + } + return params } // Spaces retries after a failed poll, so a broken channel is not re-polled at diff --git a/cmd/rds-agent/command_test.go b/cmd/rds-agent/command_test.go index 04e0ed2cc..2db49857b 100644 --- a/cmd/rds-agent/command_test.go +++ b/cmd/rds-agent/command_test.go @@ -57,7 +57,7 @@ func TestCommander_FailedHandlerRepliesWithItsError(t *testing.T) { // for. The issuer is blocked on the command ID, so it gets an answer rather // than a timeout. func TestCommander_UnknownTypeRepliesFailedNotDropped(t *testing.T) { - reply := newCommander(nil, newCommandRegistry(), time.Second).execute(context.Background(), + reply := newCommander(nil, newCommandRegistry(&fakeEngine{}), time.Second).execute(context.Background(), handlers_rds.Command{CommandID: "cmd-3", Type: "quiesce"}) if reply.CommandID != "cmd-3" || reply.Status != handlers_rds.CommandStatusFailed { @@ -107,7 +107,7 @@ func TestCommander_BacksOffAfterAFailedPoll(t *testing.T) { cp := newFakeControlPlane() cp.pollErr = errors.New("gateway unreachable") - c := newCommander(cp, newCommandRegistry(), 10*time.Millisecond) + c := newCommander(cp, newCommandRegistry(&fakeEngine{}), 10*time.Millisecond) ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) defer cancel() done := make(chan struct{}) diff --git a/cmd/rds-agent/config.go b/cmd/rds-agent/config.go index 1c95f6195..65fadc72a 100644 --- a/cmd/rds-agent/config.go +++ b/cmd/rds-agent/config.go @@ -14,6 +14,15 @@ const ( defaultEngineHost = "127.0.0.1" defaultEnginePort = 5432 defaultPGIsReady = "pg_isready" + + // The guest layout rds-init lays down. Overridable for the same reason it is + // there: a sibling engine preset points at its own paths. + defaultPGBin = "/usr/libexec/postgresql18" + defaultPGData = "/var/lib/postgresql/18/data" + defaultSocketDir = "/run/postgresql" + defaultPGUser = "postgres" + defaultRCService = "rc-service" + defaultEngineSvcName = "postgresql" // The long-poll window the agent asks the gateway to hold a request open // for. The gateway caps it at 20s. defaultPollWait = 20 * time.Second @@ -38,6 +47,16 @@ type config struct { // Overridable so a test or a sibling engine preset can point at its own. PGIsReady string PollWait time.Duration + + // Where the command handlers reach the engine: the client binaries, the + // datadir the parameter file is installed into, the socket they connect + // over, the OS user they drop to, and the service the stop goes through. + PGBin string + PGData string + SocketDir string + PGUser string + RCService string + EngineService string } // Reads the cloud-init env file, then lets real env vars override. @@ -53,6 +72,12 @@ func loadConfig(envFile string) config { HandoffDir: get("RDS_HANDOFF_DIR"), EngineHost: get("RDS_ENGINE_HOST"), PGIsReady: get("RDS_PG_ISREADY"), + PGBin: get("RDS_PG_BIN"), + PGData: get("RDS_PGDATA"), + SocketDir: get("RDS_SOCKET_DIR"), + PGUser: get("RDS_PG_USER"), + RCService: get("RDS_RC_SERVICE"), + EngineService: get("RDS_ENGINE_SERVICE"), EnginePort: defaultEnginePort, PollWait: defaultPollWait, } @@ -68,6 +93,24 @@ func loadConfig(envFile string) config { if cfg.PGIsReady == "" { cfg.PGIsReady = defaultPGIsReady } + if cfg.PGBin == "" { + cfg.PGBin = defaultPGBin + } + if cfg.PGData == "" { + cfg.PGData = defaultPGData + } + if cfg.SocketDir == "" { + cfg.SocketDir = defaultSocketDir + } + if cfg.PGUser == "" { + cfg.PGUser = defaultPGUser + } + if cfg.RCService == "" { + cfg.RCService = defaultRCService + } + if cfg.EngineService == "" { + cfg.EngineService = defaultEngineSvcName + } // The authoritative port comes from the bootstrap config; this is only what // the probe uses until that first fetch lands. if v := get("RDS_ENGINE_PORT"); v != "" { diff --git a/cmd/rds-agent/engine.go b/cmd/rds-agent/engine.go new file mode 100644 index 000000000..fd00c51aa --- /dev/null +++ b/cmd/rds-agent/engine.go @@ -0,0 +1,249 @@ +package main + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "os/user" + "path/filepath" + "strconv" + "strings" + "sync/atomic" + "syscall" + + 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. +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 (D8). + SetPassword(ctx context.Context, username, password string) error + // Installs the resolved parameter set and reloads. Returns the settings the + // engine accepted but will not honour until it restarts. + ApplyParameters(ctx context.Context, params []handlers_rds.Parameter) ([]string, error) + // Shuts the engine down cleanly, so the data volume is checkpointed before + // the VM is stopped or snapshotted. + Stop(ctx context.Context) error +} + +// One child process. Env replaces the agent's own environment rather than +// extending it, so a secret placed here reaches only the process that needs it. +type command struct { + Name string + Args []string + Env []string + Stdin string + // The OS user to drop to before exec. Empty runs as the agent itself. + User string +} + +// Returns stdout. stderr is folded into the error, because psql reports 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) { + cmd := exec.CommandContext(ctx, c.Name, c.Args...) + cmd.Env = c.Env + if c.Stdin != "" { + cmd.Stdin = strings.NewReader(c.Stdin) + } + if c.User != "" { + credential, err := lookupCredential(c.User) + if err != nil { + return "", err + } + cmd.SysProcAttr = &syscall.SysProcAttr{Credential: credential} + } + + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + if err := cmd.Run(); err != nil { + message := strings.TrimSpace(stderr.String()) + if message == "" { + message = err.Error() + } + return stdout.String(), fmt.Errorf("%s: %s", filepath.Base(c.Name), message) + } + return stdout.String(), nil +} + +func lookupCredential(username string) (*syscall.Credential, error) { + u, err := user.Lookup(username) + if err != nil { + return nil, fmt.Errorf("resolve user %s: %w", username, err) + } + uid, err := strconv.ParseUint(u.Uid, 10, 32) + if err != nil { + return nil, fmt.Errorf("parse uid of %s: %w", username, err) + } + gid, err := strconv.ParseUint(u.Gid, 10, 32) + if err != nil { + return nil, fmt.Errorf("parse gid of %s: %w", username, err) + } + 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 { + run commandRunner + psql string + rcService string + service string + pgData string + socketDir string + osUser string + // Set from the bootstrap config, on a different goroutine than the commands + // that read it. + port atomic.Int64 +} + +var _ engineOps = (*postgresEngine)(nil) + +func newPostgresEngine(cfg config, run commandRunner) *postgresEngine { + e := &postgresEngine{ + run: run, + psql: filepath.Join(cfg.PGBin, "psql"), + rcService: cfg.RCService, + service: cfg.EngineService, + pgData: cfg.PGData, + socketDir: cfg.SocketDir, + osUser: cfg.PGUser, + } + 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) + } + const sql = `\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 parameters go beside the data rather than into /etc, matching rds-init: a +// class change boots a fresh root volume, which would otherwise revert them. +func (e *postgresEngine) ApplyParameters(ctx context.Context, params []handlers_rds.Parameter) ([]string, error) { + if err := e.installParameters(params); err != nil { + return nil, err + } + if _, err := e.psqlRun(ctx, "SELECT pg_reload_conf();\n"); err != nil { + return nil, fmt.Errorf("reload the engine configuration: %w", err) + } + + // Read after the reload: a setting only becomes pending_restart once the + // engine has seen the new value and declined to adopt it live. + 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 +} + +// 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. +func (e *postgresEngine) installParameters(params []handlers_rds.Parameter) error { + dir := filepath.Join(e.pgData, "conf.d") + path := filepath.Join(dir, "10-rds-parameters.conf") + tmp := path + ".new" + + if err := os.WriteFile(tmp, []byte(renderParameters(params)), 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 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 { + if _, err := e.run(ctx, command{ + Name: e.rcService, + Args: []string{e.service, "stop"}, + Env: []string{"PATH=" + defaultGuestPath}, + }); err != nil { + return fmt.Errorf("stop the %s service: %w", e.service, err) + } + return nil +} + +// 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: []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", "-", + }, + Env: append([]string{"PATH=" + defaultGuestPath}, env...), + Stdin: sql, + User: e.osUser, + }) +} + +func redact(s, secret string) string { + if secret == "" { + return s + } + return strings.ReplaceAll(s, secret, "[REDACTED]") +} diff --git a/cmd/rds-agent/engine_test.go b/cmd/rds-agent/engine_test.go new file mode 100644 index 000000000..095689ddc --- /dev/null +++ b/cmd/rds-agent/engine_test.go @@ -0,0 +1,268 @@ +package main + +import ( + "context" + "errors" + "os" + "os/user" + "path/filepath" + "slices" + "strings" + "testing" + + handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" +) + +// Stands in for the engine in the registry tests, where what matters is that a +// command reaches the right method with the right arguments. +type fakeEngine struct { + username string + password string + params []handlers_rds.Parameter + pending []string + stopped bool + err error +} + +var _ engineOps = (*fakeEngine)(nil) + +func (f *fakeEngine) SetPassword(_ context.Context, username, password string) error { + f.username, f.password = username, password + return f.err +} + +func (f *fakeEngine) ApplyParameters(_ context.Context, params []handlers_rds.Parameter) ([]string, error) { + f.params = params + return f.pending, f.err +} + +func (f *fakeEngine) Stop(context.Context) error { + f.stopped = true + return f.err +} + +func TestCommandRegistry_SetPasswordCarriesBothParameters(t *testing.T) { + engine := &fakeEngine{} + reply := newCommander(nil, newCommandRegistry(engine), 0).execute(context.Background(), handlers_rds.Command{ + CommandID: "cmd-1", + Type: handlers_rds.CommandSetPassword, + Parameters: []handlers_rds.Parameter{ + {Name: handlers_rds.CommandParamMasterUsername, Value: "postgres"}, + {Name: handlers_rds.CommandParamMasterUserPassword, Value: "n3w-pw"}, + }, + }) + + if reply.Status != handlers_rds.CommandStatusSucceeded { + t.Fatalf("reply = %+v, want succeeded", reply) + } + if engine.username != "postgres" || engine.password != "n3w-pw" { + t.Errorf("engine got %q/%q, want postgres/n3w-pw", engine.username, engine.password) + } + // The reply is recorded by the control plane; the new password must not be + // in it. + if strings.Contains(reply.Message, "n3w-pw") { + t.Errorf("reply message %q carries the new password", reply.Message) + } +} + +// CommandReply has no structured payload, so the settings still awaiting a +// restart come back as the message the issuer parses. +func TestCommandRegistry_ApplyParamsReportsPendingRestart(t *testing.T) { + engine := &fakeEngine{pending: []string{"shared_buffers", "max_connections"}} + reply := newCommander(nil, newCommandRegistry(engine), 0).execute(context.Background(), handlers_rds.Command{ + CommandID: "cmd-2", + Type: handlers_rds.CommandApplyParams, + Parameters: []handlers_rds.Parameter{{Name: "shared_buffers", Value: "256MB"}}, + }) + + if reply.Status != handlers_rds.CommandStatusSucceeded { + t.Fatalf("reply = %+v, want succeeded", reply) + } + if reply.Message != "shared_buffers,max_connections" { + t.Errorf("reply message = %q, want the pending-restart list", reply.Message) + } + if len(engine.params) != 1 || engine.params[0].Name != "shared_buffers" { + t.Errorf("engine got %+v, want the command's parameters", engine.params) + } +} + +func TestCommandRegistry_StopEngineReachesTheEngine(t *testing.T) { + engine := &fakeEngine{} + reply := newCommander(nil, newCommandRegistry(engine), 0).execute(context.Background(), + handlers_rds.Command{CommandID: "cmd-3", Type: handlers_rds.CommandStopEngine}) + + if reply.Status != handlers_rds.CommandStatusSucceeded { + t.Fatalf("reply = %+v, want succeeded", reply) + } + if !engine.stopped { + t.Error("stop-engine did not reach the engine") + } +} + +func TestCommandRegistry_FailureRepliesFailed(t *testing.T) { + engine := &fakeEngine{err: errors.New("the engine did not stop")} + reply := newCommander(nil, newCommandRegistry(engine), 0).execute(context.Background(), + handlers_rds.Command{CommandID: "cmd-4", Type: handlers_rds.CommandStopEngine}) + + if reply.Status != handlers_rds.CommandStatusFailed { + t.Fatalf("reply = %+v, want failed", reply) + } + if !strings.Contains(reply.Message, "the engine did not stop") { + t.Errorf("reply message = %q, want the engine's error", reply.Message) + } +} + +// Records what a postgresEngine would have run, so the SQL and the argv are +// asserted without a live cluster. +type recordingRunner struct { + calls []command + out string + err error +} + +func (r *recordingRunner) run(_ context.Context, c command) (string, error) { + r.calls = append(r.calls, c) + return r.out, r.err +} + +func newTestEngine(t *testing.T, run commandRunner) *postgresEngine { + t.Helper() + cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env")) + cfg.PGData = t.TempDir() + if err := os.MkdirAll(filepath.Join(cfg.PGData, "conf.d"), 0o700); err != nil { + t.Fatalf("create conf.d: %v", err) + } + // The guest's postgres 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.PGUser = current.Username + return newPostgresEngine(cfg, run) +} + +// D8: the password reaches psql through the environment and is re-quoted there, +// so it is never a shell word or an argv another process can read. +func TestPostgresEngine_SetPasswordKeepsTheSecretOutOfArgv(t *testing.T) { + runner := &recordingRunner{} + engine := newTestEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "postgres", "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) + } + } + if strings.Contains(call.Stdin, "n3w-pw") { + t.Errorf("SQL %q carries the password literally, want it read from the environment", call.Stdin) + } + if !slices.Contains(call.Env, "RDS_MASTER_PASSWORD=n3w-pw") { + t.Errorf("env %v does not carry the password", call.Env) + } + if !strings.Contains(call.Stdin, "ALTER ROLE") { + t.Errorf("SQL %q does not alter the role", call.Stdin) + } +} + +// psql echoes the failing statement, which here would carry the new password +// back into a reply the control plane records. +func TestPostgresEngine_SetPasswordRedactsTheSecretFromAFailure(t *testing.T) { + runner := &recordingRunner{err: errors.New(`ERROR: syntax error at or near "n3w-pw"`)} + engine := newTestEngine(t, runner.run) + + err := engine.SetPassword(context.Background(), "postgres", "n3w-pw") + if err == nil { + t.Fatal("SetPassword succeeded against a failing psql") + } + if strings.Contains(err.Error(), "n3w-pw") { + t.Errorf("error %q leaks the password", err) + } + if !strings.Contains(err.Error(), "[REDACTED]") { + t.Errorf("error %q does not mark the redaction", err) + } +} + +func TestPostgresEngine_SetPasswordRejectsAnIncompleteCommand(t *testing.T) { + runner := &recordingRunner{} + engine := newTestEngine(t, runner.run) + + if err := engine.SetPassword(context.Background(), "postgres", ""); 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)) + } +} + +// The parameters land where rds-init installs them, so the next boot overwrites +// the file rather than shadowing it with a second copy. +func TestPostgresEngine_ApplyParametersInstallsAndReloads(t *testing.T) { + runner := &recordingRunner{out: "shared_buffers\nmax_connections\n"} + engine := newTestEngine(t, runner.run) + + pending, err := engine.ApplyParameters(context.Background(), + []handlers_rds.Parameter{{Name: "shared_buffers", Value: "256MB"}}) + if err != nil { + t.Fatalf("ApplyParameters: %v", err) + } + + installed, err := os.ReadFile(filepath.Join(engine.pgData, "conf.d", "10-rds-parameters.conf")) + if err != nil { + t.Fatalf("read the installed parameters: %v", err) + } + if !strings.Contains(string(installed), "shared_buffers = '256MB'") { + t.Errorf("installed parameters = %q, want the resolved value", installed) + } + if len(runner.calls) != 2 { + t.Fatalf("ran %d commands, want a reload and a pending-restart read", len(runner.calls)) + } + if !strings.Contains(runner.calls[0].Stdin, "pg_reload_conf") { + t.Errorf("first statement = %q, want the reload", runner.calls[0].Stdin) + } + if !strings.Contains(runner.calls[1].Stdin, "pending_restart") { + t.Errorf("second statement = %q, want the pending-restart read", runner.calls[1].Stdin) + } + if len(pending) != 2 || pending[0] != "shared_buffers" || pending[1] != "max_connections" { + t.Errorf("pending = %v, want both settings the engine reported", pending) + } +} + +// A reload that failed leaves the engine on its old values, so reporting the +// apply as successful would tell the customer a change took effect that did not. +func TestPostgresEngine_ApplyParametersSurfacesAFailedReload(t *testing.T) { + runner := &recordingRunner{err: errors.New("could not connect to server")} + engine := newTestEngine(t, runner.run) + + if _, err := engine.ApplyParameters(context.Background(), + []handlers_rds.Parameter{{Name: "work_mem", Value: "8MB"}}); err == nil { + t.Fatal("ApplyParameters succeeded against a failing reload") + } +} + +// Through the service manager, so the supervisor records the engine as stopped +// and does not restart it underneath a VM that is going down. +func TestPostgresEngine_StopGoesThroughTheServiceManager(t *testing.T) { + runner := &recordingRunner{} + engine := newTestEngine(t, runner.run) + + if err := engine.Stop(context.Background()); err != nil { + t.Fatalf("Stop: %v", err) + } + if len(runner.calls) != 1 { + t.Fatalf("ran %d commands, want 1", len(runner.calls)) + } + call := runner.calls[0] + if filepath.Base(call.Name) != "rc-service" { + t.Errorf("ran %q, want rc-service", call.Name) + } + if len(call.Args) != 2 || call.Args[0] != "postgresql" || call.Args[1] != "stop" { + t.Errorf("args = %v, want [postgresql stop]", call.Args) + } +} diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index 5a3c34881..89cb108ee 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -462,6 +462,8 @@ var ( // lookups, so the SDK's typed error matching round-trips. ErrorDBInstanceNotFound = "DBInstanceNotFound" ErrorDBInstanceAlreadyExists = "DBInstanceAlreadyExists" + ErrorDBInstanceInvalidState = "InvalidDBInstanceState" + ErrorDBSnapshotAlreadyExists = "DBSnapshotAlreadyExists" ErrorDBSubnetGroupNotFound = "DBSubnetGroupNotFoundFault" ErrorDBParameterGroupNotFound = "DBParameterGroupNotFound" ErrorDBInvalidVPCNetworkState = "InvalidVPCNetworkStateFault" @@ -1034,6 +1036,8 @@ var ErrorLookup = map[string]ErrorMessage{ // RDS error codes ErrorDBInstanceNotFound: {HTTPCode: 404, Message: "DBInstanceIdentifier does not refer to an existing DB instance."}, ErrorDBInstanceAlreadyExists: {HTTPCode: 400, Message: "The user already has a DB instance with the given identifier."}, + ErrorDBInstanceInvalidState: {HTTPCode: 400, Message: "The DB instance is not in a state that allows the requested operation."}, + ErrorDBSnapshotAlreadyExists: {HTTPCode: 400, Message: "The user already has a DB snapshot with the given identifier."}, ErrorDBSubnetGroupNotFound: {HTTPCode: 404, Message: "DBSubnetGroupName does not refer to an existing DB subnet group."}, ErrorDBParameterGroupNotFound: {HTTPCode: 404, Message: "DBParameterGroupName does not refer to an existing DB parameter group."}, ErrorDBInvalidVPCNetworkState: {HTTPCode: 400, Message: "The DB subnet group does not cover all Availability Zones after it is created because of changes that were made."}, diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 22cca5ace..f3b27f9e8 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1058,6 +1058,11 @@ func (d *Daemon) subscribeAll() error { natsSub{handlers_rds.SubjectGetDBBootstrapConfig, handleNATSRequest(d.rdsService.GetDBBootstrapConfig), "spinifex-workers"}, natsSub{handlers_rds.SubjectCreateDBInstance, handleNATSRequest(d.rdsService.CreateDBInstance), "spinifex-workers"}, natsSub{handlers_rds.SubjectDescribeDBInstances, handleNATSRequest(d.rdsService.DescribeDBInstances), "spinifex-workers"}, + natsSub{handlers_rds.SubjectRebootDBInstance, handleNATSRequest(d.rdsService.RebootDBInstance), "spinifex-workers"}, + natsSub{handlers_rds.SubjectStartDBInstance, handleNATSRequest(d.rdsService.StartDBInstance), "spinifex-workers"}, + natsSub{handlers_rds.SubjectStopDBInstance, handleNATSRequest(d.rdsService.StopDBInstance), "spinifex-workers"}, + natsSub{handlers_rds.SubjectDeleteDBInstance, handleNATSRequest(d.rdsService.DeleteDBInstance), "spinifex-workers"}, + natsSub{handlers_rds.SubjectDescribeEvents, handleNATSRequest(d.rdsService.DescribeEvents), "spinifex-workers"}, natsSub{handlers_rds.SubjectAddTagsToResource, handleNATSRequest(d.rdsService.AddTagsToResource), "spinifex-workers"}, natsSub{handlers_rds.SubjectRemoveTagsFromResource, handleNATSRequest(d.rdsService.RemoveTagsFromResource), "spinifex-workers"}, natsSub{handlers_rds.SubjectListTagsForResource, handleNATSRequest(d.rdsService.ListTagsForResource), "spinifex-workers"}, diff --git a/spinifex/daemon/rds_deps.go b/spinifex/daemon/rds_deps.go index c9d8ff133..ab6389739 100644 --- a/spinifex/daemon/rds_deps.go +++ b/spinifex/daemon/rds_deps.go @@ -55,6 +55,8 @@ func (d *Daemon) buildRDSDeps() handlers_rds.Deps { Network: d.vpcService, IAM: d.systemRoleEnsurer, InstanceState: handlers_rds.NewDescribeInstanceState(d.describeInstancesFanOut), + Instances: handlers_rds.NewNATSInstanceCommander(d.natsConn), + Snapshots: d.snapshotService, BaseDomain: d.dnsBaseDomain, HolderID: d.node, // A DB VM reaches the daemon only over the mgmt bridge, the same diff --git a/spinifex/gateway/rds/handler.go b/spinifex/gateway/rds/handler.go index d16492785..30bd93a9d 100644 --- a/spinifex/gateway/rds/handler.go +++ b/spinifex/gateway/rds/handler.go @@ -82,10 +82,10 @@ var actions = map[string]Handler{ "CreateDBInstance": typed(CreateDBInstance), "DescribeDBInstances": typed(DescribeDBInstances), "ModifyDBInstance": pending(), - "DeleteDBInstance": pending(), - "RebootDBInstance": pending(), - "StartDBInstance": pending(), - "StopDBInstance": pending(), + "DeleteDBInstance": typed(DeleteDBInstance), + "RebootDBInstance": typed(RebootDBInstance), + "StartDBInstance": typed(StartDBInstance), + "StopDBInstance": typed(StopDBInstance), // Snapshots. "CreateDBSnapshot": pending(), @@ -114,7 +114,7 @@ var actions = map[string]Handler{ "ListTagsForResource": typed(ListTagsForResource), // Events. - "DescribeEvents": pending(), + "DescribeEvents": typed(DescribeEvents), // Internal agent actions, callable only by the in-guest agent's system role. // They share the namespace because the agent reaches the control plane over diff --git a/spinifex/gateway/rds/handler_test.go b/spinifex/gateway/rds/handler_test.go index cce6aafd2..d7edb5cb6 100644 --- a/spinifex/gateway/rds/handler_test.go +++ b/spinifex/gateway/rds/handler_test.go @@ -104,6 +104,8 @@ func TestDispatch_UnknownAction(t *testing.T) { // dispatched against a stub responder rather than a nil connection. var liveActions = []string{ "CreateDBInstance", "DescribeDBInstances", + "RebootDBInstance", "StartDBInstance", "StopDBInstance", "DeleteDBInstance", + "DescribeEvents", "AddTagsToResource", "RemoveTagsFromResource", "ListTagsForResource", } @@ -116,6 +118,11 @@ func newStubbedNATS(t *testing.T) *nats.Conn { &rds.CreateDBInstanceOutput{DBInstance: &rds.DBInstance{DBInstanceIdentifier: aws.String("orders-db")}}) respondWith(t, nc, handlers_rds.SubjectDescribeDBInstances, &rds.DescribeDBInstancesOutput{DBInstances: []*rds.DBInstance{}}) + respondWith(t, nc, handlers_rds.SubjectRebootDBInstance, &rds.RebootDBInstanceOutput{}) + respondWith(t, nc, handlers_rds.SubjectStartDBInstance, &rds.StartDBInstanceOutput{}) + respondWith(t, nc, handlers_rds.SubjectStopDBInstance, &rds.StopDBInstanceOutput{}) + respondWith(t, nc, handlers_rds.SubjectDeleteDBInstance, &rds.DeleteDBInstanceOutput{}) + respondWith(t, nc, handlers_rds.SubjectDescribeEvents, &rds.DescribeEventsOutput{}) respondWith(t, nc, handlers_rds.SubjectAddTagsToResource, &rds.AddTagsToResourceOutput{}) respondWith(t, nc, handlers_rds.SubjectRemoveTagsFromResource, &rds.RemoveTagsFromResourceOutput{}) respondWith(t, nc, handlers_rds.SubjectListTagsForResource, diff --git a/spinifex/gateway/rds/instance.go b/spinifex/gateway/rds/instance.go index 0807d2dce..576941a1b 100644 --- a/spinifex/gateway/rds/instance.go +++ b/spinifex/gateway/rds/instance.go @@ -17,3 +17,25 @@ func CreateDBInstance(ctx context.Context, input *rds.CreateDBInstanceInput, nc func DescribeDBInstances(ctx context.Context, input *rds.DescribeDBInstancesInput, nc *nats.Conn, caller Caller) (any, error) { return handlers_rds.NewNATSService(nc).DescribeDBInstances(ctx, input, caller.AccountID) } + +func RebootDBInstance(ctx context.Context, input *rds.RebootDBInstanceInput, nc *nats.Conn, caller Caller) (any, error) { + return handlers_rds.NewNATSService(nc).RebootDBInstance(ctx, input, caller.AccountID) +} + +func StartDBInstance(ctx context.Context, input *rds.StartDBInstanceInput, nc *nats.Conn, caller Caller) (any, error) { + return handlers_rds.NewNATSService(nc).StartDBInstance(ctx, input, caller.AccountID) +} + +func StopDBInstance(ctx context.Context, input *rds.StopDBInstanceInput, nc *nats.Conn, caller Caller) (any, error) { + return handlers_rds.NewNATSService(nc).StopDBInstance(ctx, input, caller.AccountID) +} + +func DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInstanceInput, nc *nats.Conn, caller Caller) (any, error) { + return handlers_rds.NewNATSService(nc).DeleteDBInstance(ctx, input, caller.AccountID) +} + +// Scoped to the caller's account: an event ring is per-tenant, so there is no +// cross-account read even for a resource that has since been deleted. +func DescribeEvents(ctx context.Context, input *rds.DescribeEventsInput, nc *nats.Conn, caller Caller) (any, error) { + return handlers_rds.NewNATSService(nc).DescribeEvents(ctx, input, caller.AccountID) +} diff --git a/spinifex/handlers/rds/commands.go b/spinifex/handlers/rds/commands.go new file mode 100644 index 000000000..a98fc4314 --- /dev/null +++ b/spinifex/handlers/rds/commands.go @@ -0,0 +1,181 @@ +package handlers_rds + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/google/uuid" + "github.com/nats-io/nats.go" +) + +// The control-plane half of the rds-2b command channel. The agent side is a +// long poll over the gateway; this publishes on the bus subject that poll is +// subscribed to and waits for the reply the agent carries back on its next one. + +// Command types. rds-5b adds grow-filesystem and rds-8 the snapshot quiesce. +const ( + CommandSetPassword = "set-password" + CommandApplyParams = "apply-params" + CommandStopEngine = "stop-engine" +) + +// Parameter names carried on a command. They are AWS-shaped rather than +// engine-shaped: the agent maps them onto ALTER ROLE or a config write. +const ( + CommandParamMasterUsername = "MasterUsername" + CommandParamMasterUserPassword = "MasterUserPassword" +) + +// Per-command budgets. A password apply is one statement; a parameter apply +// rewrites a config file and reloads; a graceful engine stop has to checkpoint, +// which is bounded by the dirty set rather than by anything the caller knows. +const ( + setPasswordTimeout = 30 * time.Second + applyParamsTimeout = 60 * time.Second + stopEngineTimeout = 120 * time.Second + + // The channel is a live subscription rather than a durable queue (D8), so a + // command published between two of the agent's polls reaches nobody. + // Re-publishing on this interval closes that gap without queueing anything. + commandRepublishEvery = 2 * time.Second +) + +// The agent could not be reached, or did not answer inside the command's +// budget. Callers that can degrade — the graceful engine stop — check for this; +// callers that must not, like a password rotation, surface it as retryable. +var ErrCommandUnreachable = errors.New("rds: the instance agent did not answer the command") + +// Sets the engine's master password live. It is never persisted anywhere: an +// unreachable agent fails loudly rather than leaving cleartext queued for +// later (D8). +func (s *Service) setMasterPassword(ctx context.Context, accountID, dbInstanceIdentifier, username, password string) error { + _, err := s.issueCommand(ctx, accountID, dbInstanceIdentifier, CommandSetPassword, setPasswordTimeout, []Parameter{ + {Name: CommandParamMasterUsername, Value: username}, + {Name: CommandParamMasterUserPassword, Value: password}, + }) + return err +} + +// Writes the resolved parameter set into the engine's config and reloads it. +// Returns the settings the engine accepted but will not apply until it +// restarts, which is what RebootDBInstance then clears (D16). +func (s *Service) applyParameters(ctx context.Context, accountID, dbInstanceIdentifier string, params []Parameter) ([]string, error) { + reply, err := s.issueCommand(ctx, accountID, dbInstanceIdentifier, CommandApplyParams, applyParamsTimeout, params) + if err != nil { + return nil, err + } + return parsePendingRestart(reply.Message), nil +} + +// Shuts the engine down cleanly so the data volume is checkpointed before the +// VM stops. Callers treat a failure as degradation, not as an error. +func (s *Service) stopEngine(ctx context.Context, accountID, dbInstanceIdentifier string) error { + _, err := s.issueCommand(ctx, accountID, dbInstanceIdentifier, CommandStopEngine, stopEngineTimeout, nil) + return err +} + +// Publishes one directive and blocks until the agent replies to that command +// ID, the budget expires, or ctx is cancelled. A reply reporting failure is +// returned as an error carrying the agent's message, so the caller does not +// have to inspect the status itself. +func (s *Service) issueCommand(ctx context.Context, accountID, dbInstanceIdentifier, commandType string, timeout time.Duration, params []Parameter) (*CommandReply, error) { + if s.nc == nil { + return nil, errors.New("rds: nil nats connection") + } + issuedAt := time.Now().UTC() + cmd := Command{ + CommandID: uuid.NewString(), + Type: commandType, + Parameters: params, + IssuedAt: &issuedAt, + } + data, err := json.Marshal(cmd) + if err != nil { + return nil, err + } + + // Subscribed before the first publish, so a reply cannot land in the gap + // between the command going out and the issuer starting to listen. + sub, err := s.nc.SubscribeSync(BusCommandReplySubject(accountID, dbInstanceIdentifier)) + if err != nil { + return nil, fmt.Errorf("rds: subscribe command replies for %s: %w", dbInstanceIdentifier, err) + } + defer func() { + if err := sub.Unsubscribe(); err != nil { + slog.DebugContext(ctx, "rds: command reply unsubscribe failed", "dbInstance", dbInstanceIdentifier, "err", err) + } + }() + + deadlineCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + subject := BusCommandSubject(accountID, dbInstanceIdentifier) + + for { + if err := s.nc.Publish(subject, data); err != nil { + return nil, fmt.Errorf("rds: publish %s command to %s: %w", commandType, dbInstanceIdentifier, err) + } + reply, err := waitForReply(deadlineCtx, sub, cmd.CommandID) + if err != nil { + return nil, err + } + if reply == nil { + // Nothing yet: re-publish in case the agent was between polls. + continue + } + if reply.Status != CommandStatusSucceeded { + return nil, fmt.Errorf("rds: %s command failed on %s: %s", commandType, dbInstanceIdentifier, reply.Message) + } + slog.DebugContext(ctx, "rds: agent command completed", + "dbInstance", dbInstanceIdentifier, "type", commandType, "commandId", cmd.CommandID) + return reply, nil + } +} + +// Returns (nil, nil) when the republish interval elapsed with no matching +// reply, so the caller re-publishes; ErrCommandUnreachable once the whole +// budget is gone. Replies for other command IDs are stale answers to an earlier +// issuer and are skipped. +func waitForReply(ctx context.Context, sub *nats.Subscription, commandID string) (*CommandReply, error) { + window, cancel := context.WithTimeout(ctx, commandRepublishEvery) + defer cancel() + + for { + msg, err := sub.NextMsgWithContext(window) + if err != nil { + // The caller giving up is its own failure, not an unreachable agent. + if errors.Is(ctx.Err(), context.Canceled) { + return nil, ctx.Err() + } + if ctx.Err() != nil { + return nil, ErrCommandUnreachable + } + // Only the republish window closed. + return nil, nil + } + var reply CommandReply + if err := json.Unmarshal(msg.Data, &reply); err != nil { + slog.Debug("rds: undecodable command reply dropped", "err", err) + continue + } + if reply.CommandID == commandID { + return &reply, nil + } + } +} + +// The agent reports the settings still awaiting a restart as a comma-separated +// list in its reply message, since CommandReply carries no structured payload. +func parsePendingRestart(message string) []string { + var pending []string + for name := range strings.SplitSeq(message, ",") { + if trimmed := strings.TrimSpace(name); trimmed != "" { + pending = append(pending, trimmed) + } + } + return pending +} diff --git a/spinifex/handlers/rds/commands_test.go b/spinifex/handlers/rds/commands_test.go new file mode 100644 index 000000000..11ffdac97 --- /dev/null +++ b/spinifex/handlers/rds/commands_test.go @@ -0,0 +1,77 @@ +package handlers_rds + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// D8: the password is handed straight to the agent and never persisted, so the +// command is the only place it exists between the request and the engine. +func TestSetMasterPassword_CarriesTheCredentialsToTheAgent(t *testing.T) { + h := newLifecycleHarness(t, false) + + require.NoError(t, h.svc.setMasterPassword(t.Context(), testAccountID, testDBID, "postgres", "n3w-pw")) + + issued := h.agent.received() + require.Len(t, issued, 1) + assert.Equal(t, CommandSetPassword, issued[0].Type) + assert.NotEmpty(t, issued[0].CommandID) + assert.ElementsMatch(t, []Parameter{ + {Name: CommandParamMasterUsername, Value: "postgres"}, + {Name: CommandParamMasterUserPassword, Value: "n3w-pw"}, + }, issued[0].Parameters) +} + +// A rotation that could not be applied has to fail loudly: reporting success +// would leave the customer with a password the engine does not accept. +func TestSetMasterPassword_FailsWhenTheAgentRejectsIt(t *testing.T) { + h := newLifecycleHarness(t, true) + + err := h.svc.setMasterPassword(t.Context(), testAccountID, testDBID, "postgres", "n3w-pw") + require.Error(t, err) + assert.Contains(t, err.Error(), CommandSetPassword) +} + +// D16: the settings the engine accepted but will not honour until it restarts +// come back as the reply message, which is what a later reboot then clears. +func TestApplyParameters_ReportsWhatIsPendingARestart(t *testing.T) { + h := newLifecycleHarness(t, false) + h.agent.replyWith("shared_buffers, max_connections") + + pending, err := h.svc.applyParameters(t.Context(), testAccountID, testDBID, + []Parameter{{Name: "shared_buffers", Value: "256MB"}}) + require.NoError(t, err) + assert.Equal(t, []string{"shared_buffers", "max_connections"}, pending) + + issued := h.agent.received() + require.Len(t, issued, 1) + assert.Equal(t, CommandApplyParams, issued[0].Type) + assert.Equal(t, []Parameter{{Name: "shared_buffers", Value: "256MB"}}, issued[0].Parameters) +} + +// An agent that reports nothing pending has applied everything live. +func TestApplyParameters_ReportsNothingPendingOnAnEmptyReply(t *testing.T) { + h := newLifecycleHarness(t, false) + + pending, err := h.svc.applyParameters(t.Context(), testAccountID, testDBID, + []Parameter{{Name: "work_mem", Value: "8MB"}}) + require.NoError(t, err) + assert.Empty(t, pending) +} + +// The reply is matched on the command ID, so a stale answer to an earlier +// issuer cannot be mistaken for the outcome of this command. +func TestIssueCommand_CorrelatesTheReplyByCommandID(t *testing.T) { + h := newLifecycleHarness(t, false) + + reply, err := h.svc.issueCommand(t.Context(), testAccountID, testDBID, + CommandStopEngine, stopEngineTimeout, nil) + require.NoError(t, err) + require.NotNil(t, reply) + + issued := h.agent.received() + require.NotEmpty(t, issued) + assert.Equal(t, issued[0].CommandID, reply.CommandID) +} diff --git a/spinifex/handlers/rds/create.go b/spinifex/handlers/rds/create.go index eb413513e..f06bfa1d0 100644 --- a/spinifex/handlers/rds/create.go +++ b/spinifex/handlers/rds/create.go @@ -143,6 +143,7 @@ func newDBInstanceRecord(accountID string, req *validatedCreate, placement *endp VpcID: placement.VpcID, VpcSecurityGroupIDs: placement.SecurityGroupIDs, DBParameterGroupName: req.DBParameterGroupName, + DeletionProtection: req.DeletionProtection, Tags: req.Tags, Bootstrap: BootstrapState{MasterUserPassword: req.MasterPassword}, CreatedAt: now, diff --git a/spinifex/handlers/rds/create_test.go b/spinifex/handlers/rds/create_test.go index 95a5bc016..3dc0e5a13 100644 --- a/spinifex/handlers/rds/create_test.go +++ b/spinifex/handlers/rds/create_test.go @@ -379,7 +379,6 @@ func TestValidateCreateRequest_RejectsUnimplementedParameters(t *testing.T) { {"MultiAZ", func(in *rds.CreateDBInstanceInput) { in.MultiAZ = aws.Bool(true) }, "MultiAZ"}, {"PubliclyAccessible", func(in *rds.CreateDBInstanceInput) { in.PubliclyAccessible = aws.Bool(true) }, "PubliclyAccessible"}, {"StorageEncryptedFalse", func(in *rds.CreateDBInstanceInput) { in.StorageEncrypted = aws.Bool(false) }, "StorageEncrypted"}, - {"DeletionProtection", func(in *rds.CreateDBInstanceInput) { in.DeletionProtection = aws.Bool(true) }, "DeletionProtection"}, {"BackupRetentionPeriod", func(in *rds.CreateDBInstanceInput) { in.BackupRetentionPeriod = aws.Int64(7) }, "BackupRetentionPeriod"}, {"PreferredBackupWindow", func(in *rds.CreateDBInstanceInput) { in.PreferredBackupWindow = aws.String("03:00-04:00") }, "PreferredBackupWindow"}, {"PreferredMaintenanceWindow", func(in *rds.CreateDBInstanceInput) { in.PreferredMaintenanceWindow = aws.String("sun:05:00-sun:06:00") }, "PreferredMaintenanceWindow"}, @@ -406,6 +405,17 @@ func TestValidateCreateRequest_RejectsUnimplementedParameters(t *testing.T) { } } +// DeletionProtection is honoured from rds-5a onwards, so it has to be accepted +// at create and carried onto the record DeleteDBInstance checks. +func TestValidateCreateRequest_AcceptsDeletionProtection(t *testing.T) { + input := validCreateInput() + input.DeletionProtection = aws.Bool(true) + + req, err := validateCreateRequest(input) + require.NoError(t, err) + assert.True(t, req.DeletionProtection) +} + func TestValidateCreateRequest_RejectsMalformedRequests(t *testing.T) { cases := []struct { name string diff --git a/spinifex/handlers/rds/delete.go b/spinifex/handlers/rds/delete.go new file mode 100644 index 000000000..5f87552c7 --- /dev/null +++ b/spinifex/handlers/rds/delete.go @@ -0,0 +1,395 @@ +package handlers_rds + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns" + "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" + "github.com/mulgadc/spinifex/spinifex/tags" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go/jetstream" +) + +// Takes the D18 final snapshot and answers whether a volume is still +// referenced by one, which is what decides between deleting and retaining it. +type snapshotProvider interface { + CreateSnapshot(ctx context.Context, input *ec2.CreateSnapshotInput, accountID string) (*ec2.Snapshot, error) + DescribeSnapshots(ctx context.Context, input *ec2.DescribeSnapshotsInput, accountID string) (*ec2.DescribeSnapshotsOutput, error) +} + +// Tears the DB instance down. AWS requires the caller to choose explicitly +// between a final snapshot and none, so neither an accidental data loss nor an +// accidental retained volume can happen by omission. +// +// The record is moved to deleting first and every step below tolerates a +// missing resource, so a retried call or a reconciler-resumed teardown +// converges on the same end state rather than failing on what it already did. +func (s *Service) DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInstanceInput, accountID string) (*rds.DeleteDBInstanceOutput, error) { + finalSnapshot, err := validateDeleteRequest(input) + if err != nil { + return nil, err + } + id := aws.StringValue(input.DBInstanceIdentifier) + + kv, err := s.bucket(ctx, accountID) + if err != nil { + return nil, err + } + rec, rev, err := s.getDBInstance(ctx, kv, id) + if err != nil { + return nil, err + } + if rec.DeletionProtection { + return nil, fmt.Errorf("%s: DB instance %s cannot be deleted because deletion protection is enabled", + awserrors.ErrorInvalidParameterCombination, id) + } + if err := s.checkFinalSnapshotAvailable(ctx, kv, rec, finalSnapshot); err != nil { + return nil, err + } + + // A retry of a delete already under way has to repeat its snapshot choice. + // Keeping the first one silently would answer a caller who asked for a final + // snapshot with no snapshot, or snapshot for one who asked to skip. + if rec.Status == StatusDeleting && finalSnapshot != rec.FinalSnapshotIdentifier { + return nil, fmt.Errorf("%s: DB instance %s is already being deleted with %s; a retry must repeat that choice", + awserrors.ErrorDBInstanceInvalidState, id, snapshotChoice(rec.FinalSnapshotIdentifier)) + } + + // A repeat of a delete already under way re-runs the teardown rather than + // being rejected: the first call may have died partway through it. + if rec.Status != StatusDeleting { + now := time.Now().UTC() + rec.Status = StatusDeleting + rec.FailureReason = "" + rec.FinalSnapshotIdentifier = finalSnapshot + rec.TransitionStartedAt = &now + rec.UpdatedAt = now + if err := updateJSON(ctx, kv, DBInstanceKey(id), rev, rec); err != nil { + if errors.Is(err, jetstream.ErrKeyExists) { + return nil, fmt.Errorf("%s: DB instance %s changed state concurrently; retry the delete", + awserrors.ErrorDBInstanceInvalidState, id) + } + return nil, err + } + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, id, "DB instance deleted.", EventCategoryDeletion) + } + + if err := s.teardownDBInstance(ctx, kv, accountID, rec); err != nil { + return nil, err + } + + // The record is gone, so the customer is answered with the last state it + // held — as AWS does, which reports the instance as deleting. + return &rds.DeleteDBInstanceOutput{DBInstance: s.projectDBInstance(rec)}, nil +} + +// One of the two is required, exactly as AWS: omitting both is the request that +// would otherwise silently destroy the only copy of the data. +func validateDeleteRequest(input *rds.DeleteDBInstanceInput) (string, error) { + if input == nil { + return "", fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue) + } + if aws.StringValue(input.DBInstanceIdentifier) == "" { + return "", fmt.Errorf("%s: DBInstanceIdentifier is required", awserrors.ErrorInvalidParameterValue) + } + + skip := aws.BoolValue(input.SkipFinalSnapshot) + identifier := aws.StringValue(input.FinalDBSnapshotIdentifier) + switch { + case skip && identifier != "": + return "", fmt.Errorf("%s: FinalDBSnapshotIdentifier cannot be supplied with SkipFinalSnapshot", + awserrors.ErrorInvalidParameterCombination) + case !skip && identifier == "": + return "", fmt.Errorf("%s: FinalDBSnapshotIdentifier is required unless SkipFinalSnapshot is set", + awserrors.ErrorInvalidParameterCombination) + } + if identifier != "" { + if err := validateDBInstanceIdentifier(identifier); err != nil { + return "", err + } + } + return identifier, nil +} + +// How the in-flight delete's snapshot choice reads back to the customer, in the +// terms they supplied it in. +func snapshotChoice(identifier string) string { + if identifier == "" { + return "SkipFinalSnapshot" + } + return "FinalDBSnapshotIdentifier " + identifier +} + +// A snapshot record outlives the DB instance record it was taken from, so an +// identifier reused by a later instance of the same name collides with the +// earlier one. Only a record naming this same instance and its current data +// volume is a resumed teardown's own work. +func finalSnapshotIsOurs(existing *DBSnapshotRecord, rec *DBInstanceRecord) bool { + return existing.DBInstanceIdentifier == rec.DBInstanceIdentifier && + existing.SourceVolumeID == rec.DataVolumeID +} + +// Rejects a taken identifier before anything is torn down, as AWS does. Leaving +// it to the snapshot step would terminate the VM first and then fail on an +// error no retry can clear. +func (s *Service) checkFinalSnapshotAvailable(ctx context.Context, kv jetstream.KeyValue, rec *DBInstanceRecord, identifier string) error { + if identifier == "" { + return nil + } + var existing DBSnapshotRecord + found, err := getJSON(ctx, kv, DBSnapshotKey(identifier), &existing) + if err != nil || !found { + return err + } + if finalSnapshotIsOurs(&existing, rec) { + return nil + } + return fmt.Errorf("%s: DB snapshot %s already exists", awserrors.ErrorDBSnapshotAlreadyExists, identifier) +} + +// The teardown chain. Ordering is forced by what holds what: the VM has to go +// before its ENIs and volume can be released, and the final snapshot is taken +// once the VM is gone so it reads a sealed data volume rather than one a live +// engine is still writing to. +// +// Every step treats a missing resource as done, so this is safe to re-run from +// any point it stopped at. +func (s *Service) teardownDBInstance(ctx context.Context, kv jetstream.KeyValue, accountID string, rec *DBInstanceRecord) error { + // Asked to stop cleanly even though the VM is about to go: it is what makes + // the final snapshot a clean checkpoint rather than one needing WAL replay. + if rec.FinalSnapshotIdentifier != "" { + s.stopEngineOrRecordFallback(ctx, accountID, rec, "deleting") + } + + if err := s.terminateInstanceVM(ctx, rec.InstanceID); err != nil { + return fmt.Errorf("rds: terminate the VM behind %s: %w", rec.DBInstanceIdentifier, err) + } + if err := s.takeFinalSnapshot(ctx, kv, accountID, rec); err != nil { + return err + } + if err := s.releaseDataVolume(ctx, kv, accountID, rec); err != nil { + return err + } + + s.deleteInstanceENIs(ctx, accountID, rec) + s.publishDNS(ctx, accountID, rec, handlers_dns.ActionDelete) + + if rec.InstanceID != "" { + if err := s.DeleteInstanceIndex(ctx, rec.InstanceID); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("rds: delete the instance index entry for %s: %w", rec.DBInstanceIdentifier, err) + } + } + // Last: while it exists the instance is still nameable, and everything above + // is reachable only through it. + if err := kv.Delete(ctx, DBInstanceKey(rec.DBInstanceIdentifier)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("rds: delete the DB instance record for %s: %w", rec.DBInstanceIdentifier, err) + } + + slog.InfoContext(ctx, "rds: DB instance deleted", + "dbInstance", rec.DBInstanceIdentifier, "accountId", accountID, + "finalSnapshot", rec.FinalSnapshotIdentifier) + return nil +} + +// A VM no node owns is already terminated, which is the state this is trying to +// reach. +func (s *Service) terminateInstanceVM(ctx context.Context, instanceID string) error { + if instanceID == "" { + return nil + } + launcher := s.deps.Launch.Instance + if launcher == nil { + return errors.New("rds: no system-instance launcher configured") + } + if err := launcher.TerminateSystemInstance(instanceID); err != nil && + !errors.Is(err, sysinstance.ErrSystemInstanceNotFound) { + return err + } + return nil +} + +// Records the D18 final snapshot under db-snapshots/{id}. The record is written +// with Create, so a resumed teardown that already took the snapshot does not +// take a second one. +func (s *Service) takeFinalSnapshot(ctx context.Context, kv jetstream.KeyValue, accountID string, rec *DBInstanceRecord) error { + if rec.FinalSnapshotIdentifier == "" || rec.DataVolumeID == "" { + return nil + } + key := DBSnapshotKey(rec.FinalSnapshotIdentifier) + var existing DBSnapshotRecord + found, err := getJSON(ctx, kv, key, &existing) + if err != nil { + return err + } + if found { + if !finalSnapshotIsOurs(&existing, rec) { + return fmt.Errorf("%s: DB snapshot %s already exists", + awserrors.ErrorDBSnapshotAlreadyExists, rec.FinalSnapshotIdentifier) + } + return nil + } + if s.deps.Snapshots == nil { + return errors.New("rds: no snapshot service configured") + } + + // System-owned, like the volume it is taken from: the customer addresses it + // by its RDS identifier, never by the EC2 snapshot underneath. + snapshot, err := s.deps.Snapshots.CreateSnapshot(ctx, &ec2.CreateSnapshotInput{ + VolumeId: aws.String(rec.DataVolumeID), + Description: aws.String("RDS final snapshot for " + rec.DBInstanceIdentifier), + TagSpecifications: []*ec2.TagSpecification{{ + ResourceType: aws.String("snapshot"), + Tags: []*ec2.Tag{ + {Key: aws.String(tags.ManagedByKey), Value: aws.String(tags.ManagedByRDS)}, + {Key: aws.String(rdsInstanceTagKey), Value: aws.String(rec.DBInstanceIdentifier)}, + }, + }}, + }, utils.GlobalAccountID) + if err != nil { + return fmt.Errorf("rds: take the final snapshot of %s: %w", rec.DBInstanceIdentifier, err) + } + if snapshot == nil || aws.StringValue(snapshot.SnapshotId) == "" { + return fmt.Errorf("rds: take the final snapshot of %s: empty snapshot id", rec.DBInstanceIdentifier) + } + + // Everything a restore needs is copied here rather than referenced: the DB + // instance record is deleted moments later. + record := DBSnapshotRecord{ + DBSnapshotIdentifier: rec.FinalSnapshotIdentifier, + DBInstanceIdentifier: rec.DBInstanceIdentifier, + AccountID: accountID, + SnapshotType: SnapshotTypeManual, + Status: SnapshotStatusAvailable, + SnapshotID: aws.StringValue(snapshot.SnapshotId), + SourceVolumeID: rec.DataVolumeID, + Engine: rec.Engine, + EngineVersion: rec.EngineVersion, + AllocatedStorage: rec.AllocatedStorage, + StorageType: rec.StorageType, + StorageEncrypted: rec.StorageEncrypted, + MasterUsername: rec.MasterUsername, + Port: rec.Port, + VpcID: rec.VpcID, + Tags: rec.Tags, + CreatedAt: time.Now().UTC(), + } + if err := createJSON(ctx, kv, key, &record); err != nil && !errors.Is(err, jetstream.ErrKeyExists) { + return fmt.Errorf("rds: record the final snapshot of %s: %w", rec.DBInstanceIdentifier, err) + } + s.RecordEvent(ctx, accountID, EventSourceTypeDBSnapshot, rec.FinalSnapshotIdentifier, + "Final DB snapshot created.", EventCategoryBackup, EventCategoryCreation) + return nil +} + +// A viperblock snapshot references its source volume's chunk files rather than +// copying them, so the volume cannot be deleted while any snapshot survives — +// including the final one just taken. It is retained instead, recorded with the +// snapshots holding it so the last DeleteDBSnapshot can release it (D10). +func (s *Service) releaseDataVolume(ctx context.Context, kv jetstream.KeyValue, accountID string, rec *DBInstanceRecord) error { + if rec.DataVolumeID == "" { + return nil + } + holders, err := s.snapshotsHolding(ctx, rec.DataVolumeID) + if err != nil { + return err + } + if len(holders) > 0 { + return s.retainDataVolume(ctx, kv, accountID, rec, holders, false) + } + + if s.deps.Launch.Volume == nil { + return errors.New("rds: no volume service configured") + } + _, err = s.deps.Launch.Volume.DeleteVolume(ctx, &ec2.DeleteVolumeInput{ + VolumeId: aws.String(rec.DataVolumeID), + }, utils.GlobalAccountID) + switch { + case err == nil || awserrors.IsNotFound(err): + return nil + case awserrors.IsErrorCode(err, awserrors.ErrorVolumeInUse): + // The volume store's own snapshot index sees a reference the enumeration + // above did not. Retaining is the safe reading of that disagreement; + // returning the error would wedge the delete on every retry instead. + return s.retainDataVolume(ctx, kv, accountID, rec, holders, true) + default: + return fmt.Errorf("rds: delete the data volume of %s: %w", rec.DBInstanceIdentifier, err) + } +} + +// Records a volume the teardown left behind, with the snapshots holding it so +// the last DeleteDBSnapshot can release it (D10). holdersUnresolved marks the +// case where the volume store refused the delete but named nobody, so a release +// has to re-check rather than trust an empty list. +func (s *Service) retainDataVolume(ctx context.Context, kv jetstream.KeyValue, accountID string, + rec *DBInstanceRecord, holders []string, holdersUnresolved bool) error { + retained := RetainedVolumeRecord{ + VolumeID: rec.DataVolumeID, + AccountID: accountID, + DBInstanceIdentifier: rec.DBInstanceIdentifier, + Snapshots: holders, + HoldersUnresolved: holdersUnresolved, + RetainedAt: time.Now().UTC(), + } + if err := putJSON(ctx, kv, RetainedVolumeKey(rec.DataVolumeID), &retained); err != nil { + return err + } + slog.InfoContext(ctx, "rds: data volume retained; snapshots still reference its chunks", + "dbInstance", rec.DBInstanceIdentifier, "volumeId", rec.DataVolumeID, + "snapshots", holders, "holdersUnresolved", holdersUnresolved) + return nil +} + +// The RDS snapshot identifiers whose EC2 snapshot references volumeID. Read +// from EC2 rather than from the RDS key space, because that is what +// DeleteVolume itself enforces against. +func (s *Service) snapshotsHolding(ctx context.Context, volumeID string) ([]string, error) { + if s.deps.Snapshots == nil { + return nil, errors.New("rds: no snapshot service configured") + } + out, err := s.deps.Snapshots.DescribeSnapshots(ctx, &ec2.DescribeSnapshotsInput{ + Filters: []*ec2.Filter{{ + Name: aws.String("volume-id"), + Values: aws.StringSlice([]string{volumeID}), + }}, + }, utils.GlobalAccountID) + if err != nil { + if awserrors.IsNotFound(err) { + return nil, nil + } + return nil, fmt.Errorf("rds: list the snapshots of %s: %w", volumeID, err) + } + if out == nil { + return nil, nil + } + holders := make([]string, 0, len(out.Snapshots)) + for _, snapshot := range out.Snapshots { + if id := aws.StringValue(snapshot.SnapshotId); id != "" { + holders = append(holders, id) + } + } + return holders, nil +} + +// The customer ENI is deleted rather than retained: it is the endpoint of an +// instance that no longer exists. Both NICs are removed explicitly because the +// launch created them explicitly, and a leaked ENI holds its subnet address. +func (s *Service) deleteInstanceENIs(ctx context.Context, accountID string, rec *DBInstanceRecord) { + vpcSvc := s.deps.Launch.VPC + if vpcSvc == nil { + return + } + if rec.ENIID != "" { + deleteLaunchENI(ctx, vpcSvc, accountID, rec.ENIID) + } + if rec.SystemENIID != "" { + deleteLaunchENI(ctx, vpcSvc, utils.GlobalAccountID, rec.SystemENIID) + } +} diff --git a/spinifex/handlers/rds/delete_test.go b/spinifex/handlers/rds/delete_test.go new file mode 100644 index 000000000..51525d325 --- /dev/null +++ b/spinifex/handlers/rds/delete_test.go @@ -0,0 +1,312 @@ +package handlers_rds + +import ( + "errors" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func (h *lifecycleHarness) recordExists(t *testing.T, id string) bool { + t.Helper() + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var rec DBInstanceRecord + found, err := getJSON(t.Context(), kv, DBInstanceKey(id), &rec) + require.NoError(t, err) + return found +} + +func (h *lifecycleHarness) snapshotRecord(t *testing.T, id string) (DBSnapshotRecord, bool) { + t.Helper() + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var rec DBSnapshotRecord + found, err := getJSON(t.Context(), kv, DBSnapshotKey(id), &rec) + require.NoError(t, err) + return rec, found +} + +func (h *lifecycleHarness) retainedVolume(t *testing.T, volumeID string) (RetainedVolumeRecord, bool) { + t.Helper() + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var rec RetainedVolumeRecord + found, err := getJSON(t.Context(), kv, RetainedVolumeKey(volumeID), &rec) + require.NoError(t, err) + return rec, found +} + +func skipFinalSnapshot() *rds.DeleteDBInstanceInput { + return &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + SkipFinalSnapshot: aws.Bool(true), + } +} + +// The whole teardown: the VM, the data volume, both NICs, the reverse index and +// the record itself. +func TestDeleteDBInstance_TearsDownEverythingItOwns(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + require.NoError(t, h.svc.PutInstanceIndex(t.Context(), testInstance, InstanceIndexEntry{ + AccountID: testAccountID, + DBInstanceIdentifier: testDBID, + })) + + out, err := h.svc.DeleteDBInstance(t.Context(), skipFinalSnapshot(), testAccountID) + require.NoError(t, err) + + // Answered with the last state it held, as AWS does. + assert.Equal(t, string(StatusDeleting), aws.StringValue(out.DBInstance.DBInstanceStatus)) + + assert.Equal(t, []string{testInstance}, h.launcher.terminated) + assert.Equal(t, []string{"vol-rdsdata01"}, h.volumes.deleted) + assert.ElementsMatch(t, []string{"eni-cust01", "eni-sys01"}, h.enis.deleted) + assert.False(t, h.recordExists(t, testDBID)) + + entry, err := h.svc.LookupInstanceIndex(t.Context(), testInstance) + require.NoError(t, err) + assert.Nil(t, entry, "the reverse index must not outlive the instance") +} + +// AWS requires the caller to choose explicitly, because the request that omits +// both would silently destroy the only copy of the data. +func TestDeleteDBInstance_RequiresAnExplicitSnapshotChoice(t *testing.T) { + cases := []struct { + name string + input *rds.DeleteDBInstanceInput + }{ + {"NeitherSupplied", &rds.DeleteDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}}, + {"BothSupplied", &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + SkipFinalSnapshot: aws.Bool(true), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + }}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.DeleteDBInstance(t.Context(), tc.input, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterCombination) + assert.True(t, h.recordExists(t, testDBID), "a rejected delete must leave the instance alone") + assert.Empty(t, h.launcher.terminated) + }) + } +} + +// The flag exists to stop exactly this call, so honouring it has to happen +// before anything is torn down. +func TestDeleteDBInstance_HonoursDeletionProtection(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.DeletionProtection = true + seedInstance(t, h.svc, rec) + + _, err := h.svc.DeleteDBInstance(t.Context(), skipFinalSnapshot(), testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "deletion protection") + + assert.Equal(t, StatusAvailable, h.record(t).Status) + assert.Empty(t, h.launcher.terminated) + assert.Empty(t, h.volumes.deleted) +} + +// D18: the snapshot is taken once the VM is gone, so it reads a sealed data +// volume rather than one a live engine is still writing to. +func TestDeleteDBInstance_TakesTheFinalSnapshotAfterTheEngineAndVMAreDown(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + }, testAccountID) + require.NoError(t, err) + + issued := h.agent.received() + require.Len(t, issued, 1) + assert.Equal(t, CommandStopEngine, issued[0].Type) + require.Len(t, h.snaps.created, 1) + assert.Equal(t, "vol-rdsdata01", aws.StringValue(h.snaps.created[0].VolumeId)) + + // Everything a restore needs is copied onto the snapshot record: the DB + // instance record is deleted moments later. + snapshot, found := h.snapshotRecord(t, "orders-db-final") + require.True(t, found) + assert.Equal(t, testDBID, snapshot.DBInstanceIdentifier) + assert.Equal(t, SnapshotTypeManual, snapshot.SnapshotType) + assert.Equal(t, "snap-0001", snapshot.SnapshotID) + assert.Equal(t, "postgres", snapshot.Engine) + assert.Equal(t, "vol-rdsdata01", snapshot.SourceVolumeID) +} + +// D10: a snapshot references its source volume's chunks, so the volume cannot +// be deleted while the final snapshot survives. It is retained and recorded +// with the snapshots holding it. +func TestDeleteDBInstance_RetainsTheVolumeAFinalSnapshotStillHolds(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + }, testAccountID) + require.NoError(t, err) + + assert.Empty(t, h.volumes.deleted, "a volume a snapshot references must not be deleted") + retained, found := h.retainedVolume(t, "vol-rdsdata01") + require.True(t, found) + assert.Equal(t, testDBID, retained.DBInstanceIdentifier) + assert.Equal(t, []string{"snap-0001"}, retained.Snapshots) +} + +// A snapshot record outlives the instance it came from, so an instance +// re-created under the same name must not have a stale snapshot of a previous +// data volume accepted as its own — the volume would then be deleted unbacked. +func TestDeleteDBInstance_RejectsAFinalSnapshotIdentifierAlreadyTaken(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + // The snapshot left behind by the earlier instance of the same name. + require.NoError(t, putJSON(t.Context(), kv, DBSnapshotKey("orders-db-final"), &DBSnapshotRecord{ + DBSnapshotIdentifier: "orders-db-final", + DBInstanceIdentifier: testDBID, + SourceVolumeID: "vol-rdsdata00", + })) + + _, err = h.svc.DeleteDBInstance(t.Context(), &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + }, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorDBSnapshotAlreadyExists) + + // Rejected before anything was torn down, so the customer loses nothing. + assert.True(t, h.recordExists(t, testDBID)) + assert.Empty(t, h.launcher.terminated) + assert.Empty(t, h.volumes.deleted) + assert.Empty(t, h.snaps.created) +} + +// Keeping the in-flight choice silently would answer a caller who asked for a +// final snapshot with no snapshot at all. +func TestDeleteDBInstance_RejectsARetryThatChangesTheSnapshotChoice(t *testing.T) { + cases := []struct { + name string + inFlight string + retry *rds.DeleteDBInstanceInput + }{ + {"SnapshotRetryOfASkippedDelete", "", &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + }}, + {"SkipRetryOfASnapshottedDelete", "orders-db-final", skipFinalSnapshot()}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.Status = StatusDeleting + rec.FinalSnapshotIdentifier = tc.inFlight + seedInstance(t, h.svc, rec) + + _, err := h.svc.DeleteDBInstance(t.Context(), tc.retry, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceInvalidState) + assert.Contains(t, err.Error(), snapshotChoice(tc.inFlight)) + + assert.True(t, h.recordExists(t, testDBID)) + assert.Empty(t, h.launcher.terminated) + assert.Empty(t, h.volumes.deleted) + }) + } +} + +// The volume store enforces against its own snapshot index, which can see a +// reference the EC2 enumeration missed. Retaining converges; erroring would +// wedge the delete on every retry until the bound marked it failed. +func TestDeleteDBInstance_RetainsTheVolumeWhenTheVolumeStoreReportsItInUse(t *testing.T) { + h := newLifecycleHarness(t, false) + h.volumes.deleteErr = errors.New(awserrors.ErrorVolumeInUse) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.DeleteDBInstance(t.Context(), skipFinalSnapshot(), testAccountID) + require.NoError(t, err) + + retained, found := h.retainedVolume(t, "vol-rdsdata01") + require.True(t, found, "a volume the store refused to delete must be recorded, not forgotten") + assert.True(t, retained.HoldersUnresolved, + "nothing named a holder, so a release must re-check rather than trust the empty list") + assert.Equal(t, testDBID, retained.DBInstanceIdentifier) + + // The teardown still completed: the record is what makes the volume findable. + assert.False(t, h.recordExists(t, testDBID)) +} + +// A teardown that died partway through is replayed, so every step has to treat +// a resource that is already gone as work it has done. +func TestDeleteDBInstance_IsIdempotentAcrossARetriedTeardown(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + // The shape a delete leaves behind when its caller died mid-teardown. + rec.Status = StatusDeleting + rec.FinalSnapshotIdentifier = "orders-db-final" + seedInstance(t, h.svc, rec) + + input := &rds.DeleteDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + FinalDBSnapshotIdentifier: aws.String("orders-db-final"), + } + _, err := h.svc.DeleteDBInstance(t.Context(), input, testAccountID) + require.NoError(t, err) + + // The record left behind by the interrupted attempt, replayed. + seedInstance(t, h.svc, rec) + _, err = h.svc.DeleteDBInstance(t.Context(), input, testAccountID) + require.NoError(t, err) + + assert.Len(t, h.snaps.created, 1, "a replayed teardown must not take a second final snapshot") + assert.False(t, h.recordExists(t, testDBID)) +} + +// The reconciler owns a teardown whose caller never came back, so an +// interrupted delete finishes without an operator. +func TestReconciler_ResumesAnInterruptedDelete(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.Status = StatusDeleting + seedInstance(t, h.svc, rec) + + reconciler := NewReconciler(h.svc, "node-a") + require.NoError(t, reconciler.reconcileOnce(t.Context())) + + assert.Equal(t, []string{testInstance}, h.launcher.terminated) + assert.False(t, h.recordExists(t, testDBID)) +} + +// A stop whose caller died leaves the VM possibly still running, so the stop is +// re-issued rather than assumed. +func TestReconciler_ResumesAnInterruptedStop(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.Status = StatusStopping + seedInstance(t, h.svc, rec) + + reconciler := NewReconciler(h.svc, "node-a") + require.NoError(t, reconciler.reconcileOnce(t.Context())) + + assert.Equal(t, []string{"stop:" + testInstance}, h.cmdr.calls) + assert.Equal(t, StatusStopped, h.record(t).Status) +} diff --git a/spinifex/handlers/rds/describe.go b/spinifex/handlers/rds/describe.go index 547acbd97..6aaf98e7d 100644 --- a/spinifex/handlers/rds/describe.go +++ b/spinifex/handlers/rds/describe.go @@ -85,6 +85,7 @@ func (s *Service) projectDBInstance(rec *DBInstanceRecord) *rds.DBInstance { MasterUsername: aws.String(rec.MasterUsername), MultiAZ: aws.Bool(false), PubliclyAccessible: aws.Bool(false), + DeletionProtection: aws.Bool(rec.DeletionProtection), InstanceCreateTime: aws.Time(rec.CreatedAt), // The Terraform provider reads tags from the describe as well as from // ListTagsForResource, so the two have to agree. diff --git a/spinifex/handlers/rds/events.go b/spinifex/handlers/rds/events.go new file mode 100644 index 000000000..ed81d2e9a --- /dev/null +++ b/spinifex/handlers/rds/events.go @@ -0,0 +1,314 @@ +package handlers_rds + +import ( + "context" + "errors" + "fmt" + "log/slog" + "slices" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/nats-io/nats.go/jetstream" +) + +// The event ring is the only channel some facts have to the customer — a backup +// that could not be quiesced, a modify applied hours after it was requested, an +// engine that could not be shut down cleanly. Five phases write to it; this one +// owns the mechanism. +const ( + // AWS's own retention. Anything older is dropped on the next append and + // never returned by a read. + EventRetention = 14 * 24 * time.Hour + + // The per-resource bound. Deep enough that a customer polling at AWS's + // default one-hour window never misses an event, shallow enough that a + // resource's whole history stays one KV value. + eventRingSize = 100 + + // A resource can be appended to from the API path and the reconciler at + // once, so the CAS is retried rather than dropped on first contention. + eventWriteAttempts = 8 + + // AWS's defaults for the read: a one-hour window and a 100-record page. + defaultEventDuration = time.Hour + maxEventRecords = 100 +) + +// AWS's SourceType values. Only the ones some phase actually writes are +// defined; an unknown one from a request is rejected rather than matched +// against nothing, which would look like a resource with no history. +const ( + EventSourceTypeDBInstance = "db-instance" + EventSourceTypeDBSnapshot = "db-snapshot" +) + +// AWS's event categories. A customer filters and subscribes on these, so the +// strings are AWS's exactly, spaces included. +const ( + EventCategoryAvailability = "availability" + EventCategoryBackup = "backup" + EventCategoryConfigurationChange = "configuration change" + EventCategoryCreation = "creation" + EventCategoryDeletion = "deletion" + EventCategoryFailure = "failure" + EventCategoryNotification = "notification" + EventCategoryRecovery = "recovery" +) + +func validEventSourceType(sourceType string) bool { + switch sourceType { + case EventSourceTypeDBInstance, EventSourceTypeDBSnapshot: + return true + default: + return false + } +} + +// One entry in a resource's ring. +type Event struct { + SourceIdentifier string `json:"sourceIdentifier"` + SourceType string `json:"sourceType"` + Message string `json:"message"` + Categories []string `json:"categories,omitempty"` + Date time.Time `json:"date"` +} + +// Oldest first, so an append is a tail write and a trim is a head drop. +type eventRing struct { + Events []Event `json:"events"` +} + +// Best-effort by design: an event is a report about work that already happened, +// so failing the operation because its narration could not be stored would turn +// a successful stop into a failed one. Failures are logged instead. +func (s *Service) RecordEvent(ctx context.Context, accountID, sourceType, sourceIdentifier, message string, categories ...string) { + if err := s.appendEvent(ctx, accountID, sourceType, sourceIdentifier, message, categories); err != nil { + slog.WarnContext(ctx, "rds: recording an event failed", + "sourceType", sourceType, "sourceId", sourceIdentifier, "message", message, "err", err) + } +} + +func (s *Service) appendEvent(ctx context.Context, accountID, sourceType, sourceIdentifier, message string, categories []string) error { + kv, err := s.bucket(ctx, accountID) + if err != nil { + return err + } + key := EventRingKey(sourceType, sourceIdentifier) + event := Event{ + SourceIdentifier: sourceIdentifier, + SourceType: sourceType, + Message: message, + Categories: categories, + Date: time.Now().UTC(), + } + + for range eventWriteAttempts { + var ring eventRing + rev, found, err := getJSONRevision(ctx, kv, key, &ring) + if err != nil { + return err + } + ring.Events = trimEvents(append(ring.Events, event)) + + if !found { + if err := createJSON(ctx, kv, key, &ring); err == nil { + return nil + } else if !errors.Is(err, jetstream.ErrKeyExists) { + return err + } + // Another writer created the ring first; re-read and append to it. + continue + } + err = updateJSON(ctx, kv, key, rev, &ring) + if err == nil { + return nil + } + // jetstream.ErrKeyExists on Update is a revision mismatch, not a duplicate. + if !errors.Is(err, jetstream.ErrKeyExists) { + return err + } + } + return fmt.Errorf("rds: event append on %s contended after %d attempts", key, eventWriteAttempts) +} + +// Drops anything past the retention window, then anything past the ring bound. +// Age is checked first so a quiet resource's stale history is discarded even +// when the ring never fills. +func trimEvents(events []Event) []Event { + cutoff := time.Now().UTC().Add(-EventRetention) + kept := events[:0] + for _, event := range events { + if event.Date.After(cutoff) { + kept = append(kept, event) + } + } + if len(kept) > eventRingSize { + kept = kept[len(kept)-eventRingSize:] + } + return kept +} + +// The customer view of the ring. AWS scopes a read to a time window rather than +// to a resource, so an unfiltered call reports the whole account's recent +// history, including resources that have since been deleted. +func (s *Service) DescribeEvents(ctx context.Context, input *rds.DescribeEventsInput, accountID string) (*rds.DescribeEventsOutput, error) { + window, err := eventWindow(input) + if err != nil { + return nil, err + } + sourceType := aws.StringValue(input.SourceType) + sourceIdentifier := aws.StringValue(input.SourceIdentifier) + if sourceType != "" && !validEventSourceType(sourceType) { + return nil, fmt.Errorf("%s: SourceType %q is not an RDS event source", awserrors.ErrorInvalidParameterValue, sourceType) + } + // AWS's own rule: an identifier is meaningless without the type, since two + // resource kinds may share a name. + if sourceIdentifier != "" && sourceType == "" { + return nil, fmt.Errorf("%s: SourceType is required when SourceIdentifier is supplied", + awserrors.ErrorInvalidParameterCombination) + } + + kv, err := s.bucket(ctx, accountID) + if err != nil { + return nil, err + } + keys, err := eventRingKeys(ctx, kv, sourceType, sourceIdentifier) + if err != nil { + return nil, err + } + + wanted := aws.StringValueSlice(input.EventCategories) + var events []Event + for _, key := range keys { + var ring eventRing + found, err := getJSON(ctx, kv, key, &ring) + if err != nil { + return nil, err + } + if !found { + continue + } + for _, event := range ring.Events { + if event.Date.Before(window.start) || event.Date.After(window.end) { + continue + } + if !matchesCategories(event, wanted) { + continue + } + events = append(events, event) + } + } + + // Oldest first across resources, as AWS returns them, so a client that has + // already read up to a timestamp can resume from it. + slices.SortFunc(events, func(a, b Event) int { + if cmp := a.Date.Compare(b.Date); cmp != 0 { + return cmp + } + return strings.Compare(a.SourceIdentifier, b.SourceIdentifier) + }) + if limit := eventRecordLimit(input); len(events) > limit { + events = events[:limit] + } + + out := &rds.DescribeEventsOutput{Events: make([]*rds.Event, 0, len(events))} + for _, event := range events { + out.Events = append(out.Events, s.projectEvent(accountID, event)) + } + return out, nil +} + +func (s *Service) projectEvent(accountID string, event Event) *rds.Event { + out := &rds.Event{ + SourceIdentifier: aws.String(event.SourceIdentifier), + SourceType: aws.String(event.SourceType), + Message: aws.String(event.Message), + Date: aws.Time(event.Date), + EventCategories: aws.StringSlice(event.Categories), + } + switch event.SourceType { + case EventSourceTypeDBInstance: + out.SourceArn = aws.String(DBInstanceARN(s.region, accountID, event.SourceIdentifier)) + case EventSourceTypeDBSnapshot: + out.SourceArn = aws.String(DBSnapshotARN(s.region, accountID, event.SourceIdentifier)) + } + return out +} + +// The window a read covers. AWS resolves Duration against "now" and lets +// explicit StartTime/EndTime override it. +type timeWindow struct { + start, end time.Time +} + +func eventWindow(input *rds.DescribeEventsInput) (timeWindow, error) { + now := time.Now().UTC() + window := timeWindow{start: now.Add(-defaultEventDuration), end: now} + + if minutes := aws.Int64Value(input.Duration); minutes > 0 { + window.start = now.Add(-time.Duration(minutes) * time.Minute) + } + if input.StartTime != nil { + window.start = input.StartTime.UTC() + } + if input.EndTime != nil { + window.end = input.EndTime.UTC() + } + if window.end.Before(window.start) { + return timeWindow{}, fmt.Errorf("%s: EndTime is before StartTime", awserrors.ErrorInvalidParameterCombination) + } + return window, nil +} + +func eventRecordLimit(input *rds.DescribeEventsInput) int { + requested := aws.Int64Value(input.MaxRecords) + if requested <= 0 || requested > maxEventRecords { + return maxEventRecords + } + return int(requested) +} + +// A fully qualified read is one Get. Anything broader has to enumerate, because +// a ring outlives the record that would otherwise name it. +func eventRingKeys(ctx context.Context, kv jetstream.KeyValue, sourceType, sourceIdentifier string) ([]string, error) { + if sourceType != "" && sourceIdentifier != "" { + return []string{EventRingKey(sourceType, sourceIdentifier)}, nil + } + prefix := EventsPrefix() + if sourceType != "" { + prefix += sourceType + "/" + } + keys, err := kv.Keys(ctx) + if err != nil { + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return nil, fmt.Errorf("rds: list event rings: %w", err) + } + matched := make([]string, 0, len(keys)) + for _, key := range keys { + if strings.HasPrefix(key, prefix) { + matched = append(matched, key) + } + } + slices.Sort(matched) + return matched, nil +} + +// An empty filter matches everything, as AWS does; otherwise the event needs at +// least one of the requested categories. +func matchesCategories(event Event, wanted []string) bool { + if len(wanted) == 0 { + return true + } + for _, category := range wanted { + if slices.Contains(event.Categories, category) { + return true + } + } + return false +} diff --git a/spinifex/handlers/rds/events_test.go b/spinifex/handlers/rds/events_test.go new file mode 100644 index 000000000..7cfeb75fe --- /dev/null +++ b/spinifex/handlers/rds/events_test.go @@ -0,0 +1,178 @@ +package handlers_rds + +import ( + "context" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Writes an event dated at when, which the append path cannot do on its own — +// it always stamps "now". +func seedEvent(t *testing.T, svc *Service, sourceType, sourceID string, when time.Time, message string, categories ...string) { + t.Helper() + kv, err := svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + key := EventRingKey(sourceType, sourceID) + + var ring eventRing + _, err = getJSON(t.Context(), kv, key, &ring) + require.NoError(t, err) + ring.Events = append(ring.Events, Event{ + SourceIdentifier: sourceID, + SourceType: sourceType, + Message: message, + Categories: categories, + Date: when, + }) + require.NoError(t, putJSON(t.Context(), kv, key, &ring)) +} + +func describeEvents(t *testing.T, svc *Service, input *rds.DescribeEventsInput) []*rds.Event { + t.Helper() + out, err := svc.DescribeEvents(t.Context(), input, testAccountID) + require.NoError(t, err) + return out.Events +} + +func TestDescribeEvents_ReturnsARecordedEventWithItsSourceARN(t *testing.T) { + svc := newTestService(t) + svc.RecordEvent(context.Background(), testAccountID, EventSourceTypeDBInstance, testDBID, + "DB instance stopped.", EventCategoryAvailability) + + events := describeEvents(t, svc, &rds.DescribeEventsInput{}) + require.Len(t, events, 1) + assert.Equal(t, testDBID, aws.StringValue(events[0].SourceIdentifier)) + assert.Equal(t, EventSourceTypeDBInstance, aws.StringValue(events[0].SourceType)) + assert.Equal(t, "DB instance stopped.", aws.StringValue(events[0].Message)) + assert.Equal(t, []string{EventCategoryAvailability}, aws.StringValueSlice(events[0].EventCategories)) + assert.Equal(t, DBInstanceARN(testRegion, testAccountID, testDBID), aws.StringValue(events[0].SourceArn)) +} + +// AWS's default window is one hour, so an event outside it is not returned even +// though it is still inside the retention window. +func TestDescribeEvents_ScopesToTheRequestedWindow(t *testing.T) { + svc := newTestService(t) + now := time.Now().UTC() + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, now.Add(-10*time.Minute), "recent") + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, now.Add(-6*time.Hour), "older") + + assert.Equal(t, []string{"recent"}, messagesOf(describeEvents(t, svc, &rds.DescribeEventsInput{}))) + + widened := describeEvents(t, svc, &rds.DescribeEventsInput{Duration: aws.Int64(720)}) + assert.Equal(t, []string{"older", "recent"}, messagesOf(widened), "oldest first, as AWS returns them") +} + +func TestDescribeEvents_FiltersByCategory(t *testing.T) { + svc := newTestService(t) + now := time.Now().UTC() + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, now.Add(-time.Minute), "backed up", EventCategoryBackup) + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, now, "restarted", EventCategoryAvailability) + + filtered := describeEvents(t, svc, &rds.DescribeEventsInput{ + EventCategories: aws.StringSlice([]string{EventCategoryBackup}), + }) + assert.Equal(t, []string{"backed up"}, messagesOf(filtered)) +} + +// An unfiltered read is account-wide, so a resource's history is reachable even +// after the resource itself is gone. +func TestDescribeEvents_ReadsAcrossResourcesAndScopesToOneWhenAsked(t *testing.T) { + svc := newTestService(t) + now := time.Now().UTC() + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, now, "instance event") + seedEvent(t, svc, EventSourceTypeDBSnapshot, "orders-db-final", now, "snapshot event") + + assert.ElementsMatch(t, []string{"instance event", "snapshot event"}, + messagesOf(describeEvents(t, svc, &rds.DescribeEventsInput{}))) + + scoped := describeEvents(t, svc, &rds.DescribeEventsInput{ + SourceType: aws.String(EventSourceTypeDBSnapshot), + SourceIdentifier: aws.String("orders-db-final"), + }) + assert.Equal(t, []string{"snapshot event"}, messagesOf(scoped)) +} + +func TestDescribeEvents_RejectsMalformedRequests(t *testing.T) { + cases := []struct { + name string + input *rds.DescribeEventsInput + wantErr string + }{ + // Two resource kinds may share a name, so an identifier alone is + // ambiguous — AWS's own rule. + {"IdentifierWithoutType", &rds.DescribeEventsInput{ + SourceIdentifier: aws.String(testDBID), + }, awserrors.ErrorInvalidParameterCombination}, + {"UnknownSourceType", &rds.DescribeEventsInput{ + SourceType: aws.String("db-cluster"), + }, awserrors.ErrorInvalidParameterValue}, + {"EndBeforeStart", &rds.DescribeEventsInput{ + StartTime: aws.Time(time.Now()), + EndTime: aws.Time(time.Now().Add(-time.Hour)), + }, awserrors.ErrorInvalidParameterCombination}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + svc := newTestService(t) + _, err := svc.DescribeEvents(t.Context(), tc.input, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + }) + } +} + +// The ring is bounded, so a resource's whole history stays one KV value however +// long it lives. +func TestAppendEvent_BoundsTheRing(t *testing.T) { + svc := newTestService(t) + for range eventRingSize + 10 { + require.NoError(t, svc.appendEvent(context.Background(), testAccountID, + EventSourceTypeDBInstance, testDBID, "beat", nil)) + } + + kv, err := svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var ring eventRing + found, err := getJSON(t.Context(), kv, EventRingKey(EventSourceTypeDBInstance, testDBID), &ring) + require.NoError(t, err) + require.True(t, found) + assert.Len(t, ring.Events, eventRingSize) +} + +// AWS's 14-day retention: anything older is dropped on the next append and +// never returned by a read. +func TestAppendEvent_DropsEventsPastTheRetentionWindow(t *testing.T) { + svc := newTestService(t) + seedEvent(t, svc, EventSourceTypeDBInstance, testDBID, + time.Now().UTC().Add(-EventRetention-time.Hour), "ancient") + + require.NoError(t, svc.appendEvent(context.Background(), testAccountID, + EventSourceTypeDBInstance, testDBID, "fresh", nil)) + + events := describeEvents(t, svc, &rds.DescribeEventsInput{Duration: aws.Int64(60 * 24 * 30)}) + assert.Equal(t, []string{"fresh"}, messagesOf(events)) +} + +// An event narrates work that already happened, so a failed append must not +// turn a successful operation into a failed one. +func TestRecordEvent_DoesNotPropagateAFailure(t *testing.T) { + svc := NewService(nil, testRegion) + assert.NotPanics(t, func() { + svc.RecordEvent(context.Background(), testAccountID, EventSourceTypeDBInstance, testDBID, "stopped") + }) +} + +func messagesOf(events []*rds.Event) []string { + messages := make([]string, 0, len(events)) + for _, event := range events { + messages = append(messages, aws.StringValue(event.Message)) + } + return messages +} diff --git a/spinifex/handlers/rds/launch_test.go b/spinifex/handlers/rds/launch_test.go index febe28142..014093357 100644 --- a/spinifex/handlers/rds/launch_test.go +++ b/spinifex/handlers/rds/launch_test.go @@ -264,6 +264,10 @@ type fakeVolumes struct { unwind *[]string deleteCtxErr error + + // What DeleteVolume refuses with, which is how the volume store's own + // snapshot index is made to disagree with the snapshot enumeration. + deleteErr error } var _ launchVolumeProvisioner = (*fakeVolumes)(nil) @@ -283,6 +287,9 @@ func (f *fakeVolumes) DeleteVolume(ctx context.Context, in *ec2.DeleteVolumeInpu if f.unwind != nil { *f.unwind = append(*f.unwind, "delete-volume") } + if f.deleteErr != nil { + return nil, f.deleteErr + } return &ec2.DeleteVolumeOutput{}, nil } diff --git a/spinifex/handlers/rds/lifecycle.go b/spinifex/handlers/rds/lifecycle.go new file mode 100644 index 000000000..85bddd868 --- /dev/null +++ b/spinifex/handlers/rds/lifecycle.go @@ -0,0 +1,359 @@ +package handlers_rds + +import ( + "context" + "errors" + "fmt" + "log/slog" + "slices" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + "github.com/mulgadc/spinifex/spinifex/types" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// The DB VM's power state is driven over the per-instance EC2 command path +// under the system account, not over a system.* subject: the system dispatch +// surface is launch and terminate only, and the ownership check on ec2.cmd +// passes because the VM is system-owned. +type instanceCommander interface { + StopInstance(ctx context.Context, instanceID string) error + RebootInstance(ctx context.Context, instanceID string) error + // Starts a VM the owning node still holds. Returns ErrInstanceNotOnNode when + // no node has it in memory, which is the normal case after a node restart. + StartInstance(ctx context.Context, instanceID string) error + // Relaunches a VM whose owning node no longer holds it, from its persisted + // stopped-instance record. + StartStoppedInstance(ctx context.Context, instanceID string) error +} + +// No node subscribes ec2.cmd.{instanceID} for this VM, so its power state can +// only be changed through the stopped-instance path. +var ErrInstanceNotOnNode = errors.New("rds: no node is holding this DB VM") + +// How long a VM stop, start or reboot may take before the command is treated as +// lost. A stop drains and seals the data volume, which is the long pole. +const instanceCommandTimeout = 90 * time.Second + +// Reboots the engine, applying any static parameters stored pending-reboot +// (D16). ForceFailover is rejected outright: there is no standby to fail over +// to, and silently ignoring it would report a failover that never happened. +func (s *Service) RebootDBInstance(ctx context.Context, input *rds.RebootDBInstanceInput, accountID string) (*rds.RebootDBInstanceOutput, error) { + if input == nil { + return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue) + } + if aws.BoolValue(input.ForceFailover) { + return nil, unimplemented("ForceFailover", "this platform is single-AZ; there is no standby to fail over to") + } + + rec, kv, err := s.beginTransition(ctx, accountID, aws.StringValue(input.DBInstanceIdentifier), + StatusRebooting, StatusAvailable, StatusFailed) + if err != nil { + return nil, err + } + + // The engine is asked to stop cleanly first so the reboot is a restart of a + // checkpointed cluster rather than a crash it has to replay. + s.stopEngineOrRecordFallback(ctx, accountID, rec, "rebooting") + + if err := s.deps.Instances.RebootInstance(ctx, rec.InstanceID); err != nil { + return nil, s.failTransition(ctx, kv, accountID, rec, + fmt.Sprintf("the DB instance could not be rebooted: %v", err)) + } + + // The parameters are already in the engine's config on the data volume, so + // the restart is what applies them; the record only stops advertising them. + if len(rec.PendingRebootParameters) > 0 { + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, + "Applied the parameters that were pending a reboot.", EventCategoryConfigurationChange) + } + if err := s.updateInstance(ctx, kv, rec.DBInstanceIdentifier, func(stored *DBInstanceRecord) { + stored.PendingRebootParameters = nil + }); err != nil { + return nil, err + } + + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, + "DB instance restarted.", EventCategoryAvailability) + + // Returned as rebooting rather than available: the engine has to come back + // and report healthy before the reconciler calls it that. + return &rds.RebootDBInstanceOutput{DBInstance: s.projectDBInstance(rec)}, nil +} + +// Stops the engine and then the VM. The data volume, the customer ENI and its +// IP, and the DNS record are all retained, so a start comes back on the same +// datadir at the same address (D5/D9). +func (s *Service) StopDBInstance(ctx context.Context, input *rds.StopDBInstanceInput, accountID string) (*rds.StopDBInstanceOutput, error) { + if input == nil { + return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue) + } + // A pre-stop snapshot is rds-8's; accepting it here would report a snapshot + // the customer would then not find. + if aws.StringValue(input.DBSnapshotIdentifier) != "" { + return nil, unimplemented("DBSnapshotIdentifier", "taking a snapshot as part of a stop is not implemented yet") + } + + rec, kv, err := s.beginTransition(ctx, accountID, aws.StringValue(input.DBInstanceIdentifier), + StatusStopping, StatusAvailable, StatusFailed) + if err != nil { + return nil, err + } + + s.stopEngineOrRecordFallback(ctx, accountID, rec, "stopping") + + err = s.deps.Instances.StopInstance(ctx, rec.InstanceID) + if errors.Is(err, ErrInstanceNotOnNode) { + err = s.confirmVMNotRunning(ctx, accountID, rec.InstanceID) + } + if err != nil { + return nil, s.failTransition(ctx, kv, accountID, rec, + fmt.Sprintf("the DB instance could not be stopped: %v", err)) + } + + stopped, err := s.completeTransition(ctx, kv, rec.DBInstanceIdentifier, StatusStopped) + if err != nil { + return nil, err + } + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, + "DB instance stopped.", EventCategoryAvailability) + + return &rds.StopDBInstanceOutput{DBInstance: s.projectDBInstance(stopped)}, nil +} + +// Brings a stopped instance back on its retained data volume and customer ENI. +// The engine replays WAL when the graceful stop did not complete. +func (s *Service) StartDBInstance(ctx context.Context, input *rds.StartDBInstanceInput, accountID string) (*rds.StartDBInstanceOutput, error) { + if input == nil { + return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue) + } + + rec, kv, err := s.beginTransition(ctx, accountID, aws.StringValue(input.DBInstanceIdentifier), + StatusStarting, StatusStopped, StatusFailed) + if err != nil { + return nil, err + } + + if err := s.startInstanceVM(ctx, rec.InstanceID); err != nil { + return nil, s.failTransition(ctx, kv, accountID, rec, + fmt.Sprintf("the DB instance could not be started: %v", err)) + } + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, + "DB instance starting.", EventCategoryAvailability) + + // Returned as starting: the reconciler flips it to available on the first + // healthy heartbeat from the restarted agent. + return &rds.StartDBInstanceOutput{DBInstance: s.projectDBInstance(rec)}, nil +} + +// The owning node's in-memory command path first, since it is the one that can +// re-attach the volumes it already has mapped. A node restart drops that, and +// the stopped-instance path relaunches from the persisted record instead. +func (s *Service) startInstanceVM(ctx context.Context, instanceID string) error { + err := s.deps.Instances.StartInstance(ctx, instanceID) + if err == nil || !errors.Is(err, ErrInstanceNotOnNode) { + return err + } + slog.InfoContext(ctx, "rds: no node holds the DB VM; relaunching it from its stopped-instance record", + "instanceId", instanceID) + return s.deps.Instances.StartStoppedInstance(ctx, instanceID) +} + +// A power command no node answered usually means the VM is already down, which +// is where a stop was going anyway — but a partitioned or restarting node looks +// identical from here. The fleet-wide view has to agree before the record says +// stopped, or a VM still serving on the customer ENI is reported as stopped. +// +// A nil resolver disables the check, matching the reconciler's health path. +func (s *Service) confirmVMNotRunning(ctx context.Context, accountID, instanceID string) error { + if s.deps.InstanceState == nil { + return nil + } + state, err := s.deps.InstanceState.InstanceState(ctx, instanceID, accountID) + if err != nil { + return fmt.Errorf("no node is holding the DB VM and its state could not be resolved: %w", err) + } + if state == instanceStateRunning { + return errors.New("no node is holding the DB VM but it is still running") + } + return nil +} + +// Asks the engine to shut down cleanly, and records an event rather than +// failing the call when it cannot. An engine that did not stop cleanly replays +// its WAL on the next start; refusing to stop the VM over it would leave the +// customer unable to stop an instance whose agent is wedged. +func (s *Service) stopEngineOrRecordFallback(ctx context.Context, accountID string, rec *DBInstanceRecord, operation string) { + if err := s.stopEngine(ctx, accountID, rec.DBInstanceIdentifier); err != nil { + slog.WarnContext(ctx, "rds: graceful engine stop failed; continuing", + "dbInstance", rec.DBInstanceIdentifier, "operation", operation, "err", err) + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, + fmt.Sprintf("The database engine could not be shut down cleanly before %s; it will recover from its write-ahead log on the next start.", operation), + EventCategoryNotification, EventCategoryAvailability) + } +} + +// Moves the instance into a transitional state under CAS and returns the record +// as written. from lists the statuses the operation is legal from, so a stop of +// an already-stopping instance is rejected as a state error rather than racing +// the stop already running. +func (s *Service) beginTransition(ctx context.Context, accountID, id string, to Status, from ...Status) (*DBInstanceRecord, jetstream.KeyValue, error) { + if id == "" { + return nil, nil, fmt.Errorf("%s: DBInstanceIdentifier is required", awserrors.ErrorInvalidParameterValue) + } + if s.deps.Instances == nil { + return nil, nil, errors.New("rds: no instance command path configured") + } + kv, err := s.bucket(ctx, accountID) + if err != nil { + return nil, nil, err + } + rec, rev, err := s.getDBInstance(ctx, kv, id) + if err != nil { + return nil, nil, err + } + if !slices.Contains(from, rec.Status) { + return nil, nil, fmt.Errorf("%s: DB instance %s is %s; the operation requires it to be %s", + awserrors.ErrorDBInstanceInvalidState, id, rec.Status, joinStatuses(from)) + } + if rec.InstanceID == "" { + return nil, nil, fmt.Errorf("%s: DB instance %s has no VM to act on", awserrors.ErrorDBInstanceInvalidState, id) + } + + now := time.Now().UTC() + rec.Status = to + rec.FailureReason = "" + rec.TransitionStartedAt = &now + rec.UpdatedAt = now + if err := updateJSON(ctx, kv, DBInstanceKey(id), rev, rec); err != nil { + if errors.Is(err, jetstream.ErrKeyExists) { + return nil, nil, fmt.Errorf("%s: DB instance %s changed state concurrently; retry the operation", + awserrors.ErrorDBInstanceInvalidState, id) + } + return nil, nil, err + } + return rec, kv, nil +} + +// Lands the instance in its post-transition state and clears the transition +// stamp, returning the record as written. +func (s *Service) completeTransition(ctx context.Context, kv jetstream.KeyValue, id string, to Status) (*DBInstanceRecord, error) { + var stored *DBInstanceRecord + err := s.updateInstance(ctx, kv, id, func(rec *DBInstanceRecord) { + rec.Status = to + rec.TransitionStartedAt = nil + stored = rec + }) + if err != nil { + return nil, err + } + return stored, nil +} + +// Records why an operation failed and returns the error the caller surfaces, so +// a failed transition cannot be left looking like one still in progress. +func (s *Service) failTransition(ctx context.Context, kv jetstream.KeyValue, accountID string, rec *DBInstanceRecord, reason string) error { + if err := s.updateInstance(ctx, kv, rec.DBInstanceIdentifier, func(stored *DBInstanceRecord) { + stored.Status = StatusFailed + stored.FailureReason = reason + stored.TransitionStartedAt = nil + }); err != nil { + slog.ErrorContext(ctx, "rds: marking a failed lifecycle transition failed", + "dbInstance", rec.DBInstanceIdentifier, "err", err) + } + s.RecordEvent(ctx, accountID, EventSourceTypeDBInstance, rec.DBInstanceIdentifier, reason, EventCategoryFailure) + return fmt.Errorf("%s: %s", awserrors.ErrorDBInstanceInvalidState, reason) +} + +// A read-modify-write under CAS, replayed on contention so a concurrent agent +// heartbeat cannot make a lifecycle write disappear. +func (s *Service) updateInstance(ctx context.Context, kv jetstream.KeyValue, id string, mutate func(*DBInstanceRecord)) error { + key := DBInstanceKey(id) + for range tagWriteAttempts { + rec, rev, err := s.getDBInstance(ctx, kv, id) + if err != nil { + return err + } + mutate(rec) + rec.UpdatedAt = time.Now().UTC() + + err = updateJSON(ctx, kv, key, rev, rec) + if err == nil { + return nil + } + if !errors.Is(err, jetstream.ErrKeyExists) { + return err + } + } + return fmt.Errorf("rds: update of %s contended after %d attempts", key, tagWriteAttempts) +} + +func joinStatuses(statuses []Status) string { + out := "" + for i, status := range statuses { + switch { + case i == 0: + out = string(status) + case i == len(statuses)-1: + out += " or " + string(status) + default: + out += ", " + string(status) + } + } + return out +} + +// Only the node running the VM subscribes ec2.cmd.{instanceID}, so a power +// command routes itself; the fallback subject relaunches a VM no node holds. +type natsInstanceCommander struct { + nc *nats.Conn + timeout time.Duration +} + +var _ instanceCommander = (*natsInstanceCommander)(nil) + +func NewNATSInstanceCommander(nc *nats.Conn) instanceCommander { + return &natsInstanceCommander{nc: nc, timeout: instanceCommandTimeout} +} + +func (c *natsInstanceCommander) StopInstance(ctx context.Context, instanceID string) error { + return c.send(ctx, instanceID, types.EC2CommandAttributes{StopInstance: true}) +} + +func (c *natsInstanceCommander) StartInstance(ctx context.Context, instanceID string) error { + return c.send(ctx, instanceID, types.EC2CommandAttributes{StartInstance: true}) +} + +func (c *natsInstanceCommander) RebootInstance(ctx context.Context, instanceID string) error { + return c.send(ctx, instanceID, types.EC2CommandAttributes{RebootInstance: true}) +} + +// The VM runs in the system account, so every command is issued there — the +// ownership check on the far side compares against that same account. +func (c *natsInstanceCommander) send(ctx context.Context, instanceID string, attrs types.EC2CommandAttributes) error { + cmd := types.EC2InstanceCommand{ID: instanceID, Attributes: attrs} + _, err := utils.NATSRequest[struct{}](ctx, c.nc, "ec2.cmd."+instanceID, cmd, c.timeout, utils.GlobalAccountID) + if err == nil { + return nil + } + // No subscriber, or a subscriber that does not hold this VM: both mean the + // command has to go through the stopped-instance path instead. + if errors.Is(err, nats.ErrNoResponders) || + strings.Contains(err.Error(), awserrors.ErrorInvalidInstanceIDNotFound) { + return fmt.Errorf("%w: %s", ErrInstanceNotOnNode, instanceID) + } + return err +} + +func (c *natsInstanceCommander) StartStoppedInstance(ctx context.Context, instanceID string) error { + _, err := utils.NATSRequest[handlers_ec2_instance.StartStoppedInstanceOutput](ctx, c.nc, "ec2.start", + handlers_ec2_instance.StartStoppedInstanceInput{InstanceID: instanceID}, c.timeout, utils.GlobalAccountID) + return err +} diff --git a/spinifex/handlers/rds/lifecycle_test.go b/spinifex/handlers/rds/lifecycle_test.go new file mode 100644 index 000000000..80ceaa9b7 --- /dev/null +++ b/spinifex/handlers/rds/lifecycle_test.go @@ -0,0 +1,547 @@ +package handlers_rds + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/aws/aws-sdk-go/service/rds" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// fakeInstanceCommander records the power commands the lifecycle ops issue, and +// can refuse them the way a node that no longer holds the VM does. +type fakeInstanceCommander struct { + calls []string + // notOnNode makes StartInstance report the VM as held by no node, which is + // what sends the start down the stopped-instance path. + notOnNode bool + // stopNotOnNode is the same for a stop, which then has to confirm the VM is + // really down rather than assume it. + stopNotOnNode bool + err error +} + +var _ instanceCommander = (*fakeInstanceCommander)(nil) + +func (f *fakeInstanceCommander) StopInstance(_ context.Context, instanceID string) error { + f.calls = append(f.calls, "stop:"+instanceID) + if f.stopNotOnNode { + return ErrInstanceNotOnNode + } + return f.err +} + +func (f *fakeInstanceCommander) RebootInstance(_ context.Context, instanceID string) error { + f.calls = append(f.calls, "reboot:"+instanceID) + return f.err +} + +func (f *fakeInstanceCommander) StartInstance(_ context.Context, instanceID string) error { + f.calls = append(f.calls, "start:"+instanceID) + if f.notOnNode { + return ErrInstanceNotOnNode + } + return f.err +} + +func (f *fakeInstanceCommander) StartStoppedInstance(_ context.Context, instanceID string) error { + f.calls = append(f.calls, "start-stopped:"+instanceID) + return f.err +} + +// fakeSnapshots stands in for the EC2 snapshot service. holding is what +// DescribeSnapshots reports against the data volume, which is what decides +// between deleting the volume and retaining it. +type fakeSnapshots struct { + created []*ec2.CreateSnapshotInput + holding []string + createErr error + describeErr error +} + +var _ snapshotProvider = (*fakeSnapshots)(nil) + +func (f *fakeSnapshots) CreateSnapshot(_ context.Context, input *ec2.CreateSnapshotInput, _ string) (*ec2.Snapshot, error) { + if f.createErr != nil { + return nil, f.createErr + } + f.created = append(f.created, input) + id := "snap-0001" + // A snapshot pins the chunks of the volume it was taken from, so taking one + // is also what makes that volume undeletable. + f.holding = append(f.holding, id) + return &ec2.Snapshot{SnapshotId: aws.String(id), VolumeId: input.VolumeId}, nil +} + +func (f *fakeSnapshots) DescribeSnapshots(_ context.Context, _ *ec2.DescribeSnapshotsInput, _ string) (*ec2.DescribeSnapshotsOutput, error) { + if f.describeErr != nil { + return nil, f.describeErr + } + out := &ec2.DescribeSnapshotsOutput{} + for _, id := range f.holding { + out.Snapshots = append(out.Snapshots, &ec2.Snapshot{SnapshotId: aws.String(id)}) + } + return out, nil +} + +// stubAgent answers the command channel the way the in-guest agent does: it +// replies to whatever lands on the command subject, correlated by command ID. +type stubAgent struct { + mu sync.Mutex + issued []Command + fail bool + // What a successful reply carries back. The apply-params reply reports the + // settings pending a restart here, since CommandReply has no payload. + message string +} + +// Set before any command is issued; guarded because the reply is built on the +// subscription's goroutine. +func (a *stubAgent) replyWith(message string) { + a.mu.Lock() + defer a.mu.Unlock() + a.message = message +} + +func (a *stubAgent) received() []Command { + a.mu.Lock() + defer a.mu.Unlock() + return append([]Command(nil), a.issued...) +} + +func newStubAgent(t *testing.T, nc *nats.Conn, accountID, dbID string, fail bool) *stubAgent { + t.Helper() + agent := &stubAgent{fail: fail} + replySubject := BusCommandReplySubject(accountID, dbID) + + sub, err := nc.Subscribe(BusCommandSubject(accountID, dbID), func(msg *nats.Msg) { + var cmd Command + if err := json.Unmarshal(msg.Data, &cmd); err != nil { + t.Logf("stub agent: undecodable command: %v", err) + return + } + agent.mu.Lock() + agent.issued = append(agent.issued, cmd) + reply := CommandReply{CommandID: cmd.CommandID, Status: CommandStatusSucceeded, Message: agent.message} + if agent.fail { + reply = CommandReply{CommandID: cmd.CommandID, Status: CommandStatusFailed, Message: "the engine did not stop"} + } + agent.mu.Unlock() + payload, err := json.Marshal(reply) + if err != nil { + t.Logf("stub agent: marshal reply: %v", err) + return + } + if err := nc.Publish(replySubject, payload); err != nil { + t.Logf("stub agent: publish reply: %v", err) + } + }) + require.NoError(t, err) + t.Cleanup(func() { _ = sub.Unsubscribe() }) + return agent +} + +type lifecycleHarness struct { + svc *Service + nc *nats.Conn + cmdr *fakeInstanceCommander + snaps *fakeSnapshots + enis *fakeENIs + launcher *fakeLauncher + volumes *fakeVolumes + agent *stubAgent + vmState *fakeInstanceState +} + +// agentFails makes the stub agent reject every command, which is how the +// graceful-stop fallback is exercised without waiting out a real budget. +func newLifecycleHarness(t *testing.T, agentFails bool) *lifecycleHarness { + t.Helper() + _, nc, _ := testutil.StartTestJetStream(t) + + h := &lifecycleHarness{ + nc: nc, + cmdr: &fakeInstanceCommander{}, + snaps: &fakeSnapshots{}, + enis: &fakeENIs{}, + launcher: &fakeLauncher{}, + volumes: &fakeVolumes{}, + // The state an available instance's VM is in; a case that needs the VM + // down for a stop-confirmation path sets it. + vmState: &fakeInstanceState{state: instanceStateRunning}, + } + h.agent = newStubAgent(t, nc, testAccountID, testDBID, agentFails) + h.svc = NewService(nc, testRegion).WithDeps(Deps{ + LoadCA: newTestCA(t), + Instances: h.cmdr, + Snapshots: h.snaps, + InstanceState: h.vmState, + Launch: LaunchDeps{ + VPC: h.enis, + Instance: h.launcher, + Volume: h.volumes, + }, + }) + return h +} + +// availableRecord is an instance in the state every lifecycle op starts from. +func availableRecord() DBInstanceRecord { + rec := defaultRecord() + rec.Status = StatusAvailable + rec.ENIID = "eni-cust01" + rec.SystemENIID = "eni-sys01" + rec.DataVolumeID = "vol-rdsdata01" + rec.CreatedAt = time.Now().UTC().Add(-time.Hour) + return rec +} + +func (h *lifecycleHarness) record(t *testing.T) DBInstanceRecord { + t.Helper() + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var rec DBInstanceRecord + found, err := getJSON(t.Context(), kv, DBInstanceKey(testDBID), &rec) + require.NoError(t, err) + require.True(t, found) + return rec +} + +func (h *lifecycleHarness) events(t *testing.T) []Event { + t.Helper() + kv, err := h.svc.bucket(t.Context(), testAccountID) + require.NoError(t, err) + var ring eventRing + found, err := getJSON(t.Context(), kv, EventRingKey(EventSourceTypeDBInstance, testDBID), &ring) + require.NoError(t, err) + if !found { + return nil + } + return ring.Events +} + +// The engine is asked to stop cleanly first, so the reboot is a restart of a +// checkpointed cluster rather than a crash it has to replay. +func TestRebootDBInstance_StopsTheEngineThenTheVM(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + out, err := h.svc.RebootDBInstance(t.Context(), + &rds.RebootDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + issued := h.agent.received() + require.Len(t, issued, 1) + assert.Equal(t, CommandStopEngine, issued[0].Type) + assert.Equal(t, []string{"reboot:" + testInstance}, h.cmdr.calls) + + // Reported as rebooting, not available: the engine has to come back and say + // so before the reconciler calls it that. + assert.Equal(t, string(StatusRebooting), aws.StringValue(out.DBInstance.DBInstanceStatus)) + assert.Equal(t, StatusRebooting, h.record(t).Status) +} + +// D16: the static parameters are already in the engine's config, so the restart +// is what applies them and the record stops advertising them. +func TestRebootDBInstance_ClearsThePendingRebootParameters(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.PendingRebootParameters = []string{"shared_buffers"} + seedInstance(t, h.svc, rec) + + _, err := h.svc.RebootDBInstance(t.Context(), + &rds.RebootDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Empty(t, h.record(t).PendingRebootParameters) + assert.Contains(t, eventMessages(h.events(t)), "Applied the parameters that were pending a reboot.") +} + +// D19: there is no standby to fail over to, so silently ignoring the flag would +// report a failover that never happened. +func TestRebootDBInstance_RejectsForceFailover(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.RebootDBInstance(t.Context(), &rds.RebootDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + ForceFailover: aws.Bool(true), + }, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "ForceFailover") + assert.Equal(t, StatusAvailable, h.record(t).Status) +} + +// An engine that did not stop cleanly replays its WAL on the next start. +// Refusing to stop the VM over it would leave a customer unable to stop an +// instance whose agent is wedged, so it is recorded and the stop continues. +func TestStopDBInstance_RecordsAFailedGracefulStopAndContinues(t *testing.T) { + h := newLifecycleHarness(t, true) + seedInstance(t, h.svc, availableRecord()) + + out, err := h.svc.StopDBInstance(t.Context(), + &rds.StopDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Equal(t, []string{"stop:" + testInstance}, h.cmdr.calls) + assert.Equal(t, string(StatusStopped), aws.StringValue(out.DBInstance.DBInstanceStatus)) + assert.Equal(t, StatusStopped, h.record(t).Status) + + var recorded bool + for _, message := range eventMessages(h.events(t)) { + if strings.Contains(message, "could not be shut down cleanly") { + recorded = true + } + } + assert.True(t, recorded, "a degraded shutdown must reach the customer's event ring") +} + +// A node that is partitioned or restarting answers a power command exactly the +// way one that never held the VM does. Writing stopped on that alone would +// report a VM still serving on the customer ENI as stopped. +func TestStopDBInstance_FailsWhenNoNodeAnsweredButTheVMIsStillRunning(t *testing.T) { + h := newLifecycleHarness(t, false) + h.cmdr.stopNotOnNode = true + h.vmState.state = instanceStateRunning + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.StopDBInstance(t.Context(), + &rds.StopDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.Error(t, err) + + rec := h.record(t) + assert.Equal(t, StatusFailed, rec.Status) + assert.Contains(t, rec.FailureReason, "still running") +} + +// The same unanswered command is the normal shape of a VM that is genuinely +// down, and that stop has to converge rather than fail. +func TestStopDBInstance_CompletesWhenTheFleetConfirmsTheVMIsDown(t *testing.T) { + h := newLifecycleHarness(t, false) + h.cmdr.stopNotOnNode = true + h.vmState.state = "stopped" + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.StopDBInstance(t.Context(), + &rds.StopDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Equal(t, StatusStopped, h.record(t).Status) +} + +// The reconciler resumes a stop with no caller watching, so it needs the same +// confirmation before it calls an unanswered command a completed stop. +func TestReconciler_DoesNotCallAStillRunningVMStopped(t *testing.T) { + h := newLifecycleHarness(t, false) + h.cmdr.stopNotOnNode = true + h.vmState.state = instanceStateRunning + rec := availableRecord() + rec.Status = StatusStopping + started := time.Now().UTC() + rec.TransitionStartedAt = &started + seedInstance(t, h.svc, rec) + + require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + + // Left stopping so the next pass retries; the bound is what ends it. + assert.Equal(t, StatusStopping, h.record(t).Status) +} + +// D5/D9: the data volume, the customer ENI and the DNS record are all retained, +// so a start comes back on the same datadir at the same address. +func TestStopDBInstance_RetainsTheEndpointAndTheVolume(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.StopDBInstance(t.Context(), + &rds.StopDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Empty(t, h.volumes.deleted, "a stop must not delete the data volume") + assert.Empty(t, h.enis.deleted, "a stop must not delete the endpoint ENI") + + rec := h.record(t) + assert.Equal(t, "vol-rdsdata01", rec.DataVolumeID) + assert.Equal(t, "eni-cust01", rec.ENIID) +} + +// A pre-stop snapshot is a later phase's. Accepting it here would report a +// snapshot the customer would then not find. +func TestStopDBInstance_RejectsASnapshotIdentifier(t *testing.T) { + h := newLifecycleHarness(t, false) + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.StopDBInstance(t.Context(), &rds.StopDBInstanceInput{ + DBInstanceIdentifier: aws.String(testDBID), + DBSnapshotIdentifier: aws.String("orders-db-pre-stop"), + }, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "DBSnapshotIdentifier") + assert.Equal(t, StatusAvailable, h.record(t).Status) +} + +func TestStartDBInstance_StartsTheVMAndReportsStarting(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.Status = StatusStopped + seedInstance(t, h.svc, rec) + + out, err := h.svc.StartDBInstance(t.Context(), + &rds.StartDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Equal(t, []string{"start:" + testInstance}, h.cmdr.calls) + assert.Equal(t, string(StatusStarting), aws.StringValue(out.DBInstance.DBInstanceStatus)) + assert.Equal(t, StatusStarting, h.record(t).Status) +} + +// A node restart drops the VM from memory, so the start relaunches it from the +// persisted stopped-instance record rather than failing. +func TestStartDBInstance_FallsBackWhenNoNodeHoldsTheVM(t *testing.T) { + h := newLifecycleHarness(t, false) + h.cmdr.notOnNode = true + rec := availableRecord() + rec.Status = StatusStopped + seedInstance(t, h.svc, rec) + + _, err := h.svc.StartDBInstance(t.Context(), + &rds.StartDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.NoError(t, err) + + assert.Equal(t, []string{"start:" + testInstance, "start-stopped:" + testInstance}, h.cmdr.calls) +} + +// The transition is rejected rather than racing an operation already running. +func TestLifecycleOps_RejectAnIllegalStartingState(t *testing.T) { + cases := []struct { + name string + from Status + call func(*Service) error + }{ + {"StartAnAvailableInstance", StatusAvailable, func(s *Service) error { + _, err := s.StartDBInstance(context.Background(), + &rds.StartDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + return err + }}, + {"StopAStoppedInstance", StatusStopped, func(s *Service) error { + _, err := s.StopDBInstance(context.Background(), + &rds.StopDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + return err + }}, + {"RebootACreatingInstance", StatusCreating, func(s *Service) error { + _, err := s.RebootDBInstance(context.Background(), + &rds.RebootDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + return err + }}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + h := newLifecycleHarness(t, false) + rec := availableRecord() + rec.Status = tc.from + seedInstance(t, h.svc, rec) + + err := tc.call(h.svc) + require.Error(t, err) + assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceInvalidState) + assert.Equal(t, tc.from, h.record(t).Status, "a rejected op must not move the instance") + assert.Empty(t, h.cmdr.calls) + }) + } +} + +// A failed transition must not be left looking like one still in progress: the +// customer would wait on a reboot that is never coming. +func TestRebootDBInstance_MarksFailedWhenTheVMCommandFails(t *testing.T) { + h := newLifecycleHarness(t, false) + h.cmdr.err = errors.New("no node answered") + seedInstance(t, h.svc, availableRecord()) + + _, err := h.svc.RebootDBInstance(t.Context(), + &rds.RebootDBInstanceInput{DBInstanceIdentifier: aws.String(testDBID)}, testAccountID) + require.Error(t, err) + + rec := h.record(t) + assert.Equal(t, StatusFailed, rec.Status) + assert.Contains(t, rec.FailureReason, "could not be rebooted") + assert.Nil(t, rec.TransitionStartedAt) +} + +func eventMessages(events []Event) []string { + messages := make([]string, 0, len(events)) + for _, event := range events { + messages = append(messages, event.Message) + } + return messages +} + +// restartingRecord is an instance mid-reboot: the transition began at started, +// and the agent last reported at lastSeen. +func restartingRecord(status Status, started, lastSeen time.Time) DBInstanceRecord { + rec := availableRecord() + rec.Status = status + rec.TransitionStartedAt = &started + rec.Agent = AgentState{ + InstanceID: testInstance, + EngineHealth: EngineHealthHealthy, + LastSeen: &lastSeen, + } + return rec +} + +// The API call that begins a reboot or a start returns before the engine is +// back, so this is what actually lands the instance in available. +func TestReconciler_CompletesARestartOnAHeartbeatFromTheRestartedEngine(t *testing.T) { + for _, status := range []Status{StatusRebooting, StatusStarting} { + t.Run(string(status), func(t *testing.T) { + h := newLifecycleHarness(t, false) + now := time.Now().UTC() + seedInstance(t, h.svc, restartingRecord(status, now.Add(-time.Minute), now)) + + require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + + rec := h.record(t) + assert.Equal(t, StatusAvailable, rec.Status) + }) + } +} + +// The VM keeps its instance ID across a restart, so a beat sent before the +// transition began would otherwise report the reboot as finished the instant it +// started. +func TestReconciler_IgnoresAHeartbeatPredatingTheRestart(t *testing.T) { + h := newLifecycleHarness(t, false) + now := time.Now().UTC() + seedInstance(t, h.svc, restartingRecord(StatusRebooting, now, now.Add(-time.Minute))) + + require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + + assert.Equal(t, StatusRebooting, h.record(t).Status) +} + +// A restart that never comes back has to end somewhere: the customer sees a +// broken instance either way, and failed is the state they can act on. +func TestReconciler_MarksFailedWhenARestartOverrunsItsBound(t *testing.T) { + h := newLifecycleHarness(t, false) + now := time.Now().UTC() + started := now.Add(-2 * transitionTimeout) + seedInstance(t, h.svc, restartingRecord(StatusStarting, started, started.Add(-time.Minute))) + + require.NoError(t, NewReconciler(h.svc, "node-a").reconcileOnce(t.Context())) + + rec := h.record(t) + assert.Equal(t, StatusFailed, rec.Status) + assert.Contains(t, rec.FailureReason, "did not report healthy") +} diff --git a/spinifex/handlers/rds/reconciler.go b/spinifex/handlers/rds/reconciler.go index ebb1e46ae..a9ca205cb 100644 --- a/spinifex/handlers/rds/reconciler.go +++ b/spinifex/handlers/rds/reconciler.go @@ -27,6 +27,11 @@ const ( // false failed is worse than a slow one, since the customer sees a broken // create either way but only the false one is wrong. defaultBootstrapTimeout = 20 * time.Minute + + // How long a reboot, start, stop or delete may stay in flight before the + // reconciler calls it failed. It covers a cold boot plus a WAL replay, which + // is the longest a restart can honestly take. + transitionTimeout = 10 * time.Minute ) // The EC2 lifecycle states a DB VM may be in and still be on its way up. The @@ -194,42 +199,130 @@ func (r *Reconciler) reconcileAccount(ctx context.Context, kv jetstream.KeyValue return errors.Join(failures...) } -// Only creating instances are acted on in this phase. Everything else is a -// later phase's transition, and touching it here would race the owner. +// The reconciler owns every transitional state that no single API call can +// finish: the one it drives itself (creating), and the ones whose caller may +// have died partway through. A settled instance is left alone. func (r *Reconciler) reconcileInstance(ctx context.Context, kv jetstream.KeyValue, accountID, id string) error { var rec DBInstanceRecord rev, found, err := getJSONRevision(ctx, kv, DBInstanceKey(id), &rec) if err != nil || !found { return err } - if rec.Status != StatusCreating { + switch rec.Status { + case StatusCreating: + return r.reconcileCreating(ctx, kv, rev, accountID, &rec) + case StatusRebooting, StatusStarting: + return r.reconcileRestarting(ctx, kv, rev, accountID, &rec) + case StatusStopping: + return r.reconcileStopping(ctx, kv, rev, accountID, &rec) + case StatusDeleting: + return r.reconcileDeleting(ctx, kv, rev, accountID, &rec) + default: return nil } +} - healthy, err := r.bootstrapComplete(ctx, accountID, &rec) +func (r *Reconciler) reconcileCreating(ctx context.Context, kv jetstream.KeyValue, rev uint64, accountID string, rec *DBInstanceRecord) error { + // No lower bound on the heartbeat: the VM is new, so any beat naming it is + // necessarily this instance's. + healthy, err := r.engineReady(ctx, accountID, rec, time.Time{}) if err != nil { return err } if healthy { - return r.transition(ctx, kv, rev, &rec, StatusAvailable, "") + return r.transition(ctx, kv, rev, rec, StatusAvailable, "") } timeout := r.svc.bootstrapTimeout() if time.Since(rec.CreatedAt) > timeout { - return r.transition(ctx, kv, rev, &rec, StatusFailed, + return r.transition(ctx, kv, rev, rec, StatusFailed, fmt.Sprintf("the database engine did not report healthy within %s of creation", timeout)) } return nil } +// Reboot and start both end the same way: the engine comes back and says so. +// The API call that began them returns before that happens, so this is what +// actually lands the instance in available. +func (r *Reconciler) reconcileRestarting(ctx context.Context, kv jetstream.KeyValue, rev uint64, accountID string, rec *DBInstanceRecord) error { + // The VM keeps its instance ID across a restart, so only a beat sent after + // the transition began proves the engine came back rather than that it was + // up before it went down. + started := transitionStarted(rec) + healthy, err := r.engineReady(ctx, accountID, rec, started) + if err != nil { + return err + } + if healthy { + return r.transition(ctx, kv, rev, rec, StatusAvailable, "") + } + if time.Since(started) > transitionTimeout { + return r.transition(ctx, kv, rev, rec, StatusFailed, + fmt.Sprintf("the database engine did not report healthy within %s of %s", transitionTimeout, rec.Status)) + } + return nil +} + +// A stop whose caller died leaves the VM possibly still running, so the stop is +// re-issued rather than assumed: it is idempotent, and a VM no node holds is +// confirmed down against the fleet before the record calls it stopped. +func (r *Reconciler) reconcileStopping(ctx context.Context, kv jetstream.KeyValue, rev uint64, accountID string, rec *DBInstanceRecord) error { + if r.svc.deps.Instances == nil { + return errors.New("rds reconciler: no instance command path configured") + } + err := r.svc.deps.Instances.StopInstance(ctx, rec.InstanceID) + if errors.Is(err, ErrInstanceNotOnNode) { + err = r.svc.confirmVMNotRunning(ctx, accountID, rec.InstanceID) + } + if err == nil { + return r.transition(ctx, kv, rev, rec, StatusStopped, "") + } + if time.Since(transitionStarted(rec)) > transitionTimeout { + return r.transition(ctx, kv, rev, rec, StatusFailed, + fmt.Sprintf("the DB instance could not be stopped within %s: %v", transitionTimeout, err)) + } + slog.WarnContext(ctx, "rds reconciler: resuming a stop failed; retrying next pass", + "dbInstance", rec.DBInstanceIdentifier, "err", err) + return nil +} + +// Re-runs the teardown from wherever it stopped. Every step tolerates a missing +// resource, so replaying it converges rather than failing on what it already +// did; only a teardown still stuck past the bound is called failed, which is +// what lets the customer retry the delete. +func (r *Reconciler) reconcileDeleting(ctx context.Context, kv jetstream.KeyValue, rev uint64, accountID string, rec *DBInstanceRecord) error { + err := r.svc.teardownDBInstance(ctx, kv, accountID, rec) + if err == nil { + return nil + } + if time.Since(transitionStarted(rec)) > transitionTimeout { + return r.transition(ctx, kv, rev, rec, StatusFailed, + fmt.Sprintf("the DB instance could not be deleted within %s: %v", transitionTimeout, err)) + } + slog.WarnContext(ctx, "rds reconciler: resuming a delete failed; retrying next pass", + "dbInstance", rec.DBInstanceIdentifier, "err", err) + return nil +} + +// When the transition began. A record written by an older control plane carries +// no stamp, so its last write stands in — it is never earlier than the +// transition, so the bound cannot be under-counted. +func transitionStarted(rec *DBInstanceRecord) time.Time { + if rec.TransitionStartedAt != nil { + return *rec.TransitionStartedAt + } + return rec.UpdatedAt +} + // Both halves must hold: a healthy heartbeat from the record's *current* VM, // and that VM actually running. A stale beat from a superseded VM would -// otherwise report a replaced instance as ready. -func (r *Reconciler) bootstrapComplete(ctx context.Context, accountID string, rec *DBInstanceRecord) (bool, error) { +// otherwise report a replaced instance as ready. Beats at or before since are +// ignored, which is how a restart is told from the engine it restarted. +func (r *Reconciler) engineReady(ctx context.Context, accountID string, rec *DBInstanceRecord, since time.Time) (bool, error) { if rec.Agent.EngineHealth != EngineHealthHealthy || rec.InstanceID == "" || rec.Agent.InstanceID != rec.InstanceID { return false, nil } - if !r.heartbeatFresh(accountID, rec) { + if !r.heartbeatFresh(accountID, rec, since) { return false, nil } if r.svc.deps.InstanceState == nil { @@ -245,7 +338,7 @@ func (r *Reconciler) bootstrapComplete(ctx context.Context, accountID string, re // The in-memory beat is fresher than the persisted one but only this node sees // it, so a leader that has seen no beat falls back to the record — which trails // the truth by at most the persist floor. -func (r *Reconciler) heartbeatFresh(accountID string, rec *DBInstanceRecord) bool { +func (r *Reconciler) heartbeatFresh(accountID string, rec *DBInstanceRecord, since time.Time) bool { lastSeen, ok := r.svc.LastSeen(accountID, rec.DBInstanceIdentifier) if !ok { if rec.Agent.LastSeen == nil { @@ -253,6 +346,9 @@ func (r *Reconciler) heartbeatFresh(accountID string, rec *DBInstanceRecord) boo } lastSeen = *rec.Agent.LastSeen } + if !since.IsZero() && !lastSeen.After(since) { + return false + } return time.Since(lastSeen) <= HeartbeatStaleAfter } diff --git a/spinifex/handlers/rds/reconciler_test.go b/spinifex/handlers/rds/reconciler_test.go index 119ae2443..63dd37b22 100644 --- a/spinifex/handlers/rds/reconciler_test.go +++ b/spinifex/handlers/rds/reconciler_test.go @@ -163,10 +163,10 @@ func TestReconciler_LeavesASlowBootstrapAloneInsideTheWindow(t *testing.T) { assert.Equal(t, StatusCreating, status) } -// Every other status belongs to a later phase's transition, and touching one -// here would race its owner. -func TestReconciler_OnlyActsOnCreatingInstances(t *testing.T) { - for _, status := range []Status{StatusAvailable, StatusModifying, StatusDeleting, StatusFailed} { +// A settled instance, and a transition owned by a later phase, are both left +// alone: touching one would race its owner. +func TestReconciler_LeavesSettledInstancesAlone(t *testing.T) { + for _, status := range []Status{StatusAvailable, StatusStopped, StatusModifying, StatusBackingUp, StatusFailed} { t.Run(string(status), func(t *testing.T) { h := newReconcileHarness(t) rec := healthyRecord() @@ -178,7 +178,7 @@ func TestReconciler_OnlyActsOnCreatingInstances(t *testing.T) { got, _ := h.statusOf(t, testDBID) assert.Equal(t, status, got) - assert.Empty(t, h.state.calls, "a non-creating instance is not even inspected") + assert.Empty(t, h.state.calls, "a settled instance is not even inspected") }) } } diff --git a/spinifex/handlers/rds/records.go b/spinifex/handlers/rds/records.go index dbff7d5ae..4dd6d88d5 100644 --- a/spinifex/handlers/rds/records.go +++ b/spinifex/handlers/rds/records.go @@ -18,6 +18,14 @@ type DBInstanceRecord struct { MasterUsername string `json:"masterUsername"` Port int64 `json:"port"` + // A one-way marker (D8). A rotated password is handed to the agent over the + // command channel and never persisted, so this records only that it changed. + MasterPasswordUpdatedAt *time.Time `json:"masterPasswordUpdatedAt,omitempty"` + + // Blocks DeleteDBInstance outright (D18). Settable at create; rds-5b adds it + // to ModifyDBInstance. + DeletionProtection bool `json:"deletionProtection,omitempty"` + // Where the customer ENI was placed, so a replace lands the new VM's ENI in // the same subnet and security groups without re-deriving them. SubnetID string `json:"subnetId,omitempty"` @@ -52,6 +60,20 @@ type DBInstanceRecord struct { // stale reason cannot outlive the failure it describes. FailureReason string `json:"failureReason,omitempty"` + // When the lifecycle op that put the instance in its current transitional + // state began. The reconciler bounds the transition from here and ignores + // heartbeats older than it, so a beat from before a reboot cannot be read as + // the reboot having finished. + TransitionStartedAt *time.Time `json:"transitionStartedAt,omitempty"` + + // Named on the delete request and persisted before teardown starts, so a + // resumed delete takes the same final snapshot rather than none. + FinalSnapshotIdentifier string `json:"finalSnapshotIdentifier,omitempty"` + + // Static parameters written to the engine's config but not yet in effect. + // Cleared by the reboot that applies them (D16). + PendingRebootParameters []string `json:"pendingRebootParameters,omitempty"` + // Inline rather than a separate key space, so the record delete that ends the // instance also ends its tags. Tags map[string]string `json:"tags,omitempty"` @@ -83,6 +105,66 @@ type BootstrapState struct { ResolvedParameters []Parameter `json:"resolvedParameters,omitempty"` } +// Snapshot types and statuses, matching AWS. A final snapshot is manual: the +// customer named it and only the customer removes it. +const ( + SnapshotTypeManual = "manual" + SnapshotTypeAutomated = "automated" + + SnapshotStatusAvailable = "available" +) + +// The db-snapshots/{id} record. The EC2 snapshot holds the data; this is the +// RDS-level metadata a restore needs and DescribeDBSnapshots projects, captured +// at snapshot time because the DB instance it describes may be gone by then. +type DBSnapshotRecord struct { + DBSnapshotIdentifier string `json:"dbSnapshotIdentifier"` + DBInstanceIdentifier string `json:"dbInstanceIdentifier"` + AccountID string `json:"accountId"` + SnapshotType string `json:"snapshotType"` + Status string `json:"status"` + + // The EC2 snapshot the data lives in, and the volume whose chunks it + // references — which is why that volume is retained rather than deleted. + SnapshotID string `json:"snapshotId"` + SourceVolumeID string `json:"sourceVolumeId"` + + Engine string `json:"engine"` + EngineVersion string `json:"engineVersion"` + AllocatedStorage int64 `json:"allocatedStorage"` + StorageType string `json:"storageType,omitempty"` + StorageEncrypted bool `json:"storageEncrypted,omitempty"` + MasterUsername string `json:"masterUsername"` + Port int64 `json:"port"` + VpcID string `json:"vpcId,omitempty"` + + // True when the engine was still writing as it was taken, so a restore + // replays WAL. A final snapshot is taken with the engine already down, so it + // is never crash-consistent; rds-8's quiesce fallback is what sets this. + CrashConsistent bool `json:"crashConsistent,omitempty"` + + Tags map[string]string `json:"tags,omitempty"` + + CreatedAt time.Time `json:"createdAt"` +} + +// A data volume that outlived its DB instance because a COW snapshot still +// references its chunks (D10). The last DeleteDBSnapshot to empty Snapshots +// deletes it; rds-9's reaper is the backstop for a crash in between. +type RetainedVolumeRecord struct { + VolumeID string `json:"volumeId"` + AccountID string `json:"accountId"` + // The instance it belonged to, so an operator can attribute the footprint + // after the DB instance record is gone. + DBInstanceIdentifier string `json:"dbInstanceIdentifier"` + // The DB snapshot identifiers holding it alive. + Snapshots []string `json:"snapshots"` + // Set when the volume store refused the delete without naming a holder, so a + // release must re-check rather than read the empty list as "nothing holds it". + HoldersUnresolved bool `json:"holdersUnresolved,omitempty"` + RetainedAt time.Time `json:"retainedAt"` +} + // A member list rather than a map because the XML marshaller renders a map as an // AWS-foreign shape in nondeterministic order. type Parameter struct { diff --git a/spinifex/handlers/rds/service.go b/spinifex/handlers/rds/service.go index 6eb419351..3c47b4b00 100644 --- a/spinifex/handlers/rds/service.go +++ b/spinifex/handlers/rds/service.go @@ -32,6 +32,11 @@ type Deps struct { // Lets the reconciler confirm a DB VM is running before calling its // instance available. Nil leaves the transition on the heartbeat alone. InstanceState InstanceStateResolver + // Drives the DB VM's power state for reboot, stop and start. + Instances instanceCommander + // Takes the final snapshot at delete, and reports which snapshots still + // reference a data volume before it can be deleted. + Snapshots snapshotProvider // The northstar base zone. Empty means no vanity hostname, and the endpoint // is the customer-ENI IP instead (D6). BaseDomain string diff --git a/spinifex/handlers/rds/service_nats.go b/spinifex/handlers/rds/service_nats.go index 54e0cd7fb..c3a31e890 100644 --- a/spinifex/handlers/rds/service_nats.go +++ b/spinifex/handlers/rds/service_nats.go @@ -15,6 +15,14 @@ const defaultTimeout = 30 * time.Second // than the agent protocol's round-trip budget. const createTimeout = 5 * time.Minute +// A lifecycle op waits on a graceful engine stop and then on the VM, both of +// which are bounded but neither of which fits the default budget. A delete adds +// the final snapshot and the teardown on top. +const ( + lifecycleTimeout = 4 * time.Minute + deleteTimeout = 6 * time.Minute +) + // The gateway-side adapter that forwards each agent action as a NATS request to // the daemon's matching subscriber. type NATSService struct { @@ -45,6 +53,31 @@ func (s *NATSService) DescribeDBInstances(ctx context.Context, input *rds.Descri SubjectDescribeDBInstances, input, defaultTimeout, accountID) } +func (s *NATSService) RebootDBInstance(ctx context.Context, input *rds.RebootDBInstanceInput, accountID string) (*rds.RebootDBInstanceOutput, error) { + return utils.NATSRequest[rds.RebootDBInstanceOutput](ctx, s.nc, + SubjectRebootDBInstance, input, lifecycleTimeout, accountID) +} + +func (s *NATSService) StartDBInstance(ctx context.Context, input *rds.StartDBInstanceInput, accountID string) (*rds.StartDBInstanceOutput, error) { + return utils.NATSRequest[rds.StartDBInstanceOutput](ctx, s.nc, + SubjectStartDBInstance, input, lifecycleTimeout, accountID) +} + +func (s *NATSService) StopDBInstance(ctx context.Context, input *rds.StopDBInstanceInput, accountID string) (*rds.StopDBInstanceOutput, error) { + return utils.NATSRequest[rds.StopDBInstanceOutput](ctx, s.nc, + SubjectStopDBInstance, input, lifecycleTimeout, accountID) +} + +func (s *NATSService) DeleteDBInstance(ctx context.Context, input *rds.DeleteDBInstanceInput, accountID string) (*rds.DeleteDBInstanceOutput, error) { + return utils.NATSRequest[rds.DeleteDBInstanceOutput](ctx, s.nc, + SubjectDeleteDBInstance, input, deleteTimeout, accountID) +} + +func (s *NATSService) DescribeEvents(ctx context.Context, input *rds.DescribeEventsInput, accountID string) (*rds.DescribeEventsOutput, error) { + return utils.NATSRequest[rds.DescribeEventsOutput](ctx, s.nc, + SubjectDescribeEvents, input, defaultTimeout, accountID) +} + func (s *NATSService) AddTagsToResource(ctx context.Context, input *rds.AddTagsToResourceInput, accountID string) (*rds.AddTagsToResourceOutput, error) { return utils.NATSRequest[rds.AddTagsToResourceOutput](ctx, s.nc, SubjectAddTagsToResource, input, defaultTimeout, accountID) diff --git a/spinifex/handlers/rds/store.go b/spinifex/handlers/rds/store.go index 30c5db489..090df537a 100644 --- a/spinifex/handlers/rds/store.go +++ b/spinifex/handlers/rds/store.go @@ -46,6 +46,7 @@ const ( // db-parameter-groups/{name}/params/{key} // backups/{dbInstanceIdentifier}/automated/{ts} // retained-volumes/{volumeID} +// events/{sourceType}/{sourceIdentifier} // // Tags live inline on each resource's own record rather than in a separate key // space, so there is no tags/ prefix here. @@ -116,6 +117,20 @@ func RetainedVolumeKey(volumeID string) string { return RetainedVolumesPrefix() + volumeID } +// One bounded ring per resource rather than a key per event: DescribeEvents +// always reads a whole resource's history, and a key per event would make the +// 14-day trim a listing plus a delete per expired entry. +// +// The ring is deliberately outside the resource's own record, so a deleted DB +// instance's events survive it for the rest of the retention window. +func EventsPrefix() string { + return "events/" +} + +func EventRingKey(sourceType, sourceIdentifier string) string { + return fmt.Sprintf("%s%s/%s", EventsPrefix(), sourceType, sourceIdentifier) +} + // Entries are rewritten on every VM replace (each mints a new instance ID) and // removed at teardown. func InstanceIndexPrefix() string { diff --git a/spinifex/handlers/rds/subjects.go b/spinifex/handlers/rds/subjects.go index e52df65af..6617fa7ac 100644 --- a/spinifex/handlers/rds/subjects.go +++ b/spinifex/handlers/rds/subjects.go @@ -22,6 +22,15 @@ const ( SubjectCreateDBInstance = "rds.CreateDBInstance" SubjectDescribeDBInstances = "rds.DescribeDBInstances" + // Lifecycle actions. Each drives the VM and the in-guest engine inline, so + // the daemon answering one is doing the work rather than queueing it. + SubjectRebootDBInstance = "rds.RebootDBInstance" + SubjectStartDBInstance = "rds.StartDBInstance" + SubjectStopDBInstance = "rds.StopDBInstance" + SubjectDeleteDBInstance = "rds.DeleteDBInstance" + + SubjectDescribeEvents = "rds.DescribeEvents" + // Tag actions address the resource by ARN, so one subject serves every // resource type rather than one per kind. SubjectAddTagsToResource = "rds.AddTagsToResource" diff --git a/spinifex/handlers/rds/validate.go b/spinifex/handlers/rds/validate.go index 8f36d09ee..872d28f0d 100644 --- a/spinifex/handlers/rds/validate.go +++ b/spinifex/handlers/rds/validate.go @@ -48,6 +48,7 @@ type validatedCreate struct { // this with a real DB subnet group lookup. SubnetID string DBParameterGroupName string + DeletionProtection bool Tags map[string]string } @@ -150,6 +151,7 @@ func validateCreateRequest(input *rds.CreateDBInstanceInput) (*validatedCreate, DBName: aws.StringValue(input.DBName), SecurityGroupIDs: aws.StringValueSlice(input.VpcSecurityGroupIds), DBParameterGroupName: engine.DefaultParameterGroupName(), + DeletionProtection: aws.BoolValue(input.DeletionProtection), Tags: tags, }, nil } @@ -197,9 +199,6 @@ func rejectUnimplemented(input *rds.CreateDBInstanceInput) error { if input.StorageEncrypted != nil && !aws.BoolValue(input.StorageEncrypted) { return unimplemented("StorageEncrypted=false", "unencrypted storage is not offered") } - if aws.BoolValue(input.DeletionProtection) { - return unimplemented("DeletionProtection", "DeleteDBInstance does not honour it yet") - } if aws.Int64Value(input.BackupRetentionPeriod) > 0 { return unimplemented("BackupRetentionPeriod", "automated backups are not implemented yet") }