ES告警之ElastAlert

ES告警之ElastAlert

简介

ElastAlert是一个简单易用的框架，用于从Elasticsearch中的数据发现异常，或其他感兴趣的模式的警报。如果有近乎实时的数据写入Elasticsearch，并且想要在数据与某些模式匹配时收到警报，则ElasticAlert是个不错的工具。

ElastAlert包含几种具有常见监视范例的规则类型：

• “匹配Y时间内至少有X个事件的地方”（frequency类型）
• “当事件发生率增加或减少时匹配”（spike类型）
• “在Y时间内少于X个事件时匹配”（flatline类型）
• “当某个字段与黑名单/白名单匹配时匹配”（blacklist/whitelist类型）
• “任何与过滤器匹配的事件”（any类型）
• “某个字段在一段时间内具有两个不同的值时进行匹配”（change类型）
• “当字段中出现从未见过的term时进行匹配”（new_term类型）
• “当字段的唯一值数量大于或小于阈值时进行匹配”（cardinality类型）

ElastAlert周期性的查询Elastsearch并且将数据传递给规则类型，规则类型定义了需要查询哪些数据。当一个规则匹配触发，就会给到一个或者多个的告警，这些告警具体会根据规则的配置来选择告警途径，就是告警行为，比如邮件、钉钉、企业微信等。

ElastAlert具有几个功能，使其在重新启动或Elasticsearch不可用时更加可靠：

• ElastAlert将其状态保存到Elasticsearch，并且在启动时将恢复先前停止的位置
• 如果Elasticsearch没有响应，则ElastAlert将等到恢复后再继续
• 引发错误的警报可能会在一段时间内自动重试

项目地址：GitHub - Yelp/elastalert: Easy & Flexible Alerting With ElasticSearch 文档地址：ElastAlert - Easy & Flexible Alerting With Elasticsearch — ElastAlert 0.0.1 documentation 样例：elastalert/example_rules at master · Yelp/elastalert · GitHub

安装ElastAlert

git clone https://github.com/Yelp/elastalert.git && cd elastalert
 
pip install -r requirements.txt
 
python setup.py install
 
ln -s /usr/local/python/bin/elastalert* /usr/bin
 
cp config.yaml.example config.yaml

注意：如果你用的是elk5.0, elastalert master还不支持，需要切换git分支到 support_es5

安装钉钉告警插件

elastalert本身不能发送告警到钉钉，需要安装告警插件

wget https://github.com/xuyaoqiang/elastalert-dingtalk-plugin/archive/master.zip
 
unzip master.zip && cd elastalert-dingtalk-plugin-master
 
pip install pyOpenSSL==16.2.0
 
pip install setuptools==46.1.3
 
cp -r elastalert_modules /path/to/elastalert/

如何配置ElastAlert

主配置文件

主配置文件config.yaml解析：

#存放elastalert 规则的文件夹
 
rules_folder: /path/to/elastalert/example_rules

#Elastalert 多久去查询一下根据定义的规则去elasticsearch 查询是否有符合规则的字段，

#如果有就会触发报警，如果没有就等待下一次时间再检查，时间定义的单位从周到秒都可以，具体定义方法如下。

run_every:
 
  #seconds：1
 
  minutes: 1
 
  #hours：1
 
  #days：1
 
  #weeks：1

#当查询开始一直到结束，最大的缓存时间。

buffer_time:
 
  minutes: 15

  

#Elasticsearch ip地址

es_host: 1.2.3.4

#Elasticsearch 的端口

es_port: 9200

#是不是用TLS 加密

#use_ssl: True

#是不是启动TLS证书验证

#verify_certs: True

#如果Elasticsearch 有认证的话需要把这个填写上

#es_username: someusername
 
#es_password: somepassword

#配置证书存放的位置

#verify_certs: True
 
#ca_certs: /path/to/cacert.pem
 
#client_cert: /path/to/client_cert.pem
 
#client_key: /path/to/client_key.key

#就是下文中的elastalert-create-index创建的index

writeback_index: elastalert_status
 
writeback_alias: elastalert_alerts

#如果alert当时没有发出去重试多久之后放弃发送；

alert_time_limit:
 
  days: 2

创建ES索引

