This lambda function is deployed via cloud formation template. It is composed of the following parts:

• A yaml config file that is sourced from an s3 bucket at time of execution. It includes instructions on how to build job rules to extend the function's capacity.
• A library of filtering, parsing, and messaging functions
• A simple lambda_function handler that is triggered by SQS messages posted when a log file is posted to an s3 bucket.
• cloudformation template

 The function handles 10 messages at a time, filters out files we don't want, identifies file types we do want, wraps them as syslog messages, then compresses the file and posts it to a bucket, which will get pulled on premise by a local worker script.

The yaml config file

# This document is an abstraction of the pattern match and job function logic of the aws_logs_filter.py lambda function.
#
# The 'job' rules define how the aws-logs-filter.py function will parse logs.
#
# Each job is a log type that is received from AWS accounts that have implemented log aggregation and forwarding
#
# Each log is tested for gzip compression, unzipped if necessary, and loaded into memory. All files are treated as
# plain text files for the purposes of job definition.
#
# The data structure is as follows:
# job##: a simple name for the rule. Must be unique, but not necessarily sequential
# name: a plain language name for the job that is not processed by the function
# match_variable: a pattern that the function will use to identify the log
# service_variable: a service that will be added to the syslog message to determine the log type.
# action variables: an ordered list of steps the function will take to process the log.
#
# The callable methods work as follows:
#   json: treats the file as a json object on each line
#   firehose: json objects on a single line. The method looks for touching braces "}{" and breaks them into lines "}\n{"
#   text_wrap: handles text logs regardless of form, wraps them as a syslog message
#
# all logs that meet the job rules get parsed, gzipped and sent to the filtered s3 bucket.
# all logs that fail to get parsed are written to the region's DynamoDB table, AWSLogsFilterExceptions
#
# Periodic review of the table will show logs that are not being processed, and a new set of job rules can be
# added to this file. The configuration will be updated on the next execution of the lambda function.
#
# Periodic review of the dead letter queue will also be useful to identify logs that are not being processed, but these
# be a rare exception. We will need to ascertain why the log was not written to the database table.
#
# This file is kept in the s4://<redacted> bucket, which supports versioning. If a change is made that
# breaks the lambda functions, you can restore the previous version using the AWS S3 management console.
#
jobs:
    job01 :
        name: cloudtrail
        match_variable: cloudtrail/
        service_variable: 'cloudtrail.amazonaws.com'
        action_variables:
           - json

    job02 :
        name: rds
        match_variable: rds/
        service_variable: 'rds.amazonaws.com'
        action_variables:
           - text_wrap

    job03 :
        name: firehose
        match_variable: firehose/
        service_variable: 'cloudwatch.amazonaws.com'
        action_variables:
           - firehose
           - json

    job04 :
        name: config
        match_variable: config/
        service_variable: 'config.amazonaws.com'
        action_variables:
           - json

    job05 :
        name: flowlog
        match_variable: flowlog/
        service_variable: 'vpc.amazonaws.com'
        action_variables:
           - text_wrap

    job06 :
        name: elb
        match_variable: elasticloadbalancing/
        service_variable: 'elb.amazonaws.com'
        action_variables:
           - text_wrap

    job07 :
        name: serveraccesslogs
        match_variable: serveraccesslogs/
        service_variable: 'ec2.amazonaws.com'
        action_variables:
           - text_wrap

    job08 :
        name: mq
        match_variable: mq/
        service_variable: 'mq.amazonaws.com'
        action_variables:
           - text_wrap

    job09 :
        name: redshift
        match_variable: redshift/
        service_variable: 'redshift.amazonaws.com'
        action_variables:
           - text_wrap

    job10 :
        name: cloudfront
        match_variable: cloudfront/
        service_variable: 'cloudfront.amazonaws.com'
        action_variables:
           - text_wrap

    job11 :
        name: ec2
        match_variable: ec2/
        service_variable: 'ec2.amazonaws.com'
        action_variables:
           - json

    job12 :
        name: emr
        match_variable: elasticmapreduce/
        service_variable: 'elasticmapreduce.amazonaws.com'
        action_variables:
           - text_wrap

    job13 :
        name: elb2
        match_variable: elb/
        service_variable: 'elb.amazonaws.com'
        action_variables:
           - text_wrap

    job14 :
        name: etl
        match_variable: etl/
        service_variable: 'glue.amazonaws.com'
        action_variables:
           - text_wrap

    job15 :
        name: s3
        match_variable: s3/
        service_variable: 's3.amazonaws.com'
        action_variables:
           - text_wrap

    job16:
        name: aurora
        match_variable: aurora/
        service_variable: 'rds.amazonaws.com'
        action_variables:
        - text_wrap

