From 9c72bffd210debff07a42e3e5512d18abbdb5b64 Mon Sep 17 00:00:00 2001 From: iHeyTang Date: Thu, 14 Aug 2025 14:40:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E7=9B=91=E6=8E=A7?= =?UTF-8?q?=E8=B7=AF=E7=94=B1=E5=92=8C=E7=9B=B8=E5=85=B3=E5=8A=9F=E8=83=BD?= =?UTF-8?q?=EF=BC=8C=E6=9B=B4=E6=96=B0=E4=BE=9D=E8=B5=96=E9=A1=B9=E4=BB=A5?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=96=B0=E7=89=B9=E6=80=A7=EF=BC=8C=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA=E6=9C=8D=E5=8A=A1=E7=9B=91=E6=8E=A7=E8=83=BD=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- requirements.txt | 3 +- run_service.py | 6 + workflow_service/database/api.py | 18 ++ workflow_service/main.py | 3 +- workflow_service/routes/monitor.py | 428 +++++++++++++++++++++++++++++ 5 files changed, 456 insertions(+), 2 deletions(-) create mode 100644 workflow_service/routes/monitor.py diff --git a/requirements.txt b/requirements.txt index 0d66211..b07a582 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,4 +5,5 @@ websockets aiohttp boto3 aiosqlite -sqlalchemy \ No newline at end of file +sqlalchemy +psutil \ No newline at end of file diff --git a/run_service.py b/run_service.py index 22c4b3d..c92682d 100644 --- a/run_service.py +++ b/run_service.py @@ -9,6 +9,7 @@ from pathlib import Path # 加载 .env 文件 try: from dotenv import load_dotenv + env_path = Path(__file__).parent / ".env" load_dotenv(env_path) print(f"已加载环境变量文件: {env_path}") @@ -25,5 +26,10 @@ sys.path.insert(0, project_root) from workflow_service.main import web_app import uvicorn +print("🚀 启动 ComfyUI 工作流服务监控...") +print("📊 监控页面地址: http://localhost:18000/monitor") +print("🔍 健康检查: http://localhost:18000/monitor/health") +print("📈 API 文档: http://localhost:18000/docs") + if __name__ == "__main__": uvicorn.run(web_app, host="0.0.0.0", port=18000) diff --git a/workflow_service/database/api.py b/workflow_service/database/api.py index 3396344..917dda8 100644 --- a/workflow_service/database/api.py +++ b/workflow_service/database/api.py @@ -254,3 +254,21 @@ async def get_running_workflow_runs() -> List[dict]: runs = result.scalars().all() return [run.to_dict() for run in runs] + + +async def get_workflow_runs_recent(start_time: datetime, end_time: datetime) -> List[dict]: + """获取指定时间范围内的最近工作流运行记录""" + async with AsyncSessionLocal() as session: + stmt = ( + select(WorkflowRun) + .where( + WorkflowRun.created_at >= start_time, + WorkflowRun.created_at <= end_time + ) + .order_by(WorkflowRun.created_at.desc()) + ) + + result = await session.execute(stmt) + runs = result.scalars().all() + + return [run.to_dict() for run in runs] diff --git a/workflow_service/main.py b/workflow_service/main.py index d74e434..87c8ee2 100644 --- a/workflow_service/main.py +++ b/workflow_service/main.py @@ -6,7 +6,7 @@ from fastapi.middleware.cors import CORSMiddleware from workflow_service.database.api import init_db from workflow_service.config import Settings -from workflow_service.routes import service, workflow, run, runx +from workflow_service.routes import service, workflow, run, runx, monitor settings = Settings() @@ -25,6 +25,7 @@ web_app.include_router(service.service_router) web_app.include_router(workflow.workflow_router) web_app.include_router(run.run_router) web_app.include_router(runx.runx_router) +web_app.include_router(monitor.monitor_router) @web_app.on_event("startup") diff --git a/workflow_service/routes/monitor.py b/workflow_service/routes/monitor.py new file mode 100644 index 0000000..81e291c --- /dev/null +++ b/workflow_service/routes/monitor.py @@ -0,0 +1,428 @@ +from fastapi import APIRouter, HTTPException +from fastapi.responses import HTMLResponse +import psutil +import asyncio +from datetime import datetime, timedelta +from typing import Dict, Any, List +import json + +from workflow_service.config import settings +from workflow_service.database.api import get_workflow_runs_recent, get_workflow_run_nodes +from workflow_service.comfy.comfy_queue import WorkflowQueueManager + +monitor_router = APIRouter(prefix="/monitor", tags=["监控"]) + +# 全局队列管理器实例 +queue_manager = WorkflowQueueManager() + +@monitor_router.get("/", response_class=HTMLResponse) +async def monitor_dashboard(): + """监控仪表板页面""" + html_content = """ + + + + + + ComfyUI 工作流服务监控 + + + +
+
+

🚀 ComfyUI 工作流服务监控

+

实时监控系统状态、任务队列和服务器健康状态

