go-otel/internal/alerts/notify.go

package alerts

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"html"
	"io"
	"net/http"
	"net/url"
	"sort"
	"strconv"
	"strings"
	"time"

	"git.tornberg.me/go-otel/internal/features"
	"git.tornberg.me/go-otel/internal/topology"
	"git.tornberg.me/go-otel/internal/tracectx"
	"git.tornberg.me/go-otel/internal/triage"
	"git.tornberg.me/go-otel/internal/uaclass"
)

// Alert is one notification: a spike reduced to its actionable patterns.
type Alert struct {
	ID       string          `json:"id"`
	At       time.Time       `json:"at"`
	Spike    features.Spike  `json:"spike"`
	Patterns []PatternReport `json:"patterns"`
	Test     bool            `json:"test,omitempty"`
	// Laya is the laya model's shadow verdict, when it was asked. Advisory
	// only: nothing is acted on or suppressed because of it.
	Laya *LayaVerdict `json:"laya,omitempty"`
}

// PatternReport is a spike pattern with its triage verdict, ready to show.
type PatternReport struct {
	Category string            `json:"category"`
	Priority Priority          `json:"priority"`
	IssueID  string            `json:"issueId,omitempty"`
	Status   int               `json:"status,omitempty"`
	Host     string            `json:"host,omitempty"`
	Path     string            `json:"path,omitempty"`
	Services []string          `json:"services,omitempty"` // every service failing in these requests
	Routes   []topology.Target `json:"routes,omitempty"`   // ingress -> service -> deployment behind an edge error
	Pods     []PodStatus       `json:"pods,omitempty"`     // what each identified deployment's pod is doing right now
	Rule     string            `json:"rule"`
	New      bool              `json:"new"`
	Count    int               `json:"count"`
	Baseline float64           `json:"baseline"`
	Clients  int               `json:"clients,omitempty"`
	UAClass  string            `json:"uaClass,omitempty"`
	What     string            `json:"what"`
}

// PodStatus is what was found when looking for a running pod behind a
// deployment Enrich identified.
type PodStatus struct {
	Namespace  string   `json:"namespace"`
	Deployment string   `json:"deployment"`
	Pod        string   `json:"pod,omitempty"`
	Lines      []string `json:"lines,omitempty"`
	Found      bool     `json:"found"`
	Err        string   `json:"err,omitempty"`
}

// Notifier delivers an alert somewhere. Implementations must not put secrets
// in returned errors.
type Notifier interface {
	Name() string
	Send(ctx context.Context, a Alert) error
}

// NewAlert builds an Alert from a spike, classifying each pattern.
func NewAlert(spike features.Spike, at time.Time) Alert {
	a := Alert{At: at, Spike: spike, ID: alertID(spike, at)}
	for _, p := range spike.Patterns {
		res := TriageOf(p.Example)
		r := PatternReport{Category: string(res.Category), Priority: PriorityOf(p.Example, res.Category), IssueID: p.Example.IssueFingerprint, Status: p.Example.StatusCode, Host: p.Example.Host, Path: p.Example.Route, Services: p.Services, Rule: res.Rule, New: p.Novel, Count: p.Count, Baseline: p.BaselineMean, Clients: p.Clients, What: describe(p.Example)}
		if p.Example.UAClass != "" && uaclass.IsBot(p.Example.UAClass) {
			r.UAClass = p.Example.UAClass
		}
		a.Patterns = append(a.Patterns, r)
	}
	// Most urgent first; the detector's excess order is kept within a priority.
	sort.SliceStable(a.Patterns, func(i, j int) bool { return prioRank(a.Patterns[i].Priority) < prioRank(a.Patterns[j].Priority) })
	return a
}

func prioRank(p Priority) int {
	switch p {
	case PriorityHigh:
		return 0
	case PriorityMedium:
		return 1
	}
	return 2
}

// Priority is the most urgent priority among the alert's patterns.
func (a Alert) Priority() Priority {
	if len(a.Patterns) == 0 {
		return PriorityLow
	}
	return a.Patterns[0].Priority
}

