相關文章：
ELK 使用步驟：Spring Boot 日志輸出到指定目錄，Filebeat 進行采集，Logstash 進行過濾，Elasticsearch 進行存儲，Kibana 進行展示。
Filebeat 示例配置（vi /etc/filebeat/filebeat.yml）：
filebeat.prospectors:
- input_type: log
paths:
- /var/log/spring-boot-log4j2/*.log
document_type: "spring-boot-log4j2" # 定義寫入 ES 時的 _type 值
multiline:
#pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})' # 指定匹配的表達式（匹配以 2017-11-15 08:04:23:889 時間格式開頭的字符串）
pattern: '^\s*("{)' # 指定匹配的表達式（匹配以 "{ 開頭的字符串）
negate: true # 是否匹配到
match: after # 合並到上一行的末尾
max_lines: 1000 # 最大的行數
timeout: 30s # 如果在規定的時候沒有新的日志事件就不等待后面的日志
fields:
logsource: node1
logtype: spring-boot-log4j2
- input_type: log
paths:
- /var/log/messages
#- /var/log/*.log
document_type: "syslog" # 定義寫入 ES 時的 _type 值
fields:
logsource: node1
logtype: syslog
#output.elasticsearch:
#hosts: ["node1:9200"]
output.logstash:
hosts: ["node1:10515"]
上面的配置需要注意幾點：
pattern：配置的正則表達式，是為了合並異常信息（而不是單行顯示），匹配以"{開頭的字符串（判斷是否 Json 格式），如果匹配不到的話，就進行合並行。document_type：配置的是 Elasticsearch 的 Type 值，方便 Elasticsearch 對日志數據的歸類。logtype：新增的字段，用於 Filebeat 和 Logstash 之間傳遞參數，進行過濾的判斷邏輯。
Logstash 示例配置（vi /etc/logstash/conf.d/logstash.conf）：
input {
beats {
port => 10515
}
}
filter {
if [fields][logtype] == "syslog" {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
syslog_pri { }
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
}
if [fields][logtype] == "spring-boot-log4j2" {
json {
source => "message"
target => "data"
}
}
}
output {
if [fields][logtype] == "spring-boot-log4j2"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "spring-boot-log4j2-%{+YYYY.MM.dd}"
}
}
if [fields][logtype] == "syslog"{
elasticsearch {
hosts => ["127.0.0.1:9200"]
index => "filebeat-%{+YYYY.MM.dd}"
}
}
}
上面的配置需要注意幾點：
- logstash.conf：配置文件可以配置多個，
input、filter和output可以單獨文件配置。 - fields logtype：就是上面 Filebeat 配置的字段，這邊用來判斷服務來源，然后進行單獨的處理。
- filter：過濾器做了兩件事，一個是使用
grok插件，匹配數據和增加字段值，另一個就是使用json插件，將字符串轉換成 Json 對象（會創建data層級結構，如果不想新建層級的話，刪掉target配置即可）。 - output：根據
logtype判斷，輸出到指定的 Elasticsearch 地址，以及創建指定的索引。
簡單總結下， Filebeat 是客戶端，一般部署在 Service 所在服務器（有多少服務器，就有多少 Filebeat），不同 Service 配置不同的input_type（也可以配置一個），采集的數據源可以配置多個，然后 Filebeat 將采集的日志數據，傳輸到指定的 Logstash 進行過濾，最后將處理好的日志數據，存儲到指定的 Elasticsearch。
好了，下面我們測試下上面的配置，是否可行。
Logstash 和 Filebeat 配置好之后，重啟一下：
$ systemctl restart logstash &&
systemctl restart filebeat
Spring Boot 中log4j2.xml中的配置（參考文章）：
<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="OFF" monitorInterval="30">
<Properties>
<Property name="LOG_PATTERN">%m%n%ex</Property>
<Property name="LOG_FILE_PATH">/Users/xishuai/Downloads/logs</Property>
</Properties>
<Appenders>
<Console name="ConsoleAppender" target="SYSTEM_OUT" follow="true">
<PatternLayout pattern="${LOG_PATTERN}"/>
</Console>
<RollingFile name="FileAppender" fileName="${LOG_FILE_PATH}/spring-boot-log4j2-demo.log"
filePattern="${LOG_FILE_PATH}/spring-boot-log4j2-demo-%d{yyyy-MM-dd}-%i.log">
<PatternLayout>
<Pattern>${LOG_PATTERN}</Pattern>
</PatternLayout>
<Filters>
<ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY" />
</Filters>
<Policies>
<SizeBasedTriggeringPolicy size="10MB" />
<TimeBasedTriggeringPolicy interval="1" />
</Policies>
<DefaultRolloverStrategy max="10"/>
</RollingFile>
</Appenders>
<Loggers>
<Root level="ERROR">
<AppenderRef ref="ConsoleAppender" />
<AppenderRef ref="FileAppender"/>
</Root>
</Loggers>
</Configuration>
配置說明可以參考之前的文章，這邊的LOG_PATTERN配置改為了%m%n%ex，直接輸出日志信息或異常信息。
測試代碼：
@Log4j2
@RestController
@EnableDiscoveryClient
@SpringBootApplication
public class SpringBootLog4j2Application implements ApplicationRunner {
public static void main(String[] args) {
SpringApplication.run(SpringBootLog4j2Application.class, args);
}
@Override
public void run(ApplicationArguments applicationArguments) throws Exception {
logger.debug("Debugging log");
logger.info("Info log");
logger.warn("Hey, This is a warning!");
logger.error("jack! We have an Error. OK");
logger.fatal("xishuai! Fatal error. Please fix me.");
}
@RequestMapping("/log")
public String log() {
log.error("{\"msg\":\"出現一個異常錯誤：請求連接失敗\",\"level\":\"ERROR\",\"createTime\":\"2018-5-21 20:22:22\",\"provider\":\"xishuai\",\"ip\":\"192.168.1.11\",\"stackTrace\":\"java.lang.Exception\\n\\tat com.example.log_demo.LogDemoTests.logCustomField(LogDemoTests.java:33)\\n\\tat org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:191)\\n\\tat org.junit.runner.JUnitCore.run(JUnitCore.java:137)\\n\\tat com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)\\n\\tat com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)\\n\\tat com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)\\n\\tat com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)\\n\",\"tag\":\"\",\"url\":\"\"}");
return "Hello World ----spring-boot-log4j2";
}
}
啟動服務，然后訪問http://localhost:8280/log，手動產生一條日志數據（Json 格式）。
查看下 Elasticsearch 中產生的索引：
[root@node1 ~]# curl http://node1:9200/_cat/indices?v
health status index uuid pri rep docs.count docs.deleted store.size pri.store.size
yellow open .kibana _8JIwyIZTJ-y9p3_o04tdA 1 1 2 0 28.3kb 28.3kb
yellow open spring-boot-log4j2-2018.05.21 IZuVdDfjRIGXR5Rei7-gFQ 5 1 2 0 29.1kb 29.1kb
可以看到，有一條新的索引spring-boot-log4j2-2018.05.21產生。
接着，我們使用 Kibana，創建一個索引模版（``spring-boot-log4j2-*）：
然后，我們就可以看到日志信息了：
第一條日志數據中，紅框里面是我們輸出的日志信息，測試代碼中總共輸出了 5 種日志級別的信息，因為配置文件中設置的日志級別是ERROR，所以這邊ERROR以下的日志不會輸出，另外，因為設置了日志匹配規則，兩條數據都不是以"{開頭，這邊就將兩條日志數據，合並成一條了。
我們再看一下自定義輸出的日志信息（Json 格式）：
測試代碼中輸出的是 Json 字符串，經過 Logstash 過濾處理之后，就轉換成 Json對象了。
另外，我們還可以data.level:ERROR這樣格式進行搜索，或者data.msg:(錯誤)格式進行模糊搜索。
還需要注意的是，上面多行合並的配置是在 Filebeat 中，如果每個服務都是一樣的規則，那么每台服務器都需要配置，如果規則更改了，這樣每台服務器的 Filebat 配置都需要更改，就比較不方便。
Logstash 也提供了多行合並的配置功能，我們只需要這樣配置（使用codec/multiline）：
input {
beats {
port => 10515
codec => multiline {
pattern => '^\s*({")'
negate => true
what => "previous"
}
}
}
效果和 Filebeat 配置是一樣的。
如果日志組件使用 Log4j/Log4j2，Logstash 還提供了另一種處理 Log4j 的方式：input/log4j。與codec/multiline不同，這個插件是直接調用了org.apache.log4j.spi.LoggingEvent處理 TCP 端口接收的數據。
參考資料：