go-otel/internal/alerts/runner.go

package alerts

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"os"
	"strconv"
	"sync"
	"time"

	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"

	"git.tornberg.me/go-otel/internal/features"
)

// Config is read from OTEL_ALERT_* environment variables. The runner starts
// when at least one notifier is configured, or when OTEL_ALERT_DRY_RUN=true
// (evaluate and record, send nothing).
type Config struct {
	DryRun          bool
	Interval        time.Duration // how often the detector runs
	Window          time.Duration // observation window
	BaselineWindows int
	Cooldown        time.Duration // suppress repeats of a pattern unless it doubles
	MinExcess       int
	Multiple        float64
	Lag             time.Duration // ingest settle time: windows end this long before now
	BaseURL         string        // UI link appended to messages, e.g. https://otel.k6n.net

	TelegramToken, TelegramChatID, TelegramThreadID string
	// Muted, when set, reports issues a person ignored; their spikes never alert.
	Muted                     func(issueID string) bool
	WebhookURL, WebhookSecret string
	AODURL                    string
	// LayaURL, when set, asks laya (POST {LayaURL}/predict) for a shadow
	// verdict on every alert and shows it in the notification.
	LayaURL string
}

func ConfigFromEnv() Config {
	return Config{
		DryRun:          os.Getenv("OTEL_ALERT_DRY_RUN") == "true",
		Interval:        envDuration("OTEL_ALERT_INTERVAL", time.Minute),
		Window:          envDuration("OTEL_ALERT_WINDOW", 5*time.Minute),
		BaselineWindows: envInt("OTEL_ALERT_BASELINE_WINDOWS", 36),
		Cooldown:        envDuration("OTEL_ALERT_COOLDOWN", 30*time.Minute),
		MinExcess:       envInt("OTEL_ALERT_MIN_EXCESS", 10),
		Multiple:        3,
		Lag:             envDuration("OTEL_ALERT_LAG", 30*time.Second),
		BaseURL:         os.Getenv("OTEL_ALERT_BASE_URL"),

		TelegramToken:    firstEnv("OTEL_ALERT_TELEGRAM_TOKEN", "TELEGRAM_KEY"),
		TelegramChatID:   firstEnv("OTEL_ALERT_TELEGRAM_CHAT_ID", "TELEGRAM_CHATID"),
		TelegramThreadID: os.Getenv("OTEL_ALERT_TELEGRAM_THREAD_ID"),
		WebhookURL:       os.Getenv("OTEL_ALERT_WEBHOOK_URL"),
		WebhookSecret:    os.Getenv("OTEL_ALERT_WEBHOOK_SECRET"),
		AODURL:           os.Getenv("OTEL_ALERT_AOD_URL"),
		LayaURL:          os.Getenv("OTEL_ALERT_LAYA_URL"),
	}
}

// firstEnv returns the first non-empty variable; the plain TELEGRAM_* names are
// what the repo's .env uses, the OTEL_ALERT_* ones are for the deployment.
func firstEnv(keys ...string) string {
	for _, k := range keys {
		if v := os.Getenv(k); v != "" {
			return v
		}
	}
	return ""
}

func envDuration(k string, def time.Duration) time.Duration {
	if d, err := time.ParseDuration(os.Getenv(k)); err == nil && d > 0 {
		return d
	}
	return def
}
func envInt(k string, def int) int {
	if n, err := strconv.Atoi(os.Getenv(k)); err == nil && n > 0 {
		return n
	}
	return def
}

// Notifiers builds the configured delivery channels.
func (c Config) Notifiers() []Notifier {
	var out []Notifier
	if c.TelegramToken != "" && c.TelegramChatID != "" {
		out = append(out, &Telegram{Token: c.TelegramToken, ChatID: c.TelegramChatID, ThreadID: c.TelegramThreadID, BaseURL: c.BaseURL})
	}
	if c.WebhookURL != "" {
		out = append(out, &Webhook{URL: c.WebhookURL, Secret: c.WebhookSecret, BaseURL: c.BaseURL})
	}
	if c.AODURL != "" {
		out = append(out, &AlwaysOnDisplay{URL: c.AODURL})
	}
	return out
}

// Loader returns error events in [start, end); it is a field so tests can
// feed events without ClickHouse.
type Loader func(ctx context.Context, start, end time.Time) ([]features.Event, error)

