Logstashを使ってみる - Kibanaを立ててみた

Logstashとは

さまざまなデータソースから情報を収集し、さまざまなstash＝格納庫にデータを投入する機能を提供するツールです。

Elaticsearchの文脈で語る上では、「Elasticsearchにデータを投入するためのエージェント」という位置付けになりますが、Logstash自身としては、プラグインを通じてデータベースやSlack、メールや、AWS S3など様々な対象にアウトプットすることができます。

beatsとの大きな違いは、収集した情報を転送する前に加工することができる、という点です。

今回は、MySQLに対して定期的に問い合わせた「SHOW GLOBAL STATUS」の結果をファイルに出力したものをInputとし、Elasticsearchに投入するケースに挑戦します。
また、beatsと異なる機能である加工処理も入れてみたいと思います。

Logstashのセットアップ

ダウンロードとインストール

Logstashの利用にはJava8が必要なので、事前にインストールした後、Logstashのインストールを進めます。

https://www.elastic.co/guide/en/logstash/5.5/installing-logstash.html#_yum

例によってyum リポジトリからのインストールなのですが、ドキュメントの内容を見ると、Elasticsearch本体と同じリポジトリのようなので、今回はElasticsearchとLogstashが同じ場所にある構成のときには特にすることはなく、いきなりyumでインストールできます。実運用の場合には、当然離れたところにLogstashをインストールすることになるかと思いますので、その際にリポジトリを登録することにしましょう。

yum リポジトリの設定が整ったら、インストールをします。

sudo yum -y install logstash

公式ドキュメントによると、ここで一旦、簡単な稼働試験を行うことになります。
https://www.elastic.co/guide/en/logstash/current/first-event.html

logstashのインストール先ディレクトリにあるモジュールを実行しつつ、Input-Fileter-Outputのパイプラインを直接指定して実行する試験です

yum/rpmインストールだと以下のコマンドで実行することになりますが、起動にそこそこ時間がかかるので、
「The stdin plugin is now waiting for input:」
が表示されるまでしばらく待ちましょう。

/usr/share/logstash/bin/logstash -e 'input { stdin { } } output { stdout {} }'

「The stdin plugin is now waiting for input:」の後ろにも何か表示が出るかもしれませんが、そこは気にせずに、「Hello world」でもなんでもよいので入力してEnterキーを押し、入力した文字列の前にタイムスタンプが付加されることを確認してください。

確認が取れたら、「CTRL-D」で終了します。

設定ファイル

公式ドキュメントだと、このあと、「サンプルのログファイルをFilebeatからLogstash経由でElasticsearchに登録」というチュートリアルが入りますが、ここでは省略します。
サンプルデータが提供されていますので、一連の流れを体感するためには、一度やってみるのも良いと思います。
https://www.elastic.co/guide/en/logstash/current/advanced-pipeline.html

こちらでは、チュートリアルを飛ばして、Logstashで直接ファイルを読み取ってElasticsearchに登録するための流れを追いたいと思いますが、そのために編集していく設定ファイルの話に触れます。

このあたりのお話です。
https://www.elastic.co/guide/en/logstash/5.5/config-setting-files.html

まず、Logstashの設定は2段構えです。

logstash自身の設定

/etc/logstash/
logstash.yml ：logstash自身の起動構成情報。パイプライン定義の位置なども管理。
jvm.options　：logstash自身はJava VM上で稼働するため、そのJVMのオプションを管理。
log4j2.properties：logstash自身のログ出力を管理するlog4jのオプションを管理。
startup.options　：logstashのインストール構成オプションを管理。

/etc/logstash/conf.d/

XXX.conf：個別のパイプラインIn-Processing-Outの設定

今回のような使い方の場合、logstash.ymlはデフォルトのまま、必要な設定は/etc/logstash/cond.d/XXX.confを作っていく事になります。

なお、あとでもう一度出てくる話題ですが、logstashをサービスとして起動する際、/etc/logstash/conf.d/の中のファイルをすべて読み込むので、バックアップファイルなど余計なファイルを置かないように、との注意書きがドキュメントの中にあります。

先に、Logstashをserviceとして動かせるように、と思ったらハマった！

RHEL6系なら「service（または/etc/init.d) 」、RHEL7系なら「systemctl」と思っていて、serviceコマンドを叩いたら、「そんなサービスいねえぞコラ！」、/etc/init.d/logstashを叩いたら「そんなファイルが見つからねーぞコラ！」って怒られました。実際ファイル探したら、ないのよね。ググると悲鳴（英語）がたくさん。

結論としては、

service（init.d）じゃなくて、「initctl」で実行しろ！
https://www.elastic.co/guide/en/logstash/5.5/running-logstash.html#running-logstash-upstart
の意訳

とのことでした。サービス起動のための設定ファイルは「/etc/init/logstash.conf」です。

