comi-logic/modalDeploy
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化ComfyUI Worker等

Browse Source 
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* FIX 修复缓存校验路径问题
* Merge branch refs/heads/feature/modal-cluster into refs/heads/feature/temp-comfy-cluster
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* FIX webhook直接返回ComfyUI结果
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* 校验媒体文件是否存在缓存中
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* FIX 修改不规范的返回值
* 校验媒体文件是否存在缓存中
* PERF 去除多余的运行完成后重启ComfyUI操作
* 改进 Sentry 追踪系统：为每个关键步骤的 span 添加状态标记
* PERF 分离comfyui router
* Merge branch 'feature/modal-cluster' into feature/temp-comfy-cluster
* Merge branch 'feature/modal-cluster' into feature/temp-comfy-cluster
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* 修改sentry_sdk的弃用方法
* Merge remote-tracking branch 'origin/feature/temp-comfy-cluster' into feature/temp-comfy-cluster
* PERF 去除上传配置文件
* PERF 去除贝蒂上传配置文件

---------

Merge request URL: https://g-ldyi2063.coding.net/p/dev/d/modalDeploy/git/merge/4739
Co-authored-by: 康宇佳，gexianmeng
This commit is contained in:
康宇佳
2025-05-20 16:12:14 +08:00
committed by Coding
parent d3afe850ed
commit 629083210b
11 changed files with 462 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -1,3 +1,6 @@
.pypirc
.env
.idea
.runtime.env
.idea
.venv
test*

2
.runtime.env
View File

@@ -1,5 +1,5 @@
MODAL_ENVIRONMENT=dev
modal_app_name=bowong-ai-video
modal_app_name=cluster-test
S3_mount_dir=/mntS3
S3_bucket_name=modal-media-cache
S3_region=ap-northeast-2

114
src/BowongModalFunctions/api.py
View File

@@ -1,23 +1,11 @@
import modal
from typing import Annotated
from fastapi import FastAPI, Response, Depends, Header, Request
from starlette import status
from fastapi import FastAPI, Request
from scalar_fastapi import get_scalar_api_reference
import sentry_sdk
from sentry_sdk.integrations.loguru import LoguruIntegration, LoggingLevels
from sentry_sdk.integrations.fastapi import FastApiIntegration
from fastapi.middleware.cors import CORSMiddleware
from .middleware.authorization import verify_token
from .utils.KVCache import KVCache
from .utils.ModalUtils import ModalUtils
from .models.media_model import MediaSource
from .models.web_model import (SentryTransactionInfo,
SentryTransactionHeader,
ModalTaskResponse,
ComfyTaskRequest,
ComfyTaskStatusResponse
)
from .router import ffmpeg, cache
from .router import ffmpeg, cache, comfyui
from .config import WorkerConfig
config = WorkerConfig()
@@ -93,100 +81,4 @@ async def scalar():
web_app.include_router(ffmpeg.router)
web_app.include_router(cache.router)
@web_app.post("/comfyui/v1", tags=["ComfyUI"], summary="发起ComfyUIBase任务", description="发起ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
sentry_trace = None
if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUI", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path, item.start_time, item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/comfyui/v1/{task_id}", tags=["ComfyUI"], summary="查询ComfyUIBase任务",
description="查询ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_status != "success":
reason = result["msg"]
else:
media = MediaSource.from_str("s3://" + "/".join(
[config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, result["file_name"]]))
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=media.urn if media else None)
@web_app.post("/comfyui/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务",
description="发起ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
sentry_trace = None
if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUILatentSync15", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path, item.start_time, item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@web_app.get("/comfyui/v2/{task_id}", tags=["ComfyUI"], summary="查询ComfyUILatentSync1.5任务",
description="查询ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_status != "success":
reason = result["msg"]
else:
media = MediaSource.from_str("s3://" + "/".join(
[config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, result["file_name"]]))
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason,
result=media.urn if media else "")
web_app.include_router(comfyui.router)

24
src/BowongModalFunctions/models/media_model.py
View File

@@ -1,4 +1,5 @@
import os
import re
from datetime import datetime
from enum import Enum
from functools import cached_property
@@ -54,15 +55,33 @@ class MediaSource(BaseModel):
endpoint=parsed_url.netloc, # domain of http url
bucket=None,
urn=media_url)
elif media_url.startswith('s3://'): # s3://{endpoint}/{bucket}/{url}
elif media_url.startswith('s3://'):
pattern = r"^s3://[a-z]{2}-[a-z]+-\d/.*$"
if not re.match(pattern, media_url):
media_url = media_url.replace("s3://", f"s3://{s3_region}/")
# s3://{endpoint}/{bucket}/{url}
paths = media_url[5:].split('/')
if len(paths) < 3:
raise ValidationError("URN-s3 格式错误")
return MediaSource(path='/'.join(paths[2:]),
media_source = MediaSource(path='/'.join(paths[2:]),
protocol=MediaProtocol.s3,
endpoint=paths[0],
bucket=paths[1],
urn=media_url)
s3_mount_path = config.S3_mount_dir
cache_path = os.path.join(s3_mount_path, media_source.cache_filepath)
# 校验媒体文件是否存在缓存中
if not os.path.exists(cache_path):
raise ValueError(f"媒体文件 {media_source.cache_filepath} 不存在于缓存中")
return media_source
elif media_url.startswith('vod://'): # vod://{endpoint}/{subAppId}/{fileId}
paths = media_url[6:].split('/')
if len(paths) < 3:
@@ -124,6 +143,7 @@ class MediaSource(BaseModel):
case _:
return f"{self.protocol.value}/{self.endpoint}/{self.bucket}/{self.path}"
@computed_field(description="文件后缀名")
@cached_property
def file_extension(self) -> Optional[str]:

2
src/BowongModalFunctions/models/web_model.py
View File

@@ -319,7 +319,7 @@ class ComfyTaskStatusResponse(BaseModel):
class ComfyTaskRequest(BaseFFMPEGTaskRequest):
video_path: MediaSource = Field(
default=MediaSource.from_str("s3://modal-media-cache/comfyui-input/markers/1016.mp4"), description="视频源")
default=None, description="视频源")
start_time: str = Field(default="00:00:01.600", description="开始时间")
filename_prefix: str = Field(default=str(uuid.uuid4()), description="生成文件名前缀")
tts_text1: str = Field(default="好好看，这是我们家专门为女生定制的背心，", description="tts文本1")

127
src/BowongModalFunctions/router/comfyui.py Normal file
View File

