PageSpeedInsights APIを定期実行してスコアをSlack通知する

ページ読み込み速度など、サイトのパフォーマンスを定期的に監視し続け、低下したタイミングですぐに検知可能な状態にする、安価でお手軽な仕組みを作って欲しいと依頼を頂いた時に作ってみた物のメモです。

という事で、今回作る物のシステム的な機能用件を並べた内容が以下4つです。

1. サイトのパフォーマンス（今回のMUSTはページ読み込み速度）を測定する処理が、自動で定期実行されている
2. 上記処理で測定した結果が、DBに保存されている
   （後々可視化する為で、今回は保存されていればOKとの事）
3. 上記処理の測定結果で異常発生（パフォーマンス低下）を検知した際に、何らかの手段ですぐに気付ける
4. 上記全てを実現する上で、可能な限りコストが安く、運用の手間がかからない
   （稼働後に運用上の変更が発生しても最小限の変更で対応可能）

全体アーキテクチャを考える

今回の要望を満たす手段は色々考えられるので、まずはどういうアーキテクチャで実現するかを考え、以下のような全体構成で作る事に決めました。

かなり横長な図になり見辛くて申し訳ないです（一応クリックで拡大可能です）

一気に沢山の登場人物が出てきましたが、上記の図を分解していくと大まかに3つのステップに分けられます。

1. サイトパフォーマンスを測定して結果をDBに保存

ステップ1の責務範囲は、サイトパフォーマンスとしてページ読み込み速度を測定し、結果をDBに保存する処理を自動で定期実行させるところまでです。
機能用件の「運用の手間がかからない」からも分かるように、運用後に何かが変わった（通知先の場所が変わった、パフォーマンス低下を定義する閾値が変わった etc）場合の変更コストをできるだけ少なくする為にも、各処理毎に責務範囲を適切に分割して、一つの処理の責務範囲を大きくし過ぎない事はとても大事です。

• ページ読み込み速度測定方法

用件を満たす方法は色々ありますが、可能な限りコストが安く、運用の手間がかからないという点に最もマッチするGoogleのPageSpeed Insightsを利用する事にしました。
Lighthouseの方が計測には高性能ですが、今回の対象がパフォーマンス観点だけだったので、実行時環境に依存しない分正確なスコアが期待でき、分析エンジンにもLighthouseと同じ内容を組み込んだPageSpeed InsightsのAPI v5を使用しています。

• PageSpeed Insightsの実行形式/発火トリガー

PageSpeed InsightsはAPIが用意されている為、これをLambda functionで定義したコード内から呼んで測定結果を取得します。
イベントトリガーは、機能用件として自動で定期実行されている事が求められているので、CloudWatchEventsのスケジュール実行です。

• 測定結果のDBへの保存

保存先のDBには、DynamoDBを選択しました。
理由は、今回の用途に対してその他DBよりも料金が安く、コストを安くの機能用件に適している為です。

尚、実行形式/発火トリガーについて、コストをかけずにPageSpeed Insights APIを叩く処理を自動で定期実行させる機能用件を満たすだけであれば、GASなどの手軽な方法を選択しても実現可能です。
ですが、今回は保存先DBにDynamoDBを選択している為、実行ロールにDynamoDBへのアクセス許可ポリシーを付与できるLambda functionがベストプラクティスと言えそうです。
GASなどからDynamoDBにアクセスしようとすると、アクセス許可ポリシーを付与したIAM ユーザのクレデンシャル情報を使用してアクセスするセキュリティ的に非推奨な方法を選択しなければなりません。

2. DBから結果を取り出して通知用サービスに転送

ステップ2の責務範囲は、測定結果のスコアをDBから取得し、パフォーマンス低下の定義に定めた値と比較してパフォーマンス低下が発生しているかを判定、判定結果を通知用サービスに送るところまでです。
パフォーマンス低下が発生していた場合のみ通知するのではなく、判定結果によって通知先のトピックを分け、判定結果OK or NGの通知をそれぞれ行う方法を今回は取りました。

• 通知用サービス

通知用サービスには、SNSを選択しました。
理由は、通知送信先が変更された場合でもトピックに設定するサブスクライバを変えるだけで自動的に送信先が変更可能で、運用の手間がかからないの機能用件に最もマッチしている為です。

• 送信先のSNSトピック

送信先のSNSトピックは、判定結果OKの時に通知を送信するトピックと、NGの時に通知を送信するトピックをそれぞれ用意しました。
これは、同一トピックに正常時の通知と異常通知が送られると通知送信先も同一の場所になってしまい、異常検知時のアラート通知が目立たなくなり、結果として通知結果自体が見られなくなってしまうのを防ぐ為です。
判定結果NGの時に送信するトピックにはパフォーマンス低下発生時のみ通知が送信されてくるようにして、異常発生時に送るトピックだけは見てもらうようにするのが狙い

• 処理の実行形式/発火トリガー

DynamoDBのアイテム変更を検知してLambda functionで処理を行います。
イベントトリガーは、DynamoDBのアイテム変更です。

Lambda functionを実行形式に選んだのは、

• DynamoDBのレコード変更を検知して処理を開始できること
• 実行ロールに処理内でアクセスする必要があるDynamoDBとSNSへのアクセス許可ポリシーを付与できるセキュリティ面

発火条件のサービスに対するアクセス許可ポリシーは通常リソースポリシーで設定しますが、DynamoDBなどストリームをポーリングしてイベントを検知するタイプは、実行ロールへアクセス許可ポリシーを付与します

の二点が理由です。

尚、ステップ1の方のLambdaでPageSpeed Insights APIの実行結果をSNSへ送るところまでやってしまうのも一つの手ですが、「測定結果をDBに保存する」のと、「結果を見てパフォーマンス低下が発生しているか判定し、判定結果を通知する」のでは目的が違う為、責務範囲分離目的で処理を分けています。

3. 転送された通知メッセージをユーザーに送信

ステップ3の責務範囲は、SNSのトピックに送られてきた通知メッセージをユーザーに対して送信する最後の部分です。

• SNSトピックに設定するサブスクライバ

SNSのトピックに設定するサブスクライバには、一番気付きやすいとの理由から依頼いただいた方からSlackをご指定いただきました。
Slackはルームを作成できる為、PageSpeed Insights APIの実行結果がパフォーマンス低下していた場合とそうでなかった場合で通知先のルームを分けてあげれば上記狼少年になってしまうリスクも回避できそうです。

• 処理の実行形式/発火トリガー

SNSでメッセージを受信したのをアイテム変更を検知してLambda functionで処理を行います。
イベントトリガーは、SNSのメッセージ受信です。
SNSトピックのサブスクライバにSlackを指定する最も楽な方法として、完全GUIで設定を完了できるAWS Chatbotというサービスがあります。
しかし、2020年3月現在ではまだAWS Chatbotで作成したサブスクライバをSNSトピックに設定する為には、幾つかの限られた方法でSNSに対してメッセージを送信した場合にしか対応しておらず、今回のLambda functionからSNSにメッセージをpublishする方法では使用する事ができない為、SNSからSlackへの通知送信の処理もLambda functionで行います。

作ってみる

全体アーキテクチャが決まったところで、実際に作っていきます。

1. サイトパフォーマンスを測定して結果をDBに保存

まずは、PageSpeed Insights APIを実行して結果をDynamoDBに保存するLambdaの作成からですが、Lambdaを作成する前の事前準備として以下3つが必要です。

事前準備1. PageSpeed Insights API実行用のAPI Key取得/設定

PageSpeed Insights APIを実行するにはAPI Keyが必要で、まずはこれの取得です。

1. GCP プロジェクト作成

GCPコンソール画面にて、API Keyを紐付けるプロジェクトを作成します。
既に紐付けたいプロジェクトを作成済の場合は、この手順は飛ばします

2. Google DevelopersからAPI Key作成

Google DevelopersのPageSpeed Insights APIを使ってみるページ中央部にある「キーを取得する」ボタン押下後、作成したプロジェクトを選択して「Next」を押すとAPI Keyを作成できます。


3. 作成されたAPI Keyをメモ


作成したAPI KeyはPageSpeed Insights APIでのみ使用可能なAPI Keyにしておいた方が安全な為、GCP HOMEから、「APIとサービス」 → 「認証情報」と飛び、作成したAPI Keyを選択。
APIの制限欄から「キーを制限」にチェックを入れ、「PageSpeed Insights API」にだけチェックを入れて「保存」で、使用可能なサービスを制限しておく事を推奨します。

4. 作成したAPI Keyをパラメータストアに登録

API KeyはLambdaのコードから参照できる必要がありますが、クレデンシャルな情報はコード内に直接書かずセキュアに管理できるパラメータストアに保存し、コードからパラメータストアの情報を読み込む形で使用します。

パラメータストアへの登録はマネジメントコンソールから可能で、「Systems Manager」 → 「パラメータストア」 → 「パラメータの作成」と進みます。


表示された項目を入力していき、パラメータの作成ボタン押下で完了です。

名前	保存する値のKey名
説明	保存する値に付ける説明。省略可能
利用枠	通常よりも大きなサイズや有効期限を付ける場合に詳細を選択する。今回は標準で十分
タイプ	保存形式。暗号化はしておきたい為、安全な文字列推奨
KMSの主要なソース	KMSキーを使用するアカウント。自アカウントのKMSキーを使用する場合は現在のアカウントを選択
KMSキー ID	使用するKMSキー
値	保存する値。今回はPageSpeed Insights APIのAPI Keyを設定
タグ	タグ名。管理用にNameタグは設定しておく事を推奨

以上でPageSpeed Insights API実行用の、API Keyの取得/設定は完了です。

事前準備2. DynamoDBのテーブル作成

続いて、結果を保存するDynamoDBのテーブルを作成します。

DynamoDBのテーブル作成はマネジメントコンソールからDynamoDBの管理画面に移り、トップのダッシュボード画面に存在する「テーブルの作成」を選択します。

表示された項目をそれぞれ入力していき、作成ボタン押下で完了です。

テーブル名	作成するテーブル名
プライマリキー	テーブル内のレコードを一意に識別する為のPK設定。DynamoDBはこの設定が凄く大事なので、切り出して後述します
テーブル設定	デフォルト設定を使用するかの選択。今回の用途ではデフォルト設定にすると若干コストの無駄が発生する為、チェックを外してカスタム設定を使用
セカンダリインデックス	テーブルのプライマリキー以外に使用するセカンダリインデックスの設定。プライマリキーに同じく切り出して後述します
読み込み/書き込みキャパシティモード	キャパシティユニットを自分で設定するか自動設定にするかの設定。自分で設定した方がコストの無駄を抑えられる為、今回はプロビジョニングを選択
プロビジョニングされたキャパシティ	データの読み書きに対して用意するキャパシティ量設定。今回の用途では定期実行で数分おきに一回書き込みが発生するだけなので、読み書きともに最小値の1を設定
Auto Scaling	プロビジョニングしたキャパシティに対しての使用量が一定割合を超えた場合のスケーリング設定。今回の用途では不要
保管時の暗号化	保管時の暗号化方式設定。今回はコストをかけない為、デフォルトのDynamoDBが所有するキーを使用
タグの追加	タグ名。管理用にNameタグは設定しておく事を推奨

