example lambda (python) handler processing subscribed CloudWatch Logs log event (via kinesis data stream)

handler.py

import json
import gzip
import logging
from base64 import b64decode


log = logging.getLogger(__name__)


def handler(event, context):
    """example handler

    :param event: kinesis records (payload is subscribed cloudwatch logs log event)
    :type event: dict
    :param context: lambda context
    :type context: LambdaContext
    """
    log.info(event)
    result = []
    for record in event['Records']:
        encoded = record['kinesis']['data']
        payload = gzip.decompress(b64decode(encoded))
        payload = json.loads(payload)  # payload is cw logs log event
        try:
            ret = do_something(payload, context)
        except Exception as e:
            log.exception(str(e))
            raise e

        result.append(ret)

    return result


def do_something(log_events, context):
    log.info(log_events)
    log_group = log_events['logGroup']
    log_stream = log_events['logStream']

    for event in log_events['logEvents']:
        timestamp = event['timestamp']  # in milli-sec
        data = json.loads(event['message'])
        # _id = data['id']
        # _type = data['type']
        details: dict = data['details']
        error_name: str = details['error']  # built-in error or user defined exception
        error_cause: dict = json.loads(details['cause'])
        err_msg: str = error_cause['errorMessage']
        err_type: str = error_cause['errorType']
        stack_trace: list[str] = error_cause['stackTrace']
        log.info({
            'timestamp': timestamp,
            'error_name': error_name,
            'cause': {
                'err_type': err_type,
                'err_msg': err_msg,
                'stack': stack_trace
            }
        })