From 8f5c267444658235ac8e8ac963d5d1ad5fac0ac8 Mon Sep 17 00:00:00 2001 From: matst80 Date: Sat, 19 Sep 2026 06:50:19 +0000 Subject: [PATCH] ml --- .gitignore | 2 +- build.sh | 6 + cmd/issue-model/main.go | 240 +++++++++++ cmd/needle-sidecar/Dockerfile | 28 ++ cmd/needle-sidecar/server.py | 195 +++++++++ cmd/normalized-backfill/main.go | 227 ++++++++++ cmd/normalized-preview/main.go | 199 +++++++++ cmd/server/main.go | 11 +- cmd/training-sample/main.go | 130 ++++++ cmd/window-features/main.go | 104 +++++ docs/ML.md | 95 +++++ index.html | 36 ++ internal/collector/collector_test.go | 25 ++ internal/collector/handlers.go | 141 ++++++ internal/collector/issue_extract.go | 16 +- internal/features/features.go | 328 ++++++++++++++ internal/features/features_test.go | 56 +++ internal/issues/tracker.go | 9 +- internal/normalization/normalizer.go | 447 ++++++++++++++++++++ internal/normalization/normalizer_test.go | 108 +++++ k8s/deployment.yaml | 36 ++ k8s/ingress.yaml | 4 + otel-ui/src/components/IssuesVisualizer.tsx | 181 +++++++- otel-ui/src/components/NeedleTestModal.tsx | 236 +++++++++++ scripts/run_needle_test.sh | 21 + scripts/test_needle_classifier.py | 168 ++++++++ 26 files changed, 3039 insertions(+), 10 deletions(-) create mode 100644 cmd/issue-model/main.go create mode 100644 cmd/needle-sidecar/Dockerfile create mode 100755 cmd/needle-sidecar/server.py create mode 100644 cmd/normalized-backfill/main.go create mode 100644 cmd/normalized-preview/main.go create mode 100644 cmd/training-sample/main.go create mode 100644 cmd/window-features/main.go create mode 100644 internal/features/features.go create mode 100644 internal/features/features_test.go create mode 100644 internal/normalization/normalizer.go create mode 100644 internal/normalization/normalizer_test.go create mode 100644 otel-ui/src/components/NeedleTestModal.tsx create mode 100755 scripts/run_needle_test.sh create mode 100755 scripts/test_needle_classifier.py diff --git a/.gitignore b/.gitignore index 1f9fd1a..bd0b69c 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,4 @@ -./server +/server .DS_Store otel-agent.log ./semgrep-results.json diff --git a/build.sh b/build.sh index 3e7a1ef..bc01852 100755 --- a/build.sh +++ b/build.sh @@ -4,12 +4,18 @@ set -euo pipefail RELEASE="${RELEASE:-$(git rev-parse --short HEAD)}" PUBLIC_SOURCEMAPS="${PUBLIC_SOURCEMAPS:-false}" +echo "Building Needle 3 sidecar image..." +docker build -t matst80/needle-sidecar:latest cmd/needle-sidecar +docker push matst80/needle-sidecar:latest + +echo "Building otel-debug receiver image..." docker build \ --build-arg "APP_RELEASE=${RELEASE}" \ --build-arg "PUBLIC_SOURCEMAPS=${PUBLIC_SOURCEMAPS}" \ -t matst80/otel-debug:latest . docker push matst80/otel-debug:latest +kubectl apply -f k8s/ kubectl rollout restart deployment otel-debug-receiver -n monitoring kubectl rollout status deployment/otel-debug-receiver -n monitoring diff --git a/cmd/issue-model/main.go b/cmd/issue-model/main.go new file mode 100644 index 0000000..e385232 --- /dev/null +++ b/cmd/issue-model/main.go @@ -0,0 +1,240 @@ +// issue-model trains and queries an explainable prototype model for normalized +// events. It uses deterministic pattern signatures as weak labels until human +// labels are available. +package main + +import ( + "bufio" + "encoding/json" + "flag" + "fmt" + "log" + "math" + "os" + "sort" + "strconv" + "strings" +) + +type event struct { + EventID string `json:"eventId"` + Service string `json:"service"` + Kind string `json:"kind"` + Source string `json:"source"` + IsError uint8 `json:"isError"` + StatusClass string `json:"statusClass,omitempty"` + StatusCode uint16 `json:"statusCode,omitempty"` + Route string `json:"route,omitempty"` + SpanName string `json:"spanName,omitempty"` + MetricName string `json:"metricName,omitempty"` + MetricType string `json:"metricType,omitempty"` + ExceptionType string `json:"exceptionType,omitempty"` + Message string `json:"messageNormalized,omitempty"` + StackSignature string `json:"stackSignature,omitempty"` + PatternSignature string `json:"patternSignature"` +} + +type prototype struct { + Pattern string `json:"patternSignature"` + Count int `json:"count"` + Tokens map[string]float64 `json:"tokens"` +} + +type model struct { + Version string `json:"version"` + FeatureSpec string `json:"featureSpec"` + Prototypes []prototype `json:"prototypes"` +} + +type prediction struct { + EventID string `json:"eventId"` + ActualPattern string `json:"actualPattern,omitempty"` + Pattern string `json:"patternSignature"` + Score float64 `json:"score"` + Matches []string `json:"matches"` + Rank int `json:"rank"` + Correct bool `json:"correct,omitempty"` + Accepted bool `json:"accepted"` +} + +func main() { + trainPath := flag.String("train", "", "JSONL training sample") + modelPath := flag.String("model", "/tmp/issue-model-v001.json", "model path") + predictPath := flag.String("predict", "", "JSONL events to score") + top := flag.Int("top", 3, "number of predictions per event") + minScore := flag.Float64("min-score", 0.75, "minimum top similarity to accept a match") + minMargin := flag.Float64("min-margin", 0.05, "minimum gap between first and second match") + flag.Parse() + if *trainPath == "" && *predictPath == "" { + log.Fatal("provide -train or -predict") + } + if *trainPath != "" { + trained := train(*trainPath) + writeJSON(*modelPath, trained) + fmt.Printf("trained prototypes=%d model=%s\n", len(trained.Prototypes), *modelPath) + } + if *predictPath != "" { + var trained model + readJSON(*modelPath, &trained) + predict(*predictPath, trained, *top, *minScore, *minMargin) + } +} + +func train(path string) model { + groups := map[string]*prototype{} + forEachEvent(path, func(item event) { + if item.PatternSignature == "" { + return + } + p := groups[item.PatternSignature] + if p == nil { + p = &prototype{Pattern: item.PatternSignature, Tokens: map[string]float64{}} + groups[item.PatternSignature] = p + } + p.Count++ + for token := range features(item) { + p.Tokens[token]++ + } + }) + result := model{Version: "v1", FeatureSpec: "categorical tokens excluding IDs, timestamps, trace IDs, IPs, and raw values"} + for _, p := range groups { + for token, value := range p.Tokens { + p.Tokens[token] = value / float64(p.Count) + } + result.Prototypes = append(result.Prototypes, *p) + } + sort.Slice(result.Prototypes, func(i, j int) bool { return result.Prototypes[i].Pattern < result.Prototypes[j].Pattern }) + return result +} + +func predict(path string, trained model, top int, minScore, minMargin float64) { + if top < 1 { + top = 1 + } + if len(trained.Prototypes) == 0 { + log.Fatal("model contains no prototypes; train it with a non-empty normalized sample") + } + forEachEvent(path, func(item event) { + input := features(item) + results := make([]prediction, 0, len(trained.Prototypes)) + for _, p := range trained.Prototypes { + score, matches := similarity(input, p.Tokens) + results = append(results, prediction{EventID: item.EventID, Pattern: p.Pattern, Score: score, Matches: matches}) + } + sort.Slice(results, func(i, j int) bool { + if results[i].Score != results[j].Score { + return results[i].Score > results[j].Score + } + return results[i].Pattern < results[j].Pattern + }) + if len(results) > top { + results = results[:top] + } + margin := results[0].Score + if len(results) > 1 { + margin -= results[1].Score + } + for i := range results { + results[i].ActualPattern = item.PatternSignature + results[i].Rank = i + 1 + results[i].Correct = results[i].Pattern == item.PatternSignature + results[i].Accepted = results[0].Score >= minScore && margin >= minMargin + } + for _, result := range results { + writeLine(result) + } + }) +} + +func features(item event) map[string]struct{} { + result := map[string]struct{}{} + add := func(key, value string) { + if value != "" { + result[key+"="+value] = struct{}{} + } + } + add("service", item.Service) + add("kind", item.Kind) + add("source", item.Source) + add("status", item.StatusClass) + if item.StatusCode > 0 { + add("statusCode", strconv.Itoa(int(item.StatusCode))) + } + add("route", item.Route) + add("span", item.SpanName) + add("metric", item.MetricName) + add("metricType", item.MetricType) + add("exception", item.ExceptionType) + add("message", item.Message) + add("stack", item.StackSignature) + return result +} + +func similarity(input map[string]struct{}, prototype map[string]float64) (float64, []string) { + if len(input) == 0 || len(prototype) == 0 { + return 0, nil + } + var dot, inputNorm, protoNorm float64 + matches := []string{} + for token := range input { + inputNorm++ + if weight, ok := prototype[token]; ok { + dot += weight + matches = append(matches, token) + } + } + for _, weight := range prototype { + protoNorm += weight * weight + } + if protoNorm == 0 { + return 0, matches + } + score := dot / math.Sqrt(inputNorm*protoNorm) + sort.Strings(matches) + return score, matches +} + +func forEachEvent(path string, callback func(event)) { + file, err := os.Open(path) + if err != nil { + log.Fatal(err) + } + defer file.Close() + scanner := bufio.NewScanner(file) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + for scanner.Scan() { + var item event + if err := json.Unmarshal(scanner.Bytes(), &item); err != nil { + log.Fatal(err) + } + callback(item) + } + if err := scanner.Err(); err != nil { + log.Fatal(err) + } +} +func writeJSON(path string, value any) { + data, err := json.MarshalIndent(value, "", " ") + if err != nil { + log.Fatal(err) + } + if err := os.WriteFile(path, data, 0600); err != nil { + log.Fatal(err) + } +} +func readJSON(path string, value any) { + data, err := os.ReadFile(path) + if err != nil { + log.Fatal(err) + } + if err := json.Unmarshal(data, value); err != nil { + log.Fatal(err) + } +} +func writeLine(value any) { + data, err := json.Marshal(value) + if err != nil { + log.Fatal(err) + } + fmt.Println(strings.TrimSpace(string(data))) +} diff --git a/cmd/needle-sidecar/Dockerfile b/cmd/needle-sidecar/Dockerfile new file mode 100644 index 0000000..8c7f6e8 --- /dev/null +++ b/cmd/needle-sidecar/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-slim + +# Create non-root user with UID 1000 matching Kubernetes pod securityContext +RUN groupadd -g 1000 appuser && useradd -u 1000 -g appuser -m -d /app appuser + +WORKDIR /app + +ENV HOME=/app +ENV NEEDLE_PORT=5005 +ENV NEEDLE_HOST=0.0.0.0 +ENV NEEDLE_TELEMETRY=0 + +# Install dependencies +RUN pip install --no-cache-dir cactus-needle + +# Pre-cache model and shared library into /app/.cache as user 1000 +RUN python -c "import needle; needle.Needle()" + +COPY server.py /app/server.py + +# Ensure UID 1000 owns /app and all caches +RUN chown -R appuser:appuser /app + +USER appuser + +EXPOSE 5005 + +CMD ["python", "/app/server.py"] diff --git a/cmd/needle-sidecar/server.py b/cmd/needle-sidecar/server.py new file mode 100755 index 0000000..0b61c3a --- /dev/null +++ b/cmd/needle-sidecar/server.py @@ -0,0 +1,195 @@ +#!/usr/bin/env python3 +""" +Needle 3 AI Telemetry Triage Sidecar Service +Accepts OTel incident summaries and uses Cactus-Compute/needle3 to classify and extract actionable insights. +""" + +import os +import sys +import json +import time +from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler +import needle + +PORT = int(os.environ.get("NEEDLE_PORT", 5005)) +HOST = os.environ.get("NEEDLE_HOST", "0.0.0.0") + +# --- Define Triage Tools for Needle 3 --- +@needle.tool +def report_client_javascript_crash(service: str, message: str, url: str): + """Report client-side JavaScript crash or error logged by browser RUM.""" + return {"action": "client_js_crash", "service": service, "message": message, "url": url} + +@needle.tool +def alert_vulnerability_probe(scanner: str, target: str, status_code: int): + """Alert on malicious vulnerability scans or exploit crawlers.""" + return {"action": "security_alert", "scanner": scanner, "target": target, "status_code": status_code} + +@needle.tool +def record_frontend_web_vital(service: str, metric_name: str, duration_ms: int): + """Record RUM performance metrics like TTFB or FCP.""" + return {"action": "record_vital", "service": service, "metric": metric_name, "ms": duration_ms} + +@needle.tool +def log_missing_resource_404(path: str, host: str): + """Log benign 404 Not Found requests for missing pages or git commits.""" + return {"action": "log_404", "path": path, "host": host} + +@needle.tool +def report_database_error(service: str, table: str, error_message: str): + """Report database query errors, deadlocks, or connection pool timeouts.""" + return {"action": "database_error", "service": service, "table": table, "error": error_message} + +@needle.tool +def report_auth_failure(service: str, user_or_token: str, reason: str): + """Report authentication/authorization failure or invalid credentials.""" + return {"action": "auth_failure", "service": service, "user_or_token": user_or_token, "reason": reason} + +@needle.tool +def escalate_backend_outage(service: str, error_message: str): + """Escalate severe backend 5xx crash or unhandled panic to on-call.""" + return {"action": "escalate_outage", "service": service, "error": error_message} + +# Initialize Needle Agent once +print(f"Loading Needle 3 model (Cactus-Compute/needle3)...") +t0 = time.perf_counter() +agent = needle.Needle(tools=[ + report_client_javascript_crash, + alert_vulnerability_probe, + record_frontend_web_vital, + log_missing_resource_404, + report_database_error, + report_auth_failure, + escalate_backend_outage +]) +print(f"Needle 3 loaded in {(time.perf_counter() - t0):.2f}s") + + +class NeedleHandler(BaseHTTPRequestHandler): + def _send_cors(self): + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Otel-Agent-Token") + + def do_OPTIONS(self): + self.send_response(204) + self._send_cors() + self.end_headers() + + def do_GET(self): + if self.path == "/health" or self.path == "/api/needle/status": + self.send_response(200) + self._send_cors() + self.send_header("Content-Type", "application/json") + self.end_headers() + resp = { + "status": "ok", + "available": True, + "model": "Cactus-Compute/needle3", + "tools": [ + "client_js_crash", + "security_alert", + "record_vital", + "log_404", + "database_error", + "auth_failure", + "escalate_outage" + ] + } + self.wfile.write(json.dumps(resp).encode("utf-8")) + else: + self.send_response(404) + self._send_cors() + self.end_headers() + + def do_POST(self): + if self.path in ("/classify", "/api/needle/classify"): + content_len = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(content_len) + try: + data = json.loads(body) if body else {} + except Exception as e: + self.send_response(400) + self._send_cors() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"error": f"Invalid JSON: {e}"}).encode("utf-8")) + return + + text = data.get("text", "").strip() + # If text is not provided, build from structured fields: + if not text: + parts = [] + if data.get("service"): + parts.append(f"Service: {data['service']}") + if data.get("kind"): + parts.append(f"Kind: {data['kind']}") + if data.get("title"): + parts.append(f"Title: {data['title']}") + if data.get("message"): + parts.append(f"Message: {data['message']}") + if data.get("url"): + parts.append(f"URL: {data['url']}") + if data.get("status_code"): + parts.append(f"Status: {data['status_code']}") + text = " | ".join(parts) + + if not text: + self.send_response(400) + self._send_cors() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps({"error": "Missing 'text' or telemetry fields"}).encode("utf-8")) + return + + t_start = time.perf_counter() + agent.reset() + res = agent.run(text, strict=False) + latency_ms = (time.perf_counter() - t_start) * 1000 + + results = res.get("results", []) + first_result = results[0] if results and isinstance(results[0], dict) else {} + action = first_result.get("action", "unclassified") + + calls = res.get("function_calls", []) + tool_name = calls[0].get("name") if calls else "triage_tool" + + out = { + "success": True, + "input_text": text, + "action": action, + "tool": tool_name, + "arguments": first_result, + "confidence": float(res.get("confidence", 0.0)), + "reasoning": res.get("reasoning", ""), + "latency_ms": round(latency_ms, 2), + "model": "Cactus-Compute/needle3" + } + + self.send_response(200) + self._send_cors() + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(json.dumps(out).encode("utf-8")) + else: + self.send_response(404) + self._send_cors() + self.end_headers() + + def log_message(self, format, *args): + # Concise logging + print(f"[needle-sidecar] {args[0]} {args[1]} -> {args[2]}") + + +def run(): + server = ThreadingHTTPServer((HOST, PORT), NeedleHandler) + print(f"[needle-sidecar] Listening on http://{HOST}:{PORT}") + try: + server.serve_forever() + except KeyboardInterrupt: + print("\nShutting down needle sidecar...") + server.server_close() + + +if __name__ == "__main__": + run() diff --git a/cmd/normalized-backfill/main.go b/cmd/normalized-backfill/main.go new file mode 100644 index 0000000..817b5eb --- /dev/null +++ b/cmd/normalized-backfill/main.go @@ -0,0 +1,227 @@ +// normalized-backfill previews or persists deterministic normalized events. +// It is read-only unless -write is explicitly supplied. +package main + +import ( + "context" + "flag" + "fmt" + "log" + "os" + "sort" + "time" + + "git.tornberg.me/go-otel/internal/normalization" + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/ClickHouse/clickhouse-go/v2/lib/driver" +) + +type patternCount struct { + Service string + Kind string + Pattern string + Count uint64 +} + +func main() { + addr := flag.String("addr", envOr("OTEL_CLICKHOUSE_ADDR", "127.0.0.1:9000"), "ClickHouse host:port") + database := flag.String("database", envOr("OTEL_CLICKHOUSE_DB", "otel"), "ClickHouse database") + user := flag.String("user", envOr("OTEL_CLICKHOUSE_USER", "default"), "ClickHouse user") + password := flag.String("password", os.Getenv("OTEL_CLICKHOUSE_PASSWORD"), "ClickHouse password") + limit := flag.Int("limit", 0, "raw envelopes to process; 0 means all") + since := flag.String("since", "", "inclusive UTC window start, RFC3339") + until := flag.String("until", "", "exclusive UTC window end, RFC3339") + batchSize := flag.Int("batch-size", 1000, "normalized events per insert batch") + write := flag.Bool("write", false, "create the table and insert normalized events") + flag.Parse() + if *batchSize < 1 { + log.Fatal("-batch-size must be positive") + } + start, end := parseWindow(*since, *until) + + ctx, cancel := context.WithTimeout(context.Background(), 30*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() + if err := conn.Ping(ctx); err != nil { + log.Fatal(err) + } + if *write { + if err := createSchema(ctx, conn, *database); err != nil { + log.Fatal(err) + } + } + + whereSQL := "" + if !start.IsZero() { + whereSQL = fmt.Sprintf(" timestamp >= toDateTime64('%s', 9, 'UTC')", start.Format("2006-01-02 15:04:05.000000000")) + } + if !end.IsZero() { + if whereSQL != "" { + whereSQL += " AND" + } + whereSQL += fmt.Sprintf(" timestamp < toDateTime64('%s', 9, 'UTC')", end.Format("2006-01-02 15:04:05.000000000")) + } + if whereSQL != "" { + whereSQL = " WHERE" + whereSQL + } + limitSQL := "" + if *limit > 0 { + limitSQL = fmt.Sprintf(" LIMIT %d", *limit) + } + rows, err := conn.Query(ctx, fmt.Sprintf("SELECT timestamp, type, data FROM %s.entries%s%s", *database, whereSQL, limitSQL)) + if err != nil { + log.Fatal(err) + } + defer rows.Close() + + counts := map[string]*patternCount{} + rawCount, eventCount := 0, 0 + var batch driver.Batch + if *write { + batch, err = conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s.normalized_events", *database)) + if err != nil { + log.Fatal(err) + } + } + flush := func() { + if batch != nil { + if err := batch.Send(); err != nil { + log.Fatal(err) + } + batch = nil + } + } + for rows.Next() { + var timestamp time.Time + var entryType, data string + if err := rows.Scan(×tamp, &entryType, &data); err != nil { + log.Fatal(err) + } + rawCount++ + events, err := normalization.NormalizeEntry(entryType, timestamp.UnixNano(), []byte(data)) + if err != nil { + log.Printf("skip %s: %v", entryType, err) + continue + } + for _, event := range events { + eventCount++ + key := event.Service + "\x00" + event.Kind + "\x00" + event.PatternSignature + if counts[key] == nil { + counts[key] = &patternCount{Service: event.Service, Kind: event.Kind, Pattern: event.PatternSignature} + } + counts[key].Count++ + if *write { + if err := appendEvent(batch, event); err != nil { + log.Fatal(err) + } + if eventCount%*batchSize == 0 { + flush() + batch, err = conn.PrepareBatch(ctx, fmt.Sprintf("INSERT INTO %s.normalized_events", *database)) + if err != nil { + log.Fatal(err) + } + } + } + } + } + if err := rows.Err(); err != nil { + log.Fatal(err) + } + flush() + top := make([]*patternCount, 0, len(counts)) + for _, item := range counts { + top = append(top, item) + } + sort.Slice(top, func(i, j int) bool { return top[i].Count > top[j].Count }) + if len(top) > 20 { + top = top[:20] + } + fmt.Printf("raw_envelopes=%d normalized_events=%d unique_patterns=%d write=%t\n", rawCount, eventCount, len(top), *write) + for _, item := range top { + fmt.Printf("pattern_count=%d service=%s kind=%s pattern=%s\n", item.Count, item.Service, item.Kind, item.Pattern) + } +} + +func createSchema(ctx context.Context, conn clickhouse.Conn, database string) error { + query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.normalized_events ( + event_id String, + entry_type LowCardinality(String), + event_index UInt32, + ingest_timestamp DateTime64(9, 'UTC'), + timestamp DateTime64(9, 'UTC'), + service LowCardinality(String), release LowCardinality(String), environment LowCardinality(String), + kind LowCardinality(String), source LowCardinality(String), is_error UInt8, + status_class LowCardinality(String), status_code UInt16, duration_ms Float64, + route String, span_name String, metric_name String, metric_type LowCardinality(String), metric_unit String, + metric_value Float64, metric_count UInt64, metric_sum Float64, metric_min Float64, metric_max Float64, + exception_type String, message_normalized String, stack_signature String, trace_id String, issue_fingerprint String, + is_retry UInt8, is_timeout UInt8, is_cancellation UInt8, + exact_signature String, pattern_signature String, version UInt8 DEFAULT 1 + ) ENGINE = ReplacingMergeTree(version) + PARTITION BY toYYYYMM(timestamp) + ORDER BY (service, kind, pattern_signature, timestamp, event_id) + TTL toDateTime(timestamp) + INTERVAL 180 DAY`, database) + if err := conn.Exec(ctx, query); err != nil { + return err + } + // Existing developer tables predate the explicit model flags. These + // additive migrations preserve historical rows with safe zero defaults. + for _, statement := range []string{ + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS issue_fingerprint String DEFAULT ''", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_retry UInt8 DEFAULT 0", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_timeout UInt8 DEFAULT 0", database), + fmt.Sprintf("ALTER TABLE %s.normalized_events ADD COLUMN IF NOT EXISTS is_cancellation UInt8 DEFAULT 0", database), + } { + if err := conn.Exec(ctx, statement); err != nil { + return err + } + } + return nil +} + +func appendEvent(batch driver.Batch, event normalization.Event) error { + return batch.Append( + event.EventID, event.EntryType, event.EventIndex, time.Unix(0, event.IngestTimestamp), time.Unix(0, event.Timestamp), + event.Service, event.Release, event.Environment, event.Kind, event.Source, boolUint8(event.IsError), + event.StatusClass, event.StatusCode, event.DurationMS, event.Route, event.SpanName, event.MetricName, event.MetricType, event.MetricUnit, + event.MetricValue, event.MetricCount, event.MetricSum, event.MetricMin, event.MetricMax, + event.ExceptionType, event.MessageNormalized, event.StackSignature, event.TraceID, event.IssueFingerprint, + boolUint8(event.IsRetry), boolUint8(event.IsTimeout), boolUint8(event.IsCancellation), event.ExactSignature, event.PatternSignature, + uint8(1), + ) +} + +func boolUint8(value bool) uint8 { + if value { + return 1 + } + return 0 +} +func envOr(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} + +func parseWindow(since, until string) (time.Time, time.Time) { + parse := func(value, name string) time.Time { + if value == "" { + return time.Time{} + } + parsed, err := time.Parse(time.RFC3339Nano, value) + if err != nil { + log.Fatalf("invalid -%s %q: %v", name, value, err) + } + return parsed.UTC() + } + start, end := parse(since, "since"), parse(until, "until") + if !start.IsZero() && !end.IsZero() && !start.Before(end) { + log.Fatal("-since must be before -until") + } + return start, end +} diff --git a/cmd/normalized-preview/main.go b/cmd/normalized-preview/main.go new file mode 100644 index 0000000..db187b4 --- /dev/null +++ b/cmd/normalized-preview/main.go @@ -0,0 +1,199 @@ +// normalized-preview reads a small sample from ClickHouse and prints the +// normalized rows without creating tables or writing any data. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "strconv" + "time" + + "git.tornberg.me/go-otel/internal/normalization" + "github.com/ClickHouse/clickhouse-go/v2" +) + +func main() { + addr := flag.String("addr", envOr("OTEL_CLICKHOUSE_ADDR", "127.0.0.1:9000"), "ClickHouse host:port") + database := flag.String("database", envOr("OTEL_CLICKHOUSE_DB", "otel"), "ClickHouse database") + user := flag.String("user", envOr("OTEL_CLICKHOUSE_USER", "default"), "ClickHouse user") + password := flag.String("password", os.Getenv("OTEL_CLICKHOUSE_PASSWORD"), "ClickHouse password") + limit := flag.Int("limit", 10, "raw envelopes to preview") + typeFilter := flag.String("type", "", "only preview one raw type: traces, logs, or metrics") + httpMode := flag.Bool("http", false, "use ClickHouse HTTP instead of native protocol") + summary := flag.Bool("summary", false, "aggregate by service, kind, and pattern instead of printing events") + flag.Parse() + if *httpMode { + previewHTTP(*addr, *database, *user, *password, *limit, *typeFilter, *summary) + return + } + + conn, err := clickhouse.Open(&clickhouse.Options{Addr: []string{*addr}, Auth: clickhouse.Auth{Database: *database, Username: *user, Password: *password}}) + if err != nil { + log.Fatalf("clickhouse open: %v", err) + } + defer conn.Close() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := conn.Ping(ctx); err != nil { + log.Fatalf("clickhouse ping: %v", err) + } + if *typeFilter != "" && *typeFilter != "traces" && *typeFilter != "logs" && *typeFilter != "metrics" { + log.Fatalf("invalid -type %q", *typeFilter) + } + whereSQL := "" + if *typeFilter != "" { + whereSQL = fmt.Sprintf(" WHERE type = '%s'", *typeFilter) + } + rows, err := conn.Query(ctx, fmt.Sprintf("SELECT timestamp, type, data FROM %s.entries%s ORDER BY timestamp DESC LIMIT ?", *database, whereSQL), *limit) + if err != nil { + log.Fatalf("query entries: %v", err) + } + defer rows.Close() + encoder := json.NewEncoder(os.Stdout) + groups := map[string]*group{} + count := 0 + for rows.Next() { + var timestamp time.Time + var entryType, data string + if err := rows.Scan(×tamp, &entryType, &data); err != nil { + log.Fatalf("scan entry: %v", err) + } + events, err := normalization.NormalizeEntry(entryType, timestamp.UnixNano(), []byte(data)) + if err != nil { + log.Printf("skip %s at %s: %v", entryType, timestamp.Format(time.RFC3339), err) + continue + } + for _, event := range events { + if *summary { + addGroup(groups, event) + } else if err := encoder.Encode(event); err != nil { + log.Fatal(err) + } + count++ + } + } + if err := rows.Err(); err != nil { + log.Fatal(err) + } + if *summary { + printGroups(encoder, groups) + } + log.Printf("previewed %d normalized events from %d raw envelopes", count, *limit) +} + +func previewHTTP(addr, database, user, password string, limit int, typeFilter string, summary bool) { + limitSQL := "" + orderSQL := " ORDER BY timestamp DESC" + whereSQL := "" + if typeFilter != "" { + if typeFilter != "traces" && typeFilter != "logs" && typeFilter != "metrics" { + log.Fatalf("invalid -type %q", typeFilter) + } + whereSQL = fmt.Sprintf(" WHERE type = '%s'", typeFilter) + } + if limit > 0 { + limitSQL = fmt.Sprintf(" LIMIT %d", limit) + } else { + // A full dry run must stream the table. Sorting every raw payload first + // can consume more memory than the ClickHouse query limit allows. + orderSQL = "" + } + query := fmt.Sprintf("SELECT toUnixTimestamp64Nano(timestamp) AS timestamp, type, data FROM %s.entries%s%s%s FORMAT JSONEachRow", database, whereSQL, orderSQL, limitSQL) + req, err := http.NewRequest(http.MethodGet, "http://"+addr+"/?query="+url.QueryEscape(query), nil) + if err != nil { + log.Fatal(err) + } + req.SetBasicAuth(user, password) + response, err := http.DefaultClient.Do(req) + if err != nil { + log.Fatalf("clickhouse HTTP request: %v", err) + } + defer response.Body.Close() + if response.StatusCode/100 != 2 { + body, _ := io.ReadAll(response.Body) + log.Fatalf("clickhouse HTTP %s: %s", response.Status, body) + } + decoder := json.NewDecoder(response.Body) + encoder := json.NewEncoder(os.Stdout) + groups := map[string]*group{} + rawCount, eventCount := 0, 0 + for { + var row struct { + Timestamp json.RawMessage `json:"timestamp"` + Type string `json:"type"` + Data string `json:"data"` + } + if err := decoder.Decode(&row); err == io.EOF { + break + } else if err != nil { + log.Fatal(err) + } + rawCount++ + timestamp := parseTimestamp(row.Timestamp) + events, err := normalization.NormalizeEntry(row.Type, timestamp, []byte(row.Data)) + if err != nil { + log.Printf("skip %s: %v", row.Type, err) + continue + } + for _, event := range events { + if summary { + addGroup(groups, event) + } else if err := encoder.Encode(event); err != nil { + log.Fatal(err) + } + eventCount++ + } + } + if summary { + printGroups(encoder, groups) + } + log.Printf("previewed %d normalized events from %d raw envelopes", eventCount, rawCount) +} + +type group struct { + Service string `json:"service"` + Kind string `json:"kind"` + PatternSignature string `json:"patternSignature"` + Count int `json:"count"` +} + +func addGroup(groups map[string]*group, event normalization.Event) { + key := event.Service + "\x00" + event.Kind + "\x00" + event.PatternSignature + if groups[key] == nil { + groups[key] = &group{Service: event.Service, Kind: event.Kind, PatternSignature: event.PatternSignature} + } + groups[key].Count++ +} + +func printGroups(encoder *json.Encoder, groups map[string]*group) { + for _, item := range groups { + if err := encoder.Encode(item); err != nil { + log.Fatal(err) + } + } +} + +func parseTimestamp(raw json.RawMessage) int64 { + var text string + if json.Unmarshal(raw, &text) == nil { + value, _ := strconv.ParseInt(text, 10, 64) + return value + } + var value int64 + _ = json.Unmarshal(raw, &value) + return value +} + +func envOr(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 7550016..70560c6 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -293,6 +293,8 @@ func main() { // Source-map uploads are intentionally public so the release build script // can upload maps without an OIDC session or bearer token. uiMux.HandleFunc("POST /api/sourcemaps", coll.HandleSourceMap) + uiMux.HandleFunc("GET /api/needle/status", coll.HandleNeedleStatus) + uiMux.HandleFunc("POST /api/needle/classify", coll.HandleNeedleClassify) // Protected API & WebSocket routes uiMux.Handle("/api/history", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleHistory))) @@ -303,6 +305,7 @@ func main() { uiMux.Handle("/api/history/wait", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleWait))) uiMux.Handle("GET /api/traces/{traceId}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleTraceByID))) uiMux.Handle("/api/issues", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssues))) + uiMux.Handle("POST /api/ml/test-spike", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleMLDetectorTest))) uiMux.Handle("GET /api/issues/{id}", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueByID))) uiMux.Handle("GET /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleIssueSimilar))) uiMux.Handle("POST /api/issues/{id}/similar", oidcAuth.RequireAuth(http.HandlerFunc(coll.HandleSimilarityFeedback))) @@ -357,9 +360,13 @@ func main() { } }() + httpPort := os.Getenv("PORT") + if httpPort == "" { + httpPort = "8080" + } uiHandler := honeypotTracker.Middleware(uiMux) - if err := http.ListenAndServe(":8080", uiHandler); err != nil { - log.Fatalf("failed to start HTTP server: %v", err) + if err := http.ListenAndServe(":"+httpPort, uiHandler); err != nil { + log.Fatalf("failed to start HTTP server on :%s: %v", httpPort, err) } } func spaHandler(root http.FileSystem) http.Handler { diff --git a/cmd/training-sample/main.go b/cmd/training-sample/main.go new file mode 100644 index 0000000..469c8de --- /dev/null +++ b/cmd/training-sample/main.go @@ -0,0 +1,130 @@ +// training-sample exports a bounded, time-windowed JSONL dataset from +// normalized_events. It never writes to ClickHouse. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log" + "os" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" +) + +type sample struct { + EventID string `json:"eventId"` + Timestamp time.Time `json:"timestamp"` + Service string `json:"service"` + Release string `json:"release,omitempty"` + Kind string `json:"kind"` + Source string `json:"source"` + IsError uint8 `json:"isError"` + StatusClass string `json:"statusClass,omitempty"` + StatusCode uint16 `json:"statusCode,omitempty"` + DurationMS float64 `json:"durationMs,omitempty"` + Route string `json:"route,omitempty"` + SpanName string `json:"spanName,omitempty"` + MetricName string `json:"metricName,omitempty"` + MetricType string `json:"metricType,omitempty"` + MetricValue float64 `json:"metricValue,omitempty"` + MetricCount uint64 `json:"metricCount,omitempty"` + MetricSum float64 `json:"metricSum,omitempty"` + MetricMin float64 `json:"metricMin,omitempty"` + MetricMax float64 `json:"metricMax,omitempty"` + ExceptionType string `json:"exceptionType,omitempty"` + Message string `json:"messageNormalized,omitempty"` + StackSignature string `json:"stackSignature,omitempty"` + TraceID string `json:"traceId,omitempty"` + IssueFingerprint string `json:"issueFingerprint,omitempty"` + IsRetry uint8 `json:"isRetry"` + IsTimeout uint8 `json:"isTimeout"` + IsCancellation uint8 `json:"isCancellation"` + ExactSignature string `json:"exactSignature"` + PatternSignature string `json:"patternSignature"` +} + +func main() { + addr := flag.String("addr", envOr("OTEL_CLICKHOUSE_ADDR", "127.0.0.1:9000"), "ClickHouse host:port") + database := flag.String("database", envOr("OTEL_CLICKHOUSE_DB", "otel"), "ClickHouse database") + user := flag.String("user", envOr("OTEL_CLICKHOUSE_USER", "default"), "ClickHouse user") + password := flag.String("password", os.Getenv("OTEL_CLICKHOUSE_PASSWORD"), "ClickHouse password") + since := flag.String("since", "", "inclusive UTC RFC3339 timestamp") + until := flag.String("until", "", "exclusive UTC RFC3339 timestamp") + maxPerPattern := flag.Int("max-per-pattern", 25, "maximum examples per pattern") + maxEvents := flag.Int("max-events", 10000, "maximum exported events") + output := flag.String("output", "", "write JSONL to this path instead of stdout") + flag.Parse() + if *since == "" || *until == "" || *maxPerPattern < 1 || *maxEvents < 1 { + log.Fatal("-since, -until, -max-per-pattern, and -max-events are required and must be valid") + } + start, err := time.Parse(time.RFC3339Nano, *since) + if err != nil { + log.Fatal(err) + } + end, err := time.Parse(time.RFC3339Nano, *until) + if err != nil || !start.Before(end) { + log.Fatal("invalid training window") + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*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() + if err := conn.Ping(ctx); err != nil { + log.Fatal(err) + } + writer := os.Stdout + if *output != "" { + file, err := os.Create(*output) + if err != nil { + log.Fatal(err) + } + defer file.Close() + writer = file + } + query := fmt.Sprintf(`SELECT event_id, timestamp, service, release, kind, source, is_error, status_class, status_code, duration_ms, + route, span_name, metric_name, metric_type, metric_value, metric_count, metric_sum, metric_min, metric_max, + exception_type, message_normalized, stack_signature, trace_id, issue_fingerprint, is_retry, is_timeout, is_cancellation, exact_signature, pattern_signature + FROM %s.normalized_events FINAL + WHERE timestamp >= toDateTime64('%s', 9, 'UTC') AND timestamp < toDateTime64('%s', 9, 'UTC') + ORDER BY pattern_signature, timestamp + LIMIT %d BY pattern_signature + LIMIT %d`, *database, clickhouseTime(start), clickhouseTime(end), *maxPerPattern, *maxEvents) + rows, err := conn.Query(ctx, query) + if err != nil { + log.Fatal(err) + } + defer rows.Close() + encoder := json.NewEncoder(writer) + count := 0 + for rows.Next() { + var item sample + if err := rows.Scan(&item.EventID, &item.Timestamp, &item.Service, &item.Release, &item.Kind, &item.Source, &item.IsError, &item.StatusClass, &item.StatusCode, &item.DurationMS, &item.Route, &item.SpanName, &item.MetricName, &item.MetricType, &item.MetricValue, &item.MetricCount, &item.MetricSum, &item.MetricMin, &item.MetricMax, &item.ExceptionType, &item.Message, &item.StackSignature, &item.TraceID, &item.IssueFingerprint, &item.IsRetry, &item.IsTimeout, &item.IsCancellation, &item.ExactSignature, &item.PatternSignature); err != nil { + log.Fatal(err) + } + if err := encoder.Encode(item); err != nil { + log.Fatal(err) + } + count++ + } + if err := rows.Err(); err != nil { + log.Fatal(err) + } + log.Printf("exported %d training events", count) +} + +func clickhouseTime(value time.Time) string { + return value.UTC().Format("2006-01-02 15:04:05.000000000") +} +func envOr(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/cmd/window-features/main.go b/cmd/window-features/main.go new file mode 100644 index 0000000..19a8c8a --- /dev/null +++ b/cmd/window-features/main.go @@ -0,0 +1,104 @@ +// window-features exports fixed-width service/release feature rows and the +// documented future error-spike labels from a normalized JSONL export. +package main + +import ( + "bufio" + "encoding/json" + "flag" + "fmt" + "log" + "os" + "time" + + "git.tornberg.me/go-otel/internal/features" +) + +type event struct { + Timestamp time.Time `json:"timestamp"` + Service string `json:"service"` + Release string `json:"release"` + Kind string `json:"kind"` + Source string `json:"source"` + IsError uint8 `json:"isError"` + StatusClass string `json:"statusClass"` + DurationMS float64 `json:"durationMs"` + TraceID string `json:"traceId"` + PatternSignature string `json:"patternSignature"` + IssueFingerprint string `json:"issueFingerprint"` + IsRetry uint8 `json:"isRetry"` + IsTimeout uint8 `json:"isTimeout"` + IsCancellation uint8 `json:"isCancellation"` + Message string `json:"messageNormalized"` + SpanName string `json:"spanName"` +} + +func main() { + input := flag.String("input", "", "normalized JSONL input") + window := flag.Duration("window", 5*time.Minute, "observation window") + step := flag.Duration("step", 5*time.Minute, "window start interval") + future := flag.Duration("future", 15*time.Minute, "future prediction interval") + until := flag.String("until", "", "exclusive source-export watermark, RFC3339; required to label a quiet tail safely") + baselineDays := flag.Int("baseline-days", 7, "same-time-of-day trailing baseline days") + minBaselineSamples := flag.Int("min-baseline-samples", 7, "minimum nonempty baseline windows required for a label") + spikeMultiple := flag.Float64("spike-multiple", 3, "future/baseline error-rate ratio for a positive label") + minFutureErrors := flag.Int("min-future-errors", 5, "minimum future errors for a positive label") + includeUnknown := flag.Bool("include-unknown-labels", false, "emit rows lacking a complete baseline (default omits them)") + flag.Parse() + if *input == "" || *window <= 0 || *step <= 0 || *future <= 0 || *baselineDays < 1 || *minBaselineSamples < 1 || *spikeMultiple <= 0 || *minFutureErrors < 1 { + log.Fatal("invalid feature configuration") + } + var observedUntil time.Time + if *until != "" { + parsed, err := time.Parse(time.RFC3339Nano, *until) + if err != nil { + log.Fatalf("invalid -until %q: %v", *until, err) + } + observedUntil = parsed.UTC() + } + + items := readEvents(*input) + rows := features.Build(items, features.Config{ + Window: *window, Step: *step, Future: *future, ObservedUntil: observedUntil, BaselineDays: *baselineDays, + MinBaselineSamples: *minBaselineSamples, SpikeMultiple: *spikeMultiple, MinFutureErrors: *minFutureErrors, + }) + encoder := json.NewEncoder(os.Stdout) + count := 0 + for _, row := range rows { + if !*includeUnknown && !row.LabelKnown { + continue + } + if err := encoder.Encode(row); err != nil { + log.Fatal(err) + } + count++ + } + fmt.Fprintf(os.Stderr, "built %d feature rows from %d events\n", count, len(items)) +} + +func readEvents(path string) []features.Event { + file, err := os.Open(path) + if err != nil { + log.Fatal(err) + } + defer file.Close() + result := []features.Event{} + scanner := bufio.NewScanner(file) + scanner.Buffer(make([]byte, 64*1024), 4*1024*1024) + for scanner.Scan() { + var item event + if err := json.Unmarshal(scanner.Bytes(), &item); err != nil { + log.Fatal(err) + } + result = append(result, features.Event{ + Timestamp: item.Timestamp, Service: item.Service, Release: item.Release, Kind: item.Kind, Source: item.Source, + 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, + }) + } + if err := scanner.Err(); err != nil { + log.Fatal(err) + } + return result +} diff --git a/docs/ML.md b/docs/ML.md index 9f01ae0..fefa71e 100644 --- a/docs/ML.md +++ b/docs/ML.md @@ -1,5 +1,86 @@ # Error detection model +## Current handoff (2026-09-18) + +The first iteration is intentionally explainable and deterministic. Raw OTLP +events remain the source of truth; normalized rows are stored in +`otel.normalized_events` with deterministic `event_id`, `exact_signature`, and +`pattern_signature` values. + +Implemented commands: + +```bash +# Read-only normalization summary for a bounded raw window. +go run ./cmd/normalized-backfill \ + -since 2026-09-18T17:00:00Z \ + -until 2026-09-18T18:00:00Z + +# Explicit normalized-table write; use one-hour windows. +go run ./cmd/normalized-backfill \ + -since 2026-09-18T17:00:00Z \ + -until 2026-09-18T18:00:00Z -write + +# Export a capped training sample. +go run ./cmd/training-sample \ + -since 2026-09-18T17:00:00Z \ + -until 2026-09-18T22:00:00Z \ + -max-per-pattern 10 -max-events 1000 \ + -output /tmp/training-v001.jsonl + +# Train and score the explainable prototype model. +go run ./cmd/issue-model -train /tmp/training-v001.jsonl \ + -model /tmp/issue-model-v001.json +go run ./cmd/issue-model -model /tmp/issue-model-v001.json \ + -predict /tmp/sample.jsonl -min-score 0.75 -min-margin 0.05 + +# Inspect rolling features (the five-hour sample is too short to label). +go run ./cmd/window-features -input /tmp/training-v001.jsonl \ + -window 5m -step 5m -future 15m \ + -until 2026-09-18T22:00:00Z -include-unknown-labels +``` + +`window-features` emits one fixed-width row per service/release/window. It +includes counts, error rate, span duration percentiles, kind and HTTP-status +counts, unique trace and issue counts, frontend/backend error counts, and +retry/timeout/cancellation counts. Its label is generated from the following +fifteen minutes and a same-time-of-day trailing baseline. Rows without seven +available baseline days or a complete future interval are withheld by default; +pass `-include-unknown-labels` only for inspection, never as training labels. +Supply the source export's exclusive `-until` watermark so a quiet final +interval can be labelled without relying on the timestamp of its last event. + +For fast candidate/persistence analysis, override the observation and step +durations explicitly (for example `-window 15s -step 5s`). Those short rows +are not substitutes for the documented five-minute spike-prediction label. + +The current prototype was trained on 17:00–22:00 UTC and tested on a separate +05:00–06:00 UTC sample. It achieved 45% top-1 accuracy against deterministic +weak labels. This is a pipeline check, not a production model evaluation. + +The default similarity gate is provisional: score `>= 0.75` and top/runner-up +margin `>= 0.05`. Recalibrate it after human labels are available; select the +threshold for the desired precision, not by optimizing accuracy. + +Use 15 seconds with a 5-second step as the fast window for the observed +five-second polling pattern. Add 30-second confirmation and five-minute trend +windows. A single event is a candidate; persistence across windows makes it +an issue. + +## Next session + +1. Export a clean five-hour JSONL sample and review 20–50 predictions. +2. Add `same`, `different`, and `unknown/noise` labels in the UI. +3. Add persistence and rollout proximity to the window aggregates; count, + error count/rate, unique traces, issue counts, latency, sources and status + classes are already exported. +4. Re-evaluate the threshold on reviewed labels. +5. Keep the model in shadow mode until it beats the deterministic baseline. + +Do not run an unrestricted backfill from a workstation port-forward. Use +one-hour windows or an authenticated in-cluster job. The normalized table has +some smoke-test rows; always select training data by an explicit timestamp +window. + ## Current decision Keep the original OTLP payloads as the source of truth, but derive a stable @@ -40,6 +121,11 @@ PII-bearing fields such as raw messages, URLs, user agents, source IPs, and stack traces must not be model features. They can remain in the raw history and issue views subject to the existing retention and access controls. +The normalized row now carries the canonical `issue_fingerprint` for errors, +plus structured `is_retry`, `is_timeout`, and `is_cancellation` flags. The +flags are derived from OTLP attributes and protocol status—not message text—so +the corresponding aggregate features do not depend on raw user content. + ## First aggregate feature row For each service/release/window, derive: @@ -58,6 +144,15 @@ The model input should be numeric and fixed-width. Categorical service and release values can initially be filtered per service; one global model can be considered after there is enough multi-service data. +## UI verification + +The **Issues** view has a **Test spike detector** button. It calls the +authenticated `POST /api/ml/test-spike` endpoint, which creates no telemetry +and writes nothing to storage. It runs a deterministic scenario through the +same feature and label builder: seven daily 10% baseline windows followed by a +future interval containing five errors in ten events. A passing result shows +the baseline, future counts/rate, and the positive spike label in the UI. + ## Finding similar errors and spans Similarity should be a separate concern from the primary issue fingerprint. diff --git a/index.html b/index.html index 00e424e..2ec9bdb 100644 --- a/index.html +++ b/index.html @@ -141,9 +141,45 @@ Clear Logs + +
+

Needle 3 AI Telemetry Triage

+

Test on-device telemetry classification with Cactus-Compute/needle3:

+
+ + +
+
+ Click "Classify with Needle 3" to test the sidecar... +
+