CloudWatch Logs Insightsを利用したログ監視

ログ監視時にログの内容も通知したいけど、ロググループ単位でサブスクリプションフィルター設定したくないので、Logs Insightsを使ってみました。

## AWS利用サービス

AWSで以下のサービスを利用して定期的にログ抽出します。

- パラメータストア：ログ監視設定用
- EventBridge：定期実行用
- Lambda：抽出処理用
- SNS：エラー通知用
- CloudWatch Logs Insights：エラーログ抽出用

## パラメータストア設定

通知に必要となる以下の情報はすべてパラメータストアで管理します。

### 設定項目

| 項目 | 概要 |
|-|-|
| ロググループの一覧 | 監視対象のロググループを設定 |
| 抽出クエリ | CloudWatch Logs Insights で利用するクエリ文字列  |

### 設定値の例

- ロググループの一覧：1行1ロググループで設定
    ```
    /aws/lambda/hoge-func
    /aws/lambda/fuga-func
    ```
- 抽出クエリ：エラーキーワード`[ERROR]`で抽出する場合
    ```sql
    fields @timestamp, @log, @message
    | filter @message like "[ERROR]"
    | parse @log "*:*" as account, logname
    | sort @timestamp asc
    | display (@timestamp + (9*60*60*1000)) as timestamp_jst, account, logname, @message
    ```

## Lambda周り

Node20、AWS SDK v3で作成しています。
ちなみに、ロググループは最大50個まで対応しています。（Logs Insightsが1回で最大50件なので、件数チェックしてループすれば対応できるけど今回は不要なのでやってないです）

### ソースコード

```javascript
const { CloudWatchLogsClient, StartQueryCommand, GetQueryResultsCommand } = require('@aws-sdk/client-cloudwatch-logs')
const { SNSClient, PublishCommand } = require('@aws-sdk/client-sns')
const { SSMClient, GetParameterCommand } = require('@aws-sdk/client-ssm')

// CloudWatch Logsクライアントを作成
const cloudwatchlogsClient = new CloudWatchLogsClient({})

// SNSクライアントを作成
const snsClient = new SNSClient({})

// SSMクライアントを作成
const ssmClient = new SSMClient({})

// 起動間隔（分）を取得
const interval = parseInt(process.env.STARTUP_INTERVAL)

// Logインサイトでは直近数分のログが検索できないため
// 差し引く時間を設定
// ※設定は分になるため、秒に変換
const latency = parseInt(process.env.LOG_QUERY_LATENCY_TIME) * 60

// log insight のクエリで取得したフィールド値の取得
const getField = (result, fieldName) => {
  const fieldValue = result.find(field => field.field === fieldName)
  return fieldValue ? fieldValue.value : ''
}

// export const handler = async (event) => {
module.exports.handler = async (event) => {

  console.log(`event: ${JSON.stringify(event)}`)

  // Lambdaが起動した時間を取得し、Dateオブジェクトに変換
  let eventTime = new Date(event.time)

  // 秒とミリ秒を切り捨てて、分単位に丸める
  eventTime.setUTCSeconds(0, 0)

  // 分を取得
  let minute = eventTime.getUTCMinutes()

  // 現在の時間を5分単位に丸める
  let currentWindowEndMinute = Math.floor(minute / interval) * interval

  // ウィンドウの終了時間を設定
  let windowEndTime = new Date(eventTime)
  windowEndTime.setUTCMinutes(currentWindowEndMinute)

  // ウィンドウの開始時間を設定（終了時間からinterval分引く）
  let windowStartTime = new Date(windowEndTime)
  windowStartTime.setUTCMinutes(windowEndTime.getUTCMinutes() - interval)

  // 開始時間と終了時間をタイムスタンプ（ミリ秒）に変換
  let startTime = windowStartTime.getTime()
  let endTime = windowEndTime.getTime() - 1 // 終了時間ぴったりのログは含まれない

  // 開始時間と終了時間をタイムスタンプ（秒）に変換（CloudWatch Logs APIの仕様）
  let startTimeInSeconds = Math.floor(startTime / 1000) - latency
  let endTimeInSeconds = Math.floor(endTime / 1000) - latency

  try {
      // パラメータストアからロググループ名を取得
      const logGroupNamesParamName = process.env.LOG_GROUPS_PARAMETER_STORE
      const getLogGroupNamesCommand = new GetParameterCommand({
          Name: logGroupNamesParamName,
          WithDecryption: false, // 暗号化されていない場合
      })
      const logGroupNamesResponse = await ssmClient.send(getLogGroupNamesCommand)
      const logGroupNamesString = logGroupNamesResponse.Parameter.Value

      // 改行で分割して配列に変換
      const logGroupNames = logGroupNamesString.split('\n').map(name => name.trim()).filter(name => name !== '')

      // パラメータストアからクエリ文字列を取得
      const queryStringParamName = process.env.LOG_QUERY_PARAMETER_STORE
      const getQueryStringCommand = new GetParameterCommand({
          Name: queryStringParamName,
          WithDecryption: false, // 暗号化されていない場合
      })
      const queryStringResponse = await ssmClient.send(getQueryStringCommand)
      const queryString = queryStringResponse.Parameter.Value

      // クエリを開始するためのパラメータを設定
      const params = {
          logGroupNames: logGroupNames,
          startTime: startTimeInSeconds,
          endTime: endTimeInSeconds,
          queryString: queryString,
      }

      // クエリを開始
      const startQueryCommand = new StartQueryCommand(params)
      const startQueryResponse = await cloudwatchlogsClient.send(startQueryCommand)

      // クエリIDを取得
      const queryId = startQueryResponse.queryId

      // クエリの結果を取得するためのループ
      let queryStatus = 'Running'
      let queryResults = []
      while (queryStatus === 'Running' || queryStatus === 'Scheduled') {
          // 1秒待機してから再度チェック
          await new Promise(resolve => setTimeout(resolve, 1000))

          // クエリの結果を取得
          const getQueryResultsCommand = new GetQueryResultsCommand({ queryId })
          const getQueryResultsResponse = await cloudwatchlogsClient.send(getQueryResultsCommand)

          // クエリのステータスを更新
          queryStatus = getQueryResultsResponse.status

          // クエリが完了した場合、結果を取得
          if (queryStatus === 'Complete') {
              queryResults = getQueryResultsResponse.results
          }
      }

      // エラーログが存在するかチェック
      if (queryResults && queryResults.length > 0) {
          // エラーログが存在する場合、SNSでメール通知を送信

          // SNSトピックARNを環境変数から取得
          const topicArn = process.env.SNS_TOPIC_ARN

          // エラー発生のログ
          console.log(`${queryResults.length}件のエラーが検出されました`)

          for (const result of queryResults) {
              // 各結果からタイムスタンプとメッセージを抽出
              const timestampJst = getField(result, 'timestamp_jst')
              const logMessage = getField(result, '@message')
              const logname = getField(result, 'logname')
              
              // ★タイトルや通知メッセージは必要に応じて変更
              const message = `発生日時: ${timestampJst}\n対象ログ: ${logname}\n\nログメッセージ: \n${logMessage}`
              
              // SNSメッセージを送信
              const publishCommand = new PublishCommand({
                  TopicArn: topicArn,
                  Subject: `[エラーログ通知]${logname}`,
                  Message: message,
              })
              await snsClient.send(publishCommand)
          }

      } else {
          // エラーログが存在しない場合の処理（必要に応じて）
          console.log('エラーログは検出されませんでした')
      }

  } catch (error) {
      // エラーが発生した場合の処理
      console.error('エラーが発生しました：', error)
      throw error
  }
}
```

