Docker-Compose部署ELK

1.概述

1.1定義

對於大規模的日志，需要集中化的管理。而ELK提供了一整套解決方案，並且都是開源軟件，之間互相配合使用，完美銜接，高效的滿足了很多場合的應用。ELK是三種技術產品的簡稱，包括Elasticsearch、Logstash、Kibana，可在項目中作為日志框架使用。

1.2功能說明

Elasticsearch是個開源分布式搜索引擎，提供搜集、分析、存儲數據三大功能。

Logstash 主要是用來日志的搜集、分析、過濾日志的工具，支持大量的數據獲取方式。

Kibana 也是一個開源和免費的工具，Kibana可以為 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以幫助匯總、分析和搜索重要數據日志。

它們的作用如圖：

 簡單來說，應用服務生產日志，通過Logger產生日志並輸出；Logstash通過http接收應用服務產生的日志；Elasticsearch為日志提供全文檢索功能；kibana為Elasticsearch提供圖形化界面。

2.部署ELK

本文在Linux上部署，以/opt作為根目錄進行說明。

2.1創建目錄和文件

1）創建docker-elk目錄，在此目錄創建文件和其他目錄

mkdir /opt/docker_elk

2）創建logstash配置文件

mkdir /opt/docker_elk/logstash
touch  /opt/docker_elk/logstash/logstash.conf

3）配置logstash.conf，其內容如下

input {
  tcp {
    mode => "server"
    host => "0.0.0.0"
    port => 4560
    codec => json
  }
}
output {
  elasticsearch {
    hosts => "es:9200"
    index => "logstash-%{+YYYY.MM.dd}"
  }
}

在這里指定了輸入的日志的端口是4560，那么下面對外暴露的端口也必須是4560。

4）創建docker-compose.yml文件

touch /opt/docker_elk/docker-compose.yml

2.2配置docker-compose並啟動

打開docker-compose.yml，

cd /opt/docker_elk
vi docker-compose.yml

配置內容如下：

version: '3.7'
services:
  elasticsearch:
    image: elasticsearch:7.6.2
    container_name: elasticsearch
    privileged: true
    user: root
    environment:
      #設置集群名稱為elasticsearch
      - cluster.name=elasticsearch 
      #以單一節點模式啟動
      - discovery.type=single-node 
      #設置使用jvm內存大小
      - ES_JAVA_OPTS=-Xms512m -Xmx512m 
    volumes:
      - /opt/docker_elk/elasticsearch/plugins:/usr/share/elasticsearch/plugins
      - /opt/docker_elk/elasticsearch/data:/usr/share/elasticsearch/data
    ports:
      - 9200:9200
      - 9300:9300

  logstash:
    image: logstash:7.6.2
    container_name: logstash
    ports:
       - 4560:4560
    privileged: true
    environment:
      - TZ=Asia/Shanghai
    volumes:
      #掛載logstash的配置文件
      - /opt/docker_elk/logstash/logstash.conf:/usr/share/logstash/pipeline/logstash.conf 
    depends_on:
      - elasticsearch 
    links:
      #可以用es這個域名訪問elasticsearch服務
      - elasticsearch:es 
    

  kibana:
    image: kibana:7.6.2
    container_name: kibana
    ports:
        - 5601:5601
    privileged: true
    links:
      #可以用es這個域名訪問elasticsearch服務
      - elasticsearch:es 
    depends_on:
      - elasticsearch 
    environment:
      #設置訪問elasticsearch的地址
      - elasticsearch.hosts=http://es:9200 
   

這里使用privileged設置為true是賦予這個容器root權限。然后啟動

docker-compose up -d

在啟動時，如果Elasticsearch啟動報錯，說/usr/share/elasticsearch/data下的文件無權限，那么需要給宿主機授予讀寫權限

chmod 777 /opt/docker_elk/elasticsearch/data

若啟動報錯，需要先關閉並刪除容器后再重新啟動。關閉刪除命令：

docker-compose down

2.3打開kibana

輸入http://192.168.0.150:5601，訪問Kibana web界面。點擊左側設置，進入Management界面

 

3.收集日志

本文通過SpringBoot架構把日志信息記錄到logstash。

3.1環境准備

新建一個springboot的項目，需要導入web的依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

除此之外，需要導入logstash的依賴：

<!--集成logstash-->
<dependency>
    <groupId>net.logstash.logback</groupId>
    <artifactId>logstash-logback-encoder</artifactId>
    <version>6.6</version>
</dependency>

3.2使用logback記錄日志

logback是SpringBoot自帶的日志，只要導入了web的依賴即可使用。

