Using CloudWatch Application Signal With Lambda Based Applications
In this post, we’ll see how we can use CloudWatch Application Signal to monitor a serverless application.
Step 1: Launch an EC2 instance and install node 18 and serverless package
wget -nv https://d3rnber7ry90et.cloudfront.net/linux-x86_64/node-v18.17.1.tar.gz
mkdir /usr/local/lib/node
tar -xf node-v18.17.1.tar.gz
mv node-v18.17.1 /usr/local/lib/node/nodejs
### Unload NVM, use the new node in the path, then install some items globally.
echo "export NVM_DIR=''" >> /home/ec2-user/.bashrc
echo "export NODEJS_HOME=/usr/local/lib/node/nodejs" >> /home/ec2-user/.bashrc
echo "export PATH=\$NODEJS_HOME/bin:\$PATH" >> /home/ec2-user/.bashrc
### Reload environment
. /home/ec2-user/.bashrc
### Verify NodeJS v18.x is operating
node -e "console.log('Running Node.js ' + process.version)"
npm install -g serverless
pip3 install urllib3==1.26.6
Step 2: Create serverless.yml with following content
service: user-management
provider:
name: aws
runtime: python3.10
region: us-east-1
iamRoleStatements:
- Effect: Allow
Action:
- dynamodb:PutItem
- dynamodb:GetItem
- dynamodb:DeleteItem
Resource: arn:aws:dynamodb:us-east-1:*:table/Users
functions:
createUser:
handler: handler.create_user
events:
- http:
path: users
method: post
cors: true
getUser:
handler: handler.get_user
events:
- http:
path: users/{userId}
method: get
cors: true
deleteUser:
handler: handler.delete_user
events:
- http:
path: users/{userId}
method: delete
cors: true
resources:
Resources:
UsersTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: Users
AttributeDefinitions:
- AttributeName: user_id
AttributeType: S
KeySchema:
- AttributeName: user_id
KeyType: HASH
ProvisionedThroughput:
ReadCapacityUnits: 5
WriteCapacityUnits: 5
and handler.py with following content:
import boto3
import json
import uuid
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Users')
# Create User
def create_user(event, context):
user_id = str(uuid.uuid4())
body = json.loads(event['body'])
name = body['name']
email = body['email']
table.put_item(
Item={
'user_id': user_id,
'name': name,
'email': email
}
)
return {
'statusCode': 201,
'body': json.dumps({'message': 'User created', 'user_id': user_id})
}
# Get User
def get_user(event, context):
user_id = event['pathParameters']['userId']
response = table.get_item(Key={'user_id': user_id})
user = response.get('Item')
if user:
return {
'statusCode': 200,
'body': json.dumps(user)
}
else:
return {
'statusCode': 404,
'body': json.dumps({'message': 'User not found'})
}
# Delete User
def delete_user(event, context):
user_id = event['pathParameters']['userId']
table.delete_item(Key={'user_id': user_id})
return {
'statusCode': 200,
'body': json.dumps({'message': 'User deleted'})
}
and deploy the stack using following content:
serverless deploy
Step 3: Create a Python script. Replace <API_URL> with your API gateway invocation URL.
import requests
import schedule
import time
import json
# Base URL of the API Gateway (replace with your deployed API Gateway URL)
API_BASE_URL = "<API_URL>/dev"
# Function to create a user
def create_user():
url = f"{API_BASE_URL}/users"
payload = {
"name": "Test User",
"email": "testuser@example.com"
}
response = requests.post(url, json=payload)
print("Create User Response:", response.status_code, response.json())
# Return the user_id for further use
if response.status_code == 201:
return response.json().get("user_id")
return None
# Function to get a user
def get_user(user_id):
url = f"{API_BASE_URL}/users/{user_id}"
response = requests.get(url)
print("Get User Response:", response.status_code, response.json())
# Function to delete a user
def delete_user(user_id):
url = f"{API_BASE_URL}/users/{user_id}"
response = requests.delete(url)
print("Delete User Response:", response.status_code, response.json())
# Function to run the workflow
def run_workflow():
print("\nRunning API Workflow...")
# Step 1: Create a user
user_id = create_user()
if user_id:
# Step 2: Wait for a moment before fetching the user
time.sleep(2)
get_user(user_id)
# Step 3: Wait again before deleting the user
time.sleep(2)
delete_user(user_id)
else:
print("User creation failed, skipping subsequent steps.")
# Schedule the workflow to run every minute
schedule.every(1).minutes.do(run_workflow)
# Run the schedule
print("Starting the scheduled workflow...")
while True:
schedule.run_pending()
time.sleep(1)
and execute the script
python3 script.py
Step 4: Go to your Lambda functions and enable Application Signals and Lambda service traces under Monitoring and operations tools. After some time, you can see 3 services in CloudWatch application signal console
and service map for application
Create SLO for all 3 services like this. This SLO mandates that latency for our services shouldn’t be greater than 50 ms.
After a while, you will see the SLI status as healthy
Step 5: Select a service and check the dependency list and metrics like latency.