FilebeatとLogstashを使ってrails6のログをElasticsearchへ取り込もうとしたがなかなか難しかったので書く。

rails6のログをそのまま取り込んでもデータとして使えない
- 構造化されていない
- 複数行で意味をなしているものがある
対応設定（具体的な設定ファイルは下記します）
設定ファイルサンプル

rails6のログをそのまま取り込んでもデータとして使えない

構造化されていない

railsの出力ログを見てみたらこんな感じで、とくに構造化されていない。

I, [2020-11-13T17:17:40.347350 #12138]  INFO -- : [a2bb9deb-e759-4768-9b90-c8d7bd4f521c] Started GET "/favicon.ico" for 127.0.0.1 at 2020-11-13 17:17:40 +0900
...

このままElasticsearchへ取り込んでもサーチャブルなデータにならない。

また、ログにはそれが実行された時間が記録されているが、これを認識できないので「Logstashが取り込んだ時間」が基準となってElasticsearchへ記録される。
Logstashが取り込むにはタイムラグがあるので10秒ほど誤差が出る。また、サーバーメンテナンスなどで一時的にログ記録を停止した後再開したりすると、一気にログが流れ込み、全てその時間で記録されることになる。

事実と異なるデータになってしまう。

複数行で意味をなしているものがある

こんなやつ。

I, [2020-11-13T17:17:40.347350 #12138]  INFO -- : [a2bb9deb-e759-4768-9b90-c8d7bd4f521c] Started GET "/favicon.ico" for 127.0.0.1 at 2020-11-13 17:17:40 +0900
F, [2020-11-13T17:17:40.350699 #12138] FATAL -- : [a2bb9deb-e759-4768-9b90-c8d7bd4f521c]   
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] ActionController::RoutingError (No route matches [GET] "/favicon.ico"):
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c]   
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] actionpack (6.0.0) lib/action_dispatch/middleware/debug_exceptions.rb:36:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] actionpack (6.0.0) lib/action_dispatch/middleware/show_exceptions.rb:33:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] railties (6.0.0) lib/rails/rack/logger.rb:38:in `call_app'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] railties (6.0.0) lib/rails/rack/logger.rb:26:in `block in call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] activesupport (6.0.0) lib/active_support/tagged_logging.rb:80:in `block in tagged'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] activesupport (6.0.0) lib/active_support/tagged_logging.rb:28:in `tagged'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] activesupport (6.0.0) lib/active_support/tagged_logging.rb:80:in `tagged'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] railties (6.0.0) lib/rails/rack/logger.rb:26:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] actionpack (6.0.0) lib/action_dispatch/middleware/remote_ip.rb:81:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] actionpack (6.0.0) lib/action_dispatch/middleware/request_id.rb:27:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] rack (2.0.7) lib/rack/method_override.rb:22:in `call'
[a2bb9deb-e759-4768-9b90-c8d7bd4f521c] rack (2.0.7) lib/rack/runtime.rb:22:in `call'

そのまま取り込むと複数行になってしまう。

また、フォーマットも違っていて時間の記載もない。

上記と同じ問題で、そのままとりこむとlogstashへの取り込み時間が基準時間になってしまう。

対応設定（具体的な設定ファイルは下記します）

構造化して取り込みたい

Logstashのfilter設定で構造化する。

grokプラグインを使えば正規表現を使ってログへ意味付けを行える。

※logrageというgemを使えばrailsのログをjson形式で出す事も出来たようだが以下理由で最終的にはやらなかった。

・全てのログがjson形式で出るわけではない（ググってみるとみんな悩んでいるっぽい）
・logrageの仕組みがなんか無理やりやってるぽくってちょっとあとで問題になりそうで怖かった。

ログの時間を基準時間として取り込みたい

上記grokで構造化して取り出した日時データを基準時間として設定する。 Logstashのdateプラグインで出来た。

今回railsログは日本時間で出ていたけど、そのまま取り込むとUTCと判断されるのでtimezone指定して取り込まないといけない。

複数行で意味をなすものは、1行として取り込みたい

FilebeatのMultiline optionsで出来た

railsログとnginxログは別indexとして取り込みたい

今回railsのログとnginxのログをfilebeatで送出している。

$ sudo filebeat modules enable nginx # nginxログ送出設定の有効化

なにもしないとこれら2つがElasticsearchの同じindexに記録されてデータが混ざってしまうので分けたかった。

Logstashの設定でif文を使って分けた。（nginxログもgrok等を使って構造化して取り込む設定も入れている。参考)Use Logstash pipelines for parsing | Logstash Reference [7.10] | Elastic）

