複数行にまたがるログのパースについて

初心者の者です。どうかご教授ください。
2行にまたがっているログを1ドキュメントとして取り出したいと思っています。
ログの中身は以下です。

[2019/04/15 11:19:26.743] ID=1787 DEV=東京01<改行>
ls<改行>

[2019/04/15 11:19:30.743] ID=1788 DEV=東京02<改行>
ls<改行>


というように、

[日時] ID=ID番号 DEV=DEV名<改行>
コマンド名<改行>

という2行セットのデータが多数並んだログになります。作成したconfigは以下のようなものです。
(動作に関係なさそうな部分は記載を端折っています)

input {
file{
path => " ・・・"
start_position => "beginning"
sincedb_path => "・・・"
codec => multiline{
pattern => "[" ←① [ から始まる行 というつもり
negate => "true" ←② "コマンド名"は①のパターンに該当しないので true とした
what => "next" ←③ "コマンド名"は ①に後ろに続くので next とした
}
}
}

filter {
grok {
match => { "message" => "(?%{YEAR}/%{MONTHNUM}/%{MONTHDAY}%{SPACE}%{HOUR}:%{MINUTE}:%{SECOND})] ID=%{POSINT:ID} DEV=%{HOSTNAME:DEV} %{GREEDYDATA:CMD }"}
}

"DEV"までは取り出せたのですが、どうしても2行目の"コマンド名"が取り出せません。
気になるのが、
・inputの中のmultilineの記載があっているか?
・match行の中の最後でコマンド名を取り出すのをそのままの並びで{GREEDYDATA:CMD }"とするのでよいのか? DEV=%{HOSTNAME:DEV} のあと改行を考慮しないでよいのか?

という点です。
ご教授よろしくお願いします。

inputの中身について

まずinputの指定です。
https://www.elastic.co/guide/en/logstash/current/plugins-codecs-multiline.html

ここにあります「Another example is to merge lines not starting with a date up to the previous line..」のところが今回のパターンと近しいのでイメージしやすいと思います。

日付の文字列で始まってなかったら、前の行データ(previous)にくっつけていくというサンプルです。

少し上の説明文を見ると、こう書いてあります。

The negate can be true or false (defaults to false ). If true , a message not matching the pattern will constitute a match of the multiline filter and the what will be applied. (vice-versa is also true)

negateをtrueにしたとき、メッセージがパターンにマッチしなかったら、whatの内容が適用される。
今回negateはtrueになっているため、日付文字列が入った [ で始まらなかったら、前の行に追加する、 と書けるので、inputはこんな感じにしました。
nextではなくpreviousにしています。


input {
  file {
    path => "Z:/z.log"
    start_position => "beginning"
    codec => multiline {
      pattern => "^\["
      negate => true
      what => "previous"
    }
  }
}

grokの書き方について

おそらく、multilineがうまく設定されていると、logstashのoutputにstdoutを指定しますと、どのようなデータがmessageに入っているかを確認できるかと思います。

改行コードにもよりますが、今回はCRLF(\r\n)で以下のようなテストデータを作りました。

[2019/04/15 11:19:32] ID=1790 DEV=tokyo.region.04
dir

logstashのgrokでは、改行コードも含めてpatternを書きました。

filter {
  grok {
    match => { "message" => "\[%{YEAR:year}/%{MONTHNUM:month}/%{MONTHDAY:day}%{SPACE}%{HOUR:hour}:%{MINUTE:minute}:%{SECOND:sec}\] ID=%{POSINT:ID} DEV=%{HOSTNAME:DEV}\r\n%{GREEDYDATA:command}\r" }
    
  }
}

とすると、こんな感じでパースできます。

messageに入った文字列をみて、Grokのパターンを考えるのがポイントです。

参考

Grokのパターンを考える際は、KibanaのDevToolsにあるGrok Debuggerを使うと、簡易に試せて便利です。

参考2

パースする側からしてみると、 [日付]部分がTIMESTAMP_ISO8601の形式になっていればもっと簡単にGrokのパターンが書けて良さそうです。

これらの中身は、7.0.0の環境で試しています。ご参考になれば幸いです。

pattern => "[" ←① [ から始まる行 というつもり
negate => "true" ←② "コマンド名"は①のパターンに該当しないので true とした

この状態で、whatをnextにしていたらどうなるの? というところです。

input {
  file {
    path => "Z:/z.log"
    start_position => "beginning"
    codec => multiline {
      pattern => "^\["
      negate => true
      what => "next"
    }
  }
}

のようにして、テストデータはこのようにしました。

[2019/04/15 11:19:33] ID=1790 DEV=tokyo.region.03
dir 3
[2019/04/15 11:19:34] ID=1790 DEV=tokyo.region.04
dir 4
[2019/04/15 11:19:35] ID=1790 DEV=tokyo.region.05
dir 5
[2019/04/15 11:19:36] ID=1790 DEV=tokyo.region.06
dir 6

2行目のコマンド行と1行目の結びつきが分かりやすいように連番をつけてあります。
これで実行しますと、messageは、こうなりました。

image

messageのところを見ていただくと、2行目のコマンドと次の1行目のデータがmessageに入ってしまっています。
結果、期待したGrokのパターンとも合致せずエラーになっています。

このことから、当初想定されています内容ですと、whatはpreviousにしておくのがやはり妥当ではないかと思います。

具体的な実験までしていただき、ありがとうございます。
2行目のコマンド行まで取れるようになったのですが、私が説明を割愛していたのですが 各データ
の2行と2行との間には空白行が挟まれています。やってみると、2行分(1セット分)しかとれませんでした。

これまでは空白行を無視するために、

if[message]=~"["{
}
else{
drop{}
}

というような行を入れていたのですがこれが機能しなくなったようにも思えます。(そもそも今回のmutiline行を入れたことでこの行はいれてはいけないものかどうかも分かっていないのですが)

2行分をパースしつつ、その下の空白行を無視するにはどうしたらよいのでしょうか?
よろしくお願いします。

何度も申し訳ありません。自分なりにwebで調べて、
if[message]=~"^\s*$"{
drop{}
}
というように空白行の記載を変えてみましたが変わらずでした。更に、ログの空白行を削除してみましたがそれでも1セットしか取れませんした。
configのmatch行は以下です。
match => { "message" => "(?%{YEAR}/%{MONTHNUM}/%{MONTHDAY}%{SPACE}%{HOUR}:%{MINUTE}:%{SECOND})] ID=%{POSINT:ID} DEV=%{HOSTNAME:DEV}\n%{GREEDYDATA:CMD}\n" }

DB側への投入結果はhttp://localhost:9200/xxx/_search?prettyで見ています。
空白行が有りのときも無しのときも、messageは以下のように出ています。
"message" : "[2019/04/04 16:19:26.743] ID=1787 DEV=tokyo.region.04\ndir"

よろしくお願いします。

認識に相違があるといけませんので、事前に確認させてください。
投入されたいデータとしては、このようなデータということで良いでしょうか?

テストデータ

[2019/04/15 11:19:33] ID=1790 DEV=tokyo.region.03
ls |

grep hoge
[2019/04/15 11:19:34] ID=1790 DEV=tokyo.region.04
dir 

|


grep fuga

取得したい期待値(間の改行コードがなくなっている。末尾に改行コードがない)

  • ls | grep hoge
  • dir | grep fuga

具体例を書かず、申し訳ありませんでした。
[2019/04/15 11:19:33] ID=1790 DEV=tokyo.region.03<改行>
ls<改行>
<改行>
[2019/04/15 11:19:34] ID=1790 DEV=tokyo.region.04<改行>
dir<改行>
<改行>



といったようなログになります。
よろしくお願いします。

インプットデータが以下のように2件のデータが入っていたとして、期待される結果と実際におきた結果を教えてもらえますか?

[2019/04/15 11:19:33] ID=1790 DEV=tokyo.region.03
ls

[2019/04/15 11:19:34] ID=1791 DEV=tokyo.region.04
dir



やってみると、2行分(1セット分)しかとれませんでした。

とあることから、上の例でいうとID1790のデータは取得できたが、ID1791のデータが取れないということが問題なのでしょうか?
そこを教えていただけませんか?

はい、おっしゃるとおりこの例でいうとID=1790のデータのみ抽出できますが
ID=1791のデータは抽出できませんでした。
どちらのデータもとりたい、と思っています。

よろしくお願いいたします。

ここで、再度inputの確認をしておきたいです。

    codec => multiline {
      pattern => "^\["
      negate => true
      what => "previous"
    }

この設定ですと、こういう動きになると思います。

  1. 先頭が ”[" で始まる場合、新しいデータ行と見なす
  2. 先頭が ”[" で始まらない場合、前のデータの続きと見做して、前のデータに行を追加する

いま、ID=1791のデータについてみてみますと、 "["で始まる行が後ろにないため、
ずっとID=1791の続きの行がこの後もあるはず、と待っている状態であるため、2セット取れると期待しているところ1セットになっているのだと思います。

この現象で、ファイル末尾にあるID=1791が取得できていないのを確認するための簡易な方法としては、
logstashを起動中に、対象のファイルを削除してしまう、別のフォルダに移動させるなどがあります。

ご回答、どうもありがとうございます。
確かに2セットのデータだけで試していたので1セット目しか取れない、と思い違いをして
しまっていましたが、セット数を増やしてみたらファイルの最下のセットだけ取れないで
その他のデータは取得できていることがわかりました。

ファイルの最下のデータのセットが取れない問題の対処方法としては、
①pattern行を「先頭が[以外」にする
②negateをtrueとする
③whatをnextとする
でしょうか。ちょっと試しにやってみようと思います。
ありがとうございます。

一番最後のセットが取得できないというのは、
ファイルを tailモードで読み込んでいて、まだ続きがあるかもしれないからとずっと待っているからです。

なので、「もう終わり」とLogstashが次のデータセットはないと判断できることができれば良いと考えられます。

https://www.elastic.co/guide/en/logstash/7.0/plugins-inputs-file.html#plugins-inputs-file-close_older
デフォルトは1時間になっています。

input {
  file {
    path => "Z:/z.log"
    start_position => "beginning"
    close_older => "60s"
    codec => multiline {
      pattern => "^\["
      negate => true
      what => "previous"
    }
  }
}

いまここを、60秒に設定すると、最後にファイルが書かれてから60秒たつとファイルを閉じようとして、(それ以上後続のデータがないと見做して)
最終のデータセットのところが出てくると思います。

もちろん、ファイルがどのように出力されるのか、ローテーションされるのかなど含めて
運用する際は、もっと考える必要がありますが、
単一のファイルで試している段階であれば、上記の設定で動きをみてもらうと良いと思います。

どうもありがとうございます。
試してみたら確かに最終の2行についても取り込むことができました。大変助かりました。
このclose_olderというオプションは、マニュアルでさらっと読んだことがあったのですが
英語力がないこともあって、どう使うのか、なんのためのものかわかっておりませんでした。
正直言って今もよくわかっていないので教えていただきたいのですが、
この値を60sに設定し、60s後に「ファイルが閉じ」た後、再度ログのおしりにデータを追加しても取り込んでくれたのですが、だとすると「ファイルを閉じる」とはどうイメージすればよいのでしょうか?今私が検討している仕組みでは、ログデータをリアルタイムにグラフ描画をさせたいので、タイムラグはなるべく小さくしたいと考えています。ですので、例えば3sにしたとすると、

新たなログが追加される(このときにファイルを開く?)→3s後に新規分がDBに取り込まれる&ファイルを閉じる→新たなログが追加される(このときにファイルを開く?)→3s後に新規分がDBに取り込まれる&ファイルを閉じる
が繰り返される、ということになるであってますでしょうか?
また、3sといったような短時間に設定することで何か懸念されることはございますか?

宜しくお願い致します。

どのような設定値にすべきかは、
対象とするログのファイルの数、書き込み頻度やファイルサイズ、
処理しようとしているマシンのスペックなどなど、様々な要因を考慮する必要があるため、一概には言えません。
たとえば、5秒に1回ぐらいログが継続して流れてくるのであれば、close_olderをわざわざ短くする必要もないわけです。

短時間にするのが良いのかも、どのようなログ出力なのか、使われ方はどうか、
要件やトレードオフ条件がないと、なかなか難しいように思います。
簡易的には、まず自身の環境、データで試してみるのが一番確実かと思います。

最後に「ファイルが閉じた」の部分は公式のドキュメントだと、この文が該当するかと思います。

https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html#plugins-inputs-file-close_older

If tailing, and there is a large time gap in incoming data the file can be closed (allowing other files to be opened) but will be queued for reopening when new data is detected.

Tailモードのとき、データ投入に大きな時間差があると、ファイルは閉じられる。
でも、ファイルの変更を検知するため、再びオープンすることがキューイングされる。

閉じられたあとに再びログが追記されたとしても、その部分だけが出力されるのは
Logstashに差分管理DBがあるため、どこまで読んだかは分かっているため再び続きのログが出力されるから、と考えたらよいと思います。

どうもありがとうございます。
なかなかイメージが難しいですね。。
reopenとあるので、newdataが入ったタイミングで開かれるのは確か、ということですね。
何度もありがとうございました。助かりました。

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.