from typing import Optional from .monitor_heartbeat_types import HeartbeatStatus def _whole(name: str, v: object) -> None: if not isinstance(v, int) or isinstance(v, bool): raise ValueError("%s must be whole seconds, received %r" % (name, v)) def heartbeat_status(last_ping_at: Optional[int], now: int, interval_seconds: int, grace_seconds: int) -> HeartbeatStatus: """New, up, late or down, as healthchecks.io names them. Up through the due second itself, late through the last second of the grace time, down after. A ping stamped in the future (clock skew between the job and the monitor) is simply up: its due time is later still.""" if last_ping_at is not None: _whole("lastPingAt", last_ping_at) _whole("now", now) _whole("intervalSeconds", interval_seconds) _whole("graceSeconds", grace_seconds) if interval_seconds < 1: raise ValueError("intervalSeconds must be at least 1, received %d" % interval_seconds) if grace_seconds < 0: raise ValueError("graceSeconds must not be negative, received %d" % grace_seconds) if last_ping_at is None: return HeartbeatStatus(state="new", due_at=None, down_at=None, overdue_seconds=0) due_at = last_ping_at + interval_seconds down_at = due_at + grace_seconds if now <= due_at: return HeartbeatStatus(state="up", due_at=due_at, down_at=down_at, overdue_seconds=0) return HeartbeatStatus(state="late" if now <= down_at else "down", due_at=due_at, down_at=down_at, overdue_seconds=now - due_at)