部署 ElastAlert

# ElastAlert 在数据与特定模式匹配时发送警告。是可靠、模块化、易配置的工具
# 通过将 Elasticsearch 与两种类型组件: 规则、警报结合使用，定期执行查询并将数据传递到规则
# 首次运行前要使用其提供的可执行文件 "elastalert-create-index" 创建相关索引，索引名: elastalert_status
# ElastAlert 的 Kibana UI 插件: https://github.com/bitsensor/elastalert-kibana-plugin/releases

# ElastAlert 能保证在其重启或 Elasticsearch 不可用时更可靠
#   1.默认将状态数据存到 Elasticsearch 并在启动时恢复
#   2.若 Elasticsearch 无响应则等到其恢复后再继续执行
#   3.引发错误的警报可能会在一段时间内自动重试
#   4.状态索引允许对ElastAlert的操作进行某种程度的审核和调试，还能避免在重启或崩溃时丢失数据或重复报警

# ------------------------------------------------------- SETUP

# 安装依赖 & 下载源码
yum -y install wget openssl openssl-devel gcc gcc-c++
git clone https://github.com/Yelp/elastalert.git
pip install "setuptools>=11.3"
python setup.py install

# 在 Elasticsearch 端创建 elastalert 需要的索引及映射
# 关于状态索引的字段说明: https://elastalert.readthedocs.io/en/latest/elastalert_status.html
./create_index.py --host ${ES_IP} --port ${ES_PORT} --username 'xxx' --password 'xxx' --config ./config.yaml

配置文件说明

# ------------------------------------------------------- config.yaml
# https://elastalert.readthedocs.io/en/latest/elastalert.html#configuration

# 包含规则配置文件的文件夹名称，将载入其中的所有 *.yaml 文件 ( 包括其子目录 )
rules_folder: myrules

# 是否递归规则目录
scan_subdirectories：true

# 执行查询的间隔 (ElastAlert会记住上次针对给定规则运行查询的时间，并从那时起定期查询直到现在)
run_every:
  minutes: 5

# 查询的时间窗口（此处为当前时间前30分钟到当前时间的范围内的数据）
# Tips: 这可能会被个别规则替代：use_count_query 或 use_terms_query 为 true 的规则将忽略此选项
buffer_time:
  minutes: 30
# hours: <int>
# days: <int>
# seconds: <int>

# --------------------------------- # 避免重复告警 ...
# realert:                          # 5分钟内相同的报警不会重复发送
#   minutes: 5                      # 
# exponential_realert:              # 指数级扩大 realert，中间若有报警则按 5>10>20>40>60 不断增大报警时间到指定的最大时间
#     hours: 1                      # Tips: 若之后报警减少则慢慢恢复到原始的 realert 时间
# --------------------------------- # 
# aggregation:                      # 此选项允许将多个匹配项聚合到一个警报，每次找到匹配项时将等待 aggregation 时段
#   hours: 2                        # 并将在该时段内发生的所有匹配打包发给特定规则
#                                   # 若预期有大量匹配且只想要定期报告，将非常有用
# aggregation:                      # Cron 语法 ...
#   schedule: '2 4 * * mon,fri'     # 若希望汇总所有警报并定期发送，可使用 schedule 进行定义
# aggregation_key: 'mydata.user'    # 默认情况下，聚合窗口期间发生的所有事件都组合在一起，若设置了该选项
#   则共享一个公共键值的每个事件将被分组在一起，将为每个新遇到的键值创建一个单独的聚合窗口 ...
#   例如上例为：希望接收按触发事件的用户分组的警报 ...
# ---------------------------------

# Elasticsearch 的集群连接配置
es_host: localhost                  # 环境变量 ES_HOST、ES_PORT 将覆盖此字段
es_port: 9200                       # 
es_username: elasticsearch          # 环境变量 ES_USERNAME、ES_PASSWORD 将覆盖此字段
es_password: changeme               # 
es_conn_timeout: 20                 # 超时设置
use_ssl: "false"                    # 若使用TLS则还需设置: verify_certs,client_cert,client_key,ca_certs
writeback_index: elastalert_status  # Elastalert 状态索引的名称
# es_send_get_body_as: GET          # Elasticsearch 查询方法（可选）: GET、POST、source，默认 GET

