最近の砂場活動その4: CloudWatch Logs/Kinesis Firehose/S3/Athenaを利用してログ分析をやってみる

Fargateを使うようになり、EC2サーバーを管理しなくなってよくなりました(東京リージョン、7月ですね!)。一方、これまでサーバーにログとして吐かれていたファイルの閲覧や集計に困ることも増えてきました(sshでログインして、ログファイルをgrepなどができない)。AWSのサービスを組み合わせることでこれが解決できないか、ということをやってみました。個人運用のWebサービスにはオーバーキルですが、仕事で使うことも見越して...という感じです。素人が試行錯誤しながらやっているので、もっとよいやり方があると思います。

やりたいこと

一言でいうとNginxのアクセスログをいい感じで集計できるように、というのがやりたいことです。

FargateでWebアプリケーションサーバーとNginxを同一のタスクで動かしています。Nginxは静的ファイルへの振り分けとログ取りが主な目的です。Nginxのログをファイルに出力していると、タスクが終了した際にログが揮発してしまうので、標準出力に吐くようにします。そうすると、CloudWatch Logsに吐かれます。これをS3まで持っていければ、AthenaでSQLを書いたり、その結果をRedashで可視化することができます。今回はS3にログを運ぶやり方を探っていきます。

ログはltsv形式で吐いています。こんな感じ。

Nginxの設定ファイル

log_format tsv_go_active_learning "time:$time_local"
                                  "\thost:$remote_addr"
                                  "\tvhost:$host"
                                  "\tforwardedfor:$http_x_forwarded_for"
                                  "\tmethod:$request_method"
                                  "\turi:$request_uri"
                                  "\tprotocol:$server_protocol"
                                  "\tstatus:$status"
                                  "\tsize:$body_bytes_sent"
                                  "\tua:$http_user_agent"
                                  "\tapptime:$upstream_response_time"
                                  "\treqtime:$request_time"
                                  "\tupstream:$upstream_addr"
                                  "\tupstream_status:$upstream_status"
                                  "\thostname:$hostname"
                                  ;

upstream local-backend {
    server localhost:7778;
}

server {
    listen 7777 default_server;
    proxy_connect_timeout 10;
    proxy_read_timeout    10;
    proxy_send_timeout    10;
    access_log /dev/stdout tsv_go_active_learning;

    location / {
        root /www/app;
        proxy_set_header Host $http_host;
    }
    location /api {
        proxy_pass http://local-backend;
    }
}

CloudWatch LogsからS3までログを運ぶ手段

お手軽な方法としては、Lambdaを使って定期的にS3にエクスポートするのが常套手段のようです。

Lambdaを書けばいいのはお手軽ですが、リアルタイムに集計したい場合や一日のログが大きすぎてLambdaの制限時間の5分で終わらない!というの場合には困りそうですね。ある程度のログが溜まったら、それをトリガーにしてS3にエクスポートして欲しいです。調べてみると、こういう用途はどうやらKinesis Firehoseが得意なようです。CloudWatch Logsにはサブスクリプションという機能があり、これを使うとKinesis Firehoseにデータを流すことができます。

Kinesis Firehose、最初は何やってくれるのかさっぱり分からないですが、やってくれることとしては

リアルタイムにじゃんじゃか流れてくるデータを受け止める土管
[optional] 流れてきたデータをLambdaで変換できる
一定量、もしくは一定時間経過したらAWSのデータストアに横流ししてくれる
- 流す先はS3/Redshift/Amazon Elasticsearch Serviceなどの分析基盤となるサービスが中心
- 流す処理は自分でコードを書く必要がない、firehoseが面倒を見てくれる

が挙げられます。これまでのよくあるパターンとしては、fluentdで集約(aggregate)してS3などに流すというのがありましたが、集約サーバーの運用に手間がかかるというのがありました。Kinesis Firehoseを使うと、この手間をmanagedサービスが受け持ってくれるので手間が減らせるようです。

脱線: Kinesis Data Streams

もっと汎用的な土管としてはKinesis Streamsがあります。Firehoseは出力先(コンシューマーというらしい)がS3/Redshift/Amazon Elasticsearch Serviceなどに限定されていましたが、Streamsでは自前のプログラムから触ることができ、自由度が高くなります。仕事で作っているMackerelでもKinesis Streamsはお世話になっていて、mackerel-agentから投稿されるデータを引き受けてくれています。Kinesis Streamsで受け取ったデータは、Lambdaがコンシューマーとなり、RedisやDynamoDbにデータを書き込んでいます。

