Prometheus使用釘釘Webhook推送告警通知

構建釘釘Webhook鏡像

代碼依賴文件：requirements.txt

certifi==2018.10.15
chardet==3.0.4
Click==7.0
Flask==1.0.2
idna==2.7
itsdangerous==1.1.0
Jinja2==2.10
MarkupSafe==1.1.0
requests==2.20.1
urllib3==1.24.1
Werkzeug==0.15.3

釘釘Webhook代碼示例文件：app.py

import os
import json
import logging
import requests
import time
import hmac
import hashlib
import base64
import urllib.parse
from urllib.parse import urlparse

from flask import Flask
from flask import request

app = Flask(__name__)

logging.basicConfig(
    level=logging.DEBUG if os.getenv('LOG_LEVEL') == 'debug' else logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s')


@app.route('/', methods=['POST', 'GET'])
def send():
    if request.method == 'POST':
        post_data = request.get_data()
        app.logger.debug(post_data)
        send_alert(json.loads(post_data))
        return 'Success!'
    else:
        return 'Weclome to use dingtalk webhook server!'


def send_alert(data):
    token = os.getenv('ROBOT_TOKEN')
    secret = os.getenv('ROBOT_SECRET')
    if not token:
        app.logger.error('You must set ROBOT_TOKEN env!')
        return
    if not secret:
        app.logger.error('You must set ROBOT_SECRET env!')
        return
    timestamp = int(round(time.time() * 1000))
    url = 'https://oapi.dingtalk.com/robot/send?access_token=%s&timestamp=%d&sign=%s' % (token, timestamp, make_sign(timestamp, secret))

    alerts = data['alerts']
    alert_name = alerts[0]['labels']['alertname']

    def _mark_item(alert):
        labels = alert['labels']
        annotations = "> "
        for k, v in alert['annotations'].items():
            annotations += "{0}: {1}\n".format(k, v)
        if 'job' in labels:
            mark_item = "\n> job: " + labels['job'] + '\n\n' + annotations + '\n'
        else:
            mark_item = "\n> " + annotations + '\n'
        return mark_item

    title = '[雲監控] %s 有 %d 條新的報警' % (alert_name, len(alerts))

    external_url = alerts[0]['generatorURL']
    prometheus_url = os.getenv('PROME_URL')
    if prometheus_url:
        res = urlparse(external_url)
        external_url = external_url.replace(res.netloc, prometheus_url)

    send_data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": title + "\n" + _mark_item(alerts[0]) + "\n" + "[>>點擊查看完整信息](" + external_url + ")\n"
        }
    }

    req = requests.post(url, json=send_data)
    result = req.json()
    if result['errcode'] != 0:
        app.logger.error('notify dingtalk error: %s' % result['errcode'])


def make_sign(timestamp, secret):
    """新版釘釘更新了安全策略，這里我們采用簽名的方式進行安全認證。釘釘開發文檔地址如下：
    https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq"""
    secret_enc = bytes(secret, 'utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = bytes(string_to_sign, 'utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return sign


if __name__ == '__main__':
    app.debug=False
    app.run(host='0.0.0.0', port=5000)

腳本詳細解釋：

import os
import json
import logging
import requests
import time
import hmac
import hashlib
import base64
import urllib.parse
from urllib.parse import urlparse

# 導入Flask模塊
from flask import Flask
from flask import request

"""第一部分，初始化：
所有的Flask都必須創建程序實例，web服務器使用wsgi協議，把客戶端所有的請求都轉發給這個程序實例。
程序實例是Flask的對象，一般情況下用如下方法實例化。
Flask類只有一個必須指定的參數，即程序主模塊或者包的名字，__name__是系統變量，該變量指的是本py文件的文件名。"""
app = Flask(__name__)

"""日志配置部分：
這里判斷系統環境變量LOG_LEVEL的值，如果為debug，則日志輸出級別為DEBUG（即輸出所有級別日志），其他值為INFO級別。
通過自定義日志格式，使日志輸出更美觀。"""
logging.basicConfig(
    level=logging.DEBUG if os.getenv('LOG_LEVEL') == 'debug' else logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s')


"""第二部分，路由和視圖函數：
客戶端發送url給web服務器，web服務器將url轉發給flask程序實例，
程序實例需要知道對於每一個url請求啟動哪一部分代碼，所以保存了一個url和python函數的映射關系。
處理url和函數之間關系的程序，稱為路由。
在flask中，定義路由最簡便的方式，是使用程序實例的app.route裝飾器，把裝飾的函數注冊為路由。
這里請求的url地址為"/"，允許的請求方式為POST和GET，類型為可迭代對象，請求方式多達八種。"""
@app.route('/', methods=['POST', 'GET'])
def send():
    if request.method == 'POST':
        # 如果請求方式為POST，則獲取未經處理過的原始數據而不管內容類型。如果數據格式是json的，則取得的是json字符串，排序和請求參數一致。
        post_data = request.get_data()
        # 將獲取的原始數據輸出到DEBUG級別日志。
        app.logger.debug(post_data)
        # 將獲取的原始數據類型由str轉換為dict，並以此為參數，調用send_alert函數發送告警信息。
        send_alert(json.loads(post_data))
        return 'Success!'
    else:
        return 'Weclome to use dingtalk webhook server!'


"""發送告警信息函數。"""
def send_alert(data):
    # 獲取系統環境變量的值。
    token = os.getenv('ROBOT_TOKEN')
    secret = os.getenv('ROBOT_SECRET')
    if not token:
        # 如果值為空，則輸出ERROR級別日志。
        app.logger.error('You must set ROBOT_TOKEN env!')
        return
    if not secret:
        app.logger.error('You must set ROBOT_SECRET env!')
        return
    # 默認情況下python的時間戳是以秒為單位輸出的float，通過把秒轉換毫秒的方法獲得13位的時間戳，round()是四舍五入。
    timestamp = int(round(time.time() * 1000))
    # 把token、timestamp和簽名值拼接到URL中。簽名值由make_sign函數計算得到。
    url = 'https://oapi.dingtalk.com/robot/send?access_token=%s&timestamp=%d&sign=%s' % (token, timestamp, make_sign(timestamp, secret))

    # 獲取告警列表。
    alerts = data['alerts']
    # 從第一條記錄獲取告警名稱。
    alert_name = alerts[0]['labels']['alertname']

    """獲取告警摘要信息函數。"""
    def _mark_item(alert):
        # 獲取告警記錄的label列表。
        labels = alert['labels']
        # 初始化變量的值。
        annotations = "> "
        # Python字典items()方法以列表返回可遍歷的(鍵, 值)元組數組。
        for k, v in alert['annotations'].items():
            # 格式化並拼接獲取到的鍵值對。
            annotations += "{0}: {1}\n".format(k, v)
        if 'job' in labels:
            # 如果存在job的label，則將其拼接到摘要信息。
            mark_item = "\n> job: " + labels['job'] + '\n\n' + annotations + '\n'
        else:
            mark_item = "\n> " + annotations + '\n'
        return mark_item

    # 拼接告警標題，len函數獲取告警總數。
    title = '[雲監控] %s 有 %d 條新的告警' % (alert_name, len(alerts))

    # 獲取告警記錄的generatorURL。
    external_url = alerts[0]['generatorURL']
    # 獲取外部訪問prometheus的URL地址。
    prometheus_url = os.getenv('PROME_URL')
    if prometheus_url:
        # urlparse解析URL，返回元組 (scheme, netloc, path, parameters, query, fragment)。
        res = urlparse(external_url)
        # 替換netloc部分的值，即主機地址。
        external_url = external_url.replace(res.netloc, prometheus_url)
    
    """定義要發送的消息，數據格式為markdown類型。"""
    send_data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": title + "\n" + _mark_item(alerts[0]) + "\n" + "[>>點擊查看完整信息](" + external_url + ")\n"
        }
    }

    # 使用requests的post方法發送消息數據，json參數會自動將字典類型的對象轉換為json格式。
    req = requests.post(url, json=send_data)
    # 將發送出的消息轉換為json格式。
    result = req.json()
    if result['errcode'] != 0:
        # 如果errcode的值為非0，則輸出相關ERROR級別日志。
        app.logger.error('notify dingtalk error: %s' % result['errcode'])


"""簽名計算函數。"""
def make_sign(timestamp, secret):
    """新版釘釘更新了安全策略，這里采用簽名的方式進行安全認證。釘釘開發文檔地址如下：
    https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq"""
    # 將secret進行utf-8編碼。
    secret_enc = bytes(secret, 'utf-8')
    # 格式化timestamp和secret。
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    # 對格式化后的string進行utf-8編碼。
    string_to_sign_enc = bytes(string_to_sign, 'utf-8')
    # 采用SHA256進行哈希計算，digest()返回摘要，作為二進制數據字符串值。
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    # urllib.parse.quote_plus()編碼了斜線，b64encode()對hmac_code進行Base64編碼。
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return sign


"""第三部分：程序實例用run方法啟動flask集成的【開發web服務器】。
__name__ == '__main__'是python常用的方法，表示只有直接啟動本腳本時，才用app.run方法。
如果是其他腳本調用本腳本，程序假定父級腳本會啟用不同的服務器，因此不用執行app.run()。
服務器啟動后，會啟動輪詢，等待並處理請求。輪詢會一直請求，直到程序停止。"""
if __name__ == '__main__':
    app.debug=False
    app.run(host='0.0.0.0', port=5000)


