From a95e391430a085fdddfe9267d75c3ecbbc37a6d8 Mon Sep 17 00:00:00 2001 From: Bartek Tofel Date: Mon, 31 Aug 2026 15:49:08 +0200 Subject: [PATCH 1/2] chore: implement phase 6 Invariant defended: H2. The one question: can watch return success over a window that nothing is recording? Watch() records the first observation of each non-skipped rule, then detaches a child that polls at the cadence in the header. The parent returns only after the child reports ready on an inherited pipe, and writes the pidfile after that. A clean stop writes the sentinel; a hard error does not. --- grafana-alertcheck/internal/gate/log.go | 28 + grafana-alertcheck/internal/gate/schedule.go | 44 +- .../internal/gate/schedule_test.go | 8 +- .../internal/gate/source_fake_test.go | 40 +- grafana-alertcheck/internal/gate/watch.go | 764 ++++++++++++++++++ .../internal/gate/watch_daemon_test.go | 308 +++++++ .../internal/gate/watch_test.go | 702 ++++++++++++++++ .../internal/gate/watch_unix.go | 112 +++ 8 files changed, 1988 insertions(+), 18 deletions(-) create mode 100644 grafana-alertcheck/internal/gate/watch.go create mode 100644 grafana-alertcheck/internal/gate/watch_daemon_test.go create mode 100644 grafana-alertcheck/internal/gate/watch_test.go create mode 100644 grafana-alertcheck/internal/gate/watch_unix.go diff --git a/grafana-alertcheck/internal/gate/log.go b/grafana-alertcheck/internal/gate/log.go index 3f940f7d0..0d73ab092 100644 --- a/grafana-alertcheck/internal/gate/log.go +++ b/grafana-alertcheck/internal/gate/log.go @@ -237,6 +237,34 @@ func (r *Reducer) Reduce(uid string, obs Observation) Poll { return p } +// seedFrom restores the marker state above from polls that are already in the +// log, so the first poll a NEW Reducer produces compares against the last poll +// the previous one wrote instead of against an empty set. +// +// It exists for the one place a recording changes hands: watch's parent takes +// the first observation of every rule and its detached child continues from +// there (P6). Without the seed, an instance that is abnormal in the parent's +// observation and gone by the child's first poll produces no marker at all — +// it leaves the record as though it had never been bad, which is H2's +// fail-open reached through the handoff rather than through a reason string. +// +// Not-found polls are skipped, mirroring Reduce: an absent rule leaves the +// previous abnormal set untouched rather than emptying it. +func (r *Reducer) seedFrom(polls []Poll) { + r.mu.Lock() + defer r.mu.Unlock() + for _, p := range polls { + if !p.Found { + continue + } + keys := make(map[string]struct{}, len(p.Abnormal)) + for _, inst := range p.Abnormal { + keys[instanceKey(inst.Labels)] = struct{}{} + } + r.prevAbnormal[p.RuleUID] = keys + } +} + // reasonNames reports whether reason names want. Newer Grafana versions // comma-join several reasons into one string, so this tests membership rather // than equality (P7 check 9 needs the same test for KeepLast). diff --git a/grafana-alertcheck/internal/gate/schedule.go b/grafana-alertcheck/internal/gate/schedule.go index 89bcd5b3f..f39c02d24 100644 --- a/grafana-alertcheck/internal/gate/schedule.go +++ b/grafana-alertcheck/internal/gate/schedule.go @@ -193,21 +193,27 @@ type Scheduler struct { every map[string]time.Duration } -// NewScheduler builds a Scheduler over rules (keyed by UID), staggering each -// rule's initial next-due time across [0, pollEvery) so the fleet does not -// start phase-aligned (§5's burst-bound proof depends on this: an -// already-staggered fleet only re-aligns by chance, briefly, not by +// NewScheduler builds a Scheduler over per-rule cadences (keyed by UID), +// staggering each rule's initial next-due time across [0, pollEvery) so the +// fleet does not start phase-aligned (§5's burst-bound proof depends on this: +// an already-staggered fleet only re-aligns by chance, briefly, not by // construction). -func NewScheduler(rules map[string]ruleTimings, now time.Time) *Scheduler { +// +// It takes cadences rather than whole ruleTimings on purpose: a scheduler +// decides when to poll and nothing else, so it must not be handed maxGap, +// healthGrace or evalStaleAfter. Those are coverage thresholds, they are +// applied by the pure layer at classification time, and the recorder that +// drives this scheduler never applies them at all. +func NewScheduler(every map[string]time.Duration, now time.Time) *Scheduler { s := &Scheduler{ - next: make(map[string]time.Time, len(rules)), - every: make(map[string]time.Duration, len(rules)), + next: make(map[string]time.Time, len(every)), + every: make(map[string]time.Duration, len(every)), } - for uid, rt := range rules { - s.every[uid] = rt.pollEvery + for uid, pollEvery := range every { + s.every[uid] = pollEvery var offset time.Duration - if rt.pollEvery > 0 { - offset = rand.N(rt.pollEvery) + if pollEvery > 0 { + offset = rand.N(pollEvery) } s.next[uid] = now.Add(offset) } @@ -247,6 +253,22 @@ func (s *Scheduler) Mark(uid string, now time.Time) { s.next[uid] = now.Add(s.every[uid]) } +// earliestDue returns the earliest scheduled next-due time, and false when the +// scheduler holds no rules at all. The recorder's loop (P6) waits exactly that +// long instead of waking on a fixed tick: a fixed tick either polls a slack +// rule early — spending request budget the §5 formulas already accounted for — +// or wakes too late for the tightest rule and opens a gap inside its own +// maxGap. +func (s *Scheduler) earliestDue() (time.Time, bool) { + var earliest time.Time + for _, t := range s.next { + if earliest.IsZero() || t.Before(earliest) { + earliest = t + } + } + return earliest, !earliest.IsZero() +} + // CheckBudget applies §5's error-at-start check to a fully resolved schedule. // t and measured are both keyed by rule UID; measured must carry every UID in // t; a rule this run never measured can't have its budget proved, and a diff --git a/grafana-alertcheck/internal/gate/schedule_test.go b/grafana-alertcheck/internal/gate/schedule_test.go index 972300f77..67659bc2f 100644 --- a/grafana-alertcheck/internal/gate/schedule_test.go +++ b/grafana-alertcheck/internal/gate/schedule_test.go @@ -163,9 +163,9 @@ func TestScheduler_MarkAdvancesNextDue(t *testing.T) { // forced onto the tight rule's cadence. func TestScheduler_PerRuleCadenceOverTime(t *testing.T) { start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) - rules := map[string]ruleTimings{ - "tight": {pollEvery: 10 * time.Second}, - "slack": {pollEvery: 300 * time.Second}, + rules := map[string]time.Duration{ + "tight": 10 * time.Second, + "slack": 300 * time.Second, } s := NewScheduler(rules, start) @@ -196,7 +196,7 @@ func TestScheduler_PerRuleCadenceOverTime(t *testing.T) { func TestNewScheduler_StaggersWithinPollEvery(t *testing.T) { now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) - rules := map[string]ruleTimings{"r1": {pollEvery: 100 * time.Second}} + rules := map[string]time.Duration{"r1": 100 * time.Second} s := NewScheduler(rules, now) offset := s.next["r1"].Sub(now) if offset < 0 || offset >= 100*time.Second { diff --git a/grafana-alertcheck/internal/gate/source_fake_test.go b/grafana-alertcheck/internal/gate/source_fake_test.go index b3ef23cca..5f8cbc9e4 100644 --- a/grafana-alertcheck/internal/gate/source_fake_test.go +++ b/grafana-alertcheck/internal/gate/source_fake_test.go @@ -14,9 +14,9 @@ import ( // That is sufficient here: every retry/backoff test in this phase only needs // to avoid a real sleep. It is NOT sufficient for a test that must prove a // wait did not fire early — e.g. a P4 scheduler test asserting Due() doesn't -// return a rule before its next-due time. That needs a clock with a real -// waiter list keyed off Advance, which does not exist yet; build it when a -// phase actually needs it rather than guessing its shape now. +// return a rule before its next-due time. Use virtualClock below for that: it +// is the clock P6's recorder-loop tests needed, and it makes a wait and the +// passage of time the same event. type fakeClock struct { mu sync.Mutex now time.Time @@ -45,6 +45,40 @@ func (c *fakeClock) After(d time.Duration) <-chan time.Time { return ch } +// virtualClock is a Clock in which time moves only when something waits for +// it: After(d) jumps Now() forward by d and fires at once. That makes a +// recorder-loop test both instant and exact — a loop that waits for its next +// scheduled poll gets that poll's time, never an early or a late wake — and it +// terminates, which a clock whose After fires without advancing Now does not +// (the loop would spin forever on a rule that never comes due). +// +// It is goroutine-safe, but a test that advances time from two goroutines gets +// what it deserves: use it from the loop under test only. +type virtualClock struct { + mu sync.Mutex + now time.Time +} + +func newVirtualClock(now time.Time) *virtualClock { return &virtualClock{now: now} } + +func (c *virtualClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *virtualClock) After(d time.Duration) <-chan time.Time { + c.mu.Lock() + if d > 0 { + c.now = c.now.Add(d) + } + fireAt := c.now + c.mu.Unlock() + ch := make(chan time.Time, 1) + ch <- fireAt + return ch +} + // steppingClock advances by a fixed step on every Now() call, so a test can // assert exact latency/skew-bound arithmetic (doRequest's three clock reads // per attempt) without depending on real wall-clock timing. diff --git a/grafana-alertcheck/internal/gate/watch.go b/grafana-alertcheck/internal/gate/watch.go new file mode 100644 index 000000000..61e1cce63 --- /dev/null +++ b/grafana-alertcheck/internal/gate/watch.go @@ -0,0 +1,764 @@ +package gate + +import ( + "context" + "fmt" + "io" + "os" + "os/signal" + "strconv" + "strings" + "sync" + "syscall" + "time" +) + +// DaemonChildFlag is the hidden flag the parent passes when it re-execs itself +// as the detached recorder (§4.4). It is deliberately absent from the CLI's +// usage text: an operator never types it, and a child started by hand against +// a log no parent prepared fails immediately on the header read. +const DaemonChildFlag = "--daemon-child" + +// ReadyFDFlag names the inherited descriptor the child reports readiness on. +// The parent passes the write end of a pipe as descriptor 3 and waits for one +// byte, so "the recorder is running" is a POSITIVE signal from the child +// itself — it has read the header, taken the log's flock and entered its poll +// loop — and not an assumption drawn from surviving a timer. A timer cannot +// tell a healthy child from one that is about to die on a slow runner, and +// getting that wrong means watch returns success over a recording that never +// happened (§4.3). +const ReadyFDFlag = "--ready-fd" + +// childReadyTimeout bounds that wait. Everything before the signal is local — +// fork, exec, one read of a log holding a header and a handful of polls — so +// the real figure is milliseconds; this is loose enough for a badly overloaded +// runner and still fails closed rather than hanging the pipeline. +const childReadyTimeout = 30 * time.Second + +// daemonLogTailBytes bounds how much of a dead child's output the parent +// quotes back. A child dies in its first few lines or not at all. +const daemonLogTailBytes = 4096 + +// WatchConfig is the record step's whole input. +// +// It has no To field and must never gain one: watch writes the stopped +// sentinel with its OWN stop time and makes no comparison against `to`, which +// only check knows (§4.5). Passing `to` here would give two components an +// opinion about the same comparison, and the recorder's opinion is the one +// that cannot be trusted — it exits before the grace it would have to wait for. +// +// It has no States field either, and watch has no --states flag: recording is +// deliberately unfiltered. The reduction keeps every non-normal instance and +// the transition markers key off the same predicate, so neither consults +// States. The payoff is real — because the log is raw evidence, one recording +// can be re-classified under different --states without re-recording — and the +// Header carries no States field for the same reason. +type WatchConfig struct { + // URL and Token are the connection details. The CLI reads both from the + // environment and never from a flag (§20.2); Token is never logged and + // never enters an error string. + URL, Token string + + // Alerts are the operator-supplied names, one per line, in any of §17's + // forms. Empty lines are discarded by Resolve. + Alerts []string + Folder string + + // Out is the JSONL log path. PidFile and DaemonLog default to + // .pid and .daemon.log — the same convention check uses to find + // the recorder it must stop (P9), so nothing has to be wired by hand. + Out string + PidFile string + DaemonLog string + + // Until is an optional hard stop for the child. Zero means "record until + // signalled", which is the normal case: check sends SIGTERM when its + // collection loop ends. + Until time.Time + + // PollEvery is the --poll-interval override, used verbatim for every rule + // and never clamped (§5.1). Zero means each rule polls at half its own + // evaluation interval. Whatever this resolves to is written into the header + // as the cadence actually used, and that header value — never a + // re-derivation from the definitions — is what check derives maxGap from + // (P5, "two authorities"). + PollEvery time.Duration + + Concurrency int + Clock Clock + + // Notes is where the parent prints what an operator has to see before the + // deploy step runs: resolve notes, the cadence per rule, the rules it will + // not wait for. nil discards them. The library prints nothing else — the + // CLI owns presentation (§20.2). + Notes io.Writer +} + +func (cfg WatchConfig) withDefaults() WatchConfig { + if cfg.Clock == nil { + cfg.Clock = SystemClock{} + } + if cfg.Notes == nil { + cfg.Notes = io.Discard + } + if cfg.Concurrency < 1 { + cfg.Concurrency = 1 + } + if cfg.PidFile == "" && cfg.Out != "" { + cfg.PidFile = cfg.Out + ".pid" + } + if cfg.DaemonLog == "" && cfg.Out != "" { + cfg.DaemonLog = cfg.Out + ".daemon.log" + } + return cfg +} + +func (cfg WatchConfig) validate() error { + if cfg.URL == "" { + return fmt.Errorf("watch: no grafana url (it is the log's identity, which check validates)") + } + if cfg.Out == "" { + return fmt.Errorf("watch: no log path") + } + named := 0 + for _, a := range cfg.Alerts { + if strings.TrimSpace(a) != "" { + named++ + } + } + if named == 0 { + return fmt.Errorf("watch: no alert names given; there is nothing to record") + } + // An --until already in the past would make the child stop before it ever + // polled, and the parent would then report a child that never reported + // ready — a true statement about a config mistake, but a confusing one. + if !cfg.Until.IsZero() && !cfg.Until.After(cfg.Clock.Now()) { + return fmt.Errorf("watch: --until %s is not in the future", cfg.Until.Format(time.RFC3339)) + } + return nil +} + +// Watch is the record step's parent process (§4.3). It returns only once the +// window is genuinely being recorded: +// +// version gate -> resolve definitions and names -> derive timings -> +// open the log and write the header -> ONE observation of every non-skipped +// rule -> verify §3.2 -> check the schedule budget -> detach the child -> +// wait for the child to report that it is recording -> write the pidfile -> +// return. +// +// The first-observation wait is not a convenience. Returning before it would +// leave the deploy inside [from, first_poll] with no evidence — the exact +// blind interval the two-phase model exists to remove — and it is also what +// surfaces auth, name-resolution and parse failures BEFORE deploy.sh runs +// rather than ten minutes later. +func Watch(ctx context.Context, cfg WatchConfig) error { + cfg = cfg.withDefaults() + if err := cfg.validate(); err != nil { + return err + } + + src := NewHTTPSource(cfg.URL, cfg.Token, cfg.Clock) + prep, err := prepareWatch(ctx, cfg, src) + if err != nil { + return err + } + + // Hand the log over with Close, never Stop: a sentinel here would tell + // check the recording ended before the child had even started (§4.5). + // Closing also releases the flock the child is about to take. + if err := prep.writer.Close(); err != nil { + return err + } + + child, err := spawnChild(cfg) + if err != nil { + return fmt.Errorf("detach recorder: %w", err) + } + if err := waitForChildReady(cfg, child); err != nil { + return err + } + + // The PARENT writes the pidfile, not the child: check must find the pid the + // instant Watch returns, and a child writing its own would race the very + // next step of the pipeline. A deviation from P6's argv list, and the + // reason the child is never given --pidfile at all. + // + // It is written only once the child has reported ready, so no path through + // this function leaves a pidfile naming a process that is not recording. + // Pids are reused: a stale pidfile is a live process somewhere else, and a + // pipeline that ignored this function's error would SIGTERM it. + if err := writePidFile(cfg.PidFile, child.cmd.Process.Pid); err != nil { + _ = child.cmd.Process.Kill() + _ = os.Remove(cfg.PidFile) + return err + } + fmt.Fprintf(cfg.Notes, "recording to %s (pid %d, pidfile %s, output %s)\n", + cfg.Out, child.cmd.Process.Pid, cfg.PidFile, cfg.DaemonLog) + return nil +} + +// waitForChildReady waits for the child's own readiness byte, and treats every +// other outcome as a failure to record: the pipe closing without a byte (the +// child died on its way to the loop), the process exiting, or the timeout. +// Each of the three quotes the daemon log, which is the only place a detached +// process can explain itself. +func waitForChildReady(cfg WatchConfig, child detachedChild) error { + defer child.ready.Close() + + // This goroutine outlives the function when the child keeps running. It + // stays behind only to reap a child that dies while this short-lived parent + // is still alive, and costs nothing. + exited := make(chan error, 1) + go func() { exited <- child.cmd.Wait() }() + + signalled := make(chan error, 1) + go func() { + _, err := child.ready.Read(make([]byte, 1)) + signalled <- err + }() + + fail := func(format string, args ...any) error { + _ = child.cmd.Process.Kill() + return fmt.Errorf("%s; its output was:\n%s", + fmt.Sprintf(format, args...), daemonLogTail(cfg.DaemonLog, child.logOffset)) + } + + select { + case err := <-signalled: + if err == nil { + return nil + } + // The child closed the pipe — by exiting — without ever reporting that + // it had the log and was polling. + return fail("the detached recorder never reported ready (%v)", err) + case waitErr := <-exited: + status := "exit status 0" + if waitErr != nil { + status = waitErr.Error() + } + return fail("the detached recorder exited before it started recording (%s)", status) + case <-cfg.Clock.After(childReadyTimeout): + return fail("the detached recorder did not report ready within %s", childReadyTimeout) + } +} + +// daemonLogTail quotes the end of the daemon log, starting at from — the size +// the file had when THIS run opened it. The offset is what keeps the quote +// honest when several runs share one --daemon-log path: without it the tail can +// name a previous run's failure as the current one's cause. +func daemonLogTail(path string, from int64) string { + b, err := os.ReadFile(path) + if err != nil { + return fmt.Sprintf("(daemon log %s is unreadable: %v)", path, err) + } + if from > 0 && from <= int64(len(b)) { + b = b[from:] + } + if len(b) > daemonLogTailBytes { + b = b[len(b)-daemonLogTailBytes:] + } + if len(b) == 0 { + return fmt.Sprintf("(this run wrote nothing to the daemon log %s)", path) + } + return strings.TrimRight(string(b), "\n") +} + +// preparedWatch is what the parent has established by the time it is willing +// to detach: an open log with a header and one poll per non-skipped rule, the +// timings that produced them, and the latencies it measured doing so. +type preparedWatch struct { + writer *Writer + header Header + timings map[string]ruleTimings + measured map[string]time.Duration +} + +// prepareWatch is everything the parent does before it detaches. It takes a +// Source rather than building one so the paused-rule, first-observation, +// §3.2 and budget behaviours are all testable with a scripted fake — only the +// process spawning needs a real binary. +func prepareWatch(ctx context.Context, cfg WatchConfig, src Source) (*preparedWatch, error) { + version, err := src.Version(ctx) + if err != nil { + return nil, fmt.Errorf("read grafana version: %w", err) + } + if err := CheckGrafanaVersion(version); err != nil { + return nil, err + } + + defs, err := src.Definitions(ctx) + if err != nil { + return nil, fmt.Errorf("read rule definitions: %w", err) + } + resolved, notes, err := Resolve(defs, cfg.Alerts, cfg.Folder) + if err != nil { + return nil, err + } + for _, n := range notes { + fmt.Fprintf(cfg.Notes, "note: %s\n", n) + } + + rt, _, timingNotes := DeriveTimings(resolved, cfg.PollEvery) + for _, n := range timingNotes { + fmt.Fprintf(cfg.Notes, "note: %s\n", n) + } + for _, d := range resolved { + // A cadence of zero would make the child spin: every rule is due the + // instant it was marked. It also cannot be written into the header, + // where check requires a positive value to derive maxGap from (P5). + if rt[d.UID].pollEvery <= 0 { + return nil, fmt.Errorf("rule %q (%s) reports intervalSeconds=%d: there is no poll cadence to record at", + d.Title, d.UID, d.IntervalSeconds) + } + } + + writer, err := NewWriter(cfg.Out, cfg.Clock) + if err != nil { + return nil, err + } + prep, err := openRecording(ctx, cfg, src, writer, version, resolved, rt) + if err != nil { + // Close, never Stop. The log keeps whatever was written and gets no + // sentinel, so nothing can later mistake it for a finished recording. + _ = writer.Close() + return nil, err + } + return prep, nil +} + +// openRecording writes the header, takes the first observation of every rule +// the recorder will actually watch, appends those observations as the log's +// first heartbeats, and only then decides whether the schedule is feasible. +func openRecording(ctx context.Context, cfg WatchConfig, src Source, writer *Writer, + version string, resolved []Definition, rt map[string]ruleTimings) (*preparedWatch, error) { + + header := Header{ + SchemaVersion: LogSchemaVersion, + URL: cfg.URL, + GrafanaVersion: version, + StartedAt: cfg.Clock.Now(), + Rules: loggedRules(resolved, rt), + } + if err := writer.WriteHeader(header); err != nil { + return nil, err + } + + // A rule whose DEFINITION says is_paused is skipped (§12): it is not + // waited for, not scheduled and never polled. Waiting for one either hangs + // forever or errors before the deploy (§4.3), and recording polls for it + // would report an in-window pause (coverage check 7) for a rule that was + // already paused when the window opened — turning §12's exit 1 into an + // exit 2. The header still names it, with is_paused true, so check reports + // it as skipped from the definitions. + var active []Definition + activeTimings := make(map[string]ruleTimings, len(resolved)) + titles := make(map[string]string, len(resolved)) + for _, d := range resolved { + if d.IsPaused { + fmt.Fprintf(cfg.Notes, "note: rule %q (%s) is paused: recorded as skipped, not waited for (§4.3)\n", d.Title, d.UID) + continue + } + active = append(active, d) + activeTimings[d.UID] = rt[d.UID] + titles[d.UID] = d.Title + fmt.Fprintf(cfg.Notes, "recording %q (%s) every %s (maxGap %s)\n", d.Title, d.UID, rt[d.UID].pollEvery, rt[d.UID].maxGap) + } + + uids := make([]string, 0, len(active)) + for _, d := range active { + uids = append(uids, d.UID) + } + observed, err := observeAll(ctx, src, titles, uids, cfg.Concurrency) + if err != nil { + return nil, err + } + + // Verify §3.2 before anything downstream relies on it: if the state + // endpoint ever stops returning normal instances, the reduction's "keep + // the non-normal ones" silently becomes "keep everything it happened to + // send" and the transition markers lose their ground truth. + for _, d := range active { + if err := VerifyNormalInstancesVisible(observed[d.UID].Rules); err != nil { + return nil, err + } + } + + // One poll record per rule, in resolve order so the log is byte-stable for + // a given set of observations. These ARE the log's first heartbeats: they + // predate the deploy step, which is the whole point of §4.3. + reducer := NewReducer() + measured := make(map[string]time.Duration, len(active)) + for _, d := range active { + obs := observed[d.UID] + measured[d.UID] = obs.Latency + poll := reducer.Reduce(d.UID, obs) + if !poll.Found { + // Authoritative, not transient (P2 already retried transport + // failures): the rule resolved in the ruler API but the state + // endpoint does not serve it. Recorded as Found=false, which P7 + // turns into unobservable — a note rather than an error here, + // because the state endpoint can lag a freshly created rule and + // check re-resolves and fails closed either way. + fmt.Fprintf(cfg.Notes, "warning: rule %q (%s) is absent from the state endpoint; recorded as not found\n", d.Title, d.UID) + } + if err := writer.WritePoll(poll); err != nil { + return nil, err + } + } + + // Budget last, on the latencies just measured — never on a fixed estimate + // (§5.2). Only the active rules count: a skipped rule is never polled and + // consumes none of the capacity. + if err := CheckBudget(activeTimings, measured, cfg.Concurrency); err != nil { + return nil, err + } + + return &preparedWatch{writer: writer, header: header, timings: rt, measured: measured}, nil +} + +// loggedRules snapshots the resolved definitions into the header's rule list. +// Every field but PollEverySeconds is forensic — a resolve-time snapshot that +// makes an uploaded log self-describing (§21.3) — while PollEverySeconds is +// load-bearing: it is the cadence this recording actually used, and check +// derives maxGap from it rather than from the definitions (P5). +func loggedRules(defs []Definition, rt map[string]ruleTimings) []LoggedRule { + out := make([]LoggedRule, 0, len(defs)) + for _, d := range defs { + out = append(out, LoggedRule{ + UID: d.UID, + Title: d.Title, + Folder: d.Folder, + Group: d.Group, + ForSeconds: d.For.Seconds(), + IntervalSeconds: d.IntervalSeconds, + IsPaused: d.IsPaused, + NoDataState: d.NoDataState, + ExecErrState: d.ExecErrState, + PollEverySeconds: rt[d.UID].pollEvery.Seconds(), + }) + } + return out +} + +// observeAll polls every rule in uids concurrently, bounded by concurrency, +// and returns one Observation per rule that answered. Every rule is polled by +// TITLE (the ?rule_name= filter, §2.8) and selected out of the response by +// UID (§14.5) — a filtered response can carry several rules sharing one title. +// +// It returns the successful observations alongside the first error in UID +// order, so a caller that wants to keep the good heartbeats can, and the error +// message is the same on every run. +func observeAll(ctx context.Context, src Source, titles map[string]string, uids []string, concurrency int) (map[string]Observation, error) { + if concurrency < 1 { + concurrency = 1 + } + var ( + mu sync.Mutex + out = make(map[string]Observation, len(uids)) + firstErr error + firstErrUID string + ) + sem := make(chan struct{}, concurrency) + var wg sync.WaitGroup + for _, uid := range uids { + wg.Go(func() { + sem <- struct{}{} + defer func() { <-sem }() + + obs, err := src.RuleState(ctx, titles[uid]) + + mu.Lock() + defer mu.Unlock() + if err != nil { + if firstErr == nil || uid < firstErrUID { + firstErr, firstErrUID = err, uid + } + return + } + out[uid] = obs + }) + } + wg.Wait() + + if firstErr != nil { + return out, fmt.Errorf("poll rule %q (%s): %w", titles[firstErrUID], firstErrUID, firstErr) + } + return out, nil +} + +// DaemonChildConfig is the detached recorder's whole input, and it is +// deliberately tiny. The rule set and every cadence come from the header the +// parent already wrote — one source of truth, no parent/child drift, and it +// exercises ReadLog's header path — and the connection details come from the +// inherited environment. Only the run facts the header does not carry travel +// in argv (§4.4). +type DaemonChildConfig struct { + URL, Token string // from the inherited environment, never from argv (§20.2) + Out string + Until time.Time + Concurrency int + Clock Clock + // ReadyFD is the inherited descriptor to report readiness on (ReadyFDFlag). + // Zero means nobody is waiting — a hand-started child — and the report is + // then skipped rather than written to stdin. + ReadyFD int +} + +// RunDaemonChild is the detached recorder. The CLI dispatches to it when it +// sees DaemonChildFlag; nothing else ever calls it. +// +// It re-reads the log the parent wrote, restores the transition-marker state +// from the polls already in it, reopens the log for appending, takes the flock +// the parent released, and then polls until it is signalled or reaches Until. +func RunDaemonChild(ctx context.Context, cfg DaemonChildConfig) error { + if cfg.Clock == nil { + cfg.Clock = SystemClock{} + } + if cfg.Concurrency < 1 { + cfg.Concurrency = 1 + } + if cfg.Out == "" { + return fmt.Errorf("recorder: no log path") + } + + // Safe to read: the parent closed its writer before spawning this process, + // and no other writer can hold the log's flock (§4.4 step 4). + header, polls, sentinel, err := ReadLog(cfg.Out) + if err != nil { + return err + } + if sentinel != nil { + return fmt.Errorf("log %s already carries a stopped sentinel: another recorder finished it", cfg.Out) + } + // The header's URL is the log's identity (§19.1 step 3). Checking it here + // catches a child that inherited an environment pointing somewhere else, + // before it appends a single poll from the wrong Grafana. + if header.URL != cfg.URL { + return fmt.Errorf("log %s records url %q but this recorder is configured for %q", cfg.Out, header.URL, cfg.URL) + } + + titles, cadence, err := childSchedule(header) + if err != nil { + return err + } + + writer, err := NewWriter(cfg.Out, cfg.Clock) + if err != nil { + return err + } + + reducer := NewReducer() + reducer.seedFrom(polls) + + // SIGTERM is how check stops the recorder (§4.4 step 1); SIGINT is the + // same request from a human at a terminal. Both are clean stops, so both + // end with a sentinel. Registered before the readiness report, so a signal + // arriving the moment the parent unblocks is already handled. + sigCtx, stop := signal.NotifyContext(ctx, syscall.SIGTERM, syscall.SIGINT) + defer stop() + + // Everything that can fail before a single poll has now succeeded: the + // header parsed, the identity matched, the flock is held. That — and not + // the mere fact of having been started — is what the parent waits for. + if err := reportReady(cfg.ReadyFD); err != nil { + return err + } + + return watchLoop(sigCtx, watchLoopConfig{ + Src: NewHTTPSource(cfg.URL, cfg.Token, cfg.Clock), + Writer: writer, + Reducer: reducer, + Titles: titles, + Cadence: cadence, + Until: cfg.Until, + Concurrency: cfg.Concurrency, + Clock: cfg.Clock, + }) +} + +// reportReady writes one byte to the inherited readiness descriptor and closes +// it. fd 0 means no parent is waiting: descriptor 0 is stdin, so it can never +// be a readiness pipe, which makes the zero value safe as "absent". +func reportReady(fd int) error { + if fd == 0 { + return nil + } + pipe := os.NewFile(uintptr(fd), "ready") + if pipe == nil { + return fmt.Errorf("readiness descriptor %d is not open", fd) + } + defer pipe.Close() + if _, err := pipe.Write([]byte{'1'}); err != nil { + return fmt.Errorf("report ready on descriptor %d: %w", fd, err) + } + return nil +} + +// childSchedule derives what the child polls, and how often, from the header +// alone. The cadence comes from PollEverySeconds — the cadence the recording +// actually uses — and is never re-derived from the rule's evaluation interval: +// that is P5's "two authorities", and getting it wrong is fail-open in the +// faster-override direction. Paused rules are excluded here for the same +// reason the parent never polls them (§4.3, §12). +// +// It returns cadences and nothing else. maxGap, healthGrace and evalStaleAfter +// are coverage thresholds applied by the pure layer at classification time, so +// the recorder must not carry them: it would only be able to misuse them. +func childSchedule(h Header) (titles map[string]string, cadence map[string]time.Duration, err error) { + titles = make(map[string]string, len(h.Rules)) + cadence = make(map[string]time.Duration, len(h.Rules)) + for _, lr := range h.Rules { + if lr.IsPaused { + continue + } + if lr.PollEverySeconds <= 0 { + return nil, nil, fmt.Errorf("log header records poll_every_seconds=%v for rule %s (%q): there is no cadence to record at", + lr.PollEverySeconds, lr.UID, lr.Title) + } + if _, duplicate := titles[lr.UID]; duplicate { + return nil, nil, fmt.Errorf("log header names rule %s (%q) twice; its recorded cadence is ambiguous", lr.UID, lr.Title) + } + titles[lr.UID] = lr.Title + cadence[lr.UID] = time.Duration(lr.PollEverySeconds * float64(time.Second)) + } + return titles, cadence, nil +} + +// watchLoopConfig is the child's working state: what to poll, how often, and +// where to append it. There is no threshold in here and no policy — the child +// records and classifies nothing (H5). +type watchLoopConfig struct { + Src Source + Writer *Writer + Reducer *Reducer + Titles map[string]string // uid -> title: poll by title, select by UID + Cadence map[string]time.Duration // uid -> pollEvery, as recorded in the header + Until time.Time + Concurrency int + Clock Clock +} + +// watchLoop is the child's whole working life: poll the rules that are due, +// reduce each observation to one poll record, append it, and — on a clean stop +// only — finish the log with the stopped sentinel. +// +// The sentinel policy is the load-bearing part. A clean stop (a signal, or +// Until) writes it; a hard error does NOT. A recorder that died must look +// exactly like a coverage gap to check, because it is one (§4.5) — writing a +// sentinel on the way out of a failure would hand check a "recording finished" +// claim about a window that stopped being observed. +func watchLoop(ctx context.Context, cfg watchLoopConfig) error { + sched := NewScheduler(cfg.Cadence, cfg.Clock.Now()) + + for { + if ctx.Err() != nil { + return cfg.Writer.Stop() + } + now := cfg.Clock.Now() + if !cfg.Until.IsZero() && !now.Before(cfg.Until) { + return cfg.Writer.Stop() + } + + due := sched.Due(now) + if len(due) == 0 { + wait, ok := untilNextPoll(sched, cfg.Until, now) + if !ok { + // Nothing will ever come due: every watched rule is paused and + // there is no hard stop. Wait for the signal — and still write + // a sentinel, because "the recorder ran and finished" is + // exactly what check needs to prove about the window. + <-ctx.Done() + return cfg.Writer.Stop() + } + select { + case <-ctx.Done(): + return cfg.Writer.Stop() + case <-cfg.Clock.After(wait): + } + continue + } + + // Mark before polling, against the batch's own now: the next poll is + // one cadence after this one was DUE, not one cadence after it + // returned, so request latency cannot make the heartbeat spacing drift + // towards maxGap. + for _, uid := range due { + sched.Mark(uid, now) + } + + pollErr := cfg.pollBatch(ctx, due) + if ctx.Err() != nil { + // Signalled while a poll was in flight. The aborted poll's error is + // not a recorder failure, and a clean stop wins over it (§4.4 step + // 1: finish the in-flight write, then the sentinel). + return cfg.Writer.Stop() + } + if pollErr != nil { + return pollErr + } + } +} + +// pollBatch polls one round of due rules and appends every poll that +// succeeded, in due order, before returning the first failure. Writing the +// successes first is deliberate: a heartbeat that was genuinely observed is +// evidence, and dropping it because a different rule failed would turn one +// rule's transport failure into a coverage gap for the others. +func (cfg watchLoopConfig) pollBatch(ctx context.Context, uids []string) error { + observed, obsErr := observeAll(ctx, cfg.Src, cfg.Titles, uids, cfg.Concurrency) + for _, uid := range uids { + obs, ok := observed[uid] + if !ok { + continue + } + if err := cfg.Writer.WritePoll(cfg.Reducer.Reduce(uid, obs)); err != nil { + return err + } + } + return obsErr +} + +// untilNextPoll returns how long to wait for the next scheduled poll, cut +// short by Until when that comes first. ok is false when nothing will ever +// come due: no rules to poll and no hard stop. +func untilNextPoll(sched *Scheduler, until, now time.Time) (time.Duration, bool) { + next, hasNext := sched.earliestDue() + switch { + case hasNext && (until.IsZero() || next.Before(until)): + // keep next + case !until.IsZero(): + next = until + default: + return 0, false + } + return max(next.Sub(now), 0), true +} + +// writePidFile records the child's pid where check looks for it (P9's +// --pidfile, default .pid). The format is the decimal pid and a newline, +// so `kill $(cat log.jsonl.pid)` works and ReadPidFile stays trivial. +func writePidFile(path string, pid int) error { + if err := os.WriteFile(path, []byte(strconv.Itoa(pid)+"\n"), 0o644); err != nil { + return fmt.Errorf("write pidfile %s: %w", path, err) + } + return nil +} + +// ReadPidFile is the other side of that contract: the pid of the recorder +// check must stop before it may read the log (§4.4 steps 1-4). +func ReadPidFile(path string) (int, error) { + b, err := os.ReadFile(path) + if err != nil { + return 0, fmt.Errorf("read pidfile %s: %w", path, err) + } + text := strings.TrimSpace(string(b)) + pid, err := strconv.Atoi(text) + if err != nil { + return 0, fmt.Errorf("pidfile %s: unparseable pid %q", path, text) + } + if pid <= 0 { + return 0, fmt.Errorf("pidfile %s: %d is not a pid", path, pid) + } + return pid, nil +} diff --git a/grafana-alertcheck/internal/gate/watch_daemon_test.go b/grafana-alertcheck/internal/gate/watch_daemon_test.go new file mode 100644 index 000000000..8a044c7e9 --- /dev/null +++ b/grafana-alertcheck/internal/gate/watch_daemon_test.go @@ -0,0 +1,308 @@ +//go:build unix + +package gate + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "slices" + "strconv" + "strings" + "syscall" + "testing" + "time" +) + +// TestMain doubles this test binary as the detached recorder. Watch spawns +// os.Executable(), which under `go test` is this binary, so the one integration +// test below exercises the real thing — a real fork/exec, a real setsid, a real +// inherited environment, a real SIGTERM — with this function standing in for +// the CLI's `watch --daemon-child` dispatch, which lands in P10. +func TestMain(m *testing.M) { + if slices.Contains(os.Args, DaemonChildFlag) { + os.Exit(runTestDaemonChild(os.Args[1:])) + } + os.Exit(m.Run()) +} + +// runTestDaemonChild parses the child argv childArgs() writes, and reads the +// connection details from the environment — never from argv (§20.2). P10's +// `watch` FlagSet does the same four flags. +func runTestDaemonChild(args []string) int { + cfg := DaemonChildConfig{ + URL: os.Getenv("GRAFANA_URL"), + Token: os.Getenv("GRAFANA_TOKEN"), + } + for i := 0; i < len(args); i++ { + value := func() string { + if i+1 >= len(args) { + fmt.Fprintf(os.Stderr, "flag %s wants a value\n", args[i]) + os.Exit(2) + } + i++ + return args[i] + } + switch args[i] { + case "--out": + cfg.Out = value() + case "--until": + until, err := time.Parse(time.RFC3339, value()) + if err != nil { + fmt.Fprintf(os.Stderr, "--until: %v\n", err) + return 2 + } + cfg.Until = until + case "--concurrency": + n, err := strconv.Atoi(value()) + if err != nil { + fmt.Fprintf(os.Stderr, "--concurrency: %v\n", err) + return 2 + } + cfg.Concurrency = n + case ReadyFDFlag: + fd, err := strconv.Atoi(value()) + if err != nil { + fmt.Fprintf(os.Stderr, "%s: %v\n", ReadyFDFlag, err) + return 2 + } + cfg.ReadyFD = fd + } + } + if err := RunDaemonChild(context.Background(), cfg); err != nil { + fmt.Fprintln(os.Stderr, err) + return 2 + } + return 0 +} + +// testBearerToken is what every request to grafanaTestServer must carry. The +// child never receives it in argv (§20.2), so a request that arrives +// authenticated is proof that the token reached the detached process through +// the inherited environment — and a 401 is what a test sees if that ever +// breaks. +const testBearerToken = "test-token" + +// grafanaTestServer serves the three endpoints the record step reads, from the +// real captured fixtures: /api/health, the ruler definitions, and the +// rule_name-filtered state response. The state body is the one-instance +// fixture with its uid and name patched to the ruler fixture's live rule, so +// the reducer's select-by-UID finds it. +func grafanaTestServer(t *testing.T) *httptest.Server { + t.Helper() + ruler := readFixture(t, "ruler_rules.json") + state := patchedStateBody(t) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "Bearer "+testBearerToken { + // Not t.Errorf: this must reach the client as a real 401, so the + // parent fails its version gate and the child fails its polls. + http.Error(w, "unauthorized: Authorization = "+got, http.StatusUnauthorized) + return + } + w.Header().Set("Content-Type", "application/json") + switch { + case r.URL.Path == "/api/health": + fmt.Fprint(w, healthBody("13.1.0")) + case strings.HasPrefix(r.URL.Path, "/api/ruler/"): + _, _ = w.Write(ruler) + case strings.HasPrefix(r.URL.Path, "/api/prometheus/"): + if r.URL.Query().Get("rule_name") == "" { + // §2.8: the gate must never read the state endpoint unfiltered. + http.Error(w, "unfiltered state read", http.StatusBadRequest) + return + } + _, _ = w.Write(state) + default: + http.Error(w, "unexpected path "+r.URL.Path, http.StatusNotFound) + } + })) + t.Cleanup(srv.Close) + return srv +} + +func patchedStateBody(t *testing.T) []byte { + t.Helper() + var body map[string]any + if err := json.Unmarshal(readFixture(t, "state_one_instance.json"), &body); err != nil { + t.Fatalf("unmarshal state fixture: %v", err) + } + data, ok := body["data"].(map[string]any) + if !ok { + t.Fatal("state fixture: no data object") + } + groups, ok := data["groups"].([]any) + if !ok || len(groups) == 0 { + t.Fatal("state fixture: no groups") + } + group, ok := groups[0].(map[string]any) + if !ok { + t.Fatal("state fixture: group 0 is not an object") + } + rules, ok := group["rules"].([]any) + if !ok || len(rules) == 0 { + t.Fatal("state fixture: group 0 has no rules") + } + rule, ok := rules[0].(map[string]any) + if !ok { + t.Fatal("state fixture: rule 0 is not an object") + } + rule["uid"] = watchActiveUID + rule["name"] = watchActiveTitle + rule["lastEvaluation"] = time.Now().UTC().Format(time.RFC3339Nano) + + b, err := json.Marshal(body) + if err != nil { + t.Fatalf("marshal patched state fixture: %v", err) + } + return b +} + +// waitFor polls cond until it holds. This is the one tier of the project where +// a test waits on real time: it drives real processes over real HTTP, so there +// is no clock to fake. +func waitFor(t *testing.T, what string, timeout time.Duration, cond func() bool) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("timed out after %s waiting for %s", timeout, what) +} + +// TestWatchSpawnsADetachedRecorder is P6's one integration test: everything +// from the version gate to the sentinel, through a real detached process. +// +// It asserts the four things only a real spawn can show — the pidfile points +// at a live process, that process is in its own session (setsid, not a bare +// `&`), it keeps appending after Watch returned, and SIGTERM makes it finish +// the log in the §4.4 order — and it uses a 200ms --poll-interval to do it in +// about a second, which also exercises the unclamped-override path (§5.1). +func TestWatchSpawnsADetachedRecorder(t *testing.T) { + srv := grafanaTestServer(t) + t.Setenv("GRAFANA_URL", srv.URL) + t.Setenv("GRAFANA_TOKEN", testBearerToken) + + out := filepath.Join(t.TempDir(), "log.jsonl") + var notes strings.Builder + cfg := WatchConfig{ + URL: srv.URL, + Token: testBearerToken, + Alerts: []string{"uid:" + watchActiveUID}, + Out: out, + PollEvery: 200 * time.Millisecond, + Concurrency: 2, + Notes: ¬es, + } + + if err := Watch(context.Background(), cfg); err != nil { + t.Fatalf("Watch: %v\nnotes:\n%s", err, notes.String()) + } + t.Cleanup(func() { + if t.Failed() { + t.Logf("notes:\n%s", notes.String()) + t.Logf("daemon log:\n%s", daemonLogTail(out+".daemon.log", 0)) + } + }) + + pid, err := ReadPidFile(out + ".pid") + if err != nil { + t.Fatalf("ReadPidFile: %v", err) + } + if err := syscall.Kill(pid, 0); err != nil { + t.Fatalf("recorder pid %d is not running right after Watch returned: %v", pid, err) + } + // Setsid, not a bare `&`: a session leader's process group id is its own + // pid. Without this the child would still share the parent's process group + // and die with the step that started it. + if pgid, err := syscall.Getpgid(pid); err != nil { + t.Errorf("Getpgid(%d): %v", pid, err) + } else if pgid != pid { + t.Errorf("recorder pgid = %d, want %d: it did not get its own session", pgid, pid) + } + + // The parent already wrote the first heartbeat before it returned (§4.3); + // these later ones prove the detached child is the one appending now. + waitFor(t, "the detached recorder to append its own polls", 10*time.Second, func() bool { + _, polls, _, err := ReadLog(out) + return err == nil && len(polls) >= 3 + }) + + // Stop it exactly the way check does. + if err := syscall.Kill(pid, syscall.SIGTERM); err != nil { + t.Fatalf("SIGTERM %d: %v", pid, err) + } + waitFor(t, "the stopped sentinel", 10*time.Second, func() bool { + _, _, sentinel, err := ReadLog(out) + return err == nil && sentinel != nil + }) + + header, polls, sentinel, err := ReadLog(out) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if header.URL != srv.URL || header.GrafanaVersion != "13.1.0" { + t.Errorf("header identity = %q/%q, want %q/13.1.0", header.URL, header.GrafanaVersion, srv.URL) + } + if len(header.Rules) != 1 || header.Rules[0].PollEverySeconds != 0.2 { + t.Errorf("header rules = %+v, want one rule recorded at 0.2s", header.Rules) + } + for i, p := range polls { + if p.RuleUID != watchActiveUID || !p.Found { + t.Fatalf("poll %d = %+v, want a found observation of %s", i, p, watchActiveUID) + } + if p.GrafanaNow.IsZero() { + t.Fatalf("poll %d has no grafana_now; H4 needs the Date header of its own response", i) + } + } + if sentinel.Before(header.StartedAt) { + t.Errorf("sentinel at %s precedes the record start %s", sentinel, header.StartedAt) + } + + waitFor(t, "the recorder to exit", 10*time.Second, func() bool { + return syscall.Kill(pid, 0) != nil + }) +} + +// TestWatchFailsWhenTheChildCannotStartRecording is the other half of the +// readiness contract. The child dies on its identity check, so it never reports +// ready — and Watch must say so instead of returning success over a window +// nothing is recording, and must leave no pidfile naming a dead process for the +// next step to signal. +// +// The child is made to fail through the environment, which is the only channel +// it takes its connection details from: the header says one URL and the +// inherited GRAFANA_URL says another. +func TestWatchFailsWhenTheChildCannotStartRecording(t *testing.T) { + srv := grafanaTestServer(t) + t.Setenv("GRAFANA_URL", srv.URL+"/somewhere-else") + t.Setenv("GRAFANA_TOKEN", testBearerToken) + + out := filepath.Join(t.TempDir(), "log.jsonl") + var notes strings.Builder + err := Watch(context.Background(), WatchConfig{ + URL: srv.URL, // what the parent uses, and what the header records + Token: testBearerToken, + Alerts: []string{"uid:" + watchActiveUID}, + Out: out, + PollEvery: 200 * time.Millisecond, + Concurrency: 2, + Notes: ¬es, + }) + if err == nil { + t.Fatal("Watch: no error, but the child could never have started recording") + } + if !strings.Contains(err.Error(), "records url") { + t.Errorf("error does not quote the child's own reason:\n%v", err) + } + if _, statErr := os.Stat(out + ".pid"); !os.IsNotExist(statErr) { + t.Errorf("a pidfile survived a failed detach (%v); pids are reused, so the next step would signal a stranger", statErr) + } +} diff --git a/grafana-alertcheck/internal/gate/watch_test.go b/grafana-alertcheck/internal/gate/watch_test.go new file mode 100644 index 000000000..cecd3a45e --- /dev/null +++ b/grafana-alertcheck/internal/gate/watch_test.go @@ -0,0 +1,702 @@ +package gate + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "testing" + "time" +) + +// The two fixture rules every prepareWatch test below uses: one live, one +// paused in its definition. Both are addressed by uid:, because the ruler +// fixture deliberately contains a 2-way title collision and a title would make +// the tests depend on which side of it they hit. +const ( + watchActiveUID = "rule0000009" + watchActiveTitle = "Example Failure Ratio Above 10 Percent" + watchPausedUID = "rule0000007" + watchPausedTitle = "example_workflow_paused_rule" +) + +// loopSource answers every RuleState call from a responder that also sees the +// call count, so a recorder-loop test can make the answer depend on virtual +// time or fail on the Nth poll. The loop never reads Version or Definitions — +// the parent did that before detaching — so both fail loudly here. +type loopSource struct { + mu sync.Mutex + calls map[string]int + respond func(title string, call int) (Observation, error) +} + +func newLoopSource(respond func(title string, call int) (Observation, error)) *loopSource { + return &loopSource{calls: map[string]int{}, respond: respond} +} + +func (s *loopSource) Version(context.Context) (string, error) { + return "", errors.New("loopSource: the recorder loop must not read the version") +} + +func (s *loopSource) Definitions(context.Context) ([]Definition, error) { + return nil, errors.New("loopSource: the recorder loop must not read the definitions") +} + +func (s *loopSource) RuleState(_ context.Context, title string) (Observation, error) { + s.mu.Lock() + s.calls[title]++ + call := s.calls[title] + s.mu.Unlock() + return s.respond(title, call) +} + +var _ Source = (*loopSource)(nil) + +// testStateRule is one rule as the state endpoint would return it, healthy and +// evaluated at grafanaNow. +func testStateRule(uid, title string, interval time.Duration, grafanaNow time.Time, instances ...Instance) StateRule { + totals := map[string]int{"normal": len(instances)} + return StateRule{ + UID: uid, Title: title, Folder: "F", Group: "G", + Interval: interval, State: "inactive", Health: "ok", + LastEvaluation: grafanaNow, Totals: totals, Instances: instances, + } +} + +// newLoopWriter opens a log with a header already written, exactly as the +// parent hands it to the child. +func newLoopWriter(t *testing.T, path string, clock Clock) *Writer { + t.Helper() + w, err := NewWriter(path, clock) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + return w +} + +func countPolls(polls []Poll, uid string) int { + n := 0 + for _, p := range polls { + if p.RuleUID == uid { + n++ + } + } + return n +} + +// TestWatchLoopPollsEachRuleAtItsOwnCadence is §5's per-rule schedule seen +// from the recorder: a 10s rule beside a 300s one keeps its own 5s cadence +// instead of dragging the slack rule along with it or being slowed to its pace. +func TestWatchLoopPollsEachRuleAtItsOwnCadence(t *testing.T) { + const tightUID, slackUID = "tight", "slack" + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newVirtualClock(testNow) + w := newLoopWriter(t, path, clock) + + src := newLoopSource(func(title string, _ int) (Observation, error) { + uid := tightUID + if title == "Slack Rule" { + uid = slackUID + } + now := clock.Now() + return observation(now, testStateRule(uid, title, time.Minute, now)), nil + }) + + err := watchLoop(context.Background(), watchLoopConfig{ + Src: src, + Writer: w, + Reducer: NewReducer(), + Titles: map[string]string{tightUID: "Tight Rule", slackUID: "Slack Rule"}, + Cadence: map[string]time.Duration{ + tightUID: 5 * time.Second, + slackUID: 150 * time.Second, + }, + Until: testNow.Add(300 * time.Second), + Concurrency: 2, + Clock: clock, + }) + if err != nil { + t.Fatalf("watchLoop: %v", err) + } + + _, polls, sentinel, readErr := ReadLog(path) + if readErr != nil { + t.Fatalf("ReadLog: %v", readErr) + } + if sentinel == nil { + t.Fatal("no stopped sentinel after a clean stop") + } + if sentinel.Before(testNow.Add(300 * time.Second)) { + t.Errorf("sentinel at %s, want >= the stop time %s", sentinel, testNow.Add(300*time.Second)) + } + // 300s of window at 5s and 150s, minus the initial stagger offset of up to + // one cadence: 59-60 and 1-2. The assertion is the ratio, not the exact + // count — a single global cycle would give both rules the same number. + if got := countPolls(polls, tightUID); got < 59 || got > 61 { + t.Errorf("tight rule polled %d times, want ~60 (300s at 5s)", got) + } + if got := countPolls(polls, slackUID); got < 1 || got > 3 { + t.Errorf("slack rule polled %d times, want ~2 (300s at 150s)", got) + } +} + +// TestWatchLoopHardErrorLeavesNoSentinel is §4.5's fail-closed rule from the +// recorder's side: a recorder that dies must look exactly like a coverage gap, +// so it must not sign off the log on its way out. +func TestWatchLoopHardErrorLeavesNoSentinel(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newVirtualClock(testNow) + w := newLoopWriter(t, path, clock) + + boom := errors.New("grafana went away for good") + src := newLoopSource(func(title string, call int) (Observation, error) { + if call >= 2 { + return Observation{}, boom + } + now := clock.Now() + return observation(now, testStateRule("r1", title, time.Minute, now)), nil + }) + + err := watchLoop(context.Background(), watchLoopConfig{ + Src: src, + Writer: w, + Reducer: NewReducer(), + Titles: map[string]string{"r1": "Example"}, + Cadence: map[string]time.Duration{"r1": 30 * time.Second}, + Until: testNow.Add(time.Hour), + Concurrency: 1, + Clock: clock, + }) + if !errors.Is(err, boom) { + t.Fatalf("watchLoop error = %v, want %v", err, boom) + } + + _, polls, sentinel, readErr := ReadLog(path) + if readErr != nil { + t.Fatalf("ReadLog: %v", readErr) + } + if sentinel != nil { + t.Errorf("sentinel at %s after a failed recording; check would read that as a finished window", sentinel) + } + if len(polls) != 1 { + t.Errorf("kept %d polls, want the 1 that succeeded before the failure", len(polls)) + } +} + +// TestWatchLoopSignalDuringPollIsACleanStop pins §4.4 step 1: SIGTERM arriving +// while a poll is in flight is a clean stop, so the aborted poll's error must +// not suppress the sentinel — otherwise every normal check run, which stops the +// recorder exactly this way, would end unobservable. +func TestWatchLoopSignalDuringPollIsACleanStop(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newVirtualClock(testNow) + w := newLoopWriter(t, path, clock) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + src := newLoopSource(func(title string, call int) (Observation, error) { + if call >= 2 { + // The signal lands while this request is out. + cancel() + return Observation{}, ctx.Err() + } + now := clock.Now() + return observation(now, testStateRule("r1", title, time.Minute, now)), nil + }) + + if err := watchLoop(ctx, watchLoopConfig{ + Src: src, + Writer: w, + Reducer: NewReducer(), + Titles: map[string]string{"r1": "Example"}, + Cadence: map[string]time.Duration{"r1": 30 * time.Second}, + Concurrency: 1, + Clock: clock, + }); err != nil { + t.Fatalf("watchLoop: %v", err) + } + + if _, _, sentinel, err := ReadLog(path); err != nil { + t.Fatalf("ReadLog: %v", err) + } else if sentinel == nil { + t.Error("no sentinel after a signalled stop; check would call a fully observed window unobservable") + } +} + +// TestWatchLoopWithNothingToPollStillFinishesTheLog covers the every-rule-is- +// paused case: there is nothing to record, but "the recorder ran and finished" +// is still what check has to prove about the window. +func TestWatchLoopWithNothingToPollStillFinishesTheLog(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newVirtualClock(testNow) + w := newLoopWriter(t, path, clock) + + src := newLoopSource(func(title string, _ int) (Observation, error) { + return Observation{}, fmt.Errorf("nothing should be polled, got %q", title) + }) + + if err := watchLoop(context.Background(), watchLoopConfig{ + Src: src, + Writer: w, + Reducer: NewReducer(), + Titles: map[string]string{}, + Cadence: map[string]time.Duration{}, + Until: testNow.Add(time.Minute), + Concurrency: 1, + Clock: clock, + }); err != nil { + t.Fatalf("watchLoop: %v", err) + } + + _, polls, sentinel, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if len(polls) != 0 { + t.Errorf("wrote %d polls with nothing to poll", len(polls)) + } + if sentinel == nil { + t.Error("no sentinel: check cannot tell this recording from one that died") + } +} + +// TestWatchLoopPollBatchKeepsTheHeartbeatsItGot: one rule's failure must not +// discard another rule's observed heartbeat, or a single transport failure +// turns into a coverage gap for every rule that answered. +func TestWatchLoopPollBatchKeepsTheHeartbeatsItGot(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newVirtualClock(testNow) + w := newLoopWriter(t, path, clock) + + boom := errors.New("one rule is unreachable") + src := newLoopSource(func(title string, _ int) (Observation, error) { + if title == "Broken" { + return Observation{}, boom + } + now := clock.Now() + return observation(now, testStateRule("ok", title, time.Minute, now)), nil + }) + + cfg := watchLoopConfig{ + Src: src, + Writer: w, + Reducer: NewReducer(), + Titles: map[string]string{"ok": "Healthy", "bad": "Broken"}, + Cadence: map[string]time.Duration{"ok": 30 * time.Second, "bad": 30 * time.Second}, + Concurrency: 2, + Clock: clock, + } + if err := cfg.pollBatch(context.Background(), []string{"ok", "bad"}); !errors.Is(err, boom) { + t.Fatalf("pollBatch error = %v, want %v", err, boom) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + _, polls, _, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if len(polls) != 1 || polls[0].RuleUID != "ok" { + t.Errorf("polls = %+v, want the one heartbeat that was actually observed", polls) + } +} + +// TestReducerSeedFromKeepsMarkersAcrossTheHandoff is H2 at the one seam P6 +// introduces. The parent observes a firing instance; the child starts with a +// fresh Reducer and sees the instance gone. Seeded, that is a vanish — a +// discontinuity. Unseeded, it is nothing at all, and the instance silently +// leaves the record as if it had never been bad. +func TestReducerSeedFromKeepsMarkersAcrossTheHandoff(t *testing.T) { + firing := testInstance(StateFiring, "", "b") + key := instanceKey(firing.Labels) + parentPoll := Poll{RuleUID: "r1", Found: true, Abnormal: []Instance{firing}} + // The child's first response: the instance is gone from the response + // entirely, which is a vanish and never a clear (§4.7). + childObs := observation(testNow, testStateRule("r1", "Example", time.Minute, testNow)) + + t.Run("seeded", func(t *testing.T) { + r := NewReducer() + r.seedFrom([]Poll{parentPoll}) + p := r.Reduce("r1", childObs) + if !slices.Contains(p.Vanished, key) { + t.Errorf("vanished = %v, want it to contain %q", p.Vanished, key) + } + if len(p.Cleared) != 0 { + t.Errorf("cleared = %v, want none: a vanish is not a recovery", p.Cleared) + } + }) + + t.Run("unseeded loses the transition", func(t *testing.T) { + p := NewReducer().Reduce("r1", childObs) + if len(p.Vanished) != 0 { + t.Fatalf("vanished = %v; this subtest exists to show the seed is what produces the marker", p.Vanished) + } + }) + + t.Run("a not-found poll does not clear the seed", func(t *testing.T) { + r := NewReducer() + r.seedFrom([]Poll{parentPoll, {RuleUID: "r1", Found: false}}) + if p := r.Reduce("r1", childObs); !slices.Contains(p.Vanished, key) { + t.Errorf("vanished = %v, want it to contain %q: an absent rule leaves the abnormal set untouched", p.Vanished, key) + } + }) +} + +// watchTestConfig is a prepareWatch config over a temp log, with the notes +// captured so the tests can assert on what an operator is told. +func watchTestConfig(t *testing.T, notes *strings.Builder, alerts ...string) WatchConfig { + t.Helper() + return WatchConfig{ + URL: "https://grafana.example.com", + Token: "secret-token", + Alerts: alerts, + Out: filepath.Join(t.TempDir(), "log.jsonl"), + Concurrency: 2, + Clock: newFakeClock(testNow), + Notes: notes, + }.withDefaults() +} + +// watchTestSource is a fakeSource with the real ruler fixture and a scripted +// state response for the live rule only. The paused rule is deliberately +// unscripted: fakeSource errors on an unscripted title, so any attempt to poll +// it fails the test rather than passing silently. +func watchTestSource(t *testing.T, obs Observation) *fakeSource { + t.Helper() + src := newFakeSource() + src.version = "13.1.0" + src.defs = rulerDefs(t) + src.script(watchActiveTitle, obs, nil) + return src +} + +func liveObservation(grafanaNow time.Time) Observation { + return observation(grafanaNow, testStateRule(watchActiveUID, watchActiveTitle, time.Minute, grafanaNow, + testInstance(StateNormal, "", "a"))) +} + +// TestPrepareWatchDoesNotWaitForPausedRules is §22.4's regression test: a rule +// paused in its definition is skipped, never waited for. Waiting for one either +// hangs forever or errors before the deploy — and the header must still name +// it, so check can report it as skipped rather than lose it. +func TestPrepareWatchDoesNotWaitForPausedRules(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID, "uid:"+watchPausedUID) + src := watchTestSource(t, liveObservation(testNow)) + + prep, err := prepareWatch(context.Background(), cfg, src) + if err != nil { + t.Fatalf("prepareWatch: %v", err) + } + if err := prep.writer.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + header, polls, sentinel, err := ReadLog(cfg.Out) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if sentinel != nil { + t.Error("the parent wrote a sentinel; that would tell check the recording ended before the child started") + } + + if len(header.Rules) != 2 { + t.Fatalf("header names %d rules, want both the live and the paused one", len(header.Rules)) + } + for _, lr := range header.Rules { + if lr.PollEverySeconds <= 0 { + t.Errorf("header rule %s records poll_every_seconds=%v; check needs a positive cadence to derive maxGap from", lr.UID, lr.PollEverySeconds) + } + if lr.UID == watchPausedUID && !lr.IsPaused { + t.Errorf("header rule %s: is_paused = false, want the resolve-time snapshot to say true", lr.UID) + } + } + + // One poll, for the live rule only — and it is already in the log before + // prepareWatch returned, which is the whole point of §4.3. + if len(polls) != 1 || polls[0].RuleUID != watchActiveUID { + t.Fatalf("polls = %+v, want exactly one first observation of %s", polls, watchActiveUID) + } + if !polls[0].Found || !polls[0].GrafanaNow.Equal(testNow) { + t.Errorf("first poll = %+v, want a found observation at %s", polls[0], testNow) + } + if !strings.Contains(notes.String(), watchPausedTitle) || !strings.Contains(notes.String(), "paused") { + t.Errorf("notes do not mention the paused rule:\n%s", notes.String()) + } +} + +// TestPrepareWatchHeaderRecordsTheOverriddenCadence is P5's "two authorities" +// from the writing side: whatever --poll-interval resolves to is what the +// header records, because that is the only value check may derive maxGap from. +func TestPrepareWatchHeaderRecordsTheOverriddenCadence(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID) + cfg.PollEvery = 120 * time.Second // the rule evaluates every 60s + src := watchTestSource(t, liveObservation(testNow)) + + prep, err := prepareWatch(context.Background(), cfg, src) + if err != nil { + t.Fatalf("prepareWatch: %v", err) + } + defer prep.writer.Close() + + if got := prep.header.Rules[0].PollEverySeconds; got != 120 { + t.Errorf("header poll_every_seconds = %v, want 120 (the override, used verbatim and never clamped)", got) + } + if got := prep.timings[watchActiveUID].maxGap; got != 240*time.Second { + t.Errorf("maxGap = %s, want 240s (2 x the recorded cadence)", got) + } + if !strings.Contains(notes.String(), "--poll-interval") { + t.Errorf("notes do not report that the override exceeds half the evaluation interval:\n%s", notes.String()) + } +} + +// TestPrepareWatchFailsWhenTheScheduleDoesNotFit: the budget check runs on the +// latencies the parent just measured, before the deploy runs (§5.2). +func TestPrepareWatchFailsWhenTheScheduleDoesNotFit(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID) + + obs := liveObservation(testNow) + obs.Latency = 60 * time.Second // against a 30s cadence + src := watchTestSource(t, obs) + + _, err := prepareWatch(context.Background(), cfg, src) + if err == nil { + t.Fatal("prepareWatch: no error on a schedule that cannot hold its own cadence") + } + assertBudgetMessage(t, err.Error()) +} + +// TestPrepareWatchVerifiesNormalInstancesAreVisible is the §3.2 check at the +// one place it can still be cheap: the first observation. If the state endpoint +// stops returning normal instances, the reduction's predicate quietly inverts. +func TestPrepareWatchVerifiesNormalInstancesAreVisible(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID) + + rule := testStateRule(watchActiveUID, watchActiveTitle, time.Minute, testNow, testInstance(StateFiring, "", "b")) + rule.Totals = map[string]int{"alerting": 1, "normal": 4} // claims normals it did not return + src := watchTestSource(t, observation(testNow, rule)) + + _, err := prepareWatch(context.Background(), cfg, src) + if err == nil { + t.Fatal("prepareWatch: no error when totals claim normal instances the response omitted") + } + if !strings.Contains(err.Error(), "3.2") { + t.Errorf("error does not name §3.2: %v", err) + } + + // The failure happens before any poll is appended, so the log holds a + // header and nothing else. + if _, polls, _, readErr := ReadLog(cfg.Out); readErr != nil { + t.Fatalf("ReadLog: %v", readErr) + } else if len(polls) != 0 { + t.Errorf("wrote %d polls from an observation it refused to trust", len(polls)) + } +} + +func TestPrepareWatchRejectsAnUnsupportedGrafana(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID) + src := watchTestSource(t, liveObservation(testNow)) + src.version = "12.4.0" + + if _, err := prepareWatch(context.Background(), cfg, src); err == nil { + t.Fatal("prepareWatch: no error on an unsupported grafana version") + } else if !strings.Contains(err.Error(), "12.4.0") || !strings.Contains(err.Error(), "13.0.0") { + t.Errorf("error names neither what was found nor what is supported: %v", err) + } +} + +// TestPrepareWatchNotesAnAbsentRule: a rule that resolved in the ruler API but +// is absent from the state endpoint is recorded as Found=false — authoritative +// evidence P7 turns into unobservable — not silently dropped. +func TestPrepareWatchNotesAnAbsentRule(t *testing.T) { + var notes strings.Builder + cfg := watchTestConfig(t, ¬es, "uid:"+watchActiveUID) + src := watchTestSource(t, observation(testNow)) // an authoritative, empty 2xx + + prep, err := prepareWatch(context.Background(), cfg, src) + if err != nil { + t.Fatalf("prepareWatch: %v", err) + } + if err := prep.writer.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + _, polls, _, err := ReadLog(cfg.Out) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if len(polls) != 1 || polls[0].Found { + t.Fatalf("polls = %+v, want one poll recorded as not found", polls) + } + if !strings.Contains(notes.String(), "absent from the state endpoint") { + t.Errorf("notes do not warn about the absent rule:\n%s", notes.String()) + } +} + +func TestWatchConfigValidation(t *testing.T) { + base := func() WatchConfig { + return WatchConfig{ + URL: "https://grafana.example.com", + Alerts: []string{"Example"}, + Out: filepath.Join(t.TempDir(), "log.jsonl"), + Clock: newFakeClock(testNow), + } + } + + for _, tc := range []struct { + name string + mutate func(*WatchConfig) + want string + }{ + {"no url", func(c *WatchConfig) { c.URL = "" }, "url"}, + {"no log path", func(c *WatchConfig) { c.Out = "" }, "no log path"}, + {"no alerts", func(c *WatchConfig) { c.Alerts = nil }, "no alert names"}, + {"blank alerts only", func(c *WatchConfig) { c.Alerts = []string{"", " "} }, "no alert names"}, + {"until in the past", func(c *WatchConfig) { c.Until = testNow.Add(-time.Second) }, "not in the future"}, + } { + t.Run(tc.name, func(t *testing.T) { + cfg := base() + tc.mutate(&cfg) + err := cfg.withDefaults().validate() + if err == nil { + t.Fatalf("validate: no error, want one naming %q", tc.want) + } + if !strings.Contains(err.Error(), tc.want) { + t.Errorf("validate error = %v, want it to name %q", err, tc.want) + } + }) + } + + t.Run("defaults derive the pidfile and daemon log from the log path", func(t *testing.T) { + cfg := base().withDefaults() + if cfg.PidFile != cfg.Out+".pid" { + t.Errorf("PidFile = %q, want %q — check finds the recorder by this convention", cfg.PidFile, cfg.Out+".pid") + } + if cfg.DaemonLog == "" { + t.Error("DaemonLog is empty: a detached child would have nowhere to explain a failure") + } + if err := cfg.validate(); err != nil { + t.Errorf("validate: %v", err) + } + }) +} + +// TestChildScheduleUsesTheRecordedCadence is P5's fail-open direction, checked +// on the child's side: a log recorded at 5s on a 300s rule must schedule at 5s. +// Re-deriving from the interval would give 150s — and every real 250s hole in +// that recording would pass. +func TestChildScheduleUsesTheRecordedCadence(t *testing.T) { + h := Header{Rules: []LoggedRule{ + {UID: "fast", Title: "Fast", IntervalSeconds: 300, PollEverySeconds: 5}, + {UID: "paused", Title: "Paused", IntervalSeconds: 60, PollEverySeconds: 30, IsPaused: true}, + }} + + titles, cadence, err := childSchedule(h) + if err != nil { + t.Fatalf("childSchedule: %v", err) + } + if _, ok := titles["paused"]; ok { + t.Error("the child scheduled a rule that was paused when the window opened (§4.3)") + } + if got := cadence["fast"]; got != 5*time.Second { + t.Errorf("pollEvery = %s, want 5s from the header, not %s from the interval", got, defaultPollEvery(300)) + } +} + +func TestChildScheduleRejectsAnUnusableHeader(t *testing.T) { + for _, tc := range []struct { + name string + h Header + want string + }{ + { + "no recorded cadence", + Header{Rules: []LoggedRule{{UID: "r1", Title: "Example", IntervalSeconds: 60}}}, + "no cadence", + }, + { + "the same rule twice", + Header{Rules: []LoggedRule{ + {UID: "r1", Title: "Example", IntervalSeconds: 60, PollEverySeconds: 30}, + {UID: "r1", Title: "Example", IntervalSeconds: 60, PollEverySeconds: 300}, + }}, + "twice", + }, + } { + t.Run(tc.name, func(t *testing.T) { + if _, _, err := childSchedule(tc.h); err == nil { + t.Fatalf("childSchedule: no error, want one naming %q", tc.want) + } else if !strings.Contains(err.Error(), tc.want) { + t.Errorf("error = %v, want it to name %q", err, tc.want) + } + }) + } +} + +// TestChildArgsCarryNoSecretsAndNoRuleSet: the child's command line lands in +// the process table and in CI logs. Everything it needs about the rules comes +// from the header, and everything about the connection comes from the +// environment — so argv holds the log path and the two run facts only. +func TestChildArgsCarryNoSecretsAndNoRuleSet(t *testing.T) { + cfg := WatchConfig{ + URL: "https://grafana.example.com", + Token: "secret-token", + Alerts: []string{"Example"}, + Folder: "F", + Out: "/tmp/log.jsonl", + PidFile: "/tmp/log.jsonl.pid", + Until: testNow.Add(time.Hour), + PollEvery: 17 * time.Second, + Concurrency: 3, + } + args := childArgs(cfg) + joined := strings.Join(args, " ") + + for _, want := range []string{DaemonChildFlag, "--out /tmp/log.jsonl", "--concurrency 3", "--until ", ReadyFDFlag + " 3"} { + if !strings.Contains(joined, want) { + t.Errorf("child args %q do not contain %q", joined, want) + } + } + for _, forbidden := range []string{"secret-token", "Example", "--folder", "--poll-interval", "--pidfile"} { + if strings.Contains(joined, forbidden) { + t.Errorf("child args %q contain %q, which must not reach argv", joined, forbidden) + } + } +} + +func TestPidFileRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl.pid") + if err := writePidFile(path, 4242); err != nil { + t.Fatalf("writePidFile: %v", err) + } + pid, err := ReadPidFile(path) + if err != nil { + t.Fatalf("ReadPidFile: %v", err) + } + if pid != 4242 { + t.Errorf("pid = %d, want 4242", pid) + } + + t.Run("garbage is an error, never a pid", func(t *testing.T) { + bad := filepath.Join(t.TempDir(), "bad.pid") + if err := os.WriteFile(bad, []byte("not-a-pid\n"), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + if _, err := ReadPidFile(bad); err == nil { + t.Error("ReadPidFile: no error on an unparseable pidfile") + } + }) +} diff --git a/grafana-alertcheck/internal/gate/watch_unix.go b/grafana-alertcheck/internal/gate/watch_unix.go new file mode 100644 index 000000000..6d6096786 --- /dev/null +++ b/grafana-alertcheck/internal/gate/watch_unix.go @@ -0,0 +1,112 @@ +//go:build unix + +package gate + +import ( + "fmt" + "os" + "os/exec" + "strconv" + "syscall" + "time" +) + +// readyFD is where ExtraFiles[0] lands in the child: exec.Cmd starts extra +// descriptors at 3, after stdin, stdout and stderr. +const readyFD = 3 + +// detachedChild is a started recorder: the process, the read end of its +// readiness pipe, and the size the daemon log had before it wrote anything. +type detachedChild struct { + cmd *exec.Cmd + ready *os.File + // logOffset is where this run's output starts in the daemon log, so a + // failure quotes this child and never a previous run's. + logOffset int64 +} + +// spawnChild re-execs this binary as the detached recorder (§4.4). A trailing +// `&` is NOT sufficient: the child would keep the parent's session and process +// group, so it would still take the terminal's signals and, on a runner, die +// with the step that started it. Setsid gives it a new session AND a new +// process group, which is what makes it survive to the end of the window. +// +// There is deliberately no Windows counterpart — runners are Linux and +// goreleaser builds linux+darwin only (P12) — so the package does not build +// there at all rather than silently recording in the foreground. +func spawnChild(cfg WatchConfig) (detachedChild, error) { + exe, err := os.Executable() + if err != nil { + return detachedChild{}, fmt.Errorf("find own executable: %w", err) + } + + // The child is detached, so its output has nowhere to go but a file, and + // that file is the only place it can ever explain a failure. Opened + // O_APPEND, never O_TRUNC: --daemon-log is an operator-supplied path and + // this process does not get to destroy what is already in it. The offset + // below is what keeps a shared path from misattributing a failure. + logFile, err := os.OpenFile(cfg.DaemonLog, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return detachedChild{}, fmt.Errorf("open daemon log %s: %w", cfg.DaemonLog, err) + } + // The child inherits the descriptor at Start; this process does not need + // its own copy afterwards. + defer logFile.Close() + + var logOffset int64 + if info, err := logFile.Stat(); err == nil { + logOffset = info.Size() + } + + // The readiness pipe (§ReadyFDFlag): the child gets the write end as + // descriptor 3 and reports on it once it holds the log and is polling. + readyRead, readyWrite, err := os.Pipe() + if err != nil { + return detachedChild{}, fmt.Errorf("open readiness pipe: %w", err) + } + + cmd := exec.Command(exe, childArgs(cfg)...) + cmd.Stdin = nil // /dev/null + cmd.Stdout = logFile + cmd.Stderr = logFile + cmd.ExtraFiles = []*os.File{readyWrite} // descriptor 3 in the child + // The environment is how the connection details reach the child (§20.2): + // the token must never appear in argv, where it would land in the process + // table and in CI logs. + cmd.Env = os.Environ() + cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true} + + if err := cmd.Start(); err != nil { + readyRead.Close() + readyWrite.Close() + return detachedChild{}, fmt.Errorf("start recorder %s: %w", exe, err) + } + // Drop the parent's copy of the write end at once: with only the child + // holding it, a child that dies before signalling closes the pipe and the + // parent reads EOF instead of waiting out the whole timeout. + readyWrite.Close() + + return detachedChild{cmd: cmd, ready: readyRead, logOffset: logOffset}, nil +} + +// childArgs builds the child's command line. The rule set, every cadence and +// the recording's identity all come from the header the parent already wrote, +// and the connection details come from the environment — so what is left here +// is the log path plus the two run facts the header does not carry: the +// optional hard stop and the concurrency limit. +// +// Notably absent: --pidfile (the parent writes it, so check can find the pid +// the instant Watch returns), --alerts, --folder, --poll-interval, and +// anything derived from them. The CLI's `watch` FlagSet needs one flag of its +// own for this path — ReadyFDFlag — and dispatches to RunDaemonChild when it +// sees DaemonChildFlag. +func childArgs(cfg WatchConfig) []string { + args := []string{"watch", DaemonChildFlag, "--out", cfg.Out, ReadyFDFlag, strconv.Itoa(readyFD)} + if !cfg.Until.IsZero() { + args = append(args, "--until", cfg.Until.Format(time.RFC3339)) + } + if cfg.Concurrency > 0 { + args = append(args, "--concurrency", strconv.Itoa(cfg.Concurrency)) + } + return args +} From 07d2fb99d0148db19b03e1c054c0ec94a336a128 Mon Sep 17 00:00:00 2001 From: Bartek Tofel Date: Wed, 2 Sep 2026 11:44:28 +0200 Subject: [PATCH 2/2] chore: add a unit test, remove build tags --- .../internal/gate/{flock_unix.go => flock.go} | 6 ---- .../internal/gate/watch_daemon_test.go | 36 +++++++++++++++++-- .../gate/{watch_unix.go => watch_process.go} | 6 ---- 3 files changed, 34 insertions(+), 14 deletions(-) rename grafana-alertcheck/internal/gate/{flock_unix.go => flock.go} (67%) rename grafana-alertcheck/internal/gate/{watch_unix.go => watch_process.go} (94%) diff --git a/grafana-alertcheck/internal/gate/flock_unix.go b/grafana-alertcheck/internal/gate/flock.go similarity index 67% rename from grafana-alertcheck/internal/gate/flock_unix.go rename to grafana-alertcheck/internal/gate/flock.go index 7b927f315..9d7f7e750 100644 --- a/grafana-alertcheck/internal/gate/flock_unix.go +++ b/grafana-alertcheck/internal/gate/flock.go @@ -1,5 +1,3 @@ -//go:build unix - package gate import ( @@ -12,10 +10,6 @@ import ( // point (§8): a second writer must fail immediately with an error the operator // sees, not queue behind the first and start appending to a log somebody else // already finished. -// -// There is deliberately no Windows implementation — runners are Linux and -// goreleaser builds linux+darwin only (P6, P12) — so the package does not -// build there at all rather than silently skipping the lock. func lockExclusive(f *os.File) error { if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { return fmt.Errorf("flock: %w", err) diff --git a/grafana-alertcheck/internal/gate/watch_daemon_test.go b/grafana-alertcheck/internal/gate/watch_daemon_test.go index 8a044c7e9..35e9c424a 100644 --- a/grafana-alertcheck/internal/gate/watch_daemon_test.go +++ b/grafana-alertcheck/internal/gate/watch_daemon_test.go @@ -1,5 +1,3 @@ -//go:build unix - package gate import ( @@ -271,6 +269,40 @@ func TestWatchSpawnsADetachedRecorder(t *testing.T) { }) } +// TestDaemonChildRejectsAnAlreadyFinishedLog covers the RunDaemonChild guard +// against a reused --out path: a log that already carries a stopped sentinel is +// a finished recording, and a child starting against it would either append to +// a window already declared over, or take a flock over evidence that is about +// to be classified — so it must refuse before polling once. This is the +// fail-closed counterpart of §4.5 on the recorder's own startup path. +func TestDaemonChildRejectsAnAlreadyFinishedLog(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + clock := newFakeClock(testNow) + + w, err := NewWriter(path, clock) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + err = RunDaemonChild(context.Background(), DaemonChildConfig{ + URL: testHeader().URL, + Out: path, + Clock: clock, + }) + if err == nil { + t.Fatal("RunDaemonChild: no error against a log that already carries a stopped sentinel") + } + if !strings.Contains(err.Error(), "sentinel") { + t.Errorf("error = %v, want it to name the stopped sentinel", err) + } +} + // TestWatchFailsWhenTheChildCannotStartRecording is the other half of the // readiness contract. The child dies on its identity check, so it never reports // ready — and Watch must say so instead of returning success over a window diff --git a/grafana-alertcheck/internal/gate/watch_unix.go b/grafana-alertcheck/internal/gate/watch_process.go similarity index 94% rename from grafana-alertcheck/internal/gate/watch_unix.go rename to grafana-alertcheck/internal/gate/watch_process.go index 6d6096786..629606188 100644 --- a/grafana-alertcheck/internal/gate/watch_unix.go +++ b/grafana-alertcheck/internal/gate/watch_process.go @@ -1,5 +1,3 @@ -//go:build unix - package gate import ( @@ -30,10 +28,6 @@ type detachedChild struct { // group, so it would still take the terminal's signals and, on a runner, die // with the step that started it. Setsid gives it a new session AND a new // process group, which is what makes it survive to the end of the window. -// -// There is deliberately no Windows counterpart — runners are Linux and -// goreleaser builds linux+darwin only (P12) — so the package does not build -// there at all rather than silently recording in the foreground. func spawnChild(cfg WatchConfig) (detachedChild, error) { exe, err := os.Executable() if err != nil {