logsource.py

#! /usr/bin/env python3
# -*- coding: utf-8; py-indent-offset: 4 -*-
#
# Author:  Linuxfabrik GmbH, Zurich, Switzerland
# Contact: info (at) linuxfabrik (dot) ch
#          https://www.linuxfabrik.ch/
# License: The Unlicense, see LICENSE file.

# https://github.com/Linuxfabrik/lib/blob/main/CONTRIBUTING.md

"""Reads a log incrementally, wherever it is kept: a file on disk, a systemd unit in the
journal, or the log of a container.

Anything that watches a log answers the same three questions on every run, and each storage
answers them differently: where the lines come from, where the previous run stopped, and
whether that stored position still means anything after the log was rotated, truncated or
rewritten. This module answers all three behind one call and hands back an opaque position to
store until the next run.

How exactly a position can be expressed differs per storage, and that difference is reported
rather than hidden. A file is resumed at a byte offset and a unit at a journal cursor, so
neither repeats nor loses a line. A container log only offers a timestamp, and the engines
treat it inclusively, so an entry written in the very same instant as the last one read can
come back twice. A consumer that cannot afford a repeated line reads `fidelity` and
deduplicates where it says so.
"""

__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland'
__version__ = '2026101203'

import collections
import datetime
import os
import re
import stat

from . import disk, human, shell, txt
from . import time as lftime

# A position either names the exact spot the previous run stopped at, or only
# approximates it. See the module docstring.
FIDELITY_APPROXIMATE = 'approximate'
FIDELITY_EXACT = 'exact'

KIND_CONTAINER = 'container'
KIND_FILE = 'file'
KIND_JOURNALD = 'journald'

DEFAULT_TIMEOUT = 8

# Prefixes that name where a log is kept. Anything else is a path, which is why
# a Windows drive letter (`C:\log\app.log`) is read as one.
_SOURCE_PREFIXES = {
    'docker': KIND_CONTAINER,
    'kubectl': KIND_CONTAINER,
    'podman': KIND_CONTAINER,
    'systemd': KIND_JOURNALD,
}


# What a syslog daemon or journald puts in front of the message an application
# wrote: a time, the host, and the identifier with the process id. Both spellings
# of the time, because a file written by rsyslog carries the traditional one and
# `journalctl --output=short-iso` carries ISO 8601.
_SYSLOG_PREFIX_REGEX = re.compile(
    r"""^(?:
        [A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}   # Aug 29 11:15:53
        |\d{4}-\d{2}-\d{2}[T\ ]\d{2}:\d{2}:\d{2}\S*   # 2026-08-29T11:15:53+0200
    )
    \s+\S+                                            # the host
    \s+(?P<identifier>[^\s:\[]+)(?:\[\d+\])?:\s          # sshd-session[1234]:
    """,
    re.VERBOSE,
)


def count_within(lines, since, parse_line=None, key=None):
    """
    Count the lines written at or after a moment, optionally per source.

    What a log holds and what happened lately are two different numbers, and a consumer that
    alerts on how often something arrives needs the second one. A window over the whole log
    would keep reporting a burst that ended hours ago.

    Where the lines name who caused them, the total is not the number to judge by either.
    Six failures from one address within ten minutes is somebody working on this host; six
    failures from six addresses is the background of an open network going past, and only the
    first is worth reporting. `key` turns the count into the largest a single source reached,
    which is also the quantity an intrusion prevention system counts before it blocks one, so
    thresholds derived from such a system compare against the same thing.

    Parameters
    ----------
    lines : iterable of str
        The log lines to count.
    since : datetime.datetime
        The start of the window, naive and in local time.
    parse_line : callable, optional
        Handed to `timestamp()`.
    key : callable, optional
        Takes a line and returns what caused it - an address, an account, whatever the format
        offers - or None where the line names nothing. Lines returning None are counted
        together as one source, so a burst of unattributable lines still shows up. Without it
        the lines are not grouped and `count` is the plain total.

    Returns
    -------
    dict
          - `busiest` (`str | None`): The source that reached `count`. None without `key`, and
            None where the lines that reached it name no source.
          - `count` (**int**): The number to judge by: the total, or with `key` the largest any
            single source reached.
          - `sources` (**int**): How many distinct sources were seen. 0 without `key`.
          - `total` (**int**): Every line within the window, whatever its source.
          - `undated` (**int**): Lines carrying no timestamp either reader could find.

    Notes
    -----
    - The undated ones are reported separately rather than counted in or silently dropped,
      because a source whose format leaves the timestamp out would otherwise read as a quiet
      one and never raise anything.

    Examples
    --------
    >>> count_within(lines, datetime.now() - timedelta(minutes=10))
    {'busiest': None, 'count': 37, 'sources': 0, 'total': 37, 'undated': 1}
    >>> count_within(lines, datetime.now() - timedelta(minutes=10), key=peer_of)
    {'busiest': '198.51.100.7', 'count': 12, 'sources': 9, 'total': 37, 'undated': 1}
    """
    buckets = collections.Counter()
    total = 0
    undated = 0
    for line in lines:
        logged_at = timestamp(line, parse_line)
        if logged_at is None:
            undated += 1
        elif logged_at >= since:
            total += 1
            if key:
                buckets[key(line)] += 1
    busiest = None
    count = total
    if key and buckets:
        busiest, count = buckets.most_common(1)[0]
    return {
        'busiest': busiest,
        'count': count,
        'sources': len([name for name in buckets if name is not None]),
        'total': total,
        'undated': undated,
    }


def covered_window(line_groups, parse_line=None):
    """
    Return the earliest and the latest moment a set of logs was written in.

    A consumer reporting on a window of a log has to say which stretch of time that window
    covers, because the line count alone does not: on a busy host the same 30000 lines reach
    back hours, on a quiet one months.

    Each group is one log, and only its ends are read rather than every line, which keeps the
    cost at two parses per log however long it is. Taking the ends of the whole set instead
    would be wrong as soon as there is more than one: the logs arrive one after the other, so
    the newest line of the first one sits in the middle rather than at the end.

    Parameters
    ----------
    line_groups : iterable of iterable of str
        One iterable of lines per log, each
        in the order the log holds them.
    parse_line : callable, optional
        Handed to `timestamp()` for a format it cannot
        read on its own.

    Returns
    -------
    tuple
          - tuple[0] (**datetime.datetime | None**): The earliest moment found, or None where no
            line carries a time.
          - tuple[1] (**datetime.datetime | None**): The latest moment found, or None.

    Examples
    --------
    >>> covered_window(
    ...     [['Aug 29 11:15:53 host sshd[1]: Accepted publickey for alice']]
    ... )[0].hour
    11
    """
    moments = []
    for lines in line_groups:
        lines = list(lines)
        for candidate in (lines, reversed(lines)):
            for line in candidate:
                moment = timestamp(line, parse_line)
                if moment is not None:
                    moments.append(moment)
                    break
    if not moments:
        return None, None
    # Compared as they are only where they agree on being aware or naive: a log
    # written with a UTC offset next to one written without cannot be ordered,
    # and guessing an offset would move a line by hours.
    if len({moment.tzinfo is None for moment in moments}) > 1:
        moments = [moment.replace(tzinfo=None) for moment in moments]
    return min(moments), max(moments)