func alertID(s features.Spike, at time.Time) string {
	return strconv.FormatInt(at.Unix(), 36) + "-" + strings.ReplaceAll(s.Service, " ", "_")
}

func describe(e features.Event) string {
	parts := []string{}
	if e.StatusCode != 0 {
		parts = append(parts, "HTTP "+strconv.Itoa(e.StatusCode))
	}
	switch {
	case e.Route != "":
		parts = append(parts, e.Route)
	case e.SpanName != "":
		parts = append(parts, e.SpanName)
	}
	if e.ExceptionType != "" {
		parts = append(parts, e.ExceptionType)
	}
	if e.Message != "" && e.Message != e.SpanName {
		parts = append(parts, truncate(e.Message, 120))
	}
	if len(parts) == 0 {
		if e.Service != "" {
			return "Errors in " + e.Service + " without a message, route or status (see the issue for the trace)"
		}
		return "Errors without a message, route or status"
	}
	return strings.Join(parts, " | ")
}

func truncate(s string, n int) string {
	r := []rune(s)
	if len(r) <= n {
		return s
	}
	return string(r[:n-1]) + "…"
}

// advice says what a human should do about a category. Wording is plain so it
// can be acted on without knowing the detector's vocabulary.
func advice(category string, status int) string {
	if status >= 400 && status < 500 {
		return "Client errors (4xx): usually bad requests or stale links, rarely our fault. Only act if the count keeps growing."
	}
	switch triage.Category(category) {
	case triage.DependencyUnreachable:
		return "Something this service depends on (database, upstream service or network) is failing. Check the health of the services listed here, starting with the deepest one."
	case triage.CapacitySaturation:
		return "Requests are timing out or resources are exhausted. Check load, connection pools and CPU/memory limits."
	case triage.ServiceDefect:
		return "Most likely a bug or crash in this service. Open the issue for the stack trace and check whether a recent release introduced it."
	case triage.ClientCredentialIssue:
		return "Callers are being rejected as unauthenticated. Check for an expired token or key."
	}
	return "Unclassified error. Open the issue and decide whether it matters."
}

// routeLine explains where an edge error was routed and, for backends that
// reported no failing spans, how to check them directly: an uninstrumented
// service is invisible to the trace, so it must be looked at by hand.
func routeLine(t topology.Target, failing []string) string {
	where := t.Host + t.Path + " -> service " + t.Namespace + "/" + t.Service
	switch {
	case t.Missing:
		return where + ": this Service does not exist, so the ingress rule points at nothing."
	case t.External != "":
		return where + " -> external " + t.External + " (outside the cluster)."
	case len(t.Deployments) == 0:
		return where + " (no Deployment selected)."
	}
	var silent []string
	for _, d := range t.Deployments {
		if !contains(failing, d) && !contains(failing, t.Service) {
			silent = append(silent, d)
		}
	}
	line := where + " -> deployment " + strings.Join(t.Deployments, ", ")
	if len(silent) > 0 {
		line += ". No errors from it in the failing requests, it may not be instrumented: kubectl -n " + t.Namespace + " get pods -l <selector> / logs deploy/" + silent[0]
	}
	return line + "."
}

func contains(list []string, v string) bool {
	for _, x := range list {
		if x == v {
			return true
		}
	}
	return false
}

func verb(p PatternReport) string {
	if p.New {
		return "New error, never seen in the last few hours."
	}
	return fmt.Sprintf("Rising: %d now, normally about %.0f per window.", p.Count, p.Baseline)
}

