feat: implement RDS CreateDBInstance, reconciler and endpoint DNS (#594)

This commit is contained in:
Julian Sommer
2026-07-29 11:32:59 +10:00
committed by GitHub
parent 32d80b473a
commit 2569601e3d
42 changed files with 3168 additions and 53 deletions
+10
View File
@@ -318,6 +318,16 @@ suites:
# so vpcd (OVN + awsvpc ENI), viperblockd (root volume), predastore (AMI)
# are all in the path.
e2e-rds:
path: tests/e2e/rds
covers: [awsgw, spinifex_daemon, vpcd, viperblockd, predastore]
# RDS create/describe/connect: CreateDBInstance boots a dual-NIC DB VM from
# the engine AMI (predastore) with a Viperblock data volume (viperblockd) and
# a customer ENI (vpcd), and the reconciler flips it to available on the
# in-guest agent's first healthy heartbeat over awsgw. Opt-in beyond
# SPINIFEX_E2E: DeleteDBInstance is not implemented, so a run leaves the
# instance behind.
e2e-ddil:
path: tests/e2e/ddil
covers: [awsgw, spinifex_daemon, vpcd, viperblockd, predastore]
+12 -3
View File
@@ -458,8 +458,13 @@ var (
ErrorIAMMalformedPolicyDocument = "MalformedPolicyDocument"
ErrorAccessDenied = "AccessDenied"
// RDS-specific error codes.
ErrorDBInstanceNotFound = "DBInstanceNotFound"
// RDS-specific error codes. The "Fault" suffix is AWS's own for the group
// lookups, so the SDK's typed error matching round-trips.
ErrorDBInstanceNotFound = "DBInstanceNotFound"
ErrorDBInstanceAlreadyExists = "DBInstanceAlreadyExists"
ErrorDBSubnetGroupNotFound = "DBSubnetGroupNotFoundFault"
ErrorDBParameterGroupNotFound = "DBParameterGroupNotFound"
ErrorDBInvalidVPCNetworkState = "InvalidVPCNetworkStateFault"
// ECR-specific error codes.
ErrorRepositoryNotFound = "RepositoryNotFoundException"
@@ -1027,7 +1032,11 @@ var ErrorLookup = map[string]ErrorMessage{
ErrorAccessDenied: {HTTPCode: 403, Message: "User is not authorized to perform this action."},
// RDS error codes
ErrorDBInstanceNotFound: {HTTPCode: 404, Message: "DBInstanceIdentifier does not refer to an existing DB instance."},
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."},
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."},
// ECR error codes
ErrorRepositoryNotFound: {HTTPCode: 400, Message: "The repository could not be found. Check the spelling of the specified repository and ensure that you are performing operations on the correct registry."},
+5
View File
@@ -163,6 +163,11 @@ type RDSConfig struct {
// Clamped to 1..3. Zero defaults to one, which is all a single-AZ platform
// can place across.
SystemVPCPrivateSubnets int `json:"SystemVPCPrivateSubnets" mapstructure:"system_vpc_private_subnets"`
// How long a creating DB instance may go without a healthy agent heartbeat
// before the reconciler marks it failed. Zero takes the built-in default,
// which covers a cold boot plus initdb on the smallest instance class.
BootstrapTimeoutSeconds int `json:"BootstrapTimeoutSeconds" mapstructure:"bootstrap_timeout_seconds"`
}
// RDSDefaultSystemVPCSupernet anchors the RDS system VPC address space at
+18 -7
View File
@@ -150,6 +150,7 @@ type Daemon struct {
ecsService *handlers_ecs.Service
ecsScheduler *handlers_ecs.Scheduler
rdsService *handlers_rds.Service
rdsReconciler *handlers_rds.Reconciler
acmService *handlers_acm.ACMServiceImpl
ecrMetaService *handlers_ecr.MetaServiceImpl
routeTableService *handlers_ec2_routetable.RouteTableServiceImpl
@@ -1055,6 +1056,8 @@ func (d *Daemon) subscribeAll() error {
natsSub{handlers_rds.SubjectRegisterWildcard, handleNATSRequest(d.rdsService.RegisterDBInstance), "spinifex-workers"},
natsSub{handlers_rds.SubjectHealthWildcard, handleNATSRequest(d.rdsService.SubmitDBStateChange), "spinifex-workers"},
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"},
)
}
@@ -1588,13 +1591,21 @@ func (d *Daemon) startCluster() error {
})
}
// RDS control plane: KV-backed agent-protocol handlers. The cluster CA signs
// the per-instance serving certs, minted per bootstrap fetch and never
// persisted; ca.key sits beside the configured ca.pem.
d.rdsService = handlers_rds.NewService(d.natsConn, d.config.Region).WithDeps(handlers_rds.Deps{
CACertPath: d.config.NATS.CACert,
CAKeyPath: clusterCAKeyPath(d.config.NATS.CACert),
Launch: d.buildRDSLaunchDeps(),
// RDS control plane: KV-backed agent-protocol handlers plus the customer
// actions. The cluster CA signs the per-instance serving certs, minted per
// bootstrap fetch and never persisted; ca.key sits beside the configured ca.pem.
d.rdsService = handlers_rds.NewService(d.natsConn, d.config.Region).WithDeps(d.buildRDSDeps())
// One leader across the cluster derives DB instance status from the agent
// heartbeat; every node keeps serving the API.
d.rdsReconciler = handlers_rds.NewReconciler(d.rdsService, d.node)
d.shutdownWg.Go(func() {
defer func() {
if r := recover(); r != nil {
slog.Error("RDS reconciler goroutine panicked", "recover", r)
}
}()
d.rdsReconciler.Run(d.ctx)
})
d.acmService, err = initServiceWithRetry("ACM service", func() (*handlers_acm.ACMServiceImpl, error) {
+8 -2
View File
@@ -7,8 +7,8 @@ import (
)
// dnsDesiredSet builds the full desired managed-record set for the reconcile
// backstop. It spans all tenants: every running instance on this
// node plus every active load balancer and EKS cluster across all accounts.
// backstop. It spans all tenants: every running instance on this node plus
// every active load balancer, EKS cluster and DB instance across all accounts.
// Prune authority is granted per record class only when that class enumerated
// completely, so a transient store error can never delete another tenant's live
// records — the reconcile only ever repairs, never over-prunes, on a partial view.
@@ -28,6 +28,12 @@ func (d *Daemon) dnsDesiredSet() handlers_dns.DesiredSet {
ds.Prunable.EKS = true
}
}
if d.rdsService != nil {
if ch, ok := d.rdsService.DesiredDNSChanges(); ok {
ds.Changes = append(ds.Changes, ch...)
ds.Prunable.RDS = true
}
}
return ds
}
+48
View File
@@ -1,6 +1,13 @@
package daemon
import (
"context"
"log/slog"
"os"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
handlers_systemvpc "github.com/mulgadc/spinifex/spinifex/handlers/systemvpc"
)
@@ -26,3 +33,44 @@ func (d *Daemon) buildRDSLaunchDeps() handlers_rds.LaunchDeps {
Attacher: handlers_rds.NewNATSVolumeAttacher(d.natsConn),
}
}
// The control-plane deps on top of the launch primitives: where the endpoint
// ENI is placed, the instance role the agent authenticates with, the base zone
// the endpoint record lands in, and the VM-state lookup the reconciler needs.
func (d *Daemon) buildRDSDeps() handlers_rds.Deps {
gatewayCA := ""
if d.config.NATS.CACert != "" {
if caBytes, err := os.ReadFile(d.config.NATS.CACert); err == nil {
gatewayCA = string(caBytes)
} else {
slog.Warn("RDS: read gateway CACert failed; the DB VM agent will not verify the gateway over TLS",
"path", d.config.NATS.CACert, "err", err)
}
}
return handlers_rds.Deps{
CACertPath: d.config.NATS.CACert,
CAKeyPath: clusterCAKeyPath(d.config.NATS.CACert),
Launch: d.buildRDSLaunchDeps(),
Network: d.vpcService,
IAM: d.systemRoleEnsurer,
InstanceState: handlers_rds.NewDescribeInstanceState(d.describeInstancesFanOut),
BaseDomain: d.dnsBaseDomain,
HolderID: d.node,
// A DB VM reaches the daemon only over the mgmt bridge, the same
// constraint the EKS control-plane VM has.
GatewayURL: d.resolveSystemGatewayBaseURL(),
GatewayCACert: gatewayCA,
BootstrapTimeout: time.Duration(d.config.RDS.BootstrapTimeoutSeconds) * time.Second,
}
}
// Fans out across every host so a DB VM is observed wherever it landed; a
// node-local describe would report an instance on another node as missing.
func (d *Daemon) describeInstancesFanOut(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error) {
expected := 1
if d.clusterConfig != nil && len(d.clusterConfig.Nodes) > 0 {
expected = len(d.clusterConfig.Nodes)
}
return gateway_ec2_instance.DescribeInstances(context.Background(), input, d.natsConn, expected, accountID)
}
+1 -4
View File
@@ -15,9 +15,6 @@ import (
"github.com/nats-io/nats.go/jetstream"
)
// Only a session assumed from this role may call the internal agent actions.
const InstanceRoleName = "rdsInstanceRole"
// Long-poll bounds for PollDBCommands. The ceiling keeps a poll inside the
// gateway's request timeout; the floor stops a misbehaving agent turning the
// long poll into a busy loop.
@@ -40,7 +37,7 @@ type agentIdentity struct {
func authorizeAgent(ctx context.Context, nc *nats.Conn, caller Caller, requestedID string) (*agentIdentity, error) {
if caller.PrincipalType != principalTypeAssumedRole ||
caller.AccountID != utils.GlobalAccountID ||
caller.RoleName != InstanceRoleName ||
caller.RoleName != handlers_rds.InstanceRoleName ||
caller.SessionName == "" {
slog.DebugContext(ctx, "RDS: internal action rejected for non-agent caller",
"principalType", caller.PrincipalType, "accountID", caller.AccountID, "roleName", caller.RoleName)
+5 -5
View File
@@ -26,7 +26,7 @@ func agentCaller() Caller {
return Caller{
AccountID: utils.GlobalAccountID,
PrincipalType: principalTypeAssumedRole,
RoleName: InstanceRoleName,
RoleName: handlers_rds.InstanceRoleName,
SessionName: testInstanceID,
}
}
@@ -69,12 +69,12 @@ func TestAuthorizeAgent_RejectsNonAgentCallers(t *testing.T) {
name string
caller Caller
}{
{"plain IAM user", Caller{AccountID: utils.GlobalAccountID, PrincipalType: "user", RoleName: InstanceRoleName, SessionName: testInstanceID}},
{"root", Caller{AccountID: utils.GlobalAccountID, PrincipalType: "root", RoleName: InstanceRoleName, SessionName: testInstanceID}},
{"customer account", Caller{AccountID: testAccountID, PrincipalType: principalTypeAssumedRole, RoleName: InstanceRoleName, SessionName: testInstanceID}},
{"plain IAM user", Caller{AccountID: utils.GlobalAccountID, PrincipalType: "user", RoleName: handlers_rds.InstanceRoleName, SessionName: testInstanceID}},
{"root", Caller{AccountID: utils.GlobalAccountID, PrincipalType: "root", RoleName: handlers_rds.InstanceRoleName, SessionName: testInstanceID}},
{"customer account", Caller{AccountID: testAccountID, PrincipalType: principalTypeAssumedRole, RoleName: handlers_rds.InstanceRoleName, SessionName: testInstanceID}},
{"another role", Caller{AccountID: utils.GlobalAccountID, PrincipalType: principalTypeAssumedRole, RoleName: "ecsInstanceRole", SessionName: testInstanceID}},
{"unresolvable role", Caller{AccountID: utils.GlobalAccountID, PrincipalType: principalTypeAssumedRole, RoleName: "", SessionName: testInstanceID}},
{"no session name", Caller{AccountID: utils.GlobalAccountID, PrincipalType: principalTypeAssumedRole, RoleName: InstanceRoleName, SessionName: ""}},
{"no session name", Caller{AccountID: utils.GlobalAccountID, PrincipalType: principalTypeAssumedRole, RoleName: handlers_rds.InstanceRoleName, SessionName: ""}},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
+1 -1
View File
@@ -79,7 +79,7 @@ func rejectWith(code string) Handler {
// explicit stubs, so an unimplemented action stays distinct from an unknown one.
var actions = map[string]Handler{
// Instance lifecycle.
"CreateDBInstance": pending(),
"CreateDBInstance": typed(CreateDBInstance),
"DescribeDBInstances": typed(DescribeDBInstances),
"ModifyDBInstance": pending(),
"DeleteDBInstance": pending(),
+50 -4
View File
@@ -1,9 +1,15 @@
package gateway_rds
import (
"encoding/json"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -94,13 +100,43 @@ func TestDispatch_UnknownAction(t *testing.T) {
assert.Equal(t, awserrors.ErrorInvalidAction, err.Error())
}
// The customer actions this phase implements forward to the daemon, so they are
// dispatched against a stub responder rather than a nil connection.
var liveActions = []string{"CreateDBInstance", "DescribeDBInstances"}
// What is under test in this file is the action table and the XML envelope, not
// the orchestration behind the subject, so the responder returns a fixed output.
func newStubbedNATS(t *testing.T) *nats.Conn {
t.Helper()
_, nc, _ := testutil.StartTestJetStream(t)
respondWith(t, nc, handlers_rds.SubjectCreateDBInstance,
&rds.CreateDBInstanceOutput{DBInstance: &rds.DBInstance{DBInstanceIdentifier: aws.String("orders-db")}})
respondWith(t, nc, handlers_rds.SubjectDescribeDBInstances,
&rds.DescribeDBInstancesOutput{DBInstances: []*rds.DBInstance{}})
return nc
}
func respondWith(t *testing.T, nc *nats.Conn, subject string, output any) {
t.Helper()
payload, err := json.Marshal(output)
require.NoError(t, err)
sub, err := nc.Subscribe(subject, func(msg *nats.Msg) {
if err := msg.Respond(payload); err != nil {
t.Logf("respond on %s: %v", subject, err)
}
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
// Every registered action must dispatch to something: either a real response or
// one of the two deliberate rejections. Anything else means an action was wired
// to a handler that fails for an unrelated reason.
func TestDispatch_EveryActionResolves(t *testing.T) {
nc := newStubbedNATS(t)
for action := range actions {
t.Run(action, func(t *testing.T) {
body, err := Dispatch(t.Context(), action, map[string]string{"Action": action}, nil, testCaller)
body, err := Dispatch(t.Context(), action, map[string]string{"Action": action}, nc, testCaller)
if err == nil {
assert.NotEmpty(t, body, "a successful action must return an XML body")
return
@@ -115,8 +151,18 @@ func TestDispatch_EveryActionResolves(t *testing.T) {
}
}
// The actions this phase implements must have left the pending stub behind.
func TestDispatch_LiveActionsAreNotPending(t *testing.T) {
nc := newStubbedNATS(t)
for _, action := range liveActions {
body, err := Dispatch(t.Context(), action, map[string]string{"Action": action}, nc, testCaller)
require.NoError(t, err, "action %q", action)
assert.Contains(t, string(body), "<"+action+"Result", "action %q", action)
}
}
func TestDispatch_PendingActionIsNotImplemented(t *testing.T) {
_, err := Dispatch(t.Context(), "CreateDBInstance", map[string]string{"Action": "CreateDBInstance"}, nil, testCaller)
_, err := Dispatch(t.Context(), "ModifyDBInstance", map[string]string{"Action": "ModifyDBInstance"}, nil, testCaller)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
}
@@ -131,7 +177,7 @@ func TestDispatch_OutOfScopeActionIsNotSupported(t *testing.T) {
func TestDispatch_DescribeDBInstancesReturnsEmptyResultSet(t *testing.T) {
body, err := Dispatch(t.Context(), "DescribeDBInstances",
map[string]string{"Action": "DescribeDBInstances", "Version": "2014-10-31"}, nil, testCaller)
map[string]string{"Action": "DescribeDBInstances", "Version": "2014-10-31"}, newStubbedNATS(t), testCaller)
require.NoError(t, err)
// The IAM-style envelope the aws-sdk-go query unmarshaler expects, carrying
@@ -149,7 +195,7 @@ func TestDispatch_DescribeDBInstancesWithFilters(t *testing.T) {
"Action": "DescribeDBInstances",
"DBInstanceIdentifier": "orders-db",
"MaxRecords": "20",
}, nil, testCaller)
}, newStubbedNATS(t), testCaller)
require.NoError(t, err)
assert.Contains(t, string(body), "<DescribeDBInstancesResult")
}
+9 -4
View File
@@ -4,11 +4,16 @@ import (
"context"
"github.com/aws/aws-sdk-go/service/rds"
handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds"
"github.com/nats-io/nats.go"
)
// Until CreateDBInstance lands the result set is always empty, which is the
// honest answer for an account that has none.
func DescribeDBInstances(_ context.Context, _ *rds.DescribeDBInstancesInput, _ *nats.Conn, _ Caller) (any, error) {
return &rds.DescribeDBInstancesOutput{DBInstances: []*rds.DBInstance{}}, nil
// The whole orchestration runs on the daemon side, so this only carries the
// caller's account through — the request body never names the owner.
func CreateDBInstance(ctx context.Context, input *rds.CreateDBInstanceInput, nc *nats.Conn, caller Caller) (any, error) {
return handlers_rds.NewNATSService(nc).CreateDBInstance(ctx, input, caller.AccountID)
}
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)
}
+25
View File
@@ -115,6 +115,31 @@ func EKSChanges(action Action, dnsName, baseDomain, endpointIP string) []Change
}}
}
// RDSName is the account-qualified DNS name for a DB instance's endpoint:
// {dbInstanceIdentifier}.{accountID}.{region}.rds.{baseDomain}. DB instance
// identifiers are only unique within an account, so the account label prevents
// cross-tenant RRset collisions.
func RDSName(dbInstanceIdentifier, accountID, region, baseDomain string) string {
return fmt.Sprintf("%s.%s.%s.rds.%s", dbInstanceIdentifier, accountID, region, baseDomain)
}
// RDSChanges builds the record-set change for a DB instance's endpoint. The
// target is the customer ENI's private IP, which survives a VM replace, so the
// record does not have to be rewritten when the instance is rebuilt. Returns no
// change when the name, zone, or IP is unavailable.
func RDSChanges(action Action, dnsName, baseDomain, eniIP string) []Change {
if dnsName == "" || baseDomain == "" || eniIP == "" {
return nil
}
return []Change{{
Action: action,
Zone: baseDomain,
Name: dnsName,
Type: "A",
Value: eniIP,
}}
}
// privateZoneOrDefault returns the configured internal domain or the
// compute.internal default when unset.
func privateZoneOrDefault(internalDomain string) string {
+24
View File
@@ -75,6 +75,30 @@ func TestEKSNameAndChanges(t *testing.T) {
assert.Empty(t, EKSChanges(ActionUpsert, "my-cluster.ap-southeast-2.eks.spx3.net", "spx3.net", ""))
}
func TestRDSNameAndChanges(t *testing.T) {
assert.Equal(t, "orders-db.111122223333.ap-southeast-2.rds.spx3.net",
RDSName("orders-db", "111122223333", "ap-southeast-2", "spx3.net"))
assert.NotEqual(t,
RDSName("orders-db", "111122223333", "ap-southeast-2", "spx3.net"),
RDSName("orders-db", "444455556666", "ap-southeast-2", "spx3.net"),
"DB instance identifiers are unique per account, so the RRsets must differ")
// The target is the customer ENI's private IP: it survives a VM replace, so
// the record does not have to be rewritten when the instance is rebuilt.
changes := RDSChanges(ActionUpsert, "orders-db.111122223333.ap-southeast-2.rds.spx3.net", "spx3.net", "10.0.5.20")
require.Len(t, changes, 1)
assert.Equal(t, ActionUpsert, changes[0].Action)
assert.Equal(t, "spx3.net", changes[0].Zone)
assert.Equal(t, "A", changes[0].Type)
assert.Equal(t, "10.0.5.20", changes[0].Value)
// A deployment with no base domain, or an instance whose ENI IP is not known
// yet, yields no change rather than a record pointing nowhere.
assert.Empty(t, RDSChanges(ActionUpsert, "orders-db.111122223333.ap-southeast-2.rds.spx3.net", "", "10.0.5.20"))
assert.Empty(t, RDSChanges(ActionUpsert, "orders-db.111122223333.ap-southeast-2.rds.spx3.net", "spx3.net", ""))
assert.Empty(t, RDSChanges(ActionUpsert, "", "spx3.net", "10.0.5.20"))
}
func TestRelativeLabel(t *testing.T) {
assert.Empty(t, relativeLabel("spx3.net", "spx3.net"))
assert.Empty(t, relativeLabel("spx3.net.", "spx3.net"))
+6 -2
View File
@@ -52,6 +52,7 @@ type DesiredSet struct {
type PruneScope struct {
ELB bool
EKS bool
RDS bool
}
// Reconciler is the drift backstop. On a ticker it rebuilds the desired
@@ -264,8 +265,8 @@ func (r *Reconciler) computeBatch() ([]Change, error) {
}
// prunable returns the predicate deciding whether a (zone, label) record may be
// deleted when absent from the desired set: load-balancer and EKS records in the
// base domain, but only for the classes this cycle enumerated authoritatively
// deleted when absent from the desired set: load-balancer, EKS and RDS records
// in the base domain, but only for the classes this cycle enumerated authoritatively
// across all tenants. EC2 (`.compute.`) records are never pruned (a node sees
// only its own instances); structural (apex/NS/glue) records never match.
func (r *Reconciler) prunable(scope PruneScope) func(zone, label string) bool {
@@ -279,6 +280,9 @@ func (r *Reconciler) prunable(scope PruneScope) func(zone, label string) bool {
if scope.EKS && strings.Contains(label, ".eks.") {
return true
}
if scope.RDS && strings.Contains(label, ".rds.") {
return true
}
return false
}
}
+27
View File
@@ -98,6 +98,33 @@ func TestComputeConvergeNoPruneWhenNoAuthority(t *testing.T) {
assert.Empty(t, deletesOf(batch), "no authority means no deletions")
}
func TestComputeConvergePrunesRDSOnlyWhenEnumerated(t *testing.T) {
desired := []Change{upsert("orders-db.111122223333.ap-southeast-2.rds.spx3.net", "10.0.5.20")}
existing := map[string][]zoneRecord{testBase: {
existingA("orders-db.111122223333.ap-southeast-2.rds.", "10.0.5.20"),
existingA("dropped-db.111122223333.ap-southeast-2.rds.", "10.0.5.21"),
existingA("app-web-abc.ap-southeast-2.elb.", "1.1.1.1"),
existingA("ec2-4-4-4-4.ap-southeast-2.compute.", "4.4.4.4"),
}}
// Without RDS authority the deleted instance's record survives, and so does
// every other class this cycle could not enumerate.
batch, err := computeConverge(desired, existing, prunableFor(PruneScope{}))
require.NoError(t, err)
assert.Empty(t, deletesOf(batch), "RDS pruning is suppressed when the account buckets were not fully read")
batch, err = computeConverge(desired, existing, prunableFor(PruneScope{RDS: true}))
require.NoError(t, err)
deletes := deletesOf(batch)
require.Len(t, deletes, 1, "only the DB instance absent from the desired set is pruned")
assert.Equal(t, "dropped-db.111122223333.ap-southeast-2.rds.spx3.net", deletes[0].Name)
for _, d := range deletes {
assert.NotContains(t, d.Name, ".elb.", "RDS authority does not grant ELB pruning")
assert.NotContains(t, d.Name, ".compute.", "EC2 records are never pruned")
}
}
func TestComputeConvergeRejectsUnsupportedRecordType(t *testing.T) {
desired := []Change{{
Action: ActionUpsert,
+199
View File
@@ -0,0 +1,199 @@
package handlers_rds
import (
"context"
"errors"
"fmt"
"log/slog"
"time"
"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/utils"
"github.com/nats-io/nats.go/jetstream"
)
// The first VM behind a DB instance. rds-5b and rds-6 increment it on replace,
// so an agent report carrying an older generation is a superseded VM.
const firstVMGeneration = 1
// Assembles a live DB instance out of the launch primitives: validate, reserve
// the identifier, place and launch the dual-NIC VM with its data volume and
// customer ENI, seed the one-shot bootstrap config, and publish the endpoint
// record. The instance is returned at status=creating; the reconciler flips it
// to available on the first healthy agent heartbeat (D7).
func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInstanceInput, accountID string) (out *rds.CreateDBInstanceOutput, err error) {
req, err := validateCreateRequest(input)
if err != nil {
return nil, err
}
placement, err := s.resolvePlacement(ctx, accountID, req)
if err != nil {
return nil, err
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
key := DBInstanceKey(req.Identifier)
// Creating the record before anything is provisioned makes the identifier
// uniqueness check atomic against a concurrent create on another node — a
// prior read-then-write would let both pass and both launch a VM.
rec := newDBInstanceRecord(accountID, req, placement)
if createErr := createJSON(ctx, kv, key, &rec); createErr != nil {
if errors.Is(createErr, jetstream.ErrKeyExists) {
return nil, fmt.Errorf("%s: DB instance %s already exists",
awserrors.ErrorDBInstanceAlreadyExists, req.Identifier)
}
return nil, createErr
}
// The record is the reservation, so it is withdrawn on any failure below:
// leaving it behind would make the identifier permanently unusable while
// naming an instance that was never provisioned.
defer func() {
if err == nil {
return
}
rbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), rollbackTimeout)
defer cancel()
if delErr := kv.Delete(rbCtx, key); delErr != nil {
slog.WarnContext(rbCtx, "rds: rollback delete of reserved DB instance record failed",
"dbInstance", req.Identifier, "err", delErr)
}
}()
launched, err := LaunchDBInstanceVM(ctx, s.deps.Launch, LaunchInput{
DBInstanceIdentifier: req.Identifier,
AccountID: accountID,
SubnetID: placement.SubnetID,
SecurityGroupIDs: placement.SecurityGroupIDs,
Engine: req.Engine.Name,
EngineVersion: req.EngineVersion,
InstanceType: req.InstanceType,
AllocatedStorage: req.AllocatedStorage,
RequireEncryptedData: true,
UserData: buildAgentUserData(agentUserDataInput{
GatewayURL: s.deps.GatewayURL,
GatewayCACert: s.deps.GatewayCACert,
Region: s.region,
DBInstanceIdentifier: req.Identifier,
EngineVersion: req.EngineVersion,
EnginePort: req.Port,
}),
// The system account owns the VM, so the role is ensured there too — the
// gateway's agent gate requires an assumed-role session in that account.
IamInstanceProfileArn: ensureInstanceProfile(s.deps.IAM, utils.GlobalAccountID),
})
if err != nil {
return nil, err
}
// The launch already unwound everything on failure, so from here the resources
// exist and the record has to catch up with them.
stored, err := s.recordLaunch(ctx, kv, key, accountID, launched)
if err != nil {
s.unwindLaunched(ctx, launched)
return nil, err
}
// Written before the DNS publish because an agent that boots fast enough to
// call the gateway before this exists is denied and has to retry.
if err = s.PutInstanceIndex(ctx, launched.InstanceID, InstanceIndexEntry{
AccountID: accountID,
DBInstanceIdentifier: req.Identifier,
VMGeneration: firstVMGeneration,
}); err != nil {
s.unwindLaunched(ctx, launched)
return nil, fmt.Errorf("rds: write instance index for %s: %w", req.Identifier, err)
}
// Published as soon as the ENI IP is known rather than on available, so the
// name resolves the moment the engine comes up; the security group is what
// gates connectivity until then.
s.publishDNS(ctx, accountID, stored, handlers_dns.ActionUpsert)
slog.InfoContext(ctx, "rds: DB instance created",
"dbInstance", req.Identifier, "accountId", accountID, "instanceId", launched.InstanceID,
"endpoint", stored.EndpointAddress, "class", req.InstanceClass)
return &rds.CreateDBInstanceOutput{DBInstance: s.projectDBInstance(stored)}, nil
}
// The record as it stands before any resource exists: everything the request
// determined, plus the staged master password the first bootstrap fetch consumes.
func newDBInstanceRecord(accountID string, req *validatedCreate, placement *endpointPlacement) DBInstanceRecord {
now := time.Now().UTC()
return DBInstanceRecord{
DBInstanceIdentifier: req.Identifier,
AccountID: accountID,
Status: StatusCreating,
Engine: req.Engine.Name,
EngineVersion: req.EngineVersion,
DBInstanceClass: req.InstanceClass,
AllocatedStorage: req.AllocatedStorage,
StorageType: req.StorageType,
DBName: req.DBName,
MasterUsername: req.MasterUsername,
Port: req.Port,
SubnetID: placement.SubnetID,
VpcID: placement.VpcID,
VpcSecurityGroupIDs: placement.SecurityGroupIDs,
DBParameterGroupName: req.DBParameterGroupName,
Bootstrap: BootstrapState{MasterUserPassword: req.MasterPassword},
CreatedAt: now,
UpdatedAt: now,
}
}
// Folds the launch results into the reserved record and returns it as written.
// The endpoint is settled here because the ENI IP — and so the vanity name — is
// only known now.
func (s *Service) recordLaunch(ctx context.Context, kv jetstream.KeyValue, key, accountID string, launched *LaunchOutput) (*DBInstanceRecord, error) {
var rec DBInstanceRecord
rev, found, err := getJSONRevision(ctx, kv, key, &rec)
if err != nil {
return nil, err
}
if !found {
return nil, errors.New(awserrors.ErrorDBInstanceNotFound)
}
rec.InstanceID = launched.InstanceID
rec.VMGeneration = firstVMGeneration
rec.SystemENIID = launched.SystemENIID
rec.ENIID = launched.CustomerENIID
rec.ENIPrivateIP = launched.CustomerENIIP
rec.DataVolumeID = launched.DataVolumeID
rec.StorageEncrypted = launched.DataVolumeEncrypted
rec.DNSName = s.dnsName(accountID, rec.DBInstanceIdentifier)
// Without northstar there is no resolvable name, so the endpoint is the ENI
// IP itself — stable per D5 and therefore as durable as a hostname (D6).
rec.EndpointAddress = rec.DNSName
if rec.EndpointAddress == "" {
rec.EndpointAddress = rec.ENIPrivateIP
}
rec.UpdatedAt = time.Now().UTC()
if err := updateJSON(ctx, kv, key, rev, &rec); err != nil {
return nil, err
}
return &rec, nil
}
// The launch's own rollback only covers failures inside it. A failure after it
// returned has to run the same unwind: the deferred record delete removes the
// only thing naming the VM, its two ENIs and the data volume, and nothing in
// this phase reclaims them afterwards.
func (s *Service) unwindLaunched(ctx context.Context, launched *LaunchOutput) {
if launched == nil || launched.Unwind == nil {
return
}
slog.WarnContext(ctx, "rds: unwinding a DB instance that failed after launch",
"instanceId", launched.InstanceID, "dataVolumeId", launched.DataVolumeID,
"customerEniId", launched.CustomerENIID, "systemEniId", launched.SystemENIID)
launched.Unwind(ctx)
}
+484
View File
@@ -0,0 +1,484 @@
package handlers_rds
import (
"context"
"encoding/json"
"errors"
"strings"
"testing"
"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/testutil"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
testDefaultVPC = "vpc-default01"
testDefaultSG = "sg-default01"
testBaseDomain = "spx3.net"
testDBInstanceID = "orders-db"
)
// fakeNetwork is the customer account's VPC as the placement resolver reads it:
// one default VPC with two subnets and a default security group.
type fakeNetwork struct {
vpcs []*ec2.Vpc
subnets []*ec2.Subnet
groups []*ec2.SecurityGroup
vpcErr error
subErr error
groupErr error
// accts records which account each describe was issued against, which is how
// the cross-account read becomes observable.
accts []string
}
var _ networkResolver = (*fakeNetwork)(nil)
func newFakeNetwork() *fakeNetwork {
return &fakeNetwork{
vpcs: []*ec2.Vpc{{VpcId: aws.String(testDefaultVPC), IsDefault: aws.Bool(true)}},
// Deliberately out of order: the resolver sorts, so the placement is the
// same on every repeat regardless of how the describe happened to order.
subnets: []*ec2.Subnet{
{SubnetId: aws.String("subnet-zebra")},
{SubnetId: aws.String("subnet-alpha")},
},
groups: []*ec2.SecurityGroup{
{GroupId: aws.String("sg-app01"), GroupName: aws.String("app")},
{GroupId: aws.String(testDefaultSG), GroupName: aws.String("default")},
},
}
}
func (f *fakeNetwork) DescribeVpcs(_ context.Context, _ *ec2.DescribeVpcsInput, accountID string) (*ec2.DescribeVpcsOutput, error) {
f.accts = append(f.accts, accountID)
if f.vpcErr != nil {
return nil, f.vpcErr
}
return &ec2.DescribeVpcsOutput{Vpcs: f.vpcs}, nil
}
func (f *fakeNetwork) DescribeSubnets(_ context.Context, _ *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error) {
f.accts = append(f.accts, accountID)
if f.subErr != nil {
return nil, f.subErr
}
return &ec2.DescribeSubnetsOutput{Subnets: f.subnets}, nil
}
func (f *fakeNetwork) DescribeSecurityGroups(_ context.Context, _ *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) {
f.accts = append(f.accts, accountID)
if f.groupErr != nil {
return nil, f.groupErr
}
return &ec2.DescribeSecurityGroupsOutput{SecurityGroups: f.groups}, nil
}
// createHarness is a Service wired to the launch fakes plus a customer VPC, so a
// create runs end to end against KV without touching EC2 or northstar.
type createHarness struct {
svc *Service
launch *launchHarness
network *fakeNetwork
nc *nats.Conn
// dnsChanges collects what the endpoint publish put on the bus.
dnsChanges chan handlers_dns.ChangeBatch
}
func newCreateHarness(t *testing.T, baseDomain string) *createHarness {
t.Helper()
_, nc, _ := testutil.StartTestJetStream(t)
h := &createHarness{
launch: newLaunchHarness(),
network: newFakeNetwork(),
nc: nc,
dnsChanges: make(chan handlers_dns.ChangeBatch, 4),
}
// The storage key is configured on a real cluster, so the volume comes back
// encrypted; the unencrypted case is its own test.
h.launch.volumes.encrypted = true
// The DNS writer is a request-reply consumer, so without a responder the
// best-effort publish would sit out its own timeout on every create.
h.stubDNSWriter(t)
h.svc = NewService(nc, testRegion).WithDeps(Deps{
LoadCA: newTestCA(t),
Launch: h.launch.deps(),
Network: h.network,
BaseDomain: baseDomain,
})
return h
}
func (h *createHarness) stubDNSWriter(t *testing.T) {
t.Helper()
sub, err := h.nc.Subscribe(handlers_dns.SubjectRecordsetChange, func(msg *nats.Msg) {
var batch handlers_dns.ChangeBatch
if err := json.Unmarshal(msg.Data, &batch); err == nil {
h.dnsChanges <- batch
}
if err := msg.Respond([]byte(`{}`)); err != nil {
t.Logf("respond on %s: %v", handlers_dns.SubjectRecordsetChange, err)
}
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
}
func (h *createHarness) record(t *testing.T, id string) 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(id), &rec)
require.NoError(t, err)
require.True(t, found, "DB instance %s should be stored", id)
return rec
}
func (h *createHarness) 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 validCreateInput() *rds.CreateDBInstanceInput {
return &rds.CreateDBInstanceInput{
DBInstanceIdentifier: aws.String(testDBInstanceID),
Engine: aws.String("postgres"),
DBInstanceClass: aws.String("db.t3.medium"),
AllocatedStorage: aws.Int64(20),
MasterUsername: aws.String("appuser"),
MasterUserPassword: aws.String("Sup3rSecret!"),
DBName: aws.String("orders"),
}
}
func TestCreateDBInstance_ProvisionsAndRecordsTheInstance(t *testing.T) {
h := newCreateHarness(t, testBaseDomain)
out, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.NoError(t, err)
// The customer sees a creating instance immediately: the engine is not up
// yet, and the reconciler owns the flip to available.
require.NotNil(t, out.DBInstance)
assert.Equal(t, string(StatusCreating), aws.StringValue(out.DBInstance.DBInstanceStatus))
assert.Equal(t, "postgres", aws.StringValue(out.DBInstance.Engine))
assert.Equal(t, DBInstanceARN(testRegion, testAccountID, testDBInstanceID),
aws.StringValue(out.DBInstance.DBInstanceArn))
require.NotNil(t, out.DBInstance.Endpoint)
assert.Equal(t, testDBInstanceID+"."+testAccountID+"."+testRegion+".rds."+testBaseDomain,
aws.StringValue(out.DBInstance.Endpoint.Address))
assert.Equal(t, int64(5432), aws.Int64Value(out.DBInstance.Endpoint.Port),
"the engine's default port applies when the request names none")
rec := h.record(t, testDBInstanceID)
assert.Equal(t, StatusCreating, rec.Status)
assert.Equal(t, "i-rds0001", rec.InstanceID)
assert.Equal(t, int64(firstVMGeneration), rec.VMGeneration)
assert.NotEmpty(t, rec.SystemENIID)
assert.NotEmpty(t, rec.ENIID)
assert.NotEqual(t, rec.SystemENIID, rec.ENIID, "the system and customer NICs are distinct")
assert.Equal(t, "vol-rdsdata01", rec.DataVolumeID)
assert.True(t, rec.StorageEncrypted)
// The master password is staged for the one-shot bootstrap fetch, unconsumed.
assert.Equal(t, "Sup3rSecret!", rec.Bootstrap.MasterUserPassword)
assert.False(t, rec.Bootstrap.Consumed)
// The placement is the account's default VPC and its lowest-sorted subnet,
// with the VPC's default security group when the request names none.
assert.Equal(t, testDefaultVPC, rec.VpcID)
assert.Equal(t, "subnet-alpha", rec.SubnetID)
assert.Equal(t, []string{testDefaultSG}, rec.VpcSecurityGroupIDs)
for _, acct := range h.network.accts {
assert.Equal(t, testAccountID, acct, "placement is read in the customer's account")
}
// The instance index is what lets an agent's gateway call resolve its own
// DB instance, so it must exist before the VM can finish booting.
entry, err := h.svc.LookupInstanceIndex(t.Context(), "i-rds0001")
require.NoError(t, err)
require.NotNil(t, entry)
assert.Equal(t, testAccountID, entry.AccountID)
assert.Equal(t, testDBInstanceID, entry.DBInstanceIdentifier)
assert.Equal(t, int64(firstVMGeneration), entry.VMGeneration)
}
func TestCreateDBInstance_PublishesEndpointRecord(t *testing.T) {
h := newCreateHarness(t, testBaseDomain)
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.NoError(t, err)
batch := <-h.dnsChanges
require.Len(t, batch.Changes, 1)
change := batch.Changes[0]
assert.Equal(t, handlers_dns.ActionUpsert, change.Action)
assert.Equal(t, testBaseDomain, change.Zone)
assert.Equal(t, "A", change.Type)
// The record targets the customer ENI's IP, which survives a VM replace.
rec := h.record(t, testDBInstanceID)
assert.Equal(t, rec.ENIPrivateIP, change.Value)
assert.Equal(t, rec.DNSName, change.Name)
}
// Without northstar there is no name to resolve, so the endpoint has to be the
// ENI IP itself rather than an empty host the client would dial.
func TestCreateDBInstance_EndpointFallsBackToENIIPWithoutBaseDomain(t *testing.T) {
h := newCreateHarness(t, "")
out, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.NoError(t, err)
rec := h.record(t, testDBInstanceID)
assert.Empty(t, rec.DNSName)
assert.NotEmpty(t, rec.ENIPrivateIP)
assert.Equal(t, rec.ENIPrivateIP, rec.EndpointAddress)
require.NotNil(t, out.DBInstance.Endpoint)
assert.Equal(t, rec.ENIPrivateIP, aws.StringValue(out.DBInstance.Endpoint.Address))
select {
case batch := <-h.dnsChanges:
t.Fatalf("no base domain means no record to publish, got %+v", batch)
default:
}
}
func TestCreateDBInstance_DuplicateIdentifierIsRejected(t *testing.T) {
h := newCreateHarness(t, "")
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.NoError(t, err)
_, err = h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceAlreadyExists)
// The loser must not have launched anything: the record reservation is what
// makes the uniqueness check atomic, so it precedes every resource.
assert.Empty(t, h.launch.launcher.terminated)
assert.Equal(t, "i-rds0001", h.record(t, testDBInstanceID).InstanceID)
}
// The reserved record is the identifier's reservation, so a failure after it is
// written must withdraw it — otherwise the name is permanently unusable while
// naming an instance that was never provisioned.
func TestCreateDBInstance_LaunchFailureWithdrawsTheReservation(t *testing.T) {
h := newCreateHarness(t, "")
h.launch.launcher.err = errors.New("no host has capacity")
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
assert.False(t, h.recordExists(t, testDBInstanceID), "the reserved record must be withdrawn")
// The same identifier is immediately usable again.
h.launch.launcher.err = nil
_, err = h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.NoError(t, err)
}
// A failure after the launch returned has to tear down everything the launch
// built. The deferred record delete removes the only thing naming the VM, its
// two ENIs and the data volume, so anything left behind is unreclaimable.
func TestCreateDBInstance_FailureAfterLaunchUnwindsEveryResource(t *testing.T) {
h := newCreateHarness(t, "")
// Dropping the reserved record while the launch is in flight fails the
// record-launch step, which is the first thing to run after the resources exist.
h.launch.launcher.onLaunch = func() {
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, kv.Delete(t.Context(), DBInstanceKey(testDBInstanceID)))
}
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
// The VM goes first: the ENIs and volume are in-use while it runs.
assert.Equal(t, []string{"terminate", "delete-volume", "delete-eni", "delete-eni"}, h.launch.unwind)
assert.Contains(t, h.launch.launcher.terminated, "i-rds0001")
assert.Contains(t, h.launch.volumes.deleted, "vol-rdsdata01")
assert.Len(t, h.launch.enis.deleted, 2, "both the customer and system ENI are deleted")
}
// D19 honesty: the request may not be answered with an instance whose storage is
// not actually encrypted, so an unencrypted volume fails the whole create.
func TestCreateDBInstance_UnencryptedDataVolumeFailsTheCreate(t *testing.T) {
h := newCreateHarness(t, "")
h.launch.volumes.encrypted = false
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
assert.False(t, h.recordExists(t, testDBInstanceID))
assert.Contains(t, h.launch.volumes.deleted, "vol-rdsdata01", "the unencrypted volume is torn down")
}
func TestCreateDBInstance_SecurityGroupsMustBeInThePlacementVPC(t *testing.T) {
h := newCreateHarness(t, "")
input := validCreateInput()
input.VpcSecurityGroupIds = aws.StringSlice([]string{"sg-app01"})
_, err := h.svc.CreateDBInstance(t.Context(), input, testAccountID)
require.NoError(t, err)
assert.Equal(t, []string{"sg-app01"}, h.record(t, testDBInstanceID).VpcSecurityGroupIDs)
// An ENI cannot carry a group from another VPC, so this is caught before the
// launch rather than failing after the record exists.
input = validCreateInput()
input.DBInstanceIdentifier = aws.String("other-db")
input.VpcSecurityGroupIds = aws.StringSlice([]string{"sg-elsewhere"})
_, err = h.svc.CreateDBInstance(t.Context(), input, testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
assert.False(t, h.recordExists(t, "other-db"))
}
func TestCreateDBInstance_RequiresADefaultVPCWithASubnet(t *testing.T) {
h := newCreateHarness(t, "")
h.network.vpcs = nil
_, err := h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorDBInvalidVPCNetworkState)
h.network.vpcs = newFakeNetwork().vpcs
h.network.subnets = nil
_, err = h.svc.CreateDBInstance(t.Context(), validCreateInput(), testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorDBInvalidVPCNetworkState)
// A placement failure happens before the reservation, so nothing is left over.
assert.False(t, h.recordExists(t, testDBInstanceID))
}
// D19: a parameter that would create a false safety, security or availability
// guarantee is rejected rather than silently dropped.
func TestValidateCreateRequest_RejectsUnimplementedParameters(t *testing.T) {
cases := []struct {
name string
mutate func(*rds.CreateDBInstanceInput)
wantErr string
}{
{"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"},
{"IAMDatabaseAuth", func(in *rds.CreateDBInstanceInput) { in.EnableIAMDatabaseAuthentication = aws.Bool(true) }, "EnableIAMDatabaseAuthentication"},
{"Iops", func(in *rds.CreateDBInstanceInput) { in.Iops = aws.Int64(3000) }, "Iops"},
{"KmsKeyId", func(in *rds.CreateDBInstanceInput) { in.KmsKeyId = aws.String("arn:aws:kms:::key/abc") }, "KmsKeyId"},
{"AvailabilityZone", func(in *rds.CreateDBInstanceInput) { in.AvailabilityZone = aws.String("ap-southeast-2b") }, "AvailabilityZone"},
{"DBSecurityGroups", func(in *rds.CreateDBInstanceInput) { in.DBSecurityGroups = aws.StringSlice([]string{"classic"}) }, "DBSecurityGroups"},
{"DBClusterIdentifier", func(in *rds.CreateDBInstanceInput) { in.DBClusterIdentifier = aws.String("cluster-1") }, "DBClusterIdentifier"},
{"CloudwatchLogsExports", func(in *rds.CreateDBInstanceInput) {
in.EnableCloudwatchLogsExports = aws.StringSlice([]string{"postgresql"})
}, "EnableCloudwatchLogsExports"},
{"Tags", func(in *rds.CreateDBInstanceInput) {
in.Tags = []*rds.Tag{{Key: aws.String("env"), Value: aws.String("prod")}}
}, "Tags"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
input := validCreateInput()
tc.mutate(input)
_, err := validateCreateRequest(input)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
assert.Contains(t, err.Error(), tc.wantErr)
})
}
}
func TestValidateCreateRequest_RejectsMalformedRequests(t *testing.T) {
cases := []struct {
name string
mutate func(*rds.CreateDBInstanceInput)
wantError string
}{
{"NoIdentifier", func(in *rds.CreateDBInstanceInput) { in.DBInstanceIdentifier = nil }, awserrors.ErrorInvalidParameterValue},
{"IdentifierUppercase", func(in *rds.CreateDBInstanceInput) { in.DBInstanceIdentifier = aws.String("OrdersDB") }, awserrors.ErrorInvalidParameterValue},
{"IdentifierLeadingDigit", func(in *rds.CreateDBInstanceInput) { in.DBInstanceIdentifier = aws.String("1db") }, awserrors.ErrorInvalidParameterValue},
{"IdentifierTrailingHyphen", func(in *rds.CreateDBInstanceInput) { in.DBInstanceIdentifier = aws.String("db-") }, awserrors.ErrorInvalidParameterValue},
{"IdentifierDoubleHyphen", func(in *rds.CreateDBInstanceInput) { in.DBInstanceIdentifier = aws.String("or--ders") }, awserrors.ErrorInvalidParameterValue},
{"IdentifierTooLong", func(in *rds.CreateDBInstanceInput) {
in.DBInstanceIdentifier = aws.String(strings.Repeat("a", maxDBInstanceIdentifierLen+1))
}, awserrors.ErrorInvalidParameterValue},
{"UnknownEngine", func(in *rds.CreateDBInstanceInput) { in.Engine = aws.String("mysql") }, awserrors.ErrorInvalidParameterValue},
{"UnknownClass", func(in *rds.CreateDBInstanceInput) { in.DBInstanceClass = aws.String("db.x99.mega") }, awserrors.ErrorInvalidParameterValue},
{"StorageBelowMinimum", func(in *rds.CreateDBInstanceInput) {
in.AllocatedStorage = aws.Int64(minAllocatedStorageGiB - 1)
}, awserrors.ErrorInvalidParameterValue},
{"StorageAboveMaximum", func(in *rds.CreateDBInstanceInput) {
in.AllocatedStorage = aws.Int64(maxAllocatedStorageGiB + 1)
}, awserrors.ErrorInvalidParameterValue},
{"StorageType", func(in *rds.CreateDBInstanceInput) { in.StorageType = aws.String("io2") }, awserrors.ErrorInvalidParameterValue},
{"PortTooLow", func(in *rds.CreateDBInstanceInput) { in.Port = aws.Int64(80) }, awserrors.ErrorInvalidParameterValue},
{"ReservedUsername", func(in *rds.CreateDBInstanceInput) { in.MasterUsername = aws.String("rdsadmin") }, awserrors.ErrorInvalidParameterValue},
{"ShortPassword", func(in *rds.CreateDBInstanceInput) { in.MasterUserPassword = aws.String("short") }, awserrors.ErrorInvalidParameterValue},
// Named groups do not exist until DB subnet groups land, and placing the
// endpoint somewhere else would put it in a subnet nobody chose.
{"NamedSubnetGroup", func(in *rds.CreateDBInstanceInput) {
in.DBSubnetGroupName = aws.String("db-private")
}, awserrors.ErrorDBSubnetGroupNotFound},
{"NamedParameterGroup", func(in *rds.CreateDBInstanceInput) {
in.DBParameterGroupName = aws.String("tuned-pg")
}, awserrors.ErrorDBParameterGroupNotFound},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
input := validCreateInput()
tc.mutate(input)
_, err := validateCreateRequest(input)
require.Error(t, err)
assert.Contains(t, err.Error(), tc.wantError)
})
}
}
func TestValidateCreateRequest_AcceptsTheImplicitDefaultParameterGroup(t *testing.T) {
input := validCreateInput()
input.DBParameterGroupName = aws.String("default.postgres18")
input.Port = aws.Int64(6543)
req, err := validateCreateRequest(input)
require.NoError(t, err)
assert.Equal(t, "default.postgres18", req.DBParameterGroupName)
assert.Equal(t, int64(6543), req.Port)
assert.Equal(t, storageTypeGP3, req.StorageType, "gp3 is the default when the request names no storage type")
assert.Equal(t, "t3.medium", req.InstanceType)
}
// The inert parameters are accepted as no-ops rather than rejected: omitting
// them creates no false guarantee, so a Terraform config carrying them works.
func TestValidateCreateRequest_AcceptsInertParameters(t *testing.T) {
input := validCreateInput()
input.AutoMinorVersionUpgrade = aws.Bool(true)
input.CopyTagsToSnapshot = aws.Bool(true)
input.EnablePerformanceInsights = aws.Bool(true)
input.MonitoringInterval = aws.Int64(60)
input.StorageEncrypted = aws.Bool(true)
_, err := validateCreateRequest(input)
require.NoError(t, err)
}
+111
View File
@@ -0,0 +1,111 @@
package handlers_rds
import (
"context"
"errors"
"slices"
"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"
)
// Named DB instances that do not exist are an error, matching AWS: a client
// polling a create would otherwise read an empty list as "gone" rather than
// "not ready".
func (s *Service) DescribeDBInstances(ctx context.Context, input *rds.DescribeDBInstancesInput, accountID string) (*rds.DescribeDBInstancesOutput, error) {
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
if input != nil {
if id := aws.StringValue(input.DBInstanceIdentifier); id != "" {
rec, _, err := s.getDBInstance(ctx, kv, id)
if err != nil {
return nil, err
}
return &rds.DescribeDBInstancesOutput{DBInstances: []*rds.DBInstance{s.projectDBInstance(rec)}}, nil
}
}
ids, err := ListDBInstanceIDs(ctx, kv)
if err != nil {
return nil, err
}
slices.Sort(ids)
instances := make([]*rds.DBInstance, 0, len(ids))
for _, id := range ids {
var rec DBInstanceRecord
found, err := getJSON(ctx, kv, DBInstanceKey(id), &rec)
if err != nil {
return nil, err
}
// A record deleted between the key listing and this read is simply gone,
// which is the same answer a describe one tick later would give.
if !found {
continue
}
instances = append(instances, s.projectDBInstance(&rec))
}
return &rds.DescribeDBInstancesOutput{DBInstances: instances}, nil
}
// Returns the record plus its revision, for callers that follow with a CAS.
func (s *Service) getDBInstance(ctx context.Context, kv jetstream.KeyValue, id string) (*DBInstanceRecord, uint64, error) {
var rec DBInstanceRecord
rev, found, err := getJSONRevision(ctx, kv, DBInstanceKey(id), &rec)
if err != nil {
return nil, 0, err
}
if !found {
return nil, 0, errors.New(awserrors.ErrorDBInstanceNotFound)
}
return &rec, rev, nil
}
// The customer-facing view. Only fields this phase actually backs are set —
// an unset field is honestly absent rather than a fabricated default.
func (s *Service) projectDBInstance(rec *DBInstanceRecord) *rds.DBInstance {
if rec == nil {
return nil
}
out := &rds.DBInstance{
DBInstanceIdentifier: aws.String(rec.DBInstanceIdentifier),
DBInstanceArn: aws.String(DBInstanceARN(s.region, rec.AccountID, rec.DBInstanceIdentifier)),
DBInstanceStatus: aws.String(string(rec.Status)),
Engine: aws.String(rec.Engine),
EngineVersion: aws.String(rec.EngineVersion),
DBInstanceClass: aws.String(rec.DBInstanceClass),
AllocatedStorage: aws.Int64(rec.AllocatedStorage),
StorageType: aws.String(rec.StorageType),
StorageEncrypted: aws.Bool(rec.StorageEncrypted),
MasterUsername: aws.String(rec.MasterUsername),
MultiAZ: aws.Bool(false),
PubliclyAccessible: aws.Bool(false),
InstanceCreateTime: aws.Time(rec.CreatedAt),
}
if rec.DBName != "" {
out.DBName = aws.String(rec.DBName)
}
if rec.VpcID != "" {
out.DBSubnetGroup = &rds.DBSubnetGroup{VpcId: aws.String(rec.VpcID)}
}
for _, groupID := range rec.VpcSecurityGroupIDs {
out.VpcSecurityGroups = append(out.VpcSecurityGroups, &rds.VpcSecurityGroupMembership{
VpcSecurityGroupId: aws.String(groupID),
Status: aws.String("active"),
})
}
// Absent until the ENI exists: an Endpoint with no address would have a
// client dial an empty host rather than wait for the instance to come up.
if rec.EndpointAddress != "" {
out.Endpoint = &rds.Endpoint{
Address: aws.String(rec.EndpointAddress),
Port: aws.Int64(rec.Port),
}
}
return out
}
+164
View File
@@ -0,0 +1,164 @@
package handlers_rds
import (
"testing"
"github.com/aws/aws-sdk-go/aws"
"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/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// seedCreated runs a create for id and returns the stored record, so the
// describe is asserted against what a real create actually leaves behind.
func seedCreated(t *testing.T, h *createHarness, id string) DBInstanceRecord {
t.Helper()
input := validCreateInput()
input.DBInstanceIdentifier = aws.String(id)
_, err := h.svc.CreateDBInstance(t.Context(), input, testAccountID)
require.NoError(t, err)
return h.record(t, id)
}
func TestDescribeDBInstances_ListsEveryInstanceSorted(t *testing.T) {
h := newCreateHarness(t, "")
for _, id := range []string{"zulu-db", "alpha-db", "mike-db"} {
seedCreated(t, h, id)
}
out, err := h.svc.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{}, testAccountID)
require.NoError(t, err)
got := make([]string, 0, len(out.DBInstances))
for _, instance := range out.DBInstances {
got = append(got, aws.StringValue(instance.DBInstanceIdentifier))
}
assert.Equal(t, []string{"alpha-db", "mike-db", "zulu-db"}, got,
"an unfiltered describe is ordered so paging a fleet is stable")
}
func TestDescribeDBInstances_EmptyAccountIsAnEmptyList(t *testing.T) {
h := newCreateHarness(t, "")
out, err := h.svc.DescribeDBInstances(t.Context(), nil, testAccountID)
require.NoError(t, err)
assert.Empty(t, out.DBInstances)
}
// A client polling a create must be able to tell "not ready" from "gone", so a
// named instance that does not exist is an error rather than an empty list.
func TestDescribeDBInstances_NamedMissingInstanceIsAnError(t *testing.T) {
h := newCreateHarness(t, "")
seedCreated(t, h, testDBInstanceID)
_, err := h.svc.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{
DBInstanceIdentifier: aws.String("no-such-db"),
}, testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorDBInstanceNotFound)
}
func TestDescribeDBInstances_NamedInstanceProjectsTheRecord(t *testing.T) {
h := newCreateHarness(t, testBaseDomain)
rec := seedCreated(t, h, testDBInstanceID)
out, err := h.svc.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{
DBInstanceIdentifier: aws.String(testDBInstanceID),
}, testAccountID)
require.NoError(t, err)
require.Len(t, out.DBInstances, 1)
instance := out.DBInstances[0]
assert.Equal(t, testDBInstanceID, aws.StringValue(instance.DBInstanceIdentifier))
assert.Equal(t, "orders", aws.StringValue(instance.DBName))
assert.Equal(t, "appuser", aws.StringValue(instance.MasterUsername))
assert.Equal(t, int64(20), aws.Int64Value(instance.AllocatedStorage))
assert.Equal(t, storageTypeGP3, aws.StringValue(instance.StorageType))
assert.True(t, aws.BoolValue(instance.StorageEncrypted))
assert.Equal(t, rec.EndpointAddress, aws.StringValue(instance.Endpoint.Address))
// The two guarantees this platform does not make are reported as false rather
// than left unset, because an absent field reads as "unknown" to a client.
assert.False(t, aws.BoolValue(instance.MultiAZ))
assert.False(t, aws.BoolValue(instance.PubliclyAccessible))
require.NotNil(t, instance.DBSubnetGroup)
assert.Equal(t, testDefaultVPC, aws.StringValue(instance.DBSubnetGroup.VpcId))
require.Len(t, instance.VpcSecurityGroups, 1)
assert.Equal(t, testDefaultSG, aws.StringValue(instance.VpcSecurityGroups[0].VpcSecurityGroupId))
}
// Instances live in per-account buckets, so one account's describe must not see
// another's — the identifier is only unique within an account.
func TestDescribeDBInstances_IsScopedToTheCallingAccount(t *testing.T) {
h := newCreateHarness(t, "")
seedCreated(t, h, testDBInstanceID)
out, err := h.svc.DescribeDBInstances(t.Context(), &rds.DescribeDBInstancesInput{}, "999988887777")
require.NoError(t, err)
assert.Empty(t, out.DBInstances)
}
// A DB instance with no endpoint yet has no Endpoint at all: an Endpoint with an
// empty address would have a client dial an empty host.
func TestProjectDBInstance_OmitsAnUnsettledEndpoint(t *testing.T) {
h := newCreateHarness(t, "")
instance := h.svc.projectDBInstance(&DBInstanceRecord{
DBInstanceIdentifier: testDBInstanceID,
AccountID: testAccountID,
Status: StatusCreating,
Port: 5432,
})
assert.Nil(t, instance.Endpoint)
assert.Nil(t, instance.DBName, "a request that named no database has none, not an empty one")
assert.Nil(t, h.svc.projectDBInstance(nil))
}
func TestDesiredDNSChanges_CoversEveryTenantOrClaimsNoAuthority(t *testing.T) {
h := newCreateHarness(t, testBaseDomain)
rec := seedCreated(t, h, testDBInstanceID)
changes, ok := h.svc.DesiredDNSChanges()
require.True(t, ok)
require.Len(t, changes, 1)
assert.Equal(t, handlers_dns.ActionUpsert, changes[0].Action)
assert.Equal(t, rec.DNSName, changes[0].Name)
assert.Equal(t, rec.ENIPrivateIP, changes[0].Value)
}
// Without a base domain there are no managed RDS records, and claiming authority
// would let the reconcile prune records this node cannot even name.
func TestDesiredDNSChanges_ClaimsNoAuthorityWithoutABaseDomain(t *testing.T) {
h := newCreateHarness(t, "")
seedCreated(t, h, testDBInstanceID)
changes, ok := h.svc.DesiredDNSChanges()
assert.False(t, ok)
assert.Empty(t, changes)
}
// A deleted instance contributes nothing, so the reconcile prunes its record;
// anything still holding an ENI IP keeps its name resolvable, including a failed
// instance an operator is still investigating.
func TestDesiredDNSChanges_SkipsDeletedButKeepsFailed(t *testing.T) {
h := newCreateHarness(t, testBaseDomain)
seedCreated(t, h, "gone-db")
failed := seedCreated(t, h, "failed-db")
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
gone := h.record(t, "gone-db")
gone.Status = StatusDeleted
require.NoError(t, putJSON(t.Context(), kv, DBInstanceKey("gone-db"), &gone))
failed.Status = StatusFailed
require.NoError(t, putJSON(t.Context(), kv, DBInstanceKey("failed-db"), &failed))
changes, ok := h.svc.DesiredDNSChanges()
require.True(t, ok)
require.Len(t, changes, 1)
assert.Equal(t, failed.DNSName, changes[0].Name)
}
+89
View File
@@ -0,0 +1,89 @@
package handlers_rds
import (
"context"
"log/slog"
handlers_dns "github.com/mulgadc/spinifex/spinifex/handlers/dns"
"github.com/nats-io/nats.go/jetstream"
)
// The vanity hostname for a DB instance, or "" on a deployment with no base
// domain — where the endpoint is the bare ENI IP instead (D6).
func (s *Service) dnsName(accountID, dbInstanceIdentifier string) string {
if s.baseDomain == "" {
return ""
}
return handlers_dns.RDSName(dbInstanceIdentifier, accountID, s.region, s.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 == "" {
return
}
changes := handlers_dns.RDSChanges(action, rec.DNSName, s.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)
}
// The UPSERTs for every endpoint-ready DB instance across all account buckets,
// plus whether the enumeration was authoritative. Instances live in per-account
// buckets, so a complete cross-tenant view requires reading every one: any
// 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 == "" {
return nil, false
}
ctx := context.Background()
js, err := s.js()
if err != nil {
return nil, false
}
buckets, err := AccountBucketNames(ctx, js)
if err != nil {
slog.Warn("rds DesiredDNSChanges: enumerate account buckets", "err", err)
return nil, false
}
for _, bucket := range buckets {
kv, err := js.KeyValue(ctx, bucket)
if err != nil {
slog.Warn("rds DesiredDNSChanges: open account bucket", "bucket", bucket, "err", err)
return nil, false
}
bucketChanges, err := desiredBucketDNSChanges(ctx, kv, s.baseDomain)
if err != nil {
slog.Warn("rds DesiredDNSChanges: read DB instances", "bucket", bucket, "err", err)
return nil, false
}
changes = append(changes, bucketChanges...)
}
return changes, true
}
// A deleted instance's record is gone, so it contributes nothing and the
// reconcile prunes its record. Anything still holding an ENI IP keeps its name
// resolvable, including a failed instance an operator is still investigating.
func desiredBucketDNSChanges(ctx context.Context, kv jetstream.KeyValue, baseDomain string) ([]handlers_dns.Change, error) {
ids, err := ListDBInstanceIDs(ctx, kv)
if err != nil {
return nil, err
}
var changes []handlers_dns.Change
for _, id := range ids {
var rec DBInstanceRecord
found, err := getJSON(ctx, kv, DBInstanceKey(id), &rec)
if err != nil {
return nil, err
}
if !found || rec.Status == StatusDeleted || rec.DNSName == "" || rec.ENIPrivateIP == "" {
continue
}
changes = append(changes, handlers_dns.RDSChanges(
handlers_dns.ActionUpsert, rec.DNSName, baseDomain, rec.ENIPrivateIP)...)
}
return changes, nil
}
+138
View File
@@ -0,0 +1,138 @@
package handlers_rds
import (
"errors"
"fmt"
"maps"
"slices"
"strings"
"github.com/mulgadc/spinifex/spinifex/awserrors"
)
// The control-plane half of the engine seam: what CreateDBInstance needs to
// validate a request and assemble a bootstrap config. The in-guest half —
// initdb, quiesce, live password apply — lives in rds-init and rds-agent.
type Engine struct {
Name string
// Pinned for v1. A request naming another version is rejected rather than
// served by an image that is not the one asked for.
MajorVersion string
DefaultPort int64
// Identifiers the engine reserves for itself, which a master role may not
// take. Matched case-insensitively.
reservedUsernames []string
// Prefixes the engine reserves for internal roles.
reservedUsernamePrefixes []string
}
var engines = map[string]Engine{
enginePostgres.Name: enginePostgres,
}
// The pinned version an AMI lookup resolves against, which is the major alone:
// the AMI carries the major, and a minor is chosen by the image build.
func (e Engine) EngineVersion() string {
return e.MajorVersion
}
// The parameter-group name AWS clients expect when none is named. rds-7
// materialises the group itself; here it is only the name a request may carry.
func (e Engine) DefaultParameterGroupName() string {
return "default." + e.Name + e.MajorVersion
}
// An unknown engine is rejected at validation, before any volume or ENI exists.
func LookupEngine(name string) (Engine, error) {
engine, ok := engines[strings.ToLower(strings.TrimSpace(name))]
if !ok {
return Engine{}, fmt.Errorf("%s: engine %q is not supported; supported engines are %s",
awserrors.ErrorInvalidParameterValue, name, strings.Join(SupportedEngines(), ", "))
}
return engine, nil
}
func SupportedEngines() []string {
return slices.Sorted(maps.Keys(engines))
}
// An empty version takes the pin. A supplied one must name the pinned major,
// with or without a minor, since a mismatch would silently run a version the
// customer did not ask for (D19).
func (e Engine) ValidateVersion(version string) error {
v := strings.TrimSpace(version)
if v == "" || v == e.MajorVersion {
return nil
}
if major, _, ok := strings.Cut(v, "."); ok && major == e.MajorVersion {
return nil
}
return fmt.Errorf("%s: EngineVersion %q is not available; %s %s is the only supported version",
awserrors.ErrorInvalidParameterValue, version, e.Name, e.MajorVersion)
}
// Mirrors the engine's own rules rather than a generic identifier check, so a
// name the control plane accepts cannot fail at initdb time inside the guest.
func (e Engine) ValidateMasterUsername(username string) error {
if username == "" {
return fmt.Errorf("%s: MasterUsername is required", awserrors.ErrorInvalidParameterValue)
}
if len(username) > maxMasterUsernameLen {
return fmt.Errorf("%s: MasterUsername must be at most %d characters",
awserrors.ErrorInvalidParameterValue, maxMasterUsernameLen)
}
if !isLetter(rune(username[0])) {
return fmt.Errorf("%s: MasterUsername must begin with a letter", awserrors.ErrorInvalidParameterValue)
}
for _, r := range username {
if !isLetter(r) && !isDigit(r) && r != '_' {
return fmt.Errorf("%s: MasterUsername may contain only letters, digits and underscores",
awserrors.ErrorInvalidParameterValue)
}
}
lower := strings.ToLower(username)
if slices.Contains(e.reservedUsernames, lower) {
return fmt.Errorf("%s: MasterUsername %q is reserved by %s",
awserrors.ErrorInvalidParameterValue, username, e.Name)
}
for _, prefix := range e.reservedUsernamePrefixes {
if strings.HasPrefix(lower, prefix) {
return fmt.Errorf("%s: MasterUsername may not begin with %q, which %s reserves",
awserrors.ErrorInvalidParameterValue, prefix, e.Name)
}
}
return nil
}
// Bounds only. The password is never inspected beyond this and never stored in
// cleartext past the first bootstrap fetch (D8).
func ValidateMasterUserPassword(password string) error {
switch {
case password == "":
return errors.New(awserrors.ErrorInvalidParameterValue + ": MasterUserPassword is required")
case len(password) < minMasterPasswordLen || len(password) > maxMasterPasswordLen:
return fmt.Errorf("%s: MasterUserPassword must be between %d and %d characters",
awserrors.ErrorInvalidParameterValue, minMasterPasswordLen, maxMasterPasswordLen)
}
for _, r := range password {
if r == '/' || r == '"' || r == '@' || r == ' ' {
return fmt.Errorf("%s: MasterUserPassword may not contain '/', '\"', '@' or spaces",
awserrors.ErrorInvalidParameterValue)
}
}
return nil
}
const (
maxMasterUsernameLen = 63
minMasterPasswordLen = 8
maxMasterPasswordLen = 128
)
func isLetter(r rune) bool {
return (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z')
}
func isDigit(r rune) bool {
return r >= '0' && r <= '9'
}
+13
View File
@@ -0,0 +1,13 @@
package handlers_rds
// PostgreSQL 18 is the pinned v1 major, matching the rds-postgres AMI preset.
// A new major is a new AMI plus a bump here, never a runtime upgrade.
var enginePostgres = Engine{
Name: "postgres",
MajorVersion: "18",
DefaultPort: 5432,
// rdsadmin is the management role AWS reserves; postgres is the cluster
// superuser initdb creates, which the master role must not collide with.
reservedUsernames: []string{"rdsadmin", "postgres"},
reservedUsernamePrefixes: []string{"pg_"},
}
+89
View File
@@ -0,0 +1,89 @@
package handlers_rds
import (
"strings"
"testing"
"github.com/mulgadc/spinifex/spinifex/awserrors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestLookupEngine(t *testing.T) {
_, err := LookupEngine("PostgreSQL")
require.Error(t, err, "the AWS engine identifier is 'postgres', not the product name")
engine, err := LookupEngine(" POSTGRES ")
require.NoError(t, err, "the engine name is matched case-insensitively and trimmed")
assert.Equal(t, "postgres", engine.Name)
assert.Equal(t, int64(5432), engine.DefaultPort)
assert.Equal(t, "default.postgres18", engine.DefaultParameterGroupName())
_, err = LookupEngine("mysql")
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
assert.Equal(t, []string{"postgres"}, SupportedEngines())
}
// D19: a version other than the pinned one would be served by an image that is
// not the one asked for, so it is rejected rather than quietly substituted.
func TestEngineValidateVersion(t *testing.T) {
engine, err := LookupEngine("postgres")
require.NoError(t, err)
for _, version := range []string{"", "18", "18.1", "18.4"} {
assert.NoError(t, engine.ValidateVersion(version), "version %q", version)
}
for _, version := range []string{"17", "16.2", "19", "latest"} {
assert.Error(t, engine.ValidateVersion(version), "version %q", version)
}
// The pin, not the request, is what the record and the AMI lookup carry.
assert.Equal(t, "18", engine.EngineVersion())
}
func TestEngineValidateMasterUsername(t *testing.T) {
engine, err := LookupEngine("postgres")
require.NoError(t, err)
for _, username := range []string{"appuser", "app_user1", "Orders"} {
assert.NoError(t, engine.ValidateMasterUsername(username), "username %q", username)
}
cases := map[string]string{
"empty": "",
"leading digit": "1user",
"hyphen": "app-user",
"too long": strings.Repeat("a", maxMasterUsernameLen+1),
"reserved": "rdsadmin",
"reserved cased": "RDSAdmin",
// postgres reserves the pg_ prefix for its own internal roles, so a master
// role taking one would collide inside the engine, not at the API.
"reserved prefix": "pg_backup",
}
for name, username := range cases {
t.Run(name, func(t *testing.T) {
err := engine.ValidateMasterUsername(username)
require.Error(t, err)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
})
}
}
func TestValidateMasterUserPassword(t *testing.T) {
assert.NoError(t, ValidateMasterUserPassword("Sup3rSecret!"))
assert.NoError(t, ValidateMasterUserPassword(strings.Repeat("x", maxMasterPasswordLen)))
for _, password := range []string{
"",
strings.Repeat("x", minMasterPasswordLen-1),
strings.Repeat("x", maxMasterPasswordLen+1),
// The characters AWS excludes, which would otherwise break a connection
// string or the engine's own role syntax.
"pass/word1", `pass"word1`, "pass@word1", "pass word1",
} {
err := ValidateMasterUserPassword(password)
require.Error(t, err, "password %q should be rejected", password)
assert.Contains(t, err.Error(), awserrors.ErrorInvalidParameterValue)
}
}
+49
View File
@@ -0,0 +1,49 @@
package handlers_rds
import (
"log/slog"
handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam"
)
const (
// The instance role the in-guest rds-agent assumes through IMDS. The
// gateway's internal-action gate admits a caller only when the session was
// assumed from this role, so the name is part of the protocol.
InstanceRoleName = "rdsInstanceRole"
instanceRoleInlinePolicyName = "spinifex-rds-instance-internal"
// Only the four internal agent actions. The customer-facing rds:* set is
// rds-10a's; granting it here would let a DB VM manage its account's fleet.
instanceRoleInlinePolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["rds:RegisterDBInstance","rds:SubmitDBStateChange","rds:PollDBCommands","rds:GetDBBootstrapConfig"],"Resource":"*"}]}`
)
// Resolved per launch rather than held, mirroring EKS: the KV-backed IAM
// service has no responders until JetStream is up, so an eager build races
// daemon boot and fails permanently on the node that loses.
type IAMProvider func() handlers_iam.SystemInstanceRoleEnsurer
// Returns the instance-profile ARN, or "" when IAM is unwired or the ensure
// failed. Unlike EKS there is no static-credential fallback: without the role
// the agent cannot authenticate, so an empty ARN launches a VM whose agent
// never registers and the reconciler marks failed.
func ensureInstanceProfile(provider IAMProvider, accountID string) string {
if provider == nil {
slog.Warn("rds: IAM unwired; the DB VM agent will have no gateway credentials")
return ""
}
iamSvc := provider()
if iamSvc == nil {
slog.Warn("rds: IAM service unavailable; the DB VM agent will have no gateway credentials")
return ""
}
profileARN, err := handlers_iam.EnsureSystemInstanceProfile(iamSvc, accountID,
InstanceRoleName, instanceRoleInlinePolicyName, instanceRoleInlinePolicy)
if err != nil {
slog.Warn("rds: ensure instance profile failed; the DB VM agent will have no gateway credentials",
"role", InstanceRoleName, "err", err)
return ""
}
return profileARN
}
+37 -14
View File
@@ -102,6 +102,9 @@ 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
}
type LaunchOutput struct {
@@ -115,6 +118,13 @@ type LaunchOutput struct {
DataVolumeID string
DataDevice string
MgmtIP string
// The volume's own reported state, not an echo of the request, so
// DescribeDBInstances reports encryption the way EC2 does.
DataVolumeEncrypted bool
// Tears down everything this launch created, for a caller that fails after
// it returned. The launch runs it itself on its own failures, so it is only
// ever invoked once.
Unwind func(context.Context)
}
// On any failure every resource this call created is torn down, so a retried
@@ -144,12 +154,9 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
// volume are attached while it runs, and deleting those is rejected as InUse.
var terminateVM func(context.Context)
defer func() {
if err == nil {
return
}
// A context detached from the caller's, so a create that failed *because*
// the request deadline expired can still clean up.
// A context detached from the caller's, so a create that failed *because*
// the request deadline expired can still clean up.
unwind := func(ctx context.Context) {
rbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), rollbackTimeout)
defer cancel()
if terminateVM != nil {
@@ -158,6 +165,12 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
for _, undo := range slices.Backward(rollback) {
undo(rbCtx)
}
}
defer func() {
if err != nil {
unwind(ctx)
}
}()
systemENI, err := createLaunchENI(ctx, deps.VPC, utils.GlobalAccountID, systemSubnetID, nil,
@@ -236,6 +249,7 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
return nil, fmt.Errorf("rds: create data volume for %s: empty volume id", in.DBInstanceIdentifier)
}
volumeID := aws.StringValue(volume.VolumeId)
volumeEncrypted := aws.BoolValue(volume.Encrypted)
rollback = append(rollback, func(ctx context.Context) {
if _, delErr := deps.Volume.DeleteVolume(ctx, &ec2.DeleteVolumeInput{VolumeId: aws.String(volumeID)}, utils.GlobalAccountID); delErr != nil {
slog.WarnContext(ctx, "rds: rollback delete of orphaned data volume failed",
@@ -243,6 +257,13 @@ 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 {
return nil, fmt.Errorf("rds: data volume %s for %s was created unencrypted; the cluster storage key is not configured",
volumeID, in.DBInstanceIdentifier)
}
device, err := deps.Attacher.AttachVolume(ctx, utils.GlobalAccountID, instanceID, volumeID, dataVolumeDevice)
if err != nil {
return nil, fmt.Errorf("rds: attach data volume %s to %s: %w", volumeID, instanceID, err)
@@ -254,14 +275,16 @@ func LaunchDBInstanceVM(ctx context.Context, deps LaunchDeps, in LaunchInput) (o
"dataVolume", volumeID, "device", device)
return &LaunchOutput{
InstanceID: instanceID,
SystemENIID: systemENI.id,
CustomerENIID: customerENI.id,
CustomerENIIP: customerENI.ip,
CustomerENIMac: customerENI.mac,
DataVolumeID: volumeID,
DataDevice: device,
MgmtIP: sysOut.MgmtIP,
InstanceID: instanceID,
SystemENIID: systemENI.id,
CustomerENIID: customerENI.id,
CustomerENIIP: customerENI.ip,
CustomerENIMac: customerENI.mac,
DataVolumeID: volumeID,
DataDevice: device,
MgmtIP: sysOut.MgmtIP,
DataVolumeEncrypted: volumeEncrypted,
Unwind: unwind,
}, nil
}
+12 -1
View File
@@ -207,6 +207,10 @@ type fakeLauncher struct {
err error
terminated []string
unwind *[]string
// onLaunch runs once the VM exists, for tests that need to disturb state
// after the launch has committed resources but before the caller records it.
onLaunch func()
}
var _ launchInstanceLauncher = (*fakeLauncher)(nil)
@@ -216,6 +220,9 @@ func (f *fakeLauncher) LaunchSystemInstance(in *sysinstance.SystemInstanceInput)
if f.err != nil {
return nil, f.err
}
if f.onLaunch != nil {
f.onLaunch()
}
return &sysinstance.SystemInstanceOutput{InstanceID: "i-rds0001", MgmtIP: "172.30.0.9"}, nil
}
@@ -251,6 +258,10 @@ type fakeVolumes struct {
deleted []string
err error
// encrypted is what the created volume reports about itself, which is what
// the launch's encrypted-storage guard reads — not the request's echo.
encrypted bool
unwind *[]string
deleteCtxErr error
}
@@ -263,7 +274,7 @@ func (f *fakeVolumes) CreateVolume(_ context.Context, in *ec2.CreateVolumeInput,
}
f.created = append(f.created, in)
f.accts = append(f.accts, accountID)
return &ec2.Volume{VolumeId: aws.String("vol-rdsdata01")}, nil
return &ec2.Volume{VolumeId: aws.String("vol-rdsdata01"), Encrypted: aws.Bool(f.encrypted)}, nil
}
func (f *fakeVolumes) DeleteVolume(ctx context.Context, in *ec2.DeleteVolumeInput, _ string) (*ec2.DeleteVolumeOutput, error) {
+135
View File
@@ -0,0 +1,135 @@
package handlers_rds
import (
"context"
"fmt"
"slices"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
)
// The customer-account VPC surface CreateDBInstance reads to place the endpoint
// ENI. Read-only: the ENI itself is created through the launch deps.
type networkResolver interface {
DescribeVpcs(ctx context.Context, input *ec2.DescribeVpcsInput, accountID string) (*ec2.DescribeVpcsOutput, error)
DescribeSubnets(ctx context.Context, input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error)
DescribeSecurityGroups(ctx context.Context, input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error)
}
// Where the customer-facing ENI lands.
type endpointPlacement struct {
VpcID string
SubnetID string
SecurityGroupIDs []string
}
// Until DB subnet groups land in rds-7 the placement is the account's default
// VPC, mirroring AWS's own behaviour when a request names no subnet group. A
// named group is rejected in validation rather than silently resolved here.
func (s *Service) resolvePlacement(ctx context.Context, accountID string, req *validatedCreate) (*endpointPlacement, error) {
if s.deps.Network == nil {
return nil, fmt.Errorf("%s: RDS networking is not wired on this node", awserrors.ErrorServerInternal)
}
vpcID, err := s.defaultVPCID(ctx, accountID)
if err != nil {
return nil, err
}
subnetID, err := s.firstSubnetID(ctx, accountID, vpcID)
if err != nil {
return nil, err
}
groupIDs, err := s.resolveSecurityGroups(ctx, accountID, vpcID, req.SecurityGroupIDs)
if err != nil {
return nil, err
}
return &endpointPlacement{VpcID: vpcID, SubnetID: subnetID, SecurityGroupIDs: groupIDs}, nil
}
func (s *Service) defaultVPCID(ctx context.Context, accountID string) (string, error) {
out, err := s.deps.Network.DescribeVpcs(ctx, &ec2.DescribeVpcsInput{
Filters: []*ec2.Filter{{Name: aws.String("isDefault"), Values: aws.StringSlice([]string{"true"})}},
}, accountID)
if err != nil {
return "", fmt.Errorf("rds: describe default VPC: %w", err)
}
if out != nil {
for _, vpc := range out.Vpcs {
if id := aws.StringValue(vpc.VpcId); id != "" {
return id, nil
}
}
}
return "", fmt.Errorf("%s: no default VPC in this account to place the DB endpoint in",
awserrors.ErrorDBInvalidVPCNetworkState)
}
// Sorted so a repeated create in the same account is placed deterministically
// rather than wherever the describe happened to order its results.
func (s *Service) firstSubnetID(ctx context.Context, accountID, vpcID string) (string, error) {
out, err := s.deps.Network.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
Filters: []*ec2.Filter{{Name: aws.String("vpc-id"), Values: aws.StringSlice([]string{vpcID})}},
}, accountID)
if err != nil {
return "", fmt.Errorf("rds: describe subnets in %s: %w", vpcID, err)
}
var ids []string
if out != nil {
for _, subnet := range out.Subnets {
if id := aws.StringValue(subnet.SubnetId); id != "" {
ids = append(ids, id)
}
}
}
if len(ids) == 0 {
return "", fmt.Errorf("%s: default VPC %s has no subnet to place the DB endpoint in",
awserrors.ErrorDBInvalidVPCNetworkState, vpcID)
}
slices.Sort(ids)
return ids[0], nil
}
// An unset list takes the VPC's default group, matching AWS. A supplied one is
// checked against the placement VPC, because an ENI cannot carry a group from
// another VPC and the launch would otherwise fail after the record exists.
func (s *Service) resolveSecurityGroups(ctx context.Context, accountID, vpcID string, requested []string) ([]string, error) {
out, err := s.deps.Network.DescribeSecurityGroups(ctx, &ec2.DescribeSecurityGroupsInput{
Filters: []*ec2.Filter{{Name: aws.String("vpc-id"), Values: aws.StringSlice([]string{vpcID})}},
}, accountID)
if err != nil {
return nil, fmt.Errorf("rds: describe security groups in %s: %w", vpcID, err)
}
inVPC := map[string]string{}
defaultGroupID := ""
if out != nil {
for _, group := range out.SecurityGroups {
id := aws.StringValue(group.GroupId)
if id == "" {
continue
}
name := aws.StringValue(group.GroupName)
inVPC[id] = name
if name == "default" {
defaultGroupID = id
}
}
}
if len(requested) == 0 {
if defaultGroupID == "" {
return nil, fmt.Errorf("%s: VPC %s has no default security group",
awserrors.ErrorDBInvalidVPCNetworkState, vpcID)
}
return []string{defaultGroupID}, nil
}
for _, id := range requested {
if _, ok := inVPC[id]; !ok {
return nil, fmt.Errorf("%s: security group %s is not in VPC %s",
awserrors.ErrorInvalidParameterValue, id, vpcID)
}
}
return requested, nil
}
+313
View File
@@ -0,0 +1,313 @@
package handlers_rds
import (
"context"
"errors"
"fmt"
"log/slog"
"sync"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go/jetstream"
)
// Lease and sweep timing. The holder refreshes well inside the bucket's 60s TTL,
// so a leader that dies is replaced within one TTL rather than one refresh.
const (
reconcilerLeaderKey = "reconciler"
leaseRefresh = 20 * time.Second
reconcileInterval = 15 * time.Second
// How long a creating instance may go without a healthy heartbeat before it
// is called failed, unless the config overrides it. It has to cover a cold
// boot plus initdb on the slowest class, so it is deliberately generous — a
// 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
)
// The EC2 lifecycle states a DB VM may be in and still be on its way up. The
// reconciler will not call an instance available until its VM is running.
const instanceStateRunning = "running"
// The VM's EC2 lifecycle state, fanned out across every host so a DB VM is
// observed wherever it landed. Nil disables the VM half of the check.
type InstanceStateResolver interface {
InstanceState(ctx context.Context, instanceID, accountID string) (string, error)
}
// The leader-elected RDS control loop. One node holds the lease and does the
// control work; every node keeps serving the API, so a leaderless gap delays a
// status transition without failing a request.
//
// This phase's responsibilities are the creating → available transition and
// marking a stalled bootstrap failed. Auto-recovery (rds-6) and the backup
// sweep (rds-9) extend the same loop.
type Reconciler struct {
svc *Service
holder string
mu sync.Mutex
leader bool
}
// holder identifies this daemon in the lease.
func NewReconciler(svc *Service, holder string) *Reconciler {
return &Reconciler{svc: svc, holder: holder}
}
// Drives the leadership and reconcile loop until ctx is cancelled. Intended as
// a daemon-boot goroutine; panics are the caller's recover concern.
func (r *Reconciler) Run(ctx context.Context) {
leaseTicker := time.NewTicker(leaseRefresh)
reconcileTicker := time.NewTicker(reconcileInterval)
defer leaseTicker.Stop()
defer reconcileTicker.Stop()
r.evaluateLeadership(ctx)
for {
select {
case <-ctx.Done():
r.relinquish()
return
case <-leaseTicker.C:
r.evaluateLeadership(ctx)
case <-reconcileTicker.C:
if !r.isLeader() {
continue
}
if err := r.reconcileOnce(ctx); err != nil {
slog.ErrorContext(ctx, "rds reconciler: pass failed", "holder", r.holder, "err", err)
}
}
}
}
func (r *Reconciler) isLeader() bool {
r.mu.Lock()
defer r.mu.Unlock()
return r.leader
}
func (r *Reconciler) evaluateLeadership(ctx context.Context) {
won := r.acquireOrRefresh(ctx)
r.mu.Lock()
was := r.leader
r.leader = won
r.mu.Unlock()
switch {
case won && !was:
slog.Info("rds reconciler: elected leader", "holder", r.holder)
case !won && was:
slog.Info("rds reconciler: lost leadership", "holder", r.holder)
}
}
// Claims the lease, or refreshes it (resetting the TTL) when we already hold it.
func (r *Reconciler) acquireOrRefresh(ctx context.Context) bool {
kv, err := r.leaderBucket(ctx)
if err != nil {
return false
}
if _, err := kv.Create(ctx, reconcilerLeaderKey, []byte(r.holder)); err == nil {
return true
}
entry, err := kv.Get(ctx, reconcilerLeaderKey)
if err != nil {
return false
}
if string(entry.Value()) != r.holder {
return false
}
if _, err := kv.Put(ctx, reconcilerLeaderKey, []byte(r.holder)); err != nil {
return false
}
return true
}
// Releases the lease on shutdown so the next leader is elected immediately
// rather than after the TTL.
func (r *Reconciler) relinquish() {
// Run's ctx is already cancelled by the time this is called, so the release
// runs on its own — a cancelled ctx would fail the delete.
ctx := context.Background()
kv, err := r.leaderBucket(ctx)
if err != nil {
return
}
if entry, gerr := kv.Get(ctx, reconcilerLeaderKey); gerr == nil && string(entry.Value()) == r.holder {
if err := kv.Delete(ctx, reconcilerLeaderKey); err != nil {
slog.Debug("rds reconciler: release lease failed", "holder", r.holder, "err", err)
}
}
}
func (r *Reconciler) leaderBucket(ctx context.Context) (jetstream.KeyValue, error) {
js, err := r.svc.js()
if err != nil {
return nil, err
}
return InitLeaderBucket(ctx, js)
}
// One pass across every tenant. A bucket that cannot be read stops the pass
// with an error rather than being skipped silently, so a partial view shows up
// in the log instead of looking like a fleet with nothing to do.
func (r *Reconciler) reconcileOnce(ctx context.Context) error {
js, err := r.svc.js()
if err != nil {
return err
}
buckets, err := AccountBucketNames(ctx, js)
if err != nil {
return fmt.Errorf("rds reconciler: enumerate account buckets: %w", err)
}
var failures []error
for _, bucket := range buckets {
kv, err := js.KeyValue(ctx, bucket)
if err != nil {
failures = append(failures, fmt.Errorf("open %s: %w", bucket, err))
continue
}
if err := r.reconcileAccount(ctx, kv, AccountIDFromBucketName(bucket)); err != nil {
failures = append(failures, fmt.Errorf("reconcile %s: %w", bucket, err))
}
}
return errors.Join(failures...)
}
func (r *Reconciler) reconcileAccount(ctx context.Context, kv jetstream.KeyValue, accountID string) error {
ids, err := ListDBInstanceIDs(ctx, kv)
if err != nil {
return err
}
var failures []error
for _, id := range ids {
if err := r.reconcileInstance(ctx, kv, accountID, id); err != nil {
failures = append(failures, fmt.Errorf("%s: %w", id, err))
}
}
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.
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 {
return nil
}
healthy, err := r.bootstrapComplete(ctx, accountID, &rec)
if err != nil {
return err
}
if healthy {
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,
fmt.Sprintf("the database engine did not report healthy within %s of creation", timeout))
}
return nil
}
// 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) {
if rec.Agent.EngineHealth != EngineHealthHealthy || rec.InstanceID == "" ||
rec.Agent.InstanceID != rec.InstanceID {
return false, nil
}
if !r.heartbeatFresh(accountID, rec) {
return false, nil
}
if r.svc.deps.InstanceState == nil {
return true, nil
}
state, err := r.svc.deps.InstanceState.InstanceState(ctx, rec.InstanceID, accountID)
if err != nil {
return false, fmt.Errorf("resolve VM state for %s: %w", rec.InstanceID, err)
}
return state == instanceStateRunning, nil
}
// 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 {
lastSeen, ok := r.svc.LastSeen(accountID, rec.DBInstanceIdentifier)
if !ok {
if rec.Agent.LastSeen == nil {
return false
}
lastSeen = *rec.Agent.LastSeen
}
return time.Since(lastSeen) <= HeartbeatStaleAfter
}
// A CAS write, so a transition raced by an agent report or a lifecycle op is
// dropped rather than clobbering the newer state; the next pass re-reads.
func (r *Reconciler) transition(ctx context.Context, kv jetstream.KeyValue, rev uint64, rec *DBInstanceRecord, to Status, reason string) error {
if !CanTransition(rec.Status, to) {
return fmt.Errorf("illegal transition %s -> %s", rec.Status, to)
}
from := rec.Status
rec.Status = to
rec.FailureReason = reason
rec.UpdatedAt = time.Now().UTC()
if err := updateJSON(ctx, kv, DBInstanceKey(rec.DBInstanceIdentifier), rev, rec); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
slog.DebugContext(ctx, "rds reconciler: transition lost a revision race; retrying next pass",
"dbInstance", rec.DBInstanceIdentifier, "to", to)
return nil
}
return err
}
slog.InfoContext(ctx, "rds reconciler: DB instance transitioned",
"dbInstance", rec.DBInstanceIdentifier, "from", from, "to", to, "reason", reason)
return nil
}
// Adapts an EC2 describe fan-out to the reconciler's narrow state lookup.
type describeInstanceState struct {
describe func(input *ec2.DescribeInstancesInput, accountID string) (*ec2.DescribeInstancesOutput, error)
}
var _ InstanceStateResolver = (*describeInstanceState)(nil)
// The VM runs in the system account, so the describe is issued there; the
// customer account cannot see a platform-managed instance.
func NewDescribeInstanceState(describe func(*ec2.DescribeInstancesInput, string) (*ec2.DescribeInstancesOutput, error)) InstanceStateResolver {
return &describeInstanceState{describe: describe}
}
func (d *describeInstanceState) InstanceState(_ context.Context, instanceID, _ string) (string, error) {
out, err := d.describe(&ec2.DescribeInstancesInput{
InstanceIds: []*string{aws.String(instanceID)},
}, utils.GlobalAccountID)
if err != nil {
return "", err
}
for _, reservation := range out.Reservations {
for _, instance := range reservation.Instances {
if aws.StringValue(instance.InstanceId) == instanceID && instance.State != nil {
return aws.StringValue(instance.State.Name), nil
}
}
}
// A VM the platform cannot find is not running, which is the answer the
// caller needs — not an error that would stall the whole pass.
return "", nil
}
+241
View File
@@ -0,0 +1,241 @@
package handlers_rds
import (
"context"
"errors"
"testing"
"time"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// fakeInstanceState answers the VM half of the bootstrap check.
type fakeInstanceState struct {
state string
err error
calls []string
}
var _ InstanceStateResolver = (*fakeInstanceState)(nil)
func (f *fakeInstanceState) InstanceState(_ context.Context, instanceID, _ string) (string, error) {
f.calls = append(f.calls, instanceID)
if f.err != nil {
return "", f.err
}
return f.state, nil
}
type reconcileHarness struct {
svc *Service
rec *Reconciler
state *fakeInstanceState
nc *nats.Conn
}
func newReconcileHarness(t *testing.T, opts ...func(*Deps)) *reconcileHarness {
t.Helper()
_, nc, _ := testutil.StartTestJetStream(t)
state := &fakeInstanceState{state: instanceStateRunning}
deps := Deps{LoadCA: newTestCA(t), InstanceState: state}
for _, opt := range opts {
opt(&deps)
}
svc := NewService(nc, testRegion).WithDeps(deps)
return &reconcileHarness{svc: svc, rec: NewReconciler(svc, "node-a"), state: state, nc: nc}
}
// healthyRecord is a creating instance whose agent has just reported a healthy
// engine from the record's current VM — the shape the reconciler flips.
func healthyRecord() DBInstanceRecord {
now := time.Now().UTC()
rec := defaultRecord()
rec.CreatedAt = now.Add(-2 * time.Minute)
rec.Agent = AgentState{
InstanceID: testInstance,
EngineHealth: EngineHealthHealthy,
LastSeen: &now,
}
return rec
}
func (h *reconcileHarness) statusOf(t *testing.T, id string) (Status, string) {
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)
require.True(t, found)
return rec.Status, rec.FailureReason
}
func TestReconciler_MarksAvailableOnHealthyHeartbeatFromARunningVM(t *testing.T) {
h := newReconcileHarness(t)
seedInstance(t, h.svc, healthyRecord())
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, reason := h.statusOf(t, testDBID)
assert.Equal(t, StatusAvailable, status)
assert.Empty(t, reason)
assert.Equal(t, []string{testInstance}, h.state.calls)
}
// Both halves must hold. A healthy beat from a VM that is not running means the
// engine reported before the platform finished bringing the VM up, or the beat
// is stale — either way the instance is not ready.
func TestReconciler_HoldsCreatingWhenTheVMIsNotRunning(t *testing.T) {
h := newReconcileHarness(t)
h.state.state = "pending"
seedInstance(t, h.svc, healthyRecord())
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, _ := h.statusOf(t, testDBID)
assert.Equal(t, StatusCreating, status)
}
func TestReconciler_HoldsCreatingUntilTheEngineIsHealthy(t *testing.T) {
cases := []struct {
name string
mutate func(*DBInstanceRecord)
}{
{"NoHeartbeatYet", func(rec *DBInstanceRecord) { rec.Agent = AgentState{} }},
{"EngineStillStarting", func(rec *DBInstanceRecord) { rec.Agent.EngineHealth = EngineHealthStarting }},
{"EngineUnhealthy", func(rec *DBInstanceRecord) { rec.Agent.EngineHealth = EngineHealthUnhealthy }},
// A beat from a VM other than the record's current one is a superseded VM
// still running after a replace; it must not report the instance ready.
{"BeatFromASupersededVM", func(rec *DBInstanceRecord) { rec.Agent.InstanceID = "i-oldvm" }},
{"StaleHeartbeat", func(rec *DBInstanceRecord) {
stale := time.Now().UTC().Add(-2 * HeartbeatStaleAfter)
rec.Agent.LastSeen = &stale
}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
h := newReconcileHarness(t)
rec := healthyRecord()
tc.mutate(&rec)
seedInstance(t, h.svc, rec)
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, _ := h.statusOf(t, testDBID)
assert.Equal(t, StatusCreating, status)
})
}
}
// A create that never comes up has to end somewhere: the customer sees a broken
// instance either way, and failed is the state they can act on.
func TestReconciler_MarksFailedWhenTheBootstrapTimesOut(t *testing.T) {
h := newReconcileHarness(t, func(d *Deps) { d.BootstrapTimeout = time.Minute })
rec := healthyRecord()
rec.Agent = AgentState{}
rec.CreatedAt = time.Now().UTC().Add(-10 * time.Minute)
seedInstance(t, h.svc, rec)
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, reason := h.statusOf(t, testDBID)
assert.Equal(t, StatusFailed, status)
assert.Contains(t, reason, "did not report healthy")
}
// Inside the window a slow bootstrap is still a bootstrap: a false failed is
// worse than a slow one.
func TestReconciler_LeavesASlowBootstrapAloneInsideTheWindow(t *testing.T) {
h := newReconcileHarness(t, func(d *Deps) { d.BootstrapTimeout = time.Hour })
rec := healthyRecord()
rec.Agent = AgentState{}
rec.CreatedAt = time.Now().UTC().Add(-10 * time.Minute)
seedInstance(t, h.svc, rec)
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, _ := h.statusOf(t, testDBID)
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} {
t.Run(string(status), func(t *testing.T) {
h := newReconcileHarness(t)
rec := healthyRecord()
rec.Status = status
rec.CreatedAt = time.Now().UTC().Add(-24 * time.Hour)
seedInstance(t, h.svc, rec)
require.NoError(t, h.rec.reconcileOnce(t.Context()))
got, _ := h.statusOf(t, testDBID)
assert.Equal(t, status, got)
assert.Empty(t, h.state.calls, "a non-creating instance is not even inspected")
})
}
}
// A VM-state lookup that fails is not evidence the instance is unhealthy, so the
// pass reports the error rather than silently holding — or worse, failing — it.
func TestReconciler_SurfacesAVMStateLookupFailure(t *testing.T) {
h := newReconcileHarness(t)
h.state.err = errors.New("no node answered the describe")
seedInstance(t, h.svc, healthyRecord())
err := h.rec.reconcileOnce(t.Context())
require.Error(t, err)
assert.Contains(t, err.Error(), "no node answered the describe")
status, _ := h.statusOf(t, testDBID)
assert.Equal(t, StatusCreating, status)
}
// Without a VM-state resolver the heartbeat alone decides, so a node with EC2
// unwired still makes progress rather than stalling every create.
func TestReconciler_FallsBackToTheHeartbeatWhenVMStateIsUnwired(t *testing.T) {
h := newReconcileHarness(t, func(d *Deps) { d.InstanceState = nil })
seedInstance(t, h.svc, healthyRecord())
require.NoError(t, h.rec.reconcileOnce(t.Context()))
status, _ := h.statusOf(t, testDBID)
assert.Equal(t, StatusAvailable, status)
}
// One node does the control work; the rest keep serving the API. A second
// claimant must not also believe it is leader, or two nodes would transition
// the same records.
func TestReconciler_ElectsASingleLeader(t *testing.T) {
h := newReconcileHarness(t)
other := NewReconciler(h.svc, "node-b")
assert.True(t, h.rec.acquireOrRefresh(t.Context()))
assert.False(t, other.acquireOrRefresh(t.Context()))
// The holder refreshes its own lease rather than losing it to itself.
assert.True(t, h.rec.acquireOrRefresh(t.Context()))
// Releasing on shutdown hands over immediately instead of after the TTL.
h.rec.relinquish()
assert.True(t, other.acquireOrRefresh(t.Context()))
assert.False(t, h.rec.acquireOrRefresh(t.Context()))
}
// A node that never won the lease must not delete the holder's key on shutdown.
func TestReconciler_RelinquishOnlyReleasesItsOwnLease(t *testing.T) {
h := newReconcileHarness(t)
other := NewReconciler(h.svc, "node-b")
require.True(t, h.rec.acquireOrRefresh(t.Context()))
other.relinquish()
assert.False(t, other.acquireOrRefresh(t.Context()), "the original holder still owns the lease")
}
+23 -1
View File
@@ -13,15 +13,29 @@ type DBInstanceRecord struct {
EngineVersion string `json:"engineVersion"`
DBInstanceClass string `json:"dbInstanceClass"`
AllocatedStorage int64 `json:"allocatedStorage"`
StorageType string `json:"storageType,omitempty"`
DBName string `json:"dbName,omitempty"`
MasterUsername string `json:"masterUsername"`
Port int64 `json:"port"`
// 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"`
VpcID string `json:"vpcId,omitempty"`
VpcSecurityGroupIDs []string `json:"vpcSecurityGroupIds,omitempty"`
DBSubnetGroupName string `json:"dbSubnetGroupName,omitempty"`
DBParameterGroupName string `json:"dbParameterGroupName,omitempty"`
// Changes on every replace, which is why the bus subject keys off the DB
// instance identifier instead.
InstanceID string `json:"instanceId,omitempty"`
InstanceID string `json:"instanceId,omitempty"`
// Increments on every replace, so a superseded VM's agent is
// distinguishable from the current one.
VMGeneration int64 `json:"vmGeneration,omitempty"`
DataVolumeID string `json:"dataVolumeId,omitempty"`
ENIID string `json:"eniId,omitempty"`
// Disposable: a replace mints a new one, unlike the customer ENI.
SystemENIID string `json:"systemEniId,omitempty"`
// Stable across VM replace, so it serves as both the fallback endpoint and
// a durable IP SAN on the serving cert.
ENIPrivateIP string `json:"eniPrivateIp,omitempty"`
@@ -30,6 +44,14 @@ type DBInstanceRecord struct {
// EndpointAddress because the cert needs it as a DNS SAN either way.
DNSName string `json:"dnsName,omitempty"`
// Reported from the data volume's own state rather than echoed from the
// request, the way EC2 derives a volume's Encrypted.
StorageEncrypted bool `json:"storageEncrypted,omitempty"`
// Why the instance is failed. Cleared when it leaves the failed state, so a
// stale reason cannot outlive the failure it describes.
FailureReason string `json:"failureReason,omitempty"`
Bootstrap BootstrapState `json:"bootstrap"`
Agent AgentState `json:"agent"`
+45 -3
View File
@@ -24,13 +24,34 @@ type Deps struct {
// Overrides the file-backed CA loader in tests.
LoadCA CALoader
Launch LaunchDeps
// Resolves the customer subnet and validates the security groups the
// customer ENI is created with.
Network networkResolver
// Find-or-creates the rdsInstanceRole instance profile granted at launch.
IAM IAMProvider
// Lets the reconciler confirm a DB VM is running before calling its
// instance available. Nil leaves the transition on the heartbeat alone.
InstanceState InstanceStateResolver
// The northstar base zone. Empty means no vanity hostname, and the endpoint
// is the customer-ENI IP instead (D6).
BaseDomain string
// Identifies this node in the reconciler lease.
HolderID string
// Seeded into the DB VM's cloud-init so the in-guest agent can reach the
// gateway and pin its TLS. No credentials: those come from IMDS.
GatewayURL string
GatewayCACert string
// Overrides how long the reconciler waits for the first healthy heartbeat.
// Zero takes defaultBootstrapTimeout.
BootstrapTimeout time.Duration
}
// The RDS control plane's KV-backed handler set. One per daemon.
type Service struct {
nc *nats.Conn
region string
deps Deps
nc *nats.Conn
region string
baseDomain string
deps Deps
// Heartbeat state that never reaches KV: beats are counted here and
// persisted only on change or on the slower floor.
@@ -52,9 +73,17 @@ func NewService(nc *nats.Conn, region string) *Service {
func (s *Service) WithDeps(d Deps) *Service {
s.deps = d
s.baseDomain = d.BaseDomain
return s
}
func (s *Service) bootstrapTimeout() time.Duration {
if s.deps.BootstrapTimeout > 0 {
return s.deps.BootstrapTimeout
}
return defaultBootstrapTimeout
}
func (s *Service) js() (jetstream.JetStream, error) {
if s.nc == nil {
return nil, errors.New("rds service: nil nats connection")
@@ -134,6 +163,19 @@ func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) erro
return nil
}
// Writes v at key only if nothing is stored there, so the key doubles as a
// cluster-wide reservation. Returns jetstream.ErrKeyExists when it is taken.
func createJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
data, err := json.Marshal(v)
if err != nil {
return err
}
if _, err := kv.Create(ctx, key, data); err != nil {
return err
}
return nil
}
// Writes v at key only if the stored entry is still at rev.
func updateJSON(ctx context.Context, kv jetstream.KeyValue, key string, rev uint64, v any) error {
data, err := json.Marshal(v)
+15
View File
@@ -4,12 +4,17 @@ import (
"context"
"time"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
const defaultTimeout = 30 * time.Second
// A create provisions an ENI, a VM and a volume inline, so it needs far more
// than the agent protocol's round-trip budget.
const createTimeout = 5 * time.Minute
// The gateway-side adapter that forwards each agent action as a NATS request to
// the daemon's matching subscriber.
type NATSService struct {
@@ -30,6 +35,16 @@ func (s *NATSService) SubmitDBStateChange(ctx context.Context, input *SubmitDBSt
BusHealthSubject(accountID, input.DBInstanceIdentifier), input, defaultTimeout, accountID)
}
func (s *NATSService) CreateDBInstance(ctx context.Context, input *rds.CreateDBInstanceInput, accountID string) (*rds.CreateDBInstanceOutput, error) {
return utils.NATSRequest[rds.CreateDBInstanceOutput](ctx, s.nc,
SubjectCreateDBInstance, input, createTimeout, accountID)
}
func (s *NATSService) DescribeDBInstances(ctx context.Context, input *rds.DescribeDBInstancesInput, accountID string) (*rds.DescribeDBInstancesOutput, error) {
return utils.NATSRequest[rds.DescribeDBInstancesOutput](ctx, s.nc,
SubjectDescribeDBInstances, input, defaultTimeout, accountID)
}
// Requested on the Layer-1 subject, not the bus.
func (s *NATSService) GetDBBootstrapConfig(ctx context.Context, input *GetDBBootstrapConfigInput, accountID string) (*GetDBBootstrapConfigOutput, error) {
return utils.NATSRequest[GetDBBootstrapConfigOutput](ctx, s.nc,
+76
View File
@@ -8,6 +8,8 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/mulgadc/spinifex/spinifex/kvutil"
"github.com/mulgadc/spinifex/spinifex/migrate"
@@ -26,6 +28,13 @@ const (
KVBucketRDSSystem = "rds-system"
KVBucketRDSSystemVersion = 1
KVBucketRDSSystemHistory = 1
// "spinifex-rds-leader" holds the single reconciler lease. Its TTL expires a
// lease whose holder died mid-cycle, so control work resumes without an
// operator.
KVBucketRDSLeader = "spinifex-rds-leader"
KVBucketRDSLeaderVersion = 1
KVBucketRDSLeaderTTL = 60 * time.Second
)
// Key-path helpers for the per-account bucket:
@@ -158,3 +167,70 @@ func GetOrCreateSystemBucket(ctx context.Context, js jetstream.JetStream) (jetst
}
return kv, nil
}
// kvutil.GetOrCreateBucket exposes no TTL knob, so the lease bucket takes the
// direct CreateKeyValue path and falls back to an open on already-exists.
func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.KeyValue, error) {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
Bucket: KVBucketRDSLeader,
History: 1,
TTL: KVBucketRDSLeaderTTL,
})
if err != nil {
kv, err = js.KeyValue(ctx, KVBucketRDSLeader)
if err != nil {
return nil, fmt.Errorf("failed to create or open RDS leader bucket %s: %w", KVBucketRDSLeader, err)
}
}
if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketRDSLeader, kv, KVBucketRDSLeaderVersion); err != nil {
return nil, fmt.Errorf("migrate %s: %w", KVBucketRDSLeader, err)
}
return kv, nil
}
// Every RDS per-account bucket in the cluster. The reconciler and the DNS
// desired-set both need a cross-tenant view, which only this provides.
func AccountBucketNames(ctx context.Context, js jetstream.JetStream) ([]string, error) {
all, err := kvutil.BucketNames(ctx, js)
if err != nil {
return nil, err
}
names := make([]string, 0, len(all))
for _, name := range all {
if strings.HasPrefix(name, KVBucketRDSAccountPrefix) {
names = append(names, name)
}
}
return names, nil
}
// The account a per-account bucket belongs to, for callers that enumerated
// buckets rather than starting from an account.
func AccountIDFromBucketName(bucket string) string {
return strings.TrimPrefix(bucket, KVBucketRDSAccountPrefix)
}
// The DB instance identifiers held in one account bucket. An empty bucket
// yields no names rather than an error.
func ListDBInstanceIDs(ctx context.Context, kv jetstream.KeyValue) ([]string, error) {
keys, err := kv.Keys(ctx)
if err != nil {
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
return nil, fmt.Errorf("rds: list keys: %w", err)
}
prefix := DBInstancesPrefix()
names := make([]string, 0, len(keys))
for _, key := range keys {
if !strings.HasPrefix(key, prefix) {
continue
}
name := strings.TrimPrefix(key, prefix)
if name == "" || strings.Contains(name, "/") {
continue
}
names = append(names, name)
}
return names, nil
}
+5
View File
@@ -17,6 +17,11 @@ const (
// control-plane read rather than reconciler↔agent traffic.
SubjectGetDBBootstrapConfig = "rds.GetDBBootstrapConfig"
// Layer-1 customer action subjects, answered by whichever daemon the queue
// group picks — a create does its own orchestration inline.
SubjectCreateDBInstance = "rds.CreateDBInstance"
SubjectDescribeDBInstances = "rds.DescribeDBInstances"
// Queue groups are scoped per subject, so one name shares command delivery
// across gateway nodes for the whole fleet.
CommandQueueGroup = "spinifex-rds-agents"
+64
View File
@@ -0,0 +1,64 @@
package handlers_rds
import (
"fmt"
"strings"
)
// Where the AMI's rds-agent looks for its settings and the gateway CA it pins
// TLS against. Both paths are baked into the image, so they are protocol.
const (
agentEnvPath = "/etc/spinifex-rds/agent.env"
agentGatewayCAPath = "/etc/spinifex-rds/gateway-ca.pem"
)
// No credentials: the agent draws auto-rotating instance-role credentials from
// IMDS, so nothing secret is written to a guest-readable data source. The master
// password arrives only over GetDBBootstrapConfig.
type agentUserDataInput struct {
GatewayURL string
GatewayCACert string
Region string
DBInstanceIdentifier string
EngineVersion string
EnginePort int64
}
type userDataFile struct {
path string
perms string
body string
}
func buildAgentUserData(in agentUserDataInput) string {
agentBody := strings.Join([]string{
"RDS_GATEWAY_URL=" + in.GatewayURL,
"RDS_GATEWAY_CA=" + agentGatewayCAPath,
"RDS_REGION=" + in.Region,
"RDS_DB_INSTANCE_IDENTIFIER=" + in.DBInstanceIdentifier,
"RDS_ENGINE_VERSION=" + in.EngineVersion,
fmt.Sprintf("RDS_ENGINE_PORT=%d", in.EnginePort),
}, "\n")
files := []userDataFile{{path: agentEnvPath, perms: "0600", body: agentBody}}
// A deployment with no gateway CA leaves the file out rather than writing an
// empty one, which the agent would load as an empty trust store.
if ca := strings.TrimRight(in.GatewayCACert, "\n"); ca != "" {
files = append(files, userDataFile{path: agentGatewayCAPath, perms: "0644", body: ca})
}
var buf strings.Builder
buf.WriteString("#cloud-config\n")
buf.WriteString("write_files:\n")
for _, f := range files {
fmt.Fprintf(&buf, " - path: %s\n", f.path)
fmt.Fprintf(&buf, " permissions: '%s'\n", f.perms)
buf.WriteString(" content: |\n")
for line := range strings.SplitSeq(f.body, "\n") {
buf.WriteString(" ")
buf.WriteString(line)
buf.WriteString("\n")
}
}
return buf.String()
}
+69
View File
@@ -0,0 +1,69 @@
package handlers_rds
import (
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func testAgentUserDataInput() agentUserDataInput {
return agentUserDataInput{
GatewayURL: "https://172.30.0.1:9999",
GatewayCACert: "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----\n",
Region: testRegion,
DBInstanceIdentifier: testDBID,
EngineVersion: "18",
EnginePort: 5432,
}
}
func TestBuildAgentUserData(t *testing.T) {
out := buildAgentUserData(testAgentUserDataInput())
require.True(t, strings.HasPrefix(out, "#cloud-config\n"), "cloud-init only parses a document with the header")
assert.Contains(t, out, " - path: "+agentEnvPath)
assert.Contains(t, out, " - path: "+agentGatewayCAPath)
assert.Contains(t, out, "RDS_GATEWAY_URL=https://172.30.0.1:9999")
assert.Contains(t, out, "RDS_GATEWAY_CA="+agentGatewayCAPath)
assert.Contains(t, out, "RDS_REGION="+testRegion)
assert.Contains(t, out, "RDS_DB_INSTANCE_IDENTIFIER="+testDBID)
assert.Contains(t, out, "RDS_ENGINE_VERSION=18")
assert.Contains(t, out, "RDS_ENGINE_PORT=5432")
// Every content line is indented under its block, or cloud-init reads the
// second line as a new key and drops the rest of the file.
for line := range strings.SplitSeq(strings.TrimRight(out, "\n"), "\n") {
if strings.HasPrefix(line, "#") || strings.HasPrefix(line, "write_files:") ||
strings.HasPrefix(line, " - path:") || strings.HasPrefix(line, " ") {
continue
}
t.Errorf("unindented line in the cloud-config document: %q", line)
}
}
// D8: the master password reaches the guest only over GetDBBootstrapConfig, and
// the agent's gateway credentials come from IMDS — so nothing secret is written
// to a data source any process on the VM can read.
func TestBuildAgentUserDataCarriesNoCredentials(t *testing.T) {
in := testAgentUserDataInput()
out := buildAgentUserData(in)
for _, secret := range []string{"AKIA", "SecretAccessKey", "aws_secret", "MasterUserPassword", "password"} {
assert.NotContains(t, strings.ToLower(out), strings.ToLower(secret))
}
}
// An empty CA file would be loaded by the agent as an empty trust store, which
// fails the pin in a way that looks like a certificate problem rather than a
// missing configuration.
func TestBuildAgentUserDataOmitsAnAbsentCA(t *testing.T) {
in := testAgentUserDataInput()
in.GatewayCACert = ""
out := buildAgentUserData(in)
assert.NotContains(t, out, " - path: "+agentGatewayCAPath)
assert.Contains(t, out, "RDS_GATEWAY_CA="+agentGatewayCAPath,
"the path stays configured so an image with a baked-in CA still finds it")
}
+233
View File
@@ -0,0 +1,233 @@
package handlers_rds
import (
"fmt"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/awserrors"
)
// AWS's PostgreSQL range. The upper bound is not a platform limit — it is the
// point past which the customer is asking for something no engine build here
// has been exercised at.
const (
minAllocatedStorageGiB = 20
maxAllocatedStorageGiB = 65536
// The only storage type offered. Every other AWS type names a performance
// class this platform does not implement, so accepting one would promise
// IOPS that are not delivered.
storageTypeGP3 = "gp3"
// The port range AWS accepts. Below 1150 collides with the well-known
// range the guest's own services use.
minDBPort = 1150
maxDBPort = 65535
maxDBInstanceIdentifierLen = 63
)
// The request as CreateDBInstance resolved it: defaults filled in, every
// unimplemented parameter already rejected.
type validatedCreate struct {
Identifier string
Engine Engine
EngineVersion string
InstanceClass string
InstanceType string
AllocatedStorage int64
StorageType string
Port int64
MasterUsername string
MasterPassword string
DBName string
SecurityGroupIDs []string
// Empty means "resolve the account's default VPC subnet"; rds-7 replaces
// this with a real DB subnet group lookup.
SubnetID string
DBParameterGroupName string
}
// Everything that can be decided from the request alone. Network resolution
// runs afterwards, so a malformed request never reaches the VPC.
func validateCreateRequest(input *rds.CreateDBInstanceInput) (*validatedCreate, error) {
if input == nil {
return nil, fmt.Errorf("%s: empty request", awserrors.ErrorInvalidParameterValue)
}
if err := rejectUnimplemented(input); err != nil {
return nil, err
}
identifier := aws.StringValue(input.DBInstanceIdentifier)
if err := validateDBInstanceIdentifier(identifier); err != nil {
return nil, err
}
engine, err := LookupEngine(aws.StringValue(input.Engine))
if err != nil {
return nil, err
}
if err := engine.ValidateVersion(aws.StringValue(input.EngineVersion)); err != nil {
return nil, err
}
instanceClass := aws.StringValue(input.DBInstanceClass)
instanceType, err := InstanceTypeForClass(instanceClass)
if err != nil {
return nil, fmt.Errorf("%s: DBInstanceClass %q is not supported; supported classes are %s",
awserrors.ErrorInvalidParameterValue, instanceClass, strings.Join(SupportedInstanceClasses(), ", "))
}
storage := aws.Int64Value(input.AllocatedStorage)
if storage < minAllocatedStorageGiB || storage > maxAllocatedStorageGiB {
return nil, fmt.Errorf("%s: AllocatedStorage must be between %d and %d GiB",
awserrors.ErrorInvalidParameterValue, minAllocatedStorageGiB, maxAllocatedStorageGiB)
}
storageType := strings.ToLower(strings.TrimSpace(aws.StringValue(input.StorageType)))
if storageType == "" {
storageType = storageTypeGP3
}
if storageType != storageTypeGP3 {
return nil, fmt.Errorf("%s: StorageType %q is not supported; only %q is offered",
awserrors.ErrorInvalidParameterValue, storageType, storageTypeGP3)
}
masterUsername := aws.StringValue(input.MasterUsername)
if err := engine.ValidateMasterUsername(masterUsername); err != nil {
return nil, err
}
masterPassword := aws.StringValue(input.MasterUserPassword)
if err := ValidateMasterUserPassword(masterPassword); err != nil {
return nil, err
}
port := engine.DefaultPort
if input.Port != nil {
port = aws.Int64Value(input.Port)
if port < minDBPort || port > maxDBPort {
return nil, fmt.Errorf("%s: Port must be between %d and %d",
awserrors.ErrorInvalidParameterValue, minDBPort, maxDBPort)
}
}
// No subnet groups exist until rds-7, so a named one cannot resolve. Saying
// so is the honest answer; silently placing the ENI somewhere else would put
// the endpoint in a subnet the customer did not choose.
if group := aws.StringValue(input.DBSubnetGroupName); group != "" {
return nil, fmt.Errorf("%s: DB subnet group %q not found", awserrors.ErrorDBSubnetGroupNotFound, group)
}
// The implicit default group is the one name that will resolve once rds-7
// materialises it, so accepting it now keeps a Terraform config that names
// it working across both phases.
paramGroup := aws.StringValue(input.DBParameterGroupName)
if paramGroup != "" && paramGroup != engine.DefaultParameterGroupName() {
return nil, fmt.Errorf("%s: DB parameter group %q not found", awserrors.ErrorDBParameterGroupNotFound, paramGroup)
}
return &validatedCreate{
Identifier: identifier,
Engine: engine,
EngineVersion: engine.EngineVersion(),
InstanceClass: instanceClass,
InstanceType: instanceType,
AllocatedStorage: storage,
StorageType: storageType,
Port: port,
MasterUsername: masterUsername,
MasterPassword: masterPassword,
DBName: aws.StringValue(input.DBName),
SecurityGroupIDs: aws.StringValueSlice(input.VpcSecurityGroupIds),
DBParameterGroupName: engine.DefaultParameterGroupName(),
}, nil
}
// AWS's own identifier rules. Enforcing them here keeps the identifier usable
// as a DNS label, which D6 makes it.
func validateDBInstanceIdentifier(id string) error {
switch {
case id == "":
return fmt.Errorf("%s: DBInstanceIdentifier is required", awserrors.ErrorInvalidParameterValue)
case len(id) > maxDBInstanceIdentifierLen:
return fmt.Errorf("%s: DBInstanceIdentifier must be at most %d characters",
awserrors.ErrorInvalidParameterValue, maxDBInstanceIdentifierLen)
case !isLetter(rune(id[0])):
return fmt.Errorf("%s: DBInstanceIdentifier must begin with a letter", awserrors.ErrorInvalidParameterValue)
case strings.HasSuffix(id, "-"):
return fmt.Errorf("%s: DBInstanceIdentifier may not end with a hyphen", awserrors.ErrorInvalidParameterValue)
case strings.Contains(id, "--"):
return fmt.Errorf("%s: DBInstanceIdentifier may not contain consecutive hyphens", awserrors.ErrorInvalidParameterValue)
}
for _, r := range id {
if !isDigit(r) && r != '-' && (r < 'a' || r > 'z') {
return fmt.Errorf("%s: DBInstanceIdentifier may contain only lowercase letters, digits and hyphens",
awserrors.ErrorInvalidParameterValue)
}
}
return nil
}
// D19: a supported action carrying an unimplemented parameter must not silently
// drop it. Each rejection below is a parameter whose omission would create a
// false safety, security or availability guarantee. Parameters that are merely
// inert — AutoMinorVersionUpgrade, Performance Insights, Enhanced Monitoring,
// CopyTagsToSnapshot — are deliberately absent and accepted as no-ops.
func rejectUnimplemented(input *rds.CreateDBInstanceInput) error {
if aws.BoolValue(input.MultiAZ) {
return unimplemented("MultiAZ", "this platform is single-AZ; a standby would not exist")
}
if aws.BoolValue(input.PubliclyAccessible) {
return unimplemented("PubliclyAccessible",
"the endpoint is a private VPC address; a public one would not be reachable")
}
// Rejected in both directions: false asks for unencrypted storage, which is
// not offered, and omitting it entirely still yields encrypted storage.
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")
}
if aws.StringValue(input.PreferredBackupWindow) != "" {
return unimplemented("PreferredBackupWindow", "automated backups are not implemented yet")
}
if aws.StringValue(input.PreferredMaintenanceWindow) != "" {
return unimplemented("PreferredMaintenanceWindow", "maintenance windows are not implemented yet")
}
if aws.BoolValue(input.EnableIAMDatabaseAuthentication) {
return unimplemented("EnableIAMDatabaseAuthentication", "IAM database authentication is not implemented")
}
if aws.Int64Value(input.Iops) > 0 {
return unimplemented("Iops", "provisioned IOPS are not implemented; storage is gp3")
}
if aws.StringValue(input.KmsKeyId) != "" {
return unimplemented("KmsKeyId", "storage is encrypted with the cluster key, not a customer-managed one")
}
if aws.StringValue(input.AvailabilityZone) != "" {
return unimplemented("AvailabilityZone", "this platform exposes a single zone")
}
if len(input.DBSecurityGroups) > 0 {
return unimplemented("DBSecurityGroups",
"EC2-Classic security groups are not offered; use VpcSecurityGroupIds")
}
if aws.StringValue(input.DBClusterIdentifier) != "" {
return unimplemented("DBClusterIdentifier", "clustered engines are not offered")
}
if len(input.EnableCloudwatchLogsExports) > 0 {
return unimplemented("EnableCloudwatchLogsExports", "log export is not implemented")
}
if len(input.Tags) > 0 {
return unimplemented("Tags", "resource tagging is not implemented yet")
}
return nil
}
func unimplemented(parameter, why string) error {
return fmt.Errorf("%s: %s is not supported: %s", awserrors.ErrorInvalidParameterValue, parameter, why)
}
+5 -2
View File
@@ -18,9 +18,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG))))
# named suite into BIN_DIR/<suite>.test so CI can ship binaries (not source +
# go toolchain) to the remote VM.
BIN_DIR ?= $(CURDIR)/_bin
SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth partialblock
SUITES_BUILD ?= cert lb eks ecs single iam ecr bedrock gpu natuplink multinode storagegrowth partialblock rds
.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-partialblock e2e-all clean e2e-build
.PHONY: e2e e2e-cert e2e-lb e2e-eks e2e-ecs e2e-single e2e-iam e2e-ecr e2e-bedrock e2e-multinode e2e-bm e2e-reboot e2e-tofu e2e-gpu e2e-natuplink e2e-storagegrowth e2e-partialblock e2e-rds e2e-all clean e2e-build
RUN_FLAG = $(if $(RUN),-run '$(RUN)',)
@@ -57,6 +57,9 @@ e2e-gpu: ; $(MAKE) e2e PKG=./gpu/...
e2e-natuplink: ; $(MAKE) e2e PKG=./natuplink/...
e2e-storagegrowth: ; $(MAKE) e2e PKG=./storagegrowth/...
e2e-partialblock: ; $(MAKE) e2e PKG=./partialblock/...
# The RDS suite leaves its DB instance behind until DeleteDBInstance lands, so it
# carries its own opt-in on top of SPINIFEX_E2E.
e2e-rds: ; SPINIFEX_E2E_RDS=1 $(MAKE) e2e PKG=./rds/...
e2e-all: ; $(MAKE) e2e
e2e-build:
+3
View File
@@ -23,6 +23,7 @@ import (
"github.com/aws/aws-sdk-go/service/eks"
"github.com/aws/aws-sdk-go/service/elbv2"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/aws/aws-sdk-go/service/sts"
)
@@ -41,6 +42,7 @@ type AWSClient struct {
ACM *acm.ACM
ECR *ecr.ECR
ECS *ecs.ECS
RDS *rds.RDS
Bedrock *bedrock.Bedrock
BedrockRuntime *bedrockruntime.BedrockRuntime
}
@@ -141,6 +143,7 @@ func newAWSClient(t *testing.T, env *Env, accessKey, secretKey, sessionToken str
ACM: acm.New(sess),
ECR: ecr.New(sess),
ECS: ecs.New(sess),
RDS: rds.New(sess),
Bedrock: bedrock.New(sess),
BedrockRuntime: bedrockruntime.New(sess),
}
+64
View File
@@ -0,0 +1,64 @@
//go:build e2e
package harness
import (
"fmt"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
)
// The statuses a DB instance reports on its way up, and the one it lands in when
// the bootstrap never completes.
const (
DBInstanceAvailable = "available"
DBInstanceCreating = "creating"
DBInstanceFailed = "failed"
)
// WaitForDBInstanceAvailable polls DescribeDBInstances until the instance
// reports available. Reaching it means the VM booted, the engine ran initdb and
// the in-guest agent reported healthy, so the default envelope is generous.
func WaitForDBInstanceAvailable(t *testing.T, c *AWSClient, id string, opts ...PollOpt) *rds.DBInstance {
t.Helper()
cfg := applyOpts(pollCfg{timeout: 20 * time.Minute, interval: 10 * time.Second}, opts...)
var last *rds.DBInstance
EventuallyErr(t, func() error {
instance, err := DescribeDBInstance(c, id)
if err != nil {
return fmt.Errorf("describe-db-instances %s: %w", id, err)
}
last = instance
switch status := aws.StringValue(instance.DBInstanceStatus); status {
case DBInstanceAvailable:
return nil
case DBInstanceFailed:
// Terminal: the control plane already gave up waiting for the engine,
// so polling on would only burn the rest of the timeout.
t.Fatalf("DB instance %s entered %s", id, DBInstanceFailed)
return nil
default:
return fmt.Errorf("%s status=%s want=%s", id, status, DBInstanceAvailable)
}
}, cfg.timeout, cfg.interval)
t.Logf("DB instance %s reached status %s", id, DBInstanceAvailable)
return last
}
// DescribeDBInstance returns the single named DB instance. A named instance that
// does not exist is an error from the API, not an empty list.
func DescribeDBInstance(c *AWSClient, id string) (*rds.DBInstance, error) {
out, err := c.RDS.DescribeDBInstances(&rds.DescribeDBInstancesInput{
DBInstanceIdentifier: aws.String(id),
})
if err != nil {
return nil, err
}
if len(out.DBInstances) != 1 {
return nil, fmt.Errorf("describe %s returned %d instances, want 1", id, len(out.DBInstances))
}
return out.DBInstances[0], nil
}
+172
View File
@@ -0,0 +1,172 @@
//go:build e2e
package rds
import (
"fmt"
"net"
"os/exec"
"strings"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/tests/e2e/harness"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
dbInstancePfx = "rds-e2e"
dbEngine = "postgres"
dbClass = "db.t3.medium"
dbStorageGiB = 20
dbName = "orders"
dbMasterUser = "appuser"
// No '/', '"', '@' or spaces: the characters the API rejects because they
// break a connection string or the engine's own role syntax.
dbMasterPassword = "e2eSup3rSecret1"
)
// TestCreateDescribeConnect drives the first live RDS path: CreateDBInstance
// (postgres, single-AZ) → available → an endpoint that resolves → a client that
// connects with the master credentials and writes a row.
//
// The instance is created once and shared by the subtests: booting the VM and
// running initdb is by far the slowest step.
func TestCreateDescribeConnect(t *testing.T) {
f := requireRDSFixture(t)
id := fmt.Sprintf("%s-%d", dbInstancePfx, time.Now().Unix())
harness.Phase(t, "Creating DB instance %q", id)
out, err := f.AWS.RDS.CreateDBInstance(&rds.CreateDBInstanceInput{ //nolint:staticcheck // e2e:allow-create — the instance under test
DBInstanceIdentifier: aws.String(id),
Engine: aws.String(dbEngine),
DBInstanceClass: aws.String(dbClass),
AllocatedStorage: aws.Int64(dbStorageGiB),
DBName: aws.String(dbName),
MasterUsername: aws.String(dbMasterUser),
MasterUserPassword: aws.String(dbMasterPassword),
})
require.NoError(t, err, "create-db-instance")
require.NotNil(t, out.DBInstance)
// The create returns before the engine exists, so the customer sees creating
// and polls; the reconciler owns the flip to available.
assert.Equal(t, harness.DBInstanceCreating, aws.StringValue(out.DBInstance.DBInstanceStatus))
assert.Equal(t, id, aws.StringValue(out.DBInstance.DBInstanceIdentifier))
t.Logf("DB instance %s left behind for manual teardown: DeleteDBInstance is not implemented yet", id)
var instance *rds.DBInstance
t.Run("BecomesAvailable", func(t *testing.T) {
instance = harness.WaitForDBInstanceAvailable(t, f.AWS, id)
assert.Equal(t, dbEngine, aws.StringValue(instance.Engine))
assert.Equal(t, dbClass, aws.StringValue(instance.DBInstanceClass))
assert.Equal(t, int64(dbStorageGiB), aws.Int64Value(instance.AllocatedStorage))
assert.Equal(t, dbMasterUser, aws.StringValue(instance.MasterUsername))
assert.True(t, aws.BoolValue(instance.StorageEncrypted), "the data volume is encrypted with the cluster key")
assert.False(t, aws.BoolValue(instance.MultiAZ))
assert.False(t, aws.BoolValue(instance.PubliclyAccessible))
require.NotNil(t, instance.Endpoint, "an available instance must publish an endpoint")
assert.NotEmpty(t, aws.StringValue(instance.Endpoint.Address))
assert.Equal(t, int64(5432), aws.Int64Value(instance.Endpoint.Port))
})
t.Run("AppearsInTheFleetListing", func(t *testing.T) {
requireAvailable(t, instance)
list, err := f.AWS.RDS.DescribeDBInstances(&rds.DescribeDBInstancesInput{})
require.NoError(t, err, "describe-db-instances")
ids := make([]string, 0, len(list.DBInstances))
for _, i := range list.DBInstances {
ids = append(ids, aws.StringValue(i.DBInstanceIdentifier))
}
assert.Contains(t, ids, id, "an unfiltered describe must list the instance")
})
t.Run("EndpointResolves", func(t *testing.T) {
requireAvailable(t, instance)
host := aws.StringValue(instance.Endpoint.Address)
// Without northstar the endpoint is the bare ENI IP, which is a valid
// endpoint and needs no resolution.
if ip := net.ParseIP(host); ip != nil {
require.Empty(t, f.BaseDomain, "with northstar configured the endpoint must be a hostname, got %s", host)
t.Logf("endpoint is the bare ENI IP %s (no base domain configured)", host)
return
}
assert.Equal(t, fmt.Sprintf("%s.%s.%s.rds.%s", id, f.Account, f.Region, f.BaseDomain), host,
"the endpoint name is account-qualified so identifiers collide across tenants without colliding in DNS")
assertResolves(t, host)
})
t.Run("AcceptsAClientConnection", func(t *testing.T) {
requireAvailable(t, instance)
runSQLSmokeTest(t, instance)
})
}
func requireAvailable(t *testing.T, instance *rds.DBInstance) {
t.Helper()
if instance == nil {
t.Skip("DB instance never reached available (BecomesAvailable failed)")
}
}
func assertResolves(t *testing.T, host string) {
t.Helper()
// The record is published as soon as the ENI IP is known, well before the
// instance is available, so a short envelope only covers zone propagation.
harness.EventuallyErr(t, func() error {
addrs, err := net.LookupHost(host)
if err != nil {
return fmt.Errorf("lookup %s: %w", host, err)
}
if len(addrs) == 0 {
return fmt.Errorf("lookup %s returned no addresses", host)
}
t.Logf("endpoint %s resolved to %v", host, addrs)
return nil
}, 90*time.Second, 3*time.Second)
}
// runSQLSmokeTest connects with the master credentials and writes a row, which
// is the only assertion that proves the engine actually bootstrapped rather than
// the agent merely reporting that it had. Shells to psql rather than linking a
// driver so the suite adds no production dependency; skips when psql is absent.
func runSQLSmokeTest(t *testing.T, instance *rds.DBInstance) {
t.Helper()
psql, err := exec.LookPath("psql")
if err != nil {
t.Skip("psql not on PATH; skipping the client connection leg")
}
host := aws.StringValue(instance.Endpoint.Address)
port := aws.Int64Value(instance.Endpoint.Port)
table := "e2e_smoke"
statements := strings.Join([]string{
fmt.Sprintf("DROP TABLE IF EXISTS %s;", table),
fmt.Sprintf("CREATE TABLE %s (id int primary key, note text);", table),
fmt.Sprintf("INSERT INTO %s VALUES (1, 'hello from e2e');", table),
fmt.Sprintf("SELECT note FROM %s WHERE id = 1;", table),
}, " ")
// TLS is offered but not enforced (D14), so the smoke test does not pin
// verify-full — the cert chain is the cert suite's concern.
cmd := exec.Command(psql, //nolint:gosec // psql is LookPath-resolved, args test-controlled
"--no-psqlrc", "--quiet", "--tuples-only", "--no-align",
"--set", "ON_ERROR_STOP=1",
"--host", host, "--port", fmt.Sprint(port),
"--username", dbMasterUser, "--dbname", dbName,
"--command", statements,
)
cmd.Env = append(cmd.Environ(), "PGPASSWORD="+dbMasterPassword, "PGCONNECT_TIMEOUT=30")
output, err := cmd.CombinedOutput()
require.NoErrorf(t, err, "psql against %s:%d: %s", host, port, output)
assert.Contains(t, string(output), "hello from e2e", "the row written over the endpoint must read back")
}
+71
View File
@@ -0,0 +1,71 @@
//go:build e2e
// Package rds is the RDS E2E suite: the create/describe/connect path against a
// running cluster. The control-plane orchestration, the validation matrix and
// the reconciler's status transitions are covered by the handlers/rds unit
// tests; what only a live cluster can prove is that a DB VM actually boots, the
// in-guest agent reports healthy, the endpoint resolves and a client can speak
// the wire protocol to it.
//
// The suite is opt-in beyond SPINIFEX_E2E: DeleteDBInstance is not implemented
// yet, so a run leaves its DB instance and VM behind. Set SPINIFEX_E2E_RDS=1 to
// accept that on a cluster you can rebuild.
package rds
import (
"os"
"sync"
"testing"
"github.com/mulgadc/spinifex/tests/e2e/harness"
)
// Mirrors the AWS client's own default, since the endpoint name the control
// plane publishes is region-qualified.
const defaultRegion = "ap-southeast-2"
var (
pkgFixOnce sync.Once
pkgFix *Fixture
)
// Fixture carries per-process state shared across every Test* in this package.
type Fixture struct {
Env *harness.Env
AWS *harness.AWSClient
Account string
Region string
BaseDomain string
}
func TestMain(m *testing.M) {
os.Exit(m.Run())
}
// requireRDSFixture returns the package-scoped Fixture, building it on first
// call. Skips the calling test unless both gates are set.
func requireRDSFixture(t *testing.T) *Fixture {
t.Helper()
pkgFixOnce.Do(func() {
if os.Getenv("SPINIFEX_E2E") == "" || os.Getenv("SPINIFEX_E2E_RDS") != "1" {
return
}
env := harness.LoadEnv(t)
awsCli := harness.NewAWSClient(t, env)
region := os.Getenv("SPINIFEX_AWS_REGION")
if region == "" {
region = defaultRegion
}
pkgFix = &Fixture{
Env: env,
AWS: awsCli,
Account: harness.IAMAccountID(t, awsCli),
Region: region,
BaseDomain: harness.NorthstarBaseDomain(env),
}
})
if pkgFix == nil {
t.Skip("rds fixture unavailable (SPINIFEX_E2E unset, or SPINIFEX_E2E_RDS != 1 — the suite leaks a DB VM until DeleteDBInstance lands)")
}
return pkgFix
}