Logstashとは？

サーバーサイドのデータ処理パイプラインです。
多種多様なデータ入力ソースに対して、形式や複雑さに関わらずデータを動的に変換し
それぞれの出力先へ格納する仕組みです。

Logstashに様々なプラグインがあることにより、どのようなタイプのログにも対応することが可能。

以下は、一例ですが、FSやBeats,System Logなどの情報をLogstashで整形し
S3やElasticsearchなどに出力することが可能です。

Configファイルを書く

今回は、FSに格納されているCSVデータをElasticsearchに投入してみようと思います。
任意のフォルダにてlogstash-sample.confを以下のように作成します。


input {
      file{
      mode => "tail"
      path => ["/Users/hogehoge/fugafuga/elastic_test/logstash/sample.csv"]
      start_position => "beginning"
      sincedb_path => "/dev/null"
      }
}

filter {
    csv {
        columns => ["Date", "Level", "ErrorMessage","UserId"]
        convert => {
            "UserId" => "integer"
        }
        skip_header => true
    }
    date {
        match => ["Date", "yyyy-MM-dd HH:mm:ss"]
        target => "Date"
    }
}

output {
  elasticsearch {
  hosts => ['127.0.0.1:9200']
  index => 'log'
}
  
}

　各種説明

input

input {
      file{
      mode => "tail"
      path => ["/Users/hogehoge/fugafuga/elastic_test/logstash/sample.csv"]
      start_position => "beginning"
      sincedb_path => "/dev/null"
      }
}

inputにて、まず始めにfileが選択されています。
これは、入力元をFSと指定しています。
fileの中を見ると複数の設定があります。以下が簡単な説明です。

mode
- fileの読み込みモードを指定します。
  - tail
    - ファイルの追記を監視し、追記することが出来るモード
    - イメージとしては、tail -f コマンド
  - read
    - EOFを見つけると継続して読み込むことをしない。
path
- 入力元のファイルが存在する場所を指定する。
  - 完全パスでないといけない。
  - パターンで設定することも可能
    - /Users/hogehoge/fugafuga/elastic_test/logstash/*.csv
sincedb_path
- どこまで任意のファイルを読み込んだかを保存する場所
  - 保存しない場合 /dev/nullを指定すれば良い(Linuxの場合)
start_position
- 任意のファイルを初めて読み込んだ際の読み込み開始位置を指定する。
  - デフォルトではend
  - beginning
    - ファイルの先頭から読み込む
  - end
    - ファイルの最終行から読み込む

filter

filter {
    csv {
        columns => ["Date", "Level", "ErrorMessage","UserId"]
        convert => {
            "UserId" => "integer"
        }
        skip_header => true
    }
    date {
        match => ["Date", "yyyy-MM-dd HH:mm:ss"]
        target => "Date"
    }
}

Inputで入力されたデータを整形する部分
filterにて、まずcsvが設定されている。
これは、csvデータを加工することを示している。

colums
- CSVデータのカラム情報を記述
convert
- csvの各カラム毎にデータ変換を行う処理が書かれている
skip_header
- csvのheader情報を飛ばすか否かを設定する。
date
- 入力データの日付情報を任意のDateフォーマットへ変換している。
- targetでどこに格納するかを設定。
- dateで時間と定義することでESに入れた際にtimestampとして利用可能

Output

output {
  elasticsearch {
  hosts => ['127.0.0.1:9200']
  index => 'log'
}
  
}

outputにて、elasticseachが設定されている。
出力先がelasticsearchであることを示している。

hosts
- 出力先のelasticsearchのhost情報を設定。
  - 今回は、localhost (127.0.0.1:9200)を指定
idnex
- データ投入先のindexを指定している。
  - 今回はlogを指定。
    - elasticsearch上に存在しないindexを指定した場合、自動でindexが作成される。

実際に動かしてみよう

ターミナルで指定のフォルダにて以下のコマンドを実行するだけ。
(今回はconfファイルを作成したので -f オプションでファイル指定しています)

logstash -f sample_config.conf

確認方法

Postman(ブラウザでもOK)できちんとindexが作成されているかを確認しましょう。
GET http://localhost:9200/_cat/indices
上記のリクエストでelasticsearchのindex一覧が取得可能です。
下記のように今回作成したlog indexが作成されていれば成功です。

...
yellow open log                          afaZBr-dQ2WUeAWNBYj4aw 5 1      3 0  25.7kb  25.7kb
...