diff --git a/.gitignore b/.gitignore index 949fc48..e63fd59 100644 --- a/.gitignore +++ b/.gitignore @@ -11,4 +11,4 @@ otel-agent.log .env # MaxMind GeoLite2 databases (large, licensed; baked into the image at build time) -/db/*.mmdb +#/db/*.mmdb diff --git a/alert-test b/alert-test new file mode 100755 index 0000000..555e088 Binary files /dev/null and b/alert-test differ diff --git a/build.sh b/build.sh index c7a785e..dd5dc30 100755 --- a/build.sh +++ b/build.sh @@ -3,6 +3,7 @@ set -euo pipefail RELEASE="${RELEASE:-$(git rev-parse --short HEAD)}" PUBLIC_SOURCEMAPS="${PUBLIC_SOURCEMAPS:-false}" +UPLOAD_SOURCEMAPS="${UPLOAD_SOURCEMAPS:-true}" echo "Building Needle 3 sidecar image..." docker build -t matst80/needle-sidecar:latest cmd/needle-sidecar @@ -15,21 +16,12 @@ docker build \ -t matst80/otel-debug:latest . docker push matst80/otel-debug:latest -# Only the app manifests. Not `-f k8s/`: clickhouse.yaml carries a CHANGE_ME -# password Secret that would overwrite the live one, and auth-dev.yaml is a -# partial AuthServer that must be applied by hand with -# `kubectl apply --server-side --field-manager=otel-dev-auth -f k8s/auth-dev.yaml`. -kubectl apply -f k8s/deployment.yaml -f k8s/service.yaml -f k8s/ingress.yaml -kubectl rollout restart deployment otel-debug-receiver -n monitoring -kubectl rollout status deployment/otel-debug-receiver -n monitoring -# Docker builds intentionally omit public source maps. If the public receiver -# URL is present, build the same release locally and upload maps to the running -# receiver. Maps persist on the otel-sourcemaps PVC and are reloaded on start; -# verify with: node scripts/dev-auth.mjs get /api/sourcemaps -if [[ -n "${OTEL_URL:-}" ]]; then - # Take the maps from the exact UI build that went into the image (same - # bundle hashes and positions as production), not from a second local build. +# Docker builds intentionally omit public source maps. Upload maps from the +# exact UI build that went into the image before rolling it out; otherwise RUM +# events can arrive with a release that the resolver cannot map. +if [[ "${UPLOAD_SOURCEMAPS}" == "true" ]]; then + : "${OTEL_URL:?set OTEL_URL (for example https://otel.k6n.net), or set UPLOAD_SOURCEMAPS=false}" maps_dir="$(mktemp -d)" trap 'rm -rf "$maps_dir"' EXIT docker build --target ui-builder \ @@ -42,10 +34,18 @@ if [[ -n "${OTEL_URL:-}" ]]; then while IFS= read -r -d '' map; do found=1 bundle="${map%.map}" - RELEASE="$RELEASE" OTEL_URL="$OTEL_URL" \ + RELEASE="${RELEASE}" OTEL_URL="${OTEL_URL}" \ ./scripts/upload-sourcemap.sh "$bundle" done < <(find "$maps_dir" -type f -name '*.js.map' -print0) [[ "$found" == 1 ]] || { echo "no source maps found in UI build" >&2; exit 1; } fi +# Only the app manifests. Not `-f k8s/`: clickhouse.yaml carries a CHANGE_ME +# password Secret that would overwrite the live one, and auth-dev.yaml is a +# partial AuthServer that must be applied by hand with +# `kubectl apply --server-side --field-manager=otel-dev-auth -f k8s/auth-dev.yaml`. +kubectl apply -f k8s/deployment.yaml -f k8s/service.yaml -f k8s/ingress.yaml +kubectl rollout restart deployment otel-debug-receiver -n monitoring +kubectl rollout status deployment/otel-debug-receiver -n monitoring + kubectl logs -f deployment/otel-debug-receiver -n monitoring diff --git a/cmd/alert-test/main.go b/cmd/alert-test/main.go new file mode 100644 index 0000000..0ab2b94 --- /dev/null +++ b/cmd/alert-test/main.go @@ -0,0 +1,43 @@ +// alert-test sends one clearly marked test alert through every channel +// configured in the environment (see internal/alerts.ConfigFromEnv), so a new +// Telegram token / chat id or webhook can be verified. It never prints secrets. +package main + +import ( + "context" + "fmt" + "os" + "time" + + "git.tornberg.me/go-otel/internal/alerts" + "git.tornberg.me/go-otel/internal/features" +) + +func main() { + cfg := alerts.ConfigFromEnv() + notifiers := cfg.Notifiers() + if len(notifiers) == 0 { + fmt.Fprintln(os.Stderr, "no channel configured (TELEGRAM_KEY + TELEGRAM_CHATID, or OTEL_ALERT_WEBHOOK_URL)") + os.Exit(2) + } + now := time.Now().UTC() + a := alerts.NewAlert(features.Spike{Service: "alert-test", WindowStart: now.Add(-5 * time.Minute), WindowEnd: now, ErrorCount: 42, BaselineMean: 1, Excess: 41, + Patterns: []features.PatternDelta{{Signature: "test", Count: 42, BaselineMean: 1, Excess: 41, Novel: true, Clients: 3, + Example: features.Event{Service: "alert-test", Kind: "span", StatusCode: 500, Route: "/api/example", Message: "this is a test alert from otel-debug"}}}}, now) + a.Test = true + failed := false + for _, n := range notifiers { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + err := n.Send(ctx, a) + cancel() + if err != nil { + fmt.Printf("%s: FAILED: %v\n", n.Name(), err) + failed = true + continue + } + fmt.Printf("%s: sent\n", n.Name()) + } + if failed { + os.Exit(1) + } +} diff --git a/cmd/bad-actors/main.go b/cmd/bad-actors/main.go index 031a6e1..69a34d4 100644 --- a/cmd/bad-actors/main.go +++ b/cmd/bad-actors/main.go @@ -17,14 +17,19 @@ import ( "github.com/ClickHouse/clickhouse-go/v2" "git.tornberg.me/go-otel/internal/honeypot" + "git.tornberg.me/go-otel/internal/tracectx" "git.tornberg.me/go-otel/internal/triage" + "git.tornberg.me/go-otel/internal/uaclass" ) type otlp struct { ResourceSpans []struct { ScopeSpans []struct { Spans []struct { - Attributes []struct { + TraceID string `json:"traceId"` + SpanID string `json:"spanId"` + ParentSpanID string `json:"parentSpanId"` + Attributes []struct { Key string `json:"key"` Value map[string]json.RawMessage `json:"value"` } `json:"attributes"` @@ -76,8 +81,15 @@ func main() { } defer rows.Close() - actors := map[string]*actor{} - spans := 0 + // Pass 1 collects every span; pass 2 resolves parents, because the outer + // nginx span (real client) and the inner loopback span are exported in + // different envelopes and arrive in either order. + type record struct { + key tracectx.Key + ip, path, status, ua string + } + var records []record + var all []tracectx.Span for rows.Next() { var data string if err := rows.Scan(&data); err != nil { @@ -99,25 +111,9 @@ func main() { } } ip := strings.TrimSpace(strings.SplitN(first(at, "http.request.header.x-real-ip", "http.request.header.x-forwarded-for", "client.address", "net.peer.ip"), ",", 2)[0]) - if ip == "" { - continue - } - spans++ path := strings.SplitN(first(at, "http.target", "url.path", "http.route"), "?", 2)[0] - a := actors[ip] - if a == nil { - a = &actor{IP: ip, Paths: map[string]struct{}{}, Statuses: map[string]int{}} - actors[ip] = a - } - a.Total++ - if a.UserAgent == "" { - a.UserAgent = first(at, "http.user_agent") - } - if triage.IsProbePath(path) { - a.Probes++ - a.Paths[path] = struct{}{} - a.Statuses[first(at, "http.status_code")]++ - } + all = append(all, tracectx.Span{TraceID: sp.TraceID, SpanID: sp.SpanID, ParentID: sp.ParentSpanID, Service: *service, PeerIP: ip, Edge: true}) + records = append(records, record{key: tracectx.Key{TraceID: sp.TraceID, SpanID: sp.SpanID}, ip: ip, path: path, status: first(at, "http.status_code"), ua: first(at, "http.user_agent")}) } } } @@ -126,6 +122,39 @@ func main() { log.Fatal(err) } + resolved := tracectx.Resolve(all) + actors := map[string]*actor{} + spans, recovered, recoveredProbes := 0, 0, 0 + for _, r := range records { + ip := r.ip + if in, ok := resolved[r.key]; ok && in.ClientIP != "" && in.ClientHops > 0 && internal(ip) { + ip = in.ClientIP // hop behind a proxy: use the real client from an ancestor + recovered++ + if triage.IsProbePath(r.path) { + recoveredProbes++ + } + } + if ip == "" { + continue + } + spans++ + a := actors[ip] + if a == nil { + a = &actor{IP: ip, Paths: map[string]struct{}{}, Statuses: map[string]int{}} + actors[ip] = a + } + a.Total++ + if a.UserAgent == "" { + a.UserAgent = r.ua + } + if triage.IsProbePath(r.path) { + a.Probes++ + a.Paths[r.path] = struct{}{} + a.Statuses[r.status]++ + } + } + fmt.Printf("resolved %d hop spans to their real client via the parent span (%d of them probes)\n", recovered, recoveredProbes) + var list []*actor totalProbes, blockableProbes, blockable, skippedInternal, skippedInternalProbes := 0, 0, 0, 0, 0 for _, a := range actors { @@ -149,12 +178,12 @@ func main() { fmt.Printf("%d spans with an IP, %d distinct IPs, %d probe requests\n", spans, len(actors), totalProbes) fmt.Printf("blockable (public; >=%d probes/>=%d paths/>=%.0f%% probes, or >=%d probes over >=%d paths): %d IPs covering %d probes (%.0f%%)\n", *minProbes, *minPaths, *minShare*100, *broadProbes, *broadPaths, blockable, blockableProbes, pct(blockableProbes, totalProbes)) fmt.Printf("internal/loopback (never blocked): %d IPs, %d probes (%.0f%%) - real client is hidden behind a proxy hop\n\n", skippedInternal, skippedInternalProbes, pct(skippedInternalProbes, totalProbes)) - fmt.Printf("%-40s %6s %6s %6s %s\n", "ip", "probes", "total", "paths", "verdict / user-agent / statuses") + fmt.Printf("%-40s %6s %6s %6s %-14s %s\n", "ip", "probes", "total", "paths", "class", "verdict / user-agent / statuses") for i, a := range list { if i >= *top { break } - fmt.Printf("%-40s %6d %6d %6d %.60s %v\n", a.IP, a.Probes, a.Total, len(a.Paths), a.UserAgent, a.Statuses) + fmt.Printf("%-40s %6d %6d %6d %-14s %.60s %v\n", a.IP, a.Probes, a.Total, len(a.Paths), uaclass.Of(strings.TrimPrefix(strings.TrimPrefix(a.UserAgent, "[BLOCK] "), "[INTERNAL] ")), a.UserAgent, a.Statuses) } } diff --git a/cmd/server/main.go b/cmd/server/main.go index 1e812d9..d98abd2 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -9,6 +9,7 @@ import ( "os" "strings" + "git.tornberg.me/go-otel/internal/alerts" "git.tornberg.me/go-otel/internal/auth" "git.tornberg.me/go-otel/internal/collector" "git.tornberg.me/go-otel/internal/geoip" @@ -243,6 +244,12 @@ func main() { if err != nil { log.Printf("ip blocker disabled: %v", err) } + // Optional spike alerts (Telegram / webhook): off unless a channel is + // configured or OTEL_ALERT_DRY_RUN=true. + var alertRunner *alerts.Runner + if chStore, ok := store.(*storage.ClickHouseStore); ok { + alertRunner = alerts.Start(context.Background(), alerts.ConfigFromEnv(), chStore.Conn(), chStore.Database()) + } // Update collector with wsServer coll.SetBroadcaster(wsServer) // Optional MaxMind databases for the UI's IP details dialog. @@ -309,6 +316,10 @@ func main() { uiMux.HandleFunc("GET /api/auth/config", oidcAuth.HandleConfig) uiMux.Handle("GET /api/ip/{ip}", oidcAuth.RequireAuth(http.HandlerFunc(geo.Handler))) uiMux.Handle("/api/honeypot/ips", oidcAuth.RequireAuth(http.HandlerFunc(honeypotTracker.HandleIPsForUser))) + if alertRunner != nil { + uiMux.Handle("GET /api/alerts", oidcAuth.RequireAuth(http.HandlerFunc(alertRunner.HandleStatus))) + uiMux.Handle("POST /api/alerts/test", oidcAuth.RequireAuth(http.HandlerFunc(alertRunner.HandleTest))) + } if ipBlocker != nil { uiMux.Handle("GET /api/blocker/status", oidcAuth.RequireAuth(http.HandlerFunc(ipBlocker.HandleStatus))) } diff --git a/cmd/spike-alerts/main.go b/cmd/spike-alerts/main.go index 56e1851..bf6befd 100644 --- a/cmd/spike-alerts/main.go +++ b/cmd/spike-alerts/main.go @@ -17,8 +17,10 @@ import ( "github.com/ClickHouse/clickhouse-go/v2" + "git.tornberg.me/go-otel/internal/alerts" "git.tornberg.me/go-otel/internal/features" "git.tornberg.me/go-otel/internal/triage" + "git.tornberg.me/go-otel/internal/uaclass" ) func main() { @@ -51,18 +53,28 @@ func main() { } cfg := features.SpikeConfig{Window: *window, BaselineWindows: *baselineWindows, Multiple: *multiple, MinExcess: *minExcess} - events := load(*addr, *database, *user, *password, start, end) - cfg.DataSince = dataSince(*addr, *database, *user, *password) + 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() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + events, err := alerts.LoadErrorEvents(ctx, conn, *database, start, end) + if err != nil { + log.Fatal(err) + } + cfg.DataSince = alerts.DataSince(ctx, conn, *database) log.Printf("loaded %d error events", len(events)) cache := map[string]classification{} first := start.Add(time.Duration(*baselineWindows) * *window) suppress := features.NewSuppressor(*cooldown) muted := map[triage.Category]int{} - alerts := 0 + fired := 0 for at := first.Add(*window); !at.After(end); at = at.Add(*step) { - for _, spike := range suppress.Filter(triageFilter(features.DetectSpikes(events, at, cfg), *showMuted, muted)) { - alerts++ + for _, spike := range suppress.Filter(alerts.FilterActionable(features.DetectSpikes(events, at, cfg), *showMuted, muted)) { + fired++ var cls []classification for _, p := range spike.Patterns { c := classification{} @@ -86,7 +98,7 @@ func main() { for c, n := range muted { log.Printf("muted %d pattern-window(s) as %s", n, c) } - log.Printf("%d alert(s) between %s and %s", alerts, first.Format(time.RFC3339), end.Format(time.RFC3339)) + log.Printf("%d alert(s) between %s and %s", fired, first.Format(time.RFC3339), end.Format(time.RFC3339)) } type classification struct { @@ -96,35 +108,6 @@ type classification struct { Err string `json:"error,omitempty"` } -func triageOf(e features.Event) triage.Result { - return triage.Classify(triage.Input{Service: e.Service, Kind: e.Kind, Status: e.StatusCode, Route: e.Route, Title: e.SpanName, Message: e.Message, - IsTimeout: e.IsTimeout, IsCancellation: e.IsCancellation, IsRetry: e.IsRetry}) -} - -// triageFilter keeps only actionable patterns (unless showMuted), recomputing -// each spike's excess so a probe storm cannot page anyone by itself. -func triageFilter(spikes []features.Spike, showMuted bool, muted map[triage.Category]int) []features.Spike { - var out []features.Spike - for _, sp := range spikes { - var kept []features.PatternDelta - sp.Excess = 0 - for _, p := range sp.Patterns { - c := triageOf(p.Example).Category - if !showMuted && !c.Actionable() { - muted[c]++ - continue - } - kept = append(kept, p) - sp.Excess += p.Excess - } - if len(kept) > 0 { - sp.Patterns = kept - out = append(out, sp) - } - } - return out -} - func classify(base, text string) classification { body, _ := json.Marshal(map[string]string{"text": text}) resp, err := (&http.Client{Timeout: 60 * time.Second}).Post(strings.TrimRight(base, "/")+"/classify", "application/json", bytes.NewReader(body)) @@ -163,47 +146,19 @@ func printSpike(s features.Spike, cls []classification) { if p.Novel { tag = "NEW" } - what := string(triageOf(p.Example).Category) + what := string(alerts.TriageOf(p.Example).Category) if cls[i].Action != "" { what += fmt.Sprintf(" [needle:%s %.2f]", cls[i].Action, cls[i].Confidence) } - fmt.Printf(" %-6s x%-4d base %-6.1f %-24s %s\n", tag, p.Count, p.BaselineMean, what, needleText(p.Example)) - } -} - -func load(addr, database, user, password string, start, end time.Time) []features.Event { - ctx, cancel := context.WithTimeout(context.Background(), 5*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() - // Errors only, and only columns present in every deployed schema version. - 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 - 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`, database, chTime(start), chTime(end))) - if err != nil { - log.Fatal(err) - } - defer rows.Close() - var out []features.Event - for rows.Next() { - e := features.Event{IsError: true} - var status uint16 - var timeout, cancel, retry uint8 - 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); err != nil { - log.Fatal(err) + clients := "" + if p.Clients > 0 { + clients = fmt.Sprintf(" [%d client%s]", p.Clients, map[bool]string{true: "", false: "s"}[p.Clients == 1]) } - e.StatusCode = int(status) - e.IsTimeout, e.IsCancellation, e.IsRetry = timeout == 1, cancel == 1, retry == 1 - out = append(out, e) + if p.Example.UAClass != "" && uaclass.IsBot(p.Example.UAClass) { + clients += " [" + p.Example.UAClass + "]" + } + fmt.Printf(" %-6s x%-4d base %-6.1f %-24s %s%s\n", tag, p.Count, p.BaselineMean, what, needleText(p.Example), clients) } - if err := rows.Err(); err != nil { - log.Fatal(err) - } - return out } func chTime(t time.Time) string { return t.UTC().Format("2006-01-02 15:04:05.000000000") } @@ -213,23 +168,3 @@ func envOr(k, f string) string { } return f } - -// dataSince returns the earliest normalized row, used as the warm-up guard. -func dataSince(addr, database, user, password string) time.Time { - ctx, cancel := context.WithTimeout(context.Background(), 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() - 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("warm-up guard disabled: %v", err) - return time.Time{} - } - if t.Year() < 2000 { // empty table - return time.Time{} - } - return t -} diff --git a/cmd/window-features/main.go b/cmd/window-features/main.go index 19a8c8a..347233e 100644 --- a/cmd/window-features/main.go +++ b/cmd/window-features/main.go @@ -31,6 +31,12 @@ type event struct { IsCancellation uint8 `json:"isCancellation"` Message string `json:"messageNormalized"` SpanName string `json:"spanName"` + // Optional trace-resolved fields (internal/tracectx); absent in older exports. + ClientIP string `json:"clientIp"` + UAClass string `json:"uaClass"` + IsDuplicate uint8 `json:"isDuplicate"` + IsRootCause uint8 `json:"isRootCause"` + HasUpstream uint8 `json:"hasUpstream"` } func main() { @@ -95,6 +101,7 @@ func readEvents(path string) []features.Event { IsError: item.IsError != 0, StatusClass: item.StatusClass, DurationMS: item.DurationMS, TraceID: item.TraceID, PatternSignature: item.PatternSignature, IssueFingerprint: item.IssueFingerprint, IsRetry: item.IsRetry != 0, IsTimeout: item.IsTimeout != 0, IsCancellation: item.IsCancellation != 0, Message: item.Message, SpanName: item.SpanName, + ClientIP: item.ClientIP, UAClass: item.UAClass, IsDuplicate: item.IsDuplicate != 0, IsRootCause: item.IsRootCause != 0, HasUpstream: item.HasUpstream != 0, }) } if err := scanner.Err(); err != nil { diff --git a/db/GeoLite2-ASN.mmdb b/db/GeoLite2-ASN.mmdb new file mode 100644 index 0000000..28d87e7 Binary files /dev/null and b/db/GeoLite2-ASN.mmdb differ diff --git a/db/GeoLite2-City.mmdb b/db/GeoLite2-City.mmdb new file mode 100644 index 0000000..d2c5e94 Binary files /dev/null and b/db/GeoLite2-City.mmdb differ diff --git a/docs/STATUS.md b/docs/STATUS.md index a923b8f..3c4494a 100644 --- a/docs/STATUS.md +++ b/docs/STATUS.md @@ -120,6 +120,29 @@ Categories: `scanner_probe`, `dependency_unreachable`, `client_credential_issue` `uaClass` on issue samples and endpoint stats. Issues UI: add-filter offers `client: X` and `hide client: X`. Not yet a triage/spike feature. +- Trace context (`internal/tracectx`): one request is several spans in several + envelopes (nginx outer span on :442 with the real client, nginx inner span on + :8181 from 127.0.0.1, then the app span), so the ingest-time same-envelope + lookup in `normalizeTraces` almost never finds the parent (7 of 1154 nginx + spans measured). `tracectx.Resolve` works on a loaded span set instead and is + order independent. It yields `ClientIP` (first public peer up the chain), + `IsDuplicate` (repeats an ancestor's failure: same status code), `IsRootCause`, + child / upstream-child counts, `ParentMissing`. + - `bad-actors` resolves loopback hops to the real client (839 hop spans / + 583 probes recovered in a 3h sample). + - `spike-alerts` loads the spans of every trace with an error, drops duplicate + hops from the counts and reports distinct clients per pattern; it falls back + when the trace columns are absent. Historical rows have empty `span_id`, so it + only helps for data written since the trace columns shipped (~08:09 UTC 2026-09-20). + - `features.Row` gained `LogicalErrors`, `DuplicateErrors`, `RootCauseErrors`, + `DistinctClients`, `TopClientShare`, `BotErrorShare` (additive). + `window-features` reads optional `clientIp/uaClass/isDuplicate/isRootCause/hasUpstream`. + - Trace view in the UI shows the resolved client (`lib/trace-client.ts`). + - Not done: the collector still records the hop IP on issue samples at ingest; + span name `GET /` for orchestrator spans is unchanged (changing it alters every + orchestrator pattern signature); `HasUpstream` is loaded but not yet a triage + rule; trace propagation is missing in cart/checkout/profile/scheduling (roots only). + ## Commands ```bash diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go new file mode 100644 index 0000000..009ca1d --- /dev/null +++ b/internal/alerts/alerts_test.go @@ -0,0 +1,130 @@ +package alerts + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "git.tornberg.me/go-otel/internal/features" +) + +func burst(sig string, n int, at time.Time, status int, route string) []features.Event { + var out []features.Event + for i := 0; i < n; i++ { + out = append(out, features.Event{Timestamp: at.Add(time.Duration(i) * time.Second), Service: "api", Kind: "span", IsError: true, + PatternSignature: sig, StatusCode: status, Route: route, ClientIP: "203.0.113.7"}) + } + return out +} + +type fake struct { + mu sync.Mutex + sent []Alert + err error +} + +func (f *fake) Name() string { return "fake" } +func (f *fake) Send(_ context.Context, a Alert) error { + f.mu.Lock() + defer f.mu.Unlock() + f.sent = append(f.sent, a) + return f.err +} + +func newRunner(events []features.Event, n Notifier) *Runner { + cfg := Config{Interval: time.Minute, Window: 5 * time.Minute, BaselineWindows: 12, Cooldown: 30 * time.Minute, MinExcess: 5, Multiple: 3, Lag: 0} + return &Runner{ + cfg: cfg, notifiers: []Notifier{n}, sup: features.NewSuppressor(cfg.Cooldown), + load: func(context.Context, time.Time, time.Time) ([]features.Event, error) { return events, nil }, + dataSince: func(context.Context) time.Time { return time.Time{} }, + } +} + +func TestRunnerPrimesThenAlertsOnceAndFiltersProbes(t *testing.T) { + now := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + cur := now.Add(-2 * time.Minute) + f := &fake{} + // a real defect burst and a probe burst in the same window + events := append(burst("defect", 20, cur, 500, "/api/orders"), burst("probe", 50, cur, 404, "/.env")...) + r := newRunner(events, f) + + r.Tick(context.Background(), now) // first tick primes: nothing is sent + if len(f.sent) != 0 { + t.Fatalf("primed tick sent %d alerts", len(f.sent)) + } + // the same incident keeps going: suppressed by the primed cooldown + r.Tick(context.Background(), now.Add(time.Minute)) + if len(f.sent) != 0 { + t.Fatalf("ongoing incident re-alerted: %d", len(f.sent)) + } + // a new, different failure appears + r.load = func(context.Context, time.Time, time.Time) ([]features.Event, error) { + return append(events, burst("newdefect", 15, now.Add(time.Minute), 503, "/api/pay")...), nil + } + r.Tick(context.Background(), now.Add(4*time.Minute)) + if len(f.sent) != 1 { + t.Fatalf("want 1 alert, got %d", len(f.sent)) + } + txt := f.sent[0].Text("https://otel.example") + if strings.Contains(txt, "/.env") { + t.Fatalf("probe leaked into alert: %s", txt) + } + if !strings.Contains(txt, "/api/pay") || !strings.Contains(txt, "1 client") || !strings.HasSuffix(txt, "https://otel.example/issues") { + t.Fatalf("unexpected text: %s", txt) + } + if got := r.Status(); len(got.Alerts) != 1 || got.Channels[0] != "fake" { + t.Fatalf("status = %+v", got) + } +} + +func TestTelegramSendsAndNeverLeaksToken(t *testing.T) { + var gotPath string + var body map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + _ = json.NewDecoder(r.Body).Decode(&body) + w.WriteHeader(200) + })) + tg := &Telegram{Token: "SECRETTOKEN", ChatID: "42", ThreadID: "7", API: srv.URL} + a := NewAlert(features.Spike{Service: "s", ErrorCount: 9}, time.Now()) + if err := tg.Send(context.Background(), a); err != nil { + t.Fatal(err) + } + if gotPath != "/botSECRETTOKEN/sendMessage" || body["chat_id"] != "42" || body["message_thread_id"] != float64(7) { + t.Fatalf("request wrong: %s %+v", gotPath, body) + } + srv.Close() // connection refused: the transport error contains the URL + err := tg.Send(context.Background(), a) + if err == nil || strings.Contains(err.Error(), "SECRETTOKEN") { + t.Fatalf("token leaked or no error: %v", err) + } + bad := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"description":"bad token SECRETTOKEN"}`, 401) + })) + defer bad.Close() + err = (&Telegram{Token: "SECRETTOKEN", ChatID: "1", API: bad.URL}).Send(context.Background(), a) + if err == nil || strings.Contains(err.Error(), "SECRETTOKEN") || !strings.Contains(err.Error(), "401") { + t.Fatalf("error not scrubbed: %v", err) + } +} + +func TestTelegramTruncates(t *testing.T) { + var body map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _ = json.NewDecoder(r.Body).Decode(&body) })) + defer srv.Close() + spike := features.Spike{Service: "s"} + for i := 0; i < 200; i++ { + spike.Patterns = append(spike.Patterns, features.PatternDelta{Count: 1, Example: features.Event{Message: strings.Repeat("x", 100)}}) + } + if err := (&Telegram{Token: "t", ChatID: "1", API: srv.URL}).Send(context.Background(), NewAlert(spike, time.Now())); err != nil { + t.Fatal(err) + } + if n := len([]rune(body["text"].(string))); n > 4096 { + t.Fatalf("message too long: %d", n) + } +} diff --git a/internal/alerts/filter.go b/internal/alerts/filter.go new file mode 100644 index 0000000..3b2c5cd --- /dev/null +++ b/internal/alerts/filter.go @@ -0,0 +1,39 @@ +package alerts + +import ( + "git.tornberg.me/go-otel/internal/features" + "git.tornberg.me/go-otel/internal/triage" +) + +// TriageOf classifies a pattern's example event with the deterministic rules. +func TriageOf(e features.Event) triage.Result { + return triage.Classify(triage.Input{Service: e.Service, Kind: e.Kind, Status: e.StatusCode, Route: e.Route, Title: e.SpanName, Message: e.Message, + IsTimeout: e.IsTimeout, IsCancellation: e.IsCancellation, IsRetry: e.IsRetry}) +} + +// FilterActionable keeps only patterns whose category should page a human +// (unless keepMuted), recomputing each spike's excess so a probe storm cannot +// alert by itself. muted, when non-nil, counts what was dropped per category. +func FilterActionable(spikes []features.Spike, keepMuted bool, muted map[triage.Category]int) []features.Spike { + var out []features.Spike + for _, sp := range spikes { + var kept []features.PatternDelta + sp.Excess = 0 + for _, p := range sp.Patterns { + c := TriageOf(p.Example).Category + if !keepMuted && !c.Actionable() { + if muted != nil { + muted[c]++ + } + continue + } + kept = append(kept, p) + sp.Excess += p.Excess + } + if len(kept) > 0 { + sp.Patterns = kept + out = append(out, sp) + } + } + return out +} diff --git a/internal/alerts/load.go b/internal/alerts/load.go new file mode 100644 index 0000000..1d99e0b --- /dev/null +++ b/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 +} diff --git a/internal/alerts/notify.go b/internal/alerts/notify.go new file mode 100644 index 0000000..3030b1b --- /dev/null +++ b/internal/alerts/notify.go @@ -0,0 +1,235 @@ +package alerts + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "git.tornberg.me/go-otel/internal/features" + "git.tornberg.me/go-otel/internal/uaclass" +) + +// Alert is one notification: a spike reduced to its actionable patterns. +type Alert struct { + ID string `json:"id"` + At time.Time `json:"at"` + Spike features.Spike `json:"spike"` + Patterns []PatternReport `json:"patterns"` + Test bool `json:"test,omitempty"` +} + +// PatternReport is a spike pattern with its triage verdict, ready to show. +type PatternReport struct { + Category string `json:"category"` + Rule string `json:"rule"` + New bool `json:"new"` + Count int `json:"count"` + Baseline float64 `json:"baseline"` + Clients int `json:"clients,omitempty"` + UAClass string `json:"uaClass,omitempty"` + What string `json:"what"` +} + +// Notifier delivers an alert somewhere. Implementations must not put secrets +// in returned errors. +type Notifier interface { + Name() string + Send(ctx context.Context, a Alert) error +} + +// NewAlert builds an Alert from a spike, classifying each pattern. +func NewAlert(spike features.Spike, at time.Time) Alert { + a := Alert{At: at, Spike: spike, ID: alertID(spike, at)} + for _, p := range spike.Patterns { + res := TriageOf(p.Example) + r := PatternReport{Category: string(res.Category), Rule: res.Rule, New: p.Novel, Count: p.Count, Baseline: p.BaselineMean, Clients: p.Clients, What: describe(p.Example)} + if p.Example.UAClass != "" && uaclass.IsBot(p.Example.UAClass) { + r.UAClass = p.Example.UAClass + } + a.Patterns = append(a.Patterns, r) + } + return a +} + +func alertID(s features.Spike, at time.Time) string { + return strconv.FormatInt(at.Unix(), 36) + "-" + strings.ReplaceAll(s.Service, " ", "_") +} + +func describe(e features.Event) string { + parts := []string{} + if e.StatusCode != 0 { + parts = append(parts, "HTTP "+strconv.Itoa(e.StatusCode)) + } + switch { + case e.Route != "": + parts = append(parts, e.Route) + case e.SpanName != "": + parts = append(parts, e.SpanName) + } + if e.ExceptionType != "" { + parts = append(parts, e.ExceptionType) + } + if e.Message != "" && e.Message != e.SpanName { + parts = append(parts, truncate(e.Message, 120)) + } + if len(parts) == 0 { + return "(no detail)" + } + return strings.Join(parts, " | ") +} + +func truncate(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n-1]) + "…" +} + +// Text renders a plain-text message (no markup, so nothing needs escaping). +func (a Alert) Text(baseURL string) string { + var b strings.Builder + prefix := "ALERT" + if a.Test { + prefix = "TEST" + } + fmt.Fprintf(&b, "%s %s: %d errors in %s (baseline %.1f/window, excess %.0f)\n", + prefix, a.Spike.Service, a.Spike.ErrorCount, a.Spike.WindowEnd.Sub(a.Spike.WindowStart).Round(time.Minute), a.Spike.BaselineMean, a.Spike.Excess) + for _, p := range a.Patterns { + tag := "rising" + if p.New { + tag = "NEW" + } + fmt.Fprintf(&b, "\n%s x%d (base %.1f) [%s]\n%s", tag, p.Count, p.Baseline, p.Category, p.What) + var extra []string + if p.Clients > 0 { + extra = append(extra, fmt.Sprintf("%d client%s", p.Clients, plural(p.Clients))) + } + if p.UAClass != "" { + extra = append(extra, p.UAClass) + } + if len(extra) > 0 { + fmt.Fprintf(&b, "\n(%s)", strings.Join(extra, ", ")) + } + b.WriteString("\n") + } + if baseURL != "" { + fmt.Fprintf(&b, "\n%s/issues", strings.TrimRight(baseURL, "/")) + } + return strings.TrimSpace(b.String()) +} + +func plural(n int) string { + if n == 1 { + return "" + } + return "s" +} + +// Telegram posts via the Bot API sendMessage method. +type Telegram struct { + Token string + ChatID string + ThreadID string // optional forum topic + BaseURL string // link appended to messages + API string // override for tests; default https://api.telegram.org + Client *http.Client +} + +func (t *Telegram) Name() string { return "telegram" } + +// telegram limits a message to 4096 characters. +const telegramMax = 4000 + +func (t *Telegram) Send(ctx context.Context, a Alert) error { + api := t.API + if api == "" { + api = "https://api.telegram.org" + } + body := map[string]any{"chat_id": t.ChatID, "text": truncate(a.Text(t.BaseURL), telegramMax), "disable_web_page_preview": true} + if t.ThreadID != "" { + if id, err := strconv.Atoi(t.ThreadID); err == nil { + body["message_thread_id"] = id + } + } + return postJSON(ctx, t.client(), api+"/bot"+t.Token+"/sendMessage", nil, body, t.Token) +} + +func (t *Telegram) client() *http.Client { + if t.Client != nil { + return t.Client + } + return &http.Client{Timeout: 15 * time.Second} +} + +// Webhook posts {"text": ..., "alert": ...} to any URL, optionally with a +// shared secret in the X-Alert-Secret header. +type Webhook struct { + URL string + Secret string + BaseURL string + Client *http.Client +} + +func (w *Webhook) Name() string { return "webhook" } + +func (w *Webhook) Send(ctx context.Context, a Alert) error { + client := w.Client + if client == nil { + client = &http.Client{Timeout: 15 * time.Second} + } + headers := map[string]string{} + if w.Secret != "" { + headers["X-Alert-Secret"] = w.Secret + } + return postJSON(ctx, client, w.URL, headers, map[string]any{"text": a.Text(w.BaseURL), "alert": a}, w.URL, w.Secret) +} + +// postJSON posts a JSON body. Errors are rebuilt without the request URL and +// scrubbed of every secret, because both the Telegram token and webhook URLs +// commonly carry credentials. +func postJSON(ctx context.Context, client *http.Client, target string, headers map[string]string, body any, secrets ...string) error { + raw, err := json.Marshal(body) + if err != nil { + return err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, bytes.NewReader(raw)) + if err != nil { + return errors.New("invalid request") + } + req.Header.Set("Content-Type", "application/json") + for k, v := range headers { + req.Header.Set(k, v) + } + resp, err := client.Do(req) + if err != nil { + var ue *url.Error + if errors.As(err, &ue) { + err = ue.Err // drop the URL the transport put in the message + } + return errors.New(scrub(err.Error(), secrets)) + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + detail, _ := io.ReadAll(io.LimitReader(resp.Body, 300)) + return fmt.Errorf("HTTP %d: %s", resp.StatusCode, scrub(strings.TrimSpace(string(detail)), secrets)) + } + return nil +} + +func scrub(s string, secrets []string) string { + for _, sec := range secrets { + if sec != "" { + s = strings.ReplaceAll(s, sec, "") + } + } + return s +} diff --git a/internal/alerts/runner.go b/internal/alerts/runner.go new file mode 100644 index 0000000..e8180db --- /dev/null +++ b/internal/alerts/runner.go @@ -0,0 +1,281 @@ +package alerts + +import ( + "context" + "encoding/json" + "log" + "net/http" + "os" + "strconv" + "sync" + "time" + + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" + + "git.tornberg.me/go-otel/internal/features" +) + +// Config is read from OTEL_ALERT_* environment variables. The runner starts +// when at least one notifier is configured, or when OTEL_ALERT_DRY_RUN=true +// (evaluate and record, send nothing). +type Config struct { + DryRun bool + Interval time.Duration // how often the detector runs + Window time.Duration // observation window + BaselineWindows int + Cooldown time.Duration // suppress repeats of a pattern unless it doubles + MinExcess int + Multiple float64 + Lag time.Duration // ingest settle time: windows end this long before now + BaseURL string // UI link appended to messages, e.g. https://otel.k6n.net + + TelegramToken, TelegramChatID, TelegramThreadID string + WebhookURL, WebhookSecret string +} + +func ConfigFromEnv() Config { + return Config{ + DryRun: os.Getenv("OTEL_ALERT_DRY_RUN") == "true", + Interval: envDuration("OTEL_ALERT_INTERVAL", time.Minute), + Window: envDuration("OTEL_ALERT_WINDOW", 5*time.Minute), + BaselineWindows: envInt("OTEL_ALERT_BASELINE_WINDOWS", 36), + Cooldown: envDuration("OTEL_ALERT_COOLDOWN", 30*time.Minute), + MinExcess: envInt("OTEL_ALERT_MIN_EXCESS", 10), + Multiple: 3, + Lag: envDuration("OTEL_ALERT_LAG", 30*time.Second), + BaseURL: os.Getenv("OTEL_ALERT_BASE_URL"), + + TelegramToken: firstEnv("OTEL_ALERT_TELEGRAM_TOKEN", "TELEGRAM_KEY"), + TelegramChatID: firstEnv("OTEL_ALERT_TELEGRAM_CHAT_ID", "TELEGRAM_CHATID"), + TelegramThreadID: os.Getenv("OTEL_ALERT_TELEGRAM_THREAD_ID"), + WebhookURL: os.Getenv("OTEL_ALERT_WEBHOOK_URL"), + WebhookSecret: os.Getenv("OTEL_ALERT_WEBHOOK_SECRET"), + } +} + +// firstEnv returns the first non-empty variable; the plain TELEGRAM_* names are +// what the repo's .env uses, the OTEL_ALERT_* ones are for the deployment. +func firstEnv(keys ...string) string { + for _, k := range keys { + if v := os.Getenv(k); v != "" { + return v + } + } + return "" +} + +func envDuration(k string, def time.Duration) time.Duration { + if d, err := time.ParseDuration(os.Getenv(k)); err == nil && d > 0 { + return d + } + return def +} +func envInt(k string, def int) int { + if n, err := strconv.Atoi(os.Getenv(k)); err == nil && n > 0 { + return n + } + return def +} + +// Notifiers builds the configured delivery channels. +func (c Config) Notifiers() []Notifier { + var out []Notifier + if c.TelegramToken != "" && c.TelegramChatID != "" { + out = append(out, &Telegram{Token: c.TelegramToken, ChatID: c.TelegramChatID, ThreadID: c.TelegramThreadID, BaseURL: c.BaseURL}) + } + if c.WebhookURL != "" { + out = append(out, &Webhook{URL: c.WebhookURL, Secret: c.WebhookSecret, BaseURL: c.BaseURL}) + } + return out +} + +// Loader returns error events in [start, end); it is a field so tests can +// feed events without ClickHouse. +type Loader func(ctx context.Context, start, end time.Time) ([]features.Event, error) + +type Runner struct { + cfg Config + load Loader + dataSince func(ctx context.Context) time.Time + notifiers []Notifier + sup *features.Suppressor + + mu sync.Mutex + primed bool + recent []Alert // newest last + lastRun time.Time + lastError string + since time.Time + sinceAt time.Time +} + +const maxRecent = 50 + +// Start launches the runner against ClickHouse and returns it, or nil when +// alerting is not configured. +func Start(ctx context.Context, cfg Config, conn driver.Conn, database string) *Runner { + notifiers := cfg.Notifiers() + if len(notifiers) == 0 && !cfg.DryRun { + log.Printf("alerts: disabled (set OTEL_ALERT_TELEGRAM_TOKEN + OTEL_ALERT_TELEGRAM_CHAT_ID and/or OTEL_ALERT_WEBHOOK_URL)") + return nil + } + r := &Runner{ + cfg: cfg, + notifiers: notifiers, + sup: features.NewSuppressor(cfg.Cooldown), + load: func(ctx context.Context, start, end time.Time) ([]features.Event, error) { + return LoadErrorEvents(ctx, conn, database, start, end) + }, + dataSince: func(ctx context.Context) time.Time { return DataSince(ctx, conn, database) }, + } + names := make([]string, 0, len(notifiers)) + for _, n := range notifiers { + names = append(names, n.Name()) + } + log.Printf("alerts: enabled every %s (window %s, baseline %d windows, dry-run=%v, channels=%v)", cfg.Interval, cfg.Window, cfg.BaselineWindows, cfg.DryRun, names) + go r.loop(ctx) + return r +} + +func (r *Runner) loop(ctx context.Context) { + t := time.NewTicker(r.cfg.Interval) + defer t.Stop() + for { + r.Tick(ctx, time.Now()) + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// Tick runs one evaluation. The first tick after a start only primes the +// suppressor, so a deploy in the middle of an incident does not re-page it. +func (r *Runner) Tick(ctx context.Context, now time.Time) { + cctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + end := now.Add(-r.cfg.Lag).UTC() + start := end.Add(-time.Duration(r.cfg.BaselineWindows+1) * r.cfg.Window) + events, err := r.load(cctx, start, end) + r.mu.Lock() + r.lastRun = now + if err != nil { + r.lastError = err.Error() + r.mu.Unlock() + log.Printf("alerts: load failed: %v", err) + return + } + r.lastError = "" + if r.since.IsZero() || now.Sub(r.sinceAt) > time.Hour { + r.mu.Unlock() + since := r.dataSince(cctx) + r.mu.Lock() + r.since, r.sinceAt = since, now + } + since, primed := r.since, r.primed + r.mu.Unlock() + + spikes := DetectAlerts(events, end, features.SpikeConfig{ + Window: r.cfg.Window, BaselineWindows: r.cfg.BaselineWindows, Multiple: r.cfg.Multiple, MinExcess: r.cfg.MinExcess, DataSince: since, + }) + spikes = r.sup.Filter(spikes) + + r.mu.Lock() + r.primed = true + r.mu.Unlock() + if !primed { + if len(spikes) > 0 { + log.Printf("alerts: primed with %d ongoing spike(s), not sent", len(spikes)) + } + return + } + for _, spike := range spikes { + r.deliver(cctx, NewAlert(spike, end)) + } +} + +// DetectAlerts is the detection half of a tick: spikes reduced to actionable +// patterns. Exported so the CLI and the runner cannot drift apart. +func DetectAlerts(events []features.Event, at time.Time, cfg features.SpikeConfig) []features.Spike { + return FilterActionable(features.DetectSpikes(events, at, cfg), false, nil) +} + +func (r *Runner) deliver(ctx context.Context, a Alert) { + r.mu.Lock() + r.recent = append(r.recent, a) + if len(r.recent) > maxRecent { + r.recent = r.recent[len(r.recent)-maxRecent:] + } + r.mu.Unlock() + if r.cfg.DryRun { + log.Printf("alerts[dry-run]: %s", a.Text(r.cfg.BaseURL)) + return + } + for _, n := range r.notifiers { + err := n.Send(ctx, a) + if err != nil { + time.Sleep(2 * time.Second) + err = n.Send(ctx, a) + } + if err != nil { + log.Printf("alerts: %s delivery failed for %s: %v", n.Name(), a.ID, err) + } + } +} + +// Status is the JSON served by GET /api/alerts. +type Status struct { + Enabled bool `json:"enabled"` + DryRun bool `json:"dryRun"` + Channels []string `json:"channels"` + LastRun time.Time `json:"lastRun"` + LastError string `json:"lastError,omitempty"` + Alerts []Alert `json:"alerts"` +} + +func (r *Runner) Status() Status { + r.mu.Lock() + defer r.mu.Unlock() + s := Status{Enabled: true, DryRun: r.cfg.DryRun, LastRun: r.lastRun, LastError: r.lastError, Channels: []string{}} + for _, n := range r.notifiers { + s.Channels = append(s.Channels, n.Name()) + } + s.Alerts = make([]Alert, len(r.recent)) + for i, a := range r.recent { // newest first + s.Alerts[len(r.recent)-1-i] = a + } + return s +} + +// HandleStatus serves recent alerts and runner health. +func (r *Runner) HandleStatus(w http.ResponseWriter, req *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Header().Set("Cache-Control", "no-store") + _ = json.NewEncoder(w).Encode(r.Status()) +} + +// HandleTest sends a clearly marked test alert to every channel, so a new +// token or chat id can be verified without waiting for a real incident. +func (r *Runner) HandleTest(w http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + now := time.Now().UTC() + a := NewAlert(features.Spike{Service: "alert-test", WindowStart: now.Add(-r.cfg.Window), WindowEnd: now, ErrorCount: 42, BaselineMean: 1, Excess: 41, Patterns: []features.PatternDelta{{ + Signature: "test", Count: 42, BaselineMean: 1, Excess: 41, Novel: true, Clients: 3, + Example: features.Event{Service: "alert-test", Kind: "span", StatusCode: 500, Route: "/api/example", Message: "this is a test alert"}, + }}}, now) + a.Test = true + results := map[string]string{} + for _, n := range r.notifiers { + if err := n.Send(req.Context(), a); err != nil { + results[n.Name()] = err.Error() + } else { + results[n.Name()] = "ok" + } + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(results) +} diff --git a/internal/features/features.go b/internal/features/features.go index 1f6ed0f..f14ee08 100644 --- a/internal/features/features.go +++ b/internal/features/features.go @@ -7,6 +7,8 @@ package features import ( "sort" "time" + + "git.tornberg.me/go-otel/internal/uaclass" ) // Event is the subset of a normalized event used by the detector. Text fields @@ -31,6 +33,14 @@ type Event struct { Route string StatusCode int ExceptionType string + + // Trace-resolved context (see internal/tracectx). All optional: zero + // values mean "not resolved", which behaves like the pre-resolution data. + ClientIP string // real client behind proxy hops, when known + UAClass string // internal/uaclass class of the User-Agent + IsDuplicate bool // repeats an ancestor span's failure; not a new request + IsRootCause bool // error span with no erroring child + HasUpstream bool // an ingress span that had a child in another service } // Config controls feature and label windows. A zero Config receives the @@ -94,6 +104,18 @@ type Row struct { TimeoutCount int `json:"timeoutCount"` CancellationCount int `json:"cancellationCount"` + // Logical-request view: DuplicateErrors are hops repeating an ancestor's + // failure, so LogicalErrors = ErrorCount - DuplicateErrors is the number of + // failed requests. DistinctClients / TopClientShare separate one noisy + // client from broad impact; BotErrorShare is the share of errors whose + // User-Agent classifies as a self-declared non-human client. + LogicalErrors int `json:"logicalErrors"` + DuplicateErrors int `json:"duplicateErrors"` + RootCauseErrors int `json:"rootCauseErrors"` + DistinctClients int `json:"distinctClients"` + TopClientShare float64 `json:"topClientShare"` + BotErrorShare float64 `json:"botErrorShare"` + BaselineErrorRate float64 `json:"baselineErrorRate"` BaselineSamples int `json:"baselineSamples"` FutureEventCount int `json:"futureEventCount"` @@ -189,10 +211,13 @@ type accumulator struct { durations []float64 traces map[string]struct{} issues map[string]struct{} + clients map[string]int + botErrors int + classed int } func newAccumulator(start, end time.Time, service, release string) *accumulator { - return &accumulator{value: Row{WindowStart: start, WindowEnd: end, Service: service, Release: release}, traces: map[string]struct{}{}, issues: map[string]struct{}{}} + return &accumulator{value: Row{WindowStart: start, WindowEnd: end, Service: service, Release: release}, traces: map[string]struct{}{}, issues: map[string]struct{}{}, clients: map[string]int{}} } func (a *accumulator) add(e Event) { @@ -223,6 +248,20 @@ func (a *accumulator) add(e Event) { return } a.value.ErrorCount++ + if e.IsDuplicate { + a.value.DuplicateErrors++ + } else if e.ClientIP != "" { + a.clients[e.ClientIP]++ + } + if e.IsRootCause { + a.value.RootCauseErrors++ + } + if e.UAClass != "" && !e.IsDuplicate { + a.classed++ + if uaclass.IsBot(e.UAClass) { + a.botErrors++ + } + } if e.Source == "frontend" { a.value.FrontendErrors++ } else { @@ -252,6 +291,17 @@ func issueKey(e Event) string { func (a *accumulator) row() Row { a.value.ErrorRate = rate(a.value.ErrorCount, a.value.TotalEvents) a.value.UniqueTraceCount, a.value.UniqueIssueCount = len(a.traces), len(a.issues) + a.value.LogicalErrors = a.value.ErrorCount - a.value.DuplicateErrors + a.value.DistinctClients = len(a.clients) + top, known := 0, 0 + for _, n := range a.clients { + known += n + if n > top { + top = n + } + } + a.value.TopClientShare = rate(top, known) + a.value.BotErrorShare = rate(a.botErrors, a.classed) if len(a.durations) != 0 { sort.Float64s(a.durations) a.value.P50Duration = percentile(a.durations, .50) diff --git a/internal/features/features_test.go b/internal/features/features_test.go index f3e2a5d..d0a4b12 100644 --- a/internal/features/features_test.go +++ b/internal/features/features_test.go @@ -54,3 +54,24 @@ func TestBuildWithholdsLabelWithoutCompleteBaseline(t *testing.T) { t.Fatalf("short history must not be labelled: %+v", rows) } } + +func TestBuildLogicalRequestFeatures(t *testing.T) { + at := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC) + mk := func(client string, dup bool, class string) Event { + return Event{Timestamp: at, Service: "nginx", Kind: "span", IsError: true, PatternSignature: "p", ClientIP: client, IsDuplicate: dup, UAClass: class} + } + rows := Build([]Event{mk("a", false, "ai_crawler"), mk("a", true, "ai_crawler"), mk("a", false, "browser"), mk("b", false, "browser")}, Config{}) + if len(rows) == 0 { + t.Fatal("no rows") + } + r := rows[0] + if r.ErrorCount != 4 || r.DuplicateErrors != 1 || r.LogicalErrors != 3 { + t.Fatalf("logical errors wrong: %+v", r) + } + if r.DistinctClients != 2 || r.TopClientShare < 0.66 || r.TopClientShare > 0.67 { + t.Fatalf("client features wrong: %+v", r) + } + if r.BotErrorShare < 0.33 || r.BotErrorShare > 0.34 { + t.Fatalf("bot share wrong: %v", r.BotErrorShare) + } +} diff --git a/internal/features/spike.go b/internal/features/spike.go index f8e617d..5bebc51 100644 --- a/internal/features/spike.go +++ b/internal/features/spike.go @@ -33,7 +33,8 @@ type PatternDelta struct { Count int `json:"count"` BaselineMean float64 `json:"baselineMean"` Excess float64 `json:"excess"` - Novel bool `json:"novel"` // never seen in the baseline + Novel bool `json:"novel"` // never seen in the baseline + Clients int `json:"clients,omitempty"` // distinct resolved client IPs in the current window Example Event `json:"example"` } @@ -84,6 +85,7 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { cur int base []int // per baseline window example Event + clients map[string]struct{} } type svc struct { cur int @@ -92,7 +94,10 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { } services := map[string]*svc{} for _, e := range events { - if !e.IsError { + // A duplicate is a later hop of a request whose outermost error is + // already counted; counting it again multiplies every failure by the + // number of services it crossed. + if !e.IsError || e.IsDuplicate { continue } ts := e.Timestamp.UTC() @@ -123,6 +128,12 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { s.cur++ p.cur++ p.example = e + if e.ClientIP != "" { + if p.clients == nil { + p.clients = map[string]struct{}{} + } + p.clients[e.ClientIP] = struct{}{} + } } else { idx := int(ts.Sub(baseStart) / cfg.Window) if idx >= cfg.BaselineWindows { @@ -154,7 +165,7 @@ func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { } spike.Excess += excess spike.Patterns = append(spike.Patterns, PatternDelta{ - Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: !seen, Example: p.example, + Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: !seen, Example: p.example, Clients: len(p.clients), }) } if spike.Excess < float64(cfg.MinExcess) { diff --git a/internal/features/spike_test.go b/internal/features/spike_test.go index f9ae709..d8f2ad1 100644 --- a/internal/features/spike_test.go +++ b/internal/features/spike_test.go @@ -98,3 +98,26 @@ func TestDetectSpikesWarmupGuard(t *testing.T) { t.Fatalf("guard should be off when DataSince is zero") } } + +func TestDetectSpikesSkipsDuplicateHopsAndCountsClients(t *testing.T) { + at := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC) + cfg := SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 12} + var events []Event + for i := 0; i < 8; i++ { + ts := at.Add(-time.Minute).Add(time.Duration(i) * time.Second) + client := "203.0.113." + string(rune('1'+i%2)) + events = append(events, + Event{Timestamp: ts, Service: "nginx", IsError: true, PatternSignature: "p", ClientIP: client}, + // two more hops of the same request must not count + Event{Timestamp: ts, Service: "nginx", IsError: true, PatternSignature: "p", ClientIP: client, IsDuplicate: true}, + Event{Timestamp: ts, Service: "nginx", IsError: true, PatternSignature: "p", ClientIP: client, IsDuplicate: true}, + ) + } + got := DetectSpikes(events, at, cfg) + if len(got) != 1 || got[0].ErrorCount != 8 || got[0].Patterns[0].Count != 8 { + t.Fatalf("duplicates were counted: %+v", got) + } + if got[0].Patterns[0].Clients != 2 { + t.Fatalf("clients = %d, want 2", got[0].Patterns[0].Clients) + } +} diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go index ffb1827..10061c3 100644 --- a/internal/storage/clickhouse.go +++ b/internal/storage/clickhouse.go @@ -551,6 +551,11 @@ func (s *ClickHouseStore) Services(ctx context.Context) ([]string, error) { } // Close closes the underlying connection. +// Conn and Database expose the connection for read-only analysis jobs such +// as the alert runner. +func (s *ClickHouseStore) Conn() driver.Conn { return s.conn } +func (s *ClickHouseStore) Database() string { return s.db } + func (s *ClickHouseStore) Close() error { return s.conn.Close() } diff --git a/internal/tracectx/tracectx.go b/internal/tracectx/tracectx.go new file mode 100644 index 0000000..da39beb --- /dev/null +++ b/internal/tracectx/tracectx.go @@ -0,0 +1,134 @@ +// Package tracectx resolves parent/child relationships between spans that +// arrive in different OTLP envelopes. One user request is exported as several +// spans by several processes (ingress outer span, ingress inner span on a +// loopback hop, the application span), so a single span is a poor unit for +// both display and modelling. Resolution therefore works on a set of spans +// loaded together (a time window, or every span of the traces of interest) and +// is independent of arrival order. +package tracectx + +import "git.tornberg.me/go-otel/internal/honeypot" + +// Span is the subset of a normalized span needed for resolution. +type Span struct { + TraceID string + SpanID string + ParentID string + Service string + PeerIP string // immediate peer address as recorded on the span + IsError bool + StatusCode int + Edge bool // ingress/proxy service (see IsEdgeService) +} + +// Key identifies a span inside a resolved set. +type Key struct{ TraceID, SpanID string } + +// Info is what resolution learned about one span. +type Info struct { + // ClientIP is the first public peer address on the span or its ancestors. + // It is empty when every hop is internal (LAN client, NAT/firewall). + ClientIP string + // ClientHops is 0 when ClientIP came from the span itself, N when it came + // from the Nth ancestor. A non-zero value is what fixes loopback hops. + ClientHops int + // EdgePeer is the peer address of the outermost edge span, public or not. + EdgePeer string + RootSpan string // outermost ancestor present in the set + Depth int // ancestors present in the set + // Children counts direct children; UpstreamChildren counts those in a + // different service. UpstreamChildren == 0 on an edge span means the proxy + // answered on its own (probe, rate limit, unrouted host). + Children int + UpstreamChildren int + // ErrorChildren counts direct children that are errors. + ErrorChildren int + // IsRootCause marks an error span with no erroring child: the deepest + // place the failure is visible. + IsRootCause bool + // IsDuplicate marks an error that repeats the failure of an ancestor (same + // status code, or no status on either side). The outermost error owns the + // request; counting the duplicates inflates error rates by the number of + // hops a request passed through. + IsDuplicate bool + // ParentMissing is true when the span names a parent that is not in the set + // (client-supplied traceparent, or a service that was not exporting). + ParentMissing bool +} + +const maxHops = 32 + +// IsEdgeService reports whether a service name is an ingress/proxy layer. +func IsEdgeService(service string) bool { + switch service { + case "nginx", "ingress-nginx", "envoy", "traefik", "haproxy", "gateway": + return true + } + return false +} + +// Resolve derives Info for every span. Spans without a trace or span id are +// ignored. Cycles and pathological depth are cut off after maxHops. +func Resolve(spans []Span) map[Key]Info { + byKey := make(map[Key]Span, len(spans)) + for _, s := range spans { + if s.TraceID == "" || s.SpanID == "" { + continue + } + byKey[Key{s.TraceID, s.SpanID}] = s + } + children := map[Key][]Key{} + for k, s := range byKey { + if s.ParentID != "" { + p := Key{s.TraceID, s.ParentID} + children[p] = append(children[p], k) + } + } + out := make(map[Key]Info, len(byKey)) + for k, s := range byKey { + info := Info{RootSpan: s.SpanID} + if !honeypot.Untrackable(s.PeerIP) { + info.ClientIP = s.PeerIP + } + if s.Edge && s.PeerIP != "" { + info.EdgePeer = s.PeerIP + } + cur := s + for hops := 1; cur.ParentID != "" && hops <= maxHops; hops++ { + parent, ok := byKey[Key{cur.TraceID, cur.ParentID}] + if !ok { + if hops == 1 { + info.ParentMissing = true + } + break + } + info.Depth = hops + info.RootSpan = parent.SpanID + if info.ClientIP == "" && !honeypot.Untrackable(parent.PeerIP) { + info.ClientIP, info.ClientHops = parent.PeerIP, hops + } + if parent.Edge && parent.PeerIP != "" { + info.EdgePeer = parent.PeerIP // keep walking: outermost wins + } + if s.IsError && parent.IsError && info.Depth > 0 && sameFailure(s, parent) { + info.IsDuplicate = true + } + cur = parent + } + for _, ck := range children[k] { + c := byKey[ck] + info.Children++ + if c.Service != s.Service { + info.UpstreamChildren++ + } + if c.IsError { + info.ErrorChildren++ + } + } + info.IsRootCause = s.IsError && info.ErrorChildren == 0 + out[k] = info + } + return out +} + +func sameFailure(a, b Span) bool { return a.StatusCode == b.StatusCode } diff --git a/internal/tracectx/tracectx_test.go b/internal/tracectx/tracectx_test.go new file mode 100644 index 0000000..2eb9285 --- /dev/null +++ b/internal/tracectx/tracectx_test.go @@ -0,0 +1,58 @@ +package tracectx + +import "testing" + +func TestResolveLoopbackHopAndDuplicates(t *testing.T) { + // outer nginx (public client) -> inner nginx (loopback) -> app span; the + // same 401 is recorded at every hop. + spans := []Span{ + {TraceID: "t", SpanID: "outer", Service: "nginx", Edge: true, PeerIP: "203.0.113.9", IsError: true, StatusCode: 401}, + {TraceID: "t", SpanID: "inner", ParentID: "outer", Service: "nginx", Edge: true, PeerIP: "127.0.0.1", IsError: true, StatusCode: 401}, + {TraceID: "t", SpanID: "app", ParentID: "inner", Service: "orchestrator", PeerIP: "10.244.4.195", IsError: true, StatusCode: 401}, + } + got := Resolve(spans) + inner := got[Key{"t", "inner"}] + if inner.ClientIP != "203.0.113.9" || inner.ClientHops != 1 { + t.Fatalf("inner client = %+v", inner) + } + app := got[Key{"t", "app"}] + if app.ClientIP != "203.0.113.9" || app.ClientHops != 2 || app.EdgePeer != "203.0.113.9" { + t.Fatalf("app client = %+v", app) + } + if got[Key{"t", "outer"}].IsDuplicate || !got[Key{"t", "inner"}].IsDuplicate || !got[Key{"t", "app"}].IsDuplicate { + t.Fatal("only the outermost error may own the request") + } + if !app.IsRootCause || got[Key{"t", "outer"}].IsRootCause { + t.Fatal("root cause is the deepest error") + } + if outer := got[Key{"t", "outer"}]; outer.Children != 1 || outer.UpstreamChildren != 0 || outer.Depth != 0 { + t.Fatalf("outer = %+v", outer) + } + if inner.UpstreamChildren != 1 { + t.Fatalf("inner should have one upstream (other service) child: %+v", inner) + } +} + +func TestResolveDifferentStatusIsNotDuplicate(t *testing.T) { + got := Resolve([]Span{ + {TraceID: "t", SpanID: "a", Service: "nginx", Edge: true, PeerIP: "203.0.113.9", IsError: true, StatusCode: 502}, + {TraceID: "t", SpanID: "b", ParentID: "a", Service: "api", IsError: true, StatusCode: 500}, + }) + if got[Key{"t", "b"}].IsDuplicate { + t.Fatal("a 500 under a 502 is a distinct failure") + } +} + +func TestResolveMissingParentAndCycle(t *testing.T) { + got := Resolve([]Span{ + {TraceID: "t", SpanID: "x", ParentID: "gone", Service: "nginx", PeerIP: "198.51.100.7"}, + {TraceID: "t", SpanID: "c1", ParentID: "c2", Service: "s", IsError: true}, + {TraceID: "t", SpanID: "c2", ParentID: "c1", Service: "s", IsError: true}, + }) + if !got[Key{"t", "x"}].ParentMissing || got[Key{"t", "x"}].ClientIP != "198.51.100.7" { + t.Fatalf("x = %+v", got[Key{"t", "x"}]) + } + if _, ok := got[Key{"t", "c1"}]; !ok { // must terminate + t.Fatal("cycle dropped a span") + } +} diff --git a/internal/uaclass/uaclass.go b/internal/uaclass/uaclass.go index 98d2ddf..3af8402 100644 --- a/internal/uaclass/uaclass.go +++ b/internal/uaclass/uaclass.go @@ -55,6 +55,8 @@ var rules = []rule{ {AICrawler, "Google-CloudVertexBot", "google-cloudvertexbot"}, {AICrawler, "CCBot", "ccbot"}, {AICrawler, "Bytespider", "bytespider"}, + {AICrawler, "Kimi-SearchBot", "kimi-searchbot"}, + {AICrawler, "MoonshotBot", "moonshot"}, {AICrawler, "Amazonbot", "amazonbot"}, {AICrawler, "Applebot-Extended", "applebot-extended"}, {AICrawler, "meta-externalagent", "meta-externalagent"}, @@ -169,6 +171,8 @@ var rules = []rule{ // Scanners and exploit tooling. {ScanTool, "CameraScanner", "camerascanner"}, + {ScanTool, "libredtail-http", "libredtail"}, + {ScanTool, "jscrawler", "jscrawler"}, {ScanTool, "zgrab", "zgrab"}, {ScanTool, "masscan", "masscan"}, {ScanTool, "Nmap", "nmap scripting engine"}, diff --git a/internal/uaclass/uaclass_test.go b/internal/uaclass/uaclass_test.go index ffdf684..c3f895f 100644 --- a/internal/uaclass/uaclass_test.go +++ b/internal/uaclass/uaclass_test.go @@ -20,6 +20,8 @@ func TestClassify(t *testing.T) { {"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36", Browser, ""}, {"Mozilla/5.0 (compatible; SomeNewFooBot/3.1)", OtherBot, ""}, {"Mozilla/5.0 (Macintosh) Nodeblock", Browser, ""}, // "node" must not match inside a browser UA + {"Mozilla/5.0 (compatible; Kimi-SearchBot/1.0; +https://www.moonshot.cn)", AICrawler, "Kimi-SearchBot"}, + {"libredtail-http", ScanTool, "libredtail-http"}, {"weird-thing", Unknown, ""}, } for _, c := range cases { diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 58aaede..6c14ff7 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -103,6 +103,24 @@ spec: value: "true" - name: OTEL_BLOCKER_ALLOW value: "" + # Spike alerts to Telegram. The secret is created out of band (never + # commit values): kubectl create secret generic otel-alerts -n monitoring + # --from-literal=telegram-token=... --from-literal=telegram-chat-id=... + # Set OTEL_ALERT_DRY_RUN=true to evaluate and record without sending. + - name: OTEL_ALERT_TELEGRAM_TOKEN + valueFrom: + secretKeyRef: + name: otel-alerts + key: telegram-token + optional: true + - name: OTEL_ALERT_TELEGRAM_CHAT_ID + valueFrom: + secretKeyRef: + name: otel-alerts + key: telegram-chat-id + optional: true + - name: OTEL_ALERT_BASE_URL + value: "https://otel.k6n.net" - name: OTEL_STORE_WORKERS value: "4" # Optional rollout observation. The watcher is read-only and uses the diff --git a/otel-ui/index.html b/otel-ui/index.html index d2eda5c..ca36007 100644 --- a/otel-ui/index.html +++ b/otel-ui/index.html @@ -7,7 +7,7 @@ HyperScope OTel — Distributed Telemetry - = { @@ -64,6 +65,20 @@ interface HoneypotIP { promoted_until: string } +interface BlockerStatus { + enabled: boolean + dryRun: boolean + candidates?: string[] + managed?: Record +} + +/** An IP flagged by the honeypot tracker or the ingress blocker. */ +interface DetectedIP { + ip: string + sources: ("honeypot" | "blocker")[] + count?: number +} + interface DetectorTestResult { passed?: boolean scenario?: string @@ -684,6 +699,7 @@ export const IssuesVisualizer = ({ const [clientExclude, setClientExclude] = useState>(new Set()) const [showSpamIPs, setShowSpamIPs] = useState(false) const [spamIPs, setSpamIPs] = useState([]) + const [blockerStatus, setBlockerStatus] = useState(null) const [spamIPsLoading, setSpamIPsLoading] = useState(false) const [spamIPsError, setSpamIPsError] = useState(null) @@ -694,6 +710,9 @@ export const IssuesVisualizer = ({ const response = await authFetch("/api/honeypot/ips") if (!response.ok) throw new Error(`HTTP ${response.status}`) setSpamIPs((await response.json()) as HoneypotIP[]) + // The blocker route only exists when the blocker is enabled; its absence is not an error. + const blocker = await authFetch("/api/blocker/status").catch(() => null) + if (blocker?.ok) setBlockerStatus((await blocker.json()) as BlockerStatus) } catch (err) { setSpamIPsError(err instanceof Error ? err.message : String(err)) } finally { @@ -701,7 +720,32 @@ export const IssuesVisualizer = ({ } } - const spamIPSet = useMemo(() => new Set(spamIPs.map((entry) => entry.ip)), [spamIPs]) + const detectedIPs = useMemo(() => { + const byIP = new Map() + for (const entry of spamIPs) byIP.set(entry.ip, { ip: entry.ip, sources: ["honeypot"], count: entry.count }) + for (const ip of blockerStatus?.candidates ?? []) { + const known = byIP.get(ip) + if (known) known.sources.push("blocker") + else byIP.set(ip, { ip, sources: ["blocker"] }) + } + return Array.from(byIP.values()) + }, [spamIPs, blockerStatus]) + const spamIPSet = useMemo(() => new Set(detectedIPs.map((entry) => entry.ip)), [detectedIPs]) + + const crawlersHidden = CRAWLER_CLASSES.every((c) => clientExclude.has(c)) + const toggleCrawlers = () => { + setClientExclude((prev) => { + const next = new Set(prev) + if (CRAWLER_CLASSES.every((c) => prev.has(c))) CRAWLER_CLASSES.forEach((c) => next.delete(c)) + else CRAWLER_CLASSES.forEach((c) => next.add(c)) + return next + }) + setClientInclude((prev) => { + const next = new Set(prev) + CRAWLER_CLASSES.forEach((c) => next.delete(c)) + return next + }) + } const statusCounts = useMemo(() => { const counts: Record = { all: issues.length, open: 0, resolved: 0, ignored: 0 } @@ -954,7 +998,7 @@ export const IssuesVisualizer = ({ setExcludeSpamIPs((v) => !v) }} aria-pressed={excludeSpamIPs} - title="Hide HTTP errors whose recorded source IP is on the honeypot list" + title="Hide HTTP errors whose recorded source IP is on the honeypot list or the blocker's candidate list" className={`inline-flex h-9 items-center gap-1.5 rounded-[9px] border px-2.5 text-[12px] font-medium transition-colors ${ excludeSpamIPs ? "border-ds-warn/40 bg-ds-warn-bg text-ds-warn" @@ -964,6 +1008,20 @@ export const IssuesVisualizer = ({