mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
feat: DDIL E2E test harness (mulga-siv-5) (#160)
* feat: scaffold DDIL E2E harness (cluster, SSH, daemon client) Phase 1 of the DDIL E2E test harness (mulga-siv-5.1). Adds the tests/e2e/ddil/ package tree with the three harness primitives required by every sibling scenario/fault bead: - harness/cluster.go — Cluster, Node, ClusterFromEnv, NodeFromEnv - harness/ssh.go — typed SSH transport backed by golang.org/x/crypto/ssh, interface so dry-run and tests can stub it - harness/daemon_client.go — thin HTTPS wrapper over /local/status, /local/instances (placeholder types until daemon-local-autonomy lands), plus /health for scaffolding validation Also adds doc.go and README.md (run instructions + env-var contract). All harness files carry //go:build e2e so default builds skip them. Plan: docs/development/improvements/ddil-e2e-test-harness.md §0, §1. * feat: add DDIL harness fault injection and link profiles Phase 2 of the DDIL E2E test harness (mulga-siv-5.2). Adds the fault injection and netem primitives consumed by Scenarios A–F: - harness/fault.go — PartitionNode (per-peer iptables DROP with orchestrator-SSH sanity check), HealNode, KillNATS, StartNATS, RestartDaemonOnly, WaitForMode (polls /local/status for typed DaemonMode enum) - harness/netem.go — ApplyNetem (tc qdisc replace, idempotent), ClearNetem (idempotent clear) - harness/profiles.go — LinkProfile struct plus LAN, WAN, LTEDegraded, SATCOM, HFData, Flapping constants (unvalidated approximations per plan §2 table; validation tracking lives in TEST_COVERAGE.md) All files carry //go:build e2e. Flapping profile is defined but ApplyNetem rejects it — flap orchestration lands with the scenarios that need it. Plan: docs/development/improvements/ddil-e2e-test-harness.md §1, §2. * feat: add DDIL harness witness VM helpers LaunchWitnessVM launches a counter VM via the cluster's AWS gateway, locates its QEMU host via ps, and captures a baseline counter reading. ReadCounter/AssertProgressed tunnel SSH through the host's QEMU hostfwd port to verify the workload progressed across a fault. README documents the new DDIL_WITNESS_* and DDIL_GUEST_SSH_* env vars. * feat: add DDIL harness snapshot and state assertions TakeSnapshot / AssertPreserved compare /local/instances pre and post fault, failing if IDs disappear or a running/pending instance regresses to a terminal state. LocalStateRevision / AssertMonotonic read the local state file revision from /local/status so scenarios can detect the daemon rewinding its persisted state. * feat: add DDIL harness reset, retry, and quarantine ResetAllNodes brings every node back to a clean baseline (iptables -F, tc netem clear on any interface carrying one, spinifex-nats and spinifex-daemon started) then waits for cluster health — daemon /health 200 plus NATS routez peer count matching len(nodes)-1, mirroring the shell suite's verify_nats_cluster predicate. AssertCleanState is a pre-scenario baseline check that recovers a dirty state via ResetAllNodes instead of failing outright, since a dirty baseline usually means a previous scenario's cleanup was skipped. Run wraps each scenario with quarantine handling (DDIL_QUARANTINED skip) and a single reset-and-retry on failure. First-attempt failures are not hidden; the retry is a diagnostic aid rather than a flake mask. bd: mulga-siv-5.5 * feat: DDIL scenarios skeleton (6x t.Skip + TestMain + coverage drift) Phase 1 of mulga-siv-5 (DDIL E2E harness): all six scenarios land as TestScenario<L>_... stubs that t.Skipf on their hardening-epic dependency. TestMain installs a SIGINT/SIGTERM trap that ResetAllNodes before exiting; DDIL_DRY_RUN=1 short-circuits cluster setup and logs the planned scenarios. TestCoverageDrift AST-parses sibling *_test.go files and reconciles them against the DDIL table in TEST_COVERAGE.md so Phase 2/3 flips cannot silently diverge doc from code. TestHarnessSmoke covers the AssertCleanState + ResetAllNodes round trip for the "full run on real cluster" acceptance bullet. * feat: add DDIL E2E CI workflow (workflow_dispatch only) Phase 1 CI wiring for mulga-siv-5. Mirrors e2e_real_multi's provision/bootstrap/teardown shape against the same ci-multi runner and tofu-cluster, but runs the Go harness on the runner (outside the cluster) so its SSH path survives the peer-traffic iptables rules that Scenario C installs on a cluster node. Triggers are workflow_dispatch only — nightly cron and on-main-merge come in mulga-siv-5.8 once ≥4 scenarios are asserting. results.json (go test -json) uploads as an artifact so flake-rate visibility can be built on gh run view + structured output, per the design doc.
This commit is contained in:
Josh Siviour
@@ -0,0 +1,140 @@
|
||||
name: Spinifex DDIL E2E
|
||||
|
||||
# Phase 1 trigger: workflow_dispatch only. Nightly cron + on-main-merge
|
||||
# land in mulga-siv-5.8 once ≥4 scenarios are asserting. See
|
||||
# docs/development/improvements/ddil-e2e-test-harness.md §4 / §Dependencies.
|
||||
on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
quarantined:
|
||||
description: "Comma-separated scenario letters to quarantine (A..F)"
|
||||
required: false
|
||||
default: ""
|
||||
|
||||
env:
|
||||
GIT_BRANCH: ${{ github.ref_name }}
|
||||
|
||||
jobs:
|
||||
ddil:
|
||||
name: DDIL Real Multi-Node
|
||||
runs-on: [ self-hosted, ci-multi ]
|
||||
# Budget: ~45m go test + tofu provision/bootstrap/teardown on ci-multi.
|
||||
# In Phase 1 the scenarios all t.Skip, so the real-world runtime is
|
||||
# dominated by provisioning; the 60m ceiling leaves headroom for
|
||||
# Phase 2/3 when assertions come online.
|
||||
timeout-minutes: 60
|
||||
env:
|
||||
TOFU_LOG: /tmp/e2e-tofu-ddil-${{ github.run_id }}.log
|
||||
steps:
|
||||
- name: Clean Workspace
|
||||
run: rm -rf "$GITHUB_WORKSPACE"/*
|
||||
|
||||
- uses: actions/create-github-app-token@v3
|
||||
id: app-token
|
||||
with:
|
||||
client-id: ${{ secrets.APP_ID }}
|
||||
private-key: ${{ secrets.APP_PRIVATE_KEY }}
|
||||
repositories: mulga
|
||||
|
||||
- name: Resolve Mulga Branch
|
||||
id: mulga-branch
|
||||
env:
|
||||
GH_TOKEN: ${{ steps.app-token.outputs.token }}
|
||||
BRANCH: ${{ github.ref_name }}
|
||||
run: |
|
||||
if gh api "repos/mulgadc/mulga/branches/${BRANCH}" >/dev/null 2>&1; then
|
||||
echo "ref=${BRANCH}" >> "$GITHUB_OUTPUT"
|
||||
echo "mulga: using branch ${BRANCH}"
|
||||
else
|
||||
echo "ref=main" >> "$GITHUB_OUTPUT"
|
||||
echo "mulga: branch ${BRANCH} not found, falling back to main"
|
||||
fi
|
||||
|
||||
# Full spinifex checkout: the Go harness must run on the runner
|
||||
# (outside the cluster) so its SSH path survives scenario partitions
|
||||
# that drop peer-to-peer cluster traffic. Running inside the cluster
|
||||
# would self-sever when Scenario C isolates a node.
|
||||
- uses: actions/checkout@v6
|
||||
name: Checkout Spinifex
|
||||
with:
|
||||
path: spinifex
|
||||
|
||||
- uses: actions/checkout@v6
|
||||
name: Checkout Mulga (tofu-cluster)
|
||||
with:
|
||||
repository: mulgadc/mulga
|
||||
ref: ${{ steps.mulga-branch.outputs.ref }}
|
||||
token: ${{ steps.app-token.outputs.token }}
|
||||
sparse-checkout: scripts/tofu-cluster
|
||||
path: mulga
|
||||
|
||||
- name: Build Install Artifacts
|
||||
working-directory: mulga/scripts/tofu-cluster
|
||||
run: |
|
||||
./build-install-artifacts.sh "${GIT_BRANCH}" \
|
||||
"${RUNNER_TEMP}/spinifex-distro.tar.gz" \
|
||||
"${RUNNER_TEMP}/setup.sh" \
|
||||
"${RUNNER_TEMP}/spinifex-tests.tar"
|
||||
|
||||
- name: Provision VMs
|
||||
working-directory: mulga/scripts/tofu-cluster
|
||||
run: |
|
||||
export TF_VAR_git_branch="${GIT_BRANCH}"
|
||||
tofu init -input=false >> "$TOFU_LOG" 2>&1
|
||||
tofu workspace select -or-create "${SPINIFEX_CI_ENV}" >> "$TOFU_LOG" 2>&1
|
||||
tofu destroy -var-file=envs/${SPINIFEX_CI_ENV}.tfvars -auto-approve -input=false >> "$TOFU_LOG" 2>&1 || true
|
||||
./scrub-env.sh "${SPINIFEX_CI_ENV}" >> "$TOFU_LOG" 2>&1 || true
|
||||
tofu apply -var-file=envs/${SPINIFEX_CI_ENV}.tfvars -auto-approve -input=false >> "$TOFU_LOG" 2>&1
|
||||
|
||||
- name: Bootstrap via Binary Install
|
||||
working-directory: mulga/scripts/tofu-cluster
|
||||
run: |
|
||||
SPX_EXTERNAL_POOL="${SPINIFEX_CI_POOL}" \
|
||||
./bootstrap-install.sh "${SPINIFEX_CI_ENV}" \
|
||||
"${RUNNER_TEMP}/setup.sh" \
|
||||
"${RUNNER_TEMP}/spinifex-distro.tar.gz" \
|
||||
--test-tar "${RUNNER_TEMP}/spinifex-tests.tar"
|
||||
|
||||
- name: Run DDIL Scenarios
|
||||
working-directory: mulga/scripts/tofu-cluster
|
||||
run: |
|
||||
set -o pipefail
|
||||
INVENTORY=$(tofu output -json inventory)
|
||||
SSH_KEY=$(echo "$INVENTORY" | jq -r '.ssh_key_path')
|
||||
SSH_KEY="${SSH_KEY/#\~/$HOME}"
|
||||
SSH_USER=$(echo "$INVENTORY" | jq -r '.ssh_user')
|
||||
REGION=$(echo "$INVENTORY" | jq -r '.region')
|
||||
# DDIL_NODES must be the cluster-traffic IPs: PartitionNode
|
||||
# installs iptables DROP rules for each peer, so the runner's
|
||||
# SSH source IP must differ from these. On single-NIC
|
||||
# tofu-cluster, wan_ip == cluster IP and the runner's outbound
|
||||
# IP is distinct — see e2e-ddil-test-harness.md §3 Scenario C.
|
||||
NODE_IPS=$(echo "$INVENTORY" | jq -r '[.nodes[].wan_ip] | join(",")')
|
||||
echo "DDIL_NODES=${NODE_IPS}"
|
||||
|
||||
mkdir -p "${RUNNER_TEMP}/ddil-results"
|
||||
cd "${GITHUB_WORKSPACE}/spinifex"
|
||||
DDIL_NODES="${NODE_IPS}" \
|
||||
DDIL_SSH_USER="${SSH_USER}" \
|
||||
DDIL_SSH_KEY="${SSH_KEY}" \
|
||||
DDIL_QUARANTINED="${{ inputs.quarantined }}" \
|
||||
AWS_REGION="${REGION}" \
|
||||
go test -tags=e2e -timeout 45m -json \
|
||||
./tests/e2e/ddil/scenarios/... \
|
||||
| tee "${RUNNER_TEMP}/ddil-results/results.json"
|
||||
|
||||
- name: Upload DDIL Results
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: ddil-results-${{ github.run_id }}
|
||||
path: ${{ runner.temp }}/ddil-results
|
||||
retention-days: 30
|
||||
|
||||
- name: Teardown
|
||||
if: always()
|
||||
working-directory: mulga/scripts/tofu-cluster
|
||||
run: |
|
||||
tofu workspace select -or-create "${SPINIFEX_CI_ENV}" >> "$TOFU_LOG" 2>&1
|
||||
tofu destroy -var-file=envs/${SPINIFEX_CI_ENV}.tfvars -auto-approve -input=false >> "$TOFU_LOG" 2>&1
|
||||
./scrub-env.sh "${SPINIFEX_CI_ENV}" >> "$TOFU_LOG" 2>&1 || true
|
||||
@@ -18,6 +18,7 @@ require (
|
||||
github.com/spf13/cobra v1.10.2
|
||||
github.com/spf13/viper v1.21.0
|
||||
github.com/stretchr/testify v1.11.1
|
||||
golang.org/x/crypto v0.49.0
|
||||
gopkg.in/ini.v1 v1.67.1
|
||||
)
|
||||
|
||||
@@ -94,7 +95,6 @@ require (
|
||||
go.opentelemetry.io/otel/trace v1.42.0 // indirect
|
||||
go.yaml.in/yaml/v2 v2.4.4 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/crypto v0.49.0 // indirect
|
||||
golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa // indirect
|
||||
golang.org/x/mod v0.33.0 // indirect
|
||||
golang.org/x/net v0.52.0 // indirect
|
||||
|
||||
@@ -709,3 +709,64 @@ Requires pool mode with external IPAM (NOT dev_networking).
|
||||
### Phase 4: Cleanup (trap)
|
||||
- Terminate app instances, wait for terminated
|
||||
- Delete key pair, subnet, IGW, VPC
|
||||
|
||||
## DDIL (`tests/e2e/ddil/`)
|
||||
|
||||
Go-based harness and scenarios that exercise the Denied / Disrupted /
|
||||
Intermittent / Limited failure modes called out in the DDIL vulnerability
|
||||
assessment. Design and rationale live in
|
||||
[`docs/development/improvements/ddil-e2e-test-harness.md`](../../../docs/development/improvements/ddil-e2e-test-harness.md);
|
||||
the Go helper API lives in
|
||||
[`tests/e2e/ddil/harness/`](./ddil/harness/) with the authoritative link
|
||||
profile values in [`harness/profiles.go`](./ddil/harness/profiles.go).
|
||||
|
||||
Scenarios begin as `t.Skip("requires <dep>")` and are flipped to real
|
||||
assertions by the hardening epics they cover (daemon-local-autonomy,
|
||||
predastore-ddil-hardening). The status column below is the authoritative
|
||||
live state — update it in the same PR that changes the test body.
|
||||
`TestCoverageDrift` enforces the letter column against the
|
||||
`TestScenario<L>_...` functions in `ddil/scenarios/`.
|
||||
|
||||
### Scenarios
|
||||
|
||||
| Letter | Scenario | DDIL finding closed | Status |
|
||||
|--------|----------|---------------------|--------|
|
||||
| A | NATS-only kill (daemon stays up, standalone mode) | Finding 1 | SKIPPED |
|
||||
| B | Daemon restart without NATS (recovers from local state) | Finding 1 | SKIPPED |
|
||||
| C | Clean cluster-network partition (majority + isolated both progress) | Finding 1/2 | SKIPPED |
|
||||
| D | Degraded link under SATCOM profile (fan-out + Raft) | Finding 3 | SKIPPED |
|
||||
| E | Predastore write under partition (repair journal drains on heal) | Finding 2 | SKIPPED |
|
||||
| F | Raft under SATCOM latency (≤1 election over 5 min) | Finding 3 | SKIPPED |
|
||||
|
||||
### Link profile validation
|
||||
|
||||
Values in `harness/profiles.go` are unvalidated approximations derived
|
||||
from public specifications — real-hardware measurement is deferred until
|
||||
SATCOM/HF access is available. Update the last column when a measurement
|
||||
lands.
|
||||
|
||||
| Profile | Specification source | Validation status | Last measured against real hardware |
|
||||
|---------|----------------------|-------------------|-------------------------------------|
|
||||
| LAN | baseline | unvalidated | — |
|
||||
| WAN | public site-to-site norms | unvalidated | — |
|
||||
| LTEDegraded | public congested-cellular norms | unvalidated | — |
|
||||
| SATCOM | published GEO SATCOM figures | unvalidated | — |
|
||||
| HFData | HF modem specifications | unvalidated | — |
|
||||
| Flapping | synthetic (10s up / 5s down) | unvalidated | — |
|
||||
|
||||
### Known gaps
|
||||
|
||||
- Chaos framework (randomised fault injection, long-running stress) —
|
||||
follow-up plan TBD.
|
||||
- Multi-day partition — requires long-running CI infrastructure;
|
||||
follow-up plan TBD.
|
||||
- Real-hardware SATCOM/HF validation — follow-up plan TBD.
|
||||
- Clock skew injection — no hardening epic claims Finding 7 yet;
|
||||
revisit when one lands.
|
||||
- Dual-NIC cluster network — Scenario C uses peer-IP `iptables` rules on
|
||||
the single-NIC tofu-cluster today; dedicated cluster NIC is deferred
|
||||
to `docs/development/improvements/e2e-multinode-dual-nic.md` (TBC).
|
||||
- Dockerised per-node pseudo-multinode isolation — candidate under
|
||||
[`docs/development/improvements/docker-e2e-local.md`](../../../docs/development/improvements/docker-e2e-local.md).
|
||||
- Full Go migration of the rest of the E2E suite — TBD follow-up to the
|
||||
DDIL harness pilot.
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
# DDIL E2E Test Harness
|
||||
|
||||
Go-based E2E harness that exercises Spinifex and Predastore against DDIL
|
||||
(Denied, Disrupted, Intermittent, Limited) failure modes — NATS kill,
|
||||
daemon-without-NATS restart, cluster partition, degraded links, Predastore
|
||||
writes under partition, Raft under SATCOM latency.
|
||||
|
||||
Design: [`docs/development/improvements/ddil-e2e-test-harness.md`](../../../../docs/development/improvements/ddil-e2e-test-harness.md)
|
||||
(in the mulga monorepo).
|
||||
|
||||
## Layout
|
||||
|
||||
```
|
||||
tests/e2e/ddil/
|
||||
├── doc.go # package doc
|
||||
├── README.md # this file
|
||||
├── harness/ # typed primitives (cluster, SSH, daemon client, ...)
|
||||
└── scenarios/ # TestScenarioA..F, build tag e2e
|
||||
```
|
||||
|
||||
All test and helper files use `//go:build e2e` so `go build ./...` skips them.
|
||||
Run with `-tags=e2e`.
|
||||
|
||||
## Running
|
||||
|
||||
Against a pre-provisioned 3-node Proxmox cluster:
|
||||
|
||||
```bash
|
||||
DDIL_NODES=10.0.0.1,10.0.0.2,10.0.0.3 \
|
||||
DDIL_SSH_USER=ubuntu \
|
||||
DDIL_SSH_KEY=$HOME/.ssh/tf-user-ap-southeast-2 \
|
||||
AWS_REGION=ap-southeast-2 \
|
||||
go test -tags=e2e -timeout 45m ./tests/e2e/ddil/scenarios/...
|
||||
```
|
||||
|
||||
Run a single scenario:
|
||||
|
||||
```bash
|
||||
go test -tags=e2e -run TestScenarioA ./tests/e2e/ddil/scenarios/...
|
||||
```
|
||||
|
||||
Dry-run mode (no cluster ops; verifies SSH reachability, tool availability on
|
||||
remote nodes, and runs `TestCoverageDrift` only):
|
||||
|
||||
```bash
|
||||
DDIL_DRY_RUN=1 go test -tags=e2e ./tests/e2e/ddil/scenarios/...
|
||||
```
|
||||
|
||||
Quarantine listed scenarios (failures downgrade to `t.Skip` with a tagged
|
||||
reason; quarantined scenarios must have a linked bead):
|
||||
|
||||
```bash
|
||||
DDIL_QUARANTINED=D,F go test -tags=e2e ./tests/e2e/ddil/scenarios/...
|
||||
```
|
||||
|
||||
## Environment variables
|
||||
|
||||
| Var | Required | Purpose |
|
||||
|-----|----------|---------|
|
||||
| `DDIL_NODES` | yes (unless dry-run) | Comma-separated cluster node IPs, order matches scenario references (node1, node2, node3). |
|
||||
| `DDIL_SSH_USER` | yes (unless dry-run) | SSH login user on each node. |
|
||||
| `DDIL_SSH_KEY` | yes (unless dry-run) | Path to SSH private key. |
|
||||
| `DDIL_DRY_RUN` | no | `1` skips cluster ops, runs `TestCoverageDrift` only. |
|
||||
| `DDIL_QUARANTINED` | no | Comma-separated scenario letters (A..F) to quarantine. |
|
||||
| `AWS_REGION` | yes for witness scenarios | Region used by the AWS SDK when launching witness VMs via the Spinifex AWS gateway. |
|
||||
| `DDIL_WITNESS_AMI` | no | Specific AMI ID for witness VMs. If unset, the harness picks the first `ami-ubuntu-*` image registered in the cluster. |
|
||||
| `DDIL_WITNESS_INSTANCE_TYPE` | no (default `t2.micro`) | EC2 instance type for witness VMs. |
|
||||
| `DDIL_WITNESS_KEY_NAME` | no (default `spinifex-key`) | EC2 key pair name attached to witness VMs; must match an existing registered key. |
|
||||
| `DDIL_GUEST_SSH_USER` | no (default `ubuntu`) | SSH login user inside the witness guest. |
|
||||
| `DDIL_GUEST_SSH_KEY` | no (defaults to `DDIL_SSH_KEY`) | Private key for guest SSH; override when the guest image uses a different key than the host. |
|
||||
|
||||
## Scenario status
|
||||
|
||||
Scenario-level status (SKIPPED / ENABLED / QUARANTINED) plus profile
|
||||
validation tracking lives in [`tests/e2e/TEST_COVERAGE.md`](../TEST_COVERAGE.md).
|
||||
Scenarios begin as `t.Skip("requires <dep>")` and are flipped to real
|
||||
assertions by the hardening epics they exercise.
|
||||
@@ -0,0 +1,15 @@
|
||||
// Package ddil is the root of the DDIL (Denied, Disrupted, Intermittent,
|
||||
// Limited) E2E test harness.
|
||||
//
|
||||
// See docs/development/improvements/ddil-e2e-test-harness.md in the mulga
|
||||
// monorepo for the full design.
|
||||
//
|
||||
// Sub-packages:
|
||||
// - harness: typed primitives (cluster, SSH, fault injection, witness VMs,
|
||||
// snapshot/state assertions, cleanup/retry, daemon client).
|
||||
// - scenarios: test files (build tag e2e) that exercise the hardening
|
||||
// epics via the harness primitives.
|
||||
//
|
||||
// All non-scaffolding files in this tree are tagged //go:build e2e so that
|
||||
// default go build ./... skips them.
|
||||
package ddil
|
||||
@@ -0,0 +1,112 @@
|
||||
//go:build e2e
|
||||
|
||||
// Package harness exports the DDIL E2E test primitives: typed cluster/node
|
||||
// descriptors, an SSH transport, fault-injection helpers, witness-VM helpers,
|
||||
// snapshot/state assertions, and the daemon HTTP client used by scenarios.
|
||||
//
|
||||
// See docs/development/improvements/ddil-e2e-test-harness.md for the design.
|
||||
package harness
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Node identifies a single cluster member addressable over SSH and HTTPS.
|
||||
//
|
||||
// Index is the 1-based position the node occupies in the DDIL_NODES list;
|
||||
// scenarios reference nodes by index (node1, node2, node3) so a fixed
|
||||
// ordering is required.
|
||||
type Node struct {
|
||||
Index int // 1-based position in the cluster
|
||||
Name string // human-readable tag, e.g. "node1"
|
||||
Addr string // IP or hostname; used for SSH and as peer address for partition rules
|
||||
}
|
||||
|
||||
// Cluster is an ordered set of Nodes plus shared SSH credentials.
|
||||
//
|
||||
// Scenarios receive a *Cluster from main_test.go and use its Nodes slice to
|
||||
// address individual peers. All helpers that mutate node state take a Node
|
||||
// (not an index) to keep call sites self-describing.
|
||||
type Cluster struct {
|
||||
Nodes []Node
|
||||
SSHUser string
|
||||
SSHKeyPath string
|
||||
}
|
||||
|
||||
// Peers returns every node in the cluster except target. The result is used
|
||||
// by PartitionNode to build per-peer iptables DROP rules.
|
||||
func (c *Cluster) Peers(target Node) []Node {
|
||||
out := make([]Node, 0, len(c.Nodes)-1)
|
||||
for _, n := range c.Nodes {
|
||||
if n.Index == target.Index {
|
||||
continue
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ClusterFromEnv builds a Cluster from DDIL_NODES / DDIL_SSH_USER /
|
||||
// DDIL_SSH_KEY. Nodes are named node1..nodeN in list order.
|
||||
//
|
||||
// Returns an error if any required variable is missing or if DDIL_NODES
|
||||
// contains no non-empty entries. main_test.go is expected to call this once
|
||||
// and pass the result into every scenario.
|
||||
func ClusterFromEnv() (*Cluster, error) {
|
||||
nodesRaw := os.Getenv("DDIL_NODES")
|
||||
user := os.Getenv("DDIL_SSH_USER")
|
||||
key := os.Getenv("DDIL_SSH_KEY")
|
||||
|
||||
var missing []string
|
||||
if nodesRaw == "" {
|
||||
missing = append(missing, "DDIL_NODES")
|
||||
}
|
||||
if user == "" {
|
||||
missing = append(missing, "DDIL_SSH_USER")
|
||||
}
|
||||
if key == "" {
|
||||
missing = append(missing, "DDIL_SSH_KEY")
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
return nil, fmt.Errorf("ddil harness: missing required env: %s", strings.Join(missing, ", "))
|
||||
}
|
||||
|
||||
var nodes []Node
|
||||
for i, addr := range strings.Split(nodesRaw, ",") {
|
||||
addr = strings.TrimSpace(addr)
|
||||
if addr == "" {
|
||||
continue
|
||||
}
|
||||
nodes = append(nodes, Node{
|
||||
Index: i + 1,
|
||||
Name: fmt.Sprintf("node%d", i+1),
|
||||
Addr: addr,
|
||||
})
|
||||
}
|
||||
if len(nodes) == 0 {
|
||||
return nil, errors.New("ddil harness: DDIL_NODES contained no addresses")
|
||||
}
|
||||
|
||||
return &Cluster{
|
||||
Nodes: nodes,
|
||||
SSHUser: user,
|
||||
SSHKeyPath: key,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// NodeFromEnv returns the Nth node (1-based) from the cluster defined by
|
||||
// DDIL_NODES. Convenience wrapper for single-node helpers; most scenarios
|
||||
// should call ClusterFromEnv once and index c.Nodes directly.
|
||||
func NodeFromEnv(index int) (Node, error) {
|
||||
c, err := ClusterFromEnv()
|
||||
if err != nil {
|
||||
return Node{}, err
|
||||
}
|
||||
if index < 1 || index > len(c.Nodes) {
|
||||
return Node{}, fmt.Errorf("ddil harness: node index %d out of range (have %d nodes)", index, len(c.Nodes))
|
||||
}
|
||||
return c.Nodes[index-1], nil
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/spinifex/types"
|
||||
)
|
||||
|
||||
// daemonPort is the cluster-manager HTTPS port (see
|
||||
// spinifex/daemon/daemon.go ClusterManager). The harness targets the same
|
||||
// port the production daemon exposes.
|
||||
const daemonPort = 8443
|
||||
|
||||
// DaemonMode mirrors the daemon's operating mode reported by /local/status.
|
||||
//
|
||||
// The typed enum avoids stringly-typed assertions in scenarios. The concrete
|
||||
// endpoint and JSON schema land with daemon-local-autonomy §1b; this file
|
||||
// predates that PR and its types will be swapped for the daemon's own when
|
||||
// it merges.
|
||||
type DaemonMode string
|
||||
|
||||
const (
|
||||
ModeUnknown DaemonMode = ""
|
||||
ModeCluster DaemonMode = "cluster"
|
||||
ModeStandalone DaemonMode = "standalone"
|
||||
)
|
||||
|
||||
// LocalStatus is the response shape from GET /local/status.
|
||||
//
|
||||
// Placeholder fields only — the authoritative struct ships with
|
||||
// daemon-local-autonomy §1b. Scenarios that depend on these fields remain
|
||||
// t.Skip until that epic replaces this definition with an import from the
|
||||
// daemon package.
|
||||
type LocalStatus struct {
|
||||
Node string `json:"node"`
|
||||
Mode DaemonMode `json:"mode"`
|
||||
Revision uint64 `json:"revision"`
|
||||
NATS string `json:"nats"` // "connected" | "disconnected"
|
||||
}
|
||||
|
||||
// LocalInstance is one row of GET /local/instances. Placeholder — see
|
||||
// LocalStatus.
|
||||
type LocalInstance struct {
|
||||
InstanceID string `json:"instance_id"`
|
||||
State string `json:"state"`
|
||||
PID int `json:"pid,omitempty"`
|
||||
}
|
||||
|
||||
// DaemonClient is a thin HTTPS client targeting a single daemon's local
|
||||
// endpoints. It uses one *http.Client per node so connection reuse works
|
||||
// across calls within a scenario.
|
||||
type DaemonClient struct {
|
||||
http *http.Client
|
||||
}
|
||||
|
||||
// NewDaemonClient returns a DaemonClient with TLS verification disabled
|
||||
// (the tofu-cluster uses self-signed per-node certs). Timeouts are set to
|
||||
// 5s so a wedged daemon fails fast enough for scenario retry logic.
|
||||
func NewDaemonClient() *DaemonClient {
|
||||
return &DaemonClient{
|
||||
http: &http.Client{
|
||||
Timeout: 5 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // self-signed test certs
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (d *DaemonClient) url(node Node, path string) string {
|
||||
return "https://" + net.JoinHostPort(node.Addr, strconv.Itoa(daemonPort)) + path
|
||||
}
|
||||
|
||||
func (d *DaemonClient) getJSON(ctx context.Context, node Node, path string, out any) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, d.url(node, path), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("ddil harness: daemon %s %s: %w", node.Name, path, err)
|
||||
}
|
||||
resp, err := d.http.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("ddil harness: daemon %s %s: %w", node.Name, path, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
return fmt.Errorf("ddil harness: daemon %s %s: status %d: %s",
|
||||
node.Name, path, resp.StatusCode, string(body))
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
|
||||
return fmt.Errorf("ddil harness: daemon %s %s decode: %w", node.Name, path, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Status returns the daemon's self-reported local status.
|
||||
//
|
||||
// Depends on daemon-local-autonomy §1b. Until that lands, the daemon will
|
||||
// return 404 and callers should t.Skip.
|
||||
func (d *DaemonClient) Status(ctx context.Context, node Node) (LocalStatus, error) {
|
||||
var s LocalStatus
|
||||
if err := d.getJSON(ctx, node, "/local/status", &s); err != nil {
|
||||
return LocalStatus{}, err
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Instances returns the daemon's locally-known instance list.
|
||||
//
|
||||
// Depends on daemon-local-autonomy §1a. Until that lands, the daemon will
|
||||
// return 404 and callers should t.Skip.
|
||||
func (d *DaemonClient) Instances(ctx context.Context, node Node) ([]LocalInstance, error) {
|
||||
var xs []LocalInstance
|
||||
if err := d.getJSON(ctx, node, "/local/instances", &xs); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return xs, nil
|
||||
}
|
||||
|
||||
// Health hits the daemon's existing /health endpoint. Useful during
|
||||
// scaffolding validation before the /local/* endpoints exist — confirms
|
||||
// the harness can reach the daemon at all.
|
||||
func (d *DaemonClient) Health(ctx context.Context, node Node) (types.NodeHealthResponse, error) {
|
||||
var h types.NodeHealthResponse
|
||||
if err := d.getJSON(ctx, node, "/health", &h); err != nil {
|
||||
return types.NodeHealthResponse{}, err
|
||||
}
|
||||
return h, nil
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// PartitionNode isolates target from peers by installing per-peer iptables
|
||||
// DROP rules for both directions. The orchestrator's SSH source IP must not
|
||||
// be one of peers (it normally isn't: the peers are cluster members, and the
|
||||
// orchestrator is the CI runner). After applying, a sanity-check SSH echo
|
||||
// confirms the control plane is still reachable; a lost control-plane SSH
|
||||
// is a loud error rather than a silent hang.
|
||||
//
|
||||
// The rule set is additive (iptables -I INPUT/OUTPUT) so PartitionNode
|
||||
// composes with any prior non-DDIL rules. HealNode's flush is still the
|
||||
// correct teardown because the harness owns the target's firewall during a
|
||||
// scenario.
|
||||
func PartitionNode(ctx context.Context, ssh SSH, target Node, peers []Node) error {
|
||||
if len(peers) == 0 {
|
||||
return fmt.Errorf("ddil harness: PartitionNode %s: no peers supplied", target.Name)
|
||||
}
|
||||
|
||||
var lines []string
|
||||
for _, p := range peers {
|
||||
lines = append(lines,
|
||||
fmt.Sprintf("sudo iptables -I INPUT -s %s -j DROP", shellQuote(p.Addr)),
|
||||
fmt.Sprintf("sudo iptables -I OUTPUT -d %s -j DROP", shellQuote(p.Addr)),
|
||||
)
|
||||
}
|
||||
cmd := strings.Join(lines, " && ")
|
||||
if _, err := ssh.Run(ctx, target, cmd); err != nil {
|
||||
return fmt.Errorf("ddil harness: partition %s: %w", target.Name, err)
|
||||
}
|
||||
|
||||
// Sanity-check: if our SSH path ran through a peer IP, the rules we just
|
||||
// installed would have severed it. Fail loudly so a future infra change
|
||||
// (e.g. orchestrator moved onto the cluster network) surfaces immediately.
|
||||
if _, err := ssh.Run(ctx, target, "echo ok"); err != nil {
|
||||
return fmt.Errorf("ddil harness: partition %s severed orchestrator SSH "+
|
||||
"(orchestrator may share peer IPs): %w", target.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// HealNode flushes and deletes all iptables rules on target, reversing a
|
||||
// prior PartitionNode. Idempotent: safe on a node with no rules installed.
|
||||
func HealNode(ctx context.Context, ssh SSH, target Node) error {
|
||||
if _, err := ssh.Run(ctx, target, "sudo iptables -F && sudo iptables -X"); err != nil {
|
||||
return fmt.Errorf("ddil harness: heal %s: %w", target.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// KillNATS stops spinifex-nats on node without touching spinifex-daemon.
|
||||
// This is the scenario shape the current happy-path suite cannot express —
|
||||
// `systemctl stop spinifex.target` brings both down together.
|
||||
func KillNATS(ctx context.Context, ssh SSH, node Node) error {
|
||||
if _, err := ssh.Run(ctx, node, "sudo systemctl stop spinifex-nats"); err != nil {
|
||||
return fmt.Errorf("ddil harness: kill nats on %s: %w", node.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// StartNATS starts spinifex-nats on node. Paired with KillNATS.
|
||||
func StartNATS(ctx context.Context, ssh SSH, node Node) error {
|
||||
if _, err := ssh.Run(ctx, node, "sudo systemctl start spinifex-nats"); err != nil {
|
||||
return fmt.Errorf("ddil harness: start nats on %s: %w", node.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// RestartDaemonOnly restarts spinifex-daemon without touching spinifex-nats.
|
||||
// Used by Scenario B to exercise the daemon-without-NATS startup path
|
||||
// (daemon-local-autonomy §1d).
|
||||
func RestartDaemonOnly(ctx context.Context, ssh SSH, node Node) error {
|
||||
if _, err := ssh.Run(ctx, node, "sudo systemctl restart spinifex-daemon"); err != nil {
|
||||
return fmt.Errorf("ddil harness: restart daemon on %s: %w", node.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// WaitForMode polls the daemon's /local/status until it reports the expected
|
||||
// mode or timeout expires. Poll interval is 1s — fast enough that a
|
||||
// 30-second timeout sees ~30 attempts, slow enough not to flood a recovering
|
||||
// daemon.
|
||||
//
|
||||
// Depends on daemon-local-autonomy §1b. Until that endpoint ships, this
|
||||
// function will time out; callers gate on t.Skip.
|
||||
func WaitForMode(ctx context.Context, dc *DaemonClient, node Node, want DaemonMode, timeout time.Duration) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
const interval = 1 * time.Second
|
||||
|
||||
var lastErr error
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
status, err := dc.Status(ctx, node)
|
||||
if err == nil && status.Mode == want {
|
||||
return nil
|
||||
}
|
||||
lastErr = err
|
||||
|
||||
if time.Now().After(deadline) {
|
||||
if lastErr != nil {
|
||||
return fmt.Errorf("ddil harness: wait for mode %s on %s: timed out after %s: last error: %w",
|
||||
want, node.Name, timeout, lastErr)
|
||||
}
|
||||
return fmt.Errorf("ddil harness: wait for mode %s on %s: timed out after %s (still reporting another mode)",
|
||||
want, node.Name, timeout)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ApplyNetem installs a tc netem qdisc on iface of node matching profile.
|
||||
// Idempotent: any existing root qdisc on the interface is replaced.
|
||||
//
|
||||
// Flapping profiles have no static tc expression — they are driven by a
|
||||
// separate orchestrator (not yet implemented). ApplyNetem rejects them with
|
||||
// a clear error so callers don't silently apply a no-op.
|
||||
func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile LinkProfile) error {
|
||||
if profile.Flapping != nil {
|
||||
return fmt.Errorf("ddil harness: profile %q is flap-driven; use a flap orchestrator, not ApplyNetem", profile.Name)
|
||||
}
|
||||
|
||||
var netemArgs []string
|
||||
if profile.Delay > 0 {
|
||||
delay := fmt.Sprintf("%dms", profile.Delay.Milliseconds())
|
||||
if profile.Jitter > 0 {
|
||||
delay += fmt.Sprintf(" %dms distribution normal", profile.Jitter.Milliseconds())
|
||||
}
|
||||
netemArgs = append(netemArgs, "delay", delay)
|
||||
}
|
||||
if profile.Loss > 0 {
|
||||
netemArgs = append(netemArgs, "loss", fmt.Sprintf("%.3f%%", profile.Loss))
|
||||
}
|
||||
if profile.Bandwidth != "" {
|
||||
netemArgs = append(netemArgs, "rate", profile.Bandwidth)
|
||||
}
|
||||
if len(netemArgs) == 0 {
|
||||
return fmt.Errorf("ddil harness: profile %q has no netem parameters", profile.Name)
|
||||
}
|
||||
|
||||
// `tc qdisc replace` is idempotent — it adds if absent, replaces if
|
||||
// present — which matches ApplyNetem's contract.
|
||||
cmd := fmt.Sprintf("sudo tc qdisc replace dev %s root netem %s",
|
||||
shellQuote(iface), strings.Join(netemArgs, " "))
|
||||
if _, err := ssh.Run(ctx, node, cmd); err != nil {
|
||||
return fmt.Errorf("ddil harness: apply netem %s on %s (%s): %w",
|
||||
profile.Name, node.Name, iface, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ClearNetem removes any root qdisc on iface. Safe to call when no qdisc is
|
||||
// present — tc's non-zero exit is swallowed on the remote side.
|
||||
func ClearNetem(ctx context.Context, ssh SSH, node Node, iface string) error {
|
||||
cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", shellQuote(iface))
|
||||
if _, err := ssh.Run(ctx, node, cmd); err != nil {
|
||||
return fmt.Errorf("ddil harness: clear netem on %s (%s): %w", node.Name, iface, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// shellQuote wraps s in single quotes and escapes embedded single quotes, so
|
||||
// that shell metacharacters in interface names or peer addresses are treated
|
||||
// as literals on the remote shell.
|
||||
func shellQuote(s string) string {
|
||||
return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import "time"
|
||||
|
||||
// LinkProfile describes a tc netem configuration approximating a real-world
|
||||
// tactical link. Values are unvalidated approximations derived from public
|
||||
// specifications; see docs/development/improvements/ddil-e2e-test-harness.md
|
||||
// §2 and spinifex/tests/e2e/TEST_COVERAGE.md for validation tracking.
|
||||
type LinkProfile struct {
|
||||
Name string
|
||||
Delay time.Duration
|
||||
Jitter time.Duration
|
||||
Loss float64 // percent, 0..100
|
||||
Bandwidth string // tc-compatible rate, e.g. "512kbit"; empty means unshaped
|
||||
Flapping *FlapSpec
|
||||
}
|
||||
|
||||
// FlapSpec toggles link up/down for profiles that simulate jammed or
|
||||
// intermittent RF. The flap loop itself is not driven by ApplyNetem — it is
|
||||
// orchestrated by a higher-level helper (added with the scenarios that use
|
||||
// it) so scenarios keep full control over when the flap stops.
|
||||
type FlapSpec struct {
|
||||
Up time.Duration
|
||||
Down time.Duration
|
||||
}
|
||||
|
||||
// Authoritative DDIL link profile values. See the §2 table in the design doc
|
||||
// for the rationale behind each row. Replace with measured values when
|
||||
// real-hardware access is available.
|
||||
var (
|
||||
LAN = LinkProfile{
|
||||
Name: "LAN",
|
||||
Delay: 1 * time.Millisecond,
|
||||
Jitter: 0,
|
||||
Loss: 0,
|
||||
Bandwidth: "1gbit",
|
||||
}
|
||||
WAN = LinkProfile{
|
||||
Name: "WAN",
|
||||
Delay: 50 * time.Millisecond,
|
||||
Jitter: 10 * time.Millisecond,
|
||||
Loss: 0.1,
|
||||
Bandwidth: "100mbit",
|
||||
}
|
||||
LTEDegraded = LinkProfile{
|
||||
Name: "LTEDegraded",
|
||||
Delay: 100 * time.Millisecond,
|
||||
Jitter: 30 * time.Millisecond,
|
||||
Loss: 5,
|
||||
Bandwidth: "1mbit",
|
||||
}
|
||||
SATCOM = LinkProfile{
|
||||
Name: "SATCOM",
|
||||
Delay: 600 * time.Millisecond,
|
||||
Jitter: 50 * time.Millisecond,
|
||||
Loss: 2,
|
||||
Bandwidth: "512kbit",
|
||||
}
|
||||
HFData = LinkProfile{
|
||||
Name: "HFData",
|
||||
Delay: 2000 * time.Millisecond,
|
||||
Jitter: 200 * time.Millisecond,
|
||||
Loss: 10,
|
||||
Bandwidth: "9600bit",
|
||||
}
|
||||
Flapping = LinkProfile{
|
||||
Name: "Flapping",
|
||||
Flapping: &FlapSpec{
|
||||
Up: 10 * time.Second,
|
||||
Down: 5 * time.Second,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
// AllProfiles lists every predefined profile. Useful for doc-drift checks and
|
||||
// for the TEST_COVERAGE.md profile-validation table.
|
||||
var AllProfiles = []LinkProfile{LAN, WAN, LTEDegraded, SATCOM, HFData, Flapping}
|
||||
@@ -0,0 +1,207 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// routezResponse is the subset of the NATS /routez monitoring payload the
|
||||
// harness consumes. Only remote_name is required because the harness counts
|
||||
// unique peer names (not raw route count — NATS opens multiple TCP routes
|
||||
// per peer and num_routes would overcount). The shell suite's
|
||||
// verify_nats_cluster helper uses the same predicate.
|
||||
type routezResponse struct {
|
||||
Routes []struct {
|
||||
RemoteName string `json:"remote_name"`
|
||||
} `json:"routes"`
|
||||
}
|
||||
|
||||
// ResetAllNodes returns the cluster to a known clean state: every node has
|
||||
// all iptables rules flushed, every tc netem qdisc removed, and both
|
||||
// spinifex-nats and spinifex-daemon running. After the per-node reset, it
|
||||
// waits for cluster health (every daemon /health returns 200 and each node's
|
||||
// NATS monitor reports len(Nodes)-1 unique route peers) before returning.
|
||||
//
|
||||
// Idempotent — safe against a clean cluster. Three callers:
|
||||
// 1. harness.Run, between a failed scenario and its retry.
|
||||
// 2. AssertCleanState, when a pre-scenario baseline is dirty.
|
||||
// 3. The suite-level signal handler in TestMain, so SIGINT/SIGTERM do not
|
||||
// leave the cluster partitioned or netem-shaped.
|
||||
func ResetAllNodes(ctx context.Context, c *Cluster, ssh SSH) error {
|
||||
for _, n := range c.Nodes {
|
||||
if err := resetNode(ctx, ssh, n); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return waitClusterHealthy(ctx, c, ssh, 2*time.Minute)
|
||||
}
|
||||
|
||||
func resetNode(ctx context.Context, ssh SSH, n Node) error {
|
||||
// Flush every DDIL-owned firewall rule. iptables -F on the default
|
||||
// (filter) table covers INPUT/OUTPUT DROPs installed by PartitionNode;
|
||||
// iptables -X removes any transient user chains a scenario happens to
|
||||
// leave behind. This matches HealNode's teardown.
|
||||
if _, err := ssh.Run(ctx, n, "sudo iptables -F && sudo iptables -X"); err != nil {
|
||||
return fmt.Errorf("ddil harness: reset iptables on %s: %w", n.Name, err)
|
||||
}
|
||||
|
||||
// Remove netem qdiscs only on interfaces that actually carry one.
|
||||
// Scenarios may pick different NICs (eth0, ens18, br-mgmt) and a fixed
|
||||
// interface list here would silently miss whichever one was used.
|
||||
ifaces, err := netemIfaces(ctx, ssh, n)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, iface := range ifaces {
|
||||
cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", shellQuote(iface))
|
||||
if _, err := ssh.Run(ctx, n, cmd); err != nil {
|
||||
return fmt.Errorf("ddil harness: clear netem on %s (%s): %w", n.Name, iface, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Start services — noop if already active (systemd returns 0).
|
||||
if _, err := ssh.Run(ctx, n, "sudo systemctl start spinifex-nats spinifex-daemon"); err != nil {
|
||||
return fmt.Errorf("ddil harness: start services on %s: %w", n.Name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// netemIfaces lists every interface on node whose root qdisc is a netem
|
||||
// qdisc. Used by resetNode to decide what to tear down and by
|
||||
// AssertCleanState to detect a dirty baseline.
|
||||
func netemIfaces(ctx context.Context, ssh SSH, n Node) ([]string, error) {
|
||||
// `tc qdisc show` prints one row per qdisc; `qdisc netem 8003: dev eth0
|
||||
// root ...` → awk $5 yields the interface name.
|
||||
out, err := ssh.Run(ctx, n, "tc qdisc show | awk '/qdisc netem/ {print $5}' | sort -u")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: list netem qdiscs on %s: %w", n.Name, err)
|
||||
}
|
||||
return strings.Fields(string(out)), nil
|
||||
}
|
||||
|
||||
func waitClusterHealthy(ctx context.Context, c *Cluster, ssh SSH, timeout time.Duration) error {
|
||||
dc := NewDaemonClient()
|
||||
expectedPeers := len(c.Nodes) - 1
|
||||
|
||||
deadline := time.Now().Add(timeout)
|
||||
const interval = 2 * time.Second
|
||||
|
||||
var lastErr error
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
if err := checkClusterHealthy(ctx, c, ssh, dc, expectedPeers); err == nil {
|
||||
return nil
|
||||
} else {
|
||||
lastErr = err
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("ddil harness: cluster did not reach healthy within %s: %w", timeout, lastErr)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func checkClusterHealthy(ctx context.Context, c *Cluster, ssh SSH, dc *DaemonClient, expectedPeers int) error {
|
||||
for _, n := range c.Nodes {
|
||||
if _, err := dc.Health(ctx, n); err != nil {
|
||||
return fmt.Errorf("%s daemon health: %w", n.Name, err)
|
||||
}
|
||||
peers, err := natsPeerCount(ctx, ssh, n)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s nats peers: %w", n.Name, err)
|
||||
}
|
||||
if peers < expectedPeers {
|
||||
return fmt.Errorf("%s nats peers: have %d, want %d", n.Name, peers, expectedPeers)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// natsPeerCount SSHes into node, curls the node-local NATS monitor endpoint
|
||||
// (bound to 127.0.0.1:8222 by the spinifex NATS config), and returns the
|
||||
// number of unique peer names it reports.
|
||||
func natsPeerCount(ctx context.Context, ssh SSH, n Node) (int, error) {
|
||||
out, err := ssh.Run(ctx, n, "curl -fsS http://127.0.0.1:8222/routez")
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("curl routez: %w", err)
|
||||
}
|
||||
var rz routezResponse
|
||||
if err := json.Unmarshal(out, &rz); err != nil {
|
||||
return 0, fmt.Errorf("parse routez: %w", err)
|
||||
}
|
||||
seen := make(map[string]struct{})
|
||||
for _, r := range rz.Routes {
|
||||
if r.RemoteName == "" {
|
||||
continue
|
||||
}
|
||||
seen[r.RemoteName] = struct{}{}
|
||||
}
|
||||
return len(seen), nil
|
||||
}
|
||||
|
||||
// AssertCleanState verifies no node carries a DDIL-owned firewall rule or
|
||||
// netem qdisc before a scenario begins. A dirty baseline is not itself a
|
||||
// test failure — it usually means a previous scenario's t.Cleanup did not
|
||||
// run (panic, SIGKILL, CI runner killed) — so the helper logs details and
|
||||
// calls ResetAllNodes to recover. Only a failure of that reset is fatal.
|
||||
func AssertCleanState(ctx context.Context, t *testing.T, c *Cluster, ssh SSH) {
|
||||
t.Helper()
|
||||
|
||||
var dirty []string
|
||||
for _, n := range c.Nodes {
|
||||
drops, err := countDropRules(ctx, ssh, n)
|
||||
if err != nil {
|
||||
t.Fatalf("ddil harness: inspect iptables on %s: %v", n.Name, err)
|
||||
}
|
||||
ifaces, err := netemIfaces(ctx, ssh, n)
|
||||
if err != nil {
|
||||
t.Fatalf("ddil harness: inspect netem on %s: %v", n.Name, err)
|
||||
}
|
||||
if drops > 0 {
|
||||
dirty = append(dirty, fmt.Sprintf("%s: %d iptables DROP rules", n.Name, drops))
|
||||
}
|
||||
if len(ifaces) > 0 {
|
||||
dirty = append(dirty, fmt.Sprintf("%s: netem on %v", n.Name, ifaces))
|
||||
}
|
||||
}
|
||||
|
||||
if len(dirty) == 0 {
|
||||
return
|
||||
}
|
||||
t.Logf("ddil harness: dirty baseline before scenario — %s; running ResetAllNodes", strings.Join(dirty, "; "))
|
||||
if err := ResetAllNodes(ctx, c, ssh); err != nil {
|
||||
t.Fatalf("ddil harness: reset after dirty baseline: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// countDropRules reports how many DROP rules are installed in INPUT/OUTPUT
|
||||
// on node. `|| true` at the shell level keeps grep's no-match exit 1 from
|
||||
// being surfaced as an SSH error.
|
||||
func countDropRules(ctx context.Context, ssh SSH, n Node) (int, error) {
|
||||
out, err := ssh.Run(ctx, n, "sudo iptables -S INPUT -S OUTPUT | grep -c -- '-j DROP' || true")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
s := strings.TrimSpace(string(out))
|
||||
if s == "" {
|
||||
return 0, nil
|
||||
}
|
||||
count, err := strconv.Atoi(s)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("parse iptables drop count %q: %w", s, err)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Run wraps a scenario with reset-and-retry plus quarantine handling. It is
|
||||
// the primary entry point used by the files in scenarios/:
|
||||
//
|
||||
// harness.Run(t, cluster, ssh, "A", func(t *testing.T) { ... })
|
||||
//
|
||||
// Quarantine: scenarios whose letter appears in DDIL_QUARANTINED are skipped
|
||||
// instead of executed. Per the design doc, a quarantined scenario must have
|
||||
// a linked bead tracking its fix; the live status is tracked in
|
||||
// tests/e2e/TEST_COVERAGE.md.
|
||||
//
|
||||
// Retry: a first-attempt failure runs ResetAllNodes and invokes fn once more
|
||||
// as a fresh subtest. Both attempts appear in the test output so CI logs
|
||||
// preserve the transient-vs-deterministic signal. The first-attempt failure
|
||||
// is not hidden — Go's testing package propagates subtest failures to the
|
||||
// parent — which is intentional: the retry is a diagnostic aid, not a way
|
||||
// to mask flakes. Scenarios crossing the 5% nightly flake threshold are
|
||||
// moved to DDIL_QUARANTINED instead.
|
||||
func Run(t *testing.T, c *Cluster, ssh SSH, letter string, fn func(*testing.T)) {
|
||||
t.Helper()
|
||||
|
||||
if IsQuarantined(letter) {
|
||||
t.Skipf("QUARANTINED: scenario %s (DDIL_QUARANTINED=%s)", letter, os.Getenv("DDIL_QUARANTINED"))
|
||||
return
|
||||
}
|
||||
|
||||
if t.Run("attempt-1", fn) {
|
||||
return
|
||||
}
|
||||
|
||||
t.Logf("ddil harness: scenario %s failed first attempt, resetting cluster and retrying", letter)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := ResetAllNodes(ctx, c, ssh); err != nil {
|
||||
t.Errorf("ddil harness: reset before retry of %s: %v", letter, err)
|
||||
return
|
||||
}
|
||||
t.Run("attempt-2", fn)
|
||||
}
|
||||
|
||||
// IsQuarantined reports whether letter appears in DDIL_QUARANTINED.
|
||||
// Comparison is case-insensitive; whitespace and empty entries are tolerated
|
||||
// so the env var can be edited by humans without worrying about formatting.
|
||||
func IsQuarantined(letter string) bool {
|
||||
raw := os.Getenv("DDIL_QUARANTINED")
|
||||
if raw == "" {
|
||||
return false
|
||||
}
|
||||
want := strings.ToUpper(strings.TrimSpace(letter))
|
||||
for _, part := range strings.Split(raw, ",") {
|
||||
if strings.ToUpper(strings.TrimSpace(part)) == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// InstanceSnapshot captures a point-in-time view of a single daemon's locally
|
||||
// known instances. Scenarios take a snapshot before injecting a fault and a
|
||||
// second snapshot after the cluster heals, then call AssertPreserved to prove
|
||||
// the daemon did not lose or regress any instance it had been tracking.
|
||||
//
|
||||
// Reuses the LocalInstance type from daemon_client.go. Once daemon-local-
|
||||
// autonomy §1a lands, that placeholder will be replaced by the daemon's
|
||||
// authoritative struct and this alias will pick up the real schema without
|
||||
// scenario changes.
|
||||
type InstanceSnapshot []LocalInstance
|
||||
|
||||
// TakeSnapshot reads /local/instances on the given node and returns the
|
||||
// response as an InstanceSnapshot.
|
||||
//
|
||||
// The DaemonClient is passed explicitly (rather than through a package-level
|
||||
// singleton) so each scenario owns its connection reuse and so dry-run
|
||||
// scenarios can stub with a fake client once the endpoint exists.
|
||||
func TakeSnapshot(ctx context.Context, d *DaemonClient, node Node) (InstanceSnapshot, error) {
|
||||
xs, err := d.Instances(ctx, node)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: snapshot %s: %w", node.Name, err)
|
||||
}
|
||||
return InstanceSnapshot(xs), nil
|
||||
}
|
||||
|
||||
// AssertPreserved fails t if any instance present in the pre-snapshot is
|
||||
// absent from post, or if any instance regressed from a live state (pending,
|
||||
// running) to a terminal state (shutting-down, stopped, terminated) without
|
||||
// explicit scenario intent.
|
||||
//
|
||||
// EC2-style state ordering is used rather than the daemon's internal enum
|
||||
// because /local/instances reports the same `state` string the gateway
|
||||
// exposes over `ec2.DescribeInstances`. When daemon-local-autonomy §1a
|
||||
// replaces LocalInstance with the daemon type, the stateOrdinal map below
|
||||
// should be pointed at the authoritative constants.
|
||||
func (pre InstanceSnapshot) AssertPreserved(t *testing.T, post InstanceSnapshot) {
|
||||
t.Helper()
|
||||
|
||||
postByID := make(map[string]LocalInstance, len(post))
|
||||
for _, p := range post {
|
||||
postByID[p.InstanceID] = p
|
||||
}
|
||||
|
||||
var missing []string
|
||||
var regressed []string
|
||||
for _, p := range pre {
|
||||
q, ok := postByID[p.InstanceID]
|
||||
if !ok {
|
||||
missing = append(missing, p.InstanceID)
|
||||
continue
|
||||
}
|
||||
if stateRegressed(p.State, q.State) {
|
||||
regressed = append(regressed, fmt.Sprintf("%s: %s → %s", p.InstanceID, p.State, q.State))
|
||||
}
|
||||
}
|
||||
|
||||
if len(missing) == 0 && len(regressed) == 0 {
|
||||
return
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
t.Errorf("ddil harness: instances disappeared from snapshot: %v", missing)
|
||||
}
|
||||
if len(regressed) > 0 {
|
||||
t.Errorf("ddil harness: instances regressed to a terminal state: %v", regressed)
|
||||
}
|
||||
t.FailNow()
|
||||
}
|
||||
|
||||
// stateOrdinal maps EC2 instance states to a monotonic lifecycle position.
|
||||
// Unknown states return -1 so we can distinguish them from "pending" (0) and
|
||||
// refuse to flag a regression we can't reason about.
|
||||
func stateOrdinal(s string) int {
|
||||
switch s {
|
||||
case "pending":
|
||||
return 0
|
||||
case "running":
|
||||
return 1
|
||||
case "stopping":
|
||||
return 2
|
||||
case "stopped":
|
||||
return 3
|
||||
case "shutting-down":
|
||||
return 4
|
||||
case "terminated":
|
||||
return 5
|
||||
default:
|
||||
return -1
|
||||
}
|
||||
}
|
||||
|
||||
// stateRegressed reports whether a live instance (pending/running) moved to
|
||||
// a terminal state (stopping/stopped/shutting-down/terminated). Moves within
|
||||
// the live band or within the terminal band are not flagged, and unknown
|
||||
// states are treated as non-regressing so the assertion does not fail on
|
||||
// schema drift.
|
||||
func stateRegressed(pre, post string) bool {
|
||||
preOrd := stateOrdinal(pre)
|
||||
postOrd := stateOrdinal(post)
|
||||
if preOrd < 0 || postOrd < 0 {
|
||||
return false
|
||||
}
|
||||
return preOrd <= 1 && postOrd >= 2
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// SSH is the transport scenarios use to run commands on a remote Node.
|
||||
//
|
||||
// It is an interface so dry-run mode and unit tests can stub it without
|
||||
// opening real network connections. Run returns the command's combined
|
||||
// stdout+stderr and a non-nil error if the remote command exited non-zero or
|
||||
// the session failed.
|
||||
type SSH interface {
|
||||
Run(ctx context.Context, node Node, cmd string) ([]byte, error)
|
||||
Close() error
|
||||
}
|
||||
|
||||
// sshClient is the production SSH transport, backed by
|
||||
// golang.org/x/crypto/ssh. One client is opened per (user, key, node) tuple
|
||||
// on first use and cached for the harness lifetime.
|
||||
type sshClient struct {
|
||||
user string
|
||||
signer ssh.Signer
|
||||
conns map[string]*ssh.Client
|
||||
connectDur time.Duration
|
||||
runDur time.Duration
|
||||
}
|
||||
|
||||
// NewSSH returns an SSH transport configured from a Cluster.
|
||||
//
|
||||
// The private key at cluster.SSHKeyPath must be in a format understood by
|
||||
// ssh.ParsePrivateKey (OpenSSH or PEM). Host keys are not verified — the
|
||||
// harness runs against short-lived test infrastructure where TOFU would add
|
||||
// noise and no security benefit. Do not reuse this transport for production
|
||||
// access.
|
||||
func NewSSH(cluster *Cluster) (SSH, error) {
|
||||
keyBytes, err := os.ReadFile(cluster.SSHKeyPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: read ssh key %s: %w", cluster.SSHKeyPath, err)
|
||||
}
|
||||
signer, err := ssh.ParsePrivateKey(keyBytes)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: parse ssh key %s: %w", cluster.SSHKeyPath, err)
|
||||
}
|
||||
return &sshClient{
|
||||
user: cluster.SSHUser,
|
||||
signer: signer,
|
||||
conns: make(map[string]*ssh.Client),
|
||||
connectDur: 10 * time.Second,
|
||||
runDur: 2 * time.Minute,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *sshClient) dial(node Node) (*ssh.Client, error) {
|
||||
if c, ok := s.conns[node.Addr]; ok {
|
||||
return c, nil
|
||||
}
|
||||
cfg := &ssh.ClientConfig{
|
||||
User: s.user,
|
||||
Auth: []ssh.AuthMethod{ssh.PublicKeys(s.signer)},
|
||||
HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec // test harness, not production
|
||||
Timeout: s.connectDur,
|
||||
}
|
||||
c, err := ssh.Dial("tcp", net.JoinHostPort(node.Addr, "22"), cfg)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: ssh dial %s: %w", node.Addr, err)
|
||||
}
|
||||
s.conns[node.Addr] = c
|
||||
return c, nil
|
||||
}
|
||||
|
||||
func (s *sshClient) Run(ctx context.Context, node Node, cmd string) ([]byte, error) {
|
||||
client, err := s.dial(node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
session, err := client.NewSession()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: ssh new session %s: %w", node.Name, err)
|
||||
}
|
||||
defer func() { _ = session.Close() }()
|
||||
|
||||
var out bytes.Buffer
|
||||
session.Stdout = &out
|
||||
session.Stderr = &out
|
||||
|
||||
// Propagate context cancellation by signalling the session. ssh.Session
|
||||
// has no Kill; Signal(KILL) is the portable way to interrupt a running
|
||||
// command.
|
||||
done := make(chan struct{})
|
||||
defer close(done)
|
||||
go func() {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
_ = session.Signal(ssh.SIGKILL)
|
||||
_ = session.Close()
|
||||
case <-done:
|
||||
}
|
||||
}()
|
||||
|
||||
if err := session.Run(cmd); err != nil {
|
||||
return out.Bytes(), fmt.Errorf("ddil harness: ssh run on %s (%q): %w\noutput: %s",
|
||||
node.Name, cmd, err, out.String())
|
||||
}
|
||||
return out.Bytes(), nil
|
||||
}
|
||||
|
||||
func (s *sshClient) Close() error {
|
||||
var firstErr error
|
||||
for addr, c := range s.conns {
|
||||
if err := c.Close(); err != nil && firstErr == nil {
|
||||
firstErr = fmt.Errorf("ddil harness: ssh close %s: %w", addr, err)
|
||||
}
|
||||
delete(s.conns, addr)
|
||||
}
|
||||
return firstErr
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// LocalStateRevision reads the daemon's local state file revision via
|
||||
// /local/status. The revision is a monotonic counter bumped on every write
|
||||
// to the local state file (see daemon-local-autonomy §1a); AssertMonotonic
|
||||
// uses it to prove that a fault did not cause the daemon to rewind or reset
|
||||
// its persisted state.
|
||||
func LocalStateRevision(ctx context.Context, d *DaemonClient, node Node) (uint64, error) {
|
||||
s, err := d.Status(ctx, node)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ddil harness: local state revision %s: %w", node.Name, err)
|
||||
}
|
||||
return s.Revision, nil
|
||||
}
|
||||
|
||||
// AssertMonotonic fails t if after is strictly less than before. Equal
|
||||
// revisions are permitted because a scenario that exercises a read-only
|
||||
// fault (e.g. NATS kill with no instance mutations) should see the revision
|
||||
// unchanged.
|
||||
func AssertMonotonic(t *testing.T, before, after uint64) {
|
||||
t.Helper()
|
||||
if after < before {
|
||||
t.Fatalf("ddil harness: local state revision regressed: before=%d after=%d", before, after)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
#cloud-config
|
||||
# DDIL witness VM payload. Runs a monotonic counter-to-file loop so the
|
||||
# harness can prove a workload kept making progress across a fault injection.
|
||||
# See docs/development/improvements/ddil-e2e-test-harness.md §1 (witness helpers).
|
||||
|
||||
write_files:
|
||||
- path: /usr/local/bin/witness-counter.sh
|
||||
permissions: '0755'
|
||||
owner: root:root
|
||||
content: |
|
||||
#!/bin/sh
|
||||
# Resume from the last recorded value if the service restarts so the
|
||||
# counter stays monotonic across daemon/VM restarts — AssertProgressed
|
||||
# relies on the count never going backwards.
|
||||
i=0
|
||||
if [ -r /var/lib/counter ]; then
|
||||
existing=$(cat /var/lib/counter 2>/dev/null || echo 0)
|
||||
case "$existing" in
|
||||
''|*[!0-9]*) i=0 ;;
|
||||
*) i=$existing ;;
|
||||
esac
|
||||
fi
|
||||
while :; do
|
||||
echo "$i" > /var/lib/counter
|
||||
i=$((i + 1))
|
||||
sleep 1
|
||||
done
|
||||
- path: /etc/systemd/system/witness-counter.service
|
||||
content: |
|
||||
[Unit]
|
||||
Description=DDIL witness counter
|
||||
After=network.target
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStartPre=/bin/sh -c 'mkdir -p /var/lib && touch /var/lib/counter'
|
||||
ExecStart=/usr/local/bin/witness-counter.sh
|
||||
Restart=always
|
||||
RestartSec=1
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
runcmd:
|
||||
- [systemctl, daemon-reload]
|
||||
- [systemctl, enable, --now, witness-counter.service]
|
||||
@@ -0,0 +1,420 @@
|
||||
//go:build e2e
|
||||
|
||||
package harness
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
_ "embed"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go/aws"
|
||||
"github.com/aws/aws-sdk-go/aws/session"
|
||||
"github.com/aws/aws-sdk-go/service/ec2"
|
||||
"golang.org/x/crypto/ssh"
|
||||
)
|
||||
|
||||
// cloudInitWitness is the user-data payload that turns a fresh cloud-init VM
|
||||
// into a monotonic counter. See testdata/cloud-init-witness.yaml for the
|
||||
// source; embedded so the harness is self-contained.
|
||||
//
|
||||
//go:embed testdata/cloud-init-witness.yaml
|
||||
var cloudInitWitness []byte
|
||||
|
||||
// gatewayPort is the AWS gateway port on each cluster node. Matches
|
||||
// tests/e2e/run-multinode-e2e.sh's AWSGW_PORT.
|
||||
const gatewayPort = 9999
|
||||
|
||||
// Witness bundles the AWS SDK client, guest/host SSH credentials, and AMI
|
||||
// selection needed to launch and interrogate counter VMs. One Witness is
|
||||
// shared across the scenarios in a suite.
|
||||
type Witness struct {
|
||||
ec2 *ec2.EC2
|
||||
cluster *Cluster
|
||||
ssh SSH
|
||||
hostSigner ssh.Signer
|
||||
guestSigner ssh.Signer
|
||||
guestUser string
|
||||
ami string
|
||||
instanceType string
|
||||
keyName string
|
||||
}
|
||||
|
||||
// NewWitness constructs a Witness from environment-supplied settings:
|
||||
//
|
||||
// - AWS_REGION (required) — passed to the AWS SDK client.
|
||||
// - DDIL_WITNESS_AMI (optional) — specific AMI ID; if unset, an
|
||||
// ami-ubuntu-* image is discovered at launch time.
|
||||
// - DDIL_WITNESS_INSTANCE_TYPE (default t2.micro) — instance type.
|
||||
// - DDIL_WITNESS_KEY_NAME (default spinifex-key) — EC2 key pair name.
|
||||
// - DDIL_GUEST_SSH_USER (default ubuntu) — user for guest SSH.
|
||||
// - DDIL_GUEST_SSH_KEY (default cluster.SSHKeyPath) — private key for
|
||||
// guest SSH; may point at the same key baked into the AMI.
|
||||
//
|
||||
// Credentials for the SDK come from the default chain (AWS_ACCESS_KEY_ID/
|
||||
// AWS_SECRET_ACCESS_KEY or shared profile), matching the tofu-cluster
|
||||
// convention used by the shell E2E suites.
|
||||
func NewWitness(cluster *Cluster, transport SSH) (*Witness, error) {
|
||||
region := os.Getenv("AWS_REGION")
|
||||
if region == "" {
|
||||
return nil, errors.New("ddil harness: NewWitness requires AWS_REGION")
|
||||
}
|
||||
if len(cluster.Nodes) == 0 {
|
||||
return nil, errors.New("ddil harness: NewWitness requires a non-empty cluster")
|
||||
}
|
||||
|
||||
endpoint := "https://" + net.JoinHostPort(cluster.Nodes[0].Addr, strconv.Itoa(gatewayPort))
|
||||
sess, err := session.NewSession(&aws.Config{
|
||||
Region: aws.String(region),
|
||||
Endpoint: aws.String(endpoint),
|
||||
HTTPClient: &http.Client{
|
||||
Timeout: 30 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // self-signed test certs
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: aws session: %w", err)
|
||||
}
|
||||
|
||||
hostSigner, err := loadSigner(cluster.SSHKeyPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: host ssh key: %w", err)
|
||||
}
|
||||
|
||||
guestKeyPath := envDefault("DDIL_GUEST_SSH_KEY", cluster.SSHKeyPath)
|
||||
guestSigner, err := loadSigner(guestKeyPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: guest ssh key: %w", err)
|
||||
}
|
||||
|
||||
return &Witness{
|
||||
ec2: ec2.New(sess),
|
||||
cluster: cluster,
|
||||
ssh: transport,
|
||||
hostSigner: hostSigner,
|
||||
guestSigner: guestSigner,
|
||||
guestUser: envDefault("DDIL_GUEST_SSH_USER", "ubuntu"),
|
||||
ami: os.Getenv("DDIL_WITNESS_AMI"),
|
||||
instanceType: envDefault("DDIL_WITNESS_INSTANCE_TYPE", "t2.micro"),
|
||||
keyName: envDefault("DDIL_WITNESS_KEY_NAME", "spinifex-key"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// WitnessVM is the live descriptor for a single counter VM. Created by
|
||||
// LaunchWitnessVM; subsequent ReadCounter / AssertProgressed calls use the
|
||||
// back-reference to the owning Witness to avoid passing the SDK/SSH deps
|
||||
// through every call site.
|
||||
type WitnessVM struct {
|
||||
InstanceID string
|
||||
HostNode Node // cluster node the VM's QEMU process lives on
|
||||
SSHPort int // QEMU hostfwd port on HostNode mapped to guest :22
|
||||
LaunchedAt time.Time
|
||||
BaselineCounter int
|
||||
|
||||
w *Witness
|
||||
}
|
||||
|
||||
// LaunchWitnessVM launches a counter VM and waits for its QEMU process to
|
||||
// appear on host. The returned WitnessVM carries the host/port it was placed
|
||||
// on (which may differ from host if the scheduler rejected the hint — a
|
||||
// best-effort retry up to maxPlacementAttempts is performed before giving up).
|
||||
//
|
||||
// Depends on ami-ubuntu-* being registered in the cluster (the tofu-cluster
|
||||
// bootstrap handles this) and on the AWS credential chain resolving, so
|
||||
// scenarios gate on t.Skip when neither is available.
|
||||
func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, error) {
|
||||
const maxPlacementAttempts = 3
|
||||
|
||||
ami, err := w.resolveAMI(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
userData := base64.StdEncoding.EncodeToString(cloudInitWitness)
|
||||
|
||||
var lastErr error
|
||||
for attempt := 1; attempt <= maxPlacementAttempts; attempt++ {
|
||||
out, err := w.ec2.RunInstancesWithContext(ctx, &ec2.RunInstancesInput{
|
||||
ImageId: aws.String(ami),
|
||||
InstanceType: aws.String(w.instanceType),
|
||||
MinCount: aws.Int64(1),
|
||||
MaxCount: aws.Int64(1),
|
||||
KeyName: aws.String(w.keyName),
|
||||
UserData: aws.String(userData),
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: RunInstances: %w", err)
|
||||
}
|
||||
if len(out.Instances) == 0 || out.Instances[0].InstanceId == nil {
|
||||
return nil, errors.New("ddil harness: RunInstances returned no instance")
|
||||
}
|
||||
id := aws.StringValue(out.Instances[0].InstanceId)
|
||||
|
||||
if err := w.waitForRunning(ctx, id, 2*time.Minute); err != nil {
|
||||
_ = w.terminate(ctx, id)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
placed, port, err := w.findHost(ctx, id)
|
||||
if err != nil {
|
||||
_ = w.terminate(ctx, id)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if placed.Index == host.Index {
|
||||
baseline, err := readCounterViaTunnel(ctx, w, placed, port)
|
||||
if err != nil {
|
||||
// The counter service may not have started yet on a freshly
|
||||
// booted guest; a zero baseline is the expected steady state
|
||||
// since the service seeds /var/lib/counter with 0 on first
|
||||
// ExecStartPre. Treat a connection failure as fatal so
|
||||
// scenarios don't silently skip the progress assertion.
|
||||
_ = w.terminate(ctx, id)
|
||||
return nil, fmt.Errorf("ddil harness: witness baseline read: %w", err)
|
||||
}
|
||||
return &WitnessVM{
|
||||
InstanceID: id,
|
||||
HostNode: placed,
|
||||
SSHPort: port,
|
||||
LaunchedAt: time.Now(),
|
||||
BaselineCounter: baseline,
|
||||
w: w,
|
||||
}, nil
|
||||
}
|
||||
|
||||
lastErr = fmt.Errorf("witness landed on %s, wanted %s", placed.Name, host.Name)
|
||||
_ = w.terminate(ctx, id)
|
||||
}
|
||||
return nil, fmt.Errorf("ddil harness: LaunchWitnessVM: %w after %d attempts", lastErr, maxPlacementAttempts)
|
||||
}
|
||||
|
||||
// ReadCounter SSHes into the guest via its host's QEMU hostfwd port and
|
||||
// returns the current /var/lib/counter value.
|
||||
func (v *WitnessVM) ReadCounter(ctx context.Context) (int, error) {
|
||||
return readCounterViaTunnel(ctx, v.w, v.HostNode, v.SSHPort)
|
||||
}
|
||||
|
||||
// AssertProgressed fails t if the counter has not advanced beyond the value
|
||||
// captured at launch. Uses t.Helper so the caller's line number lands in the
|
||||
// failure message.
|
||||
func AssertProgressed(ctx context.Context, t *testing.T, v *WitnessVM) {
|
||||
t.Helper()
|
||||
current, err := v.ReadCounter(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("ddil harness: read witness counter on %s (%s): %v", v.HostNode.Name, v.InstanceID, err)
|
||||
}
|
||||
if current <= v.BaselineCounter {
|
||||
t.Fatalf("ddil harness: witness %s on %s did not progress: baseline=%d current=%d",
|
||||
v.InstanceID, v.HostNode.Name, v.BaselineCounter, current)
|
||||
}
|
||||
t.Logf("witness %s on %s progressed %d → %d", v.InstanceID, v.HostNode.Name, v.BaselineCounter, current)
|
||||
}
|
||||
|
||||
// --- internals ------------------------------------------------------------
|
||||
|
||||
func (w *Witness) resolveAMI(ctx context.Context) (string, error) {
|
||||
if w.ami != "" {
|
||||
return w.ami, nil
|
||||
}
|
||||
out, err := w.ec2.DescribeImagesWithContext(ctx, &ec2.DescribeImagesInput{
|
||||
Filters: []*ec2.Filter{{
|
||||
Name: aws.String("name"),
|
||||
Values: []*string{aws.String("ami-ubuntu-*")},
|
||||
}},
|
||||
})
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("ddil harness: DescribeImages: %w", err)
|
||||
}
|
||||
if len(out.Images) == 0 || out.Images[0].ImageId == nil {
|
||||
return "", errors.New("ddil harness: no ami-ubuntu-* images registered in the cluster")
|
||||
}
|
||||
w.ami = aws.StringValue(out.Images[0].ImageId)
|
||||
return w.ami, nil
|
||||
}
|
||||
|
||||
func (w *Witness) waitForRunning(ctx context.Context, id string, timeout time.Duration) error {
|
||||
deadline := time.Now().Add(timeout)
|
||||
const interval = 2 * time.Second
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
out, err := w.ec2.DescribeInstancesWithContext(ctx, &ec2.DescribeInstancesInput{
|
||||
InstanceIds: []*string{aws.String(id)},
|
||||
})
|
||||
if err == nil {
|
||||
for _, r := range out.Reservations {
|
||||
for _, inst := range r.Instances {
|
||||
if inst.State != nil && aws.StringValue(inst.State.Name) == "running" {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return fmt.Errorf("ddil harness: instance %s did not reach running within %s", id, timeout)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var hostfwdPortRE = regexp.MustCompile(`hostfwd=tcp:[^:]*:(\d+)-:22`)
|
||||
|
||||
// findHost walks every cluster node looking for the QEMU process that owns
|
||||
// instanceID. Returns the hosting node and the SSH hostfwd port extracted
|
||||
// from its command line. Matches the shell helper pattern in
|
||||
// run-multinode-e2e.sh:163-180.
|
||||
func (w *Witness) findHost(ctx context.Context, instanceID string) (Node, int, error) {
|
||||
cmd := fmt.Sprintf("ps auxw | grep %s | grep qemu-system | grep -v grep", shellQuote(instanceID))
|
||||
// Give the daemon a short window to actually spawn QEMU after the
|
||||
// EC2 state flip to running, since /aws/ec2 reports "running" before
|
||||
// the daemon has finished forking the process on some nodes.
|
||||
deadline := time.Now().Add(30 * time.Second)
|
||||
const interval = 1 * time.Second
|
||||
for {
|
||||
for _, n := range w.cluster.Nodes {
|
||||
out, err := w.ssh.Run(ctx, n, cmd)
|
||||
if err != nil {
|
||||
// `grep | grep` exits 1 when no match — our SSH wrapper treats
|
||||
// that as error; try the next node rather than bailing.
|
||||
continue
|
||||
}
|
||||
m := hostfwdPortRE.FindStringSubmatch(string(out))
|
||||
if len(m) == 2 {
|
||||
port, err := strconv.Atoi(m[1])
|
||||
if err == nil {
|
||||
return n, port, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
return Node{}, 0, fmt.Errorf("ddil harness: could not locate QEMU host for %s across %d nodes", instanceID, len(w.cluster.Nodes))
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return Node{}, 0, ctx.Err()
|
||||
case <-time.After(interval):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Witness) terminate(ctx context.Context, id string) error {
|
||||
_, err := w.ec2.TerminateInstancesWithContext(ctx, &ec2.TerminateInstancesInput{
|
||||
InstanceIds: []*string{aws.String(id)},
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("ddil harness: TerminateInstances %s: %w", id, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// readCounterViaTunnel opens an SSH connection to host, tunnels a TCP
|
||||
// channel to 127.0.0.1:port (QEMU's guest hostfwd), runs a fresh SSH
|
||||
// handshake over that channel using the guest credentials, and reads
|
||||
// /var/lib/counter.
|
||||
//
|
||||
// The hostfwd target is 127.0.0.1 because the daemon binds QEMU's user-mode
|
||||
// networking to loopback; the orchestrator cannot reach it directly without
|
||||
// the host SSH relay.
|
||||
func readCounterViaTunnel(ctx context.Context, w *Witness, host Node, port int) (int, error) {
|
||||
hostClient, err := dialHost(ctx, host, w.cluster.SSHUser, w.hostSigner)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer func() { _ = hostClient.Close() }()
|
||||
|
||||
guestAddr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port))
|
||||
tunnel, err := hostClient.DialContext(ctx, "tcp", guestAddr)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ddil harness: tunnel %s → %s: %w", host.Name, guestAddr, err)
|
||||
}
|
||||
|
||||
guestCfg := &ssh.ClientConfig{
|
||||
User: w.guestUser,
|
||||
Auth: []ssh.AuthMethod{ssh.PublicKeys(w.guestSigner)},
|
||||
HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec // ephemeral test VM
|
||||
Timeout: 10 * time.Second,
|
||||
}
|
||||
conn, chans, reqs, err := ssh.NewClientConn(tunnel, guestAddr, guestCfg)
|
||||
if err != nil {
|
||||
_ = tunnel.Close()
|
||||
return 0, fmt.Errorf("ddil harness: guest ssh handshake on %s: %w", guestAddr, err)
|
||||
}
|
||||
guestClient := ssh.NewClient(conn, chans, reqs)
|
||||
defer func() { _ = guestClient.Close() }()
|
||||
|
||||
session, err := guestClient.NewSession()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ddil harness: guest ssh session: %w", err)
|
||||
}
|
||||
defer func() { _ = session.Close() }()
|
||||
|
||||
raw, err := session.CombinedOutput("cat /var/lib/counter")
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ddil harness: read /var/lib/counter: %w (output: %s)", err, strings.TrimSpace(string(raw)))
|
||||
}
|
||||
s := strings.TrimSpace(string(raw))
|
||||
if s == "" {
|
||||
return 0, nil
|
||||
}
|
||||
n, err := strconv.Atoi(s)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("ddil harness: parse /var/lib/counter %q: %w", s, err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func dialHost(ctx context.Context, node Node, user string, signer ssh.Signer) (*ssh.Client, error) {
|
||||
cfg := &ssh.ClientConfig{
|
||||
User: user,
|
||||
Auth: []ssh.AuthMethod{ssh.PublicKeys(signer)},
|
||||
HostKeyCallback: ssh.InsecureIgnoreHostKey(), //nolint:gosec // tofu-cluster test hosts
|
||||
Timeout: 10 * time.Second,
|
||||
}
|
||||
d := net.Dialer{Timeout: cfg.Timeout}
|
||||
conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(node.Addr, "22"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ddil harness: dial host %s: %w", node.Name, err)
|
||||
}
|
||||
c, chans, reqs, err := ssh.NewClientConn(conn, node.Addr, cfg)
|
||||
if err != nil {
|
||||
_ = conn.Close()
|
||||
return nil, fmt.Errorf("ddil harness: host ssh handshake %s: %w", node.Name, err)
|
||||
}
|
||||
return ssh.NewClient(c, chans, reqs), nil
|
||||
}
|
||||
|
||||
func loadSigner(path string) (ssh.Signer, error) {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read %s: %w", path, err)
|
||||
}
|
||||
s, err := ssh.ParsePrivateKey(b)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("parse %s: %w", path, err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func envDefault(name, fallback string) string {
|
||||
if v := os.Getenv(name); v != "" {
|
||||
return v
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioA_NATSKill — stop spinifex-nats on a single node without
|
||||
// touching spinifex-daemon, verify the daemon stays up in standalone
|
||||
// mode, serves its local API, and the surviving 2-node quorum keeps
|
||||
// serving cluster-wide requests. See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario A.
|
||||
func TestScenarioA_NATSKill(t *testing.T) {
|
||||
scenarioSkip(t, "A", "daemon-local-autonomy §1a/1b/1c")
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioB_DaemonRestartWithoutNATS — kill spinifex-nats, restart
|
||||
// spinifex-daemon, verify the daemon starts within 30s (not the old
|
||||
// 5-minute NATS-wait abort) and recovers its instances from the local
|
||||
// state file. See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario B.
|
||||
func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) {
|
||||
scenarioSkip(t, "B", "daemon-local-autonomy §1a/1d")
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioC_CleanPartition — iptables-DROP node3 away from node1 and
|
||||
// node2, verify the majority keeps serving API, the isolated node
|
||||
// reports standalone mode, and heal converges state without duplicate
|
||||
// or orphaned VMs. See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario C.
|
||||
func TestScenarioC_CleanPartition(t *testing.T) {
|
||||
scenarioSkip(t, "C", "daemon-local-autonomy §1a-1e")
|
||||
}
|
||||
@@ -0,0 +1,169 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import (
|
||||
"go/ast"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// scenarioFuncRe matches the TestScenario<L>_<Name> functions that make
|
||||
// up the DDIL suite. The capture group is the single-letter identifier
|
||||
// (A..F) used everywhere in the plan, bead, and TEST_COVERAGE.md.
|
||||
var scenarioFuncRe = regexp.MustCompile(`^TestScenario([A-Z])_[A-Za-z0-9]+$`)
|
||||
|
||||
// coverageDocRe matches the scenario-letter column in the DDIL table in
|
||||
// TEST_COVERAGE.md (e.g. `| A | ...`). Anchoring on a leading pipe +
|
||||
// whitespace keeps the pattern from matching prose mentions of "A" or
|
||||
// "B" elsewhere in the doc.
|
||||
var coverageDocRe = regexp.MustCompile(`(?m)^\|\s*([A-F])\s*\|`)
|
||||
|
||||
// TestCoverageDrift enforces TEST_COVERAGE.md and the scenarios package
|
||||
// stay in lock-step. It fails if either:
|
||||
// - A TestScenario<L>_... function exists in the scenarios package but
|
||||
// TEST_COVERAGE.md's DDIL table has no row for letter <L>.
|
||||
// - TEST_COVERAGE.md's DDIL table has a row for letter <L> but no
|
||||
// TestScenario<L>_... function exists.
|
||||
//
|
||||
// Drift surfaces the most common mistake in rolling out Phase 2/3
|
||||
// hardening: flipping a scenario's Skip to a real assertion without
|
||||
// updating the coverage table (or vice versa).
|
||||
//
|
||||
// Runs unconditionally — including under DDIL_DRY_RUN=1 — because it
|
||||
// only reads local source and has no cluster dependency.
|
||||
func TestCoverageDrift(t *testing.T) {
|
||||
scenarioDir := callerDir(t)
|
||||
|
||||
codeLetters, err := lettersFromSource(scenarioDir)
|
||||
if err != nil {
|
||||
t.Fatalf("extract scenario letters from source: %v", err)
|
||||
}
|
||||
if len(codeLetters) == 0 {
|
||||
t.Fatalf("no TestScenario<L>_... functions found in %s — coverage drift check is meaningful only with scenarios present", scenarioDir)
|
||||
}
|
||||
|
||||
docPath := filepath.Join(scenarioDir, "..", "..", "TEST_COVERAGE.md")
|
||||
docLetters, err := lettersFromDoc(docPath)
|
||||
if err != nil {
|
||||
t.Fatalf("extract scenario letters from %s: %v", docPath, err)
|
||||
}
|
||||
|
||||
missing := setDiff(codeLetters, docLetters)
|
||||
extra := setDiff(docLetters, codeLetters)
|
||||
if len(missing) == 0 && len(extra) == 0 {
|
||||
return
|
||||
}
|
||||
if len(missing) > 0 {
|
||||
t.Errorf("scenarios present in code but missing from %s DDIL table: %v", filepath.Base(docPath), missing)
|
||||
}
|
||||
if len(extra) > 0 {
|
||||
t.Errorf("scenarios present in %s DDIL table but missing TestScenario<L>_... in code: %v", filepath.Base(docPath), extra)
|
||||
}
|
||||
}
|
||||
|
||||
// callerDir returns the directory holding this test file. Used to locate
|
||||
// sibling *_test.go files and to compute the relative path to
|
||||
// TEST_COVERAGE.md without hard-coding a fragile relative path at the
|
||||
// test-binary CWD.
|
||||
func callerDir(t *testing.T) string {
|
||||
t.Helper()
|
||||
_, self, _, ok := runtime.Caller(0)
|
||||
if !ok {
|
||||
t.Fatalf("runtime.Caller(0) failed — cannot locate scenarios dir")
|
||||
}
|
||||
return filepath.Dir(self)
|
||||
}
|
||||
|
||||
// lettersFromSource AST-parses every *_test.go file in dir (except this
|
||||
// file and main_test.go, which hold no scenarios) and returns the
|
||||
// deduplicated sorted set of scenario letters declared by top-level
|
||||
// TestScenario<L>_... functions.
|
||||
func lettersFromSource(dir string) ([]string, error) {
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
fset := token.NewFileSet()
|
||||
seen := make(map[string]struct{})
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
if e.IsDir() || !strings.HasSuffix(name, "_test.go") {
|
||||
continue
|
||||
}
|
||||
if name == "coverage_drift_test.go" || name == "main_test.go" {
|
||||
continue
|
||||
}
|
||||
path := filepath.Join(dir, name)
|
||||
f, err := parser.ParseFile(fset, path, nil, parser.SkipObjectResolution)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, decl := range f.Decls {
|
||||
fn, ok := decl.(*ast.FuncDecl)
|
||||
if !ok || fn.Recv != nil {
|
||||
continue
|
||||
}
|
||||
m := scenarioFuncRe.FindStringSubmatch(fn.Name.Name)
|
||||
if m == nil {
|
||||
continue
|
||||
}
|
||||
seen[m[1]] = struct{}{}
|
||||
}
|
||||
}
|
||||
return sortedKeys(seen), nil
|
||||
}
|
||||
|
||||
// lettersFromDoc reads TEST_COVERAGE.md and returns the deduplicated
|
||||
// sorted set of scenario letters its DDIL table declares. Rows are
|
||||
// matched by coverageDocRe so prose mentions elsewhere in the file do
|
||||
// not trigger false positives.
|
||||
func lettersFromDoc(path string) ([]string, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
seen := make(map[string]struct{})
|
||||
for _, m := range coverageDocRe.FindAllStringSubmatch(string(data), -1) {
|
||||
seen[m[1]] = struct{}{}
|
||||
}
|
||||
return sortedKeys(seen), nil
|
||||
}
|
||||
|
||||
func sortedKeys(m map[string]struct{}) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// setDiff returns elements in a that are not in b. Both inputs must be
|
||||
// sorted (they always are, coming from sortedKeys); the linear merge
|
||||
// keeps the drift check O(n+m) rather than O(nm).
|
||||
func setDiff(a, b []string) []string {
|
||||
var out []string
|
||||
i, j := 0, 0
|
||||
for i < len(a) && j < len(b) {
|
||||
switch {
|
||||
case a[i] < b[j]:
|
||||
out = append(out, a[i])
|
||||
i++
|
||||
case a[i] > b[j]:
|
||||
j++
|
||||
default:
|
||||
i++
|
||||
j++
|
||||
}
|
||||
}
|
||||
out = append(out, a[i:]...)
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioD_DegradedLink — apply the SATCOM netem profile to every
|
||||
// node, verify fan-out queries return complete results and Raft does not
|
||||
// enter continuous leader election. See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario D.
|
||||
func TestScenarioD_DegradedLink(t *testing.T) {
|
||||
scenarioSkip(t, "D", "ddil-concept-demonstrator §1.2 + predastore-ddil-hardening §1")
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioE_PredastoreWriteUnderPartition — partition node3, put a
|
||||
// 10MB S3 object from the majority side, verify PUT succeeds with the
|
||||
// missing-shard repair journal capturing node3's parity, then heal and
|
||||
// verify journal drain + cross-node reads. See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario E.
|
||||
func TestScenarioE_PredastoreWriteUnderPartition(t *testing.T) {
|
||||
scenarioSkip(t, "E", "predastore-ddil-hardening §2")
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
//go:build e2e
|
||||
|
||||
package scenarios
|
||||
|
||||
import "testing"
|
||||
|
||||
// TestScenarioF_RaftUnderSATCOM — apply the SATCOM netem profile
|
||||
// cluster-wide and verify Raft leader elections remain ≤1 over a
|
||||
// 5-minute window (baselined against LAN, where elections thrash). See
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md §3 Scenario F.
|
||||
func TestScenarioF_RaftUnderSATCOM(t *testing.T) {
|
||||
scenarioSkip(t, "F", "predastore-ddil-hardening §1")
|
||||
}
|
||||
@@ -0,0 +1,165 @@
|
||||
//go:build e2e
|
||||
|
||||
// Package scenarios holds the DDIL E2E scenario tests. Each scenario
|
||||
// (A..F) exercises one failure mode documented in
|
||||
// docs/development/improvements/ddil-e2e-test-harness.md. Scenarios start
|
||||
// as t.Skip stubs and are flipped to real assertions by the hardening
|
||||
// epics that make them runnable (daemon-local-autonomy,
|
||||
// predastore-ddil-hardening).
|
||||
package scenarios
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/mulgadc/spinifex/tests/e2e/ddil/harness"
|
||||
)
|
||||
|
||||
// scenarioLetters is the canonical ordered list of scenarios the suite
|
||||
// knows about. It is used for DDIL_DRY_RUN=1 logging and by
|
||||
// TestCoverageDrift as the source-of-truth set when cross-checking
|
||||
// TEST_COVERAGE.md. Keep in sync with the TestScenario<L>_* functions in
|
||||
// this package — TestCoverageDrift fails loudly if it drifts.
|
||||
var scenarioLetters = []string{"A", "B", "C", "D", "E", "F"}
|
||||
|
||||
// Package-level state shared between TestMain's signal handler and
|
||||
// TestHarnessSmoke. Populated by setupCluster when DDIL_NODES is set and
|
||||
// DDIL_DRY_RUN is unset; both remain nil otherwise.
|
||||
var (
|
||||
clusterOnce sync.Once
|
||||
cluster *harness.Cluster
|
||||
sshXport harness.SSH
|
||||
clusterErr error
|
||||
)
|
||||
|
||||
// dryRun reports whether DDIL_DRY_RUN=1 is set. Scenarios and
|
||||
// TestHarnessSmoke skip cluster-touching work in dry-run mode.
|
||||
func dryRun() bool { return os.Getenv("DDIL_DRY_RUN") == "1" }
|
||||
|
||||
// setupCluster builds the Cluster/SSH pair once per test binary. Callers
|
||||
// that need cluster access gate on both (cluster, err): cluster is nil in
|
||||
// dry-run or when DDIL_NODES is unset.
|
||||
func setupCluster() (*harness.Cluster, harness.SSH, error) {
|
||||
clusterOnce.Do(func() {
|
||||
if dryRun() {
|
||||
return
|
||||
}
|
||||
if os.Getenv("DDIL_NODES") == "" {
|
||||
return
|
||||
}
|
||||
c, err := harness.ClusterFromEnv()
|
||||
if err != nil {
|
||||
clusterErr = err
|
||||
return
|
||||
}
|
||||
s, err := harness.NewSSH(c)
|
||||
if err != nil {
|
||||
clusterErr = err
|
||||
return
|
||||
}
|
||||
cluster = c
|
||||
sshXport = s
|
||||
})
|
||||
return cluster, sshXport, clusterErr
|
||||
}
|
||||
|
||||
// TestMain is the suite-level entry point. It installs a signal handler
|
||||
// that best-effort resets the cluster on SIGINT/SIGTERM so an aborted CI
|
||||
// run does not leave iptables DROP rules or tc netem qdiscs in place,
|
||||
// then delegates to go test's default runner. DDIL_DRY_RUN=1 logs the
|
||||
// planned scenarios and skips cluster initialisation, leaving
|
||||
// TestCoverageDrift as the only meaningful assertion.
|
||||
func TestMain(m *testing.M) {
|
||||
if dryRun() {
|
||||
log.Printf("ddil: DDIL_DRY_RUN=1 — planned scenarios: %v (TestCoverageDrift will still execute)", scenarioLetters)
|
||||
}
|
||||
|
||||
c, s, err := setupCluster()
|
||||
if err != nil {
|
||||
log.Printf("ddil: cluster setup failed (continuing, scenarios will skip): %v", err)
|
||||
}
|
||||
|
||||
stop := installSignalHandler(c, s)
|
||||
defer stop()
|
||||
|
||||
code := m.Run()
|
||||
if s != nil {
|
||||
_ = s.Close()
|
||||
}
|
||||
os.Exit(code)
|
||||
}
|
||||
|
||||
// installSignalHandler traps SIGINT/SIGTERM and, when a cluster is
|
||||
// available, calls ResetAllNodes before exiting non-zero. Without a
|
||||
// cluster (dry-run or missing env) it still exits non-zero so the parent
|
||||
// shell sees the signal. Returns a stop function the caller must defer to
|
||||
// unregister the handler.
|
||||
func installSignalHandler(c *harness.Cluster, s harness.SSH) func() {
|
||||
ch := make(chan os.Signal, 1)
|
||||
signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
select {
|
||||
case sig := <-ch:
|
||||
log.Printf("ddil: received %s, cleaning up before exit", sig)
|
||||
if c != nil && s != nil {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
if err := harness.ResetAllNodes(ctx, c, s); err != nil {
|
||||
log.Printf("ddil: ResetAllNodes on signal: %v", err)
|
||||
}
|
||||
_ = s.Close()
|
||||
}
|
||||
os.Exit(1)
|
||||
case <-done:
|
||||
return
|
||||
}
|
||||
}()
|
||||
return func() {
|
||||
signal.Stop(ch)
|
||||
close(done)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHarnessSmoke exercises the reset + clean-state round trip without
|
||||
// running a scenario. It satisfies the Phase 1 acceptance criterion that
|
||||
// "full run against real cluster completes with AssertCleanState +
|
||||
// ResetAllNodes round-tripping cleanly" and gives hardening PR authors a
|
||||
// fast command for validating their tofu-cluster before touching
|
||||
// scenarios (go test -run TestHarnessSmoke).
|
||||
func TestHarnessSmoke(t *testing.T) {
|
||||
if dryRun() {
|
||||
t.Skipf("DDIL_DRY_RUN=1")
|
||||
}
|
||||
c, s, err := setupCluster()
|
||||
if err != nil {
|
||||
t.Fatalf("cluster setup: %v", err)
|
||||
}
|
||||
if c == nil || s == nil {
|
||||
t.Skipf("DDIL_NODES unset (harness smoke requires a live cluster)")
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
if err := harness.ResetAllNodes(ctx, c, s); err != nil {
|
||||
t.Fatalf("ResetAllNodes: %v", err)
|
||||
}
|
||||
harness.AssertCleanState(ctx, t, c, s)
|
||||
}
|
||||
|
||||
// scenarioSkip is the shared skip helper used by each TestScenario<L>_*
|
||||
// stub. Centralising the message format keeps Phase 2/3 hardening PRs to
|
||||
// a single place when they flip a scenario: delete the scenarioSkip call,
|
||||
// add the real body.
|
||||
func scenarioSkip(t *testing.T, letter, dep string) {
|
||||
t.Helper()
|
||||
t.Skipf("scenario %s requires %s", letter, dep)
|
||||
}
|
||||
Reference in New Issue
Block a user