| 1 | import * as Cloudflare from "@/Cloudflare/index.ts"; |
| 2 | import * as Telemetry from "@/Telemetry.ts"; |
| 3 | import * as Config from "effect/Config"; |
| 4 | import * as Duration from "effect/Duration"; |
| 5 | import * as Effect from "effect/Effect"; |
| 6 | import * as Layer from "effect/Layer"; |
| 7 | import * as HttpClientRequest from "effect/http/HttpClientRequest"; |
| 8 | import { HttpServerRequest } from "effect/http/HttpServerRequest"; |
| 9 | import * as HttpServerResponse from "effect/http/HttpServerResponse"; |
| 10 | import * as OtlpSerialization from "effect/observability/OtlpSerialization"; |
| 11 | import * as OtlpTracer from "effect/observability/OtlpTracer"; |
| 12 | |
| 13 | /** |
| 14 | * Durable Object target whose events emit child spans — one HTTP fetch |
| 15 | * event and one RPC method event, so both DurableObjectBridge paths get |
| 16 | * telemetry coverage. |
| 17 | */ |
| 18 | export class OtelEventFlushTarget extends Cloudflare.DurableObject<OtelEventFlushTarget>()( |
| 19 | "OtelEventFlushTarget", |
| 20 | Effect.succeed( |
| 21 | Effect.succeed({ |
| 22 | fetch: Effect.succeed(HttpServerResponse.text("durable-object-ok")).pipe( |
| 23 | Effect.withSpan("otel-event-flush.child"), |
| 24 | ), |
| 25 | ping: () => |
| 26 | Effect.succeed("durable-object-rpc-ok").pipe( |
| 27 | Effect.withSpan("otel-event-flush.rpc"), |
| 28 | ), |
| 29 | }), |
| 30 | ), |
| 31 | ) {} |
| 32 | |
| 33 | /** Worker that emits one Worker and one Durable Object OTLP event batch. */ |
| 34 | export default class OtelEventFlushWorker extends Cloudflare.Worker<OtelEventFlushWorker>()( |
| 35 | "OtelEventFlushWorker", |
| 36 | { |
| 37 | main: import.meta.url, |
| 38 | env: { |
| 39 | OTLP_EVENT_FLUSH_URL: Config.String("OTLP_EVENT_FLUSH_URL"), |
| 40 | OTLP_EVENT_FLUSH_SHUTDOWN_TIMEOUT: Config.String( |
| 41 | "OTLP_EVENT_FLUSH_SHUTDOWN_TIMEOUT", |
| 42 | ).pipe(Config.withDefault("0 millis")), |
| 43 | }, |
| 44 | }, |
| 45 | Effect.gen(function* () { |
| 46 | const targetNamespace = yield* OtelEventFlushTarget; |
| 47 | |
| 48 | return { |
| 49 | fetch: Effect.gen(function* () { |
| 50 | const request = yield* HttpServerRequest; |
| 51 | if (request.url.startsWith("/rpc")) { |
| 52 | const pong = yield* targetNamespace.getByName("target").ping(); |
| 53 | return HttpServerResponse.text(`worker-saw:${pong}`); |
| 54 | } |
| 55 | const targetClient = Cloudflare.toHttpClient( |
| 56 | targetNamespace.getByName("target"), |
| 57 | ); |
| 58 | const response = yield* targetClient.execute( |
| 59 | HttpClientRequest.get("http://otel-event-flush-target/"), |
| 60 | ); |
| 61 | return HttpServerResponse.text(`worker-saw:${yield* response.text}`); |
| 62 | }).pipe(Effect.withSpan("otel-event-flush.worker"), Effect.orDie), |
| 63 | }; |
| 64 | }).pipe( |
| 65 | Effect.provide( |
| 66 | Layer.unwrap( |
| 67 | Effect.gen(function* () { |
| 68 | const url = yield* Config.String("OTLP_EVENT_FLUSH_URL"); |
| 69 | const shutdownTimeout = yield* Config.Duration( |
| 70 | "OTLP_EVENT_FLUSH_SHUTDOWN_TIMEOUT", |
| 71 | ).pipe(Config.withDefault(Duration.zero)); |
| 72 | // Custom exporter selection is limited to the deadline regression. |
| 73 | return Duration.toMillis(shutdownTimeout) === 0 |
| 74 | ? Telemetry.layerOtlp({ |
| 75 | traces: { url }, |
| 76 | serviceName: "otel-event-flush-test", |
| 77 | }) |
| 78 | : Telemetry.layer( |
| 79 | OtlpTracer.layer({ |
| 80 | url, |
| 81 | resource: { serviceName: "otel-event-flush-test" }, |
| 82 | exportInterval: "1 hour", |
| 83 | shutdownTimeout, |
| 84 | }).pipe(Layer.provide(OtlpSerialization.layerJson)), |
| 85 | ); |
| 86 | }), |
| 87 | ), |
| 88 | ), |
| 89 | ), |
| 90 | ) {} |