Skip to content

sgn.health

Pipeline health reporting for process supervisors.

A pipeline that hangs still has a live process, so a supervisor watching only the PID never restarts it. :meth:sgn.apps.Pipeline.run therefore reports progress, as three lifecycle events on the run loop:

ready The graph has been checked, every element's on_startup has run, and the first stride is about to execute. alive The graph loop is about to execute a stride. The heartbeat comes from the main process's loop, so under Parallelize a stalled worker (which stalls that loop at a queue) stops it, as intended. stopping The run loop is exiting, at end-of-stream or on failure.

Every event is stamped into one :class:HealthState, and each supervisor gets it through a transport that suits it. Supervisors come in two kinds:

  • Push. :class:Systemd speaks sd_notify(3): READY=1 / WATCHDOG=1 / STOPPING=1 datagrams (with a STATUS= line) to $NOTIFY_SOCKET. systemd sets that variable for Type=notify services; Podman proxies it into containers (--sdnotify=container).
  • Pull. :class:HTTPHealth serves GET /readyz and GET /healthz (200 or 503, JSON body) from :class:sgn.control.HTTPControl's server, for Kubernetes httpGet probes, Docker HEALTHCHECK, load balancers and monitoring. A pipeline already running inside with HTTPControl() gets the routes on its control port; one that is not can have the server started for the run alone.

Pipeline.run(health=None) selects transports from the environment (:func:from_environment): NOTIFY_SOCKET enables :class:Systemd and SGN_HEALTH_PORT enables :class:HTTPHealth. Custom transports subclass :class:HealthReporter. Once a run has started, reporting never raises into it: a failing reporter is logged and skipped.

HTTPHealth

Bases: HealthReporter


              flowchart TD
              sgn.health.HTTPHealth[HTTPHealth]
              sgn.health.HealthReporter[HealthReporter]

                              sgn.health.HealthReporter --> sgn.health.HTTPHealth
                


              click sgn.health.HTTPHealth href "" "sgn.health.HTTPHealth"
              click sgn.health.HealthReporter href "" "sgn.health.HealthReporter"

Serve GET /readyz and GET /healthz on HTTPControl's server.

/readyz answers 200 between ready and stopping and 503 otherwise (starting, or draining); /healthz answers 200 while the loop has strided within max_age and 503 once it goes stale. Both return the :meth:HealthState.snapshot as JSON. A stuck loop gives a stale stamp, a wedged process gives a timeout; a supervisor treats either as failure.

If the server is already running -- the pipeline runs inside with HTTPControl() -- the routes appear on its port and host and port here are not needed. Otherwise :meth:attach enters an HTTPControl(registry_file=None, signals=False) of its own on host:port for the run alone -- the server, without the registry file or signal handlers -- and :meth:detach exits it. address holds that server's (host, port) while it runs.

Parameters:

Name Type Description Default
port int | None

Port to serve on if the server is not already running; None or 0 lets the OS choose (logged, but of little use to a probe).

 None
host str

Interface to bind; 0.0.0.0 reaches both a kubelet probing the pod IP and a HEALTHCHECK hitting localhost.

 '0.0.0.0'
max_age float | None

Overrides HealthState.max_age for this run.

 None
