This gist contains the handler function code from lambda-function.py, which is part of the blog-amazon-macie-custom-eventbridge-events repository.

blog-amazon-macie-custom-eventbridge-eventbridge_lambda-function-handler_function.py

def lambda_handler(event, context):
    """
    Main handler function.

    Parameters:
        event (dict): The CloudWatch Logs event containing base64 encoded,
                      gzip compressed log data
        context: Lambda context object

    Returns:
        dict: Response containing the status and number of events published
    """
    logger.info("Received event from CloudWatch Logs Subscription Filter")

    # Step 1: Decode and decompress the CloudWatch Logs data
    log_data = decode_cloudwatch_logs_data(event)

    if log_data is None:
        logger.error("Failed to decode CloudWatch Logs data")
        return {
            "statusCode": 500,
            "body": "Failed to decode CloudWatch Logs data"
        }

    logger.info(
        f"Processing log group: {log_data.get('logGroup')}, "
        f"log stream: {log_data.get('logStream')}, "
        f"number of log events: {len(log_data.get('logEvents', []))}"
    )

    # Step 2: Process each log event and publish to EventBridge
    events_published = 0
    events_failed = 0

    for log_event in log_data.get("logEvents", []):
        try:
            # Parse the job status event from the log message
            job_status_event = parse_job_status_event(log_event)

            if job_status_event:
                # Publish the custom event to EventBridge
                success = publish_to_eventbridge(job_status_event)
                if success:
                    events_published += 1
                    logger.info(
                        f"Successfully published event: "
                        f"eventType={job_status_event.get('eventType')}, "
                        f"jobId={job_status_event.get('jobId')}"
                    )
                else:
                    events_failed += 1
            else:
                logger.warning(
                    f"Could not parse job status from log event: "
                    f"{log_event.get('message', '')[:200]}"
                )
        except Exception as e:
            events_failed += 1
            logger.error(f"Error processing log event: {str(e)}")

    logger.info(
        f"Processing complete. Events published: {events_published}, "
        f"Events failed: {events_failed}"
    )

    return {
        "statusCode": 200,
        "body": json.dumps({
            "events_published": events_published,
            "events_failed": events_failed
        })
    }