def describe(source, size=None, size_threshold=None):
    """
    Name one source the way a report names it, together with what was read along.

    A source read from a file is named by its path in full, because that is where
    somebody copies the path from, and the rotated predecessors that were read with
    it are listed behind it. Their size is looked up here rather than passed in: a
    predecessor is not a source the caller named, the reader found it, and it stays
    out of the size a caller trends.

    Parameters
    ----------
    source : dict
        One entry of the `sources` a `read_many()` result carries,
        holding at least a `label` and a `rotated` list.
    size : int, optional
        The size of the live source in bytes, where the
        caller has it and wants it named. Defaults to `None`, which names no size.
    size_threshold : int, optional
        A size to compare against, so the fact
        arrives with the verdict a caller reaches on it. Defaults to `None`, which
        states the size on its own.

    Returns
    -------
    str
        The description, with every path in backticks.

    Examples
    --------
    >>> describe({'label': '/var/log/secure', 'rotated': []})
    '`/var/log/secure`'

    >>> describe({'label': '/var/log/secure', 'rotated': []}, size=2048)
    '`/var/log/secure` (size: 2.0KiB)'
    """
    described = f'`{source["label"]}`'
    if size is not None:
        if size_threshold is None:
            described += f' (size: {human.bytes2human(size)})'
        else:
            comparison = '>' if size >= size_threshold else '<'
            described += (
                f' (size: {human.bytes2human(size)} {comparison}'
                f' {human.bytes2human(size_threshold)})'
            )
    for predecessor in source.get('rotated') or []:
        described += f' + `{predecessor}`'
        predecessor_stat = disk.stat(predecessor)
        if predecessor_stat is not None:
            described += f' (size: {human.bytes2human(predecessor_stat.st_size)})'
    return described


def parse(source):
    """
    Split a log source specification into the storage it names and the target within it.

    Parameters
    ----------
    source : str
        Either a path to a file, or a prefixed target: `systemd:UNITNAME` for a unit in the
        journal, and `docker:NAME`, `kubectl:NAME` or `podman:NAME` for the log of a container.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the source could be read, otherwise False.
          - tuple[1] (**tuple | str**):

            - If successful, a `(kind, engine, target)` tuple. `kind` is one of `KIND_CONTAINER`,
              `KIND_FILE` or `KIND_JOURNALD`. `engine` names the container engine and is None for
              everything else. `target` is the path, the unit name or the container name.
            - If unsuccessful, an error message string.

    Examples
    --------
    >>> parse('/var/log/messages')
    (True, ('file', None, '/var/log/messages'))
    >>> parse('systemd:sshd')
    (True, ('journald', None, 'sshd'))
    >>> parse('podman:database')
    (True, ('container', 'podman', 'database'))
    """
    if not source or not isinstance(source, str):
        return False, 'No log source given.'
    prefix, separator, target = source.partition(':')
    kind = _SOURCE_PREFIXES.get(prefix) if separator else None
    if kind is None:
        return True, (KIND_FILE, None, source)
    target = target.strip()
    if not target:
        return False, f'Log source "{source}" names no target.'
    return True, (kind, prefix if kind == KIND_CONTAINER else None, target)


def systemd_booted(root=''):
    """
    Tell whether systemd is the service manager this host runs, the way `sd_booted()` does.

    A unit file says only that a package shipped one: Devuan, Alpine and other hosts with
    another init system still carry the `.service` files of the packages they install, and
    there is no journal to read for those units. systemd creates its runtime unit directory
    very early at boot, so its presence answers the question, and nothing else does.

    The path is fixed and read without following a symlink at its end, and only root can
    create it below `/run`, so no caller of a privileged consumer steers the answer.
    Verified against `sd_booted()` in systemd's `src/libsystemd/sd-daemon/sd-daemon.c`,
    which tests `access_nofollow("/run/systemd/system/", F_OK)`.

    Parameters
    ----------
    root : str, optional
        Prefix for the path, for the unit tests of a consumer that stand in a directory tree
        for the host. Defaults to the host itself.

    Returns
    -------
    bool
        True if systemd runs the host, otherwise False, also on Windows.

    Examples
    --------
    >>> systemd_booted()  # doctest: +SKIP
    True
    """
    return os.path.lexists(f'{root}/run/systemd/system')


# Enough of the head of a file to tell two lines of text apart, which is all the
# fingerprint has to do here.
_FINGERPRINT_LENGTH = 256

# What one run takes from a log. Anybody who can create a file in a log directory
# decides what a rotated file holds, and a root consumer reads it: a line without a line
# end, or a compressed file that is a megabyte on disk and a terabyte as text, must cost a
# bounded amount of memory and time.
#
# A line over `MAX_LINE_BYTES` is no log line any more. Its first `MAX_LINE_BYTES` bytes are
# reported, the rest of it up to the line end is skipped, and the offset moves past all of
# it. A last line without a line end that is shorter than that is held back for the next run
# as before; one that has reached the bound is cut off the same way, because waiting for a
# line end that never comes would read the same bytes on every run. The bytes that arrive
# after the cut are then read as a line of their own.
#
# `MAX_ROTATED_BYTES` is the most a run reads out of rotated files and of a compressed
# source, counted uncompressed. Every line is counted with `_LINE_OVERHEAD` on top, the
# memory of the string that holds it, so that a file of empty lines is bounded too. 512 MiB
# is more than a day of access log of a busy web server, which the consumers read one
# rotated file of; where the bound is reached the reading stops and the notice says so. The live file of a plain source
# is not capped: it is read from where the previous run stopped, and everything in it is
# what the consumer asked for.
MAX_LINE_BYTES = 1024 * 1024
MAX_ROTATED_BYTES = 512 * 1024 * 1024
_LINE_OVERHEAD = 64

# What a journald unit or a container may print in one read, per stream: a resumed read
# hands over everything logged since the cursor or the timestamp, and `--lines` or `--tail`
# do not cap it. A runaway unit would otherwise be held in memory whole, up to what it prints
# within the timeout. Past the cap the command is killed and the read fails with a message
# that says so. A capture costs about three times its size in memory (the chunks, the joined
# bytes, the decoded text and the split lines), which is why it is a quarter of what is
# read out of rotated files. `journalctl` says little on standard error, hints only, so
# its cap is small; a container client passes what the application wrote to standard error,
# which is log content, and gets the same cap as standard output.
MAX_COMMAND_OUTPUT_BYTES = 128 * 1024 * 1024
MAX_JOURNALCTL_STDERR_BYTES = 1024 * 1024


class _BudgetExhausted(Exception):
    """A run has read as much out of rotated files as it may."""


class _ReadState:
    """What one run has used up of its bounds, and what it has to report about them.

    `budget` is the number of bytes a run may still take, or None where there is no cap.
    """

    def __init__(self, budget=None):
        self.budget = budget
        self.shortened = 0
        self.stopped = None

    def charge(self, size):
        """Count `size` more bytes, and raise `_BudgetExhausted` once the budget is gone."""
        if self.budget is None:
            return
        self.budget -= size + _LINE_OVERHEAD
        if self.budget < 0:
            raise _BudgetExhausted


def _bounded_lines(handle, state):
    """Yield `(line, consumed, final)` for every line of `handle`, each `line` at most
    `MAX_LINE_BYTES` long.

    `consumed` is how many bytes of the file the line took, its line end and the part of a
    too long line that was skipped included, so the caller's offset stays exact. `final`
    says the line is complete: it has its line end, or it has reached the bound, which is as
    long as a line may get. A last line that is not final is for the caller to hold back. A
    line that really lost bytes to the bound is counted in `state.shortened`; one that is
    exactly as long as the bound, with or without its line end, is returned whole. The bytes
    are charged to `state` as they are read, which raises `_BudgetExhausted` when it runs out.
    """
    while True:
        line = handle.readline(MAX_LINE_BYTES)
        if not line:
            return
        consumed = len(line)
        state.charge(consumed)
        if len(line) >= MAX_LINE_BYTES and not line.endswith(b'\n'):
            # Skip the rest of the line in pieces, so none of it is held. Nothing was lost
            # if all that follows is the line end.
            lost = False
            while True:
                rest = handle.readline(MAX_LINE_BYTES)
                consumed += len(rest)
                state.charge(len(rest))
                if rest not in (b'', b'\n', b'\r\n'):
                    lost = True
                if not rest or rest.endswith(b'\n'):
                    break
            if lost:
                state.shortened += 1
        yield line, consumed, line.endswith(b'\n') or len(line) >= MAX_LINE_BYTES


