selfprivacy-rest-api/selfprivacy_api/utils/systemd.py

84 lines
2.6 KiB
Python
Raw Normal View History

"""Generic service status fetcher using systemctl"""
2024-07-26 19:59:44 +00:00
import subprocess
from typing import List
from selfprivacy_api.models.services import ServiceStatus
def get_service_status(unit: str) -> ServiceStatus:
"""
Return service status from systemd.
Use systemctl show to get the status of a service.
Get ActiveState from the output.
"""
service_status = subprocess.check_output(["systemctl", "show", unit])
if b"LoadState=not-found" in service_status:
return ServiceStatus.OFF
if b"ActiveState=active" in service_status:
return ServiceStatus.ACTIVE
if b"ActiveState=inactive" in service_status:
return ServiceStatus.INACTIVE
if b"ActiveState=activating" in service_status:
return ServiceStatus.ACTIVATING
if b"ActiveState=deactivating" in service_status:
return ServiceStatus.DEACTIVATING
if b"ActiveState=failed" in service_status:
return ServiceStatus.FAILED
if b"ActiveState=reloading" in service_status:
return ServiceStatus.RELOADING
return ServiceStatus.OFF
def get_service_status_from_several_units(services: list[str]) -> ServiceStatus:
"""
Fetch all service statuses for all services and return the worst status.
Statuses from worst to best:
- OFF
- FAILED
- RELOADING
- ACTIVATING
- DEACTIVATING
- INACTIVE
- ACTIVE
"""
service_statuses = []
for service in services:
service_statuses.append(get_service_status(service))
if ServiceStatus.OFF in service_statuses:
return ServiceStatus.OFF
if ServiceStatus.FAILED in service_statuses:
return ServiceStatus.FAILED
if ServiceStatus.RELOADING in service_statuses:
return ServiceStatus.RELOADING
if ServiceStatus.ACTIVATING in service_statuses:
return ServiceStatus.ACTIVATING
if ServiceStatus.DEACTIVATING in service_statuses:
return ServiceStatus.DEACTIVATING
if ServiceStatus.INACTIVE in service_statuses:
return ServiceStatus.INACTIVE
if ServiceStatus.ACTIVE in service_statuses:
return ServiceStatus.ACTIVE
return ServiceStatus.OFF
def get_last_log_lines(service: str, lines_count: int) -> List[str]:
if lines_count < 1:
raise ValueError("lines_count must be greater than 0")
try:
logs = subprocess.check_output(
[
"journalctl",
"-u",
service,
"-n",
str(lines_count),
"-o",
"cat",
],
shell=False,
).decode("utf-8")
return logs.splitlines()
except subprocess.CalledProcessError:
return []