Health Reporting

A pipeline that hangs -- a deadlocked thread, a worker that stopped feeding its queue, a source blocked on a dead connection -- still has a live process, so a supervisor that only watches the PID never restarts it. Pipeline.run() can report progress to the supervisor instead: it announces when the graph is ready, keeps a heartbeat as it strides, and announces when it is stopping.

Reporting is off unless a supervisor asks for it. Supervisors come in two kinds, and sgn.health speaks to both from one set of events:

Event	When	systemd (push)	HTTP (pull)
ready	graph checked, elements started, first stride next	READY=1	/readyz turns 200
alive	each stride	WATCHDOG=1	/healthz stays 200
stopping	run loop exiting, at end-of-stream or on failure	STOPPING=1	/readyz turns 503

The heartbeat comes from the main process's graph loop, so it stops when the loop stops -- including under parallelization, where a stalled worker stalls the main loop at a queue.

systemd

systemd sets NOTIFY_SOCKET for a Type=notify service, which enables the sd_notify(3) reporter:

[Service]
Type=notify
ExecStart=/usr/bin/python /opt/pipeline/run.py
WatchdogSec=30s
TimeoutStopSec=60s
Restart=on-failure

• READY=1 completes startup: systemctl start returns, and units ordered After= this one proceed, only once the graph is running.
• WATCHDOG=1 feeds WatchdogSec=. Heartbeats go out at half that interval (read from WATCHDOG_USEC), but only as the graph strides, so the timeout must exceed the longest stride -- including any time a source blocks waiting for input. When it lapses, systemd kills and restarts the service.
• STOPPING=1 tells systemd the pipeline is shutting down. systemd then treats the service as if it had sent SIGTERM: the watchdog is disarmed and TimeoutStopSec= starts, so a process that reaches end-of-stream but cannot exit (a thread that will not join, say) is killed rather than left hanging.
• Every message carries a STATUS= line, so systemctl status shows running, 12304 strides, last 0.4s ago.

Only the service's main process may notify unless NotifyAccess=all is set; if a wrapper script launches Python, exec the interpreter or set that.

The same protocol reaches into containers: Podman's default --sdnotify=container proxies the socket in, so a pipeline run by Podman or a Quadlet unit under systemd needs nothing further. An Apptainer instance started by a systemd unit can notify too, given NotifyAccess=all and the socket bound into the container (--bind /run/systemd/notify).

HTTP

Kubernetes, Docker and monitoring systems pull: they ask the process, on a schedule, whether it is fine. The pipeline answers on the HTTP server HTTPControl already provides:

Route	200 when	503 when
/readyz	between ready and stopping	starting, or stopping (draining)
/healthz	the loop strided within max_age (default 30 s)	the loop has gone stale

Both return a JSON snapshot:

{"status": "running", "ready": true, "alive": true, "strides": 12304,
 "last_stride_age_s": 0.4, "max_age_s": 30.0}

Enable it by setting SGN_HEALTH_PORT (and optionally SGN_HEALTH_HOST, default 0.0.0.0, and SGN_HEALTH_MAX_AGE), or by passing health=[HTTPHealth(port=8081)] to run(). Where the routes appear depends on the pipeline, and a manifest cannot tell the two apart:

• It already runs inside with HTTPControl(): p.run(). The routes appear on the control port, at the root and under the tag prefix; the port value is not needed.
• Otherwise the server is started on that port for the run alone -- no registry file, no signal handlers -- and stopped when run() returns.

A stuck loop leaves a stale stamp and gets a 503; a process wedged too hard to answer at all gets a probe timeout. A supervisor treats both as failure, which is the point. Before the first stride /healthz still answers 200, so a slow startup is not mistaken for a hang -- bound startup with a startup probe, as below.

Checks that must run a command rather than make a request need nothing beyond the interpreter: urllib.request.urlopen raises on a 503 and on a refused connection, so an unhandled exception is already the nonzero exit status such a check looks for.

Kubernetes

containers:
  - name: pipeline
    env:
      - name: SGN_HEALTH_PORT
        value: "8081"
    ports:
      - containerPort: 8081
    startupProbe:
      httpGet: {path: /readyz, port: 8081}
      periodSeconds: 5
      failureThreshold: 60      # up to 5 minutes to reach ready
    livenessProbe:
      httpGet: {path: /healthz, port: 8081}
      periodSeconds: 10

A pipeline using HTTPControl still sets SGN_HEALTH_PORT to enable reporting but is probed on its control port; set HTTPControl.port to something fixed. Add a readinessProbe on /readyz only if the pod sits behind a Service.

Docker and Compose

HEALTHCHECK runs a command inside the container, so localhost works:

ENV SGN_HEALTH_PORT=8081
HEALTHCHECK --interval=10s --timeout=5s \
    CMD python -c "import urllib.request as u; u.urlopen('http://localhost:8081/healthz')"

docker compose up --wait then returns once the pipeline is healthy, and depends_on: {pipeline: {condition: service_healthy}} orders other services after it. docker compose down sends SIGTERM; handle it with SignalEOS so the pipeline drains within stop_grace_period. Note that plain Docker only marks a container unhealthy -- it is Swarm, Compose ordering, or a watcher such as autoheal that acts on the status.

HTTPControl binds the host's resolved IP rather than every interface, so a HEALTHCHECK against a pipeline using it must target $(hostname), or set HTTPControl.host = "0.0.0.0".

Monitoring

Anything that can poll a URL can watch a pipeline: a Prometheus blackbox probe on /healthz, a Grafana alert on strides stalling, a cron job with curl -f URL || notify. This is the route for deployments with no supervisor at all, HTCondor jobs and Apptainer instances included.

Custom Reporters

Subclass sgn.health.HealthReporter, override the events you care about, and pass instances to run(health=...):

from collections import deque

from sgn import DequeSink, DequeSource, Pipeline
from sgn.health import HealthReporter


class EventLog(HealthReporter):
    heartbeat_interval = 5.0  # seconds between alive() calls

    def __init__(self):
        self.events = []

    def attach(self, state):
        self.state = state  # stamped by the pipeline as the run proceeds

    def ready(self):
        self.events.append("ready")

    def alive(self):
        self.events.append(f"alive after {self.state.strides} strides")

    def stopping(self):
        self.events.append("stopping")


src = DequeSource(name="src", source_pad_names=("H1",), iters={"H1": deque([1, 2, 3])})
snk = DequeSink(name="snk", sink_pad_names=("H1",))
p = Pipeline()
p.connect(src, snk)

log = EventLog()
p.run(health=[log])
assert log.events[0] == "ready" and log.events[-1] == "stopping"

An explicit health= list replaces whatever the environment would have selected; health=[] disables reporting altogether. Events arrive on the pipeline's event-loop thread in order: attach and ready once, alive repeatedly, stopping and detach once, also when the run fails. attach may raise to abort the run before startup -- that is how a port in use surfaces -- but after that a reporter that raises is logged (WARNING the first time, DEBUG while the failure persists, INFO on recovery) and never stops the pipeline.

A stamp file for a cron or Nagios check_file_age monitor is ten lines:

import os

from sgn.health import HealthReporter


class StampFile(HealthReporter):
    def __init__(self, path):
        self.path = path

    def ready(self):
        open(self.path, "a").close()

    def alive(self):
        os.utime(self.path)

    def stopping(self):
        os.unlink(self.path)