#頭條創作挑戰賽#

　　采集Apache Web服務器日志一、需求 Apache的Web Server會産生大量日志，當我們想要對這些日志檢索分析。就需要先把這些日志導入到Elasticsearch中。此處，我們就可以使用Logstash來實現日志的采集

　　打開這個文件，如下圖所示。我們發現，是一個純文本格式的日志。如下圖所示：

　　這個日志其實由一個個的字段拼接而成，參考以下表格

　　二、準備日志數據 将Apache服務器日志上傳到 /export/server/es/data/apache/ 目錄

　　mkdir -p /export/server/es/data/apache/

　　三、使用FileBeats将日志發送到Logstash 在使用Logstash進行數據解析之前，我們需要使用FileBeat将采集到的數據發送到Logstash。之前，我們使用的FileBeat是通過FileBeat的Harvester組件監控日志文件，然後将日志以一定的格式保存到Elasticsearch中，而現在我們需要配置FileBeats将數據發送到Logstash。FileBeat這一端配置以下即可：

　　#----------------------------- Logstash output --------------------------------- #output.logstash: # Boolean flag to enable or disable the output module. #enabled: true # The Logstash hosts #hosts: [localhost:5044]

　　hosts配置的是Logstash監聽的IP地址/機器名以及端口号。

　　準備FileBeat配置文件

　　cd /export/server/es/filebeat-7.6.1-linux-x86_64 vim filebeat-logstash.yml

　　因為Apache的web log日志都是以IP地址開頭的，所以我們需要修改下匹配字段

　　filebeat.inputs: - type: log enabled: true paths: - /var/apache/log/access.* multiline.pattern: ^\d \.\d \.\d \.\d multiline.negate: true multiline.match: after output.logstash: enabled: true hosts: [node1:5044]

　　啟動FileBeat，并指定使用新的配置文件

　　./filebeat -e -c filebeat-logstash.yml

　　FileBeat将嘗試建立與Logstash監聽的IP和端口号進行連接。但此時，我們并沒有開啟并配置Logstash，所以FileBeat是無法連接到Logstash的。

　　2021-12-05T11:28:47.585 0800 ERROR pIPeline/output.go:100 Failed to connect to backoff(async(tcp://node1.itcast.cn:5044)): dial tcp 192.168.88.100:5044: connect: connection refused

　　四、配置Logstash接收FileBeat數據并打印 Logstash的配置文件和FileBeat類似，它也需要有一個input、和output。基本格式如下：

　　# #号表示添加注釋 # input表示要接收的數據 input { } # file表示對接收到的數據進行過濾處理 Filter { } # output表示将數據輸出到其他位置 output { }

　　配置從FileBeat接收數據

　　cd /export/server/es/logstash-7.6.1/config vim filebeat-prINT.conf

　　input { beats { port = 5044 } } output { stdout { codec = rubydebug } }

　　測試logstash配置是否正确

　　bin/logstash -f config/filebeat-print.conf --config.test_and_exit

　　[2021-12-05T11:46:33,940][INFO ][logstash.runner ] Using config.test_and_exit mode. Config Validation Result: OK. Exiting Logstash

　　啟動logstash

　　bin/logstash -f config/filebeat-print.conf --config.reload.automatic

　　reload.automatic：修改配置文件時自動重新加載

　　測試

　　創建一個access.log.1文件，使用cat test access.log.1往日志文件中追加内容。

　　test文件中隻保存一條日志：

　　[root@node1 log]# cat test 235.9.200.242 - - [15/Apr/2015:00:27:19 0849] POST /it.cn/bigdata.html 200 45 www.baidu.comMozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249

　　當我們啟動Logstash之後，就可以發現Logstash會打印出來從FileBeat接收到的數據：

　　{ log { file { path /var/apache/log/access.log.1 }, offset 825 }, input { type log }, agent { ephemeral_id d4c3b652-4533-4ebf-81f9-a0b78c0d4b05, version 7.6.1, type filebeat, =b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64, hostname node1.itcast.cn }, @timestamp 2021-12-05T09:07:55.236Z, ecs { version 1.4.0 }, host { name node1 }, tags [ [0] beats_input_codec_plain_applied ], message 235.9.200.242 - - [15/Apr/2015:00:27:19 0849] \POST /it.cn/bigdata.html 200 45 \www.baidu.com\Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249, @version }

　　五、Logstash輸出數據到Elasticsearch 通過控制台，我們發現Logstash input接收到的數據沒有經過任何處理就發送給了output組件。而其實我們需要将數據輸出到Elasticsearch。所以，我們修改Logstash的output配置。配置輸出Elasticsearch隻需要配置以下就可以了：

　　output { elasticsearch { hosts =localhost:9200 ] }}

　　操作步驟：

　　1、重新拷貝一份配置文件 cp filebeat-print.conf filebeat-es.conf

　　2、将output修改為Elasticsearch input { beats { port = 5044 } } output { elasticsearch { hosts =node1:9200node2:9200node3:9200] } }

