ELK 日志系統

Elastic 官方文檔：https://www.elastic.co/guide/index.html

elasticsearch github：https://github.com/elastic/elasticsearch

logstash github：https://github.com/elastic/logstash

kibana github：https://github.com/elastic/kibana

從 ELK 到 EFK：https://www.sohu.com/a/198596248_748431

Elasticsearch: 權威指南：https://www.elastic.co/guide/cn/elasticsearch/guide/current/index.html

ELK 快速指南：https://blog.csdn.net/bluetjs/article/details/78770447

ELK實時日志分析平台環境部署--完整記錄：http://www.cnblogs.com/kevingrace/p/5919021.html

集中式日志分析平台 - ELK Stack - 部署篇：https://www.jianshu.com/p/934c457a333c

集中式日志系統 ELK 協定棧詳解：https://www.ibm.com/developerworks/cn/opensource/os-cn-elk/

ELK+Filebeat 集中式日志解決方案詳解：https://www.ibm.com/developerworks/cn/opensource/os-cn-elk-filebeat/index.html

ELK部署詳解：https://segmentfault.com/a/1190000007728789

Centos7 之安裝Logstash ELK stack 日志管理系統：https://www.cnblogs.com/hanyifeng/p/5509985.html

中小型研發團隊架構實踐七：集中式日志ELK：https://www.cnblogs.com/supersnowyao/p/8375926.html

ELK實時日志分析平台環境部署--完整記錄：https://www.cnblogs.com/kevingrace/p/5919021.html

ELK 之Filebeat 結合Logstash 過濾出來你想要的日志：http://blog.51cto.com/seekerwolf/2110174

我的ELK搭建筆記（阿裡雲上部署）：https://www.jianshu.com/p/797073c1913f

ELK 搭建篇：https://www.cnblogs.com/yuhuLin/p/7018858.html

通過ELK快速搭建一個你可能需要的集中化日志平台：https://www.cnblogs.com/huangxincheng/p/7918722.html

Logstat 輸入插件配置：https://www.elastic.co/guide/en/logstash/current/input-plugins.html

使用logstash的logstash-input-kafka插件讀取kafka中的資料：https://blog.csdn.net/lvyuan1234/article/details/78653324

ELK Stack （系列文章）：http://www.linuxe.cn/sort/elk

項目實戰14—ELK 企業内部日志分析系統：https://www.cnblogs.com/along21/p/8509123.html

最新Centos7.6 部署ELK日志分析系統：https://segmentfault.com/a/1190000019799137

ELK日志系統淺析與部署：https://blog.csdn.net/qq_22211217/article/details/80764568

随筆分類 - ELK：https://www.cnblogs.com/workdsz/category/1257841.html

ELK 筆記：https://blog.csdn.net/ahri_j/category_7519196.html

ES基本概念與 API 操作：https://blog.csdn.net/Ahri_J/article/details/79720824

es 基本API操作使用：https://blog.csdn.net/zhangcongyi420/article/details/94362108

ELK、FILEBEAT日志分析平台搭建：https://shuwoom.com/?p=2300

ElasticSearch總結：https://www.cnblogs.com/aiqiqi/p/11451411.html

ELK 是 Elasticsearch、Logstash、Kibana的簡稱，這三者是核心套件，但并非全部。

• Elasticsearch 是實時全文搜尋和分析引擎，提供搜集、分析、存儲資料三大功能；是一套開放 REST 和 JAVA API 等結構提供高效搜尋功能，可擴充的分布式系統。它建構于 Apache Lucene 搜尋引擎庫之上。
• Logstash 是一個用來搜集、分析、過濾日志的工具。它支援幾乎任何類型的日志，包括系統日志、錯誤日志和自定義應用程式日志。它可以從許多來源接收日志，這些來源包括 syslog、消息傳遞（例如 RabbitMQ）和 JMX，它能夠以多種方式輸出資料，包括電子郵件、websockets 和 Elasticsearch。
• Kibana 是一個基于 Web 的圖形界面，用于搜尋、分析和可視化存儲在 Elasticsearch 名額中的日志資料。它利用 Elasticsearch 的 REST 接口來檢索資料，不僅允許使用者建立他們自己的資料的定制儀表闆視圖，還允許他們以特殊的方式查詢和過濾資料

項目由來

　　（1）開發人員不能登入線上伺服器檢視詳細日志，經過運維周轉費時費力

　　（2）日志資料分散在多個系統，難以查找與整合

　　（3）日志資料量巨大，查詢速度太慢，無法滿足需求

　　（4）無法全局掌控項目運作狀況

　　（5）日志資料查詢不夠實時

　　（6）資料分析人員不會寫代碼，無法分析統計資料

　　（7）.........

架構裡包含的元件

ELK 是 elastic 公司旗下三款産品 ElasticSearch 、Logstash 、Kibana 的首字母組合。

　　Logstash + Elasticsearch + Kibana（ELK）

　　Logstash：         監控，過濾，收集日志。即傳輸和處理你的日志、事務或其他資料。

　　Elasticsearch：  存儲日志，提供搜尋功能。ElasticSearch 是一個基于 Lucene 建構的開源，分布式，RESTful 搜尋引擎。

　　kibana：             提供web界面，支援查詢，統計，和圖表展現。即 将 Elasticsearch 的資料分析并渲染為可視化的報表。

　　filebeat：            輕量級的日志收集工具。

　　很多公司都采用該架構建構分布式日志系統，包括 新浪微網誌，freewheel，暢捷通等

注意：在應用端收集日志時，建議用 filebeat 。

通過 ELK 這套解決方案，可以同時實作日志收集、日志搜尋和日志分析的功能。

架構設計

如果日志量很大，Logstash 會遇到資源占用高的問題，為解決這個問題，引入了 Filebeat。Filebeat 是基于 logstash-forwarder 的源碼改造而成，用 Golang 編寫，無需依賴 Java 環境，效率高，占用記憶體和 CPU 比較少，非常适合作為 Agent 跑在伺服器上。Filebeat 可以看做是新一代的 logstash-forward，但是性能超 logstash，部署簡單，占用資源少，可以很友善的和logstash和ES對接，作為日志檔案采集元件。Filebeat 是 Beats 家族的一員，後續可以使用 Packetbeat 進行網絡資料采集、Winlogbeat 進行 Windosw事件采集、Heartbeat 進行心跳采集、Metricbeat 進行系統名額采集。這種架構解決了 Logstash 在各伺服器節點上占用系統資源高的問題。相比 Logstash，Beats 所占系統的 CPU 和記憶體幾乎可以忽略不計。另外，Beats 和 Logstash 之間支援 SSL/TLS 加密傳輸，用戶端和伺服器雙向認證，保證了通信安全。

LogStash 可以用來對日志進行收集并進行過濾整理後輸出到 ES 中，FileBeats 是一個更加輕量級的日志收集工具。現在最常用的方式是通過 FileBeats 收集目标日志，然後統一輸出到 LogStash 做進一步的過濾，在由 LogStash 輸出到 ES 中進行存儲。

　　（1）使用filebeat

　　　　架構設計1：filebeat --> logstash(parse) --> es叢集(Elasticsearch 叢集) --> kibana--ngix

　　　　　　  缺點：如果 logstash 出問題會導緻 filebeat 收集的資料丢失

　　　　架構設計2：filebeat --> logstash(parse)[loadbalance] --> es叢集 --> kibana--ngix

　　　　　　filebeat 和 logstash 耦合性太高

　　　　架構設計3：filebeat（3台）--> redis --> logstash(parse) --> es叢集 --> kibana -- ngix（可選） 

　　　　　　裡面 redis 是一個單線程的執行個體，redis 單線程每秒處理能力一般是 10W 次左右。

　　　　架構設計4：filebeat --> redis/kafka --> logstash(parse) --> es --> kibana--ngix

