506 lines
17 KiB
Python
506 lines
17 KiB
Python
|
# =============================================================================
|
||
|
|
# IT服务台健康检查服务 - 业务监控与自愈
|
||
|
|
# =============================================================================
|
||
|
|
# 功能：
|
||
|
|
# 1. 基础设施检查：nginx/backend/redis/postgres
|
||
|
|
# 2. 业务检查：坐席API/消息发送/企微回调/AI-RAG/WebSocket
|
||
|
|
# 3. 影响级别评估
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import logging
|
||
|
|
import time
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from enum import Enum
|
||
|
|
from typing import Any, Dict, List, Optional
|
||
|
|
|
||
|
|
import aiohttp
|
||
|
|
import redis.asyncio as aioredis
|
||
|
|
from sqlalchemy import text
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from app.config import settings
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class HealthStatus(str, Enum):
|
||
|
|
"""健康状态枚举"""
|
||
|
|
HEALTHY = "healthy"
|
||
|
|
DEGRADED = "degraded"
|
||
|
|
UNHEALTHY = "unhealthy"
|
||
|
|
UNKNOWN = "unknown"
|
||
|
|
|
||
|
|
|
||
|
|
class ImpactLevel(str, Enum):
|
||
|
|
"""影响级别"""
|
||
|
|
P0 = "P0" # 紧急 - 核心业务完全不可用
|
||
|
|
P1 = "P1" # 高 - 部分功能受损
|
||
|
|
P2 = "P2" # 中 - 非核心功能异常
|
||
|
|
P3 = "P3" # 低 - 轻微异常
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class CheckResult:
|
||
|
|
"""单项检查结果"""
|
||
|
|
name: str
|
||
|
|
status: HealthStatus
|
||
|
|
latency_ms: Optional[int] = None
|
||
|
|
error: Optional[str] = None
|
||
|
|
details: Optional[Dict[str, Any]] = None
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class HealthCheckReport:
|
||
|
|
"""健康检查报告"""
|
||
|
|
timestamp: str
|
||
|
|
overall_status: HealthStatus
|
||
|
|
overall_impact: ImpactLevel
|
||
|
|
infrastructure: Dict[str, CheckResult]
|
||
|
|
business: Dict[str, CheckResult]
|
||
|
|
errors: List[Dict[str, Any]]
|
||
|
|
|
||
|
|
|
||
|
|
class HealthCheckService:
|
||
|
|
"""健康检查服务"""
|
||
|
|
|
||
|
|
# 超时配置
|
||
|
|
TIMEOUT_SHORT = 2 # 2秒 - 基础设施
|
||
|
|
TIMEOUT_MEDIUM = 5 # 5秒 - 业务API
|
||
|
|
TIMEOUT_LONG = 15 # 15秒 - AI服务
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self._redis_client: Optional[aioredis.Redis] = None
|
||
|
|
|
||
|
|
async def get_redis(self) -> aioredis.Redis:
|
||
|
|
"""获取 Redis 客户端"""
|
||
|
|
if self._redis_client is None:
|
||
|
|
self._redis_client = settings.create_redis_client()
|
||
|
|
return self._redis_client
|
||
|
|
|
||
|
|
async def close(self):
|
||
|
|
"""关闭连接"""
|
||
|
|
if self._redis_client:
|
||
|
|
await self._redis_client.close()
|
||
|
|
self._redis_client = None
|
||
|
|
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
# 基础设施检查
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
|
||
|
|
async def check_backend(self) -> CheckResult:
|
||
|
|
"""检查后端自身健康状态"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
# 直接检查自身健康端点
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(
|
||
|
|
f"http://localhost:8000/health",
|
||
|
|
timeout=aiohttp.ClientTimeout(total=self.TIMEOUT_SHORT)
|
||
|
|
) as resp:
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
if resp.status == 200:
|
||
|
|
return CheckResult(
|
||
|
|
name="backend",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return CheckResult(
|
||
|
|
name="backend",
|
||
|
|
status=HealthStatus.DEGRADED,
|
||
|
|
latency_ms=latency,
|
||
|
|
error=f"HTTP {resp.status}"
|
||
|
|
)
|
||
|
|
except asyncio.TimeoutError:
|
||
|
|
return CheckResult(
|
||
|
|
name="backend",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error="超时"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="backend",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_redis(self) -> CheckResult:
|
||
|
|
"""检查 Redis 连接"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
redis_client = await self.get_redis()
|
||
|
|
await redis_client.ping()
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
return CheckResult(
|
||
|
|
name="redis",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
except asyncio.TimeoutError:
|
||
|
|
return CheckResult(
|
||
|
|
name="redis",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error="连接超时"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="redis",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_postgres(self, db: AsyncSession) -> CheckResult:
|
||
|
|
"""检查 PostgreSQL 连接"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
await db.execute(text("SELECT 1"))
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
return CheckResult(
|
||
|
|
name="postgres",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="postgres",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_nginx(self) -> CheckResult:
|
||
|
|
"""检查 Nginx（通过本地代理）"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(
|
||
|
|
"http://localhost:80/health",
|
||
|
|
timeout=aiohttp.ClientTimeout(total=self.TIMEOUT_SHORT)
|
||
|
|
) as resp:
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
if resp.status < 500:
|
||
|
|
return CheckResult(
|
||
|
|
name="nginx",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return CheckResult(
|
||
|
|
name="nginx",
|
||
|
|
status=HealthStatus.DEGRADED,
|
||
|
|
latency_ms=latency,
|
||
|
|
error=f"HTTP {resp.status}"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
# Nginx 检查失败不阻塞，返回 unknown
|
||
|
|
return CheckResult(
|
||
|
|
name="nginx",
|
||
|
|
status=HealthStatus.UNKNOWN,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
# 业务检查
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
|
||
|
|
async def check_agents_api(self, db: AsyncSession) -> CheckResult:
|
||
|
|
"""检查坐席列表API"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
result = await db.execute(text("SELECT COUNT(*) FROM agents"))
|
||
|
|
count = result.scalar()
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
return CheckResult(
|
||
|
|
name="agents_api",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency,
|
||
|
|
details={"agent_count": count}
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="agents_api",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_message_send(self, db: AsyncSession) -> CheckResult:
|
||
|
|
"""检查消息发送能力（通过最近消息）"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
# 检查最近一条消息的状态
|
||
|
|
result = await db.execute(
|
||
|
|
text("SELECT created_at FROM messages ORDER BY created_at DESC LIMIT 1")
|
||
|
|
)
|
||
|
|
row = result.fetchone()
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
|
||
|
|
if row:
|
||
|
|
return CheckResult(
|
||
|
|
name="message_send",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency,
|
||
|
|
details={"last_message_at": str(row[0])}
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
# 无消息记录，但表可用
|
||
|
|
return CheckResult(
|
||
|
|
name="message_send",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency,
|
||
|
|
details={"last_message_at": None}
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="message_send",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_wecom_callback(self) -> CheckResult:
|
||
|
|
"""检查企微回调（通过企微服务）"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
from app.services.wecom_service import WecomService
|
||
|
|
wecom = WecomService(await self.get_redis())
|
||
|
|
# 尝试获取企微token
|
||
|
|
token = await wecom.get_access_token()
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
|
||
|
|
if token:
|
||
|
|
return CheckResult(
|
||
|
|
name="wecom_callback",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return CheckResult(
|
||
|
|
name="wecom_callback",
|
||
|
|
status=HealthStatus.DEGRADED,
|
||
|
|
latency_ms=latency,
|
||
|
|
error="无法获取access_token"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="wecom_callback",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_ai_rag(self) -> CheckResult:
|
||
|
|
"""检查AI/RAG服务"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
# 尝试连接 RAGFlow
|
||
|
|
from app.config import settings
|
||
|
|
rag_url = settings.ragflow_url if hasattr(settings, 'ragflow_url') else "http://10.80.0.85:9380"
|
||
|
|
|
||
|
|
async with aiohttp.ClientSession() as session:
|
||
|
|
async with session.get(
|
||
|
|
f"{rag_url}/health",
|
||
|
|
timeout=aiohttp.ClientTimeout(total=self.TIMEOUT_LONG)
|
||
|
|
) as resp:
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
if resp.status == 200:
|
||
|
|
return CheckResult(
|
||
|
|
name="ai_rag",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return CheckResult(
|
||
|
|
name="ai_rag",
|
||
|
|
status=HealthStatus.DEGRADED,
|
||
|
|
latency_ms=latency,
|
||
|
|
error=f"HTTP {resp.status}"
|
||
|
|
)
|
||
|
|
except asyncio.TimeoutError:
|
||
|
|
return CheckResult(
|
||
|
|
name="ai_rag",
|
||
|
|
status=HealthStatus.UNHEALTHY,
|
||
|
|
error="连接超时"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
# RAG 服务不可用不阻塞业务，返回 degraded
|
||
|
|
return CheckResult(
|
||
|
|
name="ai_rag",
|
||
|
|
status=HealthStatus.DEGRADED,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
async def check_websocket(self) -> CheckResult:
|
||
|
|
"""检查WebSocket连接状态"""
|
||
|
|
start = time.time()
|
||
|
|
try:
|
||
|
|
# 通过 Redis 获取活跃连接数
|
||
|
|
redis_client = await self.get_redis()
|
||
|
|
keys = await redis_client.keys("ws:session:*")
|
||
|
|
latency = int((time.time() - start) * 1000)
|
||
|
|
|
||
|
|
return CheckResult(
|
||
|
|
name="websocket",
|
||
|
|
status=HealthStatus.HEALTHY,
|
||
|
|
latency_ms=latency,
|
||
|
|
details={"active_connections": len(keys)}
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
return CheckResult(
|
||
|
|
name="websocket",
|
||
|
|
status=HealthStatus.UNKNOWN,
|
||
|
|
error=str(e)
|
||
|
|
)
|
||
|
|
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
# 综合检查
|
||
|
|
# -------------------------------------------------------------------------
|
||
|
|
|
||
|
|
async def check_all(self, db: Optional[AsyncSession] = None) -> HealthCheckReport:
|
||
|
|
"""执行全量健康检查"""
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
# 基础设施检查（并行）
|
||
|
|
infra_tasks = {
|
||
|
|
"backend": self.check_backend(),
|
||
|
|
"redis": self.check_redis(),
|
||
|
|
"nginx": self.check_nginx(),
|
||
|
|
}
|
||
|
|
|
||
|
|
# 如果有 DB，检查数据库
|
||
|
|
if db:
|
||
|
|
infra_tasks["postgres"] = self.check_postgres(db)
|
||
|
|
|
||
|
|
infra_results = await asyncio.gather(
|
||
|
|
*[task for task in infra_tasks.values()],
|
||
|
|
return_exceptions=True
|
||
|
|
)
|
||
|
|
|
||
|
|
infrastructure = {}
|
||
|
|
for key, result in zip(infra_tasks.keys(), infra_results):
|
||
|
|
if isinstance(result, Exception):
|
||
|
|
infrastructure[key] = CheckResult(
|
||
|
|
name=key,
|
||
|
|
status=HealthStatus.UNKNOWN,
|
||
|
|
error=str(result)
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
infrastructure[key] = result
|
||
|
|
|
||
|
|
# 业务检查（并行）
|
||
|
|
business_tasks = {}
|
||
|
|
if db:
|
||
|
|
business_tasks["agents_api"] = self.check_agents_api(db)
|
||
|
|
business_tasks["message_send"] = self.check_message_send(db)
|
||
|
|
|
||
|
|
business_tasks["wecom_callback"] = self.check_wecom_callback()
|
||
|
|
business_tasks["ai_rag"] = self.check_ai_rag()
|
||
|
|
business_tasks["websocket"] = self.check_websocket()
|
||
|
|
|
||
|
|
business_results = await asyncio.gather(
|
||
|
|
*[task for task in business_tasks.values()],
|
||
|
|
return_exceptions=True
|
||
|
|
)
|
||
|
|
|
||
|
|
business = {}
|
||
|
|
for key, result in zip(business_tasks.keys(), business_results):
|
||
|
|
if isinstance(result, Exception):
|
||
|
|
business[key] = CheckResult(
|
||
|
|
name=key,
|
||
|
|
status=HealthStatus.UNKNOWN,
|
||
|
|
error=str(result)
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
business[key] = result
|
||
|
|
|
||
|
|
# 汇总错误
|
||
|
|
errors = []
|
||
|
|
all_checks = {**infrastructure, **business}
|
||
|
|
for check in all_checks.values():
|
||
|
|
if check.status in [HealthStatus.UNHEALTHY, HealthStatus.DEGRADED]:
|
||
|
|
errors.append({
|
||
|
|
"name": check.name,
|
||
|
|
"status": check.status.value,
|
||
|
|
"error": check.error,
|
||
|
|
"latency_ms": check.latency_ms
|
||
|
|
})
|
||
|
|
|
||
|
|
# 评估整体状态和影响级别
|
||
|
|
overall_status, overall_impact = self._evaluate_status(infrastructure, business)
|
||
|
|
|
||
|
|
return HealthCheckReport(
|
||
|
|
timestamp=datetime.now(timezone.utc).isoformat(),
|
||
|
|
overall_status=overall_status,
|
||
|
|
overall_impact=overall_impact,
|
||
|
|
infrastructure={k: v for k, v in infrastructure.items()},
|
||
|
|
business={k: v for k, v in business.items()},
|
||
|
|
errors=errors
|
||
|
|
)
|
||
|
|
|
||
|
|
def _evaluate_status(
|
||
|
|
self,
|
||
|
|
infrastructure: Dict[str, CheckResult],
|
||
|
|
business: Dict[str, CheckResult]
|
||
|
|
) -> tuple[HealthStatus, ImpactLevel]:
|
||
|
|
"""评估整体状态和影响级别"""
|
||
|
|
|
||
|
|
# 检查关键服务
|
||
|
|
critical_infra = ["backend", "redis", "postgres"]
|
||
|
|
critical_business = ["agents_api", "message_send"]
|
||
|
|
|
||
|
|
has_critical_failure = False
|
||
|
|
has_high_impact = False
|
||
|
|
|
||
|
|
# 基础设施检查
|
||
|
|
for key in critical_infra:
|
||
|
|
if key in infrastructure:
|
||
|
|
if infrastructure[key].status == HealthStatus.UNHEALTHY:
|
||
|
|
has_critical_failure = True
|
||
|
|
if key in ["backend", "redis", "postgres"]:
|
||
|
|
has_high_impact = True
|
||
|
|
|
||
|
|
# 业务检查
|
||
|
|
for key in critical_business:
|
||
|
|
if key in business:
|
||
|
|
if business[key].status == HealthStatus.UNHEALTHY:
|
||
|
|
has_critical_failure = True
|
||
|
|
has_high_impact = True
|
||
|
|
|
||
|
|
# 判断状态
|
||
|
|
if has_critical_failure:
|
||
|
|
return HealthStatus.UNHEALTHY, ImpactLevel.P0
|
||
|
|
elif has_high_impact:
|
||
|
|
return HealthStatus.DEGRADED, ImpactLevel.P1
|
||
|
|
else:
|
||
|
|
# 检查是否有降级
|
||
|
|
degraded = any(
|
||
|
|
c.status == HealthStatus.DEGRADED
|
||
|
|
for c in list(infrastructure.values()) + list(business.values())
|
||
|
|
)
|
||
|
|
if degraded:
|
||
|
|
return HealthStatus.DEGRADED, ImpactLevel.P2
|
||
|
|
|
||
|
|
return HealthStatus.HEALTHY, ImpactLevel.P3
|
||
|
|
|
||
|
|
def to_dict(self, report: HealthCheckReport) -> Dict[str, Any]:
|
||
|
|
"""转换为字典格式（JSON序列化）"""
|
||
|
|
|
||
|
|
def check_to_dict(c: CheckResult) -> Dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"status": c.status.value,
|
||
|
|
"latency_ms": c.latency_ms,
|
||
|
|
"error": c.error,
|
||
|
|
"details": c.details
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
"timestamp": report.timestamp,
|
||
|
|
"overall_status": report.overall_status.value,
|
||
|
|
"overall_impact": report.overall_impact.value,
|
||
|
|
"infrastructure": {k: check_to_dict(v) for k, v in report.infrastructure.items()},
|
||
|
|
"business": {k: check_to_dict(v) for k, v in report.business.items()},
|
||
|
|
"errors": report.errors
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
# 全局单例
|
||
|
|
health_check_service = HealthCheckService()
|