fix(worker): add standalone entry point for K8s deployment
- 新增 __main__ 入口點 - 寫入 health files for K8s probes - Graceful shutdown 處理 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -292,3 +292,78 @@ def get_signal_worker() -> SignalWorker:
|
||||
"Signal worker not initialized. Call init_signal_worker() first."
|
||||
)
|
||||
return _signal_worker
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Standalone Entry Point (for K8s Worker Deployment)
|
||||
# =============================================================================
|
||||
|
||||
async def _write_health_files() -> None:
|
||||
"""
|
||||
Write health check files for K8s probes.
|
||||
"""
|
||||
from pathlib import Path
|
||||
|
||||
Path("/tmp/worker-healthy").touch()
|
||||
Path("/tmp/worker-ready").touch()
|
||||
logger.info("health_files_written")
|
||||
|
||||
|
||||
async def _main() -> None:
|
||||
"""
|
||||
Standalone worker main function.
|
||||
|
||||
用於 K8s Deployment 直接執行:
|
||||
python -m src.workers.signal_worker
|
||||
"""
|
||||
import signal
|
||||
import sys
|
||||
|
||||
# Initialize settings first (loads env vars)
|
||||
from src.core.config import settings # noqa: F401
|
||||
from src.core.redis_client import init_redis, close_redis
|
||||
|
||||
logger.info(
|
||||
"signal_worker_standalone_starting",
|
||||
environment=settings.ENVIRONMENT,
|
||||
redis_url=settings.REDIS_URL.split("@")[-1] if settings.REDIS_URL else "N/A",
|
||||
)
|
||||
|
||||
# Initialize Redis
|
||||
await init_redis()
|
||||
|
||||
# Write health files for K8s probes
|
||||
await _write_health_files()
|
||||
|
||||
# Initialize and start worker
|
||||
worker = await init_signal_worker()
|
||||
|
||||
# Setup graceful shutdown
|
||||
shutdown_event = asyncio.Event()
|
||||
|
||||
def _shutdown_handler(signum: int, frame: object) -> None:
|
||||
logger.info("shutdown_signal_received", signal=signum)
|
||||
shutdown_event.set()
|
||||
|
||||
signal.signal(signal.SIGTERM, _shutdown_handler)
|
||||
signal.signal(signal.SIGINT, _shutdown_handler)
|
||||
|
||||
# Wait for shutdown signal
|
||||
await shutdown_event.wait()
|
||||
|
||||
# Graceful shutdown
|
||||
logger.info("signal_worker_shutting_down")
|
||||
await worker.stop()
|
||||
await close_redis()
|
||||
|
||||
# Remove health files
|
||||
from pathlib import Path
|
||||
Path("/tmp/worker-healthy").unlink(missing_ok=True)
|
||||
Path("/tmp/worker-ready").unlink(missing_ok=True)
|
||||
|
||||
logger.info("signal_worker_shutdown_complete")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(_main())
|
||||
|
||||
Reference in New Issue
Block a user