　　　　　　filebeat1.3 版本不支援輸出到 kafka，5.x 版本中支援輸出到 kafka

　　（2）不使用 filebeat

　　　　　　logstash --> kafka --> logstash(parse) --> es --> kibana--ngix

　　　　　　　　裡面 kafka 支援水準擴充，可以使用多分區，支援多線程并行執行。

　　　　　　　　在應用端收集日志的話，logstash 比較重量級，性能消耗比 filebeat 大

　　（3）Filebeat 用于日志收集和傳輸，相比 Logstash 更加輕量級和易部署，對系統資源開銷更小。

為什麼做日志系統 ？

首先，什麼是日志？ 日志就是程式産生的，遵循一定格式（通常包含時間戳）的文本資料

通常日志由伺服器生成，輸出到不同的檔案中，一般會有系統日志、 應用日志、安全日志。這些日志分散地存儲在不同的機器上。

通常當系統發生故障時，工程師需要登入到各個伺服器上，使用 grep、sed、awk 等 Linux 腳本工具去日志裡查找故障原因。在沒有日志系統的情況下，首先需要定位處理請求的伺服器，如果這台伺服器部署了多個執行個體，則需要去每個應用執行個體的日志目錄下去找日志檔案。每個應用執行個體還會設定日志滾動政策（如：每天生成一個檔案），還有日志壓縮歸檔政策等。

這樣一系列流程下來，對于我們排查故障以及及時找到故障原因，造成了比較大的麻煩。是以，如果我們能把這些日志集中管理，并提供集中檢索功能，不僅可以提高診斷的效率，同時對系統情況有個全面的了解，避免事後救火的被動。

日志資料在以下幾方面具有非常重要的作用：

• 資料查找：通過檢索日志資訊，定位相應的 bug ，找出解決方案
• 服務診斷：通過對日志資訊進行統計、分析，了解伺服器的負荷和服務運作狀态
• 資料分析：可以做進一步的資料分析，比如根據請求中的課程 id ，找出 TOP10 使用者感興趣課程。

針對這些問題，為了提供分布式的實時日志搜集和分析的監控系統，我們采用了業界通用的日志資料管了解決方案 - 它主要包括 Elasticsearch 、 Logstash 和 Kibana 三個系統。通常，業界把這套方案簡稱為 ELK，取三個系統的首字母，但是我們實踐之後将其進一步優化為 EFK，F代表Filebeat，用以解決 Logstash 導緻的問題。

Elasticsearch、Logstash、Kibana、Filebeat、Kafka

Elasticsearch 和 MySQL 對比：

終于有人把Elasticsearch原理講透了：https://developer.51cto.com/art/201904/594615.htm

官網位址：https://www.elastic.co

這一套軟體可以當作一個 MVC 模型：logstash 是 controller 層，Elasticsearch 是一個 model 層，kibana 是 view 層。首先将資料傳給 logstash，它将資料進行過濾和格式化（轉成 JSON 格式），然後傳給 Elasticsearch 進行存儲、建搜尋的索引，kibana提供前端的頁面再進行搜尋和圖表可視化，它是調用 Elasticsearch 的接口傳回的資料進行可視化。logstash 和 Elasticsearch 是用 Java 寫的，kibana 使用node.js 架構。

這套軟體官網有很詳細的使用說明，https://www.elastic.co/，除了docs之外，還有視訊教程。

友情提示：安裝 ELK 時，三個應用請選擇統一的版本，避免出現一些莫名其妙的問題。例如：由于版本不統一，導緻三個應用間的通訊異常。

ELK 架構

ELK 技術棧 的一個架構圖：

說明：以上是 ELK 技術棧的一個架構圖。從圖中可以清楚的看到資料流向。

