（ショロカレ 10 日目）S3 Event notification → Lamda → Amazon ES に挫折したので SQS → Amazon ES にしたメモ

ということで、これは「初老丸の独り Advent calendar 2015」の十日目の記事です。

tl;dr

S3 で動かしている静的サイトの access_log を S3 Event notification → SQS → Amazon ES という流れで可視化してみようと思ったのでメモ。本当は SQS ではなくて Lambda で挑戦したかったんだけど Lambda から IAM role で制御されている Amazon ES にポストするのに挫折してしまって SQS に頼ることにした。（何に挫折したかにも触れる）

メモ

構成

f:id:inokara:20151210214909p:plain

作ったもの

github.com

ログを正規表現で解析するのがつらかった（最終的にググったら同じことをされている方がいらっしゃったので拝借させていただいた）
Elasticsearch の Date Format の調整に苦労した（結局は ISO Format で放り込めた）

デモ

すでに上記のリポジトリを取得している前提で...

Mapping Template を放り込む

$ cd oreno-s3-access_log/es

#
# Amazon ES の Endpoint を修正する
#
$ vim regist-mapping-template.rb

#
# Amazon ES に Mapping Template を放り込む
#
$ ruby regist-mapping-template.rb
code: 200
msg: OK
body: {"acknowledged":true}

アプリケーションを Docker build する

$ cd oreno-s3-access_log/sqs

#
# Amazon ES のエンドポイント等を修正
#
$ vim Makefile
(snip)
DOCKER_RUN := docker run -d --name s3-access-log \
              --env ES_ENDPOINT="http://YOUR-ES-ENDPOINT:9200" \
              --env ES_PREFIX="YOUR_INDEX_PREFIX" \
              --env SQS_QUEUE_NAME="YOUR_QUEUE_NAME" \
              --env AWS_ACCESS_KEY_ID="AKxxxxxxxxxxxxxxxxxxxxxxxx" \
              --env AWS_SECRET_ACCESS_KEY="xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" \
              --env AWS_REGION="YOUR_REGION" \
              -v /etc/localtime:/etc/localtime:ro s3-access-log
(snip)

#
# アプリケーションのデプロイ
#
$ make build

今回は ES_ENDPOINT には Amazon ES のエンドポイントを設定することになる。

アプリケーションを起動する

#
# docker run をラップしている
#
$ make run

#
# コンテナの軌道を確認
#
$ docker ps
CONTAINER ID        IMAGE                COMMAND                  CREATED             STATUS              PORTS                    NAMES
5915f8932f0f        s3-access-log        "/usr/local/bin/super"   About an hour ago   Up About an hour                             s3-access-log

しばらく待つ

#
# Supervisord でアプリケーションを起動しているのでログを以下のように確認できる（キューを定期的にポーリングしている図）
#
$ docker exec s3-access-log /usr/local/bin/supervisorctl -c /app/supervisord.conf tail -f app
==> Press Ctrl-C to exit <==
21:56:34,261 INFO Calling sqs:receive_message with {u'QueueUrl': 'https://ap-northeast-1.queue.amazonaws.com/xxxxxxxxxxxx/s3-notification', 'MessageAttributeNames': ['*']}
2015-12-10 21:56:34,345 INFO Event does not exists...
2015-12-10 21:56:34,346 INFO Start polling...
2015-12-10 21:57:34,402 INFO Calling sqs:get_queue_url with {'QueueName': 's3-notification'}
2015-12-10 21:57:34,444 INFO Calling sqs:receive_message with {u'QueueUrl': 'https://ap-northeast-1.queue.amazonaws.com/xxxxxxxxxxxx/s3-notification', 'MessageAttributeNames': ['*']}
2015-12-10 21:57:34,525 INFO Event does not exists...
2015-12-10 21:57:34,526 INFO Start polling...
2015-12-10 21:58:34,585 INFO Calling sqs:get_queue_url with {'QueueName': 's3-notification'}
2015-12-10 21:58:34,626 INFO Calling sqs:receive_message with {u'QueueUrl': 'https://ap-northeast-1.queue.amazonaws.com/xxxxxxxxxxxx/s3-notification', 'MessageAttributeNames': ['*']}
2015-12-10 21:58:34,707 INFO Event does not exists...
2015-12-10 21:58:34,708 INFO Start polling...

Kibana を見てみる

f:id:inokara:20151210221603p:plain

ひとまず、Kibana 3 で見てみる。

Lambda → Amazon ES について何に挫折したのか

Lambda から Amazon ES にデータをポストしようとしたら以下のようなログが出た。

AuthorizationException: TransportError(403, u'{"Message":"User: anonymous is not authorized to perform: es:ESHttpPost on resource: arn:aws:es:ap-northeast-1:xxxxxxxxxxx:domain/oreno-es/s3_log-2015-12-08/s3_log"}')

なんや、権限が不足しとーとや、ほんじゃ、Lambda Function につけている IAM role に Amazon ES へのアクセスポリシーを付与すればよかろーもんって以下のようなポリシーを付与。

(snip)
    {
      "Effect": "Allow",
      "Action": "es:ESHttpPost",
      "Resource": "arn:aws:es:*:*:*"
    }
(snip)

付与しても状況は変わらず...。悩んで CloudWatch Logs の Amazon ES へのストリーム処理に利用する Node.js のコードを見てみると何やら Amazon ES のデータをポストする際に以下のような処理を行っていることが判明。

(snip)

function buildRequest(endpoint, body) {
    var endpointParts = endpoint.match(/^([^\.]+)\.?([^\.]*)\.?([^\.]*)\.amazonaws\.com$/);
    var region = endpointParts[2];
    var service = endpointParts[3];
    var datetime = (new Date()).toISOString().replace(/[:\-]|\.\d{3}/g, '');
    var date = datetime.substr(0, 8);
    var kDate = hmac('AWS4' + process.env.AWS_SECRET_ACCESS_KEY, date);
    var kRegion = hmac(kDate, region);
    var kService = hmac(kRegion, service);
    var kSigning = hmac(kService, 'aws4_request');
    
    var request = {
        host: endpoint,
        method: 'POST',
        path: '/_bulk',
        body: body,
        headers: { 
            'Content-Type': 'application/json',
            'Host': endpoint,
            'Content-Length': Buffer.byteLength(body),
            'X-Amz-Security-Token': process.env.AWS_SESSION_TOKEN,
            'X-Amz-Date': datetime
        }
    };

    var canonicalHeaders = Object.keys(request.headers)
        .sort(function(a, b) { return a.toLowerCase() < b.toLowerCase() ? -1 : 1; })
        .map(function(k) { return k.toLowerCase() + ':' + request.headers[k]; })
        .join('\n');

    var signedHeaders = Object.keys(request.headers)
        .map(function(k) { return k.toLowerCase(); })
        .sort()
        .join(';');

    var canonicalString = [
        request.method,
        request.path, '',
        canonicalHeaders, '',
        signedHeaders,
        hash(request.body, 'hex'),
    ].join('\n');

    var credentialString = [ date, region, service, 'aws4_request' ].join('/');

    var stringToSign = [
        'AWS4-HMAC-SHA256',
        datetime,
        credentialString,
        hash(canonicalString, 'hex')
    ] .join('\n');

    request.headers.Authorization = [
        'AWS4-HMAC-SHA256 Credential=' + process.env.AWS_ACCESS_KEY_ID + '/' + credentialString,
        'SignedHeaders=' + signedHeaders,
        'Signature=' + hmac(kSigning, stringToSign, 'hex')
    ].join(', ');

    return request;
}

(snip)