diff --git a/apps/server/src/observability/HeapAttribution.test.ts b/apps/server/src/observability/HeapAttribution.test.ts new file mode 100644 index 0000000000000..b355ba5c35e1f --- /dev/null +++ b/apps/server/src/observability/HeapAttribution.test.ts @@ -0,0 +1,69 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Context from "effect/Context"; +import * as TestClock from "effect/testing/TestClock"; +import * as Tracer from "effect/Tracer"; + +import * as HeapAttribution from "./HeapAttribution.ts"; + +const MB = 1024 * 1024; + +it.effect("names the span that grew the heap in a spike, not an older open or ended one", () => + Effect.gen(function* () { + const ended: Array = []; + const inner = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + ended.push(span); + }; + return span; + }, + }); + let heap = 300 * MB; + const attribution = yield* HeapAttribution.make(inner, () => heap); + yield* attribution.run; + // The sampler's baseline: heap before the spans allocate. + yield* TestClock.adjust("1 second"); + const open = (name: string) => + attribution.tracer.span({ + name, + parent: Option.none(), + annotations: Context.empty(), + links: [], + startTime: 0n, + kind: "internal", + root: false, + sampled: true, + }); + + const subscription = open("ws.rpc.orchestration.subscribeShell"); + const old = open("ProjectionStore.getSnapshot"); + // Slow drift never spikes, but leaves the long-lived span far above its start. + for (let step = 0; step < 5; step++) { + heap += 80 * MB; + yield* TestClock.adjust("11 seconds"); + if (step === 1) old.end(1n, undefined as never); + } + const small = open("ProjectionStore.getThreadShell"); + const big = open("ProjectionStore.getThreadProjection"); + heap += 2 * MB; + small.end(1n, undefined as never); + heap += 198 * MB; + yield* TestClock.adjust("2 seconds"); + + const spike = ended.find((span) => span.name === "server.heap.spike"); + assert.isDefined(spike); + assert.deepEqual(spike!.attributes.get("suspects"), [ + "ProjectionStore.getThreadProjection +200 MB", + ]); + big.end(2n, undefined as never); + subscription.end(3n, undefined as never); + assert.equal(big.attributes.get("memory.heap_growth_bytes"), 200 * MB); + assert.isUndefined(small.attributes.get("memory.heap_growth_bytes")); + assert.equal(old.attributes.get("memory.heap_growth_bytes"), 160 * MB); + }), +); diff --git a/apps/server/src/observability/HeapAttribution.ts b/apps/server/src/observability/HeapAttribution.ts new file mode 100644 index 0000000000000..4b18efd0fb911 --- /dev/null +++ b/apps/server/src/observability/HeapAttribution.ts @@ -0,0 +1,108 @@ +// @effect-diagnostics nodeBuiltinImport:off - only node:v8 exposes a heap reading cheap enough to take per span. +import * as NodeV8 from "node:v8"; + +import * as Effect from "effect/Effect"; +import type * as Scope from "effect/Scope"; +import * as Tracer from "effect/Tracer"; + +const MB = 1024 * 1024; +// Smaller growth is noise from concurrent fibers; an attribute on every span would bloat exports. +const SPAN_GROWTH_ATTRIBUTE_BYTES = 8 * MB; +const SPIKE_GROWTH_BYTES = 128 * MB; +const SPIKE_WINDOW_SAMPLES = 10; +const SPIKE_COOLDOWN_SAMPLES = 60; +const RECENT_SPANS = 32; +const REPORTED_SPANS = 8; + +const readHeapUsed = () => NodeV8.getHeapStatistics().used_heap_size; + +interface Growth { + readonly name: string; + readonly growthBytes: number; +} + +/** + * Heap growth is approximate: it includes other fibers' allocations and GC, so it names + * suspects, not culprits. A spike lists what was open or just grew. + */ +export const make = (tracer: Tracer.Tracer, heapUsed: () => number = readHeapUsed) => + Effect.sync(() => { + const open = new Map< + Tracer.Span, + { readonly name: string; readonly startHeap: number; readonly startTick: number } + >(); + const recent: Array = []; + const samples: Array = []; + let cooldown = 0; + let tick = 0; + + const wrapped = Tracer.make({ + ...(tracer.context === undefined ? {} : { context: tracer.context }), + span(options) { + const span = tracer.span(options); + const startHeap = heapUsed(); + open.set(span, { name: options.name, startHeap, startTick: tick }); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + open.delete(span); + const growthBytes = heapUsed() - startHeap; + if (growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES) { + span.attribute("memory.heap_growth_bytes", growthBytes); + recent.push({ name: options.name, growthBytes, endTick: tick }); + if (recent.length > RECENT_SPANS) recent.shift(); + } + end(endTime, exit); + }; + return span; + }, + }); + + // Untraced: it runs every second and only emits a span when the heap spikes. + const sample = Effect.fnUntraced(function* () { + const heap = heapUsed(); + tick++; + samples.push(heap); + if (samples.length > SPIKE_WINDOW_SAMPLES) samples.shift(); + if (cooldown > 0) { + cooldown--; + return; + } + const growthBytes = heap - Math.min(...samples); + if (growthBytes < SPIKE_GROWTH_BYTES) return; + cooldown = SPIKE_COOLDOWN_SAMPLES; + const suspects = [ + // Spans open since before the window, such as subscriptions, would claim all of its growth. + ...[...open.values()] + .filter(({ startTick }) => tick - startTick <= samples.length) + .map(({ name, startHeap }) => ({ name, growthBytes: heap - startHeap })), + ...recent.filter(({ endTick }) => tick - endTick <= samples.length), + ] + .filter((suspect) => suspect.growthBytes >= SPAN_GROWTH_ATTRIBUTE_BYTES) + .toSorted((left, right) => right.growthBytes - left.growthBytes) + .slice(0, REPORTED_SPANS) + .map(({ name, growthBytes }) => `${name} +${Math.round(growthBytes / MB)} MB`); + recent.length = 0; + // A span, not a log: this fiber starts before the server logger is provided. + yield* Effect.void.pipe( + Effect.withSpan("server.heap.spike", { + root: true, + level: "Warn", + attributes: { + heapUsedMb: Math.round(heap / MB), + growthMb: Math.round(growthBytes / MB), + suspects, + }, + }), + ); + }); + + const run: Effect.Effect = sample().pipe( + Effect.andThen(Effect.sleep("1 second")), + Effect.forever, + Effect.withTracer(wrapped), + Effect.forkScoped, + Effect.asVoid, + ); + + return { tracer: wrapped, run }; + }); diff --git a/apps/server/src/observability/Observability.ts b/apps/server/src/observability/Observability.ts index 343851e3d1d11..166d8ac194a8c 100644 --- a/apps/server/src/observability/Observability.ts +++ b/apps/server/src/observability/Observability.ts @@ -16,6 +16,7 @@ import * as ServerConfig from "../config.ts"; import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts"; import * as ServerLogger from "../serverLogger.ts"; import * as BrowserTraceCollector from "./BrowserTraceCollector.ts"; +import * as HeapAttribution from "./HeapAttribution.ts"; export const layer = Layer.unwrap( Effect.gen(function* () { @@ -71,8 +72,11 @@ export const layer = Layer.unwrap( ...(delegate ? { delegate } : {}), }); + const heapAttribution = yield* HeapAttribution.make(tracer); + yield* heapAttribution.run; + return Layer.mergeAll( - Layer.succeed(Tracer.Tracer, tracer), + Layer.succeed(Tracer.Tracer, heapAttribution.tracer), BrowserTraceCollector.layer(sink), ); }),