From 2110c7dab41c63edda70938d23970f6f14cfa790 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Sat, 22 Aug 2026 10:52:13 +0100 Subject: [PATCH 1/2] Detect and alert on stuck jobs --- internal/alert/engine.go | 58 +++++++++++++++++++++++- internal/alert/rules.go | 4 ++ internal/alert/rules_test.go | 60 +++++++++++++++++++++++++ internal/server/http/metrics.go | 28 +++++++++++- internal/server/metrics/metrics.go | 9 ++++ internal/server/metrics/metrics_test.go | 4 ++ internal/store/jobs.go | 46 +++++++++++++++++++ internal/store/jobs_test.go | 42 +++++++++++++++++ 8 files changed, 249 insertions(+), 2 deletions(-) diff --git a/internal/alert/engine.go b/internal/alert/engine.go index ab6b27b..8a536c9 100644 --- a/internal/alert/engine.go +++ b/internal/alert/engine.go @@ -28,6 +28,11 @@ import ( // are evaluated — always-on hosts' stale_schedule stays a no-op. const staleBackupThreshold = 7 * 24 * time.Hour +const ( + defaultStuckJobThreshold = 6 * time.Hour + longStuckJobThreshold = 24 * time.Hour +) + // JobFinishedEvent carries everything the engine needs to evaluate // the failed-X rules. Pushed via Engine.NotifyJobFinished from the // MarkJobFinished site. @@ -53,6 +58,7 @@ type Engine struct { // we raise. Configurable for tests; default 15m. agentOfflineFloor time.Duration tickPeriod time.Duration + stuckThresholds map[string]time.Duration closeOnce sync.Once done chan struct{} @@ -69,7 +75,12 @@ func NewEngine(st *store.Store, hub *notification.Hub) *Engine { hostUp: make(chan string, 32), agentOfflineFloor: 15 * time.Minute, tickPeriod: 60 * time.Second, - done: make(chan struct{}), + stuckThresholds: map[string]time.Duration{ + "backup": longStuckJobThreshold, + "restore": longStuckJobThreshold, + "check": longStuckJobThreshold, + }, + done: make(chan struct{}), } } @@ -125,6 +136,10 @@ func (e *Engine) NotifyHostOnline(hostID string) { } func (e *Engine) handleJobFinished(ctx context.Context, ev JobFinishedEvent) { + // A late terminal message is authoritative and clears any stuck alert for + // this exact job regardless of its outcome or kind. + e.resolveAndNotify(ctx, ev.HostID, KindJobStuck, ev.JobID, ev.When) + // Determine which kind/severity pair this job maps to. Jobs not // listed here (init, unlock, restore, diff) produce no alerts in v1. var kind, severity string @@ -210,6 +225,7 @@ func (e *Engine) tick(ctx context.Context, now time.Time) { if _, err := e.store.CleanupExpiredOIDCState(ctx, now.Add(-5*time.Minute)); err != nil { slog.Warn("alert: cleanup expired oidc state", "err", err) } + e.evaluateStuckJobs(ctx, now) hosts, err := e.store.ListHosts(ctx) if err != nil { @@ -257,6 +273,46 @@ func (e *Engine) tick(ctx context.Context, now time.Time) { } } +func (e *Engine) evaluateStuckJobs(ctx context.Context, now time.Time) { + running, err := e.store.ListRunningJobActivity(ctx) + if err != nil { + slog.Warn("alert: tick list running jobs", "err", err) + return + } + active := make(map[string]store.RunningJobActivity, len(running)) + for _, job := range running { + active[job.JobID] = job + threshold := defaultStuckJobThreshold + if configured, ok := e.stuckThresholds[job.Kind]; ok { + threshold = configured + } + age := now.Sub(job.LastActivity) + if age < threshold { + continue + } + e.raiseAndNotify(ctx, job.HostID, KindJobStuck, job.JobID, "warning", + fmt.Sprintf("%s job %s is stuck: started %s, last activity %s (%s ago; threshold %s)", + job.Kind, job.JobID, job.StartedAt.Format(time.RFC3339), + job.LastActivity.Format(time.RFC3339), roundDur(age), threshold), now) + } + + // Self-heal alerts when another server path made a job terminal without + // emitting JobFinishedEvent, or after a restart missed the event. + alerts, err := e.store.ListAlerts(ctx, store.AlertFilter{Status: "open"}) + if err != nil { + return + } + acked, _ := e.store.ListAlerts(ctx, store.AlertFilter{Status: "acknowledged"}) + for _, item := range append(alerts, acked...) { + if item.Kind != KindJobStuck || item.HostID == nil { + continue + } + if _, ok := active[item.DedupKey]; !ok { + e.resolveAndNotify(ctx, *item.HostID, KindJobStuck, item.DedupKey, now) + } + } +} + // roundDur returns a human-readable duration string, rounding to the // nearest minute. Durations under a minute are reported as "less than // a minute". diff --git a/internal/alert/rules.go b/internal/alert/rules.go index f5e779c..d00c91e 100644 --- a/internal/alert/rules.go +++ b/internal/alert/rules.go @@ -36,6 +36,10 @@ const ( // KindAgentOffline is raised when a host's last_seen_at is older // than the 15-minute floor and resolved when the host reconnects. KindAgentOffline = "agent_offline" + + // KindJobStuck is raised per job when a running job has no persisted + // activity beyond its kind-specific threshold. The job ID is the dedup key. + KindJobStuck = "job_stuck" ) // raiseAndNotify is the standard raise pattern: store.RaiseOrTouch diff --git a/internal/alert/rules_test.go b/internal/alert/rules_test.go index c8f9d32..ec6a2d3 100644 --- a/internal/alert/rules_test.go +++ b/internal/alert/rules_test.go @@ -3,6 +3,7 @@ package alert import ( "context" "path/filepath" + "strings" "testing" "time" @@ -69,6 +70,65 @@ func TestEngineBackupFailedRaisesThenResolves(t *testing.T) { } } +func TestEngineStuckJobRaisesDeduplicatesAndResolves(t *testing.T) { + t.Parallel() + eng, st, hostID := setupEngine(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Second) + started := now.Add(-7 * time.Hour) + if err := st.CreateJob(ctx, store.Job{ID: "stuck-job", HostID: hostID, Kind: "prune", ActorKind: "user", CreatedAt: started}); err != nil { + t.Fatalf("create job: %v", err) + } + if err := st.MarkJobStarted(ctx, "stuck-job", started); err != nil { + t.Fatalf("start job: %v", err) + } + + eng.evaluateStuckJobs(ctx, now) + eng.evaluateStuckJobs(ctx, now.Add(time.Minute)) + open, err := st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID}) + if err != nil { + t.Fatalf("list alerts: %v", err) + } + if len(open) != 1 || open[0].Kind != KindJobStuck || open[0].DedupKey != "stuck-job" { + t.Fatalf("expected one deduplicated stuck alert, got %+v", open) + } + if !strings.Contains(open[0].Message, started.Format(time.RFC3339)) { + t.Errorf("alert does not identify start/activity time: %q", open[0].Message) + } + + if err := st.MarkJobFinished(ctx, "stuck-job", "succeeded", 0, nil, "", now); err != nil { + t.Fatalf("finish job: %v", err) + } + eng.evaluateStuckJobs(ctx, now.Add(2*time.Minute)) + open, _ = st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID}) + if len(open) != 0 { + t.Fatalf("expected terminal job alert resolved, got %+v", open) + } +} + +func TestEngineStuckJobUsesLatestLogActivity(t *testing.T) { + t.Parallel() + eng, st, hostID := setupEngine(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Second) + started := now.Add(-7 * time.Hour) + if err := st.CreateJob(ctx, store.Job{ID: "active-job", HostID: hostID, Kind: "forget", ActorKind: "user", CreatedAt: started}); err != nil { + t.Fatalf("create job: %v", err) + } + if err := st.MarkJobStarted(ctx, "active-job", started); err != nil { + t.Fatalf("start job: %v", err) + } + if err := st.AppendJobLog(ctx, "active-job", 1, now.Add(-time.Hour), "stdout", "active"); err != nil { + t.Fatalf("append log: %v", err) + } + + eng.evaluateStuckJobs(ctx, now) + open, _ := st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID}) + if len(open) != 0 { + t.Fatalf("recent activity should suppress stuck alert, got %+v", open) + } +} + func TestEngineCheckFailedSeverityCritical(t *testing.T) { t.Parallel() eng, st, hostID := setupEngine(t) diff --git a/internal/server/http/metrics.go b/internal/server/http/metrics.go index 2c65ca8..f547ff3 100644 --- a/internal/server/http/metrics.go +++ b/internal/server/http/metrics.go @@ -8,7 +8,9 @@ import ( "net/netip" "runtime" "strings" + "time" + "gitea.dcglab.co.uk/steve/restic-manager/internal/alert" "gitea.dcglab.co.uk/steve/restic-manager/internal/server/config" "gitea.dcglab.co.uk/steve/restic-manager/internal/server/metrics" "gitea.dcglab.co.uk/steve/restic-manager/internal/store" @@ -173,13 +175,37 @@ func (s *Server) gatherMetricsSnapshot(ctx context.Context) (metrics.Snapshot, e return metrics.Snapshot{}, err } bySeverity := map[string]int{"info": 0, "warning": 0, "critical": 0} + stuckJobs := 0 + var oldestStuckAge time.Duration + now := time.Now().UTC() + running, err := s.deps.Store.ListRunningJobActivity(ctx) + if err != nil { + return metrics.Snapshot{}, err + } + activityByJob := make(map[string]time.Time, len(running)) + for _, job := range running { + activityByJob[job.JobID] = job.LastActivity + } for _, a := range open { bySeverity[a.Severity]++ + if a.Kind == alert.KindJobStuck { + stuckJobs++ + lastActivity, ok := activityByJob[a.DedupKey] + if !ok { + continue + } + if age := now.Sub(lastActivity); age > oldestStuckAge { + oldestStuckAge = age + } + } } reg := s.deps.Metrics if reg == nil { reg = metrics.NewRegistry() // empty histogram block } - return reg.SnapshotWith(hostRows, bySeverity, version.Version, version.Commit, runtime.Version()), nil + snap := reg.SnapshotWith(hostRows, bySeverity, version.Version, version.Commit, runtime.Version()) + snap.StuckJobs = stuckJobs + snap.OldestStuckAge = oldestStuckAge + return snap, nil } diff --git a/internal/server/metrics/metrics.go b/internal/server/metrics/metrics.go index 588d796..e166ed6 100644 --- a/internal/server/metrics/metrics.go +++ b/internal/server/metrics/metrics.go @@ -155,6 +155,8 @@ type Snapshot struct { BuildCommit string GoVersion string JobDurationRows []HistogramRow + StuckJobs int + OldestStuckAge time.Duration } // SnapshotWith builds a Snapshot from raw inputs and the registry's @@ -205,6 +207,13 @@ func Render(w io.Writer, s Snapshot) error { fmt.Fprintf(&b, "rm_build_info{version=%q,commit=%q,go_version=%q} 1\n", s.BuildVersion, s.BuildCommit, s.GoVersion) + b.WriteString("# HELP rm_stuck_jobs Number of open job_stuck alerts.\n") + b.WriteString("# TYPE rm_stuck_jobs gauge\n") + fmt.Fprintf(&b, "rm_stuck_jobs %d\n", s.StuckJobs) + b.WriteString("# HELP rm_oldest_stuck_job_age_seconds Time since last activity for the oldest open stuck job.\n") + b.WriteString("# TYPE rm_oldest_stuck_job_age_seconds gauge\n") + fmt.Fprintf(&b, "rm_oldest_stuck_job_age_seconds %.0f\n", s.OldestStuckAge.Seconds()) + // --- Per-host gauges ------------------------------------------------- // Stable order: by host id. hosts := append([]HostRow(nil), s.Hosts...) diff --git a/internal/server/metrics/metrics_test.go b/internal/server/metrics/metrics_test.go index 70c5ed7..addd031 100644 --- a/internal/server/metrics/metrics_test.go +++ b/internal/server/metrics/metrics_test.go @@ -114,6 +114,8 @@ func TestRenderGolden(t *testing.T) { snap := r.SnapshotWith(hosts, map[string]int{"info": 0, "warning": 1, "critical": 0}, "v1.2.3", "deadbeef", "go1.25.0") + snap.StuckJobs = 2 + snap.OldestStuckAge = 90 * time.Minute var buf bytes.Buffer if err := Render(&buf, snap); err != nil { @@ -129,6 +131,8 @@ func TestRenderGolden(t *testing.T) { `rm_active_alerts{severity="info"} 0`, `rm_active_alerts{severity="critical"} 0`, `rm_build_info{version="v1.2.3",commit="deadbeef",go_version="go1.25.0"} 1`, + `rm_stuck_jobs 2`, + `rm_oldest_stuck_job_age_seconds 5400`, `rm_host_agent_online{host_id="01H0001",host="alpha"} 1`, `rm_host_agent_online{host_id="01H0002",host="bravo"} 0`, `rm_host_last_backup_timestamp_seconds{host_id="01H0001",host="alpha"} 1700000000`, diff --git a/internal/store/jobs.go b/internal/store/jobs.go index 15b71e2..cd1ff8f 100644 --- a/internal/store/jobs.go +++ b/internal/store/jobs.go @@ -27,6 +27,52 @@ type Job struct { CreatedAt time.Time } +// RunningJobActivity is the persisted activity summary used to detect jobs +// whose terminal message was lost. LastActivity is the newer of started_at and +// the most recent persisted log line; ephemeral progress events deliberately +// do not extend it. +type RunningJobActivity struct { + JobID string + HostID string + Kind string + StartedAt time.Time + LastActivity time.Time +} + +// ListRunningJobActivity returns every running job with its latest persisted +// activity. Jobs without started_at are excluded because they have not actually +// entered the running state coherently and cannot be aged safely here. +func (s *Store) ListRunningJobActivity(ctx context.Context) ([]RunningJobActivity, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT j.id, j.host_id, j.kind, j.started_at, + COALESCE(MAX(l.ts), j.started_at) AS last_activity + FROM jobs j + LEFT JOIN job_logs l ON l.job_id = j.id + WHERE j.status = 'running' AND j.started_at IS NOT NULL + GROUP BY j.id, j.host_id, j.kind, j.started_at + ORDER BY j.started_at`) + if err != nil { + return nil, fmt.Errorf("store: list running job activity: %w", err) + } + defer func() { _ = rows.Close() }() + + var out []RunningJobActivity + for rows.Next() { + var item RunningJobActivity + var started, activity string + if err := rows.Scan(&item.JobID, &item.HostID, &item.Kind, &started, &activity); err != nil { + return nil, fmt.Errorf("store: scan running job activity: %w", err) + } + item.StartedAt, _ = time.Parse(time.RFC3339Nano, started) + item.LastActivity, _ = time.Parse(time.RFC3339Nano, activity) + out = append(out, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("store: iterate running job activity: %w", err) + } + return out, nil +} + // CreateJob inserts a queued job. The agent will mark it running // when it actually starts work. ScheduledID is set when the job // originates from a cron fire (actor_kind="schedule"); nil for diff --git a/internal/store/jobs_test.go b/internal/store/jobs_test.go index 9339a66..07a3eea 100644 --- a/internal/store/jobs_test.go +++ b/internal/store/jobs_test.go @@ -7,6 +7,48 @@ import ( "time" ) +func TestListRunningJobActivity(t *testing.T) { + t.Parallel() + s := openTestStore(t) + ctx := context.Background() + hostID := makeSchedHost(t, s) + started := time.Now().UTC().Add(-8 * time.Hour).Truncate(time.Second) + + for _, id := range []string{"running-idle", "running-active", "finished"} { + if err := s.CreateJob(ctx, Job{ID: id, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil { + t.Fatalf("create %s: %v", id, err) + } + if err := s.MarkJobStarted(ctx, id, started); err != nil { + t.Fatalf("start %s: %v", id, err) + } + } + activity := started.Add(7 * time.Hour) + if err := s.AppendJobLog(ctx, "running-active", 1, activity, "stdout", "still working"); err != nil { + t.Fatalf("append log: %v", err) + } + if err := s.MarkJobFinished(ctx, "finished", "succeeded", 0, nil, "", activity); err != nil { + t.Fatalf("finish: %v", err) + } + + got, err := s.ListRunningJobActivity(ctx) + if err != nil { + t.Fatalf("list activity: %v", err) + } + if len(got) != 2 { + t.Fatalf("got %d rows, want 2: %+v", len(got), got) + } + byID := make(map[string]RunningJobActivity, len(got)) + for _, row := range got { + byID[row.JobID] = row + } + if !byID["running-idle"].LastActivity.Equal(started) { + t.Errorf("idle last activity = %s, want %s", byID["running-idle"].LastActivity, started) + } + if !byID["running-active"].LastActivity.Equal(activity) { + t.Errorf("active last activity = %s, want %s", byID["running-active"].LastActivity, activity) + } +} + func TestLatestJobByKind(t *testing.T) { t.Parallel() s := openTestStore(t) -- 2.52.0 From 750a06384b222977f1994b42074f3850c4302e04 Mon Sep 17 00:00:00 2001 From: Steve Cliff Date: Sat, 22 Aug 2026 10:53:51 +0100 Subject: [PATCH 2/2] Close websocket handshake response in test --- internal/server/ws/hub_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/server/ws/hub_test.go b/internal/server/ws/hub_test.go index f4b9248..bb90ff6 100644 --- a/internal/server/ws/hub_test.go +++ b/internal/server/ws/hub_test.go @@ -129,9 +129,12 @@ func TestWSAcceptsSnapshotReportLargerThanDefaultReadLimit(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - c, _, err := websocket.Dial(ctx, url, &websocket.DialOptions{ + c, resp, err := websocket.Dial(ctx, url, &websocket.DialOptions{ HTTPHeader: stdhttp.Header{"Authorization": []string{"Bearer " + token}}, }) + if resp != nil && resp.Body != nil { + defer resp.Body.Close() //nolint:errcheck + } if err != nil { t.Fatalf("dial: %v", err) } -- 2.52.0