How to decode CloudWatch Logs events

decode.js

const zlib = require('zlib');

exports.decode = async (data) => {
    const compressedPayload = Buffer.from(data, 'base64');
    const jsonPayload = zlib.gunzipSync(compressedPayload).toString('utf8');
    return JSON.parse(jsonPayload);
}

decode.py

import zlib
import json
from base64 import b64decode

def decode(data):
    compressed_payload = b64decode(data)
    json_payload = zlib.decompress(compressed_payload, 16+zlib.MAX_WBITS)
    return json.loads(json_payload)

Thanks for this Alex. Do you have a function that is the inverse of this? One that takes a dict and creates the compressed encoded payload? I need something like that for my unit tests.

@LarryMartell it should be fairly easy to invert the semantics and order of all operations :)

Something like this (untested):

import zlib
import json
from base64 import b64encode

def encode(data):
    json_payload = json.dumps(data)
    compressed_payload = zlib.compress(json_payload)
    return b64encode(compressed_payload)   

Thanks for the reply. I ended up having to do this to get it to work:

def encode(data):
    json_payload = json.dumps(data)
    compressed_payload = mycompress(bytes(json_payload, 'utf-8'))
    return b64encode(compressed_payload)


def mycompress(data):
    obj = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)
    result = obj.compress(data)
    result += obj.flush()
    return result