// Text renders a plain-text message (no markup, so nothing needs escaping).
// Each pattern says what failed, how it differs from normal, who and what is
// affected, what to do, and where to look.
func (a Alert) Text(baseURL string) string {
	var b strings.Builder
	prefix := strings.ToUpper(string(a.Priority()))
	if a.Test {
		prefix = "TEST"
	}
	window := a.Spike.WindowEnd.Sub(a.Spike.WindowStart).Round(time.Minute)
	fmt.Fprintf(&b, "%s: %s is failing (%d errors in the last %s, normally about %.0f)\n",
		prefix, a.Spike.Service, a.Spike.ErrorCount, window, a.Spike.BaselineMean)
	if a.Laya != nil {
		fmt.Fprintf(&b, "laya (shadow, no action taken): %s\n", a.Laya.Summary())
	}
	base := strings.TrimRight(baseURL, "/")
	for i, p := range a.Patterns {
		fmt.Fprintf(&b, "\n%d. %s\n", i+1, p.What)
		fmt.Fprintf(&b, "   %s\n", verb(p))
		var who []string
		if len(p.Services) > 1 {
			who = append(who, "Services with errors in the same requests: "+strings.Join(p.Services, ", "))
		}
		if p.Clients > 0 {
			who = append(who, fmt.Sprintf("%d client%s affected", p.Clients, plural(p.Clients)))
		}
		if p.UAClass != "" {
			who = append(who, "client looks like a bot ("+p.UAClass+")")
		}
		for _, w := range who {
			fmt.Fprintf(&b, "   %s.\n", w)
		}
		for _, t := range p.Routes {
			fmt.Fprintf(&b, "   %s\n", routeLine(t, p.Services))
		}
		for _, ps := range p.Pods {
			fmt.Fprintf(&b, "   %s\n", podLine(ps))
		}
		fmt.Fprintf(&b, "   What to do: %s\n", advice(p.Category, p.Status))
		if base != "" && p.IssueID != "" {
			fmt.Fprintf(&b, "   %s/issues/%s\n", base, url.PathEscape(p.IssueID))
		}
	}
	if base != "" {
		fmt.Fprintf(&b, "\nAll issues: %s/issues", base)
	}
	return strings.TrimSpace(b.String())
}

func plural(n int) string {
	if n == 1 {
		return ""
	}
	return "s"
}

func priorityIcon(p Priority) string {
	switch p {
	case PriorityHigh:
		return "🔴"
	case PriorityMedium:
		return "🟠"
	}
	return "🟡"
}

