ELK(elasticsearch+logstash+kibana)實現Java分佈式系統日誌分析架構

ELK(elasticsearch+logstash+kibana)實現Java分佈式系統日誌分析架構

轉載: https://blog.csdn.net/zhou280648461/article/details/77678140

日誌是分析線上問題的重要手段，一般咱們會把日誌輸出到控制檯或者本地文件中，排查問題時經過根據關鍵字搜索本地日誌，但愈來愈多的公司，項目開發中採用分佈式的架構，日誌會記錄到多個服務器或者文件中，分析問題時可能須要查看多個日誌文件才能定位問題，若是相關項目不是一個團隊維護時溝通成本更是直線上升。把各個系統的日誌聚合並經過關鍵字連接一個事務處理請求，是分析分佈式系統問題的有效的方式。

ELK(elasticsearch+logstash+kibana)是目前比較經常使用的日誌分析系統，包括日誌收集(logstash)，日誌存儲搜索(elasticsearch)，展現查詢(kibana)，咱們使用ELK做爲日誌的存儲分析系統並經過爲每一個請求分配requestId連接相關日誌。ELK具體結構以下圖所示:

一、安裝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環境

1.  
   export JAVA_HOME=/usr/local/java/jdk1.8.0_45
2.  
   export PATH=$PATH:$JAVA_HOME/bin
3.  
   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
編輯內容以下:

1.  
   input {
2.  
   stdin { }
3.  
   }
4.  
   output {
5.  
   stdout {
6.  
   codec => rubydebug {}
7.  
   }
8.  
   }

運行logstash測試：
bin/logstash -f test.conf
顯示

證實logstash已經啓動了，
輸入hello world

由於咱們配置內容爲，控制檯輸出日誌內容，因此顯示以上格式即爲成功。
二、安裝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
修改：

1.  
   path.data: /data/es #數據路徑
2.  
   path.logs: /data/logs/es #日誌路徑
3.  
   network.host: 本機地址 #服務器地址
4.  
   http.port: 9200 #端口

配置執行用戶和目錄：

1.  
   groupadd elsearch
2.  
   useradd elsearch -g elsearch -p elasticsearch
3.  
   chown -R elsearch:elsearch elasticsearch- 2.4.0
4.  
   mkdir /data/es
5.  
   mkdir /data/logs/es
6.  
   chown -R elsearch:elsearch /data/es
7.  
   chown -R elsearch:elsearch /data/logs/es

啓動elasticsearch:
su elsearch
bin/elasticsearch
經過瀏覽器訪問:

安裝成功.
集成logstash和elasticsearch，修改Logstash配置爲：

1.  
   input {
2.  
   stdin { }
3.  
   }
4.  
   output {
5.  
   elasticsearch {
6.  
   hosts => "elasticsearchIP:9200"
7.  
   index => "logstash-test"
8.  
   }
9.  
   stdout {
10.  
    codec => rubydebug {}
11.  
    }
12.  
    }

再次啓動logstash，並輸入任意文字：「hello elasticsearch」

經過elasticsearch搜索到了剛纔輸入的文字，集成成功。
可是經過elasticsearch的原生接口查詢和展現都不夠便捷直觀，下面咱們配置一下更方便的查詢分析工具kibana。
三、安裝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中的使用。
四、建立web工程
一個普通的maven java web工程，爲了測試分佈式系統日誌的連續性，咱們讓這個項目自調用n次，並部署2個項目，相互調用，關鍵代碼以下：

1.  
   @RequestMapping("http_client")
2.  
   @Controller
3.  
   public class HttpClientTestController {
4.  
    
5.  
   @Autowired
6.  
   private HttpClientTestBo httpClientTestBo;
7.  
    
8.  
   @RequestMapping(method = RequestMethod.POST)
9.  
   @ResponseBody
10.  
    public BaseResult doPost(@RequestBody HttpClientTestResult result) {
11.  
    HttpClientTestResult testPost = httpClientTestBo.testPost(result);
12.  
    return testPost;
13.  
    }
14.  
    }

1.  
   @Service
2.  
   public class HttpClientTestBo {
3.  
    
4.  
   private static Logger logger = LoggerFactory.getLogger(HttpClientTestBo.class);
5.  
    
6.  
   @Value("${test_http_client_url}")
7.  
   private String testHttpClientUrl;
8.  
    
9.  
   public HttpClientTestResult testPost(HttpClientTestResult result) {
10.  
    logger.info(JSONObject.toJSONString(result));
11.  
    result.setCount(result.getCount() + 1);
12.  
    if (result.getCount() <= 3) {
13.  
    Map<String, String> headerMap = new HashMap<String, String>();
14.  
    String requestId = RequestIdUtil.requestIdThreadLocal.get();
15.  
    headerMap.put(RequestIdUtil.REQUEST_ID_KEY, requestId);
16.  
    Map<String, String> paramMap = new HashMap<String, String>();
17.  
    paramMap.put( "status", result.getStatus() + "");
18.  
    paramMap.put( "errorCode", result.getErrorCode());
19.  
    paramMap.put( "message", result.getMessage());
20.  
    paramMap.put( "count", result.getCount() + "");
21.  
    String resultString = JsonHttpClientUtil.post(testHttpClientUrl, headerMap, paramMap, "UTF-8");
22.  
    logger.info(resultString);
23.  
    }
24.  
     
25.  
    logger.info(JSONObject.toJSONString(result));
26.  
    return result;
27.  
    }
28.  
    }

