fix(P09): migration + operational safety — job stop, retention, logs cap (REQ-158)
- 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---
This commit is contained in:
Vendored
+5
@@ -73,6 +73,11 @@ func Open(path string) (*Cache, error) {
|
|||||||
_ = db.Close()
|
_ = db.Close()
|
||||||
return nil, fmt.Errorf("ping cache sqlite: %w", err)
|
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 (
|
const schema = `CREATE TABLE IF NOT EXISTS cache_entries (
|
||||||
class TEXT NOT NULL,
|
class TEXT NOT NULL,
|
||||||
key TEXT NOT NULL,
|
key TEXT NOT NULL,
|
||||||
|
|||||||
Vendored
+47
@@ -2,6 +2,7 @@ package cache
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+75
-1
@@ -384,7 +384,81 @@ func checkOIDCHealth(ctx context.Context) []oidcCheckResult {
|
|||||||
return results
|
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 < <cutoff>).
|
||||||
|
|
||||||
|
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() {
|
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)
|
rootCmd.AddCommand(doctorCmd)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,9 +2,14 @@ package cli
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||||
|
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestDoctorText(t *testing.T) {
|
func TestDoctorText(t *testing.T) {
|
||||||
@@ -194,3 +199,92 @@ func TestDoctorProxmoxJSON(t *testing.T) {
|
|||||||
t.Errorf("doctor proxmox --json missing Name: %v", result)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+137
-6
@@ -2,6 +2,7 @@ package cli
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"database/sql"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
@@ -14,9 +15,11 @@ import (
|
|||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"github.com/spf13/cobra"
|
"github.com/spf13/cobra"
|
||||||
|
|
||||||
|
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/engine"
|
"git.cloudinit.dev/coreci/orca/internal/engine"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/jobspec"
|
"git.cloudinit.dev/coreci/orca/internal/jobspec"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
"git.cloudinit.dev/coreci/orca/internal/model"
|
||||||
|
"git.cloudinit.dev/coreci/orca/internal/sshpush"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/store"
|
"git.cloudinit.dev/coreci/orca/internal/store"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -290,11 +293,79 @@ func renderJobTable(jobs []*model.Job) string {
|
|||||||
return out
|
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{
|
var jobStopCmd = &cobra.Command{
|
||||||
Use: "stop [job-id]",
|
Use: "stop [job-id]",
|
||||||
Short: "Stop a running job",
|
Short: "Stop a running job",
|
||||||
Long: "Mark a job as stopped. Note: this is a soft stop (cancel context for the daemon).",
|
Long: `Stop a running job by sending 'systemctl stop orca-alloc-<name>-*'
|
||||||
Args: cobra.MaximumNArgs(1),
|
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 {
|
RunE: func(cmd *cobra.Command, args []string) error {
|
||||||
id := stopID
|
id := stopID
|
||||||
if id == "" && len(args) > 0 {
|
if id == "" && len(args) > 0 {
|
||||||
@@ -303,8 +374,6 @@ var jobStopCmd = &cobra.Command{
|
|||||||
if id == "" {
|
if id == "" {
|
||||||
return fmt.Errorf("job id required (--id or argument)")
|
return fmt.Errorf("job id required (--id or argument)")
|
||||||
}
|
}
|
||||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Second)
|
|
||||||
defer cancel()
|
|
||||||
|
|
||||||
db, closer, err := openDB()
|
db, closer, err := openDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -312,6 +381,9 @@ var jobStopCmd = &cobra.Command{
|
|||||||
}
|
}
|
||||||
defer closer()
|
defer closer()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
repo := store.NewJobRepo(db)
|
repo := store.NewJobRepo(db)
|
||||||
job, err := repo.Get(ctx, id)
|
job, err := repo.Get(ctx, id)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -320,6 +392,52 @@ var jobStopCmd = &cobra.Command{
|
|||||||
}
|
}
|
||||||
return err
|
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-<name>.
|
||||||
|
// Use a glob (orca-alloc-<name>-*) 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 {
|
if err := repo.UpdateStatus(ctx, id, model.JobStatusStopped, 130); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -327,9 +445,20 @@ var jobStopCmd = &cobra.Command{
|
|||||||
// `orca job list` reflects the just-stopped job.
|
// `orca job list` reflects the just-stopped job.
|
||||||
cacheInvalidate(cacheJobClass)
|
cacheInvalidate(cacheJobClass)
|
||||||
if jsonOutput {
|
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
|
return nil
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
@@ -383,6 +512,8 @@ var jobLogsCmd = &cobra.Command{
|
|||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
jobStopCmd.Flags().StringVar(&stopID, "id", "", "job id")
|
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")
|
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(&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")
|
jobRunCmd.Flags().StringVar(&runIDKey, "idempotency-key", "", "X-Orca-Idempotency-Key for cross-node dispatch dedupe")
|
||||||
|
|||||||
@@ -2,11 +2,14 @@ package cli
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
"git.cloudinit.dev/coreci/orca/internal/certpaths"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/model"
|
"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
|
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())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+43
-6
@@ -101,8 +101,20 @@ var (
|
|||||||
logsJob string
|
logsJob string
|
||||||
logsSince string
|
logsSince string
|
||||||
logsJSON bool
|
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{
|
var logsCmd = &cobra.Command{
|
||||||
Use: "logs",
|
Use: "logs",
|
||||||
Short: "Aggregate journald logs across nodes (REQ-117)",
|
Short: "Aggregate journald logs across nodes (REQ-117)",
|
||||||
@@ -139,6 +151,25 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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)
|
ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -158,7 +189,7 @@ Ctrl-C cancels the fan-out via signal.NotifyContext.`,
|
|||||||
|
|
||||||
out := cmd.OutOrStdout()
|
out := cmd.OutOrStdout()
|
||||||
multi := len(nodes) > 1
|
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 {
|
if logsJSON {
|
||||||
raw, _ := json.Marshal(line)
|
raw, _ := json.Marshal(line)
|
||||||
fmt.Fprintln(out, string(raw))
|
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
|
// JSON entry immediately. The stream ends when every node has
|
||||||
// completed (or the context is cancelled). The caller drives the
|
// completed (or the context is cancelled). The caller drives the
|
||||||
// iteration via range-over-func (D-017 iter.Seq pattern).
|
// 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) {
|
return func(yield func(LogLine) bool) {
|
||||||
merged := make(chan LogLine)
|
merged := make(chan LogLine)
|
||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
@@ -235,7 +266,7 @@ func streamLogs(ctx context.Context, ex logsExecer, nodes []*model.Node, since t
|
|||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func(n *model.Node) {
|
go func(n *model.Node) {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
streamNodeLines(ctx, ex, n, since, job, merged)
|
streamNodeLines(ctx, ex, n, since, job, lines, merged)
|
||||||
}(n)
|
}(n)
|
||||||
}
|
}
|
||||||
done := make(chan struct{})
|
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
|
// context is cancelled); the caller is responsible for waiting on the
|
||||||
// goroutine. Send is non-blocking via select on ctx.Done so a slow
|
// goroutine. Send is non-blocking via select on ctx.Done so a slow
|
||||||
// consumer does not stall the fanout forever.
|
// 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)
|
peer := peerAddrForNode(n)
|
||||||
if peer == "" {
|
if peer == "" {
|
||||||
slog.Default().Warn("logs: cannot resolve SSH address for node", "node", n.Name)
|
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 + "-*"
|
unitPattern = "orca-alloc-" + job + "-*"
|
||||||
}
|
}
|
||||||
sinceStr := since.Format("2006-01-02 15:04:05")
|
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
|
// F1: shellQuote (single-quote wrap) instead of %q — %q does not
|
||||||
// escape backticks, enabling command substitution in double quotes.
|
// 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)
|
raw, err := ex.Exec(ctx, peer, cmd)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
slog.Default().Warn("logs: exec failed", "node", n.Name, "peer", peer, "error", err)
|
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().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(&logsNode, "node", "", "restrict to a single node (name or id)")
|
||||||
logsCmd.Flags().StringVar(&logsJob, "job", "", "filter by job name (matches orca-alloc-<name>-* units)")
|
logsCmd.Flags().StringVar(&logsJob, "job", "", "filter by job name (matches orca-alloc-<name>-* 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)")
|
logsCmd.Flags().BoolVar(&logsJSON, "json", false, "output raw JSON (one LogLine per line)")
|
||||||
rootCmd.AddCommand(logsCmd)
|
rootCmd.AddCommand(logsCmd)
|
||||||
}
|
}
|
||||||
|
|||||||
+156
-1
@@ -63,6 +63,18 @@ func (m *mockLogsExec) countCalls(match string) int {
|
|||||||
return n
|
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
|
// logsTestEnv wires a mockLogsExec into logsExecOverride and returns
|
||||||
// the mock + a cleanup func. Tests MUST defer the cleanup.
|
// the mock + a cleanup func. Tests MUST defer the cleanup.
|
||||||
func logsTestEnv(t *testing.T) *mockLogsExec {
|
func logsTestEnv(t *testing.T) *mockLogsExec {
|
||||||
@@ -325,7 +337,7 @@ func TestLogsCancelStopsStream(t *testing.T) {
|
|||||||
{ID: "n1", Name: "cancelnode", Address: "cancelnode:8443"},
|
{ID: "n1", Name: "cancelnode", Address: "cancelnode:8443"},
|
||||||
}
|
}
|
||||||
consumed := 0
|
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++
|
consumed++
|
||||||
}
|
}
|
||||||
if consumed > 1 {
|
if consumed > 1 {
|
||||||
@@ -357,3 +369,146 @@ func TestLogsParseJournalLine_InvalidJSON(t *testing.T) {
|
|||||||
t.Error("expected error for invalid json, got nil")
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -57,6 +57,11 @@ func resetCommandFlags() {
|
|||||||
driftConfigPath = ""
|
driftConfigPath = ""
|
||||||
driftRemediateForce = false
|
driftRemediateForce = false
|
||||||
jobRestartPeer = ""
|
jobRestartPeer = ""
|
||||||
|
jobStopPeer = ""
|
||||||
|
jobStopTimeout = 0
|
||||||
|
jobStopTransportOverride = nil
|
||||||
|
logsLines = logsDefaultLines
|
||||||
|
cutoverFSOverride = nil
|
||||||
jobLintExplain = false
|
jobLintExplain = false
|
||||||
jobLintFormat = "text"
|
jobLintFormat = "text"
|
||||||
jobVerifyLead = ""
|
jobVerifyLead = ""
|
||||||
|
|||||||
+99
-4
@@ -69,6 +69,41 @@ var upgradeTransportOverride upgradeTransport
|
|||||||
// peers to create the orca user on. Returns a list of peer addresses.
|
// peers to create the orca user on. Returns a list of peer addresses.
|
||||||
var peersListerOverride func() ([]string, error)
|
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{
|
var upgradeCmd = &cobra.Command{
|
||||||
Use: "upgrade",
|
Use: "upgrade",
|
||||||
Short: "Upgrade orca to a new version (REQ-115, R-017 cutover)",
|
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)
|
// return 200. On failure, rolls back (restores :443, removes nft rules)
|
||||||
// and returns (false, nil). On success returns (true, nil). With
|
// and returns (false, nil). On success returns (true, nil). With
|
||||||
// force=true, verification is skipped.
|
// 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) {
|
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 {
|
cfs := cutoverFSFromCtx()
|
||||||
return false, fmt.Errorf("cutover: edit traefik.yml: %w", err)
|
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 {
|
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)
|
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`
|
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 {
|
if force {
|
||||||
fmt.Fprintf(out, " --force: skipping cutover verification\n")
|
fmt.Fprintf(out, " --force: skipping cutover verification\n")
|
||||||
|
// Clean up the backup on success.
|
||||||
|
_ = cfs.Remove(backupPath)
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -336,6 +405,8 @@ func performCutover(ctx context.Context, runner commandRunner, out interface{ Wr
|
|||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
fmt.Fprintf(out, " ✓ C-25 cutover verification passed (200 from Traefik)\n")
|
fmt.Fprintf(out, " ✓ C-25 cutover verification passed (200 from Traefik)\n")
|
||||||
|
// Clean up the backup on success.
|
||||||
|
_ = cfs.Remove(backupPath)
|
||||||
return true, nil
|
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.
|
// 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 {
|
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 {
|
cfs := cutoverFSFromCtx()
|
||||||
return fmt.Errorf("rollback: edit traefik.yml: %w", err)
|
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 {
|
if _, err := runner.Run(ctx, "systemctl", "restart", "traefik"); err != nil {
|
||||||
return fmt.Errorf("rollback: restart traefik: %w", err)
|
return fmt.Errorf("rollback: restart traefik: %w", err)
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.cloudinit.dev/coreci/orca/internal/migration"
|
"git.cloudinit.dev/coreci/orca/internal/migration"
|
||||||
"git.cloudinit.dev/coreci/orca/internal/paths"
|
"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
|
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:<path>", "write:<path>", "rename:<old>", "stat:<path>"
|
||||||
|
}
|
||||||
|
|
||||||
|
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) {
|
func setupUpgradeTest(t *testing.T) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
t.Setenv("ORCA_HOME", t.TempDir())
|
t.Setenv("ORCA_HOME", t.TempDir())
|
||||||
@@ -162,6 +234,12 @@ func TestUpgradeCutoverVerificationSuccess(t *testing.T) {
|
|||||||
upgradeRunnerOverride = runner
|
upgradeRunnerOverride = runner
|
||||||
httpClientOverride = func(url string) (int, error) { return 200, nil }
|
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"})
|
rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"})
|
||||||
if err := rootCmd.Execute(); err != nil {
|
if err := rootCmd.Execute(); err != nil {
|
||||||
t.Fatalf("upgrade with cutover: %v", err)
|
t.Fatalf("upgrade with cutover: %v", err)
|
||||||
@@ -180,6 +258,12 @@ func TestUpgradeCutoverRollback(t *testing.T) {
|
|||||||
upgradeRunnerOverride = runner
|
upgradeRunnerOverride = runner
|
||||||
httpClientOverride = func(url string) (int, error) { return 502, nil }
|
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"})
|
rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0"})
|
||||||
err := rootCmd.Execute()
|
err := rootCmd.Execute()
|
||||||
if err == nil {
|
if err == nil {
|
||||||
@@ -194,20 +278,27 @@ func TestUpgradeCutoverRollback(t *testing.T) {
|
|||||||
t.Errorf("output should mention rollback: %s", out)
|
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
|
foundRollback := false
|
||||||
for _, call := range runner.calls {
|
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" {
|
if call.name == "nft" && len(call.args) >= 2 && call.args[0] == "delete" {
|
||||||
foundRollback = true
|
foundRollback = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if !foundRollback {
|
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
|
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"})
|
rootCmd.SetArgs([]string{"upgrade", "--to", "v0.11.0", "--force"})
|
||||||
if err := rootCmd.Execute(); err != nil {
|
if err := rootCmd.Execute(); err != nil {
|
||||||
t.Fatalf("upgrade with --force: %v", err)
|
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)
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user