CloudWatch Logsに出力されたエラーログ本文のSlackへの転送

お試しで作ったアプリケーションを何日か動かせてみて、期待通りに動いているかを手軽に確認したい。
でも、まだ試作段階なので確認の為にお金はかけたくない。
そんな時に自分がよく使っている CloudWatch Logsのエラーログ本文をSlackに転送して確認する方法があったのですが、毎回「どうやって作っていたっけ？」と思い出しながらになってしまっていたので、すぐ作れるようにまとめてみます。

やりたいこと

何を達成したいかの機能要件的な内容です。

1. 想定外の挙動があった場合に何が起こっていたのか確認したい

お試しで作ってみたアプリケーションなどを実際に動かせてみて結果を見るとともに、想定外の挙動が発生した場合には、何が起こっていたのかを確認したいです。

2. 確認に手間をかけたくない

想定外の挙動があった場合に通知を受け取るだけならCloudWatch Alarmを設定するだけですぐにできますが、何が起こっていたかの詳細を確認する為にはアラーム内のリンクからマネジメントコンソールに飛んで確認しにいく必要があります。
複数のAWSアカウントを使用していたりする場合にこれが結構面倒なので、通知を受け取った場所で詳細まで見れるようにして確認の手間を最小限にしたいです。

3. 確認の為の仕組み作成にコストをかけたくない

監視をちゃんと作ろうとした場合、DatadogやKibanaにログを転送して各種ダッシュボード上での確認やフィルタを使用した検索などをやりたいですが、お試しで作った段階ではまだそこまでを必要としておらず、確認の為の仕組み作成にコストをかけたくない要求の方が強い場合がほとんどです。
そんな事情から、パターン化しておけば仕組みの作成自体がすぐにできることと、作った仕組みが無料で利用できることの2点は担保したいです。

4. しっかりした監視の仕組み作成時に作った内容を流用したい

これはサブ的な内容にはなりますが、ある程度軌道に乗ってきた後で、作ったものを捨てて1から作り直すのは勿体無いので、できれば避けたいです。
監視の仕組みを本格的に作成する際に、1から作成し直すのではなく、今回作ったものをそのまま広げていけるような作りにしておきたいです。

全体アーキテクチャ

要求を実現するためのアーキテクチャとして、以下のような構成で設計しました。

大きく分けて、3つのステップが存在します。

CloudWatch Logsのログ内容をKinesis Firehoseに転送

各種アプリケーションの実行ログを保存するCloudWatch Logsから、サブスクリプションフィルタを使用して内容をKinesis Firehoseに転送します。

Kinesis FirehoseからS3に転送

受け取ったログを一定時間毎にまとめてS3にログファイルとして転送します。
また、CloudWatch Logsのサブスクリプションフィルタから送られてきた内容はGZIP圧縮されている為、Kinesisに入ってきたタイミングで解凍する為の変換用Lambdaを挟ませています。

ファイルの出力内容をLambdaでSlackに転送

KinesisからS3へのファイルアップロードをトリガーにLambdaを発火させ、ファイルの出力内容を通知を受け取る媒体に転送します。
自分にとって確認するのに最も楽だった理由から、今回は通知を受け取る媒体としてSlackを選択しました。

作ったもの

terraformで上記構成を全て作成したサンプルコードをGitHubに置いておいたので、よければご参照ください。
文章の説明は結構長くなるので、ある程度コード読める人にはここから下の説明全てスキップしてコードを見た方が早いと思います。

MasakiMisawa/cwl-transfer-slack-sample

作成方法

早速作っていきます。

1. 転送先のS3バケット作成

全体構成的には一番初めのCloudWatch Logsのサブスクリプションフィルタから作成していきたいところですが、サブスクリプションフィルタを作成する為には転送先に指定するKinesisが必要で、Kinesisを作成する為には今度はKinesisからの転送先のS3バケットが必要になる為、まずはS3バケットを最初に作成します。

今回は、cwl-transfer-slack-sample-bucketというバケットを作成しました。
Web公開用途ではないので、パブリックアクセスのブロック設定も有効にします。

バージョニング設定も有効にして、バケットを作成ボタンを押せば作成完了です。

作成したリソース管理用にterraformで作成する場合は以下になります。
記事の都合上最小記述に縮めていますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。

provider "aws" {
  region = "ap-northeast-1"
}

resource "aws_s3_bucket" "cwl-transfer-slack-bukcet" {
  bucket        = "cwl-transfer-slack-sample-bucket"
  acl           = "private"
  force_destroy = false
  versioning {
    enabled    = true
    mfa_delete = false
  }

  tags = {
    Name = "cwl-transfer-slack-sample-bucket"
  }
}

resource "aws_s3_bucket_public_access_block" "cwl-transfer-slack-bukcet-block-public-access" {
  bucket = aws_s3_bucket.cwl-transfer-slack-bukcet.id

  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

output "bucket_arn" {
  value = aws_s3_bucket.cwl-transfer-slack-bukcet.arn
}

output "bucket_name" {
  value = aws_s3_bucket.cwl-transfer-slack-bukcet.id
}

provider "aws" {

region = "ap-northeast-1"

}

resource "aws_s3_bucket" "cwl-transfer-slack-bukcet" {

bucket = "cwl-transfer-slack-sample-bucket"

acl = "private"

force_destroy = false

versioning {

enabled = true

mfa_delete = false

}

tags = {

Name = "cwl-transfer-slack-sample-bucket"

}

resource "aws_s3_bucket_public_access_block" "cwl-transfer-slack-bukcet-block-public-access" {

bucket = aws_s3_bucket.cwl-transfer-slack-bukcet.id

block_public_acls = true

block_public_policy = true

ignore_public_acls = true

restrict_public_buckets = true

}

output "bucket_arn" {

value = aws_s3_bucket.cwl-transfer-slack-bukcet.arn

}

output "bucket_name" {

value = aws_s3_bucket.cwl-transfer-slack-bukcet.id

}

2. Kinesisに設定するCWLの解凍用Lambda作成

次はKinesis Firehoseの作成ですが、その前にFirehoseにレコードが転送されてきたタイミングで実行されるProcessorのLambda functionを先に作成します。
このLambdaは、CloudWatch Logsから転送されてきたログデータがデフォルトで圧縮されているので解凍処理を行うのが役割です。

CloudWatch Logsから送られてきた圧縮ログを解凍する為のブループリントが用意されているので、Lambda作成時はこの設計図を使用すると便利です。

今回は、cwl-compress-to-kinesisという関数を作成しました。

設定するIAM Roleとして用意したロールには、以下のポリシーを設定しています。
firehoseのリソースには、この後作成する予定のストリーム名を先に入れてしまっています。
[IAM ROLE(CWLCompressToKinesisLambdaRole).IAM Policy]

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "firehose:PutRecordBatch",
                "firehose:PutRecord"
            ],
            "Resource": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents",
                "logs:CreateLogStream"
            ],
            "Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/lambda/cwl-compress-to-kinesis:*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "",

"Effect": "Allow",