よくわからないのは、
initctl status logstash
を何度か実行すると、その度にpidが違う、というところでしょうか。

また、先に紹介したチュートリアルをやっているとき、filebeat -> ポート5043 -> logstashという動きをするのですが、双方起動しているのにfilebeatからlogstashへの接続がポート閉塞でリトライ発生しているログが見受けられるのも、仕様がよく分かっていません。

なお、この際、気をつけなくてはいけないのは

サービス（initctl）で起動する時、パイプライン定義のファイルは「/etc/logstash/conf.d/」の配下に配置したすべてのファイルを読み込むので、バックアップとかゴミファイルも置いてはいけない。
https://www.elastic.co/guide/en/logstash/5.5/config-setting-files.html#_pipeline_configuration_files から超訳

ということです。

Logstashのパイプライン設定

サービス起動の話で遠回りしてしまいましたが、パイプライン＝Input - Fillter - Outputの設定です。

文字通り、入力とフィルタリング（及び加工諸々）と出力の定義で、ファイルの基本構造もとてもシンプルです。

input {
    # input setting
}
filter {
    # fillter setting
}
output {
    # output setting
}

それぞれにプラグインがあり、プラグインごとにサポートするパラメータが違ったりスルので一概に説明は出来ませんが、公式ページで見られるLogstashの図解のイメージを表現する場所が、まさにこの設定ファイルである、と言って良いでしょう。
https://www.elastic.co/jp/products/logstash

MySQLで、定間隔でSHOW GLOBAL STATUSをロギングしたファイルをElasticsearchに登録する流れを追ってみたいと思います。

ということで、
input : データソースとなるファイルに関する情報
output : elasticsearch（hostname：9200）に関する情報
となります。

サービス起動設定済なので、/etc/logstash/conf.d/の下に、XXX.confとして作ります。

以下内容のシェルを、cronで1分間隔で実行するものとします。

mysql -u root -pパスワード -h ホスト名 -e ”SHOW GLOBAL VARIABLES;”  | sed -e 's/\t/\" \"/g' | sed -e 's/^/\"/g' | sed -e 's/$/\"/g' | sed -e "s/^/$(date '+%Y\/%m\/%d %H:%M:%S') /g" >> /var/log/mysql_global_status.log

logstashのconfを以下のように作成します。
/etc/logstash/conf.d/mysql_global_status.conf

input {
    file {
        path => "/var/log/mysql_monitor/show_global_status.log"
        start_position => "beginning"
        type => "mysql-status"
    }
}
filter {
    if [type] == "mysql-status" {
        grok {
            match => { "message" => '%{DATESTAMP:date} "%{DATA:Variable_name}" "%{INT:Variable_value}"' }
        }
    }
  }
output {
    if [type] == "mysql-status" {
      elasticsearch {
        hosts => [ "localhost:9200" ]
        index => "lgs-mysql-status-%{+YYYY.MM.dd}"
        manage_template => false
        template_name => "mysql-status"
      }
    }
#     stdout { codec => rubydebug }
}

最低限の指定しかしていませんが、各ブロックについて解説します。

■Input
最も汎用的な「file input plugin」を使います。
https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html

loglotatedのように、ログローテションしてファイル末尾に数字を加えていくような仕組みにも対応しています。

■Filter
ちょっとクセがありますが、任意でログ出力フォーマットを定義したようなファイルに値しては、汎用的に解析できる「Grok filter plugin」を使います。

ファイルから読み込む行が、どのようなフォーマットで出力されているかを定義していきます。
今回のファイルのフォーマットは
「日付時刻（yyyy/mm/dd hh:mm:ss ”ステータス名（Variable_name）” ”ステータス値(Value)”」ですので、Grokの書式定義体にもとづいて「型：名称」を宛てて、上記のように指定します。

LogstashのGrokフィルタで使用できる定義体は、
https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
を参照します。

Grokフィルタの構文チェックは、
http://grokdebug.herokuapp.com/
に、読み込ませたいデータと、Grok定義を入力すると、そのフィルタでデータをどう読み込まれるかが確認できます。

KibanaにGrok Debuggerが付いた、とのことですが、有償プロダクトである「X-Pack」に含まれる、ということらしく、試していません。

ちなみに Grok Debugger は X-Pack の機能としてリリースされました。X-Pack ってあの有償プラグインの？と思われる方いらっしゃるかもしれませんが、Grok Debbuger は BASIC サブスクリプションから利用できますので無償で利用可能です。
http://dev.classmethod.jp/server-side/elasticsearch/x-pack-grok-debugger/

はい、わけがわかりません。