@@ -0,0 +1,127 @@
from typing import Annotated
import modal
from fastapi import APIRouter, Depends, Header
from fastapi.responses import Response
from starlette import status
from ..config import WorkerConfig
from ..middleware.authorization import verify_token
from ..models.media_model import MediaSource
from ..utils.ModalUtils import ModalUtils
from ..models.web_model import (SentryTransactionHeader,
ModalTaskResponse, SentryTransactionInfo,
ComfyTaskRequest, ComfyTaskStatusResponse)
config = WorkerConfig()
router = APIRouter(prefix="/comfyui")
sentry_header_schema = {
"x-trace-id": {
"description": "Sentry Transaction ID",
"schema": {
"type": "string",
}
},
"x-baggage": {
"description": "Sentry Transaction baggage",
"schema": {
"type": "string",
}
}
}
@router.post("/v1", tags=["ComfyUI"], summary="发起ComfyUIBase任务", description="发起ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
sentry_trace = None
if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUI", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/v1/{task_id}", tags=["ComfyUI"], summary="查询ComfyUIBase任务",
description="查询ComfyUIBase任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v1_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_status != "success":
reason = result["msg"]
else:
media = MediaSource.from_str(result["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason, result=media.urn if media else None)
@router.post("/v2", tags=["ComfyUI"], summary="发起ComfyUILatentSync1.5任务",
description="发起ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2(item: ComfyTaskRequest, headers: Annotated[SentryTransactionHeader, Header()]):
sentry_trace = None
if headers.x_trace_id and headers.x_baggage:
sentry_trace = SentryTransactionInfo(x_trace_id=headers.x_trace_id, x_baggage=headers.x_baggage)
cls = modal.Cls.from_name(config.modal_app_name, "ComfyUILatentSync15", environment_name=config.modal_environment)
fn_call = cls().api.spawn(item.video_path.path.replace(config.comfyui_s3_input+"/",""), item.start_time, item.filename_prefix, item.tts_text1,
item.tts_text2, item.tts_text3, item.tts_text4, item.anchor_id, item.speed, sentry_trace,
item.webhook)
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
@router.get("/v2/{task_id}", tags=["ComfyUI"], summary="查询ComfyUILatentSync1.5任务",
description="查询ComfyUILatentSync1.5任务",
responses={
status.HTTP_200_OK: {
"description": "",
"headers": sentry_header_schema
},
},
dependencies=[Depends(verify_token)]
)
async def comfyui_v2_status(task_id: str, response: Response):
task_status, code, reason, result, transaction = await ModalUtils.get_modal_task_status(task_id)
media = None
if transaction:
response.headers["x-trace-id"] = transaction.x_trace_id
response.headers["x-baggage"] = transaction.x_baggage
if task_status == "success":
task_status = result["status"]
if task_status != "success":
reason = result["msg"]
else:
media = MediaSource.from_str(result["file_name"])
return ComfyTaskStatusResponse(taskId=task_id, status=task_status, code=code,
error=reason,
result=media.urn if media else "")

12
src/BowongModalFunctions/utils/SentryUtils.py
View File

@@ -55,6 +55,10 @@ class SentryUtils:
status = TaskStatus.success
error = None
code = ErrorCode.SUCCESS.value
if not result:
status = TaskStatus.failed
error = "Internal Error"
code = ErrorCode.BUSINESS_ERROR.value
except Exception as e:
logger.exception(e)
result = None
@@ -72,9 +76,13 @@ class SentryUtils:
code=code,
results=[result] if isinstance(result, str) else result,
).model_dump())
logger.info(f"webhook response = {response}")
case WebhookMethodEnum.GET:
response = client.post(url=webhook.endpoint.__str__())
response = client.get(url=webhook.endpoint.__str__(), params=BaseFFMPEGTaskStatusResponse(
taskId=func_id, status=status, error=error,
code=code,
results=[result] if isinstance(result, str) else result,
).model_dump())
logger.info(f"webhook response = {response}")
response.raise_for_status()
return result

4
src/cluster/app.py
View File

@@ -16,5 +16,5 @@ app = modal.App(config.modal_app_name,
app.include(media_app)
app.include(ffmpeg_app)
app.include(web_app)
# app.include(comfyui_v1_app)
# app.include(comfyui_v2_app)
app.include(comfyui_v1_app)
app.include(comfyui_v2_app)

261
src/cluster/comfyui_v1.py
View File

@@ -1,4 +1,5 @@
# ComfyUI模板--Base Auth
import backoff
import modal
from dotenv import dotenv_values
@@ -11,8 +12,7 @@ comfyui_image = (
.run_commands("add-apt-repository -y contrib "
"&& wget https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb&&dpkg -i cuda-keyring_1.1-1_all.deb")
.apt_install("git", "gcc", "libportaudio2", "cuda-toolkit", "ffmpeg")
.add_local_file("../whl/comfy_cli-0.0.0-py3-none-any.whl", "/root/comfy_cli-0.0.0-py3-none-any.whl",copy=True)
.pip_install("/root/comfy_cli-0.0.0-py3-none-any.whl")
.pip_install("comfy_cli==0.0.0", index_url="https://packages-1747622887395:0ee15474ccd7b27b57ca63a9306327678e6c2631@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple")
.run_commands("comfy --skip-prompt install --nvidia --version 0.3.10")
.env(dotenv_values("../.runtime.env"))
.workdir("/root/comfy")
@@ -50,8 +50,6 @@ comfyui_image = (
).run_commands(
"rm -rf /root/comfy/ComfyUI/models"
)
.add_local_file("../config/config.yaml", "/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",copy=True)
.add_local_file("../config/config.py","/root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/tools/config.py",copy=True)
.add_local_file("BowongModalFunctions/template/template_v1.json", "/root/src/BowongModalFunctions/template/template_v1.json", copy=True)
.add_local_python_source('BowongModalFunctions')
.add_local_python_source('cluster')
@@ -70,17 +68,12 @@ app.set_description("ComfyUI v1 Server")
with comfyui_image.imports():
import json
import os
import time
import shutil
import subprocess
import uuid
import psutil
from typing import Tuple, Any, Dict, Optional
import loguru
import sentry_sdk
import datetime
import tempfile
import traceback
from modal import current_function_call_id
from BowongModalFunctions.config import WorkerConfig
from BowongModalFunctions.models.web_model import SentryTransactionInfo, WebhookNotify
@@ -90,6 +83,7 @@ with comfyui_image.imports():
config = WorkerConfig()
vol = modal.Volume.from_name("comfyui-model", create_if_missing=True, environment_name=config.modal_environment)
secret = modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment)
custom_secret = modal.Secret.from_name("comfyui-custom-secret", environment_name=config.modal_environment)
output_dir = "/root/comfy/ComfyUI/output"
sentry_sdk.init(
@@ -124,7 +118,7 @@ with comfyui_image.imports():
cpu=(2, 64),
memory=(20480, 131072),
enable_memory_snapshot=False,
secrets=[secret],
secrets=[secret, custom_secret],
volumes={
"/root/comfy/ComfyUI/models": vol,
"/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount(
@@ -151,20 +145,25 @@ with comfyui_image.imports():
@modal.method()
async def api(self, video_path: str, start_time: str = "00:00:03.600", filename_prefix=str(uuid.uuid4()),
tts_text1: str = "测试",
tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", anchor_id: str = "dawan",
speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None, webhook:WebhookNotify = None):
tts_text1: str = "测试", tts_text2: str = "", tts_text3: str = "", tts_text4: str = "",
anchor_id: str = "dawan", speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: WebhookNotify = None):
# 导入必要模块
import datetime
import tempfile
import traceback
from loguru import logger
# 生成当前函数调用的唯一ID
function_call_id = current_function_call_id()
print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}")
logger.info(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}")
@SentryUtils.webhook_handler(webhook=webhook, func_id=function_call_id)
async def infer(workflow: Dict):
async def infer(workflow: Dict) -> str | None:
self.poll_server_health()
self.prompt_uuid = str(uuid.uuid4())
print("Workflow JSON:")
print(json.dumps(workflow, indent=4, ensure_ascii=False))
logger.info(f"Workflow JSON: \n{json.dumps(workflow, indent=4, ensure_ascii=False)}")
for node in workflow.values():
if node.get("class_type") == "ComfyUIDeployExternalText":
if node.get("inputs").get("input_id") == "file_path":
@@ -175,13 +174,16 @@ with comfyui_image.imports():
shutil.copy(file_to_move, file_to_move.replace("input_s3", "input"))
except:
try:
print("Try download file from S3 manually")
logger.warning("尝试手动从S3下载文件")
# S3 Fallback
import boto3
import yaml
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.download_file(config.S3_bucket_name,
@@ -196,15 +198,15 @@ with comfyui_image.imports():
fi.write(json.dumps(workflow, ensure_ascii=False))
cmd = f"comfy run --workflow /root/{self.prompt_uuid}.json --wait --timeout 1190 --verbose"
self.file_prefix = [
node.get("inputs")
for node in workflow.values()
if node.get("class_type") == "VHS_VideoCombine"
][0]["filename_prefix"]
self.file_prefix = \
[node.get("inputs") for node in workflow.values() if node.get("class_type") == "VHS_VideoCombine"][
0][
"filename_prefix"]
subprocess.run(cmd, shell=True, check=True, timeout=1195)
# 获取按照文件时间创建排序的列表，默认是按时间升序
file_list = os.listdir(output_dir)
# 获取按照文件时间创建排序的列表，默认是按时间升序
new_file_list = sorted(file_list, key=lambda file: os.path.getctime(os.path.join(output_dir, file)),
reverse=True)
for f in new_file_list:
@@ -216,20 +218,23 @@ with comfyui_image.imports():
os.path.join(output_dir.replace("output", "output_s3"), f))
except:
try:
print("Try move file to S3 manually")
logger.warning("尝试手动将文件移动到S3")
# S3 Fallback
import boto3
import yaml
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.upload_file(os.path.join(output_dir, f), config.S3_bucket_name,
config.comfyui_s3_output + "/" + f)
except:
raise Exception("Failed to move file to S3 manually")
return f
return "s3://" + "/".join([config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, f])
# 使用临时文件
temp_log_path = None
@@ -237,31 +242,23 @@ with comfyui_image.imports():
# 创建主Sentry事务
if sentry_trace and sentry_trace.x_trace_id and sentry_trace.x_baggage:
transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
"baggage": sentry_trace.x_baggage, })
transaction = sentry_sdk.continue_trace(
environ_or_headers={"sentry-trace": sentry_trace.x_trace_id, "baggage": sentry_trace.x_baggage, })
else:
transaction = sentry_sdk.start_transaction(op="video.processing", name=f"API调用: {video_path}")
sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id, x_baggage=transaction.containing_transaction.get_baggage().serialize())
sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id,
x_baggage=transaction.containing_transaction.get_baggage().serialize())
with transaction:
# 设置全局Sentry上下文
with sentry_sdk.configure_scope() as scope:
# 添加更丰富的上下文数据
scope.set_tag("session_id", self.session_id)
scope.set_tag("function_call_id", function_call_id)
scope.set_tag("video_path", video_path)
scope.set_context("api_parameters", {
"current_function_call_id": function_call_id,
"session_id": self.session_id,
"video_path": video_path,
"start_time": start_time,
"filename_prefix": filename_prefix,
"tts_text1": tts_text1,
"tts_text2": tts_text2,
"tts_text3": tts_text3,
"tts_text4": tts_text4,
"anchor_id": anchor_id,
"speed": speed
})
sentry_sdk.set_tag("session_id", self.session_id)
sentry_sdk.set_tag("function_call_id", function_call_id)
sentry_sdk.set_tag("video_path", video_path)
sentry_sdk.set_context("api_parameters",
{"current_function_call_id": function_call_id, "session_id": self.session_id,
"video_path": video_path, "start_time": start_time,
"filename_prefix": filename_prefix, "tts_text1": tts_text1,
"tts_text2": tts_text2, "tts_text3": tts_text3, "tts_text4": tts_text4,
"anchor_id": anchor_id, "speed": speed})
try:
# 使用Span跟踪日志文件创建
@@ -269,7 +266,7 @@ with comfyui_image.imports():
# 创建临时日志文件
temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt')
temp_log_path = temp_log_file.name
print(f"创建临时日志文件: {temp_log_path}")
logger.debug(f"创建临时日志文件: {temp_log_path}")
span.set_data("log_path", temp_log_path)
span.set_tag("file_operation", "create")
@@ -283,6 +280,9 @@ with comfyui_image.imports():
temp_log_file.flush()
temp_log_file.close()
# 设置状态为成功
span.set_status("ok")
# 使用Span跟踪模板处理
with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span:
# 正常业务逻辑
@@ -307,17 +307,17 @@ with comfyui_image.imports():
modify_span.set_data("modified_nodes",
["211", "399", "331", "406", "424", "225", "330", "515"])
# 设置状态为成功
modify_span.set_status("ok")
span.set_data("workflow_size", len(workflow_file))
span.set_tag("workflow_type", "video_processing")
# 设置状态为成功
span.set_status("ok")
try:
# 记录执行的关键步骤作为breadcrumbs
sentry_sdk.add_breadcrumb(
category="process",
message="开始调用infer方法",
level="info"
)
sentry_sdk.add_breadcrumb(category="process", message="开始调用infer方法", level="info")
# 使用Span跟踪推理过程
with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span:
@@ -328,6 +328,8 @@ with comfyui_image.imports():
with sentry_sdk.start_span(op="inference.prepare", description="准备推理环境") as prep_span:
prep_span.set_data("timestamp", datetime.datetime.now().isoformat())
prep_span.set_tag("model_type", "video_generation")
# 设置状态为成功
prep_span.set_status("ok")
# 调用推理方法
fname = await infer(workflow_file)
@@ -337,6 +339,8 @@ with comfyui_image.imports():
description="处理推理结果") as post_span:
post_span.set_data("output_file", fname)
post_span.set_data("timestamp", datetime.datetime.now().isoformat())
# 设置状态为成功
post_span.set_status("ok")
infer_end_time = datetime.datetime.now()
duration_seconds = (infer_end_time - infer_start_time).total_seconds()
@@ -346,21 +350,24 @@ with comfyui_image.imports():
infer_span.set_data("duration_seconds", duration_seconds)
infer_span.set_data("output_file", fname)
infer_span.set_tag("success", "true" if fname else "false")
# 设置状态为成功
infer_span.set_status("ok")
sentry_sdk.add_breadcrumb(
category="process",
message="infer方法调用完成",
level="info"
)
sentry_sdk.add_breadcrumb(category="process", message="infer方法调用完成", level="info")
if fname is None:
with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"):
with sentry_sdk.start_span(op="error.handling",
description="处理文件缺失错误") as error_span:
error_span.set_status("internal_error") # 设置状态为错误
raise RuntimeError("Output File not found")
# 使用Span跟踪返回结果
with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"):
with sentry_sdk.start_span(op="result.preparation", description="准备返回结果") as result_span:
j = {"status": "success", "file_name": fname}
loguru.logger.success(j)
logger.success(f"处理成功: {j}")
# 设置span状态为成功
result_span.set_status("ok")
# 设置事务状态为成功
transaction.set_status("ok")
@@ -369,20 +376,21 @@ with comfyui_image.imports():
except Exception as infer_error:
# 使用原始异常对象
sentry_sdk.add_breadcrumb(
category="error",
message=f"infer方法调用失败: {str(infer_error)}",
level="error"
)
sentry_sdk.add_breadcrumb(category="error", message=f"infer方法调用失败: {str(infer_error)}",
level="error")
with sentry_sdk.start_span(op="error.handling", description="处理推理错误") as error_span:
error_span.set_data("error_type", type(infer_error).__name__)
error_span.set_data("error_message", str(infer_error))
error_span.set_tag("error", "true")
# 设置状态为错误
error_span.set_status("internal_error")
logger.error(f"推理失败: {str(infer_error)}")
raise infer_error
except Exception as e:
# 异常处理 - 使用捕获的异常对象e
j = {"status": "fail", "msg": str(e)}
j = {"status": "failed", "msg": str(e)}
logger.error(f"处理异常: {str(e)}")
try:
# 使用Span跟踪错误记录过程
@@ -411,58 +419,67 @@ with comfyui_image.imports():
f.write("\n====== ComfyUI日志 ======\n")
f.write(cf.read())
log_span.set_tag("success", "true")
log_span.set_status("ok") # 设置状态为成功
except Exception as log_read_error:
f.write(f"\n无法读取ComfyUI日志: {str(log_read_error)}\n")
log_span.set_data("error", str(log_read_error))
log_span.set_tag("success", "false")
# 设置状态为错误
log_span.set_status("internal_error")
logger.warning(f"无法读取ComfyUI日志: {str(log_read_error)}")
else:
log_span.set_tag("file_exists", "false")
# 设置状态为通知
log_span.set_status("unavailable")
logger.warning("ComfyUI日志文件不存在")
error_span.set_data("log_path", temp_log_path)
error_span.set_tag("error_type", type(e).__name__)
# 设置状态为错误
error_span.set_status("internal_error")
# 仅当尚未报告错误时才添加附件和报告异常
if not error_reported:
with sentry_sdk.configure_scope() as scope:
with sentry_sdk.start_span(op="error.report",
description="向Sentry报告错误") as report_span:
scope.add_attachment(
path=temp_log_path,
filename=f"full_{function_call_id}.txt",
add_to_transactions=True
)
with sentry_sdk.start_span(op="error.report",
description="向Sentry报告错误") as report_span:
# 使用新的API替换configure_scope
# 添加附件
sentry_sdk.add_attachment(path=temp_log_path, filename=f"full_{function_call_id}.txt",
add_to_transactions=True)
# 添加更多异常上下文
scope.set_context("error_details", {
"error_type": type(e).__name__,
"error_message": str(e),
"time_of_error": datetime.datetime.now().isoformat()
})
# 添加更多异常上下文
sentry_sdk.set_context("error_details",
{"error_type": type(e).__name__, "error_message": str(e),
"time_of_error": datetime.datetime.now().isoformat()})
# 设置事务状态为错误
transaction.set_status("internal_error")
# 设置事务状态为错误
transaction.set_status("internal_error")
# 设置span状态为错误
report_span.set_status("internal_error")
# 使用异常对象e
sentry_sdk.capture_exception(e)
error_reported = True
print("已向Sentry报告异常")
report_span.set_tag("reported", "true")
# 使用异常对象e
sentry_sdk.capture_exception(e)
error_reported = True
logger.info("已向Sentry报告异常")
report_span.set_tag("reported", "true")
except Exception as log_error:
print(f"处理日志出错: {log_error}")
logger.error(f"处理日志出错: {log_error}")
# 设置事务状态为错误
transaction.set_status("internal_error")
# 仅当尚未报告错误时才报告原始异常
if not error_reported:
sentry_sdk.capture_exception(e)
print("已向Sentry报告原始异常（日志处理失败）")
logger.info("已向Sentry报告原始异常（日志处理失败）")
return j, sentry_trace
finally:
# 使用Span跟踪清理过程
with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span:
cleanup_successful = True # 跟踪清理是否成功的标志
# 清理临时文件和重启ComfyUI
try:
# 尝试复制日志到S3
@@ -472,56 +489,47 @@ with comfyui_image.imports():
try:
os.makedirs(log_dir, exist_ok=True)
shutil.copy(temp_log_path, f"{log_dir}/full.txt")
print(f"已复制日志到: {log_dir}/full.txt")
logger.info(f"已复制日志到: {log_dir}/full.txt")
copy_span.set_data("source", temp_log_path)
copy_span.set_data("destination", f"{log_dir}/full.txt")
copy_span.set_tag("success", "true")
# 设置状态为成功
copy_span.set_status("ok")
except Exception as copy_error:
print(f"复制日志到S3失败: {copy_error}")
logger.error(f"复制日志到S3失败: {copy_error}")
copy_span.set_data("error", str(copy_error))
copy_span.set_tag("success", "false")
# 设置状态为错误
copy_span.set_status("internal_error")
cleanup_successful = False
# 清理临时文件
with sentry_sdk.start_span(op="db.file.delete", description="删除临时文件") as del_span:
try:
os.unlink(temp_log_path)
print(f"已清理临时文件: {temp_log_path}")
logger.debug(f"已清理临时文件: {temp_log_path}")
del_span.set_data("file", temp_log_path)
del_span.set_tag("success", "true")
# 设置状态为成功
del_span.set_status("ok")
except Exception as unlink_error:
print(f"清理临时文件失败: {unlink_error}")
logger.error(f"清理临时文件失败: {unlink_error}")
del_span.set_data("error", str(unlink_error))
del_span.set_tag("success", "false")
# 设置状态为错误
del_span.set_status("internal_error")
cleanup_successful = False
except Exception as cleanup_error:
print(f"清理过程出错: {cleanup_error}")
logger.error(f"清理过程出错: {cleanup_error}")
cleanup_span.set_data("error", str(cleanup_error))
cleanup_span.set_tag("success", "false")
cleanup_successful = False
print("清理资源并重启ComfyUI")
with sentry_sdk.start_span(op="system.restart", description="重启ComfyUI") as restart_span:
try:
# 停止ComfyUI
with sentry_sdk.start_span(op="system.stop", description="停止ComfyUI") as stop_span:
cmd = "comfy stop"
subprocess.run(cmd, shell=True, check=True)
time.sleep(1)
stop_span.set_tag("success", "true")
except Exception as stop_error:
print(f"停止ComfyUI失败: {stop_error}")
restart_span.set_data("stop_error", str(stop_error))
restart_span.set_tag("stop_success", "false")
try:
# 重启ComfyUI
with sentry_sdk.start_span(op="system.start", description="启动ComfyUI") as start_span:
cmd = "comfy launch --background"
subprocess.run(cmd, shell=True, check=True)
start_span.set_tag("success", "true")
except Exception as start_error:
print(f"启动ComfyUI失败: {start_error}")
restart_span.set_data("start_error", str(start_error))
restart_span.set_tag("start_success", "false")
modal.experimental.stop_fetching_inputs()
# 根据清理和重启结果设置总体清理状态
if cleanup_successful:
cleanup_span.set_status("ok")
else:
cleanup_span.set_status("internal_error")
def poll_server_health(self):
@@ -529,7 +537,6 @@ with comfyui_image.imports():
import urllib
try:
# dummy request to check if the server is healthy
req = urllib.request.Request("http://127.0.0.1:8188/system_stats")
urllib.request.urlopen(req, timeout=5)
print("ComfyUI server is healthy")