# max_query_size: 10000             # 单个查询从Elasticsearch的最大文档数，默认 10000
# 若希望接近该值请考虑使用 use_count_query
# 若达到此限制则在直到处理所有结果之前使用设置的页面大小进行滚动: max_query_size、max_scrolling_count

# max_scrolling_count: 0            # 要滚动的最大页面数，默认 0 表示无限制
# 若将此值设为5并将 max_query_size 设 10000 则最多下载 50000 个文档

# ------------------------------------------------------- 运行

# 检查规则文件
./elastalert-test-rule --config ./config.yaml ./example_rules/example.yaml

# 运行
python elastalert/elastalert.py --config config.yaml \
    [ --verbose|--debug ] \                 # 
    [ --start YYYY-MM-DDTHH:MM:SS ] \       # 从给定时间开始进行查询
    [ --end YYYY-MM-DDTHH:MM:SS ] \         # 在指定的时间戳停止查询，默认无限制
# No handlers could be found for logger "Elasticsearch"
# INFO:root:Queried rule Example rule from 1-15 14:22 PST to 1-15 15:07 PST: 5 hits
# 定期查询最近 buffer_time（默认45分钟）的数据并与过滤器进行匹配，在这里它匹配了5个命中
# INFO:Elasticsearch:POST http://elasticsearch.example.com:14900/elastalert_status/elastalert_status?op_type=create [status:201 request:0.025s]
# 此行显示其将文档上传到 elastalert_status 索引，其中包含有关它刚进行的查询的信息
# INFO:root:Ran Example rule from 1-15 14:22 PST to 1-15 15:07 PST: 5 query hits (0 already seen), 0 matches, 0 alerts sent
# 该行表示已完成规则的处理，在较长的时间段内有时可能会运行多个查询，但是它们的数据将一起处理
# INFO:root:Sleeping for 297 seconds
# 默认 run_every 为5分钟，即ElastAlert将休眠直到从上个周期开始经过5分钟为止，然后再对每个规则查询，同时时间范围向前移动5分钟

告警规则的类型

# ElastAlert 包含几种具有通用监视范例的规则类型:

frequency:              # 匹配Y时间有X个事件的地方
spike:                  # 在事件发生率增加或减少时匹配
flatline:               # 在Y时间内少于X个事件时匹配
blacklist/whitelist:    # 当某个字段与黑名单/白名单匹配时 (常用)
any:                    # 与给定过滤器匹配的任何事件 (常用)
change:                 # 当某个字段在一段时间内具有两个不同的值时匹配 (常用)

# Elastalert 的索引中 hits 表示规则命中数，matches 表示规则命中数，并且匹配规则触发告警数量
# 告警规则通用配置项: https://elastalert.readthedocs.io/en/latest/ruletypes.html

# --------------------------------------------------------- 示例数据

# 以下规则类型均使用该文档样本作触发告警：
doc = {
    "@timestamp": get_now(),
    "codec": "nodejs",
    "tags": "31",
    "level": "high",
    "server": "nginx",
    "status": "anystatus",
    "message": ">>> [ xxx ]: valid id error ."
}

# --------------------------------------------------------- any 类型

# 说明：任何规则都会匹配，查询返回的每个命中均生成一个警报
# 规则：当 status 字段匹配为 anystatus 则触发告警
name: any_rule
type: any                   # 规则的类型
index: testalert-*          # 监控的索引
timeframe:                  # 监控时间1分钟内
    minutes: 1
filter:                     # 用过滤器查询列表，format {'filter': {'bool': {'must': [config.filter]}}}
- term:                     # 查询采用的是 Elastic DSL 语法
    status: "anystatus"
alert: post                 # 告警方式（可以是列表）
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: any_rule     # 添加到POST包中的数据，规则名称
    rule_level: medium      # 添加到POST包中的数据，告警级别

# POST 结果
# {
#     "status": "anystatus",
#     "_type": "mydata",
#     "level": "high",
#     "num_hits": 5,
#     "@timestamp": "2018-01-31T02:26:52.268477Z",
#     "server": "nginx",
#     "rule_name": "any_rule",
#     "rule_level": "medium",
#     "_index": "testalert",
#     "num_matches": 5,
#     "message": ">>> [ xxx ]: valid id error .",
#     "_id": "AWFKCd4a5xzN_sFQhZgO",
#     "codec": "nodejs",
#     "tags": "31"
# }

