From 1adfeed1ab774d2da83f9f3da5f09f04db89a8e6 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 19 Apr 2026 02:18:42 +0800 Subject: [PATCH] fix: dispatch detect start to mainland nodes --- domain-api/app/api/routes/detect.py | 201 +++++++++++++++++- domain-api/app/node_agent.py | 3 + .../app/services/ops_action_executor_core.py | 16 ++ domain-api/app/services/ops_job_service.py | 2 + .../services/ops_runtime_executor_service.py | 41 ++++ .../app/services/runtime_control_service.py | 39 +++- 6 files changed, 297 insertions(+), 5 deletions(-) diff --git a/domain-api/app/api/routes/detect.py b/domain-api/app/api/routes/detect.py index 6e260cb..e137450 100644 --- a/domain-api/app/api/routes/detect.py +++ b/domain-api/app/api/routes/detect.py @@ -15,6 +15,7 @@ from app.services.detect_job_service import ( ) from app.services.detect_service import get_detect_status from app.services.detect_run_service import create_detect_run_snapshot, finalize_detect_run, mark_detect_run_stopping +from app.services.ops_job_service import create_ops_job, list_managed_nodes from app.services.settings_service import get_settings_payload, resolve_thread_count from app.services.worker_control_service import send_worker_command, start_worker @@ -68,6 +69,172 @@ def _build_settings_summary(settings_payload: dict) -> dict: } +def _mainland_detect_targets() -> dict[str, list[dict]]: + controllers: list[dict] = [] + workers: list[dict] = [] + for node in list_managed_nodes(): + if not bool(node.get("is_enabled", True)): + continue + if str(node.get("region") or "").strip() != "mainland": + continue + if not str(node.get("last_seen_at") or "").strip(): + continue + role = str(node.get("role") or "").strip() + if role == "control": + controllers.append(node) + elif role == "worker": + workers.append(node) + return {"controllers": controllers, "workers": workers} + + +def _queue_remote_detect_job( + *, + node_code: str, + action: str, + job_summary: dict, + cycle_token: str, + requested_by: str = "api", + payload: dict | None = None, +) -> dict: + ok, message, data = create_ops_job( + { + "action": action, + "target_node_code": node_code, + "execution_mode": "remote-agent", + "requested_by": requested_by, + "auto_approve": True, + "run_now": False, + "payload": { + "job_id": int(job_summary.get("job_id") or 0), + "job_code": str(job_summary.get("job_code") or "").strip(), + "cycle_token": cycle_token, + **dict(payload or {}), + }, + "metadata": { + "source": "detect.start", + "job_id": int(job_summary.get("job_id") or 0), + "job_code": str(job_summary.get("job_code") or "").strip(), + "cycle_token": cycle_token, + }, + } + ) + return { + "node_code": node_code, + "action": action, + "ok": ok, + "message": message, + "job": dict((data or {}).get("job") or {}), + } + + +def _dispatch_remote_detect_start(*, job_summary: dict, cycle_token: str) -> dict: + targets = _mainland_detect_targets() + queued: list[dict] = [] + + for node in targets["controllers"]: + node_code = str(node.get("node_code") or "").strip() + if not node_code: + continue + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.start_sync_agent", + job_summary=job_summary, + cycle_token=cycle_token, + ) + ) + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.pull_tasks", + job_summary=job_summary, + cycle_token=cycle_token, + payload={"limit": int(job_summary.get("items_pending") or job_summary.get("items_total") or 0) or 1000}, + ) + ) + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.start_worker", + job_summary=job_summary, + cycle_token=cycle_token, + ) + ) + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.start_detection", + job_summary=job_summary, + cycle_token=cycle_token, + ) + ) + + for node in targets["workers"]: + node_code = str(node.get("node_code") or "").strip() + if not node_code: + continue + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.start_worker", + job_summary=job_summary, + cycle_token=cycle_token, + ) + ) + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.start_detection", + job_summary=job_summary, + cycle_token=cycle_token, + ) + ) + + success_jobs = [item for item in queued if item.get("ok")] + failed_jobs = [item for item in queued if not item.get("ok")] + return { + "target_summary": { + "controller_nodes": [str(item.get("node_code") or "") for item in targets["controllers"]], + "worker_nodes": [str(item.get("node_code") or "") for item in targets["workers"]], + }, + "queued_jobs": queued, + "queued_total": len(success_jobs), + "failed_total": len(failed_jobs), + } + + +def _dispatch_remote_detect_stop(*, active_job: dict | None, cycle_token: str = "") -> dict: + targets = _mainland_detect_targets() + job_summary = active_job or {} + queued: list[dict] = [] + + for node in [*targets["controllers"], *targets["workers"]]: + node_code = str(node.get("node_code") or "").strip() + if not node_code: + continue + queued.append( + _queue_remote_detect_job( + node_code=node_code, + action="runtime.stop_detection", + job_summary=job_summary, + cycle_token=cycle_token, + payload={}, + ) + ) + + success_jobs = [item for item in queued if item.get("ok")] + failed_jobs = [item for item in queued if not item.get("ok")] + return { + "target_summary": { + "controller_nodes": [str(item.get("node_code") or "") for item in targets["controllers"]], + "worker_nodes": [str(item.get("node_code") or "") for item in targets["workers"]], + }, + "queued_jobs": queued, + "queued_total": len(success_jobs), + "failed_total": len(failed_jobs), + } + + @router.get("/detect/status", response_model=ApiResponse) def detect_status() -> ApiResponse: return ApiResponse(data=get_detect_status()) @@ -165,6 +332,7 @@ def start_detect() -> ApiResponse: settings_payload = get_settings_payload() settings_summary = _build_settings_summary(settings_payload) if command_ok: + remote_dispatch = _dispatch_remote_detect_start(job_summary=job_summary, cycle_token=cycle_token) create_detect_run_snapshot( message=f"{message}；{command_message}", runtime={ @@ -177,12 +345,28 @@ def start_detect() -> ApiResponse: progress=snapshot.get("progress", {}), settings_summary=settings_summary, ) + append_detect_job_event( + job_summary["job_id"], + event_type="job_dispatch_remote_queued", + level="info" if int(remote_dispatch.get("failed_total", 0) or 0) == 0 else "warning", + message=( + f"已向大陆节点排队 {int(remote_dispatch.get('queued_total', 0) or 0)} 个远端检测动作" + if int(remote_dispatch.get("queued_total", 0) or 0) > 0 + else "当前没有可排队的大陆远端检测动作" + ), + payload={ + "cycle_token": cycle_token, + "remote_dispatch": remote_dispatch, + }, + ) + else: + remote_dispatch = {"queued_jobs": [], "queued_total": 0, "failed_total": 0, "target_summary": {"controller_nodes": [], "worker_nodes": []}} response_message = f"{message}；{command_message}" if command_ok else command_message result = _build_detect_action_result( action="start", ok=command_ok, message=response_message, - data={"job": job_summary}, + data={"job": job_summary, "remote_dispatch": remote_dispatch}, ) return ApiResponse(code=0 if command_ok else 1, message=response_message, data=result) @@ -191,6 +375,8 @@ def start_detect() -> ApiResponse: def stop_detect() -> ApiResponse: active_job = get_active_detect_job_summary(event_limit=10) ok, message = send_worker_command("stop_detection") + cycle_token = str((active_job or {}).get("current_cycle_token") or "").strip() + remote_dispatch = _dispatch_remote_detect_stop(active_job=active_job, cycle_token=cycle_token) if active_job: append_detect_job_event( active_job["job_id"], @@ -199,6 +385,17 @@ def stop_detect() -> ApiResponse: message=message, payload={"cycle_token": active_job.get("current_cycle_token", "")}, ) + append_detect_job_event( + active_job["job_id"], + event_type="job_stop_remote_queued", + level="info" if int(remote_dispatch.get("failed_total", 0) or 0) == 0 else "warning", + message=( + f"已向大陆节点排队 {int(remote_dispatch.get('queued_total', 0) or 0)} 个停止检测动作" + if int(remote_dispatch.get("queued_total", 0) or 0) > 0 + else "当前没有可排队的大陆停止检测动作" + ), + payload={"cycle_token": cycle_token, "remote_dispatch": remote_dispatch}, + ) snapshot = get_detect_status() settings_payload = get_settings_payload() settings_summary = _build_settings_summary(settings_payload) @@ -228,5 +425,5 @@ def stop_detect() -> ApiResponse: settings_summary=settings_summary, active_job=active_job, ) - result = _build_detect_action_result(action="stop", ok=ok, message=message) + result = _build_detect_action_result(action="stop", ok=ok, message=message, data={"remote_dispatch": remote_dispatch}) return ApiResponse(code=0 if ok else 1, message=message, data=result) diff --git a/domain-api/app/node_agent.py b/domain-api/app/node_agent.py index c707d45..cd84ce6 100644 --- a/domain-api/app/node_agent.py +++ b/domain-api/app/node_agent.py @@ -92,6 +92,9 @@ AGENT_CAPABILITIES = _json_env( "service.status", "runtime.start_worker", "runtime.stop_worker", + "runtime.start_detection", + "runtime.stop_detection", + "runtime.pull_tasks", "runtime.restart_api", "runtime.start_sync_agent", "runtime.stop_sync_agent", diff --git a/domain-api/app/services/ops_action_executor_core.py b/domain-api/app/services/ops_action_executor_core.py index 781e903..c615d88 100644 --- a/domain-api/app/services/ops_action_executor_core.py +++ b/domain-api/app/services/ops_action_executor_core.py @@ -16,6 +16,9 @@ STRUCTURED_ACTIONS = { "diagnostics.collect", "runtime.start_worker", "runtime.stop_worker", + "runtime.start_detection", + "runtime.stop_detection", + "runtime.pull_tasks", "runtime.restart_api", "runtime.start_sync_agent", "runtime.stop_sync_agent", @@ -186,6 +189,19 @@ def execute_structured_action( result.update(host_details) return int(code or 0) == 0, stdout or stderr or f"{service_name} status collected", result + if normalized_action in {"runtime.start_detection", "runtime.stop_detection", "runtime.pull_tasks"}: + from app.services.runtime_control_service import runtime_action + + runtime_action_name = normalized_action.split(".", 1)[1] + ok, message, result = runtime_action(runtime_action_name, normalized_payload) + result = { + **dict(result or {}), + "runtime_action": runtime_action_name, + "payload": normalized_payload, + } + result.update(host_details) + return ok, message, result + systemctl_action = systemctl_action_name(normalized_action) if systemctl_action: service_name = service_name_for_action(normalized_action, normalized_payload, normalized_service_names) diff --git a/domain-api/app/services/ops_job_service.py b/domain-api/app/services/ops_job_service.py index 104ae47..d2e9295 100644 --- a/domain-api/app/services/ops_job_service.py +++ b/domain-api/app/services/ops_job_service.py @@ -97,6 +97,8 @@ ALTER TABLE ops_jobs ADD COLUMN IF NOT EXISTS metadata_json JSONB; _LOCAL_RUNTIME_ACTIONS = { "runtime.start_worker": "start_worker", "runtime.stop_worker": "stop_worker", + "runtime.start_detection": "start_detection", + "runtime.stop_detection": "stop_detection", "runtime.restart_api": "restart_api", "runtime.start_sync_agent": "start_sync_agent", "runtime.stop_sync_agent": "stop_sync_agent", diff --git a/domain-api/app/services/ops_runtime_executor_service.py b/domain-api/app/services/ops_runtime_executor_service.py index ac84a4c..a8ccea4 100644 --- a/domain-api/app/services/ops_runtime_executor_service.py +++ b/domain-api/app/services/ops_runtime_executor_service.py @@ -25,6 +25,9 @@ _SSH_REMOTE_TIMEOUT_SECONDS = { "diagnostics.collect": 90, "runtime.start_worker": 45, "runtime.stop_worker": 45, + "runtime.start_detection": 45, + "runtime.stop_detection": 45, + "runtime.pull_tasks": 60, "runtime.restart_api": 45, "runtime.start_sync_agent": 45, "runtime.stop_sync_agent": 45, @@ -231,6 +234,18 @@ def candidate_runtime_logs_dirs(): return dirs +def runtime_module_roots(): + roots = [] + for candidate in ( + Path.cwd() / "domain-api", + Path("/opt/domaincheck/domain-api"), + Path("/www/wwwroot/getDomain/domain-api"), + ): + if candidate not in roots: + roots.append(candidate) + return roots + + def service_log_file_candidates(service_name): normalized_service_name = str(service_name or "").strip() if not normalized_service_name: @@ -297,6 +312,29 @@ def systemctl_action_name(action): return "" +def execute_runtime_action(action, payload): + runtime_action_name = str(action or "").strip().split(".", 1)[-1] + for root in runtime_module_roots(): + app_dir = root / "app" + if not app_dir.exists(): + continue + root_text = str(root) + if root_text not in sys.path: + sys.path.insert(0, root_text) + try: + from app.services.runtime_control_service import runtime_action + except Exception: + continue + ok, message, result = runtime_action(runtime_action_name, dict(payload or {{}})) + normalized_result = dict(result or {{}}) + normalized_result["runtime_action"] = runtime_action_name + normalized_result["payload"] = dict(payload or {{}}) + normalized_result["executor"] = "ssh" + normalized_result["runtime_root"] = root_text + return ok, message, normalized_result + return False, f"runtime action import failed: {{runtime_action_name}}", {{"action": action, "executor": "ssh"}} + + def execute(action, payload): if action == "health.snapshot": checks = {{}} @@ -320,6 +358,9 @@ def execute(action, payload): }} return int(code or 0) == 0, stdout or stderr or f"{{service_name}} status collected", result + if action in ("runtime.start_detection", "runtime.stop_detection", "runtime.pull_tasks"): + return execute_runtime_action(action, payload) + systemctl_action = systemctl_action_name(action) if systemctl_action: service_name = service_name_for_action(action, payload) diff --git a/domain-api/app/services/runtime_control_service.py b/domain-api/app/services/runtime_control_service.py index 6f1faf7..5f7d7ef 100644 --- a/domain-api/app/services/runtime_control_service.py +++ b/domain-api/app/services/runtime_control_service.py @@ -8,7 +8,7 @@ from app.core.config import settings from app.services.debug_event_service import push_debug_event from app.services.sync_push_service import pull_detect_task_batch_now, push_runtime_projection_now from app.services.runtime_settings_service import get_runtime_settings -from app.services.worker_control_service import _run_systemctl, normalize_systemctl_error, start_worker, stop_worker +from app.services.worker_control_service import _run_systemctl, normalize_systemctl_error, send_worker_command, start_worker, stop_worker def _workspace_root() -> Path: @@ -134,8 +134,9 @@ def stop_sync_agent() -> tuple[bool, str]: return _run_systemd_action(service_name, "stop") -def runtime_action(action: str) -> tuple[bool, str, dict]: +def runtime_action(action: str, payload: dict | None = None) -> tuple[bool, str, dict]: normalized_action = str(action or "").strip().lower().replace("-", "_") + normalized_payload = dict(payload or {}) if normalized_action != "push_debug_probe": _emit_runtime_action_event( normalized_action, @@ -174,10 +175,42 @@ def runtime_action(action: str) -> tuple[bool, str, dict]: _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "pull_tasks": - ok, message, data = pull_detect_task_batch_now() + pull_limit = normalized_payload.get("limit") + ok, message, data = pull_detect_task_batch_now(limit=int(pull_limit or 0) or None) result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message, data=data) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result + if normalized_action == "start_detection": + command_ok, command_message = send_worker_command( + "start_detection", + payload={ + key: value + for key, value in normalized_payload.items() + if value not in (None, "") + }, + ) + result = _build_runtime_action_result( + action=normalized_action, + poll_after_seconds=2, + refresh_runtime=True, + ok=command_ok, + message=command_message, + data={"payload": normalized_payload}, + ) + _emit_runtime_action_event(normalized_action, stage="finished", ok=command_ok, message=command_message, data=result) + return command_ok, command_message, result + if normalized_action == "stop_detection": + command_ok, command_message = send_worker_command("stop_detection") + result = _build_runtime_action_result( + action=normalized_action, + poll_after_seconds=2, + refresh_runtime=True, + ok=command_ok, + message=command_message, + data={}, + ) + _emit_runtime_action_event(normalized_action, stage="finished", ok=command_ok, message=command_message, data=result) + return command_ok, command_message, result if normalized_action == "push_debug_probe": ok, message, data = push_debug_event( service="domain-api",