diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index a5d070350..6d20167af 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1651,9 +1651,10 @@ func (d *Daemon) startCluster() error { // pools that hand out routable public IPs (pool mode always; nat mode only // when a public pool rides alongside the transit segment). d.vpcService.SetDefaultPublicIPMapping(d.hasPublicIPPools()) - // Project vpc/subnet/sg/eni creation-time tags into the central tag store - // so describe-tags agrees with each resource's own describe from birth. - d.vpcService.SetCentralTagWriter(d.tagsService) + // Keep the central tag store in step with vpc/subnet/sg/eni records, so + // describe-tags agrees with each resource's own describe from birth and + // stops answering for it once it is deleted. + d.vpcService.SetCentralTagStore(d.tagsService) d.routeTableService, err = initServiceWithRetry("RouteTable service", func() (*handlers_ec2_routetable.RouteTableServiceImpl, error) { return handlers_ec2_routetable.NewRouteTableServiceImplWithNATS(d.ctx, d.config, d.natsConn) diff --git a/spinifex/handlers/ec2/tags/service_impl.go b/spinifex/handlers/ec2/tags/service_impl.go index 6006264ea..94be18351 100644 --- a/spinifex/handlers/ec2/tags/service_impl.go +++ b/spinifex/handlers/ec2/tags/service_impl.go @@ -32,8 +32,8 @@ var _ TagsService = (*TagsServiceImpl)(nil) // Ensure TagsServiceImpl can project instance record tags into the store. var _ handlers_ec2_instance.InstanceTagWriter = (*TagsServiceImpl)(nil) -// Ensure TagsServiceImpl can project vpc/subnet/sg/eni record tags into the store. -var _ handlers_ec2_vpc.CentralTagWriter = (*TagsServiceImpl)(nil) +// Ensure TagsServiceImpl can both project and clear vpc/subnet/sg/eni record tags. +var _ handlers_ec2_vpc.CentralTagStore = (*TagsServiceImpl)(nil) // TagsServiceImpl implements TagsService over a JetStream KV bucket, one entry // per resource, scoped by account in the key. @@ -183,9 +183,10 @@ func (s *TagsServiceImpl) PutResourceTags(ctx context.Context, accountID, resour return s.putTags(ctx, accountID, resourceID, tags) } -// DeleteAllTags removes the stored tags for a resource. Used on instance -// terminate so describe-tags stops reporting the instance while the terminated -// record keeps its tags until TTL. Idempotent: an absent entry is not an error. +// DeleteAllTags removes the stored tags for a resource, so describe-tags stops +// reporting a resource that is gone. Instance terminate uses it while the +// terminated record keeps its own tags until TTL; the vpc delete paths use it +// to retire the entry outright. Idempotent: an absent entry is not an error. func (s *TagsServiceImpl) DeleteAllTags(ctx context.Context, accountID, resourceID string) error { if err := s.deleteTagsEntry(ctx, accountID, resourceID); err != nil { return err diff --git a/spinifex/handlers/ec2/vpc/central_tags_test.go b/spinifex/handlers/ec2/vpc/central_tags_test.go index f61954270..823fe232f 100644 --- a/spinifex/handlers/ec2/vpc/central_tags_test.go +++ b/spinifex/handlers/ec2/vpc/central_tags_test.go @@ -11,15 +11,16 @@ import ( "github.com/stretchr/testify/require" ) -// fakeCentralTagWriter records the tag map passed to PutResourceTags per -// resource, and can be made to fail so tests can prove a central-store error -// never fails the create it is projecting from. -type fakeCentralTagWriter struct { - calls map[string]map[string]string - err error +// fakeCentralTagStore records the tag map passed to PutResourceTags per +// resource and the ids passed to DeleteAllTags, and can be made to fail so +// tests can prove a central-store error never fails the operation it tracks. +type fakeCentralTagStore struct { + calls map[string]map[string]string + deleted []string + err error } -func (f *fakeCentralTagWriter) PutResourceTags(_ context.Context, _, resourceID string, tags map[string]string) error { +func (f *fakeCentralTagStore) PutResourceTags(_ context.Context, _, resourceID string, tags map[string]string) error { if f.calls == nil { f.calls = map[string]map[string]string{} } @@ -27,11 +28,16 @@ func (f *fakeCentralTagWriter) PutResourceTags(_ context.Context, _, resourceID return f.err } +func (f *fakeCentralTagStore) DeleteAllTags(_ context.Context, _, resourceID string) error { + f.deleted = append(f.deleted, resourceID) + return f.err +} + func TestCreateVpc_ProjectsTagsToCentralStore(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - writer := &fakeCentralTagWriter{} - svc.SetCentralTagWriter(writer) + writer := &fakeCentralTagStore{} + svc.SetCentralTagStore(writer) out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ CidrBlock: aws.String("10.0.0.0/16"), @@ -54,8 +60,8 @@ func TestCreateVpc_ProjectsTagsToCentralStore(t *testing.T) { func TestCreateSubnet_ProjectsTagsToCentralStore(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - writer := &fakeCentralTagWriter{} - svc.SetCentralTagWriter(writer) + writer := &fakeCentralTagStore{} + svc.SetCentralTagStore(writer) vpcID := createTestVPC(t, svc, "10.0.0.0/16") out, err := svc.CreateSubnet(context.Background(), &ec2.CreateSubnetInput{ @@ -80,8 +86,8 @@ func TestCreateSubnet_ProjectsTagsToCentralStore(t *testing.T) { func TestCreateSecurityGroup_ProjectsTagsToCentralStore(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - writer := &fakeCentralTagWriter{} - svc.SetCentralTagWriter(writer) + writer := &fakeCentralTagStore{} + svc.SetCentralTagStore(writer) vpcID := createTestVPC(t, svc, "10.0.0.0/16") out, err := svc.CreateSecurityGroup(context.Background(), &ec2.CreateSecurityGroupInput{ @@ -107,8 +113,8 @@ func TestCreateSecurityGroup_ProjectsTagsToCentralStore(t *testing.T) { func TestCreateVpc_NoTags_SkipsCentralStoreWrite(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - writer := &fakeCentralTagWriter{} - svc.SetCentralTagWriter(writer) + writer := &fakeCentralTagStore{} + svc.SetCentralTagStore(writer) out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ CidrBlock: aws.String("10.0.0.0/16"), @@ -118,10 +124,10 @@ func TestCreateVpc_NoTags_SkipsCentralStoreWrite(t *testing.T) { assert.NotContains(t, writer.calls, *out.Vpc.VpcId) } -func TestCreateVpc_NilCentralTagWriter_CreateSucceeds(t *testing.T) { +func TestCreateVpc_NilCentralTagStore_CreateSucceeds(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - // No SetCentralTagWriter call: centralTags stays nil. + // No SetCentralTagStore call: centralTags stays nil. out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ CidrBlock: aws.String("10.0.0.0/16"), @@ -138,11 +144,11 @@ func TestCreateVpc_NilCentralTagWriter_CreateSucceeds(t *testing.T) { require.Len(t, out.Vpc.Tags, 1) } -func TestCreateVpc_CentralTagWriterError_DoesNotFailCreate(t *testing.T) { +func TestCreateVpc_CentralTagStoreError_DoesNotFailCreate(t *testing.T) { t.Parallel() svc := setupTestVPCService(t) - writer := &fakeCentralTagWriter{err: errors.New("central store unavailable")} - svc.SetCentralTagWriter(writer) + writer := &fakeCentralTagStore{err: errors.New("central store unavailable")} + svc.SetCentralTagStore(writer) out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ CidrBlock: aws.String("10.0.0.0/16"), @@ -158,3 +164,168 @@ func TestCreateVpc_CentralTagWriterError_DoesNotFailCreate(t *testing.T) { require.NoError(t, err) require.Len(t, out.Vpc.Tags, 1) } + +// Deleting a resource must retire its central tag entry, whatever the type. +// The table drives each delete path through the same store so the assertion is +// about the shared teardown rather than one handler: a type added to the +// package without a clear call fails here rather than leaking quietly. +func TestDelete_ClearsCentralTagStore(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + // create returns the id of a resource of this type, already tagged. + create func(t *testing.T, svc *VPCServiceImpl) string + del func(t *testing.T, svc *VPCServiceImpl, id string) + }{ + { + name: "security group", + create: func(t *testing.T, svc *VPCServiceImpl) string { + vpcID := createTestVPC(t, svc, "10.1.0.0/16") + out, err := svc.CreateSecurityGroup(context.Background(), &ec2.CreateSecurityGroupInput{ + GroupName: aws.String("doomed-sg"), + Description: aws.String("doomed"), + VpcId: aws.String(vpcID), + TagSpecifications: []*ec2.TagSpecification{ + {ResourceType: aws.String("security-group"), Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("doomed")}, + }}, + }, + }, testAccountID) + require.NoError(t, err) + return *out.GroupId + }, + del: func(t *testing.T, svc *VPCServiceImpl, id string) { + _, err := svc.DeleteSecurityGroup(context.Background(), + &ec2.DeleteSecurityGroupInput{GroupId: aws.String(id)}, testAccountID) + require.NoError(t, err) + }, + }, + { + name: "subnet", + create: func(t *testing.T, svc *VPCServiceImpl) string { + vpcID := createTestVPC(t, svc, "10.2.0.0/16") + out, err := svc.CreateSubnet(context.Background(), &ec2.CreateSubnetInput{ + VpcId: aws.String(vpcID), + CidrBlock: aws.String("10.2.1.0/24"), + TagSpecifications: []*ec2.TagSpecification{ + {ResourceType: aws.String("subnet"), Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("doomed")}, + }}, + }, + }, testAccountID) + require.NoError(t, err) + return *out.Subnet.SubnetId + }, + del: func(t *testing.T, svc *VPCServiceImpl, id string) { + _, err := svc.DeleteSubnet(context.Background(), + &ec2.DeleteSubnetInput{SubnetId: aws.String(id)}, testAccountID) + require.NoError(t, err) + }, + }, + { + // The interface that first exposed this gap. Both the public delete + // and the forced instance teardown run the same path, so covering + // one covers the ENI reclaimed by a task going away. + name: "network interface", + create: func(t *testing.T, svc *VPCServiceImpl) string { + vpcID := createTestVPC(t, svc, "10.7.0.0/16") + subnetID := createTestSubnet(t, svc, vpcID, "10.7.1.0/24") + out, err := svc.CreateNetworkInterface(context.Background(), &ec2.CreateNetworkInterfaceInput{ + SubnetId: aws.String(subnetID), + TagSpecifications: []*ec2.TagSpecification{ + {ResourceType: aws.String("network-interface"), Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("doomed")}, + }}, + }, + }, testAccountID) + require.NoError(t, err) + return *out.NetworkInterface.NetworkInterfaceId + }, + del: func(t *testing.T, svc *VPCServiceImpl, id string) { + _, err := svc.DeleteNetworkInterface(context.Background(), + &ec2.DeleteNetworkInterfaceInput{NetworkInterfaceId: aws.String(id)}, testAccountID) + require.NoError(t, err) + }, + }, + { + name: "vpc", + create: func(t *testing.T, svc *VPCServiceImpl) string { + out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ + CidrBlock: aws.String("10.3.0.0/16"), + TagSpecifications: []*ec2.TagSpecification{ + {ResourceType: aws.String("vpc"), Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("doomed")}, + }}, + }, + }, testAccountID) + require.NoError(t, err) + return *out.Vpc.VpcId + }, + del: func(t *testing.T, svc *VPCServiceImpl, id string) { + _, err := svc.DeleteVpc(context.Background(), + &ec2.DeleteVpcInput{VpcId: aws.String(id)}, testAccountID) + require.NoError(t, err) + }, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + store := &fakeCentralTagStore{} + svc.SetCentralTagStore(store) + + id := tc.create(t, svc) + require.Contains(t, store.calls, id, "precondition: the create must have tagged it") + + tc.del(t, svc, id) + assert.Contains(t, store.deleted, id, "the deleted resource's tags must be retired") + }) + } +} + +// DeleteVpc cascades the default security group, which no caller ever deletes +// by hand. Asserted separately because the id is not the one under test. +func TestDeleteVpc_ClearsTheCascadedDefaultSecurityGroup(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + store := &fakeCentralTagStore{} + svc.SetCentralTagStore(store) + + vpcID := createTestVPC(t, svc, "10.4.0.0/16") + defaultSG, err := svc.FindDefaultSGForVPC(testAccountID, vpcID) + require.NoError(t, err) + require.NotEmpty(t, defaultSG, "precondition: CreateVpc makes a default SG") + + _, err = svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID) + require.NoError(t, err) + + assert.Contains(t, store.deleted, defaultSG, "the cascaded default SG must not keep its tags") +} + +func TestDelete_NilCentralTagStore_DeleteSucceeds(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + // No SetCentralTagStore call: centralTags stays nil. + + vpcID := createTestVPC(t, svc, "10.5.0.0/16") + _, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID) + require.NoError(t, err) +} + +// The resource is already gone by the time the store is called, so a store +// failure must not fail the delete: the caller cannot retry something that has +// happened, and would be left believing the resource survived. +func TestDelete_CentralTagStoreError_DoesNotFailDelete(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + store := &fakeCentralTagStore{err: errors.New("central store unavailable")} + svc.SetCentralTagStore(store) + + vpcID := createTestVPC(t, svc, "10.6.0.0/16") + _, err := svc.DeleteVpc(context.Background(), &ec2.DeleteVpcInput{VpcId: aws.String(vpcID)}, testAccountID) + require.NoError(t, err) + assert.Contains(t, store.deleted, vpcID, "the clear must still have been attempted") +} diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index 0e76d9fbe..cbe604360 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -275,6 +275,10 @@ func (s *VPCServiceImpl) deleteNetworkInterface(ctx context.Context, eniId, acco if err := s.eniKV.Delete(ctx, key); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } + // Both the public delete and the forced instance teardown land here, which + // is what covers an ENI reclaimed by a task or instance going away rather + // than by an API call. + s.clearRecordTags(ctx, accountID, eniId) slog.InfoContext(ctx, "DeleteNetworkInterface completed", "eniId", eniId, "accountID", accountID) diff --git a/spinifex/handlers/ec2/vpc/security_group.go b/spinifex/handlers/ec2/vpc/security_group.go index 1f440538b..165ebcc37 100644 --- a/spinifex/handlers/ec2/vpc/security_group.go +++ b/spinifex/handlers/ec2/vpc/security_group.go @@ -302,6 +302,7 @@ func (s *VPCServiceImpl) DeleteSecurityGroup(ctx context.Context, input *ec2.Del if err := s.sgKV.Delete(ctx, key); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } + s.clearRecordTags(ctx, accountID, groupId) slog.InfoContext(ctx, "DeleteSecurityGroup completed", "groupId", groupId, "accountID", accountID) @@ -1714,6 +1715,11 @@ func (s *VPCServiceImpl) deleteSecurityGroupInternal(ctx context.Context, accoun if err := s.sgKV.Delete(ctx, key); err != nil { return fmt.Errorf("delete default security group: %w", err) } + // The cascade from DeleteVpc reaches the default SG only through here, so + // without this the one group the caller never deletes by hand is the one + // that always leaks its tags. + s.clearRecordTags(ctx, accountID, groupId) + return s.requestSGEvent("vpc.delete-sg", SGEvent{ GroupId: groupId, VpcId: record.VpcId, diff --git a/spinifex/handlers/ec2/vpc/service.go b/spinifex/handlers/ec2/vpc/service.go index 584b4b6f4..bf19e09d3 100644 --- a/spinifex/handlers/ec2/vpc/service.go +++ b/spinifex/handlers/ec2/vpc/service.go @@ -34,9 +34,11 @@ type VPCService interface { UpdateSecurityGroupRuleDescriptionsEgress(ctx context.Context, input *ec2.UpdateSecurityGroupRuleDescriptionsEgressInput, accountID string) (*ec2.UpdateSecurityGroupRuleDescriptionsEgressOutput, error) } -// CentralTagWriter projects a create path's record tags into the central tag -// store so DescribeTags agrees with the record from birth. Implemented by +// CentralTagStore keeps the central tag index in step with this package's +// records, so DescribeTags agrees with a resource's own describe for as long as +// the resource exists and says nothing about it afterwards. Implemented by // handlers/ec2/tags.TagsServiceImpl. -type CentralTagWriter interface { +type CentralTagStore interface { PutResourceTags(ctx context.Context, accountID, resourceID string, tags map[string]string) error + DeleteAllTags(ctx context.Context, accountID, resourceID string) error } diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 9abeae168..1417de95a 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -92,9 +92,9 @@ type VPCServiceImpl struct { eipKV jetstream.KeyValue // Optional: injected after construction so create paths can project their - // stored tags into the central tag store. A nil value leaves create - // behaviour unchanged. - centralTags CentralTagWriter + // stored tags into the central tag store and delete paths can clear them. + // A nil value leaves both unchanged. + centralTags CentralTagStore // disableDefaultPublicIP seeds default subnets with MapPublicIpOnLaunch=false // (non-pool external modes have no public IPs to assign). Zero value keeps @@ -115,10 +115,11 @@ func (s *VPCServiceImpl) SetExternalIPAM(ipam *ExternalIPAM, eipKV jetstream.Key s.eipKV = eipKV } -// SetCentralTagWriter injects the central tag store writer so create paths -// project their record tags into it. Nil leaves create behaviour unchanged. -func (s *VPCServiceImpl) SetCentralTagWriter(w CentralTagWriter) { - s.centralTags = w +// SetCentralTagStore injects the central tag store so create paths project +// their record tags into it and delete paths clear them. Nil leaves both +// unchanged. +func (s *VPCServiceImpl) SetCentralTagStore(st CentralTagStore) { + s.centralTags = st } // localAZ returns the node's local availability zone, sourced from @@ -494,6 +495,7 @@ func (s *VPCServiceImpl) DeleteVpc(ctx context.Context, input *ec2.DeleteVpcInpu return nil, errors.New(awserrors.ErrorServerInternal) } s.releaseDefaultVPCClaim(ctx, accountID, vpcID) + s.clearRecordTags(ctx, accountID, vpcID) slog.InfoContext(ctx, "DeleteVpc completed", "vpcId", vpcID, "accountID", accountID) @@ -817,6 +819,7 @@ func (s *VPCServiceImpl) DeleteSubnet(ctx context.Context, input *ec2.DeleteSubn if err := s.subnetKV.Delete(ctx, key); err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } + s.clearRecordTags(ctx, accountID, subnetID) slog.InfoContext(ctx, "DeleteSubnet completed", "subnetId", subnetID, "accountID", accountID) @@ -1085,6 +1088,25 @@ func (s *VPCServiceImpl) projectRecordTags(ctx context.Context, accountID, resou } } +// clearRecordTags drops a deleted resource's entry from the central tag store, +// the delete-direction counterpart to projectRecordTags. On AWS a resource's +// tags go with it; without this the index answers DescribeTags for resources +// the caller can no longer describe, and a recycled id inherits them. +// +// Call it only once the record delete has succeeded. Clearing first would strip +// the tags from a resource that then fails to delete and stays live. A failure +// is logged and swallowed for the same reason the write side swallows one: the +// resource is already gone, and failing the delete over the index would leave +// the caller unable to retry something that has happened. +func (s *VPCServiceImpl) clearRecordTags(ctx context.Context, accountID, resourceID string) { + if s.centralTags == nil { + return + } + if err := s.centralTags.DeleteAllTags(ctx, accountID, resourceID); err != nil { + slog.ErrorContext(ctx, "central tag store clear failed", "resourceId", resourceID, "err", err) + } +} + // updateRecordTags applies mut to the tag map of the subnet-, vpc-, sg-, or // eni-scoped record identified by resourceID. Other resource ids are a no-op. func (s *VPCServiceImpl) updateRecordTags(ctx context.Context, accountID, resourceID string, mut func(map[string]string)) error {