DynamoDBStreams-LambdaTrigger-ElasticSearch.txt

0. Make sure you have your Cloud9 environment up and running from the previous lab

1. Create ES domain
    - Development type = Development and Testing
    - Elasticsearch Version = 7.10
    - ES Domain name = <Name>
    - Auto Tune = Disable
    - Data Nodes 
        Instance Type = t3.small.elasticsearch
        Number of nodes = 1
    - Dedicated Master nodes = No
    - Network configuration = Public Access
    - Enable fine-grained access control = Uncheck
    - Access Policy = Custom Access Policy
        - IPv4 Address = mention your laptop/local machine IP address (find your ip address https://checkip.amazonaws.com/.)
        OR you can provide Cloud 9 Public IP address
    - Do not change remaining settings
    - Confirm

From your Cloud 9 Terminal 
--------------------------
2. Create DDB Table 

    aws dynamodb create-table \
        --table-name Movies \
        --attribute-definitions AttributeName=director,AttributeType=S AttributeName=id,AttributeType=S \
        --key-schema AttributeName=director,KeyType=HASH  AttributeName=id,KeyType=RANGE \
        --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
        --stream-specification StreamEnabled=true,StreamViewType=NEW_IMAGE


3. Create a Lambda Execution role

    - Create a file named trust-relationship.json with the following contents. 

        {
        "Version": "2012-10-17",
        "Statement": [
            {
            "Effect": "Allow",
            "Principal": {
                "Service": "lambda.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
            }
        ]
        }

    - Create DDBStreamESLambdarole role

        aws iam create-role --role-name DDBStreamESLambdarole \
        --path "/service-role/" \
        --assume-role-policy-document file://trust-relationship.json

    - Create a file named new-role-policy.json with the following contents.
    
        {
        "Version": "2012-10-17",
        "Statement": [
            {
            "Effect": "Allow",
            "Action": [
                "es:ESHttpPost",
                "es:ESHttpPut",
                "dynamodb:DescribeStream",
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator",
                "dynamodb:ListStreams",
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": "*"
            }
        ]
        } 

4. Enter the following command to attach the policy to DDBStreamESLambdarole. 

    aws iam put-role-policy --role-name DDBStreamESLambdarole \
        --policy-name DDBStreamESLambdarolePolicy \
        --policy-document file://new-role-policy.json


5. Create directory 

    $ mkdir ddb-to-es

6. Install required packages 

    $ cd ddb-to-es
    $ pip install requests -t .
    $ pip install requests_aws4auth -t .

7. Create lambda function file sample.py in directory ddb-to-es

    Replace region with ap-southeast-2 and host with your ElasticSearch domain endpoint)
        
        import boto3
        import requests
        from requests_aws4auth import AWS4Auth

        region = 'ap-southeast-2' 
        service = 'es'
        credentials = boto3.Session().get_credentials()
        awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

        host = '****YOUR ELASTIC SEARCH DOMAIN ENDPOINT*****' # the Amazon ES domain, with https://
        url = host + '/movies/_doc/'

        headers = { "Content-Type": "application/json" }

        def handler(event, context):
            count = 0
            for record in event['Records']:
                # Get the primary key for use as the Elasticsearch ID
                id = record['dynamodb']['Keys']['id']['S']

                if record['eventName'] == 'REMOVE':
                    r = requests.delete(url + id, auth=awsauth)
                else:
                    document = record['dynamodb']['NewImage']
                    r = requests.put(url + id, auth=awsauth, json=document, headers=headers)
                count += 1
            return str(count) + ' records processed.'

8. Package the application code and dependencies:

$ zip -r lambda.zip *

9. Get IAM Role ARN of DDBStreamESLambdarole role that you created earlier in this lab

aws iam get-role --role-name DDBStreamESLambdarole

10. Enter the following command to create the Lambda function
Replace roleARN with the ARN of DDBStreamESLambdarole role

aws lambda create-function \
    --region ap-southeast-2 \
    --function-name dds-to-es-function \
    --zip-file fileb://lambda.zip \
    --role <roleARN> \
    --handler sample.handler \
    --timeout 5 \
    --runtime python3.8


11. Create a test event, create a file movies_payload.json with the following content
(replace region ap-southeast-2 and accountID)
    {
        "Records": [
            {
                "eventID": "7de3041dd709b024af6f29e4fa13d34c",
                "eventName": "INSERT",
                "eventVersion": "1.1",
                "eventSource": "aws:dynamodb",
                "awsRegion": "<region>",
                "dynamodb": {
                    "ApproximateCreationDateTime": 1479499740,
                    "Keys": {
                        "id": {
                            "S": "101"
                        },
                        "director": {
                            "S": "John Doe"
                        }
                    },
                    "NewImage": {
                        "Timestamp": {
                            "S": "2016-11-18:12:09:36"
                        },
                        "Message": {
                            "S": "This is a bark from the Woofer social network"
                        },
                        "director": {
                            "S": "John Doe"
                        }
                    },
                    "SequenceNumber": "13021600000000001596893679",
                    "SizeBytes": 112,
                    "StreamViewType": "NEW_IMAGE"
                },
                "eventSourceARN": "arn:aws:dynamodb:ap-southeast-2:<****accountID****>:table/Movies/stream/2021-08-30T14:53:27.617"
            }
        ]
    }

12. Test the Lambda function dds-to-es-function

aws lambda invoke --function-name dds-to-es-function --payload file://movies_payload.json output.txt

- Check output.txt file for response

13. Create a DynamoDB Trigger

    Find the StreamARN by running the following command 

    aws dynamodb describe-table --table-name Movies

    Run the following command to create a trigger. Replace streamARN with Movies StreamARN

    aws lambda create-event-source-mapping \
        --region ap-southeast-2 \
        --function-name dds-to-es-function \
        --event-source *****StreamARN*****  \
        --batch-size 1 \
        --starting-position TRIM_HORIZON

14. Test the trigger. Enter the following command to add an item to Movies. 

aws dynamodb put-item --table-name Movies --item '{"director": {"S": "Jake Kasdan"},"id": {"S": "00001"},"title": {"S": "Jumanji"}}' --region ap-southeast-2

//From your local machine try to access
curl -XGET https://search-ddb-domain-streams-ycxe4l4blylm2fygk3s52y7ovq.ap-southeast-2.es.amazonaws.com/movies/_doc/00001


aws dynamodb put-item --table-name Movies --item '{"director": {"S": "Peter Jackson"},"id": {"S": "00002"},"title": {"S": "King Kong"}}' --region ap-southeast-2

//From your local machine try to access
curl -XGET https://search-ddb-domain-streams-ycxe4l4blylm2fygk3s52y7ovq.ap-southeast-2.es.amazonaws.com/movies/_doc/00002