AWS Lambdaを使ってAmazon CloudWatch LogsからSplunkにログデータをストリーミング

このブログ記事は、アマゾン ウェブ サービス(AWS)のシニアソリューションアーキテクトであるRanjit Kalidasan氏との共同執筆です。

Amazon CloudWatch Logsでは、単一の拡張性の高いサービスを通じて、AWSの各種サービスのログ、およびAWSやオンプレミスサーバーで実行されるアプリケーションのログを一元管理できます。これにより、ログデータを簡単に表示したり、特定のエラーコードやパターンをサーチしたり、特定のフィールドに基づいてフィルタリングしたり、さらには今後の分析に向けて安全にアーカイブしたりすることができます。CloudWatch LogsのログデータをSplunkに取り込むことで、セキュリティ分析、アプリケーションのトラブルシューティング、監査コンプライアンス要件などのユースケースに活用できます。

CloudWatch Logsのサブスクリプションフィルター機能を使用して、ログイベントのリアルタイムフィードにアクセスし、Amazon Kinesisストリーム、Amazon Data Firehoseストリーム、AWS Lambdaといった他のサービスに送信することで、カスタム処理、分析、他のシステムへの取り込みを行うことができます。宛先のサービスに送信される際、ログイベントはbase64でエンコードされ、gzip形式で圧縮されます。

このブログ記事では、AWS Lambdaを使用してサブスクリプションフィルターを設定し、CloudWatch LogsデータをSplunk Cloud Platformや、AWS上で実行されるお客様管理のSplunk Enterpriseクラスター、オンプレミスのデータセンターでホストされるSplunk Enterpriseなど、さまざまなSplunkの宛先に取り込む方法についてご説明します。

サーバーレスアプリケーションのデプロイ

このブログ記事で使用するソリューションは、CloudWatch Logsのログイベントを処理し、HECエンドポイント経由でSplunkに取り込むLambda関数をデプロイします。このLambda関数にはログの取り込みをカスタマイズするオプションがあり、Splunkのソースタイプの指定、インデクサーの確認応答機能による取り込みデータの耐久性のチェック、デバッグなどを行えます。また、このLambdaアプリケーションは、サーバーレスアプリケーションとしてデプロイされます。ソースコードとデプロイ手順については、こちらのaws-samplesリポジトリを参照してください。作業には、AWS認証情報とAWS SAM CLI、およびPython 3.9がインストールされたデスクトップIDEが必要です。

このリポジトリをローカルに複製し、デプロイ手順に従います。

デプロイ中にアプリケーションのパラメーターを入力する際は、以下の参照情報を使用してください。

• SplunkSourceType：CloudWatch Logsデータのソースタイプを指定します。サポートされているすべてのAWSソースタイプについては、こちらを参照してください。

• DebugData：デフォルト値はfalseです。Lambda関数の問題をデバッグする場合は、trueに設定してください。Lambda関数のデバッグデータはCloudWatch Logsで確認でき、ネットワーク障害や確認応答の問題のトラブルシューティングに役立ちます。

• ELBCookieName：Defender ASRには、不正と判断された操作を効果的にブロックまたは監査できるExploit Guard機能が含まれています。以下に例を示します。

  • SplunkをAWS VPCでホストし、インデクサーにClassic Load Balancerを使用している場合は、Cookie名としてAWSELBを使用します。
  • AWS Application Load Balancerを使用している場合は、Cookie名としてAWSALBを使用します。Splunkをオンプレミスサーバーでホストしている場合は、ロードバランサーで設定されているCookie名を使用します。
  • Splunkがオンプレミスでホストされている場合は、ロードバランサーのカスタムCookie名を指定します。
  • Cookie名がわからない場合は、この値を空白のままにしてください。

• HTTPRequestTimeout：HTTPリクエストのタイムアウト値の秒数です。デフォルト値は5秒です。

• HTTPVerifySSL：HECエンドポイントへのHTTPリクエストのSSL接続を検証するかどうかをtrueまたはfalseで指定します。デフォルト値はtrueです。信頼できるCAなしで構成されたテストエンドポイントの場合は、falseに設定してください。

• SplunkAcknowledgementRequired：インデックスされたデータの確認応答をチェックするかどうかをtrueまたはfalseで指定します。デフォルト値はfalseです。

• SplunkAcknowledgementRetries：確認応答をチェックするための再試行回数です。デフォルト値は5です。SplunkAcknowledgementRequiredがtrueに設定されている場合にのみ適用されます。

• SplunkAcknowledgementWaitSeconds：データの取り込みから確認応答のチェックまでの待機時間の秒数です。デフォルト値は3です。SplunkAcknowledgementRequiredがtrueに設定されている場合にのみ適用されます。

• SplunkHttpEventCollectorToken：HECトークンの値です。これは必須のパラメーターです。

• SplunkHttpEventCollectorType：取り込みデータにrawまたはeventを指定します。デフォルト値はrawです。たとえば、VPCフローログの場合はrawを設定し、CloudTrailイベントの場合はeventを設定します。

• SplunkHttpEventCollectorURL：SplunkのHTTPエンドポイントのURLです。これは必須のパラメーターです。

