logstashでELBのログを2週間分だけAmazon Elasticsearch Serviceに取り込む

植木和樹

2016.04.20

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

こんにちは植木和樹＠上越妙高オフィスです。本日はAmazon Elasticsearch Serviceのお話です。

Webシステムで問題が発生した場合には、まずELBのログを調査することが多いかと思います。HTTP_5xxエラーの件数とか、日時とか、頻発しているパスとか。ログの量が少なければgrep, awkあたりで集計するのですが、量が多かったり分析の視点が多かったりする時にはツールのお世話になりましょう。

AWSではElasticsearch Serviceが提供されており、グラフィカルなログ分析もKibanaを使ってすぐに始められます。ただ、まずはS3に溜まった（大量の）ELBログファイルをElasticsearchに取り込む必要があります。

今回はログの取り込みにlogstashというツールを使ってみました。

環境構築手順

★2016/04/28追記
藤本さんの書いた CloudFrontのアクセスログをKibanaで可視化する｜ Developers.IO を参考に設定をブラッシュアップしてみました。

Elasticsearch Serviceクラスターの作成

まずはElasticsearch Serviceのクラスターを作成しましょう。手順はこちらのブログを参照してください。

[新機能]Amazon Elasticsearch Serviceがリリースされました！｜ Developers.IO

