From c8cf2e41e5ae502657f28cec92ead5c16bcdcc76 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Fri, 7 Aug 2026 05:36:30 +0000 Subject: [PATCH] feat(P06): alloc history (REQ-071) + logs --all-nodes (REQ-117) internal/store/alloc_history.go: AllocHistoryRepo (Record/List/Evict) in orca_cache.db with 7-day TTL eviction goroutine. internal/cli/logs.go: orca logs --all-nodes --since 5m with iter.Seq streaming, SSH fanout, journalctl JSON parsing, signal.NotifyContext cancellation, --json output. ---ci--- project: orca phase: 06 milestone: v0.11 status: execute ---/ci--- --- internal/cli/logs.go | 321 ++++++++++++++++++++++++ internal/cli/logs_test.go | 359 +++++++++++++++++++++++++++ internal/cli/namespace_test.go | 2 + internal/store/alloc_history.go | 198 +++++++++++++++ internal/store/alloc_history_test.go | 345 +++++++++++++++++++++++++ 5 files changed, 1225 insertions(+) create mode 100644 internal/cli/logs.go create mode 100644 internal/cli/logs_test.go create mode 100644 internal/store/alloc_history.go create mode 100644 internal/store/alloc_history_test.go diff --git a/internal/cli/logs.go b/internal/cli/logs.go new file mode 100644 index 0000000..6d7a987 --- /dev/null +++ b/internal/cli/logs.go @@ -0,0 +1,321 @@ +package cli + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "iter" + "log/slog" + "os" + "os/signal" + "strings" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/sshpush" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// logsExecer is the SSH command-execution seam used by `orca logs`. +// *sshpush.Transport satisfies it via Exec; tests inject a mock +// without a real SSH server (same pattern as drain.go / stepca mockExec). +type logsExecer interface { + Exec(ctx context.Context, peer string, cmd string) ([]byte, error) +} + +// logsExecOverride is the package-level exec seam. When non-nil it +// replaces the production transport; tests set it and restore nil in +// cleanup. nil means "build the real transport on first use". +var logsExecOverride logsExecer + +func logsExecFromCtx(_ context.Context) (logsExecer, error) { + if logsExecOverride != nil { + return logsExecOverride, nil + } + keyPath := certpaths.SSHKeyPath() + khPath := certpaths.KnownHostsPath() + return sshpush.NewTransport(keyPath, khPath), nil +} + +// LogLine is a single journald log entry parsed from journalctl --output +// json. Host is the peer the line came from (set by the aggregator). +type LogLine struct { + Host string `json:"host"` + Timestamp time.Time `json:"timestamp"` + Unit string `json:"unit"` + Message string `json:"message"` + Priority string `json:"priority"` +} + +// journalRaw is the subset of journalctl --output json fields we +// decode. Extra fields are ignored. +type journalRaw struct { + Realtime int64 `json:"__REALTIME_TIMESTAMP"` + Unit string `json:"_SYSTEMD_UNIT"` + Identifier string `json:"SYSLOG_IDENTIFIER"` + Comm string `json:"_COMM"` + Message string `json:"MESSAGE"` + Priority any `json:"PRIORITY"` +} + +func (j journalRaw) unit() string { + if j.Unit != "" { + return strings.TrimSuffix(j.Unit, ".service") + } + if j.Identifier != "" { + return j.Identifier + } + if j.Comm != "" { + return j.Comm + } + return "" +} + +func (j journalRaw) priority() string { + switch p := j.Priority.(type) { + case string: + return p + case float64: + return fmt.Sprintf("%v", int(p)) + default: + return "" + } +} + +func (j journalRaw) timestamp() time.Time { + if j.Realtime == 0 { + return time.Time{} + } + return time.Unix(0, j.Realtime).UTC() +} + +var ( + logsAllNodes bool + logsNode string + logsJob string + logsSince string + logsJSON bool +) + +var logsCmd = &cobra.Command{ + Use: "logs", + Short: "Aggregate journald logs across nodes (REQ-117)", + Long: `Aggregate journald logs across registered nodes via SSH fanout. + + orca logs --all-nodes --since 5m + orca logs --node web-1 --since 1h + orca logs --all-nodes --job web --since 30m --json + +Runs 'journalctl -u 'orca-alloc-*' --since --output json' on each +peer, parses the JSON-per-line output, and streams the entries with a +[] prefix (multi-node) or raw (single-node). --json outputs +the raw journalctl JSON lines verbatim. + +Use --all-nodes to fan out to every registered node, or --node +for a single node. --job filters units to orca-alloc--*. + +Ctrl-C cancels the fan-out via signal.NotifyContext.`, + RunE: func(cmd *cobra.Command, args []string) error { + if !logsAllNodes && logsNode == "" { + return fmt.Errorf("specify --all-nodes or --node ") + } + if logsAllNodes && logsNode != "" { + return fmt.Errorf("--all-nodes and --node are mutually exclusive") + } + since, err := parseSince(logsSince) + if err != nil { + return err + } + + ctx, cancel := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer cancel() + + nodes, err := resolveLogNodes(ctx) + if err != nil { + return err + } + if len(nodes) == 0 { + return fmt.Errorf("no nodes to query") + } + + ex, err := logsExecFromCtx(cmd.Context()) + if err != nil { + return fmt.Errorf("ssh transport: %w", err) + } + + out := cmd.OutOrStdout() + multi := len(nodes) > 1 + for line := range streamLogs(ctx, ex, nodes, since, logsJob) { + if logsJSON { + raw, _ := json.Marshal(line) + fmt.Fprintln(out, string(raw)) + continue + } + if multi { + fmt.Fprintf(out, "[%s] %s %s\n", line.Host, line.Timestamp.Format(time.RFC3339), line.Message) + } else { + fmt.Fprintln(out, line.Message) + } + } + return nil + }, +} + +// parseSince parses a duration string like "5m", "1h30m", "500ms". The +// returned time is time.Now().UTC().Add(-d). An empty string defaults +// to 5 minutes. +func parseSince(s string) (time.Time, error) { + if s == "" { + s = "5m" + } + d, err := time.ParseDuration(s) + if err != nil { + return time.Time{}, fmt.Errorf("--since %q: %w", s, err) + } + if d <= 0 { + return time.Time{}, fmt.Errorf("--since must be positive, got %s", d) + } + return time.Now().UTC().Add(-d), nil +} + +// resolveLogNodes returns the set of nodes to query. For --all-nodes it +// lists every registered node; for --node it resolves a single +// node by name or id. +func resolveLogNodes(ctx context.Context) ([]*model.Node, error) { + db, closer, err := openDB() + if err != nil { + return nil, err + } + defer closer() + reg := store.NewNodeRepo(db) + if logsAllNodes { + return reg.List(ctx) + } + n, err := reg.GetByName(ctx, logsNode) + if err == nil { + return []*model.Node{n}, nil + } + if !errors.Is(err, store.ErrNotFound) { + return nil, fmt.Errorf("lookup node %q: %w", logsNode, err) + } + n, err = reg.Get(ctx, logsNode) + if err == nil { + return []*model.Node{n}, nil + } + if errors.Is(err, store.ErrNotFound) { + return nil, fmt.Errorf("node %q not found", logsNode) + } + return nil, fmt.Errorf("lookup node %q: %w", logsNode, err) +} + +// streamLogs fans out journalctl across nodes and yields parsed +// LogLine values as they arrive. It runs each node's exec in its own +// goroutine, scans the output line-by-line, and yields each parsed +// 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] { + return func(yield func(LogLine) bool) { + merged := make(chan LogLine) + var wg sync.WaitGroup + for _, n := range nodes { + wg.Add(1) + go func(n *model.Node) { + defer wg.Done() + streamNodeLines(ctx, ex, n, since, job, merged) + }(n) + } + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + // Pump merged lines to the yield function until either all + // nodes finish or the consumer stops pulling (yield==false) + // or the context is cancelled. + for { + select { + case <-done: + return + case <-ctx.Done(): + return + case line := <-merged: + if !yield(line) { + return + } + } + } + } +} + +// streamNodeLines runs journalctl on a single node and pushes each +// parsed line into out. It blocks until the exec completes (or 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 +// 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) { + peer := peerAddrForNode(n) + if peer == "" { + slog.Default().Warn("logs: cannot resolve SSH address for node", "node", n.Name) + return + } + unitPattern := "orca-alloc-*" + if job != "" { + unitPattern = "orca-alloc-" + job + "-*" + } + sinceStr := since.Format("2006-01-02 15:04:05") + cmd := fmt.Sprintf("journalctl -u %q --since %q --output json --no-pager", unitPattern, sinceStr) + raw, err := ex.Exec(ctx, peer, cmd) + if err != nil { + slog.Default().Warn("logs: exec failed", "node", n.Name, "peer", peer, "error", err) + return + } + host := n.Name + for _, line := range strings.Split(string(raw), "\n") { + line = strings.TrimSpace(line) + if line == "" { + continue + } + ll, perr := parseJournalLine(line) + if perr != nil { + continue + } + ll.Host = host + select { + case out <- ll: + case <-ctx.Done(): + return + } + } +} + +// parseJournalLine decodes a single journalctl --output json line into +// a LogLine. Unknown fields are ignored. +func parseJournalLine(s string) (LogLine, error) { + var j journalRaw + if err := json.Unmarshal([]byte(s), &j); err != nil { + return LogLine{}, fmt.Errorf("parse journal line: %w", err) + } + return LogLine{ + Timestamp: j.timestamp(), + Unit: j.unit(), + Message: j.Message, + Priority: j.priority(), + }, nil +} + +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().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 new file mode 100644 index 0000000..19573b6 --- /dev/null +++ b/internal/cli/logs_test.go @@ -0,0 +1,359 @@ +package cli + +import ( + "bytes" + "context" + "strconv" + "strings" + "sync" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// mockLogsExec is a record-and-replay execer for `orca logs` tests +// (same pattern as internal/stepca mockExec / drain_test mockDrainExec). +// It matches each incoming command against a list of (substring, +// output) responses; the first match wins. An entry with an empty +// substring matches any command. +type mockLogsExec struct { + mu sync.Mutex + responses []logsMockResp + calls []logsMockCall +} + +type logsMockResp struct { + match string + peer string // if set, must match the peer too + out string +} + +type logsMockCall struct { + peer string + cmd string +} + +func (m *mockLogsExec) Exec(_ context.Context, peer, cmd string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.calls = append(m.calls, logsMockCall{peer: peer, cmd: cmd}) + for _, r := range m.responses { + if r.peer != "" && !strings.Contains(peer, r.peer) { + continue + } + if r.match == "" || strings.Contains(cmd, r.match) { + return []byte(r.out), nil + } + } + return nil, nil +} + +func (m *mockLogsExec) countCalls(match string) int { + m.mu.Lock() + defer m.mu.Unlock() + n := 0 + for _, c := range m.calls { + if strings.Contains(c.cmd, match) { + n++ + } + } + return n +} + +// logsTestEnv wires a mockLogsExec into logsExecOverride and returns +// the mock + a cleanup func. Tests MUST defer the cleanup. +func logsTestEnv(t *testing.T) *mockLogsExec { + t.Helper() + prev := logsExecOverride + mx := &mockLogsExec{} + logsExecOverride = mx + t.Cleanup(func() { logsExecOverride = prev }) + return mx +} + +// logsNodeForTest inserts a node with a fixed name and returns it so +// logs commands can target it by name. Uses the test ORCA_HOME db. +func logsNodeForTest(t *testing.T, name, addr string) *model.Node { + t.Helper() + db, err := store.Open(certpaths.DBPath()) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + n := &model.Node{ + ID: "node-" + name, + Name: name, + Address: addr, + State: model.NodeStateReady, + JoinedAt: time.Now().UTC(), + LastSeen: time.Now().UTC(), + Kind: string(model.NodeKindLinux), + } + if err := store.NewNodeRepo(db).Insert(context.Background(), n); err != nil { + t.Fatalf("insert node: %v", err) + } + return n +} + +// journalJSONLine renders a single journalctl --output json line. +func journalJSONLine(ts time.Time, unit, msg, prio string) string { + return `{"__REALTIME_TIMESTAMP":` + strconv.FormatInt(ts.UnixNano(), 10) + + `,"_SYSTEMD_UNIT":"` + unit + `.service","MESSAGE":"` + msg + `","PRIORITY":"` + prio + `"}` +} + +// runLogsCmd executes logsCmd with the given args against a buffered +// stdout, returning the captured output and any error. +func runLogsCmd(t *testing.T, args []string) (string, error) { + t.Helper() + resetRootFlags(t) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs(args) + err := rootCmd.Execute() + return buf.String(), err +} + +func TestLogsCommandRegistered(t *testing.T) { + found := false + for _, cmd := range rootCmd.Commands() { + if cmd.Name() == "logs" { + found = true + break + } + } + if !found { + t.Fatal("logs command not registered on rootCmd") + } +} + +func TestLogsRequiresAllNodesOrNode(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + resetRootFlags(t) + + _, err := runLogsCmd(t, []string{"logs"}) + if err == nil { + t.Fatal("expected error when neither --all-nodes nor --node is set") + } + if !strings.Contains(err.Error(), "--all-nodes") && !strings.Contains(err.Error(), "--node") { + t.Errorf("error = %q, want mention of --all-nodes/--node", err) + } +} + +func TestLogsAllNodesAndNodeMutuallyExclusive(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + resetRootFlags(t) + + _, err := runLogsCmd(t, []string{"logs", "--all-nodes", "--node", "x"}) + if err == nil { + t.Fatal("expected error for --all-nodes + --node together") + } +} + +func TestLogsSinceParsing(t *testing.T) { + cases := []struct { + in string + wantErr bool + }{ + {"5m", false}, + {"1h30m", false}, + {"500ms", false}, + {"", false}, // defaults to 5m + {"notaduration", true}, + {"-5m", true}, + {"0s", true}, + } + for _, c := range cases { + _, err := parseSince(c.in) + if c.wantErr && err == nil { + t.Errorf("parseSince(%q): expected error, got nil", c.in) + } + if !c.wantErr && err != nil { + t.Errorf("parseSince(%q): unexpected error: %v", c.in, err) + } + } +} + +func TestLogsSinceDefaultIs5m(t *testing.T) { + before := time.Now().UTC() + got, err := parseSince("") + if err != nil { + t.Fatalf("parseSince(empty): %v", err) + } + after := time.Now().UTC() + // got should be ~5m ago. The call's "now" is in [before, after], + // so got = now-5m is in [before-5m, after-5m]. + lo := before.Add(-5 * time.Minute) + hi := after.Add(-5 * time.Minute) + if got.Before(lo) || got.After(hi) { + t.Errorf("parseSince(empty) = %v, want ~5m ago (between %v and %v)", got, lo, hi) + } +} + +func TestLogsAllNodes_FansOutAndStreams(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "web-1", "web-1:8443") + logsNodeForTest(t, "web-2", "web-2:8443") + mx := logsTestEnv(t) + + ts := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + // Two nodes, each returns one journal line. + mx.responses = []logsMockResp{ + {peer: "web-1", match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "hello from web-1", "6") + "\n"}, + {peer: "web-2", match: "journalctl", out: journalJSONLine(ts, "orca-alloc-web-0", "hello from web-2", "6") + "\n"}, + } + + out, err := runLogsCmd(t, []string{"logs", "--all-nodes", "--since", "5m"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + if mx.countCalls("journalctl") != 2 { + t.Errorf("expected 2 journalctl calls, got %d", mx.countCalls("journalctl")) + } + if !strings.Contains(out, "hello from web-1") { + t.Errorf("output missing web-1 line:\n%s", out) + } + if !strings.Contains(out, "hello from web-2") { + t.Errorf("output missing web-2 line:\n%s", out) + } + if !strings.Contains(out, "[web-1]") || !strings.Contains(out, "[web-2]") { + t.Errorf("output missing [host] prefix for multi-node:\n%s", out) + } +} + +func TestLogsSingleNode_NoHostPrefix(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "solo", "solo: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", "single node msg", "6") + "\n"}, + } + + out, err := runLogsCmd(t, []string{"logs", "--node", "solo", "--since", "1h"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + if !strings.Contains(out, "single node msg") { + t.Errorf("output missing message:\n%s", out) + } + if strings.Contains(out, "[solo]") { + t.Errorf("single-node output should NOT have [host] prefix:\n%s", out) + } + if mx.countCalls("journalctl") != 1 { + t.Errorf("expected 1 journalctl call, got %d", mx.countCalls("journalctl")) + } +} + +func TestLogsJSONOutput(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "jsonnode", "jsonnode: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 line", "6") + "\n"}, + } + + out, err := runLogsCmd(t, []string{"logs", "--node", "jsonnode", "--since", "1m", "--json"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + if !strings.Contains(out, `"message":"json line"`) { + t.Errorf("json output missing message field:\n%s", out) + } + if !strings.Contains(out, `"host":"jsonnode"`) { + t.Errorf("json output missing host field:\n%s", out) + } +} + +func TestLogsJobFilterAffectsUnitPattern(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "jobnode", "jobnode:8443") + mx := logsTestEnv(t) + + mx.responses = []logsMockResp{ + {match: "orca-alloc-web-*", out: ""}, + } + + _, err := runLogsCmd(t, []string{"logs", "--node", "jobnode", "--job", "web", "--since", "1m"}) + if err != nil { + t.Fatalf("logs: %v", err) + } + if mx.countCalls("orca-alloc-web-*") != 1 { + t.Errorf("expected unit pattern orca-alloc-web-* in cmd, calls=%v", mx.calls) + } + if mx.countCalls("orca-alloc-*") != 1 { + // The job-specific pattern is a subset of the bare pattern; + // substring match counts both. Ensure the job pattern was used. + } +} + +func TestLogsCancelStopsStream(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + + logsNodeForTest(t, "cancelnode", "cancelnode:8443") + mx := logsTestEnv(t) + + mx.responses = []logsMockResp{ + {match: "journalctl", out: journalJSONLine(time.Now().UTC(), "orca-alloc-x", "msg", "6") + "\n"}, + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already-cancelled context: streamLogs should return immediately + + ex, err := logsExecFromCtx(ctx) + if err != nil { + t.Fatalf("exec: %v", err) + } + nodes := []*model.Node{ + {ID: "n1", Name: "cancelnode", Address: "cancelnode:8443"}, + } + consumed := 0 + for range streamLogs(ctx, ex, nodes, time.Now().UTC().Add(-1*time.Minute), "") { + consumed++ + } + if consumed > 1 { + t.Errorf("cancelled stream yielded %d lines, want <= 1", consumed) + } +} + +func TestLogsParseJournalLine(t *testing.T) { + ts := time.Date(2026, 1, 1, 12, 0, 0, 123456789, time.UTC) + raw := journalJSONLine(ts, "orca-alloc-web-0", "hello world", "6") + ll, err := parseJournalLine(raw) + if err != nil { + t.Fatalf("parseJournalLine: %v", err) + } + if ll.Message != "hello world" { + t.Errorf("message = %q, want hello world", ll.Message) + } + if ll.Unit != "orca-alloc-web-0" { + t.Errorf("unit = %q, want orca-alloc-web-0", ll.Unit) + } + if ll.Priority != "6" { + t.Errorf("priority = %q, want 6", ll.Priority) + } +} + +func TestLogsParseJournalLine_InvalidJSON(t *testing.T) { + _, err := parseJournalLine("not json") + if err == nil { + t.Error("expected error for invalid json, got nil") + } +} diff --git a/internal/cli/namespace_test.go b/internal/cli/namespace_test.go index 7901e95..9bba178 100644 --- a/internal/cli/namespace_test.go +++ b/internal/cli/namespace_test.go @@ -37,6 +37,7 @@ func resetCommandFlags() { stopID, runTarget, runIDKey, jobWatch = "", "", "", false migrateTarget = "" drainTimeout = 30 * time.Second + logsAllNodes, logsNode, logsJob, logsSince, logsJSON = false, "", "", "5m", false capSetCPU, capSetMem, capSetDisk, capNodeID = 0, 0, 0, "" auditLimit = 50 backupOutPath, restoreInPath, restoreTargetDir = "", "", "" @@ -48,6 +49,7 @@ func resetCommandFlags() { for _, c := range []*cobra.Command{ nodeDrainCmd, daemonCmd, daemonDrainAndStopCmd, jobCmd, jobMigrateCmd, jobRunCmd, jobListCmd, jobStopCmd, jobLogsCmd, + logsCmd, } { if c != nil { c.SetOut(nil) diff --git a/internal/store/alloc_history.go b/internal/store/alloc_history.go new file mode 100644 index 0000000..d9fb408 --- /dev/null +++ b/internal/store/alloc_history.go @@ -0,0 +1,198 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + "log/slog" + "strings" + "time" +) + +const DefaultAllocHistoryTTL = 7 * 24 * time.Hour + +var DefaultEvictorInterval = 1 * time.Hour + +type AllocHistoryEntry struct { + AllocID string `json:"alloc_id"` + JobID string `json:"job_id"` + NodeID string `json:"node_id"` + Namespace string `json:"namespace"` + FromState string `json:"from_state,omitempty"` + ToState string `json:"to_state"` + Timestamp time.Time `json:"timestamp"` + Reason string `json:"reason,omitempty"` +} + +type HistoryFilter struct { + JobID string + NodeID string + Namespace string + Since time.Time +} + +type AllocHistoryRepo struct { + db *sql.DB +} + +func NewAllocHistoryRepo(db *sql.DB) *AllocHistoryRepo { + return &AllocHistoryRepo{db: db} +} + +const allocHistorySchema = ` +CREATE TABLE IF NOT EXISTS alloc_history ( + alloc_id TEXT NOT NULL, + job_id TEXT NOT NULL, + node_id TEXT NOT NULL, + namespace TEXT NOT NULL, + from_state TEXT, + to_state TEXT NOT NULL, + timestamp INTEGER NOT NULL, + reason TEXT +); +CREATE INDEX IF NOT EXISTS idx_alloc_history_ts ON alloc_history(timestamp); +CREATE INDEX IF NOT EXISTS idx_alloc_history_job ON alloc_history(job_id); +` + +func (r *AllocHistoryRepo) EnsureSchema(ctx context.Context) error { + if _, err := r.db.ExecContext(ctx, allocHistorySchema); err != nil { + return fmt.Errorf("alloc history schema: %w", err) + } + return nil +} + +func (r *AllocHistoryRepo) Record(ctx context.Context, entry AllocHistoryEntry) error { + if err := r.EnsureSchema(ctx); err != nil { + return err + } + if entry.Timestamp.IsZero() { + entry.Timestamp = time.Now().UTC() + } + ts := entry.Timestamp.UnixNano() + _, err := r.db.ExecContext(ctx, + `INSERT INTO alloc_history (alloc_id, job_id, node_id, namespace, from_state, to_state, timestamp, reason) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + entry.AllocID, entry.JobID, entry.NodeID, entry.Namespace, + nullableStr(entry.FromState), entry.ToState, ts, nullableStr(entry.Reason)) + if err != nil { + return fmt.Errorf("alloc history record: %w", err) + } + return nil +} + +func (r *AllocHistoryRepo) List(ctx context.Context, filter HistoryFilter) ([]AllocHistoryEntry, error) { + if err := r.EnsureSchema(ctx); err != nil { + return nil, err + } + q := `SELECT alloc_id, job_id, node_id, namespace, + COALESCE(from_state, ''), to_state, timestamp, COALESCE(reason, '') + FROM alloc_history` + var ( + conds []string + args []any + ) + if filter.JobID != "" { + conds = append(conds, "job_id = ?") + args = append(args, filter.JobID) + } + if filter.NodeID != "" { + conds = append(conds, "node_id = ?") + args = append(args, filter.NodeID) + } + if filter.Namespace != "" { + conds = append(conds, "namespace = ?") + args = append(args, filter.Namespace) + } + if !filter.Since.IsZero() { + conds = append(conds, "timestamp >= ?") + args = append(args, filter.Since.UnixNano()) + } + if len(conds) > 0 { + q += " WHERE " + strings.Join(conds, " AND ") + } + q += " ORDER BY timestamp ASC" + + rows, err := r.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("alloc history list: %w", err) + } + defer rows.Close() + var out []AllocHistoryEntry + for rows.Next() { + var ( + e AllocHistoryEntry + ts int64 + from string + reas string + ) + if err := rows.Scan(&e.AllocID, &e.JobID, &e.NodeID, &e.Namespace, + &from, &e.ToState, &ts, &reas); err != nil { + return nil, fmt.Errorf("alloc history scan: %w", err) + } + e.FromState = from + e.Reason = reas + e.Timestamp = time.Unix(0, ts).UTC() + out = append(out, e) + } + return out, rows.Err() +} + +func (r *AllocHistoryRepo) Evict(ctx context.Context, before time.Time) (int64, error) { + if err := r.EnsureSchema(ctx); err != nil { + return 0, err + } + res, err := r.db.ExecContext(ctx, + `DELETE FROM alloc_history WHERE timestamp < ?`, before.UnixNano()) + if err != nil { + return 0, fmt.Errorf("alloc history evict: %w", err) + } + n, _ := res.RowsAffected() + return n, nil +} + +func (r *AllocHistoryRepo) StartEvictor(ctx context.Context, ttl, interval time.Duration) (stop func()) { + if ttl <= 0 { + ttl = DefaultAllocHistoryTTL + } + if interval <= 0 { + interval = DefaultEvictorInterval + } + ctx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(interval) + defer ticker.Stop() + sweep := func() { + cutoff := time.Now().UTC().Add(-ttl) + n, err := r.Evict(ctx, cutoff) + if err != nil { + slog.Default().Warn("alloc history evictor: sweep failed", "error", err) + return + } + if n > 0 { + slog.Default().Info("alloc history evictor: evicted", "count", n, "older_than", cutoff.Format(time.RFC3339)) + } + } + sweep() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + sweep() + } + } + }() + return func() { + cancel() + <-done + } +} + +func nullableStr(s string) any { + if s == "" { + return nil + } + return s +} diff --git a/internal/store/alloc_history_test.go b/internal/store/alloc_history_test.go new file mode 100644 index 0000000..2dab6d1 --- /dev/null +++ b/internal/store/alloc_history_test.go @@ -0,0 +1,345 @@ +package store + +import ( + "context" + "database/sql" + "path/filepath" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +func openCacheTestDB(t *testing.T) (*AllocHistoryRepo, *sql.DB, func()) { + t.Helper() + path := filepath.Join(t.TempDir(), "cache.db") + db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)") + if err != nil { + t.Fatalf("open cache sqlite: %v", err) + } + if err := db.Ping(); err != nil { + _ = db.Close() + t.Fatalf("ping cache sqlite: %v", err) + } + return NewAllocHistoryRepo(db), db, func() { _ = db.Close() } +} + +func TestAllocHistory_RecordAndList(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + + e1 := AllocHistoryEntry{ + AllocID: "alloc-1", JobID: "job-1", NodeID: "node-1", + Namespace: "default", FromState: "", ToState: "created", + Timestamp: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC), Reason: "dispatch", + } + if err := repo.Record(ctx, e1); err != nil { + t.Fatalf("record: %v", err) + } + e2 := AllocHistoryEntry{ + AllocID: "alloc-1", JobID: "job-1", NodeID: "node-1", + Namespace: "default", FromState: "created", ToState: "started", + Timestamp: time.Date(2026, 1, 1, 12, 1, 0, 0, time.UTC), + } + if err := repo.Record(ctx, e2); err != nil { + t.Fatalf("record 2: %v", err) + } + + got, err := repo.List(ctx, HistoryFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 entries, got %d", len(got)) + } + if got[0].ToState != "created" || got[1].ToState != "started" { + t.Errorf("order/states wrong: %+v", got) + } + if got[0].FromState != "" { + t.Errorf("first entry from_state = %q, want empty", got[0].FromState) + } + if got[1].FromState != "created" { + t.Errorf("second entry from_state = %q, want created", got[1].FromState) + } + if !got[0].Timestamp.Equal(e1.Timestamp) { + t.Errorf("ts = %v, want %v", got[0].Timestamp, e1.Timestamp) + } + if got[0].Reason != "dispatch" { + t.Errorf("reason = %q, want dispatch", got[0].Reason) + } +} + +func TestAllocHistory_ListFilterByJob(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + entries := []AllocHistoryEntry{ + {AllocID: "a1", JobID: "jobA", NodeID: "n1", Namespace: "ns", ToState: "created", Timestamp: base}, + {AllocID: "a2", JobID: "jobB", NodeID: "n1", Namespace: "ns", ToState: "created", Timestamp: base.Add(1 * time.Second)}, + {AllocID: "a3", JobID: "jobA", NodeID: "n2", Namespace: "ns", ToState: "created", Timestamp: base.Add(2 * time.Second)}, + } + for _, e := range entries { + if err := repo.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + got, err := repo.List(ctx, HistoryFilter{JobID: "jobA"}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 entries for jobA, got %d", len(got)) + } + for _, e := range got { + if e.JobID != "jobA" { + t.Errorf("job = %q, want jobA", e.JobID) + } + } +} + +func TestAllocHistory_ListFilterByNode(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + entries := []AllocHistoryEntry{ + {AllocID: "a1", JobID: "jobA", NodeID: "n1", Namespace: "ns", ToState: "created", Timestamp: base}, + {AllocID: "a2", JobID: "jobB", NodeID: "n1", Namespace: "ns", ToState: "created", Timestamp: base.Add(1 * time.Second)}, + {AllocID: "a3", JobID: "jobA", NodeID: "n2", Namespace: "ns", ToState: "created", Timestamp: base.Add(2 * time.Second)}, + } + for _, e := range entries { + if err := repo.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + got, err := repo.List(ctx, HistoryFilter{NodeID: "n2"}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 { + t.Fatalf("expected 1 entry for n2, got %d", len(got)) + } + if got[0].NodeID != "n2" { + t.Errorf("node = %q, want n2", got[0].NodeID) + } +} + +func TestAllocHistory_ListFilterBySince(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + entries := []AllocHistoryEntry{ + {AllocID: "a1", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "created", Timestamp: base}, + {AllocID: "a2", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "started", Timestamp: base.Add(10 * time.Second)}, + {AllocID: "a3", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "stopped", Timestamp: base.Add(20 * time.Second)}, + } + for _, e := range entries { + if err := repo.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + since := base.Add(5 * time.Second) + got, err := repo.List(ctx, HistoryFilter{Since: since}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 entries since %v, got %d", since, len(got)) + } + if got[0].ToState != "started" || got[1].ToState != "stopped" { + t.Errorf("expected started+stopped, got %+v", got) + } +} + +func TestAllocHistory_ListFilterByNamespace(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + entries := []AllocHistoryEntry{ + {AllocID: "a1", JobID: "j", NodeID: "n", Namespace: "prod", ToState: "created", Timestamp: base}, + {AllocID: "a2", JobID: "j", NodeID: "n", Namespace: "dev", ToState: "created", Timestamp: base.Add(1 * time.Second)}, + } + for _, e := range entries { + if err := repo.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + got, err := repo.List(ctx, HistoryFilter{Namespace: "prod"}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 { + t.Fatalf("expected 1 entry for prod, got %d", len(got)) + } + if got[0].Namespace != "prod" { + t.Errorf("ns = %q, want prod", got[0].Namespace) + } +} + +func TestAllocHistory_Evict(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + base := time.Now().UTC() + old := base.Add(-30 * 24 * time.Hour) + recent := base.Add(-1 * time.Hour) + entries := []AllocHistoryEntry{ + {AllocID: "old1", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "created", Timestamp: old}, + {AllocID: "old2", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "started", Timestamp: old.Add(1 * time.Second)}, + {AllocID: "new1", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "created", Timestamp: recent}, + } + for _, e := range entries { + if err := repo.Record(ctx, e); err != nil { + t.Fatalf("record: %v", err) + } + } + cutoff := base.Add(-7 * 24 * time.Hour) + n, err := repo.Evict(ctx, cutoff) + if err != nil { + t.Fatalf("evict: %v", err) + } + if n != 2 { + t.Errorf("evicted = %d, want 2", n) + } + got, err := repo.List(ctx, HistoryFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 { + t.Fatalf("after evict, expected 1 entry, got %d", len(got)) + } + if got[0].AllocID != "new1" { + t.Errorf("remaining alloc = %q, want new1", got[0].AllocID) + } +} + +func TestAllocHistory_EvictEmpty(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + n, err := repo.Evict(ctx, time.Now().UTC()) + if err != nil { + t.Fatalf("evict empty: %v", err) + } + if n != 0 { + t.Errorf("evict empty = %d, want 0", n) + } +} + +func TestAllocHistory_RecordDefaultsTimestamp(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + before := time.Now().UTC().Add(-1 * time.Second) + if err := repo.Record(ctx, AllocHistoryEntry{ + AllocID: "a", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "created", + }); err != nil { + t.Fatalf("record: %v", err) + } + got, err := repo.List(ctx, HistoryFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 { + t.Fatalf("expected 1, got %d", len(got)) + } + if got[0].Timestamp.Before(before) { + t.Errorf("default ts = %v, want >= %v", got[0].Timestamp, before) + } +} + +func TestAllocHistory_TTLEvictor(t *testing.T) { + prev := DefaultEvictorInterval + DefaultEvictorInterval = 20 * time.Millisecond + defer func() { DefaultEvictorInterval = prev }() + + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + + old := time.Now().UTC().Add(-30 * 24 * time.Hour) + if err := repo.Record(ctx, AllocHistoryEntry{ + AllocID: "old1", JobID: "j", NodeID: "n", Namespace: "ns", + ToState: "created", Timestamp: old, + }); err != nil { + t.Fatalf("record old: %v", err) + } + if err := repo.Record(ctx, AllocHistoryEntry{ + AllocID: "new1", JobID: "j", NodeID: "n", Namespace: "ns", + ToState: "created", Timestamp: time.Now().UTC(), + }); err != nil { + t.Fatalf("record new: %v", err) + } + + rootCtx, rootCancel := context.WithCancel(context.Background()) + stop := repo.StartEvictor(rootCtx, 7*24*time.Hour, 20*time.Millisecond) + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + got, err := repo.List(ctx, HistoryFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) == 1 { + break + } + time.Sleep(10 * time.Millisecond) + } + + stop() + rootCancel() + + got, err := repo.List(ctx, HistoryFilter{}) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 1 { + t.Fatalf("after evictor, expected 1 entry, got %d", len(got)) + } + if got[0].AllocID != "new1" { + t.Errorf("remaining = %q, want new1", got[0].AllocID) + } +} + +func TestAllocHistory_TTLEvictor_StopsOnCancel(t *testing.T) { + prev := DefaultEvictorInterval + DefaultEvictorInterval = 20 * time.Millisecond + defer func() { DefaultEvictorInterval = prev }() + + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx, cancel := context.WithCancel(context.Background()) + stop := repo.StartEvictor(ctx, 7*24*time.Hour, 20*time.Millisecond) + + done := make(chan struct{}) + go func() { + stop() + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("StartEvictor stop did not return within 2s") + } + cancel() +} + +func TestAllocHistory_EnsureSchemaIdempotent(t *testing.T) { + repo, _, cleanup := openCacheTestDB(t) + defer cleanup() + ctx := context.Background() + for i := 0; i < 3; i++ { + if err := repo.EnsureSchema(ctx); err != nil { + t.Fatalf("ensureSchema[%d]: %v", i, err) + } + } + if err := repo.Record(ctx, AllocHistoryEntry{ + AllocID: "x", JobID: "j", NodeID: "n", Namespace: "ns", ToState: "created", + }); err != nil { + t.Fatalf("record after repeated schema: %v", err) + } +}