// HTML renders the alert for Telegram's HTML parse mode. Every dynamic value
// is escaped, and patterns are added whole until the message would exceed max
// characters, so a cut never lands inside a tag.
func (a Alert) HTML(baseURL string, max int) string {
	esc := html.EscapeString
	window := a.Spike.WindowEnd.Sub(a.Spike.WindowStart).Round(time.Minute)
	icon, label := priorityIcon(a.Priority()), strings.ToUpper(string(a.Priority()))
	if a.Test {
		icon, label = "🧪", "TEST"
	}
	head := fmt.Sprintf("%s <b>%s</b> · <b>%s</b> is failing\n%d errors in the last %s (normally about %.0f)\n",
		icon, label, esc(a.Spike.Service), a.Spike.ErrorCount, window, a.Spike.BaselineMean)
	base := strings.TrimRight(baseURL, "/")
	foot := ""
	if base != "" {
		foot = fmt.Sprintf("\n<a href=\"%s\">All issues</a>", esc(base+"/issues"))
	}
	if a.Laya != nil {
		head += fmt.Sprintf("🤖 <b>laya</b> <i>(shadow, no action taken)</i>: <code>%s</code> %s · human %s · %s %s\n",
			esc(a.Laya.Action), pct(a.Laya.ActionConfidence), pct(a.Laya.Human), esc(a.Laya.Urgency), pct(a.Laya.UrgencyConf))
		if r := a.Laya.RunnersUp(); r != "" {
			head += "<i>   other: " + esc(r) + "</i>\n"
		}
	}
	var b strings.Builder
	b.WriteString(head)
	for i, p := range a.Patterns {
		var s strings.Builder
		fmt.Fprintf(&s, "\n<b>%d.</b> <code>%s</code>\n", i+1, esc(p.What))
		if p.New {
			s.WriteString("🆕 New error, not seen in the last few hours\n")
		} else {
			fmt.Fprintf(&s, "📈 Rising: %d now, normally about %.0f per window\n", p.Count, p.Baseline)
		}
		if len(p.Services) > 1 {
			fmt.Fprintf(&s, "🧩 Failing together: %s\n", esc(strings.Join(p.Services, ", ")))
		}
		for _, t := range p.Routes {
			fmt.Fprintf(&s, "🧭 %s\n", esc(routeLine(t, p.Services)))
		}
		for _, ps := range p.Pods {
			fmt.Fprintf(&s, "📜 %s\n", esc(podLine(ps)))
		}
		if p.Clients > 0 {
			fmt.Fprintf(&s, "👥 %d client%s affected\n", p.Clients, plural(p.Clients))
		}
		if p.UAClass != "" {
			fmt.Fprintf(&s, "🤖 Client looks like a bot (%s)\n", esc(p.UAClass))
		}
		fmt.Fprintf(&s, "🛠 <i>%s</i>\n", esc(advice(p.Category, p.Status)))
		if base != "" && p.IssueID != "" {
			fmt.Fprintf(&s, "🔗 <a href=\"%s\">Open issue</a>\n", esc(base+"/issues/"+url.PathEscape(p.IssueID)))
		}
		if plainLen(b.String())+plainLen(s.String())+plainLen(foot)+40 > max {
			remaining := len(a.Patterns) - i
			b.WriteString("\n<b>Other detected patterns</b>\n")
			for _, other := range a.Patterns[i:] {
				var compact strings.Builder
				fmt.Fprintf(&compact, "· <code>%s</code> — %d now, normally %.0f", esc(other.What), other.Count, other.Baseline)
				if base != "" && other.IssueID != "" {
					fmt.Fprintf(&compact, " · <a href=\"%s\">open issue</a>", esc(base+"/issues/"+url.PathEscape(other.IssueID)))
				}
				compact.WriteByte('\n')
				if plainLen(b.String())+plainLen(compact.String())+plainLen(foot)+10 > max {
					break
				}
				b.WriteString(compact.String())
			}
			if remaining > 0 && !strings.Contains(b.String(), "open issue") && base != "" {
				fmt.Fprintf(&b, "(%d additional pattern%s; open Issues for the full list)", remaining, plural(remaining))
			}
			break
		}
		b.WriteString(s.String())
	}
	b.WriteString(foot)
	return strings.TrimSpace(b.String())
}

// plainLen counts what Telegram counts: characters after tags are parsed away.
func plainLen(s string) int {
	n, in := 0, false
	for _, r := range html.UnescapeString(s) {
		switch {
		case r == '<':
			in = true
		case r == '>' && in:
			in = false
		case !in:
			n++
		}
	}
	return n
}

// Telegram posts via the Bot API sendMessage method.
type Telegram struct {
	Token    string
	ChatID   string
	ThreadID string // optional forum topic
	BaseURL  string // link appended to messages
	API      string // override for tests; default https://api.telegram.org
	Client   *http.Client
}

func (t *Telegram) Name() string { return "telegram" }

// telegram limits a message to 4096 characters.
const telegramMax = 4000

func (t *Telegram) Send(ctx context.Context, a Alert) error {
	api := t.API
	if api == "" {
		api = "https://api.telegram.org"
	}
	body := map[string]any{"chat_id": t.ChatID, "text": a.HTML(t.BaseURL, telegramMax), "parse_mode": "HTML", "disable_web_page_preview": true}
	if t.ThreadID != "" {
		if id, err := strconv.Atoi(t.ThreadID); err == nil {
			body["message_thread_id"] = id
		}
	}
	return postJSON(ctx, t.client(), api+"/bot"+t.Token+"/sendMessage", nil, body, t.Token)
}

func (t *Telegram) client() *http.Client {
	if t.Client != nil {
		return t.Client
	}
	return &http.Client{Timeout: 15 * time.Second}
}

// Webhook posts {"text": ..., "alert": ...} to any URL, optionally with a
// shared secret in the X-Alert-Secret header.
type Webhook struct {
	URL     string
	Secret  string
	BaseURL string
	Client  *http.Client
}

func (w *Webhook) Name() string { return "webhook" }

