feat: Phase 2 — public subnets + auto-assign public IPs

Add MapPublicIpOnLaunch field to SubnetRecord with ModifySubnetAttribute
API handler. Add PublicIpAddress/PublicIpPool fields to ENIRecord for
tracking auto-assigned public IPs.

Add vpc.add-nat / vpc.delete-nat NATS topics and handlers in vpcd
topology for creating distributed dnat_and_snat NAT rules (with
LogicalPort + ExternalMAC for chassis-local NAT processing).

Wire ModifySubnetAttribute through gateway → daemon → VPC service.
This commit is contained in:
Ben Duncan
2026-03-20 04:19:28 +00:00
parent c1cdfc07c3
commit 150bf246a4
10 changed files with 239 additions and 12 deletions
+1
View File
@@ -392,6 +392,7 @@ func (d *Daemon) subscribeAll() error {
{"ec2.CreateSubnet", d.handleEC2CreateSubnet, "spinifex-workers"},
{"ec2.DeleteSubnet", d.handleEC2DeleteSubnet, "spinifex-workers"},
{"ec2.DescribeSubnets", d.handleEC2DescribeSubnets, "spinifex-workers"},
{"ec2.ModifySubnetAttribute", d.handleEC2ModifySubnetAttribute, "spinifex-workers"},
{"ec2.CreateNetworkInterface", d.handleEC2CreateNetworkInterface, "spinifex-workers"},
{"ec2.DeleteNetworkInterface", d.handleEC2DeleteNetworkInterface, "spinifex-workers"},
{"ec2.DescribeNetworkInterfaces", d.handleEC2DescribeNetworkInterfaces, "spinifex-workers"},
+4
View File
@@ -31,6 +31,10 @@ func (d *Daemon) handleEC2DescribeSubnets(msg *nats.Msg) {
handleNATSRequest(msg, d.vpcService.DescribeSubnets)
}
func (d *Daemon) handleEC2ModifySubnetAttribute(msg *nats.Msg) {
handleNATSRequest(msg, d.vpcService.ModifySubnetAttribute)
}
func (d *Daemon) handleEC2CreateNetworkInterface(msg *nats.Msg) {
handleNATSRequest(msg, d.vpcService.CreateNetworkInterface)
}
+3
View File
@@ -215,6 +215,9 @@ var ec2Actions = map[string]EC2Handler{
"DescribeSubnets": ec2Handler(func(input *ec2.DescribeSubnetsInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_vpc.DescribeSubnets(input, gw.NATSConn, accountID)
}),
"ModifySubnetAttribute": ec2Handler(func(input *ec2.ModifySubnetAttributeInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_vpc.ModifySubnetAttribute(input, gw.NATSConn, accountID)
}),
"CreateNetworkInterface": ec2Handler(func(input *ec2.CreateNetworkInterfaceInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_vpc.CreateNetworkInterface(input, gw.NATSConn, accountID)
}),
@@ -0,0 +1,30 @@
package gateway_ec2_vpc
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/nats-io/nats.go"
)
// ModifySubnetAttribute handles the EC2 ModifySubnetAttribute API call
func ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, natsConn *nats.Conn, accountID string) (ec2.ModifySubnetAttributeOutput, error) {
var output ec2.ModifySubnetAttributeOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
if input.SubnetId == nil || *input.SubnetId == "" {
return output, errors.New(awserrors.ErrorMissingParameter)
}
svc := handlers_ec2_vpc.NewNATSVPCService(natsConn)
result, err := svc.ModifySubnetAttribute(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
+1 -1
View File
@@ -1021,7 +1021,7 @@ func TestEC2ActionMapCompleteness(t *testing.T) {
"CreateEgressOnlyInternetGateway", "DeleteEgressOnlyInternetGateway",
"DescribeEgressOnlyInternetGateways",
"CreateVpc", "DeleteVpc", "DescribeVpcs",
"CreateSubnet", "DeleteSubnet", "DescribeSubnets",
"CreateSubnet", "DeleteSubnet", "DescribeSubnets", "ModifySubnetAttribute",
"CreateNetworkInterface", "DeleteNetworkInterface", "DescribeNetworkInterfaces",
}
+8
View File
@@ -33,6 +33,8 @@ type ENIRecord struct {
AttachmentId string `json:"attachment_id,omitempty"`
InstanceId string `json:"instance_id,omitempty"`
DeviceIndex int64 `json:"device_index"`
PublicIpAddress string `json:"public_ip_address,omitempty"` // Auto-assigned or EIP
PublicIpPool string `json:"public_ip_pool,omitempty"` // Pool name the public IP came from
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
}
@@ -346,6 +348,12 @@ func (s *VPCServiceImpl) eniRecordToEC2(record *ENIRecord, accountID string) *ec
Groups: []*ec2.GroupIdentifier{},
}
if record.PublicIpAddress != "" {
eni.Association = &ec2.NetworkInterfaceAssociation{
PublicIp: aws.String(record.PublicIpAddress),
}
}
if record.AttachmentId != "" {
eni.Attachment = &ec2.NetworkInterfaceAttachment{
AttachmentId: aws.String(record.AttachmentId),
+9 -1
View File
@@ -2,7 +2,7 @@ package handlers_ec2_vpc
import "github.com/aws/aws-sdk-go/service/ec2"
// VPCService defines the interface for VPC, Subnet, and ENI operations
// VPCService defines the interface for VPC, Subnet, ENI, and Security Group operations
type VPCService interface {
CreateVpc(input *ec2.CreateVpcInput, accountID string) (*ec2.CreateVpcOutput, error)
DeleteVpc(input *ec2.DeleteVpcInput, accountID string) (*ec2.DeleteVpcOutput, error)
@@ -10,7 +10,15 @@ type VPCService interface {
CreateSubnet(input *ec2.CreateSubnetInput, accountID string) (*ec2.CreateSubnetOutput, error)
DeleteSubnet(input *ec2.DeleteSubnetInput, accountID string) (*ec2.DeleteSubnetOutput, error)
DescribeSubnets(input *ec2.DescribeSubnetsInput, accountID string) (*ec2.DescribeSubnetsOutput, error)
ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error)
CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error)
DeleteNetworkInterface(input *ec2.DeleteNetworkInterfaceInput, accountID string) (*ec2.DeleteNetworkInterfaceOutput, error)
DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error)
CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error)
DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error)
DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error)
AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error)
AuthorizeSecurityGroupEgress(input *ec2.AuthorizeSecurityGroupEgressInput, accountID string) (*ec2.AuthorizeSecurityGroupEgressOutput, error)
RevokeSecurityGroupIngress(input *ec2.RevokeSecurityGroupIngressInput, accountID string) (*ec2.RevokeSecurityGroupIngressOutput, error)
RevokeSecurityGroupEgress(input *ec2.RevokeSecurityGroupEgressInput, accountID string) (*ec2.RevokeSecurityGroupEgressOutput, error)
}
+58 -10
View File
@@ -45,14 +45,15 @@ type VPCRecord struct {
// SubnetRecord represents a stored Subnet
type SubnetRecord struct {
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
AvailabilityZone string `json:"availability_zone"`
State string `json:"state"`
IsDefault bool `json:"is_default"`
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
SubnetId string `json:"subnet_id"`
VpcId string `json:"vpc_id"`
CidrBlock string `json:"cidr_block"`
AvailabilityZone string `json:"availability_zone"`
State string `json:"state"`
IsDefault bool `json:"is_default"`
MapPublicIpOnLaunch bool `json:"map_public_ip_on_launch"`
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
}
// VPCServiceImpl implements VPC, Subnet, and ENI operations with NATS JetStream persistence
@@ -63,6 +64,7 @@ type VPCServiceImpl struct {
subnetKV nats.KeyValue
vniKV nats.KeyValue
eniKV nats.KeyValue
sgKV nats.KeyValue
ipam *IPAM
}
@@ -105,6 +107,14 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
return nil, fmt.Errorf("write version to %s: %w", KVBucketENIs, err)
}
sgKV, err := utils.GetOrCreateKVBucket(js, KVBucketSecurityGroups, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err)
}
if err := utils.WriteVersion(sgKV, KVBucketSecurityGroupsVersion); err != nil {
return nil, fmt.Errorf("write version to %s: %w", KVBucketSecurityGroups, err)
}
ipam, err := NewIPAM(js)
if err != nil {
return nil, fmt.Errorf("failed to initialize IPAM: %w", err)
@@ -114,7 +124,8 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
"vpcBucket", KVBucketVPCs,
"subnetBucket", KVBucketSubnets,
"vniBucket", KVBucketVNICounter,
"eniBucket", KVBucketENIs)
"eniBucket", KVBucketENIs,
"sgBucket", KVBucketSecurityGroups)
return &VPCServiceImpl{
config: cfg,
@@ -123,6 +134,7 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
subnetKV: subnetKV,
vniKV: vniKV,
eniKV: eniKV,
sgKV: sgKV,
ipam: ipam,
}, nil
}
@@ -632,7 +644,7 @@ func (s *VPCServiceImpl) subnetRecordToEC2(record *SubnetRecord, availableIPs in
DefaultForAz: aws.Bool(record.IsDefault),
AvailableIpAddressCount: aws.Int64(int64(availableIPs)),
OwnerId: aws.String(accountID),
MapPublicIpOnLaunch: aws.Bool(false),
MapPublicIpOnLaunch: aws.Bool(record.MapPublicIpOnLaunch),
}
if len(record.Tags) > 0 {
@@ -646,6 +658,42 @@ func (s *VPCServiceImpl) subnetRecordToEC2(record *SubnetRecord, availableIPs in
return subnet
}
// ModifySubnetAttribute modifies a subnet's attributes (e.g. MapPublicIpOnLaunch).
func (s *VPCServiceImpl) ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error) {
if input.SubnetId == nil || *input.SubnetId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
subnetID := *input.SubnetId
key := utils.AccountKey(accountID, subnetID)
entry, err := s.subnetKV.Get(key)
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound)
}
var record SubnetRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
if input.MapPublicIpOnLaunch != nil && input.MapPublicIpOnLaunch.Value != nil {
record.MapPublicIpOnLaunch = *input.MapPublicIpOnLaunch.Value
}
data, err := json.Marshal(record)
if err != nil {
return nil, fmt.Errorf("failed to marshal subnet record: %w", err)
}
if _, err := s.subnetKV.Update(key, data, entry.Revision()); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("ModifySubnetAttribute completed", "subnetId", subnetID, "mapPublicIpOnLaunch", record.MapPublicIpOnLaunch, "accountID", accountID)
return &ec2.ModifySubnetAttributeOutput{}, nil
}
// Default VPC constants matching AWS defaults.
const (
DefaultVPCCidr = "172.31.0.0/16"
+32
View File
@@ -42,6 +42,10 @@ func (s *NATSVPCService) DescribeSubnets(input *ec2.DescribeSubnetsInput, accoun
return utils.NATSRequest[ec2.DescribeSubnetsOutput](s.natsConn, "ec2.DescribeSubnets", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) ModifySubnetAttribute(input *ec2.ModifySubnetAttributeInput, accountID string) (*ec2.ModifySubnetAttributeOutput, error) {
return utils.NATSRequest[ec2.ModifySubnetAttributeOutput](s.natsConn, "ec2.ModifySubnetAttribute", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) CreateNetworkInterface(input *ec2.CreateNetworkInterfaceInput, accountID string) (*ec2.CreateNetworkInterfaceOutput, error) {
return utils.NATSRequest[ec2.CreateNetworkInterfaceOutput](s.natsConn, "ec2.CreateNetworkInterface", input, 30*time.Second, accountID)
}
@@ -53,3 +57,31 @@ func (s *NATSVPCService) DeleteNetworkInterface(input *ec2.DeleteNetworkInterfac
func (s *NATSVPCService) DescribeNetworkInterfaces(input *ec2.DescribeNetworkInterfacesInput, accountID string) (*ec2.DescribeNetworkInterfacesOutput, error) {
return utils.NATSRequest[ec2.DescribeNetworkInterfacesOutput](s.natsConn, "ec2.DescribeNetworkInterfaces", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) CreateSecurityGroup(input *ec2.CreateSecurityGroupInput, accountID string) (*ec2.CreateSecurityGroupOutput, error) {
return utils.NATSRequest[ec2.CreateSecurityGroupOutput](s.natsConn, "ec2.CreateSecurityGroup", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) DeleteSecurityGroup(input *ec2.DeleteSecurityGroupInput, accountID string) (*ec2.DeleteSecurityGroupOutput, error) {
return utils.NATSRequest[ec2.DeleteSecurityGroupOutput](s.natsConn, "ec2.DeleteSecurityGroup", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) DescribeSecurityGroups(input *ec2.DescribeSecurityGroupsInput, accountID string) (*ec2.DescribeSecurityGroupsOutput, error) {
return utils.NATSRequest[ec2.DescribeSecurityGroupsOutput](s.natsConn, "ec2.DescribeSecurityGroups", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) AuthorizeSecurityGroupIngress(input *ec2.AuthorizeSecurityGroupIngressInput, accountID string) (*ec2.AuthorizeSecurityGroupIngressOutput, error) {
return utils.NATSRequest[ec2.AuthorizeSecurityGroupIngressOutput](s.natsConn, "ec2.AuthorizeSecurityGroupIngress", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) AuthorizeSecurityGroupEgress(input *ec2.AuthorizeSecurityGroupEgressInput, accountID string) (*ec2.AuthorizeSecurityGroupEgressOutput, error) {
return utils.NATSRequest[ec2.AuthorizeSecurityGroupEgressOutput](s.natsConn, "ec2.AuthorizeSecurityGroupEgress", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) RevokeSecurityGroupIngress(input *ec2.RevokeSecurityGroupIngressInput, accountID string) (*ec2.RevokeSecurityGroupIngressOutput, error) {
return utils.NATSRequest[ec2.RevokeSecurityGroupIngressOutput](s.natsConn, "ec2.RevokeSecurityGroupIngress", input, 30*time.Second, accountID)
}
func (s *NATSVPCService) RevokeSecurityGroupEgress(input *ec2.RevokeSecurityGroupEgressInput, accountID string) (*ec2.RevokeSecurityGroupEgressOutput, error) {
return utils.NATSRequest[ec2.RevokeSecurityGroupEgressOutput](s.natsConn, "ec2.RevokeSecurityGroupEgress", input, 30*time.Second, accountID)
}
+93
View File
@@ -23,6 +23,8 @@ const (
TopicPortStatus = "vpc.port-status"
TopicIGWAttach = "vpc.igw-attach"
TopicIGWDetach = "vpc.igw-detach"
TopicAddNAT = "vpc.add-nat"
TopicDeleteNAT = "vpc.delete-nat"
)
// VPCEvent is published on vpc.create after a VPC is persisted.
@@ -48,6 +50,15 @@ type PortEvent struct {
MacAddress string `json:"mac_address"`
}
// NATEvent is published on vpc.add-nat / vpc.delete-nat for 1:1 public IP NAT.
type NATEvent struct {
VpcId string `json:"vpc_id"`
ExternalIP string `json:"external_ip"`
LogicalIP string `json:"logical_ip"`
PortName string `json:"port_name"` // logical port for distributed NAT
MAC string `json:"mac"` // external MAC for distributed NAT
}
// TopologyHandler translates VPC lifecycle NATS events into OVN NB DB operations.
type TopologyHandler struct {
ovn OVNClient
@@ -123,6 +134,11 @@ func (h *TopologyHandler) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error)
{TopicDeletePort, h.handleDeletePort, true},
{TopicIGWAttach, h.handleIGWAttach, true},
{TopicIGWDetach, h.handleIGWDetach, true},
{TopicAddNAT, h.handleAddNAT, true},
{TopicDeleteNAT, h.handleDeleteNAT, true},
{TopicCreateSG, h.handleCreateSG, true},
{TopicDeleteSG, h.handleDeleteSG, true},
{TopicUpdateSG, h.handleUpdateSG, true},
}
var result []*nats.Subscription
@@ -765,6 +781,83 @@ func (h *TopologyHandler) handleIGWDetach(msg *nats.Msg) {
respond(msg, nil)
}
// --- NAT (dnat_and_snat for public IPs) ---
func (h *TopologyHandler) handleAddNAT(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.add-nat event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
routerName := "vpc-" + evt.VpcId
logicalPort := evt.PortName
externalMAC := evt.MAC
natRule := &nbdb.NAT{
Type: "dnat_and_snat",
ExternalIP: evt.ExternalIP,
LogicalIP: evt.LogicalIP,
LogicalPort: &logicalPort,
ExternalMAC: &externalMAC,
ExternalIDs: map[string]string{
"spinifex:vpc_id": evt.VpcId,
"spinifex:public_ip": evt.ExternalIP,
},
}
if err := h.ovn.AddNAT(ctx, routerName, natRule); err != nil {
slog.Error("vpcd: failed to add dnat_and_snat rule", "router", routerName, "externalIP", evt.ExternalIP, "logicalIP", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: added dnat_and_snat rule",
"router", routerName,
"external_ip", evt.ExternalIP,
"logical_ip", evt.LogicalIP,
"port", evt.PortName,
)
respond(msg, nil)
}
func (h *TopologyHandler) handleDeleteNAT(msg *nats.Msg) {
if h.ovn == nil {
respond(msg, fmt.Errorf("OVN client not connected"))
return
}
var evt NATEvent
if err := json.Unmarshal(msg.Data, &evt); err != nil {
slog.Error("vpcd: failed to unmarshal vpc.delete-nat event", "err", err)
respond(msg, err)
return
}
ctx := context.Background()
routerName := "vpc-" + evt.VpcId
if err := h.ovn.DeleteNAT(ctx, routerName, "dnat_and_snat", evt.LogicalIP); err != nil {
slog.Error("vpcd: failed to delete dnat_and_snat rule", "router", routerName, "logicalIP", evt.LogicalIP, "err", err)
respond(msg, err)
return
}
slog.Info("vpcd: deleted dnat_and_snat rule",
"router", routerName,
"logical_ip", evt.LogicalIP,
)
respond(msg, nil)
}
// --- Helpers ---
// subnetGateway computes the gateway IP (.1) from a CIDR string.