CloudWatchの新機能「lookup processor」で VPC Flow Logs をエンリッチしてみた
はじめに
2026年7月14日、Amazon CloudWatch に lookup processor が追加されました。CSV で用意したマッピングテーブルをもとに、CloudWatch Pipelines のパイプライン内でログフィールドを自動的にエンリッチできる機能です。
本記事では VPC Flow Logs を対象に、送信元IPアドレスからチーム名・サービス名・環境名を付与する構成を試しました。
- リージョン： ap-northeast-1（東京、動作確認済み。他リージョンは未検証）
- 対象ロググループ：
/vpc/default-flow-logs - トラフィック生成： EC2 t4g.nano
検証内容
Lookup Table の作成
CSV のヘッダーに ip_address・team・service・environment を定義しました。後続の lookup processor で ip_address を照合キーとして使用し、entries で指定した team・service・environment の値をログへ付与します。
aws logs create-lookup-table \
--lookup-table-name vpc_flow_ip_lookup \
--description "IP address to team/service mapping for VPC Flow Logs enrichment" \
--table-body "ip_address,team,service,environment
203.0.113.1,platform,api-gateway,production
203.0.113.2,platform,api-gateway,production
10.0.1.203,backend,ec2-instance,development
10.0.1.185,test,flow-log-test,development
10.0.1.222,test,flow-log-test2,development
203.0.113.10,infra,ntp-server,shared
" \
--region ap-northeast-1
Lookup Table の ARN が返りました。後続の IAM ポリシーとパイプライン定義でこの ARN を使用します。
{
"lookupTableArn": "arn:aws:logs:ap-northeast-1:123456789012:lookup-table:vpc_flow_ip_lookup",
"createdAt": 1784133588924
}
CSV のサイズ上限は 10 MB、アカウント・リージョンあたり最大 100 テーブルです。今回のような 1 行 50〜100 bytes 程度のマッピングであれば 10 万行以上登録できます。
IAM ロールの準備
Trust Policy で CloudWatch Logs サービスがロールを引き受けられるようにします。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "logs.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
Permission Policy では 2 つのアクションを許可します。logs:ProcessWithPipeline は対象ロググループのログをパイプラインで処理するために必要です。logs:GetLookupTable は lookup processor がマッピングテーブルを読み取るために必要です。それぞれリソースをロググループと Lookup Table の ARN に絞っています。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "ProcessWithPipeline",
"Effect": "Allow",
"Action": "logs:ProcessWithPipeline",
"Resource": "arn:aws:logs:ap-northeast-1:123456789012:log-group:/vpc/default-flow-logs"
},
{
"Sid": "GetLookupTable",
"Effect": "Allow",
"Action": "logs:GetLookupTable",
"Resource": "arn:aws:logs:ap-northeast-1:123456789012:lookup-table:vpc_flow_ip_lookup"
}
]
}
ロググループへのデータソースタグ設定
パイプラインが対象ログを識別できるように、ロググループにデータソースタグを設定します。cw:datasource:name にデータソースの名前（amazon_vpc）、cw:datasource:type に種別（flow）を指定します。このタグ値がパイプライン定義の log_event_metadata と一致することで、処理対象として認識されます。
aws logs tag-log-group \
--log-group-name "/vpc/default-flow-logs" \
--tags '{"cw:datasource:name": "amazon_vpc", "cw:datasource:type": "flow"}'
CloudWatch Pipeline の作成
パイプラインの構成を YAML で定義します。
pipeline:
source:
cloudwatch_logs:
log_event_metadata:
data_source_name: "amazon_vpc"
data_source_type: "flow"
aws:
sts_role_arn: "arn:aws:iam::123456789012:role/vpc-flow-logs-pipeline-role"
processor:
- parse_vpc: {}
- lookup:
lookupTable: "arn:aws:logs:ap-northeast-1:123456789012:lookup-table:vpc_flow_ip_lookup"
matchKeys:
- logKey: srcAddr
lookupKey: ip_address
entries:
- source: team
target: src_team
- source: service
target: src_service
- source: environment
target: src_environment
sink:
- cloudwatch_logs:
log_group: "@original"
processor は 2 段構成です。まず parse_vpc が VPC Flow Logs のテキスト行を構造化フィールドに分解し、続く lookup がエンリッチを担います。matchKeys で srcAddr（parse_vpc 出力）と ip_address（Lookup Table のキー列）を突き合わせます。entries では CSV のカラム名（source）と出力フィールド名（target）の対応を定義します。
本検証の CloudWatch Logs ソース構成では、sink の出力先に元のロググループを示す @original を指定しました。
aws observabilityadmin validate-telemetry-pipeline-configuration \
--configuration '{"Body": "<上記YAMLをJSON文字列としてエスケープ>"}'
aws observabilityadmin create-telemetry-pipeline \
--name vpc-flow-lookup-test \
--configuration '{"Body": "<上記YAMLをJSON文字列としてエスケープ>"}'
{
"Arn": "arn:aws:observabilityadmin:ap-northeast-1:123456789012:telemetry-pipeline/xxxxxxxxxxxxxxxxxxxxxxxxxx"
}
aws observabilityadmin get-telemetry-pipeline \
--pipeline-identifier vpc-flow-lookup-test
{
"Pipeline": {
"Name": "vpc-flow-lookup-test",
"Arn": "arn:aws:observabilityadmin:ap-northeast-1:123456789012:telemetry-pipeline/xxxxxxxxxxxxxxxxxxxxxxxxxx",
"Status": "ACTIVE"
}
}
Status が ACTIVE となり、パイプラインが稼働状態になりました。
動作確認
test-telemetry-pipeline でドライランを実行します。パイプライン定義とサンプルレコードを渡し、processor による変換結果を確認できます。
aws observabilityadmin test-telemetry-pipeline \
--configuration '{"Body": "<上記YAMLをJSON文字列としてエスケープ>"}' \
--records '[{"Data": "11 123456789012 eni-xxxxxxxxxxxxxxxxx 10.0.1.222 203.0.113.50 54634 443 6 10 500 1784135000 1784135060 ACCEPT OK - - - - - - - -", "Type": "STRING"}]'
入力（送信元IP 10.0.1.222 の VPC Flow Log）:
11 123456789012 eni-xxxxxxxxxxxxxxxxx 10.0.1.222 203.0.113.50 54634 443 6 10 500 1784135000 1784135060 ACCEPT OK - - - - - - - -
出力（エンリッチ後）:
{
"version": 11,
"accountId": "123456789012",
"interfaceId": "eni-xxxxxxxxxxxxxxxxx",
"srcAddr": "10.0.1.222",
"dstAddr": "203.0.113.50",
"srcPort": 54634,
"dstPort": 443,
"protocol": 6,
"packets": 10,
"bytes": 500,
"start": 1784135000,
"end": 1784135060,
"action": "ACCEPT",
"logStatus": "OK",
"src_team": "test",
"src_service": "flow-log-test2",
"src_environment": "development"
}
Lookup Table と一致した 10.0.1.222 に対し、src_team・src_service・src_environment の 3 フィールドが付与されました。
次に、実データで確認します。EC2 インスタンスからトラフィックを発生させ、ロググループに蓄積されたログを Logs Insights で確認しました。
fields @timestamp, srcAddr, dstAddr, action, src_team, src_service, src_environment
| filter srcAddr = "10.0.1.222" and action = "ACCEPT"
| sort @timestamp desc
| limit 5
| @timestamp | srcAddr | dstAddr | action | src_team | src_service | src_environment |
|---|---|---|---|---|---|---|
| 2026-07-15 17:05:22.000 | 10.0.1.222 | 203.0.113.51 | ACCEPT | test | flow-log-test2 | development |
| 2026-07-15 17:05:22.000 | 10.0.1.222 | 203.0.113.52 | ACCEPT | test | flow-log-test2 | development |
実データでもエンリッチされたフィールドが付与されていることを確認できました。
まとめ
Lookup Table の作成からパイプラインの稼働確認まで、一連の設定を試しました。
これまで Logs Insights のクエリ内で都度対応づけたり、Lambda などでログを加工してから保存したりしていた静的なマッピングを、インジェスト時点で付与できます。Lookup Table に登録した範囲で、IP アドレスからチーム名などを自動的に解決します。
単純なエンリッチ用途であれば、クエリ側の工夫やログ加工用 Lambda の出番を減らせる可能性があります。エンリッチ済みのフィールドは Logs Insights の絞り込みやダッシュボードでそのまま利用できます。ログの加工手段の1つとして活用したいと思います。
