1. 程式人生 > >使用Logstash filter grok過濾日誌文件

使用Logstash filter grok過濾日誌文件

ref mef path nbsp .com pack 十六 elastic alt

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內置的默認類型有很多種,查看所有默認類型。

  1. 使用自定義類型
    更多時候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微調切分,下方會根據你所寫的模式將輸入切分字段。

技術分享圖片
技術分享圖片

  1. 其他常用內置方法
    • 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"]
  }
}

更多內置方法在這裏



作者:eason02
鏈接:https://www.jianshu.com/p/d46b911fb83e
來源:簡書
著作權歸作者所有。商業轉載請聯系作者獲得授權,非商業轉載請註明出處。

使用Logstash filter grok過濾日誌文件