def _is_rewritten(raw, fingerprint, length):
    """Tell whether the head of an open file still matches the fingerprint of the previous run.

    The head is the part that appending does not touch, so it identifies the file: it changes
    only once the file has become a different one. Hashes exactly as many bytes as were hashed
    back then, so appending to a file whose head was shorter than the fingerprint does not look
    like a rewrite.
    """
    if not length:
        # The file was empty on the previous run, so there is nothing to compare
        # against, and nothing to miss either: an offset of 0 stays valid no
        # matter what was written since.
        return True, False
    success, result = disk.get_fingerprint(raw, length=length)
    if not success:
        return False, result
    current, hashed = result
    if hashed < length:
        # the file no longer holds the bytes that were hashed last time, so it
        # must have been truncated
        return True, True
    return True, current != fingerprint


# What a log rotator appends to the name of the file it moved aside. Everything
# it can produce is made of digits and separators, whether it counts generations
# (`.1`) or stamps a date (`-20260828`, `-2026-08-28`), so a name carrying a
# letter is somebody's backup copy and not a rotation.
_ROTATED_SUFFIX_REGEX = re.compile(r'^[-.][-_.0-9]*[0-9][-_.0-9]*$')

# Compression a rotator applies to what it moved aside, and the module that
# reads it back. A rotator can be told to use something else (zstd, for
# example); such a file simply does not look like a rotation here and is left
# alone rather than read as binary noise.
_COMPRESSION_MODULES = {
    '.bz2': 'bz2',
    '.gz': 'gzip',
    '.xz': 'lzma',
}


def _open_log_file(path, allowed_roots=None):
    """Open a log file for reading in binary, decompressing it where its name says so.

    The file is opened once, through `disk.open_file()`, so with `allowed_roots` it is
    the very file that was checked, and everything the caller needs from it (its
    inode, its size, its fingerprint, its lines) comes from that one handle rather
    than from a second look at the path.

    Returns `(True, (handle, raw))`, where `handle` yields the uncompressed lines and
    `raw` is the file underneath it; the caller closes both. For an uncompressed file
    the two are the same object.

    The module that reads a compression is imported here rather than at the top, because a
    Python can be built without one, and that must not keep this module from being imported at
    all.
    """
    module = None
    for suffix, module_name in _COMPRESSION_MODULES.items():
        if not path.endswith(suffix):
            continue
        try:
            module = __import__(module_name)
        except ImportError:
            return False, (
                f'Cannot read "{path}": this Python has no `{module_name}` module.'
            )
        break
    success, raw = disk.open_file(
        path, binary=True, allowed_roots=allowed_roots, nofollow=bool(allowed_roots)
    )
    if not success:
        return False, raw
    if module is None:
        return True, (raw, raw)
    try:
        return True, (module.open(raw, mode='rb'), raw)
    except OSError as e:
        raw.close()
        return False, f'I/O error "{e.strerror}" while reading {path}'


def _rotated_files(path, count=None):
    """Return the files a rotator moved aside from `path`, the oldest of them first.

    `count` limits them to the most recent ones; None returns all of them.

    Ordering is by modification time rather than by name, because the two naming schemes a
    rotator uses disagree about which name is the newest: a counted generation is renamed
    upwards on every rotation while a dated one never moves. Compressing a rotated file keeps
    its modification time, so the order holds either way.

    Only a regular file is considered, and a symlink is not followed, so nothing that was
    dropped into the log directory can widen what the caller reads.
    """
    directory = os.path.dirname(path) or '.'
    base = os.path.basename(path)
    candidates = []
    try:
        entries = os.listdir(directory)
    except OSError:
        # A directory that cannot be listed simply yields no predecessor. The
        # file itself is read on its own and reports its own trouble.
        return []
    for name in entries:
        if not name.startswith(base) or name == base:
            continue
        suffix = name[len(base) :]
        for compression in _COMPRESSION_MODULES:
            if suffix.endswith(compression):
                suffix = suffix[: -len(compression)]
                break
        if not _ROTATED_SUFFIX_REGEX.match(suffix):
            continue
        candidate = os.path.join(directory, name)
        try:
            candidate_stat = os.lstat(candidate)
        except OSError:
            continue
        if not stat.S_ISREG(candidate_stat.st_mode):
            continue
        candidates.append((candidate_stat.st_mtime, candidate))
    # newest first to pick the `count` most recent, then reversed so their lines
    # arrive in the order they were written
    candidates.sort(reverse=True)
    return [candidate for _, candidate in reversed(candidates[:count])]


def _is_truncated(size, position, compressed):
    """Return whether a file of `size` bytes is shorter than where `position` stopped.

    The offset counts the bytes of the lines read, which for a compressed file are the
    uncompressed ones; its compressed size is nearly always smaller, so holding the two
    against each other took every compressed file for truncated and read it again from
    the start, run after run. A compressed file is therefore compared with its own size
    at the time the position was taken. A position from before that size was stored
    leaves the decision to the fingerprint.
    """
    if compressed:
        return size < position.get('size', 0)
    return size < position.get('offset', 0)


def _resume_rotated(path, position, allowed_roots, lines, state):
    """Read what the file held after `position` before a rotator moved it aside.

    The stored inode names the file the previous run stopped in. If a predecessor still
    carries it, the rest of that file is read from the stored offset, followed by every
    predecessor that was rotated after it, so the lines written between the last run and the
    rotation are not lost. A rotator that compressed the file wrote a new one with a new
    inode, and one that deleted it left nothing to find; then nothing is read and the caller
    restarts at the beginning of the live file.

    The candidate is checked on the handle that reads it, not on the path: it has to be the
    same inode, at least as long as the stored offset, and start with the bytes the stored
    fingerprint was taken from.

    Returns `(resumed, files, read, notices)`: whether the file was found, the files that
    were read in order, how many lines they had, and why a newer predecessor was skipped.
    The lines are appended to `lines`. What is read is charged to `state`; where its budget
    is gone, the reading stops with `state.stopped` set.
    """
    stored_inode = str(position.get('inode'))
    offset = position.get('offset', 0)
    predecessors = _rotated_files(path)
    index = None
    for number, candidate in enumerate(predecessors):
        try:
            candidate_inode = str(os.lstat(candidate).st_ino)
        except OSError:
            continue
        if candidate_inode == stored_inode:
            index = number
            break
    if index is None:
        return False, [], 0, []
    files = []
    read = 0
    notices = []
    for number, predecessor in enumerate(predecessors[index:]):
        success, opened = _open_log_file(predecessor, allowed_roots)
        if not success:
            if number == 0:
                return False, [], 0, []
            notices.append(opened)
            continue
        predecessor_handle, predecessor_raw = opened
        try:
            with predecessor_raw, predecessor_handle:
                start = 0
                if number == 0:
                    candidate_stat = os.fstat(predecessor_raw.fileno())
                    if str(candidate_stat.st_ino) != stored_inode or _is_truncated(
                        candidate_stat.st_size,
                        position,
                        compressed=predecessor_handle is not predecessor_raw,
                    ):
                        return False, [], 0, []
                    if position.get('fingerprint'):
                        success, rewritten = _is_rewritten(
                            predecessor_raw,
                            position.get('fingerprint'),
                            position.get('length', 0),
                        )
                        if not success or rewritten:
                            return False, [], 0, []
                    start = offset
                predecessor_handle.seek(start)
                # A rotated file is complete, so a last line without a line end is
                # read as it is rather than held back for a run that never comes.
                for line, _, _ in _bounded_lines(predecessor_handle, state):
                    read += 1
                    lines.append(
                        txt.to_text(line, errors='strict_or_latin1').rstrip('\r\n')
                    )
        except _BudgetExhausted:
            state.stopped = predecessor
            files.append(predecessor)
            break
        except OSError as e:
            if number == 0:
                return False, [], 0, []
            notices.append(f'I/O error "{e.strerror}" while reading {predecessor}')
            continue
        files.append(predecessor)
    return True, files, read, notices


