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

はじめに

Adventカレンダー2023として、GCP/AWS/AzureのクラウドのログをElastic Cloudに転送する方法を、紹介していきます。
それぞれに色々な方法があるのですが、テーマとして最もさくっとできそうな手順をまとめます。
実際にログ転送を構築しようとすると、ログを転送するにはそれぞれのクラウド側の設定がそれなりに必要になってきて、ボタン一つポチッと転送する、というわけにはいきません。
よって、可能な限りそこの設定もTerraformで一括設定しつつ、Elastic Cloudにログを転送するコードを紹介します。
Terraformさえ使えれば、再現できるようなものになっています。

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

この記事について

Google Cloudのログ（具体的にはCloud Loggingのログ）をElastic Stackに送りたいけど、具体的にどうすればいいのか？　
やり方は色々あり、要件によって選ぶことになりますが、最もさくっと実現できる方法としてはGoogle Dataflowを使い、サーバーレスでログをElasticに流し込む方法です。
GUI設定でも実現はできますが、Google Cloud側のPub/Sub設定、IAM設定など色々必要になることから、それらを全部まとめたTerraformスクリプト一つでできるものを紹介します。

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

Google Cloud Loggingに送られているログのうち、今回は以下の2つについてElastic Stackに転送してみます。

• 監査ログ
• アプリケーションログ

アーキテクチャー概要

Google Cloudのロギングアーキテクチャー図についてまずこちらを参照ください。
図1: https://cloud.google.com/logging/docs/routing/overview?hl=ja
図2: https://cloud.google.com/architecture/security-foundations/logging-monitoring?hl=ja

今回は図1にあるUser defined log sinksの一種を作成し、Pub/Sub経由でElastic Cloudに送ります。
Pub/Sub以降の部分のもう少し詳しいのは図2にあり、Pub/Sub -> Dataflow -> SIEM toolという流れが書いてあります。

設定コンポーネントレベルにまとめると以下のようなフローになります。
Logging Router Sink -> Pub/Subトピック -> Pub/Subサブスクライバ -> Dataflow -> Elastic Cloud

手順

前提：TerraformでGoogle Cloudにアクセスする部分は事前にセットアップしてください。

今回のTerraformからGoogle Cloudにアクセスするサービスアカウントの権限としては以下のような権限でできます。

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

main.tf

terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "4.51.0"
    }
  }
}

variable "gcp_credential_json_path" {}
variable "project" {}
variable "region" {}
variable "namespace" {}
variable "elastic_cloud_id" {}
variable "elastic_apikey" {}
variable "log_filter_app_log" {}
variable "log_filter_auth_log" {}

provider "google" {
  credentials = file(var.gcp_credential_json_path)

  project = var.project
  region  = var.region
}


# ====== Application log configuration ============
# Cloud Logging Export
resource "google_logging_project_sink" "app_logs_sink" {
  name     = "${var.namespace}-app-logs-sink"
  filter = var.log_filter_app_log
  destination = "pubsub.googleapis.com/${google_pubsub_topic.app_logs_topic.id}"
  unique_writer_identity = true
}

# Pub/Sub - Create Topic
resource "google_pubsub_topic" "app_logs_topic" {
  name = "${var.namespace}-app-logs-topic"
}

# Pub/Sub - Give a permission to write to the topic
resource "google_pubsub_topic_iam_member" "app_logs_member" {
  project = google_pubsub_topic.app_logs_topic.project
  topic = google_pubsub_topic.app_logs_topic.name
  role = "roles/pubsub.publisher"
  member = google_logging_project_sink.app_logs_sink.writer_identity
}

# Pub/Sub - Subscriber to the log exported topic
resource "google_pubsub_subscription" "app_logs_subscription" {
  name   = "${var.namespace}-app-logs-subscription"
  topic  = google_pubsub_topic.app_logs_topic.name
}

