re:Invent2025で Amazon CloudWatch に新機能 CloudWatch pipelines が発表されました。
CloudWatch pipelines とは
ログ、メトリクス、トレースなどのテレメトリデータを収集し、変換・フィルタリングなどの変換を行い、最終的な宛先へ渡すためのパイプラインを作成する機能です。
これまで複数のツールや複雑なETLパイプラインを用意する必要があったデータ管理を、CloudWatch上で一元化できるようになりました。
プロセッサの種類
パイプラインの中でプロセッサを使用することで、どのように変換・解析するかを設定できます。
解析には以下5つが執筆時点でサポートされています。
- OCSFプロセッサ
- Open Cybersecurity Schema Framework (OCSF) 標準に従ってログデータを解析
- CSVプロセッサ
- CSV 形式のデータを構造化フィールドとして解析
- Grokプロセッサ
- Grokパターンを使用して非構造化データを解析
["%{WORD:level} %{GREEDYDATA:msg}"]のようにパターンを指定します
- JSONプロセッサ（JSONを解析）
- JSON データを構造化フィールドとして解析
- Key-Value
- キーと値のペアの形式のデータを構造化フィールドとして解析
解析するデータ形式を選択した後は、変換やフィルターを選択します。
以下の画像は一部ですが、Keyを追加・削除したり、文字列を変換したりなどできます。
やってみた
今回は AWS Step Functions が出力するログから、以下のJSONのような「実行成功（ExecutionSucceeded）」イベントだけを抽出し、不要なフィールド（redrive_count等）を削除します。
Step Functions のログは以下のように \ 区切りで出力されることがあります。
{
"details": {
"output": "{}",
"outputDetails": {
"truncated": false
}
},
"redrive_count": "0",
"id": "4",
"type": "ExecutionSucceeded",
"previous_event_id": "3",
"event_timestamp": "1765379835660",
"execution_arn": "arn:aws:states:xxxxx"
}
1. 検証用 AWS Step Functions 作成
まずは検証用の AWS Step Functions を作成します。
空白から作成を選択し、ステートマシンのタイプは標準で作成します。
ログが出力できれば良いので、Pass stateだけのワークフローを作成します。
設定でログ記録を有効化し、ログを出力させます。
設定できたら保存し、ステートマシンを作成します。
作成後はログ比較ができるように一度動かしてみます。
「実行」からステートマシンを実行します。入力欄には何も入力せず、空欄のまま実行を開始します。
実行完了後、このような形式のログがCloudWatch Logsに出力されているのを確認してください。
{
"details": {
"output": "{}",
"outputDetails": {
"truncated": false
}
},
"redrive_count": "0",
"id": "4",
"type": "ExecutionSucceeded",
"previous_event_id": "3",
"event_timestamp": "1765379835660",
"execution_arn": "arn:aws:states:xxxxx"
}
2. CloudWatch Pipelineの作成
まず、CloudWatch のコンソールから「Ingestion」を選択し、パイプラインを作成します。
データソースに「AWS Step Functions ログ」を選択し、お好きなパイプライン名を設定します。「Assign tags to pipeline」は何も設定しないままで次へ進みます。
設定の詳細に表示されている「ログソースタイプ」は、ExpressとStandardが選択できます。
1で標準のステートマシンを作成したので、Standardを選択します。
Service accessはこのパイプラインへのサービスロールの選択です。特に理由が無ければ、一番上の「Auto create and use a new service role」で新たにロールを作成し、このパイプラインのサービスロールとして使用します。
次の送信先は変更できないため、このまま次へ進みます。
CloudWatch Logsのログ変換に関しては、既存のログを変換して上書きするように保存されてしまうので注意です。（そのために変更ができないようになっています）
最後にプロセッサを設定します。
今回はログの文字列のうち、特定のKeyを削除してみます。
元のログは以下のような文字列でした。messageの下に階層構造のJSONがあります。
{
"details": {
"output": "{}",
"outputDetails": {
"truncated": false
}
},
"redrive_count": "0",
"id": "4",
"type": "ExecutionSucceeded",
"previous_event_id": "3",
"event_timestamp": "1765379835660",
"execution_arn": "arn:aws:states:xxxxx"
}
まずはJSONへ変換します。
追加を選択するとどの欄も編集できないinputが表示されますが、これで大丈夫です。
プロセッサを更に追加し、「エントリを削除」を選択します。
2つのプロセッサを追加したら、次へ進みます。
確認の画面に遷移したら、設定内容を確認し、パイプラインを作成します。
3. パイプラインを動かす
これまででパイプラインの用意が出来たため、Step Functionsを実行し、パイプラインで変換されたログを確認してみます。
1で作成されたステートマシンを再度実行し、Step Functionsのログを確認します。
3つのKeyが削除され、簡素なログに変換されています。
まとめ
これまで変換処理が必要だったログ整形が、CloudWatch上の設定だけで完結できて、とても便利そうです。
今回はエントリ(key)の削除だけを試しましたが、他にも様々な変換処理があります。解析しにくいログがあれば、CloudWatch pipelinesをぜひ検討してみてください。
補足: テレメトリ設定がオンになっていません
もし「テレメトリ設定がオンになっていません」とメッセージが表示された場合は、
CloudWatch > Ingestion > データソース に表示されている「Enable resource discovery」でオンにします。
簡単30秒