DEV Community: Robert Slootjes

SQS Fair Queues with AWS EventBridge

Robert Slootjes — Wed, 03 Dec 2025 15:45:54 +0000

A while ago AWS introduced Fair Queues, a new feature for SQS. I was immediately excited and started testing this using the SDK it worked really well:

import {
  SQSClient,
  SendMessageCommand,
} from '@aws-sdk/client-sqs';

const sqsClient = new SQSClient();
await sqsClient.send(
  new SendMessageCommand({
    QueueUrl: 'https://sqs.eu-west-1.amazonaws.com/123/fair-queue-demo',
    MessageBody: JSON.stringify({
      foo: 'bar',
    }),
    MessageGroupId: 'mybrand',
  }),
);

The message MessageGroupId attribute has a value of mybrand as expected:

{
    "body": "{\"foo\":\"bar\"}",
    "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1764775946158",
        "MessageGroupId": "mybrand"
    }
}

EventBridge

After the announcement of Fair Queues I tried to make it work with EventBridge since we use this almost everywhere. Passing the MessageGroupId dynamically was not possible unfortunately yet when Fair Queues was first released. However recently it was announced that the MessageGroupId can now be passed in EventBridge SQS Targets so I decided to give it another try:

import { 
  EventBridgeClient,
  PutEventsCommand,
} from '@aws-sdk/client-eventbridge';

const eventBridgeClient = new EventBridgeClient();
await eventBridgeClient.send(new PutEventsCommand({
  Entries: [{
    Source: 'myapp',
    DetailType: 'fair-queue-demo',
    Detail: JSON.stringify({
      brand: 'mybrand',
      foo: 'bar',
    })
  }]
}));

In combination with this rule:

FairQueueDemoRule:
  Type: AWS::Events::Rule
  Properties:
    EventPattern:
      source:
        - myapp
      detail-type:
        - fair-queue-demo
    State: ENABLED
    Targets:
      - Id: FairQueueDemo
        Arn: !GetAtt FairQueueDemo.Arn
        SqsParameters:
          MessageGroupId: $.detail.brand

And now it will actually set the MessageGroupId based on the contents of the detail object sent to EventBridge:

{
    "body": "{\"version\":\"0\",\"id\":\"90a484f2-d2e2-2ff2-f8b2-3b64a4538721\",\"detail-type\":\"fair-queue-demo\",\"source\":\"myapp\",\"time\":\"2025-12-03T15:21:16Z\",\"region\":\"eu-west-1\",\"resources\":[],\"detail\":{\"brand\":\"mybrand\",\"foo\":\"bar\"}}",
    "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1764775276763",
        "MessageGroupId": "mybrand"
    }
}

As you can see, the MessageGroupId is set to mybrand as expected 🎉

AWS Lamba & RDS Proxy

Robert Slootjes — Sun, 15 Jun 2025 13:03:42 +0000

Probably like many others, I tried to avoid relational databases like Postgres, MySQL and MariaDB with Lambda as I've read many times that connection based services aren't a good fit for the potentially rapid scaling of Lambda. So ever since I am doing serverless, I try to use DynamoDB or ElastiCache Redis/Valkey as much as possible.

At some point I really needed to use a relational database and I started playing with RDS Aurora. I created an instance, connected from Lambda and it worked just fine. However when I generated a bit more load it soon started locking up, all connections were in use and new ones couldn't be created. It would take a while for the database to become available again. The warning for combining Lambda with connection based services are not there without reasons.

RDS Proxy

Then, I remembered RDS Proxy and RDS Data API to exist and now I understood why. While Data API is a great service, it does come with some limitations and I decided to go with RDS Proxy. I also liked the aspect of being able to use generic libraries for connecting and querying data.

CloudFormation

Getting everything to work with CloudFormation and using best practices took me a while as a different setting could completely lock up the creation of a database. I also ran into issues where the RDS Proxy was created and had no errors in the console but it turned out that, using CLI tools, the network config was broken which was the reason I couldn't connect. To save you from this pain, I created a demo that sets up Lambda with RDS Aurora & Proxy making use of security best practices like a managed password and IAM authentication.

Demo

The demo is using the open source fork of Serverless Framework (a drop-in replacement for Serverless Framework v3!) with Typescript that will give you the following after deploying:

Networking

A basic setup with a VPC, 2 private and 2 public subnets in different availability zones for high availability (thanks to my colleague Martijn).

RDS Aurora Postgres

An RDS Aurora Postgres cluster with Postgres 16.6 with 2 serverless instances for Multi-AZ support. Also enabled are encrypted storage, backups, IAM authentication and serverless scaling configuration.

Database:
  Type: AWS::RDS::DBCluster
  Properties:
    ManageMasterUserPassword: True
    Engine: aurora-postgresql
    EngineVersion: 16.6
    StorageEncrypted: True
    EnableIAMDatabaseAuthentication: True
    BackupRetentionPeriod: 7
    ServerlessV2ScalingConfiguration:
      MinCapacity: 0
      MaxCapacity: 1

RDS Proxy

An RDS Proxy instance with read/write and read only endpoints making use of IAM authentication.

DatabaseProxy:
  Type: AWS::RDS::DBProxy
  Properties:
    Auth:
      - AuthScheme: SECRETS
    ClientPasswordAuthType: POSTGRES_SCRAM_SHA_256
    IAMAuth: REQUIRED
    SecretArn:
      Fn::GetAtt: [ Database, MasterUserSecret.SecretArn ]
    EngineFamily: POSTGRESQL
    RequireTLS: True

Lambda

Finally a Lambda function, security group and role which can connect to the RDS Proxy using IAM authentication. Now we can just use pg with the rds-signer package to work with Postgres the way you are used to!

const db = await getClient();
const result = await db.query('SELECT version()');

console.log(result.rows[0].version);

const db = await getClient();
const result = await db.query(
  'SELECT id, title, body FROM articles WHERE type = $1',
  ['database'],
);

for (const article of result.rows) {
  console.log(article);
}

Certificate

Even though AWS points you towards a certificate for RDS per region like https://truststore.pki.rds.amazonaws.com/eu-west-1/eu-west-1-bundle.pem this does not apply to RDS Proxy for which you will need to use https://www.amazontrust.com/repository/AmazonRootCA1.pem.

Warnings

Obviously this is just a stripped down demo with the basics and you need to modify the template to your own needs. Even though this is mostly a serverless setup, you will be charged for RDS Aurora and RDS Proxy per hour. Because RDS Proxy will stay connected, RDS Aurora will not be able to pause / scale back down to 0:

If your Aurora cluster has an associated proxy through Amazon RDS Proxy, the proxy maintains an open connection to each DB instance in the cluster and the instance doesn’t automatically pause.

Warning: By deploying this demo, you acknowledge that AWS services may incur costs. I, the author, am not responsible for any charges or usage fees associated with your AWS account. Deploy at your own discretion.

Database Management in a VPC

It seems many people are still struggling using database tools because their RDS cluster is running inside a VPC. I recommend you to read one of my previous articles to see the killer tool that solved this problem for me: Port7777.

Closing

After migrating to RDS Proxy, I did not run into performance or scalability issues anymore.

To close of, again the link to the demo: https://github.com/slootjes/aws-lambda-rds-proxy. Let me know if you think I've missed something crucial. In the next article I will explain how to use S3 features with RDS Aurora and update the demo. Have fun!

AWS AppSync Events vs IoT Core

Robert Slootjes — Tue, 19 Nov 2024 18:51:34 +0000

AWS recently announced AppSync Events and it looks like very useful service. However when I was reading about it, it just felt like this is "just" a layer on top of IoT Core which exists for many years. Let's find out if this is the case...

API Gateway Websockets?
A while ago I compared API Gateway Websockets with IoT Core. It might be very useful to also give it a read if you are exploring options for bi-directional communication on AWS.

Connection Management

Both services do their own connection management. There is absolutely no need to store who connected and disconnect yourself or do any garbage collection.

AppSync Events

AppSync Events is able to act on connects but only for deciding if a client is allowed to connect. It does not seem possible to have any interaction with the outside world (ie: databases, AWS services, APIs).

IoT Core

With IoT Core you are able act on connects and disconnect using Lifecycle Events and have total freedom of doing something with this data.

Messaging

Both services use a structure based on segments separated by slashes to create a hierarchy of your own choice, for example: news/gaming, news/sports or a level deeper with news/sports/f1 and news/sports/soccer.

