diff --git a/cmd/eks-gateway-fetch/main.go b/cmd/eks-gateway-fetch/main.go index 7bfc3e619..fec9fc447 100644 --- a/cmd/eks-gateway-fetch/main.go +++ b/cmd/eks-gateway-fetch/main.go @@ -156,9 +156,10 @@ func emitAddonsTSV(out io.Writer, body []byte) error { // recoveryDirective mirrors the gateway's RecoveryDirective wire shape; duplicated // here to keep this tiny VM binary free of the handler package's dependency graph. type recoveryDirective struct { - Epoch int64 `json:"epoch"` - Action string `json:"action"` - Snapshot string `json:"snapshot"` + Epoch int64 `json:"epoch"` + Action string `json:"action"` + Snapshot string `json:"snapshot"` + SnapshotRequired bool `json:"snapshotRequired"` } type internalRecoveryResponse struct { @@ -166,8 +167,10 @@ type internalRecoveryResponse struct { } // emitRecoveryTSV decodes the recovery directive and writes a single -// epoch\taction\tsnapshot line the on-VM k3s-recovery agent consumes without a -// JSON parser. An unset action defaults to "none" (steady state). +// epoch\taction\tsnapshot\tsnapshotRequired line the on-VM k3s-recovery agent +// consumes without a JSON parser. An unset action defaults to "none" (steady +// state); snapshotRequired is 1 when the snapshot MUST restore (fresh DR seed +// aborts boot on fetch failure) else 0. func emitRecoveryTSV(out io.Writer, body []byte) error { var resp internalRecoveryResponse if err := json.Unmarshal(body, &resp); err != nil { @@ -177,7 +180,11 @@ func emitRecoveryTSV(out io.Writer, body []byte) error { if d.Action == "" { d.Action = "none" } - if _, err := fmt.Fprintf(out, "%d\t%s\t%s\n", d.Epoch, d.Action, d.Snapshot); err != nil { + required := "0" + if d.SnapshotRequired { + required = "1" + } + if _, err := fmt.Fprintf(out, "%d\t%s\t%s\t%s\n", d.Epoch, d.Action, d.Snapshot, required); err != nil { return err } return nil diff --git a/cmd/spinifex/cmd/admin_eks.go b/cmd/spinifex/cmd/admin_eks.go new file mode 100644 index 000000000..5320f2c73 --- /dev/null +++ b/cmd/spinifex/cmd/admin_eks.go @@ -0,0 +1,77 @@ +package cmd + +import ( + "context" + "fmt" + "os" + "time" + + "github.com/mulgadc/spinifex/spinifex/admin" + handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" + "github.com/spf13/cobra" +) + +var adminEksCmd = &cobra.Command{ + Use: "eks", + Short: "Manage EKS control-plane disaster recovery", +} + +var adminEksRestoreSnapshotCmd = &cobra.Command{ + Use: "restore-snapshot", + Short: "Rebuild a single-CP cluster's control plane from an etcd snapshot", + Long: `restore-snapshot drives the single-CP total-loss DR path: it launches a fresh +control-plane VM as a cluster-init seed, directs its boot-time recovery agent to +run k3s server --cluster-reset (restoring the given etcd snapshot, or the latest +one found in predastore when --snapshot is omitted), and re-points the cluster +NLB target groups at the new CP. + +Only single-CP clusters are supported. An HA cluster has a potentially +surviving etcd quorum and should be recovered via quorum reformation instead +of this destructive single-node rebuild.`, + Run: runAdminEksRestoreSnapshot, +} + +func init() { + adminCmd.AddCommand(adminEksCmd) + adminEksCmd.AddCommand(adminEksRestoreSnapshotCmd) + + adminEksRestoreSnapshotCmd.Flags().String("cluster", "", "Cluster name to restore (required)") + adminEksRestoreSnapshotCmd.Flags().String("snapshot", "", "Etcd snapshot key to restore; defaults to the latest snapshot found in predastore") + adminEksRestoreSnapshotCmd.Flags().String("account", "", "Account ID that owns the cluster; defaults to the bootstrap account") + _ = adminEksRestoreSnapshotCmd.MarkFlagRequired("cluster") +} + +func runAdminEksRestoreSnapshot(cmd *cobra.Command, _ []string) { + clusterName, _ := cmd.Flags().GetString("cluster") + snapshot, _ := cmd.Flags().GetString("snapshot") + accountID, _ := cmd.Flags().GetString("account") + if accountID == "" { + accountID = admin.DefaultAccountID() + } + + _, nc, err := loadConfigAndConnect() + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + defer nc.Close() + + svc := handlers_eks.NewNATSEKSService(nc) + // The server side launches a VM (30-60s) then fences the old CP with retries, + // so the RPC needs generous headroom over a bare launch. + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second) + defer cancel() + + fmt.Printf("Restoring control plane for cluster %q (account %s)...\n", clusterName, accountID) + out, err := svc.RestoreSnapshot(ctx, &handlers_eks.RestoreSnapshotInput{ + ClusterName: clusterName, + Snapshot: snapshot, + }, accountID) + if err != nil { + fmt.Fprintf(os.Stderr, "Error: %v\n", err) + os.Exit(1) + } + + fmt.Printf("New control plane %s launched (snapshot %s)\n", out.NewInstanceID, out.Snapshot) + fmt.Printf("Status: %s\n", out.Status) +} diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index cf870de24..df0818873 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -171,6 +171,11 @@ read. | `spx admin gpu mig enable` | `--profile ` (required, e.g. `1g.10gb`), `--gpu ` (optional, default: all MIG-capable GPUs) | Checks no GPU instances running (NATS) → discovers MIG-capable GPUs via `gpu.Discover()` (filtered by `--gpu` if set) → enables MIG mode on each target (`gpu.EnableMIGMode`) → lists available profiles (`gpu.ListProfiles`) and validates requested profile name → destroys any existing instances (`gpu.DestroyAllInstances`) → creates new instances filling GPU capacity (`gpu.CreateInstances`) → writes `mig_profile` to `spinifex.toml` via `admin.SetMIGProfile` → sends SIGHUP to `spinifex-daemon`. Must be run directly on the target host. | | `spx admin gpu mig disable` | `--gpu ` (optional, default: all MIG-capable GPUs) | Checks no GPU instances running (NATS) → discovers MIG-capable GPUs via `gpu.Discover()` (filtered by `--gpu` if set) → destroys all GPU instances (`gpu.DestroyAllInstances`) → disables MIG mode (`gpu.DisableMIGMode`) → clears `mig_profile` in `spinifex.toml` via `admin.SetMIGProfile` → sends SIGHUP to `spinifex-daemon`. Must be run directly on the target host. | +### EKS Control-Plane Disaster Recovery + +| Command | Flags | Description | +|---------|-------|-------------| +| `spx admin eks restore-snapshot` | `--cluster` (required), `--snapshot` (optional, defaults to the latest snapshot in predastore), `--account` (optional, defaults to the bootstrap account) | Single-CP total-loss DR path (fail-safe): validates the snapshot exists in predastore BEFORE any mutation (a typo'd/missing key hard-fails, never resets into an empty datastore) → launches a fresh control-plane VM as a cluster-init seed (replaying the persisted create-time launch template) → sets a required-snapshot `RecoveryDirective` (`cluster-reset`) so the boot-time recovery agent aborts rather than resets-into-empty if it cannot fetch the snapshot → persists the replacement in cluster meta BEFORE re-pointing the NLB (so an NLB failure is convergeable by the reconciler, returned as a provisional status, not a hard error) → re-points the cluster NLB's apiserver and konnectivity target groups from the old CP's ENI to the new one → fences the old CP with retries, failing loudly if it cannot be confirmed terminated (split-brain guard). Any failure before the meta commit unwinds the fresh CP (terminate + clear directive) so a re-run does not stack a second resetting control plane. The returned status is provisional — success means the sequence completed, not that etcd is restored and serving; verify cluster health. HA clusters (a spread with a potentially surviving quorum) are rejected — recover those via quorum reformation instead. | ## AWS-Compatible API diff --git a/scripts/images/eks-node/eks-node-role.sh b/scripts/images/eks-node/eks-node-role.sh index 883db78fe..a0a40eab4 100755 --- a/scripts/images/eks-node/eks-node-role.sh +++ b/scripts/images/eks-node/eks-node-role.sh @@ -48,14 +48,26 @@ if [ -z "${ROLE}" ] && [ -f "${AGENT_ENVFILE}" ]; then ROLE="agent" fi +case "${ROLE}" in + server | server-join | agent) ;; + *) + die "SPINIFEX_K3S_ROLE missing or invalid: '${ROLE}'" + ;; +esac + +# Written before the rc-service calls below: mulga-eks-k3s-recovery, started +# inline for server/server-join, reads ROLE_FILE at the top of its own logic +# and must find it already in place, not empty. +mkdir -p "$(dirname "${ROLE_FILE}")" +printf '%s\n' "${ROLE}" > "${ROLE_FILE}" + case "${ROLE}" in server) log "configuring server role" rc-update add eks-token-webhook default - # Pre-k3s etcd recovery: added to the runlevel so it runs before k3s on - # every subsequent boot, but not started now — a fresh cluster has no - # directive and the role file it guards on is written at the end of this - # script. It first acts on the reboot the reconciler triggers. + # Pre-k3s etcd recovery: enabled for later boots and started inline + # now too, so a restore-snapshot DR seed's directive applies before + # k3s starts on this first boot. A plain create has no directive, so it's a no-op. rc-update add mulga-eks-k3s-recovery default rc-update add k3s default # konnectivity-server fronts apiserver egress; every server replica runs @@ -69,6 +81,7 @@ case "${ROLE}" in # (mulga-siv-231.7); server-join nodes do not run it. rc-update add mulga-eks-addon-sync default rc-service eks-token-webhook start + rc-service mulga-eks-k3s-recovery start rc-service k3s start rc-service konnectivity-server start rc-service k3s-first-boot start @@ -79,12 +92,14 @@ case "${ROLE}" in log "configuring server-join role" rc-update add eks-token-webhook default # Pre-k3s etcd recovery (see server role): server-join members are the - # wipe-rejoin followers of a cluster-reset, so they need it too. + # wipe-rejoin followers of a cluster-reset, so they need it too, and for + # the same first-boot reason must be started here, not deferred. rc-update add mulga-eks-k3s-recovery default rc-update add k3s default rc-update add konnectivity-server default rc-update add mulga-eks-state-report default rc-service eks-token-webhook start + rc-service mulga-eks-k3s-recovery start rc-service k3s start rc-service konnectivity-server start rc-service mulga-eks-state-report start @@ -99,7 +114,5 @@ case "${ROLE}" in ;; esac -mkdir -p "$(dirname "${ROLE_FILE}")" -printf '%s\n' "${ROLE}" > "${ROLE_FILE}" rc-update del eks-node-role default 2>/dev/null || true log "role '${ROLE}' configured" diff --git a/scripts/images/eks-node/eks-node-role_test.sh b/scripts/images/eks-node/eks-node-role_test.sh index 6b088bd75..443788a58 100755 --- a/scripts/images/eks-node/eks-node-role_test.sh +++ b/scripts/images/eks-node/eks-node-role_test.sh @@ -65,8 +65,11 @@ check "server: enables k3s-first-boot" "yes" "$(grep -q 'rc-update add k3s-first check "server: starts k3s" "yes" "$(grep -q 'rc-service k3s start' "${CALLS}" && echo yes || echo no)" check "server: enables konnectivity-server" "yes" "$(grep -q 'rc-update add konnectivity-server default' "${CALLS}" && echo yes || echo no)" check "server: enables k3s-recovery" "yes" "$(grep -q 'rc-update add mulga-eks-k3s-recovery default' "${CALLS}" && echo yes || echo no)" -# Recovery is added to the runlevel (pre-k3s on later boots) but not started now. -check "server: does not start k3s-recovery" "no" "$(grep -q 'rc-service mulga-eks-k3s-recovery' "${CALLS}" && echo yes || echo no)" +# Recovery is added to the runlevel (so it also runs pre-k3s on later boots) +# AND started inline now — a restore-snapshot DR seed's directive must apply +# on this very first boot, before k3s starts. +check "server: starts k3s-recovery" "yes" "$(grep -q 'rc-service mulga-eks-k3s-recovery start' "${CALLS}" && echo yes || echo no)" +check "server: starts k3s-recovery before k3s" "yes" "$(awk '/rc-service mulga-eks-k3s-recovery start/{r=NR} /rc-service k3s start/{k=NR} END{print (r>0 && k>0 && r= 7.75 (Alpine 3.21 ships 8.x). URL="${SPINIFEX_PREDASTORE_ENDPOINT%/}/${BUCKET}/${KEY}" -curl -fsSL \ +# Capture curl's own exit: piping straight into logger masks it behind logger's +# (near-always 0), so a failed upload would log "uploaded" and delete the only +# copy — snapshots would silently never land. Keep the local snapshot and exit +# non-zero on failure so cron surfaces it and the next run still has data on disk. +if UPLOAD_OUT=$(curl -fsSL \ --aws-sigv4 "aws:amz:${AWS_REGION:-au-mel-1}:s3" \ --user "${SPINIFEX_PREDASTORE_AKID}:${SPINIFEX_PREDASTORE_SECRET}" \ -H "x-amz-content-sha256: UNSIGNED-PAYLOAD" \ --upload-file "${SNAPSHOT_FILE}" \ - "${URL}" 2>&1 | logger -t mulga-eks-etcd-snapshot - -# Remove the local snapshot to bound disk usage; the canonical copy lives in -# predastore. Restore path (`spx admin eks restore-snapshot`) pulls from there. -rm -f "${SNAPSHOT_FILE}" -logger -t mulga-eks-etcd-snapshot "uploaded ${KEY}" + "${URL}" 2>&1); then + [ -n "${UPLOAD_OUT}" ] && printf '%s\n' "${UPLOAD_OUT}" | logger -t mulga-eks-etcd-snapshot + # Remove the local snapshot only after a confirmed upload to bound disk usage; + # the canonical copy lives in predastore (`spx admin eks restore-snapshot`). + rm -f "${SNAPSHOT_FILE}" + logger -t mulga-eks-etcd-snapshot "uploaded ${KEY}" +else + UPLOAD_RC=$? + [ -n "${UPLOAD_OUT}" ] && printf '%s\n' "${UPLOAD_OUT}" | logger -t mulga-eks-etcd-snapshot + logger -t mulga-eks-etcd-snapshot "ERROR: upload of ${KEY} failed (curl rc=${UPLOAD_RC}); keeping local snapshot ${SNAPSHOT_FILE}" + exit 1 +fi diff --git a/scripts/images/eks-node/mulga-eks-etcd-snapshot_test.sh b/scripts/images/eks-node/mulga-eks-etcd-snapshot_test.sh index 873bc160f..a3a8988de 100644 --- a/scripts/images/eks-node/mulga-eks-etcd-snapshot_test.sh +++ b/scripts/images/eks-node/mulga-eks-etcd-snapshot_test.sh @@ -29,10 +29,13 @@ done : EOF -# curl stub: record the full argument line (carries --upload-file + URL). +# curl stub: record the full argument line (carries --upload-file + URL). Exits +# non-zero when CURL_FAIL=1 so the harness can exercise the failed-upload path. cat > "${STUBBIN}/curl" <<'EOF' #!/bin/sh echo "curl $*" >> "${CURL_CALLS}" +[ "${CURL_FAIL:-0}" = "1" ] && { echo "curl: (7) connection refused" >&2; exit 7; } +exit 0 EOF # logger stub: swallow. @@ -102,6 +105,24 @@ run server "${WORK}/15min/mulga-eks-etcd-snapshot" [ -s "${WORK}/curl.calls" ] && fail "no-env: must not upload" || pass "no-env: no upload without creds" ENVFILE="${ENVFILE_SAVE}" +# --- Case 5: upload FAILS -> non-zero exit + local snapshot KEPT (not masked) --- +# Set up like run() but expect a non-zero exit and assert the snapshot survives. +K3S_CALLS="${WORK}/k3s.calls"; CURL_CALLS="${WORK}/curl.calls" +SNAPSHOT_DIR="${WORK}/snapshots"; ROLE_FILE="${WORK}/role"; K3S_BIN=k3s +: > "${K3S_CALLS}"; : > "${CURL_CALLS}"; rm -rf "${SNAPSHOT_DIR}" +printf '%s' server > "${ROLE_FILE}" +export K3S_CALLS CURL_CALLS SNAPSHOT_DIR ROLE_FILE K3S_BIN ENVFILE +export TIER=frequent CURL_FAIL=1 +if sh "${WORK}/15min/mulga-eks-etcd-snapshot" /dev/null 2>&1 \ + && pass "upload-fail: local snapshot kept for the next run" \ + || fail "upload-fail: local snapshot deleted despite failed upload" +unset TIER CURL_FAIL + if [ "${FAILS}" -eq 0 ]; then echo "PASS: all mulga-eks-etcd-snapshot cases" exit 0 diff --git a/scripts/images/eks-node/mulga-eks-k3s-recovery.sh b/scripts/images/eks-node/mulga-eks-k3s-recovery.sh index 68cf3757c..0d09cac2c 100755 --- a/scripts/images/eks-node/mulga-eks-k3s-recovery.sh +++ b/scripts/images/eks-node/mulga-eks-k3s-recovery.sh @@ -7,10 +7,10 @@ set -eu # but embedded etcd never reformed quorum after a simultaneous restart). # # The reconciler cannot push into the guest, so recovery is pull-based: this -# fetches a per-member directive epochactionsnapshot through the AWS -# gateway (eks-gateway-fetch -resource recovery), epoch-guarded so a directive -# applies exactly once. Ordered `before k3s` so the etcd datastore is mutated -# while the apiserver is stopped. Actions: +# fetches a per-member directive epochactionsnapshotrequired +# through the AWS gateway (eks-gateway-fetch -resource recovery), epoch-guarded +# so a directive applies exactly once. Ordered `before k3s` so the etcd datastore +# is mutated while the apiserver is stopped. Actions: # none steady state — nothing to do # cluster-reset seed member: `k3s server --cluster-reset` reforms a # single-member etcd from this node's intact data (optionally @@ -18,6 +18,13 @@ set -eu # wipe-rejoin follower: delete the local etcd datastore so k3s re-joins the # reset seed's quorum fresh on the next k3s start # +# required=1 marks a snapshot that MUST restore (fresh DR seed via +# `spx admin eks restore-snapshot`): a fresh seed has no intact local etcd, so a +# cluster-reset without the snapshot would boot an EMPTY cluster. On a fetch +# failure this drops BLOCK_MARKER and aborts boot; k3s.initd refuses to start on +# the marker, so k3s never resets into an empty datastore. The fetch is retried +# on the next boot (BLOCK_MARKER lives in /run tmpfs, cleared on reboot). +# # Env knobs default to production paths; overridable so the logic is unit-testable # without root, IMDS, a gateway, or a real k3s. @@ -25,6 +32,7 @@ ROLE_FILE=${ROLE_FILE:-/etc/spinifex-eks/role} ENVFILE=${ENVFILE:-/etc/spinifex-eks/first-boot.env} SNAPSHOT_ENVFILE=${SNAPSHOT_ENVFILE:-/etc/spinifex-eks/etcd-snapshot.env} EPOCH_FILE=${EPOCH_FILE:-/var/lib/spinifex-eks/recovery.epoch} +BLOCK_MARKER=${BLOCK_MARKER:-/run/spinifex-eks/k3s-restore-block} K3S_BIN=${K3S_BIN:-/usr/local/bin/k3s} K3S_CONFIG=${K3S_CONFIG:-/etc/rancher/k3s/config.yaml} ETCD_DIR=${ETCD_DIR:-/var/lib/rancher/k3s/server/db/etcd} @@ -72,12 +80,26 @@ fetch_snapshot() { [ -n "${SPINIFEX_PREDASTORE_SECRET:-}" ] || return 1 _url="${SPINIFEX_PREDASTORE_ENDPOINT%/}/eks-backups-system/${EKS_ACCOUNT_ID}/${EKS_CLUSTER_NAME}/${_name}" mkdir -p "${SNAPSHOT_DIR}" - curl -fsSL \ + # Capture curl's own exit — piping straight into logger would mask it behind + # logger's (near-always 0), so a failed download would look like success and + # the caller would wrongly cluster-reset from local etcd. + if _out=$(curl -fsSL \ --aws-sigv4 "aws:amz:${AWS_REGION:-au-mel-1}:s3" \ --user "${SPINIFEX_PREDASTORE_AKID}:${SPINIFEX_PREDASTORE_SECRET}" \ -H "x-amz-content-sha256: UNSIGNED-PAYLOAD" \ -o "${SNAPSHOT_DIR}/${_name}" \ - "${_url}" 2>&1 | logger -t "${LOGTAG}" + "${_url}" 2>&1); then + [ -n "${_out}" ] && printf '%s\n' "${_out}" | logger -t "${LOGTAG}" + return 0 + else + # Capture rc as the first else statement: $? read after `fi` would be the + # if's own 0, re-masking the failure we are here to surface. + _rc=$? + [ -n "${_out}" ] && printf '%s\n' "${_out}" | logger -t "${LOGTAG}" + rm -f "${SNAPSHOT_DIR}/${_name}" + log "snapshot ${_name} download failed (curl rc=${_rc})" + return 1 + fi } # Instance ID from IMDS keys the per-member directive. IMDSv2 token if enforced, @@ -109,8 +131,11 @@ fi epoch=$(printf '%s' "${line}" | cut -f1) action=$(printf '%s' "${line}" | cut -f2) snapshot=$(printf '%s' "${line}" | cut -f3) +snapshot_required=$(printf '%s' "${line}" | cut -f4) [ -n "${action}" ] || action=none case "${epoch}" in '' | *[!0-9]*) epoch=0 ;; esac +# Absent 4th field (older gateway) reads as not-required, preserving HA-reform semantics. +case "${snapshot_required}" in 1) snapshot_required=1 ;; *) snapshot_required=0 ;; esac if [ "${action}" = "none" ]; then log "directive action=none (epoch ${epoch}); nothing to do" @@ -136,6 +161,13 @@ case "${action}" in if fetch_snapshot "${snapshot}"; then restore_arg="--cluster-reset-restore-path=${SNAPSHOT_DIR}/${snapshot}" log "cluster-reset restoring snapshot ${snapshot}" + elif [ "${snapshot_required}" = "1" ]; then + # Fresh DR seed: no intact local etcd, so a reset without the snapshot + # would boot an EMPTY cluster. Block k3s and abort; retry next boot. + log "ERROR: required snapshot ${snapshot} could not be fetched; aborting boot to avoid resetting into an EMPTY datastore (retries next boot)" + mkdir -p "$(dirname "${BLOCK_MARKER}")" + : > "${BLOCK_MARKER}" + exit 1 else log "WARN: snapshot ${snapshot} fetch failed; cluster-reset from local etcd only" fi diff --git a/scripts/images/eks-node/mulga-eks-k3s-recovery_test.sh b/scripts/images/eks-node/mulga-eks-k3s-recovery_test.sh index bd7572535..6d97c2c8a 100755 --- a/scripts/images/eks-node/mulga-eks-k3s-recovery_test.sh +++ b/scripts/images/eks-node/mulga-eks-k3s-recovery_test.sh @@ -24,7 +24,9 @@ printf '%s\n' "${DIRECTIVE_LINE}" EOF # curl stub: answers the IMDS token PUT + instance-id GET, and materialises a -# snapshot file for any `-o ` download. Records every call. +# snapshot file for any `-o ` download. Records every call. A snapshot +# download (an `-o` arg) fails with a non-zero exit when CURL_SNAP_FAIL=1, so the +# harness can exercise the fetch-failure path. cat > "${STUBBIN}/curl" <<'EOF' #!/bin/sh echo "curl $*" >> "${CURL_CALLS}" @@ -36,7 +38,10 @@ for a in "$@"; do done prev="" for a in "$@"; do - [ "$prev" = "-o" ] && echo snapdata > "$a" + if [ "$prev" = "-o" ]; then + [ "${CURL_SNAP_FAIL:-0}" = "1" ] && { echo "curl: (22) 404" >&2; exit 22; } + echo snapdata > "$a" + fi prev="$a" done exit 0 @@ -80,7 +85,7 @@ pass() { echo "ok: $*"; } # run : reset capture files + state, then run the script # with every path knob pointed into the temp dir. -run() { +_setup_run() { _role=$1 DIRECTIVE_LINE=$2 K3S_CALLS="${WORK}/k3s.calls" @@ -89,13 +94,29 @@ run() { EPOCH_FILE="${WORK}/recovery.epoch" ETCD_DIR="${WORK}/etcd" SNAPSHOT_DIR="${WORK}/snapshots" + BLOCK_MARKER="${WORK}/run/k3s-restore-block" : > "${K3S_CALLS}" : > "${CURL_CALLS}" + rm -f "${BLOCK_MARKER}" printf '%s' "${_role}" > "${ROLE_FILE}" - export DIRECTIVE_LINE K3S_CALLS CURL_CALLS ROLE_FILE EPOCH_FILE ETCD_DIR SNAPSHOT_DIR + export DIRECTIVE_LINE K3S_CALLS CURL_CALLS ROLE_FILE EPOCH_FILE ETCD_DIR SNAPSHOT_DIR BLOCK_MARKER export ENVFILE SNAPSHOT_ENVFILE export K3S_BIN=k3s FETCH_BIN=eks-gateway-fetch - sh "${SCRIPT}" : expects a zero exit (the steady-state path). +run() { + _setup_run "$1" "$2" + sh "${SCRIPT}" : expects a NON-zero exit (the required-snapshot +# fetch-failure abort). CURL_SNAP_FAIL must be set by the caller. +run_abort() { + _setup_run "$1" "$2" + if sh "${SCRIPT}" restore-path + no block marker --- +rm -f "${WORK}/recovery.epoch" +CURL_SNAP_FAIL=0; export CURL_SNAP_FAIL +run server "5${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}1" +grep -q 'cluster-reset-restore-path=.*etcd-frequent-20260709T010000Z.snap' "${WORK}/k3s.calls" \ + && pass "required-ok: restore-path passed" || fail "required-ok: no restore-path: $(cat "${WORK}/k3s.calls")" +[ -f "${WORK}/run/k3s-restore-block" ] && fail "required-ok: must not block k3s on success" || pass "required-ok: no block marker" + +# --- Case 8: REQUIRED snapshot, fetch FAILS -> abort: block marker, no k3s, no epoch --- +rm -f "${WORK}/recovery.epoch" +CURL_SNAP_FAIL=1; export CURL_SNAP_FAIL +run_abort server "6${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}1" +[ -f "${WORK}/run/k3s-restore-block" ] \ + && pass "required-fail: block marker dropped to stop k3s" || fail "required-fail: no block marker" +[ -s "${WORK}/k3s.calls" ] && fail "required-fail: must NOT cluster-reset into empty datastore" || pass "required-fail: k3s not reset" +[ -f "${WORK}/recovery.epoch" ] && fail "required-fail: epoch must not record (retry next boot)" || pass "required-fail: epoch not recorded" + +# --- Case 9: NON-required snapshot, fetch FAILS -> local reset, no block, epoch recorded --- +rm -f "${WORK}/recovery.epoch" +CURL_SNAP_FAIL=1; export CURL_SNAP_FAIL +run server "7${TAB}cluster-reset${TAB}etcd-frequent-20260709T010000Z.snap${TAB}0" +grep -q 'server --config .* --cluster-reset' "${WORK}/k3s.calls" \ + && pass "optional-fail: falls back to local cluster-reset" || fail "optional-fail: no reset: $(cat "${WORK}/k3s.calls")" +grep -q 'cluster-reset-restore-path' "${WORK}/k3s.calls" \ + && fail "optional-fail: must not pass a restore-path for a failed fetch" || pass "optional-fail: local-data reset only" +[ -f "${WORK}/run/k3s-restore-block" ] && fail "optional-fail: must not block k3s (HA-reform semantics)" || pass "optional-fail: no block marker" +[ "$(cat "${WORK}/recovery.epoch" 2>/dev/null)" = 7 ] \ + && pass "optional-fail: epoch recorded" || fail "optional-fail: epoch not recorded" +CURL_SNAP_FAIL=0; export CURL_SNAP_FAIL + if [ "${FAILS}" -eq 0 ]; then echo "PASS: all mulga-eks-k3s-recovery cases" exit 0 diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index d5e1f458e..28554f7cb 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -975,6 +975,7 @@ func (d *Daemon) subscribeAll() error { natsSub{"eks.ListStagedAddonManifests", d.handleEKSListStagedAddonManifests, "spinifex-workers"}, natsSub{"eks.GetRecoveryDirective", d.handleEKSGetRecoveryDirective, "spinifex-workers"}, natsSub{"eks.SetRecoveryDirective", d.handleEKSSetRecoveryDirective, "spinifex-workers"}, + natsSub{"eks.RestoreSnapshot", d.handleEKSRestoreSnapshot, "spinifex-workers"}, natsSub{"eks.AssociateIdentityProviderConfig", d.handleEKSAssociateIdentityProviderConfig, "spinifex-workers"}, natsSub{"eks.DescribeIdentityProviderConfig", d.handleEKSDescribeIdentityProviderConfig, "spinifex-workers"}, natsSub{"eks.ListIdentityProviderConfigs", d.handleEKSListIdentityProviderConfigs, "spinifex-workers"}, diff --git a/spinifex/daemon/daemon_handlers.go b/spinifex/daemon/daemon_handlers.go index 8944e254e..85119f8c3 100644 --- a/spinifex/daemon/daemon_handlers.go +++ b/spinifex/daemon/daemon_handlers.go @@ -84,6 +84,10 @@ func handleNATSRequest[I any, O any](msg *nats.Msg, serviceFn func(context.Conte } output, err := serviceFn(ctx, input, accountID) if err != nil { + // The error was otherwise only recorded on the OTel span, invisible + // without a trace backend. Log at Error so handler failures show up + // in the journal too. + slog.ErrorContext(ctx, "handleNATSRequest: service call failed", "subject", msg.Subject, "err", err) utils.MarkSpanError(span, err) respondWithError(msg, awserrors.ValidErrorCode(err.Error())) return diff --git a/spinifex/daemon/daemon_handlers_eks.go b/spinifex/daemon/daemon_handlers_eks.go index ce3bde437..845cd574a 100644 --- a/spinifex/daemon/daemon_handlers_eks.go +++ b/spinifex/daemon/daemon_handlers_eks.go @@ -130,6 +130,10 @@ func (d *Daemon) handleEKSSetRecoveryDirective(msg *nats.Msg) { handleNATSRequest(msg, d.eksService.SetRecoveryDirective) } +func (d *Daemon) handleEKSRestoreSnapshot(msg *nats.Msg) { + handleNATSRequest(msg, d.eksService.RestoreSnapshot) +} + // --- OIDC identity-provider configs --- func (d *Daemon) handleEKSAssociateIdentityProviderConfig(msg *nats.Msg) { diff --git a/spinifex/daemon/eks_deps.go b/spinifex/daemon/eks_deps.go index f6f7cd6d5..aca4b7362 100644 --- a/spinifex/daemon/eks_deps.go +++ b/spinifex/daemon/eks_deps.go @@ -8,6 +8,7 @@ import ( handlers_ec2_placementgroup "github.com/mulgadc/spinifex/spinifex/handlers/ec2/placementgroup" handlers_eks "github.com/mulgadc/spinifex/spinifex/handlers/eks" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/objectstore" "log/slog" ) @@ -74,33 +75,37 @@ func (d *Daemon) buildEKSServiceDeps() handlers_eks.EKSServiceDeps { } deps := handlers_eks.EKSServiceDeps{ - Config: d.config, - NATSConn: d.natsConn, - MasterKey: masterKey, - GatewayBaseURL: d.resolveGatewayBaseURL(), - Region: d.config.Region, - HolderID: d.node, - ClusterSize: clusterSize, - InternalSuffix: internalSuffix, - SystemGatewayURL: d.resolveSystemGatewayBaseURL(), - SystemAccessKey: d.config.Predastore.AccessKey, - SystemSecretKey: d.config.Predastore.SecretKey, - GatewayCACert: gatewayCA, - VPCSG: d.vpcService, - VPCK3s: d.vpcService, - VPCSubnet: &daemonEKSSubnetResolver{d: d}, - NLB: d.elbv2Service, - Instance: d, - Image: d.imageService, - EIP: d.eipService, - IGW: d.igwService, - Worker: d, - VPCMgr: d.vpcService, - NATGW: d.natGatewayService, - RouteTable: d.routeTableService, - PlacementGroup: handlers_ec2_placementgroup.NewNATSPlacementGroupService(d.natsConn), - Scheduler: handlers_eks.NewNATSHostScheduler(d.natsConn), - CPControl: d.newEKSCPControl(), + Config: d.config, + NATSConn: d.natsConn, + MasterKey: masterKey, + GatewayBaseURL: d.resolveGatewayBaseURL(), + Region: d.config.Region, + HolderID: d.node, + ClusterSize: clusterSize, + InternalSuffix: internalSuffix, + SystemGatewayURL: d.resolveSystemGatewayBaseURL(), + SystemAccessKey: d.config.Predastore.AccessKey, + SystemSecretKey: d.config.Predastore.SecretKey, + GatewayCACert: gatewayCA, + SystemPredastoreURL: d.resolveSystemPredastoreURL(), + SnapshotStore: objectstore.NewS3ObjectStoreFromConfig( + d.config.Predastore.Host, d.config.Predastore.Region, + d.config.Predastore.AccessKey, d.config.Predastore.SecretKey), + VPCSG: d.vpcService, + VPCK3s: d.vpcService, + VPCSubnet: &daemonEKSSubnetResolver{d: d}, + NLB: d.elbv2Service, + Instance: d, + Image: d.imageService, + EIP: d.eipService, + IGW: d.igwService, + Worker: d, + VPCMgr: d.vpcService, + NATGW: d.natGatewayService, + RouteTable: d.routeTableService, + PlacementGroup: handlers_ec2_placementgroup.NewNATSPlacementGroupService(d.natsConn), + Scheduler: handlers_eks.NewNATSHostScheduler(d.natsConn), + CPControl: d.newEKSCPControl(), } // A KV-backed IAM service (sharing the gateway's buckets over NATS) lets EKS @@ -176,3 +181,25 @@ func (d *Daemon) resolveSystemGatewayBaseURL() string { } return "https://" + net.JoinHostPort(d.mgmtBridgeIP, gatewayPort) } + +// resolveSystemPredastoreURL derives the predastore URL a system instance (the +// K3s server VM) fetches/PUTs etcd snapshots to. Same mgmt-bridge-only +// reachability constraint as resolveSystemGatewayBaseURL: config.Predastore.Host +// is rewritten to 127.0.0.1 for the local daemon, which a guest cannot dial, so +// the bridge IP is substituted with the configured port. Falls back to the +// configured Predastore.Host (scheme-prefixed) when no mgmt bridge exists. +func (d *Daemon) resolveSystemPredastoreURL() string { + predastorePort := "8443" + if d.config.Predastore.Host != "" { + if _, port, splitErr := net.SplitHostPort(d.config.Predastore.Host); splitErr == nil && port != "" { + predastorePort = port + } + } + if d.mgmtBridgeIP == "" { + if d.config.Predastore.Host == "" { + return "" + } + return "https://" + d.config.Predastore.Host + } + return "https://" + net.JoinHostPort(d.mgmtBridgeIP, predastorePort) +} diff --git a/spinifex/daemon/eks_deps_test.go b/spinifex/daemon/eks_deps_test.go index 3b5499440..8a60baadf 100644 --- a/spinifex/daemon/eks_deps_test.go +++ b/spinifex/daemon/eks_deps_test.go @@ -84,3 +84,74 @@ func TestResolveGatewayHost(t *testing.T) { }) } } + +// TestResolveSystemPredastoreURL covers the mgmt-bridge-present, +// mgmt-bridge-absent, and no-config branches: a guest system VM (the K3s +// server) can only reach predastore via the mgmt bridge, so the local +// loopback-rewritten Predastore.Host must never leak into the URL handed to +// a guest. +func TestResolveSystemPredastoreURL(t *testing.T) { + tests := []struct { + name string + predastoreHost string + mgmtBridge string + want string + }{ + { + name: "mgmt_bridge_present_uses_bridge_ip_and_configured_port", + predastoreHost: "127.0.0.1:9443", + mgmtBridge: "10.15.8.1", + want: "https://10.15.8.1:9443", + }, + { + name: "mgmt_bridge_present_defaults_port_when_host_has_none", + predastoreHost: "predastore-host-no-port", + mgmtBridge: "10.15.8.1", + want: "https://10.15.8.1:8443", + }, + { + name: "no_mgmt_bridge_falls_back_to_configured_host", + predastoreHost: "predastore.internal:8443", + want: "https://predastore.internal:8443", + }, + { + name: "no_mgmt_bridge_no_host_returns_empty", + want: "", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + d := &Daemon{ + mgmtBridgeIP: tc.mgmtBridge, + config: &config.Config{ + Predastore: config.PredastoreConfig{Host: tc.predastoreHost}, + }, + } + assert.Equal(t, tc.want, d.resolveSystemPredastoreURL()) + }) + } +} + +// TestBuildEKSServiceDeps pins that the mgmt-bridge-derived predastore URL and +// a non-nil snapshot object store actually reach EKSServiceDeps — RestoreSnapshot +// silently can't resolve "latest snapshot" without SnapshotStore, and CP launch +// silently can't write etcd-snapshot.env without SystemPredastoreURL. +func TestBuildEKSServiceDeps(t *testing.T) { + d := &Daemon{ + mgmtBridgeIP: "10.15.8.1", + config: &config.Config{ + Predastore: config.PredastoreConfig{ + Host: "127.0.0.1:8443", + AccessKey: "AKIAPREDASTORE", + SecretKey: "pred-s3cr3t", + Region: "us-east-1", + }, + }, + } + + deps := d.buildEKSServiceDeps() + + assert.Equal(t, "https://10.15.8.1:8443", deps.SystemPredastoreURL) + assert.NotNil(t, deps.SnapshotStore) +} diff --git a/spinifex/handlers/eks/cluster_reconciler.go b/spinifex/handlers/eks/cluster_reconciler.go index a212c9c45..cd3f786bc 100644 --- a/spinifex/handlers/eks/cluster_reconciler.go +++ b/spinifex/handlers/eks/cluster_reconciler.go @@ -684,7 +684,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu // consuming an attempt (stopping without the cluster-reset directive would just // reproduce the wedge); a follower failure logs and continues (that member stays // wedged, retried next attempt). - if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, seed.InstanceID, RecoveryActionClusterReset, ""); err != nil { + if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, seed.InstanceID, RecoveryActionClusterReset, "", false); err != nil { slog.Warn("ClusterReconciler: set cluster-reset directive failed", "cluster", r.clusterName, "seed", seed.InstanceID, "err", err) return @@ -692,7 +692,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu r.resetAttempts++ r.resetIssued = true for _, n := range members[1:] { - if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionWipeRejoin, ""); err != nil { + if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionWipeRejoin, "", false); err != nil { slog.Warn("ClusterReconciler: set wipe-rejoin directive failed", "cluster", r.clusterName, "instanceId", n.InstanceID, "err", err) } @@ -727,7 +727,7 @@ func (r *ClusterReconciler) maybeReformEtcdQuorum(ctx context.Context, meta *Clu // still prevents re-application of the already-applied directive. func (r *ClusterReconciler) clearRecoveryDirectives(meta *ClusterMeta) { for _, n := range cpMemberNodes(meta) { - if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionNone, ""); err != nil { + if _, err := StoreRecoveryDirective(r.acctKV, r.clusterName, n.InstanceID, RecoveryActionNone, "", false); err != nil { slog.Warn("ClusterReconciler: clear recovery directive failed", "cluster", r.clusterName, "instanceId", n.InstanceID, "err", err) } diff --git a/spinifex/handlers/eks/delete_cluster_test.go b/spinifex/handlers/eks/delete_cluster_test.go index 34332b57b..9ee70fc14 100644 --- a/spinifex/handlers/eks/delete_cluster_test.go +++ b/spinifex/handlers/eks/delete_cluster_test.go @@ -73,30 +73,31 @@ func newEKSServiceFixture(t *testing.T) *eksServiceFixture { rt := newFakeRouteTableProvisioner() svc, err := NewEKSServiceImpl(EKSServiceDeps{ - NATSConn: nc, - MasterKey: bootstrapTestMasterKey, - GatewayBaseURL: "https://gw.local:9999", - Region: "us-east-1", - HolderID: "node-1", - SystemGatewayURL: "https://gw.local:9999", - SystemAccessKey: "AKIAEXAMPLE", - SystemSecretKey: "s3cr3t-key", - GatewayCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", - VPCSG: sg, - VPCK3s: vpc, - VPCSubnet: fakeSubnetResolver{}, - IAM: newFakeEnsurer(), - NLB: nlb, - Instance: inst, - Image: ami, - EIP: eip, - Worker: worker, - VPCMgr: vpcMgr, - IGW: igw, - NATGW: ngw, - RouteTable: rt, - PlacementGroup: &fakePlacer{}, - Scheduler: &fakeHostScheduler{}, + NATSConn: nc, + MasterKey: bootstrapTestMasterKey, + GatewayBaseURL: "https://gw.local:9999", + Region: "us-east-1", + HolderID: "node-1", + SystemGatewayURL: "https://gw.local:9999", + SystemAccessKey: "AKIAEXAMPLE", + SystemSecretKey: "s3cr3t-key", + SystemPredastoreURL: "https://10.15.8.1:8443", + GatewayCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", + VPCSG: sg, + VPCK3s: vpc, + VPCSubnet: fakeSubnetResolver{}, + IAM: newFakeEnsurer(), + NLB: nlb, + Instance: inst, + Image: ami, + EIP: eip, + Worker: worker, + VPCMgr: vpcMgr, + IGW: igw, + NATGW: ngw, + RouteTable: rt, + PlacementGroup: &fakePlacer{}, + Scheduler: &fakeHostScheduler{}, }) require.NoError(t, err) // Tight Ready-gating knobs so nodegroup launches resolve in tests without the diff --git a/spinifex/handlers/eks/k3s_ha_control_plane.go b/spinifex/handlers/eks/k3s_ha_control_plane.go index 9edfb98c3..d50175c37 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane.go +++ b/spinifex/handlers/eks/k3s_ha_control_plane.go @@ -162,6 +162,8 @@ func (s *EKSServiceImpl) ProvisionReplacementCP(ctx context.Context, req Replace in.AccessKey = s.deps.SystemAccessKey in.SecretKey = s.deps.SystemSecretKey } + in.PredastoreAccessKey = s.deps.SystemAccessKey + in.PredastoreSecretKey = s.deps.SystemSecretKey instanceType := in.InstanceType if instanceType == "" { @@ -182,6 +184,65 @@ func (s *EKSServiceImpl) ProvisionReplacementCP(ctx context.Context, req Replace return controlPlaneNode(target, out), nil } +// FreshCPRequest carries everything ProvisionFreshControlPlane needs to launch a +// brand-new single-node control plane from a persisted template — the no-survivor +// case, where there is no live quorum member to join. +type FreshCPRequest struct { + AccountID string + ClusterName string + Template *K3sServerInput + ExcludeHosts []string +} + +// ProvisionFreshControlPlane launches a single control-plane VM as a fresh +// cluster-init seed (ServerURL empty), replaying the create-time template with +// rotating creds re-derived. It is the single-node analogue of +// ProvisionReplacementCP for the case where the entire control plane (VM + +// volume) is gone and there is no surviving etcd quorum to join — the caller +// (restore-snapshot) drives the etcd restore via a RecoveryDirective set after +// launch, not via any state this function knows about. +func (s *EKSServiceImpl) ProvisionFreshControlPlane(ctx context.Context, req FreshCPRequest) (ControlPlaneNode, error) { + if req.Template == nil { + return ControlPlaneNode{}, errors.New("eks: ProvisionFreshControlPlane nil template") + } + + in := *req.Template + in.ServerURL = "" + in.KonnServerCount = 1 + in.PrunePeerIP = "" + + sysAcct := admin.SystemAccountID() + in.IamInstanceProfileArn = "" + in.AccessKey = "" + in.SecretKey = "" + if profileARN := s.ensureCPInstanceProfile(sysAcct); profileARN != "" { + in.IamInstanceProfileArn = profileARN + } else { + in.AccessKey = s.deps.SystemAccessKey + in.SecretKey = s.deps.SystemSecretKey + } + in.PredastoreAccessKey = s.deps.SystemAccessKey + in.PredastoreSecretKey = s.deps.SystemSecretKey + + instanceType := in.InstanceType + if instanceType == "" { + instanceType = defaultK3sServerInstanceType + } + target, err := s.pickReplacementHost(ctx, instanceType, req.ExcludeHosts) + if err != nil { + return ControlPlaneNode{}, err + } + in.TargetNodeID = target + + out, err := LaunchK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, s.deps.Image, in) + if err != nil { + return ControlPlaneNode{}, err + } + slog.Info("ProvisionFreshControlPlane: fresh control plane launched", + "cluster", req.ClusterName, "instanceId", out.InstanceID, "host", target) + return controlPlaneNode(target, out), nil +} + // pickReplacementHost returns a schedulable host for the given instance type that // is not already holding a live control-plane member, preserving one-member-per-host // spread. Errors when every schedulable host is excluded (host permanently down), @@ -191,11 +252,18 @@ func (s *EKSServiceImpl) pickReplacementHost(ctx context.Context, instanceType s for _, h := range exclude { ex[h] = true } - for _, h := range s.deps.Scheduler.SchedulableHosts(ctx, instanceType) { + hosts := s.deps.Scheduler.SchedulableHosts(ctx, instanceType) + for _, h := range hosts { if !ex[h] { return h, nil } } + // On a single-node deployment the sole schedulable host is the old CP's + // own host. placeControlPlane already supports a <2-host single-CP + // topology, so fall back to it when that host is the only one available. + if len(hosts) == 1 { + return hosts[0], nil + } return "", fmt.Errorf("eks: no schedulable host for replacement control plane (excluding %d live-member host(s))", len(exclude)) } diff --git a/spinifex/handlers/eks/k3s_ha_control_plane_test.go b/spinifex/handlers/eks/k3s_ha_control_plane_test.go index 71514c010..742a0f0d2 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane_test.go +++ b/spinifex/handlers/eks/k3s_ha_control_plane_test.go @@ -583,3 +583,75 @@ func TestSpecForSystemType_RejectsNonSystem(t *testing.T) { _, _, ok = instancetypes.SpecForSystemType("sys.nonexistent") assert.False(t, ok, "unknown system size has no spec") } + +func TestProvisionFreshControlPlane_ClusterInitNoJoin(t *testing.T) { + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b"}} + inst := &seqK3sInst{} + svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, inst) + // IAM is unwired here, so the launch re-derives static system creds from + // deps (as ProvisionReplacementCP/CreateCluster do). + svc.deps.SystemAccessKey = "AKIATEST" + svc.deps.SystemSecretKey = "secret" + tmpl := validK3sInput() + + node, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{ + ClusterName: "alpha", + Template: &tmpl, + ExcludeHosts: []string{"node-a"}, + }) + + require.NoError(t, err) + assert.Equal(t, "node-b", node.NodeID, "placed on the one non-excluded host") + assert.Equal(t, "i-node-b", node.InstanceID) + assert.NotEmpty(t, node.ENIID, "fresh CP carries a fresh ENI") + ud := inst.userData["node-b"] + assert.Contains(t, ud, "cluster-init: true", "fresh CP seeds a new single-member etcd, not a join") + assert.NotContains(t, ud, "server: https://", "must not join a (nonexistent) survivor") + assert.Contains(t, ud, "EKS_ACCESS_KEY=AKIATEST", "re-derives system creds like ProvisionReplacementCP") + assert.Contains(t, ud, "SPINIFEX_PREDASTORE_AKID=AKIATEST", "re-derives predastore creds too") +} + +func TestProvisionFreshControlPlane_NoFreeHostErrors(t *testing.T) { + // Two hosts, both excluded: genuinely nothing schedulable. A single + // excluded host is covered separately since that falls back to it + // (single-node topology). + sched := &fakeHostScheduler{hosts: []string{"node-a", "node-b"}} + svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{}) + svc.deps.SystemAccessKey = "AKIATEST" + svc.deps.SystemSecretKey = "secret" + tmpl := validK3sInput() + + _, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{ + ClusterName: "alpha", + Template: &tmpl, + ExcludeHosts: []string{"node-a", "node-b"}, + }) + + require.Error(t, err) + assert.Contains(t, err.Error(), "no schedulable host") +} + +func TestProvisionFreshControlPlane_SingleHostFallsBack(t *testing.T) { + // Single-node deployment: the only schedulable host is also the excluded + // (old CP) host. pickReplacementHost falls back to it rather than erroring. + sched := &fakeHostScheduler{hosts: []string{"node-a"}} + svc := newPlacerService(sched, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{}) + svc.deps.SystemAccessKey = "AKIATEST" + svc.deps.SystemSecretKey = "secret" + tmpl := validK3sInput() + + node, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{ + ClusterName: "alpha", + Template: &tmpl, + ExcludeHosts: []string{"node-a"}, + }) + + require.NoError(t, err) + assert.Equal(t, "node-a", node.NodeID) +} + +func TestProvisionFreshControlPlane_NilTemplateRejected(t *testing.T) { + svc := newPlacerService(&fakeHostScheduler{}, &fakePlacer{}, &seqK3sVPC{}, &seqK3sInst{}) + _, err := svc.ProvisionFreshControlPlane(context.Background(), FreshCPRequest{ClusterName: "alpha"}) + require.Error(t, err, "nil template rejected") +} diff --git a/spinifex/handlers/eks/k3s_server_vm.go b/spinifex/handlers/eks/k3s_server_vm.go index 663dd5e1d..4e3339806 100644 --- a/spinifex/handlers/eks/k3s_server_vm.go +++ b/spinifex/handlers/eks/k3s_server_vm.go @@ -67,6 +67,10 @@ const ( // k3sFirstBootEnvPath is the env file k3s-first-boot.sh sources at boot. k3sFirstBootEnvPath = "/etc/spinifex-eks/first-boot.env" + // k3sSnapshotEnvPath is the env file mulga-eks-etcd-snapshot.sh (cron) and + // mulga-eks-k3s-recovery.sh (boot-time restore) source for predastore creds. + k3sSnapshotEnvPath = "/etc/spinifex-eks/etcd-snapshot.env" + // agentEnvPath is the env file the k3s-agent OpenRC service sources for // K3S_URL/K3S_TOKEN/K3S_NODE_NAME/K3S_NODE_LABEL. agentEnvPath = "/etc/spinifex-eks/agent.env" @@ -170,6 +174,13 @@ type K3sServerInput struct { // EKS_KONNECTIVITY_SERVER_COUNT so the konnectivity-server advertises // --server-count=N and every agent holds a tunnel to every replica (HA-correct). KonnServerCount int `json:"konnServerCount,omitempty"` + // PredastoreEndpoint/PredastoreAccessKey/PredastoreSecretKey are the + // mgmt-bridge-reachable predastore SigV4 creds baked into etcd-snapshot.env. + // The guest snapshot cron and boot-time recovery script have no IMDS-role + // support, so these are always static (unlike AccessKey/SecretKey above). + PredastoreEndpoint string `json:"predastoreEndpoint,omitempty"` + PredastoreAccessKey string `json:"predastoreAccessKey,omitempty"` + PredastoreSecretKey string `json:"predastoreSecretKey,omitempty"` } // K3sServerOutput carries identifiers to persist in ClusterMeta and register with the NLB. @@ -447,6 +458,12 @@ func validateK3sServerInput(in K3sServerInput) error { return errors.New("eks: K3sServerInput AccessKey and SecretKey must both be set or both empty") case strings.TrimSpace(in.GatewayCACert) == "": return errors.New("eks: K3sServerInput empty GatewayCACert") + case in.PredastoreEndpoint == "": + return errors.New("eks: K3sServerInput empty PredastoreEndpoint") + case in.PredastoreAccessKey == "": + return errors.New("eks: K3sServerInput empty PredastoreAccessKey") + case in.PredastoreSecretKey == "": + return errors.New("eks: K3sServerInput empty PredastoreSecretKey") case in.ServerURL != "" && in.JoinToken == "": return errors.New("eks: K3sServerInput join server (ServerURL set) requires JoinToken") } @@ -638,9 +655,22 @@ func buildK3sUserData(in K3sServerInput) string { ) k3sConfig := strings.Join(configLines, "\n") + snapshotEnvBody := strings.Join([]string{ + "EKS_ACCOUNT_ID=" + in.ClusterAccountID, + "EKS_CLUSTER_NAME=" + in.ClusterName, + "SPINIFEX_PREDASTORE_ENDPOINT=" + in.PredastoreEndpoint, + "SPINIFEX_PREDASTORE_AKID=" + in.PredastoreAccessKey, + "SPINIFEX_PREDASTORE_SECRET=" + in.PredastoreSecretKey, + // Without AWS_REGION the snapshot SigV4 PUT signs with the fallback + // region and predastore rejects it with 403. + "AWS_REGION=" + in.Region, + }, "\n") + files := []userDataFile{ // 0600: may contain a system SigV4 secret key (static-cred mode). {Path: k3sFirstBootEnvPath, Perms: "0600", Body: envBody}, + // 0600: carries predastore SigV4 static creds for the snapshot cron + boot-time recovery. + {Path: k3sSnapshotEnvPath, Perms: "0600", Body: snapshotEnvBody}, {Path: k3sOIDCSigningKeyPath, Perms: "0600", Body: strings.TrimRight(in.OIDCPrivateKeyPEM, "\n")}, {Path: k3sOIDCPublicKeyPath, Perms: "0644", Body: strings.TrimRight(in.OIDCPublicKeyPEM, "\n")}, {Path: k3sConfigPath, Perms: "0644", Body: k3sConfig}, diff --git a/spinifex/handlers/eks/k3s_server_vm_test.go b/spinifex/handlers/eks/k3s_server_vm_test.go index 20d565bfb..643926444 100644 --- a/spinifex/handlers/eks/k3s_server_vm_test.go +++ b/spinifex/handlers/eks/k3s_server_vm_test.go @@ -178,6 +178,10 @@ func validK3sInput() K3sServerInput { SecretKey: "s3cr3t-key", GatewayCACert: "-----BEGIN CERTIFICATE-----\nFAKECA\n-----END CERTIFICATE-----\n", JoinToken: "clustertok-deadbeef", + + PredastoreEndpoint: "https://10.15.8.1:8443", + PredastoreAccessKey: "AKIAPREDASTORE", + PredastoreSecretKey: "pred-s3cr3t-key", } } @@ -205,6 +209,9 @@ func TestLaunchK3sServerVM_EmptyInputsRejected(t *testing.T) { {"empty AccessKey", mk(func(in *K3sServerInput) { in.AccessKey = "" })}, {"empty SecretKey", mk(func(in *K3sServerInput) { in.SecretKey = "" })}, {"empty GatewayCACert", mk(func(in *K3sServerInput) { in.GatewayCACert = " \n" })}, + {"empty PredastoreEndpoint", mk(func(in *K3sServerInput) { in.PredastoreEndpoint = "" })}, + {"empty PredastoreAccessKey", mk(func(in *K3sServerInput) { in.PredastoreAccessKey = "" })}, + {"empty PredastoreSecretKey", mk(func(in *K3sServerInput) { in.PredastoreSecretKey = "" })}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -239,6 +246,16 @@ func TestBuildK3sUserData_StaticCredsBakeKeys(t *testing.T) { assert.Contains(t, ud, "EKS_SECRET_KEY=s3cr3t-key") } +func TestBuildK3sUserData_WritesEtcdSnapshotEnv(t *testing.T) { + ud := buildK3sUserData(validK3sInput()) + require.Contains(t, ud, "path: "+k3sSnapshotEnvPath) + assert.Contains(t, ud, "EKS_ACCOUNT_ID=111122223333") + assert.Contains(t, ud, "EKS_CLUSTER_NAME=alpha") + assert.Contains(t, ud, "SPINIFEX_PREDASTORE_ENDPOINT=https://10.15.8.1:8443") + assert.Contains(t, ud, "SPINIFEX_PREDASTORE_AKID=AKIAPREDASTORE") + assert.Contains(t, ud, "SPINIFEX_PREDASTORE_SECRET=pred-s3cr3t-key") +} + func TestBuildK3sUserData_ProfileModeOmitsKeys(t *testing.T) { in := validK3sInput() in.AccessKey = "" diff --git a/spinifex/handlers/eks/recovery_directive.go b/spinifex/handlers/eks/recovery_directive.go index e45f8d49f..0550b4ebf 100644 --- a/spinifex/handlers/eks/recovery_directive.go +++ b/spinifex/handlers/eks/recovery_directive.go @@ -30,10 +30,15 @@ const ( // RecoveryDirective is the per-member recovery instruction. Epoch increases on // each set so the guest applies a directive at most once: it records the // last-applied epoch and ignores an equal-or-older one across reboots. +// SnapshotRequired marks a snapshot the guest MUST restore (fresh DR seed): if +// the fetch fails the guest aborts boot rather than cluster-reset into an empty +// datastore. It stays false for the HA-reform path, whose seed cluster-resets +// from intact local etcd when no snapshot is present. type RecoveryDirective struct { - Epoch int64 `json:"epoch"` - Action RecoveryAction `json:"action"` - Snapshot string `json:"snapshot,omitempty"` + Epoch int64 `json:"epoch"` + Action RecoveryAction `json:"action"` + Snapshot string `json:"snapshot,omitempty"` + SnapshotRequired bool `json:"snapshotRequired,omitempty"` } // GetRecoveryDirectiveInput names the cluster + member whose directive to read. @@ -50,10 +55,11 @@ type GetRecoveryDirectiveOutput struct { // SetRecoveryDirectiveInput sets a member's directive; Epoch is assigned by the // store (previous+1), so callers supply only the action + optional snapshot key. type SetRecoveryDirectiveInput struct { - ClusterName string `json:"clusterName"` - InstanceID string `json:"instanceId"` - Action RecoveryAction `json:"action"` - Snapshot string `json:"snapshot,omitempty"` + ClusterName string `json:"clusterName"` + InstanceID string `json:"instanceId"` + Action RecoveryAction `json:"action"` + Snapshot string `json:"snapshot,omitempty"` + SnapshotRequired bool `json:"snapshotRequired,omitempty"` } // SetRecoveryDirectiveOutput returns the stored directive incl. the assigned epoch. @@ -83,12 +89,12 @@ func LoadRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string) (Re // StoreRecoveryDirective writes a member's directive with epoch = previous+1 and // returns the stored value. Used by the reconciler escalation and the operator CLI. -func StoreRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string, action RecoveryAction, snapshot string) (RecoveryDirective, error) { +func StoreRecoveryDirective(acctKV nats.KeyValue, cluster, instanceID string, action RecoveryAction, snapshot string, snapshotRequired bool) (RecoveryDirective, error) { prev, err := LoadRecoveryDirective(acctKV, cluster, instanceID) if err != nil { return RecoveryDirective{}, err } - next := RecoveryDirective{Epoch: prev.Epoch + 1, Action: action, Snapshot: snapshot} + next := RecoveryDirective{Epoch: prev.Epoch + 1, Action: action, Snapshot: snapshot, SnapshotRequired: snapshotRequired} data, err := json.Marshal(&next) if err != nil { return RecoveryDirective{}, fmt.Errorf("marshal recovery directive: %w", err) @@ -130,7 +136,7 @@ func (s *EKSServiceImpl) SetRecoveryDirective(_ context.Context, input *SetRecov if err != nil { return nil, err } - d, err := StoreRecoveryDirective(acctKV, input.ClusterName, input.InstanceID, input.Action, input.Snapshot) + d, err := StoreRecoveryDirective(acctKV, input.ClusterName, input.InstanceID, input.Action, input.Snapshot, input.SnapshotRequired) if err != nil { return nil, err } diff --git a/spinifex/handlers/eks/recovery_directive_test.go b/spinifex/handlers/eks/recovery_directive_test.go index c0f99a664..91d000d01 100644 --- a/spinifex/handlers/eks/recovery_directive_test.go +++ b/spinifex/handlers/eks/recovery_directive_test.go @@ -26,11 +26,11 @@ func TestStoreRecoveryDirective_EpochIncrements(t *testing.T) { acctKV, err := GetOrCreateAccountBucket(js, testAccountID, 1) require.NoError(t, err) - first, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionClusterReset, "") + first, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionClusterReset, "", false) require.NoError(t, err) assert.Equal(t, int64(1), first.Epoch, "first set starts at epoch 1") - second, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionWipeRejoin, "snap.snap") + second, err := StoreRecoveryDirective(acctKV, "alpha", "i-0", RecoveryActionWipeRejoin, "snap.snap", false) require.NoError(t, err) assert.Equal(t, int64(2), second.Epoch, "each set advances the epoch") @@ -41,6 +41,21 @@ func TestStoreRecoveryDirective_EpochIncrements(t *testing.T) { assert.Equal(t, int64(2), got.Epoch) } +func TestStoreRecoveryDirective_SnapshotRequiredRoundTrips(t *testing.T) { + _, _, js := testutil.StartTestJetStream(t) + acctKV, err := GetOrCreateAccountBucket(js, testAccountID, 1) + require.NoError(t, err) + + stored, err := StoreRecoveryDirective(acctKV, "alpha", "i-seed", RecoveryActionClusterReset, "etcd-frequent-20260709T010000Z.snap", true) + require.NoError(t, err) + assert.True(t, stored.SnapshotRequired, "a required snapshot is recorded so the guest aborts on fetch failure") + + got, err := LoadRecoveryDirective(acctKV, "alpha", "i-seed") + require.NoError(t, err) + assert.True(t, got.SnapshotRequired, "SnapshotRequired survives the KV round-trip") + assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", got.Snapshot) +} + func TestSetRecoveryDirective_RejectsBadInput(t *testing.T) { s := &EKSServiceImpl{} diff --git a/spinifex/handlers/eks/restore_snapshot.go b/spinifex/handlers/eks/restore_snapshot.go new file mode 100644 index 000000000..14c47c881 --- /dev/null +++ b/spinifex/handlers/eks/restore_snapshot.go @@ -0,0 +1,351 @@ +package handlers_eks + +import ( + "context" + "errors" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/admin" + "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/nats-io/nats.go" +) + +// eksBackupsBucket is the predastore bucket the guest etcd-snapshot cron PUTs +// to and this restore path lists/reads from. Keys are +// {accountID}/{clusterName}/etcd-{tier}-{timestamp}.snap. +const eksBackupsBucket = "eks-backups-system" + +// Old-CP fence retry budget. A wedged-but-alive old CP left running while the +// new one takes over is split-brain (two live etcds), so terminate is retried +// before the restore is declared complete. +const ( + restoreFenceAttempts = 3 + restoreFenceDelay = 2 * time.Second +) + +// RestoreSnapshotInput names the cluster to restore and, optionally, the exact +// snapshot key (basename) to apply. Empty Snapshot resolves the latest +// frequent-tier snapshot, falling back to the latest of any tier. +type RestoreSnapshotInput struct { + ClusterName string `json:"clusterName"` + Snapshot string `json:"snapshot,omitempty"` +} + +// RestoreSnapshotOutput reports the fresh control-plane instance and the +// snapshot key its boot-time recovery agent will apply. Status is deliberately +// provisional: a successful return means the new CP was launched, its restore +// directive queued, meta re-pointed and the old CP fenced — NOT that etcd is +// restored and the apiserver is serving. That is observed asynchronously by the +// reconciler, so the caller must verify cluster health before relying on it. +type RestoreSnapshotOutput struct { + ClusterName string `json:"clusterName"` + NewInstanceID string `json:"newInstanceId"` + Snapshot string `json:"snapshot"` + Status string `json:"status"` +} + +// etcdSnapshotKey is a parsed "etcd-{tier}-{timestamp}.snap" object basename. +type etcdSnapshotKey struct { + tier string + timestamp string + basename string +} + +// parseEtcdSnapshotKey parses a snapshot object basename (not the full S3 key) +// into tier + timestamp. The timestamp (YYYYMMDDTHHMMSSZ) is fixed-width, so +// lexicographic comparison of the timestamp alone is chronological — tier names +// vary in length, so comparing the whole basename would not be. +func parseEtcdSnapshotKey(basename string) (etcdSnapshotKey, bool) { + const prefix, suffix = "etcd-", ".snap" + if !strings.HasPrefix(basename, prefix) || !strings.HasSuffix(basename, suffix) { + return etcdSnapshotKey{}, false + } + mid := strings.TrimSuffix(strings.TrimPrefix(basename, prefix), suffix) + i := strings.LastIndex(mid, "-") + if i <= 0 || i == len(mid)-1 { + return etcdSnapshotKey{}, false + } + return etcdSnapshotKey{tier: mid[:i], timestamp: mid[i+1:], basename: basename}, true +} + +// resolveLatestSnapshot lists the cluster's snapshot prefix and returns the +// newest object's basename, preferring the frequent tier (tightest RPO) and +// falling back to the newest snapshot of any tier. +func resolveLatestSnapshot(ctx context.Context, store objectstore.ObjectStore, accountID, clusterName string) (string, error) { + if store == nil { + return "", errors.New("eks: no snapshot store configured; pass --snapshot explicitly") + } + prefix := accountID + "/" + clusterName + "/" + out, err := store.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(eksBackupsBucket), + Prefix: aws.String(prefix), + }) + if err != nil { + return "", fmt.Errorf("list etcd snapshots: %w", err) + } + var bestFrequent, bestAny etcdSnapshotKey + haveFrequent, haveAny := false, false + for _, obj := range out.Contents { + base := strings.TrimPrefix(aws.StringValue(obj.Key), prefix) + parsed, ok := parseEtcdSnapshotKey(base) + if !ok { + continue + } + if !haveAny || parsed.timestamp > bestAny.timestamp { + bestAny, haveAny = parsed, true + } + if parsed.tier == "frequent" && (!haveFrequent || parsed.timestamp > bestFrequent.timestamp) { + bestFrequent, haveFrequent = parsed, true + } + } + switch { + case haveFrequent: + return bestFrequent.basename, nil + case haveAny: + return bestAny.basename, nil + default: + return "", fmt.Errorf("eks: no etcd snapshots found under %s/%s", eksBackupsBucket, prefix) + } +} + +// validateSnapshotExists HEADs the snapshot object before any mutation. An +// explicit --snapshot that is typo'd or expired would otherwise be handed to the +// guest verbatim, its fetch fail, and — without the required-snapshot guard — a +// fresh seed cluster-reset into an EMPTY datastore reported as success. A nil +// store cannot verify, so it is a hard error rather than an optimistic pass. +func validateSnapshotExists(ctx context.Context, store objectstore.ObjectStore, accountID, clusterName, basename string) error { + if store == nil { + return errors.New("eks: no snapshot store configured; cannot verify the snapshot exists before a destructive restore") + } + key := accountID + "/" + clusterName + "/" + basename + if _, err := store.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: aws.String(eksBackupsBucket), + Key: aws.String(key), + }); err != nil { + if objectstore.IsNoSuchKeyError(err) { + return fmt.Errorf("eks: snapshot %q not found under %s/%s/%s/; refusing to restore (a cluster-reset with no snapshot boots an EMPTY cluster)", + basename, eksBackupsBucket, accountID, clusterName) + } + return fmt.Errorf("eks: verify snapshot %q exists: %w", basename, err) + } + return nil +} + +// replaceControlPlaneForRestore overwrites meta's single control-plane member +// with the freshly provisioned node, mirroring the create-path scalar fields +// (service_impl.go CreateCluster). restore-snapshot only ever targets a +// single-CP cluster (isHAControlPlane guards the caller), so this replaces +// wholesale rather than matching a dead instance ID like SwapControlPlaneMember. +func replaceControlPlaneForRestore(kv nats.KeyValue, name string, newNode ControlPlaneNode) error { + return casUpdateMeta(kv, name, func(m *ClusterMeta) bool { + m.ControlPlaneNodes = []ControlPlaneNode{newNode} + m.ControlPlaneInstanceID = newNode.InstanceID + m.ControlPlaneENIID = newNode.ENIID + m.ControlPlaneENIIP = newNode.ENIIP + m.ControlPlaneMgmtIP = newNode.MgmtIP + return true + }) +} + +// unwindFreshCP rolls back a just-launched fresh CP when a step before the meta +// commit fails. It supersedes the recovery directive with a no-op (so a VM that +// boots before terminate lands does not destructively reset) then terminates the +// VM. Both are best-effort and logged: leaving an orphan boot a resetting CP, +// which the operator would then compound by re-running restore into a second one. +func (s *EKSServiceImpl) unwindFreshCP(ctx context.Context, accountID, clusterName string, node ControlPlaneNode) { + if _, err := s.SetRecoveryDirective(ctx, &SetRecoveryDirectiveInput{ + ClusterName: clusterName, + InstanceID: node.InstanceID, + Action: RecoveryActionNone, + }, accountID); err != nil { + slog.WarnContext(ctx, "RestoreSnapshot: clear directive during unwind failed; guest may still reset if it boots", + "cluster", clusterName, "instanceId", node.InstanceID, "err", err) + } + if err := TerminateK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, admin.SystemAccountID(), node.InstanceID, node.ENIID); err != nil { + slog.WarnContext(ctx, "RestoreSnapshot: terminate fresh CP during unwind failed; may orphan a resetting control plane", + "cluster", clusterName, "instanceId", node.InstanceID, "err", err) + } +} + +// confirmOldCPTerminated fences the old control plane: it retries terminate until +// the instance is confirmed gone (TerminateK3sServerVM treats an already-absent +// instance as success, the primary DR case). A persistent failure means a +// wedged-but-alive old CP that would run its etcd alongside the new one — +// split-brain — so the caller must surface it loudly rather than declare success. +func (s *EKSServiceImpl) confirmOldCPTerminated(ctx context.Context, accountID string, node ControlPlaneNode, attempts int, delay time.Duration) error { + if node.InstanceID == "" && node.ENIID == "" { + return nil + } + var lastErr error + for i := range attempts { + if i > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delay): + } + } + if err := TerminateK3sServerVM(ctx, s.deps.VPCK3s, s.deps.Instance, accountID, node.InstanceID, node.ENIID); err != nil { + lastErr = err + slog.WarnContext(ctx, "RestoreSnapshot: old control plane terminate attempt failed", + "instanceId", node.InstanceID, "attempt", i+1, "err", err) + continue + } + return nil + } + return lastErr +} + +// RestoreSnapshot drives the single-CP total-loss DR path: it validates the +// snapshot exists, launches a fresh control-plane VM as a cluster-init seed, +// directs its boot-time recovery agent to `k3s server --cluster-reset` restoring +// that snapshot, persists the new CP into meta, re-points the cluster NLB target +// groups, and fences the old CP. HA clusters are out of scope — they have a +// potentially surviving quorum and should be recovered via quorum reformation, +// not a destructive single-node rebuild. The returned status is provisional: +// success means the sequence completed, not that etcd is restored and serving. +func (s *EKSServiceImpl) RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error) { + if input == nil || input.ClusterName == "" { + return nil, errors.New(awserrors.ErrorInvalidParameterValue) + } + acctKV, err := s.acctKVForCluster(accountID, input.ClusterName) + if err != nil { + return nil, err + } + meta, err := GetClusterMeta(acctKV, input.ClusterName) + if err != nil { + if errors.Is(err, ErrClusterNotFound) { + return nil, errors.New(awserrors.ErrorEKSResourceNotFound) + } + return nil, err + } + if isHAControlPlane(meta) { + return nil, errors.New("eks: restore-snapshot only supports a single control-plane cluster; " + + "this cluster runs an HA spread with a potentially surviving quorum — use quorum reformation instead") + } + if meta.ControlPlaneTemplate == nil { + return nil, errors.New("eks: cluster has no persisted control-plane template; cannot restore") + } + + // Resolve + validate the snapshot BEFORE any mutation. resolveLatestSnapshot + // already guarantees existence (it lists then picks); an explicit key is HEADed + // so a typo/expired key hard-fails here rather than booting an empty cluster. + snapshot := input.Snapshot + if snapshot == "" { + snapshot, err = resolveLatestSnapshot(ctx, s.deps.SnapshotStore, accountID, input.ClusterName) + if err != nil { + return nil, err + } + } else if verr := validateSnapshotExists(ctx, s.deps.SnapshotStore, accountID, input.ClusterName, snapshot); verr != nil { + return nil, verr + } + + var oldNode ControlPlaneNode + if oldNodes := controlPlaneTeardownNodes(meta); len(oldNodes) > 0 { + oldNode = oldNodes[0] + } + var excludeHosts []string + if oldNode.NodeID != "" { + excludeHosts = []string{oldNode.NodeID} + } + + newNode, err := s.ProvisionFreshControlPlane(ctx, FreshCPRequest{ + AccountID: accountID, + ClusterName: input.ClusterName, + Template: meta.ControlPlaneTemplate, + ExcludeHosts: excludeHosts, + }) + if err != nil { + return nil, fmt.Errorf("eks: provision fresh control plane: %w", err) + } + + // Wire the recovery directive immediately after launch: the guest applies it + // before k3s starts on this very first boot, and it is keyed by InstanceID, + // which only exists once launched. SnapshotRequired makes the guest abort + // (not reset into an empty datastore) if it cannot fetch the snapshot. + // Any failure up to and including the meta commit unwinds the fresh CP, so a + // re-run does not stack a second resetting control plane. + if _, derr := s.SetRecoveryDirective(ctx, &SetRecoveryDirectiveInput{ + ClusterName: input.ClusterName, + InstanceID: newNode.InstanceID, + Action: RecoveryActionClusterReset, + Snapshot: snapshot, + SnapshotRequired: true, + }, accountID); derr != nil { + s.unwindFreshCP(ctx, accountID, input.ClusterName, newNode) + return nil, fmt.Errorf("eks: set recovery directive: %w", derr) + } + + // Persist the new CP into meta BEFORE re-pointing the NLB. Once committed the + // new CP is canonical, so a later NLB failure is convergeable by the reconciler + // (provisional success) rather than a bare error implying nothing happened. + if perr := replaceControlPlaneForRestore(acctKV, input.ClusterName, newNode); perr != nil { + s.unwindFreshCP(ctx, accountID, input.ClusterName, newNode) + return nil, fmt.Errorf("eks: persist replacement control plane: %w", perr) + } + + sysAcct := admin.SystemAccountID() + // Re-point the cluster NLB: deregister the old (likely already-dead) CP, + // register the new one, on both the apiserver and konnectivity TGs. Deregister + // is best-effort (the old target is likely already unhealthy/absent). A failed + // register leaves meta already naming the new CP, so the reconciler converges — + // surfaced as a provisional status, not an error. + if oldNode.ENIIP != "" { + if meta.NLBTargetGroupArn != "" { + if derr := DeregisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.NLBTargetGroupArn, oldNode.ENIIP, k3sAPIServerPort); derr != nil { + slog.WarnContext(ctx, "RestoreSnapshot: deregister old apiserver target failed", "cluster", input.ClusterName, "err", derr) + } + } + if meta.KonnTargetGroupArn != "" { + if derr := DeregisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.KonnTargetGroupArn, oldNode.ENIIP, konnectivityAgentPort); derr != nil { + slog.WarnContext(ctx, "RestoreSnapshot: deregister old konnectivity target failed", "cluster", input.ClusterName, "err", derr) + } + } + } + nlbComplete := true + if meta.NLBTargetGroupArn != "" { + if rerr := RegisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.NLBTargetGroupArn, newNode.ENIIP, k3sAPIServerPort); rerr != nil { + nlbComplete = false + slog.WarnContext(ctx, "RestoreSnapshot: register new apiserver target failed; reconciler will converge", + "cluster", input.ClusterName, "err", rerr) + } + } + if meta.KonnTargetGroupArn != "" { + if rerr := RegisterClusterTarget(ctx, s.deps.NLB, sysAcct, meta.KonnTargetGroupArn, newNode.ENIIP, konnectivityAgentPort); rerr != nil { + nlbComplete = false + slog.WarnContext(ctx, "RestoreSnapshot: register new konnectivity target failed; reconciler will converge", + "cluster", input.ClusterName, "err", rerr) + } + } + + // Fence the old CP before declaring success: a wedged-but-alive old CP running + // its etcd next to the new one is split-brain. Terminate is retried; if the old + // instance cannot be confirmed gone, fail loudly (the new CP is up and canonical + // — the operator must fence the old one manually, NOT re-run restore). + if ferr := s.confirmOldCPTerminated(ctx, sysAcct, oldNode, restoreFenceAttempts, restoreFenceDelay); ferr != nil { + return nil, fmt.Errorf("eks: restore of %q brought up new control plane %s but the old control plane %s could not be confirmed terminated after %d attempts (split-brain risk) — fence it manually, do NOT re-run restore: %w", + input.ClusterName, newNode.InstanceID, oldNode.InstanceID, restoreFenceAttempts, ferr) + } + + status := fmt.Sprintf("restore initiated: new control plane %s is booting and will restore etcd from snapshot %s — verify cluster health before relying on it", + newNode.InstanceID, snapshot) + if !nlbComplete { + status = fmt.Sprintf("control plane %s restored and recorded, but NLB re-point is incomplete; the reconciler will converge — verify cluster health", + newNode.InstanceID) + } + slog.InfoContext(ctx, "RestoreSnapshot: control-plane restore sequence complete", + "cluster", input.ClusterName, "newInstanceId", newNode.InstanceID, "snapshot", snapshot, "nlbComplete", nlbComplete) + + return &RestoreSnapshotOutput{ + ClusterName: input.ClusterName, + NewInstanceID: newNode.InstanceID, + Snapshot: snapshot, + Status: status, + }, nil +} diff --git a/spinifex/handlers/eks/restore_snapshot_test.go b/spinifex/handlers/eks/restore_snapshot_test.go new file mode 100644 index 000000000..9dc650104 --- /dev/null +++ b/spinifex/handlers/eks/restore_snapshot_test.go @@ -0,0 +1,317 @@ +package handlers_eks + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseEtcdSnapshotKey(t *testing.T) { + cases := []struct { + name string + basename string + wantOK bool + wantTier string + wantTS string + }{ + {"frequent tier", "etcd-frequent-20260709T010000Z.snap", true, "frequent", "20260709T010000Z"}, + {"daily tier", "etcd-daily-20260601T000000Z.snap", true, "daily", "20260601T000000Z"}, + {"missing prefix", "frequent-20260709T010000Z.snap", false, "", ""}, + {"missing suffix", "etcd-frequent-20260709T010000Z.tar", false, "", ""}, + {"no tier separator", "etcd-20260709T010000Z.snap", false, "", ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + parsed, ok := parseEtcdSnapshotKey(tc.basename) + require.Equal(t, tc.wantOK, ok) + if tc.wantOK { + assert.Equal(t, tc.wantTier, parsed.tier) + assert.Equal(t, tc.wantTS, parsed.timestamp) + } + }) + } +} + +func seedSnapshot(t *testing.T, store objectstore.ObjectStore, accountID, cluster, basename string) { + t.Helper() + key := accountID + "/" + cluster + "/" + basename + _, err := store.PutObject(context.Background(), &s3.PutObjectInput{ + Bucket: aws.String(eksBackupsBucket), + Key: aws.String(key), + Body: strings.NewReader("fake-snapshot-data"), + }) + require.NoError(t, err) +} + +func TestResolveLatestSnapshot_PrefersFrequentTier(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260709T230000Z.snap") + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap") + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T020000Z.snap") + // A different cluster's snapshot must not be considered. + seedSnapshot(t, store, testAccountID, "beta", "etcd-frequent-20260710T230000Z.snap") + + got, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha") + require.NoError(t, err) + assert.Equal(t, "etcd-frequent-20260709T020000Z.snap", got, "newest frequent-tier snapshot wins even though a newer daily exists") +} + +func TestResolveLatestSnapshot_FallsBackToAnyTierWhenNoFrequent(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260601T000000Z.snap") + seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260602T000000Z.snap") + + got, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha") + require.NoError(t, err) + assert.Equal(t, "etcd-daily-20260602T000000Z.snap", got) +} + +func TestResolveLatestSnapshot_NoStoreConfigured(t *testing.T) { + _, err := resolveLatestSnapshot(context.Background(), nil, testAccountID, "alpha") + require.Error(t, err) +} + +func TestResolveLatestSnapshot_NoSnapshotsFound(t *testing.T) { + store := objectstore.NewMemoryObjectStore() + _, err := resolveLatestSnapshot(context.Background(), store, testAccountID, "alpha") + require.Error(t, err) +} + +// restoreSnapshotFixtureMeta builds a single-CP ClusterMeta with everything +// RestoreSnapshot needs: a persisted launch template and NLB target-group ARNs. +func restoreSnapshotFixtureMeta(name string) *ClusterMeta { + tmpl := validK3sInput() + tmpl.AccessKey = "" + tmpl.SecretKey = "" + tmpl.PredastoreAccessKey = "" + tmpl.PredastoreSecretKey = "" + + meta := sampleClusterMeta(name) + meta.NLBTargetGroupArn = "arn:aws:elasticloadbalancing:us-east-1:111122223333:targetgroup/" + ClusterTargetGroupName(name) + "/tg-001" + meta.KonnTargetGroupArn = "arn:aws:elasticloadbalancing:us-east-1:111122223333:targetgroup/" + ClusterTargetGroupName(name) + "-konn/tg-002" + meta.ControlPlaneTemplate = &tmpl + meta.ControlPlaneNodes = []ControlPlaneNode{{ + NodeID: "node-old", + InstanceID: "i-old111", + ENIID: "eni-old111", + ENIIP: "10.0.1.9", + MgmtIP: "10.255.0.5", + }} + meta.ControlPlaneInstanceID = "i-old111" + meta.ControlPlaneENIID = "eni-old111" + meta.ControlPlaneENIIP = "10.0.1.9" + return meta +} + +func TestRestoreSnapshot_HappyPathLaunchesDirectsAndRepoints(t *testing.T) { + f := newEKSServiceFixture(t) + f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}} + store := objectstore.NewMemoryObjectStore() + f.svc.deps.SnapshotStore = store + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap") + + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + out, err := f.svc.RestoreSnapshot(context.Background(), &RestoreSnapshotInput{ClusterName: "alpha"}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", out.Snapshot, "empty --snapshot resolves the latest one") + assert.NotEmpty(t, out.NewInstanceID) + + // The recovery directive must be wired for the new instance. + dirOut, err := f.svc.GetRecoveryDirective(context.Background(), + &GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: out.NewInstanceID}, testAccountID) + require.NoError(t, err) + assert.Equal(t, RecoveryActionClusterReset, dirOut.Directive.Action) + assert.Equal(t, "etcd-frequent-20260709T010000Z.snap", dirOut.Directive.Snapshot) + assert.True(t, dirOut.Directive.SnapshotRequired, "fresh DR seed must require the snapshot (abort, not reset-into-empty)") + assert.NotEmpty(t, out.Status, "success carries a provisional status, never a bare 'restored'") + + // NLB re-pointed: old CP deregistered, new CP registered, on both TGs. + require.Len(t, f.nlb.deregisterCalls, 2, "apiserver + konnectivity TGs") + require.Len(t, f.nlb.registerCalls, 2) + for _, dc := range f.nlb.deregisterCalls { + require.Len(t, dc.Targets, 1) + assert.Equal(t, "10.0.1.9", aws.StringValue(dc.Targets[0].Id), "deregisters the OLD CP's ENI IP") + } + for _, rc := range f.nlb.registerCalls { + require.Len(t, rc.Targets, 1) + assert.NotEqual(t, "10.0.1.9", aws.StringValue(rc.Targets[0].Id), "registers the NEW CP's ENI IP, not the old one") + } + + // Cluster meta reflects the replacement. + got, err := GetClusterMeta(f.kv, "alpha") + require.NoError(t, err) + assert.Equal(t, out.NewInstanceID, got.ControlPlaneInstanceID) + require.Len(t, got.ControlPlaneNodes, 1) + assert.Equal(t, out.NewInstanceID, got.ControlPlaneNodes[0].InstanceID) + + // The old CP is torn down best-effort. + assert.Contains(t, f.inst.terminateCalls, "i-old111") +} + +func TestRestoreSnapshot_ExplicitSnapshotUsedVerbatim(t *testing.T) { + f := newEKSServiceFixture(t) + f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}} + store := objectstore.NewMemoryObjectStore() + f.svc.deps.SnapshotStore = store + // A different (newer) snapshot exists; the explicit one must be used verbatim. + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260710T010000Z.snap") + seedSnapshot(t, store, testAccountID, "alpha", "etcd-daily-20260601T000000Z.snap") + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + out, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.NoError(t, err) + assert.Equal(t, "etcd-daily-20260601T000000Z.snap", out.Snapshot, "explicit snapshot is used verbatim, no latest-resolution") + + // The directive must mark the snapshot as required so the guest aborts on a + // fetch failure rather than resetting into an empty datastore. + dirOut, err := f.svc.GetRecoveryDirective(context.Background(), + &GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: out.NewInstanceID}, testAccountID) + require.NoError(t, err) + assert.True(t, dirOut.Directive.SnapshotRequired) + assert.NotEmpty(t, out.Status, "a provisional status is always returned") +} + +func TestRestoreSnapshot_ExplicitSnapshotMissingHardFailsBeforeLaunch(t *testing.T) { + f := newEKSServiceFixture(t) + f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}} + store := objectstore.NewMemoryObjectStore() + f.svc.deps.SnapshotStore = store + // Only an unrelated snapshot exists — the requested key is absent. + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap") + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + _, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "not found") + assert.Empty(t, f.inst.launchCalls, "a missing explicit snapshot must hard-fail before launching anything") +} + +func TestRestoreSnapshot_ExplicitSnapshotWithNoStoreHardFails(t *testing.T) { + f := newEKSServiceFixture(t) + f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}} + f.svc.deps.SnapshotStore = nil + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + _, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.Error(t, err) + assert.Empty(t, f.inst.launchCalls, "cannot verify snapshot exists without a store — must not launch") +} + +func TestRestoreSnapshot_HARejected(t *testing.T) { + f := newEKSServiceFixture(t) + meta := restoreSnapshotFixtureMeta("alpha") + meta.ControlPlaneSpreadGroup = "eks-cp-111122223333-alpha" + meta.ControlPlaneNodes = append(meta.ControlPlaneNodes, ControlPlaneNode{NodeID: "node-2", InstanceID: "i-cp2"}) + require.NoError(t, PutClusterMeta(f.kv, meta)) + + _, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "HA spread") + assert.Empty(t, f.inst.launchCalls, "must not launch anything for an HA cluster") +} + +func TestRestoreSnapshot_NoTemplateRejected(t *testing.T) { + f := newEKSServiceFixture(t) + meta := restoreSnapshotFixtureMeta("alpha") + meta.ControlPlaneTemplate = nil + require.NoError(t, PutClusterMeta(f.kv, meta)) + + _, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "alpha", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.Error(t, err) + assert.Contains(t, err.Error(), "template") +} + +func TestRestoreSnapshot_ClusterNotFound(t *testing.T) { + f := newEKSServiceFixture(t) + _, err := f.svc.RestoreSnapshot(context.Background(), + &RestoreSnapshotInput{ClusterName: "ghost", Snapshot: "etcd-daily-20260601T000000Z.snap"}, testAccountID) + require.Error(t, err) +} + +func TestRestoreSnapshot_NLBFailureIsProvisionalNotError(t *testing.T) { + f := newEKSServiceFixture(t) + f.svc.deps.Scheduler = &fakeHostScheduler{hosts: []string{"node-new"}} + store := objectstore.NewMemoryObjectStore() + f.svc.deps.SnapshotStore = store + seedSnapshot(t, store, testAccountID, "alpha", "etcd-frequent-20260709T010000Z.snap") + // NLB register fails after meta is persisted: the new CP is already canonical. + f.nlb.registerErr = errors.New("TargetGroupNotFound") + + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + out, err := f.svc.RestoreSnapshot(context.Background(), &RestoreSnapshotInput{ClusterName: "alpha"}, testAccountID) + require.NoError(t, err, "an NLB re-point failure after meta commit is provisional, not a hard error") + assert.Contains(t, out.Status, "NLB re-point is incomplete") + + // Meta was persisted BEFORE the NLB step, so it already names the new CP and + // the reconciler can converge the target group. + got, err := GetClusterMeta(f.kv, "alpha") + require.NoError(t, err) + assert.Equal(t, out.NewInstanceID, got.ControlPlaneInstanceID, "meta persisted before NLB re-point") + + // The new CP must NOT be torn down on an NLB failure (it is canonical now). + assert.NotContains(t, f.inst.terminateCalls, out.NewInstanceID) +} + +func TestRestoreSnapshot_FenceFailureIsLoudError(t *testing.T) { + f := newEKSServiceFixture(t) + // The old CP cannot be terminated (still alive) — split-brain risk. + f.inst.terminateErr = errors.New("InstanceStillRunning") + + oldNode := ControlPlaneNode{NodeID: "node-old", InstanceID: "i-old111", ENIID: "eni-old111", ENIIP: "10.0.1.9"} + // Drive the fence helper directly with no delay so the retry budget is exercised fast. + err := f.svc.confirmOldCPTerminated(context.Background(), testAccountID, oldNode, 3, 0) + require.Error(t, err, "a persistently un-terminable old CP is a fence failure") + require.Len(t, f.inst.terminateCalls, 3, "terminate is retried up to the attempt budget") +} + +func TestConfirmOldCPTerminated_NoOldNodeSucceeds(t *testing.T) { + f := newEKSServiceFixture(t) + err := f.svc.confirmOldCPTerminated(context.Background(), testAccountID, ControlPlaneNode{}, 3, 0) + require.NoError(t, err, "no old CP to fence is trivially confirmed") + assert.Empty(t, f.inst.terminateCalls) +} + +func TestUnwindFreshCP_TerminatesAndClearsDirective(t *testing.T) { + f := newEKSServiceFixture(t) + meta := restoreSnapshotFixtureMeta("alpha") + require.NoError(t, PutClusterMeta(f.kv, meta)) + + // Pretend a directive was already set for the fresh CP (as the real flow does + // just before a meta-commit failure), then unwind. + node := ControlPlaneNode{NodeID: "node-new", InstanceID: "i-new999", ENIID: "eni-new999", ENIIP: "10.0.1.42"} + _, err := f.svc.SetRecoveryDirective(context.Background(), &SetRecoveryDirectiveInput{ + ClusterName: "alpha", InstanceID: node.InstanceID, + Action: RecoveryActionClusterReset, Snapshot: "etcd-frequent-20260709T010000Z.snap", SnapshotRequired: true, + }, testAccountID) + require.NoError(t, err) + + f.svc.unwindFreshCP(context.Background(), testAccountID, "alpha", node) + + // The fresh CP is terminated so it cannot boot and destructively reset. + assert.Contains(t, f.inst.terminateCalls, "i-new999") + + // The directive is superseded with a no-op so a VM that boots first does not reset. + dirOut, err := f.svc.GetRecoveryDirective(context.Background(), + &GetRecoveryDirectiveInput{ClusterName: "alpha", InstanceID: node.InstanceID}, testAccountID) + require.NoError(t, err) + assert.Equal(t, RecoveryActionNone, dirOut.Directive.Action, "directive cleared to none on unwind") +} diff --git a/spinifex/handlers/eks/service.go b/spinifex/handlers/eks/service.go index 28319ef60..1f4516524 100644 --- a/spinifex/handlers/eks/service.go +++ b/spinifex/handlers/eks/service.go @@ -54,6 +54,11 @@ type EKSService interface { // reconciler / operator CLI set it to drive an etcd cluster-reset recovery. GetRecoveryDirective(ctx context.Context, input *GetRecoveryDirectiveInput, accountID string) (*GetRecoveryDirectiveOutput, error) SetRecoveryDirective(ctx context.Context, input *SetRecoveryDirectiveInput, accountID string) (*SetRecoveryDirectiveOutput, error) + // RestoreSnapshot is an internal control-plane method (not an AWS-SDK + // action): the operator CLI (spx admin eks restore-snapshot) drives the + // single-CP total-loss DR path — fresh CP launch, recovery-directive wiring, + // and NLB re-point. + RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error) // OIDC identity-provider configs. AssociateIdentityProviderConfig(ctx context.Context, input *eks.AssociateIdentityProviderConfigInput, accountID string) (*eks.AssociateIdentityProviderConfigOutput, error) diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go index d875fee33..24377221a 100644 --- a/spinifex/handlers/eks/service_impl.go +++ b/spinifex/handlers/eks/service_impl.go @@ -24,6 +24,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/config" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" "github.com/mulgadc/spinifex/spinifex/handlers/sysinstance" + "github.com/mulgadc/spinifex/spinifex/objectstore" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -60,6 +61,16 @@ type EKSServiceDeps struct { SystemSecretKey string GatewayCACert string + // SystemPredastoreURL is the mgmt-bridge-reachable predastore endpoint baked + // into the CP VM's etcd-snapshot.env (SystemAccessKey/SystemSecretKey double + // as the predastore SigV4 creds — same system credential used for the gateway). + SystemPredastoreURL string + + // SnapshotStore lists/reads etcd snapshots from the eks-backups-system bucket + // for restore-snapshot's latest-snapshot resolution. Nil disables that lookup + // (callers must pass --snapshot explicitly). + SnapshotStore objectstore.ObjectStore + VPCSG sgProvisioner VPCK3s k3sVPCProvisioner VPCSubnet SubnetVPCResolver @@ -707,24 +718,27 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc elbSubnets := dedupSubnetsByAZ(ctx, s.deps.VPCSubnet, accountID, lc.subnetIDs) serverIn := K3sServerInput{ - AccountID: sysAcct, - ClusterAccountID: accountID, - ClusterName: name, - Region: region, - SubnetID: cpRefs.PrivateSubnetIDs[0], - VpcID: meta.ResourcesVpcConfig.VpcId, - ELBSubnetIDs: elbSubnets, - ControlPlaneSGID: cpSG, - NLBDNS: nlb.DNSName, - EndpointIP: nlb.FrontendIP, - PrivateEndpointIP: meta.PrivateEndpointIP, - OIDCIssuer: oidcIssuer, - OIDCPrivateKeyPEM: privPEM, - OIDCPublicKeyPEM: pubPEM, - GatewayURL: s.deps.SystemGatewayURL, - AddonGatewayURL: s.deps.GatewayBaseURL, - GatewayCACert: s.deps.GatewayCACert, - JoinToken: joinToken, + AccountID: sysAcct, + ClusterAccountID: accountID, + ClusterName: name, + Region: region, + SubnetID: cpRefs.PrivateSubnetIDs[0], + VpcID: meta.ResourcesVpcConfig.VpcId, + ELBSubnetIDs: elbSubnets, + ControlPlaneSGID: cpSG, + NLBDNS: nlb.DNSName, + EndpointIP: nlb.FrontendIP, + PrivateEndpointIP: meta.PrivateEndpointIP, + OIDCIssuer: oidcIssuer, + OIDCPrivateKeyPEM: privPEM, + OIDCPublicKeyPEM: pubPEM, + GatewayURL: s.deps.SystemGatewayURL, + AddonGatewayURL: s.deps.GatewayBaseURL, + GatewayCACert: s.deps.GatewayCACert, + JoinToken: joinToken, + PredastoreEndpoint: s.deps.SystemPredastoreURL, + PredastoreAccessKey: s.deps.SystemAccessKey, + PredastoreSecretKey: s.deps.SystemSecretKey, } // Prefer IMDS instance-role creds: attach a system instance profile so the @@ -767,6 +781,8 @@ func (s *EKSServiceImpl) launchClusterInfra(ctx context.Context, lc clusterLaunc tmpl.AccessKey = "" tmpl.SecretKey = "" tmpl.IamInstanceProfileArn = "" + tmpl.PredastoreAccessKey = "" + tmpl.PredastoreSecretKey = "" meta.ControlPlaneTemplate = &tmpl // Persist the CP VM + ENI + spread-group refs now, before any further fallible diff --git a/spinifex/handlers/eks/service_nats.go b/spinifex/handlers/eks/service_nats.go index d54d92f86..2173615ad 100644 --- a/spinifex/handlers/eks/service_nats.go +++ b/spinifex/handlers/eks/service_nats.go @@ -153,6 +153,10 @@ func (s *NATSEKSService) SetRecoveryDirective(ctx context.Context, input *SetRec return utils.NATSRequest[SetRecoveryDirectiveOutput](ctx, s.natsConn, "eks.SetRecoveryDirective", input, defaultTimeout, accountID) } +func (s *NATSEKSService) RestoreSnapshot(ctx context.Context, input *RestoreSnapshotInput, accountID string) (*RestoreSnapshotOutput, error) { + return utils.NATSRequest[RestoreSnapshotOutput](ctx, s.natsConn, "eks.RestoreSnapshot", input, defaultTimeout, accountID) +} + // --- OIDC identity-provider configs --- func (s *NATSEKSService) AssociateIdentityProviderConfig(ctx context.Context, input *eks.AssociateIdentityProviderConfigInput, accountID string) (*eks.AssociateIdentityProviderConfigOutput, error) {