test(integration): add failure-accounting suite for timeout/crash enhanced metrics

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

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
+    })
+  };
+};

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"
+}

integration-tests/lib/stacks/failure-accounting.ts

@@ -0,0 +1,63 @@
+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 = [
+      // 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' } },
+    ];
+
+    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);
+    }
+  }
+}

integration-tests/tests/failure-accounting.test.ts

@@ -0,0 +1,182 @@
+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 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.
+ *
+ * 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 = 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;
+
+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 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';
+  }
+  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<void> {
+  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<InvocationResult> {
+  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<Counts> {
+  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<Counts> {
+  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<string, Counts> = {};
+  const observedByStrategy: Record<string, Counts> = {};
+
+  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);
+    });
+  });
+});

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,
   };
 }