さくっとAWSのログをElastic Cloudに送る方法

この記事について

AWSのログをElastic Stackに送りたいけど、具体的にどうすればいいのか？　
AWSのログと言っても、S3にあったり、CloudWatch Logsにあったりなど様々で、一つならともかく、複数を別のところに転送するのは一苦労です。
AWSのログやデータをElastic Cloudに送る方法は主なものを上げると3つあります。

• Elastic Agentを使う
• Elastic Serverless Forwarder for AWSを使う
• Amazone Kinesis Data Firehoseを使う

(その他Filebeat, Fluentdを使うなども方法としてあると思います）

今回は、Elastic Serverless Forwarderを使い、関連するAWS側のコンポーネントのセットアップも一緒にTerraformでさくっとログをElastic Cloudに転送する方法を紹介します。

他のクラウド版の記事と含め最終的にはこのように3つのクラウドからログを集めることができます。

何のログを今回Elastic Cloudに送るか?

今回は以下の2つについてElastic Stackに転送してみます。

• 監査ログ （CloudTrailログ）
• アプリケーションログ （CloudWatchに入ってきているアプリケーションログ）

アーキテクチャー概要

以下のURLにアーキテクチャー図があります。
https://www.elastic.co/guide/en/esf/current/aws-elastic-serverless-forwarder.html

Elastic Serverless Forwarderは実態はLambdaですが、作成においてはSAR(AWS Serverless Application Repository)によってCloudFormationがキックされます。

今回のデータフロー

• CloudTrail証跡 -> S3 -> S3のSQS通知 -> Elastic Serverless Forwarder -> Elastic Cloud
• アプリケーションログ -> CloudWatch Logss -> Elastic Serverless Forwarder -> Elastic Cloud

手順

前提

• TerraformでAWSにアクセスする部分は事前にセットアップしてください。今回の記事において、フル権限のユーザーで行なっておりますが本当は権限絞った方がいいです。
• 既にCloudTrailの証跡がS3に保存される設定がされています。そのS3 bucketはvariable audit-log-s3-arn/audit-log-s3-nameで指定してください。
• 既に転送したいアプリケーションログがCloudWatch Logsに送られる設定がされています。そのロググループはvariable application-log-cw-loggroup-arnで指定してください。
• CloudTrailのS3と、CloudWatch Logsは同じリージョンにある前提での手順となっています。

Step1. 以下のmain.tfを作ってください。

main.tf

terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

variable "region" {}
variable "namespace" {}
variable "elastic-cloud-id" {}
variable "elastic-cloud-apikey" {}
variable "application-log-cw-loggroup-arn" {}
variable "audit-log-s3-name" {}
variable "audit-log-s3-arn" {}

provider "aws" {
  region = var.region
}

data "aws_serverlessapplicationrepository_application" "esf_sar" {
    # この値はこのままで使います。
    application_id = "arn:aws:serverlessrepo:eu-central-1:267093732750:applications/elastic-serverless-forwarder"
}

resource "aws_serverlessapplicationrepository_cloudformation_stack" "esf_application_logs" {
    name             = "${var.namespace}-app-elastic-serverless-forwarder"
    application_id   = data.aws_serverlessapplicationrepository_application.esf_sar.application_id
    semantic_version = data.aws_serverlessapplicationrepository_application.esf_sar.semantic_version
    capabilities     = data.aws_serverlessapplicationrepository_application.esf_sar.required_capabilities
    parameters = {
        ElasticServerlessForwarderS3ConfigFile = "s3://${aws_s3_object.esf-config.bucket}/${aws_s3_object.esf-config.key}"
        ElasticServerlessForwarderSSMSecrets = ""
        ElasticServerlessForwarderKMSKeys = ""
        ElasticServerlessForwarderSQSEvents = ""
        # CloudTrail証跡のS3の通知SQSのARNを指定
        ElasticServerlessForwarderS3SQSEvents = "${aws_sqs_queue.audit_log_s3_queue.arn}"
        ElasticServerlessForwarderKinesisEvents = ""
        # 転送するアプリケーションログのロググループARNを指定
        ElasticServerlessForwarderCloudWatchLogsEvents = "${var.application-log-cw-loggroup-arn}"
        # CloudTrail証跡S3への読み取り権限をつけるためにそのS3バケットを指定
        ElasticServerlessForwarderS3Buckets = "${var.audit-log-s3-arn}"
        ElasticServerlessForwarderSecurityGroups = ""
        ElasticServerlessForwarderSubnets = ""
    }
}

#====== Serverless forwarder config yamlを格納するS3 bucketの作成 ====
resource "aws_s3_bucket" "esf_config_bucket" {
  bucket = "${var.namespace}-elastic-serverless-forwarder-config"
}

resource "aws_s3_bucket_ownership_controls" "esf_config_bucket" {
  bucket = aws_s3_bucket.esf_config_bucket.id
  rule {
    object_ownership = "BucketOwnerPreferred"
  }
}

resource "aws_s3_bucket_acl" "esf_config_bucket" {
  depends_on = [aws_s3_bucket_ownership_controls.esf_config_bucket]

  bucket = aws_s3_bucket.esf_config_bucket.id
  acl    = "private"
}

#====== Serverless forwarder config yamlの作成と上記S3 bucketへのアップロード ====
locals {
  esf_config_content = <<-EOF
inputs:
  - type: "cloudwatch-logs"
    id: "${var.application-log-cw-loggroup-arn}"
    outputs:
      - type: "elasticsearch"
        args:
          # either elasticsearch_url or cloud_id, elasticsearch_url takes precedence if both are included
          # elasticsearch_url: "http(s)://domain.tld:port"
          cloud_id: "${var.elastic-cloud-id}"
          # either api_key or username/password, username/password takes precedence if both are included
          api_key: "${var.elastic-cloud-apikey}"
          # username: "username"
          # password: "password"
          es_datastream_name: "logs-generic-${var.namespace}"
          batch_max_actions: 500 # optional: default value is 500
          batch_max_bytes: 10485760 # optional: default value is 10485760
  - type: "s3-sqs"
    id: ${aws_sqs_queue.audit_log_s3_queue.arn}
    outputs:
      - type: "elasticsearch"
        args:
          # either elasticsearch_url or cloud_id, elasticsearch_url takes precedence if both are included
          # elasticsearch_url: "http(s)://domain.tld:port"
          cloud_id: "${var.elastic-cloud-id}"
          # either api_key or username/password, username/password takes precedence if both are included
          api_key: "${var.elastic-cloud-apikey}"
          # username: "username"
          # password: "password"
          es_datastream_name: "logs-gcp.audit-${var.namespace}"
          batch_max_actions: 500 # optional: default value is 500
          batch_max_bytes: 10485760 # optional: default value is 10485760
  EOF
}

resource "aws_s3_object" "esf-config" {
  bucket = aws_s3_bucket.esf_config_bucket.bucket
  key    = "${var.namespace}-config.yml"  # Specify the desired object key
#   source = data.template_file.config-application-logs.rendered # Specify the local file path for upload
  content = local.esf_config_content
  acl    = "private"
}

#====== CloudTrail証跡のS3のSQS通知のためののSQSキューと通知を作成 ====
resource "aws_sqs_queue" "audit_log_s3_queue" {
  name = "${var.namespace}-audit-log-s3-queue"
  visibility_timeout_seconds = 900
}

resource "aws_sqs_queue_policy" "audit_log_queue_policy" {
  queue_url = aws_sqs_queue.audit_log_s3_queue.id
  policy    = jsonencode({
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": "*",
        "Action": "SQS:SendMessage",
        "Resource": aws_sqs_queue.audit_log_s3_queue.arn,
        "Condition": {
            "ArnEquals": {
                "aws:SourceArn": var.audit-log-s3-arn
            }
        }
    }]
  })
}

