From a2a95fed5798bad34f518397a7fdf4abd698c655 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Mon, 21 Sep 2026 12:24:04 -0700 Subject: [PATCH] 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)