Amazon CloudWatch pipelines の新機能、GeoIP・RDS・XML の3プロセッサを試してみた
はじめに
2026年8月19日、AWS は Amazon CloudWatch pipelines の拡張を発表しました。What's New では、今回追加されたプロセッサを「an Amazon RDS log parser, an XML parser and a GeoIP enrichment processor」と記載しています。
| プロセッサ | 用途（What's New の記載を要約） |
|---|---|
| Amazon RDS パーサー | Aurora の監査ログ・エラーログを構造化フィールドへパースする |
| XML パーサー | XML 文字列を含むフィールドを JSON へ変換する |
| GeoIP エンリッチメントプロセッサ | IP アドレスのフィールドに都市・国・座標などの地理情報を付与する |
検証内容
まず、パイプラインが処理対象のログを判別する仕組みと、データソースの宣言方法を確認します。
パイプラインの前提
パイプラインの処理対象は、ロググループ名ではなく、ログイベントに付くデータソース名と種別で決まります。RDS のような AWS サービスが出力するログは、タグを付けなくても CloudWatch Logs 側で分類されます。アプリケーション側のログでは、ロググループのタグでデータソースを宣言します。
aws logs create-log-group --log-group-name "/pipeline-test/xml-live" --region us-east-1 \
--tags 'cw:datasource:name=xml_source,cw:datasource:type=default'
上記コマンドで、ロググループに次のタグを付けました。
{"tags":{"cw:datasource:name":"xml_source","cw:datasource:type":"default"}}
AWS サービスのログには、amazon_rds のように aws や amazon で始まるデータソース名が自動で付与されます。これらとの競合を避けるため、ユーザーがタグで宣言するデータソース名には、aws や amazon で始まる名前を指定できません。
もう1つの前提が IAM ロールです。パイプラインのソースでは、sts_role_arn でロールを指定します。ロールには、logs.amazonaws.com からの引き受けを許可する信頼ポリシーと、対象のロググループを Resource に指定した logs:ProcessWithPipeline の権限ポリシーを設定します。
Aurora 監査ログのパース
今回は、監査ログを有効化し、CloudWatch Logs へエクスポートする設定にした Aurora MySQL を用意しました。実際の設定は、server_audit_logging が ON で、監査対象のイベントは CONNECT、QUERY、TABLE、QUERY_DDL、QUERY_DML、QUERY_DCL です。この設定で、ロググループ /aws/rds/cluster/<クラスター名>/audit が作られます。
パイプラインは次のように設定しました。
pipeline:
source:
cloudwatch_logs:
aws:
sts_role_arn: "arn:aws:iam::123456789012:role/example-role"
log_event_metadata:
data_source_name: "amazon_rds"
data_source_type: "aurora_mysql_audit"
processor:
- parse_rds: {}
sink:
- cloudwatch_logs:
log_group: "@original"
include_original: {}
ソースが cloudwatch_logs の場合、sink は @original 固定です。変換結果は、元のロググループの @message に書き換わります。include_original: {} を指定しているため、変換前の内容は @original_message で参照できます。
EC2 から admin ユーザーで DDL / DML / SELECT / 存在しないテーブルへの SELECT を実行しました。admin の監査イベントは36件で、内訳は CONNECT 10件、QUERY 10件、DISCONNECT 10件、READ 3件、WRITE 2件、CREATE 1件でした。
Aurora MySQL の監査ログは、10列の CSV 行として届きます。Logs Insights で、同一イベントの @original_message（変換前）と @message（変換後）を並べました。
SELECT：
1787412486446138,pipeline-test-aurora-instance,admin,10.0.2.100,65,2691,QUERY,testdb,'SELECT * FROM users WHERE id = 1',0
{"timestamp":"1787412486446138","serverhost":"pipeline-test-aurora-instance","username":"admin","host":"10.0.2.100","connectionid":65,"queryid":2691,"operation":"QUERY","database":"testdb","query":"SELECT * FROM users WHERE id = 1","retcode":0}
存在しないテーブルへの SELECT（retcode に MySQL のエラー番号 1146 が入る）：
1787412486486778,pipeline-test-aurora-instance,admin,10.0.2.100,68,2698,QUERY,testdb,'SELECT * FROM no_such_table',1146
{"timestamp":"1787412486486778","serverhost":"pipeline-test-aurora-instance","username":"admin","host":"10.0.2.100","connectionid":68,"queryid":2698,"operation":"QUERY","database":"testdb","query":"SELECT * FROM no_such_table","retcode":1146}
テーブル単位のイベント（operation が READ / WRITE / CREATE になり、query にテーブル名が入る）：
1787412486446019,pipeline-test-aurora-instance,admin,10.0.2.100,65,2691,READ,testdb,users,
{"timestamp":"1787412486446019","serverhost":"pipeline-test-aurora-instance","username":"admin","host":"10.0.2.100","connectionid":65,"queryid":2691,"operation":"READ","database":"testdb","query":"users","retcode":""}
接続イベント（query と retcode のキーが出力されない）：
1787412486342488,pipeline-test-aurora-instance,admin,10.0.2.100,61,0,CONNECT,,,0
{"timestamp":"1787412486342488","serverhost":"pipeline-test-aurora-instance","username":"admin","host":"10.0.2.100","connectionid":61,"queryid":0,"operation":"CONNECT","database":""}
テーブル単位のイベントでは戻り値が空文字列になります。接続イベントでは、クエリと戻り値のキー自体が出力されず、database は空文字列として残ります。
同じ10列形式の行（模擬データ）を CloudWatch Logs トランスフォーマーの CSV プロセッサで分解すると、次のようになりました。
20260822140000,pipeline-test-aurora-instance,admin,10.0.2.100,12345,67890,QUERY,testdb,'SELECT * FROM users WHERE id = 1',0
{"timestamp":"20260822140000","server_host":"pipeline-test-aurora-instance","username":"admin","client_host":"10.0.2.100","connection_id":"12345","query_id":"67890","operation":"QUERY","database":"testdb","query":"'SELECT * FROM users WHERE id = 1'","return_code":"0"}
parse_rds と CSV プロセッサの結果を比較すると、次のとおりです。
| 観点 | parse_rds | CSV プロセッサ |
|---|---|---|
| 接続ID・クエリID・戻り値 | 数値 | 文字列 |
| クエリのシングルクォート | 除去される | 残る |
| 空フィールド | イベント種別により異なる | 空文字列として残る |
| フィールド名 | 列定義なしで固定名 | 10列を自分で定義 |
IP アドレスのエンリッチ
geoip で、JSON ログの IP アドレスのフィールドに地理情報を付与しました。include_fields には8つのフィールドを指定しました。
pipeline:
source:
cloudwatch_logs:
aws:
sts_role_arn: "arn:aws:iam::123456789012:role/example-role"
log_event_metadata:
data_source_name: "geoip_source"
data_source_type: "default"
processor:
- parse_json: {}
- geoip:
entries:
- source: "client_ip"
target: "geo"
include_fields:
- "country_name"
- "country_iso_code"
- "city_name"
- "time_zone"
- "latitude"
- "longitude"
- "asn"
- "asn_organization"
sink:
- cloudwatch_logs:
log_group: "@original"
include_original: {}
投入した3件：
{"timestamp":"2026-08-23T00:10:00Z","client_ip":"8.8.8.8","request":"GET /api/users"}
{"timestamp":"2026-08-23T00:10:01Z","client_ip":"52.195.200.162","request":"GET /api/posts"}
{"timestamp":"2026-08-23T00:10:02Z","client_ip":"13.246.245.8","request":"POST /api/comments"}
同じロググループを読み直すと、geo が付いた状態で返ってきました。
{"timestamp":"2026-08-23T00:10:00Z","client_ip":"8.8.8.8","request":"GET /api/users","geo":{"asn_organization":"Google LLC","country_iso_code":"US","latitude":37.751,"country_name":"United States","time_zone":"America/Chicago","asn":15169,"longitude":-97.822}}
{"timestamp":"2026-08-23T00:10:01Z","client_ip":"52.195.200.162","request":"GET /api/posts","geo":{"asn_organization":"Amazon.com, Inc.","city_name":"Tokyo","country_iso_code":"JP","latitude":35.6893,"country_name":"Japan","time_zone":"Asia/Tokyo","asn":16509,"longitude":139.6899}}
{"timestamp":"2026-08-23T00:10:02Z","client_ip":"13.246.245.8","request":"POST /api/comments","geo":{"asn_organization":"Amazon.com, Inc.","city_name":"Cape Town","country_iso_code":"ZA","latitude":-33.9258,"country_name":"South Africa","time_zone":"Africa/Johannesburg","asn":16509,"longitude":18.4259}}
8.8.8.8 では city_name が返りませんでした。include_fields に指定しても取得できないフィールドは、キーごと出力されません。
変換後のイベントは、投入から1分ほどで確認できました。
リージョン差を見るため、別途 test-telemetry-pipeline に各リージョンの STS エンドポイントの実 IP を渡し、12フィールドすべてを指定して1リージョンにつき1回実行しました。
| region | country_iso_code | city_name | time_zone | asn | 欠落フィールド |
|---|---|---|---|---|---|
| us-east-1 | US | Ashburn | America/New_York | 14618 | なし |
| us-west-2 | US | Boardman | America/Los_Angeles | 16509 | なし |
| ca-central-1 | CA | Montreal | America/Toronto | 16509 | なし |
| eu-west-1 | IE | Dublin | Europe/Dublin | 16509 | なし |
| eu-central-1 | DE | Frankfurt am Main | Europe/Berlin | 16509 | なし |
| eu-north-1 | SE | Stockholm | Europe/Stockholm | 16509 | なし |
| eu-south-1 | IT | Milan | Europe/Rome | 16509 | なし |
| ap-northeast-1 | JP | Tokyo | Asia/Tokyo | 16509 | なし |
| ap-northeast-2 | KR | Incheon | Asia/Seoul | 16509 | なし |
| ap-southeast-1 | SG | Singapore | Asia/Singapore | 16509 | なし |
| ap-southeast-2 | AU | Sydney | Australia/Sydney | 16509 | なし |
| ap-south-1 | IN | Mumbai | Asia/Kolkata | 16509 | なし |
| sa-east-1 | BR | São Paulo | America/Sao_Paulo | 16509 | なし |
| me-south-1 | BH | — | Asia/Bahrain | 16509 | city_name, postal_code |
| af-south-1 | ZA | Cape Town | Africa/Johannesburg | 16509 | なし |
12フィールドが揃った例（ap-northeast-1）：
{"client_ip":"52.195.200.162","region":"ap-northeast-1","geo":{"continent_name":"Asia","asn_organization":"Amazon.com, Inc.","city_name":"Tokyo","country_iso_code":"JP","latitude":35.6893,"country_name":"Japan","continent_code":"AS","postal_code":"151-0053","time_zone":"Asia/Tokyo","asn":16509,"longitude":139.6899,"network":"52.192.0.0/13"}}
都市と郵便番号が返らなかった例（me-south-1）：
{"client_ip":"99.82.136.65","region":"me-south-1","geo":{"continent_name":"Asia","asn_organization":"Amazon.com, Inc.","country_iso_code":"BH","latitude":26.1226,"country_name":"Bahrain","continent_code":"AS","time_zone":"Asia/Bahrain","asn":16509,"longitude":50.5557,"network":"99.82.128.0/18"}}
XML 文字列のパース
parse_xml で、JSON ログの body フィールドに入った XML 文字列を JSON へ変換しました。
pipeline:
source:
cloudwatch_logs:
aws:
sts_role_arn: "arn:aws:iam::123456789012:role/example-role"
log_event_metadata:
data_source_name: "xml_source"
data_source_type: "default"
processor:
- parse_json: {}
- parse_xml:
source: "body"
destination: "parsed_xml"
sink:
- cloudwatch_logs:
log_group: "@original"
include_original: {}
parse_xml は、parse_json など先に置くパーサーの後に置きます。1つのパイプラインに置ける parse_xml は5個までです。
投入した4件は、正常な XML 2件、閉じタグのない XML 1件、body フィールドを持たないもの1件です。
ネストと同名要素の繰り返しを含む XML（変換前）：
{"event_time":"2026-08-23T00:55:00Z","source":"app-a","body":"<order id=\"A1\"><customer><name>Alice</name><email>a@example.com</email></customer><items><item sku=\"X1\"><qty>2</qty></item><item sku=\"X2\"><qty>5</qty></item></items><note>hello</note></order>"}
変換後：
{"event_time":"2026-08-23T00:55:00Z","source":"app-a","body":"<order id=\"A1\"><customer><name>Alice</name><email>a@example.com</email></customer><items><item sku=\"X1\"><qty>2</qty></item><item sku=\"X2\"><qty>5</qty></item></items><note>hello</note></order>","parsed_xml":{"id":"A1","customer":{"name":"Alice","email":"a@example.com"},"items":{"item":[{"sku":"X1","qty":"2"},{"sku":"X2","qty":"5"}]},"note":"hello"}}
CDATA を含む XML の変換後：
{"event_time":"2026-08-23T00:55:01Z","source":"app-b","body":"<msg type=\"alert\"><body2><![CDATA[<b>bold</b> & stuff]]></body2><level>3</level></msg>","parsed_xml":{"type":"alert","body2":"<b>bold</b> & stuff","level":"3"}}
閉じタグのない XML は変換されず、投入した内容のまま残りました：
{"event_time": "2026-08-23T00:55:02Z", "source": "app-c", "body": "<Person id=\"123\"><name>John</name>"}
body を持たないイベントもそのまま残りました：
{"event_time":"2026-08-23T00:55:03Z","source":"app-d","note":"body field absent"}
ルート要素の属性は子要素と同じ階層に並び、ルート要素名はキーになりません。同名要素が複数あると配列になります。CDATA の中身は文字列として入ります。
投入から70秒以内に反映を確認できました。
destination を省略すると、パース結果がイベントのルート直下へ展開されます。元の body フィールドは残ります。
{"event_time":"2026-08-23T00:45:00Z","body":"<Person id=\"123\" active=\"true\"><name>John</name><age>30</age></Person>","id":"123","active":"true","name":"John","age":"30"}
値の型と空要素の扱いも確認しました。入力は次の XML です。
<m><int>42</int><float>3.14</float><bool>true</bool><nullish></nullish><selfclose/><spaces> padded </spaces></m>
{"int":"42","float":"3.14","bool":"true","nullish":"","selfclose":"","spaces":" padded "}
値はすべて文字列で、数値や真偽値への型変換はされません。空要素と自己終了タグは空文字列になり、前後の空白は保持されます。
ネスト深度の上限
test-telemetry-pipeline で深度を変えて試したところ、深度25までは変換に成功し、深度26と30では失敗しました。公式ドキュメントに記載された上限（ネスト深度25）と一致します。深度26のときの結果は次のようになりました。
{"Results":[{"Error":{"Message":"Processor 1: Failed to parse the log event"}}]}
まとめ
Amazon CloudWatch pipelines のプロセッサを利用することで、Lambda や外部ツールを使わずに CloudWatch Logs 内でログを構造化したり、IP アドレスに地理情報を付けたりできます。Amazon RDS パーサーによる Aurora MySQL の監査ログのパースをはじめ、XML を含むアプリケーションログの構造化、IP アドレスへの地理情報の付与を確認しました。CloudWatch Logs に保存するログの前処理が必要な場合は、まずはパイプラインのプロセッサをお試しください。
