118 lines
6.2 KiB
Go
118 lines
6.2 KiB
Go
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, span_id, parent_span_id, span_kind, peer_ip, peer_ip_source, edge_peer_ip, edge_peer_source, edge_service, edge_span_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, ua_class`
|
|
|
|
// 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,
|
|
span_id String DEFAULT '', parent_span_id String DEFAULT '', span_kind UInt8 DEFAULT 0,
|
|
peer_ip String DEFAULT '', peer_ip_source LowCardinality(String) DEFAULT '',
|
|
edge_peer_ip String DEFAULT '', edge_peer_source LowCardinality(String) DEFAULT '',
|
|
edge_service LowCardinality(String) DEFAULT '', edge_span_id String DEFAULT '', 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 '', ua_class LowCardinality(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{
|
|
"span_id String DEFAULT ''", "parent_span_id String DEFAULT ''", "span_kind UInt8 DEFAULT 0",
|
|
"peer_ip String DEFAULT ''", "peer_ip_source LowCardinality(String) DEFAULT ''",
|
|
"edge_peer_ip String DEFAULT ''", "edge_peer_source LowCardinality(String) DEFAULT ''",
|
|
"edge_service LowCardinality(String) DEFAULT ''", "edge_span_id String DEFAULT ''",
|
|
"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 ''", "ua_class LowCardinality(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.SpanID, e.ParentSpanID, uint8(e.SpanKind),
|
|
e.PeerIP, e.PeerIPSource, e.EdgePeerIP, e.EdgePeerSource, e.EdgeService, e.EdgeSpanID, e.IssueFingerprint,
|
|
boolToUint8(e.IsRetry), boolToUint8(e.IsTimeout), boolToUint8(e.IsCancellation), e.ExactSignature, e.PatternSignature,
|
|
uint8(2), e.Method, e.Host, e.UserAgent, e.OriginHost, e.RefererHost, e.FetchSite, e.FetchMode, e.UAClass,
|
|
)
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
}
|