Forem: Elias Brange

6 Ways To Supercharge Your Node.js Serverless Workflow

Elias Brange — Wed, 23 Apr 2025 16:54:49 +0000

The rise of serverless has given us new ways to ship business value faster than ever. It allows us to iterate and build cost-effective solutions in record time. I've worked with serverless for six years and never felt more productive.

But it didn't start that way. When first starting with serverless, I struggled a lot. I was used to running applications and their dependencies locally during development. With serverless, this proved to be a lot more difficult. There are tools like LocalStack, but back then, it wasn't as mature as it is today. I struggled to find a good balance between rapid feedback and confidence that what I built was working.

On one end, you want to unit test as much as possible, since it offers the most rapid feedback. With the many building blocks of serverless, this can quickly evolve into a difficult mocking exercise. It's easy to spend more time crafting mocks than writing actual code.

On the other end, you could embrace the “testing in the cloud” approach, where you deploy your application to AWS to verify that everything is working as expected. This gives you more confidence but results in a slower feedback loop since you'll have to deploy every change.

I've seen many teams struggle when transitioning to serverless, with the two most common concerns being testing and local development. Do you double down on unit tests? Should you forego unit tests and focus on integration/end-to-end tests instead?

Over the years, I've tried multiple approaches, and I think the answer lies somewhere in between. I've found a workflow that I'm happy with, that allows me to iterate with rapid feedback while staying confident that what I'm shipping is working.

In this post, I'll share some tools and techniques I use. I'm writing this from the perspective of a Node.js developer building serverless applications on AWS Lambda. If you use a different language or FaaS platform, some things might not apply, but I hope you get some insights that help you improve your development experience.

Setting the stage

In this blog post, I'll use the following sample architecture as a basis. It's an order service, that allows users to create and retrieve orders. When creating an order, the service will call the product service to check that the included products are valid and available. The service then persists the order and uses DynamoDB streams to trigger a Lambda function that sends an OrderCreated event to EventBridge.

With this architecture in mind, let's look at how my workflow building it might look. I will walk through two different feedback loops, the inner and outer loops. The goal of the inner loop is to provide rapid feedback on isolated parts of the application, while the outer loop gives me confidence that the application works as a whole.

The rapid inner feedback loop

For rapid feedback, unit tests are king. With unit tests, I can test small units of code in isolation. At this stage, I don't care whether everything is wired up correctly. For example, I do not test whether a PutItem to DynamoDB results in an OrderCreated event being sent to EventBridge. Instead, I identify key parts of the application and test those in isolation:

Calling the product service
Sending an OrderCreated event to EventBridge
Saving and retrieving an order from DynamoDB

I'm using three different techniques for these three parts. Let's tackle them one by one.

Mocking HTTP requests with Mock Service Worker

Let's start by looking at the call to the product service. In code, this could look something like this:

const productServiceUrl = process.env.PRODUCT_SERVICE_URL || "https://product-service.com";

export const getProductDetails = async (productId: string) => {
  const res = await fetch(`${productServiceUrl}/products/${productId}`);
  if (res.status === 404) {
    throw new Error("Product not found");
  }
  return res.json();
};

To properly test this, I need to mock the HTTP request. In theory, I could mock fetch directly, but I've found that the tool Mock Service Worker makes it much easier to manage, especially as the number of mocks grows.

MSW intercepts all outgoing HTTP requests and allows me to configure mock responses. I can set up mocks to return a valid response for the product ID p123, and a 404 response for the product ID p456:

import { http, HttpResponse } from "msw";
import { setupServer } from "msw/node";

const handlers = [
  http.get("https://product-service.com/products/p123", () => {
    return HttpResponse.json({
      productId: "p123",
      name: "Rubber Duck",
      price: 1.99,
    });
  }),
  http.get("https://product-service.com/products/p456", () => {
    return HttpResponse.json(
      {
        error: "Product not found",
      },
      { status: 404 },
    );
  }),
];

export const server = setupServer(...handlers);

I can now test that the getProductDetails function handles both the happy path and the error path:

describe("product service", () => {
  beforeAll(async () => {
    // Start MSW
    server.listen();
  });

  it("should return product details", async () => {
    const actual = await getProductDetails("p123");

    expect(actual).toEqual({
      productId: "p123",
      name: "Rubber Duck",
      price: 1.99,
    });
  });

  it("should throw an error when product not found", async () => {
    await expect(getProductDetails("p456")).rejects.toThrow(
      "Product not found",
    );
  });
});

MSW allows me a lot of flexibility when mocking any HTTP request. I can configure mocks to return different responses based on the request path, headers, and more, and I can also use MSW to mock WebSockets and GraphQL requests. It is also easy to use in any project, since it isn't tied to a specific testing framework.

If your application includes HTTP requests to external services, I recommend adding MSW to your test setup. It will save you a lot of time and effort.

Mocking AWS SDK calls

With the call to the product service out of the way, let's tackle the sending of OrderCreated events to EventBridge.

export const sendOrderCreatedEvent = async (order: Order) => {
  await eventBridgeClient.send(
    new PutEventsCommand({
      Entries: [
        {
          Source: "com.example.order",
          DetailType: "OrderCreated",
          EventBusName: "MyBus",
          Detail: JSON.stringify(order),
        },
      ],
    }),
  );
};

Isn't this another HTTP request under the hood that MSW can handle? Well, yes, it is. But it's a bit more complex to mock than the previous example. This is because the AWS SDK does much work under the hood to send the request and parse the response. To mock this, I must know how the underlying request looks to mock it properly with MSW. I'd also have to return a response that matches what the SDK expects, with the correct format, headers, etc.

The fact that different AWS services have wildly different APIs results in mocking quickly becoming an exercise in reimplementing the AWS SDK in MSW mocks.

Instead of MSW, I'm using aws-sdk-client-mock and aws-sdk-client-mock-vitest, which makes it easy to mock and inspect calls made using the AWS SDK.

describe("Events", () => {
  const ebMock = mockClient(EventBridgeClient);

  it("should send an OrderCreated event", async () => {
    const order = {
      orderId: "o123",
      products: [
        {
          productId: "p123",
          quantity: 1,
        },
      ],
    };
    await sendOrderCreatedEvent(order);

    expect(ebMock).toHaveReceivedCommandWith(PutEventsCommand, {
      Entries: [
        expect.objectContaining({
          DetailType: "OrderCreated",
          Detail: JSON.stringify(order),
        }),
      ],
    });
  });
});

With this test, I'm confident that the function sendOrderCreatedEvent correctly sends OrderCreated events with the correct payload to EventBridge. At this stage, I don't care whether the function has permission to send the event. I'll verify that later in the outer feedback loop.

Mocking stateful behavior

Let's move on to the last part, saving and retrieving an order from DynamoDB. But, isn't this another AWS SDK call to mock with aws-sdk-client-mock? Yes, it is. But when working with stateful behavior, such as persisting data, I like to verify that the data model works as expected. For example, does the item I save to DynamoDB have the correct format? Can I read it back? Do my queries return the expected records?

On another note, when working with DynamoDB, I rarely use the AWS SDK directly. I prefer using the ElectroDB library, which is much more pleasant to work with. To use aws-sdk-client-mock together with ElectroDB, I would have to delve into the internals of ElectroDB to figure out how it uses the AWS SDK under the hood.

My approach here blurs the line between unit and integration tests. I could execute the tests against a real DynamoDB table by configuring AWS credentials locally and ensuring a table exists. Since DynamoDB is serverless, it would barely cost anything. What if I use a relational database, Redis, or something else? I prefer an approach that works locally for different stateful services, where I don't have to spin up a live database cluster. I leave that for the outer feedback loop when testing the application as a whole.

Instead, I'm using TestContainers which allows me to spin up a local DynamoDB instance for the duration of my tests, using the official amazon/dynamodb-local image.

const item = new electrodb.Entity(...);

export const getItem = async (id: string) => {
  const response = await item.get({ id }).go();
  return response.data;
};

export const createItem = async (name: string) => {
  const response = await item.create({ name }).go();
  return response.data;
};

export const listItems = async () => {
  const response = await item.scan.go();
  return response.data;
};

export const deleteItem = async (id: string) => {
  const response = await item.delete({ id }).go();
  return response.data;
};

The code shows a simple database implementation using ElectroDB. Let's look at how I can test this, and then we'll take a look at the actual TestContainers setup.

describe("Database", () => {
  beforeEach(async () => {
    await createTable();
  });

  afterEach(async () => {
    await deleteTable();
  });

  it("should create an item", async () => {
    const created = await createItem("test");
    expect(created).toMatchObject({
      id: expect.any(String),
      name: "test",
    });
  });

  it("should get an item", async () => {
    const created = await createItem("test");
    const actual = await getItem(created.id);
    expect(actual).toMatchObject(created);
  });

  it("should return a list of items", async () => {
    const items = await Promise.all([createItem("test1"), createItem("test2")]);
    const actual = await listItems();
    expect(actual.length).toBe(2);
    expect(actual).toEqual(expect.arrayContaining(items));
  });

  it("should delete an item", async () => {
    const created = await createItem("test");
    expect(await getItem(created.id)).toMatchObject(created);
    await deleteItem(created.id);
    expect(await getItem(created.id)).toBeNull();
  });
});

With these tests, I'm confident that my database implementation works as expected. After I create an item, I can retrieve it using the generated id. I know that listing and deleting items work. I create a new table before each test starts to ensure there are no side effects from earlier tests.

This all executes in a local DynamoDB instance running in Docker that holds the state in memory, ensuring the tests execute quickly.

What isn't quick, though, is the startup of the Docker container itself. When developing, I usually start a npm run test:watch process that will re-run my tests whenever I make code changes. If I had to wait 3-7 seconds for the Docker container to start every time I make a change, that would severely slow down the feedback loop.

To circumvent this, I'm using globalSetup in Vitest.

import { defineConfig } from "vitest/config";

export default defineConfig({
  test: {
    globalSetup: ["./test/setup-containers.ts"],
    ...
  },
});

import { GenericContainer, type StartedTestContainer } from "testcontainers";

let dynamodb: StartedTestContainer;

async function startDynamo() {
  return new GenericContainer("amazon/dynamodb-local")
    .withExposedPorts(8000)
    .start();
}

export async function setup() {
  dynamodb = await startDynamo();
  process.env.AWS_ENDPOINT_URL_DYNAMODB = `http://${dynamodb.getHost()}:${dynamodb.getMappedPort(8000)}`;
}

export async function teardown() {
  await dynamodb.stop();
}

Now, whenever I run npm run test or npm run test:watch, Vitest will automatically start the DynamoDB container and expose it on a random port. I then set the AWS_ENDPOINT_URL_DYNAMODB environment variable to override the URL the AWS DynamoDB SDK uses. I will only get a slight delay when starting the test watcher. All subsequent code changes will trigger the tests with no delay.

Summary

Those are the main techniques for mocking in unit tests that I use during development. To recap, I:

Mock HTTP requests with Mock Service Worker
Mock stateless AWS SDK (EventBridge, SQS, etc.) calls with aws-sdk-client-mock
Mock stateful behavior with TestContainers

With this, I'm confident that my code does what it's supposed to do. But I've not tested that the different parts of my application play well together. Let's leave the inner loop and look at integration and end-to-end tests.

The outer feedback loop

While unit tests give me rapid feedback on isolated parts, they cannot tell me if I wired everything correctly. The sample application could be broken in many ways:

The API Gateway doesn't route the request to the Lambda function.
The Lambda functions are missing required permissions.
The DynamoDB table doesn't have its stream enabled.
The product service is not available.
...

To verify that the application works as a whole, I test my application end-to-end in a real environment. This means deploying my application to AWS and running tests, e.g., by sending requests to the API Gateway. This will surface any issues with configuration, permissions, and so on.

Let's say that I locate an issue. I've confirmed that a new order is successfully created in DynamoDB, but no OrderCreated event is sent to EventBridge. What do I do? I could:

Make some changes
Deploy to AWS
Send a request
Check CloudWatch logs
Rinse and repeat

This is a painfully slow feedback loop compared to the rapid inner loop. As an application grows, so does the time it takes to deploy it. Can I speed up this feedback loop?

Debugging in the cloud

When developing, I treat the cloud as an extension of my local development environment. This means that I deploy early and often to get continuous feedback. But, deploying my application on every code change wastes time. Debugging end-to-end issues can devolve into log debugging, where I make a change, deploy, and check the logs for clues. What if I can attach a debugger locally while running the application in the cloud?

The outstanding tool Lambda Live Debugger allows me to do that. I can deploy my application, start the debugger in my IDE, and then step through the code as if it ran on my computer. I can, for example, set a breakpoint in the Lambda function that sends an OrderCreated event. When an item is created in DynamoDB, the stream will trigger the Lambda function, and the debugger will break at the breakpoint. This is especially powerful in event-driven architectures, or complex step function workflows with many Lambda functions.

With LLD, I can get rapid feedback on my changes even when running in the cloud. Changes to the code are reflected immediately, and I can step through the code of my Lambda functions as they are executed using real input events from real AWS services.

Personal Ephemeral environments

All places I've worked have had multiple environments, such as dev, staging, and production. Having only one development environment usually results in questions like "Can I deploy to dev or is anyone using it?" and "Who broke dev?".

One of the key benefits of serverless is that it enables what is generally called ephemeral environments. True serverless services, such as Lambda, DynamoDB, and EventBridge, have a billing model where you only pay for actual usage. This characteristic allows us to deploy multiple development environments without breaking the bank.

When coaching teams on how to get the most out of serverless, I always recommend that every developer on the team deploy a separate personal development environment. This eliminates the problem of a shared development environment being "used" by another developer, and allows team members to work on parallel tasks without interfering with each other.

Ephemeral environments are a requirement for the approach mentioned earlier with LLD. Once, I spent more time than I want to admit wondering why a shared test environment didn't work as expected, and why the CloudWatch logs looked like gibberish. It turned out that someone had been using a live debugger without restoring the functions when they finished. Shared environments don't lend themselves well to live debugging.

To deploy personal development environments, you must ensure that your infrastructure doesn't have resource names that might clash between environments. Take the DynamoDB table in our sample application as an example. If we were to hard-code the table name to OrderTable in our Infrastructure-as-Code, deploying two versions of the environment to the same account would be impossible. If you aren't using an IaC tool that defaults to creating a unique resource name, you have to ensure that uniqueness yourself between environments.

const table = new dynamodb.Table(this, "MyTable", {
  // Do NOT set tableName. Let CDK generate a unique name.
  partitionKey: { name: "pk", type: dynamodb.AttributeType.STRING },
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
});

Another powerful usage of ephemeral environments is in your CI/CD pipeline. For each pull request, you can deploy the application to a new environment, run tests against it, and then tear down the environment again. The ephemeral nature means you can deploy and test pull requests in parallel without conflicts.

Often, embracing ephemeral environments removes the need for shared non-production environments.

What about the product service?

What if the team behind the product service has gone all-in on ephemeral environments and removed all shared environments except production? The order service depends on it, so which product service environment should I use when developing?

Here, I take a similar approach as when unit testing, where I'm using MSW to mock external HTTP requests. I would rather not run MSW inside the Lambda function since that would entail packaging it with non-production code. It would also need a runtime switch to toggle MSW on or off depending on the environment. Instead, I'm using a technique I recently blogged about, with a Rust adoption of WireMock running in a Lambda extension.

With the extension, a WireMock server will be available on, for example, http://localhost:1234 within the Lambda function. I can mock the product service using WireMock stub files.

{
  "request": {
    "method": "GET",
    "urlPath": "/product-service/products/p123"
  },
  "response": {
    "status": 200,
    "jsonBody": {
      "productId": "p123",
      "name": "Rubber Duck",
      "price": 1.99
    }
  }
}

{
  "request": {
    "method": "GET",
    "urlPath": "/product-service/products/p456"
  },
  "response": {
    "status": 404,
    "jsonBody": {
      "error": "Product not found"
    }
  }
}

Then I set the PRODUCT_SERVICE_URL environment variable to http://localhost:1234/product-service, and I'm good to go.

Even if all teams deploy services to shared development and staging environments, I still recommend using the WireMock extension to stub requests to those services. This reduces the risk of another team blocking your progress when they have issues with their non-production environment.

It's also a great way to test how your application behaves when services you depend on are experiencing issues or abnormal latencies. Using WireMock, you can inject faults and latencies and test how your application handles them.

Lastly, it's more or less needed if you have to integrate with a service that doesn't exist in a deployed state yet. As long as an API spec exists, you can start stubbing requests to it with WireMock while the team behind the service is building it.

Summary

To summarize, I use the following strategies to speed up the feedback loop when verifying that my application works as a whole:

I use Lambda Live Debugger to debug my code, even when running in the cloud.
I use personal ephemeral environments to allow team members to work independently.
I run a Rust adoption of WireMock in a Lambda extension to mock external HTTP requests.

Live debugging ensures that I can bypass the lengthy deployment cycle. Ephemeral environments and external services mocks ensure that I'm rarely blocked by another team member or team.

Conclusion

Building with serverless has many benefits, such as allowing you to focus on shipping business value while AWS takes care of the underlying infrastructure. But I've seen developers new to serverless struggle with local development and testing.

If you are used to running your entire application and its dependencies locally, the shift to serverless and running in the cloud can feel like a step back. But it doesn't have to be.

By adopting the techniques and tools in this post, you are well on your way to a fantastic developer experience when building serverless. This workflow offers rapid feedback and the confidence you need to iterate and ship value faster than ever.

Testing Event-Driven Serverless Architectures doesn't have to be difficult. 💥🚀

Elias Brange — Thu, 26 Oct 2023 06:43:40 +0000

1. Introduction

Testing Serverless applications end-to-end is difficult. Testing distributed systems is difficult. But with the right tools, it can be made easier.

Imagine you have an OrderService that emits an OrderCreated event to EventBridge each time an order is created. To test this flow end-to-end, you must subscribe to the EventBridge bus and verify that the correct event has been published.

There is a new player in town. The team at Momento has built a Topics service. I've recently found the time to play around with this service, and I'm impressed.

In this post, I will show you how you can use Momento Topics to test that your application(s) produces the events you expect.

You can find the sample application and a step-by-step guide on GitHub.

2. Momento Topics

Momento Topics is a serverless pub/sub service that makes publishing and subscribing to events simple for any application. This is what Momento themselves say about the service:

Momento Topics is a serverless event messaging service that allows for real-time communication between parts of a distributed application. Instead of spending cycles defining topic resources and dealing with the pain of a tightly-coupled system, push to an unlimited number of channels on the fly without any topic management.

It is a fully managed service that enables a wide range of use cases in distributed systems. For example, you can use it to build an interactive live reaction app to spice up your online presentations.