"Action": [

"firehose:PutRecordBatch",

"firehose:PutRecord"

"Resource": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream"

{

"Sid": "",

"Effect": "Allow",

"Action": "logs:CreateLogGroup",

"Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:*"

{

"Sid": "",

"Effect": "Allow",

"Action": [

"logs:PutLogEvents",

"logs:CreateLogStream"

"Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/lambda/cwl-compress-to-kinesis:*"

}

]

}

今回使用したブループリントはPython2.7用しか用意されていない為、他のコードと同じ環境でテスト実行などを可能にする為にランタイムをPython3.8に上げます。

そのままランタイムをPython3.8に上げてしまうとシンタックスエラーになってしまうので、実行するコードを以下のように変更します。
Pythonが2系から3系に上がるとストリームを扱うStringIO.StringIOが使えなくなるので、そこだけIO.SttringIOに変えているだけですね。
ランタイムがPython2系のままでも別に良い場合はこの変更は不要の為、用意されたデフォルトの設計図をそのまま使用して問題ありません。

タイムアウト設定を少し伸ばしてあげれば、Kinesisに設定するCWLの解凍用Lambdaの設定は完了です。

terraformで作成する場合は以下になります。
記事の都合上最小記述に縮めていますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。
また、Lambda実行コードの functions/lambda_function.zip は、こちらにあるのでここでは省略します。

provider "aws" {
  region = "ap-northeast-1"
}

data "aws_caller_identity" "self" {}

data "aws_iam_policy_document" "principal_policy" {
  statement {
    actions = ["sts:AssumeRole"]

    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
  }
}

data "aws_iam_policy_document" "cwl_compress_to_kinesis_lambda_iam_policy" {
  statement {
    effect = "Allow"
    actions = [
      "firehose:PutRecord",
      "firehose:PutRecordBatch"
    ]
    resources = ["arn:aws:firehose:ap-northeast-1:${data.aws_caller_identity.self.account_id}:deliverystream/cwl-transfer-s3-sample-stream"]
  }

  statement {
    effect    = "Allow"
    actions   = ["logs:CreateLogGroup"]
    resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:*"]
  }

  statement {
    effect = "Allow"
    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents"
    ]
    resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/lambda/cwl-compress-to-kinesis:*"]
  }
}

data "archive_file" "cwl_compress_to_kinesis" {
  type        = "zip"
  source_dir  = "function/"
  output_path = "lambda_function.zip"
}

resource "aws_iam_role" "cwl-compress-to-kinesis-lambda-iam-role" {
  name               = "CWLCompressToKinesisLambdaRole"
  assume_role_policy = data.aws_iam_policy_document.principal_policy.json
}

resource "aws_iam_policy" "cwl-compress-to-kinesis-lambda-iam-policy" {
  name   = "CWLCompressToKinesisLambdaPolicy"
  policy = data.aws_iam_policy_document.cwl_compress_to_kinesis_lambda_iam_policy.json
}

resource "aws_iam_role_policy_attachment" "cwl-compress-to-kinesis-lambda-role-policy-attachment" {
  role       = aws_iam_role.cwl-compress-to-kinesis-lambda-iam-role.name
  policy_arn = aws_iam_policy.cwl-compress-to-kinesis-lambda-iam-policy.arn
}

resource "aws_lambda_function" "cwl-compress-to-firehose" {
  function_name = "cwl-compress-to-kinesis"
  handler       = "lambda_function.handler"
  role          = aws_iam_role.cwl-compress-to-kinesis-lambda-iam-role.iam_role.arn
  runtime       = "python3.8"

  filename         = data.archive_file.cwl_compress_to_kinesis.output_path
  source_code_hash = data.archive_file.cwl_compress_to_kinesis.output_base64sha256

  memory_size = 128
  timeout     = 180

  tags = {
    Name = "cwl-compress-to-kinesis"
  }
}

resource "aws_lambda_alias" "cwl-compress-to-firehose-prod-alias" {
  name             = "Prod"
  description      = "cwl-compress-to-kinesis Prod alias."
  function_name    = aws_lambda_function.cwl-compress-to-firehose.arn
  function_version = "$LATEST"

  lifecycle {
    ignore_changes = [function_version]
  }
}

output "function_arn" {
  value = aws_lambda_alias.cwl-compress-to-firehose-prod-alias.arn
}

provider "aws" {

region = "ap-northeast-1"

}

data "aws_caller_identity" "self" {}

