问题描述

之前在搭建ELK日志收集系统时，使用Kinaba做了异常日志统计的看板。但是今天发现有些异常日志可以被搜索，但是在统计看板中没有统计到。

通过对比可被统计和不可被统计的日志字段，发现不可被统计的日志是因为没有创建keyword字段，通过检查索映射发现这个字段配置如下：

    "content": {
        "type": "text",
        "fields": {
        "keyword": {
            "type": "keyword",
            "ignore_above": 256
        }
        }
    }

可知是异常日志超长，导致keyword字段无法创建。

解决方案

异常日志因为打印堆栈信息，通常会很长。如果只是简单的修改ignore_above的值，则需要调为很大的值，在做聚合统计时肯定会影响性能。

所以一开始想试试在ElasticSearch中能否将content.keyword截取，但是查询后发现ElasticSearch不支持对keyword字段进行截取。

然后转而试试能否在Logstash处理日志时进行截取。

最终查到通过logstash的mutate插件的gsub功能截取日志。为了不丢失原始日志，可以将截取后的日志存入一个新的字段，如content_short:

filter {
    grok {
        match => { "message" => "%{DATA:timestamp} %{LOGLEVEL:loglevel} %{DATA:logger} %{DATA:class} \[%{DATA:traceId}\] \[%{DATA:biz}\] \[%{DATA:logFollower}\] %{GREEDYDATA:content}" }
    }
    mutate {
        add_field => { "content_short" => "%{content}" }
        gsub => [
            "content_short", "(.{255}).*", "\1"
        ]
    }
}

这段配置的含义如下：

• mutate：这是 Logstash 的一个插件，用于对事件数据进行各种转换操作。
• gsub：这是 mutate 插件的一个功能，用于对字段值进行正则表达式替换。
• "(?m)(.{256}).*"：(?m) 是一个模式修饰符，使 . 匹配任何字符，包括换行符。(.{256}).* 匹配前 256 个字符和之后的所有字符。
• "\1"：这是替换字符串。\1 表示正则表达式中的第一个捕获组，即前 256 个字符。

这样就可以在Kibana中使用content_short字段进行聚合统计了。