AppSync Events

In AppSync Events this is called called "channel namespaces". A wildcard * is supported to receive messages from all sports using news/sports/* or all news events using news/*. There is a maximum of 5 segments (4 slashes).

You are required to set up all channel names and configure them with an optional function that triggers on connect and when an event is published. It does not seem to be possible to use a Lambda for this.

I was a bit surprised to find out that AppSync Events only uses the socket connection to receive messages. To publish a message, you need to use an HTTP endpoint. While this is perfect for only broadcasting from server to client, this makes it a lot less suitable for use cases where your clients often need to communicate back.

Update April 2025: AppSync Events now supports publishing over websockets.

IoT Core

As IoT Core is an MQTT Broker, it uses "topics" which are similar to the channel namespaces from AppSync Events.

You don't need to manage topics. You can just subscribe and publish to any topic on the fly. The only restriction your clients have are the IAM permissions you give them. If a client does something now allowed by it's policy, it will be disconnected. There is a maximum of 8 segments (7 slashes).

IoT Core is fully bi-directional, you can subscribe to topics, receive and publish messages over the same socket connection. Again, only if you allow this by the policy given to the client.

Authentication

AppSync Events

AppSync Events supports multiple authentication types right out of the box with no or minimal development work required. A custom flow is possible with Lambda.

IoT Core

Within IoT Core you can use a pre-signed url, a custom authorizer and/or certificates to manage client access. While this is not super difficult, it does require development and isn't as easy as AppSync Events.

The upside thing is that anything is possible. An IAM policy allows you to decide exactly which client can do what on your topics.

Event Handling

AppSync Events

AppSync Events allows you to provide an optional handler for a channel namespace:

Event handlers are functions that run on AWS AppSync's JavaScript runtime and enable you to run custom business logic. You can use an event handler to process published events or process and authorize subscribe requests.

This is very limited as it does not seem to be possible to use any AWS services, connect to a database or (third party) API. The only kind of integration that it supports is where AppSync Events is called by other services to publish a message.

Update April 2025: AppSync Events now offers several integrations (called data channels) like Lambda, DynamoDB and EventBridge.

IoT Core

In IoT Core on the other hand, you can create SQL like rules to capture messages, send them to an AWS service of choice (including Lambda) using actions to process them. This allows you to do virtually anything with the data that is sent from both clients and/or servers and also in any language of choice as long as it is supported by Lambda. Read this article if you want to learn more about Rules.

SELECT color AS rgb FROM 'topic/subtopic' WHERE temperature > 50

These queries are extremely powerful and allow data transformation and filtering.

Pricing

Both services have a generous free tier in the first 12 months but eventually you will have to pay for something of course.

AppSync Events

$0.08 per million connection minutes.
$1.00 per million Event API operations. All inbound messages published, outbound messages broadcast, event handlers invoked, and WebSockets operations, like client connection, subscription requests, and ping requests, are considered operations.

Full details and examples here.

IoT Core

IoT Core has more features, I will only list the ones relevant for this comparison (on us-west-1 / eu-west-1 regions).

$0.08 per million connection minutes.
$1.00 per million messages (first 1 billion)
- $0.80 per million messages (4 to 5 billion)
- $0.70 per million messages (5+ billion)
$0.15 per million rules triggered.

Full details and examples here.

As you can see, the pricing is very similar where IoT Core will be a bit cheaper since in AppSync Events as the rules engine is a lot cheaper and you will get a discount when having more than 4 billion messages (which a lot though if you ask me).

Service Quotas

The service quotas for AppSync Events and IoT Core have a few notable differences:

AppSync Events has a maximum message size of 1.2 Megabytes where IoT Core can do only 128 Kilobytes (hard limit).
AppSync Events has an outbound messages limit of 1.000.000 per second where IoT Core has a default (soft) limit of 20.000.
AppSync can do 2.000 connects per second, IoT core has a default (soft) limit of 500 per second.

Overall it seems that AppSync Events is way more generous with it's limits than IoT Core which makes it more suitable for a situations where you primarily want to broadcast to your clients. I had to adjust some of the IoT Core limits for a project and this was not a problem at all but it's always a hassle to do.

Infrastructure As Code

AppSync Events

With AppSync Events you are able to create multiple AppSync Events API instanced which could very useful for separation of environments or different services.

IoT Core

Every account only has a single and default IoT Core instance at time of writing which you don't need to create manually. If you have multiple applications on the same account, you would need to separate access using topic prefixes or IAM policies.

The endpoint for it is not available in CloudFormation, you would have to use a custom resource with a Lambda to retrieve it, quite cumbersome. Other than that, it is very well supported within CloudFormation.

WAF

AppSync Events

AppSync Events is supported by the AWS WAF. As far as I know, WAF does not support websockets and therefor most likely only the publish endpoint is protected.

IoT Core

IoT Core is not supported by the AWS WAF for the simple reason that there are no account level HTTP endpoints to protect for this service.

Additional Features

AppSync Events

This service is very straightforward and doesn't offer much other functionality then previously described.

IoT Core

IoT Core has many more features which can clearly be seen when looking at the difference in amount of documentation compared to AppSync Events. I don't want to pretend I have used all those features, I just want to highlight a few of them:

MQTT

Since IoT Core is a fully managed MQTT broker, it supports the following protocols: MQTT, HTTPS, MQTT over websockets and LoRaWAN. This can be very useful if you have different types of clients. When working with hardware that used MQTT, it's very useful to create a web version that emulates the hardware using the same topics and messages but instead doing it over websockets.

Retained Messages

Retained Messages is an MQTT feature supported by IoT Core where you can publish a message which the client receives when it gets back online. As of now, I could not find the possibility of doing this with AppSync Events.

Device Registry

In IoT Core there is a concept called "Things". This optional feature allows you to keep a registry of all your things/devices. You can group them and even created "shadows" for them.

Shadows provide a reliable data store for devices, apps, and other cloud services to share data. They enable devices, apps, and other cloud services to connect and disconnect without losing a device's state.

Conclusion

My take on the AppSync Events service is that it is most likely added to the portfolio to make it easier to implement push messaging from server to client which focuses on web and app development.

In general it offers a lot less functionality than IoT Core as it focuses primarily on publishing messages to clients and not the other way around. It is however (a lot) easier to get started with and part of the Amplify suite of services. Because of this, I do think that AppSync Events could become a service that will be embraced by many developers. Personally I do see some use cases where I would prefer AppSync Events over IoT Core, especially because of the more generous default quotas.

IoT Core has way more features but has a steeper learning curve and seems more aimed at hardware at first glance. This however doesn't mean it can not be used for web and app. If you are interested in using IoT Core for web and app "the easy way", read this series of articles to get started.

Update April 2025: Now that AppSync Events supports bi-directional communication and integrations with other services like Lambda, DynamoDB and EventBridge the service got a lot more interesting. Personally I still prefer IoT Core as it is still more flexible since you do not need to define your channel namespaces upfront and the policies of IoT Core seem to be more fine grained.

AppSync Events powered by IoT Core?

Looking at the different services I feel that AppSync Events technically could very well be a service that is built on top of IoT Core. However, I'm not so sure anymore looking at the absence of bi-directional communication and the very different service quotas. Maybe an AWS employee could shine a light on this?

AWS IoT Core Simplified - Part 4: Rules

Robert Slootjes — Fri, 18 Oct 2024 09:58:38 +0000

We are now able to connect clients to the server, either using a presigned url or a custom authorizer. Clients can send messages to each other but in many cases you want to do something with these messages server side. This can be done with rules.

Queries & Actions

A rule basically consists of 2 parts; first, a query to capture data. Queries use an SQL-like syntax which offer a lot of flexibility. Second, actions can send data to services like Lambda, SQS, DynamoDB, Kinesis and many more.

Rule Example

Creating a rule is not difficult but there are a few pointers. Let's look at this example to get started:

SensorUpdatesRule:
  Type: AWS::IoT::TopicRule
  Properties:
    RuleName: myapp-prod-sensor-updates
    TopicRulePayload:
      Sql: "SELECT * FROM 'updates/sensor/+'"
      AwsIotSqlVersion: '2016-03-23'
      Actions:
        - Sqs:
            QueueUrl: !GetAtt SensorEvents.QueueUrl
            RoleArn: !GetAtt RuleRole.Arn
            UseBase64: False

This will capture all messages published to wildcard topic updates/sensor/+ and sends them to an SQS queue. As we learned before, this includes topics like updates/sensor/room-1 and updates/sensor/room-2.

Permissions

As always, this rule needs the permissions to perform the configured actions, so let's add this:

RuleRole:
  Type: AWS::IAM::Role
  Properties:
    RoleName: myapp-prod-iot-rule
    AssumeRolePolicyDocument:
      Version: '2012-10-17'
      Statement:
        - Effect: Allow
          Principal:
            Service:
              - iot.amazonaws.com
          Action: sts:AssumeRole
    Policies:
      - PolicyName: myapp-prod-iot-rule-policy
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
            - Effect: Allow
              Action:
                - 'sqs:SendMessage'
              Resource:
                  - !GetAtt SensorEvents.Arn

Queue

And obviously we also need the queue we want to write to:

SensorEvents:
  Type: AWS::SQS::Queue
  Properties:
    QueueName: myapp-prod-sensor-updates

Now you can create a Lambda and add it as a trigger to the queue. You can also trigger a Lambda directly from the rule but I personally like to add a queue in between so I have more visibility and control over it (ie: batch size, concurrency, dead letter queue). If triggering Lambda directly has your preference, make sure to give your rule lambda:invoke permissions instead.

Queries

Queries are the most important part of a rule and it's very feature rich.

SELECT

SELECT is always required and defines which data you want to capture.

Capture everything
SELECT * -> {"temperature":21, "moisture": 50}

Select a specific field
SELECT temperature -> {"temperature":21}

Select multiple fields
SELECT temperature, moisture -> {"temperature":21, "moisture": 50}

Alias fields
SELECT temperature as temperature_celsius, moisture -> {"temperature_celsius":21, "moisture": 50}

Hard coded values
SELECT *, 1 as version -> {"temperature":21, "moisture": 50, "version": 1}

Single value
SELECT VALUE temperature -> 21

There are also many functions to transform your data in your query. I use this for instance to get the client ID and timestamp:

SELECT *, clientid() as clientId, timestamp() as timestamp

Another notable function is topic() which can be used to capture a segment of the topic as a variable:

SELECT *, topic(3) as room FROM 'updates/sensor/room-1'
->
{"temperature":21, "moisture": 50, "room": "room-1"}

FROM

The FROM part defines on which topic(s) the query should trigger.

Single topic
updates/sensor/room-1

Wildcard topic
updates/sensor/+ or updates/sensor/#

It's also possible to capture everything using a top level wildcard: #. This could be useful for analytics or debugging purposes but depending on how much data your clients are sending it could generate quite some data (and cost).

WHERE

The WHERE part is optional and allows you to narrow down which messages are captured. It's better to add the WHERE clause here than to discard the data in your Lambda function as it would simply save cost since the action won't run and the Lambda doesn't need to be executed.

Only capture high temperatures
SELECT * FROM 'updates/sensor/+' WHERE temperature > 25

High temperatures with low moisture
SELECT * FROM 'updates/sensor/+' WHERE temperature > 25 AND moisture < 40

All the basic operators you expect are supported out of the box.

Actions

Once you have queried the messages you want, you want to process them. In the example I've showed SQS as action but there are many more to choose from. You can for instance write directly to DynamoDB to query your data or log to CloudWatch to create nice metrics. Something I personally do is to write my events to a Firehose stream so I can easily log chunks of data to S3 per minute. Remember to always give your role the right permissions.

Saving money with Basic Ingest

A notable feature which could save you some money is basic ingest. With basic ingest you can let your clients send messages to the rule directly instead of sending it to the topic the rule is created for. A rule will create its own topic in this format: $aws/rules/{rule_name}. Sending messages to a rule directly instead of a topic saves you the cost of a message, depending on the amount of messages this could be very beneficial. The downside of this method is that other clients can not subscribe to it. Also make sure your client has permissions to publish to this topic, otherwise it will disconnect.

Closing

With this series of posts I hope to made it clear that IoT Core isn't as difficult as it seems. While IoT Core is very cheap (and has a generous free tier in the first 12 months!) there are obviously costs involved. IoT Core is however very cheap so fiddling around with it doesn't break the bank. Let me know in the comments what type of application you've built with it to inspire me and others.

AWS IoT Core Simplified - Part 3: Custom Authorizer

Robert Slootjes — Mon, 01 Jul 2024 08:54:17 +0000

Custom Authorizer

While the Presigned URL method can be very effective, sometimes you need a little bit more control. This can be done with a custom authorizer. Simply said, when a custom authorizer is used, IoT Core will invoke a Lambda function which needs to approve or deny the connection.

Creating

It's quite easy to create an authorizer with CloudFormation:

IoTAuthorizer:
  Type: AWS::IoT::Authorizer
  Properties:
    AuthorizerFunctionArn: !GetAtt IotAuthorizerLambdaFunction.Arn
    AuthorizerName: myapp-prod-custom
    SigningDisabled: True
    Status: ACTIVE

You will need to give it a unique name (this part is very important!) and point to a Lambda function that will receive the connection request and needs to decide if the client is allowed to connect, and with which permissions. I recommend to prefix with the name of the stack and the stage to be sure the name is always unique. We disable token signing as we do not need it for this use case.

Permissions

Make sure your authorizer has permission to invoke your Lambda function.

IoTAuthorizerPolicy:
  Type: AWS::Lambda::Permission
  Properties:
    FunctionName: !GetAtt IotAuthorizerLambdaFunction.Arn
    Action: lambda:InvokeFunction
    Principal: iot.amazonaws.com
    SourceArn: !GetAtt IoTAuthorizer.Arn

Without this permission, your Lambda function can not be invoked by IoT Core and your client won't be able to connect. This is something you can easily forget and it's quite a pain to figure out why it's not working.

Now, IoT Core knows which custom authorizer to use when connecting.

Connecting

MQTT clients have the option to provide a username and password when connecting, you will receive these in your custom authorizer.

Passing the authorizer

When connecting to IoT Core, you need to send along the name of the custom auhtorizer you want to use. With a websocket connection, this is a query string variable as part of the url:

wss://.../mqtt?x-amz-customauthorizer-name={authorizer}

Let's have a look at the Paho MQTT JS library again:

const url = 'wss://xxxxxxxx-ats.iot.eu-west-1.amazonaws.com/mqtt?x-amz-customauthorizer-name={authorizer}';
const client = new Paho.MQTT.Client(url, "the_client_id");
client.connect({
  useSSL: true,
  timeout: 3,
  mqttVersion: 4,
  userName: "the_username",
  password: "the_password",
  onSuccess: function () {
    console.log("connected!");
  },
  onFailure: function () {
    console.log('failed');
  }
});

Connecting by host

It is also possible to connect by host (xxxxxxxx-ats.iot.eu-west-1.amazonaws.com) and port instead of using the websocket url. Make sure to use port 443 and then add the authorizer to the username like this:

{username}?x-amz-customauthorizer-name={authorizer}

This allows you to use "regular" MQTT clients like MQTT.fx to connect as well. For you as developer there is no difference in how the connection is handled, which is good news.

Request

When connecting with the above connection details, the event from IoT Core in your custom authorizer Lambda looks something like this:

{
    "protocolData": {
        "mqtt": {
            "username": "the_username",
            "password": "dGhlX3Bhc3N3b3Jk",
            "clientId": "the_client_id"
        }
    },
    "protocols": [
        "mqtt"
    ],
    "signatureVerified": false,
    "connectionMetadata": {
        "id": "c979e717-32c4-7309-b46b-56ab0c5e36b4"
    }
}

As you can see, it contains the client ID, username and password that was sent when connecting. The password however is base64 encoded so make sure to decode it before using it. Please note that the username and password can be empty as they are not mandatory.

When connecting over a websocket connection you will also receive the query string and headers that were part of the request in an "https" block in the "protocolData". Personally I'm sticking to the username and password since it will work for both websocket clients and other MQTT clients that use the hostname and port.

Response

The response structure for a custom authorizer needs to look like this:

{
  "isAuthenticated": true,
  "principalId": "user1",
  "disconnectAfterInSeconds": 86400,
  "refreshAfterInSeconds": 3600,
  "policyDocuments": [
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": [
            "iot:Connect",
            "iot:Subscribe",
            "iot:Receive"
          ],
          "Resource": [
            "arn:aws:iot:{region}:{account-id}:client/user1-*",
            "arn:aws:iot:{region}:{account-id}:topicfilter/updates/user/user1",
            "arn:aws:iot:{region}:{account-id}:topic/updates/user/user1"
          ]
        }
      ]
    }
  ]
}

The most important keys are "isAuthenticated" and "policyDocuments".

"isAuthenticated" let's IoT Core know if the client is allowed to connect. If this is false, the connection is refused. If it's true, the connection is allowed and the policy is applied to the connection. In this example, we allow a client to connect with clientId wildcard of "user1-*" and to subscribe and receive messages on topic "updates/user/1". The principalId is only allowed to contain a-z, A-Z, 0-9 with a maximum length of 128.

An example custom authorizer Lambda:

import { IoTCustomAuthorizerEvent } from 'aws-lambda';
import { IoTCustomAuthorizerResult } from 'aws-lambda/trigger/iot-authorizer.js';

export const handle = async (event: IoTCustomAuthorizerEvent): Promise<IoTCustomAuthorizerResult> => {
  let isAuthenticated = false;
  const statement = [];

  let { username, password } = event.protocolData?.mqtt ?? {};
  if (username && password) {
    // authenticate any way you like
    isAuthenticated = true;
    statement.push({
      Action: [
        'IoT:Connect',
        'IoT:Subscribe',
        'IoT:Receive',
      ],
      Effect: 'Allow',
      Resource: [
        // allow client to connect with "{username}-{something}
        `arn:aws:iot:*:*:client/${username}-*`,
        // allow client to subscribe to "updates/user/{username}
        `arn:aws:iot:*:*:topicfilter/updates/user/${username}`,
        // allow client to receive messages from "updates/user/{username}
        `arn:aws:iot:*:*:topic/updates/user/${username}`,
      ],
    });
  }

  return {
    isAuthenticated,
    principalId: username ?? 'empty'.replace(/[^a-zA-Z0-9]/ug, '').slice(0, 128),
    disconnectAfterInSeconds: 86400, // 24 hours
    refreshAfterInSeconds: 3600, // 1 hour
    policyDocuments: [
      {
        Version: '2012-10-17',
        Statement: statement,
      },
    ],
  };
};

Authentication

Since this article focuses on IoT Core, I won't cover the authentication process in the example code. Obviously this is not production ready code and there are many ways how you can authenticate your client credentials, to name a few:

Look up the username in DynamoDB (or other database of choice) and verify the password (make sure to use something like bcrypt for storing and comparing passwords!)
Pass a Json Web Token (access or id token from Cognito or other identity provider) as the password so the Lambda only needs to verify the token and then use what's inside (ie: the subject/user ID).
Call an external API to verify the credentials.

The great thing is that you can use kind of any method for it. You receive the credentials, verify them and define exactly what the client can do. You can also set up multiple custom authorizers with their own type of authentication. It is a bit more work than the presigned url but this is more flexible if you have clients that need different permissions.

Debugging

Sometimes you might not really be sure why your client is not able to connect. Luckily IoT Core allows you to enable logging. At the IoT Core setting page, go to "Manage Logs" and choose the log level you want. When developing the Debug level will be the most helpful. Don't forget to disable the logger once you put something in production.

Common issues

IoT Core doesn't have permission to invoke your Lambda function
The name of the authorizer has a typo in it
The password wasn't base64 decoded before using it
The principal ID contains invalid characters
The policy doesn't allow the client ID to connect
The response from your Lambda is malformed in other ways

Up Next

In the next part of the series I will explain how you can leverage rules to do something with the messages sent to the server.

AWS IoT Core Simplified - Part 2: Presigned URL

Robert Slootjes — Thu, 13 Jun 2024 06:25:48 +0000

This is part 2 in a series of articles about IoT Core:

Parts coming up:
Part 3: Connect using a custom authorizer
Part 4: Topic Rules

Connect using a presigned url

Similar to S3, it's possible to create a presigned url for IoT Core. You can even cache and reuse this presigned url for multiple clients as long as the client id is different per client. If you connect a client with a client id that is already in use, the other client will be disconnected. This can be a great method for pushing content to clients but also allows them to publish things themselves if your use case requires it. While this is a super easy method, it isn't the most flexible way of using it in terms of permissions. Please note that this only works for websocket urls, not for regular MQTT connections.

Permissions

It's important to realize that the presigned url has the same permissions as the role that was used to sign it with. That means that all clients will have the exact same permissions unless you specifically create and use a different role per use case. For example, if your Lambda has a role that allows connecting with any client ID on any topic, every client will be able to connect to any topic. In some cases this is totally fine but make sure this is OK for your use case. Obviously, you can (and you should) restrict your Lambda to only allow what you want your clients to do.

For instance, if you have a website where you want to be able to push live news updates to an end user, you can have a policy with this in it:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Connect",
        "iot:Subscribe",
        "iot:Receive"
      ],
      "Resource": [
        "arn:aws:iot:{region}:{account-id}:client/user-*",
        "arn:aws:iot:{region}:{account-id}:topicfilter/news",
        "arn:aws:iot:{region}:{account-id}:topic/news"
        ]
    }
  ]
}

This will allow the client to connect with any client ID as long as it starts with user- (you could generate a uuid to make it random), and will allow to subscribe to the "news" topic and receive messages over it. This is now basically a read only connect as the client isn't allowed to publish any messages with this policy.

You can have a separate role for your publishers that allows to write updates to the topic:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Connect",
        "iot:Publish"
      ],
      "Resource": [
        "arn:aws:iot:{region}:{account-id}:client/publisher-*",
        "arn:aws:iot:{region}:{account-id}:topic/news"
      ]
    }
  ]
}

Alternatively, you can of also publish a message using the AWS SDK instead of doing this with a client that is connected with MQTT. The iot:Publish permission to the respective topic is still required obviously.

Code

This seems to be quite hidden in the documentation however I did find some example code. I converted this into something that makes use of the current AWS utilities which makes it a lot more compact and readable. The code examples are written in Typescript, I recommend using Serverless Framework with esbuild to deploy the code so you can run it with a node runtime in Lambda.

import * as crypto from 'node:crypto';
import { type BinaryLike } from 'node:crypto';
import { Sha256 } from '@aws-crypto/sha256-js';
import { type AwsCredentialIdentity } from '@aws-sdk/types';
import { SignatureV4 } from '@smithy/signature-v4';

const sha256 = (data: BinaryLike): string =>
  crypto.createHash('sha256').update(data).digest().toString('hex');

const toQueryString = (queryStrings: Record<string, string>): string => Object.entries(queryStrings)
  .map(([key, value]) => `${key}=${value}`)
  .join('&');

export const getSignedUrl = async (
  host: string,
  region: string,
  credentials: AwsCredentialIdentity,
  expiresIn = 900,
): Promise<string> => {
  const service = 'iotdevicegateway';
  const algorithm = 'AWS4-HMAC-SHA256';

  const sigV4 = new SignatureV4({
    sha256: Sha256,
    service,
    region,
    credentials,
  });

  const date = new Date().toISOString().replaceAll(/[:-]|\.\d{3}/gu, '');
  const credentialScope = `${date.slice(0, 8)}/${region}/${service}/aws4_request`;

  const parameters: Record<string, string> = {
    'X-Amz-Algorithm': algorithm,
    'X-Amz-Credential': `${encodeURIComponent(`${credentials.accessKeyId}/${credentialScope}`)}`,
    'X-Amz-Date': date,
    'X-Amz-Expires': expiresIn.toString(),
    'X-Amz-SignedHeaders': 'host',
  };

  const path = '/mqtt';
  const headers = `host:${host}\n`;
  const canonicalRequest = `GET\n${path}\n${toQueryString(parameters)}\n${headers}\nhost\n${sha256('')}`;
  const stringToSign = `${algorithm}\n${date}\n${credentialScope}\n${sha256(canonicalRequest)}`;

  parameters['X-Amz-Signature'] = await sigV4.sign(stringToSign);
  if (credentials.sessionToken) {
    parameters['X-Amz-Security-Token'] = encodeURIComponent(credentials.sessionToken);
  }

  return `wss://${host}${path}?${toQueryString(parameters)}`;
};

I can now create a Lambda with the following code:

import { type APIGatewayProxyResultV2 } from 'aws-lambda';
import { getSignedUrl } from '../../Service/IoTCore.js';

export const handle = async (): Promise<APIGatewayProxyResultV2> => ({
  statusCode: 200,
  body: JSON.stringify({
    url: await getSignedUrl(process.env.AWS_IOT_HOST ?? '', process.env.AWS_DEFAULT_REGION ?? '', {
      accessKeyId: process.env.AWS_ACCESS_KEY_ID ?? '',
      secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY ?? '',
      sessionToken: process.env.AWS_SESSION_TOKEN ?? undefined,
    }),
  }),
});

and when calling it, I receive a URL I can use to connect with something like a Paho MQTT Client. For this example I've set a AWS_IOT_HOST environment variable with the endpoint of IoT Core (xxxxxxx-ats.iot.eu-west-1.amazonaws.com). The endpoint can be found by navigating to the IoT Core service in the dashboard and then going to Settings:

You can of course also use the IoT Core SDK to retrieve the endpoint.

Connecting

Now that you have a way of getting a presigned url, you want to use it to connect to IoT Core. To do this from a browser, you can use the Paho MQTT library and following snippet of code:

const client = new Paho.MQTT.Client(url, clientId);
client.connect({
  useSSL: true,
  timeout: 3,
  mqttVersion: 4,
  onSuccess: function () {
    console.log("connected");
  },
  onFailure: function () {
    console.log("failed to connect");
  },
});
client.onMessageArrived = function (message) {
  console.log(message);
};
client.onConnectionLost = function (e) {
  console.log("lost connection", e);
};

The url is the response from your Lambda. Make sure the clientId is allowed by your policy as otherwise it will refuse to connect.

You can subscribe to a topic like this:

client.subscribe('updates/"');

and publish a message like this:

const message = new Paho.MQTT.Message(payload);
message.destinationName = 'updates/messages';
client.send(message);

Please note, once again, doing anything not allowed by your policy will result in a disconnect.

Caching

Presigned urls from a Lambda only work for a limited time due to how IAM works but it's possible to cache it in a CDN like CloudFront for a couple of minutes. This way you do not need to generate a fresh url for every visitor. With my configuration, the presigned urls are valid for a maximum of 5 minutes. In practice, I cache them for 1 minute to be on the very safe side.

Summary

You now have a basic way of working with IoT Core that allows for powerful bidirectional communication. Have fun!

In part 3 I will explain how a custom authorizer can be used to do more fine grained permissions per client.

AWS IoT Core Simplified - Part 1: Permissions

Robert Slootjes — Tue, 11 Jun 2024 18:46:17 +0000

AWS IoT Core is an amazing (and often overlooked) service as I said before when comparing IoT Core to API Gateway Websockets. To summarize, IoT Core manages it own connections, it has a powerful system of topics and rules, it scales well and, not unimportant, it's quite cheap.

However, IoT Core can also be intimidating and difficult to start with because of things, fleets, shadow devices and certificates. The good news is that those things are not at all mandatory to use this service. While maybe not obvious at first sight, there are ways to just connect some clients to IoT Core without first registering or managing them.

I will be posting a few articles to cover basic usage of IoT Core aimed at developers who are looking for an easy and flexible way of bi-directional communication between clients and server using websockets.

Part 1: Introduction & Permissions (this one!)
Part 2: Connect using a presigned url
Part 3: Connect using a custom authorizer
Part 4: Topic Rules

What is IoT Core?

Simply put, and maybe not giving enough credit, IoT Core is an MQTT broker. MQTT is a lightweight publish-subscribe protocol. The MQTT broker acts as the central hub between connected clients that can send and reach messages to each other. Messages are sent over topics. A topic is a hierarchical string separated by slashes, for instance sensor/temperature/room1. A client can send temperature updates to this topic every minute and another client might be subscribed to this topic and receives all temperatures that are sent to do something with it.

MQTT has many advanced options and use cases that I won't cover here. If you want to know more about them I recommend reading more here. For the rest of this series, it's enough to know that there is a central broker (IoT Core) and clients (ie: a web browser) that can publish and receive messages.

Topic Wildcards

In a topic, a + character can be used as a wildcard for one level while a # character can be used for a multi level wildcard. It is possible to have more than 1 single level wildcard, like sensor/+/+ however it's not possible to have more than 1 multi level wildcard (which makes sense).

Looking at the previous example, if a client is interested in temperatures of all rooms, it can subscribe to sensor/temperature/# where the # character is a wildcard in MQTT. This means that if a client wants the informations of all sensors, it can listen to sensor/#. It is also possible to subscribe to sensor/+/room1 to receive all messages (temperature and others) of room1.

Security

For simple use of IoT Core, there are 4 permissions that you need to know about. Applying least-privilege permissions obviously also applies to IoT Core. In case a client performs an action that is not allowed by it's policy, the client will be disconnected by the server.

It's important to realize that MQTT wildcards are treated as literal characters in IAM permissions. This means that you can use the ?, *, + and # in the IAM policy but the ? and * will be treated only as wildcard for the IAM policy itself while + and # are treated as wildcards for topics in IoT Core. I will explain this using several examples.

Permissions

iot:Connect

First, your client needs to be able to connect to the server. The client ID must be unique per region, otherwise the previously connected client with the same client ID will be disconnected.

Examples
arn:aws:iot:{region}:{account-id}:client/*
Allows to connect using any client ID.

arn:aws:iot:{region}:{account-id}:client/sensor-123
Allows to connect as sensor-123.

arn:aws:iot:{region}:{account-id}:client/sensor-???
Allows to connect using any client ID as long as it starts with sensor- and ends with 3 characters. This means that sensor-123, sensor-foo will work, but sensor, sensor-foobar and sensor-123456 won't work.

arn:aws:iot:{region}:{account-id}:client/sensor-*
Allows to connect using any client ID as long as it starts with sensor-.

arn:aws:iot:{region}:{account-id}:client/user-????????-????-????-????-????????????

Allows to connect using user-{uuid}.

iot:Subscribe

Before a client can receive messages, the client must first subscribe to one or multiple topics.

Examples
arn:aws:iot:{region}:{account-id}:topicfilter/updates
Allows to subscribe to updates.

arn:aws:iot:{region}:{account-id}:topicfilter/updates/sensor-???
Allows to subscribe to updates/sensor-123, updates/sensor-foo and other variants as long as the topic starts with updates/ and then has a second level starting with sensor- and then 3 characters. It won't accept updates/foo-123 or "updates/sensor-123/foo".

arn:aws:iot:{region}:{account-id}:topicfilter/updates/sensor-*
Allows to subscribe to updates/sensor-123, updates/sensor-123/foo, and/or "updates/sensor-123/foo/bar/baz" because the * allows for any level of depth in the topic.

arn:aws:iot:{region}:{account-id}:topicfilter/updates/sensor-*/???
Allows to subscribe to updates/sensor-123/foo, updates/sensor-12345678/bar and other variants as long as the topic starts with "updates/sensor-" and has a second level with exactly 3 characters. It won't accept updates/sensor-1 or updates/sensor-12345/#.

