Cloudflare R2 に Logpush したログを ALog Cloud に連携する

Last updated at 2024-11-19Posted at 2024-08-22

参考：ALog Cloud とは

無料トライアルを活用させていただきました。

ALogはオンプレからクラウドまで、あらゆるシステムのログデータを収集・分析・保管する純国産のSIEM製品です。
“わかりやすさ”を追及したALogは、あらゆるITシステムのログを自動で収集、一元的に管理するSIEM製品です。
「セキュリティの難しいをカンタンに」を、コンセプトに専門知識やノウハウがなくとも、誰でもカンタンに高度なログ分析を実現。
企業がかかえる課題に沿って、幅広いシーンで活用できるのが特徴です。

R2 に Logpush したデータをマウントしてフォワードする

ALog Cloud のログ取り込み方式を考慮し、以下のような構成を実装します。

必ずしも Linux サーバーを別に立てる必要はありませんが、切り分けのために別立てします。

R2 バケットに Event notification を設定

以下の記事を参考にして進めます。

必要となる R2, Queue, Workers を作成します。

# R2 バケットを作成
export LOGPUSH_BUCKET_NAME='alog-gzip'
export MOUNT_BUCKET_NAME='alog-cloud'
wrangler r2 bucket create $LOGPUSH_BUCKET_NAME
wrangler r2 bucket create $MOUNT_BUCKET_NAME

# Queue を作成
export QUEUE_NAME='alog-gunzip-queue'
wrangler queues create $QUEUE_NAME

# Workers を作成
npm create cloudflare@latest -- alog-gunzip-worker
cd alog-gunzip-worker

オブジェクト書き込みのイベント通知ごとに処理を起動し、*.log.gz の gzip ファイル形式から *.csv にすることで Forwarder からのアップロードに対応します。

wrangler.toml

name = "alog-gunzip-worker"
main = "src/index.ts"
compatibility_date = "2024-08-15"
compatibility_flags = ["nodejs_compat"]

[[queues.consumers]]
queue = "alog-gunzip-queue"
max_batch_size = 100
max_batch_timeout = 5

[[r2_buckets]]
binding = "LOGPUSH_BUCKET"
bucket_name = "alog-gzip"

[[r2_buckets]]
binding = "MOUNT_BUCKET"
bucket_name = "alog-cloud"

src/index.ts

import { ungzip } from 'pako';

export interface Env {
	LOGPUSH_BUCKET: R2Bucket;
	MOUNT_BUCKET: R2Bucket;
}
  
export default {
	async queue(batch, env): Promise<void> {
		// A queue consumer can make requests to other endpoints on the Internet,
		// write to R2 object storage, query a D1 Database, and much more.
		for (const message of batch.messages) {
			// Process each message (we'll just log these)
			console.log(`Message: ${JSON.stringify(message.body)}`);
			const fileName = message.body["object"]["key"];
			console.log(`object.key: ${fileName}`);
			//gunzip data into decompressedData
			const response = await env.LOGPUSH_BUCKET.get(fileName);
			const decompressedData = ungzip(new Uint8Array(await response.arrayBuffer()));
			// Write the batch of messages to R2
			console.log(`object.target: ${fileName.split('.')[0]}`);
			const fileTargetName = fileName.split('.')[0] + ".csv";
			await env.MOUNT_BUCKET.put(fileTargetName, decompressedData, {
				httpMetadata: {
					contentType: "text/csv",
				},
			});
			// Explicitly acknowledge the message as delivered
			message.ack();
		}
	},
} satisfies ExportedHandler<Env>;

最後に Workers をデプロイして、R2 バケットに Event notification を設定します。

wrangler deploy
wrangler r2 bucket notification create $LOGPUSH_BUCKET_NAME --event-type object-create --queue $QUEUE_NAME

Logpush を CSV で R2 に出力

以下のコマンドで gateway_http データセットの Logpush を作成します。

output_type で csv を指定し、destination_conf で R2 を指定します。

R2 バケット内では日付ごとにフォルダ分けはせず、timestamp_format は unix にします。

export EMAIL='YOUR_EMAIL'
export APIKEY='YOUR_APIKEY'
export ACCOUNT_ID='YOUR_ACCOUNT_ID'
export R2_ACCESS_KEY_ID='YOUR_R2_ACCESS_KEY_ID'
export R2_SECRET_ACCESS_KEY='YOUR_R2_SECRET_ACCESS_KEY'

# 全フィールドを指定
export FIELDS=$(curl -s \
-H "X-Auth-Email: $EMAIL" \
-H "X-Auth-Key: $APIKEY" \
"https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/logpush/datasets/gateway_http/fields" | jq -r '.result | keys | join("\",\"")')
echo $FIELDS

# Logpush ジョブを作成
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/logpush/jobs" -X POST -d '
{                                              
  "name": "gateway-http-r2-alog",
  "output_options": {
    "field_names": ["'$FIELDS'"],
    "output_type": "csv",
    "timestamp_format": "unix"
  },
  "destination_conf": "r2://alog-gzip?account-id='$ACCOUNT_ID'&access-key-id='$R2_ACCESS_KEY_ID'&secret-access-key='$R2_SECRET_ACCESS_KEY'",
  "max_upload_bytes": 5000000,
  "max_upload_records": 1000,
  "dataset": "gateway_http",
  "enabled": true
}' \
-H "X-Auth-Email: $EMAIL" \
-H "X-Auth-Key: $APIKEY"

s3fs を使って R2 を Linux にマウント

以下の記事を参考にします。

以下のコマンドで R2 をマウントできます。

# s3fs インストール
sudo yum install epel-release -y
sudo yum install s3fs-fuse -y

# クルデンシャルファイル作成
export R2_ACCESS_KEY_ID='YOUR_R2_ACCESS_KEY_ID'
export R2_SECRET_ACCESS_KEY='YOUR_R2_SECRET_ACCESS_KEY'
echo $R2_ACCESS_KEY_ID:$R2_SECRET_ACCESS_KEY > ${HOME}/.passwd-s3fs
chmod 600 ${HOME}/.passwd-s3fs

# マウントポイント作成
mkdir alog-cloud

# R2 をマウント
export ACCOUNT_ID='YOUR_ACCOUNT_ID'
s3fs -o passwd_file=${HOME}/.passwd-s3fs -o endpoint=auto \
-o url="https://$ACCOUNT_ID.r2.cloudflarestorage.com" alog-cloud ${HOME}/alog-cloud

以下のように R2 に Logpush された *.log.gz ファイルを、Event notification によって連携した Workers で解凍した *.csv ファイルが確認できます。

$ tree alog-cloud/
alog-cloud/
|-- test.csv
|-- 20240821T054839Z_20240821T054839Z_06b7e0be.csv
|-- 20240821T054914Z_20240821T054914Z_8f641fd1.csv
|-- 20240821T055213Z_20240821T055213Z_81de4248.csv
|-- 20240821T055338Z_20240821T055338Z_bf3b5176.csv
...