from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from multiprocessing.connection import Connection
from typing import Optional, TYPE_CHECKING, Any
__all__ = ["Status", "HealthReport", "Vitals", "Record"]
if TYPE_CHECKING:
from edri.abstract import ManagerBase
[docs]
class Status(Enum):
OK = auto()
WARNING = auto()
#: Alive and answering, but impaired — typically a dependency a component
#: needs (a database, an upstream API) is unreachable. Self-reported by a
#: component's :meth:`~edri.abstract.ManagerBase.health` probe; the Router
#: never infers it (it only escalates to WARNING/ERROR from silence).
DEGRADED = auto()
ERROR = auto()
[docs]
@dataclass
class HealthReport:
"""
A component's self-assessment, returned by its overridable ``health()`` probe.
Separates two questions the old single status conflated:
* **status** — is the component healthy, impaired, or broken? The base probe
derives this from caught exceptions; an override adds dependency checks
(e.g. ping the database, downgrade to ``DEGRADED`` if it is down).
* **ready** — can it accept traffic *right now*? A component that is alive
but still warming up or draining reports ``ready=False`` while staying
otherwise healthy.
Attributes:
status (Status):
Overall health. Defaults to OK.
ready (bool):
Whether the component is ready to accept work. Defaults to True.
reason (Optional[str]):
Human-readable explanation for a non-OK status or non-ready state,
surfaced on the health page. ``None`` when there is nothing to say.
"""
status: Status = Status.OK
ready: bool = True
reason: Optional[str] = None
[docs]
@dataclass
class Vitals:
"""
Operational metrics a component reports on every health-check round-trip.
The health check is a periodic two-way trip to every Manager that already
happens; these are the vitals that ride along on it for free. They describe
*load and progress*, complementing :class:`Status` (which only says whether
the component caught exceptions).
Some fields are point-in-time snapshots (``workers_running``,
``workers_oldest_age_seconds``, ``rss_bytes``, ``threads``, ``open_fds``),
others are deltas accumulated over the interval since the previous check
(``processed``, ``failed``, ``cpu_seconds``). The consumer interprets them
accordingly.
Attributes:
workers_running (int):
Workers currently alive (busy worker slots).
workers_pending (int):
Work queued because every worker slot is full. The component's
backpressure signal — a healthy component with a growing backlog is
an early warning before it falls behind.
workers_limit (int):
The ``max_workers`` ceiling. ``-1`` means unbounded.
workers_oldest_age_seconds (Optional[float]):
Age of the longest-running worker, or ``None`` when idle. A worker
that keeps growing old is how a deadlock surfaces while the main loop
still answers health checks — the heartbeat's truth signal.
processed (int):
Events the component dispatched to handlers since the previous health
check. Divided by the check interval, this is throughput.
failed (int):
Of those, how many left a FAILED response — the error rate per interval.
cpu_seconds (float):
CPU time (user + system) consumed since the previous check. Divided by
the interval, this is the process's CPU utilization.
rss_bytes (Optional[int]):
Resident memory of the component's process in bytes, or ``None`` when
it cannot be read on this platform.
rss_peak_bytes (Optional[int]):
Peak resident memory (high-water mark) in bytes, or ``None``.
threads (int):
Live threads in the process — a growing count flags a thread leak.
open_fds (Optional[int]):
Open file descriptors, or ``None`` when unreadable — an fd-leak detector.
pid (int):
The component process's PID, for correlation with external tools.
uptime_seconds (float):
Seconds since the component process entered its run loop. A small value
on a component that should be long-lived points at a crash-loop.
custom (dict[str, Any]):
Application-defined metrics a component adds by overriding
:meth:`~edri.abstract.ManagerBase.collect_custom_vitals`. Empty by
default. Values should be JSON-serializable primitives so they survive
both the pipe round-trip and the status endpoint's JSON encoding.
"""
workers_running: int = 0
workers_pending: int = 0
workers_limit: int = -1
workers_oldest_age_seconds: Optional[float] = None
processed: int = 0
failed: int = 0
cpu_seconds: float = 0.0
rss_bytes: Optional[int] = None
rss_peak_bytes: Optional[int] = None
threads: int = 0
open_fds: Optional[int] = None
pid: int = 0
uptime_seconds: float = 0.0
custom: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass
class Record:
"""
Represents the health status of a system component.
Attributes:
name (str):
The name of the component.
pipe (Optional[Connection]):
A connection object used for inter-process communication with the component.
Defaults to None if not applicable.
definition (Optional[ManagerBase]):
The class definition of the component, if available, providing metadata or
configuration details.
timestamp (datetime):
The timestamp of the last recorded health check. Defaults to the current time.
status (Status):
The current health status of the component, represented by a `Status` enum.
Defaults to `Status.WARNING`.
exceptions (list[tuple[str, dict, str, str, datetime]]):
A list of recorded exceptions related to the component. Each entry contains
the event name, event data, error message, traceback, and capture time.
vitals (Optional[Vitals]):
The operational metrics from the component's most recent health-check
response. ``None`` until the component has answered at least once.
health_rtt_seconds (Optional[float]):
Round-trip latency of the most recent health check, measured by the
Router (dispatch → component → back). Rises before a struggling
component goes silent, so it is a degradation signal the component
cannot report about itself. ``None`` until measured.
ready (bool):
Whether the component reported itself ready to accept traffic on its
last health check. Defaults to True.
reason (Optional[str]):
The component's explanation for a non-OK status or non-ready state.
"""
name: str
pipe: Connection | None = None
definition: Optional["ManagerBase"] = None
timestamp: datetime = datetime.now()
status: Status = Status.WARNING
exceptions: list[tuple[str, dict[str, Any], str, str, datetime]] = field(default_factory=list)
vitals: Optional[Vitals] = None
health_rtt_seconds: Optional[float] = None
ready: bool = True
reason: Optional[str] = None