You can of course use the wildcards in MQTT too if you want.

arn:aws:iot:{region}:{account-id}:topicfilter/updates/+
Allows to subscribe to topic/updates/+ only as the + is treated as a literal in the policy.

arn:aws:iot:{region}:{account-id}:topicfilter/updates/#
Allows to subscribe to topic/updates/# only as the # is treated as a literal in the policy.

iot:Receive & iot:Publish

iot:Receive allows the client to receive messages over the topics that it is subscribed to while iot:Publish allows the client to publish messages that can be received by other clients and/or topic rules.

The iot:Receive & iot:Publish permissions have the same structure as iot:Subscribe with the only difference that it uses "topic" instead of "topicfilter". Usually the iot:Subscribe and iot:Receive will be the same as it doesn't make sense to allow receiving on topics the client isn't allowed to subscribe on and vice versa.

Examples
arn:aws:iot:{region}:{account-id}:topic/updates
arn:aws:iot:{region}:{account-id}:topic/updates/sensor-???
arn:aws:iot:{region}:{account-id}:topic/updates/sensor-*
arn:aws:iot:{region}:{account-id}:topic/updates/sensor-*/???
arn:aws:iot:{region}:{account-id}:topic/updates/+
arn:aws:iot:{region}:{account-id}:topic/updates/#