# --------------------------------------------------------- blacklist 类型

# 说明：黑名单规则将检查黑名单中的某个字段，若其在黑名单中则匹配
# 规则：当该 status 字段匹配到关键字 hacker、huahua、... 则触发告警
name: blacklist_rule
type: blacklist
index: testalert
timeframe:
    minutes: 1
compare_key: status     # 用于与黑名单进行比较的字段名称。如果该字段为null，则这些事件将被忽略
blacklist:              # 列入黑名单的值的列表，和/或包含使用列入黑名单的值的文件的路径的列表
    - "hacker"
    - "huahua"
    - "!file /tmp/blacklist1.txt"
    - "!file /tmp/blacklist2.txt"
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: blacklist_rule
    rule_level: medium

# --------------------------------------------------------- whitelist 类型

# 与黑名单类似，此规则将某个字段与白名单进行比较，如果列表中不包含该字词则匹配

# --------------------------------------------------------- change 类型

# 说明：监视某字段并在其值变更时匹配，其必须相对于最后一个事件发生相同的变化
# 规则：当 server 字段值相同，但 codec 字段值不同时触发告警
name: change_rule
type: change
index: testalert
timeframe:
    minutes: 1
compare_key: codec          # 与上一条记录做对比的字段
query_key: server           # 与上一条记录相同的字段
ignore_null: true           # 忽略记录不存在 compare_key 字段的情况
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: change_rule
    rule_level: medium

# POST 结果
# {
#     "status": "up",
#     "_type": "mydata",
#     "_id": "AWFKIgZA5xzN_sFQhZh5",
#     "tags": "31",
#     "num_hits": 4,
#     "@timestamp": "2018-01-31T02:53:15.413240Z",
#     "rule_level": "medium",
#     "old_value": [
#         "nodejs"
#     ],
#     "server": "nginx",
#     "rule_name": "change_rule",
#     "_index": "testalert",
#     "new_value": [
#         "java"
#     ],
#     "num_matches": 1,
#     "message": ">>> [ xxx ]: valid id error .",
#     "level": "high",
#     "codec": "java"
# }

# --------------------------------------------------------- frequency 类型

# 说明：当给定时间范围内有至少一定数量的事件时匹配，可按每个 query_key 计数
# 规则：当 status 字段匹配到关键字 frequency >= 3次则触发
name: frequency_rule
type: frequency
index: testalert
num_events: 3                   # 3次
timeframe:
    minutes: 1
filter:
  - term:
      status: "frequency"       # 关键字匹配
# - query_string:
#     query: "system.process.cpu.total.pct: > 10%"
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: frequency_rule
    rule_level: medium

# POST 结果
# {
#     "status": "frequency",
#     "_type": "mydata",
#     "level": "high",
#     "num_hits": 3,
#     "@timestamp": "2018-01-31T03:28:00.793290Z",
#     "rule_level": "medium",
#     "server": "nginx",
#     "rule_name": "frequency_rule",
#     "_index": "testalert",
#     "num_matches": 1,
#     "message": ">>> [ xxx ]: valid id error .",
#     "_id": "AWFKQdg_5xzN_sFQhZjW",
#     "codec": "java",
#     "tags": "31"
# }
# --------------------------------------------------------- spike 类型

# 说明：当某个时间段内的事件量比上个时间段的 spike_height 时间大或小时匹配
# 它使用两个滑动窗口来比较事件的当前和参考频率，可以将这两个窗口称为 "参考" 和 "当前"
# 规则：当前窗口数据量为3，当前窗口超过参考窗口数据量次数1次则触发告警
name: spike_rule
type: spike
index: testalert
timeframe:
    minutes: 1
threshold_cur: 3        # 当前窗口初始值
spike_height: 1         # 当前窗口数据量连续比参考窗口数据量高/低的次数
spike_type: "up"        # 高或低
filter:
- term:
    status: "spike"
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: spike_rule
    rule_level: medium
