From 0e096947fc1f3cfefb7cd29547887dad3acadbb8 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 17 Apr 2026 15:13:46 +0800 Subject: [PATCH] debug --- ...mainCheck_CentOS9一键复制部署与更新文档.md | 17 ++ domain-api/app/api/routes/detect.py | 86 +++++---- domain-api/app/core/files.py | 29 ++- .../app/services/cluster_runtime_service.py | 32 +++- domain-api/app/services/dashboard.py | 15 +- domain-api/app/services/detect_service.py | 87 +++++++-- .../app/services/runtime_control_service.py | 79 ++++---- .../app/services/runtime_settings_service.py | 36 +++- .../app/services/runtime_status_service.py | 46 ++++- domain-api/app/services/settings_service.py | 14 +- domain-api/app/services/sync_push_service.py | 70 ++++++-- .../app/services/sync_record_service.py | 3 +- .../app/services/worker_control_service.py | 26 ++- domain-web/src/layouts/MainLayout.vue | 17 +- .../src/views/dashboard/DashboardView.vue | 10 +- domain-web/src/views/detect/DetectView.vue | 103 +++++++++-- domain-web/src/views/runtime/RuntimeView.vue | 72 +++++--- .../src/views/settings/SettingsView.vue | 31 +++- domainCheck/detect_worker.py | 169 ++++++++++++++++++ 19 files changed, 791 insertions(+), 151 deletions(-) diff --git a/docs/18_domainCheck_CentOS9一键复制部署与更新文档.md b/docs/18_domainCheck_CentOS9一键复制部署与更新文档.md index d840945..b83b51b 100644 --- a/docs/18_domainCheck_CentOS9一键复制部署与更新文档.md +++ b/docs/18_domainCheck_CentOS9一键复制部署与更新文档.md @@ -195,6 +195,23 @@ ln -sfn /www/wwwroot/getDomain/domainCheck /opt/domaincheck/domainCheck - 默认检测并发由后台系统设置里的 `默认检测线程数` 控制 - 若某台机器需要单独并发，可在后台系统设置里增加 `节点独立线程覆盖` +- 远端日志回传由后台 `系统设置 -> 运行配置` 或 `检测控制` 顶部按钮控制 + - `关闭回传` + - 不追加远端 Worker 日志镜像 + - 适合正式长期运行，减少额外事件写入 + - `关键回传` + - 只回传启动、停止、代理刷新、批次完成、异常、完成等关键过程 + - 适合大多数联调和线上观察 + - `全量回传` + - 在关键模式基础上，额外回传取任务、建线程、线程数变化、域名开始/完成/失败等过程 + - 适合临时排查问题，定位完成后建议切回 `关键` 或 `关闭` +- 检测控制页日志窗口现在会合并展示： + - controller 本机可见的 Worker 日志尾部 + - 远端 Worker 回传的流程事件日志 +- 如果开启了远端日志回传但页面仍无新增输出，优先检查： + - `domaincheck-worker` 是否已更新到最新代码 + - Worker 是否能连上 Redis + - 检测任务是否由当前控制面发起，并带有 `job_id / cycle_token` - 规则是： - 节点已单独配置：走该节点自己的线程数 - 节点未单独配置：回退到默认线程数 diff --git a/domain-api/app/api/routes/detect.py b/domain-api/app/api/routes/detect.py index 6d6ddfc..d6269ac 100644 --- a/domain-api/app/api/routes/detect.py +++ b/domain-api/app/api/routes/detect.py @@ -19,6 +19,39 @@ from app.services.worker_control_service import send_worker_command, start_worke router = APIRouter(tags=["detect"]) +def _build_detect_action_result( + *, + action: str, + ok: bool, + message: str, + poll_after_seconds: int = 2, + refresh_status: bool = True, + data: dict | None = None, +) -> dict: + result = { + "action": action, + "poll_after_seconds": poll_after_seconds, + "refresh_status": refresh_status, + **(data or {}), + } + lowered = str(message or "").strip().lower() + if "ui_level" not in result: + if not ok: + result["ui_level"] = "warning" if any(keyword in lowered for keyword in ("当前没有", "无需", "未运行", "未启动")) else "error" + elif any(keyword in str(message or "") for keyword in ("命令已发送", "已发送检测启动请求", "控制指令")): + result["ui_level"] = "warning" + else: + result["ui_level"] = "success" + if "poll_schedule_seconds" not in result: + if ok: + result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)] + elif poll_after_seconds > 0: + result["poll_schedule_seconds"] = [poll_after_seconds] + else: + result["poll_schedule_seconds"] = [] + return result + + def _build_settings_summary(settings_payload: dict) -> dict: thread_count_resolution = resolve_thread_count(settings_payload=settings_payload) return { @@ -65,15 +98,16 @@ def detect_job_detail(job_id: int) -> ApiResponse: def start_detect() -> ApiResponse: job_summary = create_detect_job_if_needed(limit=1000, created_by="api") if not job_summary: + result = _build_detect_action_result( + action="start", + ok=False, + message="当前没有可创建的检测任务", + data={"job": None}, + ) return ApiResponse( code=0, message="当前没有可创建的检测任务", - data={ - "action": "start", - "job": None, - "poll_after_seconds": 2, - "refresh_status": True, - }, + data=result, ) cycle_token = uuid4().hex[:10] append_detect_job_event( @@ -91,6 +125,12 @@ def start_detect() -> ApiResponse: ok, message = start_worker() if not ok: + result = _build_detect_action_result( + action="start", + ok=False, + message=message, + data={"job": job_summary}, + ) append_detect_job_event( job_summary["job_id"], event_type="job_dispatch_failed", @@ -101,12 +141,7 @@ def start_detect() -> ApiResponse: return ApiResponse( code=1, message=message, - data={ - "action": "start", - "job": job_summary, - "poll_after_seconds": 2, - "refresh_status": True, - }, + data=result, ) command_ok, command_message = send_worker_command( @@ -140,16 +175,14 @@ def start_detect() -> ApiResponse: progress=snapshot.get("progress", {}), settings_summary=settings_summary, ) - return ApiResponse( - code=0 if command_ok else 1, - message=f"{message}；{command_message}" if command_ok else command_message, - data={ - "action": "start", - "job": job_summary, - "poll_after_seconds": 2, - "refresh_status": True, - }, + 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}, ) + return ApiResponse(code=0 if command_ok else 1, message=response_message, data=result) @router.post("/detect/stop", response_model=ApiResponse) @@ -193,12 +226,5 @@ def stop_detect() -> ApiResponse: settings_summary=settings_summary, active_job=active_job, ) - return ApiResponse( - code=0 if ok else 1, - message=message, - data={ - "action": "stop", - "poll_after_seconds": 2, - "refresh_status": True, - }, - ) + result = _build_detect_action_result(action="stop", ok=ok, message=message) + return ApiResponse(code=0 if ok else 1, message=message, data=result) diff --git a/domain-api/app/core/files.py b/domain-api/app/core/files.py index 9045bed..b180be7 100644 --- a/domain-api/app/core/files.py +++ b/domain-api/app/core/files.py @@ -11,6 +11,31 @@ def domain_root() -> Path: return Path(settings.domain_root) +def resolve_domain_path(*relative_paths: str) -> Path | None: + root = domain_root() + candidates: list[Path] = [] + seen: set[str] = set() + for relative_path in relative_paths: + text = str(relative_path or "").strip() + if not text: + continue + candidate = root / text + candidate_key = str(candidate) + if candidate_key not in seen: + seen.add(candidate_key) + candidates.append(candidate) + if "/" not in text and "\\" not in text: + nested_candidate = root / "logs" / text + nested_key = str(nested_candidate) + if nested_key not in seen: + seen.add(nested_key) + candidates.append(nested_candidate) + for candidate in candidates: + if candidate.exists(): + return candidate + return candidates[0] if candidates else None + + def read_json(relative_path: str, default: dict | list | None = None): path = domain_root() / relative_path if not path.exists(): @@ -30,8 +55,8 @@ def write_json(relative_path: str, payload) -> None: def tail_lines(relative_path: str, max_lines: int = 120) -> list[str]: - path = domain_root() / relative_path - if not path.exists(): + path = resolve_domain_path(relative_path) + if path is None or not path.exists(): return [] with path.open("r", encoding="utf-8", errors="replace") as handle: return handle.read().splitlines()[-max_lines:] diff --git a/domain-api/app/services/cluster_runtime_service.py b/domain-api/app/services/cluster_runtime_service.py index 9a5f6ec..6bf6d71 100644 --- a/domain-api/app/services/cluster_runtime_service.py +++ b/domain-api/app/services/cluster_runtime_service.py @@ -86,6 +86,11 @@ CREATE TABLE IF NOT EXISTS detect_sync_records ( ); """ +_STALE_AFTER_SECONDS = 90 +_OFFLINE_AFTER_MINUTES = 5 +_PRUNE_IMPORTED_AFTER_MINUTES = 30 +_PRUNE_GENERAL_AFTER_HOURS = 6 + def _resolve_local_ip() -> str: try: @@ -190,6 +195,28 @@ def cleanup_imported_runtime_nodes(*, region: str, role: str, keep_node_code: st conn.commit() +def prune_expired_runtime_nodes() -> None: + imported_cutoff = datetime.now() - timedelta(minutes=_PRUNE_IMPORTED_AFTER_MINUTES) + general_cutoff = datetime.now() - timedelta(hours=_PRUNE_GENERAL_AFTER_HOURS) + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + DELETE FROM detect_worker_nodes + WHERE ( + (metadata_json->>'service') = 'runtime-ingest' + AND last_heartbeat_at < %s + ) + OR ( + COALESCE(metadata_json->>'service', '') <> 'runtime-ingest' + AND last_heartbeat_at < %s + ) + """, + (imported_cutoff, general_cutoff), + ) + conn.commit() + + def register_local_control_heartbeat() -> None: from app.services.detect_job_service import get_active_detect_job_summary from app.services.worker_control_service import detect_worker_runtime @@ -240,14 +267,15 @@ def _normalize_node_status(raw_status: str, last_heartbeat_at: datetime | None) return status now = datetime.now(last_heartbeat_at.tzinfo) if last_heartbeat_at.tzinfo else datetime.now() age = now - last_heartbeat_at - if age > timedelta(minutes=5): + if age > timedelta(minutes=_OFFLINE_AFTER_MINUTES): return "offline" - if age > timedelta(seconds=90): + if age > timedelta(seconds=_STALE_AFTER_SECONDS): return "stale" return status def get_cluster_snapshot() -> dict: + prune_expired_runtime_nodes() register_local_control_heartbeat() with get_db() as conn: with conn.cursor() as cur: diff --git a/domain-api/app/services/dashboard.py b/domain-api/app/services/dashboard.py index e3ee8fa..c0d1597 100644 --- a/domain-api/app/services/dashboard.py +++ b/domain-api/app/services/dashboard.py @@ -18,10 +18,21 @@ def fetch_overview() -> dict: with get_db() as conn: with conn.cursor() as cur: for key, query in queries.items(): - cur.execute(query) - result[key] = cur.fetchone()[0] + try: + cur.execute(query) + result[key] = cur.fetchone()[0] + except Exception: + result[key] = 0 runtime = get_runtime_status() + cluster_summary = ((runtime.get("cluster") or {}).get("summary") or {}) + online_worker_nodes = int(cluster_summary.get("online_worker_nodes", 0) or 0) + dedicated_online_worker_nodes = int(cluster_summary.get("dedicated_online_worker_nodes", 0) or 0) result["worker_status"] = "online" if runtime["worker"]["running"] else "offline" + result["local_worker_status"] = result["worker_status"] + result["cluster_worker_status"] = "online" if online_worker_nodes > 0 else "offline" + result["cluster_online_worker_nodes"] = online_worker_nodes + result["cluster_dedicated_online_worker_nodes"] = dedicated_online_worker_nodes + result["cluster_online_control_nodes"] = int(cluster_summary.get("online_control_nodes", 0) or 0) result["api_status"] = "online" result["worker_mode"] = runtime["worker"]["mode"] return result diff --git a/domain-api/app/services/detect_service.py b/domain-api/app/services/detect_service.py index c66d609..32cccb7 100644 --- a/domain-api/app/services/detect_service.py +++ b/domain-api/app/services/detect_service.py @@ -3,10 +3,8 @@ from __future__ import annotations import json import re from datetime import datetime, timezone -from pathlib import Path - from app.core.db import get_db -from app.core.files import tail_lines +from app.core.files import resolve_domain_path, tail_lines from app.core.redis_client import get_redis from app.services.runtime_settings_service import get_runtime_settings from app.services.detect_run_service import sync_detect_runs @@ -20,6 +18,7 @@ _PROXY_COUNT_RE = re.compile(r"当前可用代理数[:：]\s*(\d+)") _THREAD_COUNT_RE = re.compile(r"当前实际线程数量[:：]\s*(\d+)\s*/\s*(\d+)") _RUNTIME_STATE_KEY = "domain_tool:detect_runtime_state" _TIMESTAMP_FORMATS = ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S") +_REMOTE_LOG_MAX_CHARS = 500 def _extract_dependency_alerts(lines: list[str]) -> list[dict]: @@ -141,6 +140,61 @@ def _recent_event(lines: list[str]) -> str: return "" +def _build_remote_log_lines( + active_job: dict | None, + *, + enabled: bool, + mode: str, + limit: int = 240, +) -> list[str]: + if not enabled: + return [] + if not active_job: + return [] + events = list(active_job.get("current_cycle_events") or active_job.get("recent_events") or []) + if not events: + return [] + + lines: list[str] = [] + normalized_mode = str(mode or "key").strip().lower() + if normalized_mode not in {"key", "full"}: + normalized_mode = "key" + current_cycle_token = str(active_job.get("current_cycle_token") or "").strip() + for event in reversed(events): + event_type = str(event.get("event_type") or "").strip() + if event_type != "worker_log": + continue + created_at = str(event.get("created_at") or "").strip() + node_code = str(event.get("node_code") or "").strip() or "unknown" + message = str(event.get("message") or "").strip() + if not message: + continue + payload = event.get("payload") if isinstance(event.get("payload"), dict) else {} + event_cycle_token = str(payload.get("cycle_token") or "").strip() + if current_cycle_token and event_cycle_token and event_cycle_token != current_cycle_token: + continue + event_mode = str(payload.get("log_mode") or "key").strip().lower() + if event_mode not in {"key", "full"}: + event_mode = "key" + if normalized_mode != "full" and event_mode == "full": + continue + if len(message) > _REMOTE_LOG_MAX_CHARS: + message = f"{message[:_REMOTE_LOG_MAX_CHARS]}..." + lines.append(f"[{created_at}] [{node_code}] {message}") + return lines[-max(1, int(limit or 240)) :] + + +def _resolve_remote_log_lines( + active_job: dict | None, + runs: list[dict], + *, + enabled: bool, + mode: str, + limit: int = 240, +) -> list[str]: + return _build_remote_log_lines(active_job, enabled=enabled, mode=mode, limit=limit) + + def _load_runtime_state() -> dict: try: redis_client = get_redis() @@ -282,17 +336,18 @@ def get_detect_status() -> dict: with get_db() as conn: with conn.cursor() as cur: for key, query in queries.items(): - cur.execute(query) - progress[key] = cur.fetchone()[0] + try: + cur.execute(query) + progress[key] = cur.fetchone()[0] + except Exception: + progress[key] = 0 settings_payload = get_settings_payload() - worker_log = Path(__file__).resolve().parents[3] / "domainCheck" / "detect_worker.log" - if not worker_log.exists(): - worker_log = Path(__file__).resolve().parents[3] / "domainCheck" / "logs" / "detect_worker.log" + worker_log = resolve_domain_path("detect_worker.log", "logs/detect_worker.log") worker_online = False last_log_time = None - if worker_log.exists(): + if worker_log and worker_log.exists(): modified = datetime.fromtimestamp(worker_log.stat().st_mtime, tz=timezone.utc) last_log_time = modified.isoformat() worker_online = (datetime.now(timezone.utc) - modified).total_seconds() < 180 @@ -335,7 +390,7 @@ def get_detect_status() -> dict: "allow_direct": settings_payload["proxy_config"].get("allow_direct", False), "proxy_pool_count": len(settings_payload["proxy_config"].get("proxy_urls", [])), } - active_job = get_active_detect_job_summary() + active_job = get_active_detect_job_summary(event_limit=240) runtime_snapshot = { **runtime, "detecting": runtime_state.get("detecting", False), @@ -347,6 +402,15 @@ def get_detect_status() -> dict: "proxy_supplier_empty": proxy_runtime["supplier_empty"], } runs = sync_detect_runs(runtime_snapshot, progress, settings_summary, active_job=active_job) + worker_log_sync_enabled = bool(runtime_settings.get("worker_log_sync_enabled", False)) + worker_log_sync_mode = str(runtime_settings.get("worker_log_sync_mode", "key") or "key") + remote_log_lines = _resolve_remote_log_lines( + active_job, + runs, + enabled=worker_log_sync_enabled, + mode=worker_log_sync_mode, + limit=240, + ) dependency_alerts = _extract_dependency_alerts(recent_lines) append_detect_result_projection_if_changed( detect={ @@ -400,6 +464,9 @@ def get_detect_status() -> dict: "recent_event": runtime_state.get("detail") or _recent_event(recent_lines), "recent_warning": recent_proxy_warning, "log_lines": recent_lines, + "remote_log_lines": remote_log_lines, "runs": runs, "active_job": active_job, + "worker_log_sync_enabled": worker_log_sync_enabled, + "worker_log_sync_mode": worker_log_sync_mode, } diff --git a/domain-api/app/services/runtime_control_service.py b/domain-api/app/services/runtime_control_service.py index 393f90a..6f1faf7 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, start_worker, stop_worker +from app.services.worker_control_service import _run_systemctl, normalize_systemctl_error, start_worker, stop_worker def _workspace_root() -> Path: @@ -29,7 +29,7 @@ def restart_api() -> tuple[bool, str]: # blocking the HTTP request until uvicorn is torn down. result = _run_systemctl(["restart", api_service_name], timeout=5, no_block=True) if result.returncode != 0: - return False, (result.stderr or result.stdout or "重启 Linux API 失败").strip() + return False, normalize_systemctl_error(result.stderr or result.stdout or "重启 Linux API 失败", service_name=api_service_name) return True, f"Linux API 重启命令已发送: {api_service_name}" if os.name != "nt": @@ -59,7 +59,7 @@ def _run_systemd_action(service_name: str, action: str, *, no_block: bool = Fals systemctl_command.extend([action, service_name]) result = _run_systemctl(systemctl_command, timeout=10 if no_block else 30) if result.returncode != 0: - return False, (result.stderr or result.stdout or f"{action} {service_name} 失败").strip() + return False, normalize_systemctl_error(result.stderr or result.stdout or f"{action} {service_name} 失败", service_name=service_name) return True, f"{service_name} {action} 命令已发送" @@ -80,6 +80,42 @@ def _emit_runtime_action_event(action: str, *, stage: str, ok: bool, message: st pass +def _build_runtime_action_result( + *, + action: str, + poll_after_seconds: int, + refresh_runtime: bool, + ok: bool, + message: str, + data: dict | None = None, +) -> dict: + result = { + "action": action, + "poll_after_seconds": poll_after_seconds, + "refresh_runtime": refresh_runtime, + **(data or {}), + } + lowered = str(message or "").strip().lower() + if "ui_level" not in result: + if not ok: + result["ui_level"] = "warning" if any(keyword in lowered for keyword in ("当前没有", "无需", "未启用", "未配置")) else "error" + elif any(keyword in str(message or "") for keyword in ("命令已发送", "确认失败")): + result["ui_level"] = "warning" + else: + result["ui_level"] = "success" + + if "poll_schedule_seconds" not in result: + if action in {"start_worker", "stop_worker", "restart_api", "start_sync_agent", "stop_sync_agent"}: + result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)] + elif action in {"push_sync", "pull_tasks"}: + result["poll_schedule_seconds"] = [1, max(2, poll_after_seconds)] + elif poll_after_seconds > 0: + result["poll_schedule_seconds"] = [poll_after_seconds] + else: + result["poll_schedule_seconds"] = [] + return result + + def start_sync_agent() -> tuple[bool, str]: runtime = get_runtime_settings() worker_mode = runtime.get("worker_mode", settings.worker_mode) @@ -109,47 +145,37 @@ def runtime_action(action: str) -> tuple[bool, str, dict]: ) if normalized_action == "start_worker": ok, message = start_worker() - result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True} + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "stop_worker": ok, message = stop_worker() - result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True} + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "restart_api": ok, message = restart_api() - result = {"action": normalized_action, "poll_after_seconds": 4, "refresh_runtime": True} + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=4, refresh_runtime=True, ok=ok, message=message) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "start_sync_agent": ok, message = start_sync_agent() - result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True} + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "stop_sync_agent": ok, message = stop_sync_agent() - result = {"action": normalized_action, "poll_after_seconds": 2, "refresh_runtime": True} + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=2, refresh_runtime=True, ok=ok, message=message) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result if normalized_action == "push_sync": ok, message, data = push_runtime_projection_now() - result = { - "action": normalized_action, - "poll_after_seconds": 2, - "refresh_runtime": True, - **(data or {}), - } + 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 == "pull_tasks": ok, message, data = pull_detect_task_batch_now() - result = { - "action": normalized_action, - "poll_after_seconds": 2, - "refresh_runtime": True, - **(data or {}), - } + 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 == "push_debug_probe": @@ -160,18 +186,9 @@ def runtime_action(action: str) -> tuple[bool, str, dict]: message="manual debug probe", payload={"node_code": settings.node_code, "node_region": settings.node_region}, ) - result = { - "action": normalized_action, - "poll_after_seconds": 1, - "refresh_runtime": False, - **(data or {}), - } + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=1, refresh_runtime=False, ok=ok, message=message, data=data) _emit_runtime_action_event(normalized_action, stage="finished", ok=ok, message=message, data=result) return ok, message, result - result = { - "action": normalized_action, - "poll_after_seconds": 0, - "refresh_runtime": False, - } + result = _build_runtime_action_result(action=normalized_action, poll_after_seconds=0, refresh_runtime=False, ok=False, message=f"不支持的运行时动作: {action}") _emit_runtime_action_event(normalized_action, stage="finished", ok=False, message=f"不支持的运行时动作: {action}", data=result) return False, f"不支持的运行时动作: {action}", result diff --git a/domain-api/app/services/runtime_settings_service.py b/domain-api/app/services/runtime_settings_service.py index e21b8ed..cec4b46 100644 --- a/domain-api/app/services/runtime_settings_service.py +++ b/domain-api/app/services/runtime_settings_service.py @@ -9,21 +9,41 @@ DEFAULT_RUNTIME_SETTINGS = { "worker_service_name": settings.worker_service_name, "api_service_name": settings.api_service_name, "sync_agent_service_name": settings.sync_agent_service_name, + "worker_log_sync_enabled": False, + "worker_log_sync_mode": "key", } +def _normalize_worker_log_sync_mode(value: object) -> str: + return "full" if str(value or "").strip().lower() == "full" else "key" + + +def normalize_runtime_settings(payload: dict | None) -> dict: + merged = dict(DEFAULT_RUNTIME_SETTINGS) + if isinstance(payload, dict): + for key in DEFAULT_RUNTIME_SETTINGS: + if key in payload and payload[key] is not None: + merged[key] = payload[key] + + worker_mode = str(merged.get("worker_mode") or "").strip() + if worker_mode not in {"windows-local", "linux-systemd"}: + merged["worker_mode"] = DEFAULT_RUNTIME_SETTINGS["worker_mode"] + + for key in ("worker_service_name", "api_service_name", "sync_agent_service_name"): + value = str(merged.get(key) or "").strip() + merged[key] = value or DEFAULT_RUNTIME_SETTINGS[key] + + merged["worker_log_sync_enabled"] = bool(merged.get("worker_log_sync_enabled", False)) + merged["worker_log_sync_mode"] = _normalize_worker_log_sync_mode(merged.get("worker_log_sync_mode")) + return merged + + def get_runtime_settings() -> dict: stored = read_runtime_json("runtime_settings.json", default={}) - result = dict(DEFAULT_RUNTIME_SETTINGS) - result.update(stored or {}) - return result + return normalize_runtime_settings(stored) def update_runtime_settings(payload: dict) -> dict: - current = get_runtime_settings() - merged = dict(current) - for key in DEFAULT_RUNTIME_SETTINGS: - if key in payload and payload[key] is not None: - merged[key] = payload[key] + merged = normalize_runtime_settings({**get_runtime_settings(), **(payload or {})}) write_runtime_json("runtime_settings.json", merged) return merged diff --git a/domain-api/app/services/runtime_status_service.py b/domain-api/app/services/runtime_status_service.py index 3cffebd..0383e15 100644 --- a/domain-api/app/services/runtime_status_service.py +++ b/domain-api/app/services/runtime_status_service.py @@ -72,6 +72,19 @@ def _build_multi_region_readiness( warning_issues: list[str] = [] info_items: list[str] = [] + online_pairs = { + ( + str(node.get("region") or "").strip(), + str(node.get("role") or "").strip(), + ) + for node in nodes + if str(node.get("status") or "") in {"online", "busy"} + } + critical_stale_nodes: list[str] = [] + redundant_stale_nodes: list[str] = [] + critical_offline_nodes: list[str] = [] + redundant_offline_nodes: list[str] = [] + if online_control_nodes <= 0: blocking_issues.append("当前没有在线控制面节点，无法视为正式可用集群。") @@ -89,12 +102,33 @@ def _build_multi_region_readiness( if online_worker_nodes <= 0: warning_issues.append("当前没有在线 Worker 节点，检测任务无法在多机状态下继续推进。") - offline_nodes = list(summary.get("offline_nodes") or []) - stale_nodes = list(summary.get("stale_nodes") or []) - if stale_nodes: - warning_issues.append(f"存在失活节点: {'、'.join(stale_nodes)}") - if offline_nodes: - warning_issues.append(f"存在离线节点: {'、'.join(offline_nodes)}") + for node in nodes: + node_status = str(node.get("status") or "").strip() + if node_status not in {"stale", "offline"}: + continue + node_code = str(node.get("node_code") or "").strip() + node_region = str(node.get("region") or "").strip() + node_role = str(node.get("role") or "").strip() + metadata = node.get("metadata") or {} + replaced_by_online_peer = (node_region, node_role) in online_pairs + redundant = replaced_by_online_peer and ( + node_role == "control" + or not bool(node.get("is_effective_worker", False)) + or str(metadata.get("service") or "") == "runtime-ingest" + ) + if node_status == "stale": + (redundant_stale_nodes if redundant else critical_stale_nodes).append(node_code) + else: + (redundant_offline_nodes if redundant else critical_offline_nodes).append(node_code) + + if critical_stale_nodes: + warning_issues.append(f"存在失活节点: {'、'.join(critical_stale_nodes)}") + if critical_offline_nodes: + warning_issues.append(f"存在离线节点: {'、'.join(critical_offline_nodes)}") + if redundant_stale_nodes: + info_items.append(f"存在历史失活节点（已被在线同类节点覆盖）: {'、'.join(redundant_stale_nodes)}") + if redundant_offline_nodes: + info_items.append(f"存在历史离线节点（已被在线同类节点覆盖）: {'、'.join(redundant_offline_nodes)}") failed_batches = int(batch_states.get("failed", 0) or 0) projected_batches = int(batch_states.get("projected", 0) or 0) diff --git a/domain-api/app/services/settings_service.py b/domain-api/app/services/settings_service.py index a6302e3..4e1a43a 100644 --- a/domain-api/app/services/settings_service.py +++ b/domain-api/app/services/settings_service.py @@ -15,6 +15,7 @@ REDIS_KEYS = { "thread_count": "domain_tool:thread_count", "node_thread_counts": "domain_tool:node_thread_counts", "credentials": "domain_tool:credentials", + "runtime_settings": "domain_tool:runtime_settings", } DETECT_OPTION_KEYS = { @@ -198,7 +199,6 @@ def update_settings_payload(payload: dict) -> dict: write_json("proxy_config.json", proxy_config) write_json("thread_count.json", {"thread_count": str(thread_count)}) write_json("node_thread_counts.json", node_thread_counts) - redis_client = get_redis() try: redis_client.set(REDIS_KEYS["detect_options"], json.dumps(detect_options, ensure_ascii=False)) @@ -209,6 +209,12 @@ def update_settings_payload(payload: dict) -> dict: redis_client.publish("domain_tool:thread_count:update", str(thread_count)) redis_client.set(REDIS_KEYS["node_thread_counts"], json.dumps(node_thread_counts, ensure_ascii=False)) redis_client.publish("domain_tool:node_thread_counts:update", json.dumps(node_thread_counts, ensure_ascii=False)) + redis_client.set(REDIS_KEYS["runtime_settings"], json.dumps(runtime_settings, ensure_ascii=False)) + redis_client.publish("domain_tool:config_update", "runtime_settings") + redis_client.publish("domain_tool:config_update", "node_thread_counts") + redis_client.publish("domain_tool:config_update", "detect_options") + redis_client.publish("domain_tool:config_update", "proxy_config") + redis_client.publish("domain_tool:config_update", "thread_count") except Exception: pass @@ -278,6 +284,12 @@ def validate_settings_payload(payload: dict) -> None: for key in ("worker_service_name", "api_service_name", "sync_agent_service_name"): if key in runtime_settings and runtime_settings[key] is not None and not str(runtime_settings[key]).strip(): raise ValueError(f"{key} must not be empty") + if "worker_log_sync_enabled" in runtime_settings and not isinstance(runtime_settings["worker_log_sync_enabled"], bool): + raise ValueError("worker_log_sync_enabled must be a boolean") + if "worker_log_sync_mode" in runtime_settings: + worker_log_sync_mode = str(runtime_settings.get("worker_log_sync_mode") or "").strip().lower() + if worker_log_sync_mode not in {"key", "full"}: + raise ValueError("worker_log_sync_mode must be key or full") def backup_current_settings(reason: str = "manual") -> dict: diff --git a/domain-api/app/services/sync_push_service.py b/domain-api/app/services/sync_push_service.py index dce0534..5880178 100644 --- a/domain-api/app/services/sync_push_service.py +++ b/domain-api/app/services/sync_push_service.py @@ -908,11 +908,11 @@ def _push_projection_record(source_record: dict, sync_type: str, ingest_url: str def push_runtime_projection_now() -> tuple[bool, str, dict]: if not settings.sync_push_enabled: - return False, "未启用同步推送", {"action": "push_sync"} + return False, "未启用同步推送", {"action": "push_sync", "sync_state": "disabled", "ui_level": "warning", "poll_schedule_seconds": []} ingest_url = _ingest_url(settings.sync_target_api_base_url) if not ingest_url: - return False, "未配置同步目标地址", {"action": "push_sync"} + return False, "未配置同步目标地址", {"action": "push_sync", "sync_state": "misconfigured", "ui_level": "warning", "poll_schedule_seconds": []} results = [] ok, message, data = _push_projection_now("runtime_projection", ingest_url) @@ -922,19 +922,52 @@ def push_runtime_projection_now() -> tuple[bool, str, dict]: results.append({"sync_type": "detect_result_projection", "ok": ok, "message": message, "data": data}) success_count = sum(1 for item in results if item["ok"]) + warning_count = sum( + 1 + for item in results + if not item["ok"] and any(keyword in str(item.get("message") or "") for keyword in ("当前没有可推送", "已全部同步完成", "无需重复发送", "进行中", "等待下个重试窗口")) + ) if success_count == 0: - return False, "当前没有成功推送的同步投影", {"action": "push_sync", "results": results} - return True, f"同步推送完成，成功 {success_count}/{len(results)}", {"action": "push_sync", "results": results} + if warning_count == len(results): + return False, "当前没有需要立即推送的同步投影", { + "action": "push_sync", + "sync_state": "idle", + "ui_level": "warning", + "poll_schedule_seconds": [], + "results": results, + } + return False, "同步推送未成功，请检查明细结果", { + "action": "push_sync", + "sync_state": "failed", + "ui_level": "error", + "poll_schedule_seconds": [2], + "results": results, + } + if success_count < len(results): + return True, f"同步推送部分完成，成功 {success_count}/{len(results)}", { + "action": "push_sync", + "sync_state": "partial_success", + "ui_level": "warning", + "poll_schedule_seconds": [1, 3], + "results": results, + } + return True, f"同步推送完成，成功 {success_count}/{len(results)}", { + "action": "push_sync", + "sync_state": "success", + "ui_level": "success", + "poll_schedule_seconds": [1, 3], + "results": results, + } def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dict]: if settings.node_region != "mainland" or settings.node_role != "control": - return False, "当前节点无需拉取待检测任务批次", {"action": "pull_tasks"} + return False, "当前节点无需拉取待检测任务批次", {"action": "pull_tasks", "pull_state": "not_applicable", "ui_level": "warning", "poll_schedule_seconds": []} export_url = _task_export_url(settings.sync_target_api_base_url) ack_url = _task_ack_url(settings.sync_target_api_base_url) if not export_url or not ack_url: - return False, "未配置任务拉取目标地址", {"action": "pull_tasks"} + return False, "未配置任务拉取目标地址", {"action": "pull_tasks", "pull_state": "misconfigured", "ui_level": "warning", "poll_schedule_seconds": []} safe_limit = max(1, min(int(limit or settings.sync_batch_size or 200), max(1, int(settings.sync_batch_size or 200)))) request_url = f"{export_url}?limit={safe_limit}" @@ -948,27 +981,30 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic raw = response.read().decode("utf-8") data = json.loads(raw) if raw else {} except json.JSONDecodeError as exc: - return False, f"拉取待检测任务失败: 远端响应不是合法 JSON ({exc})", {"action": "pull_tasks"} + return False, f"拉取待检测任务失败: 远端响应不是合法 JSON ({exc})", {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2]} except urllib.error.HTTPError as exc: error_body = exc.read().decode("utf-8", errors="ignore") if hasattr(exc, "read") else "" return False, f"拉取待检测任务失败: HTTP {getattr(exc, 'code', 500)}", { "action": "pull_tasks", + "pull_state": "failed", + "ui_level": "error", + "poll_schedule_seconds": [2], "http_status": getattr(exc, "code", 500), "response_text": error_body[:500], } except Exception as exc: - return False, f"拉取待检测任务失败: {exc}", {"action": "pull_tasks"} + return False, f"拉取待检测任务失败: {exc}", {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2]} payload = data.get("data") or {} response_code = data.get("code", 0) if response_code not in (0, "0", None) and not payload: - return False, str(data.get("message") or "拉取待检测任务失败"), {"action": "pull_tasks", "response": data} + return False, str(data.get("message") or "拉取待检测任务失败"), {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2], "response": data} source_record_id = int(payload.get("source_record_id") or 0) projection_hash = str(payload.get("projection_hash") or "").strip() projection = payload.get("projection") or {} if source_record_id <= 0 or not projection_hash or not projection: - return False, "远端当前没有可拉取的待检测任务批次", {"action": "pull_tasks", "batch_size": 0} + return False, "远端当前没有可拉取的待检测任务批次", {"action": "pull_tasks", "pull_state": "idle", "ui_level": "warning", "poll_schedule_seconds": [], "batch_size": 0} ingest_ok, ingest_message, ingest_data = ingest_detect_task_projection( { @@ -981,7 +1017,7 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic shared_token=settings.sync_shared_token, ) if not ingest_ok: - return False, ingest_message, {"action": "pull_tasks", **(ingest_data or {})} + return False, ingest_message, {"action": "pull_tasks", "pull_state": "failed", "ui_level": "error", "poll_schedule_seconds": [2], **(ingest_data or {})} ack_request = urllib.request.Request( ack_url, @@ -1008,6 +1044,9 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic if ack_code not in (0, "0", None, ""): return True, f"{ingest_message}；但远端确认失败: {str(ack_response.get('message') or 'ack business error')}", { "action": "pull_tasks", + "pull_state": "ack_warning", + "ui_level": "warning", + "poll_schedule_seconds": [1, 3], "source_record_id": source_record_id, "projection_hash": projection_hash, **(ingest_data or {}), @@ -1017,18 +1056,27 @@ def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dic except json.JSONDecodeError as exc: return True, f"{ingest_message}；但远端确认失败: ack 响应不是合法 JSON ({exc})", { "action": "pull_tasks", + "pull_state": "ack_warning", + "ui_level": "warning", + "poll_schedule_seconds": [1, 3], "source_record_id": source_record_id, **(ingest_data or {}), } except Exception as exc: return True, f"{ingest_message}；但远端确认失败: {exc}", { "action": "pull_tasks", + "pull_state": "ack_warning", + "ui_level": "warning", + "poll_schedule_seconds": [1, 3], "source_record_id": source_record_id, **(ingest_data or {}), } return True, "待检测任务批次拉取并入库成功", { "action": "pull_tasks", + "pull_state": "success", + "ui_level": "success", + "poll_schedule_seconds": [1, 3], "source_record_id": source_record_id, "projection_hash": projection_hash, "batch_code": str(projection.get("batch_code") or "").strip(), diff --git a/domain-api/app/services/sync_record_service.py b/domain-api/app/services/sync_record_service.py index 434a286..2cc5cee 100644 --- a/domain-api/app/services/sync_record_service.py +++ b/domain-api/app/services/sync_record_service.py @@ -99,7 +99,7 @@ def _should_append_runtime_projection(previous_payload: dict, current_projection previous_cluster = previous_projection.get("cluster_summary") or {} current_cluster = current_projection.get("cluster_summary") or {} - for key in ("online_worker_nodes", "online_control_nodes", "busy_nodes", "stale_nodes", "offline_nodes"): + for key in ("online_worker_nodes", "dedicated_online_worker_nodes", "online_control_nodes", "busy_nodes", "stale_nodes", "offline_nodes"): if previous_cluster.get(key) != current_cluster.get(key): return True @@ -480,6 +480,7 @@ def append_runtime_projection_if_changed( "cluster_summary": { "nodes_total": int(cluster.get("nodes_total", 0) or 0), "online_worker_nodes": int((cluster.get("summary") or {}).get("online_worker_nodes", 0) or 0), + "dedicated_online_worker_nodes": int((cluster.get("summary") or {}).get("dedicated_online_worker_nodes", 0) or 0), "online_control_nodes": int((cluster.get("summary") or {}).get("online_control_nodes", 0) or 0), "busy_nodes": list((cluster.get("summary") or {}).get("busy_nodes") or []), "stale_nodes": list((cluster.get("summary") or {}).get("stale_nodes") or []), diff --git a/domain-api/app/services/worker_control_service.py b/domain-api/app/services/worker_control_service.py index 9503db3..470f23b 100644 --- a/domain-api/app/services/worker_control_service.py +++ b/domain-api/app/services/worker_control_service.py @@ -51,6 +51,28 @@ def _run_systemctl( return _run_shell(["sudo", "-n", *systemctl_command], timeout=timeout) +def normalize_systemctl_error(raw_message: str, *, service_name: str = "") -> str: + message = str(raw_message or "").strip() + normalized_service_name = str(service_name or "").strip() + if not message: + target = normalized_service_name or "目标服务" + return f"{target} 控制失败，未返回可用错误信息" + + lowered = message.lower() + if "sudo: a password is required" in lowered or "authentication is required" in lowered: + target = normalized_service_name or "systemd 服务" + return f"{target} 控制失败：当前运行用户没有免密 systemctl 权限，请为 API 进程授予对应 sudo/systemd 权限" + if "unit " in lowered and " could not be found" in lowered: + target = normalized_service_name or "目标服务" + return f"{target} 控制失败：systemd 中未找到该服务，请检查服务名配置是否正确" + if "access denied" in lowered or "permission denied" in lowered: + target = normalized_service_name or "systemd 服务" + return f"{target} 控制失败：权限不足，请检查当前 API 进程用户是否具备 systemctl 控制权限" + if "host is down" in lowered or "failed to connect to bus" in lowered: + return "systemctl 调用失败：当前环境未接入可用的 systemd/dbus，请确认部署方式与 worker_mode 是否匹配" + return message + + def _parse_systemd_timestamp(raw_timestamp: str) -> str: raw_timestamp = (raw_timestamp or "").strip() if not raw_timestamp: @@ -206,7 +228,7 @@ def start_worker() -> tuple[bool, str]: if worker_mode == "linux-systemd": result = _run_systemctl(["start", service_name], timeout=30) if result.returncode != 0: - return False, (result.stderr or result.stdout or "启动 Linux Worker 失败").strip() + return False, normalize_systemctl_error(result.stderr or result.stdout or "启动 Linux Worker 失败", service_name=service_name) return True, f"Linux Worker 启动命令已发送: {service_name}" if os.name != "nt": @@ -234,7 +256,7 @@ def stop_worker() -> tuple[bool, str]: if worker_mode == "linux-systemd": result = _run_systemctl(["stop", service_name], timeout=30) if result.returncode != 0: - return False, (result.stderr or result.stdout or "停止 Linux Worker 失败").strip() + return False, normalize_systemctl_error(result.stderr or result.stdout or "停止 Linux Worker 失败", service_name=service_name) return True, f"Linux Worker 停止命令已发送: {service_name}" if os.name != "nt": diff --git a/domain-web/src/layouts/MainLayout.vue b/domain-web/src/layouts/MainLayout.vue index af41d84..f334509 100644 --- a/domain-web/src/layouts/MainLayout.vue +++ b/domain-web/src/layouts/MainLayout.vue @@ -24,8 +24,11 @@ API {{ runtime.apiOnline ? "在线" : "离线" }} - - Worker {{ runtime.workerOnline ? "在线" : "离线" }} + + 本机 Worker {{ runtime.localWorkerOnline ? "在线" : "离线" }} + + + 有效执行节点 {{ runtime.effectiveWorkerNodes }} {{ runtime.workerMode }} @@ -53,7 +56,8 @@ const authStore = useAuthStore(); const runtime = ref({ apiOnline: false, - workerOnline: false, + localWorkerOnline: false, + effectiveWorkerNodes: 0, workerMode: "windows-local" }); let timer: number | null = null; @@ -77,15 +81,18 @@ const title = computed(() => String(route.meta?.title || "概览")); const loadRuntimeSummary = async () => { try { const response = await runtimeApi.status(); + const summary = response.data?.cluster?.summary || {}; runtime.value = { apiOnline: Boolean(response.data?.api?.pid), - workerOnline: Boolean(response.data?.worker?.running), + localWorkerOnline: Boolean(response.data?.worker?.running), + effectiveWorkerNodes: Number(summary.online_worker_nodes || 0), workerMode: response.data?.worker?.mode || "windows-local" }; } catch { runtime.value = { apiOnline: false, - workerOnline: false, + localWorkerOnline: false, + effectiveWorkerNodes: 0, workerMode: "unknown" }; } diff --git a/domain-web/src/views/dashboard/DashboardView.vue b/domain-web/src/views/dashboard/DashboardView.vue index 4de90a6..47daab8 100644 --- a/domain-web/src/views/dashboard/DashboardView.vue +++ b/domain-web/src/views/dashboard/DashboardView.vue @@ -50,7 +50,8 @@ const stats = ref([ { label: "待检测", value: "--", note: "当前待处理域名" }, { label: "黑名单", value: "--", note: "命中风险域名" }, { label: "API", value: "离线", note: "当前 API 运行状态" }, - { label: "Worker", value: "离线", note: "当前 Worker 运行状态" }, + { label: "本机 Worker", value: "离线", note: "当前节点本机 Worker 运行状态" }, + { label: "有效执行节点", value: "离线", note: "当前集群可承担检测任务的在线节点" }, { label: "模式", value: "--", note: "当前 Worker 运行模式" } ]); @@ -84,7 +85,12 @@ const loadOverview = async (showError = true) => { { label: "待检测", value: String(data.pending_total), note: "当前待处理域名" }, { label: "黑名单", value: String(data.blacklist_total), note: "命中风险域名" }, { label: "API", value: data.api_status, note: "当前 API 运行状态" }, - { label: "Worker", value: data.worker_status, note: "当前 Worker 运行状态" }, + { label: "本机 Worker", value: data.local_worker_status || data.worker_status, note: "当前节点本机 Worker 运行状态" }, + { + label: "有效执行节点", + value: `${data.cluster_worker_status || "offline"} / ${data.cluster_online_worker_nodes || 0}`, + note: `可执行检测的在线节点，独立 Worker ${data.cluster_dedicated_online_worker_nodes || 0}，控制面 ${data.cluster_online_control_nodes || 0}` + }, { label: "模式", value: data.worker_mode, note: "当前 Worker 运行模式" } ]; loadError.value = false; diff --git a/domain-web/src/views/detect/DetectView.vue b/domain-web/src/views/detect/DetectView.vue index 5fc1c96..ba6047f 100644 --- a/domain-web/src/views/detect/DetectView.vue +++ b/domain-web/src/views/detect/DetectView.vue @@ -5,6 +5,11 @@ 停止检测 刷新状态 前往运行中心 + + 关闭回传 + 关键回传 + 全量回传 + 最近同步：{{ lastUpdatedAt || "暂无" }} @@ -19,7 +24,7 @@ /> - {{ status.worker_online ? "是" : "否" }} + {{ status.worker_online ? "是" : "否" }} {{ status.worker_mode || "windows-local" }} {{ status.worker_service_name || "-" }} {{ status.api_service_name || "-" }} @@ -35,6 +40,8 @@ {{ status.proxy_last_refresh_time || "暂无" }} {{ status.last_worker_log_time || "暂无" }} {{ status.worker_latest_start_time || "暂无" }} + {{ status.worker_log_sync_enabled ? "已开启" : "已关闭" }} + {{ status.worker_log_sync_enabled ? (status.worker_log_sync_mode === "full" ? "全量" : "关键") : "-" }} @@ -261,7 +268,7 @@
回到底部 - Worker：{{ status.worker_online ? "在线" : "离线" }} + 本机 Worker：{{ status.worker_online ? "在线" : "离线" }} 阶段：{{ selectedRun.phase_label }} 运行中：{{ status.progress.running || 0 }} 线程：{{ threadCountSummary }} @@ -284,7 +291,7 @@ import { computed, nextTick, onBeforeUnmount, onMounted, ref, watch } from "vue" import { useRouter } from "vue-router"; import { ElMessage } from "element-plus"; import PageCard from "@/components/PageCard.vue"; -import { detectApi } from "@/api/modules"; +import { detectApi, settingsApi } from "@/api/modules"; const DETECT_LAST_ACTION_KEY = "domaincheck:detect:last-action"; const router = useRouter(); @@ -320,7 +327,10 @@ const status = ref({ proxy_last_refresh_time: "", dependency_alerts: [] as any[], log_lines: [] as string[], + remote_log_lines: [] as string[], runs: [] as any[], + worker_log_sync_enabled: false, + worker_log_sync_mode: "key", progress_percent: 0, active_job: null as any, progress: { @@ -424,13 +434,44 @@ const phaseHistory = computed(() => { if (!Array.isArray(items)) return []; return [...items].reverse(); }); +const selectedRunIsLatest = computed(() => { + if (!selectedRun.value || !runs.value.length) { + return false; + } + return selectedRun.value.run_id === runs.value[0]?.run_id; +}); + +const dedupeConsecutiveLines = (lines: string[]) => { + const result: string[] = []; + for (const rawLine of lines) { + const line = String(rawLine || ""); + if (!line) { + continue; + } + if (result[result.length - 1] === line) { + continue; + } + result.push(line); + } + return result; +}; + const selectedLogText = computed(() => { const runLogs = selectedRun.value?.logs; + const mergedLines: string[] = []; if (Array.isArray(runLogs) && runLogs.length) { - return runLogs.join("\n"); + mergedLines.push(...runLogs); + } + const remoteLogLines = status.value.remote_log_lines || []; + const canAppendRemoteLogs = !selectedRun.value || selectedRunIsLatest.value; + if (canAppendRemoteLogs && Array.isArray(remoteLogLines) && remoteLogLines.length) { + mergedLines.push(...remoteLogLines); + } + if (mergedLines.length) { + return dedupeConsecutiveLines(mergedLines).join("\n"); } const liveLogs = status.value.log_lines || []; - return liveLogs.length ? liveLogs.join("\n") : "暂无检测日志"; + return liveLogs.length ? dedupeConsecutiveLines(liveLogs).join("\n") : "暂无检测日志"; }); const proxyAlertType = computed<"success" | "warning" | "info" | "error">(() => { if (status.value.proxy_runtime_state === "healthy") return "success"; @@ -580,18 +621,35 @@ const invoke = async (action: "start" | "stop") => { actionLoading.value = action; try { const response = action === "start" ? await detectApi.start() : await detectApi.stop(); - ElMessage.success(response.message); + const responseMessage = String(response.message || ""); + const uiLevel = String(response.data?.ui_level || "").trim().toLowerCase(); + const actionType: "success" | "warning" | "error" = + uiLevel === "error" ? "error" : uiLevel === "warning" ? "warning" : "success"; + if (actionType === "error") { + ElMessage.error(responseMessage); + } else if (actionType === "warning") { + ElMessage.warning(responseMessage); + } else { + ElMessage.success(responseMessage); + } lastAction.value = { label: action === "start" ? "检测启动" : "检测停止", - message: response.message, + message: responseMessage, at: new Date().toLocaleString("zh-CN", { hour12: false }), - type: "success" + type: actionType }; persistLastAction(); - const pollDelaySeconds = Number(response.data?.poll_after_seconds || 2); - window.setTimeout(() => { - loadStatus(false); - }, pollDelaySeconds * 1000); + await loadStatus(false); + const pollSchedule = Array.isArray(response.data?.poll_schedule_seconds) + ? response.data.poll_schedule_seconds + : [Number(response.data?.poll_after_seconds || 2)]; + [...new Set(pollSchedule.map((item: unknown) => Number(item || 0)).filter((item: number) => item > 0))] + .sort((a, b) => a - b) + .forEach((seconds) => { + window.setTimeout(() => { + loadStatus(false); + }, seconds * 1000); + }); } catch (error: any) { const message = error?.message || `${action === "start" ? "启动" : "停止"}操作失败`; lastAction.value = { @@ -615,6 +673,25 @@ const goRuntime = () => { router.push("/runtime"); }; +const updateWorkerLogSync = async (enabled: boolean, mode: "key" | "full") => { + try { + const current = await settingsApi.getSettings(); + const payload = { + ...current.data, + runtime_settings: { + ...(current.data?.runtime_settings || {}), + worker_log_sync_enabled: enabled, + worker_log_sync_mode: mode + } + }; + await settingsApi.updateSettings(payload); + ElMessage.success(`远端日志回传已${enabled ? `切到${mode === "full" ? "全量" : "关键"}模式` : "关闭"}`); + await loadStatus(); + } catch { + ElMessage.error("更新远端日志回传开关失败"); + } +}; + watch(autoRefresh, ensureRefreshTimer); watch(selectedLogText, async (current, previous) => { if (current === previous) { diff --git a/domain-web/src/views/runtime/RuntimeView.vue b/domain-web/src/views/runtime/RuntimeView.vue index 82cacab..fcaa287 100644 --- a/domain-web/src/views/runtime/RuntimeView.vue +++ b/domain-web/src/views/runtime/RuntimeView.vue @@ -22,9 +22,9 @@ {{ runtime.api?.pid ? "运行中" : "未知" }}
-
Worker 状态
+
本机 Worker
{{ runtime.worker?.service_name || "-" }}
- {{ runtime.worker?.running ? "在线" : "离线" }} + {{ runtime.worker?.running ? "本机在线" : "本机离线" }}
运行模式
@@ -68,10 +68,12 @@
当前已接入控制面与执行面节点总数
-
在线 Worker
+
有效执行节点
{{ runtime.cluster?.summary?.online_worker_nodes || 0 }}
- 控制面 {{ runtime.cluster?.summary?.online_control_nodes || 0 }} + 可承担检测任务的在线节点 + / 独立 Worker {{ runtime.cluster?.summary?.dedicated_online_worker_nodes || 0 }} + / 控制面 {{ runtime.cluster?.summary?.online_control_nodes || 0 }} / 忙碌 {{ runtime.cluster?.summary?.status_counts?.busy || 0 }}
@@ -117,7 +119,10 @@ API {{ runtime.api?.pid ? "运行中" : "未知" }} - Worker {{ runtime.worker?.running ? "在线" : "离线" }} + 本机 Worker {{ runtime.worker?.running ? "在线" : "离线" }} + + + 有效执行节点 {{ runtime.cluster?.summary?.online_worker_nodes || 0 }} 预检 {{ preflight.ok ? "通过" : "异常" }} @@ -196,7 +201,7 @@ show-icon type="info" style="margin-top: 12px" - :title="`当前 Worker 状态：${runtime.worker?.running ? '运行中' : '未运行'}；当前 API 状态：${runtime.api?.pid ? '运行中' : '未知'}；Sync Agent：${runtime.sync_agent?.running ? '运行中' : '未运行'}。`" + :title="`当前本机 Worker：${runtime.worker?.running ? '运行中' : '未运行'}；有效执行节点：${runtime.cluster?.summary?.online_worker_nodes || 0}；独立 Worker：${runtime.cluster?.summary?.dedicated_online_worker_nodes || 0}；当前 API 状态：${runtime.api?.pid ? '运行中' : '未知'}；Sync Agent：${runtime.sync_agent?.running ? '运行中' : '未运行'}。`" />

扩容建议

- 基于当前积压、近窗吞吐和在线 Worker 数自动估算 + 基于当前积压、近窗吞吐和有效执行节点数自动估算
- 在线 Worker {{ runtime.detect?.capacity_plan?.online_worker_nodes || 0 }} + 有效执行节点 {{ runtime.detect?.capacity_plan?.online_worker_nodes || 0 }} 当前吞吐 {{ runtime.detect?.capacity_plan?.current_processed_per_hour || 0 }}/小时 预计剩余 {{ runtime.detect?.capacity_plan?.estimated_hours_remaining || 0 }} 小时 建议总 Worker {{ runtime.detect?.capacity_plan?.recommended_total_workers || 1 }} @@ -377,7 +382,7 @@
在线控制面 {{ runtime.cluster?.summary?.online_control_nodes || 0 }} - 在线 Worker {{ runtime.cluster?.summary?.online_worker_nodes || 0 }} + 有效执行节点 {{ runtime.cluster?.summary?.online_worker_nodes || 0 }} 独立 Worker {{ runtime.cluster?.summary?.dedicated_online_worker_nodes || 0 }} 忙碌 {{ runtime.cluster?.summary?.status_counts?.busy || 0 }} 失活 {{ runtime.cluster?.summary?.status_counts?.stale || 0 }} @@ -803,17 +808,19 @@ const clusterIssueCount = computed(() => { const clusterAlertText = computed(() => { const summary = runtime.cluster?.summary || {}; const busy = Array.isArray(summary.busy_nodes) ? summary.busy_nodes.join("、") : ""; - const stale = Array.isArray(summary.stale_nodes) ? summary.stale_nodes.join("、") : ""; - const offline = Array.isArray(summary.offline_nodes) ? summary.offline_nodes.join("、") : ""; - const parts = []; + const staleNodes = Array.isArray(summary.stale_nodes) ? summary.stale_nodes : []; + const offlineNodes = Array.isArray(summary.offline_nodes) ? summary.offline_nodes : []; + const parts = [ + `当前有效执行节点 ${summary.online_worker_nodes || 0}，其中独立 Worker ${summary.dedicated_online_worker_nodes || 0}` + ]; if (busy) { parts.push(`忙碌节点：${busy}`); } - if (stale) { - parts.push(`失活节点：${stale}`); + if (staleNodes.length) { + parts.push(`失活节点 ${staleNodes.length} 个${staleNodes.length <= 2 ? `：${staleNodes.join("、")}` : "，详见下方节点表"}`); } - if (offline) { - parts.push(`离线节点：${offline}`); + if (offlineNodes.length) { + parts.push(`离线节点 ${offlineNodes.length} 个${offlineNodes.length <= 2 ? `：${offlineNodes.join("、")}` : "，详见下方节点表"}`); } return parts.join("；"); }); @@ -824,7 +831,8 @@ const participatingNodesAlertText = computed(() => { return "当前没有节点正在领任务、执行任务或产生近窗吞吐。"; } const names = rows.map((item: Record) => String(item.node_code || "-")).join("、"); - return `当前参与检测节点 ${rows.length} 台：${names}`; + const effectiveCount = rows.filter((item: Record) => Boolean(item.is_effective_worker)).length; + return `当前参与检测节点 ${rows.length} 台，其中有效执行节点 ${effectiveCount} 台：${names}`; }); const syncAlertText = computed(() => { @@ -1034,18 +1042,36 @@ const invokeAction = async (action: "start_worker" | "stop_worker" | "restart_ap stop_sync_agent: "停止 Sync Agent", push_sync: "立即同步" }; + const responseMessage = String(response.message || ""); + const uiLevel = String(response.data?.ui_level || "").trim().toLowerCase(); + const actionType: "success" | "warning" | "error" = + uiLevel === "error" ? "error" : uiLevel === "warning" ? "warning" : "success"; lastAction.value = { label: labelMap[action], - message: response.message, + message: responseMessage, at: new Date().toLocaleString("zh-CN", { hour12: false }), - type: "success" + type: actionType }; persistLastAction(); appendActionHistory(lastAction.value); - ElMessage.success(response.message); - window.setTimeout(() => { - loadAll(false); - }, Number(response.data?.poll_after_seconds || 2) * 1000); + if (actionType === "error") { + ElMessage.error(responseMessage); + } else if (actionType === "warning") { + ElMessage.warning(responseMessage); + } else { + ElMessage.success(responseMessage); + } + await loadAll(false); + const pollSchedule = Array.isArray(response.data?.poll_schedule_seconds) + ? response.data.poll_schedule_seconds + : [Number(response.data?.poll_after_seconds || 2)]; + [...new Set(pollSchedule.map((item: unknown) => Number(item || 0)).filter((item: number) => item > 0))] + .sort((a, b) => a - b) + .forEach((seconds) => { + window.setTimeout(() => { + loadAll(false); + }, seconds * 1000); + }); } catch (error: any) { const message = error?.message || "运行时动作执行失败"; const labelMap = { diff --git a/domain-web/src/views/settings/SettingsView.vue b/domain-web/src/views/settings/SettingsView.vue index e003190..1b0b25b 100644 --- a/domain-web/src/views/settings/SettingsView.vue +++ b/domain-web/src/views/settings/SettingsView.vue @@ -54,6 +54,25 @@ + + + + + + + + + + + + + @@ -232,7 +251,9 @@ const runtimeSettings = ref({ worker_mode: "windows-local", worker_service_name: "domaincheck-worker", api_service_name: "domaincheck-api", - sync_agent_service_name: "domaincheck-sync-agent" + sync_agent_service_name: "domaincheck-sync-agent", + worker_log_sync_enabled: false, + worker_log_sync_mode: "key" }); const importInput = ref(null); const juziseoLoggingIn = ref(false); @@ -316,6 +337,10 @@ const normalizeNodeThreadOverrides = (payload: Record) .sort((a, b) => a.node_code.localeCompare(b.node_code)); }; +const normalizeWorkerLogSyncMode = (value: unknown) => { + return String(value || "").trim().toLowerCase() === "full" ? "full" : "key"; +}; + const addNodeThreadOverride = () => { nodeThreadOverrides.value.push({ node_code: "", @@ -371,7 +396,9 @@ const loadSettings = async () => { worker_mode: response.data.runtime_settings?.worker_mode || "windows-local", worker_service_name: response.data.runtime_settings?.worker_service_name || "domaincheck-worker", api_service_name: response.data.runtime_settings?.api_service_name || "domaincheck-api", - sync_agent_service_name: response.data.runtime_settings?.sync_agent_service_name || "domaincheck-sync-agent" + sync_agent_service_name: response.data.runtime_settings?.sync_agent_service_name || "domaincheck-sync-agent", + worker_log_sync_enabled: Boolean(response.data.runtime_settings?.worker_log_sync_enabled), + worker_log_sync_mode: normalizeWorkerLogSyncMode(response.data.runtime_settings?.worker_log_sync_mode) }; } catch { ElMessage.error("读取系统设置失败"); diff --git a/domainCheck/detect_worker.py b/domainCheck/detect_worker.py index 5e33531..2e1fe4f 100644 --- a/domainCheck/detect_worker.py +++ b/domainCheck/detect_worker.py @@ -41,6 +41,7 @@ CONFIG_UPDATE_CHANNEL = "domain_tool:config_update" CONTROL_CHANNEL = "domain_tool:worker_control" RUNTIME_STATE_KEY = "domain_tool:detect_runtime_state" PENDING_CONTROL_KEY = "domain_tool:worker_pending_command" +RUNTIME_SETTINGS_KEY = "domain_tool:runtime_settings" class DetectThread(QThread): """ @@ -814,6 +815,11 @@ class DetectWorker: self.current_cycle_token = "" self.current_job_id = None self.current_job_code = "" + self.runtime_settings = { + "worker_log_sync_enabled": False, + "worker_log_sync_mode": "key", + } + self._last_synced_worker_log = "" # 初始化数据库连接 self.db = Database() @@ -868,6 +874,7 @@ class DetectWorker: self.detect_options = self.load_detect_options() self.proxy_config = self.load_proxy_config() self.thread_count = self.load_thread_count() # 从配置文件加载线程数 + self.runtime_settings = self.load_runtime_settings() # 初始化代理池 self.proxy_pool = [] @@ -1042,6 +1049,7 @@ class DetectWorker: def _set_active_cycle_context(self, control_payload=None): payload = control_payload or {} + self._last_synced_worker_log = "" self.current_cycle_token = str(payload.get("cycle_token") or "").strip() job_id = payload.get("job_id") try: @@ -1051,6 +1059,7 @@ class DetectWorker: self.current_job_code = str(payload.get("job_code") or "").strip() def _clear_active_cycle_context(self): + self._last_synced_worker_log = "" self.current_cycle_token = "" self.current_job_id = None self.current_job_code = "" @@ -1091,6 +1100,7 @@ class DetectWorker: self.detecting = True try: logger.info(f"开始执行远程检测任务，来源: {source}") + self._sync_worker_log_event(f"开始执行检测任务，来源: {source}", payload={"source": source}, mode='key') self._update_runtime_state( "starting", f"开始执行检测任务，来源: {source}", @@ -1136,6 +1146,7 @@ class DetectWorker: self._update_runtime_state("idle", "当前没有运行中的检测任务") return False logger.info(f"收到停止检测指令，来源: {source}") + self._sync_worker_log_event(f"收到停止检测指令，来源: {source}", payload={"source": source}, mode='key') self.stop_requested = True self._update_runtime_state( "stopping", @@ -1351,6 +1362,79 @@ class DetectWorker: logger.info(f"使用默认检测线程数: {default_thread_count}") return default_thread_count + def load_runtime_settings(self): + default_settings = { + "worker_log_sync_enabled": False, + "worker_log_sync_mode": "key", + } + try: + if self.use_redis: + runtime_settings_raw = self.redis_client.get(RUNTIME_SETTINGS_KEY) + if runtime_settings_raw: + runtime_settings = default_settings.copy() + runtime_settings.update(json.loads(runtime_settings_raw)) + runtime_settings["worker_log_sync_enabled"] = bool(runtime_settings.get("worker_log_sync_enabled", False)) + runtime_settings["worker_log_sync_mode"] = "full" if str(runtime_settings.get("worker_log_sync_mode", "key")).strip().lower() == "full" else "key" + logger.info(f"从Redis加载运行时设置成功: {runtime_settings}") + return runtime_settings + + candidate_paths = [ + 'runtime_settings.json', + os.path.join('runtime', 'runtime_settings.json'), + ] + for candidate_path in candidate_paths: + if os.path.exists(candidate_path): + with open(candidate_path, 'r', encoding='utf-8') as f: + runtime_settings = default_settings.copy() + runtime_settings.update(json.load(f)) + runtime_settings["worker_log_sync_enabled"] = bool(runtime_settings.get("worker_log_sync_enabled", False)) + runtime_settings["worker_log_sync_mode"] = "full" if str(runtime_settings.get("worker_log_sync_mode", "key")).strip().lower() == "full" else "key" + logger.info(f"从本地文件加载运行时设置成功: {runtime_settings}") + return runtime_settings + except Exception as e: + logger.error(f"加载运行时设置失败: {e}") + return default_settings + + def _worker_log_sync_mode(self): + if not bool((self.runtime_settings or {}).get("worker_log_sync_enabled", False)): + return "off" + return "full" if str((self.runtime_settings or {}).get("worker_log_sync_mode", "key")).strip().lower() == "full" else "key" + + def _sync_worker_log_event(self, message, level='info', payload=None, *, mode='key', job_item_id=None): + current_mode = self._worker_log_sync_mode() + if current_mode == "off": + return + if mode == 'full' and current_mode != "full": + return + normalized_message = str(message or '').strip() + if not normalized_message: + return + if normalized_message == self._last_synced_worker_log and mode != 'full': + return + job_id = self.current_job_id + if not job_id: + return + event_payload = dict(payload or {}) + if self.current_cycle_token and not event_payload.get("cycle_token"): + event_payload["cycle_token"] = self.current_cycle_token + if self.current_job_code and not event_payload.get("job_code"): + event_payload["job_code"] = self.current_job_code + event_payload["log_mode"] = mode + event_payload["synced_by"] = "worker_log_callback" + try: + self.db.append_detect_run_event( + job_id, + job_item_id, + config.NODE_CODE, + event_type='worker_log', + message=normalized_message, + level=level, + payload=event_payload, + ) + self._last_synced_worker_log = normalized_message + except Exception as e: + logger.debug(f"回传Worker日志事件失败: {e}") + def test_proxy(self, proxy_item, result_queue): """ 测试单个代理的可用性 @@ -1495,6 +1579,7 @@ class DetectWorker: "idle" if not self.detecting else "running", "代理未启用，使用直接连接", ) + self._sync_worker_log_event("代理未启用，使用直接连接", mode='key') self.proxy_refresh_lock.release() return @@ -1510,6 +1595,7 @@ class DetectWorker: "refreshing_proxy" if self.detecting else "idle", f"代理池刷新冷却中，{wait_seconds} 秒后再试", ) + self._sync_worker_log_event(f"代理池刷新冷却中，{wait_seconds} 秒后再试", mode='full') return proxy_api_urls = self.proxy_config.get('proxy_urls') or [] @@ -1622,6 +1708,16 @@ class DetectWorker: "running" if self.detecting else "idle", f"代理池刷新完成，共 {len(new_proxies)} 个可用代理，来源链接 {len(proxy_api_urls)} 个，原始 {self.proxy_last_refresh_total_items} 个，验证 {self.proxy_last_validated_count} 个", ) + self._sync_worker_log_event( + f"代理池刷新完成，共 {len(new_proxies)} 个可用代理，来源链接 {len(proxy_api_urls)} 个，原始 {self.proxy_last_refresh_total_items} 个，验证 {self.proxy_last_validated_count} 个", + payload={ + "available_proxy_count": len(new_proxies), + "source_count": len(proxy_api_urls), + "raw_items": self.proxy_last_refresh_total_items, + "validated_count": self.proxy_last_validated_count, + }, + mode='key', + ) else: with self.proxy_pool_lock: self.proxy_pool = [] @@ -1636,6 +1732,7 @@ class DetectWorker: "refreshing_proxy" if self.detecting else "idle", "所有代理池链接均未返回可用代理数据", ) + self._sync_worker_log_event("所有代理池链接均未返回可用代理数据", level='warning', mode='key') else: with self.proxy_pool_lock: self.proxy_pool = [] @@ -1649,6 +1746,7 @@ class DetectWorker: "idle" if not self.detecting else "running", "未配置代理池链接", ) + self._sync_worker_log_event("未配置代理池链接", level='warning', mode='key') except Exception as e: with self.proxy_pool_lock: self.proxy_pool = [] @@ -1663,6 +1761,7 @@ class DetectWorker: "refreshing_proxy" if self.detecting else "failed", f"刷新代理池失败: {e}", ) + self._sync_worker_log_event(f"刷新代理池失败: {e}", level='error', mode='key') finally: self.proxy_refresh_lock.release() @@ -2383,6 +2482,15 @@ class DetectWorker: "job_code": job_code, }, ) + self._sync_worker_log_event( + f"开始检测域名: {domain_name}", + payload={ + "domain_id": domain_id, + "domain": domain_name, + }, + mode='full', + job_item_id=job_item_id, + ) # 使用共享的敏感词列表 sensitive_words = self.sensitive_words @@ -2433,11 +2541,32 @@ class DetectWorker: self._complete_detection(domain_id, domain_name) finalize_job_item('completed', f"域名检测完成: {domain_name}") logger.info(f"域名检测完成: {domain_name}") + self._sync_worker_log_event( + f"域名检测完成: {domain_name}", + payload={ + "domain_id": domain_id, + "domain": domain_name, + "status": "completed", + }, + mode='full', + job_item_id=job_item_id, + ) except Exception as e: logger.error(f"检测域名出错: {domain_name}, 错误: {e}") self.db.update_domain_detect_status(domain_id, DETECT_STATUS_FAILED) finalize_job_item('failed', str(e)) + self._sync_worker_log_event( + f"检测域名出错: {domain_name}, 错误: {e}", + level='error', + payload={ + "domain_id": domain_id, + "domain": domain_name, + "status": "failed", + }, + mode='full', + job_item_id=job_item_id, + ) def start_detection(self): """ @@ -2447,6 +2576,7 @@ class DetectWorker: self.detecting = True self.stop_requested = False self._mark_detection_phase("preparing", "开始执行域名检测任务，正在加载配置") + self._sync_worker_log_event("开始执行域名检测任务，正在加载配置", mode='key') # 重新加载配置，确保获取最新的配置 try: @@ -2468,6 +2598,7 @@ class DetectWorker: import traceback logger.error(traceback.format_exc()) self._mark_detection_phase("failed", f"加载配置失败: {e}") + self._sync_worker_log_event(f"加载配置失败: {e}", level='error', mode='key') return # 重新加载cookies @@ -2486,6 +2617,7 @@ class DetectWorker: if self.proxy_config.get('proxy_enable', False): logger.info("开始检测，刷新代理池") self._mark_detection_phase("refreshing_proxy", "开始检测，正在刷新代理池") + self._sync_worker_log_event("开始检测，正在刷新代理池", mode='key') self.refresh_proxy_pool() # 更新JC和Juziseo实例的代理设置 @@ -2516,6 +2648,7 @@ class DetectWorker: if self.stop_requested: logger.info("检测任务收到停止请求，停止继续领取任务") self._mark_detection_phase("stopping", "检测任务收到停止请求，准备安全退出") + self._sync_worker_log_event("检测任务收到停止请求，准备安全退出", level='warning', mode='key') break recycled_job_items = self.db.recycle_expired_detect_job_items() if recycled_job_items: @@ -2538,10 +2671,19 @@ class DetectWorker: batch_size=current_batch_size, queue_source="detect_job_items" if using_job_queue else "domains", ) + self._sync_worker_log_event( + f"从{queue_label}获取到 {current_batch_size} 个需要检测的域名", + payload={ + "batch_size": current_batch_size, + "queue_source": "detect_job_items" if using_job_queue else "domains", + }, + mode='full', + ) if not domains: logger.info("没有需要检测的域名") self._mark_detection_phase("idle", "当前没有需要检测的域名，Worker 等待下一次启动") + self._sync_worker_log_event("当前没有需要检测的域名，Worker 等待下一次启动", mode='key') break # 旧链路直接扫 domains 表时，少于 batch_size 代表已接近尾批； @@ -2560,6 +2702,11 @@ class DetectWorker: batch_size=current_batch_size, max_threads=max_threads, ) + self._sync_worker_log_event( + f"开始创建线程，当前批次域名数: {current_batch_size}，最大线程数: {max_threads}", + payload={"batch_size": current_batch_size, "max_threads": max_threads}, + mode='full', + ) for i, domain in enumerate(domains): # 检查是否需要停止 if not self.running or self.stop_requested: @@ -2627,6 +2774,15 @@ class DetectWorker: max_threads=max_threads, batch_size=current_batch_size, ) + self._sync_worker_log_event( + f"当前实际线程数量: {current_active}/{max_threads}", + payload={ + "active_threads": current_active, + "max_threads": max_threads, + "batch_size": current_batch_size, + }, + mode='full', + ) # 更新GUI显示 if self.detect_thread: @@ -2705,9 +2861,19 @@ class DetectWorker: f"当前批次域名数量 ({current_batch_size}) 少于 {batch_size}，本轮检测即将完成", processed=total_processed, ) + self._sync_worker_log_event( + f"当前批次域名数量 ({current_batch_size}) 少于 {batch_size}，本轮检测即将完成", + payload={"processed": total_processed}, + mode='key', + ) break logger.info(f"当前批次检测完成，累计处理 {total_processed} 个域名") self._mark_detection_phase("batch_completed", f"当前批次检测完成，累计处理 {total_processed} 个域名", processed=total_processed) + self._sync_worker_log_event( + f"当前批次检测完成，累计处理 {total_processed} 个域名", + payload={"processed": total_processed}, + mode='key', + ) # 完成进度 if self.detect_thread: @@ -2721,10 +2887,12 @@ class DetectWorker: logger.info("域名检测任务完成") self._mark_detection_phase("completed", "域名检测任务完成") + self._sync_worker_log_event("域名检测任务完成", payload={"processed": total_processed}, mode='key') except Exception as e: logger.error(f"执行检测任务出错: {e}") self._mark_detection_phase("failed", f"执行检测任务出错: {e}") + self._sync_worker_log_event(f"执行检测任务出错: {e}", level='error', mode='key') finally: self.detecting = False @@ -2922,6 +3090,7 @@ class DetectWorker: self.detect_options = self.load_detect_options() self.proxy_config = self.load_proxy_config() self.thread_count = self.load_thread_count() + self.runtime_settings = self.load_runtime_settings() self.load_cookies_from_remote() self.update_config_labels() logger.debug("配置已更新")