Creating a policy

Since the permissions and values have different formats, you can easily combine everything in a single policy:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "iot:Connect",
        "iot:Subscribe",
        "iot:Receive"
      ],
      "Resource": [
        "arn:aws:iot:{region}:{account-id}:client/sensor-*",
        "arn:aws:iot:{region}:{account-id}:topicfilter/sensor-*/???",
        "arn:aws:iot:{region}:{account-id}:topic/sensor-*/???"
        ]
    }
  ]
}

In the next article I will explain how to connect to IoT Core using a presigned url.

Amazon EventBridge Event Envelope

Robert Slootjes — Tue, 02 Jan 2024 10:41:55 +0000

Amazon EventBridge Event Bus is a serverless event bus that helps you receive, filter, transform, route, and deliver events. It's a very powerful service that allows developers to work with events without worrying about scaling and it offers a very flexible system of rules to send events to virtually any destination.

Standard

A standard event in EventBridge looks like this:

{
    "version": "0", // set by AWS
    "id": "9773f4b3-ff25-5ef2-b41b-791f868d424e", // set by AWS
    "detail-type": "your_event_type", // set by you
    "source": "your_app", // set by you
    "account": "123456789", // set by AWS
    "time": "2023-12-15T14:41:33Z", // set by you, optional
    "region": "eu-west-1", // set by AWS
    "resources": [], // set by you, optional
    "detail": {} // set by you, optional
}

