diff --git a/docs/service-interfaces.yaml b/docs/service-interfaces.yaml index e12744e7f..cc9be7c9c 100644 --- a/docs/service-interfaces.yaml +++ b/docs/service-interfaces.yaml @@ -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] diff --git a/spinifex/awserrors/awserrors.go b/spinifex/awserrors/awserrors.go index 85b40a3dc..5a3c34881 100644 --- a/spinifex/awserrors/awserrors.go +++ b/spinifex/awserrors/awserrors.go @@ -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."}, diff --git a/spinifex/config/config.go b/spinifex/config/config.go index 47e7a5f74..dd91545dc 100644 --- a/spinifex/config/config.go +++ b/spinifex/config/config.go @@ -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 diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index cfd9d0a1e..6d5cf7bfb 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -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) { diff --git a/spinifex/daemon/dns_reconcile.go b/spinifex/daemon/dns_reconcile.go index cf8ac8293..a270ce15d 100644 --- a/spinifex/daemon/dns_reconcile.go +++ b/spinifex/daemon/dns_reconcile.go @@ -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 } diff --git a/spinifex/daemon/rds_deps.go b/spinifex/daemon/rds_deps.go index c619d89f6..c9d8ff133 100644 --- a/spinifex/daemon/rds_deps.go +++ b/spinifex/daemon/rds_deps.go @@ -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) +} diff --git a/spinifex/gateway/rds/agent.go b/spinifex/gateway/rds/agent.go index 98617a6eb..b1e49ecb7 100644 --- a/spinifex/gateway/rds/agent.go +++ b/spinifex/gateway/rds/agent.go @@ -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) diff --git a/spinifex/gateway/rds/agent_test.go b/spinifex/gateway/rds/agent_test.go index e3d5f8742..14a28ab51 100644 --- a/spinifex/gateway/rds/agent_test.go +++ b/spinifex/gateway/rds/agent_test.go @@ -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) { diff --git a/spinifex/gateway/rds/handler.go b/spinifex/gateway/rds/handler.go index eaafc1fc2..c3cc9dd37 100644 --- a/spinifex/gateway/rds/handler.go +++ b/spinifex/gateway/rds/handler.go @@ -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(), diff --git a/spinifex/gateway/rds/handler_test.go b/spinifex/gateway/rds/handler_test.go index 988effa88..61418cf10 100644 --- a/spinifex/gateway/rds/handler_test.go +++ b/spinifex/gateway/rds/handler_test.go @@ -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), " 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' +} diff --git a/spinifex/handlers/rds/engine_postgres.go b/spinifex/handlers/rds/engine_postgres.go new file mode 100644 index 000000000..8c11944bb --- /dev/null +++ b/spinifex/handlers/rds/engine_postgres.go @@ -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_"}, +} diff --git a/spinifex/handlers/rds/engine_test.go b/spinifex/handlers/rds/engine_test.go new file mode 100644 index 000000000..8579fae15 --- /dev/null +++ b/spinifex/handlers/rds/engine_test.go @@ -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) + } +} diff --git a/spinifex/handlers/rds/iam.go b/spinifex/handlers/rds/iam.go new file mode 100644 index 000000000..c0ba95d09 --- /dev/null +++ b/spinifex/handlers/rds/iam.go @@ -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 +} diff --git a/spinifex/handlers/rds/launch.go b/spinifex/handlers/rds/launch.go index c1666f421..2da5b9a0b 100644 --- a/spinifex/handlers/rds/launch.go +++ b/spinifex/handlers/rds/launch.go @@ -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 } diff --git a/spinifex/handlers/rds/launch_test.go b/spinifex/handlers/rds/launch_test.go index fefa28898..febe28142 100644 --- a/spinifex/handlers/rds/launch_test.go +++ b/spinifex/handlers/rds/launch_test.go @@ -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) { diff --git a/spinifex/handlers/rds/network.go b/spinifex/handlers/rds/network.go new file mode 100644 index 000000000..eb1dfb76f --- /dev/null +++ b/spinifex/handlers/rds/network.go @@ -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 +} diff --git a/spinifex/handlers/rds/reconciler.go b/spinifex/handlers/rds/reconciler.go new file mode 100644 index 000000000..ebb1e46ae --- /dev/null +++ b/spinifex/handlers/rds/reconciler.go @@ -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 +} diff --git a/spinifex/handlers/rds/reconciler_test.go b/spinifex/handlers/rds/reconciler_test.go new file mode 100644 index 000000000..119ae2443 --- /dev/null +++ b/spinifex/handlers/rds/reconciler_test.go @@ -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") +} diff --git a/spinifex/handlers/rds/records.go b/spinifex/handlers/rds/records.go index 661f53d55..f90bbd1e2 100644 --- a/spinifex/handlers/rds/records.go +++ b/spinifex/handlers/rds/records.go @@ -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"` diff --git a/spinifex/handlers/rds/service.go b/spinifex/handlers/rds/service.go index bc7644f9d..6eb419351 100644 --- a/spinifex/handlers/rds/service.go +++ b/spinifex/handlers/rds/service.go @@ -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) diff --git a/spinifex/handlers/rds/service_nats.go b/spinifex/handlers/rds/service_nats.go index ad6bc766a..b6cfaf65b 100644 --- a/spinifex/handlers/rds/service_nats.go +++ b/spinifex/handlers/rds/service_nats.go @@ -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, diff --git a/spinifex/handlers/rds/store.go b/spinifex/handlers/rds/store.go index 5dbbb2f38..30c5db489 100644 --- a/spinifex/handlers/rds/store.go +++ b/spinifex/handlers/rds/store.go @@ -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 +} diff --git a/spinifex/handlers/rds/subjects.go b/spinifex/handlers/rds/subjects.go index 8444475c4..8a10eacef 100644 --- a/spinifex/handlers/rds/subjects.go +++ b/spinifex/handlers/rds/subjects.go @@ -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" diff --git a/spinifex/handlers/rds/userdata.go b/spinifex/handlers/rds/userdata.go new file mode 100644 index 000000000..384a8f018 --- /dev/null +++ b/spinifex/handlers/rds/userdata.go @@ -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() +} diff --git a/spinifex/handlers/rds/userdata_test.go b/spinifex/handlers/rds/userdata_test.go new file mode 100644 index 000000000..8114aa35b --- /dev/null +++ b/spinifex/handlers/rds/userdata_test.go @@ -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") +} diff --git a/spinifex/handlers/rds/validate.go b/spinifex/handlers/rds/validate.go new file mode 100644 index 000000000..a608775b8 --- /dev/null +++ b/spinifex/handlers/rds/validate.go @@ -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) +} diff --git a/tests/e2e/Makefile b/tests/e2e/Makefile index 23c99cb77..f33d8f911 100644 --- a/tests/e2e/Makefile +++ b/tests/e2e/Makefile @@ -18,9 +18,9 @@ SUITE_TAG ?= $(notdir $(patsubst %/...,%,$(firstword $(PKG)))) # named suite into BIN_DIR/.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: diff --git a/tests/e2e/harness/aws.go b/tests/e2e/harness/aws.go index 5aa807dfc..c70938243 100644 --- a/tests/e2e/harness/aws.go +++ b/tests/e2e/harness/aws.go @@ -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), } diff --git a/tests/e2e/harness/rds.go b/tests/e2e/harness/rds.go new file mode 100644 index 000000000..14e9d6543 --- /dev/null +++ b/tests/e2e/harness/rds.go @@ -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 +} diff --git a/tests/e2e/rds/create_describe_test.go b/tests/e2e/rds/create_describe_test.go new file mode 100644 index 000000000..62310a729 --- /dev/null +++ b/tests/e2e/rds/create_describe_test.go @@ -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") +} diff --git a/tests/e2e/rds/main_test.go b/tests/e2e/rds/main_test.go new file mode 100644 index 000000000..e41f41f16 --- /dev/null +++ b/tests/e2e/rds/main_test.go @@ -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 +}