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 {