modules/python-http-handlers: better daemon handling

This commit extends the python http handler template with the following
things:

- Integration with systemd notify
- Named systemd sockets, to avoid depending on expected FD number
- Graceful and customizable handling of SIGHUP reloads
- Graceful shutdown on SIGTERM with timeout
- Multithreaded HTTP server, one thread per request
- Better error message on undefined handler

base/default.nix

@@ -30,9 +30,9 @@
     ./services/journald-upload.nix
     ./services/logrotate.nix
     ./services/nginx.nix
-    ./services/nullmailer.nix
     ./services/openssh.nix
     ./services/polkit.nix
+    ./services/postfix.nix
     ./services/prometheus-flake-input-exporter.nix
     ./services/prometheus-node-exporter.nix
     ./services/prometheus-systemd-exporter.nix

base/services/irqbalance.nix

@@ -1,22 +1,4 @@
-{ config, pkgs, lib, ... }:
-let
-  cfg = config.services.irqbalance;
-in
+{ ... }:
 {
   services.irqbalance.enable = true;
-
-  # irqbalance only has meaningful work to do on multi-socket machines, so
-  # skip starting it pointlessly everywhere else.
-  systemd.services.irqbalance.serviceConfig.ExecCondition = let
-    isMultiSocket = pkgs.writeShellApplication {
-      name = "irqbalance-is-multi-socket";
-      runtimeInputs = [ pkgs.coreutils ];
-      text = ''
-        sockets=$(cat /sys/devices/system/cpu/cpu*/topology/physical_package_id | sort -u | wc -l)
-        [ "$sockets" -gt 1 ]
-      '';
-    };
-  in lib.mkIf cfg.enable [
-    (lib.getExe isMultiSocket)
-  ];
-}
+}

base/services/nullmailer.nix

@@ -1,15 +0,0 @@
-{ config, lib, ... }:
-{
-  services.nullmailer = {
-    enable = true;
-
-    config = {
-      adminaddr = "root@pvv.ntnu.no";
-      defaultdomain = "pvv.ntnu.no";
-      defaulthost = "pvv.ntnu.no";
-
-      me = lib.mkDefault config.networking.fqdn;
-      remotes = lib.mkDefault "smtp.pvv.ntnu.no smtp port=587 starttls";
-    };
-  };
-}

base/services/postfix.nix

@@ -0,0 +1,22 @@
+{ config, pkgs, lib, ... }:
+let
+  cfg = config.services.postfix;
+in
+{
+  services.postfix = {
+    enable = true;
+
+    settings.main = {
+      myhostname = "${config.networking.hostName}.pvv.ntnu.no";
+      mydomain = "pvv.ntnu.no";
+
+      # Nothing should be delivered to this machine
+      mydestination = [ ];
+
+      relayhost = [ "smtp.pvv.ntnu.no:465" ];
+
+      smtp_tls_wrappermode = "yes";
+      smtp_tls_security_level = "encrypt";
+    };
+  };
+}

hosts/bicep/services/postgresql/default.nix

@@ -14,7 +14,6 @@ in
     extensions = ps: with ps; [ pg_repack ];
     enableTCPIP = true;
 
-    # NOTE: md5 accepts both md5 and scram-sha-256
     authentication = ''
       host all all ${values.ipv4-space} md5
       host all all ${values.ipv6-space} md5
@@ -77,15 +76,6 @@ in
       maintenance_io_concurrency = 100;
       wal_recycle = true;
 
-      # -------------------------------- #
-
-      # Authentication
-      password_encryption = "scram-sha-256";
-
-      # Logging
-      log_connections = "authorization";
-      log_disconnections = true;
-
       # SSL
       ssl = true;
       ssl_cert_file = "/run/credentials/postgresql.service/cert";

hosts/temmie/services/userweb/mail.nix

@@ -1,6 +1,14 @@
 { config, lib, pkgs, ... }:
 {
-  services.nullmailer.enable = true;
+  services.postfix.enable = lib.mkForce false;
+
+  services.nullmailer = {
+    enable = true;
+    config = {
+      me = config.networking.fqdn;
+      remotes = "mail.pvv.ntnu.no smtp --port=25";
+    };
+  };
 
   services.bro = {
     enable = true;

modules/python-http-handlers.nix

@@ -36,6 +36,9 @@ in
           description = ''
             For each item in this list, a `ListenStream`
             option in the `[Socket]` section will be created.
+
+            Only a single `ListenStream` is currently supported by the handler script; if
+            you need more than one, you'll have to adjust {option}`handler` accordingly.
           '';
         };
 
@@ -98,8 +101,10 @@ in
             "E303" # too many blank lines
             "E305" # expected 2 blank lines after end of function or class
             "E306" # expected 1 blank line before a nested definition
+            "E402" # module level import not at top of file
             "E501" # max line length
             "E704" # multiple statements on one line (def)
+            "F811" # redefined while unused
           ];
           description = ''
             A list of flake8 rules to ignore while linting the python code.
@@ -122,6 +127,11 @@ in
                     self.send_header("Content-Length", str(len(data)))
                     self.end_headers()
                     self.wfile.write(data)
