diff --git a/README.md b/README.md index 98bdae3..db164bd 100644 --- a/README.md +++ b/README.md @@ -101,7 +101,13 @@ gRPC services for traces, metrics, and logs using standard OTLP protobuf definit filters as `search`, plus `timeout` (seconds, default 30). Rate-limited per client IP. - `GET /api/traces/{traceId}` - Retrieve all stored entries (spans and - correlated logs) referencing a given trace id. + correlated logs) referencing a given trace id. With ClickHouse this is an + indexed lookup (`entries.trace_ids`) and each payload is trimmed to that + trace's own spans/log records. +- `GET /api/traces/summaries?ids=a,b,c` - Per-trace rollups (name, start/end, + duration, span count, error count, services; max 100 ids) without reading + span payloads. Maintained on insert by the `trace_summaries` view; empty + with the file store. All endpoints above search the full persistent store, not just the in-memory ring buffer. @@ -133,6 +139,12 @@ Errors seen in traces (spans with an error status) and logs (severity Supports `limit`. - `GET /api/issues/{id}` - Fetch a single issue, including up to 5 recent sample occurrences (timestamp, trace/span id, message). +- `GET /api/issues/{id}/traces` - Every trace the issue occurred in, most + recently hit first, with each trace's summary and its occurrences, in one + request (`limit`, `before` = `nextBefore` from the previous page). + ClickHouse keeps *all* occurrences (`issue_occurrences`, 90-day TTL); + without it the response is built from the 5 in-memory samples and + `complete` is `false`. - `GET /api/patterns` - List deterministic groups of related exact issues. Pattern grouping is explainable and does not merge issue IDs. - `GET /api/issues/{id}/similar` - Return bounded, deterministic related-issue diff --git a/cmd/server/main.go b/cmd/server/main.go index 70560c6..228e470 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -303,6 +303,8 @@ func main() { uiMux.Handle("/api/history/logs", oidcAuth.RequireAuth(coll.HandleHistoryByType("logs"))) uiMux.Handle("/api/history/search", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSearch))) uiMux.Handle("/api/history/wait", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleWait))) + uiMux.Handle("GET /api/traces/summaries", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTraceSummaries))) + uiMux.Handle("GET /api/issues/{id}/traces", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueTraces))) uiMux.Handle("GET /api/traces/{traceId}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTraceByID))) uiMux.Handle("/api/issues", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssues))) uiMux.Handle("POST /api/ml/test-spike", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleMLDetectorTest))) diff --git a/internal/collector/collector.go b/internal/collector/collector.go index b81f8ab..789cb60 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -51,6 +51,13 @@ type Collector struct { sourceMaps *sourcemaps.Resolver kubeWatcher *kube.Watcher prometheus *prometheus.Client + + // Optional capabilities of the configured store (ClickHouse implements + // all of them); nil when unsupported, and the API falls back to the + // bounded in-memory data. + traceQuerier storage.TraceQuerier + occStore storage.OccurrenceStore + occChan chan storage.Occurrence } // New creates a new Collector instance. @@ -72,6 +79,7 @@ func New(broadcaster websocket.Broadcaster, historySize int, store storage.Store } if store != nil { + c.initOptionalStores(store) // Bounded worker pool prevents concurrent ClickHouse INSERTs from // multiplying query and merge memory under bursty ingest traffic. numWorkers := 4 @@ -162,6 +170,12 @@ func (c *Collector) GetIssue(id string) (issues.Issue, bool) { func (c *Collector) SetSourceMaps(resolver *sourcemaps.Resolver) { c.sourceMaps = resolver } +func resolveSamples(samples []issues.Sample, resolver *sourcemaps.Resolver) { + for i := range samples { + samples[i].ResolvedStacktrace = resolver.ResolveStack(samples[i].Release, samples[i].Stacktrace) + } +} + func resolveIssue(issue *issues.Issue, resolver *sourcemaps.Resolver) { for i := range issue.Samples { issue.Samples[i].ResolvedStacktrace = resolver.ResolveStack(issue.Samples[i].Release, issue.Samples[i].Stacktrace) diff --git a/internal/collector/handlers.go b/internal/collector/handlers.go index 9d28adc..7011678 100644 --- a/internal/collector/handlers.go +++ b/internal/collector/handlers.go @@ -360,7 +360,14 @@ func (c *Collector) HandleTraceByID(w http.ResponseWriter, r *http.Request) { return } - entries, err := c.SearchHistory(r.Context(), storage.SearchOptions{Query: traceID}, parseLimit(r, 1000)) + var entries []websocket.HistoryEntry + var err error + if c.traceQuerier != nil { + // Indexed lookup, payloads trimmed to this trace's spans/logs. + entries, err = c.traceQuerier.TraceEntries(r.Context(), traceID, parseLimit(r, 1000)) + } else { + entries, err = c.SearchHistory(r.Context(), storage.SearchOptions{Query: traceID}, parseLimit(r, 1000)) + } if err != nil { log.Printf("failed to fetch trace %s: %v", traceID, err) http.Error(w, "failed to fetch trace", http.StatusInternalServerError) diff --git a/internal/collector/issue_traces.go b/internal/collector/issue_traces.go new file mode 100644 index 0000000..428be86 --- /dev/null +++ b/internal/collector/issue_traces.go @@ -0,0 +1,242 @@ +package collector + +import ( + "context" + "encoding/json" + "log" + "net/http" + "sort" + "strconv" + "strings" + "sync/atomic" + "time" + + "git.tornberg.me/go-otel/internal/issues" + "git.tornberg.me/go-otel/internal/storage" +) + +// occurrenceBuffer bounds queued occurrence writes; beyond it, writes are +// dropped (and counted in the log) rather than slowing ingest. +const occurrenceBuffer = 4096 + +// initOptionalStores wires the store's optional capabilities: trace lookups +// and persistent issue occurrences. +func (c *Collector) initOptionalStores(store storage.Store) { + if tq, ok := store.(storage.TraceQuerier); ok { + c.traceQuerier = tq + } + occ, ok := store.(storage.OccurrenceStore) + if !ok { + return + } + c.occStore = occ + c.occChan = make(chan storage.Occurrence, occurrenceBuffer) + go func() { + for o := range c.occChan { + if err := occ.AppendOccurrence(context.Background(), o); err != nil { + log.Printf("failed to persist issue occurrence: %v", err) + } + } + }() + var dropped atomic.Int64 + c.issues.SetOccurrenceSink(func(issueID, kind, service string, sample issues.Sample) { + raw, err := json.Marshal(sample) + if err != nil { + return + } + select { + case c.occChan <- storage.Occurrence{ + IssueID: issueID, Kind: kind, Service: service, Timestamp: sample.Timestamp, + TraceID: sample.TraceID, SpanID: sample.SpanID, Sample: raw, + }: + default: + if n := dropped.Add(1); n%1000 == 1 { + log.Printf("issue occurrence buffer full; dropped %d so far", n) + } + } + }) +} + +// issueTraceJSON is one trace in the issue overview. +type issueTraceJSON struct { + TraceID string `json:"traceId"` + Hits int `json:"hits"` + FirstSeen int64 `json:"firstSeen"` + LastSeen int64 `json:"lastSeen"` + Summary *storage.TraceSummary `json:"summary,omitempty"` + Samples []issues.Sample `json:"samples"` +} + +type issueTracesResponse struct { + IssueID string `json:"issueId"` + TotalOccurrences int `json:"totalOccurrences"` + Traces []issueTraceJSON `json:"traces"` + Untraced []issues.Sample `json:"untraced"` + HasMore bool `json:"hasMore"` + // NextBefore is the LastSeen boundary to pass as ?before= for the next page. + NextBefore int64 `json:"nextBefore,omitempty"` + // Complete is false when only the in-memory samples were available. + Complete bool `json:"complete"` +} + +func decodeSamples(raw []json.RawMessage) []issues.Sample { + out := make([]issues.Sample, 0, len(raw)) + for _, r := range raw { + var s issues.Sample + if json.Unmarshal(r, &s) == nil { + out = append(out, s) + } + } + return out +} + +// HandleIssueTraces returns every trace an issue occurred in with per-trace +// summaries (duration, spans, services, errors) and the occurrences in each, +// in a single request. Spans themselves stay on GET /api/traces/{id}. +func (c *Collector) HandleIssueTraces(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + issue, known := c.GetIssue(id) + + var resp issueTracesResponse + resp.IssueID = id + if c.occStore != nil { + opts := storage.IssueTracesOptions{Limit: parseLimit(r, 50)} + if b, err := strconv.ParseInt(r.URL.Query().Get("before"), 10, 64); err == nil { + opts.Before = b + } + page, err := c.occStore.IssueTraces(r.Context(), id, opts) + if err != nil { + log.Printf("failed to load issue traces %s: %v", id, err) + http.Error(w, "failed to load issue traces", http.StatusInternalServerError) + return + } + if page.TotalOccurrences == 0 && !known { + http.Error(w, "issue not found", http.StatusNotFound) + return + } + resp.TotalOccurrences, resp.HasMore, resp.Complete = page.TotalOccurrences, page.HasMore, true + resp.Untraced = decodeSamples(page.Untraced) + for _, t := range page.Traces { + resp.Traces = append(resp.Traces, issueTraceJSON{ + TraceID: t.TraceID, Hits: t.Hits, FirstSeen: t.FirstSeen, LastSeen: t.LastSeen, + Summary: t.Summary, Samples: decodeSamples(t.Samples), + }) + } + if resp.HasMore && len(resp.Traces) > 0 { + resp.NextBefore = resp.Traces[len(resp.Traces)-1].LastSeen + } + } else { + if !known { + http.Error(w, "issue not found", http.StatusNotFound) + return + } + resp.TotalOccurrences = issue.Count + resp.Traces, resp.Untraced = groupInMemory(issue.Samples) + } + + // Source-map resolution applies to stored samples exactly as it does for + // the in-memory ones. + if c.sourceMaps != nil { + for i := range resp.Traces { + resolveSamples(resp.Traces[i].Samples, c.sourceMaps) + } + resolveSamples(resp.Untraced, c.sourceMaps) + } + if resp.Traces == nil { + resp.Traces = []issueTraceJSON{} + } + if resp.Untraced == nil { + resp.Untraced = []issues.Sample{} + } + + // Summaries for traces the store didn't summarise (fallback path, or + // traces older than the summary table): fill from the store when possible. + if c.traceQuerier != nil { + var missing []string + for _, t := range resp.Traces { + if t.Summary == nil { + missing = append(missing, t.TraceID) + } + } + if len(missing) > 0 { + if sums, err := c.traceQuerier.TraceSummaries(r.Context(), missing); err == nil { + for i := range resp.Traces { + if s, ok := sums[resp.Traces[i].TraceID]; ok { + resp.Traces[i].Summary = &s + } + } + } + } + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + log.Printf("failed to encode issue traces: %v", err) + } +} + +// groupInMemory builds the same shape from the tracker's bounded samples, +// newest trace first. +func groupInMemory(samples []issues.Sample) ([]issueTraceJSON, []issues.Sample) { + byTrace := map[string]*issueTraceJSON{} + var traces []*issueTraceJSON + var untraced []issues.Sample + for _, s := range samples { + if s.TraceID == "" { + untraced = append(untraced, s) + continue + } + t, ok := byTrace[s.TraceID] + if !ok { + t = &issueTraceJSON{TraceID: s.TraceID, FirstSeen: s.Timestamp, LastSeen: s.Timestamp} + byTrace[s.TraceID] = t + traces = append(traces, t) + } + t.Hits++ + t.Samples = append(t.Samples, s) + if s.Timestamp < t.FirstSeen { + t.FirstSeen = s.Timestamp + } + if s.Timestamp > t.LastSeen { + t.LastSeen = s.Timestamp + } + } + sort.Slice(traces, func(i, j int) bool { return traces[i].LastSeen > traces[j].LastSeen }) + out := make([]issueTraceJSON, len(traces)) + for i, t := range traces { + // Newest first, matching the store-backed path. + sort.SliceStable(t.Samples, func(a, b int) bool { return t.Samples[a].Timestamp > t.Samples[b].Timestamp }) + out[i] = *t + } + sort.Slice(untraced, func(i, j int) bool { return untraced[i].Timestamp > untraced[j].Timestamp }) + return out, untraced +} + +// HandleTraceSummaries returns per-trace summaries for ?ids=a,b,c (max 100). +// Unknown ids are omitted; stores without summaries answer an empty object. +func (c *Collector) HandleTraceSummaries(w http.ResponseWriter, r *http.Request) { + var ids []string + for _, id := range strings.Split(r.URL.Query().Get("ids"), ",") { + if id = strings.TrimSpace(id); id != "" { + ids = append(ids, id) + } + } + if len(ids) > 100 { + http.Error(w, "at most 100 ids", http.StatusBadRequest) + return + } + out := map[string]storage.TraceSummary{} + if c.traceQuerier != nil && len(ids) > 0 { + ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) + defer cancel() + sums, err := c.traceQuerier.TraceSummaries(ctx, ids) + if err != nil { + log.Printf("failed to load trace summaries: %v", err) + http.Error(w, "failed to load trace summaries", http.StatusInternalServerError) + return + } + out = sums + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(out) +} diff --git a/internal/collector/issue_traces_test.go b/internal/collector/issue_traces_test.go new file mode 100644 index 0000000..da5de8a --- /dev/null +++ b/internal/collector/issue_traces_test.go @@ -0,0 +1,133 @@ +package collector + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "git.tornberg.me/go-otel/internal/issues" + "git.tornberg.me/go-otel/internal/storage" + "git.tornberg.me/go-otel/internal/websocket" +) + +// occStore is a mockStore that also implements OccurrenceStore and +// TraceQuerier, standing in for ClickHouse. +type occStore struct { + mockStore + omu sync.Mutex + occs []storage.Occurrence + page storage.IssueTracesPage +} + +func (o *occStore) AppendOccurrence(_ context.Context, occ storage.Occurrence) error { + o.omu.Lock() + defer o.omu.Unlock() + o.occs = append(o.occs, occ) + return nil +} + +func (o *occStore) IssueTraces(context.Context, string, storage.IssueTracesOptions) (storage.IssueTracesPage, error) { + return o.page, nil +} + +func (o *occStore) TraceEntries(context.Context, string, int) ([]websocket.HistoryEntry, error) { + return nil, nil +} + +func (o *occStore) TraceSummaries(_ context.Context, ids []string) (map[string]storage.TraceSummary, error) { + out := map[string]storage.TraceSummary{} + for _, id := range ids { + out[id] = storage.TraceSummary{ID: id, Name: "sum-" + id, Duration: 42, SpanCount: 3} + } + return out, nil +} + +func getIssueTraces(t *testing.T, c *Collector, id string) (int, issueTracesResponse) { + t.Helper() + req := httptest.NewRequest(http.MethodGet, "/api/issues/"+id+"/traces", nil) + req.SetPathValue("id", id) + rec := httptest.NewRecorder() + c.HandleIssueTraces(rec, req) + var resp issueTracesResponse + _ = json.Unmarshal(rec.Body.Bytes(), &resp) + return rec.Code, resp +} + +func TestOccurrencesPersistedBeyondInMemorySamples(t *testing.T) { + store := &occStore{} + c := New(nil, 10, store) + for i := 0; i < 12; i++ { + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: int64(1000 + i), TraceID: "t1", Message: "boom"}) + } + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + store.omu.Lock() + n := len(store.occs) + store.omu.Unlock() + if n == 12 { + break + } + time.Sleep(10 * time.Millisecond) + } + store.omu.Lock() + defer store.omu.Unlock() + if len(store.occs) != 12 { + t.Fatalf("persisted %d occurrences, want all 12 (memory keeps only 5)", len(store.occs)) + } + if store.occs[0].IssueID == "" || store.occs[0].TraceID != "t1" { + t.Errorf("occurrence = %+v", store.occs[0]) + } +} + +func TestHandleIssueTracesFromStoreFillsSummaries(t *testing.T) { + sample, _ := json.Marshal(issues.Sample{Timestamp: 5, TraceID: "t9", Message: "x"}) + store := &occStore{page: storage.IssueTracesPage{ + TotalOccurrences: 7, + HasMore: true, + Traces: []storage.IssueTrace{{TraceID: "t9", Hits: 7, LastSeen: 5, Samples: []json.RawMessage{sample}}}, + }} + c := New(nil, 10, store) + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: 1, TraceID: "t9", Message: "x"}) + id := c.issues.List(10)[0].ID + + code, resp := getIssueTraces(t, c, id) + if code != http.StatusOK || !resp.Complete || resp.TotalOccurrences != 7 || !resp.HasMore || resp.NextBefore != 5 { + t.Fatalf("code=%d resp=%+v", code, resp) + } + if len(resp.Traces) != 1 || resp.Traces[0].Summary == nil || resp.Traces[0].Summary.Name != "sum-t9" { + t.Errorf("summary not filled from TraceQuerier: %+v", resp.Traces) + } + if len(resp.Traces[0].Samples) != 1 || resp.Traces[0].Samples[0].Message != "x" { + t.Errorf("samples = %+v", resp.Traces[0].Samples) + } +} + +func TestHandleIssueTracesInMemoryFallback(t *testing.T) { + c := New(nil, 10, nil) + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: 1, TraceID: "a", Message: "m"}) + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: 3, TraceID: "b", Message: "m"}) + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: 2, TraceID: "a", Message: "m"}) + c.issues.Record("span_error", "api", "boom", issues.Sample{Timestamp: 4, Message: "no trace"}) + id := c.issues.List(10)[0].ID + + code, resp := getIssueTraces(t, c, id) + if code != http.StatusOK || resp.Complete { + t.Fatalf("code=%d complete=%v (fallback must say it is incomplete)", code, resp.Complete) + } + if len(resp.Traces) != 2 || resp.Traces[0].TraceID != "b" || resp.Traces[1].TraceID != "a" || resp.Traces[1].Hits != 2 { + t.Errorf("traces = %+v", resp.Traces) + } + if len(resp.Untraced) != 1 { + t.Errorf("untraced = %d, want 1", len(resp.Untraced)) + } + if s := resp.Traces[1].Samples; len(s) != 2 || s[0].Timestamp != 2 || s[1].Timestamp != 1 { + t.Errorf("samples must be newest first: %+v", s) + } + if code, _ := getIssueTraces(t, c, "missing"); code != http.StatusNotFound { + t.Errorf("unknown issue: code=%d, want 404", code) + } +} diff --git a/internal/issues/tracker.go b/internal/issues/tracker.go index df4beac..0500647 100644 --- a/internal/issues/tracker.go +++ b/internal/issues/tracker.go @@ -117,6 +117,19 @@ type Tracker struct { mu sync.Mutex issues map[string]*Issue feedback map[string]string + sink OccurrenceSink +} + +// OccurrenceSink receives every recorded occurrence (already normalized) so +// it can be persisted beyond the bounded in-memory sample window. It is +// called outside the tracker lock and must not block. +type OccurrenceSink func(issueID, kind, service string, sample Sample) + +// SetOccurrenceSink registers the sink; nil disables it. +func (t *Tracker) SetOccurrenceSink(fn OccurrenceSink) { + t.mu.Lock() + t.sink = fn + t.mu.Unlock() } // ListOptions filters and paginates the issue list. Issues remain ordered by @@ -153,6 +166,13 @@ func NewTracker() *Tracker { // fingerprint. func (t *Tracker) Record(kind, service, title string, sample Sample) { normalizeSample(&sample, title) + id, sink := t.record(kind, service, title, sample) + if sink != nil { + sink(id, kind, service, sample) + } +} + +func (t *Tracker) record(kind, service, title string, sample Sample) (string, OccurrenceSink) { id := Fingerprint(kind, service, NormalizeTitle(title)) bucket := HourBucket(sample.Timestamp) @@ -196,6 +216,7 @@ func (t *Tracker) Record(kind, service, title string, sample Sample) { if len(issue.Samples) > maxSamples { issue.Samples = issue.Samples[len(issue.Samples)-maxSamples:] } + return id, t.sink } // Similar returns bounded, deterministic near-match suggestions for an issue. diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go index 7a7bdf9..2858798 100644 --- a/internal/storage/clickhouse.go +++ b/internal/storage/clickhouse.go @@ -198,7 +198,7 @@ func (s *ClickHouseStore) initSchema(ctx context.Context) error { // empty until this succeeds. fmt.Printf("clickhouse: failed to create rollup view (stats endpoints stay empty): %v\n", err) } - return nil + return s.initTraceSchema(ctx) } // chPayload extracts rollup columns from the marshaled OTLP JSON. Field @@ -326,17 +326,41 @@ func extractHead(data []byte) (service string, isErr bool, release, traceID stri func (s *ClickHouseStore) Append(ctx context.Context, entry websocket.HistoryEntry) error { service, isErr, release, traceID := extractHead(entry.Data) + // A batch routinely holds several traces, so keep every trace id (the + // scalar trace_id column only records the first) for indexed lookups. + var facts []spanFact + var traceIDs []string + switch entry.Type { + case "traces": + facts = extractSpanFacts(entry.Data) + seen := make(map[string]struct{}, len(facts)) + for _, f := range facts { + if _, ok := seen[f.TraceID]; !ok { + seen[f.TraceID] = struct{}{} + traceIDs = append(traceIDs, f.TraceID) + } + } + case "logs": + traceIDs = allTraceIDs(entry.Data) + } + if traceIDs == nil { + traceIDs = []string{} + } + ts := time.Unix(0, entry.Timestamp).UTC() err := s.conn.AsyncInsert(ctx, fmt.Sprintf( `INSERT INTO %s.entries - (timestamp, type, service, is_error, release, trace_id, data) - VALUES (?, ?, ?, ?, ?, ?, ?)`, s.db), + (timestamp, type, service, is_error, release, trace_id, trace_ids, data) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, s.db), false, - ts, entry.Type, service, boolToUint8(isErr), release, traceID, string(entry.Data), + ts, entry.Type, service, boolToUint8(isErr), release, traceID, traceIDs, string(entry.Data), ) if err != nil { return fmt.Errorf("clickhouse append: %w", err) } + if err := s.appendSpanFacts(ctx, ts, facts); err != nil { + return fmt.Errorf("clickhouse span facts: %w", err) + } return nil } diff --git a/internal/storage/clickhouse_test.go b/internal/storage/clickhouse_test.go index 3e1958f..1a25e0c 100644 --- a/internal/storage/clickhouse_test.go +++ b/internal/storage/clickhouse_test.go @@ -38,7 +38,17 @@ func TestClickHouseStoreIntegration(t *testing.T) { now := time.Now().UTC() mk := func(minsAgo int, typ, service string, isErr bool, release string) websocket.HistoryEntry { - raw, _ := json.Marshal(map[string]interface{}{"n": minsAgo}) + sev := 9 + if isErr { + sev = 17 + } + str := func(k, v string) map[string]any { + return map[string]any{"key": k, "value": map[string]any{"stringValue": v}} + } + raw, _ := json.Marshal(map[string]any{"resourceLogs": []any{map[string]any{ + "resource": map[string]any{"attributes": []any{str("service.name", service), str("service.release", release)}}, + "scopeLogs": []any{map[string]any{"logRecords": []any{map[string]any{"severityNumber": sev, "n": minsAgo}}}}, + }}}) return websocket.HistoryEntry{ Type: typ, Timestamp: now.Add(-time.Duration(minsAgo) * time.Minute).UnixNano(), @@ -58,7 +68,7 @@ func TestClickHouseStoreIntegration(t *testing.T) { } } // Flush async insert buffer before reading. - if err := s.conn.AsyncInsert(ctx, "INSERT INTO FUNCTION null('x UInt8') VALUES (0)", true); err != nil { + if err := s.conn.Exec(ctx, "SYSTEM FLUSH ASYNC INSERT QUEUE"); err != nil { t.Fatalf("flush: %v", err) } diff --git a/internal/storage/clickhouse_traces.go b/internal/storage/clickhouse_traces.go new file mode 100644 index 0000000..9939bdd --- /dev/null +++ b/internal/storage/clickhouse_traces.go @@ -0,0 +1,336 @@ +package storage + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "time" + + "git.tornberg.me/go-otel/internal/websocket" +) + +// initTraceSchema adds everything the trace and issue overviews need on top of +// the raw entries table: +// +// - entries.trace_ids + bloom indexes: indexed "which payloads contain trace X" +// instead of a substring scan over every payload. +// - span_facts (Null engine) -> trace_summaries (AggregatingMergeTree): one +// row per trace maintained on insert, so overview tables never read spans. +// - issue_occurrences: every occurrence of every issue, beyond the 5 samples +// the tracker keeps in memory. +// +// Failures are logged, not fatal: raw storage keeps working and the API falls +// back to the slower paths. +func (s *ClickHouseStore) initTraceSchema(ctx context.Context) error { + ddl := []string{ + fmt.Sprintf(`ALTER TABLE %s.entries ADD COLUMN IF NOT EXISTS trace_ids Array(String) DEFAULT []`, s.db), + + fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.span_facts ( + at DateTime('UTC'), + trace_id String, + span_id String, + parent_span_id String, + service LowCardinality(String), + name String, + t0_ns UInt64, + t1_ns UInt64, + is_error UInt8 + ) ENGINE = Null`, s.db), + + fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.trace_summaries ( + trace_id String, + ts SimpleAggregateFunction(min, DateTime('UTC')), + start_ns SimpleAggregateFunction(min, UInt64), + end_ns SimpleAggregateFunction(max, UInt64), + spans SimpleAggregateFunction(sum, UInt64), + errors SimpleAggregateFunction(sum, UInt64), + services SimpleAggregateFunction(groupUniqArrayArray, Array(String)), + root AggregateFunction(argMin, String, Tuple(UInt8, UInt64)) + ) ENGINE = AggregatingMergeTree + ORDER BY trace_id + TTL ts + INTERVAL 90 DAY`, s.db), + + fmt.Sprintf(`CREATE MATERIALIZED VIEW IF NOT EXISTS %s.trace_summaries_mv TO %s.trace_summaries AS + SELECT + trace_id, + min(at) AS ts, + min(t0_ns) AS start_ns, + max(t1_ns) AS end_ns, + count() AS spans, + sum(is_error) AS errors, + groupUniqArray(service) AS services, + argMinState(name, (parent_span_id != '', t0_ns)) AS root + FROM %s.span_facts + GROUP BY trace_id`, s.db, s.db, s.db), + + fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.issue_occurrences ( + issue_id String, + timestamp DateTime64(9, 'UTC'), + kind LowCardinality(String), + service LowCardinality(String), + trace_id String, + span_id String, + sample String CODEC(ZSTD(3)) + ) ENGINE = MergeTree + PARTITION BY toYYYYMM(timestamp) + ORDER BY (issue_id, timestamp) + TTL toDateTime(timestamp) + INTERVAL 90 DAY`, s.db), + } + for _, q := range ddl { + if err := s.conn.Exec(ctx, q); err != nil { + fmt.Printf("clickhouse: trace schema step failed (falling back to slow paths): %v\n", err) + return nil + } + } + + for _, col := range []string{"trace_id", "trace_ids"} { + s.ensureBloomIndex(ctx, "entries", "idx_"+col, col) + } + return nil +} + +// ensureBloomIndex adds a bloom-filter skip index and, only when it is newly +// added, materializes it over existing parts (a background mutation). +func (s *ClickHouseStore) ensureBloomIndex(ctx context.Context, table, name, expr string) { + var n uint64 + row := s.conn.QueryRow(ctx, + `SELECT count() FROM system.data_skipping_indices WHERE database = ? AND table = ? AND name = ?`, + s.db, table, name) + if err := row.Scan(&n); err != nil || n > 0 { + return + } + if err := s.conn.Exec(ctx, fmt.Sprintf( + `ALTER TABLE %s.%s ADD INDEX IF NOT EXISTS %s %s TYPE bloom_filter(0.01) GRANULARITY 1`, + s.db, table, name, expr)); err != nil { + fmt.Printf("clickhouse: add index %s: %v\n", name, err) + return + } + if err := s.conn.Exec(ctx, fmt.Sprintf(`ALTER TABLE %s.%s MATERIALIZE INDEX %s`, s.db, table, name)); err != nil { + fmt.Printf("clickhouse: materialize index %s: %v\n", name, err) + } +} + +func (s *ClickHouseStore) appendSpanFacts(ctx context.Context, ts time.Time, facts []spanFact) error { + if len(facts) == 0 { + return nil + } + batch, err := s.conn.PrepareBatch(ctx, fmt.Sprintf( + `INSERT INTO %s.span_facts (at, trace_id, span_id, parent_span_id, service, name, t0_ns, t1_ns, is_error)`, s.db)) + if err != nil { + return err + } + for _, f := range facts { + if err := batch.Append(ts, f.TraceID, f.SpanID, f.ParentSpanID, f.Service, f.Name, f.StartNs, f.EndNs, boolToUint8(f.IsError)); err != nil { + _ = batch.Abort() + return err + } + } + return batch.Send() +} + +// TraceEntries implements TraceQuerier. +func (s *ClickHouseStore) TraceEntries(ctx context.Context, traceID string, limit int) ([]websocket.HistoryEntry, error) { + if limit <= 0 { + limit = 1000 + } + rows, err := s.conn.Query(ctx, fmt.Sprintf( + `SELECT timestamp, type, data FROM %s.entries + WHERE has(trace_ids, ?) OR trace_id = ? + ORDER BY timestamp DESC LIMIT ?`, s.db), traceID, traceID, limit) + if err != nil { + return nil, fmt.Errorf("clickhouse trace entries: %w", err) + } + defer rows.Close() + + var out []websocket.HistoryEntry + for rows.Next() { + var ( + ts time.Time + typ string + raw string + ) + if err := rows.Scan(&ts, &typ, &raw); err != nil { + return nil, err + } + trimmed, ok := TrimToTrace([]byte(raw), traceID) + if !ok { + continue + } + out = append(out, websocket.HistoryEntry{Type: typ, Timestamp: ts.UnixNano(), Data: json.RawMessage(trimmed)}) + } + return out, rows.Err() +} + +// TraceSummaries implements TraceQuerier. +func (s *ClickHouseStore) TraceSummaries(ctx context.Context, ids []string) (map[string]TraceSummary, error) { + out := make(map[string]TraceSummary, len(ids)) + if len(ids) == 0 { + return out, nil + } + rows, err := s.conn.Query(ctx, fmt.Sprintf( + `SELECT trace_id, min(start_ns), max(end_ns), sum(spans), sum(errors), + groupUniqArrayArray(services), argMinMerge(root) + FROM %s.trace_summaries + WHERE trace_id IN ? + GROUP BY trace_id`, s.db), ids) + if err != nil { + return nil, fmt.Errorf("clickhouse trace summaries: %w", err) + } + defer rows.Close() + for rows.Next() { + var ( + id, name string + start, end, spans, e uint64 + services []string + ) + if err := rows.Scan(&id, &start, &end, &spans, &e, &services, &name); err != nil { + return nil, err + } + sort.Strings(services) + if name == "" { + name = "Trace " + id[:min(8, len(id))] + } + out[id] = TraceSummary{ + ID: id, Name: name, StartTime: int64(start), EndTime: int64(end), Duration: int64(end - start), + SpanCount: int(spans), ErrorCount: int(e), Services: services, + } + } + return out, rows.Err() +} + +// AppendOccurrence implements OccurrenceStore. +func (s *ClickHouseStore) AppendOccurrence(ctx context.Context, occ Occurrence) error { + err := s.conn.AsyncInsert(ctx, fmt.Sprintf( + `INSERT INTO %s.issue_occurrences (issue_id, timestamp, kind, service, trace_id, span_id, sample) + VALUES (?, ?, ?, ?, ?, ?, ?)`, s.db), + false, + occ.IssueID, time.Unix(0, occ.Timestamp).UTC(), occ.Kind, occ.Service, occ.TraceID, occ.SpanID, string(occ.Sample), + ) + if err != nil { + return fmt.Errorf("clickhouse append occurrence: %w", err) + } + return nil +} + +const maxSamplesPerIssueTrace = 20 +const maxUntraced = 20 + +// IssueTraces implements OccurrenceStore: one aggregate query for the page of +// traces, one for their samples, one batched lookup for their summaries. +func (s *ClickHouseStore) IssueTraces(ctx context.Context, issueID string, opts IssueTracesOptions) (IssueTracesPage, error) { + limit := opts.Limit + if limit <= 0 || limit > 200 { + limit = 50 + } + var page IssueTracesPage + + var total uint64 + if err := s.conn.QueryRow(ctx, fmt.Sprintf( + `SELECT count() FROM %s.issue_occurrences WHERE issue_id = ?`, s.db), issueID).Scan(&total); err != nil { + return page, fmt.Errorf("clickhouse issue totals: %w", err) + } + page.TotalOccurrences = int(total) + + having := "" + args := []any{issueID} + if opts.Before > 0 { + having = "HAVING max(timestamp) < ?" + args = append(args, time.Unix(0, opts.Before).UTC()) + } + args = append(args, limit+1) + rows, err := s.conn.Query(ctx, fmt.Sprintf( + `SELECT trace_id, count(), min(timestamp), max(timestamp) + FROM %s.issue_occurrences + WHERE issue_id = ? AND trace_id != '' + GROUP BY trace_id %s + ORDER BY max(timestamp) DESC + LIMIT ?`, s.db, having), args...) + if err != nil { + return page, fmt.Errorf("clickhouse issue traces: %w", err) + } + for rows.Next() { + var ( + t IssueTrace + hits uint64 + from, to time.Time + ) + if err := rows.Scan(&t.TraceID, &hits, &from, &to); err != nil { + _ = rows.Close() + return page, err + } + t.Hits, t.FirstSeen, t.LastSeen = int(hits), from.UnixNano(), to.UnixNano() + page.Traces = append(page.Traces, t) + } + _ = rows.Close() + if err := rows.Err(); err != nil { + return page, err + } + if len(page.Traces) > limit { + page.Traces = page.Traces[:limit] + page.HasMore = true + } + + if len(page.Traces) > 0 { + ids := make([]string, len(page.Traces)) + index := make(map[string]int, len(ids)) + for i, t := range page.Traces { + ids[i] = t.TraceID + index[t.TraceID] = i + } + srows, err := s.conn.Query(ctx, fmt.Sprintf( + `SELECT trace_id, sample FROM %s.issue_occurrences + WHERE issue_id = ? AND trace_id IN ? + ORDER BY timestamp DESC + LIMIT ? BY trace_id`, s.db), issueID, ids, maxSamplesPerIssueTrace) + if err != nil { + return page, fmt.Errorf("clickhouse issue trace samples: %w", err) + } + for srows.Next() { + var id, sample string + if err := srows.Scan(&id, &sample); err != nil { + _ = srows.Close() + return page, err + } + i := index[id] + page.Traces[i].Samples = append(page.Traces[i].Samples, json.RawMessage(sample)) + } + _ = srows.Close() + if err := srows.Err(); err != nil { + return page, err + } + + summaries, err := s.TraceSummaries(ctx, ids) + if err != nil { + return page, err + } + for i := range page.Traces { + if sum, ok := summaries[page.Traces[i].TraceID]; ok { + page.Traces[i].Summary = &sum + } + } + } + + if opts.Before == 0 { + urows, err := s.conn.Query(ctx, fmt.Sprintf( + `SELECT sample FROM %s.issue_occurrences + WHERE issue_id = ? AND trace_id = '' + ORDER BY timestamp DESC LIMIT ?`, s.db), issueID, maxUntraced) + if err != nil { + return page, fmt.Errorf("clickhouse issue untraced: %w", err) + } + for urows.Next() { + var sample string + if err := urows.Scan(&sample); err != nil { + _ = urows.Close() + return page, err + } + page.Untraced = append(page.Untraced, json.RawMessage(sample)) + } + _ = urows.Close() + if err := urows.Err(); err != nil { + return page, err + } + } + return page, nil +} diff --git a/internal/storage/clickhouse_traces_test.go b/internal/storage/clickhouse_traces_test.go new file mode 100644 index 0000000..12dfb01 --- /dev/null +++ b/internal/storage/clickhouse_traces_test.go @@ -0,0 +1,113 @@ +package storage + +import ( + "context" + "encoding/json" + "fmt" + "os" + "testing" + "time" + + "git.tornberg.me/go-otel/internal/websocket" +) + +// TestClickHouseTraceOverviewIntegration exercises the trace/issue SQL against +// a real ClickHouse: indexed trace lookup with trimming, summary rollups +// across multiple batches, and issue occurrence paging. Skipped unless +// OTEL_CLICKHOUSE_TEST_ADDR is set (see TestClickHouseStoreIntegration). +func TestClickHouseTraceOverviewIntegration(t *testing.T) { + addr := os.Getenv("OTEL_CLICKHOUSE_TEST_ADDR") + if addr == "" { + t.Skip("set OTEL_CLICKHOUSE_TEST_ADDR=host:9000 (with a disposable DB) to run") + } + ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second) + defer cancel() + db := "otel_trace_test_" + time.Now().UTC().Format("150405") + s, err := NewClickHouseStore(ctx, ClickHouseConfig{Addr: addr, Database: db, + Username: os.Getenv("OTEL_CLICKHOUSE_TEST_USER"), Password: os.Getenv("OTEL_CLICKHOUSE_TEST_PASSWORD")}) + if err != nil { + t.Skipf("ClickHouse not reachable at %s: %v", addr, err) + } + defer func() { + _ = s.conn.Exec(ctx, "DROP DATABASE IF EXISTS "+db) + _ = s.Close() + }() + + base := time.Now().UTC().Add(-time.Hour).UnixNano() + // Batch 1 holds spans of two traces; trace "aaa" continues in batch 2. + b1 := json.RawMessage(mixedPayload) + b2 := json.RawMessage(`{"resourceSpans":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"db"}}]}, + "scopeSpans":[{"spans":[{"traceId":"aaa","spanId":"s9","parentSpanId":"s3","name":"pg","startTimeUnixNano":"150","endTimeUnixNano":"900","status":{"code":2}}]}]}]}`) + for i, d := range []json.RawMessage{b1, b2} { + if err := s.Append(ctx, websocket.HistoryEntry{Type: "traces", Timestamp: base + int64(i), Data: d}); err != nil { + t.Fatalf("Append: %v", err) + } + } + // AsyncInsert buffers; wait for the flush like the stats test does. + time.Sleep(2 * time.Second) + _ = s.conn.Exec(ctx, "SYSTEM FLUSH ASYNC INSERT QUEUE") + + // Trace "bbb" is NOT the first trace in batch 1, so only trace_ids finds it. + got, err := s.TraceEntries(ctx, "bbb", 100) + if err != nil || len(got) != 1 { + t.Fatalf("TraceEntries(bbb) = %d entries, err %v; want 1", len(got), err) + } + if _, ok := TrimToTrace(got[0].Data, "aaa"); ok { + t.Error("bbb entry must not contain aaa spans after trimming") + } + got, err = s.TraceEntries(ctx, "aaa", 100) + if err != nil || len(got) != 2 { + t.Fatalf("TraceEntries(aaa) = %d entries, err %v; want 2 (both batches)", len(got), err) + } + + sums, err := s.TraceSummaries(ctx, []string{"aaa", "bbb", "nope"}) + if err != nil { + t.Fatalf("TraceSummaries: %v", err) + } + if _, ok := sums["nope"]; ok { + t.Error("unknown trace must be absent") + } + a := sums["aaa"] + // aaa spans: s1 100-300 (root, "GET /"), s3 120-200 (503 => error), s9 150-900 (error). + if a.SpanCount != 3 || a.ErrorCount != 2 || a.Duration != 800 || a.Name != "GET /" { + t.Errorf("aaa summary = %+v; want 3 spans, 2 errors, 800ns, name 'GET /'", a) + } + if len(a.Services) != 2 || a.Services[0] != "api" || a.Services[1] != "db" { + t.Errorf("aaa services = %v; want [api db]", a.Services) + } + + // Issue occurrences: 3 in aaa, 1 in bbb, 1 without a trace. + occ := func(i int, trace string) Occurrence { + raw, _ := json.Marshal(map[string]any{"timestamp": base + int64(i), "traceId": trace, "message": fmt.Sprintf("boom %d", i)}) + return Occurrence{IssueID: "issue1", Kind: "span_error", Service: "api", Timestamp: base + int64(i)*int64(time.Second), TraceID: trace, Sample: raw} + } + for i, tr := range []string{"aaa", "aaa", "bbb", "", "aaa"} { + if err := s.AppendOccurrence(ctx, occ(i, tr)); err != nil { + t.Fatalf("AppendOccurrence: %v", err) + } + } + _ = s.conn.Exec(ctx, "SYSTEM FLUSH ASYNC INSERT QUEUE") + time.Sleep(time.Second) + + page, err := s.IssueTraces(ctx, "issue1", IssueTracesOptions{Limit: 1}) + if err != nil { + t.Fatalf("IssueTraces: %v", err) + } + if page.TotalOccurrences != 5 || len(page.Traces) != 1 || !page.HasMore { + t.Fatalf("page1 total=%d traces=%d more=%v", page.TotalOccurrences, len(page.Traces), page.HasMore) + } + first := page.Traces[0] + if first.TraceID != "aaa" || first.Hits != 3 || len(first.Samples) != 3 || first.Summary == nil || first.Summary.SpanCount != 3 { + t.Errorf("first trace = %+v", first) + } + if len(page.Untraced) != 1 { + t.Errorf("untraced = %d, want 1", len(page.Untraced)) + } + next, err := s.IssueTraces(ctx, "issue1", IssueTracesOptions{Limit: 1, Before: first.LastSeen}) + if err != nil || len(next.Traces) != 1 || next.Traces[0].TraceID != "bbb" || next.HasMore { + t.Fatalf("page2 = %+v err %v", next, err) + } + if len(next.Untraced) != 0 { + t.Error("untraced only belongs on the first page") + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go index 9a7766b..7826dfe 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -2,6 +2,7 @@ package storage import ( "context" + "encoding/json" "git.tornberg.me/go-otel/internal/websocket" ) @@ -38,3 +39,73 @@ type StatsQuerier interface { // Services lists distinct services seen in the rollup. Services(ctx context.Context) ([]string, error) } + +// TraceSummary is the per-trace rollup shown in overview tables: everything +// needed to render a row without reading any span payloads. Times are unix +// nanoseconds, matching the rest of the API. +type TraceSummary struct { + ID string `json:"id"` + Name string `json:"name"` + StartTime int64 `json:"startTime"` + EndTime int64 `json:"endTime"` + Duration int64 `json:"duration"` + SpanCount int `json:"spanCount"` + ErrorCount int `json:"errorCount"` + Services []string `json:"services"` +} + +// TraceQuerier is implemented by stores that can look a trace up by id +// without scanning payloads, and can summarise many traces cheaply +// (ClickHouseStore). +type TraceQuerier interface { + // TraceEntries returns every stored entry that contains the trace, with + // each payload trimmed to that trace's spans/log records. Newest first. + TraceEntries(ctx context.Context, traceID string, limit int) ([]websocket.HistoryEntry, error) + // TraceSummaries returns summaries keyed by trace id; unknown ids are absent. + TraceSummaries(ctx context.Context, ids []string) (map[string]TraceSummary, error) +} + +// Occurrence is one persisted occurrence of an issue. Sample is the issue +// sample as JSON, so its shape can evolve without a schema change. +type Occurrence struct { + IssueID string + Kind string + Service string + Timestamp int64 + TraceID string + SpanID string + Sample json.RawMessage +} + +// IssueTrace is one trace an issue occurred in. +type IssueTrace struct { + TraceID string `json:"traceId"` + Hits int `json:"hits"` + FirstSeen int64 `json:"firstSeen"` + LastSeen int64 `json:"lastSeen"` + Summary *TraceSummary `json:"summary,omitempty"` + Samples []json.RawMessage `json:"-"` +} + +// IssueTracesOptions pages through an issue's traces, most recently hit first. +// Before is a LastSeen boundary (exclusive); zero starts from the newest. +type IssueTracesOptions struct { + Limit int + Before int64 +} + +// IssueTracesPage is one page of an issue's traces plus totals. Untraced holds +// occurrences that carried no trace id (only on the first page). +type IssueTracesPage struct { + Traces []IssueTrace + Untraced []json.RawMessage + TotalOccurrences int + HasMore bool +} + +// OccurrenceStore persists every issue occurrence and answers "which traces +// did this issue happen in" in one query (ClickHouseStore). +type OccurrenceStore interface { + AppendOccurrence(ctx context.Context, occ Occurrence) error + IssueTraces(ctx context.Context, issueID string, opts IssueTracesOptions) (IssueTracesPage, error) +} diff --git a/internal/storage/traceview.go b/internal/storage/traceview.go new file mode 100644 index 0000000..fc0a6dc --- /dev/null +++ b/internal/storage/traceview.go @@ -0,0 +1,251 @@ +package storage + +import ( + "encoding/json" + "strconv" + "strings" +) + +// spanFact is the slice of a span needed for trace summaries. +type spanFact struct { + TraceID string + SpanID string + ParentSpanID string + Service string + Name string + StartNs uint64 + EndNs uint64 + IsError bool +} + +type otlpValue struct { + StringValue string `json:"stringValue"` + IntValue json.Number `json:"intValue"` +} + +type otlpAttr struct { + Key string `json:"key"` + Value otlpValue `json:"value"` +} + +type otlpSpan struct { + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + ParentSpanID string `json:"parentSpanId"` + Name string `json:"name"` + StartTimeUnixNano json.Number `json:"startTimeUnixNano"` + EndTimeUnixNano json.Number `json:"endTimeUnixNano"` + Status struct { + Code int `json:"code"` + } `json:"status"` + Attributes []otlpAttr `json:"attributes"` +} + +// httpStatusOf reads http.status_code (old) / http.response.status_code +// (new) as a number, tolerating string- or int-encoded OTLP values. +func httpStatusOf(attrs []otlpAttr) int { + for _, a := range attrs { + if a.Key != "http.status_code" && a.Key != "http.response.status_code" { + continue + } + if n, err := a.Value.IntValue.Int64(); err == nil { + return int(n) + } + if n, err := strconv.Atoi(a.Value.StringValue); err == nil { + return n + } + } + return 0 +} + +// isErrorSpan mirrors the UI's rule: OTLP error status or HTTP status >= 400. +func isErrorSpan(s otlpSpan) bool { + return s.Status.Code == 2 || httpStatusOf(s.Attributes) >= 400 +} + +func parseUint(n json.Number) uint64 { + v, err := strconv.ParseUint(n.String(), 10, 64) + if err != nil { + return 0 + } + return v +} + +// extractSpanFacts flattens an OTLP traces payload into per-span facts. +func extractSpanFacts(data []byte) []spanFact { + var p struct { + ResourceSpans []struct { + Resource struct { + Attributes []otlpAttr `json:"attributes"` + } `json:"resource"` + ScopeSpans []struct { + Spans []otlpSpan `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` + } + if err := json.Unmarshal(data, &p); err != nil { + return nil + } + var out []spanFact + for _, rs := range p.ResourceSpans { + service := "unknown-service" + for _, a := range rs.Resource.Attributes { + if a.Key == "service.name" && a.Value.StringValue != "" { + service = a.Value.StringValue + } + } + for _, ss := range rs.ScopeSpans { + for _, sp := range ss.Spans { + if sp.TraceID == "" { + continue + } + start := parseUint(sp.StartTimeUnixNano) + end := parseUint(sp.EndTimeUnixNano) + if end < start { + end = start + } + out = append(out, spanFact{ + TraceID: sp.TraceID, SpanID: sp.SpanID, ParentSpanID: sp.ParentSpanID, + Service: service, Name: sp.Name, StartNs: start, EndNs: end, IsError: isErrorSpan(sp), + }) + } + } + } + return out +} + +// allTraceIDs lists the distinct trace ids in a payload (a single OTLP batch +// routinely carries spans/logs of many traces, so the scalar trace_id column +// alone cannot answer "which entries contain trace X"). +func allTraceIDs(data []byte) []string { + var p struct { + ResourceSpans []struct { + ScopeSpans []struct { + Spans []struct { + TraceID string `json:"traceId"` + } `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` + ResourceLogs []struct { + ScopeLogs []struct { + LogRecords []struct { + TraceID string `json:"traceId"` + } `json:"logRecords"` + } `json:"scopeLogs"` + } `json:"resourceLogs"` + } + if err := json.Unmarshal(data, &p); err != nil { + return nil + } + seen := map[string]struct{}{} + var out []string + add := func(id string) { + if id == "" { + return + } + if _, ok := seen[id]; !ok { + seen[id] = struct{}{} + out = append(out, id) + } + } + for _, rs := range p.ResourceSpans { + for _, ss := range rs.ScopeSpans { + for _, sp := range ss.Spans { + add(sp.TraceID) + } + } + } + for _, rl := range p.ResourceLogs { + for _, sl := range rl.ScopeLogs { + for _, rec := range sl.LogRecords { + add(rec.TraceID) + } + } + } + return out +} + +// filterNested walks raw (a JSON array) down path (object keys holding arrays +// at each level), keeping only leaf elements for which keep returns true and +// dropping any parent that ends up empty. Unknown fields are preserved +// verbatim. It returns the pruned array and the number of leaves kept. +func filterNested(raw json.RawMessage, path []string, keep func(json.RawMessage) bool) (json.RawMessage, int, error) { + var items []json.RawMessage + if err := json.Unmarshal(raw, &items); err != nil { + return nil, 0, err + } + kept := 0 + out := make([]json.RawMessage, 0, len(items)) + for _, item := range items { + if len(path) == 0 { + if keep(item) { + out = append(out, item) + kept++ + } + continue + } + var obj map[string]json.RawMessage + if err := json.Unmarshal(item, &obj); err != nil { + return nil, 0, err + } + child, ok := obj[path[0]] + if !ok { + continue + } + pruned, n, err := filterNested(child, path[1:], keep) + if err != nil { + return nil, 0, err + } + if n == 0 { + continue + } + prunedJSON := pruned + obj[path[0]] = prunedJSON + rebuilt, err := json.Marshal(obj) + if err != nil { + return nil, 0, err + } + out = append(out, rebuilt) + kept += n + } + b, err := json.Marshal(out) + return b, kept, err +} + +// TrimToTrace returns data reduced to the spans and log records of traceID, +// and false when nothing in it belongs to the trace. Payloads that are not +// OTLP traces/logs (metrics) report false. +func TrimToTrace(data []byte, traceID string) ([]byte, bool) { + var top map[string]json.RawMessage + if err := json.Unmarshal(data, &top); err != nil { + return nil, false + } + match := func(item json.RawMessage) bool { + var t struct { + TraceID string `json:"traceId"` + } + return json.Unmarshal(item, &t) == nil && strings.EqualFold(t.TraceID, traceID) + } + total := 0 + for key, path := range map[string][]string{ + "resourceSpans": {"scopeSpans", "spans"}, + "resourceLogs": {"scopeLogs", "logRecords"}, + } { + raw, ok := top[key] + if !ok { + continue + } + pruned, n, err := filterNested(raw, path, match) + if err != nil || n == 0 { + delete(top, key) + continue + } + top[key] = pruned + total += n + } + if total == 0 { + return nil, false + } + out, err := json.Marshal(top) + return out, err == nil +} diff --git a/internal/storage/traceview_test.go b/internal/storage/traceview_test.go new file mode 100644 index 0000000..620f85c --- /dev/null +++ b/internal/storage/traceview_test.go @@ -0,0 +1,114 @@ +package storage + +import ( + "encoding/json" + "testing" +) + +const mixedPayload = `{ + "resourceSpans":[ + {"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"api"}}]},"schemaUrl":"keep-me", + "scopeSpans":[{"scope":{"name":"lib"},"spans":[ + {"traceId":"aaa","spanId":"s1","parentSpanId":"","name":"GET /","startTimeUnixNano":"100","endTimeUnixNano":"300","status":{"code":0}}, + {"traceId":"bbb","spanId":"s2","parentSpanId":"","name":"other","startTimeUnixNano":"150","endTimeUnixNano":"160","status":{"code":2}}, + {"traceId":"aaa","spanId":"s3","parentSpanId":"s1","name":"db","startTimeUnixNano":"120","endTimeUnixNano":"200","status":{"code":0}, + "attributes":[{"key":"http.status_code","value":{"intValue":"503"}}]} + ]}]}, + {"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"only-bbb"}}]}, + "scopeSpans":[{"spans":[{"traceId":"bbb","spanId":"s4","name":"x","startTimeUnixNano":"1","endTimeUnixNano":"2"}]}]} + ], + "resourceLogs":[{"scopeLogs":[{"logRecords":[{"traceId":"aaa","body":{"stringValue":"hi"}},{"traceId":"ccc"}]}]}] +}` + +func TestTrimToTraceKeepsOnlyRequestedTrace(t *testing.T) { + out, ok := TrimToTrace([]byte(mixedPayload), "aaa") + if !ok { + t.Fatal("expected trace aaa to be found") + } + var p struct { + ResourceSpans []struct { + SchemaURL string `json:"schemaUrl"` + ScopeSpans []struct { + Scope struct{ Name string } `json:"scope"` + Spans []struct { + SpanID string `json:"spanId"` + } `json:"spans"` + } `json:"scopeSpans"` + } `json:"resourceSpans"` + ResourceLogs []struct { + ScopeLogs []struct { + LogRecords []struct { + TraceID string `json:"traceId"` + } `json:"logRecords"` + } `json:"scopeLogs"` + } `json:"resourceLogs"` + } + if err := json.Unmarshal(out, &p); err != nil { + t.Fatal(err) + } + if len(p.ResourceSpans) != 1 { + t.Fatalf("resource with only other traces should be dropped, got %d resources", len(p.ResourceSpans)) + } + rs := p.ResourceSpans[0] + if rs.SchemaURL != "keep-me" || rs.ScopeSpans[0].Scope.Name != "lib" { + t.Errorf("unknown/sibling fields must be preserved: %+v", rs) + } + got := []string{} + for _, s := range rs.ScopeSpans[0].Spans { + got = append(got, s.SpanID) + } + if len(got) != 2 || got[0] != "s1" || got[1] != "s3" { + t.Errorf("spans = %v, want [s1 s3]", got) + } + if len(p.ResourceLogs) != 1 || len(p.ResourceLogs[0].ScopeLogs[0].LogRecords) != 1 { + t.Errorf("logs not trimmed to aaa: %+v", p.ResourceLogs) + } +} + +func TestTrimToTraceMissAndNonTrace(t *testing.T) { + if _, ok := TrimToTrace([]byte(mixedPayload), "zzz"); ok { + t.Error("unknown trace should report false") + } + if _, ok := TrimToTrace([]byte(`{"resourceMetrics":[]}`), "aaa"); ok { + t.Error("metrics payload should report false") + } + if _, ok := TrimToTrace([]byte(`not json`), "aaa"); ok { + t.Error("invalid json should report false") + } +} + +func TestExtractSpanFacts(t *testing.T) { + facts := extractSpanFacts([]byte(mixedPayload)) + if len(facts) != 4 { + t.Fatalf("got %d facts, want 4", len(facts)) + } + by := map[string]spanFact{} + for _, f := range facts { + by[f.SpanID] = f + } + if f := by["s1"]; f.Service != "api" || f.StartNs != 100 || f.EndNs != 300 || f.IsError { + t.Errorf("s1 = %+v", f) + } + if !by["s2"].IsError { + t.Error("status code 2 must be an error") + } + if !by["s3"].IsError { + t.Error("http.status_code 503 (intValue) must be an error") + } + if by["s4"].Service != "only-bbb" { + t.Errorf("s4 service = %q", by["s4"].Service) + } +} + +func TestAllTraceIDsDistinctAcrossSpansAndLogs(t *testing.T) { + ids := allTraceIDs([]byte(mixedPayload)) + want := map[string]bool{"aaa": true, "bbb": true, "ccc": true} + if len(ids) != len(want) { + t.Fatalf("ids = %v", ids) + } + for _, id := range ids { + if !want[id] { + t.Errorf("unexpected id %q", id) + } + } +} diff --git a/otel-ui/src/components/IssueOccurrences.tsx b/otel-ui/src/components/IssueOccurrences.tsx new file mode 100644 index 0000000..1ab87fd --- /dev/null +++ b/otel-ui/src/components/IssueOccurrences.tsx @@ -0,0 +1,149 @@ +import { useMemo, useState } from "react" +import { MasterDetail } from "./MasterDetail" +import { BackButton } from "./common/BackButton" +import { SectionLabel } from "./common/DetailField" +import { Stat, StatHeader } from "./common/Stat" +import { TraceDetail } from "./trace/TraceDetail" +import { TraceOverviewHeader, TraceOverviewRow } from "./trace/TraceOverviewRow" +import { OccurrenceDetail } from "./issues/OccurrenceDetail" +import { useIssueTraces } from "../hooks/useIssueTraces" +import { useProcessedTraces } from "../hooks/useProcessedTraces" +import { formatDuration } from "../lib/timeUtils" +import type { Issue, IssueSample, TraceSummary } from "../lib/types" + +/** One selectable row: a trace with its occurrences, or a lone traceless occurrence. */ +interface Row { + key: string + traceId?: string + samples: IssueSample[] + ts: number + hits: number + summary?: TraceSummary +} + +/** + * Focused view of one issue: an overview of every trace it occurred in + * (server-side summaries, paged), with the selected trace's waterfall and its + * occurrences below. `initialTraceId` preselects the trace the user came from. + */ +export const IssueOccurrences = ({ + issue, + initialTraceId, + onBack, +}: { + issue: Issue + initialTraceId?: string + onBack: () => void +}) => { + const { traces, untraced, total, complete, hasMore, loading, error, loadMore } = useIssueTraces(issue.id) + + const rows: Row[] = useMemo(() => { + const traced = traces.map((t) => ({ + key: t.traceId, + traceId: t.traceId, + samples: t.samples, + ts: t.firstSeen, + hits: t.hits, + summary: t.summary, + })) + const loose = untraced.map((s, i) => ({ key: `s-${s.timestamp}-${i}`, samples: [s], ts: s.timestamp, hits: 1 })) + return [...traced, ...loose].sort((a, b) => b.ts - a.ts) + }, [traces, untraced]) + + const [picked, setPicked] = useState(initialTraceId ?? null) + const selected = rows.find((r) => r.key === picked) ?? rows[0] + const selectedTraceId = selected?.traceId + + // Spans are fetched only for the selected trace. + const { byId, loading: spansLoading, failed } = useProcessedTraces(selectedTraceId ? [selectedTraceId] : []) + + const stats = useMemo(() => { + const sums = traces.flatMap((t) => (t.summary ? [t.summary] : [])) + return { + maxDuration: Math.max(1, ...sums.map((s) => s.duration)), + slowest: Math.max(0, ...sums.map((s) => s.duration)), + errored: sums.filter((s) => s.errorCount > 0).length, + services: new Set(sums.flatMap((s) => s.services)).size, + known: sums.length > 0, + } + }, [traces]) + + return ( +
+ } + eyebrow={`service=${issue.service}`} + title={issue.title} + stats={ + <> + + + + + + + } + /> + + {error &&

Failed to load traces: {error}

} + {!complete && ( +

+ Showing only the {issue.samples.length} most recent occurrences; full history needs ClickHouse storage. +

+ )} + {loading && rows.length === 0 &&

Loading traces…

} + + r.key} + selectedKey={selected?.key ?? null} + onSelect={setPicked} + label="Traces containing this issue" + listHeader={} + renderItem={(r) => ( + + )} + renderDetail={(r) => ( +
+ {r.traceId && ( + + )} + + Occurrences in this {r.traceId ? "trace" : "record"} ({r.hits} + {r.samples.length < r.hits ? `, latest ${r.samples.length} shown` : ""}) + + {r.samples.map((s, i) => ( + + ))} +
+ )} + /> + + {hasMore && ( + + )} +
+ ) +} + +export default IssueOccurrences diff --git a/otel-ui/src/components/IssuesVisualizer.tsx b/otel-ui/src/components/IssuesVisualizer.tsx index c90b261..94abafe 100644 --- a/otel-ui/src/components/IssuesVisualizer.tsx +++ b/otel-ui/src/components/IssuesVisualizer.tsx @@ -32,6 +32,7 @@ import { strongestCorrelate, type SeverityTone, } from "../lib/issue-utils" +import { IssueOccurrences } from "./IssueOccurrences" import { NeedleTestModal } from "./NeedleTestModal" import { FilterBar, type FilterChip } from "./FilterBar" import type { Issue, IssueSample } from "../lib/types" @@ -277,12 +278,13 @@ const IssueCard = ({ issue, onStatusChange, statusBusy, + onFocus, }: { issue: Issue onStatusChange: (issue: Issue, status: string) => void statusBusy: boolean + onFocus: (issue: Issue, traceId?: string) => void }) => { - const { setActiveTraceId, setActiveSpanId, setActiveTab } = useObservabilityData() const [expanded, setExpanded] = useState(false) const [needleResult, setNeedleResult] = useState(null) const [needleLoading, setNeedleLoading] = useState(false) @@ -311,11 +313,7 @@ const IssueCard = ({ const contextSample = issue.samples.find((s) => s.httpMethod || s.httpRoute || s.httpStatus) const correlate = strongestCorrelate(issue) - const viewTrace = (traceId: string, spanId?: string) => { - setActiveTraceId(traceId) - if (spanId) setActiveSpanId(spanId) - setActiveTab("tab1") - } + const viewTrace = (traceId: string) => onFocus(issue, traceId) const metaParts = [`service=${issue.service}`] if (contextSample?.httpMethod && contextSample?.httpRoute) { @@ -360,7 +358,7 @@ const IssueCard = ({ {latestSample?.traceId && ( +) diff --git a/otel-ui/src/components/common/DetailField.tsx b/otel-ui/src/components/common/DetailField.tsx new file mode 100644 index 0000000..7c5dc41 --- /dev/null +++ b/otel-ui/src/components/common/DetailField.tsx @@ -0,0 +1,12 @@ +/** Label/value pair for a
; renders nothing for empty values. */ +export const DetailField = ({ label, value }: { label: string; value?: string | number }) => + value === undefined || value === "" ? null : ( +
+
{label}
+
{value}
+
+ ) + +export const SectionLabel = ({ children }: { children: React.ReactNode }) => ( +

{children}

+) diff --git a/otel-ui/src/components/common/Stat.tsx b/otel-ui/src/components/common/Stat.tsx new file mode 100644 index 0000000..ddba3ff --- /dev/null +++ b/otel-ui/src/components/common/Stat.tsx @@ -0,0 +1,34 @@ +export const Stat = ({ label, value, tone }: { label: string; value: string | number; tone?: "critical" }) => ( +
+ {label} + + {value} + +
+) + +/** Header strip: optional back button, title block, then stats on the right. */ +export const StatHeader = ({ + eyebrow, + title, + back, + stats, +}: { + eyebrow?: string + title: string + back?: React.ReactNode + stats: React.ReactNode +}) => ( +
+ {back} +
+ {eyebrow &&

{eyebrow}

} +

+ {title} +

+
+ {stats} +
+) diff --git a/otel-ui/src/components/issues/OccurrenceDetail.tsx b/otel-ui/src/components/issues/OccurrenceDetail.tsx new file mode 100644 index 0000000..b176294 --- /dev/null +++ b/otel-ui/src/components/issues/OccurrenceDetail.tsx @@ -0,0 +1,51 @@ +import { ExternalLink } from "lucide-react" +import { DetailField } from "../common/DetailField" +import { formatRelativeTime } from "../../lib/timeUtils" +import { parseBreadcrumbs } from "../../lib/issue-utils" +import type { IssueSample } from "../../lib/types" + +export const OccurrenceDetail = ({ sample }: { sample: IssueSample }) => { + const stack = sample.resolvedStacktrace || sample.stacktrace + const crumbs = parseBreadcrumbs(sample.breadcrumbs) + return ( +
+
+

+ {new Date(sample.timestamp / 1e6).toLocaleString()} · {formatRelativeTime(sample.timestamp / 1e6)} +

+

{sample.message}

+
+
+ + + + + + + +
+ {sample.url && ( + + + )} + {stack && ( + 
+          {stack}
+
+ )} + {crumbs.length > 0 && ( +
+ {crumbs.map((c, i) => ( +
+ {new Date(c.t).toLocaleTimeString([], { hour12: false })} + {c.category} + {c.message.slice(0, 200)} +
+ ))} +
+ )} +
+ ) +} diff --git a/otel-ui/src/components/trace/TraceDetail.tsx b/otel-ui/src/components/trace/TraceDetail.tsx new file mode 100644 index 0000000..7a3a850 --- /dev/null +++ b/otel-ui/src/components/trace/TraceDetail.tsx @@ -0,0 +1,48 @@ +import { Activity } from "lucide-react" +import { SingleTraceView, type ProcessedTrace } from "../TraceVisualization" +import { useObservabilityData } from "../../hooks/useObservabilityData" +import { SectionLabel } from "../common/DetailField" + +/** Waterfall + span inspector for one trace, with an "Open in Traces" jump. Handles the not-yet-loaded state. */ +export const TraceDetail = ({ + traceId, + trace, + loading, + spanId, + label = "Selected trace", +}: { + traceId: string + trace?: ProcessedTrace + loading?: boolean + spanId?: string + label?: string +}) => { + const { setActiveTraceId, setActiveSpanId, setActiveTab } = useObservabilityData() + const open = () => { + setActiveTraceId(traceId) + if (spanId) setActiveSpanId(spanId) + setActiveTab("tab1") + } + return ( +
+
+ {label} + +
+ {trace ? ( + + ) : ( +

+ {loading ? "Loading trace…" : "Trace spans are no longer available."} +

+ )} +
+ ) +} diff --git a/otel-ui/src/components/trace/TraceOverviewRow.tsx b/otel-ui/src/components/trace/TraceOverviewRow.tsx new file mode 100644 index 0000000..735b5ad --- /dev/null +++ b/otel-ui/src/components/trace/TraceOverviewRow.tsx @@ -0,0 +1,83 @@ +import { Loader2 } from "lucide-react" +import type { TraceSummary } from "../../lib/types" +import { formatDuration } from "../../lib/timeUtils" + +const GRID = + "grid grid-cols-[84px_minmax(0,1.6fr)_minmax(140px,1fr)_56px_minmax(0,1fr)_64px_64px] items-center gap-3" + +export const clockTime = (ns: number) => + new Date(ns / 1e6).toLocaleTimeString([], { hour12: false, hour: "2-digit", minute: "2-digit", second: "2-digit" }) + +/** Column header matching TraceOverviewRow. Pass as MasterDetail's listHeader. */ +export const TraceOverviewHeader = ({ hitsLabel = "Hits" }: { hitsLabel?: string }) => ( +
+ Time + Trace + Duration + Spans + Services + Errors + {hitsLabel} +
+) + +/** + * One trace in an overview table, rendered from a server-side TraceSummary + * (no spans needed). `summary` may be undefined for traces the server has no + * rollup for; `fallbackTitle` and `traceId` keep the row useful then. + * `maxDuration` scales the duration bar across the visible set. + */ +export const TraceOverviewRow = ({ + summary: trace, + traceId, + timestampNs, + fallbackTitle, + maxDuration, + hits, + loading, +}: { + summary?: TraceSummary + traceId?: string + timestampNs: number + fallbackTitle: string + maxDuration: number + hits?: number + loading?: boolean +}) => ( +
+ {clockTime(timestampNs)} + + {trace?.name ?? fallbackTitle} + + {traceId ? `trace:${traceId.slice(0, 12)}` : "no trace attached"} + + + + {trace ? ( + <> + + 0 ? "bg-ds-critical" : "bg-ds-accent"}`} + style={{ width: `${Math.max(2, (trace.duration / maxDuration) * 100)}%` }} + /> + + + {formatDuration(trace.duration)} + + + ) : loading ? ( + + ) : ( + – + )} + + {trace?.spanCount ?? "–"} + {trace?.services.join(", ") || "–"} + 0 ? "text-ds-critical" : "text-ds-text-3"}`}> + {trace ? trace.errorCount : "–"} + + + {hits === undefined ? "" : `×${hits}`} + +
+) diff --git a/otel-ui/src/hooks/useIssueTraces.ts b/otel-ui/src/hooks/useIssueTraces.ts new file mode 100644 index 0000000..80a6075 --- /dev/null +++ b/otel-ui/src/hooks/useIssueTraces.ts @@ -0,0 +1,55 @@ +import { useCallback, useEffect, useRef, useState } from "react" +import { authFetch } from "../lib/auth" +import type { IssueSample, IssueTrace, IssueTracesResponse } from "../lib/types" + +/** + * Loads every trace an issue occurred in (one request, summaries included) and + * pages further back with loadMore(). Spans are NOT included: load the selected + * trace's spans separately (useProcessedTraces). + */ +export function useIssueTraces(issueId: string, pageSize = 50) { + const [traces, setTraces] = useState([]) + const [untraced, setUntraced] = useState([]) + const [meta, setMeta] = useState({ total: 0, complete: true, hasMore: false }) + const [loading, setLoading] = useState(true) + const [error, setError] = useState(null) + const nextBefore = useRef(undefined) + const gen = useRef(0) + + const fetchPage = useCallback( + async (before?: number) => { + const my = gen.current + setLoading(true) + try { + const q = new URLSearchParams({ limit: String(pageSize) }) + if (before) q.set("before", String(before)) + const res = await authFetch(`/api/issues/${encodeURIComponent(issueId)}/traces?${q}`) + if (!res.ok) throw new Error(`HTTP ${res.status}`) + const data = (await res.json()) as IssueTracesResponse + if (my !== gen.current) return + setTraces((prev) => (before ? [...prev, ...data.traces] : data.traces)) + if (!before) setUntraced(data.untraced) + setMeta({ total: data.totalOccurrences, complete: data.complete, hasMore: data.hasMore }) + nextBefore.current = data.nextBefore + setError(null) + } catch (e) { + if (my === gen.current) setError(e instanceof Error ? e.message : String(e)) + } finally { + if (my === gen.current) setLoading(false) + } + }, + [issueId, pageSize], + ) + + useEffect(() => { + gen.current++ + nextBefore.current = undefined + void fetchPage() + }, [fetchPage]) + + const loadMore = useCallback(() => { + if (nextBefore.current) void fetchPage(nextBefore.current) + }, [fetchPage]) + + return { traces, untraced, ...meta, loading, error, loadMore } +} diff --git a/otel-ui/src/hooks/useProcessedTraces.ts b/otel-ui/src/hooks/useProcessedTraces.ts new file mode 100644 index 0000000..7dd44f5 --- /dev/null +++ b/otel-ui/src/hooks/useProcessedTraces.ts @@ -0,0 +1,28 @@ +import { useMemo } from "react" +import { processTraces, type ProcessedTrace } from "../components/TraceVisualization" +import { useObservabilityData } from "./useObservabilityData" +import { useTracesById } from "./useTracesById" + +/** Fetch traces by id and return them processed (spans, duration, services, errors), keyed by trace id. */ +export function useProcessedTraces(ids: string[]) { + const { logCountBySpanId } = useObservabilityData() + const { traces, loading, failed } = useTracesById(ids) + const byId = useMemo(() => { + const map = new Map() + processTraces(traces, logCountBySpanId).forEach((t) => map.set(t.id, t)) + return map + }, [traces, logCountBySpanId]) + + const stats = useMemo(() => { + const all = [...byId.values()] + return { + maxDuration: Math.max(1, ...all.map((t) => t.duration)), + slowest: all.reduce((m, t) => Math.max(m, t.duration), 0), + errored: all.filter((t) => t.hasError).length, + services: new Set(all.flatMap((t) => t.services)).size, + loaded: all.length, + } + }, [byId]) + + return { byId, stats, loading, failed } +} diff --git a/otel-ui/src/hooks/useTracesById.ts b/otel-ui/src/hooks/useTracesById.ts new file mode 100644 index 0000000..bdaba8d --- /dev/null +++ b/otel-ui/src/hooks/useTracesById.ts @@ -0,0 +1,43 @@ +import { useEffect, useMemo, useState } from "react" +import { authFetch } from "../lib/auth" +import { partitionMessages } from "../lib/ingest" +import type { TraceData } from "../lib/types" + +/** + * Fetches the given traces from /api/traces/:id (once each, cached for the + * hook's lifetime). Returns the raw payloads so callers can run them through + * processTraces(); `loading` is true while any id is still in flight. + */ +export function useTracesById(ids: string[]) { + const [byId, setById] = useState>({}) + const [failed, setFailed] = useState>(new Set()) + const key = ids.join(",") + + useEffect(() => { + let cancelled = false + for (const id of key ? key.split(",") : []) { + if (byId[id] || failed.has(id)) continue + authFetch(`/api/traces/${encodeURIComponent(id)}`) + .then((res) => { + if (!res.ok) throw new Error(`HTTP ${res.status}`) + return res.json() + }) + .then((data) => { + if (cancelled) return + setById((prev) => ({ ...prev, [id]: partitionMessages(data ?? []).traces })) + }) + .catch(() => { + if (!cancelled) setFailed((prev) => new Set(prev).add(id)) + }) + } + return () => { + cancelled = true + } + // byId/failed are read only to skip already-settled ids. + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [key]) + + const traces = useMemo(() => Object.values(byId).flat(), [byId]) + const loading = ids.some((id) => !byId[id] && !failed.has(id)) + return { traces, loading, failed } +} diff --git a/otel-ui/src/lib/issue-utils.ts b/otel-ui/src/lib/issue-utils.ts index 5413648..3645fc1 100644 --- a/otel-ui/src/lib/issue-utils.ts +++ b/otel-ui/src/lib/issue-utils.ts @@ -190,3 +190,32 @@ export function sparklineSeries(issue: Issue, hours = 24): SparklineSeries { } return { points, max, reference: mean, breachIndex } } + +/** One row per trace (all its samples grouped), plus one per traceless sample; newest first. */ +export interface SampleGroup { + key: string + traceId?: string + samples: IssueSample[] + firstTs: number +} + +export function groupSamplesByTrace(samples: IssueSample[]): SampleGroup[] { + const byTrace = new Map() + const out: SampleGroup[] = [] + samples.forEach((sample, i) => { + if (!sample.traceId) { + out.push({ key: `s-${sample.timestamp}-${i}`, samples: [sample], firstTs: sample.timestamp }) + return + } + const group = byTrace.get(sample.traceId) + if (group) { + group.samples.push(sample) + group.firstTs = Math.min(group.firstTs, sample.timestamp) + } else { + const created = { key: sample.traceId, traceId: sample.traceId, samples: [sample], firstTs: sample.timestamp } + byTrace.set(sample.traceId, created) + out.push(created) + } + }) + return out.sort((a, b) => b.firstTs - a.firstTs) +} diff --git a/otel-ui/src/lib/trace-summary.ts b/otel-ui/src/lib/trace-summary.ts new file mode 100644 index 0000000..4872731 --- /dev/null +++ b/otel-ui/src/lib/trace-summary.ts @@ -0,0 +1,16 @@ +import type { ProcessedTrace } from "../components/TraceVisualization" +import type { TraceSummary } from "./types" + +/** Same shape the server returns, derived from a fully loaded trace (for lists built from live data). */ +export function summarizeTrace(t: ProcessedTrace): TraceSummary { + return { + id: t.id, + name: t.name, + startTime: t.startTime, + endTime: t.endTime, + duration: t.duration, + spanCount: t.spans.length, + errorCount: t.errorCount, + services: t.services, + } +} diff --git a/otel-ui/src/lib/types.ts b/otel-ui/src/lib/types.ts index dc42c92..1095d15 100644 --- a/otel-ui/src/lib/types.ts +++ b/otel-ui/src/lib/types.ts @@ -266,3 +266,36 @@ export interface ErrorStatsResponse { export interface ReleaseStatsResponse { series: Record; } + +/** Per-trace rollup for overview tables (server-side, no spans). Times are unix ns. */ +export interface TraceSummary { + id: string; + name: string; + startTime: number; + endTime: number; + duration: number; + spanCount: number; + errorCount: number; + services: string[]; +} + +/** One trace an issue occurred in, from GET /api/issues/:id/traces. */ +export interface IssueTrace { + traceId: string; + hits: number; + firstSeen: number; + lastSeen: number; + summary?: TraceSummary; + samples: IssueSample[]; +} + +export interface IssueTracesResponse { + issueId: string; + totalOccurrences: number; + traces: IssueTrace[]; + untraced: IssueSample[]; + hasMore: boolean; + nextBefore?: number; + /** false when the server only had the bounded in-memory samples. */ + complete: boolean; +}