Home 数据分析 [转] 使用Logstash filter grok过滤日志文件

[转] 使用Logstash filter grok过滤日志文件

by zinan

Logstash Filter Plugin Grok

Logstash提供了一系列filter过滤plugin来处理收集到的log event,根据log event的特征去切分所需要的字段,方便kibana做visualize和dashboard的data analysis。所有logstash支持的event切分插件查看这里。下面我们主要讲grok切分。

Grok基本介绍

  1. Grok 使用文本片段切分的方式来切分日志事件,语法如下:

    %{SYNTAX:SEMANTIC}
    • SYNTAX代表匹配值的类型,例如,0.11可以NUMBER类型所匹配,10.222.22.25可以使用IP匹配。
    • SEMANTIC表示存储该值的一个变量声明,它会存储在elasticsearch当中方便kibana做字段搜索和统计,你可以将一个IP定义为客户端IP地址client_ip_address,eg:%{IP:client_ip_address},所匹配到的值就会存储到client_ip_address这个字段里边,类似数据库的列名,也可以把event log中的数字当成数字类型存储在一个指定的变量当中,比如响应时间http_response_time,假设event log record如下:

      55.3.244.1 GET /index.html 15824 0.043
    • 可以使用如下grok pattern来匹配这种记录

      %{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}

      在logstash conf.d文件夹下面创建filter conf文件,内容如下

      # /etc/logstash/conf.d/01-filter.conf filter {
      grok { match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
      }
      }

      以下是filter结果

      client_id_address: 55.3.244.1 method: GET
      request: /index.html
      bytes: 15824 http_response_time: 0.043

      grok内置的默认类型有很多种,查看所有默认类型

  2. 使用自定义类型

    更多时候logstash grok没办法提供你所需要的匹配类型,这个时候我们可以使用自定义

    • 第一种,直接使用oniguruma语法去匹配文本片段,语法如下

      (?<field_name>the pattern here)
    • 假设你需要匹配的文本片段为一个长度为10或11的十六进制的值,使用下列语法可以获取该片段,并把值赋予queue_id

      (?<queue_id>[0-9A-F]{10,11})
    • 第二种,创建自定义pattern文件

      • 创建文件夹patterns,在此文件夹下面创建一个文件,文件名随意,eg: postfix

        # contents of ./patterns/postfix:
        POSTFIX_QUEUEID [0-9A-F]{10,11}
      • 然后将patterns file引入,告诉logstash你的自定义类型文件,以下面的event log record为例子:

        55.3.244.1 GET /index.html 15824 0.043 ABC24C98567
      • 在logstash conf.d文件夹下面创建filter conf文件,内容如下

        filter {
        grok { patterns_dir => ["./patterns"] match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time} %{POSTFIX_QUEUEID:queue_id}" }
        }
        }
      • 匹配结果如下:

        client_id_address: 55.3.244.1 method: GET
        request: /index.html
        bytes: 15824 http_response_time: 0.043 queue_id: ABC24C98567
      • 推荐使用grokdebugger来写匹配模式,输入event log record,再逐步使用pattern微调切分,下方会根据你所写的模式将输入切分字段。




  3. 其他常用内置方法

    • add_field: 当pattern匹配切分成功之后,可以动态的对某些字段进行特定的修改或者添加新的字段,使用%{fieldName}来获取字段的值

      Exmaple:

      filter {
      grok{ add_field => { "foo_%{somefield}" => "Hello world, %{somefield}" }
      }
      }
      # You can also add multiple fields at once:
      filter {
      grok {
      add_field => { "foo_%{somefield}" => "Hello world, %{somefield}" "new_field" => "new_static_value" }
      }
      }
    • 如果somefield=dad,logstash会将foo_dad新字段加入elasticsearch,并将值Hello world, dad赋予该字段

    • add_tag: 为经过filter或者匹配成功的event添加标签

      Example:

      filter {
      grok { add_tag => [ "foo_%{somefield}" ]
      }
      }
      # You can also add multiple tags at once: filter {
      grok { add_tag => [ "foo_%{somefield}", "taggedy_tag"]
      }
      }
    • 更多内置方法在这里

——END——

原文地址:http://www.jianshu.com/p/d46b911fb83e

作者 : Eason,专注各种技术、平台、集成,不满现状,喜欢改改改

文章、技术合作,大胆的扫一扫,害羞的请邮件


Email : eason.lau02@hotmail.com

打赏
0 comment

You may also like

Leave a Comment

*

code