在logstash处理filebeat收集的日志数据以后我们可以发现logstash解析以后有一个@timestamp时间，然后而这个时间跟我们grok解析日志以后里面所包含的时间不是一致的，logstash收集日志的@timestamp时间是日志采集时间，而且使用的是UTC时区，相对于北京时间会晚8个小时，我们一般都不太关注日志采集时间，而是更加关注日志生成时间，所以我们需要将@timestamp字段的时间改写成日志生成时间

我们这里还是用之前收集mongodb日志的例子来继续学习下如何改写@timestamp字段的时间，mongodb收集日志的方式还是可以查看我之前的笔记：https://sulao.cn/post/750

原来的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的配置在某些场景下还是非常有用的，我们这里暂时不予介绍了