はじめに

どうも！生産技術部で製品の検査工程を担当しているエンジニアです。脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - Elastic Stackとはの続きになります。

対象読者

この記事では、Elastic Stackをそもそも知らない方、これから試そうと思っている方を対象としています。

この記事の内容

このようなcsvデータの先頭の日付を各行のtimestampとして扱う方法を説明します。

Date,2020/10/28,20:19:18
10,Test1,130.1,OK
20,Test2,1321,OK
30,Test3,50.2,OK
End
Date,2020/10/29,10:30:50
10,Test1,140.4,OK
20,Test2,1300,OK
30,Test3,50.0,OK
End
Date,2020/10/29,11:40:50
10,Test1,141.1,OK
20,Test2,1310,OK
30,Test3,55.8,NG
End

GitLabに設定ファイル一式を置いておきましたので参考にしてください。
リポジトリはこちら -> elastic-stack

方針

２つの方法で検討しました。

方針１：logstashに1行ずつ読み込む。日付がきたら保存し、後続の行に保存した日付をtimestampフィールドに設定する。
方針２：filebeatでEndまでを一つのイベントとしてまとめてlogstashに送り、logstashで1行ずつに分解し直す。

方針１：logstashで1行ずつ読み込む

結論から言うとこの方法では上手くいきませんでした。aggregateフィルタを使うと、複数のイベント間で情報を共有することが可能になるため、場合によっては可能です。

公式の例では、startからendまでの数字を加算し、TASK_ENDのイベントのフィールドに加算した結果を設定しています。

 INFO - 12345 - TASK_START - start
 INFO - 12345 - SQL - sqlQuery1 - 12
 INFO - 12345 - SQL - sqlQuery2 - 34
 INFO - 12345 - TASK_END - end

ただし、aggregateフィルタのtask_idに”12345”を設定することで、イベント間でデータの共有を行うことができます。今回のシチュエーションでは、task_idに相当するものがデータ内に存在しないため利用できませんでした。

方針２：filebeatで複数行をまとめる

うまくいった方法です。filebeatのmultilineを用いることで、複数行のテキストを\n区切りでまとめて一つのイベントとすることができます。以下、3行のルールによってまとめることができます。

  multiline.pattern: (End)
  multiline.negate: true
  multiline.match: before

@datake913様の記事Filebeatで複数行を扱うMultiline設定まとめが表になっていてわかりやすかったので参考にさせていただき、「pattern:(END)にマッチしない連続した行が、次のマッチする行の前に追加される」というルールを設定しました。下記のように1行にまとめることができます。

Date,2020/10/28,20:19:18\n10,Test1,130.1,OK\n20,Test2,1321,OK\n30,Test3,50.2,OK\nEnd

方針２の続き：logstashでmultilineを分解する

timestampのパース方法は、脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - multilineの中の"年/月/日,時:分:秒"をgrokフィルタでパースし、日本時間として扱うに記載しています。

multilineに対してcsvフィルタをそのまま適用すると、最初の\nまでのDate,2020/10/28,20:19:18しかパースすることができません。splitフィルタを用いることで、multilineを再度分解し複数のイベントに分けることができます。

最後の処理として数字以外で始まる行は、dropフィルタで削除し、mutateフィルタでString型から型変換を行っています。

logstash.conf

filter {
  grok {
    patterns_dir => ["/opt/logstash/extra_patterns"]
    match => { "message" => "%{TIMESTAMP_JP:read_timestamp}" }
  }

  date {
    match => ["read_timestamp", "yyyy/MM/dd,HH:mm:ss"]
    timezone => "Asia/Tokyo"
    target => "@timestamp"
  }

  split{}

  csv {
    columns => ["Step","TestName","Value1","Judge"]
    separator => ","
  }

  if [Step] !~ /\d+/ {
    drop{}
  }

  mutate {
    convert => {
      "Step" => "integer"
      "Value1" => "float"
    }
  }
}

最後に

multilineとsplitを用いて、複数行に同じtimestampを割り当てることができるようになりました。
今後の記事では、下記のようなヒープのエラーなどについての対策方法を紹介したいと思います。

java.lang.OutOfMemoryError: Java heap space
Dumping heap to java_pid1.hprof ...
Heap dump file created [3178685347 bytes in 34.188 secs]
warning: thread "[main]>worker11" terminated with exception (report_on_exception is true):
warning: thread "[main]>worker4" terminated with exception (report_on_exception is true):
java.lang.OutOfMemoryError: Java heap space