go-otel/cmd/normalized-preview/main.go

// normalized-preview reads a small sample from ClickHouse and prints the
// normalized rows without creating tables or writing any data.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"net/url"
	"os"
	"strconv"
	"time"

	"git.tornberg.me/go-otel/internal/normalization"
	"github.com/ClickHouse/clickhouse-go/v2"
)

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", 10, "raw envelopes to preview")
	typeFilter := flag.String("type", "", "only preview one raw type: traces, logs, or metrics")
	httpMode := flag.Bool("http", false, "use ClickHouse HTTP instead of native protocol")
	summary := flag.Bool("summary", false, "aggregate by service, kind, and pattern instead of printing events")
	flag.Parse()
	if *httpMode {
		previewHTTP(*addr, *database, *user, *password, *limit, *typeFilter, *summary)
		return
	}

	conn, err := clickhouse.Open(&clickhouse.Options{Addr: []string{*addr}, Auth: clickhouse.Auth{Database: *database, Username: *user, Password: *password}})
	if err != nil {
		log.Fatalf("clickhouse open: %v", err)
	}
	defer conn.Close()
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	if err := conn.Ping(ctx); err != nil {
		log.Fatalf("clickhouse ping: %v", err)
	}
	if *typeFilter != "" && *typeFilter != "traces" && *typeFilter != "logs" && *typeFilter != "metrics" {
		log.Fatalf("invalid -type %q", *typeFilter)
	}
	whereSQL := ""
	if *typeFilter != "" {
		whereSQL = fmt.Sprintf(" WHERE type = '%s'", *typeFilter)
	}
	rows, err := conn.Query(ctx, fmt.Sprintf("SELECT timestamp, type, data FROM %s.entries%s ORDER BY timestamp DESC LIMIT ?", *database, whereSQL), *limit)
	if err != nil {
		log.Fatalf("query entries: %v", err)
	}
	defer rows.Close()
	encoder := json.NewEncoder(os.Stdout)
	groups := map[string]*group{}
	count := 0
	for rows.Next() {
		var timestamp time.Time
		var entryType, data string
		if err := rows.Scan(&timestamp, &entryType, &data); err != nil {
			log.Fatalf("scan entry: %v", err)
		}
		events, err := normalization.NormalizeEntry(entryType, timestamp.UnixNano(), []byte(data))
		if err != nil {
			log.Printf("skip %s at %s: %v", entryType, timestamp.Format(time.RFC3339), err)
			continue
		}
		for _, event := range events {
			if *summary {
				addGroup(groups, event)
			} else if err := encoder.Encode(event); err != nil {
				log.Fatal(err)
			}
			count++
		}
	}
	if err := rows.Err(); err != nil {
		log.Fatal(err)
	}
	if *summary {
		printGroups(encoder, groups)
	}
	log.Printf("previewed %d normalized events from %d raw envelopes", count, *limit)
}

func previewHTTP(addr, database, user, password string, limit int, typeFilter string, summary bool) {
	limitSQL := ""
	orderSQL := " ORDER BY timestamp DESC"
	whereSQL := ""
	if typeFilter != "" {
		if typeFilter != "traces" && typeFilter != "logs" && typeFilter != "metrics" {
			log.Fatalf("invalid -type %q", typeFilter)
		}
		whereSQL = fmt.Sprintf(" WHERE type = '%s'", typeFilter)
	}
	if limit > 0 {
		limitSQL = fmt.Sprintf(" LIMIT %d", limit)
	} else {
		// A full dry run must stream the table. Sorting every raw payload first
		// can consume more memory than the ClickHouse query limit allows.
		orderSQL = ""
	}
	query := fmt.Sprintf("SELECT toUnixTimestamp64Nano(timestamp) AS timestamp, type, data FROM %s.entries%s%s%s FORMAT JSONEachRow", database, whereSQL, orderSQL, limitSQL)
	req, err := http.NewRequest(http.MethodGet, "http://"+addr+"/?query="+url.QueryEscape(query), nil)
	if err != nil {
		log.Fatal(err)
	}
	req.SetBasicAuth(user, password)
	response, err := http.DefaultClient.Do(req)
	if err != nil {
		log.Fatalf("clickhouse HTTP request: %v", err)
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		body, _ := io.ReadAll(response.Body)
		log.Fatalf("clickhouse HTTP %s: %s", response.Status, body)
	}
	decoder := json.NewDecoder(response.Body)
	encoder := json.NewEncoder(os.Stdout)
	groups := map[string]*group{}
	rawCount, eventCount := 0, 0
	for {
		var row struct {
			Timestamp json.RawMessage `json:"timestamp"`
			Type      string          `json:"type"`
			Data      string          `json:"data"`
		}
		if err := decoder.Decode(&row); err == io.EOF {
			break
		} else if err != nil {
			log.Fatal(err)
		}
		rawCount++
		timestamp := parseTimestamp(row.Timestamp)
		events, err := normalization.NormalizeEntry(row.Type, timestamp, []byte(row.Data))
		if err != nil {
			log.Printf("skip %s: %v", row.Type, err)
			continue
		}
		for _, event := range events {
			if summary {
				addGroup(groups, event)
			} else if err := encoder.Encode(event); err != nil {
				log.Fatal(err)
			}
			eventCount++
		}
	}
	if summary {
		printGroups(encoder, groups)
	}
	log.Printf("previewed %d normalized events from %d raw envelopes", eventCount, rawCount)
}

type group struct {
	Service          string `json:"service"`
	Kind             string `json:"kind"`
	PatternSignature string `json:"patternSignature"`
	Count            int    `json:"count"`
}

func addGroup(groups map[string]*group, event normalization.Event) {
	key := event.Service + "\x00" + event.Kind + "\x00" + event.PatternSignature
	if groups[key] == nil {
		groups[key] = &group{Service: event.Service, Kind: event.Kind, PatternSignature: event.PatternSignature}
	}
	groups[key].Count++
}

func printGroups(encoder *json.Encoder, groups map[string]*group) {
	for _, item := range groups {
		if err := encoder.Encode(item); err != nil {
			log.Fatal(err)
		}
	}
}

func parseTimestamp(raw json.RawMessage) int64 {
	var text string
	if json.Unmarshal(raw, &text) == nil {
		value, _ := strconv.ParseInt(text, 10, 64)
		return value
	}
	var value int64
	_ = json.Unmarshal(raw, &value)
	return value
}

func envOr(key, fallback string) string {
	if value := os.Getenv(key); value != "" {
		return value
	}
	return fallback
}