132 lines
4.0 KiB
Go
132 lines
4.0 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/websocket"
|
|
)
|
|
|
|
// TestClickHouseStoreIntegration is skipped unless OTEL_CLICKHOUSE_TEST_ADDR
|
|
// points at a disposable ClickHouse (native port 9000). It exercises the real
|
|
// SQL: schema creation, Append, rollup-triggered stats and Search filters.
|
|
//
|
|
// Placeholder round-trips through AsyncInsert use wait_for_async_insert so
|
|
// reads below see the writes without sleeps.
|
|
func TestClickHouseStoreIntegration(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(), 60*time.Second)
|
|
defer cancel()
|
|
|
|
// Unique database per run so repeated runs don't collide.
|
|
db := "otel_test_" + time.Now().UTC().Format("150405")
|
|
cfg := ClickHouseConfig{Addr: addr, Database: db, Username: os.Getenv("OTEL_CLICKHOUSE_TEST_USER"), Password: os.Getenv("OTEL_CLICKHOUSE_TEST_PASSWORD")}
|
|
s, err := NewClickHouseStore(ctx, cfg)
|
|
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()
|
|
}()
|
|
|
|
now := time.Now().UTC()
|
|
mk := func(minsAgo int, typ, service string, isErr bool, release string) websocket.HistoryEntry {
|
|
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(),
|
|
Data: raw,
|
|
}
|
|
}
|
|
entries := []websocket.HistoryEntry{
|
|
mk(200, "logs", "api", true, "v1.3"),
|
|
mk(190, "logs", "api", false, "v1.3"),
|
|
mk(100, "logs", "api", true, "v1.4"),
|
|
mk(90, "logs", "api", false, "v1.4"),
|
|
mk(30, "logs", "api", false, "v1.4"),
|
|
}
|
|
for i, e := range entries {
|
|
if err := s.Append(ctx, e); err != nil {
|
|
t.Fatalf("Append[%d]: %v", i, err)
|
|
}
|
|
}
|
|
// Flush async insert buffer before reading.
|
|
if err := s.conn.Exec(ctx, "SYSTEM FLUSH ASYNC INSERT QUEUE"); err != nil {
|
|
t.Fatalf("flush: %v", err)
|
|
}
|
|
|
|
// Search: newest first, limit respected.
|
|
res, err := s.Search(ctx, SearchOptions{Type: "logs"}, 3)
|
|
if err != nil {
|
|
t.Fatalf("Search: %v", err)
|
|
}
|
|
if len(res) != 3 {
|
|
t.Fatalf("expected 3 search results, got %d", len(res))
|
|
}
|
|
if res[0].Timestamp < res[2].Timestamp {
|
|
t.Errorf("expected newest first")
|
|
}
|
|
|
|
// Search: substring filter.
|
|
res, err = s.Search(ctx, SearchOptions{Query: "\"n\":200"}, 10)
|
|
if err != nil {
|
|
t.Fatalf("Search query: %v", err)
|
|
}
|
|
if len(res) != 1 {
|
|
t.Errorf("expected substring filter to match 1 entry, got %d", len(res))
|
|
}
|
|
|
|
// ErrorStats via the rollup MV: two hours with data, 2 errors total.
|
|
points, err := s.ErrorStats(ctx, "api", 24, false)
|
|
if err != nil {
|
|
t.Fatalf("ErrorStats: %v", err)
|
|
}
|
|
errSum, totSum := uint64(0), uint64(0)
|
|
for _, p := range points {
|
|
errSum += p.Errors
|
|
totSum += p.Total
|
|
}
|
|
if errSum != 2 || totSum != 5 {
|
|
t.Errorf("expected rollup totals errors=2 total=5, got errors=%d total=%d", errSum, totSum)
|
|
}
|
|
|
|
// ReleaseSeries: v1.3 has 2 entries (1 error), v1.4 has 3 (1 error).
|
|
series, err := s.ReleaseSeries(ctx, "api", 24)
|
|
if err != nil {
|
|
t.Fatalf("ReleaseSeries: %v", err)
|
|
}
|
|
var v13, v14 uint64
|
|
for _, p := range series["v1.3"] {
|
|
v13 += p.Total
|
|
}
|
|
for _, p := range series["v1.4"] {
|
|
v14 += p.Total
|
|
}
|
|
if v13 != 2 || v14 != 3 {
|
|
t.Errorf("expected per-release totals v1.3=2 v1.4=3, got %d/%d", v13, v14)
|
|
}
|
|
|
|
// Services.
|
|
svcs, err := s.Services(ctx)
|
|
if err != nil || len(svcs) == 0 || svcs[0] != "api" {
|
|
t.Errorf("expected services [api], got %v (err=%v)", svcs, err)
|
|
}
|
|
}
|