※※この記事は「Autifyアドベントカレンダー2022」の4日目記事です。
こんにちは!
Autifyに今年の10月に入社した、カスタマーサポートエンジニアのMamiです。
今回はAutifyでシナリオ実行が失敗した際に、Slackに通知するWebhookを作成してみましたので、ご紹介いたします。
作った背景
せっかく入社したので、何かつくってみようーと思い、実際にお問合せを受ける中で、ニーズがありそうだった失敗時の結果通知用Webhookを作ることにしました
※現状Autifyでは、テスト実行が完了した際にテスト結果を含む通知メッセージをSlackに送信するSlack Appを提供していますので、常に結果を受け取りたい場合は、こちらからご利用方法をご確認くださいませ。
構成
Webhookは、AWS Lambdaを利用して作成しました。
WebhookシークレットやSlackのWebhookURLの保存にはAWS Secret Managerを利用しました。
実際の作成方法
事前準備
事前に以下の準備をしてください。
- AWSアカウント
- 投稿用のSlackチャンネル
- Hattyの絵文字登録(社内のSlackチャンネルに登録いただくと、Hattyが結果をお知らせするように作成できます。)
- HattyのImageファイルはこちらからダウンロード可能です。
- 今回は、失敗通知なので、「hatty_bow」という名前でHattyがお辞儀している絵文字を登録して利用しました。
利用したコード
今回Webhook作成に利用したコードはこちらです。
Webhook用のコード
from email import header
import json, hmac, hashlib, base64, urllib,boto3,ast
from urllib.request import Request, urlopen
from botocore.exceptions import ClientError
def is_valid_signature(secret, signature, event):
body = event['body']
secret_key = secret['autify-webhook-secret']
new_hmac = "sha1=" + hmac.new(bytes(secret_key, "utf-8"), bytes(body, "utf-8"), hashlib.sha1).hexdigest()
return hmac.compare_digest(signature, new_hmac)
def get_secret():
secret_name = "autify-webhook"
region_name = "ap-northeast-1"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
# For a list of exceptions thrown, see
# https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
raise e
# Decrypts secret using the associated KMS key.
secret = get_secret_value_response['SecretString']
return secret
def post_slack_results(secret, event, context):
slack_body = json.loads(event['body'])
if slack_body['status'] == "failed":
if 'scenario_name' in slack_body:
send_data = {
"channel":secret['slack-channel'],
"username": "webhook-test-bot",
"text": ":hatty_bow: テスト失敗のお知らせです:hatty_bow: \n" + "シナリオ実行が失敗しました。URLから詳細を確認してください。\n" + "シナリオID:" + str(slack_body['scenario_id']) + "\nシナリオ名:" + slack_body['scenario_name'] + "\n実行結果:" + slack_body['url']
}
payload = "payload=" + json.dumps(send_data)
request = Request(
secret['slack-webhook-url'],
data=payload.encode("utf-8"),
method="POST"
)
with urlopen(request) as response:
response_body = response.read().decode("utf-8")
else:
return{
'statusCode': 200,
"body": "Test plan was executed."
}
else:
return {
'statusCode': 200,
"body": "The test was successful.",
}
def lambda_handler(event, context):
signature = event['headers']['x-autify-signature']
secret = ast.literal_eval(get_secret())
if is_valid_signature(secret,signature,event):
post_slack_results(secret,event,context)
return {
'statusCode': 200
}
else:
return {
"statusCode": 401,
"body": "Unauthorized"
}
1.SlackのWebhookURLを取得する。
はじめに通知用チャンネルにIncoming Webhookを追加します。
- Incoming Webhookのアプリページを開く
- 投稿用のチャンネルを選択し、Incoming Webhookインテグレーションを追加する
- 追加後に表示されるWebhookURLをメモする。
2.AWSでWebhookを作成する。
次にいよいよWebhookを作成していきます。
- AWS LambdaでWebhookを作成する。
- 設定を開き、関数URLを作成する。
※Webhookはインターネット公開する必要があるため、今回は認証タイプはNONE
を選択して作成します。
- AWS Secret Managerに必要な情報を保存する。
- AWS LambdaのIAMロールにAWS Sectet Managerの権限を付与する。
- IAMのサービスへ移動する
- ロールを開き、AWS Lambdaの名前を入力し作成したLambda関数用のロールの検索を行う
- ロールを選択し、以下ポリシーの追加を行う
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": "secretsmanager:GetSecretValue",
"Resource": "*"
}
]
}
3.AutifyにWebhookを設定する。
最後にWebhookをAutifyに設定します。
4.実際にシナリオを実施して通知を確認する。
実際にシナリオを実行しSlackに通知がされることを確認してください。
※投稿されたものを見ると、おそらくHattyの絵文字部分が残念な感じになっていると思うので、適したものに変えてください
おわりに
今回はとてもとてもシンプルな通知の内容になっておりますが、Webhookでは今回の通知に利用した内容以外も受け取っています。
こちらのページをご覧の上、必要な情報を追加してご利用いただければと思います!
また、実際作ってみるた感想としては、失敗だけの通知というのは中々切ないものがありました
作ってみたはものの、成功と失敗どちらも通知される方が精神衛生的にはいいかもしれない・・・と思ったりもしましたw
この記事がどなたかのお役に立てますと幸いです。
かなり早いですが、メリークリスマス
参考
今回SlackへのPostをする・Webhookシークレットを検証するという実装にあたって、以下の記事を参考にさせていただきました!ありがとうございました!