feat: OTP bind UI + session close knowledge generation + health check service
This commit is contained in:
Simon
@@ -0,0 +1,505 @@
|
||||
# =============================================================================
|
||||
# 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()
|
||||
Reference in New Issue
Block a user