一、ElastAlert介紹

在日志管理上我們使用Elasticsearch，Logstash和Kibana技術(shù)棧來管理不斷增長(zhǎng)的數(shù)據(jù)和日志，但是對(duì)于錯(cuò)誤日志的監(jiān)控ELK架構(gòu)并沒有提供，所以我們需要使用到第三方工具ElastAlert，來幫助我們及時(shí)發(fā)現(xiàn)業(yè)務(wù)中存在的問題。

ElastAlert通過定期查詢Elasticsearch，并將數(shù)據(jù)傳遞到規(guī)則類型，該規(guī)則類型確定何時(shí)找到匹配項(xiàng)。發(fā)生匹配時(shí)，將為該警報(bào)提供一個(gè)或多個(gè)警報(bào)，這些警報(bào)將根據(jù)匹配采取行動(dòng)。

這是由一組規(guī)則配置的，每個(gè)規(guī)則定義一個(gè)查詢，一個(gè)規(guī)則類型和一組警報(bào)。

在這里插入圖片描述

ElastAlert支持以下方式報(bào)警

• Command （可調(diào)用短信接口）
• Email
• JIRA
• OpsGenie
• SNS
• HipChat
• Slack
• Telegram
• Debug
• Stomp

除了這種基本用法外，還有許多其他功能使警報(bào)更加有用：

• 警報(bào)鏈接到Kibana儀表板
• 任意字段的合計(jì)計(jì)數(shù)
• 將警報(bào)合并為定期報(bào)告
• 通過使用唯一鍵字段來分隔警報(bào)
• 攔截并增強(qiáng)比賽數(shù)據(jù)

二、部署ElastAlert

1. 部署所需環(huán)境

ELK 環(huán)境部署

EFK6.3+kafka+logstash日志分析平臺(tái)集群

安裝依賴包

$ yum -y install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gdbm-devel db4-devel libpcap-devel xz-devel

部署python3.6

$ mkdir -p /usr/local/python3
$ cd /usr/local/python3
$ wget https://www.python.org/ftp/python/3.6.9/Python-3.6.9.tgz
$ tar xf Python-3.6.9.tgz
$ cd Python-3.6.9
$ ./configure --prefix=/usr/local/python3
$ make && make install

配置環(huán)境變量

# 將python2.7 軟鏈刪除，換成python3.6
$ rm /usr/bin/python
$ ln -s /usr/local/python3/bin/python3.6 /usr/bin/python
$ rm /usr/bin/pip
$ ln -s /usr/local/python3/bin/pip3 /usr/bin/pip

驗(yàn)證版本

$ python
Python 3.6.9 (default, Jun  2 2020, 12:12:43) 
[GCC 4.4.7 20120313 (Red Hat 4.4.7-18)] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> 
$ pip -V
pip 18.1 from /usr/local/python3/lib/python3.6/site-packages/pip (python 3.6)

2. 部署ElastAlert

$ cd /app
$ git clone https://github.com/Yelp/elastalert.git

安裝模塊：

$ pip install "setuptools>=11.3"
$ python setup.py install

根據(jù)Elasticsearch的版本，您可能需要手動(dòng)安裝正確版本的elasticsearch-py。

Elasticsearch 5.0+：

$ pip install "elasticsearch>=5.0.0"

Elasticsearch 2.X：

$ pip install "elasticsearch<3.0.0"

3. 配置ElastAlert

配置config.yaml 文件

$ cp config.yaml.example  config.yaml
$ cat config.yaml
rules_folder: example_rules
run_every:
  seconds: 10
buffer_time:
  minutes: 15
es_host: 10.1.144.208
es_port: 9201
#es_username: elastic
#es_password: 123456
writeback_index: elastalert_status
alert_time_limit:
  days: 2