複雑化しそうだったらMultiple Pipelinesを使ったほうが良さげ絶対的に使った方がいいLogstashのMultiple Pipelinesについて書いてみた - Qiita

設定ファイルサンプル

こんな感じになった。

（railsサーバー側）
$ sudo vim /etc/filebeat/filebeat.yml

- type: log

  # Change to true to enable this input configuration.
  enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /rails_app_dir/log/production.log


  ### Multiline options

  # Multiline can be used for log messages spanning multiple lines. This is common
  # for Java Stack Traces or C-Line Continuation

  # The regexp Pattern that has to be matched. The example pattern matches all lines starting with [
  multiline.pattern: ^\[

  # Defines if the pattern set under pattern should be negated or not. Default is false.
  multiline.negate: false

  # Match can be set to "after" or "before". It is used to define if lines should be append to a pattern
  # that was (not) matched before or after or as long as a pattern is not matched based on negate.
  # Note: After is the equivalent to previous and before is the equivalent to to next in Logstash
  multiline.match: after

（Logstashサーバー側）
$ sudo vim /etc/logstash/conf.d/filebeat.conf

input {
    beats {
        port => 5044
    }
}

filter {
    # nginx access 参考：https://www.elastic.co/guide/en/logstash/current/logstash-config-for-filebeat-modules.html#parsing-nginx
    if [event][dataset] == "nginx.access" {
        grok {
            match => { "message" => ["%{IPORHOST:[nginx][access][remote_ip]} - %{DATA:[nginx][access][user_name]} \[%{HTTPDATE:[nginx][access][time]}\] \"%{WORD:[nginx][access][met
hod]} %{DATA:[nginx][access][url]} HTTP/%{NUMBER:[nginx][access][http_version]}\" %{NUMBER:[nginx][access][response_code]} %{NUMBER:[nginx][access][body_sent][bytes]} \"%{DATA:[ngi
nx][access][referrer]}\" \"%{DATA:[nginx][access][agent]}\""] }
            remove_field => "message"
        }
        mutate {
            add_field => { "read_timestamp" => "%{@timestamp}" }
        }
        date {
            match => [ "[nginx][access][time]", "dd/MMM/YYYY:H:m:s Z" ]
            remove_field => "[nginx][access][time]"
        }
        useragent {
            source => "[nginx][access][agent]"
            target => "[nginx][access][user_agent]"
            remove_field => "[nginx][access][agent]"
        }
        geoip {
            source => "[nginx][access][remote_ip]"
            target => "[nginx][access][geoip]"
        }
    }

    else if [event][dataset] == "nginx.error" {
        grok {
            match => { "message" => ["%{DATA:[nginx][error][time]} \[%{DATA:[nginx][error][level]}\] %{NUMBER:[nginx][error][pid]}#%{NUMBER:[nginx][error][tid]}: (\*%{NUMBER:[nginx
][error][connection_id]} )?%{GREEDYDATA:[nginx][error][message]}"] }
            remove_field => "message"
        }
        mutate {
            rename => { "@timestamp" => "read_timestamp" }
        }
        date {
            match => [ "[nginx][error][time]", "YYYY/MM/dd H:m:s" ]
            remove_field => "[nginx][error][time]"
        }
    }


    else if [log][file][path] == "/rails_app_dir/log/production.log" {
        grok {
            patterns_dir => ["./patterns"]
            match => { "message" => "(?<timestamp>%{YEAR}-%{MONTHNUM}-%{MONTHDAY}T%{HOUR}:%{MINUTE}:%{SECOND}) #(?<pid>\d+)]\s+(?<level>\w+) -- : \[(?<tid>\w+-\w+-\w+-\w+-\w+)] (?<mess>.+)" }
        }
        mutate {
            rename => { "@timestamp" => "read_timestamp" }
        }
        date {
            match => [ "timestamp", "YYYY-MM-dd'T'HH:mm:ss.SSSSSS" ]
            timezone => "Asia/Tokyo"
            target => "@timestamp"
        }
    }
}

output {
    if [event][module] == "nginx" {
        elasticsearch {
            manage_template => false
            index => "nginx-%{+YYYY.MM.dd}"
        }
    }
    else if [log][file][path] == "/rails_app_dir/log/production.log" {
        elasticsearch {
            index => "rails-%{+YYYY.MM.dd}"
        }
    }
}

やる気がストロングZERO

やる気のストロングスタイル

Filebeat + Logstashでrails6のログをElasticsearchへ取り込む設定