Saga Pattern #3: Thinking Serverless

aws-profile-setup.sh

aws configure --profile saga-pattern-serverless

export AWS_ACCOUNT_ID=YOUR_AWS_ACCOUNT_ID
export AWS_REGION=$(aws configure get region --profile saga-pattern-serverless)

dlq-setup.sh

aws sqs create-queue \
 --queue-name route-check-time-dlq \
 --profile saga-pattern-serverless

lambda-layer-setup.sh

cd layer/
zip -r layer.zip python
cd ../


aws lambda publish-layer-version \
 --layer-name common-layer \
 --compatible-runtimes python3.9 \
 --compatible-architectures x86_64 \
 --zip-file fileb://layer/layer.zip \
 --profile saga-pattern-serverless

lambda-setup.sh

zip -j services/route/check-time/lambda.zip services/route/check-time/lambda_function.py


aws lambda create-function \
 --function-name route-check-time \
 --layers arn:aws:lambda:"$AWS_REGION":"$AWS_ACCOUNT_ID":layer:common-layer:1 \
 --runtime python3.9 \
 --handler lambda_function.lambda_handler \
 --zip-file fileb://services/route/check-time/lambda.zip \
 --environment file://services/route/check-time/environment.json \
 --role arn:aws:iam::"$AWS_ACCOUNT_ID":role/route-check-time-role \
 --dead-letter-config TargetArn=arn:aws:sqs:"$AWS_REGION":"$AWS_ACCOUNT_ID":route-check-time-dlq \
 --profile saga-pattern-serverless
 

aws lambda put-function-event-invoke-config \
 --function-name route-check-time \
 --destination-config '{"OnSuccess":{"Destination": "arn:aws:sns:$AWS_REGION:$AWS_ACCOUNT_ID:driver-cancel-trip-trigger-topic"}}' \
 --profile saga-pattern-serverless

policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:AWS_REGION:AWS_ACCOUNT_ID:*"
        },

        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:AWS_REGION:AWS_ACCOUNT_ID:log-group:/aws/lambda/route-check-time:*"
            ]
        },

        {
            "Effect": "Allow",
            "Action": [
                "sqs:SendMessage"
            ],
            "Resource": [
                "arn:aws:sqs:AWS_REGION:AWS_ACCOUNT_ID:route-check-time-dlq"
            ]
        },

        {
            "Effect": "Allow",
            "Action": [
                "sns:Publish"
            ],
            "Resource": [
                "arn:aws:sns:AWS_REGION:AWS_ACCOUNT_ID:driver-cancel-trip-trigger-topic"
            ]
        }
    ]
}

rds-setup.sh

aws rds create-db-instance \
 --db-instance-identifier route-db \
 --db-instance-class db.t3.micro \
 --engine postgres \
 --master-username postgres \
 --master-user-password some_super_secure_password \
 --no-multi-az \
 --storage-type gp2 \
 --allocated-storage 5 \
 --no-enable-iam-database-authentication \
 --no-enable-performance-insights \
 --publicly-accessible \
 --profile saga-pattern-serverless

route-check-time.py

# Route: Check Time


import json
import psycopg2
import logging
from datetime import datetime
from common import db_connection


logger = logging.getLogger()
logger.setLevel(logging.INFO)


class TimeLimitException(Exception):
    """
    Simple class to indicate an error when attempt to
    updated a route that exceed the time limit
    """
    pass


def check_time_limit(trip):
    # Trip is a tuple formed by (id, created_at)
    current_time = datetime.now()
    trip_created_at = trip[1]
    difference_in_minutes = (abs(current_time - trip_created_at)).total_seconds() / 60
    if difference_in_minutes > 5:
        raise TimeLimitException("Unable to update route due to time limit exceeded")
    else:
        logger.info("Trip is allowed to be updated")


def find_trip(external_id: str):
    try:
        connection = db_connection.connect()
        cursor = connection.cursor()
        trip_query = """
        SELECT external_id, created_at FROM trip WHERE external_id = %s
        """
        cursor.execute(trip_query, (external_id,))
        trip = cursor.fetchone()
        return trip
    except(Exception, psycopg2.Error) as error:
        logger.error("Error when attempt to find trip", error)
    finally:
        db_connection.close_connection(connection, cursor)


def lambda_handler(event, context):
    logger.info("Received event: {}".format(event))
    message = json.loads(event['Records'][0]['Sns']['Message'])
    logger.info("Converted message: {}".format(message))
    external_id = message['trip']['externalId']
    trip = find_trip(external_id)
    check_time_limit(trip)
    return {
        'statusCode': 204
    }

Saga Pattern #3: Thinking Serverless

Article Link

service-iam-setup.sh

aws iam create-role \
 --role-name route-check-time-role \
 --assume-role-policy-document file://services/route/check-time/trust-policy.json \
 --profile saga-pattern-serverless
 
 
 aws iam put-role-policy \
 --role-name route-check-time-role \
 --policy-name route-check-time-inline-policy \
 --policy-document file://services/route/check-time/policy.json \
 --profile saga-pattern-serverless

sns-topic-setup.sh

aws sns create-topic \
 --name driver-cancel-trip-trigger-topic \
 --profile saga-pattern-serverless

trust-policy.json

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