From 0bb0fcc53273fea3477222a54482ebb00ba7609b Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Fri, 22 May 2026 13:53:01 +1000 Subject: [PATCH] feat(ipam): Purpose field on every IPAM record (Slice B) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of bead mulga-siv-125.3.3. Adds a Purpose enum (handlers/ec2/vpc/purpose.go) tagging every IP allocation with its semantic role (eni-primary, eni-public, eip, igw-lrp, natgw-external, …) so multi-VPC clusters can reclaim/audit by (owner, purpose). External IPAM (spinifex-external-ipam, v1→v2): - ExternalIPAllocation.Type → .Purpose; legacy values mapped via LegacyExternalTypeToPurpose (gateway→igw-lrp, auto_assign→eni-public, elastic_ip→eip). Unknown legacy values warn and become "unknown". - AllocateIP / AllocateFromPool / allocateFromPool param rename allocType→purpose. All call sites pass Purpose* constants. - migrate/002_external_ipam_purpose.go: idempotent CAS-retry migration with v1/v2-union record shape so partial migrations resume cleanly. Internal IPAM (spinifex-vpc-ipam, v1→v2): - IPAMRecord.Allocated []string → []IPEntry{IP, Purpose, OwnerID}. - AllocateIP gains purpose+ownerID args; ReleaseIP unchanged (still keyed on IP). - AllocatedIPs returns []IPEntry. - eni.go reorders eniId generation before AllocateIP so the owning resource ID lands on the new entry. - migrate/003_ipam_purpose.go: cross-bucket backfill that reads the ENI bucket via the new RunKVWithJetStream path. Orphaned IPs (no owning ENI) warn and become Purpose=unknown. Migrate framework: - KVContext gains an optional JetStream field; new Registry.RunKVWithJetStream method threads it through. RunKV signature unchanged — all other callers are unaffected. Tests: - ipam_test.go: AllocateIP signature updates; AllocatedIPs now asserts IP / Purpose / OwnerID on the returned entries. - external_ipam_test.go: replaces "auto_assign"/"elastic_ip" literals with PurposeENIPublic/PurposeEIP and asserts alloc.Purpose. - migrate/002 + 003 tests: schema-detect, idempotent rerun, no-JetStream fallback, empty bucket. --- spinifex/daemon/daemon_system_instance.go | 3 +- spinifex/handlers/ec2/eip/service_impl.go | 4 +- .../handlers/ec2/instance/service_impl.go | 3 +- spinifex/handlers/ec2/vpc/eni.go | 6 +- spinifex/handlers/ec2/vpc/external_ipam.go | 28 ++- .../handlers/ec2/vpc/external_ipam_test.go | 42 ++-- spinifex/handlers/ec2/vpc/ipam.go | 40 ++-- spinifex/handlers/ec2/vpc/ipam_test.go | 44 ++-- spinifex/handlers/ec2/vpc/purpose.go | 31 +++ spinifex/migrate/002_external_ipam_purpose.go | 134 +++++++++++ .../migrate/002_external_ipam_purpose_test.go | 103 ++++++++ spinifex/migrate/003_ipam_purpose.go | 225 ++++++++++++++++++ spinifex/migrate/003_ipam_purpose_test.go | 158 ++++++++++++ spinifex/migrate/migrate.go | 20 +- 14 files changed, 761 insertions(+), 80 deletions(-) create mode 100644 spinifex/handlers/ec2/vpc/purpose.go create mode 100644 spinifex/migrate/002_external_ipam_purpose.go create mode 100644 spinifex/migrate/002_external_ipam_purpose_test.go create mode 100644 spinifex/migrate/003_ipam_purpose.go create mode 100644 spinifex/migrate/003_ipam_purpose_test.go diff --git a/spinifex/daemon/daemon_system_instance.go b/spinifex/daemon/daemon_system_instance.go index 830cdf670..e11720e7c 100644 --- a/spinifex/daemon/daemon_system_instance.go +++ b/spinifex/daemon/daemon_system_instance.go @@ -10,6 +10,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" handlers_elbv2 "github.com/mulgadc/spinifex/spinifex/handlers/elbv2" "github.com/mulgadc/spinifex/spinifex/tags" "github.com/mulgadc/spinifex/spinifex/utils" @@ -209,7 +210,7 @@ func (d *Daemon) LaunchSystemInstance(input *handlers_elbv2.SystemInstanceInput) region = d.config.Region az = d.config.AZ } - allocatedIP, poolName, allocErr := d.externalIPAM.AllocateIP(region, az, "auto_assign", "", instance.ENIId, instance.ID) + allocatedIP, poolName, allocErr := d.externalIPAM.AllocateIP(region, az, handlers_ec2_vpc.PurposeENIPublic, "", instance.ENIId, instance.ID) if allocErr != nil { slog.Error("LaunchSystemInstance: failed to allocate public IP for internet-facing ALB", "instanceId", instance.ID, "err", allocErr) d.cleanupFailedSystemInstance(instance, instanceType) diff --git a/spinifex/handlers/ec2/eip/service_impl.go b/spinifex/handlers/ec2/eip/service_impl.go index e5dba061d..f18231e5f 100644 --- a/spinifex/handlers/ec2/eip/service_impl.go +++ b/spinifex/handlers/ec2/eip/service_impl.go @@ -73,7 +73,7 @@ func (s *EIPServiceImpl) AllocateAddress(input *ec2.AllocateAddressInput, accoun if input.PublicIpv4Pool != nil && *input.PublicIpv4Pool != "" { // Allocate from a specific named pool. poolName = *input.PublicIpv4Pool - publicIP, err = s.externalIPAM.AllocateFromPool(poolName, "elastic_ip", allocID, "", "") + publicIP, err = s.externalIPAM.AllocateFromPool(poolName, handlers_ec2_vpc.PurposeEIP, allocID, "", "") if err != nil { slog.Error("AllocateAddress: IPAM pool allocation failed", "pool", poolName, "err", err) return nil, errors.New(awserrors.ErrorInsufficientAddressCapacity) @@ -82,7 +82,7 @@ func (s *EIPServiceImpl) AllocateAddress(input *ec2.AllocateAddressInput, accoun // Allocate from the best pool matching region/AZ (empty strings = global fallback). region := "" az := "" - publicIP, poolName, err = s.externalIPAM.AllocateIP(region, az, "elastic_ip", allocID, "", "") + publicIP, poolName, err = s.externalIPAM.AllocateIP(region, az, handlers_ec2_vpc.PurposeEIP, allocID, "", "") if err != nil { slog.Error("AllocateAddress: IPAM allocation failed", "err", err) return nil, errors.New(awserrors.ErrorInsufficientAddressCapacity) diff --git a/spinifex/handlers/ec2/instance/service_impl.go b/spinifex/handlers/ec2/instance/service_impl.go index b5aefaebb..7f7bddb6f 100644 --- a/spinifex/handlers/ec2/instance/service_impl.go +++ b/spinifex/handlers/ec2/instance/service_impl.go @@ -22,6 +22,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" + handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" "github.com/mulgadc/spinifex/spinifex/instancetypes" "github.com/mulgadc/spinifex/spinifex/objectstore" spxtypes "github.com/mulgadc/spinifex/spinifex/types" @@ -689,7 +690,7 @@ func (s *InstanceServiceImpl) PrepareRunInstances(input *ec2.RunInstancesInput, if wantPublic { region := s.config.Region az := s.config.AZ - publicIP, poolName, allocErr := s.ipAllocator.AllocateIP(region, az, "auto_assign", "", *eni.NetworkInterfaceId, instance.ID) + publicIP, poolName, allocErr := s.ipAllocator.AllocateIP(region, az, handlers_ec2_vpc.PurposeENIPublic, "", *eni.NetworkInterfaceId, instance.ID) if allocErr != nil { slog.Warn("PrepareRunInstances: failed to allocate public IP", "instanceId", instance.ID, "err", allocErr) } else { diff --git a/spinifex/handlers/ec2/vpc/eni.go b/spinifex/handlers/ec2/vpc/eni.go index 30b28f72a..ae2d3b728 100644 --- a/spinifex/handlers/ec2/vpc/eni.go +++ b/spinifex/handlers/ec2/vpc/eni.go @@ -78,21 +78,21 @@ func (s *VPCServiceImpl) CreateNetworkInterface(input *ec2.CreateNetworkInterfac return nil, err } + eniId := utils.GenerateResourceID("eni") + // Allocate IP from subnet var privateIP string if input.PrivateIpAddress != nil && *input.PrivateIpAddress != "" { // TODO: validate the requested IP is in the subnet range and not already allocated privateIP = *input.PrivateIpAddress } else { - ip, err := s.ipam.AllocateIP(subnetId, subnet.CidrBlock) + ip, err := s.ipam.AllocateIP(subnetId, subnet.CidrBlock, PurposeENIPrimary, eniId) if err != nil { return nil, errors.New(awserrors.ErrorServerInternal) } privateIP = ip } - eniId := utils.GenerateResourceID("eni") - // Generate a deterministic MAC address macAddr := generateENIMac(eniId) diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index 96cb51094..32f092066 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam.go +++ b/spinifex/handlers/ec2/vpc/external_ipam.go @@ -15,12 +15,13 @@ import ( const ( KVBucketExternalIPAM = "spinifex-external-ipam" - KVBucketExternalIPAMVersion = 1 + KVBucketExternalIPAMVersion = 2 ) -// ExternalIPAllocation describes how an external IP is being used. +// ExternalIPAllocation describes how an external IP is being used. Purpose +// values come from the Purpose enum in purpose.go. type ExternalIPAllocation struct { - Type string `json:"type"` // "gateway", "auto_assign", "elastic_ip" + Purpose string `json:"purpose"` // Purpose enum: igw-lrp, eni-public, eip, natgw-external AllocationID string `json:"allocation_id,omitempty"` // For elastic IPs Association string `json:"association,omitempty"` // ENI ID for elastic IPs ENIId string `json:"eni_id,omitempty"` // ENI owning this IP @@ -156,7 +157,7 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { GwLrpRangeStart: pool.GwLrpRangeStart, GwLrpRangeEnd: pool.GwLrpRangeEnd, Allocated: map[string]ExternalIPAllocation{ - gwIP: {Type: "gateway", Note: "OVN router SNAT address"}, + gwIP: {Purpose: PurposeIGWLRP, Note: "OVN router SNAT address"}, }, } @@ -179,13 +180,14 @@ func (m *ExternalIPAM) initPool(pool ExternalPoolConfig) error { // AllocateIP allocates the next available external IP from the best pool // matching the given region/AZ. Returns the allocated IP and pool name. -// allocID is the EIP allocation ID (for elastic IPs); pass "" for other types. -func (m *ExternalIPAM) AllocateIP(region, az, allocType, allocID, eniID, instanceID string) (string, string, error) { +// purpose is one of the Purpose* constants in purpose.go; allocID is the +// EIP allocation ID (for PurposeEIP) and "" otherwise. +func (m *ExternalIPAM) AllocateIP(region, az, purpose, allocID, eniID, instanceID string) (string, string, error) { pool := m.findPool(region, az) if pool == nil { return "", "", fmt.Errorf("InsufficientAddressCapacity: no external pool available for region=%q az=%q", region, az) } - ip, err := m.allocateFromPool(pool.Name, allocType, allocID, eniID, instanceID) + ip, err := m.allocateFromPool(pool.Name, purpose, allocID, eniID, instanceID) if err != nil { return "", "", err } @@ -193,11 +195,11 @@ func (m *ExternalIPAM) AllocateIP(region, az, allocType, allocID, eniID, instanc } // AllocateFromPool allocates an IP from a specific named pool. -func (m *ExternalIPAM) AllocateFromPool(poolName, allocType, allocID, eniID, instanceID string) (string, error) { - return m.allocateFromPool(poolName, allocType, allocID, eniID, instanceID) +func (m *ExternalIPAM) AllocateFromPool(poolName, purpose, allocID, eniID, instanceID string) (string, error) { + return m.allocateFromPool(poolName, purpose, allocID, eniID, instanceID) } -func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, instanceID string) (string, error) { +func (m *ExternalIPAM) allocateFromPool(poolName, purpose, allocID, eniID, instanceID string) (string, error) { for attempt := range 5 { record, revision, err := m.getRecord(poolName) if err != nil { @@ -210,7 +212,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins } record.Allocated[ip] = ExternalIPAllocation{ - Type: allocType, + Purpose: purpose, AllocationID: allocID, ENIId: eniID, InstanceId: instanceID, @@ -226,7 +228,7 @@ func (m *ExternalIPAM) allocateFromPool(poolName, allocType, allocID, eniID, ins continue } - slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "type", allocType) + slog.Info("external IPAM allocated IP", "pool", poolName, "ip", ip, "purpose", purpose) return ip, nil } @@ -245,7 +247,7 @@ func (m *ExternalIPAM) ReleaseIP(poolName, ip string) error { if !ok { return fmt.Errorf("IP %s not allocated in pool %s", ip, poolName) } - if alloc.Type == "gateway" { + if alloc.Purpose == PurposeIGWLRP { return fmt.Errorf("cannot release gateway IP %s in pool %s", ip, poolName) } diff --git a/spinifex/handlers/ec2/vpc/external_ipam_test.go b/spinifex/handlers/ec2/vpc/external_ipam_test.go index 71dad2eaf..cd03fa4d1 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam_test.go +++ b/spinifex/handlers/ec2/vpc/external_ipam_test.go @@ -33,16 +33,16 @@ func TestExternalIPAM_AllocateSequential(t *testing.T) { ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool}) // .150 is reserved for gateway, so first allocable is .151 - ip1, poolName, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1") + ip1, poolName, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "192.168.1.151", ip1) assert.Equal(t, "wan", poolName) - ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2") + ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2") require.NoError(t, err) assert.Equal(t, "192.168.1.152", ip2) - ip3, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3") + ip3, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3") require.NoError(t, err) assert.Equal(t, "192.168.1.153", ip3) } @@ -57,7 +57,7 @@ func TestExternalIPAM_GatewayIPReserved(t *testing.T) { // Gateway IP (.150) should be reserved alloc, ok := record.Allocated["192.168.1.150"] assert.True(t, ok, "gateway IP should be in allocated map") - assert.Equal(t, "gateway", alloc.Type) + assert.Equal(t, PurposeIGWLRP, alloc.Purpose) // Cannot release gateway IP err = ipam.ReleaseIP("wan", "192.168.1.150") @@ -79,7 +79,7 @@ func TestExternalIPAM_ExplicitGatewayIP(t *testing.T) { assert.False(t, ok) // First allocable is .150 since .155 is the gateway - ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1") + ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "192.168.1.150", ip1) } @@ -88,11 +88,11 @@ func TestExternalIPAM_Release(t *testing.T) { pool := testPool() ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool}) - ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1") + ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "192.168.1.151", ip1) - ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2") + ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2") require.NoError(t, err) assert.Equal(t, "192.168.1.152", ip2) @@ -101,7 +101,7 @@ func TestExternalIPAM_Release(t *testing.T) { require.NoError(t, err) // Next allocation reuses released IP - ip3, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3") + ip3, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3") require.NoError(t, err) assert.Equal(t, "192.168.1.151", ip3) } @@ -117,16 +117,16 @@ func TestExternalIPAM_Exhaustion(t *testing.T) { ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool}) // .1 reserved for gateway, .2 and .3 allocable - ip1, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1") + ip1, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "10.0.0.2", ip1) - ip2, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-2", "i-2") + ip2, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-2", "i-2") require.NoError(t, err) assert.Equal(t, "10.0.0.3", ip2) // Pool exhausted - _, _, err = ipam.AllocateIP("", "", "auto_assign", "", "eni-3", "i-3") + _, _, err = ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-3", "i-3") assert.ErrorContains(t, err, "InsufficientAddressCapacity") } @@ -143,7 +143,7 @@ func TestExternalIPAM_CASConflict(t *testing.T) { wg.Add(1) go func(idx int) { defer wg.Done() - ip, _, err := ipam.AllocateIP("", "", "auto_assign", "", "eni-"+itoa(idx), "i-"+itoa(idx)) + ip, _, err := ipam.AllocateIP("", "", PurposeENIPublic, "", "eni-"+itoa(idx), "i-"+itoa(idx)) results[idx] = ip errs[idx] = err }(i) @@ -187,13 +187,13 @@ func TestExternalIPAM_MultiPool(t *testing.T) { ipam := setupTestExternalIPAM(t, pools) // Allocate from US pool - ip1, poolName1, err := ipam.AllocateIP("us-east-1", "us-east-1a", "auto_assign", "", "eni-1", "i-1") + ip1, poolName1, err := ipam.AllocateIP("us-east-1", "us-east-1a", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "203.0.113.3", ip1) // .2 reserved for gateway assert.Equal(t, "us-east", poolName1) // Allocate from EU pool - ip2, poolName2, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-2", "i-2") + ip2, poolName2, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-2", "i-2") require.NoError(t, err) assert.Equal(t, "198.51.100.3", ip2) assert.Equal(t, "eu-west", poolName2) @@ -229,19 +229,19 @@ func TestExternalIPAM_PoolFallback(t *testing.T) { ipam := setupTestExternalIPAM(t, pools) // AZ match: us-east-1a → az-pool - ip1, pool1, err := ipam.AllocateIP("us-east-1", "us-east-1a", "auto_assign", "", "eni-1", "i-1") + ip1, pool1, err := ipam.AllocateIP("us-east-1", "us-east-1a", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "az-pool", pool1) assert.Equal(t, "10.0.0.2", ip1) // .1 reserved for gw // Region match (different AZ): us-east-1b → region-pool - ip2, pool2, err := ipam.AllocateIP("us-east-1", "us-east-1b", "auto_assign", "", "eni-2", "i-2") + ip2, pool2, err := ipam.AllocateIP("us-east-1", "us-east-1b", PurposeENIPublic, "", "eni-2", "i-2") require.NoError(t, err) assert.Equal(t, "region-pool", pool2) assert.Equal(t, "10.0.1.2", ip2) // No match: eu-west-1 → global-pool - ip3, pool3, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-3", "i-3") + ip3, pool3, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-3", "i-3") require.NoError(t, err) assert.Equal(t, "global-pool", pool3) assert.Equal(t, "10.0.2.2", ip3) @@ -267,7 +267,7 @@ func TestExternalIPAM_SpecificPool(t *testing.T) { ipam := setupTestExternalIPAM(t, pools) // Allocate specifically from pool-b - ip, err := ipam.AllocateFromPool("pool-b", "elastic_ip", "eipalloc-test-b", "", "") + ip, err := ipam.AllocateFromPool("pool-b", PurposeEIP, "eipalloc-test-b", "", "") require.NoError(t, err) assert.Equal(t, "10.0.1.2", ip) // .1 reserved for gw } @@ -341,7 +341,7 @@ func TestExternalIPAM_InitFromConfig(t *testing.T) { require.NoError(t, err) // Allocate an IP - ip, _, err := ipam1.AllocateIP("", "", "auto_assign", "", "eni-1", "i-1") + ip, _, err := ipam1.AllocateIP("", "", PurposeENIPublic, "", "eni-1", "i-1") require.NoError(t, err) assert.Equal(t, "192.168.1.151", ip) @@ -376,7 +376,7 @@ func TestExternalIPAM_NoPoolAvailable(t *testing.T) { ipam := setupTestExternalIPAM(t, []ExternalPoolConfig{pool}) // No pool matches eu-west - _, _, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", "auto_assign", "", "eni-1", "i-1") + _, _, err := ipam.AllocateIP("eu-west-1", "eu-west-1a", PurposeENIPublic, "", "eni-1", "i-1") assert.ErrorContains(t, err, "InsufficientAddressCapacity") } @@ -388,7 +388,7 @@ func TestExternalIPAM_FindPoolByName_NotFound(t *testing.T) { // has no KV record, so getRecord fails. kv := ipam.kv ipam2 := NewExternalIPAMWithKV(kv, []ExternalPoolConfig{pool}) - _, err := ipam2.AllocateFromPool("nonexistent", "auto_assign", "", "eni-1", "i-1") + _, err := ipam2.AllocateFromPool("nonexistent", PurposeENIPublic, "", "eni-1", "i-1") assert.Error(t, err) } diff --git a/spinifex/handlers/ec2/vpc/ipam.go b/spinifex/handlers/ec2/vpc/ipam.go index 92a40208a..b79d27f70 100644 --- a/spinifex/handlers/ec2/vpc/ipam.go +++ b/spinifex/handlers/ec2/vpc/ipam.go @@ -15,14 +15,22 @@ import ( const ( KVBucketIPAM = "spinifex-vpc-ipam" - KVBucketIPAMVersion = 1 + KVBucketIPAMVersion = 2 ) +// IPEntry tags one IP allocation with its Purpose + owner (eni-, eipalloc-, +// etc) so multi-VPC clusters can reclaim/audit by (owner, purpose). +type IPEntry struct { + IP string `json:"ip"` + Purpose string `json:"purpose"` // one of the Purpose* constants + OwnerID string `json:"owner_id,omitempty"` // ENI / EIP / NATGW / IGW resource ID +} + // IPAMRecord tracks allocated IPs for a subnet. type IPAMRecord struct { - SubnetId string `json:"subnet_id"` - CidrBlock string `json:"cidr_block"` - Allocated []string `json:"allocated"` // List of allocated IPs + SubnetId string `json:"subnet_id"` + CidrBlock string `json:"cidr_block"` + Allocated []IPEntry `json:"allocated"` } // IPAM manages IP address allocation for VPC subnets using NATS KV with CAS. @@ -36,7 +44,7 @@ func NewIPAM(js nats.JetStreamContext) (*IPAM, error) { if err != nil { return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err) } - if err := migrate.DefaultRegistry.RunKV(KVBucketIPAM, kv, KVBucketIPAMVersion); err != nil { + if err := migrate.DefaultRegistry.RunKVWithJetStream(KVBucketIPAM, kv, js, KVBucketIPAMVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketIPAM, err) } return &IPAM{kv: kv}, nil @@ -50,8 +58,10 @@ func NewIPAMWithKV(kv nats.KeyValue) *IPAM { // AllocateIP allocates an IP address from the given subnet. // It reserves the first 4 and last IP per AWS convention: // .0=network, .1=gateway, .2=DNS, .3=reserved, .255=broadcast (for /24). -// Uses CAS for conflict-free allocation across nodes. -func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) { +// Uses CAS for conflict-free allocation across nodes. purpose is one of the +// Purpose* constants in purpose.go; ownerID is the ENI / instance / EIP +// resource ID that owns this allocation. +func (m *IPAM) AllocateIP(subnetId, cidrBlock, purpose, ownerID string) (string, error) { for attempt := range 5 { record, revision, err := m.getRecord(subnetId) if err != nil && !errors.Is(err, nats.ErrKeyNotFound) { @@ -71,7 +81,7 @@ func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) { return "", err } - record.Allocated = append(record.Allocated, ip) + record.Allocated = append(record.Allocated, IPEntry{IP: ip, Purpose: purpose, OwnerID: ownerID}) // CAS write data, err := json.Marshal(record) @@ -90,7 +100,7 @@ func (m *IPAM) AllocateIP(subnetId, cidrBlock string) (string, error) { continue // CAS conflict, retry } - slog.Info("IPAM allocated IP", "subnet", subnetId, "ip", ip) + slog.Info("IPAM allocated IP", "subnet", subnetId, "ip", ip, "purpose", purpose, "owner", ownerID) return ip, nil } @@ -109,8 +119,8 @@ func (m *IPAM) ReleaseIP(subnetId, ip string) error { } found := false - for i, allocated := range record.Allocated { - if allocated == ip { + for i, entry := range record.Allocated { + if entry.IP == ip { record.Allocated = append(record.Allocated[:i], record.Allocated[i+1:]...) found = true break @@ -137,8 +147,8 @@ func (m *IPAM) ReleaseIP(subnetId, ip string) error { return fmt.Errorf("IPAM release failed after CAS retries for subnet %s", subnetId) } -// AllocatedIPs returns the list of allocated IPs for a subnet. -func (m *IPAM) AllocatedIPs(subnetId string) ([]string, error) { +// AllocatedIPs returns the list of allocated IP entries for a subnet. +func (m *IPAM) AllocatedIPs(subnetId string) ([]IPEntry, error) { record, _, err := m.getRecord(subnetId) if err != nil { if errors.Is(err, nats.ErrKeyNotFound) { @@ -177,8 +187,8 @@ func (m *IPAM) nextAvailableIP(record *IPAMRecord) (string, error) { } allocated := make(map[string]bool, len(record.Allocated)) - for _, ip := range record.Allocated { - allocated[ip] = true + for _, entry := range record.Allocated { + allocated[entry.IP] = true } ones, bits := ipNet.Mask.Size() diff --git a/spinifex/handlers/ec2/vpc/ipam_test.go b/spinifex/handlers/ec2/vpc/ipam_test.go index 7006e0e23..ac7473d4c 100644 --- a/spinifex/handlers/ec2/vpc/ipam_test.go +++ b/spinifex/handlers/ec2/vpc/ipam_test.go @@ -22,7 +22,7 @@ func setupTestIPAM(t *testing.T) *IPAM { func TestIPAM_AllocateFirst(t *testing.T) { ipam := setupTestIPAM(t) - ip, err := ipam.AllocateIP("subnet-1", "10.0.1.0/24") + ip, err := ipam.AllocateIP("subnet-1", "10.0.1.0/24", PurposeENIPrimary, "eni-1") require.NoError(t, err) // First allocable IP is .4 (skip .0 network, .1 gateway, .2 DNS, .3 reserved) assert.Equal(t, "10.0.1.4", ip) @@ -31,15 +31,15 @@ func TestIPAM_AllocateFirst(t *testing.T) { func TestIPAM_AllocateSequential(t *testing.T) { ipam := setupTestIPAM(t) - ip1, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24") + ip1, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq") require.NoError(t, err) assert.Equal(t, "10.0.2.4", ip1) - ip2, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24") + ip2, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq") require.NoError(t, err) assert.Equal(t, "10.0.2.5", ip2) - ip3, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24") + ip3, err := ipam.AllocateIP("subnet-seq", "10.0.2.0/24", PurposeENIPrimary, "eni-seq") require.NoError(t, err) assert.Equal(t, "10.0.2.6", ip3) } @@ -47,11 +47,11 @@ func TestIPAM_AllocateSequential(t *testing.T) { func TestIPAM_Release(t *testing.T) { ipam := setupTestIPAM(t) - ip1, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24") + ip1, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel") require.NoError(t, err) assert.Equal(t, "10.0.3.4", ip1) - ip2, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24") + ip2, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel") require.NoError(t, err) assert.Equal(t, "10.0.3.5", ip2) @@ -60,7 +60,7 @@ func TestIPAM_Release(t *testing.T) { require.NoError(t, err) // Next allocation should reuse the released IP - ip3, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24") + ip3, err := ipam.AllocateIP("subnet-rel", "10.0.3.0/24", PurposeENIPrimary, "eni-rel") require.NoError(t, err) assert.Equal(t, "10.0.3.4", ip3) } @@ -68,7 +68,7 @@ func TestIPAM_Release(t *testing.T) { func TestIPAM_ReleaseNotAllocated(t *testing.T) { ipam := setupTestIPAM(t) - _, err := ipam.AllocateIP("subnet-rna", "10.0.4.0/24") + _, err := ipam.AllocateIP("subnet-rna", "10.0.4.0/24", PurposeENIPrimary, "eni-rna") require.NoError(t, err) err = ipam.ReleaseIP("subnet-rna", "10.0.4.99") @@ -91,7 +91,7 @@ func TestIPAM_Exhaustion(t *testing.T) { var allocated []string for i := range 11 { - ip, err := ipam.AllocateIP(subnetId, cidr) + ip, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test") require.NoError(t, err, "allocation %d should succeed", i) allocated = append(allocated, ip) } @@ -101,7 +101,7 @@ func TestIPAM_Exhaustion(t *testing.T) { assert.Equal(t, "10.0.5.14", allocated[len(allocated)-1]) // Next allocation should fail — subnet exhausted - _, err := ipam.AllocateIP(subnetId, cidr) + _, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test") assert.ErrorContains(t, err, "exhausted") } @@ -113,7 +113,7 @@ func TestIPAM_ReservedIPs(t *testing.T) { // First 4 allocations should be .4, .5, .6, .7 (skipping .0-.3) for i := 4; i <= 7; i++ { - ip, err := ipam.AllocateIP(subnetId, cidr) + ip, err := ipam.AllocateIP(subnetId, cidr, PurposeENIPrimary, "eni-test") require.NoError(t, err) expected := "10.0.6." + itoa(i) assert.Equal(t, expected, ip) @@ -129,29 +129,31 @@ func TestIPAM_AllocatedIPs(t *testing.T) { assert.Nil(t, ips) // Allocate some IPs - _, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24") - _, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24") + _, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24", PurposeENIPrimary, "eni-list") + _, _ = ipam.AllocateIP("subnet-list", "10.0.7.0/24", PurposeENIPrimary, "eni-list") ips, err = ipam.AllocatedIPs("subnet-list") require.NoError(t, err) assert.Len(t, ips, 2) - assert.Equal(t, "10.0.7.4", ips[0]) - assert.Equal(t, "10.0.7.5", ips[1]) + assert.Equal(t, "10.0.7.4", ips[0].IP) + assert.Equal(t, PurposeENIPrimary, ips[0].Purpose) + assert.Equal(t, "eni-list", ips[0].OwnerID) + assert.Equal(t, "10.0.7.5", ips[1].IP) } func TestIPAM_MultipleSubnets(t *testing.T) { ipam := setupTestIPAM(t) - ip1, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24") + ip1, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24", PurposeENIPrimary, "eni-a") require.NoError(t, err) assert.Equal(t, "10.0.10.4", ip1) - ip2, err := ipam.AllocateIP("subnet-b", "10.0.20.0/24") + ip2, err := ipam.AllocateIP("subnet-b", "10.0.20.0/24", PurposeENIPrimary, "eni-b") require.NoError(t, err) assert.Equal(t, "10.0.20.4", ip2) // Each subnet tracks independently - ip3, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24") + ip3, err := ipam.AllocateIP("subnet-a", "10.0.10.0/24", PurposeENIPrimary, "eni-a") require.NoError(t, err) assert.Equal(t, "10.0.10.5", ip3) } @@ -160,7 +162,7 @@ func TestIPAM_LargerSubnet(t *testing.T) { ipam := setupTestIPAM(t) // /20 subnet — first allocable is still .4 - ip, err := ipam.AllocateIP("subnet-big", "172.16.0.0/20") + ip, err := ipam.AllocateIP("subnet-big", "172.16.0.0/20", PurposeENIPrimary, "eni-big") require.NoError(t, err) assert.Equal(t, "172.16.0.4", ip) } @@ -177,14 +179,14 @@ func TestIPAM_NewIPAMWithKV(t *testing.T) { require.NotNil(t, ipam) // Should work the same as regular IPAM - ip, err := ipam.AllocateIP("subnet-kv", "10.0.1.0/24") + ip, err := ipam.AllocateIP("subnet-kv", "10.0.1.0/24", PurposeENIPrimary, "eni-kv") require.NoError(t, err) assert.Equal(t, "10.0.1.4", ip) } func TestIPAM_InvalidCIDR(t *testing.T) { ipam := setupTestIPAM(t) - _, err := ipam.AllocateIP("subnet-bad", "not-a-cidr") + _, err := ipam.AllocateIP("subnet-bad", "not-a-cidr", PurposeENIPrimary, "eni-bad") assert.Error(t, err) assert.Contains(t, err.Error(), "parse CIDR") } diff --git a/spinifex/handlers/ec2/vpc/purpose.go b/spinifex/handlers/ec2/vpc/purpose.go new file mode 100644 index 000000000..88ed516dd --- /dev/null +++ b/spinifex/handlers/ec2/vpc/purpose.go @@ -0,0 +1,31 @@ +package handlers_ec2_vpc + +// Purpose tags every IP allocation with its semantic role so multi-VPC +// clusters can reclaim / audit by (owner, purpose). The constants match the +// JSON values written to KV records (mulga-siv-125.3.3 Slice B). +const ( + PurposeENIPrimary = "eni-primary" + PurposeENISecondary = "eni-secondary" + PurposeENIPublic = "eni-public" // auto-assigned public IP + PurposeEIP = "eip" + PurposeNATGWExternal = "natgw-external" + PurposeIGWLRP = "igw-lrp" + // Future-reserved values; defined now so call sites and migrations can + // reference them without a follow-up rename. + PurposeEdgeLRExternal = "edge-lr-external" // 125.3.6 + PurposeNodeMgmt = "node-mgmt" // currently .env + PurposeSubnetRouter = "subnet-router" // .1 + PurposeSubnetDNS = "subnet-dns" // .2 + PurposeSubnetReserved = "subnet-reserved" // .0, .3, .last + // PurposeUnknown tags records the migration could not classify (e.g. + // orphan IPs with no owning ENIRecord). Surfaces in logs for triage. + PurposeUnknown = "unknown" +) + +// LegacyExternalTypeToPurpose maps the old ExternalIPAllocation.Type values +// onto the new Purpose enum. Used by the v1→v2 external-IPAM migration. +var LegacyExternalTypeToPurpose = map[string]string{ + "gateway": PurposeIGWLRP, + "auto_assign": PurposeENIPublic, + "elastic_ip": PurposeEIP, +} diff --git a/spinifex/migrate/002_external_ipam_purpose.go b/spinifex/migrate/002_external_ipam_purpose.go new file mode 100644 index 000000000..b8c8488b8 --- /dev/null +++ b/spinifex/migrate/002_external_ipam_purpose.go @@ -0,0 +1,134 @@ +package migrate + +import ( + "encoding/json" + "errors" + "fmt" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// extIPAMBucket mirrors handlers/ec2/vpc.KVBucketExternalIPAM. Duplicated to +// keep the migrate package free of any handler dependency; if the bucket name +// changes, both constants must move together (mulga-siv-125.3.3 Slice B). +const extIPAMBucket = "spinifex-external-ipam" + +const extIPAMMaxRetries = 3 + +// extLegacyTypeToPurpose maps the v1 ExternalIPAllocation.Type values onto +// the v2 Purpose enum. Frozen copy of vpc.LegacyExternalTypeToPurpose to +// keep this migration self-contained. +var extLegacyTypeToPurpose = map[string]string{ + "gateway": "igw-lrp", + "auto_assign": "eni-public", + "elastic_ip": "eip", +} + +// extIPAllocV1V2 is a union of the v1 and v2 ExternalIPAllocation schemas. +// The migration reads the legacy Type field, writes the new Purpose field, +// and clears Type so the on-disk record matches the v2 struct exactly. +type extIPAllocV1V2 struct { + Type string `json:"type,omitempty"` + Purpose string `json:"purpose,omitempty"` + AllocationID string `json:"allocation_id,omitempty"` + Association string `json:"association,omitempty"` + ENIId string `json:"eni_id,omitempty"` + InstanceId string `json:"instance_id,omitempty"` + Note string `json:"note,omitempty"` +} + +type extIPAMRecordV1V2 struct { + PoolName string `json:"pool_name"` + RangeStart string `json:"range_start"` + RangeEnd string `json:"range_end"` + Gateway string `json:"gateway"` + GatewayIP string `json:"gateway_ip"` + PrefixLen int `json:"prefix_len"` + Region string `json:"region,omitempty"` + AZ string `json:"az,omitempty"` + GwLrpRangeStart string `json:"gw_lrp_range_start,omitempty"` + GwLrpRangeEnd string `json:"gw_lrp_range_end,omitempty"` + Allocated map[string]extIPAllocV1V2 `json:"allocated"` +} + +func init() { + DefaultRegistry.RegisterKV(extIPAMBucket, KVMigration{ + FromVersion: 1, + ToVersion: 2, + Description: "rename ExternalIPAllocation.Type → .Purpose with enum mapping", + Run: func(ctx KVContext) error { + keys, err := ctx.KV.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil + } + return fmt.Errorf("list keys: %w", err) + } + + for _, key := range keys { + if key == utils.VersionKey { + continue + } + if err := renameExternalIPAMType(ctx, key); err != nil { + return err + } + } + return nil + }, + }) +} + +func renameExternalIPAMType(ctx KVContext, key string) error { + var lastErr error + for attempt := range extIPAMMaxRetries { + entry, err := ctx.KV.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil + } + return fmt.Errorf("get %s: %w", key, err) + } + + var record extIPAMRecordV1V2 + if err := json.Unmarshal(entry.Value(), &record); err != nil { + return fmt.Errorf("unmarshal %s: %w", key, err) + } + + changed := false + for ip, alloc := range record.Allocated { + if alloc.Purpose != "" { + continue + } + purpose := extLegacyTypeToPurpose[alloc.Type] + if purpose == "" { + ctx.Logger.Warn("external IPAM v1→v2 migration: unknown legacy type, tagging unknown", + "pool", key, "ip", ip, "legacy_type", alloc.Type) + purpose = "unknown" + } + alloc.Purpose = purpose + alloc.Type = "" + record.Allocated[ip] = alloc + changed = true + } + + if !changed { + return nil + } + + data, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("marshal %s: %w", key, err) + } + if _, err := ctx.KV.Update(key, data, entry.Revision()); err != nil { + if errors.Is(err, nats.ErrKeyExists) { + ctx.Logger.Warn("external IPAM migration CAS conflict, retrying", "key", key, "attempt", attempt+1) + lastErr = err + continue + } + return fmt.Errorf("update %s: %w", key, err) + } + return nil + } + return fmt.Errorf("rename Type→Purpose for %s: exceeded %d CAS retries: %w", key, extIPAMMaxRetries, lastErr) +} diff --git a/spinifex/migrate/002_external_ipam_purpose_test.go b/spinifex/migrate/002_external_ipam_purpose_test.go new file mode 100644 index 000000000..2319807f0 --- /dev/null +++ b/spinifex/migrate/002_external_ipam_purpose_test.go @@ -0,0 +1,103 @@ +package migrate + +import ( + "encoding/json" + "log/slog" + "testing" + + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func extIPAMMigrationKV(t *testing.T) nats.KeyValue { + t.Helper() + _, nc := startTestNATS(t) + return createTestBucket(t, nc, "ext-ipam-migration-test") +} + +func runExtIPAMMigration(t *testing.T, kv nats.KeyValue) { + t.Helper() + for _, m := range DefaultRegistry.kvMigrations[extIPAMBucket] { + if m.FromVersion == 1 && m.ToVersion == 2 { + require.NoError(t, m.Run(KVContext{KV: kv, Logger: slog.Default()})) + return + } + } + t.Fatalf("v1→v2 migration not registered for %s", extIPAMBucket) +} + +func TestExtIPAMMigration_RenamesType(t *testing.T) { + kv := extIPAMMigrationKV(t) + + record := extIPAMRecordV1V2{ + PoolName: "wan", + RangeStart: "192.168.1.10", + RangeEnd: "192.168.1.250", + Gateway: "192.168.1.1", + PrefixLen: 24, + Allocated: map[string]extIPAllocV1V2{ + "192.168.1.10": {Type: "gateway", Note: "OVN router SNAT address"}, + "192.168.1.11": {Type: "auto_assign", ENIId: "eni-1", InstanceId: "i-1"}, + "192.168.1.12": {Type: "elastic_ip", AllocationID: "eipalloc-1"}, + "192.168.1.13": {Type: "junk-legacy", ENIId: "eni-zombie"}, + }, + } + data, err := json.Marshal(record) + require.NoError(t, err) + _, err = kv.Put("wan", data) + require.NoError(t, err) + + runExtIPAMMigration(t, kv) + + entry, err := kv.Get("wan") + require.NoError(t, err) + var got extIPAMRecordV1V2 + require.NoError(t, json.Unmarshal(entry.Value(), &got)) + + assert.Equal(t, "igw-lrp", got.Allocated["192.168.1.10"].Purpose) + assert.Equal(t, "eni-public", got.Allocated["192.168.1.11"].Purpose) + assert.Equal(t, "eip", got.Allocated["192.168.1.12"].Purpose) + assert.Equal(t, "unknown", got.Allocated["192.168.1.13"].Purpose, "unrecognized legacy types tagged unknown") + + for ip, alloc := range got.Allocated { + assert.Empty(t, alloc.Type, "legacy Type field cleared for %s", ip) + } + + assert.Equal(t, "eni-1", got.Allocated["192.168.1.11"].ENIId, "non-Type fields preserved") + assert.Equal(t, "eipalloc-1", got.Allocated["192.168.1.12"].AllocationID) +} + +func TestExtIPAMMigration_IdempotentOnAlreadyMigrated(t *testing.T) { + kv := extIPAMMigrationKV(t) + + record := extIPAMRecordV1V2{ + PoolName: "wan", + Allocated: map[string]extIPAllocV1V2{ + "192.168.1.10": {Purpose: "igw-lrp"}, + "192.168.1.11": {Purpose: "eni-public", ENIId: "eni-1"}, + }, + } + data, err := json.Marshal(record) + require.NoError(t, err) + _, err = kv.Put("wan", data) + require.NoError(t, err) + rev := getKVRevision(t, kv, "wan") + + runExtIPAMMigration(t, kv) + + newRev := getKVRevision(t, kv, "wan") + assert.Equal(t, rev, newRev, "no write when all records already have Purpose") +} + +func TestExtIPAMMigration_EmptyBucket(t *testing.T) { + kv := extIPAMMigrationKV(t) + runExtIPAMMigration(t, kv) +} + +func getKVRevision(t *testing.T, kv nats.KeyValue, key string) uint64 { + t.Helper() + entry, err := kv.Get(key) + require.NoError(t, err) + return entry.Revision() +} diff --git a/spinifex/migrate/003_ipam_purpose.go b/spinifex/migrate/003_ipam_purpose.go new file mode 100644 index 000000000..e59f39bd1 --- /dev/null +++ b/spinifex/migrate/003_ipam_purpose.go @@ -0,0 +1,225 @@ +package migrate + +import ( + "encoding/json" + "errors" + "fmt" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// ipamBucket / eniBucket mirror handlers/ec2/vpc.KVBucketIPAM and +// .KVBucketENIs. Duplicated here so the migrate package stays handler-free +// (mulga-siv-125.3.3 Slice B). +const ( + ipamBucket = "spinifex-vpc-ipam" + eniBucket = "spinifex-vpc-enis" +) + +const ipamMigrationMaxRetries = 3 + +// ipEntryV2 mirrors handlers/ec2/vpc.IPEntry (frozen snapshot). +type ipEntryV2 struct { + IP string `json:"ip"` + Purpose string `json:"purpose"` + OwnerID string `json:"owner_id,omitempty"` +} + +// ipamRecordV1V2 is a union of v1 ([]string) and v2 ([]IPEntry) schemas. +// Allocated is left as raw JSON so we can detect which schema is on-disk +// and re-marshal as IPEntry. +type ipamRecordV1V2 struct { + SubnetId string `json:"subnet_id"` + CidrBlock string `json:"cidr_block"` + Allocated json.RawMessage `json:"allocated"` +} + +// ipamRecordV2 is the post-migration shape. +type ipamRecordV2 struct { + SubnetId string `json:"subnet_id"` + CidrBlock string `json:"cidr_block"` + Allocated []ipEntryV2 `json:"allocated"` +} + +// eniRecordSnapshot is the minimum ENIRecord shape needed to attribute IPAM +// entries to their owning ENI. Frozen copy keeps the migration self-contained. +type eniRecordSnapshot struct { + NetworkInterfaceId string `json:"network_interface_id"` + SubnetId string `json:"subnet_id"` + PrivateIpAddress string `json:"private_ip_address"` +} + +func init() { + DefaultRegistry.RegisterKV(ipamBucket, KVMigration{ + FromVersion: 1, + ToVersion: 2, + Description: "convert subnet IPAM Allocated []string → []IPEntry; backfill Purpose+OwnerID from ENIs", + Run: func(ctx KVContext) error { + ownerIndex, err := buildIPAMOwnerIndex(ctx.JetStream) + if err != nil { + return fmt.Errorf("build owner index: %w", err) + } + + keys, err := ctx.KV.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return nil + } + return fmt.Errorf("list keys: %w", err) + } + + for _, key := range keys { + if key == utils.VersionKey { + continue + } + if err := convertIPAMRecord(ctx, key, ownerIndex); err != nil { + return err + } + } + return nil + }, + }) +} + +// buildIPAMOwnerIndex returns map[subnetID]map[ip]eniID by walking the ENI +// bucket. Returns an empty index if the ENI bucket is missing +// (fresh-install case) so the migration still runs cleanly. +func buildIPAMOwnerIndex(js nats.JetStreamContext) (map[string]map[string]string, error) { + if js == nil { + // Called via the bare RunKV path (e.g. in tests with no JS). Skip + // owner attribution — all entries become Purpose=unknown so the + // schema-only conversion still succeeds. + return map[string]map[string]string{}, nil + } + eniKV, err := js.KeyValue(eniBucket) + if err != nil { + if errors.Is(err, nats.ErrBucketNotFound) { + return map[string]map[string]string{}, nil + } + return nil, fmt.Errorf("open ENI bucket: %w", err) + } + + keys, err := eniKV.Keys() + if err != nil { + if errors.Is(err, nats.ErrNoKeysFound) { + return map[string]map[string]string{}, nil + } + return nil, fmt.Errorf("list ENI keys: %w", err) + } + + index := make(map[string]map[string]string) + for _, key := range keys { + if key == utils.VersionKey { + continue + } + entry, err := eniKV.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + continue + } + return nil, fmt.Errorf("get ENI %s: %w", key, err) + } + var eni eniRecordSnapshot + if err := json.Unmarshal(entry.Value(), &eni); err != nil { + continue + } + if eni.SubnetId == "" || eni.PrivateIpAddress == "" { + continue + } + if _, ok := index[eni.SubnetId]; !ok { + index[eni.SubnetId] = make(map[string]string) + } + index[eni.SubnetId][eni.PrivateIpAddress] = eni.NetworkInterfaceId + } + return index, nil +} + +func convertIPAMRecord(ctx KVContext, key string, ownerIndex map[string]map[string]string) error { + var lastErr error + for attempt := range ipamMigrationMaxRetries { + entry, err := ctx.KV.Get(key) + if err != nil { + if errors.Is(err, nats.ErrKeyNotFound) { + return nil + } + return fmt.Errorf("get %s: %w", key, err) + } + + var raw ipamRecordV1V2 + if err := json.Unmarshal(entry.Value(), &raw); err != nil { + return fmt.Errorf("unmarshal %s: %w", key, err) + } + + // Detect schema: v2 records start with `[{`, v1 with `["`. + // Idempotent skip when already migrated. + if v2Entries, ok := tryDecodeV2(raw.Allocated); ok { + _ = v2Entries + return nil + } + + var legacy []string + if len(raw.Allocated) > 0 { + if err := json.Unmarshal(raw.Allocated, &legacy); err != nil { + return fmt.Errorf("unmarshal legacy []string for %s: %w", key, err) + } + } + + ownerByIP := ownerIndex[raw.SubnetId] + converted := make([]ipEntryV2, 0, len(legacy)) + for _, ip := range legacy { + ownerID, found := ownerByIP[ip] + purpose := "eni-primary" + if !found { + ownerID = "" + purpose = "unknown" + ctx.Logger.Warn("internal IPAM v1→v2 migration: no owning ENI, tagging unknown", + "subnet", raw.SubnetId, "ip", ip) + } + converted = append(converted, ipEntryV2{IP: ip, Purpose: purpose, OwnerID: ownerID}) + } + + newRecord := ipamRecordV2{ + SubnetId: raw.SubnetId, + CidrBlock: raw.CidrBlock, + Allocated: converted, + } + data, err := json.Marshal(newRecord) + if err != nil { + return fmt.Errorf("marshal %s: %w", key, err) + } + if _, err := ctx.KV.Update(key, data, entry.Revision()); err != nil { + if errors.Is(err, nats.ErrKeyExists) { + ctx.Logger.Warn("internal IPAM migration CAS conflict, retrying", "key", key, "attempt", attempt+1) + lastErr = err + continue + } + return fmt.Errorf("update %s: %w", key, err) + } + return nil + } + return fmt.Errorf("convert IPAM record %s: exceeded %d CAS retries: %w", key, ipamMigrationMaxRetries, lastErr) +} + +// tryDecodeV2 returns true when raw decodes as []ipEntryV2 with at least one +// entry whose IP field is set — the cheapest reliable v2 detector. Empty +// arrays decode as either schema; we treat them as already migrated. +func tryDecodeV2(raw json.RawMessage) ([]ipEntryV2, bool) { + if len(raw) == 0 { + return nil, true + } + var entries []ipEntryV2 + if err := json.Unmarshal(raw, &entries); err != nil { + return nil, false + } + for _, e := range entries { + if e.IP != "" { + return entries, true + } + } + // Zero-length array — already v2-shaped. + if len(entries) == 0 { + return entries, true + } + return nil, false +} diff --git a/spinifex/migrate/003_ipam_purpose_test.go b/spinifex/migrate/003_ipam_purpose_test.go new file mode 100644 index 000000000..b479b0856 --- /dev/null +++ b/spinifex/migrate/003_ipam_purpose_test.go @@ -0,0 +1,158 @@ +package migrate + +import ( + "encoding/json" + "log/slog" + "testing" + + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func setupIPAMMigrationKV(t *testing.T) (nats.KeyValue, nats.JetStreamContext) { + t.Helper() + _, nc := startTestNATS(t) + js, err := nc.JetStream() + require.NoError(t, err) + kv := createTestBucket(t, nc, ipamBucket) + return kv, js +} + +func runIPAMMigration(t *testing.T, kv nats.KeyValue, js nats.JetStreamContext) { + t.Helper() + for _, m := range DefaultRegistry.kvMigrations[ipamBucket] { + if m.FromVersion == 1 && m.ToVersion == 2 { + require.NoError(t, m.Run(KVContext{KV: kv, JetStream: js, Logger: slog.Default()})) + return + } + } + t.Fatalf("v1→v2 migration not registered for %s", ipamBucket) +} + +func seedENIBucket(t *testing.T, js nats.JetStreamContext, enis []eniRecordSnapshot) { + t.Helper() + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: eniBucket, History: 1}) + require.NoError(t, err) + for _, eni := range enis { + data, err := json.Marshal(eni) + require.NoError(t, err) + _, err = kv.Put(eni.NetworkInterfaceId, data) + require.NoError(t, err) + } +} + +func TestIPAMMigration_ConvertsV1ToV2WithOwnerBackfill(t *testing.T) { + kv, js := setupIPAMMigrationKV(t) + seedENIBucket(t, js, []eniRecordSnapshot{ + {NetworkInterfaceId: "eni-aaa", SubnetId: "subnet-1", PrivateIpAddress: "10.0.1.4"}, + {NetworkInterfaceId: "eni-bbb", SubnetId: "subnet-1", PrivateIpAddress: "10.0.1.5"}, + {NetworkInterfaceId: "eni-ccc", SubnetId: "subnet-2", PrivateIpAddress: "10.0.2.4"}, + }) + + type legacyRecord struct { + SubnetId string `json:"subnet_id"` + CidrBlock string `json:"cidr_block"` + Allocated []string `json:"allocated"` + } + subnet1 := legacyRecord{ + SubnetId: "subnet-1", + CidrBlock: "10.0.1.0/24", + Allocated: []string{"10.0.1.4", "10.0.1.5", "10.0.1.99"}, // .99 has no owning ENI + } + data, err := json.Marshal(subnet1) + require.NoError(t, err) + _, err = kv.Put("subnet-1", data) + require.NoError(t, err) + + subnet2 := legacyRecord{ + SubnetId: "subnet-2", + CidrBlock: "10.0.2.0/24", + Allocated: []string{"10.0.2.4"}, + } + data, err = json.Marshal(subnet2) + require.NoError(t, err) + _, err = kv.Put("subnet-2", data) + require.NoError(t, err) + + runIPAMMigration(t, kv, js) + + entry, err := kv.Get("subnet-1") + require.NoError(t, err) + var got ipamRecordV2 + require.NoError(t, json.Unmarshal(entry.Value(), &got)) + + require.Len(t, got.Allocated, 3) + assert.Equal(t, "10.0.1.4", got.Allocated[0].IP) + assert.Equal(t, "eni-primary", got.Allocated[0].Purpose) + assert.Equal(t, "eni-aaa", got.Allocated[0].OwnerID) + assert.Equal(t, "10.0.1.5", got.Allocated[1].IP) + assert.Equal(t, "eni-bbb", got.Allocated[1].OwnerID) + assert.Equal(t, "10.0.1.99", got.Allocated[2].IP) + assert.Equal(t, "unknown", got.Allocated[2].Purpose, "orphan IP tagged unknown") + assert.Empty(t, got.Allocated[2].OwnerID) + + entry, err = kv.Get("subnet-2") + require.NoError(t, err) + require.NoError(t, json.Unmarshal(entry.Value(), &got)) + require.Len(t, got.Allocated, 1) + assert.Equal(t, "eni-ccc", got.Allocated[0].OwnerID) +} + +func TestIPAMMigration_IdempotentOnAlreadyV2(t *testing.T) { + kv, js := setupIPAMMigrationKV(t) + + v2 := ipamRecordV2{ + SubnetId: "subnet-1", + CidrBlock: "10.0.1.0/24", + Allocated: []ipEntryV2{ + {IP: "10.0.1.4", Purpose: "eni-primary", OwnerID: "eni-aaa"}, + }, + } + data, err := json.Marshal(v2) + require.NoError(t, err) + _, err = kv.Put("subnet-1", data) + require.NoError(t, err) + rev := getKVRevision(t, kv, "subnet-1") + + runIPAMMigration(t, kv, js) + + newRev := getKVRevision(t, kv, "subnet-1") + assert.Equal(t, rev, newRev, "no write when record is already v2") +} + +func TestIPAMMigration_NoJetStream_TagsUnknown(t *testing.T) { + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, ipamBucket) + + type legacyRecord struct { + SubnetId string `json:"subnet_id"` + CidrBlock string `json:"cidr_block"` + Allocated []string `json:"allocated"` + } + rec := legacyRecord{ + SubnetId: "subnet-1", + CidrBlock: "10.0.1.0/24", + Allocated: []string{"10.0.1.4"}, + } + data, err := json.Marshal(rec) + require.NoError(t, err) + _, err = kv.Put("subnet-1", data) + require.NoError(t, err) + + // Run migration with nil JetStream — schema conversion still runs but + // every entry becomes Purpose=unknown. + runIPAMMigration(t, kv, nil) + + entry, err := kv.Get("subnet-1") + require.NoError(t, err) + var got ipamRecordV2 + require.NoError(t, json.Unmarshal(entry.Value(), &got)) + require.Len(t, got.Allocated, 1) + assert.Equal(t, "unknown", got.Allocated[0].Purpose) +} + +func TestIPAMMigration_EmptyBucket(t *testing.T) { + kv, js := setupIPAMMigrationKV(t) + runIPAMMigration(t, kv, js) +} diff --git a/spinifex/migrate/migrate.go b/spinifex/migrate/migrate.go index 4a2f3d712..41f695352 100644 --- a/spinifex/migrate/migrate.go +++ b/spinifex/migrate/migrate.go @@ -21,9 +21,12 @@ type KVMigration struct { } // KVContext provides KV migration functions access to the bucket being migrated. +// JetStream is non-nil only when the caller used RunKVWithJetStream — used by +// migrations that need to read sibling buckets (e.g. owner-attribution lookups). type KVContext struct { - KV nats.KeyValue - Logger *slog.Logger + KV nats.KeyValue + JetStream nats.JetStreamContext + Logger *slog.Logger } // ConfigMigration represents a versioned transformation of on-disk config files. @@ -97,6 +100,13 @@ func (r *Registry) RegisterConfig(target string, m ConfigMigration) { }) } +// RunKVWithJetStream is RunKV with a JetStream handle attached to each +// migration's KVContext, enabling cross-bucket reads (e.g. owner-attribution +// during a backfill). Prefer plain RunKV when the migration is self-contained. +func (r *Registry) RunKVWithJetStream(bucket string, kv nats.KeyValue, js nats.JetStreamContext, targetVersion int) error { + return r.runKV(bucket, kv, js, targetVersion) +} + // RunKV executes all pending migrations for a KV bucket up to targetVersion. // It reads the current version via utils.ReadVersion, runs each migration in // order, and stamps the version via utils.WriteVersion after each step. @@ -109,6 +119,10 @@ func (r *Registry) RegisterConfig(target string, m ConfigMigration) { // an error — migration authors must register migrations before bumping version // constants. func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) error { + return r.runKV(bucket, kv, nil, targetVersion) +} + +func (r *Registry) runKV(bucket string, kv nats.KeyValue, js nats.JetStreamContext, targetVersion int) error { current, err := utils.ReadVersion(kv) if err != nil { return fmt.Errorf("read version for %s: %w", bucket, err) @@ -168,7 +182,7 @@ func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) err logger := slog.Default() for _, m := range pending { logger.Info("Running KV migration", "bucket", bucket, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description) - ctx := KVContext{KV: kv, Logger: logger} + ctx := KVContext{KV: kv, JetStream: js, Logger: logger} if err := m.Run(ctx); err != nil { return fmt.Errorf("KV migration %s %d→%d failed: %w", bucket, m.FromVersion, m.ToVersion, err) }