今回はお試しで m3.medium 1ノードで作成してみました。それとElasticsearchのアクセスポリシーは自アカウントのIAM Roleと特定のIPアドレスからのみアクセス可能にしています。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:ap-northeast-1:123456789012:domain/YOUR_ES_CLUSTER/*",
      "Condition": {
        "IpAddress": {
          "aws:SourceIp": [
            "203.0.113.1"
          ]
        }
      }
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::123456789012:root"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:ap-northeast-1:123456789012:domain/YOUR_ES_CLUSTER/*"
    }
  ]
}

ログ処理用EC2の起動

次に、S3からELBのログファイルをダウンロードしlogstashでElasticsearchにインポートするための、EC2を起動します。logstashはJRubyで作られており、それなりのCPUとメモリーが必要になります。t2.small（メモリー2GB）でもlogstashのレスポンスが悪かったので、c3.largeを使っています。スポットインスタンスを使えばお安く利用できるのでオススメです。

sshでログインするためEC2にはEIPを付けておきましょう。またS3とElasticsearchを利用できる権限でIAM RoleをEC2に付与しておきましょう。（ElasticsearchFullAccess Managed Policyがまだないようなので、今回はPowerUserAccessを与えています）

logstashのインストール

EC2にsshでログインしたらlogstashをインストールします。実際にElasticsearchを操作するとJSONを良く使うのでjqも入れておくと便利です。

なおAmazon Linux 2016.03だと、logstashのバージョン2.2.2を使わないと動きませんでした。古い1.5.xとかはエラーになりますので気をつけましょう。

$ wget https://download.elastic.co/logstash/logstash/packages/centos/logstash-2.2.2-1.noarch.rpm
$ sudo rpm -ivh logstash-2.2.2-1.noarch.rpm
$ sudo yum install jq -y

次にlogstashのoutput先にElasticsearch Serviceを利用できるようにするためのプラグインをインストールします。

$ sudo /opt/logstash/bin/plugin install logstash-output-amazon_es
Validating logstash-output-amazon_es
Installing logstash-output-amazon_es
Installation successful

logstashの設定

logstashがインストールできたら設定ファイルを作成します。

ELBのログファイルはS3から /home/ec2-user/elb にダウンロードします。（後述）
ログ行の解析には組み込みの%{ELB_ACCESS_LOG}パターンを使いました。
outputのamazon_esのhostsはElasticsearch Serviceのエンドポイントにあわせて変更してください。
ログの日付ごとに elblogs-2016-04-20 のようなインデックスに分割されてインポートされます。これで日付指定の検索も早くなりますし、古いログの削除も簡単になります。
ignore_olderで14日（14 * 24 * 60 * 60）以前のログは無視するようにしているので、期間を延ばす場合は修正してください。

logstash.conf

input {
    file {
        path => "/home/ec2-user/elb/*.log"
        type => "elb"
        start_position => "beginning"
        sincedb_path => "log_sincedb"
        ignore_older => 1209600
    }
}
filter {
    grok {
        match => [ "message", "%{ELB_ACCESS_LOG}" ]
    }

    date {
        match => [ "timestamp", "ISO8601" ]
        target => "@timestamp"
    }

    geoip {
        source => "clientip"
    }

    mutate {
        remove_field => ["timestamp", "message"]
    }
}
output {
    amazon_es {
        hosts => ["YOUR_ES_CLUSTER.ap-northeast-1.es.amazonaws.com"]
        region => "ap-northeast-1"
        index => "elblogs-%{+YYYY.MM.dd}"
    }
}

★2016/04/28 変更箇所

geoip を追加
timestamp を @timestamp に設定
timestamp と message を mutate
outputのdocument_type => "stream"を削除（こうするとtypeが input.file.type のままになります）
（課題）デフォルト定義の%{ELB_ACCESS_LOG}だとブラウザ情報が取得できない（独自にパターン定義する必要あり）

ElasticserachにString(not analyzed)のテンプレート設定

この状態でもインポートは可能なのですが、いざKibanaでリクエスト先のパスで集計しようとするとパスが単語分割(analyzed)されてしまいうまくいきません。そこでstringタイプのフィールドはanalyzedとnot_analyzedの2種類のフィールドが作成されるようにしました。こうしておくとリクエスト先のパスで集計したい場合はrequest.fullと指定することで単語分割されずに扱うことができます。

$ cat es_mappings.js
{
  "elb_template": {
    "order": 0,
    "template": "elblogs-*",
    "settings": {},
    "mappings": {
      "elb": {
        "_source": {
          "enabled": true
        },
        "dynamic_templates": [
          {
            "string_fields": {
              "mapping": {
                "index": "not_analyzed",
                "type": "string"
              },
              "match_mapping_type": "string",
              "match": "*"
            }
          }
        ],
        "_all": {
          "enabled": false
        },
        "properties": {
          "geoip": {
            "dynamic": true,
            "properties": {
              "ip": {
                "type": "ip"
              },
              "latitude": {
                "type": "float"
              },
              "location": {
                "type": "geo_point"
              },
              "longitude": {
                "type": "float"
              }
            }
          }
        }
      }
    },
    "aliases": {}
  }
}
$ curl -sX PUT https://YOUR_ES_CLUSTER.ap-northeast-1.es.amazonaws.com/_template/elb_template -d @es_mappings.js

★2016/04/28 変更箇所

全体的に不要フィールド削除
- string タイプをすべてnot_analyzedのみに変更
- _allをfalseに（明示的にフィールドを指定して検索するので）
geoipを追加
dynamic_templates の書き方がES v0.9xまでの定義方法（非推奨）なので、ES v1.xの方法に見直し
mappingsのtypeはelbに固定（logstash.confのinput.file.typeに対応します）
_sourceはtrueのまま（Descover画面でフィールドが表示された方が便利なので）

logstashの起動

準備ができたらlogstashを起動してプロセスを常駐させます。まだログファイルはダウンロードしていませんが、logstashは新しいファイルを（/home/ec2-user/elb/*.log）検出すると自動的に取り込みを開始してくれるので大丈夫です。

$ nohup /opt/logstash/bin/logstash agent --verbose -f logstash.conf -l logstash.log > logstash_nohup.log 2>&1 &
$ ps -ef | grep java

ログ取り込み中にsshセッションが閉じても大丈夫なように、nohupを付けて出力もlogstash_nohup.logに保存されるようにしています。tmux使え？はい、そうですね。

ELBログファイルのダウンロード

logstashが常駐したのを確認したらS3からELBのログファイルをダウンロードします。ログファイルは日付ごとにS3フォルダが分割されていて、直近1〜2週間分だけ対象にした時があったりするので、以下のようなシェルを用意して使ってます。

dl.sh

#!/bin/sh

declare FROM=4
declare TO=18
elbname=$1

declare BUCKET=$(aws elb describe-load-balancer-attributes --load-balancer-name "${elbname}" --query 'LoadBalancerAttributes.AccessLog.S3BucketName' --output text)
declare PREFIX=$(aws elb describe-load-balancer-attributes --load-balancer-name "${elbname}" --query 'LoadBalancerAttributes.AccessLog.S3BucketPrefix' --output text)
if [  x"${PREFIX}" != "x" ]
then
  BUCKET="${BUCKET}/${PREFIX}"
fi

# 2016/04/20現在 Amazon Linux 2016.03に入っているaws-cliは 1.10.8 なので
# aws sts get-caller-identity
# が使えません （要aws-cli >= 1.10.18）
user_arn=$(aws iam get-user --query 'User.Arn' --output text 2>/dev/null)
if [ $? -eq 0 ]
then
  account_id=$(echo "${user_arn}" | awk -F: '{print $5}')
else
  account_id=$(aws iam get-user 2>&1 | ruby -ne 'puts $stdin.gets.scan(/\d{12}/)[0]')
fi

for day in $(seq $FROM $TO)
do
  day=$(echo $day | sed -e 's/^\([1-9]\)$/0\1/')
  echo "${day}"
  aws s3 sync s3://${BUCKET}/AWSLogs/${account_id}/elasticloadbalancing/ap-northeast-1/2016/04/${day}/ . --exclude "*.log" --include "*${elbname}*" --exact-timestamps
done

自アカウントのELBからログファイルのバケットとプレフィックスを取得して、保存先のS3パスを組み立ててくれます。対象の日付フォルダが2016/04に固定されてて月またぎとか考慮してませんが、使う時には適当に修正してます。（妥協）

$ mkdir elb
$ cd elb
$ cat > dl.sh
（上のシェルをコピー）
$ sh ./dl.sh MY_ELB_NAME

ログファイルのインポート

S3からファイルのダウンロードが開始されると同時にlogstashがElasticsearchへの取り込みを開始してくれます。取り込みが終わったかどうかは、Elasticsearch ServiceのCPUモニターをみるか、logstashのステータスファイル（log_sincedb）の更新が止まったかで判断しています。

（おまけ）作業中に良く使ったコマンド

# ESのEndpointは良く使うので変数にいれておく
$ ES_HOST=YOUR_ES_CLUSTER.ap-northeast-1.es.amazonaws.com

# Elasticseachの状態を確認する
$ curl -sX GET "https://${ES_HOST}/" | jq "."

# 登録されているインデックス情報を確認する
$ curl -sX GET "https://${ES_HOST}/elblogs-*” | jq "."

# ELBログのインデックスを全部削除する
$ curl -sX DELETE "https://${ES_HOST}/elblogs-*” | jq "."

所要時間

今回Elasticsearch Serviceのノードは m3.medium を使用しました。ログファイルインポート処理にかかった時間を参考までに載せておきます。EC2よりもElasticsearchのCPUがボトルネックになるようなので、もっと早く取り込みたい場合はインスタンスタイプをあげると良さそうです。