all the changes

internal/storage/normalized.go

@@ -0,0 +1,107 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+
+	"git.tornberg.me/go-otel/internal/normalization"
+	"git.tornberg.me/go-otel/internal/websocket"
+)
+
+// normalizedColumns is the explicit insert column list. Tables migrated with
+// ALTER ADD COLUMN have a different physical column order than a fresh
+// CREATE, so positional inserts are not safe.
+const normalizedColumns = `event_id, entry_type, event_index, ingest_timestamp, timestamp, service, release, environment,
+	kind, source, is_error, status_class, status_code, duration_ms, route, span_name, metric_name, metric_type, metric_unit,
+	metric_value, metric_count, metric_sum, metric_min, metric_max, exception_type, message_normalized, stack_signature,
+	trace_id, issue_fingerprint, is_retry, is_timeout, is_cancellation, exact_signature, pattern_signature, version,
+	method, host, user_agent, origin_host, referer_host, fetch_site, fetch_mode`
+
+// NormalizedInsertSQL is the INSERT prefix for a normalized_events batch.
+func NormalizedInsertSQL(db string) string {
+	return fmt.Sprintf("INSERT INTO %s.normalized_events (%s)", db, normalizedColumns)
+}
+
+// EnsureNormalizedSchema creates normalized_events and applies the additive
+// migrations. Idempotent.
+func EnsureNormalizedSchema(ctx context.Context, conn driver.Conn, db string) error {
+	query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.normalized_events (
+		event_id String,
+		entry_type LowCardinality(String),
+		event_index UInt32,
+		ingest_timestamp DateTime64(9, 'UTC'),
+		timestamp DateTime64(9, 'UTC'),
+		service LowCardinality(String), release LowCardinality(String), environment LowCardinality(String),
+		kind LowCardinality(String), source LowCardinality(String), is_error UInt8,
+		status_class LowCardinality(String), status_code UInt16, duration_ms Float64,
+		route String, span_name String, metric_name String, metric_type LowCardinality(String), metric_unit String,
+		metric_value Float64, metric_count UInt64, metric_sum Float64, metric_min Float64, metric_max Float64,
+		exception_type String, message_normalized String, stack_signature String, trace_id String, issue_fingerprint String,
+		is_retry UInt8, is_timeout UInt8, is_cancellation UInt8,
+		exact_signature String, pattern_signature String, version UInt8 DEFAULT 1,
+		method String DEFAULT '', host String DEFAULT '', user_agent String DEFAULT '',
+		origin_host String DEFAULT '', referer_host String DEFAULT '', fetch_site String DEFAULT '', fetch_mode String DEFAULT ''
+	) ENGINE = ReplacingMergeTree(version)
+	PARTITION BY toYYYYMM(timestamp)
+	ORDER BY (service, kind, pattern_signature, timestamp, event_id)
+	TTL toDateTime(timestamp) + INTERVAL 180 DAY`, db)
+	if err := conn.Exec(ctx, query); err != nil {
+		return err
+	}
+	// Developer tables predate these columns; historical rows get safe defaults.
+	for _, col := range []string{
+		"issue_fingerprint String DEFAULT ''", "is_retry UInt8 DEFAULT 0", "is_timeout UInt8 DEFAULT 0",
+		"is_cancellation UInt8 DEFAULT 0", "method String DEFAULT ''", "host String DEFAULT ''",
+		"user_agent String DEFAULT ''", "origin_host String DEFAULT ''", "referer_host String DEFAULT ''",
+		"fetch_site String DEFAULT ''", "fetch_mode String DEFAULT ''",
+	} {
+		if err := conn.Exec(ctx, fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS %s", db, col)); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// AppendNormalizedEvent adds one event to a batch prepared with NormalizedInsertSQL.
+func AppendNormalizedEvent(batch driver.Batch, e normalization.Event) error {
+	return batch.Append(
+		e.EventID, e.EntryType, e.EventIndex, time.Unix(0, e.IngestTimestamp), time.Unix(0, e.Timestamp),
+		e.Service, e.Release, e.Environment, e.Kind, e.Source, boolToUint8(e.IsError),
+		e.StatusClass, uint16(e.StatusCode), e.DurationMS, e.Route, e.SpanName, e.MetricName, e.MetricType, e.MetricUnit,
+		e.MetricValue, e.MetricCount, e.MetricSum, e.MetricMin, e.MetricMax,
+		e.ExceptionType, e.MessageNormalized, e.StackSignature, e.TraceID, e.IssueFingerprint,
+		boolToUint8(e.IsRetry), boolToUint8(e.IsTimeout), boolToUint8(e.IsCancellation), e.ExactSignature, e.PatternSignature,
+		uint8(1), e.Method, e.Host, e.UserAgent, e.OriginHost, e.RefererHost, e.FetchSite, e.FetchMode,
+	)
+}
+
+// appendNormalized derives and stores normalized rows for one raw entry.
+// The raw entry is the source of truth, so a failure here is logged and never
+// fails ingest; the backfill tool can rebuild any gap.
+func (s *ClickHouseStore) appendNormalized(ctx context.Context, entry websocket.HistoryEntry) {
+	events, err := normalization.NormalizeEntry(entry.Type, entry.Timestamp, entry.Data)
+	if err != nil || len(events) == 0 {
+		return
+	}
+	ctx = clickhouse.Context(ctx, clickhouse.WithSettings(clickhouse.Settings{"async_insert": 1, "wait_for_async_insert": 0}))
+	batch, err := s.conn.PrepareBatch(ctx, NormalizedInsertSQL(s.db))
+	if err != nil {
+		log.Printf("normalized insert prepare: %v", err)
+		return
+	}
+	for _, e := range events {
+		if err := AppendNormalizedEvent(batch, e); err != nil {
+			_ = batch.Abort()
+			log.Printf("normalized insert append: %v", err)
+			return
+		}
+	}
+	if err := batch.Send(); err != nil {
+		log.Printf("normalized insert send: %v", err)
+	}
+}