feat: route tables V1 — CRUD, associations, VPC integration (mulga-760)

Implements AWS-compatible route tables for VPC subnets:

Route Table CRUD:
- CreateRouteTable, DeleteRouteTable, DescribeRouteTables
- Main route table auto-created with every VPC (CreateVpc + EnsureDefaultVPC)
- Cannot delete main table or tables with subnet associations

Route CRUD:
- CreateRoute (IGW targets), DeleteRoute, ReplaceRoute
- Immutable local route (VPC CIDR → local) cannot be deleted
- IGW ownership validation (must be attached to same VPC)
- Duplicate destination CIDR rejected

Association CRUD:
- AssociateRouteTable, DisassociateRouteTable, ReplaceRouteTableAssociation
- One explicit association per subnet
- Disassociating reverts subnet to main table

VPC Integration:
- CreateVpc auto-creates main route table with local route
- EnsureDefaultVPC creates main route table for default VPC
- ensureDefaultVPCInfrastructure adds 0.0.0.0/0 → IGW route

DescribeRouteTables filters: vpc-id, route-table-id, association.main,
association.subnet-id, route.destination-cidr-block, route.gateway-id

20 unit tests covering all CRUD operations and error paths.
This commit is contained in:
Ben Duncan
2026-03-26 12:08:21 +10:00
parent 74671a05a5
commit f8fc2b2cbb
20 changed files with 1740 additions and 0 deletions
+18
View File
@@ -39,6 +39,7 @@ import (
handlers_ec2_instance "github.com/mulgadc/spinifex/spinifex/handlers/ec2/instance"
handlers_ec2_key "github.com/mulgadc/spinifex/spinifex/handlers/ec2/key"
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"
handlers_ec2_tags "github.com/mulgadc/spinifex/spinifex/handlers/ec2/tags"
handlers_ec2_volume "github.com/mulgadc/spinifex/spinifex/handlers/ec2/volume"
@@ -111,6 +112,7 @@ type Daemon struct {
placementGroupService *handlers_ec2_placementgroup.PlacementGroupServiceImpl
vpcService *handlers_ec2_vpc.VPCServiceImpl
eipService *handlers_ec2_eip.EIPServiceImpl
routeTableService *handlers_ec2_routetable.RouteTableServiceImpl
externalIPAM *handlers_ec2_vpc.ExternalIPAM
ctx context.Context
cancel context.CancelFunc
@@ -402,6 +404,15 @@ func (d *Daemon) subscribeAll() error {
{"ec2.RemoveInstanceFromPlacementGroup", d.handleEC2RemoveInstanceFromPlacementGroup, "spinifex-workers"},
{"ec2.ReserveClusterNode", d.handleEC2ReserveClusterNode, "spinifex-workers"},
{"ec2.FinalizeClusterInstances", d.handleEC2FinalizeClusterInstances, "spinifex-workers"},
{"ec2.CreateRouteTable", d.handleEC2CreateRouteTable, "spinifex-workers"},
{"ec2.DeleteRouteTable", d.handleEC2DeleteRouteTable, "spinifex-workers"},
{"ec2.DescribeRouteTables", d.handleEC2DescribeRouteTables, "spinifex-workers"},
{"ec2.CreateRoute", d.handleEC2CreateRoute, "spinifex-workers"},
{"ec2.DeleteRoute", d.handleEC2DeleteRoute, "spinifex-workers"},
{"ec2.ReplaceRoute", d.handleEC2ReplaceRoute, "spinifex-workers"},
{"ec2.AssociateRouteTable", d.handleEC2AssociateRouteTable, "spinifex-workers"},
{"ec2.DisassociateRouteTable", d.handleEC2DisassociateRouteTable, "spinifex-workers"},
{"ec2.ReplaceRouteTableAssociation", d.handleEC2ReplaceRouteTableAssociation, "spinifex-workers"},
{"ec2.CreateVpc", d.handleEC2CreateVpc, "spinifex-workers"},
{"ec2.DeleteVpc", d.handleEC2DeleteVpc, "spinifex-workers"},
{"ec2.DescribeVpcs", d.handleEC2DescribeVpcs, "spinifex-workers"},
@@ -554,6 +565,13 @@ func (d *Daemon) Start() error {
return fmt.Errorf("failed to initialize VPC service: %w", err)
}
d.routeTableService, err = initServiceWithRetry("RouteTable service", func() (*handlers_ec2_routetable.RouteTableServiceImpl, error) {
return handlers_ec2_routetable.NewRouteTableServiceImplWithNATS(d.config, d.natsConn)
})
if err != nil {
return fmt.Errorf("failed to initialize RouteTable 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" {
@@ -0,0 +1,41 @@
package daemon
import (
"github.com/nats-io/nats.go"
)
func (d *Daemon) handleEC2CreateRouteTable(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.CreateRouteTable)
}
func (d *Daemon) handleEC2DeleteRouteTable(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.DeleteRouteTable)
}
func (d *Daemon) handleEC2DescribeRouteTables(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.DescribeRouteTables)
}
func (d *Daemon) handleEC2CreateRoute(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.CreateRoute)
}
func (d *Daemon) handleEC2DeleteRoute(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.DeleteRoute)
}
func (d *Daemon) handleEC2ReplaceRoute(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.ReplaceRoute)
}
func (d *Daemon) handleEC2AssociateRouteTable(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.AssociateRouteTable)
}
func (d *Daemon) handleEC2DisassociateRouteTable(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.DisassociateRouteTable)
}
func (d *Daemon) handleEC2ReplaceRouteTableAssociation(msg *nats.Msg) {
handleNATSRequest(msg, d.routeTableService.ReplaceRouteTableAssociation)
}
+64
View File
@@ -134,6 +134,11 @@ func (d *Daemon) ensureDefaultVPCInfrastructure() {
}
}
// Add 0.0.0.0/0 → IGW route to the main route table (if not already present)
if d.routeTableService != nil {
d.ensureDefaultIGWRoute(accountID, defaultVpcId)
}
// Check if default security group exists for this VPC
sgOut, err := d.vpcService.DescribeSecurityGroups(&ec2.DescribeSecurityGroupsInput{}, accountID)
if err != nil {
@@ -188,3 +193,62 @@ func (d *Daemon) ensureDefaultVPCInfrastructure() {
}
}
}
// ensureDefaultIGWRoute adds 0.0.0.0/0 → igw-xxx to the main route table if not present
func (d *Daemon) ensureDefaultIGWRoute(accountID, vpcID string) {
// Find the main route table for this VPC
vpcFilter := "vpc-id"
mainFilter := "association.main"
trueVal := "true"
descOut, err := d.routeTableService.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
Filters: []*ec2.Filter{
{Name: &vpcFilter, Values: []*string{&vpcID}},
{Name: &mainFilter, Values: []*string{&trueVal}},
},
}, accountID)
if err != nil || len(descOut.RouteTables) == 0 {
slog.Debug("No main route table found for default IGW route", "vpcId", vpcID, "accountID", accountID)
return
}
mainRtb := descOut.RouteTables[0]
// Check if 0.0.0.0/0 route already exists
for _, r := range mainRtb.Routes {
if r.DestinationCidrBlock != nil && *r.DestinationCidrBlock == "0.0.0.0/0" {
return // Already has a default route
}
}
// Find the attached IGW for this VPC
igwOut, err := d.igwService.DescribeInternetGateways(&ec2.DescribeInternetGatewaysInput{}, accountID)
if err != nil {
return
}
var igwID string
for _, igw := range igwOut.InternetGateways {
for _, att := range igw.Attachments {
if att.VpcId != nil && *att.VpcId == vpcID {
igwID = *igw.InternetGatewayId
break
}
}
}
if igwID == "" {
return // No IGW attached
}
// Add the default route
dest := "0.0.0.0/0"
_, err = d.routeTableService.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: mainRtb.RouteTableId,
DestinationCidrBlock: &dest,
GatewayId: &igwID,
}, accountID)
if err != nil {
slog.Warn("Failed to add default IGW route to main route table", "err", err)
} else {
slog.Info("Added default IGW route to main route table",
"routeTableId", *mainRtb.RouteTableId, "igwId", igwID, "vpcId", vpcID)
}
}
+28
View File
@@ -20,6 +20,7 @@ import (
gateway_ec2_instance "github.com/mulgadc/spinifex/spinifex/gateway/ec2/instance"
gateway_ec2_key "github.com/mulgadc/spinifex/spinifex/gateway/ec2/key"
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"
gateway_ec2_tags "github.com/mulgadc/spinifex/spinifex/gateway/ec2/tags"
gateway_ec2_volume "github.com/mulgadc/spinifex/spinifex/gateway/ec2/volume"
@@ -229,6 +230,33 @@ var ec2Actions = map[string]EC2Handler{
"ModifySubnetAttribute": ec2Handler(func(input *ec2.ModifySubnetAttributeInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_vpc.ModifySubnetAttribute(input, gw.NATSConn, accountID)
}),
"CreateRouteTable": ec2Handler(func(input *ec2.CreateRouteTableInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.CreateRouteTable(input, gw.NATSConn, accountID)
}),
"DeleteRouteTable": ec2Handler(func(input *ec2.DeleteRouteTableInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.DeleteRouteTable(input, gw.NATSConn, accountID)
}),
"DescribeRouteTables": ec2Handler(func(input *ec2.DescribeRouteTablesInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.DescribeRouteTables(input, gw.NATSConn, accountID)
}),
"CreateRoute": ec2Handler(func(input *ec2.CreateRouteInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.CreateRoute(input, gw.NATSConn, accountID)
}),
"DeleteRoute": ec2Handler(func(input *ec2.DeleteRouteInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.DeleteRoute(input, gw.NATSConn, accountID)
}),
"ReplaceRoute": ec2Handler(func(input *ec2.ReplaceRouteInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.ReplaceRoute(input, gw.NATSConn, accountID)
}),
"AssociateRouteTable": ec2Handler(func(input *ec2.AssociateRouteTableInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.AssociateRouteTable(input, gw.NATSConn, accountID)
}),
"DisassociateRouteTable": ec2Handler(func(input *ec2.DisassociateRouteTableInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.DisassociateRouteTable(input, gw.NATSConn, accountID)
}),
"ReplaceRouteTableAssociation": ec2Handler(func(input *ec2.ReplaceRouteTableAssociationInput, gw *GatewayConfig, accountID string) (any, error) {
return gateway_ec2_routetable.ReplaceRouteTableAssociation(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,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func AssociateRouteTable(input *ec2.AssociateRouteTableInput, natsConn *nats.Conn, accountID string) (ec2.AssociateRouteTableOutput, error) {
var output ec2.AssociateRouteTableOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.AssociateRouteTable(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func CreateRoute(input *ec2.CreateRouteInput, natsConn *nats.Conn, accountID string) (ec2.CreateRouteOutput, error) {
var output ec2.CreateRouteOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.CreateRoute(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func CreateRouteTable(input *ec2.CreateRouteTableInput, natsConn *nats.Conn, accountID string) (ec2.CreateRouteTableOutput, error) {
var output ec2.CreateRouteTableOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.CreateRouteTable(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func DeleteRoute(input *ec2.DeleteRouteInput, natsConn *nats.Conn, accountID string) (ec2.DeleteRouteOutput, error) {
var output ec2.DeleteRouteOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.DeleteRoute(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func DeleteRouteTable(input *ec2.DeleteRouteTableInput, natsConn *nats.Conn, accountID string) (ec2.DeleteRouteTableOutput, error) {
var output ec2.DeleteRouteTableOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.DeleteRouteTable(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func DescribeRouteTables(input *ec2.DescribeRouteTablesInput, natsConn *nats.Conn, accountID string) (ec2.DescribeRouteTablesOutput, error) {
var output ec2.DescribeRouteTablesOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.DescribeRouteTables(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func DisassociateRouteTable(input *ec2.DisassociateRouteTableInput, natsConn *nats.Conn, accountID string) (ec2.DisassociateRouteTableOutput, error) {
var output ec2.DisassociateRouteTableOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.DisassociateRouteTable(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func ReplaceRoute(input *ec2.ReplaceRouteInput, natsConn *nats.Conn, accountID string) (ec2.ReplaceRouteOutput, error) {
var output ec2.ReplaceRouteOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.ReplaceRoute(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
@@ -0,0 +1,23 @@
package gateway_ec2_routetable
import (
"errors"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_routetable "github.com/mulgadc/spinifex/spinifex/handlers/ec2/routetable"
"github.com/nats-io/nats.go"
)
func ReplaceRouteTableAssociation(input *ec2.ReplaceRouteTableAssociationInput, natsConn *nats.Conn, accountID string) (ec2.ReplaceRouteTableAssociationOutput, error) {
var output ec2.ReplaceRouteTableAssociationOutput
if input == nil {
return output, errors.New(awserrors.ErrorInvalidParameterValue)
}
svc := handlers_ec2_routetable.NewNATSRouteTableService(natsConn)
result, err := svc.ReplaceRouteTableAssociation(input, accountID)
if err != nil {
return output, err
}
return *result, nil
}
+3
View File
@@ -1028,6 +1028,9 @@ func TestEC2ActionMapCompleteness(t *testing.T) {
"AuthorizeSecurityGroupIngress", "AuthorizeSecurityGroupEgress",
"RevokeSecurityGroupIngress", "RevokeSecurityGroupEgress",
"AllocateAddress", "ReleaseAddress", "AssociateAddress", "DisassociateAddress", "DescribeAddresses",
"CreateRouteTable", "DeleteRouteTable", "DescribeRouteTables",
"CreateRoute", "DeleteRoute", "ReplaceRoute",
"AssociateRouteTable", "DisassociateRouteTable", "ReplaceRouteTableAssociation",
}
for _, action := range expectedActions {
@@ -0,0 +1,16 @@
package handlers_ec2_routetable
import "github.com/aws/aws-sdk-go/service/ec2"
// RouteTableService defines the interface for Route Table operations
type RouteTableService interface {
CreateRouteTable(input *ec2.CreateRouteTableInput, accountID string) (*ec2.CreateRouteTableOutput, error)
DeleteRouteTable(input *ec2.DeleteRouteTableInput, accountID string) (*ec2.DeleteRouteTableOutput, error)
DescribeRouteTables(input *ec2.DescribeRouteTablesInput, accountID string) (*ec2.DescribeRouteTablesOutput, error)
CreateRoute(input *ec2.CreateRouteInput, accountID string) (*ec2.CreateRouteOutput, error)
DeleteRoute(input *ec2.DeleteRouteInput, accountID string) (*ec2.DeleteRouteOutput, error)
ReplaceRoute(input *ec2.ReplaceRouteInput, accountID string) (*ec2.ReplaceRouteOutput, error)
AssociateRouteTable(input *ec2.AssociateRouteTableInput, accountID string) (*ec2.AssociateRouteTableOutput, error)
DisassociateRouteTable(input *ec2.DisassociateRouteTableInput, accountID string) (*ec2.DisassociateRouteTableOutput, error)
ReplaceRouteTableAssociation(input *ec2.ReplaceRouteTableAssociationInput, accountID string) (*ec2.ReplaceRouteTableAssociationOutput, error)
}
@@ -0,0 +1,792 @@
package handlers_ec2_routetable
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"
"github.com/mulgadc/spinifex/spinifex/config"
handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
// Ensure RouteTableServiceImpl implements RouteTableService
var _ RouteTableService = (*RouteTableServiceImpl)(nil)
// RouteTableServiceImpl implements Route Table operations with NATS JetStream persistence
type RouteTableServiceImpl struct {
config *config.Config
rtbKV nats.KeyValue
vpcKV nats.KeyValue
igwKV nats.KeyValue
subnetKV nats.KeyValue
natsConn *nats.Conn
}
// NewRouteTableServiceImplWithNATS creates a Route Table service with NATS JetStream for persistence
func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*RouteTableServiceImpl, error) {
js, err := natsConn.JetStream()
if err != nil {
return nil, fmt.Errorf("failed to get JetStream context: %w", err)
}
rtbKV, err := utils.GetOrCreateKVBucket(js, KVBucketRouteTables, 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketRouteTables, err)
}
if err := utils.WriteVersion(rtbKV, KVBucketRouteTablesVersion); err != nil {
return nil, fmt.Errorf("write version to %s: %w", KVBucketRouteTables, 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)
}
igwKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_igw.KVBucketIGW, 10)
if err != nil {
return nil, fmt.Errorf("failed to get IGW 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)
}
slog.Info("RouteTable service initialized with JetStream KV", "bucket", KVBucketRouteTables)
return &RouteTableServiceImpl{
config: cfg,
rtbKV: rtbKV,
vpcKV: vpcKV,
igwKV: igwKV,
subnetKV: subnetKV,
natsConn: natsConn,
}, nil
}
// getRouteTable retrieves a route table record from KV
func (s *RouteTableServiceImpl) getRouteTable(accountID, rtbID string) (*RouteTableRecord, error) {
entry, err := s.rtbKV.Get(utils.AccountKey(accountID, rtbID))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidRouteTableIDNotFound)
}
var record RouteTableRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
return &record, nil
}
// putRouteTable stores a route table record to KV
func (s *RouteTableServiceImpl) putRouteTable(accountID string, record *RouteTableRecord) error {
data, err := json.Marshal(record)
if err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
if _, err := s.rtbKV.Put(utils.AccountKey(accountID, record.RouteTableId), data); err != nil {
return errors.New(awserrors.ErrorServerInternal)
}
return nil
}
// getVPCCidr looks up a VPC's CIDR block from the VPC KV bucket
func (s *RouteTableServiceImpl) getVPCCidr(accountID, vpcID string) (string, error) {
entry, err := s.vpcKV.Get(utils.AccountKey(accountID, vpcID))
if err != nil {
return "", errors.New(awserrors.ErrorInvalidVpcIDNotFound)
}
var vpcRecord handlers_ec2_vpc.VPCRecord
if err := json.Unmarshal(entry.Value(), &vpcRecord); err != nil {
return "", errors.New(awserrors.ErrorServerInternal)
}
return vpcRecord.CidrBlock, nil
}
// allRouteTablesForVPC returns all route tables belonging to a VPC
func (s *RouteTableServiceImpl) allRouteTablesForVPC(accountID, vpcID string) ([]RouteTableRecord, error) {
prefix := accountID + "."
keys, err := s.rtbKV.Keys()
if err != nil {
if errors.Is(err, nats.ErrNoKeysFound) {
return nil, nil
}
return nil, errors.New(awserrors.ErrorServerInternal)
}
var results []RouteTableRecord
for _, key := range keys {
if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(key)
if err != nil {
continue
}
var record RouteTableRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
continue
}
if record.VpcId == vpcID {
results = append(results, record)
}
}
return results, nil
}
// CreateRouteTableForVPC creates a route table with the local route pre-populated.
// Exported for use by VPC service when creating default VPCs.
func (s *RouteTableServiceImpl) CreateRouteTableForVPC(vpcID, vpcCidr, accountID string, isMain bool, rtbID string) (*RouteTableRecord, error) {
if rtbID == "" {
rtbID = utils.GenerateResourceID("rtb")
}
record := RouteTableRecord{
RouteTableId: rtbID,
VpcId: vpcID,
AccountID: accountID,
IsMain: isMain,
Routes: []RouteRecord{
{
DestinationCidrBlock: vpcCidr,
GatewayId: "local",
State: "active",
Origin: "CreateRouteTable",
},
},
Tags: make(map[string]string),
CreatedAt: time.Now(),
}
if isMain {
record.Associations = []AssociationRecord{
{
AssociationId: utils.GenerateResourceID("rtbassoc"),
Main: true,
},
}
}
if err := s.putRouteTable(accountID, &record); err != nil {
return nil, err
}
slog.Info("Created route table", "routeTableId", rtbID, "vpcId", vpcID, "isMain", isMain, "accountID", accountID)
return &record, nil
}
// CreateRouteTable creates a new custom (non-main) route table for a VPC
func (s *RouteTableServiceImpl) CreateRouteTable(input *ec2.CreateRouteTableInput, accountID string) (*ec2.CreateRouteTableOutput, error) {
if input.VpcId == nil || *input.VpcId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
vpcID := *input.VpcId
vpcCidr, err := s.getVPCCidr(accountID, vpcID)
if err != nil {
return nil, err
}
record, err := s.CreateRouteTableForVPC(vpcID, vpcCidr, accountID, false, "")
if err != nil {
return nil, err
}
return &ec2.CreateRouteTableOutput{
RouteTable: recordToEC2(record),
}, nil
}
// DeleteRouteTable deletes a route table (must not be main, must have no subnet associations)
func (s *RouteTableServiceImpl) DeleteRouteTable(input *ec2.DeleteRouteTableInput, accountID string) (*ec2.DeleteRouteTableOutput, error) {
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
rtbID := *input.RouteTableId
record, err := s.getRouteTable(accountID, rtbID)
if err != nil {
return nil, err
}
if record.IsMain {
return nil, errors.New(awserrors.ErrorDependencyViolation)
}
// Check for non-main associations (subnets still using this table)
for _, assoc := range record.Associations {
if !assoc.Main && assoc.SubnetId != "" {
return nil, errors.New(awserrors.ErrorDependencyViolation)
}
}
if err := s.rtbKV.Delete(utils.AccountKey(accountID, rtbID)); err != nil {
return nil, errors.New(awserrors.ErrorServerInternal)
}
slog.Info("DeleteRouteTable completed", "routeTableId", rtbID, "accountID", accountID)
return &ec2.DeleteRouteTableOutput{}, nil
}
// DescribeRouteTables lists route tables, optionally filtered
func (s *RouteTableServiceImpl) DescribeRouteTables(input *ec2.DescribeRouteTablesInput, accountID string) (*ec2.DescribeRouteTablesOutput, error) {
rtbIDs := make(map[string]bool)
for _, id := range input.RouteTableIds {
if id != nil {
rtbIDs[*id] = true
}
}
// Build filter map
filters := make(map[string][]string)
for _, f := range input.Filters {
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.rtbKV.Keys()
if err != nil && !errors.Is(err, nats.ErrNoKeysFound) {
return nil, errors.New(awserrors.ErrorServerInternal)
}
var routeTables []*ec2.RouteTable
foundIDs := make(map[string]bool)
for _, key := range keys {
if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(key)
if err != nil {
continue
}
var record RouteTableRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
continue
}
if len(rtbIDs) > 0 && !rtbIDs[record.RouteTableId] {
continue
}
if !matchesFilters(&record, filters) {
continue
}
routeTables = append(routeTables, recordToEC2(&record))
foundIDs[record.RouteTableId] = true
}
// Return error if specific IDs were requested but not found
for id := range rtbIDs {
if !foundIDs[id] {
return nil, errors.New(awserrors.ErrorInvalidRouteTableIDNotFound)
}
}
return &ec2.DescribeRouteTablesOutput{
RouteTables: routeTables,
}, nil
}
// CreateRoute adds a route to a route table
func (s *RouteTableServiceImpl) CreateRoute(input *ec2.CreateRouteInput, accountID string) (*ec2.CreateRouteOutput, error) {
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if input.DestinationCidrBlock == nil || *input.DestinationCidrBlock == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
rtbID := *input.RouteTableId
destCidr := *input.DestinationCidrBlock
record, err := s.getRouteTable(accountID, rtbID)
if err != nil {
return nil, err
}
// Check for duplicate destination
for _, r := range record.Routes {
if r.DestinationCidrBlock == destCidr {
return nil, errors.New(awserrors.ErrorRouteAlreadyExists)
}
}
// V1: only GatewayId target supported
if input.GatewayId == nil || *input.GatewayId == "" {
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)
return &ec2.CreateRouteOutput{
Return: aws.Bool(true),
}, nil
}
// DeleteRoute removes a route from a route table (cannot delete local route)
func (s *RouteTableServiceImpl) DeleteRoute(input *ec2.DeleteRouteInput, accountID string) (*ec2.DeleteRouteOutput, error) {
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if input.DestinationCidrBlock == nil || *input.DestinationCidrBlock == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
rtbID := *input.RouteTableId
destCidr := *input.DestinationCidrBlock
record, err := s.getRouteTable(accountID, rtbID)
if err != nil {
return nil, err
}
idx := -1
for i, r := range record.Routes {
if r.DestinationCidrBlock == destCidr {
idx = i
break
}
}
if idx < 0 {
return nil, errors.New(awserrors.ErrorInvalidRouteNotFound)
}
// Cannot delete local route
if record.Routes[idx].GatewayId == "local" {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
record.Routes = append(record.Routes[:idx], record.Routes[idx+1:]...)
if err := s.putRouteTable(accountID, record); err != nil {
return nil, err
}
slog.Info("DeleteRoute completed", "routeTableId", rtbID, "destination", destCidr, "accountID", accountID)
return &ec2.DeleteRouteOutput{}, nil
}
// ReplaceRoute atomically replaces the target of an existing route
func (s *RouteTableServiceImpl) ReplaceRoute(input *ec2.ReplaceRouteInput, accountID string) (*ec2.ReplaceRouteOutput, error) {
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if input.DestinationCidrBlock == nil || *input.DestinationCidrBlock == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
rtbID := *input.RouteTableId
destCidr := *input.DestinationCidrBlock
record, err := s.getRouteTable(accountID, rtbID)
if err != nil {
return nil, err
}
idx := -1
for i, r := range record.Routes {
if r.DestinationCidrBlock == destCidr {
idx = i
break
}
}
if idx < 0 {
return nil, errors.New(awserrors.ErrorInvalidRouteNotFound)
}
if record.Routes[idx].GatewayId == "local" {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
// V1: only GatewayId target supported
if input.GatewayId == nil || *input.GatewayId == "" {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
igwID := *input.GatewayId
// Verify IGW exists and is attached to 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)
}
record.Routes[idx].GatewayId = igwID
record.Routes[idx].NatGatewayId = ""
record.Routes[idx].State = "active"
if err := s.putRouteTable(accountID, record); err != nil {
return nil, err
}
slog.Info("ReplaceRoute completed", "routeTableId", rtbID, "destination", destCidr, "gatewayId", igwID, "accountID", accountID)
return &ec2.ReplaceRouteOutput{}, nil
}
// AssociateRouteTable associates a subnet with a route table
func (s *RouteTableServiceImpl) AssociateRouteTable(input *ec2.AssociateRouteTableInput, accountID string) (*ec2.AssociateRouteTableOutput, error) {
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if input.SubnetId == nil || *input.SubnetId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
rtbID := *input.RouteTableId
subnetID := *input.SubnetId
record, err := s.getRouteTable(accountID, rtbID)
if err != nil {
return nil, err
}
// Verify subnet exists and belongs to the same 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)
}
if subnetRecord.VpcId != record.VpcId {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
// Check the subnet doesn't already have an explicit association in any route table
allTables, err := s.allRouteTablesForVPC(accountID, record.VpcId)
if err != nil {
return nil, err
}
for _, table := range allTables {
for _, assoc := range table.Associations {
if assoc.SubnetId == subnetID && !assoc.Main {
return nil, errors.New(awserrors.ErrorResourceAlreadyAssociated)
}
}
}
assocID := utils.GenerateResourceID("rtbassoc")
record.Associations = append(record.Associations, AssociationRecord{
AssociationId: assocID,
SubnetId: subnetID,
Main: false,
})
if err := s.putRouteTable(accountID, record); err != nil {
return nil, err
}
slog.Info("AssociateRouteTable completed", "routeTableId", rtbID, "subnetId", subnetID, "associationId", assocID, "accountID", accountID)
return &ec2.AssociateRouteTableOutput{
AssociationId: aws.String(assocID),
AssociationState: &ec2.RouteTableAssociationState{
State: aws.String("associated"),
},
}, nil
}
// DisassociateRouteTable removes a subnet association (cannot disassociate main)
func (s *RouteTableServiceImpl) DisassociateRouteTable(input *ec2.DisassociateRouteTableInput, accountID string) (*ec2.DisassociateRouteTableOutput, error) {
if input.AssociationId == nil || *input.AssociationId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
assocID := *input.AssociationId
// Search all route tables for this account to find the association
prefix := accountID + "."
keys, err := s.rtbKV.Keys()
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidAssociationIDNotFound)
}
for _, key := range keys {
if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(key)
if err != nil {
continue
}
var record RouteTableRecord
if err := json.Unmarshal(entry.Value(), &record); err != nil {
continue
}
for i, assoc := range record.Associations {
if assoc.AssociationId == assocID {
if assoc.Main {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
record.Associations = append(record.Associations[:i], record.Associations[i+1:]...)
if err := s.putRouteTable(accountID, &record); err != nil {
return nil, err
}
slog.Info("DisassociateRouteTable completed", "associationId", assocID, "routeTableId", record.RouteTableId, "accountID", accountID)
return &ec2.DisassociateRouteTableOutput{}, nil
}
}
}
return nil, errors.New(awserrors.ErrorInvalidAssociationIDNotFound)
}
// ReplaceRouteTableAssociation atomically moves a subnet from one route table to another
func (s *RouteTableServiceImpl) ReplaceRouteTableAssociation(input *ec2.ReplaceRouteTableAssociationInput, accountID string) (*ec2.ReplaceRouteTableAssociationOutput, error) {
if input.AssociationId == nil || *input.AssociationId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
if input.RouteTableId == nil || *input.RouteTableId == "" {
return nil, errors.New(awserrors.ErrorMissingParameter)
}
assocID := *input.AssociationId
newRtbID := *input.RouteTableId
// Verify the new route table exists
newRecord, err := s.getRouteTable(accountID, newRtbID)
if err != nil {
return nil, err
}
// Find and remove the association from its current route table
prefix := accountID + "."
keys, err := s.rtbKV.Keys()
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidAssociationIDNotFound)
}
for _, key := range keys {
if key == utils.VersionKey || !strings.HasPrefix(key, prefix) {
continue
}
entry, err := s.rtbKV.Get(key)
if err != nil {
continue
}
var oldRecord RouteTableRecord
if err := json.Unmarshal(entry.Value(), &oldRecord); err != nil {
continue
}
for i, assoc := range oldRecord.Associations {
if assoc.AssociationId != assocID {
continue
}
// If replacing a main association, swap main table for the VPC
if assoc.Main {
// New table becomes main, old table loses main status
newRecord.IsMain = true
oldRecord.IsMain = false
}
// Remove from old table
oldRecord.Associations = append(oldRecord.Associations[:i], oldRecord.Associations[i+1:]...)
if err := s.putRouteTable(accountID, &oldRecord); err != nil {
return nil, err
}
// Add to new table with new ID
newAssocID := utils.GenerateResourceID("rtbassoc")
newRecord.Associations = append(newRecord.Associations, AssociationRecord{
AssociationId: newAssocID,
SubnetId: assoc.SubnetId,
Main: assoc.Main,
})
if err := s.putRouteTable(accountID, newRecord); err != nil {
return nil, err
}
slog.Info("ReplaceRouteTableAssociation completed",
"oldAssociationId", assocID, "newAssociationId", newAssocID,
"newRouteTableId", newRtbID, "accountID", accountID)
return &ec2.ReplaceRouteTableAssociationOutput{
NewAssociationId: aws.String(newAssocID),
AssociationState: &ec2.RouteTableAssociationState{
State: aws.String("associated"),
},
}, nil
}
}
return nil, errors.New(awserrors.ErrorInvalidAssociationIDNotFound)
}
// matchesFilters checks if a route table record matches the given DescribeRouteTables filters
func matchesFilters(record *RouteTableRecord, filters map[string][]string) bool {
for name, values := range filters {
switch name {
case "vpc-id":
if !containsString(values, record.VpcId) {
return false
}
case "route-table-id":
if !containsString(values, record.RouteTableId) {
return false
}
case "association.main":
wantMain := containsString(values, "true")
hasMain := false
for _, a := range record.Associations {
if a.Main {
hasMain = true
break
}
}
if wantMain != hasMain {
return false
}
case "association.subnet-id":
found := false
for _, a := range record.Associations {
if containsString(values, a.SubnetId) {
found = true
break
}
}
if !found {
return false
}
case "route.destination-cidr-block":
found := false
for _, r := range record.Routes {
if containsString(values, r.DestinationCidrBlock) {
found = true
break
}
}
if !found {
return false
}
case "route.gateway-id":
found := false
for _, r := range record.Routes {
if containsString(values, r.GatewayId) {
found = true
break
}
}
if !found {
return false
}
}
}
return true
}
func containsString(slice []string, s string) bool {
return slices.Contains(slice, s)
}
// recordToEC2 converts an internal record to an AWS SDK RouteTable struct
func recordToEC2(record *RouteTableRecord) *ec2.RouteTable {
rtb := &ec2.RouteTable{
RouteTableId: aws.String(record.RouteTableId),
VpcId: aws.String(record.VpcId),
OwnerId: aws.String(record.AccountID),
}
for _, r := range record.Routes {
route := &ec2.Route{
DestinationCidrBlock: aws.String(r.DestinationCidrBlock),
State: aws.String(r.State),
Origin: aws.String(r.Origin),
}
if r.GatewayId != "" {
route.GatewayId = aws.String(r.GatewayId)
}
if r.NatGatewayId != "" {
route.NatGatewayId = aws.String(r.NatGatewayId)
}
rtb.Routes = append(rtb.Routes, route)
}
for _, a := range record.Associations {
assoc := &ec2.RouteTableAssociation{
RouteTableAssociationId: aws.String(a.AssociationId),
RouteTableId: aws.String(record.RouteTableId),
Main: aws.Bool(a.Main),
AssociationState: &ec2.RouteTableAssociationState{
State: aws.String("associated"),
},
}
if a.SubnetId != "" {
assoc.SubnetId = aws.String(a.SubnetId)
}
rtb.Associations = append(rtb.Associations, assoc)
}
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)})
}
rtb.Tags = tags
}
return rtb
}
@@ -0,0 +1,404 @@
package handlers_ec2_routetable
import (
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/awserrors"
handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw"
handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const testAccountID = "123456789012"
func setupTestService(t *testing.T) *RouteTableServiceImpl {
t.Helper()
opts := &server.Options{
Host: "127.0.0.1",
Port: -1,
JetStream: true,
StoreDir: t.TempDir(),
NoLog: true,
NoSigs: true,
}
ns, err := server.NewServer(opts)
require.NoError(t, err)
go ns.Start()
require.True(t, ns.ReadyForConnections(5*time.Second))
t.Cleanup(func() { ns.Shutdown() })
nc, err := nats.Connect(ns.ClientURL())
require.NoError(t, err)
t.Cleanup(func() { nc.Close() })
js, err := nc.JetStream()
require.NoError(t, err)
// Seed VPC KV
vpcKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketVPCs, History: 1})
require.NoError(t, err)
_, err = vpcKV.Put(utils.AccountKey(testAccountID, "vpc-test1"), []byte(`{"vpc_id":"vpc-test1","cidr_block":"10.0.0.0/16","state":"available"}`))
require.NoError(t, err)
// Seed IGW KV (attached to vpc-test1)
igwKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_igw.KVBucketIGW, History: 1})
require.NoError(t, err)
_, err = igwKV.Put(utils.AccountKey(testAccountID, "igw-test1"), []byte(`{"internet_gateway_id":"igw-test1","vpc_id":"vpc-test1","state":"attached"}`))
require.NoError(t, err)
// Seed subnet KV
subnetKV, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: handlers_ec2_vpc.KVBucketSubnets, History: 1})
require.NoError(t, err)
_, err = subnetKV.Put(utils.AccountKey(testAccountID, "subnet-test1"), []byte(`{"subnet_id":"subnet-test1","vpc_id":"vpc-test1","cidr_block":"10.0.1.0/24","state":"available"}`))
require.NoError(t, err)
svc, err := NewRouteTableServiceImplWithNATS(nil, nc)
require.NoError(t, err)
return svc
}
func createTestRtb(t *testing.T, svc *RouteTableServiceImpl) string {
t.Helper()
out, err := svc.CreateRouteTable(&ec2.CreateRouteTableInput{
VpcId: aws.String("vpc-test1"),
}, testAccountID)
require.NoError(t, err)
return *out.RouteTable.RouteTableId
}
func TestCreateRouteTable(t *testing.T) {
svc := setupTestService(t)
out, err := svc.CreateRouteTable(&ec2.CreateRouteTableInput{
VpcId: aws.String("vpc-test1"),
}, testAccountID)
require.NoError(t, err)
rtb := out.RouteTable
assert.NotEmpty(t, *rtb.RouteTableId)
assert.Equal(t, "vpc-test1", *rtb.VpcId)
assert.Empty(t, rtb.Associations) // custom tables have no associations initially
// Should have local route
require.Len(t, rtb.Routes, 1)
assert.Equal(t, "10.0.0.0/16", *rtb.Routes[0].DestinationCidrBlock)
assert.Equal(t, "local", *rtb.Routes[0].GatewayId)
assert.Equal(t, "active", *rtb.Routes[0].State)
}
func TestCreateRouteTable_VpcNotFound(t *testing.T) {
svc := setupTestService(t)
_, err := svc.CreateRouteTable(&ec2.CreateRouteTableInput{
VpcId: aws.String("vpc-nonexistent"),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorInvalidVpcIDNotFound)
}
func TestDeleteRouteTable(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.DeleteRouteTable(&ec2.DeleteRouteTableInput{
RouteTableId: aws.String(rtbID),
}, testAccountID)
require.NoError(t, err)
// Should be gone
_, err = svc.getRouteTable(testAccountID, rtbID)
assert.EqualError(t, err, awserrors.ErrorInvalidRouteTableIDNotFound)
}
func TestDeleteRouteTable_Main(t *testing.T) {
svc := setupTestService(t)
record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "")
require.NoError(t, err)
_, err = svc.DeleteRouteTable(&ec2.DeleteRouteTableInput{
RouteTableId: aws.String(record.RouteTableId),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorDependencyViolation)
}
func TestDeleteRouteTable_WithAssociations(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
// Associate a subnet
_, err := svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtbID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
require.NoError(t, err)
// Should fail to delete
_, err = svc.DeleteRouteTable(&ec2.DeleteRouteTableInput{
RouteTableId: aws.String(rtbID),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorDependencyViolation)
}
func TestDescribeRouteTables(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
out, err := svc.DescribeRouteTables(&ec2.DescribeRouteTablesInput{}, testAccountID)
require.NoError(t, err)
assert.Len(t, out.RouteTables, 1)
assert.Equal(t, rtbID, *out.RouteTables[0].RouteTableId)
}
func TestDescribeRouteTables_FilterByVpcId(t *testing.T) {
svc := setupTestService(t)
createTestRtb(t, svc)
name := "vpc-id"
val := "vpc-test1"
out, err := svc.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
Filters: []*ec2.Filter{{Name: &name, Values: []*string{&val}}},
}, testAccountID)
require.NoError(t, err)
assert.Len(t, out.RouteTables, 1)
// Filter by non-existent VPC
val2 := "vpc-nope"
out, err = svc.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
Filters: []*ec2.Filter{{Name: &name, Values: []*string{&val2}}},
}, testAccountID)
require.NoError(t, err)
assert.Empty(t, out.RouteTables)
}
func TestDescribeRouteTables_FilterByMain(t *testing.T) {
svc := setupTestService(t)
// Create a main route table
_, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "")
require.NoError(t, err)
// Create a non-main route table
createTestRtb(t, svc)
name := "association.main"
val := "true"
out, err := svc.DescribeRouteTables(&ec2.DescribeRouteTablesInput{
Filters: []*ec2.Filter{{Name: &name, Values: []*string{&val}}},
}, testAccountID)
require.NoError(t, err)
assert.Len(t, out.RouteTables, 1)
assert.True(t, *out.RouteTables[0].Associations[0].Main)
}
func TestCreateRoute(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
require.NoError(t, err)
// Verify route was added
record, err := svc.getRouteTable(testAccountID, rtbID)
require.NoError(t, err)
assert.Len(t, record.Routes, 2) // local + igw
assert.Equal(t, "0.0.0.0/0", record.Routes[1].DestinationCidrBlock)
assert.Equal(t, "igw-test1", record.Routes[1].GatewayId)
}
func TestCreateRoute_DuplicateDestination(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
require.NoError(t, err)
// Duplicate should fail
_, err = svc.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorRouteAlreadyExists)
}
func TestDeleteRoute(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
require.NoError(t, err)
_, err = svc.DeleteRoute(&ec2.DeleteRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
}, testAccountID)
require.NoError(t, err)
record, err := svc.getRouteTable(testAccountID, rtbID)
require.NoError(t, err)
assert.Len(t, record.Routes, 1) // only local remains
}
func TestDeleteRoute_LocalRoute(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.DeleteRoute(&ec2.DeleteRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("10.0.0.0/16"),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorInvalidParameterValue)
}
func TestDeleteRoute_NotFound(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.DeleteRoute(&ec2.DeleteRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("192.168.0.0/16"),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorInvalidRouteNotFound)
}
func TestReplaceRoute(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.CreateRoute(&ec2.CreateRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
require.NoError(t, err)
// Replace target (same IGW for simplicity — validates the swap logic)
_, err = svc.ReplaceRoute(&ec2.ReplaceRouteInput{
RouteTableId: aws.String(rtbID),
DestinationCidrBlock: aws.String("0.0.0.0/0"),
GatewayId: aws.String("igw-test1"),
}, testAccountID)
require.NoError(t, err)
}
func TestAssociateRouteTable(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
out, err := svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtbID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
require.NoError(t, err)
assert.NotEmpty(t, *out.AssociationId)
assert.Equal(t, "associated", *out.AssociationState.State)
}
func TestAssociateRouteTable_DuplicateSubnet(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
_, err := svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtbID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
require.NoError(t, err)
// Second association should fail
_, err = svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtbID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorResourceAlreadyAssociated)
}
func TestDisassociateRouteTable(t *testing.T) {
svc := setupTestService(t)
rtbID := createTestRtb(t, svc)
assocOut, err := svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtbID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
require.NoError(t, err)
_, err = svc.DisassociateRouteTable(&ec2.DisassociateRouteTableInput{
AssociationId: assocOut.AssociationId,
}, testAccountID)
require.NoError(t, err)
// Verify association removed
record, err := svc.getRouteTable(testAccountID, rtbID)
require.NoError(t, err)
assert.Empty(t, record.Associations)
}
func TestDisassociateRouteTable_Main(t *testing.T) {
svc := setupTestService(t)
record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "")
require.NoError(t, err)
_, err = svc.DisassociateRouteTable(&ec2.DisassociateRouteTableInput{
AssociationId: aws.String(record.Associations[0].AssociationId),
}, testAccountID)
assert.EqualError(t, err, awserrors.ErrorInvalidParameterValue)
}
func TestReplaceRouteTableAssociation(t *testing.T) {
svc := setupTestService(t)
rtb1ID := createTestRtb(t, svc)
rtb2ID := createTestRtb(t, svc)
assocOut, err := svc.AssociateRouteTable(&ec2.AssociateRouteTableInput{
RouteTableId: aws.String(rtb1ID),
SubnetId: aws.String("subnet-test1"),
}, testAccountID)
require.NoError(t, err)
replaceOut, err := svc.ReplaceRouteTableAssociation(&ec2.ReplaceRouteTableAssociationInput{
AssociationId: assocOut.AssociationId,
RouteTableId: aws.String(rtb2ID),
}, testAccountID)
require.NoError(t, err)
assert.NotEmpty(t, *replaceOut.NewAssociationId)
assert.NotEqual(t, *assocOut.AssociationId, *replaceOut.NewAssociationId)
// Verify old table has no associations
oldRecord, err := svc.getRouteTable(testAccountID, rtb1ID)
require.NoError(t, err)
assert.Empty(t, oldRecord.Associations)
// Verify new table has the association
newRecord, err := svc.getRouteTable(testAccountID, rtb2ID)
require.NoError(t, err)
require.Len(t, newRecord.Associations, 1)
assert.Equal(t, "subnet-test1", newRecord.Associations[0].SubnetId)
}
func TestCreateRouteTableForVPC_Main(t *testing.T) {
svc := setupTestService(t)
record, err := svc.CreateRouteTableForVPC("vpc-test1", "10.0.0.0/16", testAccountID, true, "")
require.NoError(t, err)
assert.True(t, record.IsMain)
assert.Len(t, record.Associations, 1)
assert.True(t, record.Associations[0].Main)
assert.Len(t, record.Routes, 1)
assert.Equal(t, "local", record.Routes[0].GatewayId)
}
@@ -0,0 +1,55 @@
package handlers_ec2_routetable
import (
"time"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
)
// NATSRouteTableService handles Route Table operations via NATS messaging
type NATSRouteTableService struct {
natsConn *nats.Conn
}
// NewNATSRouteTableService creates a new NATS-based Route Table service
func NewNATSRouteTableService(conn *nats.Conn) RouteTableService {
return &NATSRouteTableService{natsConn: conn}
}
func (s *NATSRouteTableService) CreateRouteTable(input *ec2.CreateRouteTableInput, accountID string) (*ec2.CreateRouteTableOutput, error) {
return utils.NATSRequest[ec2.CreateRouteTableOutput](s.natsConn, "ec2.CreateRouteTable", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) DeleteRouteTable(input *ec2.DeleteRouteTableInput, accountID string) (*ec2.DeleteRouteTableOutput, error) {
return utils.NATSRequest[ec2.DeleteRouteTableOutput](s.natsConn, "ec2.DeleteRouteTable", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) DescribeRouteTables(input *ec2.DescribeRouteTablesInput, accountID string) (*ec2.DescribeRouteTablesOutput, error) {
return utils.NATSRequest[ec2.DescribeRouteTablesOutput](s.natsConn, "ec2.DescribeRouteTables", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) CreateRoute(input *ec2.CreateRouteInput, accountID string) (*ec2.CreateRouteOutput, error) {
return utils.NATSRequest[ec2.CreateRouteOutput](s.natsConn, "ec2.CreateRoute", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) DeleteRoute(input *ec2.DeleteRouteInput, accountID string) (*ec2.DeleteRouteOutput, error) {
return utils.NATSRequest[ec2.DeleteRouteOutput](s.natsConn, "ec2.DeleteRoute", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) ReplaceRoute(input *ec2.ReplaceRouteInput, accountID string) (*ec2.ReplaceRouteOutput, error) {
return utils.NATSRequest[ec2.ReplaceRouteOutput](s.natsConn, "ec2.ReplaceRoute", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) AssociateRouteTable(input *ec2.AssociateRouteTableInput, accountID string) (*ec2.AssociateRouteTableOutput, error) {
return utils.NATSRequest[ec2.AssociateRouteTableOutput](s.natsConn, "ec2.AssociateRouteTable", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) DisassociateRouteTable(input *ec2.DisassociateRouteTableInput, accountID string) (*ec2.DisassociateRouteTableOutput, error) {
return utils.NATSRequest[ec2.DisassociateRouteTableOutput](s.natsConn, "ec2.DisassociateRouteTable", input, 30*time.Second, accountID)
}
func (s *NATSRouteTableService) ReplaceRouteTableAssociation(input *ec2.ReplaceRouteTableAssociationInput, accountID string) (*ec2.ReplaceRouteTableAssociationOutput, error) {
return utils.NATSRequest[ec2.ReplaceRouteTableAssociationOutput](s.natsConn, "ec2.ReplaceRouteTableAssociation", input, 30*time.Second, accountID)
}
+36
View File
@@ -0,0 +1,36 @@
package handlers_ec2_routetable
import "time"
const (
KVBucketRouteTables = "spinifex-vpc-route-tables"
KVBucketRouteTablesVersion = 1
)
// RouteTableRecord represents a stored Route Table
type RouteTableRecord struct {
RouteTableId string `json:"route_table_id"`
VpcId string `json:"vpc_id"`
AccountID string `json:"account_id"`
IsMain bool `json:"is_main"`
Routes []RouteRecord `json:"routes"`
Associations []AssociationRecord `json:"associations"`
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
}
// RouteRecord represents a single route in a route table
type RouteRecord struct {
DestinationCidrBlock string `json:"destination_cidr_block"`
GatewayId string `json:"gateway_id,omitempty"` // "local" or "igw-xxx"
NatGatewayId string `json:"nat_gateway_id,omitempty"` // "nat-xxx" (future)
State string `json:"state"` // "active", "blackhole"
Origin string `json:"origin"` // "CreateRouteTable", "CreateRoute"
}
// AssociationRecord represents a route table ↔ subnet association
type AssociationRecord struct {
AssociationId string `json:"association_id"`
SubnetId string `json:"subnet_id,omitempty"` // empty for main table VPC association
Main bool `json:"main"`
}
+76
View File
@@ -65,6 +65,7 @@ type VPCServiceImpl struct {
vniKV nats.KeyValue
eniKV nats.KeyValue
sgKV nats.KeyValue
rtbKV nats.KeyValue // route table bucket for auto-creating main route table
ipam *IPAM
}
@@ -115,6 +116,11 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
return nil, fmt.Errorf("write version to %s: %w", KVBucketSecurityGroups, err)
}
rtbKV, err := utils.GetOrCreateKVBucket(js, "spinifex-vpc-route-tables", 10)
if err != nil {
return nil, fmt.Errorf("failed to create KV bucket spinifex-vpc-route-tables: %w", err)
}
ipam, err := NewIPAM(js)
if err != nil {
return nil, fmt.Errorf("failed to initialize IPAM: %w", err)
@@ -135,6 +141,7 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer
vniKV: vniKV,
eniKV: eniKV,
sgKV: sgKV,
rtbKV: rtbKV,
ipam: ipam,
}, nil
}
@@ -234,6 +241,11 @@ func (s *VPCServiceImpl) CreateVpc(input *ec2.CreateVpcInput, accountID string)
// Publish vpc.create event for vpcd topology translation
s.publishVPCEvent("vpc.create", record.VpcId, record.CidrBlock, record.VNI)
// Auto-create main route table with local route (matches AWS behavior)
if s.rtbKV != nil {
s.createMainRouteTable(accountID, vpcID, record.CidrBlock)
}
return &ec2.CreateVpcOutput{
Vpc: s.vpcRecordToEC2(&record, accountID),
}, nil
@@ -823,6 +835,11 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string, bootstrap ...Bootstr
s.publishSubnetEvent("vpc.create-subnet", subnetID, vpcID, DefaultSubnetCidr)
// Create main route table with local route (written directly to KV to avoid circular import)
if s.rtbKV != nil {
s.createMainRouteTable(accountID, vpcID, DefaultVPCCidr)
}
slog.Info("Created default VPC and subnet",
"vpcId", vpcID,
"vpcCidr", DefaultVPCCidr,
@@ -839,6 +856,65 @@ func (s *VPCServiceImpl) EnsureDefaultVPC(accountID string, bootstrap ...Bootstr
}, nil
}
// createMainRouteTable writes a main route table record directly to the route table KV bucket.
// This avoids a circular import (routetable package imports vpc package for validation).
func (s *VPCServiceImpl) createMainRouteTable(accountID, vpcID, vpcCidr string) {
type rtbRecord struct {
RouteTableId string `json:"route_table_id"`
VpcId string `json:"vpc_id"`
AccountID string `json:"account_id"`
IsMain bool `json:"is_main"`
Routes []struct {
DestinationCidrBlock string `json:"destination_cidr_block"`
GatewayId string `json:"gateway_id,omitempty"`
State string `json:"state"`
Origin string `json:"origin"`
} `json:"routes"`
Associations []struct {
AssociationId string `json:"association_id"`
Main bool `json:"main"`
} `json:"associations"`
Tags map[string]string `json:"tags"`
CreatedAt time.Time `json:"created_at"`
}
rtbID := utils.GenerateResourceID("rtb")
record := rtbRecord{
RouteTableId: rtbID,
VpcId: vpcID,
AccountID: accountID,
IsMain: true,
Routes: []struct {
DestinationCidrBlock string `json:"destination_cidr_block"`
GatewayId string `json:"gateway_id,omitempty"`
State string `json:"state"`
Origin string `json:"origin"`
}{
{DestinationCidrBlock: vpcCidr, GatewayId: "local", State: "active", Origin: "CreateRouteTable"},
},
Associations: []struct {
AssociationId string `json:"association_id"`
Main bool `json:"main"`
}{
{AssociationId: utils.GenerateResourceID("rtbassoc"), Main: true},
},
Tags: make(map[string]string),
CreatedAt: time.Now(),
}
data, err := json.Marshal(record)
if err != nil {
slog.Error("Failed to marshal main route table", "vpcId", vpcID, "err", err)
return
}
if _, err := s.rtbKV.Put(utils.AccountKey(accountID, rtbID), data); err != nil {
slog.Error("Failed to store main route table", "vpcId", vpcID, "err", err)
return
}
slog.Info("Created main route table for VPC", "routeTableId", rtbID, "vpcId", vpcID, "accountID", accountID)
}
// GetDefaultSubnet returns the default subnet for RunInstances when no SubnetId is specified.
func (s *VPCServiceImpl) GetDefaultSubnet(accountID string) (*SubnetRecord, error) {
prefix := accountID + "."