# filter variable. These are key path patterns that we drop to avoid redundant logs and files that lock up the parser
FILTERS:
    - CloudTrail-Digest
    - ConfigWritabilityCheckFile
    - DS_Store
    - .trm
    - .trc
    - <example key path>
    - <rest removed for privacy>

# aggregate_accounts variable for those that combine their logs in one account before replication. Note in key path where
# actual account number shows, split by '/'
aggregate_accounts:
    agg_acct01:
        name: <redacted1>
        agg_account: '<example1>'
        match_position: '1'
    agg_acct02:
        name: <redacted2>
        agg_account: '<redacted2>'
        match_position: '2'
    agg_acct03:
        name: <redacted3>
        agg_account: '<redacted3>'
        match_position: '2'

The Library File

#!/usr/local/bin/python3
"""
DESCRIPTION
Lambda function to process sqs events, filter messages based on key paths,
load filtered messages into memory,parse formats to syslog,
and write syslog messages to a gzipped s3 object, and drop the event generating sqs message.
"""

import re
import ast
import gzip
import json
from datetime import datetime
import urllib
import boto3
import botocore
import os
import yaml
import math
import uuid


REGION = os.environ.get('AWS_REGION')


# setup vars
config_bucket = '<redacted>'
yaml_config = 'config.yaml'
s3 = boto3.client('s3')
config_handle = s3.get_object(Bucket=config_bucket, Key=yaml_config)
config_data = body = config_handle.get('Body').read()
yaml_data = yaml.load(config_data)
FILTERS = yaml_data['FILTERS']
aggregate_accounts = yaml_data['aggregate_accounts']


class Message:
    """ Pull objects and variables plus event data. """

    def __init__(self, sqs_object, record, in_queue, out_queue):
        """ receives a single record from event. """
        self.sqs = sqs_object
        self.message_id = record["messageId"]
        self.receipt_handle = record["receiptHandle"]
        # not in testing
        body = ast.literal_eval(record["body"])
        self.msg_key_path = body["Records"][0]["s3"]["object"]["key"]
        # self.msg_key_path = record["body"]["Records"]["s3"]["object"]["key"]
        self.in_queue = in_queue
        self.out_queue = out_queue
        self.tag = 'message: initialized'
        #print('key: ', self.msg_key_path)

    def filter_message(self):
        """
        filters keypath against list to drop files we don't want
        """
        return_value = ''
        global FILTERS
        for my_filter in FILTERS:
            if re.search(my_filter, self.msg_key_path):
                #print('matched: ', self.msg_key_path)
                self.drop_message()
                self.tag = 'message: filtered'
                return_value = True
        if return_value:
            return True
        else:
            #print('file not matched: ', self.msg_key_path)
            return False

    def drop_message(self):
        """
        removes sqs msg from queue after successful file processing
        """
        try:
            self.sqs.delete_message(
                QueueUrl=self.in_queue,
                ReceiptHandle=self.receipt_handle
            )
            return True

        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not drop message:', exception_string)
            return False

    def write_sqs_message_to_out_queue(self, sqs_object, out_queue, data):
        """
        Writes a message to specified / dlq queue. Have to replace single quotes to meet json formatting
        """

        try:
            replace_single_quotes = re.sub('\'', '\"', data)
            sqs = sqs_object
            sqs.send_message(
                QueueUrl=out_queue,
                MessageBody=replace_single_quotes,
            )
            return True

        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not write message to out_queue:', exception_string)
            return False


