ELK异常分类：脚本字段对多行日志不生效，求解决及更优方案

一、修改现有脚本支持多行日志 #

你的脚本失效大概率是因为多行场景下message字段的存储形式不符合预期，分两种情况调整：

情况1：多行日志被拆分为多值字段（每行一个元素） #

如果Elasticsearch中单条异常对应的文档里，message是多值字段（每个异常行对应一个值），需要遍历所有message元素进行匹配：

if (doc.containsKey('message')) {
    def messages = doc['message'];
    for (def msg : messages) {
        if (msg.value.contains("java.lang.IllegalStateException")) {
            return "Error-1";
        } else if (msg.value.contains("java.lang.NullPointerException")) {
            return "Error-2";
        } else if (msg.value.contains("java.lang.IndexOutOfBoundsException")) {
            return "Error-3";
        }
        // 可继续添加其他异常类型判断
    }
    return "Other Error";
} else {
    return "No Message";
}

情况2：多行日志合并为单值字段（含换行的完整异常栈） #

如果message是包含完整异常栈的单值字段，但原脚本用了message.keyword导致内容被截断（keyword类型默认有长度限制），直接改用message字段匹配：

if (doc.containsKey('message')) {
    if (doc['message'].size() > 0) {
        def message = doc['message'].value;
        if (message.contains("java.lang.IllegalStateException")) {
            return "Error-1";
        } else if (message.contains("java.lang.NullPointerException")) {
            return "Error-2";
        } else if (message.contains("java.lang.IndexOutOfBoundsException")) {
            return "Error-3";
        }
        return "Other Error";
    }
}
return "No Message";

二、更优实现方案：预处理阶段提取异常字段 #

脚本字段是查询时实时计算，数据量大时会有明显性能损耗。推荐在数据采集/预处理阶段就提取异常类型，存入独立字段，后续查询聚合效率大幅提升。

方案1：Filebeat + Elasticsearch Ingest Pipeline（推荐轻量架构） #

步骤1：配置Filebeat合并多行异常日志

修改Filebeat配置，确保整个Java异常栈被合并为单条文档：

filebeat.inputs:
- type: log
  paths:
    - /path/to/your/logs/*.log
  # 匹配异常栈的后续行，合并到主日志行
  multiline.pattern: '^\s+at|^Caused by:'
  multiline.negate: false
  multiline.match: after

步骤2：创建Ingest Pipeline提取异常类型

在Kibana DevTools中执行以下命令，创建自动提取异常信息的管道：

PUT _ingest/pipeline/extract-java-exception
{
  "description": "提取Java异常类型及分类",
  "processors": [
    {
      "grok": {
        "field": "message",
        "patterns": [
          "%{DATA}%{JAVACLASS:exception_type}:"
        ],
        "pattern_definitions": {
          "JAVACLASS": "[a-zA-Z0-9\\.]+"
        },
        "ignore_missing": true,
        "ignore_failure": true
      }
    },
    {
      "script": {
        "source": """
          if (ctx.containsKey('exception_type')) {
            // 提取简短异常名（如NullPointerException而非java.lang.NullPointerException）
            def classNameParts = ctx.exception_type.split('\\.');
            ctx.exception_short_type = classNameParts[classNameParts.length - 1];
            // 分类常见异常
            def commonExceptions = ['IllegalStateException', 'NullPointerException', 'IndexOutOfBoundsException'];
            ctx.exception_category = commonExceptions.contains(ctx.exception_short_type) ? '常见运行时异常' : '其他异常';
          } else {
            ctx.exception_category = '非异常日志';
          }
        """
      }
    }
  ]
}

步骤3：Filebeat关联管道

修改Filebeat输出配置，让数据导入时自动执行管道：

output.elasticsearch:
  hosts: ["localhost:9200"]
  pipeline: "extract-java-exception"

方案2：Logstash预处理（适用于已有Logstash的架构） #

用Logstash的grok和mutate过滤器提前提取异常信息：

filter {
  grok {
    match => { "message" => "%{DATA}%{JAVACLASS:exception_type}:" }
    pattern_definitions => {
      "JAVACLASS" => "[a-zA-Z0-9\.]+"
    }
    tag_on_failure => ["_grokparsefailure"]
  }

  if [exception_type] {
    mutate {
      split => { "exception_type" => "." }
      add_field => { "exception_short_type" => "%{[exception_type][-1]}" }
      remove_field => "[exception_type][0..-2]"
    }

    if [exception_short_type] in ["IllegalStateException", "NullPointerException", "IndexOutOfBoundsException"] {
      mutate {
        add_field => { "exception_category" => "常见运行时异常" }
      }
    } else {
      mutate {
        add_field => { "exception_category" => "其他异常" }
      }
    }
  } else {
    mutate {
      add_field => { "exception_category" => "非异常日志" }
    }
  }
}

处理完成后，每条日志会新增exception_type（完整类名）、exception_short_type（简短异常名）、exception_category（分类）三个字段，直接用这些字段做筛选、聚合即可。

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