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

refactor: route the remaining fire-and-forget vpc.* producers through projection

Browse Source 
CreateVPC/DeleteVPC, CreateSubnet/DeleteSubnet, AttachInternetGateway/
DetachInternetGateway, AddNATGateway/RemoveNATGateway, AddIGWRoute/
RemoveIGWRoute, GateSubnetEgress/UngateSubnetEgress and RemoveSystemEgress
were still publishing through natsmsg.PublishEvent or a raw nc.Publish
call with a bare subject string, left over from the mechanical slice 3
pass. Each now goes through its named projection.Client method instead,
so every vpc.* producer in the tree uses the typed client.
This commit is contained in:
joshsiv-mulga
2026-10-09 19:13:52 +10:00
parent 3aa82953e4
commit 46f4489631
6 changed files with 44 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
spinifex/domains/ec2/igw/service_impl.go
+6 -21
View File
@@ -15,6 +15,7 @@ import (
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -412,18 +413,10 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
}
// Publish event for vpcd to clean up OVN external switch + gateway + NAT
if s.natsConn != nil {
event := networkv1.InternetGatewayEvent{
InternetGatewayId: igwID,
VpcId: vpcID,
}
eventData, err := json.Marshal(event)
if err != nil {
slog.WarnContext(ctx, "Failed to marshal IGW detach event", "error", err)
} else if err := s.natsConn.Publish(networkv1.InternetGatewayDetachSubject, eventData); err != nil {
slog.WarnContext(ctx, "Failed to publish IGW detach event", "error", err)
}
}
projection.New(s.natsConn).DetachInternetGateway(networkv1.InternetGatewayEvent{
InternetGatewayId: igwID,
VpcId: vpcID,
})
// After detach the VPC's LR external gateway and router-wide default
// route are removed; any subnet whose effective RT still points at the
@@ -440,15 +433,7 @@ func (s *IGWServiceImpl) DetachInternetGateway(ctx context.Context, input *ec2.D
// publishAttach asks vpcd to create the OVN external switch, gateway and SNAT.
func (s *IGWServiceImpl) publishAttach(ctx context.Context, igwID, vpcID string) {
if s.natsConn == nil {
return
}
eventData, err := json.Marshal(networkv1.InternetGatewayEvent{InternetGatewayId: igwID, VpcId: vpcID})
if err != nil {
slog.WarnContext(ctx, "Failed to marshal IGW attach event", "error", err)
} else if err := s.natsConn.Publish(networkv1.InternetGatewayAttachSubject, eventData); err != nil {
slog.WarnContext(ctx, "Failed to publish IGW attach event", "error", err)
}
projection.New(s.natsConn).AttachInternetGateway(networkv1.InternetGatewayEvent{InternetGatewayId: igwID, VpcId: vpcID})
}
// CreateAttachedInternetGateway creates igwID already attached to vpcID, for a
spinifex/domains/ec2/natgw/service_impl.go
+3 -3
View File
@@ -6,7 +6,6 @@ import (
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"log/slog"
"strings"
"time"
@@ -16,6 +15,7 @@ import (
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
ec2eip "github.com/mulgadc/spinifex/spinifex/domains/ec2/eip"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -473,7 +473,7 @@ func natgwMatchesFilters(record *NatGatewayRecord, filters map[string][]string)
// PublishAddEvent publishes a vpc.add-nat-gateway event for vpcd to create the SNAT rule.
// Called by the route table service when CreateRoute targets a NAT GW.
func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
natsmsg.PublishEvent(s.natsConn, networkv1.NATGatewayAddSubject, networkv1.NATGatewayEvent{
projection.New(s.natsConn).AddNATGateway(networkv1.NATGatewayEvent{
VpcId: vpcId,
NatGatewayId: natGatewayId,
PublicIp: publicIp,
@@ -483,7 +483,7 @@ func (s *NatGatewayServiceImpl) PublishAddEvent(vpcId, natGatewayId, publicIp, s
// PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule.
func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) {
natsmsg.PublishEvent(s.natsConn, networkv1.NATGatewayDeleteSubject, networkv1.NATGatewayEvent{
projection.New(s.natsConn).RemoveNATGateway(networkv1.NATGatewayEvent{
VpcId: vpcId,
NatGatewayId: natGatewayId,
PublicIp: publicIp,
spinifex/domains/ec2/routetable/service_impl.go
+16 -27
View File
@@ -19,6 +19,7 @@ import (
ec2igw "github.com/mulgadc/spinifex/spinifex/domains/ec2/igw"
ec2natgw "github.com/mulgadc/spinifex/spinifex/domains/ec2/natgw"
ec2vpc "github.com/mulgadc/spinifex/spinifex/domains/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -332,26 +333,18 @@ func (s *RouteTableServiceImpl) publishSubnetEgressGateDecision(ctx context.Cont
"vpcId", vpcID, "subnetId", subnetID, "err", err)
return
}
client := projection.New(s.natsConn)
topic := networkv1.SubnetEgressGateSubject
var data []byte
if hasEgress {
topic = networkv1.SubnetEgressUngateSubject
data, err = json.Marshal(networkv1.SubnetEgressUngateEvent{
client.UngateSubnetEgress(networkv1.SubnetEgressUngateEvent{
VpcId: vpcID, SubnetId: subnetID, DestinationCidr: destCidr,
})
} else {
data, err = json.Marshal(networkv1.SubnetEgressGateEvent{
client.GateSubnetEgress(networkv1.SubnetEgressGateEvent{
VpcId: vpcID, SubnetId: subnetID, DestinationCidr: destCidr,
})
}
if err != nil {
slog.WarnContext(ctx, "subnet egress gate: marshal failed", "topic", topic, "err", err)
return
}
if err := s.natsConn.Publish(topic, data); err != nil {
slog.WarnContext(ctx, "subnet egress gate: publish failed", "topic", topic, "subnetId", subnetID, "err", err)
return
}
slog.InfoContext(ctx, "subnet egress gate decision published",
"topic", topic, "vpcId", vpcID, "subnetId", subnetID, "destinationCidr", destCidr)
}
@@ -1464,21 +1457,19 @@ func (s *RouteTableServiceImpl) publishNatGatewayEventForSubnet(ctx context.Cont
slog.WarnContext(ctx, "NAT GW event: subnet unmarshal failed", "topic", topic, "subnetId", subnetID, "err", err)
return
}
data, err := json.Marshal(networkv1.NATGatewayEvent{
evt := networkv1.NATGatewayEvent{
VpcId: vpcID,
NatGatewayId: natgwID,
PublicIp: publicIp,
SubnetCidr: subnet.CidrBlock,
SubnetId: subnetID,
DestinationCidr: destCidr,
})
if err != nil {
slog.WarnContext(ctx, "NAT GW event: marshal failed", "topic", topic, "err", err)
return
}
if err := s.natsConn.Publish(topic, data); err != nil {
slog.WarnContext(ctx, "NAT GW event: publish failed", "topic", topic, "subnetId", subnetID, "err", err)
return
client := projection.New(s.natsConn)
if topic == networkv1.NATGatewayDeleteSubject {
client.RemoveNATGateway(evt)
} else {
client.AddNATGateway(evt)
}
slog.InfoContext(ctx, "NAT GW event published", "topic", topic, "subnetCidr", subnet.CidrBlock, "publicIp", publicIp, "subnetId", subnetID, "destinationCidr", destCidr)
}
@@ -1533,19 +1524,17 @@ func (s *RouteTableServiceImpl) publishIGWRouteEventForSubnet(ctx context.Contex
if s.natsConn == nil {
return
}
data, err := json.Marshal(networkv1.IGWRouteEvent{
evt := networkv1.IGWRouteEvent{
VpcId: vpcID,
SubnetId: subnetID,
DestinationCidr: destCidr,
InternetGatewayId: igwID,
})
if err != nil {
slog.WarnContext(ctx, "IGW route event: marshal failed", "topic", topic, "err", err)
return
}
if err := s.natsConn.Publish(topic, data); err != nil {
slog.WarnContext(ctx, "IGW route event: publish failed", "topic", topic, "subnetId", subnetID, "err", err)
return
client := projection.New(s.natsConn)
if topic == networkv1.IGWRouteDeleteSubject {
client.RemoveIGWRoute(evt)
} else {
client.AddIGWRoute(evt)
}
slog.InfoContext(ctx, "IGW route event published", "topic", topic, "subnetId", subnetID, "destinationCidr", destCidr, "igwId", igwID)
}
spinifex/domains/ec2/vpc/service_impl.go
+15 -3
View File
@@ -6,7 +6,6 @@ import (
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"log/slog"
"net"
"strconv"
@@ -17,6 +16,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
@@ -1661,10 +1661,22 @@ func (s *VPCServiceImpl) getSubnet(ctx context.Context, accountID, subnetId stri
// publishVPCEvent publishes a VPC lifecycle event to NATS for vpcd consumption.
// This is fire-and-forget; errors are logged but do not fail the API response.
func (s *VPCServiceImpl) publishVPCEvent(topic, vpcId, cidrBlock string, vni int64) {
natsmsg.PublishEvent(s.natsConn, topic, networkv1.VPCEvent{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni})
evt := networkv1.VPCEvent{VpcId: vpcId, CidrBlock: cidrBlock, VNI: vni}
client := projection.New(s.natsConn)
if topic == networkv1.VPCDeleteSubject {
client.DeleteVPC(evt)
return
}
client.CreateVPC(evt)
}
// publishSubnetEvent publishes a subnet lifecycle event to NATS for vpcd consumption.
func (s *VPCServiceImpl) publishSubnetEvent(topic, subnetId, vpcId, cidrBlock string) {
natsmsg.PublishEvent(s.natsConn, topic, networkv1.SubnetEvent{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock})
evt := networkv1.SubnetEvent{SubnetId: subnetId, VpcId: vpcId, CidrBlock: cidrBlock}
client := projection.New(s.natsConn)
if topic == networkv1.SubnetDeleteSubject {
client.DeleteSubnet(evt)
return
}
client.CreateSubnet(evt)
}
spinifex/domains/network/systemvpc/systemvpc.go
+2 -2
View File
@@ -16,7 +16,6 @@ import (
"context"
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"hash/fnv"
"log/slog"
"net/netip"
@@ -25,6 +24,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/tags"
"github.com/nats-io/nats.go"
@@ -678,6 +678,6 @@ func gcTopology(ctx context.Context, nc *nats.Conn, name, vpcID string) {
if nc == nil || vpcID == "" {
return
}
natsmsg.PublishEvent(nc, networkv1.VPCDeleteSubject, networkv1.VPCEvent{VpcId: vpcID})
projection.New(nc).DeleteVPC(networkv1.VPCEvent{VpcId: vpcID})
slog.InfoContext(ctx, "systemvpc Delete: republished vpc.delete for OVN GC", "name", name, "vpc", vpcID)
}
spinifex/handlers/eks/service_impl.go
+2 -2
View File
@@ -8,7 +8,6 @@ import (
eksv1 "github.com/mulgadc/spinifex/contracts/eks/v1"
networkv1 "github.com/mulgadc/spinifex/contracts/network/v1"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"log/slog"
"maps"
"net"
@@ -29,6 +28,7 @@ import (
"github.com/mulgadc/spinifex/spinifex/domains/ec2/systeminstance"
"github.com/mulgadc/spinifex/spinifex/domains/eks/access"
"github.com/mulgadc/spinifex/spinifex/domains/eks/addon"
"github.com/mulgadc/spinifex/spinifex/domains/network/projection"
resourcearn "github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/lifecycle/idempotency"
@@ -1318,7 +1318,7 @@ func (s *EKSServiceImpl) purgeClusterInfra(ctx context.Context, accountID, name
if len(meta.ResourcesVpcConfig.SubnetIds) > 0 {
subnetID = meta.ResourcesVpcConfig.SubnetIds[0]
}
natsmsg.PublishEvent(s.deps.NATSConn, networkv1.SystemEgressDeleteSubject, networkv1.SystemEgressEvent{
projection.New(s.deps.NATSConn).RemoveSystemEgress(networkv1.SystemEgressEvent{
VpcId: meta.ResourcesVpcConfig.VpcId,
SubnetId: subnetID,
InstanceIp: meta.ControlPlaneENIIP,