324
src/cluster/comfyui_v2.py
View File

@@ -3,20 +3,16 @@ import modal
from dotenv import dotenv_values
comfyui_latentsync_1_5_image = (
modal.Image.debian_slim(
python_version="3.10"
)
modal.Image.debian_slim(python_version="3.10")
.pip_install_from_pyproject("../pyproject_comfyui.toml")
.apt_install("wget", "software-properties-common")
.run_commands("add-apt-repository -y contrib "
"&& wget https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb&&dpkg -i cuda-keyring_1.1-1_all.deb")
"&& wget https://developer.download.nvidia.com/compute/cuda/repos/debian12/x86_64/cuda-keyring_1.1-1_all.deb&&dpkg -i cuda-keyring_1.1-1_all.deb")
.apt_install("git", "gcc", "libportaudio2", "cuda-toolkit", "ffmpeg")
.add_local_file("../whl/comfy_cli-0.0.0-py3-none-any.whl", "/root/comfy_cli-0.0.0-py3-none-any.whl", copy=True)
.pip_install("/root/comfy_cli-0.0.0-py3-none-any.whl")
.pip_install("comfy_cli==0.0.0",index_url="https://packages-1747622887395:0ee15474ccd7b27b57ca63a9306327678e6c2631@g-ldyi2063-pypi.pkg.coding.net/dev/packages/simple")
.run_commands("comfy --skip-prompt install --nvidia --version 0.3.10")
.env(dotenv_values("../.runtime.env"))
.workdir("/root/comfy")
)
.workdir("/root/comfy"))
comfyui_latentsync_1_5_image = (
comfyui_latentsync_1_5_image.run_commands("comfy node install https://github.com/M1kep/ComfyLiterals")
@@ -37,44 +33,29 @@ comfyui_latentsync_1_5_image = (
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/ComfyUI-CustomNode.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/cosyvoice_comfyui.git")
.run_commands("comfy node install https://e.coding.net/g-ldyi2063/dev/Comfy-LatentSync-Next-Node.git")
.run_commands(
"mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model"
).run_commands(
"mkdir -p /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints && rm -rf /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints && ln -s /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints"
).run_commands(
"mkdir -p /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints && mkdir -p /root/.cache/torch/hub/checkpoints && rm -rf /root/.cache/torch/hub/checkpoints && ln -s /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints /root/.cache/torch/hub/checkpoints"
).run_commands(
"mkdir -p /root/comfy/ComfyUI/models/stabilityai && ln -s /root/comfy/ComfyUI/models/stabilityai /root/comfy/ComfyUI/stabilityai"
).run_commands(
"mkdir -p /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M && mkdir -p /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && rm -rf /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && ln -s /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M"
).run_commands(
"rm -rf /root/comfy/ComfyUI/models"
)
.add_local_file("../config/config.yaml", "/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml", copy=True)
.add_local_file("../config/config.py", "/root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/tools/config.py", copy=True)
.add_local_file("BowongModalFunctions/template/template_v2.json", "/root/src/BowongModalFunctions/template/template_v2.json", copy=True)
.run_commands("mkdir -p /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model && rm -rf /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model && ln -s /root/comfy/ComfyUI/models/ComfyUI-CustomNode/model /root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/model")
.run_commands("mkdir -p /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints && rm -rf /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints && ln -s /root/comfy/ComfyUI/models/ComfyUI-LatentSync-Next-Node/checkpoints /root/comfy/ComfyUI/custom_nodes/Comfy-LatentSync-Next-Node/checkpoints")
.run_commands("mkdir -p /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints && mkdir -p /root/.cache/torch/hub/checkpoints && rm -rf /root/.cache/torch/hub/checkpoints && ln -s /root/comfy/ComfyUI/models/.cache/torch/hub/checkpoints /root/.cache/torch/hub/checkpoints")
.run_commands("mkdir -p /root/comfy/ComfyUI/models/stabilityai && ln -s /root/comfy/ComfyUI/models/stabilityai /root/comfy/ComfyUI/stabilityai")
.run_commands("mkdir -p /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M && mkdir -p /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && rm -rf /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M && ln -s /root/comfy/ComfyUI/models/CosyVoice-ComfyUI/pretrained_models/CosyVoice-300M /root/comfy/ComfyUI/custom_nodes/cosyvoice_comfyui/pretrained_models/CosyVoice-300M")
.run_commands("rm -rf /root/comfy/ComfyUI/models")
.add_local_file("BowongModalFunctions/template/template_v2.json","/root/src/BowongModalFunctions/template/template_v2.json",copy=True)
.add_local_python_source('cluster')
.add_local_python_source('BowongModalFunctions')
)
.add_local_python_source('BowongModalFunctions'))
app = modal.App(name="ComfyUI v2", image=comfyui_latentsync_1_5_image, include_source=False)
app.set_description("ComfyUI v2 Server")
with comfyui_latentsync_1_5_image.imports():
import json
import os
import shutil
import subprocess
import time
import uuid
from typing import Dict, Tuple, Any, Optional
import loguru
import sentry_sdk
import psutil
import datetime
import tempfile
import traceback
from loguru import logger
from BowongModalFunctions.config import WorkerConfig
from modal import current_function_call_id
from sentry_sdk.integrations.loguru import LoguruIntegration
@@ -84,6 +65,7 @@ with comfyui_latentsync_1_5_image.imports():
config = WorkerConfig()
vol = modal.Volume.from_name("comfyui-model", create_if_missing=True, environment_name=config.modal_environment)
secret = modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment)
custom_secret = modal.Secret.from_name("comfyui-custom-secret", environment_name=config.modal_environment)
output_dir = "/root/comfy/ComfyUI/output"
sentry_sdk.init(
@@ -106,6 +88,7 @@ with comfyui_latentsync_1_5_image.imports():
return (cpu_freq, cpu_count, mem)
@app.cls(
max_containers=200,
min_containers=0,
@@ -113,49 +96,46 @@ with comfyui_latentsync_1_5_image.imports():
scaledown_window=120,
timeout=1200,
gpu=["L4"],
cpu=(2,64),
memory=(20480,131072),
cpu=(2, 64),
memory=(20480, 131072),
enable_memory_snapshot=False,
secrets=[secret],
volumes={
"/root/comfy/ComfyUI/models": vol,
"/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount(
bucket_name=config.S3_bucket_name,
secret=secret,
key_prefix=config.comfyui_s3_input+"/"
),
"/root/comfy/ComfyUI/output_s3": modal.CloudBucketMount(
bucket_name=config.S3_bucket_name,
secret=secret,
key_prefix=config.comfyui_s3_output+"/"
),
},
)
secrets=[secret, custom_secret],
volumes={"/root/comfy/ComfyUI/models": vol,
"/root/comfy/ComfyUI/input_s3": modal.CloudBucketMount(bucket_name=config.S3_bucket_name, secret=secret,
key_prefix=config.comfyui_s3_input + "/"),
"/root/comfy/ComfyUI/output_s3": modal.CloudBucketMount(bucket_name=config.S3_bucket_name, secret=secret,
key_prefix=config.comfyui_s3_output + "/"), }, )
@modal.concurrent(max_inputs=1)
class ComfyUILatentSync15:
@modal.enter()
def launch_comfy_background(self):
self.session_id = str(uuid.uuid4())
self.file_prefix = None
cmd = "echo client_uuid: {} && comfy launch --background".format(self.session_id,self.session_id, self.session_id)
cmd = "echo client_uuid: {} && comfy launch --background".format(self.session_id, self.session_id,
self.session_id)
subprocess.run(cmd, shell=True, check=True)
@modal.method()
async def api(self, video_path: str, start_time: str = "00:00:03.600", filename_prefix=str(uuid.uuid4()),
tts_text1: str = "测试",
tts_text2: str = "", tts_text3: str = "", tts_text4: str = "", anchor_id: str = "dawan",
speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None, webhook:WebhookNotify = None):
tts_text1: str = "测试", tts_text2: str = "", tts_text3: str = "", tts_text4: str = "",
anchor_id: str = "dawan", speed: float = 1, sentry_trace: Optional[SentryTransactionInfo] = None,
webhook: WebhookNotify = None):
# 导入必要模块
import datetime
import tempfile
import traceback
from loguru import logger
# 生成当前函数调用的唯一ID
function_call_id = current_function_call_id()
print(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}")
logger.info(f"开始处理任务，会话ID: {self.session_id}, 调用ID: {function_call_id}")
@SentryUtils.webhook_handler(webhook=webhook, func_id=function_call_id)
async def infer(workflow: Dict):
async def infer(workflow: Dict) -> str | None:
self.poll_server_health()
self.prompt_uuid = str(uuid.uuid4())
print("Workflow JSON:")
print(json.dumps(workflow, indent=4, ensure_ascii=False))
logger.info(f"Workflow JSON: \n{json.dumps(workflow, indent=4, ensure_ascii=False)}")
for node in workflow.values():
if node.get("class_type") == "ComfyUIDeployExternalText":
if node.get("inputs").get("input_id") == "file_path":
@@ -166,13 +146,16 @@ with comfyui_latentsync_1_5_image.imports():
shutil.copy(file_to_move, file_to_move.replace("input_s3", "input"))
except:
try:
print("Try download file to S3 manually")
logger.warning("尝试手动从S3下载文件")
# S3 Fallback
import boto3
import yaml
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.download_file(config.S3_bucket_name,
@@ -187,11 +170,8 @@ with comfyui_latentsync_1_5_image.imports():
fi.write(json.dumps(workflow, ensure_ascii=False))
cmd = f"comfy run --workflow /root/{self.prompt_uuid}.json --wait --timeout 1190 --verbose"
self.file_prefix = [
node.get("inputs")
for node in workflow.values()
if node.get("class_type") == "VHS_VideoCombine"
][0]["filename_prefix"]
self.file_prefix = \
[node.get("inputs") for node in workflow.values() if node.get("class_type") == "VHS_VideoCombine"][0]["filename_prefix"]
subprocess.run(cmd, shell=True, check=True, timeout=1195)
@@ -208,20 +188,23 @@ with comfyui_latentsync_1_5_image.imports():
os.path.join(output_dir.replace("output", "output_s3"), f))
except:
try:
print("Try move file to S3 manually")
logger.warning("尝试手动将文件移动到S3")
# S3 Fallback
import boto3
import yaml
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
if "aws_key_id" in list(os.environ.keys()):
yaml_config = os.environ
else:
with open("/root/comfy/ComfyUI/custom_nodes/ComfyUI-CustomNode/config.yaml",
encoding="utf-8", mode="r+") as myconfig:
yaml_config = yaml.load(myconfig, Loader=yaml.FullLoader)
awss3 = boto3.resource('s3', aws_access_key_id=yaml_config["aws_key_id"],
aws_secret_access_key=yaml_config["aws_access_key"])
awss3.meta.client.upload_file(os.path.join(output_dir, f), config.S3_bucket_name,
config.comfyui_s3_output + "/" + f)
except:
raise Exception("Failed to move file to S3 manually")
return f
return "s3://" + "/".join([config.S3_region, config.S3_bucket_name, config.comfyui_s3_output, f])
# 使用临时文件
temp_log_path = None
@@ -229,31 +212,22 @@ with comfyui_latentsync_1_5_image.imports():
# 创建主Sentry事务
if sentry_trace and sentry_trace.x_trace_id and sentry_trace.x_baggage:
transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id,
"baggage": sentry_trace.x_baggage, })
transaction = sentry_sdk.continue_trace(environ_or_headers={"sentry-trace": sentry_trace.x_trace_id, "baggage": sentry_trace.x_baggage, })
else:
transaction = sentry_sdk.start_transaction(op="video.processing", name=f"API调用: {video_path}")
sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id, x_baggage=transaction.containing_transaction.get_baggage().serialize())
sentry_trace = SentryTransactionInfo(x_trace_id=transaction.trace_id,
x_baggage=transaction.containing_transaction.get_baggage().serialize())
with transaction:
# 设置全局Sentry上下文
with sentry_sdk.configure_scope() as scope:
# 添加更丰富的上下文数据
scope.set_tag("session_id", self.session_id)
scope.set_tag("function_call_id", function_call_id)
scope.set_tag("video_path", video_path)
scope.set_context("api_parameters", {
"current_function_call_id": function_call_id,
"session_id": self.session_id,
"video_path": video_path,
"start_time": start_time,
"filename_prefix": filename_prefix,
"tts_text1": tts_text1,
"tts_text2": tts_text2,
"tts_text3": tts_text3,
"tts_text4": tts_text4,
"anchor_id": anchor_id,
"speed": speed
})
sentry_sdk.set_tag("session_id", self.session_id)
sentry_sdk.set_tag("function_call_id", function_call_id)
sentry_sdk.set_tag("video_path", video_path)
sentry_sdk.set_context("api_parameters",
{"current_function_call_id": function_call_id, "session_id": self.session_id,
"video_path": video_path, "start_time": start_time,
"filename_prefix": filename_prefix, "tts_text1": tts_text1,
"tts_text2": tts_text2, "tts_text3": tts_text3, "tts_text4": tts_text4,
"anchor_id": anchor_id, "speed": speed})
try:
# 使用Span跟踪日志文件创建
@@ -261,7 +235,7 @@ with comfyui_latentsync_1_5_image.imports():
# 创建临时日志文件
temp_log_file = tempfile.NamedTemporaryFile(mode='w+', delete=False, suffix='.txt')
temp_log_path = temp_log_file.name
print(f"创建临时日志文件: {temp_log_path}")
logger.debug(f"创建临时日志文件: {temp_log_path}")
span.set_data("log_path", temp_log_path)
span.set_tag("file_operation", "create")
@@ -275,6 +249,9 @@ with comfyui_latentsync_1_5_image.imports():
temp_log_file.flush()
temp_log_file.close()
# 设置状态为成功
span.set_status("ok")
# 使用Span跟踪模板处理
with sentry_sdk.start_span(op="template.json", description="处理工作流模板") as span:
# 正常业务逻辑
@@ -290,8 +267,7 @@ with comfyui_latentsync_1_5_image.imports():
workflow_file["399"]["inputs"]["default_value"] = tts_text2
workflow_file["331"]["inputs"]["default_value"] = tts_text3
workflow_file["406"]["inputs"]["default_value"] = tts_text4
workflow_file["424"]["inputs"][
"default_value"] = "/root/comfy/ComfyUI/input_s3/" + video_path
workflow_file["424"]["inputs"]["default_value"] = "/root/comfy/ComfyUI/input_s3/" + video_path
workflow_file["225"]["inputs"]["filename_prefix"] = filename_prefix
workflow_file["330"]["inputs"]["default_value"] = start_time
workflow_file["515"]["inputs"]["anchor_id"] = anchor_id
@@ -299,17 +275,17 @@ with comfyui_latentsync_1_5_image.imports():
modify_span.set_data("modified_nodes",
["211", "399", "331", "406", "424", "225", "330", "515"])
# 设置状态为成功
modify_span.set_status("ok")
span.set_data("workflow_size", len(workflow_file))
span.set_tag("workflow_type", "video_processing")
# 设置状态为成功
span.set_status("ok")
try:
# 记录执行的关键步骤作为breadcrumbs
sentry_sdk.add_breadcrumb(
category="process",
message="开始调用infer方法",
level="info"
)
sentry_sdk.add_breadcrumb(category="process", message="开始调用infer方法", level="info")
# 使用Span跟踪推理过程
with sentry_sdk.start_span(op="ai.inference", description="执行模型推理") as infer_span:
@@ -320,6 +296,8 @@ with comfyui_latentsync_1_5_image.imports():
with sentry_sdk.start_span(op="inference.prepare", description="准备推理环境") as prep_span:
prep_span.set_data("timestamp", datetime.datetime.now().isoformat())
prep_span.set_tag("model_type", "video_generation")
# 设置状态为成功
prep_span.set_status("ok")
# 调用推理方法
fname = await infer(workflow_file)
@@ -329,6 +307,8 @@ with comfyui_latentsync_1_5_image.imports():
description="处理推理结果") as post_span:
post_span.set_data("output_file", fname)
post_span.set_data("timestamp", datetime.datetime.now().isoformat())
# 设置状态为成功
post_span.set_status("ok")
infer_end_time = datetime.datetime.now()
duration_seconds = (infer_end_time - infer_start_time).total_seconds()
@@ -338,21 +318,24 @@ with comfyui_latentsync_1_5_image.imports():
infer_span.set_data("duration_seconds", duration_seconds)
infer_span.set_data("output_file", fname)
infer_span.set_tag("success", "true" if fname else "false")
# 设置状态为成功
infer_span.set_status("ok")
sentry_sdk.add_breadcrumb(
category="process",
message="infer方法调用完成",
level="info"
)
sentry_sdk.add_breadcrumb(category="process", message="infer方法调用完成", level="info")
if fname is None:
with sentry_sdk.start_span(op="error.handling", description="处理文件缺失错误"):
with sentry_sdk.start_span(op="error.handling",
description="处理文件缺失错误") as error_span:
error_span.set_status("internal_error") # 设置状态为错误
raise RuntimeError("Output File not found")
# 使用Span跟踪返回结果
with sentry_sdk.start_span(op="result.preparation", description="准备返回结果"):
with sentry_sdk.start_span(op="result.preparation", description="准备返回结果") as result_span:
j = {"status": "success", "file_name": fname}
loguru.logger.success(j)
logger.success(f"处理成功: {j}")
# 设置span状态为成功
result_span.set_status("ok")
# 设置事务状态为成功
transaction.set_status("ok")
@@ -361,20 +344,21 @@ with comfyui_latentsync_1_5_image.imports():
except Exception as infer_error:
# 使用原始异常对象
sentry_sdk.add_breadcrumb(
category="error",
message=f"infer方法调用失败: {str(infer_error)}",
level="error"
)
sentry_sdk.add_breadcrumb(category="error", message=f"infer方法调用失败: {str(infer_error)}",
level="error")
with sentry_sdk.start_span(op="error.handling", description="处理推理错误") as error_span:
error_span.set_data("error_type", type(infer_error).__name__)
error_span.set_data("error_message", str(infer_error))
error_span.set_tag("error", "true")
# 设置状态为错误
error_span.set_status("internal_error")
logger.error(f"推理失败: {str(infer_error)}")
raise infer_error
except Exception as e:
# 异常处理 - 使用捕获的异常对象e
j = {"status": "fail", "msg": str(e)}
j = {"status": "failed", "msg": str(e)}
logger.error(f"处理异常: {str(e)}")
try:
# 使用Span跟踪错误记录过程
@@ -403,58 +387,67 @@ with comfyui_latentsync_1_5_image.imports():
f.write("\n====== ComfyUI日志 ======\n")
f.write(cf.read())
log_span.set_tag("success", "true")
log_span.set_status("ok") # 设置状态为成功
except Exception as log_read_error:
f.write(f"\n无法读取ComfyUI日志: {str(log_read_error)}\n")
log_span.set_data("error", str(log_read_error))
log_span.set_tag("success", "false")
# 设置状态为错误
log_span.set_status("internal_error")
logger.warning(f"无法读取ComfyUI日志: {str(log_read_error)}")
else:
log_span.set_tag("file_exists", "false")
# 设置状态为通知
log_span.set_status("unavailable")
logger.warning("ComfyUI日志文件不存在")
error_span.set_data("log_path", temp_log_path)
error_span.set_tag("error_type", type(e).__name__)
# 设置状态为错误
error_span.set_status("internal_error")
# 仅当尚未报告错误时才添加附件和报告异常
if not error_reported:
with sentry_sdk.configure_scope() as scope:
with sentry_sdk.start_span(op="error.report",
description="向Sentry报告错误") as report_span:
scope.add_attachment(
path=temp_log_path,
filename=f"full_{function_call_id}.txt",
add_to_transactions=True
)
with sentry_sdk.start_span(op="error.report",
description="向Sentry报告错误") as report_span:
# 使用新的API替换configure_scope
# 添加附件
sentry_sdk.add_attachment(path=temp_log_path, filename=f"full_{function_call_id}.txt",
add_to_transactions=True)
# 添加更多异常上下文
scope.set_context("error_details", {
"error_type": type(e).__name__,
"error_message": str(e),
"time_of_error": datetime.datetime.now().isoformat()
})
# 添加更多异常上下文
sentry_sdk.set_context("error_details",
{"error_type": type(e).__name__, "error_message": str(e),
"time_of_error": datetime.datetime.now().isoformat()})
# 设置事务状态为错误
transaction.set_status("internal_error")
# 设置事务状态为错误
transaction.set_status("internal_error")
# 设置span状态为错误
report_span.set_status("internal_error")
# 使用异常对象e
sentry_sdk.capture_exception(e)
error_reported = True
print("已向Sentry报告异常")
report_span.set_tag("reported", "true")
# 使用异常对象e
sentry_sdk.capture_exception(e)
error_reported = True
logger.info("已向Sentry报告异常")
report_span.set_tag("reported", "true")
except Exception as log_error:
print(f"处理日志出错: {log_error}")
logger.error(f"处理日志出错: {log_error}")
# 设置事务状态为错误
transaction.set_status("internal_error")
# 仅当尚未报告错误时才报告原始异常
if not error_reported:
sentry_sdk.capture_exception(e)
print("已向Sentry报告原始异常（日志处理失败）")
logger.info("已向Sentry报告原始异常（日志处理失败）")
return j, sentry_trace
finally:
# 使用Span跟踪清理过程
with sentry_sdk.start_span(op="system.cleanup", description="清理资源") as cleanup_span:
cleanup_successful = True # 跟踪清理是否成功的标志
# 清理临时文件和重启ComfyUI
try:
# 尝试复制日志到S3
@@ -464,56 +457,47 @@ with comfyui_latentsync_1_5_image.imports():
try:
os.makedirs(log_dir, exist_ok=True)
shutil.copy(temp_log_path, f"{log_dir}/full.txt")
print(f"已复制日志到: {log_dir}/full.txt")
logger.info(f"已复制日志到: {log_dir}/full.txt")
copy_span.set_data("source", temp_log_path)
copy_span.set_data("destination", f"{log_dir}/full.txt")
copy_span.set_tag("success", "true")
# 设置状态为成功
copy_span.set_status("ok")
except Exception as copy_error:
print(f"复制日志到S3失败: {copy_error}")
logger.error(f"复制日志到S3失败: {copy_error}")
copy_span.set_data("error", str(copy_error))
copy_span.set_tag("success", "false")
# 设置状态为错误
copy_span.set_status("internal_error")
cleanup_successful = False
# 清理临时文件
with sentry_sdk.start_span(op="db.file.delete", description="删除临时文件") as del_span:
try:
os.unlink(temp_log_path)
print(f"已清理临时文件: {temp_log_path}")
logger.debug(f"已清理临时文件: {temp_log_path}")
del_span.set_data("file", temp_log_path)
del_span.set_tag("success", "true")
# 设置状态为成功
del_span.set_status("ok")
except Exception as unlink_error:
print(f"清理临时文件失败: {unlink_error}")
logger.error(f"清理临时文件失败: {unlink_error}")
del_span.set_data("error", str(unlink_error))
del_span.set_tag("success", "false")
# 设置状态为错误
del_span.set_status("internal_error")
cleanup_successful = False
except Exception as cleanup_error:
print(f"清理过程出错: {cleanup_error}")
logger.error(f"清理过程出错: {cleanup_error}")
cleanup_span.set_data("error", str(cleanup_error))
cleanup_span.set_tag("success", "false")
cleanup_successful = False
print("清理资源并重启ComfyUI")
with sentry_sdk.start_span(op="system.restart", description="重启ComfyUI") as restart_span:
try:
# 停止ComfyUI
with sentry_sdk.start_span(op="system.stop", description="停止ComfyUI") as stop_span:
cmd = "comfy stop"
subprocess.run(cmd, shell=True, check=True)
time.sleep(1)
stop_span.set_tag("success", "true")
except Exception as stop_error:
print(f"停止ComfyUI失败: {stop_error}")
restart_span.set_data("stop_error", str(stop_error))
restart_span.set_tag("stop_success", "false")
try:
# 重启ComfyUI
with sentry_sdk.start_span(op="system.start", description="启动ComfyUI") as start_span:
cmd = "comfy launch --background"
subprocess.run(cmd, shell=True, check=True)
start_span.set_tag("success", "true")
except Exception as start_error:
print(f"启动ComfyUI失败: {start_error}")
restart_span.set_data("start_error", str(start_error))
restart_span.set_tag("start_success", "false")
modal.experimental.stop_fetching_inputs()
# 根据清理和重启结果设置总体清理状态
if cleanup_successful:
cleanup_span.set_status("ok")
else:
cleanup_span.set_status("internal_error")
def poll_server_health(self):
@@ -524,10 +508,10 @@ with comfyui_latentsync_1_5_image.imports():
# dummy request to check if the server is healthy
req = urllib.request.Request("http://127.0.0.1:8188/system_stats")
urllib.request.urlopen(req, timeout=5)
print("ComfyUI server is healthy")
logger.info("ComfyUI server is healthy")
except (socket.timeout, urllib.error.URLError) as e:
# if no response in 5 seconds, stop the container; Modal will schedule queued inputs on a new container
print(f"Server health check failed: {str(e)} restarting ComfyUI...")
logger.error(f"Server health check failed: {str(e)} restarting ComfyUI...")
try:
cmd = "comfy stop"
try:
@@ -541,4 +525,4 @@ with comfyui_latentsync_1_5_image.imports():
raise Exception("Failed to launch ComfyUI")
except:
modal.experimental.stop_fetching_inputs()
raise Exception("ComfyUI server is not healthy, restart failed, stopping container")
raise Exception("ComfyUI server is not healthy, restart failed, stopping container")

10
src/cluster/web.py
View File

@@ -1,10 +1,6 @@
import modal
from dotenv import dotenv_values
from BowongModalFunctions.config import WorkerConfig
config = WorkerConfig()
fastapi_image = (
modal.Image
.debian_slim(python_version="3.11")
@@ -17,12 +13,13 @@ fastapi_image = (
app = modal.App(
name="web_app",
image=fastapi_image,
secrets=[modal.Secret.from_name("aws-s3-secret", environment_name=config.modal_environment)],
include_source=False)
with fastapi_image.imports():
from BowongModalFunctions.api import web_app
from BowongModalFunctions.config import WorkerConfig
config = WorkerConfig()
@app.function(scaledown_window=60,
secrets=[
@@ -34,7 +31,8 @@ with fastapi_image.imports():
secret=modal.Secret.from_name("aws-s3-secret",
environment_name=config.modal_environment),
),
}, )
},
)
@modal.concurrent(max_inputs=100)
@modal.asgi_app()
def fastapi_webapp():