diff --git a/internal/ingest/attrs_test.go b/internal/ingest/attrs_test.go index 465f9022..261491c7 100644 --- a/internal/ingest/attrs_test.go +++ b/internal/ingest/attrs_test.go @@ -71,10 +71,10 @@ func TestAttrsJSON_MatchesReflect(t *testing.T) { fast := attrsJSON(attrs) slow := attrsJSONReflect(attrs) var mf, ms map[string]any - if err := json.Unmarshal(fast, &mf); err != nil { + if err := json.Unmarshal([]byte(fast), &mf); err != nil { t.Fatalf("case %d: fast output invalid JSON: %v (%s)", i, err, fast) } - if err := json.Unmarshal(slow, &ms); err != nil { + if err := json.Unmarshal([]byte(slow), &ms); err != nil { t.Fatalf("case %d: reflect output invalid JSON: %v", i, err) } if !reflect.DeepEqual(mf, ms) { @@ -84,14 +84,15 @@ func TestAttrsJSON_MatchesReflect(t *testing.T) { } // Non-finite floats route to the reflect path (json.Marshal errors on Inf/NaN), -// so attrsJSON must never emit invalid JSON like {"k":Inf} — it returns nil for -// the whole object (the reflect encoder's documented behavior). +// so attrsJSON must never emit invalid JSON like {"k":Inf} — it returns an +// empty string for the whole object (the reflect encoder's documented +// behavior). func TestAttrsJSON_NonFiniteFloatIsSafe(t *testing.T) { for _, v := range []float64{math.Inf(1), math.Inf(-1), math.NaN()} { got := attrsJSON([]*common.KeyValue{kvDouble("k", v)}) - if got != nil { + if got != "" { var m map[string]any - if err := json.Unmarshal(got, &m); err != nil { + if err := json.Unmarshal([]byte(got), &m); err != nil { t.Errorf("attrsJSON(%v) produced invalid JSON: %s", v, got) } } @@ -110,7 +111,7 @@ func TestAttrsJSON_ConcurrentSafe(t *testing.T) { go func() { defer wg.Done() for j := 0; j < 1000; j++ { - if got := attrsJSON(attrs); !bytes.Equal(got, want) { + if got := attrsJSON(attrs); got != want { t.Errorf("concurrent attrsJSON = %s, want %s", got, want) return } @@ -179,7 +180,7 @@ func TestAttrsJSON_FlatObject(t *testing.T) { got := attrsJSON(attrs) var m map[string]any - if err := json.Unmarshal(got, &m); err != nil { + if err := json.Unmarshal([]byte(got), &m); err != nil { t.Fatalf("output is not a JSON object: %v (%s)", err, got) } if m["http.method"] != "GET" { @@ -194,10 +195,10 @@ func TestAttrsJSON_FlatObject(t *testing.T) { } func TestAttrsJSON_EmptyIsNil(t *testing.T) { - if got := attrsJSON(nil); got != nil { + if got := attrsJSON(nil); got != "" { t.Errorf("attrsJSON(nil) = %s, want nil", got) } - if got := attrsJSON([]*common.KeyValue{}); got != nil { + if got := attrsJSON([]*common.KeyValue{}); got != "" { t.Errorf("attrsJSON([]) = %s, want nil", got) } } @@ -209,7 +210,7 @@ func TestAttrsJSON_NestedKvlist(t *testing.T) { }}}, } var m map[string]any - if err := json.Unmarshal(attrsJSON(attrs), &m); err != nil { + if err := json.Unmarshal([]byte(attrsJSON(attrs)), &m); err != nil { t.Fatalf("unmarshal: %v", err) } outer, ok := m["outer"].(map[string]any) diff --git a/internal/ingest/server.go b/internal/ingest/server.go index ce2a5819..cc34276b 100644 --- a/internal/ingest/server.go +++ b/internal/ingest/server.go @@ -347,16 +347,16 @@ func spanDurationMS(startNano, endNano uint64) float64 { return float64(endNano-startNano) / 1e6 } -func toJSON(v interface{}) []byte { +func toJSON(v interface{}) string { if v == nil { - return nil + return "" } b, err := json.Marshal(v) if err != nil { slog.Error("json marshal failed", "err", err) - return []byte("null") + return "null" } - return b + return string(b) } // attrsJSON flattens an OTLP attribute list into a flat JSON object keyed by the @@ -378,9 +378,9 @@ var attrBufPool = sync.Pool{New: func() any { return new(bytes.Buffer) }} // whereas the reflect fallback (a map) sorts keys and keeps last-wins. This is // immaterial for fanout: OTLP attribute keys are unique by spec, and queries // read attributes_json by key via attr()/json_extract (order-independent). -func attrsJSON(attrs []*common.KeyValue) []byte { +func attrsJSON(attrs []*common.KeyValue) string { if len(attrs) == 0 { - return nil + return "" } if attrsNeedReflect(attrs) { return attrsJSONReflect(attrs) @@ -402,10 +402,10 @@ func attrsJSON(attrs []*common.KeyValue) []byte { appendScalarJSON(buf, kv.Value) n++ } - var out []byte + var out string if n > 0 { buf.WriteByte('}') - out = append([]byte(nil), buf.Bytes()...) // copy out before returning buf to the pool + out = buf.String() // copies out of the pooled buffer, as []byte did } attrBufPool.Put(buf) return out @@ -536,7 +536,7 @@ func appendJSONString(buf *bytes.Buffer, s string) { } // attrsJSONReflect is the reflection-based encoder, retained for nested values. -func attrsJSONReflect(attrs []*common.KeyValue) []byte { +func attrsJSONReflect(attrs []*common.KeyValue) string { m := make(map[string]any, len(attrs)) for _, kv := range attrs { if kv == nil || kv.Key == "" { @@ -545,21 +545,21 @@ func attrsJSONReflect(attrs []*common.KeyValue) []byte { m[kv.Key] = attrValue(kv.Value) } if len(m) == 0 { - return nil + return "" } b, err := json.Marshal(m) if err != nil { slog.Error("attrs json marshal failed", "err", err) - return nil + return "" } - return b + return string(b) } // resourceAttrsJSON flattens a resource's attributes into the same flat object // shape as attrsJSON, so attr(resource_json, 'key') resolves. -func resourceAttrsJSON(r *resourcepb.Resource) []byte { +func resourceAttrsJSON(r *resourcepb.Resource) string { if r == nil { - return nil + return "" } return attrsJSON(r.Attributes) } @@ -811,9 +811,9 @@ func hexOrEmpty(b []byte) string { return fmt.Sprintf("%x", b) } -func eventsToJSON(events []*tracepb.Span_Event) []byte { +func eventsToJSON(events []*tracepb.Span_Event) string { if len(events) == 0 { - return nil + return "" } type evt struct { Time int64 `json:"time_unix_nano"` @@ -837,14 +837,14 @@ func eventsToJSON(events []*tracepb.Span_Event) []byte { b, err := json.Marshal(out) if err != nil { slog.Error("json marshal events failed", "err", err) - return nil + return "" } - return b + return string(b) } -func linksToJSON(links []*tracepb.Span_Link) []byte { +func linksToJSON(links []*tracepb.Span_Link) string { if len(links) == 0 { - return nil + return "" } type link struct { TraceID string `json:"trace_id"` @@ -870,9 +870,9 @@ func linksToJSON(links []*tracepb.Span_Link) []byte { b, err := json.Marshal(out) if err != nil { slog.Error("json marshal links failed", "err", err) - return nil + return "" } - return b + return string(b) } func scopeInfo(scope *common.InstrumentationScope) (name, version string) { @@ -882,9 +882,9 @@ func scopeInfo(scope *common.InstrumentationScope) (name, version string) { return scope.Name, scope.Version } -func exemplarsToJSON(exemplars []*metricspb.Exemplar) []byte { +func exemplarsToJSON(exemplars []*metricspb.Exemplar) string { if len(exemplars) == 0 { - return nil + return "" } type ex struct { Time int64 `json:"time_unix_nano"` @@ -919,9 +919,9 @@ func exemplarsToJSON(exemplars []*metricspb.Exemplar) []byte { b, err := json.Marshal(out) if err != nil { slog.Error("json marshal exemplars failed", "err", err) - return nil + return "" } - return b + return string(b) } func expHistBuckets(dp *metricspb.ExponentialHistogramDataPoint) []float64 { diff --git a/internal/ingest/server_test.go b/internal/ingest/server_test.go index 94c64727..2c9f512a 100644 --- a/internal/ingest/server_test.go +++ b/internal/ingest/server_test.go @@ -32,10 +32,10 @@ func TestToJSON(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { result := toJSON(tc.input) - if tc.nilOut && result != nil { + if tc.nilOut && result != "" { t.Errorf("toJSON(%v) = %v, want nil", tc.input, result) } - if !tc.nilOut && result == nil { + if !tc.nilOut && result == "" { t.Errorf("toJSON(%v) = nil, want non-nil", tc.input) } }) @@ -209,10 +209,10 @@ func TestEventsToJSON(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { result := eventsToJSON(tc.events) - if tc.nilOut && result != nil { + if tc.nilOut && result != "" { t.Errorf("eventsToJSON() = %v, want nil", result) } - if !tc.nilOut && result == nil { + if !tc.nilOut && result == "" { t.Error("eventsToJSON() = nil, want non-nil") } }) @@ -247,10 +247,10 @@ func TestLinksToJSON(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { result := linksToJSON(tc.links) - if tc.nilOut && result != nil { + if tc.nilOut && result != "" { t.Errorf("linksToJSON() = %v, want nil", result) } - if !tc.nilOut && result == nil { + if !tc.nilOut && result == "" { t.Error("linksToJSON() = nil, want non-nil") } }) @@ -319,10 +319,10 @@ func TestExemplarsToJSON(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { result := exemplarsToJSON(tc.exemplars) - if tc.nilOut && result != nil { + if tc.nilOut && result != "" { t.Errorf("exemplarsToJSON() = %v, want nil", result) } - if !tc.nilOut && result == nil { + if !tc.nilOut && result == "" { t.Error("exemplarsToJSON() = nil, want non-nil") } }) diff --git a/internal/telemetry/commit_memory_test.go b/internal/telemetry/commit_memory_test.go index 6200bc5d..28505120 100644 --- a/internal/telemetry/commit_memory_test.go +++ b/internal/telemetry/commit_memory_test.go @@ -62,12 +62,12 @@ func syntheticSpan(i int, base time.Time) Span { EndUnixNanos: start.Add(3 * time.Millisecond).UnixNano(), DurationMS: 3, StatusCode: "OK", - ResourceJSON: []byte(`{"service.name":"checkout","deployment.environment":"prod","host.name":"cube-10"}`), - AttributesJSON: []byte(fmt.Sprintf( - `{"http.method":"POST","http.route":"/api/orders","http.status_code":200,"order.id":"%d","user.id":"u-%d","session":"%032x"}`, i, i*7, i)), - EventsJSON: []byte(fmt.Sprintf( - `[{"name":"validated","time":%d},{"name":"charged","time":%d,"amount":%d}]`, start.UnixNano(), start.UnixNano()+1000, i%9999)), - LinksJSON: []byte(fmt.Sprintf(`[{"trace_id":"%032x","span_id":"%016x"}]`, i+1, i+1)), + ResourceJSON: `{"service.name":"checkout","deployment.environment":"prod","host.name":"cube-10"}`, + AttributesJSON: fmt.Sprintf( + `{"http.method":"POST","http.route":"/api/orders","http.status_code":200,"order.id":"%d","user.id":"u-%d","session":"%032x"}`, i, i*7, i), + EventsJSON: fmt.Sprintf( + `[{"name":"validated","time":%d},{"name":"charged","time":%d,"amount":%d}]`, start.UnixNano(), start.UnixNano()+1000, i%9999), + LinksJSON: fmt.Sprintf(`[{"trace_id":"%032x","span_id":"%016x"}]`, i+1, i+1), } } diff --git a/internal/telemetry/conversion_alloc_test.go b/internal/telemetry/conversion_alloc_test.go new file mode 100644 index 00000000..221ea7d0 --- /dev/null +++ b/internal/telemetry/conversion_alloc_test.go @@ -0,0 +1,60 @@ +//go:build !race + +package telemetry + +import ( + "runtime" + "testing" + "time" + "unsafe" +) + +// Converting a batch into Parquet rows must not duplicate the payload it was +// handed. The JSON columns are the bulk of a span, the originals stay live +// until the commit is durably acknowledged, and four commit workers run at +// once -- so a copy here is four copies of the batch resident at the moment +// the process is most likely to be killed. +// +// The end-to-end commit gate cannot see this on its own: parquet-go and zstd +// pool their buffers, so its figure swings between 3.2x and 5.3x payload run +// to run and a copy of this size hides inside that spread. Measuring the +// conversion alone is quiet enough to assert on. +func TestSpanConversionDoesNotCopyJSONPayloads(t *testing.T) { + base := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC) + spans := make([]Span, 20_000) + for i := range spans { + spans[i] = syntheticSpan(i, base) + } + var jsonBytes int64 + for i := range spans { + jsonBytes += int64(len(spans[i].ResourceJSON) + len(spans[i].AttributesJSON) + + len(spans[i].EventsJSON) + len(spans[i].LinksJSON)) + } + + var before, after runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&before) + + rows := make([]spanParquetRow, len(spans)) + for i := range spans { + rows[i] = makeSpanParquetRow(spans[i]) + } + + runtime.ReadMemStats(&after) + runtime.KeepAlive(rows) + runtime.KeepAlive(spans) + + allocated := int64(after.TotalAlloc - before.TotalAlloc) + headers := int64(len(spans)) * int64(spanParquetRowSize()) + t.Logf("json payload %.1f MiB, row headers %.1f MiB, allocated %.1f MiB", + float64(jsonBytes)/(1<<20), float64(headers)/(1<<20), float64(allocated)/(1<<20)) + + // Allowing the row slice plus a margin, and nothing like the payload again. + limit := headers + jsonBytes/4 + if allocated > limit { + t.Errorf("conversion allocated %.1f MiB against a %.1f MiB JSON payload; it is copying the payload rather than referencing it", + float64(allocated)/(1<<20), float64(jsonBytes)/(1<<20)) + } +} + +func spanParquetRowSize() uintptr { return unsafe.Sizeof(spanParquetRow{}) } diff --git a/internal/telemetry/parquet_rows.go b/internal/telemetry/parquet_rows.go index e338c229..b9e9b97f 100644 --- a/internal/telemetry/parquet_rows.go +++ b/internal/telemetry/parquet_rows.go @@ -56,7 +56,7 @@ func makeSpanParquetRow(r Span) spanParquetRow { Service: r.ServiceName, Operation: r.Name, Kind: r.Kind, StartTime: r.StartUnixNanos, EndTime: r.EndUnixNanos, StartUnixNano: r.StartUnixNanos, EndUnixNano: r.EndUnixNanos, DurationMS: r.DurationMS, Status: r.StatusCode, StatusMessage: r.StatusMsg, - ResourceJSON: string(r.ResourceJSON), AttributesJSON: string(r.AttributesJSON), EventsJSON: string(r.EventsJSON), LinksJSON: string(r.LinksJSON), + ResourceJSON: r.ResourceJSON, AttributesJSON: r.AttributesJSON, EventsJSON: r.EventsJSON, LinksJSON: r.LinksJSON, TraceState: r.TraceState, Flags: int64(r.Flags), ScopeName: r.ScopeName, ScopeVersion: r.ScopeVersion, IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt, HTTPMethod: r.HTTPMethod, HTTPStatusCode: r.HTTPStatusCode, HTTPRoute: r.HTTPRoute, DBSystem: r.DBSystem, RPCMethod: r.RPCMethod, @@ -115,7 +115,7 @@ func makeLogParquetRow(r Log) logParquetRow { ObservedTime: FirstPositiveNanos(r.ObservedTimeNanos, r.EventUnixNanos, r.TimeUnixNanos, r.IngestedAt), TimeUnixNano: r.TimeUnixNanos, ObservedTimeUnixNano: r.ObservedTimeNanos, Severity: r.Severity, SeverityNumber: int64(r.SeverityNumber), Body: r.Body, Service: r.ServiceName, TraceID: r.TraceID, SpanID: r.SpanID, Flags: int64(r.Flags), - ResourceJSON: string(r.ResourceJSON), AttributesJSON: string(r.AttributesJSON), ScopeName: r.ScopeName, + ResourceJSON: r.ResourceJSON, AttributesJSON: r.AttributesJSON, ScopeName: r.ScopeName, ScopeVersion: r.ScopeVersion, IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt, BodyTemplate: r.BodyTemplate, } } @@ -147,9 +147,9 @@ func makeMetricParquetRow(r Metric) metricParquetRow { return metricParquetRow{ Namespace: r.Namespace, MetricTime: FirstPositiveNanos(r.EventUnixNanos, r.TimeUnixNanos, r.IngestedAt), TimeUnixNano: r.TimeUnixNanos, Name: r.Name, Description: r.Description, Unit: r.Unit, MetricType: r.Type, Service: r.ServiceName, - Value: r.Value, HistBoundsJSON: string(r.HistBoundsJSON), HistCountsJSON: string(r.HistCountsJSON), - HistCount: r.HistCount, HistSum: r.HistSum, ExemplarsJSON: string(r.ExemplarsJSON), - AttributesJSON: string(r.AttributesJSON), ResourceJSON: string(r.ResourceJSON), ScopeName: r.ScopeName, + Value: r.Value, HistBoundsJSON: r.HistBoundsJSON, HistCountsJSON: r.HistCountsJSON, + HistCount: r.HistCount, HistSum: r.HistSum, ExemplarsJSON: r.ExemplarsJSON, + AttributesJSON: r.AttributesJSON, ResourceJSON: r.ResourceJSON, ScopeName: r.ScopeName, ScopeVersion: r.ScopeVersion, IngestedAt: r.IngestedAt, IngestedUnixNano: r.IngestedAt, } } diff --git a/internal/telemetry/parquet_test.go b/internal/telemetry/parquet_test.go index 17e60df1..17e4d80c 100644 --- a/internal/telemetry/parquet_test.go +++ b/internal/telemetry/parquet_test.go @@ -182,16 +182,16 @@ func TestParquetStorePreservesCompleteLogAndMetricRows(t *testing.T) { logRow := Log{ Namespace: "tenant", EventUnixNanos: 11, TimeUnixNanos: 12, ObservedTimeNanos: 13, Severity: "ERROR", SeverityNumber: 17, Body: "declined", ServiceName: "checkout", - TraceID: "trace", SpanID: "span", Flags: 1, ResourceJSON: []byte(`{"host":"one"}`), - AttributesJSON: []byte(`{"attempt":2}`), ScopeName: "scope", ScopeVersion: "1.2.3", + TraceID: "trace", SpanID: "span", Flags: 1, ResourceJSON: `{"host":"one"}`, + AttributesJSON: `{"attempt":2}`, ScopeName: "scope", ScopeVersion: "1.2.3", IngestedAt: 14, BodyTemplate: "declined: {reason}", } metricRow := Metric{ Namespace: "tenant", EventUnixNanos: 21, TimeUnixNanos: 22, Name: "request.duration", Description: "request latency", Unit: "ms", Type: "histogram", ServiceName: "checkout", Value: 23.5, - HistBoundsJSON: []byte(`[1,5,10]`), HistCountsJSON: []byte(`[2,3,4,5]`), HistCount: 14, HistSum: 47, - ExemplarsJSON: []byte(`[{"trace_id":"trace"}]`), AttributesJSON: []byte(`{"route":"/pay"}`), - ResourceJSON: []byte(`{"host":"one"}`), ScopeName: "scope", ScopeVersion: "1.2.3", IngestedAt: 24, + HistBoundsJSON: `[1,5,10]`, HistCountsJSON: `[2,3,4,5]`, HistCount: 14, HistSum: 47, + ExemplarsJSON: `[{"trace_id":"trace"}]`, AttributesJSON: `{"route":"/pay"}`, + ResourceJSON: `{"host":"one"}`, ScopeName: "scope", ScopeVersion: "1.2.3", IngestedAt: 24, } if err := store.CommitBatch(context.Background(), BatchMetadata{ID: "complete-signals"}, nil, []Log{logRow}, []Metric{metricRow}); err != nil { t.Fatal(err) @@ -492,8 +492,8 @@ func completeTestSpan() Span { Namespace: "tenant", TraceID: "0123456789abcdef0123456789abcdef", SpanID: "0123456789abcdef", ParentSpanID: "fedcba9876543210", ServiceName: "checkout", Name: "POST /orders", Kind: "SERVER", StartUnixNanos: 10, EndUnixNanos: 20, DurationMS: 0.00001, StatusCode: "ERROR", StatusMsg: "declined", - ResourceJSON: []byte(`{"host":"one"}`), AttributesJSON: []byte(`{"http.request.method":"POST"}`), - EventsJSON: []byte(`[{"name":"exception"}]`), LinksJSON: []byte(`[{"trace_id":"linked"}]`), + ResourceJSON: `{"host":"one"}`, AttributesJSON: `{"http.request.method":"POST"}`, + EventsJSON: `[{"name":"exception"}]`, LinksJSON: `[{"trace_id":"linked"}]`, TraceState: "vendor=value", Flags: 1, ScopeName: "scope", ScopeVersion: "1.2.3", IngestedAt: 30, HTTPMethod: "POST", HTTPStatusCode: "500", HTTPRoute: "/orders", DBSystem: "postgresql", RPCMethod: "Create", RPCService: "orders.v1.Orders", PeerService: "payments", ServiceVersion: "4.5.6", diff --git a/internal/telemetry/rows.go b/internal/telemetry/rows.go index 09b7e4a7..ff7ca301 100644 --- a/internal/telemetry/rows.go +++ b/internal/telemetry/rows.go @@ -15,10 +15,10 @@ type Span struct { DurationMS float64 StatusCode string StatusMsg string - ResourceJSON []byte - AttributesJSON []byte - EventsJSON []byte - LinksJSON []byte + ResourceJSON string + AttributesJSON string + EventsJSON string + LinksJSON string TraceState string Flags uint32 ScopeName string @@ -66,8 +66,8 @@ type Log struct { TraceID string SpanID string Flags uint32 - ResourceJSON []byte - AttributesJSON []byte + ResourceJSON string + AttributesJSON string ScopeName string ScopeVersion string IngestedAt int64 @@ -84,13 +84,13 @@ type Metric struct { Type string ServiceName string Value float64 - HistBoundsJSON []byte - HistCountsJSON []byte + HistBoundsJSON string + HistCountsJSON string HistCount int64 HistSum float64 - ExemplarsJSON []byte - AttributesJSON []byte - ResourceJSON []byte + ExemplarsJSON string + AttributesJSON string + ResourceJSON string ScopeName string ScopeVersion string IngestedAt int64