elastalert的dingding告警

dingtalk.py

#!/usr/bin/env python
# encoding:utf8
"""
https://open-doc.dingtalk.com/docs/doc.htm?spm=a219a.7629140.0.0.karFPe&treeId=257&articleId=105735&docType=1
"""
from __future__ import unicode_literals

import copy
import json
import requests

from requests.exceptions import RequestException
from elastalert.alerts import Alerter, DateTimeEncoder, EAException,\
    elastalert_logger, lookup_es_key


class DingTalkAlerter(Alerter):
    """dingtalk alerter"""
    required_options = frozenset(
        ['dt_access_token'])

    def __init__(self, rule):
        super(DingTalkAlerter, self).__init__(rule)
        dt_webhook_url = self.rule.get(
            'dt_webhook_url', 'https://oapi.dingtalk.com/robot/send')
        dt_access_token = self.rule.get('dt_access_token')
        self.url = '{}?access_token={}'.format(dt_webhook_url, dt_access_token)
        self.dt_proxy = self.rule.get('dt_proxy')
        self.dt_msgtype = self.rule.get('dt_msgtype', 'text')
        dt_at_mobiles = self.rule.get('dt_at_mobiles')
        if dt_at_mobiles and isinstance(dt_at_mobiles, basestring):
            dt_at_mobiles = [dt_at_mobiles]
        self.dt_at_mobiles = dt_at_mobiles
        self.dt_is_at_all = self.rule.get('dt_is_at_all', False)

        self.dt_post_payload = self.rule.get('dt_post_payload', {})
        self.dt_post_static_payload = self.rule.get(
            'dt_post_static_payload', {})
        self.dt_post_all_values = self.rule.get(
            'dt_post_all_values', not self.dt_post_payload)
        self.dt_msg_format = self.rule.get('dt_msg_format')

    def alert(self, matches):
        headers = {
            'content-type': 'application/json',
            'Accept': 'application/json;charset=utf-8',
        }
        base_data = {
            'msgtype': self.dt_msgtype,
            'at': {
                'atMobiles': self.dt_at_mobiles,
                'isAtAll': self.dt_is_at_all,
            }
        }
        proxies = {'https': self.dt_proxy}if self.dt_proxy else None
        title = self.create_title(matches)
        for match in matches:
            payload = match if self.dt_post_all_values else {}
            payload.update(self.dt_post_static_payload)
            for post_key, es_key in self.dt_post_payload.items():
                payload[post_key] = lookup_es_key(match, es_key)
            if self.dt_msgtype == 'text':
                payload['title'] = title
            if self.dt_msg_format:
                payload = u'{}'.format(self.dt_msg_format).format(**payload)

            if self.dt_msgtype == 'text':
                content = {'content': payload}
            elif self.dt_msgtype == 'markdown':
                content = {
                    'title': title,
                    'text': payload
                }
            data = copy.copy(base_data)
            data[self.dt_msgtype] = content
            try:
                response = requests.post(
                    self.url,
                    data=json.dumps(data, cls=DateTimeEncoder),
                    headers=headers,
                    proxies=proxies,
                )
                response.raise_for_status()
            except RequestException as e:
                raise EAException("Error posting Dingtalk alert: %s" % e)
            elastalert_logger.info("Dingtalk alert sent.")

    def get_info(self):
        return {'type': "dingtalk",
                'dt_url': self.url,
                'dt_msgtype': self.dt_msgtype}

alert: "dingtalk.DingTalkAlerter"
dt_webhook_url: "https://oapi.dingtalk.com/robot/send"
dt_access_token: "token"
dt_msgtype: markdown
#dt_at_mobiles:
#  - 15999762582
dt_post_payload:
  hostname: beat.hostname
  ipaddr: fields.ipaddr
  level: level
  num_hits: num_hits
  num_matches: num_matches
  timestamp: timestamp
  message: message
dt_msg_format: "## 告警信息\n\n>级别:{level}\n\n> 主机：{hostname}-{ipaddr}\n\n> 数量：{num_matches}\n\n> 详细：{message}"