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

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

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

1.file插件收集日志文件

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

#cat logstash-zabbixserver.confinput { 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.confinput { 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.confinput {	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 514tcp6 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.confinput{ 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 onlogging 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里会将所有内容都写在messages字段里。如果我们想把不同字段做分类的话，就可以用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.1method: GETrequest: /index.htmlbytes: 15824duration: 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.confinput{ 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.confinput { 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更加美观的展示日志。