572 lines · 15.5 KB · TypeScript
RawDownload
import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import type * as Exit from "effect/Exit";import * as ExitRuntime from "effect/Exit";import * as Option from "effect/Option";import * as Tracer from "effect/Tracer";import { OtlpResource, OtlpTracer } from "effect/unstable/observability"; import { RotatingFileSink } from "./logging.ts"; const FLUSH_BUFFER_THRESHOLD = 32; export type TraceAttributes = Readonly<Record<string, unknown>>; export interface TraceRecordEvent {  readonly name: string;  readonly timeUnixNano: string;  readonly attributes: Readonly<Record<string, unknown>>;} export interface TraceRecordLink {  readonly traceId: string;  readonly spanId: string;  readonly attributes: Readonly<Record<string, unknown>>;} interface BaseTraceRecord {  readonly name: string;  readonly kind: string;  readonly traceId: string;  readonly spanId: string;  readonly parentSpanId?: string;  readonly sampled: boolean;  readonly startTimeUnixNano: string;  readonly endTimeUnixNano: string;  readonly durationMs: number;  readonly attributes: Readonly<Record<string, unknown>>;  readonly events: ReadonlyArray<TraceRecordEvent>;  readonly links: ReadonlyArray<TraceRecordLink>;} export interface EffectTraceRecord extends BaseTraceRecord {  readonly type: "effect-span";  readonly exit:    | {        readonly _tag: "Success";      }    | {        readonly _tag: "Interrupted";        readonly cause: string;      }    | {        readonly _tag: "Failure";        readonly cause: string;      };} export interface OtlpTraceRecord extends BaseTraceRecord {  readonly type: "otlp-span";  readonly resourceAttributes: Readonly<Record<string, unknown>>;  readonly scope: Readonly<{    readonly name?: string;    readonly version?: string;    readonly attributes: Readonly<Record<string, unknown>>;  }>;  readonly status?:    | {        readonly code?: string;        readonly message?: string;      }    | undefined;} export type TraceRecord = EffectTraceRecord | OtlpTraceRecord; function isStructuralTag(value: unknown): value is string {  return (    typeof value === "string" &&    value.length > 0 &&    value.length <= 128 &&    /^[A-Za-z][A-Za-z0-9._:/-]*$/.test(value)  );} export function errorTag(error: unknown): string {  try {    if (typeof error === "object" && error !== null && "_tag" in error) {      return isStructuralTag(error._tag) ? error._tag : "TaggedError";    }    if (error instanceof Error) {      return isStructuralTag(error.name) ? error.name : "Error";    }  } catch {    return "UnknownError";  }  return typeof error;} export function causeErrorTag(cause: Cause.Cause<unknown>): string {  const failure = Cause.findErrorOption(cause);  if (Option.isSome(failure)) {    return errorTag(failure.value);  }  return cause.reasons[0]?._tag ?? "Empty";} export interface TraceSinkOptions {  readonly filePath: string;  readonly maxBytes: number;  readonly maxFiles: number;  readonly batchWindowMs: number;} export interface TraceSink {  readonly filePath: string;  push: (record: TraceRecord) => void;  flush: Effect.Effect<void>;  close: () => Effect.Effect<void>;} export interface LocalFileTracerOptions extends TraceSinkOptions {  readonly delegate?: Tracer.Tracer;  readonly sink?: TraceSink;} type OtlpSpan = OtlpTracer.ScopeSpan["spans"][number];type OtlpSpanEvent = OtlpSpan["events"][number];type OtlpSpanLink = OtlpSpan["links"][number];type OtlpSpanStatus = OtlpSpan["status"]; interface SerializableSpan {  readonly name: string;  readonly traceId: string;  readonly spanId: string;  readonly parent: Option.Option<Tracer.AnySpan>;  readonly status: Tracer.SpanStatus;  readonly sampled: boolean;  readonly kind: Tracer.SpanKind;  readonly attributes: ReadonlyMap<string, unknown>;  readonly links: ReadonlyArray<Tracer.SpanLink>;  readonly events: ReadonlyArray<    readonly [name: string, startTime: bigint, attributes: Record<string, unknown>]  >;} function isPlainObject(value: unknown): value is Record<string, unknown> {  return typeof value === "object" && value !== null && !Array.isArray(value);} function markSeen(value: object, seen: WeakSet<object>): boolean {  if (seen.has(value)) {    return true;  }  seen.add(value);  return false;} function normalizeJsonValue(value: unknown, seen: WeakSet<object> = new WeakSet()): unknown {  if (    value === null ||    value === undefined ||    typeof value === "string" ||    typeof value === "number" ||    typeof value === "boolean"  ) {    return value ?? null;  }  if (typeof value === "bigint") {    return value.toString();  }  if (value instanceof Date) {    return Number.isNaN(value.getTime()) ? "Invalid Date" : value.toISOString();  }  if (value instanceof Error) {    return {      name: value.name,      message: value.message,      ...(value.stack ? { stack: value.stack } : {}),    };  }  if (Array.isArray(value)) {    if (markSeen(value, seen)) {      return "[Circular]";    }    return value.map((entry) => normalizeJsonValue(entry, seen));  }  if (value instanceof Map) {    if (markSeen(value, seen)) {      return "[Circular]";    }    return Object.fromEntries(      Array.from(value.entries(), ([key, entryValue]) => [        String(key),        normalizeJsonValue(entryValue, seen),      ]),    );  }  if (value instanceof Set) {    if (markSeen(value, seen)) {      return "[Circular]";    }    return Array.from(value.values(), (entry) => normalizeJsonValue(entry, seen));  }  if (!isPlainObject(value)) {    return String(value);  }  if (markSeen(value, seen)) {    return "[Circular]";  }  return Object.fromEntries(    Object.entries(value).map(([key, entryValue]) => [key, normalizeJsonValue(entryValue, seen)]),  );} export function compactTraceAttributes(  attributes: Readonly<Record<string, unknown>>,): TraceAttributes {  const entries: Array<[string, unknown]> = [];  for (const [key, value] of Object.entries(attributes)) {    if (value !== undefined) {      entries.push([key, normalizeJsonValue(value)]);    }  }  return Object.fromEntries(entries);} function formatTraceExit(exit: Exit.Exit<unknown, unknown>): EffectTraceRecord["exit"] {  if (ExitRuntime.isSuccess(exit)) {    return { _tag: "Success" };  }  if (Cause.hasInterruptsOnly(exit.cause)) {    return {      _tag: "Interrupted",      cause: Cause.pretty(exit.cause),    };  }  return {    _tag: "Failure",    cause: Cause.pretty(exit.cause),  };} export function spanToTraceRecord(span: SerializableSpan): EffectTraceRecord {  const status = span.status as Extract<Tracer.SpanStatus, { _tag: "Ended" }>;  const parentSpanId = Option.getOrUndefined(span.parent)?.spanId;   return {    type: "effect-span",    name: span.name,    traceId: span.traceId,    spanId: span.spanId,    ...(parentSpanId ? { parentSpanId } : {}),    sampled: span.sampled,    kind: span.kind,    startTimeUnixNano: String(status.startTime),    endTimeUnixNano: String(status.endTime),    durationMs: Number(status.endTime - status.startTime) / 1_000_000,    attributes: compactTraceAttributes(Object.fromEntries(span.attributes)),    events: span.events.map(([name, startTime, attributes]) => ({      name,      timeUnixNano: String(startTime),      attributes: compactTraceAttributes(attributes),    })),    links: span.links.map((link) => ({      traceId: link.span.traceId,      spanId: link.span.spanId,      attributes: compactTraceAttributes(link.attributes),    })),    exit: formatTraceExit(status.exit),  };} export const makeTraceSink = Effect.fn("makeTraceSink")(function* (options: TraceSinkOptions) {  const sink = new RotatingFileSink({    filePath: options.filePath,    maxBytes: options.maxBytes,    maxFiles: options.maxFiles,  });   let buffer: Array<string> = [];   const flushUnsafe = () => {    if (buffer.length === 0) {      return;    }     const chunk = buffer.join("");    buffer = [];     try {      sink.write(chunk);    } catch {      buffer.unshift(chunk);    }  };   const flush = Effect.sync(flushUnsafe).pipe(Effect.withTracerEnabled(false));   yield* Effect.addFinalizer(() => flush.pipe(Effect.ignore));  yield* Effect.forkScoped(    Effect.sleep(`${options.batchWindowMs} millis`).pipe(Effect.andThen(flush), Effect.forever),  );   return {    filePath: options.filePath,    push(record) {      try {        buffer.push(`${JSON.stringify(record)}\n`);        if (buffer.length >= FLUSH_BUFFER_THRESHOLD) {          flushUnsafe();        }      } catch {        return;      }    },    flush,    close: () => flush,  } satisfies TraceSink;}); class LocalFileSpan implements Tracer.Span {  readonly _tag = "Span";  readonly name: string;  readonly spanId: string;  readonly traceId: string;  readonly parent: Option.Option<Tracer.AnySpan>;  readonly annotations: Tracer.Span["annotations"];  readonly links: Array<Tracer.SpanLink>;  readonly sampled: boolean;  readonly kind: Tracer.SpanKind;   status: Tracer.SpanStatus;  attributes: Map<string, unknown>;  events: Array<[name: string, startTime: bigint, attributes: Record<string, unknown>]>;  private readonly delegate: Tracer.Span;  private readonly push: (record: EffectTraceRecord) => void;   constructor(    options: Parameters<Tracer.Tracer["span"]>[0],    delegate: Tracer.Span,    push: (record: EffectTraceRecord) => void,  ) {    this.delegate = delegate;    this.push = push;    this.name = delegate.name;    this.spanId = delegate.spanId;    this.traceId = delegate.traceId;    this.parent = options.parent;    this.annotations = options.annotations;    this.links = [...options.links];    this.sampled = delegate.sampled;    this.kind = delegate.kind;    this.status = {      _tag: "Started",      startTime: options.startTime,    };    this.attributes = new Map();    this.events = [];  }   end(endTime: bigint, exit: Exit.Exit<unknown, unknown>): void {    this.status = {      _tag: "Ended",      startTime: this.status.startTime,      endTime,      exit,    };    this.delegate.end(endTime, exit);     if (this.sampled) {      this.push(spanToTraceRecord(this));    }  }   attribute(key: string, value: unknown): void {    this.attributes.set(key, value);    this.delegate.attribute(key, value);  }   event(name: string, startTime: bigint, attributes?: Record<string, unknown>): void {    const nextAttributes = attributes ?? {};    this.events.push([name, startTime, nextAttributes]);    this.delegate.event(name, startTime, nextAttributes);  }   addLinks(links: ReadonlyArray<Tracer.SpanLink>): void {    this.links.push(...links);    this.delegate.addLinks(links);  }} export const makeLocalFileTracer = Effect.fn("makeLocalFileTracer")(function* (  options: LocalFileTracerOptions,) {  const sink =    options.sink ??    (yield* makeTraceSink({      filePath: options.filePath,      maxBytes: options.maxBytes,      maxFiles: options.maxFiles,      batchWindowMs: options.batchWindowMs,    }));   const delegate =    options.delegate ??    Tracer.make({      span: (spanOptions) => new Tracer.NativeSpan(spanOptions),    });   return Tracer.make({    span(spanOptions) {      return new LocalFileSpan(spanOptions, delegate.span(spanOptions), sink.push);    },    ...(delegate.context ? { context: delegate.context } : {}),  });}); const SPAN_KIND_MAP: Record<number, OtlpTraceRecord["kind"]> = {  1: "internal",  2: "server",  3: "client",  4: "producer",  5: "consumer",}; export function decodeOtlpTraceRecords(  payload: OtlpTracer.TraceData,): ReadonlyArray<OtlpTraceRecord> {  const records: Array<OtlpTraceRecord> = [];   for (const resourceSpan of payload.resourceSpans) {    const resourceAttributes = decodeAttributes(resourceSpan.resource?.attributes ?? []);     for (const scopeSpan of resourceSpan.scopeSpans) {      for (const span of scopeSpan.spans) {        records.push(          otlpSpanToTraceRecord({            resourceAttributes,            scopeAttributes: decodeAttributes(              "attributes" in scopeSpan.scope && Array.isArray(scopeSpan.scope.attributes)                ? scopeSpan.scope.attributes                : [],            ),            scopeName: scopeSpan.scope.name,            scopeVersion:              "version" in scopeSpan.scope && typeof scopeSpan.scope.version === "string"                ? scopeSpan.scope.version                : undefined,            span,          }),        );      }    }  }   return records;} function otlpSpanToTraceRecord(input: {  readonly resourceAttributes: Readonly<Record<string, unknown>>;  readonly scopeAttributes: Readonly<Record<string, unknown>>;  readonly scopeName: string | undefined;  readonly scopeVersion: string | undefined;  readonly span: OtlpSpan;}): OtlpTraceRecord {  return {    type: "otlp-span",    name: input.span.name,    traceId: input.span.traceId,    spanId: input.span.spanId,    ...(input.span.parentSpanId ? { parentSpanId: input.span.parentSpanId } : {}),    sampled: true,    kind: normalizeSpanKind(input.span.kind),    startTimeUnixNano: input.span.startTimeUnixNano,    endTimeUnixNano: input.span.endTimeUnixNano,    durationMs:      Number(parseBigInt(input.span.endTimeUnixNano) - parseBigInt(input.span.startTimeUnixNano)) /      1_000_000,    attributes: decodeAttributes(input.span.attributes),    resourceAttributes: input.resourceAttributes,    scope: {      ...(input.scopeName ? { name: input.scopeName } : {}),      ...(input.scopeVersion ? { version: input.scopeVersion } : {}),      attributes: input.scopeAttributes,    },    events: decodeEvents(input.span.events),    links: decodeLinks(input.span.links),    status: decodeStatus(input.span.status),  };} function decodeStatus(input: OtlpSpanStatus): OtlpTraceRecord["status"] {  const code = String(input.code);  const message = input.message;   return {    code,    ...(message ? { message } : {}),  };}

Lines 1–500 of 572.