Use EventBridge for your event-driven architecture in 8 steps

This article was also presented as a talk at the recently concluded APAC Community Summit 2022 in Bangkok, Thailand

Prerequisites

Before you begin, I suggest you read my introduction to EventBridge.

Scenario

In this demo, we will start with a basic repository that has 3 microservices: orders, points, and notification. The order service marks an order as delivered. The points service award points based on a delivered order, while the notification service sends a simple "your order has been delivered" email. Our baseline repository looks like this:

By the end of the post, we would use EventBridge so that the order service publishes an event once an order is marked as delivered. Then, the points and notifications receive that event. This triggers them to award points and send an email.

Clone the repository and then, run git checkout baseline-two. There are 3 folders inside, representing each of the services. Set up each one by following the setup guide on the README.md of each service.

If you want to know more about how I structured the 3 applications, you can check out my blog post that talks about how we build apps on top of Serverless Framework.

In the next section, let's test out all 5 endpoints by using the Postman collection at the root of the repository. It is ideal to have completed the full setup before proceeding to the next section.

Orders Service

The order service has 2 endpoints. The first endpoint creates an order. It outputs an order_number that we use for the next endpoint.

The second endpoint marks that order as delivered. We get the order_id generated in the first endpoint and place it as input here.

In the response, we see the order successfully marked as delivered.

Points Service

The points service also has 2 endpoints. The first endpoint awards points to a given user. If the user does not exist in the database yet, it creates an entry for it in the DB.

The second endpoint displays the existing points of the user so we can verify if the points did increase.

Notification Service

The notification service only has one endpoint. It takes a message and sends an email to an email address we have set up.

An email is then sent to our designated email:

Our Objective

When you clone the repository, you can deploy all 3 microservices and call the 5 API endpoints separately. But that's not what we are here for. We plan to connect the 3 microservices such that when we call the "Mark as Delivered" endpoint, it sends an "order_delivered" event to EventBridge.

By the end of this post, we will just have to call the order service to mark the order as delivered. Then, via EventBridge, the user will be credited his points and receives a short email.

Step 1

For the serverless.yml of all three services, add the additional permissions and environment variables as seen below:

# modify: order-service/serverless.yml
# modify: notification-service/serverless.yml
# modify: points-service/serverless.yml

provider:
  name: aws
  runtime: python3.8
  lambdaHashingVersion: '20201221'
  stage: ${opt:stage, 'dev'}
  iamRoleStatements:
    - Effect: "Allow"
      Action: "dynamodb:*"
      Resource: "*"
    - Effect: "Allow"
      Action: "sns:*"
      Resource: "*"

    # ADD THIS STATEMENT
    - Effect: "Allow"
      Action: "events:*"
      Resource: "*"
  environment:
    ORDERS_TABLE: ${self:provider.stage}-ecom-orders-table

    # ADD THESE THREE ENV VARS
    EVENT_SOURCE_NAME: ph.ecommerce.com
    EVENT_BUS_NAME:  ${self:provider.stage}-ecom-event-bus
    EVENT_BUS_ARN: arn:aws:events:${self:custom.region}:${self:custom.accountId}:event-bus/${self:provider.environment.EVENT_BUS_NAME}

Step 2

Specifically for the orders service, add the EcommerceEventBus in the resources section. This is a CloudFormation code to programmatically create an EventBridge Event Bus.

# modify: order-service/serverless.yml

resources:
  Resources:
    EcommerceEventBus:
      Type: AWS::Events::EventBus
      Properties: 
        Name: ${self:provider.environment.EVENT_BUS_NAME}

Step 3

Now, we redeploy all 3 services. These should add the environment variables and additional permissions. It should also create the event bus.

ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)

cd order-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

cd points-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

cd notification-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

Step 4

Next, let's focus on the code of the orders service. Let's start by adding the EventbridgeGateway. It uses the boto3 library to send tasks to EventBridge. We placed it in its own gateway to abstract boto3 code from the rest of our application.

# NEW FILE: order-service/gateways/eventbridge_gateway.py

import boto3

class EventbridgeGateway:
    @classmethod
    def put_event(cls, event):
        client = boto3.client('events')

        return client.put_events(Entries=[event])

And then, let's create an EventbridgeEvent model file that stores the business logic of building and sending events to EventBridge. It uses the EventbridgeGateway to send the event.

# NEW FILE: order-service/models/eventbridge_event.py

import json
import os

from gateways.eventbridge_gateway import EventbridgeGateway
from helpers.response import Response


class EventbridgeEvent:
    EVENT_BUS = os.environ.get("EVENT_BUS_NAME")
    SOURCE_URL = os.environ.get("EVENT_SOURCE_NAME")

    def __init__(self, event_name, event_body):
        self.event_name = event_name
        self.event_body = event_body

    def serialize(self):
        json_details = Response.construct_json_body(self.event_body)

        return {
            "Source": self.SOURCE_URL,
            "DetailType": self.event_name,
            "Detail": json_details,
            "EventBusName": self.EVENT_BUS,
        }

    def send(self):
        event_json = self.serialize()
        EventbridgeGateway.put_event(event_json)

