man/notify-selfcontained-example.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: MIT-0
   3 #
   4 # Implement the systemd notify protocol without external dependencies.
   5 # Supports both readiness notification on startup and on reloading,
   6 # according to the protocol defined at:
   7 # https://www.freedesktop.org/software/systemd/man/latest/sd_notify.html
   8 # This protocol is guaranteed to be stable as per:
   9 # https://systemd.io/PORTABILITY_AND_STABILITY/
  10
  11 import errno
  12 import os
  13 import signal
  14 import socket
  15 import sys
  16 import time
  17
  18 reloading = False
  19 terminating = False
  20
  21 def notify(message):
  22     if not message:
  23         raise ValueError("notify() requires a message")
  24
  25     socket_path = os.environ.get("NOTIFY_SOCKET")
  26     if not socket_path:
  27         return
  28
  29     if socket_path[0] not in ("/", "@"):
  30         raise OSError(errno.EAFNOSUPPORT, "Unsupported socket type")
  31
  32     # Handle abstract socket.
  33     if socket_path[0] == "@":
  34         socket_path = "\0" + socket_path[1:]
  35
  36     with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC) as sock:
  37         sock.connect(socket_path)
  38         sock.sendall(message)
  39
  40 def notify_ready():
  41     notify(b"READY=1")
  42
  43 def notify_reloading():
  44     microsecs = time.clock_gettime_ns(time.CLOCK_MONOTONIC) // 1000
  45     notify(f"RELOADING=1\nMONOTONIC_USEC={microsecs}".encode())
  46
  47 def notify_stopping():
  48     notify(b"STOPPING=1")
  49
  50 def reload(signum, frame):
  51     global reloading
  52     reloading = True
  53
  54 def terminate(signum, frame):
  55     global terminating
  56     terminating = True
  57
  58 def main():
  59     print("Doing initial setup")
  60     global reloading, terminating
  61
  62     # Set up signal handlers.
  63     print("Setting up signal handlers")
  64     signal.signal(signal.SIGHUP, reload)
  65     signal.signal(signal.SIGINT, terminate)
  66     signal.signal(signal.SIGTERM, terminate)
  67
  68     # Do any other setup work here.
  69
  70     # Once all setup is done, signal readiness.
  71     print("Done setting up")
  72     notify_ready()
  73
  74     print("Starting loop")
  75     while not terminating:
  76         if reloading:
  77             print("Reloading")
  78             reloading = False
  79
  80             # Support notifying the manager when reloading configuration.
  81             # This allows accurate state tracking as well as automatically
  82             # enabling 'systemctl reload' without needing to manually
  83             # specify an ExecReload= line in the unit file.
  84
  85             notify_reloading()
  86
  87             # Do some reconfiguration work here.
  88
  89             print("Done reloading")
  90             notify_ready()
  91
  92         # Do the real work here ...
  93
  94         print("Sleeping for five seconds")
  95         time.sleep(5)
  96
  97     print("Terminating")
  98     notify_stopping()
  99
 100 if __name__ == "__main__":
 101     sys.stdout.reconfigure(line_buffering=True)
 102     print("Starting app")
 103     main()
 104     print("Stopped app")