1）在測試包下新建一個測試類和測試方法

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class AppTest {

    //創建日志對象
    Logger logger = LogManager.getLogger(this.getClass());

    @Test
    public void test1() {
        logger.info("logback的日志信息過來了");
        logger.error("logback的錯誤信息過來了");
    }
}

2）在需要目錄新建logback-spring.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE configuration>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>
    <include resource="org/springframework/boot/logging/logback/console-appender.xml"/>
    <!--應用名稱-->
    <property name="APP_NAME" value="springboot-logback-elk-demo"/>
    <!--日志文件保存路徑-->
    <property name="LOG_FILE_PATH" value="${LOG_FILE:-${LOG_PATH:-${LOG_TEMP:-${java.io.tmpdir:-/tmp}}}/logs}"/>
    <contextName>${APP_NAME}</contextName>
    <!--每天記錄日志到文件appender-->
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>${LOG_FILE_PATH}/${APP_NAME}-%d{yyyy-MM-dd}.log</fileNamePattern>
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <encoder>
            <pattern>${FILE_LOG_PATTERN}</pattern>
        </encoder>
    </appender>
    <!--輸出到logstash的appender-->
    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <!--可以訪問的logstash日志收集端口-->
        <destination>192.168.86.128:4560</destination>
        <encoder charset="UTF-8" class="net.logstash.logback.encoder.LogstashEncoder"/>
    </appender>
    <root level="INFO">
        <appender-ref ref="CONSOLE"/>
        <appender-ref ref="FILE"/>
        <appender-ref ref="LOGSTASH"/>
    </root>
</configuration>

3）啟動測試方法，然后點擊index Patterns后，點擊創建索引

4）點擊創建索引，

5）創建名稱為logstash-*的index，此名字是在配置文件logstash.conf中配置的。上述配置的是logstash-%{+YYYY.MM.dd}，故這里使用*代替日期

6）然后在Next Step(下一步)中選擇@timestamp的filter

 7）創建完成之后，點擊Discover，並選擇剛才創建的index

8）查看kibana的日志信息

查看信息時，建議在左側Available fields中篩選出"message"信息，"thread_name"字段可選。篩選的字段在左側也可以看到，右側看到的信息就比較清晰。

需要注意的是，在這些日志中，時間是logstash收集日志時的創建時間，並不是原始日志的記錄時間。

3.3使用log4j2記錄日志

要使用log4j2，則必須排除SpringBoot自帶的日志。

1）排除logback並導入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
    <exclusions>
        <!-- 引入log4j日志時需去掉默認的logback -->
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </exclusion>
    </exclusions>
</dependency>

<!-- 日志管理log4j2 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-log4j2</artifactId>
    <version>2.1.0.RELEASE</version>
</dependency>

2）在資源目錄下新建log4j2.xml

<?xml version="1.0" encoding="UTF-8"?>

<configuration status="info">
    <Properties>
        <!-- 聲明日志文件存儲的目錄 -->
        <Property name="LOG_HOME">E:\logs</Property>
        <Property name="LOG_PATTERN"
                  value="%date{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%thread][%class{36}:%line] - %msg%n"></Property>
    </Properties>

    <Appenders>
        <!--輸出控制台的配置-->
        <Console name="Console" target="SYSTEM_OUT">
            <!--控制台只輸出level及以上級別的信息（onMatch），其他的直接拒絕（onMismatch）-->
            <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
            <!-- 輸出日志的格式-->
            <PatternLayout pattern="${LOG_PATTERN}"/>
        </Console>

        <!--這輸出日志到文件的配置，每次大小超過size，則這size大小的日志會自動存入按年份-月份建立的文件夾下面並進行壓縮，作為存檔-->
        <RollingFile name="RollingFile" fileName="${LOG_HOME}\app_${date:yyyy-MM-dd}.log"
                     filePattern="${LOG_HOME}\${date:yyyy-MM}\app_%d{yyyy-MM-dd}_%i.log">
            <ThresholdFilter level="info" onMatch="ACCEPT" onMismatch="DENY"/>
            <!-- 輸出日志的格式-->
            <PatternLayout pattern="${LOG_PATTERN}"/>
            <!-- 日志文件大小 -->
            <SizeBasedTriggeringPolicy size="20MB"/>
            <!-- 最多保留文件數 -->
            <DefaultRolloverStrategy max="30"/>
        </RollingFile>

        <!--輸出到logstash的appender-->
        <Socket name="Socket" host="192.168.86.128" port="4560" protocol="TCP">
            <!--輸出到logstash的日志格式-->
            <PatternLayout pattern="${LOG_PATTERN}"/>
        </Socket>
    </Appenders>

    <!--然后定義Logger，只有定義了Logger並引入的Appender，Appender才會生效。Root中level配置了日志級別，可配置其他級別-->
    <Loggers>
        <Root level="info">
            <AppenderRef ref="Console"/>
            <AppenderRef ref="RollingFile"/>
            <AppenderRef ref="Socket"/>
        </Root>
    </Loggers>

