237 lines
8.8 KiB
Go
237 lines
8.8 KiB
Go
// normalized-backfill previews or persists deterministic normalized events.
|
|
// It is read-only unless -write is explicitly supplied.
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"sort"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/normalization"
|
|
"github.com/ClickHouse/clickhouse-go/v2"
|
|
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
|
)
|
|
|
|
type patternCount struct {
|
|
Service string
|
|
Kind string
|
|
Pattern string
|
|
Count uint64
|
|
}
|
|
|
|
func main() {
|
|
addr := flag.String("addr", envOr("OTEL_CLICKHOUSE_ADDR", "127.0.0.1:9000"), "ClickHouse host:port")
|
|
database := flag.String("database", envOr("OTEL_CLICKHOUSE_DB", "otel"), "ClickHouse database")
|
|
user := flag.String("user", envOr("OTEL_CLICKHOUSE_USER", "default"), "ClickHouse user")
|
|
password := flag.String("password", os.Getenv("OTEL_CLICKHOUSE_PASSWORD"), "ClickHouse password")
|
|
limit := flag.Int("limit", 0, "raw envelopes to process; 0 means all")
|
|
since := flag.String("since", "", "inclusive UTC window start, RFC3339")
|
|
until := flag.String("until", "", "exclusive UTC window end, RFC3339")
|
|
batchSize := flag.Int("batch-size", 1000, "normalized events per insert batch")
|
|
write := flag.Bool("write", false, "create the table and insert normalized events")
|
|
flag.Parse()
|
|
if *batchSize < 1 {
|
|
log.Fatal("-batch-size must be positive")
|
|
}
|
|
start, end := parseWindow(*since, *until)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
|
|
defer cancel()
|
|
conn, err := clickhouse.Open(&clickhouse.Options{Addr: []string{*addr}, Auth: clickhouse.Auth{Database: *database, Username: *user, Password: *password}})
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
if err := conn.Ping(ctx); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if *write {
|
|
if err := createSchema(ctx, conn, *database); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
|
|
whereSQL := ""
|
|
if !start.IsZero() {
|
|
whereSQL = fmt.Sprintf(" timestamp >= toDateTime64('%s', 9, 'UTC')", start.Format("2006-01-02 15:04:05.000000000"))
|
|
}
|
|
if !end.IsZero() {
|
|
if whereSQL != "" {
|
|
whereSQL += " AND"
|
|
}
|
|
whereSQL += fmt.Sprintf(" timestamp < toDateTime64('%s', 9, 'UTC')", end.Format("2006-01-02 15:04:05.000000000"))
|
|
}
|
|
if whereSQL != "" {
|
|
whereSQL = " WHERE" + whereSQL
|
|
}
|
|
limitSQL := ""
|
|
if *limit > 0 {
|
|
limitSQL = fmt.Sprintf(" LIMIT %d", *limit)
|
|
}
|
|
rows, err := conn.Query(ctx, fmt.Sprintf("SELECT timestamp, type, data FROM %s.entries%s%s", *database, whereSQL, limitSQL))
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
counts := map[string]*patternCount{}
|
|
rawCount, eventCount := 0, 0
|
|
var batch driver.Batch
|
|
if *write {
|
|
batch, err = conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s.normalized_events", *database))
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
flush := func() {
|
|
if batch != nil {
|
|
if err := batch.Send(); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
batch = nil
|
|
}
|
|
}
|
|
for rows.Next() {
|
|
var timestamp time.Time
|
|
var entryType, data string
|
|
if err := rows.Scan(×tamp, &entryType, &data); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
rawCount++
|
|
events, err := normalization.NormalizeEntry(entryType, timestamp.UnixNano(), []byte(data))
|
|
if err != nil {
|
|
log.Printf("skip %s: %v", entryType, err)
|
|
continue
|
|
}
|
|
for _, event := range events {
|
|
eventCount++
|
|
key := event.Service + "\x00" + event.Kind + "\x00" + event.PatternSignature
|
|
if counts[key] == nil {
|
|
counts[key] = &patternCount{Service: event.Service, Kind: event.Kind, Pattern: event.PatternSignature}
|
|
}
|
|
counts[key].Count++
|
|
if *write {
|
|
if err := appendEvent(batch, event); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if eventCount%*batchSize == 0 {
|
|
flush()
|
|
batch, err = conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s.normalized_events", *database))
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
flush()
|
|
top := make([]*patternCount, 0, len(counts))
|
|
for _, item := range counts {
|
|
top = append(top, item)
|
|
}
|
|
sort.Slice(top, func(i, j int) bool { return top[i].Count > top[j].Count })
|
|
if len(top) > 20 {
|
|
top = top[:20]
|
|
}
|
|
fmt.Printf("raw_envelopes=%d normalized_events=%d unique_patterns=%d write=%t\n", rawCount, eventCount, len(top), *write)
|
|
for _, item := range top {
|
|
fmt.Printf("pattern_count=%d service=%s kind=%s pattern=%s\n", item.Count, item.Service, item.Kind, item.Pattern)
|
|
}
|
|
}
|
|
|
|
func createSchema(ctx context.Context, conn clickhouse.Conn, database 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`, database)
|
|
if err := conn.Exec(ctx, query); err != nil {
|
|
return err
|
|
}
|
|
// Existing developer tables predate the explicit model flags. These
|
|
// additive migrations preserve historical rows with safe zero defaults.
|
|
for _, statement := range []string{
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS issue_fingerprint String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_retry UInt8 DEFAULT 0", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_timeout UInt8 DEFAULT 0", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_cancellation UInt8 DEFAULT 0", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS method String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS host String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS user_agent String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS origin_host String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS referer_host String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS fetch_site String DEFAULT ''", database),
|
|
fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS fetch_mode String DEFAULT ''", database),
|
|
} {
|
|
if err := conn.Exec(ctx, statement); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func appendEvent(batch driver.Batch, event normalization.Event) error {
|
|
return batch.Append(
|
|
event.EventID, event.EntryType, event.EventIndex, time.Unix(0, event.IngestTimestamp), time.Unix(0, event.Timestamp),
|
|
event.Service, event.Release, event.Environment, event.Kind, event.Source, boolUint8(event.IsError),
|
|
event.StatusClass, event.StatusCode, event.DurationMS, event.Route, event.SpanName, event.MetricName, event.MetricType, event.MetricUnit,
|
|
event.MetricValue, event.MetricCount, event.MetricSum, event.MetricMin, event.MetricMax,
|
|
event.ExceptionType, event.MessageNormalized, event.StackSignature, event.TraceID, event.IssueFingerprint,
|
|
boolUint8(event.IsRetry), boolUint8(event.IsTimeout), boolUint8(event.IsCancellation), event.ExactSignature, event.PatternSignature,
|
|
uint8(1), event.Method, event.Host, event.UserAgent, event.OriginHost, event.RefererHost, event.FetchSite, event.FetchMode,
|
|
)
|
|
}
|
|
|
|
func boolUint8(value bool) uint8 {
|
|
if value {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|
|
func envOr(key, fallback string) string {
|
|
if value := os.Getenv(key); value != "" {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func parseWindow(since, until string) (time.Time, time.Time) {
|
|
parse := func(value, name string) time.Time {
|
|
if value == "" {
|
|
return time.Time{}
|
|
}
|
|
parsed, err := time.Parse(time.RFC3339Nano, value)
|
|
if err != nil {
|
|
log.Fatalf("invalid -%s %q: %v", name, value, err)
|
|
}
|
|
return parsed.UTC()
|
|
}
|
|
start, end := parse(since, "since"), parse(until, "until")
|
|
if !start.IsZero() && !end.IsZero() && !start.Before(end) {
|
|
log.Fatal("-since must be before -until")
|
|
}
|
|
return start, end
|
|
}
|