class File:

    """ Processes s3 object from msg_key_path. """
    def __init__(self, s3_object, msg_key_path, in_bucket, out_bucket):
        self.s3 = s3_object
        self.msg_key_path = msg_key_path
        self.in_bucket = in_bucket
        self.out_bucket = out_bucket
        self.service_name = ''
        self.owner_id = ''
        self.tag = 'file: initialized'

# check these
    def parse_body_json_to_syslog(self):
        output_data = ''
        accountID = 'account-' + self.owner_id
        my_time = datetime.now().strftime('%Y-%m-%dT%H:%M:%S%Z')
        service_name = self.service_name
        my_data = str(self.body).split('\n')
        self.datetime = my_time

        for data in my_data:
            my_data = json.loads(data)
            message = my_time + ' ' + str(accountID) + ' ' + str(service_name) + ' ' + str(my_data) + '\n'
            output_data += message
        self.body = output_data
        return True

    def determine_if_agg_account(self):
        for key in aggregate_accounts.keys():
            match = re.search(self.msg_key_path, aggregate_accounts[key]['agg_account'])
            split_path = self.msg_key_path.split('/')

            if match:
                self.owner_id = split_path[int(aggregate_accounts[key]['match_position'])]
            else:
                self.owner_id = split_path[0]
        return self.owner_id

    def determine_service_name(self, service_name):
        self.service_name = service_name
        return True

    def fix_body_firehose(self):
        data = re.sub('}{', '}\n{', self.body)
        self.tag = 'body: each line json'
        self.body = data
        return True

    def parse_ascii_body_to_syslog(self):
        output_data = ''
        accountID = 'account-' + self.owner_id
        my_time = datetime.now().strftime('%Y-%m-%dT%H:%M:%S%Z')
        service_name = self.service_name
        self.datetime = my_time
        output_data = my_time + ' ' + str(accountID) + ' ' + str(service_name) + ' ' + str(self.body)
        self.body = output_data
        return True

