feat: background remote agent updates survive SAC navigation (0.20.19)
This commit is contained in:
PTah
@@ -0,0 +1,245 @@
|
||||
"""Background SSH/WinRM host actions (survives UI navigation and multi-worker API)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Callable
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import SessionLocal
|
||||
from app.models import Host
|
||||
from app.services.agent_update import execute_agent_update_fallback
|
||||
from app.services.agent_update_types import AgentUpdateFallbackResult
|
||||
from app.services.linux_admin_settings import get_effective_linux_admin_config
|
||||
from app.services.agent_update_settings import get_effective_agent_update_config
|
||||
from app.services.ssh_connect import iter_ssh_targets, run_ssh_monitor_update, SshCommandResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
ActionRunner = Callable[[Session, Host], AgentUpdateFallbackResult | SshCommandResult]
|
||||
|
||||
_lock = threading.Lock()
|
||||
_running: set[int] = set()
|
||||
|
||||
|
||||
class RemoteActionAlreadyRunningError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _run_inline() -> bool:
|
||||
return os.environ.get("SAC_REMOTE_ACTION_INLINE", "").strip().lower() in ("1", "true", "yes")
|
||||
|
||||
|
||||
def _utcnow() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
|
||||
def _format_output(result: AgentUpdateFallbackResult | SshCommandResult) -> str:
|
||||
parts: list[str] = []
|
||||
stdout = (getattr(result, "stdout", None) or "").strip()
|
||||
stderr = (getattr(result, "stderr", None) or "").strip()
|
||||
if stdout:
|
||||
parts.append(stdout)
|
||||
if stderr:
|
||||
parts.append(stderr)
|
||||
exit_code = getattr(result, "exit_code", None)
|
||||
if exit_code is not None:
|
||||
parts.append(f"exit code: {exit_code}")
|
||||
return "\n\n".join(parts)
|
||||
|
||||
|
||||
def _result_payload(
|
||||
result: AgentUpdateFallbackResult | SshCommandResult,
|
||||
*,
|
||||
title: str,
|
||||
started_at: str,
|
||||
) -> dict[str, Any]:
|
||||
finished = _utcnow().isoformat()
|
||||
product_version = getattr(result, "product_version", None) or getattr(result, "agent_version", None)
|
||||
return {
|
||||
"title": title,
|
||||
"status": "success" if result.ok else "failed",
|
||||
"message": result.message,
|
||||
"stdout": getattr(result, "stdout", None) or "",
|
||||
"stderr": getattr(result, "stderr", None) or "",
|
||||
"output": _format_output(result),
|
||||
"ok": result.ok,
|
||||
"exit_code": getattr(result, "exit_code", None),
|
||||
"product_version": product_version,
|
||||
"target": getattr(result, "target", None) or "",
|
||||
"started_at": started_at,
|
||||
"finished_at": finished,
|
||||
}
|
||||
|
||||
|
||||
def _mark_running(db: Session, host: Host, *, title: str) -> str:
|
||||
started_at = _utcnow().isoformat()
|
||||
host.agent_update_state = "running"
|
||||
host.agent_update_last_error = None
|
||||
host.remote_action = {
|
||||
"title": title,
|
||||
"status": "running",
|
||||
"message": "Подключение к хосту и выполнение команды…",
|
||||
"stdout": "",
|
||||
"stderr": "",
|
||||
"output": "",
|
||||
"ok": None,
|
||||
"exit_code": None,
|
||||
"product_version": None,
|
||||
"target": host.hostname,
|
||||
"started_at": started_at,
|
||||
"finished_at": None,
|
||||
}
|
||||
db.flush()
|
||||
return started_at
|
||||
|
||||
|
||||
def _apply_ssh_update_result(db: Session, host: Host, result: SshCommandResult) -> None:
|
||||
now = _utcnow()
|
||||
host.agent_update_last_at = now
|
||||
if result.ok:
|
||||
host.agent_update_state = "success"
|
||||
host.agent_update_last_error = None
|
||||
if result.agent_version:
|
||||
host.product_version = result.agent_version
|
||||
host.ssh_admin_ok = True
|
||||
host.ssh_admin_checked_at = now
|
||||
else:
|
||||
host.agent_update_state = "failed"
|
||||
host.agent_update_last_error = (result.message or "SSH update failed")[:2000]
|
||||
|
||||
|
||||
def start_host_remote_action(
|
||||
db: Session,
|
||||
host: Host,
|
||||
*,
|
||||
title: str,
|
||||
runner: ActionRunner,
|
||||
) -> dict[str, Any]:
|
||||
host_id = int(host.id)
|
||||
with _lock:
|
||||
if host_id in _running or host.agent_update_state == "running":
|
||||
raise RemoteActionAlreadyRunningError(
|
||||
f"Remote action already running for host {host.hostname}"
|
||||
)
|
||||
_running.add(host_id)
|
||||
|
||||
started_at = _mark_running(db, host, title=title)
|
||||
db.commit()
|
||||
|
||||
def _worker() -> None:
|
||||
session = SessionLocal()
|
||||
try:
|
||||
row = session.get(Host, host_id)
|
||||
if row is None:
|
||||
return
|
||||
result = runner(session, row)
|
||||
started = (row.remote_action or {}).get("started_at") or started_at
|
||||
row.remote_action = _result_payload(result, title=title, started_at=started)
|
||||
if isinstance(result, SshCommandResult):
|
||||
_apply_ssh_update_result(session, row, result)
|
||||
session.commit()
|
||||
except Exception:
|
||||
logger.exception("Background remote action failed for host_id=%s", host_id)
|
||||
session.rollback()
|
||||
row = session.get(Host, host_id)
|
||||
if row is not None:
|
||||
started = (row.remote_action or {}).get("started_at") or started_at
|
||||
row.agent_update_state = "failed"
|
||||
row.agent_update_last_error = "Internal error during remote action"
|
||||
row.remote_action = {
|
||||
"title": title,
|
||||
"status": "failed",
|
||||
"message": "Внутренняя ошибка SAC при выполнении действия",
|
||||
"stdout": "",
|
||||
"stderr": "",
|
||||
"output": "",
|
||||
"ok": False,
|
||||
"exit_code": None,
|
||||
"product_version": None,
|
||||
"target": row.hostname,
|
||||
"started_at": started,
|
||||
"finished_at": _utcnow().isoformat(),
|
||||
}
|
||||
session.commit()
|
||||
finally:
|
||||
session.close()
|
||||
with _lock:
|
||||
_running.discard(host_id)
|
||||
|
||||
if _run_inline():
|
||||
_worker()
|
||||
else:
|
||||
threading.Thread(target=_worker, name=f"sac-remote-action-{host_id}", daemon=True).start()
|
||||
return dict(host.remote_action or {})
|
||||
|
||||
|
||||
def run_ssh_monitor_update_action(db: Session, host: Host) -> SshCommandResult:
|
||||
cfg = get_effective_linux_admin_config(db)
|
||||
if not cfg.configured:
|
||||
return SshCommandResult(
|
||||
ok=False,
|
||||
message="Linux SSH admin is not configured",
|
||||
target=host.hostname or "",
|
||||
)
|
||||
agent_cfg = get_effective_agent_update_config(db)
|
||||
repo_url = (agent_cfg.ssh_git_repo_url or "").strip() or None
|
||||
branch = (agent_cfg.git_branch or "main").strip() or "main"
|
||||
try:
|
||||
targets = iter_ssh_targets(host)
|
||||
except Exception as exc:
|
||||
return SshCommandResult(
|
||||
ok=False,
|
||||
message=str(exc),
|
||||
target=host.hostname or "",
|
||||
)
|
||||
last: SshCommandResult | None = None
|
||||
for target in targets:
|
||||
last = run_ssh_monitor_update(
|
||||
target=target,
|
||||
user=cfg.user,
|
||||
password=cfg.password,
|
||||
repo_url=repo_url,
|
||||
git_branch=branch,
|
||||
)
|
||||
if last.ok:
|
||||
return last
|
||||
return last or SshCommandResult(
|
||||
ok=False,
|
||||
message="SSH update failed",
|
||||
target=host.hostname or "",
|
||||
)
|
||||
|
||||
|
||||
def run_agent_fallback_action(db: Session, host: Host) -> AgentUpdateFallbackResult:
|
||||
return execute_agent_update_fallback(db, host)
|
||||
|
||||
|
||||
def get_remote_action_status(host: Host) -> dict[str, Any]:
|
||||
payload = dict(host.remote_action or {})
|
||||
if not payload:
|
||||
return {"active": False, "host_id": host.id}
|
||||
status = payload.get("status") or host.agent_update_state or "unknown"
|
||||
active = status == "running" or host.agent_update_state == "running"
|
||||
return {
|
||||
"active": active,
|
||||
"host_id": host.id,
|
||||
"hostname": host.hostname,
|
||||
"status": status,
|
||||
"title": payload.get("title") or "",
|
||||
"message": payload.get("message") or "",
|
||||
"stdout": payload.get("stdout") or "",
|
||||
"stderr": payload.get("stderr") or "",
|
||||
"output": payload.get("output") or "",
|
||||
"ok": payload.get("ok"),
|
||||
"exit_code": payload.get("exit_code"),
|
||||
"product_version": payload.get("product_version") or host.product_version,
|
||||
"target": payload.get("target") or host.hostname,
|
||||
"agent_update_state": host.agent_update_state,
|
||||
"started_at": payload.get("started_at"),
|
||||
"finished_at": payload.get("finished_at"),
|
||||
}
|
||||
Reference in New Issue
Block a user