Kinesis Streamsは汎用的に使える一方で

シャード数は自分で管理する必要がある
コンシューマーのコードは自分で書く必要がある

といった手間が増えます。コンシューマーの書き込み先がS3/Redshift/Amazon Elasticsearch Serviceなどであれば、これらの手間を肩代わりしてくれるKinesis Firehoseを使うとよいようです。

AWS Black Belt Online Seminar 2017 Amazon Kinesis from Amazon Web Services Japan

Kinesis FirehoseとCloudWatch LogsのサブスクリプションでログをS3に転送する

はい、あとはやるだけですね。CloudFormationで構成管理しているので、その断片を貼っておきます。やっていることは大まかにはこんな感じです。

CloudWatch Logsに流れてくるデータをsubscribeしておく
- subscribe先はDeliveryStream
CloudWatch Logsに前回から50MBs以上データが溜まるか60秒以上経過していたらエクスポートを開始
エクスポートする際に変換をLambdaでやる
- subscribeするとjson形式になるので、元のltsv形式に戻すだけの変換です
- AWSが用意してくれているblueprintがあるので、それをちょっと変えればいいだけ
- こういったLambdaの構成管理にはSAMが便利でした

Firehoseの構成管理

  DeliveryStream:
    Type: 'AWS::KinesisFirehose::DeliveryStream'
    Properties:
      ExtendedS3DestinationConfiguration:
        BucketARN: !Join
          - ''
          - - 'arn:aws:s3:::'
            - Fn::ImportValue:
                !Sub "${S3StackName}:NginxLogsS3Bucket"
        BufferingHints:
          IntervalInSeconds: '60'
          SizeInMBs: '50'
        CompressionFormat: UNCOMPRESSED
        Prefix: firehose/
        RoleARN:
          Fn::ImportValue: !Sub "${IAMStackName}:DeliveryRole"
        ProcessingConfiguration:
          Enabled: 'true'
          Processors:
            - Parameters:
                - ParameterName: LambdaArn
                  ParameterValue:
                    Fn::ImportValue: !Sub "${SamStackName}:LambdaExtractMessageFromCloudWatch"
              Type: Lambda
  SubscriptionFilter:
    Type: "AWS::Logs::SubscriptionFilter"
    Properties:
      RoleArn:
        Fn::ImportValue:
          !Sub "${IAMStackName}:CloudWatchIAMRole"
      LogGroupName:
        !Ref NginxLogGroup
      FilterPattern: ""
      DestinationArn:
        Fn::GetAtt:
          - DeliveryStream
          - Arn

1分くらい待っているとS3のオブジェクトができあがってきます。日付が関連付けられています。

S3のファイルをAthenaで分析

S3にデータが集まれば、あとはAthenaでなんとでもなります。例えば処理時間がかかっているuriを列挙するのは、こんな感じです。最初にテーブルを作る。フルスキャンが走らないようにパーティションを切りましょう。

テーブル定義と分析のSQL

CREATE EXTERNAL TABLE IF NOT EXISTS go_active_learning_nginx.go_active_learning_nginx (
  `time` date,
  `host` string,
  `vhost` string,
  `forwardedfor` string,
  `method` string,
  `uri` string,
  `protocol` string,
  `status` int,
  `size` int,
  `ua` string,
  `apptime` double,
  `reqtime` double,
  `upstream` string,
  `upstream_status` int,
  `hostname` string
) PARTITIONED BY (
  year string,
  month string,
  day string
)
ROW FORMAT SERDE 'org.openx.data.jsonserde.JsonSerDe'
WITH SERDEPROPERTIES (
  'serialization.format' = '1'
) LOCATION 's3://my-bucket/firehose/'
TBLPROPERTIES ('has_encrypted_data'='false');

ALTER TABLE go_active_learning_nginx ADD PARTITION (year='2018', month='06', day='09') location 's3://my-bucket/firehose/2018/06/09/';
SELECT AVG(reqtime) as t, uri FROM go_active_learning_nginx WHERE year='2018' GROUP BY uri ORDER BY t DESC LIMIT 100;

こんな感じでブラウザで実行できます。