diff --git a/cmd/server/main.go b/cmd/server/main.go index d98abd2..0522c8f 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -11,14 +11,15 @@ import ( "git.tornberg.me/go-otel/internal/alerts" "git.tornberg.me/go-otel/internal/auth" + "git.tornberg.me/go-otel/internal/blocker" "git.tornberg.me/go-otel/internal/collector" "git.tornberg.me/go-otel/internal/geoip" - "git.tornberg.me/go-otel/internal/blocker" "git.tornberg.me/go-otel/internal/honeypot" kube "git.tornberg.me/go-otel/internal/kubernetes" "git.tornberg.me/go-otel/internal/prometheus" "git.tornberg.me/go-otel/internal/sourcemaps" "git.tornberg.me/go-otel/internal/storage" + "git.tornberg.me/go-otel/internal/topology" "git.tornberg.me/go-otel/internal/websocket" "go.opentelemetry.io/collector/pdata/plog" "go.opentelemetry.io/collector/pdata/pmetric" @@ -224,6 +225,10 @@ func main() { coll.SetKubernetesWatcher(kubeWatcher) log.Printf("kubernetes: rollout watcher enabled") } + topo, topoErr := topology.NewFromEnv(context.Background()) + if topoErr != nil { + log.Printf("topology: ingress registry disabled: %v", topoErr) + } coll.SetPrometheusClient(prometheus.NewFromEnv()) sourceMaps, err := sourcemaps.New(os.Getenv("OTEL_SOURCEMAP_DIR")) if err != nil { @@ -248,7 +253,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()) + alertRunner = alerts.Start(context.Background(), alerts.ConfigFromEnv(), chStore.Conn(), chStore.Database(), topo) } // Update collector with wsServer coll.SetBroadcaster(wsServer) @@ -347,6 +352,7 @@ func main() { uiMux.Handle("GET /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueSimilar))) uiMux.Handle("POST /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSimilarityFeedback))) uiMux.Handle("GET /api/patterns", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandlePatterns))) + uiMux.Handle("GET /api/topology", oidcAuth.RequireAuth(http.HandlerFunc(topo.HandleTopology))) uiMux.Handle("GET /api/rollouts", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleRollouts))) uiMux.Handle("GET /api/rollouts/{id}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleRolloutByID))) uiMux.Handle("GET /api/rollouts/{id}/impact", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleRolloutImpact))) diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index e588ae3..5f8b774 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -11,6 +11,7 @@ import ( "time" "git.tornberg.me/go-otel/internal/features" + "git.tornberg.me/go-otel/internal/topology" ) func burst(sig string, n int, at time.Time, status int, route string) []features.Event { @@ -124,9 +125,23 @@ func TestTelegramTruncates(t *testing.T) { if err := (&Telegram{Token: "t", ChatID: "1", API: srv.URL}).Send(context.Background(), NewAlert(spike, time.Now())); err != nil { t.Fatal(err) } - if n := len([]rune(body["text"].(string))); n > 4096 { + text := body["text"].(string) + if n := plainLen(text); n > 4096 { t.Fatalf("message too long: %d", n) } + if body["parse_mode"] != "HTML" || !strings.Contains(text, "more pattern") || strings.Count(text, "") != strings.Count(text, "") { + t.Fatalf("bad html message: %.200s", text) + } +} + +func TestTelegramHTMLEscapesDynamicText(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + ev := burst("x", 12, at.Add(-2*time.Minute), 500, "/a&c") + sp := DetectAlerts(ev, at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5}) + out := NewAlert(sp[0], at).HTML("https://otel.k6n.net", 4000) + if strings.Contains(out, "/a") || !strings.Contains(out, "/a<b>&c") { + t.Fatalf("route not escaped:\n%s", out) + } } func TestScannerClientIsMutedEntirely(t *testing.T) { @@ -163,3 +178,104 @@ func TestMinExcessAppliesAfterTriage(t *testing.T) { t.Fatalf("paged on excess that was mostly probes: %+v", got) } } + +func TestClientErrorsRankBelowServerErrors(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + cur := at.Add(-2 * time.Minute) + cfg := features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 10} + // 40 x 422 discounted to 8 excess: below MinExcess 10, no page + if got := DetectAlerts(burst("bad-input", 40, cur, 422, "/api/order"), at, cfg); len(got) != 0 { + t.Fatalf("4xx alone paged: %+v", got) + } + // 5xx of the same volume does page + if got := DetectAlerts(burst("boom", 40, cur, 500, "/api/order"), at, cfg); len(got) != 1 { + t.Fatalf("5xx did not page") + } + // mixed: the 5xx is listed first and the alert is high priority + events := append(burst("bad-input", 40, cur, 422, "/api/order"), burst("boom", 8, cur, 500, "/api/pay")...) + got := DetectAlerts(events, at, cfg) + if len(got) != 1 { + t.Fatalf("mixed spike: %+v", got) + } + a := NewAlert(got[0], at) + if a.Priority() != PriorityHigh || a.Patterns[0].Priority != PriorityHigh || a.Patterns[1].Priority != PriorityLow { + t.Fatalf("priority order wrong: %+v", a.Patterns) + } +} + +func TestAlertLinksToIssue(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + ev := burst("boom", 12, at.Add(-2*time.Minute), 500, "/api") + for i := range ev { + ev[i].IssueFingerprint = "00ab12cd34ef5678" + } + sp := DetectAlerts(ev, at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5}) + txt := NewAlert(sp[0], at).Text("https://otel.k6n.net/") + if !strings.Contains(txt, "https://otel.k6n.net/issues/00ab12cd34ef5678") { + t.Fatalf("no issue link:\n%s", txt) + } +} + +func TestCrawlerClientErrorsNeverAlert(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + ev := burst("crawl", 100, at.Add(-2*time.Minute), 403, "/members") + for i := range ev { + ev[i].UAClass = "ai_crawler" + } + if got := DetectAlerts(ev, at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5}); len(got) != 0 { + t.Fatalf("crawler 4xx alerted: %+v", got) + } +} + +func TestAlertListsAffectedServices(t *testing.T) { + at := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + ev := burst("boom", 12, at.Add(-2*time.Minute), 500, "/api") + for i := range ev { + ev[i].TraceServices = []string{"payments", "api"} + } + sp := DetectAlerts(ev, at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5}) + a := NewAlert(sp[0], at) + if got := strings.Join(a.Patterns[0].Services, ","); got != "api,payments" { + t.Fatalf("services = %q", got) + } + if !strings.Contains(a.Text(""), "Services with errors in the same requests: api, payments") { + t.Fatalf("text lacks services:\n%s", a.Text("")) + } +} + +type fakeResolver struct{ got [2]string } + +func (f *fakeResolver) Resolve(host, path string) []topology.Target { + f.got = [2]string{host, path} + return []topology.Target{ + {Host: host, Path: "/api", Namespace: "shop", Service: "orders", Deployments: []string{"orders-v2"}}, + {Host: host, Path: "/api", Namespace: "shop", Service: "gone", Missing: true}, + } +} + +func TestEdgeAlertShowsIngressBackends(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) + r := &fakeResolver{} + a.Enrich(r) + if r.got != [2]string{"shop.example.com", "/api/order"} { + t.Fatalf("resolved %v", r.got) + } + txt := a.Text("") + for _, want := range []string{"service shop/orders -> deployment orders-v2", "may not be instrumented", "kubectl -n shop", "this Service does not exist"} { + if !strings.Contains(txt, want) { + t.Fatalf("missing %q in:\n%s", want, txt) + } + } + // a non-edge service is never enriched + b := NewAlert(DetectAlerts(burst("x", 12, at.Add(-2*time.Minute), 500, "/"), at, features.SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12, MinExcess: 5})[0], at) + b.Enrich(r) + if len(b.Patterns[0].Routes) != 0 { + t.Fatalf("non-edge alert enriched") + } +} diff --git a/internal/alerts/filter.go b/internal/alerts/filter.go index 3b2c5cd..93a7f4e 100644 --- a/internal/alerts/filter.go +++ b/internal/alerts/filter.go @@ -37,3 +37,44 @@ func FilterActionable(spikes []features.Spike, keepMuted bool, muted map[triage. } 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. +type Priority string + +const ( + PriorityHigh Priority = "high" + PriorityMedium Priority = "medium" + PriorityLow Priority = "low" +) + +// lowPriorityWeight is how much a low-priority pattern's excess counts toward +// the paging threshold: five 4xx errors weigh as one 5xx. +const lowPriorityWeight = 0.2 + +// PriorityOf ranks an example event: 4xx is low; server failures (5xx, or a +// service defect, saturation or unreachable dependency) are high. +func PriorityOf(e features.Event, c triage.Category) Priority { + switch { + case e.StatusCode >= 400 && e.StatusCode < 500: + return PriorityLow + case e.StatusCode >= 500, c == triage.ServiceDefect, c == triage.CapacitySaturation, c == triage.DependencyUnreachable: + return PriorityHigh + } + return PriorityMedium +} + +// PagingExcess is a spike's excess with low-priority patterns discounted, so a +// pile of 4xx cannot page on its own but still tips a real 5xx spike. +func PagingExcess(sp features.Spike) float64 { + var total float64 + for _, p := range sp.Patterns { + if PriorityOf(p.Example, TriageOf(p.Example).Category) == PriorityLow { + total += p.Excess * lowPriorityWeight + } else { + total += p.Excess + } + } + return total +} diff --git a/internal/alerts/load.go b/internal/alerts/load.go index 1d99e0b..70e5aa2 100644 --- a/internal/alerts/load.go +++ b/internal/alerts/load.go @@ -48,7 +48,7 @@ func LoadErrorEvents(ctx context.Context, conn driver.Conn, database string, sta if withTrace { extra = "span_id, parent_span_id, peer_ip, ua_class, user_agent, trace_id" } - rows, err := conn.Query(ctx, fmt.Sprintf(`SELECT timestamp, service, release, kind, source, status_code, span_name, route, exception_type, message_normalized, pattern_signature, is_timeout, is_cancellation, is_retry, %s + rows, err := conn.Query(ctx, fmt.Sprintf(`SELECT timestamp, service, release, kind, source, status_code, span_name, route, exception_type, message_normalized, pattern_signature, is_timeout, is_cancellation, is_retry, issue_fingerprint, host, %s FROM %s.normalized_events FINAL WHERE is_error = 1 AND timestamp >= toDateTime64('%s', 9, 'UTC') AND timestamp < toDateTime64('%s', 9, 'UTC') ORDER BY timestamp LIMIT 2000000`, extra, database, chTime(start), chTime(end))) @@ -64,7 +64,7 @@ func LoadErrorEvents(ctx context.Context, conn driver.Conn, database string, sta var status uint16 var timeout, cancel, retry uint8 var spanID, parentID, peer, class, ua, traceID string - if err := rows.Scan(&e.Timestamp, &e.Service, &e.Release, &e.Kind, &e.Source, &status, &e.SpanName, &e.Route, &e.ExceptionType, &e.Message, &e.PatternSignature, &timeout, &cancel, &retry, + if err := rows.Scan(&e.Timestamp, &e.Service, &e.Release, &e.Kind, &e.Source, &status, &e.SpanName, &e.Route, &e.ExceptionType, &e.Message, &e.PatternSignature, &timeout, &cancel, &retry, &e.IssueFingerprint, &e.Host, &spanID, &parentID, &peer, &class, &ua, &traceID); err != nil { return nil, err } @@ -118,8 +118,20 @@ func resolve(ctx context.Context, conn driver.Conn, database string, start, end if err := rows.Err(); err != nil { return err } + failing := map[string]map[string]struct{}{} // trace -> services with an error span + for _, sp := range spans { + if sp.IsError && sp.Service != "" { + if failing[sp.TraceID] == nil { + failing[sp.TraceID] = map[string]struct{}{} + } + failing[sp.TraceID][sp.Service] = struct{}{} + } + } info := tracectx.Resolve(spans) for i, ref := range refs { + for sv := range failing[ref.TraceID] { + events[i].TraceServices = append(events[i].TraceServices, sv) + } in, ok := info[ref] if !ok { if !honeypot.Untrackable(own[i]) { diff --git a/internal/alerts/notify.go b/internal/alerts/notify.go index 3030b1b..d249662 100644 --- a/internal/alerts/notify.go +++ b/internal/alerts/notify.go @@ -6,14 +6,19 @@ import ( "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" ) @@ -28,14 +33,21 @@ type Alert struct { // PatternReport is a spike pattern with its triage verdict, ready to show. type PatternReport struct { - Category string `json:"category"` - 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"` + 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 + 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"` } // Notifier delivers an alert somewhere. Implementations must not put secrets @@ -50,15 +62,35 @@ 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), Rule: res.Rule, New: p.Novel, Count: p.Count, Baseline: p.BaselineMean, Clients: p.Clients, What: describe(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, " ", "_") } @@ -81,7 +113,10 @@ func describe(e features.Event) string { parts = append(parts, truncate(e.Message, 120)) } if len(parts) == 0 { - return "(no detail)" + 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, " | ") } @@ -94,35 +129,106 @@ func truncate(s string, n int) string { 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 := "ALERT" + prefix := strings.ToUpper(string(a.Priority())) if a.Test { prefix = "TEST" } - fmt.Fprintf(&b, "%s %s: %d errors in %s (baseline %.1f/window, excess %.0f)\n", - prefix, a.Spike.Service, a.Spike.ErrorCount, a.Spike.WindowEnd.Sub(a.Spike.WindowStart).Round(time.Minute), a.Spike.BaselineMean, a.Spike.Excess) - for _, p := range a.Patterns { - tag := "rising" - if p.New { - tag = "NEW" + 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) + 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, ", ")) } - fmt.Fprintf(&b, "\n%s x%d (base %.1f) [%s]\n%s", tag, p.Count, p.Baseline, p.Category, p.What) - var extra []string if p.Clients > 0 { - extra = append(extra, fmt.Sprintf("%d client%s", p.Clients, plural(p.Clients))) + who = append(who, fmt.Sprintf("%d client%s affected", p.Clients, plural(p.Clients))) } if p.UAClass != "" { - extra = append(extra, p.UAClass) + who = append(who, "client looks like a bot ("+p.UAClass+")") } - if len(extra) > 0 { - fmt.Fprintf(&b, "\n(%s)", strings.Join(extra, ", ")) + for _, w := range who { + fmt.Fprintf(&b, " %s.\n", w) + } + for _, t := range p.Routes { + fmt.Fprintf(&b, " %s\n", routeLine(t, p.Services)) + } + 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)) } - b.WriteString("\n") } - if baseURL != "" { - fmt.Fprintf(&b, "\n%s/issues", strings.TrimRight(baseURL, "/")) + if base != "" { + fmt.Fprintf(&b, "\nAll issues: %s/issues", base) } return strings.TrimSpace(b.String()) } @@ -134,6 +240,85 @@ func plural(n int) string { 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")) + } + 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))) + } + 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 { + fmt.Fprintf(&b, "\nโ€ฆ and %d more pattern%s, see the issues list", len(a.Patterns)-i, plural(len(a.Patterns)-i)) + 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 @@ -154,7 +339,7 @@ func (t *Telegram) Send(ctx context.Context, a Alert) error { if api == "" { api = "https://api.telegram.org" } - body := map[string]any{"chat_id": t.ChatID, "text": truncate(a.Text(t.BaseURL), telegramMax), "disable_web_page_preview": true} + 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 @@ -233,3 +418,22 @@ func scrub(s string, secrets []string) string { } 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) + } + } +} diff --git a/internal/alerts/runner.go b/internal/alerts/runner.go index 84dd58b..72b6b60 100644 --- a/internal/alerts/runner.go +++ b/internal/alerts/runner.go @@ -98,6 +98,7 @@ type Runner struct { load Loader dataSince func(ctx context.Context) time.Time notifiers []Notifier + resolver Resolver // optional: ingress -> service -> deployment for edge errors sup *features.Suppressor mu sync.Mutex @@ -113,7 +114,7 @@ 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) *Runner { +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)") @@ -122,6 +123,7 @@ func Start(ctx context.Context, cfg Config, conn driver.Conn, database string) * 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) @@ -205,10 +207,11 @@ func DetectAlerts(events []features.Event, at time.Time, cfg features.SpikeConfi minExcess = features.DefaultSpikeConfig().MinExcess } var out []features.Spike - // The excess threshold is applied after triage: a service whose total - // excess is mostly probes must not page on the little that is left. + // 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 sp.Excess >= float64(minExcess) { + if PagingExcess(sp) >= float64(minExcess) { out = append(out, sp) } } @@ -216,6 +219,7 @@ func DetectAlerts(events []features.Event, at time.Time, cfg features.SpikeConfi } func (r *Runner) deliver(ctx context.Context, a Alert) { + a.Enrich(r.resolver) r.mu.Lock() r.recent = append(r.recent, a) if len(r.recent) > maxRecent { diff --git a/internal/features/features.go b/internal/features/features.go index 06d10ad..8aa0f82 100644 --- a/internal/features/features.go +++ b/internal/features/features.go @@ -31,17 +31,19 @@ type Event struct { Message string SpanName string Route string + Host string // request host (Host / :authority), used to find the ingress rule behind an edge error StatusCode int ExceptionType string // Trace-resolved context (see internal/tracectx). All optional: zero // values mean "not resolved", which behaves like the pre-resolution data. - ClientIP string // real client behind proxy hops, when known - UAClass string // internal/uaclass class of the User-Agent - IsDuplicate bool // repeats an ancestor span's failure; not a new request - IsRootCause bool // error span with no erroring child - HasUpstream bool // an ingress span that had a child in another service - BadActor string // reason this event is attributed to a bad actor (see MarkBadActors); empty if not + ClientIP string // real client behind proxy hops, when known + UAClass string // internal/uaclass class of the User-Agent + IsDuplicate bool // repeats an ancestor span's failure; not a new request + IsRootCause bool // error span with no erroring child + HasUpstream bool // an ingress span that had a child in another service + TraceServices []string // every service with an error span in this event's trace (includes its own) + BadActor string // reason this event is attributed to a bad actor (see MarkBadActors); empty if not } // Config controls feature and label windows. A zero Config receives the diff --git a/internal/features/spike.go b/internal/features/spike.go index e8f5944..957eaf2 100644 --- a/internal/features/spike.go +++ b/internal/features/spike.go @@ -29,13 +29,14 @@ func DefaultSpikeConfig() SpikeConfig { // PatternDelta is one error pattern's contribution to a spike. type PatternDelta struct { - Signature string `json:"signature"` - Count int `json:"count"` - BaselineMean float64 `json:"baselineMean"` - Excess float64 `json:"excess"` - Novel bool `json:"novel"` // never seen in the baseline - Clients int `json:"clients,omitempty"` // distinct resolved client IPs in the current window - Example Event `json:"example"` + Signature string `json:"signature"` + Count int `json:"count"` + BaselineMean float64 `json:"baselineMean"` + Excess float64 `json:"excess"` + Novel bool `json:"novel"` // never seen in the baseline + Clients int `json:"clients,omitempty"` // distinct resolved client IPs in the current window + Services []string `json:"services,omitempty"` // services with an error in the same requests, sorted + Example Event `json:"example"` } type Spike struct { @@ -82,10 +83,11 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { } type agg struct { - cur int - base []int // per baseline window - example Event - clients map[string]struct{} + cur int + base []int // per baseline window + example Event + clients map[string]struct{} + services map[string]struct{} } type svc struct { cur int @@ -129,6 +131,13 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { s.cur++ p.cur++ p.example = e + if p.services == nil { + p.services = map[string]struct{}{} + } + p.services[name] = struct{}{} + for _, sv := range e.TraceServices { + p.services[sv] = struct{}{} + } if e.ClientIP != "" { if p.clients == nil { p.clients = map[string]struct{}{} @@ -166,7 +175,7 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { } spike.Excess += excess spike.Patterns = append(spike.Patterns, PatternDelta{ - Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: !seen, Example: p.example, Clients: len(p.clients), + Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: !seen, Example: p.example, Clients: len(p.clients), Services: sortedKeys(p.services), }) } if spike.Excess < float64(cfg.MinExcess) { @@ -257,3 +266,12 @@ func (s *Suppressor) Filter(spikes []Spike) []Spike { } return out } + +func sortedKeys(m map[string]struct{}) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} diff --git a/internal/kubernetes/watcher.go b/internal/kubernetes/watcher.go index cdfaccd..cd14f6e 100644 --- a/internal/kubernetes/watcher.go +++ b/internal/kubernetes/watcher.go @@ -313,3 +313,7 @@ func rolloutStatus(deployment *appsv1.Deployment) (phase, reason, message string } return phase, reason, message, false } + +// RESTConfig exposes the in-cluster / KUBECONFIG configuration so sibling +// packages observe the same cluster the rollout watcher does. +func RESTConfig() (*rest.Config, error) { return kubeConfig() } diff --git a/internal/topology/topology.go b/internal/topology/topology.go new file mode 100644 index 0000000..0de25e8 --- /dev/null +++ b/internal/topology/topology.go @@ -0,0 +1,194 @@ +// Package topology keeps a read-only registry of how traffic reaches +// workloads: Ingress rule -> Service -> Deployment. An error seen at the edge +// (nginx) carries only a host and a path; the registry says which services and +// deployments sit behind them, including ones that export no telemetry. +package topology + +import ( + "sort" + "strings" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" +) + +// Target is one backend an ingress rule routes to. +type Target struct { + Ingress string `json:"ingress"` // namespace/name + Host string `json:"host"` // "*" for a default backend + Path string `json:"path"` + PathType string `json:"pathType"` + Namespace string `json:"namespace"` + Service string `json:"service"` + Port string `json:"port,omitempty"` + External string `json:"external,omitempty"` // ExternalName service target: no workload in this cluster + Deployments []string `json:"deployments,omitempty"` // names in Namespace whose pods the Service selects + Missing bool `json:"missing,omitempty"` // the Service the rule names does not exist +} + +// Snapshot is an immutable routing table. +type Snapshot struct { + Targets []Target `json:"targets"` +} + +// Build joins ingresses, services and deployments. A Service selects a +// Deployment when the Service's selector is a subset of the pod template's +// labels in the same namespace; a Service without a selector selects nothing. +func Build(ingresses []networkingv1.Ingress, services []corev1.Service, deployments []appsv1.Deployment) Snapshot { + svc := map[string]corev1.Service{} + for _, s := range services { + svc[s.Namespace+"/"+s.Name] = s + } + var out []Target + add := func(ing networkingv1.Ingress, host, path, pathType string, b *networkingv1.IngressServiceBackend) { + if b == nil { + return // resource backends (buckets etc.) are not workloads + } + t := Target{Ingress: ing.Namespace + "/" + ing.Name, Host: host, Path: path, PathType: pathType, Namespace: ing.Namespace, Service: b.Name} + if b.Port.Name != "" { + t.Port = b.Port.Name + } else if b.Port.Number != 0 { + t.Port = itoa(int(b.Port.Number)) + } + s, ok := svc[ing.Namespace+"/"+b.Name] + switch { + case !ok: + t.Missing = true + case s.Spec.Type == corev1.ServiceTypeExternalName: + t.External = s.Spec.ExternalName + default: + t.Deployments = selected(s, deployments) + } + out = append(out, t) + } + for _, ing := range ingresses { + if d := ing.Spec.DefaultBackend; d != nil { + add(ing, "*", "/", "Prefix", d.Service) + } + for _, rule := range ing.Spec.Rules { + host := rule.Host + if host == "" { + host = "*" + } + if rule.HTTP == nil { + continue + } + for _, p := range rule.HTTP.Paths { + pt := "ImplementationSpecific" + if p.PathType != nil { + pt = string(*p.PathType) + } + path := p.Path + if path == "" { + path = "/" + } + add(ing, host, path, pt, p.Backend.Service) + } + } + } + sort.Slice(out, func(i, j int) bool { + a, b := out[i], out[j] + if a.Host != b.Host { + return a.Host < b.Host + } + if len(a.Path) != len(b.Path) { + return len(a.Path) > len(b.Path) + } + return a.Path < b.Path + }) + return Snapshot{Targets: out} +} + +func selected(s corev1.Service, deployments []appsv1.Deployment) []string { + if len(s.Spec.Selector) == 0 { + return nil + } + var out []string + for _, d := range deployments { + if d.Namespace != s.Namespace { + continue + } + labels := d.Spec.Template.Labels + match := true + for k, v := range s.Spec.Selector { + if labels[k] != v { + match = false + break + } + } + if match { + out = append(out, d.Name) + } + } + sort.Strings(out) + return out +} + +// Resolve returns the targets that serve a request for host and path, best +// match first: an exact host beats a wildcard, and within a host the longest +// matching path wins. Only the best-matching rule per ingress is returned +// unless several ingresses tie, which nginx would also treat as a conflict. +func (s Snapshot) Resolve(host, path string) []Target { + host = strings.ToLower(strings.TrimSuffix(host, ".")) + if i := strings.LastIndex(host, ":"); i > 0 && !strings.Contains(host[i:], "]") { + host = host[:i] + } + if path == "" { + path = "/" + } + if i := strings.IndexAny(path, "?#"); i >= 0 { + path = path[:i] + } + bestScore := -1 + var out []Target + for _, t := range s.Targets { + hs := hostScore(t.Host, host) + if hs < 0 || !pathMatches(t, path) { + continue + } + score := hs*100000 + len(t.Path) + switch { + case score > bestScore: + bestScore, out = score, []Target{t} + case score == bestScore: + out = append(out, t) + } + } + return out +} + +// hostScore is 2 for an exact host, 1 for a wildcard or catch-all, -1 for no match. +func hostScore(rule, host string) int { + switch { + case rule == "*": + return 0 + case strings.EqualFold(rule, host): + return 2 + case strings.HasPrefix(rule, "*.") && strings.HasSuffix(host, rule[1:]) && strings.Count(host, ".") == strings.Count(rule, "."): + return 1 + } + return -1 +} + +func pathMatches(t Target, path string) bool { + if t.PathType == "Exact" { + return path == t.Path + } + p := strings.TrimSuffix(t.Path, "/") + return p == "" || path == p || strings.HasPrefix(path, p+"/") +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + var b [12]byte + i := len(b) + for n > 0 { + i-- + b[i] = byte('0' + n%10) + n /= 10 + } + return string(b[i:]) +} diff --git a/internal/topology/topology_test.go b/internal/topology/topology_test.go new file mode 100644 index 0000000..4a20f0c --- /dev/null +++ b/internal/topology/topology_test.go @@ -0,0 +1,97 @@ +package topology + +import ( + "testing" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +func ptr[T any](v T) *T { return &v } + +func path(p string, pt networkingv1.PathType, svc string) networkingv1.HTTPIngressPath { + return networkingv1.HTTPIngressPath{Path: p, PathType: &pt, Backend: networkingv1.IngressBackend{ + Service: &networkingv1.IngressServiceBackend{Name: svc, Port: networkingv1.ServiceBackendPort{Number: 80}}}} +} + +func fixture() Snapshot { + ing := networkingv1.Ingress{ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "public"}, Spec: networkingv1.IngressSpec{Rules: []networkingv1.IngressRule{ + {Host: "shop.example.com", IngressRuleValue: networkingv1.IngressRuleValue{HTTP: &networkingv1.HTTPIngressRuleValue{Paths: []networkingv1.HTTPIngressPath{ + path("/", networkingv1.PathTypePrefix, "web"), + path("/api", networkingv1.PathTypePrefix, "api"), + path("/legacy", networkingv1.PathTypeExact, "old"), + path("/pay", networkingv1.PathTypePrefix, "gone"), + path("/cdn", networkingv1.PathTypePrefix, "cdn"), + }}}}, + {Host: "*.example.com", IngressRuleValue: networkingv1.IngressRuleValue{HTTP: &networkingv1.HTTPIngressRuleValue{Paths: []networkingv1.HTTPIngressPath{ + path("/", networkingv1.PathTypePrefix, "web"), + }}}}, + }}} + svcs := []corev1.Service{ + {ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "web"}, Spec: corev1.ServiceSpec{Selector: map[string]string{"app": "web"}}}, + {ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "api"}, Spec: corev1.ServiceSpec{Selector: map[string]string{"app": "api"}}}, + {ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "old"}, Spec: corev1.ServiceSpec{Selector: map[string]string{"app": "old"}}}, + {ObjectMeta: metav1.ObjectMeta{Namespace: "shop", Name: "cdn"}, Spec: corev1.ServiceSpec{Type: corev1.ServiceTypeExternalName, ExternalName: "cdn.vendor.net"}}, + } + dep := func(ns, name, app string) appsv1.Deployment { + return appsv1.Deployment{ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name}, Spec: appsv1.DeploymentSpec{ + Template: corev1.PodTemplateSpec{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{"app": app, "tier": "x"}}}}} + } + deps := []appsv1.Deployment{dep("shop", "web-v2", "web"), dep("shop", "api", "api"), dep("other", "web-other", "web"), dep("shop", "old", "old")} + return Build([]networkingv1.Ingress{ing}, svcs, deps) +} + +func one(t *testing.T, s Snapshot, host, path string) Target { + t.Helper() + got := s.Resolve(host, path) + if len(got) != 1 { + t.Fatalf("%s%s: %d targets: %+v", host, path, len(got), got) + } + return got[0] +} + +func TestResolveLongestPathAndDeployments(t *testing.T) { + s := fixture() + if g := one(t, s, "shop.example.com", "/api/orders?x=1"); g.Service != "api" || len(g.Deployments) != 1 || g.Deployments[0] != "api" { + t.Fatalf("api: %+v", g) + } + // the selector matches the same-namespace deployment only, not other/web-other + if g := one(t, s, "shop.example.com", "/index"); g.Service != "web" || len(g.Deployments) != 1 || g.Deployments[0] != "web-v2" { + t.Fatalf("web: %+v", g) + } + // /apix must not match the /api prefix (segment-wise) + if g := one(t, s, "shop.example.com", "/apix"); g.Service != "web" { + t.Fatalf("apix: %+v", g) + } +} + +func TestResolveHostsPortsAndExact(t *testing.T) { + s := fixture() + if g := one(t, s, "SHOP.example.com:443", "/"); g.Host != "shop.example.com" { + t.Fatalf("host case/port: %+v", g) + } + if g := one(t, s, "blog.example.com", "/x"); g.Host != "*.example.com" { + t.Fatalf("wildcard: %+v", g) + } + if got := s.Resolve("a.b.example.com", "/"); len(got) != 0 { + t.Fatalf("wildcard matched two labels: %+v", got) + } + if g := one(t, s, "shop.example.com", "/legacy"); g.Service != "old" { + t.Fatalf("exact: %+v", g) + } + if g := one(t, s, "shop.example.com", "/legacy/x"); g.Service != "web" { + t.Fatalf("exact matched a subpath: %+v", g) + } +} + +func TestMissingAndExternalBackends(t *testing.T) { + s := fixture() + if g := one(t, s, "shop.example.com", "/pay"); !g.Missing || g.Service != "gone" { + t.Fatalf("missing: %+v", g) + } + if g := one(t, s, "shop.example.com", "/cdn/a.js"); g.External != "cdn.vendor.net" || len(g.Deployments) != 0 { + t.Fatalf("external: %+v", g) + } +} diff --git a/internal/topology/watcher.go b/internal/topology/watcher.go new file mode 100644 index 0000000..86b55a2 --- /dev/null +++ b/internal/topology/watcher.go @@ -0,0 +1,150 @@ +package topology + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "os" + "strings" + "sync/atomic" + "time" + + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" + + kube "git.tornberg.me/go-otel/internal/kubernetes" +) + +// 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 +} + +// Resolve returns the backends serving host+path. +func (r *Registry) Resolve(host, path string) []Target { + if r == nil { + return nil + } + s := r.snap.Load() + if s == nil { + return nil + } + return s.Resolve(host, path) +} + +// Snapshot returns the current routing table. +func (r *Registry) Snapshot() Snapshot { + if r == nil { + return Snapshot{} + } + if s := r.snap.Load(); s != nil { + return *s + } + return Snapshot{} +} + +// NewFromEnv watches ingresses, services and deployments, following the same +// switches as the rollout watcher (OTEL_KUBERNETES_ENABLED / _NAMESPACE). It +// returns nil, nil when there is no cluster to watch. +func NewFromEnv(ctx context.Context) (*Registry, error) { + if strings.EqualFold(os.Getenv("OTEL_KUBERNETES_ENABLED"), "false") { + return nil, nil + } + cfg, err := kube.RESTConfig() + if err != nil { + if os.Getenv("OTEL_KUBERNETES_ENABLED") == "true" { + return nil, err + } + return nil, nil + } + client, err := kubernetes.NewForConfig(cfg) + if err != nil { + return nil, fmt.Errorf("create kubernetes client: %w", err) + } + var opts []informers.SharedInformerOption + if ns := os.Getenv("OTEL_KUBERNETES_NAMESPACE"); ns != "" { + opts = append(opts, informers.WithNamespace(ns)) + } + factory := informers.NewSharedInformerFactoryWithOptions(client, 30*time.Minute, opts...) + ing := factory.Networking().V1().Ingresses().Informer() + svc := factory.Core().V1().Services().Informer() + dep := factory.Apps().V1().Deployments().Informer() + + r := &Registry{} + mark := cache.ResourceEventHandlerFuncs{ + AddFunc: func(any) { r.dirty.Store(true) }, + UpdateFunc: func(_, _ any) { r.dirty.Store(true) }, + DeleteFunc: func(any) { r.dirty.Store(true) }, + } + for _, i := range []cache.SharedIndexInformer{ing, svc, dep} { + _, _ = i.AddEventHandler(mark) + } + rebuild := func() { + var is []networkingv1.Ingress + for _, o := range ing.GetStore().List() { + if v, ok := o.(*networkingv1.Ingress); ok { + is = append(is, *v) + } + } + var ss []corev1.Service + for _, o := range svc.GetStore().List() { + if v, ok := o.(*corev1.Service); ok { + ss = append(ss, *v) + } + } + var ds []appsv1.Deployment + for _, o := range dep.GetStore().List() { + if v, ok := o.(*appsv1.Deployment); ok { + ds = append(ds, *v) + } + } + s := Build(is, ss, ds) + r.snap.Store(&s) + r.built.Store(time.Now().Unix()) + } + go func() { + factory.Start(ctx.Done()) + if !cache.WaitForCacheSync(ctx.Done(), ing.HasSynced, svc.HasSynced, dep.HasSynced) { + log.Printf("topology: informers did not sync") + return + } + rebuild() + log.Printf("topology: %d ingress routes", len(r.Snapshot().Targets)) + // Coalesce bursts of changes (a rollout touches many objects). + t := time.NewTicker(5 * time.Second) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if r.dirty.Swap(false) { + rebuild() + } + } + } + }() + return r, nil +} + +// HandleTopology serves GET /api/topology. With ?host=&path= it answers +// "what sits behind this request"; without them it lists every route. +func (r *Registry) HandleTopology(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "no-store") + q := req.URL.Query() + if host := q.Get("host"); host != "" { + _ = json.NewEncoder(w).Encode(map[string]any{"targets": r.Resolve(host, q.Get("path"))}) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{"targets": r.Snapshot().Targets, "updated": r.built.Load()}) +} diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 6c14ff7..10d0796 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -241,7 +241,10 @@ rules: resources: ["deployments", "replicasets"] verbs: ["get", "list", "watch"] - apiGroups: [""] - resources: ["pods", "events"] + resources: ["pods", "events", "services"] + verbs: ["get", "list", "watch"] +- apiGroups: ["networking.k8s.io"] + resources: ["ingresses"] verbs: ["get", "list", "watch"] ---