上記設定項目の中でプライマリキーの設定（セカンダリインデックス含む）がとても大事な意味を持つ為、今回の用途でどう設計すべきかを少し書いてみます。

~~ 今回の用途に対してのDynamoDBのプライマリキー設計 ~~

まず始めに、今回の用途で保存するデータで必要最小限のものだけをDynamoDBに保存した場合にどんな形になるかを考えてみます。

次に、上記データに対する読み込みリクエストのユースケースを考えてみると、以下のような使われ方が想定されます。

• ○○のURLパスの測定結果の直近xx日分を取得したい（スコア推移観測）
• yyyy/MM/dd 日の測定結果を取得したい（URLパス毎のスコア比較）

これらに対応可能なキー設計を考えて、今回は以下のように設定しました。

プライマリキー	パーティションキー: URL_PATH (対象URL:String) ソートキー: TARGET_TIME (測定日時:Number)
セカンダリインデックス	パーティションキー: TARGET_DATE (測定日:Number) * 測定日時とは別に、日付のみのフィールドを作る

URLパスのパーティションキーと測定日時のソートキーの二つでプライマリキーにしているので、同一URLパスのレコードが全て同じパーティションになってしまいレコード数が増えてきた場合に適切にパーティション分割ができなくなるなどDynamoDBの内部構造的にはあまり良い設計ではないですが、今回の用途が1日1回測定ぐらいの実行間隔を想定しておりパーティション分割が必要になるぐらいのレコード数増加はあまり考慮する必要がなかった為、想定されるユースケースをシンプルに実現可能な今回のキー設計に決めました。
PageSpeed Insightsの結果以外のパフォーマンス関連のデータを全てまとめたテーブルにする使い方（PERFORMANCE_METRICS的なテーブルにして、プライマリキーに種別のPageSpeed Insightsを設定するなど）も考えられますが、想定される読み込みリクエストのユースケースに対応する為にどうしても無理なキー設計にせざるを得なくなるなど、個人的にはあまりお薦めできません。
一つのテーブルに多用途のデータが混在する構成は、管理が煩雑になってしまうのも避けた方が良い理由です。

事前準備3. IAMロール&ポリシー作成

最後に、Lambdaの実行ロールに設定するIAMロールと、ロールにアタッチするポリシーをそれぞれ作成します。
今回作成するLambdaからはDynamoDBとSystemsManager（パラメータストアからAPI Key取得）、CloudWatchLogsへのアクセスを行う為、それぞれへのアクセスを許可したIAMポリシーと、ポリシーをアタッチするIAMロールを作成します。
Lambdaからそれぞれのサービスにアクセス可能にするだけであれば実行ロールのインラインポリシーに必要な許可設定を書けば可能ですが、インラインポリシーが増えると各サービスに対してどのロール/ポリシーからアクセスが許可されているかの管理が行き届かなくなる為、可能な限り管理ポリシーで作成する事を推奨します。

1. DynamoDBへのアクセスを許可するIAMポリシー作成

IAMポリシーの作成はマネジメントコンソールから可能で、「IAM」 → 「ポリシー」 → 「ポリシーの作成」と進みます。


今回は、上記で作成したDynamoDBテーブルへの新規アイテム追加の許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:PutItem" ], "Resource": "arn:aws:dynamodb:{リージョン名}:{アカウントID}:table/{テーブル名}" } ] }

1 2 3 4 5 6 7 8 9 10 11 12

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "dynamodb:PutItem"             ],             "Resource": "arn:aws:dynamodb:{リージョン名}:{アカウントID}:table/{テーブル名}"         }     ] }

入力を終えたらポリシーの確認ボタンで先に進み、表示された項目を入力して、ポリシーの作成ボタン押下で完了です。

名前	作成する管理ポリシー名。 カスタム管理ポリシーは接頭語で識別可能にしておくと色々便利。
説明	作成する管理ポリシーの説明。 省略可能。
概要	作成する管理ポリシーに紐づいた設定内容。 意図した設定になっているかの確認用（この画面では変更不可）

2. SystemsManagerへのアクセスを許可するIAMポリシー作成

二つ目のポリシーとして、PageSpeed Insights API 実行用のAPI Key保存先のSSMのパラメータストアから値を取得する為のIAMポリシーを作成します。

IAMポリシーの新規作成手順は上記手順と全く同じの為割合しますが、今回はSSMのパラメータストアからの値取得と、取得した値に対するKMSの複合化の許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ssm:GetParameter" ], "Resource": "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{パラメータストアに作成したAPI Keyのキー名}" }, { "Effect": "Allow", "Action": [ "kms:Decrypt" ], "Resource": [ "arn:aws:kms:{リージョン名}:{アカウントID}:key/{KMSのキーID}" ] } ] }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "ssm:GetParameter"             ],             "Resource": "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{パラメータストアに作成したAPI Keyのキー名}"         },         {             "Effect": "Allow",             "Action": [                 "kms:Decrypt"             ],             "Resource": [                 "arn:aws:kms:{リージョン名}:{アカウントID}:key/{KMSのキーID}"             ]         }     ] }

入力を終えたらポリシーの確認ボタンで先に進み、上記に同じく名前と説明、概要を入力して、ポリシーの作成ボタン押下で完了です。

3. CloudWatchLogsへのアクセスを許可するIAMポリシー作成

三つ目のポリシーとして、Lambda functionの実行結果をCloudWatchLogsのログに保存する為のIAMポリシーを作成します。

上記に同じくIAMポリシーの新規作成手順は割合し、今回はロググループの新規作成と作成したロググループに対するログ保存の許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "logs:CreateLogGroup", "logs:CreateLogStream", "logs:PutLogEvents" ], "Resource": "arn:aws:logs:{リージョン名}:{アカウントID}:*" } ] }

1 2 3 4 5 6 7 8 9 10 11 12 13 14

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "logs:CreateLogGroup",                 "logs:CreateLogStream",                 "logs:PutLogEvents"             ],             "Resource": "arn:aws:logs:{リージョン名}:{アカウントID}:*"         }     ] }

同じく入力を終えたらポリシーの確認ボタンで先に進み、名前と説明、概要を入力して、ポリシーの作成ボタン押下で全ポリシーの作成が完了です。

4. Lambdaの実行ロールに設定するIAMロール作成

上記で作成した三つのIAMポリシーをアタッチする、実行ロールとして設定するIAMロールを作成します。

IAMロールの作成もマネジメントコンソールから可能で、「IAM」 → 「ロール」 → 「ロールの作成」と進みます。


信頼されたエンティティの種類をAWSサービス、ユースケースをLambdaと選択して次のステップへ進みます。


アタッチするポリシーの選択に移るので、上記で作成した三つのIAMポリシーを選択して次のステップへ進みます。


タグの追加に移るので、付けたいタグを追加して次のステップへ進みます。
管理用にNameタグを付けておく事を推奨


最後に、作成するロール名、ロールの説明を入力して、ロールの作成ボタン押下でIAMロールの作成が完了です。

以上で実行ロールに設定するIAMロールと、ロールにアタッチするポリシーの作成が完了です。

サイトパフォーマンス測定結果をDBに保存する処理の作成

事前準備が整ったところで、サイトパフォーマンス測定結果をDBに保存する処理の作成を開始していきます。

1. 最小項目を埋めた状態のLambda function作成

Lambda functionの作成は、マネジメントコンソールから「Lambda」 → 「関数の作成」と進みます。


Lambda function作成時の最小項目入力に移るので、各項目を埋めて関数の作成ボタン押下で作成完了です。

作成オプション	作成時にAWSが用意してくれた設計図などをテンプレートとして使用するかの選択。 今回は一から作成を選択。
関数名	作成するLambda function名。
ランタイム	実行時に使用するプラグラミング言語の選択。 何でも良いですが、今回はPython3.7を選択。
実行ロールの選択/作成	処理内から他サービスへのアクセスなどの許可/拒否設定を制御するIAMロールの選択。 今回は事前準備で作成しておいたIAMロールを使用するので、既存のロールを使用するを選択。
既存のロール	上記実行ロールを作成済のIAMロールの中から選択。 事前準備で作成しておいたIAMロールを選択。


少し小さくて見辛いのでクリックで拡大可能です。

2. 実行トリガーの設定

処理を実行させる為のトリガー条件の設定は、Designer欄のトリガーを追加ボタンを押下して進むトリガーの追加画面から行います。
今回は一定間隔で定期実行させる用件の為、以下のように入力していきます。

トリガーの種類	Lambda functionの発火条件の設定。 今回はCloudWatchEventsを選択。
ルール	CloudWatchEventsの実行条件ルールを、既存ルールを使用するか新規にルール作成をするかの選択。 共通のルールで複数イベントを設定したい場合以外は新規に個別のルールを作成する方が無難です。 一つのルールを複数イベントで使い回すと、ルール内容変更時に設定されている全てのイベントに影響が出てしまいます。
ルール名	新規にルール作成をする場合に設定するルール名。
ルールの説明	新規にルール作成をする場合に設定するルールの説明。 省略可能。
ルールタイプ	イベントパターンに基づいて実行させるかスケジュール実行させるかの選択。 今回は一定間隔で定期実行させる用件の為、スケジュール式を選択。
スケジュール式	ルールタイプにスケジュール式を選択した場合に設定するcron or rate式でのスケジュール指定。 今回はcron形式を選択して毎日AM0時に実行させる事にした為、以下のように記載。 cron(0 15 * * ? *) CloudWatchEventsで指定する時刻のタイムゾーンはUTCの為、日本時間（JST）で指定したい場合は9時間前の日時で指定する必要があります。
トリガーの有効化	今すぐ設定を有効化するか、テスト用に無効化した状態で保存するかの設定。 最初は無効化しておき、一通りできたタイミングで有効化する事を推奨。



3. 各種設定項目の変更

作成時に設定した最小項目以外にも幾つか変更する必要がある項目がある為、各種設定項目を変更していきます。
設定項目は非常に多いので、今回はデフォルト値から変更した項目のみ記載します。

