feat: RDS lifecycle core — reboot, start, stop, delete, events (#596)

This commit is contained in:
Julian Sommer
2026-07-29 14:01:02 +10:00
committed by GitHub
parent 6eb76b439b
commit fc6e055ca0
33 changed files with 3285 additions and 31 deletions
+3 -1
View File
@@ -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
}
+1
View File
@@ -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.
+34 -4
View File
@@ -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
+2 -2
View File
@@ -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{})
+43
View File
@@ -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 != "" {
+249
View File
@@ -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]")
}
+268
View File
@@ -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)
}
}
+4
View File
@@ -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."},
+5
View File
@@ -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"},
+2
View File
@@ -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
+5 -5
View File
@@ -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
+7
View File
@@ -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,
+22
View File
@@ -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)
}
+181
View File
@@ -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
}
+77
View File
@@ -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)
}
+1
View File
@@ -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,
+11 -1
View File
@@ -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
+395
View File
@@ -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)
}
}
+312
View File
@@ -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)
}
+1
View File
@@ -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.
+314
View File
@@ -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
}
+178
View File
@@ -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
}
+7
View File
@@ -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
}
+359
View File
@@ -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
}
+547
View File
@@ -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")
}
+106 -10
View File
@@ -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
}
+5 -5
View File
@@ -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")
})
}
}
+82
View File
@@ -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 <entry> shape in nondeterministic order.
type Parameter struct {
+5
View File
@@ -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
+33
View File
@@ -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)
+15
View File
@@ -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 {
+9
View File
@@ -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"
+2 -3
View File
@@ -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")
}