一、概述

ELK 已經成為目前最流行的集中式日志解決方案，它主要是由 Beats、Logstash、Elasticsearch、Kibana 等組件組成，來共同完成實時日志的收集，存儲，展示等一站式的解決方案。本文將會介紹 ELK 常見的架構以及相關問題解決。

Filebeat：Filebeat 是一款輕量級，占用服務資源非常少的數據收集引擎，它是 ELK 家族的新成員，可以代替 Logstash 作為在應用服務器端的日志收集引擎，支持將收集到的數據輸出到 Kafka，Redis 等隊列。

Logstash：數據收集引擎，相較于 Filebeat 比較重量級，但它集成了大量的插件，支持豐富的數據源收集，對收集的數據可以過濾，分析，格式化日志格式。

Elasticsearch：分布式數據搜索引擎，基于 Apache Lucene 實現，可集群，提供數據的集中式存儲，分析，以及強大的數據搜索和聚合功能。

Kibana：數據的可視化平臺，通過該 web 平臺可以實時的查看 Elasticsearch 中的相關數據，并提供了豐富的圖表統計功能。

二、ELK 常見部署架構

2.1、Logstash 作為日志收集器

這種架構是比較原始的部署架構，在各應用服務器端分別部署一個 Logstash 組件，作為日志收集器，然后將 Logstash 收集到的數據過濾、分析、格式化處理后發送至 Elasticsearch 存儲，最后使用 Kibana 進行可視化展示，這種架構不足的是：Logstash 比較耗服務器資源，所以會增加應用服務器端的負載壓力。

2.2、Filebeat 作為日志收集器

該架構與第一種架構唯一不同的是：應用端日志收集器換成了 Filebeat，Filebeat 輕量，占用服務器資源少，所以使用 Filebeat 作為應用服務器端的日志收集器，一般 Filebeat 會配合 Logstash 一起使用，這種部署方式也是目前最常用的架構。

2.3、引入緩存隊列的部署架構

該架構在第二種架構的基礎上引入了 Kafka 消息隊列（還可以是其他消息隊列），將 Filebeat 收集到的數據發送至 Kafka，然后在通過 Logstasth 讀取 Kafka 中的數據，這種架構主要是解決大數據量下的日志收集方案，使用緩存隊列主要是解決數據安全與均衡 Logstash 與 Elasticsearch 負載壓力。

2.4、以上三種架構的總結

第一種部署架構由于資源占用問題，現已很少使用，目前使用最多的是第二種部署架構，至于第三種部署架構個人覺得沒有必要引入消息隊列，除非有其他需求，因為在數據量較大的情況下，Filebeat 使用壓力敏感協議向 Logstash 或 Elasticsearch 發送數據。如果 Logstash 正在繁忙地處理數據，它會告知 Filebeat 減慢讀取速度。擁塞解決后，Filebeat 將恢復初始速度并繼續發送數據。

三、問題及解決方案

問題：如何實現日志的多行合并功能？

系統應用中的日志一般都是以特定格式進行打印的，屬于同一條日志的數據可能分多行進行打印，那么在使用 ELK 收集日志的時候就需要將屬于同一條日志的多行數據進行合并。

解決方案：使用 Filebeat 或 Logstash 中的 multiline 多行合并插件來實現

在使用 multiline 多行合并插件的時候需要注意，不同的 ELK 部署架構可能 multiline 的使用方式也不同，如果是本文的第一種部署架構，那么 multiline 需要在 Logstash 中配置使用，如果是第二種部署架構，那么 multiline 需要在 Filebeat 中配置使用，無需再在 Logstash 中配置 multiline。

1、multiline 在 Filebeat 中的配置方式：

filebeat.prospectors:
-
paths:
-/home/project/elk/logs/test.log
input_type:log
multiline:
pattern:'^['
negate:true
match:after
output:
logstash:
hosts:["localhost:5044"]

pattern：正則表達式

negate：默認為 false，表示匹配 pattern 的行合并到上一行；true 表示不匹配 pattern 的行合并到上一行

match：after 表示合并到上一行的末尾，before 表示合并到上一行的行首

如：

pattern:'['
negate:true
match:after

該配置表示將不匹配 pattern 模式的行合并到上一行的末尾

2、multiline 在 Logstash 中的配置方式

input{
beats{
port=>5044
}
}

filter{
multiline{
pattern=>"%{LOGLEVEL}s*]"
negate=>true
what=>"previous"
}
}

output{
elasticsearch{
hosts=>"localhost:9200"
}
}

