1import * as Config from "effect/Config";
2import * as Context from "effect/Context";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Redacted from "effect/Redacted";
7import * as Result from "effect/Result";
8import type * as Scope from "effect/Scope";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import * as OtlpLogger from "effect/observability/OtlpLogger";
12import * as OtlpMetrics from "effect/observability/OtlpMetrics";
13import * as OtlpSerialization from "effect/observability/OtlpSerialization";
14import * as OtlpTracer from "effect/observability/OtlpTracer";
15import { unpackEnvValue } from "./RuntimeContext.ts";
16import type { layer, layerOtlp } from "./Telemetry.ts";
17
18/**
19 * The shape of the {@link Telemetry} reference: a Layer of telemetry
20 * exporters (tracer, loggers, metrics) built once per event into the
21 * event's request scope. Requirements are satisfied from the runtime's
22 * isolate context (`HttpClient`, `ConfigProvider`, …).
23 */
24export type TelemetryLayer = Layer.Layer<never, any, any>;
25
26/**
27 * Read one bound value back at runtime. `rc.set` packs values for the env
28 * var wire (plain values JSON-stringified, `Redacted` as a marker routed
29 * through the secret channel), so the raw env string must be unpacked with
30 * {@link unpackEnvValue} — it handles all three shapes (packed JSON,
31 * Redacted marker, and a raw string set directly in the environment).
32 */
33const readBoundValue = (key: string): Effect.Effect<unknown> =>
34 Config.String(key).pipe(
35 Config.withDefault(undefined),
36 Effect.orElseSucceed(() => undefined),
37 Effect.map((raw) => {
38 if (raw === undefined || raw === "") {
39 return undefined;
40 }
41 const value = unpackEnvValue<unknown>(raw);
42 return Redacted.isRedacted(value) ? Redacted.value(value) : value;
43 }),
44 );
45
46const readBound = (key: string): Effect.Effect<string | undefined> =>
47 readBoundValue(key).pipe(
48 Effect.map((inner) =>
49 inner === undefined
50 ? undefined
51 : typeof inner === "string"
52 ? inner
53 : String(inner),
54 ),
55 );
56
57/**
58 * `service.name` fallback chain for the default exporter: the standard
59 * OTEL variable, then the physical Function/Worker name, then the stack
60 * name. `OtlpResource.fromConfig` dies without a service name, so the
61 * chain must always produce one.
62 */
63const defaultServiceName = Effect.gen(function* () {
64 return (
65 (yield* readBound("OTEL_SERVICE_NAME")) ??
66 (yield* readBound("ALCHEMY_WORKER_NAME")) ??
67 (yield* readBound("AWS_LAMBDA_FUNCTION_NAME")) ??
68 (yield* readBound("ALCHEMY_STACK_NAME")) ??
69 "alchemy"
70 );
71});
72
73const defaultResource = Effect.gen(function* () {
74 const serviceName = yield* defaultServiceName;
75 const stack = yield* readBound("ALCHEMY_STACK_NAME");
76 const stage = yield* readBound("ALCHEMY_STAGE");
77 return {
78 serviceName,
79 attributes: {
80 ...(stack !== undefined ? { "alchemy.stack": stack } : undefined),
81 ...(stage !== undefined ? { "alchemy.stage": stage } : undefined),
82 },
83 };
84});
85
86/**
87 * Parse the OpenTelemetry `OTEL_EXPORTER_OTLP_HEADERS` format:
88 * `key1=value1,key2=value2` with URL-encoded values.
89 */
90const parseOtlpHeaders = (
91 raw: string | undefined,
92): Record<string, string> | undefined => {
93 if (raw === undefined || raw === "") {
94 return undefined;
95 }
96 const headers: Record<string, string> = {};
97 for (const entry of raw.split(",")) {
98 const eq = entry.indexOf("=");
99 if (eq === -1) {
100 continue;
101 }
102 const key = entry.slice(0, eq).trim();
103 if (key === "") {
104 continue;
105 }
106 const value = entry.slice(eq + 1).trim();
107 try {
108 headers[key] = decodeURIComponent(value);
109 } catch {
110 headers[key] = value;
111 }
112 }
113 return headers;
114};
115
116/**
117 * One export target for one signal, as stored in the bound destination
118 * list: a concrete OTLP/HTTP URL and the headers to send with it.
119 */
120export interface ResolvedSignal {
121 url: string;
122 headers?: Record<string, string> | undefined;
123}
124
125/**
126 * One export destination in the bound list — the resolved form of one
127 * {@link layerOtlp} layer.
128 */
129export interface ResolvedDestination {
130 traces?: ResolvedSignal | undefined;
131 logs?: ResolvedSignal | undefined;
132 metrics?: ResolvedSignal | undefined;
133}
134
135/**
136 * The single env binding carrying every configured destination as a JSON
137 * array of {@link ResolvedDestination}. One key (instead of per-signal
138 * keys) is what lets multiple `layerOtlp` layers compose — each layer
139 * build appends its destination and rebinds the full list.
140 */
141export const EXPORTERS_KEY = "ALCHEMY_OTEL_EXPORTERS";
142
143/**
144 * Resolve one signal's endpoint + headers from the standard OpenTelemetry
145 * env vars (`OTEL_EXPORTER_OTLP_{SIGNAL}_ENDPOINT` as-is, or the base
146 * endpoint with `/v1/{signal}` appended; headers per-signal → base). This
147 * forms an *implicit extra destination*, so platform-injected OTLP config
148 * exports without any layer.
149 */
150const signalConfig = (signal: "TRACES" | "LOGS" | "METRICS") =>
151 Effect.gen(function* () {
152 const specific = yield* readBound(`OTEL_EXPORTER_OTLP_${signal}_ENDPOINT`);
153 const base = yield* readBound("OTEL_EXPORTER_OTLP_ENDPOINT");
154 const url =
155 specific !== undefined && specific !== ""
156 ? specific
157 : base !== undefined && base !== ""
158 ? `${base.replace(/\/$/, "")}/v1/${signal.toLowerCase()}`
159 : undefined;
160 if (url === undefined) {
161 return undefined;
162 }
163 const rawHeaders =
164 (yield* readBound(`OTEL_EXPORTER_OTLP_${signal}_HEADERS`)) ??
165 (yield* readBound("OTEL_EXPORTER_OTLP_HEADERS"));
166 return { url, headers: parseOtlpHeaders(rawHeaders) } as ResolvedSignal;
167 });
168
169/**
170 * Sentinel URLs the single exporter set posts to; the fanout client
171 * rewrites each POST to every configured destination for that signal.
172 * Serializing once through one exporter per signal is what keeps span ids
173 * consistent across destinations — Effect has a single `Tracer` service,
174 * so per-destination tracers would generate divergent trace ids.
175 */
176const SENTINEL = {
177 traces: "http://telemetry.alchemy.internal/v1/traces",
178 logs: "http://telemetry.alchemy.internal/v1/logs",
179 metrics: "http://telemetry.alchemy.internal/v1/metrics",
180} as const;
181
182/**
183 * Wrap the ambient `HttpClient` so a POST to a sentinel URL is re-sent to
184 * every destination configured for that signal (with the destination's own
185 * headers). A single healthy destination keeps the exporter alive: the
186 * first 2xx response is returned; per-destination failures are logged at
187 * debug and only surface when every destination fails.
188 */
189const fanoutClient = (
190 routes: ReadonlyMap<string, ResolvedSignal[]>,
191): Layer.Layer<HttpClient.HttpClient, never, HttpClient.HttpClient> =>
192 Layer.effect(
193 HttpClient.HttpClient,
194 Effect.gen(function* () {
195 const base = yield* HttpClient.HttpClient;
196 return HttpClient.transform(base, (effect, request) => {
197 const targets = routes.get(request.url);
198 if (targets === undefined) {
199 return effect;
200 }
201 return Effect.gen(function* () {
202 const results = yield* Effect.all(
203 targets.map((target) =>
204 Effect.result(
205 base.execute(
206 request.pipe(
207 HttpClientRequest.setUrl(target.url),
208 HttpClientRequest.setHeaders(target.headers ?? {}),
209 ),
210 ),
211 ),
212 ),
213 { concurrency: "unbounded" },
214 );
215 const healthy = results.find(
216 (result) =>
217 Result.isSuccess(result) &&
218 result.success.status >= 200 &&
219 result.success.status < 300,
220 );
221 const anySuccess =
222 healthy ?? results.find((result) => Result.isSuccess(result));
223 if (anySuccess !== undefined && Result.isSuccess(anySuccess)) {
224 for (const result of results) {
225 if (result !== anySuccess && Result.isFailure(result)) {
226 yield* Effect.logDebug(
227 "telemetry destination failed",
228 result.failure,
229 );
230 }
231 }
232 return anySuccess.success;
233 }
234 const failure = results.find((result) => Result.isFailure(result));
235 return yield* failure !== undefined && Result.isFailure(failure)
236 ? Effect.fail(failure.failure)
237 : effect;
238 });
239 });
240 }),
241 );
242
243const makeExporterLayer = (options?: {
244 exportInterval?: Duration.Input;
245 shutdownTimeout?: Duration.Input;
246}): TelemetryLayer =>
247 Layer.unwrap(
248 Effect.gen(function* () {
249 // Depending on the secret/plain packing path, the bound list arrives
250 // as a JSON string or already parsed into an array.
251 const rawList = yield* readBoundValue(EXPORTERS_KEY);
252 const bound: ResolvedDestination[] = Array.isArray(rawList)
253 ? (rawList as ResolvedDestination[])
254 : typeof rawList === "string" && rawList !== ""
255 ? yield* Effect.try(
256 () => JSON.parse(rawList) as ResolvedDestination[],
257 )
258 : [];
259 // The standard OTEL_* env vars form an implicit extra destination.
260 const [stdTraces, stdLogs, stdMetrics] = yield* Effect.all([
261 signalConfig("TRACES"),
262 signalConfig("LOGS"),
263 signalConfig("METRICS"),
264 ]);
265 const destinations: ResolvedDestination[] = [
266 ...bound,
267 ...(stdTraces || stdLogs || stdMetrics
268 ? [{ traces: stdTraces, logs: stdLogs, metrics: stdMetrics }]
269 : []),
270 ];
271 const targets = (signal: keyof ResolvedDestination) =>
272 destinations.flatMap((destination) => {
273 const resolved = destination[signal];
274 return resolved !== undefined ? [resolved] : [];
275 });
276 const traces = targets("traces");
277 const logs = targets("logs");
278 const metrics = targets("metrics");
279 if (traces.length === 0 && logs.length === 0 && metrics.length === 0) {
280 return Layer.empty;
281 }
282 const resource = yield* defaultResource;
283 const routes = new Map<string, ResolvedSignal[]>();
284 const layers: Layer.Layer<never, never, any>[] = [];
285 if (traces.length > 0) {
286 routes.set(SENTINEL.traces, traces);
287 layers.push(
288 OtlpTracer.layer({
289 url: SENTINEL.traces,
290 resource,
291 exportInterval: options?.exportInterval,
292 shutdownTimeout: options?.shutdownTimeout,
293 }),
294 );
295 }
296 if (logs.length > 0) {
297 routes.set(SENTINEL.logs, logs);
298 layers.push(
299 OtlpLogger.layer({
300 url: SENTINEL.logs,
301 resource,
302 exportInterval: options?.exportInterval,
303 shutdownTimeout: options?.shutdownTimeout,
304 }),
305 );
306 }
307 if (metrics.length > 0) {
308 routes.set(SENTINEL.metrics, metrics);
309 layers.push(
310 OtlpMetrics.layer({
311 url: SENTINEL.metrics,
312 resource,
313 exportInterval: options?.exportInterval,
314 shutdownTimeout: options?.shutdownTimeout,
315 }),
316 );
317 }
318 return Layer.mergeAll(...(layers as [Layer.Layer<never>])).pipe(
319 Layer.provide(OtlpSerialization.layerJson),
320 Layer.provide(fanoutClient(routes)),
321 );
322 }).pipe(
323 Effect.catchCause((cause) =>
324 Effect.logWarning(
325 "Invalid telemetry configuration; telemetry disabled",
326 cause,
327 ).pipe(Effect.as(Layer.empty)),
328 ),
329 ),
330 );
331
332/**
333 * The runtime half of the {@link layerOtlp} binding, and the default
334 * per-event Layer: reads the bound `OTEL_EXPORTER_OTLP_*` values back and
335 * constructs the OTLP JSON exporters. Each signal resolves independently;
336 * only configured signals export; resolves to `Layer.empty` when nothing is
337 * bound, so telemetry is free until a layer is provided.
338 *
339 * The periodic export intervals are effectively disabled: the exporter is
340 * built per event and the request-scope flush delivers everything. An
341 * interval export firing mid-event would race the scope close — the close
342 * interrupts the exporter's in-flight batch (already spliced out of the
343 * buffer), silently dropping it. Lambda invocations regularly outlive the
344 * 1-second logger interval, which is exactly how this was discovered.
345 * Final exports get the OTLP ten-second batch budget rather than the
346 * exporter's three-second shutdown default, which can cancel slow delivery.
347 *
348 * A malformed configuration degrades to `Layer.empty` with a warning
349 * instead of failing the event.
350 */
351export const fromBoundConfig: TelemetryLayer = makeExporterLayer({
352 exportInterval: "1 hour",
353 shutdownTimeout: "10 seconds",
354});
355
356/**
357 * The process-runtime half of the binding: same bound-value resolution with
358 * the standard periodic export intervals, since a server's root scope only
359 * flushes at shutdown.
360 */
361const fromBoundConfigProcess: TelemetryLayer = makeExporterLayer();
362
363/**
364 * The per-event telemetry exporters, as a `Context.Reference` holding the
365 * Layer the runtime bridges build into every event's request scope.
366 * Provide it via {@link layerOtlp} / {@link layer} rather than directly —
367 * see the module documentation.
368 */
369export const Telemetry = Context.Reference<TelemetryLayer>(
370 "alchemy/Telemetry",
371 {
372 defaultValue: () => fromBoundConfig,
373 },
374);
375
376const reference = Telemetry;
377
378/**
379 * Build the configured {@link Telemetry} Layer into an event's request
380 * scope, returning the Context of telemetry services to provide to the
381 * event's handler effect.
382 *
383 * Called by the runtime bridges (Worker, Durable Object, Workflow, Lambda)
384 * once per event. Building into the *request* scope — not the
385 * never-finalized isolate scope — is what makes export work on workerd:
386 * the batching fiber lives inside the event's I/O context and the final
387 * flush runs from the scope's finalizer, which the bridges register with
388 * `ctx.waitUntil`.
389 *
390 * A failed build (bad user Layer, config error) degrades to an empty
391 * Context with a warning instead of failing the event.
392 *
393 * `override` is the (possibly merged) custom Layer registered on the
394 * runtime context by {@link layer} during init. It composes with
395 * — rather than replaces — the bound OTLP destinations: loggers and metric
396 * exporters merge, and a custom `Tracer` (a single Effect service) wins
397 * over the built-in one.
398 *
399 * Declared `R = never`: the Layer's actual requirements (`HttpClient`,
400 * `ConfigProvider`, …) are satisfied at runtime by the bridge's surrounding
401 * `Effect.provide` of the built runtime context.
402 */
403export const buildEventTelemetry = (
404 context: Context.Context<never>,
405 scope: Scope.Scope,
406 override?: TelemetryLayer | undefined,
407 base?: TelemetryLayer | undefined,
408): Effect.Effect<Context.Context<never>> =>
409 Effect.suspend(() => {
410 const bound = base ?? Context.get(context, reference);
411 return Layer.buildWithScope(
412 override === undefined ? bound : Layer.mergeAll(bound, override),
413 scope,
414 );
415 }).pipe(
416 Effect.catchCause((cause) =>
417 Effect.logWarning("Failed to build telemetry layer", cause).pipe(
418 Effect.as(Context.empty()),
419 ),
420 ),
421 ) as Effect.Effect<Context.Context<never>>;
422
423/**
424 * Provide telemetry to a long-running server process (Cloudflare Container,
425 * ECS Task, EC2 host, Lambda microVM).
426 *
427 * Unlike the per-event runtime bridges, server processes have no I/O-context
428 * pinning and a real shutdown: the configured {@link Telemetry} Layer is
429 * built ONCE into the ambient root scope, exporters batch on their intervals
430 * for the life of the process, and the final flush runs when the root scope
431 * closes on graceful exit.
432 *
433 * `runtimeContext` is the entrypoint's runtime context; its `telemetry`
434 * field carries the custom Layer(s) registered by {@link layer}
435 * during init, composed with the bound OTLP destinations (read with the
436 * standard periodic export intervals).
437 */
438export const provideProcessTelemetry =
439 (runtimeContext?: { telemetry?: TelemetryLayer | undefined }) =>
440 <A, E, R>(
441 effect: Effect.Effect<A, E, R>,
442 ): Effect.Effect<A, E, R | Scope.Scope> =>
443 Effect.gen(function* () {
444 const context = yield* Effect.context<never>();
445 const scope = yield* Effect.scope;
446 const telemetry = yield* buildEventTelemetry(
447 context,
448 scope,
449 runtimeContext?.telemetry,
450 fromBoundConfigProcess,
451 );
452 return yield* Effect.provideContext(effect, telemetry);
453 });