While having a source, type and time is a very good base for an event I personally like to add some more details. Occasionally multiple events are published in the same request and then it can be very useful to trace them back to see what happened in which order. Since EventBridge doesn't guarantee the order of events, it's not reliable to depend on the service to process everything in the right order.

With Metadata

This example shows how I'm populating the detail key with some extra metadata:

{
    "version": "0", // always 0
    "id": "9773f4b3-ff25-5ef2-b41b-791f868d424e",
    "detail-type": "your_event_type",
    "source": "your_app",
    "account": "123456789",
    "time": "2023-12-15T14:41:33Z",
    "region": "eu-west-1",
    "resources": [],
    "detail": {
        "createdAt": 1702651293210,
        "correlationId": "2dc5e557-dcc9-4aef-9794-be207ca1a011",
        "sequenceNr": 1,
        "idempotencyKey": "custom_idempotency_key:123",
        "persist": true,
        "context": {
            "foo": "bar"
        }
    }
}

As you can see, I'm adding a few things:

createdAt
Milisecond unix timestamp.

correlationId
A uuid which is the same for all events published in the same request.

sequenceNr
An increasing number for every event published in the same request.

idempotencyKey
A custom idempotency key so that if for some reason an event got published multiple times (EventBridge provides at-least-once event delivery to targets). The consumer can ignore an event if it was previously handled.

