ELK 實作 Java 分布式系統日志分析架構

日志是分析線上問題的重要手段，通常我們會把日志輸出到控制台或者本地檔案中，排查問題時通過根據關鍵字搜尋本地日志，但越來越多的公司，項目開發中采用分布式的架構，日志會記錄到多個伺服器或者檔案中，分析問題時可能需要檢視多個日志檔案才能定位問題，如果相關項目不是一個團隊維護時溝通成本更是直線上升。把各個系統的日志聚合并通過關鍵字連結一個事務處理請求，是分析分布式系統問題的有效的方式。

ELK(elasticsearch+logstash+kibana)是目前比較常用的日志分析系統，包括日志收集(logstash)，日志存儲搜尋(elasticsearch)，展示查詢(kibana)，我們使用ELK作為日志的存儲分析系統并通過為每個請求配置設定requestId連結相關日志。ELK具體結構如下圖所示:

1、安裝logstash

logstash需要依賴jdk，安裝logstash之前先安裝java環境。

下載下傳JDK:

在oracle的官方網站下載下傳，http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

根據作業系統的版本下載下傳對應的JDK安裝包，本次實驗下載下傳的是jdk-8u101-linux-x64.tar.gz

上傳檔案到伺服器并執行：

# mkdir /usr/local/java

# tar -zxf jdk-8u45-linux-x64.tar.gz -C /usr/local/java/

配置java環境

export JAVA_HOME=/usr/local/java/jdk1.8.0_45
export PATH=$PATH:$JAVA_HOME/bin
export CLASSPATH=.:$JAVA_HOME/lib/tools.jar:$JAVA_HOME/lib/dt.jar:$CLASSPATH      

執行java -version指令，列印出java版本資訊表示JDK配置成功。

下載下傳logstash:

wget https://download.elastic.co/logstash/logstash/logstash-2.4.0.tar.gz

tar -xzvf logstash-2.4.0.tar.gz

進入安裝目錄： cd #{dir}/logstash-2.4.0

建立logstash測試配置檔案：

vim test.conf

編輯内容如下:

input {
 stdin { }
}
output {
 stdout {
 codec => rubydebug {}
 }
}      

運作logstash測試：

bin/logstash -f test.conf

顯示

證明logstash已經啟動了，

輸入hello world

因為我們配置内容為，控制台輸出日志内容，是以顯示以上格式即為成功。

2、安裝elasticsearch

下載下傳安裝包：

wget https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.0/elasticsearch-2.4.0.tar.gz

解壓并配置：

tar -xzvf elasticsearch-2.4.0.tar.gz

cd #{dir}/elasticsearch-2.4.0

vim config/elasticsearch.yml

修改：

path.data: /data/es #資料路徑
path.logs: /data/logs/es #日志路徑
network.host: 本機位址 #伺服器位址
http.port: 9200 #端口      

配置執行使用者和目錄：

groupadd elsearch
useradd elsearch -g elsearch -p elasticsearch
chown -R elsearch:elsearch elasticsearch-2.4.0
mkdir /data/es
mkdir /data/logs/es
chown -R elsearch:elsearch /data/es
chown -R elsearch:elsearch /data/logs/es      

啟動elasticsearch:

su elsearch

bin/elasticsearch

通過浏覽器通路:

安裝成功.

內建logstash和elasticsearch，修改Logstash配置為：

input {
 stdin { } 
}
output {
 elasticsearch {
 hosts => "elasticsearchIP:9200"
 index => "logstash-test"
 } 
 stdout {
 codec => rubydebug {}
 } 
}      

再次啟動logstash，并輸入任意文字：“hello elasticsearch”

通過elasticsearch搜尋到了剛才輸入的文字，內建成功。

但是通過elasticsearch的原生接口查詢和展示都不夠便捷直覺，下面我們配置一下更友善的查詢分析工具kibana。

3、安裝kibana

下載下傳安裝包：

wget https://download.elastic.co/kibana/kibana/kibana-4.6.1-linux-x86_64.tar.gz

解壓kibana，并進入解壓後的目錄

打開config/kibana.yml,修改如下内容

#啟動端口 因為端口受限 是以變更了預設端口

server.port: 8601

#啟動服務的ip

server.host: “本機ip”

#elasticsearch位址

elasticsearch.url: “http://elasticsearchIP:9200”

啟動程式:

bin/kibana

通路配置的ip:port，在discover中搜尋剛才輸入的字元，内容非常美觀的展示了出來。

到這裡我們的elk環境已經配置完成了，我們把已java web項目試驗日志在elk中的使用。

4、建立web工程

一個普通的maven java web工程，為了測試分布式系統日志的連續性，我們讓這個項目自調用n次，并部署2個項目，互相調用，關鍵代碼如下：

@RequestMapping("http_client")
@Controller
public class HttpClientTestController {
 
    @Autowired
    private HttpClientTestBo httpClientTestBo;
 
    @RequestMapping(method = RequestMethod.POST)
    @ResponseBody
    public BaseResult doPost(@RequestBody HttpClientTestResult result) {
        HttpClientTestResult testPost = httpClientTestBo.testPost(result);
        return testPost;
    }
}      

@Service
public class HttpClientTestBo {
 
    private static Logger logger = LoggerFactory.getLogger(HttpClientTestBo.class);
 
    @Value("${test_http_client_url}")
    private String testHttpClientUrl;
 