def _read_open_file(path, handle, raw, position, allowed_roots, max_lines, rotated):
    """The part of `_read_file()` that works on the already opened file."""
    try:
        file_stat = os.fstat(raw.fileno())
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while reading {path}'
    inode = str(file_stat.st_ino)
    # Fingerprint before reading rather than after: should the file be rewritten
    # in between, the stored fingerprint is the older one, so the next run
    # rescans the file instead of missing what was written in the meantime.
    success, result = disk.get_fingerprint(raw, length=_FINGERPRINT_LENGTH)
    if not success:
        return False, result
    fingerprint, fingerprint_length = result

    stored_offset = position.get('offset', 0) if position else 0
    offset = stored_offset
    restarted = False
    moved_aside = False
    if position:
        # A state written before the position carried a kind may hold the inode
        # with INTEGER affinity and read back as int, so compare as strings.
        if str(position.get('inode')) != inode:
            # rotated or replaced; what the old file held after the stored offset is
            # looked for among the predecessors below
            moved_aside = True
            offset = 0
        elif _is_truncated(file_stat.st_size, position, compressed=handle is not raw):
            # truncated below where we stopped
            offset = 0
        elif position.get('fingerprint'):
            success, rewritten = _is_rewritten(
                raw,
                position.get('fingerprint'),
                position.get('length', 0),
            )
            if not success:
                return False, rewritten
            if rewritten:
                # An application that rewrites its log from the beginning
                # (`> logfile` instead of `>> logfile`) keeps the inode, and if
                # the new content is at least as long as what was read last
                # time, the size does not shrink either. Only the head tells us
                # that everything after our offset is gone.
                offset = 0
        # Resetting a position that was 0 anyway loses nothing, so it is not a
        # restart the consumer has to hear about.
        restarted = offset == 0 and stored_offset != 0
    # The fingerprint moved the file position. A decompressor reads on from wherever
    # the file underneath stands, so put it back to where the stream starts.
    try:
        raw.seek(0)
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while reading {path}'

    # A cap is applied to the lines that are kept, not to the lines that are
    # read: the offset still advances to the end, so a run that hits the cap
    # drops the oldest of the new lines instead of re-reading them forever.
    # Rotated predecessors share the same deque, so the cap keeps the most
    # recent lines across all of them rather than per file.
    lines = collections.deque(maxlen=max_lines) if max_lines else []
    read = 0
    notices = []
    rotated_read = []
    state = _ReadState(MAX_ROTATED_BYTES)
    for predecessor in _rotated_files(path, rotated) if rotated else []:
        success, opened = _open_log_file(predecessor, allowed_roots)
        if not success:
            notices.append(opened)
            continue
        predecessor_handle, predecessor_raw = opened
        try:
            with predecessor_raw, predecessor_handle:
                for line, _, _ in _bounded_lines(predecessor_handle, state):
                    read += 1
                    lines.append(
                        txt.to_text(line, errors='strict_or_latin1').rstrip('\r\n')
                    )
        except _BudgetExhausted:
            state.stopped = predecessor
            rotated_read.append(predecessor)
            break
        except OSError as e:
            notices.append(f'I/O error "{e.strerror}" while reading {predecessor}')
            continue
        rotated_read.append(predecessor)
    if moved_aside and state.stopped is None:
        resumed, files, resumed_read, resumed_notices = _resume_rotated(
            path, position, allowed_roots, lines, state
        )
        if resumed:
            # nothing between the stored offset and the live file is missing
            restarted = False
        read += resumed_read
        rotated_read += files
        notices += resumed_notices
    if handle is raw:
        # the live file of a plain source is read from where the previous run stopped,
        # and all of it is what the consumer asked for
        state.budget = None
    try:
        handle.seek(offset)
        for line, consumed, final in _bounded_lines(handle, state):
            if not final:
                # The writer is in the middle of this line. Reporting it now would
                # hand out half a line and the rest of it on the next run, so it
                # is left for the next run to read whole.
                break
            read += 1
            lines.append(txt.to_text(line, errors='strict_or_latin1').rstrip('\r\n'))
            offset += consumed
    except _BudgetExhausted:
        # only a compressed source gets here, which is read from its start every time
        state.stopped = path
    except OSError as e:
        return False, f'I/O error "{e.strerror}" while reading {path}'
    notice = []
    if notices:
        notice.append('Skipped a rotated file: ' + '; '.join(notices) + '.')
    if state.stopped:
        notice.append(
            f'Stopped reading {state.stopped}: more than '
            f'{MAX_ROTATED_BYTES // 1024 // 1024} MiB uncompressed.'
        )
    if state.shortened:
        notice.append(
            f'Shortened {state.shortened} line(s) longer than '
            f'{MAX_LINE_BYTES // 1024 // 1024} MiB.'
        )
    return True, {
        'fidelity': FIDELITY_EXACT,
        'kind': KIND_FILE,
        'label': path,
        'lines': list(lines),
        # What the consumer hears about the file: a rotated predecessor that could not be
        # read, a read that was stopped by its budget, and lines that were cut.
        'notice': ' '.join(notice),
        'position': {
            'fingerprint': fingerprint,
            'inode': inode,
            'kind': KIND_FILE,
            'length': fingerprint_length,
            'offset': offset,
            # the size of the file itself, for a compressed one, whose offset counts
            # the uncompressed bytes and cannot be held against it
            **({'size': file_stat.st_size} if handle is not raw else {}),
        },
        'restarted': restarted,
        'rotated': rotated_read,
        'truncated': read > len(lines) or bool(state.shortened or state.stopped),
    }


def _read_file(path, position, allowed_roots, max_lines, rotated):
    """Read the lines a file grew by since `position`, detecting rotation and rewrites."""
    # Through the decompressor where the name says the file is compressed, the
    # way a rotated predecessor is read: a caller that names a rotated file
    # directly would otherwise be handed the compressed bytes as text and told
    # that the application never wrote a line. `seek()` and `tell()` on such a
    # handle count uncompressed bytes, so the position keeps its meaning.
    success, result = _open_log_file(path, allowed_roots)
    if not success:
        if result.startswith('I/O error') and any(
            character in path for character in '*?['
        ):
            # A shell that found nothing to expand hands the pattern on
            # unchanged, and a caller who meant a set of files then sees a
            # missing file with an odd name. Saying that a source is one file
            # beats letting them hunt for a typo that is not there.
            result += (
                ' - a wildcard is not expanded here, every file is its own source.'
            )
        return False, result
    handle, raw = result
    with raw, handle:
        return _read_open_file(
            path, handle, raw, position, allowed_roots, max_lines, rotated
        )


