Cloudwatch alarms to Slack notifications with CDK

Workspace access and permissions

Before setting up the Slack notifier itself, you have to configure Slack as a 'chat client' for AWS Chatbot via the console, which gives it access to your Slack workspace. Luckily, this is a one-time process. Make sure you choose the correct workspace!

Allowing AWS Chatbot access to your Slack workspace.

Slack notifier CDK configuration

While the notifier code can be directly defined in an individual CDK stack, for any multi-stack application (or if you need to send notifications to multiple Slack channels) it makes sense to define it as its own CDK construct that can be imported into any and all stacks.

The code below is an example of such a construct, with a few things worth considering:

The SlackNotifier class exposes the SNS topic as an instance variable with self.topic. This allows the topic to be specified as the target of a CloudWatch alarm action.
As explained in the AWS Chatbot docs, the actions that users can take from within the Slack channel are restricted by the "intersection of your guardrail policies and what is allowed by their roles". In the example below, the guardrail policy is a service-agnostic ReadOnlyAccess, with the channel role allowing CloudwatchReadOnlyAccess. As the former is a superset of the latter, users will in effect be able to execute all actions allowed by the CloudwatchReadOnlyAccess policy.

class SlackNotifier(Construct):
    def __init__(
        self,
        scope: Construct,
        id: str,
        slack_channel_configuration_name: str,
        slack_workspace_id: str,
        slack_channel_id: str,
    ) -> None:
        super().__init__(scope, id)

        self.topic = sns.Topic(self, "NotificationTopic")

        slackbot = chatbot.SlackChannelConfiguration(
            self,
            "SlackChannelConfiguration",
            slack_channel_configuration_name=slack_channel_configuration_name,
            slack_workspace_id=slack_workspace_id,
            slack_channel_id=slack_channel_id,
            notification_topics=[self.topic],  # type: ignore
            guardrail_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("ReadOnlyAccess")
            ],
        )
        slackbot.role.add_managed_policy(  # type: ignore
            iam.ManagedPolicy.from_aws_managed_policy_name("CloudwatchReadOnlyAccess")
        )

Adding CloudWatch alarms

With the SlackNotifier construct defined, we can configure CloudWatch alarms to push to the Slack channel. The code below is a simple demonstration of what can be achieved:

The SlackNotifier is created with the relevant Slack credentials.
We then create the actual alarm. In this case we monitor when the number of messages in a SQS dead letter queue rises above 5.
Lastly, the notifier SNS topic is configured as the target for the OK and ALARM CloudWatch actions so that a Slack notification is created when the CloudWatch metric both enters and exits the ALARM state.

slack_notifier = SlackNotifier(
    self,
    "SlackNotifier",
    slack_channel_configuration_name="TestChannel",
    slack_workspace_id=os.environ["CDK_SLACK_WORKSPACE_ID"],
    slack_channel_id=os.environ["CDK_TEST_SLACK_CHANNEL_ID"],
)
dead_letter_queue_alarm = cloudwatch.Alarm(
    self,
    "DeadLetterQueueAlarm",
    metric=test_queue.dead_letter_queue.queue.metric_approximate_number_of_messages_visible(),  # type: ignore
    threshold=5,
    evaluation_periods=3,
)
dead_letter_queue_alarm.add_alarm_action(
    cloudwatch_actions.SnsAction(slack_notifier.topic)  # type: ignore
)
dead_letter_queue_alarm.add_ok_action(
    cloudwatch_actions.SnsAction(slack_notifier.topic)  # type: ignore
)

Conclusion

Observability and monitoring systems are crucial components in modern web applications and services, providing real-time insights into performance issues as well as error detection and notification. In this blog post we've shown how CloudWatch alarms can be propagated to Slack, enabling prompt detection of potential issues in an application.

At Weird Sheep Labs, we've implemented systems such as this for multiple clients, helping them maintain better situational awareness about their applications and enabling them to respond to failures in a timely manner.

If this sounds like something your business could benefit from, please don't hesitate to get in touch!

`1`	`class SlackNotifier(Construct):`
`2`	`def __init__(`
`3`	`self,`
`4`	`scope: Construct,`
`5`	`id: str,`
`6`	`slack_channel_configuration_name: str,`
`7`	`slack_workspace_id: str,`
`8`	`slack_channel_id: str,`
`9`	`) -> None:`
`10`	`super().__init__(scope, id)`
`11`
`12`	`self.topic = sns.Topic(self, "NotificationTopic")`
`13`
`14`	`slackbot = chatbot.SlackChannelConfiguration(`
`15`	`self,`
`16`	`"SlackChannelConfiguration",`
`17`	`slack_channel_configuration_name=slack_channel_configuration_name,`
`18`	`slack_workspace_id=slack_workspace_id,`
`19`	`slack_channel_id=slack_channel_id,`
`20`	`notification_topics=[self.topic], # type: ignore`
`21`	`guardrail_policies=[`
`22`	`iam.ManagedPolicy.from_aws_managed_policy_name("ReadOnlyAccess")`
`23`	`],`
`24`	`)`
`25`	`slackbot.role.add_managed_policy( # type: ignore`
`26`	`iam.ManagedPolicy.from_aws_managed_policy_name("CloudwatchReadOnlyAccess")`
`27`	`)`

`1`	`slack_notifier = SlackNotifier(`
`2`	`self,`
`3`	`"SlackNotifier",`
`4`	`slack_channel_configuration_name="TestChannel",`
`5`	`slack_workspace_id=os.environ["CDK_SLACK_WORKSPACE_ID"],`
`6`	`slack_channel_id=os.environ["CDK_TEST_SLACK_CHANNEL_ID"],`
`7`	`)`
`8`	`dead_letter_queue_alarm = cloudwatch.Alarm(`
`9`	`self,`
`10`	`"DeadLetterQueueAlarm",`
`11`	`metric=test_queue.dead_letter_queue.queue.metric_approximate_number_of_messages_visible(), # type: ignore`
`12`	`threshold=5,`
`13`	`evaluation_periods=3,`
`14`	`)`
`15`	`dead_letter_queue_alarm.add_alarm_action(`
`16`	`cloudwatch_actions.SnsAction(slack_notifier.topic) # type: ignore`
`17`	`)`
`18`	`dead_letter_queue_alarm.add_ok_action(`
`19`	`cloudwatch_actions.SnsAction(slack_notifier.topic) # type: ignore`
`20`	`)`