From 48c90b795aedfcf9418720a86f5803e7442b6214 Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Fri, 10 Apr 2026 14:34:15 +1000 Subject: [PATCH] feat: migration framework (#127) --- README.md | 8 +- cmd/spinifex/cmd/admin.go | 3 + cmd/spinifex/cmd/admin_upgrade.go | 92 +++ cmd/spinifex/cmd/templates/nats.conf | 1 + docs/COMMANDS.md | 6 + scripts/setup.sh | 89 +++ spinifex/daemon/jetstream.go | 18 +- spinifex/handlers/ec2/account/service_impl.go | 5 +- spinifex/handlers/ec2/eigw/service_impl.go | 5 +- spinifex/handlers/ec2/eip/service_impl.go | 5 +- spinifex/handlers/ec2/igw/service_impl.go | 5 +- spinifex/handlers/ec2/natgw/service_impl.go | 9 +- .../ec2/placementgroup/service_impl.go | 5 +- .../handlers/ec2/routetable/service_impl.go | 5 +- .../handlers/ec2/snapshot/service_impl.go | 5 +- spinifex/handlers/ec2/vpc/external_ipam.go | 5 +- spinifex/handlers/ec2/vpc/ipam.go | 5 +- spinifex/handlers/ec2/vpc/service_impl.go | 21 +- spinifex/handlers/elbv2/store.go | 5 +- spinifex/handlers/iam/service_impl.go | 21 +- spinifex/migrate/001_nats_token.go | 97 +++ spinifex/migrate/migrate.go | 330 ++++++++ spinifex/migrate/migrate_test.go | 738 ++++++++++++++++++ spinifex/migrate/version_readers.go | 128 +++ spinifex/services/nats/nats_test.go | 15 + 25 files changed, 1570 insertions(+), 56 deletions(-) create mode 100644 cmd/spinifex/cmd/admin_upgrade.go create mode 100644 spinifex/migrate/001_nats_token.go create mode 100644 spinifex/migrate/migrate.go create mode 100644 spinifex/migrate/migrate_test.go create mode 100644 spinifex/migrate/version_readers.go diff --git a/README.md b/README.md index 5cc16d892..169c9e4f7 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Spinifex: An Open Source AWS-Compatible Stack for Bare-Metal, Edge, and On-Prem Deployments -(Developer Preview) - Spinifex developed by [Mulga Defense Corporation](https://mulgadc.com/) is an open source infrastructure platform that brings the core services of AWS—like EC2, VPC, EBS, and S3—to environments where running in the cloud isn't an option. Whether you're deploying to edge sites, private data-centers, or need to operate in low-connectivity or highly contested environments, Spinifex gives you AWS-style workflows on your own hardware. +Spinifex developed by [Mulga Defense Corporation](https://mulgadc.com) is an open source infrastructure platform that brings the core services of AWS—like EC2, VPC, EBS, and S3—to environments where running in the cloud isn't an option. Whether you're deploying to edge sites, private data-centers, or need to operate in low-connectivity or highly contested environments, Spinifex gives you AWS-style workflows on your own hardware. ## What is Spinifex? @@ -61,11 +61,11 @@ Installation requires an Ubuntu / Debian system. See the detailed documentation The installation is straightforward to set up and running on a single node for testing purposes. ```bash -curl https://install.mulgadc.com | bash +curl -fsSL https://install.mulgadc.com | bash sudo /usr/local/share/spinifex/setup-ovn.sh --management -sudo spx admin init +sudo spx admin init --node node1 --nodes 1 sudo systemctl start spinifex.target @@ -76,7 +76,7 @@ aws ec2 describe-instance-types ### Development Setup -For a complete development environment see the [Source Install](https://docs.mulgadc.com/docs/source-install/) documentation +For a complete development environment see the [Source Install](https://docs.mulgadc.com/docs/install-source) documentation ### Component Repositories diff --git a/cmd/spinifex/cmd/admin.go b/cmd/spinifex/cmd/admin.go index 2a913894a..d3990ad35 100644 --- a/cmd/spinifex/cmd/admin.go +++ b/cmd/spinifex/cmd/admin.go @@ -217,6 +217,9 @@ func init() { adminCmd.AddCommand(certCmd) certCmd.AddCommand(certRenewCmd) + + adminCmd.AddCommand(upgradeCmd) + upgradeCmd.Flags().Bool("yes", false, "Apply migrations without prompting") certRenewCmd.Flags().StringSlice("extra-ip", nil, "Additional IP addresses to include in SANs") certRenewCmd.Flags().StringSlice("extra-dns", nil, "Additional DNS names to include in SANs") accountCreateCmd.Flags().String("name", "", "Account name (required)") diff --git a/cmd/spinifex/cmd/admin_upgrade.go b/cmd/spinifex/cmd/admin_upgrade.go new file mode 100644 index 000000000..4c84ce17c --- /dev/null +++ b/cmd/spinifex/cmd/admin_upgrade.go @@ -0,0 +1,92 @@ +package cmd + +import ( + "bufio" + "fmt" + "os" + "strings" + + "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/spf13/cobra" +) + +var upgradeCmd = &cobra.Command{ + Use: "upgrade", + Short: "Apply pending config migrations", + Long: `Apply pending configuration file migrations for upgrades between Spinifex versions. + +When run interactively (without --yes), shows pending migrations and prompts +for confirmation. When called from setup.sh with --yes, applies immediately. + +Operators can skip migrations during install by setting INSTALL_SPINIFEX_SKIP_MIGRATE=1, +then run 'spx admin upgrade' manually to review and apply.`, + Run: runAdminUpgrade, +} + +func runAdminUpgrade(cmd *cobra.Command, _ []string) { + configDir, _ := cmd.Root().Flags().GetString("config-dir") + dataDir, _ := cmd.Root().Flags().GetString("spinifex-dir") + yes, _ := cmd.Flags().GetBool("yes") + + // Check that the installation exists. + spinifexToml := configDir + "/spinifex.toml" + if _, err := os.Stat(spinifexToml); os.IsNotExist(err) { + fmt.Fprintf(os.Stderr, "No Spinifex installation found at %s\nRun 'spx admin init' first.\n", configDir) + os.Exit(1) + } + + // Show current versions. + versions, err := migrate.DefaultRegistry.ConfigVersions(configDir) + if err != nil { + fmt.Fprintf(os.Stderr, "Error reading config versions: %v\n", err) + os.Exit(1) + } + + fmt.Println("Reading current config versions...") + for name, v := range versions { + if v == 0 { + fmt.Printf(" %-20s %d (no version marker)\n", name+":", v) + } else { + fmt.Printf(" %-20s %d\n", name+":", v) + } + } + fmt.Println() + + // Check for pending migrations. + pending, err := migrate.DefaultRegistry.PendingConfig(configDir) + if err != nil { + fmt.Fprintf(os.Stderr, "Error checking pending migrations: %v\n", err) + os.Exit(1) + } + + if len(pending) == 0 { + fmt.Println("No pending config migrations.") + return + } + + fmt.Println("Pending config migrations:") + for _, p := range pending { + fmt.Printf(" [config] %s %d → %d: %s\n", p.Target, p.FromVersion, p.ToVersion, p.Description) + } + fmt.Println() + + // Prompt for confirmation unless --yes. + if !yes { + fmt.Print("Apply? [y/N] ") + reader := bufio.NewReader(os.Stdin) + answer, _ := reader.ReadString('\n') + answer = strings.TrimSpace(strings.ToLower(answer)) + if answer != "y" && answer != "yes" { + fmt.Println("Aborted.") + return + } + } + + // Apply migrations. + if err := migrate.DefaultRegistry.RunAllConfig(configDir, dataDir); err != nil { + fmt.Fprintf(os.Stderr, "Migration failed: %v\n", err) + os.Exit(1) + } + + fmt.Println("\nDone. Restart services to apply: sudo systemctl restart spinifex.target") +} diff --git a/cmd/spinifex/cmd/templates/nats.conf b/cmd/spinifex/cmd/templates/nats.conf index e2a438ec7..7baec86ae 100644 --- a/cmd/spinifex/cmd/templates/nats.conf +++ b/cmd/spinifex/cmd/templates/nats.conf @@ -1,3 +1,4 @@ +# spinifex-config-version: 1 # NATS Server Configuration # Generated by Spinifex admin init diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index bf5e45880..2eb0389b4 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -72,6 +72,12 @@ Operational commands for inspecting cluster state. These fan out NATS requests t |---------|-------|---------------|-------------|------------|--------| | `spx admin cert renew` | `--extra-ip` (additional IPs for SANs), `--extra-dns` (additional DNS names for SANs) | Existing CA from `spx admin init` | Reads existing CA → regenerates server certificate with all current network interface IPs and machine hostname in SANs → writes new cert. Use after adding a new network interface or changing IP addresses. | 1. Renew with auto-detected IPs
2. Renew with extra IPs
3. Renew with extra DNS names | **DONE** | +### Upgrade Management + +| Command | Flags | Prerequisites | Basic Logic | Test Cases | Status | +|---------|-------|---------------|-------------|------------|--------| +| `spx admin upgrade` | `--yes` (apply without prompting), `--config-dir` (persistent, default: `~/spinifex/config`), `--spinifex-dir` (persistent, default: `~/spinifex`) | Existing installation (`spinifex.toml` must exist in config dir) | Reads current config versions from registry → computes pending config migrations (from→to per target) → prints version summary and pending list → prompts for confirmation unless `--yes` → runs `migrate.DefaultRegistry.RunAllConfig()` to apply migrations to config files. Intended for upgrades between Spinifex versions. Operators can skip migrations during install by setting `INSTALL_SPINIFEX_SKIP_MIGRATE=1`, then run `spx admin upgrade` manually to review and apply. After completion, services must be restarted with `sudo systemctl restart spinifex.target`. Invoked non-interactively by `scripts/setup.sh` with `--yes`. | 1. No installation present (error, suggests `spx admin init`)
2. No pending migrations (prints "No pending config migrations")
3. Pending migrations with interactive confirmation (y/yes applies, anything else aborts)
4. Pending migrations with `--yes` (applies without prompt)
5. Migration failure surfaces error and exits non-zero | **DONE** | + ### Account Management | Command | Flags | Prerequisites | Basic Logic | Test Cases | Status | diff --git a/scripts/setup.sh b/scripts/setup.sh index 89d715a99..48a9fa8a8 100755 --- a/scripts/setup.sh +++ b/scripts/setup.sh @@ -370,6 +370,93 @@ EOF fi } +# --- Fix file ownership for upgrades from v1 --- +fix_file_ownership() { + info "Fixing file ownership for privilege separation..." + + # Per-service data dirs — recursive chown so existing files are accessible + for entry in \ + nats:spinifex-nats \ + predastore:spinifex-storage \ + spinifex:spinifex-daemon \ + viperblock:spinifex-viperblock \ + vpcd:spinifex-vpcd \ + awsgw:spinifex-gw; do + IFS=: read -r dir svc_user <<< "$entry" + if [ -d "/var/lib/spinifex/$dir" ]; then + $SUDO chown -R "$svc_user:$SPINIFEX_GROUP" "/var/lib/spinifex/$dir" \ + || fatal "Failed to set ownership on /var/lib/spinifex/$dir" + fi + done + + # Per-service config dirs — recursive chown + if [ -d /etc/spinifex/nats ]; then + $SUDO chown -R "spinifex-nats:$SPINIFEX_GROUP" /etc/spinifex/nats \ + || fatal "Failed to set ownership on /etc/spinifex/nats" + fi + if [ -d /etc/spinifex/predastore ]; then + $SUDO chown -R "spinifex-storage:$SPINIFEX_GROUP" /etc/spinifex/predastore \ + || fatal "Failed to set ownership on /etc/spinifex/predastore" + fi + + # Shared config files — root:spinifex with per-file modes + for f in spinifex.toml master.key server.key; do + if [ -f "/etc/spinifex/$f" ]; then + $SUDO chown "root:$SPINIFEX_GROUP" "/etc/spinifex/$f" \ + || fatal "Failed to set ownership on /etc/spinifex/$f" + $SUDO chmod 0640 "/etc/spinifex/$f" \ + || fatal "Failed to set permissions on /etc/spinifex/$f" + fi + done + for f in server.pem ca.pem; do + if [ -f "/etc/spinifex/$f" ]; then + $SUDO chown "root:$SPINIFEX_GROUP" "/etc/spinifex/$f" \ + || fatal "Failed to set ownership on /etc/spinifex/$f" + $SUDO chmod 0644 "/etc/spinifex/$f" \ + || fatal "Failed to set permissions on /etc/spinifex/$f" + fi + done + + # CA private key — root-only + if [ -f /etc/spinifex/ca.key ]; then + $SUDO chown root:root /etc/spinifex/ca.key \ + || fatal "Failed to set ownership on /etc/spinifex/ca.key" + $SUDO chmod 0600 /etc/spinifex/ca.key \ + || fatal "Failed to set permissions on /etc/spinifex/ca.key" + fi + + # Shared data dirs — root:spinifex 0770 (daemon + admin CLI write, services read) + for d in images amis volumes state; do + if [ -d "/var/lib/spinifex/$d" ]; then + $SUDO chown -R "root:$SPINIFEX_GROUP" "/var/lib/spinifex/$d" \ + || fatal "Failed to set ownership on /var/lib/spinifex/$d" + $SUDO chmod 0770 "/var/lib/spinifex/$d" \ + || fatal "Failed to set permissions on /var/lib/spinifex/$d" + fi + done + + info "File ownership updated" +} + +# --- Run config migrations --- +run_migrations() { + # Only run if spinifex is already initialized (config exists) + if [ ! -f /etc/spinifex/spinifex.toml ]; then + info "Fresh install detected, skipping migrations" + return + fi + + if [ "${INSTALL_SPINIFEX_SKIP_MIGRATE:-0}" = "1" ]; then + info "INSTALL_SPINIFEX_SKIP_MIGRATE=1, skipping migrations" + info "Run 'spx admin upgrade' manually to apply pending migrations" + return + fi + + info "Running config migrations..." + $SUDO /usr/local/bin/spx admin upgrade --yes \ + || fatal "Config migration failed. See errors above." +} + # --- Install systemd units --- install_systemd() { info "Installing systemd units..." @@ -471,7 +558,9 @@ main() { download_spinifex install_files create_directories + fix_file_ownership install_systemd + run_migrations install_logrotate rm -rf "$SPINIFEX_TMPDIR" restart_if_needed diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 328f12b72..838947f58 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" @@ -83,8 +84,8 @@ func (m *JetStreamManager) InitKVBucket() error { } m.kv = kv - if err := utils.WriteVersion(kv, InstanceStateBucketVersion); err != nil { - return fmt.Errorf("write version to %s: %w", InstanceStateBucket, err) + if err := migrate.DefaultRegistry.RunKV(InstanceStateBucket, kv, InstanceStateBucketVersion); err != nil { + return fmt.Errorf("migrate %s: %w", InstanceStateBucket, err) } return nil } @@ -113,8 +114,8 @@ func (m *JetStreamManager) InitClusterStateBucket() error { } m.clusterKV = kv - if err := utils.WriteVersion(kv, ClusterStateBucketVersion); err != nil { - return fmt.Errorf("write version to %s: %w", ClusterStateBucket, err) + if err := migrate.DefaultRegistry.RunKV(ClusterStateBucket, kv, ClusterStateBucketVersion); err != nil { + return fmt.Errorf("migrate %s: %w", ClusterStateBucket, err) } return nil } @@ -144,8 +145,8 @@ func (m *JetStreamManager) InitTerminatedInstanceBucket() error { } m.terminatedKV = kv - if err := utils.WriteVersion(kv, TerminatedInstanceBucketVersion); err != nil { - return fmt.Errorf("write version to %s: %w", TerminatedInstanceBucket, err) + if err := migrate.DefaultRegistry.RunKV(TerminatedInstanceBucket, kv, TerminatedInstanceBucketVersion); err != nil { + return fmt.Errorf("migrate %s: %w", TerminatedInstanceBucket, err) } return nil } @@ -199,8 +200,9 @@ func (m *JetStreamManager) recoverBucket(cfg *nats.KeyValueConfig, field *nats.K return nil, err } - if err := utils.WriteVersion(kv, version); err != nil { - slog.Error("Failed to write version to recreated bucket", "bucket", cfg.Bucket, "err", err) + if err := migrate.DefaultRegistry.RunKV(cfg.Bucket, kv, version); err != nil { + slog.Error("Failed to run migrations on recreated bucket", "bucket", cfg.Bucket, "err", err) + return nil, fmt.Errorf("migrate recreated bucket %s: %w", cfg.Bucket, err) } *field = kv diff --git a/spinifex/handlers/ec2/account/service_impl.go b/spinifex/handlers/ec2/account/service_impl.go index 50aec2d0f..5f0e1f380 100644 --- a/spinifex/handlers/ec2/account/service_impl.go +++ b/spinifex/handlers/ec2/account/service_impl.go @@ -9,6 +9,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -44,8 +45,8 @@ func NewAccountSettingsServiceImplWithNATS(cfg *config.Config, natsConn *nats.Co if err != nil { return nil, fmt.Errorf("failed to create account settings KV bucket: %w", err) } - if err := utils.WriteVersion(settingsKV, KVBucketAccountSettingsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketAccountSettings, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketAccountSettings, settingsKV, KVBucketAccountSettingsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountSettings, err) } slog.Info("Account settings service initialized with JetStream KV", "bucket", KVBucketAccountSettings) diff --git a/spinifex/handlers/ec2/eigw/service_impl.go b/spinifex/handlers/ec2/eigw/service_impl.go index 03530cb8f..857e998d8 100644 --- a/spinifex/handlers/ec2/eigw/service_impl.go +++ b/spinifex/handlers/ec2/eigw/service_impl.go @@ -14,6 +14,7 @@ import ( "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/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -53,8 +54,8 @@ func NewEgressOnlyIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEgressOnlyIGW, err) } - if err := utils.WriteVersion(eigwKV, KVBucketEgressOnlyIGWVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketEgressOnlyIGW, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketEgressOnlyIGW, eigwKV, KVBucketEgressOnlyIGWVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketEgressOnlyIGW, err) } // Get or create VPC KV bucket for cross-resource ownership validation diff --git a/spinifex/handlers/ec2/eip/service_impl.go b/spinifex/handlers/ec2/eip/service_impl.go index 529a35d2d..bb72e1a5a 100644 --- a/spinifex/handlers/ec2/eip/service_impl.go +++ b/spinifex/handlers/ec2/eip/service_impl.go @@ -13,6 +13,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/filterutil" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -48,8 +49,8 @@ func NewEIPServiceImpl(natsConn *nats.Conn, externalIPAM *handlers_ec2_vpc.Exter if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketEIPs, err) } - if err := utils.WriteVersion(eipKV, KVBucketEIPsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketEIPs, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketEIPs, eipKV, KVBucketEIPsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketEIPs, err) } slog.Info("EIP service initialized with JetStream KV", "bucket", KVBucketEIPs) diff --git a/spinifex/handlers/ec2/igw/service_impl.go b/spinifex/handlers/ec2/igw/service_impl.go index b2cb26fe3..52b7f7a17 100644 --- a/spinifex/handlers/ec2/igw/service_impl.go +++ b/spinifex/handlers/ec2/igw/service_impl.go @@ -14,6 +14,7 @@ import ( "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/migrate" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" @@ -55,8 +56,8 @@ func NewIGWServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*IGWSer if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketIGW, err) } - if err := utils.WriteVersion(igwKV, KVBucketIGWVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketIGW, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketIGW, igwKV, KVBucketIGWVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketIGW, err) } // Get or create VPC KV bucket for cross-resource ownership validation diff --git a/spinifex/handlers/ec2/natgw/service_impl.go b/spinifex/handlers/ec2/natgw/service_impl.go index ab0c20d5b..72c32c5f2 100644 --- a/spinifex/handlers/ec2/natgw/service_impl.go +++ b/spinifex/handlers/ec2/natgw/service_impl.go @@ -14,6 +14,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/filterutil" handlers_ec2_eip "github.com/mulgadc/spinifex/spinifex/handlers/ec2/eip" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -41,8 +42,8 @@ func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceIm if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketNatGateways, err) } - if err := utils.WriteVersion(natgwKV, KVBucketNatGatewaysVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketNatGateways, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketNatGateways, natgwKV, KVBucketNatGatewaysVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketNatGateways, err) } // Deleted NAT Gateways bucket with 1-hour TTL — keys auto-expire. @@ -59,8 +60,8 @@ func NewNatGatewayServiceImplWithNATS(natsConn *nats.Conn) (*NatGatewayServiceIm return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketDeletedNatGateways, err) } } - if err := utils.WriteVersion(deletedKV, KVBucketDeletedNatGatewaysVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketDeletedNatGateways, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketDeletedNatGateways, deletedKV, KVBucketDeletedNatGatewaysVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketDeletedNatGateways, err) } eipKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_eip.KVBucketEIPs, 10) diff --git a/spinifex/handlers/ec2/placementgroup/service_impl.go b/spinifex/handlers/ec2/placementgroup/service_impl.go index 9f3647486..51342ddf6 100644 --- a/spinifex/handlers/ec2/placementgroup/service_impl.go +++ b/spinifex/handlers/ec2/placementgroup/service_impl.go @@ -13,6 +13,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -57,8 +58,8 @@ func NewPlacementGroupServiceImplWithNATS(cfg *config.Config, natsConn *nats.Con if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketPlacementGroups, err) } - if err := utils.WriteVersion(kv, KVBucketPlacementGroupsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketPlacementGroups, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketPlacementGroups, kv, KVBucketPlacementGroupsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketPlacementGroups, err) } slog.Info("Placement group service initialized with JetStream KV", "bucket", KVBucketPlacementGroups) diff --git a/spinifex/handlers/ec2/routetable/service_impl.go b/spinifex/handlers/ec2/routetable/service_impl.go index 879977f50..b56cb97dc 100644 --- a/spinifex/handlers/ec2/routetable/service_impl.go +++ b/spinifex/handlers/ec2/routetable/service_impl.go @@ -16,6 +16,7 @@ import ( handlers_ec2_igw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/igw" handlers_ec2_natgw "github.com/mulgadc/spinifex/spinifex/handlers/ec2/natgw" handlers_ec2_vpc "github.com/mulgadc/spinifex/spinifex/handlers/ec2/vpc" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -45,8 +46,8 @@ func NewRouteTableServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) ( if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketRouteTables, err) } - if err := utils.WriteVersion(rtbKV, KVBucketRouteTablesVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketRouteTables, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketRouteTables, rtbKV, KVBucketRouteTablesVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketRouteTables, err) } vpcKV, err := utils.GetOrCreateKVBucket(js, handlers_ec2_vpc.KVBucketVPCs, 10) diff --git a/spinifex/handlers/ec2/snapshot/service_impl.go b/spinifex/handlers/ec2/snapshot/service_impl.go index 93c6b49c3..be06563a4 100644 --- a/spinifex/handlers/ec2/snapshot/service_impl.go +++ b/spinifex/handlers/ec2/snapshot/service_impl.go @@ -18,6 +18,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -75,8 +76,8 @@ func NewSnapshotServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*S if err != nil { return nil, nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVolumeSnapshots, err) } - if err := utils.WriteVersion(kv, KVBucketVolumeSnapshotsVersion); err != nil { - return nil, nil, fmt.Errorf("write version to %s: %w", KVBucketVolumeSnapshots, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketVolumeSnapshots, kv, KVBucketVolumeSnapshotsVersion); err != nil { + return nil, nil, fmt.Errorf("migrate %s: %w", KVBucketVolumeSnapshots, err) } slog.Info("Snapshot service initialized with JetStream KV", "bucket", KVBucketVolumeSnapshots) diff --git a/spinifex/handlers/ec2/vpc/external_ipam.go b/spinifex/handlers/ec2/vpc/external_ipam.go index 73cd1e99d..195041803 100644 --- a/spinifex/handlers/ec2/vpc/external_ipam.go +++ b/spinifex/handlers/ec2/vpc/external_ipam.go @@ -8,6 +8,7 @@ import ( "math/big" "net" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -71,8 +72,8 @@ func NewExternalIPAM(js nats.JetStreamContext, pools []ExternalPoolConfig) (*Ext if err != nil { return nil, fmt.Errorf("create external IPAM KV bucket: %w", err) } - if err := utils.WriteVersion(kv, KVBucketExternalIPAMVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketExternalIPAM, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketExternalIPAM, kv, KVBucketExternalIPAMVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketExternalIPAM, err) } ipam := &ExternalIPAM{kv: kv, pools: pools} if err := ipam.initPools(); err != nil { diff --git a/spinifex/handlers/ec2/vpc/ipam.go b/spinifex/handlers/ec2/vpc/ipam.go index f134fdf17..92a40208a 100644 --- a/spinifex/handlers/ec2/vpc/ipam.go +++ b/spinifex/handlers/ec2/vpc/ipam.go @@ -8,6 +8,7 @@ import ( "math/big" "net" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -35,8 +36,8 @@ func NewIPAM(js nats.JetStreamContext) (*IPAM, error) { if err != nil { return nil, fmt.Errorf("failed to create IPAM KV bucket: %w", err) } - if err := utils.WriteVersion(kv, KVBucketIPAMVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketIPAM, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketIPAM, kv, KVBucketIPAMVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketIPAM, err) } return &IPAM{kv: kv}, nil } diff --git a/spinifex/handlers/ec2/vpc/service_impl.go b/spinifex/handlers/ec2/vpc/service_impl.go index 4322d2d8f..31b405ed8 100644 --- a/spinifex/handlers/ec2/vpc/service_impl.go +++ b/spinifex/handlers/ec2/vpc/service_impl.go @@ -15,6 +15,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/awserrors" "github.com/mulgadc/spinifex/spinifex/config" "github.com/mulgadc/spinifex/spinifex/filterutil" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -96,40 +97,40 @@ func NewVPCServiceImplWithNATS(cfg *config.Config, natsConn *nats.Conn) (*VPCSer if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVPCs, err) } - if err := utils.WriteVersion(vpcKV, KVBucketVPCsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketVPCs, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketVPCs, vpcKV, KVBucketVPCsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketVPCs, err) } subnetKV, err := utils.GetOrCreateKVBucket(js, KVBucketSubnets, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSubnets, err) } - if err := utils.WriteVersion(subnetKV, KVBucketSubnetsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketSubnets, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketSubnets, subnetKV, KVBucketSubnetsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketSubnets, err) } vniKV, err := utils.GetOrCreateKVBucket(js, KVBucketVNICounter, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketVNICounter, err) } - if err := utils.WriteVersion(vniKV, KVBucketVNICounterVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketVNICounter, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketVNICounter, vniKV, KVBucketVNICounterVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketVNICounter, err) } eniKV, err := utils.GetOrCreateKVBucket(js, KVBucketENIs, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketENIs, err) } - if err := utils.WriteVersion(eniKV, KVBucketENIsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketENIs, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketENIs, eniKV, KVBucketENIsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketENIs, err) } sgKV, err := utils.GetOrCreateKVBucket(js, KVBucketSecurityGroups, 10) if err != nil { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketSecurityGroups, err) } - if err := utils.WriteVersion(sgKV, KVBucketSecurityGroupsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketSecurityGroups, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketSecurityGroups, sgKV, KVBucketSecurityGroupsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketSecurityGroups, err) } rtbKV, err := utils.GetOrCreateKVBucket(js, "spinifex-vpc-route-tables", 10) diff --git a/spinifex/handlers/elbv2/store.go b/spinifex/handlers/elbv2/store.go index cc5df4993..c52d2fb29 100644 --- a/spinifex/handlers/elbv2/store.go +++ b/spinifex/handlers/elbv2/store.go @@ -7,6 +7,7 @@ import ( "log/slog" "strings" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -38,8 +39,8 @@ func NewStore(nc *nats.Conn) (*Store, error) { return nil, fmt.Errorf("failed to create KV bucket %s: %w", KVBucketELBv2, err) } - if err := utils.WriteVersion(kv, KVBucketELBv2Version); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketELBv2, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketELBv2, kv, KVBucketELBv2Version); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketELBv2, err) } slog.Info("ELBv2 store initialized", "bucket", KVBucketELBv2) diff --git a/spinifex/handlers/iam/service_impl.go b/spinifex/handlers/iam/service_impl.go index 615655ce0..0ccf9e997 100644 --- a/spinifex/handlers/iam/service_impl.go +++ b/spinifex/handlers/iam/service_impl.go @@ -16,6 +16,7 @@ import ( "github.com/aws/aws-sdk-go/service/iam" "github.com/mulgadc/spinifex/spinifex/admin" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/migrate" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -69,40 +70,40 @@ func NewIAMServiceImpl(natsConn *nats.Conn, masterKey []byte, clusterSize int) ( if err != nil { return nil, fmt.Errorf("init users bucket: %w", err) } - if err := utils.WriteVersion(usersBucket, KVBucketUsersVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketUsers, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketUsers, usersBucket, KVBucketUsersVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketUsers, err) } accessKeysBucket, err := getOrCreateBucket(js, KVBucketAccessKeys, 5, replicas) if err != nil { return nil, fmt.Errorf("init access keys bucket: %w", err) } - if err := utils.WriteVersion(accessKeysBucket, KVBucketAccessKeysVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketAccessKeys, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketAccessKeys, accessKeysBucket, KVBucketAccessKeysVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketAccessKeys, err) } policiesBucket, err := getOrCreateBucket(js, KVBucketPolicies, 10, replicas) if err != nil { return nil, fmt.Errorf("init policies bucket: %w", err) } - if err := utils.WriteVersion(policiesBucket, KVBucketPoliciesVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketPolicies, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketPolicies, policiesBucket, KVBucketPoliciesVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketPolicies, err) } accountsBucket, err := getOrCreateBucket(js, KVBucketAccounts, 5, replicas) if err != nil { return nil, fmt.Errorf("init accounts bucket: %w", err) } - if err := utils.WriteVersion(accountsBucket, KVBucketAccountsVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketAccounts, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketAccounts, accountsBucket, KVBucketAccountsVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketAccounts, err) } accountCounterBucket, err := getOrCreateBucket(js, KVBucketAccountCounter, 5, replicas) if err != nil { return nil, fmt.Errorf("init account counter bucket: %w", err) } - if err := utils.WriteVersion(accountCounterBucket, KVBucketAccountCounterVersion); err != nil { - return nil, fmt.Errorf("write version to %s: %w", KVBucketAccountCounter, err) + if err := migrate.DefaultRegistry.RunKV(KVBucketAccountCounter, accountCounterBucket, KVBucketAccountCounterVersion); err != nil { + return nil, fmt.Errorf("migrate %s: %w", KVBucketAccountCounter, err) } decrypter, err := NewDecrypter(masterKey) diff --git a/spinifex/migrate/001_nats_token.go b/spinifex/migrate/001_nats_token.go new file mode 100644 index 000000000..8d86b5f3c --- /dev/null +++ b/spinifex/migrate/001_nats_token.go @@ -0,0 +1,97 @@ +package migrate + +import ( + "fmt" + "os" + "path/filepath" + "regexp" +) + +func init() { + DefaultRegistry.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + + DefaultRegistry.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS auth token and relocate log_file to /var/log/spinifex", + Run: func(ctx ConfigContext) error { + path := filepath.Join(ctx.ConfigDir, "nats", "nats.conf") + content, err := os.ReadFile(path) + if err != nil { + return err + } + + updated := content + var changed bool + + // 1. Uncomment the token line inside the authorization block. + // Matches "# token:" with any leading/interstitial whitespace. + commentedRe := regexp.MustCompile(`(?m)^(\s*)#(\s*token:\s*)`) + uncommentedRe := regexp.MustCompile(`(?m)^\s*token:`) + + if commentedRe.Match(updated) { + updated = commentedRe.ReplaceAll(updated, []byte("${1}${2}")) + changed = true + } else if !uncommentedRe.Match(updated) { + return fmt.Errorf("nats.conf migration failed: token line not found (neither commented nor uncommented)") + } + + // 2. Convert cluster route URLs from nats:// to nats-route://token@. + // Existing deployments have: "nats://host:port" + // Need: "nats-route://TOKEN@host:port" + // Extract the token value from the (now uncommented) token line. + tokenValueRe := regexp.MustCompile(`(?m)^\s*token:\s*"([^"]+)"`) + tokenMatch := tokenValueRe.FindSubmatch(updated) + + if tokenMatch != nil { + token := string(tokenMatch[1]) + // Match route lines: "nats://host:port" (without token already embedded). + routeRe := regexp.MustCompile(`"nats://([^@"]+)"`) + if routeRe.Match(updated) { + replacement := fmt.Sprintf(`"nats-route://%s@${1}"`, token) + updated = routeRe.ReplaceAll(updated, []byte(replacement)) + changed = true + } + } + + // 3. Relocate log_file from data dir to /var/log/spinifex. + // v1.0.2 rendered: log_file: "/var/lib/spinifex/logs/nats.log" + // New location: log_file: "/var/log/spinifex/nats.log" + // The new systemd unit sandbox only permits writes under /var/log/spinifex, + // so leaving the old path causes nats to crash-loop on "permission denied". + oldLogRe := regexp.MustCompile(`(?m)^log_file:\s*"/var/lib/spinifex/logs/nats\.log"`) + newLogLine := `log_file: "/var/log/spinifex/nats.log"` + if oldLogRe.Match(updated) { + updated = oldLogRe.ReplaceAll(updated, []byte(newLogLine)) + changed = true + } + + if !changed { + ctx.Logger.Info("nats.conf: already migrated, skipping") + return nil + } + + info, err := os.Stat(path) + if err != nil { + return err + } + if err := os.WriteFile(path, updated, info.Mode()); err != nil { + return err + } + + // Post-migration validation. + result, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("nats.conf migration: post-migration validation read failed: %w", err) + } + if !uncommentedRe.Match(result) { + return fmt.Errorf("nats.conf migration failed: token line not found after uncommenting") + } + // Validate log_file no longer points at the old data-dir location. + if oldLogRe.Match(result) { + return fmt.Errorf("nats.conf migration failed: log_file still points to /var/lib/spinifex/logs after rewrite") + } + return nil + }, + }) +} diff --git a/spinifex/migrate/migrate.go b/spinifex/migrate/migrate.go new file mode 100644 index 000000000..31ba32321 --- /dev/null +++ b/spinifex/migrate/migrate.go @@ -0,0 +1,330 @@ +package migrate + +import ( + "fmt" + "log/slog" + "os" + "path/filepath" + "sort" + "time" + + "github.com/mulgadc/spinifex/spinifex/utils" + "github.com/nats-io/nats.go" +) + +// KVMigration represents a versioned transformation of KV bucket data. +type KVMigration struct { + FromVersion int + ToVersion int + Description string + Run func(ctx KVContext) error +} + +// KVContext provides KV migration functions access to the bucket being migrated. +type KVContext struct { + KV nats.KeyValue + Logger *slog.Logger +} + +// ConfigMigration represents a versioned transformation of on-disk config files. +type ConfigMigration struct { + FromVersion int + ToVersion int + Description string + Run func(ctx ConfigContext) error +} + +// ConfigContext provides config migration functions access to the filesystem. +type ConfigContext struct { + ConfigDir string // path to /etc/spinifex or equivalent + DataDir string // path to /var/lib/spinifex or equivalent + Logger *slog.Logger +} + +// PendingMigration describes a migration that has not yet been applied. +type PendingMigration struct { + Target string + FromVersion int + ToVersion int + Description string +} + +// configTarget bundles the relative path and version reader for a config file. +type configTarget struct { + path string // relative to configDir (e.g. "nats/nats.conf") + reader ConfigVersionReader +} + +// Registry holds all registered migrations, keyed by target name. +type Registry struct { + kvMigrations map[string][]KVMigration + configMigrations map[string][]ConfigMigration + configTargets map[string]configTarget +} + +// DefaultRegistry is the global migration registry. Migrations self-register +// via init() functions so they are available before any service starts. +var DefaultRegistry = NewRegistry() + +// NewRegistry creates an empty migration registry. +func NewRegistry() *Registry { + return &Registry{ + kvMigrations: make(map[string][]KVMigration), + configMigrations: make(map[string][]ConfigMigration), + configTargets: make(map[string]configTarget), + } +} + +// RegisterKV adds a KV bucket migration. Migrations are kept sorted by FromVersion. +func (r *Registry) RegisterKV(bucket string, m KVMigration) { + r.kvMigrations[bucket] = append(r.kvMigrations[bucket], m) + sort.Slice(r.kvMigrations[bucket], func(i, j int) bool { + return r.kvMigrations[bucket][i].FromVersion < r.kvMigrations[bucket][j].FromVersion + }) +} + +// RegisterConfigTarget registers a config file target with its relative path +// and version reader. Must be called before RegisterConfig for the same target. +func (r *Registry) RegisterConfigTarget(name, relPath string, reader ConfigVersionReader) { + r.configTargets[name] = configTarget{path: relPath, reader: reader} +} + +// RegisterConfig adds a config file migration. Migrations are kept sorted by FromVersion. +func (r *Registry) RegisterConfig(target string, m ConfigMigration) { + r.configMigrations[target] = append(r.configMigrations[target], m) + sort.Slice(r.configMigrations[target], func(i, j int) bool { + return r.configMigrations[target][i].FromVersion < r.configMigrations[target][j].FromVersion + }) +} + +// 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. +// +// When no migrations are registered for a bucket, RunKV stamps targetVersion +// directly — equivalent to a bare utils.WriteVersion. This is the common case +// for buckets that have not yet needed a schema change. +// +// If targetVersion is ahead of the registered migration chain, RunKV returns +// an error — migration authors must register migrations before bumping version +// constants. +func (r *Registry) RunKV(bucket string, kv nats.KeyValue, targetVersion int) error { + current, err := utils.ReadVersion(kv) + if err != nil { + return fmt.Errorf("read version for %s: %w", bucket, err) + } + + if current >= targetVersion { + return nil + } + + all := r.kvMigrations[bucket] + + // Fresh bucket with no registered migrations — stamp directly. This is + // the common first-init path for buckets that have not yet needed a + // schema change. Equivalent to the pre-framework utils.WriteVersion call. + // Note: we do NOT stamp directly for a non-fresh bucket with no migrations + // (e.g. someone bumped a BucketVersion constant from 1 to 2 without adding + // a migration) — that case falls through and errors with "no migrations + // registered" to force the author to add the migration. + if current == 0 && len(all) == 0 { + return utils.WriteVersion(kv, targetVersion) + } + + // Migrations are registered (or an existing bucket has a missing chain) — + // require a complete chain from current to target. + var pending []KVMigration + for _, m := range all { + if m.FromVersion >= current && m.ToVersion <= targetVersion { + pending = append(pending, m) + } + } + + if len(pending) == 0 { + return fmt.Errorf("no migrations registered for %s from version %d to %d", bucket, current, targetVersion) + } + + // Validate contiguous chain. + expected := current + for _, m := range pending { + if m.FromVersion != expected { + return fmt.Errorf("migration chain gap for %s: expected from %d, got from %d", bucket, expected, m.FromVersion) + } + expected = m.ToVersion + } + if expected != targetVersion { + return fmt.Errorf("migration chain for %s ends at version %d, target is %d", bucket, expected, targetVersion) + } + + 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} + if err := m.Run(ctx); err != nil { + return fmt.Errorf("KV migration %s %d→%d failed: %w", bucket, m.FromVersion, m.ToVersion, err) + } + if err := utils.WriteVersion(kv, m.ToVersion); err != nil { + return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, bucket, err) + } + } + + return nil +} + +// RunConfig executes all pending config migrations for a target. +// Before each migration step, a timestamped backup is created. +// Returns an error on failure (setup.sh should abort). +func (r *Registry) RunConfig(target string, configDir, dataDir string) error { + t, ok := r.configTargets[target] + if !ok { + return fmt.Errorf("unknown config target: %s", target) + } + + migrations := r.configMigrations[target] + if len(migrations) == 0 { + return nil + } + + fullPath := filepath.Join(configDir, t.path) + + // If the config file doesn't exist (fresh install), skip migrations. + if _, err := os.Stat(fullPath); err != nil { + if os.IsNotExist(err) { + return nil + } + return fmt.Errorf("stat config file %s: %w", fullPath, err) + } + + current, err := t.reader.ReadVersion(fullPath) + if err != nil { + return fmt.Errorf("read version for %s: %w", target, err) + } + + var pending []ConfigMigration + for _, m := range migrations { + if m.FromVersion >= current { + pending = append(pending, m) + } + } + + if len(pending) == 0 { + return nil + } + + // Validate contiguous chain (same check as RunKV). + expected := current + for _, m := range pending { + if m.FromVersion != expected { + return fmt.Errorf("config migration chain gap for %s: expected from %d, got from %d", target, expected, m.FromVersion) + } + expected = m.ToVersion + } + + logger := slog.Default() + for _, m := range pending { + // Create backup before migration. + backupPath, err := BackupConfig(fullPath, m.FromVersion, m.ToVersion) + if err != nil { + return fmt.Errorf("backup %s before migration %d→%d: %w", target, m.FromVersion, m.ToVersion, err) + } + logger.Info("Created config backup", "target", target, "backup", backupPath) + + logger.Info("Running config migration", "target", target, "from", m.FromVersion, "to", m.ToVersion, "description", m.Description) + ctx := ConfigContext{ConfigDir: configDir, DataDir: dataDir, Logger: logger} + if err := m.Run(ctx); err != nil { + return fmt.Errorf("config migration %s %d→%d failed: %w", target, m.FromVersion, m.ToVersion, err) + } + + // Stamp new version after successful migration. + if err := t.reader.WriteVersion(fullPath, m.ToVersion); err != nil { + return fmt.Errorf("stamp version %d on %s: %w", m.ToVersion, target, err) + } + } + + return nil +} + +// RunAllConfig executes pending config migrations for all registered targets. +// Used by `spx admin upgrade`. +func (r *Registry) RunAllConfig(configDir, dataDir string) error { + for target := range r.configTargets { + if err := r.RunConfig(target, configDir, dataDir); err != nil { + return err + } + } + return nil +} + +// PendingConfig returns all pending config migrations across all targets. +func (r *Registry) PendingConfig(configDir string) ([]PendingMigration, error) { + var result []PendingMigration + for name, t := range r.configTargets { + fullPath := filepath.Join(configDir, t.path) + + if _, err := os.Stat(fullPath); err != nil { + if os.IsNotExist(err) { + continue + } + return nil, fmt.Errorf("stat config file %s: %w", fullPath, err) + } + + current, err := t.reader.ReadVersion(fullPath) + if err != nil { + return nil, fmt.Errorf("read version for %s: %w", name, err) + } + + for _, m := range r.configMigrations[name] { + if m.FromVersion >= current { + result = append(result, PendingMigration{ + Target: name, + FromVersion: m.FromVersion, + ToVersion: m.ToVersion, + Description: m.Description, + }) + } + } + } + return result, nil +} + +// ConfigVersions returns current versions for all registered config targets. +// Targets whose file does not exist are omitted. +func (r *Registry) ConfigVersions(configDir string) (map[string]int, error) { + versions := make(map[string]int) + for name, t := range r.configTargets { + fullPath := filepath.Join(configDir, t.path) + if _, err := os.Stat(fullPath); err != nil { + if os.IsNotExist(err) { + continue + } + return nil, fmt.Errorf("stat config file %s: %w", fullPath, err) + } + v, err := t.reader.ReadVersion(fullPath) + if err != nil { + return nil, fmt.Errorf("read version for %s: %w", name, err) + } + versions[name] = v + } + return versions, nil +} + +// BackupConfig creates a timestamped backup of a config file before migration. +// The backup is stored alongside the original file. +func BackupConfig(path string, fromVersion, toVersion int) (string, error) { + data, err := os.ReadFile(path) + if err != nil { + return "", fmt.Errorf("read file for backup: %w", err) + } + + info, err := os.Stat(path) + if err != nil { + return "", fmt.Errorf("stat file for backup: %w", err) + } + + backupPath := fmt.Sprintf("%s.pre-migrate-%dto%d.%d", path, fromVersion, toVersion, time.Now().Unix()) + if err := os.WriteFile(backupPath, data, info.Mode()); err != nil { + return "", fmt.Errorf("write backup: %w", err) + } + + return backupPath, nil +} diff --git a/spinifex/migrate/migrate_test.go b/spinifex/migrate/migrate_test.go new file mode 100644 index 000000000..8a8fe2066 --- /dev/null +++ b/spinifex/migrate/migrate_test.go @@ -0,0 +1,738 @@ +package migrate + +import ( + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + + "github.com/mulgadc/spinifex/spinifex/testutil" + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// startTestNATS starts an embedded NATS server with JetStream for testing. +func startTestNATS(t *testing.T) (*server.Server, *nats.Conn) { + t.Helper() + ns, nc, _ := testutil.StartTestJetStream(t) + return ns, nc +} + +func createTestBucket(t *testing.T, nc *nats.Conn, name string) nats.KeyValue { + t.Helper() + js, err := nc.JetStream() + require.NoError(t, err) + kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: 1}) + require.NoError(t, err) + return kv +} + +// --- Registry validation tests --- + +func TestRegistry_ValidatesChainNoGaps(t *testing.T) { + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(KVContext) error { return nil }}) + r.RegisterKV("test-bucket", KVMigration{FromVersion: 3, ToVersion: 4, Description: "gap", Run: func(KVContext) error { return nil }}) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + // Stamp version 1 to simulate existing bucket. + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + err = r.RunKV("test-bucket", kv, 4) + assert.Error(t, err) + assert.Contains(t, err.Error(), "gap") +} + +func TestRegistry_RejectsDuplicateVersions(t *testing.T) { + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "first", Run: func(KVContext) error { return nil }}) + r.RegisterKV("test-bucket", KVMigration{FromVersion: 1, ToVersion: 2, Description: "duplicate", Run: func(KVContext) error { return nil }}) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + // Two migrations with the same FromVersion — chain validation will catch this + // because after running first 1→2, the second 1→2 won't match expected=2. + err = r.RunKV("test-bucket", kv, 2) + // Should succeed (first 1→2 runs, second is filtered out since FromVersion < current after first runs). + // Actually with our filtering, both have FromVersion=1 >= current=1 and ToVersion=2 <= target=2, + // so both are in pending. Chain validation: expected=1, first.From=1 ✓, expected=2, second.From=1 ✗. + assert.Error(t, err) + assert.Contains(t, err.Error(), "gap") +} + +// --- RunKV tests --- + +func TestRunKV_NoPendingMigrations_NoOp(t *testing.T) { + r := NewRegistry() + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + + // Stamp version 1 directly. + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + // RunKV with target=1, current=1 → no-op. + err = r.RunKV("test-bucket", kv, 1) + assert.NoError(t, err) +} + +func TestRunKV_FreshBucket_StampsVersion(t *testing.T) { + r := NewRegistry() + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + + // No version set → version 0 (fresh bucket). + err := r.RunKV("test-bucket", kv, 1) + assert.NoError(t, err) + + // Verify version was stamped. + entry, err := kv.Get("_version") + require.NoError(t, err) + assert.Equal(t, "1", string(entry.Value())) +} + +func TestRunKV_ExecutesMigrationsInOrder(t *testing.T) { + var order []int + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 1, ToVersion: 2, Description: "step 1", + Run: func(KVContext) error { order = append(order, 1); return nil }, + }) + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 2, ToVersion: 3, Description: "step 2", + Run: func(KVContext) error { order = append(order, 2); return nil }, + }) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + err = r.RunKV("test-bucket", kv, 3) + assert.NoError(t, err) + assert.Equal(t, []int{1, 2}, order) + + // Verify final version. + entry, err := kv.Get("_version") + require.NoError(t, err) + assert.Equal(t, "3", string(entry.Value())) +} + +func TestRunKV_StampsAfterEachStep(t *testing.T) { + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 1, ToVersion: 2, Description: "step 1", + Run: func(ctx KVContext) error { + // After this runs, version should be stamped to 2 by RunKV. + return nil + }, + }) + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 2, ToVersion: 3, Description: "step 2 fails", + Run: func(KVContext) error { return errors.New("boom") }, + }) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + err = r.RunKV("test-bucket", kv, 3) + assert.Error(t, err) + assert.Contains(t, err.Error(), "boom") + + // Version should be 2 (stamped after step 1, before step 2 failed). + entry, err := kv.Get("_version") + require.NoError(t, err) + assert.Equal(t, "2", string(entry.Value())) +} + +func TestRunKV_StopsOnFailure_VersionNotBumped(t *testing.T) { + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 1, ToVersion: 2, Description: "fails", + Run: func(KVContext) error { return errors.New("migration error") }, + }) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + err = r.RunKV("test-bucket", kv, 2) + assert.Error(t, err) + + // Version remains at 1. + entry, err := kv.Get("_version") + require.NoError(t, err) + assert.Equal(t, "1", string(entry.Value())) +} + +func TestRunKV_RejectsMissingMigration(t *testing.T) { + r := NewRegistry() + // No migrations registered, but bucket is at version 1 and target is 2. + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + err = r.RunKV("test-bucket", kv, 2) + assert.Error(t, err) + assert.Contains(t, err.Error(), "no migrations registered") +} + +func TestRunKV_Idempotent(t *testing.T) { + runCount := 0 + r := NewRegistry() + r.RegisterKV("test-bucket", KVMigration{ + FromVersion: 1, ToVersion: 2, Description: "add field", + Run: func(ctx KVContext) error { + runCount++ + // Idempotent: write a key, re-running writes the same value. + _, err := ctx.KV.PutString("data.key1", `{"field":"value"}`) + return err + }, + }) + + _, nc := startTestNATS(t) + kv := createTestBucket(t, nc, "test-bucket") + _, err := kv.PutString("_version", "1") + require.NoError(t, err) + + // First run. + err = r.RunKV("test-bucket", kv, 2) + assert.NoError(t, err) + assert.Equal(t, 1, runCount) + + // Second run — already at version 2, should be no-op. + err = r.RunKV("test-bucket", kv, 2) + assert.NoError(t, err) + assert.Equal(t, 1, runCount) // Not incremented. +} + +// --- Config version reader tests --- + +func TestTOMLVersionReader_ReadStringVersion(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "spinifex.toml") + require.NoError(t, os.WriteFile(path, []byte(`version = "1.0" +[server] +port = 8080 +`), 0o644)) + + r := &TOMLVersionReader{} + v, err := r.ReadVersion(path) + require.NoError(t, err) + assert.Equal(t, 1, v) +} + +func TestTOMLVersionReader_ReadIntVersion(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + require.NoError(t, os.WriteFile(path, []byte(`version = 2 +[server] +port = 8080 +`), 0o644)) + + r := &TOMLVersionReader{} + v, err := r.ReadVersion(path) + require.NoError(t, err) + assert.Equal(t, 2, v) +} + +func TestTOMLVersionReader_WriteVersion(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.toml") + require.NoError(t, os.WriteFile(path, []byte(`version = "1.0" +[server] +port = 8080 +`), 0o644)) + + r := &TOMLVersionReader{} + require.NoError(t, r.WriteVersion(path, 2)) + + data, err := os.ReadFile(path) + require.NoError(t, err) + assert.Contains(t, string(data), "version = 2") + assert.NotContains(t, string(data), `"1.0"`) + assert.Contains(t, string(data), "port = 8080") // rest preserved +} + +func TestNATSConfVersionReader_ReadNoMarker(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "nats.conf") + require.NoError(t, os.WriteFile(path, []byte(`# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o644)) + + r := &NATSConfVersionReader{} + v, err := r.ReadVersion(path) + require.NoError(t, err) + assert.Equal(t, 0, v) +} + +func TestNATSConfVersionReader_ReadWithMarker(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "nats.conf") + require.NoError(t, os.WriteFile(path, []byte(`# spinifex-config-version: 1 +# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o644)) + + r := &NATSConfVersionReader{} + v, err := r.ReadVersion(path) + require.NoError(t, err) + assert.Equal(t, 1, v) +} + +func TestNATSConfVersionReader_WriteVersion_Prepend(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "nats.conf") + require.NoError(t, os.WriteFile(path, []byte(`# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o644)) + + r := &NATSConfVersionReader{} + require.NoError(t, r.WriteVersion(path, 1)) + + data, err := os.ReadFile(path) + require.NoError(t, err) + lines := strings.Split(string(data), "\n") + assert.Equal(t, "# spinifex-config-version: 1", lines[0]) + assert.Equal(t, "# NATS Server Configuration", lines[1]) +} + +func TestNATSConfVersionReader_WriteVersion_Replace(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "nats.conf") + require.NoError(t, os.WriteFile(path, []byte(`# spinifex-config-version: 1 +# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o644)) + + r := &NATSConfVersionReader{} + require.NoError(t, r.WriteVersion(path, 2)) + + data, err := os.ReadFile(path) + require.NoError(t, err) + lines := strings.Split(string(data), "\n") + assert.Equal(t, "# spinifex-config-version: 2", lines[0]) + assert.Equal(t, "# NATS Server Configuration", lines[1]) +} + +// --- Config migration tests --- + +func TestRunConfig_CreatesBackup(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +authorization { +# token: "testtoken123" +} +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, ToVersion: 1, Description: "test migration", + Run: func(ctx ConfigContext) error { + return nil // no-op for backup test + }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) + + // Check backup exists. + entries, err := os.ReadDir(natsDir) + require.NoError(t, err) + var found bool + for _, e := range entries { + if strings.HasPrefix(e.Name(), "nats.conf.pre-migrate-0to1.") { + found = true + break + } + } + assert.True(t, found, "backup file should exist") +} + +func TestRunConfig_SkipsFreshInstall(t *testing.T) { + dir := t.TempDir() + // No nats.conf file exists — fresh install. + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, ToVersion: 1, Description: "should not run", + Run: func(ctx ConfigContext) error { + t.Fatal("migration should not run on fresh install") + return nil + }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + assert.NoError(t, err) +} + +func TestRunConfig_AlreadyMigrated_NoOp(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# spinifex-config-version: 1 +# NATS Server Configuration +authorization { + token: "testtoken123" +} +`), 0o640)) + + ran := false + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, ToVersion: 1, Description: "already done", + Run: func(ctx ConfigContext) error { ran = true; return nil }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + assert.NoError(t, err) + assert.False(t, ran) +} + +// --- BackupConfig tests --- + +func TestBackupConfig_CreatesTimestampedBackup(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "test.conf") + content := []byte("original content") + require.NoError(t, os.WriteFile(path, content, 0o640)) + + backupPath, err := BackupConfig(path, 0, 1) + require.NoError(t, err) + assert.Contains(t, backupPath, "pre-migrate-0to1") + + backupData, err := os.ReadFile(backupPath) + require.NoError(t, err) + assert.Equal(t, content, backupData) +} + +// --- NATS token migration tests --- + +func TestNATSTokenMigration_UncommentsToken(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +authorization { +# token: "mytoken123" +} +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS authorization token", + Run: DefaultRegistry.configMigrations["nats.conf"][0].Run, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) + + data, err := os.ReadFile(confPath) + require.NoError(t, err) + assert.Contains(t, string(data), ` token: "mytoken123"`) + assert.NotContains(t, string(data), `# token:`) +} + +func TestNATSTokenMigration_ConvertsRoutes(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +cluster { + name: spinifex + listen: 10.11.12.1:4248 + + routes = [ + "nats://10.11.12.2:4248", + "nats://10.11.12.3:4248", + ] +} + +authorization { +# token: "mytoken123" +} +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS authorization token", + Run: DefaultRegistry.configMigrations["nats.conf"][0].Run, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) + + data, err := os.ReadFile(confPath) + require.NoError(t, err) + content := string(data) + assert.Contains(t, content, ` token: "mytoken123"`) + assert.Contains(t, content, `"nats-route://mytoken123@10.11.12.2:4248"`) + assert.Contains(t, content, `"nats-route://mytoken123@10.11.12.3:4248"`) + assert.NotContains(t, content, `"nats://10.11.12`) +} + +func TestNATSTokenMigration_AlreadyUncommented(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +authorization { + token: "mytoken123" +} +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS authorization token", + Run: DefaultRegistry.configMigrations["nats.conf"][0].Run, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) +} + +func TestNATSTokenMigration_RewritesLogFilePath(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + // Simulates a v1.0.2-rendered nats.conf — the log_file path lives under the + // data dir, which the new systemd sandbox blocks writes to. + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +authorization { +# token: "mytoken123" +} +log_file: "/var/lib/spinifex/logs/nats.log" +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS auth token and relocate log_file", + Run: DefaultRegistry.configMigrations["nats.conf"][0].Run, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) + + data, err := os.ReadFile(confPath) + require.NoError(t, err) + content := string(data) + assert.Contains(t, content, `log_file: "/var/log/spinifex/nats.log"`) + assert.NotContains(t, content, `"/var/lib/spinifex/logs/nats.log"`) + // Token rewrite still applied alongside the log_file rewrite. + assert.Contains(t, content, ` token: "mytoken123"`) +} + +func TestNATSTokenMigration_ToleratesWhitespaceVariations(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + + tests := []struct { + name string + input string + wantTok string + }{ + {"tab indented", "authorization {\n\t# token: \"tok1\"\n}", "\t token: \"tok1\""}, + {"extra spaces", "authorization {\n # token: \"tok2\"\n}", " token: \"tok2\""}, + } + + for i, tt := range tests { + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(tt.input), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "test " + strconv.Itoa(i), + Run: DefaultRegistry.configMigrations["nats.conf"][0].Run, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err, tt.name) + + data, err := os.ReadFile(confPath) + require.NoError(t, err) + assert.Contains(t, string(data), tt.wantTok, tt.name) + } +} + +// --- PendingConfig tests --- + +func TestPendingConfig_ReturnsPending(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +authorization { +# token: "testtoken" +} +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "Enable NATS authorization token", + Run: func(ConfigContext) error { return nil }, + }) + + pending, err := r.PendingConfig(dir) + require.NoError(t, err) + require.Len(t, pending, 1) + assert.Equal(t, "nats.conf", pending[0].Target) + assert.Equal(t, 0, pending[0].FromVersion) + assert.Equal(t, 1, pending[0].ToVersion) +} + +func TestPendingConfig_NoPending(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# spinifex-config-version: 1 +# NATS Server Configuration +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "already done", + Run: func(ConfigContext) error { return nil }, + }) + + pending, err := r.PendingConfig(dir) + require.NoError(t, err) + assert.Empty(t, pending) +} + +// --- Config migration error and chain tests --- + +func TestRunConfig_StopsOnFailure_VersionNotBumped(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, + ToVersion: 1, + Description: "fails", + Run: func(ConfigContext) error { return errors.New("migration error") }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + assert.Error(t, err) + assert.Contains(t, err.Error(), "migration error") + + // Version should remain at 0 (no version marker). + reader := &NATSConfVersionReader{} + v, err := reader.ReadVersion(confPath) + require.NoError(t, err) + assert.Equal(t, 0, v) +} + +func TestRunConfig_ExecutesMultiStepChainInOrder(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o640)) + + var order []int + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, ToVersion: 1, Description: "step 1", + Run: func(ConfigContext) error { order = append(order, 1); return nil }, + }) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 1, ToVersion: 2, Description: "step 2", + Run: func(ConfigContext) error { order = append(order, 2); return nil }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + require.NoError(t, err) + assert.Equal(t, []int{1, 2}, order) + + // Verify final version is 2. + reader := &NATSConfVersionReader{} + v, err := reader.ReadVersion(confPath) + require.NoError(t, err) + assert.Equal(t, 2, v) +} + +func TestRunConfig_RejectsChainGap(t *testing.T) { + dir := t.TempDir() + natsDir := filepath.Join(dir, "nats") + require.NoError(t, os.MkdirAll(natsDir, 0o755)) + confPath := filepath.Join(natsDir, "nats.conf") + require.NoError(t, os.WriteFile(confPath, []byte(`# NATS Server Configuration +listen: 0.0.0.0:4222 +`), 0o640)) + + r := NewRegistry() + r.RegisterConfigTarget("nats.conf", "nats/nats.conf", &NATSConfVersionReader{}) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 0, ToVersion: 1, Description: "step 1", + Run: func(ConfigContext) error { return nil }, + }) + r.RegisterConfig("nats.conf", ConfigMigration{ + FromVersion: 3, ToVersion: 4, Description: "gap", + Run: func(ConfigContext) error { return nil }, + }) + + err := r.RunConfig("nats.conf", dir, dir) + assert.Error(t, err) + assert.Contains(t, err.Error(), "gap") +} diff --git a/spinifex/migrate/version_readers.go b/spinifex/migrate/version_readers.go new file mode 100644 index 000000000..d615c19ef --- /dev/null +++ b/spinifex/migrate/version_readers.go @@ -0,0 +1,128 @@ +package migrate + +import ( + "fmt" + "os" + "regexp" + "strconv" + "strings" +) + +// ConfigVersionReader reads and writes the schema version for a config file. +type ConfigVersionReader interface { + ReadVersion(path string) (int, error) + WriteVersion(path string, version int) error +} + +// TOMLVersionReader reads/writes the version field in TOML config files. +// Handles both legacy string values ("1.0" → 1) and integer values (2 → 2). +// New writes always use integers (version = 2). +type TOMLVersionReader struct{} + +var tomlVersionRe = regexp.MustCompile(`(?m)^version\s*=\s*(?:"([^"]+)"|(\d+))\s*$`) + +func (r *TOMLVersionReader) ReadVersion(path string) (int, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + + match := tomlVersionRe.FindSubmatch(data) + if match == nil { + return 0, nil // no version field + } + + var raw string + if len(match[1]) > 0 { + raw = string(match[1]) // quoted string like "1.0" + } else { + raw = string(match[2]) // bare integer like 2 + } + + // Handle "1.0" style — take the integer part before the dot. + if idx := strings.Index(raw, "."); idx >= 0 { + raw = raw[:idx] + } + + v, err := strconv.Atoi(raw) + if err != nil { + return 0, fmt.Errorf("parse TOML version %q: %w", raw, err) + } + return v, nil +} + +func (r *TOMLVersionReader) WriteVersion(path string, version int) error { + data, err := os.ReadFile(path) + if err != nil { + return err + } + + replacement := fmt.Sprintf("version = %d", version) + + if tomlVersionRe.Match(data) { + data = tomlVersionRe.ReplaceAll(data, []byte(replacement)) + } else { + // Prepend version line if not present. + data = append([]byte(replacement+"\n"), data...) + } + + info, err := os.Stat(path) + if err != nil { + return err + } + return os.WriteFile(path, data, info.Mode()) +} + +// NATSConfVersionReader reads/writes the schema version for nats.conf files. +// Version is tracked via a comment on the first line: # spinifex-config-version: 1 +// If absent, the file is treated as version 0 (pre-versioning). +type NATSConfVersionReader struct{} + +var natsVersionRe = regexp.MustCompile(`^#\s*spinifex-config-version:\s*(\d+)\s*$`) + +func (r *NATSConfVersionReader) ReadVersion(path string) (int, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + + // Check first line only. + firstLine := strings.SplitN(string(data), "\n", 2)[0] + match := natsVersionRe.FindStringSubmatch(firstLine) + if match == nil { + return 0, nil // no version marker + } + + v, err := strconv.Atoi(match[1]) + if err != nil { + return 0, fmt.Errorf("parse NATS conf version %q: %w", match[1], err) + } + return v, nil +} + +func (r *NATSConfVersionReader) WriteVersion(path string, version int) error { + data, err := os.ReadFile(path) + if err != nil { + return err + } + + versionLine := fmt.Sprintf("# spinifex-config-version: %d", version) + content := string(data) + + // Check if first line already has a version marker. + lines := strings.SplitN(content, "\n", 2) + if natsVersionRe.MatchString(lines[0]) { + // Replace existing version line. + lines[0] = versionLine + content = strings.Join(lines, "\n") + } else { + // Prepend version line. + content = versionLine + "\n" + content + } + + info, err := os.Stat(path) + if err != nil { + return err + } + return os.WriteFile(path, []byte(content), info.Mode()) +} diff --git a/spinifex/services/nats/nats_test.go b/spinifex/services/nats/nats_test.go index 9f483155b..c84e20969 100644 --- a/spinifex/services/nats/nats_test.go +++ b/spinifex/services/nats/nats_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "runtime" + "strings" "testing" "text/template" "time" @@ -86,3 +87,17 @@ func TestRenderedConfig_EnforcesAuth(t *testing.T) { assert.True(t, nc.IsConnected()) }) } + +// TestRenderedConfig_HasMigrationVersionMarker ensures the nats.conf template +// stamps the current migration version on the first line. Without this marker, +// the migration framework treats fresh installs as version 0 and tries to run +// the 0→1 migration on next upgrade. The marker must be the literal first line +// — NATSConfVersionReader only checks line 0. +func TestRenderedConfig_HasMigrationVersionMarker(t *testing.T) { + raw, err := os.ReadFile(templatePath(t)) + require.NoError(t, err) + + firstLine := strings.SplitN(string(raw), "\n", 2)[0] + assert.Equal(t, "# spinifex-config-version: 1", firstLine, + "nats.conf template must start with the current migration version marker") +}