文章目录

步骤
1
:创建机器人
步骤
2
:创建目录和虚拟环境
步骤
3
:编写脚本
步骤
4
:书写配置文件
步骤
5
:启动服务
步骤
6
:测试

Python+钉钉搭建服务器告警系统：自动监控异常

晕头小狗

原创

发布时间: 2026-08-16 17:47:39 | 阅读数

运维/部署

钉钉

服务器运维的最大挑战并非故障本身，而是缺乏实时告警机制。无论是磁盘耗尽、服务异常还是日志报错，若无法在第一时间感知，往往会导致发现滞后，致使业务遭受实质性损害。

创建机器人

12345

进入钉钉群设置，点击“机器人” > “添加机器人” > “自定义机器人”。
填写机器人名称，并设置自定义关键词。
安全设置勾选“加签”，复制并妥善保存密钥（Secret）。
完成创建后，复制并保存 Webhook 地址。

创建目录和虚拟环境

1234

// 创建log-ding项目并进入

mkdir -p /usr/local/src/log-ding && cd /usr/local/src/log-ding

// 创建虚拟环境

python3 -m venv venv

// 激活环境后开头会出现（venv）

source venv/bin/activate

// 下载依赖

pip install requests python-dotenv

编写脚本

123

vim log_monitor.py新建文件，复制下边的内容并粘贴。然后esc退出编辑状态，:wq保存并退出。

# -*- coding: utf-8 -*-

import os, sys, time, hmac, hashlib, base64, urllib.parse, logging

from datetime import datetime

import requests

from dotenv import load_dotenv

# 加载环境变量

load_dotenv()

# 基础配置

WEBHOOK = os.getenv("DINGTALK_WEBHOOK")

SECRET = os.getenv("DINGTALK_SECRET")

LOG_FILE = os.getenv("LOG_FILE", "/var/log/ceshi/access.log")

KEYWORDS = [k.strip() for k in os.getenv("KEYWORDS", "error,failed").split(",") if k.strip()]

COOLDOWN = int(os.getenv("ALERT_COOLDOWN", 60))

# 日志配置

logging.basicConfig(

level=logging.INFO,

format="%(asctime)s [%(levelname)s] %(message)s",

stream=sys.stdout

)

logger = logging.getLogger(__name__)

def gen_sign():

"""生成钉钉签名"""

timestamp = str(round(time.time() * 1000))

string_to_sign = f"{timestamp}\n{SECRET}"

hmac_code = hmac.new(SECRET.encode(), string_to_sign.encode(), digestmod=hashlib.sha256).digest()

sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

return timestamp, sign

def send_dingtalk(content):

"""发送钉钉告警"""

if not WEBHOOK or not SECRET:

logger.error("缺少 DINGTALK_WEBHOOK 或 DINGTALK_SECRET")

return False

ts, sign = gen_sign()

url = f"{WEBHOOK}&timestamp={ts}&sign={sign}"

data = {"msgtype": "text", "text": {"content": content}}

try:

resp = requests.post(url, json=data, timeout=5)

result = resp.json()

if result.get("errcode") == 0:

logger.info("钉钉告警发送成功")

return True

else:

logger.error(f"钉钉返回错误: {result}")

return False

except Exception as e:

logger.error(f"钉钉请求异常: {e}")

return False

def match_keyword(line):

"""检查行是否包含关键词（不区分大小写）"""

lower_line = line.lower()

for kw in KEYWORDS:

if kw.lower() in lower_line:

return kw

return None

def main():

logger.info(f"监控启动 | 日志: {LOG_FILE} | 关键词: {KEYWORDS} | 冷却: {COOLDOWN}s")

last_alert_time = {}

while True:

try:

if not os.path.exists(LOG_FILE):

logger.warning(f"日志文件不存在: {LOG_FILE}，等待生成...")

time.sleep(5)

continue

with open(LOG_FILE, "r", encoding="utf-8", errors="ignore") as f:

# 首次启动跳到文件末尾，只监控新内容

f.seek(0, 2)

logger.info(f"开始监控: {LOG_FILE} (从文件末尾开始)")

while True:

line = f.readline()

if not line:

time.sleep(1)

# 检测文件是否被轮转重建

try:

if os.stat(LOG_FILE).st_ino != os.fstat(f.fileno()).st_ino:

logger.info("检测到日志文件轮转，重新打开...")

break

except OSError:

break

continue

keyword = match_keyword(line)

if keyword:

now = time.time()

last = last_alert_time.get(keyword, 0)

if now - last >= COOLDOWN:

alert_msg = (

f"⚠️ 日志告警\n"

f"主机: {os.uname().nodename}\n"

f"关键词: {keyword}\n"

f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"

f"内容: {line.strip()[:500]}"

)

send_dingtalk(alert_msg)

last_alert_time[keyword] = now

else:

remaining = COOLDOWN - int(now - last)

logger.debug(f"关键词 [{keyword}] 冷却中，剩余 {remaining}s")

except KeyboardInterrupt:

logger.info("收到退出信号，监控停止")

sys.exit(0)

except Exception as e:

logger.error(f"监控异常: {e}", exc_info=True)

time.sleep(5)

if __name__ == "__main__":

main()

书写配置文件

cat > /usr/local/src/log-ding/.env << 'EOF'

DINGTALK_WEBHOOK=https://oapi.dingtalk.com/robot/send?access_token=你创建机器人的token

DINGTALK_SECRET=你的秘钥

LOG_FILE=/var/log/ceshi/access.log

KEYWORDS=error,failed,exception,no space left,oom,panic

ALERT_COOLDOWN=60

EOF

启动服务

// 启动服务

python log_monitor.py

测试

// 创建目录

mkdir -p /var/log/ceshi

// 写入内容

echo "$(date '+%Y-%m-%d %H:%M:%S') [TEST] error: 从零部署连通性测试" >> /var/log/ceshi/access.log

qq空间

微博

复制链接

本文未经作者允许授权，禁止转载

更多相关项目