让您了解zabbix集成了ELK来采集系统异常日志以触发警报〜

优采云发布时间: 2020-08-08 02:57

　　让我们今天了解ELK的“ L” -Logstash. 是的，这就是神奇的小组成部分. 众所周知，它是ELK不可或缺的组成部分. 它完成输入，过滤和输出. （输出）工作量也是我们作为运维人员需要掌握的困难. 说到这一点，我们充满爱与恨. “爱是美好，仇恨是困难的”；这个Logstash具有强大的插件功能，除了对我们进行过滤外，高效的日志输出还可以帮助我们与Zabbix监视集成吗？

　　由于我们的Logstash支持多种输出类型，因此它可以采集Web服务日志，系统日志和内核日志；但是，有日志输出，这肯定无法避免错误日志的出现；当出现错误日志时尽管可以通过ELK找到它，但ELK无法提供实时警报，这有点尴尬. 我们要做的是既要像zabbix和nagios一样进行监控，也要发出警报. ELK仅对此进行监视，但不对其发出警报；但是没关系，我们的Logstash插件可以与zabbix结合使用，以采集需要警报的日志（例如，带有错误标识的日志）以完成日志监视并触发警报〜

　　Logstash支持多种输出介质，例如syslog，http，tcp，elasticsearch，kafka等. 如果我们将logstash采集的日志输出到zabbix警报，则必须使用logstash-output-zabbix插件，并通过此插件集成使用zabbix的logstash，过滤logstash采集的数据，将错误信息的日志输出到zabbix，最后通过zabbix告警机制触发；

　　[root@localhost ~]# /usr/local/logstash/bin/logstash-plugin install logstash-output-zabbix #安装logstash-output-zabbix插件

Validating logstash-output-zabbix

Installing logstash-output-zabbix

Installation successful

　　环境案例要求:

　　通过读取系统日志文件监控信息，过滤掉日志信息中的错误关键字，如ERR，错误，失败，警告等信息，用异常关键字过滤掉这些异常日志信息，然后输出到zabbix，通过zabbix警报机制触发警报；以下环境为filebeat作为采集终端；输出到kafaka消息队列，最后将日志由logsatsh提取和过滤，并输出到zabbix

　　[filebeat]日志采集终端

　　filebeat.inputs:

- type: log

enabled: true

paths:

- /var/log/secure

- /var/log/messages

- /var/log/cron

fields:

log_topic: system_log

processors:

- drop_fields:

fields: ["beat", "input", "source", "offset", "prospector"] #这里在filebeat中直接去掉不需要的字段。

filebeat.config.modules:

path: ${path.config}/modules.d/*.yml

reload.enabled: false

name: 192.168.37.147 #这是日志输出标识，表明日志来自哪个主机，后面再logstash会用到。

output.kafka:

enabled: true

hosts: ["192.168.37.147:9092", "192.168.37.148:9092", "192.168.37.149:9092"] #日志输出到kafka集群

version: "0.10"

topic: '%{[fields.log_topic]}'

partition.round_robin:

reachable_only: true

worker: 2

required_acks: 1

compression: gzip

max_message_bytes: 10000000

logging.level: debug

　　[Logstash端]

　　[root @ localhost〜]#vim /usr/local/logstash/config/etc/system_log.conf

　　input {

kafka {

bootstrap_servers => "192.168.37.147:9092,192.168.37.148:9092,192.168.37.149:9092"

topics => ["system_log"]

codec => "json"

}

filter {

if [fields][log_topic] == "system_log" { #指定filebeat产生的日志主题

mutate {

add_field => [ "[zabbix_key]", "oslogs" ] #新增的字段，字段名是zabbix_key，值为oslogs。

add_field => [ "[zabbix_host]", "%{[host][name]}" ] #新增的字段，字段名是zabbix_host，值可以在这里直接定义，也可以引用字段变量来获取。这里的%{[host][name]获取的就是日志数据的来源IP，这个来源IP在filebeat配置中的name选项进行定义。

}

grok {

match => { "message" => "%{SYSLOGTIMESTAMP:message_timestamp} %{SYSLOGHOST:hostname} %{DATA:message_program}(?:\[%{POSINT:message_pid}\])?: %{GREEDYDATA:message_content}" } #这里通过grok对message字段的数据进行字段划分，这里将message字段划分了5个子字段。其中，message_content字段会在output中用到。

}

mutate { #这里是删除不需要的字段

remove_field => "@version"

remove_field => "message"

}

date { #这里是对日志输出中的日期字段进行转换，其中message_timestamp字段是默认输出的时间日期字段，将这个字段的值传给 @timestamp字段。

match => [ "message_timestamp","MMM d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601"]

}

output {

if [message_content] =~ /(ERR|error|ERROR|Failed)/ { #定义在message_content字段中，需要过滤的关键字信息，也就是在message_content字段中出现给出的这些关键字，那么就将这些信息发送给zabbix。

zabbix {

zabbix_host => "[zabbix_host]" #这个zabbix_host将获取上面filter部分定义的字段变量%{[host][name]的值

zabbix_key => "[zabbix_key]" #这个zabbix_key将获取上面filter部分中给出的值

zabbix_server_host => "192.168.37.149" #这是指定zabbix server的IP地址

zabbix_server_port => "10051" #这是指定zabbix server的*敏*感*词*端口

zabbix_value => "message_content" #定要传给zabbix监控项item（oslogs）的值， zabbix_value默认的值是"message"字段，因为上面我们已经删除了"message"字段，因此，这里需要重新指定，根据上面filter部分对"message"字段的内容划分，这里指定为"message_content"字段，其实，"message_content"字段输出的就是服务器上具体的日志内容。

}

　　[root @ localhost logstash]#nohup / usr / local / logstash / bin / logstash -f config / etc / system_log.conf --path.data = / tmp /#在此，--path.data指定单词logstash进程的数据存储目录，用于在服务器上启动多个logstash进程环境