From 4ffc7785a53617f68644490611a5dbf199c75e5a Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Thu, 26 Mar 2026 14:29:14 +1000 Subject: [PATCH] =?UTF-8?q?feat:=20NAT=20Gateway=20=E2=80=94=20CRUD=20+=20?= =?UTF-8?q?vpcd=20SNAT=20rules=20+=20route=20table=20integration=20(mulga-?= =?UTF-8?q?750)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements AWS-compatible NAT Gateways for private subnet outbound internet: API (3 actions): - CreateNatGateway: validates subnet exists, EIP allocated, stores in KV - DeleteNatGateway: removes record, publishes delete events for SNAT cleanup - DescribeNatGateways: filter by ID, vpc-id, state vpcd SNAT rule management: - handleAddNATGateway: creates OVN snat rule (subnet CIDR → NAT GW EIP) - handleDeleteNATGateway: removes OVN snat rule - Subscribes to vpc.add-nat-gateway / vpc.delete-nat-gateway topics Route table integration: - CreateRoute now accepts NatGatewayId target (in addition to GatewayId) - Validates NAT GW exists and belongs to same VPC - Publishes vpc.add-nat-gateway events for associated subnets OVN mechanics: - Single snat rule per private subnet CIDR (shared by all VMs) - No policy-based routing needed — SNAT rewrites before route lookup - Existing 0.0.0.0/0 default route handles all egress - Connection tracking handles return traffic automatically --- spinifex/daemon/daemon.go | 12 + spinifex/daemon/daemon_handlers_natgw.go | 17 + spinifex/gateway/ec2.go | 10 + .../gateway/ec2/natgw/CreateNatGateway.go | 23 + .../gateway/ec2/natgw/DeleteNatGateway.go | 23 + .../gateway/ec2/natgw/DescribeNatGateways.go | 23 + spinifex/gateway/gateway_test.go | 1 + spinifex/handlers/ec2/natgw/service.go | 10 + spinifex/handlers/ec2/natgw/service_impl.go | 451 ++++++++++++++++++ spinifex/handlers/ec2/natgw/service_nats.go | 31 ++ spinifex/handlers/ec2/natgw/types.go | 21 + .../handlers/ec2/routetable/service_impl.go | 129 ++++- spinifex/services/vpcd/topology.go | 95 +++- 13 files changed, 811 insertions(+), 35 deletions(-) create mode 100644 spinifex/daemon/daemon_handlers_natgw.go create mode 100644 spinifex/gateway/ec2/natgw/CreateNatGateway.go create mode 100644 spinifex/gateway/ec2/natgw/DeleteNatGateway.go create mode 100644 spinifex/gateway/ec2/natgw/DescribeNatGateways.go create mode 100644 spinifex/handlers/ec2/natgw/service.go create mode 100644 spinifex/handlers/ec2/natgw/service_impl.go create mode 100644 spinifex/handlers/ec2/natgw/service_nats.go create mode 100644 spinifex/handlers/ec2/natgw/types.go diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 0240ebc4f..7c16326d2 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -38,6 +38,7 @@ import ( handlers_ec2_image "github.com/mulgadc/spinifex/spinifex/handlers/ec2/image" handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance" handlers_ec2_key "github.com/mulgadc/spinifex/spinifex/handlers/ec2/key" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable" handlers_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/handlers/ec2/snapshot" @@ -113,6 +114,7 @@ type Daemon struct { vpcService *handlers_ec2_vpc.VPCServiceImpl eipService *handlers_ec2_eip.EIPServiceImpl routeTableService *handlers_ec2_routetable.RouteTableServiceImpl + natGatewayService *handlers_ec2_natgw.NatGatewayServiceImpl externalIPAM *handlers_ec2_vpc.ExternalIPAM ctx context.Context cancel context.CancelFunc @@ -404,6 +406,9 @@ func (d *Daemon) subscribeAll() error { {"ec2.RemoveInstanceFromPlacementGroup", d.handleEC2RemoveInstanceFromPlacementGroup, "spinifex-workers"}, {"ec2.ReserveClusterNode", d.handleEC2ReserveClusterNode, "spinifex-workers"}, {"ec2.FinalizeClusterInstances", d.handleEC2FinalizeClusterInstances, "spinifex-workers"}, + {"ec2.CreateNatGateway", d.handleEC2CreateNatGateway, "spinifex-workers"}, + {"ec2.DeleteNatGateway", d.handleEC2DeleteNatGateway, "spinifex-workers"}, + {"ec2.DescribeNatGateways", d.handleEC2DescribeNatGateways, "spinifex-workers"}, {"ec2.CreateRouteTable", d.handleEC2CreateRouteTable, "spinifex-workers"}, {"ec2.DeleteRouteTable", d.handleEC2DeleteRouteTable, "spinifex-workers"}, {"ec2.DescribeRouteTables", d.handleEC2DescribeRouteTables, "spinifex-workers"}, @@ -576,6 +581,13 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to initialize RouteTable service: %w", err) } + d.natGatewayService, err = initServiceWithRetry("NatGateway service", func() (*handlers_ec2_natgw.NatGatewayServiceImpl, error) { + return handlers_ec2_natgw.NewNatGatewayServiceImplWithNATS(d.natsConn) + }) + if err != nil { + return fmt.Errorf("failed to initialize NatGateway service: %w", err) + } + // Initialize external IPAM if pool mode is configured (per-VM public IPs). // NAT mode only uses SNAT via a shared gateway IP — no per-VM allocation needed. if d.clusterConfig != nil && d.clusterConfig.Network.ExternalMode == "pool" { diff --git a/spinifex/daemon/daemon_handlers_natgw.go b/spinifex/daemon/daemon_handlers_natgw.go new file mode 100644 index 000000000..bfbb3b4d2 --- /dev/null +++ b/spinifex/daemon/daemon_handlers_natgw.go @@ -0,0 +1,17 @@ +package daemon + +import ( + "github.com/nats-io/nats.go" +) + +func (d *Daemon) handleEC2CreateNatGateway(msg *nats.Msg) { + handleNATSRequest(msg, d.natGatewayService.CreateNatGateway) +} + +func (d *Daemon) handleEC2DeleteNatGateway(msg *nats.Msg) { + handleNATSRequest(msg, d.natGatewayService.DeleteNatGateway) +} + +func (d *Daemon) handleEC2DescribeNatGateways(msg *nats.Msg) { + handleNATSRequest(msg, d.natGatewayService.DescribeNatGateways) +} diff --git a/spinifex/gateway/ec2.go b/spinifex/gateway/ec2.go index 720368e62..1eca2de2a 100644 --- a/spinifex/gateway/ec2.go +++ b/spinifex/gateway/ec2.go @@ -19,6 +19,7 @@ import ( gateway_ec2_image "github.com/mulgadc/spinifex/spinifex/gateway/ec2/image" gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance" gateway_ec2_key "github.com/mulgadc/spinifex/spinifex/gateway/ec2/key" + gateway_ec2_natgw "github.com/mulgadc/spinifex/spinifex/gateway/ec2/natgw" gateway_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/gateway/ec2/placementgroup" gateway_ec2_routetable "github.com/mulgadc/spinifex/spinifex/gateway/ec2/routetable" gateway_ec2_snapshot "github.com/mulgadc/spinifex/spinifex/gateway/ec2/snapshot" @@ -317,6 +318,15 @@ var ec2Actions = map[string]EC2Handler{ "DescribeAddresses": ec2Handler(func(input *ec2.DescribeAddressesInput, gw *GatewayConfig, accountID string) (any, error) { return gateway_ec2_eip.DescribeAddresses(input, gw.NATSConn, accountID) }), + "CreateNatGateway": ec2Handler(func(input *ec2.CreateNatGatewayInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_natgw.CreateNatGateway(input, gw.NATSConn, accountID) + }), + "DeleteNatGateway": ec2Handler(func(input *ec2.DeleteNatGatewayInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_natgw.DeleteNatGateway(input, gw.NATSConn, accountID) + }), + "DescribeNatGateways": ec2Handler(func(input *ec2.DescribeNatGatewaysInput, gw *GatewayConfig, accountID string) (any, error) { + return gateway_ec2_natgw.DescribeNatGateways(input, gw.NATSConn, accountID) + }), } // ec2LocalActions are actions that don't require a NATS connection. diff --git a/spinifex/gateway/ec2/natgw/CreateNatGateway.go b/spinifex/gateway/ec2/natgw/CreateNatGateway.go new file mode 100644 index 000000000..068103397 --- /dev/null +++ b/spinifex/gateway/ec2/natgw/CreateNatGateway.go @@ -0,0 +1,23 @@ +package gateway_ec2_natgw + +import ( + "errors" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" + "github.com/nats-io/nats.go" +) + +func CreateNatGateway(input *ec2.CreateNatGatewayInput, natsConn *nats.Conn, accountID string) (ec2.CreateNatGatewayOutput, error) { + var output ec2.CreateNatGatewayOutput + if input == nil { + return output, errors.New(awserrors.ErrorInvalidParameterValue) + } + svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn) + result, err := svc.CreateNatGateway(input, accountID) + if err != nil { + return output, err + } + return *result, nil +} diff --git a/spinifex/gateway/ec2/natgw/DeleteNatGateway.go b/spinifex/gateway/ec2/natgw/DeleteNatGateway.go new file mode 100644 index 000000000..19d5322fb --- /dev/null +++ b/spinifex/gateway/ec2/natgw/DeleteNatGateway.go @@ -0,0 +1,23 @@ +package gateway_ec2_natgw + +import ( + "errors" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" + "github.com/nats-io/nats.go" +) + +func DeleteNatGateway(input *ec2.DeleteNatGatewayInput, natsConn *nats.Conn, accountID string) (ec2.DeleteNatGatewayOutput, error) { + var output ec2.DeleteNatGatewayOutput + if input == nil { + return output, errors.New(awserrors.ErrorInvalidParameterValue) + } + svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn) + result, err := svc.DeleteNatGateway(input, accountID) + if err != nil { + return output, err + } + return *result, nil +} diff --git a/spinifex/gateway/ec2/natgw/DescribeNatGateways.go b/spinifex/gateway/ec2/natgw/DescribeNatGateways.go new file mode 100644 index 000000000..8e73ab49b --- /dev/null +++ b/spinifex/gateway/ec2/natgw/DescribeNatGateways.go @@ -0,0 +1,23 @@ +package gateway_ec2_natgw + +import ( + "errors" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" + "github.com/nats-io/nats.go" +) + +func DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, natsConn *nats.Conn, accountID string) (ec2.DescribeNatGatewaysOutput, error) { + var output ec2.DescribeNatGatewaysOutput + if input == nil { + return output, errors.New(awserrors.ErrorInvalidParameterValue) + } + svc := handlers_ec2_natgw.NewNATSNatGatewayService(natsConn) + result, err := svc.DescribeNatGateways(input, accountID) + if err != nil { + return output, err + } + return *result, nil +} diff --git a/spinifex/gateway/gateway_test.go b/spinifex/gateway/gateway_test.go index 0b45ada58..18e3c9bd0 100644 --- a/spinifex/gateway/gateway_test.go +++ b/spinifex/gateway/gateway_test.go @@ -1032,6 +1032,7 @@ func TestEC2ActionMapCompleteness(t *testing.T) { "CreateRouteTable", "DeleteRouteTable", "DescribeRouteTables", "CreateRoute", "DeleteRoute", "ReplaceRoute", "AssociateRouteTable", "DisassociateRouteTable", "ReplaceRouteTableAssociation", + "CreateNatGateway", "DeleteNatGateway", "DescribeNatGateways", } for _, action := range expectedActions { diff --git a/spinifex/handlers/ec2/natgw/service.go b/spinifex/handlers/ec2/natgw/service.go new file mode 100644 index 000000000..8a6aa6523 --- /dev/null +++ b/spinifex/handlers/ec2/natgw/service.go @@ -0,0 +1,10 @@ +package handlers_ec2_natgw + +import "github.com/aws/aws-sdk-go/service/ec2" + +// NatGatewayService defines the interface for NAT Gateway operations +type NatGatewayService interface { + CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error) + DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error) + DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error) +} diff --git a/spinifex/handlers/ec2/natgw/service_impl.go b/spinifex/handlers/ec2/natgw/service_impl.go new file mode 100644 index 000000000..c601a0a33 --- /dev/null +++ b/spinifex/handlers/ec2/natgw/service_impl.go @@ -0,0 +1,451 @@ +package handlers_ec2_natgw + +import ( + "encoding/json" + "errors" + "fmt" + "log/slog" + "slices" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/awserrors" + handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +var _ NatGatewayService = (*NatGatewayServiceImpl)(nil) + +// NatGatewayServiceImpl implements NAT Gateway operations with NATS JetStream persistence +type NatGatewayServiceImpl struct { + natgwKV nats.KeyValue + eipKV nats.KeyValue + subnetKV nats.KeyValue + vpcKV nats.KeyValue + natsConn *nats.Conn +} + +// NewNatGatewayServiceImplWithNATS creates a NAT Gateway service +func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceImpl, error) { + js, err := natsConn.JetStream() + if err != nil { + return nil, fmt.Errorf("failed to get JetStream context: %w", err) + } + + natgwKV, err := utils.GetOrCreateKVBucket(js, KVBucketNatGateways, 10) + if err != nil { + return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err) + } + if err := utils.WriteVersion(natgwKV, KVBucketNatGatewaysVersion); err != nil { + return nil, fmt.Errorf("write version to %s: %w", KVBucketNatGateways, err) + } + + eipKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_eip.KVBucketEIPs, 10) + if err != nil { + return nil, fmt.Errorf("failed to get EIP KV bucket: %w", err) + } + + subnetKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketSubnets, 10) + if err != nil { + return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err) + } + + vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10) + if err != nil { + return nil, fmt.Errorf("failed to get VPC KV bucket: %w", err) + } + + slog.Info("NatGateway service initialized with JetStream KV", "bucket", KVBucketNatGateways) + + return &NatGatewayServiceImpl{ + natgwKV: natgwKV, + eipKV: eipKV, + subnetKV: subnetKV, + vpcKV: vpcKV, + natsConn: natsConn, + }, nil +} + +// natGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway topics. +type natGatewayEvent struct { + VpcId string `json:"vpc_id"` + NatGatewayId string `json:"nat_gateway_id"` + PublicIp string `json:"public_ip"` + SubnetCidr string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule +} + +// CreateNatGateway creates a NAT Gateway in a public subnet with an EIP +func (s *NatGatewayServiceImpl) CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error) { + if input.SubnetId == nil || *input.SubnetId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + if input.AllocationId == nil || *input.AllocationId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + subnetID := *input.SubnetId + allocID := *input.AllocationId + + // Validate subnet exists and get its VPC + subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, subnetID)) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidSubnetIDNotFound) + } + var subnetRecord handlers_ec2_vpc.SubnetRecord + if err := json.Unmarshal(subnetEntry.Value(), &subnetRecord); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Validate EIP exists and is not already associated + eipEntry, err := s.eipKV.Get(utils.AccountKey(accountID, allocID)) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidAllocationIDNotFound) + } + var eipRecord handlers_ec2_eip.EIPRecord + if err := json.Unmarshal(eipEntry.Value(), &eipRecord); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if eipRecord.AssociationId != "" { + return nil, errors.New(awserrors.ErrorResourceAlreadyAssociated) + } + + natgwID := utils.GenerateResourceID("nat") + record := NatGatewayRecord{ + NatGatewayId: natgwID, + VpcId: subnetRecord.VpcId, + SubnetId: subnetID, + AllocationId: allocID, + PublicIp: eipRecord.PublicIp, + State: "available", + AccountID: accountID, + Tags: make(map[string]string), + CreatedAt: time.Now(), + } + + data, err := json.Marshal(record) + if err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if _, err := s.natgwKV.Put(utils.AccountKey(accountID, natgwID), data); err != nil { + slog.Error("Failed to store NAT Gateway", "natGatewayId", natgwID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("CreateNatGateway completed", "natGatewayId", natgwID, "subnetId", subnetID, + "allocationId", allocID, "publicIp", eipRecord.PublicIp, "vpcId", subnetRecord.VpcId, "accountID", accountID) + + return &ec2.CreateNatGatewayOutput{ + NatGateway: recordToEC2(&record), + }, nil +} + +// DeleteNatGateway deletes a NAT Gateway and removes OVN SNAT rules +func (s *NatGatewayServiceImpl) DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error) { + if input.NatGatewayId == nil || *input.NatGatewayId == "" { + return nil, errors.New(awserrors.ErrorMissingParameter) + } + + natgwID := *input.NatGatewayId + key := utils.AccountKey(accountID, natgwID) + + entry, err := s.natgwKV.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound) + } + slog.Error("Failed to read NAT Gateway", "natGatewayId", natgwID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + var record NatGatewayRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + // Find all private subnets whose route tables have 0.0.0.0/0 → this NAT GW + // and publish delete events for their SNAT rules + s.publishDeleteEventsForNatGateway(&record, accountID) + + // Delete from KV + if err := s.natgwKV.Delete(key); err != nil { + slog.Error("Failed to delete NAT Gateway", "natGatewayId", natgwID, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + slog.Info("DeleteNatGateway completed", "natGatewayId", natgwID, "accountID", accountID) + + return &ec2.DeleteNatGatewayOutput{ + NatGatewayId: aws.String(natgwID), + }, nil +} + +// publishDeleteEventsForNatGateway finds all SNAT rules associated with this NAT GW +// by scanning route tables for routes targeting it, and publishes delete events. +func (s *NatGatewayServiceImpl) publishDeleteEventsForNatGateway(record *NatGatewayRecord, accountID string) { + if s.natsConn == nil { + return + } + + // Scan route tables in this VPC for routes targeting this NAT GW + rtbKV, err := utils.GetOrCreateKVBucket(mustJS(s.natsConn), "spinifex-vpc-route-tables", 10) + if err != nil { + slog.Warn("Cannot scan route tables for NAT GW cleanup", "err", err) + return + } + + keys, err := rtbKV.Keys() + if err != nil { + return + } + + prefix := accountID + "." + for _, key := range keys { + if key == utils.VersionKey || !strings.HasPrefix(key, prefix) { + continue + } + entry, err := rtbKV.Get(key) + if err != nil { + continue + } + + // Minimal route table parse — just need routes and associations + var rtbData struct { + VpcId string `json:"vpc_id"` + Routes []struct { + NatGatewayId string `json:"nat_gateway_id"` + } `json:"routes"` + Associations []struct { + SubnetId string `json:"subnet_id"` + } `json:"associations"` + } + if err := json.Unmarshal(entry.Value(), &rtbData); err != nil { + continue + } + if rtbData.VpcId != record.VpcId { + continue + } + + // Check if any route targets this NAT GW + hasNatRoute := false + for _, r := range rtbData.Routes { + if r.NatGatewayId == record.NatGatewayId { + hasNatRoute = true + break + } + } + if !hasNatRoute { + continue + } + + // Publish delete events for each associated subnet's CIDR + for _, assoc := range rtbData.Associations { + if assoc.SubnetId == "" { + continue + } + subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, assoc.SubnetId)) + if err != nil { + continue + } + var subnet handlers_ec2_vpc.SubnetRecord + if err := json.Unmarshal(subnetEntry.Value(), &subnet); err != nil { + continue + } + + evt := natGatewayEvent{ + VpcId: record.VpcId, + NatGatewayId: record.NatGatewayId, + PublicIp: record.PublicIp, + SubnetCidr: subnet.CidrBlock, + } + eventData, err := json.Marshal(evt) + if err != nil { + slog.Warn("Failed to marshal NAT GW delete event", "err", err) + continue + } + if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil { + slog.Warn("Failed to publish NAT GW delete event", "subnet", subnet.CidrBlock, "err", err) + } + } + } +} + +func mustJS(nc *nats.Conn) nats.JetStreamContext { + js, _ := nc.JetStream() + return js +} + +// DescribeNatGateways lists NAT Gateways, optionally filtered +func (s *NatGatewayServiceImpl) DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error) { + natgwIDs := make(map[string]bool) + for _, id := range input.NatGatewayIds { + if id != nil { + natgwIDs[*id] = true + } + } + + // Build filter map + filters := make(map[string][]string) + for _, f := range input.Filter { + if f.Name != nil { + for _, v := range f.Values { + if v != nil { + filters[*f.Name] = append(filters[*f.Name], *v) + } + } + } + } + + prefix := accountID + "." + keys, err := s.natgwKV.Keys() + if err != nil && !errors.Is(err, nats.ErrNoKeysFound) { + return nil, errors.New(awserrors.ErrorServerInternal) + } + + var natGateways []*ec2.NatGateway + foundIDs := make(map[string]bool) + + for _, key := range keys { + if key == utils.VersionKey || !strings.HasPrefix(key, prefix) { + continue + } + + entry, err := s.natgwKV.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + continue + } + slog.Error("Failed to read NAT Gateway", "key", key, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + var record NatGatewayRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + slog.Error("Corrupt NAT Gateway record", "key", key, "err", err) + return nil, errors.New(awserrors.ErrorServerInternal) + } + + if len(natgwIDs) > 0 && !natgwIDs[record.NatGatewayId] { + continue + } + + // Apply filters + if vpcFilter, ok := filters["vpc-id"]; ok { + found := slices.Contains(vpcFilter, record.VpcId) + if !found { + continue + } + } + if stateFilter, ok := filters["state"]; ok { + found := slices.Contains(stateFilter, record.State) + if !found { + continue + } + } + + natGateways = append(natGateways, recordToEC2(&record)) + foundIDs[record.NatGatewayId] = true + } + + for id := range natgwIDs { + if !foundIDs[id] { + return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound) + } + } + + return &ec2.DescribeNatGatewaysOutput{ + NatGateways: natGateways, + }, nil +} + +// 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) { + if s.natsConn == nil { + return + } + evt := natGatewayEvent{ + VpcId: vpcId, + NatGatewayId: natGatewayId, + PublicIp: publicIp, + SubnetCidr: subnetCidr, + } + eventData, err := json.Marshal(evt) + if err != nil { + slog.Error("Failed to marshal NAT GW add event", "err", err) + return + } + if err := s.natsConn.Publish("vpc.add-nat-gateway", eventData); err != nil { + slog.Warn("Failed to publish NAT GW add event", "err", err) + } +} + +// PublishDeleteEvent publishes a vpc.delete-nat-gateway event for vpcd to remove the SNAT rule. +func (s *NatGatewayServiceImpl) PublishDeleteEvent(vpcId, natGatewayId, publicIp, subnetCidr string) { + if s.natsConn == nil { + return + } + evt := natGatewayEvent{ + VpcId: vpcId, + NatGatewayId: natGatewayId, + PublicIp: publicIp, + SubnetCidr: subnetCidr, + } + eventData, err := json.Marshal(evt) + if err != nil { + slog.Error("Failed to marshal NAT GW delete event", "err", err) + return + } + if err := s.natsConn.Publish("vpc.delete-nat-gateway", eventData); err != nil { + slog.Warn("Failed to publish NAT GW delete event", "err", err) + } +} + +// GetNatGateway retrieves a NAT Gateway record by ID +func (s *NatGatewayServiceImpl) GetNatGateway(accountID, natgwID string) (*NatGatewayRecord, error) { + entry, err := s.natgwKV.Get(utils.AccountKey(accountID, natgwID)) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound) + } + return nil, errors.New(awserrors.ErrorServerInternal) + } + var record NatGatewayRecord + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + return &record, nil +} + +func recordToEC2(record *NatGatewayRecord) *ec2.NatGateway { + ngw := &ec2.NatGateway{ + NatGatewayId: aws.String(record.NatGatewayId), + VpcId: aws.String(record.VpcId), + SubnetId: aws.String(record.SubnetId), + State: aws.String(record.State), + ConnectivityType: aws.String("public"), + CreateTime: aws.Time(record.CreatedAt), + } + + if record.PublicIp != "" { + ngw.NatGatewayAddresses = []*ec2.NatGatewayAddress{ + { + AllocationId: aws.String(record.AllocationId), + PublicIp: aws.String(record.PublicIp), + }, + } + } + + if len(record.Tags) > 0 { + tags := make([]*ec2.Tag, 0, len(record.Tags)) + for k, v := range record.Tags { + tags = append(tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)}) + } + ngw.Tags = tags + } + + return ngw +} diff --git a/spinifex/handlers/ec2/natgw/service_nats.go b/spinifex/handlers/ec2/natgw/service_nats.go new file mode 100644 index 000000000..96a4fcd0a --- /dev/null +++ b/spinifex/handlers/ec2/natgw/service_nats.go @@ -0,0 +1,31 @@ +package handlers_ec2_natgw + +import ( + "time" + + "github.com/aws/aws-sdk-go/service/ec2" + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// NATSNatGatewayService handles NAT Gateway operations via NATS messaging +type NATSNatGatewayService struct { + natsConn *nats.Conn +} + +// NewNATSNatGatewayService creates a new NATS-based NAT Gateway service +func NewNATSNatGatewayService(conn *nats.Conn) NatGatewayService { + return &NATSNatGatewayService{natsConn: conn} +} + +func (s *NATSNatGatewayService) CreateNatGateway(input *ec2.CreateNatGatewayInput, accountID string) (*ec2.CreateNatGatewayOutput, error) { + return utils.NATSRequest[ec2.CreateNatGatewayOutput](s.natsConn, "ec2.CreateNatGateway", input, 30*time.Second, accountID) +} + +func (s *NATSNatGatewayService) DeleteNatGateway(input *ec2.DeleteNatGatewayInput, accountID string) (*ec2.DeleteNatGatewayOutput, error) { + return utils.NATSRequest[ec2.DeleteNatGatewayOutput](s.natsConn, "ec2.DeleteNatGateway", input, 30*time.Second, accountID) +} + +func (s *NATSNatGatewayService) DescribeNatGateways(input *ec2.DescribeNatGatewaysInput, accountID string) (*ec2.DescribeNatGatewaysOutput, error) { + return utils.NATSRequest[ec2.DescribeNatGatewaysOutput](s.natsConn, "ec2.DescribeNatGateways", input, 30*time.Second, accountID) +} diff --git a/spinifex/handlers/ec2/natgw/types.go b/spinifex/handlers/ec2/natgw/types.go new file mode 100644 index 000000000..62ee3a0cb --- /dev/null +++ b/spinifex/handlers/ec2/natgw/types.go @@ -0,0 +1,21 @@ +package handlers_ec2_natgw + +import "time" + +const ( + KVBucketNatGateways = "spinifex-vpc-nat-gateways" + KVBucketNatGatewaysVersion = 1 +) + +// NatGatewayRecord represents a stored NAT Gateway +type NatGatewayRecord struct { + NatGatewayId string `json:"nat_gateway_id"` + VpcId string `json:"vpc_id"` + SubnetId string `json:"subnet_id"` // public subnet where NAT GW lives + AllocationId string `json:"allocation_id"` // EIP allocation + PublicIp string `json:"public_ip"` // EIP address + State string `json:"state"` // pending, available, deleting, deleted, failed + AccountID string `json:"account_id"` + Tags map[string]string `json:"tags"` + CreatedAt time.Time `json:"created_at"` +} diff --git a/spinifex/handlers/ec2/routetable/service_impl.go b/spinifex/handlers/ec2/routetable/service_impl.go index f581152f1..40e6949ee 100644 --- a/spinifex/handlers/ec2/routetable/service_impl.go +++ b/spinifex/handlers/ec2/routetable/service_impl.go @@ -14,6 +14,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" + handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -29,6 +30,7 @@ type RouteTableServiceImpl struct { vpcKV nats.KeyValue igwKV nats.KeyValue subnetKV nats.KeyValue + natgwKV nats.KeyValue natsConn *nats.Conn } @@ -62,6 +64,11 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) ( return nil, fmt.Errorf("failed to get subnet KV bucket: %w", err) } + natgwKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_natgw.KVBucketNatGateways, 10) + if err != nil { + return nil, fmt.Errorf("failed to get NAT Gateway KV bucket: %w", err) + } + slog.Info("RouteTable service initialized with JetStream KV", "bucket", KVBucketRouteTables) return &RouteTableServiceImpl{ @@ -70,6 +77,7 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) ( vpcKV: vpcKV, igwKV: igwKV, subnetKV: subnetKV, + natgwKV: natgwKV, natsConn: natsConn, }, nil } @@ -347,39 +355,69 @@ func (s *RouteTableServiceImpl) CreateRoute(input *ec2.CreateRouteInput, account } } - // V1: only GatewayId target supported - if input.GatewayId == nil || *input.GatewayId == "" { + var route RouteRecord + + switch { + case input.GatewayId != nil && *input.GatewayId != "": + igwID := *input.GatewayId + // Verify IGW exists and is attached to the same VPC + igwEntry, err := s.igwKV.Get(utils.AccountKey(accountID, igwID)) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidInternetGatewayIDNotFound) + } + var igwRecord handlers_ec2_igw.IGWRecord + if err := json.Unmarshal(igwEntry.Value(), &igwRecord); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if igwRecord.VpcId != record.VpcId { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + route = RouteRecord{ + DestinationCidrBlock: destCidr, + GatewayId: igwID, + State: "active", + Origin: "CreateRoute", + } + + case input.NatGatewayId != nil && *input.NatGatewayId != "": + natgwID := *input.NatGatewayId + // Verify NAT GW exists and belongs to the same VPC + natgwEntry, err := s.natgwKV.Get(utils.AccountKey(accountID, natgwID)) + if err != nil { + return nil, errors.New(awserrors.ErrorInvalidNatGatewayIDNotFound) + } + var natgwRecord struct { + NatGatewayId string `json:"nat_gateway_id"` + VpcId string `json:"vpc_id"` + PublicIp string `json:"public_ip"` + } + if err := json.Unmarshal(natgwEntry.Value(), &natgwRecord); err != nil { + return nil, errors.New(awserrors.ErrorServerInternal) + } + if natgwRecord.VpcId != record.VpcId { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + route = RouteRecord{ + DestinationCidrBlock: destCidr, + NatGatewayId: natgwID, + State: "active", + Origin: "CreateRoute", + } + + // Publish vpc.add-nat-gateway events for each subnet associated with this route table + s.publishNatGatewayEvents(accountID, record, natgwRecord.VpcId, natgwID, natgwRecord.PublicIp) + + default: return nil, errors.New(awserrors.ErrorInvalidParameterValue) } - igwID := *input.GatewayId - - // Verify IGW exists and is attached to the same VPC - igwEntry, err := s.igwKV.Get(utils.AccountKey(accountID, igwID)) - if err != nil { - return nil, errors.New(awserrors.ErrorInvalidInternetGatewayIDNotFound) - } - var igwRecord handlers_ec2_igw.IGWRecord - if err := json.Unmarshal(igwEntry.Value(), &igwRecord); err != nil { - return nil, errors.New(awserrors.ErrorServerInternal) - } - if igwRecord.VpcId != record.VpcId { - return nil, errors.New(awserrors.ErrorInvalidParameterValue) - } - - route := RouteRecord{ - DestinationCidrBlock: destCidr, - GatewayId: igwID, - State: "active", - Origin: "CreateRoute", - } record.Routes = append(record.Routes, route) if err := s.putRouteTable(accountID, record); err != nil { return nil, err } - slog.Info("CreateRoute completed", "routeTableId", rtbID, "destination", destCidr, "gatewayId", igwID, "accountID", accountID) + slog.Info("CreateRoute completed", "routeTableId", rtbID, "destination", destCidr, "accountID", accountID) return &ec2.CreateRouteOutput{ Return: aws.Bool(true), @@ -784,6 +822,49 @@ func containsString(slice []string, s string) bool { return slices.Contains(slice, s) } +// publishNatGatewayEvents publishes vpc.add-nat-gateway events for each subnet +// associated with this route table, so vpcd creates the SNAT rules. +func (s *RouteTableServiceImpl) publishNatGatewayEvents(accountID string, record *RouteTableRecord, vpcID, natgwID, publicIp string) { + if s.natsConn == nil { + return + } + for _, assoc := range record.Associations { + if assoc.SubnetId == "" || assoc.Main { + continue + } + subnetEntry, err := s.subnetKV.Get(utils.AccountKey(accountID, assoc.SubnetId)) + if err != nil { + continue + } + var subnet handlers_ec2_vpc.SubnetRecord + if err := json.Unmarshal(subnetEntry.Value(), &subnet); err != nil { + continue + } + + evt := struct { + VpcId string `json:"vpc_id"` + NatGatewayId string `json:"nat_gateway_id"` + PublicIp string `json:"public_ip"` + SubnetCidr string `json:"subnet_cidr"` + }{ + VpcId: vpcID, + NatGatewayId: natgwID, + PublicIp: publicIp, + SubnetCidr: subnet.CidrBlock, + } + eventData, err := json.Marshal(evt) + if err != nil { + slog.Warn("Failed to marshal NAT GW add event", "err", err) + continue + } + if err := s.natsConn.Publish("vpc.add-nat-gateway", eventData); err != nil { + slog.Warn("Failed to publish NAT GW add event", "subnet", assoc.SubnetId, "err", err) + } else { + slog.Info("Published NAT GW add event", "subnetCidr", subnet.CidrBlock, "publicIp", publicIp) + } + } +} + // recordToEC2 converts an internal record to an AWS SDK RouteTable struct func recordToEC2(record *RouteTableRecord) *ec2.RouteTable { rtb := &ec2.RouteTable{ diff --git a/spinifex/services/vpcd/topology.go b/spinifex/services/vpcd/topology.go index 7153fcfb6..047d2989e 100644 --- a/spinifex/services/vpcd/topology.go +++ b/spinifex/services/vpcd/topology.go @@ -15,17 +15,19 @@ import ( // NATS topics for VPC lifecycle events published by the daemon. const ( - TopicVPCCreate = "vpc.create" - TopicVPCDelete = "vpc.delete" - TopicSubnetCreate = "vpc.create-subnet" - TopicSubnetDelete = "vpc.delete-subnet" - TopicCreatePort = "vpc.create-port" - TopicDeletePort = "vpc.delete-port" - TopicPortStatus = "vpc.port-status" - TopicIGWAttach = "vpc.igw-attach" - TopicIGWDetach = "vpc.igw-detach" - TopicAddNAT = "vpc.add-nat" - TopicDeleteNAT = "vpc.delete-nat" + TopicVPCCreate = "vpc.create" + TopicVPCDelete = "vpc.delete" + TopicSubnetCreate = "vpc.create-subnet" + TopicSubnetDelete = "vpc.delete-subnet" + TopicCreatePort = "vpc.create-port" + TopicDeletePort = "vpc.delete-port" + TopicPortStatus = "vpc.port-status" + TopicIGWAttach = "vpc.igw-attach" + TopicIGWDetach = "vpc.igw-detach" + TopicAddNAT = "vpc.add-nat" + TopicDeleteNAT = "vpc.delete-nat" + TopicAddNATGateway = "vpc.add-nat-gateway" + TopicDeleteNATGateway = "vpc.delete-nat-gateway" ) // VPCEvent is published on vpc.create after a VPC is persisted. @@ -60,6 +62,14 @@ type NATEvent struct { MAC string `json:"mac"` // external MAC for distributed NAT } +// NATGatewayEvent is published on vpc.add-nat-gateway / vpc.delete-nat-gateway. +type NATGatewayEvent struct { + VpcId string `json:"vpc_id"` + NatGatewayId string `json:"nat_gateway_id"` + PublicIp string `json:"public_ip"` + SubnetCidr string `json:"subnet_cidr"` // private subnet CIDR for SNAT rule +} + // Bridge mode constants for external connectivity. const ( // BridgeModeMacvlan uses a macvlan sub-interface on the WAN NIC. SSH-safe @@ -185,6 +195,8 @@ func (h *TopologyHandler) Subscribe(nc *nats.Conn) ([]*nats.Subscription, error) {TopicIGWDetach, h.handleIGWDetach, true}, {TopicAddNAT, h.handleAddNAT, true}, {TopicDeleteNAT, h.handleDeleteNAT, true}, + {TopicAddNATGateway, h.handleAddNATGateway, true}, + {TopicDeleteNATGateway, h.handleDeleteNATGateway, true}, {TopicCreateSG, h.handleCreateSG, true}, {TopicDeleteSG, h.handleDeleteSG, true}, {TopicUpdateSG, h.handleUpdateSG, true}, @@ -1218,3 +1230,64 @@ func respond(msg *nats.Msg, err error) { slog.Error("vpcd: failed to respond to NATS request", "err", err) } } + +// handleAddNATGateway creates an OVN SNAT rule for a private subnet via a NAT Gateway. +// The SNAT rule rewrites source IPs from the private subnet CIDR to the NAT GW's public IP. +func (h *TopologyHandler) handleAddNATGateway(msg *nats.Msg) { + var evt NATGatewayEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.add-nat-gateway event", "err", err) + return + } + + slog.Info("vpcd: adding NAT Gateway SNAT rule", + "vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId, + "publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr) + + ctx := context.Background() + routerName := "vpc-" + evt.VpcId + + snatRule := &nbdb.NAT{ + Type: "snat", + ExternalIP: evt.PublicIp, + LogicalIP: evt.SubnetCidr, + ExternalIDs: map[string]string{ + "spinifex:vpc_id": evt.VpcId, + "spinifex:nat_gateway_id": evt.NatGatewayId, + }, + } + + if err := h.ovn.AddNAT(ctx, routerName, snatRule); err != nil { + slog.Error("vpcd: failed to add NAT Gateway SNAT rule", + "router", routerName, "publicIp", evt.PublicIp, + "subnetCidr", evt.SubnetCidr, "err", err) + return + } + + slog.Info("vpcd: NAT Gateway SNAT rule added", + "router", routerName, "publicIp", evt.PublicIp, "subnetCidr", evt.SubnetCidr) +} + +// handleDeleteNATGateway removes the OVN SNAT rule for a private subnet's NAT Gateway. +func (h *TopologyHandler) handleDeleteNATGateway(msg *nats.Msg) { + var evt NATGatewayEvent + if err := json.Unmarshal(msg.Data, &evt); err != nil { + slog.Error("vpcd: failed to unmarshal vpc.delete-nat-gateway event", "err", err) + return + } + + slog.Info("vpcd: removing NAT Gateway SNAT rule", + "vpcId", evt.VpcId, "natGatewayId", evt.NatGatewayId, "subnetCidr", evt.SubnetCidr) + + ctx := context.Background() + routerName := "vpc-" + evt.VpcId + + if err := h.ovn.DeleteNAT(ctx, routerName, "snat", evt.SubnetCidr); err != nil { + slog.Warn("vpcd: failed to delete NAT Gateway SNAT rule", + "router", routerName, "subnetCidr", evt.SubnetCidr, "err", err) + return + } + + slog.Info("vpcd: NAT Gateway SNAT rule removed", + "router", routerName, "subnetCidr", evt.SubnetCidr) +}