🔉

AWS CDK（Python）を使ってCodePipelineからSlackへ通知できるようにする

D2C

2023/06/15に公開

はじめに

D2Cでエンジニアをしている、髙橋と申します。
最近業務内でAWS CDKを使ってCI/CDパイプラインを構築する機会があり、
CodePipelineからSlackに通知する部分を記述する際に、AWS公式ドキュメントやAPIリファレンス、その他記事などを参考にさせていただきました。

ただその中の多くがUI画面から構築していたり、CDKでTypeScriptを使っているものしか見つからず、Pythonで気軽に構築するにはハードルが少し高いなと感じました。

そのため実際に構築した流れや、実際のコードについて紹介させていただこうと思います。

全体の構成

構成図

実際にCDKで構築する

環境

[ ~/ ] $ python3 --version
Python 3.7.16
[ ~/ ] $ cdk --version
2.72.0 (build 397e46c)

1. AWSコンソールからChatBotとSlackを連携する

・AWSコンソールからChatBotを開き、画面右上のプルダウンメニューを押下

・Slackを選択し、「クライアントを設定」を押下

・表示される内容に問題がなければ、「許可する」を押下

2. Slackの通知対象チャンネルにて、AWSアプリを招待する（プライベートチャンネルのみ必須）

・以下の内容をチャンネル内でメッセージとして送信

/invite @aws

3. CDKを記述していく

・app.pyとは別にスタック用のファイルを用意し、そこにCodePipelineの通知ルールやSNS,ChatBotの構築コードを記載

notificator_stack.py

from aws_cdk import (
    Stack,
    aws_iam as iam,
    aws_chatbot as chatbot,
    aws_sns as sns,
    aws_codestarnotifications as notifications,
    aws_codepipeline as codepipeline,
)

from constructs import Construct


class NotificatorStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # CodePipeline用のSNSのトピックを作成し、ポリシーを設定
        topic = sns.Topic(
            self,
            id="Topic",
            topic_name="pipeline-notification-topic"
        )
        topic.add_to_resource_policy(
            # CodePipelineからSNSへのPublishを許可
            statement=iam.PolicyStatement(
                actions=["sns:Publish"], 
                resources=[topic.topic_arn],
                principals=[
                    iam.ServicePrincipal("codepipeline.amazonaws.com") 
                ]
            )
        )

        # Slackに通知するためのChatBotを作成
        chatbot_role = iam.Role(
            self,
            id="ChatbotRole",
            role_name="pipeline-chatbot-role",
            assumed_by=iam.ServicePrincipal("chatbot.amazonaws.com")
        )
        chatbot.SlackChannelConfiguration(
            self,
            id="Chatbot",
            notification_topics=[topic],  # 通知元のSNSトピックを指定
            slack_workspace_id="<Slackの通知したいチャンネル名があるWorkspace名>",
            slack_channel_configuration_name="<Slackの通知したいチャンネル名>",
            slack_channel_id="<Slackの通知したいチャンネルID>",
            role=chatbot_role
        )

        # イベント通知ルールを作成し、既存のCodePipelineに設定する
        pipeline = codepipeline.Pipeline.from_pipeline_arn(
            self,
            id="Pipeline",
            pipeline_arn="<対象とするCodePipelineのARN>"
        )
        notifications.NotificationRule(
            self,
            id="NotificationRule",
            source=pipeline,  # 通知ルールを設定するCodePipelineを指定
            targets=[topic],  # 通知先のSNSトピックを指定
            events=[
                # CodePipelineの開始と終了時に通知
                "codepipeline-pipeline-pipeline-execution-started",
                "codepipeline-pipeline-pipeline-execution-succeeded",
                "codepipeline-pipeline-pipeline-execution-failed",
                "codepipeline-pipeline-pipeline-execution-canceled",
                "codepipeline-pipeline-pipeline-execution-resumed",
            ],
            notification_rule_name="pipeline-notification-rule",
        )

・スタック構築用のクラスを呼び出す

app.py

import aws_cdk as cdk

from notificator_stack import NotificatorStack

app = cdk.App()
NotificatorStack(scope=app,
                 construct_id="NotificatorStack",
                 env=cdk.Environment(account="<AWSのアカウントID>", region="<AWSのリージョン名>"))

app.synth()

4. CDKで記述した内容をクラウド環境へデプロイする

[ ~/cicd ] $ cdk deploy --profile <デプロイ先のAWS環境のプロファイル名>

5. パイプラインを実行し、動作確認する

・こちらの画像はパイプラインが正常終了した際の通知メッセージ
通知_動作確認

まとめ

公式ドキュメントやAPIリファレンスを見ることはとても大事なことですが、
知りたい情報が散りばめられていたり、サクッと構築したい時には時間がかかりすぎたりと、ハードルが少し高い印象です。

まだCDKを使ってない方や、初めてPythonで書くといった方には、この記事で全体のイメージを掴んでいただいて、本番導入への第一歩としていただけると嬉しいなと思います。

最後までお読みいただき、ありがとうございました。

参考

D2C m-techPublication

株式会社D2C d2c.co.jp のテックブログです。 D2Cは、NTTドコモと電通などの共同出資により設立されたデジタルマーケティング企業です。ドコモの膨大なデータを活用した最適化を行える広告配信システムの開発をしています。