Compare commits

..

17 Commits

Author SHA1 Message Date
joshsiv-mulga 076d1793b2 fix(latitude): stage and verify the CA before replacing it, treat a fresh pid-less lock as live 2026-11-03 17:42:10 +10:00
joshsiv-mulga 59dcb27d27 fix(latitude): refuse any existing AWS profile section, lock mutating commands 2026-11-03 17:41:01 +10:00
joshsiv-mulga c2b7d63a6f fix(latitude): sticky uncertain record, per-name profile guard, empty-page pagination, state-bound destroy preview 2026-11-03 17:37:11 +10:00
joshsiv-mulga 18acef43dd fix(latitude): tty console gate, per-up reconfirm, owned-set destroy preview, link pagination 2026-11-03 17:33:14 +10:00
joshsiv-mulga 0995f0a6c0 fix(latitude): console gate, owned-ID destroy, effective ssh rule gate, no secrets in argv 2026-11-03 17:29:02 +10:00
joshsiv-mulga a65116d827 feat(latitude): add deploy-latitude.sh for one-node Spinifex on latitude.sh 2026-11-03 17:22:08 +10:00
joshsiv-mulga 7bb8c79792 fix(latitude): firewall needs existing install state, verify rechecks mode and table last and reports an interval 2026-11-03 16:38:46 +10:00
joshsiv-mulga 4e5ff95027 fix(latitude): read-only firewall pre-checks, recheck baseline after daemon stop, no mtime proof in verify 2026-11-03 16:37:33 +10:00
joshsiv-mulga cbf6307012 fix(latitude): console confirm and stale-table baseline before arming, read-only verify with effective config 2026-11-03 16:35:42 +10:00
joshsiv-mulga a58e7bd918 feat(latitude): firewall and verify phases, gateway_ip on the public pool, git in prep 2026-11-03 16:30:35 +10:00
joshsiv-mulga 061cc96b82 fix: require OVS/OVN units and accept only an absent or false firewall_enabled in the latitude installer 2026-11-03 14:33:35 +10:00
joshsiv-mulga 00bbaa8998 fix(latitude): check nat transit wiring, validate config after edit, require core units active 2026-11-03 14:30:25 +10:00
joshsiv-mulga 40a7bffbe6 fix(latitude): isolate aws calls, parse config on resume, park firewall phases, prove firewall absent by nft inventory 2026-11-03 14:26:35 +10:00
joshsiv-mulga 7abe8e39a6 fix(latitude): verify routes, gateway and OVN state on every ovn run, fail closed on firewall state 2026-11-03 14:25:06 +10:00
joshsiv-mulga 108b27e326 refactor(latitude): run init and config in the start phase, after setup-ovn, as vultr does 2026-11-03 14:19:57 +10:00
joshsiv-mulga 3c578f474f feat(latitude): split install-node.sh into phases, never arm the firewall during install 2026-11-03 14:19:01 +10:00
joshsiv-mulga bd8f148879 feat(latitude): add single-node on-host install script for latitude.sh bare metal 2026-11-03 14:07:06 +10:00
248 changed files with 8012 additions and 5188 deletions
+2 -23
View File
@@ -71,16 +71,9 @@ jobs:
ARTIFACT_DIR: /tmp/spinifex-e2e-nightly-cell-${{ inputs.cell || '20' }}-${{ github.run_id }}
SPINIFEX_OTEL_ATTRS: ${{ inputs.otel-attrs || format('ci.run_id={0},ci.workflow={1},ci.cell={2},ci.matrix=baremetal/pxe/real-hw', github.run_id, github.workflow, inputs.cell || '20') }}
steps:
# A cancelled job's root steps outlive it, since the runner cannot reach
# them once it kills sudo, and would keep writing here and to the cache.
# A rootfs killed before bin/bash exists is one the build refuses to reuse.
- name: Clean Workspace
run: |
sudo systemctl stop spinifex-bm-pxe.scope spinifex-bm-e2e.scope 2>/dev/null || true
sudo umount -R -l "${BM_ISO_CACHE}/rootfs" 2>/dev/null || true
if [ -d "${BM_ISO_CACHE}/rootfs" ] && [ ! -x "${BM_ISO_CACHE}/rootfs/bin/bash" ]; then
sudo rm -rf --one-file-system "${BM_ISO_CACHE}/rootfs"
fi
if [ ! -c /dev/null ]; then sudo rm -f /dev/null; sudo mknod -m 666 /dev/null c 1 3; fi
sudo rm -rf --one-file-system "$GITHUB_WORKSPACE"/*
@@ -120,15 +113,10 @@ jobs:
"$GITHUB_WORKSPACE" \
"${RUNNER_TEMP}/spinifex-tests.tar" \
"${RUNNER_TEMP}/bm-bins"
# Built here, as the runner user, so the PXE build finds it current.
make -C "$GITHUB_WORKSPACE" build-microvm-image
- name: Build PXE Artifacts
run: |
# -o build-spinifex: the binaries come from Build Artifacts, and a
# rebuild here would write root-owned files into the workspace.
sudo systemd-run --scope --collect --quiet --unit=spinifex-bm-pxe -- \
make -C mulga/scripts/iso-builder -o build-spinifex \
sudo make -C mulga/scripts/iso-builder \
build-rootfs build-squashfs build-initramfs \
CACHE="${BM_ISO_CACHE}" \
SPINIFEX_BIN="${RUNNER_TEMP}/bm-bins/spx" \
@@ -136,10 +124,6 @@ jobs:
SPINIFEX_SRC="${GITHUB_WORKSPACE}" \
VIPERBLOCK_SRC="${RUNNER_TEMP}/bm-bins/viperblock" \
VERSION="$(cd "$GITHUB_WORKSPACE" && git describe --tags --always --exclude '*-dev' 2>/dev/null || echo dev)"
if root_owned="$(find "$GITHUB_WORKSPACE" -user root -print -quit)" && [ -n "$root_owned" ]; then
echo "::error::the PXE build wrote root-owned files into the workspace, e.g. $root_owned"
exit 1
fi
- name: Write ePDU credentials
run: |
@@ -211,18 +195,13 @@ jobs:
done
export BM_TEST_TIMEOUT
echo "suites=${BM_SUITES} test-timeout=${BM_TEST_TIMEOUT}"
sudo -E systemd-run --scope --collect --quiet --unit=spinifex-bm-e2e -- \
mulga/scripts/baremetal/run-baremetal-e2e.sh \
sudo -E mulga/scripts/baremetal/run-baremetal-e2e.sh \
--vmlinuz "${BM_ISO_CACHE}/vmlinuz" \
--initrd "${BM_ISO_CACHE}/initrd.img" \
--squashfs "${BM_ISO_CACHE}/spinifex.squashfs" \
--tests-tar "${RUNNER_TEMP}/spinifex-tests.tar" \
2>&1 | tee "${ARTIFACT_DIR}/test-baremetal.log"
- name: Stop this job's root steps
if: always()
run: sudo systemctl stop spinifex-bm-pxe.scope spinifex-bm-e2e.scope 2>/dev/null || true
- name: Generate JUnit report
if: always()
run: |
+3
View File
@@ -15,6 +15,9 @@ var allowlist = []allowance{
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/igw", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/image", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/instance", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/key", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/launchtemplate", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/placementgroup", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/routetable", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/snapshot", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
{From: "operator/accountlifecycle", To: "domains/ec2/awsapi/volume", Debt: "accountteardown (dependency)", Reason: "EC2 reapers call the awsapi action adapters; needs domain-owned cleanup capabilities"},
+1 -8
View File
@@ -23,7 +23,7 @@ subject shape are wire compatibility. Callers and receivers must use this
package rather than rebuild those details. `commands_test.go` preserves the
version-1 JSON shape and subject form.
This package owns shared command payloads. It
This package owns shared command payloads and the drain-volume response. It
does not define every action's reply body, authorize the request, store EC2
state or implement NATS transport. Those remain responsibilities of the EC2
domain and its runtime composition.
@@ -37,13 +37,6 @@ contract or an approved compatibility plan; it is not a package refactor.
Only the subject is owned here today.
The request (`account_id`) and reply (`vpc_id`, `error`) bodies are still declared as anonymous structs on each side; moving them into this package is a contract change that needs its own characterization first.
## Volume held elsewhere
A start refused because another node holds one of the instance's volumes is an `IncorrectState` error payload on the `ec2.start`, `ec2.start.<node>` and `ec2.cmd.<instance-id>` replies, and the AWS body cannot say why.
The owning node adds the `X-Volume-Held-Elsewhere: true` header to that reply, the `ec2.start` forwarder relays it, and `StartInstances` wraps `ErrVolumeHeldElsewhere` around the error it rebuilds, so a caller matches it with `errors.Is` rather than reading the message.
The owning node sets the header only from the EBS provider's `volume_held_elsewhere` code, which every provider returns for this refusal; no reader matches the message text.
`VolumeHeldElsewhereText` is the wording viperblockd puts in that refusal's message, not a signal.
## Platform inventory
`PlatformInventorySubject` (`ec2.v1.PlatformInventory`) is a request/reply route on which a daemon in the `spinifex-workers` queue group reports one account's key pairs, placement groups, launch templates and their `ec2-tags` entries from the EC2 owner stores.
+29
View File
@@ -29,6 +29,7 @@ type EC2InstanceCommand struct {
Attributes EC2CommandAttributes `json:"attributes"`
AttachVolumeData *AttachVolumeData `json:"attach_volume_data,omitempty"`
DetachVolumeData *DetachVolumeData `json:"detach_volume_data,omitempty"`
DrainVolumeData *DrainVolumeData `json:"drain_volume_data,omitempty"`
AttachENIData *AttachENIData `json:"attach_eni_data,omitempty"`
DetachENIData *DetachENIData `json:"detach_eni_data,omitempty"`
IamProfileAssociationData *IamProfileAssociationData `json:"iam_profile_association_data,omitempty"`
@@ -44,6 +45,7 @@ type EC2CommandAttributes struct {
StartInstance bool `json:"start_instance"`
AttachVolume bool `json:"attach_volume"`
DetachVolume bool `json:"detach_volume"`
DrainVolume bool `json:"drain_volume,omitempty"`
RebootInstance bool `json:"reboot_instance"`
AttachENI bool `json:"attach_eni"`
DetachENI bool `json:"detach_eni"`
@@ -67,6 +69,33 @@ type DetachVolumeData struct {
Force bool `json:"force,omitempty"`
}
// DrainVolumeData carries parameters for a drain-volume command, which flushes
// the volume's in-flight writes to S3 before a snapshot reads them. The command
// is addressed to the instance the volume is attached to because the drain
// socket only exists on the node hosting it.
type DrainVolumeData struct {
VolumeID string `json:"volume_id"`
}
const (
// DrainVolumeStatusDrained is the ack a node returns once the volume's
// in-flight writes have reached S3.
DrainVolumeStatusDrained = "drained"
// DrainVolumeStatusNotRunning is the ack a node returns when it still holds
// the instance but the VM is not running, so no process is writing to the
// volume and its sealed checkpoint is already current. An attachment record
// outlives the writer — stop deliberately leaves boot volumes attached — so
// the caller cannot tell this from the record alone.
DrainVolumeStatusNotRunning = "not-running"
)
// DrainVolumeResponse is the reply to a drain-volume command.
type DrainVolumeResponse struct {
VolumeID string `json:"volume_id"`
Status string `json:"status"`
}
// AttachENIData carries parameters for an attach-network-interface command.
type AttachENIData struct {
NetworkInterfaceID string `json:"network_interface_id"`
+14
View File
@@ -16,6 +16,7 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
StartInstance: true,
AttachVolume: true,
DetachVolume: true,
DrainVolume: true,
RebootInstance: true,
AttachENI: true,
DetachENI: true,
@@ -27,6 +28,7 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
},
AttachVolumeData: &AttachVolumeData{VolumeID: "vol-1", Device: "/dev/sdf"},
DetachVolumeData: &DetachVolumeData{VolumeID: "vol-2", Device: "/dev/sdg", Force: true},
DrainVolumeData: &DrainVolumeData{VolumeID: "vol-3"},
AttachENIData: &AttachENIData{NetworkInterfaceID: "eni-1", DeviceIndex: 2},
DetachENIData: &DetachENIData{AttachmentID: "eni-attach-1", Force: true},
IamProfileAssociationData: &IamProfileAssociationData{InstanceProfileArn: "arn:aws:iam::123456789012:instance-profile/example"},
@@ -45,6 +47,7 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
"start_instance": true,
"attach_volume": true,
"detach_volume": true,
"drain_volume": true,
"reboot_instance": true,
"attach_eni": true,
"detach_eni": true,
@@ -56,6 +59,7 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
},
"attach_volume_data": {"volume_id": "vol-1", "device": "/dev/sdf"},
"detach_volume_data": {"volume_id": "vol-2", "device": "/dev/sdg", "force": true},
"drain_volume_data": {"volume_id": "vol-3"},
"attach_eni_data": {"network_interface_id": "eni-1", "device_index": 2},
"detach_eni_data": {"attachment_id": "eni-attach-1", "force": true},
"iam_profile_association_data": {"instance_profile_arn": "arn:aws:iam::123456789012:instance-profile/example"},
@@ -65,6 +69,16 @@ func TestEC2InstanceCommandJSONContract(t *testing.T) {
}`, string(data))
}
func TestDrainVolumeResponseJSONContract(t *testing.T) {
data, err := json.Marshal(DrainVolumeResponse{
VolumeID: "vol-1",
Status: DrainVolumeStatusDrained,
})
require.NoError(t, err)
require.JSONEq(t, `{"volume_id":"vol-1","status":"drained"}`, string(data))
require.Equal(t, "not-running", DrainVolumeStatusNotRunning)
}
func TestInstanceCommandSubjects(t *testing.T) {
require.Equal(t, "ec2.cmd.i-0123456789abcdef0", InstanceCommandSubject("i-0123456789abcdef0"))
require.Equal(t, "ec2.cmd.*", InstanceCommandSubjectWildcard)
+4 -10
View File
@@ -1,13 +1,7 @@
package ec2v1
import "errors"
// VolumeHeldElsewhereText is the message viperblockd gives a lease refusal. It
// is wording for the operator; callers classify the refusal by its code.
// VolumeHeldElsewhereText is the text of a start refused because another node
// holds one of the instance's volumes. It crosses NATS as message text, so
// readers match on it; the provider that refuses and every reader share this
// string so they cannot drift.
const VolumeHeldElsewhereText = "volume is leased by another owner"
// ErrVolumeHeldElsewhere marks a start refused because another node still
// holds one of the instance's volumes. The owning node sets it from the EBS
// provider's held-elsewhere code, and a start reply carries it across NATS as
// the X-Volume-Held-Elsewhere header rather than as message text.
var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")
+2 -2
View File
@@ -84,8 +84,8 @@ type ListAccountsResponse struct {
}
// IAMQuiescence is the coordinator's recorded view of IAM quiescence for one
// fence. Recording only: it admits nothing. QuiescedScope is filled only by a run
// that guarantees a final inventory re-read, and holds as of EvaluatedAt.
// fence. Recording only: it admits nothing. QuiescedScope is filled only for a
// non-finalizing run, and holds as of EvaluatedAt: a write after that is not caught.
type IAMQuiescence struct {
OperationID string `json:"operationId"`
Fence uint64 `json:"fence"`
+3 -3
View File
@@ -112,14 +112,14 @@ When a daemon handles `RunInstances`:
1. **Resource Check**: Validates CPU/memory availability via `ResourceManager`
2. **Volume Generation**: Creates boot and EFI volumes via Viperblock
3. **Volume Mount**: Publishes each volume through the EBS provider contract (`PublishVolume` on `ebs.provider.v1.{node}.mount`), receives its export
3. **Volume Mount**: Sends a mount request to the colocated Viperblock worker on `ebs.{node}.mount`, receives NBD URI
4. **QEMU Launch**: Builds and executes QEMU command with NBD-backed disks
5. **QMP Monitoring**: Establishes QEMU Machine Protocol connection for VM management
6. **Command subscription**: Subscribes to `ec2.cmd.{instanceID}` for subsequent commands on this VM
```go
// Volume mounting through the EBS provider contract (runtime/daemon/vm_adapters.go)
published, err := a.provider.PublishVolume(ebsprovider.WithAccountID(ctx, accountID), ebsprovider.PublishVolumeRequest{Versioned: ebsprovider.NewLatestVersioned(), VolumeID: req.Name, NodeID: a.node})
// Volume mounting via NATS (runtime/daemon/vm_adapters.go)
reply, err := ebsRequestWithTrace(ctx, a.nc, instance.AccountID, viperblocklegacyv1.MountSubject(a.node), ebsMountRequest, 30*time.Second)
// QEMU launch with NBD-backed drives in args (runtime/compute/vm/vm.go)
cmd := exec.Command(qemuArchitecture, args...) // qemu-system-x86_64 or qemu-system-aarch64
+6 -12
View File
@@ -5,7 +5,6 @@ import (
"os"
"path/filepath"
"sync"
"sync/atomic"
"testing"
"github.com/stretchr/testify/require"
@@ -29,11 +28,10 @@ func SocketTempDir(t *testing.T) string {
// StartDrainSocket stands in for the NBD plugin's drain socket, at the path the
// plugin serves it on the node hosting volumeID ({dataDir}/viperblock/{volumeID}
// /snapshot.sock). Every connection is answered with ack ("OK\n" for a drain
// that reached S3, "ERR\n" for one that did not) and closed. connections
// reports how many drains have been asked for so far.
func StartDrainSocket(t *testing.T, dataDir, volumeID, ack string) (connections func() int) {
// that reached S3, "ERR\n" for one that did not) and closed.
func StartDrainSocket(t *testing.T, dataDir, volumeID, ack string) {
t.Helper()
return serveDrainSocket(t, dataDir, volumeID, ack, nil, nil)
serveDrainSocket(t, dataDir, volumeID, ack, nil, nil)
}
// StartBlockingDrainSocket stands in for a plugin whose flush is still running:
@@ -45,7 +43,7 @@ func StartBlockingDrainSocket(t *testing.T, dataDir, volumeID, ack string) (acce
acceptedCh := make(chan struct{})
releaseCh := make(chan struct{})
_ = serveDrainSocket(t, dataDir, volumeID, ack, acceptedCh, releaseCh)
serveDrainSocket(t, dataDir, volumeID, ack, acceptedCh, releaseCh)
var once sync.Once
t.Cleanup(func() { once.Do(func() { close(releaseCh) }) })
@@ -54,9 +52,8 @@ func StartBlockingDrainSocket(t *testing.T, dataDir, volumeID, ack string) (acce
}
// serveDrainSocket answers every connection with ack, optionally closing
// accepted on the first one and waiting for release before it replies. The
// returned func counts the connections accepted so far.
func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan struct{}, release <-chan struct{}) func() int {
// accepted on the first one and waiting for release before it replies.
func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan struct{}, release <-chan struct{}) {
t.Helper()
dir := filepath.Join(dataDir, "viperblock", volumeID)
@@ -66,7 +63,6 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
require.NoError(t, err)
t.Cleanup(func() { _ = ln.Close() })
var count atomic.Int64
go func() {
var once sync.Once
for {
@@ -74,7 +70,6 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
if err != nil {
return // listener closed by the cleanup above
}
count.Add(1)
if accepted != nil {
once.Do(func() { close(accepted) })
}
@@ -85,5 +80,4 @@ func serveDrainSocket(t *testing.T, dataDir, volumeID, ack string, accepted chan
_ = conn.Close()
}
}()
return func() int { return int(count.Load()) }
}
+10
View File
@@ -0,0 +1,10 @@
.terraform/
terraform.tfstate
terraform.tfstate.*
*.tfplan
*.auto.tfvars
crash.log
terraform.tfvars
*.pem
*.log
.terraform.lock.hcl
+70
View File
@@ -0,0 +1,70 @@
# latitude-spx
Builds a one-node Spinifex deployment on latitude.sh bare metal, from your laptop, with `deploy-latitude.sh`. The shape follows `vultr-spx`.
**Status: not yet run end to end.** The script was written from the plan doc findings (`docs/development/feature/latitude-bm.md`, L1 to L18) and from the provider schema. No latitude.sh API call or host run was made while building it. Treat the first `up` as the test.
## What it creates
| Resource | Notes |
|---|---|
| Project | `spinifex-<name>`, unless `--project` names an existing one (then it is never deleted) |
| SSH key | `spinifex-<name>`, account scoped |
| User data | `spinifex-<name>`, host preparation only (no `{{ }}`, latitude.sh substitutes them) |
| Server | `m4-metal-medium`, site `DAL`, Ubuntu 26.04, **hourly** billing, hostname `spx-<name>-node1` |
| Elastic IP | one, routed to node 1; it is the Spinifex public pool address |
## Requirements
OpenTofu (or Terraform) 1.5 or later, `aws` CLI, `python3`, `curl`, `ssh`, an SSH key pair, and `LATITUDESH_AUTH_TOKEN` exported.
## Usage
```
export LATITUDESH_AUTH_TOKEN=... # from the latitude.sh dashboard
./deploy-latitude.sh up --name demo
./deploy-latitude.sh status --name demo
./deploy-latitude.sh ssh --name demo node1 'sudo systemctl status spinifex.target'
./deploy-latitude.sh credentials --name demo
./deploy-latitude.sh destroy --name demo
```
`./deploy-latitude.sh --help` lists the options. `--nodes` must be 1: three nodes need the VLAN and the Elastic IP move, which are not built.
## What `up` does
`terraform` → `host` (wait for SSH, cloud-init, `/dev/kvm`) → console check → `prep` (`resolv.conf`) → `disk` (`host-data-disk.sh`) → `install` → `ovn` → `firewall` → `start` → `verify` → `access` → AWS profile.
- **Console check.** Before the first step that changes the node, `up` prints the recovery commands, runs `passwd` for the ubuntu user on the node, and waits for you to type `console-ok` after logging in over the serial console from another window. It asks again on every `up` that still has a changing step to run.
- **Data disk.** `host-data-disk.sh` picks the blank NVMe by content and by-id EUI, never by `nvmeXn1` (L1, L11). It needs exactly one candidate of at least 100 GB with no signature, partitions, holders or mounts, or it stops and lists the disks (`--data-disk-id` chooses). It mounts by UUID with `nofail`.
- **Firewall.** `setup.sh --firewall on` runs before Spinifex first starts, and the daemon writes `peers.nft` and loads the policy during the start. The start phase refuses to go on unless the active rules allow ssh: the accept rule in `spinifex.nft`, the defined ssh ports covering every port `sshd -T` reports, and `0.0.0.0/0` trusted in `custom.nft`.
- **Proof.** `verify` and the access check run on every `up`, never skipped by a step marker. The access check opens a new ssh login, asks `https://<ip>:9999` for an answer, and confirms the firewall mode is on and the policy table is loaded.
- **State** is in `~/.spinifex/latitude/<name>/`: Terraform state, `inputs`, `owned` (IDs of what exists), `step.*` markers, logs, the data disk UUID. `up`, `destroy` and `resolve-uncertain` hold a lock in the state directory (`lock.d`, a pid; a dead pid is taken over), so two runs cannot overlap. A failed `up` resumes at the failed step; delete a `step.<name>.nodeN` file to redo it. The AWS profile is `spinifex-latitude-<name>`, refused if either AWS file already has a section for it (any profile type) unless it equals exactly what `up` would write, or `--overwrite-profile` is given; the check runs again right before writing, with the CA at `~/.spinifex/latitude-<name>-ca.pem`. The secret key is written to `~/.aws/credentials` (0600, other profiles kept, the old file and `~/.aws/config` saved beside it) and never appears in a process argument.
## Token handling
The token is read from the environment by the provider and by the script's API calls. It is not a Terraform variable, is on no command line (curl reads it from stdin), is not sent to the node, and is written to no file. `TF_LOG` is unset by the script.
## Recovery
`up` prints the recovery text before the console check and again before the firewall step. With the firewall on, a lost ssh session is recovered from the latitude.sh serial console, `ssh <server_id>@<ip> -p 2222`, using the ubuntu password set in the console check. The printed disarm stops the daemon, deletes the `spinifex_filter` table and moves `peers.nft` aside. Do not reboot the node until `up` has finished or the disarm has been used: the boot unit reloads the policy from `peers.nft`.
## destroy
`destroy` builds the set to delete once, from the `owned` file plus the IDs in the Terraform state, shows it, and asks you to type the deployment name. It deletes exactly that set: `terraform destroy` first, then the API for anything left, in the order servers, Elastic IPs, keys, user data, project. If the set changes after the preview it stops and asks you to run it again.
Objects whose names match the deployment but that were never recorded are listed as `untracked` and never deleted, and `destroy` exits 1 until you remove them by hand. If an `up` failed or was interrupted while creating (`uncertain` marker), an Elastic IP may exist that nothing can attribute: `destroy` lists everything it cannot attribute and gives no all-clear. The record is append-only; after you check the dashboard, `./deploy-latitude.sh resolve-uncertain --name <name>` clears it on a typed confirmation. The preview also lists every ID in the Terraform state that the API listing does not show (`absent`), and the state checksum is rechecked before `terraform destroy` runs. It also fails on any API error, unreadable list, unfollowed pagination or refused DELETE. Elastic IP release is asynchronous and only works once the address is active or in error, so a first sweep round may need to wait.
## Unverified
- Pagination: the API docs give only `page[size]` (default 20) and `page[number]` and no response metadata. The lists follow `links.next` if the body has one, and otherwise read page numbers until an empty page.
- The OS slug `ubuntu_26_04_x64_lts` (the `latitudesh_operating_system` data source fails the plan if it is wrong; list slugs with `GET /plans/operating_systems`).
- The DELETE paths for `/user_data/{id}` and `/projects/{id}`, and the shape of the `server` field on Elastic IP list items used by `leftovers`.
- The `systemd-networkd-wait-online --any` drop-in path and effect (it first applies on the next boot).
- That the Spinifex API certificate names the node's public address, which `write_profile` needs for `aws sts get-caller-identity` to pass.
- Terraform owning the Elastic IP `server_id` fights a Spinifex-managed move; fine for one node, not for three.
## Not exercised
Any real API call, `up`, `destroy`, `host-data-disk.sh` on a host, the console check, the `--armed` start of `install-node.sh`, and the ssh-rule gate against a real `setup.sh` policy. Guest launch, public reachability of the Elastic IP, the Terraform workbooks and failover are not part of `up`; prove them by hand.
@@ -0,0 +1,20 @@
#cloud-config
# Host preparation only. Spinifex is installed and formed over SSH by deploy-latitude.sh.
# latitude.sh substitutes double-brace variables in user data, so this file holds none.
write_files:
- path: /etc/cloud/cloud.cfg.d/99-spinifex-network.cfg
content: |
network: {config: disabled}
# The image's netplan lists eno2 with no address and without optional, so boot waits two minutes
# for it. --any returns once one link, eno1, is online; it changes the wait, not the network.
# UNVERIFIED: the binary path is the Ubuntu one and the drop-in first applies on the next boot.
- path: /etc/systemd/system/systemd-networkd-wait-online.service.d/10-spinifex-any.conf
content: |
[Service]
ExecStart=
ExecStart=/usr/lib/systemd/systemd-networkd-wait-online --any
# runcmd carries on past a failing command, so each line stands alone.
runcmd:
- [sh, -c, "systemctl disable --now unattended-upgrades.service apt-daily-upgrade.timer apt-daily.timer || true"]
- [sh, -c, "systemctl daemon-reload || true"]
- [sh, -c, "touch /var/lib/cloud/spinifex-host-ready"]
+59
View File
@@ -0,0 +1,59 @@
locals {
hostname_prefix = "spx-${var.deployment_name}-node"
project = var.project != "" ? var.project : latitudesh_project.deploy[0].slug
}
# Both lookups fail the plan when a slug does not exist, which is the only check of the OS slug.
data "latitudesh_operating_system" "os" {
slug = var.operating_system
}
data "latitudesh_plan" "plan" {
slug = var.plan
}
resource "latitudesh_project" "deploy" {
count = var.project == "" ? 1 : 0
name = "spinifex-${var.deployment_name}"
description = "Spinifex deployment ${var.deployment_name}"
environment = "Development"
provisioning_type = "on_demand"
}
# The key is team scoped. The server login user is ubuntu, not root.
resource "latitudesh_ssh_key" "deploy" {
name = "spinifex-${var.deployment_name}"
public_key = trimspace(file(pathexpand(var.ssh_public_key_path)))
}
# Host preparation only. Latitude substitutes double-brace variables inside user data, so the file
# must hold none. Spinifex is installed over SSH by deploy-latitude.sh.
resource "latitudesh_user_data" "host" {
description = "spinifex-${var.deployment_name}"
content = filebase64("${path.module}/cloud-init/user-data.yaml")
}
# billing must be hourly: the default is monthly, which bills upfront with no refund.
resource "latitudesh_server" "node" {
count = var.node_count
project = local.project
hostname = "${local.hostname_prefix}${count.index + 1}"
site = var.site
plan = data.latitudesh_plan.plan.slug
operating_system = data.latitudesh_operating_system.os.slug
billing = "hourly"
ssh_keys = [latitudesh_ssh_key.deploy.id]
user_data = latitudesh_user_data.host.id
timeouts = {
create = "60m"
}
}
# One Elastic IP routed to node 1. Terraform owning server_id fights a Spinifex-managed move of the
# address (scout warning); that is fine for one node, and a multi-node deployment must stop managing
# this attachment here before Spinifex moves it.
resource "latitudesh_elastic_ip" "node1" {
project = local.project
server_id = latitudesh_server.node[0].id
}
+798
View File
@@ -0,0 +1,798 @@
#!/usr/bin/env bash
# deploy-latitude.sh — build a one-node Spinifex deployment on latitude.sh bare metal, or tear it down.
#
# deploy-latitude.sh up one m4.metal.medium with an Elastic IP
# deploy-latitude.sh status nodes, addresses, what the deployment holds, steps done
# deploy-latitude.sh credentials print the AWS access key and secret
# deploy-latitude.sh ssh [nodeN] [command] SSH to a node, node1 by default
# deploy-latitude.sh destroy
# deploy-latitude.sh resolve-uncertain clear the uncertain-create record after checking the dashboard
#
# Run it on your laptop with LATITUDESH_AUTH_TOKEN exported. OpenTofu, or Terraform when OpenTofu is
# not installed, creates the project (unless --project), SSH key, user data, server and Elastic IP.
# This script then works over SSH as the ubuntu user: it formats and mounts the blank data disk,
# and drives install-node.sh (install, ovn, firewall arm, start, verify). Each step is recorded
# under the state directory, so a rerun of up resumes at the step that failed.
#
# The token is read by the Terraform provider and by this script's own API calls from the
# environment. It is never a Terraform variable, never on a command line (curl reads it from stdin),
# and never sent to the node, so it is in no state file, log or user data.
#
# The AWS profile is "spinifex-latitude-<name>" (default name "default"). An existing profile of that
# name that this script did not write is refused unless --overwrite-profile is given.
set -euo pipefail
shopt -s inherit_errexit
# A provider trace log would carry the Authorization header.
unset TF_LOG TF_LOG_PATH TF_LOG_CORE TF_LOG_PROVIDER
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SPINIFEX_SRC="$(cd "$SCRIPT_DIR/../../.." && pwd)"
LAT_API="https://api.latitude.sh"
CMD="${1:-}"
if [ $# -gt 0 ]; then shift; fi
NODE_ARG=""
NAME="default"
SITE="DAL"
PLAN="m4-metal-medium"
OS="ubuntu_26_04_x64_lts"
PROJECT=""
DISTRO=""
CHANNEL="latest"
VERSION=""
SPX_REGION="us-east-1"
SPX_AZ="us-east-1a"
SSH_KEY="$HOME/.ssh/id_ed25519"
IMAGE="ubuntu-26.04-x86_64"
DATA_DISK_ID=""
STATE_ROOT="$HOME/.spinifex/latitude"
NODES=1
YES=0
OVERWRITE_PROFILE=0
usage() {
sed -n '2,23p' "$0" | sed 's/^# \{0,1\}//'
cat <<EOF
Options:
--name NAME deployment name (default $NAME), 1-16 characters; names every resource
--site SLUG latitude.sh site (default $SITE); needs stock and Elastic IPs, not SAO
--plan SLUG server plan (default $PLAN)
--os SLUG operating system slug (default $OS, unverified)
--project ID|SLUG deploy into this existing project; by default one is created and removed on destroy
--distro TARBALL install this local release tarball
--channel CHANNEL install from a release channel: latest (default) or dev
--version TAG install this release instead of a channel
--spx-region NAME Spinifex region (default $SPX_REGION)
--spx-az NAME Spinifex availability zone (default $SPX_AZ)
--ssh-key PATH private key; PATH.pub is installed (default $SSH_KEY)
--image NAME AMI to import (default $IMAGE)
--no-image import no AMI
--data-disk-id PATH choose the data disk by its /dev/disk/by-id/nvme-eui.* path when several are blank
--nodes N must be 1 for now
--overwrite-profile replace an AWS profile of the same name that this script did not create
--yes destroy without asking
EOF
}
while [ $# -gt 0 ]; do
case "$1" in
--name) NAME="$2"; shift ;;
--site) SITE="$2"; shift ;;
--plan) PLAN="$2"; shift ;;
--os) OS="$2"; shift ;;
--project) PROJECT="$2"; shift ;;
--distro) DISTRO="$2"; shift ;;
--channel) CHANNEL="$2"; shift ;;
--version) VERSION="$2"; shift ;;
--spx-region) SPX_REGION="$2"; shift ;;
--spx-az) SPX_AZ="$2"; shift ;;
--ssh-key) SSH_KEY="$2"; shift ;;
--image) IMAGE="$2"; shift ;;
--no-image) IMAGE="" ;;
--data-disk-id) DATA_DISK_ID="$2"; shift ;;
--nodes) NODES="$2"; shift ;;
--overwrite-profile) OVERWRITE_PROFILE=1 ;;
--yes) YES=1 ;;
-h|--help) usage; exit 0 ;;
node[1-3])
case "$CMD" in ssh) ;; *) usage >&2; exit 2 ;; esac
[ -z "$NODE_ARG" ] || { usage >&2; exit 2; }
NODE_ARG="$1" ;;
*) if [ "$CMD" = ssh ]; then break; fi
echo "unknown option: $1" >&2; usage >&2; exit 2 ;;
esac
shift
done
log() { printf '[deploy %s] %s\n' "$(date -u +%H:%M:%S)" "$*" >&2; }
die() { log "ERROR: $*"; exit 1; }
[[ "$NAME" =~ ^[a-z0-9][a-z0-9-]{0,15}$ ]] || die "--name must be 1-16 lowercase letters, digits or hyphens"
[ "$NODES" = 1 ] || die "--nodes must be 1: three nodes need the VLAN and the Elastic IP move, which are not built yet"
[[ "$SITE" =~ ^[A-Za-z0-9]+$ ]] || die "--site must be a site slug such as DAL"
[[ "$SPX_REGION" =~ ^[a-z0-9-]+$ ]] || die "--spx-region must be lowercase letters, digits and hyphens"
[[ "$SPX_AZ" =~ ^[a-z0-9-]+$ ]] || die "--spx-az must be lowercase letters, digits and hyphens"
[[ "$CHANNEL" =~ ^[a-z0-9.-]+$ ]] || die "--channel must be lowercase letters, digits, dots and hyphens"
[ -z "$VERSION" ] || [[ "$VERSION" =~ ^[A-Za-z0-9._-]+$ ]] || die "--version has characters a release tag does not"
[ -z "$IMAGE" ] || [[ "$IMAGE" =~ ^[A-Za-z0-9._-]+$ ]] || die "--image has characters an image name does not"
[ -z "$DATA_DISK_ID" ] || [[ "$DATA_DISK_ID" =~ ^/dev/disk/by-id/nvme-eui\.[A-Za-z0-9]+$ ]] || die "--data-disk-id must look like /dev/disk/by-id/nvme-eui.<hex>"
case "$CMD" in up|destroy|status) [ -n "${LATITUDESH_AUTH_TOKEN:-}" ] || die "LATITUDESH_AUTH_TOKEN is not set" ;; esac
if command -v tofu >/dev/null; then TF_BIN=tofu
elif command -v terraform >/dev/null; then TF_BIN=terraform
else die "neither OpenTofu (tofu) nor Terraform is installed"; fi
STATE_DIR="$STATE_ROOT/$NAME"
TF_STATE="$STATE_DIR/terraform.tfstate"
KNOWN_HOSTS="$STATE_DIR/known_hosts"
INPUTS="$STATE_DIR/inputs"
OWNED="$STATE_DIR/owned"
# PENDING exists while an apply runs. If the apply fails or is interrupted it is appended to UNCERTAIN, which
# only resolve-uncertain removes: a create that reached latitude.sh but not the state cannot be attributed,
# so destroy gives no all-clear while UNCERTAIN exists.
UNCERTAIN="$STATE_DIR/uncertain"
PENDING="$STATE_DIR/uncertain.pending"
HOST_PREFIX="spx-$NAME-node"
PROFILE="spinifex-latitude-$NAME"
CA_FILE="$HOME/.spinifex/latitude-$NAME-ca.pem"
SSH_OPTS=(-i "$SSH_KEY" -o BatchMode=yes -o IdentitiesOnly=yes -o ConnectTimeout=10
-o StrictHostKeyChecking=accept-new -o "UserKnownHostsFile=$KNOWN_HOSTS")
mkdir -p "$STATE_DIR"
chmod 0700 "$STATE_DIR"
# After the first up, destroy, status and ssh use the project choice that up recorded.
if [ -z "$PROJECT" ] && [ -f "$INPUTS" ]; then PROJECT="$(sed -n 's/^project=//p' "$INPUTS")"; fi
# latitude METHOD PATH prints the response body and fails on a non-2xx. The token goes to curl on
# stdin as a config line, so it is on no command line; printf is a shell builtin.
latitude() {
local out code
out="$(printf 'header = "Authorization: Bearer %s"\n' "$LATITUDESH_AUTH_TOKEN" |
curl -sS -g -K - -w '\n%{http_code}' -X "$1" -H 'Accept: application/json' "$LAT_API$2")" || return 1
code="${out##*$'\n'}"
out="${out%$'\n'*}"
case "$code" in
2??) printf '%s' "$out" ;;
*) log "latitude.sh $1 $2 returned $code: $(printf '%s' "$out" | head -c 400)"; return 1 ;;
esac
}
# lat_list PATH [QUERY] prints a JSON array of every item of a paginated list. It follows links.next
# until it is null; where the body gives no next key (the documented API has no pagination metadata, only page[size] and page[number]) it reads page numbers until an empty page, so a server-capped page cannot pass as the last. It fails on
# an API error, on a body that is not {"data": [objects with an id and attributes]}, on a next link to
# another host, on a page that repeats an earlier one, and past 200 pages.
lat_list() {
local path="$1?${2:+$2&}page[size]=100&page[number]=1" out info n next page=1 all='[]'
while :; do
out="$(latitude GET "$path")" || return 1
info="$(python3 -I -c '
import json,sys
d = json.load(sys.stdin)
data = d.get("data") if isinstance(d, dict) else None
if not isinstance(data, list) or not all(isinstance(i, dict) and isinstance(i.get("id"), str) and isinstance(i.get("attributes"), dict) for i in data):
sys.exit("unexpected list schema")
links = d.get("links")
nxt = "?" if not isinstance(links, dict) or "next" not in links else (links["next"] or "-")
print(len(data)); print(nxt)' <<<"$out")" || { log "latitude.sh list $1 returned an unexpected body"; return 1; }
n="$(sed -n 1p <<<"$info")" next="$(sed -n 2p <<<"$info")"
all="$(python3 -I -c '
import json,sys
a = json.loads(sys.argv[1]); new = json.load(sys.stdin)["data"]
if new and any(i["id"] == new[0]["id"] for i in a): sys.exit("page repeats")
print(json.dumps(a + new))' "$all" <<<"$out")" || { log "latitude.sh list $1 does not advance between pages"; return 1; }
page=$((page + 1))
[ "$page" -le 200 ] || { log "latitude.sh list $1 has more than 200 pages"; return 1; }
case "$next" in
-) break ;;
"?") if [ "$n" = 0 ]; then break; fi
path="$1?${2:+$2&}page[size]=100&page[number]=$page" ;;
"$LAT_API"/*) path="${next#"$LAT_API"}" ;;
/*) path="$next" ;;
*) log "latitude.sh list $1 gave a next link to another host"; return 1 ;;
esac
done
printf '%s' "$all"
}
tf() {
"$TF_BIN" -chdir="$SCRIPT_DIR" "$@" -state="$TF_STATE" \
-var "deployment_name=$NAME" -var "site=$SITE" -var "plan=$PLAN" -var "operating_system=$OS" \
-var "project=$PROJECT" -var "ssh_public_key_path=$SSH_KEY.pub" -var "node_count=$NODES"
}
done_step() { [ -f "$STATE_DIR/step.$1" ]; }
mark_step() { date -u +%FT%TZ > "$STATE_DIR/step.$1"; }
# node_field FIELD prints FIELD of every node, in node order.
node_field() {
"$TF_BIN" -chdir="$SCRIPT_DIR" output -state="$TF_STATE" -json nodes |
python3 -c 'import json,sys; [print(n[sys.argv[1]]) for n in json.load(sys.stdin)]' "$1"
}
# for_nodes_always FN runs FN on every node and records nothing.
for_nodes_always() {
local fn="$1" i
for i in "${!IPS[@]}"; do
IP="${IPS[$i]}" SERVER_ID="${IDS[$i]}" NODE_NAME="node$((i + 1))"
"$fn"
done
}
# for_nodes STEP FN runs FN once per node with IP, SERVER_ID and NODE_NAME set, recording
# STEP.<node> so a rerun skips the nodes already done.
for_nodes() {
local step="$1" fn="$2" i
for i in "${!IPS[@]}"; do
IP="${IPS[$i]}" SERVER_ID="${IDS[$i]}" NODE_NAME="node$((i + 1))"
if done_step "$step.$NODE_NAME"; then continue; fi
"$fn"
mark_step "$step.$NODE_NAME"
done
}
# on_tty needs -tt before the destination so passwd gets a terminal.
on_tty() { ssh -tt "${SSH_OPTS[@]}" "ubuntu@$IP" "$@"; }
# shellcheck disable=SC2029 # the command is meant to expand here
on() { ssh "${SSH_OPTS[@]}" "ubuntu@$IP" "$@"; }
# acquire_lock stops two runs on one deployment, so resolve-uncertain cannot clear a record while an
# apply is writing it. mkdir is atomic and works on macOS, which has no flock; a lock whose pid is gone is
# stale and is taken over. Holding the lock means no apply is running, so a pending entry left by an
# interrupted one is made permanent here.
acquire_lock() {
local lock="$STATE_DIR/lock.d" pid
if ! mkdir "$lock" 2>/dev/null; then
pid="$(cat "$lock/pid" 2>/dev/null || true)"
if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then die "another run (pid $pid) is working on deployment '$NAME'"; fi
# A lock with no pid yet may be a run that has just made it: stale only once it is a minute old.
if [ -z "$pid" ] && [ -z "$(find "$lock" -maxdepth 0 -mmin +1 2>/dev/null)" ]; then die "another run is starting on deployment '$NAME'"; fi
rm -rf "$lock"
mkdir "$lock" || die "could not take the lock $lock"
fi
printf '%s\n' "$$" > "$lock/pid"
trap 'rm -rf "$STATE_DIR/lock.d"' EXIT
if [ -f "$PENDING" ]; then cat "$PENDING" >> "$UNCERTAIN"; rm -f "$PENDING"; fi
}
# ── up ───────────────────────────────────────────────────────────────────────
record_inputs() {
local now
now="$(printf 'site=%s\nplan=%s\nos=%s\nproject=%s\nnodes=%s' "$SITE" "$PLAN" "$OS" "$PROJECT" "$NODES")"
if [ -f "$INPUTS" ]; then
[ "$(cat "$INPUTS")" = "$now" ] || die "inputs differ from the first up of '$NAME' (recorded in $INPUTS); use the same inputs, or destroy it first"
else
printf '%s\n' "$now" > "$INPUTS"
fi
}
preflight() {
[ -r "$SSH_KEY" ] || die "SSH key $SSH_KEY not readable"
[ -r "$SSH_KEY.pub" ] || die "SSH key $SSH_KEY.pub not readable"
command -v aws >/dev/null || die "the AWS CLI is not installed"
command -v python3 >/dev/null || die "python3 is not installed"
if [ -n "$DISTRO" ]; then [ -r "$DISTRO" ] || die "--distro $DISTRO not readable"; fi
latitude GET '/projects?page[size]=1' >/dev/null || die "the API refused the token (LATITUDESH_AUTH_TOKEN)"
if [ -n "$PROJECT" ]; then
lat_list /projects "filter[slug]=$PROJECT" | python3 -c '
import json,sys
sys.exit(0 if any(i["attributes"]["slug"] == sys.argv[1] for i in json.load(sys.stdin)) else 1)' "$PROJECT" ||
die "no project with slug $PROJECT; use its slug, or leave --project out to create one"
fi
record_inputs
# A fresh deployment refuses an existing profile now; a rerun is checked again, with the real values,
# right before the profile is written.
if [ ! -f "$TF_STATE" ] && [ "$OVERWRITE_PROFILE" != 1 ] && [ "$(profile_state)" != none ]; then
die "AWS profile $PROFILE already has a section in $(AWS_CRED_FILE) or $(AWS_CONF_FILE); pass --overwrite-profile to replace it (both files are saved first), or use another --name"
fi
}
provision() {
log "terraform apply ($PLAN at $SITE)"
"$TF_BIN" -chdir="$SCRIPT_DIR" init -input=false >/dev/null
# A pending entry left by an interrupted apply becomes a permanent one. Entries are only appended.
if [ -f "$PENDING" ]; then cat "$PENDING" >> "$UNCERTAIN"; rm -f "$PENDING"; fi
if [ -f "$UNCERTAIN" ]; then
log "an earlier create failed or was interrupted ($UNCERTAIN); destroy gives no all-clear until you check the dashboard and run: $0 resolve-uncertain --name $NAME"
fi
date -u +%FT%TZ > "$PENDING"
tf apply -input=false -auto-approve >"$STATE_DIR/apply.log" 2>&1 || {
tail -20 "$STATE_DIR/apply.log" >&2
{ cat "$PENDING" 2>/dev/null || true; date -u +%FT%TZ; } >> "$UNCERTAIN"
rm -f "$PENDING"
record_owned || true
die "terraform apply failed; log in $STATE_DIR/apply.log. What was created is recorded in $OWNED and destroy removes it. A plan with no stock at the site fails here: try another --site"
}
record_owned
rm -f "$PENDING"
}
# record_owned adds the IDs of every latitude.sh object in the Terraform state to $OWNED, as
# "kind id label". Terraform writes state as each object is created, so this also covers a partly
# failed apply. destroy deletes only what is listed here.
record_owned() {
[ -f "$TF_STATE" ] || return 0
local out
out="$("$TF_BIN" -chdir="$SCRIPT_DIR" show -json "$TF_STATE" | python3 -I -c '
import json,sys
kinds = {"latitudesh_server": "server", "latitudesh_elastic_ip": "elastic-ip", "latitudesh_ssh_key": "ssh-key",
"latitudesh_user_data": "user-data", "latitudesh_project": "project"}
st = json.load(sys.stdin)
root = (st.get("values") or {}).get("root_module") or {}
for r in root.get("resources", []):
k = kinds.get(r.get("type"))
v = r.get("values") or {}
if k and v.get("id"):
print(k, v["id"], v.get("hostname") or v.get("address") or v.get("name") or v.get("description") or "-")')" ||
die "could not read the Terraform state to record what exists"
{ [ ! -f "$OWNED" ] || cat "$OWNED"; printf '%s\n' "$out"; } | sed '/^$/d' | sort -u > "$OWNED.new"
chmod 0600 "$OWNED.new"
mv "$OWNED.new" "$OWNED"
}
# wait_host pins the host key only once the marker cloud-init writes exists.
wait_host() {
log "waiting for SSH and cloud-init on $NODE_NAME"
local i
for i in $(seq 1 120); do
if ssh -i "$SSH_KEY" -o BatchMode=yes -o IdentitiesOnly=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=no \
-o UserKnownHostsFile=/dev/null -o LogLevel=ERROR "ubuntu@$IP" \
'test -f /var/lib/cloud/spinifex-host-ready' 2>/dev/null; then break; fi
[ "$i" != 120 ] || die "host $IP was not ready after 20 minutes"
sleep 10
done
ssh-keygen -R "$IP" -f "$KNOWN_HOSTS" >/dev/null 2>&1 || true
on 'cloud-init status --wait >/dev/null; test -e /dev/kvm' || die "no /dev/kvm on $NODE_NAME"
}
# host_prep is the first change to the node, so it runs after the console gate. The image's resolv.conf
# is not a link to systemd-resolved, so the node would bypass the split DNS Spinifex sets there for its
# load balancer and EKS names.
host_prep() {
on 'if [ ! -L /etc/resolv.conf ] && systemctl is-active -q systemd-resolved; then
sudo ln -sf ../run/systemd/resolve/stub-resolv.conf /etc/resolv.conf; fi' ||
die "could not point /etc/resolv.conf at systemd-resolved on $NODE_NAME"
}
# confirm_if_pending asks for the console only when a step that changes the node is still to run.
confirm_if_pending() {
local st
for st in prep disk install ovn firewall start; do
if ! done_step "$st.$NODE_NAME"; then console_gate; return 0; fi
done
}
# console_gate runs before the first change to the node's disks, network or firewall. The person sets
# the ubuntu password here (typed into passwd on the node, never seen by this script), then proves the
# serial console works from another window, because that console is the only way back in.
console_gate() {
log "console recovery check for $NODE_NAME, before any change to its disks or firewall"
print_recovery >&2
[ -r /dev/tty ] || die "no terminal: the console check needs a typed confirmation"
printf '\nSet the ubuntu password now (typed on the node, not seen by this script):\n' >&2
on_tty 'sudo passwd ubuntu' < /dev/tty || die "setting the ubuntu password failed"
printf '\nIn ANOTHER window run: ssh %s@%s -p 2222 and log in as ubuntu with that password.\n' "$SERVER_ID" "$IP" >&2
printf 'Type "console-ok" once you are logged in over the serial console: ' >&2
local answer
read -r answer < /dev/tty || die "no answer"
[ "$answer" = console-ok ] || die "not confirmed; nothing has been changed on $NODE_NAME. Rerun up once the console works"
}
# data_disk formats the blank data disk, chosen by content on the node, and records its UUID.
data_disk() {
log "preparing the data disk on $NODE_NAME"
local tmp rc=0 uuid
scp -q "${SSH_OPTS[@]}" "$SCRIPT_DIR/host-data-disk.sh" "ubuntu@$IP:/tmp/host-data-disk.sh" || die "copy to $NODE_NAME failed"
tmp="$(mktemp "$STATE_DIR/data-disk.XXXXXX")"
on "sudo bash /tmp/host-data-disk.sh $DATA_DISK_ID" >"$tmp" 2>&1 || rc=$?
cat "$tmp" >&2
cat "$tmp" >> "$STATE_DIR/data-disk.$NODE_NAME.log"
uuid="$(sed -n 's/^DATA_UUID=//p' "$tmp" | tail -n1)"
rm -f "$tmp"
[ "$rc" = 0 ] || die "the data disk step failed (exit $rc); it may already have formatted the disk, so read the output above and $STATE_DIR/data-disk.$NODE_NAME.log before rerunning"
[[ "$uuid" =~ ^[0-9a-fA-F-]{16,36}$ ]] || die "the data disk step gave no UUID"
printf '%s\n' "$uuid" > "$STATE_DIR/data_uuid.$NODE_NAME"
}
# push_installer copies install-node.sh (and a local tarball with its setup.sh) to the node and
# proves the installer arrived intact.
push_installer() {
local want got
scp -q "${SSH_OPTS[@]}" "$SCRIPT_DIR/install-node.sh" "ubuntu@$IP:/tmp/install-node.sh" || die "copy to $NODE_NAME failed"
want="$(sha256sum "$SCRIPT_DIR/install-node.sh" | awk '{print $1}')"
got="$(on 'sha256sum /tmp/install-node.sh' | awk '{print $1}')"
[ "$want" = "$got" ] || die "install-node.sh changed in transit: local $want, node $got"
if [ -n "$DISTRO" ]; then
scp -q "${SSH_OPTS[@]}" "$DISTRO" "$SPINIFEX_SRC/scripts/setup.sh" "ubuntu@$IP:/tmp/" || die "copy of the tarball to $NODE_NAME failed"
fi
}
# install_args is every input install-node.sh records and a later phase must repeat.
install_args() {
local uuid eip
uuid="$(cat "$STATE_DIR/data_uuid.$NODE_NAME")"
eip="$(cat "$STATE_DIR/eip")"
printf -- '--eip %s --data-uuid %s --node-name %s --spx-region %s --spx-az %s --channel %s' \
"$eip" "$uuid" "$NODE_NAME" "$SPX_REGION" "$SPX_AZ" "$CHANNEL"
if [ -n "$VERSION" ]; then printf -- ' --version %s' "$VERSION"; fi
if [ -n "$DISTRO" ]; then printf -- ' --distro /tmp/%s --setup-script /tmp/setup.sh' "$(basename "$DISTRO")"; fi
}
# run_phase PHASE [EXTRA...] runs one install-node.sh phase on the node as root.
run_phase() {
local phase="$1"
shift
push_installer
log "install-node.sh --phase $phase on $NODE_NAME"
on "sudo bash /tmp/install-node.sh --phase $phase $(install_args) $*" 2>&1 | tee -a "$STATE_DIR/install-node.$NODE_NAME.log" >&2 ||
die "install-node.sh --phase $phase failed; the output above and $STATE_DIR/install-node.$NODE_NAME.log have it, and the node has /root/<step>.log"
}
phase_install() { run_phase install; }
phase_ovn() { run_phase ovn; }
# arm_firewall runs after the ovn phase and before Spinifex first starts, so the daemon never runs
# unfiltered. The daemon writes peers.nft and loads the policy during its start. The start phase
# refuses to go on unless the generated policy allows ssh: it reads the active rule, the defined ports
# against sshd -T, and the 0.0.0.0/0 trust.
arm_firewall() {
log "arming the firewall on $NODE_NAME before the first start"
print_recovery >&2
log "Do not reboot $NODE_NAME until this run has finished, or until the console disarm above has been used"
on 'sudo bash -c "env SETUP_STAGES=firewall /usr/local/share/spinifex/setup.sh --firewall on >/root/firewall.log 2>&1"' ||
die "arming the firewall failed; see /root/firewall.log on $NODE_NAME. Spinifex was not started"
}
phase_start() {
if [ -n "$IMAGE" ]; then run_phase start --armed --image "$IMAGE"; else run_phase start --armed; fi
}
phase_verify() { run_phase verify; }
# check_access proves the firewalled node still lets a NEW ssh login and the API through from here.
check_access() {
log "checking a new ssh login and the API through the firewall"
local i code=""
ssh "${SSH_OPTS[@]}" -o ControlMaster=no -o ControlPath=none "ubuntu@$IP" true ||
die "a new ssh login to $NODE_NAME failed after the firewall loaded; recover from the serial console (printed at the end of a good run; also: ssh $SERVER_ID@$IP -p 2222)"
for i in $(seq 1 12); do
code="$(curl -sk -m 10 -o /dev/null -w '%{http_code}' "https://$IP:9999/" || true)"
case "$code" in 2??|3??|4??) break ;; esac
sleep 5
done
case "$code" in
2??|3??|4??) ;;
*) die "the API on https://$IP:9999 did not answer from here (last status '$code'); ssh works, so check the firewall rules for tcp 9999" ;;
esac
on 'sudo bash -c "grep -qx on /etc/spinifex/firewall/mode && nft list table inet spinifex_filter >/dev/null"' ||
die "the firewall is not armed (mode not on or table inet spinifex_filter missing)"
log "access proven with the firewall armed"
}
AWS_CRED_FILE() { printf '%s' "${AWS_SHARED_CREDENTIALS_FILE:-$HOME/.aws/credentials}"; }
AWS_CONF_FILE() { printf '%s' "${AWS_CONFIG_FILE:-$HOME/.aws/config}"; }
# profile_state [KEY SECRET ENDPOINT] prints none when the profile has no section in either AWS file
# (credentials: [name], config: [profile name]) whatever its type (static keys, SSO, credential_process).
# With the values up would write it prints match when the sections equal exactly those, else conflict;
# without them it prints exists. The secret reaches python on stdin.
profile_state() {
printf '%s\n%s\n%s\n' "${1:-}" "${2:-}" "${3:-}" | python3 -I -c '
import configparser, os, sys
cred, conf, name, ca, region = sys.argv[1:6]
key, secret, endpoint = sys.stdin.read().split("\n")[:3]
def sect(path, sec):
cp = configparser.RawConfigParser()
cp.optionxform = str
if os.path.exists(path):
cp.read(path)
return dict(cp.items(sec)) if cp.has_section(sec) else None
c = sect(cred, name)
g = sect(conf, "profile " + name)
if c is None and g is None:
print("none")
elif not key:
print("exists")
else:
want_c = {"aws_access_key_id": key, "aws_secret_access_key": secret}
want_g = {"region": region, "endpoint_url": endpoint, "ca_bundle": ca}
print("match" if c == want_c and g == want_g else "conflict")' "$(AWS_CRED_FILE)" "$(AWS_CONF_FILE)" "$PROFILE" "$CA_FILE" "$SPX_REGION"
}
# write_credentials KEY SECRET sets the profile's keys in the shared credentials file without putting
# the secret in any process argument: it reaches python on stdin. Other profiles are kept (comments in
# the file are not), the old file is saved beside it, and the new one is 0600.
write_credentials() {
local code
code='
import configparser, os, shutil, sys, time
profile = sys.argv[1]
key, secret = sys.stdin.read().split("\n")[:2]
path = os.environ.get("AWS_SHARED_CREDENTIALS_FILE") or os.path.expanduser("~/.aws/credentials")
os.umask(0o077)
os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True)
cp = configparser.RawConfigParser()
cp.optionxform = str
if os.path.exists(path):
cp.read(path)
shutil.copy2(path, "%s.pre-latitude.%d" % (path, int(time.time())))
if not cp.has_section(profile):
cp.add_section(profile)
cp.set(profile, "aws_access_key_id", key)
cp.set(profile, "aws_secret_access_key", secret)
cp.remove_option(profile, "aws_session_token")
tmp = path + ".tmp"
with open(tmp, "w") as f:
cp.write(f)
os.chmod(tmp, 0o600)
os.replace(tmp, path)
'
printf '%s\n%s\n' "$1" "$2" | python3 -I -c "$code" "$PROFILE" || die "could not write the credentials file"
}
write_profile() {
log "writing the CA to $CA_FILE and AWS profile $PROFILE"
mkdir -p "$(dirname "$CA_FILE")"
local creds key secret ca_new
# The CA is fetched to a temporary file and checked first; the existing file is replaced only after
# the profile check passes, with the old one kept beside it.
ca_new="$(mktemp "$CA_FILE.new.XXXXXX")"
on 'sudo cat /etc/spinifex/ca.pem' > "$ca_new" || { rm -f "$ca_new"; die "could not read the node's CA"; }
grep -q -- '-----BEGIN CERTIFICATE-----' "$ca_new" || { rm -f "$ca_new"; die "the node's CA is empty or not a PEM certificate"; }
creds="$(on 'awk "/^\[spinifex\]/{f=1;next} /^\[/{f=0} f" ~/.aws/credentials')"
key="$(awk -F' *= *' '/aws_access_key_id/{print $2}' <<<"$creds")"
secret="$(awk -F' *= *' '/aws_secret_access_key/{print $2}' <<<"$creds")"
[ -n "$key" ] || die "no access key in the ubuntu user's ~/.aws/credentials on $NODE_NAME"
[ -n "$secret" ] || die "no [spinifex] credentials in the ubuntu user's ~/.aws/credentials on $NODE_NAME"
# Rechecked here, with the real values, immediately before anything is written.
case "$(profile_state "$key" "$secret" "https://$IP:9999")" in
none | match) ;;
*) [ "$OVERWRITE_PROFILE" = 1 ] || { rm -f "$ca_new"; die "AWS profile $PROFILE has a section in $(AWS_CRED_FILE) or $(AWS_CONF_FILE) that differs from what this run would write; pass --overwrite-profile to replace it (both files are saved first), or use another --name"; } ;;
esac
if [ -f "$CA_FILE" ] && ! cmp -s "$CA_FILE" "$ca_new"; then cp -p "$CA_FILE" "$CA_FILE.pre-latitude.$(date +%s)"; fi
chmod 0644 "$ca_new"
mv -f "$ca_new" "$CA_FILE"
write_credentials "$key" "$secret"
local cfg
cfg="$(AWS_CONF_FILE)"
if [ -f "$cfg" ]; then (umask 077; cp -p "$cfg" "$cfg.pre-latitude.$(date +%s)"); fi
aws configure set --profile "$PROFILE" region "$SPX_REGION"
aws configure set --profile "$PROFILE" endpoint_url "https://$IP:9999"
aws configure set --profile "$PROFILE" ca_bundle "$CA_FILE"
# The certificate must name the node's address for this to pass; that is not verified on latitude.sh before this script.
aws --profile "$PROFILE" sts get-caller-identity --query Account --output text >/dev/null ||
die "profile $PROFILE does not work against https://$IP:9999; check the API certificate names $IP"
}
print_recovery() {
printf '\nRecovery if ssh is ever lost (the firewall is on):\n'
printf ' Serial console: ssh %s@%s -p 2222 (needs an account SSH key and an OS user password: set one now with: %s ssh %s sudo passwd ubuntu)\n' \
"$SERVER_ID" "$IP" "$0" "$NODE_NAME"
printf ' Temporary disarm, in this order, from the console:\n'
printf ' sudo systemctl stop spinifex-daemon\n'
printf ' systemctl is-active spinifex-daemon (expect: inactive)\n'
printf ' sudo nft delete table inet spinifex_filter\n'
printf ' sudo mv /etc/spinifex/firewall/peers.nft /root/peers.nft.disarmed\n'
printf ' sudo nft list tables (must not list spinifex_filter)\n'
printf ' Do not reboot or restart Spinifex afterwards until a durable disarm is reviewed.\n'
}
cmd_up() {
preflight
if ! done_step terraform; then provision; mark_step terraform; fi
mapfile -t IPS < <(node_field public_ip)
mapfile -t IDS < <(node_field provider_host_id)
[ "${#IPS[@]}" = "$NODES" ] || die "the state holds ${#IPS[@]} node(s), not $NODES; destroy it or pass --nodes ${#IPS[@]}"
local eip
eip="$("$TF_BIN" -chdir="$SCRIPT_DIR" output -state="$TF_STATE" -raw elastic_ip)"
[[ "$eip" =~ ^([0-9]{1,3}\.){3}[0-9]{1,3}$ ]] || die "Terraform reported no usable Elastic IP address ('$eip')"
printf '%s\n' "$eip" > "$STATE_DIR/eip"
for_nodes host wait_host
# The console is confirmed on every up that still has a risky step to run, not once per deployment.
for_nodes_always confirm_if_pending
for_nodes prep host_prep
for_nodes disk data_disk
for_nodes install phase_install
for_nodes ovn phase_ovn
for_nodes firewall arm_firewall
for_nodes start phase_start
# Never skipped: the claim of success is always checked live by this run.
for_nodes_always phase_verify
for_nodes_always check_access
IP="${IPS[0]}" SERVER_ID="${IDS[0]}" NODE_NAME=node1
write_profile
log "ready: https://$IP:9999 (profile $PROFILE)"
printf 'endpoint=https://%s:9999\nprofile=%s\nca=%s\nconsole=https://%s:3000/\nelastic_ip=%s\n' "$IP" "$PROFILE" "$CA_FILE" "$IP" "$eip"
local i
for i in "${!IPS[@]}"; do printf 'node%d=%s\n' "$((i + 1))" "${IPS[$i]}"; done
print_recovery
printf '\nVerified by this run: the install phases, the firewall policy loaded, the ssh login and API reachable from this host, and a working AWS profile. Not shown: a guest launch and public reachability of the Elastic IP; test those.\n'
}
# ── destroy ──────────────────────────────────────────────────────────────────
# inventory prints "owned kind id label" for every object recorded in $OWNED that still exists, and
# "untracked kind id label" for every object whose name matches this deployment but is not recorded.
# With UNCERTAIN present it also prints "unattributed kind id label" for everything else in the account.
# It reads full lists, fails on any API or schema error, and never reads a failure as empty.
inventory() {
local d eip="" rc=0 uncertain=0
d="$(mktemp -d "$STATE_DIR/inv.XXXXXX")"
if [ -f "$STATE_DIR/eip" ]; then eip="$(cat "$STATE_DIR/eip")"; fi
if [ -f "$UNCERTAIN" ] || [ -f "$PENDING" ]; then uncertain=1; fi
{
lat_list /servers > "$d/server" &&
lat_list /elastic_ips > "$d/elastic-ip" &&
lat_list /ssh_keys > "$d/ssh-key" &&
lat_list /user_data > "$d/user-data" &&
lat_list /projects > "$d/project"
} || rc=1
if [ "$rc" = 0 ]; then
touch "$OWNED"
python3 -I -c '
import json,sys,os
d, owned_path, prefix, name, eip, own_project, uncertain = sys.argv[1:8]
owned = set()
for line in open(owned_path):
f = line.split()
if len(f) >= 2: owned.add((f[0], f[1]))
seen = set()
def label(a):
return a.get("hostname") or a.get("address") or a.get("name") or a.get("description") or "-"
for kind in ("server", "elastic-ip", "ssh-key", "user-data", "project"):
for it in json.load(open(os.path.join(d, kind))):
a = it["attributes"]
if (kind, it["id"]) in owned:
seen.add((kind, it["id"]))
print("owned", kind, it["id"], label(a)); continue
match = False
if kind == "server": match = str(a.get("hostname", "")).startswith(prefix)
elif kind == "elastic-ip":
host = ((a.get("server") or {}).get("hostname")) or ""
match = bool(eip and a.get("address") == eip) or host.startswith(prefix)
elif kind == "ssh-key": match = a.get("name") == "spinifex-" + name
elif kind == "project": match = own_project == "1" and a.get("name") == "spinifex-" + name
elif kind == "user-data": match = a.get("description") == "spinifex-" + name
if match: print("untracked", kind, it["id"], label(a))
elif uncertain == "1": print("unattributed", kind, it["id"], label(a))
for line in open(owned_path):
f = line.split()
if len(f) >= 2 and (f[0], f[1]) not in seen: print("absent", f[0], f[1], " ".join(f[2:]) or "-")
' "$d" "$OWNED" "$HOST_PREFIX" "$NAME" "$eip" "$([ -z "$PROJECT" ] && echo 1 || echo 0)" "$uncertain" || rc=1
fi
find "$d" -type f -delete
rmdir "$d"
return "$rc"
}
# state_sum identifies the Terraform state that destroy will act on.
state_sum() { if [ -f "$TF_STATE" ]; then sha256sum < "$TF_STATE"; else echo none; fi; }
# owned_sum identifies the recorded set, so a change after the confirmation is noticed.
owned_sum() { if [ -f "$OWNED" ]; then sha256sum < "$OWNED"; else echo none; fi; }
cmd_destroy() {
local inv set before ssum kind id label i answer failed=0
# The deletion set is the recorded IDs plus the IDs in the Terraform state, built once and shown.
record_owned
ssum="$(state_sum)"
set="$(inventory)" || die "could not list the account; nothing was deleted and nothing is proven"
before="$(owned_sum)"
if [ "$YES" != 1 ]; then
log "deployment '$NAME': every recorded object will be deleted ('owned' exists, 'absent' is recorded but not in the listing, and terraform destroy acts on it too). 'untracked' and 'unattributed' ones will NOT be:"
printf '%s\n' "$set" >&2
read -r -p "Type the deployment name to confirm: " answer < /dev/tty
[ "$answer" = "$NAME" ] || die "not confirmed"
fi
record_owned
[ "$(state_sum)" = "$ssum" ] || die "the Terraform state changed after the preview; run destroy again to confirm what it holds now"
[ "$(owned_sum)" = "$before" ] || die "the set to delete changed after the preview; run destroy again to confirm the new set"
if [ -f "$TF_STATE" ]; then
log "terraform destroy"
"$TF_BIN" -chdir="$SCRIPT_DIR" init -input=false >/dev/null
tf destroy -input=false -auto-approve >"$STATE_DIR/destroy.log" 2>&1 || {
tail -20 "$STATE_DIR/destroy.log" >&2
die "terraform destroy failed; log in $STATE_DIR/destroy.log"
}
fi
# What Terraform did not remove: delete only members of the shown set, in dependency order. The paths
# are from the API reference (servers, elastic_ips, ssh_keys, user_data, projects), untested live.
for i in 1 2 3 4 5 6; do
inv="$(inventory)" || die "could not list what remains; treat everything as still billing and check the dashboard"
grep -q '^owned ' <<<"$inv" || break
for kind in server elastic-ip ssh-key user-data project; do
while read -r _ k id label; do
[ "$k" = "$kind" ] || continue
grep -q "^owned $kind $id " <<<"$set" || die "$kind $id is recorded but was not in the confirmed set; run destroy again"
log "deleting $kind $label ($id)"
case "$kind" in
server) latitude DELETE "/servers/$id" >/dev/null || failed=1 ;;
elastic-ip) latitude DELETE "/elastic_ips/$id" >/dev/null || failed=1 ;;
ssh-key) latitude DELETE "/ssh_keys/$id" >/dev/null || failed=1 ;;
user-data) latitude DELETE "/user_data/$id" >/dev/null || failed=1 ;;
project) latitude DELETE "/projects/$id" >/dev/null || failed=1 ;;
esac
done < <(grep '^owned ' <<<"$inv")
done
sleep 15
done
inv="$(inventory)" || die "could not list what remains; treat everything as still billing and check the dashboard"
if grep -q '^owned ' <<<"$inv"; then
log "these recorded objects still exist and may be billing:"
grep '^owned ' <<<"$inv" >&2
exit 1
fi
[ "$failed" = 0 ] || die "a DELETE was refused (see above) although the objects are now absent; check the dashboard, then run destroy again"
if grep -q '^untracked ' <<<"$inv"; then
log "recorded objects are gone, but these name-matched objects are not recorded for this deployment. NOT deleted; check each in the dashboard and remove it by hand if it is yours:"
grep '^untracked ' <<<"$inv" >&2
exit 1
fi
if [ -f "$UNCERTAIN" ] || [ -f "$PENDING" ]; then
log "an earlier create failed or was interrupted before its IDs were recorded ($UNCERTAIN), so an Elastic IP or server may exist that this script cannot attribute. Everything it cannot attribute:"
grep '^unattributed ' <<<"$inv" >&2 || true
log "check the latitude.sh dashboard; delete anything of this deployment by hand, then run: $0 resolve-uncertain --name $NAME. No all-clear is given."
exit 1
fi
rm -f "$STATE_DIR"/step.* "$STATE_DIR"/eip "$STATE_DIR"/data_uuid.* "$INPUTS" "$OWNED" "$TF_STATE" "$TF_STATE.backup"
log "deployment '$NAME' destroyed: every recorded object is absent from complete account listings and none is name-matched. Objects never recorded and named differently are not covered by this check"
}
cmd_status() {
local nodes inv
if [ -f "$TF_STATE" ] && nodes="$("$TF_BIN" -chdir="$SCRIPT_DIR" output -state="$TF_STATE" -json nodes 2>/dev/null)"; then
python3 -I -c 'import json,sys; [print(n["name"], n["public_ip"]) for n in json.load(sys.stdin)]' <<<"$nodes"
fi
if inv="$(inventory)"; then
printf '%s\n' "$inv"
else
echo "UNKNOWN: the account could not be listed"
fi
find "$STATE_DIR" -maxdepth 1 -name 'step.*' | sort | sed 's/.*step\./done: /'
}
cmd_resolve_uncertain() {
local answer
[ -f "$UNCERTAIN" ] || [ -f "$PENDING" ] || { log "nothing uncertain is recorded for '$NAME'"; return 0; }
log "recorded uncertain creates:"
cat "$UNCERTAIN" "$PENDING" 2>/dev/null >&2 || true
[ -r /dev/tty ] || die "no terminal for the confirmation"
read -r -p 'Type "checked" once you have looked in the latitude.sh dashboard (servers, Elastic IPs, keys, user data, projects) and removed anything of this deployment that was not recorded: ' answer < /dev/tty
[ "$answer" = checked ] || die "not confirmed"
rm -f "$UNCERTAIN" "$PENDING"
log "uncertainty cleared for '$NAME'"
}
# node_ip NODE prints the public address of nodeN from the Terraform state.
node_ip() {
[ -f "$TF_STATE" ] || die "no deployment '$NAME'; run up first"
local ip
ip="$(node_field public_ip | sed -n "${1#node}p")"
[ -n "$ip" ] || die "deployment '$NAME' has no $1"
printf '%s' "$ip"
}
cmd_credentials() {
IP="$(node_ip node1)"
on 'awk "/^\[spinifex\]/{f=1} /^\[/&&!/^\[spinifex\]/{f=0} f" ~/.aws/credentials'
}
cmd_ssh() {
IP="$(node_ip "${NODE_ARG:-node1}")"
# shellcheck disable=SC2029
ssh "${SSH_OPTS[@]}" "ubuntu@$IP" "$@"
}
case "$CMD" in up | destroy | resolve-uncertain) acquire_lock ;; esac
case "$CMD" in
up) cmd_up ;;
destroy) cmd_destroy ;;
status) cmd_status ;;
credentials) cmd_credentials ;;
resolve-uncertain) cmd_resolve_uncertain ;;
ssh) cmd_ssh "$@" ;;
*) usage >&2; exit 2 ;;
esac
+108
View File
@@ -0,0 +1,108 @@
#!/usr/bin/env bash
# host-data-disk.sh — format and mount the blank data disk at /var/lib/spinifex. Run as root ON the host.
#
# host-data-disk.sh [/dev/disk/by-id/nvme-eui.<id>]
#
# The disk is chosen by content, never by its nvmeXn1 name, which swaps between boots. A candidate is a
# whole-disk by-id EUI link whose device is not in the root filesystem's ancestry, has no partitions,
# holders, swap or mounts, no filesystem signature and no partition table. Exactly one candidate must
# remain, or the script stops and lists the disks; pass the by-id path to choose between several.
# It never reformats: a rerun on a host already done prints the UUID and exits. The last line of stdout
# is DATA_UUID=<uuid>, which the later install phases need.
set -euo pipefail
MP=/var/lib/spinifex
LABEL=spinifex-data
FSTAB_BAK=/etc/fstab.pre-spinifex
MIN_BYTES="${MIN_DATA_BYTES:-100000000000}"
WANT="${1:-}"
die() { echo "STOP: $*" >&2; exit 1; }
[ "$(id -u)" = 0 ] || die "run as root"
# Resume: the mount is already there and is ours.
if mp_uuid="$(findmnt --mountpoint "$MP" -no UUID 2>/dev/null)" && [ -n "$mp_uuid" ]; then
src="$(findmnt --mountpoint "$MP" -no SOURCE)"
[ "$(blkid -s LABEL -o value "$src")" = "$LABEL" ] || die "$MP is mounted from $src, which is not labelled $LABEL; not touching it"
grep -q "UUID=${mp_uuid}[[:space:]]" /etc/fstab || die "$MP is mounted but /etc/fstab has no UUID=$mp_uuid entry"
echo "already done: $MP is mounted from $src (UUID $mp_uuid)"
echo "DATA_UUID=$mp_uuid"
exit 0
fi
if grep -q "$MP" /etc/fstab; then die "/etc/fstab mentions $MP but nothing is mounted there; fix that by hand first"; fi
[ ! -e "$MP" ] || [ -z "$(ls -A "$MP")" ] || die "$MP exists and is not empty"
root_src="$(findmnt -no SOURCE /)" || die "cannot find the root source"
root_anc="$(lsblk -s -nro NAME "$root_src")" || die "cannot read the root ancestry"
[ -n "$root_anc" ] || die "empty root ancestry"
# blank_reason ID prints why the disk is not a candidate and returns 1, or returns 0.
blank_reason() {
local id="$1" dev name size kids hold sw mnt rc w
dev="$(readlink -e "$id")" || { echo "link does not resolve"; return 1; }
[ -b "$dev" ] || { echo "not a block device"; return 1; }
name="$(basename "$dev")"
if grep -qx "$name" <<<"$root_anc"; then echo "in the root ancestry"; return 1; fi
size="$(blockdev --getsize64 "$id")" || { echo "size unreadable"; return 1; }
[ "$size" -ge "$MIN_BYTES" ] || { echo "only $size bytes"; return 1; }
kids="$(lsblk -nro NAME "$id")" || { echo "children unreadable"; return 1; }
[ "$kids" = "$name" ] || { echo "has children"; return 1; }
hold="$(ls -A "/sys/block/$name/holders/")" || { echo "holders unreadable"; return 1; }
[ -z "$hold" ] || { echo "has holders"; return 1; }
sw="$(swapon --show=NAME --noheadings)" || { echo "swapon failed"; return 1; }
if grep -q "$name" <<<"$sw"; then echo "is swap"; return 1; fi
mnt="$(lsblk -nro MOUNTPOINTS "$id")" || { echo "mounts unreadable"; return 1; }
[ -z "${mnt//[[:space:]]/}" ] || { echo "is mounted"; return 1; }
set +e; blkid -p "$id" >/dev/null 2>&1; rc=$?; set -e
[ "$rc" -eq 2 ] || { echo "blkid -p says it has a signature (rc $rc)"; return 1; }
w="$(wipefs -n "$id")" || { echo "wipefs failed"; return 1; }
[ -z "$w" ] || { echo "wipefs finds signatures"; return 1; }
return 0
}
candidates=()
shopt -s nullglob
for id in /dev/disk/by-id/nvme-eui.*; do
case "$id" in *-part*) continue ;; esac
if [ -n "$WANT" ] && [ "$id" != "$WANT" ]; then continue; fi
if why="$(blank_reason "$id")"; then
candidates+=("$id")
else
echo "skip $id: $why"
fi
done
shopt -u nullglob
if [ "${#candidates[@]}" -ne 1 ]; then
lsblk -o NAME,SIZE,TYPE,MODEL,SERIAL,FSTYPE,MOUNTPOINTS >&2 || true
die "expected exactly one blank data disk among the by-id EUI disks, found ${#candidates[@]}: ${candidates[*]:-none}"
fi
ID="${candidates[0]}"
DEV="$(readlink -e "$ID")"
echo "CHECKS PASSED: $ID -> $DEV"
mkdir -p "$MP"
mkfs.ext4 -q -L "$LABEL" "$ID"
U="$(blkid -p -o value -s UUID "$ID")" || die "blkid UUID failed"
[ -n "$U" ] || die "no UUID after mkfs"
[ "$(blkid -p -o value -s TYPE "$ID")" = ext4 ] || die "not ext4 after mkfs"
NEW="$(mktemp /etc/fstab.spinifex.XXXXXX)"
cp /etc/fstab "$NEW"
echo "UUID=$U $MP ext4 defaults,nofail,x-systemd.device-timeout=30s 0 2" >> "$NEW"
findmnt --verify --tab-file "$NEW" || { rm -f "$NEW"; die "staged fstab failed verify, /etc/fstab untouched"; }
BAK="$FSTAB_BAK.$(date +%s)"
cp -a /etc/fstab "$BAK"
trap 'cp -a "$BAK" /etc/fstab; systemctl daemon-reload; echo "ROLLED BACK /etc/fstab from $BAK" >&2' EXIT
chmod 644 "$NEW"
mv "$NEW" /etc/fstab
systemctl daemon-reload
mount "$MP"
[ "$(findmnt --mountpoint "$MP" -no UUID)" = "$U" ] || die "wrong UUID at $MP"
[ "$(findmnt --mountpoint "$MP" -no FSTYPE)" = ext4 ] || die "wrong fstype at $MP"
[ "$(findmnt --mountpoint "$MP" -no MAJ:MIN | tr -d ' ')" = "$(lsblk -dno MAJ:MIN "$ID" | tr -d ' ')" ] || die "wrong device at $MP"
trap - EXIT
findmnt --mountpoint "$MP" -o TARGET,SOURCE,MAJ:MIN,UUID,FSTYPE,OPTIONS
echo "STORAGE OK"
echo "DATA_UUID=$U"
File diff suppressed because it is too large Load Diff
+25
View File
@@ -0,0 +1,25 @@
output "deployment_name" {
value = var.deployment_name
}
output "site" {
value = var.site
}
output "project" {
value = local.project
}
# provider_host_id is the server ID. public_ip is the server's own /31 address, which SSH and the
# Spinifex API use; the Elastic IP is separate.
output "nodes" {
value = [for n in latitudesh_server.node : {
name = trimprefix(n.hostname, "spx-${var.deployment_name}-")
provider_host_id = n.id
public_ip = n.primary_ipv4
}]
}
output "elastic_ip" {
value = latitudesh_elastic_ip.node1.address
}
@@ -0,0 +1,4 @@
# The token comes from LATITUDESH_AUTH_TOKEN in the environment, which the provider reads
# itself (its auth_token argument documents that variable). It is never a variable here, so it
# is not written to the state or to a tfvars file. The default project is set per resource.
provider "latitudesh" {}
@@ -0,0 +1,5 @@
deployment_name = "l1"
site = "DAL"
plan = "m4-metal-medium"
node_count = 1
ssh_public_key_path = "~/.ssh/id_ed25519.pub"
@@ -0,0 +1,48 @@
variable "deployment_name" {
type = string
description = "Names every resource and the server hostnames, so two deployments in one account never collide."
validation {
# A hostname is at most 32 characters: "spx-" + name + "-node1".
condition = can(regex("^[a-z0-9][a-z0-9-]{0,15}$", var.deployment_name))
error_message = "deployment_name must be 1-16 lowercase letters, digits or hyphens."
}
}
variable "project" {
type = string
default = ""
description = "ID or slug of an existing latitude.sh project to deploy into. Empty creates a project spinifex-<deployment_name> and removes it on destroy."
}
variable "site" {
type = string
default = "DAL"
description = "latitude.sh site slug. Pick one with stock for the plan and Elastic IP support; not SAO."
}
variable "plan" {
type = string
default = "m4-metal-medium"
description = "Server plan slug (the pricing page calls it m4.metal.medium)."
}
variable "operating_system" {
type = string
default = "ubuntu_26_04_x64_lts"
description = "OS slug. UNVERIFIED: the 26.04 slug follows the 24.04 pattern (ubuntu_24_04_x64_lts) and is checked by a data source at plan time."
}
variable "node_count" {
type = number
default = 1
validation {
# Three nodes need the VLAN and the Elastic IP move, which are not built yet.
condition = var.node_count == 1
error_message = "node_count must be 1."
}
}
variable "ssh_public_key_path" {
type = string
default = "~/.ssh/id_ed25519.pub"
}
@@ -0,0 +1,7 @@
terraform {
required_version = ">= 1.5.0"
required_providers {
# Pinned exactly: 4.8.0 is the version whose schema these files were written against.
latitudesh = { source = "latitudesh/latitudesh", version = "= 4.8.0" }
}
}
+33 -3
View File
@@ -3,11 +3,14 @@ package awsapi
import (
"context"
"errors"
"io"
"log/slog"
"maps"
"net/http"
"slices"
"strings"
"github.com/mulgadc/bluebottle/pkg/sigv4"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
"github.com/nats-io/nats.go"
@@ -63,7 +66,7 @@ var Actions = map[string]Handler{
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: dispatch.ResolveTargetAction,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeJSON,
Inventory: OperationInventory(),
@@ -78,11 +81,25 @@ func OperationInventory() dispatch.Inventory {
}
}
// resolveAction is the one X-Amz-Target parser telemetry and dispatch share.
func resolveAction(r *http.Request) string {
return actionFromTarget(r.Header.Get("X-Amz-Target"))
}
// actionFromTarget extracts the action suffix from an X-Amz-Target header.
// Any "<Prefix>.<Action>" or bare "<Action>" form is accepted.
func actionFromTarget(target string) string {
if _, after, ok := strings.CutLast(target, "."); ok {
return after
}
return target
}
// dispatch resolves, scopes and authorizes the request from one read of its
// body, then runs the action. Errors are awserrors codes the gateway renders.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
action := dispatch.ResolveTargetAction(r)
action := resolveAction(r)
if action == "" {
return errors.New(awserrors.ErrorMissingAction)
}
@@ -101,7 +118,7 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
return errors.New(awserrors.ErrorInternalError)
}
body, err := dispatch.ReadBoundedBody(r)
body, err := readBoundedBody(r)
if err != nil {
slog.ErrorContext(r.Context(), "ACM dispatch: failed to read body", "err", err)
return err
@@ -127,3 +144,16 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
WriteJSONResponse(w, output)
return nil
}
// readBoundedBody reads the body under the SigV4 payload cap, so a body read
// ahead of the policy gate can neither bypass it nor exhaust memory.
func readBoundedBody(r *http.Request) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return body, nil
}
@@ -58,6 +58,18 @@ func dispatchWith(t *testing.T, deps Deps, inv dispatch.Invocation) (*httptest.R
return w, NewRegistration(deps).Dispatch(w, inv)
}
func TestActionFromTarget(t *testing.T) {
for target, want := range map[string]string{
"CertificateManager.ListCertificates": "ListCertificates",
"a.b.ListCertificates": "ListCertificates",
"ListCertificates": "ListCertificates",
"CertificateManager.": "",
"": "",
} {
assert.Equal(t, want, actionFromTarget(target), "target %q", target)
}
}
// TestActionsScopeTableIsExhaustive is what stops the next ACM action being
// added with a silent account-wide grant. It asserts both directions, so a
// scope left behind by a deleted or renamed action fails too.
+4 -11
View File
@@ -49,13 +49,11 @@ type PlacementGroupOwner interface {
CleanupAccount(ctx context.Context, accountID string, referenced func(groupID, groupName string) string) (*ec2placementgroup.AccountCleanup, error)
}
// LaunchTemplateOwner is the launch-template owner's account scan, residue
// cleanup and live-template delete; a run deletes live templates only when its
// request asks and the consumer check completed.
// LaunchTemplateOwner is the launch-template owner's account scan and residue
// cleanup; live templates are never deleted by account cleanup.
type LaunchTemplateOwner interface {
ScanAccount(ctx context.Context, accountID string) (*ec2launchtemplate.AccountScan, error)
CleanupAccountResidue(ctx context.Context, accountID string, settledBefore time.Time) (*ec2launchtemplate.AccountCleanup, error)
DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*ec2launchtemplate.AccountTemplateDeletes, error)
}
// TagIndexOwner lists and clears the account's tag-index entries.
@@ -155,11 +153,6 @@ func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, e
var keyRun *keyPairRun
var groupRun *placementGroupRun
var templateRun *launchTemplateRun
// Live templates go first when asked: a version that names a key pair or
// group blocks it, so the reference read must see what remains.
if req.DeleteLiveTemplates && slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore, true)
}
if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) || slices.Contains(kinds, ec2v1.PlatformKindPlacementGroup) {
refs := c.readReferences(ctx, acct)
if slices.Contains(kinds, ec2v1.PlatformKindKeyPair) {
@@ -169,8 +162,8 @@ func (c *Cleanup) CleanupPlatform(ctx context.Context, req *Request) (*Result, e
groupRun = c.cleanPlacementGroups(ctx, acct, refs)
}
}
if templateRun == nil && slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore, false)
if slices.Contains(kinds, ec2v1.PlatformKindLaunchTemplate) {
templateRun = c.cleanLaunchTemplates(ctx, acct, settledBefore)
}
// Tags are listed before the owner re-scans, so a resource created in
@@ -1,159 +0,0 @@
package accountcleanup
//test:in-package — shares the owner fixtures and the injected clock.
import (
"context"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
ec2v1 "github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// fakeConsumers answers what nodegroups name; a non-nil err is returned with
// whatever partial answer it also holds.
type fakeConsumers struct {
ids, names []string
err error
}
func (f *fakeConsumers) AccountTemplateConsumers(context.Context, string) (*TemplateConsumers, error) {
out := &TemplateConsumers{TemplateIDs: map[string]bool{}, TemplateNames: map[string]bool{}}
markAll(out.TemplateIDs, f.ids)
markAll(out.TemplateNames, f.names)
return out, f.err
}
var _ TemplateConsumerReader = (*fakeConsumers)(nil)
// liveDelete runs a cleanup that deletes live templates, with consumers as the
// nodegroup reader (nil means none supplied).
func (o *owners) liveDelete(t *testing.T, consumers TemplateConsumerReader) *Result {
t.Helper()
readers := o.readers()
readers.Consumers = consumers
c := New(o.keys, o.groups, o.templates, o.tags, readers)
c.now = func() time.Time { return time.Now().Add(time.Hour) }
res, err := c.CleanupPlatform(t.Context(), &Request{AccountID: testAccount, CorrelationRef: "ref-1", DeleteLiveTemplates: true})
require.NoError(t, err)
return res
}
func (o *owners) templateNamingKeyAndGroup(t *testing.T, name, key, group string) string {
t.Helper()
ltID := o.createTemplate(t, testAccount, name)
_, err := o.templates.CreateLaunchTemplateVersion(t.Context(), &ec2.CreateLaunchTemplateVersionInput{
LaunchTemplateId: aws.String(ltID),
LaunchTemplateData: &ec2.RequestLaunchTemplateData{
KeyName: aws.String(key),
Placement: &ec2.LaunchTemplatePlacementRequest{GroupName: aws.String(group)},
},
}, testAccount)
require.NoError(t, err)
return ltID
}
func (o *owners) liveTemplateIDs(t *testing.T) []string {
t.Helper()
scan, err := o.templates.ScanAccount(t.Context(), testAccount)
require.NoError(t, err)
return scan.LiveTemplateIDs
}
// A template no nodegroup names is deleted in the same run as the key pair and
// group its version names, which the reference check no longer sees.
func TestCleanupPlatformDeletesAnUnnamedTemplateAndWhatItNames(t *testing.T) {
o := newOwners(t)
key := o.importKey(t, testAccount, "tkey")
group := o.createGroup(t, testAccount, "tgroup")
lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "tgroup")
res := o.liveDelete(t, &fakeConsumers{})
assert.ElementsMatch(t, []Item{
item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionDeleted, ""),
item(key, ec2v1.PlatformKindKeyPair, DispositionDeleted, ""),
item(group, ec2v1.PlatformKindPlacementGroup, DispositionDeleted, ""),
}, res.Items)
assert.Empty(t, o.liveTemplateIDs(t))
}
// A template a nodegroup names, by ID or by name, is kept and blocked, and so
// are the key pair and group its versions name.
func TestCleanupPlatformKeepsATemplateANodegroupNames(t *testing.T) {
for name, consumers := range map[string]*fakeConsumers{
"by id": {},
"by name": {names: []string{"web"}},
} {
t.Run(name, func(t *testing.T) {
o := newOwners(t)
key := o.importKey(t, testAccount, "tkey")
group := o.createGroup(t, testAccount, "tgroup")
lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "tgroup")
other := o.createTemplate(t, testAccount, "free")
if name == "by id" {
consumers.ids = []string{lt}
}
res := o.liveDelete(t, consumers)
assert.ElementsMatch(t, []Item{
item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonLaunchTemplateReference),
item(other, ec2v1.PlatformKindLaunchTemplate, DispositionDeleted, ""),
item(key, ec2v1.PlatformKindKeyPair, DispositionBlocked, ReasonLaunchTemplateReference),
item(group, ec2v1.PlatformKindPlacementGroup, DispositionBlocked, ReasonLaunchTemplateReference),
}, res.Items)
assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
})
}
}
// A consumer read that fails, returns a partial answer with an error, or is not
// supplied deletes no template, key pair or group, and the kind is incomplete.
func TestCleanupPlatformUnreadableConsumersDeleteNoTemplate(t *testing.T) {
cases := map[string]TemplateConsumerReader{
"error": &fakeConsumers{err: errInjected},
"partial": &fakeConsumers{ids: []string{"lt-0other"}, err: errInjected},
"not supplied": nil,
}
for name, consumers := range cases {
t.Run(name, func(t *testing.T) {
o := newOwners(t)
key := o.importKey(t, testAccount, "tkey")
lt := o.templateNamingKeyAndGroup(t, "web", "tkey", "none")
res := o.liveDelete(t, consumers)
assert.Contains(t, res.Items, item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonReferenceCheckFailed))
assert.Contains(t, res.Items, item(key, ec2v1.PlatformKindKeyPair, DispositionBlocked, ReasonLaunchTemplateReference))
assert.Equal(t, ec2v1.PlatformStatusScanIncomplete, statuses(res)[ec2v1.PlatformKindLaunchTemplate])
assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
})
}
}
// A template only a DELETING nodegroup still names is as blocked as any other:
// the reader reports every status, so the cleanup never sees a status to trust.
func TestCleanupPlatformKeepsATemplateADeletingNodegroupStillNames(t *testing.T) {
o := newOwners(t)
lt := o.createTemplate(t, testAccount, "draining")
res := o.liveDelete(t, &fakeConsumers{ids: []string{lt}})
assert.Equal(t, []Item{item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonLaunchTemplateReference)}, res.Items)
assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
}
// Without the request flag a supplied consumer reader changes nothing: live
// templates are never deleted, as the exercise relies on.
func TestCleanupPlatformLiveTemplatesSurviveWithoutTheRequestFlag(t *testing.T) {
o := newOwners(t)
lt := o.createTemplate(t, testAccount, "web")
readers := o.readers()
readers.Consumers = &fakeConsumers{}
c := New(o.keys, o.groups, o.templates, o.tags, readers)
res := run(t, c)
assert.Equal(t, []Item{item(lt, ec2v1.PlatformKindLaunchTemplate, DispositionBlocked, ReasonReferenceUnknown)}, res.Items)
assert.Equal(t, []string{lt}, o.liveTemplateIDs(t))
}
+5 -52
View File
@@ -204,18 +204,13 @@ type launchTemplateRun struct {
before *ec2launchtemplate.AccountScan
cleaned *ec2launchtemplate.AccountCleanup
did attempts
// deleteLive is true when this run tried to delete live templates.
deleteLive bool
}
func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settledBefore time.Time, deleteLive bool) *launchTemplateRun {
run := &launchTemplateRun{kind: c.startKind(ec2v1.PlatformKindLaunchTemplate), did: newAttempts(), deleteLive: deleteLive}
func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settledBefore time.Time) *launchTemplateRun {
run := &launchTemplateRun{kind: c.startKind(ec2v1.PlatformKindLaunchTemplate), did: newAttempts()}
if before, err := c.templates.ScanAccount(ctx, acct); err == nil {
run.before = before
}
if deleteLive {
c.deleteLiveTemplates(ctx, acct, run)
}
cleaned, err := c.templates.CleanupAccountResidue(ctx, acct, settledBefore)
if err != nil {
slog.WarnContext(ctx, "Platform cleanup: launch-template residue cleanup failed", "err", err)
@@ -223,50 +218,10 @@ func (c *Cleanup) cleanLaunchTemplates(ctx context.Context, acct string, settled
return run
}
run.cleaned = cleaned
for _, id := range cleaned.KeptTemplateIDs {
if run.did.kept[id] == "" {
run.did.kept[id] = ReasonReferenceUnknown
}
}
keepAll(run.did.kept, cleaned.KeptTemplateIDs, ReasonReferenceUnknown)
return run
}
// deleteLiveTemplates deletes the live templates no consumer names. Without a
// complete consumer answer nothing is deleted and every live template is kept
// as unchecked, so an unreadable consumer never reads as "no references".
func (c *Cleanup) deleteLiveTemplates(ctx context.Context, acct string, run *launchTemplateRun) {
var consumers *TemplateConsumers
if c.refs.Consumers != nil {
var err error
consumers, err = c.refs.Consumers.AccountTemplateConsumers(ctx, acct)
if err != nil {
slog.WarnContext(ctx, "Platform cleanup: template consumer check failed", "err", err)
consumers = nil
}
}
if consumers == nil {
if run.before != nil {
keepAll(run.did.kept, run.before.LiveTemplateIDs, ReasonReferenceCheckFailed)
}
run.kind.incomplete(ReasonReferenceCheckFailed)
return
}
deletes, err := c.templates.DeleteLiveTemplates(ctx, acct, func(id, name string) string {
if consumers.names(id, name) {
return ReasonLaunchTemplateReference
}
return ""
})
if err != nil {
slog.WarnContext(ctx, "Platform cleanup: live template delete failed", "err", err)
run.kind.incomplete(ec2v1.PlatformReasonStoreListFailed)
return
}
markAll(run.did.deleted, deletes.DeletedTemplateIDs)
markAll(run.did.failed, deletes.FailedTemplateIDs)
maps.Copy(run.did.kept, deletes.Referenced)
}
func (c *Cleanup) finishLaunchTemplates(ctx context.Context, acct string, run *launchTemplateRun, res *Result) ownedIDs {
defer c.finishKind(run.kind)
after, err := c.templates.ScanAccount(ctx, acct)
@@ -295,10 +250,8 @@ func (c *Cleanup) finishLaunchTemplates(ctx context.Context, acct string, run *l
}
// Live templates are never deleted here, so every one still present is
// blocked on the unknown EKS nodegroup reference, whenever it appeared.
if !run.deleteLive {
for _, id := range afterLive {
run.did.kept[id] = ReasonReferenceUnknown
}
for _, id := range afterLive {
run.did.kept[id] = ReasonReferenceUnknown
}
for _, id := range sortedUnion(beforeLive, kept, afterLive) {
disposition, reason := itemDisposition(id, run.did, present[id], after != nil)
@@ -35,38 +35,17 @@ type TemplateReferenceReader interface {
AccountTemplateReferences(ctx context.Context, accountID string) (*ec2launchtemplate.TemplateReferences, error)
}
// TemplateConsumers are the IDs and names of launch templates that records
// outside EC2, such as EKS nodegroups, still name. They select what to keep and
// are never reported.
type TemplateConsumers struct {
TemplateIDs map[string]bool
TemplateNames map[string]bool
}
// TemplateConsumerReader reads which launch templates records outside EC2 name,
// every status included, complete or an error. The consumer's owner answers it;
// EC2 never reads those records.
type TemplateConsumerReader interface {
AccountTemplateConsumers(ctx context.Context, accountID string) (*TemplateConsumers, error)
}
func (c *TemplateConsumers) names(id, name string) bool {
return c.TemplateIDs[id] || c.TemplateNames[name]
}
var (
_ SpotReferenceReader = (*ec2spotinstance.SpotInstanceServiceImpl)(nil)
_ TemplateReferenceReader = (*ec2launchtemplate.LaunchTemplateServiceImpl)(nil)
)
// ReferenceReaders are the checks run before key pairs and placement groups
// are deleted. A nil reader counts as a failed check. Consumers is read only
// by a request that deletes live templates, where nil is also a failed check.
// are deleted. A nil reader counts as a failed check.
type ReferenceReaders struct {
Instances InstanceReferenceReader
Spot SpotReferenceReader
Templates TemplateReferenceReader
Consumers TemplateConsumerReader
}
func newReferences() *References {
+1 -7
View File
@@ -10,11 +10,6 @@ type Request struct {
// Kinds limits the run to these platform kinds. Empty means all three.
Kinds []string `json:"kinds,omitempty"`
// DeleteLiveTemplates lets the run delete live launch templates that no
// nodegroup names, using the TemplateConsumerReader. Without it live
// templates are never deleted.
DeleteLiveTemplates bool `json:"delete_live_templates,omitempty"`
}
// Result is what the owner stores hold after one run. Kinds reuse the ec2.v1
@@ -93,8 +88,7 @@ const (
ReasonInstanceReference = "instance-reference"
// ReasonSpotRequestReference: an open or active spot request names it.
ReasonSpotRequestReference = "spot-request-reference"
// ReasonLaunchTemplateReference: a version of a live launch template names a
// key pair or group, or a nodegroup names a template.
// ReasonLaunchTemplateReference: a version of a live launch template names it.
ReasonLaunchTemplateReference = "launch-template-reference"
// ReasonReferenceUnknown: a record EC2 cannot read may name it, so it is
// never deleted by account cleanup.
@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"log/slog"
"time"
@@ -103,7 +102,7 @@ func startLiveInstance(ctx context.Context, natsConn *nats.Conn, instanceID, acc
if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
slog.ErrorContext(ctx, "StartInstances: owner returned error", "instance_id", instanceID, "code", *responseError.Code)
return nil, false, startRefusalError(responseError, msg.Header)
return nil, false, startRefusalError(responseError)
}
slog.InfoContext(ctx, "StartInstances: restarted via owner node", "instance_id", instanceID, "response", string(msg.Data))
@@ -136,25 +135,18 @@ func startStoppedInstance(ctx context.Context, natsConn *nats.Conn, instanceID,
if responseError, parseErr := awserrors.ValidateErrorPayload(msg.Data); parseErr != nil {
slog.ErrorContext(ctx, "StartInstances: Daemon returned error", "instance_id", instanceID, "code", *responseError.Code)
return nil, startRefusalError(responseError, msg.Header)
return nil, startRefusalError(responseError)
}
slog.InfoContext(ctx, "StartInstances: Command sent successfully", "instance_id", instanceID, "response", string(msg.Data))
return newStateChange(instanceID, 0, "pending", 80, "stopped"), nil
}
// startRefusalError rebuilds the daemon's error, keeping the message it sent,
// which names the node holding the volume, and wrapping ErrVolumeHeldElsewhere
// when the reply's header says that is why the start was refused.
func startRefusalError(responseError ec2.ResponseError, header nats.Header) error {
err := startRefusalCodeError(responseError)
if header.Get(natsmsg.VolumeHeldElsewhereHeader) == "true" {
return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere, err)
}
return err
}
func startRefusalCodeError(responseError ec2.ResponseError) error {
// startRefusalError rebuilds the daemon's error, keeping the message it sent.
// Reconstructing from the code alone renders the generic ErrorLookup text
// instead, which replaces the name of the node holding the volume with
// boilerplate about attaching volumes -- the one fact the operator needed.
func startRefusalError(responseError ec2.ResponseError) error {
var code, message string
if responseError.Code != nil {
code = *responseError.Code
@@ -3,14 +3,11 @@ package instance
import (
"context"
"encoding/json"
"errors"
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -164,33 +161,3 @@ func TestStartInstances_MixedSuccessAndFailure(t *testing.T) {
require.Error(t, err)
assert.Equal(t, "InvalidInstanceID.NotFound", err.Error())
}
// The AWS body says only IncorrectState, so the header is the one thing that
// tells a held volume apart from any other refusal in that code.
func TestStartInstances_VolumeHeldElsewhereHeader(t *testing.T) {
t.Parallel()
for name, held := range map[string]bool{"with header": true, "without header": false} {
t.Run(name, func(t *testing.T) {
t.Parallel()
_, nc := startTestNATSServer(t)
const id = "i-0abc0000000000002"
_, err := nc.Subscribe(ec2v1.InstanceCommandSubject(id), func(msg *nats.Msg) {
reply := nats.NewMsg(msg.Reply)
reply.Data = awserrors.GenerateErrorPayloadWithMessage(awserrors.ErrorIncorrectState, "volume vol-1 is published to node-a")
if held {
reply.Header.Set(natsmsg.VolumeHeldElsewhereHeader, "true")
}
_ = msg.RespondMsg(reply)
})
require.NoError(t, err)
_, err = StartInstances(t.Context(), &ec2.StartInstancesInput{InstanceIds: []*string{aws.String(id)}}, nc, "123456789012")
require.Error(t, err)
assert.Equal(t, held, errors.Is(err, ec2v1.ErrVolumeHeldElsewhere))
code, message, ok := awserrors.ResolveErrorDetail(err)
require.True(t, ok)
assert.Equal(t, awserrors.ErrorIncorrectState, code)
assert.Equal(t, "volume vol-1 is published to node-a", message)
})
}
}
+13 -1
View File
@@ -1,6 +1,8 @@
package awsapi
import (
"net/http"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
)
@@ -9,7 +11,7 @@ import (
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: dispatch.ResolveQueryAction,
ResolveAction: resolveAction,
Dispatch: NewHandler(deps).Dispatch,
Errors: dispatch.ErrorEnvelopeXML,
Traits: dispatch.ServiceTraits{XMLErrorShape: dispatch.XMLErrorShapeEC2},
@@ -22,3 +24,13 @@ func NewRegistration(deps Deps) dispatch.Registration {
func OperationInventory() dispatch.Inventory {
return dispatch.Inventory{Registered: ActionNames()}
}
// resolveAction is the one Query Action parser telemetry and dispatch share.
// It leaves the body readable for the dispatcher and answers "" on any failure.
func resolveAction(r *http.Request) string {
args, err := readQueryArgs(r)
if err != nil {
return ""
}
return args["Action"]
}
@@ -1,6 +1,11 @@
//test:in-package — drives the unexported resolveAction with the registration.
package awsapi
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -22,3 +27,29 @@ func TestNewRegistration_DeclaresTheEC2Service(t *testing.T) {
b := dispatch.NewBuilder()
require.NoError(t, b.Register(reg))
}
func TestResolveAction(t *testing.T) {
post := func(body string) *http.Request {
r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
return r
}
assert.Equal(t, "DescribeVpcs", resolveAction(post("Action=DescribeVpcs&Version=2016-11-15")))
assert.Empty(t, resolveAction(post("Version=2016-11-15")))
// The body stays readable for the dispatcher after the action is resolved.
r := post("Action=DescribeVpcs")
require.Equal(t, "DescribeVpcs", resolveAction(r))
assert.Equal(t, "DescribeVpcs", resolveAction(r))
}
func TestResolveAction_UnreadableBody(t *testing.T) {
r := httptest.NewRequest(http.MethodPost, "/", errReader{})
r.Header.Set("Content-Type", "application/x-www-form-urlencoded")
assert.Empty(t, resolveAction(r))
}
type errReader struct{}
func (errReader) Read([]byte) (int, error) { return 0, http.ErrBodyReadAfterClose }
+18 -1
View File
@@ -1,7 +1,9 @@
package awsapi
import (
"bytes"
"errors"
"io"
"log/slog"
"maps"
"net/http"
@@ -10,6 +12,7 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/bluebottle/pkg/iampolicy"
"github.com/mulgadc/bluebottle/pkg/sigv4"
ec2instanceapi "github.com/mulgadc/spinifex/spinifex/domains/ec2/awsapi/instance"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -23,11 +26,25 @@ const (
ec2ServicePrincipal = "ec2.amazonaws.com"
)
// readQueryArgs parses the Query body and rewinds it, so the same bytes can be
// read again. Any read or parse failure is a malformed query string.
func readQueryArgs(r *http.Request) (map[string]string, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, err
}
r.Body = io.NopCloser(bytes.NewReader(body))
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return query.ParseArgs(string(body))
}
// Dispatch parses, authorizes and runs one EC2 request, then writes the XML
// response. A returned error is an awserrors code the gateway renders.
func (h *Handler) Dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
queryArgs, err := dispatch.ReadArgs(r)
queryArgs, err := readQueryArgs(r)
if err != nil {
slog.Debug("EC2: malformed query string", "err", err)
return errors.New(awserrors.ErrorMalformedQueryString)
+140
View File
@@ -0,0 +1,140 @@
package image
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const drainImageInstanceID = "i-image-drain-host"
// setupDrainImageService mirrors domains/ec2/snapshot's setupDrainService: a
// reachable (but rejecting) Predastore host so a drained volume falls through
// into the real snapshot path and fails fast, a data dir with no local drain
// socket, and a NATS connection to route the drain to the hosting node.
func setupDrainImageService(t *testing.T) (*ImageServiceImpl, *objectstore.MemoryObjectStore, *nats.Conn) {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
t.Cleanup(srv.Close)
_, nc := testutil.StartTestNATS(t)
store := objectstore.NewMemoryObjectStore()
cfg := &config.Config{
DataDir: testutil.SocketTempDir(t),
Predastore: config.PredastoreConfig{
Host: srv.URL,
Bucket: testBucket,
},
}
svc := NewImageServiceImplWithConfig(cfg, store, nc)
return svc, store, nc
}
func seedImageVolumeAttachment(t *testing.T, store *objectstore.MemoryObjectStore, volumeID, state, instanceID string) {
t.Helper()
require.NoError(t, volumestate.Write(context.Background(), store, testBucket, volumeID, volumestate.Record{
State: state,
AttachedInstance: instanceID,
}))
}
func drainImageResponder(t *testing.T, nc *nats.Conn, instanceID string, reply []byte) chan *nats.Msg {
t.Helper()
got := make(chan *nats.Msg, 4)
sub, err := nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
got <- msg
_ = msg.Respond(reply)
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
return got
}
func drainedImageAck(t *testing.T, volumeID string) []byte {
t.Helper()
data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
require.NoError(t, err)
return data
}
func awaitImageDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
t.Helper()
select {
case msg := <-got:
var command ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(msg.Data, &command))
return command
case <-time.After(2 * time.Second):
t.Fatal("the node hosting the volume never received a drain command")
return ec2v1.EC2InstanceCommand{}
}
}
// CreateImageFromInstance's IsRunning path must drain the volume's hosting node
// before it reads the live checkpoint — the same guarantee ec2.CreateSnapshot
// has. Without this, a checkpoint predating a guest-triggered GPT rewrite
// (e.g. growpart on first boot) can be captured silently.
func TestSnapshotRunningVolume_DrainsAttachedVolume(t *testing.T) {
svc, store, nc := setupDrainImageService(t)
createTestVolumeConfig(t, store, "vol-image-drain", 10)
seedImageVolumeAttachment(t, store, "vol-image-drain", "in-use", drainImageInstanceID)
got := drainImageResponder(t, nc, drainImageInstanceID, drainedImageAck(t, "vol-image-drain"))
err := svc.snapshotRunningVolume("vol-image-drain", "snap-image-drain", testAccountID)
// The rejecting Predastore stub fails the snapshot itself; what matters
// here is that the drain reached the hosting node first.
require.Error(t, err)
command := awaitImageDrainCommand(t, got)
assert.True(t, command.Attributes.DrainVolume)
assert.Equal(t, drainImageInstanceID, command.ID)
require.NotNil(t, command.DrainVolumeData)
assert.Equal(t, "vol-image-drain", command.DrainVolumeData.VolumeID)
}
// An attached volume whose host reports the drain failed must fail
// CreateImage rather than register an AMI built from a stale checkpoint.
func TestSnapshotRunningVolume_UndrainableAttachedVolumeFails(t *testing.T) {
svc, store, nc := setupDrainImageService(t)
createTestVolumeConfig(t, store, "vol-image-nodrain", 10)
seedImageVolumeAttachment(t, store, "vol-image-nodrain", "in-use", drainImageInstanceID)
drainImageResponder(t, nc, drainImageInstanceID,
[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
err := svc.snapshotRunningVolume("vol-image-nodrain", "snap-image-nodrain", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
// An unattached (stopped-instance) volume takes the no-drain path: the
// checkpoint Close() left behind is already current.
func TestSnapshotRunningVolume_AvailableSkipsDrain(t *testing.T) {
svc, store, _ := setupDrainImageService(t)
createTestVolumeConfig(t, store, "vol-image-available", 10)
seedImageVolumeAttachment(t, store, "vol-image-available", "available", "")
svc.natsConn = nil
// No responder exists and the socket is never dialled: proceeding straight
// to the (rejecting) snapshot path proves the drain was skipped, not stuck.
err := svc.snapshotRunningVolume("vol-image-available", "snap-image-available", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
@@ -2,20 +2,16 @@ package image
import (
"context"
"sync/atomic"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -100,15 +96,15 @@ func TestGetVolumeMetadata_Provider_UnknownVolumeIsNotFound(t *testing.T) {
assert.Equal(t, awserrors.ErrorInvalidVolumeNotFound, err.Error())
}
// TestSnapshotVolume_Provider_CreatesThroughProvider locks that a
// root-volume snapshot under a provider is created via provider.CreateSnapshot
// TestSnapshotStoppedVolume_Provider_CreatesThroughProvider locks that a
// stopped-volume snapshot under a provider is created via provider.CreateSnapshot
// and never falls through to building an embedded engine — the Predastore host
// is a closed port, so any fallback would fail with connection refused.
func TestSnapshotVolume_Provider_CreatesThroughProvider(t *testing.T) {
func TestSnapshotStoppedVolume_Provider_CreatesThroughProvider(t *testing.T) {
svc, _, provider := setupProviderSnapshotImageService(t)
doc := seedProviderVolume(t, svc, provider, "vol-stopped001", 8)
err := svc.snapshotVolume(doc.VolumeID, "snap-stopped001", testAccountID)
err := svc.snapshotStoppedVolume(doc.VolumeID, "snap-stopped001", testAccountID)
require.NoError(t, err, "the provider path must not fall through to the embedded engine against the closed Predastore port")
// CreateSnapshot is idempotent on (SnapshotID, VolumeID): re-issuing it
@@ -122,22 +118,22 @@ func TestSnapshotVolume_Provider_CreatesThroughProvider(t *testing.T) {
assert.Equal(t, doc.VolumeID, snap.SourceVolumeID)
}
// TestSnapshotVolume_Provider_PassesProviderHandleAsVolumeHandle locks
// TestSnapshotRunningVolume_Provider_PassesProviderHandleAsVolumeHandle locks
// that the ebsmetadata document's ProviderHandle is threaded through as
// CreateSnapshotRequest.VolumeHandle. A wrong handle must be rejected by the
// provider (MemoryProvider treats an empty VolumeHandle as "skip the check",
// so only a mismatched non-empty value proves the real value is forwarded).
func TestSnapshotVolume_Provider_PassesProviderHandleAsVolumeHandle(t *testing.T) {
func TestSnapshotRunningVolume_Provider_PassesProviderHandleAsVolumeHandle(t *testing.T) {
svc, _, provider := setupProviderSnapshotImageService(t)
doc := seedProviderVolume(t, svc, provider, "vol-running001", 8)
err := svc.snapshotVolume(doc.VolumeID, "snap-running001", testAccountID)
err := svc.snapshotRunningVolume(doc.VolumeID, "snap-running001", testAccountID)
require.NoError(t, err, "the correct provider handle must let CreateSnapshot succeed")
doc.ProviderHandle = "wrong-handle"
require.NoError(t, svc.MetadataStore().PutVolume(context.Background(), doc))
err = svc.snapshotVolume(doc.VolumeID, "snap-running002", testAccountID)
err = svc.snapshotRunningVolume(doc.VolumeID, "snap-running002", testAccountID)
require.Error(t, err, "a mismatched provider handle must be rejected, proving the value came from the document")
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
@@ -226,98 +222,3 @@ func TestCreateImageFromInstance_SetsEnaSupportFromInstance(t *testing.T) {
require.NotNil(t, ami.EnaSupport)
assert.True(t, *ami.EnaSupport)
}
// createUnavailableProvider refuses every CreateSnapshot as retry-later.
type createUnavailableProvider struct {
ebsprovider.EBSProvider
}
func (p *createUnavailableProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
return nil, &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnavailable, Message: "held by node-2"}
}
// TestSnapshotVolume_Provider_UnavailableIsRetryLater locks that the
// CreateImage snapshot surfaces a provider's retry-later as ServiceUnavailable.
func TestSnapshotVolume_Provider_UnavailableIsRetryLater(t *testing.T) {
svc, _, provider := setupProviderSnapshotImageService(t)
doc := seedProviderVolume(t, svc, provider, "vol-stopped-busy", 8)
svc.SetEBSProvider(&createUnavailableProvider{EBSProvider: provider})
err := svc.snapshotVolume(doc.VolumeID, "snap-stopped-busy", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServiceUnavailable, err.Error())
}
// createInUseProvider refuses every CreateSnapshot as volume in use.
type createInUseProvider struct {
ebsprovider.EBSProvider
}
func (p *createInUseProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
return nil, &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: "published"}
}
// TestSnapshotVolume_Provider_VolumeInUseIsIncorrectState locks that the
// CreateImage snapshot tells the caller to detach a volume the provider refuses as in use.
func TestSnapshotVolume_Provider_VolumeInUseIsIncorrectState(t *testing.T) {
svc, _, provider := setupProviderSnapshotImageService(t)
doc := seedProviderVolume(t, svc, provider, "vol-attached-inuse", 8)
svc.SetEBSProvider(&createInUseProvider{EBSProvider: provider})
err := svc.snapshotVolume(doc.VolumeID, "snap-attached-inuse", testAccountID)
code, message, ok := awserrors.ResolveErrorDetail(err)
require.True(t, ok, "unregistered error: %v", err)
assert.Equal(t, awserrors.ErrorIncorrectState, code)
assert.Contains(t, message, doc.VolumeID)
assert.Contains(t, message, "Detach")
}
// countingSnapshotProvider counts the CreateSnapshot calls it forwards.
type countingSnapshotProvider struct {
ebsprovider.EBSProvider
creates atomic.Int32
}
func (p *countingSnapshotProvider) CreateSnapshot(ctx context.Context, req ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
p.creates.Add(1)
return p.EBSProvider.CreateSnapshot(ctx, req)
}
// An image of a running instance is snapshotted by the provider, which drains
// a buffered volume itself, so CreateImage neither dials the drain socket nor
// asks the instance's host to drain.
func TestCreateImageFromInstance_RunningInstanceLeavesTheDrainToTheProvider(t *testing.T) {
const instanceID = "i-running-drain"
svc, _, provider := setupProviderSnapshotImageService(t)
_, nc := testutil.StartTestNATS(t)
svc.natsConn = nc
svc.config.DataDir = testutil.SocketTempDir(t)
counting := &countingSnapshotProvider{EBSProvider: provider}
svc.SetEBSProvider(counting)
doc := seedProviderVolume(t, svc, provider, "vol-running-drain", 8)
doc.State, doc.AttachedInstance = "in-use", instanceID
require.NoError(t, svc.MetadataStore().PutVolume(context.Background(), doc))
drains := testutil.StartDrainSocket(t, svc.config.DataDir, doc.VolumeID, "OK\n")
var commands atomic.Int32
sub, err := nc.Subscribe(ec2v1.InstanceCommandSubject(instanceID), func(msg *nats.Msg) {
commands.Add(1)
_ = msg.Respond([]byte(`{"Code":"InternalError","Message":"no drain socket"}`))
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
out, err := svc.CreateImageFromInstance(CreateImageParams{
Input: &ec2.CreateImageInput{InstanceId: aws.String(instanceID), Name: aws.String("running-ami")},
RootVolumeID: doc.VolumeID,
IsRunning: true,
}, testAccountID)
require.NoError(t, err)
require.NotNil(t, out.ImageId)
assert.Equal(t, int32(1), counting.creates.Load(), "the provider must take the snapshot")
assert.Zero(t, drains(), "CreateImage must not dial the drain socket")
assert.Zero(t, commands.Load(), "CreateImage must not ask the instance's host to drain")
}
+52 -13
View File
@@ -38,7 +38,7 @@ type CreateImageParams struct {
Input *ec2.CreateImageInput
RootVolumeID string
SourceImageID string
IsRunning bool // logged only: the provider's CreateSnapshot drains a live volume itself
IsRunning bool // true = use live checkpoint (instance still running), false = use numbered checkpoint from Close
}
// ImageServiceImpl handles AMI image operations with S3 storage.
@@ -69,7 +69,9 @@ func (s *ImageServiceImpl) MetadataStore() *ebsmetadata.Store {
return s.metadata
}
// NewImageServiceImpl creates a new daemon-side image service.
// NewImageServiceImpl creates a new daemon-side image service. natsConn is used
// to drain a running instance's volume (routed to the node hosting it) before
// CreateImageFromInstance reads its live checkpoint.
func NewImageServiceImpl(cfg *config.Config, natsConn *nats.Conn) *ImageServiceImpl {
store := objectstore.NewS3ObjectStoreFromConfig(
cfg.Predastore.Host,
@@ -97,7 +99,8 @@ func NewImageServiceImplWithStore(store objectstore.ObjectStore, bucketName stri
}
// NewImageServiceImplWithConfig creates an image service with an explicit
// config, object store and NATS connection (for testing).
// config and NATS connection (for testing the running-volume drain path,
// which needs config.DataDir/Predastore and a NATS connection to route to).
func NewImageServiceImplWithConfig(cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn) *ImageServiceImpl {
return &ImageServiceImpl{
config: cfg,
@@ -444,9 +447,15 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams, acc
"rootVolumeId", params.RootVolumeID, "amiId", amiID, "snapshotId", snapshotID,
"isRunning", params.IsRunning)
// Step 1: Snapshot root volume
if err := s.snapshotVolume(params.RootVolumeID, snapshotID, accountID); err != nil {
return nil, err
// Step 1: Snapshot root volume (live via NATS or offline from S3)
var snapshotErr error
if params.IsRunning {
snapshotErr = s.snapshotRunningVolume(params.RootVolumeID, snapshotID, accountID)
} else {
snapshotErr = s.snapshotStoppedVolume(params.RootVolumeID, snapshotID, accountID)
}
if snapshotErr != nil {
return nil, snapshotErr
}
// Step 2: Read source volume config for size
@@ -512,21 +521,51 @@ func (s *ImageServiceImpl) CreateImageFromInstance(params CreateImageParams, acc
}, nil
}
// snapshotVolume snapshots the root volume through the EBS provider, whether
// or not the instance is running. A provider that buffers writes drains the
// volume inside its own CreateSnapshot, so no drain is issued here.
func (s *ImageServiceImpl) snapshotVolume(volumeID, snapshotID, accountID string) error {
// snapshotRunningVolume creates a snapshot of a running instance's volume by
// draining it to the node hosting it (so any writes buffered there reach S3),
// then reading the live checkpoint written by nbdkit on every NBD Flush. This
// shares domains/ec2/snapshot's DrainVolume with ec2.CreateSnapshot so the two
// live-snapshot paths cannot diverge on when a drain is required.
func (s *ImageServiceImpl) snapshotRunningVolume(volumeID, snapshotID, accountID string) error {
if s.config == nil {
return errors.New(awserrors.ErrorServerInternal)
}
volMeta, err := s.getVolumeMetadata(context.Background(), accountID, volumeID)
if err != nil {
slog.Error("snapshotVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
slog.Error("snapshotRunningVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotVolume")
// Flush the writes still buffered by whichever node serves the volume, so
// the snapshot below is current rather than an arbitrarily stale one
// predating e.g. a guest-triggered GPT rewrite (growpart on first boot). An
// attached volume that cannot be drained fails here rather than silently
// producing an image from a stale checkpoint.
if err := ec2snapshot.DrainVolume(context.Background(), s.config, s.store, s.natsConn, volumeID, volMeta.State, volMeta.AttachedInstance, accountID); err != nil {
slog.Error("snapshotRunningVolume: drain failed", "volumeId", volumeID, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotRunningVolume")
}
// snapshotStoppedVolume creates a snapshot of a detached volume. No drain: the
// volume is stopped/detached, so nothing is buffered on a serving node.
// viperblockd's handleCreateSnapshot prefers a live mounted engine over opening
// a second one, so it handles the mounted case itself.
func (s *ImageServiceImpl) snapshotStoppedVolume(volumeID, snapshotID, accountID string) error {
if s.config == nil {
return errors.New(awserrors.ErrorServerInternal)
}
volMeta, err := s.getVolumeMetadata(context.Background(), accountID, volumeID)
if err != nil {
slog.Error("snapshotStoppedVolume: failed to read volume metadata", "volumeId", volumeID, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
return s.createSnapshot(context.Background(), volMeta, snapshotID, "snapshotStoppedVolume")
}
// createSnapshot delegates snapshot creation for volMeta to the EBS provider.
@@ -541,7 +580,7 @@ func (s *ImageServiceImpl) createSnapshot(ctx context.Context, volMeta ebsmetada
})
if err != nil || created == nil {
slog.ErrorContext(ctx, "provider CreateSnapshot failed", "op", op, "volumeId", volMeta.VolumeID, "snapshotId", snapshotID, "err", err)
return ec2snapshot.ProviderCreateError(volMeta.VolumeID, err)
return errors.New(awserrors.ErrorServerInternal)
}
slog.InfoContext(ctx, "snapshot created", "op", op, "volumeId", volMeta.VolumeID, "snapshotId", snapshotID)
return nil
@@ -2551,19 +2551,56 @@ func TestResetImageAttribute_MissingParameters(t *testing.T) {
assert.Equal(t, awserrors.ErrorMissingParameter, err.Error())
}
func TestSnapshotVolume_VolumeConfigMissing(t *testing.T) {
func TestSnapshotRunningVolume_VolumeConfigMissing(t *testing.T) {
store := objectstore.NewMemoryObjectStore()
svc := &ImageServiceImpl{
config: &config.Config{},
store: store,
bucketName: testBucket,
}
err := svc.snapshotVolume("vol-missing", "snap-s1", testAccountID)
err := svc.snapshotRunningVolume("vol-missing", "snap-r1", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotVolume_BackendInitFails(t *testing.T) {
func TestSnapshotRunningVolume_BackendInitFails(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
defer srv.Close()
store := objectstore.NewMemoryObjectStore()
svc := &ImageServiceImpl{
config: &config.Config{
Predastore: config.PredastoreConfig{
Host: srv.URL,
Region: "us-east-1",
Bucket: testBucket,
},
},
store: store,
bucketName: testBucket,
}
createTestVolumeConfig(t, store, "vol-running1", 10)
err := svc.snapshotRunningVolume("vol-running1", "snap-r2", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotStoppedVolume_VolumeConfigMissing(t *testing.T) {
store := objectstore.NewMemoryObjectStore()
svc := &ImageServiceImpl{
config: &config.Config{},
store: store,
bucketName: testBucket,
}
err := svc.snapshotStoppedVolume("vol-missing", "snap-s1", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotStoppedVolume_BackendInitFails(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
@@ -2583,7 +2620,7 @@ func TestSnapshotVolume_BackendInitFails(t *testing.T) {
}
createTestVolumeConfig(t, store, "vol-stopped1", 10)
err := svc.snapshotVolume("vol-stopped1", "snap-s2", testAccountID)
err := svc.snapshotStoppedVolume("vol-stopped1", "snap-s2", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
@@ -2609,7 +2646,60 @@ func imageS3MockServer(t *testing.T) *httptest.Server {
}))
}
func TestSnapshotVolume_EncryptionKeyLoadError(t *testing.T) {
func TestSnapshotRunningVolume_EncryptionKeyLoadError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
store := objectstore.NewMemoryObjectStore()
svc := &ImageServiceImpl{
config: &config.Config{
Predastore: config.PredastoreConfig{
Host: srv.URL,
Region: "us-east-1",
Bucket: testBucket,
},
Viperblock: config.ViperblockConfig{
EncryptionKeyFile: "/nonexistent-image-run-key.bin",
},
},
store: store,
bucketName: testBucket,
}
createTestVolumeConfig(t, store, "vol-running-enckey", 10)
err := svc.snapshotRunningVolume("vol-running-enckey", "snap-run-enckey", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotRunningVolume_LoadStateFails(t *testing.T) {
srv := imageS3MockServer(t)
defer srv.Close()
store := objectstore.NewMemoryObjectStore()
svc := &ImageServiceImpl{
config: &config.Config{
Predastore: config.PredastoreConfig{
Host: srv.URL,
Region: "us-east-1",
Bucket: testBucket,
AccessKey: "test-key",
SecretKey: "test-secret",
},
},
store: store,
bucketName: testBucket,
}
createTestVolumeConfig(t, store, "vol-running-lsf", 10)
err := svc.snapshotRunningVolume("vol-running-lsf", "snap-run-lsf", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotStoppedVolume_EncryptionKeyLoadError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
}))
@@ -2632,12 +2722,12 @@ func TestSnapshotVolume_EncryptionKeyLoadError(t *testing.T) {
}
createTestVolumeConfig(t, store, "vol-stopped-enckey", 10)
err := svc.snapshotVolume("vol-stopped-enckey", "snap-stop-enckey", testAccountID)
err := svc.snapshotStoppedVolume("vol-stopped-enckey", "snap-stop-enckey", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
func TestSnapshotVolume_LoadStateFails(t *testing.T) {
func TestSnapshotStoppedVolume_LoadStateFails(t *testing.T) {
srv := imageS3MockServer(t)
defer srv.Close()
@@ -2657,7 +2747,7 @@ func TestSnapshotVolume_LoadStateFails(t *testing.T) {
}
createTestVolumeConfig(t, store, "vol-stopped-lsf", 10)
err := svc.snapshotVolume("vol-stopped-lsf", "snap-stop-lsf", testAccountID)
err := svc.snapshotStoppedVolume("vol-stopped-lsf", "snap-stop-lsf", testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
}
+32 -12
View File
@@ -1237,12 +1237,7 @@ func (s *InstanceServiceImpl) StartInstance(ctx context.Context, instance *vm.VM
"instanceId", command.ID, "err", err)
return errors.New(awserrors.ErrorInvalidInstanceIDNotFound)
default:
// A volume held elsewhere is the caller's 400, not a fault on this node.
if refusal := volumeHeldElsewhereError(err); refusal != nil {
slog.WarnContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
return refusal
}
slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "instanceId", command.ID, "err", err)
slog.ErrorContext(ctx, "StartInstance: vmMgr.Start failed", "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
}
@@ -2843,15 +2838,40 @@ func (s *InstanceServiceImpl) RecoverInstance(ctx context.Context, instance *vm.
return s.launchClaimedInstance(ctx, instance, undo)
}
// volumeHeldElsewhereError converts a mount the EBS provider refused with its
// held-elsewhere code into IncorrectState naming the holder, wrapping
// ec2v1.ErrVolumeHeldElsewhere, and returns nil for anything else.
// volumeExclusionMarkers are the refusals that mean another node is writing this
// volume right now, or is still live and holds its unsealed writes. A previous
// owner that has stopped heartbeating is opened over with a warning instead. The
// text arrives having crossed NATS from viperblockd.
var volumeExclusionMarkers = []string{
ec2v1.VolumeHeldElsewhereText,
}
// ErrVolumeHeldElsewhere reports a launch refused because another node still
// holds one of the instance's volume leases.
//
// Recovery treats it differently from every other failure — it clears on the
// lease's own clock rather than needing anything fixed — so it has to be
// recognisable with errors.Is rather than inferred from text that has already
// crossed NATS once.
var ErrVolumeHeldElsewhere = errors.New("a volume is leased by another node")
// volumeHeldElsewhereError converts a mount refused because another node holds
// the volume into a client-facing error carrying that reason. Returns nil for
// anything else. Without it the caller gets ServerInternal and is told to
// retry, which cannot succeed while the holder is down, and the text naming the
// holder never leaves the node's log.
func volumeHeldElsewhereError(err error) error {
if !errors.Is(err, ebsprovider.ErrVolumeHeldElsewhere) {
if err == nil {
return nil
}
return fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", err.Error()))
msg := err.Error()
for _, marker := range volumeExclusionMarkers {
if strings.Contains(msg, marker) {
return fmt.Errorf("%w: %w", ErrVolumeHeldElsewhere,
awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", msg))
}
}
return nil
}
// restoreClaimedStoppedInstance writes instance back to the shared
@@ -4,19 +4,50 @@ package instance
//would widen the handler API for a test rather than for a caller.
import (
"context"
"errors"
"fmt"
"testing"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/runtime/compute/vm"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// TestVolumeHeldElsewhere_NamesTheHolder covers both refusals that mean the
// data is on another node. An operator who gets ServerInternal is told to retry
// something that cannot succeed while that node is down, so the node name and
// the reason have to survive the trip out of the daemon.
func TestVolumeHeldElsewhere_NamesTheHolder(t *testing.T) {
cases := []struct {
name string
err error
}{
{
name: "live holder, refused by the lease",
err: errors.New("failed to mount volume: volume is leased by another owner: " +
"held by node-b since 2026-09-02T17:04:39Z"),
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := volumeHeldElsewhereError(tc.err)
require.Error(t, got, "a volume held elsewhere must not fall through to ServerInternal")
code := awserrors.ValidErrorCodeFromError(got)
assert.NotEqual(t, awserrors.ErrorServerInternal, code,
"ServerInternal reads as a transient fault and invites a retry that cannot succeed")
assert.Contains(t, got.Error(), "node-b", "the caller has to learn where the data is")
assert.ErrorIs(t, got, ErrVolumeHeldElsewhere,
"recovery retries this on a shorter clock than anything else, so it has to be matchable")
resolved, _, ok := awserrors.ResolveErrorDetail(got)
require.True(t, ok, "the sentinel must not hide the code from the layer that returns it")
assert.Equal(t, awserrors.ErrorIncorrectState, resolved)
})
}
}
// TestVolumeHeldElsewhere_IgnoresUnrelatedFailures guards the other direction:
// a genuine internal fault must keep reporting as one rather than being
// relabelled as somebody else's volume.
@@ -29,76 +60,4 @@ func TestVolumeHeldElsewhere_IgnoresUnrelatedFailures(t *testing.T) {
// than refused, so this text is not an exclusion and must not become one.
assert.NoError(t, volumeHeldElsewhereError(errors.New(
"took over from node-b, which held unsealed writes since 2026-09-02T17:04:39Z")))
// Only the provider's code classifies a refusal, so its wording alone does not.
assert.NoError(t, volumeHeldElsewhereError(errors.New(
"failed to mount volume: "+ec2v1.VolumeHeldElsewhereText+": held by node-b")))
}
// A provider is free to word its refusal however it likes; the held-elsewhere
// code is what says another node has the volume.
func TestVolumeHeldElsewhere_MatchesTheProviderCodeWhateverTheWording(t *testing.T) {
refusal := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
Code: ebsprovider.ErrorCodeVolumeHeldElsewhere,
Message: "volume vol-1 is published to node-a",
})
got := volumeHeldElsewhereError(refusal)
require.Error(t, got, "a held-elsewhere code must be classified even without viperblockd's text")
assert.ErrorIs(t, got, ec2v1.ErrVolumeHeldElsewhere)
assert.Contains(t, got.Error(), "node-a", "the caller has to learn where the data is")
code, _, ok := awserrors.ResolveErrorDetail(got)
require.True(t, ok)
assert.Equal(t, awserrors.ErrorIncorrectState, code)
}
// volume_in_use is a read-only mismatch, not another node's claim, so it must
// not be relabelled as a volume held elsewhere even though the sentinels nest.
func TestVolumeHeldElsewhere_VolumeInUseAloneIsNotHeldElsewhere(t *testing.T) {
mismatch := fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
Code: ebsprovider.ErrorCodeVolumeInUse,
Message: "volume vol-1 is exported read-write, not read-only",
})
assert.NoError(t, volumeHeldElsewhereError(mismatch))
}
// heldElsewhereMounter refuses every mount the way the contract mounter
// reports a provider's held-elsewhere code, in wording that is not viperblockd's.
type heldElsewhereMounter struct{ raceVolumeMounter }
func (heldElsewhereMounter) Mount(context.Context, *vm.VM) error {
return fmt.Errorf("publish volume vol-1: %w", &ebsprovider.ProviderError{
Code: ebsprovider.ErrorCodeVolumeHeldElsewhere,
Message: "volume vol-1 is published to node-a",
})
}
// Both start paths a caller can reach must hand back the classified refusal:
// the stopped-instance claim (ec2.start) and the live owner (ec2.cmd), which
// the EKS reconciler tries first.
func TestStartPaths_ReportAVolumeHeldElsewhereByCode(t *testing.T) {
t.Setenv("XDG_RUNTIME_DIR", t.TempDir())
types, _ := defaultPrepareInstanceTypes()
t.Run("claimed relaunch", func(t *testing.T) {
mgr := vm.NewManager()
mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
err := svc.launchClaimedInstance(t.Context(), &vm.VM{ID: "i-held-claim", AccountID: "acc", InstanceType: "t3.micro", Status: vm.StatePending}, func() {})
assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
})
t.Run("live owner", func(t *testing.T) {
const id = "i-held-live"
mgr := mgrWith(map[string]*vm.VM{id: {ID: id, Status: vm.StateStopped, InstanceType: "t3.micro"}})
mgr.SetDeps(vm.Deps{NodeID: "node-b", VolumeMounter: heldElsewhereMounter{}})
v, _ := mgr.Get(id)
svc := &InstanceServiceImpl{resourceMgr: &fakeResourceCapacityProvider{instanceTypes: types}, vmMgr: mgr}
err := svc.StartInstance(t.Context(), v, ec2v1.EC2InstanceCommand{ID: id})
assert.ErrorIs(t, err, ec2v1.ErrVolumeHeldElsewhere)
assert.Equal(t, awserrors.ErrorIncorrectState, awserrors.ValidErrorCodeFromError(err))
})
}
@@ -130,51 +130,3 @@ func tallyOrphan(outcome orphanOutcome, removed, failed, unsettled *int) {
case orphanGone, orphanOwned:
}
}
// AccountTemplateDeletes is what one DeleteLiveTemplates run did, by template
// ID. It is the owner's account of its attempts, not proof of absence.
type AccountTemplateDeletes struct {
DeletedTemplateIDs []string
FailedTemplateIDs []string
// Referenced are live templates left because referenced named them, with its reason.
Referenced map[string]string
}
// DeleteLiveTemplates deletes the account's live templates through the same
// owner delete as DeleteLaunchTemplate, except those referenced names. A header
// it cannot read or decode is left for the owner scan to report. A listing
// failure is returned and nothing is deleted.
func (s *LaunchTemplateServiceImpl) DeleteLiveTemplates(ctx context.Context, accountID string, referenced func(id, name string) string) (*AccountTemplateDeletes, error) {
keys, err := kvutil.Keys(ctx, s.kv)
if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, fmt.Errorf("list launch-template keys: %w", err)
}
prefix := accountID + "."
out := &AccountTemplateDeletes{Referenced: map[string]string{}}
for _, key := range keys {
if !isHeaderKey(key, prefix) {
continue
}
entry, err := s.kv.Get(ctx, key)
if err != nil {
continue
}
var header LaunchTemplateHeader
if json.Unmarshal(entry.Value(), &header) != nil || prefix+header.LaunchTemplateId != key {
continue
}
if reason := referenced(header.LaunchTemplateId, header.LaunchTemplateName); reason != "" {
out.Referenced[header.LaunchTemplateId] = reason
continue
}
if err := s.deleteTemplate(ctx, accountID, &header, entry); err != nil {
slog.WarnContext(ctx, "Launch-template account cleanup: live template delete failed", "launchTemplateId", header.LaunchTemplateId, "err", err)
out.FailedTemplateIDs = append(out.FailedTemplateIDs, header.LaunchTemplateId)
continue
}
out.DeletedTemplateIDs = append(out.DeletedTemplateIDs, header.LaunchTemplateId)
}
slog.InfoContext(ctx, "Launch-template account live delete completed", "accountID", accountID,
"deleted", len(out.DeletedTemplateIDs), "failed", len(out.FailedTemplateIDs), "referenced", len(out.Referenced))
return out, nil
}
+304 -55
View File
@@ -2,8 +2,12 @@ package snapshot
import (
"context"
"sync/atomic"
"encoding/json"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ec2"
@@ -11,6 +15,8 @@ import (
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/providers/ebs"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/nats-io/nats.go"
@@ -18,78 +24,321 @@ import (
"github.com/stretchr/testify/require"
)
const attachedInstanceID = "i-attached-host"
const drainInstanceID = "i-drain-host"
// countingSnapshotProvider counts the CreateSnapshot calls it forwards.
type countingSnapshotProvider struct {
ebsprovider.EBSProvider
creates atomic.Int32
}
func (p *countingSnapshotProvider) CreateSnapshot(ctx context.Context, req ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
p.creates.Add(1)
return p.EBSProvider.CreateSnapshot(ctx, req)
}
// setupAttachedSnapshot is a viperblockd node answering CreateSnapshot for a
// volume attached to a running instance, with a memory provider behind it.
func setupAttachedSnapshot(t *testing.T, volumeID string) (*SnapshotServiceImpl, *countingSnapshotProvider, *nats.Conn) {
// setupDrainService returns a snapshot service configured like a daemon that
// answers ec2.CreateSnapshot: a reachable Predastore host (so drains are not
// skipped as metadata-only), a data dir with no drain socket in it, and a NATS
// connection. The Predastore stub rejects everything, so a snapshot that gets
// past the drain fails fast rather than hanging on a dead address.
func setupDrainService(t *testing.T) (*SnapshotServiceImpl, *objectstore.MemoryObjectStore, *nats.Conn) {
t.Helper()
_, nc := testutil.StartTestNATS(t)
cfg := &config.Config{
DataDir: testutil.SocketTempDir(t),
EBS: config.EBSConfig{Provider: config.EBSProviderViperblockd},
Predastore: config.PredastoreConfig{Host: "127.0.0.1:1", Bucket: "test-bucket"},
}
svc := NewSnapshotServiceImplWithStore(cfg, objectstore.NewMemoryObjectStore(), nc)
provider := &countingSnapshotProvider{EBSProvider: ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})}
svc.SetEBSProvider(provider)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusForbidden)
}))
t.Cleanup(srv.Close)
_, nc := testutil.StartTestNATS(t)
store := objectstore.NewMemoryObjectStore()
cfg := &config.Config{
DataDir: testutil.SocketTempDir(t),
Predastore: config.PredastoreConfig{
Host: srv.URL,
Bucket: "test-bucket",
},
}
return NewSnapshotServiceImplWithStore(cfg, store, nc), store, nc
}
// seedVolumeAttachment writes the control-plane state.json that decides whether
// a snapshot must drain.
func seedVolumeAttachment(t *testing.T, store *objectstore.MemoryObjectStore, volumeID, state, instanceID string) {
t.Helper()
require.NoError(t, volumestate.Write(context.Background(), store, "test-bucket", volumeID, volumestate.Record{
State: state,
AttachedInstance: instanceID,
}))
}
// drainResponder answers the per-instance command subject the way the node
// hosting the instance does, and hands each request it received to the test.
func drainResponder(t *testing.T, nc *nats.Conn, instanceID string, reply []byte) chan *nats.Msg {
t.Helper()
got := make(chan *nats.Msg, 4)
sub, err := nc.Subscribe("ec2.cmd."+instanceID, func(msg *nats.Msg) {
got <- msg
_ = msg.Respond(reply)
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
return got
}
// drainedAck is the reply of a node whose drain reached S3.
func drainedAck(t *testing.T, volumeID string) []byte {
t.Helper()
data, err := json.Marshal(ec2v1.DrainVolumeResponse{VolumeID: volumeID, Status: ec2v1.DrainVolumeStatusDrained})
require.NoError(t, err)
return data
}
// requireReturnsWithin fails the test if fn has not returned successfully by
// limit, so a drain that blocks on a socket it should not have dialed reports
// as a failure rather than as a slow test.
func requireReturnsWithin(t *testing.T, limit time.Duration, fn func() error) {
t.Helper()
done := make(chan error, 1)
go func() { done <- fn() }()
select {
case err := <-done:
require.NoError(t, err)
case <-time.After(limit):
t.Fatalf("drain did not return within %s", limit)
}
}
// awaitDrainCommand returns the command the hosting node received, failing the
// test if none arrives.
func awaitDrainCommand(t *testing.T, got chan *nats.Msg) ec2v1.EC2InstanceCommand {
t.Helper()
select {
case msg := <-got:
var command ec2v1.EC2InstanceCommand
require.NoError(t, json.Unmarshal(msg.Data, &command))
assert.Equal(t, testAccountID, msg.Header.Get(natsmsg.AccountIDHeader))
return command
case <-time.After(2 * time.Second):
t.Fatal("the node hosting the volume never received a drain command")
return ec2v1.EC2InstanceCommand{}
}
}
// An attached volume is drained on the node hosting its instance, which is the
// only node with the socket — the node answering CreateSnapshot usually is not.
func TestDrainVolume_AttachedRoutesToHostingNode(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-attached", "in-use", drainInstanceID)
got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-attached"))
require.NoError(t, svc.drainVolume(context.Background(), "vol-attached", "", "", testAccountID))
command := awaitDrainCommand(t, got)
assert.True(t, command.Attributes.DrainVolume)
assert.Equal(t, drainInstanceID, command.ID)
require.NotNil(t, command.DrainVolumeData)
assert.Equal(t, "vol-attached", command.DrainVolumeData.VolumeID)
}
// No responder on the command subject means no node is running the instance,
// which is what a stopped instance looks like: stop tears down both the plugin
// and the subscription but deliberately leaves a boot volume attached. Failing
// here would make every stopped instance's root volume unsnapshottable.
func TestDrainVolume_AttachedToStoppedInstanceTakesStoppedPath(t *testing.T) {
svc, store, _ := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-no-host", "in-use", drainInstanceID)
require.NoError(t, svc.drainVolume(context.Background(), "vol-no-host", "", "", testAccountID))
}
// A host that still holds the instance but reports it not running has nothing
// to flush, so the sealed checkpoint stands. This is the host-drain stop, which
// unmounts the volumes but keeps the subscription.
func TestDrainVolume_NotRunningAckTakesStoppedPath(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-not-running", "in-use", drainInstanceID)
ack, err := json.Marshal(ec2v1.DrainVolumeResponse{
VolumeID: "vol-not-running", Status: ec2v1.DrainVolumeStatusNotRunning,
})
require.NoError(t, err)
got := drainResponder(t, nc, drainInstanceID, ack)
require.NoError(t, svc.drainVolume(context.Background(), "vol-not-running", "", "", testAccountID))
assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
}
// A hosting node that reports the drain failed (wedged plugin, socket gone
// mid-restart) fails the snapshot rather than falling through.
func TestDrainVolume_AttachedWithFailedAckFails(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-drain-err", "in-use", drainInstanceID)
drainResponder(t, nc, drainInstanceID,
[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
err := svc.drainVolume(context.Background(), "vol-drain-err", "", "", testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), "drain failed")
}
// An ack that is not the drained ack is not evidence the writes reached S3.
func TestDrainVolume_AttachedWithUnexpectedAckFails(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-odd-ack", "in-use", drainInstanceID)
drainResponder(t, nc, drainInstanceID, []byte(`{}`))
err := svc.drainVolume(context.Background(), "vol-odd-ack", "", "", testAccountID)
require.Error(t, err)
assert.Contains(t, err.Error(), "unexpected ack")
}
// Nothing writes to an unattached volume, so the checkpoint its Close() left
// behind is current and no drain is attempted — not even locally. Dialing the
// socket is not a probe: it makes the plugin run a full flush, so an available
// volume must be decided from the record without touching it.
func TestDrainVolume_AvailableDoesNotDialLocalSocket(t *testing.T) {
svc, store, _ := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-available", "available", "")
// A socket that never answers: a dial would park here until the read
// deadline, so returning promptly is what proves none was made.
testutil.StartBlockingDrainSocket(t, svc.config.DataDir, "vol-available", "OK\n")
// A nil connection proves no command was issued: routing one would error.
svc.natsConn = nil
requireReturnsWithin(t, 2*time.Second, func() error {
return svc.drainVolume(context.Background(), "vol-available", "", "", testAccountID)
})
}
// A volume recorded as in-use but with no instance (drifted state) has no node
// to address, so it takes the stopped path rather than failing every snapshot.
func TestDrainVolume_InUseWithoutInstanceTakesStoppedPath(t *testing.T) {
svc, store, _ := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-no-instance", "in-use", "")
svc.natsConn = nil
require.NoError(t, svc.drainVolume(context.Background(), "vol-no-instance", "", "", testAccountID))
}
// When the volume is served by this node the socket is local, so the drain
// short-circuits without a hop. This is every single-node deployment.
func TestDrainVolume_LocalSocketShortCircuits(t *testing.T) {
svc, store, _ := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-local", "in-use", drainInstanceID)
testutil.StartDrainSocket(t, svc.config.DataDir, "vol-local", "OK\n")
// A nil connection proves the local ack was enough: no responder exists for
// drainInstanceID, so a routed command would fail.
svc.natsConn = nil
require.NoError(t, svc.drainVolume(context.Background(), "vol-local", "", "", testAccountID))
}
// A socket that answers without acking means this node does serve the volume
// and the flush failed. Routing that onward would come back to this same node
// and re-run the drain that has already failed, so it fails here instead.
func TestDrainVolume_LocalSocketErrorAckFailsWithoutRouting(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-local-err", "in-use", drainInstanceID)
testutil.StartDrainSocket(t, svc.config.DataDir, "vol-local-err", "ERR\n")
got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-local-err"))
err := svc.drainVolume(context.Background(), "vol-local-err", "", "", testAccountID)
require.Error(t, err)
assert.Empty(t, got, "a local drain that failed must not be re-issued to the host node")
}
// state.json is the control-plane-owned attachment record; the copy inside
// config.json is rewritten by the live NBD plugin from its stale in-memory
// state, so a volume reading "available" there can still be attached.
func TestDrainVolume_StateJSONOverridesConfigJSON(t *testing.T) {
svc, store, nc := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-stale-config", "in-use", drainInstanceID)
got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-stale-config"))
// "available" mirrors the stale State a live NBD plugin leaves in
// config.json; drainVolume must not trust it (see the test's own doc above).
require.NoError(t, svc.drainVolume(context.Background(), "vol-stale-config", "available", "", testAccountID))
assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
}
// Volumes predating the state.json split have no state object, so the state
// embedded in config.json is all there is to decide from.
func TestDrainVolume_FallsBackToConfigJSONWhenNoStateObject(t *testing.T) {
svc, _, nc := setupDrainService(t)
got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-legacy"))
// The state/instance embedded in a legacy config.json, with no state.json to overlay it.
require.NoError(t, svc.drainVolume(context.Background(), "vol-legacy", "in-use", drainInstanceID, testAccountID))
assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
}
// Without Predastore the snapshot is metadata-only and never reads a live
// checkpoint, so there is nothing for a drain to make current.
func TestDrainVolume_MetadataOnlySkipsDrain(t *testing.T) {
svc, store, _ := setupDrainService(t)
seedVolumeAttachment(t, store, "vol-meta-only", "in-use", drainInstanceID)
svc.config.Predastore.Host = ""
svc.natsConn = nil
require.NoError(t, svc.drainVolume(context.Background(), "vol-meta-only", "", "", testAccountID))
}
// putProviderVolume seeds both the control-plane document CreateSnapshot's
// provider branch reads for attachment state, and the provider's own record,
// mirroring what CreateVolume's provider branch leaves behind.
func putProviderVolume(t *testing.T, svc *SnapshotServiceImpl, provider ebsprovider.EBSProvider, volumeID, state, instanceID string) {
t.Helper()
require.NoError(t, svc.metadata.PutVolume(context.Background(), ebsmetadata.Volume{
VolumeID: volumeID, TenantID: testAccountID, CapacityGiB: 4,
State: state, AttachedInstance: instanceID, AvailabilityZone: "us-east-1a",
ProviderHandle: "memory://volume/" + volumeID,
}))
_, err := provider.CreateVolume(context.Background(), ebsprovider.CreateVolumeRequest{
Versioned: ebsprovider.NewVersioned(), VolumeID: volumeID,
CapacityRange: ebsprovider.CapacityRange{RequiredBytes: 4 * 1024 * 1024 * 1024},
AvailabilityZone: "us-east-1a",
})
require.NoError(t, err)
require.NoError(t, svc.metadata.PutVolume(context.Background(), ebsmetadata.Volume{
VolumeID: volumeID, TenantID: testAccountID, CapacityGiB: 4,
State: "in-use", AttachedInstance: attachedInstanceID, AvailabilityZone: "us-east-1a",
}))
return svc, provider, nc
}
// The provider drains a buffered volume inside its own CreateSnapshot, so the
// domain must not dial the drain socket itself: that would drain twice.
func TestCreateSnapshot_AttachedVolumeOpensNoDrainSocket(t *testing.T) {
svc, provider, _ := setupAttachedSnapshot(t, "vol-attached-local")
drains := testutil.StartDrainSocket(t, svc.config.DataDir, "vol-attached-local", "OK\n")
// The provider branch used to skip the drain entirely, silently snapshotting a
// stale checkpoint of an attached, actively-written volume. It must drain the
// same way the legacy branch does.
func TestCreateSnapshot_Provider_DrainsAttachedVolume(t *testing.T) {
svc, _, nc := setupDrainService(t)
provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
svc.SetEBSProvider(provider)
putProviderVolume(t, svc, provider, "vol-provider-drain", "in-use", drainInstanceID)
got := drainResponder(t, nc, drainInstanceID, drainedAck(t, "vol-provider-drain"))
_, err := svc.CreateSnapshot(context.Background(),
&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-attached-local")}, testAccountID)
&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-provider-drain")}, testAccountID)
require.NoError(t, err)
assert.Zero(t, drains(), "the domain must leave the drain to the provider")
assert.Equal(t, int32(1), provider.creates.Load())
command := awaitDrainCommand(t, got)
assert.True(t, command.Attributes.DrainVolume)
assert.Equal(t, "vol-provider-drain", command.DrainVolumeData.VolumeID)
}
// With no drain socket on this node, an attached volume is snapshotted by the
// provider without asking the instance's host for a drain.
func TestCreateSnapshot_AttachedVolumeSendsItsHostNoDrain(t *testing.T) {
svc, provider, nc := setupAttachedSnapshot(t, "vol-attached-remote")
var commands atomic.Int32
sub, err := nc.Subscribe(ec2v1.InstanceCommandSubject(attachedInstanceID), func(msg *nats.Msg) {
commands.Add(1)
_ = msg.Respond([]byte(`{"Code":"InternalError","Message":"no drain socket"}`))
})
require.NoError(t, err)
t.Cleanup(func() { _ = sub.Unsubscribe() })
// A provider-branch snapshot whose drain fails must fail closed rather than
// return a snapshot of stale data: the provider must never be asked to
// snapshot, and no snapshot metadata may be left behind.
func TestCreateSnapshot_Provider_UndrainableAttachedVolumeFails(t *testing.T) {
svc, _, nc := setupDrainService(t)
provider := ebsprovider.NewMemoryProvider(ebsprovider.Capabilities{})
svc.SetEBSProvider(provider)
putProviderVolume(t, svc, provider, "vol-provider-nodrain", "in-use", drainInstanceID)
snap, err := svc.CreateSnapshot(context.Background(),
&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-attached-remote")}, testAccountID)
got := drainResponder(t, nc, drainInstanceID,
[]byte(`{"Code":"`+awserrors.ErrorServerInternal+`","Message":"drain failed"}`))
_, err := svc.CreateSnapshot(context.Background(),
&ec2.CreateSnapshotInput{VolumeId: aws.String("vol-provider-nodrain")}, testAccountID)
require.Error(t, err)
assert.Equal(t, awserrors.ErrorServerInternal, err.Error())
assert.True(t, awaitDrainCommand(t, got).Attributes.DrainVolume)
out, err := svc.DescribeSnapshots(context.Background(), &ec2.DescribeSnapshotsInput{}, testAccountID)
require.NoError(t, err)
assert.Equal(t, "vol-attached-remote", aws.StringValue(snap.VolumeId))
assert.Equal(t, int32(1), provider.creates.Load())
assert.Zero(t, commands.Load(), "the instance's host must not be asked to drain")
assert.Empty(t, out.Snapshots, "a failed drain must not leave a snapshot behind")
}
+185 -17
View File
@@ -5,7 +5,10 @@ import (
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/tagmirror"
"github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"log/slog"
"net"
"path/filepath"
"strconv"
"strings"
"sync"
@@ -15,8 +18,10 @@ import (
"github.com/aws/aws-sdk-go/service/ec2"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/bluebottle/pkg/safecast"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/ebs/metadata"
"github.com/mulgadc/spinifex/spinifex/domains/ec2/volumestate"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsfilters "github.com/mulgadc/spinifex/spinifex/foundation/aws/filters"
awsidentifiers "github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
@@ -204,12 +209,19 @@ func (s *SnapshotServiceImpl) CreateSnapshot(ctx context.Context, input *ec2.Cre
return nil, errors.New(awserrors.ErrorIncorrectState)
}
// Flush the writes buffered by whichever node serves the volume, so the
// provider reads a current checkpoint. An attached volume that cannot be
// drained fails here rather than silently snapshotting stale data.
if err := s.drainVolume(ctx, volumeID, volume.State, volume.AttachedInstance, accountID); err != nil {
slog.ErrorContext(ctx, "CreateSnapshot: drain failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
created, err := s.provider.CreateSnapshot(ctx, ebsprovider.CreateSnapshotRequest{
Versioned: ebsprovider.NewVersioned(), SnapshotID: snapshotID, VolumeID: volumeID, VolumeHandle: volume.ProviderHandle,
})
if err != nil || created == nil {
slog.ErrorContext(ctx, "CreateSnapshot: provider CreateSnapshot failed", "volumeId", volumeID, "snapshotId", snapshotID, "err", err)
return nil, ProviderCreateError(volumeID, err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
now := time.Now()
snapshotCfg := ebsmetadata.Snapshot{
@@ -247,6 +259,177 @@ func (s *SnapshotServiceImpl) CreateSnapshot(ctx context.Context, input *ec2.Cre
return snapshotConfigToEC2(snapshotCfg), nil
}
const (
// drainDialTimeout bounds the connect to the local drain socket. The socket
// is either being served on this node or it is not, so a slow connect means
// a wedged plugin rather than a busy one.
drainDialTimeout = time.Second
// drainAckTimeout bounds DrainToBackend on the hosting node: it flushes the
// WAL and the live checkpoint to S3, so it scales with the dirty set rather
// than with a fixed unit of work.
drainAckTimeout = 30 * time.Second
// drainRequestTimeout bounds the ec2.cmd round-trip. It exceeds
// drainAckTimeout so a slow drain surfaces as the hosting node's error
// instead of an opaque NATS timeout here.
drainRequestTimeout = 35 * time.Second
)
// ErrNoDrainSocket reports that this node has no drain socket for the volume,
// which is how a node learns it does not serve it. It is deliberately distinct
// from a socket that answered badly: only the former is worth routing onward,
// and conflating the two makes a node re-run a drain that has already failed.
var ErrNoDrainSocket = errors.New("no drain socket on this node")
// DrainVolumeSocket asks the NBD plugin serving volumeID on this node to flush
// its WAL chunks and live checkpoint to S3, then waits for the ack. The plugin
// creates the socket under {dataDir}/viperblock/{volumeID}/, so this only
// succeeds on the node hosting the volume; the daemon calls it when a drain
// command is routed to it.
func DrainVolumeSocket(dataDir, volumeID string) error {
sockPath := filepath.Join(dataDir, "viperblock", volumeID, "snapshot.sock")
conn, err := net.DialTimeout("unix", sockPath, drainDialTimeout)
if err != nil {
return fmt.Errorf("%w: dial %s: %w", ErrNoDrainSocket, sockPath, err)
}
defer conn.Close()
_ = conn.SetDeadline(time.Now().Add(drainAckTimeout))
buf := make([]byte, 8)
n, err := conn.Read(buf)
if err != nil {
return fmt.Errorf("read drain ack for %s: %w", volumeID, err)
}
if !strings.HasPrefix(string(buf[:n]), "OK") {
return fmt.Errorf("drain of %s did not ack OK: %q", volumeID, string(buf[:n]))
}
return nil
}
// drainVolume flushes the volume's in-flight writes to S3 before the snapshot
// reads the live checkpoint from there.
func (s *SnapshotServiceImpl) drainVolume(ctx context.Context, volumeID string, fallbackState, fallbackInstance string, accountID string) error {
return DrainVolume(ctx, s.config, s.store, s.natsConn, volumeID, fallbackState, fallbackInstance, accountID)
}
// DrainVolume flushes the volume's in-flight writes to S3 before a live
// checkpoint is read from there. Shared by ec2.CreateSnapshot and
// ec2.CreateImage's running-instance path (domains/ec2/image) so the two
// live-snapshot call sites cannot diverge on when a drain is required.
//
// The drain socket lives on the node serving the volume, but both callers are
// queue-grouped across every node, so the node answering the request usually
// is not that node. Whether a drain is required is therefore decided from the
// volume's attachment record, never from whether the socket happens to be
// local: an attached volume is being written to right now, and reading its
// checkpoint without draining silently captures an older one.
func DrainVolume(ctx context.Context, cfg *config.Config, store objectstore.ObjectStore, natsConn *nats.Conn, volumeID string, fallbackState, fallbackInstance string, accountID string) error {
// A metadata-only snapshot never reads the live checkpoint, so there is
// nothing for a drain to make current.
if cfg == nil || cfg.Predastore.Host == "" {
return nil
}
state, instanceID, err := volumeAttachment(ctx, store, cfg.Predastore.Bucket, volumeID, fallbackState, fallbackInstance)
if err != nil {
return err
}
// Nothing is writing to an unattached volume: the checkpoint its Close()
// left behind is the current one. Decide this before touching the socket —
// dialing it is not a probe, it makes the plugin run a full flush.
if state != "in-use" || instanceID == "" {
slog.InfoContext(ctx, "DrainVolume: volume not attached, snapshotting the checkpoint left by Close (stopped instance path)",
"volumeId", volumeID, "state", state, "attachedInstance", instanceID)
return nil
}
// Fast path: the volume is served by this node, so no hop is needed. This is
// every single-node deployment, and the node hosting the instance on a
// cluster.
localErr := DrainVolumeSocket(cfg.DataDir, volumeID)
if localErr == nil {
return nil
}
// A socket that answered but did not ack means this node does serve the
// volume and the flush itself failed. Routing that onward would land back on
// this same node and repeat the drain that has already failed.
if !errors.Is(localErr, ErrNoDrainSocket) {
return fmt.Errorf("drain volume %s on this node: %w", volumeID, localErr)
}
slog.InfoContext(ctx, "DrainVolume: volume attached, routing drain to the node hosting it",
"volumeId", volumeID, "instanceId", instanceID)
return drainOnHostNode(ctx, natsConn, volumeID, instanceID, accountID)
}
// volumeAttachment returns the volume's authoritative state and attached
// instance. state.json is control-plane-owned and authoritative; the fallback
// pair (passed in from the caller's own volume record) is rewritten by the
// live NBD plugin from its stale in-memory state and is only used for volumes
// predating the split.
func volumeAttachment(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string, fallbackState, fallbackInstance string) (state, instanceID string, err error) {
rec, found, err := volumestate.Read(ctx, store, bucket, volumeID)
if err != nil {
return "", "", fmt.Errorf("read volume state for %s: %w", volumeID, err)
}
if found {
return rec.State, rec.AttachedInstance, nil
}
return fallbackState, fallbackInstance, nil
}
// drainOnHostNode issues the drain on the node hosting instanceID. Only that
// node subscribes ec2.cmd.{instanceID}, so the command is self-routing and no
// volume-to-node resolution is needed here.
//
// An attachment record is not proof of a live writer: stop deliberately leaves
// a boot volume attached (runtime/daemon/vm_adapters.go's volumeMounterAdapter.Unmount)
// while tearing down both the NBD plugin and this subscription. The two
// not-running signals below are therefore the stopped-instance path, not a
// failure — treating them as one would make every stopped instance's root
// volume permanently unsnapshottable.
func drainOnHostNode(ctx context.Context, natsConn *nats.Conn, volumeID, instanceID, accountID string) error {
command := ec2v1.EC2InstanceCommand{
ID: instanceID,
Attributes: ec2v1.EC2CommandAttributes{DrainVolume: true},
DrainVolumeData: &ec2v1.DrainVolumeData{VolumeID: volumeID},
}
resp, err := natsmsg.NATSRequest[ec2v1.DrainVolumeResponse](ctx, natsConn,
ec2v1.InstanceCommandSubject(instanceID), command, drainRequestTimeout, accountID)
if err != nil {
// No subscriber at all: the instance runs nowhere in the cluster, so it
// was stopped (stop migrates it to shared KV before unsubscribing) and
// its volumes were sealed by Close. Warn rather than Info because the
// same shape appears if the hosting node has lost NATS while its VM
// keeps writing, which yields a stale snapshot as it did before routing
// existed. A live host that answers and fails is still a hard error.
if errors.Is(err, nats.ErrNoResponders) {
slog.WarnContext(ctx, "drainVolume: no node hosts the instance, treating the volume as stopped and snapshotting its sealed checkpoint",
"volumeId", volumeID, "instanceId", instanceID)
return nil
}
return fmt.Errorf("drain volume %s on the node hosting %s: %w", volumeID, instanceID, err)
}
switch resp.Status {
case ec2v1.DrainVolumeStatusDrained:
return nil
// The host still holds the instance but it is not running: nothing is
// writing, so there is nothing to flush. This is a host-drain stop, which
// keeps the VM (and this subscription) in place after unmounting.
case ec2v1.DrainVolumeStatusNotRunning:
slog.InfoContext(ctx, "drainVolume: instance is not running on its host, snapshotting the volume's sealed checkpoint",
"volumeId", volumeID, "instanceId", instanceID)
return nil
default:
return fmt.Errorf("drain volume %s on the node hosting %s: unexpected ack %q", volumeID, instanceID, resp.Status)
}
}
// describeSnapshotsValidFilters defines the set of filter names accepted by DescribeSnapshots.
var describeSnapshotsValidFilters = map[string]bool{
"snapshot-id": true,
@@ -557,21 +740,6 @@ func unreadableRecordError(ctx context.Context, snapshotID, accountID, kind stri
"Cannot confirm that snapshot '%s' is not in use: %d %s record(s) could not be read", snapshotID, len(unreadable), kind)
}
// ProviderCreateError maps a provider CreateSnapshot failure of volumeID to
// the code a client can act on: retry-later for a provider that cannot answer
// now, IncorrectState for one that needs the volume detached, else internal.
func ProviderCreateError(volumeID string, err error) error {
switch {
case errors.Is(err, ebsprovider.ErrUnavailable):
return errors.New(awserrors.ErrorServiceUnavailable)
case errors.Is(err, ebsprovider.ErrVolumeInUse):
return awserrors.Errorf(awserrors.ErrorIncorrectState,
"The volume '%s' is in use. Detach it before creating a snapshot.", volumeID)
default:
return errors.New(awserrors.ErrorServerInternal)
}
}
// providerDeleteError maps a provider DeleteSnapshot failure to the code a
// client can act on. A transport failure is retryable: the delete is idempotent.
func providerDeleteError(snapshotID string, err error) error {
@@ -1519,55 +1519,3 @@ func TestDropEmptySnapshotRefsDoesNotDropAConcurrentAdd(t *testing.T) {
assert.True(t, has, "iteration %d: the concurrent add was dropped by the empty-key delete", i)
}
}
// createFailingProvider fails only CreateSnapshot, with err.
type createFailingProvider struct {
ebsprovider.EBSProvider
err error
}
func (p *createFailingProvider) CreateSnapshot(context.Context, ebsprovider.CreateSnapshotRequest) (*ebsprovider.Snapshot, error) {
return nil, p.err
}
// A provider that cannot snapshot a volume while it is attached refuses it
// as in use, which the client can act on by detaching the volume first.
func TestCreateSnapshot_ProviderVolumeInUseIsIncorrectState(t *testing.T) {
svc, store := setupTestSnapshotService(t)
createTestVolume(t, svc, store, "vol-create-inuse", 10)
svc.SetEBSProvider(&createFailingProvider{EBSProvider: svc.provider,
err: &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeVolumeInUse, Message: "published"}})
_, err := svc.CreateSnapshot(context.Background(), &ec2.CreateSnapshotInput{VolumeId: aws.String("vol-create-inuse")}, testAccountID)
code, message := clientErrorDetail(t, err)
assert.Equal(t, awserrors.ErrorIncorrectState, code)
assert.Contains(t, message, "vol-create-inuse")
assert.Contains(t, message, "Detach")
}
// A provider that cannot answer now, such as one refusing a volume another
// node holds, is retry-later for the client; any other failure is internal.
func TestCreateSnapshot_ProviderFailureMapsToSpecificError(t *testing.T) {
cases := []struct {
name string
err error
wantCode string
}{
{"provider unavailable", &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeUnavailable, Message: "held by node-2"}, awserrors.ErrorServiceUnavailable},
{"internal failure", &ebsprovider.ProviderError{Code: ebsprovider.ErrorCodeInternal, Message: "boom"}, awserrors.ErrorServerInternal},
{"transport failure", errors.New("connection reset"), awserrors.ErrorServerInternal},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
svc, store := setupTestSnapshotService(t)
createTestVolume(t, svc, store, "vol-create-fail", 10)
svc.SetEBSProvider(&createFailingProvider{EBSProvider: svc.provider, err: tc.err})
_, err := svc.CreateSnapshot(context.Background(), &ec2.CreateSnapshotInput{VolumeId: aws.String("vol-create-fail")}, testAccountID)
require.Error(t, err)
assert.Equal(t, tc.wantCode, err.Error())
})
}
}
@@ -0,0 +1,78 @@
// Package volumestate reads and writes the control-plane-owned attachment
// state for an EBS volume, persisted to a per-volume state.json object.
//
// It is kept out of config.json because config.json is rewritten by the live
// nbdkit VB on every SaveState (clobbering any State the control plane wrote
// there) and is a sealed object for encrypted volumes (a second writer reuses
// the AES-GCM nonce). state.json is plaintext, viperblock never touches it, so
// the control plane is its single writer. Readers outside the volume service
// (the snapshot path) need the same record, which is why it lives here.
package volumestate
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
)
// Record is the control-plane-owned attachment state of a volume.
type Record struct {
State string `json:"state"`
AttachedInstance string `json:"attachedInstance"`
DeviceName string `json:"deviceName"`
AttachedAt time.Time `json:"attachedAt"`
}
// Key is the S3 key for a volume's control-plane state object.
func Key(volumeID string) string { return volumeID + "/state.json" }
// Write persists the attachment state, replacing any previous record.
func Write(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string, rec Record) error {
data, err := json.Marshal(rec)
if err != nil {
return fmt.Errorf("marshal volume state: %w", err)
}
_, err = store.PutObject(ctx, &s3.PutObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(Key(volumeID)),
Body: bytes.NewReader(data),
})
if err != nil {
return fmt.Errorf("failed to write volume state to S3: %w", err)
}
return nil
}
// Read returns the attachment state. found=false with a nil error means the
// object is absent (a volume predating the state.json split), in which case the
// caller falls back to the State embedded in config.json.
func Read(ctx context.Context, store objectstore.ObjectStore, bucket, volumeID string) (Record, bool, error) {
getResult, err := store.GetObject(ctx, &s3.GetObjectInput{
Bucket: aws.String(bucket),
Key: aws.String(Key(volumeID)),
})
if err != nil {
if objectstore.IsNoSuchKeyError(err) {
return Record{}, false, nil
}
return Record{}, false, fmt.Errorf("failed to get volume state: %w", err)
}
defer getResult.Body.Close()
body, err := io.ReadAll(getResult.Body)
if err != nil {
return Record{}, false, fmt.Errorf("failed to read volume state body: %w", err)
}
var rec Record
if err := json.Unmarshal(body, &rec); err != nil {
return Record{}, false, fmt.Errorf("failed to unmarshal volume state: %w", err)
}
return rec, true, nil
}
+25 -1
View File
@@ -122,7 +122,7 @@ func (m *ExternalIPAM) AllocateIP(ctx context.Context, region, az, purpose, allo
if m == nil {
return "", "", fmt.Errorf("external IPAM unavailable on this node: %w", errors.New(awserrors.ErrorInsufficientAddressCapacity))
}
pool := external.FindPool(m.pools, region, az)
pool := m.findPool(region, az)
if pool == nil {
return "", "", fmt.Errorf("no external pool available for region=%q az=%q: %w", region, az, errors.New(awserrors.ErrorInsufficientAddressCapacity))
}
@@ -221,6 +221,30 @@ func (m *ExternalIPAM) allocatorFor(poolName string) (external.Allocator, error)
return m.static, nil
}
// findPool returns the best pool for the given region/AZ using the same
// fallback order as topology.go: AZ-scoped → region-scoped → unscoped.
func (m *ExternalIPAM) findPool(region, az string) *external.ExternalPoolConfig {
for i := range m.pools {
p := &m.pools[i]
if p.AZ != "" && p.AZ == az && p.Region == region {
return p
}
}
for i := range m.pools {
p := &m.pools[i]
if p.AZ == "" && p.Region != "" && p.Region == region {
return p
}
}
for i := range m.pools {
p := &m.pools[i]
if p.Region == "" && p.AZ == "" {
return p
}
}
return nil
}
// filterStatic returns only the static pools (Source unset or "static").
func filterStatic(pools []external.ExternalPoolConfig) []external.ExternalPoolConfig {
out := make([]external.ExternalPoolConfig, 0, len(pools))
+3 -10
View File
@@ -99,7 +99,7 @@ func TestRotateOnce_MintsAndSwaps(t *testing.T) {
r.rotateOnce(t.Context(), time.Now().UTC())
// A new active key took over.
assert.NotEqual(t, key.Kid, activeKid(issuer), "issuer should sign with the rotated key")
assert.NotEqual(t, key.Kid, issuer.ActiveKid(), "issuer should sign with the rotated key")
// Both old and new tokens verify (previous key retained).
_, err = verifier.Verify(oldTok)
@@ -142,7 +142,7 @@ func TestRotateOnce_PrunesExpiredKey(t *testing.T) {
require.NoError(t, err)
require.Len(t, metas, 1, "the older key should be pruned")
assert.Equal(t, second.Kid, active.Kid, "newest key remains active")
assert.Equal(t, second.Kid, activeKid(issuer))
assert.Equal(t, second.Kid, issuer.ActiveKid())
}
// TestRotator_RunRotatesThenStops drives the scheduler loop on a fast interval
@@ -169,7 +169,7 @@ func TestRotator_RunRotatesThenStops(t *testing.T) {
stopped := make(chan struct{})
go func() { r.Run(ctx); close(stopped) }()
require.Eventually(t, func() bool { return activeKid(issuer) != key.Kid },
require.Eventually(t, func() bool { return issuer.ActiveKid() != key.Kid },
2*time.Second, 5*time.Millisecond, "rotator should mint a new active key")
cancel()
@@ -218,10 +218,3 @@ func TestIssuerVerifier_ConcurrentSwap(t *testing.T) {
}
<-done
}
// activeKid reads the kid the issuer currently signs with.
func activeKid(i *Issuer) string {
i.mu.RLock()
defer i.mu.RUnlock()
return i.key.Kid
}
+10
View File
@@ -91,6 +91,16 @@ func (i *Issuer) SetActiveKey(key *SigningKey) {
i.mu.Unlock()
}
// ActiveKid returns the kid of the current signing key, or "" if none.
func (i *Issuer) ActiveKid() string {
i.mu.RLock()
defer i.mu.RUnlock()
if i.key == nil {
return ""
}
return i.key.Kid
}
// Mint returns a signed ES256 JWT for p and its expiry. The kid header lets any
// verifier resolve the public key from the shared signing-key set.
func (i *Issuer) Mint(p Principal) (token string, expiresAt time.Time, err error) {
+32 -3
View File
@@ -2,12 +2,14 @@ package awsapi
import (
"errors"
"io"
"log/slog"
"maps"
"net/http"
"slices"
"strings"
"github.com/mulgadc/bluebottle/pkg/sigv4"
ecrauth "github.com/mulgadc/spinifex/spinifex/domains/ecr/auth"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -32,7 +34,7 @@ type Deps struct {
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: dispatch.ResolveTargetAction,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeJSON,
Inventory: OperationInventory(),
@@ -53,11 +55,25 @@ func OperationInventory() dispatch.Inventory {
}
}
// resolveAction is the one X-Amz-Target parser telemetry and dispatch share.
func resolveAction(r *http.Request) string {
return actionFromTarget(r.Header.Get("X-Amz-Target"))
}
// actionFromTarget extracts the action suffix from an X-Amz-Target header.
// Any "<Prefix>.<Action>" or bare "<Action>" form is accepted.
func actionFromTarget(target string) string {
if _, after, ok := strings.CutLast(target, "."); ok {
return after
}
return target
}
// dispatch resolves, scopes and authorizes the request from one read of its
// body, then runs the action. Errors are awserrors codes the gateway renders.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
action := dispatch.ResolveTargetAction(r)
action := resolveAction(r)
if action == "" {
return errors.New(awserrors.ErrorMissingAction)
}
@@ -76,7 +92,7 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
return errors.New(awserrors.ErrorInternalError)
}
body, err := dispatch.ReadBoundedBody(r)
body, err := readBoundedBody(r)
if err != nil {
slog.Error("ECR dispatch: failed to read body", "err", err)
return err
@@ -162,6 +178,19 @@ func roleIDOf(actor dispatch.Actor) string {
return roleID
}
// readBoundedBody reads the body under the SigV4 payload cap, so a body read
// ahead of the policy gate can neither bypass it nor exhaust memory.
func readBoundedBody(r *http.Request) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return body, nil
}
func without(values, excluded []string) []string {
return slices.DeleteFunc(slices.Clone(values), func(value string) bool {
return slices.Contains(excluded, value)
@@ -172,6 +172,18 @@ var composedCases = []struct{ action, body string }{
{"DeleteRepository", `{"repositoryName":"app"}`},
}
func TestActionFromTarget(t *testing.T) {
for target, want := range map[string]string{
TargetPrefix + ".CreateRepository": "CreateRepository",
"a.b.ListRepositories": "ListRepositories",
"GetAuthorizationToken": "GetAuthorizationToken",
TargetPrefix + ".": "",
"": "",
} {
assert.Equal(t, want, actionFromTarget(target), "target %q", target)
}
}
// Every composed action must be in the Actions namespace, or dispatch rejects
// it as InvalidAction before its resources are ever scoped.
func TestComposedActionsAreInTheNamespace(t *testing.T) {
+20
View File
@@ -28,22 +28,42 @@ func AccountBucket(accountID string) string {
// shapes every ECR storage path must use. Blobs are pooled per account for
// dedup; manifests and tags are indexed per repository.
const (
// repoMetaKeyFmt is the repo config object (scan-on-push, tag mutability).
repoMetaKeyFmt = "repos/%s/_meta.json"
// tagKeyFmt maps a tag to a digest pointer (body is "sha256:...").
tagKeyFmt = "repos/%s/manifests/by-tag/%s"
// manifestKeyFmt holds manifest JSON keyed by its own digest.
manifestKeyFmt = "repos/%s/manifests/by-digest/%s"
// indexKeyFmt is the rolling tag->digest index used by ListImages.
indexKeyFmt = "repos/%s/manifests/index.json"
// blobKeyFmt is the account-wide blob pool, two-char hex shard.
blobKeyFmt = "blobs/%s/%s"
// uploadKeyFmt holds in-progress chunked upload bytes.
uploadKeyFmt = "uploads/%s"
// uploadChunkKeyFmt holds a single revision's assembled bytes under a unique
// per-write token, so a CAS-winning PATCH always points at exactly the bytes
// it hashed (a losing concurrent write lands on a different, orphaned token).
uploadChunkKeyFmt = "uploads/%s/%s"
)
// RepoMetaKey returns the key for a repository's config object.
func RepoMetaKey(repo string) string { return fmt.Sprintf(repoMetaKeyFmt, repo) }
// TagKey returns the key for a tag's digest pointer.
func TagKey(repo, tag string) string { return fmt.Sprintf(tagKeyFmt, repo, tag) }
// ManifestKey returns the key for a manifest stored by its digest. The digest
// is sanitized via DigestToken so the object key carries no ':' separator.
func ManifestKey(repo, digest string) string {
return fmt.Sprintf(manifestKeyFmt, repo, DigestToken(digest))
}
// IndexKey returns the key for a repository's tag->digest index.
func IndexKey(repo string) string { return fmt.Sprintf(indexKeyFmt, repo) }
// UploadKey returns the key for an in-progress upload's bytes.
func UploadKey(uploadID string) string { return fmt.Sprintf(uploadKeyFmt, uploadID) }
// UploadChunkKey returns the key for one revision's assembled upload bytes,
// addressed by a unique per-write token.
func UploadChunkKey(uploadID, token string) string {
+4
View File
@@ -17,7 +17,11 @@ func TestKeyHelpers(t *testing.T) {
got string
want string
}{
{"repo meta", RepoMetaKey("team/app"), "repos/team/app/_meta.json"},
{"tag", TagKey("team/app", "v1"), "repos/team/app/manifests/by-tag/v1"},
{"manifest", ManifestKey("team/app", "sha256:abcd"), "repos/team/app/manifests/by-digest/sha256-abcd"},
{"index", IndexKey("team/app"), "repos/team/app/manifests/index.json"},
{"upload", UploadKey("u-1"), "uploads/u-1"},
}
for _, c := range cases {
if c.got != c.want {
+30 -3
View File
@@ -2,11 +2,14 @@ package awsapi
import (
"errors"
"io"
"log/slog"
"maps"
"net/http"
"slices"
"strings"
"github.com/mulgadc/bluebottle/pkg/sigv4"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -32,7 +35,7 @@ type Deps struct {
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: dispatch.ResolveTargetAction,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeJSON,
Inventory: OperationInventory(),
@@ -47,11 +50,35 @@ func OperationInventory() dispatch.Inventory {
}
}
func resolveAction(r *http.Request) string {
return actionFromTarget(r.Header.Get("X-Amz-Target"))
}
// actionFromTarget extracts the action suffix from an X-Amz-Target header.
// Any "<Prefix>.<Action>" or bare "<Action>" form is accepted.
func actionFromTarget(target string) string {
if _, after, ok := strings.CutLast(target, "."); ok {
return after
}
return target
}
func readBoundedBody(r *http.Request) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return body, nil
}
// dispatch resolves, scopes and authorizes the request, then runs the action.
// Errors are awserrors codes the gateway renders.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
action := dispatch.ResolveTargetAction(r)
action := resolveAction(r)
if action == "" {
return errors.New(awserrors.ErrorMissingAction)
}
@@ -72,7 +99,7 @@ func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
return errors.New(awserrors.ErrorInternalError)
}
body, err := dispatch.ReadBoundedBody(r)
body, err := readBoundedBody(r)
if err != nil {
slog.ErrorContext(r.Context(), "ECS dispatch: failed to read body", "err", err)
return err
@@ -0,0 +1,13 @@
package awsapi
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestActionFromTarget(t *testing.T) {
assert.Equal(t, "RunTask", actionFromTarget("AmazonEC2ContainerServiceV20141113.RunTask"))
assert.Equal(t, "CreateCluster", actionFromTarget("CreateCluster"))
assert.Empty(t, actionFromTarget(""))
}
+8 -9
View File
@@ -3,7 +3,6 @@ package ecs
import (
"context"
"errors"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"time"
"github.com/aws/aws-sdk-go/aws"
@@ -24,7 +23,7 @@ func (s *Service) PutClusterCapacityProviders(ctx context.Context, input *ecs.Pu
return nil, err
}
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
if err != nil {
return nil, err
}
@@ -34,7 +33,7 @@ func (s *Service) PutClusterCapacityProviders(ctx context.Context, input *ecs.Pu
rec.CapacityProviders = awsStringSlice(input.CapacityProviders)
rec.DefaultCapacityProviderStrategy = strategyFromAWS(input.DefaultCapacityProviderStrategy)
if err := taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(cluster), &rec); err != nil {
if err := putJSON(ctx, kv, ClusterMetaKey(cluster), &rec); err != nil {
return nil, err
}
return &ecs.PutClusterCapacityProvidersOutput{Cluster: rec.toAWS()}, nil
@@ -54,7 +53,7 @@ func (s *Service) CreateCapacityProvider(ctx context.Context, input *ecs.CreateC
}
var existing CapacityProviderRecord
found, err := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &existing)
found, err := getJSON(ctx, kv, CapacityProviderKey(name), &existing)
if err != nil {
return nil, err
}
@@ -70,7 +69,7 @@ func (s *Service) CreateCapacityProvider(ctx context.Context, input *ecs.CreateC
Tags: tagsToMap(input.Tags),
CreatedAt: time.Now().UTC(),
}
if err := taskdefinition.PutJSON(ctx, kv, CapacityProviderKey(name), &rec); err != nil {
if err := putJSON(ctx, kv, CapacityProviderKey(name), &rec); err != nil {
return nil, err
}
return &ecs.CreateCapacityProviderOutput{CapacityProvider: rec.toAWS()}, nil
@@ -87,13 +86,13 @@ func (s *Service) DescribeCapacityProviders(ctx context.Context, input *ecs.Desc
names := awsStringSlice(input.CapacityProviders)
if len(names) == 0 {
keys, kerr := taskdefinition.KeysWithPrefix(ctx, kv, CapacityProvidersPrefix())
keys, kerr := keysWithPrefix(ctx, kv, CapacityProvidersPrefix())
if kerr != nil {
return nil, kerr
}
for _, k := range keys {
var rec CapacityProviderRecord
ok, gerr := taskdefinition.GetJSON(ctx, kv, k, &rec)
ok, gerr := getJSON(ctx, kv, k, &rec)
if gerr != nil {
return nil, gerr
}
@@ -107,7 +106,7 @@ func (s *Service) DescribeCapacityProviders(ctx context.Context, input *ecs.Desc
for _, ref := range names {
name := CapacityProviderShortName(ref)
var rec CapacityProviderRecord
ok, gerr := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &rec)
ok, gerr := getJSON(ctx, kv, CapacityProviderKey(name), &rec)
if gerr != nil {
return nil, gerr
}
@@ -133,7 +132,7 @@ func (s *Service) DeleteCapacityProvider(ctx context.Context, input *ecs.DeleteC
return nil, err
}
var rec CapacityProviderRecord
found, err := taskdefinition.GetJSON(ctx, kv, CapacityProviderKey(name), &rec)
found, err := getJSON(ctx, kv, CapacityProviderKey(name), &rec)
if err != nil {
return nil, err
}
+4 -5
View File
@@ -3,7 +3,6 @@ package ecs
import (
"context"
"errors"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"strings"
"github.com/aws/aws-sdk-go/aws"
@@ -35,7 +34,7 @@ func (s *Service) DeleteCluster(ctx context.Context, input *ecs.DeleteClusterInp
return nil, err
}
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &rec)
if err != nil {
return nil, err
}
@@ -72,7 +71,7 @@ func (s *Service) DeregisterContainerInstance(ctx context.Context, input *ecs.De
return nil, err
}
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
if err != nil {
return nil, err
}
@@ -125,7 +124,7 @@ func (s *Service) UpdateContainerInstancesState(ctx context.Context, input *ecs.
for _, ref := range awsStringSlice(input.ContainerInstances) {
id := ContainerInstanceShortID(ref)
var rec InstanceRecord
found, gerr := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
found, gerr := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
if gerr != nil {
return nil, gerr
}
@@ -135,7 +134,7 @@ func (s *Service) UpdateContainerInstancesState(ctx context.Context, input *ecs.
}
rec.Status = status
rec.Reaped = false
if perr := taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, id), &rec); perr != nil {
if perr := putJSON(ctx, kv, InstanceKey(cluster, id), &rec); perr != nil {
return nil, perr
}
if status == InstanceStatusDraining {
@@ -28,7 +28,7 @@ func TestService_DeleteCluster_CascadesAndSweeps(t *testing.T) {
list, err := svc.ListClusters(context.Background(), &ecs.ListClustersInput{}, testAccountID)
require.NoError(t, err)
assert.Empty(t, list.ClusterArns)
keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, clusterKeyPrefix("web"))
keys, err := keysWithPrefix(t.Context(), kv, clusterKeyPrefix("web"))
require.NoError(t, err)
assert.Empty(t, keys)
}
+3 -3
View File
@@ -19,7 +19,7 @@ import (
func reloadService(t *testing.T, kv jetstream.KeyValue, cluster, name string) *ServiceRecord {
t.Helper()
var rec ServiceRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, ServiceKey(cluster, name), &rec)
found, err := getJSON(t.Context(), kv, ServiceKey(cluster, name), &rec)
require.NoError(t, err)
require.True(t, found)
return &rec
@@ -329,7 +329,7 @@ func TestDeployment_LegacyServiceSynthesizesPrimary(t *testing.T) {
DesiredCount: 1, Status: ServiceStatusActive,
SchedulingStrategy: SchedulingStrategyReplica, DeploymentID: "legacy-dep",
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "legacy"), rec))
require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "legacy"), rec))
require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, rec))
reloaded := reloadService(t, kv, "web", "legacy")
@@ -388,7 +388,7 @@ func TestDeployment_RepeatedStartFailuresBackOffWithoutTheCircuitBreaker(t *test
// Once it passes, launching resumes: the backoff is a brake, not a stop.
rec = reloadService(t, kv, "web", "web")
rec.primaryDeployment().NextLaunchAt = time.Now().UTC().Add(-time.Second)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), rec))
require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), rec))
require.NoError(t, svc.reconcileService(context.Background(), kv, testAccountID, reloadService(t, kv, "web", "web")))
assert.Positive(t, liveServiceTasks(t, svc, kv, "web", "web"), "backoff never released the deployment")
+4 -5
View File
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"github.com/stretchr/testify/assert"
@@ -38,14 +37,14 @@ func TestAssignTaskPublicIP_EnabledAssociatesAndPersists(t *testing.T) {
eips := &stubEIPs{publicIP: "192.0.2.50", allocID: "eipalloc-1"}
svc.eips = eips
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
Name: "web", Cluster: "web", AssignPublicIP: "ENABLED",
}))
task := &TaskRecord{
TaskID: "t-1", Cluster: "web", Group: serviceTaskGroup("web"),
ENIID: "eni-1", ENIPrivateIP: "172.31.0.8",
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
svc.assignTaskPublicIP(context.Background(), kv, testAccountID, task)
@@ -54,7 +53,7 @@ func TestAssignTaskPublicIP_EnabledAssociatesAndPersists(t *testing.T) {
assert.Equal(t, "eipalloc-1", task.ENIEIPAllocationID)
var persisted TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &persisted)
found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &persisted)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, "192.0.2.50", persisted.ENIPublicIP)
@@ -65,7 +64,7 @@ func TestAssignTaskPublicIP_DisabledNoOp(t *testing.T) {
eips := &stubEIPs{publicIP: "192.0.2.50", allocID: "eipalloc-1"}
svc.eips = eips
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
require.NoError(t, putJSON(t.Context(), kv, ServiceKey("web", "web"), &ServiceRecord{
Name: "web", Cluster: "web", // AssignPublicIP unset
}))
task := &TaskRecord{
+4 -4
View File
@@ -189,14 +189,14 @@ func singleTaskRecord(t *testing.T, svc *Service, cluster string) (TaskRecord, b
t.Helper()
kv, err := svc.bucket(context.Background(), testAccountID)
require.NoError(t, err)
keys, err := taskdefinition.KeysWithPrefix(context.Background(), kv, TasksPrefix(cluster))
keys, err := keysWithPrefix(context.Background(), kv, TasksPrefix(cluster))
require.NoError(t, err)
require.LessOrEqual(t, len(keys), 1, "test expects at most one task record")
if len(keys) == 0 {
return TaskRecord{}, false
}
var rec TaskRecord
found, err := taskdefinition.GetJSON(context.Background(), kv, keys[0], &rec)
found, err := getJSON(context.Background(), kv, keys[0], &rec)
require.NoError(t, err)
return rec, found
}
@@ -317,10 +317,10 @@ func TestReaper_Awsvpc_ReleasesENI(t *testing.T) {
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
var rec InstanceRecord
_, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
_, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
require.NoError(t, err)
rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
sc := NewScheduler(nc, svc, "test-holder")
sc.reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
+2
View File
@@ -15,6 +15,8 @@ const TestAccountID = testAccountID
var (
NewTestService = newTestService
RegisterInstance = registerInstance
PutJSON = putJSON
GetJSON = getJSON
ServiceTaskGroup = serviceTaskGroup
)
+3 -7
View File
@@ -1,10 +1,6 @@
package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
)
import "context"
// ReportTaskGPUInput is the agent's per-task GPU device-assignment report
// (agent → gateway). Internal agent↔gateway shape, not an AWS SDK action:
@@ -40,7 +36,7 @@ func (s *Service) ReportTaskGPU(ctx context.Context, input *ReportTaskGPUInput,
return nil, err
}
var task TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &task)
found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &task)
if err != nil {
return nil, err
}
@@ -59,7 +55,7 @@ func (s *Service) ReportTaskGPU(ctx context.Context, input *ReportTaskGPUInput,
}
}
if changed {
if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), &task); err != nil {
if err := putJSON(ctx, kv, TaskKey(cluster, taskID), &task); err != nil {
return nil, err
}
}
+10 -11
View File
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"github.com/aws/aws-sdk-go/aws"
@@ -18,7 +17,7 @@ func TestReportTaskGPU_MergesOntoTaskContainers(t *testing.T) {
svc, _ := newTestService(t)
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
TaskID: "t-1", Cluster: "web",
Containers: []ContainerState{
{Name: "web", Status: "RUNNING"},
@@ -34,7 +33,7 @@ func TestReportTaskGPU_MergesOntoTaskContainers(t *testing.T) {
assert.Equal(t, "OK", out.Acknowledgment)
var rec TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
for _, c := range rec.Containers {
@@ -64,7 +63,7 @@ func TestDescribeTasks_GPUIDsPopulatedFromReport(t *testing.T) {
svc, _ := newTestService(t)
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
Containers: []ContainerState{{Name: "trainer", Status: "RUNNING", GPUIDs: []string{"GPU-aaa", "GPU-bbb"}}},
}))
@@ -86,7 +85,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
svc, _ := newTestService(t)
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
ContainerInstanceID: "i-1",
Containers: []ContainerState{{Name: "trainer", Status: "PENDING"}},
@@ -100,7 +99,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
}))
var rec TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
require.Len(t, rec.Containers, 1)
@@ -115,7 +114,7 @@ func TestRecordTaskState_PreservesGPUIDsAcrossStoppedReport(t *testing.T) {
Containers: []ecsv1.ContainerStatus{{Name: "trainer", Status: ecsv1.TaskStatusStopped, ExitCode: &exit}},
}))
found, err = taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
found, err = getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
require.Len(t, rec.Containers, 1)
@@ -138,7 +137,7 @@ func TestRecordTaskState_ReportedGPUIDsOverwritePinned(t *testing.T) {
svc, _ := newTestService(t)
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
ContainerInstanceID: "i-1",
Containers: []ContainerState{{Name: "trainer", Status: "RUNNING", GPUIDs: []string{"GPU-aaa"}}},
@@ -151,7 +150,7 @@ func TestRecordTaskState_ReportedGPUIDsOverwritePinned(t *testing.T) {
}))
var rec TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
require.Len(t, rec.Containers, 1)
@@ -164,7 +163,7 @@ func TestRecordTaskState_NonGPUContainerStaysEmpty(t *testing.T) {
svc, _ := newTestService(t)
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &TaskRecord{
TaskID: "t-1", ARN: TaskARN(svc.region, testAccountID, "web", "t-1"), Cluster: "web",
ContainerInstanceID: "i-1",
Containers: []ContainerState{{Name: "web", Status: "PENDING"}},
@@ -184,7 +183,7 @@ func TestRecordTaskState_NonGPUContainerStaysEmpty(t *testing.T) {
}))
var rec TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
found, err := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
require.Len(t, rec.Containers, 1)
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"time"
@@ -28,7 +27,7 @@ func instanceStatus(t *testing.T, svc *Service, cluster, id string) InstanceReco
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
var rec InstanceRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey(cluster, id), &rec)
found, err := getJSON(t.Context(), kv, InstanceKey(cluster, id), &rec)
require.NoError(t, err)
require.True(t, found)
return rec
@@ -47,7 +46,7 @@ func TestRegister_RestoresReapedInstanceToActive(t *testing.T) {
require.NoError(t, err)
rec := instanceStatus(t, svc, "web", "i-1")
rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
NewScheduler(nc, svc, "test-holder").reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
@@ -87,7 +86,7 @@ func TestReaper_ReleasesCapacityAndRecoversClean(t *testing.T) {
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
seeded.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &seeded))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &seeded))
NewScheduler(nc, svc, "test-holder").reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
+12 -13
View File
@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"errors"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"log/slog"
"slices"
"time"
@@ -18,14 +17,14 @@ import (
// listInstanceRecords returns all container-instance records in a cluster.
func (s *Service) listInstanceRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]InstanceRecord, error) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, InstancesPrefix(cluster))
keys, err := keysWithPrefix(ctx, kv, InstancesPrefix(cluster))
if err != nil {
return nil, err
}
out := make([]InstanceRecord, 0, len(keys))
for _, k := range keys {
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
@@ -139,7 +138,7 @@ func (s *Service) DescribeContainerInstances(ctx context.Context, input *ecs.Des
for _, ref := range awsStringSlice(input.ContainerInstances) {
id := ContainerInstanceShortID(ref)
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
if err != nil {
return nil, err
}
@@ -195,7 +194,7 @@ func (s *Service) ListContainerInstances(ctx context.Context, input *ecs.ListCon
// it back. Used by both the AWS register path and the bus register handler.
func (s *Service) upsertInstance(ctx context.Context, kv jetstream.KeyValue, accountID, cluster, instanceID string, mutate func(*InstanceRecord)) (*InstanceRecord, error) {
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec)
found, err := getJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec)
if err != nil {
return nil, err
}
@@ -213,7 +212,7 @@ func (s *Service) upsertInstance(ctx context.Context, kv jetstream.KeyValue, acc
if mutate != nil {
mutate(&rec)
}
if err := taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec); err != nil {
if err := putJSON(ctx, kv, InstanceKey(cluster, instanceID), &rec); err != nil {
return nil, err
}
return &rec, nil
@@ -346,7 +345,7 @@ func (s *Service) recordHeartbeat(ctx context.Context, msg *ecsv1.Heartbeat) err
return err
}
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
found, err := getJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
if err != nil || !found {
return err
}
@@ -354,7 +353,7 @@ func (s *Service) recordHeartbeat(ctx context.Context, msg *ecsv1.Heartbeat) err
if msg.Status != "" {
rec.Status = msg.Status
}
return taskdefinition.PutJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
return putJSON(ctx, kv, InstanceKey(msg.ClusterName, msg.InstanceID), &rec)
}
// recordTaskState applies an agent task-state report: it updates the task record
@@ -365,7 +364,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
return err
}
var task TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task)
found, err := getJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task)
if err != nil || !found {
return err
}
@@ -407,7 +406,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
// filtering on desired status would otherwise see it as still wanted.
task.DesiredStatus = TaskStatusStopped
}
if err := taskdefinition.PutJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil {
if err := putJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); err != nil {
return err
}
@@ -431,7 +430,7 @@ func (s *Service) recordTaskState(ctx context.Context, msg *ecsv1.TaskState) err
s.reclaimAssignInbox(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
s.reclaimStopInbox(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID)
s.reclaimTaskENI(ctx, msg.AccountID, &task)
if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil {
if perr := putJSON(ctx, kv, TaskKey(msg.ClusterName, msg.TaskID), &task); perr != nil {
slog.ErrorContext(ctx, "ECS task STOPPED: persist after EIP release failed", "task", msg.TaskID, "err", perr)
}
return s.releaseReservation(ctx, kv, msg.ClusterName, task.ContainerInstanceID, msg.TaskID, task.ReservedCPU, task.ReservedMemoryMiB, task.GPU, eniReservationFor(task.NetworkMode))
@@ -486,7 +485,7 @@ func (s *Service) recordDeploymentFailure(ctx context.Context, kv jetstream.KeyV
return
}
var svc ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &svc)
found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &svc)
if err != nil || !found {
return
}
@@ -494,7 +493,7 @@ func (s *Service) recordDeploymentFailure(ctx context.Context, kv jetstream.KeyV
if svc.Deployments[i].ID == depID {
svc.Deployments[i].FailedTasks++
svc.Deployments[i].UpdatedAt = time.Now().UTC()
if perr := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &svc); perr != nil {
if perr := putJSON(ctx, kv, ServiceKey(cluster, name), &svc); perr != nil {
slog.ErrorContext(ctx, "ECS deployment failure accounting: persist failed", "service", name, "err", perr)
}
return
+4 -5
View File
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
"github.com/nats-io/nats.go/jetstream"
@@ -56,14 +55,14 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
_ = kv.Delete(ctx, StopKey(cluster, instanceID, TaskShortID(taskID)))
}
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, AssignmentsPrefix(cluster, instanceID))
keys, err := keysWithPrefix(ctx, kv, AssignmentsPrefix(cluster, instanceID))
if err != nil {
return nil, err
}
out := &PollAssignmentsOutput{}
for _, key := range keys {
var as ecsv1.Assign
found, err := taskdefinition.GetJSON(ctx, kv, key, &as)
found, err := getJSON(ctx, kv, key, &as)
if err != nil {
return nil, err
}
@@ -72,13 +71,13 @@ func (s *Service) PollAssignments(ctx context.Context, input *PollAssignmentsInp
}
}
stopKeys, err := taskdefinition.KeysWithPrefix(ctx, kv, StopsPrefix(cluster, instanceID))
stopKeys, err := keysWithPrefix(ctx, kv, StopsPrefix(cluster, instanceID))
if err != nil {
return nil, err
}
for _, key := range stopKeys {
var sd ecsv1.StopDirective
found, err := taskdefinition.GetJSON(ctx, kv, key, &sd)
found, err := getJSON(ctx, kv, key, &sd)
if err != nil {
return nil, err
}
+3 -4
View File
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
@@ -17,9 +16,9 @@ import (
func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
svc, _, kv := serviceTestRig(t)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, AssignmentKey("web", "i-1", "t-1"),
require.NoError(t, putJSON(t.Context(), kv, AssignmentKey("web", "i-1", "t-1"),
&ecsv1.Assign{TaskID: "t-1", ClusterName: "web", InstanceID: "i-1"}))
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"),
require.NoError(t, putJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"),
&ecsv1.StopDirective{TaskID: "t-2", Reason: "drain"}))
out, err := svc.PollAssignments(context.Background(), &PollAssignmentsInput{
@@ -42,7 +41,7 @@ func TestPollAssignments_DrainsStopsAndAcks(t *testing.T) {
assert.Empty(t, out.Assignments)
assert.Empty(t, out.Stops)
found, err := taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &ecsv1.StopDirective{})
found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-2"), &ecsv1.StopDirective{})
require.NoError(t, err)
assert.False(t, found)
}
+11 -12
View File
@@ -7,7 +7,6 @@ package ecs
import (
"context"
"encoding/json"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"time"
@@ -61,11 +60,11 @@ func runOneTask(t *testing.T, svc *Service, kv jetstream.KeyValue) string {
Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1),
}, testAccountID)
require.NoError(t, err)
keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, TasksPrefix("web"))
keys, err := keysWithPrefix(t.Context(), kv, TasksPrefix("web"))
require.NoError(t, err)
require.Len(t, keys, 1)
var task TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, keys[0], &task)
found, err := getJSON(t.Context(), kv, keys[0], &task)
require.NoError(t, err)
require.True(t, found)
return task.TaskID
@@ -129,11 +128,11 @@ func TestScheduler_TheReaperAsksForTheSoonestHeartbeatWindow(t *testing.T) {
now := time.Now().UTC()
var inst InstanceRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
require.NoError(t, err)
require.True(t, found)
inst.LastSeen = now.Add(-30 * time.Second)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
due, err := sc.reap(t.Context())
require.NoError(t, err)
@@ -143,12 +142,12 @@ func TestScheduler_TheReaperAsksForTheSoonestHeartbeatWindow(t *testing.T) {
// Past the window the reaper acts, and with nothing live left it asks for
// nothing: the resync is the only thing that need bring it back.
inst.LastSeen = now.Add(-2 * heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
due, err = sc.reap(t.Context())
require.NoError(t, err)
assert.Zero(t, due)
found, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
found, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, InstanceStatusDraining, inst.Status, "an instance past its window is still reaped")
@@ -196,17 +195,17 @@ func TestScheduler_TheTimerPassReturnsTheSoonestOfItsJobs(t *testing.T) {
// A live instance due in 10s, and a stopped task due much later: the pass must
// come back for the sooner of the two.
var inst InstanceRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst)
require.NoError(t, err)
require.True(t, found)
inst.LastSeen = now.Add(10*time.Second - heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &inst))
stopped := TaskRecord{
TaskID: "t-1", Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", "t-1"),
LastStatus: TaskStatusStopped, DesiredStatus: TaskStatusStopped, StoppedAt: now,
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &stopped))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &stopped))
reapDue, err := sc.reap(t.Context())
require.NoError(t, err)
@@ -232,11 +231,11 @@ func TestScheduler_AnInFlightServiceAsksToBeRevisited(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, reconcileInterval, due)
keys, err := taskdefinition.KeysWithPrefix(t.Context(), kv, TasksPrefix("web"))
keys, err := keysWithPrefix(t.Context(), kv, TasksPrefix("web"))
require.NoError(t, err)
require.Len(t, keys, 1)
var task TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, keys[0], &task)
found, err := getJSON(t.Context(), kv, keys[0], &task)
require.NoError(t, err)
require.True(t, found)
require.NoError(t, svc.recordTaskState(t.Context(), &ecsv1.TaskState{
+5 -6
View File
@@ -5,7 +5,6 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"log/slog"
"strings"
"sync"
@@ -363,7 +362,7 @@ func (sc *Scheduler) reap(ctx context.Context) (time.Duration, error) {
// reapBucket reaps one account's disconnected instances and reports when the
// soonest instance it left alone falls due.
func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, accountID string, now time.Time) time.Duration {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
keys, err := keysWithPrefix(ctx, kv, "clusters/")
if err != nil {
return 0
}
@@ -373,7 +372,7 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
continue
}
var inst InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &inst)
found, err := getJSON(ctx, kv, k, &inst)
if err != nil || !found {
continue
}
@@ -388,7 +387,7 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
"lastSeen", inst.LastSeen.Format(time.RFC3339))
inst.Status = InstanceStatusDraining
inst.Reaped = true
if perr := taskdefinition.PutJSON(ctx, kv, k, &inst); perr != nil {
if perr := putJSON(ctx, kv, k, &inst); perr != nil {
continue
}
sc.stopInstanceTasks(ctx, kv, accountID, inst.Cluster, inst.InstanceID)
@@ -399,13 +398,13 @@ func (sc *Scheduler) reapBucket(ctx context.Context, kv jetstream.KeyValue, acco
// stopInstanceTasks transitions a reaped instance's non-stopped tasks to STOPPED
// and reclaims each awsvpc task's ENI (leak guard for a dead agent).
func (sc *Scheduler) stopInstanceTasks(ctx context.Context, kv jetstream.KeyValue, accountID, cluster, instanceID string) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
if err != nil {
return
}
for _, k := range keys {
var task TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &task)
found, err := getJSON(ctx, kv, k, &task)
if err != nil || !found {
continue
}
+54 -6
View File
@@ -5,7 +5,9 @@ package ecs
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"maps"
"slices"
@@ -166,12 +168,39 @@ func (s *Service) bucket(ctx context.Context, accountID string) (jetstream.KeyVa
return GetOrCreateAccountBucket(ctx, js, accountID)
}
// getJSON reads key into out. Returns (false, nil) when the key is absent.
func getJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
entry, err := kv.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return false, nil
}
if err != nil {
return false, err
}
if err := json.Unmarshal(entry.Value(), out); err != nil {
return false, fmt.Errorf("unmarshal %s: %w", key, err)
}
return true, nil
}
// putJSON marshals v and writes it at key.
func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
data, err := json.Marshal(v)
if err != nil {
return err
}
if _, err := kv.Put(ctx, key, data); err != nil {
return fmt.Errorf("put %s: %w", key, err)
}
return nil
}
// deleteKeysWithPrefix deletes every KV key under prefix. Used by DeleteCluster
// to sweep a cluster's whole key subtree (meta/instances/tasks/services/
// assignments) after its tasks are stopped. Best-effort per key; the first
// delete error is returned after attempting the rest.
func deleteKeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) error {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, prefix)
keys, err := keysWithPrefix(ctx, kv, prefix)
if err != nil {
return err
}
@@ -184,6 +213,25 @@ func deleteKeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix str
return firstErr
}
// keysWithPrefix returns all KV keys under prefix.
func keysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
}
if err != nil {
return nil, err
}
out := make([]string, 0, len(keys))
for _, k := range keys {
if strings.HasPrefix(k, prefix) {
out = append(out, k)
}
}
slices.Sort(out)
return out, nil
}
// --- Cluster ---
// CreateCluster persists a cluster meta record. Idempotent: re-creating an
@@ -199,7 +247,7 @@ func (s *Service) CreateCluster(ctx context.Context, input *ecs.CreateClusterInp
}
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(name), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(name), &rec)
if err != nil {
return nil, err
}
@@ -221,7 +269,7 @@ func (s *Service) CreateCluster(ctx context.Context, input *ecs.CreateClusterInp
rec.Tags[aws.StringValue(t.Key)] = aws.StringValue(t.Value)
}
}
if err := taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(name), &rec); err != nil {
if err := putJSON(ctx, kv, ClusterMetaKey(name), &rec); err != nil {
return nil, err
}
}
@@ -244,7 +292,7 @@ func (s *Service) DescribeClusters(ctx context.Context, input *ecs.DescribeClust
for _, name := range names {
name = ClusterShortName(name)
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(name), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(name), &rec)
if err != nil {
return nil, err
}
@@ -310,7 +358,7 @@ func (s *Service) ListClusters(ctx context.Context, _ *ecs.ListClustersInput, ac
if err != nil {
return nil, err
}
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
keys, err := keysWithPrefix(ctx, kv, "clusters/")
if err != nil {
return nil, err
}
@@ -320,7 +368,7 @@ func (s *Service) ListClusters(ctx context.Context, _ *ecs.ListClustersInput, ac
continue
}
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
+5 -5
View File
@@ -378,7 +378,7 @@ func TestService_TaskCredIDIsMintedSecret(t *testing.T) {
assert.Equal(t, as.CredID, second[taskID].CredID, "re-delivered assign changed credential ID")
var rec TaskRecord
found, gerr := taskdefinition.GetJSON(ctx, kv, TaskKey("web", taskID), &rec)
found, gerr := getJSON(ctx, kv, TaskKey("web", taskID), &rec)
require.NoError(t, gerr)
require.True(t, found)
assert.Equal(t, as.CredID, rec.CredID)
@@ -558,10 +558,10 @@ func TestScheduler_ReapBucket_StopsStaleInstanceTasks(t *testing.T) {
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
var rec InstanceRecord
_, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
_, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
require.NoError(t, err)
rec.LastSeen = time.Now().UTC().Add(-2 * heartbeatTimeout)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec))
sc := NewScheduler(nc, svc, "test-holder")
sc.reapBucket(context.Background(), kv, testAccountID, time.Now().UTC())
@@ -721,7 +721,7 @@ func TestService_RunTask_GPU_ReservesOnPlacement(t *testing.T) {
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
var rec InstanceRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
found, err := getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, 2, rec.TotalGPU)
@@ -732,7 +732,7 @@ func TestService_RunTask_GPU_ReservesOnPlacement(t *testing.T) {
AccountID: testAccountID, ClusterName: "web", InstanceID: "i-1", TaskID: taskID,
LastStatus: ecsv1.TaskStatusStopped, Reason: "exited",
}))
found, err = taskdefinition.GetJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
found, err = getJSON(t.Context(), kv, InstanceKey("web", "i-1"), &rec)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, 0, rec.ReservedGPU)
+15 -16
View File
@@ -6,7 +6,6 @@ import (
"encoding/json"
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"log/slog"
"strings"
"time"
@@ -146,7 +145,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
return nil, err
}
var clusterRec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
if err != nil {
return nil, err
}
@@ -155,7 +154,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
}
var existing ServiceRecord
if ok, gerr := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &existing); gerr != nil {
if ok, gerr := getJSON(ctx, kv, ServiceKey(cluster, name), &existing); gerr != nil {
return nil, gerr
} else if ok && existing.Status == ServiceStatusActive {
return &ecs.CreateServiceOutput{Service: s.serviceToAWS(accountID, &existing)}, nil
@@ -196,7 +195,7 @@ func (s *Service) CreateService(ctx context.Context, input *ecs.CreateServiceInp
applyDeploymentConfig(&rec, input.DeploymentConfiguration)
rec.LastGoodTaskDefARN = taskDef.ARN
rec.Deployments = []Deployment{newPrimaryDeployment(rec.DeploymentID, taskDef, rec.DesiredCount)}
if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
return nil, err
}
@@ -215,7 +214,7 @@ func (s *Service) UpdateService(ctx context.Context, input *ecs.UpdateServiceInp
return nil, err
}
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
if err != nil {
return nil, err
}
@@ -254,7 +253,7 @@ func (s *Service) UpdateService(ctx context.Context, input *ecs.UpdateServiceInp
}
}
rec.UpdatedAt = time.Now().UTC()
if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
return nil, err
}
if err := s.reconcileService(ctx, kv, accountID, &rec); err != nil {
@@ -273,7 +272,7 @@ func (s *Service) DeleteService(ctx context.Context, input *ecs.DeleteServiceInp
return nil, err
}
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
found, err := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
if err != nil {
return nil, err
}
@@ -293,7 +292,7 @@ func (s *Service) DeleteService(ctx context.Context, input *ecs.DeleteServiceInp
rec.RunningCount = 0
rec.PendingCount = 0
rec.UpdatedAt = time.Now().UTC()
if err := taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
if err := putJSON(ctx, kv, ServiceKey(cluster, name), &rec); err != nil {
return nil, err
}
return &ecs.DeleteServiceOutput{Service: s.serviceToAWS(accountID, &rec)}, nil
@@ -310,7 +309,7 @@ func (s *Service) DescribeServices(ctx context.Context, input *ecs.DescribeServi
for _, ref := range awsStringSlice(input.Services) {
name := ServiceShortName(ref)
var rec ServiceRecord
found, gerr := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, name), &rec)
found, gerr := getJSON(ctx, kv, ServiceKey(cluster, name), &rec)
if gerr != nil {
return nil, gerr
}
@@ -431,7 +430,7 @@ func persistServiceIfChanged(ctx context.Context, kv jetstream.KeyValue, svc *Se
if bytes.Equal(before, after) {
return nil
}
return taskdefinition.PutJSON(ctx, kv, ServiceKey(svc.Cluster, svc.Name), svc)
return putJSON(ctx, kv, ServiceKey(svc.Cluster, svc.Name), svc)
}
// reconcileAllServices reconciles every ACTIVE service in every ECS account
@@ -604,14 +603,14 @@ func (s *Service) listServiceTasks(ctx context.Context, kv jetstream.KeyValue, c
// listTaskRecords returns every task record in a cluster.
func (s *Service) listTaskRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]TaskRecord, error) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
if err != nil {
return nil, err
}
out := make([]TaskRecord, 0, len(keys))
for _, k := range keys {
var rec TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
@@ -624,14 +623,14 @@ func (s *Service) listTaskRecords(ctx context.Context, kv jetstream.KeyValue, cl
// listServiceRecords returns every service record in a cluster.
func (s *Service) listServiceRecords(ctx context.Context, kv jetstream.KeyValue, cluster string) ([]ServiceRecord, error) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, ServicesPrefix(cluster))
keys, err := keysWithPrefix(ctx, kv, ServicesPrefix(cluster))
if err != nil {
return nil, err
}
out := make([]ServiceRecord, 0, len(keys))
for _, k := range keys {
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
@@ -644,7 +643,7 @@ func (s *Service) listServiceRecords(ctx context.Context, kv jetstream.KeyValue,
// allServiceRecords returns every service record across all clusters in a bucket.
func (s *Service) allServiceRecords(ctx context.Context, kv jetstream.KeyValue) ([]ServiceRecord, error) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
keys, err := keysWithPrefix(ctx, kv, "clusters/")
if err != nil {
return nil, err
}
@@ -654,7 +653,7 @@ func (s *Service) allServiceRecords(ctx context.Context, kv jetstream.KeyValue)
continue
}
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
+5 -6
View File
@@ -3,7 +3,6 @@ package ecs
import (
"context"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"time"
@@ -470,7 +469,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.9",
ReservedCPU: 64, ReservedMemoryMiB: 64,
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-9"), task))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-9"), task))
// Cooperative stop: the task stays RUNNING with desiredStatus=STOPPED and a
// stop directive lands in the instance inbox; nothing is deregistered until the
@@ -484,7 +483,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
assert.Empty(t, reg.deregistered)
var sd ecsv1.StopDirective
found, err := taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
found, err := getJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, "bye", sd.Reason)
@@ -495,7 +494,7 @@ func TestService_StopTask_StopsAndDeregisters(t *testing.T) {
TaskID: "t-9", LastStatus: ecsv1.TaskStatusStopped,
}))
assert.Equal(t, []string{key(tgARN, "10.0.1.9", 80)}, reg.deregistered)
found, err = taskdefinition.GetJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
found, err = getJSON(t.Context(), kv, StopKey("web", "i-1", "t-9"), &sd)
require.NoError(t, err)
assert.False(t, found)
@@ -519,7 +518,7 @@ func TestService_StopTask_UnplacedForceStops(t *testing.T) {
TaskID: "t-np", Cluster: "web",
LastStatus: TaskStatusPending, DesiredStatus: TaskStatusRunning,
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-np"), task))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-np"), task))
out, err := svc.StopTask(context.Background(), &ecs.StopTaskInput{
Cluster: aws.String("web"), Task: aws.String("t-np"), Reason: aws.String("cancel"),
@@ -551,7 +550,7 @@ func TestService_TargetGroup_RegisterOnRunningDeregisterOnStopped(t *testing.T)
ContainerInstanceID: "i-1", LastStatus: TaskStatusPending,
NetworkMode: NetworkModeAwsvpc, ENIPrivateIP: "10.0.1.5",
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), task))
// RUNNING → register.
require.NoError(t, svc.recordTaskState(context.Background(), &ecsv1.TaskState{
+6
View File
@@ -107,6 +107,12 @@ func ServiceKey(cluster, serviceName string) string {
return ServicesPrefix(cluster) + serviceName
}
// LeaderLeaseKey returns the per-cluster scheduler leader-lease key in the shared
// leader bucket (Q3).
func LeaderLeaseKey(accountID, clusterName string) string {
return fmt.Sprintf("%s/%s", accountID, clusterName)
}
// CapacityProvidersPrefix returns the KV key prefix under which all
// capacity-provider records live. Account-scoped (not cluster-scoped),
// matching the AWS ARN shape; a cluster references providers by name.
+2
View File
@@ -124,6 +124,8 @@ func TestKeyPaths(t *testing.T) {
assert.Equal(t, "clusters/web/services/", ServicesPrefix("web"))
assert.Equal(t, "clusters/web/services/api", ServiceKey("web", "api"))
assert.Equal(t, "123456789012/web", LeaderLeaseKey("123456789012", "web"))
}
// Prefix helpers must be a true prefix of their per-record key so a KV
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"strings"
"sync"
"testing"
@@ -58,11 +57,11 @@ func TestCurrentBehaviour_Tagging_ECSTagWriteRevertsInterleavedRecordWrite(t *te
key := ServiceKey("web", "api")
wrapped := &interleavingKV{KeyValue: kv, key: key, between: func() {
var rec ServiceRecord
found, gerr := taskdefinition.GetJSON(ctx, kv, key, &rec)
found, gerr := getJSON(ctx, kv, key, &rec)
require.NoError(t, gerr)
require.True(t, found)
rec.DesiredCount = 5
require.NoError(t, taskdefinition.PutJSON(ctx, kv, key, &rec))
require.NoError(t, putJSON(ctx, kv, key, &rec))
}}
err = svc.mutateResourceTags(ctx, wrapped, ecsResourceService, "web", "api", func(existing map[string]string) map[string]string {
@@ -75,7 +74,7 @@ func TestCurrentBehaviour_Tagging_ECSTagWriteRevertsInterleavedRecordWrite(t *te
require.NoError(t, err)
var final ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, key, &final)
found, err := getJSON(ctx, kv, key, &final)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, 0, final.DesiredCount, "the interleaved DesiredCount=5 write is lost")
+15 -15
View File
@@ -78,7 +78,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
switch kind {
case ecsResourceCluster:
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(id), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(id), &rec)
if err != nil {
return nil, err
}
@@ -92,7 +92,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
return nil, errors.New(awserrors.ErrorECSInvalidParameter)
}
var rec taskdefinition.Record
found, err := taskdefinition.GetJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
found, err := getJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
if err != nil {
return nil, err
}
@@ -102,7 +102,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
return rec.Tags, nil
case ecsResourceService:
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, ServiceKey(cluster, id), &rec)
if err != nil {
return nil, err
}
@@ -112,7 +112,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
return rec.Tags, nil
case ecsResourceTask:
var rec TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, TaskKey(cluster, id), &rec)
if err != nil {
return nil, err
}
@@ -122,7 +122,7 @@ func (s *Service) resourceTags(ctx context.Context, kv jetstream.KeyValue, kind
return rec.Tags, nil
case ecsResourceContainerInstance:
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
if err != nil {
return nil, err
}
@@ -142,7 +142,7 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
switch kind {
case ecsResourceCluster:
var rec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(id), &rec)
found, err := getJSON(ctx, kv, ClusterMetaKey(id), &rec)
if err != nil {
return err
}
@@ -150,14 +150,14 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
return errors.New(awserrors.ErrorECSClusterNotFound)
}
rec.Tags = mutate(rec.Tags)
return taskdefinition.PutJSON(ctx, kv, ClusterMetaKey(id), &rec)
return putJSON(ctx, kv, ClusterMetaKey(id), &rec)
case ecsResourceTaskDefinition:
family, rev := taskdefinition.ParseRef(id)
if family == "" || rev == 0 {
return errors.New(awserrors.ErrorECSInvalidParameter)
}
var rec taskdefinition.Record
found, err := taskdefinition.GetJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
found, err := getJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
if err != nil {
return err
}
@@ -165,10 +165,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
return errors.New(awserrors.ErrorECSInvalidParameter)
}
rec.Tags = mutate(rec.Tags)
return taskdefinition.PutJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
return putJSON(ctx, kv, taskdefinition.RevKey(family, rev), &rec)
case ecsResourceService:
var rec ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, ServiceKey(cluster, id), &rec)
if err != nil {
return err
}
@@ -176,10 +176,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
return errors.New(awserrors.ErrorECSServiceNotFound)
}
rec.Tags = mutate(rec.Tags)
return taskdefinition.PutJSON(ctx, kv, ServiceKey(cluster, id), &rec)
return putJSON(ctx, kv, ServiceKey(cluster, id), &rec)
case ecsResourceTask:
var rec TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, TaskKey(cluster, id), &rec)
if err != nil {
return err
}
@@ -187,10 +187,10 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
return errors.New(awserrors.ErrorECSInvalidParameter)
}
rec.Tags = mutate(rec.Tags)
return taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, id), &rec)
return putJSON(ctx, kv, TaskKey(cluster, id), &rec)
case ecsResourceContainerInstance:
var rec InstanceRecord
found, err := taskdefinition.GetJSON(ctx, kv, InstanceKey(cluster, id), &rec)
found, err := getJSON(ctx, kv, InstanceKey(cluster, id), &rec)
if err != nil {
return err
}
@@ -198,7 +198,7 @@ func (s *Service) mutateResourceTags(ctx context.Context, kv jetstream.KeyValue,
return errors.New(awserrors.ErrorECSInvalidParameter)
}
rec.Tags = mutate(rec.Tags)
return taskdefinition.PutJSON(ctx, kv, InstanceKey(cluster, id), &rec)
return putJSON(ctx, kv, InstanceKey(cluster, id), &rec)
default:
return errors.New(awserrors.ErrorECSInvalidParameter)
}
+2 -3
View File
@@ -1,7 +1,6 @@
package ecs_test
import (
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"github.com/aws/aws-sdk-go/aws"
@@ -24,7 +23,7 @@ func seedTasks(t *testing.T, svc *ecsdomain.Service, cluster string, recs ...*ec
for _, r := range recs {
r.Cluster = cluster
r.ARN = ecsdomain.TaskARN(svc.Region(), ecsdomain.TestAccountID, cluster, r.TaskID)
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, ecsdomain.TaskKey(cluster, r.TaskID), r))
require.NoError(t, ecsdomain.PutJSON(t.Context(), kv, ecsdomain.TaskKey(cluster, r.TaskID), r))
}
}
@@ -125,7 +124,7 @@ func TestRecordTaskState_SelfExitMovesDesiredStatusToStopped(t *testing.T) {
kv, err := svc.Bucket(t.Context(), ecsdomain.TestAccountID)
require.NoError(t, err)
var rec ecsdomain.TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, ecsdomain.TaskKey("web", "t-1"), &rec)
found, err := ecsdomain.GetJSON(t.Context(), kv, ecsdomain.TaskKey("web", "t-1"), &rec)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, ecsdomain.TaskStatusStopped, rec.DesiredStatus)
+9 -10
View File
@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"errors"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"log/slog"
"strings"
"time"
@@ -34,7 +33,7 @@ func (s *Service) StopTask(ctx context.Context, input *ecs.StopTaskInput, accoun
return nil, err
}
var task TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &task)
found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &task)
if err != nil {
return nil, err
}
@@ -58,7 +57,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco
return nil, err
}
var clusterRec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
if err != nil {
return nil, err
}
@@ -102,7 +101,7 @@ func (s *Service) StartTask(ctx context.Context, input *ecs.StartTaskInput, acco
continue
}
}
if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
if err := putJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
return nil, err
}
if err := s.publishAssign(ctx, kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
@@ -169,12 +168,12 @@ func (s *Service) requestStopTask(ctx context.Context, kv jetstream.KeyValue, ac
return
}
task.DesiredStatus = TaskStatusStopped
if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
slog.ErrorContext(ctx, "ECS requestStopTask: persist failed", "task", task.TaskID, "err", perr)
return
}
sd := ecsv1.StopDirective{TaskID: task.TaskID, Reason: reason}
if perr := taskdefinition.PutJSON(ctx, kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
if perr := putJSON(ctx, kv, StopKey(task.Cluster, task.ContainerInstanceID, task.TaskID), &sd); perr != nil {
slog.ErrorContext(ctx, "ECS requestStopTask: post stop directive failed", "task", task.TaskID, "err", perr)
}
}
@@ -197,7 +196,7 @@ func (s *Service) forceStopTask(ctx context.Context, kv jetstream.KeyValue, acco
// cleared fields land in this single write.
s.releaseTaskPublicIP(ctx, accountID, task)
s.reclaimTaskENI(ctx, accountID, task)
if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
slog.ErrorContext(ctx, "ECS forceStopTask: persist failed", "task", task.TaskID, "err", perr)
return
}
@@ -239,7 +238,7 @@ func (s *Service) applyServiceTargets(ctx context.Context, kv jetstream.KeyValue
return
}
var svc ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
found, err := getJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
if err != nil || !found || len(svc.LoadBalancers) == 0 {
return
}
@@ -271,7 +270,7 @@ func (s *Service) assignTaskPublicIP(ctx context.Context, kv jetstream.KeyValue,
return
}
var svc ServiceRecord
found, err := taskdefinition.GetJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
found, err := getJSON(ctx, kv, ServiceKey(task.Cluster, name), &svc)
if err != nil || !found || !strings.EqualFold(svc.AssignPublicIP, "ENABLED") {
return
}
@@ -282,7 +281,7 @@ func (s *Service) assignTaskPublicIP(ctx context.Context, kv jetstream.KeyValue,
}
task.ENIPublicIP = publicIP
task.ENIEIPAllocationID = allocID
if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
if perr := putJSON(ctx, kv, TaskKey(task.Cluster, task.TaskID), task); perr != nil {
slog.ErrorContext(ctx, "ECS auto-EIP: persist public IP failed", "task", task.TaskID, "err", perr)
}
}
@@ -3,7 +3,6 @@ package ecs
import (
"context"
ecsv1 "github.com/mulgadc/spinifex/contracts/ecs/v1"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"github.com/aws/aws-sdk-go/aws"
@@ -241,7 +240,7 @@ func TestRunTask_AssignCarriesGPU(t *testing.T) {
kv, err := svc.bucket(t.Context(), testAccountID)
require.NoError(t, err)
var rec TaskRecord
found, err := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", taskID), &rec)
found, err := getJSON(t.Context(), kv, TaskKey("web", taskID), &rec)
require.NoError(t, err)
require.True(t, found)
assert.Equal(t, 1, rec.GPU)
+15 -15
View File
@@ -63,10 +63,10 @@ func (o *Owner) Register(ctx context.Context, kv jetstream.KeyValue, accountID s
RequiresCompatibilities: spec.RequiresCompatibilities,
RuntimePlatform: spec.RuntimePlatform,
}
if err := PutJSON(ctx, kv, RevKey(spec.Family, rev), rec); err != nil {
if err := putJSON(ctx, kv, RevKey(spec.Family, rev), rec); err != nil {
return nil, err
}
if err := PutJSON(ctx, kv, LatestRevKey(spec.Family), rev); err != nil {
if err := putJSON(ctx, kv, LatestRevKey(spec.Family), rev); err != nil {
return nil, err
}
return rec, nil
@@ -75,7 +75,7 @@ func (o *Owner) Register(ctx context.Context, kv jetstream.KeyValue, accountID s
// nextRevision reads the family's latest-rev and returns latest+1 (1 if absent).
func nextRevision(ctx context.Context, kv jetstream.KeyValue, family string) (int, error) {
var latest int
found, err := GetJSON(ctx, kv, LatestRevKey(family), &latest)
found, err := getJSON(ctx, kv, LatestRevKey(family), &latest)
if err != nil {
return 0, err
}
@@ -88,7 +88,7 @@ func nextRevision(ctx context.Context, kv jetstream.KeyValue, family string) (in
// Deregister marks one revision INACTIVE; the revision stays describable.
func (o *Owner) Deregister(ctx context.Context, kv jetstream.KeyValue, family string, rev int) (*Record, error) {
var rec Record
found, err := GetJSON(ctx, kv, RevKey(family, rev), &rec)
found, err := getJSON(ctx, kv, RevKey(family, rev), &rec)
if err != nil {
return nil, err
}
@@ -100,7 +100,7 @@ func (o *Owner) Deregister(ctx context.Context, kv jetstream.KeyValue, family st
rec.DeregisteredAt = &now
}
rec.Status = StatusInactive
if err := PutJSON(ctx, kv, RevKey(family, rev), &rec); err != nil {
if err := putJSON(ctx, kv, RevKey(family, rev), &rec); err != nil {
return nil, err
}
return &rec, nil
@@ -115,7 +115,7 @@ func (o *Owner) Resolve(ctx context.Context, kv jetstream.KeyValue, ref string)
}
if rev == 0 {
var latest int
found, err := GetJSON(ctx, kv, LatestRevKey(family), &latest)
found, err := getJSON(ctx, kv, LatestRevKey(family), &latest)
if err != nil {
return nil, err
}
@@ -125,7 +125,7 @@ func (o *Owner) Resolve(ctx context.Context, kv jetstream.KeyValue, ref string)
rev = latest
}
var rec Record
found, err := GetJSON(ctx, kv, RevKey(family, rev), &rec)
found, err := getJSON(ctx, kv, RevKey(family, rev), &rec)
if err != nil {
return nil, err
}
@@ -142,7 +142,7 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
if family != "" {
prefix = RevsPrefix(family)
}
keys, err := KeysWithPrefix(ctx, kv, prefix)
keys, err := keysWithPrefix(ctx, kv, prefix)
if err != nil {
return nil, err
}
@@ -152,7 +152,7 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
continue
}
var rec Record
found, err := GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
@@ -163,8 +163,8 @@ func (o *Owner) List(ctx context.Context, kv jetstream.KeyValue, family, status
return arns, nil
}
// GetJSON reads key into out. Returns (false, nil) when the key is absent.
func GetJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
// getJSON reads key into out. Returns (false, nil) when the key is absent.
func getJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (bool, error) {
entry, err := kv.Get(ctx, key)
if errors.Is(err, jetstream.ErrKeyNotFound) {
return false, nil
@@ -178,8 +178,8 @@ func GetJSON(ctx context.Context, kv jetstream.KeyValue, key string, out any) (b
return true, nil
}
// PutJSON marshals v and writes it at key.
func PutJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
// putJSON marshals v and writes it at key.
func putJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) error {
data, err := json.Marshal(v)
if err != nil {
return err
@@ -190,8 +190,8 @@ func PutJSON(ctx context.Context, kv jetstream.KeyValue, key string, v any) erro
return nil
}
// KeysWithPrefix returns all KV keys under prefix, sorted.
func KeysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
// keysWithPrefix returns all KV keys under prefix, sorted.
func keysWithPrefix(ctx context.Context, kv jetstream.KeyValue, prefix string) ([]string, error) {
keys, err := kv.Keys(ctx)
if errors.Is(err, jetstream.ErrNoKeysFound) {
return nil, nil
+8 -8
View File
@@ -34,7 +34,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI
}
var clusterRec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
if err != nil {
return nil, err
}
@@ -82,7 +82,7 @@ func (s *Service) RunTask(ctx context.Context, input *ecs.RunTaskInput, accountI
}
}
if err := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
if err := putJSON(ctx, kv, TaskKey(cluster, taskID), rec); err != nil {
return nil, err
}
if err := s.publishAssign(ctx, kv, accountID, cluster, inst.InstanceID, rec, taskDef); err != nil {
@@ -166,7 +166,7 @@ func (s *Service) provisionTaskENI(ctx context.Context, kv jetstream.KeyValue, a
rec.DesiredStatus = TaskStatusStopped
rec.StoppedReason = fmt.Sprintf("ENI attach failed: %s", err)
rec.StoppedAt = time.Now().UTC()
if perr := taskdefinition.PutJSON(ctx, kv, TaskKey(cluster, rec.TaskID), rec); perr != nil {
if perr := putJSON(ctx, kv, TaskKey(cluster, rec.TaskID), rec); perr != nil {
slog.ErrorContext(ctx, "ECS RunTask: persist ENI rollback failed", "task", rec.TaskID, "err", perr)
}
}
@@ -271,7 +271,7 @@ func (s *Service) publishAssign(ctx context.Context, kv jetstream.KeyValue, acco
for _, c := range td.Containers {
msg.Containers = append(msg.Containers, containerToAssign(c))
}
return taskdefinition.PutJSON(ctx, kv, AssignmentKey(cluster, instanceID, rec.TaskID), &msg)
return putJSON(ctx, kv, AssignmentKey(cluster, instanceID, rec.TaskID), &msg)
}
// DescribeTasks returns task records for the named tasks in a cluster.
@@ -285,7 +285,7 @@ func (s *Service) DescribeTasks(ctx context.Context, input *ecs.DescribeTasksInp
for _, ref := range awsStringSlice(input.Tasks) {
taskID := ContainerInstanceShortID(ref)
var rec TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, TaskKey(cluster, taskID), &rec)
found, err := getJSON(ctx, kv, TaskKey(cluster, taskID), &rec)
if err != nil {
return nil, err
}
@@ -308,14 +308,14 @@ func (s *Service) ListTasks(ctx context.Context, input *ecs.ListTasksInput, acco
return nil, err
}
var clusterRec ClusterRecord
found, err := taskdefinition.GetJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
found, err := getJSON(ctx, kv, ClusterMetaKey(cluster), &clusterRec)
if err != nil {
return nil, err
}
if !found {
return nil, errors.New(awserrors.ErrorECSClusterNotFound)
}
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, TasksPrefix(cluster))
keys, err := keysWithPrefix(ctx, kv, TasksPrefix(cluster))
if err != nil {
return nil, err
}
@@ -323,7 +323,7 @@ func (s *Service) ListTasks(ctx context.Context, input *ecs.ListTasksInput, acco
arns := make([]string, 0, len(keys))
for _, k := range keys {
var rec TaskRecord
found, err := taskdefinition.GetJSON(ctx, kv, k, &rec)
found, err := getJSON(ctx, kv, k, &rec)
if err != nil {
return nil, err
}
+3 -4
View File
@@ -2,7 +2,6 @@ package ecs
import (
"context"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"log/slog"
"strings"
"time"
@@ -53,7 +52,7 @@ func (sc *Scheduler) sweepStoppedTasks(ctx context.Context) (time.Duration, erro
// the ENI release for any that still owe one. A missing StoppedAt or an owed
// (unreleased) ENI both block deletion; the latter until the release succeeds.
func (s *Service) sweepStoppedBucket(ctx context.Context, kv jetstream.KeyValue, accountID string, now time.Time, retention time.Duration) (int, time.Duration, error) {
keys, err := taskdefinition.KeysWithPrefix(ctx, kv, "clusters/")
keys, err := keysWithPrefix(ctx, kv, "clusters/")
if err != nil {
return 0, 0, err
}
@@ -65,7 +64,7 @@ func (s *Service) sweepStoppedBucket(ctx context.Context, kv jetstream.KeyValue,
continue
}
var task TaskRecord
found, gerr := taskdefinition.GetJSON(ctx, kv, k, &task)
found, gerr := getJSON(ctx, kv, k, &task)
if gerr != nil || !found {
continue
}
@@ -117,7 +116,7 @@ func (s *Service) retryTaskENIRelease(ctx context.Context, kv jetstream.KeyValue
task.ENIReleaseAttempts++
task.ENIReleaseNextTry = now.Add(eniReleaseBackoff(task.ENIReleaseAttempts))
}
if perr := taskdefinition.PutJSON(ctx, kv, key, task); perr != nil {
if perr := putJSON(ctx, kv, key, task); perr != nil {
slog.Warn("ECS sweep: persist ENI retry state failed", "key", key, "err", perr)
}
}
+12 -13
View File
@@ -3,7 +3,6 @@ package ecs
import (
"errors"
"fmt"
"github.com/mulgadc/spinifex/spinifex/domains/ecs/taskdefinition"
"testing"
"time"
@@ -20,7 +19,7 @@ func TestSweepStoppedBucket_PrunesOnlyStale(t *testing.T) {
TaskID: id, Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", id),
LastStatus: status, DesiredStatus: status, StoppedAt: stoppedAt,
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
}
put("stale", TaskStatusStopped, now.Add(-2*time.Hour)) // older than retention -> prune
@@ -37,7 +36,7 @@ func TestSweepStoppedBucket_PrunesOnlyStale(t *testing.T) {
exists := func(id string) bool {
var rec TaskRecord
found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", id), &rec)
found, gerr := getJSON(t.Context(), kv, TaskKey("web", id), &rec)
require.NoError(t, gerr)
return found
}
@@ -59,7 +58,7 @@ func TestSweepStoppedBucket_RetriesOwedENI_Succeeds(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-1", ENIAttachmentID: "att-1",
StoppedAt: now.Add(-2 * time.Hour), // well past retention
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
require.NoError(t, err)
@@ -68,7 +67,7 @@ func TestSweepStoppedBucket_RetriesOwedENI_Succeeds(t *testing.T) {
assert.Equal(t, 1, eni.releaseCalls)
var got TaskRecord
found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
found, gerr := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
require.NoError(t, gerr)
require.True(t, found)
assert.Equal(t, "eni-1", got.ENIID, "the identity survives release as the forensic record")
@@ -89,7 +88,7 @@ func TestSweepStoppedBucket_RetryFails_BacksOff(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-1",
StoppedAt: now.Add(-2 * time.Hour),
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
require.NoError(t, err)
@@ -97,7 +96,7 @@ func TestSweepStoppedBucket_RetryFails_BacksOff(t *testing.T) {
assert.Equal(t, eniReleaseBackoffBase, due)
var got TaskRecord
found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
found, gerr := getJSON(t.Context(), kv, TaskKey("web", "t-1"), &got)
require.NoError(t, gerr)
require.True(t, found)
assert.Equal(t, "eni-1", got.ENIID, "a failed retry is still owed")
@@ -119,7 +118,7 @@ func TestSweepStoppedBucket_RetryNotDueYet_Skipped(t *testing.T) {
StoppedAt: now.Add(-2 * time.Hour),
ENIReleaseNextTry: now.Add(10 * time.Minute),
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "t-1"), &rec))
pruned, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
require.NoError(t, err)
@@ -143,7 +142,7 @@ func TestSweepStoppedBucket_HonoursPerPassCap(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-" + id,
StoppedAt: now.Add(-2 * time.Hour),
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
}
_, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
@@ -174,7 +173,7 @@ func TestSweepStoppedBucket_CappedOverflow_AsksForSweepInterval(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-" + id,
StoppedAt: now.Add(-2 * time.Hour),
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", id), &rec))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", id), &rec))
}
_, due, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
@@ -197,7 +196,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-released", ENIReleased: true,
StoppedAt: now.Add(-2 * time.Hour), // well past retention
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "released"), &released))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "released"), &released))
owed := TaskRecord{
TaskID: "owed", Cluster: "web", ARN: TaskARN(testRegion, testAccountID, "web", "owed"),
@@ -205,7 +204,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
NetworkMode: NetworkModeAwsvpc, ENIID: "eni-owed", ENIReleased: false,
StoppedAt: now.Add(-2 * time.Hour), // also well past retention, but still owed
}
require.NoError(t, taskdefinition.PutJSON(t.Context(), kv, TaskKey("web", "owed"), &owed))
require.NoError(t, putJSON(t.Context(), kv, TaskKey("web", "owed"), &owed))
pruned, _, err := svc.sweepStoppedBucket(t.Context(), kv, testAccountID, now, time.Hour)
require.NoError(t, err)
@@ -213,7 +212,7 @@ func TestSweepStoppedBucket_PrunesReleased_KeepsOwed(t *testing.T) {
exists := func(id string) bool {
var rec TaskRecord
found, gerr := taskdefinition.GetJSON(t.Context(), kv, TaskKey("web", id), &rec)
found, gerr := getJSON(t.Context(), kv, TaskKey("web", id), &rec)
require.NoError(t, gerr)
return found
}
+14 -1
View File
@@ -3,9 +3,11 @@ package awsapi
import (
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"github.com/mulgadc/bluebottle/pkg/sigv4"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
@@ -62,6 +64,17 @@ func CallerFromActor(actor dispatch.Actor) Caller {
}
}
func readBoundedBody(r *http.Request) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidParameterValue)
}
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return body, nil
}
// dispatch runs the action and renames the generic parameter error to the code
// EKS declares. The internal routes keep theirs: the CP-VM agents match on it.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
@@ -126,7 +139,7 @@ func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
return errors.New(awserrors.ErrorInternalError)
}
body, err := dispatch.ReadBoundedBody(r)
body, err := readBoundedBody(r)
if err != nil {
slog.ErrorContext(r.Context(), "EKS dispatch: failed to read body", "err", err)
return err
@@ -795,7 +795,8 @@ func retryableStartRefusal(err error) bool {
if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorInsufficientInstanceCapacity {
return true
}
return errors.Is(err, ec2v1.ErrVolumeHeldElsewhere)
_, message, _ := awserrors.ResolveErrorDetail(err)
return strings.Contains(message, ec2v1.VolumeHeldElsewhereText) || strings.Contains(err.Error(), ec2v1.VolumeHeldElsewhereText)
}
// maybeReformEtcdQuorum escalates a wedged HA control plane that the in-place
@@ -3,11 +3,9 @@ package cluster
import (
"context"
"errors"
"fmt"
"testing"
"time"
"github.com/mulgadc/spinifex/contracts/ec2/v1"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -32,12 +30,12 @@ func (c *refusingCPControl) StartInstance(context.Context, string) error {
func (c *refusingCPControl) StopInstance(context.Context, string) error { return nil }
// What the start path returns after crossing NATS: StartInstances wraps the
// sentinel back on when the owner's reply carries the held-elsewhere header.
// What the start path returns after crossing NATS: the code and the message the
// owning node sent, with no sentinel left to match on.
var (
errStartCapacity = errors.New(awserrors.ErrorInsufficientInstanceCapacity)
errStartHeld = fmt.Errorf("%w: %w", ec2v1.ErrVolumeHeldElsewhere,
awserrors.Errorf(awserrors.ErrorIncorrectState, "%s", "volume vol-1 is published to node-a"))
errStartHeld = awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
"volume is leased by another owner: held by node-a since 2026-10-30T01:00:00Z")
)
// A refusal that clears by itself must not use up the restart budget, or the
@@ -74,30 +72,18 @@ func TestMaybeRecoverCP_RetryableStartRefusalIsRetriedNotReplaced(t *testing.T)
}
}
// The cap still bounds the faults nothing says will clear. IncorrectState
// without the held-elsewhere sentinel is one, whatever its message says.
// The cap still bounds the faults nothing says will clear.
func TestMaybeRecoverCP_UnknownStartFailureStillCountsAgainstTheBudget(t *testing.T) {
t.Parallel()
for name, refusal := range map[string]error{
"server internal": errors.New(awserrors.ErrorServerInternal),
"unrelated incorrect state": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
"instance i-cp is not in a state from which it can be started"),
"lease text without the sentinel": awserrors.Errorf(awserrors.ErrorIncorrectState, "%s",
ec2v1.VolumeHeldElsewhereText+": held by node-a since 2026-10-30T01:00:00Z"),
} {
t.Run(name, func(t *testing.T) {
t.Parallel()
cp := &refusingCPControl{err: refusal}
r := newRecoveryReconciler(t, cp)
r.degradedSince = time.Now().Add(-10 * time.Minute)
cp := &refusingCPControl{err: errors.New(awserrors.ErrorServerInternal)}
r := newRecoveryReconciler(t, cp)
r.degradedSince = time.Now().Add(-10 * time.Minute)
for range r.maxRestartAttempts + 2 {
r.lastRestartAt = time.Time{}
r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
}
require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
})
for range r.maxRestartAttempts + 2 {
r.lastRestartAt = time.Time{}
r.maybeRecoverControlPlane(t.Context(), metaWithCP("i-cp"), "stale")
}
require.Equal(t, r.maxRestartAttempts, cp.startCalls, "stops at the attempt cap")
assert.Equal(t, r.maxRestartAttempts, r.restartAttempts)
}
+29 -2
View File
@@ -1,14 +1,17 @@
package awsapi
import (
"bytes"
"context"
"errors"
"io"
"log/slog"
"maps"
"net/http"
"slices"
"github.com/aws/aws-sdk-go/service/elbv2"
"github.com/mulgadc/bluebottle/pkg/sigv4"
elbv2domain "github.com/mulgadc/spinifex/spinifex/domains/elbv2"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
awsxml "github.com/mulgadc/spinifex/spinifex/foundation/aws/xml"
@@ -214,7 +217,7 @@ func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
InventoryName: InventoryName,
ResolveAction: dispatch.ResolveQueryAction,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeXML,
Inventory: OperationInventory(),
@@ -232,11 +235,35 @@ func OperationInventory() dispatch.Inventory {
return dispatch.Inventory{Registered: ActionNames()}
}
// resolveAction is the one Query Action parser telemetry and dispatch share.
// It leaves the body readable for the dispatcher and answers "" on any failure.
func resolveAction(r *http.Request) string {
args, err := readQueryArgs(r)
if err != nil {
return ""
}
return args["Action"]
}
// readQueryArgs parses the Query body and rewinds it, so the same bytes can be
// read again. Any read or parse failure is a malformed query string.
func readQueryArgs(r *http.Request) (map[string]string, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, err
}
r.Body = io.NopCloser(bytes.NewReader(body))
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return query.ParseArgs(string(body))
}
// dispatch parses, scopes and authorizes the request from one parse of its
// body, then runs the action. Errors are awserrors codes the gateway renders.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
queryArgs, err := dispatch.ReadArgs(r)
queryArgs, err := readQueryArgs(r)
if err != nil {
slog.DebugContext(r.Context(), "ELBv2: malformed query string", "err", err)
return errors.New(awserrors.ErrorMalformedQueryString)
@@ -3,6 +3,10 @@ package awsapi
import (
"context"
"encoding/xml"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/aws/aws-sdk-go/service/elbv2"
@@ -27,6 +31,22 @@ func TestNewRegistration_RegistersUnderCredentialScopeAndInventoryName(t *testin
assert.Empty(t, inv.Unsupported)
}
func TestResolveAction_ReadsActionAndRewindsBody(t *testing.T) {
body := "Action=DescribeLoadBalancers&Version=2015-12-01"
r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
assert.Equal(t, "DescribeLoadBalancers", resolveAction(r))
again, err := io.ReadAll(r.Body)
require.NoError(t, err)
assert.Equal(t, body, string(again), "the dispatcher must still see the whole body")
}
func TestResolveAction_UnparseableBodyIsEmpty(t *testing.T) {
r := httptest.NewRequest(http.MethodPost, "/", strings.NewReader("Action=%zz"))
assert.Empty(t, resolveAction(r))
}
// A registration with no quota reports the AWS default cap and enforces none.
func TestDescribeAccountLimits_NoQuotaReportsDefault(t *testing.T) {
out, err := actions["DescribeAccountLimits"].dispatch(t.Context(), "DescribeAccountLimits",
+61 -6
View File
@@ -3293,12 +3293,66 @@ func (s *ELBv2ServiceImpl) DescribeTags(ctx context.Context, input *elbv2.Descri
}
arn := *arnPtr
h, notFoundError, found, err := s.resolveTaggable(ctx, arn)
resourceType, err := elbv2ResourceTypeFromArn(arn)
if err != nil {
return nil, err
}
if !found {
if lArn, isDefault := listenerArnFromDefaultRuleArn(arn); isDefault {
var (
tags map[string]string
ownerAccount string
notFoundError string
found bool
)
switch resourceType {
case resourcearn.ELBv2LoadBalancer:
notFoundError = awserrors.ErrorELBv2LoadBalancerNotFound
lb, lbErr := s.store.GetLoadBalancerByArn(ctx, arn)
if lbErr != nil {
slog.ErrorContext(ctx, "DescribeTags: failed to get LB", "arn", arn, "err", lbErr)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if lb != nil {
found = true
tags = lb.Tags
ownerAccount = lb.AccountID
}
case resourcearn.ELBv2TargetGroup:
notFoundError = awserrors.ErrorELBv2TargetGroupNotFound
tg, tgErr := s.store.GetTargetGroupByArn(ctx, arn)
if tgErr != nil {
slog.ErrorContext(ctx, "DescribeTags: failed to get target group", "arn", arn, "err", tgErr)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if tg != nil {
found = true
tags = tg.Tags
ownerAccount = tg.AccountID
}
case resourcearn.ELBv2Listener:
notFoundError = awserrors.ErrorELBv2ListenerNotFound
l, lErr := s.store.GetListenerByArn(ctx, arn)
if lErr != nil {
slog.ErrorContext(ctx, "DescribeTags: failed to get listener", "arn", arn, "err", lErr)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if l != nil {
found = true
tags = l.Tags
ownerAccount = l.AccountID
}
case resourcearn.ELBv2ListenerRule:
notFoundError = awserrors.ErrorELBv2RuleNotFound
r, rErr := s.store.GetRuleByArn(ctx, arn)
if rErr != nil {
slog.ErrorContext(ctx, "DescribeTags: failed to get rule", "arn", arn, "err", rErr)
return nil, errors.New(awserrors.ErrorServerInternal)
}
if r != nil {
found = true
tags = r.Tags
ownerAccount = r.AccountID
} else if lArn, isDefault := listenerArnFromDefaultRuleArn(arn); isDefault {
// Synthetic default rule: not stored, carries no tags. Resolve via
// its parent listener so a controller's post-create rule-tag sync
// gets an empty TagDescription instead of an error.
@@ -3309,18 +3363,19 @@ func (s *ELBv2ServiceImpl) DescribeTags(ctx context.Context, input *elbv2.Descri
}
if l != nil {
found = true
h = taggable{owner: l.AccountID}
tags = nil
ownerAccount = l.AccountID
}
}
}
if !found || h.owner != accountID {
if !found || ownerAccount != accountID {
return nil, errors.New(notFoundError)
}
descriptions = append(descriptions, &elbv2.TagDescription{
ResourceArn: aws.String(arn),
Tags: tagsMapToSDK(h.tags),
Tags: tagsMapToSDK(tags),
})
}
+16
View File
@@ -578,6 +578,22 @@ func (s *Store) PutRule(ctx context.Context, r *RuleRecord) error {
return err
}
// GetRule retrieves a rule by its short ID.
func (s *Store) GetRule(ctx context.Context, ruleID string) (*RuleRecord, error) {
entry, err := s.kv.Get(ctx, KeyPrefixRule+ruleID)
if err != nil {
if errors.Is(err, jetstream.ErrKeyNotFound) {
return nil, nil
}
return nil, err
}
var r RuleRecord
if err := json.Unmarshal(entry.Value(), &r); err != nil {
return nil, fmt.Errorf("unmarshal rule: %w", err)
}
return &r, nil
}
// DeleteRule removes a rule by its short ID.
func (s *Store) DeleteRule(ctx context.Context, ruleID string) error {
err := s.kv.Delete(ctx, KeyPrefixRule+ruleID)
+27 -2
View File
@@ -1,8 +1,10 @@
package awsapi
import (
"bytes"
"context"
"errors"
"io"
"log/slog"
"maps"
"net/http"
@@ -11,6 +13,7 @@ import (
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/mulgadc/bluebottle/pkg/iampolicy"
"github.com/mulgadc/bluebottle/pkg/sigv4"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
@@ -373,7 +376,7 @@ var getRootUser = handler(func(accountID string, _ *iam.GetUserInput, d Deps) (a
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: dispatch.ResolveQueryAction,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeXML,
Traits: dispatch.ServiceTraits{XMLThrottleCode: awserrors.ErrorThrottling},
@@ -392,11 +395,33 @@ func OperationInventory() dispatch.Inventory {
return dispatch.Inventory{Registered: ActionNames()}
}
// resolveAction is the one Query Action parser telemetry and dispatch share.
// It leaves the body readable for the dispatcher and answers "" on any failure.
func resolveAction(r *http.Request) string {
args, err := readQueryArgs(r)
if err != nil {
return ""
}
return args["Action"]
}
func readQueryArgs(r *http.Request) (map[string]string, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, sigv4.MaxPayloadLen+1))
if err != nil {
return nil, err
}
r.Body = io.NopCloser(bytes.NewReader(body))
if int64(len(body)) > sigv4.MaxPayloadLen {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return query.ParseArgs(string(body))
}
// dispatch parses, scopes and authorizes the request from one parse of its
// body, then runs the action. Errors are awserrors codes the gateway renders.
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
queryArgs, err := dispatch.ReadArgs(r)
queryArgs, err := readQueryArgs(r)
if err != nil {
slog.DebugContext(r.Context(), "IAM: malformed query string", "err", err)
return errors.New(awserrors.ErrorMalformedQueryString)
-33
View File
@@ -1,33 +0,0 @@
// Package backoff holds the jitter and context-aware sleep shared by the
// external-network allocators' retry and poll loops.
package backoff
import (
"context"
"math/rand/v2"
"time"
)
// Jitter returns a uniform offset in [-span, span), or 0 when span <= 0.
func Jitter(span time.Duration) time.Duration {
if span <= 0 {
return 0
}
return time.Duration(rand.Int64N(int64(2*span))) - span //nolint:gosec // jitter, not a secret
}
// Sleep waits for d or until ctx is done, returning ctx.Err() in the latter
// case. A non-positive d still reports an already-cancelled ctx.
func Sleep(ctx context.Context, d time.Duration) error {
if d <= 0 {
return ctx.Err()
}
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-t.C:
return nil
}
}
@@ -1,30 +0,0 @@
package backoff
import (
"context"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestJitter_StaysWithinSpan(t *testing.T) {
assert.Zero(t, Jitter(0))
assert.Zero(t, Jitter(-time.Second))
for range 1000 {
j := Jitter(10 * time.Millisecond)
require.GreaterOrEqual(t, j, -10*time.Millisecond)
require.Less(t, j, 10*time.Millisecond)
}
}
func TestSleep_ReportsCancellation(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
cancel()
require.ErrorIs(t, Sleep(ctx, 0), context.Canceled, "a zero wait must still see a cancelled ctx")
require.ErrorIs(t, Sleep(ctx, time.Hour), context.Canceled)
require.NoError(t, Sleep(context.Background(), 0))
require.NoError(t, Sleep(context.Background(), time.Millisecond))
}
+29 -2
View File
@@ -232,8 +232,35 @@ func fromWireEntry(w wireEntry) (Entry, error) {
RawOffer: w.RawOffer,
RawACK: w.RawACK,
}
if err := decodeLeaseAddrs(lease, w.IP, w.SubnetMask, w.Routers, w.DNS, w.ServerID, w.HWAddr); err != nil {
return Entry{}, err
if w.IP != "" {
lease.IP = net.ParseIP(w.IP)
}
if w.SubnetMask != "" {
mask, err := parseMask(w.SubnetMask)
if err != nil {
return Entry{}, fmt.Errorf("parse subnet_mask %q: %w", w.SubnetMask, err)
}
lease.SubnetMask = mask
}
for _, s := range w.Routers {
if ip := net.ParseIP(s); ip != nil {
lease.Routers = append(lease.Routers, ip)
}
}
for _, s := range w.DNS {
if ip := net.ParseIP(s); ip != nil {
lease.DNS = append(lease.DNS, ip)
}
}
if w.ServerID != "" {
lease.ServerID = net.ParseIP(w.ServerID)
}
if w.HWAddr != "" {
hw, err := net.ParseMAC(w.HWAddr)
if err != nil {
return Entry{}, fmt.Errorf("parse hwaddr %q: %w", w.HWAddr, err)
}
lease.HWAddr = hw
}
if !w.Expiry.IsZero() {
lease.LeaseDuration = w.Expiry.Sub(w.Acquired)
@@ -1,59 +0,0 @@
package dhcp
import (
"net"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestWireAndKVDecodeSameLeaseAddrs(t *testing.T) {
wl := &wireLease{
IP: "10.0.0.5",
SubnetMask: "255.255.255.0",
Routers: []string{"10.0.0.1", "bogus"},
DNS: []string{"1.1.1.1", "8.8.8.8"},
ServerID: "10.0.0.2",
HWAddr: "02:00:00:00:00:01",
}
we := wireEntry{
ClientID: "c",
IP: wl.IP,
SubnetMask: wl.SubnetMask,
Routers: wl.Routers,
DNS: wl.DNS,
ServerID: wl.ServerID,
HWAddr: wl.HWAddr,
}
fromWire, err := fromWireLease(wl)
require.NoError(t, err)
fromKV, err := fromWireEntry(we)
require.NoError(t, err)
for _, l := range []*Lease{fromWire, fromKV.Lease} {
assert.True(t, l.IP.Equal(net.ParseIP("10.0.0.5")))
assert.Equal(t, net.CIDRMask(24, 32), l.SubnetMask)
require.Len(t, l.Routers, 1, "unparseable router is dropped")
assert.True(t, l.Routers[0].Equal(net.ParseIP("10.0.0.1")))
assert.Len(t, l.DNS, 2)
assert.True(t, l.ServerID.Equal(net.ParseIP("10.0.0.2")))
assert.Equal(t, "02:00:00:00:00:01", l.HWAddr.String())
}
}
func TestWireAndKVRejectSameBadLeaseAddrs(t *testing.T) {
cases := map[string]struct{ mask, hw, want string }{
"bad mask": {mask: "not-a-mask", want: `parse subnet_mask "not-a-mask"`},
"bad hwaddr": {hw: "zz", want: `parse hwaddr "zz"`},
}
for name, tc := range cases {
t.Run(name, func(t *testing.T) {
_, err := fromWireLease(&wireLease{SubnetMask: tc.mask, HWAddr: tc.hw})
require.ErrorContains(t, err, tc.want)
_, err = fromWireEntry(wireEntry{SubnetMask: tc.mask, HWAddr: tc.hw})
require.ErrorContains(t, err, tc.want)
})
}
}
+31 -5
View File
@@ -6,12 +6,12 @@ import (
"errors"
"fmt"
"log/slog"
"math/rand/v2"
"net"
"os"
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/domains/network/external/backoff"
"github.com/mulgadc/spinifex/spinifex/foundation/telemetry"
"github.com/nats-io/nats.go"
@@ -395,7 +395,7 @@ func (m *Manager) run(ctx context.Context, self *leaseLoop, e Entry, reaffirm bo
return
}
slog.Warn("dhcp manager: re-DORA after expiry failed; backing off", "client_id", e.Lease.ClientID, "err", err)
if backoff.Sleep(ctx, postExpiryBackoff) != nil {
if !sleepWithCtx(ctx, postExpiryBackoff) {
return
}
}
@@ -515,7 +515,7 @@ func (m *Manager) acquireWithBackoff(ctx context.Context, purpose string, req Ac
if remaining <= 0 {
break
}
attempt := base + backoff.Jitter(acquireAttemptJitter)
attempt := base + jitter(acquireAttemptJitter)
if attempt <= 0 {
attempt = base
}
@@ -882,12 +882,38 @@ func decodeHWAddr(s string) (net.HardwareAddr, error) {
func sleepUntil(ctx context.Context, now func() time.Time, deadline time.Time) bool {
d := deadline.Sub(now())
if d > 0 {
d += backoff.Jitter(renewJitter)
d += jitter(renewJitter)
if d < 0 {
d = 0
}
} else {
d = 0
}
return backoff.Sleep(ctx, d) == nil
return sleepWithCtx(ctx, d)
}
func sleepWithCtx(ctx context.Context, d time.Duration) bool {
if d <= 0 {
select {
case <-ctx.Done():
return false
default:
return true
}
}
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return false
case <-t.C:
return true
}
}
func jitter(span time.Duration) time.Duration {
if span <= 0 {
return 0
}
return time.Duration(rand.Int64N(int64(span*2))) - span //nolint:gosec // jitter, not cryptographic
}
+3 -4
View File
@@ -435,7 +435,7 @@ func TestNATSClientAcquireReleaseRoundtrip(t *testing.T) {
assert.Equal(t, "eip", got.Purpose)
assert.Equal(t, "default", got.PoolName)
require.NoError(t, client.RequestReleaseByIP(context.Background(), "default", lease.IP.String(), false))
require.NoError(t, client.RequestRelease(context.Background(), "eipalloc-A"))
_, err = store.Get(t.Context(), "eipalloc-A")
assert.ErrorIs(t, err, jetstream.ErrKeyNotFound)
assert.Equal(t, 1, fake.ReleaseCount())
@@ -676,9 +676,8 @@ func TestManagerReleaseUnknownClientIsNoop(t *testing.T) {
_ = s.Unsubscribe()
}
})
msg, err := nc.Request(dhcp.TopicRelease, []byte(`{"client_id":"never-existed"}`), time.Second)
require.NoError(t, err)
assert.JSONEq(t, `{}`, string(msg.Data))
client := dhcp.NewNATSClient(nc, time.Second)
assert.NoError(t, client.RequestRelease(context.Background(), "never-existed"))
assert.Equal(t, 0, fake.ReleaseCount())
}
+9
View File
@@ -89,6 +89,15 @@ func (c *NATSClient) RequestAcquire(ctx context.Context, p AcquireParams) (*Leas
return fromWireLease(reply.Lease)
}
// RequestRelease issues an idempotent release RPC. Unknown client-ids
// return nil (vpcd treats them as already-released).
func (c *NATSClient) RequestRelease(ctx context.Context, clientID string) error {
if clientID == "" {
return errors.New("dhcp NATSClient: clientID required")
}
return c.requestRelease(ctx, releaseWireRequest{ClientID: clientID})
}
// RequestReleaseByIP releases the lease that holds (poolName, ip).
// Used by DHCPPoolAllocator.Release, whose external.Allocator signature
// only carries the IP — vpcd looks up the client-id from the lease store.
+18 -27
View File
@@ -199,44 +199,35 @@ func fromWireLease(w *wireLease) (*Lease, error) {
RawOffer: w.RawOffer,
RawACK: w.RawACK,
}
if err := decodeLeaseAddrs(l, w.IP, w.SubnetMask, w.Routers, w.DNS, w.ServerID, w.HWAddr); err != nil {
return nil, err
if w.IP != "" {
l.IP = net.ParseIP(w.IP)
}
return l, nil
}
// decodeLeaseAddrs parses the address fields shared by the wire and KV lease
// formats into l. Unparseable routers and DNS servers are dropped.
func decodeLeaseAddrs(l *Lease, ip, mask string, routers, dns []string, serverID, hw string) error {
if ip != "" {
l.IP = net.ParseIP(ip)
}
if mask != "" {
m, err := parseMask(mask)
if w.SubnetMask != "" {
m, err := parseMask(w.SubnetMask)
if err != nil {
return fmt.Errorf("parse subnet_mask %q: %w", mask, err)
return nil, fmt.Errorf("subnet_mask: %w", err)
}
l.SubnetMask = m
}
for _, s := range routers {
if r := net.ParseIP(s); r != nil {
l.Routers = append(l.Routers, r)
for _, s := range w.Routers {
if ip := net.ParseIP(s); ip != nil {
l.Routers = append(l.Routers, ip)
}
}
for _, s := range dns {
if d := net.ParseIP(s); d != nil {
l.DNS = append(l.DNS, d)
for _, s := range w.DNS {
if ip := net.ParseIP(s); ip != nil {
l.DNS = append(l.DNS, ip)
}
}
if serverID != "" {
l.ServerID = net.ParseIP(serverID)
if w.ServerID != "" {
l.ServerID = net.ParseIP(w.ServerID)
}
if hw != "" {
mac, err := net.ParseMAC(hw)
if w.HWAddr != "" {
hw, err := net.ParseMAC(w.HWAddr)
if err != nil {
return fmt.Errorf("parse hwaddr %q: %w", hw, err)
return nil, fmt.Errorf("hwaddr: %w", err)
}
l.HWAddr = mac
l.HWAddr = hw
}
return nil
return l, nil
}
+26 -3
View File
@@ -27,12 +27,12 @@ import (
"errors"
"fmt"
"log/slog"
"math/rand/v2"
"net/netip"
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/domains/network/external"
"github.com/mulgadc/spinifex/spinifex/domains/network/external/backoff"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvstore"
"github.com/mulgadc/spinifex/spinifex/providers/cloud/oci"
@@ -203,7 +203,7 @@ func New(client oci.Client, store Store, cfg Config) (*PoolAllocator, error) {
a.budget = assignBudget
}
if a.sleep == nil {
a.sleep = backoff.Sleep
a.sleep = sleepCtx
}
if a.now == nil {
a.now = time.Now
@@ -410,7 +410,7 @@ func (a *PoolAllocator) waitAssigned(ctx context.Context, pub oci.PublicIP) (oci
deadline := time.Now().Add(a.budget)
for i := 0; ; i++ {
wait := a.schedule[min(i, len(a.schedule)-1)]
if j := backoff.Jitter(assignJitter); wait+j > 0 {
if j := jitter(assignJitter); wait+j > 0 {
wait += j
}
if time.Now().Add(wait).After(deadline) {
@@ -501,3 +501,26 @@ func displayName(req external.AllocateRequest) string {
return DisplayNamePrefix + "unclaimed"
}
}
func jitter(span time.Duration) time.Duration {
if span <= 0 {
return 0
}
// Spreading the poll ladder, not generating a secret: math/rand is the
// right tool and crypto/rand would be cargo cult here.
return time.Duration(rand.Int64N(int64(2*span))) - span //nolint:gosec // jitter, not a secret
}
func sleepCtx(ctx context.Context, d time.Duration) error {
if d <= 0 {
return nil
}
t := time.NewTimer(d)
defer t.Stop()
select {
case <-ctx.Done():
return ctx.Err()
case <-t.C:
return nil
}
}

Some files were not shown because too many files have changed in this diff Show More