Source code in src/sgn/health.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
class HTTPHealth(HealthReporter):
    """Serve ``GET /readyz`` and ``GET /healthz`` on ``HTTPControl``'s server.

    ``/readyz`` answers 200 between ``ready`` and ``stopping`` and 503
    otherwise (starting, or draining); ``/healthz`` answers 200 while the
    loop has strided within ``max_age`` and 503 once it goes stale. Both
    return the :meth:`HealthState.snapshot` as JSON. A stuck loop gives a
    stale stamp, a wedged process gives a timeout; a supervisor treats either
    as failure.

    If the server is already running -- the pipeline runs inside
    ``with HTTPControl()`` -- the routes appear on its port and ``host`` and
    ``port`` here are not needed. Otherwise :meth:`attach` enters an
    ``HTTPControl(registry_file=None, signals=False)`` of its own on
    ``host:port`` for the run alone -- the server, without the registry
    file or signal handlers -- and :meth:`detach` exits it. ``address``
    holds that server's ``(host, port)`` while it runs.

    Args:
        port: Port to serve on if the server is not already running; ``None``
            or ``0`` lets the OS choose (logged, but of little use to a probe).
        host: Interface to bind; ``0.0.0.0`` reaches both a kubelet probing
            the pod IP and a ``HEALTHCHECK`` hitting ``localhost``.
        max_age: Overrides ``HealthState.max_age`` for this run.
    """

    heartbeat_interval = 0.0

    def __init__(
        self,
        port: int | None = None,
        host: str = "0.0.0.0",  # noqa: S104 -- probes come from outside
        max_age: float | None = None,
    ) -> None:
        self.port = port
        self.host = host
        self.max_age = max_age
        self.address: tuple[str, int] | None = None
        self._control: HTTPControl | None = None

    def attach(self, state: HealthState) -> None:
        # control imports the sgn package, so it cannot be imported at module
        # level from here (apps imports this module).
        from sgn.control import HTTPControl

        if self.max_age is not None:
            state.max_age = self.max_age
        if _http_control_serving():
            logger.info(
                "health routes at /readyz and /healthz on the HTTPControl server"
            )
        else:
            control = HTTPControl(
                registry_file=None, host=self.host, port=self.port or 0, signals=False
            )
            control.__enter__()
            self._control = control
            self.address = (self.host, control.bound_port)
            logger.info(
                "health routes at http://%s:%d/readyz and /healthz", *self.address
            )
        HTTPControl.health = state

    def detach(self) -> None:
        from sgn.control import HTTPControl

        HTTPControl.health = None
        if self._control is not None:
            self._control.__exit__(None, None, None)
            self._control = None
        self.address = None

    def __repr__(self) -> str:
        return (
            f"{type(self).__name__}(port={self.port!r}, host={self.host!r}, "
            f"max_age={self.max_age!r})"
        )

HealthReporter

Base class for a health transport; override what the supervisor needs.

The pipeline calls these from its event-loop thread, in order: :meth:attach and :meth:ready once, :meth:alive repeatedly, :meth:stopping and :meth:detach once (also when the run fails). :meth:attach may raise to abort the run before startup (a port in use, say); the others may raise too, but the pipeline logs that and continues.

Source code in src/sgn/health.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class HealthReporter:
    """Base class for a health transport; override what the supervisor needs.

    The pipeline calls these from its event-loop thread, in order:
    :meth:`attach` and :meth:`ready` once, :meth:`alive` repeatedly,
    :meth:`stopping` and :meth:`detach` once (also when the run fails).
    :meth:`attach` may raise to abort the run before startup (a port in use,
    say); the others may raise too, but the pipeline logs that and continues.
    """

    heartbeat_interval: float = 1.0
    """Minimum seconds between :meth:`alive` calls; ``0`` reports every stride."""

    def attach(self, state: HealthState) -> None:
        """The run is about to start; ``state`` will be stamped as it goes."""

    def ready(self) -> None:
        """The pipeline has started and is about to execute its first stride."""

    def alive(self) -> None:
        """The pipeline is about to execute a stride (a liveness heartbeat)."""

    def stopping(self) -> None:
        """The pipeline's run loop is exiting (end-of-stream or failure)."""

    def detach(self) -> None:
        """``run()`` is returning; release anything :meth:`attach` acquired."""

heartbeat_interval = 1.0 class-attribute instance-attribute

Minimum seconds between :meth:alive calls; 0 reports every stride.

alive()

The pipeline is about to execute a stride (a liveness heartbeat).

Source code in src/sgn/health.py 
139
140
def alive(self) -> None:
    """The pipeline is about to execute a stride (a liveness heartbeat)."""

attach(state)

The run is about to start; state will be stamped as it goes.

Source code in src/sgn/health.py 
133
134
def attach(self, state: HealthState) -> None:
    """The run is about to start; ``state`` will be stamped as it goes."""

detach()

run() is returning; release anything :meth:attach acquired.

Source code in src/sgn/health.py 
145
146
def detach(self) -> None:
    """``run()`` is returning; release anything :meth:`attach` acquired."""

ready()

The pipeline has started and is about to execute its first stride.

Source code in src/sgn/health.py 
136
137
def ready(self) -> None:
    """The pipeline has started and is about to execute its first stride."""

stopping()

The pipeline's run loop is exiting (end-of-stream or failure).

Source code in src/sgn/health.py 
142
143
def stopping(self) -> None:
    """The pipeline's run loop is exiting (end-of-stream or failure)."""

HealthState dataclass

What the run loop has done so far; stamped by :class:PipelineHealth.

Timestamps are time.monotonic() values. max_age is how long the loop may go without a stride before :meth:is_alive turns false.

