elasticsearch - 过滤数据时出现logstash grok问题
问题描述
我有一个基本上是通过 rm 命令删除数据的数据,如下所示。
ttmv516,19/05/21,03:59,00-mins,dvcm,dvcm 166820 4.1 0.0 4212 736 ? DN 03:59 0:01 rm -rf /dv/project/agile/mce_dev_folic/test/install.asan/install,/dv/svgwwt/commander/workspace4/dvfcronrun_IL-SFV-RHEL6.5-K4_kinite_agile_invoke_dvfcronrun_at_given_site_50322
我在下面使用logstash grok,它工作正常,但直到最近我才看到两个奇怪的问题1)_grokparsefailure
另一个2)Hostname Field
没有正确显示,即它的初始字符不存在,就像ttmv516
看起来一样mv516
。
%{HOSTNAME:Hostname},%{DATE:Date},%{HOUR:dt_h}:%{MINUTE:dt_m},%{NUMBER:duration}-%{WORD:hm},%{USER:User},%{USER:User_1} %{NUMBER:Pid} %{NUMBER:float} %{NUMBER:float} %{NUMBER:Num_1} %{NUMBER:Num_2} %{DATA} (?:%{HOUR:dt_h1}:|)(?:%{MINUTE:dt_m1}|) (?:%{HOUR:dt_h2}:|)(?:%{MINUTE:dt_m2}|)%{GREEDYDATA:CMD},%{GREEDYDATA:PWD_PATH}
但是,在 Kibana 数据中使用 grok 调试器进行相同的测试显示正确。
我的logstash文件如下。
cat /etc/logstash/conf.d/rmlog.conf
input {
file {
path => [ "/data/rm_logs/*.txt" ]
start_position => beginning
sincedb_path => "/data/registry-1"
max_open_files => 64000
type => "rmlog"
}
}
filter {
if [type] == "rmlog" {
grok {
match => { "message" => "%{HOSTNAME:Hostname},%{DATE:Date},%{HOUR:dt_h}:%{MINUTE:dt_m},%{NUMBER:duration}-%{WORD:hm},%{USER:User},%{USER:User_1} %{NUMBER:Pid} %{NUMBER:float} %{NUMBER:float} %{NUMBER:Num_1} %{NUMBER:Num_2} %{DATA} (?:%{HOUR:dt_h1}:|)(?:%{MINUTE:dt_m1}|) (?:%{HOUR:dt_h2}:|)(?:%{MINUTE:dt_m2}|)%{GREEDYDATA:CMD},%{GREEDYDATA:PWD_PATH}" }
add_field => [ "received_at", "%{@timestamp}" ]
remove_field => [ "@version", "host", "message", "_type", "_index", "_score" ]
}
}
}
output {
if [type] == "rmlog" {
elasticsearch {
hosts => ["myhost.xyz.com:9200"]
manage_template => false
index => "pt-rmlog-%{+YYYY.MM.dd}"
}
}
}
任何帮助建议将不胜感激。
编辑:
根据我的观察,它失败的消息..
ttmv540,19/05/21,03:59,00-hrs,USER,USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND,/local/ntr/ttmv540.373
ttmv541,19/05/21,03:43,-mins,USER,USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND,/local/ntr/ttmv541.373
但是,我尝试使用以下条件编辑 grok,但它仍然删除了少数字段..
input {
file {
path => [ "/data/rm_logs/*.txt" ]
start_position => beginning
max_open_files => 64000
sincedb_path => "/data/registry-1"
type => "rmlog"
}
}
filter {
if [type] == "rmlog" {
grok {
match => { "message" => "%{HOSTNAME:hostname},%{DATE:date},%{HOUR:time_h}:%{MINUTE:time_m},%{NUMBER:duration}-%{WORD:hm},%{USER:user},%{USER:group} %{NUMBER:pid} %{NUMBER:float} %{NUMBER:float} %{NUMBER:num_1} %{NUMBER:num_2} %{DATA} (?:%{HOUR:time_h1}:|)(?:%{MINUTE:time_m1}|) (?:%{HOUR:time_h2}:|)(?:%{MINUTE:time_m2}|)%{GREEDYDATA:cmd},%{GREEDYDATA:pwd}" }
add_field => [ "received_at", "%{@timestamp}" ]
remove_field => [ "@version", "host", "message", "_type", "_index", "_score" ]
}
}
if "_grokparsefailure" in [tags] {
grok {
match => { "message" => "%{HOSTNAME:hostname},%{DATE:date},%{HOUR:time_h}:%{MINUTE:time_m},-%{WORD:duration},%{USER:user},%{USER:group}%{GREEDYDATA:cmd}" }
add_field => [ "received_at", "%{@timestamp}" ]
remove_field => [ "@version", "host", "message", "_type", "_index", "_score" ]
}
}
}
output {
if [type] == "rmlog" {
elasticsearch {
hosts => ["myhost.xyz.com:9200"]
manage_template => false
index => "pt-rmlog-%{+YYYY.MM.dd}"
}
}
}
注意:看起来_grokparsefailure
标签对以下消息有效,但在另一个消息上仍然失败..
1)这有效..
ttmv541,19/05/21,03:43,-mins,USER,USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND,/local/ntr/ttmv541.373
ttmv540,19/05/21,03:59,00-hrs,USER,USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND,/local/ntr/ttmv540.373
2)第二行文本日志失败,因为它有00-hrs
与之关联的数字,现在无法满足以下两个条件 grok ..
%{HOSTNAME:hostname},%{DATE:date},%{HOUR:time_h}:%{MINUTE:time_m},-%{WORD:duration},%{USER:user},%{USER:group}%{GREEDYDATA:cmd}
解决方案
我将处理分为两部分,一部分处理主机名和时间戳问题,另一部分处理行的其余部分。我发现这使维护更容易。
所以你留下了这两个输入:
ttmv541,19/05/21,03:43,-mins
ttmv540,19/05/21,03:59,00-hrs
您的两个模式将很好地匹配第一个部分,所以问题是您希望如何在时间之后解析出这些东西。在您的原始模式中,您使用duration
的是数字部分和hm
单位。在您的第二种模式中,您似乎将单位放入duration
,这可能是不正确的。
如果没有更多信息,看起来持续时间是可选的,但您将始终拥有单位。这可以反映在您的模式中,例如:
(%{NUMBER:duration})?-%{WORD:hm}
另请注意,如果您最终需要多个模式,则不必依赖 grokparsefailure 来使用它们 - match->message 可以采用数组。有关示例,请参见文档。
推荐阅读
- flutter - 如何做列表元素的随机生成器?
- csv - 用于将 xlsb 转换为 csv 的 Nifi 处理器
- html - 添加网站图标 -
不是 - workflow - 设置字段值工作流
- javascript - 如何在javascript中将数字格式更改为字母月份
- javascript - Youtube Iframe API 多个视频使用自定义按钮开始每个视频
- java - 如何关闭在 startActivityForResult 上打开的意图
- c# - .NET Framework 与 .NET Core - 将证书添加到 RESTful API 调用中的一个,而不是另一个
- python - 优化距离矩阵计算
- javascript - Amchart:绘制不同时区的图表(例如浏览器时区 = 美国/多伦多,图表目标时区 = 美国/芝加哥)