systemd is a wonderful piece of software, that has liberated us from writing ugly bash-based init scripts and has given us features like a structured and unified log and one of the best local DNS resolvers on Linux.

In this post, I'm demonstrating how to integrate a service (also known as daemons on UNIX-like systems) written in Python into systemd to leverage some advanced features of the service manager.

What advanced features?

Classic init systems pretty much did one thing: Start up the system and it's services, and do a ordered shutdown. Apart from that, everything else had to be done by other services, which led to a ton of fragmentation in the Linux world, as every distribution had different ideas on how to do things.

systemd is advanced in the regard, that the entire lifecycle of a service is taken into account: Start, reload, status, monitoring and cleanup. But for all of this to work in sync, applications need to work together with the service manager and inform it about their internal state.

This is especially useful when you have multiple services that have dependencies on each other, and you want to make sure they're started and restarted in the right order.

For this, a C function called sd_notify is provided, but it's functionality is so simple that it can be implemented in any language with just a few lines of code.

The example application

For this example, I've created a simple FastAPI based webservice, that tells you the current time.

from datetime import datetime
from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def index():
    now = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
    return {"time": now}

To run this using systemd, you also need a simple service unit file:

[Unit]
Description=Basic Webserver

[Service]
WorkingDirectory=/home/server/app/
ExecStart=/home/server/app/.venv/bin/uvicorn basic_webserver:app
Restart=always
User=server
Group=server

[Install]
WantedBy=multi-user.service

In this case we assume that the application code is placed into /home/server/app/, a virtualenv was created at /home/server/.venv/, and a user named server was created. Do not ever run your webservices as root!

Put this file at the place systemd is expecting it (/etc/systemd/system/webserver.service) and enable and start it using the following commands:

$ systemctl enable webserver.service
Created symlink /etc/systemd/system/multi-user.service.wants/webserver.service → /etc/systemd/system/webserver.service.
Unit /etc/systemd/system/webserver.service is added as a dependency to a non-existent unit multi-user.service.

$ systemctl start webserver.service

After this is done, you can ask systemd about it's status:

$ systemctl status webserver.service

If everything is correct, the output should look like this:

As you can see in the screenshot, modern versions of systemd can track all processes of our services and even the resources used like CPU and Memory.

systemd also has a ton of other useful features to increase the security of services by limiting their capabilities.

Building the basic blocks

Before we begin with the integration into the application, let's implement the communication with the service manager first. To communicate with it's services, systemd is providing a UNIX socket with the path provided in the environment variable NOTIFY_SOCKET. You can connect to it and send simple text-based commands.

As FastAPI is at it's core an async framework, we also write our integration using asyncio.

import asyncio
import socket
import os


async def sd_notify(*commands: str):
    """
    Send one or more commands to systemd, if available.
    """
    if path := os.environ.get("NOTIFY_SOCKET"):
        # If the path starts with @, it's a Linux abstract namespace socket.
        # To connect, we need to replace the @ with a NUL byte.
        if path.startswith("@"):
            path = "\0" + path[1:]
        loop = asyncio.get_running_loop()

        with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
            sock.connect(path)
            sock.setblocking(False)
            await loop.sock_sendall(sock, "\n".join(commands).encode())

While there could be ways to make this more efficient (like caching and re-using the socket), this function is never on a critical path in the application, therefore we don't need to bother with this aspect.

Next, let's implement the commands, which are all one-liners. The comments for each function were taken from the systemd documentation.

from sd_notify_async import sd_notify


async def ready():
    """
    Tells the service manager that service startup is finished,
    or the service finished re-loading its configuration.
    """
    await sd_notify("READY=1")


async def reloading():
    """
    Tells the service manager that the service is beginning to reload its configuration.
    This is useful to allow the service manager to track the service's internal state,
    and present it to the user. Note that a service that sends this notification
    must also send a "READY=1" notification when it completed reloading its configuration.
    """
    await sd_notify("RELOADING=1")


async def stopping():
    """
    Tells the service manager that the service is beginning its shutdown.
    This is useful to allow the service manager to track the service's internal state,
    and present it to the user.
    """
    await sd_notify("STOPPING=1")


async def status(text: str):
    """
    Passes a single-line UTF-8 status string back to the service manager
    that describes the service state.
    """
    await sd_notify(f"STATUS={text}")

Now, we can integrate these functions into our service:

from datetime import datetime
from contextlib import asynccontextmanager

from fastapi import FastAPI

import sd_notify_async_basic as systemd


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Code to be run when the application is booting up
    await systemd.ready()
    await systemd.status("Application ready!")
    # Give back control to FastAPI
    yield
    # Code to be run when the application is stopping
    await systemd.stopping()


app = FastAPI(lifespan=lifespan)


@app.get("/")
async def index():
    now = datetime.now().strftime("%d.%m.%Y %H:%M:%S")
    return {"time": now}

As last thing, we need to tell systemd that we want it to expect these commands:

[Service]
Type=notify

After re-deploying and restarting, you'll see more information in the status output:

After the application has successfully initialized, the state is changed and the status message also displayed:

Recap

By adding just a few lines of code, you made the life of administrators managing your applications a bit easier.