Then, on the order model, let's add the send_order_delivered_event() function that creates an EventBridgeEvent and sends it to the event bus.

# UPDATE: order-service/models/order.py

class Order(DynamodbModelBase):
    DYNAMODB_TABLE_NAME = os.getenv("ORDERS_TABLE")

    # ADD THESE TWO FUNCTIONS
    def serialize_for_eventbridge(self):
        return {
            "order_number": self.data["order_number"],
            "user_id": self.data["user_id"],
            "total": self.data["total"]
        }

    def send_order_delivered_event(self):
        event = EventbridgeEvent("order_delivered", self.serialize_for_eventbridge())
        event.send()

Finally, let's call the send_order_delivered_event() in our handler.

As you can see in the code, an order object is instantiated, marked as delivered, saved to DB, and then, we call the send_order_delivered_event() to send an event to EventBridge.

# UPDATE: order-service/handlers/mark_order_as_delivered.py

def handler(event, context):
    try:
        ...

        order = Order.find(primary_key)
        order.mark_as_delivered()
        order.save()

        # ADD THIS LINE
        order.send_order_delivered_event()

Step 5

Now, let's shift our attention to the points service. Let's add a new Lambda function whose event source is our Event Bus. Notice how we added a filter where we only want to receive "order_delivered" events.

# UPDATE: points-service/serverless.yml

functions:
  awardPointsSix:
    handler: handlers/award_points.handler
    events:
      - eventBridge:
          eventBus: ${self:provider.environment.EVENT_BUS_ARN}
          pattern:
            source:
              - ${self:provider.environment.EVENT_SOURCE_NAME}
            detail-type:
              - order_delivered

In the case you want to change the eventBridge parameters we defined above (or you made a mistake somewhere), you have to change the name of the function. It's my sixth attempt at getting it right, hence my function's name is awardPointsSix.

In the handler for our award points endpoint, let's add an if statement so we can accommodate both calls from the API endpoint and EventBridge. Notice the difference in how we extract the payload.

# UPDATE: points-service/handlers/award_points.py

def handler(event, context):
    try:
        print("AND THE EVENT IS NOWWWW!!!!<3 <3")
        print(event)

        if 'detail' in event:
            body = event['detail']

            primary_key = {}
            primary_key["user_id"] = body['user_id']
        else:
            body = json.loads(event['body'])

            primary_key = {}
            primary_key["user_id"] = event['pathParameters']['user_id']

        user = User.create_if_not_exist(primary_key)
        user.award_points(body)
        user.save()

Step 6

Going to the notification service, let's add another Lambda function that receives an event from EventBridge. Notice how it's similar to the original API-based Lambda function, sendEmailNotification, except for the eventBridge part.

# UPDATE: notification-service/serverless.yml

functions:
  sendEmailNotificationSix:
    handler: handlers/send_email_notification.handler
    events:
      - eventBridge:
          eventBus: ${self:provider.environment.EVENT_BUS_ARN}
          pattern:
            source:
              - ${self:provider.environment.EVENT_SOURCE_NAME}
            detail-type:
              - order_delivered

Like with the award points API, let's add this if statement to accommodate both EventBridge event and API calls.

# UPDATE: notification-service/handlers/send_email_notification.py

def handler(event, context):
    try:
        if 'detail' in event:
            body = event['detail']

            primary_key = {}
            primary_key["user_id"] = body['user_id']
        else:
            body = json.loads(event['body'])

            primary_key = {}
            primary_key["user_id"] = event['pathParameters']['user_id']

        topic_arn = os.getenv("SNS_TOPIC_ARN")

        SnsNotificationGateway.publish_message(
            message=body["message"],
            subject="Order marked as delivered",
            topic_arn=topic_arn,
            region="ap-southeast-1"
        )

Step 7

With all that code change, let's deploy once again!

You can also opt to do local testing first via serverless offline. Learn more about that in this blog post

ACCOUNT_ID=$(aws sts get-caller-identity --query "Account" --output text)

cd order-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

cd points-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

cd notification-service
serverless deploy --region ap-southeast-1 --stage dev --accountId $ACCOUNT_ID

Step 8

Now, let's test it out. Check first how many points the user has by calling the "Check User Points" API Endpoint:

Then, create an order and mark it as delivered. When you call the "Check User Points" API endpoint again, the points should have increased on their own:

Also, you should have received an email.

That's it!

You have used EventBridge to connect your event-driven serverless architecture. In the next post, we'll tackle the problems we encountered while using EventBridge in production and how you can address them. Stay tuned!

Photo by Kelly Sikkema on Unsplash