Send CloudWatch alarms to Slack via AWS Lambda

Introduction

One of the key factors in maintaining production uptime is the implementation of effective monitoring and alert systems. Integrating AWS CloudWatch with Slack using AWS Lambda allows you to receive real-time alerts in your Slack channels, keeping your team informed and responsive. This blog will guide you do this integration using a AWS Lambda function for sending AWS CloudWatch alarms to Slack.

Objective

The article covers setting up an incoming webhook in your Slack workspace to receive messages from AWS and how to create an AWS Lambda function to process CloudWatch alarms and send formatted notifications to the Slack channel via the webhook URL.

Prerequisites

• AWS Account
• Slack

Steps by Step Procedure –

1. Creation of a SNS Topic

Cloudwatch uses SNS topics to publish messages. So, In your AWS console, visit Simple Notification Service, then enter the name of the topic that you want to create, and click on Next Step under the topic name.

Note: Select type as Standard and the topic name that you have entered earlier should be visible in the Name section, confirm that and create a topic.

2. Creation of Slack channel and Slack App

Now, Create a slack channel for receiving cloudwatch alerts. Let’s give it a name Test and choose visibility as per your need.

For creating a slack app, visit https://api.slack.com/apps and click on Create an app. Choose From scratch option for creating a slack app

Give a name to your app and select the same workspace in you have created the channel for receiving the alerts to develop your app.

We will be using webhooks for receiving alerts from cloudwatch. So, let’s activate incoming webhooks and the webhook to your channel in which you want to receive alerts.

Go to Basic Information and give basic details like the App icon and description to your app. It’s an optional step

3. Creation of Lambda Function

Go to the AWS Management Console and navigate to Lambda and click Create function, and choose Author from scratch. Provide basic details to your lambda function and select Python3.9 runtime as this the python version which our lambda function will be using.

After the creation of the Lambda, we need a trigger to invoke our lambda function whenever any Cloudwatch event takes place.

Select the topic that we have created earlier to send alerts to slack. AWS will add a resource-based policy to the function and add a subscription to the topic.

Now, Replace the below prewritten function with the below function and give your webhook url to slack_url variable, give your region to region variable and Deploy the function.

import urllib3
import urllib3
import json
import logging

slack_url = "Your WebHook URL"
http = urllib3.PoolManager()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()

def get_alarm_attributes(sns_message):
    alarm = dict()
    
    alarm['name'] = sns_message['AlarmName']
    alarm['description'] = sns_message.get('AlarmDescription', 'No description provided.')
    alarm['reason'] = sns_message['NewStateReason']
    alarm['region'] = sns_message['Region']
    alarm['state'] = sns_message['NewStateValue']
    alarm['previous_state'] = sns_message['OldStateValue']
    alarm['timestamp'] = sns_message['StateChangeTime']

    # Use the correct AWS region code
    region = "us-east-1"
    alarm_name = sns_message['AlarmName']
    alarm['link'] = f"https://{region}.console.aws.amazon.com/cloudwatch/home?region={region}#alarmsV2:alarm/{alarm_name}"

    return alarm

def register_alarm(alarm):
    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": ":warning: " + alarm['name'] + " alarm was registered"
                }
            },
            {
                "type": "divider"
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Alarm Name:* {alarm['name']}\n\n*Description:* {alarm['description']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Region:* {alarm['region']}"},
                    {"type": "mrkdwn", "text": f"*Timestamp:* {alarm['timestamp']}"}
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f""
                }
            }
        ]
    }

def activate_alarm(alarm):
    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": ":red_circle: Alarm: " + alarm['name'],
                }
            },
            {
                "type": "divider"
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Alarm Name:* {alarm['name']}\n\n*Description:* {alarm['description']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Region:* {alarm['region']}"},
                    {"type": "mrkdwn", "text": f"*Timestamp:* {alarm['timestamp']}"}
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f""
                }
            }
        ]
    }

def resolve_alarm(alarm):
    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": ":large_green_circle: Alarm: " + alarm['name'] + " was resolved",
                }
            },
            {
                "type": "divider"
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Alarm Name:* {alarm['name']}\n\n*Description:* {alarm['description']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Region:* {alarm['region']}"},
                    {"type": "mrkdwn", "text": f"*Timestamp:* {alarm['timestamp']}"}
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f""
                }
            }
        ]
    }

def insufficient_data_alarm(alarm):
    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": ":grey_question: Alarm: " + alarm['name'] + " is in INSUFFICIENT_DATA state",
                }
            },
            {
                "type": "divider"
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Alarm Name:* {alarm['name']}\n\n*Description:* {alarm['description']}"
                }
            },
            {
                "type": "section",
                "fields": [
                    {"type": "mrkdwn", "text": f"*Region:* {alarm['region']}"},
                    {"type": "mrkdwn", "text": f"*Timestamp:* {alarm['timestamp']}"}
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f""
                }
            }
        ]
    }

def lambda_handler(event, context):
    sns_message = json.loads(event["Records"][0]["Sns"]["Message"])
    alarm = get_alarm_attributes(sns_message)

    msg = str()

    if alarm['previous_state'] == "INSUFFICIENT_DATA" and alarm['state'] == 'OK':
        msg = register_alarm(alarm)
    elif alarm['previous_state'] == 'OK' and alarm['state'] == 'ALARM':
        msg = activate_alarm(alarm)
    elif alarm['previous_state'] == 'ALARM' and alarm['state'] == 'OK':
        msg = resolve_alarm(alarm)
    elif alarm['state'] == 'INSUFFICIENT_DATA':
        msg = insufficient_data_alarm(alarm)
    elif alarm['previous_state'] == 'INSUFFICIENT_DATA' and alarm['state'] == 'ALARM':
        msg = activate_alarm(alarm)

    encoded_msg = json.dumps(msg).encode("utf-8")
    try:
        resp = http.request("POST", slack_url, body=encoded_msg, headers={'Content-Type': 'application/json'})
        logger.info(
            {
                "message": msg,
                "status_code": resp.status,
                "response": resp.data,
            }
        )
    except Exception as e:
        logger.error(f"Error sending message to Slack: {e}")

        

Note: You can find your webhook url in the Incoming Webhooks tab in your Slack application.

4. Create a Cloudwatch Alarm

Now, create a CloudWatch alarm and configure it to send notifications to the Lambda function via an SNS topic.

Go to the AWS CloudWatch console and navigate to the Alarms section. Click “Create Alarm” and choose the appropriate metric for your use case (e.g., CPU utilisation, memory usage, or custom metrics).

Set the desired threshold and configure the alarm’s behaviour (e.g., send a notification when the metric exceeds a certain value).

Under “Send a notification to,” select “SNS Topic” and choose the SNS Topic you created in the previous steps.

Review the alarm configuration and click “Create Alarm.”

For testing the CPU utilization, the Stress Linux utility can be used. Install it by running sudo apt install stress, and then for testing, execute stress –cpu 2.

Conclusion

You have now successfully set up an AWS Lambda function to send CloudWatch alarms to a Slack channel. This integration allows you to stay informed about your AWS resources’ health and performance in real-time, directly over your team slack channel.

Feel free to customize the Lambda function further as per your specific needs. For example, you can add more details to the Slack messages, handle different types of alerts differently, or integrate with other monitoring tools.