AWS

Create SQS Queue

Use the following values for Type, Name, and Access policy. Leave the other options at their defaults or set them according to your environment.

• Type: Select FIFO

• Name: Enter logpresso-aws-aurora-das.fifo

• Access policy: Select Advanced and configure it based on the following access policy (replace REGION and ACCOUNT_ID with your values)

  {
    "Version": "2012-10-17",
    "Id": "__default_policy_ID",
    "Statement": [
      {
        "Sid": "__owner_statement",
        "Effect": "Allow",
        "Principal": {
          "AWS": "arn:aws:iam::ACCOUNT_ID:root"
        },
        "Action": "SQS:*",
        "Resource": "arn:aws:sqs:REGION:ACCOUNT_ID:logpresso-aws-aurora-das.fifo"
      },
      {
        "Sid": "__receiver_statement",
        "Effect": "Allow",
        "Principal": {
          "AWS": "arn:aws:iam::ACCOUNT_ID:user/logpresso"
        },
        "Action": [
          "SQS:ChangeMessageVisibility",
          "SQS:DeleteMessage",
          "SQS:ReceiveMessage"
        ],
        "Resource": "arn:aws:sqs:REGION:ACCOUNT_ID:logpresso-aws-aurora-das.fifo"
      }
    ]
  }
  

The page right after creating the SQS queue looks like this.

Create Lambda Function

In the Lambda console, go to Event Functions and click Create Event Function. This function is invoked whenever a log file is stored in S3 and sends the logs to the SQS queue.

1. Select Author from scratch, enter the Basic information options as follows, and click Create function. Leave the other options at their defaults or set them according to your environment.

   • Function name: Enter logpresso-aws-aurora-das-log-producer

   • Runtime: Python 3.12

2. The page right after creating the function looks like this.

   

   On the Code tab, replace the contents of lambda_function.py with the following code. Change the string assigned to the sqs_queue_url variable to the URL of the logpresso-aws-aurora-das.fifo SQS queue.

   import json
   import boto3
   import gzip
   
   s3_client = boto3.client('s3')
   sqs_client = boto3.client('sqs')
   sqs_queue_url = 'https://sqs.REGION.amazonaws.com/ACCOUNT_ID/QUEUE'
   
   def lambda_handler(event, context):
       # process_s3_object('BUCKET_NAME', 'FILE_NAME')
       for record in event['Records']:
            bucket_name = record['s3']['bucket']['name']
            object_key = record['s3']['object']['key']
            process_s3_object(bucket_name, object_key)
   
       return {
           'statusCode': 200,
           'body': json.dumps('Completed')
       }
   
   def process_s3_object(bucket_name, object_key):
       response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
       data = gzip.decompress(response['Body'].read()).decode('utf-8')
   
       lines = data.splitlines()
       filtered = []
       i = 0
       for line in lines:
           try:
               i = i + 1
               log_entry = json.loads(line)
   
               # NOTE: Add your own filtering condition below
               if 'type' in log_entry:
                   filtered.append(json.dumps(log_entry))
           except: 
               print('error object key ' +  object_key + ' row #' + str(i))
               continue
   
       dup_id = 0
       if filtered:
           max_message_size = 128 * 1024  # 128KB
           current_message = []
           current_size = 0
   
           for log in filtered:
               log_size = len(log.encode('utf-8')) + 1 
               if current_size + log_size > max_message_size:
                   send_message_to_sqs(sqs_queue_url, "\n".join(current_message), object_key, dup_id)
                   dup_id = dup_id + 1
                   current_message = [log]
                   current_size = log_size
               else:
                   current_message.append(log)
                   current_size += log_size
   
           if current_message:
               send_message_to_sqs(sqs_queue_url, "\n".join(current_message), object_key, dup_id)
               dup_id = dup_id + 1
   
   
   
   def send_message_to_sqs(queue_url, message_body, object_key, dup_id):
       message_group_id = 'logpresso'
       message_deduplication_id = object_key.split('/')[-1]
   
       response = sqs_client.send_message(
           QueueUrl=queue_url,
           MessageBody=message_body,
           MessageGroupId=message_group_id,
           MessageDeduplicationId=object_key + str(dup_id)
       )
   
       print(response)
       return response
   

Test with a Simulated Event

Test that the Lambda function works correctly. On the Test tab, select Create new event and edit the simulated event as shown below.

Click Test at the top right to verify that the test runs successfully.

Lambda Inline Policy

If the test runs successfully, grant the Lambda function permission to access S3.

1. On the Configuration tab, select Permissions and click the Role name under Execution role. You are taken to the IAM console.

   

2. On the IAM role page, click Add permissions > Create inline policy.

   

3. In the policy editor, paste the JSON below to create the IAM policy (replace REGION, ACCOUNT_ID, BUCKET_NAME, and QUEUE with your values).

   {
   	"Version": "2012-10-17",
   	"Statement": [
   		{
   			"Sid": "Statement1",
   			"Effect": "Allow",
   			"Action": [
   				"s3:GetObject",
   				"sqs:SendMessage"
   			],
   			"Resource": [
   				"arn:aws:s3:::BUCKET_NAME",
   				"arn:aws:s3:::BUCKET_NAME/*",
   				"arn:aws:sqs:REGION:ACCOUNT_ID:QUEUE.fifo"
   			]
   		}
   	]
   }
   

Add Lambda Function Trigger

Now add a function trigger so that the Lambda function is invoked whenever a log file is stored in S3.

1. In the Function overview, click Add trigger.

2. On the Add trigger page, select the S3 bucket and select All object create events as the Event types. If a warning appears, check the acknowledgment box and click Add.

   

Verify Execution

On the Monitor tab, confirm that the Lambda function is invoked whenever logs are written to S3.