一文吃透企业级elk技术栈:11. zabbix报警实现

本文涉及的产品
检索分析服务 Elasticsearch 版,2核4GB开发者规格 1个月
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
简介: 一文吃透企业级elk技术栈:11. zabbix报警实现

十一、zabbix报警实现

mysql多次登录失败告警 // 存在问题 日志信息无法打印到zabbix

[root@elk-02 ~]# cat /etc/logstash/conf.d/mysql.conf
input {
  kafka {
    bootstrap_servers => "10.10.8.10:9092,10.10.8.11:9092,10.10.8.12:9092"
    topics => ["elktest"]
    group_id => "elkgroup"
    codec => "json"
  }
}
filter {
  if [type] == "mysql-slow-logs" {
    grok {
        # 有ID有use
        match => [ "message", "^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id:\s+%{NUMBER:id:int}\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nuse\s(?<dbname>\w+);\nSET\s+timestamp=%{NUMBER:timestamp_mysql};\n(?<query>[\s\S]*)" ]
        # 有ID无use
        match => [ "message", "^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id:\s+%{NUMBER:id:int}\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nSET\s+timestamp=%{NUMBER:timestamp_mysql};\n(?<query>[\s\S]*)" ]
        # 无ID有use
        match => [ "message", "^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nuse\s(?<dbname>\w+);\nSET\s+timestamp=%{NUMBER:timestamp_mysql};\n(?<query>[\s\S]*)" ]
        # 无ID无use
        match => [ "message", "^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nSET\s+timestamp=%{NUMBER:timestamp_mysql};\n(?<query>[\s\S]*)" ]
        # mariadb慢日志获取
        match => [ "message", "^#\s+User@Host:\s+%{USER:user}\[[^\]]+\]\s+@\s+(?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\n# Thread_id:%{NUMBER:thread_id:int}\s+Schema: %{DATA:schema}\s+QC_hit: %{DATA:qc_hit}\n# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\nSET\s+timestamp=%{NUMBER:timestamp_mysql};\n(?<query>[\s\S]*)" ]
    }
    date {
            match => ["timestamp_mysql","UNIX"]
            target => "@timestamp"
    }
    mutate {
            remove_field => ["@version","message","timestamp_mysql"]
    }
  }
  else if [type] == "mysql-err-logs" {
    grok {
     # mysql 5.7 err
      match => [ "message", "(?m)^%{NOTSPACE:date} %{NUMBER:bytes} \[%{GREEDYDATA:log_level}\] %{GREEDYDATA:messages}" ]
      # mysql 5.6 err
      match => [ "message", "(?<timestamp>\d+ \d+:\d+:\d+) \[%{LOGLEVEL:log_level}\] %{GREEDYDATA:messages}\s*$" ]
      # mariadb  err1
      match => [ "message", "(?<timestamp>\d+ \d+:\d+:\d+) \[%{DATA:log_level}\] %{GREEDYDATA:messages}\s*$" ]
      # mariadb err2
      match => [ "message", "(?<timestamp>\d+ \d+:\d+:\d+) %{GREEDYDATA:messages}\s*$" ]
    }
    mutate {
      add_field => ["[zabbix_host]","10.10.8.152"]
      add_field => ["[zabbix_key]","mysql_passwd_verification_failed"]
    }
    mutate {
      add_field => ["count","%{[agent][name]}_%{messages}"]
    }
    mutate {
     remove_field => [ "@version" ]
    }
  }
}
output {
  if [type] == "mysql-slow-logs" {
  # 推送到es
    elasticsearch {
      hosts => ["http://10.10.8.10:9200","http://10.10.8.11:9200","http://10.10.8.12:9200"]
      user => "elastic"
      password => "123456"
      index => "mysql-slow-%{+YYYY.MM.dd}"
    }
  }
  else if [type] == "mysql-err-logs" {
    elasticsearch {
      hosts => ["http://10.10.8.10:9200","http://10.10.8.11:9200","http://10.10.8.12:9200"]
      user => "elastic"
      password => "123456"
      index => "mysql-err-%{+YYYY.MM.dd}"
    }
    if [count]  =~ /(Access denied)/ {
      zabbix {
        zabbix_host => "[zabbix_host]"
        zabbix_key => "[zabbix_key]"
        zabbix_server_host => "10.10.8.166"
        zabbix_server_port => "10051"
        zabbix_value => "count"
      }
    }
  }
}

客户端配置

centos 7安装

systemctl status filebeat
cd /root && \
wget  https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.9.2-x86_64.rpm && \
rpm -ivh filebeat-7.9.2-x86_64.rpm && \
systemctl enable filebeat.service && \
systemctl start filebeat.service && \
mkdir /etc/filebeat/conf.d

centos 6安装

systemctl status filebeat
cd /root && \
wget  https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.9.2-x86_64.rpm && \
rpm -ivh filebeat-7.9.2-x86_64.rpm && \
service filebeat start && \
chkconfig  --add  filebeat && \
mkdir /etc/filebeat/conf.d

配置文件配置

cat >/etc/filebeat/filebeat.yml<<'EOF'
filebeat.config.inputs:
  enabled: true
  # 配置多配置文件路径
  path: /etc/filebeat/conf.d/*.yml
processors:
  - drop_fields:
      fields: ["source","input","beat","prospector","offset"]
# 区分日志来自哪台主机 // 自行修改当前主机IP
name: 103.29.16.83
output:
  # 输出到kafka
  kafka:
    hosts: ["10.10.8.164:9092", "10.10.8.165:9092", "10.10.8.166:9092"]
    topic: elktest
logging.level: info
logging.to_files: true
logging.files:
  path: /var/log/filebeat
  name: filebeat
  keepfiles: 7
  permissions: 0644
#output.file:
#  path: "/tmp/filebeat"
#  filename: filebeat
EOF

mysql日志配置

netstat -lntp|grep mysqld
tcp6       0      0 :::3306                 :::*                    LISTEN      166237/mysqld
tcp        0      0 0.0.0.0:3307            0.0.0.0:*               LISTEN      1853/mysqld
tcp        0      0 0.0.0.0:3309            0.0.0.0:*               LISTEN      181145/mysqld
tcp        0      0 0.0.0.0:3310            0.0.0.0:*               LISTEN      261336/mysqld
tcp6       0      0 :::3311                 :::*                    LISTEN      85491/mysqld
tcp        0      0 0.0.0.0:3316            0.0.0.0:*               LISTEN      127655/mysqld
tcp        0      0 0.0.0.0:3326            0.0.0.0:*               LISTEN      27277/mysqld
cat >/etc/filebeat/conf.d/mysql.yml<<'EOF'
# mysql 慢日志推送
- type: log
  tail_files: true
  backoff: "1s"
  paths:
  # 多实例推送 // 自行修改慢日志路径
  - /data/mysql3309/log/mysql-slow.log
  - /data/mysql3310/log/mysql-slow.log  
  - /data/mysql3316/log/mysql-slow.log
  - /data/mysql3326/log/slow.log
  tags: ["mysql-slow-logs"]
  # 排除列
  exclude_lines: ['^\# Time']
  fields:
    # 配置logstash区分
    type: "mysql-slow-logs"
  fields_under_root: true
  multiline:
    # 多行匹配
    pattern: '^\# Time|^\# User'
    negate: true
    match: after
# mysql 错误日志推送
- type: log
  tail_files: true
  backoff: "1s"
  paths:
  # 多实例推送 // 自行修改错误日志路径
  - /var/log/mysqld.log
  - /data/mysql3307/log/mysql-error.log
  - /data/mysql3309/log/mysql-error.log
  - /data/mysql3310/log/mysql-error.log
  - /data/mysql3311/error.log
  - /data/mysql3316/log/mysql-error.log
  - /data/mysql3326/log/mysql-error.log
  tags: ["mysql-err-logs"]
  fields:
    # 配置logstash区分
    type: "mysql-err-logs"
  fields_under_root: true
EOF

系统日志

cat >/etc/filebeat/conf.d/system.yml<<'EOF'
# 系统messages日志推送
- type: log
  tail_files: true
  backoff: "1s"
  paths:
   - /var/log/messages
  tags: ["system-messages-logs"]
  fields:
    # 配置logstash区分
    type: "system-messages-logs"
  fields_under_root: true
# 系统secure日志推送
- type: log
  tail_files: true
  backoff: "1s"
  paths:
   - /var/log/secure
  tags: ["system-secure-logs"]
  fields:
    # 配置logstash区分
    type: "system-secure-logs"
  fields_under_root: true
EOF

重启

systemctl restart filebeat.service

防火墙规则

telnet  10.10.8.164 9092
ip a

问题解决

1. “error”=>{“type”=>“validation_exception”, “reason”=>“Validation Failed: 1: this action would add [2] shards, but this cluster currently has [999]/[1000] maximum normal shards open;”}

PUT _cluster/settings
{
  "persistent": {
    "cluster": {
      "max_shards_per_node":20000
    }
  }
}

es的每台机器的默认最大分片数位1000,如果集群有3台机器,那就有3000当满了之后就无法创建索引,所以需要设置更大的参数

vim /etc/elasticsearch/elasticsearch.yml
cluster.max_shards_per_node: 20000


相关实践学习
使用阿里云Elasticsearch体验信息检索加速
通过创建登录阿里云Elasticsearch集群,使用DataWorks将MySQL数据同步至Elasticsearch,体验多条件检索效果,简单展示数据同步和信息检索加速的过程和操作。
ElasticSearch 入门精讲
ElasticSearch是一个开源的、基于Lucene的、分布式、高扩展、高实时的搜索与数据分析引擎。根据DB-Engines的排名显示,Elasticsearch是最受欢迎的企业搜索引擎,其次是Apache Solr(也是基于Lucene)。 ElasticSearch的实现原理主要分为以下几个步骤: 用户将数据提交到Elastic Search 数据库中 通过分词控制器去将对应的语句分词,将其权重和分词结果一并存入数据 当用户搜索数据时候,再根据权重将结果排名、打分 将返回结果呈现给用户 Elasticsearch可以用于搜索各种文档。它提供可扩展的搜索,具有接近实时的搜索,并支持多租户。
相关文章
|
2月前
|
监控
一文吃透企业级elk技术栈:9. zabbix结合logstash告警
一文吃透企业级elk技术栈:9. zabbix结合logstash告警
|
2月前
|
消息中间件 Kafka 网络安全
一文吃透企业级elk技术栈:elk 各组件调试
调试需先理解逻辑与程序调用顺序。本文介绍filebeat、kafka、logstash和es的数据推送流程及调试方法:filebeat传输数据检查包括服务状态、配置与日志;kafka调试涵盖服务状态、端口与日志;logstash调试需检查配置文件、日志与流量;es直接通过kibana查看。还介绍了使用rsyslog接收防火墙/waf/交换机日志的方法。
一文吃透企业级elk技术栈:10. es数据生命周期管理
一文吃透企业级elk技术栈:10. es数据生命周期管理
|
5月前
|
存储 监控 数据可视化
日志分析对决:揭示 ELK 与 GrayLog 的优势和差异
日志分析对决:揭示 ELK 与 GrayLog 的优势和差异
1054 0
|
5月前
|
存储 Prometheus 监控
Prometheus vs. ELK Stack:容器监控与日志管理工具的较量
随着容器化技术的广泛应用,容器监控与日志管理成为了关键任务。本文将对两种常用工具进行比较与选择,分别是Prometheus和ELK Stack。Prometheus是一款开源的监控系统,专注于时序数据的收集和告警。而ELK Stack则是一套完整的日志管理解决方案,由Elasticsearch、Logstash和Kibana三个组件组成。通过比较它们的特点、优势和适用场景,读者可以更好地了解如何选择适合自己需求的工具。
|
8天前
|
存储 消息中间件 网络协议
日志平台-ELK实操系列(一)
日志平台-ELK实操系列(一)
|
2月前
|
消息中间件 Kafka 开发工具
rsyslog+ELK收集Cisco日志
rsyslog+ELK收集Cisco日志
|
2月前
|
运维 监控 Ubuntu
一键启动日志魔法:揭秘ELK自动安装脚本的神秘面纱!
【8月更文挑战第9天】在数据驱动时代,高效处理日志至关重要。ELK Stack(Elasticsearch、Logstash、Kibana)是强大的日志分析工具,但其复杂的安装配置常让初学者望而却步。本文介绍如何编写ELK自动安装脚本,简化部署流程。脚本适用于Ubuntu系统,自动完成ELK下载、安装及基本配置,包括依赖项安装、服务启动及自启设置,极大降低了使用门槛,助力运维人员和开发者轻松构建日志分析平台。
106 6
|
2月前
|
存储 消息中间件 监控
Java日志详解:日志级别,优先级、配置文件、常见日志管理系统ELK、日志收集分析
Java日志详解:日志级别,优先级、配置文件、常见日志管理系统、日志收集分析。日志级别从小到大的关系(优先级从低到高): ALL < TRACE < DEBUG < INFO < WARN < ERROR < FATAL < OFF 低级别的会输出高级别的信息,高级别的不会输出低级别的信息
|
2月前
|
存储 应用服务中间件 nginx
部署ELK+filebeat收集nginx日志
部署ELK+filebeat收集nginx日志
部署ELK+filebeat收集nginx日志

推荐镜像

更多
下一篇
无影云桌面