如何配置gitlab runner实现每次runner运行后通过钉钉机器人webhook提示更新

🛠工具与编程 💻编程
, ,
1

提示

如果你的gitlab runner无法拉取docker跑不起来，可以参考 gitlab runner如何自部署，以及如何解决拉取docker镜像时失败

背景

希望每次运行完gitlab runner后，通过webhook，让钉钉机器人提示本次runner做了什么，用的是哪个版本的commit，效果如图

image
image1036×438 79.8 KB

配置好钉钉机器人

在群聊设置里找到机器人

image
image1012×1622 110 KB

新增一个自定义机器人

image
image1424×1266 148 KB

添加好以后点开机器人头衔-机器人设置

image
image1144×912 128 KB

勾上安全设置-加签，之后我们需要webhook、和加签secrete，后面要放在gitlab CI/CD的variable里

image
image1424×1266 100 KB

写好调用钉钉机器人webhook的代码

参考钉钉官方的python代码

# !/usr/bin/env python

import argparse
import logging
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests

def setup_logger():
    logger = logging.getLogger()
    handler = logging.StreamHandler()
    handler.setFormatter(
        logging.Formatter('%(asctime)s %(name)-8s %(levelname)-8s %(message)s [%(filename)s:%(lineno)d]'))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger


def define_options():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--access_token', dest='access_token', required=True,
        help='机器人webhook的access_token from https://open.dingtalk.com/document/orgapp/obtain-the-webhook-address-of-a-custom-robot '
    )
    parser.add_argument(
        '--secret', dest='secret', required=True,
        help='secret from https://open.dingtalk.com/document/orgapp/customize-robot-security-settings#title-7fs-kgs-36x'
    )
    parser.add_argument(
        '--userid', dest='userid',
        help='待 @ 的钉钉用户ID，多个用逗号分隔 from https://open.dingtalk.com/document/orgapp/basic-concepts-beta#title-o8w-yj2-t8x '
    )
    parser.add_argument(
        '--at_mobiles', dest='at_mobiles',
        help='待 @ 的手机号，多个用逗号分隔'
    )
    parser.add_argument(
        '--is_at_all', dest='is_at_all', action='store_true',
        help='是否@所有人，指定则为True，不指定为False'
    )
    parser.add_argument(
        '--msg', dest='msg', default='钉钉，让进步发生',
        help='要发送的消息内容'
    )
    return parser.parse_args()


def send_custom_robot_group_message(access_token, secret, msg, at_user_ids=None, at_mobiles=None, is_at_all=False):
    """
    发送钉钉自定义机器人群消息
    :param access_token: 机器人webhook的access_token
    :param secret: 机器人安全设置的加签secret
    :param msg: 消息内容
    :param at_user_ids: @的用户ID列表
    :param at_mobiles: @的手机号列表
    :param is_at_all: 是否@所有人
    :return: 钉钉API响应
    """
    timestamp = str(round(time.time() * 1000))
    string_to_sign = f'{timestamp}\n{secret}'
    hmac_code = hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

    url = f'https://oapi.dingtalk.com/robot/send?access_token={access_token}&timestamp={timestamp}&sign={sign}'

    body = {
        "at": {
            "isAtAll": str(is_at_all).lower(),
            "atUserIds": at_user_ids or [],
            "atMobiles": at_mobiles or []
        },
        "text": {
            "content": msg
        },
        "msgtype": "text"
    }
    headers = {'Content-Type': 'application/json'}
    resp = requests.post(url, json=body, headers=headers)
    logging.info("钉钉自定义机器人群消息响应：%s", resp.text)
    return resp.json()


def main():
    options = define_options()
    # 处理 @用户ID
    at_user_ids = []
    if options.userid:
        at_user_ids = [u.strip() for u in options.userid.split(',') if u.strip()]
    # 处理 @手机号
    at_mobiles = []
    if options.at_mobiles:
        at_mobiles = [m.strip() for m in options.at_mobiles.split(',') if m.strip()]
    send_custom_robot_group_message(
        options.access_token,
        options.secret,
        options.msg,
        at_user_ids=at_user_ids,
        at_mobiles=at_mobiles,
        is_at_all=options.is_at_all
    )


if __name__ == '__main__':
    main()

写一个python实现我们的需求

import os
import sys
import json
import time
import hmac
import hashlib
import base64
import urllib.parse
import urllib.request


def main():
    webhook = os.environ.get("DINGTALK_WEBHOOK", "")
    if not webhook:
        print("DINGTALK_WEBHOOK not set, skipping notification.")
        return 0

    secret = os.environ.get("DINGTALK_SECRET", "")
    branch = os.environ.get("CI_COMMIT_REF_NAME", "")
    commit_sha = os.environ.get("CI_COMMIT_SHORT_SHA", "")
    commit_msg = os.environ.get("CI_COMMIT_MESSAGE", "").strip()


    project_url = '你的gitlab地址'
    commit_full_sha = os.environ.get("CI_COMMIT_SHA", "")
    commit_link = f"{project_url}/-/commit/{commit_full_sha}"
    content = f"已经更新了，本次更新使用的分支: {branch}\n commit SHA: {commit_sha}\n commit message: {commit_msg}\n commit link: {commit_link}"

    payload = {"msgtype": "text", "text": {"content": content}}
    data = json.dumps(payload).encode("utf-8")

    url = webhook
    if secret:
        ts = str(int(time.time() * 1000))
        string_to_sign = f"{ts}\n{secret}"
        h = hmac.new(secret.encode("utf-8"), string_to_sign.encode("utf-8"), digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(h).decode("utf-8"))
        url = f"{webhook}&timestamp={ts}&sign={sign}"

    try:
        req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=10) as f:
            body = f.read().decode("utf-8")
            print("DingTalk notify success:", body)
            return 0
    except Exception as e:
        print("DingTalk notify failed:", e, file=sys.stderr)
        return 2


if __name__ == "__main__":
    sys.exit(main())

在gitlab runner里调用这个python代码

stages:
  - notify  


notify-success:
  stage: notify
  image: python:3.11-alpine
  only:
    - dev
  script:
    - |
      # 通过钉钉自定义机器人发送部署成功通知（使用项目内脚本）
      python3 scripts/ci_dingtalk_notify.py

配置gitlab CI/CD variable

把前面设置机器人时获取的webhook、和加签secrete放进去

image
image3318×1756 449 KB

结束

run！