# journalctl prints the cursor of the last entry on a line of its own when asked
# for it with --show-cursor.
_JOURNALD_CURSOR_PREFIX = '-- cursor: '


def _read_journald(unit, position, max_lines, since, timeout):
    """Read the entries a unit logged since `position`, resuming at a journal cursor."""
    cursor = position.get('cursor') if position else None
    # `cursor` moves on to the one journalctl prints below; whether the caller
    # passed one in decides how --lines= behaves and how a cut is detected
    resumed = bool(cursor)
    cmd = [
        'journalctl',
        '--no-pager',
        '--output=short-iso',
        '--show-cursor',
        # Every value is bound to its option with `=`, so a unit name starting
        # with `-` cannot be picked up as an option of journalctl itself.
        f'--unit={unit}',
    ]
    if resumed:
        cmd.append(f'--after-cursor={cursor}')
    elif since:
        cmd.append(f'--since={since}')
    else:
        # Without a stored position and without a window the caller named, the
        # current boot is the one bounded answer that needs no clock arithmetic.
        cmd.append('--boot')
    if max_lines and not resumed:
        # Without a cursor journalctl jumps back from the end, so --lines= keeps the
        # most recent entries. After a cursor it reads forward from there and stops
        # after N, which are the oldest ones, and the cursor it prints then follows
        # them: a busy unit would fall further behind on every run. The cap is applied
        # below instead. Verified against systemd 259, src/journal/journalctl-show.c,
        # seek_journal().
        cmd.append(f'--lines={max_lines}')
    success, result = shell.shell_exec(
        cmd,
        timeout=timeout,
        max_bytes=(MAX_COMMAND_OUTPUT_BYTES, MAX_JOURNALCTL_STDERR_BYTES),
    )
    if not success:
        return False, result
    stdout, stderr, retc = result
    if 'insufficient permissions' in stderr:
        # journalctl ends with this, and with an exit code of 1, when the caller may
        # not open a single journal file (src/shared/journal-util.c). Passing the raw
        # text on would bury the one thing to do under three lines of hint.
        # Verified against systemd 252 on Rocky 9.
        return False, (
            f'Not allowed to read `systemd:{unit}`. Run with elevated privileges, '
            f'or add the account to the `adm` or `systemd-journal` group.'
        )
    if retc != 0 and stderr.strip():
        return False, (
            f'`{" ".join(cmd)}` exited with error ({retc}, {stderr.strip()}).'
        )
    # An exit code with nothing said about it means the journal holds nothing for
    # this unit rather than that the read went wrong: `--show-cursor` exits 1 on a
    # unit that logged nothing since the boot, because there is no cursor to
    # print, and journalctl says so on standard output only ("-- No entries --").
    # Failing here would put a permanent WARNING on every host with a quiet unit.
    # Measured against systemd 239 on Rocky 8.
    lines = collections.deque(maxlen=max_lines) if max_lines else []
    read = 0
    for line in stdout.splitlines():
        if line.startswith(_JOURNALD_CURSOR_PREFIX):
            # Keep the previous cursor when journalctl prints none, which is what
            # it does when the unit logged nothing at all since last time.
            cursor = line[len(_JOURNALD_CURSOR_PREFIX) :].strip() or cursor
            continue
        if line.startswith('-- '):
            # journalctl frames its output with informational lines such as
            # `-- No entries --` and `-- Boot ... --`. With --output=short-iso
            # every real entry starts with its timestamp, so a line starting
            # with `-- ` is never one of them.
            continue
        read += 1
        lines.append(line)
    notice = stderr.strip()
    if notice and not lines:
        # journalctl says on stderr, and with an exit code of 0, that it is only
        # showing what the caller is allowed to see. Where that leaves nothing at
        # all, returning an empty result would report a quiet log on a host whose
        # log simply cannot be read, which is the worst answer a check can give.
        # Verified against systemd 257 on Debian 13, where an account outside the
        # `adm` and `systemd-journal` groups sees no system entries at all.
        return False, (
            f'Read no entries of `systemd:{unit}`, and journalctl reported: {notice} '
            f'Run with elevated privileges, or add the account to the `adm` or '
            f'`systemd-journal` group.'
        )
    return True, {
        'fidelity': FIDELITY_EXACT,
        'kind': KIND_JOURNALD,
        'label': f'systemd:{unit}',
        'lines': list(lines),
        # What journalctl said next to the entries it did return, so a consumer
        # can pass on a partial read instead of presenting it as a whole one.
        'notice': notice,
        'position': {'cursor': cursor, 'kind': KIND_JOURNALD},
        # A cursor whose entry has rotated out of the journal makes journalctl
        # resume at the closest one it still holds, without saying so, so there
        # is nothing here that could be reported as a restart.
        'restarted': False,
        # Neither storage has a rotation of its own a caller could ask for.
        'rotated': [],
        # After a cursor every entry was read and the cap dropped the oldest; without
        # one journalctl applied the cap itself, and a full result may have been cut.
        'truncated': read > len(lines)
        if resumed
        else bool(max_lines) and len(lines) >= max_lines,
    }


# `<engine> logs --timestamps` prefixes every line with an RFC 3339 timestamp.
# Verified against podman 5 on Fedora 44, which renders it in local time with an
# explicit offset (`2026-08-28T15:30:26.241557000+02:00`) rather than in UTC.
_CONTAINER_TIMESTAMP_REGEX = re.compile(
    r'^(\d{4}-\d{2}-\d{2}T[\d:.]+(?:Z|[+-][\d:]+)) ?'
)


# What an unstamped entry, or a stored timestamp that cannot be read, sorts as.
_CONTAINER_TIME_UNKNOWN = (
    datetime.datetime.min.replace(tzinfo=datetime.timezone.utc),
    0,
)

# The fractional seconds of a container timestamp, taken apart from the rest
# because a datetime holds microseconds only.
_CONTAINER_FRACTION_REGEX = re.compile(r'\.(\d+)')


def _container_time(stamp):
    """The moment a container timestamp names, comparable across UTC offsets.

    Podman writes local time with its offset, so the same wall clock hour comes twice when
    daylight saving time ends (`02:59+02:00`, then `02:10+01:00`) and the strings no longer
    sort in the order they were written. It also writes nine fractional digits (libpod
    `LogTimeFormat`), more than a datetime holds, and a burst can put two entries into the
    same microsecond. Returns the second as an aware datetime together with the
    nanoseconds, so both the offset and the full precision count.
    """
    if not stamp:
        return _CONTAINER_TIME_UNKNOWN
    match = _CONTAINER_FRACTION_REGEX.search(stamp)
    nanoseconds = int(match.group(1)[:9].ljust(9, '0')) if match else 0
    whole = _CONTAINER_FRACTION_REGEX.sub('', stamp, count=1)
    try:
        return lftime.timestr2datetime(whole, pattern='iso8601'), nanoseconds
    except ValueError:
        return _CONTAINER_TIME_UNKNOWN


