Elk + filebeat log parsing, log warehousing optimization, logstash filter configuration attribute

Preface

Some time ago, I did not build a set of ELK Log analysis system , Then the log is through beats Read landing log , Push to logstash, And then again from logstash Pushed to the elasticsearch The index library , Last adopt kibana Visual tools are used to analyze and view logs , See... For the construction process Springboot/Springcloud Integrate ELK platform ,（Filebeat The way ） Log collection and management （Elasticsearch+Logstash+Filebeat+Kibana）

Here is the graph kibana The result of the presentation , Here we find a lot of useless and duplicate fields , And my log fields have to be expanded to see , This part can certainly be optimized . And then my Springboot/Springcloud Integrate ELK platform ,（Filebeat The way ） Log collection and management （Elasticsearch+Logstash+Filebeat+Kibana） There was also a friend in the comment area of this article who asked me beats It can be pushed directly to elasticsearch, Why go through logstash？

Before this little friend comments , I have optimized the log , This article records the process of parsing the optimization log ！

Optimization results

Realization

 First , We know logstash There are many filter plug-ins 

plug-in unit	explain
date	Date resolution
grok	Regular matching parsing
dissect	Separator resolution
mutate	Process fields , Such as renaming 、 Delete 、 Replace
json	according to json Parse the field contents into the specified field
geoip	Add geographic location data
ruby	utilize ruby Code to dynamically modify Logstash Event

Here I refer to the examples of these two articles , Then to my logstash Modify the log
Logstash filter Use
logstash Configuration details

• This is my output log

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

• This is me log4j Defined format , Here I am logstash The configuration file is parsed according to this format , See... For log integration What is a log facade ？ SpringBoot Integrate log4j2 , Log landing

# log4j2 
%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight{%-5level} [%thread] %style{%logger{36}}{cyan} : %msg%n
# logstash Resolution in configuration 
%{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 {
    
  // Parsing log 
  grok{
    
    match => {
     "message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{JAVALOGMESSAGE:thread} %{JAVALOGMESSAGE:style} : %{JAVALOGMESSAGE:msg}" }
  }
  // Replace the warehousing time with the time in our log 
  date {
    
    match => [ "timestamp" , "yyyy-MM-dd HH:mm:ss,S", "ISO8601" ]
  } 
  // Delete redundant fields 
  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

Add the configuration
multiline:
pattern: ‘^\s*(\d{4}|\d{2})-(\d{2}|[a-zA-Z]{3})-(\d{2}|\d{4})’ accord with java Log line breaking rules

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

After these configuration changes are completed , restart filebeat、logstash

If you have any questions, please feel free to contact us by private letter ！
Originality is not easy. , If it helps you, please give me a compliment before you leave ！ thank ！