在logstash处理filebeat收集的日志数据以后我们可以发现logstash解析以后有一个@timestamp时间，然后而这个时间跟我们grok解析日志以后里面所包含的时间不是一致的，logstash收集日志的@timestamp时间是日志采集时间，而且使用的是UTC时区，相对于北京时间会晚8个小时，我们一般都不太关注日志采集时间，而是更加关注日志生成时间，所以我们需要将@timestamp字段的时间改写成日志生成时间
我们这里还是用之前收集mongodb日志的例子来继续学习下如何改写@timestamp字段的时间，mongodb收集日志的方式还是可以查看我之前的笔记https://sulao.cn/post/753.html
原来的filter如下
filter {
if "mongo-logs" in [tags] {
grok {
match => ["message", '%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:i} %{NOTSPACE:opeariton} (?<info>(.)*)']
}
mutate {
remove_field => ["message"]
remove_field => ["i"]
}
}
if "beats_input_codec_plain_applied" in [tags] {
mutate {
remove_tag => ["beats_input_codec_plain_applied"]
}
}
}我们直接接着这个例子改写@timestamp字段，上面的例子我们收集到一个timestamp和@timestamp很像，我们来替换@timestamp字段的值，这里需要使用logstash另外一个插件Date，具体改写如下
filter {
if "mongo-logs" in [tags] {
grok {
match => ["message", '%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:i} %{NOTSPACE:opeariton} (?<info>(.)*)']
}
date {
match => ["timestamp", "ISO8601"]
}
mutate {
remove_field => ["message"]
remove_field => ["timestamp"]
remove_field => ["i"]
}
}
if "beats_input_codec_plain_applied" in [tags] {
mutate {
remove_tag => ["beats_input_codec_plain_applied"]
}
}
}然后我们重启logstash可以看到新进来的@timestamp字段的时间就是日志生成的时间了，logstash也可以进行单独的测试
input { stdin {} }
filter {
if "mongo-logs" in [tags] {
grok {
match => ["message", '%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:i} %{NOTSPACE:opeariton} (?<info>(.)*)']
}
date {
match => ["timestamp", "ISO8601"]
}
mutate {
remove_field => ["message"]
remove_field => ["timestamp"]
remove_field => ["i"]
}
}
if "beats_input_codec_plain_applied" in [tags] {
mutate {
remove_tag => ["beats_input_codec_plain_applied"]
}
}
}
output {
stdout { codec => rubydebug }
}启动Logstash：bin/logstash -f test.conf
logstash中的Date插件介绍
date {
match => [ "time_field", "yyyyMMdd HH:mm:ss.SSS" ]
# timezone => "UTC"
target => "end_time"
}这里可以将time_field字段解析成指定格式以后然后存储到target指定的字段中，如果不指定字段默认就是@timestamp字段，至于timezone的配置在某些场景下还是非常有用的，我们这里暂时不予介绍了