diff --git a/cmd/needle-sidecar/eval/__pycache__/run_eval.cpython-314.pyc b/cmd/needle-sidecar/eval/__pycache__/run_eval.cpython-314.pyc new file mode 100644 index 0000000..6cb7261 Binary files /dev/null and b/cmd/needle-sidecar/eval/__pycache__/run_eval.cpython-314.pyc differ diff --git a/cmd/needle-sidecar/eval/run_eval.py b/cmd/needle-sidecar/eval/run_eval.py index 93169ad..67d0d06 100755 --- a/cmd/needle-sidecar/eval/run_eval.py +++ b/cmd/needle-sidecar/eval/run_eval.py @@ -63,6 +63,10 @@ def main(): print(f"\naccuracy {acc:.2%} over {len(results)} rows; " f"median latency {sorted(lat)[len(lat)//2]:.0f}ms") + for src in sorted({r.get("src", "?") for r in results}): + sub = [r for r in results if r.get("src", "?") == src] + print(f" {src:<11} {sum(r['ok'] for r in sub)}/{len(sub)}") + wrong = [r for r in results if not r["ok"]] if wrong: print("\nmisses:") diff --git a/cmd/server/main.go b/cmd/server/main.go index 7afafac..f5677d8 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -234,6 +234,7 @@ func main() { } wsServer := websocket.New(coll) honeypotTracker := honeypot.New(honeypot.ConfigFromEnv()) + coll.SetHoneypotRecorder(honeypotTracker) // Update collector with wsServer coll.SetBroadcaster(wsServer) ingestKey := os.Getenv("OTEL_INGEST_KEY") diff --git a/internal/collector/collector.go b/internal/collector/collector.go index 789cb60..c8f9556 100644 --- a/internal/collector/collector.go +++ b/internal/collector/collector.go @@ -51,6 +51,7 @@ type Collector struct { sourceMaps *sourcemaps.Resolver kubeWatcher *kube.Watcher prometheus *prometheus.Client + honeypot HoneypotRecorder // Optional capabilities of the configured store (ClickHouse implements // all of them); nil when unsupported, and the API falls back to the @@ -109,6 +110,17 @@ func (c *Collector) SetIngestKey(key string) { c.ingestKey = key } +// HoneypotRecorder receives suspicious request observations extracted from +// telemetry. It is intentionally a small interface so the collector does not +// depend on the HTTP honeypot implementation. +type HoneypotRecorder interface { + RecordTelemetry(ip, path string) bool +} + +func (c *Collector) SetHoneypotRecorder(recorder HoneypotRecorder) { + c.honeypot = recorder +} + // SetStatsQuerier enables the aggregate stats endpoints (error trends, // release comparisons) when a store supporting them is configured. func (c *Collector) SetStatsQuerier(q storage.StatsQuerier) { @@ -353,6 +365,7 @@ func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error { // Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage c.addToHistory(websocket.DataTypeTraces, buf) + recordSuspiciousTraceIPs(c.honeypot, td) if extractSpanErrors(c.issues, td, time.Now().UnixNano()) { c.notifyIssues() } @@ -398,6 +411,7 @@ func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error { // Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage c.addToHistory(websocket.DataTypeLogs, buf) + recordSuspiciousLogIPs(c.honeypot, ld) if extractLogErrors(c.issues, ld, time.Now().UnixNano()) { c.notifyIssues() } diff --git a/internal/collector/issue_extract.go b/internal/collector/issue_extract.go index bc9c66e..007bfae 100644 --- a/internal/collector/issue_extract.go +++ b/internal/collector/issue_extract.go @@ -44,6 +44,42 @@ func extractSpanErrors(tracker *issues.Tracker, td ptrace.Traces, timestamp int6 return found } +func recordSuspiciousTraceIPs(recorder HoneypotRecorder, td ptrace.Traces) { + if recorder == nil { + return + } + resourceSpans := td.ResourceSpans() + for i := 0; i < resourceSpans.Len(); i++ { + rs := resourceSpans.At(i) + scopeSpans := rs.ScopeSpans() + for j := 0; j < scopeSpans.Len(); j++ { + spans := scopeSpans.At(j).Spans() + for k := 0; k < spans.Len(); k++ { + span := spans.At(k) + _ = recorder.RecordTelemetry(sourceIP(span.Attributes(), rs.Resource().Attributes()), httpRoute(span.Attributes())) + } + } + } +} + +func recordSuspiciousLogIPs(recorder HoneypotRecorder, ld plog.Logs) { + if recorder == nil { + return + } + resourceLogs := ld.ResourceLogs() + for i := 0; i < resourceLogs.Len(); i++ { + rl := resourceLogs.At(i) + scopeLogs := rl.ScopeLogs() + for j := 0; j < scopeLogs.Len(); j++ { + records := scopeLogs.At(j).LogRecords() + for k := 0; k < records.Len(); k++ { + record := records.At(k) + _ = recorder.RecordTelemetry(sourceIP(record.Attributes(), rl.Resource().Attributes()), httpRoute(record.Attributes())) + } + } + } +} + // extractLogErrors scans logs for error/fatal severity records and records // one issue occurrence per error found. It reports whether any error was // seen, so the caller can notify live clients. @@ -131,23 +167,27 @@ func logSample(record plog.LogRecord, res pcommon.Resource, timestamp int64, tit // its resource. func sourceIP(attrs, resource pcommon.Map) string { ip := firstAttr(attrs, + "http.request.header.x-real-ip", + "http.request.header.x_real_ip", + "http.request.header.x-forwarded-for", + "http.request.header.x_forwarded_for", "client.address", "client.ip", "http.client_ip", - "net.peer.ip", "network.peer.address", - "http.request.header.x-real-ip", - "http.request.header.x-forwarded-for", - "http.request.header.x_real_ip", - "http.request.header.x_forwarded_for", + "net.peer.ip", ) if ip == "" { ip = firstAttr(resource, + "http.request.header.x-real-ip", + "http.request.header.x_real_ip", + "http.request.header.x-forwarded-for", + "http.request.header.x_forwarded_for", "client.address", "client.ip", "http.client_ip", - "net.peer.ip", "network.peer.address", + "net.peer.ip", ) } return strings.TrimSpace(strings.SplitN(ip, ",", 2)[0]) diff --git a/internal/features/features.go b/internal/features/features.go index 3feaa59..1f6ed0f 100644 --- a/internal/features/features.go +++ b/internal/features/features.go @@ -28,6 +28,9 @@ type Event struct { IsCancellation bool Message string SpanName string + Route string + StatusCode int + ExceptionType string } // Config controls feature and label windows. A zero Config receives the diff --git a/internal/features/spike.go b/internal/features/spike.go new file mode 100644 index 0000000..ebcf693 --- /dev/null +++ b/internal/features/spike.go @@ -0,0 +1,155 @@ +package features + +import ( + "sort" + "time" +) + +// SpikeConfig controls the deterministic "is this service misbehaving right +// now" rule. It works on excess errors: a pattern that fires at its usual +// rate (a client polling a 401 endpoint every few seconds) contributes zero, +// so only new or rising patterns can raise an alert. +type SpikeConfig struct { + Window time.Duration // current observation window + BaselineWindows int // preceding windows used as the baseline + Multiple float64 // a pattern must be novel or >= Multiple * its own baseline mean + MinExcess int // and excess errors must be >= MinExcess + TopPatterns int +} + +func DefaultSpikeConfig() SpikeConfig { + return SpikeConfig{Window: 5 * time.Minute, BaselineWindows: 36, Multiple: 3, MinExcess: 5, TopPatterns: 5} +} + +// PatternDelta is one error pattern's contribution to a spike. +type PatternDelta struct { + Signature string `json:"signature"` + Count int `json:"count"` + BaselineMean float64 `json:"baselineMean"` + Excess float64 `json:"excess"` + Novel bool `json:"novel"` // never seen in the baseline + Example Event `json:"example"` +} + +type Spike struct { + Service string `json:"service"` + WindowStart time.Time `json:"windowStart"` + WindowEnd time.Time `json:"windowEnd"` + ErrorCount int `json:"errorCount"` + BaselineMean float64 `json:"baselineMean"` + Excess float64 `json:"excess"` + Patterns []PatternDelta `json:"patterns"` +} + +// DetectSpikes evaluates the window ending at `at` against the preceding +// BaselineWindows windows of the same length. Baseline data never includes +// the current window. Only error events are inspected. +func DetectSpikes(events []Event, at time.Time, cfg SpikeConfig) []Spike { + d := DefaultSpikeConfig() + if cfg.Window <= 0 { + cfg.Window = d.Window + } + if cfg.BaselineWindows <= 0 { + cfg.BaselineWindows = d.BaselineWindows + } + if cfg.Multiple <= 0 { + cfg.Multiple = d.Multiple + } + if cfg.MinExcess <= 0 { + cfg.MinExcess = d.MinExcess + } + if cfg.TopPatterns <= 0 { + cfg.TopPatterns = d.TopPatterns + } + at = at.UTC() + start := at.Add(-cfg.Window) + baseStart := start.Add(-time.Duration(cfg.BaselineWindows) * cfg.Window) + + type agg struct { + cur, base int + example Event + } + type svc struct { + cur, base int + patterns map[string]*agg + } + services := map[string]*svc{} + for _, e := range events { + if !e.IsError { + continue + } + ts := e.Timestamp.UTC() + inCur := !ts.Before(start) && ts.Before(at) + inBase := !ts.Before(baseStart) && ts.Before(start) + if !inCur && !inBase { + continue + } + name := e.Service + if name == "" { + name = "unknown" + } + s := services[name] + if s == nil { + s = &svc{patterns: map[string]*agg{}} + services[name] = s + } + key := e.PatternSignature + if key == "" { + key = issueKey(e) + } + p := s.patterns[key] + if p == nil { + p = &agg{} + s.patterns[key] = p + } + if inCur { + s.cur++ + p.cur++ + p.example = e + } else { + s.base++ + p.base++ + if p.example.PatternSignature == "" && p.cur == 0 { + p.example = e + } + } + } + + var spikes []Spike + n := float64(cfg.BaselineWindows) + for name, s := range services { + baseMean := float64(s.base) / n + if s.cur == 0 { + continue + } + spike := Spike{Service: name, WindowStart: start, WindowEnd: at, ErrorCount: s.cur, BaselineMean: baseMean} + for sig, p := range s.patterns { + mean := float64(p.base) / n + excess := float64(p.cur) - mean + // A pattern only counts when it is novel or well above its own + // baseline; the service-wide total is masked by chronic noise. + if p.cur == 0 || excess <= 0 || (p.base > 0 && float64(p.cur) < cfg.Multiple*mean) { + continue + } + spike.Excess += excess + spike.Patterns = append(spike.Patterns, PatternDelta{ + Signature: sig, Count: p.cur, BaselineMean: mean, Excess: excess, Novel: p.base == 0, Example: p.example, + }) + } + if spike.Excess < float64(cfg.MinExcess) { + continue + } + sort.Slice(spike.Patterns, func(i, j int) bool { + if spike.Patterns[i].Excess != spike.Patterns[j].Excess { + return spike.Patterns[i].Excess > spike.Patterns[j].Excess + } + return spike.Patterns[i].Signature < spike.Patterns[j].Signature + }) + if len(spike.Patterns) > cfg.TopPatterns { + spike.Patterns = spike.Patterns[:cfg.TopPatterns] + } + spikes = append(spikes, spike) + } + sort.Slice(spikes, func(i, j int) bool { return spikes[i].Excess > spikes[j].Excess }) + return spikes +} diff --git a/internal/features/spike_test.go b/internal/features/spike_test.go new file mode 100644 index 0000000..fbe4baf --- /dev/null +++ b/internal/features/spike_test.go @@ -0,0 +1,41 @@ +package features + +import ( + "testing" + "time" +) + +func TestDetectSpikesIgnoresChronicPatterns(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 + add := func(sig string, n int, when time.Time) { + for i := 0; i < n; i++ { + events = append(events, Event{Timestamp: when.Add(time.Duration(i) * time.Second), Service: "api", IsError: true, PatternSignature: sig}) + } + } + // chronic poller: 50 errors in every window, including the current one + for w := 0; w <= 12; w++ { + add("poll-401", 50, at.Add(-time.Duration(w+1)*cfg.Window)) + } + if got := DetectSpikes(events, at, cfg); len(got) != 0 { + t.Fatalf("chronic pattern alone alerted: %+v", got) + } + // a novel failure appears in the current window + add("db-timeout", 20, at.Add(-cfg.Window)) + got := DetectSpikes(events, at, cfg) + if len(got) != 1 || got[0].Patterns[0].Signature != "db-timeout" || !got[0].Patterns[0].Novel { + t.Fatalf("expected db-timeout spike, got %+v", got) + } +} + +func TestDetectSpikesRequiresMinimumExcess(t *testing.T) { + at := time.Date(2026, 9, 19, 12, 0, 0, 0, time.UTC) + events := []Event{} + for i := 0; i < 3; i++ { + events = append(events, Event{Timestamp: at.Add(-time.Minute), Service: "api", IsError: true, PatternSignature: "x"}) + } + if got := DetectSpikes(events, at, SpikeConfig{}); len(got) != 0 { + t.Fatalf("3 errors should stay under the floor: %+v", got) + } +} diff --git a/internal/honeypot/honeypot.go b/internal/honeypot/honeypot.go index e53b3ff..1e3d4d5 100644 --- a/internal/honeypot/honeypot.go +++ b/internal/honeypot/honeypot.go @@ -135,7 +135,7 @@ func (t *Tracker) record(ip string) bool { } state.Count++ state.LastSeen = now - if state.Count > t.threshold { + if state.Count >= t.threshold { state.PromotedUntil = now.Add(t.ttl) } t.clients[ip] = state @@ -145,6 +145,19 @@ func (t *Tracker) record(ip string) bool { return !state.PromotedUntil.IsZero() && now.Before(state.PromotedUntil) } +// RecordTelemetry records a suspicious request observed in an OTLP span. This +// is separate from Middleware because the request may have been handled by a +// different application/ingress host while its telemetry is sent to this +// collector. No decoy response is written here; this only updates the +// promoted-IP feed. +func (t *Tracker) RecordTelemetry(ip, path string) bool { + ip = strings.TrimSpace(strings.SplitN(ip, ",", 2)[0]) + if ip == "" || !isSuspiciousPath(path) { + return false + } + return t.record(ip) +} + // PromotedIPs returns a snapshot suitable for an ingress synchronizer. func (t *Tracker) PromotedIPs() []IP { now := time.Now() diff --git a/internal/honeypot/honeypot_test.go b/internal/honeypot/honeypot_test.go index 6e23121..d89c5ce 100644 --- a/internal/honeypot/honeypot_test.go +++ b/internal/honeypot/honeypot_test.go @@ -39,3 +39,20 @@ func TestHandleIPsRequiresToken(t *testing.T) { t.Fatalf("got %d and %q", recorder.Code, recorder.Body.String()) } } + +func TestRecordTelemetryPromotesTheObservedSourceIP(t *testing.T) { + tracker := New(Config{Threshold: 2, TTL: time.Minute}) + if tracker.RecordTelemetry("203.0.113.20", "/mail.php") { + t.Fatal("must not promote before threshold") + } + if !tracker.RecordTelemetry("203.0.113.20", "/settings.php") { + t.Fatal("must promote at threshold") + } + if tracker.RecordTelemetry("203.0.113.20", "/index.html") { + t.Fatal("normal paths must not be tracked") + } + ips := tracker.PromotedIPs() + if len(ips) != 1 || ips[0].Address != "203.0.113.20" { + t.Fatalf("unexpected promoted IPs: %+v", ips) + } +} diff --git a/otel-ui/src/components/IssuesVisualizer.tsx b/otel-ui/src/components/IssuesVisualizer.tsx index 51e6ba2..0c16c01 100644 --- a/otel-ui/src/components/IssuesVisualizer.tsx +++ b/otel-ui/src/components/IssuesVisualizer.tsx @@ -885,7 +885,10 @@ export const IssuesVisualizer = ({