脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - （1行目：日付、2行目以降：csvデータ）の日付を2行目以降の各行にtimestampフィールドとして追加

## はじめに
どうも！生産技術部で製品の検査工程を担当しているエンジニアです。[脱Excel Elastic Stack（docker-compose）でcsvログを分析・可視化 - Elastic Stackとは](https://qiita.com/TomoyukiSugiyama/items/619ab258ba230dfcf0a0)の続きになります。

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

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

```c
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](https://gitlab.com/iyama_yuki/elastic-stack/-/tree/master/)

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

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

### 方針１：logstashで1行ずつ読み込む
結論から言うとこの方法では上手くいきませんでした。[aggregate](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html)フィルタを使うと、複数のイベント間で情報を共有することが可能になるため、場合によっては可能です。

公式の例では、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](https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html)を用いることで、複数行のテキストを\n区切りでまとめて一つのイベントとすることができます。以下、3行のルールによってまとめることができます。

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


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

```c
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フィルタでパースし、日本時間として扱う](https://qiita.com/TomoyukiSugiyama/items/1633a1ce72cfc7ab38b9)に記載しています。

multilineに対してcsvフィルタをそのまま適用すると、最初の\nまでの`Date,2020/10/28,20:19:18`しかパースすることができません。[split](https://www.elastic.co/guide/en/logstash/current/plugins-filters-split.html)フィルタを用いることで、multilineを再度分解し複数のイベントに分けることができます。

最後の処理として数字以外で始まる行は、[drop](https://www.elastic.co/guide/en/logstash/current/plugins-filters-drop.html)フィルタで削除し、[mutate](https://www.elastic.co/guide/en/logstash/current/plugins-filters-mutate.html)フィルタでString型から型変換を行っています。

```c: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を割り当てることができるようになりました。
今後の記事では、下記のようなヒープのエラーなどについての対策方法を紹介したいと思います。

```systemd
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
```