如何使用FileBeat将整份日志作为单个事件发送至Elasticsearch

这个需求完全可以实现，核心是在日志采集侧配置多行合并规则，根据你的日志特征匹配需要聚合的行即可，两种常见场景的实现方式如下：

场景1：按单条日志条目合并（包含多行异常栈） #

你提供的日志特征非常明确：正常日志的首行以DD/MM/YYYY HH:MM:SS格式的时间戳开头，异常栈的后续行带缩进、无时间戳前缀，只需要配置采集规则将无时间戳的行合并到上一条日志条目即可。

最常用的Filebeat采集器配置示例如下：

filebeat.inputs:
- type: filestream
  paths:
    - /你的/日志/目录/*.log
    - /你的/日志/目录/*.txt
  # 多行合并配置
  multiline.type: pattern
  # 匹配首行的时间戳格式
  multiline.pattern: '^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}'
  multiline.negate: true
  multiline.match: after

配置逻辑：所有不匹配「时间戳开头」规则的行，都会被追加到上一条匹配规则的日志末尾，刚好适配异常栈缩进行的合并需求。

如果用Logstash作为采集/处理组件，配置multiline编码插件即可：

input {
  file {
    path => ["/你的/日志/目录/*.log", "/你的/日志/目录/*.txt"]
    codec => multiline {
      pattern => "^[0-9]{2}/[0-9]{2}/[0-9]{4} [0-9]{2}:[0-9]{2}:[0-9]{2}"
      negate => true
      what => "previous"
    }
  }
}

场景2：整个日志文件内容合并为单个事件 #

如果你的单份日志对应应用的一次完整运行（比如一次性批处理任务的单次运行日志），需要将整个文件的所有行封装为一条事件上报，可以通过采集器的文件完成触发规则实现：

• Filebeat可配置file_complete事件触发上报，结合内容聚合规则将整个文件内容合并为单个字段
• 也可以自定义简单脚本监听日志目录，检测到日志文件写完不再修改后，读取整个文件内容作为单个字段发送到Elasticsearch，再同步到Kibana即可。

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