Being serverless, it also comes with the pay-as-you-go pricing model that we all love and a generous free tier.

Momento has SDKs for a wide range of languages and an HTTP API, which enables the use case in this post. We will integrate EventBridge with this API with the help of API Destinations.

3. Sample application

Our sample application mimics a simple Order Service. It consists of an API Gateway with a single POST / endpoint that triggers a Lambda function. This Lambda function generates a new order with a random ID and publishes an OrderCreated event to EventBridge.

The Lambda function, in this case, is a simple function with the following handler:

export const handler = async (): Promise<APIGatewayProxyResult> => {
  const order = {
    id: ulid(),
    name: 'test order',
  };

  await eventBridgeClient.send(
    new PutEventsCommand({
      Entries: [
        {
          EventBusName: EVENT_BUS,
          Source: 'OrderService',
          DetailType: 'OrderCreated',
          Detail: JSON.stringify(order),
        },
      ],
    }),
  );
  return {
    statusCode: 201,
    body: JSON.stringify(order),
  };
};

When the API receives a POST / request, an event with the following structure is published to EventBridge:

{
  "version": "0",
  "id": "11111111-2222-4444-5555-666666666666",
  "detail-type": "OrderCreated",
  "source": "OrderService",
  "account": "123456789012",
  "time": "2023-10-19T08:00:00Z",
  "region": "eu-west-1",
  "resources": [],
  "detail": {
    "id": "01F9ZQZJZJZJZJZJZJZJZJZJZJ",
    "name": "test order"
  }
}

How can you test this application? In a more real-life scenario, you would probably persist an order to a database. If that's the case, you could verify that the service has persisted the order in the database. Your service might have external consumers that rely on the events it emits. You must ensure that your service publishes the OrderCreated event to EventBridge and has the correct format.

There are different ways to do this. You could integrate EventBridge with services like SNS or AppSync subscriptions and subscribe to those in your tests. But, there is an easier way: Momento Topics.

4. Integrate EventBridge and Momento Topics

The first thing that enables this pattern is the API Destinations feature of EventBridge. API Destinations allow you to send events from EventBridge to an HTTP endpoint. You can configure an API Destination to use Basic, OAuth, and API Key authorization. The authorization configuration, if any, is securely stored in Secrets Manager.

The second thing that enables this pattern is the Momento HTTP API. The API lets you authenticate via an API Key and publish events to a Momento Topic. For a deeper dive into integrating EventBridge and Momento, refer to the official Momento documentation.

The following standalone CloudFormation template shows the required resources to integrate EventBridge and Momento Topics:

AWSTemplateFormatVersion: '2010-09-09'
Description: Momento destination for EventBridge

Parameters:
  EventBusName:
    Type: String
  MomentoEndpoint:
    Type: String
  MomentoAuthToken:
    Type: String
    NoEcho: true
  TopicName:
    Type: String
  CacheName:
    Type: String

Resources:
  DLQ:
    Type: AWS::SQS::Queue

  Connection:
    Type: AWS::Events::Connection
    Properties:
      AuthorizationType: API_KEY
      AuthParameters:
        ApiKeyAuthParameters:
          ApiKeyName: Authorization
          ApiKeyValue: !Sub ${MomentoAuthToken}

  Destination:
    Type: AWS::Events::ApiDestination
    Properties:
      ConnectionArn: !GetAtt Connection.Arn
      HttpMethod: POST
      InvocationEndpoint: !Sub ${MomentoEndpoint}/topics/${CacheName}/${TopicName}
      InvocationRateLimitPerSecond: 300

  TargetRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: sts:AssumeRole
      Path: /service-role/
      Policies:
        - PolicyName: destinationinvoke
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - events:InvokeApiDestination
                Resource: !GetAtt Destination.Arn

  Rule:
    Type: AWS::Events::Rule
    Properties:
      EventBusName: !Ref EventBusName
      EventPattern:
        source:
          - prefix: ''
      State: ENABLED
      Targets:
        - Id: topicPublish-rule
          Arn: !GetAtt Destination.Arn
          RoleArn: !GetAtt TargetRole.Arn
          DeadLetterConfig:
            Arn: !GetAtt DLQ.Arn

The template consists of a few resources:

DLQ: A dead-letter queue to catch errors when publishing events to Momento.
Connection: Specifies the authorization type and parameters to use with a destination endpoint. Here, we specify that it should use API_KEY validation, and we provide the API key value.
Destination: Specifies the endpoint to send events to. Here, we specify the method and endpoint to call. In this case, we construct the URL from the MomentoEndPoint, CacheName, and TopicName parameters.
TargetRole: An IAM role that allows EventBridge to invoke the API destination.
Rule: An EventBridge rule on the EventBusName bus that sends events to the API destination. Here we use the pattern "source": [{"prefix": ""}] to match all events.

With this stack in place, every event sent to the EventBridge bus will be sent to the Momento Topic. We can now use this in our End-to-End tests to verify that the application publishes the events we expect to EventBridge.

5. Test the application End-to-End

Now, let's get to the magic part. With the EventBridge and Momento integration in place, we can now test our application and be confident that it emits the correct event(s). The architecture below shows the flow of events:

A test calls the API Gateway endpoint that triggers the Lambda to send an event.
The API destination relays the event to the Momento topic.
The test receives the event from the Momento topic and verifies that it matches the expected event.

So, we need a test that subscribes to the topic, calls the API, and verifies that the event is received. The test could look like this:

describe('When an order is created', async () => {
  const subscription = await subscribe();
  let order: { id: string; name: string };

  beforeAll(async () => {
    const response = await fetch(API_URL, {
      method: 'POST',
    });
    expect(response.status).toBe(201);

    order = await response.json();
  });

  afterAll(async () => {
    subscription.unsubscribe();
  });

  it('It should publish an OrderCreated event to EventBridge', async () => {
    const message = await subscription.waitForMessageMatching({
      source: 'OrderService',
      'detail-type': 'OrderCreated',
      detail: {
        id: order.id,
        name: order.name,
      },
    });

    expect(message).not.toBeNull();
  }, 5000);
});

The test uses a subscribe function to set up a subscription to the topic. It then calls the API and saves the ID of the created order. It then verifies that it receives an OrderCreated event with the format via the waitForMessageMatching function.

The test explicitly sets a timeout of five seconds. This should be ample time for the service to publish the event and let it propagate to the Momento topic. If no matching event is received within that duration, the test fails.

So, what magic is going on in these functions? Not that much, actually. The Momento SDK is a joy to work with:

import 'dotenv/config';
import {
  TopicClient,
  TopicConfigurations,
  CredentialProvider,
  TopicItem,
  TopicSubscribe,
} from '@gomomento/sdk';
import { ReplaySubject, firstValueFrom } from 'rxjs';
import { filter } from 'rxjs/operators';
import * as _ from 'lodash';

const topicClient = new TopicClient({
  configuration: TopicConfigurations.Default.latest(),
  credentialProvider: CredentialProvider.fromEnvironmentVariable({
    environmentVariableName: 'MOMENTO_API_KEY',
  }),
});
const CACHE_NAME = process.env.CACHE_NAME || '';
const TOPIC_NAME = process.env.TOPIC_NAME || '';

export const subscribe = async () => {
  const messages = new ReplaySubject(100);

  const subscription = await topicClient.subscribe(CACHE_NAME, TOPIC_NAME, {
    onError: (error) => {
      throw error;
    },
    onItem: (item: TopicItem) => {
      try {
        const message = JSON.parse(item.valueString());
        messages.next(message);
      } catch (error) {
        console.log('Error parsing message from Momento Topic', item);
      }
    },
  });

  if (!(subscription instanceof TopicSubscribe.Subscription)) {
    throw new Error('Failed to subscribe to topic');
  }

  const unsubscribe = async () => {
    subscription.unsubscribe();
  };

  const waitForMessageMatching = async (expected: object) => {
    const predicate = (message: unknown) => {
      if (typeof message !== 'object' || message === null) {
        return false;
      }

      return _.isMatch(message, expected);
    };

    const data = messages.pipe(filter((message) => predicate(message)));
    return firstValueFrom(data);
  };

  return {
    unsubscribe,
    waitForMessageMatching,
  };
};

Let's break this down.

We first create a TopicClient and configure it to read the Momento API Key from the environment variable MOMENTO_API_KEY. We also fetch the name of the cache and topic from environment variables.

Inside the subscribe function, we first create an RxJS ReplaySubject. We use this to store incoming events from the Momento topic.

Then, we use the client to subscribe to our topic. In onItem, we parse the incoming event and store it by calling messages.next(message). Every event sent to the topic will be stored during the test.

In waitForMessageMatching, we filter out any messages that do not match the expected event. It returns the first matching event that is received, if any. The test fails if no matching event is received within the test timeout.

Looking back at the actual test case, we are expecting an OrderCreated event from the OrderService that contains the id and name of the order that was created:

const message = await subscription.waitForMessageMatching({
  source: 'OrderService',
  'detail-type': 'OrderCreated',
  detail: {
    id: order.id,
    name: order.name,
  },
});

expect(message).not.toBeNull();

With this, we can be confident that our application works correctly end-to-end.

6. Conclusion

One of the most common arguments against Serverless is that testing is difficult. This partly comes from a misconception that you should test everything locally. With serverless, you should embrace testing in the cloud. This is where the application runs, and this is where you should test it.

Distributed systems are complex to test, no matter if they are built with Serverless technologies or not. But, with the right tools and patterns, you can make it easier. In this post, you have learned how to use EventBridge and Momento Topics to test that your applications produce the events you expect.

You can find the sample application and a step-by-step guide on GitHub.

Don't Lose Your Events! Use the Transactional Outbox Pattern.

Elias Brange — Mon, 23 Oct 2023 07:22:01 +0000

1. Introduction

Events are the key part of an Event-driven Architecture (EDA). They are the mechanism that allows services to communicate with each other in a decoupled way.

For example, an Order Service emits an OrderCreated event when a new order is created. The Shipping Service then consumes this event that starts the shipping process.

But how can you ensure that you can reliably publish events? Take a look at the following code:

exports.handler = async (event, context) => {
  const order = domain.NewOrder(...);

  await persistOrder(order);
  await publishDomainEvent(new OrderCreated(order));
}

This looks straightforward, right? The order service receives a request (or a command) to create a new order. It validates the order, persists it, and publishes a domain event.

When building distributed systems, eight fallacies describe false assumptions that developers often make. One of which is the network is reliable.

Look at the above code again. What happens if the publishDomainEvent call fails? The order will be persisted, but no domain event will be published. If you don't handle this case, you will end up with an inconsistent state. The shipping service will not initiate the shipping process.

There are many solutions of varying complexity to this problem. In this article, you will see how to implement the Transactional Outbox Pattern to solve it in a Serverless environment.

2. What is the Transactional Outbox Pattern?

In the transactional outbox pattern, you persist updates to business entities and any domain events related to those updates in the same transaction. A separate process reads the persisted domain events and publishes them to a message broker. The transactional nature ensures that the entity and related events are either all persisted or none of them are.

In the diagram above, when the Order Service receives a request, it:

Persists any updates to business entities in the Order Table and any related domain events to the Outbox Table in a single transaction.
The Message relay process gets the domain events from the Outbox Table.
The Message relay process publishes the domain events to a message broker.

3. Using DynamoDB Streams for Change Data Capture

DynamoDB Streams is a powerful feature of DynamoDB that lets you capture all item-level changes made to a table. When you enable a stream on a table, you can configure a Lambda function to be invoked on every change to the table. Streams are fully managed by AWS and are a perfect fit to implement the Transactional Outbox Pattern.

Looking back at the code in the Introduction, we were worried about what happens if the publishDomainEvent call fails after the persistOrder call succeeds. DynamoDB streams can help us solve this.

Using streams, the Lambda function only needs to persist the order to the OrdersTable. This change will be sent through the stream to another Lambda function, which publishes the domain events. If there are any errors when publishing the events, the Lambda function will retry the operation until it succeeds or until the records expire.

4. Implementing the Pattern in Serverless Applications

4.1. Naive approach

Let's start with the naive approach. We will use the following architecture:

In this pattern, the CreateOrder Lambda function persists the order to the OrdersTable. The table is configured with a DynamoDB stream that triggers the PublishEvents Lambda function on every change to the table.

The PublishEvents Lambda must parse the change and determine what kind of domain event to send. The code might look something like this:

exports.handler = async (event, context) => {
  for (const record of event.Records) {
    const domainEvent = parseDynamoRecord(record);

    if (domainEvent) {
      await publishDomainEvent(event);
    }
  }
};

const parseDynamoRecord = (record) => {
  if (isOrderCreated(record)) {
    return new OrderCreated(record.dynamodb.NewImage);
  }
  return null;
};

const isOrderCreated = (record) => {
  // Check if the record is an INSERT event for an order.
  return (
    record.eventName === 'INSERT' &&
    record.dynamodb.NewImage &&
    record.dynamodb.NewImage.type === 'ORDER'
  );
};

This might look like a good and simple solution at first. But it has some drawbacks:

Coupling between the PublishEvents Lambda and the OrdersTable

The PublishEvents Lambda function is coupled to the schema of OrdersTable. If you change the schema, you must also change the PublishEvents Lambda function.
Complex logic in PublishEvents to handle different events

If you want to publish other events, such as OrderCancelled and OrderUpdated, you must add complex logic to the PublishEvents Lambda function. This means that you will inevitably have to duplicate the core domain logic since you must correctly determine which domain event to publish depending on what the change contains.
If you are using Single-Table Design in DynamoDB

If you use a single-table design for your DynamoDB table, the complexity in PublishEvents will be even greater. You might have multiple entities in the OrdersTable, such as Orders and OrderItems. The PublishEvents Lambda function must now determine what kind of entity has changed and what type of event to emit for that entity.

There are a lot of advocates for single-table design in DynamoDB. It is, however, not a silver bullet that solves all problems. This article by Pete Naylor tells a different story and is well worth a read.

The above drawbacks are not ideal. They force you to move core domain logic to the outer bounds of your application while simultaneously coupling the logic to your database schema.

There's a better approach.

4.2. Use a Separate Outbox Table and DynamoDB Transactions

This pattern introduces another table, the OutboxTable. It relies on DynamoDB transactions to atomically persist both the entity and domain event in a transaction. This means that either both the entity and domain events are persisted or none of them are.

The PublishEvents Lambda function only relays events from the OutboxTable to EventBridge. The code might look something like this:

exports.handler = async (event, context) => {
  for (const record of event.Records) {
    if (record.eventName !== 'INSERT') {
      // We only want to publish events when they are inserted in the outbox table.
      continue;
    }
    await publishDomainEvent(record.dynamodb.newImage.event);
  }
};

The logic to determine what events to publish is kept within the core domain logic inside the CreateOrder function. The outer bounds of the application are responsible for persisting the order and any domain events in an atomic transaction.

exports.handler = async (event, context) => {
  const { order, events } = domain.NewOrder(...);

  await persistNewOrder(order, events);
}

const persistNewOrder = async (order, events) => {
  await dynamoClient.send(
    new TransactWriteItemsCommand({
      TransactItems: [
        {
          Put: {
            TableName: "OrdersTable",
            Item: order,
          },
        },
        ...events.map((event) => ({
          Put: {
            TableName: "OutboxTable",
            Item: {
              event,
            },
          },
        })),
      ],
    }),
  );
};

This approach has several benefits:

The PublishEvents Lambda is decoupled

The PublishEvents Lambda function is now decoupled from the schema of the OrdersTable. It doesn't need to know how entities are persisted.
The PublishEvents Lambda has low complexity

The PublishEvents Lambda function is now straightforward. It only needs to relay events from the OutboxTable to EventBridge.
Core domain logic doesn't leak

The core domain logic is kept inside the domain layer. It doesn't leak to the outer bounds of the application.
The OutboxTable can support different events and entities

The OutboxTable can support different events and entities. You can add new events and entities without changing the PublishEvents Lambda function.

5. Handling failures

The Transactional Outbox Pattern is all about reliably publishing events. Even though most of the services used are fully managed by AWS, errors can still occur. Let's look at what happens if the PublishEvents Lambda function fails to publish events.

DynamoDB streams send batches of records to the PublishEvents Lambda function. If an invocation of the PublishEvents function fails, the entire batch will be retried until it succeeds or until the records expire.

For example, imagine that a batch of ten records is sent to the PublishEvents Lambda function, and the function fails to publish the 10th event. The entire batch is retried, and this time it succeeds. Events 1-9 will now have been published twice.

There are a couple of different ways to handle this:

Making clients idempotent

Make sure that clients that consume the events implement idempotency into their operations. This means that they must be able to handle the same event being published multiple times. I'd recommend always doing this since many AWS services use at-least-once delivery.

Even if you can guarantee that you only publish an event once, the client could still receive the same event more than once from EventBridge itself.

For a deeper dive on idempotency, refer to this excellent article written by Allen Helton.
Using a batch size of 1

You can configure the Lambda function to only receive one record at a time. If the function fails, it will only retry that single record. This is the simplest option, but it could increase the number of invocations and the cost of the Lambda function. This could also increase latency if you write many events to the OutboxTable.
Bisecting batch failures

A more advanced concept in DynamoDB streams lets you bisect a batch when an invocation fails. When an invocation fails in the middle of a batch, a function can report the last successfully processed record. The batch is then bisected at this record so that only the unprocessed records are retried.

6. Limitations

Since this pattern relies on DynamoDB transactions, it inherits the same limitations. Transactions are limited to 100 items. This means that you cannot publish, say, 150 domain events in a single transaction.

I can't really see a reason for publishing that many events from a single operation, so this shouldn't be a problem in practice.

7. Conclusion

The Transactional Outbox Pattern is a great pattern to publish domain events reliably. DynamoDB streams let you easily implement the pattern in your Serverless applications. Combined with DynamoDB transactions, your domain logic can be kept nicely encapsulated in the domain layer, while the outer parts of the application can be kept as simple as possible.

8. Further reading

Level up your Lambda Game with Canary Deployments using SST

Elias Brange — Mon, 11 Sep 2023 07:52:17 +0000

Have you ever deployed a new version of your Lambda function to production and immediately regretted it? Have you heard people advocating "testing in production" and wondered how that is possible?

In this tutorial, you will learn how to use canary deployments in a Serverless environment with SST to safely expose a small percentage of users to new versions. If sh*t happens, which it inevitably will, automatic rollbacks got your back.

1. How it works

Canary deployments is a deployment strategy that releases a new application version to a small subset of users. Let's say that you deploy a new version of your application, and your new version includes a defect. If you send 100% of traffic to this version, you could break the application for all users.

Canary deployments aim to solve this problem by validating your new version against a small fraction of the traffic. You can, for example, opt to send 5% of traffic to the latest version for a set period while monitoring its behavior. If all metrics look good during this period, you can switch over 100% of traffic to the new version. But, if error metrics for the new version start rising, you can automatically roll back to the previous version.

The simplest way to implement canary deployments of Lambda functions is with Lambda aliases and AWS CodeDeploy.

1.1. Lambda Aliases

AWS Lambda lets you have multiple versions of a Lambda function deployed at the same time. A Lambda alias functions as a pointer to a specific version. Aliases also let you configure a second version and define the percentage of incoming events it should route to each version. CodeDeploy utilizes this feature to control the traffic weighting of the alias during a deployment.

You can configure API Gateways to point to aliases instead of functions. This means that all traffic entering your API Gateway will be routed to the different versions based on the weights in the alias.

1.2. AWS CodeDeploy

AWS CodeDeploy is a service that helps you automate releases to AWS Lambda, AWS ECS, and AWS EC2. To use CodeDeploy with AWS Lambda, you need a few resources:

A CodeDeploy Application
A CodeDeploy Deployment Group
A CodeDeploy Deployment Configuration

CodeDeploy Application

An application is simply a container for deployments and deployment groups.

CodeDeploy Deployment Group

A deployment group defines the target of a deployment. For Lambda deployments, the target is a Lambda alias. It also defines the deployment configuration to use, as well as any alarms that should trigger a rollback.

CodeDeploy Deployment Configuration

A deployment configuration defines the deployment strategy. CodeDeploy comes with a few built-in strategies, but you can also create your own. The built-in strategies for Lambda functions are:

Linear: Shift traffic in equal increments with an equal number of minutes between each increment. For example, 10% every 10 minutes.
Canary: Shift traffic in two increments. For example, first 10% for 5 minutes and then 100% afterward.
All at once: Shift all traffic to the new version immediately.

2. Tutorial

This tutorial assumes that your local environment is configured with AWS credentials. The tutorial uses SST to define and deploy infrastructure. You should be able to accomplish the same with any of the other CloudFormation-based tools (SAM, CDK, etc.).

2.1. Create a new SST Project

Initialize a new SST project. We use the standard/api as a starting point for illustrative purposes.

$ npx create-sst@latest --template standard/api sst-lambda-canary

Open up the generated project in your favorite editor and open the stacks/MyStack.ts file. Start by removing everything from the stack so that you start from a clean slate:

import { StackContext } from 'sst/constructs';

export function API({ stack }: StackContext) {
  // empty stack
}

SST defaults to the us-east-1 region. You can specify another region in sst.config.ts.

2.2. Add a simple Lambda-backed API

Let's add a simple API backed by a Lambda function to the stack. The standard/api template you used comes pre-baked with a simple handler function in packages/functions/src/lambda.ts that you can use for the purpose of this tutorial. This handler will be invoked when a GET request hits the root path (/) of the API Gateway.

In stacks/MyStack.ts, add the following:

import { StackContext, Api, Function as SSTFunction } from 'sst/constructs';
// I usually import Function like above to keep linters happy about shadowing.

export function API({ stack }: StackContext) {
  const func = new SSTFunction(stack, 'MyFunc', {
    handler: 'packages/functions/src/lambda.handler',
  });

  const api = new Api(stack, 'MyApi', {
    routes: {
      'GET /': func,
    },
  });

  stack.addOutputs({
    ApiEndpoint: api.url,
  });
}

Deploy your application to your AWS account using the SST CLI:

$ npx sst deploy --stage prod

SST v2.24.25

➜  App:     sst-test
   Stage:   prod
   Region:  us-east-1
   Account: 123456789012

✔  Building...
...

✔  Deployed:
   API
   ApiEndpoint: https://YOUR_API_ID.execute-api.eu-west-1.amazonaws.com

Test your API to make sure everything is setup correctly:

$ curl https://YOUR_API_ID.execute-api.eu-west-1.amazonaws.com

Hello world. The time is 2023-09-07T19:18:24.360Z

2.3. Create a Lambda Alias

In stacks/MyStack.ts, add a Lambda alias to your Lambda function and update your API Gateway to route traffic to the alias instead:

const func = ...

const alias = func.addAlias('live');

const api = new Api(stack, "MyApi", {
  routes: {
    'GET /': {
      cdk: {
        function: alias
      }
    },
  },
});

Deploy again with npx sst deploy --stage prod to ensure the alias works.

2.4. Create an Alarm

CodeDeploy can automatically roll back a deployment in case one or more specified alarms get triggered during the deployment. To illustrate this, you will add an alarm that triggers when the new Lambda version produces any errors.

It is a bit annoying to have to specify the values of dimensionsMap manually. It would be great if CDK could infer this automatically if you use alias.metricErrors or func.currentVersion.metricErrors. This is reported as a bug in the following GitHub issue.

In the stack, add the following:

import { Alarm } from "aws-cdk-lib/aws-cloudwatch";
import { Duration } from 'aws-cdk-lib/core';
...

export function API({ stack }: StackContext) {
  ...

  const alarm = new Alarm(stack, 'MyAlarm', {
    alarmName: `${func.functionName}-${func.currentVersion.version}-errors`,
    metric: func.metricErrors({
      period: Duration.minutes(1),
      dimensionsMap: {
        FunctionName: func.functionName,
        Resource: `${func.functionName}:${alias.aliasName}`,
        ExecutedVersion: func.currentVersion.version,
      },
    }),
    threshold: 1,
    evaluationPeriods: 1,
  });
}

This alarm will check for any errors in the newly deployed Lambda version. If your current Lambda version is X, a new deploy will create a version X+1 alongside X and send a percentage of traffic to it. Using the dimensions above ensures only the X+1 version can trigger the alarm.

The alarm is named after the function name and version, which ensures that the alarm is recreated during a deployment. If the currently deployed version is experiencing errors, the alarm could be in an alarm state when you deploy a new version, resulting in an instant rollback. Updating the underlying metric of the alarm does not reset the alarm status. This workaround creates a new alarm with the correct state and underlying metric for each deployment.

2.5. Add CodeDeploy Configuration

You must first create a CodeDeploy Application. In your stack, add:

import { LambdaApplication } from 'aws-cdk-lib/aws-codedeploy'
...

export function API({ stack }: StackContext) {
  ...

  const application = new LambdaApplication(stack, 'MyApplication');
}

Next, add a new Deployment Group to your CodeDeploy Application, referencing the Lambda alias and alarm you created earlier:

import {
  LambdaApplication,
  LambdaDeploymentConfig,
  LambdaDeploymentGroup,
} from 'aws-cdk-lib/aws-codedeploy';
...

export function API({ stack }: StackContext) {
  ...

  const deploymentGroup = new LambdaDeploymentGroup(stack, 'MyDeploymentGroup', {
    application,
    alias,
    deploymentConfig: LambdaDeploymentConfig.CANARY_10PERCENT_5MINUTES,
    alarms: [alarm],
  });
}

This creates a CodeDeploy Deployment Group that uses a built-in deployment strategy. The strategy sends 10% of the traffic to the new version and it keeps that weight for a duration of five minutes. If any specified alarm is triggered during this time window, CodeDeploy automatically rolls back the deployment and sends 100% of traffic to the old version. If this happens, CloudFormation will roll back the stack to the previous state.

CloudFormation will be stuck in the UPDATE_IN_PROGRESS state until the CodeDeploy deployment is complete.

2.6. Trying it out

Change the return value of your Lambda function (packages/functions/src/lambda.ts):

export const handler = ApiHandler(async (_evt) => {
  return {
    statusCode: 200,
    body: "Hello there. I'm a canary.",
  };
});

Deploy your stack again with npx sst deploy --stage prod. After some time you should notice the deployment getting stuck at API MyFunc/Aliaslive AWS::Lambda::Alias UPDATE_IN_PROGRESS. This means that the CodeDeploy deployment is in progress.

Hit your endpoint a few times with curl or similar. Most of the responses should be Hello world. The time is ... but you should see a few Hello there. I'm a canary. as well. After five minutes have passed, the deployment should be complete, and all future requests will return the new message.

Let's simulate a failed deployment. Inject an error in your handler:

export const handler = ApiHandler(async (_evt) => {
  throw new Error("Oops!");
  return { ... };
});

Deploy again and hit your endpoint again while the CodeDeploy deployment is in progress. Some of the requests should hit the canary and return a 500 Internal Server Error. Shortly after this happens, the alarm will be triggered and the deployment will be rolled back. All requests you do should now return Hello there. I'm a canary.. The CloudFormation stack will also be rolled back to its previous state.

2.7. Customizing the Deployment Strategy

CodeDeploy comes with a couple of built-in deployment strategies. If these do not fit your needs, you can create a custom deployment config. Imagine you want to send 50% of the traffic to the new version, and let it "bake" for 10 minutes. Simply create a new deployment configuration and update the deployment group:

import {
  LambdaApplication,
  LambdaDeploymentConfig,
  LambdaDeploymentGroup,
  TimeBasedCanaryTrafficRouting
} from 'aws-cdk-lib/aws-codedeploy';
...

export function API({ stack }: StackContext) {
  ...

  const deploymentConfig = new LambdaDeploymentConfig(stack, 'MyDeploymentConfig', {
    trafficRouting: new TimeBasedCanaryTrafficRouting({
      interval: Duration.minutes(10),
      percentage: 50,
    }),
  });

  const deploymentGroup = new LambdaDeploymentGroup(stack, 'MyDeploymentGroup', {
    application,
    alias,
    deploymentConfig,
    alarms: [alarm],
  });
}

2.8. Managing Different Environments

Currently, your application will use the same deployment strategy in all environments. This is probably not what you want. In ephemeral environments and when using SST's Live Lambda Development you most likely want instant deploys without any canary. You can use the stage parameter to conditionally define the deployment configuration to use:

const deploymentGroup = new LambdaDeploymentGroup(stack, 'MyDeploymentGroup', {
  application,
  alias,
  deploymentConfig:
    stack.stage === 'prod'
      ? deploymentConfig
      : LambdaDeploymentConfig.ALL_AT_ONCE,
  alarms: [alarm],
});

3. Conclusion

In this tutorial you have learned how to use Lambda aliases and CodeDeploy to do canary deployments of your Lambda functions. You have also learned how to create custom deployment strategies as well as how to use different strategy for different environments.

With this knowledge, you can improve the robustness and resilience of your serverless architectures, and you are one step closer to testing in production.

How to Separate your Serverless Infrastructure

Elias Brange — Sun, 17 Jul 2022 13:56:48 +0000

Some time ago, you had a brilliant business idea. You started building it on AWS and read that Serverless was the way to go. Good on you! After quickly prototyping an MVP and pitching it to investors, you found yourself surrounded by a team of developers. You worked hard to get the beta version ready. By iterating quickly and listening to customers, the beta got tremendously popular, and you had a successful launch. Money was pouring in, and you felt unstoppable, so you set out to add new features and services to your product to continue to please customers and investors alike.

To move fast, you kept all code and infrastructure in the same repository. Suddenly, something happened. You were not releasing features as quickly as before. As time passed, making changes got more challenging and time-consuming. After continuously adding services and resources, your infrastructure was now a mess. You found yourself with a tangled web of Lambda Functions, API Gateways, Dynamo Tables, SQS Queues, and other resources.

To avoid the tangled web of dependent services, you can separate your infrastructure into more manageable chunks. These chunks can then be deployed individually and have different lifecycles. Even within a single service, say a Lambda-backed API with a DynamoDB table, you might want to separate the stateless API Gateway and Lambda Function(s) from the stateful DynamoDB table.

All of your services are most likely not living in total isolation and will have dependencies on other services and resources. In this article, you will learn different ways to separate and share resources between services in your Serverless infrastructure, with examples for AWS CDK, Serverless Framework, AWS SAM, and Terraform. You can even mix and match frameworks. Perhaps you want to use Terraform for your DynamoDB Table and Serverless framework for your API Gateway and Lambda functions.

1. CloudFormation Outputs

CloudFormation allows you to declare output values that you can import into other stacks. There are some restrictions to keep in mind when using outputs:

The export names of outputs must be unique within a region and AWS account.
You can’t create cross-stack references across AWS regions.
You can’t modify or remove an output value if another stack references it.
You can’t delete a stack if another stack references one of its outputs.

1.1. Exporting outputs

AWS CDK

If you have multiple stacks in the same CDK app, you can directly pass resources between stacks.

Use the CfnOutput construct to define an output.

import { Stack, StackProps, CfnOutput } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';

export class ExportStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const table = new dynamodb.Table(this, 'Table', { ... });

    new CfnOutput(this, 'TableNameOutput', {
      value: table.tableName,
      exportName: 'ExportedTableName',
    });
  }
}

AWS SAM

You define your outputs in the Outputs section of your template.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: SAM Export

Resources:
  Table:
    Type: AWS::DynamoDB::Table
    Properties: ...

Outputs:
  TableName:
    Description: 'DynamoDB Table name'
    Value: !Ref Table
    Export:
      Name: ExportedTableName

Serverless Framework

If you have multiple services in the same Serverless project, you can use Serverless compose to share resources between services.

Non-function resources in Serverless framework is defined using CloudFormation syntax in the resources section. Thus, it looks very similar to the SAM example above.

service: sls-export
provider:
  name: aws

resources:
  Resources:
    Table:
      Type: AWS::DynamoDB::Table
      Properties: ...

  Outputs:
    TableName:
      Description: 'DynamoDB Table name'
      Value: !Ref Table
      Export:
        Name: ExportedTableName

Terraform

Terraform differs from the other frameworks and does not use CloudFormation as an engine to manage your infrastructure. It is thus not possible to create CloudFormation outputs when using Terraform.

1.2. Importing outputs

AWS CDK

You can use Fn.importValue to import the value of an output exported by another stack.

import { Stack, StackProps, Fn } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';

export class ImportStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const tableName = Fn.importValue('ExportedTableName');

    new lambda.NodejsFunction(this, 'MyFunction', {
      ...
      environment: {
        TABLE_NAME: tableName,
      },
    });
  }
}

AWS SAM

You can use the intrinsic function ImportValue to import the value of an output exported by another stack.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: SAM Import

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
      ...
      Environment:
        Variables:
          TABLE_NAME: !ImportValue 'ExportedTableName'

Serverless Framework

You can import exported outputs with the ImportValue intrinsic function. For non-exported outputs, you can use the syntax ${cf:stackName.outputName}.

service: sls-import
provider:
  name: aws

functions:
  hello:
    handler: src/functions/app.lambdaHandler
    environment:
      TABLE_NAME: !ImportValue ExportedTableName

Terraform

Fetching an exported CloudFormation outputs is done with the aws_cloudformation_export data source.

data "aws_cloudformation_export" "table_name" {
  name = "ExportedTableName"
}

2. SSM Parameters

You can also use the Systems Manager Parameter Store to pass values between stacks by creating a parameter and dynamically reference it in a CloudFormation stack or Terraform configuration.

Compared to CloudFormation outputs, SSM does not give you the same guardrails. Nothing stops you from deleting a parameter, even if another stack dynamically references it. By doing so, the next update of the other stack will result in errors. You will also most likely break the service if you also remove the resources referenced by the SSM parameter.

So why would you use SSM parameters over CloudFormation outputs? If you want to mix Infrastructure-as-Code tools and, for example, export values from Terraform and use them in a CloudFormation-based tool.

You can also use SSM parameters during runtime instead of deploy time. You can use the AWS SDK and dynamically fetch parameters in your Lambda functions. It gives you the possibility of updating parameters without having to redeploy consuming functions. However, it makes it harder to follow the principle of least privilege. You will have to use less granular permissions to account for different parameter values (if you, for example, have created a new DynamoDB from a snapshot and intend to switch the parameter to point to the restored table).

Be aware that reading during runtime will result in some overhead and longer execution times in cold starts. It will also increase the execution time of warm lambdas if you do the fetch inside the handler code.

2.1. Exporting parameters

AWS CDK

Create an SSM parameter with the StringParameter construct.

import { Stack, StackProps } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as ssm from 'aws-cdk-lib/aws-ssm';

export class ExportStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const table = new dynamodb.Table(this, 'Table', { ... });

    new ssm.StringParameter(this, 'SSMParam', {
      parameterName: '/some/path/tableName',
      type: ssm.ParameterType.STRING,
      stringValue: table.tableName,
    });
  }
}

AWS SAM

Create an SSM parameter with the AWS::SSM::Parameter resource.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: SAM Export

Resources:
  Table:
    Type: AWS::DynamoDB::Table
    Properties: ...

  SSMParam:
    Type: AWS::SSM::Parameter
    Properties:
      Type: String
      Name: '/some/path/tableName'
      Value: !Ref Table

Serverless Framework

Creating a parameter in Serverless framework is done the same way as in AWS SAM.

service: sls-export
provider:
  name: aws

resources:
  Resources:
    Table:
      Type: AWS::DynamoDB::Table
      Properties: ...

  SSMParam:
    Type: AWS::SSM::Parameter
    Properties:
      Type: String
      Name: '/some/path/tableName'
      Value: !Ref Table

Terraform

Use the aws_ssm_parameter in the AWS provider to create a parameter.

resource "aws_dynamodb_table" "dynamo_table" {
  name = "dynamodb-table"
  ...
}

resource "aws_ssm_parameter" "ssm_dynamo_table_name" {
  name  = "/some/path/tableName"
  type  = "String"
  value = aws_dynamodb_table.dynamo_table.name
}

2.2. Importing parameters

AWS CDK

Use StringParameter.valueForStringParameter to fetch the value of a parameter.

import { Stack, StackProps } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as ssm from 'aws-cdk-lib/aws-ssm';
import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';

export class ImportStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const tableName = ssm.StringParameter.valueForStringParameter(
      this,
      '/some/path/tableName',
    );

    new lambda.NodejsFunction(this, 'MyFunction', {
      ...
      environment: {
        TABLE_NAME: tableName,
      },
    });
  }
}

AWS SAM

You can use SSM parameters in your template with the '{{resolve:ssm:parameter-name:version}}' pattern, where version is optional. If you do not specify a version, CloudFormation will use the latest version of the parameter whenever you create or update the stack.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: SAM Import

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
      ...
      Environment:
        Variables:
          TABLE_NAME: '{{resolve:ssm:/some/path/tableName}}'

Serverless Framework

Reading parameters in Serverless framework is done with the ${ssm:/path/to/param} syntax:

service: sls-import
provider:
  name: aws

functions:
  hello:
    handler: src/functions/app.lambdaHandler
    environment:
      TABLE_NAME: ${ssm:/some/path/tableName}

Terraform

You can use the data source aws_ssm_parameter to fetch a parameter value.

