filebeat+ELK配置及常用操作
背景介紹
最近工作涉及幾臺新伺服器的日誌需要接入ELK系統，配置思路如下：
使用Filebeat收集本地日誌資料，Filebeat監視日誌目錄或特定的日誌檔案，再發送到訊息佇列到kafka，然後logstash去獲取消費，利用filter功能過濾分析，最終儲存到elasticsearch中。
filebeat和flume都具有日誌收集功能，不過filebeat更輕量，使用go語言編寫佔用資源更少，可以有很高的併發，帶有內部模組（auditd，Apache，Nginx，System和MySQL），可通過一個指定命令來簡化通用日誌格式的收集，解析和視覺化；flume使用java開發，需要安裝java環境，相對會比較重。
當然兩者也存在區別：Filebeat收集資料的速度大於寫入速度的時候可能出現數據丟失的現象，而flume會在收集資料和寫入資料之間做出調整，保證能在兩者之間提供一種平穩的資料狀態。可以實時的將分析資料並將資料儲存在資料庫或者其他系統中，不會出現資料丟失的現象。
以下僅記錄配置過程及常見的幾種排錯命令，安裝篇會獨立一篇做詳細介紹。
配置資訊
filebeat配置
我是直接yum install filebeat一鍵安裝的，這裡不做具體講解官網有詳細介紹：
https://www.elastic.co/guide/en/beats/filebeat/current/index.html
安裝完成後我們以配置採集/var/log/messages為例，配置如下
# egrep -v '#|^$' /etc/filebeat/filebeat.yml filebeat.inputs: - type: log enabled: true paths: - /var/log/messages filebeat.config.modules: path: ${path.config}/modules.d/*.yml reload.enabled: false setup.template.settings: index.number_of_shards: 1 setup.kibana: output.kafka: hosts: ["10.114.102.30:9092", "10.114.102.31:9092", "10.114.102.32:9092", "10.114.102.33:9092", "10.114.102.34:9092"] topic: T621_messages partition.round_robin: reachable_only: false required_acks: 1 compression: gzip processors: - add_host_metadata: when.not.contains.tags: forwarded - add_cloud_metadata: ~ - add_docker_metadata: ~ - add_kubernetes_metadata: ~
有幾個引數需要注意的：
paths表示需要提取的日誌的路徑，將日誌輸出到kafka中，建立topic
- required_acks
0：這意味著生產者producer不等待來自broker同步完成的確認繼續傳送下一條（批）訊息。此選項提供最低的延遲但最弱的耐久性保證（當伺服器發生故障時某些資料會丟失，如leader已死，但producer並不知情，發出去的資訊broker就收不到）。
1：這意味著producer在leader已成功收到的資料並得到確認後傳送下一條message。此選項提供了更好的耐久性為客戶等待伺服器確認請求成功（被寫入死亡leader但尚未複製將失去了唯一的訊息）。
-1：這意味著producer在follower副本確認接收到資料後才算一次傳送完成。 此選項提供最好的耐久性，我們保證沒有資訊將丟失，只要至少一個同步副本保持存活。 三種機制，效能依次遞減 (producer吞吐量降低)，資料健壯性則依次遞增。
- json.keys_under_root: true
- json.add_error_key: true
- json.message_key: log
這三行是識別json格式日誌的配置，若日誌格式不為json格式，需要註釋掉，否則收集到的日誌為filebeat的錯誤日誌。
kafka配置
kafka原來已經安裝並配置好了，這裡不再說明具體安裝過程，後續會出一篇ELK完整搭建過程。
這裡不做重點講解，可直接查官網：https://kafka.apache.org/documentation/#quickstart
因為有5臺配合zookeeper做了叢集，選其中一臺配置如下：
# egrep -v '#|^$' /home/kafka/kafka/config/server.properties
broker.id=1 #按順序寫，不要亂
listeners=PLAINTEXT://0.0.0.0:9092 #自己的ip
advertised.listeners=PLAINTEXT://10.114.102.30:9092
num.network.threads=24
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka
num.partitions=8
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=48
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=10.114.102.30:2181,10.114.102.31:2181,10.114.102.32:2181,10.114.102.33:2181,10.114.102.34:2181
zookeeper.connection.timeout.ms=6000
auto.create.topics.enable=false
group.initial.rebalance.delay.ms=0
注意：每臺伺服器除broker.id需要修改之外，其他屬性保持一致。
logstash配置
logstash安裝也是直接參考官網就可以了
https://www.elastic.co/guide/en/logstash/7.x/index.html
不過有個地方要注意，kafka和logstash的版本相容問題，以下是kafka使用的版本：
find /home/kafka/kafka/libs/ -name \*kafka_\* | head -1 | grep -o '\kafka[^\n]*'
kafka_2.11-1.1.0.jar
通過查詢rpm包可以看到logstash用的是7.8.0
/home/cxhchusr/logstash-7.8.0.rpm
在conf.d目錄下配置消費messages的檔案如下
# egrep -v '#|^$' /etc/logstash/conf.d/T621_messages.conf
input {
kafka {
bootstrap_servers => "10.114.102.30:9092,10.114.102.31:9092,10.114.102.32:9092,10.114.102.33:9092,10.114.102.34:9092"
client_id => "T621_messages"
group_id => "T621_messages"
auto_offset_reset => "latest"
consumer_threads => 10
decorate_events => true
topics => ["T621_messages"]
decorate_events => true
type => syslog
}
}
filter{
grok {
match => { "message" => "%{SYSLOGLINE}" }
}
date {
match => [ "logdate", "YYYY-MM-dd HH:mm:ss.SSS" ]
target => "@timestamp"
timezone =>"+00:00"
}
mutate{
remove_field => "logdate"
}
}
output {
elasticsearch {
hosts => ["10.114.102.30:9200", "10.114.102.31:9200", "10.114.102.32:9200", "10.114.102.33:9200", "10.114.102.34:9200"]
index => "t621_messages-%{+YYYY.MM.dd}"
user => caixun
password => "******()90"
}
}
注意：es索引需要全部為小寫。
最後啟動即可，並加入開機自啟動/etc/rc.local：
nohup /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/ > /dev/null 2>&1 &
kibana建立索引
logstash配置完成後即可在kibana建立索引
建立完成效果如下，表示接入成功：
常用排查命令
配置過程中除了在kafka上建立topic，還需查詢topic是否建立成功、消費情況、以及訊息處理情況。
以及es是否正常入庫並建立了索引。下面列出幾個ELK運維常用命令。
kafka常用運維指令
- 查詢當前topic列表
/home/kafka/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --list
- 建立topic
/home/kafka/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --create --partitions 10 --replication-factor 1 --topic T621_messages
- topic描述(某topic詳細資訊)
/home/kafka/kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe --topic T621_messages
- topic消費情況(測試訊息是否正常生產)
傳送：/home/kafka/kafka/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic T621_messages
接收：/home/kafka/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic T621_messages --from-beginning
- 檢視topic堆積情況
/home/kafka/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list ##檢視組列表
/home/kafka/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group T621_messages ##偏移量
/data/kafka/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic T621_messages --from-beginning
ES常用運維指令
- 檢視ES叢集健康情況
curl -u caixxx:"CAIxxx()90" '10.114.102.30:9200/_cluster/health'
- 檢視索引儲存情況
curl -u caixxx:"CAIxxx()90" '10.114.102.30:9200/_cat/indices?v'
- 檢視幫助命令
# curl -u caixun:"CAIXUN()90" '10.114.102.30:9200/_cat'
=^.^=
/_cat/allocation
/_cat/shards
/_cat/shards/{index}
/_cat/master
/_cat/nodes
/_cat/tasks
/_cat/indices
/_cat/indices/{index}
/_cat/segments
/_cat/segments/{index}
/_cat/count
/_cat/count/{index}
/_cat/recovery
/_cat/recovery/{index}
/_cat/health
/_cat/pending_tasks
/_cat/aliases
/_cat/aliases/{alias}
/_cat/thread_pool
/_cat/thread_pool/{thread_pools}
/_cat/plugins
/_cat/fielddata
/_cat/fielddata/{fields}
/_cat/nodeattrs
/_cat/repositories
/_cat/snapshots/{repository}
/_cat/templates
/_cat/transforms
/_cat/transforms/{transform_id}
過濾索引檢視訊息是否成功儲存在es，有的話代表配置成功。
---- 鋼鐵 [email protected] 2021.08.20
參考鳴謝
官方kafka：http://kafka.apache.org/