前言

前段时间不是搭建了一套ELK日志分析系统嘛，然后日志是通过beats读取落地日志，推送给logstash，然后再由logstash推送到elasticsearch索引库，最后通过kibana可视化工具进行日志的分析查看，搭建过程详见Springboot/Springcloud整合ELK平台，（Filebeat方式）日志采集及管理（Elasticsearch+Logstash+Filebeat+Kibana）

下面这张图是kibana展示的结果，这里我们发现有好多无用和重复的字段，并且我的日志字段还得展开才能看到，这部分肯定可以优化。然后我的Springboot/Springcloud整合ELK平台，（Filebeat方式）日志采集及管理（Elasticsearch+Logstash+Filebeat+Kibana）这篇文章评论区也有个小伙伴问我beats可以直接推送到elasticsearch，为什么还要通过logstash？

在这个小伙伴评论之前，我就已经优化好了日志， 这篇文章记录一下我解析优化日志的过程！

优化结果

实现

首先，我们知道logstash有很多的过滤插件

插件	说明
date	日期解析
grok	正则匹配解析
dissect	分隔符解析
mutate	对字段作处理，比如重命名、删除、替换
json	按照json解析字段内容到指定字段中
geoip	增加地理位置数据
ruby	利用ruby代码来动态修改Logstash Event

这里我参考了这两篇文章的示例，之后对我的logstash日志进行修改
Logstash filter 的使用
logstash配置详解

• 这是我的输出日志

2022-06-10 11:00:47.974 ERROR [main] com.alibaba.nacos.client.config.http.ServerHttpAgent : [NACOS SocketTimeoutException httpGet] currentServerAddr:http://127.0.0.1:8848， err : connect timed out

• 这是我log4j定义的格式，这里我在logstash配置文件中根据这个格式进行解析，日志整合详见什么是日志门面？ SpringBoot整合log4j2 ，日志落地

# log4j2 
%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight{%-5level} [%thread] %style{%logger{36}}{cyan} : %msg%n
# logstash配置中解析
%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{JAVALOGMESSAGE:thread} %{JAVALOGMESSAGE:style} : %{JAVALOGMESSAGE:msg}

logstash

input {
  beats {
    port => 5044
    type => "logs"
  }
  tcp {
    mode => "server"
    host => "127.0.0.1"
    port => 4560
    codec => json_lines
  }
}
filter {
  //解析日志
  grok{
    match => { "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{JAVALOGMESSAGE:thread} %{JAVALOGMESSAGE:style} : %{JAVALOGMESSAGE:msg}" }
  }
  //将入库时间替换成我们日志中的时间
  date {
    match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,S", "ISO8601" ]
  } 
  //删除多余字段
  mutate { 
    remove_field => "agent"
    remove_field => "ecs" 
    remove_field => "@version"
    remove_field => "host"    
    remove_field => "path"
    remove_field => "log"
    remove_field => "message"
  }
}
output {
  elasticsearch {
	hosts => ["http://127.0.0.1:9200"]
    user => "elastic"
    passwrod => "123456"
	index => "%{[fields][servicename]}-%{+yyyy.MM.dd}"
  }
}

filebeat

添加配置
multiline:
pattern: ‘^\s*(\d{4}|\d{2})-(\d{2}|[a-zA-Z]{3})-(\d{2}|\d{4})’ 符合java日志换行规则

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - E:\ideaProject\SpringCloudAlibaba2022\logs\order-service\info.log
    #- c:\programdata\elasticsearch\logs\*
  fields:
    servicename: order-service
  multiline:
    pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'
    negate: true
    match: after
    timeout: 5s
- type: log
  enabled: true
  paths:
    - E:\ideaProject\SpringCloudAlibaba2022\logs\user-service\info.log
    #- c:\programdata\elasticsearch\logs\*
  fields:
    servicename: user-service
  multiline:
    pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'
    negate: true
    match: after
    timeout: 5s

完成以上这些配置修改之后，重启filebeat、logstash

有任何问题随时私信联系！
原创不易，如果有帮助到你的话点个赞再走吧！感谢！