　　3、重新啟動Logstash bin/logstash -f config/filebeat-es.conf --config.reload.automatic

　　4、追加一條日志到監控的文件中，并查看Elasticsearch中的索引、文檔 cat test access.log.1

　　// 查看索引數據

　　GET /_cat/indices?v

　　我們在Elasticsearch中發現一個以logstash開頭的索引

　　{ healthgreen, statusopen, indexlogstash-2021.12.05-000001, uuid147Uwl1LRb-HMFERUyNEBw, pri1, rep1, docs.count2, docs.deleted0, store.size44.8kb, pri.store.size22.4kb }

　　// 查看索引庫的數據

　　GET /logstash-2021.12.05-000001/_search?format=txt

　　{ from: 0, size: 1 }

　　我們可以獲取到以下數據：

　　@timestamp2021-12-05T09:38:00.402Z, tags: [ beats_input_codec_plain_applied ], host: { namenode1 }, @version1, log: { file: { path/var/apache/log/access.log.1 }, offset: 1343 }, agent: { version7.6.1, ephemeral_idd4c3b652-4533-4ebf-81f9-a0b78c0d4b05, : b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64, hostnamenode1, typefilebeat }, input: { typelog }, message235.9.200.242 - - [15/Apr/2015:00:27:19 0849] \POST /it.cn/bigdata.html 200 45 \www.baidu.com\Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249, ecs: { version1.4.0 }

　　從輸出返回結果，我們可以看到，日志确實已經保存到了Elasticsearch中，而且我們看到消息數據是封裝在名為message中的，其他的數據也封裝在一個個的字段中。我們其實更想要把消息解析成一個個的字段。例如：IP字段、時間、請求方式、請求URL、響應結果，這樣

　　六、Logstash過濾器 在Logstash中可以配置過濾器Filter對采集到的數據進行中間處理，在Logstash中，有大量的插件供我們使用。參考官網：

　　[Filter plugins | Logstash Reference 7.6] | Elastic

　　此處，我們重點來講解Grok插件。

　　1、查看Logstash已經安裝的插件 bin/logstash-plugin list

　　2、Grok插件 Grok是一種将非結構化日志解析為結構化的插件。這個工具非常适合用來解析系統日志、Web服務器日志、MySQL或者是任意其他的日志格式。

　　Grok官網：[Grok filter plugin | Logstash Reference 7.6] | Elastic

　　3、Grok語法 Grok是通過模式匹配的方式來識别日志中的數據,可以把Grok插件簡單理解為升級版本的正則表達式。它擁有更多的模式，默認，Logstash擁有120個模式。如果這些模式不滿足我們解析日志的需求，我們可以直接使用正則表達式來進行匹配。

　　官網：htt match =message %{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} } } }

　　七、匹配日志中的IP、日期并打印 235.9.200.242 - - [15/Apr/2015:00:27:19 0849] POST /it.cn/bigdata.html 200 45 www.baidu.comMozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249

　　我們使用IP就可以把前面的IP字段匹配出來，使用HTTPDATE可以将後面的日期匹配出來

　　配置Grok過濾插件

　　1、配置Logstash input { beats { port = 5044 } } filter { grok { match = { message %{IP:ip} - - \[%{HTTPDATE:date}\] } } } output { stdout { codec = rubydebug } }

　　2、啟動Logstash bin/logstash -f config/filebeat-filter-print.conf --config.reload.automatic

　　{ log { offset 1861, file { path /var/apache/log/access.log.1 } }, input { type log }, tags [ [0] beats_input_codec_plain_applied ], date 15/Apr/2015:00:27:19 0849, ecs { version 1.4.0 }, @timestamp 2021-12-05T11:02:05.809Z, message 235.9.200.242 - - [15/Apr/2015:00:27:19 0849] \POST /it.cn/bigdata.html 200 45 \www.baidu.com\Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900 144.180.122.249, host { name node1 }, =235.9.200.242, agent { hostname node1, version 7.6.1, ephemeral_id d4c3b652-4533-4ebf-81f9-a0b78c0d4b05, =b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64, type filebeat }, @version }

　　我們看到，經過Grok過濾器插件處理之後，我們已經獲取到了ip和date兩個字段。接下來，我們就可以繼續解析其他的字段

　　八、解析所有字段 将日志解析成以下字段：

　　2、測試并啟動Logstash 我們可以看到，8個字段都已經成功解析。