# these are reliable
    def load_s3_object_to_body(self):
        """
       Loads the file's contents to memory for manipulation
        """
        self.msg_key_path = urllib.parse.unquote(self.msg_key_path)
        file_handle = self.s3.get_object(Bucket=self.in_bucket, Key=self.msg_key_path)
        try:
            self.body = file_handle.get('Body').read()
            self.tag = 'body: loaded'
            return True
        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not load file from path: ', self.msg_key_path, ' ', exception_string)
            return False

    def put_s3_object(self):
        """ Write loaded data to filtered bucket. """
        try:
            self.s3.put_object(Bucket=self.out_bucket, Key=self.msg_key_path, Body=self.body)
            return True
        except:
            print('could not put: ', self.msg_key_path)
            return False

    def check_body_is_gzip(self):
        try:
            test_gzip = gzip.decompress(self.body)
            self.tag = 'body: gzip'
            return True
        except:
            return False

    def check_body_is_ascii(self):
        try:
            data = self.body.decode()
            test_ascii = isinstance(data, data)
            if test_ascii:
                self.tag = 'body: ascii'
                return True
        except:
            return False

    def gunzip_body(self):
        data = gzip.decompress(self.body)
        self.body = data.decode('utf-8')
        self.tag = 'body: unzipped'
        return True

    def gzip_body(self):
        data = self.body
        encoded_data = data.encode('utf-8')
        self.body = gzip.compress(encoded_data)
        self.tag = 'body: compressed to gz'
        return True

    def add_syslog_suffix_to_path(self):
        self.msg_key_path = self.msg_key_path + '.syslog'
        self.tag = 'file: .syslog'
        return True

    def add_gz_suffix_to_path(self):
        self.tag = 'file: .gz'
        self.msg_key_path = self.msg_key_path + '.gz'
        return True

    def strip_path_of_suffix(self):
        try:
            self.msg_key_path = re.sub('.gz', '', self.msg_key_path)
            self.msg_key_path = re.sub('.json', '', self.msg_key_path)
            self.msg_key_path = re.sub('.syslog', '', self.msg_key_path)
            return True
        except:
            return False

    def buffer_msg_size(self):
        content_length = len(self.body)
        content = self.body.split('\n')
        owner_id = self.owner_id
        service_name = self.service_name
        my_timestamp = self.datetime
        main_body = ''

        for line in content:
            line_length = len(line)
            if line_length < 8000:
                main_body += line + "\n\n"
            else:
                my_uuid = self.my_random_sequence()
                part_count = math.ceil(line_length / 8000)
                print('part_count: ', part_count)
                for x in range(0, part_count):
                    if x == part_count:
                        submessage = line[(x * 8000):0]
                        submessage = str(my_timestamp) + " account-" + str(owner_id) + " " + service_name + " " + submessage + " sequence " + str(my_uuid) + " part " + str(x + 1) + " of " + str(part_count) + "\n\n"
                        main_body += submessage
                    else:
                        submessage = line[(x * 8000):((x + 1) * 8000)]
                        submessage = str(my_timestamp) + " account-" + str(owner_id) + " " + service_name + " " + submessage + " sequence " + str(my_uuid) + " part " + str(x + 1) + " of " + str(part_count) + "\n\n"
                        main_body += submessage
        self.body = main_body
        try:
            self.transport_syslog_body_to_gz_s3()
        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not write message to out_queue:', exception_string)

    def parse_exception(self):
        # can let message process 4 times and fail to dlq, or copy directly to dlq, &/or write to ddb table
        line_split = self.msg_key_path.split("/")
        accountID = line_split[0]
        now = datetime.now().today().strftime("%Y-%m-%dT%H:%M:%S:%f")
        ddb = boto3.client('dynamodb')
        table_name = 'AwsLogsFilterExceptions'
        if self.check_body_is_gzip() == True:
            self.gunzip_body()
        # limiting body message to < 400kb, max limit of ddb entry
        self.body = str(self.body[0:3000])
        item = {
            'accountID': {'S': accountID},
            'eventDateTime': {'S': now},
            'region': {'S': REGION},
            'msg_key_path': {'S': self.msg_key_path},
            'tag': {'S': self.tag},
            'body_data': {'S': self.body},
        }

        try:
            ddb.put_item(
                TableName=table_name,
                Item=item
            )
            return True
        except botocore.exceptions.ClientError as error:
            print('can not write to ddb' + 'error: {0}'.format(error))
            return False

    def my_random_sequence(self,string_length=5):
        sequence = str(uuid.uuid4()).upper()
        return sequence[0:string_length]

    def transport_syslog_body_to_gz_s3(self):
        self.add_syslog_suffix_to_path()
        self.gzip_body()
        self.add_gz_suffix_to_path()
        self.put_s3_object()
        return True

The Lambda Handler

