diff --git a/cmd/server/main.go b/cmd/server/main.go index a97cb3e..8f0e80b 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -282,7 +282,7 @@ func main() { // configured or OTEL_ALERT_DRY_RUN=true. var alertRunner *alerts.Runner if chStore, ok := store.(*storage.ClickHouseStore); ok { - alertRunner = alerts.Start(context.Background(), alerts.ConfigFromEnv(), chStore.Conn(), chStore.Database(), topo) + alertRunner = alerts.Start(context.Background(), alertConfig(coll), chStore.Conn(), chStore.Database(), topo) } // Update collector with wsServer coll.SetBroadcaster(wsServer) @@ -534,3 +534,10 @@ func spaHandler(root http.FileSystem) http.Handler { fileServer.ServeHTTP(w, r) }) } + +// alertConfig is the env-driven alert config plus muting of ignored issues. +func alertConfig(coll *collector.Collector) alerts.Config { + cfg := alerts.ConfigFromEnv() + cfg.Muted = coll.IsIssueIgnored + return cfg +} diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index 5f8b774..6c54251 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -279,3 +279,64 @@ func TestEdgeAlertShowsIngressBackends(t *testing.T) { t.Fatalf("non-edge alert enriched") } } + +type fakePodResolver struct { + got []struct{ namespace, deployment string } +} + +func (f *fakePodResolver) PodLogs(ctx context.Context, namespace, deployment string, tail int) topology.PodLog { + f.got = append(f.got, struct{ namespace, deployment string }{namespace, deployment}) + switch deployment { + case "orders-v2": + return topology.PodLog{Pod: "orders-v2-7d8f9-abcde", Found: true, Lines: []string{"line1", "line2", "line3"}} + default: + return topology.PodLog{} // scaled to 0 + } +} + +func TestAlertShowsPodLogsOrScaledToZero(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + ev := burst("bad-gw", 16, at.Add(-2*time.Minute), 502, "/api/order") + for i := range ev { + ev[i].Service, ev[i].Host = "nginx", "shop.example.com" + } + sp := DetectAlerts(ev, at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5}) + a := NewAlert(sp[0], at) + a.Enrich(&fakeResolver{}) + pr := &fakePodResolver{} + a.EnrichPods(context.Background(), pr) + + txt := a.Text("") + if !strings.Contains(txt, "orders-v2: pod orders-v2-7d8f9-abcde, last 3 log line(s):") { + t.Fatalf("missing pod logs in:\n%s", txt) + } + for _, want := range []string{"line1", "line2", "line3"} { + if !strings.Contains(txt, want) { + t.Fatalf("missing log line %q in:\n%s", want, txt) + } + } + if len(pr.got) != 1 || pr.got[0].namespace != "shop" || pr.got[0].deployment != "orders-v2" { + t.Fatalf("unexpected lookups: %+v", pr.got) + } +} + +func TestPodLineNoPod(t *testing.T) { + got := podLine(PodStatus{Deployment: "orders-v2", Found: false}) + if got != "orders-v2: no pod running (scaled to 0?)" { + t.Fatalf("got %q", got) + } +} + +func TestDropMuted(t *testing.T) { + mk := func(id string, excess float64) features.PatternDelta { + return features.PatternDelta{Excess: excess, Example: features.Event{IssueFingerprint: id}} + } + spikes := []features.Spike{ + {Excess: 10, Patterns: []features.PatternDelta{mk("ignored", 6), mk("live", 4)}}, + {Excess: 5, Patterns: []features.PatternDelta{mk("ignored", 5)}}, + } + got := DropMuted(spikes, func(id string) bool { return id == "ignored" }) + if len(got) != 1 || len(got[0].Patterns) != 1 || got[0].Patterns[0].Example.IssueFingerprint != "live" || got[0].Excess != 4 { + t.Fatalf("unexpected result: %+v", got) + } +} diff --git a/internal/alerts/filter.go b/internal/alerts/filter.go index 93a7f4e..0b6cc4c 100644 --- a/internal/alerts/filter.go +++ b/internal/alerts/filter.go @@ -38,6 +38,32 @@ func FilterActionable(spikes []features.Spike, keepMuted bool, muted map[triage. return out } +// DropMuted removes patterns whose issue the user ignored, recomputing each +// spike's excess and dropping spikes left with nothing to report. A nil +// muted keeps everything. +func DropMuted(spikes []features.Spike, muted func(issueID string) bool) []features.Spike { + if muted == nil { + return spikes + } + var out []features.Spike + for _, sp := range spikes { + var kept []features.PatternDelta + sp.Excess = 0 + for _, p := range sp.Patterns { + if id := p.Example.IssueFingerprint; id != "" && muted(id) { + continue + } + kept = append(kept, p) + sp.Excess += p.Excess + } + if len(kept) > 0 { + sp.Patterns = kept + out = append(out, sp) + } + } + return out +} + // Priority ranks how urgently a pattern needs a human. Client errors (4xx) // are the caller's problem more often than ours, so they rank below server // failures and count for less when deciding whether to page. diff --git a/internal/alerts/notify.go b/internal/alerts/notify.go index d249662..789d5ea 100644 --- a/internal/alerts/notify.go +++ b/internal/alerts/notify.go @@ -41,6 +41,7 @@ type PatternReport struct { 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"` @@ -50,6 +51,17 @@ type PatternReport struct { 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 { @@ -222,6 +234,9 @@ func (a Alert) Text(baseURL string) string { 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)) @@ -283,6 +298,9 @@ func (a Alert) HTML(baseURL string, max int) string { 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)) } @@ -437,3 +455,48 @@ func (a *Alert) Enrich(r Resolver) { } } } + +// 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 ")) + } +} diff --git a/internal/alerts/runner.go b/internal/alerts/runner.go index 72b6b60..c3db39e 100644 --- a/internal/alerts/runner.go +++ b/internal/alerts/runner.go @@ -30,7 +30,9 @@ type Config struct { BaseURL string // UI link appended to messages, e.g. https://otel.k6n.net TelegramToken, TelegramChatID, TelegramThreadID string - WebhookURL, WebhookSecret string + // Muted, when set, reports issues a person ignored; their spikes never alert. + Muted func(issueID string) bool + WebhookURL, WebhookSecret string } func ConfigFromEnv() Config { @@ -181,6 +183,15 @@ func (r *Runner) Tick(ctx context.Context, now time.Time) { 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() @@ -220,6 +231,9 @@ func DetectAlerts(events []features.Event, at time.Time, cfg features.SpikeConfi 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.mu.Lock() r.recent = append(r.recent, a) if len(r.recent) > maxRecent { diff --git a/internal/collector/collector.go b/internal/collector/collector.go index e5aac47..2ae7346 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -222,6 +222,9 @@ func (c *Collector) SetIssueStatus(id, status string) (issues.Issue, bool) { return c.issues.SetStatus(id, status) } +// IsIssueIgnored reports whether the issue was ignored (used to mute alerts). +func (c *Collector) IsIssueIgnored(id string) bool { return c.issues.IsIgnored(id) } + // SetIssueStatusStore makes resolved/ignored decisions survive restarts. func (c *Collector) SetIssueStatusStore(ctx context.Context, store issues.StatusStore) error { return c.issues.SetStatusStore(ctx, store) diff --git a/internal/issues/tracker.go b/internal/issues/tracker.go index 95c30af..fc2d15a 100644 --- a/internal/issues/tracker.go +++ b/internal/issues/tracker.go @@ -190,6 +190,14 @@ func (t *Tracker) SetStatusStore(ctx context.Context, store StatusStore) error { return nil } +// IsIgnored reports whether a person ignored the issue. It also answers for +// issues not yet recreated since a restart, which alerting needs. +func (t *Tracker) IsIgnored(id string) bool { + t.mu.Lock() + defer t.mu.Unlock() + return t.triage[id].Status == StatusIgnored +} + // persistLocked records a decision. The write happens off the lock. func (t *Tracker) persistLocked(id string, st TriageState) { if st.Status == StatusOpen { diff --git a/internal/topology/podlogs.go b/internal/topology/podlogs.go new file mode 100644 index 0000000..c5a4e22 --- /dev/null +++ b/internal/topology/podlogs.go @@ -0,0 +1,87 @@ +package topology + +import ( + "bufio" + "context" + "sort" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// PodLog is what looking up a running pod behind a Deployment found. +type PodLog struct { + Pod string // empty unless Found + Lines []string // most recent log lines, oldest first + Found bool // a pod exists and was queried (Lines may still be empty: a quiet pod) + Err string // set when the lookup itself failed; Found is false +} + +// podLogTimeout bounds one lookup so a slow or unreachable API server never +// stalls alert delivery. +const podLogTimeout = 5 * time.Second + +// PodLogs finds a pod behind namespace/deployment and returns its last `tail` +// log lines. A Deployment with no live pods right now - scaled to 0, or mid +// rollout with none ready yet - comes back with Found=false and no Err: that +// absence is itself the answer, not a failure. +func (r *Registry) PodLogs(ctx context.Context, namespace, deployment string, tail int) PodLog { + if r == nil || r.client == nil { + return PodLog{} + } + ctx, cancel := context.WithTimeout(ctx, podLogTimeout) + defer cancel() + + dep, err := r.client.AppsV1().Deployments(namespace).Get(ctx, deployment, metav1.GetOptions{}) + if err != nil { + return PodLog{Err: err.Error()} + } + sel, err := metav1.LabelSelectorAsSelector(dep.Spec.Selector) + if err != nil { + return PodLog{Err: err.Error()} + } + pods, err := r.client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: sel.String()}) + if err != nil { + return PodLog{Err: err.Error()} + } + live := make([]corev1.Pod, 0, len(pods.Items)) + for _, p := range pods.Items { + if p.DeletionTimestamp == nil { + live = append(live, p) + } + } + if len(live) == 0 { + return PodLog{} // scaled to 0, or nothing ready yet + } + // Prefer a Running pod; among ties prefer the newest, most likely to hold + // the log lines behind whatever just triggered the alert. + sort.Slice(live, func(i, j int) bool { + ri, rj := live[i].Status.Phase == corev1.PodRunning, live[j].Status.Phase == corev1.PodRunning + if ri != rj { + return ri + } + return live[i].CreationTimestamp.After(live[j].CreationTimestamp.Time) + }) + pod := live[0] + container := "" + if len(pod.Spec.Containers) > 0 { + container = pod.Spec.Containers[0].Name + } + tailLines := int64(tail) + stream, err := r.client.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{Container: container, TailLines: &tailLines}).Stream(ctx) + if err != nil { + return PodLog{Pod: pod.Name, Found: true, Err: err.Error()} + } + defer stream.Close() + var lines []string + sc := bufio.NewScanner(stream) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for sc.Scan() { + lines = append(lines, sc.Text()) + } + if len(lines) > tail { + lines = lines[len(lines)-tail:] + } + return PodLog{Pod: pod.Name, Lines: lines, Found: true} +} diff --git a/internal/topology/watcher.go b/internal/topology/watcher.go index 86b55a2..53ec078 100644 --- a/internal/topology/watcher.go +++ b/internal/topology/watcher.go @@ -24,9 +24,10 @@ import ( // Registry serves the latest Snapshot. A nil *Registry resolves nothing, so // callers need no cluster to run. type Registry struct { - snap atomic.Pointer[Snapshot] - dirty atomic.Bool - built atomic.Int64 // unix seconds of the last rebuild + snap atomic.Pointer[Snapshot] + dirty atomic.Bool + built atomic.Int64 // unix seconds of the last rebuild + client kubernetes.Interface // for on-demand pod lookups; nil outside a cluster } // Resolve returns the backends serving host+path. @@ -79,7 +80,7 @@ func NewFromEnv(ctx context.Context) (*Registry, error) { svc := factory.Core().V1().Services().Informer() dep := factory.Apps().V1().Deployments().Informer() - r := &Registry{} + r := &Registry{client: client} mark := cache.ResourceEventHandlerFuncs{ AddFunc: func(any) { r.dirty.Store(true) }, UpdateFunc: func(_, _ any) { r.dirty.Store(true) }, diff --git a/otel-ui/src/components/Tabs.tsx b/otel-ui/src/components/Tabs.tsx index cfcd82c..dd2d3b0 100644 --- a/otel-ui/src/components/Tabs.tsx +++ b/otel-ui/src/components/Tabs.tsx @@ -291,7 +291,7 @@ export const TabsComponent = () => { body: JSON.stringify({ status }), }) if (!res.ok) throw new Error(`HTTP ${res.status}`) - const verb = status === "resolved" ? "Resolved" : status === "ignored" ? "Ignored" : "Reopened" + const verb = status === "resolved" ? "Resolved" : status === "ignored" ? "Ignored (alerts muted)" : "Reopened" setStatusNotice({ text: `${verb}: ${issue.title}`, undo: status === "open" ? undefined : () => void handleIssueStatusChange({ ...issue, status }, previous),