　　{ reference www.baidu.com, @version , ecs { version 1.4.0 }, @timestamp 2021-12-05T03:30:10.048Z, =235.9.200.241, method POST, uri /it.cn/bigdata.html, agent { =b4c5c4dc-03c3-4ba4-9400-dc6afcb36d64, ephemeral_id 734ae9d8-bcdc-4be6-8f97-34387fcde972, version 7.6.1, hostname node1, type filebeat }, length , status 200, log { file { path /var/apache/log/access.log }, offset 1 }, input { type log }, host { name node1 }, tags [ [0] beats_input_codec_plain_applied ], browser Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900, date 15/Apr/2015:00:27:19 0849, message 235.9.200.241 - - [15/Apr/2015:00:27:19 0849] \POST /it.cn/bigdata.html HTTP/1.1\ 200 45 \www.baidu.com\Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900\ }

　　九、将數據輸出到Elasticsearch 到目前為止，我們已經通過了Grok Filter可以将日志消息解析成一個一個的字段，那現在我們需要将這些字段保存到Elasticsearch中。我們看到了Logstash的輸出中，有大量的字段，但如果我們隻需要保存我們需要的8個，該如何處理呢？而且，如果我們需要将日期的格式進行轉換，我們又該如何處理呢？

　　1、過濾出來需要的字段 要過濾出來我們需要的字段。我們需要使用mutate插件。mutate插件主要是作用在字段上，例如：它可以對字段進行重命名、删除、替換或者修改結構。

　　官方文檔：[Mutate filter plugin | Logstash Reference 7.6] | Elastic

　　例如，mutate插件可以支持以下常用操作

　　配置：

　　注意：此處為了方便進行類型的處理，将status、length指定為int類型

　　input { beats { port = 5044 } } filter { grok { match = { message %{IP:ip} - - \[%{HTTPDATE:date}\] \%{WORD:method} %{PATH:uri} %{DATA}\ %{INT:status:int} %{INT:length:int} \%{DATA:reference}\%{DATA:browser}\ } } mutate { enable_metric =false remove_field =messagelogtags@timestampinputagenthostecs@version] } } output { stdout { codec = rubydebug } }

　　2、轉換日期格式 要将日期格式進行轉換，我們可以使用Date插件來實現。該插件專門用來解析字段中的日期，官方說明文檔：[Date filter plugin | Logstash Reference 7.6] | Elastic

　　用法如下：

　　将date字段轉換為「年月日 時分秒」格式。默認字段經過date插件處理後，會輸出到@timestamp字段，所以，我們可以通過修改target屬性來重新定義輸出字段。

　　Logstash配置修改為如下：

　　input { beats { port = 5044 } } filter { grok { match = { message %{IP:ip} - - \[%{HTTPDATE:date}\] \%{WORD:method} %{PATH:uri} %{DATA}\ %{INT:status:int} %{INT:length:int} \%{DATA:reference}\%{DATA:browser}\ } } mutate { enable_metric =false remove_field =messagelogtags@timestampinputagenthostecs@version] } date { match =datedd/MMM/yyyy:HH:mm:ss Zyyyy-MM-dd HH:mm:ss] target =date } } output { stdout { codec = rubydebug } }

　　啟動Logstash：

　　bin/logstash -f config/filebeat-filter-print.conf --config.reload.automatic

　　{ status 200, reference www.baidu.com, method POST, browser Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/53.0.2785.104 Safari/537.36 Core/1.53.4549.400 QQBrowser/9.7.12900, =235.9.200.241, length , uri /it.cn/bigdata.html, date 2015-04-14T15:38:19.000Z }

　　3、輸出到Elasticsearch指定索引 我們可以通過

　　elasticsearch { hosts =node1:9200node2:9200node3:9200] index =xxx }

　　index來指定索引名稱，默認輸出的index名稱為：logstash-%{ yyyy.MM.dd}。但注意，要在index中使用時間格式化，filter的輸出必須包含 @timestamp字段，否則将無法解析日期。

　　input { beats { port = 5044 } } filter { grok { match = { message %{IP:ip} - - \[%{HTTPDATE:date}\] \%{WORD:method} %{PATH:uri} %{DATA}\ %{INT:status:int} %{INT:length:int} \%{DATA:reference}\%{DATA:browser}\ } } mutate { enable_metric =false remove_field =messagelogtagsinputagenthostecs@version] } date { match =datedd/MMM/yyyy:HH:mm:ss Zyyyy-MM-dd HH:mm:ss] target =date } } output { stdout { codec = rubydebug } elasticsearch { hosts =node1:9200node2:9200node3:9200] index =apache_web_log_%{ YYYY-MM} } }

　　啟動Logstash

　　bin/logstash -f config/filebeat-apache-weblog.conf --config.test_and_exit bin/logstash -f config/filebeat-apache-weblog.conf --config.reload.automatic

　　注意：

　　index名稱中，不能出現大寫字符,