114 lines
4.6 KiB
Go
114 lines
4.6 KiB
Go
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")
|
|
}
|
|
}
|