建站
咨詢
售后
建站咨詢
新聞中心
Filebeat是用于轉(zhuǎn)發(fā)和集中日志數(shù)據(jù)的輕量級(jí)傳送程序。作為服務(wù)器上的代理安裝，F(xiàn)ilebeat監(jiān)視您指定的日志文件或位置，收集日志事件，并將它們轉(zhuǎn)發(fā)到Elasticsearch或Logstash進(jìn)行索引。
Filebeat System Module
啟用 System Module Filebeat 支持的模塊默認(rèn)都是未啟用的，我們可以通過(guò)下面的方式啟用模塊。找到 filebeat 程序，執(zhí)行 moudles enable 命令：
$ sudo ./filebeat modules enable system
上面的命令啟用了 system 模塊，用下面的命令可以查看當(dāng)前已經(jīng)啟用的模塊有哪些：
$ sudo ./filebeat modules list
把數(shù)據(jù)發(fā)送給 logstash 配置 Filebeat 將日志行發(fā)送到 Logstash。要做到這一點(diǎn)，在配置文件 filebeat.yml 中禁用 Elasticsearch 輸出，并啟用 Logstash 輸出：
#output.elasticsearch:
#hosts: ["xxx.xxx.xxx.xxx:9200"]
output.logstash:
hosts: ["xxx.xxx.xxx.xxx:5044"]
重啟 filebeat 服務(wù)
$ sudo systemctl restart filebeat.service
配置 Logstash 處理數(shù)據(jù)
要讓 logstash 接受 Filebeat System Module 發(fā)送來(lái)的數(shù)據(jù)還是有些難度的，至少我們需要一個(gè)看上去有點(diǎn)復(fù)雜的配置：
input {
beats {
port => 5064
host => "0.0.0.0"
}
}
filter {
if [fileset][module] == "system" {
if [fileset][name] == "auth" {
grok {
match => { "message" => ["%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} %{DATA:[system][auth][ssh][method]} for (invalid user )?%{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]} port %{NUMBER:[system][auth][ssh][port]} ssh2(: %{GREEDYDATA:[system][auth][ssh][signature]})?",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} user %{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]}",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: Did not receive identification string from %{IPORHOST:[system][auth][ssh][dropped_ip]}",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sudo(?:\[%{POSINT:[system][auth][pid]}\])?: \s*%{DATA:[system][auth][user]} :( %{DATA:[system][auth][sudo][error]} ;)? TTY=%{DATA:[system][auth][sudo][tty]} ; PWD=%{DATA:[system][auth][sudo][pwd]} ; USER=%{DATA:[system][auth][sudo][user]} ; COMMAND=%{GREEDYDATA:[system][auth][sudo][command]}",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} groupadd(?:\[%{POSINT:[system][auth][pid]}\])?: new group: name=%{DATA:system.auth.groupadd.name}, GID=%{NUMBER:system.auth.groupadd.gid}",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} useradd(?:\[%{POSINT:[system][auth][pid]}\])?: new user: name=%{DATA:[system][auth][user][add][name]}, UID=%{NUMBER:[system][auth][user][add][uid]}, GID=%{NUMBER:[system][auth][user][add][gid]}, home=%{DATA:[system][auth][user][add][home]}, shell=%{DATA:[system][auth][user][add][shell]}$",
"%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} %{DATA:[system][auth][program]}(?:\[%{POSINT:[system][auth][pid]}\])?: %{GREEDYMULTILINE:[system][auth][message]}"] }
pattern_definitions => {
"GREEDYMULTILINE"=> "(.|\n)*"
}
remove_field => "message"
}
date {
match => [ "[system][auth][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
geoip {
source => "[system][auth][ssh][ip]"
target => "[system][auth][ssh][geoip]"
}
}
else if [fileset][name] == "syslog" {
grok {
match => { "message" => ["%{SYSLOGTIMESTAMP:[system][syslog][timestamp]} %{SYSLOGHOST:[system][syslog][hostname]} %{DATA:[system][syslog][program]}(?:\[%{POSINT:[system][syslog][pid]}\])?: %{GREEDYMULTILINE:[system][syslog][message]}"] }
pattern_definitions => { "GREEDYMULTILINE" => "(.|\n)*" }
remove_field => "message"
}
date {
match => [ "[system][syslog][timestamp]", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
}
}
}
output {
elasticsearch {
hosts => xxx.xxx.xxx.xxx
manage_template => false
index => "%{[@metadata][beat]}-%{[@metadata][version]}-%{+YYYY.MM.dd}"
}
}
處理時(shí)區(qū)問(wèn)題 看到這段配置我多么希望它能夠直接工作??！不幸的是它并不能很好的工作，至少在我的 ubuntu 18.04 上不行。問(wèn)題的核心是無(wú)論 auth.log 還是 syslog，記錄的都是本地時(shí)區(qū)的區(qū)時(shí)：
而上面的配置中把這些時(shí)間都當(dāng)成 UTC 時(shí)間來(lái)處理了。搞清楚了原因，糾正起來(lái)就很容易了，在 date 插件中添加本地的時(shí)區(qū)信息就可以了。比如筆者所在時(shí)區(qū)為東八區(qū)，那么就分別在兩個(gè) date 的配置中添加下面的信息：
timezone => "Asia/Chongqing"
讓獨(dú)立的 pipeline 處理該數(shù)據(jù)流 下面創(chuàng)建一個(gè)新的目錄 /etc/logstash/myconf.d，并在 /etc/logstash/myconf.d 目錄下創(chuàng)建 Logstash 配置文件 krtest.conf。然后在 /etc/logstash/pipelines.yml 文件中添加新的 pipeline 配置： – pipeline.id: main path.config: “/etc/logstash/conf.d/*.conf” – pipeline.id: krtest path.config: “/etc/logstash/myconf.d/krtest.conf” 其中 pipeline.id 為 main 的管道是默認(rèn)的配置，我們新添加了 id 為 krtest 的管道并指定了對(duì)應(yīng)的配置文件路徑。把上面的配置寫入到 /etc/logstash/myconf.d/krtest.conf 文件中。然后重啟 logstash 服務(wù)：
$ sudo systemctl restart logstash.service
在 Kibana 中查看日志
最后在 kibana 中添加 filebeat 開頭的 index pattern，就可以通過(guò)圖形界面查看 ubuntu 的系統(tǒng)日志了：
本文題目：使用Filebeat收集Ubuntu系統(tǒng)日志
標(biāo)題來(lái)源：http://www.5511xx.com/article/djcoisc.html
