定期的にCloudWatchLogsを集計してSlackに流すLambdaを作る

目的

CloudWatchLogsに以下のようなログをためています。

{
    "error_message": "NotLogined(\"session not found\")",
    "error_name": "NotLogined",
    "path": "/api/v1/member/me"
}

これを集計してerror_nameごとに何件エラーがあるのかを5分ごとに通知します。

Slackには以下のように通知します。

web1
NotLogined : 3

構成

AWS LambdaにRubyのコードで集計してSlackに通知するプログラムを作りました。
5分ごとに定期的に実行させたいのでEventBridgeのSchedulerを設定します。
どのCloudWatchLogsを集計して、どのSlackに通知するのかはスケジューラーに設定しています。
以下のような感じです。

{
  "user_name": "slack_notifier",
  "targets": [
    {
      "channel": "#log-my-project-production",
      "log_groups": [
        {
          "title": "web1",
          "name": "my-project-web1-esc"
        }
      ]
    }
  ]
}

コード

Ruby

Gemfile

# frozen_string_literal: true

source "https://rubygems.org"

# gem "rails"
gem 'aws-sdk'
gem 'httparty'
gem 'rexml'

main.rb

require 'httparty'
require 'json'
require 'aws-sdk-ssm'
require 'aws-sdk-cloudwatchlogs'
require 'time'

# SystemManagerのパラメーターストアーからSlackのトークンを取得する
def get_token
  ssm_client = Aws::SSM::Client.new
  request = {
    name: 'SLACK_TOKEN',
    with_decryption: true
  }
  response = ssm_client.get_parameter(request)
  response.parameter.value
end

# スラックに通知する
def send_message(channel, username, text)
  content = {
    channel: channel,
    username: username,
    text: text
  }
  url = "https://slack.com/api/chat.postMessage"
  response = HTTParty.post(url, 
    body: content.to_json,
    headers: {
      'Content-Type' => 'application/json',
      'Authorization' => "Bearer #{get_token}"
    }
  )
  JSON.parse(response.body.to_s)
end

# CloudWatchLogsからログを集める
def get_logs(log_group_name)
  client = Aws::CloudWatchLogs::Client.new
  now = Time.now.getgm
  # 直近の0,5,10,15,20,25,30,35,40,45,50,55分の時間を探す
  end_time = Time.gm(now.year, now.month, now.day, now.hour, now.min - now.min % 5, 0)
  # 終了時間から5分前が開始時間
  start_time = end_time - 5 * 60
  resp = client.filter_log_events({
    log_group_name: log_group_name,
    start_time: start_time.to_i * 1000,
    end_time: end_time.to_i * 1000,
    filter_pattern: 'error_message',
    limit: 100,
  })
  resp.events
end

# ログを集計する
def summary_logs(logs)
  map = {}
  logs.each do |log|
    begin
      message = JSON.parse(log.message)
      map[message["error_name"]] = (map[message["error_name"]] || 0) + 1
    rescue
    end
  end
  res = []
  map.each_pair do |key, value|
    res << "#{key} : #{value}"
  end
  res
end

# Lambdaのエントリーポイント
def handler(event:, context:)
  event["targets"].each do |target|
    result = []
    target["log_groups"].each do |log_group|
      res = summary_logs(get_logs(log_group["name"]))
      if !res.empty?
        result << "#{log_group["title"]}\n#{res.join("\n")}"
      end
    end
    if !result.empty?
      send_message(target["channel"], event["user_name"], result.join("\n\n"))
    end
  end
end

CDK

cdk-notify-slack-stack.ts

import { Stack, StackProps } from 'aws-cdk-lib';
import { Construct } from 'constructs';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as scheduler from 'aws-cdk-lib/aws-scheduler';
import * as cdk from 'aws-cdk-lib';

export class CdkNotifySlackStack extends Stack {
  constructor(scope: Construct, id: string, props?: StackProps) {
    super(scope, id, props);

    const procejctName = 'my-project'
    const region = cdk.Stack.of(this).region
    const accountId = cdk.Stack.of(this).account

    const lambdaRole = new iam.Role(this, 'LambdaRole', {
      assumedBy: new iam.ServicePrincipal('lambda.amazonaws.com'),
      roleName: `${procejctName}-role-slack-notifier-lambda`,
      inlinePolicies: {
        CloudWatch: iam.PolicyDocument.fromJson({
          'Version': '2012-10-17',
          'Statement': [
            {
              'Effect': 'Allow',
              'Action': [
                'logs:FilterLogEvents',
              ],
              'Resource': '*'
            },
          ]
        }),
        Ssm: iam.PolicyDocument.fromJson({
          'Version': '2012-10-17',
          'Statement': [
            {
              'Effect': 'Allow',
              'Action': [
                'ssm:GetParameters',
                'ssm:GetParameter'
              ],
              'Resource': [
                `arn:aws:ssm:${region}:${accountId}:parameter/SLACK_TOKEN`
              ]
            },
          ]
        })
      }
    })

    // スラック通知
    const notifyFunction = new lambda.Function(this, 'NotifySlackHandler', {
      runtime: lambda.Runtime.RUBY_3_2,
      code: lambda.Code.fromAsset('./slack_notifier'),
      role: lambdaRole,
      handler: 'main.handler',
      timeout: cdk.Duration.seconds(60),
      functionName: `${procejctName}-slack-notifier`,
    });

    const scheduleRole = new iam.Role(this, 'ScheduleRole', {
      assumedBy: new iam.ServicePrincipal('scheduler.amazonaws.com'),
      roleName: `${procejctName}-role-slack-notifier-schedule`,
      inlinePolicies: {
        Lambda: iam.PolicyDocument.fromJson({
          'Version': '2012-10-17',
          'Statement': [
            {
              'Effect': 'Allow',
              'Action': [
                'lambda:InvokeFunction',
              ],
              'Resource': notifyFunction.functionArn,
            },
          ]
        })
      }
    })

    const event = new scheduler.CfnSchedule(this, 'MyCfnSchedule', {
      name: `${procejctName}-slack-notifier`,
      flexibleTimeWindow: {
        mode: 'OFF',
      },
      scheduleExpression: "cron(0/5 * * * ? *)",
      target: {
        arn: notifyFunction.functionArn,
        roleArn: scheduleRole.roleArn,
        retryPolicy: {
          maximumEventAgeInSeconds: 60,
          maximumRetryAttempts: 0,
        },
        input: JSON.stringify({
          user_name: "slack_notifier",
          targets: [
            {
              channel: '#log-my-project-production',
              log_groups: [
                {
                  title: 'web1',
                  name: "my-project-web1-esc"
                }
              ]
            }
          ]
        })
      }
    })
  }
}

おまけ

最初RubyのHttpクライアントにHTTP.rbを使おうと思ったんですがこれがffiを参照していて以下のようなエラーがでました。

"libffi.so.7: cannot open shared object file: No such file or directory - /var/task/vendor/bundle/ruby/3.2.0/gems/ffi-1.15.5/lib/ffi_c.so"

どうやらOS側に存在しないようで、調べると現状はlayerをつっこむくらいしかできなくて、面倒なのでffiをつかっていなそうなhttpartyにしました。