debug(lipsync): v2 deep diag - broker/queue/DB/test-apply
Debug Lipsync SSH Diag / SSH Staging Diagnostics (push) Successful in 59s
Debug Lipsync SSH Diag / SSH Staging Diagnostics (push) Successful in 59s
This commit is contained in:
xiaoxia-agent
@@ -38,6 +38,6 @@ jobs:
|
||||
set -euo pipefail
|
||||
H=47.98.113.167; P=22222
|
||||
scp -P $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no \
|
||||
infra/scripts/server-lipsync-diag.sh root@$H:/tmp/server-lipsync-diag.sh
|
||||
infra/scripts/server-lipsync-diag-v2.sh root@$H:/tmp/server-lipsync-diag-v2.sh
|
||||
ssh -p $P -i ~/.ssh/id_rsa -o StrictHostKeyChecking=no root@$H \
|
||||
"bash /tmp/server-lipsync-diag.sh 2>&1"
|
||||
"bash /tmp/server-lipsync-diag-v2.sh 2>&1"
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
#!/bin/bash
|
||||
# Lipsync 深度诊断脚本 v2 - 重点排查broker连通/消息投递/DB状态
|
||||
set +e
|
||||
echo "#########################################################"
|
||||
echo "# Lipsync 深度诊断 v2"
|
||||
echo "# Date: $(date)"
|
||||
echo "#########################################################"
|
||||
|
||||
W=$(docker ps --format '{{.Names}}' | grep -E 'worker' | head -1)
|
||||
A=$(docker ps --format '{{.Names}}' | grep -E 'api' | grep -v 'web' | head -1)
|
||||
R=$(docker ps --format '{{.Names}}' | grep -E 'redis' | head -1)
|
||||
echo "Containers: W=$W A=$A R=$R"
|
||||
|
||||
echo ""
|
||||
echo "=== 1. 容器 uptime + image tag ==="
|
||||
for c in $W $A; do
|
||||
echo "--- $c ---"
|
||||
docker inspect "$c" --format 'Image={{.Config.Image}} Created={{.Created}} Started={{.State.StartedAt}} Restarts={{.RestartCount}}'
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo "=== 2. API+Worker 的 BROKER/REDIS 地址（隐藏密码） ==="
|
||||
echo "--- API BROKER ---"
|
||||
docker exec "$A" env | grep -iE "broker|celery_broker|redis_url|backend" | sed -E 's|(:)?//[^:]+:([^@]+)@|\1//***:***@|g'
|
||||
echo "--- Worker BROKER ---"
|
||||
docker exec "$W" env | grep -iE "broker|celery_broker|redis_url|backend" | sed -E 's|(:)?//[^:]+:([^@]+)@|\1//***:***@|g'
|
||||
|
||||
echo ""
|
||||
echo "=== 3. API容器 完整日志（近1000行） lipsync/TTS 相关 ==="
|
||||
docker logs --tail=1000 "$A" 2>&1 | grep -iE "lipsync|tts_synth|tts_processing|celery.*task|apply_async|NotRegistered|OperationalError|ConnectionError|error.*submit|traceback|submitted" | tail -200
|
||||
|
||||
echo ""
|
||||
echo "=== 4. API 容器最近的 ERROR/Exception ==="
|
||||
docker logs --tail=2000 "$A" 2>&1 | grep -iE "error|exception|traceback|critical" | grep -v "health\|/health" | tail -80
|
||||
|
||||
echo ""
|
||||
echo "=== 5. 测试1：在API容器内实际投递一条测试消息，看Worker是否消费 ==="
|
||||
# 投递后立即检查队列和Worker日志
|
||||
docker exec "$A" python - <<'PYEOF'
|
||||
import sys, time, traceback
|
||||
try:
|
||||
from app.tasks.lipsync_tts import tts_synthesize_and_submit
|
||||
from app.core.celery_app import celery_app as api_app
|
||||
# 检查broker连接
|
||||
conn = api_app.connection()
|
||||
conn.ensure_connection(max_retries=2)
|
||||
print("API celery broker connected:", conn.as_uri())
|
||||
# 投递到celery默认队列
|
||||
result = tts_synthesize_and_submit.apply_async(
|
||||
args=["diag-test-job-id", "diag-user-id", "diag-voice", "diagnostic script text", 1.0, "neutral"],
|
||||
queue="celery",
|
||||
)
|
||||
print("APPLY_ASYNC_OK task_id:", result.id)
|
||||
print("task name:", result.name)
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
PYEOF
|
||||
echo ""
|
||||
sleep 3
|
||||
echo "--- After apply_async: queue lengths ---"
|
||||
for q in celery generation transcode; do
|
||||
echo " $q: $(docker exec "$R" redis-cli LLEN $q)"
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo "=== 6. 等8秒看Worker是否收到并消费 ==="
|
||||
sleep 8
|
||||
echo "--- Queue lengths after 8s ---"
|
||||
for q in celery generation transcode; do
|
||||
echo " $q: $(docker exec "$R" redis-cli LLEN $q)"
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo "=== 7. Worker 日志最新记录（查看是否消费了测试消息） ==="
|
||||
docker logs --since=30s "$W" 2>&1 | tail -60
|
||||
|
||||
echo ""
|
||||
echo "=== 8. Worker 内 Python 直接连接broker测试 ==="
|
||||
docker exec "$W" python - <<'PYEOF'
|
||||
import traceback
|
||||
try:
|
||||
from worker_app.celery_app import celery_app
|
||||
conn = celery_app.connection()
|
||||
conn.ensure_connection(max_retries=2)
|
||||
print("Worker celery broker connected:", conn.as_uri())
|
||||
insp = celery_app.control.inspect(timeout=3)
|
||||
reg = insp.registered()
|
||||
print("Registered from worker inspect (via broker):")
|
||||
for node, tasks in (reg or {}).items():
|
||||
has_l = any('lipsync' in t for t in tasks)
|
||||
print(f" {node}: {len(tasks)} tasks, lipsync registered: {has_l}")
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
PYEOF
|
||||
|
||||
echo ""
|
||||
echo "=== 9. DB 查询最近10条lipsync_jobs状态 ==="
|
||||
docker exec "$A" python - <<'PYEOF' 2>&1
|
||||
import os, traceback
|
||||
db_url = os.environ.get("DATABASE_URL","")
|
||||
print("DATABASE_URL prefix:", (db_url[:50]+"...") if db_url else "(empty)")
|
||||
try:
|
||||
# 通过现有代码路径
|
||||
from app.db.session import SessionLocal
|
||||
from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
|
||||
db = SessionLocal()
|
||||
jobs = db.query(LipsyncJobModel).order_by(LipsyncJobModel.created_at.desc()).limit(10).all()
|
||||
print(f"Found {len(jobs)} recent lipsync jobs:")
|
||||
for j in jobs:
|
||||
err = getattr(j, 'error_message', '') or ''
|
||||
t_id = getattr(j, 'celery_task_id', '') or ''
|
||||
print(f" id={j.id} status={j.status} mode={getattr(j,'mode','?')} "
|
||||
f"created={j.created_at} celery_task_id={t_id} "
|
||||
f"error={(err[:120]+'...') if len(err)>120 else err!r}")
|
||||
db.close()
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
PYEOF
|
||||
|
||||
echo ""
|
||||
echo "=== 10. Worker 启动完整日志（前60行）==="
|
||||
docker logs "$W" 2>&1 | head -60
|
||||
|
||||
echo ""
|
||||
echo "=== 11. lipsync_service.py 完整 create_job 相关段 ==="
|
||||
docker exec "$A" sed -n '170,240p' /app/apps/api/app/services/lipsync_service.py 2>&1
|
||||
|
||||
echo ""
|
||||
echo "=== 12. celery_app task_routes / queue 配置（API侧）==="
|
||||
docker exec "$A" python - <<'PYEOF'
|
||||
from app.core.celery_app import celery_app
|
||||
print("task_routes:", getattr(celery_app.conf, 'task_routes', None))
|
||||
print("task_default_queue:", celery_app.conf.task_default_queue)
|
||||
print("task_queues:", celery_app.conf.task_queues)
|
||||
print("broker_url prefix:", celery_app.conf.broker_url[:60])
|
||||
print("result_backend prefix:", str(celery_app.conf.result_backend)[:60] if celery_app.conf.result_backend else None)
|
||||
PYEOF
|
||||
|
||||
echo ""
|
||||
echo "=== 13. Worker celery_app queue 配置 ==="
|
||||
docker exec "$W" python - <<'PYEOF'
|
||||
from worker_app.celery_app import celery_app
|
||||
print("task_routes:", getattr(celery_app.conf, 'task_routes', None))
|
||||
print("task_default_queue:", celery_app.conf.task_default_queue)
|
||||
print("task_queues:", celery_app.conf.task_queues)
|
||||
print("include/imports count:", len(celery_app.conf.imports))
|
||||
PYEOF
|
||||
|
||||
echo ""
|
||||
echo "#########################################################"
|
||||
echo "# v2 诊断完成"
|
||||
echo "#########################################################"
|
||||
Reference in New Issue
Block a user