diff --git a/apps/api/src/main.py b/apps/api/src/main.py index cbb1758b5..db5853595 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -654,10 +654,14 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: from src.jobs.awooop_ansible_check_mode_job import ( run_awooop_ansible_check_mode_loop, ) - schedule_api_background_task(run_awooop_ansible_check_mode_loop()) + scheduled_task = schedule_api_background_task( + run_awooop_ansible_check_mode_loop() + ) logger.info( - "awooop_ansible_check_mode_worker_scheduled", + "awooop_ansible_check_mode_worker_schedule_evaluated", enabled=settings.ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER, + scheduled_in_api_process=scheduled_task is not None, + production_process_owner="awoooi-worker", interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS, ) except Exception as e: diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index c1b66ba0d..77f06f0d3 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -390,6 +390,10 @@ async def _run_ansible_command(spec: AnsibleCommandSpec, *, timeout_seconds: int timed_out = True process.kill() stdout_bytes, stderr_bytes = await process.communicate() + except asyncio.CancelledError: + process.kill() + await process.communicate() + raise duration_ms = int((time.monotonic() - started) * 1000) return AnsibleRunResult( returncode=124 if timed_out else int(process.returncode or 0), diff --git a/apps/api/src/workers/signal_worker.py b/apps/api/src/workers/signal_worker.py index ad3136cc0..78697d803 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_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 ( + run_awooop_ansible_check_mode_loop, + ) + + ansible_check_mode_task = asyncio.create_task( + run_awooop_ansible_check_mode_loop(), + name="run_awooop_ansible_check_mode_loop", + ) + logger.info( + "signal_worker_ansible_check_mode_loop_started", + interval_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS, + batch_limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, + ) # Setup graceful shutdown shutdown_event = asyncio.Event() @@ -634,6 +649,12 @@ async def _main() -> None: # Graceful shutdown logger.info("signal_worker_shutting_down") + if ansible_check_mode_task is not None: + ansible_check_mode_task.cancel() + try: + await ansible_check_mode_task + except asyncio.CancelledError: + pass await worker.stop() await close_worker_redis_pool() # 關閉 Worker 專屬連線 await close_redis_pool() diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 2fa3b3bfa..17616f4ab 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -23,6 +23,7 @@ from src.services.awooop_ansible_check_mode_service import ( _post_apply_verification_result, _record_auto_repair_execution_receipt, _record_learning_writeback_receipt, + _run_ansible_command, _send_controlled_apply_telegram_receipt, backfill_missing_auto_repair_execution_receipts_once, build_ansible_apply_command, @@ -1785,6 +1786,14 @@ def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> N assert '"automation_run_id": str(' in apply_source +def test_ansible_subprocess_is_terminated_when_worker_task_is_cancelled() -> None: + source = inspect.getsource(_run_ansible_command) + + assert "except asyncio.CancelledError" in source + assert "process.kill()" in source + assert "await process.communicate()" in source + + def test_ansible_post_apply_verifier_helpers_are_deterministic() -> None: assert _post_apply_km_path_type("03ca6836-1b76-4da2-8e3e-6d3b6df9254a") == ( "ansible_apply_receipt:03ca6836" diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py new file mode 100644 index 000000000..f08234b9c --- /dev/null +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import inspect +from pathlib import Path + +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_check_mode_loop" in source + assert "asyncio.create_task" in source + assert "signal_worker_ansible_check_mode_loop_started" in source + assert "ansible_check_mode_task.cancel()" in source + assert "await ansible_check_mode_task" in source + + +def test_worker_manifest_mounts_existing_ssh_mcp_transport() -> None: + repo_root = Path(__file__).resolve().parents[3] + manifest = (repo_root / "k8s/awoooi-prod/08-deployment-worker.yaml").read_text() + + assert 'command: ["python", "-m", "src.workers.signal_worker"]' in manifest + assert "serviceAccountName: awoooi-executor" 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 + assert "mountPath: /run/secrets/ssh_mcp_key" in manifest + assert "mountPath: /etc/ssh-mcp/known_hosts" in manifest + assert "secretName: ssh-mcp-key" in manifest + assert "optional: true" in manifest + + +def test_api_log_does_not_claim_skipped_executor_was_scheduled() -> None: + repo_root = Path(__file__).resolve().parents[3] + main_source = (repo_root / "apps/api/src/main.py").read_text() + + 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 diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml index 8fc97f45d..a02463f01 100644 --- a/k8s/awoooi-prod/08-deployment-worker.yaml +++ b/k8s/awoooi-prod/08-deployment-worker.yaml @@ -91,6 +91,35 @@ spec: value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" + - name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER + value: "true" + - name: ENABLE_AWOOOP_ANSIBLE_CONTROLLED_APPLY + value: "true" + - name: AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS + value: "low,medium,high" + - name: AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS + value: "300" + - name: AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT + value: "1" + - name: AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS + value: "180" + - name: AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS + value: "30" + - name: AWOOOP_ANSIBLE_CHECK_MODE_TRANSPORT_PROFILE + value: "ssh_mcp" + - name: AWOOOP_ANSIBLE_CHECK_MODE_SSH_KEY_PATH + value: "/run/secrets/ssh_mcp_key" + - name: AWOOOP_ANSIBLE_CHECK_MODE_KNOWN_HOSTS_PATH + value: "/etc/ssh-mcp/known_hosts" + volumeMounts: + - name: ssh-mcp-key + mountPath: /run/secrets/ssh_mcp_key + subPath: ssh_mcp_key + readOnly: true + - name: ssh-mcp-key + mountPath: /etc/ssh-mcp/known_hosts + subPath: known_hosts + readOnly: true resources: requests: cpu: "100m" @@ -133,6 +162,12 @@ spec: periodSeconds: 5 timeoutSeconds: 5 failureThreshold: 12 + volumes: + - name: ssh-mcp-key + secret: + secretName: ssh-mcp-key + defaultMode: 0400 + optional: true # 反親和性 - 分散到不同節點 affinity: podAntiAffinity: