DEV Community: Alina Dima

A Tool to Validate AWS IoT Rules SQL Statements

Alina Dima — Wed, 12 Jul 2023 15:22:45 +0000

Rules Engine is a feature in AWS IoT Core that allows engineers to filter, decode, and process IoT device data and route this data to 15+ AWS and third-party services. AWS IoT Core Rules Engine currently has support for over 70 distinct SQL functions which can be used in either SELECT or WHERE clauses, 14 distinct operators which can be used in either SELECT or WHERE clauses, all JSON data types, specifying Literal objects in the SELECT and WHERE clauses, case statements, JSON extensions, substitution templates and nested object queries and more.

In this post, we explore how to validate AWS IoT Core Rules Engine Rules SQL Statements, by introducing a validation tool which:

Encapsulates the heavy lifting of creating, configuring and cleaning up the AWS resources needed to create, run and validate IoT Rule SQL payload transformations.
Enables friction free validation of Rules syntax and payload transformations.
Provides an easily extensible library of sample SQL statements, with input payloads and expected output, allowing you to build your own use-cases.

This tool is available on GitHub, here.

Why do we need a tool?

To validate SQL syntax and the input/output expectations, developers would normally need to:

Create a rule.
Create and assign actions to the rule.
Create and assign an IAM Role with valid permissions for the actions.
Subscribe to the output topic (or monitor the output system), and then publish an MQTT message, to see if the rule works.
If the Rule execution does not work, they have to check in Amazon CloudWatch IoT Logs or the downstream service logs to see what went wrong.
If Amazon CloudWatch Logs for IoT are not enabled, developers need to enable them and then try again.

This is quite some heavy lifting, if the ultimate purpose is to validate that the payload transformations from provided input into expected output work as expected.

So this called for creating a tool that enables developers to do a ‘closed box ’- type of validation, where input payload, SQL statement and expected output are provided. The tool executes validation scenarios, and if the Rules Engine parses the input as expected, with the expected output, scenarios succeed, if not they fail and expected versus actual output payload are printed for references.

How does the tool work?

The diagram below shows the architecture of the validation tool. You can imagine this tool working like an integration test suite, where you only need to define the inputs, configuration and expected outputs (components you need to define are marked yellow). This inputs are provided as a new JSON file and also referred to as creating a new test case or validation scenario. See the “Adding a new test case” section below for details.

The integration test itself (set-up, execution, assertions and expectations and tear down - marked gray on the diagram) is already part of the tool. Of course, if you discover that you need to implement different behaviour to the one the tool provides for set up/tear down/test execution and assertions/validation, you can extend this tool by creating further integration tests to fit your use case.

Additionally, the tool also includes a sample library with validation scenarios, comprised to date, of 6 examples from the more popularly reported Rules SQL statement questions. These scenarios can be explored, executed or modified and adapted.

To better understand the format of the input, have a look at the current JSON schema:

{
  "$schema": "https://json-schema.org/draft/2019-09/schema",
  "type": "object",
  "properties": {
    "sqlVersion": {
      "type": "string",
      "description": "SQL version needed for the test. Defaults to latest."
    },
    "topic": {
      "type": "string",
      "description":"This is the AWS IoT MQTT topic that the input payload will be published on. The same topic must be used in the Rule SQL FROM clause. "
    },
    "inputPayload": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "object"
        }
      ],
      "description":"This is the test input payload which will be published during the test execution. "
    },
    "inputSql": {
      "type": "string",
      "description":"This is the SQL statement of the IoT Rule under evaluation. "
    },
    "expectedOutput": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "object"
        },
        {
          "type": "array"
        }
      ]
    },
    "description":"This is the expected output that the input payload will be transformed into after scenario validation execution. "

  },
  "required": [
    "topic",
    "inputPayload",
    "inputSql",
    "expectedOutput"
  ]
}

Getting started

Pre-Requisites

To be able to run the existing or new validation scenarios, you need to:

Install npm. This project was tested with Node v18.16.0.
Have an AWS Account and provide Node.js with credentials.

Repository Structure

The project repository is structured as per screenshot below:

A util folder containing the configuration, environment and resource set-up and clean-up utilities. The utilities are called from the validation test automatically as needed.
A validation folder containing:
- validation-data: this is a sample library of working examples of SQL statements which validate successfully. You can extend this folder with more validation scenario files as needed.
- input-json-schema: providing type and description information for mandatory and optional input fields.
- validate-iot-rules: which is the framework for set up and execution of the provided and new validation scenarios.

To get started

Clone the GitHub repository: https://github.com/aws-iot-builder-tools/validation-tool-for-aws-iot-rules
Make sure you add your AWS account id in the configuration file: util/config.js.
Run:npm install
By default, the default validation scenario is executed: casting-sensor-payload.json. To run the default scenario, run: npm run test
To change the default scenario, got to config.js, and re-set the defaultInputFile to a file of your choice from the provided files in the validation/validation-data directory. You can choose the validation scenario you want to execute, by running: npm test -- -inputFile=<existing or newly created file name>
You can also choose to execute all the provides scenarios, by running: npm test -- -inputFile=ALL

Note that:

This is a development tool, which is designed to support with faster, friction free testing and validation of SQL statements based on desired inputs and outputs. It is therefore not recommended to run this tool against production environments.
Running all validation scenarios takes longer as, the tool creates and needed rules, IAM roles and policies before executing any of the tests.
As tests are executed, expectations are evaluated to either success or failure (with the printed expected and actual result).
All AWS resources are automatically cleaned up post execution, so there is no manual action needed.

If there is a test failure, you see the comparison between expected and actual output, as shown below:

Adding a new validation scenario

Adding a new validation scenario based on your use-case is straight-forward. You need to add a new JSON file in the verification-tool-for-aws-iot-rules/validation/validation-datafolder.

The following requirements hold when adding your own new validation scenario:

The contents of the file must follow the JSON schema (validation/input-json-schema.json) mentioned in the paragraph above and include all mandatory fields, with the correct data types.
You need to also make sure that the topic name is test uses is unique.
If you add more validation scenarios, bear in mind that you might need to adjust the Jest timeout value. If you want to execute all validation scenarios together in the same test suite after you added more scenarios, you should adjust the timeout value in the test itself, or in the Jest configuration (jest.config.js).

Conclusion and Future Improvements

This blog shows an approach which allows developers to experiment with and validate AWS IoT Rules SQL statements faster and friction-free. This is achieved by encapsulating the complexities of AWS resource creation and clean-up, publishing data on topics and republishing actions in a configurable and extensible tool developers can use (available on GitHub), which does this heavy lifting.

This tool is currently in its first iteration. Below is a list of improvements could be considered for future iterations:

In the current version, the tool is designed to run locally. Ideally it would be integrated in a CI/CD pipeline.
Add support for MQTT 5 and protobuf.
Add ability to mock Rules SQL functions which call other AWS and Amazon services, like Amazon DynamoDB or AWS Lambda.
In the current version, the tool assumes that the rule executes (i.e. that the payload satisfies the WHERE clause). To validate scenarios where input payloads do not satisfy the WHERE clause, a new test case needs to be created, with modified expectations.
Improve overall execution time and resilience.

For more information about AWS IoT Core and Rules Engine, have a look at the AWS IoT Developer Guide.

If you have validation scenarios you would like to share, or additions to this tool, feel free reach out to me on LinkedIn or Twitter, or provide feedback on GitHub.

To get notified about more IoT content, you can additionally subscribe to the IoT Builders YouTube channel.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Batch Ingestion of IoT Device Metrics into Amazon CloudWatch Metrics, using Embedded Metric Format (EMF)

Alina Dima — Thu, 22 Jun 2023 10:37:34 +0000

Introduction

This blog demonstrates how to generate, ingest, store and vizualize IoT device metrics by using the Embedded Metric Format (EMF) and native integrations between AWS IoT Core Rules Engine and Amazon CloudWatch, as well as CloudWatch Logs and Metrics. The Amazon CloudWatch Embedded Metric Format is a JSON specification used to instruct Amazon CloudWatch Logs to automatically extract metric values embedded in structured log events.

In this post, we will see how to:

Ingest metric values embedded in structured log events in batch mode via the AWS IoT Core Rules Engine and the Basic Ingest feature.
Store these log events in Amazon CloudWatch.
View the metrics in Amazon CloudWatch Metrics and create graphs on the extracted metric values.

For a detailed walk-through a live demo of this solution, you can watch the video linked below on the IoT Builders YouTube channel:

Batch Ingestion of IoT Device Metrics into Amazon CloudWatch Metrics

Benefits of this Approach

The benefit of using Basic Ingest is optimizing the data flow by removing the AWS IoT Core MQTT Broker from the ingestion path, thus removing the messaging costs.

On the device-side, sending metrics in batches is more efficient and caters for temporary connectivity loss. The EMF timestamps ensure that metrics are stored in an eventually consistent manner. The integration between AWS IoT Rules Engine and Amazon CloudWatch happens via a Rules Action, therefore reducing the need for bespoke code. Another benefit of this approach is that Amazon CloudWatch automatically extracts the metrics from logs.

Overview

In this post, we will walk through the steps to generate IoT device metrics in the EMF format, like system or OS information at a sampling interval, batch them and ingest them using Basic Ingest, at a chosen reporting interval. This means that metrics are consolidated and sent in batches (independently of the collection time), instead of being routed event by event (one-by-one as they occur).

Once the batched metrics arrive at the IoT Rule, they are routed to Amazon CloudWatch using batchMode. batchMode is a Boolean parameter within the AWS IoT CloudWatch Logs rule action. This parameter is optional and is off (false) by default. To upload device-side log files in batches, you must turn this parameter on (true) when you create the AWS IoT rule.

Because the logs follow the EMF specification, Amazon CloudWatch automatically extracts the metric values embedded in structured log events, and we can create graphs and alarms on the extracted metric values.

The diagram below shows the ingestion flow:

