From 0052822e49b74210881a9fc3c2a51138379ef128 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mats=20T=C3=B6rnberg?= Date: Tue, 22 Sep 2026 22:09:32 +0200 Subject: [PATCH] fix --- cmd/server/main.go | 10 +++++++ internal/alerts/notify.go | 18 ++++++++++++- internal/collector/collector.go | 17 ++++++++++++ internal/issues/tracker.go | 8 ++++++ internal/storage/clickhouse_traces.go | 38 +++++++++++++++++++++++++++ internal/storage/store.go | 1 + k8s/deployment.yaml | 3 +++ 7 files changed, 94 insertions(+), 1 deletion(-) diff --git a/cmd/server/main.go b/cmd/server/main.go index 8f0e80b..b24702c 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -244,6 +244,16 @@ func main() { historySize := 100 store := setupStore(context.Background()) coll := collector.New(nil, historySize, store) // Temporary nil for wsServer + if occurrenceStore, ok := store.(storage.OccurrenceStore); ok { + occurrences, err := occurrenceStore.RecentOccurrences(context.Background(), 100000) + if err != nil { + log.Printf("issues: persisted index restore skipped: %v", err) + } else if n, err := coll.RestoreIssues(context.Background(), occurrences); err != nil { + log.Printf("issues: persisted index restore failed after %d occurrences: %v", n, err) + } else if len(occurrences) > 0 { + log.Printf("issues: restored %d persisted occurrences", n) + } + } kubeWatcher, kubeErr := kube.NewFromEnv(context.Background()) if kubeErr != nil { log.Printf("kubernetes: rollout watcher disabled: %v", kubeErr) diff --git a/internal/alerts/notify.go b/internal/alerts/notify.go index 789d5ea..07215c2 100644 --- a/internal/alerts/notify.go +++ b/internal/alerts/notify.go @@ -312,7 +312,23 @@ func (a Alert) HTML(baseURL string, max int) string { 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)) + 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()) diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 2ae7346..484af74 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -6,6 +6,7 @@ import ( "container/ring" "context" "encoding/json" + "fmt" "log" "os" "strconv" @@ -230,6 +231,22 @@ func (c *Collector) SetIssueStatusStore(ctx context.Context, store issues.Status return c.issues.SetStatusStore(ctx, store) } +// RestoreIssues rebuilds the bounded in-memory issue index from persisted +// occurrences. This keeps /issues aligned with alerting after a restart. +func (c *Collector) RestoreIssues(ctx context.Context, occurrences []storage.Occurrence) (int, error) { + for _, occurrence := range occurrences { + var sample issues.Sample + if err := json.Unmarshal(occurrence.Sample, &sample); err != nil { + return 0, fmt.Errorf("decode persisted issue sample: %w", err) + } + if sample.Message == "" { + continue + } + c.issues.Restore(occurrence.Kind, occurrence.Service, sample.Message, sample) + } + return len(occurrences), nil +} + func (c *Collector) SimilarIssues(id string, limit int) ([]issues.SimilarIssue, bool) { return c.issues.Similar(id, limit) } diff --git a/internal/issues/tracker.go b/internal/issues/tracker.go index fc2d15a..2ab2eaf 100644 --- a/internal/issues/tracker.go +++ b/internal/issues/tracker.go @@ -227,6 +227,14 @@ func (t *Tracker) Record(kind, service, title string, sample Sample) { } } +// Restore records an occurrence loaded from the persistent occurrence store. +// It deliberately does not invoke the occurrence sink, avoiding duplicate +// writes while rebuilding the in-memory index during startup. +func (t *Tracker) Restore(kind, service, title string, sample Sample) { + normalizeSample(&sample, title) + _, _ = t.record(kind, service, title, sample) +} + func (t *Tracker) record(kind, service, title string, sample Sample) (string, OccurrenceSink) { id := Fingerprint(kind, service, NormalizeTitle(title)) bucket := HourBucket(sample.Timestamp) diff --git a/internal/storage/clickhouse_traces.go b/internal/storage/clickhouse_traces.go index dca9008..e14c8ce 100644 --- a/internal/storage/clickhouse_traces.go +++ b/internal/storage/clickhouse_traces.go @@ -213,6 +213,44 @@ func (s *ClickHouseStore) AppendOccurrence(ctx context.Context, occ Occurrence) return nil } +// RecentOccurrences returns persisted issue occurrences in chronological +// order so the in-memory issue tracker can be rebuilt after a restart. +func (s *ClickHouseStore) RecentOccurrences(ctx context.Context, limit int) ([]Occurrence, error) { + if limit <= 0 { + limit = 100000 + } + rows, err := s.conn.Query(ctx, fmt.Sprintf(` + SELECT issue_id, kind, service, toUnixTimestamp64Nano(timestamp), trace_id, span_id, sample + FROM %s.issue_occurrences + ORDER BY timestamp DESC + LIMIT ?`, s.db), limit) + if err != nil { + return nil, fmt.Errorf("clickhouse recent issue occurrences: %w", err) + } + defer rows.Close() + var out []Occurrence + for rows.Next() { + var o Occurrence + var ts int64 + var sample string + if err := rows.Scan(&o.IssueID, &o.Kind, &o.Service, &ts, &o.TraceID, &o.SpanID, &sample); err != nil { + return nil, fmt.Errorf("scan issue occurrence: %w", err) + } + o.Timestamp = ts + o.Sample = json.RawMessage(sample) + out = append(out, o) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate issue occurrences: %w", err) + } + // The query is newest-first; replay oldest-first to preserve issue titles + // and trend buckets as they were originally observed. + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} + const maxSamplesPerIssueTrace = 20 const maxUntraced = 20 diff --git a/internal/storage/store.go b/internal/storage/store.go index 8824982..3515f0e 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -134,6 +134,7 @@ type IssueTracesPage struct { // did this issue happen in" in one query (ClickHouseStore). type OccurrenceStore interface { AppendOccurrence(ctx context.Context, occ Occurrence) error + RecentOccurrences(ctx context.Context, limit int) ([]Occurrence, error) IssueTraces(ctx context.Context, issueID string, opts IssueTracesOptions) (IssueTracesPage, error) IssueSources(ctx context.Context, issueID string) ([]IssueSource, error) } diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 091e33c..d385b9c 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -264,6 +264,9 @@ rules: - apiGroups: [""] resources: ["pods", "events", "services"] verbs: ["get", "list", "watch"] +- apiGroups: [""] + resources: ["pods/log"] + verbs: ["get"] - apiGroups: ["networking.k8s.io"] resources: ["ingresses"] verbs: ["get", "list", "watch"]