Log Processing Lambda for Kinesis Firehose

Build a Lambda function that transforms log records from Kinesis Data Firehose by filtering, enriching, and formatting them before delivery to S3.

Requirements

Your Lambda function should:

1. Filter logs by severity: Only process records with log level "ERROR" or "WARN". Drop all other records.

2. Parse and validate: Parse each record as JSON. If parsing fails, mark the record as processing failed.

3. Enrich records: Add a processed_at timestamp field (ISO 8601 format) to each valid record.

4. Format output: Return transformed records as JSON strings, properly base64-encoded.

5. Handle status codes: Return appropriate result codes:

   • "Ok" for successfully transformed records
   • "Dropped" for records that don't meet the filter criteria
   • "ProcessingFailed" for records that can't be parsed

Input Format

Each incoming record contains:

• recordId: Unique identifier for the record
• data: Base64-encoded JSON string with fields:
  • level: Log level (DEBUG, INFO, WARN, ERROR)
  • message: Log message text
  • timestamp: Original log timestamp

Output Format

Return a response with transformed records containing:

• recordId: Original record ID
• result: Status code (Ok, Dropped, or ProcessingFailed)
• data: Base64-encoded transformed JSON (for Ok status)

Implementation

@generates

API

def lambda_handler(event: dict, context: Any) -> dict:
    """
    Transform Kinesis Firehose records.

    Args:
        event: Firehose transformation event
        context: Lambda context

    Returns:
        Firehose transformation response with records
    """

Test Cases

• It successfully transforms ERROR level logs @test
• It successfully transforms WARN level logs @test
• It drops INFO level logs @test
• It drops DEBUG level logs @test
• It handles invalid JSON with ProcessingFailed @test
• It adds processed_at timestamp to transformed records @test

Dependencies { .dependencies }

aws-lambda-powertools { .dependency }

Provides event handling and data transformation utilities for AWS Lambda.