Forward each invocation's logs before Lambda freezes, with an E2E of the AWS Lambda docs

.github/workflows/notify_linear_failure.yml

@@ -0,0 +1,105 @@
+name: "Notify Linear on failure"
+
+on:
+  workflow_call:
+
+jobs:
+  create-linear-issue:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      actions: read
+    steps:
+      - name: "Gather failure context"
+        id: ctx
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          RUN_ID: ${{ github.event.workflow_run.id || github.run_id }}
+          RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt || github.run_attempt }}
+        run: |
+          FAILED=$(gh api \
+            "repos/${{ github.repository }}/actions/runs/$RUN_ID/attempts/$RUN_ATTEMPT/jobs" \
+            --paginate \
+            --jq '[.jobs[]
+                    | select(.conclusion == "failure")
+                    | "- **\(.name)** — failed at _\([.steps[] | select(.conclusion == "failure") | .name] | join(", "))_"]
+                  | join("\n")')
+
+          if [ -z "$FAILED" ]; then
+            FAILED="_(no failed jobs reported by the API yet)_"
+          fi
+
+          {
+            echo "failed_jobs<<CTXEOF"
+            echo "$FAILED"
+            echo "CTXEOF"
+          } >> "$GITHUB_OUTPUT"
+      - name: "Create Linear issue"
+        env:
+          LINEAR_API_KEY: ${{ secrets.LINEAR_API_KEY }}
+          LINEAR_TEAM_ID: ${{ secrets.LINEAR_TEAM_TELEMETRY_ID }}
+          LINEAR_STATUS_PLAN_ID: ${{ secrets.LINEAR_STATUS_PLAN_ID }}
+          LINEAR_PROJECT_INGESTION_FOR_EVERYTHING_ID: ${{ secrets.LINEAR_PROJECT_INGESTION_FOR_EVERYTHING_ID }}
+          LINEAR_ASSIGNEE_PETR_ID: ${{ secrets.LINEAR_ASSIGNEE_PETR_ID }}
+          WORKFLOW: ${{ github.event.workflow_run.name || github.workflow }}
+          REF_NAME: ${{ github.event.workflow_run.head_branch || github.ref_name }}
+          RUN_URL: ${{ github.event.workflow_run.html_url || format('{0}/{1}/actions/runs/{2}', github.server_url, github.repository, github.run_id) }}
+          RUN_ATTEMPT: ${{ github.event.workflow_run.run_attempt || github.run_attempt }}
+          EVENT_NAME: ${{ github.event.workflow_run.event || github.event_name }}
+          SHA: ${{ github.event.workflow_run.head_sha || github.sha }}
+          COMMIT_MESSAGE: ${{ github.event.workflow_run.head_commit.message || github.event.head_commit.message }}
+          FAILED_JOBS: ${{ steps.ctx.outputs.failed_jobs }}
+        run: |
+          SHORT_SHA="${SHA:0:7}"
+
+          # head_commit is only populated on push events; skip the line otherwise
+          COMMIT_LINE="**Commit:** \`$SHORT_SHA\`"
+          if [ -n "$COMMIT_MESSAGE" ]; then
+            COMMIT_SUBJECT="$(printf '%s' "$COMMIT_MESSAGE" | head -n1)"
+            COMMIT_LINE="**Commit:** \`$SHORT_SHA\` — $COMMIT_SUBJECT"
+          fi
+
+          ATTEMPT_NOTE=""
+          if [ "$RUN_ATTEMPT" != "1" ]; then
+            ATTEMPT_NOTE=" (attempt #$RUN_ATTEMPT)"
+          fi
+
+          TITLE="E2E failure: $WORKFLOW"
+          DESCRIPTION=$(cat <<EOF
+          **$WORKFLOW** failed on \`$REF_NAME\` via \`$EVENT_NAME\`$ATTEMPT_NOTE
+
+          **Failing jobs:**
+          $FAILED_JOBS
+
+          $COMMIT_LINE
+
+          [View failing run]($RUN_URL)
+          EOF
+          )
+
+          PAYLOAD=$(jq -n \
+            --arg title "$TITLE" \
+            --arg description "$DESCRIPTION" \
+            --arg teamId "$LINEAR_TEAM_ID" \
+            --arg assigneeId "$LINEAR_ASSIGNEE_PETR_ID" \
+            --arg stateId "$LINEAR_STATUS_PLAN_ID" \
+            --arg projectId "$LINEAR_PROJECT_INGESTION_FOR_EVERYTHING_ID" \
+            '{
+              query: "mutation IssueCreate($input: IssueCreateInput!) { issueCreate(input: $input) { success issue { identifier url } } }",
+              variables: {
+                input: {
+                  title: $title,
+                  description: $description,
+                  teamId: $teamId,
+                  assigneeId: $assigneeId,
+                  stateId: $stateId,
+                  projectId: $projectId,
+                  priority: 2
+                }
+              }
+            }')
+
+          curl -sSf -X POST https://api.linear.app/graphql \
+            -H "Authorization: $LINEAR_API_KEY" \
+            -H 'Content-Type: application/json' \
+            -d "$PAYLOAD"

README.md

@@ -1,5 +1,6 @@
 # logtail-lambda-extension
 [![Test logtail-lambda-extension](https://github.com/logtail/logtail-lambda-extension/actions/workflows/test.yml/badge.svg)](https://github.com/logtail/logtail-lambda-extension/actions/workflows/test.yml)
+[![E2E](https://github.com/logtail/logtail-lambda-extension/actions/workflows/e2e.yml/badge.svg)](https://github.com/logtail/logtail-lambda-extension/actions/workflows/e2e.yml)
 
 [AWS Lambda Extension](https://docs.aws.amazon.com/lambda/latest/dg/lambda-extensions.html) that uses the [Lambda Telemetry API](https://docs.aws.amazon.com/lambda/latest/dg/telemetry-api-reference.html) to forward logs to a [Better Stack HTTP source](https://betterstack.com/docs/logs/http-rest-api/).
 
@@ -34,7 +35,7 @@ On next invocation your Lambda logs should start appearing in **Telemetry** -> [
 | RECEIVER_PORT                            | Port of the logs http receiver                          | `4243`                               |
 | MAX_ITEMS                                | Maximum number of events that can be buffered in memory | `10000`                              |
 | MAX_BYTES                                | Maximum size in bytes of events that can be buffered    | `262144`                             |
-| TIMEOUT_MS                               | Maximum time (in milliseconds) that a batch is buffered | `1000`                               |
+| TIMEOUT_MS                               | Maximum time (in milliseconds) that a batch is buffered | `25`                                 |
 | AWS_LAMBDA_RUNTIME_API                   | HTTP base URI to the lambda runtime API                 | _Provided by AWS Lambda environment_ |
 | AWS_LAMBDA_RUNTIME_EXTENSION_API_VERSION | Lambda Extension API version date                       | `2020-01-01`                         |
 | AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION | Telemetry API version date                              | `2022-07-01`                         |

src/aws/events.ts

@@ -24,3 +24,12 @@ export const functionLogEventSchema = z.object({
   time: z.preprocess((value) => (typeof value === 'string' ? new Date(value) : undefined), z.date()), // "2022-10-12T00:03:50.000Z"
   record: z.string(), // "[INFO] Hello world, I am a function!"
 });
+
+// Lambda reports with this event that the runtime finished an invocation; the Telemetry API delivers events in order,
+// so the invocation's function logs have been delivered by then.
+export const platformRuntimeDoneEventSchema = z.object({
+  type: z.literal('platform.runtimeDone'),
+  record: z.object({
+    requestId: z.string(),
+  }),
+});

src/env.test.ts

@@ -23,7 +23,7 @@ describe('parseEnvironmentVariables', () => {
         RECEIVER_ADDRESS: 'sandbox',
         MAX_ITEMS: 10000,
         MAX_BYTES: 262144,
-        TIMEOUT_MS: 1000,
+        TIMEOUT_MS: 25,
         RECEIVER_PORT: 4243,
         sourceToken: 'source_token',
         ingestionUrl: 'https://s123456.lu-trq-1.betterstackdata.com/',

src/env.ts

@@ -18,7 +18,10 @@ const baseSchema = z.object({
   RECEIVER_ADDRESS: z.string().default('sandbox'), // `sandbox` seems to be the default accepted address
   MAX_ITEMS: envNumber.default('10000'), // Maximum number of events that are buffered in memory.
   MAX_BYTES: envNumber.default('262144'), // Maximum size in bytes that the logs are buffered in memory.
-  TIMEOUT_MS: envNumber.default('1000'), // Maximum time (in milliseconds) that a batch is buffered.
+  // Maximum time (in milliseconds) Lambda buffers telemetry before it hands it to the extension; the extension waits for
+  // this hand-off at the end of every invocation, so Lambda's minimum keeps that wait short. Raise it to get fewer, larger
+  // hand-offs in long invocations that log more than 40 lines per second, at that much more time per invocation.
+  TIMEOUT_MS: envNumber.default('25'),
   RECEIVER_PORT: envNumber.default('4243'), // HTTP server receiving port
   // Better Stack source token; SOURCE_TOKEN is preferred, LOGTAIL_TOKEN is kept as a fallback.
   SOURCE_TOKEN: z.string().optional(),

src/extension.test.ts

@@ -0,0 +1,60 @@
+import { forwardInvocationLogs, telemetrySubscription } from '~/extension';
+
+describe('forwardInvocationLogs', () => {
+  test('forwards the logs only after the runtime finished the invocation', async () => {
+    let finishRuntime = () => undefined as void;
+    const runtimeDone = jest.fn(
+      () =>
+        new Promise<void>((resolve) => {
+          finishRuntime = resolve;
+        }),
+    );
+    const forwardLogs = jest.fn(() => Promise.resolve());
+
+    const forwarded = forwardInvocationLogs({ runtimeDone }, forwardLogs, {
+      requestId: 'request-1',
+      deadlineMs: Date.now() + 60_000,
+    });
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(runtimeDone).toHaveBeenCalledWith('request-1');
+    expect(forwardLogs).not.toHaveBeenCalled();
+
+    finishRuntime();
+    await forwarded;
+    expect(forwardLogs).toHaveBeenCalledTimes(1);
+  });
+
+  test('forwards the logs anyway, well before the deadline, when the runtime never reports', async () => {
+    const forwardLogs = jest.fn(() => Promise.resolve());
+    const start = Date.now();
+
+    await forwardInvocationLogs({ runtimeDone: () => new Promise<void>(() => undefined) }, forwardLogs, {
+      requestId: 'request-1',
+      deadlineMs: start + 1_200,
+    });
+
+    expect(forwardLogs).toHaveBeenCalledTimes(1);
+    expect(Date.now() - start).toBeLessThan(1_000);
+  });
+});
+
+describe('telemetrySubscription', () => {
+  test('subscribes to platform events, for platform.runtimeDone, along with the function logs', () => {
+    const subscription = telemetrySubscription({
+      AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION: '2022-07-01',
+      RECEIVER_ADDRESS: 'sandbox',
+      RECEIVER_PORT: 4243,
+      TIMEOUT_MS: 25,
+      MAX_BYTES: 262144,
+      MAX_ITEMS: 10000,
+    });
+
+    expect(subscription).toStrictEqual({
+      schemaVersion: '2022-07-01',
+      destination: { protocol: 'HTTP', URI: 'http://sandbox:4243' },
+      types: ['platform', 'function'],
+      buffering: { timeoutMs: 25, maxBytes: 262144, maxItems: 10000 },
+    });
+  });
+});

src/extension.ts

@@ -1,10 +1,62 @@
 import { function as F, taskEither as TE } from 'fp-ts';
 import { EnvironmentVars } from '~/env';
 import { pollForNextEvent, registerExtension } from '~/aws/api';
-import { startTelemetryHttpListener } from '~/httpListener';
+import { startTelemetryHttpListener, TelemetryHttpListener } from '~/httpListener';
 import { subscribeTelemetry, SubscriptionBody } from '~/aws/subscribe';
-import { betterStackLogForwarder } from '~/forwarders/betterStack';
+import { betterStackLogForwarder, REQUEST_TIMEOUT_MS } from '~/forwarders/betterStack';
 import { AbortHandler } from '~/abortHandler';
+import { InvokeEvent } from '~/aws/events';
+
+export const telemetrySubscription = ({
+  AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION,
+  RECEIVER_ADDRESS,
+  RECEIVER_PORT,
+  TIMEOUT_MS,
+  MAX_BYTES,
+  MAX_ITEMS,
+}: Pick<
+  EnvironmentVars,
+  | 'AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION'
+  | 'RECEIVER_ADDRESS'
+  | 'RECEIVER_PORT'
+  | 'TIMEOUT_MS'
+  | 'MAX_BYTES'
+  | 'MAX_ITEMS'
+>): SubscriptionBody => ({
+  schemaVersion: AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION,
+  destination: {
+    protocol: 'HTTP',
+    URI: `http://${RECEIVER_ADDRESS}:${RECEIVER_PORT}`,
+  },
+  // Platform events for platform.runtimeDone, see forwardInvocationLogs
+  types: ['platform', 'function'],
+  buffering: {
+    timeoutMs: TIMEOUT_MS,
+    maxBytes: MAX_BYTES,
+    maxItems: MAX_ITEMS,
+  },
+});
+
+// Lambda freezes the execution environment as soon as the runtime and every extension asked for the next event, and
+// the Telemetry API hands over an invocation's logs only after its runtime finished. So the extension waits for
+// platform.runtimeDone of the invocation before it forwards and asks for the next event; forwarding right away would
+// send the logs one or two invocations later, or never for the last invocation before a shutdown.
+export const forwardInvocationLogs = (
+  listener: Pick<TelemetryHttpListener, 'runtimeDone'>,
+  forwardLogs: () => Promise<void>,
+  event: Pick<InvokeEvent, 'requestId' | 'deadlineMs'>,
+): Promise<void> => {
+  let timeout: NodeJS.Timeout | undefined;
+  const beforeDeadline = new Promise<void>((resolve) => {
+    // When the runtime has not reported that it finished, forward anyway early enough for the request to time out
+    // before the invocation's deadline, so the extension never makes the invocation time out.
+    timeout = setTimeout(resolve, Math.max(0, event.deadlineMs - Date.now() - REQUEST_TIMEOUT_MS));
+  });
+
+  return Promise.race([listener.runtimeDone(event.requestId), beforeDeadline])
+    .finally(() => clearTimeout(timeout))
+    .then(forwardLogs);
+};
 
 // noinspection HttpUrlsUsage
 export const createExtension = (
@@ -24,34 +76,7 @@ export const createExtension = (
     TE.bind('listener', ({ EXTENSION_NAME, RECEIVER_ADDRESS, RECEIVER_PORT }) =>
       startTelemetryHttpListener(EXTENSION_NAME, RECEIVER_ADDRESS, RECEIVER_PORT),
     ),
-    TE.chainFirst(
-      ({
-        telemetryBaseUrl,
-        extensionId,
-        RECEIVER_ADDRESS,
-        RECEIVER_PORT,
-        AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION,
-        TIMEOUT_MS,
-        MAX_BYTES,
-        MAX_ITEMS,
-      }) =>
-        F.pipe(
-          <SubscriptionBody>{
-            schemaVersion: AWS_LAMBDA_RUNTIME_TELEMETRY_API_VERSION,
-            destination: {
-              protocol: 'HTTP',
-              URI: `http://${RECEIVER_ADDRESS}:${RECEIVER_PORT}`,
-            },
-            types: ['function'],
-            buffering: {
-              timeoutMs: TIMEOUT_MS,
-              maxBytes: MAX_BYTES,
-              maxItems: MAX_ITEMS,
-            },
-          },
-          (body) => subscribeTelemetry(telemetryBaseUrl, extensionId, body),
-        ),
-    ),
+    TE.chainFirst((env) => subscribeTelemetry(env.telemetryBaseUrl, env.extensionId, telemetrySubscription(env))),
     TE.bind('forwardLogs', ({ sourceToken, ingestionUrl, listener }) =>
       TE.of(betterStackLogForwarder(sourceToken, ingestionUrl, listener)),
     ),
@@ -62,12 +87,12 @@ export const createExtension = (
           TE.tryCatch(
             () => {
               switch (event.eventType) {
-                // Received when the Lambda first invokes
+                // Received when the Lambda invokes the function
                 case 'INVOKE':
                   console.log(`[${EXTENSION_NAME}] Received INVOKE event`);
 
-                  // Forward any logs pending from the last execution
-                  return forwardLogs();
+                  // Forward the invocation's logs once its runtime finished, before Lambda freezes the environment
+                  return forwardInvocationLogs(listener, forwardLogs, event);
                 // Received when the runtime asks us to shut down (i.e. no lambdas running in a set time-frame)
                 case 'SHUTDOWN':
                 default:

src/forwarders/betterStack.test.ts

@@ -71,6 +71,25 @@ describe('test Better Stack log forwarding', () => {
     );
     expect(listener.logsQueue.length).toBe(3);
   });
+
+  test('forwarder should give up on a request Better Stack does not answer in time and re-queue the logs', async () => {
+    fetchMock.mockResponse(
+      (request: Request) =>
+        new Promise((_resolve, reject) => {
+          request.signal?.addEventListener('abort', () => reject(new Error('The operation was aborted.')));
+        }),
+    );
+
+    const listener: { logsQueue: FunctionLogEvent[] } = {
+      logsQueue: [log, log],
+    };
+    const start = Date.now();
+
+    await betterStackLogForwarder(token, ingestionUrl, listener, 100)();
+
+    expect(Date.now() - start).toBeLessThan(1_000);
+    expect(listener.logsQueue.length).toBe(2);
+  });
 });
 
 describe('test parseMessageWithPowertoolsLogFormat`', () => {

src/forwarders/betterStack.ts

@@ -35,8 +35,13 @@ export const parseMessageWithPowertoolsLogFormat = (message: string): E.Either<E
     }),
   );
 
+// How long a request to Better Stack may take before the extension gives up and keeps the logs queued for the next
+// forward, so an unresponsive endpoint holds an invocation for at most this long.
+export const REQUEST_TIMEOUT_MS = 1000;
+
 export const betterStackLogForwarder =
-  (token: string, ingestionUrl: string, listener: { logsQueue: FunctionLogEvent[] }) => (): Promise<void> => {
+  (token: string, ingestionUrl: string, listener: { logsQueue: FunctionLogEvent[] }, timeoutMs = REQUEST_TIMEOUT_MS) =>
+  (): Promise<void> => {
     const logs = listener.logsQueue.splice(0);
 
     if (logs.length === 0) {
@@ -71,6 +76,7 @@ export const betterStackLogForwarder =
         'Content-Type': 'application/json',
         Authorization: `Bearer ${token}`,
       },
+      signal: AbortSignal.timeout(timeoutMs),
     })
       .then(async (response) => {
         if (!response.ok) {