在 Kibana 仪表板中创建单独部分的 Grok 模式

Question

长期以来，我一直在尝试使用 logstash 从我的自定义日志中提取和标记数据，但没有得到任何结果，我有一个自定义的 haproxy 日志，如下所示：

Feb 22 21:17:32 ap haproxy[1235]: 10.172.80.45:32071 10.31.33.34:44541 10.31.33.34:32772 13.127.229.72:443 [22/Feb/2020:21:17:32.006] this_machine~ backend_test-tui/test-tui_32772 40/0/5/1/836 200 701381 - - ---- 0/0/0/0/0 0/0 {testtui.net} {cache_hit} "GET /ob/720/output00007.ts HTTP/1.1"

我想从日志中提取并标记kibana仪表板中的特定内容，例如：

• 从“40/0/5/1/836”部分我想将唯一的最后一个部分数字 (836) 标记为“response_time”
• “701381”为“response_bytes”
• “/ob/720/output00007.ts”为“content_url”
• 并且想使用日志文件中的时间戳而不是默认的

我使用https://grokdebug.herokuapp.com/ 创建了一个 grok 过滤器，但每当我应用它时，我都会看到“_grokparsefailure”消息，并且 kibana 仪表板停止填充

下面是logstash调试日志

{
      "@version" => "1",
       "message" => "Mar  8 13:53:59 ap haproxy[22158]: 10.172.80.45:30835 10.31.33.34:57886 10.31.33.34:32771 43.252.91.147:443 [08/Mar/2020:13:53:59.827] this_machine~ backend_noida/noida_32771 55/0/1/0/145 200 2146931 - - ---- 0/0/0/0/0 0/0 {testalef1.adcontentamtsolutions.} {cache_hit} \"GET /felaapp/virtual_videos/og/1080/output00006.ts HTTP/1.1\"",
    "@timestamp" => 2020-03-08T10:24:07.348Z,
          "path" => "/home/alef/haproxy.log",
          "host" => "com1",
          "tags" => [
        [0] "_grokparsefailure"
    ]
}

下面是我创建的过滤器

%{MONTH:[Month]} %{MONTHDAY:[date]} %{TIME:[time]} %{WORD:[source]} %{WORD:[app]}\[%{DATA:[class]}\]: %{IPORHOST:[UE_IP]}:%{NUMBER:[UE_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Source_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Destination_Port]} %{IPORHOST:[WAN_IP]}:%{NUMBER:[WAN_Port]} \[%{HAPROXYDATE:[accept_date]}\] %{NOTSPACE:[frontend_name]}~ %{NOTSPACE:[backend_name]} %{NOTSPACE:[ty_name]}/%{NUMBER:[response_time]} %{NUMBER:[http_status_code]} %{INT:[response_bytes]} - - ---- %{NOTSPACE:[df]} %{NOTSPACE:[df]} %{DATA:[domain_name]} %{DATA:[cache_status]} %{DATA:[domain_name]} %{NOTSPACE:[content]} HTTP/%{NUMBER:[http_version]}

下面是我的logstash conf文件：

    input {
  beats {
    port => 5044
  }
 }

filter {
  grok {
    match => { "message" => "%{MONTH:[Month]} %{MONTHDAY:[date]} %{TIME:[time]} %{WORD:[source]} %{WORD:[app]}\[%{DATA:[class]}\]: %{IPORHOST:[UE_IP]}:%{NUMBER:[UE_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Source_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Destination_Port]} %{IPORHOST:[WAN_IP]}:%{NUMBER:[WAN_Port]} \[%{HAPROXYDATE:[accept_date]}\] %{NOTSPACE:[frontend_name]}~ %{NOTSPACE:[backend_name]} %{NOTSPACE:[ty_name]}/%{NUMBER:[response_time]} %{NUMBER:[http_status_code]} %{INT:[response_bytes]} - - ---- %{NOTSPACE:[df]} %{NOTSPACE:[df]} %{DATA:[domain_name]} %{DATA:[cache_status]} %{DATA:[domain_name]} %{NOTSPACE:[content]} HTTP/%{NUMBER:[http_version]} " }
  }
  date {
    match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
 }

output {
  elasticsearch { hosts => ["localhost:9200"] }
  }

Answer 1

使用以下过滤器解决了我的问题，我必须在 logstash 本身中进行调试才能获得正确的过滤器：

输入{节拍{ 端口 => 5044 } }

过滤器{ grok { match => { "message" => "%{MONTH:month} %{MONTHDAY:date} %{TIME:time} %{WORD:[source]} %{WORD:[app]}[%{DATA:[班级]}]： %{IPORHOST:[UE_IP]}:%{NUMBER:[UE_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Source_Port]} %{IPORHOST:[NATTED_IP]}:%{NUMBER:[NATTED_Destination_Port]} %{IPORHOST:[WAN_IP]}:%{NUMBER:[WAN_Port]} [%{HAPROXYDATE:[accept_date]}] %{NOTSPACE:[frontend_name]}~ %{NOTSPACE:[后端名称]} %{NOTSPACE:[ty_name]}/%{NUMBER:[response_time]:int} %{NUMBER:[http_status_code]} %{NUMBER:[response_bytes]:int} - - ---- %{NOTSPACE:[df]} %{NOTSPACE:[df]} %{DATA:[domain_name]} %{DATA:[cache_status]} %{DATA:[domain_name]} %{URIPATHPARAM:[content]} HTTP/%{NUMBER:[http_version]}" } add_tag => [ "response_time", "response_time" ]

}日期{ 匹配 => [ "时间戳" , "dd/MMM/yyyy:HH:mm:ss Z" ] } }

输出 { elasticsearch { hosts => ["localhost:9200"] }

标准输出 {

编解码器 => ruby​​debug

} }