通过 Filebeat 收集 ubuntu 系统日志
收藏

 点击上方

壹伴助手关注我们




需求:收集 ubuntu 系统日志,发送给 logstash,再由 logstash 把数据传递给 elasticsearch,最后通过 kibana 展示日志数据。


Filebeat System Module


Filebeat Modules 可以简化常见日志格式的收集、解析和可视化。一个典型的模块(例如,对于 Nginx 日志)由一个或多个日志文件(fileset)组成(对于 Nginx 来说,默认是 access.log 和 error.log)。这里我们可以使用 Filebeat 的 System Module 完成 ubuntu 的系统日志。


下面介绍配置 System Module 的步骤(假如你已经安装好了 Filebeat)。


启用 System Module
Filebeat 支持的模块默认都是未启用的,我们可以通过下面的方式启用模块。找到 filebeat 程序,执行 moudles enable 命令:

$ sudo ./filebeat modules enable system


上面的命令启用了 system 模块,用下面的命令可以查看当前已经启用的模块有哪些:

$ sudo ./filebeat modules list


把数据发送给 logstash
配置 Filebeat 将日志行发送到 Logstash。要做到这一点,在配置文件 filebeat.yml 中禁用 Elasticsearch 输出,并启用 Logstash 输出:

#output.elasticsearch:  #hosts: ["xxx.xxx.xxx.xxx:9200"]output.logstash:  hosts: ["xxx.xxx.xxx.xxx:5044"]


重启 filebeat 服务

$ sudo systemctl restart filebeat.service

配置 Logstash 处理数据


要让 logstash 接受 Filebeat System Module 发送来的数据还是有些难度的,至少我们需要一个看上去有点复杂的配置:

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}"  }}


处理时区问题

看到这段配置我多么希望它能够直接工作啊!不幸的是它并不能很好的工作,至少在我的 ubuntu 18.04 上不行。问题的核心是无论 auth.log 还是 syslog,记录的都是本地时区的区时:


而上面的配置中把这些时间都当成 UTC 时间来处理了。搞清楚了原因,纠正起来就很容易了,在 date 插件中添加本地的时区信息就可以了。比如笔者所在时区为东八区,那么就分别在两个 date 的配置中添加下面的信息:

timezone => "Asia/Chongqing"


让独立的 pipeline 处理该数据流
下面创建一个新的目录 /etc/logstash/myconf.d,并在 /etc/logstash/myconf.d 目录下创建 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 的管道是默认的配置,我们新添加了 id 为 krtest 的管道并指定了对应的配置文件路径。把上面的配置写入到 /etc/logstash/myconf.d/krtest.conf 文件中。然后重启 logstash 服务:

$ sudo systemctl restart logstash.service


在 Kibana 中查看日志


最后在 kibana 中添加 filebeat 开头的 index pattern,就可以通过图形界面查看 ubuntu 的系统日志了:


参考:

  • Filebeat Modules

    (https://www.elastic.co/guide/en/beats/filebeat/6.2/filebeat-modules.html

  • System module

    (https://www.elastic.co/guide/en/beats/filebeat/6.2/filebeat-module-system.html)

  • Working with Filebeat Modules

    (https://www.elastic.co/guide/en/logstash/6.2/filebeat-modules.html#graduating-to-Logstash)


出处:http://www.cnblogs.com/sparkdev/



最新课程通知


Golang 零基础入门课程

本课程为:网络班(腾讯课堂)

https://ke.qq.com/course/406096?tuin=6ceacb80


51reboot 运维前端课程正在火热招生和中,

7.27号开课


想要学习的小伙伴可以扫码咨询

添加小助手WeChat:17812796384



 

    公众号
    关注公众号订阅更多技术干货!