ELK+K8S
ELK+K8Scss
| 做者html |
劉暢java |
| 時間node |
2021-07-26linux |
環境: CentOS7.5nginx
| 主機名稱git |
IPgithub |
軟件web |
| controlnode正則表達式 |
172.16.1.120 [4核8G] |
kafka環境[zookeeper-3.7.0、kafka_2.13-2.8.0、jdk1.8.0_45] ELK環境[nginx、tomcat、ES-7.13.4、ElasticHD、logstash-7.13.4、filebeat-7.13.4、kibana-7.13.4] |
| slavenode1 |
172.16.1.121 [2核4G] |
kafka環境[zookeeper-3.7.0、kafka_2.13-2.8.0、jdk1.8.0_45] ELK環境[ES-7.13.4、logstash-7.13.4] |
| slavenode2 |
172.16.1.122 [2核4G] |
kafka環境[zookeeper-3.7.0、kafka_2.13-2.8.0、jdk1.8.0_45] ELK環境[ES-7.13.4] |
| k8s-admin |
172.16.1.70 [2核4G] |
K8S控制端 |
| k8s-node1 |
172.16.1.71 [4核8G] |
k8s slave端01 |
| k8s-node2 |
172.16.1.72 [4核8G] |
k8s slave端02 |
(1) 注: 本文檔不記錄zookeeper+kafka、k8s的搭建過程。
(2) ELK官方文檔: https://www.elastic.co/guide/index.html
(3) ELK下載地址: https://www.elastic.co/downloads/
(4) elasticsearch、logstash自帶jdk環境。
(5) elk屬於比較消耗資源的服務，不建議採用docker方式部署，docker主要是簡化部署，有助於後面的升級。該文檔主要採用二進制方式部署，也能夠採用rpm包的方式部署。
(6) 測試文件建立
# touch /tmp/test.log && chmod 777 /tmp/test.log
# touch /tmp/prod.log && chmod 777 /tmp/prod.log
# touch /tmp/unknown.log && chmod 777 /tmp/unknown.log
# touch /tmp/out_result.txt && chmod 777 /tmp/out_result.txt
# touch /tmp/filebeat_out.txt && chmod 777 /tmp/filebeat_out.txt
目錄
2.3 將Elasticsearch服務加入systemd 3
3.5 輸入插件input(輸入階段，從哪裏獲取日誌) 12
3.6 過濾插件filter(過濾階段，將日誌格式化處理) 15
3.7 輸出插件output(將處理完成的日誌推送到遠程數據庫存儲) 24
8 Filebeat->Logstash->Kafka->Logstash->ES->Kibana 53
9.3 kibana、logstash帶密碼方式訪問elasticsearch 68
9.8 elasticsearch索引分片及副本的設置 81
1 ELK介紹
1.1 需求背景
1 業務發展愈來愈龐大，服務器愈來愈多。
2 各類訪問日誌、應用日誌、錯誤日誌量愈來愈多。
3 開發人員排查問題，須要到服務器上查日誌，效率低、權限很差控制。
4 運維需實時關注業務訪問狀況。
1.2 ELK介紹
1 ELK是三個開源軟件的縮寫，提供一套完整的企業級日誌平臺解決方案。
2 ELK分別是
(1) Elasticsearch # 搜索、分析和存儲數據。
(2) Logstash # 採集日誌、格式化、過濾，最後將數據推送到Elasticsearch存儲。
(3) Kibana # 數據可視化。
3 Beats
# 集合了多種單一用途數據採集器，用於實現從邊緣機器向Logstash和Elasticsearch發送數據。是應用最多的，是一個輕量級日誌採集器。
1.3 ELK架構
2 Elasticsearch集羣部署
在172.16.1.120、121、122節點上操做
2.1 Elasticsearch介紹
1 Elasticsearch(簡稱ES)是一個分佈式、RESTful風格的搜索和數據分析引擎，用於集中存儲日誌數據。
2 Elasticsearch術語
(1) Index # 索引是多個文檔的集合。
(2) Document # Index裏每條記錄稱爲Document，若干文檔構建一個Index。
(3) Type # 一個Index能夠定義一種或多種類型，將Document邏輯分組。
(4) Field # ES存儲的最小單元。
3 ES與關係型數據庫術語對比
| Elasticsearch |
關係型數據庫 |
| Index |
Database |
| Type |
Table |
| Document |
Row |
| Filed |
Colume |
2.2 安裝
1 下載二進制包
https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.13.4-linux-x86_64.tar.gz
2 修改配置文件
# tar -xzf elasticsearch-7.13.4-linux-x86_64.tar.gz
# mv elasticsearch-7.13.4/ /usr/local/elasticsearch/
# mkdir -p /usr/local/elasticsearch/logs
# mkdir -p /usr/local/elasticsearch/data
# useradd -M -s /sbin/nologin elasticsearch
# id elasticsearch
uid=1002(elasticsearch) gid=1002(elasticsearch) groups=1002(elasticsearch)
# chown -R elasticsearch.elasticsearch /usr/local/elasticsearch/
# grep "^[a-Z]" /usr/local/elasticsearch/config/elasticsearch.yml
cluster.name: elk-cluster
# 集羣名稱
node.name: es01
# 集羣節點名稱，172.16.1.120節點爲es01，172.16.1.121節點爲es02，172.16.1.122節點爲es03
path.data: /usr/local/elasticsearch/data
# 數據目錄
path.logs: /usr/local/elasticsearch/logs
# 日誌目錄
bootstrap.memory_lock: true
# 鎖定物理內存地址，防止es內存和swap進行交換
network.host: 172.16.1.120
# 監聽地址，三個節點分別爲172.16.1.120，172.16.1.121，172.16.1.122
http.port: 9200
# 監聽地址端口
transport.tcp.port: 9300
# 集羣內部節點之間通訊端口
discovery.seed_hosts: ["172.16.1.120", "172.16.1.121", "172.16.1.122"]
# 集羣節點列表
cluster.initial_master_nodes: ["172.16.1.120", "172.16.1.121", "172.16.1.122"]
# 首次啓動指定的Master節點，這裏指定全部node節點，會自動選舉
http.cors.enabled: true
http.cors.allow-origin: "*"
# 開啓elasticsearch跨域訪問功能
2.3 將Elasticsearch服務加入systemd
1 調整進程最大打開文件數數量
# ulimit -SHn 65535
# cat >> /etc/security/limits.conf << EOF
* soft nofile 65535
* hard nofile 65535
EOF
2 調整進程最大虛擬內存區域數量
# echo "vm.max_map_count=262144" >> /etc/sysctl.conf
# sysctl -p
3 將jvm堆大小設置爲大約可用內存的一半
# vim /usr/local/elasticsearch/config/jvm.options
-Xms512m
-Xmx512m
4 elasticsearch.service文件
# vim /usr/lib/systemd/system/elasticsearch.service
[Unit]
Description=elasticsearch server
Requires=network.target
After=network.target
[Service]
# 不限制內存大小
LimitMEMLOCK=infinity
# 指定此進程能夠打開的最大文件描述符
LimitNOFILE=65535
# 指定最大進程數
LimitNPROC=4096
# 不限制虛擬內存大小
LimitAS=infinity
# 不限制最大文件大小
LimitFSIZE=infinity
# 當JVM接收到SIGTERM信號時，它將退出，並返回代碼143
SuccessExitStatus=143
Type=simple
Environment=JAVA_HOME=/usr/local/elasticsearch/jdk
ExecStart=/usr/local/elasticsearch/bin/elasticsearch
ExecReload=/bin/kill -HUP $MAINPID
KillMode=process
Restart=on-failure
User=elasticsearch
Group=elasticsearch
[Install]
WantedBy=multi-user.target
5 啓動服務
# systemctl daemon-reload
# systemctl start elasticsearch.service
# systemctl enable elasticsearch.service
6 查看集羣信息
(1) 查看端口信息
# netstat -tunlp
tcp6 0 0 172.16.1.120:9200 :::* LISTEN 3177/java
tcp6 0 0 172.16.1.120:9300 :::* LISTEN 3177/java
(2) 查看集羣狀態
1) 查看集羣節點信息
[root@controlnode tools]# curl -XGET 'http://172.16.1.120:9200/_cat/nodes?pretty'
172.16.1.120 54 35 2 0.02 0.25 0.24 cdfhilmrstw * es01
172.16.1.121 17 96 5 0.05 0.59 0.51 cdfhilmrstw - es02
172.16.1.122 24 97 5 0.07 0.56 0.49 cdfhilmrstw - es03
2) 查詢集羣健康狀態
[root@controlnode tools]# curl -i -XGET http://172.16.1.120:9200/_cluster/health?pretty
3) 查看指定es節點基本信息
[root@controlnode ~]# curl -XGET '172.16.1.120:9200/?pretty'
2.4 圖形頁面管理ES
在172.16.1.120節點上操做
1 下載
推薦插件: ElasticHD、cerebro、elasticsearch-head，這裏使用ElasticHD
https://github.com/360EntSecGroup-Skylar/ElasticHD/releases/download/1.4/elasticHD_linux_amd64.zip
2 啓動
# unzip -q elasticHD_linux_amd64.zip
# mv ElasticHD /usr/bin/
# nohup ElasticHD -p 172.16.1.120:9800 >/dev/null 2>&1 &
# chmod +x /etc/rc.d/rc.local
# cat >>/etc/rc.local<< EOF
source /etc/profile
nohup ElasticHD -p 172.16.1.120:9800 >/dev/null 2>&1 &
EOF
3 訪問
URL: http://172.16.1.120:9800/
4 補充: cerebro的使用
(1) 下載
https://github.com/lmenezes/cerebro/releases/download/v0.9.4/cerebro-0.9.4.zip
(2) 安裝
# unzip -q cerebro-0.9.4.zip
# mv cerebro-0.9.4/ /usr/local/cerebro/
(3) 啓動
# nohup /usr/local/cerebro/bin/cerebro -Dhttp.port=9000 >/dev/null 2>&1 &
# chmod +x /etc/rc.d/rc.local
# cat >>/etc/rc.local<< EOF
source /etc/profile
nohup /usr/local/cerebro/bin/cerebro -Dhttp.port=9000 >/dev/null 2>&1 &
EOF
# netstat -tunlp | grep 9000
tcp6 0 0 :::9000 :::* LISTEN 2553/java
(4) 訪問
URI: http://172.16.1.120:9000
概要信息:
3 Logstash部署
在172.16.1.120節點上操做
3.1 Logstash介紹
Logstash可以將採集日誌、格式化、過濾，最後將數據推送到Elasticsearch存儲。
Logstash不僅是一個input|filter|output的數據流，而是一個input|decode|filter|encode|output的數據流。codec就是用來decode，encode 事件的。因此codec經常使用在input和output中，經常使用的codec插件有plain，json，multiline等。
| 服務名 |
logstash input默認解碼方式 |
logstash output默認編碼方式 |
| redis |
json |
json |
| kafka |
plain |
plain |
| file |
plain |
json_lines |
| beats |
plain |
不支持 |
| tcp |
line |
json |
| elasticsearch |
json |
- |
| rabbitmq |
json |
json |
| stdin |
line |
不支持 |
| stdout |
不支持 |
rubydebug |
(1) json
1) 此編解碼器可用於解碼(經過輸入)和編碼(經過輸出)完整的JSON消息。若是發送的數據是其根處的JSON 數組，則將建立多個事件(每一個元素一個）。若是您正在流式傳輸由"\n"分隔的JSON消息，請參閱"json_lines"編解碼器。
2) 編碼將產生一個緊湊的JSON表示(沒有行終止符或縮進)，若是此編解碼器從無效JSON的輸入中收到有效負載，則它將回退到純文本並添加標籤"_jsonparsefailure"。JSON失敗時，有效負載將存儲在該message字段中。
3) "charset"默認值爲"UTF-8"。
4) 解碼和編碼都是json，json數據不添加任何內容。
(2) plain
1) "plain"編解碼器用於純文本，事件之間沒有分隔。這主要用於在其傳輸協議中已經定義了框架的輸入和輸出(例如zeromq、rabbitmq、redis等）。
2) "charset"默認值爲"UTF-8"。
(3) json_lines
1) 此編解碼器將解碼以換行符分隔的流式JSON。編碼將發出一個以"@delimiter"註釋結尾的JSON字符串，若是您的源輸入是面向行的JSON，例如redis或文件輸入，請不要使用此編解碼器。而是使用json編解碼器。
2) 此編解碼器指望接收換行符終止行的流(字符串)。文件輸入將產生一個沒有換行符的行串。所以，此編解碼器不能與面向行的輸入一塊兒使用。
3) "charset"默認值爲"UTF-8"。
4) "delimiter"默認值爲 "\n"。
(4) line
1) 面向行的文本數據。
2) 解碼行爲：只會發出整行事件。
3) 編碼行爲：每一個事件都將與尾隨換行符一塊兒發出。
4) "charset"默認值爲"UTF-8"。
4) "delimiter"默認值爲 "\n"。
(5) rubydebug
rubydebug編解碼器將使用 Ruby Amazing Print 庫輸出您的 Logstash 事件數據。
(1) Input： 輸入，輸入數據能夠是Stdin、File、TCP、Redis、Syslog等。
(2) Filter： 過濾，將日誌格式化。有豐富的過濾插件：Grok正則捕獲、Date時間處理、Json編解碼
、Mutate數據修改等。
(3) Output：輸出，輸出目標能夠是Stdout、File、TCP、Redis、ES等。
3.2 安裝
1 下載二進制軟件包
https://artifacts.elastic.co/downloads/logstash/logstash-7.13.4-linux-x86_64.tar.gz
2 修改配置文件
# tar -xzf logstash-7.13.4-linux-x86_64.tar.gz
# mv logstash-7.13.4/ /usr/local/logstash/
# useradd -M -s /sbin/nologin logstash
# id logstash
uid=1003(logstash) gid=1003(logstash) groups=1003(logstash)
# mkdir -p /usr/local/logstash/conf.d/
# mkdir -p /usr/local/logstash/logs/
# chown -R logstash.logstash /usr/local/logstash/
# egrep -v "^$|^#" /usr/local/logstash/config/logstash.yml
pipeline:
batch:
size: 125
delay: 5
# 管道配置
pipeline.ordered: auto
path.config: /usr/local/logstash/conf.d
# 自定義input，output流處理文件路徑
config.reload.automatic: false
# 關閉按期檢查配置是否修改，並從新加載管道(咱們使用SIGHUP信號手動觸發)。
config.reload.interval: 3s
# 檢查配置是否修改的時間間隔。
http.enabled: true
# 開啓http API。
http.host: 172.16.1.120
# 監聽的IP
http.port: 9600-9700
# 監聽的端口
log.level: info
# 日誌級別
path.logs: /usr/local/logstash/logs
# 日誌路徑
補充:
logstash默認數據存儲目錄爲: "path.data: LOGSTASH_HOME/data"
3.3 將Logstash服務加入systemd
1 logstash.service配置文件
# vim /usr/lib/systemd/system/logstash.service
[Unit]
Description=logstash server
After=network.target
[Service]
SuccessExitStatus=143
Type=simple
Environment=JAVA_HOME=/usr/local/logstash/jdk
ExecStart=/usr/local/logstash/bin/logstash
ExecReload=/bin/kill -HUP $MAINPID
KillMode=process
Restart=on-failure
User=logstash
Group=logstash
[Install]
WantedBy=multi-user.target
2 啓動服務
# systemctl daemon-reload
# systemctl start logstash.service
# systemctl enable logstash.service
注意:
上面啓動了logstash服務，但因爲logstash服務進程沒有可處理的數據，過一會logstash服務會自動中止運行，因此看不到logstash的9600端口號。
3.4 Logstash標準輸入輸出
1 編寫配置文件
# cat /usr/local/logstash/conf.d/input_stdin-output_stdout.conf
input{
stdin{}
}
output{
stdout{
codec=>rubydebug
}
}
2 驗證配置文件是否有語法錯誤
# /usr/local/logstash/bin/logstash -f /usr/local/logstash/conf.d/input_stdin-output_stdout.conf -t
3 從配置文件啓動logstash
# /usr/local/logstash/bin/logstash -f /usr/local/logstash/conf.d/input_stdin-output_stdout.conf
說明:
message # 輸入的信息
默認給日誌加了以下三個字段:
(1) "@timestamp" # 標記事件發生的時間點，寫入es時索引用到的" %{+YYYY.MM.dd} "變量就來自這裏。
(2) "host" # 標記事件發生的主機
(3) "@version"
4 查看logstash節點基本信息
# curl -XGET '172.16.1.120:9600/?pretty'
5 logstash命令行參數說明
(1) -f
# 指定配置文件在shell終端啓動logstash
logstash -f <file_name>.conf
(2) -t
# 驗證配置文件是否有語法錯誤
logstash -f <file_name>.conf -t
(3) -e
# 在shell終端啓動配置的logstash
logstash -e 'input{stdin{}}output{stdout{codec=>rubydebug}}'
注意:
使用該方式啓動logstash，須要先註釋掉"/usr/local/logstash/config/logstash.yml"配置文件中的
"path.config:"參數，不然會報以下錯誤。
ERROR: Settings 'path.config' (-f) and 'config.string' (-e) can't be used simultaneously.
3.5 輸入插件input(輸入階段，從哪裏獲取日誌)
3.5.1 經常使用插件
1 stdin # 通常用於調試
2 File
3 Redis
4 beats # 列如filebeat
3.5.2 通用配置字段
1 add_field # 添加一個字段到一個事件，放到事件頂部，通常標記日誌來源。例如屬於哪一個項目，哪一個應用。值類型是hash。
add_field => {
"project" => "microservice"
"app" => "product"
}
2 tags # 添加任意數量的標籤，用於標記日誌的其它屬性。例如表名是訪問日誌仍是錯誤日誌。
值類型是array，tags => ["web","nginx"]
3 type # 爲全部輸入添加一個字段，例如代表日誌的類型。
值類型是string，type => "access"
3.5.3 輸入插件file
file插件用於讀取指定日誌的文件。
1 經常使用字段
(1) path # 日誌文件路徑，可使用通配符。
值類型是array，path => ["/tmp/test.log","/tmp/test1.log"]
(2) exclude # 排除採集的日誌文件。
(3) discover_interval # 多久檢查被監聽的目錄下是否有新文件，默認15s。
(4) sincedb_write_interval # 多久寫一次sincedb文件，默認15s。
(5) stat_interval # 多久檢查被監聽文件的狀態默認1s。
(6) start_position # 指定日誌文件從什麼位置開始讀，默認從結尾開始，通常僅在初次讀取文件時起做用，若是文件已被記錄在sincedb中，則根據pos。
注意:
(1) logstash默認從日誌文件的末尾開始讀取，指定beginning表示從頭開始讀日誌文件，但這也僅限於第一次讀取文件時有效，讀取完成後會記錄日誌的位置(.sincedb_xxxx)，下次從記錄的位置開始讀。若是日誌被清理致使實際位置點比記錄的位置點小，那麼logstash會從頭開始讀取日誌。若是想要全部收集的日誌文件都從開頭讀取，先stop logstash，而後刪除logstash數據目錄下的全部數據，再啓動logstash便可(rm -rf /usr/local/logstash/data/*)。
(2) 在logstash收集日誌配置文件中設置以下內容，能夠實現讀取的目標文件未經修改，而僅修改了conf文件，實現每次從新運行logstash都從文件頭開始收集日誌。若是日誌收集完成，重啓logstash不想再從頭收集日誌，把添加的配置去掉，實現重啓logstash後從日誌末尾開始收集日誌。
input {
file {
path => ["/tmp/test.log"]
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
2 示例: 讀取日誌文件並輸出到文件
(1) 配置文件
# cat /usr/local/logstash/conf.d/input_file-output_file.conf
input {
file {
exclude => "error.log"
start_position => "beginning"
tags => "nginx"
type => "access"
add_field => {
"project" => "microservice"
"app" => "product"
}
}
}
output {
file {
path => "/tmp/out_result.txt"
}
}
(2) 檢查配置文件
/usr/local/logstash/bin/logstash -f /usr/local/logstash/conf.d/input_file-out_file.conf -t
……
Configuration OK
……
(3) 重啓logstash
# systemctl status logstash.service
# kill -HUP 3386
查看logstash日誌，肯定logstash已經從新啓動
# tailf /usr/local/logstash/logs/logstash-plain.log
(4) 測試
# echo "input_file-test" >>/tmp/test.log
經過/tmp/out_result.txt文件找到輸出日誌，經過在線json效驗網站將輸出內容進行格式化。
https://www.bejson.com/
(5) 說明
下面的相關測試按照此方式進行，但會省略一些重複的步驟，只貼出配置文件和輸出結果。
3.5.4 輸入插件Beats
Beats插件接收來自Beats數據採集器發來的數據，例如Filebeat。
經常使用字段: host—監聽地址，port—監聽端口。
用法:
# vim input_beats-output_file.conf
input {
beats {
host => "0.0.0.0"
port => 5044
}
}
filter {
}
output {
file {
path => "/tmp/out_result.txt"
}
}
3.6 過濾插件filter(過濾階段，將日誌格式化處理)
3.6.1 經常使用插件
json、kv、grok、geoip、date
3.6.2 通用配置字段
1 add_field # 若是過濾成功，添加一個字段到這個事件。
2 add_tags # 若是過濾成功，添加任意數量的標籤到這個事件。
3 remove_field # 若是過濾成功，從這個事件移除任意字段。
4 remove_tag # 若是過濾成功，從這個事件移除任意標籤。
3.6.3 過濾插件json
1 說明
JSON插件接收一個json數據，將其展開爲Logstash事件中的數據結構，放到事件的頂層。
2 經常使用字段
(1) source # 值類型爲string，指定要解析的字段，通常是原始消息message字段。
(2) targe t # 值類型爲string，將解析的結果放到指定字段，若是不指定，默認在事件的頂層。若是該target字段已經存在，它將被覆蓋。
(3) remove_field # 值類型爲array，若是此過濾器成功，則今後事件中刪除任意字段。
(1) 配置文件
# cat /usr/local/logstash/conf.d/filter_json-output_file.conf
input {
file {
path => "/tmp/test.log"
start_position => "beginning"
}
}
filter {
json {
source => "message"
}
}
output {
file {
path => "/tmp/out_result.txt"
}
}
(2) 收集的日誌
echo '{"remote_addr":"223.5.5.5","url":"/index.html","status":"200"}' >>/tmp/test.log
3.6.4 過濾插件kv
1 說明
kv插件接收一個鍵值數據，按照指定分割符解析爲logstash事件中的數據結構，放到事件頂層。
2 經常使用字段
field_split # 指定鍵值分隔符，默認是空
3 示例: 解析URL中的參數
(1) 配置文件
# cat /usr/local/logstash/conf.d/filter_kv-output_file.conf
input {
file {
path => "/tmp/test.log"
start_position => "beginning"
}
}
filter {
kv {
field_split => "?&"
}
}
output {
file {
path => "/tmp/out_result.txt"
}
}
(2) 收集日誌
echo 'www.baidu.com?id=19&name=cctv1&content=show' >>/tmp/test.log
3.6.5 過濾插件grok
1 說明
(1) 若是採集的日誌格式是非結構化的，能夠寫正則表達式提取，grok是正則表達式支持的實現。
(2) Logstash內置的正則匹配模式，在安裝目錄下能夠看到，路徑以下:
vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.3.1/patterns/legacy/grok-patterns
(3) 正則匹配模式語法格式
%{SYNTAX:SEMANTIC}
1) SYNTAX # 模式名稱，模式文件中的第一列
2) SEMANTIC # 匹配文件的字段名
例如: %{IP:client}
(4) 正則符號說明
2 經常使用字段
1) match # 正則匹配模式
2) patterns_dir # 自定義正則模式文件
3 示例
(1) 使用內置正則匹配http請求日誌
1) 模擬數據
223.5.5.5 GET /index.html 55632 0.013
2) 正則
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
(2) 自定義正則匹配http請求日誌
1) 自定義匹配規則
CID [0-9]{5,6}
TAG \w+
2) 自定義匹配日誌示例1
223.5.5.5 GET /index.html 55632 0.013 123456
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}
3) 自定義匹配日誌示例2
223.5.5.5 GET /index.html 55632 0.013 123456 abcdef
%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid} %{TAG:tag}
(3) 在logstash中匹配多格式日誌
若是一個日誌文件下有多個日誌格式怎麼辦，例如項目新版本添加一個日誌字段，須要兼容舊日誌匹配使用多模式匹配，寫多個正則表達式，只要知足其中一條就能匹配成功。
注意:
logstash多模式匹配時，正則匹配規則遵循從前日後進行匹配，若是輸入的日誌字段比較長，而短日誌字段匹配正則排在前面，那麼會丟失日誌字段，因此在排列正則時匹配字段較長的正則排在匹配字段較小的正則以前。
1) 配置文件
# mkdir -p /usr/local/logstash/patterns/
# cat /usr/local/logstash/patterns/patterns.conf
CID [0-9]{5,6}
TAG \w+
# cat /usr/local/logstash/conf.d/filter_grok-output_file.conf
input {
file {
path => "/tmp/test.log"
start_position => "beginning"
}
}
filter {
grok {
patterns_dir => "/usr/local/logstash/patterns"
# 單模式匹配
#match => {
# "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}"
#}
# 多模式匹配
match => [
"message","%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid} %{TAG:tag}",
"message","%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{CID:cid}"
]
}
}
output {
file {
path => "/tmp/out_result.txt"
}
}
2) 查看輸出
echo '223.5.5.5 GET /index.html 55632 0.013 123456' >>/tmp/test.log
echo '223.5.5.5 GET /index.html 55632 0.013 123456 abcdef' >>/tmp/test.log
3.6.6 過濾插件geoip
(1) GeoIP插件根據Maxmind GeoLite2數據庫中的數據添加有關IP地址位置信息。使用多模式匹配，寫多個正則表達式，只要知足其中一條就能匹配成功。
(2) 下載地址[須要註冊登陸才能下載]
https://www.maxmind.com/en/accounts/current/geoip/downloads
2 經常使用字段
(1) source # 指定要解析的IP字段，結果默認保存到geoip字段
(2) database GeoLite2 # 數據庫文件的路徑
(3) fields # 保留解析的指定字段
3 示例
(1) 配置文件
# tar -xzf GeoLite2-City_20210727.tar.gz
# mv GeoLite2-City_20210727/ /usr/local/logstash/
# cat /usr/local/logstash/conf.d/filter_geoip-output_file.conf
input {
file {
path => "/tmp/test.log"
start_position => "beginning"
}
}
filter {
grok {
patterns_dir => "/usr/local/logstash/patterns"
# 單模式匹配
match => {
"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"
}
}
geoip {
source => "client"
database => "/usr/local/logstash/GeoLite2-City_20210727/GeoLite2-City.mmdb"
# 指定保留geoip解析的字段，國家編碼、國家名、省會名、城市名
fields => ["country_code2", "country_name", "region_name", "city_name"]
# 保存到geoip字段(默認)
target => "geoip"
}
}
output {
file {
path => "/tmp/out_result.txt"
}
}
(2) 輸出結果
echo '223.5.5.5 GET /index.html 55632 0.013' >>/tmp/test.log
# 未指定保留geoip字段內特定內容的geoip字段輸出結果。
3.7 輸出插件output(將處理完成的日誌推送到遠程數據庫存儲)
3.7.1 經常使用插件
file、Elasticsearch
3.7.2 輸出插件elasticsearch
1 說明
Elasticsearch插件將數據推送到ES存儲。
2 經常使用字段
(1) hosts # 指定ES主機地址。
值類型是array，hosts => ["172.16.1.120","172.16.1.121","172.16.1.122"]
設置遠程實例的主機。若是給定一個數組，它將在hosts參數中指定的主機之間對請求進行負載平衡。
(2) index # 指定寫入的ES索引名稱，通常按日期ec劃分。
3 示例
(1) 配置文件
# cat /usr/local/logstash/conf.d/input_file-output_es.conf
input {
file {
path => "/tmp/test.log"
start_position => "beginning"
tags => "web"
tags => "nginx"
type => "access"
add_field => {
"project" => "microservice"
"app" => "product"
}
}
}
output {
elasticsearch {
hosts => ["172.16.1.120","172.16.1.121","172.16.1.122"]
index => "microservice-product-%{+YYYY.MM.dd}"
}
}
(2) 查看輸出
# echo "hello elasticsearch" >>/tmp/test.log
3.8 條件判斷
注意:
(1) /usr/local/logstash/conf.d/目錄下自定義的全部日誌收集配置文件，實際上全部的配置都在一個文件內，若是logstash在收集日誌時不根據日誌類型作條件判斷，會致使一個input輸入同時寫入多個output中的狀況(若是有的output作了判斷，有的output沒有作判斷，那麼沒有作判斷的output是全部日誌的總和)。logstash能夠同時有多個input和多個output。
(2) 如何在一臺服務器上起多臺logstash實例
1) 方法一
另安裝一個logstash實例，修改配置文件，啓動logstash便可。多個logstash實例的端口號依次排序爲9600、9601、……、9700。
2) 方法二
使用"/usr/local/logstash/bin/logstash -f /usr/local/logstash/conf.d/logstash1.conf --path.data=/tmp/data1"方式指定配置文件啓動logstash
實例，" --path.data=/tmp/data1 "表示實例的數據目錄(該目錄須要本身建立，注意權限)，每一個實例的數據目錄都不相同，若是不指
定該參數啓動logstash，會報' Logstash could not be started because there is already another instance using the configured data directory. If
you wish to run multiple instances, you must change the "path.data" setting. '的錯誤。
-f # 覆蓋logstash.yml文件中的" path.config: /usr/local/logstash/conf.d "參數。
--path.data # 覆蓋logstash.yml文件中的" path.data: LOGSTASH_HOME/data "參數。
3.8.1 操做符
1 比較操做符
==、!=、<、>、<=、>=
if [type] == " access" {}
# []表明引用的字段值，該值多是單值或是一個列表。
# 單值字段 # type字段、經過field添加的字段，或者其它字段。
# 列表字段 # 好比tags字段。
2 正則匹配操做符
=~(匹配正則)、!~(不匹配正則)
3 成員操做符
in(包含)、not in(不包含)
if "nginx " in [tags]{}
4 邏輯操做符
and(與)、or(或)、nand(非與)、nor(非或)
5 一元運算符
!(取反)、()(複合表達式)、!()(對複合表達式結果取反)
3.8.2 示例
1 典型應用場景
根據日誌來源字段寫入不一樣索引名稱。
日誌來源字段能夠爲項目名稱、應用名或測試環境、生產環境等。
2 配置文件
# cat /usr/local/logstash/conf.d/input_mutate-output_es.conf
input {
file {
path => "/tmp/test.log"
add_field => {
"log_type" => "test"
}
start_position => "beginning"
}
file {
path => "/tmp/prod.log"
add_field => {
"log_type" => "prod"
}
start_position => "beginning"
}
file {
path => "/tmp/unknown.log"
add_field => {
"log_type" => "unknown"
}
start_position => "beginning"
}
}
filter {
if [log_type] in ["test","dev"] {
mutate {
add_field => {
"[@metadata][target_index]" => "test-%{+YYYY.MM}"
}
}
} else if [log_type] == "prod" {
mutate {
add_field => {
"[@metadata][target_index]" => "prod-%{+YYYY.MM.dd}"
}
}
} else {
mutate {
add_field => {
"[@metadata][target_index]" => "unknown-%{+YYYY}"
}
}
}
}
output {
elasticsearch {
hosts => ["172.16.1.120","172.16.1.121","172.16.1.122"]
index => "%{[@metadata][target_index]}"
}
}
說明:
從Logstash 1.5開始，咱們能夠在logstash配置中使用metadata。metadata不會在output中被序列化輸出，這樣咱們即可以在metadata中添加一些臨時的中間數據，而不須要去刪除它。
3 查看輸出
# echo "test" >>/tmp/test.log
# echo "prod" >>/tmp/prod.log
# echo "unknown" >>/tmp/unknown.log
4 Filebeat部署
在172.16.1.120節點上操做
4.1 Filebeat介紹
Filebeat是一個輕量級的日誌採集器，將採集的數據推送到Logstash、ES存儲。
| 服務名 |
filebeat output默認編碼方式 |
| redis |
json |
| kafka |
json |
| file |
json |
| elasticsearch |
- |
| logstash |
- |
4.2 安裝
1 安裝包下載
(1) 二進制包
https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.13.4-linux-x86_64.tar.gz
(2) RPM包(生產建議使用)
https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.13.4-x86_64.rpm
2 解壓安裝包
# tar -xzf filebeat-7.13.4-linux-x86_64.tar.gz
# mv filebeat-7.13.4-linux-x86_64/ /usr/local/filebeat/
# mkdir -p /usr/local/filebeat/data/
# mkdir -p /usr/local/filebeat/logs/
# useradd -M -s /sbin/nologin filebeat
# id filebeat
uid=1005(filebeat) gid=1005(filebeat) groups=1005(filebeat)
# chown -R filebeat.filebeat /usr/local/filebeat/
4.3 將filebeat服務加入systemd
1 filebeat.service配置文件
# cat filebeat.service
[Unit]
Description=filebeat server
After=network.target
[Service]
Type=simple
ExecStart=/usr/local/filebeat/filebeat --path.config /usr/local/filebeat -c filebeat.yml --path.data /usr/local/filebeat/data --path.home /usr/local/filebeat --path.logs /usr/local/filebeat/logs
ExecReload=/bin/kill -HUP $MAINPID
KillMode=control-group
Restart=on-failure
User=filebeat
Group=filebeat
[Install]
WantedBy=multi-user.target
2 啓動服務
# systemctl daemon-reload
# systemctl start filebeat.service
# systemctl enable filebeat.service
4.4 推送日誌到Logstash
(1) filebeat日誌輸入類型能夠有多個，輸出只能有一個，不然會報以下錯誤。
Exiting: error unpacking config data: more than one namespace configured accessing 'output' (source:'/usr/local/filebeat/filebeat.yml')
(2) filebeat默認第一次讀取文件時從頭開始讀取，並記錄日誌的位置，下次從記錄的位置開始讀，若是日誌被清理致使實際位置點比記錄的位置點小，那麼filebeat會從頭開始讀取日誌。若是想要全部收集的日誌文件都從開頭讀取，先stop filebeat，而後刪除
filebeat數據目錄下的全部數據，再啓動filebeat便可。rm -rf /usr/local/filebeat/data/*
1 filebeat配置文件
# cat /usr/local/filebeat/filebeat.yml
filebeat.inputs:
# 配置不一樣的輸入
- type: log
# 是否啓用該輸入配置
enabled: true
# 採集的日誌文件路徑，能夠通配
- /tmp/test.log
# 正則匹配要排除的行，這裏以DBG開頭的行都過濾掉
exclude_lines: ['^DBG']
# 正則匹配要採集的行，這裏以ERR/WARN開頭的行都採集
#include_lines: ['^ERR', '^WARN']
# 排除的文件，默認採集全部
exclude_files: ['.gz$']
# 添加標籤
tags: ["web","nginx"]
# 添加類型，改參數已經在filebeat6.0時被取消
# document_type: "access"
# 下面fields添加的字段默認是在fields.xxx，能夠設置在頂級對象下
fields_under_root: true
# 自定義添加的字段，通常用於標記日誌來源
fields:
project: microservice
app: product
# 輸出到文件
output.file:
path: "/tmp"
enabled: false
filename: "filebeat_out.txt"
# 輸出到logstash
hosts: ["172.16.1.120:5044"]
# logstash服務器地址，能夠有多個
enabled: true
# 是否開啓輸出至logstash，默認即爲true
# 工做線程數
compression_level: 3
# 壓縮級別
# loadbalance: true
# 多個輸出的時候開啓負載
# 輸出到es
setup.ilm.enabled: false
setup.template.name: "filebeat-es"
setup.template.pattern: "filebeat-es-*"
output.elasticsearch:
hosts: ["172.16.1.120:9200","172.16.1.121:9200","172.16.1.122:9200"]
index: "filebeat-es-%{+yyyy.MM.dd}"
enabled: false
# 添加模塊
-
每一个你不满意的现在，都有一个你没有努力的曾经。