Filebeatでアプリの独自のログをパースしてElasticsearchに送りたい

ファイルのパーミッションの問題でWSLでFilebeatが使えなかったので、Windows版で。

ログファイル my_app.log。

2020-07-02 08:00:00 成功した
2020-07-02 10:01:12 ダメだった
2020-07-02 20:03:04 こんにちは

↑このログに↓このインデックスをつけたい。

フィールド	値
@timestamp	時間
myapp.success	"成功"のときtrue、"ダメ"のときfalse、それ以外はnull
event.dataset	myapp.log

filebeat.yml

filebeat.inputs:
 - type: log
   enabled: true
   paths:
    - ./my_app.log
   encoding: utf-8
   processors:
    - script:
       lang: javascript
       id: my_log_parser
       source: >
         function process(event) {
           var m = event.Get("message").match(/^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*)/);
           if (m !== null) {
             var success = null;
             if (m[2].match(/成功/)) success = true;
             if (m[2].match(/ダメ/)) success = false;
             event.Put('myapp', { time: m[1], success: success });
             event.Put('event.dataset', 'myapp.result');
           }
         }
    - timestamp:
       field: myapp.time
       layouts:
        - '2006-01-02 15:04:05'
       timezone: '+0900'

output.console:
  enabled: true
  pretty: true

#output.elasticsearch:
#  hosts: ["localhost:9200"]

パースの処理を javascript で書けるのは便利ですな。.jsファイルに分けることもできるので、複雑になってもデバッグしやすそう。

output は console に。Elasticsearch に流す前にコンソールで確認できる。

テスト用に test.bat を作った。ログファイルの読み込み位置が保存されている data.json を毎回削除するようにして、繰り返し実行しても常に新しいファイルとして処理されるように。
(もっとちゃんとしたやり方があるかもしれないけどとりあえず)

> del data\registry\filebeat\data.json
> filebeat -v --path.config %~dp0 --path.data %~dp0data --path.logs %~dp0log

実行結果 (長いので関係ない @metadata, agent, host, log, input,ecs は削除)

{
  "@timestamp": "2020-07-01T23:00:00.000Z",
  "message": "2020-07-02 08:00:00 成功した",
  "myapp": {
    "time": "2020-07-02 08:00:00",
    "success": true
  },
  "event": {
    "dataset": "myapp.result"
  },
}
{
  "@timestamp": "2020-07-02T01:01:12.000Z",
  "message": "2020-07-02 10:01:12 ダメだった",
  "myapp": {
    "time": "2020-07-02 10:01:12",
    "success": false
  },
  "event": {
    "dataset": "myapp.result"
  },
}
{
  "@timestamp": "2020-07-02T11:03:04.000Z",
  "message": "2020-07-02 20:03:04 こんにちは",
  "myapp": {
    "time": "2020-07-02 20:03:04",
    "success": null
  },
  "event": {
    "dataset": "myapp.result"
  },
}

期待通りに @timestamp と myapp.success が設定された。

Kibana でログ確認。

フィールドを success ではなく myapp.success としたのは、既存のフィールドとの衝突を避けるため。
もしも、success というフィールドがすでに存在していて number 型だったりすると、bool は入れられないので reject されてしまう。
↓ここに共通フィールドのスキーマがあるので、目的にあうものがあればそれを使うか、インデックス自体をアプリ専用に作った方がよさげ。

インデックスを分ける方法は、こんな感じでいけるみたい。

output.elasticsearch:
  hosts: ["localhost:9200"]
  indices:
    - index: "myapp-log-%{+yyyy.MM.dd}"
      when.equals:
        event.dataset: 'myapp.result'

まだほとんどまともに使ってないけど、javascriptでパースの処理を書けるのは学習コストが低くて良いですな。