rules_folder：ElastAlert從中加載規(guī)則配置文件的位置。它將嘗試加載文件夾中的每個(gè).yaml文件。沒有任何有效規(guī)則，ElastAlert將無法啟動(dòng)。
run_every： ElastAlert多久查詢一次Elasticsearch的時(shí)間。
buffer_time：查詢窗口的大小，從運(yùn)行每個(gè)查詢的時(shí)間開始向后延伸。對(duì)于其中use_count_query或use_terms_query設(shè)置為true的規(guī)則，將忽略此值。
es_host：是Elasticsearch群集的地址，ElastAlert將在其中存儲(chǔ)有關(guān)其狀態(tài)，查詢運(yùn)行，警報(bào)和錯(cuò)誤的數(shù)據(jù)。
es_port：es對(duì)應(yīng)的端口。
es_username： 可選的; 用于連接的basic-auth用戶名es_host。
es_password： 可選的; 用于連接的basic-auth密碼es_host。
es_send_get_body_as： 可選的; 方法查詢Elasticsearch - GET，POST或source。默認(rèn)是GET
writeback_index：ElastAlert將在其中存儲(chǔ)數(shù)據(jù)的索引的名稱。我們稍后將創(chuàng)建此索引。
alert_time_limit： 失敗警報(bào)的重試窗口。

創(chuàng)建elastalert-create-index索引

$ elastalert-create-index
New index name (Default elastalert_status)
Name of existing index to copy (Default None)
New index elastalert_status created
Done!

三、使用微信報(bào)警

由于ElastAlert沒有內(nèi)置企業(yè)微信的報(bào)警方式，我們還需要使用一個(gè)開源插件elastalert-wechat-plugin來實(shí)現(xiàn)微信的報(bào)警，Github項(xiàng)目地址

1. 下載項(xiàng)目文件

$ cd elastalert
$ wget -P ~/elastalert/elastalert_modules/ wget https://raw.githubusercontent.com/anjia0532/elastalert-wechat-plugin/master/elastalert_modules/wechat_qiye_alert.py
$ touch ~/elastalert/elastalert_modules/__init__.py

2. 修改插件源碼

由于這個(gè)插件是基于python2.x版本開發(fā)的，而ElastAlert的最新版本使用的是python3.6版本開發(fā)，所以需要改一些代碼，以便正常運(yùn)行，另外還添添加了轉(zhuǎn)中文字符功能。
wechat_qiye_alert.py修改后如下：

#! /usr/bin/env python
# -*- coding: utf-8 -*-

import json
import datetime
from elastalert.alerts import Alerter, BasicMatchString
from requests.exceptions import RequestException
from elastalert.util import elastalert_logger,EAException #[感謝minminmsn分享](https://github.com/anjia0532/elastalert-wechat-plugin/issues/2#issuecomment-311014492)
import requests
from elastalert_modules.MyEncoder import MyEncoder

'''
#################################################################
# 微信企業(yè)號(hào)推送消息                                              #
#                                                               #
# 作者: AnJia <anjia0532@gmail.com>                              #
# 作者博客: https://anjia.ml/                                    #
# Github: https://github.com/anjia0532/elastalert-wechat-plugin #
#                                                               #
#################################################################
'''
class WeChatAlerter(Alerter):

    #企業(yè)號(hào)id，secret，應(yīng)用id必填

    required_options = frozenset(['corp_id','secret','agent_id'])

    def __init__(self, *args):
        super(WeChatAlerter, self).__init__(*args)
        self.corp_id = self.rule.get('corp_id', '')     #企業(yè)號(hào)id
        self.secret = self.rule.get('secret', '')       #secret
        self.agent_id = self.rule.get('agent_id', '')   #應(yīng)用id

        self.party_id = self.rule.get('party_id')       #部門id
        self.user_id = self.rule.get('user_id', '')     #用戶id，多人用 | 分割，全部用 @all
        self.tag_id = self.rule.get('tag_id', '')       #標(biāo)簽id
        self.access_token = ''                          #微信身份令牌
        self.expires_in=datetime.datetime.now() - datetime.timedelta(seconds=60)

    def create_default_title(self, matches):
        subject = 'ElastAlert: %s' % (self.rule['name'])
        return subject

    def alert(self, matches):

        if not self.party_id and not self.user_id and not self.tag_id:
            elastalert_logger.warn("All touser & toparty & totag invalid")

        # 參考elastalert的寫法
        # https://github.com/Yelp/elastalert/blob/master/elastalert/alerts.py#L236-L243
        body = self.create_alert_body(matches)

        #matches 是json格式
        #self.create_alert_body(matches)是String格式,詳見 [create_alert_body 函數(shù)](https://github.com/Yelp/elastalert/blob/master/elastalert/alerts.py)

        # 微信企業(yè)號(hào)獲取Token文檔
        # http://qydev.weixin.qq.com/wiki/index.php?title=AccessToken
        self.get_token()

        self.senddata(body)

        elastalert_logger.info("send message to %s" % (self.corp_id))

    def get_token(self):

        #獲取token是有次數(shù)限制的,本想本地緩存過期時(shí)間和token，但是elastalert每次調(diào)用都是一次性的，不能全局緩存
        if self.expires_in >= datetime.datetime.now() and self.access_token:
            return self.access_token

        #構(gòu)建獲取token的url
        get_token_url = 'https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid=%s&corpsecret=%s' %(self.corp_id,self.secret)

        try:
            response = requests.get(get_token_url)
            response.raise_for_status()
        except RequestException as e:
            raise EAException("get access_token failed , stacktrace:%s" % e)
            #sys.exit("get access_token failed, system exit")

        token_json = response.json()

        if 'access_token' not in token_json :
            raise EAException("get access_token failed , , the response is :%s" % response.text())
            #sys.exit("get access_token failed, system exit")

        #獲取access_token和expires_in
        self.access_token = token_json['access_token']
        self.expires_in = datetime.datetime.now() + datetime.timedelta(seconds=token_json['expires_in'])

        return self.access_token

    def senddata(self, content):

        #如果需要原始json，需要傳入matches

        # http://qydev.weixin.qq.com/wiki/index.php?title=%E6%B6%88%E6%81%AF%E7%B1%BB%E5%9E%8B%E5%8F%8A%E6%95%B0%E6%8D%AE%E6%A0%BC%E5%BC%8F
        # 微信企業(yè)號(hào)有字符長(zhǎng)度限制（2048），超長(zhǎng)自動(dòng)截?cái)?
        # 參考 http://blog.csdn.net/handsomekang/article/details/9397025
        #len utf8 3字節(jié)，gbk2 字節(jié)，ascii 1字節(jié)
        if len(content) > 512 :
            content = content[:512] + "..."

        # 微信發(fā)送消息文檔
        # http://qydev.weixin.qq.com/wiki/index.php?title=%E6%B6%88%E6%81%AF%E7%B1%BB%E5%9E%8B%E5%8F%8A%E6%95%B0%E6%8D%AE%E6%A0%BC%E5%BC%8F
        send_url = 'https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=%s' %( self.access_token)

        headers = {'content-type': 'application/json'}
      
        # 替換消息標(biāo)題為中文,下面的字段為logstash切分的日志字段
        title_dict = { 
#            "At least": "報(bào)警規(guī)則:At least",
            "@timestamp": "報(bào)警時(shí)間",
            "_index": "索引名稱",
            "_type": "索引類型",
            "ServerIP": "報(bào)警主機(jī)",
            "hostname": "報(bào)警機(jī)器",
            "message": "報(bào)警內(nèi)容",
            "class": "報(bào)錯(cuò)類",
            "lineNum": "報(bào)錯(cuò)行"
           "num_hits": "文檔命中數(shù)",
           "num_matches": "文檔匹配數(shù)"
        }

        #print(f"type:{type(content)}")
        for k, v in title_dict.items():
            content = content.replace(k, v, 1 )

        # 最新微信企業(yè)號(hào)調(diào)整校驗(yàn)規(guī)則，tagid必須是string類型，如果是數(shù)字類型會(huì)報(bào)錯(cuò)，故而使用str()函數(shù)進(jìn)行轉(zhuǎn)換
        payload = {
            "touser": self.user_id and str(self.user_id) or '', #用戶賬戶，建議使用tag
            "toparty": self.party_id and str(self.party_id) or '', #部門id，建議使用tag
            "totag": self.tag_id and str(self.tag_id) or '', #tag可以很靈活的控制發(fā)送群體細(xì)粒度。比較理想的推送應(yīng)該是，在heartbeat或者其他elastic工具自定義字段，添加標(biāo)簽id。這邊根據(jù)自定義的標(biāo)簽id，進(jìn)行推送
            'msgtype': "text",
            "agentid": self.agent_id,
            "text":{
                "content": content.encode('UTF-8').decode("latin1") #避免中文字符發(fā)送失敗
               },
            "safe":"0"
        }

        # set https proxy, if it was provided
        # 如果需要設(shè)置代理，可修改此參數(shù)并傳入requests
        # proxies = {'https': self.pagerduty_proxy} if self.pagerduty_proxy else None
        try:
            #response = requests.post(send_url, data=json.dumps(payload, ensure_ascii=False), headers=headers)
            response = requests.post(send_url, data=json.dumps(payload, cls=MyEncoder, indent=4, ensure_ascii=False), headers=headers)
            response.raise_for_status()
        except RequestException as e:
            raise EAException("send message has error: %s" % e)

        elastalert_logger.info("send msg and response: %s" % response.text)


    def get_info(self):
        return {'type': 'WeChatAlerter'}

在同級(jí)目錄下創(chuàng)建MyEncoder.py文件

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
 
import json

class MyEncoder(json.JSONEncoder):
     def default(self, obj):
         if isinstance(obj, bytes):
            return str(obj, encoding='utf-8')
         return json.JSONEncoder.default(self, obj)

3. 申請(qǐng)企業(yè)微信賬號(hào)

step 1: 訪問網(wǎng)站 注冊(cè)企業(yè)微信賬號(hào)（不需要企業(yè)認(rèn)證）。
step 2: 訪問apps 創(chuàng)建第三方應(yīng)用，點(diǎn)擊創(chuàng)建應(yīng)用按鈕 -> 填寫應(yīng)用信息：
Step3: 創(chuàng)建部門，獲取部門ID

4. 配置報(bào)警規(guī)則

配置規(guī)則文件

$ cd  elastalert/example_rules/
$ cp example_frequency.yaml   applog.yaml
$ cat sms-applog.yaml  | grep -v ^#
name: 【日志報(bào)警】
use_strftine_index: true
type: frequency

index: applog-*  
num_events: 1
timeframe:
  minutes: 1
filter: 
- query:
    query_string:
      query: '"\[ERROR\]" NOT "發(fā)送郵件失敗"'

alert:
 - "elastalert_modules.wechat_qiye_alert.WeChatAlerter"
corp_id: wxdxxx40b4720f24
secret: xa4pWq63sxxtaZzzEg8X860ZBIoOkToCbh_oNc
agent_id: 1000002
party_id: 2

index：要查詢的索引的名稱, ES中存在的索引。
num_events：此參數(shù)特定于frequency類型，并且是觸發(fā)警報(bào)時(shí)的閾值。
filter：用于過濾結(jié)果的Elasticsearch過濾器列表，這里的規(guī)則定義是除了包含“發(fā)送郵件失敗”的錯(cuò)誤日志，其他所有ERROR的日志都會(huì)觸發(fā)報(bào)警。
alert：定義報(bào)警方式，我們這里采用企業(yè)微信報(bào)警。
corp_id： 企業(yè)微信的接口認(rèn)證信息

5. 運(yùn)行ElastAlert

$ python -m elastalert.elastalert --verbose --config /app/elastalert/config.yaml --rule /app/elastalert/example_rules/sms-applog.yaml 
1 rules loaded
INFO:elastalert:Starting up
INFO:elastalert:Disabled rules are: []
INFO:elastalert:Sleeping for 9.999904 seconds
INFO:elastalert:Queried rule 【日志報(bào)警】 from 2020-06-05 17:47 CST to 2020-06-05 17:47 CST: 0 / 0 hits
INFO:elastalert:Ran 【日志報(bào)警】 from 2020-06-05 17:47 CST to 2020-06-05 17:47 CST: 0 query hits (0 already seen), 0 matches, 0 alerts sent
后臺(tái)運(yùn)行
$ nohup python -m elastalert.elastalert --verbose --config /app/elastalert/config.yaml --rule /app/elastalert/example_rules/sms-applog.yaml > nohup.txt 2>&1 

微信端添加企業(yè)號(hào)報(bào)警格式如下：

image.png

關(guān)注公眾號(hào)獲取視頻教程及更多資料：

image.png