HeartbeatChecker Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

class core.heartbeatchecker.HeartbeatChecker(*args: Any, **kwargs: Any)

Bases: BaseSatelliteFrame

Checks periodically Satellites’ state via subscription to its Heartbeat.

Individual heartbeat checks run in separate threads. In case of a failure (either the Satellite is in ERROR/SAFE state or has missed several heartbeats) the corresponding thread will set a failed event. Alternatively, an action can be triggered via a callback.

HB_INIT_LIVES = 3

HB_INIT_PERIOD = 2000

close() → None

property fail_events: dict[str, Event]

Return a dictionary of Events triggered for failed Satellites.

get_failed() → list[str]

Get a list of the names of all failed Satellites.

heartbeat_host_is_registered(host: UUID) → bool

Check whether a given Satellite is already registered.

heartbeat_name_is_registered(name: str) → bool

Check whether a given Satellite is already registered.

property heartbeat_states: dict[str, SatelliteState]

Return a dictionary of the monitored Satellites’ state.

register_heartbeat_callback(callback: Callable[[str, SatelliteState], None] | None = None) → None

register_heartbeat_host(host: UUID, address: str, name: str = '', context: Context | None = None) → Event

Register a heartbeat check for a specific Satellite.

Returns threading.Event that will be set when a failure occurs.

unregister_heartbeat_host(host: UUID) → None

Unregister a heartbeat check for a specific Satellite.

class core.heartbeatchecker.HeartbeatState(host: UUID, name: str, evt: Event, lives: int, interval: int, role: CHPRole = CHPRole.DYNAMIC)

Bases: object

refresh(ts: datetime | None = None) → None

property seconds_since_refresh: float