From 948aadbe577e37a59e552982326424a0a81639ca Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Mon, 21 Sep 2026 12:07:25 -0700 Subject: [PATCH 1/2] fix(intelligence): score error-rate spikes against the rate's own variance The z-score divides a difference of rates, so its denominator has to be the scatter of the rate from bucket to bucket. It was STDDEV(CASE WHEN status IN ('STATUS_CODE_ERROR','ERROR') THEN 1.0 ELSE 0.0 END) over raw spans, which is the scatter of individual span outcomes -- sqrt(p(1-p)), about 0.34 at a 14% error rate. Those are different quantities, and the mismatch punished exactly the services worth watching: the noisier the service, the larger the denominator. At the 13.8% baseline the live demo actually ran, clearing the 2.0 threshold required the rate to jump 69 percentage points. A tripling from 13.8% to 41% scored 0.80. The denominator is now the standard deviation of the per-bucket error rate across the baseline window, and both windows average over the same 5-minute buckets -- the same asymmetry the volume detector had. It is also floored, because a healthy service's error rate is flat at zero and its bucket-to-bucket stddev is therefore exactly zero. Dividing by that hit the CASE's 0.0 fallback, so the service that had just started failing was the one that could not alert: going from no errors to 42% scored 0.00. One percentage point is the least noise worth assuming. It keeps a single stray error in three thousand spans at z=0.03 while a real break reaches z=42. Measured on the fixtures in error_rate_sql_test.go, before -> after: 13.8% -> 41% (noisy baseline) 0.80 -> 27.40 now fires 13.8% -> 13.8% (noisy baseline) 0.01 -> 0.20 stays quiet 0% -> 42% (flat baseline) 0.00 -> 42.33 now fires 0% -> 0.03% (one stray error) 0.00 -> 0.03 stays quiet Verified on the demo across five detector cycles: 1-2 anomalies each, health 85-95, no errors. The concern that a more sensitive denominator would produce an alert flood does not survive contact with the data. --- internal/intelligence/detector.go | 89 +++++++++---- internal/intelligence/error_rate_sql_test.go | 130 +++++++++++++++++++ 2 files changed, 192 insertions(+), 27 deletions(-) create mode 100644 internal/intelligence/error_rate_sql_test.go diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go index 10de6c5a..3bf09abf 100644 --- a/internal/intelligence/detector.go +++ b/internal/intelligence/detector.go @@ -148,37 +148,61 @@ func (d *Detector) detectAnomalies(ctx context.Context, start, end time.Time) [] return anomalies } -// detectErrorRateAnomalies detects error rate spikes -func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time.Time) []Anomaly { - startNano := start.UnixNano() - endNano := end.UnixNano() - namespace := d.duck.DefaultNamespace() - scope := detectorScopeClause(namespace) - - // Compare current error rate to baseline (previous period) - sql := fmt.Sprintf(` +// errorRateAnomalySQL compares the error rate in [startNano, endNano) against +// the window of equal length immediately before it. +// +// The z-score divides a difference of rates, so its denominator has to be the +// scatter of the RATE from bucket to bucket. It used to be +// STDDEV(CASE WHEN status = error THEN 1.0 ELSE 0.0 END) over raw spans, which +// is the scatter of individual span outcomes -- sqrt(p(1-p)), about 0.34 at a +// 14% error rate. Those are different quantities, and the mismatch punished +// exactly the services worth watching: the noisier the service, the larger the +// denominator, so at the 13.8% baseline the live demo ran, clearing the 2.0 +// threshold needed the rate to jump 69 percentage points. A tripling to 41% +// scored 0.80. +// +// The denominator is floored because a healthy service's rate is flat at zero, +// giving it a bucket-to-bucket stddev of exactly zero -- so the service that +// just started failing was the one that could not alert (z=0.00 going from no +// errors to 42%). One percentage point is the least noise worth assuming: it +// keeps a single stray error in a few thousand spans below the threshold while +// letting a real break through. +// minErrorRateStddev floors the error-rate z-score denominator at one +// percentage point. See errorRateAnomalySQL. +const minErrorRateStddev = 0.01 + +func errorRateAnomalySQL(startNano, endNano int64, scope string) string { + return fmt.Sprintf(` WITH current_period AS ( SELECT - service as service_name, - COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count, - COUNT(*) AS total_count, - (COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate - FROM spans - WHERE start_unix_nano >= %d AND start_unix_nano < %d - %s - GROUP BY service + service_name, + AVG(bucket_rate) AS error_rate + FROM ( + SELECT + service as service_name, + (COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS bucket_rate + FROM spans + WHERE start_unix_nano >= %d AND start_unix_nano < %d + %s + GROUP BY service, time_bucket(INTERVAL '5 minutes', start_time) + ) buckets + GROUP BY service_name ), baseline_period AS ( SELECT - service as service_name, - COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count, - COUNT(*) AS total_count, - (COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate, - STDDEV(CASE WHEN status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_stddev - FROM spans - WHERE start_unix_nano >= %d AND start_unix_nano < %d - %s - GROUP BY service + service_name, + AVG(bucket_rate) AS error_rate, + GREATEST(COALESCE(STDDEV(bucket_rate), 0.0), %f) AS error_stddev + FROM ( + SELECT + service as service_name, + (COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS bucket_rate + FROM spans + WHERE start_unix_nano >= %d AND start_unix_nano < %d + %s + GROUP BY service, time_bucket(INTERVAL '5 minutes', start_time) + ) buckets + GROUP BY service_name ) SELECT c.service_name, @@ -191,7 +215,18 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time FROM current_period c LEFT JOIN baseline_period b ON c.service_name = b.service_name WHERE c.error_rate > 0 - `, startNano, endNano, scope, startNano-endNano+startNano, startNano, scope) + `, startNano, endNano, scope, minErrorRateStddev, startNano-endNano+startNano, startNano, scope) +} + +// detectErrorRateAnomalies detects error rate spikes +func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time.Time) []Anomaly { + startNano := start.UnixNano() + endNano := end.UnixNano() + namespace := d.duck.DefaultNamespace() + scope := detectorScopeClause(namespace) + + // Compare current error rate to baseline (previous period) + sql := errorRateAnomalySQL(startNano, endNano, scope) resp := d.duck.ExecuteSQL(ctx, query.SQLRequest{Query: sql}) if resp.Error != "" { diff --git a/internal/intelligence/error_rate_sql_test.go b/internal/intelligence/error_rate_sql_test.go new file mode 100644 index 00000000..b324343e --- /dev/null +++ b/internal/intelligence/error_rate_sql_test.go @@ -0,0 +1,130 @@ +package intelligence + +import ( + "database/sql" + "math" + "testing" + "time" +) + +// errorRateFixture writes six 5-minute buckets: the first three are the +// baseline window, the last three the current one. Each entry is a bucket's +// error percentage, so a service's whole history is one readable slice. +func errorRateFixture(t *testing.T, db *sql.DB, start time.Time, services map[string][]float64) { + t.Helper() + if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS spans ( + service TEXT, namespace TEXT, status TEXT, start_time TIMESTAMP, start_unix_nano BIGINT + )`); err != nil { + t.Fatal(err) + } + const perBucket = 1000 + for service, rates := range services { + for bucket, rate := range rates { + at := start.Add(time.Duration(bucket) * 5 * time.Minute) + errors := int(math.Round(rate * perBucket)) + for i := range perBucket { + status := "STATUS_CODE_OK" + if i < errors { + status = "STATUS_CODE_ERROR" + } + ts := at.Add(time.Duration(i) * time.Millisecond) + if _, err := db.Exec(`INSERT INTO spans VALUES (?, 'default', ?, ?, ?)`, + service, status, ts, ts.UnixNano()); err != nil { + t.Fatal(err) + } + } + } + } +} + +func errorRateScores(t *testing.T, db *sql.DB, start, end time.Time) map[string]float64 { + t.Helper() + rows, err := db.Query(errorRateAnomalySQL(start.UnixNano(), end.UnixNano(), "")) + if err != nil { + t.Fatalf("error rate query: %v", err) + } + defer rows.Close() + scores := map[string]float64{} + for rows.Next() { + var service string + var current, baseline, zScore float64 + if err := rows.Scan(&service, ¤t, &baseline, &zScore); err != nil { + t.Fatal(err) + } + scores[service] = zScore + t.Logf("%-14s current=%.3f baseline=%.3f z=%.2f", service, current, baseline, zScore) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + return scores +} + +// The z-score divided a difference of RATES by the STDDEV of a per-span 0/1 +// indicator. That indicator's spread is sqrt(p(1-p)) -- the scatter of +// individual span outcomes, about 0.34 at a 14% error rate -- and not the +// scatter of the rate itself from bucket to bucket, which is what a z-score on +// a rate needs. The units did not match, and the mismatch got worse the noisier +// the service: at the 13.8% baseline the live demo actually ran, clearing the +// 2.0 threshold required the rate to jump by 69 percentage points. +// +// So the services most worth watching were the ones least able to alert. +const errorRateThreshold = 2.0 + +func TestErrorRateFiresOnASpikeFromANoisyBaseline(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC) + start := end.Add(-15 * time.Minute) + errorRateFixture(t, db, start.Add(-15*time.Minute), map[string][]float64{ + // Sat at 13.8% and tripled. The old denominator scored this 0.47. + "spiking-noisy": {0.13, 0.14, 0.138, 0.40, 0.42, 0.41}, + // Same noisy baseline, no spike. Must stay quiet. + "steady-noisy": {0.13, 0.14, 0.138, 0.135, 0.142, 0.137}, + }) + + scores := errorRateScores(t, db, start, end) + if z := scores["spiking-noisy"]; math.Abs(z) < errorRateThreshold { + t.Errorf("a service that went from 13.8%% to 41%% errors scored z=%.2f, below the %.1f threshold", z, errorRateThreshold) + } + if z := scores["steady-noisy"]; math.Abs(z) >= errorRateThreshold { + t.Errorf("a service holding steady at 13.8%% scored z=%.2f and would alert", z) + } +} + +// A healthy service has a flat 0% error rate, so the bucket-to-bucket stddev of +// its rate is exactly zero. Dividing by it yields the CASE's 0.0 fallback, and +// the service that just started failing is the one that cannot alert. The +// denominator needs a floor. +func TestErrorRateFiresWhenAFlatlineBreaks(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC) + start := end.Add(-15 * time.Minute) + errorRateFixture(t, db, start.Add(-15*time.Minute), map[string][]float64{ + "clean-then-broken": {0, 0, 0, 0.40, 0.45, 0.42}, + "clean-throughout": {0, 0, 0, 0, 0, 0}, + // One stray error in a thousand spans is not an incident. + "clean-with-a-blip": {0, 0, 0, 0, 0.001, 0}, + }) + + scores := errorRateScores(t, db, start, end) + if z := scores["clean-then-broken"]; math.Abs(z) < errorRateThreshold { + t.Errorf("a service that went from no errors to 42%% scored z=%.2f, below the %.1f threshold", z, errorRateThreshold) + } + if z, ok := scores["clean-with-a-blip"]; ok && math.Abs(z) >= errorRateThreshold { + t.Errorf("one error in three thousand spans scored z=%.2f and would alert", z) + } + // A service with no errors at all is filtered out by `WHERE c.error_rate > 0`. + if z, ok := scores["clean-throughout"]; ok && math.Abs(z) >= errorRateThreshold { + t.Errorf("a service with no errors scored z=%.2f", z) + } +} From d880396eaf623eb82318b628503f83ab3d581612 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Mon, 21 Sep 2026 12:24:04 -0700 Subject: [PATCH 2/2] fix(observability,ingest): stop MCP paying for the browser's comparison grid service_performance answers a question about one service. It also returned the latency heatmap -- the twelve busiest services by every bucket in the window -- because the browser's performance page draws that grid and both callers shared one method. On the live demo that was about 85% of a 113 KB response describing twelve services the caller had not asked about, plus the second DuckDB query that built it. The grid is now opt-in through PerformanceOptions. The browser asks for it; the MCP tool does not, and skips the query rather than blanking the field afterwards, so the cost actually goes away. Also reattaches three doc comments that had been separated from what they document, and adds the guard that found the third. A declaration inserted between a doc comment and its subject is invisible to gofmt, go vet and the compiler: the new declaration inherits a comment describing something else and the original is left undocumented. errorRateAnomalySQL's doc had landed on minErrorRateStddev, ParquetStore.Trace's on TraceTotals, and attrsJSON's on attrBufPool -- where it had also gone stale, still promising a nil return from a function that returns "". Requiring every doc comment to begin with its own declaration's name would flag hundreds that simply do not follow that convention, and a guard that is mostly false positives gets muted; the first cut of this one flagged 17, nearly all of them legitimate. It checks the exact signature instead -- a doc comment whose first word names the declaration immediately below it -- which is what such an insertion produces and what nobody writes on purpose. One flag across 2,378 declarations, and it was real. The first version of that guard did not catch the bug it was written for: it walked only documented declarations, so it compared against the next documented one and stepped straight over the undocumented victim. It now walks every declaration in source order, verified by reintroducing the defect. --- internal/agent/tools_integration_test.go | 2 +- internal/api/observability.go | 6 +- internal/api/observability_test.go | 4 +- internal/cmd/lintdocs/doc_placement_test.go | 124 ++++++++++++++++++ internal/ingest/server.go | 18 +-- internal/intelligence/detector.go | 8 +- internal/mcp/server.go | 4 +- internal/mcp/server_test.go | 2 +- internal/observability/performance.go | 44 +++++-- .../observability/performance_heatmap_test.go | 84 ++++++++++++ internal/observability/service_test.go | 2 +- internal/telemetry/parquet.go | 7 +- 12 files changed, 267 insertions(+), 38 deletions(-) create mode 100644 internal/cmd/lintdocs/doc_placement_test.go create mode 100644 internal/observability/performance_heatmap_test.go diff --git a/internal/agent/tools_integration_test.go b/internal/agent/tools_integration_test.go index d4d1e408..bacd6575 100644 --- a/internal/agent/tools_integration_test.go +++ b/internal/agent/tools_integration_test.go @@ -25,7 +25,7 @@ func (registryQueries) Topology(context.Context, observability.Scope, int) (obse return observability.Result[observability.Topology]{}, nil } -func (registryQueries) Performance(context.Context, observability.Scope, string, int) (observability.Result[observability.Performance], error) { +func (registryQueries) Performance(context.Context, observability.Scope, observability.PerformanceOptions) (observability.Result[observability.Performance], error) { return observability.Result[observability.Performance]{}, nil } diff --git a/internal/api/observability.go b/internal/api/observability.go index fd8bda4b..7da6cae1 100644 --- a/internal/api/observability.go +++ b/internal/api/observability.go @@ -15,7 +15,7 @@ import ( type ObservabilityQueries interface { Overview(context.Context, observability.Scope, int) (observability.Result[observability.Overview], error) Topology(context.Context, observability.Scope, int) (observability.Result[observability.Topology], error) - Performance(context.Context, observability.Scope, string, int) (observability.Result[observability.Performance], error) + Performance(context.Context, observability.Scope, observability.PerformanceOptions) (observability.Result[observability.Performance], error) Trace(context.Context, observability.Scope, string, string, int) (observability.Result[observability.TraceDetail], error) Logs(context.Context, observability.Scope, string, string, string, int) (observability.Result[observability.Logs], error) } @@ -72,7 +72,9 @@ func (h *ObservabilityHandler) performance(c *echo.Context) error { if err != nil { return err } - result, err := h.queries.Performance(c.Request().Context(), scope, c.QueryParam("service"), limit) + result, err := h.queries.Performance(c.Request().Context(), scope, observability.PerformanceOptions{ + Service: c.QueryParam("service"), Limit: limit, Heatmap: true, + }) if err != nil { return mapQueryError(err) } diff --git a/internal/api/observability_test.go b/internal/api/observability_test.go index bebfe451..c6023eaf 100644 --- a/internal/api/observability_test.go +++ b/internal/api/observability_test.go @@ -35,7 +35,7 @@ func (f *fakeQueries) Topology(_ context.Context, scope observability.Scope, _ i }, nil } -func (f *fakeQueries) Performance(_ context.Context, _ observability.Scope, _ string, _ int) (observability.Result[observability.Performance], error) { +func (f *fakeQueries) Performance(_ context.Context, _ observability.Scope, _ observability.PerformanceOptions) (observability.Result[observability.Performance], error) { return observability.Result[observability.Performance]{Schema: observability.PerformanceSchema}, nil } @@ -151,7 +151,7 @@ func (p deadlineProbe) Topology(ctx context.Context, scope observability.Scope, return observability.Result[observability.Topology]{}, nil } -func (p deadlineProbe) Performance(ctx context.Context, scope observability.Scope, service string, limit int) (observability.Result[observability.Performance], error) { +func (p deadlineProbe) Performance(ctx context.Context, scope observability.Scope, opts observability.PerformanceOptions) (observability.Result[observability.Performance], error) { p.note(ctx, "/api/observability/performance") return observability.Result[observability.Performance]{}, nil } diff --git a/internal/cmd/lintdocs/doc_placement_test.go b/internal/cmd/lintdocs/doc_placement_test.go new file mode 100644 index 00000000..489727eb --- /dev/null +++ b/internal/cmd/lintdocs/doc_placement_test.go @@ -0,0 +1,124 @@ +package lintdocs + +import ( + "go/ast" + "go/parser" + "go/token" + "os" + "path/filepath" + "strings" + "testing" +) + +// Inserting a declaration between a doc comment and the thing it documents is +// invisible to gofmt, go vet and the compiler, and it silently reassigns the +// comment: the new declaration inherits a block describing something else, and +// the original is left undocumented. It happened three times in this tree, twice +// in one afternoon, every time from a scripted edit that matched on the +// declaration line and inserted above it. +// +// Requiring every doc comment to start with its own declaration's name would +// flag hundreds that simply do not follow that convention, and a guard that is +// mostly false positives gets muted. This checks the exact signature of the bug +// instead: a doc comment whose first word names the declaration immediately +// BELOW the one it is attached to. That is what such an insertion produces, and +// nobody writes it on purpose. +func TestDocCommentsDocumentWhatTheySitOn(t *testing.T) { + root := "../../.." + fset := token.NewFileSet() + + // Every declaration in source order, documented or not. Tracking only the + // documented ones would compare against the next *documented* declaration + // and skip straight past the undocumented victim -- which is precisely the + // declaration this is looking for. + type decl struct { + file string + line int + name string + doc string + } + var decls []decl + + err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return err + } + if info.IsDir() { + switch info.Name() { + case ".git", "node_modules", "vendor", "dist", "site", "experiments": + return filepath.SkipDir + } + return nil + } + if !strings.HasSuffix(path, ".go") { + return nil + } + parsed, err := parser.ParseFile(fset, path, nil, parser.ParseComments) + if err != nil { + return nil // not this guard's job to report unparseable files + } + add := func(name string, doc *ast.CommentGroup, pos token.Pos) { + text := "" + if doc != nil { + text = doc.Text() + } + decls = append(decls, decl{file: path, line: fset.Position(pos).Line, name: name, doc: text}) + } + for _, d := range parsed.Decls { + switch d := d.(type) { + case *ast.FuncDecl: + add(d.Name.Name, d.Doc, d.Pos()) + case *ast.GenDecl: + // A grouped block is one declaration: its doc describes the + // block, which conventionally means its first name. + for si, spec := range d.Specs { + var name string + switch spec := spec.(type) { + case *ast.TypeSpec: + name = spec.Name.Name + case *ast.ValueSpec: + if len(spec.Names) > 0 { + name = spec.Names[0].Name + } + } + if name == "" { + continue + } + if si == 0 { + add(name, d.Doc, d.Pos()) + } else { + add(name, nil, spec.Pos()) + } + } + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + if len(decls) < 500 { + t.Fatalf("only %d declarations found; this guard is not looking at the right tree", len(decls)) + } + + documented, flagged := 0, 0 + for i, d := range decls { + if d.doc == "" { + continue + } + documented++ + if i+1 >= len(decls) || decls[i+1].file != d.file { + continue + } + first, _, _ := strings.Cut(strings.TrimSpace(d.doc), " ") + first = strings.TrimRight(first, ".,:") + if first == d.name || first != decls[i+1].name { + continue + } + flagged++ + t.Errorf("%s:%d: the doc comment on %s describes %s, which is declared immediately below it\n"+ + "a declaration was inserted between that comment and what it documents", + d.file, d.line, d.name, first) + } + t.Logf("checked %d declarations, %d documented, %d flagged", len(decls), documented, flagged) +} diff --git a/internal/ingest/server.go b/internal/ingest/server.go index 488b2d1d..648e1a1b 100644 --- a/internal/ingest/server.go +++ b/internal/ingest/server.go @@ -373,16 +373,18 @@ func toJSON(v interface{}) string { return string(b) } -// attrsJSON flattens an OTLP attribute list into a flat JSON object keyed by the -// literal (dotted) attribute name, e.g. {"http.method":"GET","http.status_code":200}. -// This is the shape the attr() macro and json_extract_string(col, '$."key"') paths -// expect. Marshaling the raw []*KeyValue (as the old toJSON path did) produced an -// array of {Key,Value} structs that no JSON-path query could read. Returns nil for -// an empty list so the column stays NULL. +// attrBufPool holds the buffers attrsJSON's fast path writes into. var attrBufPool = sync.Pool{New: func() any { return new(bytes.Buffer) }} -// attrsJSON flattens an OTLP attribute list into a flat JSON object. The fast -// path writes the object directly into a pooled buffer, skipping the +// attrsJSON flattens an OTLP attribute list into a flat JSON object keyed by +// the literal (dotted) attribute name, e.g. +// {"http.method":"GET","http.status_code":200}. That is the shape the attr() +// macro and json_extract_string(col, '$."key"') paths expect; marshaling the +// raw []*KeyValue, as the old toJSON path did, produced an array of +// {Key,Value} structs that no JSON-path query could read. Returns "" for an +// empty list so the column stays NULL. +// +// The fast path writes the object directly into a pooled buffer, skipping the // map[string]any + interface boxing + reflection that json.Marshal needs — that // path dominated ingest allocations under load (profiled: ~7GB / 14% of // alloc_space at 175k rows/s). Attributes whose values are nested (array/kvlist) diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go index 3bf09abf..e92f9c1a 100644 --- a/internal/intelligence/detector.go +++ b/internal/intelligence/detector.go @@ -148,6 +148,10 @@ func (d *Detector) detectAnomalies(ctx context.Context, start, end time.Time) [] return anomalies } +// minErrorRateStddev floors the error-rate z-score denominator at one +// percentage point. See errorRateAnomalySQL. +const minErrorRateStddev = 0.01 + // errorRateAnomalySQL compares the error rate in [startNano, endNano) against // the window of equal length immediately before it. // @@ -167,10 +171,6 @@ func (d *Detector) detectAnomalies(ctx context.Context, start, end time.Time) [] // errors to 42%). One percentage point is the least noise worth assuming: it // keeps a single stray error in a few thousand spans below the threshold while // letting a real break through. -// minErrorRateStddev floors the error-rate z-score denominator at one -// percentage point. See errorRateAnomalySQL. -const minErrorRateStddev = 0.01 - func errorRateAnomalySQL(startNano, endNano int64, scope string) string { return fmt.Sprintf(` WITH current_period AS ( diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 5b1ba284..a01e6d97 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -23,7 +23,7 @@ const serverInstructions = "Start with observability_overview for system health, type Observability interface { Overview(context.Context, observability.Scope, int) (observability.Result[observability.Overview], error) Topology(context.Context, observability.Scope, int) (observability.Result[observability.Topology], error) - Performance(context.Context, observability.Scope, string, int) (observability.Result[observability.Performance], error) + Performance(context.Context, observability.Scope, observability.PerformanceOptions) (observability.Result[observability.Performance], error) Trace(context.Context, observability.Scope, string, string, int) (observability.Result[observability.TraceDetail], error) Logs(context.Context, observability.Scope, string, string, string, int) (observability.Result[observability.Logs], error) } @@ -287,7 +287,7 @@ func (s *Server) performance(ctx context.Context, _ *mcp.CallToolRequest, input if err != nil { return nil, observability.Result[observability.Performance]{}, err } - output, err := s.queries.Performance(ctx, scope, input.Service, input.Limit) + output, err := s.queries.Performance(ctx, scope, observability.PerformanceOptions{Service: input.Service, Limit: input.Limit}) if err != nil { return nil, output, err } diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go index ef33e379..67ab775f 100644 --- a/internal/mcp/server_test.go +++ b/internal/mcp/server_test.go @@ -127,7 +127,7 @@ func (f *fakeObservability) Topology(_ context.Context, scope observability.Scop }, nil } -func (f *fakeObservability) Performance(_ context.Context, scope observability.Scope, _ string, _ int) (observability.Result[observability.Performance], error) { +func (f *fakeObservability) Performance(_ context.Context, scope observability.Scope, _ observability.PerformanceOptions) (observability.Result[observability.Performance], error) { f.scope = scope return observability.Result[observability.Performance]{Schema: observability.PerformanceSchema, Summary: "performance"}, nil } diff --git a/internal/observability/performance.go b/internal/observability/performance.go index f0378304..cf2fb8fb 100644 --- a/internal/observability/performance.go +++ b/internal/observability/performance.go @@ -203,7 +203,21 @@ SELECT FROM service_rollup WHERE bucket >= ? AND bucket < ? AND (? = '' OR namespace = ?) AND (? = '' OR service = ?)` -func (s *Service) Performance(ctx context.Context, scope Scope, service string, limit int) (Result[Performance], error) { +// PerformanceOptions selects what a performance read costs. +// +// Heatmap is the cross-service comparison grid: 12 services by every bucket in +// the window. The browser's performance page draws it. An MCP caller asked +// about one service, so for that path it is a second DuckDB query and a payload +// -- about 85% of a 113 KB response on the live demo -- covering 12 services it +// did not ask about. +type PerformanceOptions struct { + Service string + Limit int + Heatmap bool +} + +func (s *Service) Performance(ctx context.Context, scope Scope, opts PerformanceOptions) (Result[Performance], error) { + service, limit := opts.Service, opts.Limit scope, err := s.normalizeScope(scope) if err != nil { return Result[Performance]{}, err @@ -240,23 +254,25 @@ func (s *Service) Performance(ctx context.Context, scope Scope, service string, } data.Endpoints = endpoints - rows, err = s.db.QueryContext(ctx, performanceHeatmapSQL(window), scope.Start, scope.End, scope.Namespace, scope.Namespace, scope.Start, scope.End, scope.Namespace, scope.Namespace) - if err != nil { - return Result[Performance]{}, fmt.Errorf("query latency heatmap: %w", err) - } - for rows.Next() { - var point HeatmapPoint - if err := rows.Scan(&point.Time, &point.Service, &point.P95MS); err != nil { + if opts.Heatmap { + rows, err = s.db.QueryContext(ctx, performanceHeatmapSQL(window), scope.Start, scope.End, scope.Namespace, scope.Namespace, scope.Start, scope.End, scope.Namespace, scope.Namespace) + if err != nil { + return Result[Performance]{}, fmt.Errorf("query latency heatmap: %w", err) + } + for rows.Next() { + var point HeatmapPoint + if err := rows.Scan(&point.Time, &point.Service, &point.P95MS); err != nil { + rows.Close() + return Result[Performance]{}, fmt.Errorf("scan latency heatmap: %w", err) + } + data.Heatmap = append(data.Heatmap, point) + } + if err := rows.Err(); err != nil { rows.Close() - return Result[Performance]{}, fmt.Errorf("scan latency heatmap: %w", err) + return Result[Performance]{}, fmt.Errorf("iterate latency heatmap: %w", err) } - data.Heatmap = append(data.Heatmap, point) - } - if err := rows.Err(); err != nil { rows.Close() - return Result[Performance]{}, fmt.Errorf("iterate latency heatmap: %w", err) } - rows.Close() midpoint := scope.Start.Add(scope.End.Sub(scope.Start) / 2) before, err := s.performanceAggregate(ctx, Scope{Namespace: scope.Namespace, Start: scope.Start, End: midpoint}, service) diff --git a/internal/observability/performance_heatmap_test.go b/internal/observability/performance_heatmap_test.go new file mode 100644 index 00000000..c3406bb4 --- /dev/null +++ b/internal/observability/performance_heatmap_test.go @@ -0,0 +1,84 @@ +package observability + +import ( + "context" + "database/sql" + "testing" + "time" + + _ "github.com/duckdb/duckdb-go/v2" +) + +// The heatmap is a cross-service comparison grid: the twelve busiest services +// by every bucket in the window. The browser's performance page draws it. +// +// An MCP caller asks about one service and got that grid too -- on the live +// demo about 85% of a 113 KB response, describing twelve services it did not +// ask about, plus the second DuckDB query that built it. So it is opt-in, and +// the query has to be skipped rather than the field blanked afterwards. +func TestPerformanceOmitsTheHeatmapUnlessAsked(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + if _, err := db.Exec(` +CREATE TABLE service_rollup ( + bucket TIMESTAMP, namespace VARCHAR, service VARCHAR, + spans BIGINT, served_spans BIGINT, error_rate DOUBLE, + p50_ms DOUBLE, p95_ms DOUBLE, log_count BIGINT, metric_count BIGINT +)`); err != nil { + t.Fatal(err) + } + // queryEndpoints reads both the rollup and raw spans at the window edges. + if _, err := db.Exec(` +CREATE TABLE endpoint_rollup ( + bucket TIMESTAMP, namespace VARCHAR, service VARCHAR, method VARCHAR, path VARCHAR, + calls BIGINT, error_rate DOUBLE, p50_ms DOUBLE, p95_ms DOUBLE, p99_ms DOUBLE +); +CREATE TABLE spans ( + namespace VARCHAR, service VARCHAR, operation VARCHAR, kind VARCHAR, status VARCHAR, + http_method VARCHAR, http_route VARCHAR, + duration_ms DOUBLE, start_time TIMESTAMP, attributes_json VARCHAR +)`); err != nil { + t.Fatal(err) + } + + start := time.Date(2026, 9, 21, 20, 0, 0, 0, time.UTC) + for bucket := range 6 { + at := start.Add(time.Duration(bucket) * time.Minute) + for _, service := range []string{"checkout", "cart", "frontend", "payment"} { + if _, err := db.Exec(`INSERT INTO service_rollup VALUES (?, 'prod', ?, 10, 10, 0, 5, 20, 1, 1)`, + at, service); err != nil { + t.Fatal(err) + } + } + } + + svc := New(SQLDB(db), newTestRepository(t).Parquet, 30) + scope := Scope{Namespace: "prod", Start: start, End: start.Add(10 * time.Minute)} + ctx := context.Background() + + withGrid, err := svc.Performance(ctx, scope, PerformanceOptions{Service: "checkout", Limit: 50, Heatmap: true}) + if err != nil { + t.Fatalf("Performance with heatmap: %v", err) + } + if len(withGrid.Data.Heatmap) == 0 { + t.Fatal("the fixture produced no heatmap, so this test compares nothing") + } + + withoutGrid, err := svc.Performance(ctx, scope, PerformanceOptions{Service: "checkout", Limit: 50}) + if err != nil { + t.Fatalf("Performance without heatmap: %v", err) + } + if got := len(withoutGrid.Data.Heatmap); got != 0 { + t.Errorf("heatmap carries %d points when it was not requested", got) + } + // Everything the caller did ask for is untouched. + if len(withoutGrid.Data.Points) != len(withGrid.Data.Points) { + t.Errorf("points = %d without the grid, %d with it", len(withoutGrid.Data.Points), len(withGrid.Data.Points)) + } + if withoutGrid.Data.Totals != withGrid.Data.Totals { + t.Errorf("totals differ without the grid:\n %+v\n %+v", withoutGrid.Data.Totals, withGrid.Data.Totals) + } +} diff --git a/internal/observability/service_test.go b/internal/observability/service_test.go index caeaaffa..e4d36bb9 100644 --- a/internal/observability/service_test.go +++ b/internal/observability/service_test.go @@ -241,7 +241,7 @@ func TestPerformanceReturnsAllVisualizationDatasets(t *testing.T) { WithArgs(midpoint, end, "prod", "prod", "checkout", "checkout"). WillReturnRows(sqlmock.NewRows([]string{"spans", "served_spans", "error_rate", "p50_ms", "p95_ms"}).AddRow(70.0, 70.0, 0.06, 70.0, 180.0)) - result, err := svc.Performance(context.Background(), Scope{Namespace: "prod", Start: start, End: end}, "checkout", 25) + result, err := svc.Performance(context.Background(), Scope{Namespace: "prod", Start: start, End: end}, PerformanceOptions{Service: "checkout", Limit: 25, Heatmap: true}) if err != nil { t.Fatalf("Performance: %v", err) } diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go index 03970629..050c5d25 100644 --- a/internal/telemetry/parquet.go +++ b/internal/telemetry/parquet.go @@ -436,9 +436,6 @@ func (p *ParquetStore) RestoreRetiredInputs(inputs []string, replacementID strin }) } -// Trace reads only ranges selected by the persistent hash index. Scope filters -// and the limit are applied while decoding so a pathological trace cannot grow -// request memory without bound. // TraceTotals describes the whole trace, not the page a limit admitted. // // These come free: readIndexedTrace already decodes every row of the trace and @@ -454,6 +451,10 @@ type TraceTotals struct { MaxEndNano int64 } +// Trace reads only ranges selected by the persistent hash index. Scope filters +// and the limit are applied while decoding so a pathological trace cannot grow +// request memory without bound. The totals it returns describe the whole trace, +// not the page the limit admitted. func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSpan, TraceTotals, error) { var totals TraceTotals services := make(map[string]struct{}, 8)