logstash Apache 日志的自定义日志过滤器

Question

我是 ELK 堆栈的新手。我有一个filebeat服务将日志发送到logstash，并且在logstash中使用grok过滤器，数据被推送到elasticsearch索引。

我正在使用gork 过滤器和match => { "message" => "%{COMBINEDAPACHELOG}"} 来解析数据。

我的问题是，我希望将字段名称及其值存储在弹性搜索索引中。我的不同版本的日志如下：

---

27.60.18.21 - - [27/Aug/2017:10:28:49 +0530] "GET /api/v1.2/places/search/json?username=pradeep.pgu&location=28.5359586,77.3677936&query=atm&explain=true&bridge=true HTTP/1.1" 200 3284
27.60.18.21 - - [27/Aug/2017:10:28:49 +0530] "GET /api/v1.2/places/search/json?username=pradeep.pgu&location=28.5359586,77.3677936&query=atms&explain=true&bridge=true HTTP/1.1" 200 1452
27.60.18.21 - - [27/Aug/2017:10:28:52 +0530] "GET /api/v1.2/places/nearby/json?&refLocation=28.5359586,77.3677936&keyword=FINATM HTTP/1.1" 200 3283
27.60.18.21 - - [27/Aug/2017:10:29:06 +0530] "GET /api/v1.2/places/search/json?username=pradeep.pgu&location=28.5359586,77.3677936&query=co&explain=true&bridge=true HTTP/1.1" 200 3415
27.60.18.21 - - [27/Aug/2017:10:29:06 +0530] "GET /api/v1.2/places/search/json?username=pradeep.pgu&location=28.5359586,77.3677936&query=cof&explain=true&bridge HTTP/1.1" 200 2476

---

弹性索引中我想要的字段如下：

1. client_ip => 类型必须与 kibana 用于 IP 映射的兼容。
2. 时间戳 => 日期时间格式。 => 日志的时间
3. method => text => 被调用的方法，例如获取、发布
4. 版本 => 十进制数 => 例如1.2 / 1.0（在示例日志中为 v1.2）
5. username => text => username= 之后的文本（在示例日志中为 pradeep.pgu）
6. location =>geo_point type => 该值同时具有纬度和经度，因此 kibana 可以在地图上绘制它们。
7. search_query => text => 被搜索的内容（在示例中来自“keyword=”或“query=”这两个字段之一）。这两个字段中的任何一个都将存在，并且存在的一个必须使用它的值。
8. response_code => number => 响应的代码。 （在样本中为 200）
9. data_transfered => number => 传输的数据量（样本中的最后一个数字）。

这样的事情可能吗？ gork 过滤器是否对此有规定？问题是参数不是特定于顺序的。

Answer 1

从HTTPD_COMMONLOG 开始，您可以使用此模式（您可以在grok tester 进行测试）：

grok {
 match => { 
  "message" => "%{IPORHOST:client_ip} %{HTTPDUSER:ident} %{HTTPDUSER:auth} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:method} /api/v%{NUMBER:version}/places/search/json\?%{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})\" %{NUMBER:response_code} (?:%{NUMBER:data_transfered}|-)"
 } 
}

一旦 grok 过滤器提取了请求，您就可以在其上使用 kv 过滤器，它将提取参数（并忽略参数不是特定于顺序的问题）。您必须将 field_split 选项放在 &:

kv { 
  source => "request"
  field_split => "&"
}

对于search_query，根据存在的字段，我们使用mutate 过滤器和add_field 选项来创建字段。

---

filter {
    grok {
        match => { 
            "message" => "%{IPORHOST:client_ip} %{HTTPDUSER:ident} %{HTTPDUSER:auth} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:method} /api/v%{NUMBER:version}/.*/json\?%{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})\" %{NUMBER:response_code} (?:%{NUMBER:data_transfered}|-)"
        } 
    }
    kv { 
        source => "request"
        field_split => "&"
    }

    if [query] {
        mutate {
            add_field => { "search_query" => "%{query}" }
        }
    } else if [keyword] {
        mutate {
            add_field => { "search_query" => "%{keyword}" }
        }
    }

    if [refLocation] {
        mutate {
            rename => { "refLocation" => "location" }
        }
    }
}