Sending Python CLI Logs to journald and syslog

When a CLI runs as a systemd service, everything it writes to stdout and stderr lands in the journal automatically. That is a good start — and a surprisingly lossy one. Every line arrives with the same priority, so journalctl -p warning cannot find your warnings. A traceback becomes twenty separate entries. The logging level, the logger name and any context you had (which target, which file, how long it took) are flattened into text that has to be parsed back out. journald is a structured log store: each entry is a set of fields, and it can be filtered by any of them. This guide makes a Python CLI use it properly — priority prefixes when logging to stderr, detecting whether stderr really is the journal, a small dependency-free handler for journald’s native protocol with structured fields, and syslog for hosts without systemd — and tests the protocol against a local socket. It belongs to the long-running and watch-mode CLIs topic.

Prerequisites

• A CLI that uses the logging module, and ideally runs as a service as in running a CLI as a systemd service.
• A Linux system with systemd to try the journal; the tests run on any Unix.

Three ways into the journal

There are three routes, from least to most effort. Plain stderr: systemd connects the service’s output to journald, which stores each line as a MESSAGE with the unit’s default priority. Stderr with priority prefixes: journald recognises a <N> at the start of each line (the kernel’s printk convention) and records it as the priority, so warnings and errors become filterable. The native protocol: the program sends datagrams to /run/systemd/journal/socket, each containing any fields it likes — multi-line messages stay one entry, and custom fields such as TARGET=eu-west become searchable with journalctl TARGET=eu-west.

The recipe

# src/mytool/journal.py
from __future__ import annotations

import logging
import os
import socket
import struct
import sys

JOURNAL_SOCKET = "/run/systemd/journal/socket"
# syslog priorities: 3 = err, 4 = warning, 6 = info, 7 = debug
PRIORITY = {logging.CRITICAL: 2, logging.ERROR: 3, logging.WARNING: 4, logging.INFO: 6, logging.DEBUG: 7}


def priority(levelno: int) -> int:
    return next((p for level, p in sorted(PRIORITY.items(), reverse=True) if levelno >= level), 7)


def stream_is_journal(stream=None) -> bool:
    """True if STREAM (default stderr) is connected to journald, per $JOURNAL_STREAM."""
    value = os.environ.get("JOURNAL_STREAM")
    if not value:
        return False
    stream = stream or sys.stderr
    try:
        st = os.fstat(stream.fileno())
    except (OSError, ValueError, AttributeError):
        return False
    device, _, inode = value.partition(":")
    return (st.st_dev, st.st_ino) == (int(device), int(inode))


class PrefixFormatter(logging.Formatter):
    """Prefix each line with <N> so journald records the right priority for stderr logs."""

    def format(self, record: logging.LogRecord) -> str:
        text = super().format(record)
        prefix = f"<{priority(record.levelno)}>"
        return "\n".join(prefix + line for line in text.splitlines())


class JournalHandler(logging.Handler):
    """Send records to journald's native protocol, with structured fields."""

    def __init__(self, identifier: str, address: str = JOURNAL_SOCKET) -> None:
        super().__init__()
        self.identifier = identifier
        self.address = address
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM | socket.SOCK_CLOEXEC)

    @staticmethod
    def _field(name: str, value: object) -> bytes:
        data = str(value).encode("utf-8", "replace")
        key = name.upper().encode()
        if b"\n" in data:                              # multi-line values use a length prefix
            return key + b"\n" + struct.pack("<Q", len(data)) + data + b"\n"
        return key + b"=" + data + b"\n"

    def emit(self, record: logging.LogRecord) -> None:
        try:
            fields = {
                "MESSAGE": self.format(record),
                "PRIORITY": priority(record.levelno),
                "SYSLOG_IDENTIFIER": self.identifier,
                "LOGGER": record.name,
                "CODE_FILE": record.pathname,
                "CODE_LINE": record.lineno,
                "CODE_FUNC": record.funcName,
            }
            fields.update(getattr(record, "journal", {}))  # extra={"journal": {"TARGET": "eu"}}
            self.sock.sendto(b"".join(self._field(k, v) for k, v in fields.items()), self.address)
        except Exception:
            self.handleError(record)

    def close(self) -> None:
        self.sock.close()
        super().close()


def configure(identifier: str = "mytool", level: int = logging.INFO) -> logging.Handler:
    """Pick the best destination: native journal, journal-aware stderr, or plain stderr."""
    root = logging.getLogger()
    if stream_is_journal() and os.path.exists(JOURNAL_SOCKET):
        handler: logging.Handler = JournalHandler(identifier)
        handler.setFormatter(logging.Formatter("%(message)s"))
    else:
        handler = logging.StreamHandler(sys.stderr)
        handler.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s"))
    root.handlers[:] = [handler]
    root.setLevel(level)
    return handler

Is stderr actually the journal?

A CLI cannot assume where its output goes: the same mytool agent runs in a terminal, in Docker, under cron and under systemd. systemd announces the connection through the JOURNAL_STREAM environment variable, which holds the device and inode numbers of the stream it connected. Comparing them with os.fstat(sys.stderr.fileno()) tells you whether stderr is that stream — and not just that some ancestor process was a service, since the variable is inherited by children whose stderr may be redirected elsewhere. Only when the check passes does configure switch to journal-specific output; otherwise it uses a normal human-readable stderr format.

Priority prefixes for stderr logging

If you prefer to keep logging through stderr — simpler, and visible in systemctl status either way — PrefixFormatter adds <3>, <4>, <6> or <7> to each line according to the record’s level. The prefix goes on every line, because journald splits multi-line output into separate entries and each needs its own priority. systemd’s SyslogLevelPrefix= setting, on by default, makes journald strip the prefix and store the priority. Use the formatter only when stream_is_journal() is true; in a terminal, <4> in front of every line is noise.

The native protocol

JournalHandler writes each record as one datagram of KEY=value lines. Values containing a newline — tracebacks, multi-line messages — use the protocol’s binary form: the key, a newline, the value’s length as a little-endian 64-bit integer, the value and a final newline. Field names must be uppercase letters, digits and underscores, and must not start with an underscore, which is reserved for fields journald adds itself (_PID, _SYSTEMD_UNIT, _HOSTNAME and others are recorded automatically and cannot be forged).

The handler sends the standard fields — MESSAGE, PRIORITY, SYSLOG_IDENTIFIER, CODE_FILE, CODE_LINE, CODE_FUNC — plus whatever a call adds through extra={"journal": {...}}. Structured context is the point: log.warning("sync slow", extra={"journal": {"TARGET": "eu-west", "DURATION_MS": 5400}}) produces an entry that journalctl TARGET=eu-west finds directly. Datagrams are limited by the socket buffer size (typically around 200 KB); for larger payloads, the protocol allows passing a memfd file descriptor, which is rarely worth implementing for a CLI — truncate instead.

syslog for everything else

On systems without systemd — Alpine containers, BSDs, older servers — or when logs must go to a central syslog server, the standard library’s SysLogHandler is enough:

import logging
import logging.handlers

handler = logging.handlers.SysLogHandler(address="/dev/log")       # local daemon
# handler = logging.handlers.SysLogHandler(address=("logs.example.com", 514))   # remote, UDP
handler.ident = "mytool: "
handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
logging.getLogger().addHandler(handler)

The handler maps logging levels to syslog priorities itself, and ident sets the program name shown in log files. On systemd hosts /dev/log is served by journald too, so syslog messages end up in the journal with correct priorities — a reasonable portable choice when you do not need custom fields. On macOS the socket is /var/run/syslog. Remote syslog over UDP loses messages silently under load; for anything important, ship logs with a dedicated agent instead.

UX considerations

• Keep terminal output human. The journal-specific paths activate only when JOURNAL_STREAM matches; interactive runs keep the familiar format from adding verbose and quiet logging flags.
• Choose a stable identifier. SYSLOG_IDENTIFIER=mytool makes journalctl -t mytool work regardless of how the program was launched.
• Name fields consistently. TARGET, DURATION_MS, RUN_ID — documented, uppercase, stable across releases — so administrators can build queries and alerts on them.
• Never log secrets into the journal. It is readable by administrators and often by the adm or systemd-journal groups, and it is retained for weeks; redact as in redacting secrets from CLI output and logs.
• Do not double-log. If the native handler is active, do not also write the same records to stderr, or every entry appears twice.

