mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
chore: delete stale migrations since we forced clean re init anyway
This commit is contained in:
Julian Sommer
@@ -15,32 +15,27 @@ import (
|
||||
)
|
||||
|
||||
// TestUpgradeSeesEveryConfigTarget pins the registrations `spx admin upgrade`
|
||||
// depends on. A migration whose package is no longer linked here would leave
|
||||
// the command reporting nothing pending on an install that needs migrating.
|
||||
// depends on. A target whose package is no longer linked here would leave the
|
||||
// command reporting nothing at all for a config file it installs.
|
||||
//
|
||||
// spinifex.toml is the only config target: predastore.toml is not migrated at
|
||||
// all, so a predastore config on disk must contribute neither a version nor a
|
||||
// pending step.
|
||||
// pending step. No migrations are registered, so nothing is ever pending.
|
||||
func TestUpgradeSeesEveryConfigTarget(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
|
||||
// A target must exist on disk to be reported at all.
|
||||
require.NoError(t, os.MkdirAll(filepath.Join(configDir, "predastore"), 0750))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"4\"\n"), 0640))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "predastore", "predastore.toml"), []byte("version = 1\n"), 0640))
|
||||
|
||||
versions, err := migrate.DefaultRegistry.ConfigVersions(configDir)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, map[string]int{"spinifex.toml": 3}, versions)
|
||||
assert.Equal(t, map[string]int{"spinifex.toml": 4}, versions)
|
||||
|
||||
pending, err := migrate.DefaultRegistry.PendingConfig(configDir)
|
||||
require.NoError(t, err)
|
||||
|
||||
targets := make(map[string][2]int, len(pending))
|
||||
for _, p := range pending {
|
||||
targets[p.Target] = [2]int{p.FromVersion, p.ToVersion}
|
||||
}
|
||||
assert.Equal(t, map[string][2]int{"spinifex.toml": {3, 4}}, targets)
|
||||
assert.Empty(t, pending)
|
||||
}
|
||||
|
||||
func TestValidateUpgradeFlags(t *testing.T) {
|
||||
@@ -140,19 +135,6 @@ func TestReportConfigStatus_NoPendingMigrations(t *testing.T) {
|
||||
assert.Contains(t, out, "spinifex.toml:")
|
||||
}
|
||||
|
||||
func TestReportConfigStatus_PendingMigrationIsListed(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
|
||||
var pending []migrate.PendingMigration
|
||||
out := captureStdout(t, func() { pending = reportConfigStatus(configDir) })
|
||||
|
||||
require.Len(t, pending, 1)
|
||||
assert.Equal(t, "spinifex.toml", pending[0].Target)
|
||||
assert.Contains(t, out, "Pending config migrations:")
|
||||
assert.Contains(t, out, "spinifex.toml 3 → 4")
|
||||
}
|
||||
|
||||
// captureStderr redirects os.Stderr for the duration of fn and returns what
|
||||
// was written. runAdminUpgrade's error paths print to stderr, not stdout.
|
||||
func captureStderr(t *testing.T, fn func()) string {
|
||||
@@ -293,9 +275,9 @@ func TestRunAdminUpgrade_DryRunNothingPending(t *testing.T) {
|
||||
assert.Contains(t, out, "Nothing pending — config and units are current.")
|
||||
}
|
||||
|
||||
func TestRunAdminUpgrade_DryRunReportsBothPending(t *testing.T) {
|
||||
func TestRunAdminUpgrade_DryRunReportsUnitsPending(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"4\"\n"), 0640))
|
||||
withUnitRoot(t, t.TempDir()) // empty: every unit reports missing
|
||||
|
||||
cmd := newUpgradeTestCmd(t, configDir, t.TempDir())
|
||||
@@ -307,15 +289,12 @@ func TestRunAdminUpgrade_DryRunReportsBothPending(t *testing.T) {
|
||||
})
|
||||
assert.Equal(t, -1, code)
|
||||
assert.NotContains(t, out, "Nothing pending")
|
||||
assert.Contains(t, out, "Pending config migrations:")
|
||||
assert.Contains(t, out, "will install")
|
||||
}
|
||||
|
||||
func TestRunAdminUpgrade_UnitsOnlySkipsConfigReporting(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
// Version 3 would normally report a pending config migration; --units-only
|
||||
// must never look at it.
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"4\"\n"), 0640))
|
||||
withUnitRoot(t, t.TempDir())
|
||||
|
||||
cmd := newUpgradeTestCmd(t, configDir, t.TempDir())
|
||||
@@ -332,7 +311,7 @@ func TestRunAdminUpgrade_UnitsOnlySkipsConfigReporting(t *testing.T) {
|
||||
|
||||
func TestRunAdminUpgrade_SkipUnitsSkipsUnitReporting(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"4\"\n"), 0640))
|
||||
// Point systemdUnitRoot at a path Reconcile must never touch: if
|
||||
// --skip-units leaked a call through, a nonexistent root would still
|
||||
// "work" for a dry run, so the real proof is that its output never
|
||||
@@ -348,31 +327,7 @@ func TestRunAdminUpgrade_SkipUnitsSkipsUnitReporting(t *testing.T) {
|
||||
})
|
||||
assert.Equal(t, -1, code)
|
||||
assert.NotContains(t, out, "Reading installed systemd unit versions", "--skip-units must skip unit reporting")
|
||||
assert.Contains(t, out, "Pending config migrations:")
|
||||
}
|
||||
|
||||
func TestRunAdminUpgrade_ApplyRunsConfigMigrationAndUnitReconcile(t *testing.T) {
|
||||
configDir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, "spinifex.toml"), []byte("version = \"3\"\n"), 0640))
|
||||
// Every unit already current: the real (non-dry-run) Reconcile call
|
||||
// below finds nothing to change, so it never touches daemon-reload.
|
||||
unitRoot := t.TempDir()
|
||||
seedUnitRoot(t, unitRoot)
|
||||
withUnitRoot(t, unitRoot)
|
||||
|
||||
cmd := newUpgradeTestCmd(t, configDir, t.TempDir())
|
||||
setUpgradeFlags(t, cmd, map[string]string{"yes": "true"})
|
||||
|
||||
var code int
|
||||
out := captureStdout(t, func() {
|
||||
code = withUpgradeExitCapture(t, func() { runAdminUpgrade(cmd, nil) })
|
||||
})
|
||||
assert.Equal(t, -1, code, "the apply must succeed without exiting")
|
||||
assert.Contains(t, out, "Done.")
|
||||
|
||||
migrated, err := os.ReadFile(filepath.Join(configDir, "spinifex.toml"))
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(migrated), `version = "4"`, "spinifex.toml must have been migrated to the latest version")
|
||||
assert.Contains(t, out, "Reading current config versions")
|
||||
}
|
||||
|
||||
func TestRunAdminUpgrade_ApplyWithNonWritableUnitRootExitsWithRootHint(t *testing.T) {
|
||||
|
||||
@@ -1,124 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// sgBucket mirrors handlers/ec2/vpc.KVBucketSecurityGroups. Duplicated here to
|
||||
// keep the migrate package free of any handler dependency; if the bucket name
|
||||
// ever changes, both constants must move together.
|
||||
const sgBucket = "spinifex-vpc-security-groups"
|
||||
|
||||
const sgMigrationMaxRetries = 3
|
||||
|
||||
// sgRule is the v1→v2 schema snapshot of handlers/ec2/vpc.SGRule. A frozen
|
||||
// copy decouples this migration from future struct evolution.
|
||||
type sgRule struct {
|
||||
RuleId string `json:"rule_id"`
|
||||
IpProtocol string `json:"ip_protocol"`
|
||||
FromPort int64 `json:"from_port"`
|
||||
ToPort int64 `json:"to_port"`
|
||||
CidrIp string `json:"cidr_ip,omitempty"`
|
||||
SourceSG string `json:"source_sg,omitempty"`
|
||||
}
|
||||
|
||||
// sgRecord is the v1→v2 schema snapshot of handlers/ec2/vpc.SecurityGroupRecord.
|
||||
type sgRecord struct {
|
||||
GroupId string `json:"group_id"`
|
||||
GroupName string `json:"group_name"`
|
||||
Description string `json:"description"`
|
||||
VpcId string `json:"vpc_id"`
|
||||
IngressRules []sgRule `json:"ingress_rules"`
|
||||
EgressRules []sgRule `json:"egress_rules"`
|
||||
Tags map[string]string `json:"tags"`
|
||||
IsDefault bool `json:"is_default,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
func init() {
|
||||
DefaultRegistry.RegisterKV(sgBucket, KVMigration{
|
||||
FromVersion: 1,
|
||||
ToVersion: 2,
|
||||
Description: "assign sgr- IDs to security group rules",
|
||||
Run: func(ctx context.Context, kvc KVContext) error {
|
||||
keys, err := kvc.KV.Keys(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("list keys: %w", err)
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if key == utils.VersionKey {
|
||||
continue
|
||||
}
|
||||
if err := backfillSGRuleIDs(ctx, kvc, key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// backfillSGRuleIDs assigns sgr- IDs to any rule missing one. Skips records
|
||||
// already fully populated. CAS conflicts retry up to sgMigrationMaxRetries.
|
||||
func backfillSGRuleIDs(ctx context.Context, kvc KVContext, key string) error {
|
||||
var lastErr error
|
||||
for attempt := range sgMigrationMaxRetries {
|
||||
entry, err := kvc.KV.Get(ctx, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("get %s: %w", key, err)
|
||||
}
|
||||
|
||||
var record sgRecord
|
||||
if err := json.Unmarshal(entry.Value(), &record); err != nil {
|
||||
return fmt.Errorf("unmarshal %s: %w", key, err)
|
||||
}
|
||||
|
||||
changed := false
|
||||
for i := range record.IngressRules {
|
||||
if record.IngressRules[i].RuleId == "" {
|
||||
record.IngressRules[i].RuleId = utils.GenerateResourceID("sgr")
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
for i := range record.EgressRules {
|
||||
if record.EgressRules[i].RuleId == "" {
|
||||
record.EgressRules[i].RuleId = utils.GenerateResourceID("sgr")
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
|
||||
if !changed {
|
||||
return nil
|
||||
}
|
||||
|
||||
data, err := json.Marshal(record)
|
||||
if err != nil {
|
||||
return fmt.Errorf("marshal %s: %w", key, err)
|
||||
}
|
||||
if _, err := kvc.KV.Update(ctx, key, data, entry.Revision()); err != nil {
|
||||
// A concurrent writer moved the record between the Get and here.
|
||||
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
kvc.Logger.Warn("SG 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("backfill rule IDs for %s: exceeded %d CAS retries: %w", key, sgMigrationMaxRetries, lastErr)
|
||||
}
|
||||
@@ -1,244 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const sgTestAccountID = "123456789012"
|
||||
|
||||
func sgMigrationKV(t *testing.T) jetstream.KeyValue {
|
||||
t.Helper()
|
||||
_, nc := startTestNATS(t)
|
||||
return createTestBucket(t, nc, "sg-migration-test")
|
||||
}
|
||||
|
||||
func runSGMigration(t *testing.T, kv jetstream.KeyValue) {
|
||||
t.Helper()
|
||||
for _, m := range DefaultRegistry.kvMigrations[sgBucket] {
|
||||
if m.FromVersion == 1 && m.ToVersion == 2 {
|
||||
require.NoError(t, m.Run(t.Context(), KVContext{KV: kv, Logger: slog.Default()}))
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Fatalf("v1→v2 migration not registered for %s", sgBucket)
|
||||
}
|
||||
|
||||
func TestSGMigration_AssignsBlankRuleIDs(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
|
||||
record := sgRecord{
|
||||
GroupId: "sg-0123456789abcdef0",
|
||||
GroupName: "pre-v2",
|
||||
VpcId: "vpc-0123456789abcdef0",
|
||||
IngressRules: []sgRule{
|
||||
{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/24"},
|
||||
{IpProtocol: "tcp", FromPort: 443, ToPort: 443, CidrIp: "0.0.0.0/0"},
|
||||
},
|
||||
EgressRules: []sgRule{
|
||||
{IpProtocol: "-1", FromPort: 0, ToPort: 0, CidrIp: "0.0.0.0/0"},
|
||||
},
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(record)
|
||||
require.NoError(t, err)
|
||||
_, err = kv.Put(t.Context(), utils.AccountKey(sgTestAccountID, record.GroupId), data)
|
||||
require.NoError(t, err)
|
||||
|
||||
runSGMigration(t, kv)
|
||||
|
||||
entry, err := kv.Get(t.Context(), utils.AccountKey(sgTestAccountID, record.GroupId))
|
||||
require.NoError(t, err)
|
||||
var got sgRecord
|
||||
require.NoError(t, json.Unmarshal(entry.Value(), &got))
|
||||
|
||||
assert.Equal(t, record.GroupId, got.GroupId, "non-rule fields must be preserved")
|
||||
assert.Equal(t, record.VpcId, got.VpcId)
|
||||
require.Len(t, got.IngressRules, 2)
|
||||
require.Len(t, got.EgressRules, 1)
|
||||
seen := make(map[string]struct{})
|
||||
for _, r := range got.IngressRules {
|
||||
assert.Regexp(t, `^sgr-[0-9a-f]{17}$`, r.RuleId)
|
||||
seen[r.RuleId] = struct{}{}
|
||||
}
|
||||
for _, r := range got.EgressRules {
|
||||
assert.Regexp(t, `^sgr-[0-9a-f]{17}$`, r.RuleId)
|
||||
seen[r.RuleId] = struct{}{}
|
||||
}
|
||||
assert.Len(t, seen, 3, "all assigned IDs must be unique")
|
||||
}
|
||||
|
||||
func TestSGMigration_IdempotentOnAlreadyMigrated(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
|
||||
record := sgRecord{
|
||||
GroupId: "sg-0123456789abcdef0",
|
||||
GroupName: "already-v2",
|
||||
VpcId: "vpc-0123456789abcdef0",
|
||||
IngressRules: []sgRule{
|
||||
{RuleId: "sgr-aaaaaaaaaaaaaaaaa", IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/24"},
|
||||
},
|
||||
EgressRules: []sgRule{
|
||||
{RuleId: "sgr-bbbbbbbbbbbbbbbbb", IpProtocol: "-1", FromPort: 0, ToPort: 0, CidrIp: "0.0.0.0/0"},
|
||||
},
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(record)
|
||||
require.NoError(t, err)
|
||||
key := utils.AccountKey(sgTestAccountID, record.GroupId)
|
||||
_, err = kv.Put(t.Context(), key, data)
|
||||
require.NoError(t, err)
|
||||
|
||||
entryBefore, err := kv.Get(t.Context(), key)
|
||||
require.NoError(t, err)
|
||||
revBefore := entryBefore.Revision()
|
||||
|
||||
runSGMigration(t, kv)
|
||||
|
||||
entryAfter, err := kv.Get(t.Context(), key)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, revBefore, entryAfter.Revision(), "fully-populated records must not be rewritten")
|
||||
}
|
||||
|
||||
func TestSGMigration_PartialBackfill(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
|
||||
record := sgRecord{
|
||||
GroupId: "sg-0123456789abcdef0",
|
||||
GroupName: "partial",
|
||||
VpcId: "vpc-0123456789abcdef0",
|
||||
IngressRules: []sgRule{
|
||||
{RuleId: "sgr-aaaaaaaaaaaaaaaaa", IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/24"},
|
||||
{IpProtocol: "tcp", FromPort: 80, ToPort: 80, CidrIp: "0.0.0.0/0"},
|
||||
},
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(record)
|
||||
require.NoError(t, err)
|
||||
_, err = kv.Put(t.Context(), utils.AccountKey(sgTestAccountID, record.GroupId), data)
|
||||
require.NoError(t, err)
|
||||
|
||||
runSGMigration(t, kv)
|
||||
|
||||
entry, err := kv.Get(t.Context(), utils.AccountKey(sgTestAccountID, record.GroupId))
|
||||
require.NoError(t, err)
|
||||
var got sgRecord
|
||||
require.NoError(t, json.Unmarshal(entry.Value(), &got))
|
||||
assert.Equal(t, "sgr-aaaaaaaaaaaaaaaaa", got.IngressRules[0].RuleId)
|
||||
assert.Regexp(t, `^sgr-[0-9a-f]{17}$`, got.IngressRules[1].RuleId)
|
||||
}
|
||||
|
||||
func TestSGMigration_EmptyBucket(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
runSGMigration(t, kv)
|
||||
}
|
||||
|
||||
// wrongLastSequence builds the error kvs.Update returns for a lost CAS race on
|
||||
// a bucket whose stream reports the conflict under code. Single-replica streams
|
||||
// report 10071, replicated ones 10164; both are wrapped with
|
||||
// ErrKeyRevisionMismatch, and only 10071 also satisfies ErrKeyExists.
|
||||
func wrongLastSequence(code jetstream.ErrorCode) error {
|
||||
apiErr := &jetstream.APIError{
|
||||
ErrorCode: code,
|
||||
Code: 400,
|
||||
Description: "wrong last sequence: 3",
|
||||
}
|
||||
return fmt.Errorf("%w: %w", apiErr, jetstream.ErrKeyRevisionMismatch)
|
||||
}
|
||||
|
||||
// casConflictKV wraps a jetstream.KeyValue and forces the next failuresLeft
|
||||
// Update calls to fail with a wrong-last-sequence response, simulating the
|
||||
// concurrent writer that backfillSGRuleIDs retries on.
|
||||
type casConflictKV struct {
|
||||
jetstream.KeyValue
|
||||
|
||||
failuresLeft int
|
||||
conflictCode jetstream.ErrorCode
|
||||
}
|
||||
|
||||
func (k *casConflictKV) Update(ctx context.Context, key string, value []byte, last uint64) (uint64, error) {
|
||||
if k.failuresLeft > 0 {
|
||||
k.failuresLeft--
|
||||
return 0, wrongLastSequence(k.conflictCode)
|
||||
}
|
||||
return k.KeyValue.Update(ctx, key, value, last)
|
||||
}
|
||||
|
||||
// conflictCodes names the two wrong-last-sequence codes a KV bucket can report,
|
||||
// so every CAS test runs against a replicated bucket as well as a single-replica
|
||||
// one. A predicate that only matches ErrKeyExists passes R1 and fails R3.
|
||||
var conflictCodes = map[string]jetstream.ErrorCode{
|
||||
"single replica": jetstream.JSErrCodeStreamWrongLastSequence,
|
||||
"replicated": jetstream.JSErrCodeStreamWrongLastSequenceConstant,
|
||||
}
|
||||
|
||||
func seedSGRecord(t *testing.T, kv jetstream.KeyValue) string {
|
||||
t.Helper()
|
||||
record := sgRecord{
|
||||
GroupId: "sg-0123456789abcdef0",
|
||||
GroupName: "cas",
|
||||
VpcId: "vpc-0123456789abcdef0",
|
||||
IngressRules: []sgRule{
|
||||
{IpProtocol: "tcp", FromPort: 22, ToPort: 22, CidrIp: "10.0.0.0/24"},
|
||||
},
|
||||
CreatedAt: time.Now(),
|
||||
}
|
||||
data, err := json.Marshal(record)
|
||||
require.NoError(t, err)
|
||||
key := utils.AccountKey(sgTestAccountID, record.GroupId)
|
||||
_, err = kv.Put(t.Context(), key, data)
|
||||
require.NoError(t, err)
|
||||
return key
|
||||
}
|
||||
|
||||
func TestSGMigration_RetriesOnCASConflict(t *testing.T) {
|
||||
for name, code := range conflictCodes {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
key := seedSGRecord(t, kv)
|
||||
|
||||
stub := &casConflictKV{KeyValue: kv, failuresLeft: sgMigrationMaxRetries - 1, conflictCode: code}
|
||||
err := backfillSGRuleIDs(t.Context(), KVContext{KV: stub, Logger: slog.Default()}, key)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 0, stub.failuresLeft, "all injected conflicts must have been consumed")
|
||||
|
||||
entry, err := kv.Get(t.Context(), key)
|
||||
require.NoError(t, err)
|
||||
var got sgRecord
|
||||
require.NoError(t, json.Unmarshal(entry.Value(), &got))
|
||||
require.Len(t, got.IngressRules, 1)
|
||||
assert.Regexp(t, `^sgr-[0-9a-f]{17}$`, got.IngressRules[0].RuleId)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSGMigration_FailsAfterCASRetryExhaustion(t *testing.T) {
|
||||
for name, code := range conflictCodes {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
kv := sgMigrationKV(t)
|
||||
key := seedSGRecord(t, kv)
|
||||
|
||||
stub := &casConflictKV{KeyValue: kv, failuresLeft: sgMigrationMaxRetries + 1, conflictCode: code}
|
||||
err := backfillSGRuleIDs(t.Context(), KVContext{KV: stub, Logger: slog.Default()}, key)
|
||||
require.Error(t, err)
|
||||
assert.Contains(t, err.Error(), "exceeded 3 CAS retries")
|
||||
assert.ErrorIs(t, err, jetstream.ErrKeyRevisionMismatch,
|
||||
"exhaustion error must wrap the underlying CAS conflict")
|
||||
|
||||
entry, err := kv.Get(t.Context(), key)
|
||||
require.NoError(t, err)
|
||||
var got sgRecord
|
||||
require.NoError(t, json.Unmarshal(entry.Value(), &got))
|
||||
assert.Empty(t, got.IngressRules[0].RuleId, "record must be untouched after retry exhaustion")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,135 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// 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.
|
||||
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 context.Context, kvc KVContext) error {
|
||||
keys, err := kvc.KV.Keys(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("list keys: %w", err)
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if key == utils.VersionKey {
|
||||
continue
|
||||
}
|
||||
if err := renameExternalIPAMType(ctx, kvc, key); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func renameExternalIPAMType(ctx context.Context, kvc KVContext, key string) error {
|
||||
var lastErr error
|
||||
for attempt := range extIPAMMaxRetries {
|
||||
entry, err := kvc.KV.Get(ctx, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.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 == "" {
|
||||
kvc.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 := kvc.KV.Update(ctx, key, data, entry.Revision()); err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
kvc.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)
|
||||
}
|
||||
@@ -1,103 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func extIPAMMigrationKV(t *testing.T) jetstream.KeyValue {
|
||||
t.Helper()
|
||||
_, nc := startTestNATS(t)
|
||||
return createTestBucket(t, nc, "ext-ipam-migration-test")
|
||||
}
|
||||
|
||||
func runExtIPAMMigration(t *testing.T, kv jetstream.KeyValue) {
|
||||
t.Helper()
|
||||
for _, m := range DefaultRegistry.kvMigrations[extIPAMBucket] {
|
||||
if m.FromVersion == 1 && m.ToVersion == 2 {
|
||||
require.NoError(t, m.Run(t.Context(), 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(t.Context(), "wan", data)
|
||||
require.NoError(t, err)
|
||||
|
||||
runExtIPAMMigration(t, kv)
|
||||
|
||||
entry, err := kv.Get(t.Context(), "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(t.Context(), "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 jetstream.KeyValue, key string) uint64 {
|
||||
t.Helper()
|
||||
entry, err := kv.Get(t.Context(), key)
|
||||
require.NoError(t, err)
|
||||
return entry.Revision()
|
||||
}
|
||||
@@ -1,225 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/utils"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// ipamBucket / eniBucket mirror handlers/ec2/vpc.KVBucketIPAM and
|
||||
// .KVBucketENIs. Duplicated here so the migrate package stays handler-free.
|
||||
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 context.Context, kvc KVContext) error {
|
||||
ownerIndex, err := buildIPAMOwnerIndex(ctx, kvc.JetStream)
|
||||
if err != nil {
|
||||
return fmt.Errorf("build owner index: %w", err)
|
||||
}
|
||||
|
||||
keys, err := kvc.KV.Keys(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("list keys: %w", err)
|
||||
}
|
||||
|
||||
for _, key := range keys {
|
||||
if key == utils.VersionKey {
|
||||
continue
|
||||
}
|
||||
if err := convertIPAMRecord(ctx, kvc, 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(ctx context.Context, js jetstream.JetStream) (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(ctx, eniBucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
return map[string]map[string]string{}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("open ENI bucket: %w", err)
|
||||
}
|
||||
|
||||
keys, err := eniKV.Keys(ctx)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.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(ctx, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.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 context.Context, kvc KVContext, key string, ownerIndex map[string]map[string]string) error {
|
||||
var lastErr error
|
||||
for attempt := range ipamMigrationMaxRetries {
|
||||
entry, err := kvc.KV.Get(ctx, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, jetstream.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"
|
||||
kvc.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 := kvc.KV.Update(ctx, key, data, entry.Revision()); err != nil {
|
||||
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
kvc.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
|
||||
}
|
||||
@@ -1,158 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"testing"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/testutil"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func setupIPAMMigrationKV(t *testing.T) (jetstream.KeyValue, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
_, nc := startTestNATS(t)
|
||||
js := testutil.NewJetStream(t, nc)
|
||||
kv := createTestBucket(t, nc, ipamBucket)
|
||||
return kv, js
|
||||
}
|
||||
|
||||
func runIPAMMigration(t *testing.T, kv jetstream.KeyValue, js jetstream.JetStream) {
|
||||
t.Helper()
|
||||
for _, m := range DefaultRegistry.kvMigrations[ipamBucket] {
|
||||
if m.FromVersion == 1 && m.ToVersion == 2 {
|
||||
require.NoError(t, m.Run(t.Context(), 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 jetstream.JetStream, enis []eniRecordSnapshot) {
|
||||
t.Helper()
|
||||
kv, err := js.CreateKeyValue(t.Context(), jetstream.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(t.Context(), 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(t.Context(), "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(t.Context(), "subnet-2", data)
|
||||
require.NoError(t, err)
|
||||
|
||||
runIPAMMigration(t, kv, js)
|
||||
|
||||
entry, err := kv.Get(t.Context(), "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(t.Context(), "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(t.Context(), "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(t.Context(), "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(t.Context(), "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)
|
||||
}
|
||||
@@ -1,91 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The spinifex config target, relative to the Spinifex config directory.
|
||||
const (
|
||||
spinifexTarget = "spinifex.toml"
|
||||
spinifexRelPath = "spinifex.toml"
|
||||
// spinifexHostIDVersion is the schema where the predastore section names a
|
||||
// cluster host rather than a node.
|
||||
spinifexHostIDVersion = 4
|
||||
)
|
||||
|
||||
// spinifexPredastoreTableRe matches the per-node predastore section header,
|
||||
// [nodes.<name>.predastore], which is the only place node_id ever appeared.
|
||||
var spinifexPredastoreTableRe = regexp.MustCompile(`^\[nodes\.[^\[\]]+\.predastore\]$`)
|
||||
|
||||
// spinifexNodeIDKeyRe matches the key to rename, preserving its indentation
|
||||
// and spacing so the rest of the line is untouched.
|
||||
var spinifexNodeIDKeyRe = regexp.MustCompile(`^(\s*)node_id(\s*=)`)
|
||||
|
||||
// spinifexTableHeaderRe captures the name of a table or table-array header.
|
||||
var spinifexTableHeaderRe = regexp.MustCompile(`^\[\[?([^\[\]]+)\]?\]$`)
|
||||
|
||||
func init() {
|
||||
DefaultRegistry.RegisterConfigTarget(spinifexTarget, spinifexRelPath, &TOMLVersionReader{})
|
||||
DefaultRegistry.RegisterConfig(spinifexTarget, ConfigMigration{
|
||||
FromVersion: 3,
|
||||
ToVersion: spinifexHostIDVersion,
|
||||
Description: "rename [nodes.*.predastore] node_id → host_id",
|
||||
Run: migrateSpinifexPredastoreHostID,
|
||||
})
|
||||
}
|
||||
|
||||
// migrateSpinifexPredastoreHostID renames the predastore node selector.
|
||||
//
|
||||
// A machine is one predastore host now, and derivePredastoreBind reads this
|
||||
// key to pick the process's host. Left as node_id, a multi-node member falls
|
||||
// back to host 0 and tries to run every node in the cluster in its own process.
|
||||
//
|
||||
// Single-node installs never had the key — the template only emits it for
|
||||
// multi-node — so for them this is a version bump and nothing else.
|
||||
func migrateSpinifexPredastoreHostID(ctx ConfigContext) error {
|
||||
path := filepath.Join(ctx.ConfigDir, spinifexRelPath)
|
||||
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read %s: %w", path, err)
|
||||
}
|
||||
|
||||
lines := strings.Split(string(data), "\n")
|
||||
inPredastore := false
|
||||
renamed := 0
|
||||
|
||||
for i, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
|
||||
if spinifexTableHeaderRe.MatchString(trimmed) {
|
||||
inPredastore = spinifexPredastoreTableRe.MatchString(trimmed)
|
||||
continue
|
||||
}
|
||||
if !inPredastore {
|
||||
continue
|
||||
}
|
||||
if spinifexNodeIDKeyRe.MatchString(line) {
|
||||
lines[i] = spinifexNodeIDKeyRe.ReplaceAllString(line, "${1}host_id${2}")
|
||||
renamed++
|
||||
}
|
||||
}
|
||||
|
||||
if renamed == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("stat %s: %w", path, err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")), info.Mode()); err != nil {
|
||||
return fmt.Errorf("write %s: %w", path, err)
|
||||
}
|
||||
|
||||
ctx.Logger.Info("Renamed predastore node_id to host_id", "path", path, "keys", renamed)
|
||||
return nil
|
||||
}
|
||||
@@ -1,99 +0,0 @@
|
||||
package migrate
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// spinifexTomlFixture writes a spinifex.toml and returns its config directory.
|
||||
func spinifexTomlFixture(t *testing.T, body string) string {
|
||||
t.Helper()
|
||||
configDir := t.TempDir()
|
||||
require.NoError(t, os.WriteFile(filepath.Join(configDir, spinifexRelPath), []byte(body), 0640))
|
||||
return configDir
|
||||
}
|
||||
|
||||
func readSpinifexToml(t *testing.T, configDir string) string {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(filepath.Join(configDir, spinifexRelPath))
|
||||
require.NoError(t, err)
|
||||
return string(data)
|
||||
}
|
||||
|
||||
// TestSpinifexPredastoreHostIDRename covers a multi-node member, where the key
|
||||
// selects which predastore host this machine runs.
|
||||
func TestSpinifexPredastoreHostIDRename(t *testing.T) {
|
||||
configDir := spinifexTomlFixture(t, `version = "3"
|
||||
|
||||
[nodes.node1]
|
||||
node = "node1"
|
||||
host = "10.0.0.1"
|
||||
|
||||
[nodes.node1.predastore]
|
||||
host = "10.0.0.1:8443"
|
||||
region = "ap-southeast-2"
|
||||
bucket = "predastore"
|
||||
node_id = 2
|
||||
|
||||
[nodes.node2]
|
||||
node = "node2"
|
||||
`)
|
||||
|
||||
require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir()))
|
||||
|
||||
got := readSpinifexToml(t, configDir)
|
||||
assert.Contains(t, got, "host_id = 2")
|
||||
assert.NotContains(t, got, "node_id")
|
||||
assert.Contains(t, got, `version = "4"`)
|
||||
|
||||
// The surrounding tables are untouched.
|
||||
assert.Contains(t, got, `host = "10.0.0.1:8443"`)
|
||||
assert.Contains(t, got, "[nodes.node2]")
|
||||
}
|
||||
|
||||
// TestSpinifexPredastoreHostIDLeavesOtherTables guards against renaming a
|
||||
// node_id that belongs to something else — only the predastore section moved
|
||||
// from naming a node to naming a host.
|
||||
func TestSpinifexPredastoreHostIDLeavesOtherTables(t *testing.T) {
|
||||
configDir := spinifexTomlFixture(t, `version = "3"
|
||||
|
||||
[cluster]
|
||||
node_id = 7
|
||||
|
||||
[nodes.node1.viperblock]
|
||||
node_id = 9
|
||||
|
||||
[nodes.node1.predastore]
|
||||
node_id = 2
|
||||
`)
|
||||
|
||||
require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir()))
|
||||
|
||||
got := readSpinifexToml(t, configDir)
|
||||
assert.Contains(t, got, "[cluster]\nnode_id = 7")
|
||||
assert.Contains(t, got, "[nodes.node1.viperblock]\nnode_id = 9")
|
||||
assert.Contains(t, got, "[nodes.node1.predastore]\nhost_id = 2")
|
||||
}
|
||||
|
||||
// TestSpinifexPredastoreHostIDSingleNode is the common install: the key is
|
||||
// only emitted for multi-node, so there is nothing to rename.
|
||||
func TestSpinifexPredastoreHostIDSingleNode(t *testing.T) {
|
||||
body := `version = "3"
|
||||
|
||||
[nodes.node1.predastore]
|
||||
host = "0.0.0.0:8443"
|
||||
bucket = "predastore"
|
||||
`
|
||||
configDir := spinifexTomlFixture(t, body)
|
||||
|
||||
require.NoError(t, DefaultRegistry.RunConfig(spinifexTarget, configDir, t.TempDir()))
|
||||
|
||||
got := readSpinifexToml(t, configDir)
|
||||
assert.Contains(t, got, `version = "4"`)
|
||||
assert.Contains(t, got, `host = "0.0.0.0:8443"`)
|
||||
assert.NotContains(t, got, "host_id")
|
||||
}
|
||||
@@ -5,7 +5,7 @@ Single source of truth for the current schema version of every config file Spini
|
||||
| Target | Current version | Canonical template |
|
||||
|---|---|---|
|
||||
| `nats.conf` | `3` | `cmd/spinifex/cmd/templates/nats.conf` |
|
||||
| `awsgw.toml` | `2` | `cmd/spinifex/cmd/templates/awsgw.toml` |
|
||||
| `awsgw.toml` | `3` | `cmd/spinifex/cmd/templates/awsgw.toml` |
|
||||
| `spinifex.toml` | `4` | `cmd/spinifex/cmd/templates/spinifex.toml` |
|
||||
| `predastore.toml` | `1` | `cmd/spinifex/cmd/templates/predastore.toml` |
|
||||
| `predastore-multinode.toml` | `1` | `cmd/spinifex/cmd/templates/predastore-multinode.toml` |
|
||||
@@ -15,26 +15,23 @@ Single source of truth for the current schema version of every config file Spini
|
||||
1. **Fresh install (`spx admin init`)** — the version string is baked into the embedded template file and written verbatim to disk. There is no Go constant; the template is the source of truth.
|
||||
2. **Upgrade (`spx admin upgrade`)** — the migration framework (`Registry.RunConfig`) calls `ConfigVersionReader.WriteVersion` after each registered migration step. A target with no registered migration is left alone, and the upgrade command reports nothing pending for it.
|
||||
|
||||
## Registered config migrations
|
||||
## Registered migrations
|
||||
|
||||
| Target | Migration | File |
|
||||
|---|---|---|
|
||||
| `spinifex.toml` | `3` → `4`: rename `[nodes.*.predastore]` `node_id` → `host_id` | `spinifex/migrate/005_spinifex_predastore_host_id.go` |
|
||||
None. The migrations that used to live here predated a breaking change that required `spx admin init --force`, so no install can reach the current versions by migrating and the steps were dropped rather than left as dead code.
|
||||
|
||||
## Registered object-store migrations
|
||||
|
||||
These migrate data in Predastore rather than a config file. Predastore has no conditional write, so their version is stamped in a shared JetStream KV bucket instead of in an object; see `RunObject`.
|
||||
|
||||
| Target | Migration | File | Version bucket |
|
||||
|---|---|---|---|
|
||||
| `ebsmetadata` | `0` → `1`: backfill `spinifex/ebsmetadata/v1` documents from legacy viperblock `config.json` | `spinifex/migrate/ebsmetadatabackfill/006_ebsmetadata_backfill.go` | `spinifex-ebsmetadata-migrate` |
|
||||
|
||||
Unlike config migrations, these do not run from `spx admin upgrade`. The ebsmetadata backfill runs from `Daemon.configureEBSProvider`, and only when `[ebs] provider = "viperblockd"` is selected: backfilling while the embedded engine is still authoritative would produce documents that immediately go stale.
|
||||
`spinifex.toml` is still registered as a config target in `migrate.go` so `spx admin upgrade` reports its on-disk version. KV and object-store buckets have no registered migrations either, so `RunKV` and `RunObject` stamp their target version directly on first init.
|
||||
|
||||
## Where migrations live
|
||||
|
||||
Register a `ConfigMigration` against `DefaultRegistry` in a new numbered file under `spinifex/migrate/`, and bump the version in both the template and the table above.
|
||||
|
||||
There are no migrations in the tree to copy from. For worked examples of all three kinds, read the deleted files out of git history:
|
||||
|
||||
```bash
|
||||
git log --diff-filter=D --name-only -- 'spinifex/migrate/0*.go'
|
||||
git show <commit>^:spinifex/migrate/003_ipam_purpose.go
|
||||
```
|
||||
|
||||
## Framework
|
||||
|
||||
The framework (`migrate.go`, `version_readers.go`) covers three kinds of target:
|
||||
|
||||
@@ -91,6 +91,14 @@ type Registry struct {
|
||||
// via init() functions so they are available before any service starts.
|
||||
var DefaultRegistry = NewRegistry()
|
||||
|
||||
// spinifexTarget is registered even with no migrations against it, so
|
||||
// `spx admin upgrade` still reports the version on disk.
|
||||
const spinifexTarget = "spinifex.toml"
|
||||
|
||||
func init() {
|
||||
DefaultRegistry.RegisterConfigTarget(spinifexTarget, spinifexTarget, &TOMLVersionReader{})
|
||||
}
|
||||
|
||||
// NewRegistry creates an empty migration registry.
|
||||
func NewRegistry() *Registry {
|
||||
return &Registry{
|
||||
|
||||
Reference in New Issue
Block a user