e793718707
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m31s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""Dedicated allowlisted Ansible execution broker process."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import signal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import structlog
|
|
|
|
from src.core.config import settings
|
|
from src.core.redis_client import close_redis_pool, init_redis_pool
|
|
from src.db.base import close_db
|
|
from src.jobs.awooop_ansible_check_mode_job import (
|
|
run_awooop_ansible_check_mode_loop,
|
|
)
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
_HEALTH_PATH = Path("/tmp/executor-broker-healthy")
|
|
_READY_PATH = Path("/tmp/executor-broker-ready")
|
|
|
|
|
|
async def _heartbeat_loop(shutdown_event: asyncio.Event) -> None:
|
|
while not shutdown_event.is_set():
|
|
_HEALTH_PATH.touch()
|
|
try:
|
|
await asyncio.wait_for(shutdown_event.wait(), timeout=15)
|
|
except TimeoutError:
|
|
pass
|
|
|
|
|
|
async def _init_optional_projection_redis() -> bool:
|
|
"""Keep the controlled executor available when the Redis read model is down."""
|
|
|
|
try:
|
|
await init_redis_pool()
|
|
return True
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"ansible_execution_broker_redis_projection_degraded",
|
|
error_type=type(exc).__name__,
|
|
controlled_executor_available=True,
|
|
durable_source="postgresql",
|
|
)
|
|
return False
|
|
|
|
|
|
async def _main() -> None:
|
|
if not settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER:
|
|
raise RuntimeError("ansible_execution_broker_disabled_by_config")
|
|
|
|
ssh_key_path = Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH)
|
|
known_hosts_path = Path(settings.AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH)
|
|
if not ssh_key_path.is_file() or not known_hosts_path.is_file():
|
|
raise RuntimeError("ansible_execution_broker_transport_not_mounted")
|
|
|
|
redis_projection_ready = await _init_optional_projection_redis()
|
|
|
|
shutdown_event = asyncio.Event()
|
|
|
|
def _shutdown_handler(signum: int, _frame: Any) -> None:
|
|
logger.info("ansible_execution_broker_shutdown_requested", signal=signum)
|
|
shutdown_event.set()
|
|
|
|
signal.signal(signal.SIGTERM, _shutdown_handler)
|
|
signal.signal(signal.SIGINT, _shutdown_handler)
|
|
|
|
_HEALTH_PATH.touch()
|
|
_READY_PATH.touch()
|
|
heartbeat_task = asyncio.create_task(
|
|
_heartbeat_loop(shutdown_event),
|
|
name="ansible_execution_broker_heartbeat",
|
|
)
|
|
execution_task = asyncio.create_task(
|
|
run_awooop_ansible_check_mode_loop(),
|
|
name="run_awooop_ansible_check_mode_loop",
|
|
)
|
|
shutdown_wait_task = asyncio.create_task(
|
|
shutdown_event.wait(),
|
|
name="ansible_execution_broker_shutdown_wait",
|
|
)
|
|
logger.info(
|
|
"ansible_execution_broker_started",
|
|
interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS,
|
|
batch_limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT,
|
|
allowed_risk_levels=(
|
|
settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS
|
|
),
|
|
worker_raw_credential_access=False,
|
|
redis_projection_ready=redis_projection_ready,
|
|
redis_projection_blocks_executor=False,
|
|
)
|
|
|
|
try:
|
|
done, _pending = await asyncio.wait(
|
|
{execution_task, shutdown_wait_task},
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
if execution_task in done and not shutdown_event.is_set():
|
|
_READY_PATH.unlink(missing_ok=True)
|
|
await execution_task
|
|
raise RuntimeError("ansible_execution_broker_loop_exited_unexpectedly")
|
|
finally:
|
|
shutdown_wait_task.cancel()
|
|
execution_task.cancel()
|
|
heartbeat_task.cancel()
|
|
for task in (shutdown_wait_task, execution_task, heartbeat_task):
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
await close_redis_pool()
|
|
await close_db()
|
|
_HEALTH_PATH.unlink(missing_ok=True)
|
|
_READY_PATH.unlink(missing_ok=True)
|
|
logger.info("ansible_execution_broker_stopped")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_main())
|