package api import ( "context" "database/sql" "encoding/json" "log" "net/http" "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" // resolutionDeadman marks a heartbeat the dead man's switch sweeper declared // dead. Distinct from expiry because it is load-bearing, not just // descriptive: it is the one resolution the ingest upsert will let a // same-instance re-fire undo, so a switch that comes back can be heard. resolutionDeadman = "deadman" ) // amPayload mirrors the Alertmanager webhook v4 payload. type amPayload struct { Version string `json:"version"` Status string `json:"status"` // GroupKey and GroupLabels are how alerts get correlated into incidents. // Alertmanager has already done the grouping work according to the group_by // routing tree the operator configured, so we adopt its answer instead of // inventing a second grouping scheme here. GroupKey string `json:"groupKey"` GroupLabels map[string]string `json:"groupLabels"` Alerts []amAlert `json:"alerts"` } type amAlert struct { Status string `json:"status"` Labels map[string]string `json:"labels"` Annotations map[string]string `json:"annotations"` StartsAt time.Time `json:"startsAt"` EndsAt time.Time `json:"endsAt"` GeneratorURL string `json:"generatorURL"` Fingerprint string `json:"fingerprint"` } // ingested records what actually happened to one alert of a payload, which is // what decides whether an incident opens. type ingested struct { id int64 name string firing bool // newOccurrence marks an alert that transitioned *into* firing: a // fingerprint we had never seen, a newer startsAt, or a resolved alert that // started again. A repeat_interval re-send of an already-firing alert is // none of these, which is what keeps a manually resolved incident closed. newOccurrence bool // justResolved marks the firing → resolved edge, worth a timeline entry. justResolved bool // deadman marks a heartbeat: an alert whose arrival means everything is // fine. It is stored like any other alert — received_at is the heartbeat — // but it never reaches an incident. Its absence is what opens one, which // sweepDeadman decides later and elsewhere. deadman bool } func handleAlertmanagerWebhook(db *sql.DB, notify NotifyConfig, deadman DeadmanConfig) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var payload amPayload if err := decodeJSON(r, &payload); err != nil { respond(w, http.StatusBadRequest, errResp("invalid payload")) return } // Alertmanager retries anything that is not 2xx, and a retry of a payload // we failed to store is more useful than an error it cannot act on — so // failures are logged, not surfaced. if err := ingest(r.Context(), db, notify, deadman, payload); err != nil { log.Printf("webhook ingest (group %q): %v", payload.GroupKey, err) } w.WriteHeader(http.StatusOK) } } // ingest stores a payload's alerts and reconciles the incident for its group. // The whole payload is one transaction: an incident that opened but whose alerts // failed to link would be a work item nobody could act on. func ingest(ctx context.Context, db *sql.DB, notify NotifyConfig, deadman DeadmanConfig, payload amPayload) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() //nolint:errcheck accepted, err := upsertAlerts(ctx, tx, deadman, payload.Alerts) if err != nil { return err } // touched collects every incident this payload affected, so severity and the // resolution cascade are recomputed once per incident at the end. touched := map[int64]bool{} incidentID, err := incidentForGroup(ctx, tx, notify, payload, accepted) if err != nil { return err } if incidentID != 0 { touched[incidentID] = true for _, a := range accepted { if !a.firing || a.deadman { continue } if err := linkAlert(ctx, tx, incidentID, a.id); err != nil { return err } } } for _, a := range accepted { if !a.justResolved || a.deadman { continue } id, err := openIncidentForAlert(ctx, tx, a.id) if err != nil { return err } if id == 0 { continue } touched[id] = true alertID := a.id if err := logEvent(ctx, tx, id, evAlertResolved, nil, &alertID, nil); err != nil { return err } } for id := range touched { if err := refreshSeverity(ctx, tx, id); err != nil { return err } if _, err := resolveIfSettled(ctx, tx, id); err != nil { return err } } return tx.Commit() } // upsertAlerts stores each alert of a payload and reports what changed. Payloads // the ordering guard rejected are left out entirely. func upsertAlerts(ctx context.Context, tx *sql.Tx, deadman DeadmanConfig, alerts []amAlert) ([]ingested, error) { now := time.Now().Unix() accepted := make([]ingested, 0, len(alerts)) for _, a := range alerts { name := a.Labels["alertname"] labelsJSON, _ := json.Marshal(a.Labels) annotationsJSON, _ := json.Marshal(a.Annotations) // The stored state has to be read before the upsert overwrites it: it is // the only way to tell a genuine new occurrence from a re-send. var prevStatus string var prevStartsAt int64 existed := true switch err := tx.QueryRowContext(ctx, "SELECT status, starts_at FROM alerts WHERE fingerprint = $1", a.Fingerprint, ).Scan(&prevStatus, &prevStartsAt); { case err == sql.ErrNoRows: existed = false case err != nil: return nil, err } // 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 if a.EndsAt.Year() > 1 { t := a.EndsAt.Unix() 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 — with one exception. // // A resolution this server synthesised for a dead man's switch is not // Alertmanager's word that the instance ended; it is our inference from // silence. The heartbeat that proves us wrong carries the unchanged // startsAt of an alert that never stopped firing, so without the // exemption a switch could go dead exactly once and never be heard from // again. Scoped to 'deadman' so no resolution anybody else wrote can be // undone by a stale retry. if _, err := tx.ExecContext(ctx, ` INSERT INTO alerts (fingerprint, name, status, labels, annotations, starts_at, ends_at, generator_url, received_at, resolution_source) VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6, $7, $8, $9, $10) ON CONFLICT (fingerprint) DO UPDATE SET status = excluded.status, labels = excluded.labels, annotations = excluded.annotations, starts_at = excluded.starts_at, ends_at = excluded.ends_at, generator_url = excluded.generator_url, -- Load-bearing: advancing received_at on every accepted -- payload, re-sends included, is the documented liveness -- heartbeat clients and the sweeper both read. Removing it -- is a breaking API change — see models.Alert.ReceivedAt. 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 (alerts.resolution_source = '`+resolutionDeadman+`' OR NOT (alerts.status = 'resolved' AND excluded.status = 'firing')))`, a.Fingerprint, name, a.Status, string(labelsJSON), string(annotationsJSON), a.StartsAt.Unix(), endsAtUnix, a.GeneratorURL, now, resolutionSource, ); err != nil { return nil, err } var id int64 var curStatus string var curStartsAt int64 if err := tx.QueryRowContext(ctx, "SELECT id, status, starts_at FROM alerts WHERE fingerprint = $1", a.Fingerprint, ).Scan(&id, &curStatus, &curStartsAt); err != nil { return nil, err } // The upsert copies status and starts_at straight from the payload, so a // row that does not match it is one the ordering guard rejected. A // discarded payload describes a past instance and must not touch the // incident state either. if existed && (curStatus != a.Status || curStartsAt != a.StartsAt.Unix()) { continue } firing := a.Status == "firing" accepted = append(accepted, ingested{ id: id, name: name, firing: firing, newOccurrence: firing && (!existed || a.StartsAt.Unix() > prevStartsAt || prevStatus == "resolved"), justResolved: !firing && existed && prevStatus == "firing", deadman: deadman.isDeadman(a.Labels), }) } return accepted, nil } // incidentForGroup returns the open incident that this payload's firing alerts // belong to, opening one if the group has none. It returns 0 when the payload // warrants no incident at all. // // The rule that matters: a group with no open incident gets a new one only if // something actually started firing. Without that, a manually resolved incident // would reappear on the next repeat_interval re-send of an alert that never // stopped, and manual resolution would be meaningless. // // Heartbeats do not count as anything here. A group of nothing but dead man's // switch alerts opens no incident at all, and a mixed group gets an incident for // its real alerts only. func incidentForGroup(ctx context.Context, tx *sql.Tx, notify NotifyConfig, payload amPayload, accepted []ingested) (int64, error) { var firstName string anyFiring, anyNew := false, false for _, a := range accepted { if a.deadman { continue } if a.firing { if !anyFiring { firstName = a.name } anyFiring = true } if a.newOccurrence { anyNew = true } } if !anyFiring { // A payload of nothing but resolutions never opens an incident. return 0, nil } groupKey := payload.GroupKey if groupKey == "" { // Alertmanager always sends groupKey; a sender that does not still gets // one incident per alert name rather than one giant shared incident. groupKey = "groupless:" + firstName } var id int64 switch err := tx.QueryRowContext(ctx, "SELECT id FROM incidents WHERE group_key = $1 AND resolved_at IS NULL", groupKey, ).Scan(&id); { case err == nil: return id, nil case err != sql.ErrNoRows: return 0, err } if !anyNew { return 0, nil } return openIncident(ctx, tx, notify, groupKey, incidentTitle(payload.GroupLabels, firstName), payload.GroupLabels, nil) } // openIncident creates an incident and assigns it to whoever is on call today, // which is the point at which the schedule stops being decorative. // // The one place an incident is born, for both of the things that can raise one: // the webhook, inside its transaction, and the dead man's switch sweeper, inside // its own. Hence the querier rather than a *sql.Tx. A nil severity leaves the // column for refreshSeverity to fill from the member alerts; the sweeper passes // one because its incidents have no members to derive it from. func openIncident(ctx context.Context, q querier, notify NotifyConfig, groupKey, title string, groupLabels map[string]string, severity *string) (int64, error) { onCall, err := currentOnCall(ctx, q) if err != nil { return 0, err } labelsJSON, _ := json.Marshal(groupLabels) if groupLabels == nil { labelsJSON = []byte("{}") } var id int64 err = q.QueryRowContext(ctx, ` INSERT INTO incidents (group_key, title, group_labels, status, severity, triggered_at, assigned_to) VALUES ($1, $2, $3::jsonb, 'triggered', $4, $5, $6) RETURNING id`, groupKey, title, string(labelsJSON), severity, time.Now().Unix(), onCall).Scan(&id) if err != nil { return 0, err } if err := logEvent(ctx, q, id, evTriggered, nil, nil, nil); err != nil { return 0, err } if onCall != nil { // On an "assigned" event user_id is the assignee, not the actor. if err := logEvent(ctx, q, id, evAssigned, onCall, nil, nil); err != nil { return 0, err } } // 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. if err := enqueueOpened(ctx, q, notify, id, onCall); err != nil { return 0, err } return id, nil } // linkAlert adds an alert to an incident, emitting a timeline entry only the // first time. Re-sends of an already-linked alert are silent. func linkAlert(ctx context.Context, tx *sql.Tx, incidentID, alertID int64) error { res, err := tx.ExecContext(ctx, ` INSERT INTO incident_alerts (incident_id, alert_id, added_at) VALUES ($1, $2, $3) ON CONFLICT (incident_id, alert_id) DO NOTHING`, incidentID, alertID, time.Now().Unix()) if err != nil { return err } if n, _ := res.RowsAffected(); n == 0 { return nil } return logEvent(ctx, tx, incidentID, evAlertAdded, nil, &alertID, nil) }