filebeat で ISO8601 形式のタイムスタンプを @timestamp に変換して elasticsearch にインデキシキングする

この記事は、みらい翻訳 Advent Calendar 2020 3日目の記事です。

こんにちは、株式会社みらい翻訳 プラットフォーム部の@kobarasukimaroです。
バックエンドエンジニアですが最近は SRE をやっています。
ちょうどアドベントカレンダーに空きがあったのでねじ込んでみました。

記事について

filebeat を使ってログを読み込むときに logback のデフォルトタイムスタンプ ISO8601 形式のタイムスタンプのパースにハマったのでメモ

環境

Filebeat 7.8.0
Elasticsearch 7.8.0

内容

filebeat で Logstash と同じ感覚で ISO8601 形式のタイムスタンプをそのままパースできると思ったら出来ませんでした。
Logstash とかとフォーマットが違うよというのはドキュメントに書いてありました。

filebeat.yml 内で色々と試行錯誤したのでその結果を残しておきます。

先に結論

以下のように定義を書いたら上手くパースできました

filebeat.inputs:
- type: s3
  processors:
    - dissect:
        tokenizer: '%{timestamp} %{message}'
        field: "message"
        target_prefix: "dissect"
    - script:
        lang: javascript
        id: convert_timestamp
        source: >
          function process(event) {
              var timestamp = event.Get("dissect.timestamp");
              event.Put("dissect.new_timestamp", Date.parse(timestamp));
          }
    - timestamp:
        field: dissect.new_timestamp
        layouts:
          - 'UNIX_MS'

yaml について

dissect

filebeat が受け取ったデータをトークナイズします。ここは受け取ったものによって内容は変わってきますが、今回だと↓のようなログを想定しています。

2020-12-03T05:35:32.010+0000 hogehoge

script

filebeat でデータを加工する時 ECMAScript が使えます。
https://www.elastic.co/guide/en/beats/filebeat/master/processor-script.html

ISO8601 の日時だとそのままパースできないため、ここで UNIX タイムにパースします。
さらにそのあと elastic search の timestamp に変換するため event.Put を使ってイベントに値を入れます。

timestamp

最後に elastic search でタイムスタンプとして取り扱えるように変換します。
new_timestamp は UNIX タイムが入っているので layouts で UNIX_MS を指定します。
そうすると elastic search の @timestamp に dissect で 取り出した timestamp と同じ日時のデータを入れることができます（別のフィールドに入れたければ target_field でフィールド名指定）
https://www.elastic.co/guide/en/beats/filebeat/master/processor-timestamp.html

最後に

filebeat 色々できるので楽しいですね。
他、良いパースの方法ありましたら教えていただけると幸いです（logback のタイムスタンプ直した方が早いというのもありますね・・・）。

明日は @m-chika が何か書くそうです！