diff --git a/apps/api/src/main.py b/apps/api/src/main.py index db5853595..316abe65a 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -638,10 +638,14 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_candidate_backfill_job import ( run_awooop_ansible_candidate_backfill_loop, ) - schedule_api_background_task(run_awooop_ansible_candidate_backfill_loop()) + scheduled_task = schedule_api_background_task( + run_awooop_ansible_candidate_backfill_loop() + ) logger.info( - "awooop_ansible_candidate_backfill_worker_scheduled", + "awooop_ansible_candidate_backfill_worker_schedule_evaluated", enabled=settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER, + scheduled_in_api_process=scheduled_task is not None, + production_process_owner="awoooi-worker", interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS, ) except Exception as e: diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index 78697d803..7ff1df05b 100644 --- a/apps/api/src/workers/signal_worker.py +++ b/apps/api/src/workers/signal_worker.py @@ -608,6 +608,21 @@ async def _main() -> None: # Initialize and start worker worker = await init_signal_worker() + ansible_candidate_backfill_task: asyncio.Task[Any] | None = None + if settings.ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER: + from src.jobs.awooop_ansible_candidate_backfill_job import ( + run_awooop_ansible_candidate_backfill_loop, + ) + + ansible_candidate_backfill_task = asyncio.create_task( + run_awooop_ansible_candidate_backfill_loop(), + name="run_awooop_ansible_candidate_backfill_loop", + ) + logger.info( + "signal_worker_ansible_candidate_backfill_loop_started", + interval_seconds=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS, + batch_limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT, + ) ansible_check_mode_task: asyncio.Task[Any] | None = None if settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER: from src.jobs.awooop_ansible_check_mode_job import ( @@ -649,6 +664,12 @@ async def _main() -> None: # Graceful shutdown logger.info("signal_worker_shutting_down") + if ansible_candidate_backfill_task is not None: + ansible_candidate_backfill_task.cancel() + try: + await ansible_candidate_backfill_task + except asyncio.CancelledError: + pass if ansible_check_mode_task is not None: ansible_check_mode_task.cancel() try: diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 1a18c20d0..d56df2af0 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -9,6 +9,10 @@ from src.workers import signal_worker def test_signal_worker_owns_ansible_executor_loop() -> None: source = inspect.getsource(signal_worker._main) + assert "run_awooop_ansible_candidate_backfill_loop" in source + assert "signal_worker_ansible_candidate_backfill_loop_started" in source + assert "ansible_candidate_backfill_task.cancel()" in source + assert "await ansible_candidate_backfill_task" in source assert "run_awooop_ansible_check_mode_loop" in source assert "asyncio.create_task" in source assert "signal_worker_ansible_check_mode_loop_started" in source @@ -23,6 +27,8 @@ def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None: assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest assert "serviceAccountName: awoooi-executor" in manifest assert "fsGroup: 1000" in manifest + assert "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER" in manifest + assert "AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER" in manifest assert "ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY" in manifest assert "AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS" in manifest @@ -39,3 +45,4 @@ def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None: assert "awooop_ansible_check_mode_worker_schedule_evaluated" in main_source assert "scheduled_in_api_process=scheduled_task is not None" in main_source assert 'production_process_owner="awoooi-worker"' in main_source + assert "awooop_ansible_candidate_backfill_worker_schedule_evaluated" in main_source diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index ab14b6acc..cac10f36a 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -93,6 +93,16 @@ spec: value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" + - name: ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER + value: "true" + - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS + value: "600" + - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT + value: "2" + - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS + value: "24" + - name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS + value: "10" - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER value: "true" - name: ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY