From 07ef7d3e142fa2702111d7cf34a42d12bfa2e505 Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Thu, 22 Oct 2026 13:55:36 -0400 Subject: [PATCH 1/3] test(integration): add failure-accounting suite for timeout/crash enhanced metrics Invokes a Node function 310 times sequentially with a timeout or crash every 50th invocation, under both the Default and End flush strategies, and asserts aws.lambda.enhanced.invocations/errors/timeouts match the Invoke responses. Co-Authored-By: Claude Opus 5.5 --- .gitlab/datasources/test-suites.yaml | 1 + integration-tests/bin/app.ts | 4 + .../lambda/failure-accounting-node/index.js | 18 ++ .../failure-accounting-node/package.json | 6 + .../lib/stacks/failure-accounting.ts | 62 ++++++ .../tests/failure-accounting.test.ts | 181 ++++++++++++++++++ integration-tests/tests/utils/lambda.ts | 6 + 7 files changed, 278 insertions(+) create mode 100644 integration-tests/lambda/failure-accounting-node/index.js create mode 100644 integration-tests/lambda/failure-accounting-node/package.json create mode 100644 integration-tests/lib/stacks/failure-accounting.ts create mode 100644 integration-tests/tests/failure-accounting.test.ts diff --git a/.gitlab/datasources/test-suites.yaml b/.gitlab/datasources/test-suites.yaml index 54a176c41..7da1e67fe 100644 --- a/.gitlab/datasources/test-suites.yaml +++ b/.gitlab/datasources/test-suites.yaml @@ -10,3 +10,4 @@ test_suites: - name: payload-size-e2e - name: dsm - name: durable-cold-start + - name: failure-accounting diff --git a/integration-tests/bin/app.ts b/integration-tests/bin/app.ts index 49465b977..cb72d611b 100644 --- a/integration-tests/bin/app.ts +++ b/integration-tests/bin/app.ts @@ -12,6 +12,7 @@ import {CustomMetrics} from '../lib/stacks/custom-metrics'; import {PayloadSize} from '../lib/stacks/payload-size'; import {Dsm} from '../lib/stacks/dsm'; import {DurableColdStart} from '../lib/stacks/durable-cold-start'; +import {FailureAccounting} from '../lib/stacks/failure-accounting'; import {AuthRoleStack} from '../lib/auth-role'; import {ACCOUNT, IDENTIFIER, REGION} from '../config'; import {CapacityProviderStack} from "../lib/capacity-provider"; @@ -67,6 +68,9 @@ const stacks = [ new Dsm(app, `${IDENTIFIER}-dsm`, { env, }), + new FailureAccounting(app, `${IDENTIFIER}-failure-accounting`, { + env, + }), ] // Tag all stacks so we can easily clean them up diff --git a/integration-tests/lambda/failure-accounting-node/index.js b/integration-tests/lambda/failure-accounting-node/index.js new file mode 100644 index 000000000..c8a3e32c8 --- /dev/null +++ b/integration-tests/lambda/failure-accounting-node/index.js @@ -0,0 +1,18 @@ +exports.handler = async (event, context) => { + const mode = event && event.mode; + + if (mode === 'timeout') { + // Exceeds the 10 s function timeout. + await new Promise(resolve => setTimeout(resolve, 30000)); + } else if (mode === 'crash') { + process.exit(1); + } + + return { + statusCode: 200, + body: JSON.stringify({ + message: 'Success', + requestId: context.awsRequestId + }) + }; +}; diff --git a/integration-tests/lambda/failure-accounting-node/package.json b/integration-tests/lambda/failure-accounting-node/package.json new file mode 100644 index 000000000..69128c4d6 --- /dev/null +++ b/integration-tests/lambda/failure-accounting-node/package.json @@ -0,0 +1,6 @@ +{ + "name": "failure-accounting-node", + "version": "1.0.0", + "description": "Lambda that times out or crashes on demand for failure-accounting integration tests", + "main": "index.js" +} diff --git a/integration-tests/lib/stacks/failure-accounting.ts b/integration-tests/lib/stacks/failure-accounting.ts new file mode 100644 index 000000000..ef238bcef --- /dev/null +++ b/integration-tests/lib/stacks/failure-accounting.ts @@ -0,0 +1,62 @@ +import * as cdk from 'aws-cdk-lib'; +import * as lambda from 'aws-cdk-lib/aws-lambda'; +import { Construct } from 'constructs'; +import { + createLogGroup, + defaultDatadogEnvVariables, + defaultDatadogSecretPolicy, + getExtensionLayer, + getDefaultNodeLayer, + defaultNodeRuntime, +} from '../util'; + +/** + * Failure-accounting test stack. + * + * Deploys the same Node function twice: once with the Default (adaptive) + * flush strategy and once pinned to `end` as a control. Each function has + * reserved concurrency 1 so sequential invocations share a single execution + * environment, letting the Default strategy adapt to periodic flushing before + * the test triggers a timeout or crash. + */ +export class FailureAccounting extends cdk.Stack { + constructor(scope: Construct, id: string, props: cdk.StackProps) { + super(scope, id, props); + + const extensionLayer = getExtensionLayer(this); + const nodeLayer = getDefaultNodeLayer(this); + + // Drop the shared `end` override so the extension uses its Default strategy. + const { DD_SERVERLESS_FLUSH_STRATEGY: _, ...envWithoutFlushStrategy } = defaultDatadogEnvVariables; + + const variants = [ + { name: 'default', env: envWithoutFlushStrategy }, + { name: 'end', env: { ...envWithoutFlushStrategy, DD_SERVERLESS_FLUSH_STRATEGY: 'end' } }, + ]; + + for (const variant of variants) { + const functionName = `${id}-${variant.name}-lambda`; + const fn = new lambda.Function(this, functionName, { + runtime: defaultNodeRuntime, + architecture: lambda.Architecture.ARM_64, + handler: '/opt/nodejs/node_modules/datadog-lambda-js/handler.handler', + code: lambda.Code.fromAsset('./lambda/failure-accounting-node'), + functionName, + // Leaves headroom for End-mode flushes, which count against the timeout. + timeout: cdk.Duration.seconds(10), + memorySize: 1024, + reservedConcurrentExecutions: 1, + environment: { + ...variant.env, + DD_SERVICE: functionName, + DD_TRACE_ENABLED: 'true', + DD_LAMBDA_HANDLER: 'index.handler', + }, + logGroup: createLogGroup(this, functionName), + }); + fn.addToRolePolicy(defaultDatadogSecretPolicy); + fn.addLayers(extensionLayer); + fn.addLayers(nodeLayer); + } + } +} diff --git a/integration-tests/tests/failure-accounting.test.ts b/integration-tests/tests/failure-accounting.test.ts new file mode 100644 index 000000000..90c80994c --- /dev/null +++ b/integration-tests/tests/failure-accounting.test.ts @@ -0,0 +1,181 @@ +import { TooManyRequestsException } from '@aws-sdk/client-lambda'; +import { invokeLambda, InvocationResult } from './utils/lambda'; +import { getMetricCount } from './utils/datadog'; +import { IDENTIFIER } from '../config'; + +/** + * Failure-accounting test: every timeout / crash must be reflected in + * `aws.lambda.enhanced.errors` and `aws.lambda.enhanced.timeouts`. + * + * The extension can lose these metrics when an invocation fails after it has + * switched from End-of-invocation flushing to periodic/continuous flushing. + * The Default strategy only adapts after ~20 invocations in one execution + * environment, and a timeout/crash resets the environment (new extension + * process, back in End mode). Failures are therefore spaced 50 invocations + * apart so each fresh environment has adapted before the next failure. + * Reserved concurrency 1 plus sequential invokes keep everything in one + * environment at a time. + * + * The `end`-strategy function never adapts, so it is the control: if it also + * miscounts, the problem is in the test or ingestion rather than the + * flush-mode transition. + */ +const stackName = `${IDENTIFIER}-failure-accounting`; + +interface FailureCase { + strategy: string; + functionName: string; +} + +const cases: FailureCase[] = [ + { strategy: 'default', functionName: `${stackName}-default-lambda` }, + { strategy: 'end', functionName: `${stackName}-end-lambda` }, +]; + +type Mode = 'ok' | 'timeout' | 'crash'; + +const FAILURE_SPACING = 50; +const FAILURE_COUNT = 6; +// Trailing successes so the environment created after the last failure also runs. +const TOTAL_INVOCATIONS = FAILURE_SPACING * FAILURE_COUNT + 10; + +const INVOCATIONS_METRIC = 'aws.lambda.enhanced.invocations'; +const ERRORS_METRIC = 'aws.lambda.enhanced.errors'; +const TIMEOUTS_METRIC = 'aws.lambda.enhanced.timeouts'; + +const INITIAL_WAIT_MS = 90 * 1000; // wait before first query +const POLL_INTERVAL_MS = 30 * 1000; // re-query cadence +const TOTAL_BUDGET_MS = 12 * 60 * 1000; // polling ceiling, measured from the last invoke +const INVOKE_PHASE_BUDGET_MS = 15 * 60 * 1000; + +// Failure k (1-based) lands on invocation 50k; odd k times out, even k crashes. +function modeFor(invocation: number): Mode { + if (invocation % FAILURE_SPACING !== 0 || invocation / FAILURE_SPACING > FAILURE_COUNT) { + return 'ok'; + } + return (invocation / FAILURE_SPACING) % 2 === 1 ? 'timeout' : 'crash'; +} + +interface Counts { + invocations: number; + errors: number; + timeouts: number; +} + +function isTimeout(result: InvocationResult): boolean { + try { + const body = JSON.parse(result.payload ?? '{}'); + return body.errorType === 'Sandbox.Timedout' || /timed out/i.test(body.errorMessage ?? ''); + } catch { + return false; + } +} + +async function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +// With reserved concurrency 1, the invoke right after a timeout/crash can be +// throttled while the old environment shuts down. Throttled invokes never run, +// so retrying does not affect the ground truth. +async function invokeWithThrottleRetry(functionName: string, payload: object): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await invokeLambda(functionName, payload); + } catch (err) { + if (!(err instanceof TooManyRequestsException) || attempt >= 10) { + throw err; + } + await sleep(1000 * attempt); + } + } +} + +async function runSchedule(functionName: string): Promise { + const truth: Counts = { invocations: 0, errors: 0, timeouts: 0 }; + for (let i = 1; i <= TOTAL_INVOCATIONS; i++) { + const mode = modeFor(i); + const result = await invokeWithThrottleRetry(functionName, { mode }); + truth.invocations++; + if (result.functionError) { + truth.errors++; + if (isTimeout(result)) { + truth.timeouts++; + } + console.log(`${functionName} #${i} mode=${mode} functionError=${result.functionError} payload=${result.payload}`); + } else if (mode !== 'ok') { + console.warn(`${functionName} #${i} mode=${mode} unexpectedly succeeded`); + } + } + return truth; +} + +async function fetchCounts(functionName: string, start: number, end: number): Promise { + const [invocations, errors, timeouts] = await Promise.all([ + getMetricCount(INVOCATIONS_METRIC, functionName, start, end), + getMetricCount(ERRORS_METRIC, functionName, start, end), + getMetricCount(TIMEOUTS_METRIC, functionName, start, end), + ]); + return { invocations, errors, timeouts }; +} + +function reachedExpected(observed: Counts, expected: Counts): boolean { + return observed.invocations >= expected.invocations + && observed.errors >= expected.errors + && observed.timeouts >= expected.timeouts; +} + +describe('Failure Accounting Integration Tests', () => { + const truthByStrategy: Record = {}; + const observedByStrategy: Record = {}; + + beforeAll(async () => { + // Same bucket-alignment margin as oom.test.ts. + const windowStart = Date.now() - 60 * 1000; + + const truths = await Promise.all(cases.map((c) => runSchedule(c.functionName))); + cases.forEach((c, i) => { truthByStrategy[c.strategy] = truths[i]; }); + console.log('Ground truth by strategy:', truthByStrategy); + + const invokesDone = Date.now(); + await sleep(INITIAL_WAIT_MS); + + const deadline = invokesDone + TOTAL_BUDGET_MS; + let attempt = 0; + while (Date.now() < deadline) { + attempt++; + const now = Date.now(); + const observed = await Promise.all(cases.map((c) => fetchCounts(c.functionName, windowStart, now))); + cases.forEach((c, i) => { observedByStrategy[c.strategy] = observed[i]; }); + const pending = cases.filter((c) => !reachedExpected(observedByStrategy[c.strategy], truthByStrategy[c.strategy])); + console.log(`Failure-accounting poll #${attempt}:`, observedByStrategy, pending.length ? `(pending: ${pending.map((c) => c.strategy).join(', ')})` : '(all reached expected)'); + if (pending.length === 0) { + break; + } + await sleep(POLL_INTERVAL_MS); + } + + console.log('Observed counts by strategy (final):', observedByStrategy); + }, INVOKE_PHASE_BUDGET_MS + TOTAL_BUDGET_MS + 60 * 1000); + + describe.each(cases)('$strategy flush strategy', ({ strategy }) => { + it('should produce the scheduled failures (test harness sanity check)', () => { + const truth = truthByStrategy[strategy]; + expect(truth.invocations).toBe(TOTAL_INVOCATIONS); + expect(truth.errors).toBe(FAILURE_COUNT); + expect(truth.timeouts).toBe(FAILURE_COUNT / 2); + }); + + it('should report one invocations metric per invocation', () => { + expect(observedByStrategy[strategy]?.invocations).toBe(truthByStrategy[strategy].invocations); + }); + + it('should report one errors metric per timeout or crash', () => { + expect(observedByStrategy[strategy]?.errors).toBe(truthByStrategy[strategy].errors); + }); + + it('should report one timeouts metric per timeout', () => { + expect(observedByStrategy[strategy]?.timeouts).toBe(truthByStrategy[strategy].timeouts); + }); + }); +}); diff --git a/integration-tests/tests/utils/lambda.ts b/integration-tests/tests/utils/lambda.ts index 5ac14ac59..fc2c114ee 100644 --- a/integration-tests/tests/utils/lambda.ts +++ b/integration-tests/tests/utils/lambda.ts @@ -37,6 +37,10 @@ export interface InvocationResult { functionName: string; requestId: string; statusCode?: number; + // Set by Lambda when the function errored, timed out, or crashed (e.g. "Unhandled"). + functionError?: string; + // Decoded response payload; on failure this is Lambda's `{errorType, errorMessage}` body. + payload?: string; } export async function invokeLambda( @@ -62,6 +66,8 @@ export async function invokeLambda( functionName, requestId, statusCode: response.StatusCode || 200, + functionError: response.FunctionError, + payload: response.Payload ? Buffer.from(response.Payload).toString('utf-8') : undefined, }; } From 71aa0143d638a5da7f923c16561b903a089b5ddf Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Thu, 22 Oct 2026 17:09:30 -0400 Subject: [PATCH 2/3] test(integration): run 12 failures per failure-accounting run and enable debug logs The first CI run lost one timeout's errors/timeouts counts on the default-strategy function in one of two attempts, so the loss is a race. More resets per run make it more likely to show up, and debug logs show when runtimeDone reaches the extension relative to shutdown. Co-Authored-By: Claude Opus 5.5 --- integration-tests/lib/stacks/failure-accounting.ts | 3 ++- integration-tests/tests/failure-accounting.test.ts | 9 +++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/integration-tests/lib/stacks/failure-accounting.ts b/integration-tests/lib/stacks/failure-accounting.ts index ef238bcef..8b406f335 100644 --- a/integration-tests/lib/stacks/failure-accounting.ts +++ b/integration-tests/lib/stacks/failure-accounting.ts @@ -30,7 +30,8 @@ export class FailureAccounting extends cdk.Stack { const { DD_SERVERLESS_FLUSH_STRATEGY: _, ...envWithoutFlushStrategy } = defaultDatadogEnvVariables; const variants = [ - { name: 'default', env: envWithoutFlushStrategy }, + // Debug logs show whether runtimeDone reaches the extension before or after shutdown. + { name: 'default', env: { ...envWithoutFlushStrategy, DD_LOG_LEVEL: 'debug' } }, { name: 'end', env: { ...envWithoutFlushStrategy, DD_SERVERLESS_FLUSH_STRATEGY: 'end' } }, ]; diff --git a/integration-tests/tests/failure-accounting.test.ts b/integration-tests/tests/failure-accounting.test.ts index 90c80994c..db697c897 100644 --- a/integration-tests/tests/failure-accounting.test.ts +++ b/integration-tests/tests/failure-accounting.test.ts @@ -11,7 +11,7 @@ import { IDENTIFIER } from '../config'; * switched from End-of-invocation flushing to periodic/continuous flushing. * The Default strategy only adapts after ~20 invocations in one execution * environment, and a timeout/crash resets the environment (new extension - * process, back in End mode). Failures are therefore spaced 50 invocations + * process, back in End mode). Failures are therefore spaced 30 invocations * apart so each fresh environment has adapted before the next failure. * Reserved concurrency 1 plus sequential invokes keep everything in one * environment at a time. @@ -34,8 +34,9 @@ const cases: FailureCase[] = [ type Mode = 'ok' | 'timeout' | 'crash'; -const FAILURE_SPACING = 50; -const FAILURE_COUNT = 6; +const FAILURE_SPACING = 30; +// The loss is a race, so run enough resets to hit it reliably. +const FAILURE_COUNT = 12; // Trailing successes so the environment created after the last failure also runs. const TOTAL_INVOCATIONS = FAILURE_SPACING * FAILURE_COUNT + 10; @@ -48,7 +49,7 @@ const POLL_INTERVAL_MS = 30 * 1000; // re-query cadence const TOTAL_BUDGET_MS = 12 * 60 * 1000; // polling ceiling, measured from the last invoke const INVOKE_PHASE_BUDGET_MS = 15 * 60 * 1000; -// Failure k (1-based) lands on invocation 50k; odd k times out, even k crashes. +// Failure k (1-based) lands on invocation 30k; odd k times out, even k crashes. function modeFor(invocation: number): Mode { if (invocation % FAILURE_SPACING !== 0 || invocation / FAILURE_SPACING > FAILURE_COUNT) { return 'ok'; From a086bc4122026a0288630243f9f737fb6c0e5f3b Mon Sep 17 00:00:00 2001 From: Darcy Rayner Date: Fri, 23 Oct 2026 12:46:21 -0400 Subject: [PATCH 3/3] ci: keep failure-accounting out of the integration matrix until the fix lands The suite reproduces a race that the current extension loses intermittently, so registering it before the fix would make unrelated pipelines flaky. The fix PR registers it. Co-Authored-By: Claude Opus 5.5 --- .gitlab/datasources/test-suites.yaml | 1 - 1 file changed, 1 deletion(-) diff --git a/.gitlab/datasources/test-suites.yaml b/.gitlab/datasources/test-suites.yaml index 7da1e67fe..54a176c41 100644 --- a/.gitlab/datasources/test-suites.yaml +++ b/.gitlab/datasources/test-suites.yaml @@ -10,4 +10,3 @@ test_suites: - name: payload-size-e2e - name: dsm - name: durable-cold-start - - name: failure-accounting