AWS CloudWatch LogsからFirehose経由でログを収集

CloudWatch Logsを使用すれば、AWS環境からすべてのパフォーマンスログと運用ログを1カ所に収集することができます。この柔軟なプラットフォームのおかげで、多くのログソースを、ソースとログ形式ごとに複数のロググループにまとめて収集することができます。たとえば、VPCフローログ、CloudTrailログ、RDSログはすべてログ構造が異なります。この記事では、あらゆる形式のログファイルをCloudWatchからSplunkに取り込む方法、そして別のユースケースのために用意されたサンプルを拡張あるいは変更できるケースについてご紹介します。

Splunkは、AWSアドオンを使用する方法、Lambda 関数 （ラムダ式）やKinesis Firehoseを介してサーバーレスで取り込む方法、さらにはProject Trumpetを使用して自動的にサーバーレスで取り込む方法など、AWSデータを取り込むためのさまざまな方法を提供しています。しかし、CloudWatch Logsからの取り込みには、大規模なログ収集を実行でき、複数のAWSアカウントからログを収集する柔軟性を備えたKinesis Firehoseがおすすめです。

Firehoseの機能の1つに、Lambda関数を呼び出してログコンテンツの変換や処理を行うオプションがあります。これにより、CloudWatchログから取得したメッセージパッケージを開いてSplunkイベントに変換できるほか、ログコンテンツに基づいて、ソース、ソースタイプ、ホスト、インデックスなどのイベント情報を、柔軟に設定できるようになります。

この変換のバージョンの1つは、SplunkとAWSがFirehoseの統合を発表した際にすでにご紹介しています。この例で使用されたLambda関数は、VPCフローログを抽出したあと、それらのログをSplunkに送信します。この関数については、AWS LambdaのBlueprint、kinesis-firehose-cloudwatch-logs-processorまたはkinesis-firehose-cloudwatch-logs-processor-pythonでご覧いただけます。このブログではさらに一歩進み、CloudWatchのあらゆるログに使用できる、Splunkでの一般的なログ収集方法の基本事項についてご説明します。

以下の図は、Firehoseでログを取り込む方法のアーキテクチャをまとめたものです。

FirehoseとSplunkの設定

FirehoseとSplunkの設定に必要な手順のほとんどは、以前のブログでご確認いただけます。また、こちらに記載されている設定プロセスも参照してください。このページには、前述のブログに記載された手順のあとに実行する別の手順が記載されているほか、新しいLambda関数について追記されています。

新しいLambda 関数 （ラムダ式）の機能

以前のブログでは、Lambda関数のテンプレートを使ってログストリームから個々のログイベントを抽出し、そのイベントを変更することなくFirehoseに送信しています。これらはJSON形式ではなく単純なVPCフローログであるため、ログのコンテンツをRawイベントとして簡単にSplunkに送信できます。この標準テンプレートでは、注意するべきいくつかの重要なポイントがあります。

• イベントの詳細であるソースタイプ、ソース、ホスト、およびインデックスは、HEC設定にのみ設定できます。
• ログのソース情報の一部も失われます。たとえば、ロググループ名やリージョンはSplunkには送信されません。これはたとえば、AWSアカウント番号、ロググループ、またはログの発生場所への参照がログに含まれていない場合などに重要です。このようなログの場合、ロググループごとにFirehoseおよびHECのトークンまたは入力を個別に作成しない限り、ログのソースを追跡する方法がありません。

新しいLambda関数を使用すると、CloudWatchからログを収集し、JSON形式のSplunk HECイベントとしてパッケージできます。これにより、さらに以下のメリットが得られます。

• イベントの詳細であるソースタイプ、ソース、ホスト、およびインデックスを設定すると、HECトークンに設定された値をオーバーライドできます。
• アカウント、リージョン、ロググループ名やストリーム名などのログソース情報が保持され、Splunkに渡されます。

この関数テンプレートのサンプルは、以下のことを実行します。

1. CloudWatchからFirehoseに送信されたメッセージバンドルを受け取り、データを解凍します。
2. 各「message」ペイロードが「transformLogEvent」関数へと送信され、ペイロードメッセージがSplunk JSONイベント形式に変換されます。
   • イベントのHost値が、FirehoseストリームのARNに設定されます。
   • Sourceは、サブスクリプションフィルター名とCloudWatchのロググループ名の両方の値に設定されます。
   • ロググループ名に「CloudTrail」が含まれている場合はSourcetypeが「aws:cloudtrail」に設定され、「VPC」が含まれている場合は「aws:cloudwatchlogs:vpcflow」に設定されます。それ以外の場合はすべて、Lambda関数設定の環境変数の値に設定されます。
   • Indexは関数内では設定されませんが、ロググループ名またはサブスクリプションフィルター名の内容によって簡単に設定できます。
3. 結果をFirehoseに返します。

HECイベントの形式を設定する方法について詳しくは、こちらをご覧ください。

次のステップ

この変換関数の例では、Firehoseに送信されるイベントの形式をSplunk HEC JSON形式に変換し、ログ情報に基づいて、イベントの詳細の一部を設定する方法を示しています。関数を変更すれば、柔軟性を高めたり、要件に合わせたりできます。簡単な変更を行うだけで、Splunkのインデックス、ホスト、ソース、ソースタイプを独自の方法で設定できるようになります。以下に、2つの例を示します。

1) CloudWatchにログを送信するRDSインスタンスがある場合、1つのFirehoseで複数のRDSインスタンスやログタイプを使用できるようにロググループ名を使用できます。たとえば、2つのRDSインスタンスがあり、それらのログが以下のロググループに送信されるとします。

/aws/rds/mydatabse1/audit 

/aws/rds/mydatabse1/error 

/aws/rds/mydatabse2/audit

/aws/rds/mydatabse2/error 

監査ログとエラーログではソースタイプが異なるため、そのログが監査ログなのかエラーログなのかによって簡単にソースタイプの値を設定できます。この場合、同じFirehoseにサブスクリプションフィルターを追加するだけで、他のデータベースログからロググループを追加することもできます。Splunk側で変更を行う必要はありません。

2) もう1つの例は、複数のAWSアカウントからログを収集する際、セキュリティ上の理由からこれらのログを別々のSplunkインデックスに保存する必要がある場合です。変換関数にインデックスの値を設定すれば、関数によって各アカウントに異なるインデックス名が設定されます。

Splunkのメリットをどうぞお試しください。