Expire stale firing alerts
Release / build (amd64, darwin) (push) Failing after 12s
Release / build (arm64, darwin) (push) Failing after 11s
Release / build (arm64, linux) (push) Failing after 11s
Release / release (push) Has been skipped
Release / docker (push) Failing after 19s
Release / build (amd64, linux) (push) Failing after 12s
Release / chart (push) Failing after 9s
Release / build (amd64, darwin) (push) Failing after 12s
Release / build (arm64, darwin) (push) Failing after 11s
Release / build (arm64, linux) (push) Failing after 11s
Release / release (push) Has been skipped
Release / docker (push) Failing after 19s
Release / build (amd64, linux) (push) Failing after 12s
Release / chart (push) Failing after 9s
A resolved webhook was the only path out of the firing state, so a
notification that was dropped, silenced, or lost to a restart pinned an
alert as firing forever — Prometheus showed it resolved while
terdut-server kept listing it. The archiver only ever touched resolved
alerts, and both the list and stats queries compared status with plain
equality, so a stale row was indistinguishable from a live one.
A sweeper pass now resolves firing alerts on either of two signals: the
ends_at watermark Alertmanager sets on outgoing firing notifications has
passed (plus a grace period for clock skew), or no webhook has refreshed
the alert within TERDUT_STALE_AFTER (default 6h, above Alertmanager's 4h
repeat_interval). Such alerts get resolution_source = 'expiry',
distinguishing them from a real 'alertmanager' resolve.
Two related webhook bugs fixed alongside:
- The upsert had no ordering guard, so a retried firing notification
arriving after the resolved one resurrected the alert. Payloads for
an older alert instance are now discarded: a stale retry carries the
same startsAt, a genuine re-fire a newer one.
- archived_at was never cleared on re-fire, leaving a re-fired alert
archived and invisible in the default list.
Stats now exclude archived alerts to match the default list view; this
lowers historical firing/resolved totals.
The chart exposes both sweeper durations via sweeper.staleAfter and
sweeper.archiveAfter.
This commit is contained in:
@@ -57,6 +57,12 @@ docker run -p 8080:8080 -v $(pwd)/data:/data \
|
|||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `TERDUT_ADDR` | `:8080` | TCP address to listen on |
|
| `TERDUT_ADDR` | `:8080` | TCP address to listen on |
|
||||||
| `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file |
|
| `TERDUT_DB_PATH` | `terdut.db` | Path to the SQLite database file |
|
||||||
|
| `TERDUT_ARCHIVE_AFTER` | `168h` (7d) | How long a resolved alert stays in the default list before being auto-archived |
|
||||||
|
| `TERDUT_STALE_AFTER` | `6h` | How long a firing alert may go without a refreshing webhook before it is treated as resolved — **must exceed your Alertmanager `repeat_interval`** |
|
||||||
|
|
||||||
|
Durations use Go syntax (`30m`, `12h`, `168h`). An unparseable value falls back to the default.
|
||||||
|
|
||||||
|
In the Helm chart the two sweeper durations are set via `sweeper.staleAfter` and `sweeper.archiveAfter`.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
@@ -77,6 +83,22 @@ route:
|
|||||||
|
|
||||||
The webhook endpoint requires no authentication.
|
The webhook endpoint requires no authentication.
|
||||||
|
|
||||||
|
### Stale alert expiry
|
||||||
|
|
||||||
|
A resolved webhook is the only signal that an alert has stopped firing, so a
|
||||||
|
notification that is dropped, silenced, or lost to a restart would otherwise pin
|
||||||
|
that alert as firing forever. A background sweeper resolves firing alerts that
|
||||||
|
Alertmanager has stopped refreshing, using either signal:
|
||||||
|
|
||||||
|
- the `endsAt` watermark on the last notification has passed, or
|
||||||
|
- no webhook has refreshed the alert within `TERDUT_STALE_AFTER`.
|
||||||
|
|
||||||
|
Alertmanager re-sends firing notifications every `repeat_interval`, which is what
|
||||||
|
keeps a live alert fresh — so `TERDUT_STALE_AFTER` must be comfortably larger
|
||||||
|
than your `repeat_interval` (default 4h), or live alerts will be resolved
|
||||||
|
prematurely. Alerts resolved this way are marked `"resolution_source": "expiry"`
|
||||||
|
to distinguish them from a real Alertmanager resolve (`"alertmanager"`).
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## API reference
|
## API reference
|
||||||
@@ -110,14 +132,21 @@ Authorization: Bearer <api-key>
|
|||||||
|
|
||||||
| Method | Path | Description |
|
| Method | Path | Description |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `GET` | `/api/alerts` | List alerts. Filters: `?status=firing\|resolved`, `?name=`, `?from=YYYY-MM-DD`, `?to=YYYY-MM-DD`, `?limit=` (default 50, max 500) |
|
| `GET` | `/api/alerts` | List alerts. Filters: `?status=firing\|resolved`, `?name=`, `?archived=true`, `?from=YYYY-MM-DD`, `?to=YYYY-MM-DD`, `?limit=` (default 50, max 500) |
|
||||||
| `GET` | `/api/alerts/{id}` | Get single alert |
|
| `GET` | `/api/alerts/{id}` | Get single alert |
|
||||||
| `POST` | `/api/alerts/{id}/acknowledge` | Acknowledge alert (stamps authed user + time) |
|
| `POST` | `/api/alerts/{id}/acknowledge` | Acknowledge alert (stamps authed user + time) |
|
||||||
| `DELETE` | `/api/alerts/{id}/acknowledge` | Clear acknowledgement |
|
| `DELETE` | `/api/alerts/{id}/acknowledge` | Clear acknowledgement |
|
||||||
|
| `POST` | `/api/alerts/{id}/archive` | Archive alert (hides it from the default list) |
|
||||||
|
| `DELETE` | `/api/alerts/{id}/archive` | Un-archive alert |
|
||||||
| `GET` | `/api/alerts/{id}/comments` | List comments (chronological) |
|
| `GET` | `/api/alerts/{id}/comments` | List comments (chronological) |
|
||||||
| `POST` | `/api/alerts/{id}/comments` | Add comment `{"content"}` |
|
| `POST` | `/api/alerts/{id}/comments` | Add comment `{"content"}` |
|
||||||
| `DELETE` | `/api/alerts/{id}/comments/{commentID}` | Delete own comment |
|
| `DELETE` | `/api/alerts/{id}/comments/{commentID}` | Delete own comment |
|
||||||
|
|
||||||
|
Archived alerts are hidden from `GET /api/alerts` unless `?archived=true` is
|
||||||
|
passed. Resolved alerts carry `resolution_source`: `"alertmanager"` for a real
|
||||||
|
resolved webhook, `"expiry"` when the sweeper inferred it (see
|
||||||
|
[Stale alert expiry](#stale-alert-expiry)).
|
||||||
|
|
||||||
### On-call schedule
|
### On-call schedule
|
||||||
|
|
||||||
| Method | Path | Description |
|
| Method | Path | Description |
|
||||||
@@ -129,7 +158,7 @@ Authorization: Bearer <api-key>
|
|||||||
|
|
||||||
### Statistics
|
### Statistics
|
||||||
|
|
||||||
All stat endpoints accept optional `?from=YYYY-MM-DD` and `?to=YYYY-MM-DD` to filter by `received_at`.
|
All stat endpoints accept optional `?from=YYYY-MM-DD` and `?to=YYYY-MM-DD` to filter by `received_at`. Archived alerts are excluded, matching the default alert list.
|
||||||
|
|
||||||
| Method | Path | Description |
|
| Method | Path | Description |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
|
|||||||
@@ -29,6 +29,10 @@ spec:
|
|||||||
value: ":{{ .Values.service.port }}"
|
value: ":{{ .Values.service.port }}"
|
||||||
- name: TERDUT_DB_PATH
|
- name: TERDUT_DB_PATH
|
||||||
value: "/data/terdut.db"
|
value: "/data/terdut.db"
|
||||||
|
- name: TERDUT_STALE_AFTER
|
||||||
|
value: "{{ .Values.sweeper.staleAfter }}"
|
||||||
|
- name: TERDUT_ARCHIVE_AFTER
|
||||||
|
value: "{{ .Values.sweeper.archiveAfter }}"
|
||||||
volumeMounts:
|
volumeMounts:
|
||||||
- name: data
|
- name: data
|
||||||
mountPath: /data
|
mountPath: /data
|
||||||
|
|||||||
@@ -15,6 +15,13 @@ service:
|
|||||||
type: ClusterIP
|
type: ClusterIP
|
||||||
port: 8080
|
port: 8080
|
||||||
|
|
||||||
|
sweeper:
|
||||||
|
# How long a firing alert may go without a refreshing webhook before it is
|
||||||
|
# treated as resolved. Must exceed your Alertmanager repeat_interval.
|
||||||
|
staleAfter: 6h
|
||||||
|
# How long a resolved alert stays in the default list before auto-archiving.
|
||||||
|
archiveAfter: 168h
|
||||||
|
|
||||||
bootstrap:
|
bootstrap:
|
||||||
enabled: true
|
enabled: true
|
||||||
username: admin
|
username: admin
|
||||||
|
|||||||
+1
-1
@@ -41,7 +41,7 @@ func main() {
|
|||||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
go api.StartArchiver(ctx, database, cfg.ArchiveAfter)
|
go api.StartArchiver(ctx, database, cfg.ArchiveAfter, cfg.StaleAfter)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
log.Printf("terdut-server %s listening on %s", version, cfg.Addr)
|
log.Printf("terdut-server %s listening on %s", version, cfg.Addr)
|
||||||
|
|||||||
@@ -8,6 +8,13 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Values for alerts.resolution_source, recording why an alert left the firing
|
||||||
|
// state: a real Alertmanager notification, or inference by the sweeper.
|
||||||
|
const (
|
||||||
|
resolutionAlertmanager = "alertmanager"
|
||||||
|
resolutionExpiry = "expiry"
|
||||||
|
)
|
||||||
|
|
||||||
// amPayload mirrors the Alertmanager webhook v4 payload.
|
// amPayload mirrors the Alertmanager webhook v4 payload.
|
||||||
type amPayload struct {
|
type amPayload struct {
|
||||||
Version string `json:"version"`
|
Version string `json:"version"`
|
||||||
@@ -39,28 +46,51 @@ func handleAlertmanagerWebhook(db *sql.DB) http.HandlerFunc {
|
|||||||
labelsJSON, _ := json.Marshal(a.Labels)
|
labelsJSON, _ := json.Marshal(a.Labels)
|
||||||
annotationsJSON, _ := json.Marshal(a.Annotations)
|
annotationsJSON, _ := json.Marshal(a.Annotations)
|
||||||
|
|
||||||
// Alertmanager uses zero time ("0001-01-01T00:00:00Z") to mean "still firing".
|
// Zero time ("0001-01-01T00:00:00Z") means "no end known" — that is the
|
||||||
|
// convention of Alertmanager's ingest API. Outgoing notifications
|
||||||
|
// normally carry a real future endsAt instead, which is the watermark
|
||||||
|
// the sweeper uses to expire alerts that stop being refreshed.
|
||||||
var endsAtUnix *int64
|
var endsAtUnix *int64
|
||||||
if a.EndsAt.Year() > 1 {
|
if a.EndsAt.Year() > 1 {
|
||||||
t := a.EndsAt.Unix()
|
t := a.EndsAt.Unix()
|
||||||
endsAtUnix = &t
|
endsAtUnix = &t
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var resolutionSource *string
|
||||||
|
if a.Status == "resolved" {
|
||||||
|
s := resolutionAlertmanager
|
||||||
|
resolutionSource = &s
|
||||||
|
}
|
||||||
|
|
||||||
|
// The WHERE clause discards payloads that describe an alert instance
|
||||||
|
// older than the stored one. Alertmanager retries failed notifications,
|
||||||
|
// so a stale firing retry can arrive after the resolved one; it carries
|
||||||
|
// the same startsAt, whereas a genuine re-fire carries a newer one.
|
||||||
|
// Within a single instance, resolution is terminal.
|
||||||
_, err := db.ExecContext(r.Context(), `
|
_, err := db.ExecContext(r.Context(), `
|
||||||
INSERT INTO alerts
|
INSERT INTO alerts
|
||||||
(fingerprint, name, status, labels, annotations, starts_at, ends_at, generator_url, received_at)
|
(fingerprint, name, status, labels, annotations, starts_at, ends_at,
|
||||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
generator_url, received_at, resolution_source)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
ON CONFLICT(fingerprint) DO UPDATE SET
|
ON CONFLICT(fingerprint) DO UPDATE SET
|
||||||
status = excluded.status,
|
status = excluded.status,
|
||||||
labels = excluded.labels,
|
labels = excluded.labels,
|
||||||
annotations = excluded.annotations,
|
annotations = excluded.annotations,
|
||||||
|
starts_at = excluded.starts_at,
|
||||||
ends_at = excluded.ends_at,
|
ends_at = excluded.ends_at,
|
||||||
generator_url = excluded.generator_url,
|
generator_url = excluded.generator_url,
|
||||||
received_at = excluded.received_at`,
|
received_at = excluded.received_at,
|
||||||
|
resolution_source = excluded.resolution_source,
|
||||||
|
-- A re-fire makes the alert current again, so it leaves the archive.
|
||||||
|
archived_at = CASE WHEN excluded.status = 'firing'
|
||||||
|
THEN NULL ELSE alerts.archived_at END
|
||||||
|
WHERE excluded.starts_at > alerts.starts_at
|
||||||
|
OR (excluded.starts_at = alerts.starts_at
|
||||||
|
AND NOT (alerts.status = 'resolved' AND excluded.status = 'firing'))`,
|
||||||
a.Fingerprint, name, a.Status,
|
a.Fingerprint, name, a.Status,
|
||||||
string(labelsJSON), string(annotationsJSON),
|
string(labelsJSON), string(annotationsJSON),
|
||||||
a.StartsAt.Unix(), endsAtUnix,
|
a.StartsAt.Unix(), endsAtUnix,
|
||||||
a.GeneratorURL, now,
|
a.GeneratorURL, now, resolutionSource,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("upsert alert %s: %v", a.Fingerprint, err)
|
log.Printf("upsert alert %s: %v", a.Fingerprint, err)
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ const alertSelectFrom = `
|
|||||||
a.labels, a.annotations,
|
a.labels, a.annotations,
|
||||||
a.starts_at, a.ends_at, a.generator_url, a.received_at,
|
a.starts_at, a.ends_at, a.generator_url, a.received_at,
|
||||||
a.acknowledged_by, a.acknowledged_at, u.username,
|
a.acknowledged_by, a.acknowledged_at, u.username,
|
||||||
a.archived_at
|
a.resolution_source, a.archived_at
|
||||||
FROM alerts a
|
FROM alerts a
|
||||||
LEFT JOIN users u ON u.id = a.acknowledged_by`
|
LEFT JOIN users u ON u.id = a.acknowledged_by`
|
||||||
|
|
||||||
@@ -186,7 +186,7 @@ func scanAlert(s scanner) (models.Alert, error) {
|
|||||||
&startsAtUnix, &endsAtUnix,
|
&startsAtUnix, &endsAtUnix,
|
||||||
&a.GeneratorURL, &receivedAtUnix,
|
&a.GeneratorURL, &receivedAtUnix,
|
||||||
&ackByID, &ackAtUnix, &ackByUser,
|
&ackByID, &ackAtUnix, &ackByUser,
|
||||||
&archivedAtUnix,
|
&a.ResolutionSource, &archivedAtUnix,
|
||||||
); err != nil {
|
); err != nil {
|
||||||
return a, err
|
return a, err
|
||||||
}
|
}
|
||||||
|
|||||||
+211
-2
@@ -2,21 +2,26 @@ package api_test
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"database/sql"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/yeniklas/terdut-server/internal/api"
|
"github.com/yeniklas/terdut-server/internal/api"
|
||||||
"github.com/yeniklas/terdut-server/internal/db"
|
"github.com/yeniklas/terdut-server/internal/db"
|
||||||
)
|
)
|
||||||
|
|
||||||
// ts wraps httptest.Server with a pre-bootstrapped API key.
|
// ts wraps httptest.Server with a pre-bootstrapped API key. db is exposed so
|
||||||
|
// tests can age rows directly — the sweeper's inputs are wall-clock timestamps.
|
||||||
type ts struct {
|
type ts struct {
|
||||||
*httptest.Server
|
*httptest.Server
|
||||||
key string
|
key string
|
||||||
|
db *sql.DB
|
||||||
}
|
}
|
||||||
|
|
||||||
func newTS(t *testing.T) *ts {
|
func newTS(t *testing.T) *ts {
|
||||||
@@ -44,7 +49,27 @@ func newTS(t *testing.T) *ts {
|
|||||||
json.NewDecoder(resp.Body).Decode(&result)
|
json.NewDecoder(resp.Body).Decode(&result)
|
||||||
key := result["api_key"].(map[string]any)["key"].(string)
|
key := result["api_key"].(map[string]any)["key"].(string)
|
||||||
|
|
||||||
return &ts{Server: srv, key: key}
|
return &ts{Server: srv, key: key, db: database}
|
||||||
|
}
|
||||||
|
|
||||||
|
// exec runs a statement against the test database.
|
||||||
|
func (s *ts) exec(t *testing.T, query string, args ...any) {
|
||||||
|
t.Helper()
|
||||||
|
if _, err := s.db.Exec(query, args...); err != nil {
|
||||||
|
t.Fatalf("exec %q: %v", query, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// alertRow reads the sweeper-relevant columns of one alert straight from the DB.
|
||||||
|
func (s *ts) alertRow(t *testing.T, fingerprint string) (status string, source *string, archivedAt *int64) {
|
||||||
|
t.Helper()
|
||||||
|
err := s.db.QueryRow(
|
||||||
|
"SELECT status, resolution_source, archived_at FROM alerts WHERE fingerprint = ?",
|
||||||
|
fingerprint).Scan(&status, &source, &archivedAt)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("read alert %s: %v", fingerprint, err)
|
||||||
|
}
|
||||||
|
return status, source, archivedAt
|
||||||
}
|
}
|
||||||
|
|
||||||
// req sends an authenticated request, optionally with a JSON body.
|
// req sends an authenticated request, optionally with a JSON body.
|
||||||
@@ -417,6 +442,190 @@ func TestArchive_RoundTrip(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Stale-alert expiry
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
// noArchive is long enough that archiving never interferes with expiry tests.
|
||||||
|
const noArchive = 365 * 24 * time.Hour
|
||||||
|
|
||||||
|
// zeroTime is Alertmanager's "no end known" sentinel, which stores ends_at NULL.
|
||||||
|
const zeroTime = "0001-01-01T00:00:00Z"
|
||||||
|
|
||||||
|
// postAlert sends a single-alert webhook.
|
||||||
|
func postAlert(t *testing.T, s *ts, fingerprint, status, startsAt, endsAt string) {
|
||||||
|
t.Helper()
|
||||||
|
postWebhook(t, s, []map[string]any{{
|
||||||
|
"status": status,
|
||||||
|
"labels": map[string]string{"alertname": "Stale"},
|
||||||
|
"annotations": map[string]string{},
|
||||||
|
"startsAt": startsAt,
|
||||||
|
"endsAt": endsAt,
|
||||||
|
"generatorURL": "",
|
||||||
|
"fingerprint": fingerprint,
|
||||||
|
}})
|
||||||
|
}
|
||||||
|
|
||||||
|
func sweep(t *testing.T, s *ts, staleAfter time.Duration) {
|
||||||
|
t.Helper()
|
||||||
|
api.Sweep(context.Background(), s.db, noArchive, staleAfter)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A firing alert Alertmanager stopped refreshing is resolved via the
|
||||||
|
// received_at heartbeat, even with no ends_at watermark to go on.
|
||||||
|
func TestExpiry_StaleFiringAlert(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
postAlert(t, s, "stale1", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime)
|
||||||
|
|
||||||
|
// Age the last-seen timestamp past the staleness window.
|
||||||
|
s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'stale1'",
|
||||||
|
time.Now().Add(-10*time.Hour).Unix())
|
||||||
|
|
||||||
|
sweep(t, s, 6*time.Hour)
|
||||||
|
|
||||||
|
status, source, _ := s.alertRow(t, "stale1")
|
||||||
|
if status != "resolved" {
|
||||||
|
t.Errorf("expected status resolved, got %q", status)
|
||||||
|
}
|
||||||
|
if source == nil || *source != "expiry" {
|
||||||
|
t.Errorf("expected resolution_source=expiry, got %v", source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A fresh webhook whose ends_at watermark has already passed is expired without
|
||||||
|
// waiting out the full staleness window.
|
||||||
|
func TestExpiry_PastEndsAt(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
postAlert(t, s, "stale2", "firing",
|
||||||
|
time.Now().Add(-2*time.Hour).Format(time.RFC3339),
|
||||||
|
time.Now().Add(-30*time.Minute).Format(time.RFC3339))
|
||||||
|
|
||||||
|
sweep(t, s, 6*time.Hour) // received_at is fresh; only ends_at can trigger
|
||||||
|
|
||||||
|
status, source, _ := s.alertRow(t, "stale2")
|
||||||
|
if status != "resolved" {
|
||||||
|
t.Errorf("expected status resolved, got %q", status)
|
||||||
|
}
|
||||||
|
if source == nil || *source != "expiry" {
|
||||||
|
t.Errorf("expected resolution_source=expiry, got %v", source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The regression that matters most: a genuinely firing alert must survive a
|
||||||
|
// sweep untouched.
|
||||||
|
func TestExpiry_LeavesFreshAlertsAlone(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
postAlert(t, s, "fresh1", "firing",
|
||||||
|
time.Now().Add(-10*time.Minute).Format(time.RFC3339),
|
||||||
|
time.Now().Add(1*time.Hour).Format(time.RFC3339))
|
||||||
|
|
||||||
|
sweep(t, s, 6*time.Hour)
|
||||||
|
|
||||||
|
status, source, _ := s.alertRow(t, "fresh1")
|
||||||
|
if status != "firing" {
|
||||||
|
t.Errorf("expected fresh alert to stay firing, got %q", status)
|
||||||
|
}
|
||||||
|
if source != nil {
|
||||||
|
t.Errorf("expected no resolution_source, got %q", *source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// An ends_at only just past must not trip expiry — that grace absorbs clock skew.
|
||||||
|
func TestExpiry_RespectsGraceOnEndsAt(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
postAlert(t, s, "grace1", "firing",
|
||||||
|
time.Now().Add(-time.Hour).Format(time.RFC3339),
|
||||||
|
time.Now().Add(-1*time.Minute).Format(time.RFC3339))
|
||||||
|
|
||||||
|
sweep(t, s, 6*time.Hour)
|
||||||
|
|
||||||
|
if status, _, _ := s.alertRow(t, "grace1"); status != "firing" {
|
||||||
|
t.Errorf("expected alert within grace period to stay firing, got %q", status)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Webhook resolution bookkeeping
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
func TestWebhook_ResolvedSetsSource(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
start := time.Now().Add(-time.Hour).Format(time.RFC3339)
|
||||||
|
postAlert(t, s, "src1", "firing", start, zeroTime)
|
||||||
|
|
||||||
|
if _, source, _ := s.alertRow(t, "src1"); source != nil {
|
||||||
|
t.Errorf("expected firing alert to have no resolution_source, got %q", *source)
|
||||||
|
}
|
||||||
|
|
||||||
|
postAlert(t, s, "src1", "resolved", start, time.Now().Format(time.RFC3339))
|
||||||
|
|
||||||
|
status, source, _ := s.alertRow(t, "src1")
|
||||||
|
if status != "resolved" {
|
||||||
|
t.Errorf("expected status resolved, got %q", status)
|
||||||
|
}
|
||||||
|
if source == nil || *source != "alertmanager" {
|
||||||
|
t.Errorf("expected resolution_source=alertmanager, got %v", source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A re-fire under the same fingerprint must leave the archive and clear the
|
||||||
|
// stale expiry marker, otherwise the alert stays invisible in the default list.
|
||||||
|
func TestWebhook_RefireUnarchivesAndClearsSource(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
postAlert(t, s, "refire1", "firing", time.Now().Add(-24*time.Hour).Format(time.RFC3339), zeroTime)
|
||||||
|
|
||||||
|
// Expire it, then archive it.
|
||||||
|
s.exec(t, "UPDATE alerts SET received_at = ? WHERE fingerprint = 'refire1'",
|
||||||
|
time.Now().Add(-10*time.Hour).Unix())
|
||||||
|
sweep(t, s, 6*time.Hour)
|
||||||
|
s.exec(t, "UPDATE alerts SET archived_at = unixepoch() WHERE fingerprint = 'refire1'")
|
||||||
|
|
||||||
|
var alerts []map[string]any
|
||||||
|
decode(t, s.req(t, http.MethodGet, "/api/alerts", nil), &alerts)
|
||||||
|
if len(alerts) != 0 {
|
||||||
|
t.Fatalf("expected archived alert to be hidden, got %d", len(alerts))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fires again: a new alert instance, so a newer startsAt.
|
||||||
|
postAlert(t, s, "refire1", "firing", time.Now().Format(time.RFC3339), zeroTime)
|
||||||
|
|
||||||
|
status, source, archivedAt := s.alertRow(t, "refire1")
|
||||||
|
if status != "firing" {
|
||||||
|
t.Errorf("expected status firing after re-fire, got %q", status)
|
||||||
|
}
|
||||||
|
if source != nil {
|
||||||
|
t.Errorf("expected resolution_source cleared on re-fire, got %q", *source)
|
||||||
|
}
|
||||||
|
if archivedAt != nil {
|
||||||
|
t.Errorf("expected archived_at cleared on re-fire, got %d", *archivedAt)
|
||||||
|
}
|
||||||
|
|
||||||
|
decode(t, s.req(t, http.MethodGet, "/api/alerts", nil), &alerts)
|
||||||
|
if len(alerts) != 1 {
|
||||||
|
t.Errorf("expected re-fired alert back in default list, got %d", len(alerts))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Alertmanager retries failed notifications, so a firing payload for an
|
||||||
|
// already-resolved instance can arrive late. It must not resurrect the alert.
|
||||||
|
func TestWebhook_IgnoresOutOfOrderRetry(t *testing.T) {
|
||||||
|
s := newTS(t)
|
||||||
|
start := time.Now().Add(-time.Hour).Format(time.RFC3339)
|
||||||
|
end := time.Now().Format(time.RFC3339)
|
||||||
|
|
||||||
|
postAlert(t, s, "ooo1", "firing", start, zeroTime)
|
||||||
|
postAlert(t, s, "ooo1", "resolved", start, end)
|
||||||
|
postAlert(t, s, "ooo1", "firing", start, zeroTime) // stale retry, same instance
|
||||||
|
|
||||||
|
status, source, _ := s.alertRow(t, "ooo1")
|
||||||
|
if status != "resolved" {
|
||||||
|
t.Errorf("expected alert to stay resolved after stale retry, got %q", status)
|
||||||
|
}
|
||||||
|
if source == nil || *source != "alertmanager" {
|
||||||
|
t.Errorf("expected resolution_source=alertmanager, got %v", source)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestStats_ByDayReturnsSevenSlots(t *testing.T) {
|
func TestStats_ByDayReturnsSevenSlots(t *testing.T) {
|
||||||
s := newTS(t)
|
s := newTS(t)
|
||||||
resp := s.req(t, http.MethodGet, "/api/stats/alerts/by-day", nil)
|
resp := s.req(t, http.MethodGet, "/api/stats/alerts/by-day", nil)
|
||||||
|
|||||||
+66
-14
@@ -7,11 +7,74 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
func StartArchiver(ctx context.Context, db *sql.DB, archiveAfter time.Duration) {
|
const (
|
||||||
ticker := time.NewTicker(15 * time.Minute)
|
// sweepInterval is how often the background sweeper runs.
|
||||||
|
sweepInterval = 15 * time.Minute
|
||||||
|
|
||||||
|
// expiryGrace absorbs clock skew and notification latency before an alert
|
||||||
|
// whose ends_at watermark has passed is treated as stale.
|
||||||
|
expiryGrace = 5 * time.Minute
|
||||||
|
)
|
||||||
|
|
||||||
|
// StartArchiver runs the alert sweeper until ctx is cancelled, starting with an
|
||||||
|
// immediate pass so a restart reconciles state right away.
|
||||||
|
func StartArchiver(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Duration) {
|
||||||
|
ticker := time.NewTicker(sweepInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
run := func() {
|
Sweep(ctx, db, archiveAfter, staleAfter)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ticker.C:
|
||||||
|
Sweep(ctx, db, archiveAfter, staleAfter)
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Sweep runs a single pass: expire stale firing alerts, then archive resolved
|
||||||
|
// ones. Expiry runs first so an alert can expire and be archived in one pass.
|
||||||
|
// Exported so tests can drive a pass without waiting on the ticker.
|
||||||
|
func Sweep(ctx context.Context, db *sql.DB, archiveAfter, staleAfter time.Duration) {
|
||||||
|
expireStale(ctx, db, staleAfter)
|
||||||
|
archiveResolved(ctx, db, archiveAfter)
|
||||||
|
}
|
||||||
|
|
||||||
|
// expireStale resolves firing alerts that Alertmanager has stopped refreshing.
|
||||||
|
//
|
||||||
|
// A resolved webhook is otherwise the only way out of the firing state, so a
|
||||||
|
// notification that is dropped, silenced, or lost to a restart would pin the
|
||||||
|
// alert as firing forever. Two independent signals mark an alert stale:
|
||||||
|
//
|
||||||
|
// - ends_at, the "valid until" watermark Alertmanager sets on outgoing firing
|
||||||
|
// notifications, has passed (plus expiryGrace for clock skew). Absent on
|
||||||
|
// rows whose payload carried no ends_at, hence the second signal.
|
||||||
|
// - received_at is older than staleAfter. Alertmanager re-sends firing
|
||||||
|
// notifications every repeat_interval, making received_at a liveness
|
||||||
|
// heartbeat — provided staleAfter exceeds that interval.
|
||||||
|
func expireStale(ctx context.Context, db *sql.DB, staleAfter time.Duration) {
|
||||||
|
now := time.Now()
|
||||||
|
res, err := db.ExecContext(ctx, `
|
||||||
|
UPDATE alerts
|
||||||
|
SET status = 'resolved',
|
||||||
|
resolution_source = ?,
|
||||||
|
ends_at = COALESCE(ends_at, unixepoch())
|
||||||
|
WHERE status = 'firing'
|
||||||
|
AND archived_at IS NULL
|
||||||
|
AND ((ends_at IS NOT NULL AND ends_at < ?) OR received_at < ?)`,
|
||||||
|
resolutionExpiry, now.Add(-expiryGrace).Unix(), now.Add(-staleAfter).Unix())
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("sweeper: expire stale: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if n, _ := res.RowsAffected(); n > 0 {
|
||||||
|
log.Printf("sweeper: expired %d stale firing alert(s)", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// archiveResolved hides resolved alerts that have been settled for archiveAfter.
|
||||||
|
func archiveResolved(ctx context.Context, db *sql.DB, archiveAfter time.Duration) {
|
||||||
cutoff := time.Now().Add(-archiveAfter).Unix()
|
cutoff := time.Now().Add(-archiveAfter).Unix()
|
||||||
res, err := db.ExecContext(ctx,
|
res, err := db.ExecContext(ctx,
|
||||||
`UPDATE alerts SET archived_at = unixepoch()
|
`UPDATE alerts SET archived_at = unixepoch()
|
||||||
@@ -26,14 +89,3 @@ func StartArchiver(ctx context.Context, db *sql.DB, archiveAfter time.Duration)
|
|||||||
log.Printf("archiver: archived %d resolved alert(s)", n)
|
log.Printf("archiver: archived %d resolved alert(s)", n)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
run()
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-ticker.C:
|
|
||||||
run()
|
|
||||||
case <-ctx.Done():
|
|
||||||
return
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -160,8 +160,9 @@ func handleStatsByDay(db *sql.DB) http.HandlerFunc {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// statsFilter builds a WHERE clause and args from optional ?from and ?to query params.
|
// statsFilter builds a WHERE clause and args from optional ?from and ?to query params.
|
||||||
|
// Archived alerts are always excluded, matching the default GET /api/alerts view.
|
||||||
func statsFilter(q url.Values) (where string, args []any) {
|
func statsFilter(q url.Values) (where string, args []any) {
|
||||||
clauses := []string{}
|
clauses := []string{"archived_at IS NULL"}
|
||||||
if from := q.Get("from"); from != "" {
|
if from := q.Get("from"); from != "" {
|
||||||
if t, err := time.Parse("2006-01-02", from); err == nil {
|
if t, err := time.Parse("2006-01-02", from); err == nil {
|
||||||
clauses = append(clauses, "received_at >= ?")
|
clauses = append(clauses, "received_at >= ?")
|
||||||
@@ -174,8 +175,5 @@ func statsFilter(q url.Values) (where string, args []any) {
|
|||||||
args = append(args, t.UTC().AddDate(0, 0, 1).Unix())
|
args = append(args, t.UTC().AddDate(0, 0, 1).Unix())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if len(clauses) == 0 {
|
|
||||||
return "1=1", args
|
|
||||||
}
|
|
||||||
return strings.Join(clauses, " AND "), args
|
return strings.Join(clauses, " AND "), args
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,6 +9,11 @@ type Config struct {
|
|||||||
Addr string
|
Addr string
|
||||||
DBPath string
|
DBPath string
|
||||||
ArchiveAfter time.Duration
|
ArchiveAfter time.Duration
|
||||||
|
|
||||||
|
// StaleAfter is how long a firing alert may go without a refreshing webhook
|
||||||
|
// before the sweeper treats it as resolved. It must exceed Alertmanager's
|
||||||
|
// repeat_interval (default 4h), which is what refreshes the alert.
|
||||||
|
StaleAfter time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
func Load() Config {
|
func Load() Config {
|
||||||
@@ -26,5 +31,11 @@ func Load() Config {
|
|||||||
archiveAfter = d
|
archiveAfter = d
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return Config{Addr: addr, DBPath: dbPath, ArchiveAfter: archiveAfter}
|
staleAfter := 6 * time.Hour
|
||||||
|
if s := os.Getenv("TERDUT_STALE_AFTER"); s != "" {
|
||||||
|
if d, err := time.ParseDuration(s); err == nil {
|
||||||
|
staleAfter = d
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return Config{Addr: addr, DBPath: dbPath, ArchiveAfter: archiveAfter, StaleAfter: staleAfter}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,4 @@
|
|||||||
|
-- Records why an alert left the firing state: 'alertmanager' when a resolved
|
||||||
|
-- webhook set it, 'expiry' when the sweeper inferred it from staleness.
|
||||||
|
-- NULL for firing alerts and for rows that predate this migration.
|
||||||
|
ALTER TABLE alerts ADD COLUMN resolution_source TEXT;
|
||||||
@@ -19,5 +19,10 @@ type Alert struct {
|
|||||||
AcknowledgedByUser *string `json:"acknowledged_by,omitempty"`
|
AcknowledgedByUser *string `json:"acknowledged_by,omitempty"`
|
||||||
AcknowledgedAt *time.Time `json:"acknowledged_at,omitempty"`
|
AcknowledgedAt *time.Time `json:"acknowledged_at,omitempty"`
|
||||||
|
|
||||||
|
// ResolutionSource records why a resolved alert left the firing state:
|
||||||
|
// "alertmanager" for a real resolved webhook, "expiry" when the sweeper
|
||||||
|
// inferred it after the alert stopped being refreshed.
|
||||||
|
ResolutionSource *string `json:"resolution_source,omitempty"`
|
||||||
|
|
||||||
ArchivedAt *time.Time `json:"archived_at,omitempty"`
|
ArchivedAt *time.Time `json:"archived_at,omitempty"`
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user