feat(aws): Add support for streaming handlers

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/.npmrc

@@ -0,0 +1,2 @@
+@sentry:registry=http://127.0.0.1:4873
+@sentry-internal:registry=http://127.0.0.1:4873

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/package.json

@@ -0,0 +1,27 @@
+{
+  "name": "aws-serverless-streaming",
+  "version": "1.0.0",
+  "private": true,
+  "scripts": {
+    "start": "node src/run.mjs",
+    "test": "playwright test",
+    "clean": "npx rimraf node_modules pnpm-lock.yaml",
+    "test:build": "pnpm install",
+    "test:assert": "pnpm test"
+  },
+  "dependencies": {
+    "@sentry/aws-serverless": "* || latest"
+  },
+  "devDependencies": {
+    "@sentry-internal/test-utils": "link:../../../test-utils",
+    "@playwright/test": "~1.53.2"
+  },
+  "volta": {
+    "extends": "../../package.json"
+  },
+  "pnpm": {
+    "overrides": {
+      "@opentelemetry/instrumentation-aws-lambda": "file:/Users/martin/code/opentelemetry-js-contrib/packages/instrumentation-aws-lambda"
+    }
+  }
+}

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/playwright.config.ts

@@ -0,0 +1,3 @@
+import { getPlaywrightConfig } from '@sentry-internal/test-utils';
+
+export default getPlaywrightConfig();

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/src/lambda-function.mjs

@@ -0,0 +1,16 @@
+import * as Sentry from '@sentry/aws-serverless';
+
+let handler = async (event, responseStream, context) => {
+  Sentry.startSpan({ name: 'streaming-span', op: 'stream' }, () => {
+    responseStream.write('Starting stream\n');
+  });
+
+  responseStream.write('Continuing stream\n');
+  responseStream.write('Stream completed\n');
+  responseStream.end();
+};
+
+handler[Symbol.for('aws.lambda.runtime.handler.streaming')] = 'response';
+handler = Sentry.wrapHandler(handler);
+
+export { handler };

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/src/package.json

@@ -0,0 +1,5 @@
+{
+  "//": "This is a mock package.json file which is usually created by AWS when deploying the lambda. OTEL instrumentation tries to read this file to get the lambda version",
+  "name": "lambda",
+  "version": "1.0.0"
+}

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/src/run-lambda.mjs

@@ -0,0 +1,18 @@
+import { handler } from './lambda-function.mjs';
+import { Writable } from 'node:stream';
+
+const event = {};
+
+const context = {
+  invokedFunctionArn: 'arn:aws:lambda:us-east-1:123453789012:function:my-streaming-lambda',
+  functionName: 'my-streaming-lambda',
+};
+
+const responseStream = new Writable({
+  write: (chunk, encoding, callback) => {
+    console.log('Streamed chunk:', chunk.toString());
+    callback();
+  },
+});
+
+await handler(event, responseStream, context);

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/src/run.mjs

@@ -0,0 +1,16 @@
+import child_process from 'child_process';
+
+child_process.execSync('node ./src/run-lambda.mjs', {
+  stdio: 'inherit',
+  env: {
+    ...process.env,
+    // On AWS, LAMBDA_TASK_ROOT is usually /var/task but for testing, we set it to the CWD to correctly apply our handler
+    LAMBDA_TASK_ROOT: process.cwd(),
+    _HANDLER: 'src/lambda-function.handler',
+
+    NODE_OPTIONS: '--import @sentry/aws-serverless/awslambda-auto',
+    SENTRY_DSN: 'http://public@localhost:3031/1337',
+    SENTRY_TRACES_SAMPLE_RATE: '1.0',
+  },
+  cwd: process.cwd(),
+});

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/start-event-proxy.mjs

@@ -0,0 +1,6 @@
+import { startEventProxyServer } from '@sentry-internal/test-utils';
+
+startEventProxyServer({
+  port: 3031,
+  proxyServerName: 'aws-serverless-streaming',
+});

dev-packages/e2e-tests/test-applications/aws-serverless-streaming/tests/basic.test.ts

@@ -0,0 +1,54 @@
+import * as child_process from 'child_process';
+import { expect, test } from '@playwright/test';
+import { waitForTransaction } from '@sentry-internal/test-utils';
+
+test('AWS Serverless SDK sends events from streaming handler', async ({ request }) => {
+  const transactionEventPromise = waitForTransaction('aws-serverless-streaming', transactionEvent => {
+    return transactionEvent?.transaction === 'my-streaming-lambda';
+  });
+
+  // Waiting for 1s here because attaching the listener for events in `waitForTransaction` is not synchronous
+  // Since in this test, we don't start a browser via playwright, we don't have the usual delays (page.goto, etc)
+  // which are usually enough for us to never have noticed this race condition before.
+  // This is a workaround but probably sufficient as long as we only experience it in this test.
+  await new Promise<void>(resolve =>
+    setTimeout(() => {
+      resolve();
+    }, 1000),
+  );
+
+  child_process.execSync('pnpm start', {
+    stdio: 'inherit',
+  });
+
+  const transactionEvent = await transactionEventPromise;
+
+  expect(transactionEvent.transaction).toEqual('my-streaming-lambda');
+  expect(transactionEvent.contexts?.trace).toEqual({
+    data: {
+      'sentry.sample_rate': 1,
+      'sentry.source': 'custom',
+      'sentry.origin': 'auto.otel.aws-lambda',
+      'sentry.op': 'function.aws.lambda',
+      'cloud.account.id': '123453789012',
+      'faas.id': 'arn:aws:lambda:us-east-1:123453789012:function:my-streaming-lambda',
+      'faas.coldstart': true,
+      'otel.kind': 'SERVER',
+    },
+    op: 'function.aws.lambda',
+    origin: 'auto.otel.aws-lambda',
+    span_id: expect.stringMatching(/[a-f0-9]{16}/),
+    status: 'ok',
+    trace_id: expect.stringMatching(/[a-f0-9]{32}/),
+  });
+
+  expect(transactionEvent.spans).toHaveLength(1);
+
+  const streamingSpan = transactionEvent.spans?.[0];
+  expect(streamingSpan).toMatchObject({
+    description: 'streaming-span',
+    op: 'stream',
+    status: 'ok',
+    origin: 'manual',
+  });
+});

packages/aws-serverless/src/integration/awslambda.ts

@@ -1,8 +1,8 @@
 import { AwsLambdaInstrumentation } from '@opentelemetry/instrumentation-aws-lambda';
 import type { IntegrationFn } from '@sentry/core';
 import { defineIntegration, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN } from '@sentry/core';
-import { generateInstrumentOnce } from '@sentry/node';
-import { eventContextExtractor } from '../utils';
+import { captureException, generateInstrumentOnce } from '@sentry/node';
+import { eventContextExtractor, markEventUnhandled } from '../utils';
 
 interface AwsLambdaOptions {
   /**
@@ -27,6 +27,11 @@ export const instrumentAwsLambda = generateInstrumentOnce(
         span.setAttribute(SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, 'auto.otel.aws-lambda');
         span.setAttribute(SEMANTIC_ATTRIBUTE_SENTRY_OP, 'function.aws.lambda');
       },
+      responseHook(_span, { err }) {
+        if (err) {
+          captureException(err, scope => markEventUnhandled(scope, 'auto.function.aws-serverless.otel'));
+        }
+      },
     };
   },
 );