Real-Time AWS CloudFormation Stack Monitoring with Slack Notifications

Introduction

Monitoring AWS CloudFormation stack events is crucial to track deployments, updates, and failures. Instead of manually checking AWS Console or Amazon CloudWatch logs, you can automate notifications to a Slack channel.

This post will guide you through setting up an AWS Lambda function that listens to Amazon SNS messages from AWS CloudFormation and sends notifications to Slack.

Pioneers in Cloud Consulting & Migration Services

Reduced infrastructural costs
Accelerated application deployment

Get Started

How It Works?

An Amazon SNS Topic is created to receive AWS CloudFormation stack events.
AWS Lambda function subscribes to this Amazon SNS topic and processes messages.
The function extracts details like stack name, resource status, and physical resource ID.
If the event is important (e.g., failure or rollback), it is formatted and sent to Slack.

Setting Up the Slack Notification Bot

You must set up a Slack bot with a webhook URL to send messages to a Slack channel. You can follow the guide in this reference to create a Slack webhook and obtain the required URL.

AWS CloudFormation Template

This template provisions:

An AWS IAM Role for the Lambda function.
AWS Lambda function to process SNS events and send Slack notifications.
An Amazon SNS Topic to receive AWS CloudFormation stack events.
AWS Lambda Subscription to Amazon SNS for automatic invocation.

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: "Template to create Lambda function, SNS topic, SNS subscription, IAM role, and SSM parameter for CFN Stack notification for Slack."

Parameters:
  EnvironmentType:
    Type: String
    Description: "Environment of Deployment"
    AllowedValues: ["dev", "qa", "qa2", "uat", "int", "int2", "production"]
    Default: "dev"

Resources:
  CFNStatusNotifierRole:
    Type: "AWS::IAM::Role"
    Properties:
      RoleName: "CFNStatusNotifierRole"
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service: "lambda.amazonaws.com"
            Action: "sts:AssumeRole"
      Policies:
        - PolicyName: "CloudWatchPutLogsPolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action:
                  - "logs:CreateLogGroup"
                  - "logs:CreateLogStream"
                  - "logs:PutLogEvents"
                Resource: "*"
        - PolicyName: "CloudFormationAccessPolicy"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action:
                  - "cloudformation:DescribeStackResources"
                  - "cloudformation:DescribeStacks"
                Resource: "*"

  CFNStatusNotifierFunction:
    Type: 'AWS::Serverless::Function'
    Properties:
      FunctionName: "CFNStatusNotifier"
      CodeUri: LambdaFunction/
      Handler: "index.lambda_handler"
      Role: !GetAtt CFNStatusNotifierRole.Arn
      Runtime: "python3.12"
      Timeout: 60
      Environment:
        Variables:
          SLACK_WEBHOOK_URL: 
            !Sub '{{resolve:ssm:/CFN/slack/webhook/url}}'

  CFNStatusSNSTopic:
    Type: "AWS::SNS::Topic"
    Properties:
      TopicName: "CFNStatusNotificationTopic"

  CFNStatusSNSSubscription:
    Type: "AWS::SNS::Subscription"
    Properties:
      TopicArn: !Ref CFNStatusSNSTopic
      Protocol: "lambda"
      Endpoint: !GetAtt CFNStatusNotifierFunction.Arn

  CFNStatusNotifierLambdaPermission:
    Type: "AWS::Lambda::Permission"
    Properties:
      FunctionName: !Ref CFNStatusNotifierFunction
      Action: "lambda:InvokeFunction"
      Principal: "sns.amazonaws.com"
      SourceArn: !Ref CFNStatusSNSTopic

AWSTemplateFormatVersion: "2010-09-09"

Transform: AWS::Serverless-2016-10-31

Description: "Template to create Lambda function, SNS topic, SNS subscription, IAM role, and SSM parameter for CFN Stack notification for Slack."

Parameters:

EnvironmentType:

Type: String

Description: "Environment of Deployment"

AllowedValues: ["dev", "qa", "qa2", "uat", "int", "int2", "production"]

Default: "dev"

Resources:

CFNStatusNotifierRole:

Type: "AWS::IAM::Role"

Properties:

RoleName: "CFNStatusNotifierRole"

AssumeRolePolicyDocument:

Version: "2012-10-17"

Statement:

- Effect: "Allow"

Principal:

Service: "lambda.amazonaws.com"

Action: "sts:AssumeRole"

Policies:

- PolicyName: "CloudWatchPutLogsPolicy"

PolicyDocument:

Version: "2012-10-17"

Statement:

- Effect: "Allow"

Action:

- "logs:CreateLogGroup"

- "logs:CreateLogStream"

- "logs:PutLogEvents"

Resource: "*"

- PolicyName: "CloudFormationAccessPolicy"

PolicyDocument:

Version: "2012-10-17"

Statement:

- Effect: "Allow"

Action:

- "cloudformation:DescribeStackResources"

- "cloudformation:DescribeStacks"