# S3イベント通知の設定
resource "aws_s3_bucket_notification" "audit_log_bucket_notification" {
  bucket = var.audit-log-s3-name

  queue {
    queue_arn     = aws_sqs_queue.audit_log_s3_queue.arn
    events        = ["s3:ObjectCreated:*"]
    # filter_suffix = ".txt" # 適切なフィルター条件に変更してください
  }
}

Step2. 環境の値をvariableファイルに書いてください。

terraform.tfvarsの例

region = "us-west-2"
namespace = "test1203"
elastic_cloud_id = "xxxx:YXAtbm9 ....省略 w=="
elastic_apikey = "aUt ... 省略 QQ=="
audit-log-s3-name="aws-cloudtrail-logs-xx-省略"
audit-log-s3-arn="arn:aws:s3:::aws-cloudtrail-logs-省略"
application-log-cw-loggroup-arn="arn:aws:logs:us-west-2:アカウントID:log-group:firelens-container:*"

Step3. あとはterraform applyするだけです。

結果

アプリケーションログ： CloudWatch Logとそれを転送したElasticのES|QLを使った表示

ログが一致してます

CloudTrailログ:

ほぼ一致してますが、CloudTrailのイベント履歴がデフォルトで表示しないイベントがあるためか、Elastic側のデータが多く表示されました。

ES|QL

from logs-* 
| where data_stream.namespace=="test1203" and data_stream.dataset like "aws.cloudtrail" 
and @timestamp >= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-12-03T11:40:00+09:00") and @timestamp <= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-12-03T11:45:00+09:00")
| keep event.action, @timestamp, user.name, aws.cloudtrail.user_identity.type, event.provider, aws.cloudtrail.user_identity.access_key_id, aws.cloudtrail.response_elements.text
| sort @timestamp desc

#　トラブルシューティング
各フローのポイントでメトリック確認し、データが流れてそうかを確認してください。
Serverless ForwarderのLambda関数のメトリックを確認
Serverless ForwarderのLambda関数のログを確認
SQSのメトリック確認

おわり

Serverless Forwarderより、Kinesis Firehoseを使う方が設定がわかりやすいかもしれないので、時間があったらそちらも書きたいと思います。