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---
This commit is contained in:
Jon Chery
@@ -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 <dur> --output json' on each
|
||||
peer, parses the JSON-per-line output, and streams the entries with a
|
||||
[<hostname>] 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 <host>
|
||||
for a single node. --job <name> filters units to orca-alloc-<name>-*.
|
||||
|
||||
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 <host>")
|
||||
}
|
||||
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 <host> 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-<name>-* 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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user