+
+ + + +
+
+
--
+
CPU 使用率
+
+
+
--
+
内存使用率
+
+
+
--
+
运行中任务
+
+
+
--
+
等待中任务
+
+
+ +
+

🖥️ 服务器状态

+
+ +
+
+ +
+

📋 最近任务

+
+ +
+
+
+ + + + + """ + return HTMLResponse(content=html_content) + +@monitor_router.get("/system-stats") +async def get_system_stats() -> Dict[str, Any]: + """获取系统统计信息""" + try: + cpu_percent = psutil.cpu_percent(interval=1) + memory = psutil.virtual_memory() + disk = psutil.disk_usage('/') + + return { + "cpu_percent": round(cpu_percent, 1), + "memory_percent": round(memory.percent, 1), + "memory_used_gb": round(memory.used / (1024**3), 2), + "memory_total_gb": round(memory.total / (1024**3), 2), + "disk_percent": round(disk.percent, 1), + "disk_used_gb": round(disk.used / (1024**3), 2), + "disk_total_gb": round(disk.total / (1024**3), 2), + "timestamp": datetime.now().isoformat() + } + except Exception as e: + raise HTTPException(status_code=500, detail=f"获取系统统计信息失败: {str(e)}") + +@monitor_router.get("/task-stats") +async def get_task_stats() -> Dict[str, Any]: + """获取任务统计信息""" + try: + # 获取最近24小时的任务统计 + end_time = datetime.now() + start_time = end_time - timedelta(hours=24) + + recent_runs = await get_workflow_runs_recent(start_time, end_time) + + # 统计各种状态的任务数量 + status_counts = {} + for run in recent_runs: + status = run.get('status', 'unknown') + status_counts[status] = status_counts.get(status, 0) + 1 + + return { + "running_tasks": status_counts.get('running', 0), + "pending_tasks": status_counts.get('pending', 0), + "completed_tasks": status_counts.get('completed', 0), + "failed_tasks": status_counts.get('failed', 0), + "total_tasks_24h": len(recent_runs), + "timestamp": datetime.now().isoformat() + } + except Exception as e: + raise HTTPException(status_code=500, detail=f"获取任务统计信息失败: {str(e)}") + +@monitor_router.get("/server-status") +async def get_server_status() -> List[Dict[str, Any]]: + """获取服务器状态信息""" + try: + import aiohttp + import asyncio + + async def check_server_status(server): + try: + timeout = aiohttp.ClientTimeout(total=5) + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.get(f"{server.http_url}/system_stats") as response: + if response.status == 200: + return { + "http_url": server.http_url, + "ws_url": server.ws_url, + "input_dir": server.input_dir, + "output_dir": server.output_dir, + "status": "online", + "response_time": response.headers.get('X-Response-Time', 'N/A') + } + else: + return { + "http_url": server.http_url, + "ws_url": server.ws_url, + "input_dir": server.input_dir, + "output_dir": server.output_dir, + "status": "offline", + "error": f"HTTP {response.status}" + } + except Exception as e: + return { + "http_url": server.http_url, + "ws_url": server.ws_url, + "input_dir": server.input_dir, + "output_dir": server.output_dir, + "status": "offline", + "error": str(e) + } + + # 并发检查所有服务器状态 + tasks = [check_server_status(server) for server in settings.SERVERS] + results = await asyncio.gather(*tasks, return_exceptions=True) + + # 过滤掉异常结果 + valid_results = [] + for result in results: + if isinstance(result, dict): + valid_results.append(result) + + return valid_results + + except Exception as e: + raise HTTPException(status_code=500, detail=f"获取服务器状态失败: {str(e)}") + +@monitor_router.get("/recent-tasks") +async def get_recent_tasks(limit: int = 10) -> List[Dict[str, Any]]: + """获取最近的任务列表""" + try: + end_time = datetime.now() + start_time = end_time - timedelta(hours=24) + + recent_runs = await get_workflow_runs_recent(start_time, end_time) + + # 限制返回数量 + limited_runs = recent_runs[:limit] + + # 格式化返回数据 + formatted_runs = [] + for run in limited_runs: + formatted_runs.append({ + "workflow_run_id": run.get('workflow_run_id'), + "workflow_name": run.get('workflow_name'), + "status": run.get('status', 'unknown'), + "created_at": run.get('created_at'), + "updated_at": run.get('updated_at'), + "api_spec": run.get('api_spec') + }) + + return formatted_runs + + except Exception as e: + raise HTTPException(status_code=500, detail=f"获取最近任务失败: {str(e)}") + +@monitor_router.get("/health") +async def health_check() -> Dict[str, Any]: + """健康检查端点""" + try: + # 检查数据库连接 + from workflow_service.database.connection import get_db + db = get_db() + + # 检查配置 + servers_count = len(settings.SERVERS) + + return { + "status": "healthy", + "timestamp": datetime.now().isoformat(), + "version": "1.0.0", + "database": "connected", + "servers_configured": servers_count, + "uptime": "N/A" # 可以添加启动时间跟踪 + } + except Exception as e: + return { + "status": "unhealthy", + "timestamp": datetime.now().isoformat(), + "error": str(e) + }