LOGSTASH 7.8.x无限循环且仅向Elasticsearch插入单条数据求助

问题背景 #

作为ELK新手，使用Logstash 7.8.x处理日志时出现两个异常：Elasticsearch的目标索引仅插入一条数据，同时Logstash陷入无限循环，需手动终止。

示例日志 #

2021-07-18 09:15:30,000 INFO Sample log message 01
2022-07-18 10:20:45,111 ERROR Sample log message 02
2023-07-18 11:20:45,222 DEBUG Sample log message 03
2024-07-18 12:20:45,333 WARN Sample log message 04

Logstash配置文件 #

input {
  file {
    path => "/home/sample.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
    close_older=> "1 second"
 }
}

filter {
    grok {
          match => { "resource" => "%{TIMESTAMP_ISO8601:log_timestamp} %{LOGLEVEL:log_level} %{GREEDYDATA:log_message}" }
         }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "sample_logs"
    document_id => "%{my_doc_for_logs}"
    document_type => "_doc"
    codec => "json"
 }
  stdout { codec => rubydebug }
}

启动命令 #

bin/logstash -f /etc/logstash/conf.d/sshd.conf --log.level debug

问题现象 #

• Elasticsearch的sample_logs索引中仅存在1条数据，可通过以下命令验证：

  curl -XGET "http://localhost:9200/sample_logs/_search?size=1000" | jq .
  

• Logstash持续运行陷入无限循环，必须按CTRL + C才能终止进程。

原因分析 #

1. Grok匹配字段错误：Filter中grok插件指定匹配resource字段，但File输入插件默认将日志内容存入message字段，导致grok匹配失败，后续字段无法正确解析。
2. Document ID配置错误：Output中document_id使用了不存在的字段%{my_doc_for_logs}，所有文档会生成相同的空ID或默认ID，新数据持续覆盖旧数据，最终仅保留一条。
3. 文件读取配置不合理：sincedb_path => "/dev/null"会禁用进度记录，close_older => "1 second"会快速关闭文件，Logstash会反复从头读取日志文件，引发无限循环。

解决步骤 #

1. 修正Grok匹配字段：将grok的匹配目标改为message字段：

   filter {
       grok {
             match => { "message" => "%{TIMESTAMP_ISO8601:log_timestamp} %{LOGLEVEL:log_level} %{GREEDYDATA:log_message}" }
            }
   }
   

2. 调整Document ID配置：
   • 若不需要自定义ID，直接删除document_id配置，Elasticsearch会自动生成唯一ID；
   • 若需自定义ID，使用日志中唯一标识字段组合（如时间+日志级别）：

     document_id => "%{log_timestamp}-%{log_level}"
     

3. 优化文件读取配置：
   • 若只需读取一次日志，可保留start_position => "beginning"，但去掉sincedb_path => "/dev/null"，让Logstash记录读取进度；
   • 调整close_older为合理值（如"1 hour"），避免频繁关闭重启文件读取：

     input {
       file {
         path => "/home/sample.log"
         start_position => "beginning"
         close_older=> "1 hour"
      }
     }
     

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