我一直在尝试使用logstash解析我的python回溯日志。我的日志如下：
[pid: 26422|app: 0|req: 73/73] 192.168.1.1 () {34 vars in 592 bytes} [Wed Feb 18 13:35:55 2015] GET /data => generated 2538923 bytes in 4078 msecs (HTTP/1.1 200) 2 headers in 85 bytes (1 switches on core 0)
Traceback (most recent call last):
File "/var/www/analytics/parser.py", line 257, in parselogfile
parselogline(basedir, lne)
File "/var/www/analytics/parser.py", line 157, in parselogline
pval = understandpost(parts[3])
File "/var/www/analytics/parser.py", line 98, in understandpost
val = json.loads(dct["events"])
File "/usr/lib/python2.7/json/__init__.py", line 338, in loads
return _default_decoder.decode(s)
File "/usr/lib/python2.7/json/decoder.py", line 366, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
File "/usr/lib/python2.7/json/decoder.py", line 382, in raw_decode
obj, end = self.scan_once(s, idx)
ValueError: Unterminated string starting at: line 1 column 355 (char 354)
到目前为止，我已经能够解析日志，除了最后一行，即
ValueError: Unterminated string starting at: line 1 column 355 (char 354)
我正在使用多行过滤器来实现这一点。我的logstash配置看起来像这样：
filter {
multiline {
pattern => "^Traceback"
what => "previous"
}
multiline {
pattern => "^ "
what => "previous"
}
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})%{GREEDYDATA:traceback}"
]
}
if "_grokparsefailure" in [tags] {
multiline {
pattern => "^.*$"
what => "previous"
negate => "true"
}
}
if "_grokparsefailure" in [tags] {
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})%{GREEDYDATA:traceback}"
]
remove_tag => ["_grokparsefailure"]
}
}
}
但我的最后一行不是解析。相反，它仍然给我一个错误，并且还成倍地增加了处理时间。关于如何解析回溯的最后一行，有什么建议吗？
好吧，我找到了一个解决方案。因此，我采用的方法是，我将忽略以"["开头的日志消息的开头，所有其他行都将附加在前一条消息的末尾。然后可以应用grok过滤器，并可以解析回溯。注意，我必须应用两个grok过滤器：
-
对于使用GREEDYDATA进行回溯以获取回溯的情况。
-
因为当没有回溯时，GREEDYDATA解析失败，我将不得不删除_grokparsefailure标记，然后再次应用没有GREEDYATA的grok。这是在if块的帮助下完成的。
最后的logstash过滤器看起来像这样：
filter {
multiline {
pattern => "^[^[]"
what => "previous"
}
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})%{GREEDYDATA:traceback}"
]
}
if "_grokparsefailure" in [tags] {
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})"
]
remove_tag => ["_grokparsefailure"]
}
}
else {
mutate {
convert => {"traceback" => "string"}
}
}
date {
match => ["timestamp", "dd/MM/YYYY:HH:MM:ss Z"]
locale => en
}
geoip {
source => "clientip"
}
useragent {
source => "agent"
target => "Useragent"
}
}
或者，如果您不想使用if块来检查另一个grok模式并删除_grokparsefailure，则可以使用第一个grok筛选器来检查这两种消息类型，方法是在grok筛选器的match数组中包括多个消息模式检查。可以这样做：
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})",
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})%{GREEDYDATA:traceback}"
]
}
还有第三种方法（可能是最优雅的方法）。它看起来像这样：
grok {
match => [
"message", "[pid: %{NUMBER:process_id:int}|app: 0|req: %{NUMBER}/%{NUMBER}] %{IPORHOST:clientip} () {%{NUMBER:vars:int} vars in %{NUMBER:bytes:int} bytes} [%{GREEDYDATA:timestamp}] %{WORD:method} /%{GREEDYDATA:referrer} => generated %{NUMBER:generated_bytes:int} bytes in %{NUMBER} msecs (HTTP/%{NUMBER} %{NUMBER:status_code:int}) %{NUMBER:headers:int} headers in %{NUMBER:header_bytes:int} bytes (%{NUMBER:switches:int} switches on core %{NUMBER:core:int})(%{GREEDYDATA:traceback})?"
]
}
请注意，在这种方法中，存在性是可选的字段必须包含在"（）？"中。此处，(%{GREEDYDATA:traceback})?
因此，grok过滤器会发现，如果字段可用，就会对其进行解析。否则，它将被跳过。