From f3b8b5cfa946b5d7d11558528e0df11274633d73 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Sun, 20 Sep 2026 10:53:49 -0700 Subject: [PATCH] fix(observability): report the trace, not the page, in trace_detail trace_detail computed services, duration_ms and has_error from the span slice that `limit` admitted, then stated them as facts about the trace. A call with limit=1 against a five-span trace returned "contains 1 spans across 1 services" and named only the service of the one span it happened to return, while the actual error sat three levels down in a service the page omitted. The failure is silent and the output reads as authoritative, so a caller has no reason to doubt it. Those fields now come from a single-row aggregate over the whole trace. spans and services still describe the page; span_count and service_count describe the trace; truncated says whether the two differ, and the summary gains "; showing N" when they do. The aggregate counts in DuckDB and returns one row, so it does not materialise the trace in this process. The aggregate reads the BIGINT start_unix_nano/end_unix_nano columns rather than start_time/end_time: those are TIMESTAMP, and subtracting them yields an INTERVAL that DuckDB will not divide. A DuckDB-backed test covers the query directly, because the sqlmock tests never execute the SQL and cannot catch that class of error. --- internal/observability/contracts.go | 5 + internal/observability/service_test.go | 24 ++++ internal/observability/trace.go | 61 +++++++-- internal/observability/trace_test.go | 181 +++++++++++++++++++++++++ ui/contracts.ts | 5 + 5 files changed, 262 insertions(+), 14 deletions(-) create mode 100644 internal/observability/trace_test.go diff --git a/internal/observability/contracts.go b/internal/observability/contracts.go index 19112a43..6c83d425 100644 --- a/internal/observability/contracts.go +++ b/internal/observability/contracts.go @@ -190,6 +190,11 @@ type TraceDetail struct { Services []string `json:"services"` Spans []TraceSpan `json:"spans"` Logs []LogEntry `json:"logs"` + // SpanCount and ServiceCount describe the trace. Spans and Services + // describe the page the limit admitted, which may be narrower. + SpanCount int `json:"span_count"` + ServiceCount int `json:"service_count"` + Truncated bool `json:"truncated"` } type LogBucket struct { diff --git a/internal/observability/service_test.go b/internal/observability/service_test.go index caeaaffa..8fe59615 100644 --- a/internal/observability/service_test.go +++ b/internal/observability/service_test.go @@ -330,6 +330,10 @@ func TestTraceSelectsRecentErrorAndCorrelatesLogs(t *testing.T) { }}); err != nil { t.Fatalf("commit trace fixture: %v", err) } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("trace-1", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(2), int64(2), 200.0, true)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("trace-1", start, end, "prod", "prod", 20). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}). @@ -510,6 +514,10 @@ func TestTraceLogsUseFullScopeEventTimeAcrossBatches(t *testing.T) { t.Fatal(err) } } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(1), int64(1), 1.0, false)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod", 10). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}). @@ -536,6 +544,10 @@ func TestTraceUsesIndexedParquet(t *testing.T) { }}}); err != nil { t.Fatal(err) } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("parquet-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(1), int64(1), 25.0, true)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("parquet-trace", start, end, "prod", "prod", 10). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}). @@ -571,6 +583,10 @@ func TestTraceCombinesIndexedSpansAcrossBatches(t *testing.T) { }}}); err != nil { t.Fatal(err) } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("split-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(2), int64(2), 2400025.0, true)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("split-trace", start, end, "prod", "prod", 10). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"})) @@ -597,6 +613,10 @@ func TestTraceReadsRecentRootFromParquetIndex(t *testing.T) { }}}); err != nil { t.Fatal(err) } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("new-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(1), int64(1), 10.0, false)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("new-trace", start, end, "prod", "prod", 10). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"})) @@ -623,6 +643,10 @@ func TestTraceFiltersIndexedSpansByNamespace(t *testing.T) { }}); err != nil { t.Fatal(err) } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("shared-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(1), int64(1), 0.0, false)) mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). WithArgs("shared-trace", start, end, "prod", "prod", 10). WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"})) diff --git a/internal/observability/trace.go b/internal/observability/trace.go index 01e9f16b..aff01314 100644 --- a/internal/observability/trace.go +++ b/internal/observability/trace.go @@ -19,6 +19,15 @@ ORDER BY MAX(CASE WHEN upper(status) IN ('ERROR', 'STATUS_CODE_ERROR') THEN 1 EL MAX(end_time) - MIN(start_time) DESC LIMIT 1` +const traceSummaryQuery = ` +SELECT + CAST(count(*) AS BIGINT), + CAST(count(DISTINCT service) AS BIGINT), + COALESCE((max(end_unix_nano) - min(start_unix_nano)) / 1000000.0, 0), + COALESCE(bool_or(upper(status) IN ('ERROR', 'STATUS_CODE_ERROR')), false) +FROM spans +WHERE trace_id = ? AND start_time >= ? AND start_time < ? AND (? = '' OR namespace = ?)` + const traceLogsQuery = ` SELECT time, severity, coalesce(service, ''), body, coalesce(trace_id, ''), coalesce(span_id, '') FROM logs @@ -68,29 +77,28 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin data.Spans = append(data.Spans, TraceSpan{SpanID: row.SpanID, ParentSpanID: row.ParentSpanID, Service: row.ServiceName, Operation: row.Name, Kind: row.Kind, Start: time.Unix(0, row.StartUnixNanos).UTC(), DurationMS: row.DurationMS, Status: row.StatusCode, StatusMessage: row.StatusMsg}) } + // Only the page's own service list is derived here. Duration and + // has_error describe the whole trace and come from the aggregate below. serviceSet := make(map[string]struct{}) - var first, last time.Time for _, span := range data.Spans { - if first.IsZero() || span.Start.Before(first) { - first = span.Start - } - if end := span.Start.Add(time.Duration(span.DurationMS * float64(time.Millisecond))); end.After(last) { - last = end - } - if strings.Contains(strings.ToUpper(span.Status), "ERROR") { - data.HasError = true - } if span.Service != "" { serviceSet[span.Service] = struct{}{} } } - if !first.IsZero() { - data.DurationMS = last.Sub(first).Seconds() * 1000 - } for name := range serviceSet { data.Services = append(data.Services, name) } sort.Strings(data.Services) + // The page above is what limit admitted. Everything the caller reads as + // a fact about the trace — how many spans it has, how many services it + // crosses, how long it took, whether it failed — comes from an + // aggregate over the whole trace instead, so a narrow page cannot + // silently redefine the trace. The aggregate reads no rows into this + // process: it is counted in DuckDB and returns one row. + if err := s.traceTotals(ctx, scope, traceID, &data); err != nil { + return Result[TraceDetail]{}, err + } + data.Truncated = data.SpanCount > len(data.Spans) data.Logs, err = s.traceLogsFromParquet(ctx, scope, traceID, limit) if err != nil { return Result[TraceDetail]{}, err @@ -99,11 +107,36 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin summary := "No traces found in this telemetry window" if traceID != "" { - summary = fmt.Sprintf("Trace %s contains %d spans across %d services", traceID, len(data.Spans), len(data.Services)) + summary = fmt.Sprintf("Trace %s contains %d spans across %d services", traceID, data.SpanCount, data.ServiceCount) + if data.Truncated { + summary += fmt.Sprintf("; showing %d", len(data.Spans)) + } } return Result[TraceDetail]{Schema: TraceSchema, Summary: summary, Data: data, Provenance: s.provenanceFor(scope, dataSource)}, nil } +// traceTotals fills the fields that describe the trace rather than the page. +// A trace that has aged out of the window reports zeros, which leaves the +// summary saying the trace holds no spans — true for the window asked about. +func (s *Service) traceTotals(ctx context.Context, scope Scope, traceID string, data *TraceDetail) error { + rows, err := s.db.QueryContext(ctx, traceSummaryQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace) + if err != nil { + return fmt.Errorf("query trace totals: %w", err) + } + defer rows.Close() + if rows.Next() { + var spanCount, serviceCount int64 + if err := rows.Scan(&spanCount, &serviceCount, &data.DurationMS, &data.HasError); err != nil { + return fmt.Errorf("scan trace totals: %w", err) + } + data.SpanCount, data.ServiceCount = int(spanCount), int(serviceCount) + } + if err := rows.Err(); err != nil { + return fmt.Errorf("iterate trace totals: %w", err) + } + return nil +} + func (s *Service) traceLogsFromParquet(ctx context.Context, scope Scope, traceID string, limit int) ([]LogEntry, error) { rows, err := s.db.QueryContext(ctx, traceLogsQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace, limit) if err != nil { diff --git a/internal/observability/trace_test.go b/internal/observability/trace_test.go new file mode 100644 index 00000000..6821b745 --- /dev/null +++ b/internal/observability/trace_test.go @@ -0,0 +1,181 @@ +package observability + +import ( + "context" + "database/sql" + "regexp" + "strings" + "testing" + "time" + + "github.com/DATA-DOG/go-sqlmock" + _ "github.com/duckdb/duckdb-go/v2" + "github.com/labstack/fanout/internal/telemetry" + telemetrystore "github.com/labstack/fanout/internal/telemetry/store" +) + +// A paged trace must still describe the trace. Limit bounds the spans that are +// returned; it must not silently redefine what the trace contains. Reporting a +// page's service list and span count as the trace's own is the failure this +// test exists to prevent: an agent reading "1 spans across 1 services" for a +// trace whose fault lies in a service the page omitted concludes the wrong +// service is broken, and the output gives it no reason to doubt that. +func TestTracePagedSpansStillDescribeTheWholeTrace(t *testing.T) { + svc, mock, repository := newMockService(t) + start := time.Date(2026, 7, 20, 11, 0, 0, 0, time.UTC) + end := start.Add(time.Hour) + if err := repository.Commit(context.Background(), telemetrystore.Batch{ID: "paged-trace", Spans: []telemetry.Span{ + { + Namespace: "prod", TraceID: "wide-trace", SpanID: "root", ServiceName: "cart", + Name: "ResolveBoolean", Kind: "CLIENT", + StartUnixNanos: start.UnixNano(), DurationMS: 2, StatusCode: "OK", + }, + { + Namespace: "prod", TraceID: "wide-trace", SpanID: "child", ParentSpanID: "root", ServiceName: "cart", + Name: "POST", Kind: "CLIENT", + StartUnixNanos: start.Add(time.Millisecond).UnixNano(), DurationMS: 1, StatusCode: "OK", + }, + { + Namespace: "prod", TraceID: "wide-trace", SpanID: "leaf", ParentSpanID: "child", ServiceName: "flagd", + Name: "resolveBoolean", Kind: "SERVER", + StartUnixNanos: start.Add(2 * time.Millisecond).UnixNano(), DurationMS: 1, StatusCode: "ERROR", + StatusMsg: "error evaluating flag with key failedReadinessProbe", + }, + }}); err != nil { + t.Fatal(err) + } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("wide-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(3), int64(2), 3.0, true)) + mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). + WithArgs("wide-trace", start, end, "prod", "prod", 1). + WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"})) + + result, err := svc.Trace(context.Background(), Scope{Namespace: "prod", Start: start, End: end}, "wide-trace", "", 1) + if err != nil { + t.Fatal(err) + } + + if len(result.Data.Spans) != 1 { + t.Fatalf("limit must still bound the returned spans: got %d", len(result.Data.Spans)) + } + if result.Data.SpanCount != 3 { + t.Errorf("span_count = %d, want 3: the count describes the trace, not the page", result.Data.SpanCount) + } + if result.Data.ServiceCount != 2 { + t.Errorf("service_count = %d, want 2: the count describes the trace, not the page", result.Data.ServiceCount) + } + if !result.Data.Truncated { + t.Error("truncated = false, want true: 1 of 3 spans were returned") + } + if !result.Data.HasError { + t.Error("has_error = false, want true: the trace errors in a span the page omitted") + } + if strings.Contains(result.Summary, "1 spans") || strings.Contains(result.Summary, "1 services") { + t.Errorf("summary reports the page as the trace: %q", result.Summary) + } + if !strings.Contains(result.Summary, "3") { + t.Errorf("summary omits the trace's real span count: %q", result.Summary) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} + +// An untruncated trace must not be labelled truncated, and its summary must not +// carry a qualifier it has not earned. +func TestTraceWholeTraceIsNotReportedAsTruncated(t *testing.T) { + svc, mock, repository := newMockService(t) + start := time.Date(2026, 7, 20, 11, 0, 0, 0, time.UTC) + end := start.Add(time.Hour) + if err := repository.Commit(context.Background(), telemetrystore.Batch{ID: "whole-trace", Spans: []telemetry.Span{{ + Namespace: "prod", TraceID: "small-trace", SpanID: "root", ServiceName: "checkout", + Name: "pay", Kind: "SERVER", + StartUnixNanos: start.UnixNano(), DurationMS: 25, StatusCode: "OK", + }}}); err != nil { + t.Fatal(err) + } + mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)). + WithArgs("small-trace", start, end, "prod", "prod"). + WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}). + AddRow(int64(1), int64(1), 25.0, false)) + mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)). + WithArgs("small-trace", start, end, "prod", "prod", 10). + WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"})) + + result, err := svc.Trace(context.Background(), Scope{Namespace: "prod", Start: start, End: end}, "small-trace", "", 10) + if err != nil { + t.Fatal(err) + } + + if result.Data.Truncated { + t.Error("truncated = true for a trace returned in full") + } + if result.Data.SpanCount != 1 { + t.Errorf("span_count = %d, want 1", result.Data.SpanCount) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatal(err) + } +} + +// The sqlmock tests above prove the plumbing but never execute the SQL, so the +// aggregate's own arithmetic needs DuckDB to check it. start_time and end_time +// are TIMESTAMP (internal/query/views.go:19-20); subtracting them yields an +// INTERVAL, which is not a nanosecond count and cannot be scaled to +// milliseconds by division. Only the BIGINT *_unix_nano columns can. +func TestTraceSummaryQueryComputesDurationInMilliseconds(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatalf("open duckdb: %v", err) + } + defer db.Close() + if _, err := db.Exec(`CREATE TABLE spans ( + namespace VARCHAR, trace_id VARCHAR, service VARCHAR, status VARCHAR, + start_time TIMESTAMP, end_time TIMESTAMP, + start_unix_nano BIGINT, end_unix_nano BIGINT)`); err != nil { + t.Fatalf("create spans: %v", err) + } + base := time.Date(2026, 7, 20, 11, 0, 0, 0, time.UTC) + // The trace spans 3ms end-to-end: first span starts at base, last ends 3ms later. + rows := []struct { + service string + status string + startMS int64 + endMS int64 + }{ + {"cart", "OK", 0, 3}, + {"cart", "OK", 1, 2}, + {"flagd", "ERROR", 2, 3}, + } + for _, r := range rows { + start := base.Add(time.Duration(r.startMS) * time.Millisecond) + end := base.Add(time.Duration(r.endMS) * time.Millisecond) + if _, err := db.Exec(`INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + "prod", "t1", r.service, r.status, start, end, start.UnixNano(), end.UnixNano()); err != nil { + t.Fatalf("insert span: %v", err) + } + } + + var spanCount, serviceCount int64 + var durationMS float64 + var hasError bool + if err := db.QueryRow(traceSummaryQuery, "t1", base, base.Add(time.Hour), "prod", "prod"). + Scan(&spanCount, &serviceCount, &durationMS, &hasError); err != nil { + t.Fatalf("trace summary query: %v", err) + } + + if spanCount != 3 { + t.Errorf("span_count = %d, want 3", spanCount) + } + if serviceCount != 2 { + t.Errorf("service_count = %d, want 2", serviceCount) + } + if durationMS != 3 { + t.Errorf("duration_ms = %v, want 3", durationMS) + } + if !hasError { + t.Error("has_error = false, want true") + } +} diff --git a/ui/contracts.ts b/ui/contracts.ts index 65dfa796..2e348e9f 100644 --- a/ui/contracts.ts +++ b/ui/contracts.ts @@ -132,6 +132,11 @@ export interface TraceDetail { services: string[]; spans: TraceSpan[]; logs: LogEntry[]; + // span_count and service_count describe the trace; spans and services + // describe the page the limit admitted, which may be narrower. + span_count: number; + service_count: number; + truncated: boolean; } export interface LogBucket {