1/**
2 * Shared batch exporter for OTLP/HTTP observability modules.
3 *
4 * Signal modules use this exporter to buffer already-encoded telemetry and post
5 * it to a configured OTLP endpoint. It owns the scoped transport loop, batching,
6 * retry behavior, temporary disabling after repeated failures, and final flush
7 * during shutdown.
8 *
9 * @stability unstable
10 * @since 4.0.0
11 */
12import { Clock } from "../Clock.ts"
13import * as Context from "../Context.ts"
14import * as Duration from "../Duration.ts"
15import * as Effect from "../Effect.ts"
16import * as Fiber from "../Fiber.ts"
17import * as FiberSet from "../FiberSet.ts"
18import { identity } from "../Function.ts"
19import * as Headers from "../http/Headers.ts"
20import type { HttpBody } from "../http/HttpBody.ts"
21import * as HttpClient from "../http/HttpClient.ts"
22import * as HttpClientError from "../http/HttpClientError.ts"
23import * as HttpClientRequest from "../http/HttpClientRequest.ts"
24import * as Layer from "../Layer.ts"
25import * as Num from "../Number.ts"
26import * as Option from "../Option.ts"
27import * as Schedule from "../Schedule.ts"
28import * as Scope from "../Scope.ts"
29
30const retryAfterDelay = (value: string | undefined): Effect.Effect<Duration.Duration> => {
31 const seconds = Option.fromUndefinedOr(value).pipe(Option.flatMap(Num.parse))
32 if (Option.isSome(seconds)) {
33 return Effect.succeed(Duration.seconds(seconds.value))
34 }
35 if (value === undefined) {
36 return Effect.succeed(Duration.seconds(5))
37 }
38 const timestamp = Date.parse(value)
39 if (Number.isNaN(timestamp)) {
40 return Effect.succeed(Duration.seconds(5))
41 }
42 return Effect.map(
43 Clock,
44 (clock) => Duration.millis(Math.max(timestamp - clock.currentTimeMillisUnsafe(), 1))
45 )
46}
47
48const policy = Schedule.forever.pipe(
49 Schedule.passthrough,
50 Schedule.addDelay(({ output: error }) => {
51 if (
52 HttpClientError.isHttpClientError(error)
53 && error.reason._tag === "StatusCodeError"
54 && error.reason.response.status === 429
55 ) {
56 return retryAfterDelay(error.reason.response.headers["retry-after"])
57 }
58 return Effect.succeed(Duration.seconds(1))
59 })
60)
61
62/**
63 * Registry of exporter flush operations, used to manually drain buffered
64 * telemetry before the surrounding scope closes.
65 *
66 * **Details**
67 *
68 * Every exporter created by `make` registers its export operation here, so a
69 * single `flush` drains all signals sharing the registry. `flush` returns only
70 * after the exports it initiated have settled, cannot fail, and respects each
71 * exporter's temporary-disable window. Wrap it with `Effect.timeoutOption` to
72 * bound its duration at the call site.
73 *
74 * @stability unstable
75 * @category services
76 * @since 4.0.0
77 */
78export class Flusher extends Context.Service<Flusher, {
79 /**
80 * Drains all registered exporters concurrently and cannot fail.
81 *
82 * **Details**
83 *
84 * There is no built-in timeout; use `Effect.timeoutOption` to bound the
85 * operation. Exporters in their 60-second `disabledUntil` window are skipped.
86 *
87 * **Example** (Flushing exporters)
88 *
89 * ```ts import.meta.vitest
90 * import { Effect } from "effect"
91 * import { OtlpExporter } from "effect/observability"
92 *
93 * const program = Effect.gen(function*() {
94 * const flusher = yield* OtlpExporter.Flusher
95 * yield* flusher.flush
96 * return "flushed"
97 * }).pipe(Effect.provide(OtlpExporter.layerFlusher))
98 *
99 * await Effect.runPromise(program) // => "flushed"
100 * ```
101 */
102 readonly flush: Effect.Effect<void>
103 readonly register: (run: Effect.Effect<void>) => Effect.Effect<void, never, Scope.Scope>
104}>()(
105 "effect/observability/OtlpExporter/Flusher"
106) {}
107
108/**
109 * Provides a `Flusher` backed by a fresh registry.
110 *
111 * **Details**
112 *
113 * This is intentionally a single module-level constant rather than a factory:
114 * layer memoization is keyed by layer instance, so every signal layer
115 * referencing this same constant shares one registry per layer build, and one
116 * `flush` drains traces, logs and metrics together. A factory returning a new
117 * layer per call would silently create one registry per signal.
118 *
119 * Registration is scoped — an exporter is removed from the registry when its
120 * own scope closes. Flushing with an empty registry is a no-op.
121 *
122 * Note that `flush` cannot await an export that was already in flight when it
123 * was called (for example one started by the export interval); it only waits
124 * for the exports it initiates.
125 *
126 * @stability unstable
127 * @category layers
128 * @since 4.0.0
129 */
130export const layerFlusher: Layer.Layer<Flusher> = Layer.sync(Flusher, () => {
131 const registry = new Set<Effect.Effect<void>>()
132 return {
133 flush: Effect.suspend(() => {
134 if (registry.size === 0) {
135 return Effect.void
136 }
137 return Effect.forEach(registry, identity, {
138 concurrency: "unbounded",
139 discard: true
140 })
141 }),
142 register: (run) =>
143 Effect.flatMap(Scope.Scope, (scope) => {
144 registry.add(run)
145 return Scope.addFinalizer(
146 scope,
147 Effect.sync(() => registry.delete(run))
148 )
149 })
150 }
151})
152
153/**
154 * Creates a scoped OTLP batch exporter.
155 *
156 * **Details**
157 *
158 * The exporter buffers pushed data, periodically posts encoded batches to the
159 * configured URL, retries transient failures, temporarily disables exporting
160 * after unhandled failures, and flushes during scope finalization up to
161 * `shutdownTimeout`.
162 *
163 * @stability unstable
164 * @category constructors
165 * @since 4.0.0
166 */
167export const make: (
168 options: {
169 readonly url: string
170 readonly headers: Headers.Input | undefined
171 readonly label: string
172 readonly exportInterval: Duration.Input
173 readonly maxBatchSize: number | "disabled"
174 readonly exportEmpty?: boolean | undefined
175 readonly body: (data: Array<any>) => readonly [body: HttpBody, onSuccess: Effect.Effect<void>]
176 readonly shutdownTimeout: Duration.Input
177 }
178) => Effect.Effect<
179 { readonly push: (data: unknown) => void },
180 never,
181 Flusher | HttpClient.HttpClient | Scope.Scope
182> = Effect.fnUntraced(function*(options) {
183 const services = yield* Effect.context<Scope.Scope | HttpClient.HttpClient>()
184 const clock = Context.get(services, Clock)
185 const scope = Context.get(services, Scope.Scope)
186 const exportInterval = Duration.max(Duration.fromInputUnsafe(options.exportInterval), Duration.zero)
187 let disabledUntil: number | undefined = undefined
188
189 const client = Context.get(services, HttpClient.HttpClient).pipe(
190 HttpClient.tap((response) => Effect.ignore(response.arrayBuffer)),
191 HttpClient.tapError((error) => error.response ? Effect.ignore(error.response.arrayBuffer) : Effect.void),
192 HttpClient.filterStatusOk,
193 HttpClient.transformResponse(Effect.provideService(HttpClient.TracerPropagationEnabled, false)),
194 HttpClient.retryTransient({ schedule: policy, times: 3 })
195 )
196
197 let headers = Headers.fromRecordUnsafe({
198 "user-agent": `effect-opentelemetry-${options.label}/0.0.0`
199 })
200 if (options.headers) {
201 headers = Headers.merge(Headers.fromInput(options.headers), headers)
202 }
203
204 const request = HttpClientRequest.post(options.url, { headers })
205 let buffer: Array<any> = []
206 const runExport = Effect.suspend(() => {
207 if (disabledUntil !== undefined && clock.currentTimeMillisUnsafe() < disabledUntil) {
208 return Effect.void
209 } else if (disabledUntil !== undefined) {
210 disabledUntil = undefined
211 }
212 if (buffer.length === 0 && (options.maxBatchSize !== "disabled" || options.exportEmpty !== true)) {
213 return Effect.void
214 }
215 const items = buffer
216 buffer = []
217 const [body, onSuccess] = options.body(items)
218 return client.execute(
219 HttpClientRequest.setBody(request, body)
220 ).pipe(
221 Effect.andThen(onSuccess),
222 Effect.asVoid,
223 Effect.withTracerEnabled(false)
224 )
225 }).pipe(
226 Effect.catchCause((cause) => {
227 if (disabledUntil !== undefined) return Effect.void
228 disabledUntil = clock.currentTimeMillisUnsafe() + 60_000
229 buffer = []
230 return Effect.logDebug("Disabling exporter for 60 seconds", cause)
231 }),
232 Effect.annotateLogs({
233 package: "@effect/opentelemetry",
234 module: options.label
235 })
236 )
237
238 const exportFibers = yield* FiberSet.make<void, never>()
239 const runExportFork = yield* FiberSet.runtime(exportFibers)<never>()
240
241 const flusher = yield* Flusher
242 yield* flusher.register(runExport)
243
244 yield* Scope.addFinalizer(
245 scope,
246 Effect.suspend(() => {
247 if (disabledUntil !== undefined) return Effect.void
248 runExportFork(runExport)
249 return FiberSet.awaitEmpty(exportFibers)
250 }).pipe(
251 Effect.ignore,
252 Effect.interruptible,
253 Effect.timeoutOption(options.shutdownTimeout)
254 )
255 )
256
257 yield* Effect.sleep(exportInterval).pipe(
258 Effect.andThen(FiberSet.run(exportFibers, runExport)),
259 Effect.flatMap(Fiber.await),
260 Effect.forever,
261 Effect.forkIn(scope)
262 )
263
264 return {
265 push(data) {
266 if (disabledUntil !== undefined) return
267 buffer.push(data)
268 if (options.maxBatchSize !== "disabled" && buffer.length >= options.maxBatchSize) {
269 runExportFork(runExport)
270 }
271 }
272 }
273})