creating raw AWS log event
Peter Otten
__peter__ at web.de
Thu Jun 24 03:19:14 EDT 2021
On 23/06/2021 19:42, Larry Martell wrote:
> When an AWS cloudwatch event is passed to a consumer it looks like this:
>
> {
> "awslogs": {
> "data": "ewogICAgIm1l..."
> }
> }
>
> To get the actual message I do this:
>
> def _decode(data):
> compressed_payload = b64decode(data)
> json_payload = zlib.decompress(compressed_payload, 16+zlib.MAX_WBITS)
> return json.loads(json_payload)
>
> message = _decode(json.dumps(event['awslogs']['data']))
>
> This returns the log message as a string.
>
> For my unit tests I need to reverse this - given a message as a string
> I want to generate the compressed, encoded event structure.
>
> I have not been able to get this to work. I have this:
>
> message = b'test message'
> compressed= zlib.compress(message)
> event['awslogs']['data'] = str(compressed)
>
> message = _decode(json.dumps(event['awslogs']['data']))
> Traceback (most recent call last):
> File "<stdin>", line 1, in <module>
> File "<stdin>", line 3, in _decode
> zlib.error: Error -3 while decompressing data: incorrect header check
>
> Anyone see how to make this work?
The json/bas64 parts are not involved in the problem:
>>> zlib.decompress(zlib.compress(b"foo"), 16 + zlib.MAX_WBITS)
Traceback (most recent call last):
File "<pyshell#4>", line 1, in <module>
zlib.decompress(zlib.compress(b"foo"), 16 + zlib.MAX_WBITS)
zlib.error: Error -3 while decompressing data: incorrect header check
whereas:
>>> zlib.decompress(zlib.compress(b"foo"))
b'foo'
Unfortunately compress() doesn't accept the flags you seem to require.
However, reading around a bit in the zlib docs turns up the compressobj
which does. So
>>> def mycompress(data):
obj = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)
result = obj.compress(data)
result += obj.flush()
return result
>>> zlib.decompress(mycompress(b"foo"), 16 + zlib.MAX_WBITS)
b'foo'
More information about the Python-list
mailing list