Using CloudWatch Application Signal With Lambda Based Applications

4 min readDec 2, 2024

Reference: https://aws.amazon.com/blogs/aws/track-performance-of-serverless-applications-built-using-aws-lambda-with-application-signals/

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