适合读者:能运行 Python 脚本,想把“偶尔手动 curl”变成可被定时任务或流水线使用的检查。示例使用 Python 3 标准库,判定规则明确为“最终响应状态是 200”。

1. 巡检脚本要给人和程序各一个答案
一个能打印“访问失败”的脚本,方便人看,却未必方便 cron 或 Jenkins 判断。一个只返回 1 的脚本,能让流水线失败,却又不容易知道哪个目标出了问题。
这篇练习同时提供两个出口:每个目标输出一行 JSON,方便查原因;整个脚本返回退出码,方便外层任务判断成功或失败。
我们先控制需求:逐个发起 GET 请求,只有最终 HTTP 状态 200 才算通过,不读取完整响应正文,不自动重试。它适合练习和少量端点检查,尚不覆盖完整业务交易、并发调度和严格的总时限控制。
2. 先约定状态和退出码
| 情况 | 单个结果 | 脚本最终退出码 |
|---|---|---|
| 所有 URL 最终响应 200 | 每项 ok=true |
0 |
| 任意 URL 非 200 或网络失败 | 对应项 ok=false |
1 |
| 参数不合法,例如 timeout 为负数 | 参数错误信息 | 2 |
urllib 默认会跟随它支持的 HTTP 重定向,因此 /redirect → /ok 最终可能得到 200。如果健康检查要求“不能跳转到登录页”,应继续增加最终地址或业务正文校验,而不是只看状态码。URL 打开和重定向行为见 urllib.request 官方文档。
3. 完整脚本:health_check.py
使用编辑器保存为 health_check.py:
import argparse
import http.client
import json
import math
import time
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit, urlunsplit
from urllib.request import Request, urlopen
def positive_timeout(value):
try:
number = float(value)
except ValueError as exc:
raise argparse.ArgumentTypeError("timeout must be a number") from exc
if not math.isfinite(number) or number <= 0:
raise argparse.ArgumentTypeError("timeout must be finite and greater than zero")
return number
def checked_url(value):
try:
parts = urlsplit(value)
if parts.scheme not in ("http", "https") or not parts.hostname:
raise ValueError("unsupported URL")
if parts.username is not None or parts.password is not None:
raise ValueError("credentials in URL")
if any(ch.isspace() for ch in value):
raise ValueError("whitespace in URL")
_ = parts.port
except ValueError as exc:
raise argparse.ArgumentTypeError(
"use an http/https URL with a valid host/port and no user credentials or whitespace"
) from exc
return value
def log_url(value):
parts = urlsplit(value)
host = parts.hostname or ""
if ":" in host:
host = "[" + host + "]"
if parts.port is not None:
host += ":" + str(parts.port)
return urlunsplit((parts.scheme, host, parts.path, "", ""))
def probe(url, timeout):
started = time.monotonic()
result = {
"target": log_url(url),
"final_target": None,
"ok": False,
"http_status": None,
"error": None,
}
request = Request(url, headers={"User-Agent": "ops-health-check/1.0"})
try:
with urlopen(request, timeout=timeout) as response:
result["http_status"] = response.status
result["final_target"] = log_url(response.geturl())
result["ok"] = response.status == 200
if not result["ok"]:
result["error"] = "unexpected_http_status"
except HTTPError as exc:
result["http_status"] = exc.code
result["final_target"] = log_url(exc.geturl())
result["error"] = "http_error"
exc.close()
except URLError as exc:
result["error"] = type(exc.reason).__name__
except (TimeoutError, OSError, ValueError, http.client.HTTPException) as exc:
result["error"] = type(exc).__name__
result["elapsed_ms"] = round((time.monotonic() - started) * 1000)
return result
def main():
parser = argparse.ArgumentParser(description="Check HTTP endpoints; only status 200 is healthy.")
parser.add_argument("--timeout", type=positive_timeout, default=3.0)
parser.add_argument("urls", type=checked_url, nargs="+")
args = parser.parse_args()
failed = False
for url in args.urls:
result = probe(url, args.timeout)
print(json.dumps(result, ensure_ascii=False), flush=True)
failed = failed or not result["ok"]
return 1 if failed else 0
if __name__ == "__main__":
raise SystemExit(main())
脚本对输入作了几项检查:协议限定 HTTP/HTTPS;需要有效主机和端口;禁止把用户名密码直接写进 URL;超时必须是大于零的有限数值。
urlsplit 负责拆分 URL,但它本身不是完整的业务输入校验器,因此仍要明确检查协议和字段。相关边界见 urllib.parse 官方说明。
日志去掉了查询参数、片段和用户凭据,不输出异常完整文本。路径本身仍可能含敏感业务信息,实际接入日志平台前还需要按自己的 URL 规范处理。
4. 为什么把 HTTPError 和 URLError 分开
收到 503 时,服务器确实返回了 HTTP 响应,只是没有满足健康条件;域名解析失败、连接拒绝或 TLS 验证失败,则可能连有效 HTTP 响应都没拿到。
HTTPError 是 URLError 的子类,因此先处理 HTTPError,才方便保留状态码。脚本另外记录网络异常的类型,避免只留下一个笼统的“失败”。异常关系见 urllib.error 文档。
HTTPS 请求继续使用默认的证书验证。证书错误属于需要定位的问题,不会被脚本悄悄跳过。
timeout 限制的是相关阻塞操作等待,不是“整个 URL 从解析到响应绝不超过三秒”的严格总时限;多个目标还是顺序执行。elapsed_ms 记录拿到响应头或发生错误之前的耗时,不代表下载完整页面的耗时。
5. 在本机准备一个能成功、失败、重定向和延迟的服务
保存 http_demo.py:
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == "/redirect":
self.send_response(302)
self.send_header("Location", "/ok")
self.end_headers()
return
if self.path == "/slow":
time.sleep(1)
ok = self.path in ("/ok", "/slow")
body = b"ok\n" if ok else b"unavailable\n"
self.send_response(200 if ok else 503)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
try:
self.wfile.write(body)
except ConnectionError:
pass
if __name__ == "__main__":
ThreadingHTTPServer(("127.0.0.1", 18082), Handler).serve_forever()
终端 A 运行它,服务只监听本机:
python3 http_demo.py
终端 B 先检查正常接口:
python3 health_check.py --timeout 2 http://127.0.0.1:18082/ok
printf 'exit_code=%s\n' "$?"
预期 JSON 中 ok=true、http_status=200,脚本退出码为 0。耗时由环境决定,不要把某个固定毫秒数当成正确答案。
6. 分别验证四种结果
HTTP 失败:
python3 health_check.py --timeout 2 http://127.0.0.1:18082/fail
printf 'exit_code=%s\n' "$?"
预期 http_status=503、error=http_error,退出码为 1。
重定向:
python3 health_check.py --timeout 2 http://127.0.0.1:18082/redirect
预期最终地址为 /ok,结果通过。原始目标和最终目标都被记录,方便发现“健康检查跳去了另一个地方”。
等待超时:
python3 health_check.py --timeout 0.2 http://127.0.0.1:18082/slow
printf 'exit_code=%s\n' "$?"
服务会延迟一秒才发响应,客户端预期先遇到超时,退出码为 1。异常类型字符串可能因 Python 和系统有所不同,判断时重点看失败状态和超时证据。
多个目标:
python3 health_check.py --timeout 2 \
http://127.0.0.1:18082/ok \
http://127.0.0.1:18082/fail
printf 'exit_code=%s\n' "$?"
预期输出两行结果,其中一个失败便使最终退出码为 1。失败后仍检查后面的目标,可以一次收集整组端点的情况。
还可以使用 --timeout 0 验证参数错误,预期退出码为 2。读取 $? 要紧跟被检查的命令。
7. 接到 cron 或 Jenkins 前,再核对三个条件
第一是路径。调度环境中使用 Python 和脚本的明确路径,提前创建可写的日志目录。可以结合站内的 Shell 定时任务排查文章 检查环境差异。
第二是运行位置。脚本默认使用 urllib 的环境代理机制;它在个人电脑、agent 或容器内看到的 DNS、代理和路由可能不同。你要监测的是哪个访问视角,就从相应位置发起请求。
第三是退出码处理。在 Jenkins 普通 sh 步骤中,返回 1 会使步骤失败;如果外层使用 || true 或忽略 returnStatus,脚本正确报告的失败也可能被隐藏。
需要重试时,明确限制次数和间隔,再评估是否会放大故障期请求。401、403 等持续权限问题,通常不会靠频繁重试自行修复。
8. 下一步如何扩展
真实健康检查可以增加预期正文、JSON 字段、证书有效期或依赖状态,但应让每个结果仍然回答清楚“检查了什么、在哪一步失败”。
实验结束后在终端 A 按 Ctrl+C 停止服务。保留脚本和本机服务,今后修改检查逻辑时,就能先验证成功、失败、超时和重定向这几条路径,再交给定时任务。