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 / 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 %s ยท %s 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("\nAll issues", esc(base+"/issues")) } if a.Laya != nil { head += fmt.Sprintf("๐Ÿค– laya (shadow, no action taken): %s %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 += " other: " + esc(r) + "\n" } } var b strings.Builder b.WriteString(head) for i, p := range a.Patterns { var s strings.Builder fmt.Fprintf(&s, "\n%d. %s\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, "๐Ÿ›  %s\n", esc(advice(p.Category, p.Status))) if base != "" && p.IssueID != "" { fmt.Fprintf(&s, "๐Ÿ”— Open issue\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("\nOther detected patterns\n") for _, other := range a.Patterns[i:] { var compact strings.Builder fmt.Fprintf(&compact, "ยท %s โ€” %d now, normally %.0f", esc(other.What), other.Count, other.Baseline) if base != "" && other.IssueID != "" { fmt.Fprintf(&compact, " ยท open issue", 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, "") } } 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 ")) } }