def lambda_handler(event, context):
    """
    Flow control for aws-logs-filter.
    """
    import aws_logs_filter
    import boto3
    import botocore
    import re
    import os
    import yaml


    # setup vars
    config_bucket = '<redacted>'
    yaml_config = 'config.yaml'
    s3 = boto3.client('s3')
    config_handle = s3.get_object(Bucket=config_bucket, Key=yaml_config)
    config_data = config_handle.get('Body').read()
    yaml_data = yaml.load(config_data)

    REGION = os.environ.get('AWS_REGION')
    FILTERS = yaml_data['FILTERS'] # TODO see if this use of FILTERS is necessary, called as function in aws_logs_filter
    jobs = yaml_data['jobs']
    aggregate_accounts = yaml_data['aggregate_accounts']

    in_bucket = '<redacted>' + REGION
    out_bucket = '<redacted>-filtered-' + REGION

    sqs = boto3.client('sqs')
    in_queue = 'https://sqs.' + REGION + '.amazonaws.com/<redacted>/Log-Aggregation-Queue'
    out_queue = 'https://sqs.' + REGION + '.amazonaws.com/<redacted>/Log-Aggregation-Dead-Letter-Queue'

    ddb = boto3.client('dynamodb')
    table_name = 'AwsLogsFilterExceptions'

    sent_flag = 'no'

    for record in event['Records']:

        try:
            message = aws_logs_filter.Message(sqs, record, in_queue, out_queue)
            # print(message.tag)
        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not retrieve message:', exception_string)

        try:
            filtered_result = message.filter_message()
            # print(message.tag)
        except botocore.exceptions.ClientError as error:
            exception_string = 'Raised exception: error: {0}'.format(error) + '\n'
            print('can not filter:', exception_string)

        # simple filter function, mostly relevant to us-west-2, but some global entries
        if filtered_result is True:
            print('dropped message: ', message.msg_key_path)
            message.drop_message()

        # main processing flow control
        if filtered_result is False:
            print('loading message:', message.msg_key_path)
            try:
                file = aws_logs_filter.File(s3, message.msg_key_path, in_bucket, out_bucket)
                file.load_s3_object_to_body()
            except:
                print('could not load file into memory')
                break
                # could result in dlq messages from unprocessed from break.
                # TODO fix so this fails gracefully

            try:
                if file.check_body_is_gzip() == True:
                    file.gunzip_body()
                    file.strip_path_of_suffix()
                elif file.check_body_is_gzip() == False:
                    file.check_body_is_ascii()

            except:
                print('file is unparsable: ', file.msg_key_path)
                file.parse_exception()

            for key in jobs.keys():
                file.determine_if_agg_account()
                #print('account id: ',file.owner_id)
                match_variable = jobs[key]['match_variable']
                action_variables = jobs[key]['action_variables']
                service_variable = jobs[key]['service_variable']
                # TODO make sure ownerID variables are in place


                match = re.search(match_variable, message.msg_key_path, re.IGNORECASE)

                if match:
                    file.determine_service_name(service_variable)
                    print("service: ", file.service_name)
                    action_count = len(action_variables)
                    #print('action_count: ', action_count)
                    for i in range(0, action_count):
                        #print(action_variables[i])

                        if action_variables[i] == 'firehose':
                            file.strip_path_of_suffix()
                            file.fix_body_firehose()

                        elif action_variables[i] == 'json':
                            file.strip_path_of_suffix()
                            file.parse_body_json_to_syslog()

                        elif action_variables[i] == 'text_wrap':
                            file.parse_ascii_body_to_syslog()
                        #print('end of action #', i)
                    file.buffer_msg_size()
                    sent_flag = "yes"

            if sent_flag == "yes":
                print('sent: ', sent_flag)
            else:
                print('write to ddb as error')
                result = file.parse_exception()
                if result is True:
                    message.drop_message()
                # if false, no db entry, message goes to dead letter for further processing

Cloudformation template

AWSTemplateFormatVersion: 2010-09-09
Description: >-
    Creates an instance of aws_logs_filter.py for filtering, wrapping and transporting
    log files so they can be pulled on-premise for further processing.

Parameters:
    LambdaCodeBucket:
        Description: The bucket where you uploaded the zip file
        Type: String
    LambdaCodeFile:
        Description: the name of the zip file you uploaded
        Type: String
    FunctionInstanceName:
        Description: Name of this instance of the function (must be unique).
        Type: String
Resources:
    AWSLogsFilter:
        Type: AWS::Lambda::Function
        Properties:
            Code:
                S3Bucket:
                    Ref: LambdaCodeBucket
                S3Key:
                    Ref: LambdaCodeFile
            FunctionName:
                Ref: FunctionInstanceName
            Handler: lambda_function.py
            MemorySize: 512
            Role: arn:aws:iam::<redacted>:role/lambda_filter_log_agg_sqs_queue
            Timeout: 900
            Description: >-
                pulls log file, wraps as syslog, segments and sequences long messages
                and puts compressed file to filter bucket
            Runtime: python3.6
    LambdaFunctionEventSourceMapping:
        Type: AWS::Lambda::EventSourceMapping
        Properties:
            BatchSize: 10
            Enabled: true
            #change above to true for production
            EventSourceArn:
                Fn::Join:
                    - ''
                    - 
                      - 'arn:aws:sqs:'
                      - Ref: 'AWS::Region'
                      - ':<redacted>:Log-Aggregation-Queue'
            FunctionName: 
                Ref: AWSLogsFilter