diff --git a/spinifex/foundation/state/kvutil/account_key.go b/spinifex/foundation/state/kvutil/account_key.go new file mode 100644 index 000000000..6ec650021 --- /dev/null +++ b/spinifex/foundation/state/kvutil/account_key.go @@ -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 +} diff --git a/spinifex/foundation/state/kvutil/account_key_test.go b/spinifex/foundation/state/kvutil/account_key_test.go new file mode 100644 index 000000000..2617c0cb4 --- /dev/null +++ b/spinifex/foundation/state/kvutil/account_key_test.go @@ -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) + } + } +} diff --git a/spinifex/handlers/ec2/eigw/service_impl.go b/spinifex/handlers/ec2/eigw/service_impl.go index 081963f87..feb403088 100644 --- a/spinifex/handlers/ec2/eigw/service_impl.go +++ b/spinifex/handlers/ec2/eigw/service_impl.go @@ -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 { diff --git a/spinifex/handlers/ec2/eigw/service_impl_test.go b/spinifex/handlers/ec2/eigw/service_impl_test.go index b49b45d8d..5bbcad58c 100644 --- a/spinifex/handlers/ec2/eigw/service_impl_test.go +++ b/spinifex/handlers/ec2/eigw/service_impl_test.go @@ -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 diff --git a/spinifex/handlers/ec2/eip/allocate_idempotency.go b/spinifex/handlers/ec2/eip/allocate_idempotency.go index a9ab06766..63a21c8e7 100644 --- a/spinifex/handlers/ec2/eip/allocate_idempotency.go +++ b/spinifex/handlers/ec2/eip/allocate_idempotency.go @@ -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", diff --git a/spinifex/handlers/ec2/eip/service_impl.go b/spinifex/handlers/ec2/eip/service_impl.go index ae42f61e0..d020fd39f 100644 --- a/spinifex/handlers/ec2/eip/service_impl.go +++ b/spinifex/handlers/ec2/eip/service_impl.go @@ -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 diff --git a/spinifex/handlers/ec2/igw/service_impl.go b/spinifex/handlers/ec2/igw/service_impl.go index aa3567e87..5d39cd5be 100644 --- a/spinifex/handlers/ec2/igw/service_impl.go +++ b/spinifex/handlers/ec2/igw/service_impl.go @@ -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, diff --git a/spinifex/handlers/ec2/igw/service_impl_test.go b/spinifex/handlers/ec2/igw/service_impl_test.go index 324b9be68..6474a8469 100644 --- a/spinifex/handlers/ec2/igw/service_impl_test.go +++ b/spinifex/handlers/ec2/igw/service_impl_test.go @@ -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 diff --git a/spinifex/handlers/ec2/launchtemplate/service_impl.go b/spinifex/handlers/ec2/launchtemplate/service_impl.go index 2e3e2dbba..07a2ded7f 100644 --- a/spinifex/handlers/ec2/launchtemplate/service_impl.go +++ b/spinifex/handlers/ec2/launchtemplate/service_impl.go @@ -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-.v. 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 { diff --git a/spinifex/handlers/ec2/launchtemplate/service_impl_test.go b/spinifex/handlers/ec2/launchtemplate/service_impl_test.go index 30e7ede3b..03a161705 100644 --- a/spinifex/handlers/ec2/launchtemplate/service_impl_test.go +++ b/spinifex/handlers/ec2/launchtemplate/service_impl_test.go @@ -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{ diff --git a/spinifex/handlers/ec2/natgw/lifecycle_invariants_test.go b/spinifex/handlers/ec2/natgw/lifecycle_invariants_test.go index b3334c1f7..e6078883e 100644 --- a/spinifex/handlers/ec2/natgw/lifecycle_invariants_test.go +++ b/spinifex/handlers/ec2/natgw/lifecycle_invariants_test.go @@ -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)) diff --git a/spinifex/handlers/ec2/natgw/service_impl.go b/spinifex/handlers/ec2/natgw/service_impl.go index ea81ca84a..050f1ae4a 100644 --- a/spinifex/handlers/ec2/natgw/service_impl.go +++ b/spinifex/handlers/ec2/natgw/service_impl.go @@ -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) diff --git a/spinifex/handlers/ec2/natgw/service_impl_test.go b/spinifex/handlers/ec2/natgw/service_impl_test.go index abd0f1bfc..e751c66c2 100644 --- a/spinifex/handlers/ec2/natgw/service_impl_test.go +++ b/spinifex/handlers/ec2/natgw/service_impl_test.go @@ -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) diff --git a/spinifex/handlers/ec2/placementgroup/service_impl.go b/spinifex/handlers/ec2/placementgroup/service_impl.go index dcb88db77..64d9b5988 100644 --- a/spinifex/handlers/ec2/placementgroup/service_impl.go +++ b/spinifex/handlers/ec2/placementgroup/service_impl.go @@ -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 { diff --git a/spinifex/handlers/ec2/placementgroup/service_impl_test.go b/spinifex/handlers/ec2/placementgroup/service_impl_test.go index 2261e6b91..65091959e 100644 --- a/spinifex/handlers/ec2/placementgroup/service_impl_test.go +++ b/spinifex/handlers/ec2/placementgroup/service_impl_test.go @@ -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 diff --git a/spinifex/handlers/ec2/routetable/service_impl.go b/spinifex/handlers/ec2/routetable/service_impl.go index 0a998f5b5..746f63d93 100644 --- a/spinifex/handlers/ec2/routetable/service_impl.go +++ b/spinifex/handlers/ec2/routetable/service_impl.go @@ -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 diff --git a/spinifex/handlers/ec2/routetable/service_impl_test.go b/spinifex/handlers/ec2/routetable/service_impl_test.go index 2de2bfdf0..9f8d7c93c 100644 --- a/spinifex/handlers/ec2/routetable/service_impl_test.go +++ b/spinifex/handlers/ec2/routetable/service_impl_test.go @@ -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) diff --git a/spinifex/handlers/ec2/spotinstance/service_impl.go b/spinifex/handlers/ec2/spotinstance/service_impl.go index 8b3db783a..3bafb9a29 100644 --- a/spinifex/handlers/ec2/spotinstance/service_impl.go +++ b/spinifex/handlers/ec2/spotinstance/service_impl.go @@ -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 { diff --git a/spinifex/handlers/ec2/vpc/default_vpc.go b/spinifex/handlers/ec2/vpc/default_vpc.go index b8a0e89bf..4cfef33eb 100644 --- a/spinifex/handlers/ec2/vpc/default_vpc.go +++ b/spinifex/handlers/ec2/vpc/default_vpc.go @@ -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, diff --git a/spinifex/handlers/ec2/vpc/default_vpc_test.go b/spinifex/handlers/ec2/vpc/default_vpc_test.go index 781e95d80..974e09447 100644 --- a/spinifex/handlers/ec2/vpc/default_vpc_test.go +++ b/spinifex/handlers/ec2/vpc/default_vpc_test.go @@ -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, }) diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index 5f84447ab..eac97b29d 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -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) } diff --git a/spinifex/handlers/ec2/vpc/security_group.go b/spinifex/handlers/ec2/vpc/security_group.go index 96f7f788c..a5c537576 100644 --- a/spinifex/handlers/ec2/vpc/security_group.go +++ b/spinifex/handlers/ec2/vpc/security_group.go @@ -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) diff --git a/spinifex/handlers/ec2/vpc/security_group_modify_test.go b/spinifex/handlers/ec2/vpc/security_group_modify_test.go index c735a7988..927346054 100644 --- a/spinifex/handlers/ec2/vpc/security_group_modify_test.go +++ b/spinifex/handlers/ec2/vpc/security_group_modify_test.go @@ -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{ diff --git a/spinifex/handlers/ec2/vpc/security_group_test.go b/spinifex/handlers/ec2/vpc/security_group_test.go index 2b6ce671e..724335487 100644 --- a/spinifex/handlers/ec2/vpc/security_group_test.go +++ b/spinifex/handlers/ec2/vpc/security_group_test.go @@ -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)) diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 25bc94849..30e3e0976 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -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) diff --git a/spinifex/utils/account.go b/spinifex/utils/account.go index 3c67b5078..507a2c14f 100644 --- a/spinifex/utils/account.go +++ b/spinifex/utils/account.go @@ -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 { diff --git a/spinifex/utils/account_test.go b/spinifex/utils/account_test.go index a59fbdecd..1b749d828 100644 --- a/spinifex/utils/account_test.go +++ b/spinifex/utils/account_test.go @@ -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 diff --git a/spinifex/utils/tagmirror.go b/spinifex/utils/tagmirror.go index 395faa5a2..cf28b787e 100644 --- a/spinifex/utils/tagmirror.go +++ b/spinifex/utils/tagmirror.go @@ -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) { diff --git a/spinifex/utils/tagmirror_test.go b/spinifex/utils/tagmirror_test.go index 385005801..5cc00a176 100644 --- a/spinifex/utils/tagmirror_test.go +++ b/spinifex/utils/tagmirror_test.go @@ -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{ diff --git a/tests/e2e/single/account_test.go b/tests/e2e/single/account_test.go index f3f267d98..e6abe0e43 100644 --- a/tests/e2e/single/account_test.go +++ b/tests/e2e/single/account_test.go @@ -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