elastalert-create-index： ElastAlert 会把执行记录存放到一个 ES 索引中，该命令就是用来创建这个索引的，默认情况下，索引名叫 elastalert_status。其中有 4 个 _type，都有自己的 @timestamp 字段，所以同样也可以用 kibana 来查看这个索引的日志记录情况。这不是必须的步骤，但是强烈建议创建。因为对于审计和测试很有用，并且重启ES不影响计数和发送alert。

elastalert-create-index --config config.yaml

配置告警频率和过滤器

告警频率

#限定时间内，发生事件次数
 
num_events: 3
 
#与上面参数结合使用，表示在2分钟内发生3次就报警
 
timeframe:
 
  minutes: 2

避免重复告警

避免一定时间段中重复告警，可以配置realert和exponential_realert这两个选项：

# 5分钟内相同的报警不会重复发送
 
realert:
 
  minutes: 5
 
# 指数级扩大 realert 时间，中间如果有报警，
 
# 则按照5->10->20->40->60不断增大报警时间到制定的最大时间，
 
# 如果之后报警减少，则会慢慢恢复原始realert时间
 
exponential_realert:
 
  hours: 1

聚合相同告警

根据报警的内容将相同的报警按照 name 来聚合

aggregation_key: name
 
# 聚合报警的内容，只展示 name 与 message
 
summary_table_fields:
 
  - name
 
  - message

为规则编写条件过滤器

过滤器完全按照以下方式传递给Elasticsearch：

filter:
 
  and:
 
    filters:
 
      - [filters from rule.yaml]

与这些过滤器匹配的每个结果都将传递到规则进行处理。

常用过滤器类型

query_string匹配

query_string类型遵循Lucene查询格式，可用于部分或完全匹配多个字段。

filter:
 
- query:
 
    query_string:
 
      query: "username: bob"
 
- query:
 
    query_string:
 
      query: "_type: login_logs"
 
- query:
 
    query_string:
 
      query: "field: value OR otherfield: othervalue"
 
- query:
 
    query_string:
 
       query: "this: that AND these: those"

term匹配

filter:
 
- term:
 
    name_field: "bob"
 
- term:
 
    _type: "login_logs"

terms匹配

filter:
 
- terms:
 
    field: ["value1", "value2"] # value1 OR value2
 
也可以在多个字段进行匹配
 
- terms:
 
    fieldX: ["value1", "value2"]
 
    fieldY: ["something", "something_else"]
 
    fieldZ: ["foo", "bar", "baz"]

通配符匹配

filter:
 
- query:
 
    wildcard:
 
      field: "foo*bar"

范围匹配

filter:
 
- range:
 
    status_code:
 
      from: 500
 
      to: 599

条件匹配

filter:
 
- or:
 
    - term:
 
        field: "value"
 
    - wildcard:
 
        field: "foo*bar"
 
    - and:
 
        - not:
 
            term:
 
              field: "value"
 
        - not:
 
            term:
 
              _type: "something"

下面是一个简单的例子，检查nginx 5xx状态，1分钟内大于5次就发送钉钉告警

name: nginx rule
 
index: xm-nginx-*
 
type: frequency
 
num_events: 5
 
timeframe: {minutes: 1}
 
 
 
filter:
 
- range:
 
    data.status:
 
       from: 500
 
       to: 599
 
 
 
alert_subject: "nginx data.status 500-599"
 
alert_text: "
域   名: {}\n
告警时间: {}\n
调用方式: {}\n
请求链接: {}\n
状 态 码: {}\n
服 务 器: {}\n
"
 
alert_text_type: alert_text_only
 
 
 
alert_text_args:
 
- data.nginx_host
 
- data.timestamp
 
- data.method
 
- data.uri
 
- data.status
 
- data.upstream_addr
 
 
 
alert:
 
- "elastalert_modules.dingtalk_alert.DingTalkAlerter"
 
dingtalk_webhook: "https://oapi.dingtalk.com/robot/send?access_token=你的token"
 
dingtalk_msgtype: "text"

测试规则

elastalert-test-rule rules/nginx.yaml

使用supervisor管理elastalert

Supervisor是用python开发的一套通用的进程管理程序，能将一个普通的命令行进程变为后台daemon，并监控进程状态，异常退出时能自动重启。

[program:elastic-alert]
 
command = /usr/bin/python -m elastalert.elastalert --rule /path/to/elastalert/rules/nginx.yaml --verbose
 
directory= /usr/local/elastalert
 
autostart = true
 
autorestart = true
 
startsecs = 5
 
startretries = 3
 
user = root
 
redirect_stderr = true
 
stdout_logfile=/data/logs/elk/elastic-std.log
 
stderr_logfile=/data/logs/elk/elastic-error.log

常见问题

为什么我的规则没有运行成功？

编写了一条规则并运行了该规则，但是没有任何反应，或者显示为0 query hits。首先建议使用该命令elastalert-test-rule rule.yaml进行调试。它会显示在过去24小时内有多少文档与您的过滤器匹配（或更多，请参阅--help）。如果在Kibana中有一个过滤器，并想在ElastAlert中重新创建它，则可能要使用查询字符串。例如，

filter:
 
- query:
 
    query_string:
 
      query: "foo: bar AND baz: abc*"

如果收到错误，表明Elasticsearch无法对其进行解析，则可能是YAML的间距不正确，或者过滤器的格式不正确。如果你使用Logstash，这是默认设置。例如，

filter:
 
- term:
 
    foo: "Test Document"

我命中了规则，为什么没有收到警报？

如果日志中包含X query hits, 0 matches, 0 alerts sent，则发送警报的动作取决于type。如果你使用type: frequency，则num_events必须与timeframe相互之间发生匹配。不同的规则适用于不同的规则类型。

如果看到X matches, 0 alerts sent，则可能由于多种原因而发生。如果设置了aggregation，则直到该时间过去后才会发送警报。如果你之前已收到针对同一规则的警报，则该规则可能会在一段时间内保持沉默。默认值为两次警报之间的一分钟。如果某个规则被沉默，在Ignoring match for silenced rule日志中可以看到。

如果看到X alerts sent，但没有收到任何警报，则可能与邮件、钉钉的警报配置有关。

当我希望收到多个警报时，为什么只收到一个警报？

设置realert，它是同一规则的两次警报之间的最短时间。在此时间内发生的任何警报都将被丢弃。默认值为一分钟。如果想针对每一个匹配项接收警报，即使它们紧接着发生，也请使用

realert:
 
  minutes: 0

如何防止重复警报？

通过设置realert，可以防止同一条规则在一段时间内发出两次警报。例如，

realert:
 
  days: 1

还可以使用来防止基于某个字段的重复项query_key。例如，为防止同一用户出现多个警报，你可以使用

realert:
 
  hours: 8
 
query_key: user

请注意，这也会影响许多规则类型的工作方式。例如，如果使用type: frequency，则在发送警报之前num_events，query_key必须出现单个值。您也可以将此键使用多个字段的组合。例如，如果只想接收一次针对特定错误和主机名的警报，则可以使用

query_key: [error, hostname]

如何更改警报中的内容？

可以使用字段alert_text将自定义文本添加到警报中。通过设置alert_text_type: alert_text_only，将成为整个警报。还可以使用Python样式字符串格式设置和从警报中添加其他字段alert_text_args。例如

alert_text: "Something happened with {0} at {1}"
 
alert_text_type: alert_text_only
 
alert_text_args: ["username", "@timestamp"]
 
还可以使用来将警报限制为仅包含文档中的某些字段include。
 
include: ["ip_address", "hostname", "status"]

如何在特定时间发出警报？

aggregation功能将记录一段时间内发生的所有警报，并将它们一起发送到一个警报中。可以使用cron样式语法来发送自上一次以来的所有警报

aggregation:
 
  schedule: '2 4 * * mon,fri'

我有很多文件，而且查询很慢，如何加快速度？

有几种方法可以加快查询速度。如果使用index: logstash-*，Elasticsearch将查询所有分片，即使它们可能不包含带有正确时间戳的数据也是如此。相反，你可以使用Python时间格式字符串并设置

use_strftime_index
 
index: logstash-%Y.%m
 
use_strftime_index: true

另外一个设置是buffer_time。默认情况下，ElastAlert将查询大的重叠窗口，以确保它不会丢失任何事件，即使它们是实时索引的。在config.yaml中，您可以调整buffer_time为较小的数字以仅查询最近几分钟。

buffer_time:
 
  minutes: 5