</configuration>

主要的上面紅色的部分，需要指定logstash服務的ip和記錄日志的端口。　　

3）在測試類新建測試方法

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class AppTest {

    //創建日志對象
    Logger logger = LogManager.getLogger(this.getClass());

    ...

    @Test
    public void test2() {
        logger.info("我是log4j2的日志信息");
        logger.error("我是log4j2的錯誤信息");
    }
}

4）啟動測試方法，查看kibana的日志信息

查看信息時，建議在左側Available fields中篩選出"message"信息，"thread_name"字段可選。篩選的字段在左側也可以看到，右側看到的信息就比較清晰，包含了日志本身的時間，這是是配置日志配置的。

4.功能提升

4.1漢化Kibana

在使用Kibana界面時，發現都是英文的，實際上是可以配置中文版。方法如下：

1）進入容器內部

docker exec -it kibana bash

2）打開配置文件

vi /opt/kibana/config/kibana.yml

3）在配置文件最后添加一行，設置語言是中文

i18n.locale: zh-CN

需要注意的是，zhe-CN和:號之間必須有個空格，否則kibana無法啟動。

4）保存后重啟此容器即可。

4.2自定義日志過濾

對應logstash，默認是格式是使用的logback的默認格式，一旦使用log4j2或其他日志框架來指定日志的輸出格式時，那么logstash便無法解析，需要進行自定義過濾。

4.2.1情景說明

如上述章節使用log4j2配置的輸出格式進行日志收集時，定義的格式：

 它把所有的信息都記錄在message上了如下圖：

 但實際上並吧需要這么多的信息，只需具體的信息，因此需要手動對日志格式過濾。

4.2.2日志過濾

1）修改日志輸出的格式。緊接着時間的后面的一個參數必須使用中括號包裹，否則解析有問題

將log4j2.xml的日志輸出格式修改如下：

<Property name="LOG_PATTERN"
                  value="%date{yyyy-MM-dd HH:mm:ss.SSS} [%-5level] [%thread][%class{36}:%line] - %msg%n"></Property>

2）在logstash.conf中添加過濾規則：

filter {

    grok {
        match => {
            message => "%{DATA:datetime}\ \[%{DATA:level}\]\ \[%{DATA:class}\] - %{GREEDYDATA:logger}"
            remove_field =>[ "message" ]
        }
    }
    date {
        match => ["datetime", "yyyy-MM-dd HH:mm:ss.SSS"]
    }
    if "_grokparsefailure" in [tags]    {
        drop {}
    }
}

截圖如下

其中grok是根據正則來匹配日志的，data是指定上面時間的格式，if是用於將解析失敗的信息刪除。

grok的正則語法在次略，其可以在kibana界面進行驗證。打開在dev Tools，選擇Grok Debugger，輸入數據和規則點擊模擬進行驗證和解析

 3）重啟logstash后，再打印日志，把日志推送到過去，在kibana中篩選level和logger進行查看（logger是具體的信息，在配置grok時指定的），看到信息已正常解析

 4.2.3安裝插件multiline

1）情景重現

測試類：

package com.zys.example;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

@RunWith(SpringRunner.class)
@SpringBootTest
public class MyTest2 {

    //創建日志對象
    Logger logger = LogManager.getLogger(this.getClass());

    @Test
    public void test() {
        try {
            logger.info("日志信息過來了");
            int i = 1 / 0;
        } catch (Exception e) {
            logger.error("發生異常：", e);
        }
    }
}

運行時會發生異常，把異常的錯誤信息通過日志推送到logstash

但在kibana中卻看不到詳細的信息

主要原因是錯誤信息換行輸出的，對於上述配置的logstash會解析失敗，故不會顯示，因此需要安裝插件multiline將多行合並為一行輸出。

2）安裝multiline

進入logstash容器內部，查看是否已安裝multiline

logstash-plugin list

若沒有安裝過，則進行安裝

logstash-plugin install logstash-filter-multiline

安裝完成后在logstash.conf中的filter配置

multiline {
        pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}"
        negate => true
        what => "previous"
    }

截圖如下：

配置后重啟logstash，再次執行測試方法，會發現錯誤日志已正常記錄

需要注意的是，必須先安裝multiline才能重啟logstash，否則只配置logstash.conf會導致logstash無法啟動。