Skip to content

Instantly share code, notes, and snippets.

Created May 2, 2022 16:08
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save hassaku63/a7d6a3a870216f6bb5e5042831ba9603 to your computer and use it in GitHub Desktop.
example lambda (python) handler processing subscribed CloudWatch Logs log event (via kinesis data stream)
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
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
ret = do_something(payload, context)
except Exception as e:
raise e
return result
def do_something(log_events, context):
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']{
'timestamp': timestamp,
'error_name': error_name,
'cause': {
'err_type': err_type,
'err_msg': err_msg,
'stack': stack_trace
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment