[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

filebeat.inputs:
- type: logenabled: truepaths:- /var/log/secure- /var/log/messages- /var/log/cronfields:log_topic: system_log
processors:- drop_fields:fields: ["beat", "input", "source", "offset", "prospector"] #这里在filebeat中直接去掉不需要的字段。
filebeat.config.modules:path: ${path.config}/modules.d/*.ymlreload.enabled: false
name: 192.168.37.147        #这是日志输出标识，表明日志来自哪个主机，后面再logstash会用到。
output.kafka:enabled: truehosts: ["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: trueworker: 2required_acks: 1compression: gzipmax_message_bytes: 10000000
logging.level: debug

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"字段输出的就是服务器上具体的日志内容。}}
}