dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
"""systemd: юниты, действия, journalctl."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List
|
|
|
|
from app.core.ssh import SSHClient
|
|
|
|
CRITICAL_UNITS = {"ssh.service", "sshd.service", "ssh.socket", "docker.service",
|
|
"networking.service", "systemd-networkd.service", "dbus.service"}
|
|
|
|
|
|
@dataclass
|
|
class Unit:
|
|
name: str
|
|
load: str
|
|
active: str
|
|
sub: str
|
|
description: str
|
|
enabled: str = ""
|
|
|
|
|
|
async def list_units(c: SSHClient) -> List[Unit]:
|
|
r = await c.sudo(
|
|
"systemctl list-units --type=service --all --no-legend --no-pager "
|
|
"--plain | awk '{$1=$1};1'", timeout=60
|
|
)
|
|
enabled_r = await c.sudo(
|
|
"systemctl list-unit-files --type=service --no-legend --no-pager", timeout=60
|
|
)
|
|
enabled_map = {}
|
|
for line in enabled_r.stdout.splitlines():
|
|
p = line.split()
|
|
if len(p) >= 2:
|
|
enabled_map[p[0]] = p[1]
|
|
|
|
units: List[Unit] = []
|
|
for line in r.stdout.splitlines():
|
|
p = line.split(None, 4)
|
|
if len(p) >= 4:
|
|
units.append(Unit(
|
|
name=p[0], load=p[1], active=p[2], sub=p[3],
|
|
description=p[4] if len(p) > 4 else "",
|
|
enabled=enabled_map.get(p[0], ""),
|
|
))
|
|
return units
|
|
|
|
|
|
async def failed_units(c: SSHClient) -> List[str]:
|
|
r = await c.sudo("systemctl --failed --no-legend --no-pager --plain", timeout=30)
|
|
return [l.split()[0] for l in r.stdout.splitlines() if l.strip() and not l.startswith("0 ")]
|
|
|
|
|
|
async def action(c: SSHClient, unit: str, act: str) -> str:
|
|
assert act in ("start", "stop", "restart", "reload", "enable", "disable", "mask", "unmask")
|
|
r = await c.sudo(f"systemctl {act} {unit}", timeout=120)
|
|
if not r.ok:
|
|
raise RuntimeError(r.stderr.strip() or f"rc={r.returncode}")
|
|
return r.stdout
|
|
|
|
|
|
async def logs(c: SSHClient, unit: str, lines: int = 200) -> str:
|
|
r = await c.sudo(f"journalctl -u {unit} -n {lines} --no-pager", timeout=60)
|
|
return r.stdout
|
|
|
|
|
|
async def cat_unit(c: SSHClient, unit: str) -> str:
|
|
r = await c.sudo(f"systemctl cat {unit}", timeout=30)
|
|
return r.stdout
|
|
|
|
|
|
async def daemon_reload(c: SSHClient) -> None:
|
|
await c.sudo("systemctl daemon-reload")
|