type Runner struct {
	cfg       Config
	load      Loader
	dataSince func(ctx context.Context) time.Time
	notifiers []Notifier
	resolver  Resolver // optional: ingress -> service -> deployment for edge errors
	laya      *Laya    // optional: shadow verdict attached to each alert
	sup       *features.Suppressor

	mu        sync.Mutex
	primed    bool
	recent    []Alert // newest last
	lastRun   time.Time
	lastError string
	since     time.Time
	sinceAt   time.Time
}

const maxRecent = 50

// Start launches the runner against ClickHouse and returns it, or nil when
// alerting is not configured.
func Start(ctx context.Context, cfg Config, conn driver.Conn, database string, resolver Resolver) *Runner {
	notifiers := cfg.Notifiers()
	if len(notifiers) == 0 && !cfg.DryRun {
		log.Printf("alerts: disabled (set OTEL_ALERT_TELEGRAM_TOKEN + OTEL_ALERT_TELEGRAM_CHAT_ID and/or OTEL_ALERT_WEBHOOK_URL)")
		return nil
	}
	r := &Runner{
		cfg:       cfg,
		notifiers: notifiers,
		resolver:  resolver,
		sup:       features.NewSuppressor(cfg.Cooldown),
		load: func(ctx context.Context, start, end time.Time) ([]features.Event, error) {
			return LoadErrorEvents(ctx, conn, database, start, end)
		},
		dataSince: func(ctx context.Context) time.Time { return DataSince(ctx, conn, database) },
	}
	if cfg.LayaURL != "" {
		r.laya = &Laya{URL: cfg.LayaURL}
		log.Printf("alerts: laya shadow verdicts from %s", cfg.LayaURL)
	}
	names := make([]string, 0, len(notifiers))
	for _, n := range notifiers {
		names = append(names, n.Name())
	}
	log.Printf("alerts: enabled every %s (window %s, baseline %d windows, dry-run=%v, channels=%v)", cfg.Interval, cfg.Window, cfg.BaselineWindows, cfg.DryRun, names)
	go r.loop(ctx)
	return r
}

