terdut-server/internal/api/metrics.go

package api

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"log"
	"net/http"
	"sync"
	"time"

	"git.ryuvia.com/niklas/terdut-server/internal/config"
	"git.ryuvia.com/niklas/terdut-server/internal/prom"
)

const (
	// metricsInterval is how often the worker looks for incidents that need a
	// snapshot. Short, because the "opened" snapshot is most useful when it is
	// taken close to the moment the incident opened.
	metricsInterval = 15 * time.Second

	// metricsBatch caps one pass, and metricsConcurrency how many Prometheus
	// requests run at once within it.
	metricsBatch       = 50
	metricsConcurrency = 4

	// metricsMaxAttempts stops a failing capture being retried forever, and
	// metricsRetryStep spaces the retries: attempt n waits n*metricsRetryStep.
	metricsMaxAttempts = 4
	metricsRetryStep   = time.Minute

	// metricsLookback is how far back the worker goes for incidents that are
	// already closed. Without it, switching the feature on would try to
	// reconstruct the whole history from a Prometheus that has probably aged it
	// out. Incidents that are still open are always eligible.
	metricsLookback = 24 * time.Hour

	// maxAlertsPerIncident bounds the Prometheus work one incident can cause.
	maxAlertsPerIncident = 10

	// metricsLiveTTL is how long a live answer is reused, so a room full of
	// people watching one incident is one query per alert, not one each.
	metricsLiveTTL = 15 * time.Second

	// metricsLockKey is the advisory lock a pass holds, distinct from the
	// archiver's and the notifier's.
	metricsLockKey int64 = 7265_0004
)

// metricsService is the Prometheus side of the incident page: a client that only
// reaches allowed hosts, and a short-lived cache of live answers.
type metricsService struct {
	client  *prom.Client
	preroll time.Duration

	mu    sync.Mutex
	cache map[int64]liveEntry
}

type liveEntry struct {
	at   time.Time
	live *liveMetrics
}

func newMetricsService(cfg config.Config) *metricsService {
	// Validate has already refused a bad list at startup; a failure here can only
	// be a test building a Config by hand, and no allowlist is the safe reading.
	allow, err := prom.ParseAllowlist(cfg.PrometheusAllowedHosts)
	if err != nil {
		allow = nil
	}
	preroll := cfg.PrometheusPreroll
	if preroll <= 0 {
		preroll = time.Hour
	}
	return &metricsService{
		client:  prom.NewClient(allow, cfg.PrometheusTimeout),
		preroll: preroll,
		cache:   map[int64]liveEntry{},
	}
}

func (m *metricsService) enabled() bool { return m.client.Enabled() }

// StartMetricsWorker captures the opened and closed snapshots until ctx is
// cancelled. With no allowed hosts configured it does nothing.
func StartMetricsWorker(ctx context.Context, db *sql.DB, cfg config.Config) {
	m := newMetricsService(cfg)
	if !m.enabled() {
		log.Print("metrics: disabled (TERDUT_PROMETHEUS_ALLOWED_HOSTS is empty)")
		return
	}
	log.Printf("metrics: capturing from %d allowed Prometheus host entries", len(cfg.PrometheusAllowedHosts))

	ticker := time.NewTicker(metricsInterval)
	defer ticker.Stop()
	sweep := func() {
		withAdvisoryLock(ctx, db, metricsLockKey, "metrics", func() {
			m.sweep(ctx, db)
		})
	}
	sweep()
	for {
		select {
		case <-ticker.C:
			sweep()
		case <-ctx.Done():
			return
		}
	}
}

// snapshotJob is one (incident, alert, kind) that still needs a snapshot.
type snapshotJob struct {
	incidentID   int64
	alertID      int64
	kind         string
	resolvedAt   sql.NullInt64
	startsAt     int64
	labels       map[string]string
	generatorURL string
}

