go-otel/cmd/normalized-backfill/main.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"
	"git.tornberg.me/go-otel/internal/storage"
	"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 := storage.EnsureNormalizedSchema(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, storage.NormalizedInsertSQL(*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(&timestamp, &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 := storage.AppendNormalizedEvent(batch, event); err != nil {
					log.Fatal(err)
				}
				if eventCount%*batchSize == 0 {
					flush()
					batch, err = conn.PrepareBatch(ctx, storage.NormalizedInsertSQL(*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 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
}