func (w *Webhook) Send(ctx context.Context, a Alert) error {
	client := w.Client
	if client == nil {
		client = &http.Client{Timeout: 15 * time.Second}
	}
	headers := map[string]string{}
	if w.Secret != "" {
		headers["X-Alert-Secret"] = w.Secret
	}
	return postJSON(ctx, client, w.URL, headers, map[string]any{"text": a.Text(w.BaseURL), "alert": a}, w.URL, w.Secret)
}

// postJSON posts a JSON body. Errors are rebuilt without the request URL and
// scrubbed of every secret, because both the Telegram token and webhook URLs
// commonly carry credentials.
func postJSON(ctx context.Context, client *http.Client, target string, headers map[string]string, body any, secrets ...string) error {
	raw, err := json.Marshal(body)
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, bytes.NewReader(raw))
	if err != nil {
		return errors.New("invalid request")
	}
	req.Header.Set("Content-Type", "application/json")
	for k, v := range headers {
		req.Header.Set(k, v)
	}
	resp, err := client.Do(req)
	if err != nil {
		var ue *url.Error
		if errors.As(err, &ue) {
			err = ue.Err // drop the URL the transport put in the message
		}
		return errors.New(scrub(err.Error(), secrets))
	}
	defer resp.Body.Close()
	if resp.StatusCode/100 != 2 {
		detail, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
		return fmt.Errorf("HTTP %d: %s", resp.StatusCode, scrub(strings.TrimSpace(string(detail)), secrets))
	}
	return nil
}

func scrub(s string, secrets []string) string {
	for _, sec := range secrets {
		if sec != "" {
			s = strings.ReplaceAll(s, sec, "<redacted>")
		}
	}
	return s
}

// Resolver maps a request host and path to the backends behind it.
type Resolver interface {
	Resolve(host, path string) []topology.Target
}

// Enrich attaches the ingress -> service -> deployment path to patterns of an
// edge service (nginx, ingress-nginx, ...), whose errors say little about
// which workload is at fault.
func (a *Alert) Enrich(r Resolver) {
	if r == nil || !tracectx.IsEdgeService(a.Spike.Service) {
		return
	}
	for i := range a.Patterns {
		if p := &a.Patterns[i]; p.Host != "" {
			p.Routes = r.Resolve(p.Host, p.Path)
		}
	}
}

// podLogTail is how many trailing log lines a notification carries per pod.
const podLogTail = 3

// PodResolver looks up a running pod behind a namespace/deployment and its
// most recent log lines.
type PodResolver interface {
	PodLogs(ctx context.Context, namespace, deployment string, tail int) topology.PodLog
}

// EnrichPods looks up a running pod for each deployment Enrich identified and
// attaches its last few log lines, so the notification shows what the
// backend is actually doing instead of just naming it. A deployment with no
// pod right now - scaled to 0, most likely - is reported as such.
func (a *Alert) EnrichPods(ctx context.Context, pr PodResolver) {
	if pr == nil {
		return
	}
	for i := range a.Patterns {
		p := &a.Patterns[i]
		for _, t := range p.Routes {
			if t.Missing || t.External != "" {
				continue
			}
			for _, d := range t.Deployments {
				pl := pr.PodLogs(ctx, t.Namespace, d, podLogTail)
				p.Pods = append(p.Pods, PodStatus{Namespace: t.Namespace, Deployment: d, Pod: pl.Pod, Lines: pl.Lines, Found: pl.Found, Err: pl.Err})
			}
		}
	}
}

// podLine renders one deployment's pod status for a notification.
func podLine(ps PodStatus) string {
	switch {
	case ps.Err != "":
		return fmt.Sprintf("%s: could not check for a pod (%s)", ps.Deployment, ps.Err)
	case !ps.Found:
		return fmt.Sprintf("%s: no pod running (scaled to 0?)", ps.Deployment)
	case len(ps.Lines) == 0:
		return fmt.Sprintf("%s: pod %s, no recent log lines", ps.Deployment, ps.Pod)
	default:
		return fmt.Sprintf("%s: pod %s, last %d log line(s):\n     %s", ps.Deployment, ps.Pod, len(ps.Lines), strings.Join(ps.Lines, "\n     "))
	}
}