    public HttpClientTestResult testPost(HttpClientTestResult result) {
        logger.info(JSONObject.toJSONString(result));
        result.setCount(result.getCount() + 1);
        if (result.getCount() <= 3) {
            Map<String, String> headerMap = new HashMap<String, String>();
            String requestId = RequestIdUtil.requestIdThreadLocal.get();
            headerMap.put(RequestIdUtil.REQUEST_ID_KEY, requestId);
            Map<String, String> paramMap = new HashMap<String, String>();
            paramMap.put("status", result.getStatus() + "");
            paramMap.put("errorCode", result.getErrorCode());
            paramMap.put("message", result.getMessage());
            paramMap.put("count", result.getCount() + "");
            String resultString = JsonHttpClientUtil.post(testHttpClientUrl, headerMap, paramMap, "UTF-8");
            logger.info(resultString);
        }
 
        logger.info(JSONObject.toJSONString(result));
        return result;
    }
}      

為了表示調用的連結性我們在web.xml中配置requestId的filter，用于建立requestId：

<filter>
 <filter-name>requestIdFilter</filter-name>
 <filter-class>com.virxue.baseweb.utils.RequestIdFilter</filter-class>
</filter>
<filter-mapping>
 <filter-name>requestIdFilter</filter-name>
 <url-pattern>/*</url-pattern>
</filter-mapping>      

public class RequestIdFilter implements Filter {
    private static final Logger logger = LoggerFactory.getLogger(RequestIdFilter.class);
 
    /* (non-Javadoc)
     * @see javax.servlet.Filter#init(javax.servlet.FilterConfig)
     */
    public void init(FilterConfig filterConfig) throws ServletException {
        logger.info("RequestIdFilter init");
    }
 
    /* (non-Javadoc)
     * @see javax.servlet.Filter#doFilter(javax.servlet.ServletRequest, javax.servlet.ServletResponse, javax.servlet.FilterChain)
     */
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException,
        ServletException {
        String requestId = RequestIdUtil.getRequestId((HttpServletRequest) request);
        MDC.put("requestId", requestId);
        chain.doFilter(request, response);
        RequestIdUtil.requestIdThreadLocal.remove();
        MDC.remove("requestId");
    }
 
    /* (non-Javadoc)
     * @see javax.servlet.Filter#destroy()
     */
    public void destroy() {
 
    }
}      

public class RequestIdUtil {
    public static final String REQUEST_ID_KEY = "requestId";
    public static ThreadLocal<String> requestIdThreadLocal = new ThreadLocal<String>();
 
    private static final Logger logger = LoggerFactory.getLogger(RequestIdUtil.class);
 
    /**
     * 擷取requestId
     * @Title getRequestId
     * @Description TODO
     * @return
     *
     * @author sunhaojie [email protected]
     * @date 2016年8月31日 上午7:58:28
     */
    public static String getRequestId(HttpServletRequest request) {
        String requestId = null;
        String parameterRequestId = request.getParameter(REQUEST_ID_KEY);
        String headerRequestId = request.getHeader(REQUEST_ID_KEY);
 
        if (parameterRequestId == null && headerRequestId == null) {
            logger.info("request parameter 和header 都沒有requestId入參");
            requestId = UUID.randomUUID().toString();
        } else {
            requestId = parameterRequestId != null ? parameterRequestId : headerRequestId;
        }
 
        requestIdThreadLocal.set(requestId);
 
        return requestId;
    }
}      

我們使使用了Logback作為日志輸出的插件，并且使用它的MDC類，可以無侵入的在任何地方輸出requestId，具體的配置如下:

<configuration> 
 <appender name="logfile" class="ch.qos.logback.core.rolling.RollingFileAppender">
 <Encoding>UTF-8</Encoding>
 <File>${log_base}/java-base-web.log</File>
 <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
 <FileNamePattern>${log_base}/java-base-web-%d{yyyy-MM-dd}-%i.log</FileNamePattern>
 <MaxHistory>10</MaxHistory>
 <TimeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
 <MaxFileSize>200MB</MaxFileSize>
 </TimeBasedFileNamingAndTriggeringPolicy> 
 </rollingPolicy>
 <layout class="ch.qos.logback.classic.PatternLayout">
 <pattern>%d^|^%X{requestId}^|^%-5level^|^%logger{36}%M^|^%msg%n</pattern>
 </layout>
 </appender>
 <root level="info"> 
 <appender-ref ref="logfile" /> 
 </root> 
</configuration>      

這裡的日志格式使用了“^|^”做為分隔符，友善logstash進行切分。在測試伺服器部署2個web項目，并且修改日志輸出位置，并修改url調用連結使項目互相調用。

5、修改logstash讀取項目輸出日志：

新增stdin.conf，内容如下:

input {
 file {
 path => ["/data/logs/java-base-web1/java-base-web.log", "/data/logs/java-base-web2/java-base-web.log"]
 type => "logs"
 start_position => "beginning"
 codec => multiline {
 pattern => "^\[\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}"
 negate => true
 what => "next"
 } 
 } 
}
filter{
 mutate{
 split=>["message","^|^"]
 add_field => {
 "messageJson" => "{datetime:%{[message][0]}, requestId:%{[message][1]},level:%{[message][2]}, class:%{[message][3]}, content:%{[message][4]}}"
 } 
 remove_field => ["message"]
 } 
  
}
output {
 elasticsearch {
 hosts => "10.160.110.48:9200"
 index => "logstash-${type}"
 } 
 stdout {
 codec => rubydebug {}
 } 
}      

其中path為日志檔案位址；codec => multiline為處理Exception日志，使換行的異常内容和異常頭分割在同一個日志中；filter為日志内容切分，把日志内容做為json格式，友善查詢分析；

測試一下：

使用POSTMan模拟調用，提示伺服器端異常：

通過界面搜尋”調用接口異常”，共兩條資料。

使用其中一條資料的requestId搜尋，展示出了請求再系統中和系統間的執行過程，友善了我們排查錯誤。

到這裡我們實驗了使用elk配置日志分析，其中很多細節需要更好的處理，歡迎更多的同學交流學習。