+
+            def on_reload():
+                # This function is called when the service receives SIGHUP (e.g. via `systemctl reload`).
+                # You can use it to clear caches or re-read state. Completely optional
+                pass
           '';
           description = ''
             Python code including the HTTP handler for the server.
@@ -145,6 +155,7 @@ in
           inherit (v) listenStreams;
           socketConfig = {
             Accept = false;
+            FileDescriptorName = v.name;
           } // v.socketConfig;
         };
       }))
@@ -157,30 +168,126 @@ in
         inherit (v) name;
         value = {
           serviceConfig = {
-            Type = "simple";
+            Type = "notify-reload";
+            NotifyAccess = "main";
             DynamicUser = true;
+            TimeoutStopSec = "35s";
 
             ExecStart = let
               package = pkgs.writers.writePython3Bin "${v.name}-bin" {
                 inherit (v) libraries flakeIgnore;
               } ''
+                import os
+                import signal
                 import socket
+                import socketserver
+                import threading
+                import time
                 from http.server import HTTPServer, BaseHTTPRequestHandler
 
-                ${v.handler}
+                SOCKET_NAME = "${v.name}"
+                SHUTDOWN_TIMEOUT = 30
+
+                def sd_notify(message: str):
+                    addr = os.environ.get("NOTIFY_SOCKET")
+                    if not addr:
+                        return
+                    if addr[0] == "@":
+                        addr = "\0" + addr[1:]
+                    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC) as sock:
+                        sock.connect(addr)
+                        sock.sendall(message.encode())
+
+                def sd_listen_fd(name: str) -> int:
+                    if os.environ.get("LISTEN_PID") != str(os.getpid()):
+                        raise RuntimeError("No sockets were passed to this service by systemd")
+
+                    try:
+                        count = int(os.environ.get("LISTEN_FDS", "0"))
+                    except ValueError:
+                        count = 0
+
+                    raw_names = os.environ.get("LISTEN_FDNAMES")
+                    names = raw_names.split(":") if raw_names else []
+
+                    for i in range(count):
+                        if i < len(names) and names[i] == name:
+                            return 3 + i
+
+                    raise RuntimeError(
+                        "No systemd socket named %r was passed to this service; check the "
+                        "FileDescriptorName= of the corresponding .socket unit" % name
+                    )
+
+                class Server(socketserver.ThreadingMixIn, HTTPServer):
+                    daemon_threads = True
 
-                class NoBindHTTPServer(HTTPServer):
                     def server_bind(): pass
                     def server_activate(): pass
 
+                    def __init__(self, *args, **kwargs):
+                        super().__init__(*args, **kwargs)
+                        self._request_threads = []
+                        self._request_threads_lock = threading.Lock()
+
+                    def process_request(self, request, client_address):
+                        thread = threading.Thread(
+                            target=self.process_request_thread,
+                            args=(request, client_address),
+                        )
+                        thread.daemon = self.daemon_threads
+                        with self._request_threads_lock:
+                            self._request_threads.append(thread)
+                        thread.start()
+
+                    def join_request_threads(self, timeout):
+                        deadline = time.monotonic() + timeout
+                        with self._request_threads_lock:
+                            threads = list(self._request_threads)
+                        for thread in threads:
+                            thread.join(max(deadline - time.monotonic(), 0))
+
+                def handle_reload(signum, frame):
+                    monotonic_usec = time.clock_gettime_ns(time.CLOCK_MONOTONIC) // 1000
+                    sd_notify("RELOADING=1\nMONOTONIC_USEC=%d" % monotonic_usec)
+
+                    on_reload = globals().get("on_reload")
+                    if callable(on_reload):
+                        on_reload()
+
+                    sd_notify("READY=1")
+
+                shutdown_requested = threading.Event()
+
+                def handle_sigterm(signum, frame):
+                    sd_notify("STOPPING=1")
+                    shutdown_requested.set()
+
+                ${v.handler}
+
+                assert "Handler" in globals(), "You must define a class Handler(BaseHTTPRequestHandler) in the handler code"
+
                 def main():
-                    httpd = NoBindHTTPServer(
-                        ("", 0),
-                        Handler,
-                        bind_and_activate=False,
-                    )
-                    httpd.socket = socket.fromfd(3, socket.AF_INET, socket.SOCK_STREAM)
-                    httpd.serve_forever()
+                    signal.signal(signal.SIGHUP, handle_reload)
+                    signal.signal(signal.SIGTERM, handle_sigterm)
+
+                    fd = sd_listen_fd(SOCKET_NAME)
+
+                    httpd = Server(("", 0), Handler, bind_and_activate=False)
+                    httpd.socket = socket.socket(fileno=fd)
+
+                    server_thread = threading.Thread(target=httpd.serve_forever, name="http-server", daemon=True)
+                    server_thread.start()
+
+                    sd_notify("READY=1")
+                    shutdown_requested.wait()
+
+                    deadline = time.monotonic() + SHUTDOWN_TIMEOUT
+
+                    httpd.shutdown()
+                    server_thread.join(max(deadline - time.monotonic(), 0))
+                    httpd.join_request_threads(max(deadline - time.monotonic(), 0))
+                    httpd.server_close()
 
                 if __name__ == '__main__':
                     main()