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 a leading provider of Cloud Training and Consulting services with a global presence in India, the USA, Asia, Europe, and Africa. Specializing in AWS, Microsoft Azure, GCP, VMware, Databricks, and more, the company serves mid-market and enterprise clients, offering comprehensive expertise in Cloud Migration, Data Platforms, DevOps, IoT, AI/ML, and more.

CloudThat is the first Indian Company to win the prestigious Microsoft Partner 2024 Award and is recognized as a top-tier partner with AWS and Microsoft, including the prestigious ‘Think Big’ partner award from AWS and the Microsoft Superstars FY 2023 award in Asia & India. Having trained 650k+ professionals in 500+ cloud certifications and completed 300+ consulting projects globally, CloudThat is an official AWS Advanced Consulting Partner, Microsoft Gold Partner, AWS Training Partner, AWS Migration Partner, AWS Data and Analytics Partner, AWS DevOps Competency Partner, AWS GenAI Competency Partner, Amazon QuickSight Service Delivery Partner, Amazon EKS Service Delivery Partner, AWS Microsoft Workload Partners, Amazon EC2 Service Delivery Partner, Amazon ECS Service Delivery Partner, AWS Glue Service Delivery Partner, Amazon Redshift Service Delivery Partner, AWS Control Tower Service Delivery Partner, AWS WAF Service Delivery Partner, Amazon CloudFront, Amazon OpenSearch, AWS DMS, AWS Systems Manager, Amazon RDS, and many more.

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 works as a Research Intern at CloudThat. His expertise lies in AWS's services. Deepak is good at haunting new technologies and automobile enthusiasts.