277 lines
9.4 KiB
Python
277 lines
9.4 KiB
Python
"""
|
||
系统监控服务
|
||
提供系统状态、执行统计、性能指标等监控数据
|
||
"""
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import func, and_, case
|
||
from datetime import datetime, timedelta
|
||
from typing import Dict, Any, List
|
||
from app.models.user import User
|
||
from app.models.workflow import Workflow
|
||
from app.models.agent import Agent
|
||
from app.models.execution import Execution
|
||
from app.models.execution_log import ExecutionLog
|
||
from app.models.data_source import DataSource
|
||
from app.models.model_config import ModelConfig
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MonitoringService:
|
||
"""系统监控服务"""
|
||
|
||
@staticmethod
|
||
def get_system_overview(db: Session, user_id: str = None) -> Dict[str, Any]:
|
||
"""
|
||
获取系统概览统计
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user_id: 用户ID（如果提供，则只统计该用户的数据）
|
||
|
||
Returns:
|
||
系统概览数据
|
||
"""
|
||
# 构建基础查询条件
|
||
user_filter = Workflow.user_id == user_id if user_id else True
|
||
|
||
# 统计工作流数量
|
||
workflow_count = db.query(func.count(Workflow.id)).filter(user_filter).scalar() or 0
|
||
|
||
# 统计Agent数量
|
||
agent_filter = Agent.user_id == user_id if user_id else True
|
||
agent_count = db.query(func.count(Agent.id)).filter(agent_filter).scalar() or 0
|
||
|
||
# 统计执行记录数量
|
||
execution_filter = None
|
||
if user_id:
|
||
execution_filter = Execution.workflow_id.in_(
|
||
db.query(Workflow.id).filter(Workflow.user_id == user_id)
|
||
)
|
||
execution_count = db.query(func.count(Execution.id)).filter(
|
||
execution_filter if execution_filter else True
|
||
).scalar() or 0
|
||
|
||
# 统计数据源数量
|
||
data_source_filter = DataSource.user_id == user_id if user_id else True
|
||
data_source_count = db.query(func.count(DataSource.id)).filter(
|
||
data_source_filter
|
||
).scalar() or 0
|
||
|
||
# 统计模型配置数量
|
||
model_config_filter = ModelConfig.user_id == user_id if user_id else True
|
||
model_config_count = db.query(func.count(ModelConfig.id)).filter(
|
||
model_config_filter
|
||
).scalar() or 0
|
||
|
||
# 统计用户数量（仅管理员可见）
|
||
user_count = None
|
||
if not user_id:
|
||
user_count = db.query(func.count(User.id)).scalar() or 0
|
||
|
||
return {
|
||
"workflows": workflow_count,
|
||
"agents": agent_count,
|
||
"executions": execution_count,
|
||
"data_sources": data_source_count,
|
||
"model_configs": model_config_count,
|
||
"users": user_count
|
||
}
|
||
|
||
@staticmethod
|
||
def get_execution_statistics(
|
||
db: Session,
|
||
user_id: str = None,
|
||
days: int = 7
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
获取执行统计信息
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user_id: 用户ID（如果提供，则只统计该用户的数据）
|
||
days: 统计天数（默认7天）
|
||
|
||
Returns:
|
||
执行统计数据
|
||
"""
|
||
# 构建时间范围
|
||
end_time = datetime.utcnow()
|
||
start_time = end_time - timedelta(days=days)
|
||
|
||
# 构建查询条件
|
||
execution_filter = Execution.created_at >= start_time
|
||
if user_id:
|
||
execution_filter = and_(
|
||
execution_filter,
|
||
Execution.workflow_id.in_(
|
||
db.query(Workflow.id).filter(Workflow.user_id == user_id)
|
||
)
|
||
)
|
||
|
||
# 统计总执行数
|
||
total_executions = db.query(func.count(Execution.id)).filter(
|
||
execution_filter
|
||
).scalar() or 0
|
||
|
||
# 统计各状态执行数
|
||
status_stats = db.query(
|
||
Execution.status,
|
||
func.count(Execution.id).label('count')
|
||
).filter(execution_filter).group_by(Execution.status).all()
|
||
|
||
status_counts = {status: count for status, count in status_stats}
|
||
|
||
# 计算成功率
|
||
completed = status_counts.get('completed', 0)
|
||
failed = status_counts.get('failed', 0)
|
||
success_rate = (completed / total_executions * 100) if total_executions > 0 else 0
|
||
|
||
# 统计平均执行时间
|
||
avg_execution_time = db.query(
|
||
func.avg(Execution.execution_time)
|
||
).filter(
|
||
and_(execution_filter, Execution.execution_time.isnot(None))
|
||
).scalar() or 0
|
||
|
||
# 统计最近24小时的执行趋势
|
||
hourly_trends = []
|
||
for i in range(24):
|
||
hour_start = end_time - timedelta(hours=24-i)
|
||
hour_end = hour_start + timedelta(hours=1)
|
||
hour_filter = and_(
|
||
execution_filter,
|
||
Execution.created_at >= hour_start,
|
||
Execution.created_at < hour_end
|
||
)
|
||
hour_count = db.query(func.count(Execution.id)).filter(
|
||
hour_filter
|
||
).scalar() or 0
|
||
hourly_trends.append({
|
||
"hour": hour_start.strftime("%H:00"),
|
||
"count": hour_count
|
||
})
|
||
|
||
return {
|
||
"total": total_executions,
|
||
"status_counts": status_counts,
|
||
"success_rate": round(success_rate, 2),
|
||
"avg_execution_time": round(avg_execution_time, 2) if avg_execution_time else 0,
|
||
"hourly_trends": hourly_trends
|
||
}
|
||
|
||
@staticmethod
|
||
def get_node_type_statistics(
|
||
db: Session,
|
||
user_id: str = None,
|
||
days: int = 7
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取节点类型统计
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user_id: 用户ID（如果提供，则只统计该用户的数据）
|
||
days: 统计天数（默认7天）
|
||
|
||
Returns:
|
||
节点类型统计数据
|
||
"""
|
||
# 构建时间范围
|
||
end_time = datetime.utcnow()
|
||
start_time = end_time - timedelta(days=days)
|
||
|
||
# 构建查询条件
|
||
execution_filter = Execution.created_at >= start_time
|
||
if user_id:
|
||
execution_filter = and_(
|
||
execution_filter,
|
||
Execution.workflow_id.in_(
|
||
db.query(Workflow.id).filter(Workflow.user_id == user_id)
|
||
)
|
||
)
|
||
|
||
# 获取符合条件的执行ID列表
|
||
execution_ids_query = db.query(Execution.id).filter(execution_filter)
|
||
execution_ids = [row[0] for row in execution_ids_query.all()]
|
||
|
||
if not execution_ids:
|
||
return []
|
||
|
||
# 统计各节点类型的执行情况
|
||
node_stats = db.query(
|
||
ExecutionLog.node_type,
|
||
func.count(ExecutionLog.id).label('execution_count'),
|
||
func.sum(ExecutionLog.duration).label('total_duration'),
|
||
func.avg(ExecutionLog.duration).label('avg_duration'),
|
||
func.count(
|
||
case((ExecutionLog.level == 'ERROR', 1))
|
||
).label('error_count')
|
||
).filter(
|
||
and_(
|
||
ExecutionLog.execution_id.in_(execution_ids),
|
||
ExecutionLog.node_type.isnot(None),
|
||
ExecutionLog.duration.isnot(None)
|
||
)
|
||
).group_by(ExecutionLog.node_type).all()
|
||
|
||
result = []
|
||
for node_type, exec_count, total_dur, avg_dur, error_count in node_stats:
|
||
result.append({
|
||
"node_type": node_type,
|
||
"execution_count": exec_count,
|
||
"total_duration": round(total_dur or 0, 2),
|
||
"avg_duration": round(avg_dur or 0, 2),
|
||
"error_count": error_count,
|
||
"success_rate": round((exec_count - error_count) / exec_count * 100, 2) if exec_count > 0 else 0
|
||
})
|
||
|
||
return result
|
||
|
||
@staticmethod
|
||
def get_recent_activities(
|
||
db: Session,
|
||
user_id: str = None,
|
||
limit: int = 10
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
获取最近的活动记录
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
user_id: 用户ID（如果提供，则只统计该用户的数据）
|
||
limit: 返回数量限制
|
||
|
||
Returns:
|
||
最近活动列表
|
||
"""
|
||
# 构建查询条件
|
||
execution_filter = True
|
||
if user_id:
|
||
execution_filter = Execution.workflow_id.in_(
|
||
db.query(Workflow.id).filter(Workflow.user_id == user_id)
|
||
)
|
||
|
||
# 获取最近的执行记录
|
||
recent_executions = db.query(Execution).filter(
|
||
execution_filter
|
||
).order_by(Execution.created_at.desc()).limit(limit).all()
|
||
|
||
result = []
|
||
for execution in recent_executions:
|
||
workflow = db.query(Workflow).filter(
|
||
Workflow.id == execution.workflow_id
|
||
).first() if execution.workflow_id else None
|
||
|
||
result.append({
|
||
"id": execution.id,
|
||
"type": "execution",
|
||
"workflow_name": workflow.name if workflow else "未知工作流",
|
||
"status": execution.status,
|
||
"created_at": execution.created_at.isoformat() if execution.created_at else None,
|
||
"execution_time": execution.execution_time
|
||
})
|
||
|
||
return result
|