data "aws_ssm_parameter" "table_name" {
  name = "/some/path/tableName"
}

2.3. Reading at Runtime in a Lambda function

The following code uses the AWS SDK for TypeScript to fetch parameter values. Be aware that the function IAM role will need permission to fetch the parameter.

import { APIGatewayProxyEvent, APIGatewayProxyResult } from 'aws-lambda';
import { SSMClient, GetParameterCommand } from '@aws-sdk/client-ssm';

const ssmClient = new SSMClient({});
const input = { Name: '/some/path/tableName' };
const command = new GetParameterCommand(input);
const parameterPromise = ssmClient.send(command);

export const lambdaHandler = async (event: APIGatewayProxyEvent): Promise<APIGatewayProxyResult> => {
    const parameter = await parameterPromise;
    ...

3. Summary

You have now learned how to separate your infrastructure stacks into more manageable chunks with the use of CloudFormation Outputs and/or SSM Parameters. You've seen examples for AWS CDK, AWS SAM, Serverless Framework and Terraform.

Now go build something awesome!

Lawn weeds and Technical debt

Elias Brange — Mon, 30 May 2022 15:51:52 +0000

I was spending some time this weekend (as usual) clearing out weeds on my lawn when I suddenly realized how similar weeds are to technical debt.

Clearing weeds as they appear and before they spread requires time and patience, but it is manageable if tackled continuously.

The same applies to technical debt, which is much cheaper and faster to pay back in the early stages of a software project.

Let weeds spiral out of control, and the cost and time to fix the problem increase to a point where you begin to wonder whether it’s more cost-effective to start over from scratch.

This is also true for technical debt.

How do you prefer to tackle technical debt?

PSA: Your dev.to post will most likely get stolen by this blog with no link to the original source or other form of credit.

Elias Brange — Fri, 29 Apr 2022 07:03:35 +0000

When investigating the SEO of my own blog I stumbled upon my own article, copied from dev.to to the "blog" https://lzomedia.com/.

The blog seems to gather articles from different webpages, and for some of them it includes backlinks to the original source. However, for dev.to posts, there is only a footnote at the bottom of the page stating Source: Dev Community. No link to the original source. No canonical URL. No name. Nothing.

EDIT: The blog owner even seems to be a member of this community...

.ltag__user__id__130694 .follow-action-button { background-color: #0030b3 !important; color: #ffffff !important; border-color: #0030b3 !important; }

StefanFollow

Creator of @lzomedia I am a 30 years old Romanian freelance developer who likes beautiful code.

Some examples

lzomedia:
https://lzomedia.com/blog/how-i-automatically-generate-grocery-shopping-lists-with-trello-aws-lambda/

original:

How I automatically generate grocery shopping lists with Trello & AWS Lambda

Elias Brange for AWS Community Builders ・ Apr 26 ・ 8 min read

#aws #serverless #tutorial #python

lzomedia:
https://lzomedia.com/blog/add-a-smart-back-to-top-button-to-your-blog-%f0%9f%91%93%f0%9f%94%9d/

original:

Add a smart 'back to top' button to your website 👓🔝

Rob OLeary ・ Apr 27 ・ 9 min read

#webdev #tutorial #javascript #beginners

lzomedia:
https://lzomedia.com/blog/steps-to-delete-aws-resources-ec2-vpc-using-ansible/

original:

Steps To Delete AWS Resources (EC2 & VPC) Using Ansible

Nurul Ramadhona for AWS Community Builders ・ Apr 27 ・ 3 min read

#aws #ec2 #vpc #ansible

lzomedia:
https://lzomedia.com/blog/what-does-your-dream-work-setup-look-like/

original:

What does your dream work setup look like?

Tina Huynh ・ Apr 28 ・ 1 min read

#workstations #setup #discuss

How I automatically generate grocery shopping lists with Trello & AWS Lambda

Elias Brange — Tue, 26 Apr 2022 09:16:42 +0000

I use Trello for all kinds of personal stuff, one of which is keeping track of recipes and grocery lists. After planning which meals to cook and buy groceries for, a lot of manual work goes into checking every recipe and adding the ingredients to the grocery shopping list. Manual stuff is tedious, so I automated it with AWS Lambda. Read on to find out how!

The Trello board

NOTE: Our Trello board is in Swedish, so the screenshots will be in Swedish as well.

We have a Trello board that contains a lot of recipes grouped into lists, such as Pasta Dishes, Soups & Stews, and so on. We also have a list of groceries that needs to be bought, so that we always have the shopping list available when shopping.

In the above picture, you can get a glimpse of the Trello board, with each card containing a picture of the dish as well as a link to the external site where the recipe can be found.

What I wanted to automate was the following:

Add a button to each card, that when clicked, adds the recipe to the Recipes to buy list (named Recept att handla in my board).
Add a button to the board, that when clicked, reads through the Recipes to buy list and adds the ingredients (and amount of each) to the Grocery Shopping List (named AUTO Handlingslista in my board).

Below you can see the completed automation in action:

Automation setup

Trello part 1

I started by adding two new lists to my board, one to hold recipes that should be bought, and one for the generated grocery shopping list.

I then proceeded to add a checklist to recipe cards. This checklist contains the ingredients required for a single serving of the recipe. For example, if a recipe requires one onion for four servings, I would add onion:0.25. Here I had to be consistent between recipes so that the units used for ingredients would be the same across all recipes.

The unit for onion here is number of onions, while for pasta I use gram and for cooking cream I use deciliter.

For ingredients where I do not care about the unit and just need to make sure that I buy it if needed, such as cooking oil and certain spices, I set the amount needed to 0.

With the checklist(s) completed, I went on and created a Card button in the Automation menu. When clicked, this button triggers an action that adds a new card to the recipes to buy list. The created card gets the following title ANTAL_PORTIONER:{cardidlong}:{cardname}. The ANTAL_PORTIONER (number of servings in Swedish) is a placeholder that needs to be changed to an actual number before generating the shopping list.

The cards now look like this:

Clicking the button creates a new card in the recipes to buy list, and I then edit the text ANTAL_PORTIONER in the title to the number of servings I need to purchase.

I also needed to get an API key and token for the Trello API, as described here.

I also needed to fetch the IDs of both the recipes to buy list, the grocery shopping list, and the board itself. This was easiest done by creating a new card in each of the lists, clicking on them and then adding .json to the end of the URL in the browser. From there I could then extract the idList and idBoard variables.

AWS part

This whole thing really started with me wanting to find a use case for the newly released function URLs of AWS Lambda. So I decided to go with FastAPI in a Lambda function (as I usually do).

Since the API should be reachable from Trello I decided to keep the function URL open to the public. To secure it, I added code to the Lambda function that validates the X-Api-Key header in incoming requests.

Secrets

To not require hard-coded secrets in the Lambda function, I added the following parameters to the AWS Systems Manager Parameter Store:

/trellomat/api_key: Trello API Key
/trellomat/token: Trello token
/trellomat/board_id: ID of the Trello board
/trellomat/shopping_list_id: ID of the grocery shopping list
/trellomat/recipe_list_id: ID of the recipes to buy list
/trellomat/aws_api_key: The API key that is required in the X-Api-Key header

Lambda code

Folder structure

The code examples below are structured as follows:

TrelloMatApi/
├── template.yml
├── api-function
│   ├── requirements.txt
│   ├── src
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── trello.py
│   │   ├── utils.py

requirements.txt

The Lambda function requires the following libraries.

mangum
fastapi
pydantic
py-trello
aws-lambda-powertools

init.py

The handler function is a FastAPI application wrapped by a Mangum adapter. The API includes a single route POST /generate, that neither requires nor returns any payload.

Authorization is done with the X-Api-Key header, which is validated by the auth method. The valid key is fetched from AWS Systems Manager Parameter Store and cached between invocations.

from fastapi import FastAPI, HTTPException, Header, Depends
from mangum import Mangum
from aws_lambda_powertools.utilities import parameters
from src.utils import logger, tracer
from src import trello

app = FastAPI()
API_KEY = parameters.get_parameter("/trellomat/aws_api_key")


async def auth(x_api_key: str = Header(...)):
    if x_api_key != API_KEY:
        raise HTTPException(status_code=401, detail="Unauthorized")


@app.post("/generate", status_code=204, dependencies=[Depends(auth)])
def generate():
    trello.generate()


handler = Mangum(app)
handler.__name__ = "handler"
handler = tracer.capture_lambda_handler(handler)
handler = logger.inject_lambda_context(handler, clear_state=True)

utils.py

This is a simple file that sets up a logger and tracer using AWS Lambda Power Tools.

from aws_lambda_powertools import Logger, Tracer


logger: Logger = Logger()
tracer: Tracer = Tracer()

models.py

This is where I define my ingredients.

In the INGREDIENTS dict, I specify ingredients their units as well as which category the ingredient is part of.

The Category decides which label an ingredient gets, and it is also used for sorting the generated shopping list. The default values in the Ingredient class are there to not break the API whenever an ingredient is added on the Trello side before adding it to the Lambda function itself.

from enum import Enum
from pydantic import BaseModel


class Category(Enum):
    GREEN = 0
    FRIDGE = 1
    BREAD = 2
    FROZEN = 3
    DRY = 4
    OTHER = 5
    NONE = 6


class Ingredient(BaseModel):
    name: str
    unit: str = "NOUNIT"
    cat: Category = Category.NONE
    amount: int = 0


INGREDIENTS = {
    "vitlök": {  # garlic
        "unit": "klyftor",  # cloves 
        "cat": Category.GREEN,
    },
    "gul lök": {  # onion
        "unit": "st",  #  pcs
        "cat": Category.GREEN,
    },
    "matlagningsgrädde": {  # cooking cream
        "unit": "dl",
        "cat": Category.FRIDGE,
    },
    ...
}

trello.py

This is where the magic happens. First, all required information is fetched from AWS Systems Manager Parameter Store.

We then initialize a TrelloClient from the py-trello library, as well as instantiate objects for the board and lists.

The generate function then:

Archives all existing cards in the grocery shopping list.
For each card in the recipes to buy list (which has the format {servings}:{recipe_card_id}:{recipe_name}), it fetches the card with the recipe_card_id.
On this card, it looks for the checklist named Ingredienser.
For each checklist item, it adds the required ingredient amount to the ingredients dict. It also adds the ingredient unit and category from the model defined earlier.
The ingredients dict is then sorted by category.
For each ingredient in the ingredients dict, it creates a new card in the grocery shopping list.

from aws_lambda_powertools.utilities import parameters
from trello import TrelloClient
from src.utils import logger
from src.models import INGREDIENTS, Ingredient, Category


TOKEN = parameters.get_parameter("/trellomat/token")
API_KEY = parameters.get_parameter("/trellomat/api_key")
BOARD_ID = parameters.get_parameter("/trellomat/board_id")
RECIPE_LIST_ID = parameters.get_parameter("/trellomat/recipe_list_id")
SHOPPING_LIST_ID = parameters.get_parameter("/trellomat/shopping_list_id")
CHECKLIST_KEY = "Ingredienser"

client = TrelloClient(api_key=API_KEY, token=TOKEN)
board = client.get_board(board_id=BOARD_ID)
recipe_list = client.get_list(list_id=RECIPE_LIST_ID)
shopping_list = client.get_list(list_id=SHOPPING_LIST_ID)
labels = board.get_labels()


def _get_label(cat: Category) -> list:
    try:
        return [labels[cat.value]]
    except IndexError:
        return []


def _archive_old_cards():
    logger.info("Clearing AUTO shopping list")
    shopping_list.archive_all_cards()


def _get_ingredient_list() -> list[Ingredient]:
    logger.info("Getting ingredients")

    cards = {card.id: card for card in board.get_cards()}
    ingredients = {}

    for card in recipe_list.list_cards_iter():
        servings, ref_card_id, name = card.name.split(":")
        ref_card = cards[ref_card_id]

        logger.info(f"Reading recipe for {name}")

        card_ingredients = next(
            (cl for cl in ref_card.checklists if cl.name == CHECKLIST_KEY), None
        )

        if not card_ingredients:
            logger.warning(f"No checklist named {CHECKLIST_KEY} found on card {name}")
            continue

        for item in card_ingredients.items:
            name, amount = item["name"].split(":")

            if name not in ingredients:
                ingredients[name] = Ingredient(name=name, **INGREDIENTS.get(name, {}))

            ingredients[name].amount += int(servings) * float(amount)

    # Return list of Ingredients sorted by category
    return sorted(ingredients.values(), key=lambda item: item.cat.value)


def _create_cards(ingredients: list[Ingredient]):
    for ingredient in ingredients:
        if ingredient.amount == 0:
            title = f"{ingredient.name}"
        elif ingredient.amount == int(ingredient.amount):
            title = f"{int(ingredient.amount)} {ingredient.unit} {ingredient.name}"
        else:
            title = f"{ingredient.amount:.2f} {ingredient.unit} {ingredient.name}"

        logger.info("Adding card: %s", title)
        shopping_list.add_card(title, labels=_get_label(ingredient.cat))


def generate():
    _archive_old_cards()
    ingredients = _get_ingredient_list()
    _create_cards(ingredients)
    logger.info("Grocery list generated successfully")

template.yml

I deployed the Lambda function with AWS SAM. The SAM template looks like the following:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Trello Mat API

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      MemorySize: 256
      Timeout: 180
      Tracing: Active
      FunctionUrlConfig:
        AuthType: NONE
      CodeUri: api-function
      Handler: src.handler
      Runtime: python3.9
      Policies:
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/aws_api_key"
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/token"
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/api_key"
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/board_id"
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/recipe_list_id"
        - SSMParameterReadPolicy:
            ParameterName: "trellomat/shopping_list_id"
      Environment:
        Variables:
          POWERTOOLS_SERVICE_NAME: TrelloMatAPI

Outputs:
  FunctionUrl:
    Description: URL of the Function
    Value: !GetAtt ApiFunctionUrl.FunctionUrl

Trello part 2

With the API complete, I created a new Board button in the Automation menu. When clicked, a post request is sent to the Lambda function URL, with the API key specified in the X-Api-Key header. The Lambda function then reads the recipes to buy list and generates a grocery shopping list.

Final result

Lo and behold, the final result! After adding a few recipes to the recipes to buy list, I can now automatically generate a grocery shopping list.

Have you built anything similar to aid you in your regular day-to-day tasks? Let me know in the comments!

Secure AWS deploys from Github Actions with OIDC

Elias Brange — Wed, 13 Apr 2022 09:26:19 +0000

Long gone are the days when you had to keep long-lived access keys in your CI/CD pipelines to deploy to AWS. Learn how to use OIDC (OpenID Connect) to securely deploy to AWS from Github Actions, and how to use GitHub Environments to secure deployments to specific AWS environments.

Introduction

Managing access from your CI/CD systems to your cloud environments in a secure manner can often be a tedious challenge. For a long time, one common way to do so for AWS was to set up an IAM user for this purpose and then store the access keys for that user in e.g. GitHub secrets for use with GitHub Actions. This required a lot of manual work, such as setting up different users for different projects, rotating keys after a certain period, etc.

By utilizing OIDC you can configure AWS to trust GitHub as a federated identity provider, and then use ID tokens in Github Actions workflows to authenticate to AWS and access resources. You can create separate IAM roles for different purposes and allow workflows to assume those roles in a granular way.

Each job in a GitHub Actions workflow can request an OIDC token from GitHub's OIDC provider. The contents of the token are described well in the GitHub documentation. Depending on the job's configuration, as well as what event triggered the workflow, the token will contain a different set of claims. One important claim is the sub claim, which will be used in our IAM Role trust policies to grant permission for workflows to use a role.

For example, if a workflow is triggered by a push to the main branch of a repository named repo-org/repo-name, the subject claim will be set to repo:repo-org/repo-name:ref:refs/heads/main. If a job references a GitHub environment named Production, the subject claim will be set to repo:repo-org/repo-name:environment:Production.

For a full list of possible subject claims, check the GitHub documentation.

Example setup

In this example, we will use two different S3 buckets to mimic two different AWS environments. We will create two different workflows, one that will be triggered by pushes to the main branch, and another that will be triggered by pull requests.

On pull requests, we want to be able to read our environments. In a real example, this could perhaps be a terraform plan job that could visualize any proposed changes to the environment. We will also deploy to the development environment from pull requests, to speed up the feedback loop. We do, however, NOT want to deploy to the production environment from pull requests. We also need to make sure that contributors cannot update the workflow as part of the pull request to include a deployment to the production environment.

For pushes to main, we want to be able to read and deploy to both environments. We will also require manual approval from a repository admin to deploy to the production environment.

In the example workflows, read and deploy operations will be demonstrated by download and upload operations to the respective S3 buckets.

1. Create buckets

Start by creating two buckets in your account. I will refer to these below as YOUR_DEV_BUCKET and YOUR_PROD_BUCKET.

2. Add GitHub as an identity provider

To be able to authenticate with OIDC from GitHub you will first need to set up GitHub as a federated identity provider in your AWS account. To do that, navigate to the AWS IAM console and click on Identity Providers on the left-hand side. Then, click on the Add provider button.

For Provider type, select OpenID Connect.
For Provider URL, enter https://token.actions.githubusercontent.com
Click on Get thumbprint to get the thumbprint of the provider
For Audience, enter sts.amazonaws.com

3. Create roles and policies

Now that GitHub is set up as an identity provider, it is time to create the roles and policies that will be assumed by the respective workflows. You will create three roles:

Reader role: Will have read permissions on both the buckets. This role will be available both for pull_request events as well as push events to the main branch.
Dev Deploy role: Will have both read and write permissions to the dev bucket. This role will require a workflow to use the Development GitHub environment.
Prod Deploy role: Will have both read and write permissions to the prod bucket. This role will require a workflow to use the Production GitHub environment.

Reader role

Start by creating a new role that will be used to read from the buckets. The role should be able to be assumed by both workflows triggered by the pull_request and push events. To allow this, the role will need to have the following trust policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::0123456789:oidc-provider/token.actions.githubusercontent.com"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "ForAllValues:StringEquals": {
          "token.actions.githubusercontent.com:sub": [
            "repo:ORG_OR_USER_NAME/REPOSITORY:pull_request",
            "repo:ORG_OR_USER_NAME/REPOSITORY:ref:refs/heads/main"
          ],
          "token.actions.githubusercontent.com:aud": "sts.amazonaws.com"
        }
      }
    }
  ]
}

You need to replace the ORG_OR_USER_NAME and REPOSITORY with your own values. For my example repository eliasbrange/aws-github-actions-oidc it would be "repo:eliasbrange/aws-github-actions-oidc:pull_request"

Next, add an IAM policy to this role that grants the role permission to read from the buckets.

