CloudWatch pipelines の新機能「条件付きログルーティング」を試してみた
はじめに
2026 年 10 月 14 日、CloudWatch pipelines が条件付きログルーティングと連鎖パーサーに対応しました。
条件付きルーティングは、パイプラインで抽出・加工したフィールドの条件に応じて、ログイベントの送信先ロググループを振り分ける機能です。送信先ごとにロググループを分けることで、保持期間やアクセス制御を用途に合わせて設定できます。たとえば、1 つのロググループに届く JSON ログから、後段の ETL や分析の対象から外すイベントを別のロググループへ分離できます。
この記事では、東京リージョンで、ダミーの JSON ログを level と type の条件で 2 つのロググループへ振り分け、次の 3 点を確かめました。
- どの条件のイベントがどのロググループに届くか
- 振り分けたイベントが元のロググループに残るか
- 振り分け先でタイムスタンプがどうなるか
あわせて、JSON をパースした後に grok でフィールドを抽出する、連鎖パーサーの動作も確認しました。
検証の構成
題材は Lambda Web Functions のプラットフォームログに似せた JSON で、1 件目は次の形です。
{"level":"WARN","type":"platform.invokeReport","seq":1,"region":"ap-northeast-1","record":{"requestId":"req-001","durationMs":11}}
ロググループは /test/pipelines-routing-1015/ の下に作り、以下では末尾の名前で source（送信元）、warn、summary と書きます。
パイプラインには、プロセッサーとして parse_json を 1 つ、シンクとして 3 つを設定しました。level が WARN のイベントは warn に、type が platform.invokeSummary のイベントは summary に送ります。どちらの条件にも一致しないイベントは、デフォルトルートとして元のロググループ（@original）に送ります。設定は次の YAML のとおりです。<ACCOUNT_ID> は、自分の 12 桁の AWS アカウント ID に置き換えてください。
pipeline:
source:
cloudwatch_logs:
aws:
sts_role_arn: "arn:aws:iam::<ACCOUNT_ID>:role/pipelines-routing-1015-role"
log_event_metadata:
data_source_name: "routing1015"
data_source_type: "default"
processor:
- parse_json: {}
sink:
- cloudwatch_logs:
log_group: "/test/pipelines-routing-1015/warn"
when: 'level == "WARN"'
- cloudwatch_logs:
log_group: "/test/pipelines-routing-1015/summary"
when: 'type == "platform.invokeSummary"'
- cloudwatch_logs:
log_group: "@original"
is_default: true
条件は、シンクを YAML に記述した順に評価されます。複数の条件に一致するイベントは、最初に一致したシンクへ送られます。
シンクが複数あるパイプラインでは、デフォルトルートを 1 つだけ指定し、最後に記述します。1 つのパイプラインに設定できるシンクは、デフォルトルートを含めて最大 10 個です。条件式は 4,096 文字まで指定できます。
送信元ロググループとパイプラインは、データソースタグで関連付けます。この例では、送信元の source に cw:datasource:name と cw:datasource:type のタグを設定します。
各タグの値は、YAML の log_event_metadata にある対応する値と一致させます。つまり、cw:datasource:name は data_source_name に、cw:datasource:type は data_source_type に対応します。両方の値が一致するロググループのイベントが、そのパイプラインの入力になります。
aws logs tag-resource --resource-arn arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source --tags cw:datasource:name=routing1015,cw:datasource:type=default
--configuration には、YAML 全体を文字列にして Body に入れた JSON ファイルを渡します。作成前に validate-telemetry-pipeline-configuration で検証できます。
jq -Rs '{Body: .}' routing-pipeline.yaml > routing-config.json
aws observabilityadmin validate-telemetry-pipeline-configuration --configuration file://routing-config.json
aws observabilityadmin create-telemetry-pipeline --name routing-1015-test --configuration file://routing-config.json
ソースロールの信頼ポリシーと権限ポリシー
本番では、信頼ポリシーに aws:SourceAccount や aws:SourceArn の条件を付け、権限ポリシーのリソースも対象の ARN に制限することを推奨します。この記事では、これらの条件や ARN を制限した構成での動作は確認していません。
信頼ポリシー（trust-policy.json）は、CloudWatch Logs にロールの引き受けを許可します。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": { "Service": "logs.amazonaws.com" },
"Action": "sts:AssumeRole"
}
]
}
権限ポリシー（source-role-policy.json）は、送信元のロググループへの ProcessWithPipeline と、振り分け先の作成・書き込みを許可します。送信元は、振り分けの検証用の source と、後述する連鎖パーサーの検証用の source-chain の 2 つです。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "LogsProcessing",
"Effect": "Allow",
"Action": "logs:ProcessWithPipeline",
"Resource": [
"arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source",
"arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/source-chain"
]
},
{
"Sid": "CreateRoutedLogGroups",
"Effect": "Allow",
"Action": "logs:CreateLogGroup",
"Resource": "*"
},
{
"Sid": "WriteRoutedLogEvents",
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "arn:aws:logs:ap-northeast-1:<ACCOUNT_ID>:log-group:/test/pipelines-routing-1015/*:*"
}
]
}
条件ごとの振り分け結果
17 件のダミーイベントを 1 回の put-log-events で送り、入力のパターンごとに振り分け先を数えました。
| 入力イベント | 件数 | 振り分け先 |
|---|---|---|
| level が WARN、type が platform.invokeReport | 5 | warn |
| level が INFO、type が platform.invokeSummary | 4 | summary |
| level が INFO、type が platform.invokeReport | 3 | 元のロググループ |
| level が WARN、type が platform.invokeSummary（両方の条件に合致） | 2 | warn |
| level なし | 1 | 元のロググループ |
| level が小文字の warn | 1 | 元のロググループ |
| JSON ではない文字列 | 1 | 元のロググループ |
振り分け先ごとの合計は warn 7 件、summary 4 件、元のロググループ 6 件で、入力の 17 件と一致しました。振り分けたイベントは元のロググループに残りません。
両方の条件に合うイベントは、YAML で先に書いた warn のルートに届きました。小文字の warn のイベントは level == "WARN" の条件に合致せず、元のロググループに届きました。
parse_json だけの今回の設定では、warn と summary に届いた本文は入力と同じ文字列でした。
振り分け先のタイムスタンプ
WARN と INFO を交互に計 4 件、それぞれ 1 時間前、30 分前、10 分前、現在のタイムスタンプを付けて送りました。時刻は JST です。
| 送ったイベント | 元のタイムスタンプ | 振り分け先 | 保存されたタイムスタンプ |
|---|---|---|---|
| WARN（seq 100）、1 時間前 | 10:08:55.756 | warn |
11:08:56.805 |
| INFO（seq 101）、30 分前 | 10:38:55.757 | 元のロググループ | 10:38:55.757 |
| WARN（seq 102）、10 分前 | 10:58:55.758 | warn |
11:08:56.805 |
| INFO（seq 103）、現在 | 11:08:55.759 | 元のロググループ | 11:08:55.759 |
warn に届いた 2 件は、元の時刻が 1 時間前でも 10 分前でも同じタイムスタンプになりました。この値は取り込み時刻（ingestionTime）の約 0.2 秒前で、パイプラインが処理した時点の時刻と見られます。前節で一括送信した 17 件は、送信時のタイムスタンプが 1 ミリ秒ずつ異なっていましたが、振り分け先に届いた 11 件はすべて同じタイムスタンプでした。
振り分け先のログを時刻で集計したりクエリしたりするなら、送る側でログの本文に時刻フィールドを含めておく必要があります。
不要なログの扱い
CloudWatch Logs をソースにするパイプラインでは、drop_events でイベントを破棄できませんでした。不要なログを後段の ETL や分析の対象から外すには、保持期間を短く設定した別のロググループへ振り分けます。
cloudwatch_logs ソースに drop_events を設定すると、validate-telemetry-pipeline-configuration で PROCESSOR_SOURCE_INCOMPATIBLE となり、create-telemetry-pipeline も ValidationException で失敗しました。今回は、保持期間を 1 日に設定した summary を事前に作成し、level が INFO の platform.invokeSummary イベント 4 件を振り分けました。
保持期間は、振り分け先ロググループの設定に従います。条件ルートの振り分け先ロググループは、存在しなければ初めてイベントを送る際に自動作成されます。今回、自動作成された warn には保持期間が設定されず、ログが無期限に保持される状態でした。保持期間を管理したい振り分け先は、パイプラインの作成前にロググループを作成しておくことを推奨します。
一方、デフォルトルートの振り分け先は自動作成されません。パイプラインを作成する前に、あらかじめ作成しておく必要があります。
連鎖パーサー
連鎖パーサーでは、parse_json の後に grok を配置し、JSON の message フィールドを解析する構成を試しました。送信元には別のロググループ source-chain を用意し、前述と同じ手順でデータソースタグを付けました。タグの値は、YAML の data_source_name に対応する chain1015 と、data_source_type に対応する default です。設定は次の YAML です。
pipeline:
source:
cloudwatch_logs:
aws:
sts_role_arn: "arn:aws:iam::<ACCOUNT_ID>:role/pipelines-routing-1015-role"
log_event_metadata:
data_source_name: "chain1015"
data_source_type: "default"
processor:
- parse_json: {}
- grok:
match:
message: ["%{WORD:method} %{URIPATH:path} %{NUMBER:status} %{NUMBER:duration_ms}ms"]
sink:
- cloudwatch_logs:
log_group: "@original"
シンクが 1 つだけのこの設定では、is_default を付けなくても検証を通り、動作しました。
message に文字列を持つ JSON を 3 件送った結果です。
| 入力の message | 結果 |
|---|---|
| GET /api/items 200 12ms | method が GET、path が /api/items、status が "200"、duration_ms が "12" |
| POST /api/orders 500 340ms | method が POST、path が /api/orders、status が "500"、duration_ms が "340" |
| not matching text | フィールドは追加されず、そのまま届いた |
grok が抽出した値はすべて文字列でした。パターンに一致したイベントの本文は、追加フィールドを含む JSON に書き換えられて保存されました。これは、parse_json だけの振り分け設定で本文が変わらなかった挙動とは異なります。
プロセッサーの配置や使用数の制限など、詳細は次のドキュメントを参照してください。
料金
CloudWatch pipelines 自体に追加料金はありませんが、通常の CloudWatch Logs の取り込み料金がかかります。CloudWatch Logs をソースにする場合、取り込み料金はパイプラインで処理する前に発生します。そのため、短期保持のロググループへ振り分けて後段の処理対象から外したログにも、取り込み料金はかかります。
まとめ
CloudWatch pipelines の条件付きログルーティングを使うと、ログの内容に応じて取り込み時に別のロググループへ振り分けられます。保持期間やアクセス制御が異なるログを分けられるほか、後段の ETL や CloudWatch Logs Insights で対象ロググループを絞ることで、処理対象やクエリのスキャン量を抑えられます。
CloudWatch pipelines 自体に追加料金はありませんが、取り込み料金は振り分け前に発生します。短期保持で抑えられるのはストレージコストです。
振り分け先では元イベントのタイムスタンプが保持されないため、元の時刻で集計・検索する必要がある場合は、ログ本文にも時刻フィールドを含める必要がある点にご留意ください。
