はじめに
Kubernetesでのログ収集にfluentを使うのは結構メジャーな方法だと思います。
実際にやってみて分かったのですが、kubernetes上で動いているアプリケーションが複数業のログを出した場合、
kubernetes上ではログが複数に分割されてしまいます。
Fluentdではログの固定フォーマットに従って、ログ内容をパースできるのがメリットの一つだと思うのですが、
このままだとパースができなくなってしまいます。
具体的にはこんな感じ。
2020-10-04T14:07:22.234Z ERR Authorization error because you have to be a member of following groups
XXXX
YYYY
ZZZZ
これがこんな感じに分割されてしまいます。
{"log":"2020-10-04T14:07:22.234Z ERR Authorization error because you have to be a member of following groups\n","stream":"stderr","time":"2020-10-04T14:07:22.234Z"}
{"log":"XXXX\n","stream":"stderr","time":"2020-10-04T14:07:22.434Z"}
{"log":"YYYY\n","stream":"stderr","time":"2020-10-04T14:07:22.634Z"}
{"log":"ZZZZ\n","stream":"stderr","time":"2020-10-04T14:07:22.639Z"}
最終的にログをDBなり、Elasticsearchなりに集約すると思いますが、これだと可読性が下がってしまいますよね。。。
色々調べてみましたが、ログの結合とパースを（私のやりたいように）両方やっている事例はなかったので、試行錯誤しなんとかして記事にしてみました。
どなたかのお役に立つと嬉しいです。
ちなみに、fluent内ではrubyによるコードを使うことができるのですが、
筆者はRubyの知識がありません。
最後のパース周りでRubyを使っているのですが行き当たりばったりベースでやっているので、
Rubyのエキスパートには改善点が見つかるかもしれません。
その際には是非、優しく有意義なご意見をいただけると幸いです。
環境
kubectl: v1.16.13
k8s server: AWS EKS
fluentd: 1.11.3
Fluend config
早速解決策です。
まず、使用するpluginです。
fluent-plugin-concat (2.4.0)
fluent-plugin-record-reformer (0.9.1)
調べる時間もなかったので、既存のイメージを編集して使いました。
FROM fluent/fluentd-kubernetes-daemonset:v1.11.3-debian-kinesis-1.0
RUN fluent-gem install fluent-plugin-record-reformer
ここで作ったイメージをAWS ECRにアップロードし、Daemonsetとして使用しています。
このイメージはkubernetes上のログもデフォルトで収集対象としていますが、
上記の分割されてしまったログの結合対応はされておらず、1行ずつ出力されるようになっています。
これを変更します。
もともとのロジック
メインのコンフィグファイルは/fluentd/etc/fluent.confです。
@include "#{ENV['FLUENTD_SYSTEMD_CONF'] || 'systemd'}.conf"
@include "#{ENV['FLUENTD_PROMETHEUS_CONF'] || 'prometheus'}.conf"
@include kubernetes.conf
Kubernetes上のアプリログは、/var/log/containers/*.logに格納されます。
kubernets.confが上記に対応するコンフィグなので、こちらを上書きます。
元々のファイルに以下を追記してます。
<filter kubernetes.**>
@type concat
key log
multiline_start_regexp /^\d{4}-\d{1,2}-\d{1,2}.*/
separator ''
flush_interval 5
timeout_label @NORMAL
</filter>
<match **>
@type relabel
@label @NORMAL
</match>
<label @NORMAL>
<match kubernetes.**>
@type record_reformer
tag ref.kubernetes
enable_ruby true
time ${begin log.scan(/^(\d{4}-\d{1,2}-\d{1,2}T\d{1,2}:\d{1,2}:\d{1,2}.\d{1,3}Z?)/).first.first rescue "" end}
level ${begin log.scan(/^\d{4}-\d{1,2}-\d{1,2}T\d{1,2}:\d{1,2}:\d{1,2}.\d{1,3}Z (\w{3}?)/).first.first rescue "" end}
message ${begin log.gsub(/\n/, 'NNN').scan(/^\d{4}-\d{1,2}-\d{1,2}T\d{1,2}:\d{1,2}:\d{1,2}.\d{1,3}Z \w{3} (.*?)$/).first.first.gsub(/NNN/, '\n') rescue log end}
</match>
<filter ref.kubernetes>
@type record_transformer
enable_ruby
<record>
container ${record.dig("kubernetes", "container_name")}
</record>
</filter>
<match **>
@type kinesis_streams
@id out_kinesis_streams
region "ap-northeast-1"
stream_name "xxxxx"
include_time_key "xxxxx"
<buffer>
@type file
path /var/log
flush_at_shutdown true
flush_interval 1
chunk_limit_size "1m"
flush_thread_interval 0.1
flush_thread_burst_interval 0.01
flush_thread_count 15
</buffer>
</store>
</match>
</label>
以下、解説です。
1. @type concat
GithubのREADMEにある、kubernetesでの活用事例ほぼそのままです。
アプリケーションログに規則性（改行されたとしても、必ず日付がログの一番最初に来る）を利用し、multiline_start_regexpを使って、multilineの終了を判別しています。
また、例外によるエラーはこのフォーマットにのらないことが多いので、timeoutとtimeout_labelを設定し、このregexpに当てはまらないものも後続に流れるようにしています。
キーに設定してあるlogフィールドがコンカチされて、次のプロセスに流れます。
2. @type record_reformer
前プロセスでラベルがつけられているので、<label @NORMAL>内で処理を記載します。
ここでは、パースするための処理を記載しています。
Rubyのscanというメソッドを使い、regexpで値を抽出しています。
また、前プロセスのmultilineのregexpに合致しないものはそもそもこのscanの対象にならず、
scan(xxx)が値を持たないため、scan(xxx).first.firstが例外を発生させてします。
そこで、Rubyの例外対応である、begin rescue endをこの中に入れています。
最後、ログ内のメイン部分は改行されていたものを結合しているため、もちろん改行を含みます。
ここで、通常のregexpとscanで色々と試してみたのですが、どうしても改行部分で決裂してしまったり、
改行を無視しようとすると、終わりが判別できなくなったりうまく行きませんでした。
若干汚いですが、gsubを使い一旦改行をエスケープし、最後に改行を戻す（今回の例ではkinesisに改行として送りたいため。スペースでいい、などであればスペースに変更するだけで良い）ようなことをしています。
終わりに
以上です。
同じことをやろうとしている方は多いのかと思いつつ、日本語、英語共にダイレクトに当てはまるものがなかったので、
色々調べて試行錯誤した内容を記事にしてみました。
質問・改善点など、お待ちしております。
参考
concatenateについて：fluent-plugin-concat
concatenateしたあとのパースに関するヒント：Parse Ruby on Rails logs with FluentD