Source code in src/sgn/health.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@dataclass
class HealthState:
    """What the run loop has done so far; stamped by :class:`PipelineHealth`.

    Timestamps are ``time.monotonic()`` values. ``max_age`` is how long the
    loop may go without a stride before :meth:`is_alive` turns false.
    """

    max_age: float = 30.0
    ready_at: float | None = None
    last_alive: float | None = None
    strides: int = 0
    stopping_at: float | None = None

    def age(self, now: float | None = None) -> float | None:
        """Seconds since the last stride, or ``None`` before the first."""
        if self.last_alive is None:
            return None
        return (time.monotonic() if now is None else now) - self.last_alive

    def is_ready(self) -> bool:
        """Between ``ready`` and ``stopping``: fit to receive work."""
        return self.ready_at is not None and self.stopping_at is None

    def is_alive(self, now: float | None = None) -> bool:
        """The loop has strided within ``max_age``.

        Counts as alive before the first stride, so a slow startup is not
        mistaken for a hang; bound startup with a startup probe or
        ``TimeoutStartSec=`` instead. After ``stopping`` the stamp goes
        stale on its own, which is how a hung shutdown gets noticed.
        """
        age = self.age(now)
        if age is None:
            return self.stopping_at is None
        return age < self.max_age

    def status(self, now: float | None = None) -> str:
        """``starting``, ``running``, ``stale`` or ``stopping``."""
        if self.stopping_at is not None:
            return "stopping"
        if self.ready_at is None:
            return "starting"
        return "running" if self.is_alive(now) else "stale"

    def summary(self, now: float | None = None) -> str:
        """One line for humans: ``running, 12304 strides, last 0.4s ago``."""
        parts = [self.status(now), f"{self.strides} strides"]
        age = self.age(now)
        if age is not None:
            parts.append(f"last {age:.1f}s ago")
        return ", ".join(parts)

    def snapshot(self, now: float | None = None) -> dict:
        """The JSON body of the HTTP routes."""
        age = self.age(now)
        return {
            "status": self.status(now),
            "ready": self.is_ready(),
            "alive": self.is_alive(now),
            "strides": self.strides,
            "last_stride_age_s": None if age is None else round(age, 3),
            "max_age_s": self.max_age,
        }

age(now=None)

Seconds since the last stride, or None before the first.

Source code in src/sgn/health.py 
68
69
70
71
72
def age(self, now: float | None = None) -> float | None:
    """Seconds since the last stride, or ``None`` before the first."""
    if self.last_alive is None:
        return None
    return (time.monotonic() if now is None else now) - self.last_alive

is_alive(now=None)

The loop has strided within max_age.

Counts as alive before the first stride, so a slow startup is not mistaken for a hang; bound startup with a startup probe or TimeoutStartSec= instead. After stopping the stamp goes stale on its own, which is how a hung shutdown gets noticed.

Source code in src/sgn/health.py 
78
79
80
81
82
83
84
85
86
87
88
89
def is_alive(self, now: float | None = None) -> bool:
    """The loop has strided within ``max_age``.

    Counts as alive before the first stride, so a slow startup is not
    mistaken for a hang; bound startup with a startup probe or
    ``TimeoutStartSec=`` instead. After ``stopping`` the stamp goes
    stale on its own, which is how a hung shutdown gets noticed.
    """
    age = self.age(now)
    if age is None:
        return self.stopping_at is None
    return age < self.max_age

is_ready()

Between ready and stopping: fit to receive work.

Source code in src/sgn/health.py 
74
75
76
def is_ready(self) -> bool:
    """Between ``ready`` and ``stopping``: fit to receive work."""
    return self.ready_at is not None and self.stopping_at is None

snapshot(now=None)

The JSON body of the HTTP routes.

Source code in src/sgn/health.py 
107
108
109
110
111
112
113
114
115
116
117
def snapshot(self, now: float | None = None) -> dict:
    """The JSON body of the HTTP routes."""
    age = self.age(now)
    return {
        "status": self.status(now),
        "ready": self.is_ready(),
        "alive": self.is_alive(now),
        "strides": self.strides,
        "last_stride_age_s": None if age is None else round(age, 3),
        "max_age_s": self.max_age,
    }

status(now=None)

starting, running, stale or stopping.

Source code in src/sgn/health.py 
91
92
93
94
95
96
97
def status(self, now: float | None = None) -> str:
    """``starting``, ``running``, ``stale`` or ``stopping``."""
    if self.stopping_at is not None:
        return "stopping"
    if self.ready_at is None:
        return "starting"
    return "running" if self.is_alive(now) else "stale"