persist
A boolean to indicate if I want to persist this event to S3 using a rule that sends it to Firehose. If you want to learn more, I recommend you to read this article: Logging EventBridge events to S3 with Firehose.

For some projects, I use the same rule to persist the events in DynamoDB so I can easily look up events by ID, correlation ID (sorted by the Sequence Nr) and event type (sorted by timestamp) using Global Secondary Indexes.

context
This is the data that belongs to the event, you can put here what ever you like.

Benefits

Having a correlationId, sequenceNr and createdAt allows you to see which events were published in the same request and in which order. The idempotencyKey adds reliability to the application as it can prevent to do double work (depending on the type of event, this can have quite some impact).

Persisting events can be very useful. It's super useful for debugging but also a gold mine for auditing and analytical purposes.

AWS RDS & ElastiCache remote access with Port7777

Robert Slootjes — Fri, 10 Nov 2023 13:18:04 +0000

Sometimes you just need to be able to quickly access a database that's inside a VPC. I stumbled upon a tool called Port7777 and it promises that it can do exactly that for both RDS and ElastiCache.

Disclaimer This is not a sponsored post. My employer paid for a team license earlier this year and I am genuinely happy with it and I think it can greatly benefit others too.

Port7777

So, what is it and how does it work?

7777 automates the work of creating and using a jump server to reach private databases in AWS VPC.

By internally using AWS Fargate containers instead of EC2 instances, the setup time and costs are drastically reduced. 7777 creates jump servers using containers on the fly, and deletes them after you are finished.

Finally, everything runs in your AWS account, including the Fargate containers. Nothing is sent to 3rd party servers.

Simply said, the tool uses a Fargate container to setup an SSH tunnel between the remote database and your device and cleans it up automatically afterwards so you don't have a server running 24/7.

Pricing

This tool is cheap, very cheap, maybe a bit too cheap for what it solves if you ask me:

$19 for a solo license, 1 user with unlimited databases, unlimited tunnels
$99 for a team license, unlimited users allowing it to be used in your team and CI/CD pipelines

Note that both licenses are a one-time purchase (!).

Obviously running the Fargate container itself costs some money but this costs only a few dollar per month per tunnel assuming you don't need it all the time.

Security

While this tool is developed by 2 very trustworthy people I didn't like the aspect of this tool grabbing my AWS credentials and doing something with it in a black box. I gave myself the task to come up with a solution where we are in control of the credentials the tool could access.

Isolation with Docker

To increase security I looked into the offered option of running 7777 in Docker and manually setting up the stack. This way I can be sure that the tool runs completely isolated and only has access to the AWS credentials that I specifically specified with the correct policy. While there was some documentation on manual installation available it took me a while to figure it out.

Basically the steps are:

Install Port7777 stack using CloudFormation
Create a policy + user with minimal permissions
Store credentials for this user in an env file
Start Port7777 inside a Docker container passing the env file containing the AWS permissions

Sharing is caring

I created a repository which contains the templates to set up this tool manually in your AWS account and a shell script that makes it similar in use to the original tool. I shared it previously with the other developers within my company but I thought it might also be helpful to others so I put it on my GitHub account for everyone to use: https://github.com/slootjes/port-7777-isolated.

Enterprise

Recently they also added an enterprise license which is a subscription-based plan where they allow a customer to fork the project in a private GitHub repository. This allows you to analyse the code, run security scanners and self-build the tool. I don't know about pricing details on this subscription so you will have to contact them about it yourself if you're interested.

Optimizing Typescript packages in Serverless Framework with esbuild

Robert Slootjes — Fri, 12 May 2023 12:44:54 +0000

Update April 14, 2025
The /u modifier is breaking for the latest Go-based typescript compiler as the modifier is not valid in Go. Luckily just removing the modifier is enough to make it work again.

Update December 16, 2024
This article applies to Serverless Framework v3 which is no longer maintained. You can seamlessly switch to the open source fork which is a drop-in replacement.

I'm a big fan of Serverless Framework as it takes care of a lot of things behind the scenes and offers many plugins. I do my work in Typescript and I was using serverless-plugin-typescript that takes care of transpiling into Javascript. Next to that, I was using a combination of serverless-plugin-common-excludes and a huge custom exclude list to reduce the size of the package (zip files) being uploaded to Lambda.

Doing a deploy would take around 2 minutes (on a Intel Core i5-13600K processor with 32GB RAM for who is interested) and in my project resulted in a 4MB package. While 4MB isn't a lot, it still contained a lot of code which wasn't actually used. The cold start time of the Lambda was around 4 seconds which bothered me a lot. While this seems to be caused by an unresolved issue at AWS with the node 18 runtime, I felt like I shouldn't sit around and wait for a solution.

When looking for options to reduce the package size I stumbled upon the serverless-esbuild plugin which uses esbuild which is supposed to be super fast (spoiler alert: it is!). While the plugin works without configuration it felt like it was capable of more.

Goals

I hoped to achieve the following things:

Fast builds

While local invocation and tools like LocalStack are awesome, I prefer to just upload to AWS and run everything on the actual (dev) environment. Waiting for a deploy of several minutes can be really annoying so the faster it builds, the faster I can work.

Small package size

Having a small package size has multiple advantages:

Faster deployments
Lower cold start time

Proper stack traces

When something goes wrong unexpectedly and you go to CloudWatch it's very painful to only see unreadable stack traces because of bundled, minified code. This is where source maps come in. Source maps can be used to transform bundled code back into their original file locations and line numbers.

The problem I faced with source maps is that once generated, they were huge (1 MB code, 8 MB source map) as it included all the data from all external packages while usually the errors are within my own code. In case something went wrong, it would take around 4 seconds to load and process the source map to generate the stack trace, that was unacceptable to me as Lambda is paid for by the millisecond. The slow processing of source maps seems to be an issue from nodejs itself.

Result

After fiddling around with configurations and reading many issues on Github I eventually ended up with the following configuration that seems to tick off the boxes for me:

Configuration

serverless.yaml

plugins:
  - serverless-esbuild # enables esbuild plugin

package:
  individually: true # an optimized package per function