環境変数	Lambda functionの処理で使用する環境変数の設定。 今回の処理では、以下を環境変数で用意。 SSM_REGION_NAME 使用するSystemsManagerのリージョン名。 東京リージョンならap-northeast-1。 PAGE_SPEED_INSIGHTS_API_KEY_ PARAMETER_NAME PageSpeed Insights API Keyのパラメータストアキー名。 DOMAIN_URL 測定対象サイトのドメインURL。 TARGET_URL_PATHS 測定対象のURLパス。 測定対象が複数存在する場合は、カンマで繋いで保存する。 DYNAMODB_REGION_NAME 使用するDynamoDBのリージョン名。 東京リージョンならap-northeast-1。 DYNAMODB_TARGET_ TABLE_NAME 測定結果を保存するDynamoDBのテーブル名。 DYNAMODB_PRIMARY_ PARTITION_KEY_NAME 上記テーブルのプライマリパーティションキー名。 今回のテーブルでは、URL_PATH。 DYNAMODB_PRIMARY_ PARTITION_KEY_DATA_TYPE 上記プライマリパーティションキーのデータタイプ。 今回のテーブルでは、String型なのでS。 DYNAMODB_PRIMARY_ SORT_KEY_NAME 上記テーブルのプライマリソートキー名。 今回のテーブルでは、TARGET_TIME。 DYNAMODB_PRIMARY_ SORT_KEY_DATA_TYPE 上記プライマリソートキーのデータタイプ。 今回のテーブルでは、Number型なのでN。 DYNAMODB_SECONDARY_ PARTITION_KEY_NAME 上記テーブルのセカンダリパーティションキー名。 今回のテーブルでは、TARGET_DATE。 DYNAMODB_SECONDARY_ PARTITION_KEY_DATA_TYPE 上記セカンダリパーティションキーのデータタイプ。 今回のテーブルでは、Number型なのでN。 DYNAMODB_DESKTOP_SCORE_ COLUMN_NAME 上記テーブルのデスクトップスコア保存先カラム名。 今回のテーブルでは、DESKTOP_SCORE。 DYNAMODB_DESKTOP_SCORE_ COLUMN_DATA_TYPE 上記デスクトップスコア保存先カラムのデータタイプ。 今回のテーブルでは、Number型なのでN。 DYNAMODB_MOBILE_SCORE_ COLUMN_NAME 上記テーブルのモバイルスコア保存先カラム名。 今回のテーブルでは、MOBILE_SCORE。 DYNAMODB_MOBILE_SCORE_ COLUMN_DATA_TYPE 上記モバイルスコア保存先カラムのデータタイプ。 今回のテーブルでは、Number型なのでN。
タグ	作成するLambda functionに紐付けるタグ。 管理用にNameタグは設定しておく事を推奨。
基本設定.タイムアウト	処理の実行時間上限設定。 デフォルトは3秒でPageSpeed Insights APIを実行する毎に10秒前後かかる為、変更が必要です。 測定対象URLの数次第で実行時間も変わってきますが、今回は上限時間5分で設定。 基本的考え方として、タイムアウト設定の上限を増やすのではなく、使用メモリを増やして実行時間を短縮させるのが望ましいです。 が、今回の処理で時間がかかっているのはAPIリクエストを投げてから結果が返ってくるまでの時間です。 よって、Lambda側の処理速度を速めても実行時間は短縮できない為、タイムアウト設定の上限を増やします。

4. 関数コードの編集

実行させる処理の内容を記述する為、関数コードを編集していきます。

lambda page speed insights api result to dynamodb

Python

import json import boto3 import logging import urllib.request import os from datetime import datetime, timedelta, timezone ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME = 'PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME' DOMAIN_URL = 'DOMAIN_URL' TARGET_URL_PATHS = 'TARGET_URL_PATHS' DYNAMODB_REGION_NAME = 'DYNAMODB_REGION_NAME' DYNAMODB_TARGET_TABLE_NAME = 'DYNAMODB_TARGET_TABLE_NAME' DYNAMODB_PRIMARY_PARTITION_KEY_NAME = 'DYNAMODB_PRIMARY_PARTITION_KEY_NAME' DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE = 'DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE' DYNAMODB_PRIMARY_SORT_KEY_NAME = 'DYNAMODB_PRIMARY_SORT_KEY_NAME' DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE = 'DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE' DYNAMODB_SECONDARY_PARTITION_KEY_NAME = 'DYNAMODB_SECONDARY_PARTITION_KEY_NAME' DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE = 'DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE' DYNAMODB_DESKTOP_SCORE_COLUMN_NAME = 'DYNAMODB_DESKTOP_SCORE_COLUMN_NAME' DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE' DYNAMODB_MOBILE_SCORE_COLUMN_NAME = 'DYNAMODB_MOBILE_SCORE_COLUMN_NAME' DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE' ## const value PAGE_SPEED_INSIGHTS_API_EXECUTE_URL_PREFIX = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed' ## singleton object logger = None ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class SsmParameterStore: def __init__(self, region_name): """ Parameters ---------- region_name : str SystemsManager region name. """ self.ssm = boto3.client( 'ssm', region_name = region_name ) def get_parameter(self, parameter_name, with_decryption): """ Get ParameterStore value. Parameters ---------- parameter_name : str target parameter name. with_decryption : bool need decryption. Returns ------- parameter_store_value : str target parameter store value. """ response = self.ssm.get_parameter( Name = parameter_name, WithDecryption = with_decryption ) return response['Parameter']['Value'] ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class DynamoDB: def __init__(self, region_name): """ Parameters ---------- region_name : str DynamoDB region name. """ self.dynamodb = boto3.client( 'dynamodb', region_name = region_name ) def put_item(self, table_name, item): """ DynamoDB PutItem operation. Parameters ---------- table_name : str target table name. item : dict dynamodb record. ext. { "foo": {"S": "foo value"}, "bar": {"N": "bar value"} } Returns ------- response_syntax : dict Represents the output of a PutItem operation. """ return self.dynamodb.put_item( TableName = table_name, Item = item ) def initialize_logger_setting(): """ Initialize logger setting. """ global logger if not logger: logger = logging.getLogger() logger.setLevel(logging.INFO) def get_page_speed_insights_api_key(): """ Get PageSpeed Insights API Key. Returns ------- api_key : str api key value. """ ssm = SsmParameterStore(os.environ[SSM_REGION_NAME]) return ssm.get_parameter( os.environ[PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME], True ) def generate_get_url(url_path, strategy, domain_url, api_key): """ Generate get url for PageSpeed Insights API. Parameters ---------- url_path : str target url path. strategy : str target device name(desktop or mobile). default value is desktop. domain_url : str target domain url. api_key : str PageSpeed Insights API Key. Returns ------- get_url : str get url for PageSpeed Insights API. """ target_url = '' if url_path.startswith('/') == False: target_url = domain_url + '/' + url_path else: target_url = domain_url + url_path qs = { 'url': target_url, 'key': api_key, 'category': 'performance', 'strategy': strategy } return '{}?{}'.format( PAGE_SPEED_INSIGHTS_API_EXECUTE_URL_PREFIX, urllib.parse.urlencode(qs) ) def execute_page_speed_insights_api(url_path, strategy, domain_url, api_key): """ Execute PageSpeed Insights API. Parameters ---------- url_path : str target url path. strategy : str target device name(desktop or mobile). default value is desktop. domain_url : str target domain url. api_key : str PageSpeed Insights API Key. Returns ------- result_score : float performance score value. """ get_url = generate_get_url( url_path, strategy, domain_url, api_key ) with urllib.request.urlopen(get_url) as result: return json.load(result)['lighthouseResult']['categories']['performance']['score'] def create_put_item(url_path, desktop_score, mobile_score): """ Create put item for DynamoDB. Parameters ---------- url_path : str target url path. desktop_score : float desktop score value. mobile_score : float mobile score value. Returns ------- put_item : dict dynamodb record. ext. { "foo": {"S": "foo value"}, "bar": {"N": "bar value"} } """ utc_datetime_now = datetime.now() jst_datetime_now = utc_datetime_now.astimezone(timezone(timedelta(hours=+9), 'JST')) return { os.environ[DYNAMODB_PRIMARY_PARTITION_KEY_NAME]: {os.environ[DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE]: url_path}, os.environ[DYNAMODB_PRIMARY_SORT_KEY_NAME]: {os.environ[DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE]: utc_datetime_now.strftime('%s')}, os.environ[DYNAMODB_SECONDARY_PARTITION_KEY_NAME]: {os.environ[DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE]: jst_datetime_now.strftime('%Y%m%d')}, os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_NAME]: {os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE]: str(desktop_score * 100)}, os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_NAME]: {os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE]: str(mobile_score * 100)} } def save_performance_score(dynamodb, url_path, desktop_score, mobile_score): """ Save performance score to DynamoDB. Parameters ---------- dynamodb : DynamoDB DynamoDB object. url_path : str target url path. desktop_score : float desktop score value. mobile_score : float mobile score value. """ dynamodb.put_item( os.environ[DYNAMODB_TARGET_TABLE_NAME], create_put_item( url_path, desktop_score, mobile_score ) ) def lambda_handler(event, context): initialize_logger_setting() api_key = get_page_speed_insights_api_key() domain_url = os.environ[DOMAIN_URL] dynamodb = DynamoDB(os.environ[DYNAMODB_REGION_NAME]) for url_path in os.environ[TARGET_URL_PATHS].split(','): desktop_score = execute_page_speed_insights_api( url_path, 'desktop', domain_url, api_key ) logger.info('url: ' + url_path + ', desktop score is = ' + str(desktop_score)) mobile_score = execute_page_speed_insights_api( url_path, 'mobile', domain_url, api_key ) logger.info('url: ' + url_path + ', mobile score is = ' + str(mobile_score)) save_performance_score( dynamodb, url_path, desktop_score, mobile_score ) logger.info('url: ' + url_path + ', save performance score success.')

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288

