添加监控接口
This commit is contained in:
@@ -137,6 +137,104 @@ async def health():
|
||||
return await gateway_health()
|
||||
|
||||
|
||||
# 监控演练开关。只影响 GET /monitor 的响应,不摘除 worker、不改变派发。
|
||||
# 进程内存态,重启网关后自动关闭。
|
||||
_monitor_drill = {"enabled": False}
|
||||
|
||||
|
||||
def monitor_payload(status: dict, *, simulated: bool = False) -> dict:
|
||||
"""根据池状态生成监控响应。至少一台 worker 在线才算服务正常。
|
||||
|
||||
simulated 为真时强制 unhealthy,用于监控报警演练;计数字段仍是真实池状态。
|
||||
"""
|
||||
healthy = int(status.get("healthy") or 0)
|
||||
ok = healthy > 0 and not simulated
|
||||
body = {
|
||||
"status": "ok" if ok else "unhealthy",
|
||||
"service": "gateway",
|
||||
"workers_total": status.get("total", 0),
|
||||
"workers_healthy": healthy,
|
||||
"workers_primary_total": status.get("primary_total", 0),
|
||||
"workers_primary_healthy": status.get("primary_healthy", 0),
|
||||
}
|
||||
if simulated:
|
||||
body["simulated"] = True
|
||||
if healthy > 0:
|
||||
body["message"] = "模拟故障:监控测试中,真实 worker 未摘除"
|
||||
else:
|
||||
body["message"] = "模拟故障:监控测试中;同时当前没有可用的 worker"
|
||||
elif not ok:
|
||||
body["message"] = "没有可用的 worker"
|
||||
return body
|
||||
|
||||
|
||||
def _monitor_token_error(request: Request):
|
||||
"""校验 X-Internal-Token 与网关 shared_password 一致。通过时返回 None。"""
|
||||
from gateway.config import get_config
|
||||
|
||||
try:
|
||||
expected = get_config().get("shared_password") or ""
|
||||
except Exception: # noqa: BLE001
|
||||
expected = ""
|
||||
token = request.headers.get("X-Internal-Token")
|
||||
if not expected or token != expected:
|
||||
return JSONResponse(status_code=401, content={
|
||||
"status": "unauthorized",
|
||||
"message": "未授权:缺少或错误的 X-Internal-Token",
|
||||
})
|
||||
return None
|
||||
|
||||
|
||||
@app.get("/monitor", include_in_schema=False)
|
||||
async def monitor():
|
||||
"""供外部监控探测当前服务是否可接业务。
|
||||
|
||||
有在线 worker 返回 200,全部下线返回 503(监控侧据此报警)。
|
||||
演练开关打开时也返回 503,但 worker 池保持原样。
|
||||
只读内存中的池状态,后台健康检查已按配置间隔探测过 worker,这里不再额外发包。
|
||||
"""
|
||||
body = monitor_payload(_get_pool_status_safe(), simulated=_monitor_drill["enabled"])
|
||||
if body["status"] != "ok":
|
||||
return JSONResponse(status_code=503, content=body)
|
||||
return body
|
||||
|
||||
|
||||
@app.post("/monitor/fault", include_in_schema=False)
|
||||
async def monitor_fault(request: Request):
|
||||
"""打开或关闭监控故障模拟。
|
||||
|
||||
Body 必须是 ``{"enabled": true}`` 或 ``{"enabled": false}``(JSON 布尔)。
|
||||
打开后 GET /monitor 返回 503,业务派发不变。需要请求头 X-Internal-Token。
|
||||
"""
|
||||
denied = _monitor_token_error(request)
|
||||
if denied is not None:
|
||||
return denied
|
||||
try:
|
||||
payload = await request.json()
|
||||
except Exception: # noqa: BLE001
|
||||
payload = None
|
||||
if not isinstance(payload, dict) or type(payload.get("enabled")) is not bool:
|
||||
return JSONResponse(status_code=400, content={
|
||||
"status": "error",
|
||||
"message": '需要 JSON:{"enabled": true} 或 {"enabled": false}',
|
||||
})
|
||||
enabled = payload["enabled"]
|
||||
_monitor_drill["enabled"] = enabled
|
||||
if enabled:
|
||||
logger.warning("监控故障模拟已开启:GET /monitor 将返回 503,worker 池未改动")
|
||||
else:
|
||||
logger.warning("监控故障模拟已关闭:GET /monitor 恢复真实池状态")
|
||||
return {
|
||||
"status": "ok",
|
||||
"simulated": enabled,
|
||||
"message": (
|
||||
"已开启模拟故障,GET /monitor 将返回 503"
|
||||
if enabled else
|
||||
"已关闭模拟故障,GET /monitor 恢复真实状态"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@app.get("/", include_in_schema=False)
|
||||
async def index():
|
||||
return {
|
||||
|
||||
Reference in New Issue
Block a user