我真的很喜欢ELK解析日志。但是,我陷入了需要解析字典列表的问题。以下是我的日志:-
IP - - 0.000 0.000 [24/May/2015:06:51:13 +0000] *"POST /c.gif HTTP/1.1"* 200 4 * user_id=UserID&package_name=SomePackageName&model=Titanium+S202&country_code=in&android_id=AndroidID&eT=1432450271859&eTz=GMT%2B05%3A30&events=%5B%7B%22eV%22%3A%22com.olx.southasia%22%2C%22eC%22%3A%22appUpdate%22%2C%22eA%22%3A%22app_activated%22%2C%22eTz%22%3A%22GMT%2B05%3A30%22%2C%22eT%22%3A%221432386324909%22%2C%22eL%22%3A%22packageName%22%7D%5D * "-" "-" "-"
上述日志的URL解码版本为
IP - - 0.000 0.000 [24/May/2015:06:51:13 0000] *"POST /c.gif HTTP/1.1"* 200 4 * user_id=UserID&package_name=SomePackageName&model=Titanium S202&country_code=in&android_id=AndroidID&eT=1432450271859&eTz=GMT+05:30&events=[{"eV":"com.olx.southasia","eC":"appUpdate","eA":"app_activated","eTz":"GMT+05:30","eT":"1432386324909","eL":"packageName"}] * "-" "-" "-"
无论我在哪里解析,它都会显示
_jsonparsefailure
。我也经历过this问题,也经历过各种论坛,但没有找到完美的解决方案。如何解析Logstash中的JSON列表?如果到现在为止都不存在,那么可以解决该问题。以下是我的配置文件。
filter {
mutate {
gsub => [
"message", "\+", "%20"
]
}
urldecode{
field => "message"
}
grok {
match => [
'message', '%{IP:clientip}%{GREEDYDATA} \[%{GREEDYDATA:timestamp}\] \*"%{WORD:method}%{GREEDYDATA}'
]
}
kv {
field_split => "&?"
}
json{
source => "events"
}
geoip {
source => "clientip"
}
}
最佳答案
这个问题是Parse json in a list in logstash的精确副本。即使有相同的日志条目?有人能理解吗?
您可以在那里看到我的答案,但我会为您总结一下... 选项e)可能是的最佳方法
显然,由于方括号,您得到jsonparsefailure。解决方法是您可以手动将其删除。在kv之后和json过滤器之前添加以下mutate过滤器:
mutate {
gsub => [ "events","\]",""]
gsub => [ "events","\[",""]
}
但是,这不适用于
[{"foo":"bar"},{"foo":"bar1"}]
这样的输入。因此,这里有4个选项:选项a)丑陋的gsub
一个丑陋的解决方法将是另一个gsub:
gsub => [ "event","\},\{",","]
但这会消除内部关系,所以我想您不想这样做。
选项b)分割
更好的方法可能是使用拆分过滤器:
split {
field => "event"
terminator => ","
}
mutate {
gsub => [ "event","\]",""]
gsub => [ "event","\[",""]
}
json{
source=> "event"
}
这将生成多个事件。 (第一个使用
foo = bar
,第二个使用foo1 = bar1
。)选项c)变异分割
您可能希望将所有值都包含在一个logstash事件中。您可以使用mutate => split过滤器生成一个数组,并在存在条目的情况下解析json。不幸的是,您必须为每个条目设置一个条件,因为logstash在其配置中不支持循环。
mutate {
gsub => [ "event","\]",""]
gsub => [ "event","\[",""]
split => [ "event", "," ]
}
json{
source=> "event[0]"
target => "result[0]"
}
if 'event[1]' {
json{
source=> "event[1]"
target => "result[1]"
}
if 'event[2]' {
json{
source=> "event[2]"
target => "result[2]"
}
}
# You would have to specify more conditionals if you expect even more dictionaries
}
选项d)Ruby1
以下工作(在您的kv过滤器之后):宁可使用选项e)
mutate {
gsub => [ "event","\]",""]
gsub => [ "event","\[",""]
}
ruby {
init => "require 'json'"
code => "
e = event['event'].split(',')
ary = Array.new
e.each do |x|
hash = JSON.parse(x)
hash.each do |key, value|
ary.push( { key => value } )
end
end
event['result'] = ary
"
}
更新
选项e)Ruby2
经过一些测试,这可能是最好的方法。在您的kv过滤器之后使用此功能:
ruby {
init => "require 'json'"
code => "event['result'] = JSON.parse(event['event'])"
}
关于elasticsearch - Logstash-JSON解析列表,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/31801838/