Ingest data into Elasticsearch with AWS Lambda and SNS using LocalStack
Tags: aws, elasticsearch, sns, lambda
Published: 2022-06-07 15:05:09

In this post I will explain and show the implementation of an architecture that ingest data into Elasticsearch with Lambda and SNS using LocalStack. In simple terms, the architecture looks like this:

![Architecture](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/kk87fgkpeebm6jlt23e0.png)

The ideia is to monitor the business events of an application. So once the application wants to send a business event, this event is sent as a message to a SNS topic. After that a lambda function receives the message and send it to elasticsearch. Finally we can see the event using Kibana interface.

### LocalStack

LocalStack is a framework originally developed by AWS that we can use to simulate AWS environment locally. With LocalStack we can set up SNS topics and Lambda functions.

### SNS

SNS is an AWS service that works like a pub/sub, so we create SNS topics and send messages to these topics, once these messages are received by SNS, it pushes notifications to subscribers. SNS subscribers can be SQS (AWS queue service), Lambda functions, HTTP endpoints and Amazon Kinesis.

### Lambda

Lambda is a serverless service that we can run on AWS environment, that is, we upload the code and AWS takes care of all compute resources needed. 

### Elasticsearch

Elasticsearch is a powerful engine that store any kind of data with the purpose of analyzing and searching with high performance and escalability. 

### Kibana

Kibana provides a graphic interface to elasticsearch and lets us visualize the data with charts and graphs.

### Architecture implementation

First of all we will configure a `docker-compose.yml` file so we can run LocalStack, Elasticsearch and Kibana locally:

```yaml
version: '3.5'
services:
  localstack:
    container_name: localstack
    image: localstack/localstack:latest
    networks:
      - default
    ports:
      - '4566:4566'
    environment:
      - EDGE_PORT=4566
      - SERVICES=sns,lambda

  elasticsearch:
    container_name: elasticsearch
    image: docker.elastic.co/elasticsearch/elasticsearch:7.7.0
    environment:
      discovery.type: "single-node"
      ES_JAVA_OPTS: "-Xms2g -Xmx2g"
      xpack.monitoring.enabled: "true"
    ports:
      - "9200:9200"
    networks:
      - default

  kibana:
    container_name: kibana
    image: docker.elastic.co/kibana/kibana:7.7.0
    ports:
      - "5601:5601"
    networks:
      - default

networks:
  default:
    driver: bridge
```

Now run the `docker-compose.yml` file:

```shell
docker-compose up -d
```

Install the [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html), so we can communicate with the LocalStack container easier and then configure AWS running:

```shell
aws configure
```
You can use dummy credentials for configuration because LocalStack doesn't really validate them.

#### Creating the SNS topic

Now create a SNS topic to send the business events. We will name our topic `events-topic`:

```shell
aws --no-cli-pager --endpoint-url=http://0.0.0.0:4566 sns create-topic --name events-topic
```
#### Creating the Lambda function

Before creating the lambda in LocalStack, we need to implement the code. Here I will be using `node.js`:

**1)** Choose a directory and initiate a node project:

```javascript
npm init
```

**2)** Install the elasticsearch library inside your project: 

```javascript
npm i elasticsearch
```

**3)** Create a directory (I called it elastic) and add the `elastic.js` file inside:

```javascript
var elasticsearch = require('elasticsearch');

const elasticClient = elasticsearch.Client({
    host: 'http://elasticsearch:9200'
});

var indexName = 'business_events';

function initIndex() {    
    try {
        return elasticClient.indices.create({
            index: indexName
        });
    } catch (error) {
        console.log('Error creating elasticsearch index')
    }
}
exports.initIndex = initIndex;

function addDocument(message) {
    console.log('Adding document to elastic');

    return elasticClient.index({
        index: indexName,
        type: "document",
        body: {
            event_name: message.name ?? null,
            properties: message.properties ?? null,
            created_at: message.created_at ?? null,
        }
    });
}
exports.addDocument = addDocument;
```
    
This file implements the two main functions to manipulate elasticsearch. First we connect to the local elasticsearch `http://elasticsearch:9200`, then we create the function `initIndex()` responsible to create the elasticsearch index (`business_events`). Ultimately we create the `addDocument()` function to send the data to elastic.

**4)** Create the `index.js` file at the project root:

```javascript
var elasticsearch = require('./elastic/elastic');

elasticsearch.initIndex();

exports.handler = function(event) {
    var message = event.Records[0].Sns.Message;

    elasticsearch.addDocument(JSON.parse(message));

    return 'Success';
};    
```

Here we implement the main function `handler()` which will be invoked by SNS every time a new message arrives. This function just retrieves the message from SNS topic and add it to elasticsearch.

**5)** Access your code directory, select all files and zip them with the name: `function.zip`.

Now the code is done and you can find the complete implementation in my github: [lambda-sns-elasticsearch](https://github.com/jainec/lambda-sns-elasticsearch).

**6)** The next step is to create a lambda function in LocalStack and upload the compressed code to it:

```shell
aws --no-cli-pager --endpoint-url=http://0.0.0.0:4566 lambda create-function --function-name events-lambda \
    	  --zip-file fileb://function.zip --handler index.handler --runtime nodejs12.x --role _
```
#### Subscribing lambda to SNS topic

Finally subscribe the lambda to the sns topic we created:

```shell
aws --no-cli-pager --endpoint-url=http://0.0.0.0:4566 sns subscribe --topic-arn arn:aws:sns:us-east-1:000000000000:events-topic --protocol lambda \
    	  --notification-endpoint arn:aws:lambda:us-east-1:000000000000:function:events-lambda
```

From now on every message you send to the SNS topic, will be received by the lambda function which will send it to elasticsearch. We can see the messages accessing kibana: `http://localhost:5601/`

### Testing

Let's make a test sending a message to the SNS topic via command line:

```shell
aws --no-cli-pager sns publish --endpoint-url=http://0.0.0.0:4566 --topic-arn arn:aws:sns:us-east-1:000000000000:events-topic --message '{"name":"test", "user_id":1, "properties": {"nickname": "test-user1", "job": "Software engineer"}}'
```
Now access Kibana and make sure that you have configured the index pattern:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/e1jd8cjpzd3dlw9sniq0.png)

Then go to Discover section and your message should appear like that:

![Image description](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/6gdp8s6n4gjxzfk77y36.png)

Instead of sending messages via command line, you can set your services to send messages to a SNS topic using [AWS SDK](https://aws.amazon.com/pt/tools/).

Thank you for reading :)