summary(now=None)

One line for humans: running, 12304 strides, last 0.4s ago.

Source code in src/sgn/health.py 
 99
100
101
102
103
104
105
def summary(self, now: float | None = None) -> str:
    """One line for humans: ``running, 12304 strides, last 0.4s ago``."""
    parts = [self.status(now), f"{self.strides} strides"]
    age = self.age(now)
    if age is not None:
        parts.append(f"last {age:.1f}s ago")
    return ", ".join(parts)

PipelineHealth

Bases: HealthReporter


              flowchart TD
              sgn.health.PipelineHealth[PipelineHealth]
              sgn.health.HealthReporter[HealthReporter]

                              sgn.health.HealthReporter --> sgn.health.PipelineHealth
                


              click sgn.health.PipelineHealth href "" "sgn.health.PipelineHealth"
              click sgn.health.HealthReporter href "" "sgn.health.HealthReporter"

The pipeline's side of the seam: one state, many reporters.

Stamps :attr:state on every event, then delivers the event to each reporter -- :meth:alive throttled to the reporter's heartbeat_interval. :meth:attach propagates a failure (and detaches what was attached) so a misconfigured transport aborts the run before startup; every later event isolates the pipeline from reporter failures, logging them at WARNING the first time, DEBUG while they persist, and INFO on recovery.

Source code in src/sgn/health.py 
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
class PipelineHealth(HealthReporter):
    """The pipeline's side of the seam: one state, many reporters.

    Stamps :attr:`state` on every event, then delivers the event to each
    reporter -- :meth:`alive` throttled to the reporter's
    ``heartbeat_interval``. :meth:`attach` propagates a failure (and detaches
    what was attached) so a misconfigured transport aborts the run before
    startup; every later event isolates the pipeline from reporter failures,
    logging them at WARNING the first time, DEBUG while they persist, and
    INFO on recovery.
    """

    heartbeat_interval = 0.0

    def __init__(
        self, reporters: Iterable[HealthReporter] = (), state: HealthState | None = None
    ) -> None:
        self._slots = [_Slot(reporter) for reporter in reporters]
        self.state = HealthState() if state is None else state

    @classmethod
    def resolve(
        cls, health: HealthReporter | Iterable[HealthReporter] | None
    ) -> PipelineHealth:
        """Build from ``Pipeline.run``'s ``health`` argument.

        ``None`` consults the environment (:func:`from_environment`). A single
        :class:`HealthReporter` or an iterable of them is used as given, so an
        empty iterable disables reporting whatever the environment says.
        """
        if health is None:
            reporters = from_environment()
        elif isinstance(health, HealthReporter):
            reporters = [health]
        else:
            reporters = list(health)
        for reporter in reporters:
            if not isinstance(reporter, HealthReporter):
                raise TypeError(
                    f"health reporters must be sgn.health.HealthReporter instances, "
                    f"got {reporter!r}"
                )
            logger.info("reporting pipeline health to %r", reporter)
        return cls(reporters)

    @property
    def reporters(self) -> list[HealthReporter]:
        return [slot.reporter for slot in self._slots]

    def __bool__(self) -> bool:
        return bool(self._slots)

    def attach(self, state: HealthState | None = None) -> None:
        if state is not None:
            self.state = state
        for slot in self._slots:
            try:
                slot.reporter.attach(self.state)
            except BaseException:
                self.detach()
                raise
            slot.attached = True

    def ready(self) -> None:
        self.state.ready_at = time.monotonic()
        for slot in self._slots:
            self._deliver(slot, "ready")

    def alive(self) -> None:
        now = time.monotonic()
        self.state.last_alive = now
        self.state.strides += 1
        for slot in self._slots:
            if now - slot.last_alive >= slot.reporter.heartbeat_interval:
                slot.last_alive = now
                self._deliver(slot, "alive")

    def stopping(self) -> None:
        self.state.stopping_at = time.monotonic()
        for slot in self._slots:
            self._deliver(slot, "stopping")

    def detach(self) -> None:
        for slot in self._slots:
            if slot.attached:
                slot.attached = False
                self._deliver(slot, "detach")

    def _deliver(self, slot: _Slot, event: str) -> None:
        try:
            getattr(slot.reporter, event)()
        except Exception as exc:  # noqa: BLE001 -- must never take down the run
            level = logging.DEBUG if slot.failing else logging.WARNING
            slot.failing = True
            logger.log(
                level, "health reporter %r failed on %s: %s", slot.reporter, event, exc
            )
            return
        if slot.failing:
            slot.failing = False
            logger.info("health reporter %r recovered", slot.reporter)

