はじめに：それぞれのツールについてふわっと理解

検索エンジンが〜とか、ログの整形が〜とかいう言葉を使うと猿にはわからないので、以下の言葉でふわっと理解。

• Logstash: 　ログを要素に分解するツール
• Elasticsearch: Logstashで分解したログの要素をフィルタリングするツール
• Kibana: Elasticsearchでフィルタリングした結果をグラフとか可視化するツール

環境づくり

まずは、ELKの環境準備。Macに直接brewでインストールする。
https://logz.io/blog/elk-mac/を参考に構築。

% brew install elasticsearch
% brew install kibana
% brew install logstash

ElasticsearchとKibanaを起動する

% brew services start elasticsearch
% brew services start kibana

起動しているか確認

% brew services list
Name          Status  User     Plist
elasticsearch started hogeee /hogeee/Library/LaunchAgents/homebrew.mxcl.elasticsearch.plist
kibana        started hogeee /hogeee/Library/LaunchAgents/homebrew.mxcl.kibana.plist

Kibanaの設定ファイル編集
/usr/local/etc/kibana/kibana.yml　（多分人によってパス違う）の、以下二行のコメントアウト外しておく。

server.port: 5601
elasticsearch.url: "http://localhost:9200"

これでElasticSearchとKibanaは完了。
アクセスできるか確認する。

Elasticsearch: http://localhost:9200

Kibana: http://localhost:5601

Logstashについては後の手順にて起動します。

ログの用意

何か分析するログがなければいけないが、無いので、自分で作る。

% cat log_genelator.sh
export LANG=en_US

while true; do

  d=`date '+%b %e %T'`

  r=`echo $((RANDOM%+101))`

  l=`echo $((RANDOM%+4))`
  h_list=(xxx aaa bbb ccc)
  h=${h_list[$l]}

  echo $d $r $h >> sample_log/sample.log
  echo $d $r $h
  sleep 10

done;

実行。

% sh log_genelator.sh

こんな感じでログがでる。

% tail -f sample_log/sample.log
Feb 11 12:08:21 17 xxx
Feb 11 12:08:31 68 bbb
Feb 11 12:08:41 62 ccc

Logstashの設定

先ほど出したログをLogstashで、Elasticsearchに送れる形に整形する。
logstashの設定ファイルを作る。今回はlogtest.confにするが、場所とファイル名は何でも良い。

結果としては以下の通り。

% cat logstash/conf.d/logtest.conf
input {
  file {
    path => [ "hogeee/sample_log/sample.log" ]
    type => "sample-log"
  }
}

filter {
  if [type] == "sample-log" {

    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{BASE10NUM:syslog_number} %{GREEDYDATA:syslog_message}" }
    }

    syslog_pri { }

    date {
      match => [ "syslog_timestamp", "yyyy-MM-dd-hh:mm:ss", "MMM  d HH:mm:ss" ]
    }

    mutate {
      convert => {
        syslog_number => integer
      }
    }
  }
}

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "syslog-test"
  }
  stdout { codec => rubydebug }
}

Logstashの設定ファイルは、以下のように、input、filter、outputの3セクションに分かれている。

input {
 #ログの読み込み方を記載する
}
filter {
 #ログの分解の仕方を記載する
}
output {
 #ログの書き出し方を記載する
}

inputについて

pathで、どこにあるログを読み込むか指定。（絶対パスでの記載が必要）
typeの名前は任意の名前、後のfilterセクションで使う。

input {
  file {
    path => [ "hogeee/sample_log/sample.log" ]
    type => "sample-log"
  }
}

filterについて

ここでの肝は、grokプラグイン。ログの内容を後々Elasticsearchで分析するために要素に分類する。

今回は、「Feb 11 12:08:21 17 xxx」のようにログが出るので、
「タイムスタンプ（Feb 11 12:08:21）　数字（17） メッセージ（xxx）」の三要素に分類する。

それぞれの要素は、「%{形式: タグ}」という風に定義づける。
タグの名前は任意の名前。後でKibanaにプロットする時に使う。
形式については、自分でゴリゴリ書くこともできるが、あらかじめ用意された形式を使うのが楽。

あらかじめ用意された形式については、以下リンクを参考にした。
https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns

もしゴリゴリ書くなら、こちらのツールを使うとgrokの文法チェックができるので便利。
http://grokconstructor.appspot.com/do/match#result

filter {
  if [type] == "sample-log" {

    grok {
      match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{BASE10NUM:syslog_number} %{GREEDYDATA:syslog_message}" }
    }

    syslog_pri { }

    date {
      match => [ "syslog_timestamp", "yyyy-MM-dd-hh:mm:ss", "MMM  d HH:mm:ss" ]
    }

    mutate {
      convert => {
        syslog_number => integer
      }
    }
  }
}

outputについて

Elasticに送るという定義はここで書く。
indexは任意の名前。後でKibanaに取り込む時に使う。

output {
  elasticsearch {
    hosts => ["127.0.0.1:9200"]
    index => "syslog-test"
  }
  stdout { codec => rubydebug }
}

文法チェック

以下コマンドで文法チェックができる。色々出るが、結果として「Configuration OK」が出力されればOK。

% logstash -f logstash/conf.d/logtest.conf --config.test_and_exit

※実施は注意※ Elasticsearchの情報削除

もし今までのデータを削除したい場合は、以下コマンド。
危険コマンドなので注意する。

% curl -XDELETE 'http://localhost:9200/*'
{"acknowledged":true}%

LogstashからElasticsearchにデータを送る

以下コマンドで、Logstashを使ってデータをElasticsearchに送る。
※先ほどkibanaの設定ファイルにElasticsearchのリンクを書いているので、Elasticsearchに送れば、Kibanaにプロットできる。

% logstash -f logstash/conf.d/logtest.conf

色々出た後に、以下のようにデータが取り込めていることを確認できる。
先ほどfilterで指定した、syslog_message、syslog_numberなどのタグの値についても無事取得できている。

{
                    "tags" => [
        [0] "_dateparsefailure"
    ],
        "syslog_timestamp" => "Feb 11 13:55:53",
                "@version" => "1",
         "syslog_facility" => "user-level",
         "syslog_severity" => "notice",
                 "message" => "Feb 11 13:55:53 12 ccc",
                    "type" => "sample-log",
          "syslog_message" => "ccc",
                    "path" => "/hogeee/sample_log/sample.log",
                    "host" => "hogeee",
           "syslog_number" => 12,
              "@timestamp" => 2018-02-11T04:55:53.960Z,
    "syslog_severity_code" => 5,
    "syslog_facility_code" => 1
}
{
                    "tags" => [
        [0] "_dateparsefailure"
    ],
        "syslog_timestamp" => "Feb 11 13:56:03",
                "@version" => "1",
         "syslog_facility" => "user-level",
         "syslog_severity" => "notice",
                 "message" => "Feb 11 13:56:03 35 bbb",
                    "type" => "sample-log",
          "syslog_message" => "bbb",
                    "path" => "/sample_log/sample.log",
                    "host" => "hogeee",
           "syslog_number" => 35,
              "@timestamp" => 2018-02-11T04:56:03.996Z,
    "syslog_severity_code" => 5,
    "syslog_facility_code" => 1
}

Kibanaにデータを取り込む

Kibanaにアクセスし、「Management」タブを開く。
先ほど、Logstashの設定ファイルのindexで指定した、syslog-testをindexに入力する。（アスタリスクについてはワイルドカードなので、今回のケースはあってもなくても良い）

次に進んで、@timestampを指定して、「Create index pattern」。

これで、データの取り込みが完了。

Kibanaで可視化する