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

refactor: move account key to state foundation

Browse Source
This commit is contained in:
joshsiv-mulga
2026-10-02 14:46:22 +10:00
parent 347ae8d9c8
commit d8985c2873
30 changed files with 173 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/foundation/state/kvutil/account_key.go
+7
View File
@@ -0,0 +1,7 @@
package kvutil
// AccountKey returns the conventional account-scoped KV key:
// "{accountID}.{resourceID}".
func AccountKey(accountID, resourceID string) string {
return accountID + "." + resourceID
}
spinifex/foundation/state/kvutil/account_key_test.go
+21
View File
@@ -0,0 +1,21 @@
package kvutil
import "testing"
func TestAccountKey(t *testing.T) {
tests := []struct {
accountID string
resourceID string
want string
}{
{"000000000000", "vpc-123", "000000000000.vpc-123"},
{"123456789012", "igw-abc", "123456789012.igw-abc"},
{"", "vol-1", ".vol-1"},
}
for _, tt := range tests {
got := AccountKey(tt.accountID, tt.resourceID)
if got != tt.want {
t.Errorf("AccountKey(%q, %q) = %q, want %q", tt.accountID, tt.resourceID, got, tt.want)
}
}
}
spinifex/handlers/ec2/eigw/service_impl.go
+3 -3
View File
@@ -89,7 +89,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, *input.VpcId)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, *input.VpcId)); err != nil {
slog.WarnContext(ctx, "CreateEgressOnlyInternetGateway: VPC not found for account", "vpcId", *input.VpcId, "accountID", accountID)
return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", *input.VpcId)
}
@@ -108,7 +108,7 @@ func (s *EgressOnlyIGWServiceImpl) CreateEgressOnlyInternetGateway(ctx context.C
if err != nil {
return nil, fmt.Errorf("failed to marshal Egress-only IGW record: %w", err)
}
if _, err := s.eigwKV.Put(ctx, utils.AccountKey(accountID, eigwID), data); err != nil {
if _, err := s.eigwKV.Put(ctx, kvutil.AccountKey(accountID, eigwID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -126,7 +126,7 @@ func (s *EgressOnlyIGWServiceImpl) DeleteEgressOnlyInternetGateway(ctx context.C
}
eigwID := *input.EgressOnlyInternetGatewayId
key := utils.AccountKey(accountID, eigwID)
key := kvutil.AccountKey(accountID, eigwID)
// Verify the EIGW exists before deleting
if _, err := s.eigwKV.Get(ctx, key); err != nil {
spinifex/handlers/ec2/eigw/service_impl_test.go
+3 -3
View File
@@ -8,8 +8,8 @@ import (
"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/kvutil"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -23,7 +23,7 @@ func setupTestEIGWService(t *testing.T) *EgressOnlyIGWServiceImpl {
// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
vpcEntries := map[string][]byte{}
for _, vpcID := range []string{"vpc-test123", "vpc-tagged", "vpc-tagged2"} {
vpcEntries[utils.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
}
testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, vpcEntries)
@@ -312,7 +312,7 @@ func TestCreateEgressOnlyInternetGateway_CrossAccountVPCRejected(t *testing.T) {
// Create VPC KV bucket with a VPC owned by testAccountID
vpcID := "vpc-alpha456"
testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
utils.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
kvutil.AccountKey(testAccountID, vpcID): []byte(`{"vpc_id":"vpc-alpha456","state":"available"}`),
})
var err error
spinifex/handlers/ec2/eip/allocate_idempotency.go
+2 -2
View File
@@ -12,7 +12,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
@@ -36,7 +36,7 @@ func (s *EIPServiceImpl) allocateOnce(ctx context.Context, accountID string, all
if key == "" || s.idemKV == nil {
return alloc()
}
kvKey := utils.AccountKey(accountID, idempotencyKeyHash(key))
kvKey := kvutil.AccountKey(accountID, idempotencyKeyHash(key))
if out := s.cachedAllocation(ctx, kvKey); out != nil {
slog.WarnContext(ctx, "AllocateAddress: returning first result for a retried request",
spinifex/handlers/ec2/eip/service_impl.go
+4 -4
View File
@@ -152,7 +152,7 @@ func (s *EIPServiceImpl) allocateAddress(ctx context.Context, input *ec2.Allocat
if err != nil {
return nil, fmt.Errorf("failed to marshal EIP record: %w", err)
}
if _, err := s.eipKV.Put(ctx, utils.AccountKey(accountID, allocID), data); err != nil {
if _, err := s.eipKV.Put(ctx, kvutil.AccountKey(accountID, allocID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -173,7 +173,7 @@ func (s *EIPServiceImpl) ReleaseAddress(ctx context.Context, input *ec2.ReleaseA
}
allocID := *input.AllocationId
key := utils.AccountKey(accountID, allocID)
key := kvutil.AccountKey(accountID, allocID)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
@@ -212,7 +212,7 @@ func (s *EIPServiceImpl) AssociateAddress(ctx context.Context, input *ec2.Associ
}
allocID := *input.AllocationId
key := utils.AccountKey(accountID, allocID)
key := kvutil.AccountKey(accountID, allocID)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
@@ -665,7 +665,7 @@ func (s *EIPServiceImpl) DescribeAddressesAttribute(ctx context.Context, input *
if id == nil {
continue
}
key := utils.AccountKey(accountID, *id)
key := kvutil.AccountKey(accountID, *id)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
continue // not found, skip
spinifex/handlers/ec2/igw/service_impl.go
+7 -7
View File
@@ -131,7 +131,7 @@ func (s *IGWServiceImpl) createIGW(ctx context.Context, input *ec2.CreateInterne
if err != nil {
return nil, fmt.Errorf("failed to marshal IGW record: %w", err)
}
if _, err := s.igwKV.Put(ctx, utils.AccountKey(accountID, igwID), data); err != nil {
if _, err := s.igwKV.Put(ctx, kvutil.AccountKey(accountID, igwID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -149,7 +149,7 @@ func (s *IGWServiceImpl) DeleteInternetGateway(ctx context.Context, input *ec2.D
}
igwID := *input.InternetGatewayId
key := utils.AccountKey(accountID, igwID)
key := kvutil.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -320,7 +320,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
igwID := *input.InternetGatewayId
vpcID := *input.VpcId
key := utils.AccountKey(accountID, igwID)
key := kvutil.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -341,7 +341,7 @@ func (s *IGWServiceImpl) AttachInternetGateway(ctx context.Context, input *ec2.A
slog.ErrorContext(ctx, "VPC KV unavailable, cannot verify VPC ownership")
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, vpcID)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
slog.WarnContext(ctx, "AttachInternetGateway: VPC not found for account", "vpcId", vpcID, "accountID", accountID)
return nil, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
@@ -382,7 +382,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
igwID := *input.InternetGatewayId
vpcID := *input.VpcId
key := utils.AccountKey(accountID, igwID)
key := kvutil.AccountKey(accountID, igwID)
entry, err := s.igwKV.Get(ctx, key)
if err != nil {
@@ -457,11 +457,11 @@ func (s *IGWServiceImpl) CreateAttachedInternetGateway(ctx context.Context, acco
if s.vpcKV == nil {
return false, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, vpcID)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID)); err != nil {
return false, awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
key := utils.AccountKey(accountID, igwID)
key := kvutil.AccountKey(accountID, igwID)
data, err := json.Marshal(IGWRecord{
InternetGatewayId: igwID,
VpcId: vpcID,
spinifex/handlers/ec2/igw/service_impl_test.go
+6 -6
View File
@@ -11,8 +11,8 @@ import (
"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/kvutil"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -54,7 +54,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
// Create VPC KV bucket and register test VPCs so fail-closed ownership checks pass
vpcEntries := map[string][]byte{}
for _, vpcID := range []string{"vpc-test123", "vpc-other", "vpc-lifecycle", "vpc-event-test"} {
vpcEntries[utils.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
vpcEntries[kvutil.AccountKey(testAccountID, vpcID)] = []byte(`{"vpc_id":"` + vpcID + `","state":"available"}`)
}
testutil.SeedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, vpcEntries)
@@ -67,7 +67,7 @@ func setupTestIGWService(t *testing.T) (*IGWServiceImpl, *nats.Conn) {
// gateway come up. Without it an attach stays pending and reports no attachment.
func confirmAttach(t *testing.T, svc *IGWServiceImpl, igwID, vpcID string) {
t.Helper()
require.NoError(t, MarkAttached(context.Background(), svc.igwKV, utils.AccountKey(testAccountID, igwID), vpcID))
require.NoError(t, MarkAttached(context.Background(), svc.igwKV, kvutil.AccountKey(testAccountID, igwID), vpcID))
}
func createTestIGW(t *testing.T, svc *IGWServiceImpl) string {
@@ -338,7 +338,7 @@ func TestDescribeInternetGateways_LegacyRecordReportsAttachment(t *testing.T) {
svc, _ := setupTestIGWService(t)
igwID := createTestIGW(t, svc)
key := utils.AccountKey(testAccountID, igwID)
key := kvutil.AccountKey(testAccountID, igwID)
entry, err := svc.igwKV.Get(context.Background(), key)
require.NoError(t, err)
var record IGWRecord
@@ -405,7 +405,7 @@ func TestMarkAttached_NoRewriteWhenAlreadyAttached(t *testing.T) {
}, testAccountID)
require.NoError(t, err)
key := utils.AccountKey(testAccountID, igwID)
key := kvutil.AccountKey(testAccountID, igwID)
confirmAttach(t, svc, igwID, "vpc-test123")
first, err := svc.igwKV.Get(context.Background(), key)
require.NoError(t, err)
@@ -782,7 +782,7 @@ func TestAttachInternetGateway_CrossAccountVPCRejected(t *testing.T) {
require.NoError(t, err)
vpcID := "vpc-alpha123"
_, err = vpcKV.Put(t.Context(), utils.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
_, err = vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), []byte(`{"vpc_id":"vpc-alpha123","state":"available"}`))
require.NoError(t, err)
// Refresh service to pick up the VPC KV bucket
spinifex/handlers/ec2/launchtemplate/service_impl.go
+8 -8
View File
@@ -84,24 +84,24 @@ func NewLaunchTemplateServiceImplWithNATS(ctx context.Context, cfg *config.Confi
// Launch template names allow characters such as parentheses that are not valid
// in NATS KV keys, so the name must not be used in its raw form here.
func nameKey(accountID, name string) string {
return utils.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
return kvutil.AccountKey(accountID, "name."+hex.EncodeToString([]byte(name)))
}
// versionKey returns a version-body key: account.lt-<id>.v<n>.
func versionKey(accountID, ltID string, n int64) string {
return utils.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
return kvutil.AccountKey(accountID, ltID+".v"+strconv.FormatInt(n, 10))
}
// versionPrefix returns the scan prefix for a template's version bodies.
func versionPrefix(accountID, ltID string) string {
return utils.AccountKey(accountID, ltID+".v")
return kvutil.AccountKey(accountID, ltID+".v")
}
// --- record load helpers ---
// getHeaderByID reads a header and its KV entry (for CAS) by launch template id.
func (s *LaunchTemplateServiceImpl) getHeaderByID(ctx context.Context, accountID, ltID string) (*LaunchTemplateHeader, jetstream.KeyValueEntry, error) {
entry, err := s.kv.Get(ctx, utils.AccountKey(accountID, ltID))
entry, err := s.kv.Get(ctx, kvutil.AccountKey(accountID, ltID))
if err != nil {
return nil, nil, errors.New(awserrors.ErrorInvalidLaunchTemplateIdNotFound)
}
@@ -336,7 +336,7 @@ func (s *LaunchTemplateServiceImpl) claimName(ctx context.Context, accountID, na
// header means the name is taken; any other read error (transient fault, or a
// concurrent in-flight create whose header is not yet written) fails closed so
// the name is never stolen from a possibly-live template.
_, herr := s.kv.Get(ctx, utils.AccountKey(accountID, string(entry.Value())))
_, herr := s.kv.Get(ctx, kvutil.AccountKey(accountID, string(entry.Value())))
switch {
case herr == nil:
return errNameInUse()
@@ -354,7 +354,7 @@ func (s *LaunchTemplateServiceImpl) putHeader(ctx context.Context, accountID str
if err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.kv.Put(ctx, utils.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
if _, err := s.kv.Put(ctx, kvutil.AccountKey(accountID, h.LaunchTemplateId), data); err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
return nil
@@ -466,7 +466,7 @@ func (s *LaunchTemplateServiceImpl) ModifyLaunchTemplate(ctx context.Context, in
if err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.kv.Update(ctx, utils.AccountKey(accountID, header.LaunchTemplateId), data, entry.Revision()); err != nil {
if _, err := s.kv.Update(ctx, kvutil.AccountKey(accountID, header.LaunchTemplateId), data, entry.Revision()); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
}
@@ -495,7 +495,7 @@ func (s *LaunchTemplateServiceImpl) DeleteLaunchTemplate(ctx context.Context, in
// Delete the header first: the template immediately vanishes from every
// describe. Version bodies and the name index are best-effort cleanup.
if err := s.kv.Delete(ctx, utils.AccountKey(accountID, header.LaunchTemplateId)); err != nil {
if err := s.kv.Delete(ctx, kvutil.AccountKey(accountID, header.LaunchTemplateId)); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if err := s.kv.Delete(ctx, nameKey(accountID, header.LaunchTemplateName)); err != nil {
spinifex/handlers/ec2/launchtemplate/service_impl_test.go
+2 -2
View File
@@ -9,7 +9,7 @@ import (
"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/utils"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -116,7 +116,7 @@ func TestCreateLaunchTemplate_OrphanNameReclaim(t *testing.T) {
lt := createTemplate(t, svc, "orphan", "t3.micro")
// Simulate a crash after the name claim but with the header lost.
require.NoError(t, svc.kv.Delete(t.Context(), utils.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
require.NoError(t, svc.kv.Delete(t.Context(), kvutil.AccountKey(testAccountID, aws.StringValue(lt.LaunchTemplateId))))
// The name index still points at the now-orphaned id; create must reclaim it.
out, err := svc.CreateLaunchTemplate(context.Background(), &ec2.CreateLaunchTemplateInput{
spinifex/handlers/ec2/natgw/lifecycle_invariants_test.go
+3 -3
View File
@@ -8,8 +8,8 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -41,7 +41,7 @@ func TestRLC3_NatGatewayDeletesWhileRouted(t *testing.T) {
natgwID := createTestNatGateway(t, svc)
seedKV(t, js, kvBucketRouteTables, map[string][]byte{
utils.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
kvutil.AccountKey(testAccountID, "rtb-routed"): fmt.Appendf(nil,
`{"route_table_id":"rtb-routed","vpc_id":"vpc-test1","routes":[{"destination_cidr_block":"0.0.0.0/0","nat_gateway_id":%q}]}`, natgwID),
})
@@ -92,7 +92,7 @@ func createTestNatGateway(t *testing.T, svc *NatGatewayServiceImpl) string {
func readEIP(t *testing.T, svc *NatGatewayServiceImpl, allocID string) handlers_ec2_eip.EIPRecord {
t.Helper()
entry, err := svc.eipKV.Get(t.Context(), utils.AccountKey(testAccountID, allocID))
entry, err := svc.eipKV.Get(t.Context(), kvutil.AccountKey(testAccountID, allocID))
require.NoError(t, err)
var eip handlers_ec2_eip.EIPRecord
require.NoError(t, json.Unmarshal(entry.Value(), &eip))
spinifex/handlers/ec2/natgw/service_impl.go
+9 -9
View File
@@ -118,7 +118,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
allocID := *input.AllocationId
// Validate subnet exists and get its VPC
subnetEntry, err := s.subnetKV.Get(ctx, utils.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -128,7 +128,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
}
// Validate EIP exists and is not already associated
eipEntry, err := s.eipKV.Get(ctx, utils.AccountKey(accountID, allocID))
eipEntry, err := s.eipKV.Get(ctx, kvutil.AccountKey(accountID, allocID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidAllocationIDNotFound)
}
@@ -157,7 +157,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
if err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.natgwKV.Put(ctx, utils.AccountKey(accountID, natgwID), data); err != nil {
if _, err := s.natgwKV.Put(ctx, kvutil.AccountKey(accountID, natgwID), data); err != nil {
slog.ErrorContext(ctx, "Failed to store NAT Gateway", "natGatewayId", natgwID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -170,7 +170,7 @@ func (s *NatGatewayServiceImpl) CreateNatGateway(ctx context.Context, input *ec2
eipRecord.VpcId = subnetRecord.VpcId
eipRecord.State = "associated"
if eipData, err := json.Marshal(eipRecord); err == nil {
if _, err := s.eipKV.Update(ctx, utils.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
if _, err := s.eipKV.Update(ctx, kvutil.AccountKey(accountID, allocID), eipData, eipEntry.Revision()); err != nil {
slog.WarnContext(ctx, "CreateNatGateway: failed to mark EIP associated", "natGatewayId", natgwID, "allocationId", allocID, "err", err)
}
}
@@ -190,7 +190,7 @@ func (s *NatGatewayServiceImpl) DeleteNatGateway(ctx context.Context, input *ec2
}
natgwID := *input.NatGatewayId
key := utils.AccountKey(accountID, natgwID)
key := kvutil.AccountKey(accountID, natgwID)
entry, err := s.natgwKV.Get(ctx, key)
if err != nil {
@@ -307,7 +307,7 @@ func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(ctx context.Con
if assoc.SubnetId == "" {
continue
}
subnetEntry, err := s.subnetKV.Get(ctx, utils.AccountKey(accountID, assoc.SubnetId))
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, assoc.SubnetId))
if err != nil {
continue
}
@@ -327,7 +327,7 @@ func (s *NatGatewayServiceImpl) disassociateEIP(ctx context.Context, record *Nat
if record.AllocationId == "" {
return
}
key := utils.AccountKey(accountID, record.AllocationId)
key := kvutil.AccountKey(accountID, record.AllocationId)
entry, err := s.eipKV.Get(ctx, key)
if err != nil {
if !errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -425,7 +425,7 @@ func (s *NatGatewayServiceImpl) DescribeNatGateways(ctx context.Context, input *
if foundIDs[id] {
continue
}
key := utils.AccountKey(accountID, id)
key := kvutil.AccountKey(accountID, id)
entry, err := s.deletedNatgwKV.Get(ctx, key)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
@@ -499,7 +499,7 @@ func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp
// GetNatGateway retrieves a NAT Gateway record by ID.
func (s *NatGatewayServiceImpl) GetNatGateway(accountID, natgwID string) (*NatGatewayRecord, error) {
ctx := context.Background()
entry, err := s.natgwKV.Get(ctx, utils.AccountKey(accountID, natgwID))
entry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
spinifex/handlers/ec2/natgw/service_impl_test.go
+6 -6
View File
@@ -6,12 +6,12 @@ import (
"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"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -31,18 +31,18 @@ func setupTestServiceJS(t *testing.T) (*NatGatewayServiceImpl, jetstream.JetStre
// Seed VPC KV
seedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
utils.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
})
// Seed subnet KV (public subnet)
seedKV(t, js, handlers_ec2_vpc.KVBucketSubnets, map[string][]byte{
utils.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
kvutil.AccountKey(testAccountID, "subnet-pub1"): []byte(`{"subnet_id":"subnet-pub1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available","map_public_ip_on_launch":true}`),
})
// Seed EIP KV
seedKV(t, js, handlers_ec2_eip.KVBucketEIPs, map[string][]byte{
utils.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
utils.AccountKey(testAccountID, "eipalloc-used"): []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
kvutil.AccountKey(testAccountID, "eipalloc-test1"): []byte(`{"allocation_id":"eipalloc-test1","public_ip":"203.0.113.50","state":"allocated"}`),
kvutil.AccountKey(testAccountID, "eipalloc-used"): []byte(`{"allocation_id":"eipalloc-used","public_ip":"203.0.113.51","state":"associated","association_id":"eipassoc-xxx"}`),
})
svc, err := NewNatGatewayServiceImplWithNATS(t.Context(), nc)
spinifex/handlers/ec2/placementgroup/service_impl.go
+8 -8
View File
@@ -111,7 +111,7 @@ func (s *PlacementGroupServiceImpl) CreatePlacementGroup(ctx context.Context, in
}
groupName := *input.GroupName
key := utils.AccountKey(accountID, groupName)
key := kvutil.AccountKey(accountID, groupName)
groupID := utils.GenerateResourceID("pg")
record := PlacementGroupRecord{
@@ -149,7 +149,7 @@ func (s *PlacementGroupServiceImpl) DeletePlacementGroup(ctx context.Context, in
}
groupName := *input.GroupName
key := utils.AccountKey(accountID, groupName)
key := kvutil.AccountKey(accountID, groupName)
entry, err := s.kv.Get(ctx, key)
if err != nil {
@@ -375,7 +375,7 @@ func (s *PlacementGroupServiceImpl) ReserveSpreadNodes(ctx context.Context, inpu
}
var selected []string
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
if record.State != ec2.PlacementGroupStateAvailable {
return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -418,7 +418,7 @@ func (s *PlacementGroupServiceImpl) FinalizeSpreadInstances(ctx context.Context,
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
maps.Copy(record.NodeInstances, input.NodeInstances)
return true, nil
@@ -443,7 +443,7 @@ func (s *PlacementGroupServiceImpl) ReleaseSpreadNodes(ctx context.Context, inpu
releaseSet[n] = true
}
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
changed := false
for node := range releaseSet {
@@ -471,7 +471,7 @@ func (s *PlacementGroupServiceImpl) RemoveInstance(ctx context.Context, input *R
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
instances, exists := record.NodeInstances[input.NodeName]
if !exists {
@@ -514,7 +514,7 @@ func (s *PlacementGroupServiceImpl) ReserveClusterNode(ctx context.Context, inpu
}
var targetNode string
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
if record.State != ec2.PlacementGroupStateAvailable {
return false, errors.New(awserrors.ErrorInvalidPlacementGroupUnknown)
@@ -551,7 +551,7 @@ func (s *PlacementGroupServiceImpl) FinalizeClusterInstances(ctx context.Context
return nil, errors.New(awserrors.ErrorMissingParameter)
}
_, err := kvutil.Update(ctx, s.kv, utils.AccountKey(accountID, input.GroupName), casConfig(),
_, err := kvutil.Update(ctx, s.kv, kvutil.AccountKey(accountID, input.GroupName), casConfig(),
func(record *PlacementGroupRecord) (bool, error) {
// Append rather than replace: a cluster group can have several launches in flight at once.
for node, ids := range input.NodeInstances {
spinifex/handlers/ec2/placementgroup/service_impl_test.go
+2 -3
View File
@@ -11,7 +11,6 @@ import (
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -40,7 +39,7 @@ func createTestGroup(t *testing.T, svc *PlacementGroupServiceImpl, name, strateg
// readRecord returns the stored record for assertions.
func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string) *PlacementGroupRecord {
t.Helper()
entry, err := svc.kv.Get(t.Context(), utils.AccountKey(testAccountID, groupName))
entry, err := svc.kv.Get(t.Context(), kvutil.AccountKey(testAccountID, groupName))
require.NoError(t, err)
var record PlacementGroupRecord
require.NoError(t, json.Unmarshal(entry.Value(), &record))
@@ -51,7 +50,7 @@ func readRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string)
// starting state the public API cannot produce.
func seedRecord(t *testing.T, svc *PlacementGroupServiceImpl, groupName string, mutate func(*PlacementGroupRecord)) {
t.Helper()
_, err := kvutil.Update(t.Context(), svc.kv, utils.AccountKey(testAccountID, groupName), kvutil.CASConfig{},
_, err := kvutil.Update(t.Context(), svc.kv, kvutil.AccountKey(testAccountID, groupName), kvutil.CASConfig{},
func(record *PlacementGroupRecord) (bool, error) {
mutate(record)
return true, nil
spinifex/handlers/ec2/routetable/service_impl.go
+12 -12
View File
@@ -92,7 +92,7 @@ func NewRouteTableServiceImplWithNATS(ctx context.Context, cfg *config.Config, n
// getRouteTable retrieves a route table record from KV.
func (s *RouteTableServiceImpl) getRouteTable(ctx context.Context, accountID, rtbID string) (*RouteTableRecord, error) {
entry, err := s.rtbKV.Get(ctx, utils.AccountKey(accountID, rtbID))
entry, err := s.rtbKV.Get(ctx, kvutil.AccountKey(accountID, rtbID))
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, rtbNotFoundError(rtbID)
@@ -115,7 +115,7 @@ func (s *RouteTableServiceImpl) putRouteTable(ctx context.Context, accountID str
slog.ErrorContext(ctx, "Failed to marshal route table record", "routeTableId", record.RouteTableId, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.rtbKV.Put(ctx, utils.AccountKey(accountID, record.RouteTableId), data); err != nil {
if _, err := s.rtbKV.Put(ctx, kvutil.AccountKey(accountID, record.RouteTableId), data); err != nil {
slog.ErrorContext(ctx, "Failed to write route table to KV", "routeTableId", record.RouteTableId, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
@@ -140,7 +140,7 @@ var (
// several subnets with one route table at once. mutate reports whether
// it changed the record; a false return commits nothing.
func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, accountID, rtbID string, mutate func(*RouteTableRecord) (bool, error)) error {
_, err := kvutil.Update(ctx, s.rtbKV, utils.AccountKey(accountID, rtbID), kvutil.CASConfig{
_, err := kvutil.Update(ctx, s.rtbKV, kvutil.AccountKey(accountID, rtbID), kvutil.CASConfig{
Attempts: rtbCASMaxRetries,
NotFound: errRTBAbsent,
Exhausted: func(string, int) error { return errRTBContended },
@@ -170,7 +170,7 @@ func (s *RouteTableServiceImpl) mutateRouteTableCAS(ctx context.Context, account
// getVPCCidr looks up a VPC's CIDR block from the VPC KV bucket.
func (s *RouteTableServiceImpl) getVPCCidr(ctx context.Context, accountID, vpcID string) (string, error) {
entry, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, vpcID))
entry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
if err != nil {
return "", awserrors.IDNotFound(awserrors.ErrorInvalidVpcIDNotFound, "vpc", vpcID)
}
@@ -584,7 +584,7 @@ func (s *RouteTableServiceImpl) DeleteRouteTable(ctx context.Context, input *ec2
}
}
if err := s.rtbKV.Delete(ctx, utils.AccountKey(accountID, rtbID)); err != nil {
if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -752,7 +752,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
case input.GatewayId != nil && *input.GatewayId != "":
igwID := *input.GatewayId
// Verify IGW exists and is attached to the same VPC
igwEntry, err := s.igwKV.Get(ctx, utils.AccountKey(accountID, igwID))
igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
if err != nil {
return nil, gatewayNotFoundError(igwID)
}
@@ -782,7 +782,7 @@ func (s *RouteTableServiceImpl) CreateRoute(ctx context.Context, input *ec2.Crea
case input.NatGatewayId != nil && *input.NatGatewayId != "":
natgwID := *input.NatGatewayId
// Verify NAT GW exists and belongs to the same VPC
natgwEntry, err := s.natgwKV.Get(ctx, utils.AccountKey(accountID, natgwID))
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound)
}
@@ -936,7 +936,7 @@ func (s *RouteTableServiceImpl) ReplaceRoute(ctx context.Context, input *ec2.Rep
igwID := *input.GatewayId
// AWS reports a missing gateway ahead of a missing route.
igwEntry, err := s.igwKV.Get(ctx, utils.AccountKey(accountID, igwID))
igwEntry, err := s.igwKV.Get(ctx, kvutil.AccountKey(accountID, igwID))
if err != nil {
return nil, gatewayNotFoundError(igwID)
}
@@ -995,7 +995,7 @@ func (s *RouteTableServiceImpl) AssociateRouteTable(ctx context.Context, input *
}
// Verify subnet exists and belongs to the same VPC
subnetEntry, err := s.subnetKV.Get(ctx, utils.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -1381,7 +1381,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayDeleteEvents(ctx context.Contex
if s.natsConn == nil {
return
}
natgwEntry, err := s.natgwKV.Get(ctx, utils.AccountKey(accountID, natgwID))
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, natgwID))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", "vpc.delete-nat-gateway", "natGatewayId", natgwID, "err", err)
return
@@ -1429,7 +1429,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventsForAssociation(ctx contex
if r.NatGatewayId == "" {
continue
}
natgwEntry, err := s.natgwKV.Get(ctx, utils.AccountKey(accountID, r.NatGatewayId))
natgwEntry, err := s.natgwKV.Get(ctx, kvutil.AccountKey(accountID, r.NatGatewayId))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: natgw lookup failed", "topic", topic, "natGatewayId", r.NatGatewayId, "err", err)
continue
@@ -1454,7 +1454,7 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventForSubnet(ctx context.Cont
if s.natsConn == nil {
return
}
subnetEntry, err := s.subnetKV.Get(ctx, utils.AccountKey(accountID, subnetID))
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetID))
if err != nil {
slog.WarnContext(ctx, "NAT GW event: subnet lookup failed", "topic", topic, "subnetId", subnetID, "err", err)
return
spinifex/handlers/ec2/routetable/service_impl_test.go
+7 -7
View File
@@ -8,12 +8,12 @@ import (
"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/kvutil"
handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -29,23 +29,23 @@ func setupTestService(t *testing.T) *RouteTableServiceImpl {
// Seed VPC KV
seedKV(t, js, handlers_ec2_vpc.KVBucketVPCs, map[string][]byte{
utils.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
kvutil.AccountKey(testAccountID, "vpc-test1"): []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`),
})
// Seed IGW KV (attached to vpc-test1)
seedKV(t, js, handlers_ec2_igw.KVBucketIGW, map[string][]byte{
utils.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
kvutil.AccountKey(testAccountID, "igw-test1"): []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"available"}`),
})
// Seed subnet KV
seedKV(t, js, handlers_ec2_vpc.KVBucketSubnets, map[string][]byte{
utils.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
utils.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
kvutil.AccountKey(testAccountID, "subnet-test1"): []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`),
kvutil.AccountKey(testAccountID, "subnet-priv1"): []byte(`{"subnet_id":"subnet-priv1","vpc_id":"vpc-test1","cidr_block":"10.0.11.0/24","state":"available"}`),
})
// Seed NAT Gateway KV
seedKV(t, js, handlers_ec2_natgw.KVBucketNatGateways, map[string][]byte{
utils.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
kvutil.AccountKey(testAccountID, "nat-test1"): []byte(`{"nat_gateway_id":"nat-test1","vpc_id":"vpc-test1","subnet_id":"subnet-test1","public_ip":"192.168.1.100","state":"available"}`),
})
svc, err := NewRouteTableServiceImplWithNATS(t.Context(), nil, nc)
spinifex/handlers/ec2/spotinstance/service_impl.go
+2 -3
View File
@@ -17,7 +17,6 @@ import (
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/foundation/state/migrate"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
@@ -92,7 +91,7 @@ func (s *SpotInstanceServiceImpl) PutSpotInstanceRequests(ctx context.Context, i
if err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.activeKV.Put(ctx, utils.AccountKey(accountID, sirID), data); err != nil {
if _, err := s.activeKV.Put(ctx, kvutil.AccountKey(accountID, sirID), data); err != nil {
slog.ErrorContext(ctx, "PutSpotInstanceRequests: KV put failed", "sirId", sirID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -177,7 +176,7 @@ func (s *SpotInstanceServiceImpl) CancelSpotInstanceRequests(ctx context.Context
cancelled := make([]*ec2.CancelledSpotInstanceRequest, 0, len(input.SpotInstanceRequestIds))
for _, idPtr := range input.SpotInstanceRequestIds {
sirID := aws.StringValue(idPtr)
key := utils.AccountKey(accountID, sirID)
key := kvutil.AccountKey(accountID, sirID)
record, err := s.readRecord(ctx, s.activeKV, key)
if err == nil {
spinifex/handlers/ec2/vpc/default_vpc.go
+3 -3
View File
@@ -104,7 +104,7 @@ func (s *VPCServiceImpl) ensureDefaultVPC(ctx context.Context, accountID string,
return claim.info(), nil
}
if _, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, claim.VpcId)); err == nil {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, claim.VpcId)); err == nil {
return claim.info(), nil
} else if !errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, fmt.Errorf("read default VPC %s: %w", claim.VpcId, err)
@@ -261,7 +261,7 @@ func oldestAccountRecord[T any](ctx context.Context, kv jetstream.KeyValue, acco
// marks the claim complete. Records are written with Create, so only the caller
// that actually wrote one announces it to vpcd.
func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string, claim *defaultVPCClaim) error {
created, err := createRecord(ctx, s.vpcKV, utils.AccountKey(accountID, claim.VpcId), VPCRecord{
created, err := createRecord(ctx, s.vpcKV, kvutil.AccountKey(accountID, claim.VpcId), VPCRecord{
VpcId: claim.VpcId,
CidrBlock: claim.Cidr,
State: "available",
@@ -283,7 +283,7 @@ func (s *VPCServiceImpl) buildDefaultVPC(ctx context.Context, accountID string,
if s.config != nil && s.config.AZ != "" {
az = s.config.AZ
}
created, err = createRecord(ctx, s.subnetKV, utils.AccountKey(accountID, claim.SubnetId), SubnetRecord{
created, err = createRecord(ctx, s.subnetKV, kvutil.AccountKey(accountID, claim.SubnetId), SubnetRecord{
SubnetId: claim.SubnetId,
VpcId: claim.VpcId,
CidrBlock: claim.SubnetCidr,
spinifex/handlers/ec2/vpc/default_vpc_test.go
+5 -5
View File
@@ -11,9 +11,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
@@ -149,12 +149,12 @@ func TestEnsureDefaultVPC_AdoptsOldestExistingDefault(t *testing.T) {
const accountID = "000000000051"
created := time.Date(2026, 9, 11, 14, 13, 40, 0, time.UTC)
for i, id := range []string{"vpc-0000000000000newer", "vpc-0000000000000older"} {
putRecord(t, js, handlers_ec2_vpc.KVBucketVPCs, utils.AccountKey(accountID, id), handlers_ec2_vpc.VPCRecord{
putRecord(t, js, handlers_ec2_vpc.KVBucketVPCs, kvutil.AccountKey(accountID, id), handlers_ec2_vpc.VPCRecord{
VpcId: id, CidrBlock: handlers_ec2_vpc.DefaultVPCCidr, IsDefault: true, VNI: int64(200 + i),
CreatedAt: created.Add(time.Duration(1-i) * time.Minute),
})
}
putRecord(t, js, handlers_ec2_vpc.KVBucketSubnets, utils.AccountKey(accountID, "subnet-00000000000older"), handlers_ec2_vpc.SubnetRecord{
putRecord(t, js, handlers_ec2_vpc.KVBucketSubnets, kvutil.AccountKey(accountID, "subnet-00000000000older"), handlers_ec2_vpc.SubnetRecord{
SubnetId: "subnet-00000000000older", VpcId: "vpc-0000000000000older", CidrBlock: handlers_ec2_vpc.DefaultSubnetCidr,
IsDefault: true, CreatedAt: created,
})
@@ -183,7 +183,7 @@ func TestEnsureDefaultVPC_FinishesHalfBuiltClaim(t *testing.T) {
"route_table_id": "rtb-00000000000half", "internet_gateway_id": "igw-00000000000half",
"cidr": handlers_ec2_vpc.DefaultVPCCidr, "subnet_cidr": handlers_ec2_vpc.DefaultSubnetCidr, "vni": 4242,
})
putRecord(t, js, handlers_ec2_vpc.KVBucketVPCs, utils.AccountKey(accountID, "vpc-00000000000000half"), handlers_ec2_vpc.VPCRecord{
putRecord(t, js, handlers_ec2_vpc.KVBucketVPCs, kvutil.AccountKey(accountID, "vpc-00000000000000half"), handlers_ec2_vpc.VPCRecord{
VpcId: "vpc-00000000000000half", CidrBlock: handlers_ec2_vpc.DefaultVPCCidr, IsDefault: true, VNI: 4242,
})
spinifex/handlers/ec2/vpc/eni.go
+12 -12
View File
@@ -115,7 +115,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
subnetId := *input.SubnetId
// Verify subnet exists and belongs to this account
subnetEntry, err := s.subnetKV.Get(ctx, utils.AccountKey(accountID, subnetId))
subnetEntry, err := s.subnetKV.Get(ctx, kvutil.AccountKey(accountID, subnetId))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
@@ -202,7 +202,7 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2.
if err != nil {
return nil, fmt.Errorf("failed to marshal ENI record: %w", err)
}
if _, err := s.eniKV.Put(ctx, utils.AccountKey(accountID, eniId), data); err != nil {
if _, err := s.eniKV.Put(ctx, kvutil.AccountKey(accountID, eniId), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -248,7 +248,7 @@ func (s *VPCServiceImpl) ForceDeleteInstanceENI(ctx context.Context, accountID,
}
func (s *VPCServiceImpl) deleteNetworkInterface(ctx context.Context, eniId, accountID string, force bool) (*ec2.DeleteNetworkInterfaceOutput, error) {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
// Get the ENI record
entry, err := s.eniKV.Get(ctx, key)
@@ -343,7 +343,7 @@ func (s *VPCServiceImpl) releaseENISideEffects(ctx context.Context, eniId, accou
// deletion that did not happen.
func (s *VPCServiceImpl) DetachAndDeleteENI(ctx context.Context, accountID, eniID string, force bool) (deleted bool, err error) {
const maxAttempts = 3
key := utils.AccountKey(accountID, eniID)
key := kvutil.AccountKey(accountID, eniID)
var lastErr error
for range maxAttempts {
@@ -455,7 +455,7 @@ func (s *VPCServiceImpl) ModifyNetworkInterfaceAttribute(ctx context.Context, in
}
eniId := *input.NetworkInterfaceId
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
@@ -696,7 +696,7 @@ func (s *VPCServiceImpl) AttachENI(accountID, eniId, instanceId string, deviceIn
}
func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instanceId string, deviceIndex int64) (string, error) {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return "", eniNotFoundError(eniId)
@@ -738,7 +738,7 @@ func (s *VPCServiceImpl) attachENI(ctx context.Context, accountID, eniId, instan
// DetachENI marks an ENI as detached from an instance (internal use by TerminateInstances).
// accountID scopes the lookup to the correct KV key.
func (s *VPCServiceImpl) DetachENI(ctx context.Context, accountID, eniId string) error {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return eniNotFoundError(eniId)
@@ -774,7 +774,7 @@ func (s *VPCServiceImpl) GetENIRecord(accountID, eniId string) (ENIRecord, error
}
func (s *VPCServiceImpl) getENIRecord(ctx context.Context, accountID, eniId string) (ENIRecord, error) {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return ENIRecord{}, eniNotFoundError(eniId)
@@ -794,7 +794,7 @@ func (s *VPCServiceImpl) UpdateENI(accountID, eniId string, fn func(*ENIRecord))
}
func (s *VPCServiceImpl) updateENI(ctx context.Context, accountID, eniId string, fn func(*ENIRecord)) error {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return eniNotFoundError(eniId)
@@ -984,7 +984,7 @@ func (s *VPCServiceImpl) UpdateENIPublicIP(accountID, eniId, publicIP, poolName
}
func (s *VPCServiceImpl) updateENIPublicIP(ctx context.Context, accountID, eniId, publicIP, poolName string) error {
key := utils.AccountKey(accountID, eniId)
key := kvutil.AccountKey(accountID, eniId)
entry, err := s.eniKV.Get(ctx, key)
if err != nil {
return fmt.Errorf("ENI %s not found: %w", eniId, err)
@@ -1019,7 +1019,7 @@ func (s *VPCServiceImpl) fillSGNames(ctx context.Context, accountID string, ids
continue
}
names[id] = ""
entry, err := s.sgKV.Get(ctx, utils.AccountKey(accountID, id))
entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, id))
if err != nil {
slog.WarnContext(ctx, "fillSGNames: SG read failed", "groupId", id, "err", err)
continue
@@ -1189,7 +1189,7 @@ func (s *VPCServiceImpl) validateSGAttachment(ctx context.Context, accountID str
// Each SG must exist in the caller's account and belong to the same VPC.
for _, sgId := range sgIds {
sgEntry, err := s.sgKV.Get(ctx, utils.AccountKey(accountID, sgId))
sgEntry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, sgId))
if err != nil {
return sgNotFoundError(sgId)
}
spinifex/handlers/ec2/vpc/security_group.go
+11 -11
View File
@@ -268,7 +268,7 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre
if err != nil {
return nil, fmt.Errorf("failed to marshal security group record: %w", err)
}
if _, err := s.sgKV.Put(ctx, utils.AccountKey(accountID, groupId), data); err != nil {
if _, err := s.sgKV.Put(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -299,7 +299,7 @@ func (s *VPCServiceImpl) DeleteSecurityGroup(ctx context.Context, input *ec2.Del
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -351,7 +351,7 @@ func (s *VPCServiceImpl) validateSGRuleReferences(ctx context.Context, accountID
if r.SourceSG == "" {
continue
}
entry, err := s.sgKV.Get(ctx, utils.AccountKey(accountID, r.SourceSG))
entry, err := s.sgKV.Get(ctx, kvutil.AccountKey(accountID, r.SourceSG))
if err != nil {
return sgNotFoundError(r.SourceSG)
}
@@ -1062,7 +1062,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupIngress(ctx context.Context, inpu
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1145,7 +1145,7 @@ func (s *VPCServiceImpl) AuthorizeSecurityGroupEgress(ctx context.Context, input
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1228,7 +1228,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupIngress(ctx context.Context, input *
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1295,7 +1295,7 @@ func (s *VPCServiceImpl) RevokeSecurityGroupEgress(ctx context.Context, input *e
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -1417,7 +1417,7 @@ func (s *VPCServiceImpl) updateSGRuleDescriptions(ctx context.Context, accountID
return errors.New(awserrors.ErrorInvalidParameterCombination)
}
key := utils.AccountKey(accountID, req.groupId)
key := kvutil.AccountKey(accountID, req.groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
return sgNotFoundError(req.groupId)
@@ -1609,7 +1609,7 @@ func (s *VPCServiceImpl) ModifySecurityGroupRules(ctx context.Context, input *ec
}
groupId := *input.GroupId
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
@@ -2186,7 +2186,7 @@ func (s *VPCServiceImpl) storeDefaultSecurityGroup(ctx context.Context, accountI
if err != nil {
return nil, false, fmt.Errorf("marshal default security group: %w", err)
}
if _, err := s.sgKV.Create(ctx, utils.AccountKey(accountID, groupId), data); err != nil {
if _, err := s.sgKV.Create(ctx, kvutil.AccountKey(accountID, groupId), data); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return record, false, nil
}
@@ -2215,7 +2215,7 @@ func (s *VPCServiceImpl) announceDefaultSecurityGroup(record *SecurityGroupRecor
// CannotDelete guard for default SGs. Used by DeleteVpc to cascade-delete the
// per-VPC default SG. Surfaces vpcd tear-down failures to the caller.
func (s *VPCServiceImpl) deleteSecurityGroupInternal(ctx context.Context, accountID, groupId string) error {
key := utils.AccountKey(accountID, groupId)
key := kvutil.AccountKey(accountID, groupId)
entry, err := s.sgKV.Get(ctx, key)
if err != nil {
return fmt.Errorf("read default security group: %w", err)
spinifex/handlers/ec2/vpc/security_group_modify_test.go
+3 -3
View File
@@ -15,7 +15,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -70,7 +70,7 @@ func tcpPeerRequest(port int64, peer string) *ec2.SecurityGroupRuleRequest {
// modify wrote nothing.
func storedSGRecord(t *testing.T, svc *VPCServiceImpl, sgID string) []byte {
t.Helper()
entry, err := svc.sgKV.Get(t.Context(), utils.AccountKey(testAccountID, sgID))
entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
require.NoError(t, err)
return entry.Value()
}
@@ -697,7 +697,7 @@ func TestRevokeSecurityGroupIngress_InvertedPortRangeStillRevocable(t *testing.T
vpcID := createTestVPC(t, svc, "10.0.0.0/16")
sgID := createTestSG(t, svc, vpcID, "stale-inverted-ports")
key := utils.AccountKey(testAccountID, sgID)
key := kvutil.AccountKey(testAccountID, sgID)
var rec SecurityGroupRecord
require.NoError(t, json.Unmarshal(storedSGRecord(t, svc, sgID), &rec))
rec.IngressRules = append(rec.IngressRules, SGRule{
spinifex/handlers/ec2/vpc/security_group_test.go
+2 -3
View File
@@ -14,7 +14,6 @@ import (
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -2123,7 +2122,7 @@ func seedAvailableVPC(t *testing.T, svc *VPCServiceImpl, vpcID string) {
rec := VPCRecord{VpcId: vpcID, CidrBlock: "10.0.0.0/16", State: "available"}
data, err := json.Marshal(rec)
require.NoError(t, err)
_, err = svc.vpcKV.Put(t.Context(), utils.AccountKey(testAccountID, vpcID), data)
_, err = svc.vpcKV.Put(t.Context(), kvutil.AccountKey(testAccountID, vpcID), data)
require.NoError(t, err)
}
@@ -2360,7 +2359,7 @@ func TestAuthorizeSecurityGroupIngress_AssignsRuleIDs(t *testing.T) {
}, testAccountID)
require.NoError(t, err)
entry, err := svc.sgKV.Get(t.Context(), utils.AccountKey(testAccountID, sgID))
entry, err := svc.sgKV.Get(t.Context(), kvutil.AccountKey(testAccountID, sgID))
require.NoError(t, err)
var record SecurityGroupRecord
require.NoError(t, json.Unmarshal(entry.Value(), &record))
spinifex/handlers/ec2/vpc/service_impl.go
+12 -12
View File
@@ -315,7 +315,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
if err != nil {
return nil, fmt.Errorf("failed to marshal VPC record: %w", err)
}
if _, err := s.vpcKV.Put(ctx, utils.AccountKey(accountID, vpcID), data); err != nil {
if _, err := s.vpcKV.Put(ctx, kvutil.AccountKey(accountID, vpcID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -350,7 +350,7 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu
// requireVPCExists returns InvalidVpcID.NotFound if the VPC doesn't exist for
// this account.
func (s *VPCServiceImpl) requireVPCExists(ctx context.Context, accountID, vpcId string) error {
if _, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, vpcId)); err != nil {
if _, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcId)); err != nil {
return vpcNotFoundError(vpcId)
}
return nil
@@ -363,7 +363,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
}
vpcID := *input.VpcId
key := utils.AccountKey(accountID, vpcID)
key := kvutil.AccountKey(accountID, vpcID)
if _, err := s.vpcKV.Get(ctx, key); err != nil {
// AWS-faithful: an absent VPC is NotFound (the tofu/SDK provider
@@ -514,7 +514,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu
return nil, errors.New(awserrors.ErrorServerInternal)
}
if rtbID != "" {
if err := s.rtbKV.Delete(ctx, utils.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
if err := s.rtbKV.Delete(ctx, kvutil.AccountKey(accountID, rtbID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
slog.ErrorContext(ctx, "DeleteVpc: main route table reap failed", "vpcId", vpcID, "routeTableId", rtbID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -710,7 +710,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
}
// Verify VPC exists and belongs to this account
vpcEntry, err := s.vpcKV.Get(ctx, utils.AccountKey(accountID, vpcID))
vpcEntry, err := s.vpcKV.Get(ctx, kvutil.AccountKey(accountID, vpcID))
if err != nil {
return nil, vpcNotFoundError(vpcID)
}
@@ -799,7 +799,7 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn
if err != nil {
return nil, fmt.Errorf("failed to marshal subnet record: %w", err)
}
if _, err := s.subnetKV.Put(ctx, utils.AccountKey(accountID, subnetID), data); err != nil {
if _, err := s.subnetKV.Put(ctx, kvutil.AccountKey(accountID, subnetID), data); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
@@ -822,7 +822,7 @@ func (s *VPCServiceImpl) DeleteSubnet(ctx context.Context, input *ec2.DeleteSubn
}
subnetID := *input.SubnetId
key := utils.AccountKey(accountID, subnetID)
key := kvutil.AccountKey(accountID, subnetID)
// Read subnet record before deletion (needed for vpcd event)
subnetEntry, err := s.subnetKV.Get(ctx, key)
@@ -1349,7 +1349,7 @@ func (s *VPCServiceImpl) ModifySubnetAttribute(ctx context.Context, input *ec2.M
}
subnetID := *input.SubnetId
key := utils.AccountKey(accountID, subnetID)
key := kvutil.AccountKey(accountID, subnetID)
entry, err := s.subnetKV.Get(ctx, key)
if err != nil {
@@ -1401,7 +1401,7 @@ func (s *VPCServiceImpl) ModifyVpcAttribute(ctx context.Context, input *ec2.Modi
}
vpcID := *input.VpcId
key := utils.AccountKey(accountID, vpcID)
key := kvutil.AccountKey(accountID, vpcID)
entry, err := s.vpcKV.Get(ctx, key)
if err != nil {
@@ -1449,7 +1449,7 @@ func (s *VPCServiceImpl) DescribeVpcAttribute(ctx context.Context, input *ec2.De
}
vpcID := *input.VpcId
key := utils.AccountKey(accountID, vpcID)
key := kvutil.AccountKey(accountID, vpcID)
entry, err := s.vpcKV.Get(ctx, key)
if err != nil {
@@ -1550,7 +1550,7 @@ func (s *VPCServiceImpl) writeMainRouteTable(ctx context.Context, accountID, vpc
if err != nil {
return false, fmt.Errorf("marshal main route table: %w", err)
}
if _, err := s.rtbKV.Create(ctx, utils.AccountKey(accountID, rtbID), data); err != nil {
if _, err := s.rtbKV.Create(ctx, kvutil.AccountKey(accountID, rtbID), data); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return false, nil
}
@@ -1643,7 +1643,7 @@ func (s *VPCServiceImpl) GetSubnet(accountID, subnetId string) (*SubnetRecord, e
}
func (s *VPCServiceImpl) getSubnet(ctx context.Context, accountID, subnetId string) (*SubnetRecord, error) {
key := utils.AccountKey(accountID, subnetId)
key := kvutil.AccountKey(accountID, subnetId)
entry, err := s.subnetKV.Get(ctx, key)
if err != nil {
return nil, fmt.Errorf("subnet %s not found: %w", subnetId, err)
spinifex/utils/account.go
-5
View File
@@ -3,11 +3,6 @@ package utils
// GlobalAccountID is the root/system account ID.
const GlobalAccountID = "000000000000"
// AccountKey returns a KV key scoped to an account: "{accountID}.{resourceID}".
func AccountKey(accountID, resourceID string) string {
return accountID + "." + resourceID
}
// IsAccountID checks if a string is a valid 12-digit AWS account ID.
func IsAccountID(s string) bool {
if len(s) != 12 {
spinifex/utils/account_test.go
-18
View File
@@ -4,24 +4,6 @@ import (
"testing"
)
func TestAccountKey(t *testing.T) {
tests := []struct {
accountID string
resourceID string
want string
}{
{"000000000000", "vpc-123", "000000000000.vpc-123"},
{"123456789012", "igw-abc", "123456789012.igw-abc"},
{"", "vol-1", ".vol-1"},
}
for _, tt := range tests {
got := AccountKey(tt.accountID, tt.resourceID)
if got != tt.want {
t.Errorf("AccountKey(%q, %q) = %q, want %q", tt.accountID, tt.resourceID, got, tt.want)
}
}
}
func TestIsAccountID(t *testing.T) {
tests := []struct {
input string
spinifex/utils/tagmirror.go
+2 -1
View File
@@ -9,6 +9,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
@@ -78,7 +79,7 @@ func RemoveTagsMut(input *ec2.DeleteTagsInput) func(map[string]string) {
// applying mut. A resource absent from this store is skipped (its tags live
// elsewhere).
func UpdateKVRecordTags[R any](ctx context.Context, kv jetstream.KeyValue, accountID, resourceID string, mut func(*R)) error {
key := AccountKey(accountID, resourceID)
key := kvutil.AccountKey(accountID, resourceID)
entry, err := kv.Get(ctx, key)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
spinifex/utils/tagmirror_test.go
+10 -11
View File
@@ -6,8 +6,9 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -18,9 +19,7 @@ type tagMirrorRecord struct {
Tags map[string]string `json:"tags"`
}
// seedKV creates a bucket on the jetstream API and populates it. Package utils
// cannot use kvutil's helpers — kvutil imports utils — so a fresh JetStream
// server is bootstrapped here directly.
// seedKV creates a bucket on the JetStream API and populates it.
func seedKV(t *testing.T, bucket string, entries map[string][]byte) jetstream.KeyValue {
t.Helper()
_, nc, _ := testutil.StartTestJetStream(t)
@@ -76,15 +75,15 @@ func TestUpdateKVRecordTags(t *testing.T) {
rec, err := json.Marshal(&tagMirrorRecord{Name: "r1", Tags: map[string]string{"a": "1"}})
require.NoError(t, err)
kv := seedKV(t, "tagmirror-test", map[string][]byte{
AccountKey("acct", "res-1"): rec,
AccountKey("acct", "res-bad"): []byte("not-json"),
kvutil.AccountKey("acct", "res-1"): rec,
kvutil.AccountKey("acct", "res-bad"): []byte("not-json"),
})
t.Run("mutates and persists", func(t *testing.T) {
require.NoError(t, UpdateKVRecordTags(t.Context(), kv, "acct", "res-1", func(r *tagMirrorRecord) {
r.Tags["b"] = "2"
}))
entry, err := kv.Get(t.Context(), AccountKey("acct", "res-1"))
entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", "res-1"))
require.NoError(t, err)
var got tagMirrorRecord
require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -111,14 +110,14 @@ func TestMirrorKVRecordTags(t *testing.T) {
nilTags, err := json.Marshal(&tagMirrorRecord{})
require.NoError(t, err)
kv := seedKV(t, "tagmirror-mirror-test", map[string][]byte{
AccountKey("acct", "vol-1"): withTags,
AccountKey("acct", "vol-2"): nilTags,
kvutil.AccountKey("acct", "vol-1"): withTags,
kvutil.AccountKey("acct", "vol-2"): nilTags,
})
tagsOf := func(r *tagMirrorRecord) *map[string]string { return &r.Tags }
readTags := func(t *testing.T, id string) map[string]string {
entry, err := kv.Get(t.Context(), AccountKey("acct", id))
entry, err := kv.Get(t.Context(), kvutil.AccountKey("acct", id))
require.NoError(t, err)
var got tagMirrorRecord
require.NoError(t, json.Unmarshal(entry.Value(), &got))
@@ -151,7 +150,7 @@ func TestMirrorKVRecordTags(t *testing.T) {
// must never be touched even though it exists in this bucket.
otherRec, err := json.Marshal(&tagMirrorRecord{Tags: map[string]string{"x": "y"}})
require.NoError(t, err)
_, err = kv.Put(t.Context(), AccountKey("acct", "snap-other"), otherRec)
_, err = kv.Put(t.Context(), kvutil.AccountKey("acct", "snap-other"), otherRec)
require.NoError(t, err)
require.NoError(t, MirrorKVRecordTags(t.Context(), kv, "acct", "vol-", resources, tagsOf,
MergeTagsMut(&ec2.CreateTagsInput{Tags: []*ec2.Tag{
tests/e2e/single/account_test.go
+1 -1
View File
@@ -506,7 +506,7 @@ func runAccountScoping(t *testing.T, fix *Fixture) {
// TestAccountScoping_IGWEIGW and TestAccountScoping_Settings.
// VPCServiceImpl/IGWServiceImpl/EgressOnlyIGWServiceImpl/
// AccountSettingsServiceImpl all resolve ownership from a plain
// account-scoped KV key (utils.AccountKey(accountID, id) or an
// account-scoped KV key (kvutil.AccountKey(accountID, id) or an
// account-keyed settings record) — no vm.Manager, no real guest, no OVN
// state that only exists once a VPC has an attached instance. The live
// variants proved nothing beyond what the integration tier now asserts