如何在Elasticsearch中自定义时间戳？Logstash配置求助

当然可以实现！问题的核心是让Logstash解析日志里的时间戳，并替换掉默认的@timestamp字段（Logstash默认用事件创建时间填充这个字段，这就是你现在看到“创建日期作为时间戳”的原因）。下面是具体的解决步骤和配置示例：

1. 提取日志中的时间戳字段 #

首先用grok插件从你的日志消息里把时间戳单独提取出来。你的日志格式是"t message 2018-01-21 23:55:00.266 INFO"，对应的grok匹配模式可以这样写：

grok {
  match => { "message" => "%{DATA:log_content} %{TIMESTAMP_ISO8601:log_timestamp} %{LOGLEVEL:loglevel}" }
  # 可选：匹配失败时添加标记方便排查
  tag_on_failure => ["_grokparsefailure"]
}

这里%{DATA:log_content}会匹配时间戳之前的所有内容（也就是t message），%{TIMESTAMP_ISO8601:log_timestamp}专门匹配你日志里的标准ISO时间格式，%{LOGLEVEL:loglevel}匹配最后的日志级别（比如INFO）。

2. 将提取的时间戳替换为Logstash默认时间字段 #

Logstash默认用@timestamp字段作为Elasticsearch索引的时间依据，所以我们用date插件把刚才提取的log_timestamp转换成@timestamp：

date {
  match => [ "log_timestamp", "yyyy-MM-dd HH:mm:ss.SSS" ]
  target => "@timestamp"
  timezone => "Asia/Shanghai" # 根据你的日志时区调整，比如UTC、Europe/London等
}

match参数指定了log_timestamp的格式，target设置为@timestamp后，就会覆盖原来的事件创建时间。如果日志时间不是UTC时区，一定要加上timezone参数避免时间偏移。

3. 完整的Logstash配置示例 #

把上述插件整合到你的配置中，完整结构大概是这样：

input {
  # 你的输入配置（比如file、beats、tcp等）
  file {
    path => "/path/to/your/logs/*.log"
    start_position => "beginning"
  }
}

filter {
  grok {
    match => { "message" => "%{DATA:log_content} %{TIMESTAMP_ISO8601:log_timestamp} %{LOGLEVEL:loglevel}" }
    tag_on_failure => ["_grokparsefailure"]
  }
  
  date {
    match => [ "log_timestamp", "yyyy-MM-dd HH:mm:ss.SSS" ]
    target => "@timestamp"
    timezone => "Asia/Shanghai"
  }
  
  # 可选：如果不需要原始message字段，可以删除它节省空间
  mutate {
    remove_field => ["message"]
  }
}

output {
  elasticsearch {
    hosts => ["your-es-host:9200"]
    index => "app-logs-%{+YYYY.MM.dd}" # 如果是按时间分片的索引，现在会用日志时间生成索引名
  }
}

4. 验证效果 #

重启Logstash后，发送一条测试日志到输入源，然后查看Elasticsearch中的文档：你会发现@timestamp字段的值已经变成日志里的时间，而不是Logstash接收事件的时间了，这样索引就会基于这个日志时间来处理。

如果grok匹配出现问题，可以用Logstash自带的grokdebug工具（或者本地调试工具）测试你的匹配模式，调整到能正确提取出log_timestamp为止。

内容的提问来源于stack exchange，提问作者rashidcmb