Processing Cloudwatch Logs from Kinesis Streams

[blewinter]

Hi,
We are pushing cloudwatch logs to a subscription filter, and setting the subscription filter's target to a kinesis stream. When subsequently processing those logs using Lambda, we'd like to use the cloud_watch_logs_event data class, but it seems we're unable to because the log payloads are not encapsulated with '{awslogs': { 'data': }}`. It appears that encapsulation only happens when the subscription filters target is lambda directly. Am I understanding this correctly? Would it be appropriate to put in a feature request to be able to handle Cloudwatch Logs that don't have that encapsulation?

[heitorlessa]

Yes! Please open up a feature request. We’d also appreciate if you have a sample event (no sensitive data) ;-) Thank you!!

…

On Wed, 24 Aug 2022 at 20:20, blewinter ***@***.***> wrote: Hi, We are pushing cloudwatch logs to a subscription filter, and setting the subscription filter's target to a kinesis stream. When subsequently processing those logs using Lambda, we'd like to use the cloud_watch_logs_event data class, but it seems we're unable to because the log payloads are not encapsulated with '{awslogs': { 'data': }}`. It appears that encapsulation only happens when the subscription filters target is lambda directly. Am I understanding this correctly? Would it be appropriate to put in a feature request to be able to handle Cloudwatch Logs that don't have that encapsulation? — Reply to this email directly, view it on GitHub <#1476>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZPQBEZ3IVBD672TQDKU6TV2ZRWJANCNFSM57QJYMXA> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.*** com>

[leandrodamascena]

Hi all! I saw this conversation and remembered a few months ago I sent logs from Cloudwatch to an OnPrem solution and the best and resilient way was using a Cloudwatch subscription to a Kinesis Stream and consuming this stream on the OnPrem solution.

Actually you're right @blewinter, Cloudwatch subscription doesn't wrap the log/stream with {awslogs': { 'data': }} when the target is a Kinesis, this sends the message in raw format. After decoded, the Kinesis record looks like the code below and you cannot parse this using the cloud_watch_logs_event data class.

[{
    "messageType": "DATA_MESSAGE",
    "owner": "123456789012",
    "logGroup": "/aws/lambda/lambda-url",
    "logStream": "2022/08/25/[$LATEST]a8dbdd664e5846a8bf66bc0c528e907f",
    "subscriptionFilters": [
        "lambdapt"
    ],
    "logEvents": [
        {
            "id": "37052029857236301225838331276741826638572548226118057984",
            "timestamp": 1661470481250,
            "message": "START RequestId: 76119103-f86b-4a37-bdb5-af82858d0179 Version: $LATEST\n",
            "extractedFields": {
                "message": "START RequestId: 76119103-f86b-4a37-bdb5-af82858d0179 Version: $LATEST\n"
            }
        },
        {
            "id": "37052029857258601971036861899883362356845196587624038401",
            "timestamp": 1661470481251,
            "message": "END RequestId: 76119103-f86b-4a37-bdb5-af82858d0179 Version: $LATEST\n",
            "extractedFields": {
                "message": "END RequestId: 76119103-f86b-4a37-bdb5-af82858d0179 Version: $LATEST\n"
            }
        }
    ]
}]

For testing purposes only (this is probably not the best solution), I created in my local environment a new method to return the raw event as a CloudWatchLogsDecodedData class and I could parse the logs using the cloud_watch_logs_event data class.

New property in aws_lambda_powertools/utilities/data_classes/cloud_watch_logs_event.py file:

    @property
    def parse_logs_raw(self) -> CloudWatchLogsDecodedData:
        """Return the raw payload"""
        return CloudWatchLogsDecodedData(self[0])

Lambda function:

from aws_lambda_powertools.utilities.data_classes import event_source, CloudWatchLogsEvent
from aws_lambda_powertools.utilities.data_classes.cloud_watch_logs_event import CloudWatchLogsDecodedData

@event_source(data_class=CloudWatchLogsEvent)
def lambda_handler(event: CloudWatchLogsEvent, context):

    kinesis_logs: CloudWatchLogsDecodedData = event.parse_logs_raw
    log_events = kinesis_logs.log_events
    for event in log_events:
        print(f"LOG ID -> {event['id']}")

Execution:

What do you think opening a feature request to implement this as a first contributor?

Thank you and I hope this information can help in this discussion.

[blewinter]

Thanks for the input! I've submitted the feature request here: #1488

I'm honestly not sure what the most elegant solution is. @leandrodamascena your solution is definitely nice, but I can see it being confusing that the original raw_logs_data property is still present, which would throw an exception when used on a Cloudwatch->Kinesis log.

[heitorlessa]

Closing discussion as this was released in https://github.com/awslabs/aws-lambda-powertools-python/releases/tag/v2.3.0