taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity

feat: 实现平台监控功能并生成API文档

Browse Source 
功能实现:
1. 创建监控模块 (monitoring.py)
   - 系统健康检查
   - 系统性能指标（CPU、内存、磁盘）
   - 服务统计信息（Agent、执行、工具、用户）
   - 性能趋势分析
   - 系统告警

2. 添加监控API端点
   - GET /api/v1/monitoring/metrics - 系统性能指标
   - GET /api/v1/monitoring/stats - 服务统计信息
   - GET /api/v1/monitoring/trends - 性能趋势数据
   - GET /api/v1/monitoring/alerts - 系统告警
   - GET /api/v1/monitoring/dashboard - 监控仪表板

3. 优化健康检查API
   - 使用监控模块统一管理健康检查

4. 添加依赖
   - psutil==5.9.6 (系统资源监控)

5. 生成API文档
   - Docs/项目文档/API接口文档_监控功能.md
   - 包含完整的API说明、请求/响应示例、集成示例

监控功能:
- 实时系统资源监控（CPU、内存、磁盘）
- 服务指标统计（24小时/7天）
- 性能趋势分析（支持多种时间范围和间隔）
- 智能告警系统（资源告警、服务告警）
- 聚合监控仪表板
This commit is contained in:
xiaohei
2025-12-23 07:36:57 +00:00
parent 326d95c0ab
commit 70608bed16
4 changed files with 1228 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Docs/项目文档/API接口文档_监控功能.md
+723
View File
@@ -0,0 +1,723 @@
# taiji-AI-PAD API 接口文档 - 监控功能
**版本**: v2.0
**创建时间**: 2025年12月23日
**最后更新**: 2025年12月23日
---
## 📋 目录
1. [概述](#概述)
2. [基础信息](#基础信息)
3. [监控API端点](#监控api端点)
4. [请求/响应示例](#请求响应示例)
5. [错误处理](#错误处理)
6. [集成示例](#集成示例)
---
## 1. 概述
本文档描述了 taiji-AI-PAD 平台监控功能的 API 接口。监控功能提供系统健康检查、性能指标、资源使用、服务统计、性能趋势和系统告警等功能。
### 1.1 功能特性
- ✅ 系统健康检查
- ✅ 实时性能指标（CPU、内存、磁盘）
- ✅ 服务统计信息（Agent、执行、工具、用户）
- ✅ 性能趋势分析
- ✅ 系统告警
- ✅ 监控仪表板（聚合数据）
### 1.2 监控指标
- **系统资源**: CPU使用率、内存使用、磁盘使用
- **服务指标**: 活跃Agent数、执行次数、成功率、平均响应时间
- **业务指标**: 日活用户、EU消耗、成本统计
- **告警信息**: 资源告警、服务告警
---
## 2. 基础信息
### 2.1 基础URL
```
http://localhost:8002
```
### 2.2 认证方式
当前版本无需认证，未来版本将支持 JWT Token 认证。
### 2.3 响应格式
所有API响应均为 JSON 格式，使用 UTF-8 编码。
### 2.4 HTTP状态码
| 状态码 | 说明 |
|--------|------|
| 200 | 请求成功 |
| 400 | 请求参数错误 |
| 500 | 服务器内部错误 |
---
## 3. 监控API端点
### 3.1 系统健康检查
#### GET /health
获取系统健康状态。
**请求参数**: 无
**响应示例**:
```json
{
"status": "healthy",
"timestamp": "2025-12-23T07:30:00.000000",
"services": {
"database": "healthy",
"redis": "healthy",
"nats": "healthy"
}
}
```
**响应字段说明**:
- `status`: 系统整体状态 (`healthy`, `degraded`, `unhealthy`)
- `timestamp`: 检查时间戳
- `services`: 各服务健康状态
---
### 3.2 系统性能指标
#### GET /api/v1/monitoring/metrics
获取系统实时性能指标。
**请求参数**: 无
**响应示例**:
```json
{
"timestamp": "2025-12-23T07:30:00.000000",
"system": {
"cpu_usage_percent": 15.5,
"memory_usage_percent": 45.2,
"memory_used_mb": 2048.5,
"memory_total_mb": 4096.0,
"disk_usage_percent": 32.1,
"disk_used_gb": 128.5,
"disk_total_gb": 400.0
},
"services": {
"active_agents": 10,
"total_executions_24h": 1250,
"success_rate_percent": 98.5,
"avg_execution_time_ms": 125.5,
"daily_active_users": 25
},
"billing": {
"total_eu_consumed_24h": 1250.5,
"total_cost_24h": 12.50
}
}
```
**响应字段说明**:
- `system`: 系统资源使用情况
- `cpu_usage_percent`: CPU使用率（%）
- `memory_usage_percent`: 内存使用率（%）
- `memory_used_mb`: 已使用内存（MB）
- `memory_total_mb`: 总内存（MB）
- `disk_usage_percent`: 磁盘使用率（%）
- `disk_used_gb`: 已使用磁盘（GB）
- `disk_total_gb`: 总磁盘空间（GB）
- `services`: 服务指标（过去24小时）
- `active_agents`: 活跃Agent数量
- `total_executions_24h`: 总执行次数
- `success_rate_percent`: 成功率（%）
- `avg_execution_time_ms`: 平均执行时间（毫秒）
- `daily_active_users`: 日活用户数
- `billing`: 计费统计（过去24小时）
- `total_eu_consumed_24h`: 总EU消耗
- `total_cost_24h`: 总成本
---
### 3.3 服务统计信息
#### GET /api/v1/monitoring/stats
获取服务统计信息。
**请求参数**:
| 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| service | string | 否 | 服务类型，可选值: `all`, `agents`, `executions`, `tools`, `users`，默认: `all` |
**请求示例**:
```
GET /api/v1/monitoring/stats?service=agents
```
**响应示例**:
```json
{
"timestamp": "2025-12-23T07:30:00.000000",
"stats": {
"agents": {
"total": 50,
"active": 45,
"inactive": 5,
"avg_executions": 125.5,
"avg_success_rate": 98.2
},
"executions": {
"total_7d": 8750,
"completed": 8600,
"failed": 100,
"running": 50,
"avg_time_ms": 125.5,
"total_eu": 8750.5
},
"tools": {
"total": 20,
"active": 18,
"total_calls": 12500,
"avg_success_rate": 99.5,
"avg_response_time_ms": 50.2
},
"users": {
"total": 100,
"active": 95,
"admins": 5
}
}
}
```
**响应字段说明**:
- `agents`: Agent统计
- `total`: 总Agent数
- `active`: 活跃Agent数
- `inactive`: 非活跃Agent数
- `avg_executions`: 平均执行次数
- `avg_success_rate`: 平均成功率
- `executions`: 执行统计（过去7天）
- `total_7d`: 总执行次数
- `completed`: 成功完成数
- `failed`: 失败数
- `running`: 运行中数
- `avg_time_ms`: 平均执行时间（毫秒）
- `total_eu`: 总EU消耗
- `tools`: 工具统计
- `total`: 总工具数
- `active`: 活跃工具数
- `total_calls`: 总调用次数
- `avg_success_rate`: 平均成功率
- `avg_response_time_ms`: 平均响应时间（毫秒）
- `users`: 用户统计
- `total`: 总用户数
- `active`: 活跃用户数
- `admins`: 管理员数
---
### 3.4 性能趋势数据
#### GET /api/v1/monitoring/trends
获取性能趋势数据。
**请求参数**:
| 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| metric | string | 否 | 指标类型，可选值: `executions`, `eu_consumption`，默认: `executions` |
| period | string | 否 | 时间范围，可选值: `24h`, `7d`, `30d`，默认: `24h` |
| interval | string | 否 | 时间间隔，可选值: `1h`, `6h`, `1d`，默认: `1h` |
**请求示例**:
```
GET /api/v1/monitoring/trends?metric=executions&period=7d&interval=6h
```
**响应示例** (metric=executions):
```json
{
"metric": "executions",
"period": "7d",
"interval": "6h",
"data": [
{
"timestamp": "2025-12-23T00:00:00",
"count": 125,
"avg_time_ms": 120.5,
"success_rate": 98.5
},
{
"timestamp": "2025-12-23T06:00:00",
"count": 150,
"avg_time_ms": 125.2,
"success_rate": 99.0
}
]
}
```
**响应示例** (metric=eu_consumption):
```json
{
"metric": "eu_consumption",
"period": "24h",
"interval": "1h",
"data": [
{
"timestamp": "2025-12-23T00:00:00",
"eu_consumed": 50.5,
"cost": 0.50
},
{
"timestamp": "2025-12-23T01:00:00",
"eu_consumed": 52.3,
"cost": 0.52
}
]
}
```
**响应字段说明**:
- `metric`: 指标类型
- `period`: 时间范围
- `interval`: 时间间隔
- `data`: 趋势数据数组
- `timestamp`: 时间点
- `count`: 执行次数（executions指标）
- `avg_time_ms`: 平均执行时间（executions指标）
- `success_rate`: 成功率（executions指标）
- `eu_consumed`: EU消耗（eu_consumption指标）
- `cost`: 成本（eu_consumption指标）
---
### 3.5 系统告警
#### GET /api/v1/monitoring/alerts
获取系统告警信息。
**请求参数**:
| 参数名 | 类型 | 必填 | 说明 |
|--------|------|------|------|
| severity | string | 否 | 严重程度过滤，可选值: `warning`, `critical`, `info` |
**请求示例**:
```
GET /api/v1/monitoring/alerts?severity=critical
```
**响应示例**:
```json
{
"timestamp": "2025-12-23T07:30:00.000000",
"alerts": [
{
"severity": "warning",
"type": "high_cpu",
"message": "CPU使用率过高: 85.5%",
"timestamp": "2025-12-23T07:29:00.000000"
},
{
"severity": "critical",
"type": "low_disk",
"message": "磁盘空间不足: 92.1%",
"timestamp": "2025-12-23T07:25:00.000000"
}
],
"count": 2
}
```
**响应字段说明**:
- `timestamp`: 查询时间
- `alerts`: 告警列表
- `severity`: 严重程度 (`warning`, `critical`, `info`)
- `type`: 告警类型 (`high_cpu`, `high_memory`, `low_disk`, `high_failure_rate`)
- `message`: 告警消息
- `timestamp`: 告警时间
- `count`: 告警总数
**告警类型说明**:
- `high_cpu`: CPU使用率 > 80%
- `high_memory`: 内存使用率 > 85%
- `low_disk`: 磁盘使用率 > 90%
- `high_failure_rate`: 过去1小时内失败执行 > 10次
---
### 3.6 监控仪表板
#### GET /api/v1/monitoring/dashboard
获取监控仪表板数据（聚合所有监控信息）。
**请求参数**: 无
**响应示例**:
```json
{
"timestamp": "2025-12-23T07:30:00.000000",
"health": {
"status": "healthy",
"timestamp": "2025-12-23T07:30:00.000000",
"uptime_seconds": 86400,
"services": {
"database": "healthy",
"redis": "healthy",
"nats": "healthy"
}
},
"metrics": {
"timestamp": "2025-12-23T07:30:00.000000",
"system": {
"cpu_usage_percent": 15.5,
"memory_usage_percent": 45.2,
"disk_usage_percent": 32.1
},
"services": {
"active_agents": 10,
"total_executions_24h": 1250,
"success_rate_percent": 98.5
},
"billing": {
"total_eu_consumed_24h": 1250.5,
"total_cost_24h": 12.50
}
},
"stats": {
"agents": {
"total": 50,
"active": 45
},
"executions": {
"total_7d": 8750,
"completed": 8600
},
"tools": {
"total": 20,
"active": 18
},
"users": {
"total": 100,
"active": 95
}
},
"alerts": {
"items": [
{
"severity": "warning",
"type": "high_cpu",
"message": "CPU使用率过高: 85.5%",
"timestamp": "2025-12-23T07:29:00.000000"
}
],
"count": 1,
"critical_count": 0,
"warning_count": 1
}
}
```
**响应字段说明**:
- `health`: 系统健康状态
- `metrics`: 系统性能指标
- `stats`: 服务统计信息
- `alerts`: 系统告警
- `items`: 告警列表
- `count`: 告警总数
- `critical_count`: 严重告警数
- `warning_count`: 警告告警数
---
## 4. 请求/响应示例
### 4.1 cURL 示例
#### 获取系统性能指标
```bash
curl -X GET "http://localhost:8002/api/v1/monitoring/metrics"
```
#### 获取Agent统计
```bash
curl -X GET "http://localhost:8002/api/v1/monitoring/stats?service=agents"
```
#### 获取执行趋势（7天，6小时间隔）
```bash
curl -X GET "http://localhost:8002/api/v1/monitoring/trends?metric=executions&period=7d&interval=6h"
```
#### 获取严重告警
```bash
curl -X GET "http://localhost:8002/api/v1/monitoring/alerts?severity=critical"
```
#### 获取监控仪表板
```bash
curl -X GET "http://localhost:8002/api/v1/monitoring/dashboard"
```
### 4.2 Python 示例
```python
import httpx
import asyncio
async def get_monitoring_data():
base_url = "http://localhost:8002"
async with httpx.AsyncClient() as client:
# 获取系统指标
metrics = await client.get(f"{base_url}/api/v1/monitoring/metrics")
print("系统指标:", metrics.json())
# 获取服务统计
stats = await client.get(f"{base_url}/api/v1/monitoring/stats?service=all")
print("服务统计:", stats.json())
# 获取性能趋势
trends = await client.get(
f"{base_url}/api/v1/monitoring/trends",
params={"metric": "executions", "period": "24h", "interval": "1h"}
)
print("性能趋势:", trends.json())
# 获取告警
alerts = await client.get(f"{base_url}/api/v1/monitoring/alerts")
print("系统告警:", alerts.json())
# 获取监控仪表板
dashboard = await client.get(f"{base_url}/api/v1/monitoring/dashboard")
print("监控仪表板:", dashboard.json())
asyncio.run(get_monitoring_data())
```
### 4.3 JavaScript 示例
```javascript
const baseUrl = 'http://localhost:8002';
// 获取系统指标
async function getMetrics() {
const response = await fetch(`${baseUrl}/api/v1/monitoring/metrics`);
const data = await response.json();
console.log('系统指标:', data);
}
// 获取服务统计
async function getStats(service = 'all') {
const response = await fetch(`${baseUrl}/api/v1/monitoring/stats?service=${service}`);
const data = await response.json();
console.log('服务统计:', data);
}
// 获取性能趋势
async function getTrends(metric = 'executions', period = '24h', interval = '1h') {
const url = new URL(`${baseUrl}/api/v1/monitoring/trends`);
url.searchParams.append('metric', metric);
url.searchParams.append('period', period);
url.searchParams.append('interval', interval);
const response = await fetch(url);
const data = await response.json();
console.log('性能趋势:', data);
}
// 获取告警
async function getAlerts(severity = null) {
let url = `${baseUrl}/api/v1/monitoring/alerts`;
if (severity) {
url += `?severity=${severity}`;
}
const response = await fetch(url);
const data = await response.json();
console.log('系统告警:', data);
}
// 获取监控仪表板
async function getDashboard() {
const response = await fetch(`${baseUrl}/api/v1/monitoring/dashboard`);
const data = await response.json();
console.log('监控仪表板:', data);
}
// 使用示例
getMetrics();
getStats('agents');
getTrends('executions', '7d', '6h');
getAlerts('critical');
getDashboard();
```
---
## 5. 错误处理
### 5.1 错误响应格式
```json
{
"detail": "错误描述信息"
}
```
### 5.2 常见错误
| HTTP状态码 | 错误类型 | 说明 |
|-----------|---------|------|
| 400 | Bad Request | 请求参数错误 |
| 500 | Internal Server Error | 服务器内部错误 |
### 5.3 错误处理示例
```python
import httpx
async def get_metrics_safe():
try:
async with httpx.AsyncClient() as client:
response = await client.get("http://localhost:8002/api/v1/monitoring/metrics")
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
print(f"HTTP错误: {e.response.status_code}")
print(f"错误信息: {e.response.text}")
except Exception as e:
print(f"其他错误: {e}")
```
---
## 6. 集成示例
### 6.1 实时监控仪表板
```python
import asyncio
import httpx
from datetime import datetime
async def update_dashboard():
"""每30秒更新一次监控仪表板"""
base_url = "http://localhost:8002"
while True:
try:
async with httpx.AsyncClient() as client:
response = await client.get(f"{base_url}/api/v1/monitoring/dashboard")
data = response.json()
# 显示关键指标
print(f"\n[{datetime.now()}] 监控仪表板")
print(f"系统状态: {data['health']['status']}")
print(f"CPU使用率: {data['metrics']['system']['cpu_usage_percent']:.1f}%")
print(f"内存使用率: {data['metrics']['system']['memory_usage_percent']:.1f}%")
print(f"活跃Agent: {data['metrics']['services']['active_agents']}")
print(f"24小时执行次数: {data['metrics']['services']['total_executions_24h']}")
print(f"成功率: {data['metrics']['services']['success_rate_percent']:.2f}%")
print(f"告警数量: {data['alerts']['count']} (严重: {data['alerts']['critical_count']})")
except Exception as e:
print(f"获取监控数据失败: {e}")
await asyncio.sleep(30)
# 运行监控
asyncio.run(update_dashboard())
```
### 6.2 告警通知
```python
import httpx
import asyncio
async def check_alerts():
"""检查系统告警并发送通知"""
base_url = "http://localhost:8002"
async with httpx.AsyncClient() as client:
# 获取严重告警
response = await client.get(f"{base_url}/api/v1/monitoring/alerts?severity=critical")
alerts = response.json()
if alerts['count'] > 0:
print(f"⚠️ 发现 {alerts['count']} 个严重告警:")
for alert in alerts['alerts']:
print(f" - {alert['message']} ({alert['type']})")
# 这里可以添加通知逻辑（邮件、短信、Slack等）
# 获取警告告警
response = await client.get(f"{base_url}/api/v1/monitoring/alerts?severity=warning")
alerts = response.json()
if alerts['count'] > 0:
print(f"⚠️ 发现 {alerts['count']} 个警告:")
for alert in alerts['alerts']:
print(f" - {alert['message']} ({alert['type']})")
asyncio.run(check_alerts())
```
---
## 7. 最佳实践
### 7.1 监控频率建议
- **系统指标**: 每30秒-1分钟查询一次
- **服务统计**: 每5-10分钟查询一次
- **性能趋势**: 根据需求，建议每1小时查询一次
- **系统告警**: 每1-5分钟检查一次
### 7.2 性能优化
- 使用 `/api/v1/monitoring/dashboard` 端点获取聚合数据，减少请求次数
- 对于趋势数据，合理选择时间范围和间隔，避免查询过大数据集
- 使用缓存机制，避免频繁查询数据库
### 7.3 告警阈值建议
- **CPU使用率**: > 80% 警告，> 90% 严重
- **内存使用率**: > 85% 警告，> 95% 严重
- **磁盘使用率**: > 85% 警告，> 90% 严重
- **失败率**: > 5% 警告，> 10% 严重
---
## 8. 更新日志
### v2.0 (2025-12-23)
- ✅ 新增系统性能指标API
- ✅ 新增服务统计信息API
- ✅ 新增性能趋势数据API
- ✅ 新增系统告警API
- ✅ 新增监控仪表板API
- ✅ 优化健康检查API
---
**文档版本**: v2.0
**最后更新**: 2025年12月23日
**维护者**: taiji-AI-PAD 开发团队
services/mcp-server/main.py
+105 -32
View File
@@ -9,6 +9,7 @@ import logging
import os
from datetime import datetime
from typing import Any, Dict, List, Optional
from fastapi import Query
from uuid import UUID
import structlog
@@ -39,6 +40,7 @@ from schemas import (
from mcp_protocol import MCPProtocolHandler
from database import get_db, init_db
from config import Settings
from monitoring import system_monitor
# 配置日志
structlog.configure(
@@ -354,39 +356,11 @@ async def handle_system_event(msg):
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""健康检查端点"""
services = {
"mcp_server": "healthy",
"redis": "unknown",
"nats": "unknown",
"database": "unknown"
}
# 检查Redis
try:
if redis_client:
await redis_client.ping()
services["redis"] = "healthy"
except Exception:
services["redis"] = "unhealthy"
# 检查NATS
try:
if nats_client and nats_client.is_connected:
services["nats"] = "healthy"
except Exception:
services["nats"] = "unhealthy"
# 检查数据库连接
try:
# 这里应该有数据库连接检查
services["database"] = "healthy"
except Exception:
services["database"] = "unhealthy"
health_data = await system_monitor.get_system_health()
return HealthResponse(
status="healthy",
timestamp=datetime.utcnow().isoformat(),
services=services
status=health_data["status"],
timestamp=health_data["timestamp"],
services=health_data["services"]
)
@app.post("/agents", response_model=AgentCard)
@@ -770,6 +744,105 @@ async def get_metrics():
status_code=500
)
# ========== 监控API端点 ==========
@app.get("/api/v1/monitoring/metrics")
async def get_system_metrics():
"""获取系统性能指标"""
try:
metrics = await system_monitor.get_system_metrics()
return metrics
except Exception as e:
logger.error(f"获取系统指标失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/monitoring/stats")
async def get_service_stats(service: str = "all"):
"""获取服务统计信息
Args:
service: 服务类型 (all, agents, executions, tools, users)
"""
try:
stats = await system_monitor.get_service_stats(service)
return stats
except Exception as e:
logger.error(f"获取服务统计失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/monitoring/trends")
async def get_performance_trends(
metric: str = "executions",
period: str = "24h",
interval: str = "1h"
):
"""获取性能趋势数据
Args:
metric: 指标类型 (executions, eu_consumption)
period: 时间范围 (24h, 7d, 30d)
interval: 时间间隔 (1h, 6h, 1d)
"""
try:
trends = await system_monitor.get_performance_trends(metric, period, interval)
return trends
except Exception as e:
logger.error(f"获取性能趋势失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/monitoring/alerts")
async def get_system_alerts(severity: Optional[str] = None):
"""获取系统告警
Args:
severity: 严重程度过滤 (warning, critical, info)
"""
try:
alerts = await system_monitor.get_alerts(severity)
return {
"timestamp": datetime.utcnow().isoformat(),
"alerts": alerts,
"count": len(alerts)
}
except Exception as e:
logger.error(f"获取系统告警失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/api/v1/monitoring/dashboard")
async def get_monitoring_dashboard():
"""获取监控仪表板数据（聚合所有监控信息）"""
try:
# 并行获取所有监控数据
health_task = system_monitor.get_system_health()
metrics_task = system_monitor.get_system_metrics()
stats_task = system_monitor.get_service_stats("all")
alerts_task = system_monitor.get_alerts()
health, metrics, stats, alerts = await asyncio.gather(
health_task, metrics_task, stats_task, alerts_task
)
return {
"timestamp": datetime.utcnow().isoformat(),
"health": health,
"metrics": metrics,
"stats": stats["stats"],
"alerts": {
"items": alerts,
"count": len(alerts),
"critical_count": len([a for a in alerts if a.get("severity") == "critical"]),
"warning_count": len([a for a in alerts if a.get("severity") == "warning"]),
}
}
except Exception as e:
logger.error(f"获取监控仪表板失败: {e}")
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
import uvicorn
uvicorn.run(
services/mcp-server/monitoring.py
+399
View File
@@ -0,0 +1,399 @@
"""
平台监控模块
提供系统健康检查、性能指标、资源使用等监控功能
"""
import asyncio
import json
import logging
import psutil
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text, func
from sqlalchemy.orm import selectinload
from models import Agent, Execution, User, Tool, Session
from database import AsyncSessionLocal
logger = logging.getLogger(__name__)
class SystemMonitor:
"""系统监控类"""
def __init__(self):
self.start_time = datetime.utcnow()
async def get_system_health(self) -> Dict[str, Any]:
"""获取系统健康状态"""
health = {
"status": "healthy",
"timestamp": datetime.utcnow().isoformat(),
"uptime_seconds": (datetime.utcnow() - self.start_time).total_seconds(),
"services": {}
}
# 检查数据库
try:
async with AsyncSessionLocal() as session:
await session.execute(text("SELECT 1"))
health["services"]["database"] = "healthy"
except Exception as e:
health["services"]["database"] = f"unhealthy: {str(e)}"
health["status"] = "degraded"
return health
async def get_system_metrics(self) -> Dict[str, Any]:
"""获取系统性能指标"""
try:
# 系统资源使用
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
# 数据库统计
async with AsyncSessionLocal() as session:
# Agent统计
agent_result = await session.execute(
text("SELECT COUNT(*) FROM agents WHERE status = 'active'")
)
active_agents = agent_result.scalar() or 0
# 执行统计
execution_result = await session.execute(
text("""
SELECT
COUNT(*) as total,
AVG(execution_time) as avg_time,
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END)::float / NULLIF(COUNT(*), 0) * 100 as success_rate
FROM executions
WHERE started_at > NOW() - INTERVAL '24 hours'
""")
)
exec_stats = execution_result.fetchone()
total_executions = exec_stats[0] or 0 if exec_stats else 0
avg_execution_time = float(exec_stats[1] or 0) if exec_stats and exec_stats[1] else 0.0
success_rate = float(exec_stats[2] or 0) if exec_stats and exec_stats[2] else 0.0
# 用户统计
user_result = await session.execute(
text("""
SELECT COUNT(DISTINCT user_id)
FROM sessions
WHERE created_at > NOW() - INTERVAL '24 hours'
""")
)
daily_active_users = user_result.scalar() or 0
# EU消耗统计
eu_result = await session.execute(
text("""
SELECT
SUM(eu_consumed) as total_eu,
SUM(cost) as total_cost
FROM billing
WHERE created_at > NOW() - INTERVAL '24 hours'
""")
)
eu_stats = eu_result.fetchone()
total_eu = float(eu_stats[0] or 0) if eu_stats and eu_stats[0] else 0.0
total_cost = float(eu_stats[1] or 0) if eu_stats and eu_stats[1] else 0.0
return {
"timestamp": datetime.utcnow().isoformat(),
"system": {
"cpu_usage_percent": cpu_percent,
"memory_usage_percent": memory.percent,
"memory_used_mb": memory.used / 1024 / 1024,
"memory_total_mb": memory.total / 1024 / 1024,
"disk_usage_percent": disk.percent,
"disk_used_gb": disk.used / 1024 / 1024 / 1024,
"disk_total_gb": disk.total / 1024 / 1024 / 1024,
},
"services": {
"active_agents": active_agents,
"total_executions_24h": total_executions,
"success_rate_percent": round(success_rate, 2),
"avg_execution_time_ms": round(avg_execution_time, 2),
"daily_active_users": daily_active_users,
},
"billing": {
"total_eu_consumed_24h": round(total_eu, 4),
"total_cost_24h": round(total_cost, 4),
}
}
except Exception as e:
logger.error(f"获取系统指标失败: {e}")
raise
async def get_service_stats(self, service: str = "all") -> Dict[str, Any]:
"""获取服务统计信息"""
try:
async with AsyncSessionLocal() as session:
stats = {}
if service == "all" or service == "agents":
# Agent统计
agent_stats = await session.execute(
text("""
SELECT
COUNT(*) as total,
COUNT(CASE WHEN status = 'active' THEN 1 END) as active,
COUNT(CASE WHEN status = 'inactive' THEN 1 END) as inactive,
AVG(total_executions) as avg_executions,
AVG(success_rate) as avg_success_rate
FROM agents
""")
)
row = agent_stats.fetchone()
if row:
stats["agents"] = {
"total": row[0] or 0,
"active": row[1] or 0,
"inactive": row[2] or 0,
"avg_executions": float(row[3] or 0),
"avg_success_rate": float(row[4] or 0),
}
if service == "all" or service == "executions":
# 执行统计
exec_stats = await session.execute(
text("""
SELECT
COUNT(*) as total,
COUNT(CASE WHEN status = 'completed' THEN 1 END) as completed,
COUNT(CASE WHEN status = 'failed' THEN 1 END) as failed,
COUNT(CASE WHEN status = 'running' THEN 1 END) as running,
AVG(execution_time) as avg_time,
SUM(eu_consumed) as total_eu
FROM executions
WHERE started_at > NOW() - INTERVAL '7 days'
""")
)
row = exec_stats.fetchone()
if row:
stats["executions"] = {
"total_7d": row[0] or 0,
"completed": row[1] or 0,
"failed": row[2] or 0,
"running": row[3] or 0,
"avg_time_ms": float(row[4] or 0),
"total_eu": float(row[5] or 0),
}
if service == "all" or service == "tools":
# 工具统计
tool_stats = await session.execute(
text("""
SELECT
COUNT(*) as total,
COUNT(CASE WHEN is_active = true THEN 1 END) as active,
SUM(total_calls) as total_calls,
AVG(success_rate) as avg_success_rate,
AVG(avg_response_time) as avg_response_time
FROM tools
""")
)
row = tool_stats.fetchone()
if row:
stats["tools"] = {
"total": row[0] or 0,
"active": row[1] or 0,
"total_calls": row[2] or 0,
"avg_success_rate": float(row[3] or 0),
"avg_response_time_ms": float(row[4] or 0),
}
if service == "all" or service == "users":
# 用户统计
user_stats = await session.execute(
text("""
SELECT
COUNT(*) as total,
COUNT(CASE WHEN is_active = true THEN 1 END) as active,
COUNT(CASE WHEN is_admin = true THEN 1 END) as admins
FROM users
""")
)
row = user_stats.fetchone()
if row:
stats["users"] = {
"total": row[0] or 0,
"active": row[1] or 0,
"admins": row[2] or 0,
}
return {
"timestamp": datetime.utcnow().isoformat(),
"stats": stats
}
except Exception as e:
logger.error(f"获取服务统计失败: {e}")
raise
async def get_performance_trends(
self,
metric: str = "executions",
period: str = "24h",
interval: str = "1h"
) -> Dict[str, Any]:
"""获取性能趋势数据"""
try:
async with AsyncSessionLocal() as session:
# 计算时间范围
if period == "24h":
hours = 24
elif period == "7d":
hours = 168
elif period == "30d":
hours = 720
else:
hours = 24
if interval == "1h":
interval_sql = "1 hour"
elif interval == "6h":
interval_sql = "6 hours"
elif interval == "1d":
interval_sql = "1 day"
else:
interval_sql = "1 hour"
if metric == "executions":
result = await session.execute(
text(f"""
SELECT
DATE_TRUNC('hour', started_at) as time_bucket,
COUNT(*) as count,
AVG(execution_time) as avg_time,
SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END)::float / NULLIF(COUNT(*), 0) * 100 as success_rate
FROM executions
WHERE started_at > NOW() - INTERVAL '{hours} hours'
GROUP BY time_bucket
ORDER BY time_bucket
""")
)
rows = result.fetchall()
return {
"metric": metric,
"period": period,
"interval": interval,
"data": [
{
"timestamp": row[0].isoformat() if row[0] else None,
"count": row[1] or 0,
"avg_time_ms": float(row[2] or 0),
"success_rate": float(row[3] or 0),
}
for row in rows
]
}
elif metric == "eu_consumption":
result = await session.execute(
text(f"""
SELECT
DATE_TRUNC('hour', created_at) as time_bucket,
SUM(eu_consumed) as total_eu,
SUM(cost) as total_cost
FROM billing
WHERE created_at > NOW() - INTERVAL '{hours} hours'
GROUP BY time_bucket
ORDER BY time_bucket
""")
)
rows = result.fetchall()
return {
"metric": metric,
"period": period,
"interval": interval,
"data": [
{
"timestamp": row[0].isoformat() if row[0] else None,
"eu_consumed": float(row[1] or 0),
"cost": float(row[2] or 0),
}
for row in rows
]
}
else:
return {
"metric": metric,
"period": period,
"interval": interval,
"data": []
}
except Exception as e:
logger.error(f"获取性能趋势失败: {e}")
raise
async def get_alerts(self, severity: Optional[str] = None) -> List[Dict[str, Any]]:
"""获取系统告警"""
alerts = []
try:
# 检查系统资源
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
if cpu_percent > 80:
alerts.append({
"severity": "warning",
"type": "high_cpu",
"message": f"CPU使用率过高: {cpu_percent:.1f}%",
"timestamp": datetime.utcnow().isoformat(),
})
if memory.percent > 85:
alerts.append({
"severity": "warning",
"type": "high_memory",
"message": f"内存使用率过高: {memory.percent:.1f}%",
"timestamp": datetime.utcnow().isoformat(),
})
if disk.percent > 90:
alerts.append({
"severity": "critical",
"type": "low_disk",
"message": f"磁盘空间不足: {disk.percent:.1f}%",
"timestamp": datetime.utcnow().isoformat(),
})
# 检查服务健康
async with AsyncSessionLocal() as session:
# 检查失败的执行
failed_result = await session.execute(
text("""
SELECT COUNT(*)
FROM executions
WHERE status = 'failed'
AND started_at > NOW() - INTERVAL '1 hour'
""")
)
failed_count = failed_result.scalar() or 0
if failed_count > 10:
alerts.append({
"severity": "warning",
"type": "high_failure_rate",
"message": f"过去1小时内有 {failed_count} 次执行失败",
"timestamp": datetime.utcnow().isoformat(),
})
# 过滤严重程度
if severity:
alerts = [a for a in alerts if a["severity"] == severity]
return alerts
except Exception as e:
logger.error(f"获取告警失败: {e}")
return []
# 全局监控实例
system_monitor = SystemMonitor()
services/mcp-server/requirements.txt
+1
View File
@@ -35,6 +35,7 @@ bcrypt==4.1.2
prometheus-client==0.19.0
structlog==23.2.0
rich==13.7.0
psutil==5.9.6
# 配置管理
python-dotenv==1.0.0