Testing the behaviour

The native protocol is tested by binding a Unix datagram socket in a temporary directory, pointing the handler at it and parsing what arrives — including the binary encoding for multi-line values:

# tests/test_journal.py
import logging
import os
import socket
import struct
import sys

import pytest

from mytool.journal import JournalHandler, PrefixFormatter, priority, stream_is_journal

pytestmark = pytest.mark.skipif(not hasattr(socket, "AF_UNIX"), reason="Unix sockets")


def parse(datagram: bytes) -> dict[str, str]:
    fields, i = {}, 0
    while i < len(datagram):
        end = datagram.index(b"\n", i)
        line = datagram[i:end]
        if b"=" in line:
            key, _, value = line.partition(b"=")
            i = end + 1
        else:
            key = line
            (size,) = struct.unpack("<Q", datagram[end + 1:end + 9])
            value = datagram[end + 9:end + 9 + size]
            i = end + 9 + size + 1
        fields[key.decode()] = value.decode()
    return fields


@pytest.fixture
def journal(tmp_path):
    path = str(tmp_path / "journal.sock")
    server = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    server.bind(path)
    server.settimeout(2)
    handler = JournalHandler("mytool", address=path)
    logger = logging.getLogger("mytool.test")
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    yield logger, server
    logger.removeHandler(handler)
    handler.close()
    server.close()


def test_structured_fields_reach_the_journal(journal):
    logger, server = journal
    logger.warning("sync slow", extra={"journal": {"TARGET": "eu-west", "DURATION_MS": 5400}})
    fields = parse(server.recv(65536))
    assert fields["MESSAGE"] == "sync slow"
    assert fields["PRIORITY"] == "4"
    assert fields["SYSLOG_IDENTIFIER"] == "mytool"
    assert fields["TARGET"] == "eu-west" and fields["DURATION_MS"] == "5400"


def test_multiline_messages_use_binary_fields(journal):
    logger, server = journal
    try:
        1 / 0
    except ZeroDivisionError:
        logger.exception("crashed")
    fields = parse(server.recv(65536))
    assert fields["MESSAGE"].startswith("crashed\nTraceback")
    assert fields["PRIORITY"] == "3"


def test_priority_mapping():
    assert [priority(n) for n in (10, 20, 25, 30, 40, 50)] == [7, 6, 6, 4, 3, 2]


def test_prefix_formatter_marks_every_line():
    record = logging.LogRecord("x", logging.ERROR, __file__, 1, "a\nb", None, None)
    assert PrefixFormatter("%(message)s").format(record) == "<3>a\n<3>b"


def test_journal_stream_detection(tmp_path, monkeypatch):
    with open(tmp_path / "f", "w") as f:
        st = os.fstat(f.fileno())
        monkeypatch.setenv("JOURNAL_STREAM", f"{st.st_dev}:{st.st_ino}")
        assert stream_is_journal(f)
        monkeypatch.setenv("JOURNAL_STREAM", "1:2")
        assert not stream_is_journal(f)

The tests never touch the real journal, so they run in CI containers without systemd. A manual check on a real system is still worthwhile once: send a message and inspect it with journalctl -t mytool -o json-pretty, which shows every field exactly as stored.

Conclusion

journald stores fields, not just lines. Detect it reliably by matching JOURNAL_STREAM against stderr’s device and inode; if you log to stderr, prefix every line with a <N> priority; for full structure, send records over the native datagram protocol with standard fields plus your own uppercase context fields, using the binary form for multi-line values; fall back to SysLogHandler where systemd is absent; keep terminal output unchanged; and test the protocol against a temporary socket.

Frequently asked questions

Related

• Up: Long-running and watch-mode CLIs
• Running a CLI as a systemd service
• Structured JSON logging in Python CLIs
• Separating logs from program output
• Health checks and heartbeats for long-running CLIs