resolve(health) classmethod

Build from Pipeline.run's health argument.

None consults the environment (:func:from_environment). A single :class:HealthReporter or an iterable of them is used as given, so an empty iterable disables reporting whatever the environment says.

Source code in src/sgn/health.py 
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
@classmethod
def resolve(
    cls, health: HealthReporter | Iterable[HealthReporter] | None
) -> PipelineHealth:
    """Build from ``Pipeline.run``'s ``health`` argument.

    ``None`` consults the environment (:func:`from_environment`). A single
    :class:`HealthReporter` or an iterable of them is used as given, so an
    empty iterable disables reporting whatever the environment says.
    """
    if health is None:
        reporters = from_environment()
    elif isinstance(health, HealthReporter):
        reporters = [health]
    else:
        reporters = list(health)
    for reporter in reporters:
        if not isinstance(reporter, HealthReporter):
            raise TypeError(
                f"health reporters must be sgn.health.HealthReporter instances, "
                f"got {reporter!r}"
            )
        logger.info("reporting pipeline health to %r", reporter)
    return cls(reporters)

Systemd

Bases: HealthReporter


              flowchart TD
              sgn.health.Systemd[Systemd]
              sgn.health.HealthReporter[HealthReporter]

                              sgn.health.HealthReporter --> sgn.health.Systemd
                


              click sgn.health.Systemd href "" "sgn.health.Systemd"
              click sgn.health.HealthReporter href "" "sgn.health.HealthReporter"

Report to systemd over the sd_notify(3) notification socket.

ready sends READY=1, completing startup of a Type=notify service. alive sends WATCHDOG=1, feeding WatchdogSec=: the heartbeat only goes out when the graph strides, so a stride slower than that timeout (including time a source blocks on input) gets the service killed and restarted. stopping sends STOPPING=1; systemd then disarms the watchdog and starts TimeoutStopSec=, as if it had sent SIGTERM, so a process that reaches end-of-stream but cannot exit -- a thread that will not join, say -- is killed rather than left hanging. Each message carries a STATUS= line for systemctl status.

Parameters:

Name Type Description Default
socket_path str

The notification socket as systemd passes it in NOTIFY_SOCKET: an absolute filesystem path, or @name for a Linux abstract socket.

 required
heartbeat_interval float

Seconds between WATCHDOG=1 messages. :meth:from_environment derives this from WATCHDOG_USEC.

 1.0
Source code in src/sgn/health.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class Systemd(HealthReporter):
    """Report to systemd over the sd_notify(3) notification socket.

    ``ready`` sends ``READY=1``, completing startup of a ``Type=notify``
    service. ``alive`` sends ``WATCHDOG=1``, feeding ``WatchdogSec=``: the
    heartbeat only goes out when the graph strides, so a stride slower than
    that timeout (including time a source blocks on input) gets the service
    killed and restarted. ``stopping`` sends ``STOPPING=1``; systemd then
    disarms the watchdog and starts ``TimeoutStopSec=``, as if it had sent
    SIGTERM, so a process that reaches end-of-stream but cannot exit -- a
    thread that will not join, say -- is killed rather than left hanging.
    Each message carries a ``STATUS=`` line for ``systemctl status``.

    Args:
        socket_path:
            The notification socket as systemd passes it in ``NOTIFY_SOCKET``:
            an absolute filesystem path, or ``@name`` for a Linux abstract
            socket.
        heartbeat_interval:
            Seconds between ``WATCHDOG=1`` messages. :meth:`from_environment`
            derives this from ``WATCHDOG_USEC``.
    """

    def __init__(self, socket_path: str, heartbeat_interval: float = 1.0) -> None:
        if not socket_path or socket_path[0] not in ("/", "@"):
            raise ValueError(f"unsupported NOTIFY_SOCKET path {socket_path!r}")
        self.socket_path = socket_path
        # A leading '@' names an abstract socket, whose address is the name
        # prefixed with a NUL byte.
        if socket_path[0] == "@":
            self._address = "\0" + socket_path[1:]
        else:
            self._address = socket_path
        self.heartbeat_interval = heartbeat_interval
        self._state: HealthState | None = None

    @classmethod
    def from_environment(
        cls, environ: Mapping[str, str] | None = None
    ) -> Systemd | None:
        """Build from the variables systemd sets for a service, or ``None``.

        Requires ``NOTIFY_SOCKET``. When ``WATCHDOG_USEC`` is also set
        (systemd exports the ``WatchdogSec=`` timeout in microseconds),
        heartbeats go out at half that interval, as sd_watchdog_enabled(3)
        recommends; otherwise once per second.
        """
        env = os.environ if environ is None else environ
        socket_path = env.get("NOTIFY_SOCKET")
        if not socket_path:
            return None
        heartbeat_interval = 1.0
        usec = env.get("WATCHDOG_USEC")
        if usec:
            try:
                heartbeat_interval = int(usec) / 2e6
            except ValueError:
                logger.warning("ignoring unparseable WATCHDOG_USEC=%r", usec)
        return cls(socket_path, heartbeat_interval=heartbeat_interval)

    def attach(self, state: HealthState) -> None:
        self._state = state

    def ready(self) -> None:
        self._send("READY=1")

    def alive(self) -> None:
        self._send("WATCHDOG=1")

    def stopping(self) -> None:
        self._send("STOPPING=1")

    def detach(self) -> None:
        self._state = None

    def _send(self, state_line: str) -> None:
        message = state_line
        if self._state is not None:
            message += f"\nSTATUS={self._state.summary()}"
        # A fresh socket per message, as the sd_notify(3) reference does; it
        # is close-on-exec by default (PEP 446). Non-blocking, so a supervisor
        # that stops draining its socket can never stall the event loop.
        with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
            sock.settimeout(0)
            sock.connect(self._address)
            sock.send(message.encode())

    def __repr__(self) -> str:
        return (
            f"{type(self).__name__}({self.socket_path!r}, "
            f"heartbeat_interval={self.heartbeat_interval})"
        )

