これは Tech Do Advent Calendar 2018 の20日目の記事です。
はじめに
弊社ではAkamai社のCDNサービスにお世話になっています。
Akamaiには各種AnalyticsやMonitoring系のサービスもいろいろありますが、
「Log Delivery Service」 という機能があって生のアクセスログをもらうこともできます。
生ログがあるのなら手元で集計や分析をしたいというのが今回の目的です。
概要
この「Log Delivery Service」ですが、残念ながらログの提供方法はメール送付もしくはFTPでPUTしてもらうしかないようなので、一旦ログはどこかに溜めて、逐次バルクロードする形で行きます。
以前はこういう場合Embulkを使っていましたが、今回はElastic Stackで完結してみます。
バージョンはElastic Stack 6.3の環境で行っていますが、インストール方法などは割愛します。
ざっくり以下の流れです。
１．Akamaiから送付されるログをある場所に保存する
２．Filebeatで 定期的に標準入力でLogstashへ送信しElasticsearchへ入れる
３．Kibanaで可視化する
なぜ標準入力なんかするのかというと、、、
- Filebeatでデータ送信したかったが、時間の都合で割愛・・
- Logstashはデーモンで動かしてログ保存ディレクトリを指定して置く方法もあるようだが、gzip形式のログ読み込みがうまくいかず割愛・・・（ドキュメント見るとgzipでも大丈夫なようなので後日あらためて挑戦）
Logstashのconfファイル作成
実はAkamaiのログだからといって特殊なことはなにもなく
いくつか追加のフィールドが選べますが、ApacheでいうCombined形式とほぼ同じです。（LTSVにしてほしい・・）
Logstashは、「input」「filter」「output」 の３セクションで構成されます。
- 「input」 : いろいろと方法があるが、今回は標準入力とした（stdin）
- 「filter」 : ログをパースしたり不要なfieldを削除したりする
- 「grok」 : フィールド名を指定してパース
- 「date」 : タイムスタンプのフィールド指定
- 「geoip」 : IPアドレスのフィールド指定（位置情報解析）
- 「mutate」 : 数値型を指定したり不要フィールド削除したり
- 「output」 : Elasticsearchのアドレスとインデックス名を指定
以下のようなconfファイルを作成しました。
input {
stdin {}
}
filter {
grok {
match => { "message" => "%{IP:clientip} - - \[%{HTTPDATE:timestamp}\] %{HOSTNAME:hostname} \"%{WORD:verb} /%{HOSTNAME:origin}%{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:response:int} %{NUMBER:bytes:int} \"(?:%{URI:referrer}|-)\" %{QS:agent} %{QS:cookie}" }
break_on_match => false
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
geoip {
source => ["clientip"]
}
useragent {
source => "agent"
target => "useragent"
}
mutate {
convert => {
"bytes" => "integer"
}
}
mutate {
remove_field => ["message","cookie"]
}
}
output {
# stdout { codec => rubydebug }
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-akamai_accesslog-%{+YYYYMMdd}"
}
}
ログをElasticsearchへ流し込む
いきなりElasticsearchへ入れる前に、Logstashのconfの「output」セクションを標準出力に変えて動作確認します。
stdout { codec => rubydebug }
output {
stdout { codec => rubydebug }
# elasticsearch {
# hosts => ["localhost:9200"]
# index => "logstash-akamai_accesslog-%{+YYYYMMdd}"
# }
}
以下のようにログファイルをzcatしてLogstashへ渡してみます。
設定ファイルは -f で指定します。
$ zcat access_log_hogehoge.gz | /usr/share/logstash/bin/logstash -f logstash_akamai.conf
〜〜〜(中略）〜〜〜〜
{
"bytes" => 131322,
"verb" => "GET",
"agent" => "\"hogehogehoge/SC-02J;Android26 8.0.0)\"",
"@version" => "1",
"origin" => "example.com",
"clientip" => "***.***.***.227",
"@timestamp" => 2018-11-30T16:57:46.000Z,
"geoip" => {
"timezone" => "Asia/Tokyo",
"continent_code" => "AS",
"country_code3" => "JP",
"country_code2" => "JP",
"country_name" => "Japan",
"region_name" => "Saitama",
"postal_code" => "350-0852",
"location" => {
"lon" => 1**.4853,
"lat" => 3*.9086
},
"region_code" => "11",
"longitude" => 139.4853,
"ip" => "***.***.***.227",
"latitude" => 35.9086,
"city_name" => "Kawagoe"
},
"timestamp" => "30/Nov/2018:16:57:46 +0000",
"response" => "206",
"host" => "elst-server.＊＊＊。＊＊＊",
"hostname" => "hogehoge.akamaized.net",
"request" => "/＊＊＊＊＊/****/****/***.jpg",
"httpversion" => "1.1",
"useragent" => {
"os_name" => "Android",
"os" => "Android",
"device" => "Generic Smartphone",
"build" => "",
"name" => "Other"
}
}
〜〜〜(以下略）〜〜〜〜
上記の感じで、各ログデータがいい感じにパースできました。
Logstashの設定ファイルで「output」セクションの設定を戻して、Elasticsearchにログを流していきます。
output {
# stdout { codec => rubydebug }
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-akamai_accesslog-%{+YYYYMMdd}"
}
}
kibanaで可視化
**「Management」→「Index Patterns」→「Create Index Patterns」と進み以下の設定をして「Create」**をしました。
「Index name or pattern」： logstash-akamai_accesslog-*
「Time Filter field name」： @timestamp
次に、**「Discover」**で先程指定したIndex Patternの 「logstash-akamai_accesslog-*」を選択してElasticsearchへ正常にログが入っていっているのを確認します。
着々と入っていっていてよさそうです。
あとは、**「Visualize」で、円グラフやデータテーブルなどを作成して、「Dashboard」**で、作成したグラフ等をいい感じに並べます。
最後に
継続的に可視化していくにはBeatsなども組み合わせて自動でElasticsearchに流していきたいところですが、
今回は一旦手動で流して可視化するところまでとしました。
試行錯誤しながらでしたので、あまりスマートな方法ではなかったかもしれませんが
CDNへのアクセスの傾向などが把握しやすくなりました。
参考記事