（1）Logstash 中配置的 what 屬性值為 previous，相當于 Filebeat 中的 after，Logstash 中配置的 what 屬性值為 next，相當于 Filebeat 中的 before。(2）pattern => "%{LOGLEVEL}s*]" 中的 LOGLEVEL 是 Logstash 預制的正則匹配模式，預制的還有好多常用的正則匹配模式，詳細請看：https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns

問題：如何將 Kibana 中顯示日志的時間字段替換為日志信息中的時間？

默認情況下，我們在 Kibana 中查看的時間字段與日志信息中的時間不一致，因為默認的時間字段值是日志收集時的當前時間，所以需要將該字段的時間替換為日志信息中的時間。

解決方案：使用 grok 分詞插件與 date 時間格式化插件來實現

在 Logstash 的配置文件的過濾器中配置 grok 分詞插件與 date 時間格式化插件，如：

input{
beats{
port=>5044
}
}

filter{
multiline{
pattern=>"%{LOGLEVEL}s*][%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME}]"
negate=>true
what=>"previous"
}

grok{
match=>["message","(?%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME})"]
}

date{
match=>["customer_time","yyyyMMddHHss,SSS"]//格式化時間
target=>"@timestamp"http://替換默認的時間字段
}
}

output{
elasticsearch{
hosts=>"localhost:9200"
}
}

如,要匹配的日志格式為：“[DEBUG][20170811 1031,359][DefaultBeanDefinitionDocumentReader:106] Loading bean definitions”，解析出該日志的時間字段的方式有：

通過引入寫好的表達式文件，如表達式文件為 customer_patterns，內容為：CUSTOMER_TIME %{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME} 注：內容格式為：[自定義表達式名稱] [正則表達式] 然后 logstash 中就可以這樣引用：

filter{
grok{
patterns_dir=>["./customer-patterms/mypatterns"]//引用表達式文件路徑
match=>["message","%{CUSTOMER_TIME:customer_time}"]//使用自定義的grok表達式
}
}

以配置項的方式，規則為：(?< 自定義表達式名稱> 正則匹配規則)，如：

filter{
grok{
match=>["message","(?%{YEAR}%{MONTHNUM}%{MONTHDAY}s+%{TIME})"]
}
}

問題：如何在 Kibana 中通過選擇不同的系統日志模塊來查看數據

一般在 Kibana 中顯示的日志數據混合了來自不同系統模塊的數據，那么如何來選擇或者過濾只查看指定的系統模塊的日志數據？

解決方案：新增標識不同系統模塊的字段或根據不同系統模塊建 ES 索引

1、新增標識不同系統模塊的字段，然后在 Kibana 中可以根據該字段來過濾查詢不同模塊的數據 這里以第二種部署架構講解，在 Filebeat 中的配置內容為：

filebeat.prospectors:
-
paths:
-/home/project/elk/logs/account.log
input_type:log
multiline:
pattern:'^['
negate:true
match:after
fields://新增log_from字段
log_from:account

-
paths:
-/home/project/elk/logs/customer.log
input_type:log
multiline:
pattern:'^['
negate:true
match:after
fields:
log_from:customer
output:
logstash:
hosts:["localhost:5044"]

通過新增：log_from 字段來標識不同的系統模塊日志

2、根據不同的系統模塊配置對應的 ES 索引，然后在 Kibana 中創建對應的索引模式匹配，即可在頁面通過索引模式下拉框選擇不同的系統模塊數據。這里以第二種部署架構講解，分為兩步：① 在 Filebeat 中的配置內容為：

filebeat.prospectors:
-
paths:
-/home/project/elk/logs/account.log
input_type:log
multiline:
pattern:'^['
negate:true
match:after
document_type:account

-
paths:
-/home/project/elk/logs/customer.log
input_type:log
multiline:
pattern:'^['
negate:true
match:after
document_type:customer
output:
logstash:
hosts:["localhost:5044"]

通過 document_type 來標識不同系統模塊

② 修改 Logstash 中 output 的配置內容為：

output{
elasticsearch{
hosts=>"localhost:9200"
index=>"%{type}"
}
}

在 output 中增加 index 屬性，%{type} 表示按不同的 document_type 值建 ES 索引

四、總結

本文主要介紹了 ELK 實時日志分析的三種部署架構，以及不同架構所能解決的問題，這三種架構中第二種部署方式是時下最流行也是最常用的部署方式，最后介紹了 ELK 作在日志分析中的一些問題與解決方案，說在最后，ELK 不僅僅可以用來作為分布式日志數據集中式查詢和管理，還可以用來作為項目應用以及服務器資源監控等場景，更多內容請看官網。