クラスメソッドさんは普段から良質な情報を提供してくださっていますし、Elasticの代理店でもあるので、ウソ言っているとは全く思っていませんが、公式ページからもそんな風に読み取れる部分がなかったです。<追記>
https://www.elastic.co/subscriptions
こちらに、Subscriptionの種類と対応の一覧表があるのですが、「Basic-Free」のところに「Grok Debugger」のチェックが入っていましたので、X-Packを入れつつ無償で使える！ということでした。

日本語サイト
https://www.elastic.co/jp/subscriptions
だと「Grok Debugger」の表記がないので混乱していましたが、単に更新が間に合ってないだけのようです。

■output
Elasticsearchにデータを登録するので、「Elasticsearch output plugin」を使います。
登録先となるElasticsearchのサーバーの場所と、格納するindex名を指定します。

設定ファイルを保存後、いきなり起動させるのではなく、confファイルに構文エラーがないか、チェックすることができます。

/usr/share/logstash/bin/logstash -f confファイルの場所 --config.test_and_exit

括弧のとじ忘れ、誤字脱字で存在しないキーワードを指定、などといったものを検出し、該当箇所を案内してくれます。

正しい構文であれば、実際にLogstashを起動させると、Elasticsearchにindexが作成されてデータが投入されますので、Kibanaから「Management」>「Index Pattern」>「Create Index Pattern」の画面からIndex name or patternに「"lgs-mysql-status-*”」を入力してindexが検出されることを確認しましょう。そのままCreateすれば、「Discover」>「gs-mysql-status-*」を選択することで、実際に登録されてきたデータを確認することができます。

また、このままだと、どうやらValueの値が文字列として処理されてしまうようです。
文字列で取得された値は、条件検索して検出した件数（count)で処理する分には使えるのですが、その数字そのものをメトリクスとしてグラフに反映することが出来ないようなので、数値として扱えるようにしてあげます。

Beatsのところでtemplate.jsonを適用したのと同じように、テンプレートを定義してあげます。

今回はインデックスを「 index => "lgs-mysql-status-%{+YYYY.MM.dd}"」で定義していますので、「"lgs-mysql-status-”」に該当するインデックスへの登録にはこのマッピングを適用する、というテンプレートを登録します。

curl -XPUT localhost:9200/_template/mysql-status -d '
{
  "template": "lgs-mysql-status-*", 
   "mappings": {
     "mysql-status": {
       "date_detection": false,
       "properties": {
         "@timestamp": {
           "type": "date"
         },
         "Variable_name": {
           "type": "text"
         },
         "Variable_value": {
           "type": "long"
         }
       }
      }
   }
}'

実際には、一度自動登録で作成されたマッピングを問い合わせて、出力結果を編集して、再登録という形でおこなっています。

コマンド一発でのテンプレート登録になっていますが、.jsonファイルに保存して、コマンドからファイルを読み込んで登録するのでも構いません。

仕様なのか、時間の関係なのか、うまく反映されないので、Logstashの停止、インデックスの削除、テンプレートの適用、Logstashの停止、Kibana->Managemen->Index PatternsからIndexの再検出、という手順を踏んでいます。

複数ファイルを取り扱う場合の注意！

前述の設定ファイルの記述において、Inputでtypeを定義し、Filter、Outputでif [type]=={}ブロックで記述しています。

同じ環境で１ファイルしか扱わないのであれば省略可能ですが、複数のファイルを扱い、それぞれファイルフォーマットが違ったり、Filterでの取り扱い方法、Output先のインデックスが異なる場合などは、インプットするファイル毎に、typeを使用して識別子をつけて上げる必要があります。

Logstashをサービスとして起動する際には/etc/logstash/conf.d/配下のファイルをすべて読み込むので、定義自体は１つのファイルの中で複数書いても、複数のファイルに分けて書いてもよいです。ただし、全体として１つの定義として機能する＝複数ファイルに書いた場合、マージして１つのファイルになったものと同様に処理されるので、１ファイル内に１つしかInput/Fileter/Outputを書いていないからと言ってTypeを定義していないと、ファイルから取り込んだデータがeralticsearchに登録されるタイミングで自動マッピングが作動する際に、それぞれのFilterによるフォーマットの解釈結果がマージされたインデックスが作成される事があります。ですので、「あとで２つ目のファイルを扱うことになったから、最初に登録した設定ファイルも書き換えなくては！」とならないよう、常にType定義とIfブロックを使う習慣をつけておいたほうが良いかと思います。

公式ドキュメントでもifブロックを省略した記述例ばかりですので、それに倣ってしばらく試していたのですが、上記の他にもう１つ設定ファイルを作っていて、ずっとそちらと混在したマッピングが生成される状況から抜け出せず、Elasticの中の人にTwitterから公式フォーラム（https://discuss.elastic.co）経由で解決支援していただきました。

Elasticさんすばらしい。これからお客さんその他ネット界隈にプッシュしていきます！