def _read_container(engine, target, position, max_lines, since, timeout):
    """Read the lines a container logged since `position`, resuming at a timestamp."""
    success, target = shell.safe_cli_value(target, 'container name')
    if not success:
        return False, target
    last = position.get('timestamp') if position else None
    start = last or since
    cmd = [engine, 'logs', '--timestamps']
    if start:
        # kubectl spells the same thing differently: its own `--since` takes a
        # duration, and only `--since-time` takes a timestamp.
        cmd.append(
            f'--since-time={start}' if engine == 'kubectl' else f'--since={start}'
        )
    if max_lines:
        cmd.append(f'--tail={max_lines}')
    cmd.append(target)
    success, result = shell.shell_exec(
        cmd, timeout=timeout, max_bytes=MAX_COMMAND_OUTPUT_BYTES
    )
    if not success:
        return False, result
    stdout, stderr, retc = result
    if retc != 0:
        return False, (
            f'`{" ".join(cmd)}` exited with error ({retc}, {stderr.strip()}).'
        )
    # An application in a container commonly logs to stderr, so both streams are
    # log content. A failure of the engine itself is caught by the exit code
    # above and never reaches this point.
    entries = []
    stamp = last or ''
    for line in stdout.splitlines() + stderr.splitlines():
        match = _CONTAINER_TIMESTAMP_REGEX.match(line)
        if match:
            stamp = match.group(1)
            line = line[match.end() :]
        # A line the engine did not stamp keeps the stamp of the one before it,
        # so it stays next to the entry it belongs to instead of sorting away.
        entries.append((stamp, line))
    # The two streams are read one after the other, so put them back into the
    # order they were written in, by the moment and not by the string, which
    # changes its offset when daylight saving time ends.
    entries.sort(key=lambda entry: _container_time(entry[0]))
    if last:
        # `--since` is inclusive, verified against podman 5: asking for the
        # timestamp of the last line read returns that very line again. Dropping
        # it here is what keeps the repeat away, rather than moving the stored
        # timestamp on, which would skip an entry written in the same instant.
        last_time = _container_time(last)
        entries = [entry for entry in entries if _container_time(entry[0]) > last_time]
    return True, {
        'fidelity': FIDELITY_APPROXIMATE,
        'kind': KIND_CONTAINER,
        'label': f'{engine}:{target}',
        'lines': [line for _, line in entries],
        # An engine writes its diagnostics to the same stream the container logs
        # to, so there is no way to tell them apart and nothing to report here.
        'notice': '',
        'position': {
            'kind': KIND_CONTAINER,
            'timestamp': entries[-1][0] if entries else last,
        },
        'restarted': False,
        # Neither storage has a rotation of its own a caller could ask for.
        'rotated': [],
        'truncated': bool(max_lines) and len(entries) >= max_lines,
    }


def read(
    source,
    position=None,
    allowed_roots=None,
    max_lines=None,
    rotated=0,
    since=None,
    timeout=DEFAULT_TIMEOUT,
):
    """
    Read the lines a log grew by since the previous run.

    Hands back the lines together with an opaque position. Store that position and pass it in
    again on the next run, and only what was written in between comes back. Pass none, and the
    log is read from a starting point that `since` and `max_lines` bound.

    Rotation, truncation and a rewrite in place are recognized for a file. After a rotation the
    file the previous run stopped in is looked for among the predecessors and read on from the
    stored offset, followed by the predecessors rotated after it and the live file, so nothing
    written in between is lost. Where that file is gone, was compressed into a new one, or was
    rewritten, and after a truncation or a rewrite in place, the file is read from its
    beginning with `restarted` set, because everything after the stored offset is gone. The
    other storages have no equivalent, which is why `restarted` is never set for them.

    A last line without a line end is held back until it is complete, because its writer is in
    the middle of it. A rotated file never grows again, so there it is read as it is.

    One run is bounded. A line over `MAX_LINE_BYTES` is cut to that many bytes and the rest of
    it is skipped, and what is read out of rotated files and of a compressed source stops at
    `MAX_ROTATED_BYTES` uncompressed. Both are reported in `notice`, and `truncated` is set,
    because whoever can create a file in a log directory decides what is in it.

    Parameters
    ----------
    source : str
        Where the log is kept. Either a path to a file, or a prefixed target: `systemd:UNITNAME`,
        `docker:NAME`, `kubectl:NAME` or `podman:NAME`. See `parse()`.
    position : dict, optional
        The `position` of the previous call, or None to start fresh. A position taken from
        another storage than the one `source` names is ignored rather than misread.
    allowed_roots : iterable of str, optional
        Directories a file source is allowed to resolve into. A path outside them is refused
        instead of read. Applies to file sources only, and confines nothing when left at None,
        which is why anything running with elevated privileges should name them.
    max_lines : int, optional
        At most this many lines are returned, the most recent ones. The position still advances
        past everything that was there, so the lines a cap dropped do not come back on the next
        run. Defaults to None, which is no cap. Where rotated files are read as well, the cap
        applies to all of them together rather than to each one. A unit resumed at a cursor
        hands over everything logged since, and the cap is applied to that, so a unit that
        logged a lot while the caller was not running costs a correspondingly long read.
    rotated : int, optional
        How many rotated predecessors of a file source to read ahead of it, most recent first.
        Defaults to 0, which reads the file alone. Only a read that has no `position` honours
        this, because a caller that resumes where it stopped has seen those lines already. See
        the notes on which files count as a predecessor.
    since : str, optional
        Where to start when there is no position: a timestamp for a container, and anything
        `journalctl --since` accepts for a unit, for example `-8h`. Ignored for a file, which
        always starts at its beginning. Defaults to None, which reads the current boot of a unit
        and the whole log of a container.
    timeout : int, optional
        Seconds to wait for `journalctl` or the container engine. Defaults to 8. Not used for a
        file, which is read directly.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if the log could be read, otherwise False.
          - tuple[1] (**dict | str**): If unsuccessful, an error message string. If successful, a
            dict:

            - `fidelity` (`str`): `FIDELITY_EXACT` if the position names the exact spot the run
              stopped at, `FIDELITY_APPROXIMATE` if a line can repeat across runs.
            - `kind` (`str`): the storage the lines came from.
            - `label` (`str`): the source in words, for a consumer that reports where it read.
            - `lines` (`list` of `str`): the new lines, without their line endings, and for a
              container without the timestamp the engine prefixes them with.
            - `notice` (`str`): what the source said about the read itself rather than about what
              was logged, for example that it only showed what the caller is allowed to see.
              Empty where there is nothing to report. A source that says so and returns nothing
              at all fails instead, because an empty result would read as a quiet log.
            - `position` (`dict`): to store and pass back on the next call.
            - `restarted` (`bool`): True if the stored position had become meaningless and the
              source was read from its beginning, so the lines are not only the new ones.
            - `rotated` (`list` of `str`): the rotated files that were read ahead of the source,
              in the order their lines appear. Empty unless `rotated` asked for them, or a
              rotation since the stored position was resumed in.
            - `truncated` (`bool`): True if `max_lines` dropped lines from the result.

    Notes
    -----
    - The position is a plain dict of strings and numbers, so it serializes to JSON and fits in
      a single column of a state database.
    - A consumer that stores the position has to survive not finding one, because the first run
      after deploying it has none, and so has the run after the log was rotated. Reading a whole
      log at once is what `max_lines` and `since` are there to bound.
    - A file is read in binary and decoded as UTF-8, falling back to Latin-1 for the whole line
      on any invalid byte, so a log that is not valid UTF-8 is reported rather than raising at
      the point where the result is printed.
    - A predecessor is a regular file in the same directory whose name is the source's plus a
      suffix of digits and separators, as a rotator writes it (`error.log.1`,
      `error.log-20260828`), optionally compressed with gzip, xz or bzip2. Ordering is by
      modification time, so both a counted and a dated scheme come out chronologically. A
      symlink is never followed, a rotator told to compress with something else is left alone,
      and a rotator configured to move its output to another directory is out of reach. One
      that could not be read is named in `notice` instead of failing the whole call, so the
      live file is still reported.

    Examples
    --------
    >>> success, result = read('/var/log/messages', allowed_roots=['/var/log'])
    >>> success, result = read('/var/log/messages', max_lines=30000, rotated=1)
    >>> success, result = read('systemd:sshd', position=stored, since='-8h')
    >>> success, result = read('podman:database', position=stored, max_lines=1000)
    """
    success, result = parse(source)
    if not success:
        return False, result
    kind, engine, target = result
    if position and position.get('kind') and position.get('kind') != kind:
        # The source changed shape, a file became a unit for example. The stored
        # position says nothing about the new one, so start over rather than
        # read a byte offset as a timestamp.
        position = None
    if kind == KIND_CONTAINER:
        return _read_container(engine, target, position, max_lines, since, timeout)
    if kind == KIND_JOURNALD:
        return _read_journald(target, position, max_lines, since, timeout)
    # A caller that resumes where it stopped has seen the predecessors already,
    # and the position it gets back describes the live file alone, so there
    # would be no way to remember how much of a predecessor was consumed.
    return _read_file(
        target, position, allowed_roots, max_lines, 0 if position else rotated
    )