• Beats (https://www.elastic.co/products/beats) 是單一用途的資料傳輸平台，它可以将多台機器的資料發送到 Logstash 或 ElasticSearch。但 Beats 并不是不可或缺的一環，是以本文中暫不介紹。

• Logstash：資料收集處理引擎。支援動态的從各種資料源搜集資料，并對資料進行過濾、分析、豐富、統一格式等操作，然後存儲以供後續使用。logstash 其實就是一個資料分析軟體，主要目的是分析 log 日志。Logstash 是一個動态資料收集管道。支援以 TCP/UDP/HTTP 多種方式收集資料（也可以接受 Beats 傳輸來的資料），并對資料做進一步豐富或提取字段處理。

  logstash 日志分析的配置和使用：https://www.cnblogs.com/yincheng/p/logstash.html

  ELK 之 Logstash 安裝與配置及使用: https://blog.csdn.net/CleverCode/article/details/78632887

• Kibana：可視化化平台。它能夠搜尋、展示存儲在 Elasticsearch 中索引資料。使用它可以很友善的用圖表、表格、地圖展示和分析資料。Kibana 是一個基于浏覽器頁面的 Elasticsearch 前端展示工具，是 ELK 的使用者界面，也是一個開源和免費的工具。它将收集的資料進行可視化展示（各種報表、圖形化資料），并提供配置、管理 ELK 的界面。Kibana可以為 Logstash 和 ElasticSearch 提供的日志分析友好的 Web 界面，可以幫助您彙總、分析和搜尋重要資料日志。
• Elasticsearch：分布式搜尋引擎。具有高可伸縮、高可靠、易管理等特點。可以用于全文檢索、結構化檢索和分析。Elasticsearch、Logstash 和 Kibana 這三個可以結合起來。Elasticsearch 基于 Lucene 開發，現在是使用最廣的開源搜尋引擎之一，Wikipedia 、StackOverflow、Github 等都基于它來建構自己的搜尋引擎。ElasticSearch 是一個基于 JSON 的分布式的搜尋和分析引擎。作為 ELK 的核心，它集中存儲資料。
• Filebeat：輕量級資料收集引擎。基于原先 Logstash-fowarder 的源碼改造出來。換句話說：Filebeat就是新版的 Logstash-fowarder，也會是 ELK Stack 在 shipper 端的第一選擇。

ELK 工作原理展示圖：

如上圖：Logstash 收集 AppServer 産生的 Log，并存放到 ElasticSearch 叢集中，而 Kibana 則從 ES 叢集中查詢資料生成圖表，再傳回給 Browser。

Logstash 工作原理：

Logstash事件處理有三個階段：inputs → filters → outputs。是一個接收，處理，轉發日志的工具。支援系統日志，webserver日志，錯誤日志，應用日志，總之包括所有可以抛出來的日志類型。

Input：輸入資料到logstash。

Logstash Reference：https://www.elastic.co/guide/en/logstash/current/index.html

一些常用的輸入為：

Logstat 輸入插件配置：https://www.elastic.co/guide/en/logstash/current/input-plugins.html

file：從檔案系統的檔案中讀取，類似于tial -f指令

syslog：在514端口上監聽系統日志消息，并根據RFC3164标準進行解析

redis：從redis service中讀取

beats：從filebeat中讀取

Filters：資料中間處理，對資料進行操作。

一些常用的過濾器為：

Logstat 過濾插件：https://www.elastic.co/guide/en/logstash/current/filter-plugins.html

grok：解析任意文本資料，Grok 是 Logstash 最重要的插件。它的主要作用就是将文本格式的字元串，轉換成為具體的結構化的資料，配合正規表達式使用。内置120多個解析文法。

mutate：對字段進行轉換。例如對字段進行删除、替換、修改、重命名等。

drop：丢棄一部分events不進行處理。

clone：拷貝 event，這個過程中也可以添加或移除字段。

geoip：添加地理資訊(為前台kibana圖形化展示使用)

Outputs：outputs是logstash處理管道的最末端元件。一個event可以在處理過程中經過多重輸出，但是一旦所有的outputs都執行結束，這個event也就完成生命周期。

一些常見的outputs為：

Logstat 輸出插件：https://www.elastic.co/guide/en/logstash/current/output-plugins.html

elasticsearch：可以高效的儲存資料，并且能夠友善和簡單的進行查詢。

file：将event資料儲存到檔案中。

graphite：将event資料發送到圖形化元件中，一個很流行的開源存儲圖形化展示的元件。

Codecs：codecs 是基于資料流的過濾器，它可以作為input，output的一部配置設定置。Codecs可以幫助你輕松的分割發送過來已經被序列化的資料。

一些常見的codecs：

Logstat codes ：https://www.elastic.co/guide/en/logstash/current/codec-plugins.html

json：使用json格式對資料進行編碼/解碼。

multiline：将彙多個事件中資料彙總為一個單一的行。比如：java異常資訊和堆棧資訊。

簡單版架構

圖 1

圖 2

這種架構下把 Logstash 執行個體與 Elasticsearch 執行個體直接相連。Logstash 執行個體直接通過 Input 插件讀取資料源資料(比如 Java 日志， Nginx 日志等)，經過 Filter 插件進行過濾日志，最後通過 Output 插件将資料寫入到 ElasticSearch 執行個體中（Logstash部署在各個節點上搜集相關日志、資料，并經過分析、過濾後發送給遠端伺服器上的Elasticsearch進行存儲）。Elasticsearch 再将資料以分片的形式壓縮存儲，并提供多種API供使用者查詢、操作。使用者可以通過Kibana Web直覺的對日志進行查詢，并根據需求生成資料報表。

這個階段，日志的收集、過濾、輸出等功能，主要由這三個核心元件組成 Input 、Filter、Output

• Input：輸入，輸入資料可以是 File 、 Stdin（直接從控制台輸入） 、TCP、Syslog 、Redis 、Collectd 等
• Filter：過濾，将日志輸出成我們想要的格式。Logstash 存在豐富的過濾插件：Grok 正則捕獲、時間處理、JSON 編解碼、資料修改 Mutate 。Grok 是 Logstash 中最重要的插件，強烈建議每個人都要使用 Grok Debugger 來調試自己的 Grok 表達式：grok { match => ["message", "(?m)[%{LOGLEVEL:level}] [%{TIMESTAMP_ISO8601:timestamp}] [%{DATA:logger}] [%{DATA:threadId}] [%{DATA:requestId}] %{GREEDYDATA:msgRawData}"] }
• Output：輸出，輸出目标可以是 Stdout （直接從控制台輸出）、Elasticsearch 、Redis 、TCP 、File 等

這是最簡單的一種ELK架構方式，Logstash 執行個體直接與 Elasticsearch 執行個體連接配接。優點是搭建簡單，易于上手。建議供初學者學習與參考，不能用于線上的環境。

叢集版架構

圖 1

此架構适合大型叢集、海量資料的業務場景，它通過将前端Logstash Agent替換成filebeat，有效降低了收集日志對業務系統資源的消耗。同時，消息隊列使用kafka叢集架構，有效保障了收集資料的安全性和穩定性，而後端Logstash和Elasticsearch均采用叢集模式搭建，從整體上提高了ELK系統的高效性、擴充性和吞吐量。

圖 2

這種架構下我們采用多個 Elasticsearch 節點組成 Elasticsearch 叢集，由于 Logstash 與 Elasticsearch 采用叢集模式運作，叢集模式可以避免單執行個體壓力過重的問題，同時線上上各個伺服器上部署 Logstash Agent，來滿足資料量不大且可靠性不強的場景。

資料收集端：每台伺服器上面部署 Logstash Shipper Agent 來收集目前伺服器上日志，日志經過 Logstash Shipper 中 Input插件、Filter插件、Output 插件傳輸到 Elasticsearch 叢集

資料存儲與搜尋：Elasticsearch 配置預設即可滿足，同時我們看資料重要性來決定是否添加副本，如果需要的話，最多一個副本即可

資料展示：Kibana 可以根據 Elasticsearch 的資料來做各種各樣的圖表來直覺的展示業務實時狀況

這種架構使用場景非常有限，主要存在以下兩個問題

• 消耗伺服器資源：Logstash 的收集、過濾都在伺服器上完成，這就造成伺服器上占用系統資源較高、性能方面不是很好，調試、跟蹤困難，異常處理困難
• 資料丢失：大并發情況下，由于日志傳輸峰值比較大，沒有消息隊列來做緩沖，就會導緻 Elasticsearch 叢集丢失資料

這個架構相對上個版本略微複雜，不過維護起來同樣比較友善，同時可以滿足資料量不大且可靠性不強的業務使用。

引入消息隊列

圖 1

此架構主要特點是引入了消息隊列機制，位于各個節點上的Logstash Agent（一級Logstash，主要用來傳輸資料）先将資料傳遞給消息隊列（常見的有Kafka、Redis等），接着，Logstash server（二級Logstash，主要用來拉取消息隊列資料，過濾并分析資料）将格式化的資料傳遞給Elasticsearch進行存儲。最後，由Kibana将日志和資料呈現給使用者。由于引入了Kafka（或者Redis）緩存機制，即使遠端Logstash server因故障停止運作，資料也不會丢失，因為資料已經被存儲下來了。

這種架構适合于較大叢集、資料量一般的應用環境，但由于二級Logstash要分析處理大量資料，同時Elasticsearch也要存儲和索引大量資料，是以它們的負荷會比較重，解決的方法是将它們配置為叢集模式，以分擔負載。

此架構的優點在于引入了消息隊列機制，均衡了網絡傳輸，進而降低了網絡閉塞尤其是丢失資料的可能性，但依然存在Logstash占用系統資源過多的問題，在海量資料應用場景下，可能會出現性能瓶頸。

圖 2

該場景下面，多個資料首先通過 Logstash Shipper Agent 來收集資料，然後經過 Output 插件将資料投遞到 Kafka 叢集中，這樣當遇到 Logstash 接收資料的能力超過 Elasticsearch 叢集處理能力的時候，就可以通過隊列就能起到削峰填谷的作用， Elasticsearch 叢集就不存在丢失資料的問題。

目前業界在日志服務場景中，使用比較多的兩種消息隊列為 ：Kafka 和 Redis。盡管 ELK Stack 官網建議使用 Redis 來做消息隊列，但是我們建議采用 Kafka 。主要從下面兩個方面考慮：

• 資料丢失：Redis 隊列多用于實時性較高的消息推送，并不保證可靠。Kafka保證可靠但有點延時。
• 資料堆積：Redis 隊列容量取決于機器記憶體大小，如果超過設定的Max memory，資料就會抛棄。Kafka 的堆積能力取決于機器硬碟大小。

綜合上述的理由，我們決定采用 Kafka 來緩沖隊列。不過在這種架構下仍然存在一系列問題

• Logstash shipper 收集資料同樣會消耗 CPU 和記憶體資源
• 不支援多機房部署

這種架構适合較大叢集的應用部署，通過消息隊列解決了消息丢失、網絡堵塞的問題。

多機房部署

随着業務的飛速增長，單機房的架構已經不能滿足需求，不可避免的需要将業務分布到不同機房中，對于日志服務來說也是不小的挑戰。當然業界也有不少成熟的方法，比如阿裡的單元化、騰訊的 SET 方案等等。單元化在這邊不詳細展開，大家可以參考微網誌的【單元化架構】。

最終我們決定采用單元化部署的方式來解決 ELK 多機房中遇到的問題(延時、專線流量過大等)，從日志的産生、收集、傳輸、存儲、展示都是在同機房裡面閉環消化，不存在跨機房傳輸與調用的問題。因為互動緊密的應用盡量部署在同機房，是以這種方案并不會給業務查詢造成困擾。

Logstash、Elasticsearch、Kafka、Kibana 四個叢集都部署到同一機房中，每個機房都要每個機房自己的日志服務叢集，比如A機房業務的日志隻能傳輸給本機房 Kafka ，而A機房 Indexer 叢集消費并寫入到A機房 Elasticsearch 叢集中，并由A機房 Kibana 叢集展示，中間任何一個步驟不依賴B機房任何服務。

引入 Filebeat

如果日志量很大，Logstash 會遇到資源占用高的問題，為解決這個問題，引入了 Filebeat。Filebeat 是基于 logstash-forwarder 的源碼改造而成，用 Golang 編寫，無需依賴 Java 環境，效率高，占用記憶體 和 CPU 比較少，非常适合作為 Agent 跑在伺服器上。

下面看看 Filebeat 的基本用法。編寫配置檔案，從 Nginx access.log 中解析日志資料：

# filebeat.ymlfilebeat.prospectors:- input_type: log paths: /var/log/nginx/access.log json.message_key:output.elasticsearch: hosts: ["localhost"] index: "filebeat-nginx-%{+yyyy.MM.dd}"

我們來看看壓測資料：

壓測環境

• 虛拟機 8 cores 64G記憶體 540G SATA盤
• Logstash 版本 2.3.1
• Filebeat 版本 5.5.0

壓測方案

Logstash / Filebeat 讀取 350W 條日志 到 console，單行資料 580B，8個程序寫入采集檔案

壓測結果

項目	workers	cpu usr	總共耗時	收集速度
Logstash	8	53.7%	210s	1.6w line/s
Filebeat	8	38.0%	30s	11w line/s

Filebeat 所消耗的CPU隻有 Logstash 的70%，但收集速度為 Logstash 的7倍。從我們的應用實踐來看，Filebeat 确實用較低的成本和穩定的服務品質，解決了 Logstash 的資源消耗問題。

最後，分享給大家一些血淚教訓，希望大家以我為鑒。

1. Indexer 運作一段時間後自動挂掉

突然有一天監控發現日志不消費了，排查下來發現消費 Kafka 資料的 indexer 挂掉了。是以，Indexer 程序也是需要用 supervisor 來監控的，保證它時刻都在運作。

2. Java異常日志輸出

開始我們在通過 grok 切割日志的時候，發現 Java 的 Exception 日志輸出之後，會出現換行的問題。後來使用 Logstash codec/multiline插件來解決。

input { stdin { codec => multiline { pattern => "^[" negate => true what => "previous" } }} 3. 由于時區導緻日志8小時時差

Logstash 2.3版本 date插件配置如下，檢視解析結果發現@timestamp比中國時間早了8小時。

解決方案 Kibana 讀取浏覽器的目前時區，然後在頁面上轉換時間内容的顯示。

date { match => [ "log_timestamp", "YYYY-MM-dd HH:mm:ss.SSS" ] target => "@timestamp" } 4.Grok parse failure

我們遇到線上 node 日志突然有幾天日志檢視不出來。後來拉出原始日志對比才發現生成出來的日志格式不正确，同時包含 JSON 格式和非 JSON 格式的日志。但是我們用grok解析的時候采用是 json 格式。建議大家輸出日志保證格式一緻同時不要出現空格等異常字元，可以使用線上 grok debug (http://grokdebug.herokuapp.com/) 來調試正則。

總結

基于 ELK stack 的日志解決方案的優勢主要展現于：

• 可擴充性：采用高可擴充性的分布式系統架構設計，可以支援每日 TB 級别的新增資料。
• 使用簡單：通過使用者圖形界面實作各種統計分析功能，簡單易用，上手快
• 快速響應：從日志産生到查詢可見，能達到秒級完成資料的采集、處理和搜尋統計。
• 界面炫麗：Kibana 界面上，隻需要點選滑鼠，就可以完成搜尋、聚合功能，生成炫麗的儀表闆。

參考資料

• https://www.elastic.co/guide/en/beats/filebeat/1.3/filebeat-overview.html
• https://zhuanlan.zhihu.com/p/26399963

安裝

ELK 要求本地環境中安裝了 JDK 。如果不确定是否已安裝，可使用下面的指令檢查：java -version

友情提示：安裝 ELK 時，三個應用請選擇統一的版本，避免出現一些莫名其妙的問題。

                  例如：由于版本不統一，導緻三個應用間的通訊異常。

安裝 FAQ

elasticsearch 不允許以 root 權限來運作。

問題：在 Linux 環境中，elasticsearch 不允許以 root 權限來運作。如果以 root 身份運作 elasticsearch，會提示這樣的錯誤：can not run elasticsearch as root

解決方法：使用非 root 權限賬号運作 elasticsearch

# 建立使用者組
groupadd elk
# 建立新使用者，-g elk 設定其使用者組為 elk，-p elk 設定其密碼為 elk
useradd elk -g elk -p elk
# 更改 /opt 檔案夾及内部檔案的所屬使用者及組為 elk:elk
chown -R elk:elk /opt # 假設你的 elasticsearch 安裝在 opt 目錄下
# 切換賬号
su elk
           

vm.max_map_count 不低于 262144

問題：vm.max_map_count 表示虛拟記憶體大小，它是一個核心參數。elasticsearch 預設要求 vm.max_map_count 不低于 262144。

max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]

解決方法：你可以執行以下指令，設定 vm.max_map_count ，但是重新開機後又會恢複為原值。

sysctl -w vm.max_map_count=262144
           

持久性的做法是在 /etc/sysctl.conf 檔案中修改 vm.max_map_count 參數：

echo "vm.max_map_count=262144" > /etc/sysctl.conf
sysctl -p
           

注意：如果運作環境為 docker 容器，可能會限制執行 sysctl 來修改核心參數。這種情況下，你隻能選擇直接修改主控端上的參數了。

nofile 不低于 65536

問題： nofile 表示程序允許打開的最大檔案數。elasticsearch 程序要求可以打開的最大檔案數不低于 65536。

max file descriptors [4096] for elasticsearch process is too low, increase to at least [65536]

解決方法：在 /etc/security/limits.conf 檔案中修改 nofile 參數

echo "* soft nofile 65536" > /etc/security/limits.conf
echo "* hard nofile 131072" > /etc/security/limits.conf
           

nproc 不低于 2048

問題： nproc 表示最大線程數。elasticsearch 要求最大線程數不低于 2048。

max number of threads [1024] for user [user] is too low, increase to at least [2048]

解決方法：在 /etc/security/limits.conf 檔案中修改 nproc 參數

echo "* soft nproc 2048" > /etc/security/limits.conf
echo "* hard nproc 4096" > /etc/security/limits.conf
           

Kibana No Default Index Pattern Warning

問題：安裝 ELK 後，通路 kibana 頁面時，提示以下錯誤資訊：

Warning No default index pattern. You must select or create one to continue.

...

Unable to fetch mapping. Do you have indices matching the pattern?

這就說明 logstash 沒有把日志寫入到 elasticsearch。

解決方法：檢查 logstash 與 elasticsearch 之間的通訊是否有問題，一般問題就出在這。

Elasticsearch

安裝步驟：

确定機器已經安裝了 Java 環境後，就可以安裝 ES 了。官網提供了壓縮包可以直接下載下傳，

1. elasticsearch 官方下載下傳位址下載下傳所需版本包并解壓到本地。本次安裝以Ubuntu為例。

右鍵選擇 複制連結位址 ，使用 wget 下載下傳 Filebeat.tar.gz 壓縮包。然後使用 tar -zxvf 指令解壓。

# 下載下傳壓縮包
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.2.2.tar.gz
# 解壓
tar -xzf elasticsearch-6.2.2.tar.gz
# 進入檔案
cd elasticsearch-6.2.2/
           

Windows、MAC、Linux 下載下傳安裝：

Linux 上可以執行下面的指令來下載下傳壓縮包：
curl -L -O https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.0.0.tar.gz

Mac 上可以執行以下指令來進行安裝：
brew install elasticsearch

Windows 上可以選擇 MSI 可執行安裝程式，将應用安裝到本地
           

2. 解壓完成後進入目錄，啟動指令在 bin 目錄下，直接運作指令就可以啟動了。

        Linux 下運作 

bin/elasticsearch

 （ Windows 上運作 

bin\elasticsearch.bat

）

        注意：在 Linux 環境中，elasticsearch 不允許以 root 權限來運作。使用非 root 權限賬号運作 elasticsearch

# 建立使用者組
groupadd elk
# 建立新使用者，-g elk 設定其使用者組為 elk，-p elk 設定其密碼為 elk
useradd elk -g elk -p elk
# 更改 /opt 檔案夾及内部檔案的所屬使用者及組為 elk:elk
chown -R elk:elk /opt # 假設你的 elasticsearch 安裝在 opt 目錄下
# 切換賬号
su elk
           

3. 驗證運作成功：linux 上可以執行 

curl http://localhost:9200/

 ；windows 上可以用通路 REST 接口的方式來通路 http://localhost:9200/

上圖表示Elasticsearch正常啟動。

ElasticSearch 的預設啟動端口是 9200。手動通路出現如下資訊說明啟動成功。

Logstash

安裝步驟：

1. 在 logstash 官方下載下傳位址下載下傳所需版本包并解壓到本地。

右鍵 選擇 複制連結位址，使用 wget 下載下傳 tar.gz 壓縮包。然後使用 tar 解壓

tar -zxvf logstash-6.2.2.tar.gz
# 進入目錄
cd logstash-6.2.2
           

2.配置 logstash。需要建立一個配置檔案，并指定要使用的插件和每個插件的設定。我們建立名為 logstash-simple.conf 的檔案并将其儲存在與Logstash相同的目錄中。LogStash 的運作需要指定一個配置檔案，來指定資料的流向。

建立配置檔案

cd logstash-6.4.0/
vi logstash-simple.conf
           

在 logstash-simple.conf 中輸入以下内容

input { stdin { } }
output {
elasticsearch { hosts => ["localhost:9200"] }
stdout { codec => rubydebug }
}
           

或者：

# 配置輸入為 beats
input {
    beats {
            port => "5044"
    }
}

# 資料過濾
filter {
    grok {
            match => { "message" => "%{COMBINEDAPACHELOG}" }
    }
    geoip {
            source => "clientip"
    }
}

# 輸出到本機的 ES
output {
    elasticsearch {
            hosts => [ "localhost:9200"  ]
    }
}
           

3. 上面配置了 LogStash 輸出日志到 ES 中，具體字段在後面的筆記中會詳細介紹，這裡先用起來再說。

配置完成後就可以通過如下方式啟動 LogStash 了（Linux 下運作 

bin/logstash -f

logstash-simple.conf （Windows 上運作

bin/logstash.bat -f

logstash-simple.conf））

bin/logstash -f logstash-simple.conf --config.reload.automatic
           

可以看到指令行會列印出如下資訊， 可以看到 LogStash 預設端口為 5044:

[2018-03-08T23:12:44,087][WARN ][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified
[2018-03-08T23:12:44,925][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.2.2"}
[2018-03-08T23:12:45,623][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-03-08T23:12:49,960][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2018-03-08T23:12:50,882][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
[2018-03-08T23:12:50,894][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"}
[2018-03-08T23:12:51,303][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://localhost:9200/"}
[2018-03-08T23:12:51,595][INFO ][logstash.outputs.elasticsearch] ES Output version determined {:es_version=>nil}
[2018-03-08T23:12:51,604][WARN ][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document _type {:es_version=>6}
[2018-03-08T23:12:51,641][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=>nil}
[2018-03-08T23:12:51,676][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=>{"template"=>"logstash-*", "version"=>60001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"_default_"=>{"dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"*", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword", "ignore_above"=>256}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date"}, "@version"=>{"type"=>"keyword"}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
[2018-03-08T23:12:51,773][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["//localhost:9200"]}
[2018-03-08T23:12:52,176][INFO ][logstash.filters.geoip   ] Using geoip database {:path=>"/Users/zouyingjie/soft/study/ELK/logstash-6.2.2/vendor/bundle/jruby/2.3.0/gems/logstash-filter-geoip-5.0.3-java/vendor/GeoLite2-City.mmdb"}
[2018-03-08T23:12:53,026][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"}
[2018-03-08T23:12:53,195][INFO ][logstash.pipeline        ] Pipeline started succesfully {:pipeline_id=>"main", :thread=>"#<Thread:0x66461e40 run>"}
[2018-03-08T23:12:53,290][INFO ][org.logstash.beats.Server] Starting server on port: 5044
[2018-03-08T23:12:53,401][INFO ][logstash.agent           ] Pipelines running {:count=>1, :pipelines=>["main"]}
           

FileBeats

FileBeats 也提供了下載下傳包，位址為 https://www.elastic.co/downloads/beats/filebeat 。找到系統對應的包下載下傳後解壓即可。

右鍵 選擇 複制連結位址 使用 wget 下載下傳 Filebeat.tar.gz 壓縮包。

然後使用 tar -zxvf 指令解壓。

tar -zxvf filebeat-6.2.2-darwin-x86_64.tar.gz
cd filebeat-6.2.2-darwin-x86_64
           

進入目錄編輯 filebeat.yml 找到對應的配置項，配置如下

- type: log
   # Change to true to enable this prospector configuration.
    enabled: True

    # Paths that should be crawled and fetched. Glob based paths.
    # 讀取 Nginx 的日志
    paths:
      - /usr/local/nginx/logs/*.log

#----------------------------- Logstash output --------------------------------
# 輸出到本機的 LogStash
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]
           

配置完成後執行如下指令，啟動 FileBeat 即可

# FileBeat 需要以 root 身份啟動，是以先更改配置檔案的權限
sudo chown root filebeat.yml
sudo ./filebeat -e -c filebeat.yml -d "publish"
           

Kibana

百度   kibana 使用教程

Kibana查詢語言增強功能：https://www.elastic.co/guide/en/kibana/6.3/kuery-query.html#kuery-query

Kibana 使用者手冊：https://www.elastic.co/guide/cn/kibana/current/index.html

安裝步驟：

1. 在 kibana 官方下載下傳位址下載下傳所需版本包并解壓到本地。
2. 修改 

   config/kibana.yml

    配置檔案，設定 

   elasticsearch.url

    指向 Elasticsearch 執行個體。設定 server.host: "0.0.0.0"
3. 運作 

   bin/kibana

    （Windows 上運作 

   bin\kibana.bat

   ）
4. 在浏覽器上通路 http://localhost:5601

Kibana 也提供了對應的安裝包下載下傳，連結為 https://www.elastic.co/downloads/kibana , Mac、Linux、Win 都有對應的安裝包，直接下載下傳解壓即可

tar zxvf kibana-6.2.2-darwin-x86_64.tar.gz
cd kibana-6.2.2-darwin-x86_64
# 直接啟動即可
bin/kibana
           

Kibana 預設連結了本機的 9200 端口，其綁定的端口為 5601，啟動成功後直接通路 127.0.0.1:5601 端口即可,界面如下。我因為安裝了 x-pack 插件是以顯示的項可能會多一些，這個暫時忽略.

點選右上方的 Discover. 界面會提示建立索引模式，通過這個來檢索 ES 中的索引，可以看到已經有一個 Logstash 的索引了，輸入名稱進行完全比對，

這裡標明一個時間戳，使用預設的 timestamp 即可，設定完成後我們可以根據時間範圍篩選資料。

設定完成後建立後顯示如下

這時在點選 Discover 就可以看到我們建立的索引了，此時輸入 UnicodeEncodeError 已經傳回了比對到的日志資訊。

至此就完成了簡單的日志分析平台的搭建。OK，關于安裝就介紹到這裡，後面正式開始對整個 ELK Stack 的學習。

注意：如果在 discover 裡面找不到資料，有可能是時區的問題，可以通過 Management -> Advanced Settings -> dateFormat:tz 更改時區，改為 UTC 即可。

快速搭建ELK日志分析系統

From：https://blog.csdn.net/e_wsq/article/details/81303713

環境

2 台 Centos 主機
IP：192.168.1.202 安裝： elasticsearch、logstash、Kibana、Nginx、Http、Redis
    192.168.1.201 安裝:  logstash
           

安裝

安裝elasticsearch的yum源的密鑰（這個需要在所有伺服器上都配置）
# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
 
配置elasticsearch的yum源
# vim /etc/yum.repos.d/elasticsearch.repo
 
在elasticsearch.repo檔案中添加如下内容
[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
           

安裝elasticsearch的環境

安裝elasticsearch
# yum install -y elasticsearch
 
安裝java環境(java環境必須是1.8版本以上的)
 
wget http://download.oracle.com/otn-pub/java/jdk/8u131-b11/d54c1d3a095b4ff2b6607d096fa80163/jdk-8u131-linux-x64.rpm
 
rpm -ivh jdk-8u131-linux-x64.rpm 
 
驗證java安裝成功
java -version
java version "1.8.0_131"
Java(TM) SE Runtime Environment (build 1.8.0_131-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.131-b11, mixed mode)
           

建立elasticsearch data的存放目錄，并修改該目錄的屬主屬組

# mkdir -p /data/es-data   (自定義用于存放data資料的目錄)
# chown -R elasticsearch:elasticsearch /data/es-data
           

修改elasticsearch的日志屬主屬組

# chown -R elasticsearch:elasticsearch /var/log/elasticsearch/
           

修改elasticsearch的配置檔案

# vim /etc/elasticsearch/elasticsearch.yml
 
找到配置檔案中的cluster.name，打開該配置并設定叢集名稱
cluster.name: demon
 
找到配置檔案中的node.name，打開該配置并設定節點名稱
node.name: elk-1
 
修改data存放的路徑
path.data: /data/es-data
 
修改logs日志的路徑
path.logs: /var/log/elasticsearch/
 
配置記憶體使用用交換分區
bootstrap.memory_lock: true
 
監聽的網絡位址
network.host: 0.0.0.0
 
開啟監聽的端口
http.port: 9200
 
增加新的參數，這樣head插件可以通路es (5.x版本，如果沒有可以自己手動加)
http.cors.enabled: true
http.cors.allow-origin: "*"
啟動elasticsearch服務
           

啟動服務

/etc/init.d/elasticsearch start
 
Starting elasticsearch: Java HotSpot(TM) 64-Bit Server VM warning: INFO: os::commit_memory(0x0000000085330000, 2060255232, 0) failed; error='Cannot allocate memory' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 2060255232 bytes for committing reserved memory.
# An error report file with more information is saved as:
# /tmp/hs_err_pid2616.log
                                                           [FAILED]
這個報錯是因為預設使用的記憶體大小為2G，虛拟機沒有那麼多的空間
 
修改參數：
vim /etc/elasticsearch/jvm.options
-Xms512m
-Xmx512m
 
再次啟動
/etc/init.d/elasticsearch start
 
檢視服務狀态，如果有報錯可以去看錯誤日志 less /var/log/elasticsearch/demon.log(日志的名稱是以叢集名稱命名的)
 
建立開機自啟動服務
# chkconfig elasticsearch on
           

注意事項

需要修改幾個參數，不然啟動會報錯
 
vim /etc/security/limits.conf
在末尾追加以下内容（elk為啟動使用者，當然也可以指定為*）
elk soft nofile 65536
elk hard nofile 65536
elk soft nproc 2048
elk hard nproc 2048
elk soft memlock unlimited
elk hard memlock unlimited
 
繼續再修改一個參數
vim /etc/security/limits.d/90-nproc.conf
将裡面的1024改為2048（ES最少要求為2048）
*          soft    nproc     2048
 
另外還需注意一個問題（在日志發現如下内容，這樣也會導緻啟動失敗，這一問題困擾了很久）
[2017-06-14T19:19:01,641][INFO ][o.e.b.BootstrapChecks    ] [elk-1] bound or publishing to a non-loopback or non-link-local address, enforcing bootstrap checks
[2017-06-14T19:19:01,658][ERROR][o.e.b.Bootstrap          ] [elk-1] node validation exception
[1] bootstrap checks failed
[1]: system call filters failed to install; check the logs and fix your configuration or disable system call filters at your own risk
    
解決：修改配置檔案，在配置檔案添加一項參數（目前還沒明白此參數的作用）
vim /etc/elasticsearch/elasticsearch.yml 
bootstrap.system_call_filter: false
           

通過浏覽器請求下9200的端口，看下是否成功

先檢查9200端口是否起來
netstat -antp |grep 9200
tcp        0      0 :::9200                     :::*                        LISTEN      2934/java
 
浏覽器通路測試是否正常（以下為正常）
# curl http://127.0.0.1:9200/
{
  "name" : "linux-node1",
  "cluster_name" : "demon",
  "cluster_uuid" : "kM0GMFrsQ8K_cl5Fn7BF-g",
  "version" : {
    "number" : "5.4.0",
    "build_hash" : "780f8c4",
    "build_date" : "2017-04-28T17:43:27.229Z",
    "build_snapshot" : false,
    "lucene_version" : "6.5.0"
  },
  "tagline" : "You Know, for Search"
}
           

如何和elasticsearch互動

JavaAPI
 
RESTful API
Javascript,.Net,PHP,Perl,Python
 
利用API檢視狀态
# curl -i -XGET 'localhost:9200/_count?pretty' 
    HTTP/1.1 200 OK
    content-type: application/json; charset=UTF-8
    content-length: 95
 
    {
      "count" : 0,
      "_shards" : {
        "total" : 0,
        "successful" : 0,
        "failed" : 0
      }
    }
           

安裝插件

安裝elasticsearch-head插件
 
安裝docker鏡像或者通過github下載下傳elasticsearch-head項目都是可以的，1或者2兩種方式選擇一種安裝使用即可
 
1. 使用docker的內建好的elasticsearch-head
    # docker run -p 9100:9100 mobz/elasticsearch-head:5
    
    docker容器下載下傳成功并啟動以後，運作浏覽器打開http://localhost:9100/
 
2. 使用git安裝elasticsearch-head
    # yum install -y npm
    # git clone git://github.com/mobz/elasticsearch-head.git
    # cd elasticsearch-head
    # npm install
    # npm run start
    檢查端口是否起來
    netstat -antp |grep 9100
    浏覽器通路測試是否正常
    http://IP:9100/
           

截圖：

LogStash的使用

安裝Logstash環境：
 
官方安裝手冊:
https://www.elastic.co/guide/en/logstash/current/installing-logstash.html
 
下載下傳yum源的密鑰認證：
# rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
 
利用yum安裝logstash
# yum install -y logstash
 
檢視下logstash的安裝目錄
# rpm -ql logstash
 
建立一個軟連接配接，每次執行指令的時候不用在寫安裝路勁（預設安裝在/usr/share下）
ln -s /usr/share/logstash/bin/logstash /bin/
 
執行logstash的指令
# logstash -e 'input { stdin { } } output { stdout {} }'
 
運作成功以後輸入:
nihao
 
stdout傳回的結果:
           

截圖：

注:
   -e          執行操作
   input       标準輸入
   { input }   插件
   output      标準輸出
   { stdout }  插件  
 
通過rubydebug來輸出下更詳細的資訊
# logstash -e 'input { stdin { } } output { stdout {codec => rubydebug} }'
 
執行成功輸入:
nihao
 
stdout輸出的結果:
           

如果标準輸出還有elasticsearch中都需要保留應該怎麼玩，看下面
# /usr/share/logstash/bin/logstash -e 'input { stdin { } } output { elasticsearch { hosts => ["192.168.1.202:9200"] } stdout { codec => rubydebug }}'
 
運作成功以後輸入:
I am elk
 
傳回的結果(标準輸出中的結果):
           

logstash使用配置檔案

官方指南:
https://www.elastic.co/guide/en/logstash/current/configuration.html
 
建立配置檔案01-logstash.conf
# vim /etc/logstash/conf.d/elk.conf
 
檔案中添加以下内容
input { stdin { } }
output {
  elasticsearch { hosts => ["192.168.1.202:9200"] }
  stdout { codec => rubydebug }
}
 
使用配置檔案運作logstash
# logstash -f ./elk.conf
 
運作成功以後輸入以及标準輸出結果
           

logstash的資料庫類型

1. Input插件
    權威指南:https://www.elastic.co/guide/en/logstash/current/input-plugins.html
    
    file插件的使用
    # vim /etc/logstash/conf.d/elk.conf
 
    添加如下配置
    input {
        file {
            path => "/var/log/messages"
            type => "system"
            start_position => "beginning"
            }
    }
    output {    
             elasticsearch {
                    hosts => ["192.168.1.202:9200"]
                    index => "system-%{+YYYY.MM.dd}"
                }
    }
 
 
運作logstash指定elk.conf配置檔案，進行過濾比對
#logstash -f /etc/logstash/conf.d/elk.conf
           

來一發配置安全日志的并且把日志的索引按類型做存放，繼續編輯elk.conf檔案

# vim /etc/logstash/conf.d/elk.conf
 
添加secure日志的路徑
input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }
 
    file {
        path => "/var/log/secure"
        type => "secure"
        start_position => "beginning"
    }
}
 
output {
 
    if [type] == "system" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-system-%{+YYYY.MM.dd}"
        }
    }
 
    if [type] == "secure" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-secure-%{+YYYY.MM.dd}"
        }
    }
}
 
運作logstash指定elk.conf配置檔案，進行過濾比對
# logstash -f ./elk.conf 
           

這些設定都沒有問題之後，接下來安裝下kibana，可以讓在前台展示

Kibana的安裝及使用

安裝kibana環境
 
官方安裝手冊:https://www.elastic.co/guide/en/kibana/current/install.html
 
下載下傳kibana的tar.gz的軟體包
# wget https://artifacts.elastic.co/downloads/kibana/kibana-5.4.0-linux-x86_64.tar.gz
 
解壓kibana的tar包
# tar -xzf kibana-5.4.0-linux-x86_64.tar.gz
 
進入解壓好的kibana
# mv kibana-5.4.0-linux-x86_64 /usr/local
 
建立kibana的軟連接配接
# ln -s /usr/local/kibana-5.4.0-linux-x86_64/ /usr/local/kibana 
 
編輯kibana的配置檔案
# vim /usr/local/kibana/config/kibana.yml
 
修改配置檔案如下，開啟以下的配置
server.port: 5601
 
server.host: "0.0.0.0"
 
elasticsearch.url: "http://192.168.1.202:9200"
 
kibana.index: ".kibana" 
 
安裝screen,以便于kibana在背景運作（當然也可以不用安裝，用其他方式進行背景啟動）
# yum -y install screen
 
# screen
 
# /usr/local/kibana/bin/kibana
netstat -antp |grep 5601
tcp        0      0 0.0.0.0:5601                0.0.0.0:*                   LISTEN      17007/node 
打開浏覽器并設定對應的index
http://IP:5601
           

二、ELK實戰篇

好，現在索引也可以建立了，現在可以來輸出nginx、apache、message、secrue的日志到前台展示（Nginx有的話直接修改，沒有自行安裝）

編輯nginx配置檔案，修改以下内容（在http子產品下添加）
 
log_format json '{"@timestamp":"$time_iso8601",'
             '"@version":"1",'
             '"client":"$remote_addr",'
             '"url":"$uri",'
             '"status":"$status",'
             '"domian":"$host",'
             '"host":"$server_addr",'
             '"size":"$body_bytes_sent",'
             '"responsetime":"$request_time",'
             '"referer":"$http_referer",'
             '"ua":"$http_user_agent"'
          '}';
 
修改access_log的輸出格式為剛才定義的json 
access_log  logs/elk.access.log  json;
 
繼續修改apache的配置檔案
 
LogFormat "{ \
        \"@timestamp\": \"%{%Y-%m-%dT%H:%M:%S%z}t\", \
        \"@version\": \"1\", \
        \"tags\":[\"apache\"], \
        \"message\": \"%h %l %u %t \\\"%r\\\" %>s %b\", \
        \"clientip\": \"%a\", \
        \"duration\": %D, \
        \"status\": %>s, \
        \"request\": \"%U%q\", \
        \"urlpath\": \"%U\", \
        \"urlquery\": \"%q\", \
        \"bytes\": %B, \
        \"method\": \"%m\", \
        \"site\": \"%{Host}i\", \
        \"referer\": \"%{Referer}i\", \
        \"useragent\": \"%{User-agent}i\" \
       }" ls_apache_json
 
一樣修改輸出格式為上面定義的json格式
CustomLog logs/access_log ls_apache_json
 
編輯logstash配置檔案，進行日志收集
vim /etc/logstash/conf.d/full.conf
 
    input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }   
 
    file {
        path => "/var/log/secure"
        type => "secure"
        start_position => "beginning"
    }   
 
    file {
        path => "/var/log/httpd/access_log"
        type => "http"
        start_position => "beginning"
    }   
 
    file {
        path => "/usr/local/nginx/logs/elk.access.log"
        type => "nginx"
        start_position => "beginning"
    }   
 
}
    
output {
 
    if [type] == "system" { 
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-system-%{+YYYY.MM.dd}"
        }       
    }   
 
    if [type] == "secure" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-secure-%{+YYYY.MM.dd}"
        }
    }
 
    if [type] == "http" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-http-%{+YYYY.MM.dd}"
        }
    }
 
    if [type] == "nginx" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-nginx-%{+YYYY.MM.dd}"
        }
    }
 
}
 
運作看看效果如何
logstash -f /etc/logstash/conf.d/full.conf
           

可以發現所有建立日志的索引都已存在，接下來就去Kibana建立日志索引，進行展示（按照上面的方法進行建立索引即可），看下展示的效果

接下來再來一發MySQL慢日志的展示

由于MySQL的慢日志查詢格式比較特殊，是以需要用正則進行比對，并使用multiline能夠進行多行比對（看具體配置）
input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }   
 
    file {
        path => "/var/log/secure"
        type => "secure"
        start_position => "beginning"
    }   
 
    file {
        path => "/var/log/httpd/access_log"
        type => "http"
        start_position => "beginning"
    }   
 
    file {
        path => "/usr/local/nginx/logs/elk.access.log"
        type => "nginx"
        start_position => "beginning"
    }   
    
    file {
        path => "/var/log/mysql/mysql.slow.log"
        type => "mysql"
        start_position => "beginning"   
    codec => multiline {
            pattern => "^# [email protected]:"
            negate => true
            what => "previous"
        }
    }
}
 
filter {
 
    grok {
        match => { "message" => "SELECT SLEEP" }
            add_tag => [ "sleep_drop" ]
            tag_on_failure => []
    }
 
 
    if "sleep_drop" in [tags] {
            drop {}
    }
    
    grok {
        match => { "message" => "(?m)^# [email protected]: %{USER:User}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:Client_IP})?\]\s.*# Query_time: %{NUMBER:Query_Time:float}\s+Lock_time: %{NUMBER:Lock_Time:float}\s+Rows_sent: %{NUMBER:Rows_Sent:int}\s+Rows_examined: %{NUMBER:Rows_Examined:int}\s*(?:use %{DATA:Database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<Query>(?<Action>\w+)\s+.*)\n# Time:.*$" }
        }
 
     date {
            match => [ "timestamp", "UNIX" ]
            remove_field => [ "timestamp" ]
    }
 
 
}
 
 
   
output {
 
    if [type] == "system" { 
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-system-%{+YYYY.MM.dd}"
        }       
    }   
 
    if [type] == "secure" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-secure-%{+YYYY.MM.dd}"
        }
    }
 
    if [type] == "http" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-http-%{+YYYY.MM.dd}"
        }
    }
 
    if [type] == "nginx" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-nginx-%{+YYYY.MM.dd}"
        }
    }
    
    if [type] == "mysql" {
 
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-mysql-slow-%{+YYYY.MM.dd}"
        }
    }
}
           

檢視效果(一條慢日志查詢會顯示一條，如果不進行正則比對，那麼一行就會顯示一條)

具體的日志輸出需求，進行具體的分析

三：ELK終極篇

安裝reids 
# yum install -y redis
 
修改redis的配置檔案
# vim /etc/redis.conf
 
修改内容如下
daemonize yes
 
bind 192.168.1.202
 
啟動redis服務
# /etc/init.d/redis restart
 
測試redis的是否啟用成功
# redis-cli -h 192.168.1.202
 
輸入info如果有不報錯即可
redis 192.168.1.202:6379> info
 
redis_version:2.4.10
....
 
編輯配置redis-out.conf配置檔案，把标準輸入的資料存儲到redis中
# vim /etc/logstash/conf.d/redis-out.conf
 
添加如下内容
 
input {
            stdin {}
}
 
output {
 
        redis {
                host => "192.168.1.202"
                port => "6379"
                password => 'test'
                db => '1'
                data_type => "list"
                key => 'elk-test'
        }
}   
 
運作logstash指定redis-out.conf的配置檔案
# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis-out.conf
           

運作成功以後，在logstash中輸入内容（檢視下效果）

編輯配置redis-in.conf配置檔案，把reids的存儲的資料輸出到elasticsearch中
# vim /etc/logstash/conf.d/redis-out.conf
 
添加如下内容
input{
    redis {
                    host => "192.168.1.202"
                    port => "6379"
                    password => 'test'
                    db => '1'
                    data_type => "list"
                    key => 'elk-test'
                    batch_count => 1 #這個值是指從隊列中讀取資料時，一次性取出多少條，預設125條（如果redis中沒有125條，就會報錯，是以在測試期間加上這個值）
            }
 
}
 
output {
        elasticsearch {
                hosts => ['192.168.1.202:9200']
                index => 'redis-test-%{+YYYY.MM.dd}'
        }
}
 
運作logstash指定redis-in.conf的配置檔案
# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis-out.conf
           

把之前的配置檔案修改一下，變成所有的日志監控的來源檔案都存放到redis中，然後通過redis在輸出到elasticsearch中
 
更改為如下，編輯full.conf
input {
    file {
            path => "/var/log/httpd/access_log"
            type => "http"
            start_position => "beginning"
    }
 
    file {
            path => "/usr/local/nginx/logs/elk.access.log"
            type => "nginx"
            start_position => "beginning"
    }
 
    file {
            path => "/var/log/secure"
            type => "secure"
            start_position => "beginning"
    }
 
    file {
            path => "/var/log/messages"
            type => "system"
            start_position => "beginning"
    }
}
 
 
output {
    if [type] == "http" {
        redis {
            host => "192.168.1.202"
            password => 'test'
            port => "6379"
            db => "6"
            data_type => "list"
            key => 'nagios_http' 
        }
    }
 
    if [type] == "nginx" {
        redis {
            host => "192.168.1.202"
            password => 'test'
            port => "6379"
            db => "6"
            data_type => "list"
            key => 'nagios_nginx' 
        }
    }
 
    if [type] == "secure" {
        redis {
            host => "192.168.1.202"
            password => 'test'
            port => "6379"
            db => "6"
            data_type => "list"
            key => 'nagios_secure' 
        }
    }
 
    if [type] == "system" {
        redis {
            host => "192.168.1.202"
            password => 'test'
            port => "6379"
            db => "6"
            data_type => "list"
            key => 'nagios_system' 
        }
    }
} 
 
 
運作logstash指定shipper.conf的配置檔案
# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/full.conf
 
在redis中檢視是否已經将資料寫到裡面(有時候輸入的日志檔案不産生日志，會導緻redis裡面也沒有寫入日志)
           

把redis中的資料讀取出來，寫入到elasticsearch中(需要另外一台主機做實驗)
 
編輯配置檔案
# vim /etc/logstash/conf.d/redis-out.conf
 
添加如下内容
input {
    redis {
        type => "system"
        host => "192.168.1.202"
        password => 'test'
        port => "6379"
        db => "6"
        data_type => "list"
        key => 'nagios_system' 
    batch_count => 1
     }
    
    redis {
        type => "http"
        host => "192.168.1.202"
        password => 'test'
        port => "6379"
        db => "6"
        data_type => "list"
        key => 'nagios_http' 
    batch_count => 1
     }
 
    redis {
        type => "nginx"
        host => "192.168.1.202"
        password => 'test'
        port => "6379"
        db => "6"
        data_type => "list"
        key => 'nagios_nginx'
    batch_count => 1
     }
    
    redis {
        type => "secure"
        host => "192.168.1.202"
        password => 'test'
        port => "6379"
        db => "6"
        data_type => "list"
        key => 'nagios_secure' 
    batch_count => 1
    }
}
    
output {
    
    if [type] == "system" {
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-system-%{+YYYY.MM.dd}"
        }
    }   
 
    if [type] == "http" {
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-http-%{+YYYY.MM.dd}"
        }   
    }   
 
    if [type] == "nginx" {
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-nginx-%{+YYYY.MM.dd}"
        }   
    }  
 
    if [type] == "secure" {
        elasticsearch {
            hosts => ["192.168.1.202:9200"]
            index => "nagios-secure-%{+YYYY.MM.dd}"
        }   
    }  
}
 
 
注意:
input是從用戶端收集的
output是同樣也儲存到192.168.1.202中的elasticsearch中，如果要儲存到目前的主機上，可以把output中的hosts修改成localhost，如果還需要在kibana中顯示，需要在本機上部署kabana，為何要這樣做，起到一個松耦合的目的
說白了，就是在用戶端收集日志，寫到服務端的redis裡或是本地的redis裡面，輸出的時候對接ES伺服器即可
 
運作指令看看效果
# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/redis-out.conf
           

效果是和直接往ES伺服器輸出一樣的（這樣是先将日志存到redis資料庫，然後再從redis資料庫裡取出日志）

上線 ELK

１.　日志分類
        系統日志        rsyslog     logstash    syslog插件
        通路日志        nginx       logstash    codec   json
        錯誤日志        file        logstash    mulitline
        運作日志        file        logstash    codec   json
        裝置日志        syslog      logstash    syslog插件
        Debug日志      file         logstash   json 或者 mulitline  
 
2.  日志标準化
        路徑      固定
        格式      盡量json
 
3.   系統個日志開始-->錯誤日志-->運作日志-->通路日志
           

因為ES儲存日志是永久儲存，是以需要定期删除一下日志，下面指令為删除指定時間前的日志

curl -X DELETE http://xx.xx.com:9200/logstash-*-`date +%Y-%m-%d -d "-$n days"`