diff --git a/grafana-alertcheck/internal/gate/flock_unix.go b/grafana-alertcheck/internal/gate/flock_unix.go new file mode 100644 index 000000000..7b927f315 --- /dev/null +++ b/grafana-alertcheck/internal/gate/flock_unix.go @@ -0,0 +1,24 @@ +//go:build unix + +package gate + +import ( + "fmt" + "os" + "syscall" +) + +// lockExclusive takes a non-blocking exclusive lock on f. Non-blocking is the +// point (§8): a second writer must fail immediately with an error the operator +// sees, not queue behind the first and start appending to a log somebody else +// already finished. +// +// There is deliberately no Windows implementation — runners are Linux and +// goreleaser builds linux+darwin only (P6, P12) — so the package does not +// build there at all rather than silently skipping the lock. +func lockExclusive(f *os.File) error { + if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil { + return fmt.Errorf("flock: %w", err) + } + return nil +} diff --git a/grafana-alertcheck/internal/gate/log.go b/grafana-alertcheck/internal/gate/log.go new file mode 100644 index 000000000..3f940f7d0 --- /dev/null +++ b/grafana-alertcheck/internal/gate/log.go @@ -0,0 +1,521 @@ +package gate + +import ( + "encoding/json" + "fmt" + "os" + "sort" + "strings" + "sync" + "time" +) + +// LogSchemaVersion is the version stamped into every log header. A log with +// any other value is a read error, never a best-effort read: the log is the +// gate's only evidence, and misreading a stale shape is a fail-open (§5). +const LogSchemaVersion = 1 + +// RecordType tags each JSONL line. There are exactly three, and a poll record +// IS the heartbeat — there is deliberately no separate heartbeat type (§4.6). +type RecordType string + +const ( + RecordHeader RecordType = "header" + RecordPoll RecordType = "poll" + RecordStopped RecordType = "stopped" +) + +// missingSeriesReason is the reason Grafana parks a disappearing series at +// ("Normal (MissingSeries)") for a couple of evaluations before deleting the +// instance. Reading that as a recovery is H2's named bug, so the markers below +// route it to Vanished (P1.2a). +const missingSeriesReason = "MissingSeries" + +// LoggedRule is the per-rule identity written into the header. Together with +// the header URL it IS the log's identity, which check validates (§19.1 step +// 3), and it supplies the alert set in check mode. +type LoggedRule struct { + UID string `json:"uid"` + 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 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"` + // 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 + // that wrong is fail-open in the faster-override direction — a real + // recorder gap would pass silently (see "Two authorities", P5). + PollEverySeconds float64 `json:"poll_every_seconds"` +} + +// Header is the log's first line: what was recorded, from where, and when the +// recording started. It carries no States field — recording is deliberately +// unfiltered, so the same log can be re-classified under different --states +// without re-recording (P6). +type Header struct { + SchemaVersion int `json:"schema_version"` + URL string `json:"url"` // the log's identity (§19.1 step 3) + GrafanaVersion string `json:"grafana_version"` + StartedAt time.Time `json:"started_at"` // the record start (§7 validation) + Rules []LoggedRule `json:"rules"` // THE alert set (§19.1 step 3) +} + +// 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 { + RuleUID string `json:"rule_uid"` + GrafanaNow time.Time `json:"grafana_now"` // the Date header — H4 + // SkewMS, SkewBoundMS and LatencyMS are milliseconds for JSONL + // compactness ONLY. The pure layer never touches raw ms: it reads + // Skew(), SkewBound() and Latency() below, which convert at the + // (de)serialization boundary. + SkewMS int64 `json:"skew_ms"` + SkewBoundMS int64 `json:"skew_bound_ms"` + LatencyMS int64 `json:"latency_ms"` + // Found false means an authoritative 2xx in which this rule was absent + // (§14.5) — never a transport failure, which P2 retried and never turns + // into a Poll. P7 check 8 turns it into unobservable. + Found bool `json:"found"` + // State, Health and LastError are the raw rule-level strings, reporting + // only and never classified (P1.2a). + State string `json:"state,omitempty"` + Health string `json:"health,omitempty"` + LastError string `json:"last_error,omitempty"` + // omitzero, not omitempty: a not-found poll (and a paused rule, §2.3) has + // no evaluation time, and writing "0001-01-01T00:00:00Z" into an artifact + // humans and jq read (§21.3) invites reading it as a real timestamp. + LastEvaluation time.Time `json:"last_evaluation,omitzero"` + IsPaused bool `json:"is_paused"` + Histogram map[string]int `json:"histogram,omitempty"` // §4.9 — written, never analysed + // Reasons counts this poll's non-empty instance reasons, e.g. + // {"NoData":1091,"Error":14}; nil when none. Reporting-only, and the ONLY + // place composite states stay visible: they are canonical normal (so they + // are dropped from Abnormal) and `totals` never carries composite keys. + // + // The KEYS are raw reason strings and can be comma-joined composites + // ("KeepLast, MissingSeries") — newer Grafana versions join several + // reasons into one. So any consumer, P7 check 9's KeepLast note included, + // must test membership across the keys with reasonNames and must NEVER + // index a literal key: reasons["KeepLast"] misses every composite. + Reasons map[string]int `json:"reasons,omitempty"` + // Abnormal holds the instances whose CANONICAL state is not normal + // (§4.6). "Normal (NoData)" and "Normal (Error)" are canonical normal and + // are deliberately not retained here (P1.2a). + Abnormal []Instance `json:"abnormal,omitempty"` + // Cleared and Vanished are instance keys (§4.7): keys that left the + // abnormal set, resolved against the SAME response — a clear and a + // discontinuity are not the same fact (H2). + Cleared []string `json:"cleared,omitempty"` + Vanished []string `json:"vanished,omitempty"` +} + +// Skew is the signed clock skew of this poll (§16). +func (p Poll) Skew() time.Duration { return time.Duration(p.SkewMS) * time.Millisecond } + +// SkewBound is the uncertainty on Skew — the tolerance every cross-domain +// comparison in P7 applies alongside it. +func (p Poll) SkewBound() time.Duration { return time.Duration(p.SkewBoundMS) * time.Millisecond } + +// Latency is the wall time this poll's request took, feeding §5.2's budget check. +func (p Poll) Latency() time.Duration { return time.Duration(p.LatencyMS) * time.Millisecond } + +// Reducer turns each Observation into the single Poll record that goes into +// the log. It holds the previous poll's abnormal instance keys per rule, which +// is all the state the transition markers need (§4.7). +// +// A Reducer is safe for concurrent use: watch polls a fleet of rules +// concurrently (P6) and every one of those goroutines reduces through the same +// instance, because the per-rule marker state has to live in one place. The +// lock is per-Reducer rather than per-rule — Reduce only touches maps and +// slices, so it never blocks on I/O while holding it. +type Reducer struct { + mu sync.Mutex + prevAbnormal map[string]map[string]struct{} +} + +func NewReducer() *Reducer { + return &Reducer{prevAbnormal: make(map[string]map[string]struct{})} +} + +// Reduce selects the rule identified by uid out of obs and reduces it to a +// Poll. Selection is BY UID, never by title: a filtered response can carry +// several rules sharing one title (the known 2-way collision, §14.5), and +// picking the first would silently watch the wrong rule. +// +// The reduction (§4.6) keeps the rule-level fields, the raw totals histogram, +// the reason counts, and only the instances whose canonical state is not +// normal. That makes per-poll size independent of NORMAL cardinality — not of +// cardinality outright: a rule with 449 firing instances still stores all 449. +func (r *Reducer) Reduce(uid string, obs Observation) Poll { + r.mu.Lock() + defer r.mu.Unlock() + + p := Poll{ + RuleUID: uid, + GrafanaNow: obs.GrafanaNow, + SkewMS: obs.Skew.Milliseconds(), + SkewBoundMS: obs.SkewBound.Milliseconds(), + LatencyMS: obs.Latency.Milliseconds(), + } + + var rule *StateRule + for i := range obs.Rules { + if obs.Rules[i].UID == uid { + rule = &obs.Rules[i] + break + } + } + 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 + // with an instance missing, the next poll still reports that instance + // as vanished rather than losing the transition entirely. + return p + } + + p.Found = true + p.State = rule.State + p.Health = rule.Health + p.LastError = rule.LastError + p.LastEvaluation = rule.LastEvaluation + p.IsPaused = rule.IsPaused + p.Histogram = rule.Totals + + // present indexes every instance in THIS response, normal ones included — + // the markers below must resolve a departed key against the same response + // (H2), which is impossible from the abnormal subset alone. + present := make(map[string]Instance, len(rule.Instances)) + curAbnormal := make(map[string]struct{}) + for _, inst := range rule.Instances { + key := instanceKey(inst.Labels) + present[key] = inst + if inst.Reason != "" { + if p.Reasons == nil { + p.Reasons = make(map[string]int) + } + p.Reasons[inst.Reason]++ + } + if inst.State != StateNormal { + p.Abnormal = append(p.Abnormal, inst) + curAbnormal[key] = struct{}{} + } + } + + for key := range r.prevAbnormal[uid] { + if _, still := curAbnormal[key]; still { + continue + } + inst, found := present[key] + switch { + case !found: + // Fully absent from the response: a discontinuity, not a recovery. + p.Vanished = append(p.Vanished, key) + case reasonNames(inst.Reason, missingSeriesReason): + // The vanish in disguise, caught one poll earlier than the fully + // absent case — H2's named bug. + p.Vanished = append(p.Vanished, key) + default: + // Present as canonical normal without a MissingSeries reason. + p.Cleared = append(p.Cleared, key) + } + } + // Map iteration is unordered; sort so a log line is byte-stable for a + // given poll and a golden fixture stays meaningful. + sort.Strings(p.Cleared) + sort.Strings(p.Vanished) + + r.prevAbnormal[uid] = curAbnormal + return p +} + +// 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). +func reasonNames(reason, want string) bool { + for part := range strings.SplitSeq(reason, ",") { + if strings.TrimSpace(part) == want { + return true + } + } + return false +} + +// VerifyNormalInstancesVisible checks §3.2's assumption on a first +// observation: that the state endpoint really does return normal instances, +// not only the abnormal ones. If it ever stops doing so, the reduction's +// "keep the non-normal instances" becomes "keep everything the API happened to +// send" and the transition markers lose their ground truth — a silent +// fail-open. So this is verified at start, never assumed. +// +// The counts are summed over every totals key whose LOWERCASED name is +// "normal" or "inactive". Never index one literal key: the captured +// vocabulary is mixed across rules ({"alerting":445,"normal":2004} on one, +// {"firing":2,"inactive":363} on another) and its case has already drifted +// from the original recon. Composite states never appear in totals — Grafana +// counts a "Normal (NoData)" instance under normal. +func VerifyNormalInstancesVisible(rules []StateRule) error { + for _, r := range rules { + var claimed int + for k, v := range r.Totals { + switch strings.ToLower(k) { + case "normal", "inactive": + claimed += v + } + } + if claimed == 0 { + continue + } + if hasNormalInstance(r.Instances) { + continue + } + return fmt.Errorf( + "rule %q (%s): totals claim %d normal instances but the response returned none — "+ + "the state endpoint no longer returns normal instances, which the §3.2 reduction depends on", + r.Title, r.UID, claimed) + } + return nil +} + +func hasNormalInstance(instances []Instance) bool { + for _, inst := range instances { + if inst.State == StateNormal { + return true + } + } + return false +} + +// headerRecord, pollRecord and stoppedRecord are the three wire shapes. The +// type tag is a real field on each line rather than an envelope, so a human +// (or jq) reading an uploaded log sees flat records. +type headerRecord struct { + Type RecordType `json:"type"` + Header +} + +type pollRecord struct { + Type RecordType `json:"type"` + Poll +} + +type stoppedRecord struct { + Type RecordType `json:"type"` + At time.Time `json:"at"` +} + +// Writer appends records to the JSONL log. It is append-only by construction +// (§8): O_APPEND|O_CREATE|O_WRONLY, never O_TRUNC, so no writer can ever +// destroy evidence a previous one recorded. An exclusive non-blocking flock +// makes a second writer fail immediately rather than interleave. +type Writer struct { + mu sync.Mutex + f *os.File + enc *json.Encoder + clock Clock + stopped bool +} + +// NewWriter opens path for appending and takes the exclusive lock. A second +// writer on the same path fails here, immediately — it never blocks and never +// waits, because two recorders on one log means one of them is recording a +// window nobody will classify. +func NewWriter(path string, clock Clock) (*Writer, error) { + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return nil, fmt.Errorf("open log %s: %w", path, err) + } + if err := lockExclusive(f); err != nil { + f.Close() + return nil, fmt.Errorf("lock log %s: %w (another writer holds it)", path, err) + } + return &Writer{f: f, enc: json.NewEncoder(f), clock: clock}, nil +} + +// WriteHeader writes line 1 and stamps the current schema version, so no +// caller can leave it at zero. It refuses a non-empty file: the log already +// has a header, and a second one would make ReadLog's "header is line 1" +// contract a lie. In the P6 handoff the parent writes the header and the child +// only appends polls. +func (w *Writer) WriteHeader(h Header) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return fmt.Errorf("log writer already stopped") + } + info, err := w.f.Stat() + if err != nil { + return fmt.Errorf("stat log: %w", err) + } + if info.Size() != 0 { + return fmt.Errorf("log %s is not empty: it already has a header", w.f.Name()) + } + h.SchemaVersion = LogSchemaVersion + if err := w.enc.Encode(headerRecord{Type: RecordHeader, Header: h}); err != nil { + return fmt.Errorf("write log header: %w", err) + } + return nil +} + +// WritePoll appends one poll record — the heartbeat. +func (w *Writer) WritePoll(p Poll) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return fmt.Errorf("log writer already stopped") + } + if err := w.enc.Encode(pollRecord{Type: RecordPoll, Poll: p}); err != nil { + return fmt.Errorf("write poll for rule %s: %w", p.RuleUID, err) + } + return nil +} + +// Stop finishes recording in the fixed §4.4 order, which must not be +// reordered: let the in-flight write finish (the mutex), append the stopped +// sentinel, fsync, then release. Any other order can leave a log whose last +// durable byte is a sentinel that was never actually preceded by the polls it +// vouches for. +// +// Stop writes the sentinel with the recorder's OWN stop time and makes no +// comparison against `to` — watch never knows `to` or the transition grace. +// check does that comparison, after this writer has exited (§4.5). +// +// Calling Stop twice is a no-op: watch reaches it from both a signal handler +// and a defer, and a second sentinel would be indistinguishable from a second +// writer. +func (w *Writer) Stop() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return nil + } + w.stopped = true + + encErr := w.enc.Encode(stoppedRecord{Type: RecordStopped, At: w.clock.Now()}) + syncErr := w.f.Sync() + closeErr := w.f.Close() + + switch { + case encErr != nil: + return fmt.Errorf("write stopped sentinel: %w", encErr) + case syncErr != nil: + return fmt.Errorf("fsync log: %w", syncErr) + case closeErr != nil: + return fmt.Errorf("close log: %w", closeErr) + } + return nil +} + +// Close releases the file and the lock WITHOUT writing a sentinel. It exists +// for exactly one caller: watch's parent, which writes the header and then +// hands the log to the detached child that will finish it (P6). A sentinel +// here would tell check the recording ended before the child had even +// started. +func (w *Writer) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return nil + } + w.stopped = true + if err := w.f.Close(); err != nil { + return fmt.Errorf("close log: %w", err) + } + return 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 +// pass). +// +// Call this only after the writer has exited (§4.4 step 4). Reading a log a +// writer can still append to can only produce a shorter window than the one +// that was recorded. +// +// The parse rules are deliberately the crudest possible (§24.2): the header +// must be line 1 with a matching schema version, and ANY unparseable line — +// including the last one, and including a last line that follows a sentinel — +// is an error, full stop. No heuristics, no discarding an untidy tail: a +// truncated log is evidence that something killed the recorder, which is +// exactly what must not pass. +func ReadLog(path string) (Header, []Poll, *time.Time, error) { + b, err := os.ReadFile(path) + if err != nil { + return Header{}, nil, nil, fmt.Errorf("read log %s: %w", path, err) + } + + lines := strings.Split(string(b), "\n") + // A complete record always ends with the encoder's newline, so the split + // leaves one trailing empty element. Drop exactly that one; any other + // empty line stays and fails below as unparseable. + if len(lines) > 0 && lines[len(lines)-1] == "" { + lines = lines[:len(lines)-1] + } + if len(lines) == 0 { + return Header{}, nil, nil, fmt.Errorf("log %s is empty: the header must be line 1", path) + } + + var ( + header Header + polls []Poll + sentinel *time.Time + ) + for i, line := range lines { + var probe struct { + Type RecordType `json:"type"` + } + if err := json.Unmarshal([]byte(line), &probe); err != nil { + return Header{}, nil, nil, fmt.Errorf("log %s line %d: unparseable record: %w", path, i+1, err) + } + if sentinel != nil { + return Header{}, nil, nil, fmt.Errorf( + "log %s line %d: a %q record follows the stopped sentinel — the log had a second writer", + path, i+1, probe.Type) + } + if (i == 0) != (probe.Type == RecordHeader) { + return Header{}, nil, nil, fmt.Errorf( + "log %s line %d: got record type %q; the header must be line 1 and appear only once", + path, i+1, probe.Type) + } + + switch probe.Type { + case RecordHeader: + var rec headerRecord + if err := json.Unmarshal([]byte(line), &rec); err != nil { + return Header{}, nil, nil, fmt.Errorf("log %s line 1: unparseable header: %w", path, err) + } + if rec.SchemaVersion != LogSchemaVersion { + return Header{}, nil, nil, 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) + } + header = rec.Header + case RecordPoll: + var rec pollRecord + if err := json.Unmarshal([]byte(line), &rec); err != nil { + return Header{}, nil, nil, fmt.Errorf("log %s line %d: unparseable poll: %w", path, i+1, err) + } + polls = append(polls, rec.Poll) + case RecordStopped: + var rec stoppedRecord + if err := json.Unmarshal([]byte(line), &rec); err != nil { + return Header{}, nil, nil, fmt.Errorf("log %s line %d: unparseable sentinel: %w", path, i+1, err) + } + at := rec.At + sentinel = &at + default: + return Header{}, nil, nil, fmt.Errorf("log %s line %d: unknown record type %q", path, i+1, probe.Type) + } + } + + return header, polls, sentinel, nil +} diff --git a/grafana-alertcheck/internal/gate/log_test.go b/grafana-alertcheck/internal/gate/log_test.go new file mode 100644 index 000000000..494c9535c --- /dev/null +++ b/grafana-alertcheck/internal/gate/log_test.go @@ -0,0 +1,894 @@ +package gate + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + "time" +) + +// Every time literal in this file is UTC and built with time.Date, so it +// carries no monotonic reading and survives a JSON round trip byte-identical — +// which is what lets the round-trip tests below use reflect.DeepEqual on whole +// Poll values instead of comparing field by field. +var testNow = time.Date(2026, 8, 31, 9, 0, 0, 0, time.UTC) + +func testInstance(state State, reason, instanceLabel string) Instance { + return Instance{ + Labels: map[string]string{"alertname": "Example", "instance": instanceLabel}, + State: state, + Reason: reason, + ActiveAt: testNow, + } +} + +// observation wraps rules into an Observation with plausible timing numbers, +// deliberately not round millisecond values so a lost conversion at the ms +// boundary shows up as a wrong number rather than a coincidentally equal one. +func observation(grafanaNow time.Time, rules ...StateRule) Observation { + return Observation{ + Rules: rules, + GrafanaNow: grafanaNow, + Skew: 1500 * time.Millisecond, + SkewBound: 40 * time.Millisecond, + Latency: 1800 * time.Millisecond, + } +} + +func TestLogReduceKeepsOnlyAbnormalInstances(t *testing.T) { + rule := StateRule{ + UID: "rule1", Title: "Example", Folder: "F", Group: "G", + Interval: time.Minute, State: "firing", Health: "ok", + LastEvaluation: testNow, Totals: map[string]int{"alerting": 1, "normal": 2}, + Instances: []Instance{ + testInstance(StateNormal, "", "a"), + testInstance(StateFiring, "", "b"), + // Both composites are canonical normal (P1.2a): they must NOT be + // retained as abnormal, and their reasons must still be counted. + testInstance(StateNormal, "NoData", "c"), + testInstance(StateNormal, "Error", "d"), + }, + } + + p := NewReducer().Reduce("rule1", observation(testNow, rule)) + + if !p.Found { + t.Fatalf("Found = false, want true") + } + if len(p.Abnormal) != 1 || p.Abnormal[0].Labels["instance"] != "b" { + t.Errorf("Abnormal = %+v, want only the firing instance b", p.Abnormal) + } + if want := map[string]int{"NoData": 1, "Error": 1}; !reflect.DeepEqual(p.Reasons, want) { + t.Errorf("Reasons = %v, want %v", p.Reasons, want) + } + // The histogram is a verbatim copy of the response totals — raw keys, no + // normalization (§4.9). + if want := map[string]int{"alerting": 1, "normal": 2}; !reflect.DeepEqual(p.Histogram, want) { + t.Errorf("Histogram = %v, want %v", p.Histogram, want) + } + // Rule-level state and health stay raw and unnormalized (P1.2a). + if p.State != "firing" || p.Health != "ok" { + t.Errorf("State/Health = %q/%q, want firing/ok", p.State, p.Health) + } + if p.Skew() != 1500*time.Millisecond || p.SkewBound() != 40*time.Millisecond || p.Latency() != 1800*time.Millisecond { + t.Errorf("durations = %s/%s/%s, want 1.5s/40ms/1.8s", p.Skew(), p.SkewBound(), p.Latency()) + } + if p.Reasons["MissingSeries"] != 0 { + t.Errorf("unexpected MissingSeries count") + } +} + +// A filtered response can hold several rules sharing one title (the known +// 2-way collision, §14.5), so the reducer must select by UID. +func TestLogReduceSelectsRuleByUID(t *testing.T) { + first := StateRule{UID: "ruleA", Title: "Same Title", Health: "ok", State: "inactive", LastEvaluation: testNow} + second := StateRule{ + UID: "ruleB", Title: "Same Title", Health: "error", State: "firing", LastEvaluation: testNow, + Instances: []Instance{testInstance(StateFiring, "", "x")}, + } + + p := NewReducer().Reduce("ruleB", observation(testNow, first, second)) + + if p.Health != "error" || len(p.Abnormal) != 1 { + t.Errorf("reduced the wrong rule: %+v", p) + } +} + +func TestLogReduceRuleAbsentIsAuthoritative(t *testing.T) { + other := StateRule{UID: "other", Title: "Other", Health: "ok", State: "inactive", LastEvaluation: testNow} + + p := NewReducer().Reduce("rule1", observation(testNow, other)) + + if p.Found { + t.Errorf("Found = true, want false for a rule absent from an authoritative 2xx") + } + if p.RuleUID != "rule1" { + t.Errorf("RuleUID = %q, want rule1 — an absent rule is still attributed", p.RuleUID) + } + // The heartbeat still exists: a not-found poll is evidence that Grafana + // answered at this time, which the coverage proof reads. + if !p.GrafanaNow.Equal(testNow) || p.Latency() == 0 { + t.Errorf("absent-rule poll lost its timing evidence: %+v", p) + } + if p.Health != "" || p.Abnormal != nil { + t.Errorf("absent-rule poll carries rule fields: %+v", p) + } +} + +// H2: an instance that leaves the abnormal set is resolved against the SAME +// response, and MissingSeries is a vanish, never a recovery. +func TestTransitionMarkersClearedVersusVanished(t *testing.T) { + badKey := instanceKey(testInstance(StateFiring, "", "b").Labels) + + cases := []struct { + name string + second []Instance + wantCleared []string + wantVanished []string + }{ + { + name: "present as canonical normal is a clear", + second: []Instance{testInstance(StateNormal, "", "b")}, + wantCleared: []string{badKey}, + }, + { + name: "fully absent is a discontinuity", + second: nil, + wantVanished: []string{badKey}, + }, + { + name: "Normal (MissingSeries) is the vanish in disguise", + second: []Instance{testInstance(StateNormal, "MissingSeries", "b")}, + wantVanished: []string{badKey}, + }, + { + name: "a comma-joined reason naming MissingSeries still vanishes", + second: []Instance{testInstance(StateNormal, "KeepLast, MissingSeries", "b")}, + wantVanished: []string{badKey}, + }, + { + name: "an unrelated comma-joined reason still clears", + second: []Instance{testInstance(StateNormal, "KeepLast, Updated", "b")}, + wantCleared: []string{badKey}, + }, + { + name: "still abnormal is neither", + second: []Instance{testInstance(StateFiring, "", "b")}, + }, + { + name: "abnormal under a different state, then gone, still vanishes", + second: []Instance{testInstance(StateNormal, "", "unrelated")}, + wantVanished: []string{badKey}, + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + r := NewReducer() + firing := StateRule{ + UID: "rule1", Health: "ok", State: "firing", LastEvaluation: testNow, + Instances: []Instance{testInstance(StateFiring, "", "b")}, + } + first := r.Reduce("rule1", observation(testNow, firing)) + if first.Cleared != nil || first.Vanished != nil { + t.Fatalf("first poll produced markers with no previous poll: %+v", first) + } + + next := StateRule{UID: "rule1", Health: "ok", State: "inactive", LastEvaluation: testNow, Instances: c.second} + p := r.Reduce("rule1", observation(testNow.Add(30*time.Second), next)) + + if !reflect.DeepEqual(p.Cleared, c.wantCleared) { + t.Errorf("Cleared = %q, want %q", p.Cleared, c.wantCleared) + } + if !reflect.DeepEqual(p.Vanished, c.wantVanished) { + t.Errorf("Vanished = %q, want %q", p.Vanished, c.wantVanished) + } + }) + } +} + +// A departed key must be resolved against the response the reducer is holding, +// not against a later one — so a rule that goes absent and comes back with the +// instance missing still reports the vanish rather than losing it. +func TestTransitionMarkersSurviveAnAbsentPoll(t *testing.T) { + r := NewReducer() + firing := StateRule{ + UID: "rule1", Health: "ok", State: "firing", LastEvaluation: testNow, + Instances: []Instance{testInstance(StateFiring, "", "b")}, + } + r.Reduce("rule1", observation(testNow, firing)) + + absent := r.Reduce("rule1", observation(testNow.Add(30*time.Second))) + if absent.Vanished != nil || absent.Cleared != nil { + t.Fatalf("an absent rule produced markers: %+v", absent) + } + + back := StateRule{UID: "rule1", Health: "ok", State: "inactive", LastEvaluation: testNow} + p := r.Reduce("rule1", observation(testNow.Add(60*time.Second), back)) + if len(p.Vanished) != 1 { + t.Errorf("Vanished = %q, want the instance that disappeared across the absent poll", p.Vanished) + } +} + +func TestTransitionMarkersAreSortedAndPerRule(t *testing.T) { + r := NewReducer() + ruleOne := StateRule{ + UID: "rule1", Health: "ok", State: "firing", LastEvaluation: testNow, + Instances: []Instance{ + testInstance(StateFiring, "", "z"), + testInstance(StateFiring, "", "a"), + testInstance(StateFiring, "", "m"), + }, + } + ruleTwo := StateRule{ + UID: "rule2", Health: "ok", State: "firing", LastEvaluation: testNow, + Instances: []Instance{testInstance(StateFiring, "", "q")}, + } + r.Reduce("rule1", observation(testNow, ruleOne, ruleTwo)) + r.Reduce("rule2", observation(testNow, ruleOne, ruleTwo)) + + clearedOne := StateRule{UID: "rule1", Health: "ok", State: "inactive", LastEvaluation: testNow} + p := r.Reduce("rule1", observation(testNow.Add(time.Minute), clearedOne, ruleTwo)) + if len(p.Vanished) != 3 { + t.Fatalf("Vanished = %q, want 3 keys", p.Vanished) + } + for i := 1; i < len(p.Vanished); i++ { + if p.Vanished[i-1] > p.Vanished[i] { + t.Errorf("Vanished is not sorted: %q", p.Vanished) + } + } + // rule2's own abnormal set is untouched by rule1's transitions. + q := r.Reduce("rule2", observation(testNow.Add(time.Minute), clearedOne, ruleTwo)) + if q.Cleared != nil || q.Vanished != nil { + t.Errorf("rule2 picked up rule1's transitions: %+v", q) + } +} + +// §3.2: the reduction depends on the state endpoint returning normal instances. +// If it ever stops, that must fail loudly at start, never be assumed. +func TestLogVerifyNormalInstancesVisible(t *testing.T) { + cases := []struct { + fixture string + wantError bool + }{ + {"state_one_instance.json", false}, + {"state_reason_composite.json", false}, // composites plus one plain Normal + {"state_paused.json", false}, // no totals, no instances + {"state_missing_optional.json", false}, // no totals key at all + {"state_health_error.json", false}, // totals {"error":1} claims no normal + {"state_only_active_instances.json", true}, + } + + for _, c := range cases { + t.Run(c.fixture, func(t *testing.T) { + rules, err := ParseState(readFixture(t, c.fixture)) + if err != nil { + t.Fatalf("ParseState: %v", err) + } + err = VerifyNormalInstancesVisible(rules) + if c.wantError { + if err == nil { + t.Fatalf("VerifyNormalInstancesVisible: want an error, got nil") + } + if !strings.Contains(err.Error(), "§3.2") { + t.Errorf("error does not name §3.2: %v", err) + } + return + } + if err != nil { + t.Fatalf("VerifyNormalInstancesVisible: unexpected error: %v", err) + } + }) + } +} + +// The totals vocabulary is mixed and its case has drifted, so the check sums +// every key that lowercases to normal or inactive rather than indexing one +// literal key. +func TestLogVerifyNormalInstancesVisibleVocabularies(t *testing.T) { + cases := []struct { + name string + totals map[string]int + wantError bool + }{ + {"lowercase normal", map[string]int{"alerting": 1, "normal": 4}, true}, + {"capitalized Normal", map[string]int{"Alerting": 1, "Normal": 4}, true}, + {"rule vocabulary inactive", map[string]int{"firing": 2, "inactive": 363}, true}, + {"no normal claimed", map[string]int{"alerting": 1}, false}, + {"nil totals", nil, false}, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + rules := []StateRule{{ + UID: "rule1", Title: "Example", Totals: c.totals, + Instances: []Instance{testInstance(StateFiring, "", "b")}, + }} + err := VerifyNormalInstancesVisible(rules) + if (err != nil) != c.wantError { + t.Errorf("VerifyNormalInstancesVisible: error = %v, want error = %v", err, c.wantError) + } + }) + } +} + +// The two authorities: the header owns the recording facts (the cadence +// actually used), the ruler API owns the rule facts. Mixing them up is +// fail-open in the faster-override direction, so this pins both. +func TestLogModeCadenceComesFromTheHeader(t *testing.T) { + defs := []Definition{{UID: "rule1", Title: "Example", IntervalSeconds: 300, For: time.Minute}} + + h := testHeader() + h.Rules[0].IntervalSeconds = 300 + h.Rules[0].PollEverySeconds = 5 // an operator override far tighter than the default 150s + + rt, _, err := DeriveTimingsFromLog(h, defs) + if err != nil { + t.Fatalf("DeriveTimingsFromLog: %v", err) + } + got := rt["rule1"] + if got.pollEvery != 5*time.Second { + t.Errorf("pollEvery = %s, want the header's 5s, not the default 150s", got.pollEvery) + } + // maxGap and healthGrace follow the recorded cadence; without this a 250s + // hole in a log recorded at 5s would pass silently. + if got.maxGap != 10*time.Second { + t.Errorf("maxGap = %s, want 10s (2 x the recorded cadence)", got.maxGap) + } + if got.healthGrace != 300*time.Second { + t.Errorf("healthGrace = %s, want 300s (max(maxGap, interval))", got.healthGrace) + } + // evalStaleAfter is a rule fact, so it stays 2 x intervalSeconds from the + // definitions regardless of how often the gate polled. + if got.evalStaleAfter != 600*time.Second { + t.Errorf("evalStaleAfter = %s, want 600s from the definition's interval", got.evalStaleAfter) + } + + // A log that cannot say how often it was written cannot have its coverage + // proved, and neither can one naming a rule that no longer resolves. + missingCadence := testHeader() + missingCadence.Rules[0].PollEverySeconds = 0 + if _, _, err := DeriveTimingsFromLog(missingCadence, defs); err == nil { + t.Errorf("a header with no recorded cadence was accepted") + } + if _, _, err := DeriveTimingsFromLog(testHeader(), nil); err == nil { + t.Errorf("a header naming an unresolvable rule was accepted") + } + + // A duplicated UID must not resolve last-one-wins: the slower duplicate + // would widen maxGap, which is fail-open through log corruption alone. + duplicated := testHeader() + slower := duplicated.Rules[0] + slower.PollEverySeconds = 600 + duplicated.Rules = append(duplicated.Rules, slower) + if _, _, err := DeriveTimingsFromLog(duplicated, defs); err == nil { + t.Errorf("a header naming one rule twice was accepted") + } +} + +// watch polls a fleet concurrently through one Reducer (P6), so the marker +// state it holds per rule must be safe under -race — a latent data race here +// surfaces as a wrong transition, which is the one thing markers exist to get +// right. +func TestLogReduceIsSafeForConcurrentUse(t *testing.T) { + r := NewReducer() + rules := make([]StateRule, 0, 8) + for i := range 8 { + rules = append(rules, StateRule{ + UID: fmt.Sprintf("rule%d", i), Health: "ok", State: "firing", LastEvaluation: testNow, + Instances: []Instance{testInstance(StateFiring, "", "b")}, + }) + } + obs := observation(testNow, rules...) + + var wg sync.WaitGroup + for range 3 { + for _, rule := range rules { + wg.Go(func() { r.Reduce(rule.UID, obs) }) + } + } + wg.Wait() + + // Each rule's abnormal instance never left, so no round may invent a + // transition — the concurrency must not corrupt the per-rule state either. + for _, rule := range rules { + p := r.Reduce(rule.UID, obs) + if p.Cleared != nil || p.Vanished != nil { + t.Errorf("rule %s: markers after concurrent reduction: %+v", rule.UID, p) + } + } +} + +// A not-found poll has no evaluation time, and the artifact is read by humans +// and jq (§21.3) — the zero time must not appear as though it were real. +func TestLogPollOmitsTheZeroEvaluationTime(t *testing.T) { + absent := NewReducer().Reduce("rule1", observation(testNow)) + b, err := json.Marshal(pollRecord{Type: RecordPoll, Poll: absent}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if strings.Contains(string(b), "0001-01-01") { + t.Errorf("a not-found poll wrote the zero time: %s", b) + } + if strings.Contains(string(b), "last_evaluation") { + t.Errorf("a not-found poll wrote last_evaluation at all: %s", b) + } + + // A real evaluation time still round-trips. + found := StateRule{UID: "rule1", Health: "ok", State: "inactive", LastEvaluation: testNow} + p := NewReducer().Reduce("rule1", observation(testNow, found)) + b, err = json.Marshal(pollRecord{Type: RecordPoll, Poll: p}) + if err != nil { + t.Fatalf("marshal: %v", err) + } + var back pollRecord + if err := json.Unmarshal(b, &back); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !back.LastEvaluation.Equal(testNow) { + t.Errorf("last_evaluation = %s, want %s", back.LastEvaluation, testNow) + } +} + +func testHeader() Header { + return Header{ + URL: "https://grafana.example.com", + GrafanaVersion: "13.1.0", + StartedAt: testNow, + Rules: []LoggedRule{{ + UID: "rule1", Title: "Example", Folder: "F", Group: "G", + ForSeconds: 300, IntervalSeconds: 60, NoDataState: "OK", ExecErrState: "OK", + PollEverySeconds: 30, + }}, + } +} + +func newTestWriter(t *testing.T, path string) (*Writer, *fakeClock) { + t.Helper() + clock := newFakeClock(testNow) + w, err := NewWriter(path, clock) + if err != nil { + t.Fatalf("NewWriter: %v", err) + } + return w, clock +} + +func TestWriterReadLogRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, clock := newTestWriter(t, path) + + h := testHeader() + if err := w.WriteHeader(h); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + + r := NewReducer() + firing := StateRule{ + UID: "rule1", Health: "ok", State: "firing", LastError: "", LastEvaluation: testNow, + Totals: map[string]int{"alerting": 1}, + Instances: []Instance{testInstance(StateFiring, "", "b")}, + } + cleared := StateRule{UID: "rule1", Health: "ok", State: "inactive", LastEvaluation: testNow.Add(time.Minute)} + want := []Poll{ + r.Reduce("rule1", observation(testNow, firing)), + r.Reduce("rule1", observation(testNow.Add(time.Minute), cleared)), + } + for _, p := range want { + if err := w.WritePoll(p); err != nil { + t.Fatalf("WritePoll: %v", err) + } + } + + clock.Advance(2 * time.Minute) + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + gotHeader, gotPolls, sentinel, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + h.SchemaVersion = LogSchemaVersion // WriteHeader stamps it + if !reflect.DeepEqual(gotHeader, h) { + t.Errorf("header round trip:\n got %+v\nwant %+v", gotHeader, h) + } + if !reflect.DeepEqual(gotPolls, want) { + t.Errorf("poll round trip:\n got %+v\nwant %+v", gotPolls, want) + } + if sentinel == nil { + t.Fatalf("sentinel is nil after Stop") + } + // Stop stamps the recorder's own stop time and makes no comparison + // against `to` — watch never knows it (§4.5). + if !sentinel.Equal(testNow.Add(2 * time.Minute)) { + t.Errorf("sentinel = %s, want the writer's stop time %s", sentinel, testNow.Add(2*time.Minute)) + } +} + +// §8: the log is append-only. A second run against the same path must never +// destroy the evidence the first one recorded. +func TestWriterAppendsAndNeverTruncates(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.WritePoll(Poll{RuleUID: "rule1", Found: true, GrafanaNow: testNow}); err != nil { + t.Fatalf("WritePoll: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + before, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + + // The P6 handoff: the parent wrote the header and closed; the child + // reopens the same path and appends without a second header. + child, _ := newTestWriter(t, path) + if err := child.WritePoll(Poll{RuleUID: "rule1", Found: true, GrafanaNow: testNow.Add(time.Minute)}); err != nil { + t.Fatalf("child WritePoll: %v", err) + } + if err := child.Stop(); err != nil { + t.Fatalf("child Stop: %v", err) + } + + after, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if !strings.HasPrefix(string(after), string(before)) { + t.Fatalf("reopening the log rewrote earlier records:\n%s", after) + } + _, polls, sentinel, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if len(polls) != 2 || sentinel == nil { + t.Errorf("got %d polls, sentinel %v; want 2 polls and a sentinel", len(polls), sentinel) + } +} + +// Two recorders on one log means one of them is recording a window nobody +// will classify, so the second writer fails immediately — it never blocks. +func TestWriterSecondWriterFails(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + first, _ := newTestWriter(t, path) + defer first.Close() + + done := make(chan error, 1) + go func() { + _, err := NewWriter(path, newFakeClock(testNow)) + done <- err + }() + + select { + case err := <-done: + if err == nil { + t.Fatalf("a second writer took the lock") + } + if !strings.Contains(err.Error(), "another writer") { + t.Errorf("error does not name the conflict: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatalf("the second NewWriter blocked instead of failing immediately") + } +} + +func TestWriterHeaderRefusesANonEmptyLog(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.WriteHeader(testHeader()); err == nil { + t.Fatalf("a second header was accepted") + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + reopened, _ := newTestWriter(t, path) + defer reopened.Close() + if err := reopened.WriteHeader(testHeader()); err == nil { + t.Fatalf("a header was accepted on a non-empty log") + } +} + +func TestSentinelStopIsIdempotentAndLast(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + // watch reaches Stop from both a signal handler and a defer; a second + // sentinel would be indistinguishable from a second writer. + if err := w.Stop(); err != nil { + t.Errorf("second Stop: %v", err) + } + // Nothing may be appended after the sentinel — not even by the same writer. + if err := w.WritePoll(Poll{RuleUID: "rule1"}); err == nil { + t.Errorf("WritePoll after Stop was accepted") + } + + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + lines := strings.Split(strings.TrimSuffix(string(b), "\n"), "\n") + if len(lines) != 2 { + t.Fatalf("got %d lines, want header + one sentinel:\n%s", len(lines), b) + } + if !strings.Contains(lines[1], `"type":"stopped"`) { + t.Errorf("last line is not the sentinel: %s", lines[1]) + } +} + +// Close is the parent's handoff path in P6: a sentinel there would tell check +// the recording ended before the child had even started. +func TestSentinelCloseWritesNone(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + _, polls, sentinel, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if sentinel != nil { + t.Errorf("Close wrote a sentinel: %s", sentinel) + } + if polls != nil { + t.Errorf("polls = %+v, want none", polls) + } +} + +// An unfinished recording reads cleanly with a nil sentinel — ReadLog reports +// the absence and P7 turns it into unobservable. It is never ReadLog's job to +// call that a failure, and never anyone's job to call it a pass. +func TestReadLogWithoutASentinel(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.WritePoll(Poll{RuleUID: "rule1", Found: true, GrafanaNow: testNow}); err != nil { + t.Fatalf("WritePoll: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + _, polls, sentinel, err := ReadLog(path) + if err != nil { + t.Fatalf("ReadLog: %v", err) + } + if sentinel != nil || len(polls) != 1 { + t.Errorf("got %d polls, sentinel %v; want 1 poll and no sentinel", len(polls), sentinel) + } +} + +// The read rules are deliberately the crudest possible (§24.2): any unparseable +// line is an error, full stop — including the last one, and including a last +// line that follows a sentinel. +func TestReadLogRejectsBadLogs(t *testing.T) { + header := func(version int) string { + h := testHeader() + h.SchemaVersion = version + b, err := json.Marshal(headerRecord{Type: RecordHeader, Header: h}) + if err != nil { + t.Fatalf("marshal header: %v", err) + } + return string(b) + } + poll := func() string { + b, err := json.Marshal(pollRecord{Type: RecordPoll, Poll: Poll{RuleUID: "rule1", Found: true, GrafanaNow: testNow}}) + if err != nil { + t.Fatalf("marshal poll: %v", err) + } + return string(b) + } + sentinel := func() string { + b, err := json.Marshal(stoppedRecord{Type: RecordStopped, At: testNow}) + if err != nil { + t.Fatalf("marshal sentinel: %v", err) + } + return string(b) + } + + cases := []struct { + name string + content string + wantIn string + }{ + {"empty file", "", "empty"}, + {"no header", poll() + "\n", "header must be line 1"}, + {"header not first", poll() + "\n" + header(LogSchemaVersion) + "\n", "header must be line 1"}, + {"second header", header(LogSchemaVersion) + "\n" + header(LogSchemaVersion) + "\n", "appear only once"}, + {"wrong schema version", header(2) + "\n" + poll() + "\n", "schema version"}, + { + name: "unparseable last line", + content: header(LogSchemaVersion) + "\n" + poll() + "\n" + `{"type":"poll","rule_ui`, + wantIn: "unparseable", + }, + { + // A preceding sentinel makes no difference: a truncated tail is + // evidence that something killed the recorder. + name: "unparseable line after the sentinel", + content: header(LogSchemaVersion) + "\n" + sentinel() + "\n" + `{"type":"pol`, + wantIn: "unparseable", + }, + { + name: "unparseable middle line", + content: header(LogSchemaVersion) + "\n" + `{"type":` + "\n" + poll() + "\n", + wantIn: "unparseable", + }, + { + name: "empty middle line", + content: header(LogSchemaVersion) + "\n\n" + poll() + "\n", + wantIn: "unparseable", + }, + { + name: "a record after the sentinel", + content: header(LogSchemaVersion) + "\n" + sentinel() + "\n" + poll() + "\n", + wantIn: "second writer", + }, + { + name: "unknown record type", + content: header(LogSchemaVersion) + "\n" + `{"type":"heartbeat"}` + "\n", + wantIn: "unknown record type", + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + if err := os.WriteFile(path, []byte(c.content), 0o600); err != nil { + t.Fatalf("write: %v", err) + } + _, _, _, err := ReadLog(path) + if err == nil { + t.Fatalf("ReadLog: want an error, got nil") + } + if !strings.Contains(err.Error(), c.wantIn) { + t.Errorf("error %q does not contain %q", err, c.wantIn) + } + }) + } +} + +func TestReadLogMissingFile(t *testing.T) { + _, _, _, err := ReadLog(filepath.Join(t.TempDir(), "absent.jsonl")) + if err == nil { + t.Fatalf("ReadLog on a missing log: want an error, got nil") + } +} + +// §22.3: per-poll log size must not grow across polls on a high-cardinality +// rule, and the one firing instance among 2446 must still be attributed by its +// labels. The reduction makes size independent of NORMAL cardinality — the +// firing instances are still stored, which is why a clear shrinks the record. +func TestLogSizeIsFlatAcrossPollsOnAHighCardinalityRule(t *testing.T) { + body := synthesizeHighCardinalityState(t, 1, 2445) + rules, err := ParseState(body) + if err != nil { + t.Fatalf("ParseState: %v", err) + } + if len(rules[0].Instances) != 2446 { + t.Fatalf("got %d instances, want 2446", len(rules[0].Instances)) + } + uid := rules[0].UID + + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + + r := NewReducer() + var sizes []int64 + // Measure from the end of the header line, so sizes[0] is the first poll + // record alone rather than the header plus it. + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + previous := info.Size() + for i := range 5 { + p := r.Reduce(uid, observation(testNow.Add(time.Duration(i)*30*time.Second), rules[0])) + if len(p.Abnormal) != 1 { + t.Fatalf("poll %d: Abnormal = %d instances, want the single firing one", i, len(p.Abnormal)) + } + if got := p.Abnormal[0].Labels["instance"]; got != "alerting-0" { + t.Fatalf("poll %d: the firing instance lost its identity: %q", i, got) + } + if err := w.WritePoll(p); err != nil { + t.Fatalf("WritePoll: %v", err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat: %v", err) + } + sizes = append(sizes, info.Size()-previous) + previous = info.Size() + } + + for i := 1; i < len(sizes); i++ { + if sizes[i] != sizes[0] { + t.Errorf("per-poll size grew across polls: %v", sizes) + } + } + // One firing instance among 2446 costs a few hundred bytes, against the + // ~600 KB the unreduced response carries. + if sizes[0] > 2048 { + t.Errorf("per-poll size %d bytes is not a reduction of a %d-byte response", sizes[0], len(body)) + } + + // When the firing instance clears, the record collapses further and the + // transition is still attributed. + rules[0].Instances[0].State = StateNormal + p := r.Reduce(uid, observation(testNow.Add(5*30*time.Second), rules[0])) + if len(p.Cleared) != 1 || len(p.Abnormal) != 0 { + t.Errorf("cleared poll = %+v, want exactly one cleared key and no abnormal instances", p) + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } +} + +// The log must stay readable by anything that reads JSONL, one flat object per +// line with its type tag — an uploaded artifact (§21.3) is read by humans and +// by jq, not only by ReadLog. +func TestLogRecordsAreFlatOneLineObjects(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + w, _ := newTestWriter(t, path) + if err := w.WriteHeader(testHeader()); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + if err := w.WritePoll(Poll{RuleUID: "rule1", Found: true, GrafanaNow: testNow}); err != nil { + t.Fatalf("WritePoll: %v", err) + } + if err := w.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + lines := strings.Split(strings.TrimSuffix(string(b), "\n"), "\n") + wantTypes := []RecordType{RecordHeader, RecordPoll, RecordStopped} + if len(lines) != len(wantTypes) { + t.Fatalf("got %d lines, want %d:\n%s", len(lines), len(wantTypes), b) + } + for i, line := range lines { + var m map[string]json.RawMessage + if err := json.Unmarshal([]byte(line), &m); err != nil { + t.Fatalf("line %d is not one JSON object: %v", i+1, err) + } + var gotType RecordType + if err := json.Unmarshal(m["type"], &gotType); err != nil { + t.Fatalf("line %d has no type tag: %v", i+1, err) + } + if gotType != wantTypes[i] { + t.Errorf("line %d type = %q, want %q", i+1, gotType, wantTypes[i]) + } + if _, nested := m["header"]; nested { + t.Errorf("line %d wraps its payload instead of being flat: %s", i+1, line) + } + } +} diff --git a/grafana-alertcheck/internal/gate/parse_state.go b/grafana-alertcheck/internal/gate/parse_state.go index ebd30d7a5..8b2464aef 100644 --- a/grafana-alertcheck/internal/gate/parse_state.go +++ b/grafana-alertcheck/internal/gate/parse_state.go @@ -25,12 +25,16 @@ const ( // is the opaque suffix of a "State (Reason)" composite ("" when the API gave a // bare state). Reason is reporting-only except for the H2 MissingSeries routing // done downstream in the log markers. +// +// The json tags are for the JSONL log's abnormal-instance list (P5) only — +// parsing an API response never goes through them, because parseInstance +// decodes field by field through req/opt to keep H1's presence checks explicit. type Instance struct { - Labels map[string]string - State State - Reason string - ActiveAt time.Time - Value string + Labels map[string]string `json:"labels"` + State State `json:"state"` + Reason string `json:"reason,omitempty"` + ActiveAt time.Time `json:"active_at"` + Value string `json:"value,omitempty"` } // StateRule is one rule from the state endpoint diff --git a/grafana-alertcheck/internal/gate/schedule.go b/grafana-alertcheck/internal/gate/schedule.go index bf860ddef..89bcd5b3f 100644 --- a/grafana-alertcheck/internal/gate/schedule.go +++ b/grafana-alertcheck/internal/gate/schedule.go @@ -97,6 +97,67 @@ func DeriveTimings(defs []Definition, override time.Duration) (rules map[string] return rules, deriveGlobalTimings(defs), notes } +// DeriveTimingsFromLog is DeriveTimings' log-mode counterpart, and the two +// authorities of P5 are the whole reason it exists as a separate function. +// pollEvery comes from the header — the cadence the recording ACTUALLY used, +// after any --poll-interval override — and maxGap and healthGrace follow from +// it. Re-deriving pollEvery from defs here would compare gaps recorded at the +// override cadence against thresholds computed from the default: exit 2 on a +// clean window when the override was slower, and, worse, a real recorder gap +// passing silently when it was faster. +// +// evalStaleAfter still comes from defs (2 x intervalSeconds): it is a property +// of the rule's own evaluation cadence and is unaffected by how often the gate +// polled. +// +// Three shapes of header are errors rather than a best-effort derivation, +// because each one would otherwise widen a threshold silently: +// +// - a rule with no matching definition — a log that names a rule nobody can +// resolve cannot have that rule's coverage proved; +// - a non-positive recorded cadence — a log that cannot say how often it was +// written cannot have maxGap derived, and defaulting the cadence would +// prove a window that was never observed; +// - the same UID twice — last-one-wins would take whichever cadence happened +// to be written last, and a slower duplicate widens maxGap. That is a +// fail-open reachable through nothing but log corruption. +// +// It checks only the header-to-defs direction. The opposite direction — a +// resolved definition absent from the header — is NOT this function's to +// judge: it is §19.1 step 3's log-identity validation, and it belongs to P9's +// Check, which is the only caller that knows both sets and can name the +// mismatch. Without that check a definition simply gets no timings entry, and +// a downstream lookup would read a zero maxGap: fail-closed (every gap +// exceeds it) but silent, so P9 must reject the set mismatch by name rather +// than let a rule fail for an unexplained reason. +func DeriveTimingsFromLog(h Header, defs []Definition) (rules map[string]ruleTimings, global globalTimings, err error) { + byUID := make(map[string]Definition, len(defs)) + for _, d := range defs { + byUID[d.UID] = d + } + + rules = make(map[string]ruleTimings, len(h.Rules)) + for _, lr := range h.Rules { + def, ok := byUID[lr.UID] + if !ok { + return nil, globalTimings{}, fmt.Errorf( + "log header names rule %s (%q), which no current definition matches", lr.UID, lr.Title) + } + if _, duplicate := rules[lr.UID]; duplicate { + return nil, globalTimings{}, fmt.Errorf( + "log header names rule %s (%q) twice; its recorded cadence is ambiguous", lr.UID, lr.Title) + } + if lr.PollEverySeconds <= 0 { + return nil, globalTimings{}, fmt.Errorf( + "log header records poll_every_seconds=%v for rule %s (%q); the recorded cadence is required to derive maxGap", + lr.PollEverySeconds, lr.UID, lr.Title) + } + pollEvery := time.Duration(lr.PollEverySeconds * float64(time.Second)) + rules[lr.UID] = newRuleTimings(pollEvery, def.IntervalSeconds) + } + return rules, deriveGlobalTimings(defs), 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