def read_many(
    sources,
    allowed_roots=None,
    dedup_key=None,
    max_lines=None,
    rotated=0,
    since=None,
    timeout=DEFAULT_TIMEOUT,
):
    """
    Read several logs as one window.

    An application does not always keep what it has to say in one file. A web server writes a
    log per virtual host next to the one it writes about itself, and an administrator watching
    a service may want the file and the unit read together. Reading them one call at a time and
    stitching the results together is the same work every time, and getting it slightly wrong
    is easy: the same file named twice, or reached once by its own path and once through a
    symlink, would count everything in it twice.

    For a consumer that reads a window rather than resumes where it stopped, which is why this
    takes no position; `read()` is the one to use for reading a log incrementally.

    Parameters
    ----------
    sources : iterable of str
        What to read, each as `read()` takes it.
    allowed_roots : iterable of str, optional
        Handed to `read()` for every source.
    dedup_key : callable, optional
        Called with one line and returning what identifies
        the event in it, for example its time and the message without the prefix a log transport
        put in front of it (`strip_syslog_prefix()`). A line of a later source whose key a line
        of an earlier one already produced is dropped, which is what keeps a file and the journal
        of the same unit from counting the same event twice. Lines of one and the same source are
        never dropped against each other: an application repeating a message is what a rate is
        counted from. Without it nothing is dropped.
    max_lines : int, optional
        At most this many lines **per source**, the most recent ones. A window over several logs
        is therefore at most this many times the number of sources, which is what keeps one busy
        log from crowding out a quiet one that had the interesting line.
    rotated : int, optional
        Handed to `read()` for every source.
    since : str, optional
        Handed to `read()` for every source.
    timeout : int, optional
        Handed to `read()` for every source, each in its own right.

    Notes
    -----
    - Every source has its own byte budget (see `read()`), so N sources can read N times
      `MAX_ROTATED_BYTES` out of their rotated files.

    Returns
    -------
    tuple
          - tuple[0] (**bool**): True if at least one source could be read, otherwise False.
          - tuple[1] (**dict | str**): If unsuccessful, a message naming what went wrong with
            each source. If successful, a dict:

            - `lines` (`list` of `str`): Every source's lines, in the order the sources were
              given.
            - `duplicates` (`int`): How many lines `dedup_key` identified as an event an
              earlier source had already delivered. 0 without `dedup_key`.
            - `failed` (`list` of `str`): One message per source that could not be read at all.
              A source that failed is skipped rather than fatal, because a log an application
              has not written yet is normal where several are read - but the window then has a
              hole in it, which is a different matter from `notice` and is reported apart from
              it for that reason.
            - `notice` (`str`): What the sources said about the read itself rather than about
              what they hold, for a consumer to pass on.
            - `sources` (`list` of `dict`): What `read()` returned per source that could be read,
              in the same order, so a consumer can name each one and what came from it.
            - `truncated` (`bool`): Whether any source stopped at `max_lines`.

    Notes
    -----
    - A source named twice is read once, and so is one reached by two paths that resolve to the
      same file.

    Examples
    --------
    >>> success, result = read_many(
    ...     ['/var/log/httpd/error_log', 'systemd:httpd.service']
    ... )
    """
    lines = []
    notices = []
    failures = []
    read_sources = []
    seen = set()
    seen_keys = set()
    duplicates = 0
    for source in sources:
        success, parsed = parse(source)
        if not success:
            failures.append(parsed)
            continue
        kind, _, target = parsed
        # A file is identified by where it resolves to, so naming it twice, or
        # reaching it once directly and once through a symlink, reads it once.
        identity = os.path.realpath(target) if kind == KIND_FILE else source
        if identity in seen:
            continue
        seen.add(identity)
        success, result = read(
            source,
            allowed_roots=allowed_roots,
            max_lines=max_lines,
            rotated=rotated,
            since=since,
            timeout=timeout,
        )
        if not success:
            failures.append(result)
            continue
        kept = result['lines']
        if dedup_key is not None:
            # Against the sources read so far, never against this one itself, so
            # an application repeating a message keeps being counted as often as
            # it repeated it.
            kept = []
            keys = []
            for line in result['lines']:
                key = dedup_key(line)
                if key in seen_keys:
                    duplicates += 1
                    continue
                keys.append(key)
                kept.append(line)
            seen_keys.update(keys)
        lines.extend(kept)
        if result['notice']:
            notices.append(result['notice'])
        read_sources.append(result)
    if not read_sources:
        return False, ' '.join(failures) if failures else 'No log source given.'
    return True, {
        'duplicates': duplicates,
        'failed': failures,
        'lines': lines,
        'notice': ' '.join(notices),
        'sources': read_sources,
        'truncated': any(item['truncated'] for item in read_sources),
    }


# The timestamp a transport puts in front of what the application wrote:
# `journalctl --output=short-iso` writes `2026-08-28T17:16:18+0200 host unit[pid]:`
# and this module strips the container engine's equivalent while reading, so
# what is left to recognize here is an ISO 8601 moment at the very start of a
# line. Deliberately anchored: an ISO timestamp further along is part of the
# message, not of the line's own header.
# The hour is allowed to be padded with a space rather than with a zero, which
# is not ISO 8601 but is what MariaDB writes: `2026-08-29  6:00:21` before ten
# in the morning and `2026-08-29 15:01:47` after it. Without this a consumer
# reading such a log would find no time on any line for a third of every day.
# Verified against MariaDB 11.8.
_LINE_TIMESTAMP_REGEX = re.compile(
    r'^(\d{4}-\d{2}-\d{2}[T ][ \d]\d:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?)'
)

# The other timestamp a syslog daemon puts at the head of a line, the traditional
# one of RFC 3164: `Aug 28 19:34:14`, with the day padded to two columns by a
# space. Which of the two formats a host writes is up to the daemon and its
# configuration rather than up to the application, so both are read here.
# Verified against rsyslog with the configuration its package ships: 8.2102 on
# Rocky 8 and 8.2510 on Rocky 9 write this format, 8.2504 on Debian 13 writes the
# ISO one above.
_SYSLOG_TIMESTAMP_REGEX = re.compile(
    r'^([A-Z][a-z]{2}) {1,2}(\d{1,2}) (\d{2}):(\d{2}):(\d{2})(?:\s|$)'
)

# The month names of that format, read from a table rather than with `%b`,
# because `strptime()` reads them in whatever locale the process happens to be
# in, while a syslog daemon writes them in English on every host.
_SYSLOG_MONTHS = {
    'Apr': 4,
    'Aug': 8,
    'Dec': 12,
    'Feb': 2,
    'Jan': 1,
    'Jul': 7,
    'Jun': 6,
    'Mar': 3,
    'May': 5,
    'Nov': 11,
    'Oct': 10,
    'Sep': 9,
}