resource "google_dataflow_flex_template_job" "app_logs_to_elastic" {
  provider = google-beta
  project = var.project
  region  = var.region
  name                    = "${var.namespace}-app-logs-to-elastic"
  container_spec_gcs_path = "gs://dataflow-templates-us-central1/latest/flex/PubSub_to_Elasticsearch"
  on_delete               = "cancel"
  parameters = {
      inputSubscription = google_pubsub_subscription.app_logs_subscription.id
      connectionUrl = var.elastic_cloud_id
      errorOutputTopic = google_pubsub_topic.dataflow_error_app_logs.id
      apiKey = var.elastic_apikey
      namespace = var.namespace
      dataset =  "pubsub"
      workerMachineType = "e2-medium" #default without this will be 4 vcpu 15GB (n1-standard-4)
  }
}

resource "google_pubsub_topic" "dataflow_error_app_logs" {
  name = "${var.namespace}-dataflow-error-app-logs"
}

# ====== Audit log configuration ============
resource "google_logging_project_sink" "audit_logs_sink" {
  name     = "${var.namespace}-audit-logs-sink"
  filter      = var.log_filter_auth_log
  destination = "pubsub.googleapis.com/${google_pubsub_topic.audit_logs_topic.id}"
  unique_writer_identity = true
}

# Pub/Sub - Create Topic
resource "google_pubsub_topic" "audit_logs_topic" {
  name = "${var.namespace}-audit-logs-topic"
}

# Pub/Sub - Give a permission to write to the topic
resource "google_pubsub_topic_iam_member" "audit_logs_member" {
  project = google_pubsub_topic.audit_logs_topic.project
  topic = google_pubsub_topic.audit_logs_topic.name
  role = "roles/pubsub.publisher"
  member = google_logging_project_sink.audit_logs_sink.writer_identity
}

# Pub/Sub - Subscriber to the log exported topic
resource "google_pubsub_subscription" "audit_logs_subscription" {
  name   = "${var.namespace}-audit-logs-subscription"
  topic  = google_pubsub_topic.audit_logs_topic.name
}

resource "google_dataflow_flex_template_job" "audit_logs_to_elastic" {
  provider = google-beta
  project = var.project
  region  = var.region
  name                    = "${var.namespace}-audit-logs-to-elastic"
  container_spec_gcs_path = "gs://dataflow-templates-us-central1/latest/flex/PubSub_to_Elasticsearch"
  on_delete               = "cancel"
  parameters = {
      inputSubscription = google_pubsub_subscription.audit_logs_subscription.id
      connectionUrl = var.elastic_cloud_id
      errorOutputTopic = google_pubsub_topic.dataflow_error_audit_logs.id
      apiKey = var.elastic_apikey
      dataset =  "audit"
      namespace = var.namespace
      workerMachineType = "e2-medium" #default without this will be 4 vcpu 15GB (n1-standard-4)
  }
}

resource "google_pubsub_topic" "dataflow_error_audit_logs" {
  name = "${var.namespace}-dataflow-error-audit-logs"
}

補足

• 上のコードでは、アプリケーションログの転送部分と、監査ログ転送部分に分かれてます。
• 大きな違いはlog sinkで指定するGoogle Cloudのログフィルターのところと、datasetの値です。
  • 監査ログの場合はdataset = "audit"にしてください。Elastic Cloud側で監査ログに関しては監査ログ専用の加工処理（Ingest Pipeline）の上取り込まれ、すぐに組み込みの[Logs GCP] Auditというダッシュボードでデータを確認できます。
  • アプリケーションログの場合はdataset = "pubsub"にしてください。汎用的なログはpubsub指定すればそのまま取り込まれます。
• コード見るとわかりますが、Pub/Subトピックに書き込むための権限をSinkのwriter_identityのサービスアカウントに与えないといけません。GUIで同じことを設定していくと自動的にここはやってくれてあまり意識しないのですが、今回はここがハマりポイントでした。
• DataflowのworkerMachineTypeを明示的に小さく指定しないと、デフォルトではコストのかかるインスタンスタイプが使われて課金が多く発生してしまいます。

Step 2: 環境固有情報をこちらのtfvarsファイルに入れていきます。

terraform.tfvars

gcp_credential_json_path = ""
project = ""
region = ""
namespace = ""
elastic_cloud_id = ""
elastic_apikey = ""
log_filter_app_log = ""
log_filter_auth_log = ""