Set-up

To run this demo, clone the repo: https://github.com/aws-iot-builder-tools/emf-metrics-with-iot-rules

This repository is composed of two folders:

app - containing the application code and configuration.
infra - containing the infrastructure CDK code and configuration.

You will need both to set up the demo.

To run the demo, the following steps should be performed:

1. Deploy the required AWS resources:

The AWS resources for this demo are created and deployed using AWS CDK. The CDK Typescript code creates and deploys:

An IoT device policy, allowing the device to connect to AWS IoT Core and publish data on the Basic Ingest Rule topic. This policy looks as below:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": [
        "iot:Connect"
      ],
      "Resource": [
        "arn:aws:iot:<AWS_REGION>:<AWS_ACCOUNT>:client/${iot:ClientId}"
      ],
      "Effect": "Allow"
    },
    {
      "Action": [
        "iot:Publish"
      ],
      "Resource": [
        "arn:aws:iot:<AWS_REGION>:<AWS_ACCOUNT>:topic/$aws/rules/emf/${iot:ClientId}/logs"
      ],
      "Effect": "Allow"
    }
  ]
}

An IoT Rule for Basic Ingest, configured with an action to batch ingest log entries into Amazon CloudWatch.
An IAM Role with the correct IAM policy allowing the IoT Rule to ingest into CloudWatch.

Pre-requisites:

You must have AWS CDK installed and configured with the required credentials for your AWS Account. For help with this, follow the steps in the documentation.

Steps:

In the infra directory , run npm run build.
Run cdk deploy.

2. Run the IoT Device Simulation Application:

The IoT application connects to AWS IoT Core using an MQTT client implemented with MQTT.Js, reads operating system metrics on a sampling interval of 5 seconds, stores them in an in-memory array and ingests them in batches at a reporting interval of 15 seconds. In the app folder, there is also a utility function which, in an idempotent manner, creates the IoT thing, certificate and keys.

Pre-requisites:

To run the IoT application, you need to ensure:

