go-otel/internal/alerts/load.go

// Package alerts turns the deterministic spike detector into notifications:
// it loads recent error events, runs features.DetectSpikes, keeps only
// actionable triage categories, suppresses repeats and posts what is left to
// a Telegram chat and/or a generic webhook.
package alerts

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"

	"git.tornberg.me/go-otel/internal/features"
	"git.tornberg.me/go-otel/internal/honeypot"
	"git.tornberg.me/go-otel/internal/tracectx"
	"git.tornberg.me/go-otel/internal/uaclass"
)

func chTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05.000000000") }

// DataSince returns the earliest normalized row, the input to the detector's
// warm-up guard. Zero means unknown (empty table or query failure).
func DataSince(ctx context.Context, conn driver.Conn, database string) time.Time {
	var t time.Time
	if err := conn.QueryRow(ctx, fmt.Sprintf(`SELECT min(timestamp) FROM %s.normalized_events`, database)).Scan(&t); err != nil {
		log.Printf("alerts: warm-up guard disabled: %v", err)
		return time.Time{}
	}
	if t.Year() < 2000 { // empty table
		return time.Time{}
	}
	return t
}

// LoadErrorEvents reads error events in [start, end) from normalized_events
// and, when the trace columns exist, resolves duplicate hops and real client
// IPs through each error's trace (see internal/tracectx).
func LoadErrorEvents(ctx context.Context, conn driver.Conn, database string, start, end time.Time) ([]features.Event, error) {
	// Trace columns exist only after the server has migrated the table.
	var traceCols uint64
	if err := conn.QueryRow(ctx, `SELECT count() FROM system.columns WHERE database = ? AND table = 'normalized_events' AND name IN ('span_id','parent_span_id','peer_ip','ua_class')`, database).Scan(&traceCols); err != nil {
		return nil, err
	}
	withTrace := traceCols == 4
	extra := "'' AS span_id, '' AS parent_span_id, '' AS peer_ip, '' AS ua_class, '' AS user_agent, '' AS trace_id"
	if withTrace {
		extra = "span_id, parent_span_id, peer_ip, ua_class, user_agent, trace_id"
	}
	rows, err := conn.Query(ctx, fmt.Sprintf(`SELECT timestamp, service, release, kind, source, status_code, span_name, route, exception_type, message_normalized, pattern_signature, is_timeout, is_cancellation, is_retry, %s
		FROM %s.normalized_events FINAL
		WHERE is_error = 1 AND timestamp >= toDateTime64('%s', 9, 'UTC') AND timestamp < toDateTime64('%s', 9, 'UTC')
		ORDER BY timestamp LIMIT 2000000`, extra, database, chTime(start), chTime(end)))
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []features.Event
	var refs []tracectx.Key
	var own []string // the span's own peer address, the fallback client
	for rows.Next() {
		e := features.Event{IsError: true}
		var status uint16
		var timeout, cancel, retry uint8
		var spanID, parentID, peer, class, ua, traceID string
		if err := rows.Scan(&e.Timestamp, &e.Service, &e.Release, &e.Kind, &e.Source, &status, &e.SpanName, &e.Route, &e.ExceptionType, &e.Message, &e.PatternSignature, &timeout, &cancel, &retry,
			&spanID, &parentID, &peer, &class, &ua, &traceID); err != nil {
			return nil, err
		}
		e.StatusCode = int(status)
		e.IsTimeout, e.IsCancellation, e.IsRetry = timeout == 1, cancel == 1, retry == 1
		e.TraceID, e.UAClass = traceID, class
		if e.UAClass == "" && ua != "" {
			e.UAClass = uaclass.Of(ua) // rows written before ua_class existed
		}
		out = append(out, e)
		refs = append(refs, tracectx.Key{TraceID: traceID, SpanID: spanID})
		own = append(own, peer)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if withTrace && len(out) > 0 {
		if err := resolve(ctx, conn, database, start, end, out, refs, own); err != nil {
			// Unresolved events are still usable; only the dedupe is lost.
			log.Printf("alerts: trace resolution skipped: %v", err)
		}
	}
	return out, nil
}

// resolve loads every span of the traces that contain an error (ancestors are
// usually not errors themselves) and copies the resolved client / duplicate /
// root-cause facts onto the error events.
func resolve(ctx context.Context, conn driver.Conn, database string, start, end time.Time, events []features.Event, refs []tracectx.Key, own []string) error {
	rows, err := conn.Query(ctx, fmt.Sprintf(`SELECT trace_id, span_id, parent_span_id, service, peer_ip, is_error, status_code
		FROM %[1]s.normalized_events
		WHERE kind = 'span' AND span_id != '' AND timestamp >= toDateTime64('%[2]s', 9, 'UTC') AND timestamp < toDateTime64('%[3]s', 9, 'UTC')
		  AND trace_id IN (SELECT trace_id FROM %[1]s.normalized_events WHERE kind = 'span' AND is_error = 1 AND trace_id != ''
		                   AND timestamp >= toDateTime64('%[2]s', 9, 'UTC') AND timestamp < toDateTime64('%[3]s', 9, 'UTC'))
		LIMIT 5000000`, database, chTime(start.Add(-time.Minute)), chTime(end.Add(time.Minute))))
	if err != nil {
		return err
	}
	defer rows.Close()
	var spans []tracectx.Span
	for rows.Next() {
		var s tracectx.Span
		var isErr uint8
		var status uint16
		if err := rows.Scan(&s.TraceID, &s.SpanID, &s.ParentID, &s.Service, &s.PeerIP, &isErr, &status); err != nil {
			return err
		}
		s.IsError, s.StatusCode, s.Edge = isErr == 1, int(status), tracectx.IsEdgeService(s.Service)
		spans = append(spans, s)
	}
	if err := rows.Err(); err != nil {
		return err
	}
	info := tracectx.Resolve(spans)
	for i, ref := range refs {
		in, ok := info[ref]
		if !ok {
			if !honeypot.Untrackable(own[i]) {
				events[i].ClientIP = own[i]
			}
			continue
		}
		events[i].ClientIP, events[i].IsDuplicate, events[i].IsRootCause = in.ClientIP, in.IsDuplicate, in.IsRootCause
		events[i].HasUpstream = in.UpstreamChildren > 0
	}
	return nil
}