Logstash on DockerでCSVデータ前処理ツールを作る

業務でLogstashを用いてCSV形式のログデータを加工するツールを作りました。これが予想外に大苦戦。。。「Pandasだと一瞬で終わりそうなのに。。。」という考えが何度も頭をよぎりましたが、ここは諸事情でLogstashオンリーという制約下で格闘。。。 :japanese_ogre::japanese_ogre::japanese_ogre: また同様の業務をやる時のために、備忘録を残しておこうと思います。:dancer::dancer_tone1::dancer_tone2::dancer_tone3::dancer_tone4::dancer_tone5:

# データの形式
今回は以下の様な入力データをLogstashで読み込み、各種filter処理を実行し、出力イメージとして示している様な形式のデータを出力します。

```csvs:入力データ
a,b,c,d,e: あいうえお: 日本語: カタカナ
```

```csvs:出力イメージ
{"test1":"a","test2":"b","test3":"c","test4":"d","tag1":"e", "tag2":"あいうえお","tag3":"日本語","tag4":"カタカナ"}
```

# Logstash on Dockerの準備
今回はLogstashのフィルター機能などを研究しながら作業を進める必要が有ったため、環境構築ややり直しを容易に実行出来るDockerのコンテナを利用することにしました。LogstashのコンテナイメージをコンテナとしてDocker Composeで起動し、所望のログ編集処理を実行します。環境構築には以下の2つのファイルを用意しまし、以下のディレクトリ構成の様に格納しました。

```bash:ディレクトリ構成
logstashtest/
|-- docker
|   |-- config
|   |   `-- logstash.conf
|   `-- docker-compose.yml
`-- logfile
    |-- input_logs.csv
    `-- output_logs.csv

3 directories, 4 files
```

入力データと出力データを格納するディレクトリ(`logfile`)と、Logstashの内部処理を定義する`logstash.conf`ファイルをを`volumes`項でコンテナ外のファイルと紐づけています。

```yaml:docker-compose.yml
version: "2.4"
services:
    logstash:
        image: docker.elastic.co/logstash/logstash:7.13.2
        volumes:
            - ../logfile:/usr/share/logstash/logfiles
            - ./config/logstash.conf:/usr/share/logstash/pipeline/logstash.conf
```

データの読み込み→加工→出力の順に処理を記載します。

```conf:logstash.conf
input {
    file {
        path => ["/usr/share/logstash/logfiles/input_logs.csv"]
        start_position => "beginning"
        sincedb_path => "/usr/share/logstash/mysincedb"
        codec => plain {
            charset => "SJIS"
        }
    }
}
filter {
  # Add column names
  csv {
      columns => [
          "test1",
          "test2",
          "test3",
          "test4",
          "test5"
      ]
  }
  # Create some new tags
  grok {
      match => {
          "test5" => "%{DATA:tag1}: %{DATA:tag2}: %{DATA:tag3}: %{GREEDYDATA:tag4}"
      }
  }
  # Remove unnecessary field names
  mutate {
      remove_field => [
          "@version",
          "host",
          "path",
          "@timestamp",
          "message",
          "tags",
          "test5"
      ]
  }
}
output {
    if ([test1])  {
        file {
            path => ["/usr/share/logstash/logfiles/output_logs.csv"]
        }
    }
}
```

## 読み込み部(input)
`start_position => "beginning"`でファイルの先頭から読み込みます。`sincedb_path`に任意のファイルパスを指定することで、どこまでファイルを読み込んだかをバイト単位で記録してくれるsicedbにすぐにアクセス出来ます。`codec`項では文字コードを指定できます。

## 処理部(filter)
3種類のフィルターを用います。csvフィルターで読み込んだデータにカラム名を追加します。このカラム名を用いて、以降の行程を実行します。続けて、grokフィルターで"test5"項で正規パターンに一致する部分の分割方法を定義します。ここではDATA(任意の文字の繰り返し)形式の項目が`: `区切りになっている部分を分割し、定義したカラム名`tag〇`を追加します。最後に`mutate`フィルターで出力データには不要なカラム名を削除します。

## 出力部(output)
if文の使い方の備忘録として条件分岐を記載していますが、この分岐は無くても動作します。この書き方でtest1項が存在する場合、パスで指定したファイルに加工されたデータが書き込まれます。

# 実行結果
実行結果は以下の様になりました。辞書形式なので並びはランダムになっていますが、イメージ通りの結果が得られました。

```json
{"test4":"d","tag1":"e","tag2":"あいうえお","test3":"c","test1":"a","tag3":"日本語","tag4":"カタカナ","test2":"b"}
```

# Reference
* [Logstash File input plugin](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html#plugins-inputs-file)
* [Logstash File output plugin](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-file.html#plugins-outputs-file)
* [Logstash Grok filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-file.html#plugins-inputs-file)
* [Logstash CSV filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-csv.html)
* [Logstash on Docker config](https://www.elastic.co/guide/en/logstash/current/docker-config.html#docker-config)
* [はじめての Logstash](https://qiita.com/nskydiving/items/0cb598de7ffb5c22424d)
* [Logstash File input: sincedb_path](https://stackoverflow.com/questions/29573097/logstash-file-input-sincedb-path)
* [Logstashの実践的な説明](https://www.elastic.co/jp/blog/a-practical-introduction-to-logstash)
* [Kibana(Elasticsearch)＋Logstashでオタクの出費を可視化してみる](https://qiita.com/estaro/items/8d4e295da0be00827947)
* [logstash-pattern @ GitHub](https://github.com/hpcugent/logstash-patterns/blob/master/files/grok-patterns)
* [logstash check if field exists](https://stackoverflow.com/questions/30309096/logstash-check-if-field-exists)
* [logstash条件付き出力をelasticsearch（filebeatホスト名ごとのインデックス） - elasticsearch、logstash、logstash-grok、logstash-configuration](https://living-sun.com/ja/elasticsearch/244316-logstash-conditional-output-to-elasticsearch-index-per-filebeat-hostname-elasticsearch-logstash-logstash-grok-logstash-configuration.html)
* [i-FilterのアクセスログをLogstashで正規化してセキュリティログ分析に活用する方法](https://qiita.com/hssh2_bin/items/d494ad110f8a73bd7e37)
* [[正規表現] .*?は最短マッチではない](https://qiita.com/anqooqie/items/191ad215e93237c77811)