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

refactor: simplify RDS internals (#692)

Browse Source
This commit is contained in:
Julian Sommer
2026-08-04 14:42:45 +10:00
committed by GitHub
parent 2e09461080
commit 615578a2e7
14 changed files with 32 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/rds-agent/storage.go
+3 -9
View File
@@ -40,9 +40,6 @@ type guestStorage struct {
dataMount string
mountsFile string
sysBlock string
growpart string
resize2fs string
xfsGrowfs string
}
var _ storageOps = (*guestStorage)(nil)
@@ -53,9 +50,6 @@ func newGuestStorage(cfg config, run commandRunner) *guestStorage {
dataMount: cfg.DataMount,
mountsFile: cfg.MountsFile,
sysBlock: cfg.SysBlock,
growpart: defaultGrowpart,
resize2fs: defaultResize2fs,
xfsGrowfs: defaultXFSGrowfs,
}
}
@@ -82,7 +76,7 @@ func (g *guestStorage) GrowFilesystem(ctx context.Context) (string, error) {
case strings.HasPrefix(mount.fstype, "ext"):
// resize2fs grows a mounted ext filesystem online, and takes the device.
if _, err := g.run(ctx, command{
Name: g.resize2fs,
Name: defaultResize2fs,
Args: []string{mount.device},
Env: []string{"PATH=" + defaultGuestPath},
}); err != nil {
@@ -91,7 +85,7 @@ func (g *guestStorage) GrowFilesystem(ctx context.Context) (string, error) {
case mount.fstype == "xfs":
// xfs_growfs is addressed by mount point, and XFS can only grow mounted.
if _, err := g.run(ctx, command{
Name: g.xfsGrowfs,
Name: defaultXFSGrowfs,
Args: []string{g.dataMount},
Env: []string{"PATH=" + defaultGuestPath},
}); err != nil {
@@ -116,7 +110,7 @@ func (g *guestStorage) growPartition(ctx context.Context, device string) error {
return nil
}
out, err := g.run(ctx, command{
Name: g.growpart,
Name: defaultGrowpart,
Args: []string{disk, number},
Env: []string{"PATH=" + defaultGuestPath},
})
internal/rdsgw/client.go
-1
View File
@@ -111,7 +111,6 @@ type errorResponse struct {
Code string `xml:"Code"`
Message string `xml:"Message"`
} `xml:"Error"`
RequestID string `xml:"RequestId"`
}
// params carries the action's own arguments; Action and Version are set here.
scripts/images/rds-postgres/rds-datadir
+2 -2
View File
@@ -83,7 +83,7 @@ await_data_disk() {
# when it could not find out: a blkid that is missing, errored or pointed at an
# absent node prints exactly what a blank disk prints, so only the status decides.
fstype_of() {
if _out=$(blkid "$1" 2>/dev/null); then
if _out=$(blkid -o value -s TYPE "$1" 2>/dev/null); then
_rc=0
else
_rc=$?
@@ -91,7 +91,7 @@ fstype_of() {
# 0 is a signature found, 2 a clean probe that found none. Anything else
# means the probe did not run, which is not an answer.
[ "${_rc}" -eq 0 ] || [ "${_rc}" -eq 2 ] || return 1
printf '%s' "${_out}" | sed -n 's/.*[[:space:]]TYPE="\([^"]*\)".*/\1/p'
printf '%s' "${_out}"
}
# Partitions appear as child directories of the disk in sysfs, which needs no
scripts/images/rds-postgres/rds-datadir_test.sh
+5 -4
View File
@@ -16,18 +16,19 @@ trap 'rm -rf "${WORK}"' EXIT
STUBBIN="${WORK}/bin"
mkdir -p "${STUBBIN}"
# blkid stub: prints a real blkid's line for any device listed in FS_TABLE as
# "<dev> <fstype>", and exits non-zero for one that holds no filesystem.
# blkid stub: prints the requested filesystem type value for any device listed
# in FS_TABLE as "<dev> <fstype>", and exits non-zero when none is present.
cat > "${STUBBIN}/blkid" <<'EOF'
#!/bin/sh
# BLKID_RC simulates a probe that could not run at all — 127 is a blkid missing
# from the image, 4 an internal error. Both print nothing, which is precisely
# what a genuinely blank disk prints, so only the exit status separates them.
[ "${BLKID_RC:-0}" = "0" ] || exit "${BLKID_RC}"
dev="$1"
[ "$1" = "-o" ] && [ "$2" = "value" ] && [ "$3" = "-s" ] && [ "$4" = "TYPE" ] || exit 4
dev="$5"
fstype=$(awk -v d="${dev}" '$1 == d { print $2 }' "${FS_TABLE}" 2>/dev/null)
[ -n "${fstype}" ] || exit 2
echo "${dev}: UUID=\"1234-5678\" TYPE=\"${fstype}\""
printf '%s\n' "${fstype}"
EOF
# mkfs.ext4 stub: records the call and marks the device formatted, so a mount
spinifex/handlers/rds/create.go
-1
View File
@@ -88,7 +88,6 @@ func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInsta
EngineVersion: req.EngineVersion,
InstanceType: req.InstanceType,
AllocatedStorage: req.AllocatedStorage,
RequireEncryptedData: true,
UserData: buildAgentUserData(agentUserDataInput{
GatewayURL: s.deps.GatewayURL,
GatewayCACert: s.deps.GatewayCACert,
spinifex/handlers/rds/dns.go
+6 -6
View File
@@ -11,19 +11,19 @@ import (
// The vanity hostname for a DB instance, or "" on a deployment with no base
// domain — where the endpoint is the bare ENI IP instead.
func (s *Service) dnsName(accountID, dbInstanceIdentifier string) string {
if s.baseDomain == "" {
if s.deps.BaseDomain == "" {
return ""
}
return handlers_dns.RDSName(dbInstanceIdentifier, accountID, s.region, s.baseDomain)
return handlers_dns.RDSName(dbInstanceIdentifier, accountID, s.region, s.deps.BaseDomain)
}
// Best-effort: a miss is repaired by the reconcile backstop within a cycle, and
// a create must not fail because northstar was briefly unreachable.
func (s *Service) publishDNS(ctx context.Context, accountID string, rec *DBInstanceRecord, action handlers_dns.Action) {
if s.baseDomain == "" || rec == nil || rec.DNSName == "" || rec.ENIPrivateIP == "" {
if s.deps.BaseDomain == "" || rec == nil || rec.DNSName == "" || rec.ENIPrivateIP == "" {
return
}
changes := handlers_dns.RDSChanges(action, rec.DNSName, s.baseDomain, rec.ENIPrivateIP)
changes := handlers_dns.RDSChanges(action, rec.DNSName, s.deps.BaseDomain, rec.ENIPrivateIP)
slog.DebugContext(ctx, "rds: publishing endpoint record",
"dbInstance", rec.DBInstanceIdentifier, "name", rec.DNSName, "action", action)
handlers_dns.PublishChangesBestEffort(s.nc, accountID, changes)
@@ -35,7 +35,7 @@ func (s *Service) publishDNS(ctx context.Context, accountID string, rec *DBInsta
// failure yields ok=false, which suppresses RDS pruning rather than deleting a
// tenant's endpoint on a partial view.
func (s *Service) DesiredDNSChanges() (changes []handlers_dns.Change, ok bool) {
if s == nil || s.baseDomain == "" {
if s == nil || s.deps.BaseDomain == "" {
return nil, false
}
ctx := context.Background()
@@ -54,7 +54,7 @@ func (s *Service) DesiredDNSChanges() (changes []handlers_dns.Change, ok bool) {
slog.Warn("rds DesiredDNSChanges: open account bucket", "bucket", bucket, "err", err)
return nil, false
}
bucketChanges, err := desiredBucketDNSChanges(ctx, kv, s.baseDomain)
bucketChanges, err := desiredBucketDNSChanges(ctx, kv, s.deps.BaseDomain)
if err != nil {
slog.Warn("rds DesiredDNSChanges: read DB instances", "bucket", bucket, "err", err)
return nil, false
spinifex/handlers/rds/launch.go
+5 -14
View File
@@ -108,9 +108,6 @@ type LaunchInput struct {
AllocatedStorage int64
UserData string
IamInstanceProfileArn string
// Fails the launch when the created data volume comes back unencrypted,
// rather than reporting an encryption the customer did not get.
RequireEncryptedData bool
// A replace re-uses the DB instance's persisted customer ENI and data
// volume instead of minting new ones: the endpoint address and the
@@ -124,13 +121,10 @@ type LaunchOutput struct {
InstanceID string
// The system ENI is disposable — a replace makes a new one. The customer
// ENI is the stable endpoint: its IP is the DNS target and survives.
SystemENIID string
CustomerENIID string
CustomerENIIP string
CustomerENIMac string
DataVolumeID string
DataDevice string
MgmtIP string
SystemENIID string
CustomerENIID string
CustomerENIIP string
DataVolumeID string
// The volume's own reported state, not an echo of the request, so
// DescribeDBInstances reports encryption the way EC2 does. Meaningful only
// when this launch created the volume; a replace or restore attaches one
@@ -283,7 +277,7 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
// Checked before the attach so an unencrypted volume is unwound rather than
// mounted: the cluster storage key is unset, which no retry here can fix.
if in.RequireEncryptedData && !volumeEncrypted {
if !volumeEncrypted {
return nil, fmt.Errorf("rds: data volume %s for %s was created unencrypted; the cluster storage key is not configured",
volumeID, in.DBInstanceIdentifier)
}
@@ -304,10 +298,7 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
SystemENIID: systemENI.id,
CustomerENIID: customerENI.id,
CustomerENIIP: customerENI.ip,
CustomerENIMac: customerENI.mac,
DataVolumeID: volumeID,
DataDevice: device,
MgmtIP: sysOut.MgmtIP,
DataVolumeEncrypted: volumeEncrypted,
CreatedDataVolume: in.ExistingDataVolume == "",
Unwind: unwind,
spinifex/handlers/rds/launch_test.go
+3 -5
View File
@@ -275,7 +275,7 @@ func (f *fakeLauncher) LaunchSystemInstance(in *sysinstance.SystemInstanceInput)
if f.onLaunch != nil {
f.onLaunch()
}
return &sysinstance.SystemInstanceOutput{InstanceID: "i-rds0001", MgmtIP: "172.30.0.9"}, nil
return &sysinstance.SystemInstanceOutput{InstanceID: "i-rds0001"}, nil
}
func (f *fakeLauncher) TerminateSystemInstance(instanceID string) error {
@@ -390,7 +390,7 @@ func newLaunchHarness() *launchHarness {
images: &fakeImages{images: []*ec2.Image{
{ImageId: aws.String(testEngineAMI), CreationDate: aws.String("2026-01-02T00:00:00Z")},
}},
volumes: &fakeVolumes{},
volumes: &fakeVolumes{encrypted: true},
attacher: &fakeAttacher{},
}
h.enis.unwind, h.launcher.unwind, h.volumes.unwind = &h.unwind, &h.unwind, &h.unwind
@@ -480,7 +480,7 @@ func TestLaunchDBInstanceVMWiresBothNICs(t *testing.T) {
require.Len(t, in.ExtraENIs, 1)
assert.Equal(t, sysinstance.ExtraENIInput{
ENIID: out.CustomerENIID,
ENIMac: out.CustomerENIMac,
ENIMac: "02:00:00:00:00:02",
ENIIP: out.CustomerENIIP,
SubnetID: testDBSubnet,
AccountID: testCustomerAccount,
@@ -489,7 +489,6 @@ func TestLaunchDBInstanceVMWiresBothNICs(t *testing.T) {
assert.Equal(t, "arn:aws:iam::000000000000:instance-profile/rdsInstanceRole", in.IamInstanceProfileArn)
assert.Equal(t, "i-rds0001", out.InstanceID)
assert.Equal(t, "172.30.0.9", out.MgmtIP)
}
func TestLaunchDBInstanceVMAttachesTheDataVolume(t *testing.T) {
@@ -515,7 +514,6 @@ func TestLaunchDBInstanceVMAttachesTheDataVolume(t *testing.T) {
assert.Equal(t, utils.GlobalAccountID, h.attacher.accountID)
assert.Equal(t, "vol-rdsdata01", out.DataVolumeID)
assert.Equal(t, dataVolumeDevice, out.DataDevice)
}
func TestLaunchDBInstanceVMRollsBackEverythingItCreated(t *testing.T) {
spinifex/handlers/rds/modify.go
+2 -9
View File
@@ -112,7 +112,7 @@ func (s *Service) ModifyDBInstance(ctx context.Context, input *rds.ModifyDBInsta
return nil, err
}
if !plan.disruptive() {
stored, err := s.reloadInstance(ctx, kv, id)
stored, _, err := s.getDBInstance(ctx, kv, id)
if err != nil {
return nil, err
}
@@ -157,20 +157,13 @@ func (s *Service) ModifyDBInstance(ctx context.Context, input *rds.ModifyDBInsta
// Returned as modifying: the replacement or restarted VM has to come back
// and report healthy before the reconciler calls it available.
stored, err := s.reloadInstance(ctx, kv, id)
stored, _, err := s.getDBInstance(ctx, kv, id)
if err != nil {
return nil, err
}
return &rds.ModifyDBInstanceOutput{DBInstance: s.projectDBInstance(stored)}, nil
}
// The record as stored, for a response that reports what actually landed rather
// than what the handler believed it wrote.
func (s *Service) reloadInstance(ctx context.Context, kv jetstream.KeyValue, id string) (*DBInstanceRecord, error) {
rec, _, err := s.getDBInstance(ctx, kv, id)
return rec, err
}
// Resolves the request against the stored record: drops every field that
// repeats its current value, and rejects everything that cannot be delivered.
func (s *Service) planModify(ctx context.Context, input *rds.ModifyDBInstanceInput, accountID string, rec *DBInstanceRecord) (*modifyPlan, error) {
spinifex/handlers/rds/parametergroup.go
+1 -9
View File
@@ -214,7 +214,7 @@ func (s *Service) DescribeDBParameters(ctx context.Context, input *rds.DescribeD
if err != nil {
return nil, err
}
memoryMiB, err := classMemoryMiB(describeParametersClass())
memoryMiB, err := classMemoryMiB(smallestInstanceClass())
if err != nil {
return nil, err
}
@@ -412,14 +412,6 @@ func validateParameterGroupFamily(family string) (string, error) {
"DBParameterGroupFamily %q is not offered; %s is the only supported family", family, enginePostgres.ParameterGroupFamily())
}
// The class DescribeDBParameters evaluates the size-derived defaults at. A group
// has no instance, so the smallest supported class is the honest choice: it is
// the one whose computed values a customer most needs to see before attaching
// the group to anything.
func describeParametersClass() string {
return smallestInstanceClass()
}
func parameterSource(isOverride bool) string {
if isOverride {
return ParameterSourceUser
spinifex/handlers/rds/parametergroup_test.go
+2 -2
View File
@@ -278,8 +278,8 @@ func TestDescribeDBParameters_ReportsComputedDefaultsAsLiterals(t *testing.T) {
shared := params["shared_buffers"]
require.NotNil(t, shared)
// The class is named rather than read back from describeParametersClass, which
// would make this assert the code agrees with itself.
// The class is named rather than derived from the class catalog, which would
// make this assert the code agrees with itself.
memoryMiB, err := classMemoryMiB("db.t3.micro")
require.NoError(t, err)
assert.Equal(t, sharedBuffersFor(memoryMiB), aws.StringValue(shared.ParameterValue))
spinifex/handlers/rds/service.go
+3 -5
View File
@@ -64,10 +64,9 @@ type Deps struct {
// The RDS control plane's KV-backed handler set. One per daemon.
type Service struct {
nc *nats.Conn
region string
baseDomain string
deps Deps
nc *nats.Conn
region string
deps Deps
// Heartbeat state that never reaches KV: beats are counted here and
// persisted only on change or on the slower floor.
@@ -89,7 +88,6 @@ func NewService(nc *nats.Conn, region string) *Service {
func (s *Service) WithDeps(d Deps) *Service {
s.deps = d
s.baseDomain = d.BaseDomain
return s
}
spinifex/handlers/rds/status.go
-46
View File
@@ -49,19 +49,6 @@ var transitions = map[Status][]Status{
StatusDeleted: nil,
}
// Statuses meaning the control plane is already acting on this instance, so the
// recovery classifier does not recover a VM out from under a lifecycle op.
var transitional = map[Status]bool{
StatusCreating: true,
StatusModifying: true,
StatusBackingUp: true,
StatusRebooting: true,
StatusStopping: true,
StatusStarting: true,
StatusRecovering: true,
StatusDeleting: true,
}
// Anything read back from KV that fails this check was written by a newer or
// corrupted control plane, not a state to act on.
func (s Status) Valid() bool {
@@ -69,14 +56,6 @@ func (s Status) Valid() bool {
return ok
}
func (s Status) Terminal() bool {
return s == StatusDeleted
}
func (s Status) Transitional() bool {
return transitional[s]
}
// A no-op is legal for any valid status, so a repeated observation of the same
// state is not treated as an illegal transition.
func CanTransition(from, to Status) bool {
@@ -88,28 +67,3 @@ func CanTransition(from, to Status) bool {
}
return slices.Contains(transitions[from], to)
}
// The snapshot's own lifecycle, separate from the instance's backing-up state
// The instance is derived from an in-flight control-plane operation, while
// the snapshot record moves creating → available once the data exists. There is
// no failed state — a snapshot that never completed is removed rather than kept
// as an unrestorable record.
var snapshotTransitions = map[string][]string{
SnapshotStatusCreating: {SnapshotStatusAvailable},
SnapshotStatusAvailable: nil,
}
func ValidSnapshotStatus(status string) bool {
_, ok := snapshotTransitions[status]
return ok
}
func CanTransitionSnapshot(from, to string) bool {
if !ValidSnapshotStatus(from) || !ValidSnapshotStatus(to) {
return false
}
if from == to {
return true
}
return slices.Contains(snapshotTransitions[from], to)
}
spinifex/handlers/rds/status_test.go
-18
View File
@@ -21,24 +21,6 @@ func TestStatus_Valid(t *testing.T) {
}
}
func TestStatus_TerminalAndTransitional(t *testing.T) {
assert.True(t, StatusDeleted.Terminal())
assert.False(t, StatusFailed.Terminal())
assert.False(t, StatusStopped.Terminal())
// A lifecycle operation is already driving these, so the recovery classifier
// must leave them alone.
for _, s := range []Status{
StatusCreating, StatusModifying, StatusBackingUp, StatusRebooting,
StatusStopping, StatusStarting, StatusRecovering, StatusDeleting,
} {
assert.True(t, s.Transitional(), "status %q should be transitional", s)
}
for _, s := range []Status{StatusAvailable, StatusStopped, StatusFailed, StatusDeleted} {
assert.False(t, s.Transitional(), "status %q should be settled, not transitional", s)
}
}
func TestCanTransition_Lifecycle(t *testing.T) {
tests := []struct {
from, to Status