custom:
  esbuild:
    config: './esbuild.config.cjs' # external config file

environment:
  NODE_OPTIONS: '--enable-source-maps' # use source map if available

esbuild.config.cjs

const {Buffer} = require('node:buffer');
const fs = require('node:fs');
const path = require('node:path');

// inspired by https://github.com/evanw/esbuild/issues/1685
const excludeVendorFromSourceMap = (includes = []) => ({
    name: 'excludeVendorFromSourceMap',
    setup(build) {
        const emptySourceMap = '\n//# sourceMappingURL=data:application/json;base64,' + Buffer.from(JSON.stringify({
            version: 3,
            sources: [''],
            mappings: 'A',
        })).toString('base64');

        build.onLoad({filter: /node_modules/}, async (args) => {
            if (/\.[mc]?js$/.test(args.path)
                && !new RegExp(includes.join('|'), 'u').test(args.path.split(path.sep).join(path.posix.sep))
            ) {
                return {
                    contents: `${await fs.promises.readFile(args.path, 'utf8')}${emptySourceMap}`,
                    loader: 'default',
                };
            }
        });
    },
});

module.exports = () => {
    return {
        format: 'esm',
        minify: true,
        sourcemap: true,
        sourcesContent: false,
        keepNames: false,
        outputFileExtension: '.mjs',
        plugins: [excludeVendorFromSourceMap(['@my-vendor', 'other/package'])],
        banner: {
            // https://github.com/evanw/esbuild/issues/1921
            js: "import { createRequire } from 'module';const require = createRequire(import.meta.url);",
        }
    };
};

What this does:

Targets the node version as configured in Serverless Framework (in my case: node 18)
Package every function into a separate, optimized single file using tree shaking and minification
Using ESM (instead of CJS) allowing things like top level await which can be very useful
Create empty source maps for external packages except for manually enabled ones to prevent having a giant source map
Includes a "banner" to support packages that don't support ESM yet.
Works on Windows and *nix machines

The Good

Extremely fast build process - it only takes miliseconds to transpile Typescript and package
Really small package size
Cold start went from around 4 seconds to 1.5 second, a massive improvement!
Readable stack traces for my own source code and manually enabled vendor code

The Less Good

Even with a smaller source map it still takes node quite some time to process an error - this delay is around 100ms - 400ms for my use cases. Not super fast but way more acceptable than the 4 to 5 seconds without optimization obviously.
If something unexpected now happens within vendor code, it won't show up in the stack trace unless it was configured to be included in the source maps.

Verdict

While I was a bit skeptical at first to switch to esbuild, I'm quite happy with the end result. I hope this helps someone else to find a good configuration for their use case. If you have some tips or tricks that I've missed, please share them in the comments.

QLDB to EventBridge using Pipes

Robert Slootjes — Sat, 04 Mar 2023 12:26:56 +0000

Last year I did a project with AWS Quantum Ledger Database or simply QLDB. QLDB is a fully managed ledger database which keeps track of every change through a journal. This makes it possible to see the complete history inserts, updates and deletes but also queries being fired at it. As a database it has its limitations but this is a very nice product if having a reliable audit trail is crucial.

For this project I was looking for a reliable way to act on changes in the database using EventBridge without the application publishing it. I prefer to do everything async as much as possible to avoid the database record to have changed and have the application fail to publish the event as that means I lost valuable information. The good news is that QLBD supports Streams which sends it data to a Kinesis Data Stream. This way, you do not have to worry about losing any data. It sounds pretty easy and it kind of is but the bad news is that the records from QLDB are in the ion format and on top of that, Kinesis also aggregates the records.

An event from QLDB in a Kinesis Data Stream looks something like this:

[
  {
    "eventSource": "aws:kinesis",
    "eventVersion": "1.0",
    "eventID": "shardId-000000000000:49636282114612585653977882531504014001922575972444405762",
    "eventName": "aws:kinesis:record",
    "invokeIdentityArn": "arn:aws:iam::123456:role/qldbpipe-ledger-stream-pipe-role",
    "awsRegion": "eu-west-1",
    "eventSourceARN": "arn:aws:kinesis:eu-west-1:123456:stream/qldbpipe-dev-ledger",
    "kinesisSchemaVersion": "1.0",
    "partitionKey": "aa2f52e2b20be04ac9e38683dff18447fecf9e89a62bf5744428e6b650138f37",
    "sequenceNumber": "49636282114612585653977882531504014001922575972444405762",
    "data": "4RVR6l4S5ZXU3xYwy74S24ZosXIzL3IpQNWkHOAlzeAcP29pQWI5tXNYtXc5sX9yQZozsX9arFWbQYAct3FZt3IpPN5bJNJBt2MouNMlP2MFs4ZFtdWlt2WauXcmsbcbaf5zsX9arZIgsNMquXWktZczsX9arFyyt2zCQN5FtdcctFyyt2zFbOSpQOQgs3MqHdomP2kZPOEfaf9cseIprNMqJXWqrVogt3JFa3IpPN5qPNEFrN9lJN5ds4gquXWFyNZcseIqzOEFPOIcsNMluZcquXWpuWIgsNNFa3EFPOIcsNM9uVIgQ2MquE4UnPhFqdWpsagyu3D68NobPagcuJZ3QOEFCKV6FUbpDab3FKLpDUtoFeEFtdMysJ9osXIztXcnQJZbQOPmDYuQKZtnDKMmNOPqEWgkIMckJbgXDPlEHboGHFkwLZM3LWJNPqvRk2E3g2FagQWJ3uVuOW6rZQKJKAArOMWPaSLsYRnapZTW51FcbgZud9nrdHpJL9ADZQZEWEZrcANQqrIrfRG5fpLa5FxnpTJihThCZCzjxmxJjeayfGw8PIY/j+vzrPi9OIVBFr2LSFGE5FlfWdZUhLOZX34TgjOBGTyb7vGCYOOPeF/3Fck2ehCuuUucB6xd7uSYSN/v92EwhOF4pQZZNDOKmy3ZIlFEjlGLUpSdK2MmjNlfEdbg//sUl838Ny84FwbqcrgmIEolPaJ69vF7n+MiJ6ZfB6xuU2SGVGaxanFqZBQGHLRs9ZSxIVH7Ymd2ZzsEHqbwllN3lJOmlYv35zFiWEWKVMUMTRhZVQJKFFxrN5ds3AkPOIgs25wt2EfQNZyCeMqQOAwuXWzsX7qdNlRU+rDcZ+Ke8DUn5hlfC67Zfb4/laSZmVAdyY5sSZY+BuFxQFoFIBRh4f4Jmh3",
    "approximateArrivalTimestamp": 1671549572.187
  }
]

Great, now what? If this was published to EventBridge, adding filter rules to this data isn't really possible...

What I did a year ago is that I had a Lambda function as a trigger on the Kinesis Data Stream that deaggregates the records and converts ion into json. Then I would publish it to EventBridge. This worked perfectly fine but when EventBridge Pipes launched I wanted to see if I could reduce the amount of code I needed to achieve the same result.

In this scenario a Kinesis Data Stream is the source and the EventBridge is the target. The only glue required is a Lambda that takes care of the enrichment which in my case is deaggregating and converting from ion to json.

As I strongly believe you shouldn't be clicking things in the console but rather use CloudFormation I created a fully working proof of concept using Serverless Framework which you can find on my Github. In this proof of concept I'm just the logging the events to S3 but you can tweak this to suit your needs of course. Also, if anyone has a better way of converting ion to json, please share it as I went for quick and dirty (it is very effective however :-).

Data flow

With the proof of concept a SELECT query done on QLDB would look something like this when published to EventBridge:

