Feature request: Data Class for Cloudwatch Logs passed through Kinesis Stream

[blewinter]

Use case

We are currently passing Cloudwatch logs to a Kinesis stream to be subsequently processed by a Lambda. We'd like to use the cloud_watch_logs_event data class but it only works when the source is Cloudwatch directly, not Kinesis. It seems when Cloudwatch targets Lambda, it wraps the payload in the following superstructure:
'awslogs': {'data': {<PAYLOAD>}}

You can see that the existing CloudWatchLogsEvent class' raw_logs_data property expects this structure to unpack the data.

This superstructure doesn't show up when the logs are passed through kinesis

Solution/User Experience

I think a new method/property within the CloudWatchLogsEvent class, perhaps named parse_logs_kinesis() or something, which can be used to unpack this type of payload, could work.

However that may be less than ideal since accidentally using the existing raw_logs_data property would then cause issues, so it's possible that an entirely new class (like CloudWatchLogsKinesisEvent, ex) would be preferable.

A third option would be to rewrite the existing raw_logs_data property, such that it checks for the presence of the Cloudwatch->Lambda superstructure:
return self['awslogs']['data'] if self.get('awslogs') else self[0]
(safe retrieval from nested dicts can get a bit klunky)

Note: Originally posted in discussions forum here.

Alternative solutions

No response

Acknowledgment

• This feature request meets Lambda Powertools Tenets
• Should this be considered in other Lambda Powertools languages? i.e. Java, TypeScript

[boring-cyborg]

Thanks for opening your first issue here! We'll come back to you as soon as we can.

[leandrodamascena]

Hey @blewinter thanks for following up with a feature request on this - would you have the bandwidth to contribute it? We can offer guidance sync or async on our Discord channel.

If you don't, we can ping you to review a PR once we clear other activities.

🚀

[blewinter]

Hi! Apologies, I wasn't receiving notifications on this thread. I'd love to contribute! Will just need a bit of time to get set up. Do you have thoughts on what the best approach might be, or should I post on Discord?

[leandrodamascena]

However that may be less than ideal since accidentally using the existing raw_logs_data property would then cause issues, so it's possible that an entirely new class (like CloudWatchLogsKinesisEvent, ex) would be preferable.

Hi @blewinter! We think this is the best and safest way to implement this. A new class to handle events coming from Kinesis is a great idea.

We look forward to your contribution. If you have any questions or want help testing something, please let me know and I'll be available for that. 🤖

[leandrodamascena]

Hi @blewinter! I hope things are going well there.

We are planning to include this change in the next release, which I think will be in 2 or 3 weeks. But it's okay to move this change to another release if you need more time to implement this or we can even implement this and ping it for you to review before the merge. We'd love to hear the best option for you 👂.

Thank you

[blewinter]

Hey, I had to put it aside for a bit but I might be able to have it by the end of next week! I will keep you posted

[blewinter]

Hi,

So I had played around with the idea of a separate class for this feature, but I think a simple conversion function might be preferable? Of course can wrap in a class. Lots of ways to implement this. What do you think of this approach?

https://github.com/blewinter/aws-lambda-powertools-python/blob/feat/cloudwatch-kinesis-log/aws_lambda_powertools/utilities/data_classes/cloud_watch_kinesis_log.py