クックパッド開発者ブログスマートフォン用の表示で見るクックパッド開発者ブログ2018-05-02S3に保存したログファイルをストリーム処理するサーバーレスアプリケーションの紹介インフラストラクチャー部セキュリティグループの水谷(@m_mizutani)です。クックパッドでは現在セキュリティ監視の高度化に取り組んでおり、その一環としてセキュリティ関連のログ収集およびその分析に力を入れています。ログ収集の部分では可用性などの観点からAWSのオブジェクトストレージサービスであるS3に一部のサービスやサーバのログをまず保存し、後から保存されたファイルを読み込んで分析などに利用しています。Glueなどを用いたバッチ処理がありますが、到着したログをなるべくストリームデータのように扱いたい場合もあります。セキュリティ関連のログでは以下のようなユースケースで利用しています。ログを検査してその中から危険度の高いと考えられるログを探し出し、アラートとして発報します。アラートの具体的な例としてはオフィス環境からの不審な通信やスタッフ用クラウドサービスでの不審な活動、スタッフPC上の不審イベントなどが挙げられ、ログの各種パラメータをチェックして危険度を判定します。S3に蓄積されたログの形式を変換して別のデータソースへ転送します。具体的にはGraylogに転送してログ分析をしやすいようにしたり、Athenaによって検索できるように変換して別のS3バケットに配置するといったことをしています。3分程度の遅延であれば許容できても、1時間単位の遅れが発生するのは少し困ります。できる限り到着したものから順次処理して分析までのレイテンシを短くしたいところです。本記事ではこのような処理に対し、S3に保存したログファイルをなるべくレイテンシが短くなるように処理するためのアーキテクチャ、そしてそのアーキテクチャのためのフレームワークを紹介したいと思います。Lambdaが起動し、その後起動したLambdaがオブジェクトそのものをダウンロードして処理する、という構成かと思います。この構成はシンプルで構築しやすいのですが、実際に運用してみるといくつかの課題が見えてきました。単独のサービスしか動かしていない場合は直接Lambdaを起動するだけで事足ります。S3にオブジェクトが生成されたタイミングで何か処理をしたいという要求が2つ以上でてくると、直接Lambdaを起動するのではなく別の方法を考えないとなりません。S3に保存されたオブジェクトを変換して別にデータストアに投入するというタスクでは、投入する先のデータストアに対する流量を気遣う必要があります。データストアの種類にもよりますが流量を超えすぎると全体のパフォーマンスが低下したり、データをロスするといったことが起きる可能性があります。予想される流量に対して受け側のキャパシティを確保してくのは重要なのですが、ログの種類によっては一過性のバーストがしばしば起こります。S3から直接Lambdaを起動しているとこういった任意のタイミングで外部から流量をコントロールするのは難しいです。AWSの機能でLambdaの同時実行数を制限してスロットリングさせる方法もありますが、処理順序が時系列と大幅にずれたり複数回失敗して自前で再試行しないとならないといったところで煩雑になってしまいます。システムやサービスを常に更新・拡張していると実運用におけるエラーの発生は避けられません。よってエラーが起きたリクエストに対して再試行するという場面はたびたびあるのですが、これをなるべく簡単にやりたいという思いがあります。Logsに実行時ログとして残すことができますが、実行時ログの量が多くなるとそもそも検索が難しくなるなどの問題があります。どのエラーに対して再試行したのか・していないのかを手動で管理することになり、煩雑になってしまいます。実際に使われているサーバーレスアプリケーションの構成図いくつかの課題を解決してスムーズな運用を試行錯誤した結果、上記の図のような構成に落ち着きました。シンプルな構成に比べるとそれなりに複雑なサーバーレスアプリケーションに仕上がりましたが、課題となっていた部分を解決しています。S3にオブジェクトが作成されてからどのように処理されるかというワークフローを、順を追って以下で説明します。S3のイベントはLambdaで直で受けるのではなくSNSで配信するようにしています。これによって複数の処理を実行したい場合でもSNSからイベントを受け取ることができるようになります。Streamにイベントを流します。Streamは一度データを投入すると指定時間(デフォルトで24時間)保持されるので、これ以降の処理を一時的に止めなければいけなくなってもイベントを蓄積し、あとから自由に読み出すことが可能です。いう2つを用意しています。送られてきたイベントの中身に応じてどちらかに振り分けます。それぞれのStreamに設定の違いはありませんが、この先にあるDispatcherの役割が異なります。非同期で呼び出します。名前をつけていますが、構成に違いはありません。StreamからはLambdaは直列&同期的に呼び出されるため、時間単位で呼び出される回数が抑制され、全体の流量が抑制されます。遅延を増やすことを想定しています。Alarmのような仕組みで遅延を調整するという方法が考えられます。イベント情報として作成されたS3オブジェクトのバケット名とキーが引き渡されるので、実際にS3からオブジェクトをダウンロードして必要な処理をします。処理が適切に終了しなかった場合、2回まで再試行されますが、それでも正常終了しなかった場合はDLQにエラーの情報が引き渡されます。ここでは実装上の都合などからSNSを使っています。ここまでで自動的に処理されるデータフローは一度終了します。これ以降は保存されたエラーをどのように対応するか、というワークフローになります。2回までの再試行である程度解消されると期待されますが、それ以外のコード由来のエラーなどは何度試行しても同じ結果になるためループさせるとシステム全体の負荷になってしまします。ここまでエラーで残ったものについてはユーザが確認してから再度処理のワークフローに戻すのが望ましいため手動で発火させる仕様にしています。エラーを全て、あるいは選択的に吸い出してます。Streamに放流します。Stream投入時にどちらのstreamに投入したかの情報を付与しているため、もともとのstreamに戻されます。このようなワークフローで処理することで、エラーが起きた場合でも少ない手間で修正・再実行し開発と運用のサイクルを回していくことができます。いうフレームワークを実装して利用しています。このフレームワークはPythonのライブラリとして動作し、導入したプロジェクトを先程のサーバーレスアプリケーションの構成で動かすための大部分をサポートしてくれます。サーバーレスアプリケーションを作成する時に苦労するポイントの1つにコンポーネントの設定や各コンポーネント同士のつなぎ込みといった作業があるかと思います。slipsは「S3からログファイルを読み込んで順次処理をする」というタスクを実行するサーバーレスアプリケーション作成におけるコンポーネントの設定、および統合をサポートしてくれます。これによって開発者がサーバーレスアプリケーションのコンポーネントを構成する際の消耗を最小限におさえ、ログデータを処理する部分の開発に集中できます。slipsの実態としてはCloudFormationのラッパーになっています。Frameworkが挙げられますが、ここの部分を自作したのは以下のような理由によります。冒頭で複数の処理に利用したいと書きましたが、最終的に読み取ったログの処理する部分以外はほとんどが共通した構成のサーバーレスアプリケーションになります。構成そのものをアップデートしてよりよい処理に変えていく必要があるため、複数のほぼ同じ構成を管理するのは冗長でメンテナンスの負荷が大きくなってしまいます。そのため、大部分の構成の定義を自動的に生成してCloudFormationで展開するというアプローチになっています。ほとんどが同じ構成と上述しましたが、一方で環境ごとに微妙な構成の差もあるためこれを吸収する必要があります。ある環境ではコンポーネントを自動生成できるが、別の環境では専用の承認フローに基づいて作成されたコンポーネントを利用する必要がある、といった状況を吸収する必要があります。既存フレームワーク内にこういった環境ごとの差分を簡単なロジックで書き込むことは可能ですが、数が多くなったりロジックが複雑になったりすると管理が難しくなるので、割り切ってPythonのコードとしてロジックを作成しています。このフレームワークを利用することで、先程のサーバーレスアプリケーションにでてきたAWS上の構成をほぼ自動で展開してくれます。先程のアーキテクチャの図を元に説明すると以下のような構成になります。S3へのログ保存とSNSへの通知設定、そしてS3に蓄えられたログをパースした後の処理(例えば、形式を変換したり、どこかへ転送したり、ログの内容を検査したりなど)を実施するためのコードは自前で用意する必要がありますが、それ以外の部分のコンポーネントをCloudFormationを利用してシュッと構築してくれます。これを利用することで、S3のログで何か処理をしたいユーザ(開発者)はS3に保存されたログが途中どのように扱われるかを意識することなく、最終的な処理の部分に集中することができるようになります。導入と設定ファイルの作成まずPythonのプロジェクト用ディレクトリを作成します。'git+https://github.com/m-mizutani/slips.git#egg=slips'次に設定ファイルを作成します。ありますが、本記事では要点のみ解説します。arn:aws:sns:ap-northeast-1:1234xxxxxx:seclog-event上記は構築するサーバーレスアプリケーションの各要素の設定を記述します。項では既存のIAMロールを割り当てることもできますし、省略すると必要なIAMロールをCloudFormationで自動的に追加します。StreamやDLQに使われるSNSも、ARNを指定すれば既存のものが利用され、指定しなければ必要なものが自動的に生成されます。これは組織や環境によって各リソースの管理方法が異なるため、別途作成されたリソースを使う場合とCloudFormationで自動的に生成する場合を切り替えることができるようにしています。Enterprise上のレポジトリで厳密に管理されていますが、開発環境では比較的自由にリソースを作成できるようになっています。このような環境にあわせて容易に展開できるような実装になっています。sample上記の設定は自分が実行させたい処理についての定義をしています。実行時に渡したい引数の内容を指定します。g-suite-login]上記のパートでどのS3バケットにどのようなフォーマットのオブジェクトがアップロードされるかを指定します。これを記述しておくことで、自分で書くスクリプトはパース済みのデータを受け取れるようになります。以下のキーがS3バケット名、その下のリストでS3キーのプレフィックスとパースするためのフォーマットを指定します。上記例にでてくる指定は以下のような意味になります。設定の準備ができたら、自分が処理したいコードを用意します。いうファイルを作成します。サンプルとしてログの件数をCloudWatchのカスタムメトリクスに送信するコードを書いてみます。クラスを継承してクラスを定義すると、これが読み出されて実行されます。新しく作成するクラスには以下のメソッドをそれぞれ定義します。最初に1度だけ呼び出されます。パースしたログ1件に対して1度呼び出されます。パースする過程で得られた付加情報が格納されています。終了時に呼び出されます。returnで値を返してテストなど利用しますデプロイ以上の準備ができたらAWSの環境にデプロイできます。ディレクトリには以下のようになっているかと思います。参考までに、サンプルの構成をgithubにあげておきます。コマンドを実行します。実行にはCLI用のAWSのCredential情報が必要です(詳しくはこちら)以下が実行結果になります。exitこれがうまくいくとバックグラウンドでCloudFormationが実行され、対象のアカウントに必要なリソースがバーンと展開されます。同時に、CloudWatchのダッシュボードも自動生成されるので、それを見るとこのサーバーレスアプリケーションの実行状況がある程度把握できます。Lambdaや各種マネージドサービスを使ったサーバーレスアプリケーションは便利に利用できる反面、実行の制御やエラー処理などの勝手がインスタンス上とは異なり、実行の制御やエラー処理をスムーズに実施できるような構成にする必要があります。これをサポートする機能やサービスもAWSには充実しており、クックパッドでは日々それらを活用しながら改善に取り組んでいます。今後、新たにサーバーレスアプリケーションを構築する方の参考になれば幸いです。