从零构建skill自动化巡检系统：新手避坑指南与最佳实践

29次阅读

共计 3493 个字符，预计需要花费 9 分钟才能阅读完成。

在运维和开发工作中，手动巡检一直是让人头疼的问题。我曾经负责过一个中型项目的运维工作，每天要手动检查几十个服务的状态，不仅效率低下，还经常因为疲劳而漏掉关键问题。更糟糕的是，半夜被叫起来处理线上问题，结果发现其实只是服务短暂波动，这种误报让人抓狂。

常见的自动化巡检系统也有自己的问题：

很多现成工具配置复杂，学习曲线陡峭
误报率居高不下，导致告警疲劳
扩展性差，难以适应业务快速变化
缺乏统一的可视化，问题难定位

在评估了多种技术栈后，我最终选择了 Python + Prometheus + Grafana 的组合，原因如下：

Python 生态优势 ：
丰富的库支持，开发效率高
学习曲线平缓，团队上手快
适合快速原型开发
Prometheus 的优势 ：
专为监控设计的时间序列数据库
强大的查询语言 PromQL
原生支持服务发现
Grafana 的优势 ：
出色的可视化能力
丰富的仪表板模板
灵活的告警配置

相比 Go 语言方案，Python 更适合我们这样需要快速迭代的业务场景。虽然 Go 在性能上更优，但对于巡检系统来说，Python 的性能已经足够，而开发效率的优势更为重要。

让我们从一个基础的 HTTP 服务巡检脚本开始：

#!/usr/bin/env python3
import requests
from prometheus_client import start_http_server, Gauge
import time

# 定义 Prometheus 指标
SERVICE_STATUS = Gauge('service_status', 'HTTP 服务状态', ['service_name'])
RESPONSE_TIME = Gauge('response_time', '服务响应时间 (ms)', ['service_name'])

def check_http_service(url, service_name):
    """
    检查 HTTP 服务状态
    :param url: 服务 URL
    :param service_name: 服务名称 (用于打标签)
    :return: (status_code, response_time_ms)
    """
    try:
        start_time = time.time()
        response = requests.get(url, timeout=5)
        elapsed = (time.time() - start_time) * 1000  # 转换为毫秒

        # 设置指标值
        SERVICE_STATUS.labels(service_name=service_name).set(1 if response.ok else 0)
        RESPONSE_TIME.labels(service_name=service_name).set(elapsed)

        return response.status_code, elapsed
    except Exception as e:
        SERVICE_STATUS.labels(service_name=service_name).set(0)
        return None, None

if __name__ == '__main__':
    # 启动 Prometheus 指标服务器 (默认端口 8000)
    start_http_server(8000)

    # 模拟持续巡检
    while True:
        check_http_service('https://example.com/api/health', 'example_api')
        time.sleep(60)  # 每分钟检查一次

这个脚本实现了：

基本的 HTTP 服务可用性检查
响应时间监控
异常捕获和处理
Prometheus 指标暴露

接下来配置 Prometheus 采集这些指标。创建 prometheus.yml:

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: 'service_monitor'
    static_configs:
      - targets: ['localhost:8000']  # Python 脚本暴露的端口

在 Grafana 中，我们可以创建一个简单的看板：

添加 Prometheus 数据源
新建 Dashboard
添加 Panel，使用 PromQL 查询：
服务状态：service_status{service_name="example_api"}
响应时间：response_time{service_name="example_api"}
设置合适的可视化类型 (状态用 Singlestat，响应时间用 Graph)

网络不稳定是误报的主要原因之一。我们可以采用以下策略：

重试机制 ：短暂失败后自动重试
滑动窗口检测 ：基于一段时间内的状态判断
异常值过滤 ：忽略明显不合理的响应时间

改进后的检查函数：

def robust_check(url, service_name, retries=3):
    for i in range(retries):
        try:
            status, elapsed = check_http_service(url, service_name)
            if status and status < 500:  # 忽略服务端错误
                return status, elapsed
        except Exception:
            pass
        time.sleep(1)  # 重试间隔
    return None, None

当需要检查大量服务时，顺序执行效率太低。我们可以使用线程池：

from concurrent.futures import ThreadPoolExecutor

services = [('https://service1/health', 'service1'),
    ('https://service2/health', 'service2'),
    # 更多服务...
]

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(robust_check, url, name) for url, name in services]
    # 可以在这里处理结果

好的告警策略应该：

避免 ” 狼来了 ” 效应
快速发现真正的问题
提供足够的上下文

推荐的分级告警策略：

Warning 级别 ：
单个实例失败
响应时间超过阈值的 50%
通知到 IM 工具
Critical 级别 ：
同一服务的多个实例失败
响应时间持续超过阈值
电话通知值班人员

好的巡检系统也需要被测试。使用 pytest 编写测试用例：

import pytest
from unittest.mock import patch
from your_module import check_http_service

@patch('requests.get')
def test_check_http_service_success(mock_get):
    mock_get.return_value.status_code = 200
    mock_get.return_value.ok = True

    status, elapsed = check_http_service('http://test', 'test')
    assert status == 200
    assert elapsed is not None

@patch('requests.get')
def test_check_http_service_failure(mock_get):
    mock_get.side_effect = Exception("Timeout")

    status, elapsed = check_http_service('http://test', 'test')
    assert status is None
    assert elapsed is None

随着业务增长，单机巡检会遇到瓶颈。我们可以考虑：