爲了表示調用的連接性咱們在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>

1.  
   public class RequestIdFilter implements Filter {
2.  
   private static final Logger logger = LoggerFactory.getLogger(RequestIdFilter.class);
3.  
    
4.  
   /* (non-Javadoc)
5.  
   * @see javax.servlet.Filter#init(javax.servlet.FilterConfig)
6.  
   */
7.  
   public void init(FilterConfig filterConfig) throws ServletException {
8.  
   logger.info( "RequestIdFilter init");
9.  
   }
10.  
     
11.  
    /* (non-Javadoc)
12.  
    * @see javax.servlet.Filter#doFilter(javax.servlet.ServletRequest, javax.servlet.ServletResponse, javax.servlet.FilterChain)
13.  
    */
14.  
    public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException,
15.  
    ServletException {
16.  
    String requestId = RequestIdUtil.getRequestId((HttpServletRequest) request);
17.  
    MDC.put( "requestId", requestId);
18.  
    chain.doFilter(request, response);
19.  
    RequestIdUtil.requestIdThreadLocal.remove();
20.  
    MDC.remove( "requestId");
21.  
    }
22.  
     
23.  
    /* (non-Javadoc)
24.  
    * @see javax.servlet.Filter#destroy()
25.  
    */
26.  
    public void destroy() {
27.  
     
28.  
    }
29.  
    }

1.  
   public class RequestIdUtil {
2.  
   public static final String REQUEST_ID_KEY = "requestId";
3.  
   public static ThreadLocal<String> requestIdThreadLocal = new ThreadLocal<String>();
4.  
    
5.  
   private static final Logger logger = LoggerFactory.getLogger(RequestIdUtil.class);
6.  
    
7.  
   /**
8.  
   * 獲取requestId
9.  
   * @Title getRequestId
10.  
    * @Description TODO
11.  
    * @return
12.  
    *
13.  
    * @author sunhaojie 3113751575@qq.com
14.  
    * @date 2016年8月31日 上午7:58:28
15.  
    */
16.  
    public static String getRequestId(HttpServletRequest request) {
17.  
    String requestId = null;
18.  
    String parameterRequestId = request.getParameter(REQUEST_ID_KEY);
19.  
    String headerRequestId = request.getHeader(REQUEST_ID_KEY);
20.  
     
21.  
    if (parameterRequestId == null && headerRequestId == null) {
22.  
    logger.info( "request parameter 和header 都沒有requestId入參");
23.  
    requestId = UUID.randomUUID().toString();
24.  
    } else {
25.  
    requestId = parameterRequestId != null ? parameterRequestId : headerRequestId;
26.  
    }
27.  
     
28.  
    requestIdThreadLocal.set(requestId);
29.  
     
30.  
    return requestId;
31.  
    }
32.  
    }

咱們使使用了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調用連接使項目相互調用。

五、修改logstash讀取項目輸出日誌：
新增stdin.conf，內容以下:

1.  
   input {
2.  
   file {
3.  
   path => [ "/data/logs/java-base-web1/java-base-web.log", "/data/logs/java-base-web2/java-base-web.log"]
4.  
   type => "logs"
5.  
   start_position => "beginning"
6.  
   codec => multiline {
7.  
   pattern => "^\[\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}"
8.  
   negate => true
9.  
   what => "next"
10.  
    }
11.  
    }
12.  
    }
13.  
    filter{
14.  
    mutate{
15.  
    split=>[ "message","^|^"]
16.  
    add_field => {
17.  
    "messageJson" => "{datetime:%{[message][0]}, requestId:%{[message][1]},level:%{[message][2]}, class:%{[message][3]}, content:%{[message][4]}}"
18.  
    }
19.  
    remove_field => [ "message"]
20.  
    }
21.  
     
22.  
    }
23.  
    output {
24.  
    elasticsearch {
25.  
    hosts => "10.160.110.48:9200"
26.  
    index => "logstash-${type}"
27.  
    }
28.  
    stdout {
29.  
    codec => rubydebug {}
30.  
    }
31.  
    }

其中path爲日誌文件地址；codec => multiline爲處理Exception日誌，使換行的異常內容和異常頭分割在同一個日誌中；filter爲日誌內容切分，把日誌內容作爲json格式，方便查詢分析；

測試一下：

使用POSTMan模擬調用，提示服務器端異常：
經過界面搜索」調用接口異常」，共兩條數據。

使用其中一條數據的requestId搜索，展現出了請求再系統中和系統間的執行過程，方便了咱們排查錯誤。

到這裏咱們實驗了使用elk配置日誌分析，其中不少細節須要更好的處理，歡迎更多的同窗交流學習。

 

轉載請註明：孫豪傑的博客 » ELK(elasticsearch+logstash+kibana)實現Java分佈式系統日誌分析架構

 

做者：孫豪傑 http://www.sunhaojie.com