# POST 结果
# {
#     "status": "spike",
#     "_type": "mydata",
#     "_id": "AWFLMbye5xzN_sFQhZlk",
#     "tags": "31",
#     "num_hits": 13,
#     "@timestamp": "2018-01-31T07:50:02.382708Z",
#     "rule_level": "medium",
#     "server": "nginx",
#     "rule_name": "spike_rule",
#     "_index": "testalert",
#     "spike_count": 8,
#     "reference_count": 0,
#     "num_matches": 1,
#     "message": ">>> [ xxx ]: valid id error .",
#     "level": "high",
#     "codec": "java"
# }

# --------------------------------------------------------- flatline 类型

# 说明：当一个时间段内的事件总数低于给定阈值时匹配
# 规则：当信息量低于3条时触发告警
name: flatline_rule
type: flatline
index: testalert
timeframe:
    minutes: 1          # 监控时间1分钟内
threshold: 3            # 阈值为3
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: flatline_rule
    rule_level: medium
# POST 结果
# {
#     "count": 1,
#     "num_hits": 1,
#     "@timestamp": "2018-01-31T09:02:35.720517Z",
#     "rule_level": "medium",
#     "rule_name": "flatline_rule",
#     "key": "all",
#     "num_matches": 1
# }

# --------------------------------------------------------- cardinality 类型

# 说明：当一个时间范围内的特定字段的唯一值的总数高于或低于阈值时匹配
# 规则：1分钟内 level 的唯一数量 > 2 则触发告警
name: test_rule
index: testalert
type: cardinality
timeframe:
    minutes: 1                  # 区间
cardinality_field: level        # 字段
max_cardinality: 2              # 阈值
alert: post
http_post_url: "http://localhost:8088/api/alert"
http_post_static_payload:
    rule_name: test_rule
    rule_level: medium
# POST 结果
# {
#     "status": "cardinality",
#     "_type": "mydata",
#     "level": "info",
#     "num_hits": 3,
#     "@timestamp": "2018-01-31T09:17:02.276937Z",
#     "rule_level": "medium",
#     "server": "nginx",
#     "rule_name": "cardinality_rule",
#     "_index": "testalert",
#     "num_matches": 1,
#     "message": ">>> [ xxx ]: valid id error .",
#     "_id": "AWFLgWKw5xzN_sFQhZvg",
#     "codec": "java",
#     "tags": "31"
# }
# --------------------------------------------------------- percentage match 类型

# 说明：当计算窗口内的匹配桶中的文档的百分比高于或低于阈值时匹配。计算窗口默认为 buffer_time
# 规则：当level字段为high，时间窗口内日志量高于前一个时间窗口95%，触发告警（未完整测试）
name: percentage_match_rule
type: percentage_match
index: testalert
buffer_time:
    minutes: 1
max_percentage: 95          # 高于前一个时间窗口95%
match_bucket_filter:
- term:
    level: high             # 当 level 字段为 high
doc_type: mydata            # _doc ? ...
alert: post
http_post_url: "http://localhost:8088/alertapi"
http_post_static_payload:
    rule_name: percentage_match_rule
    rule_level: medium
# POST 结果
# {
#     "num_hits": 10,
#     "@timestamp": "2018-01-31T09:39:05.199394Z",
#     "rule_level": "medium",
#     "rule_name": "percentage_match_rule",
#     "num_matches": 1,
#     "percentage": 100.0
# }

# --------------------------------------------------------- Other

new_term
# 字段的值与30天前的数据是否是新出现，如比较后是新值，则触发报警

metric_aggregation
# 计算窗口中的度量值高于或低于阈值时进行匹配

告警方式 ( 类型 )

# 支持的告警方式: https://elastalert.readthedocs.io/en/latest/ruletypes.html?highlight=Command#alerts
# 当前支持的告警类型:
#   Command
#   Email
#   JIRA
#   OpsGenie
#   SNS
#   HipChat
#   Slack
#   Telegram
#   GoogleChat
#   Debug
#   Stomp
#   theHive

# --------------------------------------------------------- command

# 使用 command 方式告警:  https://elastalert.readthedocs.io/en/latest/ruletypes.html?highlight=timeframe#command
# ......
alert:
  - command                                                     # 告警的动作类型
command: ["/bin/send_alert", "--username", "%(username)s"]      # 特定动作的具体告警方式
# ......

# --------------------------------------------------------- HTTP POST

