Logstashの複数行に渡るログの処理方法について

前回に続いてlogastash+datadogの話になります。
今回はlogstashを使う上での複数行に渡るログの処理方法についてです。
例えば、下記のようなjavaのスタックトレースが出力されているとします。

2018-10-29 06:29:58.690 hostname ERROR Log4j2DemoApplication: - This is Error Message.
    java.lang.Exception: null
    at com.example.log4j2demo.Log4j2DemoApplication.run(Log4j2DemoApplication.java:29) ~[classes/:?]
    at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:781) ~[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE]
    at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:771) ~[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:335) ~[spring-boot-2.0.0.RELEASE.jar:2.0.0.RELEASE]
    at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:138) ~[spring-boot-test-2.0.0.RELEASE.jar:2.0.0.RELEASE]' >> /var/tmp/test.log

これをそのままDatadogに送信すると下記のように別のレコードとして登録されてしまいます。
※LogstashでなくDatadogエージェントで直接送る場合はJSONの形で送っているため問題はありません。

スタックトレースも含めて１レコードとして扱うためにはマルチラインの設定を記述しておきます。
マルチラインはinputの記述の中でcodecの項目で指定をします。
下記の設定では、先頭に日付が存在しない行は直前の日付が存在する行の末尾に追記されます。

input {
  file {
    path => "/var/tmp/test.log"
    start_position => "beginning"
    codec => multiline {
      pattern => "%{TIMESTAMP_ISO8601}"
      negate => true
      what => "previous"
    }
  }
}

この状態でログを出力すると下記のようにjavaのスタックトレースが１レコードとしてまとめって管理がされます。

普通にログを閲覧するだけであればここまでする必要はないですが、Datadogでエラーログを監視して通知したいとき等にはまとめておくと情報量を増やすことができます。