How to Turn ELK into a Real‑Time Network Log Alert System with Zabbix

Introduction

This article upgrades the "ELK Deployment Visual Network Log Analysis Monitoring Platform" to add Zabbix alert integration for abnormal network logs. The open‑source ELK version lacks a native alert module, so logs can only be viewed manually, leading to missed alerts.

Implementation Idea

Use the logstash-output-zabbix plugin to send filtered abnormal logs from Logstash to Zabbix, enabling real‑time alert push.

Install logstash-output-zabbix Plugin

/usr/share/logstash/bin/logstash-plugin install logstash-output-zabbix

Filebeat Configuration

# egrep -v "*#|^$"  /etc/filebeat/filebeat.yml
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /mnt/huawei/*
  tags: ["huawei"]
  include_lines: ['Failed','failed','error','ERROR','\bDOWN\b','\bdown\b','\bUP\b','\bup\b']
  drop_fields:
    fields: ["beat","input_type","source","offset","prospector"]
- type: log
  paths:
    - /mnt/h3c/*
  tags: ["h3c"]
  include_lines: ['Failed','failed','error','ERROR','\bDOWN\b','\bdown\b','\bUP\b','\bup\b']
  drop_fields:
    fields: ["beat","input_type","source","offset","prospector"]
- type: log
  paths:
    - /mnt/ruijie/*
  tags: ["ruijie"]
  include_lines: ['Failed','failed','error','ERROR','\bDOWN\b','\bdown\b','\bUP\b','\bup\b']
  drop_fields:
    fields: ["beat","input_type","source","offset","prospector"]
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
name: 192.168.99.185
setup.template.settings:
  index.number_of_shards: 3
setup.kibana:
output.logstash:
  hosts: ["192.168.99.185:5044"]
processors:
  - add_host_metadata: ~
  - add_cloud_metadata: ~

Logstash Configuration (networklog.conf)

input {
  beats {
    port => 5044
  }
}

filter {
  if "huawei" in [tags] {
    grok {
      match => {"message" => "%{SYSLOGTIMESTAMP:time} %{DATA:hostname} %{GREEDYDATA:info}"}
    }
  } else if "h3c" in [tags] {
    grok {
      match => {"message" => "%{SYSLOGTIMESTAMP:time} %{YEAR:year} %{DATA:hostname} %{GREEDYDATA:info}"}
    }
  } else if "ruijie" in [tags] {
    grok {
      match => {"message" => "%{SYSLOGTIMESTAMP:time} %{DATA:hostname} %{GREEDYDATA:info}"}
    }
  }
  mutate {
    add_field => ["[zabbix_key]", "networklogs"]
    add_field => ["[zabbix_host]", "192.168.99.185"]
    add_field => ["count", "%{hostname}%{info}"]
    remove_field => ["message","time","year","offset","tags","path","host","@version","[log]","[prospector]","[beat]","[input][type]","[source]"]
  }
}

output {
  stdout { codec => rubydebug }
  elasticsearch {
    index => "networklogs-%{+YYYY.MM.dd}"
    hosts => ["192.168.99.185:9200"]
    sniffing => false
  }
  if [count] =~ /(error|ERROR|Failed|failed)/ {
    zabbix {
      zabbix_host => "[zabbix_host]"
      zabbix_key => "[zabbix_key]"
      zabbix_server_host => "192.168.99.200"
      zabbix_server_port => "10051"
      zabbix_value => "count"
    }
  }
}

Logstash Parameter Explanation

add_field => ["[zabbix_key]", "networklogs"]   # adds field zabbix_key with value networklogs
add_field => ["[zabbix_host]", "192.168.99.185"] # host name must match the host defined in Zabbix template
add_field => ["count", "%{hostname}%{info}"]   # combines hostname and log info for distinguishing devices in Zabbix
if [count] =~ /(error|ERROR|Failed|failed)/ {   # filters abnormal keywords and pushes to Zabbix }

Testing Field Splitting and Merging

Zabbix Web Configuration

Create a template, application set, monitoring item, and trigger as shown in the screenshots below.

Zabbix‑sender Installation and Test

yum install zabbix-sender
# Test sending data to Zabbix
zabbix_sender -s 192.168.99.185 -z 192.168.99.200 -k "networklogs" -o 1 -vv
# Expected response
zabbix_sender [2444]: DEBUG: answer [{"response":"success","info":"processed: 1; failed: 0; total: 1; seconds spent: 0.000057"}]
Response from "192.168.99.200:10051": "processed: 1; failed: 0; total: 1; seconds spent: 0.000057"
Sent: 1; Skipped: 0; Total: 1

Viewing Data in Zabbix and Kibana