AWS: Automatic Subscription Confirmation from SQS Queue to SNS Topic

 We have an architecture in AWS where different events from different accounts need to be sent to one central SQS queue. Since the events will cross both accounts and regions, one way to do it is to send them to a local SNS Topic. 

The SQS queue will have to subscribe to all those Topics, but we can not do it on the SQS side, since it does not know each time someone pops out a new account. However, the problem with having the SNS Topics create the subscriptions, is that they are waiting for confirmation from the SQS queue.

Since we already have a lambda waiting on the other side of the queue, handling all the events, we added a small code to handle the subscription confirmation as well. Here it is:

import json

import urllib.request

def lambda_handler(event, context):

    for record in event["Records"]:

        body = json.loads(record["body"])

        if body.get("Type") == "SubscriptionConfirmation":

            handle_subscription_confirmation(body)

def handle_subscription_confirmation(message):

    url = message["SubscribeURL"]

    with urllib.request.urlopen(url) as response:

        print(response.read())

I find it strange that the Cloudformation template that we use to create the subscription does not handle the confirmation as well. Or maybe not cross-account?

Send Batch of Messages to SQS

We have a simple code that can send up to several thousand messages to an SQS queue. Using Python and boto3, the code looks like this:

sqs = boto3.resource('sqs')
queue = sqs.get_queue_by_name(QueueName=SQS_QUEUE_NAME)
for message in messages:
    queue.send_message(message)

When you have really a lot of messages in an array, it is possible to send them by batch of up to 10 messages, using the send_message_batch method. When doing this, there are two problems to solve: creating the batches of 10 messages, and generating an ID for each message. AWS enforce the generation of this ID so it can send back a response containing the list of messages that failed or succeeded, identified by the ID.

Here is the new code:

sqs = boto3.resource('sqs')
queue = sqs.get_queue_by_name(QueueName=SQS_QUEUE_NAME)

for i in range(0, len(messages), 10):
    chunk = messages[i:i+10]
    queue.send_messages(Entries=[
        {
            "Id": "MSG" + str(id), 
            "MessageBody": message
        } for id, message in zip(range(10), chunk)
    ])

Our loop now jump to every tenth message. Inside the loop, we create a chunk of 10 messages, and use the send_message_batch method to send it. You can see that we use a list comprehension the runs over both the chunk and a range to generate the ID.