From 06ca05d6119fef6b9ccedfb3a0f7c5288708f083 Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Mon, 18 May 2026 15:16:03 +1000 Subject: [PATCH] refactor(e2e): hoist generic DDIL harness primitives to shared tests/e2e/harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 1.5 of docs/development/improvements/e2e-go-harness.md. Promotes Cluster/Node/SSH/DaemonClient/LocalStateRevision/ResetAllNodes/Witness into the shared package so the LB, multinode, and BM scenarios in steps 4-5 can build on the same primitives instead of duplicating SSH + cluster plumbing. Moves - ddil/harness/{cluster,ssh,daemon_client,state,reset,witness}.go -> harness/ (plus the witness testdata embed dir). - ddil/harness/{fault,netem,profiles,snapshot}.go -> ddil/fault/ as a new package; fault.PartitionNode/HealNode/KillNATS/StartNATS/ RestartDaemonOnly/ApplyNetem/ClearNetem/LinkProfile*/TakeSnapshot now take harness.Cluster / harness.Node / harness.SSH explicitly. - ddil/harness/retry.go::Run retry-with-reset extracted as harness.RetryWithReset in harness/wait.go; ddil/harness/quarantine.go keeps the DDIL-only quarantine wrapper that calls into it. - WaitForMode moved from fault.go to harness/daemon_client.go since it polls daemon state and has no fault-injection concern. - The pre-hoist harness/ssh.go (PeerSSH, subprocess ssh) is preserved as harness/peer_ssh.go so the lb_test suite keeps working unchanged; the typed x/crypto/ssh client from DDIL now owns harness/ssh.go. - New harness/util.go exports ShellQuote (was unexported shellQuote in netem.go) so both ddil/fault and reset.go/witness.go can use it. Env var rename - SPINIFEX_NODES / SPINIFEX_SSH_USER / SPINIFEX_SSH_KEY replace DDIL_NODES / DDIL_SSH_USER / DDIL_SSH_KEY. Old names still honoured with a stderr deprecation note (envWithLegacy) so the DDIL CI keeps passing until .github/workflows/e2e-ddil.yml drops the old names. DDIL_QUARANTINED stays DDIL-specific. DDIL parity preserved — scenarios still call ddilh.Run for quarantine semantics; the underlying retry-with-reset is the same code path, just shared. Closes mulga-siv-74. --- tests/e2e/ddil/{harness => fault}/fault.go | 77 +++------ tests/e2e/ddil/{harness => fault}/netem.go | 27 ++-- tests/e2e/ddil/{harness => fault}/profiles.go | 2 +- tests/e2e/ddil/{harness => fault}/snapshot.go | 24 +-- .../ddil/harness/{retry.go => quarantine.go} | 33 ++-- tests/e2e/ddil/harness/ssh.go | 128 --------------- tests/e2e/ddil/scenarios/a_nats_kill_test.go | 16 +- .../ddil/scenarios/b_daemon_restart_test.go | 18 ++- tests/e2e/ddil/scenarios/c_partition_test.go | 16 +- tests/e2e/ddil/scenarios/main_test.go | 2 +- tests/e2e/{ddil => }/harness/cluster.go | 50 +++--- tests/e2e/{ddil => }/harness/daemon_client.go | 47 +++++- tests/e2e/harness/peer_ssh.go | 59 +++++++ tests/e2e/{ddil => }/harness/reset.go | 20 +-- tests/e2e/harness/ssh.go | 147 +++++++++++++----- tests/e2e/{ddil => }/harness/state.go | 4 +- .../harness/testdata/cloud-init-witness.yaml | 0 tests/e2e/harness/util.go | 12 ++ tests/e2e/harness/wait.go | 24 +++ tests/e2e/{ddil => }/harness/witness.go | 60 +++---- 20 files changed, 405 insertions(+), 361 deletions(-) rename tests/e2e/ddil/{harness => fault}/fault.go (51%) rename tests/e2e/ddil/{harness => fault}/netem.go (62%) rename tests/e2e/ddil/{harness => fault}/profiles.go (99%) rename tests/e2e/ddil/{harness => fault}/snapshot.go (79%) rename tests/e2e/ddil/harness/{retry.go => quarantine.go} (55%) delete mode 100644 tests/e2e/ddil/harness/ssh.go rename tests/e2e/{ddil => }/harness/cluster.go (56%) rename tests/e2e/{ddil => }/harness/daemon_client.go (74%) create mode 100644 tests/e2e/harness/peer_ssh.go rename tests/e2e/{ddil => }/harness/reset.go (89%) rename tests/e2e/{ddil => }/harness/state.go (83%) rename tests/e2e/{ddil => }/harness/testdata/cloud-init-witness.yaml (100%) create mode 100644 tests/e2e/harness/util.go rename tests/e2e/{ddil => }/harness/witness.go (86%) diff --git a/tests/e2e/ddil/harness/fault.go b/tests/e2e/ddil/fault/fault.go similarity index 51% rename from tests/e2e/ddil/harness/fault.go rename to tests/e2e/ddil/fault/fault.go index 069d1e884..54b1c6ad4 100644 --- a/tests/e2e/ddil/harness/fault.go +++ b/tests/e2e/ddil/fault/fault.go @@ -1,12 +1,18 @@ //go:build e2e -package harness +// Package fault provides DDIL-only fault-injection helpers — iptables-based +// peer partitioning, NATS/daemon lifecycle toggles, netem link-condition +// shaping, and instance-state snapshots used to assert that nothing rewound +// across a partition/heal cycle. Generic cluster/SSH/daemon primitives live +// in github.com/mulgadc/spinifex/tests/e2e/harness. +package fault import ( "context" "fmt" "strings" - "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" ) // PartitionNode isolates target from peers by installing per-peer iptables @@ -20,28 +26,28 @@ import ( // 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 { +func PartitionNode(ctx context.Context, ssh harness.SSH, target harness.Node, peers []harness.Node) error { if len(peers) == 0 { - return fmt.Errorf("ddil harness: PartitionNode %s: no peers supplied", target.Name) + return fmt.Errorf("ddil fault: 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)), + fmt.Sprintf("sudo iptables -I INPUT -s %s -j DROP", harness.ShellQuote(p.Addr)), + fmt.Sprintf("sudo iptables -I OUTPUT -d %s -j DROP", harness.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) + return fmt.Errorf("ddil fault: 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 "+ + return fmt.Errorf("ddil fault: partition %s severed orchestrator SSH "+ "(orchestrator may share peer IPs): %w", target.Name, err) } return nil @@ -49,9 +55,9 @@ func PartitionNode(ctx context.Context, ssh SSH, target Node, peers []Node) erro // 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 { +func HealNode(ctx context.Context, ssh harness.SSH, target harness.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 fmt.Errorf("ddil fault: heal %s: %w", target.Name, err) } return nil } @@ -59,17 +65,17 @@ func HealNode(ctx context.Context, ssh SSH, target Node) error { // 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 { +func KillNATS(ctx context.Context, ssh harness.SSH, node harness.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 fmt.Errorf("ddil fault: 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 { +func StartNATS(ctx context.Context, ssh harness.SSH, node harness.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 fmt.Errorf("ddil fault: start nats on %s: %w", node.Name, err) } return nil } @@ -77,48 +83,9 @@ func StartNATS(ctx context.Context, ssh SSH, node Node) error { // 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 { +func RestartDaemonOnly(ctx context.Context, ssh harness.SSH, node harness.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 fmt.Errorf("ddil fault: 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): - } - } -} diff --git a/tests/e2e/ddil/harness/netem.go b/tests/e2e/ddil/fault/netem.go similarity index 62% rename from tests/e2e/ddil/harness/netem.go rename to tests/e2e/ddil/fault/netem.go index 5acd4db45..ae384c76e 100644 --- a/tests/e2e/ddil/harness/netem.go +++ b/tests/e2e/ddil/fault/netem.go @@ -1,11 +1,13 @@ //go:build e2e -package harness +package fault import ( "context" "fmt" "strings" + + "github.com/mulgadc/spinifex/tests/e2e/harness" ) // ApplyNetem installs a tc netem qdisc on iface of node matching profile. @@ -14,9 +16,9 @@ import ( // 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 { +func ApplyNetem(ctx context.Context, ssh harness.SSH, node harness.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) + return fmt.Errorf("ddil fault: profile %q is flap-driven; use a flap orchestrator, not ApplyNetem", profile.Name) } var netemArgs []string @@ -34,15 +36,15 @@ func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile L netemArgs = append(netemArgs, "rate", profile.Bandwidth) } if len(netemArgs) == 0 { - return fmt.Errorf("ddil harness: profile %q has no netem parameters", profile.Name) + return fmt.Errorf("ddil fault: 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, " ")) + harness.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", + return fmt.Errorf("ddil fault: apply netem %s on %s (%s): %w", profile.Name, node.Name, iface, err) } return nil @@ -50,17 +52,10 @@ func ApplyNetem(ctx context.Context, ssh SSH, node Node, iface string, profile L // 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)) +func ClearNetem(ctx context.Context, ssh harness.SSH, node harness.Node, iface string) error { + cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", harness.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 fmt.Errorf("ddil fault: 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, "'", `'\''`) + "'" -} diff --git a/tests/e2e/ddil/harness/profiles.go b/tests/e2e/ddil/fault/profiles.go similarity index 99% rename from tests/e2e/ddil/harness/profiles.go rename to tests/e2e/ddil/fault/profiles.go index e51b60a26..65eaf8a4f 100644 --- a/tests/e2e/ddil/harness/profiles.go +++ b/tests/e2e/ddil/fault/profiles.go @@ -1,6 +1,6 @@ //go:build e2e -package harness +package fault import "time" diff --git a/tests/e2e/ddil/harness/snapshot.go b/tests/e2e/ddil/fault/snapshot.go similarity index 79% rename from tests/e2e/ddil/harness/snapshot.go rename to tests/e2e/ddil/fault/snapshot.go index ffa20206d..d22a7ff40 100644 --- a/tests/e2e/ddil/harness/snapshot.go +++ b/tests/e2e/ddil/fault/snapshot.go @@ -1,11 +1,13 @@ //go:build e2e -package harness +package fault import ( "context" "fmt" "testing" + + "github.com/mulgadc/spinifex/tests/e2e/harness" ) // InstanceSnapshot captures a point-in-time view of a single daemon's locally @@ -13,11 +15,11 @@ import ( // 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 +// Reuses the LocalInstance type from harness/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 []harness.LocalInstance // TakeSnapshot reads /local/instances on the given node and returns the // response as an InstanceSnapshot. @@ -25,10 +27,10 @@ type InstanceSnapshot []LocalInstance // 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) { +func TakeSnapshot(ctx context.Context, d *harness.DaemonClient, node harness.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 nil, fmt.Errorf("ddil fault: snapshot %s: %w", node.Name, err) } return InstanceSnapshot(xs), nil } @@ -46,7 +48,7 @@ func TakeSnapshot(ctx context.Context, d *DaemonClient, node Node) (InstanceSnap func (pre InstanceSnapshot) AssertPreserved(t *testing.T, post InstanceSnapshot) { t.Helper() - postByID := make(map[string]LocalInstance, len(post)) + postByID := make(map[string]harness.LocalInstance, len(post)) for _, p := range post { postByID[p.InstanceID] = p } @@ -68,10 +70,10 @@ func (pre InstanceSnapshot) AssertPreserved(t *testing.T, post InstanceSnapshot) return } if len(missing) > 0 { - t.Errorf("ddil harness: instances disappeared from snapshot: %v", missing) + t.Errorf("ddil fault: instances disappeared from snapshot: %v", missing) } if len(regressed) > 0 { - t.Errorf("ddil harness: instances regressed to a terminal state: %v", regressed) + t.Errorf("ddil fault: instances regressed to a terminal state: %v", regressed) } t.FailNow() } diff --git a/tests/e2e/ddil/harness/retry.go b/tests/e2e/ddil/harness/quarantine.go similarity index 55% rename from tests/e2e/ddil/harness/retry.go rename to tests/e2e/ddil/harness/quarantine.go index 3de94de57..a9dcefc14 100644 --- a/tests/e2e/ddil/harness/retry.go +++ b/tests/e2e/ddil/harness/quarantine.go @@ -1,13 +1,18 @@ //go:build e2e +// Package harness is DDIL's thin shim over the shared +// github.com/mulgadc/spinifex/tests/e2e/harness primitives. It hosts only the +// quarantine + retry wrapper that DDIL scenarios use; everything generic +// (Cluster, SSH, DaemonClient, ResetAllNodes, Witness, etc.) lives in the +// shared harness package. package harness import ( - "context" "os" "strings" "testing" - "time" + + "github.com/mulgadc/spinifex/tests/e2e/harness" ) // Run wraps a scenario with reset-and-retry plus quarantine handling. It is @@ -20,14 +25,9 @@ import ( // 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)) { +// Retry: delegates to harness.RetryWithReset so the shared package owns the +// reset-and-rerun mechanics; only quarantine policy stays DDIL-specific. +func Run(t *testing.T, c *harness.Cluster, ssh harness.SSH, letter string, fn func(*testing.T)) { t.Helper() if IsQuarantined(letter) { @@ -35,18 +35,7 @@ func Run(t *testing.T, c *Cluster, ssh SSH, letter string, fn func(*testing.T)) 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) + harness.RetryWithReset(t, c, ssh, "scenario "+letter, fn) } // IsQuarantined reports whether letter appears in DDIL_QUARANTINED. diff --git a/tests/e2e/ddil/harness/ssh.go b/tests/e2e/ddil/harness/ssh.go deleted file mode 100644 index 161df8b07..000000000 --- a/tests/e2e/ddil/harness/ssh.go +++ /dev/null @@ -1,128 +0,0 @@ -//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 -} - -var _ SSH = (*sshClient)(nil) - -// 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 -} diff --git a/tests/e2e/ddil/scenarios/a_nats_kill_test.go b/tests/e2e/ddil/scenarios/a_nats_kill_test.go index 412d5e944..f49916651 100644 --- a/tests/e2e/ddil/scenarios/a_nats_kill_test.go +++ b/tests/e2e/ddil/scenarios/a_nats_kill_test.go @@ -7,7 +7,9 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/harness" + ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/ddil/fault" ) // TestScenarioA_NATSKill — stop spinifex-nats on a single node without @@ -19,7 +21,7 @@ func TestScenarioA_NATSKill(t *testing.T) { deps := requireLiveCluster(t) c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness - harness.Run(t, c, ssh, "A", func(t *testing.T) { + ddilh.Run(t, c, ssh, "A", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) defer cancel() @@ -28,12 +30,12 @@ func TestScenarioA_NATSKill(t *testing.T) { node1, node2, node3 := c.Nodes[0], c.Nodes[1], c.Nodes[2] witnesses := launchWitnesses(ctx, t, w, node1, node2, node3) - pre, err := harness.TakeSnapshot(ctx, dc, node3) + pre, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err) } - if err := harness.KillNATS(ctx, ssh, node3); err != nil { + if err := fault.KillNATS(ctx, ssh, node3); err != nil { t.Fatalf("kill nats on %s: %v", node3.Name, err) } // Restore NATS in cleanup so a mid-scenario failure does not leave @@ -41,7 +43,7 @@ func TestScenarioA_NATSKill(t *testing.T) { t.Cleanup(func() { cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute) defer ccancel() - _ = harness.StartNATS(cctx, ssh, node3) + _ = fault.StartNATS(cctx, ssh, node3) }) if err := harness.WaitForMode(ctx, dc, node3, harness.ModeStandalone, 30*time.Second); err != nil { @@ -68,14 +70,14 @@ func TestScenarioA_NATSKill(t *testing.T) { } } - if err := harness.StartNATS(ctx, ssh, node3); err != nil { + if err := fault.StartNATS(ctx, ssh, node3); err != nil { t.Fatalf("restart nats on %s: %v", node3.Name, err) } if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil { t.Fatalf("%s did not return to cluster mode: %v", node3.Name, err) } - post, err := harness.TakeSnapshot(ctx, dc, node3) + post, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("post-heal snapshot on %s: %v", node3.Name, err) } diff --git a/tests/e2e/ddil/scenarios/b_daemon_restart_test.go b/tests/e2e/ddil/scenarios/b_daemon_restart_test.go index 2ae53c0a9..ebaaf237d 100644 --- a/tests/e2e/ddil/scenarios/b_daemon_restart_test.go +++ b/tests/e2e/ddil/scenarios/b_daemon_restart_test.go @@ -7,7 +7,9 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/harness" + ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/ddil/fault" ) // TestScenarioB_DaemonRestartWithoutNATS — kill spinifex-nats, restart @@ -26,7 +28,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) { deps := requireLiveCluster(t) c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness - harness.Run(t, c, ssh, "B", func(t *testing.T) { + ddilh.Run(t, c, ssh, "B", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) defer cancel() @@ -35,7 +37,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) { node3 := c.Nodes[2] _ = launchWitnesses(ctx, t, w, node3) - pre, err := harness.TakeSnapshot(ctx, dc, node3) + pre, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err) } @@ -43,20 +45,20 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) { t.Fatalf("pre-fault /local/instances on %s empty — witness launch did not register", node3.Name) } - if err := harness.KillNATS(ctx, ssh, node3); err != nil { + if err := fault.KillNATS(ctx, ssh, node3); err != nil { t.Fatalf("kill nats on %s: %v", node3.Name, err) } t.Cleanup(func() { cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute) defer ccancel() - _ = harness.StartNATS(cctx, ssh, node3) + _ = fault.StartNATS(cctx, ssh, node3) }) // 1d's promise is "daemon comes up without NATS". The 30s budget is // the plan's threshold — if it slips, the new startup path has // regressed back to the 5-minute NATS-wait abort. restartStart := time.Now() - if err := harness.RestartDaemonOnly(ctx, ssh, node3); err != nil { + if err := fault.RestartDaemonOnly(ctx, ssh, node3); err != nil { t.Fatalf("restart daemon on %s: %v", node3.Name, err) } if err := harness.WaitForMode(ctx, dc, node3, harness.ModeStandalone, 30*time.Second); err != nil { @@ -64,7 +66,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) { } t.Logf("daemon on %s reached standalone in %s", node3.Name, time.Since(restartStart)) - post, err := harness.TakeSnapshot(ctx, dc, node3) + post, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("post-restart snapshot on %s: %v", node3.Name, err) } @@ -80,7 +82,7 @@ func TestScenarioB_DaemonRestartWithoutNATS(t *testing.T) { } } - if err := harness.StartNATS(ctx, ssh, node3); err != nil { + if err := fault.StartNATS(ctx, ssh, node3); err != nil { t.Fatalf("restart nats on %s: %v", node3.Name, err) } if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil { diff --git a/tests/e2e/ddil/scenarios/c_partition_test.go b/tests/e2e/ddil/scenarios/c_partition_test.go index b87a58c4c..e7f171a33 100644 --- a/tests/e2e/ddil/scenarios/c_partition_test.go +++ b/tests/e2e/ddil/scenarios/c_partition_test.go @@ -7,7 +7,9 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/harness" + ddilh "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/ddil/fault" ) // TestScenarioC_CleanPartition — iptables-DROP node3 away from node1 and @@ -19,7 +21,7 @@ func TestScenarioC_CleanPartition(t *testing.T) { deps := requireLiveCluster(t) c, ssh, dc, w := deps.cluster, deps.ssh, deps.dc, deps.witness - harness.Run(t, c, ssh, "C", func(t *testing.T) { + ddilh.Run(t, c, ssh, "C", func(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) defer cancel() @@ -31,12 +33,12 @@ func TestScenarioC_CleanPartition(t *testing.T) { // only node-3. witnesses := launchWitnesses(ctx, t, w, node3) - pre, err := harness.TakeSnapshot(ctx, dc, node3) + pre, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("pre-fault snapshot on %s: %v", node3.Name, err) } - if err := harness.PartitionNode(ctx, ssh, node3, c.Peers(node3)); err != nil { + if err := fault.PartitionNode(ctx, ssh, node3, c.Peers(node3)); err != nil { t.Fatalf("partition %s: %v", node3.Name, err) } // HealNode flushes node3's iptables on cleanup; safe even if a later @@ -44,7 +46,7 @@ func TestScenarioC_CleanPartition(t *testing.T) { t.Cleanup(func() { cctx, ccancel := context.WithTimeout(context.Background(), 1*time.Minute) defer ccancel() - _ = harness.HealNode(cctx, ssh, node3) + _ = fault.HealNode(cctx, ssh, node3) }) // The orchestrator dials peers directly from outside the cluster @@ -74,14 +76,14 @@ func TestScenarioC_CleanPartition(t *testing.T) { // proving QEMU on node3 kept executing despite the network split. harness.AssertProgressed(ctx, t, witnesses[0]) - if err := harness.HealNode(ctx, ssh, node3); err != nil { + if err := fault.HealNode(ctx, ssh, node3); err != nil { t.Fatalf("heal %s: %v", node3.Name, err) } if err := harness.WaitForMode(ctx, dc, node3, harness.ModeCluster, 60*time.Second); err != nil { t.Fatalf("%s did not return to cluster mode after heal: %v", node3.Name, err) } - post, err := harness.TakeSnapshot(ctx, dc, node3) + post, err := fault.TakeSnapshot(ctx, dc, node3) if err != nil { t.Fatalf("post-heal snapshot on %s: %v", node3.Name, err) } diff --git a/tests/e2e/ddil/scenarios/main_test.go b/tests/e2e/ddil/scenarios/main_test.go index 1a2911a0a..b1a397929 100644 --- a/tests/e2e/ddil/scenarios/main_test.go +++ b/tests/e2e/ddil/scenarios/main_test.go @@ -18,7 +18,7 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/tests/e2e/ddil/harness" + "github.com/mulgadc/spinifex/tests/e2e/harness" ) // scenarioLetters is the canonical ordered list of scenarios the suite diff --git a/tests/e2e/ddil/harness/cluster.go b/tests/e2e/harness/cluster.go similarity index 56% rename from tests/e2e/ddil/harness/cluster.go rename to tests/e2e/harness/cluster.go index 507b98a32..43e1d9f86 100644 --- a/tests/e2e/ddil/harness/cluster.go +++ b/tests/e2e/harness/cluster.go @@ -1,10 +1,5 @@ //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 ( @@ -16,7 +11,7 @@ import ( // 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; +// Index is the 1-based position the node occupies in the SPINIFEX_NODES list; // scenarios reference nodes by index (node1, node2, node3) so a fixed // ordering is required. type Node struct { @@ -37,7 +32,7 @@ type Cluster struct { } // Peers returns every node in the cluster except target. The result is used -// by PartitionNode to build per-peer iptables DROP rules. +// by fault.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 { @@ -49,29 +44,29 @@ func (c *Cluster) Peers(target Node) []Node { return out } -// ClusterFromEnv builds a Cluster from DDIL_NODES / DDIL_SSH_USER / -// DDIL_SSH_KEY. Nodes are named node1..nodeN in list order. +// ClusterFromEnv builds a Cluster from SPINIFEX_NODES / SPINIFEX_SSH_USER / +// SPINIFEX_SSH_KEY. Nodes are named node1..nodeN in list order. // -// Returns an error if any required variable is missing or if DDIL_NODES +// Returns an error if any required variable is missing or if SPINIFEX_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") + nodesRaw := envWithLegacy("SPINIFEX_NODES", "DDIL_NODES") + user := envWithLegacy("SPINIFEX_SSH_USER", "DDIL_SSH_USER") + key := envWithLegacy("SPINIFEX_SSH_KEY", "DDIL_SSH_KEY") var missing []string if nodesRaw == "" { - missing = append(missing, "DDIL_NODES") + missing = append(missing, "SPINIFEX_NODES") } if user == "" { - missing = append(missing, "DDIL_SSH_USER") + missing = append(missing, "SPINIFEX_SSH_USER") } if key == "" { - missing = append(missing, "DDIL_SSH_KEY") + missing = append(missing, "SPINIFEX_SSH_KEY") } if len(missing) > 0 { - return nil, fmt.Errorf("ddil harness: missing required env: %s", strings.Join(missing, ", ")) + return nil, fmt.Errorf("e2e harness: missing required env: %s", strings.Join(missing, ", ")) } var nodes []Node @@ -87,7 +82,7 @@ func ClusterFromEnv() (*Cluster, error) { }) } if len(nodes) == 0 { - return nil, errors.New("ddil harness: DDIL_NODES contained no addresses") + return nil, errors.New("e2e harness: SPINIFEX_NODES contained no addresses") } return &Cluster{ @@ -98,7 +93,7 @@ func ClusterFromEnv() (*Cluster, error) { } // NodeFromEnv returns the Nth node (1-based) from the cluster defined by -// DDIL_NODES. Convenience wrapper for single-node helpers; most scenarios +// SPINIFEX_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() @@ -106,7 +101,22 @@ func NodeFromEnv(index int) (Node, error) { 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 Node{}, fmt.Errorf("e2e harness: node index %d out of range (have %d nodes)", index, len(c.Nodes)) } return c.Nodes[index-1], nil } + +// envWithLegacy returns os.Getenv(canonical), falling back to legacy on miss. +// DDIL_* names are accepted for back-compat; rename to SPINIFEX_* once all +// CI workflows are migrated. A one-line stderr note flags the deprecation +// so leftover callers surface in CI logs. +func envWithLegacy(canonical, legacy string) string { + if v := os.Getenv(canonical); v != "" { + return v + } + if v := os.Getenv(legacy); v != "" { + fmt.Fprintf(os.Stderr, "e2e harness: %s is deprecated; set %s instead\n", legacy, canonical) + return v + } + return "" +} diff --git a/tests/e2e/ddil/harness/daemon_client.go b/tests/e2e/harness/daemon_client.go similarity index 74% rename from tests/e2e/ddil/harness/daemon_client.go rename to tests/e2e/harness/daemon_client.go index c788685d8..8c146bd2c 100644 --- a/tests/e2e/ddil/harness/daemon_client.go +++ b/tests/e2e/harness/daemon_client.go @@ -16,6 +16,45 @@ import ( "github.com/mulgadc/spinifex/spinifex/types" ) +// 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("e2e harness: wait for mode %s on %s: timed out after %s: last error: %w", + want, node.Name, timeout, lastErr) + } + return fmt.Errorf("e2e 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): + } + } +} + // daemonPort is the cluster-manager HTTPS port the daemon binds for /health // and /local/* (see spinifex/daemon/daemon.go ClusterManager, configured via // [nodes..daemon].host). 4432 is the deployed default; 8443 is predastore. @@ -84,21 +123,21 @@ func (d *DaemonClient) url(node Node, path string) string { 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) + return fmt.Errorf("e2e 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) + return fmt.Errorf("e2e 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", + return fmt.Errorf("e2e 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 fmt.Errorf("e2e harness: daemon %s %s decode: %w", node.Name, path, err) } return nil } diff --git a/tests/e2e/harness/peer_ssh.go b/tests/e2e/harness/peer_ssh.go new file mode 100644 index 000000000..630cedc5a --- /dev/null +++ b/tests/e2e/harness/peer_ssh.go @@ -0,0 +1,59 @@ +//go:build e2e + +package harness + +import ( + "context" + "fmt" + "os" + "os/exec" + "strings" + "time" +) + +// PeerSSH is a minimal SSH wrapper that shells out to /usr/bin/ssh with the +// flags the bash E2E scripts use. It exists so LB/multinode scenarios can +// drive remote nodes without depending on the DDIL harness. Step 1.5 of +// docs/development/improvements/e2e-go-harness.md replaces this with the +// typed x/crypto/ssh client hoisted from ddil/harness/ssh.go. +type PeerSSH struct { + User string + KeyPath string + Timeout time.Duration +} + +// NewPeerSSH reads SPINIFEX_SSH_KEY (default ~/.ssh/tf-user-ap-southeast-2) +// and SPINIFEX_SSH_USER (default tf-user). Matches the bash peer_ssh helper. +func NewPeerSSH() *PeerSSH { + return &PeerSSH{ + User: getenv("SPINIFEX_SSH_USER", "tf-user"), + KeyPath: getenv("SPINIFEX_SSH_KEY", os.ExpandEnv("$HOME/.ssh/tf-user-ap-southeast-2")), + Timeout: 10 * time.Second, + } +} + +// Run executes cmd on host over SSH and returns combined stdout+stderr. +// Non-zero exit becomes a non-nil error with stderr in the message. +func (p *PeerSSH) Run(ctx context.Context, host, cmd string) ([]byte, error) { + args := []string{ + "-i", p.KeyPath, + "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null", + "-o", fmt.Sprintf("ConnectTimeout=%d", int(p.Timeout.Seconds())), + "-o", "LogLevel=ERROR", + fmt.Sprintf("%s@%s", p.User, host), + cmd, + } + out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput() + if err != nil { + return out, fmt.Errorf("ssh %s@%s: %w: %s", p.User, host, err, strings.TrimSpace(string(out))) + } + return out, nil +} + +// Ping returns nil if `hostname` succeeds over SSH. Used as a probe before +// running real remote tests. +func (p *PeerSSH) Ping(ctx context.Context, host string) error { + _, err := p.Run(ctx, host, "hostname") + return err +} diff --git a/tests/e2e/ddil/harness/reset.go b/tests/e2e/harness/reset.go similarity index 89% rename from tests/e2e/ddil/harness/reset.go rename to tests/e2e/harness/reset.go index adee3de4c..34351d232 100644 --- a/tests/e2e/ddil/harness/reset.go +++ b/tests/e2e/harness/reset.go @@ -49,7 +49,7 @@ func resetNode(ctx context.Context, ssh SSH, n Node) error { // 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) + return fmt.Errorf("e2e harness: reset iptables on %s: %w", n.Name, err) } // Remove netem qdiscs only on interfaces that actually carry one. @@ -60,15 +60,15 @@ func resetNode(ctx context.Context, ssh SSH, n Node) error { return err } for _, iface := range ifaces { - cmd := fmt.Sprintf("sudo tc qdisc del dev %s root 2>/dev/null || true", shellQuote(iface)) + 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) + return fmt.Errorf("e2e 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 fmt.Errorf("e2e harness: start services on %s: %w", n.Name, err) } return nil } @@ -81,7 +81,7 @@ func netemIfaces(ctx context.Context, ssh SSH, n Node) ([]string, error) { // 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 nil, fmt.Errorf("e2e harness: list netem qdiscs on %s: %w", n.Name, err) } return strings.Fields(string(out)), nil } @@ -104,7 +104,7 @@ func waitClusterHealthy(ctx context.Context, c *Cluster, ssh SSH, timeout time.D lastErr = err } if time.Now().After(deadline) { - return fmt.Errorf("ddil harness: cluster did not reach healthy within %s: %w", timeout, lastErr) + return fmt.Errorf("e2e harness: cluster did not reach healthy within %s: %w", timeout, lastErr) } select { case <-ctx.Done(): @@ -164,11 +164,11 @@ func AssertCleanState(ctx context.Context, t *testing.T, c *Cluster, ssh SSH) { 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) + t.Fatalf("e2e 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) + t.Fatalf("e2e 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)) @@ -181,9 +181,9 @@ func AssertCleanState(ctx context.Context, t *testing.T, c *Cluster, ssh SSH) { if len(dirty) == 0 { return } - t.Logf("ddil harness: dirty baseline before scenario — %s; running ResetAllNodes", strings.Join(dirty, "; ")) + t.Logf("e2e 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) + t.Fatalf("e2e harness: reset after dirty baseline: %v", err) } } diff --git a/tests/e2e/harness/ssh.go b/tests/e2e/harness/ssh.go index 630cedc5a..6971f4dfb 100644 --- a/tests/e2e/harness/ssh.go +++ b/tests/e2e/harness/ssh.go @@ -3,57 +3,126 @@ package harness import ( + "bytes" "context" "fmt" + "net" "os" - "os/exec" - "strings" "time" + + "golang.org/x/crypto/ssh" ) -// PeerSSH is a minimal SSH wrapper that shells out to /usr/bin/ssh with the -// flags the bash E2E scripts use. It exists so LB/multinode scenarios can -// drive remote nodes without depending on the DDIL harness. Step 1.5 of -// docs/development/improvements/e2e-go-harness.md replaces this with the -// typed x/crypto/ssh client hoisted from ddil/harness/ssh.go. -type PeerSSH struct { - User string - KeyPath string - Timeout time.Duration +// 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 } -// NewPeerSSH reads SPINIFEX_SSH_KEY (default ~/.ssh/tf-user-ap-southeast-2) -// and SPINIFEX_SSH_USER (default tf-user). Matches the bash peer_ssh helper. -func NewPeerSSH() *PeerSSH { - return &PeerSSH{ - User: getenv("SPINIFEX_SSH_USER", "tf-user"), - KeyPath: getenv("SPINIFEX_SSH_KEY", os.ExpandEnv("$HOME/.ssh/tf-user-ap-southeast-2")), - Timeout: 10 * time.Second, - } +// 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 } -// Run executes cmd on host over SSH and returns combined stdout+stderr. -// Non-zero exit becomes a non-nil error with stderr in the message. -func (p *PeerSSH) Run(ctx context.Context, host, cmd string) ([]byte, error) { - args := []string{ - "-i", p.KeyPath, - "-o", "StrictHostKeyChecking=no", - "-o", "UserKnownHostsFile=/dev/null", - "-o", fmt.Sprintf("ConnectTimeout=%d", int(p.Timeout.Seconds())), - "-o", "LogLevel=ERROR", - fmt.Sprintf("%s@%s", p.User, host), - cmd, - } - out, err := exec.CommandContext(ctx, "ssh", args...).CombinedOutput() +var _ SSH = (*sshClient)(nil) + +// 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 out, fmt.Errorf("ssh %s@%s: %w: %s", p.User, host, err, strings.TrimSpace(string(out))) + return nil, fmt.Errorf("e2e harness: read ssh key %s: %w", cluster.SSHKeyPath, err) } - return out, nil + signer, err := ssh.ParsePrivateKey(keyBytes) + if err != nil { + return nil, fmt.Errorf("e2e 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 } -// Ping returns nil if `hostname` succeeds over SSH. Used as a probe before -// running real remote tests. -func (p *PeerSSH) Ping(ctx context.Context, host string) error { - _, err := p.Run(ctx, host, "hostname") - return err +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("e2e 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("e2e 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("e2e 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("e2e harness: ssh close %s: %w", addr, err) + } + delete(s.conns, addr) + } + return firstErr } diff --git a/tests/e2e/ddil/harness/state.go b/tests/e2e/harness/state.go similarity index 83% rename from tests/e2e/ddil/harness/state.go rename to tests/e2e/harness/state.go index 2dc89ca11..91dd95a39 100644 --- a/tests/e2e/ddil/harness/state.go +++ b/tests/e2e/harness/state.go @@ -16,7 +16,7 @@ import ( 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 0, fmt.Errorf("e2e harness: local state revision %s: %w", node.Name, err) } return s.Revision, nil } @@ -28,6 +28,6 @@ func LocalStateRevision(ctx context.Context, d *DaemonClient, node Node) (uint64 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) + t.Fatalf("e2e harness: local state revision regressed: before=%d after=%d", before, after) } } diff --git a/tests/e2e/ddil/harness/testdata/cloud-init-witness.yaml b/tests/e2e/harness/testdata/cloud-init-witness.yaml similarity index 100% rename from tests/e2e/ddil/harness/testdata/cloud-init-witness.yaml rename to tests/e2e/harness/testdata/cloud-init-witness.yaml diff --git a/tests/e2e/harness/util.go b/tests/e2e/harness/util.go new file mode 100644 index 000000000..0494e9261 --- /dev/null +++ b/tests/e2e/harness/util.go @@ -0,0 +1,12 @@ +//go:build e2e + +package harness + +import "strings" + +// 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, "'", `'\''`) + "'" +} diff --git a/tests/e2e/harness/wait.go b/tests/e2e/harness/wait.go index db36be947..9189dafc9 100644 --- a/tests/e2e/harness/wait.go +++ b/tests/e2e/harness/wait.go @@ -3,6 +3,7 @@ package harness import ( + "context" "testing" "time" ) @@ -44,3 +45,26 @@ func EventuallyErr(t *testing.T, cond func() error, timeout, interval time.Durat time.Sleep(interval) } } + +// RetryWithReset runs fn as an "attempt-1" subtest; on failure it calls +// ResetAllNodes and retries fn as "attempt-2". Both attempts surface in the +// test output so CI logs preserve the transient-vs-deterministic signal — +// the retry is a diagnostic aid, not a way to mask flakes. +// +// Used by DDIL's quarantine wrapper (ddil/harness.Run) and any other scenario +// that wants the same retry-after-reset shape without taking on DDIL's +// quarantine semantics. +func RetryWithReset(t *testing.T, c *Cluster, ssh SSH, label string, fn func(*testing.T)) { + t.Helper() + if t.Run("attempt-1", fn) { + return + } + t.Logf("e2e harness: %s failed first attempt, resetting cluster and retrying", label) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + if err := ResetAllNodes(ctx, c, ssh); err != nil { + t.Errorf("e2e harness: reset before retry of %s: %v", label, err) + return + } + t.Run("attempt-2", fn) +} diff --git a/tests/e2e/ddil/harness/witness.go b/tests/e2e/harness/witness.go similarity index 86% rename from tests/e2e/ddil/harness/witness.go rename to tests/e2e/harness/witness.go index dd1e590a0..de4a184ad 100644 --- a/tests/e2e/ddil/harness/witness.go +++ b/tests/e2e/harness/witness.go @@ -75,10 +75,10 @@ type Witness struct { 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") + return nil, errors.New("e2e harness: NewWitness requires AWS_REGION") } if len(cluster.Nodes) == 0 { - return nil, errors.New("ddil harness: NewWitness requires a non-empty cluster") + return nil, errors.New("e2e harness: NewWitness requires a non-empty cluster") } endpoint := "https://" + net.JoinHostPort(cluster.Nodes[0].Addr, strconv.Itoa(gatewayPort)) @@ -93,21 +93,21 @@ func NewWitness(cluster *Cluster, transport SSH) (*Witness, error) { }, }) if err != nil { - return nil, fmt.Errorf("ddil harness: aws session: %w", err) + return nil, fmt.Errorf("e2e harness: aws session: %w", err) } hostSigner, err := loadSigner(cluster.SSHKeyPath) if err != nil { - return nil, fmt.Errorf("ddil harness: host ssh key: %w", err) + return nil, fmt.Errorf("e2e harness: host ssh key: %w", err) } guestKeyPath := os.Getenv("DDIL_GUEST_SSH_KEY") if guestKeyPath == "" { - return nil, errors.New("ddil harness: NewWitness requires DDIL_GUEST_SSH_KEY (path to the private key paired with DDIL_WITNESS_KEY_NAME's registered material)") + return nil, errors.New("e2e harness: NewWitness requires DDIL_GUEST_SSH_KEY (path to the private key paired with DDIL_WITNESS_KEY_NAME's registered material)") } guestSigner, err := loadSigner(guestKeyPath) if err != nil { - return nil, fmt.Errorf("ddil harness: guest ssh key: %w", err) + return nil, fmt.Errorf("e2e harness: guest ssh key: %w", err) } return &Witness{ @@ -170,10 +170,10 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er UserData: aws.String(userData), }) if err != nil { - return nil, fmt.Errorf("ddil harness: RunInstances: %w", err) + return nil, fmt.Errorf("e2e harness: RunInstances: %w", err) } if len(out.Instances) == 0 || out.Instances[0].InstanceId == nil { - return nil, errors.New("ddil harness: RunInstances returned no instance") + return nil, errors.New("e2e harness: RunInstances returned no instance") } id := aws.StringValue(out.Instances[0].InstanceId) @@ -197,7 +197,7 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er baseline, err := awaitBaseline(ctx, w, placed, publicIP, 2*time.Minute) if err != nil { _ = w.terminate(ctx, id) - return nil, fmt.Errorf("ddil harness: witness baseline read: %w", err) + return nil, fmt.Errorf("e2e harness: witness baseline read: %w", err) } return &WitnessVM{ InstanceID: id, @@ -212,7 +212,7 @@ func LaunchWitnessVM(ctx context.Context, w *Witness, host Node) (*WitnessVM, er 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) + return nil, fmt.Errorf("e2e harness: LaunchWitnessVM: %w after %d attempts", lastErr, maxPlacementAttempts) } // ReadCounter SSHes the guest at its public IP (tunnelled through the @@ -235,10 +235,10 @@ 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) + t.Fatalf("e2e 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", + t.Fatalf("e2e 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) @@ -257,10 +257,10 @@ func (w *Witness) resolveAMI(ctx context.Context) (string, error) { }}, }) if err != nil { - return "", fmt.Errorf("ddil harness: DescribeImages: %w", err) + return "", fmt.Errorf("e2e 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") + return "", errors.New("e2e harness: no ami-ubuntu-* images registered in the cluster") } w.ami = aws.StringValue(out.Images[0].ImageId) return w.ami, nil @@ -279,10 +279,10 @@ func (w *Witness) resolveInstanceType(ctx context.Context) (string, error) { } out, err := w.ec2.DescribeInstanceTypesWithContext(ctx, &ec2.DescribeInstanceTypesInput{}) if err != nil { - return "", fmt.Errorf("ddil harness: DescribeInstanceTypes: %w", err) + return "", fmt.Errorf("e2e harness: DescribeInstanceTypes: %w", err) } if len(out.InstanceTypes) == 0 { - return "", errors.New("ddil harness: cluster registered no instance types") + return "", errors.New("e2e harness: cluster registered no instance types") } sort.Slice(out.InstanceTypes, func(i, j int) bool { a, b := out.InstanceTypes[i], out.InstanceTypes[j] @@ -307,7 +307,7 @@ func (w *Witness) resolveInstanceType(ctx context.Context) (string, error) { }) w.instanceType = aws.StringValue(out.InstanceTypes[0].InstanceType) if w.instanceType == "" { - return "", errors.New("ddil harness: smallest instance type has empty name") + return "", errors.New("e2e harness: smallest instance type has empty name") } return w.instanceType, nil } @@ -332,7 +332,7 @@ func (w *Witness) waitForRunning(ctx context.Context, id string, timeout time.Du } } if time.Now().After(deadline) { - return fmt.Errorf("ddil harness: instance %s did not reach running within %s", id, timeout) + return fmt.Errorf("e2e harness: instance %s did not reach running within %s", id, timeout) } select { case <-ctx.Done(): @@ -348,7 +348,7 @@ func (w *Witness) waitForRunning(ctx context.Context, id string, timeout time.Du // exactly one process per cluster regardless of net mode (user-hostfwd vs // tap-mode bridge). func (w *Witness) findHost(ctx context.Context, instanceID string) (Node, error) { - cmd := fmt.Sprintf("ps auxw | grep %s | grep qemu-system | grep -v grep", shellQuote(instanceID)) + 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. @@ -367,7 +367,7 @@ func (w *Witness) findHost(ctx context.Context, instanceID string) (Node, error) } } if time.Now().After(deadline) { - return Node{}, fmt.Errorf("ddil harness: could not locate QEMU host for %s across %d nodes", instanceID, len(w.cluster.Nodes)) + return Node{}, fmt.Errorf("e2e harness: could not locate QEMU host for %s across %d nodes", instanceID, len(w.cluster.Nodes)) } select { case <-ctx.Done(): @@ -400,7 +400,7 @@ func (w *Witness) resolvePublicIP(ctx context.Context, instanceID string) (strin } } if time.Now().After(deadline) { - return "", fmt.Errorf("ddil harness: instance %s has no PublicIpAddress after %s (default subnet MapPublicIpOnLaunch may be off)", instanceID, 30*time.Second) + return "", fmt.Errorf("e2e harness: instance %s has no PublicIpAddress after %s (default subnet MapPublicIpOnLaunch may be off)", instanceID, 30*time.Second) } select { case <-ctx.Done(): @@ -415,7 +415,7 @@ func (w *Witness) terminate(ctx context.Context, id string) error { InstanceIds: []*string{aws.String(id)}, }) if err != nil { - return fmt.Errorf("ddil harness: TerminateInstances %s: %w", id, err) + return fmt.Errorf("e2e harness: TerminateInstances %s: %w", id, err) } return nil } @@ -435,7 +435,7 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i guestAddr := net.JoinHostPort(publicIP, "22") tunnel, err := hostClient.DialContext(ctx, "tcp", guestAddr) if err != nil { - return 0, fmt.Errorf("ddil harness: tunnel %s → %s: %w", host.Name, guestAddr, err) + return 0, fmt.Errorf("e2e harness: tunnel %s → %s: %w", host.Name, guestAddr, err) } guestCfg := &ssh.ClientConfig{ @@ -447,20 +447,20 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i c, 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) + return 0, fmt.Errorf("e2e harness: guest ssh handshake on %s: %w", guestAddr, err) } guestClient := ssh.NewClient(c, chans, reqs) defer func() { _ = guestClient.Close() }() session, err := guestClient.NewSession() if err != nil { - return 0, fmt.Errorf("ddil harness: guest ssh session: %w", err) + return 0, fmt.Errorf("e2e 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))) + return 0, fmt.Errorf("e2e harness: read /var/lib/counter: %w (output: %s)", err, strings.TrimSpace(string(raw))) } s := strings.TrimSpace(string(raw)) if s == "" { @@ -468,7 +468,7 @@ func readCounter(ctx context.Context, w *Witness, host Node, publicIP string) (i } n, err := strconv.Atoi(s) if err != nil { - return 0, fmt.Errorf("ddil harness: parse /var/lib/counter %q: %w", s, err) + return 0, fmt.Errorf("e2e harness: parse /var/lib/counter %q: %w", s, err) } return n, nil } @@ -509,7 +509,7 @@ func awaitBaseline(ctx context.Context, w *Witness, host Node, publicIP string, return 0, err } if time.Now().After(deadline) { - return 0, fmt.Errorf("ddil harness: %s baseline not ready after %s: %w", publicIP, timeout, lastErr) + return 0, fmt.Errorf("e2e harness: %s baseline not ready after %s: %w", publicIP, timeout, lastErr) } select { case <-ctx.Done(): @@ -529,12 +529,12 @@ func dialHost(ctx context.Context, node Node, user string, signer ssh.Signer) (* 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) + return nil, fmt.Errorf("e2e 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 nil, fmt.Errorf("e2e harness: host ssh handshake %s: %w", node.Name, err) } return ssh.NewClient(c, chans, reqs), nil }