はじめに
CloudWatch Insightsの分析結果を監視するLambdaを作成しました。
CloudWatch Logs Insightsとは
CloudWatch Logs Insightsとは、2018年のre:Inventで発表されたサービスです。
詳細に関しては、以下のサイトをご覧いただければと思いますが、一言で説明すればCloudWatch Logs内のログデータを簡単に集計・分析できるサービスです。
ロググループに対する専用のクエリ言語がサポートされており、そのクエリを投げることで簡単にログの集計・分析が行え、さらにグラフを作成して結果を可視化することもできます。これらの分析結果をCloudWatchダッシュボードに追加することもできます。
新機能 – Amazon CloudWatch Logs Insights – 高速でインタラクティブなログ分析
CloudWatch Logs Insights を使用したログデータの分析
やったこと
今回、CloudWatch Insightsで得られた分析結果を監視するLambdaを作成しました。
具体的には、APIGatwayのアクセスログのステータスコードに400以上のものがある場合にSlackに通知する
ということを行いました。
注：「LambdaでInsightsの分析結果を使う」ことをしたかったため、そのままCloudWatchのメトリクスを使えばもっと簡単にできます。
以下、構成図です。
APIGatewayのアクセスログをCloudWatch Logsに吐き出し、
LambdaでCloudWatch Logs Insightsのクエリを実行し分析結果を取得、
その分析結果の内容にしたがってSlackに通知するということを行います。
なお、2019年8月現在、Insightsの分析結果をCloudWatch Alermのメトリクスに指定することはできなかったため、LambdaはCloudwatch Eventsによって5分ごとに動かすようにしました。
事前準備
Lambdaを作成する前に、いくつかの事前準備を行います。
SlackWebhookのURLを取得する。
Slack通知を行う場合、SlackWebhookのURLを取得しておいてください。
IAMロールの作成
今回使用するIAMロールは以下の2つあるので、それらを作成します。
1.APIGateway用IAMロール
後述しますが、APIGatewayはデフォルト状態ではアクセスログをCloudWatch Logsに書き出す仕様ではないためその設定を行う必要があります。その際、APIGatewayがCloudWatch Logsにログを書き込む権限が必要となるためそのIAMロールを作成します。
デフォルトで存在するポリシー「AmazonAPIGatewayPushToCloudWatchLogs」をアタッチしたロールを作成し、ARNをコピーしておいてください。
2.Lambda用IAMロール
Lambdaに割り当てるロールを作成します。今回、CloudWatchLogsへのアクセス権限が必要なため、以下のIAMポリシーをアタッチしたロールを作成しました。
{
"Version": "2012-10-17",
"Statement": [
{
"Action": [
"logs:*"
],
"Effect": "Allow",
"Resource": "*"
}
]
}
ロググループの作成
APIGatewayのアクセスログを書き出すためのCloudWatch Logsのロググループを作成しておきます。
APIGatewayの設定
APIGatewayの対象APIのアクセスログの設定を行います。デフォルト状態では、アクセスログは取得しない設定になっているためアクセスログをCloudWatch Logsに吐き出す設定を行います。
サービス > APIGateway > 設定
でARNにAPIGatway用に作成したIAMロールのARNを設定します。
アクセスログを取得するAPIを選択 > ステージを選択 > ログのトレース
CloudWatch ログを有効化し、アクセスログを有効化、ログの形式を設定 (今回は、JSONにしました)。
これでAPIGatwayの設定は完了です。
Lambdaの設定
環境変数
今回、以下の環境変数を設定しました。
CHANNEL_LIST ：通知するSlackのチャンネル名をカンマ区切りで設定。
LOGGROUP_NAMES ： 監視するCloudWatchLogsのロググループをカンマ区切りで設定。
import os
import datetime
import time
import ast
import slackweb
import boto3
at_color = "danger"
client = boto3.client('logs')
channel_list = os.environ.get('CHANNEL_LIST').split(',')
logGroupNames = os.environ.get('LOGGROUP_NAMES').split(',')
slack_api = "Slack WebhookのURLを記載"
def lambda_handler(event, context):
five_minutes_ago = datetime.datetime.now() - datetime.timedelta(minutes = 5)
startTime = five_minutes_ago.replace(second = 0, microsecond = 0)
endTime = startTime + datetime.timedelta(minutes = 5) - datetime.timedelta(milliseconds = 1)
queryString = 'fields @timestamp, @message | sort @timestamp desc | filter (status >= 400)'
for logGroupName in logGroupNames:
error_logs = results(logGroupName, startTime, endTime, queryString)
if len(error_logs) > 0:
for error_log in error_logs:
dict_message = ast.literal_eval(error_log[1]['value'])
text = str(logGroupName) + '\n' + str(dict_message)
for channel in channel_list:
try:
post_slack(channel,text)
except Exception as e:
post_slack(channel,text)
def results(logGroupName, startTime, endTime, queryString):
start_query_res = client.start_query(
logGroupName=logGroupName,
startTime=int(startTime.timestamp()),
endTime=int(endTime.timestamp()),
queryString=queryString
)
queryId = start_query_res['queryId']
get_results_res = client.get_query_results(
queryId=queryId
)
while get_results_res['status'] == 'Running':
time.sleep(5)
get_results_res = client.get_query_results(
queryId=queryId
)
return get_results_res['results']
def post_slack(channel,text):
slack = slackweb.Slack(url=slack_api)
slack.notify(
username="API-accesslog",
channel=channel,
attachments = [{
"color" : at_color,
"text" : text
}]
)