AWS Lambdaでアプリケーションログを処理する

こんにちは、技術開発グループの松原です。

AWSでログを処理するときは以下の方法が考えられるかと思います。

CloudWatch LogsのデータをAmazon Elasticsearch Serviceにストリーミングする
FluentdやLogstashを使ってログの収集、処理をする

Fluentdを利用する場合はAggregatorがあった方が、リトライやバッファリング、タグを追加したいなどの場合にAggregatorの設定を変更するだけですむなど利点が多いです。
ただし、サーバの台数が少ない場合はAggregatorを構築するほどでも無いと思う場合もあるかと思います。
そこで、今回は以下のような構成を試してみます*1。

ここではログの格納先をS3にしていますが、Kinesisに置き換えても同様に実現可能かと思います。

Fluentdの設定

Fluentdで集めたログデータをS3に送信するためにAmazon S3 Output Pluginを利用します*2。

td-agent.confは大体以下のようになります。
ここではApacheのログをS3に送信する設定例です。

<source>
  type tail
  path /var/log/httpd/access_log
  format ltsv
  time_format %d/%b/%Y:%H:%M:%S %z
  pos_file /var/log/td-agent/access.pos
  tag s3.td_database.td_table
</source>

# S3に入れる際にはタグ名を取得できないためテーブル名とDB名を指定する
<filter s3.*.*>
  type record_transformer
  enable_ruby false
  <record>
    dbname ${tag_parts[1]}
    tablename ${tag_parts[2]}
  </record>
</filter>

<match s3.*.*>
  type s3
  s3_bucket testbucket
  buffer_path /var/log/td-agent/s3
  format json
  flush_interval 5m 
  path s3/log/log_#{Socket.gethostname}_
  buffer_chunk_limit 5m   
</match>

td-agent.confの設定で重要な点は下記の通りになります。

S3にログを入れる際にはタグ名を取得できないためテーブル名とDB名をログに埋め込む*3
圧縮するかどうか。（S3に入れる際には圧縮した方が費用対効果が高くなる）
ログをS3に送付するサイズと間隔*4
どのサーバから送付されたログか識別できるようにする

Lambdaの構成

ログを処理する側のLambdaの構成は以下のようにします。

図にあるように2つのLambda Functionを用意します。

Bucketに対してObjectが作成された時（Event type:Object Created(All))に実行するLambda Function(上図のlambda (1))
1がInvokeされなかった場合などに削除されなかったObjectに対して 1 をInvokeするLambda Function(上図のlambda (2))

Fluentdの設定で触れたようにS3にObjectを作成時にLambda FunctionがInvokeされ無い場合に備えて、スケジュール実行する 2のLambda Functionを用意しています。

処理の流れは以下のようになります。

Lambdaの実行対象となるBucketにObject(ログファイル)がFluentdにより作成される。
1. Lambda FunctionがInvokeされる
2. ログ処理してTresureDataにログを格納する
3. 処理が完了するとInvokeの対象となったObjectを削除する
スケジュールタスクにより実行される
1. 1の対象となっているBucketのObject一覧を取得する
2. 作成され一定時間以上経っているObjectが存在する場合は、そのObjectを対象として 1 をInvokeする

今までの処理の流れの説明を追加すると下図のようになります。