今までの悩み
AWS を使っていると、こんな経験はありませんか？
- CloudTrail（誰が何をした？）のログはここ
- VPC フローログ（ネットワークの通信記録）はあそこ
- WAF ログ（不正アクセス対策）はまた別の場所
- セキュリティツールのログも、アプリのログも、全部バラバラ...
「あれ、さっきの怪しいアクセス、何のログをどこで見ればいいんだっけ？」
こんな状況では、障害対応もセキュリティ調査も大変で時間がかかりすぎますよね。
忙しい方向けに 3 行でポイントを説明
- 統合: バラバラだったログを一箇所に集約
- 自動変換: 形式がバラバラなログを統一フォーマットに
- 分析: CloudWatch 以外のサービスからも分析可能
具体的に説明
1. 統合
アカウントと AWS リージョン全体で AWS 提供のログやサードパーティのサービスを自動的に収集して 1 箇所に纏めることが可能です。
例として下記は一部です。CloudWatch > 取り込み > データソースのタブから確認ください。
では、実際に統合されている画面を少しだけ確認してみます。
前提として、CloudTrail のログを CloudWatch Logs に送っていることとします。
CloudWatch > Log management > Data sources で aws_cloudtrail.management が自動で検出してくれています。
aws_cloudtrail.management の中を確認していくと、自動的に CloudTrail の属性
(Fields）を検出してくれているみたいです。また、画面右側ではロググループが表示されており、カテゴリごとにロググループを確認できる点が嬉しいところですね。
2. 自動変換
CloudWatch パイプラインという機能を使って、ログデータを解析、変換、文字操作をしてくれる機能です。
例えば、ログを OCSF という共通フォーマットにしたい場合、自動変換されるので、「形式が違う」という悩みから解放されます。
では、CloudTrail のログを OCSF 形式に変換してみます。
CloudWatch > 取り込み > パイプライン > 「Create pipeline」。
ステップ 1: 一般設定
「AWS CloudTrail ログ」を選択し、任意のパイプライン名を入力し「次へ」。
ステップ 2: データソースの選択
「Management」を選択し、IAM ロールを新規に作成し、「次へ」。
ステップ 3: 送信先を設定
「次へ」。
既に CloudWatch にあるデータの場合は、元のロググループを宛先として使用されます。
ステップ 4: プロセッサを設定
「OCSF プロセッサ」を選択し、「追加」。
バージョンとマッピングバージョンを選択し、「次へ」。
ステップ 5: プレビューおよび作成
「パイプラインを作成」。
確認
CloudWatch Logs を確認すると次の通り、OCSF に変換されて保存されていることが確認できます。
CloudTrail Logs（OCSF）
{
"resources": [
{
"type": "AWS::DynamoDB::Table",
"uid": "arn:aws:dynamodb:ap-southeast-1:123456789012:table/naata-personal",
"owner": {
"account": {
"uid": "123456789012"
}
}
},
{
"type": "AWS::DynamoDB::Stream",
"uid": "arn:aws:dynamodb:ap-southeast-1:123456789012:table/naata-personal/stream/2019-05-10T10:57:45.349",
"owner": {
"account": {
"uid": "123456789012"
}
}
}
],
"time": 1765292638000,
"time_dt": "2025-12-09 15:03:58.000000",
"class_name": "API Activity",
"class_uid": 6003,
"category_name": "Application Activity",
"category_uid": 6,
"cloud": {
"provider": "AWS",
"region": "ap-southeast-1",
"account": {
"uid": "123456789012"
}
},
"severity_id": 1,
"severity": "Informational",
"metadata": {
"product": {
"name": "CloudTrail",
"vendor_name": "AWS",
"version": "1.11",
"feature": {
"name": "Management"
}
},
"profiles": [
"cloud",
"datetime"
],
"version": "1.5.0",
"event_code": "AwsApiCall",
"uid": "993e5bab-c692-408c-aa37-5eb12080617f"
},
"api": {
"request": {
"data": "{\"streamArn\":\"arn:aws:dynamodb:ap-southeast-1:123456789012:table/naata-personal/stream/2019-05-10T10:57:45.349\",\"limit\":100}",
"uid": "ANSNDUQD66UK7UDG8LDM8MJ64FVV4KQNSO5AEMVJF66Q9ASUAAJG"
},
"operation": "DescribeStream",
"version": "2012-08-10",
"service": {
"name": "dynamodb.amazonaws.com"
},
"response": null
},
"actor": {
"user": {
"type": "AssumedRole",
"uid_alt": "AROAT5BUXK4UP47NN7NB7:ProdReplDynamoSession",
"uid": "arn:aws:sts::123456789012:assumed-role/AWSServiceRoleForDynamoDBReplication/ProdReplDynamoSession",
"account": {
"uid": "123456789012"
},
"credential_uid": "ASIAT5BUXK4UOMPPBIHX",
"name": null
},
"session": {
"created_time_dt": "2025-12-09 14:54:11.000000",
"is_mfa": false,
"issuer": "arn:aws:iam::123456789012:role/aws-service-role/replication.dynamodb.amazonaws.com/AWSServiceRoleForDynamoDBReplication"
},
"invoked_by": "replication.dynamodb.amazonaws.com",
"idp": null
},
"http_request": {
"user_agent": "replication.dynamodb.amazonaws.com"
},
"status": "Success",
"src_endpoint": {
"domain": "replication.dynamodb.amazonaws.com",
"ip": null,
"uid": null
},
"observables": [
{
"name": "resources[].uid",
"type": "Resource UID",
"type_id": 10,
"value": "arn:aws:dynamodb:ap-southeast-1:123456789012:table/naata-personal"
},
{
"name": "src_endpoint.domain",
"type": "Hostname",
"type_id": 1,
"value": "replication.dynamodb.amazonaws.com"
}
],
"dst_endpoint": null,
"is_mfa": null,
"session": null,
"policy": null,
"user": null,
"activity_name": "Read",
"activity_id": 2,
"type_uid": 600302,
"type_name": "API Activity: Read",
"unmapped": {
"userIdentity.sessionContext.sessionIssuer.type": "Role",
"userIdentity.sessionContext.sessionIssuer.principalId": "AROAT5BUXK4UP47NN7NB7",
"userIdentity.sessionContext.sessionIssuer.accountId": "1234567890124",
"userIdentity.sessionContext.sessionIssuer.userName": "AWSServiceRoleForDynamoDBReplication",
"readOnly": "true",
"managementEvent": "true"
}
}
ところで、全然関係ありませんがシンガポールに DynamoDB 作った覚えないのにあるのはなぜ？
3. 分析
ファセット（Facets）機能
ファセット（Facets） 機能を使えば、ログを直感的にフィルタリングして、クエリすることが可能です。
ただし、これでは
簡易的なものしかできなさそうなのと、
カラムを理解している必要があるのと、
使い慣れている必要がある
ので、CloudWatch Logs Insights では自然言語でのクエリ生成が、日本語に対応してくれると個人的には嬉しいなといったところです。
S3 Tables に統合し、Athena などからクエリ
CloudWatch のログを S3 Tables と統合することで、Athena からクエリすることが可能です。
なお、本機能は東京リージョンでも対応しているはずですが、AWS マネージメントコンソールから確認したところ、CloudWatch のログと S3 Table を連携する機能が見当たらなかったのでバージニア北部で検証します。
CloudWatch > Log management > Data sources で aws_cloudtrail.management を選択し、「Associate data sources」。
「Create S3 Tables integration」。
新しい IAM ロールを作成し、「Create S3 Tables Integration」。
「aws_cloudtrail」「management」を選択し、「Associate data source(s)」。
Amazon S3 > テーブルバケット > aws-cloudwatch > aws_cloudtrail__management を確認するともちろん既にスキーマが定義されています。
Athena から S3 テーブルバケットへクエリを実行するためには、Amazon S3 > テーブルバケット で、「統合を有効にする」必要があります。
裏では自動的に、Glue カタログなどを自動的に作成してくれるようです。
では、準備が整いましたので後は Athena からクエリを実行するだけです。
AWS アカウントなどの情報が含まれるため、「完了済み」となっており、成功しているんだなと思ってください。
料金について
以下の AWS ドキュメントの記載によると、CloudWatch Log から S3 Tables への転送料金やストレージの料金は発生しないと解釈しています。(CloudWatch Log から削除されると、S3 Tables からも削除される)
Data retention in the S3 table bucket matches the retention policy set for the log group. For example, if you set a log group to 1-day retention, CloudWatch Logs removes the data from both CloudWatch Logs and the S3 Table after one day. When you delete a log group or log stream, CloudWatch Logs also removes the data from the S3 table bucket.
There are no additional storage or table maintenance charges for S3 tables created through this integration, beyond existing CloudWatch ingestion and storage pricing.
ただし **CloudWatch Logs への保管料金は 通常の S3 への保管料金よりも高いため、S3 へ長期保管している場合は、今まで通り S3 へ送り Athena などで分析する必要があるのかと思います。