filebeat日志收集

以nginx錯誤日志為例，演示日志處理流程

filebeat--logstash--es

filebeat--kafka--logstash--es

#filebeat使用systemd管理
/usr/lib/systemd/system/filebeat.service?
[Unit]
Description=Filebeat
Documentation=http://www.elastic.co
Wants=network-online.target
After=network-online.target
[Service]
ExecStart=/usr/local/filebeat/filebeat?-c?/usr/local/filebeat/filebeat.yml
Restart=always
[Install]
WantedBy=multi-user.target

#logstash使用systemd管理
#如果有多個logstash配置文件,可以使用-f指定目錄
/usr/lib/systemd/system/logstash.service?
[Unit]
Description=logstash
Documentation=http://www.elastic.co
Wants=network-online.target
After=network-online.target
[Service]
Environment=JAVA_HOME=/usr/java/jdk1.8.0_211
ExecStart=/usr/local/logstash/bin/logstash?-f?/usr/local/logstash/config/logstash.conf?-l?/usr/local/logstash/logs
Restart=always
[Install]
WantedBy=multi-user.target

#啟動nginx容器，映射日志目錄
docker?run?-d?--name=nginx?--net=host?-v?/tmp/nginx_log:/var/log/nginx?nginx

#nginx錯誤日志：

2019/09/21 17:00:08 [error] 7#7: *9 open() "/usr/share/nginx/html/api" failed (2: No such file or directory), client: 192.168.3.102, server: localhost, request: "GET /api HTTP/1.1", host: "192.168.3.100"

• filebeat--logstash--es示例

#filebeat輸出logstash示例
/usr/local/filebeat/filebeat.yml?
filebeat.inputs:
-?type:?log
??paths:
????-?/tmp/nginx_log/error.log
??
??multiline.pattern:?^\d{4}/\d{2}/\d{2}\s\d{2}:\d{2}:\d{2}
??#匹配nginx日志時間格式??2019/09/21?17:00:08
??multiline.negate:?true
??multiline.match:?after
??exclude_files:?[".gz$"]
??tail_files:?true
??
??#增加輸出字段，tags為數組形式，fields.id為鍵值對形式
??tags:?["nginx-100"]
??fields:
????id:?"nginx-100"
output.logstash:
??hosts:?["192.168.3.100:5044","192.168.3.101:5044"]
??loadbalance:?true

#輸出到單個logstash
#output.logstash:
#??hosts:?["127.0.0.1:5044"]

#logstash輸出到es示例；根據fileds.id來劃分索引
/usr/local/logstash/config/logstash.conf?
input?{
??beats?{
????port?=>?5044
??}
}
output?{
??elasticsearch?{
????hosts?=>?["http://192.168.3.100:9200","http://192.168.3.101:9200","http://192.168.3.102:9200"]
????index?=>?"%{[fields][id]}-%{+YYYY.MM.dd}"
????user?=>?"elastic"
????password?=>?"HkqZIHZsuXSv6B5OwqJ7"
??}
}

• filebeat--kafka--logstash--es示例

#filebeat輸出到kafka示例
/usr/local/filebeat/filebeat.yml?
filebeat.inputs:
-?type:?log
??paths:
????-?/tmp/nginx_log/error.log
??
??multiline.pattern:?^\d{4}/\d{2}/\d{2}\s\d{2}:\d{2}:\d{2}
??#匹配nginx日志時間格式??2019/09/21?17:00:08
??multiline.negate:?true
??multiline.match:?after
??exclude_files:?[".gz$"]
??tail_files:?true
??
??#增加輸出字段，tags為數組形式，fields.id為鍵值對形式
??tags:?["nginx-kafka-100"]
??fields:
????id:?"nginx-kafka-100"
output.kafka:
??hosts:?["192.168.3.100:9092",?"192.168.3.101:9092",?"192.168.3.102:9092"]

??topic:?'%{[fields.id]}'
??partition.round_robin:
????reachable_only:?false

??required_acks:?1
??compression:?gzip
??max_message_bytes:?1000000

#kafka輸出到es示例
/usr/local/logstash/config/logstash.conf
input?{
????kafka?{
??????group_id?=>?"logstash"
??????topics?=>?["nginx-kafka-100"]
??????bootstrap_servers?=>?"192.168.3.100:9092,192.168.3.101:9092,192.168.3.102:9092"
??????consumer_threads?=>?"1"
??????fetch_max_bytes?=>?"26214400"
??????codec?=>?plain
??}
}
filter?{
??json?{
????source?=>?"message"
??}
}
output?{
??elasticsearch?{
????hosts?=>?["http://192.168.3.100:9200","http://192.168.3.101:9200","http://192.168.3.102:9200"]
????index?=>?"%{[fields][id]}-%{+YYYY.MM.dd}"
????user?=>?"elastic"
????password?=>?"HkqZIHZsuXSv6B5OwqJ7"
??}
}

參考：

https://www.elastic.co/guide/en/beats/filebeat/current/kafka-output.html

https://www.elastic.co/guide/en/beats/filebeat/current/logstash-output.html

https://www.elastic.co/guide/en/logstash/current/plugins-filters-json.html