Use Golang for Data Processing With Amazon SNS and AWS Lambda

In this blog post, you will be using the aws-lambda-go library along with the AWS Go SDK v2 for an application that will process records from an Amazon SNS topic and store them in a DynamoDB table.

You will also learn how to use Go bindings for AWS CDK to implement “Infrastructure-as-code” for the entire solution and deploy it with the AWS CDK CLI.

The code is available on GitHub.

Introduction

Amazon Simple Notification Service (SNS) is a highly available, durable, and scalable messaging service that enables the exchange of messages between applications or microservices. It uses a publish/subscribe model where publishers send messages to topics, and subscribers receive messages from topics they are interested in. Clients can subscribe to the SNS topic and receive published messages using a supported endpoint type, such as Amazon Kinesis Data Firehose, Amazon SQS, AWS Lambda, HTTP, email, mobile push notifications, and mobile text messages (SMS).

AWS Lambda and Amazon SNS integration enable developers to build event-driven architectures that can scale automatically and respond to changes in real time. When a new message is published to an SNS topic, it can trigger a Lambda function (Amazon SNS invokes your function asynchronously with an event that contains a message and metadata) which can perform a set of actions, such as processing the message, storing data in a database, sending emails or SMS messages, or invoking other AWS services.

Prerequisites

Before you proceed, make sure you have the Go programming language (v1.18 or higher) and AWS CDK installed.

Clone the project and change it to the right directory:

git clone https://github.com/abhirockzz/sns-lambda-events-golang

cd sns-lambda-events-golang

Use CDK To Deploy the Solution

To start the deployment, simply invoke cdk deploy and wait for a bit. You will see a list of resources that will be created and will need to provide your confirmation to proceed.

cd cdk

cdk deploy

# output

Bundling asset SNSLambdaGolangStack/sns-function/Code/Stage...

✨  Synthesis time: 5.94s

This deployment will make potentially sensitive changes according to your current security approval level (--require-approval broadening).
Please confirm you intend to make the following modifications:

//.... omitted

Do you wish to deploy these changes (y/n)? y

This will start creating the AWS resources required for our application.

If you want to see the AWS CloudFormation template which will be used behind the scenes, run cdk synth and check the cdk.out folder.

You can keep track of the progress in the terminal or navigate to AWS console: CloudFormation > Stacks > SNSLambdaGolangStack

• A Lambda function
• A SNS topic
• A DynamoDB table
• Along with a few other components (like IAM roles etc.)

Verify the Solution

You can check the table and SNS info in the stack output (in the terminal or the Outputs tab in the AWS CloudFormation console for your Stack):

Send few messages to the SNS topic. For the purposes of this demo, you can use the AWS CLI:

export SNS_TOPIC_ARN=<enter the queue url from cloudformation output>

aws sns publish --topic-arn $SNS_TOPIC_ARN --message "user1@foo.com" --message-attributes 'name={DataType=String, StringValue="user1"}, city={DataType=String,StringValue="seattle"}'

aws sns publish --topic-arn $SNS_TOPIC_ARN --message "user2@foo.com" --message-attributes 'name={DataType=String, StringValue="user2"}, city={DataType=String,StringValue="new delhi"}'

aws sns publish --topic-arn $SNS_TOPIC_ARN --message "user3@foo.com" --message-attributes 'name={DataType=String, StringValue="user3"}, city={DataType=String,StringValue="new york"}'

You can also use the AWS console to send SQS messages.

Check the DynamoDB table to confirm that the file metadata has been stored. You can use the AWS console or the AWS CLI aws dynamodb scan --table-name <enter the table name from cloudformation output>

Don’t Forget To Clean Up

Once you’re done, to delete all the services, simply use:

cdk destroy

#output prompt (choose 'y' to continue)

Are you sure you want to delete: SQSLambdaGolangStack (y/n)?

You were able to setup and try the complete solution. Before we wrap up, let’s quickly walk through some of important parts of the code to get a better understanding of what’s going the behind the scenes.

Code Walk Through

Some of the code (error handling, logging etc.) has been omitted for brevity since we only want to focus on the important parts.

CDK

You can refer to the CDK code here.

We start by creating a DynamoDB table:

table := awsdynamodb.NewTable(stack, jsii.String("dynamodb-table"),
        &awsdynamodb.TableProps{
            PartitionKey: &awsdynamodb.Attribute{
                Name: jsii.String("email"),
                Type: awsdynamodb.AttributeType_STRING},
        })

    table.ApplyRemovalPolicy(awscdk.RemovalPolicy_DESTROY)

Then, we handle the Lambda function (CDK will take care of building and deploying the function) and make sure we provide it appropriate permissions to write to the DynamoDB table.

function := awscdklambdagoalpha.NewGoFunction(stack, jsii.String("sns-function"),
        &awscdklambdagoalpha.GoFunctionProps{
            Runtime:     awslambda.Runtime_GO_1_X(),
            Environment: &map[string]*string{"TABLE_NAME": table.TableName()},
            Entry:       jsii.String(functionDir),
        })

    table.GrantWriteData(function)

Then, we create the SNS topic and add that as an event source to the Lambda function.

snsTopic := awssns.NewTopic(stack, jsii.String("sns-topic"), nil)
    function.AddEventSource(awslambdaeventsources.NewSnsEventSource(snsTopic, nil))

Finally, we export the SNS topic and DynamoDB table name as CloudFormation outputs.

awscdk.NewCfnOutput(stack, jsii.String("sns-topic-name"),
        &awscdk.CfnOutputProps{
            ExportName: jsii.String("sns-topic-name"),
            Value:      snsTopic.TopicName()})

    awscdk.NewCfnOutput(stack, jsii.String("dynamodb-table-name"),
        &awscdk.CfnOutputProps{
            ExportName: jsii.String("dynamodb-table-name"),
            Value:      table.TableName()})

Lambda Function

You can refer to the Lambda Function code here.

The Lambda function handler iterates over each SNS topic, and for each of them:

• Stores the message body in the primary key attribute (email) of the DynamoDB table
• Rest of the message attributes are stored as is.

func handler(ctx context.Context, snsEvent events.SNSEvent) {
    for _, record := range snsEvent.Records {

        snsRecord := record.SNS

        item := make(map[string]types.AttributeValue)
        item["email"] = &types.AttributeValueMemberS{Value: snsRecord.Message}

        for attrName, attrVal := range snsRecord.MessageAttributes {
            fmt.Println(attrName, "=", attrVal)
            attrValMap := attrVal.(map[string]interface{})

            dataType := attrValMap["Type"]

            val := attrValMap["Value"]

            switch dataType.(string) {
            case "String":
                item[attrName] = &types.AttributeValueMemberS{Value: val.(string)}
            }
        }

        _, err := client.PutItem(context.Background(), &dynamodb.PutItemInput{
            TableName: aws.String(table),
            Item:      item,
        })
    }
}

Wrap Up

In this blog, you saw an example of how to use Lambda to process messages sent to SNS and store them in DynamoDB, thanks to the SNS and Lamdba integration. The entire infrastructure life-cycle was automated using AWS CDK.

All this was done using the Go programming language, which is well-supported in DynamoDB, AWS Lambda, and AWS CDK.

Happy building!