ELK日志管理深度實戰

Logstash收集Rsyslog/syslog日志

syslog預設是通過514端口去發送日志，可以使用logstash安裝對應的插件，監聽514端口來收集日志。如果隻是監聽系統日志可以不用安裝logstash的agent,隻需要監聽對應的514端口，收集系統資料即可。

logstash在INPUT插件中提供了syslog的子產品，可以用于不安裝本地日志收集agent的裝置（如硬體防火牆等），當然支援此類協定的普通伺服器也适用。

注意：此INPUT插件會同時監聽TCP和UDP端口。

可以在安裝logstash伺服器node2(172.16.10.21)上編寫logstash配置檔案,指定端口（預設為514）同時指定寫入的ES伺服器：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/syslog.conf </code>

<code>input {</code>

<code> </code><code>syslog{</code>

<code> </code><code>type</code> <code>=> </code><code>"system-syslog"</code>

<code>output {</code>

<code> </code><code>elasticsearch {</code>

<code> </code><code>hosts => [</code><code>"172.16.10.21:9200"</code><code>] </code><code># 指定寫入的ES伺服器</code>

<code> </code><code>index => </code><code>"system-syslog-%{+YYYY.MM}"</code>

在node1(172.16.10.20)使syslog自動發送日志資訊到logstash，需要修改/etc/rsyslog.conf （CentOS7）的配置檔案：

在最後添加一行，指定遠端logstash伺服器位址和端口：

重新開機rsyslog服務：

<code>systemctl restart rsyslog</code>

在logstash伺服器上指定配置檔案啟動：

<code>/opt/logstash/bin/logstash</code> <code>-f </code><code>/etc/logstash/conf</code><code>.d</code><code>/syslog</code><code>.conf</code>

在node1使用logger 指令可以生成系統日志，使Elasticsearch上産生索引：

<code>logger trying</code>

登入Elasticsearch上就可查詢到syslog的日志了，可以在kibana上添加此項。

TCP、UDP日志收集

在有時要對一些日志進行補充或者發送一些檔案内容到es上時，可以通過使用TCP子產品的方式。

編輯tcp.conf的logstash檔案：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/tcp.conf </code>

<code> </code><code>mode => </code><code>"server"</code>

<code>output{</code>

<code> </code><code>stdout {</code>

<code> </code><code>codec => rubydebug</code>

啟動服務，在其他任何主機上可以通過nc 指令或者僞裝置輸出的方式，将需要發送的内容發送給node2:

<code>[root@node1 ~]</code><code># echo "trying"|nc 172.16.10.21 6666</code>

<code>[root@node1 ~]</code><code># echo "stuff" > /dev/tcp/172.16.10.21/6666</code>

在node2上，顯示接受的資訊：

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/tcp.conf </code>

<code>Settings: Default pipeline workers: 1</code>

<code>Pipeline main started</code>

<code> </code><code>"message"</code> <code>=> </code><code>"trying"</code><code>,</code>

<code> </code><code>"@version"</code> <code>=> </code><code>"1"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:12:10.630Z"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"# Generated by NetworkManager"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:13:11.081Z"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"nameserver 114.114.114.114"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"stuff"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:15:14.224Z"</code><code>,</code>

使用Filter子產品對Apache日志進行格式化

在logstash中可以對一些輸入的格式進行格式化，如記錄Apache日志時，由于自身不具有像nginx提供的JSON格式化的輸出，是以在進行記錄時，需要使用filter插件的grok來進行正則比對。

在logstash中，已經預設提供了友善使用的正則表示方式,在/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns儲存了相關的配置。

編寫logstash的配置檔案：

<code>[root@node2 patterns]</code><code># cat /etc/logstash/conf.d/grok.conf </code>

<code>input{</code>

<code> </code><code>stdin {}</code>

<code>filter { </code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"</code> <code>}</code>

<code> </code><code>stdout{</code>

啟動服務，輸入測試示例，傳回格式化的輸出：

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf </code>

<code>55.3.244.1 GET </code><code>/index</code><code>.html 15824 0.043 </code><code>#輸入的示例</code>

<code> </code><code>"message"</code> <code>=> </code><code>"55.3.244.1 GET /index.html 15824 0.043"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:59:44.264Z"</code><code>,</code>

<code> </code><code>"client"</code> <code>=> </code><code>"55.3.244.1"</code><code>,</code>

<code> </code><code>"method"</code> <code>=> </code><code>"GET"</code><code>,</code>

<code> </code><code>"request"</code> <code>=> </code><code>"/index.html"</code><code>,</code>

<code> </code><code>"bytes"</code> <code>=> </code><code>"15824"</code><code>,</code>

<code> </code><code>"duration"</code> <code>=> </code><code>"0.043"</code>

使用grok收集Apache日志到Elasticsearch.

在預設的Apache日志格式下，logstash提供了一個已經編寫好的日志格式正則filter:

<code>[root@node2 ~]</code><code># cat /opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/grok-patterns \</code>

<code>|</code><code>grep</code> <code>"COMBINEDAPACHELOG"</code>

<code>COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}</code>

編寫logstash配置檔案,引用此參數COMBINEDAPACHELOG：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/grok.conf </code>

<code> </code><code>path => </code><code>"/var/log/httpd/access_log"</code>

<code> </code><code>start_position => </code><code>"beginning"</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>} </code><code>#調用此參數</code>

<code> </code><code>hosts => [</code><code>"172.16.10.20:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"apache-access-log-%{+YYYY.MM.dd}"</code>

<code># stdout{ codec => rubydebug }</code>

這樣，預設的日志檔案就可以被格式化輸出到Elasticsearch上了。

使用grok會有一些問題，比如會非常影響性能，而且很不靈活，除非對ruby掌控很好。

在實際的生産環境中，為了實作松耦合和性能問題，一般将日志寫入redis,再使用python對其進行格式化處理。

使用消息隊列擴充ELK

資料 >> lostash >> MQ/Redis >> logstash >> ES

在實作低耦合和高可靠性的ELK部署架構中，使用消息隊列的方式來作為一個中間轉存的資料交換中心，将資料檔案需要寫入到消息隊列，而對消息隊列中資料的處理可有其它任意的應用去實作，整個過程寫入和輸出系統沒有任何依賴關系。

這裡使用最簡單的redis來實作這一功能。

在node2上安裝redis,将收集的資料發送到redis中，不做任何處理，這裡需要先修改redis的配置檔案：

<code># vim /etc/redis.conf </code>

<code>daemonize </code><code>yes</code> <code># 使用台運作模式</code>

<code>bind 172.16.10.21 </code><code># 綁定監聽的IP</code>

通過使用logstash将資料寫入redis,建立logstash的配置檔案，用于将資料寫入redis:

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/redis.conf </code>

<code> </code><code>stdin {</code>

<code> </code><code>redis {</code>

啟動redis,指定配置檔案啟動logstash:

<code>systemctl start redis</code>

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis.conf </code>

在redis上可以檢視到輸入的資料：

<code>[root@node2 ~]</code><code># redis-cli -h 172.16.10.21 -p 6379</code>

<code># Keyspace</code>

<code>db6:keys=2,expires=0,avg_ttl=0</code>

<code>172.16.10.21:6379> </code><code>select</code> <code>6</code>

<code>1) </code><code>"apache-accesslog"</code>

<code>172.16.10.21:6379[6]> lindex demo -1</code>

<code>"{\"message\":\"try\",\"@version\":\"1\",\"@timestamp\":\"2017-01-04T03:29:28.337Z\",\"host\":\"node2\"}"</code>

使用另一個logstash将資料從redis中取出：

在node1上配置logstash的配置檔案，指定檔案啟動，此時會自動将redis上的資料讀出：

<code>[root@node1 ~]</code><code># cat /etc/logstash/conf.d/redis-get.conf </code>

<code> </code><code>redis {</code>

<code>filter {}</code>

<code>[root@node1 conf.d]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis-get.conf </code>

<code> </code><code>"message"</code> <code>=> </code><code>"try"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T12:10:20.635Z"</code><code>,</code>

上面使用的是手動輸入的方式，在實際收集日志時，可以在輸入時指定檔案，輸出時指定到Elasticsearch。

使用Redis收集Apache通路日志

在上面的示例中，我們使用了直接通過grok來收集Apache的日志，但是在實際的生産環境，由于這種方案會使性能下降，而且在整個架構上過度依賴于logstash,無法做到解耦，并不是一個好的解決方案。這裡引入消息隊列，作為一個解耦，上日志的存取分開，友善後期擴充，提高了穩定性。

在node2上配置logstash 配置檔案，指定Apache的access檔案，并啟動：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/apache-redis.conf </code>

<code> </code><code>key => </code><code>"apache-log"</code>

此處直接完整輸入内容到redis,不做任何處理。

在node1上從redis中取出日志，并通過grok進行日志格式化處理，之後輸出到Elasticsearch中。

node1上logstash的配置：

<code>filter {</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>}</code>

<code> </code><code>elasticsearch{</code>

<code> </code><code>index => </code><code>"apache-log-%{+YYYY.MM.dd}"</code>

這樣，啟動node1和node2上的logstash,在指定的Elasticsearch上就可以看到被格式化後的日志輸出了。

ELK項目生産規劃

在實際的生産環境中要對日志進行收集，首先需要對日志分類：

通路日志： apache，nginx, tomcat 等web通路日志通過file插件收集，apache通過filter插件處理格式

錯誤日志：error log,java 日志等可以直接收取，對于一條多行的日志（java異常）使用多行處理插件

系統日志： /var/log syslog 等直接使用 syslog插件收取

網絡日志：防火牆，交換機，路由器等使用 syslog日志

運作日志：應用程式定義使用file 插件，最好為json格式

标準化：統一存放路徑，命名規則，格式（json）,日志切割（crontab）. 需要儲存的日志先在本地讀寫，提高性能，通過rsync推送到指定存儲，防止挂載故障而使服務終止。删除本地日志（确認保留時間）

工具化：可以使用logstash進行日志收集方案。

如果使用redis作為消息隊列，那麼要對所有的list key的長度進行監控，防止溢出和資料丢失

llen key_name

根據實際情況，例如超過 10萬就報警。

如将之前的所有配置合并到一處，在使用redis作為存儲時，可以這樣編輯logstash配置檔案：

從redis伺服器接受日志,處理後存入Elasticsearch：

<code> </code><code>syslog {</code>

<code> </code><code>redis {</code>

<code> </code><code>type</code> <code>=> </code><code>"apache-accesslog"</code>

<code> </code><code>key => </code><code>"apache-accesslog"</code>

<code> </code><code>redis {</code>

<code> </code><code>if</code> <code>[</code><code>type</code><code>] == </code><code>"apache-accesslog"</code> <code>{</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>}</code>

<code> </code><code>hosts => [</code><code>"192.168.56.11:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"apache-accesslog-%{+YYYY.MM.dd}"</code>

<code> </code><code>elasticsearch {</code>

<code> </code><code>hosts => [</code><code>"192.168.56.11:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"es-log-%{+YYYY.MM}"</code>

<code> </code><code>if</code> <code>[</code><code>type</code><code>] == </code><code>"system-syslog"</code> <code>{</code>

<code> </code><code>index => </code><code>"system-syslog-%{+YYYY.MM}"</code>

從伺服器接收日志，将日志存入redis/MQ:

<code> </code><code>path => </code><code>"/var/log/elasticsearch/elasticsearch.log"</code>

<code> </code><code>codec => multiline{</code>

<code> </code><code>pattern => </code><code>"^\["</code>

<code> </code><code>negate => </code><code>true</code>

<code> </code><code>what => </code><code>"previous"</code>

<code> </code><code>key => </code><code>"apache-accesslog"</code>

這裡使用了if做為了條件的判斷，來對不同的日志進行選擇處理，在一些系統日志如syslog等，可以單獨進行處理。

提示：在使用系統自帶的腳本進行啟動時，會出現無法收集日志的情況，腳本中預設的是使用logstash使用者，在收集一些日志時可能存在沒有權限的情況。可以修改腳本中的預設使用者，或者給logstash添加其他使用者組的權限。

本文轉自酥心糖 51CTO部落格，原文連結:http://blog.51cto.com/tryingstuff/1888978

ELK日志管理深度實戰

繼續閱讀

httpd dead but subsys locked;No space left on device:Couldn't create accept loc

apache (httpd)不支援中文路徑問題先解除安裝yum安裝的httpd再用源碼安裝，重裝httpd再安裝支援中文的插件遇到問題

搭建httpd服務

windows下配置Apache的vhost初次接觸，強烈歡迎拍磚，指出錯誤

Apache與PHP環境下配置本地虛拟主機

Linux 7 中配置Apache服務，及禁止ip通路，删除apache廣告頁面。

Apache配置檔案中的deny和allow的使用

Apache 配置預設編碼

伺服器配置——Apache

Apache靜态檔案通路配置（書封伺服器）

apache httpd 配置

Ubuntu16.04安裝Apache+MySQL+PHP1. 安裝Apache2. 安裝MySQL3. 安裝PHP4. 安裝phpMyAdmin

Apache配置SSLApache配置SSL

Windows下配置Apache的SSL服務

Apache2.4.x 配置檔案詳解Apache配置需要了解如下：開始講解：

配置apache支援PHP（win7）