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)