これらのサーバーレスアプリケーションの入力パラメーターは、以下のようにLambda環境変数にマッピングされます。

デプロイ手順を実行すると、CloudFormationテンプレートが作成され、指定のAWSアカウントとリージョンに送信されます。CloudFormationスタックの作成が完了すると、Lambda関数が作成されます。デプロイされたLambda関数の物理IDをメモしてください。次のセクションでCloudWatch Logsのサブスクリプションフィルターを作成するときに、この物理IDが必要になります。

ソリューションのデプロイ

サブスクリプションフィルターを作成するには、CloudWatch Logsコンソールに移動し、ロググループを選択します。[Subscription filters]タブに移動し、Lambda関数用のサブスクリプションフィルターを作成します。

前の手順でサーバーレスアプリを使用して作成したLambda関数を選択し、サブスクリプションフィルターの名前を入力して[Start Streaming]を選択し、サブスクリプションフィルターを作成します。

これで、Splunkに取り込まれたデータを確認できるようになります。

ベストプラクティス

1. Lambda関数を拡張するには、予約済みまたはプロビジョニングされた同時実行数の設定を使用します。
2. SplunkインデクサーがVPC内、またはDirect ConnectやVPNなどを通じてAmazon VPCに接続されたオンプレミスのデータセンター内でプライベートにホストされている場合、VPCアクセスを有効にするようLambda関数を設定することで、CloudWatch Logsのデータを取り込むことができます。Lambda関数がSplunkインデクサーにアクセスするには、ルートテーブルエントリ、セキュリティグループルール、NACLルールなどを設定してネットワークアクセスを適切に構成する必要があります。
3. Lambda関数を単一のソースタイプ向けにデプロイし、そのソースタイプのデータがロググループに含まれていることを確認します。たとえば、CloudTrailログデータ用にLambda関数をデプロイし、CloudTrail用のロググループに設定した場合、VPCフローログには同じLambda関数を使用しないでください。VPCフロー用のロググループには別の関数をデプロイしてください。
4. デバッグにはDEBUG_DATA Lambda環境変数を使用し、トラブルシューティングにはCloudWatch Logs Insightsを使用します。CloudWatch Logs Insightsを使用したトラブルシューティングに役立つ情報を以下にご紹介します。

CloudWatch Logs Insightsを使用したトラブルシューティングと監視

CloudWatch Logs Insightsを使用することで、Lambda関数の実行に関する問題のトラブルシューティングや監視が可能です。さまざまなトラブルシューティングの場面で活用できるクエリーの例をいくつかご紹介しましょう。

エラーメッセージの確認

fields @message
    | parse @message "[*] *" as loggingType, loggingMessage
    | filter loggingType = "ERROR"
    | display loggingMessage

5分間隔でエラーメッセージの数を取得する場合：

fields @message
    | parse @message "[*] *" as loggingType, loggingMessage
    | filter loggingType = "ERROR"
    | stats count() by bin(5m)

接続エラーの確認

fields @timestamp, @message  |
filter @message like /Connection Error/ 

5分間隔で接続エラーの数を取得する場合：

fields @timestamp, @message  
| filter @message like /Connection Error/ 
| stats count() by bin(5m)

ネットワーク接続に接続エラーがないか確認します。Lambda関数がパブリック関数である場合、Splunkエンドポイントがインターネット経由でアクセス可能なパブリックエンドポイントであることを確認してください。Splunkエンドポイントがファイアウォールで保護されているため、Lambda関数からのアクセスを有効にする必要がある場合は、こちらのURLでAWSリージョン別のAWSサービスのパブリックエンドポイント情報をご確認ください。Lambda関数をVPCアクセス用に設定している場合は、それが設定されているVPCからSplunkエンドポイントへのネットワーク接続が可能であることをご確認ください。

確認応答の失敗のチェック

fields @timestamp, @message, @logStream, @log
| filter @message Like /Acknowledgement Failed/

5分間隔で失敗の数を取得する場合：

fields @timestamp, @message, @logStream, @log
| filter @message Like /Acknowledgement Failed/
| stats count() by bin(5m)

Lambda関数は、サーバーレスパラメーター SplunkAcknowledgementRequired またはLambda環境変数 ACK_REQUIRED がtrueに設定されている場合に、取り込みの確認応答をチェックします。確認応答が失敗する場合、関連するLambda環境変数 (ACK_RETRIESとACK_WAIT_SECS) の値を大きくしてみてください。

クリーンアップ

今後の課金が発生しないように、作成したリソースを次の順序で削除します。

1. CloudWatch Logsのサブスクリプションフィルターを削除する
2. サーバーレスアプリケーションのCloudFormationスタックを削除する

まとめ

このブログ記事では、Lambdaを使用してCloudWatch LogsのログデータをSplunkに取り込む方法について説明しています。このサーバーレスアプリケーションは非常に拡張性が高く、CloudWatchからAWSやサードパーティのあらゆる種類のログを、任意の場所で実行されているSplunkに取り込むことができます。Lambdaを使ってCloudWatchからSplunkに大量のログデータを取り込みたいとお考えのお客様にとって、効率的でコスト効果の高いソリューションといえるでしょう。