diff --git a/grafana-alertcheck/internal/gate/check.go b/grafana-alertcheck/internal/gate/check.go new file mode 100644 index 000000000..6c3bf8805 --- /dev/null +++ b/grafana-alertcheck/internal/gate/check.go @@ -0,0 +1,970 @@ +package gate + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "sort" + "strings" + "time" +) + +// Obligations this phase leaves for P10, carried forward the way P6 and P7 +// carried theirs so a later review has something concrete to check against: +// +// - Exit codes are the CLI's (§20.3, §9.1, H6/H7). Check returns +// (Result, error) and nothing else: err != nil is exit 2 unconditionally, +// never 0 and never 1, even alongside real violations; len(Violations) > 0 +// with err == nil is exit 1; both empty is exit 0. Check deliberately does +// not return a code, because a code is a presentation decision and the +// library must not make it. +// - §12.1 wants the paused rule AND --allow-paused both named to the user. +// decide names the rule in the shortfall Violation's Note (classify.go); +// the flag hint belongs to the CLI's renderer. +// - Config.Notes carries the running commentary (§13.2's planned run time, +// the countdown, the blind-interval warning). §20.2 puts the human output +// on stderr and reserves stdout for --output json, so the CLI must pass +// stderr here. +// - Check never reads the environment. GRAFANA_URL and GRAFANA_TOKEN are +// read by the CLI and passed in as fields, and the token must never reach +// a *flag.FlagSet (§20.2). + +// countdownEvery is how often the collection loop reports what it is waiting +// for (§13.2: "then print a countdown at regular intervals"). A silent wait is +// indistinguishable from a hung process, and the wait after `to` is the +// longest silence in the whole run. +const countdownEvery = 30 * time.Second + +// recorderStopTimeout bounds §4.4 step 3, the wait for the recorder's exit. +// Not in the source plan's table of values — a judgment call, on the same +// reasoning as childReadyTimeout (P6): everything the recorder does after +// SIGTERM is local (finish the in-flight write, append the sentinel, fsync) +// and an in-flight poll aborts through the child's own context, so the real +// figure is milliseconds. Loose enough for an overloaded runner, and a +// timeout is a hard error rather than a longer wait — a log a writer may +// still hold cannot be read at all (§4.4 step 4). +const recorderStopTimeout = 30 * time.Second + +// recorderStopPoll is how often that wait re-checks the pid. There is no +// wait(2) available: the recorder is a detached session leader, not this +// process's child (P6), so its exit can only be observed by polling. +const recorderStopPoll = 100 * time.Millisecond + +// Config is check's whole input. It is the CLI's view of a run, and it is +// deliberately wider than Policy: Policy is the narrowed, pure-layer subset +// that reaches decide (classify.go), and the token is the field that must +// never cross that line. +type Config struct { + // URL and Token are the connection details, read from the environment by + // the CLI and never registered as flags (§20.2). Token never enters the + // pure layer, an error string, or a Result. + URL, Token string + + // Alerts is REQUIRED in single-step mode and must be EMPTY in log mode: + // with a log, the header IS the alert set (§19.1 step 3), and there is + // nothing to compare a second list against. Both directions are encoded, + // resolving the source plan's §19.1 step 1 / step 3 contradiction. + Alerts []string + Folder string + + States []State + Preexisting PreexistingPolicy + MinObserved int + AllowPaused bool + NodataIsUnobservable bool + + // From is the moment the deploy finished and To is the end of the work + // (§7). They are different moments and both come from the work. In + // recorder mode an absent From is a hard error; in single-step mode it + // falls back to the start of this step, with the blind-interval warning + // §4.2 requires. + From, To time.Time + + // Log is the path of a recording made by watch; "" selects single-step + // mode. PidFile defaults to .pid, the convention watch's parent + // writes (P6) and the only way check can reach the recorder it must stop + // before it may read the log (§4.4 steps 1-4). + Log string + PidFile string + + // There is deliberately NO PollEvery here, and `check` has no + // --poll-interval flag. In log mode the cadence comes from the header — + // the cadence the recording actually used — and a second authority would + // let an operator silently widen maxGap over evidence that was recorded at + // a different rate (P5, "two authorities"); in single-step mode the same + // process records and classifies, so §5's default is the only cadence + // there is. + Concurrency int + Clock Clock + + // Notes is where the shell prints what an operator has to see while the + // run is in progress: the planned run time, the grace and its source, the + // countdown, the blind-interval warning. nil discards them. The library + // renders no table — the CLI owns presentation (§20.2). + Notes io.Writer +} + +func (cfg Config) withDefaults() Config { + 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.Log != "" { + cfg.PidFile = cfg.Log + ".pid" + } + return cfg +} + +// namedAlerts returns the alert names that survive §17.3's trim-and-discard, +// so validation counts what Resolve will actually see rather than what the +// caller happened to pass (a file ending in a newline yields an empty line). +func (cfg Config) namedAlerts() []string { + out := make([]string, 0, len(cfg.Alerts)) + for _, a := range cfg.Alerts { + if strings.TrimSpace(a) != "" { + out = append(out, a) + } + } + return out +} + +// Check is the I/O shell: HTTP, signals, the pidfile, file reads, the +// countdown print. Every correctness question it touches is answered +// elsewhere — by proveCoverage and decide, which are pure — and that split is +// the most important seam in the project (§2). Check therefore needs two +// integration tests; decide carries the suite. +// +// H7 governs the return: a pass is exactly len(Violations) == 0 && err == nil. +// Every error path below leaves err non-nil, and no path anywhere in this file +// converts an error into an empty Result with a nil error. +func Check(ctx context.Context, cfg Config) (Result, error) { + cfg = cfg.withDefaults() + if err := cfg.validate(); err != nil { + return Result{}, err + } + // The Source is built here and injected into check() so every behaviour + // below is testable against a scripted fake — the same seam prepareWatch + // uses (P6), and the reason this file needs no test-only setter. + return check(ctx, cfg, NewHTTPSource(cfg.URL, cfg.Token, cfg.Clock)) +} + +// validate is §19.1 step 1. It runs before any network call, so a +// configuration mistake costs nothing and, more importantly, is never +// discovered after a ten-minute wait. +func (cfg Config) validate() error { + if cfg.URL == "" { + return errors.New("check: no grafana url") + } + if cfg.To.IsZero() { + return errors.New("check: no `to`: the end of the window is required (§7)") + } + + named := cfg.namedAlerts() + if cfg.Log == "" { + // §19.1 step 1: an empty Alerts is an error — but only without a log. + if len(named) == 0 { + return errors.New("check: no alert names given and no recorded log to take them from") + } + } else if len(named) > 0 { + // §19.1 step 3, the other direction: the alert set comes from the log. + // Accepting both would mean reconciling two sets, which is the subset + // arithmetic the source plan removes by making the log the one source. + return fmt.Errorf("check: --alerts is refused with a recorded log: %s already names the alert set it recorded (§19.1 step 3)", cfg.Log) + } + + now := cfg.Clock.Now() + // from mirrors what check() will use, so the two window checks below judge + // the window that will really be classified. The fallback is not written + // back into cfg: check() re-reads the clock at the same point, and one + // authority for that value is better than two that could disagree. + from := cfg.From + switch { + case from.IsZero() && cfg.Log != "": + // §7, and never a warning-and-continue: falling back to the start of + // the check step reinstates exactly the blind interval the recorder + // exists to remove, which is the fail-open shape this design refuses. + return errors.New("check: no `from` in recorder mode: the deploy step must emit a completion timestamp (§7)") + case from.IsZero(): + // Single-step only. The caller sees the resulting blind interval named + // exactly, once the first observation has fixed its end (§4.2). + from = now + } + + if cfg.To.Before(from) { + return fmt.Errorf("check: `to` %s is before `from` %s", cfg.To.Format(time.RFC3339), from.Format(time.RFC3339)) + } + if from.After(now.Add(fromFutureTolerance)) { + return fmt.Errorf("check: `from` %s is more than %s ahead of this runner's clock %s (§7)", + from.Format(time.RFC3339), fromFutureTolerance, now.Format(time.RFC3339)) + } + + // A `to` already in the past is not a special mode WITH a log (§7, §24.3): + // the collection loop's condition is simply already true and the evidence + // is classified immediately. Without one it is a different thing entirely + // — a request to prove a window that nothing observed. Refusing it is not + // pedantry: the coverage window would end before the first observation, + // every heartbeat gap inside it would measure negative, and the run would + // report a proved window it never saw. + if cfg.Log == "" && !cfg.To.After(now) { + return fmt.Errorf("check: `to` %s has already passed and there is no recorded log: a window that ended before check started can only be classified from a recording (§4.2)", + cfg.To.Format(time.RFC3339)) + } + return nil +} + +// check is Check with the Source injected. Its body is §19.1 steps 1-9, one +// commented block each and in that order, so a review can diff it against the +// source plan line by line. +func check(ctx context.Context, cfg Config, src Source) (Result, error) { + // ---- §19.1 step 1 — validate the configuration. ----------------------- + // Done by Check before this function is reached, except for the one part + // that needs a clock reading kept for later: the single-step fallback for + // an absent `from`. + from := cfg.From + if from.IsZero() { + from = cfg.Clock.Now() + fmt.Fprintf(cfg.Notes, "note: no `from` given; the window starts at the start of this step, %s (§4.2)\n", + from.Format(time.RFC3339)) + } + + // ---- §19.1 step 2 — resolve the definitions from the ruler API. ------- + // Unconditional, in BOTH modes. A log's header supplies the alert set as + // UIDs and the recording facts, never the rule facts: `for`, + // intervalSeconds and Kind always come from a fresh ruler read, which is + // why LoggedRule.ForSeconds is never converted back into a Definition. + version, err := src.Version(ctx) + if err != nil { + return Result{}, fmt.Errorf("read grafana version: %w", err) + } + if err := CheckGrafanaVersion(version); err != nil { + return Result{}, err + } + allDefs, err := src.Definitions(ctx) + if err != nil { + // §19.3 case 2: resolution of the definitions failed. + return Result{}, fmt.Errorf("read rule definitions: %w", err) + } + + // ---- §19.1 step 3 — with a log, validate its identity. ---------------- + // The header is read early — line 1 only, the one line a writer can never + // change (ReadLogHeader) — so a wrong URL or a rule that no longer + // resolves fails closed NOW rather than after the whole window has + // elapsed. It is advisory: the authoritative header comes from the single + // full ReadLog in step 6, after the writer has exited, and the identity is + // validated again against that one. + var ( + resolved []Definition + notes []string + earlyHdr Header + logHasHdr bool + rt map[string]ruleTimings + gt globalTimings + timingNote []string + ) + if cfg.Log != "" { + earlyHdr, err = ReadLogHeader(cfg.Log) + if err != nil { + // §19.3 case 3. + return Result{}, fmt.Errorf("log identity: %w", err) + } + logHasHdr = true + resolved, notes, err = resolveFromLog(allDefs, earlyHdr, cfg) + } else { + resolved, notes, err = Resolve(allDefs, cfg.namedAlerts(), cfg.Folder) + } + if err != nil { + return Result{}, err + } + if len(resolved) == 0 { + // Reachable only from a header with an empty rule list. Left to run, + // MinObserved would default to zero, no rule would be judged, and the + // gate would return a pass over nothing at all. + return Result{}, fmt.Errorf("check: no rules to classify") + } + for _, n := range notes { + fmt.Fprintf(cfg.Notes, "note: %s\n", n) + } + + // ---- §19.1 step 4 — derive the timings, print the plan, fit the budget. + if logHasHdr { + // The header is the authority for the cadence actually recorded at; + // re-deriving it from defs would compare gaps recorded at an override + // cadence against thresholds computed from the default — fail-open in + // the faster-override direction (P5). + rt, gt, err = DeriveTimingsFromLog(earlyHdr, resolved) + if err != nil { + return Result{}, fmt.Errorf("log identity: %w", err) + } + } else { + rt, gt, timingNote = DeriveTimings(resolved, 0) + for _, n := range timingNote { + fmt.Fprintf(cfg.Notes, "note: %s\n", n) + } + } + summary, warning := StartupSummary(from, cfg.To, gt) + fmt.Fprintln(cfg.Notes, summary) + if warning != "" { + fmt.Fprintf(cfg.Notes, "warning: %s\n", warning) + } + + // The measurement pass and the budget check belong to single-step mode + // alone (§5.2): in recorder mode watch already took one observation of + // every rule and checked the budget against those measured latencies + // before it detached, and repeating it here would spend a second poll of + // every rule to re-answer a question already answered. + var ( + header Header + initial []Poll + reducer = NewReducer() + ) + if !logHasHdr { + // StartedAt is fixed before the pass rather than after it, so the + // interval it claims to have observed can only be wider than the one + // it really saw — and the first heartbeat's own boundary gap (P7 check + // 3) is what proves that interval, not this timestamp. + startedAt := cfg.Clock.Now() + active := activeRules(resolved) + var measured map[string]time.Duration + initial, measured, err = firstObservations(ctx, src, active, reducer, cfg.Concurrency, cfg.Notes) + if err != nil { + return Result{}, err + } + if err := CheckBudget(activeTimingsOf(active, rt), measured, cfg.Concurrency); err != nil { + return Result{}, err + } + + // Single-step synthesis — how the pure layer stays unconditional (§2). + // The shell builds the Header and later stamps the sentinel itself, so + // P7 checks 1 and 2 run exactly as they do over a recording and no + // mode flag ever reaches proveCoverage or decide. + header = Header{ + SchemaVersion: LogSchemaVersion, + URL: cfg.URL, + GrafanaVersion: version, + StartedAt: startedAt, + Rules: loggedRules(resolved, rt), + } + if from.Before(startedAt) { + // §4.2/§22.4's declared blind interval: in single-step mode this + // is a warning and a pass, and ONLY here. Recorder mode keeps P7 + // check 2 strict (§22.9), because there the recorder was supposed + // to be watching and the gap means it was not. + fmt.Fprintf(cfg.Notes, "warning: cannot see [%s, %s) — %s before the first observation; the window is classified from %s (§4.2)\n", + from.Format(time.RFC3339), startedAt.Format(time.RFC3339), + startedAt.Sub(from).Round(time.Second), startedAt.Format(time.RFC3339)) + from = startedAt + } + } + + // ---- §19.1 step 5 — apply MinObserved. -------------------------------- + // Its default is the resolved rule count AFTER the collapse (§17.3), which + // is len(resolved) by construction. decide defaults it identically; it is + // resolved here as well so the value the run will judge against is printed + // before the wait rather than inferred from the verdict afterwards. + minObserved := cfg.MinObserved + if minObserved == 0 { + minObserved = len(resolved) + } + fmt.Fprintf(cfg.Notes, "min-observed: %d of %d resolved rule(s)\n", minObserved, len(resolved)) + + // ---- §19.1 step 6 — collect the evidence. ----------------------------- + // Collect ONLY. No classification happens here and there is no early exit, + // even once a violation is certain (H5, §19.2): the loop always runs to + // to + transitionGrace, which is what makes "did the early exit lose the + // coverage proof?" a question that cannot be asked. + windowEnd := cfg.To.Add(gt.transitionGrace) + + var poller *livePoller + if !logHasHdr { + poller = newLivePoller(src, reducer, activeRules(resolved), rt, cfg.Concurrency, cfg.Clock.Now()) + } + collected, err := collectUntil(ctx, cfg, windowEnd, poller) + if err != nil { + // §19.3 case 1: the failure limit was exceeded (retryTransport already + // gave every transient failure its backoff), or the context ended. + // Nothing collected is classified — the count is there so an operator + // can tell a run that failed at once from one that failed at minute + // nine. + return Result{}, fmt.Errorf("collect evidence after %d poll(s): %w", len(collected), err) + } + + var ( + polls []Poll + sentinel *time.Time + ) + if logHasHdr { + // §4.4 steps 2-4, in this order and no other: signal the writer, wait + // for its exit, and only THEN read the log once. A log read while a + // writer can still append can only yield a shorter window than the one + // that was actually recorded. + heldLog, err := stopRecorder(ctx, cfg) + if err != nil { + return Result{}, err + } + header, polls, sentinel, err = ReadLog(cfg.Log) + // The lock stays held across the read, so no writer can appear between + // the proof that there was none and the read itself. Released here + // rather than deferred: everything past this point works from bytes + // already in memory, and the drain wait below can take minutes. + _ = heldLog.Close() + if err != nil { + return Result{}, err + } + // The authoritative header, validated the same way the advisory one + // was — and its result is KEPT. Everything from here on judges the + // header ReadLog returned, so nothing downstream rests on the advisory + // read having been right. That read is what it claims to be: a + // fail-fast, and no part of the verdict depends on it. + resolved, _, err = resolveFromLog(allDefs, header, cfg) + if err != nil { + return Result{}, err + } + // rt is re-derived because it depends on the header: PollEverySeconds + // is the one load-bearing value the advisory read supplied. windowEnd + // is deliberately NOT recomputed from the gt this returns: the + // collection loop has already stopped at the earlier value, and moving + // the end of the window afterwards would prove a window this run did + // not collect. + if rt, gt, err = DeriveTimingsFromLog(header, resolved); err != nil { + return Result{}, fmt.Errorf("log identity: %w", err) + } + } else { + polls = make([]Poll, 0, len(initial)+len(collected)) + polls = append(polls, initial...) + polls = append(polls, collected...) + // The shell stamps the sentinel itself, when the collection loop + // exits: by construction that is at or after to + transitionGrace, so + // P7 check 1 passes for the same reason a clean recorder stop does, + // and for no other. + stoppedAt := cfg.Clock.Now() + sentinel = &stoppedAt + } + + // ---- §19.1 step 7 — the drain wait. ----------------------------------- + // The last instance of the liveness check (§14.6): did this rule evaluate + // through the end of the window? It is I/O and it is deliberately NOT part + // of proveCoverage — adding it there would put HTTP inside the pure layer + // and destroy the seam §2 depends on. + drained, err := drainWait(ctx, cfg, src, resolved, header.pausedAtStart(), rt, polls, windowEnd, gt.drainTimeout) + if err != nil { + return Result{}, err + } + + // ---- §19.1 step 8 — classify. ----------------------------------------- + pol := Policy{ + States: cfg.States, + Preexisting: cfg.Preexisting, + MinObserved: minObserved, + AllowPaused: cfg.AllowPaused, + NodataIsUnobservable: cfg.NodataIsUnobservable, + From: from, + To: cfg.To, + } + result, decideErr := decide(header, polls, sentinel, resolved, rt, gt, pol) + result, drainErr := mergeDrainTimeouts(result, drained) + + // ---- §19.1 step 9 — return Result. ------------------------------------ + // Both errors are joined rather than one shadowing the other: each names + // rules the other does not, and on exit 2 that list IS the answer to + // "why". H7 needs only that err be non-nil when either fired. + return result, errors.Join(decideErr, drainErr) +} + +// resolveFromLog turns a log header into the resolved definitions, and is +// §19.1 step 3's identity check in practice. Three things are verified: the +// URL matches, the schema version matches (ReadLog/ReadLogHeader own that), +// and every header UID still resolves against the fresh ruler read. The alert +// set is TAKEN from the log, never compared — with Alerts required empty in +// log mode there is nothing to compare it against, and §22.4's "different +// alert set" refusal is exactly this URL-and-UID failure. +// +// Resolving through Resolve, by uid:, rather than by a private lookup, keeps +// one implementation of §17: a header naming a recording or datasource-managed +// rule gets the same specific refusal an operator would, and a header naming +// the same UID twice collapses with a note (DeriveTimingsFromLog rejects that +// case outright, so the note is belt and braces). +// +// Only the header-to-defs direction needs checking. The opposite direction +// cannot fail here: resolved is BUILT from the header, so no resolved +// definition can be absent from it. +func resolveFromLog(allDefs []Definition, h Header, cfg Config) ([]Definition, []string, error) { + if h.URL != cfg.URL { + return nil, nil, fmt.Errorf("log identity: %s recorded url %q but this run is configured for %q (§19.1 step 3)", + cfg.Log, h.URL, cfg.URL) + } + names := make([]string, 0, len(h.Rules)) + for _, lr := range h.Rules { + names = append(names, "uid:"+lr.UID) + } + resolved, notes, err := Resolve(allDefs, names, "") + if err != nil { + return nil, nil, fmt.Errorf("log identity: %s names a rule that no longer resolves: %w (§19.1 step 3)", cfg.Log, err) + } + return resolved, notes, nil +} + +// activeRules drops the rules whose DEFINITION says paused. They are skipped +// (§12): never polled, never waited for, and reported from the definitions +// alone — a skipped rule has no poll records at all, so it has no heartbeats +// to prove and no IsPaused poll to detect (P6 deviation 4, and the obligation +// it left P7/P8). +func activeRules(defs []Definition) []Definition { + out := make([]Definition, 0, len(defs)) + for _, d := range defs { + if !d.IsPaused { + out = append(out, d) + } + } + return out +} + +// activeTimingsOf narrows the timings map to the rules that will actually be +// polled, which is what §5.2's budget is spent on: a skipped rule consumes +// none of the capacity, so counting it would refuse schedules that fit. +func activeTimingsOf(active []Definition, rt map[string]ruleTimings) map[string]ruleTimings { + out := make(map[string]ruleTimings, len(active)) + for _, d := range active { + out[d.UID] = rt[d.UID] + } + return out +} + +// livePoller is single-step mode's collection engine: the same per-rule +// scheduler and the same Reducer the recorder uses (P4/P6), writing into +// memory instead of a log. Log mode has none — the recorder is doing this +// work in another process — and collectUntil takes a nil poller for it. +type livePoller struct { + src Source + reducer *Reducer + sched *Scheduler + titles map[string]string // uid -> title: poll by title, select by UID (§14.5) + concurrency int +} + +func newLivePoller(src Source, reducer *Reducer, active []Definition, rt map[string]ruleTimings, + concurrency int, now time.Time) *livePoller { + + titles := make(map[string]string, len(active)) + cadence := make(map[string]time.Duration, len(active)) + for _, d := range active { + titles[d.UID] = d.Title + cadence[d.UID] = rt[d.UID].pollEvery + } + return &livePoller{ + src: src, + reducer: reducer, + sched: NewScheduler(cadence, now), + titles: titles, + concurrency: concurrency, + } +} + +// poll runs one round of due rules and returns every poll that succeeded, +// alongside the first failure. +// +// The successes are NOT kept for the reason watchLoopConfig.pollBatch keeps +// its own: those go into a durable log that a later check will read, so +// dropping one would turn a single rule's transport failure into a coverage +// gap for the others. Here there is no later reader. A terminal failure +// during collection is exit 2 (§19.3 case 1) and check discards the whole +// collection, so these come back only to let the error say how far the run +// got before it stopped — which is the one part of it an operator can act on. +func (p *livePoller) poll(ctx context.Context, uids []string) ([]Poll, error) { + observed, obsErr := observeAll(ctx, p.src, p.titles, uids, p.concurrency) + out := make([]Poll, 0, len(uids)) + for _, uid := range uids { + obs, ok := observed[uid] + if !ok { + continue + } + out = append(out, p.reducer.Reduce(uid, obs)) + } + return out, obsErr +} + +// collectUntil is §19.1 step 6's loop, shared by both modes. With a poller it +// polls each rule on its own cadence; with nil it only waits, because in +// recorder mode the evidence is being written by another process. Both print +// the same countdown, because both are the same silence to an operator +// watching a job (§13.2). +// +// It never classifies and never exits early (H5). +func collectUntil(ctx context.Context, cfg Config, deadline time.Time, p *livePoller) ([]Poll, error) { + var ( + polls []Poll + lastPrint time.Time + ) + for { + now := cfg.Clock.Now() + if !now.Before(deadline) { + return polls, nil + } + if lastPrint.IsZero() || now.Sub(lastPrint) >= countdownEvery { + fmt.Fprintf(cfg.Notes, "collecting: %s until the window closes at %s\n", + deadline.Sub(now).Round(time.Second), deadline.Format(time.RFC3339)) + lastPrint = now + } + + wait := min(deadline.Sub(now), countdownEvery) + if p != nil { + // Mark before polling, against the batch's own `now`: the next poll + // is one cadence after this one was DUE, not after it returned, so + // request latency cannot make the heartbeat spacing drift towards + // maxGap (the same rule watchLoop follows). + due := p.sched.Due(now) + for _, uid := range due { + p.sched.Mark(uid, now) + } + if len(due) > 0 { + batch, err := p.poll(ctx, due) + polls = append(polls, batch...) + if err != nil { + return polls, err + } + } + if next, ok := p.sched.earliestDue(); ok { + wait = min(wait, next.Sub(cfg.Clock.Now())) + } + } + + select { + case <-ctx.Done(): + return polls, ctx.Err() + case <-cfg.Clock.After(max(wait, 0)): + } + } +} + +// stopRecorder is §4.4 steps 2 and 3. Nothing here is best-effort: the log may +// not be read until the writer has provably gone, so every failure to reach +// that state is a hard error. +// +// It returns the log held under an exclusive flock. The caller must keep that +// file open across ReadLog and close it afterwards — the lock is the proof +// that no writer exists, and holding it across the read also shuts out a new +// one appearing between the proof and the read. +// +// Two authorities, and only one of them is evidence: +// +// - The PIDFILE says whether a recording was ever started. An absent or +// unparseable one is the load-bearing case, and P6's obligation on this +// phase: it must never read as "there was nothing to stop". The parent +// writes the pidfile only AFTER the child reports that it holds the log +// and is polling, and removes it on every failing path, so a missing one +// means watch failed and this run has no evidence at all. +// - The FLOCK says whether a writer exists RIGHT NOW. Nothing removes the +// pidfile when a recorder exits cleanly — the parent has long returned and +// the child never learns the path — so after a --until run, a supported +// flow, the pidfile names a pid nobody owns. Signalling it would SIGTERM +// whatever same-user process inherited that pid. The kernel releases a +// flock when its holder exits, crash included, so the lock cannot go +// stale that way. +// +// So: read the pidfile to learn that a recording happened, then ask the lock +// whether it is still running, and signal only if it is. +func stopRecorder(ctx context.Context, cfg Config) (*os.File, error) { + pid, err := ReadPidFile(cfg.PidFile) + if err != nil { + return nil, fmt.Errorf("cannot stop the recorder: %w; a pidfile is written only once a recorder reports that it is running, so an unreadable one means the recording never started (§4.4)", err) + } + + log, err := os.Open(cfg.Log) + if err != nil { + return nil, fmt.Errorf("open %s to check for a writer: %w", cfg.Log, err) + } + + held, err := tryLockExclusive(log) + if err != nil { + log.Close() + return nil, err + } + if held { + // No writer. Send no signal, whatever the pidfile says — the pid may + // belong to somebody else entirely by now. Which of --until, a clean + // stop and a death ended the recording is the sentinel's question, + // answered by P7 check 1 over the log this unblocks. + fmt.Fprintf(cfg.Notes, "note: no writer holds %s; the recorder has already finished\n", cfg.Log) + return log, nil + } + + // The lock is held, so a writer is alive and the pidfile's pid cannot be + // stale — the recorder that took the lock is the one the parent recorded. + gone, err := signalRecorder(pid) + if err != nil { + log.Close() + return nil, err + } + if gone { + // A live writer holds the log and the pidfile names a process that + // does not exist. That is a broken contract, not a case to reason + // around: signalling the real holder would mean guessing who it is. + log.Close() + return nil, fmt.Errorf("a writer holds %s but pidfile %s names pid %d, which does not exist: the pidfile does not name the process that holds the log", + cfg.Log, cfg.PidFile, pid) + } + + // Wait on the LOCK, not on the pid: its release is the kernel-guaranteed + // writer-is-gone event, and it carries no pid-reuse hazard. + deadline := cfg.Clock.Now().Add(recorderStopTimeout) + for { + select { + case <-ctx.Done(): + log.Close() + return nil, ctx.Err() + case <-cfg.Clock.After(recorderStopPoll): + } + + held, err := tryLockExclusive(log) + if err != nil { + log.Close() + return nil, err + } + if held { + return log, nil + } + if !cfg.Clock.Now().Before(deadline) { + log.Close() + return nil, fmt.Errorf("recorder pid %d still holds %s %s after SIGTERM; refusing to read a log a writer can still append to (§4.4 step 4)", + pid, cfg.Log, recorderStopTimeout) + } + } +} + +// drainVerdict is what the drain wait concluded about one rule it could not +// clear. It carries the reason as well as the prose because the two outcomes +// are genuinely different faults: drain_timeout means the rule is still there +// and still behind, rule_absent means it is gone. Collapsing both into +// drain_timeout would name the wait instead of the fault, and Reason is a +// published vocabulary that reaches the action's JSON (§19.0). +type drainVerdict struct { + reason UnobservableReason + note string +} + +// drainWait is §19.1 step 7 and §14.6: the final instance of the liveness +// check, asking each rule the last question — did you evaluate through the end +// of the window? A rule that cannot answer within drainTimeout is +// unobservable, never a pass. +// +// It returns one verdict per rule it could not clear, keyed by UID, which the +// caller folds into the Result. It returns an error only for a hard failure of +// the wait itself (§19.3 case 1); a rule that simply never catches up is +// reported, not raised. +// +// Two rules are excluded from the wait before it starts, and both are +// exclusions of work that could not change a verdict: +// +// - a rule the HEADER says was already paused when the recording opened +// (§12): it is skipped, it was not evaluating, and it never was — there is +// no evaluation to wait for. The header and not the definition, for +// decide's reason (Header.pausedAtStart): a rule the header says was +// active must be drained or faulted, because a pause somebody applied +// after the window is not evidence about the window; +// - a rule whose last poll says Found == false: P7 check 8 already makes it +// unobservable, so the only thing draining it could add is drainTimeout of +// waiting before the same answer. +func drainWait(ctx context.Context, cfg Config, src Source, defs []Definition, pausedAtStart map[string]bool, + rt map[string]ruleTimings, polls []Poll, windowEnd time.Time, timeout time.Duration) (map[string]drainVerdict, error) { + + pending := make(map[string]string) // uid -> title, the shape observeAll wants + for _, d := range defs { + if pausedAtStart[d.UID] { + continue + } + rulePolls := pollsForRule(polls, d.UID) + if n := len(rulePolls); n > 0 && !rulePolls[n-1].Found { + continue + } + // Evidence already in hand can satisfy the wait outright: a rule whose + // recorded evaluations already reach past the end of the window has + // answered the question, and polling it again asks nothing new. + if !anyPollEvaluatedThrough(rulePolls, windowEnd) { + pending[d.UID] = d.Title + } + } + if len(pending) == 0 { + return nil, nil + } + + fmt.Fprintf(cfg.Notes, "drain wait: %d rule(s) have not yet evaluated through %s (limit %s)\n", + len(pending), windowEnd.Format(time.RFC3339), timeout) + + deadline := cfg.Clock.Now().Add(timeout) + verdicts := make(map[string]drainVerdict) + for { + uids := make([]string, 0, len(pending)) + for uid := range pending { + uids = append(uids, uid) + } + sort.Strings(uids) // deterministic request order and message order + + observed, err := observeAll(ctx, src, pending, uids, cfg.Concurrency) + if err != nil { + return nil, fmt.Errorf("drain wait: %w", err) + } + for _, uid := range uids { + obs, ok := observed[uid] + if !ok { + continue + } + rule := stateRuleByUID(obs.Rules, uid) + if rule == nil { + // §14.5: a 2xx that parsed and carries no matching rule is an + // authoritative "the rule is gone" — P2 retried every transport + // failure long before this Observation existed. It is knowable + // on the FIRST poll, so waiting the rest of drainTimeout would + // spend two minutes to reach the same verdict under a name that + // describes the wait rather than the fault. + verdicts[uid] = drainVerdict{ + reason: ReasonRuleAbsent, + note: fmt.Sprintf("rule %q: absent from the state endpoint during the drain wait; there is no evaluation to wait for (§14.5)", + pending[uid]), + } + delete(pending, uid) + continue + } + if rule.IsPaused { + // A paused rule does not evaluate, so this one can never catch + // up and the rest of drainTimeout would buy nothing. The reason + // stays drain_timeout: UnobservableReason is a published + // vocabulary that reaches the action's JSON (§19.0), and the + // prose below is where the detail belongs. + verdicts[uid] = drainVerdict{ + reason: ReasonDrainTimeout, + note: fmt.Sprintf("rule %q: paused before it evaluated through %s, so it never will (§14.8)", + pending[uid], windowEnd.Format(time.RFC3339)), + } + delete(pending, uid) + continue + } + if evaluatedThrough(rule.LastEvaluation, obs.Skew, obs.SkewBound, windowEnd) { + delete(pending, uid) + } + } + if len(pending) == 0 { + return verdicts, nil + } + + now := cfg.Clock.Now() + if !now.Before(deadline) { + for uid, title := range pending { + verdicts[uid] = drainVerdict{ + reason: ReasonDrainTimeout, + note: fmt.Sprintf("rule %q: did not evaluate through %s within the %s drain limit (§19.1 step 7)", + title, windowEnd.Format(time.RFC3339), timeout), + } + } + return verdicts, nil + } + + // Re-ask no faster than the tightest cadence among the rules still + // pending: a rule evaluating every 60s cannot answer differently 200ms + // later, and hammering it would spend the request budget §5 accounts + // for on nothing. + wait := deadline.Sub(now) + for uid := range pending { + if every := rt[uid].pollEvery; every > 0 { + wait = min(wait, every) + } + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-cfg.Clock.After(max(wait, 0)): + } + } +} + +// anyPollEvaluatedThrough reports whether any recorded poll of a rule already +// proves it evaluated through windowEnd. +func anyPollEvaluatedThrough(polls []Poll, windowEnd time.Time) bool { + for _, p := range polls { + if !p.Found { + continue + } + if evaluatedThrough(p.LastEvaluation, p.Skew(), p.SkewBound(), windowEnd) { + return true + } + } + return false +} + +// evaluatedThrough is the drain wait's one comparison, and it is cross-domain +// (§16): lastEvaluation is a Grafana timestamp and windowEnd is runner-domain, +// so the Grafana value is translated by its own poll's skew. The skew BOUND is +// then subtracted rather than added — the pessimistic end of the uncertainty — +// so an evaluation that only might have reached the end of the window does not +// count as one that did. Understating it costs a few more seconds of waiting; +// overstating it would pass an unproven window. +// +// A zero lastEvaluation never satisfies the wait: only a paused rule may +// legitimately report it (§2.3), and a paused rule has nothing to drain. +func evaluatedThrough(lastEval time.Time, skew, bound time.Duration, windowEnd time.Time) bool { + if lastEval.IsZero() { + return false + } + return !lastEval.Add(-skew).Add(-bound).Before(windowEnd) +} + +// mergeDrainTimeouts folds the I/O drain wait's verdicts into the pure layer's +// Result. P7 places this merge "before decide runs"; it cannot be, because +// decide owns proveCoverage and therefore builds the Coverage map itself — so +// the merge happens immediately after, which is the same thing from every +// caller's point of view and keeps decide's signature a pure function of its +// arguments. +// +// It returns its own error rather than mutating decide's, so neither hides the +// other: a run with one rule unobservable from the coverage proof and another +// from the drain wait must name both (H6 — inability beats violation, and it +// beats a second inability being dropped from the message too). The error says +// "at the drain wait" for that reason: the two are joined into one message, and +// two counts under one identical phrase read as a contradiction rather than as +// two findings. +func mergeDrainTimeouts(res Result, drained map[string]drainVerdict) (Result, error) { + if len(drained) == 0 { + return res, nil + } + if res.Coverage == nil { + res.Coverage = make(map[string]CoverageResult, len(drained)) + } + + var names []string + for i := range res.Verdicts { + uid := res.Verdicts[i].RuleUID + verdict, ok := drained[uid] + if !ok { + continue + } + cov := res.Coverage[uid] + cov.Unobservable = true + cov.Proved = false + if cov.Reason == "" { + // The FIRST reason wins, as it does inside proveCoverage: a rule + // the coverage proof already faulted keeps the fault it was + // actually caught by. + cov.Reason = verdict.reason + } + cov.Notes = append(cov.Notes, verdict.note) + res.Coverage[uid] = cov + + if res.Verdicts[i].Outcome != OutcomeUnobservable { + names = append(names, fmt.Sprintf("%s (%s)", res.Verdicts[i].Alert, verdict.reason)) + } + res.Verdicts[i].Outcome = OutcomeUnobservable + res.Verdicts[i].Note = strings.Join(cov.Notes, "; ") + } + if len(names) == 0 { + // Every drained rule was already unobservable for an earlier reason, + // so decide's own error already stops the run. Adding a second error + // saying the same thing would only make the message longer. + return res, nil + } + return res, fmt.Errorf("gate: %d rule(s) unobservable at the drain wait: %s", len(names), strings.Join(names, "; ")) +} diff --git a/grafana-alertcheck/internal/gate/check_process.go b/grafana-alertcheck/internal/gate/check_process.go new file mode 100644 index 000000000..580a2a143 --- /dev/null +++ b/grafana-alertcheck/internal/gate/check_process.go @@ -0,0 +1,29 @@ +package gate + +import ( + "errors" + "fmt" + "syscall" +) + +// signalRecorder asks the recorder to stop (§4.4 step 2). +// +// The caller must have established that a writer is alive — by taking the +// log's flock and being refused — before it calls this. Nothing removes the +// pidfile when a recorder exits cleanly, so a pid read without that proof can +// name any same-user process that has since inherited it. +// +// gone reports ESRCH. Given the lock proof, that is a broken contract rather +// than a clean stop, and stopRecorder treats it as one; the value is reported +// instead of raised here because this function knows the errno and not what +// it means. +func signalRecorder(pid int) (gone bool, err error) { + switch err := syscall.Kill(pid, syscall.SIGTERM); { + case err == nil: + return false, nil + case errors.Is(err, syscall.ESRCH): + return true, nil + default: + return false, fmt.Errorf("signal recorder pid %d: %w", pid, err) + } +} diff --git a/grafana-alertcheck/internal/gate/check_test.go b/grafana-alertcheck/internal/gate/check_test.go new file mode 100644 index 000000000..2971c30e4 --- /dev/null +++ b/grafana-alertcheck/internal/gate/check_test.go @@ -0,0 +1,1216 @@ +package gate + +import ( + "bufio" + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "syscall" + "testing" + "time" +) + +// The one rule every test in this file watches, unless it says otherwise: a +// 60s evaluation interval, no `for`, not paused. Every derived value follows +// from those three numbers, and the tests assert against them by name rather +// than by magic constant: +// +// pollEvery 30s (§5: intervalSeconds/2) +// maxGap 60s (2 x pollEvery) +// healthGrace 60s (max(maxGap, interval)) +// evalStaleAfter 120s (2 x interval) +// transitionGrace 60s (for + interval) +// drainTimeout 2m (max(2 x interval, 2m)) +const ( + checkUID = "rule-one" + checkTitle = "Rule One" + + checkPollEvery = 30 * time.Second + checkGrace = 60 * time.Second + checkDrainLimit = 2 * time.Minute +) + +func checkDef() Definition { + return Definition{ + UID: checkUID, Title: checkTitle, Folder: "F", Group: "G", + IntervalSeconds: 60, NoDataState: "OK", ExecErrState: "OK", + Kind: KindGrafanaManaged, + } +} + +// checkSource is a Source whose state answers depend on virtual time and on +// the call count, which is what a collection-loop test needs: fakeSource's +// static script cannot express "healthy for the whole window" without +// scripting every poll, and loopSource (watch_test.go) deliberately refuses +// Version and Definitions because the recorder's child never reads them. +type checkSource struct { + mu sync.Mutex + + version string + versionErr error + defs []Definition + defsErr error + + calls map[string]int + respond func(title string, call int) (Observation, error) +} + +func newCheckSource(respond func(title string, call int) (Observation, error)) *checkSource { + return &checkSource{ + version: "13.1.0", + defs: []Definition{checkDef()}, + calls: map[string]int{}, + respond: respond, + } +} + +func (s *checkSource) Version(context.Context) (string, error) { return s.version, s.versionErr } + +func (s *checkSource) Definitions(context.Context) ([]Definition, error) { return s.defs, s.defsErr } + +// RuleState answers from the responder. A nil responder means the test +// expects no state read at all — it fails with a message rather than a nil +// dereference, because "this path must not poll" is an assertion several tests +// here make on purpose. +func (s *checkSource) RuleState(_ context.Context, title string) (Observation, error) { + s.mu.Lock() + s.calls[title]++ + call := s.calls[title] + s.mu.Unlock() + if s.respond == nil { + return Observation{}, fmt.Errorf("checkSource: this test expects no state read, but %q was polled", title) + } + return s.respond(title, call) +} + +func (s *checkSource) callCount(title string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.calls[title] +} + +var _ Source = (*checkSource)(nil) + +// checkStateRule builds one state-endpoint rule whose totals agree with the +// instances it carries. That agreement is load-bearing: a totals map claiming +// normal instances that the instance list does not contain fails §3.2's +// verification (VerifyNormalInstancesVisible), which is a different failure +// from the one most of these tests are about. +func checkStateRule(lastEval time.Time, insts ...Instance) StateRule { + totals := map[string]int{} + for _, i := range insts { + totals[string(i.State)]++ + } + return StateRule{ + UID: checkUID, Title: checkTitle, Folder: "F", Group: "G", + Interval: time.Minute, State: "inactive", Health: "ok", + LastEvaluation: lastEval, Totals: totals, Instances: insts, + } +} + +// healthyObservation is a poll of a rule that evaluated at this instant, with +// no skew at all — every test that is not ABOUT skew uses zero so its +// arithmetic reads directly off the timestamps. +func healthyObservation(now time.Time, insts ...Instance) Observation { + return Observation{ + Rules: []StateRule{checkStateRule(now, insts...)}, + GrafanaNow: now, + Latency: 200 * time.Millisecond, + } +} + +// baseConfig is a single-step run over [now, now+5m]: window 5m, grace 60s, so +// the collection loop ends at now+6m. +func baseConfig(t *testing.T, clock Clock) Config { + t.Helper() + now := clock.Now() + return Config{ + URL: "https://grafana.example.com", + Alerts: []string{"uid:" + checkUID}, + From: now, + To: now.Add(5 * time.Minute), + Clock: clock, + Notes: &strings.Builder{}, + }.withDefaults() +} + +func notesOf(cfg Config) string { return cfg.Notes.(*strings.Builder).String() } + +// --------------------------------------------------------------------------- +// §19.1 step 1 — configuration validation +// --------------------------------------------------------------------------- + +func TestCheckValidateRejectsBadConfigurations(t *testing.T) { + clock := newFakeClock(testNow) + base := func() Config { + return Config{ + URL: "https://grafana.example.com", + From: testNow, + To: testNow.Add(5 * time.Minute), + Clock: clock, + } + } + + tests := []struct { + name string + mutate func(*Config) + wantErr string + }{ + { + name: "no url", + mutate: func(c *Config) { c.URL = ""; c.Alerts = []string{"A"} }, + wantErr: "no grafana url", + }, + { + name: "no to", + mutate: func(c *Config) { c.To = time.Time{}; c.Alerts = []string{"A"} }, + wantErr: "no `to`", + }, + { + // §19.1 step 1: an empty Alerts is an error — but only without a log. + name: "single-step without alerts", + mutate: func(c *Config) {}, + wantErr: "no alert names given", + }, + { + // An alerts file ending in a newline must not read as a named alert. + name: "single-step with only blank alert lines", + mutate: func(c *Config) { c.Alerts = []string{"", " "} }, + wantErr: "no alert names given", + }, + { + // §19.1 step 3, the other direction: the log names the alert set. + name: "log mode with alerts", + mutate: func(c *Config) { c.Log = "log.jsonl"; c.Alerts = []string{"A"} }, + wantErr: "--alerts is refused with a recorded log", + }, + { + // §7 — never a warning-and-continue. + name: "log mode without from", + mutate: func(c *Config) { c.Log = "log.jsonl"; c.From = time.Time{} }, + wantErr: "the deploy step must emit a completion timestamp", + }, + { + name: "from beyond the future tolerance", + mutate: func(c *Config) { + c.Alerts = []string{"A"} + c.From = testNow.Add(2 * time.Minute) + c.To = testNow.Add(10 * time.Minute) + }, + wantErr: "ahead of this runner's clock", + }, + { + name: "to before from", + mutate: func(c *Config) { c.Alerts = []string{"A"}; c.To = testNow.Add(-time.Minute) }, + wantErr: "is before `from`", + }, + { + // A past `to` is only "not a special mode" WITH a log: without one + // the coverage window ends before the first observation exists. + name: "single-step with a to already past", + mutate: func(c *Config) { + c.Alerts = []string{"A"} + c.From = testNow.Add(-10 * time.Minute) + c.To = testNow.Add(-time.Minute) + }, + wantErr: "can only be classified from a recording", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := base() + tc.mutate(&cfg) + err := cfg.withDefaults().validate() + if err == nil { + t.Fatalf("validate() = nil, want an error containing %q", tc.wantErr) + } + if !strings.Contains(err.Error(), tc.wantErr) { + t.Fatalf("validate() = %q, want it to contain %q", err, tc.wantErr) + } + }) + } +} + +// A past `to` WITH a log is explicitly not a special mode (§7, §24.3): the +// collection loop's condition is already true and the evidence classifies +// immediately. No branch, and no refusal. +func TestCheckValidateAcceptsAPastToWithALog(t *testing.T) { + cfg := Config{ + URL: "https://grafana.example.com", + Log: "log.jsonl", + From: testNow.Add(-10 * time.Minute), + To: testNow.Add(-time.Minute), + Clock: newFakeClock(testNow), + }.withDefaults() + + if err := cfg.validate(); err != nil { + t.Fatalf("validate() = %v, want nil", err) + } + if cfg.PidFile != "log.jsonl.pid" { + t.Errorf("PidFile = %q, want the .pid default", cfg.PidFile) + } +} + +// --------------------------------------------------------------------------- +// Single-step mode +// --------------------------------------------------------------------------- + +func TestCheckSingleStepCleanWindowPasses(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + src := newCheckSource(func(_ string, _ int) (Observation, error) { + return healthyObservation(clock.Now()), nil + }) + + res, err := check(context.Background(), cfg, src) + if err != nil { + t.Fatalf("check() = %v, want nil\nnotes:\n%s", err, notesOf(cfg)) + } + // H7: a pass is exactly this shape. + if len(res.Violations) != 0 { + t.Fatalf("Violations = %+v, want none", res.Violations) + } + if len(res.Verdicts) != 1 || res.Verdicts[0].Outcome != OutcomeClean { + t.Fatalf("Verdicts = %+v, want one clean verdict", res.Verdicts) + } + if cov := res.Coverage[checkUID]; !cov.Proved || cov.Unobservable { + t.Fatalf("Coverage = %+v, want proved", cov) + } + + // The collection loop ran to to+transitionGrace and no further (H5). + windowEnd := cfg.To.Add(checkGrace) + if clock.Now().Before(windowEnd) { + t.Errorf("stopped collecting at %s, before to+grace %s", clock.Now(), windowEnd) + } + // One measurement-pass poll plus one every 30s across the 6-minute + // collection, plus the drain wait's own polls. The exact count depends on + // the scheduler's random stagger, so assert the order of magnitude a full + // window implies rather than an exact number. + if got := src.callCount(checkTitle); got < 12 { + t.Errorf("polled %d times, want at least the ~13 a full 6-minute window at 30s implies", got) + } + if notes := notesOf(cfg); !strings.Contains(notes, "planned run time") { + t.Errorf("§13.2 requires the planned run time at start; notes were:\n%s", notes) + } +} + +// H5: a certain violation does not release the runner early, and it does not +// stop the gate reporting exit-1 shape — violations with a nil error. +func TestCheckSingleStepFiringInstanceReportsWithoutExitingEarly(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + firing := Instance{ + Labels: map[string]string{"alertname": "Rule One", "instance": "a"}, + State: StateFiring, + ActiveAt: testNow.Add(-10 * time.Minute), // bad before the window opened + } + src := newCheckSource(func(_ string, _ int) (Observation, error) { + return healthyObservation(clock.Now(), firing), nil + }) + + res, err := check(context.Background(), cfg, src) + if err != nil { + t.Fatalf("check() = %v, want nil (a violation is exit 1, not an error)", err) + } + if len(res.Violations) != 1 { + t.Fatalf("Violations = %+v, want exactly one", res.Violations) + } + if got := res.Violations[0].Outcome; got != OutcomePersistentlyBad { + t.Errorf("Outcome = %q, want %q", got, OutcomePersistentlyBad) + } + if windowEnd := cfg.To.Add(checkGrace); clock.Now().Before(windowEnd) { + t.Errorf("exited early at %s; H5 requires collecting to %s", clock.Now(), windowEnd) + } +} + +// §4.2/§22.4: in single-step mode an explicit `from` earlier than the first +// observation is a DECLARED blind interval — a warning and a pass, naming the +// exact interval it cannot see. Recorder mode keeps P7 check 2 strict. +func TestCheckSingleStepFromBeforeFirstObservationWarnsAndPasses(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + cfg.From = testNow.Add(-2 * time.Minute) + src := newCheckSource(func(_ string, _ int) (Observation, error) { + return healthyObservation(clock.Now()), nil + }) + + res, err := check(context.Background(), cfg, src) + if err != nil { + t.Fatalf("check() = %v, want a pass with a warning\nnotes:\n%s", err, notesOf(cfg)) + } + notes := notesOf(cfg) + if !strings.Contains(notes, "cannot see [") || !strings.Contains(notes, testNow.Format(time.RFC3339)) { + t.Errorf("want a warning naming the unseen interval; notes were:\n%s", notes) + } + // The classified window is the clamped one, and Result says so rather than + // reporting a window the run never proved. + if !res.From.Equal(testNow) { + t.Errorf("Result.From = %s, want the clamped %s", res.From, testNow) + } +} + +// §19.3 case 1: the failure limit was exceeded. The measurement pass succeeds +// and the collection loop then hits a terminal failure, so this exercises the +// path a live run really takes. +func TestCheckFailClosedOnExhaustedRetries(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + src := newCheckSource(func(_ string, call int) (Observation, error) { + if call > 1 { + return Observation{}, &RetryExhaustedError{Failures: 6, Cause: errors.New("connection refused")} + } + return healthyObservation(clock.Now()), nil + }) + + res, err := check(context.Background(), cfg, src) + if err == nil { + t.Fatalf("check() = nil, want the collection failure to fail closed") + } + if !strings.Contains(err.Error(), "collect evidence") { + t.Errorf("err = %q, want it to name the collection step", err) + } + if len(res.Violations) != 0 { + t.Errorf("Violations = %+v; an error must never be reported as a verdict", res.Violations) + } +} + +// §19.3 case 2: the resolution of the definitions failed. Both shapes — the +// ruler read itself failing, and a name that resolves to nothing. +func TestCheckFailClosedOnDefinitionResolution(t *testing.T) { + t.Run("ruler read fails", func(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + src := newCheckSource(nil) + src.defsErr = errors.New("502 bad gateway") + + if _, err := check(context.Background(), cfg, src); err == nil || + !strings.Contains(err.Error(), "read rule definitions") { + t.Fatalf("check() = %v, want a definitions-read failure", err) + } + }) + + t.Run("unknown alert name", func(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + cfg.Alerts = []string{"No Such Rule"} + src := newCheckSource(nil) + + if _, err := check(context.Background(), cfg, src); err == nil || + !strings.Contains(err.Error(), "no rule matched") { + t.Fatalf("check() = %v, want a no-match failure", err) + } + }) +} + +// The version gate (§2.7 control 2): an unsupported Grafana is exit 2 before +// anything else is attempted. +func TestCheckRefusesUnsupportedGrafanaVersion(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + src := newCheckSource(nil) + src.version = "12.4.0" + + if _, err := check(context.Background(), cfg, src); err == nil || + !strings.Contains(err.Error(), "unsupported grafana version") { + t.Fatalf("check() = %v, want the version gate to refuse 12.4.0", err) + } +} + +// §5.2: the budget is checked against the latencies the measurement pass +// actually measured, and a schedule that cannot fit errors at START rather +// than producing a gap-riddled recording nobody can classify. +func TestCheckSingleStepRefusesAScheduleThatDoesNotFit(t *testing.T) { + clock := newVirtualClock(testNow) + cfg := baseConfig(t, clock) + src := newCheckSource(func(_ string, _ int) (Observation, error) { + obs := healthyObservation(clock.Now()) + obs.Latency = 45 * time.Second // longer than the rule's own 30s cadence + return obs, nil + }) + + _, err := check(context.Background(), cfg, src) + if err == nil { + t.Fatalf("check() = nil, want the budget check to refuse the schedule") + } + for _, want := range []string{"raising concurrency", "raising poll-interval", "watching fewer alerts"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("err = %q, want it to name the control %q (§5.1)", err, want) + } + } +} + +// --------------------------------------------------------------------------- +// Recorder mode +// --------------------------------------------------------------------------- + +// recordedLog writes a log the way watch would have: a header, one poll every +// 30s over [start, end], and a stopped sentinel at sentinelAt. lastEvalLag is +// how far behind each poll's own GrafanaNow its lastEvaluation sits, which is +// what the drain-wait tests vary. +func recordedLog(t *testing.T, dir string, url string, startedAt, start, end, sentinelAt time.Time, lastEvalLag time.Duration) string { + t.Helper() + path := filepath.Join(dir, "log.jsonl") + clock := newFakeClock(sentinelAt) + w, err := NewWriter(path, clock) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + header := Header{ + URL: url, + GrafanaVersion: "13.1.0", + StartedAt: startedAt, + Rules: []LoggedRule{{ + UID: checkUID, Title: checkTitle, Folder: "F", Group: "G", + IntervalSeconds: 60, NoDataState: "OK", ExecErrState: "OK", + PollEverySeconds: checkPollEvery.Seconds(), + }}, + } + if err := w.WriteHeader(header); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + for at := start; !at.After(end); at = at.Add(checkPollEvery) { + if err := w.WritePoll(Poll{ + RuleUID: checkUID, GrafanaNow: at, Found: true, + State: "inactive", Health: "ok", LastEvaluation: at.Add(-lastEvalLag), + }); err != nil { + t.Fatalf("WritePoll: %v", err) + } + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + return path +} + +// deadPid returns a pid that is guaranteed to have exited — the normal state +// of a recorder by the time check signals it, since a recorder given --until +// (or one that finished cleanly) is already gone. +func deadPid(t *testing.T) int { + t.Helper() + cmd := exec.Command("/bin/sh", "-c", "exit 0") + if err := cmd.Start(); err != nil { + t.Fatalf("start a throwaway process: %v", err) + } + pid := cmd.Process.Pid + if err := cmd.Wait(); err != nil { + t.Fatalf("wait for the throwaway process: %v", err) + } + return pid +} + +func writePid(t *testing.T, path, contents string) { + t.Helper() + if err := os.WriteFile(path, []byte(contents), 0o644); err != nil { + t.Fatalf("write pidfile: %v", err) + } +} + +// recorderConfig points check at a recording of [testNow-1m, windowEnd+30s] +// over the window [testNow, testNow+5m]. +func recorderConfig(t *testing.T, clock Clock, logPath string) Config { + t.Helper() + return Config{ + URL: "https://grafana.example.com", + Log: logPath, + From: testNow, + To: testNow.Add(5 * time.Minute), + Clock: clock, + Notes: &strings.Builder{}, + }.withDefaults() +} + +func TestCheckRecorderModeCleanWindowPasses(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd.Add(30*time.Second), windowEnd.Add(30*time.Second), 0) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + clock := newVirtualClock(testNow.Add(time.Minute)) + cfg := recorderConfig(t, clock, logPath) + // The drain wait is satisfied from the log's own evidence, so the source + // must never be asked for a state — asserted by the nil responder. + src := newCheckSource(func(title string, _ int) (Observation, error) { + t.Errorf("the drain wait polled %q although the log already proves the evaluations", title) + return Observation{}, errors.New("unexpected poll") + }) + + res, err := check(context.Background(), cfg, src) + if err != nil { + t.Fatalf("check() = %v, want nil\nnotes:\n%s", err, notesOf(cfg)) + } + if len(res.Violations) != 0 { + t.Fatalf("Violations = %+v, want none", res.Violations) + } + if len(res.Verdicts) != 1 || res.Verdicts[0].Outcome != OutcomeClean { + t.Fatalf("Verdicts = %+v, want one clean verdict", res.Verdicts) + } + if res.GrafanaVersion != "13.1.0" { + t.Errorf("GrafanaVersion = %q, want the recorded one", res.GrafanaVersion) + } + // The collection loop still waited out to+transitionGrace (H5) even though + // the recorder had already finished. + if clock.Now().Before(windowEnd) { + t.Errorf("returned at %s, before to+grace %s", clock.Now(), windowEnd) + } +} + +// §19.3 case 3: the identity of the log is not correct. The check runs against +// the header read EARLY, so it fails before the window's wait rather than +// after it. +func TestCheckFailClosedOnWrongLogIdentity(t *testing.T) { + t.Run("different url", func(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://other.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 0) + + clock := newVirtualClock(testNow) + cfg := recorderConfig(t, clock, logPath) + _, err := check(context.Background(), cfg, newCheckSource(nil)) + if err == nil || !strings.Contains(err.Error(), "log identity") { + t.Fatalf("check() = %v, want a log-identity failure", err) + } + if !clock.Now().Equal(testNow) { + t.Errorf("the identity check waited out the window (now %s); it must fail before the wait", clock.Now()) + } + }) + + t.Run("rule no longer resolves", func(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 0) + + cfg := recorderConfig(t, newVirtualClock(testNow), logPath) + src := newCheckSource(nil) + src.defs = []Definition{{UID: "somebody-else", Title: "Other", Kind: KindGrafanaManaged, IntervalSeconds: 60}} + + _, err := check(context.Background(), cfg, src) + if err == nil || !strings.Contains(err.Error(), "log identity") { + t.Fatalf("check() = %v, want a log-identity failure", err) + } + }) +} + +// §19.3 case 4: the coverage proof failed. A hole in the middle of the +// recording is not saved by healthy data at both ends (§22.4). +func TestCheckFailClosedOnCoverageGap(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + path := filepath.Join(dir, "log.jsonl") + clock := newFakeClock(windowEnd.Add(30 * time.Second)) + w, err := NewWriter(path, clock) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + if err := w.WriteHeader(Header{ + URL: "https://grafana.example.com", GrafanaVersion: "13.1.0", StartedAt: testNow.Add(-time.Minute), + Rules: []LoggedRule{{UID: checkUID, Title: checkTitle, IntervalSeconds: 60, PollEverySeconds: checkPollEvery.Seconds()}}, + }); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + for at := testNow.Add(-time.Minute); !at.After(windowEnd.Add(30 * time.Second)); at = at.Add(checkPollEvery) { + // A three-minute hole in the middle of the window. + if at.After(testNow.Add(time.Minute)) && at.Before(testNow.Add(4*time.Minute)) { + continue + } + if err := w.WritePoll(Poll{ + RuleUID: checkUID, GrafanaNow: at, Found: true, State: "inactive", Health: "ok", LastEvaluation: at, + }); err != nil { + t.Fatalf("WritePoll: %v", err) + } + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + writePid(t, path+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + cfg := recorderConfig(t, newVirtualClock(testNow), path) + res, err := check(context.Background(), cfg, newCheckSource(nil)) + if err == nil { + t.Fatalf("check() = nil, want the coverage gap to fail closed") + } + if got := res.Coverage[checkUID].Reason; got != ReasonHeartbeatGap { + t.Errorf("Reason = %q, want %q", got, ReasonHeartbeatGap) + } + if got := res.Verdicts[0].Outcome; got != OutcomeUnobservable { + t.Errorf("Outcome = %q, want %q", got, OutcomeUnobservable) + } +} + +// §19.3 case 5: the drain limit passed. The recording itself is clean, so this +// isolates the drain wait — the rule simply never evaluates through the end of +// the window, and a rule that cannot answer that question is unobservable. +func TestCheckFailClosedOnDrainTimeout(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + // A 45s lag keeps every poll inside evalStaleAfter (120s), so P7 check 6 + // is silent and only the drain wait can fail. + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 45*time.Second) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + clock := newVirtualClock(testNow) + cfg := recorderConfig(t, clock, logPath) + frozen := windowEnd.Add(-45 * time.Second) + src := newCheckSource(func(_ string, _ int) (Observation, error) { + now := clock.Now() + return Observation{ + Rules: []StateRule{checkStateRule(frozen)}, + GrafanaNow: now, + }, nil + }) + + res, err := check(context.Background(), cfg, src) + if err == nil { + t.Fatalf("check() = nil, want the drain limit to fail closed") + } + if got := res.Coverage[checkUID].Reason; got != ReasonDrainTimeout { + t.Errorf("Reason = %q, want %q", got, ReasonDrainTimeout) + } + if got := res.Verdicts[0].Outcome; got != OutcomeUnobservable { + t.Errorf("Outcome = %q, want %q", got, OutcomeUnobservable) + } + if !strings.Contains(res.Verdicts[0].Note, "drain limit") { + t.Errorf("Note = %q, want it to explain the drain limit", res.Verdicts[0].Note) + } + if waited := clock.Now().Sub(windowEnd); waited < checkDrainLimit { + t.Errorf("gave up after %s of drain wait, want the full %s", waited, checkDrainLimit) + } +} + +// §14.5: a rule the state endpoint no longer serves is knowable on the FIRST +// drain poll, and the answer is rule_absent — the fault — rather than +// drain_timeout, which would only name the wait. It must not spend the whole +// drain limit to reach it. +func TestCheckDrainWaitNamesADeletedRuleAtOnce(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 45*time.Second) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + clock := newVirtualClock(testNow) + cfg := recorderConfig(t, clock, logPath) + // An authoritative 2xx that parsed and carries no matching rule. P2 + // retried every transport failure long before an Observation exists, so + // this is a deletion, not a hiccup. + src := newCheckSource(func(_ string, _ int) (Observation, error) { + return Observation{GrafanaNow: clock.Now()}, nil + }) + + res, err := check(context.Background(), cfg, src) + if err == nil { + t.Fatalf("check() = nil, want a deleted rule to fail closed") + } + if got := res.Coverage[checkUID].Reason; got != ReasonRuleAbsent { + t.Errorf("Reason = %q, want %q — the fault, not the wait", got, ReasonRuleAbsent) + } + if got := src.callCount(checkTitle); got != 1 { + t.Errorf("polled %d times, want exactly 1: the absence is knowable on the first poll", got) + } + if waited := clock.Now().Sub(windowEnd); waited >= checkDrainLimit { + t.Errorf("spent %s in the drain wait, want it to conclude at once", waited) + } +} + +// --------------------------------------------------------------------------- +// `skipped` comes from the header, not from a definition read after the window +// --------------------------------------------------------------------------- + +// pausedAfterWindowLog records a rule that was ACTIVE at record start and that +// fired inside the window. The caller then tells check that the rule's current +// definition says paused — the state somebody set after the fact. +func pausedAfterWindowLog(t *testing.T, dir string, firesAt time.Time, end, sentinelAt time.Time) string { + t.Helper() + path := filepath.Join(dir, "log.jsonl") + w, err := NewWriter(path, newFakeClock(sentinelAt)) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + if err := w.WriteHeader(Header{ + URL: "https://grafana.example.com", GrafanaVersion: "13.1.0", StartedAt: testNow.Add(-time.Minute), + Rules: []LoggedRule{{ + UID: checkUID, Title: checkTitle, IntervalSeconds: 60, + IsPaused: false, PollEverySeconds: checkPollEvery.Seconds(), + }}, + }); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + firing := Instance{ + Labels: map[string]string{"alertname": checkTitle, "instance": "a"}, + State: StateFiring, + ActiveAt: firesAt, + } + for at := testNow.Add(-time.Minute); !at.After(end); at = at.Add(checkPollEvery) { + p := Poll{RuleUID: checkUID, GrafanaNow: at, Found: true, State: "inactive", Health: "ok", LastEvaluation: at} + if !at.Before(firesAt) { + p.State = "firing" + p.Abnormal = []Instance{firing} + } + if err := w.WritePoll(p); err != nil { + t.Fatalf("WritePoll: %v", err) + } + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + return path +} + +// pausedAfterWindowCheck runs the timeline above. The recording reaches past +// to + transitionGrace, which for this 60s rule is to + 60s: the fresh +// definition says paused, but that no longer shrinks the grace — the header +// does, and the header says the rule was active (deriveGlobalTimings). +func pausedAfterWindowCheck(t *testing.T, allowPaused bool) (Result, error, Config) { + t.Helper() + dir := t.TempDir() + to := testNow.Add(5 * time.Minute) + end := to.Add(checkGrace + 30*time.Second) + logPath := pausedAfterWindowLog(t, dir, testNow.Add(2*time.Minute), end, end) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + cfg := recorderConfig(t, newVirtualClock(testNow), logPath) + cfg.AllowPaused = allowPaused + + src := newCheckSource(nil) + paused := checkDef() + paused.IsPaused = true // somebody paused it after the alert started paging + src.defs = []Definition{paused} + + res, err := check(context.Background(), cfg, src) + return res, err, cfg +} + +// The window's own evidence outranks a definition read after it closed: a rule +// that was active at record start is classified, whatever its pause state is +// by the time check resolves the definitions. +func TestCheckPausingARuleAfterTheWindowDoesNotMakeItSkipped(t *testing.T) { + res, err, cfg := pausedAfterWindowCheck(t, false) + if err != nil { + t.Fatalf("check() = %v, want a classified verdict\nnotes:\n%s", err, notesOf(cfg)) + } + if got := res.Verdicts[0].Outcome; got != OutcomeNewlyBad { + t.Fatalf("Outcome = %q, want %q: the rule was active for the whole window and fired inside it", got, OutcomeNewlyBad) + } + if len(res.Violations) != 1 || res.Violations[0].Outcome != OutcomeNewlyBad { + t.Fatalf("Violations = %+v, want the firing reported", res.Violations) + } + if strings.Contains(res.Verdicts[0].Note, "paused before the window opened") { + t.Errorf("Note = %q, which the log's own polls contradict", res.Verdicts[0].Note) + } +} + +// The regression pin for the loophole this fix closed. Reading skipped from +// the post-window definition made the rule skipped; --allow-paused then made +// skipped free; and a window in which the alert fired reported exit 0. The +// default message names --allow-paused, so an operator was led straight to it. +func TestCheckAllowPausedCannotExcuseARulePausedAfterItFired(t *testing.T) { + res, err, cfg := pausedAfterWindowCheck(t, true) + if err != nil { + t.Fatalf("check() = %v, want a classified verdict\nnotes:\n%s", err, notesOf(cfg)) + } + if len(res.Violations) == 0 { + t.Fatalf("Violations = none with --allow-paused: the run passed over a window in which the alert fired") + } +} + +// The other direction, unchanged: a rule the HEADER says was paused when the +// recording opened is genuinely skipped. It has no polls, so no coverage is +// attempted for it, and --allow-paused behaves as it always did. +func TestCheckHeaderPausedRuleStaysSkipped(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + path := filepath.Join(dir, "log.jsonl") + w, err := NewWriter(path, newFakeClock(windowEnd.Add(30*time.Second))) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + // Named in the header, is_paused true, and no poll records at all — the + // shape watch writes for a rule paused before the window opened (P6 + // deviation 4). + if err := w.WriteHeader(Header{ + URL: "https://grafana.example.com", GrafanaVersion: "13.1.0", StartedAt: testNow.Add(-time.Minute), + Rules: []LoggedRule{{ + UID: checkUID, Title: checkTitle, IntervalSeconds: 60, + IsPaused: true, PollEverySeconds: checkPollEvery.Seconds(), + }}, + }); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + writePid(t, path+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + run := func(allowPaused bool) (Result, error) { + cfg := recorderConfig(t, newVirtualClock(testNow), path) + cfg.AllowPaused = allowPaused + // The definition is unpaused now; the header still decides. + src := newCheckSource(func(title string, _ int) (Observation, error) { + t.Errorf("the drain wait polled skipped rule %q", title) + return Observation{}, errors.New("unexpected poll") + }) + return check(context.Background(), cfg, src) + } + + res, err := run(false) + if err != nil { + t.Fatalf("check() = %v, want exit-1 shape: a skipped rule is a known condition, not an inability", err) + } + if got := res.Verdicts[0].Outcome; got != OutcomeSkipped { + t.Fatalf("Outcome = %q, want %q", got, OutcomeSkipped) + } + if _, ok := res.Coverage[checkUID]; ok { + t.Errorf("Coverage[%s] present, want absent: a skipped rule has no coverage to prove", checkUID) + } + if len(res.Violations) != 1 { + t.Errorf("Violations = %+v, want the MinObserved shortfall (§12.1)", res.Violations) + } + + res, err = run(true) + if err != nil || len(res.Violations) != 0 { + t.Errorf("with --allow-paused: err = %v, Violations = %+v, want a pass", err, res.Violations) + } +} + +// A paused rule does not evaluate, so it can never catch up: the drain wait +// must conclude on the first poll instead of spending the whole limit. +func TestCheckDrainWaitConcludesAtOnceOnAPausedRule(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 45*time.Second) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + clock := newVirtualClock(testNow) + cfg := recorderConfig(t, clock, logPath) + src := newCheckSource(func(_ string, _ int) (Observation, error) { + now := clock.Now() + rule := checkStateRule(windowEnd.Add(-45 * time.Second)) + rule.IsPaused = true + return Observation{Rules: []StateRule{rule}, GrafanaNow: now}, nil + }) + + res, err := check(context.Background(), cfg, src) + if err == nil { + t.Fatalf("check() = nil, want a rule that stopped evaluating to fail closed") + } + if got := src.callCount(checkTitle); got != 1 { + t.Errorf("polled %d times, want exactly 1: a paused rule can never catch up", got) + } + if got := res.Coverage[checkUID].Reason; got != ReasonDrainTimeout { + t.Errorf("Reason = %q, want %q — the vocabulary is published (§19.0), so the detail goes in the note", got, ReasonDrainTimeout) + } + if !strings.Contains(res.Verdicts[0].Note, "paused before it evaluated through") { + t.Errorf("Note = %q, want it to say the rule was paused", res.Verdicts[0].Note) + } + if waited := clock.Now().Sub(windowEnd); waited >= checkDrainLimit { + t.Errorf("spent %s in the drain wait, want it to conclude at once", waited) + } +} + +// P6's obligation on this phase: an absent or unparseable pidfile is never +// "there was nothing to stop". The parent writes the pidfile only once the +// child reports that it is recording, so a missing one means the recording +// never started — and the log must not be read at all. +func TestCheckRefusesToReadALogItCannotStop(t *testing.T) { + windowEnd := testNow.Add(5*time.Minute + checkGrace) + + tests := []struct { + name string + pidfile string // "" = do not create one + }{ + {name: "missing pidfile"}, + {name: "unparseable pidfile", pidfile: "not-a-pid\n"}, + {name: "empty pidfile", pidfile: ""}, + } + + for i, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 0) + if i != 0 { + writePid(t, logPath+".pid", tc.pidfile) + } + + cfg := recorderConfig(t, newVirtualClock(testNow), logPath) + _, err := check(context.Background(), cfg, newCheckSource(nil)) + if err == nil || !strings.Contains(err.Error(), "cannot stop the recorder") { + t.Fatalf("check() = %v, want a refusal to stop the recorder", err) + } + }) + } +} + +// startLockHolder re-execs this test binary as a process that holds the log's +// flock and ignores SIGTERM, and returns its pid once the lock is genuinely +// held. See lockHolderEnv (watch_daemon_test.go) for why it must be a separate +// real process rather than a shell one-liner. +func startLockHolder(t *testing.T, logPath string) int { + t.Helper() + cmd := exec.Command(os.Args[0]) + cmd.Env = append(os.Environ(), lockHolderEnv+"="+logPath) + cmd.Stderr = os.Stderr + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("pipe: %v", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("start the lock holder: %v", err) + } + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + }) + if _, err := bufio.NewReader(stdout).ReadString('\n'); err != nil { + t.Fatalf("the lock holder never reported holding the lock: %v", err) + } + return cmd.Process.Pid +} + +// §4.4 step 4: a recorder that will not let go of the log means the log may +// still be appended to, and a log a writer can change cannot be read at all. +func TestCheckFailsWhenTheRecorderWillNotExit(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd, windowEnd.Add(30*time.Second), 0) + + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", startLockHolder(t, logPath))) + + cfg := recorderConfig(t, newVirtualClock(testNow), logPath) + _, err := check(context.Background(), cfg, newCheckSource(nil)) + if err == nil || !strings.Contains(err.Error(), "still holds") { + t.Fatalf("check() = %v, want the stop wait to time out on the lock", err) + } +} + +// The regression pin for a stray SIGTERM. Nothing removes the pidfile when a +// recorder exits cleanly — the parent has returned and the child never learns +// the path — so after a --until run, a supported flow, the pidfile names a pid +// the operating system is free to hand to somebody else. The flock, not the +// pid, is what says whether a writer exists. +func TestCheckDoesNotSignalABystanderHoldingAReusedPid(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + logPath := recordedLog(t, dir, "https://grafana.example.com", + testNow.Add(-time.Minute), testNow.Add(-time.Minute), windowEnd.Add(30*time.Second), windowEnd.Add(30*time.Second), 0) + + // An innocent process that happens to hold the pid the finished recorder + // left behind. It does not hold the log's lock, because it is not a + // recorder. + bystander := exec.Command("sleep", "30") + if err := bystander.Start(); err != nil { + t.Fatalf("start the bystander: %v", err) + } + t.Cleanup(func() { + _ = bystander.Process.Kill() + _ = bystander.Wait() + }) + writePid(t, logPath+".pid", fmt.Sprintf("%d\n", bystander.Process.Pid)) + + cfg := recorderConfig(t, newVirtualClock(testNow), logPath) + if _, err := check(context.Background(), cfg, newCheckSource(nil)); err != nil { + t.Fatalf("check() = %v, want nil\nnotes:\n%s", err, notesOf(cfg)) + } + if err := syscall.Kill(bystander.Process.Pid, 0); err != nil { + t.Fatalf("the bystander is gone (%v): check signalled a process that was not the recorder", err) + } +} + +// P5's "two authorities", from check's side: maxGap comes from the cadence the +// header records, never from a re-derivation off intervalSeconds. The +// fail-open direction is the one asserted — a log recorded at 5s on a 60s rule +// must still fail on a hole a re-derived 30s maxGap would have forgiven. +func TestCheckDerivesMaxGapFromTheRecordedCadence(t *testing.T) { + dir := t.TempDir() + windowEnd := testNow.Add(5*time.Minute + checkGrace) + path := filepath.Join(dir, "log.jsonl") + w, err := NewWriter(path, newFakeClock(windowEnd.Add(30*time.Second))) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + if err := w.WriteHeader(Header{ + URL: "https://grafana.example.com", GrafanaVersion: "13.1.0", StartedAt: testNow.Add(-time.Minute), + Rules: []LoggedRule{{UID: checkUID, Title: checkTitle, IntervalSeconds: 60, PollEverySeconds: 5}}, + }); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + for at := testNow.Add(-time.Minute); !at.After(windowEnd.Add(30 * time.Second)); at = at.Add(5 * time.Second) { + // A 20s hole: under the recorded 5s cadence maxGap is 10s and this + // fails; under a cadence re-derived from intervalSeconds it would be + // 60s and the hole would pass unseen. + if at.After(testNow.Add(time.Minute)) && at.Before(testNow.Add(80*time.Second)) { + continue + } + if err := w.WritePoll(Poll{ + RuleUID: checkUID, GrafanaNow: at, Found: true, State: "inactive", Health: "ok", LastEvaluation: at, + }); err != nil { + t.Fatalf("WritePoll: %v", err) + } + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + writePid(t, path+".pid", fmt.Sprintf("%d\n", deadPid(t))) + + cfg := recorderConfig(t, newVirtualClock(testNow), path) + res, err := check(context.Background(), cfg, newCheckSource(nil)) + if err == nil { + t.Fatalf("check() = nil; a 20s hole exceeds the 10s maxGap the recorded 5s cadence implies") + } + if got := res.Coverage[checkUID].Reason; got != ReasonHeartbeatGap { + t.Errorf("Reason = %q, want %q", got, ReasonHeartbeatGap) + } +} + +// --------------------------------------------------------------------------- +// The pieces, in isolation +// --------------------------------------------------------------------------- + +// The drain wait's one comparison is cross-domain (§16), and its uncertainty +// is spent in the fail-closed direction: an evaluation that only MIGHT have +// reached the end of the window does not count as one that did. +func TestEvaluatedThroughSpendsItsUncertaintyFailingClosed(t *testing.T) { + end := testNow + + tests := []struct { + name string + lastEval time.Time + skew, bound time.Duration + wantSatisfied bool + }{ + {name: "zero lastEvaluation never satisfies", lastEval: time.Time{}, wantSatisfied: false}, + {name: "exactly at the end, no skew", lastEval: end, wantSatisfied: true}, + {name: "one second short", lastEval: end.Add(-time.Second), wantSatisfied: false}, + { + name: "far enough past the end to absorb the bound", + // Grafana runs 10s fast; the reading translates back to end+5s and + // the 1s bound still leaves it past the end. + lastEval: end.Add(16 * time.Second), skew: 10 * time.Second, bound: time.Second, + wantSatisfied: true, + }, + { + name: "inside the bound is not proof", + // Translated it lands exactly on the end, so the bound can put it + // either side — which is not an evaluation THROUGH the end. + lastEval: end.Add(10 * time.Second), skew: 10 * time.Second, bound: time.Second, + wantSatisfied: false, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := evaluatedThrough(tc.lastEval, tc.skew, tc.bound, end); got != tc.wantSatisfied { + t.Errorf("evaluatedThrough() = %v, want %v", got, tc.wantSatisfied) + } + }) + } +} + +// H6 through the merge: a drain timeout on one rule and a coverage failure on +// another must both reach the message. Neither error may shadow the other. +func TestMergeDrainTimeoutsNamesEveryUnobservableRule(t *testing.T) { + res := Result{ + Coverage: map[string]CoverageResult{ + "a": {Proved: true}, + "b": {Unobservable: true, Reason: ReasonHeartbeatGap, Notes: []string{"rule \"B\": gap"}}, + }, + Verdicts: []RuleVerdict{ + {Alert: "A", RuleUID: "a", Outcome: OutcomeClean}, + {Alert: "B", RuleUID: "b", Outcome: OutcomeUnobservable}, + }, + } + + merged, err := mergeDrainTimeouts(res, map[string]drainVerdict{ + "a": {reason: ReasonDrainTimeout, note: "rule \"A\": did not evaluate through the end within the drain limit"}, + "b": {reason: ReasonDrainTimeout, note: "rule \"B\": did not evaluate through the end within the drain limit"}, + }) + if err == nil { + t.Fatal("mergeDrainTimeouts() = nil, want an error naming the newly unobservable rule") + } + // Its own shape: joined with decide's, two counts under one identical + // phrase would read as a contradiction rather than as two findings. + if !strings.Contains(err.Error(), "unobservable at the drain wait") { + t.Errorf("err = %q, want the drain wait's own error shape", err) + } + // Only A is newly unobservable; B was already, so naming it twice would + // only lengthen the message. + if !strings.Contains(err.Error(), "A ("+string(ReasonDrainTimeout)+")") { + t.Errorf("err = %q, want it to name A's drain timeout", err) + } + if strings.Contains(err.Error(), "B (") { + t.Errorf("err = %q, want it not to re-report B, which decide already reported", err) + } + if got := merged.Coverage["a"].Reason; got != ReasonDrainTimeout { + t.Errorf("Coverage[a].Reason = %q, want %q", got, ReasonDrainTimeout) + } + // B keeps the reason the coverage proof gave it — the FIRST reason wins, + // as it does inside proveCoverage. + if got := merged.Coverage["b"].Reason; got != ReasonHeartbeatGap { + t.Errorf("Coverage[b].Reason = %q, want the earlier %q", got, ReasonHeartbeatGap) + } + if merged.Verdicts[0].Outcome != OutcomeUnobservable { + t.Errorf("Verdicts[0].Outcome = %q, want %q", merged.Verdicts[0].Outcome, OutcomeUnobservable) + } +} + +// ReadLogHeader is the one read of a log a writer may still hold, so its +// refusals matter as much as its successes. +func TestReadLogHeader(t *testing.T) { + dir := t.TempDir() + + t.Run("reads line 1 while the log keeps growing", func(t *testing.T) { + path := filepath.Join(dir, "growing.jsonl") + w, err := NewWriter(path, newFakeClock(testNow)) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + defer w.Close() + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.WritePoll(Poll{RuleUID: "rule1", GrafanaNow: testNow, Found: true}); err != nil { + t.Fatalf("WritePoll: %v", err) + } + + h, err := ReadLogHeader(path) + if err != nil { + t.Fatalf("ReadLogHeader: %v", err) + } + if h.URL != testHeader().URL || len(h.Rules) != 1 { + t.Errorf("header = %+v, want the written one", h) + } + }) + + t.Run("a half-written header is not a header", func(t *testing.T) { + path := filepath.Join(dir, "torn.jsonl") + if err := os.WriteFile(path, []byte(`{"type":"header","url":"htt`), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + if _, err := ReadLogHeader(path); err == nil || + !strings.Contains(err.Error(), "no complete header") { + t.Fatalf("ReadLogHeader() = %v, want a refusal", err) + } + }) + + t.Run("a wrong schema version is refused", func(t *testing.T) { + path := filepath.Join(dir, "old.jsonl") + if err := os.WriteFile(path, []byte(`{"type":"header","schema_version":99,"url":"u"}`+"\n"), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + if _, err := ReadLogHeader(path); err == nil || + !strings.Contains(err.Error(), "schema version 99") { + t.Fatalf("ReadLogHeader() = %v, want a schema refusal", err) + } + }) +} diff --git a/grafana-alertcheck/internal/gate/classify.go b/grafana-alertcheck/internal/gate/classify.go index 11ddc8ba2..c6b32fc25 100644 --- a/grafana-alertcheck/internal/gate/classify.go +++ b/grafana-alertcheck/internal/gate/classify.go @@ -507,8 +507,17 @@ func decide(h Header, polls []Poll, sentinel *time.Time, defs []Definition, unobservableNames []string ) + // `skipped` is decided from the header, never from defs (§12). defs are + // resolved after the window has closed, so Definition.IsPaused describes + // the present; Header.pausedAtStart describes the moment the recording + // opened, which is the only moment "paused before the window opened" can + // mean. Reading the late definition instead let a rule that fired and was + // then paused report as skipped, with its firing never classified — and + // under AllowPaused that was a pass. + pausedAtStart := h.pausedAtStart() + for _, def := range defs { - if def.IsPaused { + if pausedAtStart[def.UID] { skippedRules = append(skippedRules, def) result.Verdicts = append(result.Verdicts, RuleVerdict{ Alert: def.Title, RuleUID: def.UID, Outcome: OutcomeSkipped, diff --git a/grafana-alertcheck/internal/gate/classify_test.go b/grafana-alertcheck/internal/gate/classify_test.go index 7176a9167..25d766b6c 100644 --- a/grafana-alertcheck/internal/gate/classify_test.go +++ b/grafana-alertcheck/internal/gate/classify_test.go @@ -34,6 +34,18 @@ func quietPoll(uid string, at time.Time) Poll { var defaultBad = badStateSet(nil) // {firing} +// pausedHeader builds the header decide reads `skipped` from: the pause state +// as of record start. Definition.IsPaused is deliberately NOT that authority +// — it comes from a ruler read taken after the window closed — so a test that +// wants a rule treated as skipped must say so HERE (Header.pausedAtStart). +func pausedHeader(startedAt time.Time, pausedUIDs ...string) Header { + h := Header{SchemaVersion: LogSchemaVersion, StartedAt: startedAt} + for _, uid := range pausedUIDs { + h.Rules = append(h.Rules, LoggedRule{UID: uid, IsPaused: true}) + } + return h +} + // --- clean / newly_bad --- func TestClassifyRule_NoEvidenceIsClean(t *testing.T) { @@ -310,7 +322,9 @@ func TestDecide_SkippedRuleNeverReachesProveCoverage(t *testing.T) { // No polls, no sentinel at all: a heartbeat_gap/no_sentinel misclassification // here would mean proveCoverage ran for a skipped rule (§4.3's obligation). - res, err := decide(Header{StartedAt: from.Add(-time.Hour)}, nil, nil, defs, rt, gt, pol) + // The HEADER is what says paused — decide reads skipped from there, not + // from def.IsPaused, which is a post-window reading (Header.pausedAtStart). + res, err := decide(pausedHeader(from.Add(-time.Hour), "r1"), nil, nil, defs, rt, gt, pol) if err != nil { t.Fatalf("err = %v, want nil: a rule paused before the window is skipped, not unobservable", err) } @@ -445,7 +459,7 @@ func TestDecide_SkippedOnlyShortfallProducesAViolationWithoutAnError(t *testing. } sentinel := to - res, err := decide(Header{StartedAt: from.Add(-time.Hour)}, polls, &sentinel, defs, rt, gt, pol) + res, err := decide(pausedHeader(from.Add(-time.Hour), "paused"), polls, &sentinel, defs, rt, gt, pol) if err != nil { t.Fatalf("err = %v, want nil: a shortfall caused only by a skipped rule is exit 1, not exit 2 (§9.1)", err) } @@ -516,7 +530,7 @@ func TestDecide_AllowPausedSuppressesTheShortfall(t *testing.T) { } sentinel := to - res, err := decide(Header{StartedAt: from.Add(-time.Hour)}, polls, &sentinel, defs, rt, gt, pol) + res, err := decide(pausedHeader(from.Add(-time.Hour), "paused"), polls, &sentinel, defs, rt, gt, pol) if err != nil { t.Fatalf("err = %v, want nil", err) } diff --git a/grafana-alertcheck/internal/gate/coverage.go b/grafana-alertcheck/internal/gate/coverage.go index 3b5425b6d..59a5dfb0f 100644 --- a/grafana-alertcheck/internal/gate/coverage.go +++ b/grafana-alertcheck/internal/gate/coverage.go @@ -8,22 +8,25 @@ import ( // keepLastReason is the instance Reason that check 9 watches for (§10.2). const keepLastReason = "KeepLast" -// Obligations this phase leaves for later ones — carried forward the same -// way P6's own deviations list did, so a later review has something concrete -// to check against: +// Two things this file deliberately does not do, and where they are done +// instead — both were open obligations when P7 was written, and both are now +// discharged: // -// - fromFutureTolerance (§5: 60s) has no constant and no hard-error check -// anywhere yet. Check 2 below implements only "from < StartedAt"; the -// second clause — from more than fromFutureTolerance ahead is a hard -// error — is once-per-run input validation, not a per-rule coverage -// check, and belongs to Check's construction in a later phase (P9). -// - decide (P8) must read a rule's skipped status from the definitions -// (LoggedRule.IsPaused / Definition.IsPaused), never from the polls, and -// must do so BEFORE calling proveCoverage for that rule: a rule paused -// before the window opened is never scheduled or polled (§4.3), so it -// reaches this function with zero polls and today reads as one large -// heartbeat_gap, not skipped (pinned by +// - §7's second clause, "from more than fromFutureTolerance ahead is a hard +// error", is once-per-run input validation rather than a per-rule +// coverage check, and this function has no error return. Discharged by +// P9: the constant is fromFutureTolerance (schedule.go) and Config.validate +// (check.go) applies it. Check 2 below still owns the first clause, +// "from < StartedAt". +// - A rule paused before the window opened is never scheduled or polled +// (§4.3), so it reaches this function with zero polls and reads as one +// large heartbeat_gap, not as skipped (pinned by // TestProveCoverage_SkippedRuleWithZeroPollsPinnedAsHeartbeatGap). +// Discharged by P8: decide returns before it ever calls proveCoverage for +// such a rule (classify.go). It reads skipped from the log header +// (Header.pausedAtStart), NOT from Definition.IsPaused — the definitions +// are re-resolved after the window closed, so they cannot answer what was +// paused when it opened. // UnobservableReason names why proveCoverage could not prove a rule's window. // It is machine-readable — this reaches the action's JSON outputs, so it is a diff --git a/grafana-alertcheck/internal/gate/flock.go b/grafana-alertcheck/internal/gate/flock.go index 9d7f7e750..865c6019a 100644 --- a/grafana-alertcheck/internal/gate/flock.go +++ b/grafana-alertcheck/internal/gate/flock.go @@ -1,6 +1,7 @@ package gate import ( + "errors" "fmt" "os" "syscall" @@ -16,3 +17,23 @@ func lockExclusive(f *os.File) error { } return nil } + +// tryLockExclusive is the same call read as a question rather than as a +// demand: held is false when another process holds the lock, and err is +// non-nil only for a failure that is not contention. +// +// check needs that distinction where NewWriter does not. NewWriter is entitled +// to treat any refusal as "another writer has it", because it wants the lock; +// check only wants to know whether a writer EXISTS (§4.4). The lock answers +// that directly, where a pid can only infer it — the kernel releases a flock +// when the holder exits, crash included, and pids get reused. +func tryLockExclusive(f *os.File) (held bool, err error) { + switch err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); { + case err == nil: + return true, nil + case errors.Is(err, syscall.EWOULDBLOCK): + return false, nil + default: + return false, fmt.Errorf("flock %s: %w", f.Name(), err) + } +} diff --git a/grafana-alertcheck/internal/gate/log.go b/grafana-alertcheck/internal/gate/log.go index 0d73ab092..fa9838aed 100644 --- a/grafana-alertcheck/internal/gate/log.go +++ b/grafana-alertcheck/internal/gate/log.go @@ -1,6 +1,7 @@ package gate import ( + "bufio" "encoding/json" "fmt" "os" @@ -39,16 +40,22 @@ type LoggedRule struct { Title string `json:"title"` Folder string `json:"folder"` Group string `json:"group"` - // ForSeconds, IntervalSeconds, IsPaused, NoDataState and ExecErrState are - // purely forensic: a resolve-time snapshot that makes the uploaded - // artifact self-describing to a human reading it after the runner is gone - // (§21.3). check never converts them back into a Definition — it always - // re-resolves definitions from the ruler API (§19.1 step 2). + // ForSeconds, IntervalSeconds, NoDataState and ExecErrState are purely + // forensic: a resolve-time snapshot that makes the uploaded artifact + // self-describing to a human reading it after the runner is gone (§21.3). + // check never converts them back into a Definition — it always re-resolves + // definitions from the ruler API (§19.1 step 2). ForSeconds float64 `json:"for_seconds"` IntervalSeconds int `json:"interval_seconds"` - IsPaused bool `json:"is_paused"` - NoDataState string `json:"no_data_state"` - ExecErrState string `json:"exec_err_state"` + // IsPaused is NOT forensic, and is the second load-bearing field here + // beside PollEverySeconds. It is the pause state at record start, which is + // the only moment `skipped` can honestly mean (§12), and decide reads it + // through Header.pausedAtStart rather than reading Definition.IsPaused off + // a ruler read taken after the window had already closed. See that method + // for what goes wrong the other way. + IsPaused bool `json:"is_paused"` + NoDataState string `json:"no_data_state"` + ExecErrState string `json:"exec_err_state"` // PollEverySeconds is the cadence this recording ACTUALLY used, after any // --poll-interval override. Load-bearing, not forensic: check derives // maxGap from it and never re-derives it from the definitions. Getting @@ -69,6 +76,30 @@ type Header struct { Rules []LoggedRule `json:"rules"` // THE alert set (§19.1 step 3) } +// pausedAtStart reports, per rule UID, whether the rule was paused when the +// recording opened. That instant — and no other — is what `skipped` means +// (§12): a rule nobody was watching on purpose. +// +// It is the authority for `skipped` in BOTH modes, and the reason is that no +// other source knows the right moment. `check` re-resolves the definitions +// AFTER the window closed (§19.1 step 2), so Definition.IsPaused there +// describes the present, not the window: a rule that fired and was then +// paused would read as skipped, its firing would never be classified, and +// under --allow-paused the run would pass. The header cannot drift that way, +// because watch stamps it before the deploy step runs and single-step check +// stamps it from definitions resolved at the start of its own step. +// +// A UID the header does not name is reported NOT paused, which is the safe +// direction: it then reaches proveCoverage with no polls and fails closed as +// a heartbeat gap, rather than being waved through as legitimately unwatched. +func (h Header) pausedAtStart() map[string]bool { + paused := make(map[string]bool, len(h.Rules)) + for _, lr := range h.Rules { + paused[lr.UID] = lr.IsPaused + } + return paused +} + // Poll is one reduced observation of one rule — the log's heartbeat and the // only input the pure coverage and classification layers ever see. type Poll struct { @@ -167,13 +198,7 @@ func (r *Reducer) Reduce(uid string, obs Observation) Poll { LatencyMS: obs.Latency.Milliseconds(), } - var rule *StateRule - for i := range obs.Rules { - if obs.Rules[i].UID == uid { - rule = &obs.Rules[i] - break - } - } + rule := stateRuleByUID(obs.Rules, uid) if rule == nil { // An authoritative "the rule is absent". No markers are computed and // the previous abnormal set is kept untouched: if the rule comes back @@ -265,6 +290,25 @@ func (r *Reducer) seedFrom(polls []Poll) { } } +// stateRuleByUID picks one rule out of a state-endpoint response BY UID, and +// nil means the response is an authoritative "the rule is absent" (§14.5). +// +// Never by title: the ?rule_name= filter is a title filter, and a filtered +// response can carry several rules sharing one title (the known 2-way +// collision), so picking the first would silently watch the wrong rule. This +// is the single implementation of that selection for the package — Reduce +// above and the drain wait (check.go) both call it, for the same reason +// pollsForRule (classify.go) is shared between proveCoverage and classifyRule: +// two copies of a membership test are two chances for one to drift. +func stateRuleByUID(rules []StateRule, uid string) *StateRule { + for i := range rules { + if rules[i].UID == uid { + return &rules[i] + } + } + return nil +} + // 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). @@ -460,6 +504,47 @@ func (w *Writer) Close() error { return nil } +// ReadLogHeader reads ONLY line 1 and is the one read of a log that a writer +// may still hold. That is safe for exactly one line and for no other: the +// header is written once, by watch's parent, before any child appends a byte, +// the file is opened O_APPEND and never O_TRUNC (§8), so line 1 is complete +// and immutable for the whole life of the recording. +// +// It exists so check can fail closed EARLY (§19.1 steps 3-4): the log's +// identity, the rule set and the cadences are all knowable at the start, and +// discovering a wrong URL or an unresolvable rule after a ten-minute wait +// helps nobody. It is advisory only — the authoritative read is still ReadLog, +// once, after the writer has exited (§4.4 step 4), and check re-validates the +// identity against that header rather than trusting this one. +func ReadLogHeader(path string) (Header, error) { + f, err := os.Open(path) + if err != nil { + return Header{}, fmt.Errorf("read log header %s: %w", path, err) + } + defer f.Close() + + line, err := bufio.NewReader(f).ReadString('\n') + if err != nil { + // io.EOF included: a log whose first line has no terminating newline is + // a log whose header was never fully written, which is not a header. + return Header{}, fmt.Errorf("log %s: no complete header on line 1: %w", path, err) + } + + var rec headerRecord + if err := json.Unmarshal([]byte(line), &rec); err != nil { + return Header{}, fmt.Errorf("log %s line 1: unparseable header: %w", path, err) + } + if rec.Type != RecordHeader { + return Header{}, fmt.Errorf("log %s line 1: got record type %q; the header must be line 1", path, rec.Type) + } + if rec.SchemaVersion != LogSchemaVersion { + return Header{}, fmt.Errorf( + "log %s: schema version %d is not %d — this log was written by a different version of the gate", + path, rec.SchemaVersion, LogSchemaVersion) + } + return rec.Header, nil +} + // ReadLog reads the whole log once and returns its header, its polls in // recorded order, and the sentinel time when one is present (nil when the // recording never finished — check turns that into unobservable, never a diff --git a/grafana-alertcheck/internal/gate/schedule.go b/grafana-alertcheck/internal/gate/schedule.go index f39c02d24..87b014375 100644 --- a/grafana-alertcheck/internal/gate/schedule.go +++ b/grafana-alertcheck/internal/gate/schedule.go @@ -14,6 +14,16 @@ import ( // P2 needed it before this file did, so it started there. const skewHardLimit = 60 * time.Second +// fromFutureTolerance is how far ahead of the runner's own clock a supplied +// `from` may sit before check refuses it (§7: "from in the future, more than +// the skew tolerance — error"). §7 names no number, so this is the judgment +// call §5's table records: the same 60s as skewHardLimit, because the only +// legitimate reason for a `from` in the future is clock disagreement between +// the deploy step and the check step, and that is bounded by the same figure. +// It is once-per-run input validation, not a per-rule coverage check, so +// Check applies it (P9) and proveCoverage does not. +const fromFutureTolerance = 60 * time.Second + // minDrainTimeout is §5's floor on drainTimeout: max(2 x max(intervalSeconds), // 2m). Without the floor, a fleet of very tight rules would derive a // drainTimeout too short to let a healthy in-flight poll land. @@ -94,7 +104,22 @@ func DeriveTimings(defs []Definition, override time.Duration) (rules map[string] } rules[d.UID] = newRuleTimings(pollEvery, d.IntervalSeconds) } - return rules, deriveGlobalTimings(defs), notes + // In this mode the defs ARE the start-of-step snapshot — watch resolves + // them before it detaches, and single-step check before its first + // observation — so they can answer what was paused when the window opened. + // Only the log-mode counterpart below has to look elsewhere. + return rules, deriveGlobalTimings(defs, pausedSet(defs)), notes +} + +// pausedSet is Header.pausedAtStart's counterpart for a set of definitions +// resolved at the start of the step, which is the one moment a definition can +// answer "was this paused when the window opened". +func pausedSet(defs []Definition) map[string]bool { + paused := make(map[string]bool, len(defs)) + for _, d := range defs { + paused[d.UID] = d.IsPaused + } + return paused } // DeriveTimingsFromLog is DeriveTimings' log-mode counterpart, and the two @@ -155,17 +180,31 @@ func DeriveTimingsFromLog(h Header, defs []Definition) (rules map[string]ruleTim pollEvery := time.Duration(lr.PollEverySeconds * float64(time.Second)) rules[lr.UID] = newRuleTimings(pollEvery, def.IntervalSeconds) } - return rules, deriveGlobalTimings(defs), nil + // The header, not defs, decides which rules are excluded from the grace: + // defs were resolved after the window closed. See deriveGlobalTimings. + return rules, deriveGlobalTimings(defs, h.pausedAtStart()), nil } // deriveGlobalTimings computes transitionGrace and drainTimeout over defs -// (§5, §13.1, §19). A rule paused before the window opened — skipped, §12 — -// is excluded from the transitionGrace max: its `for` value can never fire -// during the window, so counting it would only inflate the wait past what any -// watched rule actually needs (a judgment call the v2 plan makes explicitly -// for this formula; §19's drainTimeout carries no such exclusion, so it still -// runs over every resolved rule). -func deriveGlobalTimings(defs []Definition) globalTimings { +// (§5, §13.1, §19). +// +// A rule paused before the window opened — skipped, §12 — is excluded from the +// transitionGrace max: its `for` value can never fire during the window, so +// counting it would only inflate the wait past what any watched rule actually +// needs (a judgment call the v2 plan makes explicitly for this formula; §19's +// drainTimeout carries no such exclusion, so it still runs over every resolved +// rule). +// +// "Before the window opened" is the whole content of that exclusion, so the +// authority is pausedAtStart and NEVER Definition.IsPaused: in log mode the +// definitions are re-resolved after the window closed. Reading them instead +// was a fail-open, and a quiet one. transitionGrace is what lets a condition +// arising just before `to` be seen when it surfaces at to + `for`, and +// windowEnd is BOTH the classification bound and the collection deadline — so +// a rule somebody paused after `to` dropped out of the max, the grace +// collapsed, the surfacing poll was never even recorded, and the run reported +// clean. With one watched rule the shrink is total. +func deriveGlobalTimings(defs []Definition, pausedAtStart map[string]bool) globalTimings { var g globalTimings var maxInterval time.Duration for _, d := range defs { @@ -173,7 +212,7 @@ func deriveGlobalTimings(defs []Definition) globalTimings { if interval > maxInterval { maxInterval = interval } - if d.IsPaused { + if pausedAtStart[d.UID] { continue } if candidate := d.For + interval; candidate > g.transitionGrace { diff --git a/grafana-alertcheck/internal/gate/schedule_test.go b/grafana-alertcheck/internal/gate/schedule_test.go index 67659bc2f..1a992f8bd 100644 --- a/grafana-alertcheck/internal/gate/schedule_test.go +++ b/grafana-alertcheck/internal/gate/schedule_test.go @@ -77,6 +77,70 @@ func TestDeriveTimings_TransitionGraceZeroWhenAllSkipped(t *testing.T) { } } +// TestDeriveTimingsFromLog_TransitionGraceFollowsTheHeaderNotTheDefinition +// pins the log-mode authority for the grace exclusion. Definitions are +// re-resolved AFTER the window closed, so "paused" in a definition says +// nothing about whether the rule was watched during it. +// +// The fail-open direction is the first case. transitionGrace exists so a +// condition arising just before `to` is still seen when it surfaces at +// to + `for`, and windowEnd is both the classification bound and the +// collection deadline — so a rule somebody paused after `to` dropping out of +// the max collapses the grace, the surfacing poll is never recorded, and the +// run reports clean. +func TestDeriveTimingsFromLog_TransitionGraceFollowsTheHeaderNotTheDefinition(t *testing.T) { + loggedRule := func(uid string, pausedAtStart bool) LoggedRule { + return LoggedRule{UID: uid, Title: uid, IntervalSeconds: 60, PollEverySeconds: 30, IsPaused: pausedAtStart} + } + // The definition says paused in BOTH cases: it is the post-window reading, + // and it must change nothing. + defs := []Definition{{UID: "r1", Title: "R1", IntervalSeconds: 60, For: 5 * time.Minute, IsPaused: true}} + want := 5*time.Minute + 60*time.Second + + t.Run("header says active: the rule stays in the max", func(t *testing.T) { + h := Header{Rules: []LoggedRule{loggedRule("r1", false)}} + _, global, err := DeriveTimingsFromLog(h, defs) + if err != nil { + t.Fatalf("DeriveTimingsFromLog: %v", err) + } + if global.transitionGrace != want { + t.Fatalf("transitionGrace = %s, want %s: the rule was active when the recording opened, "+ + "so a pause applied afterwards must not shrink the window", global.transitionGrace, want) + } + if !strings.Contains(global.graceSource, "R1") { + t.Errorf("graceSource = %q, want it to name R1", global.graceSource) + } + }) + + t.Run("header says paused: the rule stays out", func(t *testing.T) { + h := Header{Rules: []LoggedRule{loggedRule("r1", true)}} + _, global, err := DeriveTimingsFromLog(h, defs) + if err != nil { + t.Fatalf("DeriveTimingsFromLog: %v", err) + } + if global.transitionGrace != 0 { + t.Fatalf("transitionGrace = %s, want 0: a rule paused before the window opened can never fire during it", + global.transitionGrace) + } + }) + + t.Run("drainTimeout counts every rule either way", func(t *testing.T) { + // §19 puts no pause exclusion on drainTimeout, so both headers give the + // same floor-bound value. + for _, pausedAtStart := range []bool{false, true} { + h := Header{Rules: []LoggedRule{loggedRule("r1", pausedAtStart)}} + _, global, err := DeriveTimingsFromLog(h, defs) + if err != nil { + t.Fatalf("DeriveTimingsFromLog: %v", err) + } + if global.drainTimeout != minDrainTimeout { + t.Fatalf("drainTimeout = %s with pausedAtStart=%v, want the %s floor", + global.drainTimeout, pausedAtStart, minDrainTimeout) + } + } + }) +} + func TestDeriveTimings_DrainTimeoutIncludesPaused(t *testing.T) { defs := []Definition{{UID: "r1", Title: "R1", IntervalSeconds: 10}} _, global, _ := DeriveTimings(defs, 0) diff --git a/grafana-alertcheck/internal/gate/watch.go b/grafana-alertcheck/internal/gate/watch.go index 61e1cce63..67f60954d 100644 --- a/grafana-alertcheck/internal/gate/watch.go +++ b/grafana-alertcheck/internal/gate/watch.go @@ -353,7 +353,6 @@ func openRecording(ctx context.Context, cfg WatchConfig, src Source, writer *Wri // 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) @@ -361,48 +360,15 @@ func openRecording(ctx context.Context, cfg WatchConfig, src Source, writer *Wri } 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) + polls, measured, err := firstObservations(ctx, src, active, NewReducer(), cfg.Concurrency, cfg.Notes) 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 { + for _, p := range polls { + if err := writer.WritePoll(p); err != nil { return nil, err } } @@ -441,6 +407,66 @@ func loggedRules(defs []Definition, rt map[string]ruleTimings) []LoggedRule { return out } +// firstObservations takes one observation of every rule in active, verifies +// §3.2 against those very responses, and reduces each into the poll record +// that IS the window's first heartbeat — plus the measured latency of each, +// which is the only honest input to §5.2's budget check (a fixed estimate is +// worthless when one rule's payload is ~230x another's). +// +// Both entry paths share it: watch's parent, before it detaches (§4.3), and +// single-step check's measurement pass, which keeps the polls as evidence +// rather than writing them to a log (P9). Keeping one implementation is the +// point — the §3.2 verification and the "absent is a warning, not an error" +// rule are exactly the places where two copies would silently drift, and a +// drift in either direction is fail-open. +// +// polls come back in `active` order, so a log written from them is byte-stable +// for a given set of observations. +func firstObservations(ctx context.Context, src Source, active []Definition, reducer *Reducer, + concurrency int, notes io.Writer) ([]Poll, map[string]time.Duration, error) { + + titles := make(map[string]string, len(active)) + uids := make([]string, 0, len(active)) + for _, d := range active { + titles[d.UID] = d.Title + uids = append(uids, d.UID) + } + + observed, err := observeAll(ctx, src, titles, uids, concurrency) + if err != nil { + return nil, 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, nil, err + } + } + + polls := make([]Poll, 0, len(active)) + 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 + // check 8 turns into unobservable — a note rather than an error + // here, because the state endpoint can lag a freshly created rule + // and the coverage proof fails closed either way. + fmt.Fprintf(notes, "warning: rule %q (%s) is absent from the state endpoint; recorded as not found\n", d.Title, d.UID) + } + polls = append(polls, poll) + } + return polls, measured, nil +} + // 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 diff --git a/grafana-alertcheck/internal/gate/watch_daemon_test.go b/grafana-alertcheck/internal/gate/watch_daemon_test.go index 35e9c424a..db35c505b 100644 --- a/grafana-alertcheck/internal/gate/watch_daemon_test.go +++ b/grafana-alertcheck/internal/gate/watch_daemon_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "os" + "os/signal" "path/filepath" "slices" "strconv" @@ -22,12 +23,48 @@ import ( // 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 path := os.Getenv(lockHolderEnv); path != "" { + os.Exit(runTestLockHolder(path)) + } if slices.Contains(os.Args, DaemonChildFlag) { os.Exit(runTestDaemonChild(os.Args[1:])) } os.Exit(m.Run()) } +// lockHolderEnv turns this test binary into a stand-in recorder that holds the +// log's flock and refuses to die: a process check's stop protocol must wait +// for and, on a timeout, refuse to read around. +// +// It has to be a real second process. flock is what stopRecorder probes, and +// there is no flock(1) on darwin, so a shell one-liner cannot take the lock — +// while a lock taken in the test process itself would be granted to the probe +// on some platforms and prove nothing. +const lockHolderEnv = "GRAFANA_ALERTCHECK_TEST_LOCK_HOLDER" + +// runTestLockHolder takes the log's exclusive lock, reports that it has it on +// stdout, ignores SIGTERM, and waits to be killed. The report is what lets the +// test start only once the lock is genuinely held, rather than racing it. +func runTestLockHolder(path string) int { + signal.Ignore(syscall.SIGTERM) + + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + if err := lockExclusive(f); err != nil { + fmt.Fprintln(os.Stderr, err) + return 1 + } + fmt.Println("locked") + + // Long enough to outlive any test that starts it; the test kills it, and + // SIGKILL is not ignorable. + time.Sleep(5 * time.Minute) + return 0 +} + // 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.