diff --git a/README.md b/README.md index 9787731..b1102c4 100644 --- a/README.md +++ b/README.md @@ -377,6 +377,50 @@ exhausts its retries. Written from the result rather than at enqueue, so the timeline says what actually happened — and a page that never landed is visible instead of looking the same as one that did. +### Escalation + +Without a ladder, an unacknowledged incident re-pages the same topic every +`notify_repeat` forever. That is a louder version of the same silence: if the +person on call is asleep, out of signal, or has left, nothing else happens. + +A team can configure an ordered ladder instead. Each level has a timeout and a +set of targets, and a target is either a named person or **whoever the team's +rota says is on call today** — the target that keeps working when the rota +changes and nobody remembers to edit the policy. + +``` +level 1 5m oncall the rota gets first refusal +level 2 5m user:bob then a named second + then repeat_count more rounds + then the team's fallback topic, once +``` + +When a level's timeout passes with the incident still `triggered`, the next +level is paged. Off the end of the ladder the whole thing runs again +`repeat_count` times, and after that the team's `fallback_topic` is paged once +as the end of the line. The incident stays open throughout: running out of +people to wake is not the same as somebody answering. + +**Acknowledging or resolving stops it**, which is the point — continuing to wake +people after somebody has said "I have this" is how a tool teaches people to +mute it. **Snoozing pauses it**: a deliberate "not now" holds the ladder where +it is, and it resumes when the snooze runs out. + +Every step is on the incident's timeline with the level and the names it woke, +so somebody reading it afterwards can tell why their phone rang at 04:00. A +level whose targets are all unreachable — no ntfy topic, a disabled account, an +empty rota — is recorded as `nobody reachable` and the ladder moves on rather +than stalling on a rung that cannot ring. + +**Reminders and escalation never both run.** A team with a ladder gets +escalation; a team without keeps the reminder behaviour exactly as it was. Two +pages for one silence is the surest way to get a tool muted. + +The ladder's `fallback_topic` is per team, unlike `TERDUT_NTFY_FALLBACK_TOPIC`, +which is the install-wide topic used when an incident opens with nobody on call. +They answer different questions: one is "nobody was scheduled", the other is +"everybody scheduled has been tried". + ### Stale alert expiry A resolved webhook is the only signal that an alert has stopped firing, so a @@ -578,6 +622,8 @@ and was removed in v0.13.0 once senders had moved onto keys. | `GET` | `/api/teams/{teamID}/integrations` | member | List integrations. Never returns keys | | `POST` | `/api/teams/{teamID}/integrations` | **owner** | Mint an integration `{"name","kind"}` — key and URL shown once | | `DELETE` | `/api/teams/{teamID}/integrations/{integrationID}` | **owner** | Revoke an integration | +| `GET` | `/api/teams/{teamID}/escalation` | member | The team's [escalation ladder](#escalation) `{repeat_count, fallback_topic, levels[]}`. Empty levels means the team has none | +| `PUT` | `/api/teams/{teamID}/escalation` | **owner** | Replace it wholesale. `400` for a level with no targets or no timeout — a rung that pages nobody is a silence with a number on it | | `GET` | `/api/teams/{teamID}/deadman` | member | The team's [dead man's switch](#dead-mans-switch) configuration `{matchers, timeout_seconds, severity}` | | `PUT` | `/api/teams/{teamID}/deadman` | **owner** | Replace it. `400` when no matcher names an `alertname`, because a switch that silently watches nothing is the failure this feature exists to prevent | diff --git a/internal/api/alertmanager.go b/internal/api/alertmanager.go index 2fd1e8e..a630b77 100644 --- a/internal/api/alertmanager.go +++ b/internal/api/alertmanager.go @@ -403,12 +403,18 @@ func openIncident(ctx context.Context, q querier, notify NotifyConfig, teamID in } } - // Queue the page, but do not send it here: this runs inside a transaction on - // a single-connection pool, so an HTTP call would hold up every other - // request. The notifier picks the row up within a tick. + // Queue the page, but do not send it here: this runs inside the webhook's + // transaction, and an HTTP call would hold a connection open across a + // network round trip. The notifier picks the row up within a tick. if err := enqueueOpened(ctx, q, notify, id, onCall); err != nil { return 0, err } + + // And start the escalation clock, if the team keeps one. In the same + // transaction, so an incident is never briefly open with nobody counting. + if err := startEscalation(ctx, q, id, teamID); err != nil { + return 0, err + } return id, nil } diff --git a/internal/api/escalation.go b/internal/api/escalation.go new file mode 100644 index 0000000..0b20c9e --- /dev/null +++ b/internal/api/escalation.go @@ -0,0 +1,506 @@ +package api + +import ( + "context" + "database/sql" + "log" + "net/http" + "strconv" + "strings" + "time" +) + +// evEscalated records a rung of the ladder on the incident's timeline: which +// level, and who it woke. +const evEscalated = "escalated" + +// escalationPolicy is a team's ladder, loaded whole. It is small — a handful of +// levels with a few targets each — and every use needs all of it, so there is +// no point reading it a level at a time. +type escalationPolicy struct { + teamID int64 + repeatCount int64 + fallbackTopic string + levels []escalationLevel +} + +type escalationLevel struct { + id int64 + position int64 + timeout time.Duration + targets []escalationTarget +} + +type escalationTarget struct { + kind string // "user" or "oncall" + userID *int64 +} + +// configured reports whether this team has anything to escalate through. A +// policy row with no levels is the same as no policy: the team gets the +// pre-escalation behaviour, which is reminders on the assignee's topic. +func (p *escalationPolicy) configured() bool { return p != nil && len(p.levels) > 0 } + +// level returns the level at a 1-based position. +func (p *escalationPolicy) level(pos int64) (escalationLevel, bool) { + for _, l := range p.levels { + if l.position == pos { + return l, true + } + } + return escalationLevel{}, false +} + +// loadEscalationPolicy reads one team's ladder. A team with no policy row +// returns nil, which every caller treats as "not configured" rather than as an +// error: most teams will never set one up. +func loadEscalationPolicy(ctx context.Context, q querier, teamID int64) (*escalationPolicy, error) { + p := &escalationPolicy{teamID: teamID} + err := q.QueryRowContext(ctx, + "SELECT repeat_count, fallback_topic FROM escalation_policies WHERE team_id = $1", + teamID).Scan(&p.repeatCount, &p.fallbackTopic) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + + rows, err := q.QueryContext(ctx, ` + SELECT l.id, l.position, l.timeout_seconds, t.kind, t.user_id + FROM escalation_levels l + LEFT JOIN escalation_targets t ON t.level_id = l.id + WHERE l.team_id = $1 + ORDER BY l.position, t.id`, teamID) + if err != nil { + return nil, err + } + defer rows.Close() + + byPosition := map[int64]int{} // position -> index in p.levels + for rows.Next() { + var id, position, timeout int64 + var kind *string + var userID *int64 + if err := rows.Scan(&id, &position, &timeout, &kind, &userID); err != nil { + return nil, err + } + idx, seen := byPosition[position] + if !seen { + p.levels = append(p.levels, escalationLevel{ + id: id, + position: position, + timeout: time.Duration(timeout) * time.Second, + }) + idx = len(p.levels) - 1 + byPosition[position] = idx + } + // LEFT JOIN: a level with no targets yet still produces a row, with a + // NULL kind. It is a rung that pages nobody, which the API refuses to + // store but an older row could still hold. + if kind != nil { + p.levels[idx].targets = append(p.levels[idx].targets, + escalationTarget{kind: *kind, userID: userID}) + } + } + return p, rows.Err() +} + +// escalate advances every incident whose current level has run out of time. +// +// Runs on the notifier's tick, beside the reminder pass, because it is the same +// question asked differently: reminders ask "has this been ignored long +// enough to say it again", escalation asks "long enough to say it to somebody +// else". Sharing the tick means one query cadence and one outbox. +func escalate(ctx context.Context, db *sql.DB, cfg NotifyConfig) { + rows, err := db.QueryContext(ctx, ` + SELECT i.id, i.team_id, i.escalation_level, i.escalation_level_at, i.escalation_round + FROM incidents i + JOIN escalation_policies p ON p.team_id = i.team_id + WHERE i.resolved_at IS NULL + AND i.archived_at IS NULL + AND i.status = 'triggered' + AND (i.snoozed_until IS NULL OR i.snoozed_until <= $1) + AND i.escalation_level > 0`, time.Now().Unix()) + if err != nil { + log.Printf("escalation: find due: %v", err) + return + } + + type pending struct { + incidentID, teamID, level, round int64 + levelAt int64 + } + var due []pending + for rows.Next() { + var p pending + var levelAt *int64 + if err := rows.Scan(&p.incidentID, &p.teamID, &p.level, &levelAt, &p.round); err != nil { + rows.Close() + log.Printf("escalation: scan: %v", err) + return + } + if levelAt == nil { + continue + } + p.levelAt = *levelAt + due = append(due, p) + } + rows.Close() + if err := rows.Err(); err != nil { + log.Printf("escalation: iterate: %v", err) + return + } + + now := time.Now() + for _, d := range due { + policy, err := loadEscalationPolicy(ctx, db, d.teamID) + if err != nil { + log.Printf("escalation: load policy for team %d: %v", d.teamID, err) + continue + } + if !policy.configured() { + continue + } + current, ok := policy.level(d.level) + if !ok { + continue + } + if now.Sub(time.Unix(d.levelAt, 0)) < current.timeout { + continue + } + if err := advanceEscalation(ctx, db, cfg, policy, d.incidentID, d.level, d.round, now); err != nil { + log.Printf("escalation: advance incident %d: %v", d.incidentID, err) + } + } +} + +// advanceEscalation moves one incident to its next rung, or off the end of the +// ladder. +// +// The whole move is one transaction: the level, the page and the timeline entry +// are one event, and an incident recorded as being at level 3 that nobody at +// level 3 was told about is the worst of the possible half-states. +func advanceEscalation(ctx context.Context, db *sql.DB, cfg NotifyConfig, policy *escalationPolicy, incidentID, level, round int64, now time.Time) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck + + next := level + 1 + nextRound := round + if _, ok := policy.level(next); !ok { + // Off the end. Either start the chain again, or make the last call. + if round < policy.repeatCount { + next, nextRound = 1, round+1 + } else { + if err := escalationExhausted(ctx, tx, policy, incidentID, now); err != nil { + return err + } + return tx.Commit() + } + } + + target, ok := policy.level(next) + if !ok { + return nil + } + paged, err := pageLevel(ctx, tx, cfg, policy, incidentID, target) + if err != nil { + return err + } + + if _, err := tx.ExecContext(ctx, ` + UPDATE incidents + SET escalation_level = $1, escalation_level_at = $2, escalation_round = $3 + WHERE id = $4`, next, now.Unix(), nextRound, incidentID); err != nil { + return err + } + + detail := "level " + strconv.FormatInt(next, 10) + if nextRound > round { + detail += " (round " + strconv.FormatInt(nextRound+1, 10) + ")" + } + if len(paged) > 0 { + detail += ": " + strings.Join(paged, ", ") + } else { + // Worth recording loudly: the rung exists, its turn came, and it woke + // nobody. That is a policy that looks configured and is not. + detail += ": nobody reachable" + } + if err := logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail); err != nil { + return err + } + return tx.Commit() +} + +// escalationExhausted is the end of the line: the fallback topic, once, and a +// timeline entry saying the ladder is finished. The incident stays triggered — +// escalation running out is not the same as somebody answering. +func escalationExhausted(ctx context.Context, tx *sql.Tx, policy *escalationPolicy, incidentID int64, now time.Time) error { + detail := "escalation exhausted" + if policy.fallbackTopic != "" { + if err := enqueueNotification(ctx, tx, incidentID, nil, policy.fallbackTopic, notifyEscalated); err != nil { + return err + } + detail += ": paged " + policy.fallbackTopic + } else { + detail += ": no fallback topic configured" + } + + // Level 0 again, so the sweep stops considering it. The round counter is + // left where it is, as the record of how far it got. + if _, err := tx.ExecContext(ctx, + "UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1", + incidentID); err != nil { + return err + } + return logEvent(ctx, tx, incidentID, evEscalated, nil, nil, &detail) +} + +// pageLevel notifies every target of one level and reports who was woken. +// +// Each target gets its own outbox row, so each gets its own Acknowledge token: +// the button in a notification must acknowledge as the person holding the +// phone, not as whoever was paged first. +func pageLevel(ctx context.Context, tx *sql.Tx, cfg NotifyConfig, policy *escalationPolicy, incidentID int64, level escalationLevel) ([]string, error) { + var paged []string + seen := map[int64]bool{} + + for _, t := range level.targets { + userID := t.userID + if t.kind == "oncall" { + onCall, err := currentOnCall(ctx, tx, policy.teamID) + if err != nil { + return nil, err + } + if onCall == nil { + continue + } + userID = onCall + } + if userID == nil || seen[*userID] { + continue + } + seen[*userID] = true + + var topic *string + var username string + if err := tx.QueryRowContext(ctx, + "SELECT ntfy_topic, username FROM users WHERE id = $1 AND disabled_at IS NULL", + *userID).Scan(&topic, &username); err != nil { + // A disabled or deleted account is not an error in the middle of an + // escalation: it is a target that cannot be woken, and the next + // level is the answer to that. + continue + } + if topic == nil || *topic == "" { + continue + } + if err := enqueueNotification(ctx, tx, incidentID, userID, *topic, notifyEscalated); err != nil { + return nil, err + } + paged = append(paged, username) + } + return paged, nil +} + +// startEscalation puts a newly opened incident on the first rung, when its team +// has a ladder. Called from openIncident, inside the same transaction, so an +// incident is never briefly open with no escalation clock running. +func startEscalation(ctx context.Context, q querier, incidentID, teamID int64) error { + policy, err := loadEscalationPolicy(ctx, q, teamID) + if err != nil || !policy.configured() { + return err + } + _, err = q.ExecContext(ctx, + "UPDATE incidents SET escalation_level = 1, escalation_level_at = $1 WHERE id = $2", + time.Now().Unix(), incidentID) + return err +} + +// stopEscalation takes an incident off the ladder. Acknowledging or resolving +// is somebody saying "I have this", and continuing to wake people after that is +// the behaviour that teaches people to ignore the tool. +func stopEscalation(ctx context.Context, q querier, incidentID int64) error { + _, err := q.ExecContext(ctx, + "UPDATE incidents SET escalation_level = 0, escalation_level_at = NULL WHERE id = $1", + incidentID) + return err +} + +// handleGetEscalation returns a team's ladder. +func handleGetEscalation(db *sql.DB) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + teamID, ok := teamParam(w, r) + if !ok { + return + } + if !requireTeamMember(w, r, teamID) { + return + } + + policy, err := loadEscalationPolicy(r.Context(), db, teamID) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + respond(w, http.StatusOK, escalationResponse(policy, teamID)) + } +} + +type escalationLevelJSON struct { + Position int64 `json:"position"` + TimeoutSeconds int64 `json:"timeout_seconds"` + Targets []escalationTargetJSON `json:"targets"` +} + +type escalationTargetJSON struct { + Kind string `json:"kind"` + UserID *int64 `json:"user_id,omitempty"` +} + +type escalationJSON struct { + TeamID int64 `json:"team_id"` + RepeatCount int64 `json:"repeat_count"` + FallbackTopic string `json:"fallback_topic"` + Levels []escalationLevelJSON `json:"levels"` +} + +func escalationResponse(p *escalationPolicy, teamID int64) escalationJSON { + out := escalationJSON{TeamID: teamID, Levels: []escalationLevelJSON{}} + if p == nil { + return out + } + out.RepeatCount = p.repeatCount + out.FallbackTopic = p.fallbackTopic + for _, l := range p.levels { + level := escalationLevelJSON{ + Position: l.position, + TimeoutSeconds: int64(l.timeout.Seconds()), + Targets: []escalationTargetJSON{}, + } + for _, t := range l.targets { + level.Targets = append(level.Targets, escalationTargetJSON{Kind: t.kind, UserID: t.userID}) + } + out.Levels = append(out.Levels, level) + } + return out +} + +// handleSetEscalation replaces a team's ladder wholesale. +// +// Replace rather than patch: the levels are an order, and an API that edits one +// rung has to answer what happens to the numbering of the others. Sending the +// whole ladder makes the order the client's to decide and the server's to +// store, and makes an edit atomic — there is no moment where level 2 exists +// twice. +func handleSetEscalation(db *sql.DB) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + teamID, ok := teamParam(w, r) + if !ok { + return + } + if !requireTeamOwner(w, r, teamID) { + return + } + + var req escalationJSON + if err := decodeJSON(r, &req); err != nil { + respond(w, http.StatusBadRequest, errResp("invalid request body")) + return + } + if req.RepeatCount < 0 || req.RepeatCount > 10 { + respond(w, http.StatusBadRequest, errResp("repeat_count must be between 0 and 10")) + return + } + for i, l := range req.Levels { + if l.TimeoutSeconds <= 0 { + respond(w, http.StatusBadRequest, errResp("every level needs a timeout")) + return + } + if len(l.Targets) == 0 { + // A rung that pages nobody is not a delay, it is a silence with + // a number on it. + respond(w, http.StatusBadRequest, + errResp("level "+strconv.FormatInt(int64(i+1), 10)+" has no targets")) + return + } + for _, t := range l.Targets { + switch t.Kind { + case "oncall": + if t.UserID != nil { + respond(w, http.StatusBadRequest, errResp("an oncall target takes no user_id")) + return + } + case "user": + if t.UserID == nil { + respond(w, http.StatusBadRequest, errResp("a user target needs a user_id")) + return + } + default: + respond(w, http.StatusBadRequest, errResp("target kind must be user or oncall")) + return + } + } + } + + tx, err := db.BeginTx(r.Context(), nil) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + defer tx.Rollback() //nolint:errcheck + + if _, err := tx.ExecContext(r.Context(), ` + INSERT INTO escalation_policies (team_id, repeat_count, fallback_topic, updated_at) + VALUES ($1, $2, $3, `+nowEpoch+`) + ON CONFLICT (team_id) DO UPDATE SET + repeat_count = excluded.repeat_count, + fallback_topic = excluded.fallback_topic, + updated_at = excluded.updated_at`, + teamID, req.RepeatCount, req.FallbackTopic); err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + // The levels are replaced, not merged; the cascade takes the targets. + if _, err := tx.ExecContext(r.Context(), + "DELETE FROM escalation_levels WHERE team_id = $1", teamID); err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + + for i, l := range req.Levels { + var levelID int64 + if err := tx.QueryRowContext(r.Context(), ` + INSERT INTO escalation_levels (team_id, position, timeout_seconds) + VALUES ($1, $2, $3) RETURNING id`, + teamID, int64(i+1), l.TimeoutSeconds).Scan(&levelID); err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + for _, t := range l.Targets { + if _, err := tx.ExecContext(r.Context(), ` + INSERT INTO escalation_targets (level_id, kind, user_id) + VALUES ($1, $2, $3)`, levelID, t.Kind, t.UserID); err != nil { + // The only foreign key here is the user. + respond(w, http.StatusBadRequest, errResp("unknown user in targets")) + return + } + } + } + + if err := tx.Commit(); err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + + policy, err := loadEscalationPolicy(r.Context(), db, teamID) + if err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } + respond(w, http.StatusOK, escalationResponse(policy, teamID)) + } +} diff --git a/internal/api/escalation_test.go b/internal/api/escalation_test.go new file mode 100644 index 0000000..f1c22f3 --- /dev/null +++ b/internal/api/escalation_test.go @@ -0,0 +1,385 @@ +package api_test + +import ( + "net/http" + "strings" + "testing" + "time" + + "git.ryuvia.com/niklas/terdut-server/internal/api" +) + +// teamUser creates a user in the default team with an ntfy topic, so they can +// actually be paged. +func teamUser(t *testing.T, s *ts, username, topic string) int64 { + t.Helper() + var user struct { + ID int64 `json:"id"` + } + decode(t, s.req(t, http.MethodPost, "/api/users", + map[string]string{"username": username, "email": username + "@test.com"}), &user) + resp := s.req(t, http.MethodPost, "/api/teams/"+defaultTeam+"/members", + map[string]any{"user_id": user.ID, "role": "member"}) + resp.Body.Close() + setTopic(t, s, int(user.ID), topic) + return user.ID +} + +// Escalation is all timeouts, and there is no fake clock in this package. The +// tests back-date escalation_level_at instead, which is the same trick the dead +// man's switch tests use on received_at: the sweeper reads a stored timestamp, +// so moving the timestamp is moving the clock. + +// ladder configures the default team with two levels: the rota first, then a +// named person, then the fallback topic. +func ladder(t *testing.T, s *ts, secondUserID int64, repeat int64, fallback string) { + t.Helper() + resp := s.req(t, http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", map[string]any{ + "repeat_count": repeat, + "fallback_topic": fallback, + "levels": []map[string]any{ + {"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}}, + {"timeout_seconds": 300, "targets": []map[string]any{{"kind": "user", "user_id": secondUserID}}}, + }, + }) + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("configure the ladder: %d", resp.StatusCode) + } +} + +// overdue back-dates an incident's current level so its timeout has passed. +func overdue(t *testing.T, s *ts, incidentID int64) { + t.Helper() + s.exec(t, "UPDATE incidents SET escalation_level_at = $1 WHERE id = $2", + time.Now().Add(-time.Hour).Unix(), incidentID) +} + +func escalationLevel(t *testing.T, s *ts, incidentID int64) (level, round int64) { + t.Helper() + if err := s.db.QueryRow( + "SELECT escalation_level, escalation_round FROM incidents WHERE id = $1", + incidentID).Scan(&level, &round); err != nil { + t.Fatalf("read escalation state: %v", err) + } + return level, round +} + +// The whole point: nobody answers, so somebody else is woken. +func TestEscalation_PagesTheNextLevel(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-esc", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + // Level 1 is the rota, so the first page went to the admin. + if level, _ := escalationLevel(t, s, 1); level != 1 { + t.Fatalf("a new incident should start at level 1, got %d", level) + } + if got := f.topicsSince(t); len(got) == 0 || got[0] != "terdut-admin" { + t.Fatalf("the first page should go to the on-call user, went to %v", got) + } + + // Time passes with no acknowledgement. + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) + + if level, _ := escalationLevel(t, s, 1); level != 2 { + t.Errorf("expected level 2, got %d", level) + } + if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-second" { + t.Errorf("level 2 should page the named user, paged %v", got) + } + + // And the timeline says so, which is what somebody reads afterwards to + // understand why their phone rang at 04:00. + timeline := list(t, s.req(t, http.MethodGet, "/api/incidents/1/timeline", nil)) + found := "" + for _, e := range timeline { + if e["type"] == "escalated" { + found, _ = e["detail"].(string) + } + } + if found == "" { + t.Error("the timeline should record the escalation") + } else if !strings.HasPrefix(found, "level 2") || !strings.Contains(found, "second") { + t.Errorf("the escalation entry should say which level and who: %q", found) + } +} + +// Acknowledging is somebody saying "I have this". Nobody else should be woken. +func TestEscalation_AcknowledgementStopsIt(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-ack", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + s.req(t, http.MethodPost, "/api/incidents/1/acknowledge", nil).Body.Close() + if level, _ := escalationLevel(t, s, 1); level != 0 { + t.Errorf("acknowledging should take the incident off the ladder, level is %d", level) + } + + f.forget() + overdue(t, s, 1) // no-op: level is 0, so there is nothing due + s.sweepNotify(t) + if got := f.topicsSince(t); len(got) != 0 { + t.Errorf("an acknowledged incident should page nobody, paged %v", got) + } +} + +// Resolving stops it too, and by the same mechanism. +func TestEscalation_ResolutionStopsIt(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-res", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + s.req(t, http.MethodPost, "/api/incidents/1/resolve", nil).Body.Close() + + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) + if level, _ := escalationLevel(t, s, 1); level != 0 { + t.Errorf("a resolved incident should be off the ladder, level is %d", level) + } +} + +// Snoozing is a deliberate "not now", so the ladder waits rather than carrying +// on without the person who asked for quiet. +func TestEscalation_SnoozePausesIt(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-snooze", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + resp := s.req(t, http.MethodPost, "/api/incidents/1/snooze", map[string]any{"duration": "1h"}) + resp.Body.Close() + + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) + + if level, _ := escalationLevel(t, s, 1); level != 1 { + t.Errorf("a snoozed incident should stay where it is, level is %d", level) + } + if got := f.topicsSince(t); len(got) != 0 { + t.Errorf("a snoozed incident should page nobody, paged %v", got) + } + + // When the snooze ends, the ladder picks up where it left off. + s.exec(t, "UPDATE incidents SET snoozed_until = $1 WHERE id = 1", time.Now().Add(-time.Minute).Unix()) + s.sweepNotify(t) + if level, _ := escalationLevel(t, s, 1); level != 2 { + t.Errorf("after the snooze the ladder should resume, level is %d", level) + } +} + +// Running out of ladder pages the team's fallback topic once, and says so. +func TestEscalation_ExhaustionPagesTheFallback(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-end", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + overdue(t, s, 1) + s.sweepNotify(t) // level 2 + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) // off the end + + if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-fallback" { + t.Errorf("exhaustion should page the fallback topic once, paged %v", got) + } + level, _ := escalationLevel(t, s, 1) + if level != 0 { + t.Errorf("an exhausted ladder should stop asking, level is %d", level) + } + + // The incident is still open: running out of people is not an answer. + var status string + if err := s.db.QueryRow("SELECT status FROM incidents WHERE id = 1").Scan(&status); err != nil { + t.Fatal(err) + } + if status != "triggered" { + t.Errorf("exhaustion must not resolve the incident, status is %q", status) + } +} + +// repeat_count walks the whole ladder again before giving up. +func TestEscalation_RepeatsTheChain(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 1, "terdut-fallback") // one extra round + + postWebhook(t, s, []map[string]any{ + amAlert("fp-repeat", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + overdue(t, s, 1) + s.sweepNotify(t) // level 2 + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) // back to level 1, round 2 + + level, round := escalationLevel(t, s, 1) + if level != 1 || round != 1 { + t.Errorf("expected level 1 round 1, got level %d round %d", level, round) + } + if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-admin" { + t.Errorf("the second round should start at the top again, paged %v", got) + } +} + +// A team without a ladder keeps exactly the behaviour it had, and never gets +// both a reminder and an escalation for the same silence. +func TestEscalation_WithoutAPolicyRemindersStillRun(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + postWebhook(t, s, []map[string]any{ + amAlert("fp-noesc", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + // Age the first notification past the repeat interval. + f.forget() + s.exec(t, "UPDATE notifications SET created_at = $1, sent_at = $1", + time.Now().Add(-time.Hour).Unix()) + s.sweepNotify(t) + + if got := f.topicsSince(t); len(got) != 1 || got[0] != "terdut-admin" { + t.Errorf("without a ladder the reminder should still fire, paged %v", got) + } + if level, _ := escalationLevel(t, s, 1); level != 0 { + t.Errorf("an incident in a team with no ladder should not be on one, level is %d", level) + } +} + +// With a ladder, reminders stop: two pages for one silence is how people learn +// to mute the tool. +func TestEscalation_WithAPolicyRemindersDoNotAlsoFire(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + second := teamUser(t, s, "second", "terdut-second") + ladder(t, s, second, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-both", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + f.forget() + // Old enough for a reminder, but not yet due for escalation. + s.exec(t, "UPDATE notifications SET created_at = $1, sent_at = $1", + time.Now().Add(-time.Hour).Unix()) + s.sweepNotify(t) + + if got := f.topicsSince(t); len(got) != 0 { + t.Errorf("a team with a ladder should not also get reminders, paged %v", got) + } +} + +// The API refuses a ladder that cannot page anybody. +func TestEscalation_RejectsAnUnusablePolicy(t *testing.T) { + s := newTS(t) + + for _, c := range []struct { + name string + body map[string]any + }{ + {"a level with no targets", map[string]any{ + "levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{}}}, + }}, + {"a level with no timeout", map[string]any{ + "levels": []map[string]any{{"timeout_seconds": 0, "targets": []map[string]any{{"kind": "oncall"}}}}, + }}, + {"a user target with no user", map[string]any{ + "levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "user"}}}}, + }}, + {"an unknown target kind", map[string]any{ + "levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "everybody"}}}}, + }}, + {"an absurd repeat count", map[string]any{ + "repeat_count": 99, + "levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}}}, + }}, + } { + resp := s.req(t, http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", c.body) + resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("%s: expected 400, got %d", c.name, resp.StatusCode) + } + } +} + +// Editing the ladder is an owner's job; reading it is any member's. +func TestEscalation_OwnerOnlyToEdit(t *testing.T) { + s := newTS(t) + _, call := member(t, s, "plain") + + resp := call(http.MethodPut, "/api/teams/"+defaultTeam+"/escalation", map[string]any{ + "levels": []map[string]any{{"timeout_seconds": 300, "targets": []map[string]any{{"kind": "oncall"}}}}, + }) + resp.Body.Close() + if resp.StatusCode != http.StatusForbidden { + t.Errorf("a member editing the ladder: expected 403, got %d", resp.StatusCode) + } + + resp = call(http.MethodGet, "/api/teams/"+defaultTeam+"/escalation", nil) + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("a member reading the ladder: expected 200, got %d", resp.StatusCode) + } +} + +// A target who cannot be woken is not a reason to stop: the next level is the +// answer to an unreachable one. +func TestEscalation_SkipsUnreachableTargets(t *testing.T) { + s, f := notifyTS(t, api.NotifyConfig{PublicURL: "https://terdut.example.com", RepeatEvery: 15 * time.Minute}) + // Second user has no ntfy topic at all. + var user struct { + ID int64 `json:"id"` + } + decode(t, s.req(t, http.MethodPost, "/api/users", + map[string]string{"username": "silent", "email": "silent@test.com"}), &user) + s.req(t, http.MethodPost, "/api/teams/"+defaultTeam+"/members", + map[string]any{"user_id": user.ID, "role": "member"}).Body.Close() + + ladder(t, s, user.ID, 0, "terdut-fallback") + + postWebhook(t, s, []map[string]any{ + amAlert("fp-silent", "DiskFull", "firing", "2026-05-20T10:00:00Z", zeroTime, nil), + }) + s.sweepNotify(t) + + f.forget() + overdue(t, s, 1) + s.sweepNotify(t) + + // Level 2 was entered even though it woke nobody, so the ladder keeps + // moving toward the fallback rather than stalling on a silent rung. + if level, _ := escalationLevel(t, s, 1); level != 2 { + t.Errorf("expected the ladder to advance past an unreachable target, level is %d", level) + } + if got := f.topicsSince(t); len(got) != 0 { + t.Errorf("a target with no topic should page nothing, paged %v", got) + } +} diff --git a/internal/api/incident_store.go b/internal/api/incident_store.go index 63d2d1f..d065ea7 100644 --- a/internal/api/incident_store.go +++ b/internal/api/incident_store.go @@ -235,6 +235,9 @@ func resolveIfSettled(ctx context.Context, q querier, incidentID int64) (bool, e if n == 0 { return false, nil } + if err := stopEscalation(ctx, q, incidentID); err != nil { + return false, err + } if err := logEvent(ctx, q, incidentID, evResolved, nil, nil, nil); err != nil { return false, err } @@ -260,6 +263,10 @@ func acknowledgeIncident(ctx context.Context, q querier, incidentID, userID int6 if n, _ := res.RowsAffected(); n == 0 { return false, nil } + // Somebody has it: stop waking anybody else. + if err := stopEscalation(ctx, q, incidentID); err != nil { + return false, err + } return true, logEvent(ctx, q, incidentID, evAcknowledged, &userID, nil, nil) } diff --git a/internal/api/incidents.go b/internal/api/incidents.go index 756b037..1e353d8 100644 --- a/internal/api/incidents.go +++ b/internal/api/incidents.go @@ -243,6 +243,11 @@ func handleIncidentResolve(db *sql.DB) http.HandlerFunc { time.Now().Unix(), incidentResolutionManual, id) { return } + // A person closing an incident is the clearest possible "I have this". + if err := stopEscalation(r.Context(), db, id); err != nil { + respond(w, http.StatusInternalServerError, errResp("internal error")) + return + } if err := logEvent(r.Context(), db, id, evResolved, &user.ID, nil, nil); err != nil { respond(w, http.StatusInternalServerError, errResp("internal error")) return diff --git a/internal/api/notifier.go b/internal/api/notifier.go index c955263..51fd2b5 100644 --- a/internal/api/notifier.go +++ b/internal/api/notifier.go @@ -43,6 +43,10 @@ const ( notifyTriggered = "triggered" notifyReminder = "reminder" notifyResolved = "resolved" + + // notifyEscalated is a page that went out because nobody answered the last + // one. Told apart from a reminder because it goes to somebody else. + notifyEscalated = "escalated" ) // Timeline event types the notifier writes, so an incident's history says who @@ -120,6 +124,9 @@ func StartNotifier(ctx context.Context, db *sql.DB, cfg NotifyConfig) { // Exported so tests can drive a pass without waiting on the ticker. func NotifySweep(ctx context.Context, db *sql.DB, cfg NotifyConfig) { enqueueReminders(ctx, db, cfg) + // Escalation before delivery, so a level that comes due on this tick is + // paged on this tick rather than waiting for the next one. + escalate(ctx, db, cfg) deliverPending(ctx, db, cfg) } @@ -159,7 +166,12 @@ func enqueueReminders(ctx context.Context, db *sql.DB, cfg NotifyConfig) { AND i.resolved_at IS NULL AND i.archived_at IS NULL AND i.status = 'triggered' - AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2)`, + AND (i.snoozed_until IS NULL OR i.snoozed_until <= $2) + -- A team with an escalation ladder gets escalation instead. Both + -- would mean two pages for one silence, which is how people learn to + -- mute a tool. + AND NOT EXISTS ( + SELECT 1 FROM escalation_levels el WHERE el.team_id = i.team_id)`, now.Add(-repeat).Unix(), now.Unix()) if err != nil { log.Printf("notifier: find reminders: %v", err) diff --git a/internal/api/notify_test.go b/internal/api/notify_test.go index 2abc935..c3b6ef0 100644 --- a/internal/api/notify_test.go +++ b/internal/api/notify_test.go @@ -69,6 +69,27 @@ func (f *fakeNtfy) messages() []pushed { return append([]pushed(nil), f.got...) } +// topicsSince lists the topics published to since the last forget, which is how +// the escalation tests ask "who did this tick wake". +func (f *fakeNtfy) topicsSince(t *testing.T) []string { + t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + out := make([]string, 0, len(f.got)) + for _, m := range f.got { + out = append(out, m.Topic) + } + return out +} + +// forget drops what has been published so far, so the next assertion is about +// this tick rather than the whole test. +func (f *fakeNtfy) forget() { + f.mu.Lock() + defer f.mu.Unlock() + f.got = nil +} + func (f *fakeNtfy) failWith(status int) { f.mu.Lock() defer f.mu.Unlock() diff --git a/internal/api/router.go b/internal/api/router.go index 3dd8fb3..9fe79a4 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -109,6 +109,10 @@ func NewRouter(db *sql.DB, notify NotifyConfig, cfg config.Config) http.Handler r.Post("/api/teams/{teamID}/members", handleAddTeamMember(db)) r.Delete("/api/teams/{teamID}/members/{userID}", handleRemoveTeamMember(db)) + // A team's escalation ladder: who is paged when nobody answers. + r.Get("/api/teams/{teamID}/escalation", handleGetEscalation(db)) + r.Put("/api/teams/{teamID}/escalation", handleSetEscalation(db)) + // A team's own dead man's switches: which of its alerts are heartbeats, // and how long a silence has to last before somebody is paged. r.Get("/api/teams/{teamID}/deadman", handleGetTeamDeadman(db)) diff --git a/internal/db/migrations/006_escalation.sql b/internal/db/migrations/006_escalation.sql new file mode 100644 index 0000000..245567d --- /dev/null +++ b/internal/db/migrations/006_escalation.sql @@ -0,0 +1,95 @@ +-- Escalation: page somebody else when the first person does not answer. +-- +-- This is the gap the whole multi-tenancy line of work was opened to close. +-- Until now an unacknowledged incident re-paged the same topic every +-- notify_repeat forever, which is a louder version of the same silence: if the +-- person on call is asleep, has no signal, or has left, nothing else happens. +-- +-- Shape: one policy per team, an ordered list of levels, each level with a +-- timeout and a set of targets. When a level's timeout passes and the incident +-- is still triggered, the next level is paged. When the last level passes, the +-- chain repeats repeat_count times, and then the team's fallback topic is paged +-- once as the end of the line. +-- +-- A team WITHOUT a policy keeps exactly today's behaviour: page the assignee, +-- then remind on the same topic. Escalation is opt-in per team, and the two +-- never both run for one incident -- see enqueueReminders. +CREATE TABLE escalation_policies ( + -- One per team for now, hence the team as the key rather than an id with a + -- unique index: routing different alerts to different chains needs the + -- alert to carry something to route ON, which is a separate question. + team_id BIGINT PRIMARY KEY REFERENCES teams(id) ON DELETE CASCADE, + + -- How many extra times to run the whole chain after it has been walked + -- once. 0 means walk it once and stop at the fallback. + repeat_count BIGINT NOT NULL DEFAULT 0 CHECK (repeat_count >= 0 AND repeat_count <= 10), + + -- Where the last page goes when every level has been tried. Per team now: + -- TERDUT_NTFY_FALLBACK_TOPIC was one topic for the whole install, which in + -- a multi-team server pages the wrong people. Empty means the chain simply + -- ends. + fallback_topic TEXT NOT NULL DEFAULT '', + + updated_at BIGINT NOT NULL DEFAULT FLOOR(EXTRACT(EPOCH FROM now()))::bigint +); + +CREATE TABLE escalation_levels ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + team_id BIGINT NOT NULL REFERENCES escalation_policies(team_id) ON DELETE CASCADE, + -- 1-based, dense. The API rewrites the whole ladder on every edit rather + -- than patching one rung, so there is no way to leave a gap. + position BIGINT NOT NULL, + -- How long this level has to produce an acknowledgement before the next one + -- is paged. Seconds, like every other duration in this schema. + timeout_seconds BIGINT NOT NULL CHECK (timeout_seconds > 0), + + UNIQUE (team_id, position) +); + +-- Who a level pages. Either a named person, or whoever the team's rota says is +-- on call today -- which is the target that keeps working when the rota +-- changes and nobody remembers to edit the policy. +CREATE TABLE escalation_targets ( + id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, + level_id BIGINT NOT NULL REFERENCES escalation_levels(id) ON DELETE CASCADE, + kind TEXT NOT NULL CHECK (kind IN ('user', 'oncall')), + -- Set for kind='user', NULL for kind='oncall'. + user_id BIGINT REFERENCES users(id) ON DELETE CASCADE, + + CHECK ((kind = 'user' AND user_id IS NOT NULL) OR (kind = 'oncall' AND user_id IS NULL)) +); + +CREATE INDEX escalation_targets_level_idx ON escalation_targets(level_id); + +-- --------------------------------------------------------------------------- +-- Where an incident is in its chain. +-- +-- On the incident rather than in a side table: it is read on every notifier +-- tick alongside the incident's status, and one row per incident is exactly +-- what the state is. +-- --------------------------------------------------------------------------- + +-- 0 means no level has been paged yet, which is the state of every incident +-- that existed before escalation and of every incident in a team with no +-- policy. 1 is the first level. +ALTER TABLE incidents ADD COLUMN escalation_level BIGINT NOT NULL DEFAULT 0; + +-- When the current level was entered, and therefore what its timeout is +-- measured from. NULL while escalation_level is 0. +ALTER TABLE incidents ADD COLUMN escalation_level_at BIGINT; + +-- How many times the chain has been walked in full. Compared against the +-- policy's repeat_count. +ALTER TABLE incidents ADD COLUMN escalation_round BIGINT NOT NULL DEFAULT 0; + +-- The notifier's escalation query: incidents still waiting, oldest level first. +CREATE INDEX incidents_escalation_idx + ON incidents(escalation_level_at) + WHERE resolved_at IS NULL AND status = 'triggered'; + +-- 'escalated' joins the outbox kinds: a page that went out because nobody +-- answered the last one, which is worth telling apart from the first page and +-- from a reminder when reading the timeline or debugging a delivery. +ALTER TABLE notifications DROP CONSTRAINT notifications_kind_check; +ALTER TABLE notifications ADD CONSTRAINT notifications_kind_check + CHECK (kind IN ('triggered', 'reminder', 'resolved', 'escalated'));