# 使用 HTTP POST 方式告警: https://elastalert.readthedocs.io/en/latest/ruletypes.html#http-post
# ......
alert: post
http_post_url: "http://example.com/api"     # 要发布的网址
http_post_payload:                          # 键列表：用作POST内容的值
  ip: clientip                              # 将从Elasticsearch查询到的key映射到名为IP的JSON对象，若未定义则发送所有的key
http_post_static_payload:                   # 要发送的静态参数值对，以及Elasticsearch结果。在此处输入身份验证或其他信息
  apikey: abc123
http_post_headers:                          # 作为请求的一部分发送的标头值对
  authorization: Basic 123dr3234

# http_post_proxy:                          # 代理的网址（若需要）
# http_post_all_values:                     # 是否包含除 http_post_payload、http_post_static_payload 中键值对之外的所有键值对
#                                           # 若未指定 http_post_payload 则默认为 True
# ......

以 python 模块方式运行 elastalert

https://elastalert.readthedocs.io/en/latest/elastalert.html

# 将elastalert以python模块的方式安装
pip install elastalert

# 创建需要的相关索引和映射配置信息到Elasticsearch
# 关于其创建的索引的用途说明: https://elastalert.readthedocs.io/en/latest/elastalert_status.html
./elastalert-create-index --host <ES_IP> --port <ES_PORT> --username 账号 --password '密码' --config ./config.yaml

# 启动前先检查规则文件
./elastalert-test-rule ./123.yaml  --config ./config.yaml

# ------------------------------------------------------------------------------ demo-1

# 启动elastalert
./bin/elastalert --config ./config.yaml --verbose

# ------------------------------------------------------------------------------ demo-2

# 启动elastalert
# 这里的选项: --rule example_frequency.yaml 表示只运行 example_frequency.yaml 这一个rule文件
# 若不使用上述选项则会运行主配置文件中rules_folder字段指定的路径下所有的rule文件
python -m elastalert.elastalert --config ./config.yaml --verbose [ --rule example_frequency.yaml ]

# 为了让服务后台运行并且达到守护进程的效果，在生产环境建议使用supervisor管理

补充

# docker
# https://github.com/bitsensor/elastalert

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

# -------------------------------------------------------------- summary_table_fields

# 当使用 elastalert 所谓的"聚合"时，有时会在查看媒体（电子邮件、jira 票证等）中存在大量文档
# 如果设置了 summary_table_fields 字段，则将提供所有结果中指定字段的摘要:

# 例如希望汇总出现在文档中的 usernames 和 event_types，以便快速浏览最相关的字段:
summary_table_fields:
    - my_data.username
    - my_data.event_type

# Elastalert 将在警报介质中提供以下汇总表:
# +------------------+--------------------+
# | my_data.username | my_data.event_type |
# +------------------+--------------------+
# |      alice       |       login        |
# |       bob        |     something      |
# |      alice       |   something else   |
# +------------------+--------------------+

# -------------------------------------------------------------- SMTP 配置

smtp_host: smtp.qq.com
smtp_port: 25
smtp_auth_file: /root/elastalert/rule_templates/smtp_auth_file.yaml   # 帐号密码配置在此
from_addr: "xxxx@qq.com"
alert:
- "email"
email:
- "xxxx@qq.com"

# cat /root/elastalert/rule_templates/smtp_auth_file.yaml
# user: xxxx@qq.com
# password: xxxxxxxxxx

# -------------------------------------------------------------- 告警内容格式化

# 可以自定义告警内容，内部是使用Python的format来实现的
alert_subject: "Error {} @{}"
alert_subject_args:
  - name
  - "@timestamp"

alert_text_type: alert_text_only
alert_text: |
  ### Error frequency exceeds
  > Name: {}
  > Message: {}
  > Host: {} ({})
alert_text_args:
  - name
  - message
  - hostname
  - host

# --------------------------------------------------------------

alert_text: "           # 自定义需要报警发送的内容
域    名: {}\n
调用方式: {}\n
请求链接: {}\n
状 态 码: {}\n
后端服务器: {}\n
数     量: {}
"

alert_text_type: alert_text_only
alert_text_args:        # 对应alert_text的内容
- host
- method
- request
- status
- upstream
- num_hits

alert: 
- "elastalert_modules.dingtalk_alert.DingTalkAlerter"
dingtalk_webhook: "XXXXXX"
dingtalk_msgtype: "text"