materialized view and a bit better ui

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)

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)

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)
+}

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)
+	}
+}