1import { assert, describe, it } from "@effect/vitest"
2import { Clock, ConfigProvider, Deferred, type Duration, Effect, Exit, Fiber, Layer, Metric, Ref, Scope } from "effect"
3import { HttpBody, HttpClient, HttpClientResponse } from "effect/http"
4import type { HttpClientError } from "effect/http"
5import { OtlpExporter, OtlpLogger, OtlpMetrics, OtlpSerialization, OtlpTracer } from "effect/observability"
6import { TestClock } from "effect/testing"
7
8const makeHttpClient = Effect.fnUntraced(function*(retryAfter: string | undefined) {
9 const attempts = yield* Ref.make(0)
10
11 const httpClient = HttpClient.makeWith(
12 Effect.fnUntraced(function*(requestEffect) {
13 const request = yield* requestEffect
14 const attempt = yield* Ref.updateAndGet(attempts, (attempts) => attempts + 1)
15 if (attempt === 1) {
16 return HttpClientResponse.fromWeb(
17 request,
18 retryAfter === undefined
19 ? new Response(null, { status: 429 })
20 : new Response(null, { status: 429, headers: { "retry-after": retryAfter } })
21 )
22 }
23 return HttpClientResponse.fromWeb(request, new Response())
24 }),
25 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
26 )
27
28 return { attempts, httpClient } as const
29})
30
31const makeExporter = (
32 httpClient: HttpClient.HttpClient,
33 options?: {
34 readonly exportInterval?: Duration.Input
35 readonly maxBatchSize?: number | "disabled"
36 readonly shutdownTimeout?: Duration.Input
37 readonly body?: (data: Array<any>) => HttpBody.HttpBody
38 }
39) =>
40 OtlpExporter.make({
41 label: "OtlpExporterTest",
42 url: "http://localhost:4318/v1/logs",
43 headers: undefined,
44 exportInterval: options?.exportInterval ?? "1 hour",
45 maxBatchSize: options?.maxBatchSize ?? 1,
46 body: (data) => [options?.body?.(data) ?? HttpBody.empty, Effect.void],
47 shutdownTimeout: options?.shutdownTimeout ?? "1 second"
48 }).pipe(
49 Effect.provideService(HttpClient.HttpClient, httpClient),
50 Effect.provide(OtlpExporter.layerFlusher)
51 )
52
53const makeExporterRaw = (
54 maxBatchSize: number | "disabled" = 1,
55 body: (data: Array<any>) => HttpBody.HttpBody = () => HttpBody.empty
56) =>
57 OtlpExporter.make({
58 label: "OtlpExporterTest",
59 url: "http://localhost:4318/v1/logs",
60 headers: undefined,
61 exportInterval: "1 hour",
62 maxBatchSize,
63 body: (data) => [body(data), Effect.void],
64 shutdownTimeout: "1 second"
65 })
66
67const makeStatusHttpClient = Effect.fnUntraced(function*(status: number) {
68 const attempts = yield* Ref.make(0)
69 const urls = yield* Ref.make<ReadonlyArray<string>>([])
70
71 const httpClient = HttpClient.makeWith(
72 Effect.fnUntraced(function*(requestEffect) {
73 const request = yield* requestEffect
74 yield* Ref.update(attempts, (attempts) => attempts + 1)
75 yield* Ref.update(urls, (urls) => [...urls, request.url])
76 return HttpClientResponse.fromWeb(request, new Response(null, { status }))
77 }),
78 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
79 )
80
81 return { attempts, httpClient, urls } as const
82})
83
84const yieldNowN = (times: number) =>
85 Effect.forEach(Array.from({ length: times }), () => Effect.yieldNow, { discard: true })
86
87const makeControlledHttpClient = Effect.fnUntraced(function*(requestCount: number) {
88 const started = yield* Effect.forEach(Array.from({ length: requestCount }), () => Deferred.make<void>())
89 const releases = yield* Effect.forEach(Array.from({ length: requestCount }), () => Deferred.make<void>())
90 const interrupted = yield* Ref.make(0)
91 let requestIndex = 0
92
93 const httpClient = HttpClient.makeWith(
94 Effect.fnUntraced(function*(requestEffect) {
95 const request = yield* requestEffect
96 const index = requestIndex++
97 yield* Deferred.succeed(started[index], undefined)
98 yield* Deferred.await(releases[index]).pipe(
99 Effect.onInterrupt(() => Ref.update(interrupted, (count) => count + 1))
100 )
101 return HttpClientResponse.fromWeb(request, new Response())
102 }),
103 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
104 )
105
106 return { httpClient, interrupted, releases, started } as const
107})
108
109describe("OtlpExporter", () => {
110 it.effect("allows an in-flight timer export to finish during shutdown", () =>
111 Effect.gen(function*() {
112 const scope = yield* Scope.make()
113 const { httpClient, releases, started } = yield* makeControlledHttpClient(1)
114 const exporter = yield* makeExporter(httpClient, {
115 exportInterval: "1 second",
116 maxBatchSize: 10
117 }).pipe(Scope.provide(scope))
118
119 exporter.push({ value: 1 })
120 yield* TestClock.adjust("1 second")
121 yield* Deferred.await(started[0])
122
123 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void))
124 yield* Effect.yieldNow
125 assert.isUndefined(closeFiber.pollUnsafe())
126
127 yield* Deferred.succeed(releases[0], undefined)
128 yield* Fiber.join(closeFiber)
129 }))
130
131 it.effect("waits for each periodic export before starting the next interval", () =>
132 Effect.gen(function*() {
133 const scope = yield* Scope.make()
134 const { httpClient, releases, started } = yield* makeControlledHttpClient(3)
135 const exporter = yield* makeExporter(httpClient, {
136 exportInterval: "1 second",
137 maxBatchSize: "disabled"
138 }).pipe(Scope.provide(scope))
139
140 exporter.push({ value: 1 })
141 yield* TestClock.adjust("1 second")
142 yield* Deferred.await(started[0])
143
144 yield* TestClock.adjust("1 second")
145 assert.isFalse(yield* Deferred.isDone(started[1]))
146
147 yield* Deferred.succeed(releases[0], undefined)
148 exporter.push({ value: 2 })
149 yield* TestClock.adjust("1 second")
150 yield* Deferred.await(started[1])
151
152 exporter.push({ value: 3 })
153 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void))
154 yield* Deferred.succeed(releases[1], undefined)
155 yield* Deferred.await(started[2])
156 yield* Deferred.succeed(releases[2], undefined)
157 yield* Fiber.join(closeFiber)
158 }))
159
160 it.effect("allows an in-flight batch export to finish during shutdown", () =>
161 Effect.gen(function*() {
162 const scope = yield* Scope.make()
163 const { httpClient, releases, started } = yield* makeControlledHttpClient(1)
164 const exporter = yield* makeExporter(httpClient).pipe(Scope.provide(scope))
165
166 exporter.push({ value: 1 })
167 yield* Deferred.await(started[0])
168
169 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void))
170 yield* Effect.yieldNow
171 assert.isUndefined(closeFiber.pollUnsafe())
172
173 yield* Deferred.succeed(releases[0], undefined)
174 yield* Fiber.join(closeFiber)
175 }))
176
177 it.effect("exports remaining telemetry concurrently and waits for all requests", () =>
178 Effect.gen(function*() {
179 const scope = yield* Scope.make()
180 const { httpClient, releases, started } = yield* makeControlledHttpClient(2)
181 const batches: Array<Array<number>> = []
182 const exporter = yield* makeExporter(httpClient, {
183 maxBatchSize: 2,
184 body(data) {
185 batches.push(data.map((item) => item.value))
186 return HttpBody.empty
187 }
188 }).pipe(Scope.provide(scope))
189
190 exporter.push({ value: 1 })
191 exporter.push({ value: 2 })
192 yield* Deferred.await(started[0])
193 exporter.push({ value: 3 })
194
195 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void))
196 yield* Deferred.await(started[1])
197 assert.deepStrictEqual(batches, [[1, 2], [3]])
198
199 yield* Deferred.succeed(releases[0], undefined)
200 yield* Effect.yieldNow
201 assert.isUndefined(closeFiber.pollUnsafe())
202 yield* Deferred.succeed(releases[1], undefined)
203 yield* Fiber.join(closeFiber)
204 }))
205
206 it.effect("bounds shutdown waiting and interrupts remaining requests", () =>
207 Effect.gen(function*() {
208 const scope = yield* Scope.make()
209 const { httpClient, interrupted, started } = yield* makeControlledHttpClient(1)
210 const exporter = yield* makeExporter(httpClient).pipe(Scope.provide(scope))
211
212 exporter.push({ value: 1 })
213 yield* Deferred.await(started[0])
214
215 const closeFiber = yield* Effect.forkChild(Scope.close(scope, Exit.void))
216 yield* TestClock.adjust("1 second")
217 yield* Fiber.join(closeFiber)
218 assert.strictEqual(yield* Ref.get(interrupted), 1)
219 }))
220
221 it.effect("does not initiate or wait for delivery when disabled", () =>
222 Effect.gen(function*() {
223 const scope = yield* Scope.make()
224 const blocked = yield* Deferred.make<void>()
225 const blockedInterrupted = yield* Deferred.make<void>()
226 const failed = yield* Deferred.make<void>()
227 const attempts = yield* Ref.make(0)
228 const httpClient = HttpClient.makeWith(
229 Effect.fnUntraced(function*(requestEffect) {
230 const request = yield* requestEffect
231 const attempt = yield* Ref.updateAndGet(attempts, (count) => count + 1)
232 if (attempt === 1) {
233 yield* Deferred.succeed(blocked, undefined)
234 return yield* Effect.never.pipe(
235 Effect.onInterrupt(() => Deferred.succeed(blockedInterrupted, undefined))
236 )
237 }
238 yield* Deferred.succeed(failed, undefined)
239 return HttpClientResponse.fromWeb(request, new Response(null, { status: 400 }))
240 }),
241 Effect.succeed as HttpClient.HttpClient.Preprocess<HttpClientError.HttpClientError, never>
242 )
243 const exporter = yield* makeExporter(httpClient).pipe(Scope.provide(scope))
244
245 exporter.push({ value: 1 })
246 yield* Deferred.await(blocked)
247 exporter.push({ value: 2 })
248 yield* Deferred.await(failed)
249 yield* yieldNowN(3)
250 exporter.push({ value: 3 })
251
252 yield* Scope.close(scope, Exit.void)
253 yield* Deferred.await(blockedInterrupted)
254 assert.strictEqual(yield* Ref.get(attempts), 2)
255 }))
256
257 it.effect("retries status 429 with numeric retry-after delay", () =>
258 Effect.gen(function*() {
259 const { attempts, httpClient } = yield* makeHttpClient("2")
260 const exporter = yield* makeExporter(httpClient)
261
262 exporter.push({ value: 1 })
263 yield* yieldNowN(3)
264
265 assert.strictEqual(yield* Ref.get(attempts), 1)
266
267 yield* TestClock.adjust("1 second")
268 yield* yieldNowN(2)
269 assert.strictEqual(yield* Ref.get(attempts), 1)
270
271 yield* TestClock.adjust("1 second")
272 yield* yieldNowN(2)
273 assert.strictEqual(yield* Ref.get(attempts), 2)
274 }))
275
276 it.effect("retries status 429 with HTTP-date retry-after delay", () =>
277 Effect.gen(function*() {
278 const { attempts, httpClient } = yield* makeHttpClient("Thu, 01 Jan 1970 00:01:00 GMT")
279 const exporter = yield* makeExporter(httpClient)
280
281 exporter.push({ value: 1 })
282 yield* yieldNowN(3)
283
284 assert.strictEqual(yield* Ref.get(attempts), 1)
285
286 yield* TestClock.adjust("5 seconds")
287 yield* yieldNowN(2)
288 assert.strictEqual(yield* Ref.get(attempts), 1)
289
290 yield* TestClock.adjust("55 seconds")
291 yield* yieldNowN(2)
292 assert.strictEqual(yield* Ref.get(attempts), 2)
293 }))
294
295 it.effect("uses fallback retry-after delay when header is non-numeric", () =>
296 Effect.gen(function*() {
297 const { attempts, httpClient } = yield* makeHttpClient("soon")
298 const exporter = yield* makeExporter(httpClient)
299
300 exporter.push({ value: 1 })
301 yield* yieldNowN(3)
302
303 assert.strictEqual(yield* Ref.get(attempts), 1)
304
305 yield* TestClock.adjust("4 seconds")
306 yield* yieldNowN(2)
307 assert.strictEqual(yield* Ref.get(attempts), 1)
308
309 yield* TestClock.adjust("1 second")
310 yield* yieldNowN(2)
311 assert.strictEqual(yield* Ref.get(attempts), 2)
312 }))
313
314 it.effect("uses fallback retry-after delay when header is missing", () =>
315 Effect.gen(function*() {
316 const { attempts, httpClient } = yield* makeHttpClient(undefined)
317 const exporter = yield* makeExporter(httpClient)
318
319 exporter.push({ value: 1 })
320 yield* yieldNowN(3)
321
322 assert.strictEqual(yield* Ref.get(attempts), 1)
323
324 yield* TestClock.adjust("4 seconds")
325 yield* yieldNowN(2)
326 assert.strictEqual(yield* Ref.get(attempts), 1)
327
328 yield* TestClock.adjust("1 second")
329 yield* yieldNowN(2)
330 assert.strictEqual(yield* Ref.get(attempts), 2)
331 }))
332
333 describe("flush", () => {
334 it.effect("exports buffered items without advancing to the export interval", () =>
335 Effect.gen(function*() {
336 const { attempts, httpClient } = yield* makeStatusHttpClient(200)
337 yield* Effect.scoped(
338 Effect.gen(function*() {
339 const exporter = yield* makeExporterRaw(10)
340 const flusher = yield* OtlpExporter.Flusher
341 const before = yield* Clock.currentTimeMillis
342
343 exporter.push({ value: 1 })
344 yield* flusher.flush
345
346 assert.strictEqual(yield* Clock.currentTimeMillis, before)
347 assert.strictEqual(yield* Ref.get(attempts), 1)
348 }).pipe(
349 Effect.provideService(HttpClient.HttpClient, httpClient),
350 Effect.provide(OtlpExporter.layerFlusher)
351 )
352 )
353 }))
354
355 for (const filterStatus of [false, true]) {
356 it.effect(`drains a 429 response before retrying${filterStatus ? " from a status-filtered client" : ""}`, () =>
357 Effect.gen(function*() {
358 const responses: Array<Response> = []
359 const rawClient = HttpClient.make((request) => {
360 const response = new Response("body", {
361 status: responses.length === 0 ? 429 : 200,
362 headers: { "retry-after": "1" }
363 })
364 responses.push(response)
365 return Effect.succeed(HttpClientResponse.fromWeb(request, response))
366 })
367 const httpClient = filterStatus ? HttpClient.filterStatusOk(rawClient) : rawClient
368 yield* Effect.scoped(
369 Effect.gen(function*() {
370 const exporter = yield* makeExporterRaw(10)
371 const flusher = yield* OtlpExporter.Flusher
372
373 exporter.push({ value: 1 })
374 const flush = yield* Effect.forkChild(flusher.flush)
375 yield* yieldNowN(3)
376 assert.strictEqual(responses.length, 1)
377 assert.isTrue(responses[0].bodyUsed)
378
379 yield* TestClock.adjust("1 second")
380 yield* Fiber.join(flush)
381 assert.strictEqual(responses.length, 2)
382 assert.isTrue(responses[1].bodyUsed)
383 }).pipe(
384 Effect.provideService(HttpClient.HttpClient, httpClient),
385 Effect.provide(OtlpExporter.layerFlusher)
386 )
387 )
388 }))
389 }
390
391 it.effect("keeps exporting when reading a successful response body fails", () =>
392 Effect.gen(function*() {
393 let attempts = 0
394 const httpClient = HttpClient.make((request) => {
395 attempts++
396 const response = new Response(
397 new ReadableStream({
398 start(controller) {
399 controller.error(new Error("body read failed"))
400 }
401 })
402 )
403 return Effect.succeed(HttpClientResponse.fromWeb(request, response))
404 })
405 yield* Effect.scoped(
406 Effect.gen(function*() {
407 const exporter = yield* makeExporterRaw(10)
408 const flusher = yield* OtlpExporter.Flusher
409
410 exporter.push({ value: 1 })
411 yield* flusher.flush
412 exporter.push({ value: 2 })
413 yield* flusher.flush
414 assert.strictEqual(attempts, 2)
415 }).pipe(
416 Effect.provideService(HttpClient.HttpClient, httpClient),
417 Effect.provide(OtlpExporter.layerFlusher)
418 )
419 )
420 }))
421
422 it.effect("does not export empty payloads when batching is disabled", () =>
423 Effect.gen(function*() {
424 const { httpClient } = yield* makeStatusHttpClient(200)
425 const payloads: Array<ReadonlyArray<unknown>> = []
426
427 yield* Effect.scoped(
428 Effect.gen(function*() {
429 yield* makeExporterRaw("disabled", (items) => {
430 payloads.push(items)
431 return HttpBody.empty
432 })
433 const flusher = yield* OtlpExporter.Flusher
434 yield* flusher.flush
435 }).pipe(
436 Effect.provideService(HttpClient.HttpClient, httpClient),
437 Effect.provide(OtlpExporter.layerFlusher)
438 )
439 )
440
441 assert.deepStrictEqual(payloads, [])
442 }))
443
444 it.effect("exports buffered items once when batching is disabled", () =>
445 Effect.gen(function*() {
446 const { httpClient } = yield* makeStatusHttpClient(200)
447 const payloads: Array<ReadonlyArray<unknown>> = []
448
449 yield* Effect.scoped(
450 Effect.gen(function*() {
451 const exporter = yield* makeExporterRaw("disabled", (items) => {
452 payloads.push(items)
453 return HttpBody.empty
454 })
455 const flusher = yield* OtlpExporter.Flusher
456
457 exporter.push({ value: 1 })
458 yield* flusher.flush
459 yield* flusher.flush
460 }).pipe(
461 Effect.provideService(HttpClient.HttpClient, httpClient),
462 Effect.provide(OtlpExporter.layerFlusher)
463 )
464 )
465
466 assert.deepStrictEqual(payloads, [[{ value: 1 }]])
467 }))
468
469 it.effect("shares one registry across traces, logs and metrics", () =>
470 Effect.gen(function*() {
471 const { httpClient, urls } = yield* makeStatusHttpClient(200)
472 const signals = Layer.mergeAll(
473 OtlpTracer.layer({
474 url: "http://localhost:4318/v1/traces",
475 resource: { serviceName: "test" },
476 exportInterval: "1 hour",
477 maxBatchSize: 100
478 }),
479 OtlpLogger.layer({
480 url: "http://localhost:4318/v1/logs",
481 resource: { serviceName: "test" },
482 exportInterval: "1 hour",
483 maxBatchSize: 100,
484 mergeWithExisting: false
485 }),
486 OtlpMetrics.layer({
487 url: "http://localhost:4318/v1/metrics",
488 resource: { serviceName: "test" },
489 exportInterval: "1 hour"
490 })
491 ).pipe(
492 Layer.provide(OtlpSerialization.layerJson),
493 Layer.provideMerge(Layer.succeed(HttpClient.HttpClient, httpClient))
494 )
495
496 yield* Effect.gen(function*() {
497 yield* Effect.log("flush test")
498 yield* Metric.update(Metric.counter("otlp_flush_test"), 1)
499 yield* Effect.void.pipe(Effect.withSpan("flush test"))
500
501 const flusher = yield* OtlpExporter.Flusher
502 yield* flusher.flush
503
504 assert.deepStrictEqual(
505 [...yield* Ref.get(urls)].sort(),
506 [
507 "http://localhost:4318/v1/logs",
508 "http://localhost:4318/v1/metrics",
509 "http://localhost:4318/v1/traces"
510 ]
511 )
512 }).pipe(Effect.provide(signals))
513 }))
514
515 it.effect("is a no-op while the exporter is disabled", () =>
516 Effect.gen(function*() {
517 const { attempts, httpClient } = yield* makeStatusHttpClient(400)
518 yield* Effect.scoped(
519 Effect.gen(function*() {
520 const exporter = yield* makeExporterRaw(10)
521 const flusher = yield* OtlpExporter.Flusher
522
523 exporter.push({ value: 1 })
524 yield* flusher.flush
525 assert.strictEqual(yield* Ref.get(attempts), 1)
526
527 exporter.push({ value: 2 })
528 yield* flusher.flush
529 assert.strictEqual(yield* Ref.get(attempts), 1)
530 }).pipe(
531 Effect.provideService(HttpClient.HttpClient, httpClient),
532 Effect.provide(OtlpExporter.layerFlusher)
533 )
534 )
535 }))
536
537 it.effect("deregisters an exporter when its scope closes", () =>
538 Effect.gen(function*() {
539 const { attempts, httpClient } = yield* makeStatusHttpClient(200)
540 const flusher = yield* OtlpExporter.Flusher
541 const scope = yield* Scope.make()
542
543 const exporter = yield* makeExporterRaw("disabled").pipe(
544 Effect.provideService(HttpClient.HttpClient, httpClient),
545 Effect.provideService(Scope.Scope, scope)
546 )
547 exporter.push({ value: 1 })
548
549 yield* Scope.close(scope, Exit.void)
550 assert.strictEqual(yield* Ref.get(attempts), 1)
551
552 yield* flusher.flush
553 assert.strictEqual(yield* Ref.get(attempts), 1)
554 }).pipe(Effect.provide(OtlpExporter.layerFlusher)))
555
556 it.effect("succeeds with no registered exporters", () =>
557 Effect.gen(function*() {
558 const flusher = yield* OtlpExporter.Flusher
559 yield* flusher.flush
560 }).pipe(Effect.provide(OtlpExporter.layerFlusher)))
561
562 it.effect("is available when the SDK is disabled", () =>
563 Effect.gen(function*() {
564 const { attempts, httpClient } = yield* makeStatusHttpClient(200)
565 const layer = OtlpTracer.layerFromConfig().pipe(
566 Layer.provide(OtlpSerialization.layerJson),
567 Layer.provideMerge(Layer.succeed(HttpClient.HttpClient, httpClient)),
568 Layer.provideMerge(ConfigProvider.layer(ConfigProvider.fromEnv({
569 env: { OTEL_SDK_DISABLED: "true" }
570 })))
571 )
572
573 yield* Effect.gen(function*() {
574 const flusher = yield* OtlpExporter.Flusher
575 yield* flusher.flush
576 }).pipe(Effect.provide(layer))
577
578 assert.strictEqual(yield* Ref.get(attempts), 0)
579 }))
580
581 it.effect("does not fail when the collector returns 500", () =>
582 Effect.gen(function*() {
583 const { attempts, httpClient } = yield* makeStatusHttpClient(500)
584 yield* Effect.scoped(
585 Effect.gen(function*() {
586 const exporter = yield* makeExporterRaw(10)
587 const flusher = yield* OtlpExporter.Flusher
588
589 exporter.push({ value: 1 })
590 const fiber = yield* Effect.forkChild(flusher.flush)
591 yield* yieldNowN(3)
592 yield* TestClock.adjust("3 seconds")
593 yield* Fiber.join(fiber)
594 assert.strictEqual(yield* Ref.get(attempts), 4)
595
596 yield* flusher.flush
597 assert.strictEqual(yield* Ref.get(attempts), 4)
598 }).pipe(
599 Effect.provideService(HttpClient.HttpClient, httpClient),
600 Effect.provide(OtlpExporter.layerFlusher)
601 )
602 )
603 }))
604 })
605})