// 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, issue_fingerprint, host, %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, &e.IssueFingerprint, &e.Host, &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 } failing := map[string]map[string]struct{}{} // trace -> services with an error span for _, sp := range spans { if sp.IsError && sp.Service != "" { if failing[sp.TraceID] == nil { failing[sp.TraceID] = map[string]struct{}{} } failing[sp.TraceID][sp.Service] = struct{}{} } } info := tracectx.Resolve(spans) for i, ref := range refs { for sv := range failing[ref.TraceID] { events[i].TraceServices = append(events[i].TraceServices, sv) } 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 }