filebeat和logstash收集處理java多行日志

java的異常日志通常是多行的，使用logstash和filebeat收集的時候每行就會當成一條日志（事件），這樣是不連貫的。所以，我們需要對這種日志進行合並.

比如一個java應用產生的異常日志是這樣：

2017-11-15 08:04:23:889 ERROR com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver 173 send - 商戶回調網關--
發送HTTP異常！參數：requestNo=101201711151000062271 java.io.FileNotFoundException: http://192.168..139:8000/mrchantDemo/callback.htm
        at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1624) ~[?:1.7.0_51]
        at com.weconex.pay.commons.callback.utils.HttpClient.response(HttpClient.java:198) ~[callback-gateway-commons-2.1.0-SNAPSHOT.jar:?]
        at com.weconex.pay.commons.callback.utils.HttpClient.send(HttpClient.java:109) ~[callback-gateway-commons-2.1.0-SNAPSHOT.jar:?]
        at com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver.send(MerchantCallbackReceiver.java:165) [callback-gateway-service-2.1.0-SNAPSHOT.jar:?]
        at com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver.onMessage(MerchantCallbackReceiver.java:79) [callback-gateway-service-2.1.0-SNAPSHOT.jar:?]
        at org.springframework.jms.listener.adapter.MessageListenerAdapter.onMessage(MessageListenerAdapter.java:214) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:721) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:681) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:651) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:317) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1166) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1158) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1055) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]
        at java.lang.Thread.run(Thread.java:744) [?:1.7.0_51]

可以看到一個異常的日志是多行的，我們目的是要把有這樣的日志合並成一條

logstash做法

---

增加一個匹配java日志的partten文件，目的是匹配以 “2017-11-15 08:04:23:889” 這種時間格式開頭的日志

# vim  /usr/share/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.1.2/patterns/catalina
WORDS [a-zA-Z]{3}
CATALINAOUT ^(\s*%{YEAR}|%{MONTHDAY})-(%{MONTHNUM}|%{WORDS})-(%{MONTHDAY}|%{YEAR}) %{HOUR}:?%{MINUTE}(?::?%{SECOND}) 

 通過codec的mutiline插件將多行合並成一行

# vim /etc/logstash/conf.d/server.conf
input {
  file {
    id => "input-file"
    type => "catalina.out"
    path => ["/data/logs/catalina.out"]
    codec => multiline {             # 使用codec/multiline插件
      pattern => "%{CATALINAOUT}"    # 指定匹配的表達式
      negate => true                 # 是否匹配到
      what => "previous"             # 可選previous或next, previous是合並到匹配的上一行末尾
      max_lines => 1000              # 最大允許的行
      max_bytes => "10MiB"           # 允許的大小
      auto_flush_interval => 30      # 如果在規定時候內沒有新的日志事件就不等待后面的日志事件
    }  
  }
}

output {
  stdout {
    codec => "rubydebug"
  }

上面的意思是如果一個日志事件沒有被 {CATALINAOUT} 這個pattern匹配到則合並到上一行的末尾，最后進行測試

[root@ops38 conf.d]# logstash -rf server.conf
Sending Logstash's logs to /var/log/logstash which is now configured via log4j2.properties
{
          "path" => "/data/logs/catalina.out",
    "@timestamp" => 2017-12-03T12:53:32.687Z,
      "@version" => "1",
          "host" => "ops38",
       "message" => "2017-11-15 08:04:23:889 ERROR com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver 173 send - 商戶回調網關--\n發送HTTP異常！參數：requestNo=101201711151000062271 java.io.FileNotFoundException: http://222.143.53.139:8000/MerchantDemo/callback.htm\n        at sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1624) ~[?:1.7.0_51]\n        at com.weconex.pay.commons.callback.utils.HttpClient.response(HttpClient.java:198) ~[callback-gateway-commons-2.1.0-SNAPSHOT.jar:?]\n        at com.weconex.pay.commons.callback.utils.HttpClient.send(HttpClient.java:109) ~[callback-gateway-commons-2.1.0-SNAPSHOT.jar:?]\n        at com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver.send(MerchantCallbackReceiver.java:165) [callback-gateway-service-2.1.0-SNAPSHOT.jar:?]\n        at com.weconex.pay.callback.gateway.service.mq.receive.MerchantCallbackReceiver.onMessage(MerchantCallbackReceiver.java:79) [callback-gateway-service-2.1.0-SNAPSHOT.jar:?]\n        at org.springframework.jms.listener.adapter.MessageListenerAdapter.onMessage(MessageListenerAdapter.java:214) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:721) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:681) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:651) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:317) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:255) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1166) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1158) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1055) [spring-jms-4.3.4.RELEASE.jar:4.3.4.RELEASE]\n        at java.lang.Thread.run(Thread.java:744) [?:1.7.0_51]",
          "type" => "catalina.out",
          "tags" => [
        [0] "multiline"
    ]
}

解析時，最后一行不會輸出。只有當再追加一條日志時，才會輸出最后一條日志。 可以指定 auto_flush_interval值，如果在auto_flush_interval時間內沒有新的日志事件就不等待后面的日志

filebeat做法

---

 修改filebeat的配置文件 /etc/filebeat/filebeat.yml 

filebeat.prospectors:
- input_type: log
  enable: yes
  name: "catalina-out"
  paths:
      - /data/logs/catalina.out
  multiline:
      pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表達式
      negate: true                                                       # 是否匹配到
      match: after                                                       # 合並到上一行的末尾
      max_lines: 1000                                                    # 最大的行數
      timeout: 30s                                                       # 如果在規定的時候沒有新的日志事件就不等待后面的日志
  fields:                                                                # 添加type字段
      type: "catalina-out"
  fields_under_root: true                                                # 將type變為頂級字段

上面的效果和logstash一樣， 如果沒有被pattern匹配到則合並到上一行的末尾。 

同樣解析時最后一行不會輸出。只有當再追加一條日志時，才會輸出最后一條日志，可以指定timeout時間，與auto_flush_interval效果一樣.