From 968efb14f1f6b59855ed34730cd908f0baf59565 Mon Sep 17 00:00:00 2001 From: Bartek Tofel Date: Mon, 31 Aug 2026 13:43:36 +0200 Subject: [PATCH 1/2] chore: implement phase 2 Fix retry-error conflation, measure full poll latency, and harden Source test doubles for concurrency. --- grafana-alertcheck/internal/gate/source.go | 373 ++++++++++++ .../internal/gate/source_fake_test.go | 142 +++++ .../internal/gate/source_test.go | 559 ++++++++++++++++++ 3 files changed, 1074 insertions(+) create mode 100644 grafana-alertcheck/internal/gate/source.go create mode 100644 grafana-alertcheck/internal/gate/source_fake_test.go create mode 100644 grafana-alertcheck/internal/gate/source_test.go diff --git a/grafana-alertcheck/internal/gate/source.go b/grafana-alertcheck/internal/gate/source.go new file mode 100644 index 000000000..46de818bb --- /dev/null +++ b/grafana-alertcheck/internal/gate/source.go @@ -0,0 +1,373 @@ +package gate + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "math/rand/v2" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +// skewHardLimit is one of §5's filled-in values (basis: §16; §22.11 asserts +// 120s errors, 30s does not). It belongs in schedule.go's named-constants +// block once P4 exists; defined here because P2 needs it first. +const skewHardLimit = 60 * time.Second + +// Clock is the seam that lets tests advance time without sleeping (§22) — the +// only two operations the gate ever needs from a clock. +type Clock interface { + Now() time.Time + After(d time.Duration) <-chan time.Time +} + +// SystemClock is the production Clock: the real wall clock. +type SystemClock struct{} + +func (SystemClock) Now() time.Time { return time.Now() } +func (SystemClock) After(d time.Duration) <-chan time.Time { return time.After(d) } + +// Observation is one successful poll of the state endpoint for a single rule. +type Observation struct { + Rules []StateRule // may be empty — an authoritative 2xx saying the rule is absent (§14.5) + GrafanaNow time.Time // the Date header — H4 + Skew time.Duration // serverDate - (t_send+t_headers)/2, signed (§16) + SkewBound time.Duration // (t_headers-t_send)/2 — RTT/2 to the response headers + Latency time.Duration // t_send through the full body read — see requestResult.Latency +} + +// TransportError marks a failure worth retrying: a non-2xx response, a +// network failure, or a body that failed to parse. It is never a deleted rule +// (an authoritative 2xx with no matching rule is not this) and never a clock +// problem (a missing/unparseable Date header or an out-of-bounds skew is a +// hard error instead — see doRequest). Never conflate them (§14.5). +type TransportError struct { + Err error + Status int // 0 when the failure never got a status (network/transport failure) +} + +func (e *TransportError) Error() string { + if e.Status != 0 { + return fmt.Sprintf("transport error: status %d: %v", e.Status, e.Err) + } + return fmt.Sprintf("transport error: %v", e.Err) +} + +func (e *TransportError) Unwrap() error { return e.Err } + +// RetryExhaustedError is what retryTransport returns once it gives up after +// too many sequential *TransportError failures. It deliberately does not +// implement Unwrap into the underlying *TransportError: once retries are +// exhausted the result is a hard, terminal failure, and +// errors.AsType[*TransportError] must never re-classify it as retryable — +// that is the exact conflation §19.3 case 1 forbids. Cause is still exposed +// as a plain field (and folded into Error()'s text) so a caller can log or +// inspect it; it just cannot flow back into the retry classification. +type RetryExhaustedError struct { + Failures int + Cause error +} + +func (e *RetryExhaustedError) Error() string { + return fmt.Sprintf("gave up after %d sequential failures: %v", e.Failures, e.Cause) +} + +// Source is everything the gate reads from Grafana. httpSource is the one +// production implementation; every later phase's tests use a scripted fake +// (source_fake_test.go) instead of real HTTP. +type Source interface { + Version(ctx context.Context) (string, error) + Definitions(ctx context.Context) ([]Definition, error) + RuleState(ctx context.Context, title string) (Observation, error) +} + +// grafanaVersion is a parsed major.minor.patch triple. +type grafanaVersion struct{ major, minor, patch int } + +func (v grafanaVersion) String() string { return fmt.Sprintf("%d.%d.%d", v.major, v.minor, v.patch) } + +// ord encodes the triple as a single comparable integer. Safe as long as +// minor and patch stay under 1000, true of every real Grafana version. +func (v grafanaVersion) ord() int64 { + return int64(v.major)*1_000_000 + int64(v.minor)*1_000 + int64(v.patch) +} + +func parseGrafanaVersion(s string) (grafanaVersion, error) { + s = strings.TrimSpace(s) + if s == "" { + return grafanaVersion{}, errors.New("empty version string") + } + // Grafana's /api/health always reports exactly three components + // (health.json: "13.1.0"). Require all three explicitly rather than + // defaulting missing ones to zero or silently dropping extras — either + // would accept a value ("13", "13.1.0.5") that was never actually seen + // and never verified against. + parts := strings.Split(s, ".") + if len(parts) != 3 { + return grafanaVersion{}, fmt.Errorf("unparseable version %q: want exactly 3 dot-separated components, got %d", s, len(parts)) + } + var v grafanaVersion + fields := [3]*int{&v.major, &v.minor, &v.patch} + for i, field := range fields { + // Trim any trailing non-digit suffix (prerelease/build metadata, e.g. + // "0+security") rather than requiring an exact numeric match. + digits := parts[i] + j := 0 + for j < len(digits) && digits[j] >= '0' && digits[j] <= '9' { + j++ + } + if j == 0 { + return grafanaVersion{}, fmt.Errorf("unparseable version %q", s) + } + n, err := strconv.Atoi(digits[:j]) + if err != nil { + return grafanaVersion{}, fmt.Errorf("unparseable version %q: %w", s, err) + } + *field = n + } + return v, nil +} + +// supportedGrafanaMin and supportedGrafanaMax bound the platform this gate is +// verified against (§2.7 control 2, §21.5): >= 13.0.0, < 14.0.0. +var ( + supportedGrafanaMin = grafanaVersion{13, 0, 0} + supportedGrafanaMax = grafanaVersion{14, 0, 0} // exclusive +) + +// CheckGrafanaVersion enforces the supported range. An unparseable or +// out-of-range version is a hard error naming both what was found and what is +// supported — trusting an unverified schema is exactly the deprecation risk +// §2.7 control 2 exists to catch. +func CheckGrafanaVersion(version string) error { + v, err := parseGrafanaVersion(version) + if err != nil { + return fmt.Errorf("grafana version %q: %w (supported: >=%s, <%s)", + version, err, supportedGrafanaMin, supportedGrafanaMax) + } + if v.ord() < supportedGrafanaMin.ord() || v.ord() >= supportedGrafanaMax.ord() { + return fmt.Errorf("unsupported grafana version %q (supported: >=%s, <%s)", + version, supportedGrafanaMin, supportedGrafanaMax) + } + return nil +} + +// httpSource is the production Source: stdlib net/http only, bearer auth +// from a token supplied at construction (the caller reads it from the +// environment — §20.2 — this type never touches env itself), and manual +// strict decoding via ParseState/ParseDefinitions (H1). The retry limit and +// backoff parameters are struct fields with production defaults set here, +// not package constants, so a test can shrink them without a hook. +type httpSource struct { + baseURL string + token string + client *http.Client + clock Clock + + maxSequentialFailures int + backoffBase time.Duration + backoffCap time.Duration +} + +// NewHTTPSource builds the production Source. token is never logged and +// never enters an error string (§20.2) — it is used only to set the +// Authorization header. +func NewHTTPSource(baseURL, token string, clock Clock) Source { + return &httpSource{ + baseURL: strings.TrimSuffix(baseURL, "/"), + token: token, + client: &http.Client{Timeout: 30 * time.Second}, + clock: clock, + maxSequentialFailures: 5, + backoffBase: time.Second, + backoffCap: 30 * time.Second, + } +} + +func (s *httpSource) Version(ctx context.Context) (string, error) { + return retryTransport(ctx, s.clock, s.maxSequentialFailures, s.backoffBase, s.backoffCap, func() (string, error) { + r, err := s.doRequest(ctx, "/api/health") + if err != nil { + return "", err + } + var health struct { + Version string `json:"version"` + } + if err := json.Unmarshal(r.Body, &health); err != nil { + return "", &TransportError{Err: fmt.Errorf("parse /api/health: %w", err)} + } + if health.Version == "" { + return "", &TransportError{Err: errors.New("/api/health: empty version")} + } + return health.Version, nil + }) +} + +func (s *httpSource) Definitions(ctx context.Context) ([]Definition, error) { + return retryTransport(ctx, s.clock, s.maxSequentialFailures, s.backoffBase, s.backoffCap, func() ([]Definition, error) { + r, err := s.doRequest(ctx, "/api/ruler/grafana/api/v1/rules") + if err != nil { + return nil, err + } + defs, parseErr := ParseDefinitions(r.Body) + if parseErr != nil { + return nil, &TransportError{Err: fmt.Errorf("parse ruler definitions: %w", parseErr)} + } + return defs, nil + }) +} + +func (s *httpSource) RuleState(ctx context.Context, title string) (Observation, error) { + path := "/api/prometheus/grafana/api/v1/rules?rule_name=" + url.QueryEscape(title) + return retryTransport(ctx, s.clock, s.maxSequentialFailures, s.backoffBase, s.backoffCap, func() (Observation, error) { + r, err := s.doRequest(ctx, path) + if err != nil { + return Observation{}, err + } + rules, parseErr := ParseState(r.Body) + if parseErr != nil { + // Treated as transient, not a schema break: an unparseable 2xx + // is far more likely a mid-stream hiccup than a permanent shape + // change, and H1's strict parser already turns a real shape + // change into a loud per-field error the moment it's visible. + return Observation{}, &TransportError{Err: fmt.Errorf("parse rule state: %w", parseErr)} + } + return Observation{ + Rules: rules, + GrafanaNow: r.ServerDate, + Skew: r.Skew, + SkewBound: r.SkewBound, + Latency: r.Latency, + }, nil + }) +} + +// requestResult is the outcome of one successful HTTP attempt in doRequest: +// the raw body plus everything derived from timing the round trip against +// the response's own clock (§16). +type requestResult struct { + Body []byte + ServerDate time.Time // the Date header — H4 + Skew time.Duration // serverDate - (t_send+t_headers)/2, signed + SkewBound time.Duration // (t_headers-t_send)/2 — RTT/2 to the response headers + // Latency spans t_send through the full body read (§5.2's budget check + // needs the whole poll's wall time, or a schedule feasibility check that + // only sees header latency goes optimistic — fail-open). It does not + // include the caller's subsequent JSON parse (ParseState/ParseDefinitions + // run outside doRequest); if P4's budget accounting needs parse time + // folded in too, extend here rather than approximating it at the call + // site. + Latency time.Duration +} + +// doRequest performs one HTTP GET and classifies the outcome (§14.5, §16): +// a network failure, a non-2xx status, or a body-read failure is retryable +// (*TransportError); a missing or unparseable Date header, or a skew beyond +// skewHardLimit, is a hard error — retrying can never fix either, so neither +// may enter the backoff loop (H4). +// +// The Date-header/skew check runs for every endpoint this hits, including +// /api/health — broader than §16's own scope, which only discusses the state +// endpoint. Deliberate: a skewed clock discovered only once RuleState starts +// polling is a skew that has already masked whatever /api/health and the +// ruler read reported; failing closed at the first response catches it +// before any of that is trusted, and every response comes with a Date header +// for free. +func (s *httpSource) doRequest(ctx context.Context, path string) (requestResult, error) { + req, buildErr := http.NewRequestWithContext(ctx, http.MethodGet, s.baseURL+path, nil) + if buildErr != nil { + return requestResult{}, fmt.Errorf("build request for %s: %w", path, buildErr) + } + if s.token != "" { + req.Header.Set("Authorization", "Bearer "+s.token) + } + + tSend := s.clock.Now() + resp, doErr := s.client.Do(req) + tHeaders := s.clock.Now() + if doErr != nil { + return requestResult{}, &TransportError{Err: doErr} + } + defer resp.Body.Close() + + b, readErr := io.ReadAll(resp.Body) + tBodyRead := s.clock.Now() + if readErr != nil { + return requestResult{}, &TransportError{Err: fmt.Errorf("read response body (status %d): %w", resp.StatusCode, readErr)} + } + latency := tBodyRead.Sub(tSend) + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return requestResult{}, &TransportError{Err: fmt.Errorf("unexpected status %d", resp.StatusCode), Status: resp.StatusCode} + } + + dateHeader := resp.Header.Get("Date") + if dateHeader == "" { + return requestResult{}, fmt.Errorf("%s: response has no Date header (H4)", path) + } + serverDate, parseErr := http.ParseTime(dateHeader) + if parseErr != nil { + return requestResult{}, fmt.Errorf("%s: unparseable Date header %q: %w", path, dateHeader, parseErr) + } + + bound := tHeaders.Sub(tSend) / 2 + mid := tSend.Add(bound) + signedSkew := serverDate.Sub(mid) + absSkew := signedSkew + if absSkew < 0 { + absSkew = -absSkew + } + if absSkew > skewHardLimit { + return requestResult{}, fmt.Errorf("%s: clock skew %s exceeds hard limit %s (§16)", path, absSkew, skewHardLimit) + } + + return requestResult{Body: b, ServerDate: serverDate, Skew: signedSkew, SkewBound: bound, Latency: latency}, nil +} + +// retryTransport runs fn, retrying with backoff only while it fails with a +// *TransportError — any other error is a hard error and returns immediately, +// never retried. failures counts consecutive *TransportError results; +// exceeding maxFailures gives up with a wrapped hard error (§19.3 case 1). +// The wait between attempts goes through clock.After so a test with a fake +// Clock never sleeps on real time (§22). +func retryTransport[T any](ctx context.Context, clock Clock, maxFailures int, backoffBase, backoffCap time.Duration, fn func() (T, error)) (T, error) { + var zero T + failures := 0 + for { + v, err := fn() + if err == nil { + return v, nil + } + if _, ok := errors.AsType[*TransportError](err); !ok { + return zero, err + } + failures++ + if failures > maxFailures { + return zero, &RetryExhaustedError{Failures: failures, Cause: err} + } + select { + case <-ctx.Done(): + return zero, ctx.Err() + case <-clock.After(backoffDelay(backoffBase, backoffCap, failures)): + } + } +} + +// backoffDelay is 1s base, doubling per failure, capped at maxDelay, with +// ±20% jitter (§5's filled-in value for maxSequentialFailures). +func backoffDelay(base, maxDelay time.Duration, failureCount int) time.Duration { + d := base + for i := 1; i < failureCount && d < maxDelay; i++ { + d *= 2 + } + if d > maxDelay { + d = maxDelay + } + jitter := 0.8 + rand.Float64()*0.4 // [0.8, 1.2] + return time.Duration(float64(d) * jitter) +} diff --git a/grafana-alertcheck/internal/gate/source_fake_test.go b/grafana-alertcheck/internal/gate/source_fake_test.go new file mode 100644 index 000000000..b3ef23cca --- /dev/null +++ b/grafana-alertcheck/internal/gate/source_fake_test.go @@ -0,0 +1,142 @@ +package gate + +import ( + "context" + "fmt" + "sync" + "time" +) + +// fakeClock is a manually-advanced Clock — no test in this package sleeps on +// real time (§22). It is goroutine-safe (a concurrent fleet under -race must +// not trip on the double itself), but After always fires immediately, +// regardless of the requested duration or whether Advance was ever called. +// That is sufficient here: every retry/backoff test in this phase only needs +// to avoid a real sleep. It is NOT sufficient for a test that must prove a +// wait did not fire early — e.g. a P4 scheduler test asserting Due() doesn't +// return a rule before its next-due time. That needs a clock with a real +// waiter list keyed off Advance, which does not exist yet; build it when a +// phase actually needs it rather than guessing its shape now. +type fakeClock struct { + mu sync.Mutex + now time.Time +} + +func newFakeClock(now time.Time) *fakeClock { return &fakeClock{now: now} } + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) Advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.now = c.now.Add(d) +} + +func (c *fakeClock) After(d time.Duration) <-chan time.Time { + c.mu.Lock() + fireAt := c.now.Add(d) + c.mu.Unlock() + ch := make(chan time.Time, 1) + ch <- fireAt + return ch +} + +// steppingClock advances by a fixed step on every Now() call, so a test can +// assert exact latency/skew-bound arithmetic (doRequest's three clock reads +// per attempt) without depending on real wall-clock timing. +type steppingClock struct { + mu sync.Mutex + now time.Time + step time.Duration +} + +func (c *steppingClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + t := c.now + c.now = c.now.Add(c.step) + return t +} + +func (c *steppingClock) After(d time.Duration) <-chan time.Time { + c.mu.Lock() + fireAt := c.now.Add(d) + c.mu.Unlock() + ch := make(chan time.Time, 1) + ch <- fireAt + return ch +} + +// scriptedObservation is one canned (Observation, error) pair a fakeSource +// returns from RuleState, in the order scripted. +type scriptedObservation struct { + obs Observation + err error +} + +// fakeSource is a scripted Source with no HTTP, goroutine-safe so a phase +// that polls several rules concurrently (P6) can share one instance across +// goroutines without tripping -race. P3 through at least P5 can construct +// one directly instead of talking to HTTP; a phase that needs it to behave +// like a live server under concurrent load beyond simple locking should +// verify that assumption rather than take this comment's word for it. +type fakeSource struct { + mu sync.Mutex + + version string + versionErr error + + defs []Definition + defsErr error + + // states maps a rule title to a queue of scripted results, popped one + // per call to RuleState. Once the queue is down to its last entry, that + // entry repeats — so a test can script the interesting transitions and + // let a long collection loop settle into steady state without scripting + // every single poll. + states map[string][]scriptedObservation +} + +func newFakeSource() *fakeSource { + return &fakeSource{states: make(map[string][]scriptedObservation)} +} + +func (f *fakeSource) Version(_ context.Context) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.version, f.versionErr +} + +func (f *fakeSource) Definitions(_ context.Context) ([]Definition, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.defs, f.defsErr +} + +func (f *fakeSource) RuleState(_ context.Context, title string) (Observation, error) { + f.mu.Lock() + defer f.mu.Unlock() + q := f.states[title] + if len(q) == 0 { + return Observation{}, fmt.Errorf("fakeSource: no scripted response for %q", title) + } + next := q[0] + if len(q) > 1 { + f.states[title] = q[1:] + } + return next.obs, next.err +} + +// script appends one scripted (Observation, error) pair to be returned, in +// order, by RuleState(ctx, title). +func (f *fakeSource) script(title string, obs Observation, err error) { + f.mu.Lock() + defer f.mu.Unlock() + f.states[title] = append(f.states[title], scriptedObservation{obs: obs, err: err}) +} + +var _ Source = (*fakeSource)(nil) diff --git a/grafana-alertcheck/internal/gate/source_test.go b/grafana-alertcheck/internal/gate/source_test.go new file mode 100644 index 000000000..41bae2dfa --- /dev/null +++ b/grafana-alertcheck/internal/gate/source_test.go @@ -0,0 +1,559 @@ +package gate + +import ( + "context" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +func healthBody(version string) string { + return fmt.Sprintf(`{"database":"ok","version":%q,"commit":"abc123"}`, version) +} + +func emptyStateBody() string { + return `{"status":"success","data":{"groups":[]}}` +} + +// rawHTTPServer starts an httptest server whose handler hijacks the +// connection and writes exactly the bytes respond returns, bypassing +// net/http's automatic Date-header insertion — the only way to test a +// response with no Date header at all, or a deliberately garbled one. +func rawHTTPServer(t *testing.T, respond func(r *http.Request) []byte) *httptest.Server { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hj, ok := w.(http.Hijacker) + if !ok { + t.Errorf("ResponseWriter does not support Hijacker") + return + } + conn, buf, err := hj.Hijack() + if err != nil { + t.Errorf("hijack: %v", err) + return + } + defer conn.Close() + if _, err := buf.Write(respond(r)); err != nil { + t.Errorf("write raw response: %v", err) + return + } + _ = buf.Flush() + })) + t.Cleanup(srv.Close) + return srv +} + +// rawResponse builds a minimal, fully-controlled HTTP/1.1 response: no +// header net/http would add unasked, in particular no automatic Date. +func rawResponse(status int, statusText string, headers map[string]string, body string) []byte { + out := fmt.Sprintf("HTTP/1.1 %d %s\r\n", status, statusText) + for k, v := range headers { + out += fmt.Sprintf("%s: %s\r\n", k, v) + } + out += fmt.Sprintf("Content-Length: %d\r\n", len(body)) + out += "Connection: close\r\n\r\n" + out += body + return []byte(out) +} + +func TestCheckGrafanaVersion(t *testing.T) { + cases := []struct { + version string + wantErr bool + wantContains []string // required substrings of the error message, per case, when wantErr + }{ + {"13.1.0", false, nil}, + {"13.0.0", false, nil}, + {"13.99.99", false, nil}, + {"12.9.9", true, []string{`"12.9.9"`, "13.0.0", "14.0.0"}}, + {"14.0.0", true, []string{`"14.0.0"`, "13.0.0", "14.0.0"}}, + {"14.1.0", true, []string{`"14.1.0"`, "13.0.0", "14.0.0"}}, + {"not-a-version", true, []string{`"not-a-version"`}}, + {"", true, nil}, + } + for _, c := range cases { + err := CheckGrafanaVersion(c.version) + if c.wantErr && err == nil { + t.Errorf("CheckGrafanaVersion(%q): want error, got nil", c.version) + continue + } + if !c.wantErr && err != nil { + t.Errorf("CheckGrafanaVersion(%q): unexpected error: %v", c.version, err) + continue + } + for _, want := range c.wantContains { + if !strings.Contains(err.Error(), want) { + t.Errorf("CheckGrafanaVersion(%q): error %q does not mention %q (the plan requires naming both what was found and what is supported)", c.version, err.Error(), want) + } + } + } +} + +func TestBackoffDelay(t *testing.T) { + base := time.Second + maxDelay := 30 * time.Second + maxWithJitter := maxDelay + maxDelay/5 + time.Millisecond + for n := 1; n <= 10; n++ { + d := backoffDelay(base, maxDelay, n) + if d <= 0 { + t.Fatalf("backoffDelay(_, _, %d) = %v, want > 0", n, d) + } + if d > maxWithJitter { + t.Fatalf("backoffDelay(_, _, %d) = %v, want <= ~%v", n, d, maxWithJitter) + } + } +} + +func TestHTTPSource_Version_HappyPath(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/health" { + t.Errorf("path = %q, want /api/health", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(healthBody("13.1.0"))) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + v, err := src.Version(context.Background()) + if err != nil { + t.Fatalf("Version(): unexpected error: %v", err) + } + if v != "13.1.0" { + t.Fatalf("Version() = %q, want 13.1.0", v) + } +} + +func TestHTTPSource_Version_NeverLogsToken(t *testing.T) { + const secret = "super-secret-token" + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "Bearer "+secret { + t.Errorf("Authorization = %q, want Bearer %s", got, secret) + } + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, secret, clock) + _, err := src.Version(context.Background()) + if err == nil { + t.Fatalf("Version(): want error, got nil") + } + if strings.Contains(err.Error(), secret) { + t.Fatalf("error %q leaks the token", err.Error()) + } +} + +func TestHTTPSource_RuleState_EmptyIsNotAnError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(emptyStateBody())) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + obs, err := src.RuleState(context.Background(), "Anything") + if err != nil { + t.Fatalf("RuleState(): unexpected error: %v", err) + } + if len(obs.Rules) != 0 { + t.Fatalf("Rules = %+v, want empty (an authoritative 2xx is not a transport error)", obs.Rules) + } + if obs.GrafanaNow.IsZero() { + t.Fatalf("GrafanaNow is zero, want the response's Date header value (H4)") + } +} + +func TestHTTPSource_RuleState_EscapesRuleName(t *testing.T) { + var gotQuery string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotQuery = r.URL.RawQuery + if r.URL.Path != "/api/prometheus/grafana/api/v1/rules" { + t.Errorf("path = %q, want /api/prometheus/grafana/api/v1/rules", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(emptyStateBody())) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + title := "[JD] No Job Proposals & More" + if _, err := src.RuleState(context.Background(), title); err != nil { + t.Fatalf("RuleState(): unexpected error: %v", err) + } + want := "rule_name=" + url.QueryEscape(title) + if gotQuery != want { + t.Fatalf("query = %q, want %q", gotQuery, want) + } +} + +func TestHTTPSource_Definitions_HappyPath(t *testing.T) { + body := readFixture(t, "ruler_rules.json") + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/ruler/grafana/api/v1/rules" { + t.Errorf("path = %q, want /api/ruler/grafana/api/v1/rules", r.URL.Path) + } + if r.URL.RawQuery != "" { + t.Errorf("query = %q, want none — Definitions reads the ruler API unfiltered", r.URL.RawQuery) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write(body) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + defs, err := src.Definitions(context.Background()) + if err != nil { + t.Fatalf("Definitions(): unexpected error: %v", err) + } + if len(defs) == 0 { + t.Fatalf("Definitions(): got 0 definitions from a fixture known to have some") + } +} + +func TestHTTPSource_Skew(t *testing.T) { + cases := []struct { + name string + drift time.Duration + wantErr bool + }{ + {"30s skew is fine", 30 * time.Second, false}, + {"120s skew is a hard error", 120 * time.Second, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + anchor := time.Now() + clock := newFakeClock(anchor) + var calls atomic.Int32 + srv := rawHTTPServer(t, func(r *http.Request) []byte { + calls.Add(1) + date := anchor.Add(c.drift).UTC().Format(http.TimeFormat) + return rawResponse(200, "OK", map[string]string{ + "Content-Type": "application/json", + "Date": date, + }, healthBody("13.1.0")) + }) + src := NewHTTPSource(srv.URL, "", clock) + _, err := src.Version(context.Background()) + if c.wantErr && err == nil { + t.Fatalf("Version(): want error, got nil") + } + if !c.wantErr && err != nil { + t.Fatalf("Version(): unexpected error: %v", err) + } + if c.wantErr && calls.Load() != 1 { + t.Fatalf("calls = %d, want 1 — a skew hard error must never be retried", calls.Load()) + } + }) + } +} + +func TestHTTPSource_MissingDateHeader(t *testing.T) { + var calls atomic.Int32 + srv := rawHTTPServer(t, func(r *http.Request) []byte { + calls.Add(1) + return rawResponse(200, "OK", map[string]string{ + "Content-Type": "application/json", + }, healthBody("13.1.0")) + }) + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + _, err := src.Version(context.Background()) + if err == nil { + t.Fatalf("Version(): want error, got nil (H4: a missing Date header is a hard error)") + } + if calls.Load() != 1 { + t.Fatalf("calls = %d, want 1 — a missing Date header must never be retried", calls.Load()) + } +} + +func TestHTTPSource_UnparseableDateHeader(t *testing.T) { + var calls atomic.Int32 + srv := rawHTTPServer(t, func(r *http.Request) []byte { + calls.Add(1) + return rawResponse(200, "OK", map[string]string{ + "Content-Type": "application/json", + "Date": "definitely not a date", + }, healthBody("13.1.0")) + }) + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + _, err := src.Version(context.Background()) + if err == nil { + t.Fatalf("Version(): want error, got nil (H4: an unparseable Date header is a hard error)") + } + if calls.Load() != 1 { + t.Fatalf("calls = %d, want 1 — an unparseable Date header must never be retried", calls.Load()) + } +} + +// TestHTTPSource_ObservationTiming pins the arithmetic behind Observation's +// three derived fields, not just the hard-limit behavior TestHTTPSource_Skew +// already covers: the sign of Skew, SkewBound as exactly RTT/2 to the +// response headers, and Latency as the full send-through-body-read span +// (not just the header round trip). steppingClock advances by a fixed 2s on +// every clock.Now() call, and doRequest calls Now() exactly three times per +// attempt (before send, after headers, after the body read), so the +// arithmetic is exact rather than a real-time approximation. +func TestHTTPSource_ObservationTiming(t *testing.T) { + cases := []struct { + name string + drift time.Duration + }{ + {"positive skew", 5 * time.Second}, + {"negative skew", -5 * time.Second}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + anchor := time.Now().Truncate(time.Second) + clock := &steppingClock{now: anchor, step: 2 * time.Second} + // With step=2s: t_send=anchor, t_headers=anchor+2s, t_bodyRead=anchor+4s. + // mid = t_send + (t_headers-t_send)/2 = anchor+1s, so serverDate = + // anchor+1s+drift makes Skew land on exactly `drift`. + srv := rawHTTPServer(t, func(r *http.Request) []byte { + date := anchor.Add(time.Second + c.drift).UTC().Format(http.TimeFormat) + return rawResponse(200, "OK", map[string]string{ + "Content-Type": "application/json", + "Date": date, + }, emptyStateBody()) + }) + src := NewHTTPSource(srv.URL, "", clock) + obs, err := src.RuleState(context.Background(), "Anything") + if err != nil { + t.Fatalf("RuleState(): unexpected error: %v", err) + } + if obs.Skew != c.drift { + t.Errorf("Skew = %v, want %v", obs.Skew, c.drift) + } + if obs.SkewBound != time.Second { + t.Errorf("SkewBound = %v, want 1s (RTT/2 with a 2s round trip to headers)", obs.SkewBound) + } + if obs.Latency != 4*time.Second { + t.Errorf("Latency = %v, want 4s (send through full body read, §5.2) — not just the 2s header round trip", obs.Latency) + } + }) + } +} + +func TestHTTPSource_Retry_TransientRecovers(t *testing.T) { + var mu sync.Mutex + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + if n <= 2 { + w.WriteHeader(http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(healthBody("13.1.0"))) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + v, err := src.Version(context.Background()) + if err != nil { + t.Fatalf("Version(): unexpected error after a transient failure: %v", err) + } + if v != "13.1.0" { + t.Fatalf("Version() = %q, want 13.1.0", v) + } + mu.Lock() + n := calls + mu.Unlock() + if n != 3 { + t.Fatalf("calls = %d, want 3 (2 failures + 1 success)", n) + } +} + +func TestHTTPSource_Retry_ExceedsLimit(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.WriteHeader(http.StatusInternalServerError) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + _, err := src.Version(context.Background()) + if err == nil { + t.Fatalf("Version(): want error, got nil") + } + if n := calls.Load(); n != 6 { + t.Fatalf("calls = %d, want 6 (maxSequentialFailures=5 tolerates 5, gives up on the 6th)", n) + } + assertRetryExhausted(t, err, 6) +} + +func TestHTTPSource_RuleState_GarbageBodyRetries(t *testing.T) { + var mu sync.Mutex + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + n := calls + mu.Unlock() + w.Header().Set("Content-Type", "application/json") + if n <= 2 { + // A 2xx with a body that fails ParseState — classified as a + // transient *TransportError (source.go), not a hard schema + // break, so it must retry rather than fail immediately. + _, _ = w.Write([]byte("{not valid json")) + return + } + _, _ = w.Write([]byte(emptyStateBody())) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + obs, err := src.RuleState(context.Background(), "Anything") + if err != nil { + t.Fatalf("RuleState(): unexpected error after a transient garbage body: %v", err) + } + if len(obs.Rules) != 0 { + t.Fatalf("Rules = %+v, want empty", obs.Rules) + } + mu.Lock() + n := calls + mu.Unlock() + if n != 3 { + t.Fatalf("calls = %d, want 3 (2 unparseable bodies + 1 valid one)", n) + } +} + +func TestHTTPSource_Definitions_GarbageBodyGivesUp(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte("{not valid json")) + })) + defer srv.Close() + + clock := newFakeClock(time.Now()) + src := NewHTTPSource(srv.URL, "", clock) + _, err := src.Definitions(context.Background()) + if err == nil { + t.Fatalf("Definitions(): want error, got nil") + } + if n := calls.Load(); n != 6 { + t.Fatalf("calls = %d, want 6 — a persistently unparseable 2xx body retries like any other transport failure", n) + } + assertRetryExhausted(t, err, 6) +} + +func TestHTTPSource_NetworkFailureRetries(t *testing.T) { + // A server that is never listening: every attempt is a network failure, + // classified as *TransportError, so this exercises the same retry path + // as a 5xx without needing a real listening socket per failure. + clock := newFakeClock(time.Now()) + src := NewHTTPSource("http://127.0.0.1:1", "", clock) + _, err := src.Version(context.Background()) + if err == nil { + t.Fatalf("Version(): want error, got nil") + } + assertRetryExhausted(t, err, 6) +} + +// assertRetryExhausted checks the two properties a retry give-up must have: +// it names how many failures it gave up after, and — the regression this +// pins — it is never itself classified as a *TransportError. If it were, +// something one layer up that also retries on *TransportError would treat an +// already-exhausted give-up as retryable again, the exact conflation §19.3 +// case 1 forbids. +func assertRetryExhausted(t *testing.T, err error, wantFailures int) { + t.Helper() + var reErr *RetryExhaustedError + if !errors.As(err, &reErr) { + t.Fatalf("error %v (%T): want a *RetryExhaustedError", err, err) + } + if reErr.Failures != wantFailures { + t.Errorf("RetryExhaustedError.Failures = %d, want %d", reErr.Failures, wantFailures) + } + if !strings.Contains(err.Error(), fmt.Sprintf("gave up after %d", wantFailures)) { + t.Errorf("error %q does not name the failure count", err.Error()) + } + if _, ok := errors.AsType[*TransportError](err); ok { + t.Fatalf("error %v (%T) is classified as *TransportError — an exhausted retry must be a terminal, non-retryable error", err, err) + } +} + +func TestFakeClock(t *testing.T) { + start := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + c := newFakeClock(start) + if !c.Now().Equal(start) { + t.Fatalf("Now() = %v, want %v", c.Now(), start) + } + c.Advance(5 * time.Minute) + want := start.Add(5 * time.Minute) + if !c.Now().Equal(want) { + t.Fatalf("Now() after Advance = %v, want %v", c.Now(), want) + } + + select { + case fired := <-c.After(time.Hour): + if !fired.Equal(want.Add(time.Hour)) { + t.Fatalf("After fired with %v, want %v", fired, want.Add(time.Hour)) + } + default: + t.Fatalf("After(1h) did not fire immediately") + } +} + +func TestFakeSource(t *testing.T) { + f := newFakeSource() + f.version = "13.1.0" + f.defs = []Definition{{UID: "u1", Title: "Rule One"}} + + ctx := context.Background() + if v, err := f.Version(ctx); err != nil || v != "13.1.0" { + t.Fatalf("Version() = (%q, %v), want (13.1.0, nil)", v, err) + } + if defs, err := f.Definitions(ctx); err != nil || len(defs) != 1 { + t.Fatalf("Definitions() = (%v, %v), want one definition", defs, err) + } + + f.script("Rule One", Observation{Rules: []StateRule{{UID: "u1"}}}, nil) + f.script("Rule One", Observation{}, fmt.Errorf("boom")) + f.script("Rule One", Observation{Rules: nil}, nil) + + obs, err := f.RuleState(ctx, "Rule One") + if err != nil || len(obs.Rules) != 1 { + t.Fatalf("RuleState() call 1 = (%v, %v), want one rule, no error", obs, err) + } + if _, err := f.RuleState(ctx, "Rule One"); err == nil { + t.Fatalf("RuleState() call 2: want the scripted error, got nil") + } + obs, err = f.RuleState(ctx, "Rule One") + if err != nil { + t.Fatalf("RuleState() call 3: unexpected error: %v", err) + } + if obs.Rules != nil { + t.Fatalf("RuleState() call 3: Rules = %v, want nil (last script entry, then repeats)", obs.Rules) + } + obs, err = f.RuleState(ctx, "Rule One") + if err != nil || obs.Rules != nil { + t.Fatalf("RuleState() call 4: want the last scripted entry to repeat, got (%v, %v)", obs, err) + } + + if _, err := f.RuleState(ctx, "Unscripted Rule"); err == nil { + t.Fatalf("RuleState() for an unscripted title: want an error, got nil") + } +} From 3894d9b23a7c1c25bd32e010ef20228c678db619 Mon Sep 17 00:00:00 2001 From: Bartek Tofel Date: Tue, 1 Sep 2026 16:17:29 +0200 Subject: [PATCH 2/2] chore: enhance unit tests --- .../internal/gate/duration_test.go | 20 ++++++++++--------- .../internal/gate/source_test.go | 4 +++- 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/grafana-alertcheck/internal/gate/duration_test.go b/grafana-alertcheck/internal/gate/duration_test.go index 73a0c00e1..ba3a1629a 100644 --- a/grafana-alertcheck/internal/gate/duration_test.go +++ b/grafana-alertcheck/internal/gate/duration_test.go @@ -47,15 +47,17 @@ func TestParsePromDuration(t *testing.T) { func TestParsePromDuration_Errors(t *testing.T) { cases := []string{ - "5", // bare number, no unit - "-5m", // negative - "5x", // unknown unit - "30m1h", // ascending order (must be descending) - "1h1h", // duplicate unit - "m", // unit with no number - "1.5h", // fractional number not supported by this grammar - "1 h", // whitespace - "300y", // overflows time.Duration (int64 nanoseconds) — must error, not wrap negative + "5", // bare number, no unit + "-5m", // negative + "5x", // unknown unit + "30m1h", // ascending order (must be descending) + "1h1h", // duplicate unit + "m", // unit with no number + "1.5h", // fractional number not supported by this grammar + "1 h", // whitespace + "300y", // overflows time.Duration (int64 nanoseconds) — must error, not wrap negative + "1w2d3h4m5s6ms7us8ns", // too many units + "carrot", // completely invalid } for _, in := range cases { if _, err := ParsePromDuration(in); err == nil { diff --git a/grafana-alertcheck/internal/gate/source_test.go b/grafana-alertcheck/internal/gate/source_test.go index 41bae2dfa..d25f1f2ec 100644 --- a/grafana-alertcheck/internal/gate/source_test.go +++ b/grafana-alertcheck/internal/gate/source_test.go @@ -229,8 +229,10 @@ func TestHTTPSource_Skew(t *testing.T) { drift time.Duration wantErr bool }{ + {"0s skew is fine", 0 * time.Second, false}, {"30s skew is fine", 30 * time.Second, false}, - {"120s skew is a hard error", 120 * time.Second, true}, + {"60s skew is fine", 60 * time.Second, false}, + {"61s skew is a hard error", 61 * time.Second, true}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) {