### 環境変数

ソースコード上の環境変数ごとに説明は以下の通り

| 環境変数 | 設定値説明 |
|-|-|
| LOG_GROUPS_PARAMETER_STORE | 対象ロググループ一覧のパラメータストア名  |
| LOG_QUERY_PARAMETER_STORE | 抽出クエリのパラメータストア名 |
| STARTUP_INTERVAL | EventBridgeでの呼び出し間隔時間（分）を設定 |
| LOG_QUERY_LATENCY_TIME | Logインサイトでは直近数分のログが検索できないため差し引く時間（分）を設定 |
| SNS_TOPIC_ARN | 通知先のSNSトピックARN |
| TZ | `Asia/Tokyo` 固定 |

### 必要権限

Lambdaに必要なIAM設定は以下の通り

| 権限 | 必要な設定 |
|-|-|
| ログ記録 | logs:CreateLogGroup<br>logs:CreateLogStream<br>logs:PutLogEvents |
| ログ抽出 | logs:StartQuery<br>logs:GetQueryResults |
| SNS通知 | sns:Publish |
| パラメータストア取得 | ssm:GetParameter |

## 注意点

CloudWatch Logsにログが出力されてからインデックス化が行われて、CloudWatch Logs Insightsで抽出できるまでタイムラグが2分ぐらいあるので、タイムラグを注意して設定する必要があります。

例えば、環境変数に以下の値が設定されている場合、Lambda起動時にチェックするログは3分前～8分前が対象になります。

- タイムラグ：3分　※`LOG_QUERY_LATENCY_TIME`のこと
- ログチェック間隔：5分　※`STARTUP_INTERVAL`のこと

## 特定キーワードの除外設定をする場合

パラメータストアでクエリを管理しているため、対象キーワードで除外する`filter`を`not like`で追加するのみで対応できます。
※正規表現で指定することもできるので結構柔軟にできるはずです。

```sql
fields @timestamp, @log, @message
| filter @message like "ERROR"
| filter @message not like "hogehoge"
| filter @message not like "fugafuga"
| sort @timestamp asc
```