import json import boto3 import logging import urllib.request import os from datetime import datetime, timedelta, timezone   ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME = 'PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME' DOMAIN_URL = 'DOMAIN_URL' TARGET_URL_PATHS = 'TARGET_URL_PATHS' DYNAMODB_REGION_NAME = 'DYNAMODB_REGION_NAME' DYNAMODB_TARGET_TABLE_NAME = 'DYNAMODB_TARGET_TABLE_NAME' DYNAMODB_PRIMARY_PARTITION_KEY_NAME = 'DYNAMODB_PRIMARY_PARTITION_KEY_NAME' DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE = 'DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE' DYNAMODB_PRIMARY_SORT_KEY_NAME = 'DYNAMODB_PRIMARY_SORT_KEY_NAME' DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE = 'DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE' DYNAMODB_SECONDARY_PARTITION_KEY_NAME = 'DYNAMODB_SECONDARY_PARTITION_KEY_NAME' DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE = 'DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE' DYNAMODB_DESKTOP_SCORE_COLUMN_NAME = 'DYNAMODB_DESKTOP_SCORE_COLUMN_NAME' DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE' DYNAMODB_MOBILE_SCORE_COLUMN_NAME = 'DYNAMODB_MOBILE_SCORE_COLUMN_NAME' DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE'   ## const value PAGE_SPEED_INSIGHTS_API_EXECUTE_URL_PREFIX = 'https://www.googleapis.com/pagespeedonline/v5/runPagespeed'   ## singleton object logger = None     ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class SsmParameterStore:     def __init__(self, region_name):         """         Parameters         ----------         region_name : str             SystemsManager region name.         """         self.ssm = boto3.client(             'ssm',             region_name = region_name         )       def get_parameter(self, parameter_name, with_decryption):         """         Get ParameterStore value.           Parameters         ----------         parameter_name : str             target parameter name.         with_decryption : bool             need decryption.           Returns         -------         parameter_store_value : str             target parameter store value.         """         response = self.ssm.get_parameter(             Name = parameter_name,             WithDecryption = with_decryption         )         return response['Parameter']['Value']     ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class DynamoDB:     def __init__(self, region_name):         """         Parameters         ----------         region_name : str             DynamoDB region name.         """         self.dynamodb = boto3.client(             'dynamodb',             region_name = region_name         )       def put_item(self, table_name, item):         """         DynamoDB PutItem operation.           Parameters         ----------         table_name : str             target table name.         item : dict             dynamodb record.             ext. {                 "foo": {"S": "foo value"},                 "bar": {"N": "bar value"}             }           Returns         -------         response_syntax : dict             Represents the output of a PutItem operation.         """         return self.dynamodb.put_item(             TableName = table_name,             Item = item         )     def initialize_logger_setting():     """     Initialize logger setting.     """     global logger     if not logger:         logger = logging.getLogger()         logger.setLevel(logging.INFO)   def get_page_speed_insights_api_key():     """     Get PageSpeed Insights API Key.       Returns     -------     api_key : str         api key value.     """     ssm = SsmParameterStore(os.environ[SSM_REGION_NAME])     return ssm.get_parameter(         os.environ[PAGE_SPEED_INSIGHTS_API_KEY_PARAMETER_NAME],         True     )   def generate_get_url(url_path, strategy, domain_url, api_key):     """     Generate get url for PageSpeed Insights API.       Parameters     ----------     url_path : str         target url path.     strategy : str         target device name(desktop or mobile).         default value is desktop.     domain_url : str         target domain url.     api_key : str         PageSpeed Insights API Key.       Returns     -------     get_url : str         get url for PageSpeed Insights API.     """     target_url = ''     if url_path.startswith('/') == False:         target_url = domain_url + '/' + url_path     else:         target_url = domain_url + url_path       qs = {         'url': target_url,         'key': api_key,         'category': 'performance',         'strategy': strategy     }     return '{}?{}'.format(         PAGE_SPEED_INSIGHTS_API_EXECUTE_URL_PREFIX,         urllib.parse.urlencode(qs)     )   def execute_page_speed_insights_api(url_path, strategy, domain_url, api_key):     """     Execute PageSpeed Insights API.       Parameters     ----------     url_path : str         target url path.     strategy : str         target device name(desktop or mobile).         default value is desktop.     domain_url : str         target domain url.     api_key : str         PageSpeed Insights API Key.       Returns     -------     result_score : float         performance score value.     """     get_url = generate_get_url(         url_path,         strategy,         domain_url,         api_key     )     with urllib.request.urlopen(get_url) as result:         return json.load(result)['lighthouseResult']['categories']['performance']['score']   def create_put_item(url_path, desktop_score, mobile_score):     """     Create put item for DynamoDB.       Parameters     ----------     url_path : str         target url path.     desktop_score : float         desktop score value.     mobile_score : float         mobile score value.       Returns     -------     put_item : dict         dynamodb record.         ext. {             "foo": {"S": "foo value"},             "bar": {"N": "bar value"}         }     """     utc_datetime_now = datetime.now()     jst_datetime_now = utc_datetime_now.astimezone(timezone(timedelta(hours=+9), 'JST'))     return {         os.environ[DYNAMODB_PRIMARY_PARTITION_KEY_NAME]: {os.environ[DYNAMODB_PRIMARY_PARTITION_KEY_DATA_TYPE]: url_path},         os.environ[DYNAMODB_PRIMARY_SORT_KEY_NAME]: {os.environ[DYNAMODB_PRIMARY_SORT_KEY_DATA_TYPE]: utc_datetime_now.strftime('%s')},         os.environ[DYNAMODB_SECONDARY_PARTITION_KEY_NAME]: {os.environ[DYNAMODB_SECONDARY_PARTITION_KEY_DATA_TYPE]: jst_datetime_now.strftime('%Y%m%d')},         os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_NAME]: {os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE]: str(desktop_score * 100)},         os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_NAME]: {os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE]: str(mobile_score * 100)}     }   def save_performance_score(dynamodb, url_path, desktop_score, mobile_score):     """     Save performance score to DynamoDB.       Parameters     ----------     dynamodb : DynamoDB         DynamoDB object.     url_path : str         target url path.     desktop_score : float         desktop score value.     mobile_score : float         mobile score value.     """     dynamodb.put_item(         os.environ[DYNAMODB_TARGET_TABLE_NAME],         create_put_item(             url_path,             desktop_score,             mobile_score         )     )     def lambda_handler(event, context):     initialize_logger_setting()       api_key = get_page_speed_insights_api_key()     domain_url = os.environ[DOMAIN_URL]     dynamodb = DynamoDB(os.environ[DYNAMODB_REGION_NAME])       for url_path in os.environ[TARGET_URL_PATHS].split(','):         desktop_score = execute_page_speed_insights_api(             url_path,             'desktop',             domain_url,             api_key         )         logger.info('url: ' + url_path + ', desktop score is = ' + str(desktop_score))         mobile_score = execute_page_speed_insights_api(             url_path,             'mobile',             domain_url,             api_key         )         logger.info('url: ' + url_path + ', mobile score is = ' + str(mobile_score))           save_performance_score(             dynamodb,             url_path,             desktop_score,             mobile_score         )         logger.info('url: ' + url_path + ', save performance score success.')

詳細説明は省きますが、パラメータストアからPageSpeed Insights APIのAPI Keyを取得し、測定対象URLパス毎にデスクトップとモバイルそれぞれのパフォーマンス測定を実施、結果をDynamoDBに保存するのが大まかな流れです。

以上で、サイトパフォーマンス測定結果をDBに保存する処理は完成です。

2. DBから結果を取り出して通知用サービスに転送

続いて、保存結果をDBから取り出して通知用サービスに転送する処理の作成です。
手順1と同じく、Lambdaを作成する前の事前準備として以下3つを先に行います。

事前準備1. DynamoDBへの項目追加をイベント検知可能に変更

今回作成する処理は、DynamoDBに新規項目が追加されたタイミングで通知用サービスに転送する処理を開始させる必要がある為、まずはDynamoDBへの項目追加をイベントトリガーとして使用可能に変更します。
DynamoDBにはDynamoDB Streamsという項目の追加/更新/削除の履歴を保存する場所が用意されており、また、Streamsへの保存がLambdaのイベントトリガーとして設定可能になっている為、今回はこれを処理のトリガーとして利用します。
Streamsに保存するかどうかの設定はテーブル毎に行う為、今回作成したテーブルのStreamsへのイベント保存を有効化します。

DynamoDB Streamsの有効化はマネジメントコンソールから可能で、「DynamoDB」 → 「テーブル」 → 「一覧から対象テーブル選択」 → 「ストリームの管理」と進みます。

ストリームの管理ダイアログが出現するので、新しいイメージにチェックを付けて有効化ボタン押下で完了です。
項目の更新/削除時の変更前データも保存する場合は古いイメージも必要になりますが、今回は新規追加時の追加されたデータがあれば用件を満たせる為、新しいイメージだけで十分です。

以上で、DynamoDBへの項目追加をイベント検知可能に変更する準備は完了です。

事前準備2. SNSのトピック作成

続いて、転送する通知用サービスとして使用するSNSのトピックを作成します。

SNSトピックの作成はマネジメントコンソールから可能で、「Simple Notification Service」 → 「トピック」 → 「トピックの作成」と進みます。

表示された項目をそれぞれ入力していき、トピックの作成ボタン押下で完了です。

名前	作成するトピック名
表示名	受信側に表示させる送信者名。 今回は、PageSpeed Insights Resultを設定。 SMSメッセージは、先頭10文字しか表示されません。
暗号化	暗号化の有効化/無効化設定。 今回は、有効化した上でデフォルトのKMSマスターキーを使用。
アクセスポリシー メソッド選択	トピックのリソースポリシーのメソッド選択。 今回は複雑なポリシーは作らない為、基本を選択。
アクセスポリシー トピックへのメッセージ送信可能者	トピックにメッセージ送信可能なアカウント設定。 今回は、トピックの所有者のみで設定。
アクセスポリシー トピックへのサブスクライブ	トピックにサブスクライブ可能なアカウント設定。 今回は、トピックの所有者のみで設定。
配信再試行ポリシー	SNSの送信失敗時の再試行ポリシー設定。 今回は、デフォルトの配信再試行ポリシーを使用。
配信ステータスのログ記録	SMSからの配信記録をログ保存するかの設定。 SMSトピックから受信者へメッセージが届かない場合のデバッグ用途等で使うが、今回は不特定多数にpublishする使い方ではないのでログ記録は行わないで設定。
タグ	作成するトピックに紐付けるタグ。 管理用にNameタグは設定しておく事を推奨。

今回はパフォーマンス低下が発生していた場合のみ通知するのではなく、判定結果によって通知先のトピックを分けてそれぞれの通知を行う方法を取る為、判定結果ok用のトピックとng用のトピックで合計二つのトピックを作成します。

事前準備3. IAMロール&ポリシー作成

最後に、Lambdaの実行ロールに設定するIAMロールと、ロールにアタッチするポリシーをそれぞれ作成します。
今回作成するLambdaからはDynamoDB StreamsとSNS、CloudWatchLogsへのアクセスを行う為、それぞれへのアクセスを許可したIAMポリシーと、ポリシーをアタッチするIAMロールを作成します。
CloudWatchLogsへのアクセスを可能にするポリシーは前回作成したポリシーを流用できる為、新規作成はせずに今回作成するIAMロールへ前回作成したポリシーをアタッチします。

1. DynamoDB Streamsへのアクセスを許可するIAMポリシー作成

処理を開始するトリガーのDynamoDB Streamsから新規追加された項目を取り出す為のIAMポリシーを作成します。
通常、発火条件のサービスに対するアクセス許可はリソースポリシーで設定しますが、各種Streamsへの項目追加などLambda側がポーリングしてイベントを検知するタイプは、実行ロールへアクセス許可ポリシーを設定します。