// sweep finds the snapshots that are missing, or failed and are due a retry, and
// takes them.
func (m *metricsService) sweep(ctx context.Context, db *sql.DB) {
	now := time.Now()
	rows, err := db.QueryContext(ctx, `
		SELECT i.id, a.id, k.kind, i.resolved_at, a.starts_at, a.labels, a.generator_url
		FROM   incidents i
		JOIN   incident_alerts ia ON ia.incident_id = i.id
		JOIN   alerts a ON a.id = ia.alert_id
		CROSS  JOIN (VALUES ('opened'), ('closed')) AS k(kind)
		LEFT   JOIN incident_metric_snapshots s
		       ON s.incident_id = i.id AND s.alert_id = a.id AND s.kind = k.kind
		WHERE  a.generator_url <> ''
		  AND  i.archived_at IS NULL
		  AND  (i.resolved_at IS NULL OR i.resolved_at >= $1)
		  AND  (k.kind = 'opened' OR i.resolved_at IS NOT NULL)
		  AND  (s.id IS NULL
		        OR (s.error IS NOT NULL AND s.attempts < $2
		            AND s.captured_at + s.attempts * $3 <= $4))
		ORDER  BY i.id, a.id, k.kind
		LIMIT  $5`,
		now.Add(-metricsLookback).Unix(), metricsMaxAttempts, int64(metricsRetryStep/time.Second), now.Unix(), metricsBatch)
	if err != nil {
		log.Printf("metrics: find work: %v", err)
		return
	}
	var jobs []snapshotJob
	for rows.Next() {
		var j snapshotJob
		var labels []byte
		if err := rows.Scan(&j.incidentID, &j.alertID, &j.kind, &j.resolvedAt, &j.startsAt, &labels, &j.generatorURL); err != nil {
			log.Printf("metrics: scan: %v", err)
			rows.Close()
			return
		}
		_ = json.Unmarshal(labels, &j.labels)
		jobs = append(jobs, j)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		log.Printf("metrics: find work: %v", err)
		return
	}

	sem := make(chan struct{}, metricsConcurrency)
	var wg sync.WaitGroup
	for _, j := range jobs {
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer func() { <-sem; wg.Done() }()
			m.capture(ctx, db, j)
		}()
	}
	wg.Wait()
}

// window is the span a snapshot covers: from the pre-roll before the alert
// started to the end of the incident, or to now while it is open.
func (m *metricsService) window(startsAt int64, resolvedAt sql.NullInt64, now time.Time) (time.Time, time.Time) {
	start := time.Unix(startsAt, 0).Add(-m.preroll)
	end := now
	if resolvedAt.Valid {
		end = time.Unix(resolvedAt.Int64, 0)
	}
	if !end.After(start) {
		end = start.Add(time.Minute)
	}
	return start, end
}

func (m *metricsService) capture(ctx context.Context, db *sql.DB, j snapshotJob) {
	now := time.Now()
	start, end := m.window(j.startsAt, j.resolvedAt, now)
	step := prom.StepFor(start, end)

	series, expr, err := m.fetchRange(ctx, j.generatorURL, j.labels, start, end, step)

	attempts := 1
	var errText any
	if err != nil {
		errText = err.Error()
		if errors.Is(err, prom.ErrNotAllowed) || errors.Is(err, errNotGraphLink) {
			attempts = metricsMaxAttempts // retrying cannot change the answer
		}
	}
	if series == nil {
		series = []prom.Series{}
	}
	body, _ := json.Marshal(series)

	_, dbErr := db.ExecContext(ctx, `
		INSERT INTO incident_metric_snapshots
		       (incident_id, alert_id, kind, expr, window_start, window_end, step_seconds, series, error, attempts, captured_at)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
		ON CONFLICT (incident_id, alert_id, kind) DO UPDATE SET
		       expr = excluded.expr, window_start = excluded.window_start, window_end = excluded.window_end,
		       step_seconds = excluded.step_seconds, series = excluded.series, error = excluded.error,
		       attempts = CASE WHEN excluded.error IS NULL THEN 1
		                       ELSE GREATEST(incident_metric_snapshots.attempts + 1, excluded.attempts) END,
		       captured_at = excluded.captured_at`,
		j.incidentID, j.alertID, j.kind, expr, start.Unix(), end.Unix(), int(step/time.Second),
		string(body), errText, attempts, now.Unix())
	if dbErr != nil {
		log.Printf("metrics: store snapshot (incident %d, alert %d, %s): %v", j.incidentID, j.alertID, j.kind, dbErr)
	}
}