"""如上述代碼所示，app是flask的實例，功能就是接受來自web服務器的請求，整個流程如下：
1. 瀏覽器將請求給web服務器，web服務器將請求給app;
2. app收到請求，通過路由找到對應的視圖函數，然后將請求處理，得到一個響應response;
3. 然后app將響應返回給web服務器;
4. web服務器返回給瀏覽器;
5. 瀏覽器展示給用戶觀看。"""

鏡像構建模板文件：Dockerfile

FROM python:3.6.4-alpine3.4

MAINTAINER varden

RUN echo "https://mirrors.aliyun.com/alpine/v3.4/main/" > /etc/apk/repositories
RUN echo "https://mirrors.aliyun.com/alpine/v3.4/community/" >> /etc/apk/repositories
RUN apk update
RUN apk upgrade

RUN apk add --no-cache ca-certificates tzdata curl bash && rm -rf /var/cache/apk/*
RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
RUN echo "Asia/Shanghai" > /etc/timezone

WORKDIR /app

COPY requirements.txt /app/requirements.txt
RUN pip install -r requirements.txt -i http://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com

COPY app.py /app/app.py

CMD python app.py

構建命令

docker build -t dingtalk-hook:1.0.1 .

docker-compose部署

version: "3.3"

networks:
  bridge_net:
    external:
      name: bridge

services:
  dingtalk_hook:
    image: dingtalk-hook:1.0.1
    ports:
      - "15000:5000"
    networks:
      - bridge_net
    environment:
      LOG_LEVEL: debug
      PROME_URL: 10.10.10.200:32101
    env_file: robot.env
    deploy:
      mode: replicated
      replicas: 1
      resources:
        limits:
          cpus: '0.2'
          memory: 128M
        reservations:
          cpus: '0.1'
          memory: 64M
    healthcheck:
      test: curl -f http://localhost:5000 || exit 1
      interval: 30s
      timeout: 30s
      retries: 5

環境變量文件：robot.env

ROBOT_TOKEN=248fc95536c33cbba9c6d2418d651766a7e8060078a0cffxxxxxxxxxxxxxxxxx
ROBOT_SECRET=SECe76d2fedf79602b265dc103494e1d8e87e7999cbe73xxxxxxxxxxxxxxxxxxxxxxx

K8s部署

創建Secret資源對象

kubectl create secret generic dingtalk-secret --from-literal=token=248fc95536c33cbba9c6d2418d651766a7e8060078a0cff833409xxxxxxxxxxxx --from-literal=secret=SECe76d2fedf79602b265dc103494e1d8e87e7999cbe73badbaa8c3xxxxxxxxxxxx -n monitoring

squid代理訪問外網：

kubectl create secret generic proxy-secret \
--from-literal=http_proxy=http://<username>:<password>@10.10.10.15:3128 \
--from-literal=https_proxy=http://<username>:<password>@10.10.10.15:3128 \
-n monitoring

K8s部署清單

apiVersion: apps/v1
kind: Deployment
metadata:
  name: dingtalk-hook
  namespace: monitoring
spec:
  replicas: 1
  selector:
    matchLabels:
      app: dingtalk-hook
  template:
    metadata:
      labels:
        app: dingtalk-hook
    spec:
      containers:
      - name: dingtalk-hook
        image: dingtalk-hook:1.0.1
        imagePullPolicy: IfNotPresent
        ports:
        - containerPort: 5000
          name: http
        env:
        - name: PROME_URL
          value: 10.10.10.200:32101
        - name: LOG_LEVEL
          value: debug
        - name: ROBOT_TOKEN
          valueFrom:
            secretKeyRef:
              name: dingtalk-secret
              key: token
        - name: ROBOT_SECRET
          valueFrom:
            secretKeyRef:
              name: dingtalk-secret
              key: secret
        - name: http_proxy
          valueFrom:
            secretKeyRef:
              name: proxy-secret
              key: http_proxy
        - name: https_proxy
          valueFrom:
            secretKeyRef:
              name: proxy-secret
              key: https_proxy
        resources:
          requests:
            cpu: 100m
            memory: 200Mi
          limits:
            cpu: 100m
            memory: 200Mi
        livenessProbe:
          httpGet:
            scheme: HTTP
            path: /
            port: 5000
          initialDelaySeconds: 30
          timeoutSeconds: 30
        readinessProbe:
          httpGet:
            scheme: HTTP
            path: /
            port: 5000
          initialDelaySeconds: 30
          timeoutSeconds: 30
---
apiVersion: v1
kind: Service
metadata:
  name: dingtalk-hook
  namespace: monitoring
spec:
  selector:
    app: dingtalk-hook
  ports:
  - name: hook
    port: 5000
    targetPort: http

在AlertManager中webhook地址直接通過DNS形式訪問即可：

receivers:
- name: 'webhook'
  webhook_configs:
  - url: 'http://dingtalk-hook:5000'
    send_resolved: true