ELK 架構之 Logstash 和 Filebeat 配置使用（採集過濾）

相關文章：

• ELK 架構之 Elasticsearch 和 Kibana 安裝配置
• ELK 架構之 Logstash 和 Filebeat 安裝配置

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 埠接收的數據。

參考資料：

• ELK 使用 - 多個日誌源 filebeat 配置詳解
• filebeat和logstash收集處理java多行日誌
• JSON 編解碼（官網推薦）
• 合併多行數據(Multiline)（官網推薦）