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" ) // 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 } func handleAlertmanagerWebhook(db *sql.DB, notify NotifyConfig) 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, 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, payload amPayload) error { tx, err := db.BeginTx(ctx, nil) if err != nil { return err } defer tx.Rollback() //nolint:errcheck accepted, err := upsertAlerts(ctx, tx, 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 { continue } if err := linkAlert(ctx, tx, incidentID, a.id); err != nil { return err } } } for _, a := range accepted { if !a.justResolved { 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, 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 = ?", 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. if _, err := tx.ExecContext(ctx, ` INSERT INTO alerts (fingerprint, name, status, labels, annotations, starts_at, ends_at, generator_url, received_at, resolution_source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 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 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 = ?", 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", }) } 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. 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.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 = ? 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, payload.GroupLabels, firstName) } // openIncident creates an incident for a group and assigns it to whoever is on // call today, which is the point at which the schedule stops being decorative. func openIncident(ctx context.Context, tx *sql.Tx, notify NotifyConfig, groupKey string, groupLabels map[string]string, fallbackName string) (int64, error) { onCall, err := currentOnCall(ctx, tx) if err != nil { return 0, err } labelsJSON, _ := json.Marshal(groupLabels) if groupLabels == nil { labelsJSON = []byte("{}") } res, err := tx.ExecContext(ctx, ` INSERT INTO incidents (group_key, title, group_labels, status, triggered_at, assigned_to) VALUES (?, ?, ?, 'triggered', ?, ?)`, groupKey, incidentTitle(groupLabels, fallbackName), string(labelsJSON), time.Now().Unix(), onCall) if err != nil { return 0, err } id, err := res.LastInsertId() if err != nil { return 0, err } if err := logEvent(ctx, tx, 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, tx, id, evAssigned, onCall, nil, nil); err != nil { return 0, err } } // Queue the page, but do not send it here: this runs inside the webhook's // 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, tx, 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 OR IGNORE INTO incident_alerts (incident_id, alert_id, added_at) VALUES (?, ?, ?)`, 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) }