include db for gitea build

internal/alerts/load.go

@@ -0,0 +1,134 @@
+// 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
+}