ELK 日志采集日志截断问题与优化方案记录

在部署 ELK（Elasticsearch + Logstash + Kibana）日志平台过程中，日志采集经常出现以下问题：

• 日志收集不完整，部分异常日志无法收集。

• 单条日志被截断为多条，在 Kibana 中展示不一致。

这些问题大多由 Filebeat 在日志采集阶段配置不当所引起。本文将详细说明排查与优化过程。

一、问题分析

1. 多行日志被截断原因

多行日志（如 Java 异常堆栈）被 Filebeat 拆分为多条日志，原因包括：

• multiline 配置未生效：原配置中 multiline 配置错误地放在 output.kafka 下，应放在 filebeat.inputs 级别。

• 前后端日志格式不一致：如后端以时间戳开头，前端以 IP 开头。

2. 原始错误配置（示例）

 filebeat.inputs:
 - type: log
   paths:
   - /var/log/*
   encoding: utf-8
   fields:
     module_name: "${MODULE_NAME}"
     module_namespace: "${POD_NAMESPACE}"
   harvester_buffer_size: 16384
 ​
 output.kafka:
   hosts: ["10.36.6.71:9092","10.36.6.72:9092","10.36.6.73:9092"]
   topic: "cargo"
   multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} ...'
   ...  # ❌ multiline 错误配置位置

二、优化后的 Filebeat 配置

1. 后端日志采集配置

 - type: log
   enabled: true
   paths:
     - /var/log/*
   fields:
     log_type: backend
     module_name: "${MODULE_NAME}"
     module_namespace: "${POD_NAMESPACE}"
   fields_under_root: true
   multiline.pattern: '^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[,\.]\d{3}'
   multiline.negate: true
   multiline.match: after
   harvester_buffer_size: 65536

2. 前端日志采集配置

 - type: log
   enabled: true
   paths:
     - /var/log/*
   fields:
     log_type: frontend
     module_name: "${MODULE_NAME}"
     module_namespace: "${POD_NAMESPACE}"
   fields_under_root: true
   multiline.pattern: '^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'
   multiline.negate: true
   multiline.match: after
   harvester_buffer_size: 65536

3. Kafka 输出配置

 output.kafka:
   hosts: ["10.36.6.71:9092", "10.36.6.72:9092", "10.36.6.73:9092"]
   topic: "cargo"
   compression: gzip
   max_message_bytes: 1000000
   version: 0.11.0.0
   partition.round_robin:
     reachable_only: true
   required_acks: 1
   bulk_max_size: 1000
   worker: 4

三、优化后的 Logstash 配置

1. 输入部分

 input {
   kafka {
     bootstrap_servers => "10.36.6.71:9092,10.36.6.72:9092,10.36.6.73:9092"
     topics => ["wulaoer"]
     group_id => "wulaoer"
     client_id => "wulaoer"
     type => "wulaoer"
     auto_offset_reset => "latest"
     tags => ["wulaoer_data"]
     codec => "json"
   }
 }

2. 过滤器部分

 filter {
   if [fields][log_type] == "backend" {
     grok {
       match => { "message" => "(?<time>%{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{HOUR}:%{MINUTE}:%{SECOND}(?:[.,]%{INT})?)" }
       add_field => { "extracted_timestamp" => "%{time}" }
     }
 ​
     date {
       match => ["extracted_timestamp", "yyyy-MM-dd HH:mm:ss.SSS", "yyyy-MM-dd HH:mm:ss,SSS", "yyyy-MM-dd HH:mm:ss"]
       target => "@timestamp"
       timezone => "Asia/Shanghai"
     }
   }
 ​
   if [fields][log_type] == "frontend" {
     grok {
       match => { "message" => "^%{IP:client_ip} %{DATA:ident} %{DATA:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{URIPATH:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:response_code} %{NUMBER:bytes}" }
     }
 ​
     date {
       match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
       target => "@timestamp"
       timezone => "Asia/Shanghai"
     }
   }
 ​
   mutate {
     rename => { "[fields][module_name]" => "module_name" }
     rename => { "[fields][module_namespace]" => "module_namespace" }
     remove_field => ["host", "event", "agent", "ecs"]
   }
 }

3. 输出部分

 output {
   if [type] == "cargo" {
     elasticsearch {
       hosts => ["http://10.36.6.71:9200", "http://10.36.6.72:9200", "http://10.36.6.73:9200"]
       index => "%{module_namespace}-%{module_name}-%{+YYYY.MM.dd}"
       user => "elastic"
       password => "xxxxxxxxxxx"
       manage_template => false
     }
   }
 }

四、总结与建议

1. Filebeat 的 multiline 应配置在 inputs 中，而不是 output。

2. 根据不同类型日志（前端、后端）配置不同的 multiline pattern。

3. Kafka 传输时使用 JSON 编码，在 Logstash 中用 **codec => "json"** 自动解析字段。

4. 根据日志类型在 Logstash 中进行分流和解析，统一处理公共字段。

5. 调整 **harvester_buffer_size**，防止截断大日志。

6. 修改配置后需重启 Filebeat 和 Logstash 服务才能生效。

部署完成后，通过 Kibana 查看日志可以确保日志不再被截断，格式完整且易于分析。