ELK logstash 处理多行事件

实际效果如下：

       按行显示：

       

 

 

       多源文件合并显示：

       

 

有些日志是分多行输出的，为了能正确的处理这些多行的事件，logstash必须哪些行是单一事件的一部分。多行事件处理是比较复杂的，需依赖正确的事件顺序。最佳方法是尽早的在管道中处理以保证日志有序。这就是本文要说到的logstash管道multiline codec，专门来处理合并多行事件的。

multiline插件最重要的方面：

•  pattern 选项指定一个正则表达式。 事件匹配指定的正则表达式来确定是前一个事件的内容还是新的事件的内容。可以使用grok正则表达式的模板来配置该选项。
•  what 选项有两个选择值：previous 或者 next。previous 值指定行匹配pattern选项的内容是上一行的一部分。 next 指定行匹配pattern选项的内容是下一行的一部分。* negate选项适用于 multiline codec 行不匹配pattern选项指定的正则表达式。

了解更多还需要看看multiline codec 和 multiline filter 插件的配置项。

下面介绍三个例子：

• java 堆栈跟踪单个事件
• C-style 线连续成单个事件
• 从时间标记事件

java stack traces

Java stack traces  包含多行内容，后面几行与初始行缩进，如下所示：

 

1 2 3 4

Exception in thread "main" java.lang.NullPointerException         at com.example.myproject.Book.getTitle(Book.java:16)         at com.example.myproject.Author.getBookTitles(Author.java:25)         at com.example.myproject.Bootstrap.main(Bootstrap.java:14)

logstash配置如下：

 

1 2 3 4 5 6 7 8

input {   stdin {     codec => multiline {       pattern => "^\s"       what => "previous"     }   } }

该配置将任何以空白开始的行与上一行合并。

输出的结果如下：

 

1 2 3 4 5 6 7 8 9

{     "@timestamp" => "2015-12-11T07:32:20.872Z",        "message" => "Exception in thread \"main\" java.lang.NullPointerException\n        at com.example.myproject.Book.getTitle(Book.java:16)\n        at com.example.myproject.Author.getBookTitles(Author.java:25)\n        at com.example.myproject.Bootstrap.main(Bootstrap.java:14)Settings: Default filter workers: 4",       "@version" => "1",           "tags" => [         [0] "multiline"     ],           "host" => "localhost" }

 

line continuations

多种编程语言使用\字符在一行末尾表示该行继续，如：

 

1 2	printf ("%10.10ld  \t %10.10ld \t %s\   %f", w, x, y, z );

logstash配置如下：

 

1 2 3 4 5 6 7 8

input {   stdin {     codec => multiline {       pattern => "\\$"       what => "next"     }   } }

该配置将以\字符结尾的任何行合并到下一行。

输出的结果如下：

 

1 2 3 4 5 6 7 8 9

{     "@timestamp" => "2015-12-11T07:46:02.116Z",        "message" => "printf (\"%10.10ld  \\t %10.10ld \\t %s\\\n  %f\", w, x, y, z );",       "@version" => "1",           "tags" => [         [0] "multiline"     ],           "host" => "localhost" }

 

timestamps

有些日志是以时间戳开始的，如：

 

1 2	[2015-08-24 11:49:14,389][INFO ][env                      ] [Letha] using [1] data paths, mounts [[/ (/dev/disk1)]], net usable_space [34.5gb], net total_space [118.9gb], types [hfs]

logstash配置如下：

 

1 2 3 4 5 6 7 8 9

input { stdin {     codec => multiline {       pattern => "^%{TIMESTAMP_ISO8601} "       negate => true       what => previous     }   } }

该配置使用negate => true 选项来指定任何不是以时间戳开始的行属于前行。也就是不匹配pattern的行都属于前行的内容的一部分。

输出的结果如下：

 

1 2 3 4 5 6 7 8 9

{     "@timestamp" => "2015-12-11T08:08:37.013Z",        "message" => "[2015-08-24 11:49:14,389][INFO ][env                      ] [Letha] using [1] data paths, mounts [[/\n(/dev/disk1)]], net usable_space [34.5gb], net total_space [118.9gb], types [hfs]",       "@version" => "1",           "tags" => [         [0] "multiline"     ],           "host" => "localhost" }

后面看应用处理mysql慢查询日志。