Skip to content

Instantly share code, notes, and snippets.

@hassaku63
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
"""
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
}
})
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment