|
| 1 | +import { spawn } from "node:child_process"; |
| 2 | +import { once } from "node:events"; |
| 3 | +import * as http from "node:http"; |
| 4 | +import * as Module from "node:module"; |
| 5 | +import * as path from "node:path"; |
| 6 | +import { pathToFileURL } from "node:url"; |
| 7 | +import * as zlib from "node:zlib"; |
| 8 | + |
| 9 | +const fixtureDirectory = path.join(__dirname, "runtime", "fixtures"); |
| 10 | +const handlerEntry = pathToFileURL(path.join(__dirname, "..", "dist", "handler.mjs")).href; |
| 11 | +const runnerPath = path.join(fixtureDirectory, "published-handler-runner.mjs"); |
| 12 | +const tracePayloads: Buffer[] = []; |
| 13 | + |
| 14 | +let agent: http.Server; |
| 15 | +let resolveTrace: (() => void) | undefined; |
| 16 | +let traceReceived: Promise<void>; |
| 17 | + |
| 18 | +/** |
| 19 | + * @param {Buffer} body Encoded trace body. |
| 20 | + * @param {string | string[] | undefined} encoding Content encoding from the trace request. |
| 21 | + */ |
| 22 | +function recordTrace(body: Buffer, encoding: string | string[] | undefined): void { |
| 23 | + let payload = body; |
| 24 | + if (encoding === "gzip") { |
| 25 | + payload = zlib.gunzipSync(body); |
| 26 | + } else if (encoding === "deflate") { |
| 27 | + payload = zlib.inflateSync(body); |
| 28 | + } |
| 29 | + |
| 30 | + tracePayloads.push(payload); |
| 31 | + resolveTrace?.(); |
| 32 | +} |
| 33 | + |
| 34 | +/** |
| 35 | + * @param {http.IncomingMessage} request Agent request. |
| 36 | + * @param {http.ServerResponse} response Agent response. |
| 37 | + */ |
| 38 | +function respond(request: http.IncomingMessage, response: http.ServerResponse): void { |
| 39 | + const chunks: Buffer[] = []; |
| 40 | + |
| 41 | + /** @param {Buffer} chunk Request body chunk. */ |
| 42 | + function recordChunk(chunk: Buffer): void { |
| 43 | + chunks.push(chunk); |
| 44 | + } |
| 45 | + |
| 46 | + request.on("data", recordChunk); |
| 47 | + request.once("end", () => { |
| 48 | + response.setHeader("content-type", "application/json"); |
| 49 | + |
| 50 | + if (request.url === "/info") { |
| 51 | + response.end(JSON.stringify({ endpoints: ["/v0.4/traces", "/v0.5/traces"] })); |
| 52 | + return; |
| 53 | + } |
| 54 | + |
| 55 | + if (request.url?.startsWith("/v0.4/traces") || request.url?.startsWith("/v0.5/traces")) { |
| 56 | + recordTrace(Buffer.concat(chunks), request.headers["content-encoding"]); |
| 57 | + } |
| 58 | + |
| 59 | + response.end(JSON.stringify({ rate_by_service: {} })); |
| 60 | + }); |
| 61 | +} |
| 62 | + |
| 63 | +/** @param {() => void} resolve Trace promise resolver. */ |
| 64 | +function captureTraceResolver(resolve: () => void): void { |
| 65 | + resolveTrace = resolve; |
| 66 | +} |
| 67 | + |
| 68 | +/** @param {Promise<[unknown]>} messagePromise Child IPC message signal. */ |
| 69 | +async function readChildMessage(messagePromise: Promise<[unknown]>): Promise<unknown> { |
| 70 | + const [message] = await messagePromise; |
| 71 | + return message; |
| 72 | +} |
| 73 | + |
| 74 | +/** @param {boolean} traceEnabled Whether tracing is enabled for the child process. */ |
| 75 | +async function runPublishedHandler(traceEnabled: boolean): Promise<unknown> { |
| 76 | + const address = agent.address(); |
| 77 | + if (address === null || typeof address === "string") { |
| 78 | + throw new Error("Mock agent is not listening on a TCP port"); |
| 79 | + } |
| 80 | + |
| 81 | + const env: NodeJS.ProcessEnv = { |
| 82 | + ...process.env, |
| 83 | + AWS_ACCESS_KEY_ID: "test-access-key", |
| 84 | + AWS_EXECUTION_ENV: `AWS_Lambda_nodejs${process.versions.node.split(".")[0]}.x`, |
| 85 | + AWS_LAMBDA_FUNCTION_MEMORY_SIZE: "1024", |
| 86 | + AWS_LAMBDA_FUNCTION_VERSION: "$LATEST", |
| 87 | + AWS_REGION: "us-east-1", |
| 88 | + AWS_SECRET_ACCESS_KEY: "test-secret-key", |
| 89 | + DD_COLD_START_TRACING: "false", |
| 90 | + DD_INSTRUMENTATION_TELEMETRY_ENABLED: "false", |
| 91 | + DD_LAMBDA_HANDLER: "esm-instrumentation-handler.handle", |
| 92 | + DD_PROFILING_ENABLED: "false", |
| 93 | + DD_REMOTE_CONFIGURATION_ENABLED: "false", |
| 94 | + DD_RUNTIME_METRICS_ENABLED: "false", |
| 95 | + DD_TELEMETRY_ENABLED: "false", |
| 96 | + DD_TRACE_AGENT_URL: `http://127.0.0.1:${address.port}`, |
| 97 | + DD_TRACE_ENABLED: String(traceEnabled), |
| 98 | + DD_TRACE_STARTUP_LOGS: "false", |
| 99 | + LAMBDA_TASK_ROOT: fixtureDirectory, |
| 100 | + HANDLER_TEST_ENTRY: handlerEntry, |
| 101 | + }; |
| 102 | + delete env.NODE_OPTIONS; |
| 103 | + delete env.OTEL_LOGS_EXPORTER; |
| 104 | + delete env.OTEL_METRICS_EXPORTER; |
| 105 | + delete env.OTEL_TRACES_EXPORTER; |
| 106 | + |
| 107 | + const child = spawn(process.execPath, [runnerPath], { |
| 108 | + env, |
| 109 | + stdio: ["ignore", "pipe", "pipe", "ipc"], |
| 110 | + }); |
| 111 | + const closePromise = once(child, "close") as Promise<[number | null, NodeJS.Signals | null]>; |
| 112 | + const messagePromise = once(child, "message") as Promise<[unknown]>; |
| 113 | + let standardOutput = ""; |
| 114 | + let standardError = ""; |
| 115 | + |
| 116 | + child.stdout?.setEncoding("utf8"); |
| 117 | + child.stderr?.setEncoding("utf8"); |
| 118 | + |
| 119 | + /** @param {string} chunk Standard output chunk. */ |
| 120 | + function recordStandardOutput(chunk: string): void { |
| 121 | + standardOutput += chunk; |
| 122 | + } |
| 123 | + |
| 124 | + /** @param {string} chunk Standard error chunk. */ |
| 125 | + function recordStandardError(chunk: string): void { |
| 126 | + standardError += chunk; |
| 127 | + } |
| 128 | + |
| 129 | + async function failOnEarlyClose(): Promise<never> { |
| 130 | + const [exitCode, signal] = await closePromise; |
| 131 | + throw new Error( |
| 132 | + `Published handler exited before completing the test: code=${exitCode}, signal=${signal}` + |
| 133 | + `\nstdout:\n${standardOutput}\nstderr:\n${standardError}`, |
| 134 | + ); |
| 135 | + } |
| 136 | + |
| 137 | + child.stdout?.on("data", recordStandardOutput); |
| 138 | + child.stderr?.on("data", recordStandardError); |
| 139 | + const watchdog = setTimeout(() => child.kill("SIGKILL"), 15_000); |
| 140 | + watchdog.unref(); |
| 141 | + |
| 142 | + try { |
| 143 | + const message = await Promise.race([readChildMessage(messagePromise), failOnEarlyClose()]); |
| 144 | + |
| 145 | + if (traceEnabled) { |
| 146 | + await Promise.race([traceReceived, failOnEarlyClose()]); |
| 147 | + } |
| 148 | + |
| 149 | + child.send("exit"); |
| 150 | + const [exitCode, signal] = await closePromise; |
| 151 | + if (exitCode !== 0 || signal !== null) { |
| 152 | + throw new Error(`Published handler failed: code=${exitCode}, signal=${signal}\n${standardError}`); |
| 153 | + } |
| 154 | + |
| 155 | + return message; |
| 156 | + } finally { |
| 157 | + clearTimeout(watchdog); |
| 158 | + if (child.exitCode === null && child.signalCode === null) { |
| 159 | + child.kill("SIGKILL"); |
| 160 | + } |
| 161 | + } |
| 162 | +} |
| 163 | + |
| 164 | +const moduleWithRegister = Module as typeof Module & { register?: unknown }; |
| 165 | +const nodeMajor = Number(process.versions.node.split(".")[0]); |
| 166 | +const supportsDurableFixture = nodeMajor >= 22 && typeof moduleWithRegister.register === "function"; |
| 167 | +const describeWithESMLoader = supportsDurableFixture ? describe : describe.skip; |
| 168 | + |
| 169 | +describeWithESMLoader("published ESM handler", () => { |
| 170 | + jest.setTimeout(30_000); |
| 171 | + |
| 172 | + beforeAll(async () => { |
| 173 | + agent = http.createServer(respond); |
| 174 | + const listening = once(agent, "listening"); |
| 175 | + agent.listen(0, "127.0.0.1"); |
| 176 | + await listening; |
| 177 | + }); |
| 178 | + |
| 179 | + afterAll(async () => { |
| 180 | + const closed = once(agent, "close"); |
| 181 | + agent.close(); |
| 182 | + await closed; |
| 183 | + }); |
| 184 | + |
| 185 | + beforeEach(() => { |
| 186 | + tracePayloads.length = 0; |
| 187 | + traceReceived = new Promise(captureTraceResolver); |
| 188 | + }); |
| 189 | + |
| 190 | + it("instruments an ESM durable handler through the published handler", async () => { |
| 191 | + const message = await runPublishedHandler(true); |
| 192 | + const payload = Buffer.concat(tracePayloads); |
| 193 | + |
| 194 | + expect(message).toEqual({ |
| 195 | + registerLoaded: true, |
| 196 | + result: expect.objectContaining({ Status: "SUCCEEDED" }), |
| 197 | + }); |
| 198 | + expect(payload.includes(Buffer.from("aws.lambda"))).toBe(true); |
| 199 | + expect(payload.includes(Buffer.from("aws.durable.execute"))).toBe(true); |
| 200 | + }); |
| 201 | + |
| 202 | + it("does not register tracing when tracing is disabled", async () => { |
| 203 | + const message = await runPublishedHandler(false); |
| 204 | + |
| 205 | + expect(message).toEqual({ |
| 206 | + registerLoaded: false, |
| 207 | + result: expect.objectContaining({ Status: "SUCCEEDED" }), |
| 208 | + }); |
| 209 | + expect(tracePayloads).toHaveLength(0); |
| 210 | + }); |
| 211 | +}); |
0 commit comments