From 531b36924c82242ee60c09ad085e1a8b8a360ef9 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Mon, 10 Aug 2026 13:37:28 +0000 Subject: [PATCH] =?UTF-8?q?fix(P09):=20migration=20+=20operational=20safet?= =?UTF-8?q?y=20=E2=80=94=20job=20stop,=20retention,=20logs=20cap=20(REQ-15?= =?UTF-8?q?8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - job stop: real systemctl stop via SSH (was DB-only soft stop) resolves node from alloc_history or --peer flag - doctor db-retention: row count check for jobs/tasks/audit_log warns at 100k rows, suggests backup + cleanup - logs --lines: cap at 50000 (default 1000); --since upper bound 7d prevents OOM from unbounded journalctl - cache DB mode 0600 (was 0644; matches store.Open) - upgrade cutover: backup file + atomic rename (was sed -i) rollback restores from backup on failure Tests: job stop SSH, DB retention warning, logs lines cap, cache mode, cutover backup-restore + atomic rename. ---ci--- project: orca phase: 9 milestone: v0.13 status: complete requirements: covered: [158] ---/ci--- --- internal/cache/cache.go | 5 + internal/cache/cache_test.go | 47 +++++ internal/cli/doctor.go | 76 +++++++- internal/cli/doctor_test.go | 94 ++++++++++ internal/cli/job.go | 143 ++++++++++++++- internal/cli/job_test.go | 193 ++++++++++++++++++++ internal/cli/logs.go | 49 ++++- internal/cli/logs_test.go | 157 +++++++++++++++- internal/cli/namespace_test.go | 5 + internal/cli/upgrade.go | 103 ++++++++++- internal/cli/upgrade_test.go | 319 ++++++++++++++++++++++++++++++++- 11 files changed, 1166 insertions(+), 25 deletions(-) diff --git a/internal/cache/cache.go b/internal/cache/cache.go index f22e98e..bfb2366 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -73,6 +73,11 @@ func Open(path string) (*Cache, error) { _ = db.Close() return nil, fmt.Errorf("ping cache sqlite: %w", err) } + // REQ-158 / P09 T4: enforce 0600 on the cache DB file (SQLite + // creates it at umask, typically 0644). Match store.Open which + // chmods after open+ping (the file exists at this point). Non-fatal + // if chmod fails (e.g. the DB is at a path we don't own). + _ = os.Chmod(path, 0o600) const schema = `CREATE TABLE IF NOT EXISTS cache_entries ( class TEXT NOT NULL, key TEXT NOT NULL, diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 4d34d38..0209909 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -2,6 +2,7 @@ package cache import ( "errors" + "os" "path/filepath" "testing" "time" @@ -225,3 +226,49 @@ func BenchmarkCacheHit(b *testing.B) { } } } + + +// TestCache_FileMode0600 verifies that the cache DB file is created +// with mode 0600 (not the default umask 0644) (REQ-158, P09 T4). +func TestCache_FileMode0600(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "orca_cache.db") + c, err := Open(path) + if err != nil { + t.Fatalf("open: %v", err) + } + defer c.Close() + + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat cache db: %v", err) + } + got := info.Mode().Perm() + if got != 0o600 { + t.Errorf("cache db mode = %04o, want 0600", got) + } +} + +// TestCache_FileMode0600DefaultPath verifies that the cache DB at the +// default path (ORCA_HOME) also gets 0600 (REQ-158, P09 T4). +func TestCache_FileMode0600DefaultPath(t *testing.T) { + dir := t.TempDir() + t.Setenv("ORCA_HOME", dir) + c, err := Open("") + if err != nil { + t.Fatalf("open default path: %v", err) + } + defer c.Close() + + // The default path is paths.CacheDB() which is under ORCA_HOME. + // Find the db file. + dbPath := filepath.Join(dir, "orca_cache.db") + info, err := os.Stat(dbPath) + if err != nil { + t.Fatalf("stat cache db at %s: %v", dbPath, err) + } + got := info.Mode().Perm() + if got != 0o600 { + t.Errorf("cache db mode = %04o, want 0600", got) + } +} diff --git a/internal/cli/doctor.go b/internal/cli/doctor.go index b79e32a..3cfe93a 100644 --- a/internal/cli/doctor.go +++ b/internal/cli/doctor.go @@ -384,7 +384,81 @@ func checkOIDCHealth(ctx context.Context) []oidcCheckResult { return results } +// doctorDBRetentionCmd implements `orca doctor db-retention` (REQ-158, +// P09 T2). Counts rows in the jobs, tasks, and audit_log tables and +// warns if any exceeds 100k rows (unbounded growth risk). Suggests +// `orca backup` + manual cleanup. +var doctorDBRetentionCmd = &cobra.Command{ + Use: "db-retention", + Short: "Check DB row counts for unbounded growth (REQ-158)", + Long: `Count rows in the jobs, tasks, and audit_log tables and warn +if any table exceeds 100,000 rows (unbounded growth risk). + +Large tables degrade query performance and inflate backup size. Run +'orca backup' to capture a snapshot, then prune old rows manually +(e.g. DELETE FROM tasks WHERE created_at < ). + +Exits 0 if all tables are under the threshold, exits 0 with WARN if any +table exceeds it (the check is advisory, not a hard failure).`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, args []string) error { + ctx, cancel := context.WithTimeout(cmd.Context(), 10*time.Second) + defer cancel() + + db, closer, err := openDB() + if err != nil { + return fmt.Errorf("doctor db-retention: open db: %w", err) + } + defer closer() + + tables := []string{"jobs", "tasks", "audit_log"} + const threshold = 100_000 + type rowCount struct { + Table string `json:"table"` + Count int64 `json:"count"` + Warn bool `json:"warn"` + } + var results []rowCount + anyWarn := false + for _, table := range tables { + var count int64 + q := fmt.Sprintf("SELECT COUNT(*) FROM %s", table) + if err := db.QueryRowContext(ctx, q).Scan(&count); err != nil { + return fmt.Errorf("doctor db-retention: count %s: %w", table, err) + } + warn := count > threshold + if warn { + anyWarn = true + } + results = append(results, rowCount{Table: table, Count: count, Warn: warn}) + } + + if jsonOutput { + return printJSON(map[string]any{ + "results": results, + "threshold": threshold, + "any_warn": anyWarn, + }) + } + + out := cmd.OutOrStdout() + for _, r := range results { + status := "ok" + if r.Warn { + status = "WARN" + } + fmt.Fprintf(out, "%-12s %-5s %d rows (threshold: %d)\n", r.Table, status, r.Count, threshold) + } + if anyWarn { + fmt.Fprintf(out, "\n⚠ one or more tables exceed %d rows — run 'orca backup' then prune old rows\n", threshold) + } else { + fmt.Fprintln(out, "\n✓ all tables under retention threshold") + } + return nil + }, +} + func init() { - doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd, doctorNftCmd, doctorAuditCmd, doctorModesCmd, doctorOIDCCmd) + doctorCmd.AddCommand(doctorCertCmd, doctorNetworkCmd, doctorDBCmd, doctorOSCmd, doctorProxmoxCmd, noOrcaOnServerCmd, doctorNftCmd, doctorAuditCmd, doctorModesCmd, doctorOIDCCmd, doctorDBRetentionCmd) rootCmd.AddCommand(doctorCmd) } diff --git a/internal/cli/doctor_test.go b/internal/cli/doctor_test.go index 344e1d7..8508e8d 100644 --- a/internal/cli/doctor_test.go +++ b/internal/cli/doctor_test.go @@ -2,9 +2,14 @@ package cli import ( "bytes" + "context" "encoding/json" "strings" "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/store" ) func TestDoctorText(t *testing.T) { @@ -194,3 +199,92 @@ func TestDoctorProxmoxJSON(t *testing.T) { t.Errorf("doctor proxmox --json missing Name: %v", result) } } + + +// TestDoctorDBRetention verifies that `orca doctor db-retention` counts +// rows in jobs, tasks, and audit_log and warns when a table exceeds +// 100k rows (REQ-158, P09 T7). +func TestDoctorDBRetention(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + if err := runInit(discardWriter{}); err != nil { + t.Fatalf("init: %v", err) + } + + // Insert 100001 rows into the audit_log table to trigger the warning. + // Use a multi-row VALUES insert in batches for speed. + db, err := store.Open(certpaths.DBPath()) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + ctx := context.Background() + // Build a batch insert: 500 rows per INSERT in a transaction. + // SQLite handles this much faster than 100k individual inserts. + const totalRows = 100001 + const batchSize = 500 + inserted := 0 + for inserted < totalRows { + remaining := totalRows - inserted + batch := batchSize + if remaining < batch { + batch = remaining + } + var placeholders strings.Builder + var args []any + for j := 0; j < batch; j++ { + if j > 0 { + placeholders.WriteString(",") + } + placeholders.WriteString("(?, 'test', 'test.action', 'test-resource', 'success')") + args = append(args, time.Now().UTC()) + } + q := "INSERT INTO audit_log (timestamp, actor, action, resource, result) VALUES " + placeholders.String() + if _, err := db.ExecContext(ctx, q, args...); err != nil { + t.Fatalf("batch insert at offset %d: %v", inserted, err) + } + inserted += batch + } + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"doctor", "db-retention"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("doctor db-retention: %v", err) + } + out := buf.String() + if !strings.Contains(out, "audit_log") { + t.Errorf("output missing audit_log table: %s", out) + } + if !strings.Contains(out, "WARN") { + t.Errorf("output should contain WARN for audit_log exceeding threshold: %s", out) + } + if !strings.Contains(out, "backup") { + t.Errorf("output should suggest 'orca backup': %s", out) + } +} + +// TestDoctorDBRetentionNoWarn verifies that with a small DB no warning +// is emitted (REQ-158, P09 T7). +func TestDoctorDBRetentionNoWarn(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + if err := runInit(discardWriter{}); err != nil { + t.Fatalf("init: %v", err) + } + + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"doctor", "db-retention"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("doctor db-retention: %v", err) + } + out := buf.String() + if strings.Contains(out, "WARN") { + t.Errorf("output should NOT contain WARN for small DB: %s", out) + } +} diff --git a/internal/cli/job.go b/internal/cli/job.go index b0fba22..2936725 100644 --- a/internal/cli/job.go +++ b/internal/cli/job.go @@ -2,6 +2,7 @@ package cli import ( "context" + "database/sql" "encoding/json" "errors" "fmt" @@ -14,9 +15,11 @@ import ( "github.com/google/uuid" "github.com/spf13/cobra" + "git.cloudinit.dev/coreci/orca/internal/certpaths" "git.cloudinit.dev/coreci/orca/internal/engine" "git.cloudinit.dev/coreci/orca/internal/jobspec" "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/sshpush" "git.cloudinit.dev/coreci/orca/internal/store" ) @@ -290,11 +293,79 @@ func renderJobTable(jobs []*model.Job) string { return out } +// jobStopTransport is the SSH command-execution seam used by +// `orca job stop`. *sshpush.Transport satisfies it via Exec; tests +// inject a mock (same pattern as driftTransport / drainExecer). +type jobStopTransport interface { + Exec(ctx context.Context, peer string, cmd string) ([]byte, error) +} + +// jobStopTransportOverride is the package-level test seam for the +// SSH transport used by `orca job stop`. When non-nil it replaces the +// production transport; tests set it and restore nil in cleanup. +var jobStopTransportOverride jobStopTransport + +// jobStopTimeout is the SSH command timeout for `orca job stop`. +var jobStopTimeout time.Duration + +// jobStopPeer is the optional --peer override for `orca job stop`. +// When empty, the node is looked up from the alloc_history table +// (latest entry for the job id). When set, the SSH stop targets that +// peer directly. +var jobStopPeer string + +func jobStopTransportFromCtx() (jobStopTransport, error) { + if jobStopTransportOverride != nil { + return jobStopTransportOverride, nil + } + keyPath := certpaths.SSHKeyPath() + khPath := certpaths.KnownHostsPath() + return sshpush.NewTransport(keyPath, khPath), nil +} + +// nodeForJob looks up the node that ran (or is running) a job by +// searching the alloc_history table for the latest entry for the +// given job id. Returns nil if no history entry exists (the job may +// have been run locally or pre-dates alloc_history). +func nodeForJob(ctx context.Context, db *sql.DB, jobID string) (*model.Node, error) { + hist := store.NewAllocHistoryRepo(db) + if err := hist.EnsureSchema(ctx); err != nil { + return nil, fmt.Errorf("alloc history schema: %w", err) + } + entries, err := hist.List(ctx, store.HistoryFilter{JobID: jobID}) + if err != nil { + return nil, fmt.Errorf("alloc history list: %w", err) + } + if len(entries) == 0 { + return nil, nil + } + // Pick the latest entry (List returns ASC; take the last). + latest := entries[len(entries)-1] + if latest.NodeID == "" { + return nil, nil + } + nodeRepo := store.NewNodeRepo(db) + n, err := nodeRepo.Get(ctx, latest.NodeID) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + return nil, nil + } + return nil, fmt.Errorf("lookup node %s: %w", latest.NodeID, err) + } + return n, nil +} + var jobStopCmd = &cobra.Command{ Use: "stop [job-id]", Short: "Stop a running job", - Long: "Mark a job as stopped. Note: this is a soft stop (cancel context for the daemon).", - Args: cobra.MaximumNArgs(1), + Long: `Stop a running job by sending 'systemctl stop orca-alloc--*' +to the node running the allocation via SSH, then mark the job as +stopped in the DB (REQ-158, P09 T1). + +If --peer is not given, the node is looked up from the allocation +history. If no node is found, the DB status is updated anyway (soft +stop fallback for local-run jobs).`, + Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error { id := stopID if id == "" && len(args) > 0 { @@ -303,8 +374,6 @@ var jobStopCmd = &cobra.Command{ if id == "" { return fmt.Errorf("job id required (--id or argument)") } - ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second) - defer cancel() db, closer, err := openDB() if err != nil { @@ -312,6 +381,9 @@ var jobStopCmd = &cobra.Command{ } defer closer() + ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second) + defer cancel() + repo := store.NewJobRepo(db) job, err := repo.Get(ctx, id) if err != nil { @@ -320,6 +392,52 @@ var jobStopCmd = &cobra.Command{ } return err } + + // Determine the peer to SSH to. --peer takes precedence; + // otherwise look up the node from alloc_history. + peer := jobStopPeer + var node *model.Node + if peer == "" { + node, err = nodeForJob(ctx, db, id) + if err != nil { + return fmt.Errorf("lookup node for job %s: %w", id, err) + } + if node != nil { + peer = peerAddrForNode(node) + } + } + + // Validate the job name before interpolation into the shell + // command (same injection guard as logs --job / stopAlloc). + jobName := job.Name + if !validSafeName(jobName) { + return fmt.Errorf("job stop: invalid job name %q (allowed: A-Z a-z 0-9 _ -)", jobName) + } + + sshRan := false + if peer != "" { + transport, terr := jobStopTransportFromCtx() + if terr != nil { + return fmt.Errorf("job stop: ssh transport: %w", terr) + } + stopCtx, stopCancel := sshCmdCtx(ctx, jobStopTimeout) + defer stopCancel() + // Match the drift.go job restart unit pattern: orca-alloc-. + // Use a glob (orca-alloc--*) to stop all task units in + // a multi-task allocation group. + unitPattern := fmt.Sprintf("orca-alloc-%s-*", jobName) + stopCmd := fmt.Sprintf("systemctl stop %s", shellQuote(unitPattern)) + out, sErr := transport.Exec(stopCtx, peer, stopCmd) + if sErr != nil { + // Non-fatal: the unit may not be running (already + // stopped) or SSH may fail. We still update the DB + // status so the operator's intent is recorded. + fmt.Fprintf(cmd.ErrOrStderr(), "⚠ job stop: SSH systemctl stop failed on %s: %v (output: %s)\n", peer, sErr, strings.TrimSpace(string(out))) + } else { + sshRan = true + } + } + if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil { return err } @@ -327,9 +445,20 @@ var jobStopCmd = &cobra.Command{ // `orca job list` reflects the just-stopped job. cacheInvalidate(cacheJobClass) if jsonOutput { - return printJSON(map[string]any{"id": id, "status": "stopped", "previous_status": job.Status}) + result := map[string]any{"id": id, "status": "stopped", "previous_status": job.Status} + if peer != "" { + result["peer"] = peer + result["ssh_stop"] = sshRan + } + return printJSON(result) + } + if peer != "" && sshRan { + fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (systemctl stop on %s)\n", id, peer) + } else if peer != "" { + fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; SSH stop failed — see stderr)\n", id) + } else { + fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s (DB only; no node found)\n", id) } - fmt.Fprintf(cmd.OutOrStdout(), "✓ Job stopped: %s\n", id) return nil }, } @@ -383,6 +512,8 @@ var jobLogsCmd = &cobra.Command{ func init() { jobStopCmd.Flags().StringVar(&stopID, "id", "", "job id") + jobStopCmd.Flags().StringVar(&jobStopPeer, "peer", "", "peer address (host:port) running the allocation (auto-detected from alloc history if empty)") + jobStopCmd.Flags().DurationVar(&jobStopTimeout, "timeout", sshCmdDefaultTimeout, "SSH command timeout") jobLogsCmd.Flags().StringVar(&stopID, "id", "", "job id") jobRunCmd.Flags().StringVar(&runTarget, "target", "", "pin job to a specific node id (overrides bin-packing)") jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe") diff --git a/internal/cli/job_test.go b/internal/cli/job_test.go index 4f4f159..e938a7a 100644 --- a/internal/cli/job_test.go +++ b/internal/cli/job_test.go @@ -2,11 +2,14 @@ package cli import ( "bytes" + "context" "encoding/json" "os" "path/filepath" "strings" + "sync" "testing" + "time" "git.cloudinit.dev/coreci/orca/internal/certpaths" "git.cloudinit.dev/coreci/orca/internal/model" @@ -310,3 +313,193 @@ func seedJob(t *testing.T, name string, status model.JobStatus) string { } return j.ID } + + +// mockJobStopExec is a record-and-replay SSH execer for `orca job stop` +// tests (same pattern as mockDrainExec / mockLogsExec). +type mockJobStopExec struct { + mu sync.Mutex + responses []jobStopMockResp + calls []jobStopMockCall +} + +type jobStopMockResp struct { + match string + out string + exit int +} + +type jobStopMockCall struct { + peer string + cmd string +} + +func (m *mockJobStopExec) Exec(_ context.Context, peer, cmd string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.calls = append(m.calls, jobStopMockCall{peer: peer, cmd: cmd}) + for _, r := range m.responses { + if r.match == "" || strings.Contains(cmd, r.match) { + return []byte(r.out), nil + } + } + return []byte(""), nil +} + +func (m *mockJobStopExec) callsFor(match string) []jobStopMockCall { + m.mu.Lock() + defer m.mu.Unlock() + var out []jobStopMockCall + for _, c := range m.calls { + if strings.Contains(c.cmd, match) { + out = append(out, c) + } + } + return out +} + +// TestJobStopSSH verifies that `orca job stop` sends a real +// 'systemctl stop' via SSH to the target node when the job has a +// recorded allocation history (REQ-158, P09 T6). +func TestJobStopSSH(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + if err := runInit(discardWriter{}); err != nil { + t.Fatalf("init: %v", err) + } + + // Seed a node and a job, then record an alloc_history entry + // linking the job to the node. + nodeID := seedNode(t, "worker-1", "worker-1:8443") + jobID := seedJob(t, "webapp", model.JobStatusRunning) + + db, err := store.Open(certpaths.DBPath()) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + hist := store.NewAllocHistoryRepo(db) + ctx := context.Background() + if err := hist.EnsureSchema(ctx); err != nil { + t.Fatalf("ensure schema: %v", err) + } + if err := hist.Record(ctx, store.AllocHistoryEntry{ + AllocID: "default/webapp-0", + JobID: jobID, + NodeID: nodeID, + Namespace: "default", + ToState: "created", + Timestamp: time.Now().UTC(), + }); err != nil { + t.Fatalf("record alloc history: %v", err) + } + + // Wire the mock SSH transport (must be after resetRootFlags so + // resetCommandFlags doesn't nil it out). + resetRootFlags(t) + mock := &mockJobStopExec{} + prev := jobStopTransportOverride + jobStopTransportOverride = mock + t.Cleanup(func() { jobStopTransportOverride = prev }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "stop", jobID}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job stop: %v", err) + } + + // Verify systemctl stop was called via SSH. + stopCalls := mock.callsFor("systemctl stop") + if len(stopCalls) == 0 { + t.Fatalf("expected systemctl stop SSH call, got %d calls: %v", len(mock.calls), mock.calls) + } + if !strings.Contains(stopCalls[0].cmd, "orca-alloc-webapp-*") { + t.Errorf("expected 'orca-alloc-webapp-*' in cmd, got: %s", stopCalls[0].cmd) + } + if !strings.Contains(stopCalls[0].peer, "worker-1") { + t.Errorf("expected peer to contain 'worker-1', got: %s", stopCalls[0].peer) + } + + // Verify the DB status was updated. + repo := store.NewJobRepo(db) + job, err := repo.Get(ctx, jobID) + if err != nil { + t.Fatalf("get job: %v", err) + } + if job.Status != model.JobStatusStopped { + t.Errorf("job status = %v, want stopped", job.Status) + } +} + +// TestJobStopSSHPeerOverride verifies that --peer bypasses the +// alloc_history lookup and uses the given peer directly (REQ-158). +func TestJobStopSSHPeerOverride(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + if err := runInit(discardWriter{}); err != nil { + t.Fatalf("init: %v", err) + } + + jobID := seedJob(t, "webapp2", model.JobStatusRunning) + + resetRootFlags(t) + mock := &mockJobStopExec{} + prev := jobStopTransportOverride + jobStopTransportOverride = mock + t.Cleanup(func() { jobStopTransportOverride = prev }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "stop", jobID, "--peer", "10.0.0.5:22"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job stop: %v", err) + } + + stopCalls := mock.callsFor("systemctl stop") + if len(stopCalls) == 0 { + t.Fatalf("expected systemctl stop SSH call, got %d calls", len(mock.calls)) + } + if stopCalls[0].peer != "10.0.0.5:22" { + t.Errorf("peer = %s, want 10.0.0.5:22", stopCalls[0].peer) + } +} + +// TestJobStopNoNodeFallback verifies that when no node is found in +// alloc_history, the job is still stopped in the DB (soft stop +// fallback) without attempting SSH (REQ-158). +func TestJobStopNoNodeFallback(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + if err := runInit(discardWriter{}); err != nil { + t.Fatalf("init: %v", err) + } + + jobID := seedJob(t, "localjob", model.JobStatusRunning) + + resetRootFlags(t) + mock := &mockJobStopExec{} + prev := jobStopTransportOverride + jobStopTransportOverride = mock + t.Cleanup(func() { jobStopTransportOverride = prev }) + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "stop", jobID}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job stop: %v", err) + } + + // No SSH calls should have been made (no node found). + if len(mock.calls) > 0 { + t.Errorf("expected 0 SSH calls, got %d: %v", len(mock.calls), mock.calls) + } + + // Verify the output mentions "DB only". + if !strings.Contains(buf.String(), "DB only") { + t.Errorf("output should mention 'DB only', got: %s", buf.String()) + } +} diff --git a/internal/cli/logs.go b/internal/cli/logs.go index 611bdae..3c3dbe7 100644 --- a/internal/cli/logs.go +++ b/internal/cli/logs.go @@ -101,8 +101,20 @@ var ( logsJob string logsSince string logsJSON bool + logsLines int ) +// logsMaxLines is the hard cap on --lines to prevent OOM from +// unbounded journalctl output (REQ-158, P09 T3). +const logsMaxLines = 50000 + +// logsDefaultLines is the default --lines value. +const logsDefaultLines = 1000 + +// logsMaxSince is the maximum lookback for --since (7 days) to +// prevent OOM from unbounded journalctl queries (REQ-158, P09 T3). +const logsMaxSince = 7 * 24 * time.Hour + var logsCmd = &cobra.Command{ Use: "logs", Short: "Aggregate journald logs across nodes (REQ-117)", @@ -139,6 +151,25 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`, if err != nil { return err } + // REQ-158 / P09 T3: clamp --since to 7 days max to prevent + // OOM from unbounded journalctl queries. If the requested + // lookback exceeds the cap, clamp it and warn. + now := time.Now().UTC() + maxSince := now.Add(-logsMaxSince) + if since.Before(maxSince) { + fmt.Fprintf(cmd.ErrOrStderr(), "⚠ --since %s exceeds 7d cap; clamping to 7d\n", logsSince) + since = maxSince + } + + // REQ-158 / P09 T3: clamp --lines to [1, logsMaxLines]. + lines := logsLines + if lines <= 0 { + lines = logsDefaultLines + } + if lines > logsMaxLines { + fmt.Fprintf(cmd.ErrOrStderr(), "⚠ --lines %d exceeds max %d; clamping\n", lines, logsMaxLines) + lines = logsMaxLines + } ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) defer cancel() @@ -158,7 +189,7 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`, out := cmd.OutOrStdout() multi := len(nodes) > 1 - for line := range streamLogs(ctx, ex, nodes, since, logsJob) { + for line := range streamLogs(ctx, ex, nodes, since, logsJob, lines) { if logsJSON { raw, _ := json.Marshal(line) fmt.Fprintln(out, string(raw)) @@ -227,7 +258,7 @@ func resolveLogNodes(ctx context.Context) ([]*model.Node, error) { // JSON entry immediately. The stream ends when every node has // completed (or the context is cancelled). The caller drives the // iteration via range-over-func (D-017 iter.Seq pattern). -func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since time.Time, job string) iter.Seq[LogLine] { +func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since time.Time, job string, lines int) iter.Seq[LogLine] { return func(yield func(LogLine) bool) { merged := make(chan LogLine) var wg sync.WaitGroup @@ -235,7 +266,7 @@ func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since t wg.Add(1) go func(n *model.Node) { defer wg.Done() - streamNodeLines(ctx, ex, n, since, job, merged) + streamNodeLines(ctx, ex, n, since, job, lines, merged) }(n) } done := make(chan struct{}) @@ -267,7 +298,7 @@ func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since t // context is cancelled); the caller is responsible for waiting on the // goroutine. Send is non-blocking via select on ctx.Done so a slow // consumer does not stall the fanout forever. -func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since time.Time, job string, out chan<- LogLine) { +func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since time.Time, job string, lines int, out chan<- LogLine) { peer := peerAddrForNode(n) if peer == "" { slog.Default().Warn("logs: cannot resolve SSH address for node", "node", n.Name) @@ -278,9 +309,14 @@ func streamNodeLines(ctx context.Context, ex logsExecer, n *model.Node, since ti unitPattern = "orca-alloc-" + job + "-*" } sinceStr := since.Format("2006-01-02 15:04:05") + // REQ-158 / P09 T3: pass --lines=N to journalctl to cap output + // and prevent OOM from unbounded log queries. + if lines <= 0 { + lines = logsDefaultLines + } // F1: shellQuote (single-quote wrap) instead of %q — %q does not // escape backticks, enabling command substitution in double quotes. - cmd := fmt.Sprintf("journalctl -u %s --since %s --output json --no-pager", shellQuote(unitPattern), shellQuote(sinceStr)) + cmd := fmt.Sprintf("journalctl -u %s --since %s --lines %d --output json --no-pager", shellQuote(unitPattern), shellQuote(sinceStr), lines) raw, err := ex.Exec(ctx, peer, cmd) if err != nil { slog.Default().Warn("logs: exec failed", "node", n.Name, "peer", peer, "error", err) @@ -324,7 +360,8 @@ func init() { logsCmd.Flags().BoolVar(&logsAllNodes, "all-nodes", false, "fan out to all registered nodes") logsCmd.Flags().StringVar(&logsNode, "node", "", "restrict to a single node (name or id)") logsCmd.Flags().StringVar(&logsJob, "job", "", "filter by job name (matches orca-alloc--* units)") - logsCmd.Flags().StringVar(&logsSince, "since", "5m", "duration lookback (e.g. 5m, 1h, 30m); default 5m") + logsCmd.Flags().StringVar(&logsSince, "since", "5m", "duration lookback (e.g. 5m, 1h, 30m); default 5m; max 7d") + logsCmd.Flags().IntVar(&logsLines, "lines", logsDefaultLines, fmt.Sprintf("max number of journal lines per node (default %d, max %d)", logsDefaultLines, logsMaxLines)) logsCmd.Flags().BoolVar(&logsJSON, "json", false, "output raw JSON (one LogLine per line)") rootCmd.AddCommand(logsCmd) } diff --git a/internal/cli/logs_test.go b/internal/cli/logs_test.go index 19573b6..8c171f9 100644 --- a/internal/cli/logs_test.go +++ b/internal/cli/logs_test.go @@ -63,6 +63,18 @@ func (m *mockLogsExec) countCalls(match string) int { return n } +func (m *mockLogsExec) callsFor(match string) []logsMockCall { + m.mu.Lock() + defer m.mu.Unlock() + var out []logsMockCall + for _, c := range m.calls { + if strings.Contains(c.cmd, match) { + out = append(out, c) + } + } + return out +} + // logsTestEnv wires a mockLogsExec into logsExecOverride and returns // the mock + a cleanup func. Tests MUST defer the cleanup. func logsTestEnv(t *testing.T) *mockLogsExec { @@ -325,7 +337,7 @@ func TestLogsCancelStopsStream(t *testing.T) { {ID: "n1", Name: "cancelnode", Address: "cancelnode:8443"}, } consumed := 0 - for range streamLogs(ctx, ex, nodes, time.Now().UTC().Add(-1*time.Minute), "") { + for range streamLogs(ctx, ex, nodes, time.Now().UTC().Add(-1*time.Minute), "", 1000) { consumed++ } if consumed > 1 { @@ -357,3 +369,146 @@ func TestLogsParseJournalLine_InvalidJSON(t *testing.T) { t.Error("expected error for invalid json, got nil") } } + + +// TestLogsLinesFlag verifies that --lines is passed through to the +// journalctl command as --lines=N (REQ-158, P09 T8). +func TestLogsLinesFlag(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "linesnode", "linesnode:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "line test", "6") + "\n"}, + } + + _, err := runLogsCmd(t, []string{"logs", "--node", "linesnode", "--since", "1m", "--lines", "500"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + calls := mx.callsFor("journalctl") + if len(calls) == 0 { + t.Fatal("expected journalctl call") + } + if !strings.Contains(calls[0].cmd, "--lines 500") { + t.Errorf("expected '--lines 500' in cmd, got: %s", calls[0].cmd) + } +} + +// TestLogsLinesDefault verifies that the default --lines value (1000) +// is passed to journalctl when --lines is not specified (REQ-158). +func TestLogsLinesDefault(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "defnode", "defnode:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "default lines", "6") + "\n"}, + } + + _, err := runLogsCmd(t, []string{"logs", "--node", "defnode", "--since", "1m"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + calls := mx.callsFor("journalctl") + if len(calls) == 0 { + t.Fatal("expected journalctl call") + } + if !strings.Contains(calls[0].cmd, "--lines 1000") { + t.Errorf("expected default '--lines 1000' in cmd, got: %s", calls[0].cmd) + } +} + +// TestLogsLinesClamp verifies that --lines exceeding the max (50000) is +// clamped (REQ-158, P09 T8). +func TestLogsLinesClamp(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "clampnode", "clampnode:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "clamp test", "6") + "\n"}, + } + + _, err := runLogsCmd(t, []string{"logs", "--node", "clampnode", "--since", "1m", "--lines", "999999"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + calls := mx.callsFor("journalctl") + if len(calls) == 0 { + t.Fatal("expected journalctl call") + } + if !strings.Contains(calls[0].cmd, "--lines 50000") { + t.Errorf("expected clamped '--lines 50000' in cmd, got: %s", calls[0].cmd) + } +} + +// TestLogsSinceClamp verifies that --since exceeding 7 days is +// clamped and a warning is printed (REQ-158, P09 T8). +func TestLogsSinceClamp(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "sincenode", "sincenode:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "since test", "6") + "\n"}, + } + + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + resetRootFlags(t) + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"logs", "--node", "sincenode", "--since", "720h"}) + err := rootCmd.Execute() + if err != nil { + t.Fatalf("logs: %v", err) + } + out := buf.String() + if !strings.Contains(out, "clamping to 7d") { + t.Errorf("expected warning about clamping --since to 7d, got: %s", out) + } + calls := mx.callsFor("journalctl") + if len(calls) == 0 { + t.Fatal("expected journalctl call") + } +} + +// TestLogsLinesFlagJSON verifies --lines is passed through in JSON mode. +func TestLogsLinesFlagJSON(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "jsonlines", "jsonlines:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "json lines test", "6") + "\n"}, + } + + _, err := runLogsCmd(t, []string{"logs", "--node", "jsonlines", "--since", "1m", "--lines", "200", "--json"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + calls := mx.callsFor("journalctl") + if len(calls) == 0 { + t.Fatal("expected journalctl call") + } + if !strings.Contains(calls[0].cmd, "--lines 200") { + t.Errorf("expected '--lines 200' in cmd, got: %s", calls[0].cmd) + } +} diff --git a/internal/cli/namespace_test.go b/internal/cli/namespace_test.go index 59c518c..fce4137 100644 --- a/internal/cli/namespace_test.go +++ b/internal/cli/namespace_test.go @@ -57,6 +57,11 @@ func resetCommandFlags() { driftConfigPath = "" driftRemediateForce = false jobRestartPeer = "" + jobStopPeer = "" + jobStopTimeout = 0 + jobStopTransportOverride = nil + logsLines = logsDefaultLines + cutoverFSOverride = nil jobLintExplain = false jobLintFormat = "text" jobVerifyLead = "" diff --git a/internal/cli/upgrade.go b/internal/cli/upgrade.go index ebe2cec..96d1268 100644 --- a/internal/cli/upgrade.go +++ b/internal/cli/upgrade.go @@ -69,6 +69,41 @@ var upgradeTransportOverride upgradeTransport // peers to create the orca user on. Returns a list of peer addresses. var peersListerOverride func() ([]string, error) +// cutoverFS is the filesystem seam used by performCutover / +// rollbackCutover for Traefik config editing (REQ-158, P09 T5). The +// production implementation uses real os calls; tests inject a mock +// so they don't need /etc/traefik/traefik.yml to exist. +type cutoverFS interface { + ReadFile(path string) ([]byte, error) + WriteFile(path string, content []byte, mode os.FileMode) error + Rename(old, new string) error + Remove(path string) error + Stat(path string) (os.FileInfo, error) +} + +// realCutoverFS is the production cutoverFS backed by the real os. +type realCutoverFS struct{} + +func (realCutoverFS) ReadFile(path string) ([]byte, error) { return os.ReadFile(path) } +func (realCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error { + return os.WriteFile(path, content, mode) +} +func (realCutoverFS) Rename(old, new string) error { return os.Rename(old, new) } +func (realCutoverFS) Remove(path string) error { return os.Remove(path) } +func (realCutoverFS) Stat(path string) (os.FileInfo, error) { return os.Stat(path) } + +// cutoverFSOverride is the package-level test seam for the cutover +// filesystem. When non-nil it replaces the production FS; tests set +// it and restore nil in cleanup. +var cutoverFSOverride cutoverFS + +func cutoverFSFromCtx() cutoverFS { + if cutoverFSOverride != nil { + return cutoverFSOverride + } + return realCutoverFS{} +} + var upgradeCmd = &cobra.Command{ Use: "upgrade", Short: "Upgrade orca to a new version (REQ-115, R-017 cutover)", @@ -310,11 +345,43 @@ func detectOldTraefikBinding() bool { // return 200. On failure, rolls back (restores :443, removes nft rules) // and returns (false, nil). On success returns (true, nil). With // force=true, verification is skipped. +// +// REQ-158 / P09 T5: the cutover now uses a backup-file + atomic-rename +// strategy instead of `sed -i` (which edits in-place with no backup). +// The Traefik config is copied to traefik.yml.bak, the new content is +// written to a temp file, then atomically renamed over the original. +// If any step fails, the backup is restored. This prevents a partial +// edit from leaving Traefik in a broken state. func performCutover(ctx context.Context, runner commandRunner, out interface{ Write([]byte) (int, error) }, force bool) (bool, error) { - if _, err := runner.Run(ctx, "sed", "-i", "s/:443/127.0.0.1:8443/g", "/etc/traefik/traefik.yml"); err != nil { - return false, fmt.Errorf("cutover: edit traefik.yml: %w", err) + cfs := cutoverFSFromCtx() + traefikYml := "/etc/traefik/traefik.yml" + backupPath := traefikYml + ".bak" + + // Step 1: read the current config and create a backup. + original, err := cfs.ReadFile(traefikYml) + if err != nil { + return false, fmt.Errorf("cutover: read traefik.yml: %w", err) } + if err := cfs.WriteFile(backupPath, original, 0o644); err != nil { + return false, fmt.Errorf("cutover: write backup %s: %w", backupPath, err) + } + + // Step 2: write the new config to a temp file, then atomically rename. + newContent := strings.ReplaceAll(string(original), ":443", "127.0.0.1:8443") + tmpPath := traefikYml + ".tmp" + if err := cfs.WriteFile(tmpPath, []byte(newContent), 0o644); err != nil { + return false, fmt.Errorf("cutover: write temp %s: %w", tmpPath, err) + } + if err := cfs.Rename(tmpPath, traefikYml); err != nil { + // Rename failed — restore from backup and clean up the temp file. + _ = cfs.Remove(tmpPath) + _ = cfs.Rename(backupPath, traefikYml) + return false, fmt.Errorf("cutover: atomic rename %s → %s: %w", tmpPath, traefikYml, err) + } + if _, err := runner.Run(ctx, "systemctl", "restart", "traefik"); err != nil { + // Restart failed — restore from backup. + _ = cfs.Rename(backupPath, traefikYml) return false, fmt.Errorf("cutover: restart traefik: %w", err) } nftCmd := `nft add table inet orca_redirect; nft 'add chain inet orca_redirect prerouting { type nat hook prerouting priority -100; }'; nft add rule inet orca_redirect prerouting tcp dport 443 dnat to 127.0.0.1:8443` @@ -324,6 +391,8 @@ func performCutover(ctx context.Context, runner commandRunner, out interface{ Wr if force { fmt.Fprintf(out, " --force: skipping cutover verification\n") + // Clean up the backup on success. + _ = cfs.Remove(backupPath) return true, nil } @@ -336,6 +405,8 @@ func performCutover(ctx context.Context, runner commandRunner, out interface{ Wr return false, nil } fmt.Fprintf(out, " ✓ C-25 cutover verification passed (200 from Traefik)\n") + // Clean up the backup on success. + _ = cfs.Remove(backupPath) return true, nil } @@ -390,9 +461,33 @@ func verifyCutover(out interface{ Write([]byte) (int, error) }) error { } // rollbackCutover restores Traefik to :443 and removes nftables rules. +// REQ-158 / P09 T5: restore from the backup file (traefik.yml.bak) +// created by performCutover, falling back to an in-place replacement +// if the backup is missing. func rollbackCutover(ctx context.Context, runner commandRunner) error { - if _, err := runner.Run(ctx, "sed", "-i", "s/127.0.0.1:8443/:443/g", "/etc/traefik/traefik.yml"); err != nil { - return fmt.Errorf("rollback: edit traefik.yml: %w", err) + cfs := cutoverFSFromCtx() + traefikYml := "/etc/traefik/traefik.yml" + backupPath := traefikYml + ".bak" + // Try restoring from the backup first. + if _, err := cfs.Stat(backupPath); err == nil { + if err := cfs.Rename(backupPath, traefikYml); err != nil { + return fmt.Errorf("rollback: restore backup %s → %s: %w", backupPath, traefikYml, err) + } + } else { + // No backup — do an in-place replacement as a fallback. + current, rErr := cfs.ReadFile(traefikYml) + if rErr != nil { + return fmt.Errorf("rollback: read traefik.yml: %w", rErr) + } + restored := strings.ReplaceAll(string(current), "127.0.0.1:8443", ":443") + tmpPath := traefikYml + ".tmp" + if err := cfs.WriteFile(tmpPath, []byte(restored), 0o644); err != nil { + return fmt.Errorf("rollback: write temp %s: %w", tmpPath, err) + } + if err := cfs.Rename(tmpPath, traefikYml); err != nil { + _ = cfs.Remove(tmpPath) + return fmt.Errorf("rollback: atomic rename: %w", err) + } } if _, err := runner.Run(ctx, "systemctl", "restart", "traefik"); err != nil { return fmt.Errorf("rollback: restart traefik: %w", err) diff --git a/internal/cli/upgrade_test.go b/internal/cli/upgrade_test.go index 625b9a1..46f20a2 100644 --- a/internal/cli/upgrade_test.go +++ b/internal/cli/upgrade_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "testing" + "time" "git.cloudinit.dev/coreci/orca/internal/migration" "git.cloudinit.dev/coreci/orca/internal/paths" @@ -61,6 +62,77 @@ func (m *mockUpgradeTransport) Exec(ctx context.Context, peer string, cmd string return []byte(""), nil } +// mockCutoverFS is an in-memory cutoverFS for testing performCutover / +// rollbackCutover without touching /etc/traefik (REQ-158, P09 T5). +type mockCutoverFS struct { + files map[string][]byte + errs map[string]error // keyed by operation: "read:", "write:", "rename:", "stat:" +} + +func newMockCutoverFS() *mockCutoverFS { + return &mockCutoverFS{ + files: make(map[string][]byte), + errs: make(map[string]error), + } +} + +func (m *mockCutoverFS) ReadFile(path string) ([]byte, error) { + if err, ok := m.errs["read:"+path]; ok { + return nil, err + } + if data, ok := m.files[path]; ok { + return data, nil + } + return nil, fmt.Errorf("mock: %s not found", path) +} + +func (m *mockCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error { + if err, ok := m.errs["write:"+path]; ok { + return err + } + cp := make([]byte, len(content)) + copy(cp, content) + m.files[path] = cp + return nil +} + +func (m *mockCutoverFS) Rename(old, new string) error { + if err, ok := m.errs["rename:"+old]; ok { + return err + } + data, ok := m.files[old] + if !ok { + return fmt.Errorf("mock: rename source %s not found", old) + } + m.files[new] = data + delete(m.files, old) + return nil +} + +func (m *mockCutoverFS) Remove(path string) error { + delete(m.files, path) + return nil +} + +func (m *mockCutoverFS) Stat(path string) (os.FileInfo, error) { + if err, ok := m.errs["stat:"+path]; ok { + return nil, err + } + if _, ok := m.files[path]; ok { + return mockFileInfo{name: path}, nil + } + return nil, fmt.Errorf("mock: %s not found", path) +} + +type mockFileInfo struct{ name string } + +func (m mockFileInfo) Name() string { return m.name } +func (m mockFileInfo) Size() int64 { return 0 } +func (m mockFileInfo) Mode() os.FileMode { return 0o644 } +func (m mockFileInfo) ModTime() time.Time { return time.Now() } +func (m mockFileInfo) IsDir() bool { return false } +func (m mockFileInfo) Sys() any { return nil } + func setupUpgradeTest(t *testing.T) { t.Helper() t.Setenv("ORCA_HOME", t.TempDir()) @@ -162,6 +234,12 @@ func TestUpgradeCutoverVerificationSuccess(t *testing.T) { upgradeRunnerOverride = runner httpClientOverride = func(url string) (int, error) { return 200, nil } + // Provide a mock Traefik config so performCutover can read it. + cfs := newMockCutoverFS() + cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n") + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"}) if err := rootCmd.Execute(); err != nil { t.Fatalf("upgrade with cutover: %v", err) @@ -180,6 +258,12 @@ func TestUpgradeCutoverRollback(t *testing.T) { upgradeRunnerOverride = runner httpClientOverride = func(url string) (int, error) { return 502, nil } + // Provide a mock Traefik config so performCutover can read it. + cfs := newMockCutoverFS() + cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n") + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0"}) err := rootCmd.Execute() if err == nil { @@ -194,20 +278,27 @@ func TestUpgradeCutoverRollback(t *testing.T) { t.Errorf("output should mention rollback: %s", out) } + // Verify rollback: the traefik.yml content should be restored to + // :443 (the backup was renamed back over the modified file). + restored, ok := cfs.files["/etc/traefik/traefik.yml"] + if !ok { + t.Fatal("rollback: traefik.yml missing after rollback") + } + if !strings.Contains(string(restored), ":443") { + t.Errorf("rollback: traefik.yml not restored to :443, got: %s", string(restored)) + } + if strings.Contains(string(restored), "127.0.0.1:8443") { + t.Errorf("rollback: traefik.yml still has 127.0.0.1:8443 after rollback: %s", string(restored)) + } + foundRollback := false for _, call := range runner.calls { - if call.name == "sed" && len(call.args) >= 2 { - joined := strings.Join(call.args, " ") - if strings.Contains(joined, "127.0.0.1:8443") && strings.Contains(joined, ":443") { - foundRollback = true - } - } if call.name == "nft" && len(call.args) >= 2 && call.args[0] == "delete" { foundRollback = true } } if !foundRollback { - t.Errorf("rollback commands not detected (calls: %v)", runner.calls) + t.Errorf("rollback nft delete command not detected (calls: %v)", runner.calls) } } @@ -227,6 +318,12 @@ func TestUpgradeCutoverForceSkipsVerification(t *testing.T) { return 200, nil } + // Provide a mock Traefik config so performCutover can read it. + cfs := newMockCutoverFS() + cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint: :443\n") + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"}) if err := rootCmd.Execute(); err != nil { t.Fatalf("upgrade with --force: %v", err) @@ -339,3 +436,211 @@ func TestUpgradeFullMigration(t *testing.T) { t.Errorf("install.sh was not invoked (calls: %v)", runner.calls) } } + + +// TestCutoverBackupRestoreOnFailure verifies that when the cutover +// verification fails, the Traefik config is restored from the backup +// file (REQ-158, P09 T10). This is a unit-level test that calls +// performCutover directly with a mock FS. +func TestCutoverBackupRestoreOnFailure(t *testing.T) { + // Set up a mock FS with a Traefik config containing :443. + cfs := newMockCutoverFS() + original := []byte("entrypoint:\n - :443\n") + cfs.files["/etc/traefik/traefik.yml"] = original + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + + // Mock runner that succeeds for systemctl restart. + runner := &mockUpgradeRunner{ + outputs: make(map[string][]byte), + } + // Mock HTTP check returns 502 (failure). + prevHTTP := httpClientOverride + httpClientOverride = func(url string) (int, error) { return 502, nil } + t.Cleanup(func() { httpClientOverride = prevHTTP }) + + var buf bytes.Buffer + ok, err := performCutover(context.Background(), runner, &buf, false) + if err != nil { + t.Fatalf("performCutover: %v", err) + } + if ok { + t.Fatal("expected cutover to fail (ok=false)") + } + + // Verify the Traefik config was restored from backup. + restored, exists := cfs.files["/etc/traefik/traefik.yml"] + if !exists { + t.Fatal("traefik.yml missing after rollback") + } + if string(restored) != string(original) { + t.Errorf("traefik.yml not restored to original, got: %s", string(restored)) + } + // Verify 127.0.0.1:8443 is NOT in the restored file. + if strings.Contains(string(restored), "127.0.0.1:8443") { + t.Errorf("traefik.yml still has 127.0.0.1:8443 after rollback: %s", string(restored)) + } + // The backup file should have been consumed by rollbackCutover's rename. + if _, bakExists := cfs.files["/etc/traefik/traefik.yml.bak"]; bakExists { + t.Error("backup file still exists after rollback (should have been renamed)") + } +} + +// TestCutoverAtomicRenameSuccess verifies that the cutover writes the +// new config via atomic rename (temp file → original) and cleans up +// the backup on success (REQ-158, P09 T10). +func TestCutoverAtomicRenameSuccess(t *testing.T) { + cfs := newMockCutoverFS() + original := []byte("entrypoint:\n - :443\n") + cfs.files["/etc/traefik/traefik.yml"] = original + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + + runner := &mockUpgradeRunner{ + outputs: make(map[string][]byte), + } + prevHTTP := httpClientOverride + httpClientOverride = func(url string) (int, error) { return 200, nil } + t.Cleanup(func() { httpClientOverride = prevHTTP }) + + var buf bytes.Buffer + ok, err := performCutover(context.Background(), runner, &buf, false) + if err != nil { + t.Fatalf("performCutover: %v", err) + } + if !ok { + t.Fatal("expected cutover to succeed (ok=true)") + } + + // Verify the config was updated to 127.0.0.1:8443. + updated, exists := cfs.files["/etc/traefik/traefik.yml"] + if !exists { + t.Fatal("traefik.yml missing after cutover") + } + if !strings.Contains(string(updated), "127.0.0.1:8443") { + t.Errorf("traefik.yml should have 127.0.0.1:8443, got: %s", string(updated)) + } + if strings.Contains(string(updated), ":443\n") && !strings.Contains(string(updated), "127.0.0.1:8443") { + t.Errorf("traefik.yml should not have bare :443 anymore, got: %s", string(updated)) + } + // The temp file should not exist. + if _, tmpExists := cfs.files["/etc/traefik/traefik.yml.tmp"]; tmpExists { + t.Error("temp file still exists after atomic rename") + } + // The backup should have been cleaned up on success. + if _, bakExists := cfs.files["/etc/traefik/traefik.yml.bak"]; bakExists { + t.Error("backup file still exists after successful cutover (should be cleaned up)") + } +} + +// TestCutoverBackupCreated verifies that a backup file is created +// before the cutover edits the config (REQ-158, P09 T10). Uses a +// custom mock FS that records the sequence of operations so we can +// assert the backup was written before the temp file. +func TestCutoverBackupCreated(t *testing.T) { + // Use a recording mock FS that fails on the rename step so the + // backup write is observable before the rollback consumes it. + cfs := newMockCutoverFS() + original := []byte("entrypoint:\n - :443\n") + cfs.files["/etc/traefik/traefik.yml"] = original + // Track write order via a custom FS that records operations. + var writeOrder []string + recordingCFS := &recordingCutoverFS{ + inner: cfs, + writeOrder: &writeOrder, + } + // Make the rename of the temp file fail so the cutover aborts. + cfs.errs["rename:/etc/traefik/traefik.yml.tmp"] = fmt.Errorf("rename failed") + cutoverFSOverride = recordingCFS + t.Cleanup(func() { cutoverFSOverride = nil }) + + runner := &mockUpgradeRunner{ + outputs: make(map[string][]byte), + } + + var buf bytes.Buffer + _, err := performCutover(context.Background(), runner, &buf, false) + if err == nil { + t.Fatal("expected error from failed rename") + } + + // Verify the backup was written BEFORE the temp file. + // writeOrder records WriteFile calls in order. + bakIdx := -1 + tmpIdx := -1 + for i, p := range writeOrder { + if p == "/etc/traefik/traefik.yml.bak" { + bakIdx = i + } + if p == "/etc/traefik/traefik.yml.tmp" { + tmpIdx = i + } + } + if bakIdx == -1 { + t.Fatal("backup file was not written before cutover") + } + if tmpIdx == -1 { + t.Fatal("temp file was not written") + } + if bakIdx > tmpIdx { + t.Errorf("backup written after temp file (bakIdx=%d, tmpIdx=%d) — backup should come first", bakIdx, tmpIdx) + } + // The original should have been restored from backup on failure. + restored, exists := cfs.files["/etc/traefik/traefik.yml"] + if !exists { + t.Fatal("traefik.yml missing after failed rename + restore") + } + if string(restored) != string(original) { + t.Errorf("traefik.yml not restored to original after failed rename, got: %s", string(restored)) + } +} + +// recordingCutoverFS wraps a cutoverFS and records WriteFile call +// paths so tests can assert the order of operations (REQ-158, P09 T10). +type recordingCutoverFS struct { + inner cutoverFS + writeOrder *[]string +} + +func (r *recordingCutoverFS) ReadFile(path string) ([]byte, error) { + return r.inner.ReadFile(path) +} +func (r *recordingCutoverFS) WriteFile(path string, content []byte, mode os.FileMode) error { + *r.writeOrder = append(*r.writeOrder, path) + return r.inner.WriteFile(path, content, mode) +} +func (r *recordingCutoverFS) Rename(old, new string) error { + return r.inner.Rename(old, new) +} +func (r *recordingCutoverFS) Remove(path string) error { + return r.inner.Remove(path) +} +func (r *recordingCutoverFS) Stat(path string) (os.FileInfo, error) { + return r.inner.Stat(path) +} + +// TestCutoverNoSedDirectly verifies that the cutover does NOT use +// `sed -i` (the old unsafe approach). The mock runner records all +// calls; none should be `sed` (REQ-158, P09 T5). +func TestCutoverNoSedDirectly(t *testing.T) { + cfs := newMockCutoverFS() + cfs.files["/etc/traefik/traefik.yml"] = []byte("entrypoint:\n - :443\n") + cutoverFSOverride = cfs + t.Cleanup(func() { cutoverFSOverride = nil }) + + runner := &mockUpgradeRunner{ + outputs: make(map[string][]byte), + } + prevHTTP := httpClientOverride + httpClientOverride = func(url string) (int, error) { return 200, nil } + t.Cleanup(func() { httpClientOverride = prevHTTP }) + + var buf bytes.Buffer + _, _ = performCutover(context.Background(), runner, &buf, false) + + for _, call := range runner.calls { + if call.name == "sed" { + t.Errorf("cutover should not use 'sed' (uses atomic rename now), found call: %s %v", call.name, call.args) + } + } +}