from_environment(environ=None) classmethod

Build from the variables systemd sets for a service, or None.

Requires NOTIFY_SOCKET. When WATCHDOG_USEC is also set (systemd exports the WatchdogSec= timeout in microseconds), heartbeats go out at half that interval, as sd_watchdog_enabled(3) recommends; otherwise once per second.

Source code in src/sgn/health.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
@classmethod
def from_environment(
    cls, environ: Mapping[str, str] | None = None
) -> Systemd | None:
    """Build from the variables systemd sets for a service, or ``None``.

    Requires ``NOTIFY_SOCKET``. When ``WATCHDOG_USEC`` is also set
    (systemd exports the ``WatchdogSec=`` timeout in microseconds),
    heartbeats go out at half that interval, as sd_watchdog_enabled(3)
    recommends; otherwise once per second.
    """
    env = os.environ if environ is None else environ
    socket_path = env.get("NOTIFY_SOCKET")
    if not socket_path:
        return None
    heartbeat_interval = 1.0
    usec = env.get("WATCHDOG_USEC")
    if usec:
        try:
            heartbeat_interval = int(usec) / 2e6
        except ValueError:
            logger.warning("ignoring unparseable WATCHDOG_USEC=%r", usec)
    return cls(socket_path, heartbeat_interval=heartbeat_interval)

from_environment(environ=None)

The reporters the environment asks for; empty when it asks for none.

NOTIFY_SOCKET (set by systemd for Type=notify services) selects :class:Systemd. SGN_HEALTH_PORT selects :class:HTTPHealth, with SGN_HEALTH_HOST (default 0.0.0.0) and SGN_HEALTH_MAX_AGE (seconds, default 30). Both may be set.

Source code in src/sgn/health.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def from_environment(environ: Mapping[str, str] | None = None) -> list[HealthReporter]:
    """The reporters the environment asks for; empty when it asks for none.

    ``NOTIFY_SOCKET`` (set by systemd for ``Type=notify`` services) selects
    :class:`Systemd`. ``SGN_HEALTH_PORT`` selects :class:`HTTPHealth`, with
    ``SGN_HEALTH_HOST`` (default ``0.0.0.0``) and ``SGN_HEALTH_MAX_AGE``
    (seconds, default 30). Both may be set.
    """
    env = os.environ if environ is None else environ
    reporters: list[HealthReporter] = []
    systemd = Systemd.from_environment(env)
    if systemd is not None:
        reporters.append(systemd)
    port = env.get("SGN_HEALTH_PORT")
    if port:
        max_age = env.get("SGN_HEALTH_MAX_AGE")
        reporters.append(
            HTTPHealth(
                port=int(port),
                host=env.get("SGN_HEALTH_HOST") or "0.0.0.0",  # noqa: S104
                max_age=float(max_age) if max_age else None,
            )
        )
    return reporters