filebeat + logstash + elasticsearch + granfa
filebeat + logstash + elasticsearch + granfa
https://www.cnblogs.com/wenchengxiaopenyou/p/9034213.html
一。背景
前端web服務器為nginx，采用filebeat + logstash + elasticsearch + granfa 進行數據采集與展示，對客戶端ip進行地域統計，監控服務器響應時間等。二。業務整體架構：
nginx日誌落地——》filebear——》logstash——》elasticsearch——》grafna（展示）三。先上個效果圖，慢慢去一步步實現
如上只是簡單的幾個實用的例子，實際上有多維度的數據之後還可以定制很多需要的內容，如終端ip訪問數，國家、地區占比，訪問前十的省份，請求方法占比，referer統計，user_agent統計，慢響應時間統計，更有世界地圖坐標展示等等，只要有數據就能多維度進行展示。這裏提供模板搜索位置大家可以查找參考：https://grafana.com/dashboards
四，準備條件
需要具備如下條件：
1.nginx日誌落地，需要主要落地格式，以及各個字段對應的含義。
2.安裝filebeat。 filebeat輕量，功能相比較logstash而言比較單一。
3.安裝logstash 作為中繼服務器。這裏需要說明一下的是，起初設計階段並沒有計劃使用filebeat，而是直接使用logstash發往elasticsearch，但是當前端機數量增加之後logstash數量也隨之增加，同時發往elasticsearch的數量增大，logstash則會拋出由於elasticsearch 限制導致的錯誤，大家遇到後搜索相應錯誤的代碼即可。為此只采用logstash作為中繼。
4.elasticsearch 集群。坑點是index templates的創建會影響新手操作 geoip模塊。後文會有。
5.grafana安裝，取代傳統的kibana，grafana有更友好、美觀的展示界面。
五。實現過程
1.nginx日誌落地配置
nginx日誌格式、字段的內容和順序都是高度可定制化的，將需要收集的字段內容排列好。定義一個log_format
定義的形勢實際上直接決定了logstash配置中對於字段抽取的模式，這裏有兩種可用，一種是直接在nginx日誌中拼接成json的格式，在logstash中用codec => "json"來轉換，
一種是常規的甚至是默認的分隔符的格式，在logstash中需要用到grok來進行匹配，這個會是相對麻煩些。兩種方法各有優點。直接拼接成json的操作比較簡單，但是在轉碼過程中
會遇到諸如 \x 無法解析的情況。 這裏我也遇到過，如有必要後續會詳談。采用grok來匹配的方法相對難操作，但是準確性有提升。我們這裏采用的是第一種方法，下面logstash部分
也會給出采用grok的例子。 nginx日誌中日誌格式定義如下：
1
2
3
4
5
6
7
8
9
10
log_format access_json ‘{"timestamp":"$time_iso8601",‘
‘"hostname":"$hostname",‘
‘"ip":"$remote_addrx",‘
‘"request_method":"$request_method",‘
‘"domain":"XXXX",‘
‘"size":$body_bytes_sent,‘
‘"status": $status,‘
‘"responsetime":$request_time,‘
‘"sum":"1"‘
‘}‘;
2.filebeat配置文件
關於filebeat更多內容請參考https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-overview.html
配置文件內容：filebeat.yml 這裏應該不會遇到坑。
1
2
3
4
5
6
7
filebeat.prospectors:
- input_type: log
paths:- /data0/logs/log_json/*.log #nginx日誌路徑
output.logstash:
hosts: ["xxxx.xxxx.xxxx.xxx:12121"] #logstash 服務器地址
3.logstahs配置文件內容：
這裏是針對json已經拼接號，直接進行json轉碼的情況：
需要註意如下：
1）date模塊必須有，否則會造成數據無法回填導致最終的圖像出現鋸齒狀影響穩定性（原因是排列時間並不是日誌產生的時間，而是進入logstash的時間）。這裏後面的 yyyy-MM-dd‘T‘HH:mm:ssZZ 需要根據你日誌中的日期格式進行匹配匹配規則見：https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html
2）需要說明的是我下面去掉了好多的字段（remove_field），原因是我們數據量大，es服務器有限。 可以根據需要隨時調整收集的字段。
3) geoip 僅需要指定源ip字段名稱即可，fields並不是必須的，我加入的原因還是由於資源有限導致的。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
input {
beats {
port => 12121
host => "10.13.0.80"
codec => "json"
}
}
filter {
date {
match => [ "timestamp", "yyyy-MM-dd‘T‘HH:mm:ssZZ" ]
#timezone => "Asia/Shanghai"
#timezone => "+00:00"
}
mutate {
convert => [ "status","integer" ]
convert => [ "sum","integer" ]
convert => [ "size","integer" ]
remove_field => "message"
remove_field => "source"
remove_field => "tags"
remove_field => "beat"
remove_field => "offset"
remove_field => "type"
remove_field => "@source"
remove_field => "input_type"
remove_field => "@version"
remove_field => "host"
remove_field => "client"
#remove_field => "request_method"
remove_field => "size"
remove_field => "timestamp"
#remove_field => "domain"
#remove_field => "ip"
}
geoip {
source => "ip"
fields => ["country_name", "city_name", "timezone","region_name","location"]
}
}
output {
elasticsearch {
hosts => ["xxx:19200","xxx:19200","xxx:19200"]
user => "xxx"
password => "xxx"
index => "logstash-suda-alllog-%{+YYYY.MM.dd}"
flush_size => 10000
idle_flush_time => 35
}
}
下面給出一個采用grok的例子：
其中match內的內容不用參考我的，需要根據你的字段個數，以及格式來定制。 這裏其實是正則表達式。自帶了一部分幾乎就可以滿足所有的需求了無需自己寫了，可以參考：https://github.com/elastic/logstash/blob/v1.4.0/patterns/grok-patterns 直接用即可。復制代碼
input {
file {
type => "access"
path => ["/usr/local/nginx/logs/main/*.log"]
}
}
filter {
if [type] == "access" {
if [message] =~ "^#" {
drop {}
}
grok {
match => ["message", "[%{HTTPDATE:log_timestamp}] %{HOSTNAME:server_name} "%{WORD:request_method} %{NOTSPACE:query_string} HTTP/%{NUMBER:httpversion}" "%{GREEDYDATA:http_user_agent}" %{NUMBER:status} %{IPORHOST:server_addr} "%{IPORHOST:remote_addr}" "%{NOTSPACE:http_referer}" %{NUMBER:body_bytes_sent} %{NUMBER:time_taken} %{GREEDYDATA:clf_body_bytes_sent} %{NOTSPACE:uri} %{NUMBER:m_request_time}"]
}
date {
match => [ "log_timestamp", "dd/MMM/yyyy:mm:ss:SS Z" ]
timezone => "Etc/UTC"
}
mutate {
convert => [ "status","integer" ]
convert => [ "body_bytes_sent","integer" ]
convert => [ "m_request_time","float" ]
}
復制代碼
在提供一個高級的用法：
ruby：強大的模塊， 可以進行諸如時間轉換，單位計算等，多個可以用分號隔開。
復制代碼
ruby {
code => "event.set(‘logdateunix‘,event.get(‘@timestamp‘).to_i);event.set(‘request_time‘, event.get(‘m_request_time‘) / 1000000 )"
}
mutate {
add_field => {
"http_host" => "%{server_name}"
"request_uri" => "%{uri}"
}
復制代碼
在windowns上使用lgostsh需要註意的是：（win上收集iis的日誌，我想正常環境是不會用到的，但是我確實用到了。。。。。。）
path路徑一定要采用 linux中的分割符來拼接路徑，用win的格式則正則不能實現，大家可以測試下。其他配置則無區別。
復制代碼
input {
file {
type => "access"
path => ["C:/WINDOWS/system32/LogFiles/W3SVC614874788/*.log"]
}
}
復制代碼
4.elasticsearch配置，集群的安裝以及啟動，調優這裏不多說（說不完），需要註意的一個是，geoip location的格式，我這裏采用的是index templates來實現的如下：
最重要的是 "location" 定義 （否則geoip_location字段格式有問題，無法拼接成坐標），其他可以根據情況自定：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
{
"order": 0,
"version": 50001,
"index_patterns": [
"suda-*"
],
"settings": {
"index": {
"number_of_shards": "5",
"number_of_replicas": "1",
"refresh_interval": "200s"
}
},
"mappings": {
"default": {
"dynamic_templates": [
{
"message_field": {
"path_match": "message",
"mapping": {
"norms": false,
"type": "text"
},
"match_mapping_type": "string"
}
},
{
"string_fields": {
"mapping": {
"norms": false,
"type": "text",
"fields": {
"keyword": {
"ignore_above": 256,
"type": "keyword"
}
}
},
"match_mapping_type": "string",
"match": "*"
}
}
],
"properties": {
"@timestamp": {
"type": "date"
},
"geoip": {
"dynamic": true,
"properties": {
"ip": {
"type": "ip"
},
"latitude": {
"type": "half_float"
},
"location": {
"type": "geo_point"
},
"longitude": {
"type": "half_float"
}
}
},
"@version": {
"type": "keyword"
}
}
}
},
"aliases": {}
}
5 grafana 安裝以及模板創建，這個比較簡單，安裝完直接寫語句即可附一個例子如下：
這裏的變量需要自定義：
通過上面應該能夠完成一個完整的收集和展示情況，這裏實際上提供了一種可行的方法，並麽有太大的具體操作。
希望多多交流
推薦內容：kibana中文指南（有ibook版本，看著挺方便） 三鬥室著
https://www.elastic.co/cn/
https://grafana.com/dashboards
https://grafana.com/
filebeat + logstash + elasticsearch + granfa