AWS Edge computing example with Lambda and IoT Greengrass (version 1)
Table of contents
Introduction: AWS IoT Greengrass
Nowadays, with the huge development of different smart devices from personal smartphones to cars and starship modules, edge comping becomes an important consideration to make smart devices autonomous, not to rely on internet connection all the time and to use computing capabilities of the device to execute some operations directly on it. This is what edge computing is. In this post, I describe a simple example of the implementation of edge computing with AWS IoT Greengrass and AWS lambda written in Java.
Greengrass is the software from Amazon that extends cloud capabilities to local devices. It has different features that allow arranging edge computing. To get more on that, please refer to the official AWS documentation by the link given below:
Test scenario
Let’s use Greengrass to implement the simple architecture in the picture given below.
The device has some YAML files. It reads them and publishes the content to the Greengrass topic via MQTT. The Core device uses the Lambda function deployed on it to subscribe to the topic, consume the messages with file content and convert it into JSON. The converted message is sent to the SQS queue for further processing. For both devices, AWS EC2 instances will be used to make the setup available for anyone.
Implementation
The Classic (V1) Greengrass will be used for implementation.
The application code and scripts can be found in the GitHub repository: https://github.com/rimironenko/aws-iot-greengrass-project
1. Setup and configure Greengrass group
- Open the AWS console and search for “Greengrass”.
2. Go to IoT Greengrass and select the “Classic (V1)” menu item from the left nav menu.
3. Click “Create Group” and then click “Use default creation”.
4. Fill in the Group name (e.g. “MyFirstGroup”) and click “Next”.
5. The setup flow automatically creates the name for the Core device, rename it if you want, and click “Next”.
6. Click “Create Group and Core”.
Greengrass executes all the listed steps and marks them as done.
7. Download the Core’s security resources by clicking the “Download these resources as a tar.gz” button. It will download the archive to your computer.
8. Upload the archive to an existing S3 bucket or create a new one and upload it there (to be used for the automated Core Device setup through EC2 user data).
9. Create a Greengrass group. To allow it to interact with the AWS resources, assign a necessary IAM role to it: select the “Settings” group menu and assign an IAM role with access to the AWS resources required by business functionality.
2. Setup EC2 instance as the Greengrass Core device
In the AWS console, go to EC2 and provision a new EC2 instance.
- Search for “EC2” in the search bar and click “EC2”.
2. Click the “Launch instance” button on the right top.
3. Select the “Amazon Linux 2 AMI” image marked as “Free tier eligible”
4. Click “Select”. On the next screen please ensure that the “t2.micro” type is used (marked as “Free tier eligible”) and click the “Next: Configure Instance Details” button on the bottom left.
5. Fill in the user data text field with the script that will configure the instance as the Greengrass Core device. Please replace “ACCESS_KEY” and “SECRET_ACCCESS_KEY” with valid developer keys for your AWS account and “S3_URI” with the S3 URI of the archive with the Core’s secure resources created at step 1.7.
#!/bin/bash
# Create Greengrass user and group
adduser --system ggc_user
groupadd --system ggc_group
# Extract and run the following script to mount cgroups.
# This allows AWS IoT Greengrass to set the memory limit for Lambda functions.
# Cgroups are also required to run AWS IoT Greengrass in the default containerization mode.
cd /home/ec2-user
curl https://raw.githubusercontent.com/tianon/cgroupfs-mount/951c38ee8d802330454bdede20d85ec1c0f8d312/cgroupfs-mount > cgroupfs-mount.sh
chmod +x cgroupfs-mount.sh
bash ./cgroupfs-mount.sh
# Install Java
amazon-linux-extras enable corretto8
yum -y install java-1.8.0-amazon-corretto-devel
# Greengrass Lambdas require 'java8' executable, not 'java'
# See https://gist.github.com/noahcoad/92133670d6189440f883d9369211aeca
mv /usr/bin/java /usr/bin/java8
# Download and install Core software
curl https://d1onfpft10uf5o.cloudfront.net/greengrass-core/downloads/1.11.4/greengrass-linux-x86-64-1.11.4.tar.gz > greengrass-linux-x86-64-1.11.4.tar.gz
tar -xzvf greengrass-linux-x86-64-1.11.4.tar.gz -C /
# Download and install Core device certificates
export AWS_ACCESS_KEY_ID=ACCESS_KEY
export AWS_SECRET_ACCESS_KEY=SECRET_ACCCESS_KEY
aws s3 cp S3_URI certs.tar.gz
tar -xzvf certs.tar.gz -C /greengrass
# Download and install Root CA certificate
cd /greengrass/certs/
wget -O root.ca.pem https://www.amazontrust.com/repository/AmazonRootCA1.pem
# Start the Greengrass daemon
cd /greengrass/ggc/core/
./greengrassd start6. Click the “Next: Add Storage” button on the bottom right.
7. Click the “Next: Add Tags” button on the bottom right.
8. Add the tags to your instance (e.g. “app: Greengrass”) and click the “Next: Configure Security Group” button on the bottom right.
9. Choose the “Create a new security group” radio button. Name the group as “Greengrass” (feel free to use your name). Add a custom TCP rule with the 8883 port and the source as “Custom 0.0.0.0/0” (it is required by Greengrass for MMQT communications).
10. Click the “Review and Launch” button on the bottom right.
11. Click the “Launch” button on the bottom right, create a new key pair or choose an existing one, tick the “I acknowledge…” checkbox, and click “Launch instances”.
12 Wait until the instance gets the “Running” state, then connect to it in any way and ensure that the Greengrass is running by the command given below:
ps aux | grep "greengrass"
The Greengrass process should be displayed.
3. Register Greengrass Device
- Go to Greengrass in the AWS console.
- Select the group that we created before.
- Select the “Devices” menu item and click the “Add Device” button.
4. Click “Create New Device”.
5. Type in the device name and click “Next”.
6. Click “Use Defaults”.
7. Download the Device’s security resources by clicking the “Download these resources as a tar.gz” button and click “Finish”.
8. Upload the archive to the S3 bucket that was used to upload the archive for the core device before (to be used for the automated Core Device setup through EC2 user data).
9. Provision another EC2 instance following the same steps as were given above for the Core device, but with the following differences:
- When configuring the instance details, ensure that the same subnet as for the instance with the Core device is used
- The user data script is given below. As before, please replace “ACCESS_KEY” and “SECRET_ACCESS_KEY” with valid developer keys and S3_URI by the S3 URI of the archive with the device certificates.
#!/bin/bash
# Make a folder for the Greengrass device data
cd /home/ec2-user
mkdir publisher_device
cd publisher_device
# Download the Device certificates
export AWS_ACCESS_KEY_ID=ACCESS_KEY
export AWS_SECRET_ACCESS_KEY=SECRET_ACCCESS_KEY
aws s3 cp S3_URI device-certs.tar.gz
tar -xzvf device-certs.tar.gz
wget -O root.ca.pem https://www.amazontrust.com/repository/AmazonRootCA1.pem
# Install the Python SDK to run the script that will send a message to the Greengrass Core
yum -y install git
git clone https://github.com/aws/aws-iot-device-sdk-python.git
cd aws-iot-device-sdk-python
python3 setup.py install
cp -R AWSIoTPythonSDK ../AWSIoTPythonSDK
cd ..
rm -rf aws-iot-device-sdk-python10. If everything was done right, once you connect to the instance by SSH, there will be a “/home/ec2-user/publisher_device” folder containing the files as in the picture given below (you will have the different hash in your certificate names).
4. Create a Lambda function for the message transforming
Lambda function was bootstrapped by the AWS SAM framework. I will not dive into details about it here, please see my post about AWS SAM to learn more.
The code is given below.
package com.home.amazon.iot.lamnda;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import com.home.amazon.iot.model.Item;
import software.amazon.awssdk.core.SdkSystemSetting;
import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.SendMessageRequest;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
public class SubscribeFunction implements RequestHandler<Map<String, String>, String> {
private static final ObjectMapper YAML_MAPPER = new ObjectMapper(new YAMLFactory());
private static final ObjectMapper JSON_MAPPER = new ObjectMapper();
private final SqsClient sqsClient;
private final String queueUrl;
public SubscribeFunction() {
sqsClient = SqsClient.builder()
.region(Region.of(System.getenv(SdkSystemSetting.AWS_REGION.environmentVariable())))
.httpClientBuilder(UrlConnectionHttpClient.builder())
.build();
queueUrl = System.getenv("QueueUrl");
}
@Override
public String handleRequest(Map<String, String> input, Context context) {
if (input != null && input.get("message") != null) {
try {
String message = input.get("message");
String fileName = "test-" + System.currentTimeMillis();
Path tempFile = Files.createTempFile(fileName, ".yaml");
Files.write(tempFile, message.getBytes());
Item item = YAML_MAPPER.readValue(tempFile.toFile(), Item.class);
String json = JSON_MAPPER.writeValueAsString(item);
System.out.println("Converted to JSON: " + json);
sendSqsMessage(json);
} catch (IOException e) {
System.out.println("Failed to save message: " + e);
}
}
return "Subscribe function is invoked";
}
private void sendSqsMessage(String json) {
SendMessageRequest sendMsgRequest = SendMessageRequest.builder()
.queueUrl(queueUrl)
.messageBody(json)
.delaySeconds(5)
.build();
sqsClient.sendMessage(sendMsgRequest);
}
}Highlights:
- Jackson libraries are used to work with JSON and YAML.
System.outis used for logging instead of Lambda logger because the function is executed on the Core devices and the system output goes to the log file on the device’s file system.- The Lambda function is deployed to the cloud by
sam buildandsam deploy --guidedcommands. - The “Greengrass” Lambda Alias is auto-published by the SAM template to point the Greengrass group to the latest version of the function.
- The URL of the SQS queue is put to the Lambda environment variables by AWS SAM
5. Configure the group and subscriptions
- Go to Greengrass in the AWS console.
- Select the group that we created before.
- Click on the “Lambdas” menu item and click the “Add Lambda” button.
4. Click the “Use existing Lambda” button.
5. Select the “Subscribe function” Lambda from the list and select the “Alias: Greengrass” radio button.
6. Click “Finish”.
7. Now, we need to update the configuration of the function in the Greengrass group. Click on the triple dot and click the “Edit configuration” option.
8. Change the memory limit to 128MB.
9. Add the “QueueUrl” Environment variable and put the URL of the SQS queue as the value. Please note that the import of a Lambda function to a Greengrass group does not import the Environment variables. Thus, the environment variable and its value should be defined manually in the Lambda configuration in the Greengrass group.
9. Click “Update”.
10. Now, select the “Subscriptions” menu item in the Greengrass group and click the “Add Subscription” button.
11. Select the device as the source and the Lambda as the target and click “Next”. Type in any topic and click “Next”.
12. Click “Finish”.
13. Now, we need to configure logging for troubleshooting purposes. Please click the “settings” menu item, find the “Local logs configuration” item in the end, and click “Edit”.
14. Select both “User lambda logs” and “Greengrass system logs” options and click “Save”.
15. Finally, we need to assign an IAM role to the Greengrass group to give it permissions to send messages to the SQS.
16. Select the “Settings” group menu item and edit the Group role on top. To experiment with Greengrass, I assigned the admin role created before.
6. Deploy the function and subscription to the Core device
- Click the “Actions” item on the top right of the group description and click “Deploy”.
2. Select the automatic detection by pressing the corresponding button.
3. Click the “Deployments” menu item and wait until the deployment is completed.
7. Test the communication
- Log in to the Device EC2 instance by SSH and navigate to the “/home/ec2-user/publisher_device” folder.
2. Create a Python file named “sendMessage.py” with code that sends messages to the Core device:
import os
import sys
import time
import uuid
import json
import logging
import argparse
from AWSIoTPythonSDK.core.greengrass.discovery.providers import DiscoveryInfoProvider
from AWSIoTPythonSDK.core.protocol.connection.cores import ProgressiveBackOffCore
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
from AWSIoTPythonSDK.exception.AWSIoTExceptions import DiscoveryInvalidRequestException
# General message notification callback
def customOnMessage(message):
print('Received message on topic %s: %s\n' % (message.topic, message.payload))
MAX_DISCOVERY_RETRIES = 10
GROUP_CA_PATH = "./groupCA/"
# Read in command-line parameters
parser = argparse.ArgumentParser()
parser.add_argument("-e", "--endpoint", action="store", required=True, dest="host", help="Your AWS IoT custom endpoint")
parser.add_argument("-r", "--rootCA", action="store", required=True, dest="rootCAPath", help="Root CA file path")
parser.add_argument("-c", "--cert", action="store", dest="certificatePath", help="Certificate file path")
parser.add_argument("-k", "--key", action="store", dest="privateKeyPath", help="Private key file path")
parser.add_argument("-n", "--thingName", action="store", dest="thingName", default="Bot", help="Targeted thing name")
parser.add_argument("-t", "--topic", action="store", dest="topic", default="sdk/test/Python", help="Targeted topic")
parser.add_argument("-f", "--file", action="store", dest="file", default="/test.json", help="File to send content to topic")
args = parser.parse_args()
host = args.host
rootCAPath = args.rootCAPath
certificatePath = args.certificatePath
privateKeyPath = args.privateKeyPath
clientId = args.thingName
thingName = args.thingName
topic = args.topic
if not args.certificatePath or not args.privateKeyPath:
parser.error("Missing credentials for authentication, you must specify --cert and --key args.")
exit(2)
if not os.path.isfile(rootCAPath):
parser.error("Root CA path does not exist {}".format(rootCAPath))
exit(3)
if not os.path.isfile(certificatePath):
parser.error("No certificate found at {}".format(certificatePath))
exit(3)
if not os.path.isfile(privateKeyPath):
parser.error("No private key found at {}".format(privateKeyPath))
exit(3)
# Configure logging
logger = logging.getLogger("AWSIoTPythonSDK.core")
logger.setLevel(logging.DEBUG)
streamHandler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)
# Progressive back off core
backOffCore = ProgressiveBackOffCore()
# Discover GGCs
discoveryInfoProvider = DiscoveryInfoProvider()
discoveryInfoProvider.configureEndpoint(host)
discoveryInfoProvider.configureCredentials(rootCAPath, certificatePath, privateKeyPath)
discoveryInfoProvider.configureTimeout(10) # 10 sec
retryCount = MAX_DISCOVERY_RETRIES
discovered = False
groupCA = None
coreInfo = None
while retryCount != 0:
try:
discoveryInfo = discoveryInfoProvider.discover(thingName)
caList = discoveryInfo.getAllCas()
coreList = discoveryInfo.getAllCores()
# We only pick the first ca and core info
groupId, ca = caList[0]
coreInfo = coreList[0]
print("Discovered GGC: %s from Group: %s" % (coreInfo.coreThingArn, groupId))
print("Now we persist the connectivity/identity information...")
groupCA = GROUP_CA_PATH + groupId + "_CA_" + str(uuid.uuid4()) + ".crt"
if not os.path.exists(GROUP_CA_PATH):
os.makedirs(GROUP_CA_PATH)
groupCAFile = open(groupCA, "w")
groupCAFile.write(ca)
groupCAFile.close()
discovered = True
print("Now proceed to the connecting flow...")
break
except DiscoveryInvalidRequestException as e:
print("Invalid discovery request detected!")
print("Type: %s" % str(type(e)))
print("Error message: %s" % str(e))
print("Stopping...")
break
except BaseException as e:
print("Error in discovery!")
print("Type: %s" % str(type(e)))
print("Error message: %s" % str(e))
retryCount -= 1
print("\n%d/%d retries left\n" % (retryCount, MAX_DISCOVERY_RETRIES))
print("Backing off...\n")
backOffCore.backOff()
if not discovered:
print("Discovery failed after %d retries. Exiting...\n" % (MAX_DISCOVERY_RETRIES))
sys.exit(-1)
# Iterate through all connection options for the core and use the first successful one
myAWSIoTMQTTClient = AWSIoTMQTTClient(clientId)
myAWSIoTMQTTClient.configureCredentials(groupCA, privateKeyPath, certificatePath)
myAWSIoTMQTTClient.onMessage = customOnMessage
connected = False
for connectivityInfo in coreInfo.connectivityInfoList:
currentHost = connectivityInfo.host
currentPort = connectivityInfo.port
print("Trying to connect to core at %s:%d" % (currentHost, currentPort))
myAWSIoTMQTTClient.configureEndpoint(currentHost, currentPort)
try:
myAWSIoTMQTTClient.connect()
connected = True
break
except BaseException as e:
print("Error in connect!")
print("Type: %s" % str(type(e)))
print("Error message: %s" % str(e))
if not connected:
print("Cannot connect to core %s. Exiting..." % coreInfo.coreThingArn)
sys.exit(-2)
message = {}
file = open(args.file)
fileContent = file.read()
file.close()
message['message'] = fileContent
messageJson = json.dumps(message)
myAWSIoTMQTTClient.publish(topic, messageJson, 0)
print('Published topic %s: %s\n' % (topic, messageJson))
time.sleep(15)3. Create a YAML file with test content in the device folder and name it “test.yaml”. Put the following content inside it:
id: 1
message: TestAs a result, the folder will have the content as in the picture given below:
4. Run the following command:
python3 sendMessage.py \
--endpoint ENDPOINT \
--rootCA root.ca.pem \
--cert HASH.cert.pem \
--key HASH.private.key \
--thingName THING_NAME \
--topic TOPIC \
--file test.yamlReplace the HASH with the hash of your device, TOPIC with the topic that was configured in the Greengrass Group subscriptions (e.g., “greengrass/test/pubsub”) as was given above and THING_NAME by the Device name given to it in the group.
To obtain the endpoint, go to the Greengrass settings in the AWS Console (in the left menu).
For my particular device, the command would be as given below:
python3 sendMessage.py \
--endpoint a2jbo9ibf7ygli-ats.iot.us-east-1.amazonaws.com \
--rootCA root.ca.pem \
--cert 7caa99279d.cert.pem \
--key 7caa99279d.private.key \
--thingName PublisherDevice \
--topic greengrass/test/pubsub \
--file test.yamlIf the message was sent correctly, you should see the “Published topic…” message at the end of the output.
8. In the AWS console, go to SQS, select the created SQS queue, and poll for messages. The JSON messages from the Lambda on the Core device will appear there.
9. Clean up your resources:
- Terminate the EC2 instances
- Remove the Greengrass Group. It will ask you to reset the deployments first, just click the “Actions” in the group and click “Reset deployments”. Then remove the group.
- Remove the devices from the “Manage / Things” left menu item.
- Remove the certificates from the “Secure / Certificates” left menu item.
- Remove the policies from the “Secure / Policies” left menu item.
- Remove the application resources by
sam deletecommand (or manually delete the CloudFormation stack in the AWS console).
Troubleshooting
In this section, I will mention the main issues that you may face if you will follow the guideline given above.
- To investigate any issue with the Core device, please go through the logs in the “/greengrass/ggc/var/log” folder (use
sudoto have access to all the files).
- system/runtime.log — the info about the Greengrass and registration of the Lambda function
- system/localwatch/localwatch.log — the info about local issues on the Core device that may cause connectivity issues
- user/{AWS_REGION}/{AWS_ACCOUNT_ID} — the folder with the Lambda log file. All the
System.outoutputs from the Java code go to this file.
These logs should provide you all the information that is required for troubleshooting.
If you do not see logs, please ensure that the logging was configured in the Greengrass group in the AWS Console.
2. If you noticed “java8: executable not found” in the runtime.log file, please ensure that Java is installed and the executable was moved from Java to Java8 by mv /usr/bin/java /usr/bin/java8 (currently, this command is a part of the user data script for the Core device).
3. If a message from the Device is not sent, please ensure that the Greengrass daemon is running on the core device. It does not start automatically, so after restarting the EC2 instance you have to start it with the commands given below:
sudo su
cd /greengrass/ggc/core/
./greengrassd start4. If the Greengrass daemon is running but the message is still not sent, please ensure that your EC2 instances are in the same subnet (otherwise the Device instance may not see the Core Device instance).
5. if you are experimenting with your device, please note that the Greengrass takes the private IPv4 instance of the EC2 Core device instance. You need to change the address to the public IPv4 address by the following steps:
- Go to the Core settings of the Device group in the AWS console.
- Choose the “Connectivity” menu item.
- Click “Edit”, change the IPv4 address and click “Update”.
Please note that every deployment will reset the IPv4 address to the private one and you have to change it back again manually.
Conclusion
Advantages of AWS IoT Greengrass
- Greengrass works fine for the cases that require Edge computing and allows establishing the whole infrastructure.
- Greengrass allows using Lambda functions that is the straightforward way to use a custom programming code with the custom business logic.
Disadvantages of AWS IoT Greengrass
- The configuration of Greengrass may be an overhead even for a simple application.
- If you want to use the Lambda functions written in Java, it will require more effort comparing to Python. Also, such functions cannot co-exist if you are going to use Stream Manager (https://docs.aws.amazon.com/greengrass/v1/developerguide/stream-manager.html).
Constraints of AWS IoT Greengrass
- Lambda functions for Greengrass require the language-specific SDK to be packaged with the Lambda function.
- As of today (September 2021), only Python, Java, Node.js, and C are supported and you cannot use any other languages (https://docs.aws.amazon.com/greengrass/v1/developerguide/what-is-gg.html#gg-core-sdk-download).
- The environment variables for a Lambda function are not copied automatically when the function is added to a Greengrass group and should be copied manually.
- Stream Manager requires Java executable and cannot co-exist with Lambda functions written in Java, because the functions require Java8 executable instead.
In the next post, I will consider the implementation of the same setup using the newest version of AWS IoT Greengrass (version 2).
