refactor: extract the RDS DB subnet group lifecycle owner

domains/rds/subnetgroup owns the db-subnet-groups/{name} record, member
subnet validation and the delete in-use guard. handlers/rds keeps AWS
validation, projection and paging and binds two consumer-owned
capabilities: the owner's Dependants (instances naming the group) and the
instance side's subnetGroupPlacement. Stored shape and behaviour are
unchanged.
This commit is contained in:
joshsiv-mulga
2026-10-08 10:03:01 +10:00
parent 56384f9046
commit 17dc29251d
14 changed files with 512 additions and 273 deletions
+1
View File
@@ -75,6 +75,7 @@ services:
- spinifex/domains/acm
- spinifex/handlers/iam
- spinifex/handlers/rds
- spinifex/domains/rds/subnetgroup
- spinifex/domains/ochre/vector
subscribes:
# EC2 instance lifecycle.
+212
View File
@@ -0,0 +1,212 @@
package subnetgroup
import (
"context"
"errors"
"fmt"
"log/slog"
"slices"
"strings"
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// How many times a modify replays against a fresh record after losing its CAS.
const modifyAttempts = 16
// Subnets is the EC2 read the group validates its members through, issued as the caller.
type Subnets interface {
DescribeSubnets(ctx context.Context, input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error)
}
// Dependants reports the sorted identifiers of the DB instances in the account that name the group.
type Dependants interface {
InstancesUsing(ctx context.Context, accountID, name string) ([]string, error)
}
// Bucket opens the account's RDS bucket, which holds the group's records.
type Bucket func(ctx context.Context, accountID string) (*kvstore.Bucket, error)
// Owner is the sole writer of DB subnet group records.
type Owner struct {
subnets Subnets
dependants Dependants
bucket Bucket
}
// New returns an Owner. A nil subnets refuses every create and modify, as on a node without RDS networking.
func New(subnets Subnets, dependants Dependants, bucket Bucket) *Owner {
return &Owner{subnets: subnets, dependants: dependants, bucket: bucket}
}
// Spec is a validated create request.
type Spec struct {
Name string
Description string
SubnetIDs []string
Tags map[string]string
}
// Create stores a new group, or fails with DBSubnetGroupAlreadyExists when the name is taken.
func (o *Owner) Create(ctx context.Context, accountID string, spec Spec) (*Record, error) {
subnets, vpcID, err := o.resolveSubnets(ctx, accountID, spec.SubnetIDs)
if err != nil {
return nil, err
}
kv, err := o.bucket(ctx, accountID)
if err != nil {
return nil, err
}
now := time.Now().UTC()
rec := &Record{
Name: spec.Name,
AccountID: accountID,
Description: spec.Description,
Subnets: subnets,
VpcID: vpcID,
Tags: spec.Tags,
CreatedAt: now,
UpdatedAt: now,
}
// Create rather than Put, so two concurrent creates of one name have exactly
// one winner instead of both reporting success over each other's subnets.
if _, err := kvstore.On[Record](kv).Create(ctx, Key(spec.Name), rec); err != nil {
if errors.Is(err, kvstore.ErrExists) {
return nil, awserrors.Errorf(awserrors.ErrorDBSubnetGroupAlreadyExists,
"DB subnet group %s already exists", spec.Name)
}
return nil, err
}
slog.InfoContext(ctx, "rds: DB subnet group created",
"dbSubnetGroup", spec.Name, "accountId", accountID, "vpcId", vpcID, "subnets", len(subnets))
return rec, nil
}
// Get returns the group, or DBSubnetGroupNotFoundFault.
func (o *Owner) Get(ctx context.Context, accountID, name string) (*Record, error) {
kv, err := o.bucket(ctx, accountID)
if err != nil {
return nil, err
}
rec, _, err := get(ctx, kv, name)
return rec, err
}
// List returns every group in the account, sorted by name.
func (o *Owner) List(ctx context.Context, accountID string) ([]Record, error) {
kv, err := o.bucket(ctx, accountID)
if err != nil {
return nil, err
}
recs, err := kvstore.On[Record](kv).List(ctx, Prefix())
if err != nil {
return nil, err
}
slices.SortFunc(recs, func(a, b Record) int { return strings.Compare(a.Name, b.Name) })
return recs, nil
}
// Placement returns the VPC and member subnet IDs an instance created against the group may land in.
func (o *Owner) Placement(ctx context.Context, accountID, name string) (string, []string, error) {
rec, err := o.Get(ctx, accountID, name)
if err != nil {
return "", nil, err
}
ids := make([]string, 0, len(rec.Subnets))
for _, subnet := range rec.Subnets {
ids = append(ids, subnet.SubnetID)
}
return rec.VpcID, ids, nil
}
// Modify replaces the group's subnet set, which must stay in the group's VPC, as AWS requires. An
// empty description keeps the stored one. A lost CAS replays against the fresh record, so a
// concurrent write is not undone.
func (o *Owner) Modify(ctx context.Context, accountID, name, description string, subnetIDs []string) (*Record, error) {
kv, err := o.bucket(ctx, accountID)
if err != nil {
return nil, err
}
// Read first so a missing group is reported as such rather than as whatever
// the subnet resolution would have objected to.
if _, _, err := get(ctx, kv, name); err != nil {
return nil, err
}
subnets, vpcID, err := o.resolveSubnets(ctx, accountID, subnetIDs)
if err != nil {
return nil, err
}
key := Key(name)
for range modifyAttempts {
rec, rev, err := get(ctx, kv, name)
if err != nil {
return nil, err
}
if vpcID != rec.VpcID {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"The new Subnets are not in the same Vpc as the existing subnet group")
}
if description != "" {
rec.Description = description
}
rec.Subnets = subnets
rec.UpdatedAt = time.Now().UTC()
_, err = kvstore.On[Record](kv).CompareAndSet(ctx, key, rec, rev)
if err == nil {
slog.InfoContext(ctx, "rds: DB subnet group modified",
"dbSubnetGroup", name, "accountId", accountID, "vpcId", vpcID, "subnets", len(subnets))
return rec, nil
}
if !errors.Is(err, kvstore.ErrConflict) {
return nil, err
}
}
return nil, fmt.Errorf("rds: update of %s contended after %d attempts", key, modifyAttempts)
}
// Delete is refused while any instance still names the group, including one that is only deleting:
// releasing it early would let a teardown lose the record of where its ENI was placed, and would
// make destroy ordering ambiguous.
func (o *Owner) Delete(ctx context.Context, accountID, name string) error {
kv, err := o.bucket(ctx, accountID)
if err != nil {
return err
}
if _, _, err := get(ctx, kv, name); err != nil {
return err
}
users, err := o.dependants.InstancesUsing(ctx, accountID, name)
if err != nil {
return err
}
if len(users) > 0 {
return awserrors.Errorf(awserrors.ErrorDBSubnetGroupInvalidState,
"DB subnet group %s is still used by %s", name, strings.Join(users, ", "))
}
if err := kv.Delete(ctx, Key(name)); err != nil {
return fmt.Errorf("rds: delete DB subnet group %s: %w", name, err)
}
slog.InfoContext(ctx, "rds: DB subnet group deleted", "dbSubnetGroup", name, "accountId", accountID)
return nil
}
// The record plus its revision. A missing group raises AWS's own fault, so a
// well-formed name that resolves to nothing is distinguishable from a bad one.
func get(ctx context.Context, kv *kvstore.Bucket, name string) (*Record, uint64, error) {
rec, rev, err := kvstore.On[Record](kv).Get(ctx, Key(name))
if errors.Is(err, kvstore.ErrNotFound) {
return nil, 0, awserrors.Errorf(awserrors.ErrorDBSubnetGroupNotFound, "DB subnet group %s not found", name)
}
if err != nil {
return nil, 0, err
}
return rec, rev, nil
}
@@ -0,0 +1,101 @@
package subnetgroup
import (
"context"
"errors"
"strings"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
testAccount = "123456789012"
testGroup = "db-private"
)
type fakeSubnets struct{}
func (fakeSubnets) DescribeSubnets(_ context.Context, input *ec2.DescribeSubnetsInput, _ string) (*ec2.DescribeSubnetsOutput, error) {
out := &ec2.DescribeSubnetsOutput{}
for _, id := range aws.StringValueSlice(input.SubnetIds) {
out.Subnets = append(out.Subnets, &ec2.Subnet{SubnetId: aws.String(id), VpcId: aws.String("vpc-a")})
}
return out, nil
}
type fakeDependants struct {
users []string
err error
}
func (f fakeDependants) InstancesUsing(context.Context, string, string) ([]string, error) {
return f.users, f.err
}
func newOwner(t *testing.T, subnets Subnets, dependants Dependants) *Owner {
t.Helper()
_, _, js := testutil.StartTestJetStream(t)
b := kvstore.NewBucket(js, kvstore.Config{Name: "rds-account-" + testAccount, History: 1})
return New(subnets, dependants, func(context.Context, string) (*kvstore.Bucket, error) { return b, nil })
}
func TestKey_SitsUnderPrefix(t *testing.T) {
t.Parallel()
assert.Equal(t, "db-subnet-groups/prod-db-subnets", Key("prod-db-subnets"))
assert.True(t, strings.HasPrefix(Key("x"), Prefix()))
}
// The in-use guard fails closed: when the instance side cannot answer, the
// group stays rather than being deleted on an unknown answer.
func TestDelete_KeepsTheGroupWhenTheInUseCheckFails(t *testing.T) {
t.Parallel()
owner := newOwner(t, fakeSubnets{}, fakeDependants{err: errors.New("instance scan failed")})
_, err := owner.Create(t.Context(), testAccount, Spec{Name: testGroup, Description: "d", SubnetIDs: []string{"subnet-a"}})
require.NoError(t, err)
require.ErrorContains(t, owner.Delete(t.Context(), testAccount, testGroup), "instance scan failed")
_, err = owner.Get(t.Context(), testAccount, testGroup)
require.NoError(t, err)
}
func TestDelete_RefusesWhileDependantsNameTheGroup(t *testing.T) {
t.Parallel()
owner := newOwner(t, fakeSubnets{}, fakeDependants{users: []string{"orders-db", "users-db"}})
_, err := owner.Create(t.Context(), testAccount, Spec{Name: testGroup, Description: "d", SubnetIDs: []string{"subnet-a"}})
require.NoError(t, err)
err = owner.Delete(t.Context(), testAccount, testGroup)
assert.Equal(t, awserrors.ErrorDBSubnetGroupInvalidState, awserrors.ValidErrorCodeFromError(err))
assert.ErrorContains(t, err, "orders-db, users-db")
}
// A node without RDS networking cannot validate members, so it stores nothing.
func TestCreate_RefusesWithoutSubnetResolution(t *testing.T) {
t.Parallel()
owner := newOwner(t, nil, fakeDependants{})
_, err := owner.Create(t.Context(), testAccount, Spec{Name: testGroup, Description: "d", SubnetIDs: []string{"subnet-a"}})
assert.Equal(t, awserrors.ErrorServerInternal, awserrors.ValidErrorCodeFromError(err))
_, err = owner.Get(t.Context(), testAccount, testGroup)
assert.Equal(t, awserrors.ErrorDBSubnetGroupNotFound, awserrors.ValidErrorCodeFromError(err))
}
// Placement reports member subnets in request order; choosing among them is the instance's job.
func TestPlacement_ReportsTheVPCAndMembers(t *testing.T) {
t.Parallel()
owner := newOwner(t, fakeSubnets{}, fakeDependants{})
_, err := owner.Create(t.Context(), testAccount, Spec{Name: testGroup, Description: "d", SubnetIDs: []string{"subnet-z", "subnet-a"}})
require.NoError(t, err)
vpcID, ids, err := owner.Placement(t.Context(), testAccount, testGroup)
require.NoError(t, err)
assert.Equal(t, "vpc-a", vpcID)
assert.Equal(t, []string{"subnet-z", "subnet-a"}, ids)
}
@@ -0,0 +1,48 @@
// Package subnetgroup owns the RDS DB subnet group: its durable record, the
// validation of its member subnets, and the in-use guard on its delete.
package subnetgroup
import "time"
const prefix = "db-subnet-groups/"
// Prefix returns the per-account key prefix under which each DB subnet group record lives.
func Prefix() string {
return prefix
}
// Key returns the per-account KV key of a DB subnet group's record.
func Key(name string) string {
return prefix + name
}
// Record is the db-subnet-groups/{name} record. The subnet list is stored verbatim rather than
// reduced to a placement, so when V2 makes AZs real the group needs no migration — only the code
// that chooses among its subnets changes.
type Record struct {
Name string `json:"name"`
AccountID string `json:"accountId"`
Description string `json:"description"`
// Every subnet the customer supplied, in request order, each with the AZ
// recorded on the subnet itself rather than a hardcoded zone.
Subnets []Subnet `json:"subnets"`
// The one VPC they all share, which is what makes the group usable for a
// placement at all.
VpcID string `json:"vpcId"`
Tags map[string]string `json:"tags,omitempty"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
// Subnet is one subnet of a DB subnet group, with the AZ recorded on the subnet itself;
// AvailabilityZone is empty when the subnet carries none.
type Subnet struct {
SubnetID string `json:"subnetId"`
AvailabilityZone string `json:"availabilityZone,omitempty"`
}
func (r *Record) GetTags() map[string]string { return r.Tags }
func (r *Record) SetTags(tags map[string]string) { r.Tags = tags }
@@ -0,0 +1,82 @@
package subnetgroup
import (
"context"
"fmt"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
// AWS's cap on the subnets one group may hold.
const maxSubnets = 20
// Validates the requested subnets against the caller's own account and returns
// them with the AZ each subnet itself records. The describe is issued as the
// caller, so a subnet in another account simply does not come back and is
// reported as not found rather than leaked.
func (o *Owner) resolveSubnets(ctx context.Context, accountID string, requested []string) ([]Subnet, string, error) {
if o.subnets == nil {
return nil, "", awserrors.Errorf(awserrors.ErrorServerInternal, "RDS networking is not wired on this node")
}
switch {
case len(requested) == 0:
return nil, "", awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"SubnetIds must name at least one subnet")
case len(requested) > maxSubnets:
return nil, "", awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"a DB subnet group may hold at most %d subnets, got %d", maxSubnets, len(requested))
}
out, err := o.subnets.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
SubnetIds: aws.StringSlice(requested),
}, accountID)
if err != nil {
return nil, "", fmt.Errorf("rds: describe the subnets of the DB subnet group: %w", err)
}
found := map[string]*ec2.Subnet{}
if out != nil {
for _, subnet := range out.Subnets {
if id := aws.StringValue(subnet.SubnetId); id != "" {
found[id] = subnet
}
}
}
vpcID := ""
subnets := make([]Subnet, 0, len(requested))
seen := make(map[string]bool, len(requested))
for _, id := range requested {
if seen[id] {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid, "subnet %s is named more than once", id)
}
seen[id] = true
subnet, ok := found[id]
if !ok {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid,
"subnet %s does not exist in this account", id)
}
subnetVPC := aws.StringValue(subnet.VpcId)
if vpcID == "" {
vpcID = subnetVPC
}
if subnetVPC != vpcID {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid,
"subnet %s is in VPC %s while the group's other subnets are in %s; "+
"a DB subnet group must span one VPC", id, subnetVPC, vpcID)
}
// Read off the subnet rather than stamped from the platform's single zone,
// so the response needs no change when V2 makes AZs real.
subnets = append(subnets, Subnet{
SubnetID: id,
AvailabilityZone: aws.StringValue(subnet.AvailabilityZone),
})
}
if vpcID == "" {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid, "the named subnets report no VPC")
}
return subnets, vpcID, nil
}
+1 -1
View File
@@ -34,7 +34,7 @@ func (s *Service) CreateDBInstance(ctx context.Context, input *rds.CreateDBInsta
if err != nil {
return nil, err
}
placement, err := s.resolvePlacement(ctx, kv, accountID, req)
placement, err := s.resolvePlacement(ctx, accountID, req)
if err != nil {
return nil, err
}
+20 -11
View File
@@ -7,8 +7,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/domains/rds/subnetgroup"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
// The customer-account VPC surface CreateDBInstance reads to place the endpoint
@@ -19,6 +19,14 @@ type networkResolver interface {
DescribeSecurityGroups(ctx context.Context, input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error)
}
// The subnet group's answer to where an endpoint may land: its VPC and member
// subnet IDs, or the group's own not-found fault.
type subnetGroupPlacement interface {
Placement(ctx context.Context, accountID, name string) (vpcID string, subnetIDs []string, err error)
}
var _ subnetGroupPlacement = (*subnetgroup.Owner)(nil)
// Where the customer-facing ENI lands.
type endpointPlacement struct {
VpcID string
@@ -33,22 +41,23 @@ type endpointPlacement struct {
// back to the account's default VPC, mirroring AWS's own behaviour. The security
// groups are validated against whichever VPC that resolved to, because an ENI
// cannot carry a group from another one.
func (s *Service) resolvePlacement(ctx context.Context, kv *kvstore.Bucket, accountID string, req *validatedCreate) (*endpointPlacement, error) {
func (s *Service) resolvePlacement(ctx context.Context, accountID string, req *validatedCreate) (*endpointPlacement, error) {
if s.deps.Network == nil {
return nil, awserrors.Errorf(awserrors.ErrorServerInternal, "RDS networking is not wired on this node")
}
var vpcID, subnetID string
if req.DBSubnetGroupName != "" {
group, _, err := getDBSubnetGroup(ctx, kv, req.DBSubnetGroupName)
var groups subnetGroupPlacement = s.subnetGroups()
groupVPC, groupSubnets, err := groups.Placement(ctx, accountID, req.DBSubnetGroupName)
if err != nil {
return nil, err
}
subnetID, err = subnetFromGroup(group)
subnetID, err = subnetFromGroup(req.DBSubnetGroupName, groupSubnets)
if err != nil {
return nil, err
}
vpcID = group.VpcID
vpcID = groupVPC
} else {
var err error
if vpcID, err = s.defaultVPCID(ctx, accountID); err != nil {
@@ -98,16 +107,16 @@ func (s *Service) vpcCIDR(ctx context.Context, accountID, vpcID string) (string,
// two instances created against the same group land the same way regardless of
// the order the subnets were supplied in. Single-AZ makes the choice arbitrary
// today; when V2 makes AZs real this is the one function that changes.
func subnetFromGroup(group *DBSubnetGroupRecord) (string, error) {
ids := make([]string, 0, len(group.Subnets))
for _, subnet := range group.Subnets {
if subnet.SubnetID != "" {
ids = append(ids, subnet.SubnetID)
func subnetFromGroup(name string, subnetIDs []string) (string, error) {
ids := make([]string, 0, len(subnetIDs))
for _, id := range subnetIDs {
if id != "" {
ids = append(ids, id)
}
}
if len(ids) == 0 {
return "", awserrors.Errorf(awserrors.ErrorDBInvalidVPCNetworkState,
"DB subnet group %s holds no subnet to place the DB endpoint in", group.Name)
"DB subnet group %s holds no subnet to place the DB endpoint in", name)
}
slices.Sort(ids)
return ids[0], nil
-33
View File
@@ -378,39 +378,6 @@ type Parameter struct {
Value string `json:"value" locationName:"Value"`
}
// DBSubnetGroupRecord is the db-subnet-groups/{name} record. The subnet list is stored verbatim rather
// than reduced to a placement, so when V2 makes AZs real the group needs no migration — only the code
// that chooses among its subnets changes.
type DBSubnetGroupRecord struct {
Name string `json:"name"`
AccountID string `json:"accountId"`
Description string `json:"description"`
// Every subnet the customer supplied, in request order, each with the AZ
// recorded on the subnet itself rather than a hardcoded zone.
Subnets []DBSubnetGroupSubnet `json:"subnets"`
// The one VPC they all share, which is what makes the group usable for a
// placement at all.
VpcID string `json:"vpcId"`
Tags map[string]string `json:"tags,omitempty"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
// DBSubnetGroupSubnet is one subnet of a DB subnet group, with the AZ recorded on the subnet itself;
// AvailabilityZone is empty when the subnet carries none.
type DBSubnetGroupSubnet struct {
SubnetID string `json:"subnetId"`
AvailabilityZone string `json:"availabilityZone,omitempty"`
}
var _ TaggedRecord = (*DBSubnetGroupRecord)(nil)
func (r *DBSubnetGroupRecord) GetTags() map[string]string { return r.Tags }
func (r *DBSubnetGroupRecord) SetTags(tags map[string]string) { r.Tags = tags }
// DBParameterGroupRecord is the db-parameter-groups/{name}/meta record. The values themselves live one
// key each under .../params/, so a modify touching one parameter cannot clobber a concurrent change to
// another.
+1 -1
View File
@@ -52,7 +52,7 @@ func (s *Service) RestoreDBInstanceFromDBSnapshot(ctx context.Context, input *rd
if err != nil {
return nil, err
}
placement, err := s.resolvePlacement(ctx, kv, accountID, req)
placement, err := s.resolvePlacement(ctx, accountID, req)
if err != nil {
return nil, err
}
-10
View File
@@ -101,16 +101,6 @@ func kvKeySegmentIdentifier(segment string) string {
return strings.ReplaceAll(segment, kvKeyColon, ":")
}
// DBSubnetGroupsPrefix returns the per-account key prefix under which each DB subnet group record lives.
func DBSubnetGroupsPrefix() string {
return "db-subnet-groups/"
}
// DBSubnetGroupKey returns the per-account KV key of a DB subnet group's record.
func DBSubnetGroupKey(name string) string {
return DBSubnetGroupsPrefix() + name
}
// DBParameterGroupsPrefix returns the parameter group prefix. A group's own record is at .../meta and
// its values hang off .../params/, so listing groups walks the meta keys.
func DBParameterGroupsPrefix() string {
-2
View File
@@ -18,7 +18,6 @@ func TestKeyPaths(t *testing.T) {
t.Parallel()
assert.Equal(t, "db-instances/orders-db", DBInstanceKey("orders-db"))
assert.Equal(t, "db-snapshots/orders-db-final", DBSnapshotKey("orders-db-final"))
assert.Equal(t, "db-subnet-groups/prod-db-subnets", DBSubnetGroupKey("prod-db-subnets"))
assert.Equal(t, "db-parameter-groups/pg16/meta", DBParameterGroupMetaKey("pg16"))
assert.Equal(t, "db-parameter-groups/pg16/params/shared_buffers",
DBParameterGroupParamKey("pg16", "shared_buffers"))
@@ -34,7 +33,6 @@ func TestKeyPathsSitUnderTheirPrefix(t *testing.T) {
t.Parallel()
assert.True(t, strings.HasPrefix(DBInstanceKey("x"), DBInstancesPrefix()))
assert.True(t, strings.HasPrefix(DBSnapshotKey("x"), DBSnapshotsPrefix()))
assert.True(t, strings.HasPrefix(DBSubnetGroupKey("x"), DBSubnetGroupsPrefix()))
assert.True(t, strings.HasPrefix(DBParameterGroupMetaKey("x"), DBParameterGroupsPrefix()))
assert.True(t, strings.HasPrefix(DBParameterGroupParamKey("x", "p"), DBParameterGroupParamsPrefix("x")))
assert.True(t, strings.HasPrefix(AutomatedBackupKey("x", "t"), AutomatedBackupsPrefix("x")))
+34 -199
View File
@@ -2,16 +2,12 @@ package handlers_rds
import (
"context"
"errors"
"fmt"
"log/slog"
"slices"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/domains/rds/subnetgroup"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
@@ -21,8 +17,6 @@ import (
const (
maxDBGroupNameLen = 255
maxDBGroupDescriptionLen = 255
// AWS's cap on the subnets one group may hold.
maxSubnetsPerGroup = 20
)
// The status AWS reports on a usable group. There is no asynchronous
@@ -33,8 +27,8 @@ const subnetGroupStatusComplete = "Complete"
// CreateDBSubnetGroup implements the RDS CreateDBSubnetGroup action. AWS requires a group to span at
// least two AZs, but this platform is single-AZ and multi-AZ is a V2 milestone, so that rule is not
// enforced: rejecting on it would fail every stock Terraform module for no safety benefit. The
// constraint that does matter is enforced below, because a group spanning two VPCs cannot host an
// instance at all.
// constraint that does matter is enforced by the owner, because a group spanning two VPCs cannot host
// an instance at all.
func (s *Service) CreateDBSubnetGroup(ctx context.Context, input *rds.CreateDBSubnetGroupInput, accountID string) (*rds.CreateDBSubnetGroupOutput, error) {
if input == nil {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "empty request")
@@ -52,39 +46,16 @@ func (s *Service) CreateDBSubnetGroup(ctx context.Context, input *rds.CreateDBSu
return nil, err
}
subnets, vpcID, err := s.resolveGroupSubnets(ctx, accountID, aws.StringValueSlice(input.SubnetIds))
if err != nil {
return nil, err
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
now := time.Now().UTC()
rec := DBSubnetGroupRecord{
rec, err := s.subnetGroups().Create(ctx, accountID, subnetgroup.Spec{
Name: name,
AccountID: accountID,
Description: description,
Subnets: subnets,
VpcID: vpcID,
SubnetIDs: aws.StringValueSlice(input.SubnetIds),
Tags: tags,
CreatedAt: now,
UpdatedAt: now,
}
// Create rather than Put, so two concurrent creates of one name have exactly
// one winner instead of both reporting success over each other's subnets.
if err := createJSON(ctx, kv, DBSubnetGroupKey(name), &rec); err != nil {
if errors.Is(err, kvstore.ErrExists) {
return nil, awserrors.Errorf(awserrors.ErrorDBSubnetGroupAlreadyExists,
"DB subnet group %s already exists", name)
}
})
if err != nil {
return nil, err
}
slog.InfoContext(ctx, "rds: DB subnet group created",
"dbSubnetGroup", name, "accountId", accountID, "vpcId", vpcID, "subnets", len(subnets))
return &rds.CreateDBSubnetGroupOutput{DBSubnetGroup: s.projectSubnetGroup(&rec)}, nil
return &rds.CreateDBSubnetGroupOutput{DBSubnetGroup: s.projectSubnetGroup(rec)}, nil
}
// DescribeDBSubnetGroups errors on a named group that does not exist, matching AWS; an unnamed request
@@ -93,35 +64,21 @@ func (s *Service) DescribeDBSubnetGroups(ctx context.Context, input *rds.Describ
if input == nil {
input = &rds.DescribeDBSubnetGroupsInput{}
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
if name := aws.StringValue(input.DBSubnetGroupName); name != "" {
rec, _, err := getDBSubnetGroup(ctx, kv, name)
rec, err := s.subnetGroups().Get(ctx, accountID, name)
if err != nil {
return nil, err
}
return &rds.DescribeDBSubnetGroupsOutput{DBSubnetGroups: []*rds.DBSubnetGroup{s.projectSubnetGroup(rec)}}, nil
}
names, err := listNames(ctx, kv, DBSubnetGroupsPrefix())
recs, err := s.subnetGroups().List(ctx, accountID)
if err != nil {
return nil, err
}
slices.Sort(names)
groups := make([]*rds.DBSubnetGroup, 0, len(names))
for _, name := range names {
var rec DBSubnetGroupRecord
found, err := getJSON(ctx, kv, DBSubnetGroupKey(name), &rec)
if err != nil {
return nil, err
}
if !found {
continue
}
groups = append(groups, s.projectSubnetGroup(&rec))
groups := make([]*rds.DBSubnetGroup, 0, len(recs))
for i := range recs {
groups = append(groups, s.projectSubnetGroup(&recs[i]))
}
groups, next, err := Page(groups, subnetGroupPageKey, input.MaxRecords, input.Marker)
if err != nil {
@@ -130,9 +87,8 @@ func (s *Service) DescribeDBSubnetGroups(ctx context.Context, input *rds.Describ
return &rds.DescribeDBSubnetGroupsOutput{DBSubnetGroups: groups, Marker: next}, nil
}
// ModifyDBSubnetGroup replaces the group's subnet set with SubnetIds, which must stay in the group's
// VPC, as AWS requires. An omitted or empty description keeps the stored one. A lost CAS replays
// against the fresh record, so a concurrent write is not undone.
// ModifyDBSubnetGroup replaces the group's subnet set with SubnetIds. An omitted or empty description
// keeps the stored one.
func (s *Service) ModifyDBSubnetGroup(ctx context.Context, input *rds.ModifyDBSubnetGroupInput, accountID string) (*rds.ModifyDBSubnetGroupOutput, error) {
if input == nil {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "empty request")
@@ -148,52 +104,13 @@ func (s *Service) ModifyDBSubnetGroup(ctx context.Context, input *rds.ModifyDBSu
}
}
kv, err := s.bucket(ctx, accountID)
rec, err := s.subnetGroups().Modify(ctx, accountID, name, description, aws.StringValueSlice(input.SubnetIds))
if err != nil {
return nil, err
}
// Read first so a missing group is reported as such rather than as whatever
// the subnet resolution would have objected to.
if _, _, err := getDBSubnetGroup(ctx, kv, name); err != nil {
return nil, err
}
subnets, vpcID, err := s.resolveGroupSubnets(ctx, accountID, aws.StringValueSlice(input.SubnetIds))
if err != nil {
return nil, err
}
key := DBSubnetGroupKey(name)
for range tagWriteAttempts {
rec, rev, err := getDBSubnetGroup(ctx, kv, name)
if err != nil {
return nil, err
}
if vpcID != rec.VpcID {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"The new Subnets are not in the same Vpc as the existing subnet group")
}
if description != "" {
rec.Description = description
}
rec.Subnets = subnets
rec.UpdatedAt = time.Now().UTC()
err = updateJSON(ctx, kv, key, rev, rec)
if err == nil {
slog.InfoContext(ctx, "rds: DB subnet group modified",
"dbSubnetGroup", name, "accountId", accountID, "vpcId", vpcID, "subnets", len(subnets))
return &rds.ModifyDBSubnetGroupOutput{DBSubnetGroup: s.projectSubnetGroup(rec)}, nil
}
if !errors.Is(err, kvstore.ErrConflict) {
return nil, err
}
}
return nil, fmt.Errorf("rds: update of %s contended after %d attempts", key, tagWriteAttempts)
return &rds.ModifyDBSubnetGroupOutput{DBSubnetGroup: s.projectSubnetGroup(rec)}, nil
}
// DeleteDBSubnetGroup is refused while any instance still names the group, including one that is only
// deleting: releasing it early would let a teardown lose the record of where its ENI was placed, and
// would make destroy ordering ambiguous.
func (s *Service) DeleteDBSubnetGroup(ctx context.Context, input *rds.DeleteDBSubnetGroupInput, accountID string) (*rds.DeleteDBSubnetGroupOutput, error) {
if input == nil {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "empty request")
@@ -202,113 +119,31 @@ func (s *Service) DeleteDBSubnetGroup(ctx context.Context, input *rds.DeleteDBSu
if name == "" {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValue, "DBSubnetGroupName is required")
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
if err := s.subnetGroups().Delete(ctx, accountID, name); err != nil {
return nil, err
}
if _, _, err := getDBSubnetGroup(ctx, kv, name); err != nil {
return nil, err
}
users, err := instancesUsingGroup(ctx, kv, func(rec *DBInstanceRecord) bool {
return rec.DBSubnetGroupName == name
})
if err != nil {
return nil, err
}
if len(users) > 0 {
return nil, awserrors.Errorf(awserrors.ErrorDBSubnetGroupInvalidState,
"DB subnet group %s is still used by %s", name, strings.Join(users, ", "))
}
if err := kv.Delete(ctx, DBSubnetGroupKey(name)); err != nil {
return nil, fmt.Errorf("rds: delete DB subnet group %s: %w", name, err)
}
slog.InfoContext(ctx, "rds: DB subnet group deleted", "dbSubnetGroup", name, "accountId", accountID)
return &rds.DeleteDBSubnetGroupOutput{}, nil
}
// Validates the requested subnets against the caller's own account and returns
// them with the AZ each subnet itself records. The describe is issued as the
// caller, so a subnet in another account simply does not come back and is
// reported as not found rather than leaked.
func (s *Service) resolveGroupSubnets(ctx context.Context, accountID string, requested []string) ([]DBSubnetGroupSubnet, string, error) {
if s.deps.Network == nil {
return nil, "", awserrors.Errorf(awserrors.ErrorServerInternal, "RDS networking is not wired on this node")
}
switch {
case len(requested) == 0:
return nil, "", awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"SubnetIds must name at least one subnet")
case len(requested) > maxSubnetsPerGroup:
return nil, "", awserrors.Errorf(awserrors.ErrorInvalidParameterValue,
"a DB subnet group may hold at most %d subnets, got %d", maxSubnetsPerGroup, len(requested))
}
out, err := s.deps.Network.DescribeSubnets(ctx, &ec2.DescribeSubnetsInput{
SubnetIds: aws.StringSlice(requested),
}, accountID)
if err != nil {
return nil, "", fmt.Errorf("rds: describe the subnets of the DB subnet group: %w", err)
}
found := map[string]*ec2.Subnet{}
if out != nil {
for _, subnet := range out.Subnets {
if id := aws.StringValue(subnet.SubnetId); id != "" {
found[id] = subnet
}
}
}
vpcID := ""
subnets := make([]DBSubnetGroupSubnet, 0, len(requested))
seen := make(map[string]bool, len(requested))
for _, id := range requested {
if seen[id] {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid, "subnet %s is named more than once", id)
}
seen[id] = true
subnet, ok := found[id]
if !ok {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid,
"subnet %s does not exist in this account", id)
}
subnetVPC := aws.StringValue(subnet.VpcId)
if vpcID == "" {
vpcID = subnetVPC
}
if subnetVPC != vpcID {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid,
"subnet %s is in VPC %s while the group's other subnets are in %s; "+
"a DB subnet group must span one VPC", id, subnetVPC, vpcID)
}
// Read off the subnet rather than stamped from the platform's single zone,
// so the response needs no change when V2 makes AZs real.
subnets = append(subnets, DBSubnetGroupSubnet{
SubnetID: id,
AvailabilityZone: aws.StringValue(subnet.AvailabilityZone),
})
}
if vpcID == "" {
return nil, "", awserrors.Errorf(awserrors.ErrorDBSubnetInvalid, "the named subnets report no VPC")
}
return subnets, vpcID, nil
// The group's lifecycle owner, bound to this Service's networking and bucket.
func (s *Service) subnetGroups() *subnetgroup.Owner {
return subnetgroup.New(s.deps.Network, instanceReferences{s}, s.bucket)
}
// The record plus its revision. A missing group raises AWS's own fault, so a
// well-formed name that resolves to nothing is distinguishable from a bad one.
func getDBSubnetGroup(ctx context.Context, kv *kvstore.Bucket, name string) (*DBSubnetGroupRecord, uint64, error) {
var rec DBSubnetGroupRecord
rev, found, err := getJSONRevision(ctx, kv, DBSubnetGroupKey(name), &rec)
// The instance side of the group's in-use guard. It reads instance records
// directly until instances have their own lifecycle owner.
type instanceReferences struct{ s *Service }
var _ subnetgroup.Dependants = instanceReferences{}
func (r instanceReferences) InstancesUsing(ctx context.Context, accountID, name string) ([]string, error) {
kv, err := r.s.bucket(ctx, accountID)
if err != nil {
return nil, 0, err
return nil, err
}
if !found {
return nil, 0, awserrors.Errorf(awserrors.ErrorDBSubnetGroupNotFound, "DB subnet group %s not found", name)
}
return &rec, rev, nil
return instancesUsingGroup(ctx, kv, func(rec *DBInstanceRecord) bool {
return rec.DBSubnetGroupName == name
})
}
// The identifiers of every instance in the account matching uses, sorted. Both
@@ -335,7 +170,7 @@ func instancesUsingGroup(ctx context.Context, kv *kvstore.Bucket, uses func(*DBI
return users, nil
}
func (s *Service) projectSubnetGroup(rec *DBSubnetGroupRecord) *rds.DBSubnetGroup {
func (s *Service) projectSubnetGroup(rec *subnetgroup.Record) *rds.DBSubnetGroup {
if rec == nil {
return nil
}
+7 -14
View File
@@ -11,6 +11,7 @@ import (
"github.com/aws/aws-sdk-go/private/protocol/xml/xmlutil"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/domains/rds/subnetgroup"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsxml "github.com/mulgadc/spinifex/spinifex/foundation/aws/xml"
"github.com/stretchr/testify/assert"
@@ -61,9 +62,7 @@ func TestCreateDBSubnetGroup_StoresEverySubnetSupplied(t *testing.T) {
subnetGroupInput(testSubnetGroup, "subnet-zebra", "subnet-alpha"), testAccountID)
require.NoError(t, err)
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
rec, _, err := getDBSubnetGroup(t.Context(), kv, testSubnetGroup)
rec, err := h.svc.subnetGroups().Get(t.Context(), testAccountID, testSubnetGroup)
require.NoError(t, err)
require.Len(t, rec.Subnets, 2)
@@ -382,16 +381,14 @@ func TestModifyDBSubnetGroup_KeepsTheOmittedDescriptionTagsAndCreatedAt(t *testi
_, err := h.svc.CreateDBSubnetGroup(t.Context(), create, testAccountID)
require.NoError(t, err)
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
before, _, err := getDBSubnetGroup(t.Context(), kv, testSubnetGroup)
before, err := h.svc.subnetGroups().Get(t.Context(), testAccountID, testSubnetGroup)
require.NoError(t, err)
_, err = h.svc.ModifyDBSubnetGroup(t.Context(),
modifySubnetGroupInput(testSubnetGroup, "subnet-alpha", "subnet-zebra"), testAccountID)
require.NoError(t, err)
after, _, err := getDBSubnetGroup(t.Context(), kv, testSubnetGroup)
after, err := h.svc.subnetGroups().Get(t.Context(), testAccountID, testSubnetGroup)
require.NoError(t, err)
assert.Equal(t, "Database subnets", after.Description)
assert.Equal(t, map[string]string{"env": "prod"}, after.Tags)
@@ -419,9 +416,7 @@ func TestModifyDBSubnetGroup_RefusesAMoveToAnotherVPC(t *testing.T) {
assert.Equal(t, awserrors.ErrorInvalidParameterValue, awserrors.ValidErrorCodeFromError(err),
"the code has to survive resolution or the client sees a 500")
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
rec, _, err := getDBSubnetGroup(t.Context(), kv, testSubnetGroup)
rec, err := h.svc.subnetGroups().Get(t.Context(), testAccountID, testSubnetGroup)
require.NoError(t, err)
assert.Equal(t, testDefaultVPC, rec.VpcID)
}
@@ -483,9 +478,7 @@ func TestModifyDBSubnetGroup_RejectsWhatCreateRejects(t *testing.T) {
assert.Equal(t, tc.want, awserrors.ValidErrorCodeFromError(err),
"the code has to survive resolution or the client sees a 500")
kv, err := h.svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
rec, _, err := getDBSubnetGroup(t.Context(), kv, testSubnetGroup)
rec, err := h.svc.subnetGroups().Get(t.Context(), testAccountID, testSubnetGroup)
require.NoError(t, err)
require.Len(t, rec.Subnets, 1)
assert.Equal(t, "subnet-alpha", rec.Subnets[0].SubnetID)
@@ -649,7 +642,7 @@ func TestDBSubnetGroupRecord_PersistedFieldNames(t *testing.T) {
require.NoError(t, err)
js, err := kv.KV(t.Context())
require.NoError(t, err)
entry, err := js.Get(t.Context(), DBSubnetGroupKey(testSubnetGroup))
entry, err := js.Get(t.Context(), subnetgroup.Key(testSubnetGroup))
require.NoError(t, err)
var raw map[string]json.RawMessage
+5 -2
View File
@@ -10,6 +10,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/rds"
"github.com/mulgadc/spinifex/spinifex/domains/rds/subnetgroup"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
)
@@ -48,6 +49,8 @@ type taggableResource struct {
// resource. A default parameter group is deliberately absent: it has no stored
// record, so a tag written to it would have nowhere to live and would read back
// as absent on the next apply.
var _ TaggedRecord = (*subnetgroup.Record)(nil)
var taggableResources = map[ResourceKind]taggableResource{
ResourceKindDBInstance: {
key: DBInstanceKey,
@@ -60,8 +63,8 @@ var taggableResources = map[ResourceKind]taggableResource{
notFound: awserrors.ErrorDBSnapshotNotFound,
},
ResourceKindDBSubnetGroup: {
key: DBSubnetGroupKey,
newRecord: func() TaggedRecord { return &DBSubnetGroupRecord{} },
key: subnetgroup.Key,
newRecord: func() TaggedRecord { return &subnetgroup.Record{} },
notFound: awserrors.ErrorDBSubnetGroupNotFound,
},
ResourceKindDBParameterGroup: {