var errNotGraphLink = errors.New("the alert's generator URL is not a Prometheus graph link")

// fetchRange parses the generator URL and returns the series of the alert's own
// expression over [start, end]. expr is returned even on error where known.
func (m *metricsService) fetchRange(ctx context.Context, generatorURL string, labels map[string]string, start, end time.Time, step time.Duration) ([]prom.Series, string, error) {
	base, expr, ok := prom.ParseGeneratorURL(generatorURL)
	if !ok {
		return nil, "", errNotGraphLink
	}
	series, err := m.client.QueryRange(ctx, base, expr, start, end, step)
	if err != nil {
		return nil, expr, err
	}
	return prom.SelectSeries(series, labels), expr, nil
}

// --- REST ---------------------------------------------------------------

type metricsResponse struct {
	// Enabled is false when no Prometheus host is allowed, so a client can hide
	// the section rather than show every alert as unavailable.
	Enabled bool           `json:"enabled"`
	Alerts  []alertMetrics `json:"alerts"`
}

type alertMetrics struct {
	AlertID   int64  `json:"alert_id"`
	AlertName string `json:"alert_name"`
	Expr      string `json:"expr,omitempty"`

	// Available says whether this alert can be queried at all; Reason says why
	// not when it cannot.
	Available bool   `json:"available"`
	Reason    string `json:"reason,omitempty"`

	Snapshots map[string]*snapshotView `json:"snapshots"`
	Live      *liveMetrics             `json:"live,omitempty"`
}

type snapshotView struct {
	WindowStart int64           `json:"window_start"`
	WindowEnd   int64           `json:"window_end"`
	Step        int             `json:"step_seconds"`
	Series      json.RawMessage `json:"series"`
	Error       string          `json:"error,omitempty"`
	CapturedAt  int64           `json:"captured_at"`
}

type liveMetrics struct {
	WindowStart int64         `json:"window_start"`
	WindowEnd   int64         `json:"window_end"`
	Step        int           `json:"step_seconds"`
	Series      []prom.Series `json:"series"`
	// Current is the expression's value right now, one point per series. Empty
	// means the condition that fires the alert no longer holds.
	Current []prom.Series `json:"current"`
	Error   string        `json:"error,omitempty"`
}

func handleIncidentMetrics(db *sql.DB, m *metricsService) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		id, ok := incidentIDParam(w, r, db)
		if !ok {
			return
		}
		resp := metricsResponse{Enabled: m.enabled(), Alerts: []alertMetrics{}}
		if !resp.Enabled {
			respond(w, http.StatusOK, resp)
			return
		}

		var resolvedAt sql.NullInt64
		if err := db.QueryRowContext(r.Context(), "SELECT resolved_at FROM incidents WHERE id = $1", id).Scan(&resolvedAt); err != nil {
			serverError(w, r, err)
			return
		}

		rows, err := db.QueryContext(r.Context(), `
			SELECT a.id, a.name, a.labels, a.generator_url, a.starts_at
			FROM   alerts a
			JOIN   incident_alerts ia ON ia.alert_id = a.id
			WHERE  ia.incident_id = $1
			ORDER  BY a.starts_at, a.id
			LIMIT  $2`, id, maxAlertsPerIncident)
		if err != nil {
			serverError(w, r, err)
			return
		}
		type alertRow struct {
			am       alertMetrics
			labels   map[string]string
			url      string
			startsAt int64
		}
		var alerts []*alertRow
		for rows.Next() {
			var a alertRow
			var labels []byte
			if err := rows.Scan(&a.am.AlertID, &a.am.AlertName, &labels, &a.url, &a.startsAt); err != nil {
				rows.Close()
				serverError(w, r, err)
				return
			}
			_ = json.Unmarshal(labels, &a.labels)
			a.am.Snapshots = map[string]*snapshotView{}
			alerts = append(alerts, &a)
		}
		rows.Close()
		if err := rows.Err(); err != nil {
			serverError(w, r, err)
			return
		}

		snaps, err := loadSnapshots(r.Context(), db, id)
		if err != nil {
			serverError(w, r, err)
			return
		}

		var wg sync.WaitGroup
		for _, a := range alerts {
			a := a
			if base, expr, ok := prom.ParseGeneratorURL(a.url); !ok {
				a.am.Reason = "no Prometheus link on this alert"
			} else if !m.client.Allowed(base) {
				a.am.Reason = "Prometheus host is not allowed"
				a.am.Expr = expr
			} else {
				a.am.Available = true
				a.am.Expr = expr
			}
			for kind, s := range snaps[a.am.AlertID] {
				a.am.Snapshots[kind] = s
			}
			if a.am.Available && !resolvedAt.Valid {
				wg.Add(1)
				go func() {
					defer wg.Done()
					a.am.Live = m.live(r.Context(), a.am.AlertID, a.url, a.labels, a.startsAt)
				}()
			}
		}
		wg.Wait()

		for _, a := range alerts {
			resp.Alerts = append(resp.Alerts, a.am)
		}
		respond(w, http.StatusOK, resp)
	}
}