{
  "version": "0",
  "id": "4dad5941-44b2-d3cc-4373-5763ebec04cb",
  "detail-type": "Event from aws:kinesis",
  "source": "Pipe qldbpipe-dev-ledger",
  "account": "123456",
  "time": "2023-03-04T09:41:35Z",
  "region": "eu-west-1",
  "resources": [],
  "detail": {
    "qldbStreamArn": "arn:aws:qldb:eu-west-1:123456:stream/qldbpipe-dev/0wYOW015oYv34ZmEYmJJF2",
    "recordType": "BLOCK_SUMMARY",
    "payload": {
      "blockAddress": {
        "strandId": "EKwDuqzkVSI2IiuEb0Tlp1",
        "sequenceNo": 653
      },
      "transactionId": "EFcCQMkOEPXDj5aVyogJbj",
      "blockTimestamp": "2023-03-04T09:41:35.292Z",
      "blockHash": "Z6gKNWpJ+kmeHLVmr2ZjT05ca4XhdbL6YU6Y5phGcEY=",
      "entriesHash": "AhDroVPQiFYltthOkVoGlwONAmG+tIBrvlzHVColmO7=",
      "previousBlockHash": "rGsA9+leZON3iZGRqW8PB2CaWVc0Lminy1DOSrbpKZ7=",
      "entriesHashList": ["AhCD3Slvf3F+wCazKP12zK0upXJ44/weIsfd7ceaFvI=", "", "a60i/33Xewkp2koI7+06NkuAt8v5qRGpKjCBVqdfe17="],
      "transactionInfo": {
        "statements": [{
          "statement": "SELECT * FROM tickets",
          "startTime": "2023-03-04T09:41:35.259Z",
          "statementDigest": "ux//sy9PGxtDcFw6VzZkzxssE74KCOA3vCULzGs/vCc="
        }]
      }
    }
  }
}

Of course this is still very verbose but at least I can add filter rules for this in EventBridge. Optionally you could also adjust the Lambda code to filter records or remove information you don't need. For instance, you might be interested only in inserts and updates.

Overall I'm quite happy with the end result as the only code I need to maintain now is a very simple transformer. What bothers me though is that it currently doesn't seem possible is to override the source and detail-type, these will be set by the Pipe itself:

"detail-type": "Event from aws:kinesis",
"source": "Pipe qldbpipe-dev-ledger",

I think it would be a very nice feature to for instance set the source to "qldb" and the detail type to something like "select" or "insert". We could change the target of the pipe to be an SQS queue where we attach a Lambda that manually publishes the records to EventBridge with this configuration but that would add more glue code which I tried to avoid with this proof of concept. At this moment I am not yet sure what I will be using in production in my next project but for sure this gave me some good insights in the possibilities of EventBridge Pipes. Let me know what you would prefer in the comments.

To close off I'm excited and proud to announce that I was recently selected as Community Builder by AWS. Follow me to not miss any upcoming articles.

API Gateway Websockets vs IoT Core

Robert Slootjes — Wed, 23 Nov 2022 16:00:37 +0000

When you need websockets in a project on AWS most likely API Gateway Websockets (I will refer to it as API Gateway from now on) is the first service coming to mind. At some point when looking into options, I ran into IoT Core instead. I thought this was meant only for very specific scenarios involving hardware; however it also supports MQTT over websockets which makes it an amazing choice for web and app. I think this is a hidden gem in the AWS ecosystem and in this post I will explain why.

Connection Management

API Gateway

With API Gateway you need to manage your own connections. This means you need to listen to $connect and $disconnect events and store connection IDs in a database (DynamoDB or Redis would be very good choices). As they do not guarantee a $disconnect for every connection you are responsible for your own garbage collection on top of this.

IoT Core

IoT Core manages everything for you! Optionally you can still listen for connect and disconnect events using Lifecycle events.

Messaging

API Gateway

Sending a message is done through the API but allows to send to 1 connection at a time.

If you want to send a message to all connected users, you will need to loop over all connections in your database and send them a message individually. You could offload this work to an SQS queue + Lambda function but it's not ideal.

A noteable feature is data transformation which allows to transform inputs and outputs.

IoT Core

Because IoT Core is an MQTT broker, your clients can subscribe (including wildcards!) and publish to topics. Sending a message to all connected clients from the server is as easy as letting your clients subscribe to the same topic and then doing a single publish API request to the topic to send the message to everyone at once. You can also create a topic for an individual user by their user ID (eg: user/2d543007-2951-43cc-bec4-96f6649b4f53) and send individual messages if needed of course. Keep in mind however that depending on how you set up access control, others could potentially read along.

Integrations

API Gateway

Using custom routes you can create your own integrations based on the body of the received message. Any non-matched messages can be routed to a default route. It is possible to trigger a Lambda, call HTTP endpoints but also there is a possibility to call other AWS Services (especially useful in combination with data transformation).

IoT Core

Handing received messages can be done with Rules which uses an SQL-like syntax. Rule Actions specify what to do when a rule is hit and there are tons of out of the box integrations available. These integrations include Lambda, SQS, SNS & Kinesis to name a few.

Optionally you can send messages to a rule directly to save on messaging costs!

Security

API Gateway

Controlling access for websockets is similar to REST APIs. The option to use a Lambda authorizers and IAM authorization is very flexible.

Other useful security features include request validation and account level and route-level throttling. Using a custom authorizer you can also implement your own throttling mechanism.

IoT Core

Security in IoT Core has much more options than API Gateway since it has many different use cases. Luckily custom authentication allows for a lot of flexibility. This option doesn't only make it possible to allow or deny access to IoT Core but also specify exactly what is allowed to be done (ie: read only on specific topics). I will create a separate post about subject this soon.

There is also another option that doesn't seem to be documented; a pre-signed url! I found a library that does the trick perfectly. To control what the user can do you must create a IAM user for it with the right permissions on iot:Connect, iot:Subscribe, iot:Receive and iot:Publish and use that to sign the request. Your client can then simply connect to the pre-signed url and do it's thing. The pre-signed url is reusable for multiple users which means you can cache it on your CDN. Keep in mind this does give every user the same permissions but in some cases this is exactly what you need.

In all cases, whenever the client tries to do something it isn't allowed (ie: publish to a topic which the IAM user doesn't have permission to) the server will disconnect the client.

IoT Core does not seem to have any request validation or throttling features as of this moment.

Pricing

Pricing obviously is a bit different per region but overall the differences are similar. For this post I've chosen eu-west-1 since that is my region of choice.

API Gateway

1 Million connection minutes: $0.285
Messages measured per: 32 KB
First 1 billion: $1.14 per million
Over 1 billion: $0.94 per million

Source

IoT Core

1 Million connection minutes: $0.08
Messages measured per: 5 KB
Up to 1 billion messages: $1.00 per million
Next 4 billion messages: $0.80 per million
Over 5 billion messages: $0.70 per million
Rules triggered: $0.15 per million
Rule Actions executed: $0.15 per million

Messages sent or received using the reserved topics of Basic Ingest or Alexa Voice Service Integration for AWS IoT Core are free.

Source

Key Differences

IoT Core connection minutes are ~3 times cheaper than API Gateway
Message pricing looks similar but API Gateway meters messages per 32 KB while IoT Core meters messages per 5 KB, a ~6.5 times difference
Receiving and sending messages in IoT Core can be free in some cases
IoT Core charges a fee for Rules and Rule actions, in API Gateway you will be charged for Lambda execution time instead.

Depending on how many messages, the size of your messages and your use case you can calculate easily which one would be cheaper to use. Don't forget though that with API Gateway you do need a database to manage your connections which also costs you extra. Also for sending messages in API Gateway you will most likely need a queue and a Lambda which adds more cost.

Scalability

I've selected some of the most important limits for both services.

API Gateway

New connections per second per account per region: 500
Concurrent connections: no enforced limit (!)

Source

500 new connections per second is a lot and can be increased if needed. Having no limit is very cool but managing these connections in your own database might also add some challenges.

IoT Core

New connections per second per account: 500
Subscriptions per second per account: 500
Inbound publish requests per second per account: 20.000
Outbound publish requests per second per account: 20.000
Maximum concurrent client connections per account: 500.000

Source

This seems to be a bit strict compared to API Gateway, luckily all these limits can be increased. I've had to do this a few times and I was able to set both inbound and outbound publish requests per second to 50.000 but it took some effort with the AWS support team to get it done (and it's understandable why).

I couldn't really find what happens if you exceed these publish limits. Does the message get sent to a limited amount of users, does no one receive it, will it get spread or queued automatically? If anyone has more information on this, please let me know about it as I would love to know more about it.

A possible solution could be to split users in multiple groups round robin and send a message to every group with some delays. On the client side, when sending messages, adding a random delay and doing a exponential back off retry mechanism can be very effective.

For most scenarios the default limits will probably be more than enough and you don't need to worry about these things at all.

Summary

Both are great services with their ups and downs. The main reasons why I prefer IoT Core over API Gateway:

No connection management
Working with topics instead of individual connections
Much cheaper if you can keep the size under 5 KB per message
Less code required

Thanks

Thanks to my colleague Iran Reyes who provided some good insights on the usage of API Gateway Websockets.