619 lines
23 KiB
Python
619 lines
23 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import shlex
|
||
import subprocess
|
||
from io import StringIO
|
||
|
||
try:
|
||
import paramiko
|
||
except ImportError: # pragma: no cover - exercised via graceful fallback tests
|
||
paramiko = None
|
||
|
||
from app.core.db import get_db
|
||
from app.core.config import settings
|
||
from app.services.ops_action_executor_core import (
|
||
STRUCTURED_ACTIONS,
|
||
build_service_name_map,
|
||
execute_structured_action,
|
||
trim_output,
|
||
)
|
||
from app.services.ops_release_executor_core import build_remote_release_action_script, execute_release_action
|
||
|
||
|
||
_SSH_SUPPORTED_ACTIONS = set(STRUCTURED_ACTIONS) | {"deploy.release"}
|
||
_SSH_CONNECT_TIMEOUT_SECONDS = 12
|
||
_SSH_REMOTE_TIMEOUT_SECONDS = {
|
||
"health.snapshot": 30,
|
||
"service.status": 45,
|
||
"service.restart": 45,
|
||
"service.start": 45,
|
||
"service.stop": 45,
|
||
"logs.collect": 45,
|
||
"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,
|
||
"deploy.release": 180,
|
||
}
|
||
|
||
|
||
def supports_ssh_execution(action: str) -> bool:
|
||
return str(action or "").strip() in _SSH_SUPPORTED_ACTIONS
|
||
|
||
|
||
def execute_local_support_action(action: str, payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized_action = str(action or "").strip()
|
||
if normalized_action not in _SSH_SUPPORTED_ACTIONS:
|
||
return False, f"当前未实现本机支持动作: {normalized_action}", {}
|
||
if normalized_action == "deploy.release":
|
||
return execute_release_action(
|
||
dict(payload or {}),
|
||
run_command=_run_local_command,
|
||
default_api_service_name=settings.api_service_name,
|
||
user_agent="domaincheck-local-runtime/0.1",
|
||
)
|
||
return execute_structured_action(
|
||
normalized_action,
|
||
dict(payload or {}),
|
||
service_names=_service_name_map(),
|
||
runner=_run_local_command,
|
||
)
|
||
|
||
|
||
def execute_ssh_action(node: dict, action: str, payload: dict | None = None) -> tuple[bool, str, dict]:
|
||
normalized_action = str(action or "").strip()
|
||
if normalized_action not in _SSH_SUPPORTED_ACTIONS:
|
||
return False, f"当前 SSH 执行器暂不支持动作: {normalized_action}", {}
|
||
|
||
ssh_host = str(node.get("ssh_host") or "").strip()
|
||
ssh_user = str(node.get("ssh_user") or "").strip()
|
||
ssh_port = max(1, int(node.get("ssh_port") or 22))
|
||
node_code = str(node.get("node_code") or "").strip()
|
||
if not ssh_host or not ssh_user:
|
||
return False, "目标节点缺少 SSH 主机或用户信息", {
|
||
"node_code": node_code,
|
||
"ssh_host": ssh_host,
|
||
"ssh_user": ssh_user,
|
||
"ssh_port": ssh_port,
|
||
}
|
||
|
||
normalized_payload = dict(payload or {})
|
||
node_secret = _load_ssh_secret(node_code)
|
||
service_names = _service_name_map()
|
||
if normalized_action == "deploy.release":
|
||
remote_script = build_remote_release_action_script(
|
||
normalized_payload,
|
||
default_api_service_name=service_names.get("api") or "domaincheck-api",
|
||
user_agent="domaincheck-ssh/0.1",
|
||
)
|
||
else:
|
||
remote_script = _build_remote_python_script(
|
||
action=normalized_action,
|
||
payload=normalized_payload,
|
||
service_names=service_names,
|
||
)
|
||
remote_command = "\n".join(
|
||
[
|
||
"set -euo pipefail",
|
||
"if command -v python3 >/dev/null 2>&1; then",
|
||
" PYTHON_BIN=python3",
|
||
"elif command -v python >/dev/null 2>&1; then",
|
||
" PYTHON_BIN=python",
|
||
"else",
|
||
' echo "{\\"ok\\": false, \\"message\\": \\"python interpreter missing on remote node\\", \\"result\\": {\\"executor\\": \\"ssh\\"}}"',
|
||
" exit 127",
|
||
"fi",
|
||
'"${PYTHON_BIN}" - <<\'PY\'',
|
||
remote_script.rstrip("\n"),
|
||
"PY",
|
||
]
|
||
)
|
||
timeout_seconds = int(_SSH_REMOTE_TIMEOUT_SECONDS.get(normalized_action, 45) or 45) + _SSH_CONNECT_TIMEOUT_SECONDS
|
||
auth_mode = str(node.get("auth_mode") or "").strip() or ("password" if node_secret.get("ssh_password") else "key")
|
||
ssh_password = str(node_secret.get("ssh_password") or "").strip()
|
||
ssh_private_key = str(node_secret.get("ssh_private_key") or "").strip()
|
||
if auth_mode == "key" and not ssh_private_key and ssh_password:
|
||
auth_mode = "password"
|
||
elif auth_mode == "password" and not ssh_password and ssh_private_key:
|
||
auth_mode = "key"
|
||
if (auth_mode == "password" and ssh_password) or ssh_private_key:
|
||
if paramiko is None:
|
||
return False, "当前环境未安装 paramiko，无法使用密码或私钥 SSH 执行", {
|
||
"executor": "ssh",
|
||
"action": normalized_action,
|
||
"transport": {
|
||
"executor": "ssh",
|
||
"node_code": node_code,
|
||
"ssh_host": ssh_host,
|
||
"ssh_user": ssh_user,
|
||
"ssh_port": ssh_port,
|
||
"auth_mode": auth_mode,
|
||
"action": normalized_action,
|
||
},
|
||
}
|
||
if auth_mode == "password" and ssh_password:
|
||
completed = _run_paramiko_command(
|
||
ssh_host=ssh_host,
|
||
ssh_port=ssh_port,
|
||
ssh_user=ssh_user,
|
||
remote_command=remote_command,
|
||
timeout_seconds=timeout_seconds,
|
||
ssh_password=ssh_password,
|
||
)
|
||
elif ssh_private_key:
|
||
completed = _run_paramiko_command(
|
||
ssh_host=ssh_host,
|
||
ssh_port=ssh_port,
|
||
ssh_user=ssh_user,
|
||
remote_command=remote_command,
|
||
timeout_seconds=timeout_seconds,
|
||
ssh_private_key=ssh_private_key,
|
||
)
|
||
else:
|
||
ssh_command = [
|
||
"ssh",
|
||
"-o",
|
||
"BatchMode=yes",
|
||
"-o",
|
||
"PreferredAuthentications=publickey",
|
||
"-o",
|
||
"StrictHostKeyChecking=accept-new",
|
||
"-o",
|
||
f"ConnectTimeout={_SSH_CONNECT_TIMEOUT_SECONDS}",
|
||
"-p",
|
||
str(ssh_port),
|
||
f"{ssh_user}@{ssh_host}",
|
||
remote_command,
|
||
]
|
||
completed = subprocess.run(
|
||
ssh_command,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout_seconds,
|
||
)
|
||
|
||
transport = {
|
||
"executor": "ssh",
|
||
"node_code": node_code,
|
||
"ssh_host": ssh_host,
|
||
"ssh_user": ssh_user,
|
||
"ssh_port": ssh_port,
|
||
"auth_mode": auth_mode,
|
||
"action": normalized_action,
|
||
"returncode": int(completed.returncode or 0),
|
||
}
|
||
parsed_ok, parsed_payload = _extract_json_from_output(completed.stdout or "")
|
||
if parsed_ok:
|
||
ok = bool(parsed_payload.get("ok", False))
|
||
message = str(parsed_payload.get("message") or "").strip() or ("SSH 动作执行成功" if ok else "SSH 动作执行失败")
|
||
result = dict(parsed_payload.get("result") or {})
|
||
result["transport"] = {
|
||
**transport,
|
||
"stdout_preview": trim_output(completed.stdout, 4000),
|
||
"stderr_preview": trim_output(completed.stderr, 2000),
|
||
}
|
||
if completed.returncode != 0 and ok:
|
||
ok = False
|
||
message = f"SSH 命令返回码异常: {completed.returncode}"
|
||
return ok, message, result
|
||
|
||
stderr_preview = trim_output(completed.stderr, 4000)
|
||
stdout_preview = trim_output(completed.stdout, 4000)
|
||
message = stderr_preview or stdout_preview or f"SSH 执行失败，返回码 {completed.returncode}"
|
||
return False, message, {
|
||
"executor": "ssh",
|
||
"action": normalized_action,
|
||
"transport": {
|
||
**transport,
|
||
"stdout_preview": stdout_preview,
|
||
"stderr_preview": stderr_preview,
|
||
},
|
||
}
|
||
|
||
|
||
def _load_ssh_secret(node_code: str) -> dict:
|
||
normalized_node_code = str(node_code or "").strip()
|
||
if not normalized_node_code:
|
||
return {}
|
||
try:
|
||
with get_db() as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT ssh_password, ssh_private_key
|
||
FROM ops_managed_node_secrets
|
||
WHERE node_code = %s
|
||
LIMIT 1
|
||
""",
|
||
(normalized_node_code,),
|
||
)
|
||
row = cur.fetchone()
|
||
except Exception:
|
||
return {}
|
||
if not row:
|
||
return {}
|
||
return {
|
||
"ssh_password": str(row[0] or ""),
|
||
"ssh_private_key": str(row[1] or ""),
|
||
}
|
||
|
||
|
||
def _load_private_key(private_key_text: str) -> paramiko.PKey:
|
||
if paramiko is None:
|
||
raise RuntimeError("paramiko is not installed")
|
||
key_text = str(private_key_text or "")
|
||
for key_cls in (paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey, paramiko.DSSKey):
|
||
try:
|
||
return key_cls.from_private_key(StringIO(key_text))
|
||
except Exception:
|
||
continue
|
||
raise ValueError("无法识别 SSH 私钥格式")
|
||
|
||
|
||
def _run_paramiko_command(
|
||
*,
|
||
ssh_host: str,
|
||
ssh_port: int,
|
||
ssh_user: str,
|
||
remote_command: str,
|
||
timeout_seconds: int,
|
||
ssh_password: str = "",
|
||
ssh_private_key: str = "",
|
||
) -> subprocess.CompletedProcess:
|
||
if paramiko is None:
|
||
raise RuntimeError("paramiko is not installed")
|
||
client = paramiko.SSHClient()
|
||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
connect_kwargs = {
|
||
"hostname": ssh_host,
|
||
"port": int(ssh_port),
|
||
"username": ssh_user,
|
||
"timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"banner_timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"auth_timeout": _SSH_CONNECT_TIMEOUT_SECONDS,
|
||
"look_for_keys": False,
|
||
"allow_agent": False,
|
||
}
|
||
if ssh_password:
|
||
connect_kwargs["password"] = ssh_password
|
||
elif ssh_private_key:
|
||
connect_kwargs["pkey"] = _load_private_key(ssh_private_key)
|
||
else:
|
||
connect_kwargs["look_for_keys"] = True
|
||
connect_kwargs["allow_agent"] = True
|
||
try:
|
||
client.connect(**connect_kwargs)
|
||
wrapped_command = f"bash -lc {shlex.quote(remote_command)}"
|
||
_, stdout, stderr = client.exec_command(wrapped_command, timeout=timeout_seconds)
|
||
returncode = int(stdout.channel.recv_exit_status())
|
||
stdout_text = stdout.read().decode("utf-8", errors="replace")
|
||
stderr_text = stderr.read().decode("utf-8", errors="replace")
|
||
return subprocess.CompletedProcess(
|
||
args=["paramiko", f"{ssh_user}@{ssh_host}"],
|
||
returncode=returncode,
|
||
stdout=stdout_text,
|
||
stderr=stderr_text,
|
||
)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def _service_name_map() -> dict[str, str]:
|
||
return build_service_name_map(
|
||
api_service_name=settings.api_service_name,
|
||
worker_service_name=settings.worker_service_name,
|
||
sync_agent_service_name=settings.sync_agent_service_name,
|
||
)
|
||
|
||
|
||
def _run_local_command(command: list[str], *, timeout: int = 60) -> tuple[int, str, str]:
|
||
completed = subprocess.run(command, capture_output=True, text=True, timeout=timeout)
|
||
return int(completed.returncode or 0), str(completed.stdout or "").strip(), str(completed.stderr or "").strip()
|
||
|
||
|
||
def _build_remote_python_script(*, action: str, payload: dict, service_names: dict[str, str]) -> str:
|
||
return f"""import json
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
SERVICE_NAMES = {json.dumps(service_names, ensure_ascii=False)}
|
||
ACTION = {json.dumps(str(action or '').strip(), ensure_ascii=False)}
|
||
PAYLOAD = {json.dumps(dict(payload or {}), ensure_ascii=False)}
|
||
|
||
|
||
def trim_output(text, limit):
|
||
value = str(text or "").strip()
|
||
if len(value) <= int(limit):
|
||
return value
|
||
return value[-int(limit):]
|
||
|
||
|
||
def run(cmd, timeout=60):
|
||
completed = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
|
||
normalized_cmd = [str(part or "").strip() for part in cmd]
|
||
combined_output = f"{{completed.stdout or ''}}\\n{{completed.stderr or ''}}".lower()
|
||
needs_sudo_retry = (
|
||
normalized_cmd
|
||
and normalized_cmd[0] == "systemctl"
|
||
and completed.returncode != 0
|
||
and "sudo" not in normalized_cmd
|
||
and any(
|
||
marker in combined_output
|
||
for marker in (
|
||
"interactive authentication required",
|
||
"authentication is required",
|
||
"authorization not available",
|
||
"polkit",
|
||
)
|
||
)
|
||
)
|
||
if needs_sudo_retry:
|
||
completed = subprocess.run(["sudo", "-n", *normalized_cmd], capture_output=True, text=True, timeout=timeout)
|
||
return int(completed.returncode or 0), str(completed.stdout or "").strip(), str(completed.stderr or "").strip()
|
||
|
||
|
||
def read_tail_lines(path, max_lines):
|
||
file_path = Path(path)
|
||
if not file_path.exists() or not file_path.is_file():
|
||
return []
|
||
with file_path.open("r", encoding="utf-8", errors="replace") as handle:
|
||
return handle.read().splitlines()[-int(max_lines):]
|
||
|
||
|
||
def candidate_domain_roots():
|
||
roots = []
|
||
for candidate in (
|
||
Path.cwd() / "domainCheck",
|
||
Path("/opt/domaincheck/domainCheck"),
|
||
Path("/www/wwwroot/getDomain/domainCheck"),
|
||
):
|
||
if candidate not in roots:
|
||
roots.append(candidate)
|
||
return roots
|
||
|
||
|
||
def candidate_runtime_logs_dirs():
|
||
dirs = []
|
||
for candidate in (
|
||
Path.cwd() / "domain-api" / "runtime" / "logs",
|
||
Path("/opt/domaincheck/domain-api/runtime/logs"),
|
||
Path("/www/wwwroot/getDomain/domain-api/runtime/logs"),
|
||
):
|
||
if candidate not in dirs:
|
||
dirs.append(candidate)
|
||
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:
|
||
return []
|
||
candidates = []
|
||
|
||
def append_candidate(path):
|
||
if path not in candidates:
|
||
candidates.append(path)
|
||
|
||
if normalized_service_name == (SERVICE_NAMES.get("worker") or "domaincheck-worker"):
|
||
for domain_root in candidate_domain_roots():
|
||
append_candidate(domain_root / "detect_worker.log")
|
||
append_candidate(domain_root / "logs" / "detect_worker.log")
|
||
elif normalized_service_name == (SERVICE_NAMES.get("api") or "domaincheck-api"):
|
||
for runtime_logs_dir in candidate_runtime_logs_dirs():
|
||
append_candidate(runtime_logs_dir / "domain-api.stderr.log")
|
||
append_candidate(runtime_logs_dir / "domain-api.stdout.log")
|
||
for domain_root in candidate_domain_roots():
|
||
append_candidate(domain_root / "logs" / "app.log")
|
||
|
||
return candidates
|
||
|
||
|
||
def collect_log_file_fallback(service_name, lines):
|
||
aggregated_lines = []
|
||
used_paths = []
|
||
for path in service_log_file_candidates(service_name):
|
||
current_lines = read_tail_lines(path, lines)
|
||
if not current_lines:
|
||
continue
|
||
aggregated_lines.extend(current_lines)
|
||
used_paths.append(str(path))
|
||
if not aggregated_lines:
|
||
return "", []
|
||
return "\\n".join(aggregated_lines[-int(lines):]), used_paths
|
||
|
||
|
||
def service_name_from_payload(payload):
|
||
value = str((payload or {{}}).get("service_name") or "").strip()
|
||
return value or SERVICE_NAMES.get("worker") or "domaincheck-worker"
|
||
|
||
|
||
def service_name_for_action(action, payload):
|
||
if action in ("service.status", "service.restart", "service.start", "service.stop"):
|
||
return service_name_from_payload(payload)
|
||
runtime_action_map = {{
|
||
"runtime.start_worker": SERVICE_NAMES.get("worker") or "domaincheck-worker",
|
||
"runtime.stop_worker": SERVICE_NAMES.get("worker") or "domaincheck-worker",
|
||
"runtime.restart_api": SERVICE_NAMES.get("api") or "domaincheck-api",
|
||
"runtime.start_sync_agent": SERVICE_NAMES.get("sync_agent") or "domaincheck-sync-agent",
|
||
"runtime.stop_sync_agent": SERVICE_NAMES.get("sync_agent") or "domaincheck-sync-agent",
|
||
}}
|
||
return str(runtime_action_map.get(action) or "").strip()
|
||
|
||
|
||
def systemctl_action_name(action):
|
||
if action in ("service.restart", "runtime.restart_api"):
|
||
return "restart"
|
||
if action in ("service.start", "runtime.start_worker", "runtime.start_sync_agent"):
|
||
return "start"
|
||
if action in ("service.stop", "runtime.stop_worker", "runtime.stop_sync_agent"):
|
||
return "stop"
|
||
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 = {{}}
|
||
for key, service_name in SERVICE_NAMES.items():
|
||
code, stdout, stderr = run(["systemctl", "is-active", service_name], timeout=15)
|
||
checks[key] = {{
|
||
"service_name": service_name,
|
||
"returncode": int(code or 0),
|
||
"state": stdout or stderr,
|
||
"ok": int(code or 0) == 0 and (stdout or stderr) == "active",
|
||
}}
|
||
return True, "health snapshot collected", {{"checks": checks}}
|
||
|
||
if action == "service.status":
|
||
service_name = service_name_from_payload(payload)
|
||
code, stdout, stderr = run(["systemctl", "status", service_name, "--no-pager", "-l"], timeout=45)
|
||
result = {{
|
||
"service_name": service_name,
|
||
"stdout": trim_output(stdout, 12000),
|
||
"stderr": trim_output(stderr, 4000),
|
||
}}
|
||
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)
|
||
if not service_name:
|
||
return False, f"missing service name for action: {{action}}", {{"action": action}}
|
||
code, stdout, stderr = run(["systemctl", systemctl_action, service_name], timeout=45)
|
||
result = {{
|
||
"service_name": service_name,
|
||
"systemctl_action": systemctl_action,
|
||
"stdout": trim_output(stdout, 12000),
|
||
"stderr": trim_output(stderr, 4000),
|
||
}}
|
||
return int(code or 0) == 0, stdout or stderr or f"{{service_name}} {{systemctl_action}} completed", result
|
||
|
||
if action == "logs.collect":
|
||
service_name = service_name_from_payload(payload)
|
||
lines = max(20, min(int((payload or {{}}).get("lines") or 120), 500))
|
||
code, stdout, stderr = run(["journalctl", "-u", service_name, "-n", str(lines), "--no-pager"], timeout=45)
|
||
journal_output = stdout or stderr
|
||
fallback_output, fallback_paths = collect_log_file_fallback(service_name, lines)
|
||
collection_source = "journal"
|
||
effective_output = journal_output
|
||
ok = int(code or 0) == 0
|
||
if (not ok or not stdout.strip()) and fallback_output:
|
||
collection_source = "file_fallback"
|
||
effective_output = fallback_output
|
||
ok = True
|
||
result = {{
|
||
"service_name": service_name,
|
||
"lines": lines,
|
||
"collection_source": collection_source,
|
||
"fallback_used": bool(collection_source == "file_fallback"),
|
||
"fallback_reason": trim_output(journal_output, 4000) if collection_source == "file_fallback" else "",
|
||
"log_paths": fallback_paths,
|
||
"journal_returncode": int(code or 0),
|
||
"stdout": trim_output(effective_output, 20000),
|
||
"stderr": trim_output(stderr, 4000),
|
||
}}
|
||
return ok, effective_output or f"{{service_name}} logs collected", result
|
||
|
||
if action == "diagnostics.collect":
|
||
lines = max(20, min(int((payload or {{}}).get("lines") or 200), 800))
|
||
diagnostics = {{
|
||
"services": {{}},
|
||
"lines": lines,
|
||
}}
|
||
for key, service_name in SERVICE_NAMES.items():
|
||
active_code, active_stdout, active_stderr = run(["systemctl", "is-active", service_name], timeout=15)
|
||
status_code, status_stdout, status_stderr = run(["systemctl", "status", service_name, "--no-pager", "-l"], timeout=45)
|
||
log_code, log_stdout, log_stderr = run(["journalctl", "-u", service_name, "-n", str(lines), "--no-pager"], timeout=45)
|
||
log_output = log_stdout or log_stderr
|
||
fallback_output, fallback_paths = collect_log_file_fallback(service_name, lines)
|
||
log_source = "journal"
|
||
if (int(log_code or 0) != 0 or not log_stdout.strip()) and fallback_output:
|
||
log_output = fallback_output
|
||
log_source = "file_fallback"
|
||
diagnostics["services"][key] = {{
|
||
"service_name": service_name,
|
||
"active_returncode": int(active_code or 0),
|
||
"active_state": active_stdout or active_stderr,
|
||
"status_returncode": int(status_code or 0),
|
||
"status_output": trim_output(status_stdout or status_stderr, 12000),
|
||
"log_returncode": int(log_code or 0),
|
||
"log_source": log_source,
|
||
"log_paths": fallback_paths,
|
||
"log_output": trim_output(log_output, 20000),
|
||
}}
|
||
return True, "diagnostics collected", diagnostics
|
||
|
||
return False, f"unsupported action: {{action}}", {{"action": action}}
|
||
|
||
|
||
def main():
|
||
try:
|
||
ok, message, result = execute(ACTION, PAYLOAD)
|
||
except Exception as exc:
|
||
ok, message, result = False, str(exc), {{"exception": str(exc), "action": ACTION}}
|
||
print(json.dumps({{"ok": ok, "message": message, "result": result}}, ensure_ascii=False))
|
||
raise SystemExit(0 if ok else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
"""
|
||
|
||
|
||
def _extract_json_from_output(text: str) -> tuple[bool, dict]:
|
||
for raw_line in reversed(str(text or "").splitlines()):
|
||
line = str(raw_line or "").strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
data = json.loads(line)
|
||
except Exception:
|
||
continue
|
||
if isinstance(data, dict) and "ok" in data:
|
||
return True, data
|
||
return False, {}
|