mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-19 19:01:42 +00:00
feat: build the guest agent's engine from the image it is baked into
Define the composite engine interface and a newEngine() factory keyed by a per-engine layout table, so the agent's defaults, health probe and command implementation all come from the engine the image bakes rather than from PostgreSQL-shaped constants. Rename PGBin/PGData/PGUser to EngineBinDir/EngineDataDir/EngineUser, delete PGIsReady, and give engineProbe an injected state function with the pg_isready exit-code mapping moved into the PostgreSQL implementation. setup.sh now stamps /etc/spinifex-rds/engine and buildAgentUserData emits RDS_ENGINE. The agent compares the two after registering and refuses to run when they disagree, reporting unhealthy before the bootstrap handoff is written so rds-init never unblocks and no datadir is touched.
This commit is contained in:
+40
-8
@@ -180,7 +180,7 @@ type Agent struct {
|
||||
id identity
|
||||
cp controlPlane
|
||||
probe *engineProbe
|
||||
engine *postgresEngine
|
||||
engine engine
|
||||
handoffWriter func(string, *handlers_rds.GetDBBootstrapConfigOutput) error
|
||||
dataMountWaiter func(context.Context, string, string) error
|
||||
|
||||
@@ -195,10 +195,7 @@ type Agent struct {
|
||||
|
||||
// Assembles from already-built parts, so tests can pass fakes; New builds the
|
||||
// production control plane and delegates here.
|
||||
func newAgent(cfg config, cp controlPlane, probe *engineProbe) *Agent {
|
||||
if cfg.DataMount == "" {
|
||||
cfg.DataMount = defaultDataMount
|
||||
}
|
||||
func newAgent(cfg config, cp controlPlane, probe *engineProbe) (*Agent, error) {
|
||||
if cfg.MountsFile == "" {
|
||||
cfg.MountsFile = defaultMountsFile
|
||||
}
|
||||
@@ -211,11 +208,15 @@ func newAgent(cfg config, cp controlPlane, probe *engineProbe) *Agent {
|
||||
cfg: cfg, id: id, cp: cp, probe: probe,
|
||||
handoffWriter: writeHandoff, dataMountWaiter: waitForDataMount,
|
||||
}
|
||||
a.engine = newPostgresEngine(cfg, execCommandRunner, execSessionRunner, probe)
|
||||
eng, err := newEngine(cfg, execCommandRunner, execSessionRunner, probe)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a.engine = eng
|
||||
a.hb = newHeartbeater(cp, probe, a.engine, handlers_rds.HeartbeatInterval)
|
||||
a.cmd = newCommander(cp, newCommandRegistry(a.engine, newGuestStorage(cfg, execCommandRunner)), cfg.PollWait)
|
||||
a.guard = newParamGuard(a.engine, probe, cp)
|
||||
return a
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// Does not wait for IMDS: the register loop rides out a datapath still coming up.
|
||||
@@ -227,7 +228,11 @@ func New(cfg config) (*Agent, error) {
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return newAgent(cfg, cp, newEngineProbe(cfg, execProbeRunner)), nil
|
||||
probe, err := newProbe(cfg, execProbeRunner)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return newAgent(cfg, cp, probe)
|
||||
}
|
||||
|
||||
func (a *Agent) Run(ctx context.Context) error {
|
||||
@@ -245,6 +250,13 @@ func (a *Agent) Run(ctx context.Context) error {
|
||||
}
|
||||
a.hb.id, a.cmd.id, a.guard.id = a.id, a.id, a.id
|
||||
|
||||
// Ahead of the handoff, so rds-init never unblocks and no datadir is touched.
|
||||
// An agent running on the wrong image would initialise one engine over
|
||||
// another's data, coming up healthy and empty beside data nothing references.
|
||||
if err := a.refuseEngineMismatch(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Beating before the bootstrap keeps a stuck boot visible as a live VM with
|
||||
// a down engine rather than as silence.
|
||||
go a.hb.Run(ctx)
|
||||
@@ -279,6 +291,26 @@ func (a *Agent) Run(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// The engine cloud-init says this VM was launched as, against the engine the
|
||||
// image bakes. An empty expectation is no assertion rather than a disagreement:
|
||||
// a VM launched before the control plane emitted RDS_ENGINE carries none.
|
||||
func (a *Agent) refuseEngineMismatch(ctx context.Context) error {
|
||||
expected := strings.ToLower(strings.TrimSpace(a.cfg.Engine))
|
||||
if expected == "" || expected == a.cfg.BakedEngine {
|
||||
return nil
|
||||
}
|
||||
|
||||
message := fmt.Sprintf("this VM was launched as engine %q but its image bakes %q; refusing to bootstrap",
|
||||
expected, a.cfg.BakedEngine)
|
||||
slog.ErrorContext(ctx, "rds-agent: "+message)
|
||||
// Reported before returning, so the control plane learns why the instance
|
||||
// never came up rather than only that it never did.
|
||||
if _, err := a.cp.SubmitState(ctx, a.id, handlers_rds.EngineHealthUnhealthy, message); err != nil {
|
||||
slog.ErrorContext(ctx, "rds-agent: reporting the engine mismatch failed", "err", err)
|
||||
}
|
||||
return errors.New(message)
|
||||
}
|
||||
|
||||
const dataMountPollInterval = time.Second
|
||||
|
||||
// waitForDataMount blocks the stateful agent loops until the kernel reports the
|
||||
|
||||
+84
-27
@@ -168,18 +168,26 @@ func (f *fakeControlPlane) snapshotReplies() [][]handlers_rds.CommandReply {
|
||||
return out
|
||||
}
|
||||
|
||||
// testConfig points the agent's file output at a temp dir and its probe at a
|
||||
// name no real binary answers to.
|
||||
// testConfig stamps the engine a real rds-postgres image bakes, so the agent
|
||||
// resolves the same guest layout it would in a guest, and points its file output
|
||||
// at a temp dir.
|
||||
func testConfig(t *testing.T) config {
|
||||
t.Helper()
|
||||
return config{
|
||||
GatewayURL: "https://gw.test",
|
||||
HandoffDir: filepath.Join(t.TempDir(), "spinifex-rds"),
|
||||
EngineHost: defaultEngineHost,
|
||||
EnginePort: defaultEnginePort,
|
||||
PGIsReady: "pg_isready",
|
||||
PollWait: time.Second,
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
cfg.GatewayURL = "https://gw.test"
|
||||
cfg.HandoffDir = filepath.Join(t.TempDir(), "spinifex-rds")
|
||||
cfg.PollWait = time.Second
|
||||
return cfg
|
||||
}
|
||||
|
||||
// newTestAgent builds the agent over a probe the test drives directly.
|
||||
func newTestAgent(t *testing.T, cfg config, cp controlPlane, run probeRunner) *Agent {
|
||||
t.Helper()
|
||||
a, err := newAgent(cfg, cp, newPostgresProbe(cfg, run))
|
||||
if err != nil {
|
||||
t.Fatalf("newAgent: %v", err)
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// staticProbe is a probeRunner that always reports the same exit code.
|
||||
@@ -272,7 +280,7 @@ func TestRun_RegistersThenDeliversBootstrap(t *testing.T) {
|
||||
cp.bootstrapOut = bootstrapOutput(&password)
|
||||
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
if err := runAgent(t, a); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
@@ -303,7 +311,7 @@ func TestRun_AttachModeNeedsNoPassword(t *testing.T) {
|
||||
cp.bootstrapOut = bootstrapOutput(nil)
|
||||
|
||||
cfg := testConfig(t)
|
||||
if err := runAgent(t, newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))); err != nil {
|
||||
if err := runAgent(t, newTestAgent(t, cfg, cp, staticProbe(0))); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
|
||||
@@ -325,7 +333,7 @@ func TestRun_RetriesRegisterAndBootstrap(t *testing.T) {
|
||||
cp.bootstrapErrs = []error{errors.New("ServerInternal")}
|
||||
|
||||
cfg := testConfig(t)
|
||||
if err := runAgent(t, newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))); err != nil {
|
||||
if err := runAgent(t, newTestAgent(t, cfg, cp, staticProbe(0))); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
|
||||
@@ -344,7 +352,7 @@ func TestBootstrap_RetriesHandoffWithoutRefetching(t *testing.T) {
|
||||
password := "staged-secret"
|
||||
cp.bootstrapOut = bootstrapOutput(&password)
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
|
||||
writes := 0
|
||||
a.handoffWriter = func(_ string, got *handlers_rds.GetDBBootstrapConfigOutput) error {
|
||||
@@ -385,7 +393,7 @@ func TestBootstrap_RetriesInitializeWithNoPassword(t *testing.T) {
|
||||
}}
|
||||
cp.bootstrapOut = bootstrapOutput(&password)
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
|
||||
var handed []*handlers_rds.GetDBBootstrapConfigOutput
|
||||
a.handoffWriter = func(_ string, got *handlers_rds.GetDBBootstrapConfigOutput) error {
|
||||
@@ -408,7 +416,7 @@ func TestAcknowledgeBootstrap_WaitsForTheReceiptThenRetiresIt(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cfg := testConfig(t)
|
||||
cfg.DataMount = t.TempDir()
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
a.id = identity{DBInstanceIdentifier: "db-resolved"}
|
||||
a.pending = &pendingBootstrap{payloadID: testPayloadID, vmGeneration: 1, dataVolumeID: "vol-data-01"}
|
||||
|
||||
@@ -447,7 +455,7 @@ func TestAcknowledgeBootstrap_IgnoresAForeignReceipt(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cfg := testConfig(t)
|
||||
cfg.DataMount = t.TempDir()
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
a.id = identity{DBInstanceIdentifier: "db-resolved"}
|
||||
a.pending = &pendingBootstrap{payloadID: testPayloadID, vmGeneration: 1}
|
||||
writeReceipt(t, a, tc.payloadID, tc.dbID)
|
||||
@@ -470,7 +478,7 @@ func TestAcknowledgeBootstrap_IsANoOpOnAttach(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cfg := testConfig(t)
|
||||
cfg.DataMount = t.TempDir()
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
|
||||
if err := a.acknowledgeBootstrap(t.Context()); err != nil {
|
||||
t.Fatalf("acknowledgeBootstrap: %v", err)
|
||||
@@ -480,6 +488,52 @@ func TestAcknowledgeBootstrap_IsANoOpOnAttach(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The image is the authority on which engine the agent runs. A VM launched as
|
||||
// another engine has to stop ahead of the handoff: rds-init would otherwise
|
||||
// initialise one engine over the other's datadir and come up healthy and empty
|
||||
// beside data nothing references.
|
||||
func TestRun_RefusesAnEngineTheImageDoesNotBake(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cp.bootstrapOut = bootstrapOutput(nil)
|
||||
|
||||
cfg := testConfig(t)
|
||||
cfg.Engine = "mariadb"
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
|
||||
err := a.Run(t.Context())
|
||||
if err == nil {
|
||||
t.Fatal("Run returned nil for a VM launched as an engine this image does not bake")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "mariadb") {
|
||||
t.Errorf("error = %v, want it to name the engine the VM was launched as", err)
|
||||
}
|
||||
if _, statErr := os.Stat(filepath.Join(cfg.HandoffDir, handoffEnvFile)); !os.IsNotExist(statErr) {
|
||||
t.Error("the agent wrote a bootstrap handoff for an engine it refused to run")
|
||||
}
|
||||
if n := len(cp.bootstrapReqs); n != 0 {
|
||||
t.Errorf("bootstrap fetches = %d, want none before the refusal", n)
|
||||
}
|
||||
// Reported rather than only logged, so the control plane learns why the
|
||||
// instance never came up rather than only that it never did.
|
||||
states := cp.snapshotStates()
|
||||
if len(states) != 1 || states[0].health != handlers_rds.EngineHealthUnhealthy {
|
||||
t.Fatalf("submitted states = %+v, want a single unhealthy report", states)
|
||||
}
|
||||
}
|
||||
|
||||
// The same assertion when it agrees is not a gate: the boot goes on exactly as
|
||||
// it does for a VM launched before the control plane sent one at all.
|
||||
func TestRun_ProceedsWhenTheDeliveredEngineMatchesTheImage(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cp.bootstrapOut = bootstrapOutput(nil)
|
||||
|
||||
cfg := testConfig(t)
|
||||
cfg.Engine = enginePostgres
|
||||
if err := runAgent(t, newTestAgent(t, cfg, cp, staticProbe(0))); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A configured identifier is asserted on the wire so a mis-provisioned VM is
|
||||
// rejected by the gateway instead of adopting whatever it resolves to.
|
||||
func TestRun_SendsConfiguredIdentifier(t *testing.T) {
|
||||
@@ -489,7 +543,7 @@ func TestRun_SendsConfiguredIdentifier(t *testing.T) {
|
||||
cfg := testConfig(t)
|
||||
cfg.DBInstanceIdentifier = "db-configured"
|
||||
cfg.EngineVersion = "18.1"
|
||||
if err := runAgent(t, newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))); err != nil {
|
||||
if err := runAgent(t, newTestAgent(t, cfg, cp, staticProbe(0))); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
|
||||
@@ -503,12 +557,12 @@ func TestRun_WaitsForDataMountBeforePollingCommands(t *testing.T) {
|
||||
cp := newFakeControlPlane()
|
||||
cp.bootstrapOut = bootstrapOutput(nil)
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(2)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(2))
|
||||
|
||||
waiting := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
a.dataMountWaiter = func(ctx context.Context, mountsFile, target string) error {
|
||||
if mountsFile != defaultMountsFile || target != defaultDataMount {
|
||||
if mountsFile != defaultMountsFile || target != cfg.DataMount {
|
||||
t.Errorf("mount waiter got %q/%q, want configured defaults", mountsFile, target)
|
||||
}
|
||||
close(waiting)
|
||||
@@ -580,7 +634,7 @@ func TestRun_HeartbeatIncludesBootstrapFailure(t *testing.T) {
|
||||
cp.registerOut.HeartbeatIntervalSeconds = 0
|
||||
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(2)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(2))
|
||||
a.hb.interval = 10 * time.Millisecond
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
@@ -609,7 +663,7 @@ func TestRun_HeartbeatsEngineHealth(t *testing.T) {
|
||||
cp.registerOut.HeartbeatIntervalSeconds = 0
|
||||
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
a.hb.interval = 10 * time.Millisecond
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
@@ -639,7 +693,7 @@ func TestRegister_AdoptsControlPlaneCadence(t *testing.T) {
|
||||
cp.registerOut.HeartbeatIntervalSeconds = 45
|
||||
|
||||
cfg := testConfig(t)
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
if err := runAgent(t, a); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
@@ -656,12 +710,15 @@ func TestRun_HeartbeatReportsDownEngine(t *testing.T) {
|
||||
cp.registerOut.HeartbeatIntervalSeconds = 0
|
||||
|
||||
cfg := testConfig(t)
|
||||
probe := newEngineProbe(cfg, staticProbe(2))
|
||||
probe := newPostgresProbe(cfg, staticProbe(2))
|
||||
// The engine has served before, so a silent one now is a failure rather
|
||||
// than a boot still in progress.
|
||||
probe.seenHealthy = true
|
||||
|
||||
a := newAgent(cfg, cp, probe)
|
||||
a, err := newAgent(cfg, cp, probe)
|
||||
if err != nil {
|
||||
t.Fatalf("newAgent: %v", err)
|
||||
}
|
||||
a.hb.interval = 10 * time.Millisecond
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
@@ -752,7 +809,7 @@ func TestAcknowledgeBootstrap_StopsOnATerminalDenial(t *testing.T) {
|
||||
}}
|
||||
cfg := testConfig(t)
|
||||
cfg.DataMount = t.TempDir()
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
a.id = identity{DBInstanceIdentifier: "db-resolved"}
|
||||
a.pending = &pendingBootstrap{payloadID: testPayloadID, vmGeneration: 1}
|
||||
writeReceipt(t, a, testPayloadID, "db-resolved")
|
||||
@@ -780,7 +837,7 @@ func TestAcknowledgeBootstrap_DoesNotReportAnAbsentReceiptAsAFailure(t *testing.
|
||||
cp := newFakeControlPlane()
|
||||
cfg := testConfig(t)
|
||||
cfg.DataMount = t.TempDir()
|
||||
a := newAgent(cfg, cp, newEngineProbe(cfg, staticProbe(0)))
|
||||
a := newTestAgent(t, cfg, cp, staticProbe(0))
|
||||
a.id = identity{DBInstanceIdentifier: "db-resolved"}
|
||||
a.pending = &pendingBootstrap{payloadID: testPayloadID, vmGeneration: 1}
|
||||
|
||||
|
||||
+67
-42
@@ -1,7 +1,9 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/internal/guestenv"
|
||||
@@ -12,17 +14,10 @@ const (
|
||||
defaultGatewayCA = "/etc/spinifex-rds/gateway-ca.pem"
|
||||
defaultHandoffDir = "/run/spinifex-rds"
|
||||
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"
|
||||
// Where setup.sh stamps the engine the image bakes. The agent builds its
|
||||
// engine implementation from this file rather than from anything delivered.
|
||||
defaultEngineFile = "/etc/spinifex-rds/engine"
|
||||
defaultRCService = "rc-service"
|
||||
// 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
|
||||
@@ -44,17 +39,22 @@ type config struct {
|
||||
HandoffDir string
|
||||
EngineHost string
|
||||
EnginePort int
|
||||
// Overridable so a test or a sibling engine preset can point at its own.
|
||||
PGIsReady string
|
||||
PollWait time.Duration
|
||||
PollWait time.Duration
|
||||
|
||||
// The engine the control plane launched this VM as, delivered by cloud-init,
|
||||
// and the engine the image itself bakes. Two independent statements of the
|
||||
// same fact: the agent refuses to bootstrap when they disagree.
|
||||
Engine string
|
||||
EngineFile string
|
||||
BakedEngine string
|
||||
|
||||
// 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
|
||||
EngineBinDir string
|
||||
EngineDataDir string
|
||||
SocketDir string
|
||||
PGUser string
|
||||
EngineUser string
|
||||
RCService string
|
||||
EngineService string
|
||||
|
||||
@@ -75,18 +75,18 @@ func loadConfig(envFile string) config {
|
||||
GatewayCA: get("RDS_GATEWAY_CA"),
|
||||
Region: get("RDS_REGION"),
|
||||
DBInstanceIdentifier: get("RDS_DB_INSTANCE_IDENTIFIER"),
|
||||
Engine: get("RDS_ENGINE"),
|
||||
EngineFile: get("RDS_ENGINE_FILE"),
|
||||
EngineVersion: get("RDS_ENGINE_VERSION"),
|
||||
HandoffDir: get("RDS_HANDOFF_DIR"),
|
||||
EngineHost: get("RDS_ENGINE_HOST"),
|
||||
PGIsReady: get("RDS_PG_ISREADY"),
|
||||
PGBin: get("RDS_PG_BIN"),
|
||||
PGData: get("RDS_PGDATA"),
|
||||
EngineBinDir: get("RDS_ENGINE_BIN"),
|
||||
EngineDataDir: get("RDS_ENGINE_DATA"),
|
||||
SocketDir: get("RDS_SOCKET_DIR"),
|
||||
PGUser: get("RDS_PG_USER"),
|
||||
EngineUser: get("RDS_ENGINE_USER"),
|
||||
RCService: get("RDS_RC_SERVICE"),
|
||||
EngineService: get("RDS_ENGINE_SERVICE"),
|
||||
DataMount: get("RDS_DATA_MOUNT"),
|
||||
EnginePort: defaultEnginePort,
|
||||
PollWait: defaultPollWait,
|
||||
}
|
||||
if cfg.GatewayCA == "" {
|
||||
@@ -98,32 +98,20 @@ func loadConfig(envFile string) config {
|
||||
if cfg.EngineHost == "" {
|
||||
cfg.EngineHost = defaultEngineHost
|
||||
}
|
||||
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.EngineFile == "" {
|
||||
cfg.EngineFile = defaultEngineFile
|
||||
}
|
||||
if cfg.RCService == "" {
|
||||
cfg.RCService = defaultRCService
|
||||
}
|
||||
if cfg.EngineService == "" {
|
||||
cfg.EngineService = defaultEngineSvcName
|
||||
}
|
||||
// The image's own stamp rather than the delivered engine: an agent that took
|
||||
// its layout from the payload could run one engine's implementation against
|
||||
// another engine's datadir, which is the failure the stamp exists to prevent.
|
||||
cfg.BakedEngine = readBakedEngine(cfg.EngineFile)
|
||||
cfg.applyLayout(engineLayouts[cfg.BakedEngine])
|
||||
|
||||
// rds-datadir reads the same RDS_DATA_MOUNT, so the two agree on where the
|
||||
// volume landed without either asserting it to the other.
|
||||
if cfg.DataMount == "" {
|
||||
cfg.DataMount = defaultDataMount
|
||||
}
|
||||
cfg.MountsFile = defaultMountsFile
|
||||
cfg.SysBlock = defaultSysBlock
|
||||
// The authoritative port comes from the bootstrap config; this is only what
|
||||
@@ -140,3 +128,40 @@ func loadConfig(envFile string) config {
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
// Fills in whatever the delivered configuration left unset. An override always
|
||||
// wins, so a test can point any of these at a fixture. An unrecognised engine
|
||||
// leaves the layout empty, and New refuses rather than guessing at one.
|
||||
func (c *config) applyLayout(layout engineLayout) {
|
||||
if c.EngineBinDir == "" {
|
||||
c.EngineBinDir = layout.binDir
|
||||
}
|
||||
if c.EngineDataDir == "" {
|
||||
c.EngineDataDir = layout.dataDir
|
||||
}
|
||||
if c.SocketDir == "" {
|
||||
c.SocketDir = layout.socketDir
|
||||
}
|
||||
if c.EngineUser == "" {
|
||||
c.EngineUser = layout.osUser
|
||||
}
|
||||
if c.EngineService == "" {
|
||||
c.EngineService = layout.service
|
||||
}
|
||||
if c.DataMount == "" {
|
||||
c.DataMount = layout.dataMount
|
||||
}
|
||||
if c.EnginePort == 0 {
|
||||
c.EnginePort = layout.port
|
||||
}
|
||||
}
|
||||
|
||||
// The engine the image bakes. An absent or unreadable stamp reads as empty,
|
||||
// which New refuses: an agent that guessed would defeat the stamp.
|
||||
func readBakedEngine(path string) string {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return strings.ToLower(strings.TrimSpace(string(raw)))
|
||||
}
|
||||
|
||||
@@ -3,21 +3,31 @@ package main
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Stamps the engine a real image bakes and loads the configuration against it,
|
||||
// which is what selects the guest layout the agent's defaults come from.
|
||||
func testLoadConfig(t *testing.T, engine string) config {
|
||||
t.Helper()
|
||||
stamp := filepath.Join(t.TempDir(), "engine")
|
||||
if err := os.WriteFile(stamp, []byte(engine+"\n"), 0o444); err != nil {
|
||||
t.Fatalf("write the engine stamp: %v", err)
|
||||
}
|
||||
t.Setenv("RDS_ENGINE_FILE", stamp)
|
||||
return loadConfig(filepath.Join(t.TempDir(), "absent.env"))
|
||||
}
|
||||
|
||||
func TestLoadConfig_Defaults(t *testing.T) {
|
||||
cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
|
||||
if cfg.GatewayCA != defaultGatewayCA || cfg.HandoffDir != defaultHandoffDir {
|
||||
t.Errorf("CA/handoff = %q/%q, want the built-in defaults", cfg.GatewayCA, cfg.HandoffDir)
|
||||
}
|
||||
if cfg.EngineHost != defaultEngineHost || cfg.EnginePort != defaultEnginePort {
|
||||
t.Errorf("engine endpoint = %s:%d, want the local default", cfg.EngineHost, cfg.EnginePort)
|
||||
}
|
||||
if cfg.PGIsReady != defaultPGIsReady || cfg.PollWait != defaultPollWait {
|
||||
t.Errorf("probe/pollWait = %q/%v, want the built-in defaults", cfg.PGIsReady, cfg.PollWait)
|
||||
if cfg.EngineHost != defaultEngineHost || cfg.PollWait != defaultPollWait {
|
||||
t.Errorf("engine host/pollWait = %q/%v, want the built-in defaults", cfg.EngineHost, cfg.PollWait)
|
||||
}
|
||||
// Identity is optional: the gateway resolves it from the caller's
|
||||
// credentials, so an agent with no configured identifier is normal.
|
||||
@@ -26,11 +36,62 @@ func TestLoadConfig_Defaults(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The guest layout is the image's to state, not the control plane's: everything
|
||||
// the agent reaches the engine through comes from the stamp it was built with.
|
||||
func TestLoadConfig_TakesTheLayoutFromTheBakedEngine(t *testing.T) {
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
|
||||
if cfg.BakedEngine != enginePostgres {
|
||||
t.Fatalf("baked engine = %q, want %q", cfg.BakedEngine, enginePostgres)
|
||||
}
|
||||
layout := engineLayouts[enginePostgres]
|
||||
if cfg.EngineBinDir != layout.binDir || cfg.EngineDataDir != layout.dataDir {
|
||||
t.Errorf("bin/data = %q/%q, want the PostgreSQL layout", cfg.EngineBinDir, cfg.EngineDataDir)
|
||||
}
|
||||
if cfg.EngineUser != layout.osUser || cfg.EngineService != layout.service {
|
||||
t.Errorf("user/service = %q/%q, want the PostgreSQL layout", cfg.EngineUser, cfg.EngineService)
|
||||
}
|
||||
if cfg.SocketDir != layout.socketDir || cfg.DataMount != layout.dataMount {
|
||||
t.Errorf("socket/mount = %q/%q, want the PostgreSQL layout", cfg.SocketDir, cfg.DataMount)
|
||||
}
|
||||
if cfg.EnginePort != layout.port {
|
||||
t.Errorf("port = %d, want the PostgreSQL default %d", cfg.EnginePort, layout.port)
|
||||
}
|
||||
}
|
||||
|
||||
// An image with no stamp cannot say which engine it is, and an agent that
|
||||
// guessed would be the failure the stamp exists to prevent.
|
||||
func TestNewEngine_RefusesAnImageWithNoEngineStamp(t *testing.T) {
|
||||
t.Setenv("RDS_ENGINE_FILE", filepath.Join(t.TempDir(), "absent-engine"))
|
||||
cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
|
||||
|
||||
if _, err := newProbe(cfg, staticProbe(0)); err == nil {
|
||||
t.Error("newProbe built a probe for an image carrying no engine stamp")
|
||||
}
|
||||
if _, err := newEngine(cfg, nil, nil, nil); err == nil {
|
||||
t.Error("newEngine built an engine for an image carrying no engine stamp")
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewEngine_RefusesAnEngineItDoesNotImplement(t *testing.T) {
|
||||
cfg := testLoadConfig(t, "mariadb")
|
||||
|
||||
_, err := newEngine(cfg, nil, nil, nil)
|
||||
if err == nil {
|
||||
t.Fatal("newEngine built an engine this agent does not implement")
|
||||
}
|
||||
if !strings.Contains(err.Error(), "mariadb") {
|
||||
t.Errorf("error = %v, want it to name the engine the image bakes", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadConfig_ReadsEnvFile(t *testing.T) {
|
||||
t.Setenv("RDS_ENGINE_FILE", filepath.Join(t.TempDir(), "absent-engine"))
|
||||
path := filepath.Join(t.TempDir(), "agent.env")
|
||||
body := "RDS_GATEWAY_URL=https://gw.internal:9999\n" +
|
||||
"RDS_REGION=ap-southeast-2\n" +
|
||||
"RDS_DB_INSTANCE_IDENTIFIER=db-1\n" +
|
||||
"RDS_ENGINE=postgres\n" +
|
||||
"RDS_ENGINE_VERSION=18.1\n" +
|
||||
"RDS_ENGINE_PORT=6543\n" +
|
||||
"RDS_POLL_WAIT=5s\n"
|
||||
@@ -45,19 +106,22 @@ func TestLoadConfig_ReadsEnvFile(t *testing.T) {
|
||||
if cfg.DBInstanceIdentifier != "db-1" || cfg.EngineVersion != "18.1" {
|
||||
t.Errorf("identity = %q/%q, want db-1/18.1", cfg.DBInstanceIdentifier, cfg.EngineVersion)
|
||||
}
|
||||
if cfg.Engine != enginePostgres {
|
||||
t.Errorf("delivered engine = %q, want the one cloud-init carried", cfg.Engine)
|
||||
}
|
||||
if cfg.EnginePort != 6543 || cfg.PollWait != 5*time.Second {
|
||||
t.Errorf("port/pollWait = %d/%v, want 6543/5s", cfg.EnginePort, cfg.PollWait)
|
||||
}
|
||||
}
|
||||
|
||||
// An unparseable override falls back to the default rather than leaving the
|
||||
// agent with a zero port or a zero-length long poll.
|
||||
// An unparseable override falls back to the layout's own default rather than
|
||||
// leaving the agent with a zero port or a zero-length long poll.
|
||||
func TestLoadConfig_IgnoresUnusableOverrides(t *testing.T) {
|
||||
t.Setenv("RDS_ENGINE_PORT", "not-a-port")
|
||||
t.Setenv("RDS_POLL_WAIT", "forever")
|
||||
|
||||
cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
|
||||
if cfg.EnginePort != defaultEnginePort || cfg.PollWait != defaultPollWait {
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
if cfg.EnginePort != engineLayouts[enginePostgres].port || cfg.PollWait != defaultPollWait {
|
||||
t.Errorf("port/pollWait = %d/%v, want the defaults", cfg.EnginePort, cfg.PollWait)
|
||||
}
|
||||
}
|
||||
|
||||
+115
-8
@@ -56,6 +56,87 @@ type servingParameterRecorder interface {
|
||||
RecordServingParameters(ctx context.Context) error
|
||||
}
|
||||
|
||||
// Everything one engine implementation owns. The narrow interfaces above are
|
||||
// what the heartbeat, the commander and the rollback guard each hold; this is
|
||||
// only what the agent has to construct.
|
||||
type engine interface {
|
||||
engineOps
|
||||
parameterRecovery
|
||||
servingParameterRecorder
|
||||
// Follows the port the control plane assigned, once the bootstrap fetch
|
||||
// lands. Called from a different goroutine than the commands that read it.
|
||||
setPort(port int)
|
||||
}
|
||||
|
||||
// The guest layout an engine preset lays down, keyed by the engine the image
|
||||
// bakes. Every default that belongs to one engine lives here, so nothing
|
||||
// outside an implementation names its paths, service, port or probe.
|
||||
type engineLayout struct {
|
||||
binDir string
|
||||
dataDir string
|
||||
socketDir string
|
||||
osUser string
|
||||
service string
|
||||
dataMount string
|
||||
port int
|
||||
// The engine's own liveness signal, which is the only part of engineProbe
|
||||
// that is not shared.
|
||||
newProbe func(cfg config, run probeRunner) *engineProbe
|
||||
// The implementation every control-plane directive is served by.
|
||||
newEngine func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) engine
|
||||
}
|
||||
|
||||
// The name the control plane knows this implementation by, which is also what
|
||||
// setup.sh stamps into the image.
|
||||
const enginePostgres = "postgres"
|
||||
|
||||
var engineLayouts = map[string]engineLayout{
|
||||
enginePostgres: {
|
||||
binDir: "/usr/libexec/postgresql18",
|
||||
dataDir: "/var/lib/postgresql/18/data",
|
||||
socketDir: "/run/postgresql",
|
||||
osUser: "postgres",
|
||||
service: "postgresql",
|
||||
dataMount: "/var/lib/postgresql",
|
||||
port: 5432,
|
||||
newProbe: newPostgresProbe,
|
||||
newEngine: func(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) engine {
|
||||
return newPostgresEngine(cfg, run, startSess, probe)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// Builds the health probe the image's engine supplies.
|
||||
func newProbe(cfg config, run probeRunner) (*engineProbe, error) {
|
||||
layout, err := layoutFor(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return layout.newProbe(cfg, run), nil
|
||||
}
|
||||
|
||||
// Builds the implementation the image bakes. Nothing here consults the control
|
||||
// plane: the agent runs the engine it is made of, and the delivered engine is
|
||||
// only ever checked against it.
|
||||
func newEngine(cfg config, run commandRunner, startSess sessionRunner, probe *engineProbe) (engine, error) {
|
||||
layout, err := layoutFor(cfg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return layout.newEngine(cfg, run, startSess, probe), nil
|
||||
}
|
||||
|
||||
func layoutFor(cfg config) (engineLayout, error) {
|
||||
if cfg.BakedEngine == "" {
|
||||
return engineLayout{}, fmt.Errorf("this image carries no engine stamp at %s", cfg.EngineFile)
|
||||
}
|
||||
layout, ok := engineLayouts[cfg.BakedEngine]
|
||||
if !ok {
|
||||
return engineLayout{}, fmt.Errorf("this image bakes engine %q, which this agent does not implement", cfg.BakedEngine)
|
||||
}
|
||||
return layout, nil
|
||||
}
|
||||
|
||||
// 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 {
|
||||
@@ -145,17 +226,43 @@ type postgresEngine struct {
|
||||
held *quiesceHold
|
||||
}
|
||||
|
||||
var (
|
||||
_ engineOps = (*postgresEngine)(nil)
|
||||
_ parameterRecovery = (*postgresEngine)(nil)
|
||||
_ servingParameterRecorder = (*postgresEngine)(nil)
|
||||
)
|
||||
var _ engine = (*postgresEngine)(nil)
|
||||
|
||||
const (
|
||||
parameterRepairTimeout = 90 * time.Second
|
||||
parameterRepairPoll = time.Second
|
||||
)
|
||||
|
||||
// pg_isready is what gives PostgreSQL three states for free: exit 0 is serving,
|
||||
// exit 1 is a postmaster up and rejecting connections, and anything else is an
|
||||
// engine that is not there at all. Resolved on PATH, where the client package
|
||||
// puts it.
|
||||
const postgresProbeBinary = "pg_isready"
|
||||
|
||||
func newPostgresProbe(cfg config, run probeRunner) *engineProbe {
|
||||
return newEngineProbe(cfg.EnginePort, postgresProbeState(cfg.EngineHost, run))
|
||||
}
|
||||
|
||||
func postgresProbeState(host string, run probeRunner) probeStateFn {
|
||||
return func(ctx context.Context, port int64) (engineState, string) {
|
||||
portArg := strconv.FormatInt(port, 10)
|
||||
code, err := run(ctx, postgresProbeBinary, "-h", host, "-p", portArg, "-q")
|
||||
switch {
|
||||
case err != nil:
|
||||
// A missing binary or broken image. Reporting healthy on the strength of
|
||||
// nothing would hide it, so this reads as absent like an engine that did
|
||||
// not answer.
|
||||
return engineAbsent, fmt.Sprintf("engine probe could not run: %v", err)
|
||||
case code == 0:
|
||||
return engineServing, ""
|
||||
case code == 1:
|
||||
return engineRecovering, "engine is rejecting connections (startup or recovery)"
|
||||
default:
|
||||
return engineAbsent, fmt.Sprintf("engine did not respond on %s:%s", host, portArg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Resolved once, under the name the control plane knows this implementation by.
|
||||
// A lookup failure is a mismatch between the agent and the control plane it was
|
||||
// built against, so it fails at startup rather than at the first rotation.
|
||||
@@ -174,12 +281,12 @@ func newPostgresEngine(cfg config, run commandRunner, startSess sessionRunner, p
|
||||
meta: postgresEngineMeta,
|
||||
run: run,
|
||||
startSess: startSess,
|
||||
psql: filepath.Join(cfg.PGBin, "psql"),
|
||||
psql: filepath.Join(cfg.EngineBinDir, "psql"),
|
||||
rcService: cfg.RCService,
|
||||
service: cfg.EngineService,
|
||||
pgData: cfg.PGData,
|
||||
pgData: cfg.EngineDataDir,
|
||||
socketDir: cfg.SocketDir,
|
||||
osUser: cfg.PGUser,
|
||||
osUser: cfg.EngineUser,
|
||||
probe: probe,
|
||||
repairTimeout: parameterRepairTimeout,
|
||||
repairPoll: parameterRepairPoll,
|
||||
|
||||
@@ -141,9 +141,9 @@ func (r *recordingRunner) run(_ context.Context, c command) (string, error) {
|
||||
|
||||
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 {
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
cfg.EngineDataDir = t.TempDir()
|
||||
if err := os.MkdirAll(filepath.Join(cfg.EngineDataDir, "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
|
||||
@@ -152,8 +152,8 @@ func newTestEngine(t *testing.T, run commandRunner) *postgresEngine {
|
||||
if err != nil {
|
||||
t.Fatalf("resolve the current user: %v", err)
|
||||
}
|
||||
cfg.PGUser = current.Username
|
||||
return newPostgresEngine(cfg, run, nil, newEngineProbe(cfg, staticProbe(0)))
|
||||
cfg.EngineUser = current.Username
|
||||
return newPostgresEngine(cfg, run, nil, newPostgresProbe(cfg, staticProbe(0)))
|
||||
}
|
||||
|
||||
// The password reaches psql through the environment and is re-quoted there,
|
||||
@@ -341,7 +341,7 @@ func TestPostgresEngine_ApplyParametersRestartsOnARepairSetWhileEngineIsDown(t *
|
||||
codes := []int{2, 2, 0}
|
||||
probeCall := 0
|
||||
cfg := testProbeConfig()
|
||||
engine.probe = newEngineProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
engine.probe = newPostgresProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
code := codes[min(probeCall, len(codes)-1)]
|
||||
probeCall++
|
||||
return code, nil
|
||||
|
||||
+14
-26
@@ -3,9 +3,7 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"strconv"
|
||||
"sync/atomic"
|
||||
|
||||
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
|
||||
@@ -27,12 +25,16 @@ func execProbeRunner(ctx context.Context, name string, args ...string) (int, err
|
||||
return -1, err
|
||||
}
|
||||
|
||||
// Shells out to pg_isready rather than dialling the port, since only a startup
|
||||
// exchange separates an engine that is serving from one still in recovery.
|
||||
// What the engine reports about itself, which is the only part of the probe
|
||||
// that differs between engines. The port is passed in rather than read from the
|
||||
// engine, so the atomic below stays the one place it is followed from.
|
||||
type probeStateFn func(ctx context.Context, port int64) (engineState, string)
|
||||
|
||||
// Engine-neutral: the latch, the port, the degraded mapping and the translation
|
||||
// to a reported health are shared, and only the state function is supplied by
|
||||
// the engine the image bakes.
|
||||
type engineProbe struct {
|
||||
run probeRunner
|
||||
binary string
|
||||
host string
|
||||
stateFn probeStateFn
|
||||
// Set from the bootstrap config once it lands, from a different goroutine
|
||||
// than the heartbeat that reads it.
|
||||
port atomic.Int64
|
||||
@@ -41,9 +43,9 @@ type engineProbe struct {
|
||||
seenHealthy bool
|
||||
}
|
||||
|
||||
func newEngineProbe(cfg config, run probeRunner) *engineProbe {
|
||||
p := &engineProbe{run: run, binary: cfg.PGIsReady, host: cfg.EngineHost}
|
||||
p.port.Store(int64(cfg.EnginePort))
|
||||
func newEngineProbe(port int, stateFn probeStateFn) *engineProbe {
|
||||
p := &engineProbe{stateFn: stateFn}
|
||||
p.port.Store(int64(port))
|
||||
return p
|
||||
}
|
||||
|
||||
@@ -60,7 +62,7 @@ type engineState int
|
||||
const (
|
||||
// Not answering at all, or the probe could not run.
|
||||
engineAbsent engineState = iota
|
||||
// The postmaster is up and rejecting connections: startup or crash recovery.
|
||||
// The process is up but not yet serving: startup or crash recovery.
|
||||
engineRecovering
|
||||
engineServing
|
||||
)
|
||||
@@ -83,21 +85,7 @@ func (p *engineProbe) Check(ctx context.Context) (handlers_rds.EngineHealth, str
|
||||
// The raw probe result. Safe to call from a goroutine other than the
|
||||
// heartbeat's: it touches no latched state.
|
||||
func (p *engineProbe) state(ctx context.Context) (engineState, string) {
|
||||
port := strconv.FormatInt(p.port.Load(), 10)
|
||||
code, err := p.run(ctx, p.binary, "-h", p.host, "-p", port, "-q")
|
||||
switch {
|
||||
case err != nil:
|
||||
// A missing binary or broken image. Reporting healthy on the strength of
|
||||
// nothing would hide it, so this reads as absent like an engine that did
|
||||
// not answer.
|
||||
return engineAbsent, fmt.Sprintf("engine probe could not run: %v", err)
|
||||
case code == 0:
|
||||
return engineServing, ""
|
||||
case code == 1:
|
||||
return engineRecovering, "engine is rejecting connections (startup or recovery)"
|
||||
default:
|
||||
return engineAbsent, fmt.Sprintf("engine did not respond on %s:%s", p.host, port)
|
||||
}
|
||||
return p.stateFn(ctx, p.port.Load())
|
||||
}
|
||||
|
||||
// A non-answering engine: starting until it has answered once, unhealthy after.
|
||||
|
||||
@@ -28,7 +28,7 @@ func TestEngineProbe_MapsExitCodesToHealth(t *testing.T) {
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
probe := newEngineProbe(testProbeConfig(), func(context.Context, string, ...string) (int, error) {
|
||||
probe := newPostgresProbe(testProbeConfig(), func(context.Context, string, ...string) (int, error) {
|
||||
return tt.code, tt.runErr
|
||||
})
|
||||
probe.seenHealthy = tt.seenHealthy
|
||||
@@ -51,7 +51,7 @@ func TestEngineProbe_MapsExitCodesToHealth(t *testing.T) {
|
||||
// has to flip the first time the engine answers.
|
||||
func TestEngineProbe_LatchesAfterFirstHealthy(t *testing.T) {
|
||||
code := 0
|
||||
probe := newEngineProbe(testProbeConfig(), func(context.Context, string, ...string) (int, error) {
|
||||
probe := newPostgresProbe(testProbeConfig(), func(context.Context, string, ...string) (int, error) {
|
||||
return code, nil
|
||||
})
|
||||
|
||||
@@ -68,7 +68,7 @@ func TestEngineProbe_LatchesAfterFirstHealthy(t *testing.T) {
|
||||
// must follow it rather than keep asking the default.
|
||||
func TestEngineProbe_ProbesTheAssignedPort(t *testing.T) {
|
||||
var gotArgs []string
|
||||
probe := newEngineProbe(testProbeConfig(), func(_ context.Context, _ string, args ...string) (int, error) {
|
||||
probe := newPostgresProbe(testProbeConfig(), func(_ context.Context, _ string, args ...string) (int, error) {
|
||||
gotArgs = args
|
||||
return 0, nil
|
||||
})
|
||||
@@ -82,5 +82,5 @@ func TestEngineProbe_ProbesTheAssignedPort(t *testing.T) {
|
||||
}
|
||||
|
||||
func testProbeConfig() config {
|
||||
return config{EngineHost: defaultEngineHost, EnginePort: defaultEnginePort, PGIsReady: defaultPGIsReady}
|
||||
return config{EngineHost: defaultEngineHost, EnginePort: engineLayouts[enginePostgres].port}
|
||||
}
|
||||
|
||||
@@ -39,7 +39,7 @@ func TestHeartbeater_FirstServingProbeSeedsLastKnownGood(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestHeartbeater_BoundsBootstrapFailure(t *testing.T) {
|
||||
h := newHeartbeater(newFakeControlPlane(), newEngineProbe(testProbeConfig(), staticProbe(2)), nil, 0)
|
||||
h := newHeartbeater(newFakeControlPlane(), newPostgresProbe(testProbeConfig(), staticProbe(2)), nil, 0)
|
||||
h.setBootstrapFailure("bootstrap fetch", errors.New(strings.Repeat("界", 1000)))
|
||||
|
||||
failure := h.bootstrapFailure.Load()
|
||||
@@ -60,7 +60,7 @@ func TestHeartbeater_BoundsBootstrapFailure(t *testing.T) {
|
||||
func TestHeartbeater_ChecksServingParametersOnEveryHealthyProbe(t *testing.T) {
|
||||
code := 0
|
||||
cfg := testProbeConfig()
|
||||
probe := newEngineProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
probe := newPostgresProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
return code, nil
|
||||
})
|
||||
recorder := &countingServingRecorder{}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -369,8 +368,8 @@ func (f *fakeRecovery) Restart(context.Context) error {
|
||||
// A probe whose result the test drives directly, standing in for pg_isready.
|
||||
func stubProbe(t *testing.T, code int, err error) *engineProbe {
|
||||
t.Helper()
|
||||
cfg := loadConfig(filepath.Join(t.TempDir(), "absent.env"))
|
||||
return newEngineProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
cfg := testLoadConfig(t, enginePostgres)
|
||||
return newPostgresProbe(cfg, func(context.Context, string, ...string) (int, error) {
|
||||
return code, err
|
||||
})
|
||||
}
|
||||
|
||||
@@ -28,8 +28,6 @@ const (
|
||||
defaultXFSGrowfs = "xfs_growfs"
|
||||
defaultMountsFile = "/proc/mounts"
|
||||
defaultSysBlock = "/sys/class/block"
|
||||
// Where rds-datadir mounts the data volume; the datadir sits one level in.
|
||||
defaultDataMount = "/var/lib/postgresql"
|
||||
// growpart says this when the partition already fills the disk, which a
|
||||
// resumed grow is the normal way to reach.
|
||||
growpartNoChange = "NOCHANGE"
|
||||
|
||||
@@ -61,12 +61,12 @@ func newTestStorage(t *testing.T, device, fstype string, run commandRunner) *gue
|
||||
mounts := filepath.Join(dir, "mounts")
|
||||
content := "proc /proc proc rw 0 0\n"
|
||||
if device != "" {
|
||||
content += device + " " + defaultDataMount + " " + fstype + " rw,relatime 0 0\n"
|
||||
content += device + " " + testDataMount + " " + fstype + " rw,relatime 0 0\n"
|
||||
}
|
||||
if err := os.WriteFile(mounts, []byte(content), 0o600); err != nil {
|
||||
t.Fatalf("write the mount table: %v", err)
|
||||
}
|
||||
cfg := config{DataMount: defaultDataMount, MountsFile: mounts, SysBlock: filepath.Join(dir, "block")}
|
||||
cfg := config{DataMount: testDataMount, MountsFile: mounts, SysBlock: filepath.Join(dir, "block")}
|
||||
if err := os.MkdirAll(cfg.SysBlock, 0o755); err != nil {
|
||||
t.Fatalf("create the sysfs tree: %v", err)
|
||||
}
|
||||
@@ -93,6 +93,9 @@ func seedPartition(t *testing.T, g *guestStorage, disk, name, number string) {
|
||||
|
||||
// The data volume is formatted whole, so the usual grow is resize2fs against
|
||||
// the device with no partition to push out first.
|
||||
// Where rds-datadir mounts the data volume in the PostgreSQL preset.
|
||||
const testDataMount = "/var/lib/postgresql"
|
||||
|
||||
func TestGrowFilesystem_ResizesAnExt4DeviceOnline(t *testing.T) {
|
||||
log := &runLog{}
|
||||
g := newTestStorage(t, "/dev/vdb", "ext4", log.run)
|
||||
@@ -108,7 +111,7 @@ func TestGrowFilesystem_ResizesAnExt4DeviceOnline(t *testing.T) {
|
||||
if args := log.calls[0].Args; len(args) != 1 || args[0] != "/dev/vdb" {
|
||||
t.Errorf("resize2fs args = %v, want the device", args)
|
||||
}
|
||||
if !strings.Contains(message, "/dev/vdb") || !strings.Contains(message, defaultDataMount) {
|
||||
if !strings.Contains(message, "/dev/vdb") || !strings.Contains(message, testDataMount) {
|
||||
t.Errorf("reply message = %q, want the device and mount it grew", message)
|
||||
}
|
||||
}
|
||||
@@ -125,7 +128,7 @@ func TestGrowFilesystem_GrowsXFSByMountPoint(t *testing.T) {
|
||||
if got := log.names(); len(got) != 1 || got[0] != defaultXFSGrowfs {
|
||||
t.Fatalf("ran %v, want just %s", got, defaultXFSGrowfs)
|
||||
}
|
||||
if args := log.calls[0].Args; len(args) != 1 || args[0] != defaultDataMount {
|
||||
if args := log.calls[0].Args; len(args) != 1 || args[0] != testDataMount {
|
||||
t.Errorf("xfs_growfs args = %v, want the mount point", args)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -35,8 +35,8 @@ start_pre() {
|
||||
# shellcheck disable=SC1090
|
||||
. "${AGENT_ENV}"
|
||||
export RDS_GATEWAY_URL RDS_GATEWAY_CA RDS_REGION RDS_DB_INSTANCE_IDENTIFIER \
|
||||
RDS_ENGINE_VERSION RDS_HANDOFF_DIR RDS_ENGINE_HOST RDS_ENGINE_PORT \
|
||||
RDS_PG_ISREADY RDS_POLL_WAIT RDS_DATA_MOUNT
|
||||
RDS_ENGINE RDS_ENGINE_VERSION RDS_HANDOFF_DIR RDS_ENGINE_HOST RDS_ENGINE_PORT \
|
||||
RDS_POLL_WAIT RDS_DATA_MOUNT
|
||||
fi
|
||||
}
|
||||
|
||||
|
||||
@@ -9,7 +9,10 @@ source_time_handoff="${CASE}/source-time"
|
||||
configured_handoff="${CASE}/configured"
|
||||
mkdir -p "${configured_handoff}"
|
||||
touch "${configured_handoff}/bootstrap.env"
|
||||
printf 'RDS_HANDOFF_DIR=%s\n' "${configured_handoff}" >"${CASE}/agent.env"
|
||||
{
|
||||
printf 'RDS_HANDOFF_DIR=%s\n' "${configured_handoff}"
|
||||
printf 'RDS_ENGINE=postgres\n'
|
||||
} >"${CASE}/agent.env"
|
||||
|
||||
# Simulate OpenRC sourcing the service before start_pre loads agent.env.
|
||||
export RDS_HANDOFF_DIR="${source_time_handoff}"
|
||||
@@ -34,5 +37,12 @@ if [ "${EEND_STATUS}" != "0" ]; then
|
||||
echo "FAIL: post-start wait did not find the configured handoff" >&2
|
||||
exit 1
|
||||
fi
|
||||
# The agent checks this against the engine its own image bakes. Left out of the
|
||||
# export list it would read as no assertion at all, so a VM launched as the
|
||||
# wrong engine would bootstrap instead of refusing.
|
||||
if [ "${RDS_ENGINE:-}" != "postgres" ]; then
|
||||
echo "FAIL: start_pre did not export RDS_ENGINE from agent.env" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "rds-agent.initd: all tests passed"
|
||||
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
#!/bin/sh
|
||||
set -eu
|
||||
|
||||
# The engine this preset bakes is stated in three places that have to agree: the
|
||||
# stamp setup.sh writes, the manifest tag resolveEngineAMI selects on, and the
|
||||
# name rds-agent's layout table is keyed by. A drift between them launches the
|
||||
# wrong image or refuses the right one.
|
||||
|
||||
SCRIPT_DIR=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd)
|
||||
REPO_ROOT=$(CDPATH='' cd -- "${SCRIPT_DIR}/../../.." && pwd)
|
||||
ENGINE="postgres"
|
||||
|
||||
stamped=$(sed -n "s|^printf '\\(.*\\)\\\\n' > /etc/spinifex-rds/engine\$|\\1|p" "${SCRIPT_DIR}/setup.sh")
|
||||
if [ "${stamped}" != "${ENGINE}" ]; then
|
||||
echo "FAIL: setup.sh stamps engine '${stamped}', want '${ENGINE}'" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if ! grep -q "engine=${ENGINE} " "${SCRIPT_DIR}/manifest.conf"; then
|
||||
echo "FAIL: manifest.conf SYSTEM_TAG does not carry engine=${ENGINE}" >&2
|
||||
grep '^SYSTEM_TAG' "${SCRIPT_DIR}/manifest.conf" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if ! grep -q "enginePostgres = \"${ENGINE}\"" "${REPO_ROOT}/cmd/rds-agent/engine.go"; then
|
||||
echo "FAIL: rds-agent does not implement an engine named ${ENGINE}" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
echo "rds-engine-stamp: all tests passed"
|
||||
@@ -22,6 +22,13 @@ rc-update add mulga-mgmt-net boot
|
||||
# so the delivery lands in a root-only directory.
|
||||
install -d -m 0700 /etc/spinifex-rds
|
||||
|
||||
# The engine this image bakes. rds-agent builds its engine implementation from
|
||||
# this file, so running the wrong implementation is structurally impossible
|
||||
# rather than merely validated against; a VM launched as another engine is
|
||||
# refused before anything touches the datadir.
|
||||
printf 'postgres\n' > /etc/spinifex-rds/engine
|
||||
chmod 0444 /etc/spinifex-rds/engine
|
||||
|
||||
# Empty in the image: rds-datadir mounts over it at boot. Postgres-owned so the
|
||||
# engine can traverse it if the volume's own root is stricter.
|
||||
install -d -m 0750 -o postgres -g postgres /var/lib/postgresql
|
||||
|
||||
@@ -99,6 +99,7 @@ func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInsta
|
||||
GatewayCACert: s.deps.GatewayCACert,
|
||||
Region: s.region,
|
||||
DBInstanceIdentifier: req.Identifier,
|
||||
Engine: req.Engine.Name,
|
||||
EngineVersion: req.EngineVersion,
|
||||
EnginePort: req.Port,
|
||||
}),
|
||||
|
||||
@@ -180,6 +180,7 @@ func (s *Service) launchReplacementVM(ctx context.Context, accountID string, rec
|
||||
GatewayCACert: s.deps.GatewayCACert,
|
||||
Region: s.region,
|
||||
DBInstanceIdentifier: rec.DBInstanceIdentifier,
|
||||
Engine: rec.Engine,
|
||||
EngineVersion: rec.EngineVersion,
|
||||
EnginePort: rec.Port,
|
||||
}),
|
||||
|
||||
@@ -124,6 +124,7 @@ func (s *Service) RestoreDBInstanceFromDBSnapshot(ctx context.Context, input *rd
|
||||
GatewayCACert: s.deps.GatewayCACert,
|
||||
Region: s.region,
|
||||
DBInstanceIdentifier: req.Identifier,
|
||||
Engine: req.Engine.Name,
|
||||
EngineVersion: req.EngineVersion,
|
||||
EnginePort: req.Port,
|
||||
}),
|
||||
|
||||
@@ -20,8 +20,11 @@ type agentUserDataInput struct {
|
||||
GatewayCACert string
|
||||
Region string
|
||||
DBInstanceIdentifier string
|
||||
EngineVersion string
|
||||
EnginePort int64
|
||||
// The engine this VM is launched as. The agent checks it against the engine
|
||||
// its own image bakes and refuses to bootstrap when the two disagree.
|
||||
Engine string
|
||||
EngineVersion string
|
||||
EnginePort int64
|
||||
}
|
||||
|
||||
type userDataFile struct {
|
||||
@@ -36,6 +39,7 @@ func buildAgentUserData(in agentUserDataInput) string {
|
||||
"RDS_GATEWAY_CA=" + agentGatewayCAPath,
|
||||
"RDS_REGION=" + in.Region,
|
||||
"RDS_DB_INSTANCE_IDENTIFIER=" + in.DBInstanceIdentifier,
|
||||
"RDS_ENGINE=" + in.Engine,
|
||||
"RDS_ENGINE_VERSION=" + in.EngineVersion,
|
||||
fmt.Sprintf("RDS_ENGINE_PORT=%d", in.EnginePort),
|
||||
}, "\n")
|
||||
|
||||
@@ -14,6 +14,7 @@ func testAgentUserDataInput() agentUserDataInput {
|
||||
GatewayCACert: "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n",
|
||||
Region: testRegion,
|
||||
DBInstanceIdentifier: testDBID,
|
||||
Engine: enginePostgres.Name,
|
||||
EngineVersion: "18",
|
||||
EnginePort: 5432,
|
||||
}
|
||||
@@ -29,6 +30,9 @@ func TestBuildAgentUserData(t *testing.T) {
|
||||
assert.Contains(t, out, "RDS_GATEWAY_CA="+agentGatewayCAPath)
|
||||
assert.Contains(t, out, "RDS_REGION="+testRegion)
|
||||
assert.Contains(t, out, "RDS_DB_INSTANCE_IDENTIFIER="+testDBID)
|
||||
// The agent checks this against the engine its own image bakes and refuses
|
||||
// to bootstrap when the two disagree.
|
||||
assert.Contains(t, out, "RDS_ENGINE=postgres")
|
||||
assert.Contains(t, out, "RDS_ENGINE_VERSION=18")
|
||||
assert.Contains(t, out, "RDS_ENGINE_PORT=5432")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user