您的位置 首页 php

ELK日志分析平台Logstash常用收集各类日志举例

Logstash 是一个开源的数据收集引擎,它具有备实时数据传输能力。它可以统一过滤来自不同源的数据,并按照开发者的制定的规范输出到目的地。顾名思义,Logstash 收集数据对象就是日志文件。由于日志文件来源多(如:系统日志、服务器日志、网络设备日志等),且内容杂乱,不便于人类进行观察。因此,我们可以使用 Logstash 对日志文件进行收集和统一过滤,变成可读性高的内容,方便开发者或运维人员观察,从而有效的分析系统/项目运行的性能,做好监控和预警的准备工作等。

Logstash 通过管道进行运作,管道有两个必需的元素,输入和输出,还有一个可选的元素,过滤器。输入插件从数据源获取数据,过滤器插件根据用户指定的数据格式修改数据,输出插件则将数据写入到目的地。在实际工作中Logstash数据收集的一个流程大概是:数据源→通过input插件(如 file 、redis、stdin)→ output 插件→写到 Elasticsearch

下面我通过ELK平台收集下图所示的日志,举例说明Logstash是怎么收集一些常用各类日志的。这些日志对zabbix服务器的正常运行至关重要,这也是大部分应用系统正常运行会包含的一些日志。关于ELK的搭建可以参考 ,对zabbix监控技术有兴趣的同学可以参考搭建 。

1.file插件收集日志文件

在zabbix服务器172.18.20.28上编辑logstash配置文件,并将收集到的zabbix_server.log文件数据提交到elasticsearch中:

#cat logstash-zabbixserver.conf
input {
 file{
 path => " / var /log/zabbix/zabbix_server.log" #文件路径
 type => "zabbix-server" #类似打个标记,自定义 
 start_position => "beginning" #文件头部读取,相反还有end
 }
}
output {
 elasticsearch {
 hosts => ["172.28.29.211:9200"] #输出到elasticsearch
 index => "zabbix-%{+YYYY.MM.dd}" #按年月日格式生成索引
 }
}
 

运行logstash:

/usr/local/logstash-6.2.4/ bin /logstash -f logstash-zabbixserver.conf
 

在elasticsearch上查看数据:

2. if判断收集多个日志

现在需要在收集另外一个日志文件 mariadb .log,我们修改logstash配置文件使用if判断进行收集:

#cat logstash-zabbixserver.conf
input {
 file{
 path => " /var/log/zabbix/zabbix_server.log" 
 type => "zabbix-server" 
 start_position => "beginning" 
}
file{
 path => " /var/log/mariadb/mariadb.log" 
 type => "mysql" 
 start_position => "beginning" 
}
}
output {
if [type] == " zabbix-server " { #根据type来匹配
elasticsearch {
hosts => ["172.28.29.211:9200"]
 index => " zabbix -%{+YYYY.MM.dd}"
 }
 }
 if [type] == " mysql " {
 elasticsearch {
 hosts => ["172.28.29.211:9200"]
 index => " zabbix-mysql -%{+YYYY.MM}" #按年月
 } 
 } 
}
 

再次运行logstash:

/usr/local/logstash-6.2.4/bin/logstash -f logstash-zabbixserver.conf
 

在elasticsearch上查看另一个收集的日志数据:

3.syslog插件收集系统网络日志

syslog默认是通过514端口去发送日志,可以使用logstash安装对应的syslog插件,监听514端口来收集日志。如果只是监听系统日志可以不用安装logstash的agent,只需要监听对应的514端口,收集系统数据即可。logstash在INPUT插件中提供了syslog的模块,可以用于不安装本地日志收集agent的设备(如硬件防火墙等),当然支持此类协议的普通服务器也适用。

注意:此INPUT插件会同时监听TCP和UDP端口。

服务器rsyslog收集

创建编辑logstash-syslog配置文件,使用syslog插件:

#cat logstash-syslog.conf
input {
syslog {
 type => "system-syslog"
 port => 514 #默认为514端口,可自行修改
 }
}
output {
 elasticsearch {
 hosts => ["172.28.29.211:9200"] #输出到elasticsearch
 index => "zabbix-syslog-%{+YYYY.MM}" #名称按年月保存收集
 }
}
 

运行logstash:

/usr/local/logstash-6.2.4/bin/logstash -f logstash-syslog.conf
 

重新开启一个窗口,查看服务是否启动:

# netstat –ntlp | grep 514
tcp6 0 0 :::514 :::* LISTEN 21544/java 
 

修改rsyslog配置文件:

# vi /etc/rsyslog.conf
…
#*.* @@remote-host:514 
*.* @@172.18.20.28:514 #添加远程syslog服务器IP,这里是本机
 

重启rsyslog:

systemctl restart rsyslog
 

在elasticsearch上查看收集到的服务器rsyslog日志:

网络设备syslog收集

收集交换机网和防火墙syslog日志,配置如下:

#cat wl-syslog.conf
input{
 syslog {
 type => "wl-syslog"
 port => 514
 }
}
output {
 if [host] == "172.18.20.254"{ #根据host来匹配生成相应的索引
 elasticsearch {
 hosts => ["172.28.29.211:9200"]
 index => "jhj-%{+YYYY.MM}"
 }
 }
 if [host] == "172.18.16.254"{
 elasticsearch {
 hosts => ["172.28.29.211:9200"]
 index => "fhq-%{+YYYY.MM}"
 }
 }
}
 

相应的网络设备上开启并指定syslog服务器,以Cisco设备为例:

logging on
logging host 172.28.29.211
 

在elasticsearch上查看收集这两台网络设备的日志:

4.grok插件收集Apache访问日志

一般系统或服务生成的日志都是一大长串。每个字段之间用空格隔开。logstash在获取日志是整个一串获取,如果把日志中每个字段代表的意思分割开来在传给elasticsearch。这样呈现出来的数据更加清晰,而且也能让kibana更方便的绘制图形。Grok是Logstash最重要的插件。它的主要作用就是将文本格式的字符串,转换成为具体的结构化的数据,配合 正则表达式 使用。

grok事先已经预定义好了许多正则表达式规则,该规则文件存放路径:

/usr/local/logstash-6.2.4/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-
core-4.1.2/patterns
 

grok插件语法说明, 以一个简单的访问日志为例:

55.3.244.1 GET /index.html 15824 0.043
 

这条日志可切分为5个部分,IP(55.3.244.1)、方法(GET)、请求文件路径(/index.html)、字节数(15824)、访问时长(0.043) ,如果不做拆分的话Kibana里会将所有内容都写在 message s字段里。如果我们想把不同字段做分类的话,就可以用grok来进行定义,在默认配置文件里其实很多配置都已经写好了,只需要引用下:

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{ NUMBER : bytes } %{NUMBER:duration}
 

大写的单词是预定义的正则表达式,可以在grok-patterns文件看到其具体匹配的内容。如果匹配的话,就将该字段名称重新定义为冒号后面的小写的单词。用上面那条访问日志的客户端IP来说明,它将在Kibana里被定义为client字段并单独显示出来,而不是全部塞在了message里。

写到filter中:

filter {
 grok {
 match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"}
 }
}
 

解析后:

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043
 

下面是一条zabbix服务器Apache日志:

192.168.33.203 - - [11/Jul/2018:09:37:07 +0800] "POST
/zabbix/jsrpc.php?output=json-rpc HTTP/1.1" 200 65 "" "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:61.0) Gecko/20100101 Firefox/61.0"
 

创建编辑logstash-apache配置文件,使用filter中的grok插件:

# cat logstash-apache.conf
input{
 file{
 path => "/etc/ httpd /logs/access_log"
 type => "apache"
 start_position => "beginning"
 }
}
filter{
 if[type] == "apache"{
 grok{
 match => {"message" => "%{COMBINEDAPACHELOG}"}
 }
 }
}
output{
 if[type] == "apache"{
 elasticsearch{
 hosts => ["172.28.29.211:9200"]
 index => "zabbix_apache-%{+YYYY.MM}"
 }
 }
}
#fileter中的message代表一条一条的日志,%{COMBINEDAPACHELOG}代表解析日志的正则表达式
 

运行logstash:

/usr/local/logstash-6.2.4/bin/logstash -f logstash-apache.conf
 

在elasticsearch上查看收集到的Apache日志:

整合到一个logstash配置文件运行

我们已成功收集到各项日志,现在我们需要将这些单独的logstash配置文件整合到一个配置文件中,配置如下:

#cat logstash-all.conf
input {
 syslog{
 type => "system-syslog"
 port => 514 
 }
 file{
 path => "/var/log/zabbix/zabbix_server.log" 
 type => "zabbix-server" 
 start_position => "beginning" 
 }
 file{
 path => "/var/log/mariadb/mariadb.log" 
 type => "mysql" 
 start_position => "beginning" 
 }
 file{
 path => "/etc/httpd/logs/access_log"
 type => "apache"
 start_position => "beginning"
}
}
filter{
 if[type] == "apache"{
 grok{
 match => {"message" => "%{COMBINEDAPACHELOG}"}
 }
 }
}
output {
if [type] == "system-syslog" {
 elasticsearch {
 hosts => ["172.28.29.211:9200"] 
 index => "zabbix-syslog-%{+YYYY.MM}" 
 }
 }
if [type] == "zabbix-server" { 
elasticsearch {
hosts => ["172.28.29.211:9200"]
 index => "zabbix _server-%{+YYYY.MM}"
 }
 }
 if [type] == "mysql" {
 elasticsearch {
 hosts => ["172.28.29.211:9200"]
 index => "zabbix-mysql -%{+YYYY.MM}" 
 } 
 } 
if[type] == "apache"{
 elasticsearch{
 hosts => ["172.28.29.211:9200"]
 index => "zabbix_apache-%{+YYYY.MM}"
 }
 }
}
 

至此我们配置完成收集到所有zabbix服务器上日志的logstash文件,最后在后台运行logstash:

# nohup /usr/local/logstash-6.2.4/bin/logstash -f logstash-all.conf -w 4 &
 

Kibana上日志展示

Kibana是为Elasticsearch提供的可视化平台,负责数据的美观展示。Kibana服务默认监控在5601端口,浏览器访问可以打开Kibana的界面。左侧导航栏有很多选项,Discover用于和Elasticsearch交互和展示搜索结果,Visualize用于报表生成。

我们新建一个收集zabbix-sever的运行日志。点击工具栏中的Management –> 选择index patterns –> 点击Create Index Pattern按钮。

然后选择一个包含了时间戳的索引字段,可以用来做基于时间的处理。Kibana会读取索引的映射,然后列出所有包含了时间戳的字段。

点击create之后,创建完成。

在DISCOVER就可以选择查看并搜索相应的日志了。

同理我们创建其他的日志,可以在discover左边栏选择fields更加美观的展示日志。

文章来源:智云一二三科技

文章标题:ELK日志分析平台Logstash常用收集各类日志举例

文章地址:https://www.zhihuclub.com/151098.shtml

关于作者: 智云科技

热门文章

网站地图