From c674e4b5ce6b8b0cfd710c20cfd6a04edfcec51e Mon Sep 17 00:00:00 2001 From: Josh Siviour Date: Thu, 17 Sep 2026 14:28:29 +1000 Subject: [PATCH] fix(ec2): project creation-time tags into the central tag store (#1063) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A tag set at creation time through TagSpecifications was written only onto the resource's own KV record. DescribeTags reads a separate central index at tags/{accountID}/{resourceID}.json, which only CreateTags writes, so a resource born with tags described as having none until something tagged it again. Terraform finds this immediately: terraform-aws-modules/eks reads a cluster security group's tags back after creating it, and aws_ec2_tag fails against a resource whose tags the tag API cannot see. The four VPC create paths now project their stored tags into the central store through an optional CentralTagWriter, injected after construction like the external IPAM. A nil writer leaves create behaviour unchanged, and a write failure is logged rather than failing the create — the resource exists either way, and failing a create over a tag index would be worse than the divergence it repairs. Only vpc, subnet, security group and network interface are covered here. Image, volume, snapshot, route table, internet gateway, egress-only internet gateway, elastic IP, NAT gateway, key pair and launch template have the same split and are follow-up work. --- spinifex/daemon/daemon.go | 3 + spinifex/handlers/ec2/tags/service_impl.go | 4 + .../handlers/ec2/vpc/central_tags_test.go | 160 ++++++++++++++++++ spinifex/handlers/ec2/vpc/eni.go | 2 + spinifex/handlers/ec2/vpc/security_group.go | 2 + spinifex/handlers/ec2/vpc/service.go | 7 + spinifex/handlers/ec2/vpc/service_impl.go | 29 ++++ 7 files changed, 207 insertions(+) create mode 100644 spinifex/handlers/ec2/vpc/central_tags_test.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 347677413..a5d070350 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1651,6 +1651,9 @@ 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) 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 b4011f64d..6006264ea 100644 --- a/spinifex/handlers/ec2/tags/service_impl.go +++ b/spinifex/handlers/ec2/tags/service_impl.go @@ -19,6 +19,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/kvutil" "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/utils" @@ -31,6 +32,9 @@ 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) + // TagsServiceImpl implements TagsService over a JetStream KV bucket, one entry // per resource, scoped by account in the key. // diff --git a/spinifex/handlers/ec2/vpc/central_tags_test.go b/spinifex/handlers/ec2/vpc/central_tags_test.go new file mode 100644 index 000000000..f61954270 --- /dev/null +++ b/spinifex/handlers/ec2/vpc/central_tags_test.go @@ -0,0 +1,160 @@ +package handlers_ec2_vpc + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/stretchr/testify/assert" + "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 +} + +func (f *fakeCentralTagWriter) PutResourceTags(_ context.Context, _, resourceID string, tags map[string]string) error { + if f.calls == nil { + f.calls = map[string]map[string]string{} + } + f.calls[resourceID] = tags + return f.err +} + +func TestCreateVpc_ProjectsTagsToCentralStore(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + writer := &fakeCentralTagWriter{} + svc.SetCentralTagWriter(writer) + + out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ + CidrBlock: aws.String("10.0.0.0/16"), + TagSpecifications: []*ec2.TagSpecification{ + { + ResourceType: aws.String("vpc"), + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("my-vpc")}, + }, + }, + }, + }, testAccountID) + require.NoError(t, err) + + vpcID := *out.Vpc.VpcId + require.Contains(t, writer.calls, vpcID) + assert.Equal(t, map[string]string{"Name": "my-vpc"}, writer.calls[vpcID]) +} + +func TestCreateSubnet_ProjectsTagsToCentralStore(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + writer := &fakeCentralTagWriter{} + svc.SetCentralTagWriter(writer) + vpcID := createTestVPC(t, svc, "10.0.0.0/16") + + out, err := svc.CreateSubnet(context.Background(), &ec2.CreateSubnetInput{ + VpcId: aws.String(vpcID), + CidrBlock: aws.String("10.0.1.0/24"), + TagSpecifications: []*ec2.TagSpecification{ + { + ResourceType: aws.String("subnet"), + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("my-subnet")}, + }, + }, + }, + }, testAccountID) + require.NoError(t, err) + + subnetID := *out.Subnet.SubnetId + require.Contains(t, writer.calls, subnetID) + assert.Equal(t, map[string]string{"Name": "my-subnet"}, writer.calls[subnetID]) +} + +func TestCreateSecurityGroup_ProjectsTagsToCentralStore(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + writer := &fakeCentralTagWriter{} + svc.SetCentralTagWriter(writer) + vpcID := createTestVPC(t, svc, "10.0.0.0/16") + + out, err := svc.CreateSecurityGroup(context.Background(), &ec2.CreateSecurityGroupInput{ + GroupName: aws.String("web-sg"), + Description: aws.String("web sg"), + VpcId: aws.String(vpcID), + TagSpecifications: []*ec2.TagSpecification{ + { + ResourceType: aws.String("security-group"), + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("my-sg")}, + }, + }, + }, + }, testAccountID) + require.NoError(t, err) + + groupID := *out.GroupId + require.Contains(t, writer.calls, groupID) + assert.Equal(t, map[string]string{"Name": "my-sg"}, writer.calls[groupID]) +} + +func TestCreateVpc_NoTags_SkipsCentralStoreWrite(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + writer := &fakeCentralTagWriter{} + svc.SetCentralTagWriter(writer) + + out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ + CidrBlock: aws.String("10.0.0.0/16"), + }, testAccountID) + require.NoError(t, err) + + assert.NotContains(t, writer.calls, *out.Vpc.VpcId) +} + +func TestCreateVpc_NilCentralTagWriter_CreateSucceeds(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + // No SetCentralTagWriter call: centralTags stays nil. + + out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ + CidrBlock: aws.String("10.0.0.0/16"), + TagSpecifications: []*ec2.TagSpecification{ + { + ResourceType: aws.String("vpc"), + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("my-vpc")}, + }, + }, + }, + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Vpc.Tags, 1) +} + +func TestCreateVpc_CentralTagWriterError_DoesNotFailCreate(t *testing.T) { + t.Parallel() + svc := setupTestVPCService(t) + writer := &fakeCentralTagWriter{err: errors.New("central store unavailable")} + svc.SetCentralTagWriter(writer) + + out, err := svc.CreateVpc(context.Background(), &ec2.CreateVpcInput{ + CidrBlock: aws.String("10.0.0.0/16"), + TagSpecifications: []*ec2.TagSpecification{ + { + ResourceType: aws.String("vpc"), + Tags: []*ec2.Tag{ + {Key: aws.String("Name"), Value: aws.String("my-vpc")}, + }, + }, + }, + }, testAccountID) + require.NoError(t, err) + require.Len(t, out.Vpc.Tags, 1) +} diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index 6484bbd72..0e76d9fbe 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -203,6 +203,8 @@ func (s *VPCServiceImpl) CreateNetworkInterface(ctx context.Context, input *ec2. slog.InfoContext(ctx, "CreateNetworkInterface completed", "eniId", eniId, "subnetId", subnetId, "ip", privateIP, "accountID", accountID) + s.projectRecordTags(ctx, accountID, eniId, record.Tags) + // Send vpc.create-port synchronously so vpcd OVSDB errors surface to the // caller. Fire-and-forget would let CreateNetworkInterface return success // while the LSP joins zero port groups (NATS hiccup or vpcd OVSDB error), diff --git a/spinifex/handlers/ec2/vpc/security_group.go b/spinifex/handlers/ec2/vpc/security_group.go index a37b6b9bb..1f440538b 100644 --- a/spinifex/handlers/ec2/vpc/security_group.go +++ b/spinifex/handlers/ec2/vpc/security_group.go @@ -249,6 +249,8 @@ func (s *VPCServiceImpl) CreateSecurityGroup(ctx context.Context, input *ec2.Cre slog.InfoContext(ctx, "CreateSecurityGroup completed", "groupId", groupId, "groupName", groupName, "vpcId", vpcId, "accountID", accountID) + s.projectRecordTags(ctx, accountID, groupId, record.Tags) + if err := s.requestSGEvent("vpc.create-sg", SGEvent{ GroupId: groupId, VpcId: vpcId, diff --git a/spinifex/handlers/ec2/vpc/service.go b/spinifex/handlers/ec2/vpc/service.go index 24eec8e32..584b4b6f4 100644 --- a/spinifex/handlers/ec2/vpc/service.go +++ b/spinifex/handlers/ec2/vpc/service.go @@ -33,3 +33,10 @@ type VPCService interface { UpdateSecurityGroupRuleDescriptionsIngress(ctx context.Context, input *ec2.UpdateSecurityGroupRuleDescriptionsIngressInput, accountID string) (*ec2.UpdateSecurityGroupRuleDescriptionsIngressOutput, error) 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 +// handlers/ec2/tags.TagsServiceImpl. +type CentralTagWriter interface { + PutResourceTags(ctx context.Context, accountID, resourceID string, tags map[string]string) error +} diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 1d0256134..9abeae168 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -91,6 +91,11 @@ type VPCServiceImpl struct { externalIPAM *ExternalIPAM 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 + // disableDefaultPublicIP seeds default subnets with MapPublicIpOnLaunch=false // (non-pool external modes have no public IPs to assign). Zero value keeps // the AWS-faithful default of true. @@ -110,6 +115,12 @@ 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 +} + // localAZ returns the node's local availability zone, sourced from // s.config.AZ. Returns "" when no config is wired (mostly test paths), // which the reconciler treats as a legacy record matching every AZ. @@ -281,6 +292,8 @@ func (s *VPCServiceImpl) CreateVpc(ctx context.Context, input *ec2.CreateVpcInpu slog.InfoContext(ctx, "CreateVpc completed", "vpcId", vpcID, "cidrBlock", record.CidrBlock, "vni", vni, "accountID", accountID) + s.projectRecordTags(ctx, accountID, vpcID, record.Tags) + // Publish vpc.create event for vpcd topology translation s.publishVPCEvent("vpc.create", record.VpcId, record.CidrBlock, record.VNI) @@ -752,6 +765,8 @@ func (s *VPCServiceImpl) CreateSubnet(ctx context.Context, input *ec2.CreateSubn slog.InfoContext(ctx, "CreateSubnet completed", "subnetId", subnetID, "vpcId", vpcID, "cidrBlock", record.CidrBlock, "accountID", accountID) + s.projectRecordTags(ctx, accountID, subnetID, record.Tags) + // Publish vpc.create-subnet event for vpcd topology translation s.publishSubnetEvent("vpc.create-subnet", record.SubnetId, record.VpcId, record.CidrBlock) @@ -1056,6 +1071,20 @@ func (s *VPCServiceImpl) RemoveRecordTags(input *ec2.DeleteTagsInput, accountID return nil } +// projectRecordTags writes a create path's tags into the central tag store, the +// create-direction counterpart to updateRecordTags, so DescribeTags agrees with +// the resource's own describe from birth. A write failure is logged and +// swallowed: failing the create over a tag index would be worse than the +// divergence it repairs. +func (s *VPCServiceImpl) projectRecordTags(ctx context.Context, accountID, resourceID string, tags map[string]string) { + if s.centralTags == nil || len(tags) == 0 { + return + } + if err := s.centralTags.PutResourceTags(ctx, accountID, resourceID, tags); err != nil { + slog.ErrorContext(ctx, "central tag store write 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 {