That your application has the correct credentials to make calls to AWS IoT to create the IoT thing, certificate and keys (More info here.
That the correct AWS Region is also configured.

Steps:

Fill in config.js with your IoT endpoint configuration:

export const config = {
    iotEndpoint: "<YOUR_AWS_IOT_ENDPOINT>",
    region: "<YOUR_AWS_REGION>"
}

Run npm install in the app directory.
Run node app.js

Metrics Format and Ingestion

On the device, metric values are embedded in structured log events, so that Amazon CloudWatch can automatically extract them. In the example, memory metrics and network metrics are stored in 2 different namespaces iot-device-memory and iot-device-network. The format looks as follows:

const statObject = {
    "_aws": {
        "Timestamp": Date.now(),
        "CloudWatchMetrics": [
            {
                "Namespace": "iot-device-memory",
                "Dimensions": [["thingName"]],
                "Metrics": [
                    {
                        "Name": "total",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "free",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "used",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "active",
                        "Unit": "Kb",
                        "StorageResolution": 60
                    }, {
                        "Name": "available",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                ]
            },
            {
                "Namespace": "iot-device-network",
                "Dimensions": [["thingName"]],
                "Metrics": [
                    {
                        "Name": "operstate",
                         "Unit": "String",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "rx_bytes",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "rx_dropped",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "rx_errors",
                        "Unit": "Count",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "tx_bytes",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    }, {
                        "Name": "tx_dropped",
                        "Unit": "Kb",
                        "StorageResolution": 1
                    },
                    {
                        "Name": "tx_errors",
                        "Unit": "Count",
                        "StorageResolution": 1
                    }, {
                        "Name": "ms",
                        "Unit": "Milliseconds",
                        "StorageResolution": 1
                    },
                ]
            }
        ]
    },
    "thingName": CLIENT_ID,
    "total": convertSize(memory.total, "KB"),
    "free": convertSize(memory.free,"KB"),
    "used": convertSize(memory.used,"KB"),
    "active": convertSize(memory.active,"KB"),
    "available": convertSize(memory.available,"KB"),
    "iface": iface,
    "operstate": network_0.operstate,
    "rx_bytes": convertSize(network_0.rx_bytes,"KB"),
    "rx_dropped": convertSize(network_0.rx_dropped,"KB"),
    "rx_errors": network_0.rx_errors,
    "tx_bytes": convertSize(network_0.tx_bytes,"KB"),
    "tx_dropped": convertSize(network_0.tx_dropped,"KB"),
    "tx_errors": network_0.tx_errors,
    "ms": network_0.ms,
    "requestId": v4()
}

The "convert-size" JavaScript library is used to convert from bytes to kilobytes. Every 5 seconds (as configured via the sampling interval), a new entry is collected in the statsObject, and added to an in-memory array. Every 15 seconds (as configured by the reporting interval), a batch object is constructed containing the array, and this object is published into the Basic Ingest IoT Rule Topic, as below:

let message = {
                batch: []
            }
            metrics.forEach(metric => {
                message.batch.push(
                    {"timestamp": Date.now(), "message": JSON.stringify(metric)});
            })

client.publish(METRICS_PUB_TOPIC, JSON.stringify(message), {
                qos: 1,
                properties: {
                    contentType: 'application/json',
                    correlationData: JSON.stringify({messageId: messageId, appId: appId})
                }
            });

On the AWS cloud-side, an IoT Rule is created as per CDK code below:

new CfnTopicRule(this, 'BasicIngestEMFIoTRule', {
            ruleName: 'emf',
            topicRulePayload: {
                actions: [
                    {
                        cloudwatchLogs: {
                            logGroupName: LOG_GROUP_NAME,
                            roleArn: CWRole.roleArn,
                            batchMode: true,
                        }
                    }
                ],
                description: 'IoT Rule',
                sql: `SELECT VALUE *.batch
                      FROM '${METRICS_RULE_TOPIC}'`,
                ruleDisabled: false,
                awsIotSqlVersion: '2016-03-23'
            }
        });

The video below shows the application log for storing and publishing the metrics in batches:

Amazon CloudWatch Logs and CloudWatch Metrics

Upon batch ingestion, each item in the batch will be stored as a separate entry in CloudWatch logs. Because the format used is EMF, the metrics are extracted by Amazon CloudWatch and available in CloudWatch Metrics to create charts and alarms. Below is a video of how the log entries look with the LiveTail view:

Each element in the batch sent from the IoT device is stored as a separate log entry, with the correct reporting timestamped passed from the device. The sampling timestamps for the metrics become relevant when the metrics are extracted. Below is a log entry view from the LiveTail:

Navigating to Amazon CloudWatch Metrics, you can see the two newly created namespaces with thing name as the dimension. By clicking one of the namespaces, you can plot the metrics values over time, with the desired aggregations, as shown in the video below:

Conclusion

In this post, we looked at how to leverage Basic Ingest and the AWS IoT CloudWatch Logs Action in Batch Mode, in order to ingest and route device metrics in EMF format. The benefit of this approach is that Amazon CloudWatch automatically extracts the metrics from logs. Additionally, ingestion costs are reduced by using Basic Ingest. In case of temporary loss of connectivity, the batching functionality will ensure eventual data consistency.

Once the metrics are in the cloud, they can be viewed in CloudWatch metrics, to prepare charts or set alarms. For more information on uploading IoT device logs to Amazon CloudWatch, have a look at the developer documentation. To learn more about EMF, check the specification.

To get notified about more IoT content, you can additionally subscribe to the IoT Builders YouTube channel.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Enriching Payloads with MQTT 5 Metadata, using AWS IoT Core Rules Engine

Alina Dima — Thu, 08 Jun 2023 13:18:13 +0000

Introduction

This blog post explains how to build a Vehicle Command Log Store to keep track of command requests and responses sent to vehicles by application clients. To this purpose, we look at extracting MQTT 5 metadata information from MQTT messages published using MQTT v5, and enrich payloads via an AWS IoT Core Rules Engine Rules. Processed data is stored in an Amazon DynamoDB table.

The goal is to use as little bespoke code on the cloud side as possible and lean on native integrations on the AWS Cloud side, like the IoT Rule with Dynamo DB Action.

For a detailed walk-through a live demo of this solution, you can watch the video linked below on the IoT Builders YouTube channel:

Enriching Payloads with MQTT 5 Metadata

MQTT 5 Request/Response Pattern

As part of this blog post, we exploring a feature of MQTT 5: the Request/Response pattern.

The Request/Response messaging pattern is a method to track responses to client requests in an asynchronous way. It’s a mechanism implemented in MQTTv5 to allow the publisher to specify a topic for the response to be sent on a particular request. When the subscriber receives the request, it also receives the topic to send the response on. This pattern supports a correlation data field that allows tracking of packets, e.g. request or device identification parameters.

Let's look at an example:

We want to send commands to vehicles over MQTT from client applications. The flow is as follows:

Vehicles subscribe to the request topic, and client applications subscribe to their decided response topics, which can be dependent on the application instance id, for example.
App clients publish requests on the request topic. In our case, the payload is a simple text DOOR_LOCK indicating the command to lock the vehicle doors. Following the MQTT 5 pattern, in addition to the payload, we are sending metadata like the Response Topic, Content Type and Correlation Data:
- a request id,
- a timestamp
- a user id.
This data helps correlate MQTT requests and their responses.
As the device receives the MQTT message, it executes the command and publishes the response on the specified response topic. Upon publishing, it re-send the correlation data, but also appends response specific information, such as:
- a response timestamp,
- the command it is responding to, marked red on the diagram.
The app instance receives this information on the response topic it subscribed to previously.

Prerequisites and Approach

Prerequisites

To recreate this demo locally, one needs to already have in place:

An AWS Account with permissions to create IoT resources.
Two created AWS IoT things for the app and car simulators, with locally stored certificates and keys to connect to AWS IoT Core.
An IoT Core policy allowing connections, subscriptions and data publishing for both IoT Things on the configured topics.
An Amazon DynamoDB table needs to be created beforehand, with a unique identifier ‘msgId’ string as primary key.

This section looks at the steps to be performed:

Step 1: Setting up the the MQTTJS clients

First, we build two client simulators, for the car and the application client. We use MQTTJS for Javascript and its MQTT v5 implementation support. Step 1 below will describe the details of the MQTT client implementation.

Upon running the application simulator, it connects, subscribes to the response topic and stays connected. Every 10 seconds, the application simulator publishes a DOOR_LOCK text message command. Upon running the car simulator, it connects and stays connected, then receives a command from the app client, simulates its execution with a JavaScript timer and sends back a DOOR_LOCK_SUCCESS response.

The MQTT Request/Response pattern implementation is exemplified in the code snippets below:

Car Simulator

Connection options and connection creation:

//create options for MQTT v5 client
        const options = {
            clientId: CLIENT_ID,
            host: ENDPOINT,
            port: PORT,
            protocol: 'mqtts',
            protocolVersion: 5,
            cert: fs.readFileSync(CERT_FILE),
            key: fs.readFileSync(KEY_FILE),
            reconnectPeriod: 0,
            enableTrace: false
        }
        //connect
        const client = mqtt.connect(options);

MQTT Event Handlers implementation:

client.on('connect', (packet) => {
            console.log('connected');
            console.log('subscribing to', SUB_TOPIC);
            client.subscribe(SUB_TOPIC);
        });

//handle Messages
        client.on('message', (topic, message, properties) => {
            console.log('Received Message',message.toString());
            console.log('Received Message Properties', properties );

            if(message && message.toString() === 'DOOR_LOCK' ) {
                console.log('Executing', JSON.stringify(message));
                setTimeout(() => {
                    const response = 'DOOR_LOCK_SUCCESS';
                    const responseTopic = properties.properties.responseTopic;
                    console.log("ResponseTopic: ", responseTopic);
                    console.log('Publishing Response: ', response," to Topic: ", responseTopic.toString());
                    const cDataString = properties.properties.correlationData.toString();
                    console.log('Correlation Data String: ', cDataString)
                    let correlationDataJSON = JSON.parse(cDataString);
                    correlationDataJSON.resp_ts = new Date().toISOString();
                    correlationDataJSON.cmd = message.toString();

                    client.publish(responseTopic.toString(), response, {
                        qos: 1,
                        properties: {
                            contentType: 'text/plain',
                            correlationData: JSON.stringify(correlationDataJSON)
                        }
                    }, (error, packet) => {
                          //ERROR Handlers here.
                    });
                }, 5000);
            }
        });

App Client Simulator

Connection settings are the same as above.

MQTT Event Handlers implementation:

//handle the message
        client.on('message', (topic, message,  properties) => {
        //Only log for now.
            console.log('Received message: ', message.toString());
            console.log('Received message properties: ', properties );   
        });

Publish the command on Interval for testing purposes:

//publish
        setInterval(() => {
            const requestId = v4();
            // const message = JSON.stringify({ping: 'pong'});
            console.log('Publishing message ');
            client.publish(PUB_TOPIC, 'DOOR_LOCK', {
                    qos: 1, properties: {
                        responseTopic: SUB_TOPIC,
                        contentType: 'text/plain',
                        correlationData: JSON.stringify({
                            requestId: requestId,
                            userId: userId,
                            req_ts: new Date().toISOString()
                        })
                    }
                }
                , (error, packet) => {
                    console.log(JSON.stringify(packet))
                })
        }, 10000);

Step 2: AWS IoT Rule with Amazon DynamoDB Action

An AWS IoT Rule can be used to enrich the message payload with MQTT 5 response topic information, as well as correlation data and content type. We are interested in both command requests and responses, each of them published to different MQTT topics. Therefore, the IoT Rule we create must select data from both topics, as show in the diagram below.

Because we are building a Vehicle Command Log Store and in this scenario such payloads are sent with content-type text/plain, we are filtering only for such messages in the Rule SQL statement.

As the desired behavior is storing the command messages and metadata in a Dynamo DB table, the rule will prepare a new JSON object to be stored. Each key will be stored as a separate column in Amazon Dynamo DB. A new unique message identifier (msgId) is created to be the primary key of the table. For extracting the MQTT 5 metadata, the Rule SQL uses the get_mqtt_property(name) SQL function. Encoding/Decoding functions are used to manipulate data from and to base64 encoded strings.

The Rule Action to be specified is Amazon DynamoDB, pointing to the previously created table.

The IoT Rule SQL is shown below:

SELECT 
{"msgId": newuuid(), 
"name": decode(encode(*, 'base64'), 'base64'), 
"requestId": decode(get_mqtt_property('correlation_data'), 'base64').requestId, "req_ts": decode(get_mqtt_property('correlation_data'), 'base64').req_ts, 
"cmd": decode(get_mqtt_property('correlation_data'), 'base64').cmd, 
"resp_ts": decode(get_mqtt_property('correlation_data'), 'base64').resp_ts, 
"userId": decode(get_mqtt_property('correlation_data'), 'base64').userId, "responseTopic": get_mqtt_property('response_topic') } 
FROM 'cmd/#' 
where get_mqtt_property('content_type') = 'text/plain'

After running the simulation for some minutes, you should see your Amazon Dynamo DB table populated with data entries as showed in the image below:

Conclusion

This blog post shows how to use the MQTT 5 IoT Rules SQL functions to extract MQTT 5 metadata from messages and use them to enrich device payloads. One of the design goals was to achieve enrichment and storage with native cloud integrations and no bespoke message processing code.

The demonstrated Vehicle Command Log Store use-case is just an example, to explore the art of the possible. For more information about the AWS IoT Core Rules Engine SQL functions for MQTT 5 support, have a look at the documentation.

If you are interested in detailed walk-through a live demo of this solution, you can watch the YouTube video. To get notified about more IoT content, you can subscribe to the IoT Builders YouTube channel.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Coding IoT with Amazon CodeWhisperer AI Coding Companion

Alina Dima — Fri, 12 May 2023 13:51:19 +0000

Amazon CodeWhisperer AI Coding Companion has been generally available since April 13th, 2023. Naturally, curiosity got the best of me, so I decided to see if I could rapidly build an IoT application, and have the AI do most of the coding.

The goal of this experiment was to set up the foundations of a simple IoT application, in about 30 minutes. In terms of tools, I used AWS CDK for infrastructure (TypeScript). I had not used CDK before, even if I had years of experience with AWS CloudFormation, SAM, Amplify. For the IoT app itself, I used JavaScript, the MQTT.js library, and the AWS SDK v3 for the AWS IoT Control Plane calls. The idea behind mixing JavaScript and TypeScript was to evaluate how well CodeWhisper could assist me in both languages.

Curious how good a team CodeWhisperer and me were in this experiment? Keep reading, because I will cover some of the great and no so great experiences of working with CodeWhisperer.

For those interested in the build details, I have recorded the entire experience. It is available in a 2-episode series on the IoT Builders YouTube channel.

Part 1 - CDK for IoT

Part 1 - Building the IoT App

The code is also available on GitHub: http://bit.ly/cdk-iot-sample

What did I build?

The IoT application is shown in the diagram below. We have an AWS IoT Thing, connected to AWS IoT Core, subscribed to an MQTT topic, and publishing data on another MQTT topic. An AWS IoT Rule picks up the data and invokes a Lambda action which prints out the MQTT message.

This application was built in 2 steps:

The AWS Resources were created using CDK: an IoT Thing policy based on least privilege best practices, an IoT Rule, a Lambda action and of course the Resource policy allowing Lambda invocation from the IoT Rule.
The IoT application code was written to: create an IoT Thing, create the identity, attach the IoT policy to the certificate, and the certificate to the Thing, create and configure the MQTT client, connect the client to IoT Core, subscribe and publish data.

Amazon CodeWhisperer helped in both steps listed above. It was used to speed up development, by making whole-line and full-function code completions, automatically generating functions and code from natural language comments, and accelerating work with third party libraries like MQTT.js.

Getting started with Amazon CodeWhisperer

As a user of intellij IDEA, I have integrated CodeWhisperer into my IDE in 3 easy steps:

Step 1: Installed the latest_ AWS Toolkit plugin_ in IDEA.
Step 2: In the IDE, open the AWS extension panel and click the “Start” button under Developer Tools → CodeWhisperer
Step 3: In the resulting pop-up, select the “Sign in with Builder ID” option. Use your personal email address to sign up and sign in with AWS Builder ID.

What was the Developer Experience like with CodeWhisperer for IoT?

First off, as we saw above, CodeWhisperer is super easy to set up and start.

Generating Code from Comments

CodeWhisperer was efficient at generating code from comments. Some interesting examples were:

When prompted with the comment: // Create AWS Lambda function, with inline code.

the result was:

Here is another example of CodeWhisperer coming up with the code to create an IoT thing when prompted by comment:

Auto-Completion

CodeWhisperer is a clever tool for auto-completion. Below is an example of how I fixed the initially incorrect IoT policy using the auto-completion assistance of CodeWhisperer:

A Problematic IoT Policy

While trying to get assistance from CodeWhisperer, I struggled with the creation of an IoT policy, using the CDK SDK for Typescript. It took a few attempts at changing my comment prompts to get to an overly permissive and incorrect IoT policy (see below).

The comment I provided was:

// Create IoT policy for IoT Core. Restrict to account id and region. Allow connect with thing name, subscribe to topic and publish on topic.

Coming up with a correct policy, available here, was eventually more effort than expected, although, as shown in the video above, CodeWhisperer did a very good job with auto-completion.

Coaxing CodeWhisperer into Creating Correct Code

Sometimes, CodeWhisperer generates code that looks just about right, but actually it is not. In fact, sometimes I got tricked into skipping the correct suggestion and went for the incorrect one.

So, as a developer, you need to take care to pick the correct suggestions. Below is an example:

How to make the most of CodeWhisperer?

If you want to make the most of working with CodeWhisperer, the following helps:

In general, the more code you already have, the better CodeWhisperer will do.
Do your main imports first - if you want to use the Control Plane calls for IoT Core, like in my example, import the ClientIoT as a first thing. Afterwards, you CodeWhisperer will anticipate better what you are prompting it to do in your comments.
Be very explicit in your comments - if you want a JavaScript Lambda function, write that in your comments. Be explicit with the type of resources, the nature of code (if you want a function, say so). The more detailed and explicit your comments are, the better success you will have.
Pay attention - sometimes CodeWhisperer provides multiple suggestions, and if you do not pay a lot of attention, you get tricked into choosing the incorrect one (as in the example above). Scroll fast between the proposals, and choose the correct one.
Know when to give up and move on - if full code generation from comments does not work out-of-the-box (like in my example with the IoT policy), count your loses on this one, and use the AWS documentation to start it off correctly, and then use CodeWhisperer for auto-completion, once you are on the right track.

Conclusion and Useful Links

To sum up, the experiment of building an IoT application with the assistance of CodeWhisperer was a success. I aimed for 30 minutes, but in reality it probably took about 40, excluding explanations, introduction and conclusions parts of the video, or written this blog.

If you want to watch the entire experience, it is available in 2 episodes on YouTube:

The code for the IoT application is available on GitHub: http://bit.ly/cdk-iot-sample.

Amazon CodeWhisperer is an AI-powered coding assistant that provides real-time recommendations in your IDE based on your existing code and comments. The tool is available for and can integrate well with various IDEs, including JetBrains. This post talks about my developer experience building an IoT application in about 30 minutes, with the assistance of CodeWhisperer, with some examples of what went well and what did not got well.

For more information about CodeWhisperer, have a look at the documentation. There is also a guide on how to get started with CodeWhisperer and JetBrains IDEs.

To watch more such content, subscribe to IoT Builders on YouTube or follow IoT Builders on dev.to.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Automatically Applying Configuration to IoT Devices with AWS IoT and AWS Step Functions - Part 1

Alina Dima — Thu, 06 Apr 2023 15:59:23 +0000

Introduction
Assumptions
Tools and Services
What workflow are we building?
How it works
Why this Developer Technique
Conclusion
Author

Introduction

Explicitly defining processes involving IoT devices and externalizing their management and orchestration is a useful mechanism to break down complexity, centralize state management, ensure bounded context and encapsulation in your system design. It is also a useful technique for modelling workflows that must survive external factors which impact IoT device operation in the field, such as intermittent connectivity, reboots or battery loss, causing temporary offline behavior.

In this blog post series, we will look at a simple example of modeling an IoT device process as a workflow, using primarily AWS IoT and AWS Step Functions. Our example is a system where, when a device comes online, you need to get external settings based on the profile of the user the device belongs to and push that configuration to the device. The system that holds the external settings is often a third party system you need to integrate with at API level (HTTPS), based on the API specification and system SLAs.

A real world example is home automation. A user sets up a default desired temperature for his apartment, based on a User Settings Profile they created when they purchased the gateway responsible for managing the different sensors, including temperature.

Assumptions

To narrow down this implementation example, here is a list of assumptions:

We refer to the gateway that controls the different sensors as the device.
The workflow starts every time a device comes online.
Nothing happens when devices disconnect. The profile is checked only upon reconnects.
AWS IoT Core interacts only with the gateway, and not with individual sensors. The gateway reports back to confirm the settings have been correctly configured by the respective sensors.
We will simulate the device behavior using an MQTT client application built with the MQTT.js open-source library.

Note that:

This is a sample implementation to get you started. It is not meant to be lifted and shifted for a production environment. It can definitely be used as a starting point.
The focus point in this solution is on development techniques and not on the type of device or type of configuration that needs to be applied. Therefore, this solution simulates the device device behavior, the config, and the external API.

Tools and Services

To build the above described scenario, we use the following technologies:

AWS Step Functions for workflow management and execution. The state machine is available in GitHub. The following AWS Step Function features are used in the sample project:
- Tasks - to break down each unit of work.
- Choice state
- AWS SDK service integrations.
- Wait for a Callback with the Task Token.
AWS IoT Core and AWS IoT Rules Engine for the communication with the IoT Device over MQTT and routing of messages from the device and automatically triggering actions based on message content.
Amazon DynamoDB for storing of Task Tokens needed for the callback service integration pattern of AWS Step Functions.
AWS Lambda for publishing messages to the device. AWS Lambda is also used for processing responses to temperature set requests from the IoT Device, as well as simulating the Third Party system call to retrieve the user default desired temperature.
The device simulator is build in JavaScript, using the open-source Node.js MQTT.js client library.
All the AWS resources are created and deployed using AWS SAM. The AWS SAM template is available in GitHub

You can have a look at the GitHub repository for the entire solution, here: https://github.com/aws-iot-builder-tools/iot-config-management-sample.

What workflow are we building?

The flow is as follows:

Every time the device comes online, the workflow for retrieving the user profile, to understand their desired room temperature kicks off. The desired temperature is stored in the user profile, available via an API call to an external system. We simulate this system call using an AWS Lambda function.
The workflow retrieves the external ID connecting the device to the user it belongs to from IoT thing attributes. It then makes the call to the third party system to retrieve the user profile with the desired temperature value, publishes a message with the configuration setting to the MQTT topic the device is subscribed to, and enters a callback Task, which pauses waiting for the event from the device to come back on the configured response MQTT topic.
Upon receiving and parsing the response, success or failure is determined, the paused Task is terminated on callback with the respective result, and the AWS Step Functions workflow terminates.

Building an automatically triggered workflow contributes to a nice user experience. There is no need for manual intervention to synchronize the IoT device with user specific configuration settings. This is a simple example to introduce the concept. The workflow can get more complex, with more settings and more involved systems.

How it works

Fig. 1 below shows the interactions between the different components.

Fig. 1 - How it works diagram

The interactions can be described as follows:

The device comes online and an MQTT presence event is published on the AWS topic $aws/events/presence/connected/+. (1 in Fig. 1 )
An AWS IoT Rule is configured on the above topic, with the Rule Action being an AWS Step Function State Machine execution trigger. The Rule SQL looks as follows: SELECT *, topic(5) AS thingName FROM $aws/events/presence/connected/+ (2 in Fig. 1 )
On MQTT connect presence event, the Rule will pick up the event and pass it as input to the state machine, which will be started on Rule execution (3 in Fig. 1 ). The workflow steps will execute as follows:
- Firstly, the IoT DescribeThing SDK call will be made to retrieve the thing from the Device Registry. The external ID thing attribute will be retrieved and passed on to the next step. Note that the AWS Step Functions SDK integration is used at this step.
- 3Party_GetUserProfileForDevice is executed as a next step. This is an AWS Lambda function which simulates a call to a third party system. 3Party_GetUserProfileForDevice returns a randomly generated desired temperature value for an external ID and thing name. (4 in the Fig. 1 ).
- The SetConfigurationOnDevice AWS Lambda function uses the IoTDataPlaneClient to publish a message to the IoT Device on the MQTT request topic with the desired temperature value. (5 in Fig. 1 ), and go next into a “Wait for Callback” Task, waiting for confirmation of the configuration setting from the device.
- The “Wait for Callback” Task integrates an AWS DynamoDB PutItem SDK call. A task token is generated by AWS Step Functions and sent into the Task as input. We store the token so that we can identify and correlate which token corresponds to the request the device sends the response for. The token is stored in an Amazon DynamoDB table, together with the name of the operation (in this case SET_TEMP), a unique operation ID (passed as input from the Rules Engine MQTT event session ID ), the device ID (thing name) and a status value of ACTIVE. The DynamoDB table is configuration is shown below:

  CMTaskTokens:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName:  CMTaskTokens
      AttributeDefinitions:
        - AttributeName: operationId
          AttributeType: S
        - AttributeName: token
          AttributeType: S

      KeySchema:
        - AttributeName: operationId
          KeyType: HASH
        - AttributeName: token
          KeyType: RANGE
      BillingMode: PAY_PER_REQUEST

The device receives the request (6 in Fig. 1 ), attempts the operation, and sends the event confirming the result on the MQTT response topic (7 in Fig. 1 ). An IoT Rule picks up the incoming event, and invokes an AWS Lambda Function (8 in Fig. 1 ) which :
- Parses the response and establishes success or failure.
- Retrieves the token entry based on the operation ID from the database table and uses it to terminate the AWS Step Function WaitForConfirmationEventFromDevice state with either success or failure.(9 in Fig. 1 ).
The result from the evaluation of the MQTT response event from the IoT device is passed further into a Choice Task, which determines state machine termination with with either success or failure.

How to Deploy and Run this in your AWS Account

Pre-requisites:

Clone the GitHub repo: iot-config-management-sample
In the root directory, iot-config-management-sample, run the following commands:

      sam build 
      sam deploy --guided

Start the device simulator:
- Add and validate your configuration in iot-config-management-sample/device/config.js:

export const config =
{
    iotEndpoint: '<YOUR IoT Endpoint>',
    clientId: '<YOUR Thing Name>',
    policyName: '<YOUR IoT Policy>',
    verbosity: 'warn',
    region: '<YOUR AWS Region>',
    shouldCreateThingAndIdentity: <true or false> // if true, the simulator will create the AWS IoT Thing and unique identity. Certificate and Key will be stored in certs/
}

If your shouldCreateThingAndIdentity flag is set to false, you need to make sure the IoT thing, certificate and key have already been created, and store the certificate and key in the certs folder prior to running the MQTT client.

Start the simulator: node simulator.js
If the simulator is successfully started, you should already see that an AWS Step Functions workflow was triggered automatically, once your device successfully connected to AWS IoT. Note that the simulator implements a delay of 10 seconds between receiving the request and sending a successful response.
Verify that the workflow was successfully triggered:
- Log into the AWS Console, navigate to AWS Step Functions, find the ConfigManagement State Machine, and explore the execution, as shown in the 2 images below.

Fig. 2 - Waiting for MQTT response from the device

Fig 3- Successful workflow execution

Why this Developer Technique?

The advantages of modeling the workflow using AWS Step Functions are as follows:

The workflow is explicitly defined and exists outside of various system interactions or the IoT device.
Flexibility in configuring retries, error handling and timeouts, at each step:
- DescribeThing can run into TPS limits caused by too many concurrent calls to AWS IoT. You can configure your state machine to retry this step, with an exponential back-off strategy.
- Calls to the Third Party systems can fail with various errors, some of which are worth retrying on. Exception Handling and Retries can be configured at Task level.
- If the device goes offline temporarily during the workflow execution for whatever reason, you can configure a reasonable Heartbeat on the Callback Task. The workflow then pauses until either a message from the device arrives or the timeout is reached. In this way, workflows can exist in the cloud outside of whatever is going on with the device, for as long as needed.
AWS Step Functions centralizes state management for each step in the workflow, as well as the workflow as a whole. Tracing, logging and metrics are available as execution level, and at task level. These observability features, as well as the integration with AWS X-Ray and Amazon CloudWatch will be covered in Part 2 of this blog series.

Conclusion

In the first part of this blog series, we looked at modelling and building the process of automatically applying configuration to an IoT device, based on default settings in a user profile, using AWS Steps Functions, AWS IoT, Amazon Dynamo DB, AWS Lambda and the open-source MQTT.js client library. This is a developer technique used to reduce complexity in workflows involving IoT devices. To understand more about the implementation, have a look at the code in GitHub.

For a more complex example of an IoT Workflow, you can have a look at the firmware upgrade (OTA) flow built using AWS IoT and AWS Step Functions in the iot-workflow-management-and-execution GitHub repo.

If you are curious to learn more about IoT workflow orchestration, watch our IoT Builders YouTube episodes:

Stay tuned for the next blog of the series, where we explore in detail observability aspects such as logging, tracing and metrics for this workflow.

If you have any feedback about this post, or you would like to see more related content, please reach out to me here, or on Twitter or LinkedIn.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Decoding USP protobuf data sent by USP agents, using the recent AWS IoT Rules Engine native SQL decode function

Alina Dima — Tue, 03 Jan 2023 10:41:13 +0000

Introduction
Context
Set-Up Steps
Exploring the payloads and testing with the AWS IoT Console MQTT Client
Conclusion
Author

Introduction

This post is a follow-up to:

Connect the Open Broadband USP Agent to AWS IoT Core using MQTT5, and Validate the Integration with AWS IoT Core Device Advisor

Alina Dima for IoT Builders ・ Dec 9 '22

#awsiot #tr369 #telco #mqtt5

, which showed how to connect a device running obuspa to AWS IoT Core, over MQTT 5, ingest data and validate the MQTT 5 connection using AWS IoT Core Device Advisor. It is recommended you read the preceding post, in order to get a better understanding of integrating a USP agent with AWS IoT.

In this post, you will see how to set up and use the native AWS IoT Rules Engine protobuf decoding SQL function, in order to decode USP records and messages sent by USP agents, on the fly, without the need for additional AWS Lambda functions with bespoke decoding logic.

Context

Protocol Buffers (protobuf) is an open-source data format used to serialize structured data in binary form. It is used for transmitting data over networks or storing it in files. Protobuf allows you to send data in small packet sizes and at a faster rate than other messaging formats. USP specifies Protocol Buffers Version 3 as a mechanism to serialize data to be sent over a Message Transfer Protocol, such as MQTT.

The protobuf decoding support in the AWS IoT Rules Engine was recently introduced (https://aws.amazon.com/about-aws/whats-new/2022/12/aws-iot-core-rules-engine-google-protocol-buffer-messaging-format/) to enable developers to decode protobuf payloads to JSON format directly in the rule, and route them to downstream services, by means of an AWS IoT Rules SQL function: decode(value, decodingScheme) (https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-functions.html#iot-sql-decode-base64).

This feature is interesting because it removes the need to write and maintain custom protobuf decoding logic, which was, prior to this feature, usually done using AWS Lambda functions, invoked either as Rule Actions, or as part of IoT Rule SQL expressions (see the diagram below - before and after the native protobuf support in AWS IoT Rules Engine).

!(https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ssc6t3oihgvwqpp24ggp.png)

Set-Up Steps

Prepare the required protobuf files

As part of the AWS IoT Rule decoding pre-requisites, the protobuf descriptor file needs to be created and uploaded to an S3 bucket.

Install the Protobuf Compiler (protoc) on your system. On MacOS, you could use Homebrew, as described here.
Download the 2 proto files record and message from the open source USP GitHub repository, corresponding to the USP version that you want to use in your project. This post will work with the latest version, USP 1.2.
Create the descriptor file required by the AWS IoT Rules Engine, as described in the docs

You can use the following command:

protoc —descriptor_set_out=usp-1-2.desc —proto_path=<PATH TO YOUR PROTO FILES> —include_imports usp-record-1-2.proto usp-msg-1-2.proto

Create the AWS Cloud resources

You will need to create the AWS IoT Rule, correct Roles and Policies, and S3 Bucket where the descriptor file will be uploded. You can use the AWS CloudFormation template below for this set-up. To create the AWS resources using this template, you can use the AWS CloudFormation CLI commands or the AWS SAM CLI commands.

For the purpose of this example, the Rule will decode the protobuf record, then decode the protobuf message (both using the decode () SQL function), and then republish the resulting JSON object, on a new topic, so we can examine correctness.

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

Resources:
  USPProtobufDecodingRule:
    Type: AWS::IoT::TopicRule
    Properties:
      RuleName: USPProtobufDecodingRule
      TopicRulePayload:
        AwsIotSqlVersion: 2016-03-23
        RuleDisabled: False
        Sql: SELECT VALUE decode(decode(encode(*, 'base64'), 'proto', 'usp-1-2', 'usp-1-2.desc', 'usp-record-1-2.proto', 'Record').noSessionContext.payload, 'proto', 'usp-1-2', 'usp-1-2.desc', 'usp-msg-1-2.proto', 'Msg') FROM '/usp/controller'
        Actions:
          - Republish:
              RoleArn: !GetAtt RepublishDecodedRole.Arn
              Topic: '/usp/controller/decoded'

  RepublishDecodedRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action:
              - sts:AssumeRole
            Principal:
              Service:
              - iot.amazonaws.com
      Policies:
        - PolicyName: allowRepublish
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action: iot:Publish
                Resource: !Join [ "", [ "arn:aws:iot:", !Ref "AWS::Region", ":", !Ref "AWS::AccountId", ":topic//usp/controller/decoded" ] ]

  USPBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: usp-1-2-bucket

  USPBucketPolicy:
    Type: AWS::S3::BucketPolicy
    Properties:
      Bucket: !Ref USPBucket
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Action:
            - 's3:Get*'
            Effect: Allow
            Resource: !Join
              - ''
              - - 'arn:aws:s3:::'
                - !Ref USPBucket
                - /*
            Principal:
              Service:
                - iot.amazonaws.com

The AWS IoT Rule SQL expression for decoding could look like below:

SELECT VALUE decode(decode(encode(*, 'base64'), 'proto', 'usp-1-2', 'usp-1-2.desc', 'usp-record-1-2.proto', 'Record').noSessionContext.payload, 'proto', 'usp-1-2', 'usp-1-2.desc', 'usp-msg-1-2.proto', 'Msg') FROM '/usp/controller'

The SQL rule behaves the following way:

First, the binary MQTT payload is transformed in a based64-encoded string,
Then, the base64 encoded string is protobuf-decoded at USP record level, as specified by the descriptor,
Finally, the USP message inside the record, located in the payload object, is protobuf decoded.

Because USP messages are encoded and wrapped inside encoded USP records, you will need to call the decode function twice.

Upload the descriptor file and start obuspa

Upload the descriptor file created in step 3 into your S3 bucket. You can do this using the AWS CLI or AWS Console.
You can now set up and start the obuspa, as described in the following sections: Integrate obuspa with AWS IoT Core over MQTT5 and Verify MQTT 5 Integration/Protobuf Data Publishing.

Once the agent is sending data, the set-up is complete.

Exploring the payloads and testing with the AWS IoT Console MQTT Client

As the protobuf records arrive on the topic, the IoT Rule will pick them up, decode them, and republish the JSON objects on the /usp/controller/decoded topic. If you subscribe with the test MQTT Client in the AWS IoT Console, you will be able to see the protobuf records on the /usp/controller topic, and the JSON payloads on the /usp/controller/decoded topic, as shown in the images below. The first image shows the original protobuf encoded USP record, arriving on the topic, and the second image the decoded USP message, in JSON format.

Conclusion

This post shows how to to decode protobuf records and messages sent by a USP agent, directly in the AWS IoT Rules Engine, using the decode function, as part of the rule expression. This is an AWS IoT Rules Engine feature released recently, making it easier to use protobuf directly in your IoT Rule, without the need to write and maintain custom decoding code in AWS Lambda functions.

Author

Alina DimaFollow

An engineer for about 20 years, love solving real world problems with code. I simplify complex problems to help developer communities build better and faster with AWS IoT.

Connect the Open Broadband USP Agent to AWS IoT Core using MQTT5, and Validate the Integration with AWS IoT Core Device Advisor

Alina Dima — Fri, 09 Dec 2022 10:13:05 +0000

Introduction
Scope
Integrate obuspa with AWS IoT Core over MQTT5
Verify MQTT 5 Integration/Protobuf Data Publishing
Validate MQTT 5 Integration with AWS IoT Core Device Advisor “Qualification Program Test Suite”
Conclusion

Introduction

The User Services Platform (USP), also called TR-369, is a standardized protocol for managing, monitoring, upgrading, and controlling connected devices. It is widely used for customer premise equipment (CPE), such as modem routers. This protocol is the next generation follow-up for Broadband Forum’s CPE WAN Management Protocol (CWMP), commonly known as TR-069.

The Broadband Forum specification defines USP as "consisting of a collection of Endpoints (Agents and Controllers) that allow applications to manipulate Service Elements (Objects and Parameters that model a given service, such as network interfaces, software modules, device firmware, remote elements proxied through another interface, virtual elements, or other managed services)".

Generally, Agents run on CPE devices and can communicate with Controllers over a variety of message transport protocols (MTPs), one of which is MQTT. MQTT allows a secure, 24/7 open communication channel between Agents and Controllers. USP has specified the usage of both MQTT 5 and MQTT 3.1.1 from early specification versions.

With the recent general availability of MQTT 5 support in AWS IoT Core, I was eager to test the MQTT 5 integration between the AWS IoT Core MQTT Broker and USP Agents. This post is summarizing my findings.

For testing purposes, I will be using the Open Source, GitHub available, Open Broadband User Services Platform Agent (also referred to as OB-USP-Agent or obuspa), a reference C-based implementation of the USP protocol for Agents.

Scope

In this post, we will cover the following:

Configuring the OB-USP-Agent to connect to AWS IoT using MQTT 5.
Verifying successful connection to AWS IoT Core over MQTT 5, as well as subscriptions and data publishing.
Creating and running an AWS IoT Core Device Advisor “Qualification Program Test Suite”, to validate the MQTT 5 client/broker communication.

For those not familiar with AWS IoT Device Advisor, it is a Cloud-based, fully managed test capability for validating IoT devices during device software development. Device Advisor provides pre-built tests that you can use to validate IoT devices for reliable and secure connectivity with AWS IoT Core, before deploying devices to production. Device Advisor’s pre-built tests help you validate your device software against best practices for usage of TLS, MQTT, Device Shadow, and IoT Jobs. With the recent support for MQTT 5 in AWS IoT Core, MQTT 5 test suites have also been added to Device Advisor.

This blog post is a demonstration that obuspa can be configured to integrated with AWS IoT Core over MQTT 5. It is not a solution to be lifted and shifted for production use. The content also assumes some familiarity with USP, especially overall architecture, getting/setting parameters, and configuring USP subscriptions. If you are not already familiar with USP, these concepts are very well explained in the USP technical specification, provided by the Broadband Forum.

Integrate obuspa with AWS IoT Core over MQTT5

Our first goal is to get obuspa to connect successfully to AWS IoT Core, and perform the pub/sub over MQTT 5.

Before getting started, make sure you have the following list of per-requisites in place:

Download and install Docker. It is very convenient to run obuspa in Docker.
Make sure you have an AWS account, and select a region where the AWS IoT Device Advisor is available, for example: eu-west-1 (Ireland).
Enable AWS IoT Logs, for example via the AWS IoT Console. You can find instructions here.

To this purpose, it is worth pointing out that support for MQTT 5 client-side is already implemented as part of obuspa. For more details, have a look at the mqtt.c file in obuspa. With this in mind, getting obuspa running and connecting to AWS IoT Core using MQTT 5 should be a matter of configuration, not code.

To create and test this configuration, we will be using the test/mqtt folder, the existing shared.sh script (with some minor changes required, which we will discuss later). We will create another test script, inspired by the existing MQTT test scripts. We can call it test-mqtt-aws.sh, and it can look as below:

#!/bin/bash
CURRENT_DIR="$(dirname $0)"

echo $CURRENT_DIR
CERTS_ARGS="-t tests/mqtt/certs-iot.pem -a tests/mqtt/client-iot.pem"
source "$CURRENT_DIR"/shared.sh || exit 1

#CERTS_ARGS="-t /etc/ssl/certs"

rm -rf "$LOG_FILE"
rm -rf "$GREP_FILE"
rm -rf "$DB_FILE"

trap cleanup EXIT
trap cleanup SIGINT

stop_obuspa
sleep 1
start_obuspa $CERTS_ARGS

techo
techo "Preparing USP Agent for AWS IoT MQTT V5.0."

add_client 1
verify_client 1

add_local_agent_mtp 1 || fail
add_local_agent_controller 1  || fail
add_local_agent_controller_mtp 1 || fail

set_parameter_client "ClientID" "uspTestAgent"
set_parameter_client "ProtocolVersion" "5.0"

configure_client "1" "<YOUR_AWS_IOT_CORE_ENDPOINT>" "8883"

obuspa_cmd -c set "Device.MQTT.Client.1.TransportProtocol" "TLS"

# Enable boot notify (Controller 1)

obuspa_cmd -c add Device.LocalAgent.Subscription
obuspa_cmd -c add Device.LocalAgent.Subscription.1

obuspa_cmd -c set "Device.LocalAgent.Subscription.1.NotifType" "ValueChange"
obuspa_cmd -c set "Device.LocalAgent.Subscription.1.ReferenceList" "Device."
obuspa_cmd -c set "Device.LocalAgent.Subscription.1.Enable" "true"
obuspa_cmd -c set "Device.LocalAgent.Subscription.1.Recipient" "true"


# Finally, enable client.
enable_client 1

sleep 2
# Log and wipe grep file
cat "$GREP_FILE" >> $LOG_FILE
echo "" > "$GREP_FILE"

techo "USP Agent Ready"
wait

Before you can run this script, the following extra steps are needed:

Replace the placeholder YOUR_AWS_IOT_CORE_ENDPOINT with the endpoint from your AWS account. You can read about the IoT endpoint here.
Create the following AWS IoT Resources:
- AWS IoT Device Policy (allowing iot:Connect,iot:Subscribe, iot:Publish, iot:Receive on the correct topics),
- Your AWS IoT Thing, with thingName: uspTestAgent, a new certificate, and attach the newly created IoT Policy, either via the AWS Console, CLI, or SDK.
You then need to save the device certificate, private key, as well as the root CA, in 2 separate files, which you will reference when starting the obuspa agent.

Note: certs-iot.pem must contain the root CA, and client-iot.pem the private key, followed by unique certificate, like this:
```
-----BEGIN RSA PRIVATE KEY-----
<body>
-----END RSA PRIVATE KEY-----
-----BEGIN CERTIFICATE-----
<body>
-----END CERTIFICATE-----
```
In the shared.sh script, in the add_local_agent_mtp(), the ResponseTopicConfigured parameter is configured to '/usp/endpoint/#’. According to the TR-181 Data Model specification, the value MUST NOT contain any wild card characters (“+”, “#”). So, you need to modify the Set cli call, and remove the # in the topic (Not doing this will force an MQTT client disconnect):
```
 obuspa_cmd -c set "Device.LocalAgent.MTP.$1.MQTT.ResponseTopicConfigured" "/usp/endpoint"
```
Additionally, set the QoS 1 on publish:
```
 obuspa_cmd -c set "Device.LocalAgent.MTP.$1.MQTT.PublishQoS" "1"
```
As a last step, to be able to start up the test script as you run your Docker container, you can replace the existing Docker file, with the following:

FROM ubuntu:focal

ENV MAKE_JOBS=8
ENV OBUSPA_ARGS="-v4"

# Add mosquitto latest ppa
RUN apt-get update && \
    apt-get install -y software-properties-common && \
    add-apt-repository ppa:mosquitto-dev/mosquitto-ppa

# Install dependencies
RUN apt-get update &&\
    apt-get -y install \
        libssl-dev \
        libcurl4-openssl-dev\
        libsqlite3-dev \
        libc-ares-dev \
        libz-dev \
        autoconf \
        automake \
        libtool \
        libmosquitto-dev \
        libwebsockets-dev \
        pkg-config \
        make \
        &&\
        apt-get clean

# Copy in all of the code
# Then compile, as root.
COPY . /obuspa/
RUN cd /obuspa/ && \
    autoreconf -fi && \
    ./configure && \
    make -j${MAKE_JOBS} && \
    make install

CMD cd obuspa &&  tests/mqtt/test_mqtt_aws.sh

Build and run your Docker image, and your test will run and start obuspa, connected to AWS IoT Core, using MQTT 5.

Verify MQTT 5 Integration/Protobuf Data Publishing

Once you have successfully run obuspa in Docker, you should see in the log file of the Agent log statements around the successful transition of the MQTT client state:

SendingConnect → [[ Connect sent ]] → AwaitingConnect, to AwaitingConnect → [[ Connect Callback Received ]] → Running

On the AWS IoT Core MQTT Broker side, you can validate that the USP agent is successfully connected, subscribed and publishing data, by looking at the AWS IoT Logs in Amazon CloudWatch, and searching for the following events:

CONNECT:

{
  "timestamp": "xxx",
  "logLevel": "INFO",
  "traceId": "xxx",
  "accountId": "xxxx",
  "status": "Success",
  "eventType": "Connect",
  "protocol": "MQTT",
  "clientId": "uspTestAgent",
  "principalId": "xxxx",
  "sourceIp": "xxxx",
  "sourcePort": 26259
}

PUBLISH:

{
"timestamp": "xxx",
"logLevel": "INFO",
"traceId": "xxx",
"accountId": "xxx",
"status":"Success",
"eventType":"Publish-In",
"protocol": "MQTT",
"topicName": "/usp/controller",
"clientId": "uspTestAgent",
"principalId": "xxx",
"sourceIp": "54.240.197.234",
"sourcePort": 26259
}

SUBSCRIBE:

{
"timestamp": "xxx",
"logLevel": "INFO",
"traceId": "xxx",
"accountId": "xxx",
"status": "Success",
"eventType": "Subscribe",
"protocol": "MQTT",
"topicName": "/usp/endpoint",
"clientId": "uspTestAgent",
"principalId": "xxx",
"sourceIp": "xxx",
"sourcePort": 26259
}

You can verify that the USP agent has successfully subscribed to the topic configured in your modified shared.sh(/usp/endpoint), by looking out for the following log statements in the agent logs:

Subscribe: Sending subscribe to /usp/endpoint 0 0
LogCallback: MQTT Debug: Client uspTestAgent sending SUBSCRIBE (Mid: 1, Topic: /usp/endpoint, QoS: 0, Options: 0x00)

Because of the USP subscription to changes in the Device. data model parameters, you should see MQTT publishing happening from the USP agent, for instantiated data model parameters changing frequently, such as Device.DeviceInfo.UpTime or Device.Time.CurrentLocalTime Examples of successful publishing in the agent logs will look like this:

NOTIFY sending at time 2022-12-08T11:27:23Z, to host <YOUR_AWS_IOT_CORE_ENDPOINT> over MQTT
LogCallback: MQTT Debug: Client uspTestAgent sending PUBLISH (d0, q1, r0, m8, '/usp/controller', ... (164 bytes))
Sending NotifyRequest (ValueChange for path=Device.DeviceInfo.UpTime)
....
LogCallback: MQTT Debug: Client uspTestAgent received PUBACK (Mid: 4, RC:0)
PublishCallback: Sent MID 4

If you log into your AWS account, navigate to AWS IoT, select the test MQTT client, and subscribe to the /usp/controller topic, you should see data coming in. Please note that the data is protobuf-encoded, as specified by the USP message encoding specification.

You can, for now, stop the Docker container in which obuspa is running. This will also ensure the termination of the MQTT connection.

Validate MQTT 5 Integration with AWS IoT Core Device Advisor “Qualification Program Test Suite"

After seeing successful MQTT connections, you can use AWS IoT Core Device Advisor, with its “Qualification Program Test Suite”, to validate MQTT 5 functionality and set-up.

Note: To use AWS IoT Core Device Advisor, you will need to confgure your device with the IoT Endpoint of Device Advisor, which is different than the production IoT Core Endpoint. Hence, this should be done at test phases in your development.

This step is extremely valuable during device software development, before on-boarding devices to AWS IoT Core. AWS IoT Core Device Advisor also allows you to create your own test suite, not as part of the qualification program, and configure the appropriate tests for your use case. We will however not cover custom suites in this blog post.

Let’s walk through the steps for setting up AWS IoT Core Device Advisor to run the test suite against the USP agent.

Navigate to the AWS Console, to IoT Core Device Advisor. Use one of the regions where Device Advisor is supported, such as Ireland (eu-west-1).
Under Test Suites, create a new Qualification Test Suite:
Choose the “AWS IoT Core qualification test suite” for MQTT 5, and click Next:
On the Configuration page, verify that all the tests in the suite are present (they should be there by default, so no action), and click Next.
Create a new Role for AWS IoT Core Device Advisor to perform the needed MQTT actions, and make sure you configure the correct Action permissions, on the correct resources:
Review the test suite, and click Create Test Suite.
Your newly create test suite should appear in the Test Suites section, and you can select it:
Open it, and click Run test suite under Actions:
Select your IoT thing, and copy the IoT Endpoint, as outlined below:
In your obuspa test script test-mqtt-aws.sh, replace the IoT endpoint with the Device Advisor endpoint, and rebuild your Docker image.
You can now click Run test, and immediately after run your USP Agent Docker container.

You will notice that Cloud-side, the test suite is running, and also, that in the obuspa logs, MQTT tests are being executed.

Be patient until the obuspa's MQTT client managed to successfully connect to the IoT endpoint of the Device Advisor, and observe the test suite execution steps, progress, and look at the logs.

If all the tests in the suites have passed, the validation has been successful. At the end, you can also download the report with the test results.

Conclusion

This post shows how the Open Source reference implementation of an USP Agent, obuspa, can be configured to connect, subscribe and publish data, over MQTT 5, to AWS IoT Core, following the release of MQTT 5 support for IoT Core. It also shows how to create and configure a “Qualification Program” MQTT 5 test suite, and run it against the agent during development phases.

For more information around USP, have a look at the specification released by the Broadband Forum, and the Open Source USP Agent reference implementation. If you would like to dive deeper into AWS IoT Core Device Advisor, here is a link to the developer guide. To understand better MQTT 5 support in AWS IoT Core, have a look at the technical documentation.

If you have any feedback about this post, or you would like to see more IoT related content, please reach out to me here, or on Twitter or LinkedIn.

How to ensure resilience for your AWS IoT Rules Engine to AWS Lambda integration

Alina Dima — Mon, 21 Nov 2022 09:12:30 +0000

If you are using AWS Lambda rule actions with your AWS IoT Rules, then this post might be of interest to you. I have a question to you: do you know for a fact that all your events are processed successfully, and the function execution is successful for all your events? How do you know that? If these questions make you curious, keep on reading.

This post will focus on explaining the asynchronous nature of the Lambda invocation from AWS IoT Rules Engine, and the impact it can have on your IoT application, potentially leading to message loss. We will also walk through a working set-up of some resilience steps, using AWS SAM, and AWS Lambda PowerTools for Typescript, and AWS X-Ray for observability.

Asynchronous Lambda invocations

When AWS IoT Rules Engine invokes AWS Lambda, the invocations are asynchronous. This means that AWS Lambda places the event in a queue and AWS IoT Rules records success, regardless of what happens next in your Lambda execution. A separate process reads events from the queue and sends them to your configured function.

Your AWS Lambda function consists of both your own code and the Lambda runtime itself, and both can be a source of errors that you should handle. So what happens if your function execution leads to an error, be it an error in the function’s code, or a runtime error, such as a timeout?

Well, in error case, as described here, AWS Lambda by default will attempt to run the function with the event 2 times more, with 1 minute wait between the first 2 attempts, and 2 minutes delays before the second and third attempt.

Additional situations can occur:

Your function does not have enough concurrency available to process all incoming events, which can well happen in an IoT application at scale. As a result, additional requests will get throttled (429 HTTP status) by the Lambda service, returned to the event queue, and retried as per Lambda strategy. The more events in the queue, the higher the retry interval set by Lambda, and the lower the rate at which it reads events from the queue.
The same event is received multiple times, due to eventual consistency of the event queue.
The function cannot keep up with the incoming events, and events are deleted from the queue without being sent to the function.
With events at scale, if the AWS Lambda event queue is very long, new events might expire before AWS Lambda gets to send them to your function. If events expire or processing fails, AWS Lambda discards them.

So clearly this can lead to data loss in your application.

Can the Error Action help?

The short answer is no, not in this situation.

The Error Action of the AWS IoT Rules Engine is a great tool for errors that prevent AWS Lambda from accepting the invocation, such as lack of permissions for the rule to trigger your function, or if AWS Lambda is not able to add the event to the queue, but it will not help with errors that happen in your Lambda function’s code, or Lambda runtime (like timeouts). Remember the asynchronous nature of the AWS Lambda function trigger from the IoT Rule.

So what can you do?

Luckily, there are a things you can do to ensure that your events do not get lost, and also to try and mitigate the above mentioned situations:

Ensure that you have enough concurrency to handle invocations. If you allow the default Lambda retries (2), this also means that these invocations will add to your concurrent invocation count, so keep this in mind.
Make sure your IoT Cloud application can handle duplicate events as required by your design, and don’t let duplicate events catch you by surprise. For example, ensure that your application code is idempotent, and you design for eventual consistency in your data storage.
You could reduce the number of retries that the Lambda service performs, or discard unprocessed events quicker. Here are the docs on how to do this. And if your application code is making calls to other services, you should build a retry strategy, as well as error handling in your application code, inside your Lambda function.
You can configure destinations for asynchronous invocation, for both successful and failed events. Destinations are used by Lambda to send your events to other services, as per configuration. Setting up destinations for failed executions in particular allows you to decouple the handling of your failures without data loss, even if you cannot accommodate Lambda service retries or you figure that your errors are not recoverable with retries. You can configure Amazon SQS, Amazon SNS, Lambda or EventBridge as destinations.
An alternative to destinations is configuring a dead-letter queue(part of your functions version-specific configuration) in your function’s configuration. Destinations however are more flexible (they are not locked in when you publish a version of your function), support additional targets, and include details about the function's response in the invocation record.
Using tracing tools like AWS X-Ray, or logging and metrics with Amazon CloudWatch, is great for visibility as to what happens to your AWS Lambda function invocations.

Let’s look at this in an example

The architecture of the example we are building is in the diagram below:

We have an AWS IoT Serverless application using AWS SAM, which is composed of an AWS Lambda function invoked from the Rules Engine on every message published by an IoT device on topic 'device/<thingName>/from-device'. For the purpose of our example, the Lambda function will simply throw an error.

Here is the AWS SAM template for this:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >
  rules-engine-to-lambda

  Sample SAM Template for rules-engine-to-lambda

Globals:
  Function:
    Timeout: 60
    Tracing: Active

Resources:
  RulesHandlingFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: app/
      Handler: app.lambdaHandler
      Runtime: nodejs14.x
      Architectures:
        - x86_64
      Tracing: Active
      Environment:
        Variables:
          POWERTOOLS_SERVICE_NAME: RulesToLambdaService
          POWERTOOLS_METRICS_NAMESPACE: rules-engine-to-lambda
          LOG_LEVEL: INFO
      EventInvokeConfig:
        DestinationConfig:
          OnFailure:
            Type: SQS
            Destination: !GetAtt FailedRequestsQueue.Arn
        MaximumEventAgeInSeconds: 120
        MaximumRetryAttempts: 1
      Events:
        IoTLambda:
          Type: IoTRule
          Properties:
            AwsIotSqlVersion: 2016-03-23
            Sql: SELECT * FROM 'device/+/from-device'

    Metadata:
      BuildMethod: esbuild
      BuildProperties:
        Minify: true
        Target: "es2020"
        EntryPoints:
        - app.ts

  FailedRequestsQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: "FailedRequestsQueue"
      VisibilityTimeout: 300
      KmsMasterKeyId: alias/aws/sqs

Outputs:
  RulesHandlingFunction:
    Description: "RulesHandlingFunction  ARN"
    Value: !GetAtt   RulesHandlingFunction.Arn
  RulesHandlingFunctionIamRole:
    Description: "Implicit IAM Role for RulesHandlingFunction"
    Value: !GetAtt   RulesHandlingFunction.Arn

And the example AWS Lambda function (app.ts), in Typescript. The Typescript function utilizes the Lambda Powertools for Typescript for an easy, descriptor based integration with AWS X-Ray and Amazon CloudWatch.

import { Context } from 'aws-lambda';
import { captureLambdaHandler, Tracer } from '@aws-lambda-powertools/tracer';
import middy from "@middy/core";
import {injectLambdaContext, Logger} from "@aws-lambda-powertools/logger";

const tracer = new Tracer();
const logger = new Logger();

export const lambdaHandler = middy(async (event: any, context: Context): Promise<any> => {
    logger.info('Event', event)
    throw new Error('This is a test error');
})
    .use(captureLambdaHandler(tracer))
    .use(injectLambdaContext(logger, {clearState: true}));

If you publish an event from your IoT device on the topic mentioned above, you can have a look in AWS X-Ray. The service map will look like below.

You can see the light brown circle around your Lambda function invocation, showing the 4xx Error. However, if you look at the trace, you will see the Response code is 202, as expected for a successful asynchronous invocation. As a reminder here, setting up an Error Action will not help, as from the perspective of your IoT Rule, the AWS Lambda invocation was successful, because AWS Lambda responded with a 202 status code, and there is no reason to invoke the Error Action.

Because we have set MaximumRetryAttempts to 1 and the MaximumEventAgeInSeconds to 120 seconds, if we open the trace we will see that Lambda attempted to invoke the function twice, both times with the same expected failure. If your application is processing large numbers of events, allowing the Lambda service to retry for you, especially with a high value of the MaximumEventAgeInSeconds, might not be the best strategy, due to the potential loss of newer events. You can also notice the different log entries in Amazon CloudWatch for each invocation, in the image below.

Because we have configured an Amazon SQS queue for failure cases, our event is not lost, but was sent to the SQS queue. The destination is configured with DestinationConfig in the SAM template. Once the messages land in the SQS queue, you can process them with an idempotent consumer, such as another AWS Lambda function, for example.

Have a look also at the GitHub repository with all the resources.

Conclusion

This post explains the behaviour of the AWS IoT Rules Engine invoking your AWS Lambda function, the asynchronous nature of the invocation, and the impact it can have on your IoT application, especially if it leads to potential data loss.

Tools like AWS X-Ray and the Lambda PowerTools help with tracing visibility for your Lambda invocations. Setting up a destination for failed events is a good strategy to ensure data is not getting lost, and you have the possibility for a decoupled handling of failed events, in a parallel path in your application. Programmatic error handling and retries in your Lambda function application code can also work towards your fallback strategy.

Have a look at the links throughout the blog post, to understand better how asynchronous invocations with AWS Lambda work.

If you find this interesting or have suggestions for future topics, feel free to reach out here or @fay_ette on Twitter or LinkedIn.

Some Tips & Tricks when working with AWS IoT Rules Engine

Alina Dima — Tue, 08 Nov 2022 13:46:46 +0000

The AWS IoT Rules Engine is widely used for routing IoT device messages to different downstream services, based on evaluation of rule statements and conditions. Even if working with the AWS IoT Rules Engine can be straightforward, there are a few tricks to keep in mind when things don’t go according to plan.

In this post, I will put forward a few tips that you should apply to your AWS IoT Rules-based implementation, to prevent failures, increase observability, and verify that everything works as expected.

Enable the AWS IoT Logs

Debugging IoT rule executions is not trivial. Incorrect syntax is usually detected at creation time, however, a lot of issues can only be detected at runtime. When and if something goes wrong, your best bet of identifying the problem is looking in the AWS IoT Logs. Once you enable the AWS IoT Logs, which you can do using the AWS CLI, AWS CloudFormation, CDK or the AWS Console, you should be able to explore the log streams in Amazon CloudWatch Logs, or, at an even more granular level, in CloudWatch Insights.

With Amazon CloudWatch Insights, for instance, you could examine all the failures in your rule execution with the following query:

fields @timestamp, @message | filter ruleName="<your-rule-name>" and status = 'Failure' | sort @timestamp desc

Don’t forget to configure an Error Action for your Rule

Any of the actions you configure for execution could fail at any point. This can be caused by various reasons, such as missing execution permissions (a very common error, especially in cross account executions), runtime errors, throughput exceeded exceptions appearing at high TPS, just to name a few.

To ensure resilience, you need to make sure that that you build a mitigation for this point of failure. This mitigation should come in the form of an Error Action, which log your error, ensure you don’t lose data, and allow you to build a decoupled error handling flow. Setting up an error action is easy.

Below is an AWS CloudFormation S3 Error Action example.

ActivityToKinesisTopicRule:
  Type: AWS::IoT::TopicRule
  Properties:
    RuleName: "<your-rule-name>"
    TopicRulePayload:
      RuleDisabled: false
      AwsIotSqlVersion: "2016-03-23"
      Sql: "Some SQL"
      Actions:
        - Kinesis:
            RoleArn:<your-kinesis-role>
            StreamName: "your-kinesis-data-stream"
            PartitionKey: "<your key>"
      ErrorAction:
        S3:
          BucketName: "<your-bucket-name>"
          Key: "error-${timestamp()}.json"
          RoleArn: "<your-action-role>"

Ensure that the SQL Version is set to "Latest" for your IoT Rule

So let me share with you a little anecdote to kick start this best practice section.

Only just last week, I was building a data ingestion pipeline routing time-series data from IoT devices into Amazon Kinesis Data Streams, via the Rules Engine’s Kinesis Action. As I was doing due diligence and checking if ingestion worked, I noticed that all the array key/value pairs from the device JSON payload were missing from the Kinesis records. No error, no warning, just missing.

I looked in the IoT logs, checked for Kinesis errors, built an Error Action. But nothing helped. All the executions were cleanly performed, nothing seemed wrong. As I looked closely at my IoT Rule as it was displayed in the AWS IoT console, I noticed that my rule’s SQL version was set to "2015-10-08". That was clearly not what I wanted, but, as I forgot to set the desired version in my AWS CloudFormation template, my rule resource got created with the default SQL version, which is the old SQL version "2015-10-08". If you create your rule using the AWS SDK, you will see the same behaviour. And, as the old version does not have the right support for arrays, my array key/value pairs in the payload were silently ignored .

I remembered later that, every once in a while, when I work with the AWS IoT Rules Engine, I found myself encountering one problem or another that ends up being related to using an old SQL version. The behaviour can be unpredictable, so don’t expect a straight forward error message.

The conclusion here is, as a best practice, make sure you set the SQL version to the desired version explicitly (for example the latest current version "2016-03-23") when you create your AWS IoT Rule.

Does your IoT Rule have the right permissions?

Failures in your rule execution can be caused by lack of permissions. Your AWS IoT Rule needs an IAM Rule and a policy that allows the AWS IoT service to perform the operations needed to execute the action you specified. If you forget to add this, your rule execution will fail with authorization errors. Below is an example of an AWS IAM role, that allows your rule to write to an AWS Kinesis Data Stream.

KinesisRuleRole:
  Type: AWS::IAM::Role
  Properties:
    AssumeRolePolicyDocument:
      Version: "2012-10-17"
      Statement:
        - Effect: Allow
          Principal:
            Service:
              - iot.amazonaws.com
          Action:
            - sts:AssumeRole
    Path: "/"
    Policies:
      - PolicyName: AllowKinesis
        PolicyDocument:
          Version: "2012-10-17"
          Statement:
            - Effect: Allow
              Action:
                - kinesis:PutRecord
              Resource: "<your-kinesis-data-stream-arn"

In conclusion, working with the AWS IoT Rules Engine can be straightforward, but, when issues occur, debugging is not always trivial. Having a few tricks up your sleeve, like the ones above, can help save some development and debugging time.
The four take-aways from this post are: enabling AWS IoT Logs, adding an Error Action, verifying the SQL version, and giving the AWS IoT Service the needed permissions for your action.

If you liked this post and would like to see more, follow me on Twitter (@fay_yette), or LinkedIn.

IoT Builders - Live Stream on Twitter Spaces

Alina Dima — Mon, 10 Oct 2022 12:02:10 +0000

What is IoT Builders
When?
How does it work?
What's coming up this week?
What did you miss so far?
- Episode #1
- Episode #2
- Episode #3
- Episode #4
- Episode #5
- Episode #6

What is IoT Builders?

IoT Builders is a live stream hosted by four AWS IoT Developer Advocates - 👩‍💻👨‍💻👨‍💻👨‍💻 (Alina, Dan, Nenad, and Syed), aiming to foster engaging conversations with community builders, who tune in weekly to share their experiences building IoT products.

When?

Weekly, on Wednesdays, at 5pm CET/ 8am PT.

How does it work?

We invite a new speaker weekly, and we keep the same set or sub-set of hosts. We go into interesting details, ranging from Embedded to Cloud, discuss IoT challenges, walk through educational material etc.

What's coming up this week?

This Wednesday, Oct 26th, we will be talking to Ali Benfattoum, Principal Tech Evangelist for #iot at #aws. Ali is an expert in #SmartCity #SmartBuilding, #LoRaWAN, and the creator of the #opensource Smart Territory Framework.

Here's the link to the space 🚀.

What did you miss so far?

If you have missed our live streams so far, don't worry! Here is a summary:

Episode #1: Live September 21st, with Ed Miller from Arm, about Arm Virtual Hardware and other IoT related topics. Below is the list of meaningful topics we covered in the stream.