def sort_by_time(lines, parse_line=None):
    """
    Return the lines in the order they were written, whichever log each of them came from.

    A consumer reading several logs as one window holds them one log after the other, so the
    newest line of the first log sits in the middle: "the last line" is then not the newest
    one, and reporting it as the latest event names something days old. Sorting once puts that
    right for everything built on the order - the line a summary quotes as the last, and the
    order a listing renders in.

    A line whose time cannot be read keeps its place relative to the other unreadable ones and
    sorts before everything dated, because the alternative - guessing a time for it - would
    move it somewhere it does not belong.

    Parameters
    ----------
    lines : iterable of str
        The lines, in the order they were read.
    parse_line : callable, optional
        Handed to `timestamp()` for a format it cannot
        read on its own.

    Returns
    -------
    list of str
        The lines, oldest first.

    Examples
    --------
    >>> sort_by_time(['Aug 29 11:15:53 h x[1]: b', 'Aug 28 11:15:53 h x[1]: a'])[0][:6]
    'Aug 28'
    """
    decorated = []
    for index, line in enumerate(lines):
        moment = timestamp(line, parse_line)
        decorated.append((moment is not None, moment, index, line))
    # Naive and aware values cannot be compared, so the sort runs over the naive
    # form of each: an offset moves a line by hours at worst, a `TypeError`
    # takes the whole check down.
    if len({moment.tzinfo is None for _, moment, _, _ in decorated if moment}) > 1:
        decorated = [
            (dated, moment.replace(tzinfo=None) if moment else None, index, line)
            for dated, moment, index, line in decorated
        ]
    decorated.sort(
        key=lambda item: (item[0], item[1] or datetime.datetime.min, item[2])
    )
    return [line for _, _, _, line in decorated]


def strip_syslog_prefix(line):
    """
    Return what an application wrote, without the prefix a log transport put in front of it.

    The same event reaches a consumer twice where a file and the journal of the same unit are
    both read, and in two different prefixes: `Aug 29 11:15:53 host sshd[1]: ...` from what
    rsyslog wrote, `2026-08-29T11:15:53+0200 host sshd[1]: ...` from `journalctl`. What
    follows the prefix is byte for byte the same, which is what makes it comparable.

    Parameters
    ----------
    line : str
        One line of a log.

    Returns
    -------
    str
        The message, or the line unchanged where it carries no such prefix - which is
        what an application writing its own timestamps into a file of its own does.

    Examples
    --------
    >>> strip_syslog_prefix(
    ...     'Aug 29 11:15:53 host sshd[1]: Server listening on 0.0.0.0.'
    ... )
    'Server listening on 0.0.0.0.'

    >>> strip_syslog_prefix('[28-Aug-2026 15:20:15] ERROR: failed to ptrace(ATTACH)')
    '[28-Aug-2026 15:20:15] ERROR: failed to ptrace(ATTACH)'
    """
    return _SYSLOG_PREFIX_REGEX.sub('', line, count=1)


def syslog_identifier(line):
    """
    Return the identifier a log transport wrote in front of a line, or None.

    That is the program name a syslog daemon or journald puts between the host and the
    message, without the process id: `sshd` in `Aug 29 11:15:53 host sshd[1]: ...`. A consumer
    reading the journal of a unit needs it to tell the application's own lines from the ones
    systemd writes about the unit ("Starting ...", "Started ..."), which are about the service
    rather than from it.

    Parameters
    ----------
    line : str
        One line of a log.

    Returns
    -------
    str | None
        The identifier, or None where the line carries no transport prefix.

    Examples
    --------
    >>> syslog_identifier('Aug 29 11:15:53 host sshd[1]: Server listening on 0.0.0.0.')
    'sshd'

    >>> syslog_identifier('[28-Aug-2026 15:20:15] ERROR: failed') is None
    True
    """
    match = _SYSLOG_PREFIX_REGEX.match(line)
    if not match:
        return None
    return match.group('identifier')


def _syslog_datetime(match):
    """Turn a match of `_SYSLOG_TIMESTAMP_REGEX` into a moment, or into None.

    The format carries no year, so it is inferred: a line cannot have been written in the
    future, so a date that lies ahead belongs to the year before. A day of slack absorbs a
    host whose clock runs a little behind the one that wrote the line, and a line that lands
    outside the year this leaves is reported as undated rather than mis-dated, which is all
    a format without a year can honestly say about it.
    """
    month = _SYSLOG_MONTHS.get(match.group(1))
    if month is None:
        return None
    now = datetime.datetime.now()
    earliest = now - datetime.timedelta(days=366)
    latest = now + datetime.timedelta(days=1)
    for year in (now.year, now.year - 1):
        try:
            logged_at = datetime.datetime(
                year,
                month,
                int(match.group(2)),
                int(match.group(3)),
                int(match.group(4)),
                int(match.group(5)),
            )
        except ValueError:
            # 31 April, and 29 February in a year that has no such day. The
            # latter is why the year before is tried as well rather than only
            # when the date lies ahead.
            continue
        if earliest <= logged_at <= latest:
            return logged_at
    return None


def timestamp(line, parse_line=None):
    """
    Return the moment a log line was written, as a naive datetime in local time.

    An application that stamps its own lines is the better source, because that stamp says when
    the event happened rather than when the line reached the transport, and because a transport
    is not always there. It is also the one this module cannot know, so a consumer passes it in.
    What is read here without being told is the timestamp the transport prefixes, which is what
    stays when an application logs through syslog and leaves its own out.

    Parameters
    ----------
    line : str
        One log line.
    parse_line : callable, optional
        The consumer's reader for the application's own timestamp. Takes the line and returns a
        `datetime.datetime` or None. Tried first; the transport's timestamp is the fallback.

    Returns
    -------
    datetime.datetime | None
        When the line was written, in local time and without a timezone attached, so it compares
        against `datetime.now()`. None where neither reader found a timestamp, which is not an
        error: a line an application wrote while starting up commonly carries none.

    Notes
    -----
    - A transport timestamp that carries an offset is converted to local time, so lines written
      before and after a daylight saving change stay in order.
    - Both formats a syslog daemon writes are read: the ISO 8601 moment, and the traditional
      one of RFC 3164 (`Aug 28 19:34:14`). The traditional one names no year, so the year is
      inferred from the assumption that a line was not written in the future.

    Examples
    --------
    The result is naive local time, so the same line reads differently depending on
    the host's timezone. On a host running UTC:

    >>> timestamp('2026-08-28T17:16:18+0200 host httpd[20]: [ssl:error] AH02032: ...')
    datetime.datetime(2026, 8, 28, 15, 16, 18)
    >>> timestamp('Aug 28 19:34:14 host sshd[193]: Server listening on port 22.')
    datetime.datetime(2026, 8, 28, 19, 34, 14)
    """
    if parse_line:
        logged_at = parse_line(line)
        if logged_at is not None:
            return logged_at
    match = _LINE_TIMESTAMP_REGEX.match(line)
    if not match:
        match = _SYSLOG_TIMESTAMP_REGEX.match(line)
        return _syslog_datetime(match) if match else None
    try:
        # A space-padded hour is put back to what the parsers expect.
        return lftime.datetime2local(
            lftime.timestr2datetime(
                match.group(1).replace('  ', ' 0').replace('T ', 'T0'),
                pattern='iso8601',
            )
        )
    except ValueError:
        # also a moment that cannot be expressed in local time, such as Go's zero
        # time `0001-01-01T00:00:00Z` west of UTC
        return None