| from __future__ import annotations |
|
|
| from collections.abc import Callable |
| from dataclasses import dataclass |
| import os |
| import signal |
|
|
| from .models import DashboardSnapshot |
| from .procfs import sample_process |
| from .snapshot import build_snapshot |
|
|
|
|
| SignalName = str |
| SnapshotBuilder = Callable[[], DashboardSnapshot] |
| SignalSender = Callable[[int, int], None] |
|
|
|
|
| SUPPORTED_SIGNALS: dict[SignalName, int] = { |
| "TERM": signal.SIGTERM, |
| "KILL": signal.SIGKILL, |
| } |
|
|
|
|
| @dataclass(frozen=True) |
| class ProcessActionRequest: |
| pid: int |
| process_key: str |
| signal_name: SignalName = "TERM" |
|
|
|
|
| @dataclass(frozen=True) |
| class ProcessActionResult: |
| ok: bool |
| status: str |
| message: str |
| pid: int |
| process_key: str |
| signal_name: SignalName |
|
|
|
|
| def send_signal_to_candidate( |
| request: ProcessActionRequest, |
| *, |
| snapshot_builder: SnapshotBuilder | None = None, |
| signal_sender: SignalSender = os.kill, |
| current_pid: int | None = None, |
| ) -> ProcessActionResult: |
| """Send a signal only if the PID still matches a current dry-run candidate.""" |
|
|
| current_pid = os.getpid() if current_pid is None else current_pid |
| snapshot_builder = snapshot_builder or build_snapshot |
| signal_number = SUPPORTED_SIGNALS.get(request.signal_name) |
|
|
| if signal_number is None: |
| return _result( |
| request, |
| ok=False, |
| status="invalid_signal", |
| message=f"Signal {request.signal_name} wird nicht unterstützt.", |
| ) |
|
|
| if request.pid == current_pid: |
| return _result( |
| request, |
| ok=False, |
| status="self_process", |
| message="Der Dashboard-Prozess selbst wird nicht beendet.", |
| ) |
|
|
| snapshot = snapshot_builder() |
| live = next((sample for sample in snapshot.processes if sample.pid == request.pid), None) |
| if live is None: |
| return _result( |
| request, |
| ok=False, |
| status="not_found", |
| message=( |
| f"PID {request.pid} läuft nicht mehr oder ist für den aktuellen User " |
| "nicht sichtbar." |
| ), |
| ) |
|
|
| if live.process_key != request.process_key: |
| return _result( |
| request, |
| ok=False, |
| status="pid_reused", |
| message=( |
| f"PID {request.pid} gehört inzwischen zu einem anderen Prozess. " |
| "Es wurde kein Signal gesendet." |
| ), |
| ) |
|
|
| candidate = next( |
| ( |
| item |
| for item in snapshot.dry_run.candidates |
| if item.pid == request.pid and item.process_key == request.process_key |
| ), |
| None, |
| ) |
| if candidate is None: |
| return _result( |
| request, |
| ok=False, |
| status="not_candidate", |
| message=( |
| f"PID {request.pid} ist kein aktueller Dry-run-Kandidat. " |
| "Es wurde kein Signal gesendet." |
| ), |
| ) |
|
|
| try: |
| signal_sender(request.pid, signal_number) |
| except ProcessLookupError: |
| return _result( |
| request, |
| ok=False, |
| status="not_found", |
| message=f"PID {request.pid} ist vor dem Signal verschwunden.", |
| ) |
| except PermissionError: |
| return _result( |
| request, |
| ok=False, |
| status="permission_denied", |
| message=f"Keine Berechtigung, Signal {request.signal_name} an PID {request.pid} zu senden.", |
| ) |
| except OSError as exc: |
| return _result( |
| request, |
| ok=False, |
| status="signal_failed", |
| message=f"Signal {request.signal_name} an PID {request.pid} ist fehlgeschlagen: {exc}", |
| ) |
|
|
| return _result( |
| request, |
| ok=True, |
| status="sent", |
| message=f"Signal {request.signal_name} wurde an PID {request.pid} gesendet.", |
| ) |
|
|
|
|
| def same_process_is_running( |
| pid: int, |
| process_key: str, |
| *, |
| proc_root: str = "/proc", |
| ) -> bool: |
| sample = sample_process(pid, proc_root=proc_root, current_uid_only=True) |
| return sample is not None and sample.process_key == process_key |
|
|
|
|
| def _result( |
| request: ProcessActionRequest, |
| *, |
| ok: bool, |
| status: str, |
| message: str, |
| ) -> ProcessActionResult: |
| return ProcessActionResult( |
| ok=ok, |
| status=status, |
| message=message, |
| pid=request.pid, |
| process_key=request.process_key, |
| signal_name=request.signal_name, |
| ) |
|
|