Resource: "*"

CFNStatusNotifierFunction:

Type: 'AWS::Serverless::Function'

Properties:

FunctionName: "CFNStatusNotifier"

CodeUri: LambdaFunction/

Handler: "index.lambda_handler"

Role: !GetAtt CFNStatusNotifierRole.Arn

Runtime: "python3.12"

Timeout: 60

Environment:

Variables:

SLACK_WEBHOOK_URL:

!Sub '{{resolve:ssm:/CFN/slack/webhook/url}}'

CFNStatusSNSTopic:

Type: "AWS::SNS::Topic"

Properties:

TopicName: "CFNStatusNotificationTopic"

CFNStatusSNSSubscription:

Type: "AWS::SNS::Subscription"

Properties:

TopicArn: !Ref CFNStatusSNSTopic

Protocol: "lambda"

Endpoint: !GetAtt CFNStatusNotifierFunction.Arn

CFNStatusNotifierLambdaPermission:

Type: "AWS::Lambda::Permission"

Properties:

FunctionName: !Ref CFNStatusNotifierFunction

Action: "lambda:InvokeFunction"

Principal: "sns.amazonaws.com"

SourceArn: !Ref CFNStatusSNSTopic

AWS Lambda Function Code

This function:

Parses Amazon SNS messages from AWS CloudFormation.
Extract stack details and resource statuses.
Sends notifications to Slack for relevant events.

import os
import json
import urllib.request

def parse_message_body(message_body):
    """Parses the SNS message body (key-value formatted string) into a dictionary."""
    message_dict = {}
    for line in message_body.split("\n"):
        if "=" in line:
            key, value = line.split("=", 1)
            message_dict[key.strip()] = value.strip().strip("'")  # Remove leading/trailing spaces & quotes
    return message_dict

def extract_stack_name(stack_id):
    """Extracts the stack name from the full StackId ARN."""
    parts = stack_id.split("/")
    if len(parts) >= 2:
        return parts[1]  # Stack name is the second part
    return stack_id  # Fallback

def extract_physical_resource_id(full_id):
    """Extracts the required part of PhysicalResourceId."""
    parts = full_id.split("/")
    if len(parts) >= 3:
        return parts[1]  # Extract only the required part
    return full_id  # Fallback

def lambda_handler(event, context):
    print("Received event:", json.dumps(event, indent=2))

    for record in event.get("Records", []):
        try:
            # Extract raw SNS message
            sns_message_raw = record.get("Sns", {}).get("Message", "")
            if not sns_message_raw:
                raise ValueError("SNS Message is missing or empty")
            
            # Parse SNS message body
            sns_message = parse_message_body(sns_message_raw)
            print("Parsed SNS Message:", json.dumps(sns_message, indent=2))

            # Extract relevant details
            stack_id_raw = sns_message.get("StackId", "Unknown")
            stack_name = extract_stack_name(stack_id_raw)
            
            physical_resource_id_raw = sns_message.get("PhysicalResourceId", "Unknown")
            physical_resource_id = extract_physical_resource_id(physical_resource_id_raw)

            resource_status = sns_message.get("ResourceStatus", "Unknown")

            # **Skip sending message if status is non-critical**
            if resource_status in ["CREATE_COMPLETE", "CREATE_IN_PROGRESS", "REVIEW_IN_PROGRESS", "ROLLBACK_IN_PROGRESS", "UPDATE_COMPLETE", "UPDATE_COMPLETE_CLEANUP_IN_PROGRESS", "UPDATE_IN_PROGRESS", "UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS", "UPDATE_ROLLBACK_IN_PROGRESS"]:
                print(f"Skipping message as Resource Status is {resource_status}")
                continue  # Skip sending the message
            
            # Format Slack message
            slack_message = {
                "text": "*AWS CloudFormation Notification* :zap:\n\n"
                        f":bookmark: *Stack Name:* `{stack_name}`\n"
                        f":link: *Physical Resource ID:* `{physical_resource_id}`\n"
                        f":traffic_light: *Resource Status:* `{resource_status}`\n"
            }

            # Send to Slack
            slack_webhook_url = os.getenv("SLACK_WEBHOOK_URL")
            if not slack_webhook_url:
                raise ValueError("SLACK_WEBHOOK_URL is not set in the environment variables")

            json_message = json.dumps(slack_message).encode("utf-8")
            req = urllib.request.Request(
                url=slack_webhook_url,
                data=json_message,
                headers={"Content-Type": "application/json"}
            )
            with urllib.request.urlopen(req) as response:
                print("Message sent to Slack successfully:", response.read().decode("utf-8"))

        except Exception as e:
            print("Failed to process record:", e)

import os

import json

import urllib.request

def parse_message_body(message_body):

"""Parses the SNS message body (key-value formatted string) into a dictionary."""

message_dict = {}

for line in message_body.split("\n"):

if "=" in line:

key, value = line.split("=", 1)

message_dict[key.strip()] = value.strip().strip("'") # Remove leading/trailing spaces & quotes

return message_dict

def extract_stack_name(stack_id):

"""Extracts the stack name from the full StackId ARN."""

parts = stack_id.split("/")

if len(parts) >= 2:

return parts[1] # Stack name is the second part

return stack_id # Fallback

def extract_physical_resource_id(full_id):

"""Extracts the required part of PhysicalResourceId."""

parts = full_id.split("/")

if len(parts) >= 3:

return parts[1] # Extract only the required part

return full_id # Fallback

def lambda_handler(event, context):

print("Received event:", json.dumps(event, indent=2))

for record in event.get("Records", []):

try:

# Extract raw SNS message

sns_message_raw = record.get("Sns", {}).get("Message", "")

if not sns_message_raw:

raise ValueError("SNS Message is missing or empty")

# Parse SNS message body

sns_message = parse_message_body(sns_message_raw)

print("Parsed SNS Message:", json.dumps(sns_message, indent=2))

# Extract relevant details

stack_id_raw = sns_message.get("StackId", "Unknown")

stack_name = extract_stack_name(stack_id_raw)

physical_resource_id_raw = sns_message.get("PhysicalResourceId", "Unknown")

physical_resource_id = extract_physical_resource_id(physical_resource_id_raw)

resource_status = sns_message.get("ResourceStatus", "Unknown")

# **Skip sending message if status is non-critical**

if resource_status in ["CREATE_COMPLETE", "CREATE_IN_PROGRESS", "REVIEW_IN_PROGRESS", "ROLLBACK_IN_PROGRESS", "UPDATE_COMPLETE", "UPDATE_COMPLETE_CLEANUP_IN_PROGRESS", "UPDATE_IN_PROGRESS", "UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS", "UPDATE_ROLLBACK_IN_PROGRESS"]:

print(f"Skipping message as Resource Status is {resource_status}")

continue # Skip sending the message

# Format Slack message

slack_message = {

"text": "*AWS CloudFormation Notification* :zap:\n\n"

f":bookmark: *Stack Name:* `{stack_name}`\n"

f":link: *Physical Resource ID:* `{physical_resource_id}`\n"

f":traffic_light: *Resource Status:* `{resource_status}`\n"

}

# Send to Slack

slack_webhook_url = os.getenv("SLACK_WEBHOOK_URL")

if not slack_webhook_url:

raise ValueError("SLACK_WEBHOOK_URL is not set in the environment variables")

json_message = json.dumps(slack_message).encode("utf-8")

req = urllib.request.Request(

url=slack_webhook_url,

data=json_message,

headers={"Content-Type": "application/json"}

)

with urllib.request.urlopen(req) as response:

print("Message sent to Slack successfully:", response.read().decode("utf-8"))

except Exception as e:

print("Failed to process record:", e)

Conclusion

With this setup:

AWS CloudFormation stack events are pushed to an Amazon SNS topic.
The AWS Lambda function listens to events and posts critical updates to Slack.
You get real-time notifications for failures and rollback actions.

This solution improves visibility and response time for AWS CloudFormation changes in your AWS environment.

Drop a query if you have any questions regarding AWS CloudFormation and we will get back to you quickly.

Making IT Networks Enterprise-ready – Cloud Management Services

Accelerated cloud migration
End-to-end view of the cloud environment

Get Started

About CloudThat

CloudThat is an award-winning company and the first in India to offer cloud training and consulting services worldwide. As an AWS Premier Tier Services Partner, AWS Advanced Training Partner, Microsoft Solutions Partner, and Google Cloud Platform Partner, CloudThat has empowered over 1.1 million professionals through 1000+ cloud certifications, winning global recognition for its training excellence, including 20 MCT Trainers in Microsoft’s Global Top 100 and an impressive 14 awards in the last 9 years. CloudThat specializes in Cloud Migration, Data Platforms, DevOps, Security, IoT, and advanced technologies like Gen AI & AI/ML. It has delivered over 750 consulting projects for 850+ organizations in 30+ countries as it continues to empower professionals and enterprises to thrive in the digital-first world.

FAQs

1. How do I set up the Slack Webhook URL?

ANS: – You need to create a Slack app and generate an incoming webhook URL. Follow these steps:

Go to Slack API Apps
Click “Create New App”
Select “From Scratch” and provide an app name
Enable “Incoming Webhooks”
Add a new webhook and choose a Slack channel
Copy the generated webhook URL and store it in the AWS Systems Manager Parameter Store under /CFN/slack/webhook/url

2. Can I filter which AWS CloudFormation events trigger Slack notifications?

ANS: – Yes, you can modify the AWS Lambda function to exclude events like CREATE_COMPLETE or UPDATE_COMPLETE and only send failure or rollback notifications.

WRITTEN BY Deepak S

Deepak S is a Senior Research Associate at CloudThat, specializing in AWS services. He is passionate about exploring new technologies in cloud and is also an automobile enthusiast.