今回はDynamoDB Streamsからの項目取り出しの許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:GetRecords", "dynamodb:GetShardIterator", "dynamodb:DescribeStream", "dynamodb:ListStreams" ], "Resource": "arn:aws:dynamodb:{リージョン名}:{アカウントID}:table/{テーブル名}/stream/*" } ] }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "dynamodb:GetRecords",                 "dynamodb:GetShardIterator",                 "dynamodb:DescribeStream",                 "dynamodb:ListStreams"             ],             "Resource": "arn:aws:dynamodb:{リージョン名}:{アカウントID}:table/{テーブル名}/stream/*"         }     ] }

2. SNSへのアクセスを許可するIAMポリシー作成

二つ目のポリシーとして、DynamoDB Streamsから取り出した値の測定結果を通知用サービスのSNSへ転送する為のIAMポリシーを作成します。

今回は二つのSNSトピックに対してのpublishの許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "sns:Publish" ], "Resource": [ "arn:aws:sns:{リージョン名}:{アカウントID}:{判定結果ok用トピック名}", "arn:aws:sns:{リージョン名}:{アカウントID}:{判定結果ng用トピック名}" ] } ] }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "sns:Publish"             ],             "Resource": [                 "arn:aws:sns:{リージョン名}:{アカウントID}:{判定結果ok用トピック名}",                 "arn:aws:sns:{リージョン名}:{アカウントID}:{判定結果ng用トピック名}"             ]         }     ] }

3. Lambdaの実行ロールに設定するIAMロール作成

上記で作成した二つのIAMポリシーと、前回作成したCloudWatchLogsへのアクセス許可を設定したポリシーの合計三つをアタッチする、実行ロールとして設定するIAMロールを作成します。

以上で今回のIAMロール/ポリシーの作成は完了です。

測定結果を通知用サービスに転送する処理の作成

事前準備が整ったところで、測定結果をDBから取り出して通知用サービスに転送する処理の作成を開始していきます。

1. 最小項目を埋めた状態のLambda function作成

Lambda functionの作成に入り、今回はDynamoDB Streamsへの項目追加を実行トリガーに使用するので、前回のLambda function作成時とは異なり予め用意されている各イベントソース用の設計図を使用して作成を行います。

用意されている各イベントソース用の設計図を使用するには、

1. Lambda function作成時の最初の設定画面で設計図を使用を選択
2. キーワード入力欄にdynamodbを入力してフィルタ
3. dynamodb-process-stream-pythonを選択
4. 設定ボタン押下

と選択して進みます。


最小項目入力に移るので、各項目を埋めて関数の作成ボタン押下で完了です。

関数名	作成するLambda function名。
実行ロールの選択/作成	処理内から他サービスへのアクセスなどの許可/拒否設定を制御するIAMロールの選択。 今回は事前準備で作成しておいたIAMロールを使用するので、既存のロールを使用するを選択。
既存のロール	上記実行ロールを作成済のIAMロールの中から選択。 事前準備で作成しておいたIAMロールを選択。
DynamoDBテーブル	対象のDynamoDBテーブル名。 今回Streamsを有効化したテーブルを選択。
バッチサイズ	Streamsから一度に読み取る最大レコード数。 デフォルト設定の100を使用。
バッチウィンドウオプション	関数を呼び出すまでにレコードを収集する最大時間。 オプション設定なので今回は未使用。
開始位置	Lambda有効化時の、Streamsからの読み取り開始位置。 今回はLambda有効化後のDynamoDB変更分だけ取得できれば良い為、最新を選択。

2. 各種設定項目の変更

作成時に設定した最小項目以外にも幾つか変更する必要がある項目がある為、各種設定項目を変更していきます。
設定項目は非常に多いので、今回はデフォルト値から変更した項目のみ記載します。

環境変数	Lambda functionの処理で使用する環境変数の設定。 今回の処理では、以下を環境変数で用意。 SNS_REGION_NAME 使用するSNSのリージョン名。 東京リージョンならap-northeast-1。 SNS_ALERT_TOPIC_ARN スコア判定結果がng時のアラート通知用SNSトピック名。 SNS_NOTIFICATION_TOPIC_ARN スコア判定結果がok時の通常通知用SNSトピック名。 ALERT_BORDER_SCORE スコアが正常か異常かの判定に使用する境界値。 測定対象サイトの状況により境界値は変動する（目標スコアが変わる）為、環境変数で持たせます。 今回は、デスクトップ or モバイルの片方でも基準値未満だった場合はアラート用トピックに、両方基準値以上だった場合は通常通知用トピックに送ります。 DYNAMODB_URL_PATH_ COLUMN_NAME DynamoDBテーブルのターゲットURLパス保存先カラム名。 今回のテーブルでは、URL_PATH。 DYNAMODB_URL_PATH_ COLUMN_DATA_TYPE 上記ターゲットURLパス保存先カラムのデータタイプ。 今回のテーブルでは、String型なのでS。 DYNAMODB_TARGET_TIME_ COLUMN_NAME DynamoDBテーブルのターゲット日時保存先カラム名。 今回のテーブルでは、TARGET_TIME。 DYNAMODB_TARGET_TIME_ COLUMN_DATA_TYPE 上記ターゲット日時保存先カラムのデータタイプ。 今回のテーブルでは、Number型なのでN。 DYNAMODB_DESKTOP_SCORE_ COLUMN_NAME DynamoDBテーブルのデスクトップスコア保存先カラム名。 今回のテーブルでは、DESKTOP_SCORE。 DYNAMODB_DESKTOP_SCORE_ COLUMN_DATA_TYPE 上記デスクトップスコア保存先カラムのデータタイプ。 今回のテーブルでは、Number型なのでN。 DYNAMODB_MOBILE_SCORE_ COLUMN_NAME DynamoDBテーブルのモバイルスコア保存先カラム名。 今回のテーブルでは、MOBILE_SCORE。 DYNAMODB_MOBILE_SCORE_ COLUMN_DATA_TYPE 上記モバイルスコア保存先カラムのデータタイプ。 今回のテーブルでは、Number型なのでN。
タグ	作成するLambda functionに紐付けるタグ。 管理用にNameタグは設定しておく事を推奨。
基本設定.タイムアウト	処理の実行時間上限設定。 デフォルトは3秒で短過ぎる為、少し伸ばしておきます。 今回は、SNSとの通信に時間がかかった場合でもタイムアウトしないように10秒で設定。

3. 関数コードの編集

実行させる処理の内容を記述する為、関数コードを編集していきます。

lambda performance score dynamodb to sns

Python

import boto3 import logging import os from datetime import datetime, timedelta, timezone ## environment variable key name SNS_REGION_NAME = 'SNS_REGION_NAME' SNS_ALERT_TOPIC_ARN = 'SNS_ALERT_TOPIC_ARN' SNS_NOTIFICATION_TOPIC_ARN = 'SNS_NOTIFICATION_TOPIC_ARN' ALERT_BORDER_SCORE = 'ALERT_BORDER_SCORE' DYNAMODB_URL_PATH_COLUMN_NAME = 'DYNAMODB_URL_PATH_COLUMN_NAME' DYNAMODB_URL_PATH_COLUMN_DATA_TYPE = 'DYNAMODB_URL_PATH_COLUMN_DATA_TYPE' DYNAMODB_TARGET_TIME_COLUMN_NAME = 'DYNAMODB_TARGET_TIME_COLUMN_NAME' DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE = 'DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE' DYNAMODB_DESKTOP_SCORE_COLUMN_NAME = 'DYNAMODB_DESKTOP_SCORE_COLUMN_NAME' DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE' DYNAMODB_MOBILE_SCORE_COLUMN_NAME = 'DYNAMODB_MOBILE_SCORE_COLUMN_NAME' DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE' ## singleton object logger = None ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class SimpleNotificationService: def __init__(self, region_name): """ Parameters ---------- region_name : str SimpleNotificationService region name. """ self.sns = boto3.client( 'sns', region_name = region_name ) def publish(self, topic_arn, message, subject): """ SimpleNotificationService Publish action. Parameters ---------- topic_arn : str target topic arn. message : str send message detail. subject : str send message subject. Returns ------- response_syntax : dict Response for Publish action. """ return self.sns.publish( TopicArn = topic_arn, Message = message, Subject = subject ) def initialize_logger_setting(): """ Initialize logger setting. """ global logger if not logger: logger = logging.getLogger() logger.setLevel(logging.INFO) def performance_score_degradation(desktop_score, mobile_score): """ Check performance score degradation. Parameters ---------- desktop_score : int performance score of desktop. mobile_score : int performance score of mobile. Returns ------- performance score degradation : bool If true, performance score degradation. """ alert_border_score = int(os.environ[ALERT_BORDER_SCORE]) return desktop_score < alert_border_score or mobile_score < alert_border_score def send_alert_message(sns, url_path, target_time, desktop_score, mobile_score): """ Send performance score alert message to Simple Notification Service. Parameters ---------- sns : SimpleNotificationService SimpleNotificationService object. url_path : str target url path. target_time : datetime target datetime of jst. desktop_score : int performance score of desktop. mobile_score : int performance score of mobile. Returns ------- notify response : dict Response from SimpleNotificationService publish action. """ message = 'URL: ' + url_path + '\nDesktop score: ' + str(desktop_score) + '\nMobile score: ' + str(mobile_score) subject = '[Urgent] Performance score alert. (' + target_time.strftime('%Y/%m/%d %H:%M:%S') + ')' return sns.publish( os.environ[SNS_ALERT_TOPIC_ARN], message, subject ) def send_notification_message(sns, url_path, target_time, desktop_score, mobile_score): """ Send performance score notification message to Simple Notification Service. Parameters ---------- sns : SimpleNotificationService SimpleNotificationService object. url_path : str target url path. target_time : datetime target datetime of jst. desktop_score : int performance score of desktop. mobile_score : int performance score of mobile. Returns ------- notify response : dict Response from SimpleNotificationService publish action. """ message = 'URL: ' + url_path + '\nDesktop score: ' + str(desktop_score) + '\nMobile score: ' + str(mobile_score) subject = 'Performance score notification. (' + target_time.strftime('%Y/%m/%d %H:%M:%S') + ')' return sns.publish( os.environ[SNS_NOTIFICATION_TOPIC_ARN], message, subject ) def notify_performance_score(sns, dynamodb_record): """ Notify performance score. Parameters ---------- sns : SimpleNotificationService SimpleNotificationService object. dynamodb_record : dict dynamodb record. ext. { "foo": {"S": "foo value"}, "bar": {"N": "bar value"} } Returns ------- notify response : dict Response from SimpleNotificationService publish action. """ url_path = dynamodb_record[os.environ[DYNAMODB_URL_PATH_COLUMN_NAME]][os.environ[DYNAMODB_URL_PATH_COLUMN_DATA_TYPE]] target_time = datetime.fromtimestamp( int(dynamodb_record[os.environ[DYNAMODB_TARGET_TIME_COLUMN_NAME]][os.environ[DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE]]), timezone(timedelta(hours=+9), 'JST') ) desktop_score = int(dynamodb_record[os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_NAME]][os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE]]) mobile_score = int(dynamodb_record[os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_NAME]][os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE]]) if performance_score_degradation(desktop_score, mobile_score): return send_alert_message( sns, url_path, target_time, desktop_score, mobile_score ) else: return send_notification_message( sns, url_path, target_time, desktop_score, mobile_score ) def lambda_handler(event, context): initialize_logger_setting() sns = SimpleNotificationService(os.environ[SNS_REGION_NAME]) for record in event['Records']: response = notify_performance_score( sns, record['dynamodb']['NewImage'] ) logger.info('notify_performance_score successed. response = ' + str(response))

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207

import boto3 import logging import os from datetime import datetime, timedelta, timezone   ## environment variable key name SNS_REGION_NAME = 'SNS_REGION_NAME' SNS_ALERT_TOPIC_ARN = 'SNS_ALERT_TOPIC_ARN' SNS_NOTIFICATION_TOPIC_ARN = 'SNS_NOTIFICATION_TOPIC_ARN' ALERT_BORDER_SCORE = 'ALERT_BORDER_SCORE'   DYNAMODB_URL_PATH_COLUMN_NAME = 'DYNAMODB_URL_PATH_COLUMN_NAME' DYNAMODB_URL_PATH_COLUMN_DATA_TYPE = 'DYNAMODB_URL_PATH_COLUMN_DATA_TYPE' DYNAMODB_TARGET_TIME_COLUMN_NAME = 'DYNAMODB_TARGET_TIME_COLUMN_NAME' DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE = 'DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE' DYNAMODB_DESKTOP_SCORE_COLUMN_NAME = 'DYNAMODB_DESKTOP_SCORE_COLUMN_NAME' DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE' DYNAMODB_MOBILE_SCORE_COLUMN_NAME = 'DYNAMODB_MOBILE_SCORE_COLUMN_NAME' DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE = 'DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE'   ## singleton object logger = None     ## todo 記事が長くなり過ぎるのと今回の本題ではない為Lambda function内に入れたが、汎用的な処理はLayersに移す class SimpleNotificationService:     def __init__(self, region_name):         """         Parameters         ----------         region_name : str             SimpleNotificationService region name.         """         self.sns = boto3.client(             'sns',             region_name = region_name         )       def publish(self, topic_arn, message, subject):         """         SimpleNotificationService Publish action.           Parameters         ----------         topic_arn : str             target topic arn.         message : str             send message detail.         subject : str             send message subject.           Returns         -------         response_syntax : dict             Response for Publish action.         """         return self.sns.publish(             TopicArn = topic_arn,             Message = message,             Subject = subject         )     def initialize_logger_setting():     """     Initialize logger setting.     """     global logger     if not logger:         logger = logging.getLogger()         logger.setLevel(logging.INFO)   def performance_score_degradation(desktop_score, mobile_score):     """     Check performance score degradation.       Parameters     ----------     desktop_score : int         performance score of desktop.     mobile_score : int         performance score of mobile.       Returns     -------     performance score degradation : bool         If true, performance score degradation.     """     alert_border_score = int(os.environ[ALERT_BORDER_SCORE])     return desktop_score < alert_border_score or mobile_score < alert_border_score   def send_alert_message(sns, url_path, target_time, desktop_score, mobile_score):     """     Send performance score alert message to Simple Notification Service.       Parameters     ----------     sns : SimpleNotificationService         SimpleNotificationService object.     url_path : str         target url path.     target_time : datetime         target datetime of jst.     desktop_score : int         performance score of desktop.     mobile_score : int         performance score of mobile.       Returns     -------     notify response : dict         Response from SimpleNotificationService publish action.     """     message = 'URL: ' + url_path + '\nDesktop score: ' + str(desktop_score) + '\nMobile score: ' + str(mobile_score)     subject = '[Urgent] Performance score alert. (' + target_time.strftime('%Y/%m/%d %H:%M:%S') + ')'     return sns.publish(         os.environ[SNS_ALERT_TOPIC_ARN],         message,         subject     )   def send_notification_message(sns, url_path, target_time, desktop_score, mobile_score):     """     Send performance score notification message to Simple Notification Service.       Parameters     ----------     sns : SimpleNotificationService         SimpleNotificationService object.     url_path : str         target url path.     target_time : datetime         target datetime of jst.     desktop_score : int         performance score of desktop.     mobile_score : int         performance score of mobile.       Returns     -------     notify response : dict         Response from SimpleNotificationService publish action.     """     message = 'URL: ' + url_path + '\nDesktop score: ' + str(desktop_score) + '\nMobile score: ' + str(mobile_score)     subject = 'Performance score notification. (' + target_time.strftime('%Y/%m/%d %H:%M:%S') + ')'     return sns.publish(         os.environ[SNS_NOTIFICATION_TOPIC_ARN],         message,         subject     )   def notify_performance_score(sns, dynamodb_record):     """     Notify performance score.       Parameters     ----------     sns : SimpleNotificationService         SimpleNotificationService object.     dynamodb_record : dict         dynamodb record.         ext. {             "foo": {"S": "foo value"},             "bar": {"N": "bar value"}         }       Returns     -------     notify response : dict         Response from SimpleNotificationService publish action.     """     url_path = dynamodb_record[os.environ[DYNAMODB_URL_PATH_COLUMN_NAME]][os.environ[DYNAMODB_URL_PATH_COLUMN_DATA_TYPE]]     target_time = datetime.fromtimestamp(         int(dynamodb_record[os.environ[DYNAMODB_TARGET_TIME_COLUMN_NAME]][os.environ[DYNAMODB_TARGET_TIME_COLUMN_DATA_TYPE]]),         timezone(timedelta(hours=+9), 'JST')     )     desktop_score = int(dynamodb_record[os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_NAME]][os.environ[DYNAMODB_DESKTOP_SCORE_COLUMN_DATA_TYPE]])     mobile_score = int(dynamodb_record[os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_NAME]][os.environ[DYNAMODB_MOBILE_SCORE_COLUMN_DATA_TYPE]])       if performance_score_degradation(desktop_score, mobile_score):         return send_alert_message(             sns,             url_path,             target_time,             desktop_score,             mobile_score         )     else:         return send_notification_message(             sns,             url_path,             target_time,             desktop_score,             mobile_score         )     def lambda_handler(event, context):     initialize_logger_setting()          sns = SimpleNotificationService(os.environ[SNS_REGION_NAME])     for record in event['Records']:         response = notify_performance_score(             sns,             record['dynamodb']['NewImage']         )         logger.info('notify_performance_score successed. response = ' + str(response))

詳細説明は省きますが、DynamoDBからパフォーマンス測定結果を取得し、スコアが正常か異常かの判定結果をSNSに送るのが大まかな流れです。

以上で、保存結果をDBから取り出して通知用サービスに転送する処理は完成です。

3. 転送された通知メッセージをユーザーに送信

SNSのトピックに届いた通知メッセージをユーザに送信する最終処理の作成です。
ここでも、Lambdaを作成する前の事前準備を2つ先に行います。

事前準備1. 通知先SlackチャンネルのIncoming WebHooks用意

今回は通知先サービスにSlackを使う為、まずはSlack側の用意をしていきます。
外部からSlackに通知を送信する為には、Slack側で用意された機能のIncoming WebHooksを使用して実現します。

1. 通知送信先のワークスペース & チャンネル指定

Incoming WebHooksを使用するには、以下のURLから設定を開始します。
https://slack.com/services/new/incoming-webhook

通知先のワークスペース選択画面に移るので、対象ワークスペースのワークスペース名を入力します。
Slackアカウント作成やワークスペース参加がまだの場合は先に済ませる必要がありますが、長くなるのと今回の本題ではない為割合し、アカウント作成/ワークスペースへの参加済の状態を前提に進めます。


ログイン情報の入力画面に移るので、対象ワークスペースに参加済アカウントのメールアドレスとパスワードを入れてサインインします。


通知先のチャンネル選択画面に移るので、対象チャンネルのチャンネル名を選択します。


選択後、Add Incoming WebHooks integrationボタン押下で対象チャンネルにIncoming WebHooksが使用可能になります。

2. Integration Settings

通知を送信する際の表示名やアイコンなどを設定していきます。

Post to Channel	通知送信先のチャンネル名。 前の画面で選択したチャンネル名が選択されているのでそのままにします。
Webhook URL	Webhook用のURL。 通知を送信する際に使用するのでコピーしてメモしておきます。
Descriptive Label	一覧で表示されるメモ。 任意項目で省略可能。
Customize Name	通知送信者として表示させる表示名。
Customize Icon	通知送信者として表示させるアイコン。
Preview Message	入力した表示名やアイコンの表示確認用プレビュー。

Save Settingsボタン押下で完了です。

3. 作成したWebhook URLをパラメータストアに登録

Webhook URLはLambdaのコードから参照できる必要がありますが、クレデンシャルな情報はコード内に直接書かずセキュアに管理できるパラメータストアに保存し、コードからパラメータストアの情報を読み込む形で使用します。
パラメータストアへの保存方法は、PageSpeed Insights API Keyを保存した時と同じの為、ここでは割合します。

以上でSlack側の用意は終わりですが、今回はパフォーマンス測定結果のスコアが基準値未満だった場合に通知するアラート通知送信先チャンネルと、基準値以上だった場合に通知する通常通知送信先チャンネルの二つが必要になる為、上記手順を繰り返して二つのチャンネルのIncoming WebHooks設定を行います。

事前準備2. IAMロール&ポリシー作成

次に、Lambdaの実行ロールに設定するIAMロールと、ロールにアタッチするポリシーをそれぞれ作成します。
今回作成するLambdaからは、SystemsManager（パラメータストアからWebhook URL取得）とCloudWatchLogsへのアクセスを行う為、それぞれへのアクセスを許可したIAMポリシーと、ポリシーをアタッチするIAMロールを作成します。
CloudWatchLogsへのアクセスを可能にするポリシーは前回作成したポリシーを流用できる為、新規作成はせずに今回作成するIAMロールへ前回作成したポリシーをアタッチします。

1. SystemsManagerへのアクセスを許可するIAMポリシー作成

Webhook URL保存先のSSMのパラメータストアから値を取得する為のIAMポリシーを作成します。

今回はSSMのパラメータストアから2つのWebhook URLの値取得と、取得した値に対するKMSの複合化の許可設定を追加すれば良いので、JSONエディタに以下を記述します。

ZSH

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ssm:GetParameter" ], "Resource": [ "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{アラート通知送信先チャンネルのWebhook URLのキー名}", "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{通常通知送信先チャンネルのWebhook URLのキー名}" ] }, { "Effect": "Allow", "Action": [ "kms:Decrypt" ], "Resource": [ "arn:aws:kms:{リージョン名}:{アカウントID}:key/{KMSのキーID}" ] } ] }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

{     "Version": "2012-10-17",     "Statement": [         {             "Effect": "Allow",             "Action": [                 "ssm:GetParameter"             ],             "Resource": [                 "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{アラート通知送信先チャンネルのWebhook URLのキー名}",                 "arn:aws:ssm:{リージョン名}:{アカウントID}:parameter/{通常通知送信先チャンネルのWebhook URLのキー名}"             ]         },         {             "Effect": "Allow",             "Action": [                 "kms:Decrypt"             ],             "Resource": [                 "arn:aws:kms:{リージョン名}:{アカウントID}:key/{KMSのキーID}"             ]         }     ] }

2. Lambdaの実行ロールに設定するIAMロール作成

上記で作成したIAMポリシーと、前回作成したCloudWatchLogsへのアクセス許可を設定したポリシーの二つをアタッチする、実行ロールとして設定するIAMロールを作成します。

以上で今回のIAMロール/ポリシーの作成は完了です。

転送された通知メッセージをユーザーに送信する処理の作成

事前準備が整ったところで、転送された通知メッセージをユーザーに送信する処理の作成を開始していきます。

1. 共通処理を集めたLambda Layers作成

今回は、パフォーマンス測定結果のスコアが基準値未満だった場合のアラート通知を送るSNSトピックと、基準値以上だった場合の通常通知を送るSNSトピックで、二つのトピックを対象に設定するサブスクライバのLambda functionを作らなければなりません。
一つのLambdaを二つのSNSトピックのサブスクライバに設定する事も可能ですが、通知送信先のSlackチャンネル、およびWebhook URLがそれぞれ異なったりで、一つのLambdaで両方に対応しようとするとどうしてもコード内のif文で無理やり通知先を分けたりといった分岐が必要になってしまいます。

また、アラート通知と通常の通知はたまたま今が同じような処理内容になっているだけで本来意味が違うものでもある為、アラート通知だけSMSにも通知を送るようにしたい。通常通知は通知送信自体を止めたいなど、分岐が増えていくと運用管理が難しくなってしまう為、それぞれのSNSトピックに対してLambdaを一つずつ作成するのが望ましいです。

とは言え、複数のLambdaでほぼ同じような処理を何度も書くのは非効率でコアロジックの変わらない部分は共通化したい。
そんな時に便利なのがLambda Layersで、共通処理を外部に切り出して保存しておき、使いたい時だけ読み込んで使用する事ができます。
今回は、Slackの指定されたWebhook URLに通知を飛ばす処理と、パラメータストアから値を取り出す処理の二つをLayersで作成します。

Lambda Layersの作成はマネジメントコンソールから可能で、「Lambda」 → 「Layers」 → 「レイヤーの作成」と進みます。


レイヤーの作成画面に移るので、各項目を入力していきます。

名前	作成するLambda Layers名。
説明	作成するLambda Layersに付ける説明文。 任意項目で省略可能。
対象ファイル	zipアップロード or S3のパスを指定するかの選択。 対象ファイルの中身は後述する*1と*2を参照。
互換性のあるランタイム	作成したLayersを読み込み可能なランタイムの指定。 今回はPython3.7を指定。
ライセンス	作成するレイヤーのソフトウェアライセンス。 任意項目で省略可能。

*1 Slackの指定されたWebhook URLに通知を飛ばす処理のLayer

lambda layers slack notification sender

Python

import urllib.request import json ## const value. HTTP_METHOD_NAME = 'POST' def generate_send_text(subject, message): """ Generate send text for slack incoming webooks http request data. Parameters ---------- subject : str subject from sns topic. message : str message from sns topic. Returns ------- send_text : str Send text for slack incoming webooks http request data. """ return '*' + subject + '*\n```\n' + message + '\n```' def create_send_data(sns_published_item): """ Create send data for slack incoming webooks http request. Parameters ---------- sns_published_item : dict published item from sns. ext. { "Subject": "example subject" "Message": "example message" } Returns ------- send_data : str Send data for slack incoming webhooks http request. """ send_data = 'payload=' + json.dumps( { 'text': generate_send_text( sns_published_item['Subject'], sns_published_item['Message'] ) } ) return send_data.encode('utf-8') def send_notification(webhook_url, sns_published_item): """ Send notification to slack from sns topic. Parameters ---------- webhook_url : str slack incoming webhooks url. sns_published_item : dict published item from sns. ext. { "Subject": "example subject" "Message": "example message" } Returns ------- response_syntax : HttpResponse Http response from slack incoming webhooks url. """ request = urllib.request.Request( webhook_url, data = create_send_data(sns_published_item), method = HTTP_METHOD_NAME ) with urllib.request.urlopen(request) as response: return response.read().decode('utf-8')

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

import urllib.request import json   ## const value. HTTP_METHOD_NAME = 'POST'   def generate_send_text(subject, message):     """     Generate send text for slack incoming webooks http request data.       Parameters     ----------     subject : str         subject from sns topic.     message : str         message from sns topic.       Returns     -------     send_text : str         Send text for slack incoming webooks http request data.     """     return '*' + subject + '*\n```\n' + message + '\n```'   def create_send_data(sns_published_item):     """     Create send data for slack incoming webooks http request.       Parameters     ----------     sns_published_item : dict         published item from sns.         ext. {             "Subject": "example subject"             "Message": "example message"         }       Returns     -------     send_data : str         Send data for slack incoming webhooks http request.     """     send_data = 'payload=' + json.dumps(         {             'text': generate_send_text(                 sns_published_item['Subject'],                 sns_published_item['Message']             )         }     )     return send_data.encode('utf-8')   def send_notification(webhook_url, sns_published_item):     """     Send notification to slack from sns topic.       Parameters     ----------     webhook_url : str         slack incoming webhooks url.     sns_published_item : dict         published item from sns.         ext. {             "Subject": "example subject"             "Message": "example message"         }       Returns     -------     response_syntax : HttpResponse         Http response from slack incoming webhooks url.     """     request = urllib.request.Request(         webhook_url,         data = create_send_data(sns_published_item),         method = HTTP_METHOD_NAME     )     with urllib.request.urlopen(request) as response:         return response.read().decode('utf-8')

*2 パラメータストアから値を取り出す処理のLayer

lambda layers ssm parameter store

Python

import boto3 class SsmParameterStore: def __init__(self, region_name): """ Parameters ---------- region_name : str SystemsManager region name. """ self.ssm = boto3.client( 'ssm', region_name = region_name ) def get_parameter(self, parameter_name, with_decryption): """ Get ParameterStore Value. Parameters ---------- parameter_name : str target parameter name. with_decryption : bool need decryption. Returns ------- parameter_value : str target parameter store value. """ response = self.ssm.get_parameter( Name = parameter_name, WithDecryption = with_decryption ) return response['Parameter']['Value']

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36

import boto3   class SsmParameterStore:     def __init__(self, region_name):         """         Parameters         ----------         region_name : str             SystemsManager region name.         """         self.ssm = boto3.client(             'ssm',             region_name = region_name         )       def get_parameter(self, parameter_name, with_decryption):         """         Get ParameterStore Value.           Parameters         ----------         parameter_name : str             target parameter name.         with_decryption : bool             need decryption.           Returns         -------         parameter_value : str             target parameter store value.         """         response = self.ssm.get_parameter(             Name = parameter_name,             WithDecryption = with_decryption         )         return response['Parameter']['Value']

作成ボタン押下で完了ですが、一回の作成につき一つのLayerしか作れない為、本手順を二回繰り返して二つのLayersを作成します。

2. 最小項目を埋めた状態のLambda function作成

Lambda functionの作成に入り、今回はSNSトピックへメッセージがpublishされた事を実行トリガーに使用するので、前回と同じく予め用意されている各イベントソース用の設計図を使用して作成を行います。

用意されている各イベントソース用の設計図を使用するには、

1. Lambda function作成時の最初の設定画面で設計図を使用を選択
2. キーワード入力欄にsnsを入力してフィルタ
3. sns-message-pythonを選択
4. 設定ボタン押下

と選択して進みます。


最小項目入力に移るので、各項目を埋めて関数の作成ボタン押下で完了です。

関数名	作成するLambda function名。
実行ロールの選択/作成	処理内から他サービスへのアクセスなどの許可/拒否設定を制御するIAMロールの選択。 今回は事前準備で作成しておいたIAMロールを使用するので、既存のロールを使用するを選択。
既存のロール	上記実行ロールを作成済のIAMロールの中から選択。 事前準備で作成しておいたIAMロールを選択。
SNSトピック	対象のSNSトピック名。 パフォーマンス測定結果を受信する二つのトピックのどちらかを選択。
トリガーの有効化	今すぐ設定を有効化するか、テスト用に無効化した状態で保存するかの設定。 最初は無効化しておき、一通りできたタイミングで有効化する事を推奨。

アラート通知用のLambdaと通常通知用のLambdaが二つ必要になる為、本手順についても二回繰り返して二つのLambdaを作成します。

3. 各種設定項目の変更

作成時に設定した最小項目以外にも幾つか変更する必要がある項目がある為、各種設定項目を変更していきます。
設定項目は非常に多いので、今回はデフォルト値から変更した項目のみ記載します。

ランタイム	処理を実行するプログラム言語。 作成時の設計図にsns-message-pythonを選択した場合はPython2.7になっているので、Python3.7にバージョンを上げておきます。
環境変数	Lambda functionの処理で使用する環境変数の設定。 今回の処理では、以下を環境変数で用意。 SLACK_ALERT_CHANNEL_ WEBHOOK_URL_ PARAMETER_NAME SlackのIncoming WebHooks用のWebhook URLのパラメータストアキー名。 この環境変数を作るのはアラート通知用のLambdaのみ。 SLACK_NOTIFICATION_CHANNEL_ WEBHOOK_URL_ PARAMETER_NAME SlackのIncoming WebHooks用のWebhook URLのパラメータストアキー名。 この環境変数を作るのは通常通知用のLambdaのみ。 SSM_REGION_NAME 使用するSystemsManagerのリージョン名。 東京リージョンならap-northeast-1。
タグ	作成するLambda functionに紐付けるタグ。 管理用にNameタグは設定しておく事を推奨。
基本設定.タイムアウト	処理の実行時間上限設定。 デフォルトは3秒で短過ぎる為、少し伸ばしておきます。 今回は、Slackとの通信に時間がかかった場合でもタイムアウトしないように10秒で設定。

本手順についても同じく、アラート通知用のLambdaと通常通知用のLambdaで、それぞれ設定を変更します。

4. 使用するLayersの指定

今回はLayersを用意していた為、使用するLayersを読み込みます。
Layersの設定は、「読み込む対象のLambdaの詳細画面のLayersタブ選択」 → 「レイヤーの追加」と進みます。


レイヤーの追加画面に移るので、互換性のあるレイヤーから事前に用意したLayersを選択して追加ボタンで追加します。
別途バージョンを作成している場合以外は最新バージョンのみ選択可能なので、使用したいバージョンを選択します。


今回は、SlackのWebhook URLに通知を飛ばす処理と、パラメータストアから値を取得する処理で二つのLayerを用意した為、二つを追加して保存します。
また、本手順についても同じく、アラート通知用のLambdaと通常通知用のLambdaで、それぞれLayersを二つずつ追加します。

5. 関数コードの編集

実行させる処理の内容を記述する為、関数コードを編集していきます。

アラート通知用のSNSトピックサブスクライバに設定するLambda

lambda sns alert topic to slack

Python

import os import logging import ssm_parameter_store import slack_notification_sender ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME = 'SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME' ## singleton object logger = None def initialize_logger_setting(): """ Initialize logger setting. """ global logger if not logger: logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): initialize_logger_setting() ssm = ssm_parameter_store.SsmParameterStore(os.environ[SSM_REGION_NAME]) webhook_url = ssm.get_parameter( os.environ[SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME], True ) response_body = slack_notification_sender.send_notification( webhook_url, event['Records'][0]['Sns'] ) logger.info('send_notification successed. response body = ' + response_body)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

import os import logging import ssm_parameter_store import slack_notification_sender   ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME = 'SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME'   ## singleton object logger = None     def initialize_logger_setting():     """     Initialize logger setting.     """     global logger     if not logger:         logger = logging.getLogger()         logger.setLevel(logging.INFO)   def lambda_handler(event, context):     initialize_logger_setting()          ssm = ssm_parameter_store.SsmParameterStore(os.environ[SSM_REGION_NAME])     webhook_url = ssm.get_parameter(         os.environ[SLACK_ALERT_CHANNEL_WEBHOOK_URL_PARAMETER_NAME],         True     )     response_body = slack_notification_sender.send_notification(         webhook_url,         event['Records'][0]['Sns']     )     logger.info('send_notification successed. response body = ' + response_body)

通常通知用のSNSトピックサブスクライバに設定するLambda

lambda sns notification topic to slack

Python

import os import logging import ssm_parameter_store import slack_notification_sender ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME = 'SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME' ## singleton object logger = None def initialize_logger_setting(): """ Initialize logger setting. """ global logger if not logger: logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): initialize_logger_setting() ssm = ssm_parameter_store.SsmParameterStore(os.environ[SSM_REGION_NAME]) webhook_url = ssm.get_parameter( os.environ[SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME], True ) response_body = slack_notification_sender.send_notification( webhook_url, event['Records'][0]['Sns'] ) logger.info('send_notification successed. response body = ' + response_body)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35

import os import logging import ssm_parameter_store import slack_notification_sender   ## environment variable key name SSM_REGION_NAME = 'SSM_REGION_NAME' SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME = 'SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME'   ## singleton object logger = None     def initialize_logger_setting():     """     Initialize logger setting.     """     global logger     if not logger:         logger = logging.getLogger()         logger.setLevel(logging.INFO)   def lambda_handler(event, context):     initialize_logger_setting()          ssm = ssm_parameter_store.SsmParameterStore(os.environ[SSM_REGION_NAME])     webhook_url = ssm.get_parameter(         os.environ[SLACK_NOTIFICATION_CHANNEL_WEBHOOK_URL_PARAMETER_NAME],         True     )     response_body = slack_notification_sender.send_notification(         webhook_url,         event['Records'][0]['Sns']     )     logger.info('send_notification successed. response body = ' + response_body)

ここでも、アラート通知用と通常通知用で二つのLambdaを編集します。

以上で転送された通知メッセージをユーザーに送信する処理は完成です。

4. 処理を動かして結果を確認

ここまでで下記3ステップが全て完了した為、処理を動かして期待通りに動いているかを確認してみます。

ここまでに作成した3ステップの処理

1. サイトパフォーマンスを測定して結果をDBに保存
2. DBから結果を取り出して通知用サービスに転送
3. 転送された通知メッセージをユーザーに送信

全ての処理は繋がっている為、起点となる一番最初の処理のPageSpeed Insights APIを実行して結果をDynamoDBに保存する処理のLambdaを動かしてみます。
上記Lambdaの実行トリガーはcron起点の定期間隔実行で開始する為、意図したタイミングで処理を開始したい場合には同Lambdaの詳細画面に入りテストの実行ボタンを押下して実行させます。

テスト実行後、以下2点を確認しましょう。

1. DynamoDBへの意図した内容での結果レコード保存

一点目の確認として、結果の保存先として使用したDynamoDBに意図した内容のレコードが保存されている事を確認します。

DynamoDBのレコード確認は、

1. マネジメントコンソールのDynamoDBの管理画面に遷移
2. サイドメニューから「テーブル」選択
3. 一覧から「対象テーブル」を選択
4. ヘッダから「項目」のタブを選択
5. 検索セレクトボックスのチェックを「スキャン」から「クエリ」に変更
6. クエリ対象を「セカンダリインデックスのインデックス名」に変更
7. 値の入力欄に「今日の日付」を入力
8. 表示されたレコードの内容を確認

で確認可能です。


意図した内容のレコードが保存されている事が確認できれば一つ目の確認観点はOKです。

2. Slackチャンネルでの通知受信

二点目の確認として、結果がSlackのチャンネルに通知されてくる事を確認します。

こちらの確認方法は非常に単純で、通知対象のSlackチャンネルの新着メッセージを確認するだけです。


上記のような新着メッセージが受信できていれば、通知の確認もOKです。

以上で確認は終わりですが、もし期待した結果が得られていない場合はどこかの処理で失敗している可能性が高いので、CloudWatchLogsから1ステップずつ各処理が正常に完了しているかを確認してみてください。
どこで処理が止まってしまっているか、何故処理が失敗してしまっているかを調べる上で、各Lambda処理内に入れるログ出力はとても有益な情報になるので、省かずに必ず出力させる事を推奨します。

5. エラー検知

以上で全工程が完了です！と言いたいところではありますが、とても重要なエラー検知可能な状態にする対応がまだできていません。
今回のそもそもの目的はパフォーマンス劣化をすぐに検知できるようにしたいという用件でしたが、検知する為の仕組みが失敗している、尚且つ失敗している事を検知できないのでは本末転倒です。
Lambda functionの処理失敗を検知する方法は非常に簡単ですぐに実現できるので、さくっと設定してしまいましょう。

Lambda functionの実行失敗を検知する為の方法を考える

Lambda functionの実行失敗を検知する方法は色々ありますが、今回は以下の最も設定が楽で運用コストがかからない方法を選択しました。

1. Lambda functionの実行失敗をCloudWatch Alarmで検知

まずは大元の検知で、Lambdaの実行失敗をシステムとして検知できるようにする為にCloudWatch Alarmで対象Lambdaのエラー発生件数が1件以上発生した場合にイベント検知可能な状態にします。

2. CloudWatch Alarmから処理失敗のアラート通知をSNSに転送

続いて、ユーザが処理が失敗している事を検知できるようにする為に、CloudWatch AlarmからSNSに処理が失敗している情報を転送します。

3. SNSからAWS Chatbotを介してSlackチャンネルに通知

最後に、CloudWatch Alarmから処理の失敗通知を受信したSNSのトピックに設定するサブスクライバにSlackをセットし、処理が失敗した事実をユーザがSlack上で気付けるようにします。
尚、SNSにメッセージを送信するサービスがCloudWatch Alarmという事もあり、完全GUIでSlackに通知を送信可能なAWS Chatbotが使用可能な為、今回はコードを一切書かずにAWS Chatbotで通知を送信させる方法を取ります。

以上が、今回採用したLambda functionの実行失敗を検知する仕組みになります。

エラー検知の仕組みを実装

それでは、仕組みを実際に作っていきましょう。

1. SNSトピックの作成

SNSトピックの新規作成方法は、DynamoDBの項目変更を検知してパフォーマンス測定結果をSNSに転送処理の説明で一度書いたので割合しますが、それらとは別に今回新規に一つSNSトピックを作成します。
作成する内容は、全てデフォルト設定で大丈夫です。
扱う内容がセキュアな内容ではないことと、暗号化を有効化するとCloudWatchLogs側の設定で諸々面倒な設定が必要になる為、暗号化を無効化したトピックを作成します。

2. AWS Chatbotでサブスクライバを作成

続いて、作成したトピックに設定するサブスクライバをAWS Chatbotから作成していきます。

AWS Chatbotでのサブスクライバ作成は、マネジメントコンソールからAWS Chatbotの管理画面に入り、「チャットクライアントにSlackを選択」 → 「クライアントを設定」と進みます。


AWS Chatbotとの連携確認画面に移るので、Allowを選択して許可します。


詳細設定画面に移るので、各項目を入力していきます。

設定名	作成するAWS Chatbotのチャネル名。
ログ記録	設定の記録をログ記録させるかどうかの設定。
Slackチャネルタイプ	通知送信先のチャンネルがパブリックチャンネルかプライベートチャンネルかの選択。
パブリックチャンネル	通知送信先チャンネルの選択。
IAMロール	Chatbotに設定する他サービスへのアクセスなどの許可/拒否設定を制御するIAMロールの選択。 非常に弱い権限しか必要としない為、今回はテンプレートを使用してIAMロールを作成するを選択。
ロール名	新規作成する際のIAMロール名。
ポリシーテンプレート	上記IAMロールに設定するポリシー選択。 今回は、通知のアクセス許可だけで十分。
SNSトピック-リージョン	通知送信先のSNSトピックが存在するリージョン。 東京リージョンにSNSを選択した場合は、アジアパシフィック-東京。
SNSトピック-トピック	通知送信先のSNSトピック選択。 今回新しく作成したトピックを選択。

設定ボタン押下でAWS Chatbot経由でのサブスクライバ作成が完了です。

3. CloudWatch Alarmの作成

最後に、作成したSNSトピック/サブスクライバに通知をpubishするCloudWatch Alarmを作成します。

CloudWatch Alarmの作成は、マネジメントコンソールからCloudWatchの管理画面に入り、「アラーム」 → 「アラームの作成」と進みます。


メトリクスの選択画面に移るので、「Lambda」 → 「関数名別」と進み、今回作成したLambdaのエラー(Errors)にチェックを付けてメトリクスの選択ボタン押下で先に進みます。


詳細項目の入力画面に移るので、各項目を入力していきます。

メトリクス名	アラーム対象のメトリクス名。 デフォルトのErrorsのまま入力。
FunctionName	対象のLambda Function名。
統計	アラーム状態に遷移させる為の集計対象。 1件でもエラーが発生したら検知したいので合計を選択。
期間	集計期間の設定。 デフォルトでは最短1分間隔の為、1分で設定。
しきい値の種類	値をしきい値として使用するかバンドをしきい値として使用するかの設定。 値をしきい値として使用するを選択。
Errorsが次の時…	アラーム状態に遷移させる条件。 1件でもエラーが発生したら検知したいので以上を選択。
…よりも	アラーム状態に遷移させる条件の比較対象件数。 1件でもエラーが発生したら検知したいので1を選択。
アラームを実行するデータポイント	アラーム状態に遷移させる条件の評価期間とデータポイント数。 1分間に1件でも発生したら即検知したいので1/1を選択。
欠陥データの処理	データ不足数の際の挙動設定。 欠陥データを見つかりませんとして処理するように選択
アラーム状態トリガー	アラームステータスに遷移させる条件。 アラーム状態になった際にアラームステータスになるように設定。
SNSトピックの選択	アラームステータスに遷移時の通知送信先SNSトピック選択。 今回新しく用意したトピックを選択。
Auto Scalingアクション	アラームステータスに遷移時のオートスケール設定。 今回は使用しない。
EC2アクション	アラームステータスに遷移時のEC2アクション設定。 今回は使用しない。
アラーム名	今回作成するアラーム名。
アラームの説明	今回作成するアラームに付ける説明。 任意項目で省略可能。

最後にプレビュー画面が表示されるので、一通り内容を確認して問題がなければアラームの作成ボタン押下で作成完了です。

また、今回作成したLambda Functionは全部で4つあるはずなので、全てのLambda Functionに対して上記CloudWatch Alarm設定を行います。

以上でエラー検知の仕組み作成は完了です。

確認の為、Lambda Functionのコードにバグを仕込んでテスト実行させてみて、CloudWatchから下記のような通知がSlackに届いていればエラー検知の仕組みが正常に動いている事の確認もOKです。
テスト実行後、バグを埋め込む前の元の状態に戻すのを忘れないでください。

以上で今回の用件を満たす全対応が完了です。
お疲れ様でした！