logstash拆分日志字段-JamKing工作站

1、需求背景

2、过滤器

1、使用dissect过滤器：因为日志行由固定的分隔符 | 分割，dissect是最高效、最直接的选择。它比 grok 更快，因为 grok 是基于正则表达式的，而dissect是纯文本分割。
2、使用 date 过滤器：将日志中的时间戳字符串转换为 Logstash 内部标准的 @timestamp字段。
3、使用 jsom过滤器：解析日志末尾的 JSON 字符串。这里有个小技巧，因为 JSON 内部还有一个 JSON 字符串 (extInfo，我们需要解析两次。
4、使用 mutate过滤器：进行字段重命名、数据类型转换（如将状态码 "200" 转为整数 200）和清理无用字段。

3、日志拆分配置

java日志样例

包含了时间、线程、日志等级、信息等

2025-11-14 13:48:32.835|http-nio-8085-exec-30|INFO |regexTypeLogger.regexPrintLog:51|INTERFACE|fe80:0:0:0:824:feff:fecd:cc53%eth0|10.xx.xx.xx|yun-user|yun-user-field|/auth/getUserInfoByToken|0b704643-f125-4db2-b62c-7d0404fbc963||f25187aa46064522b8bf69fa59f50b6d|17815514553|PHONE|200|1763099312815|1763099312835|20|||||||||NONE|0000||||89127dcc-eac4-4906-be67-b2de3e16aad1|{"input":{"arg0":{"token":"****"}},"output":{"success":true,"code":"0000","message":"请求成功","data":{"cutoverStatus":1,"belongsPlatform":2,"phoneNumber":"xxxxxxx","userDomainId":"xxxxxxxx","siteCode":"10003","nationCode":"+86","userAssetFlag":2}},"extInfo":"{\"contentType\":0,\"app_type\":\"getUserInfoByToken\"}"}我有段日志，怎么用logstash拆分每个字段，例如ip，日志等级这些

拆分

2025-11-14 13:48:32.835 -> 时间戳
http-nio-8085-exec-30 -> 线程名
INFO -> 日志级别
regexTypeLogger.regexPrintLog:51 -> Logger名称和行号
INTERFACE -> 日志类型
fe80:0:0:0:824:feff:fecd:cc53%eth0 -> ipv6的host地址
10.xx.xx.xx -> 远程ip
yun-user -> 系统
yun-user-field -> 服务
/auth/getUserInfoByToken -> 请求url
0b704643-f125-4db2-b62c-7d0404fbc963 -> trace ID
`` -> 空字段
f25187aa46064522b8bf69fa59f50b6d -> span ID
178xxxxxx -> 电话号码
PHONE -> 号码类型
200 -> HTTP状态码
1763099312815 -> 开始时间戳（Unix毫秒）
1763099312835 -> 结束时间戳（Unix毫秒）
20 -> 持续时间（毫秒）
`` -> 空字段
`` -> 空字段
`` -> 空字段
`` -> 空字段
`` -> 空字段
`` -> 空字段
`` -> 空字段
NONE -> 某个标志或值
0000 -> 错误码
`` -> 空字段
`` -> 空字段
`` -> 空字段
`` -> 空字段
89127dcc-eac4-4906-be67-b2de3e16aad1 -> 另一个ID
{"input":...} -> JSON 载荷。

4、logstash.conf配置

4.1 调试日志

input {
  file {
    #测试监听es的日志
    path => "/root/test.log"
    #从文件末尾开始，只读取新增的内容（适合已经运行的日志）
    start_position => "beginning"
    sincedb_path => "/data/logstash/data/sincedb"
  }
}
filter {

}
output {
        #输出调试日志，查看结构
        stdout {
                codec => rubydebug
        }
}

启动后的日志输出

4.2 grok表达式

时间戳

2025-11-14 13:48:32.835是一个标准的 ISO8601 时间戳格式,grok内置

#表达式：%{TIMESTAMP_ISO8601}

分隔符|

#表达式： \s*\|\s*

\s* 匹配0个或多个空白字符

\\| 为|本身，\作为转义符

日志级别

INFO日志级别，内置模式LOGLEVEL
#表达式 %{LOGLEVEL:log_level}

线程名

没有非常标准的内置模式，但通常由字母、数字(.、-、_)组成，可以自定义成(?<thread>[\w\.-]+)
#表达式(?<thread>[\w\.-]+)

logger类名

通常是 Java 包名和类名，可以用 (?<logger>[a-zA-Z0-9.]+) 匹配
#表达式(?<logger>[a-zA-Z0-9.]+)

日志消息主体

这是最灵活的部分，可能包含任何字符，包括我们用作分隔符的 |。对于最后一个字段，我们应该使用 GREEDYDATA，它会匹配行尾的所有剩余内容。
#表达式：%{GREEDYDATA:message}

最终得出如下表达式

%{TIMESTAMP_ISO8601:timestamp}\s*\|\s*%{LOGLEVEL:log_level}\s*\|\s*(?<thread>[\w\.-]+)\s*\|\s*(?<logger>[a-zA-Z0-9.:]+)\s*\|\s*(?<log_type>[a-zA-Z]+)\s*\|\s*(?<host>[\w:%-]+)\s*\|\s*(?<remote_ip>[\w.-]+)\s*\|\s*(?<system>[a-z-]+)\s*\|\s*(?<service>[a-z-]+)\s*\|\s*(?<request_url>[a-zA-Z0-9/]+)\s*\|\s*(?<trace_id>[a-z0-9-]+)\s*\|\s*\s*\|\s*(?<span_id>[a-z0-9]{32})\s*\|\s*(?<phone_number>\d+)\s*\|\s*(?<device_type>[a-zA-Z]+)\s*\|\s*(?<request_status>\d+)\s*\|\s*(?<duration>\d+)s*\|\s*%{GREEDYDATA:msg}

目录CONTENT

logstash拆分日志字段

评论区