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 \"大众http://192.168.33.29/zabbix/zabbix.php?action=dashboard.view\"大众 \"大众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端口，浏览器访问http://IP:5601可以打开Kibana的界面。左侧导航栏有很多选项，Discover用于和Elasticsearch交互和展示搜索结果，Visualize用于报表天生。
我们新建一个网络zabbix-sever的运行日志。点击工具栏中的Management --> 选择index patterns --> 点击Create Index Pattern按钮。
然后选择一个包含了韶光戳的索引字段，可以用来做基于韶光的处理。Kibana会读取索引的映射，然后列出所有包含了韶光戳的字段。
点击create之后，创建完成。
在DISCOVER就可以选择查看并搜索相应的日志了。
同理我们创建其他的日志，可以在discover左边栏选择fields更加都雅的展示日志。
