Open History to see the last commit
1// @effect-diagnostics-next-line nodeBuiltinImport:off - flushSync must write before the process exits.
2import * as NodeFS from "node:fs";
3import * as Cause from "effect/Cause";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import type * as Exit from "effect/Exit";
7import * as ExitRuntime from "effect/Exit";
8import * as Option from "effect/Option";
9import * as Queue from "effect/Queue";
10import * as Result from "effect/Result";
11import * as Semaphore from "effect/Semaphore";
12import * as Schema from "effect/Schema";
13import * as SchemaIssue from "effect/SchemaIssue";
14import * as SchemaTransformation from "effect/SchemaTransformation";
15import * as Tracer from "effect/Tracer";
16import { OtlpResource, OtlpTracer, OtlpSerialization } from "effect/observability";
17
18import { makeAsyncRotatingFileSink } from "./logging.ts";
19
20export const OtlpProtocol = Schema.Literals(["http/json", "http/protobuf"]);
21export type OtlpProtocol = typeof OtlpProtocol.Type;
22export const layerOtlpSerialization = (protocol: OtlpProtocol) =>
23 protocol === "http/protobuf" ? OtlpSerialization.layerProtobuf : OtlpSerialization.layerJson;
24
25export interface SignalExport {
26 readonly protocol: OtlpProtocol;
27 readonly headers: Readonly<Record<string, string>> | undefined;
28 readonly exportIntervalMs: number;
29}
30
31/**
32 * `process.runtime.*` resource attributes for a Node process, matching the
33 * OpenTelemetry Node process detector. Electron embeds Node, so an Electron
34 * process reports Node and names Electron in the description.
35 */
36export const nodeProcessRuntimeAttributes = (): Record<string, string> => ({
37 "process.runtime.name": "nodejs",
38 "process.runtime.version": process.versions.node,
39 "process.runtime.description": process.versions.electron
40 ? `Electron ${process.versions.electron}`
41 : "Node.js",
42});
43let generatedServiceInstanceId: string | undefined;
44
45/**
46 * `service.instance.id` shared by every resource this process exports. Effect lets code
47 * attributes beat `OTEL_RESOURCE_ATTRIBUTES`, so callers pass the operator's in to keep it winning.
48 */
49export const processServiceInstanceId = (
50 operatorAttributes: Readonly<Record<string, string>> = {},
51): string => {
52 if (operatorAttributes["service.instance.id"]) return operatorAttributes["service.instance.id"];
53 // @effect-diagnostics-next-line cryptoRandomUUID:off -- Resources are built synchronously, outside any Effect.
54 generatedServiceInstanceId ??= crypto.randomUUID();
55 return generatedServiceInstanceId;
56};
57
58export const DEFAULT_SIGNAL_EXPORT: SignalExport = {
59 protocol: "http/json",
60 headers: undefined,
61 exportIntervalMs: 10_000,
62};
63
64const FLUSH_BUFFER_THRESHOLD = 256;
65
66export type TraceAttributes = Readonly<Record<string, unknown>>;
67
68export interface TraceRecordEvent {
69 readonly name: string;
70 readonly timeUnixNano: string;
71 readonly attributes: Readonly<Record<string, unknown>>;
72}
73
74export interface TraceRecordLink {
75 readonly traceId: string;
76 readonly spanId: string;
77 readonly attributes: Readonly<Record<string, unknown>>;
78}
79
80interface BaseTraceRecord {
81 readonly name: string;
82 readonly kind: string;
83 readonly traceId: string;
84 readonly spanId: string;
85 readonly parentSpanId?: string;
86 readonly sampled: boolean;
87 readonly startTimeUnixNano: string;
88 readonly endTimeUnixNano: string;
89 readonly durationMs: number;
90 readonly attributes: Readonly<Record<string, unknown>>;
91 readonly events: ReadonlyArray<TraceRecordEvent>;
92 readonly links: ReadonlyArray<TraceRecordLink>;
93}
94
95export interface EffectTraceRecord extends BaseTraceRecord {
96 readonly type: "effect-span";
97 readonly exit:
98 | {
99 readonly _tag: "Success";
100 }
101 | {
102 readonly _tag: "Interrupted";
103 readonly cause: string;
104 }
105 | {
106 readonly _tag: "Failure";
107 readonly cause: string;
108 };
109}
110
111export interface OtlpTraceRecord extends BaseTraceRecord {
112 readonly type: "otlp-span";
113 readonly resourceAttributes: Readonly<Record<string, unknown>>;
114 readonly scope: Readonly<{
115 readonly name?: string;
116 readonly version?: string;
117 readonly attributes: Readonly<Record<string, unknown>>;
118 }>;
119 readonly status?:
120 | {
121 readonly code?: string;
122 readonly message?: string;
123 }
124 | undefined;
125}
126
127export type TraceRecord = EffectTraceRecord | OtlpTraceRecord;
128
129function isStructuralTag(value: unknown): value is string {
130 return (
131 typeof value === "string" &&
132 value.length > 0 &&
133 value.length <= 128 &&
134 /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value)
135 );
136}
137
138export function errorTag(error: unknown): string {
139 try {
140 if (typeof error === "object" && error !== null && "_tag" in error) {
141 return isStructuralTag(error._tag) ? error._tag : "TaggedError";
142 }
143 if (error instanceof Error) {
144 return isStructuralTag(error.name) ? error.name : "Error";
145 }
146 } catch {
147 return "UnknownError";
148 }
149 return typeof error;
150}
151
152export function causeErrorTag(cause: Cause.Cause<unknown>): string {
153 const failure = Cause.findErrorOption(cause);
154 if (Option.isSome(failure)) {
155 return errorTag(failure.value);
156 }
157 return cause.reasons[0]?._tag ?? "Empty";
158}
159
160export interface TraceSinkOptions {
161 readonly filePath: string;
162 readonly maxBytes: number;
163 readonly maxFiles: number;
164 readonly batchWindowMs: number;
165 readonly onFlush?: (stats: TraceSinkFlushStats) => Effect.Effect<void>;
166}
167
168export interface TraceSinkFlushStats {
169 readonly logicalWriteBytes: number;
170 readonly count: number;
171 readonly durationMs: number;
172}
173
174export interface TraceSink {
175 readonly filePath: string;
176 push: (record: TraceRecord) => void;
177 flush: Effect.Effect<void>;
178 /** Appends buffered records synchronously, for process exit where async writes never land. */
179 flushSync: () => void;
180 close: () => Effect.Effect<void>;
181}
182
183export interface LocalFileTracerOptions extends TraceSinkOptions {
184 readonly delegate?: Tracer.Tracer;
185 readonly sink?: TraceSink;
186}
187
188type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number];
189type OtlpSpanEvent = OtlpSpan["events"][number];
190type OtlpSpanLink = OtlpSpan["links"][number];
191type OtlpSpanStatus = OtlpSpan["status"];
192type OtlpScope = OtlpTracer.ScopeSpan["scope"];
193
194interface SerializableSpan {
195 readonly name: string;
196 readonly traceId: string;
197 readonly spanId: string;
198 readonly parent: Option.Option<Tracer.AnySpan>;
199 readonly status: Tracer.SpanStatus;
200 readonly sampled: boolean;
201 readonly kind: Tracer.SpanKind;
202 readonly attributes: ReadonlyMap<string, unknown>;
203 readonly links: ReadonlyArray<Tracer.SpanLink>;
204 readonly events: ReadonlyArray<
205 readonly [name: string, startTime: bigint, attributes: Record<string, unknown>]
206 >;
207}
208
209function isPlainObject(value: unknown): value is Record<string, unknown> {
210 return typeof value === "object" && value !== null && !Array.isArray(value);
211}
212
213// Repeated references expand once per path, so a small shared graph can grow exponentially.
214const TRACE_ATTRIBUTE_MAX_NODES = 10_000;
215
216interface NormalizeState {
217 // Only the current ancestors, so a value repeated under two keys is not a cycle.
218 readonly seen: WeakSet<object>;
219 remainingNodes: number;
220}
221
222function enterNode(value: object, state: NormalizeState): string | undefined {
223 if (state.seen.has(value)) {
224 return "[Circular]";
225 }
226 if (state.remainingNodes <= 0) {
227 return "[Truncated]";
228 }
229 state.remainingNodes--;
230 state.seen.add(value);
231 return undefined;
232}
233
234function normalizeJsonValue(value: unknown, state: NormalizeState): unknown {
235 if (
236 value === null ||
237 value === undefined ||
238 typeof value === "string" ||
239 typeof value === "number" ||
240 typeof value === "boolean"
241 ) {
242 return value ?? null;
243 }
244 if (typeof value === "bigint") {
245 return value.toString();
246 }
247 if (value instanceof Date) {
248 return Number.isNaN(value.getTime()) ? "Invalid Date" : value.toISOString();
249 }
250 if (value instanceof Error) {
251 return {
252 name: value.name,
253 message: value.message,
254 ...(value.stack ? { stack: value.stack } : {}),
255 };
256 }
257 if (Array.isArray(value)) {
258 const marker = enterNode(value, state);
259 if (marker) {
260 return marker;
261 }
262 const normalized = value.map((entry) => normalizeJsonValue(entry, state));
263 state.seen.delete(value);
264 return normalized;
265 }
266 if (value instanceof Map) {
267 const marker = enterNode(value, state);
268 if (marker) {
269 return marker;
270 }
271 const normalized = Object.fromEntries(
272 Array.from(value.entries(), ([key, entryValue]) => [
273 String(key),
274 normalizeJsonValue(entryValue, state),
275 ]),
276 );
277 state.seen.delete(value);
278 return normalized;
279 }
280 if (value instanceof Set) {
281 const marker = enterNode(value, state);
282 if (marker) {
283 return marker;
284 }
285 const normalized = Array.from(value.values(), (entry) => normalizeJsonValue(entry, state));
286 state.seen.delete(value);
287 return normalized;
288 }
289 if (!isPlainObject(value)) {
290 return String(value);
291 }
292 const marker = enterNode(value, state);
293 if (marker) {
294 return marker;
295 }
296 const normalized = Object.fromEntries(
297 Object.entries(value).map(([key, entryValue]) => [key, normalizeJsonValue(entryValue, state)]),
298 );
299 state.seen.delete(value);
300 return normalized;
301}
302
303export function compactTraceAttributes(
304 attributes: Readonly<Record<string, unknown>>,
305): TraceAttributes {
306 const state: NormalizeState = { seen: new WeakSet(), remainingNodes: TRACE_ATTRIBUTE_MAX_NODES };
307 const entries: Array<[string, unknown]> = [];
308 for (const [key, value] of Object.entries(attributes)) {
309 if (value !== undefined) {
310 entries.push([key, normalizeJsonValue(value, state)]);
311 }
312 }
313 return Object.fromEntries(entries);
314}
315
316function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] {
317 if (ExitRuntime.isSuccess(exit)) {
318 return { _tag: "Success" };
319 }
320 if (Cause.hasInterruptsOnly(exit.cause)) {
321 return {
322 _tag: "Interrupted",
323 cause: Cause.pretty(exit.cause),
324 };
325 }
326 return {
327 _tag: "Failure",
328 cause: Cause.pretty(exit.cause),
329 };
330}
331
332const TRACE_ATTRIBUTE_MAX_LENGTH = 500;
333const TRACE_ATTRIBUTE_TRUNCATED_LENGTH = 200;
334const TRACE_ATTRIBUTE_TRUNCATION_SUFFIX = "…[truncated]";
335const ALWAYS_TRUNCATED_TRACE_ATTRIBUTES: ReadonlySet<string> = new Set(["db.query.text"]);
336
337function truncateNestedValue(value: unknown): unknown {
338 if (typeof value === "string") {
339 return value.length <= TRACE_ATTRIBUTE_MAX_LENGTH
340 ? value
341 : `${value.slice(0, TRACE_ATTRIBUTE_MAX_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`;
342 }
343 if (Array.isArray(value)) {
344 const truncated = value.map(truncateNestedValue);
345 return truncated.some((entry, index) => entry !== value[index]) ? truncated : value;
346 }
347 if (isPlainObject(value)) {
348 let truncated: Record<string, unknown> | undefined;
349 for (const [key, entry] of Object.entries(value)) {
350 const next = truncateNestedValue(entry);
351 if (next === entry) continue;
352 truncated ??= { ...value };
353 truncated[key] = next;
354 }
355 return truncated ?? value;
356 }
357 return value;
358}
359
360/** Clones only when clamping; never mutates attributes shared with other tracers. */
361export function truncateTraceAttributes(attributes: TraceAttributes): TraceAttributes {
362 let truncated: Record<string, unknown> | undefined;
363 for (const [key, value] of Object.entries(attributes)) {
364 if (typeof value === "string" && ALWAYS_TRUNCATED_TRACE_ATTRIBUTES.has(key)) {
365 if (value.length <= TRACE_ATTRIBUTE_TRUNCATED_LENGTH) continue;
366 truncated ??= { ...attributes };
367 truncated[key] =
368 `${value.slice(0, TRACE_ATTRIBUTE_TRUNCATED_LENGTH)}${TRACE_ATTRIBUTE_TRUNCATION_SUFFIX}`;
369 continue;
370 }
371 const next = truncateNestedValue(value);
372 if (next === value) continue;
373 truncated ??= { ...attributes };
374 truncated[key] = next;
375 }
376 return truncated ?? attributes;
377}
378
379function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {
380 const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>;
381 const parentSpanId = Option.getOrUndefined(span.parent)?.spanId;
382
383 return {
384 type: "effect-span",
385 name: span.name,
386 traceId: span.traceId,
387 spanId: span.spanId,
388 ...(parentSpanId ? { parentSpanId } : {}),
389 sampled: span.sampled,
390 kind: span.kind,
391 startTimeUnixNano: String(status.startTime),
392 endTimeUnixNano: String(status.endTime),
393 durationMs: Number(status.endTime - status.startTime) / 1_000_000,
394 attributes: truncateTraceAttributes(
395 compactTraceAttributes(Object.fromEntries(span.attributes)),
396 ),
397 events: span.events.map(([name, startTime, attributes]) => ({
398 name,
399 timeUnixNano: String(startTime),
400 attributes: truncateTraceAttributes(compactTraceAttributes(attributes)),
401 })),
402 links: span.links.map((link) => ({
403 traceId: link.span.traceId,
404 spanId: link.span.spanId,
405 attributes: truncateTraceAttributes(compactTraceAttributes(link.attributes)),
406 })),
407 exit: formatTraceExit(status.exit),
408 };
409}
410
411// WebSocket tickets and their DPoP proofs ride in URL queries and stay valid after the request.
412const URL_CREDENTIAL_PARAMS = /(^|[?&"\s])(wsTicket|dpop)=[^&#"\\\s]*/g;
413
414const redactUrlCredentials = (text: string) =>
415 text.replace(URL_CREDENTIAL_PARAMS, "$1$2=<redacted>");
416
417export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) {
418 const sink = yield* makeAsyncRotatingFileSink({
419 filePath: options.filePath,
420 maxBytes: options.maxBytes,
421 maxFiles: options.maxFiles,
422 }).pipe(Effect.orDie);
423
424 let buffer: Array<{ readonly text: string; readonly bytes: number }> = [];
425 let bufferBytes = 0;
426 const maxBufferBytes = Math.max(options.maxBytes, 4 * 1024 * 1024);
427 const flushRequests = yield* Queue.sliding<void>(1);
428 const flushSemaphore = yield* Semaphore.make(1);
429 let writeFailing = false;
430 let droppedCount = 0;
431 let failureReported = false;
432 let pendingFlushStats: TraceSinkFlushStats = {
433 logicalWriteBytes: 0,
434 count: 0,
435 durationMs: 0,
436 };
437
438 const flushBuffered = Effect.gen(function* () {
439 if (buffer.length === 0) {
440 return;
441 }
442
443 const records = buffer;
444 buffer = [];
445 bufferBytes = 0;
446 let persistedCount = 0;
447
448 while (persistedCount < records.length) {
449 const firstRecordBytes = records[persistedCount]!.bytes;
450 let nextIndex = persistedCount + 1;
451 let chunkBytes = firstRecordBytes;
452 while (nextIndex < records.length) {
453 const nextRecordBytes = records[nextIndex]!.bytes;
454 if (chunkBytes + nextRecordBytes > options.maxBytes) break;
455 chunkBytes += nextRecordBytes;
456 nextIndex += 1;
457 }
458
459 const chunk = records
460 .slice(persistedCount, nextIndex)
461 .map((record) => record.text)
462 .join("");
463 const startedAt = performance.now();
464 const written = yield* sink.write(chunk).pipe(Effect.result);
465 if (Result.isFailure(written)) {
466 // Drop the batch after disk failure to avoid an ever-growing retry backlog.
467 writeFailing = true;
468 droppedCount += records.length - persistedCount;
469 return;
470 }
471 writeFailing = false;
472 pendingFlushStats = {
473 logicalWriteBytes: pendingFlushStats.logicalWriteBytes + chunkBytes,
474 count: pendingFlushStats.count + nextIndex - persistedCount,
475 durationMs: pendingFlushStats.durationMs + Math.max(0, performance.now() - startedAt),
476 };
477 persistedCount = nextIndex;
478 }
479 });
480
481 const flush = flushSemaphore
482 .withPermits(1)(
483 Effect.gen(function* () {
484 const droppedAtStart = droppedCount;
485 yield* flushBuffered;
486 const stats = pendingFlushStats;
487 pendingFlushStats = {
488 logicalWriteBytes: 0,
489 count: 0,
490 durationMs: 0,
491 };
492 if (stats.count > 0 && options.onFlush) {
493 yield* options.onFlush(stats).pipe(Effect.ignore);
494 }
495 if (droppedCount > 0 && !failureReported) {
496 failureReported = true;
497 yield* Effect.logWarning(
498 "Trace writer could not keep up, dropping records until it recovers",
499 {
500 filePath: options.filePath,
501 },
502 );
503 } else if (failureReported && !writeFailing && droppedCount === droppedAtStart) {
504 yield* Effect.logInfo("Trace writes recovered", {
505 filePath: options.filePath,
506 droppedCount,
507 });
508 failureReported = false;
509 droppedCount = 0;
510 }
511 }),
512 )
513 .pipe(
514 // Cancelled OS writes still finish; await completion before rotation or finalizer drain.
515 Effect.uninterruptible,
516 Effect.withTracerEnabled(false),
517 // Drop the ended makeTraceSink parent span so timed-fiber logs cannot accumulate on it.
518 Effect.updateContext(Context.omit(Tracer.ParentSpan)<never>),
519 );
520
521 yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));
522 yield* Effect.forkScoped(
523 Queue.take(flushRequests).pipe(
524 Effect.timeoutOption(`${options.batchWindowMs} millis`),
525 Effect.andThen(flush),
526 Effect.forever,
527 ),
528 );
529
530 return {
531 filePath: options.filePath,
532 push(record) {
533 try {
534 const text = `${redactUrlCredentials(JSON.stringify(record))}\n`;
535 const bytes = Buffer.byteLength(text);
536 if (bytes > options.maxBytes) return;
537 if (bufferBytes + bytes > maxBufferBytes) {
538 droppedCount++;
539 return;
540 }
541 buffer.push({ text, bytes });
542 bufferBytes += bytes;
543 if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {
544 Queue.offerUnsafe(flushRequests, undefined);
545 }
546 } catch {
547 return;
548 }
549 },
550 flush,
551 flushSync() {
552 if (buffer.length === 0) return;
553 const text = buffer.map((record) => record.text).join("");
554 buffer = [];
555 bufferBytes = 0;
556 try {
557 NodeFS.appendFileSync(options.filePath, text);
558 } catch {
559 return;
560 }
561 },
562 close: () => flush,
563 } satisfies TraceSink;
564});
565
566// Effect's HTTP errors, including failed upgrades, put the request URL in their message.
567export function redactCauseCredentials(cause: Cause.Cause<unknown>): Cause.Cause<unknown> {
568 if (cause.reasons.length === 0 || Cause.hasInterruptsOnly(cause)) {
569 return cause;
570 }
571 const pretty = Cause.pretty(cause);
572 const redacted = redactUrlCredentials(pretty);
573 return redacted === pretty ? cause : Cause.die(new Error(redacted));
574}
575
576function redactExitCredentials(exit: Exit.Exit<unknown, unknown>): Exit.Exit<unknown, unknown> {
577 if (ExitRuntime.isSuccess(exit)) {
578 return exit;
579 }
580 const cause = redactCauseCredentials(exit.cause);
581 return cause === exit.cause ? exit : ExitRuntime.failCause(cause);
582}
583
584class LocalFileSpan implements Tracer.Span {
585 readonly _tag = "Span";
586 readonly name: string;
587 readonly spanId: string;
588 readonly traceId: string;
589 readonly parent: Option.Option<Tracer.AnySpan>;
590 readonly annotations: Tracer.Span["annotations"];
591 readonly links: Array<Tracer.SpanLink>;
592 readonly sampled: boolean;
593 readonly kind: Tracer.SpanKind;
594
595 status: Tracer.SpanStatus;
596 attributes: Map<string, unknown>;
597 events: Array<[name: string, startTime: bigint, attributes: Record<string, unknown>]>;
598 private readonly delegate: Tracer.Span;
599 private readonly push: (record: EffectTraceRecord) => void;
600
601 constructor(
602 options: Parameters<Tracer.Tracer["span"]>[0],
603 delegate: Tracer.Span,
604 push: (record: EffectTraceRecord) => void,
605 ) {
606 this.delegate = delegate;
607 this.push = push;
608 this.name = delegate.name;
609 this.spanId = delegate.spanId;
610 this.traceId = delegate.traceId;
611 this.parent = options.parent;
612 this.annotations = options.annotations;
613 this.links = [...options.links];
614 this.sampled = delegate.sampled;
615 this.kind = delegate.kind;
616 this.status = {
617 _tag: "Started",
618 startTime: options.startTime,
619 };
620 this.attributes = new Map();
621 this.events = [];
622 }
623
624 end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void {
625 // Cached lookup fibers retain parent spans. Traces need success status, not result payloads.
626 const traceExit = ExitRuntime.isSuccess(exit) ? ExitRuntime.void : exit;
627 this.status = {
628 _tag: "Ended",
629 startTime: this.status.startTime,
630 endTime,
631 exit: traceExit,
632 };
633 this.delegate.end(endTime, redactExitCredentials(traceExit));
634
635 if (this.sampled) {
636 this.push(spanToTraceRecord(this));
637 }
638 }
639
640 attribute(key: string, value: unknown): void {
641 const recorded =
642 (key === "url.full" || key === "url.query") && typeof value === "string"
643 ? redactUrlCredentials(value)
644 : value;
645 this.attributes.set(key, recorded);
646 this.delegate.attribute(key, recorded);
647 }
648
649 event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void {
650 const cause = attributes?.["effect.cause"];
651 const nextAttributes =
652 typeof cause === "string"
653 ? { ...attributes, "effect.cause": redactUrlCredentials(cause) }
654 : (attributes ?? {});
655 this.events.push([name, startTime, nextAttributes]);
656 this.delegate.event(name, startTime, nextAttributes);
657 }
658
659 addLinks(links: ReadonlyArray<Tracer.SpanLink>): void {
660 this.links.push(...links);
661 this.delegate.addLinks(links);
662 }
663}
664
665export const makeLocalFileTracer = Effect.fn("makeLocalFileTracer")(function* (
666 options: LocalFileTracerOptions,
667) {
668 const sink =
669 options.sink ??
670 (yield* makeTraceSink({
671 filePath: options.filePath,
672 maxBytes: options.maxBytes,
673 maxFiles: options.maxFiles,
674 batchWindowMs: options.batchWindowMs,
675 ...(options.onFlush ? { onFlush: options.onFlush } : {}),
676 }));
677
678 const delegate =
679 options.delegate ??
680 Tracer.make({
681 span: (spanOptions) => new Tracer.NativeSpan(spanOptions),
682 });
683
684 return Tracer.make({
685 span(spanOptions) {
686 return new LocalFileSpan(spanOptions, delegate.span(spanOptions), sink.push);
687 },
688 ...(delegate.context ? { context: delegate.context } : {}),
689 });
690});
691
692const SPAN_KIND_MAP: Record<number, OtlpTraceRecord["kind"]> = {
693 1: "internal",
694 2: "server",
695 3: "client",
696 4: "producer",
697 5: "consumer",
698};
699
700export function decodeOtlpTraceRecords(
701 payload: OtlpTracer.TraceData,
702): ReadonlyArray<OtlpTraceRecord> {
703 const records: Array<OtlpTraceRecord> = [];
704
705 // OTLP/JSON follows protobuf JSON, which omits unset and empty fields, so a
706 // valid batch can leave out any field read here.
707 for (const resourceSpan of payload.resourceSpans ?? []) {
708 const resourceAttributes = decodeAttributes(resourceSpan.resource?.attributes);
709
710 for (const scopeSpan of resourceSpan.scopeSpans ?? []) {
711 const scope: Partial<OtlpScope> = scopeSpan.scope ?? {};
712 for (const span of scopeSpan.spans ?? []) {
713 records.push(
714 otlpSpanToTraceRecord({
715 resourceAttributes,
716 scopeAttributes: decodeAttributes(
717 "attributes" in scope && Array.isArray(scope.attributes) ? scope.attributes : [],
718 ),
719 scopeName: scope.name,
720 scopeVersion: typeof scope.version === "string" ? scope.version : undefined,
721 span,
722 }),
723 );
724 }
725 }
726 }
727
728 return records;
729}
730
731function otlpSpanToTraceRecord(input: {
732 readonly resourceAttributes: Readonly<Record<string, unknown>>;
733 readonly scopeAttributes: Readonly<Record<string, unknown>>;
734 readonly scopeName: string | undefined;
735 readonly scopeVersion: string | undefined;
736 readonly span: OtlpSpan;
737}): OtlpTraceRecord {
738 return {
739 type: "otlp-span",
740 name: input.span.name,
741 traceId: input.span.traceId,
742 spanId: input.span.spanId,
743 ...(input.span.parentSpanId ? { parentSpanId: input.span.parentSpanId } : {}),
744 sampled: true,
745 kind: normalizeSpanKind(input.span.kind),
746 startTimeUnixNano: input.span.startTimeUnixNano,
747 endTimeUnixNano: input.span.endTimeUnixNano,
748 durationMs:
749 Number(parseBigInt(input.span.endTimeUnixNano) - parseBigInt(input.span.startTimeUnixNano)) /
750 1_000_000,
751 attributes: decodeAttributes(input.span.attributes),
752 resourceAttributes: input.resourceAttributes,
753 scope: {
754 ...(input.scopeName ? { name: input.scopeName } : {}),
755 ...(input.scopeVersion ? { version: input.scopeVersion } : {}),
756 attributes: input.scopeAttributes,
757 },
758 events: decodeEvents(input.span.events),
759 links: decodeLinks(input.span.links),
760 status: decodeStatus(input.span.status),
761 };
762}
763
764function decodeStatus(input: OtlpSpanStatus | undefined): OtlpTraceRecord["status"] {
765 const code = String(input?.code ?? 0);
766 const message = input?.message;
767
768 return {
769 code,
770 ...(message ? { message } : {}),
771 };
772}
773
774function decodeEvents(
775 input: ReadonlyArray<OtlpSpanEvent> | undefined,
776): ReadonlyArray<TraceRecordEvent> {
777 return (input ?? []).map((current) => ({
778 name: current.name,
779 timeUnixNano: current.timeUnixNano,
780 attributes: decodeAttributes(current.attributes),
781 }));
782}
783
784function decodeLinks(
785 input: ReadonlyArray<OtlpSpanLink> | undefined,
786): ReadonlyArray<TraceRecordLink> {
787 return (input ?? []).flatMap((current) => {
788 const traceId = current.traceId;
789 const spanId = current.spanId;
790 return {
791 traceId,
792 spanId,
793 attributes: decodeAttributes(current.attributes),
794 };
795 });
796}
797
798function decodeAttributes(
799 input: ReadonlyArray<OtlpResource.KeyValue> | undefined,
800): Readonly<Record<string, unknown>> {
801 const entries: Record<string, unknown> = {};
802
803 for (const attribute of input ?? []) {
804 entries[attribute.key] = decodeValue(attribute.value);
805 }
806
807 return truncateTraceAttributes(compactTraceAttributes(entries));
808}
809
810function decodeValue(input: OtlpResource.AnyValue | null | undefined): unknown {
811 if (input == null) {
812 return null;
813 }
814 if ("stringValue" in input) {
815 return input.stringValue;
816 }
817 if ("boolValue" in input) {
818 return input.boolValue;
819 }
820 if ("intValue" in input) {
821 return input.intValue;
822 }
823 if ("doubleValue" in input) {
824 return input.doubleValue;
825 }
826 if ("bytesValue" in input) {
827 return input.bytesValue;
828 }
829 if (input.arrayValue) {
830 return (input.arrayValue.values ?? []).map((entry) => decodeValue(entry));
831 }
832 if (input.kvlistValue) {
833 return decodeAttributes(input.kvlistValue.values);
834 }
835 return null;
836}
837
838function normalizeSpanKind(input: number): OtlpTraceRecord["kind"] {
839 return SPAN_KIND_MAP[input] || "internal";
840}
841
842function parseBigInt(input: string): bigint {
843 try {
844 return BigInt(input);
845 } catch {
846 return 0n;
847 }
848}
849
850export const OtlpHeadersFromString = Schema.String.pipe(
851 Schema.decodeTo(
852 Schema.Record(Schema.String, Schema.String),
853 SchemaTransformation.transformEffect({
854 decode: (input) => {
855 const headers: Record<string, string> = {};
856 for (const pair of input.split(",")) {
857 if (pair.trim() === "") {
858 continue;
859 }
860 const separator = pair.indexOf("=");
861 const key = separator === -1 ? "" : pair.slice(0, separator).trim();
862 if (key === "") {
863 return Effect.fail(
864 new SchemaIssue.InvalidValue({
865 message: `Expected key=value but received ${JSON.stringify(pair.trim())}.`,
866 }),
867 );
868 }
869 try {
870 headers[key] = decodeURIComponent(pair.slice(separator + 1).trim());
871 } catch {
872 return Effect.fail(
873 new SchemaIssue.InvalidValue({
874 message: `Header ${JSON.stringify(key)} has a malformed percent-encoded value.`,
875 }),
876 );
877 }
878 }
879 return Effect.succeed(headers);
880 },
881 encode: (headers) =>
882 Effect.succeed(
883 Object.entries(headers)
884 .map(([key, value]) => `${key}=${encodeURIComponent(value)}`)
885 .join(","),
886 ),
887 }),
888 ),
889);