func loadSnapshots(ctx context.Context, db *sql.DB, incidentID int64) (map[int64]map[string]*snapshotView, error) {
	rows, err := db.QueryContext(ctx, `
		SELECT alert_id, kind, window_start, window_end, step_seconds, series, COALESCE(error, ''), captured_at
		FROM   incident_metric_snapshots
		WHERE  incident_id = $1`, incidentID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := map[int64]map[string]*snapshotView{}
	for rows.Next() {
		var alertID int64
		var kind string
		var s snapshotView
		var series []byte
		if err := rows.Scan(&alertID, &kind, &s.WindowStart, &s.WindowEnd, &s.Step, &series, &s.Error, &s.CapturedAt); err != nil {
			return nil, err
		}
		s.Series = json.RawMessage(series)
		if out[alertID] == nil {
			out[alertID] = map[string]*snapshotView{}
		}
		out[alertID][kind] = &s
	}
	return out, rows.Err()
}

// live queries Prometheus now, reusing an answer younger than metricsLiveTTL.
// The cache is per alert; the window start is fixed by the alert, only its end
// moves, so one entry serves every viewer.
func (m *metricsService) live(ctx context.Context, alertID int64, generatorURL string, labels map[string]string, startsAt int64) *liveMetrics {
	m.mu.Lock()
	if e, ok := m.cache[alertID]; ok && time.Since(e.at) < metricsLiveTTL {
		m.mu.Unlock()
		return e.live
	}
	m.mu.Unlock()

	now := time.Now()
	start, end := m.window(startsAt, sql.NullInt64{}, now)
	step := prom.StepFor(start, end)
	lm := &liveMetrics{WindowStart: start.Unix(), WindowEnd: end.Unix(), Step: int(step / time.Second),
		Series: []prom.Series{}, Current: []prom.Series{}}

	base, expr, ok := prom.ParseGeneratorURL(generatorURL)
	if !ok {
		lm.Error = errNotGraphLink.Error()
		return lm
	}
	series, err := m.client.QueryRange(ctx, base, expr, start, end, step)
	if err != nil {
		lm.Error = err.Error()
	} else {
		lm.Series = prom.SelectSeries(series, labels)
		if cur, err := m.client.Query(ctx, base, expr, now); err == nil {
			lm.Current = prom.SelectSeries(cur, labels)
		} else {
			lm.Error = err.Error()
		}
	}

	// A failed answer is cached too, for the same TTL: a Prometheus that is down
	// should not be hit once per viewer per poll.
	m.mu.Lock()
	for k, e := range m.cache {
		if time.Since(e.at) > 10*metricsLiveTTL {
			delete(m.cache, k)
		}
	}
	m.cache[alertID] = liveEntry{at: now, live: lm}
	m.mu.Unlock()
	return lm
}