Stage 3: Alertmanager webhook ingestion and alert query API
- Migration 003: alerts table with fingerprint UNIQUE, JSON label/annotation
columns, nullable ends_at, and indexed status/name/received_at
- POST /api/alertmanager/webhook — upserts each alert by fingerprint;
zero endsAt ("0001-01-01") stored as NULL (still firing)
- GET /api/alerts — filtered list (?status, ?name, ?from, ?to, ?limit)
- GET /api/alerts/{id} — single alert lookup
This commit is contained in:
@@ -0,0 +1,72 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"log"
|
||||||
|
"net/http"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// amPayload mirrors the Alertmanager webhook v4 payload.
|
||||||
|
type amPayload struct {
|
||||||
|
Version string `json:"version"`
|
||||||
|
Status string `json:"status"`
|
||||||
|
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"`
|
||||||
|
}
|
||||||
|
|
||||||
|
func handleAlertmanagerWebhook(db *sql.DB) 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
|
||||||
|
}
|
||||||
|
|
||||||
|
now := time.Now().Unix()
|
||||||
|
for _, a := range payload.Alerts {
|
||||||
|
name := a.Labels["alertname"]
|
||||||
|
labelsJSON, _ := json.Marshal(a.Labels)
|
||||||
|
annotationsJSON, _ := json.Marshal(a.Annotations)
|
||||||
|
|
||||||
|
// Alertmanager uses zero time ("0001-01-01T00:00:00Z") to mean "still firing".
|
||||||
|
var endsAtUnix *int64
|
||||||
|
if a.EndsAt.Year() > 1 {
|
||||||
|
t := a.EndsAt.Unix()
|
||||||
|
endsAtUnix = &t
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err := db.ExecContext(r.Context(), `
|
||||||
|
INSERT INTO alerts
|
||||||
|
(fingerprint, name, status, labels, annotations, starts_at, ends_at, generator_url, received_at)
|
||||||
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||||||
|
ON CONFLICT(fingerprint) DO UPDATE SET
|
||||||
|
status = excluded.status,
|
||||||
|
labels = excluded.labels,
|
||||||
|
annotations = excluded.annotations,
|
||||||
|
ends_at = excluded.ends_at,
|
||||||
|
generator_url = excluded.generator_url,
|
||||||
|
received_at = excluded.received_at`,
|
||||||
|
a.Fingerprint, name, a.Status,
|
||||||
|
string(labelsJSON), string(annotationsJSON),
|
||||||
|
a.StartsAt.Unix(), endsAtUnix,
|
||||||
|
a.GeneratorURL, now,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("upsert alert %s: %v", a.Fingerprint, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,137 @@
|
|||||||
|
package api
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/go-chi/chi/v5"
|
||||||
|
"github.com/yeniklas/terdut-server/internal/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
func handleListAlerts(db *sql.DB) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
q := r.URL.Query()
|
||||||
|
|
||||||
|
where := []string{}
|
||||||
|
args := []any{}
|
||||||
|
|
||||||
|
if status := q.Get("status"); status != "" {
|
||||||
|
where = append(where, "status = ?")
|
||||||
|
args = append(args, status)
|
||||||
|
}
|
||||||
|
if name := q.Get("name"); name != "" {
|
||||||
|
where = append(where, "name = ?")
|
||||||
|
args = append(args, name)
|
||||||
|
}
|
||||||
|
if from := q.Get("from"); from != "" {
|
||||||
|
if t, err := time.Parse("2006-01-02", from); err == nil {
|
||||||
|
where = append(where, "received_at >= ?")
|
||||||
|
args = append(args, t.UTC().Unix())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if to := q.Get("to"); to != "" {
|
||||||
|
if t, err := time.Parse("2006-01-02", to); err == nil {
|
||||||
|
// include the full to-day
|
||||||
|
where = append(where, "received_at < ?")
|
||||||
|
args = append(args, t.UTC().AddDate(0, 0, 1).Unix())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
limit := 50
|
||||||
|
if l := q.Get("limit"); l != "" {
|
||||||
|
if n, err := strconv.Atoi(l); err == nil && n > 0 && n <= 500 {
|
||||||
|
limit = n
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
clause := "1=1"
|
||||||
|
if len(where) > 0 {
|
||||||
|
clause = strings.Join(where, " AND ")
|
||||||
|
}
|
||||||
|
args = append(args, limit)
|
||||||
|
|
||||||
|
rows, err := db.QueryContext(r.Context(), fmt.Sprintf(`
|
||||||
|
SELECT id, fingerprint, name, status, labels, annotations,
|
||||||
|
starts_at, ends_at, generator_url, received_at
|
||||||
|
FROM alerts
|
||||||
|
WHERE %s
|
||||||
|
ORDER BY received_at DESC
|
||||||
|
LIMIT ?`, clause), args...)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
|
||||||
|
alerts := []models.Alert{}
|
||||||
|
for rows.Next() {
|
||||||
|
a, err := scanAlert(rows)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
alerts = append(alerts, a)
|
||||||
|
}
|
||||||
|
respond(w, http.StatusOK, alerts)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func handleGetAlert(db *sql.DB) http.HandlerFunc {
|
||||||
|
return func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64)
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusBadRequest, errResp("invalid alert id"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
row := db.QueryRowContext(r.Context(), `
|
||||||
|
SELECT id, fingerprint, name, status, labels, annotations,
|
||||||
|
starts_at, ends_at, generator_url, received_at
|
||||||
|
FROM alerts WHERE id = ?`, id)
|
||||||
|
a, err := scanAlert(row)
|
||||||
|
if err == sql.ErrNoRows {
|
||||||
|
respond(w, http.StatusNotFound, errResp("alert not found"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
respond(w, http.StatusInternalServerError, errResp("internal error"))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
respond(w, http.StatusOK, a)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// scanner is satisfied by both *sql.Row and *sql.Rows.
|
||||||
|
type scanner interface {
|
||||||
|
Scan(dest ...any) error
|
||||||
|
}
|
||||||
|
|
||||||
|
func scanAlert(s scanner) (models.Alert, error) {
|
||||||
|
var a models.Alert
|
||||||
|
var labelsJSON, annotationsJSON string
|
||||||
|
var startsAtUnix, receivedAtUnix int64
|
||||||
|
var endsAtUnix *int64
|
||||||
|
|
||||||
|
if err := s.Scan(
|
||||||
|
&a.ID, &a.Fingerprint, &a.Name, &a.Status,
|
||||||
|
&labelsJSON, &annotationsJSON,
|
||||||
|
&startsAtUnix, &endsAtUnix,
|
||||||
|
&a.GeneratorURL, &receivedAtUnix,
|
||||||
|
); err != nil {
|
||||||
|
return a, err
|
||||||
|
}
|
||||||
|
|
||||||
|
json.Unmarshal([]byte(labelsJSON), &a.Labels) //nolint:errcheck
|
||||||
|
json.Unmarshal([]byte(annotationsJSON), &a.Annotations) //nolint:errcheck
|
||||||
|
a.StartsAt = time.Unix(startsAtUnix, 0).UTC()
|
||||||
|
a.ReceivedAt = time.Unix(receivedAtUnix, 0).UTC()
|
||||||
|
if endsAtUnix != nil {
|
||||||
|
t := time.Unix(*endsAtUnix, 0).UTC()
|
||||||
|
a.EndsAt = &t
|
||||||
|
}
|
||||||
|
return a, nil
|
||||||
|
}
|
||||||
@@ -17,8 +17,9 @@ func NewRouter(db *sql.DB) http.Handler {
|
|||||||
respond(w, http.StatusOK, map[string]string{"status": "ok"})
|
respond(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||||
})
|
})
|
||||||
|
|
||||||
// Unauthenticated: bootstrap (only works when user table is empty).
|
// Unauthenticated: bootstrap and Alertmanager webhook receiver.
|
||||||
r.Post("/api/bootstrap", handleBootstrap(db))
|
r.Post("/api/bootstrap", handleBootstrap(db))
|
||||||
|
r.Post("/api/alertmanager/webhook", handleAlertmanagerWebhook(db))
|
||||||
|
|
||||||
// All other /api routes require a valid API key.
|
// All other /api routes require a valid API key.
|
||||||
r.Group(func(r chi.Router) {
|
r.Group(func(r chi.Router) {
|
||||||
@@ -29,6 +30,9 @@ func NewRouter(db *sql.DB) http.Handler {
|
|||||||
r.Delete("/api/users/{id}", handleDeleteUser(db))
|
r.Delete("/api/users/{id}", handleDeleteUser(db))
|
||||||
r.Post("/api/users/{id}/api-keys", handleCreateAPIKey(db))
|
r.Post("/api/users/{id}/api-keys", handleCreateAPIKey(db))
|
||||||
r.Delete("/api/users/{id}/api-keys/{keyID}", handleDeleteAPIKey(db))
|
r.Delete("/api/users/{id}/api-keys/{keyID}", handleDeleteAPIKey(db))
|
||||||
|
|
||||||
|
r.Get("/api/alerts", handleListAlerts(db))
|
||||||
|
r.Get("/api/alerts/{id}", handleGetAlert(db))
|
||||||
})
|
})
|
||||||
|
|
||||||
return r
|
return r
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
CREATE TABLE alerts (
|
||||||
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||||
|
fingerprint TEXT NOT NULL UNIQUE,
|
||||||
|
name TEXT NOT NULL,
|
||||||
|
status TEXT NOT NULL CHECK(status IN ('firing', 'resolved')),
|
||||||
|
labels TEXT NOT NULL DEFAULT '{}',
|
||||||
|
annotations TEXT NOT NULL DEFAULT '{}',
|
||||||
|
starts_at INTEGER NOT NULL,
|
||||||
|
ends_at INTEGER,
|
||||||
|
generator_url TEXT NOT NULL DEFAULT '',
|
||||||
|
received_at INTEGER NOT NULL DEFAULT (strftime('%s', 'now'))
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX alerts_status_idx ON alerts(status);
|
||||||
|
CREATE INDEX alerts_name_idx ON alerts(name);
|
||||||
|
CREATE INDEX alerts_received_at_idx ON alerts(received_at DESC);
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
package models
|
||||||
|
|
||||||
|
import "time"
|
||||||
|
|
||||||
|
type Alert struct {
|
||||||
|
ID int64 `json:"id"`
|
||||||
|
Fingerprint string `json:"fingerprint"`
|
||||||
|
Name string `json:"name"`
|
||||||
|
Status string `json:"status"` // "firing" or "resolved"
|
||||||
|
Labels map[string]string `json:"labels"`
|
||||||
|
Annotations map[string]string `json:"annotations"`
|
||||||
|
StartsAt time.Time `json:"starts_at"`
|
||||||
|
EndsAt *time.Time `json:"ends_at,omitempty"`
|
||||||
|
GeneratorURL string `json:"generator_url"`
|
||||||
|
ReceivedAt time.Time `json:"received_at"`
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user