如何配置Logstash将旧日志按文件日期写入对应日期的Elasticsearch索引

问题根因 #

你当前配置的核心问题是date过滤器的时间格式和日志实际时间格式不匹配：日志中的时间毫秒部分用逗号分隔（如2021-10-27 09:13:16,225），但你配置的匹配格式用的是点作为分隔符，导致时间解析失败。Logstash会默认用日志处理的当前时间作为@timestamp的值，最终索引后缀就变成了索引创建的日期，和日志实际日期不符。

调整方案 #

方案1：基于日志内容时间生成索引（推荐，日志时间准确时使用） #

仅需要修正date过滤器的匹配格式即可，调整后%{+yyyy.MM.dd}会自动取解析成功的@timestamp日期作为索引后缀，完全和日志实际时间对齐：

filter {
  grok {
    match => { "message" => "(?<timestamp>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) %{LOGLEVEL:logLevel} %{JAVACLASS:className} \[(?<threadname>[A-Za-z0-9_-]+)\] %{GREEDYDATA:API_request_and_response}" }
  }
  date {
    match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,SSS" ]
    target => "@timestamp"
    tag_on_failure => ["_date_parse_failed"]
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "admins-%{+yyyy.MM.dd}"
  }
}

新增的tag_on_failure配置用于标记时间解析失败的日志，方便后续排查异常数据。

方案2：基于日志文件名日期生成索引（严格对齐文件日期时使用） #

如果需要保证单个日志文件的所有日志都进入同一天的索引，避免单文件内少量跨天日志分散到不同索引，可以额外从日志文件路径中提取日期作为索引命名基准：
以下示例假设你的日志文件名格式为admin.2021-10-27.log，可根据实际文件名规则调整grok匹配规则：

filter {
  # 提取日志内容字段
  grok {
    match => { "message" => "(?<timestamp>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}) %{LOGLEVEL:logLevel} %{JAVACLASS:className} \[(?<threadname>[A-Za-z0-9_-]+)\] %{GREEDYDATA:API_request_and_response}" }
  }
  # 从文件路径中提取日志所属日期
  grok {
    match => { "path" => "admin\.(?<file_date>\d{4}-\d{2}-\d{2})\.log" }
  }
  # 可选：仍将@timestamp设置为日志内容中的实际时间，不影响索引命名
  date {
    match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,SSS" ]
    target => "@timestamp"
    tag_on_failure => ["_date_parse_failed"]
  }
  # 替换日期分隔符适配索引命名格式
  mutate {
    gsub => [ "file_date", "-", "." ]
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "admins-%{file_date}"
  }
}

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