func (r *Runner) loop(ctx context.Context) {
	t := time.NewTicker(r.cfg.Interval)
	defer t.Stop()
	for {
		r.Tick(ctx, time.Now())
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// Tick runs one evaluation. The first tick after a start only primes the
// suppressor, so a deploy in the middle of an incident does not re-page it.
func (r *Runner) Tick(ctx context.Context, now time.Time) {
	cctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
	defer cancel()
	end := now.Add(-r.cfg.Lag).UTC()
	start := end.Add(-time.Duration(r.cfg.BaselineWindows+1) * r.cfg.Window)
	events, err := r.load(cctx, start, end)
	r.mu.Lock()
	r.lastRun = now
	if err != nil {
		r.lastError = err.Error()
		r.mu.Unlock()
		log.Printf("alerts: load failed: %v", err)
		return
	}
	r.lastError = ""
	if r.since.IsZero() || now.Sub(r.sinceAt) > time.Hour {
		r.mu.Unlock()
		since := r.dataSince(cctx)
		r.mu.Lock()
		r.since, r.sinceAt = since, now
	}
	since, primed := r.since, r.primed
	r.mu.Unlock()

	spikes := DetectAlerts(events, end, features.SpikeConfig{
		Window: r.cfg.Window, BaselineWindows: r.cfg.BaselineWindows, Multiple: r.cfg.Multiple, MinExcess: r.cfg.MinExcess, DataSince: since,
	})
	// Ignored issues are muted before the cooldown suppressor, so they never
	// consume a cooldown slot or page anyone.
	if r.cfg.Muted != nil {
		before := len(spikes)
		spikes = DropMuted(spikes, r.cfg.Muted)
		if len(spikes) != before {
			log.Printf("alerts: muted spike(s) for ignored issues (%d -> %d)", before, len(spikes))
		}
	}
	spikes = r.sup.Filter(spikes)

	r.mu.Lock()
	r.primed = true
	r.mu.Unlock()
	if !primed {
		if len(spikes) > 0 {
			log.Printf("alerts: primed with %d ongoing spike(s), not sent", len(spikes))
		}
		return
	}
	for _, spike := range spikes {
		r.deliver(cctx, NewAlert(spike, end))
	}
}

// DetectAlerts is the detection half of a tick: spikes reduced to actionable
// patterns. Exported so the CLI and the runner cannot drift apart.
func DetectAlerts(events []features.Event, at time.Time, cfg features.SpikeConfig) []features.Spike {
	// Scanners, route enumerators and crawlers getting 4xx are noise for paging.
	features.MarkBadActors(events, features.BadActorConfig{})
	minExcess := cfg.MinExcess
	if minExcess <= 0 {
		minExcess = features.DefaultSpikeConfig().MinExcess
	}
	var out []features.Spike
	// The excess threshold is applied after triage, with 4xx discounted: a
	// service whose excess is mostly probes or client errors must not page on
	// the little that is left.
	for _, sp := range FilterActionable(features.DetectSpikes(events, at, cfg), false, nil) {
		if PagingExcess(sp) >= float64(minExcess) {
			out = append(out, sp)
		}
	}
	return out
}

func (r *Runner) deliver(ctx context.Context, a Alert) {
	a.Enrich(r.resolver)
	if pr, ok := r.resolver.(PodResolver); ok {
		a.EnrichPods(ctx, pr)
	}
	r.askLaya(ctx, &a)
	r.mu.Lock()
	r.recent = append(r.recent, a)
	if len(r.recent) > maxRecent {
		r.recent = r.recent[len(r.recent)-maxRecent:]
	}
	r.mu.Unlock()
	if r.cfg.DryRun {
		log.Printf("alerts[dry-run]: %s", a.Text(r.cfg.BaseURL))
		return
	}
	for _, n := range r.notifiers {
		err := n.Send(ctx, a)
		if err != nil {
			time.Sleep(2 * time.Second)
			err = n.Send(ctx, a)
		}
		if err != nil {
			log.Printf("alerts: %s delivery failed for %s: %v", n.Name(), a.ID, err)
		}
	}
}

// askLaya attaches laya's shadow verdict. It is bounded so a slow model
// delays a page by seconds at most, and a failure only costs the verdict.
func (r *Runner) askLaya(ctx context.Context, a *Alert) {
	if r.laya == nil {
		return
	}
	lctx, cancel := context.WithTimeout(ctx, 8*time.Second)
	defer cancel()
	v, err := r.laya.Classify(lctx, *a)
	if err != nil {
		log.Printf("alerts: laya verdict for %s failed: %v", a.ID, err)
		return
	}
	a.Laya = v
	log.Printf("alerts: laya[shadow] %s: %s", a.ID, v.Summary())
}

// Status is the JSON served by GET /api/alerts.
type Status struct {
	Enabled   bool      `json:"enabled"`
	DryRun    bool      `json:"dryRun"`
	Channels  []string  `json:"channels"`
	LastRun   time.Time `json:"lastRun"`
	LastError string    `json:"lastError,omitempty"`
	Alerts    []Alert   `json:"alerts"`
}

func (r *Runner) Status() Status {
	r.mu.Lock()
	defer r.mu.Unlock()
	s := Status{Enabled: true, DryRun: r.cfg.DryRun, LastRun: r.lastRun, LastError: r.lastError, Channels: []string{}}
	for _, n := range r.notifiers {
		s.Channels = append(s.Channels, n.Name())
	}
	s.Alerts = make([]Alert, len(r.recent))
	for i, a := range r.recent { // newest first
		s.Alerts[len(r.recent)-1-i] = a
	}
	return s
}

// HandleStatus serves recent alerts and runner health.
func (r *Runner) HandleStatus(w http.ResponseWriter, req *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(r.Status())
}

// HandleTest sends a clearly marked test alert to every channel, so a new
// token or chat id can be verified without waiting for a real incident.
func (r *Runner) HandleTest(w http.ResponseWriter, req *http.Request) {
	if req.Method != http.MethodPost {
		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
		return
	}
	now := time.Now().UTC()
	a := NewAlert(features.Spike{Service: "alert-test", WindowStart: now.Add(-r.cfg.Window), WindowEnd: now, ErrorCount: 42, BaselineMean: 1, Excess: 41, Patterns: []features.PatternDelta{{
		Signature: "test", Count: 42, BaselineMean: 1, Excess: 41, Novel: true, Clients: 3,
		Example: features.Event{Service: "alert-test", Kind: "span", StatusCode: 500, Route: "/api/example", Message: "this is a test alert"},
	}}}, now)
	a.Test = true
	r.askLaya(req.Context(), &a)
	results := map[string]string{}
	for _, n := range r.notifiers {
		if err := n.Send(req.Context(), a); err != nil {
			results[n.Name()] = err.Error()
		} else {
			results[n.Name()] = "ok"
		}
	}
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(results)
}