Logstash配置问题：过滤空行导致多行日志解析失效

问题背景 #

原本通过multiline codec实现的多行日志提取功能正常运行，为过滤日志中空首行导致的Kibana空条目，在filter段末尾添加空行过滤规则后，空行虽被成功过滤，但多行日志提取功能失效。

核心原因分析 #

问题出在处理顺序颠倒：

1. 你在input阶段先使用multiline codec合并多行日志，此时日志中的空行会被multiline视为独立事件（空行不匹配^\[%{NOTSPACE:timestamp}\]模式，且无前置事件可合并）。
2. 若空行之后存在属于后续正常日志的延续行（即不匹配时间戳开头的行），multiline会将这些延续行合并到空行事件中，形成包含空行+延续行的无效事件。
3. 该无效事件的message并非全空白，不会被filter末尾的规则drop，但会因开头无时间戳导致grok解析失败（出现_grokparsefailure标签），最终表现为多行日志提取失效——原本应合并到正常日志的行变成了独立的无效事件。

解决方案 #

调整处理顺序，先过滤空行，再合并多行日志，推荐使用multiline filter替代multiline codec，具体配置修改如下：

修改后的Logstash配置 #

input {
    beats {
        port => 5044
    }
    
    file {
        path => "/usr/share/logstash/iway_logs/*"       
        start_position => "beginning"
        sincedb_path => "/dev/null"
        # 改用plain codec读取每行原始内容
        codec => plain { charset => "UTF-8" }
  }
}

filter {    
    # 第一步：过滤空行，直接drop全空白的行
    if [message] =~ /^\s*$/ {
        drop { }
    }
    
    # 第二步：处理多行合并
    multiline {
        pattern => "^\[%{NOTSPACE:timestamp}\]"
        negate => true
        what => "previous"
        max_lines => 2500
    }
    
    # 后续原有过滤逻辑保持不变
    grok {
        match => {  "message" => 
            ['(?m)\[%{NOTSPACE:timestamp}\]%{SPACE}%{WORD:level}%{SPACE}\(%{NOTSPACE:entity}\)%{SPACE}%{GREEDYDATA:rawlog}'         
            ]
        }
    }
                    
    date {
      match => [ "timestamp", "yyyy-MM-dd'T'HH:mm:ss.SSS"]
      target => "@timestamp"
    }
        
    grok {
        match => { "entity" =>          ['(?:W.%{GREEDYDATA:channel}:%{GREEDYDATA:inlet}:%{GREEDYDATA:listener}\.%{GREEDYDATA:workerid}|W.%{GREEDYDATA:channel}\.%{GREEDYDATA:workerid}|%{GREEDYDATA:channel}:%{GREEDYDATA:inlet}:%{GREEDYDATA:listener}\.%{GREEDYDATA:workerid}|%{GREEDYDATA:channel}:%{GREEDYDATA:inlet}:%{GREEDYDATA:listener}|%{GREEDYDATA:channel})']
        }
    }
    
    dissect {
        mapping => {
            "[log][file][path]" => "/usr/share/logstash/iway_logs/%{serverName}#%{configName}#%{?ignore}.log"
        }
    }
    
}

output {
    elasticsearch {
        hosts => "${ELASTICSEARCH_HOST_PORT}"
        index => "iway_"
        user => "${ELASTIC_USERNAME}"
        password => "${ELASTIC_PASSWORD}"
        ssl => true
        ssl_certificate_verification => false
        cacert => "/certs/ca.crt"
    }
}

配置说明 #

1. input阶段：将file输入的multiline codec替换为plain codec，确保每行日志都被单独读取，不提前合并。
2. filter阶段：
   • 先添加空行过滤规则，直接drop全空白的行，避免空行进入后续处理流程。
   • 再使用multiline filter合并多行日志，此时所有待处理的行均为非空白行，延续行可正确合并到前置的正常日志行中。
3. 原有grok、date、dissect等逻辑保持不变，确保日志解析流程正常运行。

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