Use Golang for Data Processing With Amazon SNS and AWS Lambda

gowebing 2024-12-02

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:

     Shell 
   
   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.

     Shell 
   
   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

Once all the resources are created, you can try out the application. You should have:

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):

     Shell 
   
   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:

     Shell 
   
   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:

     Shell 
   
 
 
   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.

     Shell 
   
 
 
   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.

     Shell 
   
   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.

     Shell 
   
 
 
   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.

     Shell 
   
 
 
   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!