data "aws_iam_policy_document" "principal_policy" {

statement {

actions = ["sts:AssumeRole"]

principals {

type = "Service"

identifiers = ["lambda.amazonaws.com"]

}

data "aws_iam_policy_document" "cwl_compress_to_kinesis_lambda_iam_policy" {

statement {

effect = "Allow"

actions = [

"firehose:PutRecord",

"firehose:PutRecordBatch"

]

resources = ["arn:aws:firehose:ap-northeast-1:${data.aws_caller_identity.self.account_id}:deliverystream/cwl-transfer-s3-sample-stream"]

}

statement {

effect = "Allow"

actions = ["logs:CreateLogGroup"]

resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:*"]

}

statement {

effect = "Allow"

actions = [

"logs:CreateLogStream",

"logs:PutLogEvents"

]

resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/lambda/cwl-compress-to-kinesis:*"]

}

data "archive_file" "cwl_compress_to_kinesis" {

type = "zip"

source_dir = "function/"

output_path = "lambda_function.zip"

}

resource "aws_iam_role" "cwl-compress-to-kinesis-lambda-iam-role" {

name = "CWLCompressToKinesisLambdaRole"

assume_role_policy = data.aws_iam_policy_document.principal_policy.json

}

resource "aws_iam_policy" "cwl-compress-to-kinesis-lambda-iam-policy" {

name = "CWLCompressToKinesisLambdaPolicy"

policy = data.aws_iam_policy_document.cwl_compress_to_kinesis_lambda_iam_policy.json

}

resource "aws_iam_role_policy_attachment" "cwl-compress-to-kinesis-lambda-role-policy-attachment" {

role = aws_iam_role.cwl-compress-to-kinesis-lambda-iam-role.name

policy_arn = aws_iam_policy.cwl-compress-to-kinesis-lambda-iam-policy.arn

}

resource "aws_lambda_function" "cwl-compress-to-firehose" {

function_name = "cwl-compress-to-kinesis"

handler = "lambda_function.handler"

role = aws_iam_role.cwl-compress-to-kinesis-lambda-iam-role.iam_role.arn

runtime = "python3.8"

filename = data.archive_file.cwl_compress_to_kinesis.output_path

source_code_hash = data.archive_file.cwl_compress_to_kinesis.output_base64sha256

memory_size = 128

timeout = 180

tags = {

Name = "cwl-compress-to-kinesis"

}

resource "aws_lambda_alias" "cwl-compress-to-firehose-prod-alias" {

name = "Prod"

description = "cwl-compress-to-kinesis Prod alias."

function_name = aws_lambda_function.cwl-compress-to-firehose.arn

function_version = "$LATEST"

lifecycle {

ignore_changes = [function_version]

}

output "function_arn" {

value = aws_lambda_alias.cwl-compress-to-firehose-prod-alias.arn

}

3. KinesisFirehoseの作成

processorに設定するLambdaができたので、次はKinesisのリソース作成です。

今回は、cwl-transfer-s3-sample-streamというストリームを作成しました。
上記LambdaのIAMポリシーで指定したFirehoseのリソース名と合わせる必要がある点に注意が必要です。

Processorに登録するLambdaを指定します。
今回は、Step2で作成したcwl-compress-to-kinesisが対象です。

Lambdaのバージョンは、バージョン番号でしか指定できないようなので最新の$LATESTを指定しています。
エイリアス名を指定できると理想ですが、今はまだできないようです。

転送先の指定です。
今回は転送先はS3で、転送先バケットはStep1で作成したcwl-transfer-slack-sample-bucketが対象です。

S3バケットの保存先パスです。
正常時の保存先にlambda/、エラー時の保存先にerror/lambda/を指定しました。

Firehoseに溜まったデータをS3へ転送するタイミングの指定です。
Firehoseに5MiB以上のデータが溜まるか、データが入ってから300秒経過時に転送するデフォルト設定を使用します。

また、S3の保存容量削減の為、転送するファイルは圧縮して保存させます。

Firehoseに設定するIAMロールの指定です。
予め作成しておいたCWLTransferS3FirehoseRoleを使用します。
設定したポリシー内容は後述。

Firehoseに設定したIAM Roleにアタッチしたポリシーは、以下になります。
[IAM ROLE(CWLTransferS3FirehoseRole).IAM Policy]

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "glue:GetTableVersions",
                "glue:GetTableVersion",
                "glue:GetTable"
            ],
            "Resource": "*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:ListBucketMultipartUploads",
                "s3:ListBucket",
                "s3:GetObject",
                "s3:GetBucketLocation",
                "s3:AbortMultipartUpload"
            ],
            "Resource": [
                "arn:aws:s3:::cwl-transfer-slack-sample-bucket/*",
                "arn:aws:s3:::cwl-transfer-slack-sample-bucket"
            ]
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction",
                "lambda:GetFunctionConfiguration"
            ],
            "Resource": "arn:aws:lambda:ap-northeast-1:{アカウントID}:function:cwl-compress-to-kinesis:Prod"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "logs:PutLogEvents",
            "Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/kinesisfirehose/*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "kinesis:ListShards",
                "kinesis:GetShardIterator",
                "kinesis:GetRecords",
                "kinesis:DescribeStream"
            ],
            "Resource": "arn:aws:kinesis:ap-northeast-1:{アカウントID}:stream/%FIREHOSE_STREAM_NAME%"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "kms:Decrypt",
            "Resource": "arn:aws:kms:ap-northeast-1:{アカウントID}:key/%SSE_KEY_ID%",
            "Condition": {
                "StringEquals": {
                    "kms:ViaService": "kinesis.ap-northeast-1.amazonaws.com"
                }
            }
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "kms:Decrypt",
            "Resource": "arn:aws:kms:ap-northeast-1:{アカウントID}:key/%SSE_KEY_ID%",
            "Condition": {
                "StringLike": {
                    "kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:ap-northeast-1:{アカウントID}:stream/%FIREHOSE_STREAM_NAME%"
                }
            }
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "",

"Effect": "Allow",

"Action": [

"glue:GetTableVersions",

"glue:GetTableVersion",

"glue:GetTable"

"Resource": "*"

{

"Sid": "",

"Effect": "Allow",

"Action": [

"s3:PutObject",

"s3:ListBucketMultipartUploads",

"s3:ListBucket",

"s3:GetObject",

"s3:GetBucketLocation",

"s3:AbortMultipartUpload"

"Resource": [

"arn:aws:s3:::cwl-transfer-slack-sample-bucket/*",

"arn:aws:s3:::cwl-transfer-slack-sample-bucket"

]

{

"Sid": "",

"Effect": "Allow",

"Action": [

"lambda:InvokeFunction",

"lambda:GetFunctionConfiguration"

"Resource": "arn:aws:lambda:ap-northeast-1:{アカウントID}:function:cwl-compress-to-kinesis:Prod"

{

"Sid": "",

"Effect": "Allow",

"Action": "logs:PutLogEvents",

"Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/kinesisfirehose/*"

{

"Sid": "",

"Effect": "Allow",

"Action": [

"kinesis:ListShards",

"kinesis:GetShardIterator",

"kinesis:GetRecords",

"kinesis:DescribeStream"

"Resource": "arn:aws:kinesis:ap-northeast-1:{アカウントID}:stream/%FIREHOSE_STREAM_NAME%"

{

"Sid": "",

"Effect": "Allow",

"Action": "kms:Decrypt",

"Resource": "arn:aws:kms:ap-northeast-1:{アカウントID}:key/%SSE_KEY_ID%",

"Condition": {

"StringEquals": {

"kms:ViaService": "kinesis.ap-northeast-1.amazonaws.com"

}

{

"Sid": "",

"Effect": "Allow",

"Action": "kms:Decrypt",

"Resource": "arn:aws:kms:ap-northeast-1:{アカウントID}:key/%SSE_KEY_ID%",

"Condition": {

"StringLike": {

"kms:EncryptionContext:aws:kinesis:arn": "arn:aws:kinesis:ap-northeast-1:{アカウントID}:stream/%FIREHOSE_STREAM_NAME%"

}

]

}

作成したKinesisFirehoseのterraformのコードは、以下になります。
記事の都合上最小記述に縮めていますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。

provider "aws" {
  region = "ap-northeast-1"
}

data "aws_caller_identity" "self" {}

data "terraform_remote_state" "s3" {
  backend = "local"

  config = {
    path = "{s3のリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"
  }
}

data "terraform_remote_state" "lambda" {
  backend = "local"

  config = {
    path = "{CWLの解凍用Lambdaのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"
  }
}

data "aws_iam_policy_document" "principal_policy" {
  statement {
    actions = ["sts:AssumeRole"]
 
    principals {
      type        = "Service"
      identifiers = ["firehose.amazonaws.com"]
    }
  }
}

data "aws_iam_policy_document" "cwl_transfer_s3_firehose_policy" {
  statement {
    effect = "Allow"
    actions = [
      "glue:GetTable",
      "glue:GetTableVersion",
      "glue:GetTableVersions"
    ]
    resources = ["*"]
  }

  statement {
    effect = "Allow"
    actions = [
      "s3:AbortMultipartUpload",
      "s3:GetBucketLocation",
      "s3:GetObject",
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
      "s3:PutObject"
    ]
    resources = [
      data.terraform_remote_state.s3.outputs.bucket_arn,
      "${data.terraform_remote_state.s3.outputs.bucket_arn}/*"
    ]
  }

  statement {
    effect = "Allow"
    actions = [
      "lambda:InvokeFunction",
      "lambda:GetFunctionConfiguration"
    ]
    resources = [data.terraform_remote_state.lambda.outputs.function_arn]
  }

  statement {
    effect    = "Allow"
    actions   = ["logs:PutLogEvents"]
    resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/kinesisfirehose/*"]
  }

  statement {
    effect = "Allow"
    actions = [
      "kinesis:DescribeStream",
      "kinesis:GetShardIterator",
      "kinesis:GetRecords",
      "kinesis:ListShards"
    ]
    resources = ["arn:aws:kinesis:ap-northeast-1:${data.aws_caller_identity.self.account_id}:stream/%FIREHOSE_STREAM_NAME%"]
  }

  statement {
    effect    = "Allow"
    actions   = ["kms:Decrypt"]
    resources = ["arn:aws:kms:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/%SSE_KEY_ID%"]
    condition {
      test     = "StringEquals"
      variable = "kms:ViaService"
      values   = ["kinesis.ap-northeast-1.amazonaws.com"]
    }
  }

  statement {
    effect    = "Allow"
    actions   = ["kms:Decrypt"]
    resources = ["arn:aws:kms:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/%SSE_KEY_ID%"]
    condition {
      test     = "StringLike"
      variable = "kms:EncryptionContext:aws:kinesis:arn"
      values   = ["arn:aws:kinesis:ap-northeast-1:${data.aws_caller_identity.self.account_id}:stream/%FIREHOSE_STREAM_NAME%"]
    }
  }
}

resource "aws_iam_role" "cwl-transfer-s3-firehose-iam-role" {
  name               = "CWLTransferS3FirehoseRole"
  assume_role_policy = data.aws_iam_policy_document.principal_policy.json
}
 
resource "aws_iam_policy" "cwl-transfer-s3-firehose-iam-policy" {
  name   = "CWLTransferS3FirehosePolicy"
  policy = data.aws_iam_policy_document.cwl_transfer_s3_firehose_policy.json
}
 
resource "aws_iam_role_policy_attachment" "cwl-transfer-s3-firehose-role-policy-attachment" {
  role       = aws_iam_role.cwl-transfer-s3-firehose-iam-role.name
  policy_arn = aws_iam_policy.cwl-transfer-s3-firehose-iam-policy.arn
}

resource "aws_cloudwatch_log_group" "firehose-cwl-log-group" {
  name = "/aws/kinesisfirehose/cwl-transfer-s3-sample-stream/"
}

resource "aws_cloudwatch_log_stream" "firehose-cwl-log-stream" {
  name           = "S3Delivery"
  log_group_name = aws_cloudwatch_log_group.firehose-cwl-log-group.name
}

resource "aws_kinesis_firehose_delivery_stream" "cwl-transfer-s3" {
  name        = "cwl-transfer-s3-sample-stream"
  destination = "extended_s3"
  extended_s3_configuration {
    role_arn            = aws_iam_role.cwl-transfer-s3-firehose-iam-role.arn
    bucket_arn          = data.terraform_remote_state.s3.outputs.bucket_arn
    prefix              = "lambda/"
    error_output_prefix = "error/lambda/"
    buffer_size         = 5
    buffer_interval     = 300
    compression_format  = "GZIP"
    processing_configuration {
      enabled = true

      processors {
        type = "Lambda"

        parameters {
          parameter_name  = "LambdaArn"
          parameter_value = data.terraform_remote_state.lambda.outputs.function_arn
        }
      }
    }
    cloudwatch_logging_options {
      enabled         = true
      log_group_name  = aws_cloudwatch_log_group.firehose-cwl-log-group.name
      log_stream_name = aws_cloudwatch_log_stream.firehose-cwl-log-stream.name
    }
    s3_backup_mode = "Disabled"
  }

  tags = {
    Name = "cwl-transfer-s3-sample-stream"
  }
}

output "stream_arn" {
  value = aws_kinesis_firehose_delivery_stream.cwl-transfer-s3.arn
}

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

provider "aws" {

region = "ap-northeast-1"

}

data "aws_caller_identity" "self" {}

data "terraform_remote_state" "s3" {

backend = "local"

config = {

path = "{s3のリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"

}

data "terraform_remote_state" "lambda" {

backend = "local"

config = {

path = "{CWLの解凍用Lambdaのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"

}

data "aws_iam_policy_document" "principal_policy" {

statement {

actions = ["sts:AssumeRole"]

principals {

type = "Service"

identifiers = ["firehose.amazonaws.com"]

}

data "aws_iam_policy_document" "cwl_transfer_s3_firehose_policy" {

statement {

effect = "Allow"

actions = [

"glue:GetTable",

"glue:GetTableVersion",

"glue:GetTableVersions"

]

resources = ["*"]

}

statement {

effect = "Allow"

actions = [

"s3:AbortMultipartUpload",

"s3:GetBucketLocation",

"s3:GetObject",

"s3:ListBucket",

"s3:ListBucketMultipartUploads",

"s3:PutObject"

]

resources = [

data.terraform_remote_state.s3.outputs.bucket_arn,

"${data.terraform_remote_state.s3.outputs.bucket_arn}/*"

]

}

statement {

effect = "Allow"

actions = [

"lambda:InvokeFunction",

"lambda:GetFunctionConfiguration"

]

resources = [data.terraform_remote_state.lambda.outputs.function_arn]

}

statement {

effect = "Allow"

actions = ["logs:PutLogEvents"]

resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/kinesisfirehose/*"]

}

statement {

effect = "Allow"

actions = [

"kinesis:DescribeStream",

"kinesis:GetShardIterator",

"kinesis:GetRecords",

"kinesis:ListShards"

]

resources = ["arn:aws:kinesis:ap-northeast-1:${data.aws_caller_identity.self.account_id}:stream/%FIREHOSE_STREAM_NAME%"]

}

statement {

effect = "Allow"

actions = ["kms:Decrypt"]

resources = ["arn:aws:kms:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/%SSE_KEY_ID%"]

condition {

test = "StringEquals"

variable = "kms:ViaService"

values = ["kinesis.ap-northeast-1.amazonaws.com"]

}

statement {

effect = "Allow"

actions = ["kms:Decrypt"]

resources = ["arn:aws:kms:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/%SSE_KEY_ID%"]

condition {

test = "StringLike"

variable = "kms:EncryptionContext:aws:kinesis:arn"

values = ["arn:aws:kinesis:ap-northeast-1:${data.aws_caller_identity.self.account_id}:stream/%FIREHOSE_STREAM_NAME%"]

}

resource "aws_iam_role" "cwl-transfer-s3-firehose-iam-role" {

name = "CWLTransferS3FirehoseRole"

assume_role_policy = data.aws_iam_policy_document.principal_policy.json

}

resource "aws_iam_policy" "cwl-transfer-s3-firehose-iam-policy" {

name = "CWLTransferS3FirehosePolicy"

policy = data.aws_iam_policy_document.cwl_transfer_s3_firehose_policy.json

}

resource "aws_iam_role_policy_attachment" "cwl-transfer-s3-firehose-role-policy-attachment" {

role = aws_iam_role.cwl-transfer-s3-firehose-iam-role.name

policy_arn = aws_iam_policy.cwl-transfer-s3-firehose-iam-policy.arn

}

resource "aws_cloudwatch_log_group" "firehose-cwl-log-group" {

name = "/aws/kinesisfirehose/cwl-transfer-s3-sample-stream/"

}

resource "aws_cloudwatch_log_stream" "firehose-cwl-log-stream" {

name = "S3Delivery"

log_group_name = aws_cloudwatch_log_group.firehose-cwl-log-group.name

}

resource "aws_kinesis_firehose_delivery_stream" "cwl-transfer-s3" {

name = "cwl-transfer-s3-sample-stream"

destination = "extended_s3"

extended_s3_configuration {

role_arn = aws_iam_role.cwl-transfer-s3-firehose-iam-role.arn

bucket_arn = data.terraform_remote_state.s3.outputs.bucket_arn

prefix = "lambda/"

error_output_prefix = "error/lambda/"

buffer_size = 5

buffer_interval = 300

compression_format = "GZIP"

processing_configuration {

enabled = true

processors {

type = "Lambda"

parameters {

parameter_name = "LambdaArn"

parameter_value = data.terraform_remote_state.lambda.outputs.function_arn

}

cloudwatch_logging_options {

enabled = true

log_group_name = aws_cloudwatch_log_group.firehose-cwl-log-group.name

log_stream_name = aws_cloudwatch_log_stream.firehose-cwl-log-stream.name

}

s3_backup_mode = "Disabled"

}

tags = {

Name = "cwl-transfer-s3-sample-stream"

}

output "stream_arn" {

value = aws_kinesis_firehose_delivery_stream.cwl-transfer-s3.arn

}

4. CloudWatch Logs.ロググループにサブスクリプションフィルタ作成

作成したKinesisFirehoseにCloudWatch Logsからエラーログを転送させます。
サブスクリプションフィルタを作成する対象は、ロググループです。

作成方法ですが、CloudWatch Logsからの転送先がLambda or Elasticsearch以外の場合はGUIからサブスクリプションフィルターの作成ができない為、作成はCLIからおこないます。

尚、転送する対象のCloudWatch Logsのロググループは予め作成されている前提で進めますが、ロググループがまだ存在しない場合は先にそちらを作成します。
今回は、output-cloudwatch-logs-sample-functionというLambdaの出力するロググループを転送する前提で進めていきます。

まずは、サブスクリプションフィルタがまだ存在していない事の確認です。

aws logs describe-subscription-filters \
  --log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function"
{
    "subscriptionFilters": []
}

aws logs describe-subscription-filters \

--log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function"

{

"subscriptionFilters": []

}

続いて、サブスクリプションフィルタを作成します。
今回はエラーログのみを転送したい要件だった為、フィルタパターンでエラーログ以外を除外しています。

aws logs put-subscription-filter \
  --log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function" \
  --filter-name "cwl-transfer-firehose-subscription-filter" \
  --filter-pattern "?Error ?ERROR" \
  --destination-arn "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream" \
  --role-arn "arn:aws:iam::{アカウントID}:role/CWLTransferFirehoseSubscriptionFilterRole"

aws logs put-subscription-filter \

--log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function" \

--filter-name "cwl-transfer-firehose-subscription-filter" \

--filter-pattern "?Error ?ERROR" \

--destination-arn "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream" \

--role-arn "arn:aws:iam::{アカウントID}:role/CWLTransferFirehoseSubscriptionFilterRole"

CWLのロググループには予め用意しておいたCWLTransferFirehoseSubscriptionFilterRoleのIAM Roleを設定していますが、同ロールにアタッチしたポリシーは、以下になります。
[IAM ROLE(CWLTransferFirehoseSubscriptionFilterRole).IAM Policy]

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "firehose:PutRecordBatch",
                "firehose:PutRecord"
            ],
            "Resource": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "",

"Effect": "Allow",

"Action": [

"firehose:PutRecordBatch",

"firehose:PutRecord"

"Resource": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream"

}

]

}

もう一度サブスクリプションフィルタを確認すると、正常に作成できているのが分かります。

aws logs describe-subscription-filters \
  --log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function"
{
    "subscriptionFilters": [
        {
            "filterName": "cwl-transfer-firehose-subscription-filter",
            "logGroupName": "/aws/lambda/output-cloudwatch-logs-sample-function",
            "filterPattern": "?Error ?ERROR",
            "destinationArn": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream",
            "roleArn": "arn:aws:iam::{アカウントID}:role/CWLTransferFirehoseSubscriptionFilterRole",
            "distribution": "ByLogStream",
            "creationTime": 1609425714435
        }
    ]
}

aws logs describe-subscription-filters \

--log-group-name "/aws/lambda/output-cloudwatch-logs-sample-function"

{

"subscriptionFilters": [

{

"filterName": "cwl-transfer-firehose-subscription-filter",

"logGroupName": "/aws/lambda/output-cloudwatch-logs-sample-function",

"filterPattern": "?Error ?ERROR",

"destinationArn": "arn:aws:firehose:ap-northeast-1:{アカウントID}:deliverystream/cwl-transfer-s3-sample-stream",

"roleArn": "arn:aws:iam::{アカウントID}:role/CWLTransferFirehoseSubscriptionFilterRole",

"distribution": "ByLogStream",

"creationTime": 1609425714435

}

]

}

作成したサブスクリプションフィルタのterraformのコードは、以下になります。
記事の都合上最小記述に縮めていますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。

provider "aws" {
  region = "ap-northeast-1"
}

data "terraform_remote_state" "firehose" {
  backend = "local"

  config = {
    path = "{firehoseのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"
  }
}

data "aws_iam_policy_document" "principal_policy" {
  statement {
    actions = ["sts:AssumeRole"]
 
    principals {
      type        = "Service"
      identifiers = ["logs.ap-northeast-1.amazonaws.com"]
    }
  }
}

data "aws_iam_policy_document" "cwl_transfer_firehose_subscription_filter_policy" {
  statement {
    effect = "Allow"
    actions = [
      "firehose:PutRecord",
      "firehose:PutRecordBatch"
    ]
    resources = [data.terraform_remote_state.firehose.outputs.stream_arn]
  }
}

resource "aws_iam_role" "cwl-transfer-firehose-subscription-filter-iam-role" {
  name               = "CWLTransferFirehoseSubscriptionFilterRole"
  assume_role_policy = data.aws_iam_policy_document.principal_policy.json
}
 
resource "aws_iam_policy" "cwl-transfer-firehose-subscription-filter-iam-policy" {
  name   = "CWLTransferFirehoseSubscriptionFilterPolicy"
  policy = data.aws_iam_policy_document.cwl_transfer_firehose_subscription_filter_policy.json
}
 
resource "aws_iam_role_policy_attachment" "cwl-transfer-firehose-subscription-filter-role-policy-attachment" {
  role       = aws_iam_role.cwl-transfer-firehose-subscription-filter-iam-role.name
  policy_arn = aws_iam_policy.cwl-transfer-firehose-subscription-filter-iam-policy.arn
}

resource "aws_cloudwatch_log_subscription_filter" "cwl-transfer-firehose" {
  name            = "cwl-transfer-firehose-subscription-filter"
  role_arn        = aws_iam_role.cwl-transfer-firehose-subscription-filter-iam-role.arn
  log_group_name  = "/aws/lambda/output-cloudwatch-logs-sample-function"
  filter_pattern  = "?Error ?ERROR"
  destination_arn = data.terraform_remote_state.firehose.outputs.stream_arn
  distribution    = "ByLogStream"
}

provider "aws" {

region = "ap-northeast-1"

}

data "terraform_remote_state" "firehose" {

backend = "local"

config = {

path = "{firehoseのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"

}

data "aws_iam_policy_document" "principal_policy" {

statement {

actions = ["sts:AssumeRole"]

principals {

type = "Service"

identifiers = ["logs.ap-northeast-1.amazonaws.com"]

}

data "aws_iam_policy_document" "cwl_transfer_firehose_subscription_filter_policy" {

statement {

effect = "Allow"

actions = [

"firehose:PutRecord",

"firehose:PutRecordBatch"

]

resources = [data.terraform_remote_state.firehose.outputs.stream_arn]

}

resource "aws_iam_role" "cwl-transfer-firehose-subscription-filter-iam-role" {

name = "CWLTransferFirehoseSubscriptionFilterRole"

assume_role_policy = data.aws_iam_policy_document.principal_policy.json

}

resource "aws_iam_policy" "cwl-transfer-firehose-subscription-filter-iam-policy" {

name = "CWLTransferFirehoseSubscriptionFilterPolicy"

policy = data.aws_iam_policy_document.cwl_transfer_firehose_subscription_filter_policy.json

}

resource "aws_iam_role_policy_attachment" "cwl-transfer-firehose-subscription-filter-role-policy-attachment" {

role = aws_iam_role.cwl-transfer-firehose-subscription-filter-iam-role.name

policy_arn = aws_iam_policy.cwl-transfer-firehose-subscription-filter-iam-policy.arn

}

resource "aws_cloudwatch_log_subscription_filter" "cwl-transfer-firehose" {

name = "cwl-transfer-firehose-subscription-filter"

role_arn = aws_iam_role.cwl-transfer-firehose-subscription-filter-iam-role.arn

log_group_name = "/aws/lambda/output-cloudwatch-logs-sample-function"

filter_pattern = "?Error ?ERROR"

destination_arn = data.terraform_remote_state.firehose.outputs.stream_arn

distribution = "ByLogStream"

}

5. ログ転送先SlackチャネルのIncomming Webhook URL取得 / パラメータストアへの保存

CloudWatch Logsに出力された内容をS3にファイル出力するところまで貫通したので、最後のSlack転送に入ります。
が、その前にSlackに通知を送るにはIncommingWebhookのURLを取得する必要があるので、先にこちらを済ませます。

ログ転送先のワークスペース & チャンネル指定

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

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

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

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

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

Integration Settings

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

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

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

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

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

今回は、cwl-transfer-slack-sample-webhook-urlのキーで作成しました。
クレデンシャル情報なので種別はSecure String、利用枠は標準です。

terraformのコードは以下です。
記事の都合上最小記述にしますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。

provider "aws" {
  region = "ap-northeast-1"
}

resource "aws_ssm_parameter" "slack-incomming-webhook-url" {
  name            = "cwl-transfer-slack-sample-webhook-url"
  type            = "SecureString"
  value           = "dummy"
  description     = "cwl-transfer-slack sample incomming webhook url."
  tier            = "Standard"
  overwrite       = false
  allowed_pattern = ""

  tags = {
    Name = "cwl-transfer-slack-sample-webhook-url"
  }

  lifecycle {
    ignore_changes = [value]
  }
}

output "parameter_name" {
  value = aws_ssm_parameter.slack-incomming-webhook-url.name
}

provider "aws" {

region = "ap-northeast-1"

}

resource "aws_ssm_parameter" "slack-incomming-webhook-url" {

name = "cwl-transfer-slack-sample-webhook-url"

type = "SecureString"

value = "dummy"

description = "cwl-transfer-slack sample incomming webhook url."

tier = "Standard"

overwrite = false

allowed_pattern = ""

tags = {

Name = "cwl-transfer-slack-sample-webhook-url"

}

lifecycle {

ignore_changes = [value]

}

output "parameter_name" {

value = aws_ssm_parameter.slack-incomming-webhook-url.name

}

コード管理上は適当な値で一時作成してライフサイクル管理から外す形を取るので、作成後に保存しておいたIncomming WebhookのURLに上書きします。

6. S3へのファイルputをトリガーに発火するSlackに転送用Lambda作成

S3に保存されたエラーログ本文をSlackに転送するLambdaを作成します。

S3にオブジェクトが置かれたタイミングで発火するLambdaもブループリントが用意されているので、設計図を使用して作成していきます。

今回は、cwl-transfer-slackという関数を作成しました。
対象バケットは、今回作成したcwl-transfer-slack-sample-bucketを設定します。
タイプは全オブジェクト作成で、prefixにlambda、suffixに.gzを設定します。
prefix/suffixの指定は、対象バケットのオブジェクト全てではなくFirehoseから転送されたオブジェクトだけを対象にする為に設定しています。

環境変数の設定です。
S3とSSMのリージョン名の二つに加えて、先程作成したパラメータストアのキー名を設定します。

タイムアウト設定がデフォルトだと3秒で短いので、少し伸ばします。

今回もLambdaに設定するIAM RoleとしてCWLTransferSlackLambdaRoleを予め用意しており、アタッチ済のポリシーは以下になります。
[IAM ROLE(CWLTransferSlackLambdaRole).IAM Policy]

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::cwl-transfer-slack-sample-bucket/*",
                "arn:aws:s3:::cwl-transfer-slack-sample-bucket"
            ]
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "ssm:GetParameter",
            "Resource": "arn:aws:ssm:ap-northeast-1:{アカウントID}:parameter/cwl-transfer-slack-sample-webhook-url"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "kms:Decrypt",
            "Resource": "arn:aws:ssm:ap-northeast-1:{アカウントID}:key/*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:*"
        },
        {
            "Sid": "",
            "Effect": "Allow",
            "Action": [
                "logs:PutLogEvents",
                "logs:CreateLogStream"
            ],
            "Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/lambda/cwl-transfer-slack:*"
        }
    ]
}

{

"Version": "2012-10-17",

"Statement": [

{

"Sid": "",

"Effect": "Allow",

"Action": [

"s3:ListBucket",

"s3:GetObject"

"Resource": [

"arn:aws:s3:::cwl-transfer-slack-sample-bucket/*",

"arn:aws:s3:::cwl-transfer-slack-sample-bucket"

]

{

"Sid": "",

"Effect": "Allow",

"Action": "ssm:GetParameter",

"Resource": "arn:aws:ssm:ap-northeast-1:{アカウントID}:parameter/cwl-transfer-slack-sample-webhook-url"

{

"Sid": "",

"Effect": "Allow",

"Action": "kms:Decrypt",

"Resource": "arn:aws:ssm:ap-northeast-1:{アカウントID}:key/*"

{

"Sid": "",

"Effect": "Allow",

"Action": "logs:CreateLogGroup",

"Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:*"

{

"Sid": "",

"Effect": "Allow",

"Action": [

"logs:PutLogEvents",

"logs:CreateLogStream"

"Resource": "arn:aws:logs:ap-northeast-1:{アカウントID}:log-group:/aws/lambda/cwl-transfer-slack:*"

}

]

}

実行コードを以下の感じに更新して完了です。
あくまでもサンプルコードなので適当に変更してください。

import boto3
import os
import logging
import time
import urllib.request
import urllib.parse
import json
import gzip
import io

## environment variable key name
S3_REGION_NAME = "S3_REGION_NAME"
SSM_REGION_NAME = "SSM_REGION_NAME"
SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME = "SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME"

## const value.
PRETEXT_TEMPLATE = "【---------- CWL_LOG START ----------】"
MESSAGE_BORDER_COLOR = "#FF0000"
OUTPUT_MAX_LINE_NUM = 50
HTTP_METHOD_NAME = "POST"

## singleton object
logger = None
s3 = None
ssm = None
webhook_url = None


def get_ssm_parameter_store_value(ssm, parameter_name, with_decryption):
    response = ssm.get_parameter(
        Name=parameter_name,
        WithDecryption=with_decryption
    )
    return response["Parameter"]["Value"]

def initialize_singleton_object():
    """
    Initialize singleton object.
    """
    global logger
    if not logger:
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)

    global s3
    if not s3:
        s3 = boto3.resource("s3", region_name=os.environ[S3_REGION_NAME])

    global ssm
    if not ssm:
        ssm = boto3.client("ssm", region_name=os.environ[SSM_REGION_NAME])

    global webhook_url
    if not webhook_url:
        webhook_url = get_ssm_parameter_store_value(
            ssm,
            os.environ[SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME],
            True
        )

def create_send_data(transfer_logs):
    send_data = "payload=" + json.dumps(
        {
            "attachments": [
                {
                    "fallback": PRETEXT_TEMPLATE,
                    "pretext": PRETEXT_TEMPLATE,
                    "color": MESSAGE_BORDER_COLOR,
                    "text": transfer_logs,
                    "mrkdwn_in": ["text"]
                }
            ]
        }
    )
    return send_data.encode("utf-8")

def post_request_to_slack(webhook_url, transfer_logs):
    request = urllib.request.Request(
        webhook_url,
        data=create_send_data(transfer_logs),
        method=HTTP_METHOD_NAME,
    )
    with urllib.request.urlopen(request) as response:
        return response.read().decode("utf-8")

def get_s3_object(s3_resource, bucket_name, file_path):
    return s3_resource.Object(bucket_name, file_path).get()

def unzip_compressed_object(compressed_object):
    return gzip.open(io.BytesIO(compressed_object), 'rt')

def cwl_transfer_slack(s3, bucket_name, file_path, webhook_url, logger):
    transfer_logs = ""
    exist_fail_request = False

    s3_object = get_s3_object(s3, bucket_name, file_path)
    cwl_content = unzip_compressed_object(s3_object["Body"].read())
    logs_line = cwl_content.read().split("\n")
    for i in range(len(logs_line)):
        log = logs_line[i].replace("'", "\\'").replace("&", "%26")
        if len(log) > 0:
            transfer_logs = transfer_logs + log + "\n"

        if i > 0 and i % OUTPUT_MAX_LINE_NUM == 0:
            try:
                post_request_to_slack(webhook_url, transfer_logs)
                logger.info(
                    "post_request_to_slack successed. transfer_logs = "
                    + transfer_logs
                )
            except Exception as e:
                logger.exception(
                    "post_request_to_slack failed. transfer_logs = "
                    + transfer_logs
                )
                exist_fail_request = True

            transfer_logs = ""
            time.sleep(3)

    if len(transfer_logs) > 0:
        try:
            post_request_to_slack(webhook_url, transfer_logs)
            logger.info(
                "post_request_to_slack successed. transfer_logs = "
                + transfer_logs
            )
        except Exception as e:
            logger.exception(
                "post_request_to_slack failed. transfer_logs = "
                + transfer_logs
            )
            exist_fail_request = True

    return exist_fail_request

def lambda_handler(event, context):
    initialize_singleton_object()

    bucket_name = event["Records"][0]["s3"]["bucket"]["name"]
    file_path = urllib.parse.unquote(event["Records"][0]["s3"]['object']['key'])
    logger.info(
        "cwl_transfer_slack start. bucket_name = "
        + bucket_name
        + " file_path = "
        + file_path
    )
    exist_fail_request = cwl_transfer_slack(
        s3, bucket_name, file_path, webhook_url, logger
    )
    if exist_fail_request:
        raise Exception("Exist fail request into cwl transfer slack.")

    logger.info(
        "cwl transfer slack successed. bucket_name = "
        + bucket_name
        + " file_path = "
        + file_path
    )

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

import boto3

import os

import logging

import time

import urllib.request

import urllib.parse

import json

import gzip

import io

## environment variable key name

S3_REGION_NAME = "S3_REGION_NAME"

SSM_REGION_NAME = "SSM_REGION_NAME"

SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME = "SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME"

## const value.

PRETEXT_TEMPLATE = "【---------- CWL_LOG START ----------】"

MESSAGE_BORDER_COLOR = "#FF0000"

OUTPUT_MAX_LINE_NUM = 50

HTTP_METHOD_NAME = "POST"

## singleton object

logger = None

s3 = None

ssm = None

webhook_url = None

def get_ssm_parameter_store_value(ssm, parameter_name, with_decryption):

response = ssm.get_parameter(

Name=parameter_name,

WithDecryption=with_decryption

)

return response["Parameter"]["Value"]

def initialize_singleton_object():

"""

Initialize singleton object.

"""

global logger

if not logger:

logger = logging.getLogger()

logger.setLevel(logging.INFO)

global s3

if not s3:

s3 = boto3.resource("s3", region_name=os.environ[S3_REGION_NAME])

global ssm

if not ssm:

ssm = boto3.client("ssm", region_name=os.environ[SSM_REGION_NAME])

global webhook_url

if not webhook_url:

webhook_url = get_ssm_parameter_store_value(

ssm,

os.environ[SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME],

True

)

def create_send_data(transfer_logs):

send_data = "payload=" + json.dumps(

{

"attachments": [

{

"fallback": PRETEXT_TEMPLATE,

"pretext": PRETEXT_TEMPLATE,

"color": MESSAGE_BORDER_COLOR,

"text": transfer_logs,

"mrkdwn_in": ["text"]

}

]

}

)

return send_data.encode("utf-8")

def post_request_to_slack(webhook_url, transfer_logs):

request = urllib.request.Request(

webhook_url,

data=create_send_data(transfer_logs),

method=HTTP_METHOD_NAME,

)

with urllib.request.urlopen(request) as response:

return response.read().decode("utf-8")

def get_s3_object(s3_resource, bucket_name, file_path):

return s3_resource.Object(bucket_name, file_path).get()

def unzip_compressed_object(compressed_object):

return gzip.open(io.BytesIO(compressed_object), 'rt')

def cwl_transfer_slack(s3, bucket_name, file_path, webhook_url, logger):

transfer_logs = ""

exist_fail_request = False

s3_object = get_s3_object(s3, bucket_name, file_path)

cwl_content = unzip_compressed_object(s3_object["Body"].read())

logs_line = cwl_content.read().split("\n")

for i in range(len(logs_line)):

log = logs_line[i].replace("'", "\\'").replace("&", "%26")

if len(log) > 0:

transfer_logs = transfer_logs + log + "\n"

if i > 0 and i % OUTPUT_MAX_LINE_NUM == 0:

try:

post_request_to_slack(webhook_url, transfer_logs)

logger.info(

"post_request_to_slack successed. transfer_logs = "

+ transfer_logs

)

except Exception as e:

logger.exception(

"post_request_to_slack failed. transfer_logs = "

+ transfer_logs

)

exist_fail_request = True

transfer_logs = ""

time.sleep(3)

if len(transfer_logs) > 0:

try:

post_request_to_slack(webhook_url, transfer_logs)

logger.info(

"post_request_to_slack successed. transfer_logs = "

+ transfer_logs

)

except Exception as e:

logger.exception(

"post_request_to_slack failed. transfer_logs = "

+ transfer_logs

)

exist_fail_request = True

return exist_fail_request

def lambda_handler(event, context):

initialize_singleton_object()

bucket_name = event["Records"][0]["s3"]["bucket"]["name"]

file_path = urllib.parse.unquote(event["Records"][0]["s3"]['object']['key'])

logger.info(

"cwl_transfer_slack start. bucket_name = "

+ bucket_name

+ " file_path = "

+ file_path

)

exist_fail_request = cwl_transfer_slack(

s3, bucket_name, file_path, webhook_url, logger

)

if exist_fail_request:

raise Exception("Exist fail request into cwl transfer slack.")

logger.info(

"cwl transfer slack successed. bucket_name = "

+ bucket_name

+ " file_path = "

+ file_path

)

最後に、Lambda作成部分をterraformで書いたのが以下になります。
記事の都合上最小記述に縮めていますが、値の変数化や責務に応じたファイル分割などを適宜行って下さい。
また、Lambda functionの実行コードも上記と同様の為割合します。

provider "aws" {
  region = "ap-northeast-1"
}

data "aws_caller_identity" "self" {}

data "terraform_remote_state" "ssm_slack_webhook_url" {
  backend = "local"

  config = {
    path = "{ssm.parameter-storeのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}e"
  }
}

data "terraform_remote_state" "s3" {
  backend = "local"

  config = {
    path = "{s3のリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"
  }
}

data "aws_iam_policy_document" "principal_policy" {
  statement {
    actions = ["sts:AssumeRole"]
 
    principals {
      type        = "Service"
      identifiers = ["lambda.amazonaws.com"]
    }
  }
}

data "aws_iam_policy_document" "cwl_transfer_slack_lambda_iam_policy" {
  statement {
    effect = "Allow"
    actions = [
      "s3:ListBucket",
      "s3:GetObject"
    ]
    resources = [
      data.terraform_remote_state.s3.outputs.bucket_arn,
      "${data.terraform_remote_state.s3.outputs.bucket_arn}/*"
    ]
  }

  statement {
    effect    = "Allow"
    actions   = ["ssm:GetParameter"]
    resources = ["arn:aws:ssm:ap-northeast-1:${data.aws_caller_identity.self.account_id}:parameter/${data.terraform_remote_state.ssm_slack_webhook_url.outputs.parameter_name}"]
  }

  statement {
    effect    = "Allow"
    actions   = ["kms:Decrypt"]
    resources = ["arn:aws:ssm:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/*"]
  }

  statement {
    effect    = "Allow"
    actions   = ["logs:CreateLogGroup"]
    resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:*"]
  }

  statement {
    effect = "Allow"
    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents"
    ]
    resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/lambda/cwl-transfer-slack:*"]
  }
}

data "archive_file" "cwl_transfer_slack" {
  type        = "zip"
  source_dir  = "function/"
  output_path = "lambda_function.zip"
}

resource "aws_iam_role" "cwl-transfer-slack-iam-role" {
  name               = "CWLTransferSlackLambdaRole"
  assume_role_policy = data.aws_iam_policy_document.principal_policy.json
}
 
resource "aws_iam_policy" "cwl-transfer-slack-iam-policy" {
  name   = "CWLTransferSlackLambdaPolicy"
  policy = data.aws_iam_policy_document.cwl_transfer_slack_lambda_iam_policy.json
}
 
resource "aws_iam_role_policy_attachment" "cwl-transfer-slack-role-policy-attachment" {
  role       = aws_iam_role.cwl-transfer-slack-iam-role.name
  policy_arn = aws_iam_policy.cwl-transfer-slack-iam-policy.arn
}

resource "aws_lambda_function" "cwl-transfer-slack" {
  function_name = "cwl-transfer-slack"
  handler       = "lambda_function.lambda_handler"
  role          = aws_iam_role.cwl-transfer-slack-iam-role.arn
  runtime       = "python3.8"

  filename         = data.archive_file.cwl_transfer_slack.output_path
  source_code_hash = data.archive_file.cwl_transfer_slack.output_base64sha256

  memory_size = 128
  timeout     = 300

  environment {
    variables = {
      S3_REGION_NAME                       = "ap-northeast-1"
      SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME = data.terraform_remote_state.ssm_slack_webhook_url.outputs.parameter_name
      SSM_REGION_NAME                      = "ap-northeast-1"
    }
  }

  tags = {
    Name = "cwl-transfer-slack"
  }
}

resource "aws_lambda_alias" "cwl-transfer-slack-prod-alias" {
  name             = "Prod"
  description      = "cwl-transfer-slack Prod alias."
  function_name    = aws_lambda_function.cwl-transfer-slack.arn
  function_version = "$LATEST"

  lifecycle {
    ignore_changes = [function_version]
  }
}

resource "aws_lambda_permission" "cwl-transfer-slack-resource-policy" {
  function_name  = aws_lambda_function.cwl-transfer-slack.function_name
  qualifier      = aws_lambda_alias.cwl-transfer-slack-prod-alias.name
  statement_id   = "AllowExecutionFromS3Bucket"
  action         = "lambda:InvokeFunction"
  principal      = "s3.amazonaws.com"
  source_account = data.aws_caller_identity.self.account_id
  source_arn     = data.terraform_remote_state.s3.outputs.bucket_arn
}

resource "aws_s3_bucket_notification" "cwl-transfer-slack-event-trigger" {
  bucket = data.terraform_remote_state.s3.outputs.bucket_name

  lambda_function {
    lambda_function_arn = aws_lambda_alias.cwl-transfer-slack-prod-alias.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "lambda"
    filter_suffix       = ".gz"
  }
}

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

provider "aws" {

region = "ap-northeast-1"

}

data "aws_caller_identity" "self" {}

data "terraform_remote_state" "ssm_slack_webhook_url" {

backend = "local"

config = {

path = "{ssm.parameter-storeのリソース作成時のoutput出力ディレクトリのtfstateファイルパス}e"

}

data "terraform_remote_state" "s3" {

backend = "local"

config = {

path = "{s3のリソース作成時のoutput出力ディレクトリのtfstateファイルパス}"

}

data "aws_iam_policy_document" "principal_policy" {

statement {

actions = ["sts:AssumeRole"]

principals {

type = "Service"

identifiers = ["lambda.amazonaws.com"]

}

data "aws_iam_policy_document" "cwl_transfer_slack_lambda_iam_policy" {

statement {

effect = "Allow"

actions = [

"s3:ListBucket",

"s3:GetObject"

]

resources = [

data.terraform_remote_state.s3.outputs.bucket_arn,

"${data.terraform_remote_state.s3.outputs.bucket_arn}/*"

]

}

statement {

effect = "Allow"

actions = ["ssm:GetParameter"]

resources = ["arn:aws:ssm:ap-northeast-1:${data.aws_caller_identity.self.account_id}:parameter/${data.terraform_remote_state.ssm_slack_webhook_url.outputs.parameter_name}"]

}

statement {

effect = "Allow"

actions = ["kms:Decrypt"]

resources = ["arn:aws:ssm:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/*"]

}

statement {

effect = "Allow"

actions = ["logs:CreateLogGroup"]

resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:*"]

}

statement {

effect = "Allow"

actions = [

"logs:CreateLogStream",

"logs:PutLogEvents"

]

resources = ["arn:aws:logs:ap-northeast-1:${data.aws_caller_identity.self.account_id}:log-group:/aws/lambda/cwl-transfer-slack:*"]

}

data "archive_file" "cwl_transfer_slack" {

type = "zip"

source_dir = "function/"

output_path = "lambda_function.zip"

}

resource "aws_iam_role" "cwl-transfer-slack-iam-role" {

name = "CWLTransferSlackLambdaRole"

assume_role_policy = data.aws_iam_policy_document.principal_policy.json

}

resource "aws_iam_policy" "cwl-transfer-slack-iam-policy" {

name = "CWLTransferSlackLambdaPolicy"

policy = data.aws_iam_policy_document.cwl_transfer_slack_lambda_iam_policy.json

}

resource "aws_iam_role_policy_attachment" "cwl-transfer-slack-role-policy-attachment" {

role = aws_iam_role.cwl-transfer-slack-iam-role.name

policy_arn = aws_iam_policy.cwl-transfer-slack-iam-policy.arn

}

resource "aws_lambda_function" "cwl-transfer-slack" {

function_name = "cwl-transfer-slack"

handler = "lambda_function.lambda_handler"

role = aws_iam_role.cwl-transfer-slack-iam-role.arn

runtime = "python3.8"

filename = data.archive_file.cwl_transfer_slack.output_path

source_code_hash = data.archive_file.cwl_transfer_slack.output_base64sha256

memory_size = 128

timeout = 300

environment {

variables = {

S3_REGION_NAME = "ap-northeast-1"

SSM_SLACK_WEBHOOK_URL_PARAMETER_NAME = data.terraform_remote_state.ssm_slack_webhook_url.outputs.parameter_name

SSM_REGION_NAME = "ap-northeast-1"

}

tags = {

Name = "cwl-transfer-slack"

}

resource "aws_lambda_alias" "cwl-transfer-slack-prod-alias" {

name = "Prod"

description = "cwl-transfer-slack Prod alias."

function_name = aws_lambda_function.cwl-transfer-slack.arn

function_version = "$LATEST"

lifecycle {

ignore_changes = [function_version]

}

resource "aws_lambda_permission" "cwl-transfer-slack-resource-policy" {

function_name = aws_lambda_function.cwl-transfer-slack.function_name

qualifier = aws_lambda_alias.cwl-transfer-slack-prod-alias.name

statement_id = "AllowExecutionFromS3Bucket"

action = "lambda:InvokeFunction"

principal = "s3.amazonaws.com"

source_account = data.aws_caller_identity.self.account_id

source_arn = data.terraform_remote_state.s3.outputs.bucket_arn

}

resource "aws_s3_bucket_notification" "cwl-transfer-slack-event-trigger" {

bucket = data.terraform_remote_state.s3.outputs.bucket_name

lambda_function {

lambda_function_arn = aws_lambda_alias.cwl-transfer-slack-prod-alias.arn

events = ["s3:ObjectCreated:*"]

filter_prefix = "lambda"

filter_suffix = ".gz"

}

動作確認

一通りできたところで動かせてみます。
今回の一番最初の起点は、Lambda funtion.output-cloudwatch-logs-sample-function実行時に出力されるCloudWatch Logsだった為、このfunctionで適当にエラーログを吐かせて数分待っていると、通知受信場所のSlackチャネルに以下のような通知が転送されてくれば成功です。

以上で今回は全て終わりです！
冒頭にも書いた通り、今回の構成を作成する為のサンプルコードをGitHubに置いたのでよければご参照ください。
MasakiMisawa/cwl-transfer-slack-sample

Pocket

public memo

エンジニア向け小ネタ書き溜め用。公開日記だけど親切な文章とは程遠いかもしれない。

CloudWatch Logsに出力されたエラーログ本文のSlackへの転送

やりたいこと

1. 想定外の挙動があった場合に何が起こっていたのか確認したい

2. 確認に手間をかけたくない

3. 確認の為の仕組み作成にコストをかけたくない

4. しっかりした監視の仕組み作成時に作った内容を流用したい

全体アーキテクチャ

作ったもの

作成方法

1. 転送先のS3バケット作成

2. Kinesisに設定するCWLの解凍用Lambda作成

3. KinesisFirehoseの作成

4. CloudWatch Logs.ロググループにサブスクリプションフィルタ作成

5. ログ転送先SlackチャネルのIncomming Webhook URL取得 / パラメータストアへの保存

6. S3へのファイルputをトリガーに発火するSlackに転送用Lambda作成

動作確認