Logstash——multiline 插件，匹配多行日志

时间  2019-11-06
标签 logstash multiline 插件 匹配 日志 栏目 日志分析 繁體版
原文   原文链接

本文内容

  • 测试数据
  • 字段属性
  • 按多行解析运行时日志
  • 把多行日志解析到字段
  • 参考资料

在处理日志时，除了访问日志外，还要处理运行时日志，该日志大都用程序写的，好比 log4j。运行时日志跟访问日志最大的不一样是，运行时日志是多行，也就是说，连续的多行才能表达一个意思。html

本文主要说明，如何用 multiline 出来运行日志。正则表达式

若是能按多行处理，那么把他们拆分到字段就很容易了。ruby

迁移到：http://www.bdata-cap.com/newsinfo/1712113.html

测试数据

[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.
[16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category
where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null
order by product_category.ORDERS asc
[16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.
[16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.
[16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category
where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null
order by product_category.ORDERS desc
[16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category
where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null
order by product_category.ORDERS asc
[16-04-12 03:40:07 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.

测试是在7秒内发生的（固然是假数据）。能够看到，第2、5、六秒的日志是多行的，有条SQL语句。其余是单行的。app

字段属性

 

对 multiline 插件来讲，有三个设置比较重要：negate、pattern 和 what。ide

negate测试

  • 类型是 booleanui

  • 默认为 falsespa

否认正则表达式（若是没有匹配的话）。插件

patterndebug

  • 必须设置

  • 类型为 string

  • 没有默认值

要匹配的正则表达式。

what

  • 必须设置

  • 能够为 previous 或 next

  • 没有默认值

若是正则表达式匹配了，那么该事件是属于下一个或是前一个事件？

按多行解析运行时日志

示例1：若配置文件以下所示，

input {
        file{
                path=>"/usr/local/elk/logstash/logs/c.out"
                type=>"runtimelog"
                codec=> multiline {
                        pattern => "^\["
                        negate => true
                        what => "previous"
                }
                start_position=>"beginning"
                sincedb_path=>"/usr/local/elk/logstash/sincedb-access"
                ignore_older=>0
        }
}
output{
        stdout{
                codec=>rubydebug
        }
}

说明：匹配以“[”开头的行，若是不是，那确定是属于前一行的。

解析结果以下所示，能解析出6个JSON：

{
    "@timestamp" => "2016-06-01T04:37:43.147Z",
       "message" => "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:37:43.152Z",
       "message" => "[16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:37:43.152Z",
       "message" => "[16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:37:43.155Z",
       "message" => "[16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:37:43.157Z",
       "message" => "[16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS desc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:37:43.159Z",
       "message" => "[16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}

解析时，最后一行日志，不会解析。只有当再追加一条日志时，才会解析最后一条日志。

示例2：若将配置文件修改成，

input {
        file{
                path=>"/usr/local/elk/logstash/logs/c.out"
                type=>"runtimelog"
                codec=>multiline  {
                        pattern => "^\["
                        negate => true
                        what => "next"
                }
                start_position=>"beginning"
                sincedb_path=>"/usr/local/elk/logstash/sincedb-access"
                ignore_older=>0
        }
}
output{
        stdout{
                codec=>rubydebug
        }
}

解析结果为，能解析出7个JSON：

{
    "@timestamp" => "2016-06-01T04:40:43.232Z",
       "message" => "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.237Z",
       "message" => "[16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.238Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc\n[16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.239Z",
       "message" => "[16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.244Z",
       "message" => "[16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.245Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS desc\n[16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T04:40:43.249Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc\n[16-04-12 03:40:07 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}

示例3：若将配置文件修改成，

codec=>multiline  {
        pattern => "^\["
        negate => false
        what => "previous"
}

则解析结果为：

{
    "@timestamp" => "2016-06-01T05:38:50.853Z",
       "message" => "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.\n[16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T05:38:50.856Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T05:38:50.858Z",
       "message" => "order by product_category.ORDERS asc\n[16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.\n[16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.\n[16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T05:38:50.860Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T05:38:50.861Z",
       "message" => "order by product_category.ORDERS desc\n[16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}
{
    "@timestamp" => "2016-06-01T05:38:50.863Z",
       "message" => "where product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog"
}

 

 

 

 

 

 

 

把多行日志解析到字段

配置文件以下所示：

input {
        file{
                path=>"/usr/local/elk/logstash/logs/c.out"
                type=>"runtimelog"
                codec=>multiline  {
                        pattern => "^\["
                        negate => true
                        what => "previous"
                }
                start_position=>"beginning"
                sincedb_path=>"/usr/local/elk/logstash/sincedb-access"
                ignore_older=>0
        }
}
filter {
        grok {
                match=>["message","\[%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level}\] %{GREEDYDATA:msg}"]
        }
}
output{
        stdout{
                codec=>rubydebug
        }
}

解析后结果：

{
    "@timestamp" => "2016-06-01T06:33:26.426Z",
       "message" => "[16-04-12 03:40:01 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:01",
         "level" => "DEBUG",
           "msg" => "model.MappingNode:- ['/store/shopclass'] matched over."
}
{
    "@timestamp" => "2016-06-01T06:33:26.485Z",
       "message" => "[16-04-12 03:40:02 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:02",
         "level" => "DEBUG",
           "msg" => "impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc"
}
{
    "@timestamp" => "2016-06-01T06:33:26.491Z",
       "message" => "[16-04-12 03:40:03 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:03",
         "level" => "DEBUG",
           "msg" => "model.MappingNode:- ['/store/shopclass'] matched over."
}
{
    "@timestamp" => "2016-06-01T06:33:26.492Z",
       "message" => "[16-04-12 03:40:04 DEBUG] model.MappingNode:- ['/store/shopclass'] matched over.",
      "@version" => "1",
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:04",
         "level" => "DEBUG",
           "msg" => "model.MappingNode:- ['/store/shopclass'] matched over."
}
{
    "@timestamp" => "2016-06-01T06:33:26.494Z",
       "message" => "[16-04-12 03:40:05 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS desc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:05",
         "level" => "DEBUG",
           "msg" => "impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS desc"
}
{
    "@timestamp" => "2016-06-01T06:33:26.495Z",
       "message" => "[16-04-12 03:40:06 DEBUG] impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc",
      "@version" => "1",
          "tags" => [
        [0] "multiline"
    ],
          "path" => "/usr/local/elk/logstash/logs/c.out",
          "host" => "vcyber",
          "type" => "runtimelog",
     "timestamp" => "16-04-12 03:40:06",
         "level" => "DEBUG",
           "msg" => "impl.JdbcEntityInserter:- from product_category product_category\nwhere product_category.PARENT_ID is null and product_category.STATUS = ? and product_category.DEALER_ID is null\norder by product_category.ORDERS asc"
}

参考资料

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程