{
  "Statement": [
    {
      "Action": [
        "s3:ListBucket"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_DEV_BUCKET",
        "arn:aws:s3:::YOUR_PROD_BUCKET"
      ]
    },
    {
      "Action": [
        "s3:GetObject"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_DEV_BUCKET/*",
        "arn:aws:s3:::YOUR_PROD_BUCKET/*"
      ]
    }
  ],
  "Version": "2012-10-17"
}

Development Deploy Role

Create another role that will be used to "deploy" to the development environment, in this case by uploading a file to the development bucket. This role should require the workflow to use a GitHub Environment named Development, so it will require the following trust policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::0123456789:oidc-provider/token.actions.githubusercontent.com"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "ForAllValues:StringEquals": {
          "token.actions.githubusercontent.com:sub": "repo:ORG_OR_USER_NAME/REPOSITORY:environment:Development",
          "token.actions.githubusercontent.com:aud": "sts.amazonaws.com"
        }
      }
    }
  ]
}

Give it the following IAM policy:

{
  "Statement": [
    {
      "Action": [
        "s3:ListBucket"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_DEV_BUCKET"
      ]
    },
    {
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_DEV_BUCKET/*"
      ]
    }
  ],
  "Version": "2012-10-17"
}

Production Deploy Role

Create the final role, which will be used for production deploys. It should look similar to the development role, but with some changes to the permissions.

Add the following trust policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::0123456789:oidc-provider/token.actions.githubusercontent.com"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "ForAllValues:StringEquals": {
          "token.actions.githubusercontent.com:sub": "repo:ORG_OR_USER_NAME/REPOSITORY:environment:Production",
          "token.actions.githubusercontent.com:aud": "sts.amazonaws.com"
        }
      }
    }
  ]
}

Give it the following IAM policy:

{
  "Statement": [
    {
      "Action": [
        "s3:ListBucket"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_PROD_BUCKET"
      ]
    },
    {
      "Action": [
        "s3:GetObject",
        "s3:PutObject"
      ],
      "Effect": "Allow",
      "Resource": [
        "arn:aws:s3:::YOUR_PROD_BUCKET/*"
      ]
    }
  ],
  "Version": "2012-10-17"
}

4. Create GitHub Environments

In your GitHub repository, create two environments, Development and Production. For Development, we will not add any safeguards so that we can deploy to it directly from pull requests. However, for the Production environment, we will add a few protection rules.

In the environment configuration under Environment protection rules, tick the Required reviewers box and add your own GitHub username.
In the environment configuration under Deployment branches, pick Selected branches in the dropdown and add main as an allowed branch.

The above rules will prevent contributors to modify the actual workflow as part of a pull request to deploy to production. It will also force workflows that target the Production environment to require manual approval.

5. Add GitHub secrets

While you could use secrets scoped to each environment, I want to show that even if the workflow has access to the secret, it will not be able to assume roles it shouldn't have access to.

In your GitHub repository, add the following secrets:

DEV_DEPLOY_ROLE: The ARN of your development deploy role.
PROD_DEPLOY_ROLE: The ARN of your production deploy role.
READ_ROLE: The ARN of your reader role.

6. Create workflows

Time to create the deployment workflows. You will create a total of three workflows, one for pull requests, one for pushes to main, as well as a bonus workflow for visualizing which workflow event triggers that has access to which IAM roles.

The workflows will both try to download a file named README.md from your buckets, as well as upload a file from the repository root named README.md to the buckets. Either change the workflows to use another file or make sure that you have a README.md file in your repository root. To allow the first run of the workflows to succeed, you also need to upload said file to both buckets.

Pull request workflow

Add the following workflow to your repository:

name: "Pull Request"

# The workflow should only trigger on pull requests to the main branch
on:
  pull_request:
    branches:
      - main

# Required to get the ID Token that will be used for OIDC
permissions:
  id-token: write

jobs:
  read-dev:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.READ_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_DEV_BUCKET/README.md README.md
        shell: bash

  write-dev:
    runs-on: ubuntu-latest
    needs: read-dev
    environment: Development
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.DEV_DEPLOY_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp README.md s3://YOUR_DEV_BUCKET/README.md
        shell: bash

  read-prod:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.READ_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_PROD_BUCKET/README.md README.md
        shell: bash

This is a very basic workflow that:

First downloads the README.md file from the Development bucket.
Then uploads the README.md from the repository to the Development bucket.
Finally downloads the README.md file from the Production bucket.

Not a very exciting, nor useful, workflow, but it demonstrates the usage of different roles and environments.

In both read-dev and read-prod there is no environment specified, so the ID token in these jobs will have a subject claim of repo:ORG_OR_USER_NAME/REPOSITORY:pull_request. They are also both using the READ_ROLE secret, which should point to the ARN of your reader IAM role.

In write-dev, there is environment: Development specified, which means that the ID token will have a subject claim of repo:ORG_OR_USER_NAME/REPOSITORY:environment:Development. The job uses the DEV_DEPLOY_ROLE secret instead of READ_ROLE.

Push to `main` workflow

Add the following workflow to your repository:

name: "Push"

# The workflow should only trigger on push events to the main branch
on:
  push:
    branches:
      - main

# Required to get the ID Token that will be used for OIDC
permissions:
  id-token: write

jobs:
  read-dev:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.READ_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_DEV_BUCKET/README.md README.md
        shell: bash

  write-dev:
    runs-on: ubuntu-latest
    needs: read-dev
    environment: Development
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.DEV_DEPLOY_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp README.md s3://YOUR_DEV_BUCKET/README.md
        shell: bash

  read-prod:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.READ_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_PROD_BUCKET/README.md README.md
        shell: bash

  write-prod:
    needs: [read-prod, write-dev]
    runs-on: ubuntu-latest
    environment: Production
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets.PROD_DEPLOY_ROLE }}
          role-session-name: OIDCSession

      - run: aws s3 cp README.md s3://YOUR_PROD_BUCKET/README.md
        shell: bash

Almost the same as the previous workflow, but with the addition of a write-prod job.

Download README.md from the Development bucket.
Upload README.md from the repository to the Development bucket.
Download README.md from the Production bucket.
Upload README.md from the repository to the Production bucket.

read-prod uses the same setup as read-dev, i.e. no environment and both use the READ_ROLE secret. The write-prod job is similar to the write-dev job, but uses environment: Production and the PROD_DEPLOY_ROLE secret.

Bonus workflow

This workflow is a bonus workflow that uses a matrix strategy to show the result of different combinations of subject claim, environment and role.

Add the following workflow to your repository:

name: "Bonus"

on:
  pull_request:
    branches:
      - main

  push:
    branches:
      - main

permissions:
  id-token: write

jobs:
  read-dev:
    strategy:
      fail-fast: false
      matrix:
        environment:
          - ""
          - Development
          - Production
        role_secret:
          - READ_ROLE
          - DEV_DEPLOY_ROLE
          - PROD_DEPLOY_ROLE
    runs-on: ubuntu-latest
    environment: ${{ matrix.environment }}
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets[matrix.role_secret] }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_DEV_BUCKET/README.md README.md
        shell: bash

  write-dev:
    strategy:
      fail-fast: false
      matrix:
        environment:
          - ""
          - Development
          - Production
        role_secret:
          - READ_ROLE
          - DEV_DEPLOY_ROLE
          - PROD_DEPLOY_ROLE
    runs-on: ubuntu-latest
    environment: ${{ matrix.environment }}
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets[matrix.role_secret] }}
          role-session-name: OIDCSession

      - run: aws s3 cp README.md s3://YOUR_DEV_BUCKET/README.md
        shell: bash


  read-prod:
    strategy:
      fail-fast: false
      matrix:
        environment:
          - ""
          - Development
          - Production
        role_secret:
          - READ_ROLE
          - DEV_DEPLOY_ROLE
          - PROD_DEPLOY_ROLE
    runs-on: ubuntu-latest
    environment: ${{ matrix.environment }}
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets[matrix.role_secret] }}
          role-session-name: OIDCSession

      - run: aws s3 cp s3://YOUR_PROD_BUCKET/README.md README.md
        shell: bash

  write-prod:
    strategy:
      fail-fast: false
      matrix:
        environment:
          - ""
          - Development
          - Production
        role_secret:
          - READ_ROLE
          - DEV_DEPLOY_ROLE
          - PROD_DEPLOY_ROLE
    runs-on: ubuntu-latest
    environment: ${{ matrix.environment }}
    steps:
      - name: Checkout
        uses: actions/checkout@v3

      - name: Configure AWS Credentials
        uses: aws-actions/configure-aws-credentials@v1
        with:
          aws-region: eu-west-1
          role-to-assume: ${{ secrets[matrix.role_secret] }}
          role-session-name: OIDCSession

      - run: aws s3 cp README.md s3://YOUR_PROD_BUCKET/README.md
        shell: bash

This workflow will be triggered by both pushes to main and pull requests. It will then try to perform reads and writes on both buckets with different combinations of environment and role_secret.

Results

With all the workflows pushed to main in your repository, create a new branch and create a pull request. If everything works correctly, the workflow named Pull Request should execute successfully. The Bonus workflow should fail for a lot of combinations (and it might take some time to fail due to the back-off strategy of the configure AWS credentials action).

If you merge the pull request, the workflow named Push should execute successfully.

Workflow combinations

Looking at the Bonus workflow we should be able to see which combinations of workflow event, environment and role that is possible with the OIDC setup we have. The working combinations should also be different depending on whether the workflow started from a push to main or a pull request.

For each of the 4 jobs, we should have 9 different combinations. These combinations are:

#	Environment	Role
1	NONE	READ_ROLE
2	NONE	DEV_DEPLOY_ROLE
3	NONE	PROD_DEPLOY_ROLE
4	Development	READ_ROLE
5	Development	DEV_DEPLOY_ROLE
6	Development	PROD_DEPLOY_ROLE
7	Production	READ_ROLE
8	Production	DEV_DEPLOY_ROLE
9	Production	PROD_DEPLOY_ROLE

Pull request

If we start by looking at a pull request, the workflow should have access to the READ_ROLE when the environment is NONE (which means that the subject claim will be repo:ORG_OR_USER_NAME/REPOSITORY:pull_request). This should allow the workflow to read both development and production buckets when using the combination of no environment and READ_ROLE.

It should also be able to use the Development environment since there are no protection rules for that environment. This should allow the workflow to both read and write to the development bucket when the environment is Development and role is DEV_DEPLOY_ROLE.

Due to the environment protection rule on Production, the workflow should not be able to use the Production environment at all when triggered by pull requests.

read-dev job

#	Environment	Role	Result
1	NONE	READ_ROLE	Success
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Success
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: environment protection rule
8	Production	DEV_DEPLOY_ROLE	Fail: environment protection rule
9	Production	PROD_DEPLOY_ROLE	Fail: environment protection rule

write-dev job

#	Environment	Role	Result
1	NONE	READ_ROLE	Fail: insufficient permissions
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Success
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: environment protection rule
8	Production	DEV_DEPLOY_ROLE	Fail: environment protection rule
9	Production	PROD_DEPLOY_ROLE	Fail: environment protection rule

read-prod job

#	Environment	Role	Result
1	NONE	READ_ROLE	Success
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Fail: insufficient permissions
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: environment protection rule
8	Production	DEV_DEPLOY_ROLE	Fail: environment protection rule
9	Production	PROD_DEPLOY_ROLE	Fail: environment protection rule

write-prod job

#	Environment	Role	Result
1	NONE	READ_ROLE	Fail: insufficient permissions
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Fail: insufficient permissions
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: environment protection rule
8	Production	DEV_DEPLOY_ROLE	Fail: environment protection rule
9	Production	PROD_DEPLOY_ROLE	Fail: environment protection rule

Push

Continuing to the workflow when triggered by a push to main, we should be able to again use the READ_ROLE when the environment is NONE. This should give read access to both buckets for that combination.

For writing to development, a combination of Development environment with DEV_DEPLOY_ROLE is required.

Finally, to deploy to production, a combination of Production environment with PROD_DEPLOY_ROLE is required. This should also trigger a manual approval step to continue the deployment.

read-dev job

#	Environment	Role	Result
1	NONE	READ_ROLE	Success
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Success
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: invalid claim
8	Production	DEV_DEPLOY_ROLE	Fail: invalid claim
9	Production	PROD_DEPLOY_ROLE	Fail: invalid claim

write-dev job

#	Environment	Role	Result
1	NONE	READ_ROLE	Fail: insufficient permissions
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Success
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: invalid claim
8	Production	DEV_DEPLOY_ROLE	Fail: invalid claim
9	Production	PROD_DEPLOY_ROLE	Fail: invalid claim

read-prod job

#	Environment	Role	Result
1	NONE	READ_ROLE	Success
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Fail: insufficient permissions
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Success
8	Production	DEV_DEPLOY_ROLE	Fail: invalid claim
9	Production	PROD_DEPLOY_ROLE	Fail: invalid claim

write-prod job

#	Environment	Role	Result
1	NONE	READ_ROLE	Fail: insufficient permissions
2	NONE	DEV_DEPLOY_ROLE	Fail: invalid claim
3	NONE	PROD_DEPLOY_ROLE	Fail: invalid claim
4	Development	READ_ROLE	Fail: invalid claim
5	Development	DEV_DEPLOY_ROLE	Fail: insufficient permissions
6	Development	PROD_DEPLOY_ROLE	Fail: invalid claim
7	Production	READ_ROLE	Fail: invalid claim
8	Production	DEV_DEPLOY_ROLE	Fail: invalid claim
9	Production	PROD_DEPLOY_ROLE	Success

Conclusions

Starting from this example, you should now be able to use GitHub OIDC as a federated identity in your own GitHub Actions workflows to get rid of long-lived credentials once and for all. You have learned how to:

Enable federated identity for GitHub on the AWS side
Define which repositories and workflow events are allowed to access your roles
Keep your production environments secure by requiring additional steps

Hope you learned a thing or two. Now go build something awesome.

GitHub repository

I have a companion repository available on GitHub for this blog post, where you can find the workflows themselves, as well as terraform configuration for setting up the AWS side.

Observability Best Practices when running FastAPI in a Lambda

Elias Brange — Mon, 11 Apr 2022 07:47:52 +0000

Serverless makes it incredibly easy to get an API up and running in record time. But how do you achieve a high level of observability in a distributed architecture? In this post, we will be looking at instrumenting logging, metrics, and tracing capabilities for a FastAPI application using AWS Lambda Powertools for Python.

This guide assumes that you know how to build and deploy a SAM application using the AWS SAM CLI. If you need a refresher on that, please refer to my previous blog post.

0. Table of Contents

1. What is observability?
2. Sample FastAPI Application
3. Checking the baseline
4. AWS Lambda Powertools Python
5. Logging
6. Metrics
7. Tracing
8. Conclusion

1. What is observability?

The rise of cloud and serverless computing has made a big impact on how we build, scale, and manage applications. Developers can now independently build and deploy applications with unmatched agility and speed. With distributed applications becoming more and more common, we must take care to not lose observability in our systems and services. Before long, we will get to a point where a single request traverses our architecture through any number of functions, containers, queues, and other services.

By implementing best practices when it comes to logging and metrics, as well as being able to trace a request through the entire system, we can gain a better understanding of how our applications are performing. The goal of observability is to instrument our applications so that when an issue occurs somewhere in our distributed system, we can quickly locate the root cause of the issue. Good observability also gives great insights into how our applications and services are being used, valuable insights that can guide future business decisions.

Observability can be achieved in several ways, and in this article you will learn how to instrument a FastAPI application, running inside a Lambda function, using the AWS Lambda Powertools for Python library. We will focus on implementing three different capabilities in our application.

Structured Logging where we add contextual information to logs, such as request and correlation IDs, Lambda and FastAPI context, service name, exception information, and more.
Metrics so that we can track how our application is used and how it is performing.
Tracing so that we can trace requests as they travel through our systems.

2. Sample FastAPI Application

In this example we will use a simple FastAPI application, a Pets API that allows you to add, remove, list, fetch and update pets that are stored in DynamoDB. We will use a REST API Gateway to act as a proxy in front of the application.

As shown in the AWS Docs, HTTP APIs do not (at the time of writing) support tracing with AWS X-ray, which is why we have to use a REST API instead.

Example files

The required files to follow along in this example are available on GitHub, or here on my blog.

3. Checking the baseline

Before we add more advanced features to the application, let's deploy the API in its current state to establish a baseline. While testing the API, I have used this Python script. The script uses boto3 to fetch the API URL from the CloudFormation Stack outputs. It then runs a sequence of requests towards the API, touching all the endpoints. You could also issue requests with cURL, through the Swagger UI available at https://API_GATEWAY_URL/STAGE_NAME/docs, or with something like Postman. The choice is yours.

After making a few requests, navigate to the CloudWatch console and go to Logs Insights. You should see a log group called /aws/lambda/FUNCTION_NAME in the Select log group(s) dropdown.

The resulting log records contain almost no useful information, which is because right now we are only using simple print(...) statements in the code. Let's change that, shall we?

4. AWS Lambda Powertools Python

Lambda Powertools Python is a library that comes fully packed with utilities that make it easy to adopt best practices when it comes to the observability of AWS Lambda functions. We will focus on the three core utilities, which are: Logger, Metrics and Tracer.

First, in requirements.txt, add the library.

mangum
fastapi
boto3
pydantic
+ aws-lambda-powertools

In the application code, you will need to import and initialize the three utilities. In a new file example/src/app/utils.py, add the following:

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit  # noqa: F401


logger: Logger = Logger()
metrics: Metrics = Metrics()
tracer: Tracer = Tracer()

5. Logging

Let's start by adding some structured logging to the application. We will use the Logger utility from the Lambda Powertools library for this.

Upgrade from print statements

In the dynamo.py file, add from .utils import logger and change all the print(...) statements to logger.info(...). After a few requests, the experience in CloudWatch should have gotten a tad better.

Here we can see the power of structured logging. CloudWatch has identified fields such as level, location, message, service, and so on. We can now use these fields to filter queries. If we wanted to see only error logs, we could add a filter to the query:

fields @timestamp, message
| sort @timestamp desc
| limit 200
| filter level = "ERROR"

What is that service_undefined value in the service field you ask? That's right, we forgot one thing in the SAM template. We can control what service name Lambda Powertools uses by setting environment variables in a function. If you have a lot of different services, being able to easily filter logs by service name will be critical when you are troubleshooting a problem in production.

Function:
  Type: AWS::Serverless::Function
  Properties:
    ...
    Environment:
      Variables:
+       POWERTOOLS_SERVICE_NAME: FastAPIPowerToolsExample
+       POWERTOOLS_METRICS_NAMESPACE: FastAPINamespace
        TABLE_NAME: !Ref Table
    ...

That wasn't so hard right? Just a few lines of code, and we now have nicely structured logs in the FastAPI application. Let's explore some more features in the Lambda Powertools Logger.

Lambda context

Powertools makes it easy to add information from the Lambda context to logs with the inject_lambda_context decorator, like this:

@logger.inject_lambda_context
def handler(event, context):
    ...

But we do not have a handler function, do we? We have a Mangum object wrapping the FastAPI application. Luckily, the Mangum object acts as a handler function, so we can just add the following in example/src/app/__init__.py:

from .utils import logger  # add this import

...

handler = Mangum(app)
# add this line
handler = logger.inject_lambda_context(handler, clear_state=True)

Here, we set the parameter clear_state=True to clear the state on each invocation. This is useful if you want to ensure that the logs are not polluted with the state from previous invocations.

So, one more line of code. What did that line give us in the CloudWatch console?

Some information about the Lambda context, such as function name, memory configuration, as well as whether the invocation required a cold start or not. Nice!

Correlation ID

Correlation IDs can be used to uniquely identify a request as it traverses multiple services. This is a key component to being able to correlate logs from different services in distributed systems.

In our example, the API will accept an optional X-Correlation-Id header, and if it is not present, it will use the request ID from the Lambda context. The correlation ID can then be added to requests towards downstream services (if any), to be able to get a complete view of the request flow. The Logger utility includes two helper functions to handle correlation IDs, logger.set_correlation_id("ID") and logger.get_correlation_id().

Since we want to extract the correlation ID in every request, as well as return it in every response, we will implement this using a FastAPI middleware.

In example/src/app/__init__.py, add the following:

from fastapi import FastAPI, HTTPException, Request
...

@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    # Get correlation id from X-Correlation-Id header
    corr_id = request.headers.get("x-correlation-id")
    if not corr_id:
        # If empty, use request id from aws context
        corr_id = request.scope["aws.context"].aws_request_id

    # Add correlation id to logs
    logger.set_correlation_id(corr_id)

    response = await call_next(request)

    # Return correlation header in response
    response.headers["X-Correlation-Id"] = corr_id
    return response

Hit the API with a few requests. If you supply a X-Correlation-Id header in the request, you should see the same X-Correlation-Id in the response. If you do not supply one, one will be generated for you. If the application makes further requests to downstream services, the correlation ID could be retrieved with logger.get_correlation_id() and passed on to the downstream service. Navigate to CloudWatch and take a look at the shiny new correlation_id field.

FastAPI context

Let's add some information about the FastAPI request context to the logs. On the top of my head, it would be nice to be able to filter logs by the request path, request route, and the request method. Since we have the Request object in the middleware created above, couldn't we add this functionality there? We can't. This is because middlewares are executed before any routing is made in the FastAPI application. So if we want to add the routes as they are declared in the application, such as /pets/{pet_id}, we need to use a custom APIRoute class.

Add a new file, example/src/app/router.py, and add the following:

from fastapi import Request, Response
from fastapi.routing import APIRoute
from typing import Callable
from .utils import logger


class LoggerRouteHandler(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def route_handler(request: Request) -> Response:
            # Add fastapi context to logs
            ctx = {
                "path": request.url.path,
                "route": self.path,
                "method": request.method,
            }
            logger.append_keys(fastapi=ctx)
            logger.info("Received request")

            return await original_route_handler(request)

        return route_handler

Here we add a few fields to the logger so that e.g. every future logger.info(...) call includes those fields. If you do not fully understand how the custom APIRoute works, please read the FastAPI documentation. He explains it a lot better than I possibly could.

Now, to use the custom APIRoute class, we need to add the following in example/src/app/__init__.py:

...
from .router import LoggerRouteHandler  # add this import


app = FastAPI()
# add this line
app.router.route_class = LoggerRouteHandler

...

Let's see what we have in CloudWatch this time.

Look at that beauty. We now have filterable fields for the method, route, and actual path. Want to get all logs that occurred when someone fetched a specific pet with ID 123456? Simply add a filter to the query like | filter fastapi.method = "GET" and fastapi.path = "pets/123456". Want to see all logs from all calls to the DELETE /pets/{pet_id} route? You know what to do.

Exceptions

The Logger utility can also help us understand and debug errors easier by logging exceptions with logger.exception(...). To illustrate how Lambda Powertools can help us here, let us first add an endpoint in which we inject an error. Add the following code to example/src/app/__init__.py:

@app.get("/fail")
def fail():
    some_dict = {}
    return some_dict["missing_key"]

Call the endpoint, you should receive a 500 Internal Server Error. Head over to CloudWatch and see how the exception was logged.

Looks like we are missing a lot of information here, such as correlation ID, route, service, and all the other fields we added previously. Let's fix that, by adding a custom exception handler.

Creating an exception handler that catches all exceptions does not seem to be fully supported. I've tried just creating an exception handler for exceptions of type Exception, and it seems to enter the handler fine and the logging works. However, it does not catch the exception fully, so the exception is still propagated and finally caught in the Mangum event loop. It also does not enter the middleware that sets the correlation ID in the response.

The issue seems to be resolved by adding a starlette.exceptions.ExceptionMiddleware, though I'm unsure if this can have any unforeseen side effects. Use at your own risk! :)

Further information on FastAPI GitHub Issue and Starlette GitHub Issue:

In example/src/app/__init__.py, add the following:

from fastapi import FastAPI, HTTPException
# add the following imports
from fastapi.responses import JSONResponse
from starlette.exceptions import ExceptionMiddleware

...

app = FastAPI()
app.router.route_class = LoggerRouteHandler
# add the below code
# +++
app.add_middleware(ExceptionMiddleware, handlers=app.exception_handlers)


@app.exception_handler(Exception)
async def unhandled_exception_handler(request, err):
    logger.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

# +++

Call the /fail endpoint again and head over to CloudWatch.

Now we have a lot of relevant information in logs whenever an unhandled exception occurs, which makes it easy to find what caused a request to fail if you for example have the correlation ID at hand.

The logs are now a lot more useful, and we did not need to add that much code to the application. Most of the code resides in either middleware, exception handlers, or the custom APIRoute class. The changes to the actual business logic have been minimal, which shows that Lambda Powertools can easily be added to existing applications to improve logging practices.

6. Metrics

Let's explore the next core utility in Lambda Powertools, the Metrics utility. This utility lets you easily push metrics to CloudWatch by taking care of all the necessary boilerplate. It works asynchronously by using Amazon CloudWatch Embedded Metrics Format, by logging the metrics to stdout. It also aggregates all metrics from each invocation to save on the number of calls to CloudWatch.

There is some terminology to be aware of here. CloudWatch metrics are grouped in containers called namespaces. If an application comprises multiple services, you could for example use the same namespace for all of them to group all the metrics for that application. Metrics can also have dimensions, which are key-value pairs added as metadata to metrics which allows you to filter and aggregate metrics depending on the dimension values.

The default configuration for the Metrics utility utilizes the two environment variables POWERTOOLS_METRICS_NAMESPACE and POWERTOOLS_SERVICE_NAME, where the former specifies the metric namespace and the latter adds a dimension service=POWERTOOLS_SERVICE_NAME to all metrics.

Instrumenting

If you followed along with the guide so far you should already have a metrics object in example/src/app/utils.py. There we used the default configuration, which sets the namespace and service dimension from environment variables. You could also specify those explicitly, like this:

from aws_lambda_powertools import Metrics

metrics = Metrics() # Sets metric namespace and service dimension via environment variables
metrics = Metrics(namespace="Namespace", service="ServiceA") # Sets metric namespace to "Namespace" and service dimension to "ServiceA"

If you want to add another dimension to all metrics, such as environment, you can do so with the set_default_dimensions method:

from aws_lambda_powertools import Metrics

metrics = Metrics()
metrics.set_default_dimensions(environment="development")

Now, to enable the functionality in the application, add the following to example/src/app/__init__.py:

...
# Add metrics to the following import
from .utils import logger, metrics

...

handler = Mangum(app)
handler = logger.inject_lambda_context(handler, clear_state=True)
# Add the following (last to properly flush metrics):
handler = metrics.log_metrics(handler, capture_cold_start_metric=True)

Make sure that the log_metrics decorator is added last to the handler function.

Adding metrics

Let's add the following metrics to the application:

Number of times a pet is created
Number of times an unhandled exception occurs

Created pets metric

To add a metric that counts how many times a pet is created, we can add the following to the POST /pets route. We add it after the call to DynamoDB, to only record a metric if the pet was created successfully.

@app.post("/pets", status_code=201, response_model=models.PetResponse)
def post_pet(payload: models.CreatePayload):
    res = dynamo.create_pet(kind=payload.kind, name=payload.name)
    # Add the following line
    metrics.add_metric(name="CreatedPets", unit=MetricUnit.Count, value=1)
    return res

Unhandled exception metric

To count the number of unhandled exceptions, we can add the following to the exception handler:

@app.exception_handler(Exception)
async def validation_exception_handler(request, err):
    # add the following line
    metrics.add_metric(name="UnhandledExceptions", unit=MetricUnit.Count, value=1)
    logger.exception("Unhandled exception")
    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})

Using different dimensions

When using CloudWatch EMF, all metrics in a document must have the same dimensions. So, all calls to add_metric will generate metrics with identical dimensions, service plus any additional default dimensions you've added. To add different dimensions for specific metrics, we need to use single_metric. We might for example want to have a metric named RequestCount, that has both service and route dimensions.

First, in example/src/app/utils.py, update the following import to include single_metric:

from aws_lambda_powertools import Logger, Metrics, Tracer
# Add single_metric to the following import
from aws_lambda_powertools.metrics import MetricUnit, single_metric  # noqa: F401

Then, in example/src/app/router.py, add the following code:

from fastapi import Request, Response
from fastapi.routing import APIRoute
from typing import Callable
# Add single_metric to the following import
from .utils import metrics, MetricUnit, single_metric


class LoggerRouteHandler(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def route_handler(request: Request) -> Response:
            # Add fastapi context to logs
            ctx = {
                "path": request.url.path,
                "route": self.path,
                "method": request.method,
            }
            logger.append_keys(fastapi=ctx)
            logger.info("Received request")

            # Add the following:
            # +++
            # Add count metric with method + route as dimension
            with single_metric(name="RequestCount", unit=MetricUnit.Count, value=1) as metric:
                metric.add_dimension(name="route", value=f"{request.method} {self.path}")
            # +++

            return await original_route_handler(request)

        return route_handler

CloudWatch showtime

Let's hit the API a few times on different routes, as well as the /fail endpoint. Then, navigate to CloudWatch and click on All metrics on the left-hand side. Then, click on the Namespace you specified. You should see the following.

First, click on function_name, service. Here you should see a metric that counts the number of times a function has been subject to a cold start which is automatically added by the Lambda Powertools metrics utility when you add the capture_cold_start_metric=True parameter.

Next, click on service under the namespace. Here you will be able to see the number of pets that have been created, as well as the number of unhandled exceptions.

Finally, check out route, service. Here you should have metrics describing the total amount of requests to each route.

This showcases how easy it is to add metrics to a FastAPI Lambda application when using Lambda Powertools to take care of the heavy lifting.

7. Tracing

Now, let's move on to the final core utility, the Tracer. The utility makes it easy to instrument your Lambda function to capture traces and send them to AWS X-Ray. AWS X-Ray is a distributed tracing system that can help you analyze and debug distributed applications, and provides a way to track requests as they travel through your different services.

Powertools also has the convenience of auto-patching modules supported by X-Ray, such as boto3. This will automatically create segments in your traces whenever you use boto3 to call DynamoDB or another AWS service from within your Lambda function, and likewise for other supported modules such as requests.

Instrumenting

First, we need to enable tracing on the Lambda function and API Gateway. In the SAM template, add the following:

Function:
  Type: AWS::Serverless::Function
  Properties:
+   Tracing: Active
    MemorySize: 128
...

RestApi:
  Type: AWS::Serverless::Api
  Properties:
    StageName: Prod
+   TracingEnabled: true

Then, as with the other utilities, we need to decorate our handler. Add the following to example/src/app/__init__.py:

...
# Add tracer to the following import
from .utils import tracer, logger, metrics, MetricUnit

...

handler = Mangum(app)

# Add the following lines
# +++
handler.__name__ = "handler"  # tracer requires __name__ to be set
handler = tracer.capture_lambda_handler(handler)
# +++

handler = logger.inject_lambda_context(handler, clear_state=True)
handler = metrics.log_metrics(handler, capture_cold_start_metric=True)

If you want to measure a specific method call and add it to your traces, you can decorate functions with the capture_method decorator. For illustrative purposes, add the following to example/src/app/dynamo.py:

...
# Add the following import
from .utils import tracer

...

@tracer.capture_method  # add this
def create_pet(kind: str, name: str) -> dict:
...

@tracer.capture_method  # and this
def get_pet(pet_id: str) -> dict:
...

@tracer.capture_method  # this as well
def update_pet(pet_id: str, kind: str = None, name: str = None):
...

@tracer.capture_method  # and this
def list_pets(next_token: str = None) -> dict:
...

@tracer.capture_method # and finally this
def delete_pet(pet_id: str):
...

Deploy the API, and send a couple of request to it. Mix it up a little and hit the /fail endpoint a few times, as well as try to generate some 4xx responses by trying to access non-existent pets, or create pets with invalid payloads. Go to the CloudWatch console and click on Service Map on the left-hand side.

Here you can see that AWS X-Ray generated a service map of our entire architecture. We can see that a client has made a request to the API Gateway, which forwarded the request to the Lambda service, which in turn invoked a Lambda function, which then made further requests to the DynamoDB Table. As the system grows, the service map will be an invaluable tool to get a full picture of the entire system. For each node in the map, it is also possible to see the percentage of successful requests, average latency, and other useful metrics.

You can also drill down into individual traces. Below you can see timeline for a request to create a pet.

Advanced features

There are a lot of advanced features in the tracer utility, such as adding metadata and annotations, recording responses, and capturing exceptions.

For example, when we raise PetNotFoundError, it will be visible in the relevant segment details.

We could also add the correlation ID as an annotation, to allow us to filter and query traces by correlation ID. In the add_correlation_id middleware, add tracer.put_annotation(key="correlation_id", value=corr_id):

@app.middleware("http")
async def add_correlation_id(request: Request, call_next):
    # Get correlation id from X-Correlation-Id header
    corr_id = request.headers.get("x-correlation-id")
    if not corr_id:
        # If empty, use request id from aws context
        corr_id = request.scope["aws.context"].aws_request_id

    # Add correlation id to logs
    logger.set_correlation_id(corr_id)

    # Add correlation id to traces
    tracer.put_annotation(key="correlation_id", value=corr_id)

    response = await call_next(request)

    # Return correlation header in response
    response.headers["X-Correlation-Id"] = corr_id
    return response

Now, by drilling down into the ##handler segment, you should see the correlation ID in the correlation_id annotation.

If you want to find traces for a specific ID, you can use the query functionality in the console.

8. Conclusion

That's all for now. Hopefully, you have learned a thing or two about how you can use the Lambda Powertools library to implement best practices when it comes to observability for FastAPI applications (and Lambda functions in general). AWS CloudWatch and AWS X-Ray are wonderful tools to help you analyze and monitor serverless applications. Lambda Powertools makes it incredibly easy to start using those services, allowing you to focus on your application logic while leaving the implementation of metrics, logs, and traces to the library.

AWS Recipe: Build an Asynchronous Serverless Task API

Elias Brange — Mon, 28 Mar 2022 18:12:04 +0000

In this article, you will learn how to build an asynchronous serverless task API from scratch on AWS using Lambda, DynamoDB, API Gateway, SQS, and SNS. The Lambda functions will be implemented in Python, and the REST API will use the FastAPI framework. The entire application will be deployed using AWS SAM.

0. Table of Contents

1. Introduction
2. Recipe ingredients
3. Requirements
4. Build the API stack
5. Build the Handler stack(s)
6. Showtime
7. Cleaning up
8. Potential improvements
9. Conclusion

1. Introduction

This article will guide you through the process of building an asynchronous serverless task API on AWS. With complete code examples, you will learn how to:

Run FastAPI in a Lambda function behind an API Gateway using DynamoDB for storage.
Using DynamoDB streams with filters together with a Lambda function to publish events to SNS.
Implementing a fan-out pattern with SNS and SQS to distribute tasks to different handlers.
Using Systems Manager Parameter Store to share variables between CloudFormation stacks.

We will first build a stack that includes an API that allows clients to create and list tasks. This API will be deployed to AWS Lambda behind a HTTP API Gateway. The application will store tasks in DynamoDB, and whenever a new task is created it will be sent via a DynamoDB Stream to another Lambda that will publish the task to SNS.

In another stack, we will create handlers where each handler will be responsible for processing a specific type of task. Each handler will comprise of an SQS queue that will receive tasks from SNS, and a Lambda function that will process the task and report the task status back to the REST API.

2. Recipe ingredients

API Gateway
Lambda
DynamoDB
DynamoDB Stream
SNS
SQS
FastAPI

3. Requirements

To follow along in this article, you will need:

Python
AWS SAM CLI

4. Build the API stack

Let's begin by setting up everything required on the API side of the architecture. The API stack will include the following:

An API Gateway to act as an entry point for the API.
A Lambda Function running FastAPI for the backend logic.
A DynamoDB Table for storing data about tasks.
A DynamoDB Stream that records changes made to the table.
A Lambda Function that is invoked whenever a new task is created, via the stream.
An SNS Topic, on which new tasks are published for further processing by handlers.

4.1. Create folder structure and required SAM files

To make it easier to understand in which file everything below should go into, this is how the API stack directory will look when you are finished with this section.

api-stack/
  api-function/
    requirements.txt
    app/
      __init__.py
      dynamo.py
      models.py
  publish-function
    requirements.txt
    app/
      __init__.py
  template.yml
  samconfig.toml

Start by creating and adding the following to the api-stack/template.yml file.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Task API

Resources:

To make SAM deployment easier and avoid parameters to the sam deploy command, add the following to api-stack/sam-config.toml.

version = 0.1

[default.global.parameters]
stack_name = "TaskAPI"  # Or choose your own stack name

[default.deploy]
[default.deploy.parameters]
capabilities = "CAPABILITY_IAM"
s3_bucket = "BUCKET_NAME"  # A bucket your credentials have access to
s3_prefix = "task-api"  # The prefix that will be used for your s3 assets
region = "eu-west-1"  # Change to your preferred region

4.2. DynamoDB Table

To start off, we will create a table that will store all of our tasks. In this example we will keep the table rather simple, and use only a Primary Key named id.

If you, for example, want to have tasks scoped to e.g. a user, an application, an organization, or some other kind of entity in your systems, you could opt for a schema where you use the entity ID (such as User ID) as the Primary Key and the Task ID as the Sort Key. But for now, let's keep it simple and continue with just using the Task ID as Primary Key.

We will use the CloudFormation resource AWS::DynamoDB::Table to define the table, since AWS::Serverless::SimpleTable does not support advanced features such as streams.

In Resources in the SAM template, add the following.

  Table:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: tasks
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      ProvisionedThroughput:
        ReadCapacityUnits: 1
        WriteCapacityUnits: 1

I've used the minimum provisioned throughput for testing purposes. Feel free to modify it to your needs, or change it if you want to use on-demand billing.

Feel free to deploy your SAM application after every step to verify that everything works. To do so, from the api-stack directory, first build the application with sam build and deploy it with sam deploy.

4.3. API Gateway

To front our API we will use API Gateway. More specifically, we will use a HTTP API since it is both cheaper and we do not need the more advanced features of a REST API.

For now, we will not use any features such as Authorization, CORS, or any other advanced features of an API Gateway. Authorization is mentioned in the potential improvements section.

In Resources in the SAM template, add the following.

  Api:
    Type: AWS::Serverless::HttpApi

  ApiUrlParameter:
    Type: AWS::SSM::Parameter
    Properties:
      Name: "/tasks/api_url"
      Type: String
      Value: !Sub "https://${Api}.execute-api.${AWS::Region}.${AWS::URLSuffix}"

The resource ApiUrlParameter will output the URL of the API Gateway to AWS Systems Manager Parameter Store. This will be used as input to the task handler stack(s).

To easily find the auto-generated API URL, you can take advantage of CloudFormation Outputs. Add the following to the SAM template. Remember, Outputs is a top-level key.

Outputs:
  ApiUrl:
    Description: URL of the Task API
    Value:
      Fn::Sub: "https://${Api}.execute-api.${AWS::Region}.${AWS::URLSuffix}"

The next time you run sam deploy, you should see the URL in the output.

4.4. Lambda API

Finally it is time to write some actual code and we will be using Python for this. We will use the FastAPI framwork to build our API, and use an adapter library called Mangum to make FastAPI play nice inside a Lambda function.

SAM Template

We need to define our Lambda function in the SAM template. Add the following resource.

ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: "TaskAPI"
      MemorySize: 128
      CodeUri: api-function
      Handler: app.handler
      Runtime: python3.9
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref Table
      Environment:
        Variables:
          TABLE_NAME: !Ref Table
      Events:
        DefaultEndpoint:
          Type: HttpApi
          Properties:
            ApiId: !Ref Api

The above template references the Table resource created before to add the table name as an environment variable, as well as adding an IAM policy to the Lambda execution role that allows it to perform CRUD operations on the DynamoDB table. We also add the API Gateway created earlier as an event source.

Boilerplate

First let's create a requirements.txt file in the api-stack/api-function directory. This should include the following packages.

mangum
fastapi
boto3
pydantic

Then, in the the api-stack/api-function/app/__init__.py file, add the following.

from fastapi import FastAPI, HTTPException
from mangum import Mangum


app = FastAPI()
handler = Mangum(app)

HTTPException will be used later on, so might aswell add it in now.

Routes

We will implement the following routes:

GET /tasks: Returns a list of all tasks.
GET /tasks/{id}: Returns a task with the given ID.
POST /tasks: Creates a new task.
PATCH /tasks/{id}: Updates the status of a task with the given ID.

GET /tasks/{id}

Let's start with implementing the Get /tasks/{id} endpoint. First, in the api-stack/api-function/app/dynamo.py file, add the following.

import os
import boto3


table = boto3.resource("dynamodb").Table(os.environ["TABLE_NAME"])


class Error(Exception):
    pass


class TaskNotFoundError(Error):
    pass


def get_task(task_id: str) -> dict:
    res = table.get_item(
        Key={
            "id": task_id,
        },
    )

    item = res.get("Item")
    if not item:
        raise TaskNotFoundError

    return item

Here we use the boto3 resource client to call the GetItem API with the task_id as the Primary Key. If the task is not found, we raise an error, otherwise we return the task data. We do not filter the data here, this is handled by FastAPI response models as you will see soon.

Before creating our route logic, let's create a response model that will define what data is returned by the API when a task is retrieved. We will use pydantic to declare our models

In api-stack/api-function/app/models.py, add the following.

from pydantic import BaseModel
from typing import Literal


task_types = Literal["TASK1", "TASK2", "TASK3"]
status_types = Literal["CREATED", "IN_PROGRESS", "COMPLETED", "FAILED"]


class TaskResponse(BaseModel):
    id: str
    task_type: task_types
    status: status_types
    status_msg: str = ""
    created_time: int = None
    updated_time: int = None

Now, let's create the route logic. In api-stack/api-function/app/__init__.py, add the following.

from . import models, dynamo

...

@app.get("/tasks/{task_id}", response_model=models.TaskResponse)
def get_task(task_id: str):
    try:
        return dynamo.get_task(task_id)
    except dynamo.TaskNotFoundError:
        raise HTTPException(status_code=404, detail="Task not found")

Let's deploy what we have so far and see if it works. Run sam build and sam deploy.

If you haven't used FastAPI before this might come as a surprise, but fire up your favourite browser and navigate to https://YOUR_API_URL/docs. The URL should be listed in the output of sam deploy.

FastAPI comes with built-in support for generating and serving documentation for your API using Swagger. Try your new route from the Swagger UI or with something like curl.

$ curl https://YOUR_API_URL/tasks/12345
{
  "detail": "Task not found"
}

Obviously, no tasks exist yet since we haven't created any yet. Let's change that and implement the create route.

POST /tasks

It is time to implement the logic for creating tasks. First, we need to decide what kind of input the user should include in the request. We want to support different kinds of tasks, so we will need to include a task_type field in the request. Different tasks might require different payloads, so let's add a data field in the request which accepts generic json.

When a task is created, the response we will send the user will include the task ID.

Add the following models to api-stack/api-function/app/models.py.

class CreatePayload(BaseModel):
    task_type: task_types
    data: dict


class CreateResponse(BaseModel):
    id: str

When creating a task, we want to generate an ID for the task. We can use the uuid library to generate a random UUID. We will also add the current timestamp in the attribute created_time. And finally, to avoid handling json in payloads when publishing tasks over SNS and SQS we will base64 encode the payload and store the encoded payload in DynamoDB.

Add the following to api-stack/api-function/app/dynamo.py.

...
import base64
import json
from uuid import uuid4
from datetime import datetime

...

def create_task(task_type: str, payload: dict) -> str:
    task_id = str(uuid4())
    table.put_item(
        Item={
            "id": task_id,
            "task_type": task_type,
            "status": "CREATED",
            "payload": _encode(payload),
            "created_time": _get_timestamp(),
        }
    )

    return {"id": task_id}



def _encode(data: dict) -> str:
    json_string = json.dumps(data)
    return base64.b64encode(json_string.encode("utf-8")).decode("utf-8")


def _get_timestamp() -> int:
    return int(datetime.utcnow().timestamp())

And finally, add the following to api-stack/api-function/app/__init__.py.

@app.post("/tasks", status_code=201, response_model=models.CreateResponse)
def post_task(payload: models.CreatePayload):
    return dynamo.create_task(payload.task_type, payload.data)

Here we can see how FastAPI takes care of input validation and serialization when we specify the response_model=models.CreateResponse as well as the expected request payload with payload: models.CreatePayload.

Let's deploy and try creating a task.

$ curl -X POST \
  -H "Content-Type: application/json" \
  -d '{"task_type": "TASK1", "data": {"foo": "bar"}}' \
  https://YOUR_API_URL/tasks
{
  "id": "12345678-abcd-1234-abcd-112233445566"
}

...

$ curl https://YOUR_API_URL/tasks/12345678-abcd-1234-abcd-112233445566
{
  "id": "12345678-abcd-1234-abcd-112233445566",
  "task_type": "TASK1",
  "status": "CREATED",
  "status_msg": "",
  "created_time": 1648229203,
  "updated_time": null
}

Did you accidentally forgot to copy the task ID after creating a few tasks? If there only was a way to list all tasks without opening up the DynamoDB console. Let's implement that next.

GET /tasks

The list route will be very simple, without any filter or sort queries. We will allow a maximum of 10 tasks to be returned at a time. If there are more than 10 tasks, we will return a next_token in the response. The next request should then include the next_token in the query string to fetch the next 10 tasks. If there are no more tasks left, next_token will be null.

Since all items in the DynamoDB table have unique primary keys, we will need to use the scan operation to fetch items.

Let's start with the model for the response which will be a list of tasks and a next_token field.

In api-stack/api-function/app/models.py, add the following.

class TaskListResponse(BaseModel):
    tasks: list[TaskResponse]
    next_token: str = None

Moving on the the actual database logic, we will need to conditionally add the next_token to the scan operation in case it is provided. The next_token will be base64 encoded before it is returned to the client, so we will need to decode it before we can use it.

The scan response will include a LastEvaluatedKey if there are more items left to fetch, so we use that to set the next_token.

Add the following to api-stack/api-function/app/dynamo.py.

def list_tasks(next_token: str = None) -> dict:
    scan_args = {
        "Limit": 10,
    }

    if next_token:
        scan_args["ExclusiveStartKey"] = _decode(next_token)

    res = table.scan(**scan_args)
    response = {"tasks": res["Items"]}

    if "LastEvaluatedKey" in res:
        response["next_token"] = _encode(res["LastEvaluatedKey"])

    return response

...

def _decode(data: str) -> dict:
    json_string = base64.b64decode(data.encode("utf-8")).decode("utf-8")
    return json.loads(json_string)

Finally, add the following to api-stack/api-function/app/__init__.py.

@app.get("/tasks", response_model=models.TaskListResponse)
def list_tasks(next_token: str = None):
    return dynamo.list_tasks(next_token)

Now deploy, and you should be able to list all tasks you've created.

$ curl https://YOUR_API_URL/tasks
{
  "tasks": [
    {
      "id": "12345678-abcd-1234-abcd-112233445566",
      "task_type": "TASK1",
      "status": "CREATED",
      "status_msg": "",
      "created_time": 1648229203,
      "updated_time": null
    },
    ... more tasks ...
  ],
  "next_token": null
}

We can now create and list tasks, but what good is that if we cannot update their status? Let's go ahead and implement the final route.

PATCH /tasks/{id}

This endpoint is supposed to be used internally by the task handlers. Thus, it would be preferable to add some kind of authorization here so that users cannot update tasks directly. But, to keep the scope small, I have ignored it for now.

We want task handlers to be able to update the status of the tasks they handle, as well as an optional message string. Add the following model to api-stack/api-function/app/models.py.

class UpdatePayload(BaseModel):
    status: status_types
    status_msg: str = ""

The logic for updating a task will be a bit more complex than the other operations. First, we need to make sure the task actually exists. We also want to guard ourselves against multiple handlers trying to start the same task. This could for example happen due to a side effect of the SQS at least once delivery mechanism, or if the visiblity timeout on the queue is shorter than the time it takes to process a task. These two checks are made with conditional updates in DynamoDB.

Add the following code to api-stack/api-function/app/dynamo.py.

from boto3.dynamodb.conditions import And, Attr

...

class InvalidTaskStateError(Error):
    pass

...

def update_task(task_id: str, status: str, status_msg: str):
    cond = Attr("id").exists()

    if status == "IN_PROGRESS":
        cond = And(cond, Attr("status").eq("CREATED"))

    try:
        table.update_item(
            Key={
                "id": task_id,
            },
            UpdateExpression="set #S=:s, status_msg=:m, updated_time=:t",
            # status is reserved
            ExpressionAttributeNames={
                "#S": "status",
            },
            ExpressionAttributeValues={
                ":s": status,
                ":m": status_msg,
                ":t": _get_timestamp(),
            },
            ConditionExpression=cond,
        )
    except table.meta.client.exceptions.ConditionalCheckFailedException:
        raise InvalidTaskStateError

We first create a condition that requires the item to exist already. Then, if we are setting the state to IN_PROGRESS, we will also require the task to have the status CREATED. This way, if another requests comes in that tries to set the state to IN_PROGRESS while it already is in progress, the request will fail.

We then use the update_item method which will throw an exception if the condition evaluates to false. As far as I know, you cannot see which part of the condition failed, so we cannot differentiate between a task that doesn't exist and a task that is already in progress.

Finally, add the route to api-stack/api-function/app/__init__.py and we should be good to go.

@app.patch("/tasks/{task_id}", status_code=204)
def update_task(task_id: str, payload: models.UpdatePayload):
    try:
        return dynamo.update_task(task_id, payload.status, payload.status_msg)
    except dynamo.InvalidTaskStateError:
        raise HTTPException(
            status_code=400, detail="Task does not exist or is already in progress."
        )

Deploy and try it out! You are now finished with the API.

4.5. DynamoDB Stream

We now want to enable a Stream on the DynamoDB table. I choose to use streams to remove the need to handle transactions. Imagine the following scenario:

User creates a task through the API
Task is created in DynamoDB
Request to SNS fails for some reason

We would then need to roll back the task we created in DynamoDB. By using streams, we make sure that we only send a task to SNS after it has been commited to the database. And, if the request to SNS then fails, the function can be configured to retry a set amount of times.

To enable the stream, add StreamSpecification under properties in the Table resource in the SAM template.

  Table:
    Type: AWS::DynamoDB::Table
    Properties:
      ...
      StreamSpecification:
        StreamViewType: NEW_IMAGE

For now, we only care about the current state of the items in the stream. If we wanted to act on specific changes in the DynamoDB table, such as send a notification somewhere when a task went from IN_PROGRESS to FAILED, we would need to use NEW_AND_OLD_IMAGES instead of NEW_IMAGE. That way we could compare the old and new values of the item in the stream handler.

4.6. SNS Topic

We also need an SNS Topic to send our tasks to. Add the following to the SAM template.

  Topic:
    Type: AWS::SNS::Topic

  TopicArnParameter:
    Type: AWS::SSM::Parameter
    Properties:
      Name: "/tasks/topic_arn"
      Type: String
      Value: !Ref Topic

The resource TopicArnParameter will output the Topic ARN to AWS Systems Manager Parameter Store. This will be used as input to the task handler stack(s).

4.7. Publisher Lambda

It is time to create our second Lambda function. This function will be responsible for publishing the task to SNS.

Required packages

Add the following to api-stack/publish-function/requirements.txt.

boto3
aws-lambda-powertools

Lambda Powertools for Python is a package that provides a lot of useful features when working with Lambda functions, such as logging, tracing, and data classes for common event source payloads.

SAM Template

Let's define the publisher function in the SAM template.

  PublishFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: "TaskPublisher"
      MemorySize: 128
      CodeUri: publish-function
      Handler: app.handler
      Runtime: python3.9
      Policies:
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt Topic.TopicName
      Environment:
        Variables:
          TOPIC_ARN: !Ref Topic
      Events:
        Stream:
          Type: DynamoDB
          Properties:
              Stream: !GetAtt Table.StreamArn
              StartingPosition: TRIM_HORIZON
              MaximumRetryAttempts: 5
              FilterCriteria:
                Filters:
                  - Pattern: '{"eventName": ["INSERT"]}'

Here we can see that we are doing almost the same thing with the SNS Topic as we did with the DynamoDB table in the API function. We are using the built in SNSPublishMessagePolicy policy to give the function permissions to publish to the topic.

We also define a Stream event that will trigger the function when some operation is done in the DynamoDB table. We also define a filter criteria to only trigger the function when a new item is added to the table, i.e. when a task is first created.

Handler

The publisher handler will be quite simple. Here we utilize the package aws-lambda-powertools to provide us with data classes for the event payloads. For each message received from the stream (which is filtered to only include INSERT events), we publish a message to SNS.

We send the payload (which is already base64 encoded) as the message body, and we add the task ID and task type as message attributes.

All in all, it looks like this. Add it to api-stack/publish-function/app/__init__.py and we are done with the entire API stack.

import os
import boto3
from aws_lambda_powertools.utilities.data_classes import (
    event_source,
    DynamoDBStreamEvent,
)


topic = boto3.resource("sns").Topic(os.environ["TOPIC_ARN"])


@event_source(data_class=DynamoDBStreamEvent)
def handler(event: DynamoDBStreamEvent, _):
    for record in event.records:
        task_id = record.dynamodb.keys["id"].get_value
        task_type = record.dynamodb.new_image["task_type"].get_value
        payload = record.dynamodb.new_image["payload"].get_value

        res = topic.publish(
            MessageAttributes={
                "TaskId": {
                    "DataType": "String",
                    "StringValue": task_id,
                },
                "TaskType": {
                    "DataType": "String",
                    "StringValue": task_type,
                },
            },
            Message=payload,
        )

        print(f"Message {res['MessageId']} published.")

Great job so far! We have now finished the API part of the system, and it is time to start building our handlers.

5. Build the Handler stack(s)

With the API complete, we can now build the handler stack(s). In this example, I will only create a single stack with a single handler for events with task_type set to TASK1. Implementing handlers for TASK2 and TASK3, or other events, is left as an exercise for the reader.

The stack we will build will include the following resources:

A Lambda Function that will be responsible for handling the task.
An SQS Queue that will be used as a middleware between the SNS Topic and Lambda function.
An SQS Queue Policy that will be used to grant the SNS Topic permission to send messages to the SQS Queue.
An SNS Subscription between the SNS Topic and the SQS Queue.
Another SQS Queue that will act as a dead letter queue.

5.1. Create folder structure and required SAM files

To make it easier to understand in which file everything below should go into, this is how the Handler stack directory will look when you are finished with this section.

handler-stack/
  task1-function/
    requirements.txt
    app/
      __init__.py
  template.yml
  samconfig.toml

As before, add some boilerplate to the handler-stack/template.yml file.

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Task API

Resources:

We will also use a handler-stack/samconfig.toml file for easier CLI usage.

version = 0.1

[default.global.parameters]
stack_name = "TaskHandlers"  # Or choose your own stack name

[default.deploy]
[default.deploy.parameters]
capabilities = "CAPABILITY_IAM"
s3_bucket = "BUCKET_NAME"  # A bucket your credentials have access to
s3_prefix = "task-handlers"  # The prefix that will be used for your s3 assets
region = "eu-west-1"  # Change to your preferred region

5.2. SQS Queue

First, let's create the SQS Queue as well as the dead letter queue. Add the following resources to the SAM template.

TaskHandler1Queue:
  Type: AWS::SQS::Queue
  Properties:
    RedrivePolicy:
      deadLetterTargetArn: !GetAtt TaskHandler1DLQ.Arn
      maxReceiveCount: 1

TaskHandler1DLQ:
  Type: AWS::SQS::Queue

By using a DLQ, we will be able to capture events that failed to be processed by the Lambda function. We can analyze the events and either discard them or send them back to the main queue to be processed again. maxReceiveCount specifies how many times you want to retry the event in case of failure before sending it to the dead letter queue. In this example, we will keep it at one to disable retries.

5.3. SNS Subscription

We now want to set up a subscription to the SNS Topic that was created in the API stack. Remember how we created an SSM parameter in the API stack with the topic arn? We will now import that value in the handler stack. We will also need to create a Queue Policy that will allow the SNS Topic to send messages to the SQS Queue. In the handler we are creating now, we only want to process events which have the message attribute TaskType set to TASK1. To do this, we will use a filter policy on the subscription.

In the SAM template, first add the Parameters section.

Parameters:
  TasksTopic:
    Type: AWS::SSM::Parameter::Value<String>
    Description: Tasks Topic Arn
    Default: /tasks/topic_arn

Now, under resources, add the following.

TaskHandler1QueuePolicy:
  Type: AWS::SQS::QueuePolicy
  Properties:
    Queues:
      - !Ref TaskHandler1Queue
    PolicyDocument:
      Version: "2012-10-17"
      Statement:
        - Effect: Allow
          Action: sqs:SendMessage
          Resource: !GetAtt TaskHandler1Queue.Arn
          Principal:
            Service: "sns.amazonaws.com"
          Condition:
            ArnEquals:
              aws:SourceArn: !Ref TasksTopic

TaskHandler1Subscription:
  Type: AWS::SNS::Subscription
  Properties:
    Protocol: sqs
    TopicArn: !Ref TasksTopic
    Endpoint: !GetAtt TaskHandler1Queue.Arn
    RawMessageDelivery: True
    FilterPolicy:
      TaskType:
        - "TASK1"

5.4. Handler function

SAM Template

We need to define our Lambda function in the SAM template. Add the following resource.

TaskHandler1Function:
  Type: AWS::Serverless::Function
  Properties:
    FunctionName: "TaskHandler1"
    MemorySize: 128
    Timeout: 30
    CodeUri: task1-function
    Handler: app.handler
    Runtime: python3.9
    Policies:
      - SQSPollerPolicy:
          QueueName: !GetAtt TaskHandler1Queue.QueueName
    Environment:
      Variables:
        TASKS_API_URL: !Ref TasksApiUrl
    Events:
      Stream:
        Type: SQS
        Properties:
            Queue: !GetAtt TaskHandler1Queue.Arn
            BatchSize: 1

As you can see, the lambda function requires the URL of the Task API. Since we exported the URL to the parameter store from the API stack, we should add the following under the Parameters section.

TasksApiUrl:
  Type: AWS::SSM::Parameter::Value<String>
  Description: Tasks
  Default: /tasks/api_url

Code

Now it is time to implement the actual task handler for the TASK1 events. The example I will show here is very minimal (and completely useless), but it should be enough to get you started. Again, we will use the aws-lambda-powertools to deserialize the event payload from SQS to make our life a little easier. For each record that the lambda receives, we will do the following.

Read Task ID and Task Type from the message attributes.
Decode the base64 encoded payload that the client provided when creating the task.
Call the Update endpoint in the Task API to set the task to IN_PROGRESS.
Perform the actual task. In this case, sleep for 10 seconds...
If the task was successful, call the Update endpoint in the Task API to set the task to COMPLETED. If an exception was raised, update the status to FAILED.

In the code example, I also randomly raise exceptions to simulate failures.

All in all, the handler function looks like this. Add it to the handler-stack/task1-function/app/__init__.py file.

import base64
import json
import os
import time
import requests
import random
from aws_lambda_powertools.utilities.data_classes import (
    event_source,
    SQSEvent,
)


API_URL = os.environ["TASKS_API_URL"]


@event_source(data_class=SQSEvent)
def handler(event: SQSEvent, context):
    for record in event.records:
        task_id = record.message_attributes["TaskId"].string_value
        task_type = record.message_attributes["TaskType"].string_value
        payload = _decode_payload(record.body)

        print(f"Starting task {task_type} with id {task_id}")
        _update_task_status(task_id, "IN_PROGRESS", "Task started")
        try:
            _do_task(payload)
        except Exception as e:
            print(f"Task with id {task_id} failed: {str(e)}")
            _update_task_status(task_id, "FAILED", str(e))
            continue

        print(f"Task with id {task_id} completed successfully.")
        _update_task_status(task_id, "COMPLETED", "Task completed")


def _do_task(payload: dict):
    # Do something here.
    print(f"Payload: {payload}")
    time.sleep(10)
    if random.randint(1, 4) == 1:
        # Simulate failure in some invocations
        raise Exception("Task failed somehow")


def _decode_payload(payload: str) -> dict:
    json_string = base64.b64decode(payload.encode("utf-8")).decode("utf-8")
    return json.loads(json_string)


def _update_task_status(task_id: str, status: str, status_msg: str):
    data = {
        "status": status,
        "status_msg": status_msg,
    }

    url = f"{API_URL}/tasks/{task_id}"
    res = requests.patch(url, json=data)

    if res.status_code != 204:
        print(f"Request to API failed: {res.json()}")
        raise Exception("Update task status failed")

Also, don't forget the to add the required packages in the handler-stack/task1-function/requirements.txt file.

aws-lambda-powertools
requests

The handler stack is done for now, I'll leave you to implement the actual task and perhaps create handlers for the other task types as well. Time to deploy what we have so far!

6. Showtime

It's showtime. With both stacks deployed, try creating a few tasks through the API and see the magic happen. Try creating both tasks with Task Type TASK1 and TASK2 and see what happens. If everything works as it's supposed to, you should see the TASK1 tasks change status to IN_PROGRESS and then COMPLETED/FAILED after a few seconds. Tasks with other task types should be ignored and be left with status CREATED.

# Create a task with type TASK1
$ curl -X POST \
  -H "Content-Type: application/json" \
  -d '{"task_type": "TASK1", "data": {"foo": "bar"}}' \
  https://YOUR_API_URL/tasks
{
  "id": "11111111-abcd-1111-abcd-111111111111"
}

# Create a task with type TASK2
$ curl -X POST \
  -H "Content-Type: application/json" \
  -d '{"task_type": "TASK2", "data": {"foo": "bar"}}' \
  https://YOUR_API_URL/tasks
{
  "id": "22222222-abcd-2222-abcd-222222222222"
}

# List tasks. The task with type TASK1 should have status IN_PROGRESS and the other
# should still have status CREATED. If it isn't IN_PROGRESS, try again after a few seconds.
$ curl https://YOUR_API_URL/tasks
{
  "tasks": [
    {
      "id": "11111111-abcd-1111-abcd-111111111111",
      "task_type": "TASK1",
      "status": "IN_PROGRESS",
      "status_msg": "",
      "created_time": 1648229203,
      "updated_time": null
    },
    {
      "id": "22222222-abcd-2222-abcd-222222222222",
      "task_type": "TASK1",
      "status": "CREATED",
      "status_msg": "",
      "created_time": 1648229203,
      "updated_time": null
    },
  ],
  "next_token": null
}

# Allow 10 seconds for the task to complete and then list tasks again. The task should
# now have the status COMPLETED or FAILED.
$ curl https://YOUR_API_URL/tasks
{
  "tasks": [
    {
      "id": "11111111-abcd-1111-abcd-111111111111",
      "task_type": "TASK1",
      "status": "COMPLETED",
      "status_msg": "",
      "created_time": 1648229203,
      "updated_time": 1648229218
    },
    {
      "id": "22222222-abcd-2222-abcd-222222222222",
      "task_type": "TASK1",
      "status": "CREATED",
      "status_msg": "",
      "created_time": 1648229203,
      "updated_time": null
    },
  ],
  "next_token": null
}

7. Cleaning up

To remove everything we have created, simply run the sam delete command in the api-stack and handler-stack directories.

8. Potential improvements

While this is a simple example, there are many things we could do to make it better. Below are some ideas that I can think of from the top of my head. Why don't give one of them a try?

8.1. Authentication and Authorization

Right now there is no authentication or authorization on the Task API. This means that any client can create tasks and see the status of any task, and also update the status of tasks. First, I would make sure that only the handlers themselves are allowed to use the PATCH /tasks/{id} endpoint. This could for example be done by setting up IAM authorization on the API Gateway. Secondly, we might want to require that the client is authenticated before creating and listing tasks. If you want to do this in a serverless fashion, you could look into AWS Cognito and use a JWT authorizer.

8.2. DynamoDB TTL

Since our list endpoint retrieves all tasks, this list could grow very large. Perhaps we want to remove tasks when they are older than a certain amount of time. This could be done by setting a TTL on the DynamoDB table to automatically delete old tasks after a set period of time.

8.3. Logging

Right now we do not have much logging in place. And in the few places we have, it is only simple print statements that aren't as structured as we want. aws-lambda-powertools has a great logging library that helps with setting up structured logs for your lambda functions.

8.4. Tracing

Monitoring in distributed serverless systems can be quite daunting. AWS provides X-Ray for this purpose, which is a distributed tracing system. This can help you visualize the flow of events in your application going from the API Gateway -> Lambda -> DynamoDB -> Lambda -> SNS -> SQS -> Lambda and so on. aws-lambda-powertools has a great tracer library that helps with setting up X-Ray for your lambda functions.

8.5. Error handling

I have not included any kind of error handling in the example. You could for example implement functionality to allow a task to be retried a set amount of times in case of failure. Right now, if a task fails, it will be updated in Dynamo to have a status of FAILED. The current implementation of the update endpoint requires the task to be have a status of CREATED when updating the status to IN_PROGRESS. If that logic is left unchanged, retried invocations will fail on the first request to the update endpoint.

8.6. Webhooks

Most of the improvements above have been about securing, managing, and monitoring the API. We could also extend it with new features. One example would be to include webhooks. Clients could for example include a webhook URL in their task creation request. We could then add another lambda function that reads from the DynamoDB stream and sends a notification to the webhook URL when the task goes from IN_PROGRESS to COMPLETED or from IN_PROGRESS to FAILED.

To be able to react on certain changes in a DynamoDB item, you must update the stream view type to be NEW_AND_OLD_IMAGES instead of NEW_IMAGE. This way the lambda will receive both the old and the new version of the item.

8.7. Scoped events

Perhaps you want to scope tasks to users, applications, or some other entity in your system. Then, I would do the following changes.

Change the DynamoDB schema to use both a HASH key and SORT key, and set the HASH key to be the entity ID and SORT key to be the task ID.
Change the API to include the entity ID in the request, such as POST /{entity_id}/tasks.
Update the boto3 DynamoDB client calls to use the correct key. The scan operation would for example be replaced with a query operation, using the Entity ID as the query key.

9. Conclusion

Congratulations for making it this far. You have now managed to build an asynchronous task API running entirely on serverless services on AWS. Hopefully you have learned a thing or two, I know I definitely did by creating this. If you have any questions, please feel free to reach out to me in any way you see fit.

All code in this guide is available on GitHub. It might drift a bit if I decide to build upon it, but if I do, I will try to keep this blog entry up to date.

Now go build something awesome! Why not try implementing one of the ideas from the potential improvements section?

Deploy FastAPI on AWS Part 2: Fargate & ALB

Elias Brange — Wed, 09 Mar 2022 10:25:55 +0000

In the previous part you learned how to deploy FastAPI on Lambda. This time we will use containers, but still in a serverless fashion. Read on to learn how to deploy a FastAPI application on AWS Fargate behind an Application Load Balancer using AWS CDK.

Introduction

FastAPI is a Python framework for creating production-ready APIs, made by Sebastián Ramírez AKA tiangolo. It's fast (both when it comes to performance and developer speed), easy to get started with, and an absolute joy to work with.

In this two-part series you will learn how to create and deploy a minimal FastAPI application on AWS in a serverless fashion.

In part 1, you learned how to deploy FastAPI on Lambda using SAM.

In part 2, you will instead learn how to package FastAPI inside a container and deploy it on AWS Fargate, the serverless container platform. Here we will use an Application Load Balancer to front the application instead of API Gateway, and this time we will define and deploy the application using AWS Cloud Development Kit (CDK).

Are you ready to test out serverless containers? Keep reading!

Requirements

Python
AWS CDK installed and bootstrapped
Docker
Docker Compose (for running locally)

Tutorial

1. Create a directory for your application

To start, create a new directory and cd into it.

$ mkdir fastapi-on-fargate
$ cd fastapi-on-fargate

2. Install AWS CDK for Python

In this example I have used Python, so to follow along the tutorial you must ensure that aws-cdk-lib is installed. This is preferably done in a virtualenv, to avoid polluting the global python installation.

i. Create a virtualenv

$ python3 -m venv .venv

ii. Activate the virtualenv

$ source .venv/bin/activate

iii. Install aws-cdk-lib

$ pip3 install aws-cdk-lib

3. Create a simple FastAPI application

The sample application we will use is similar to the one used in part 1. However, this time we will not use mangum as an adapter.

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def get_root():
    return {"message": "FastAPI running in a Docker container"}

Save the above in a file with the path src/app/__init__.py.

4. Specify runtime dependencies

To run our function we will need to install a couple of third-party dependencies. We define these in a requirements.txt file.

fastapi
uvicorn

Save the above in a file with the path src/requirements.txt.

5. Create a Dockerfile

Now it is time to dockerize our FastAPI application. In the src/ directory, create a Dockerfile with the following contents:

FROM python:3.9-alpine
WORKDIR /src
COPY ./requirements.txt /src/requirements.txt
RUN pip install --no-cache-dir --upgrade -r /src/requirements.txt
COPY ./app /src/app
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "80"]

Save the above in a file with the path src/Dockerfile.

6. Test your API locally with Docker Compose

Now that we have an application and a Dockerfile, let's try running it locally using Docker Compose. First, create a Compose file with the following contents:

services:
  app:
    build: src
    ports:
      - "3000:80"

Save the above as docker-compose.yml in the root of the directory you created in step 1.

You should now be able to start your API locally:

$ docker-compose up

The above command should build a Docker image from your Dockerfile, and start the API locally on port 3000. Let's try calling it:

$ curl localhost:3000

{"message":"FastAPI running in a Docker container"}

7. Define your application using CDK

Now that we have verified that our dockerized API works as expected, it is time to get building with AWS CDK. The current directory structure should look like the following:

fastapi-on-fargate/
  docker-compose.yml
  src/
    requirements.txt
    Dockerfile
    app/
      __init__.py

Next to the src/ directory, create a new directory named cdk/.

In the cdk/ directory, add a file cdk.json with the following contents:

{
  "app": "python3 cdk.py"
}

This tells cdk that our application is defined in cdk.py, so let's create that file:

from aws_cdk import App
from fastapi_stack import FastAPIStack

app = App()
FastAPIStack(app, "FastAPIStack")
app.synth()

In the above file, we have imported FastAPIStack from fastapi_stack, but that file does not exist (yet). Save the following in fastapi_stack.py (next to cdk.py):

from aws_cdk import Stack
from constructs import Construct

import aws_cdk.aws_ec2 as ec2
import aws_cdk.aws_ecs as ecs
import aws_cdk.aws_ecs_patterns as ecs_patterns


class FastAPIStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # (i) Create VPC
        self.vpc = ec2.Vpc(self, "MyVPC", max_azs=3)

        # (ii) Create Fargate Cluster
        self.ecs_cluster = ecs.Cluster(
            self,
            "MyECSCluster",
            vpc=self.vpc,
        )

        # (iii) Define Docker image for the Service
        image = ecs_patterns.ApplicationLoadBalancedTaskImageOptions(
            image=ecs.ContainerImage.from_asset(
                directory="../src",
            )
        )

        # (iv) Create Fargate Service and ALB
        self.ecs_service = ecs_patterns.ApplicationLoadBalancedFargateService(
            self,
            "FastAPIService",
            cluster=self.ecs_cluster,
            cpu=256,
            memory_limit_mib=512,
            desired_count=2,
            task_image_options=image,
        )

This is where the magic happens. Lets go through the resource one by one.

i. Create VPC

This statement creates a VPC for us to use. Under the hood, CDK will generate a lot of resources (Subnets, Route Tables, Security Groups, etc.)

To see the power of CDK and how much it abstracts away from us, try executing cdk synth and check out the generated CloudFormation template.

ii. Create Fargate Cluster

This statement creates a Fargate Cluster inside the VPC defined above.

iii. Define Docker image for the Service

Here we define which Docker image the Service should use. In a real world scenario, this should preferably reference an image from a registry such as DockerHub. But for the sake of this tutorial, here we build the image locally by specifying the path to the Dockerfile (which is ../src).

iv. Create Fargate Service and ALB

This statement illustrates a high-level construct in CDK. Here we specify which cluster our service should run in, how many instances of the service we want, which image to use, and how much resources to allocate. The construct also sets up an Application Load Balancer in front of the Fargate Service and sets up the connection between them. There are more configuration options available for this particular construct than shown here.

8. Deploy your API to AWS

To deploy the API to AWS you need the following:

An AWS account (duh!).
Credentials configured for said account.
Your account needs to be bootstrapped for CDK.

With the above in place we can now use CDK to deploy our FastAPI application. Issue the following command inside the cdk/ directory:

$ cdk deploy

  ...

  ✅ FastAPIStack

  Outputs:
  FastAPIStack.FastAPIServiceLoadBalancerDNS12341234 = FastA-FastA-XXXXXXXXXXXX-1111111111.eu-west-1.elb.amazonaws.com
  FastAPIStack.FastAPIServiceServiceURL12341234 = http://FastA-FastA-XXXXXXXXXXXX-1111111111.eu-west-1.elb.amazonaws.com

  Stack ARN:
  arn:aws:cloudformation:eu-west-1:1234567890:stack/FastAPIStack/11111111-aaaa-2222-bbbb-333333333333

If everything works, and hopefully it does, you should see an output similar to the one above. This means that we have successfully deployed the API.

9. Call your newly deployed API

From the cdk deploy output, you should be able to locate the URL for the Application Load Balancer. Let's try it out in the browser, or from the command line:

$ curl http://FastA-FastA-XXXXXXXXXXXX-1111111111.eu-west-1.elb.amazonaws.com

{"message":"FastAPI running in a Docker container"}

Cleaning up

To clean up, simply issue the following CDK command:

$ cdk destroy

Conclusion

You have now learned how to run a FastAPI application on Fargate, and how to configure and deploy it using AWS CDK. You have also seen how to use high-level constructs in CDK for common use-cases, such as deploying a Fargate Service behind an Application Load Balancer.

This wraps up this short two-part series, I hope you have learned a thing or two.

Now go build something awesome!