補足

• log_filter_app_logとlog_filter_auth_logのフィルター条件はLogging Explorerのこの部分を使います。
  
  auditd_resourceが監査ログを指定しているところです。

Terraformとしてこの画面のクエリー文を使うには、ダブルクオートや改行については少し変換してこのような例のフォーマットにします。

log_filter_auth_log = "resource.type=audited_resource AND log_name=projects/<project>/logs/cloudaudit.googleapis.com%2Factivity"

• gcp_credential_json_path は、Google のサービスアカウントのキーをJSON形式でダウンロードしたもののファイルパスです。

• namespaceはElastic Cloud側のログのnamespaceで、環境名(test, prodなど自由に）をつけるものです。また、今回はこのnamespaceはGoogle Cloud側の設定名に関してもプレフィックスとして使いました。

例としてこのようになります。

terraform.tfvarsの例

gcp_credential_json_path = "myproject-99e102ce5ce0.json"
project = "myproject"
region = "asia-northeast1"
namespace = "test1130"
elastic_cloud_id = "xxxx:YXAtbm9 ....省略 w=="
elastic_apikey = "aUt ... 省略 QQ=="
log_filter_app_log = "logName=projects/myproject/logs/stderr AND resource.labels.namespace_name=otel-demo AND resource.type=k8s_container AND severity=ERROR AND resource.labels.container_name=productcatalogservice"
log_filter_auth_log = "resource.type=audited_resource AND log_name=projects/myproject/logs/cloudaudit.googleapis.com%2Factivity"

Step3. Terraformを実行し、設定を反映してください。

terraform applyで反映です。

結果：Elastic Cloudにログが取り込まれた様子

ES|QLを使うとGoogle Cloudと同じようにクエリーできます。件数も一致しています。

アプリケーションログの比較

Google Logging Query

logName="projects/xxxxx/logs/stderr"
resource.labels.namespace_name="otel-demo"
resource.type="k8s_container" AND severity=ERROR
resource.labels.container_name="productcatalogservice"
textPayload:"code = ResourceExhauste"
timestamp>="2023-11-30T13:30:00Z" AND timestamp<="2023-11-30T14:00:00Z" 

Elsatic ES|QL

from logs-*
| where logName == "projects/xxxxx/logs/stderr" 
and resource.labels.namespace_name == "otel-demo" 
and resource.type == "k8s_container"
and severity == "ERROR" and resource.labels.container_name == "productcatalogservice"
and textPayload like "*code = ResourceExhausted*" 
and @timestamp >= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-11-29T13:00:00+09:00") and @timestamp <= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-11-29T14:00:00+09:00")
| keep @timestamp, resource.labels.container_name, textPayload

監査ログの比較

Google Logging Query

resource.type="audited_resource"
logName="projects/xxxxx/logs/cloudaudit.googleapis.com%2Factivity"
timestamp>="2023-11-30T13:30:00+0900" AND timestamp<="2023-11-30T14:00:00+0900"

Elsatic ES|QL

from logs-*
| where resource.type == "audited_resource" 
and logName == "projects/xxxxx/logs/cloudaudit.googleapis.com%2Factivity"
and @timestamp >= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-11-30T13:30:00+09:00") and @timestamp <= date_parse("yyyy-MM-dd'T'HH:mm:sszzzz", "2023-11-30T14:00:00+09:00")
| keep severity, @timestamp, protoPayload.serviceName, protoPayload.methodName,protoPayload.authenticationInfo.principalEmail

うまくいかない時のトラブルシューティング

Elastic Cloud側のデータが表示されない場合、以下のフローのステップ毎のメトリックを確認し、どこがうまくいっていないかを確認する必要があります。

Logging Router Sink -> Pub/Subトピック -> Pub/Subサブスクライバ -> Dataflow -> Elastic Cloud

例えば、今回最初のSinkのところで、作成したSinkが次のPub/Subトピックへの書き込み権限がなかったため、Sinkのメトリックでエラーカウントを確認して追うことができました。

おわり

この記事の手順を作り上げるのは全くさくっとといきませんでしたが、この記事を参考にしていただければきっとさくっとできると思います。。