Streaming big logs with generators

Generators, rotated gzip files, Ubuntu 26.04 log lines and constant memory.

A log you have to search during an incident is usually too big to read into memory in one go, split over rotated and compressed files, and not entirely clean. This lesson builds a failed-login counter that reads every rotation of an Ubuntu 26.04 auth.log one line at a time with generators, survives bytes that are not UTF-8, writes one JSON object per line for the next tool, and reports the lines it could not parse instead of hiding them. You will then run it and a read-everything version under the same memory limit and watch one of them get killed, and read the systemd journal as a structured source with journalctl -o json.

Generators: values on demand

A function with return computes its whole result and hands it back once. A function that contains yield is a generator function: calling it returns a generator object, and the body runs only when someone asks for the next value, stopping again at each yield. The lab works in ~/ps-logs; unpack the lesson files in your home directory, or create the files as they appear:

"""Watch a generator run: each value is produced only when the loop asks for it."""
 
 
def numbered(lines):
    for number, line in enumerate(lines, start=1):
        print(f"  (generator: producing line {number})")
        yield number, line
 
 
lines = ["sshd-session: Failed password", "CRON: session opened", "sudo: session closed"]
gen = numbered(lines)
print("created:", type(gen).__name__)
print("next():", next(gen))
for number, line in gen:
    print("loop got:", number, line)
print("second pass:", list(gen))

Calling numbered(lines) printed nothing: the body had not started. next(gen) ran it up to the first yield, and the for loop took the other two values the same way, each produced just before it was used. The last line is the trap. A generator runs once; after it is exhausted, every further loop over it gets nothing, silently. If you need the values twice, keep them in a list or create a new generator. A generator expression, (line for line in f if "Failed" in line), is the one-line form, and sum(1 for line in f) counts lines without storing any of them.

A test log in the Ubuntu 26.04 format

On Ubuntu 26.04, rsyslog writes /var/log/auth.log with RFC 3339 timestamps (2026-09-27T15:36:52.936856+00:00: date, time with microseconds, offset from UTC). Since OpenSSH 9.8 the process that handles a connection is sshd-session, so a failed login reads sshd-session[PID]: Failed password for invalid user ...; the lab confirmed both on the machine itself, with a failed login against a throwaway sshd. auth.log belongs to syslog:adm with mode 0640, so reading it needs the adm group. Ubuntu's logrotate rules rotate it weekly and keep four old copies, compressing all but the newest (auth.log.1 stays plain, auth.log.2.gz and older are gzip files).

So that every count in this lesson is the same on your machine, the lab generates a log in that format. fake_events.py, in the lesson files, makes random but repeatable events: failed passwords from five test addresses with fixed weights, accepted keys, cron and sudo sessions. make_authlog.py writes auth.log and three rotations from those events. It is setup code, so skim it; the part that matters is the PLANTED list, three lines it plants on purpose:

"""Write a lab auth.log and three rotations in the Ubuntu 26.04 format.
 
Usage: make_authlog.py DIR LINES_PER_FILE
The same arguments always produce the same files.
"""
import gzip
import random
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
 
from fake_events import event
 
# (file, line number, raw bytes): lines the generator plants on purpose
PLANTED = [
    ("auth.log.2.gz", 7, b"<ts> web01 sshd-session[4410]: Failed password for invalid "
                         b"user adm\xe9n from 192.0.2.201 port 50122 ssh2"),
    ("auth.log.3.gz", 5, b"<ts> web01 sshd[812]: Failed password for root from "
                         b"203.0.113.42 port 40022 ssh2"),
    ("auth.log.3.gz", 9, b"Sep  6 00:01:40 web01 sshd[813]: Failed password for root from "
                         b"198.51.100.99 port 40023 ssh2"),
]
 
 
def main(argv: list[str]) -> int:
    out, per_file = Path(argv[0]), int(argv[1])
    rng = random.Random(per_file)
    when = datetime(2026, 9, 6, tzinfo=timezone.utc)
    out.mkdir(exist_ok=True)
    for name in ["auth.log.3.gz", "auth.log.2.gz", "auth.log.1", "auth.log"]:  # oldest first
        opener = gzip.open if name.endswith(".gz") else open
        with opener(out / name, "wb") as f:
            for number in range(1, per_file + 1):
                when += timedelta(seconds=rng.randrange(1, 40), microseconds=rng.randrange(10**6))
                ts = when.isoformat(timespec="microseconds")
                line = event(rng, ts).encode()
                for planted_name, planted_number, raw in PLANTED:
                    if (planted_name, planted_number) == (name, number):
                        line = raw.replace(b"<ts>", ts.encode())
                f.write(line + b"\n")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

random.Random(per_file) is a random number generator with a fixed seed, which is why the same arguments give the same files. The first planted line has the byte 0xe9 in a user name, "é" in Latin-1 and invalid in UTF-8; cat -v shows it as M-i. The second is written by sshd, as OpenSSH wrote before 9.8, which a parser must still accept in old rotations. The third is in the traditional syslog format: no year, no time zone, the way rsyslog writes when a configuration selects its traditional template, and the way many older hosts and appliances still send logs.

Reading rotations as one stream

"""Generators that read auth logs one line at a time, plain or gzip-compressed."""
import gzip
import re
from collections.abc import Iterable, Iterator
from pathlib import Path
 
FAILED = re.compile(
    r"""
    (?P<ts>[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9:.]+[+-][0-9]{2}:[0-9]{2})  # RFC 3339
    \ \S+\ sshd(?:-session)?\[[0-9]+\]:      # sshd-session since OpenSSH 9.8
    \ Failed\ password\ for\ (?:invalid\ user\ )?(?P<user>.+)
    \ from\ (?P<ip>[0-9A-Fa-f:.]+)\ port\ [0-9]+\ ssh2$
    """,
    re.VERBOSE | re.ASCII,
)
 
 
def read_lines(path: Path) -> Iterator[str]:
    """Yield the lines of one log; bytes that are not UTF-8 become U+FFFD."""
    opener = gzip.open if path.suffix == ".gz" else open
    with opener(path, "rt", encoding="utf-8", errors="replace") as f:
        for line in f:
            yield line.rstrip("\n")
 
 
def numbered_lines(paths: Iterable[Path]) -> Iterator[tuple[Path, int, str]]:
    """Yield (path, line number, line) for every line of every file, in order."""
    for path in paths:
        for number, line in enumerate(read_lines(path), start=1):
            yield path, number, line

read_lines() picks gzip.open for .gz files and open for the rest. In mode "rt" both return text lines, so the rest of the code never knows the difference, and errors="replace" turns bad bytes into U+FFFD, as in the files lesson. The yield sits inside the with block: the file stays open while the loop is consuming lines and is closed when the generator finishes. numbered_lines() feeds from read_lines() for one file after another, so a chain of generators passes a single line through at a time, whatever the total size.

The pattern is the one from the regex lesson, now anchored at the start on the RFC 3339 timestamp, accepting sshd and sshd-session, and with an address class wide enough for IPv6. Forgetting the gzip half is a common first bug:

Every gzip file starts with the bytes 1f 8b, and 0x8b is not valid UTF-8, hence the error at position 1. With errors="replace" added, the same mistake would not fail at all: you would get lines of binary noise and a count of zero. Choose the reader by file type before you decide how to handle bad bytes.

Counting with small state, NDJSON out

"""Count failed SSH logins per source address across rotated auth logs.
 
Prints one JSON object per address (NDJSON) on stdout. Lines that mention
"Failed password" but do not parse are reported on stderr, and the exit
status is then 1.
"""
import json
import sys
from collections import Counter
from pathlib import Path
 
from logread import FAILED, numbered_lines
 
 
def main(argv: list[str]) -> int:
    if not argv:
        print("usage: failed_logins.py LOG...", file=sys.stderr)
        return 2
    failures = Counter()
    users = {}
    last_seen = {}
    unparsed = 0
    try:
        for path, number, line in numbered_lines(Path(arg) for arg in argv):
            if "Failed password" not in line:
                continue
            m = FAILED.match(line)
            if m is None:
                unparsed += 1
                print(f"unparsed: {path}:{number}: {line}", file=sys.stderr)
                continue
            ip = m["ip"]
            failures[ip] += 1
            users.setdefault(ip, set()).add(m["user"])
            last_seen[ip] = max(last_seen.get(ip, ""), m["ts"])
    except OSError as err:
        print(f"failed_logins: {err}", file=sys.stderr)
        return 2
    for ip, count in failures.most_common():
        record = {"ip": ip, "failures": count, "users": sorted(users[ip]), "last_seen": last_seen[ip]}
        print(json.dumps(record))
    return 1 if unparsed else 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

The state the loop keeps is one counter, one set of user names and one timestamp per source address. Its size depends on how many distinct addresses attacked, not on how many lines the logs hold. max() on the timestamp strings works because RFC 3339 times with the same offset sort as text in time order.

The output is NDJSON (newline-delimited JSON): one complete JSON object per line, written with json.dumps. A consumer can read it line by line, as it was produced, and json.dumps escapes whatever a user name contains, so the name with the bad byte arrives as "adm\ufffdn" and a name with quotes or control characters cannot break the line. The traditional-format line did not match, so it was printed on stderr with its file and line number and the exit status is 1: the next person learns that part of the input was not counted. The sshd[812] line was counted, against 203.0.113.42.

2>/dev/null drops the unparsed-line report, and jq filters the stream: select(.failures > 500) keeps the objects with more than 500 failures, and {ip, failures} prints only those two fields (-M turns off colour, -c keeps one object per line). jq is not part of a default Ubuntu Server install; sudo apt install jq adds it. A file that cannot be opened stops the run with exit status 2 and one line instead of a traceback.

Streaming versus reading everything

slurp.py does the same count the way many first versions do: readlines() loads every line of every file into one list, and the loop runs over the list:

"""The same count as failed_logins.py, but every file is read into one list first."""
import gzip
import sys
from collections import Counter
 
from logread import FAILED
 
 
def read_all(path: str) -> list[str]:
    opener = gzip.open if path.endswith(".gz") else open
    with opener(path, "rt", encoding="utf-8", errors="replace") as f:
        return f.readlines()  # every line of the file, in memory at once
 
 
lines = []
for path in sys.argv[1:]:
    lines += read_all(path)
failures = Counter()
for line in lines:
    m = FAILED.match(line.rstrip("\n"))
    if m:
        failures[m["ip"]] += 1
print(f"{len(lines)} lines read, {len(failures)} source addresses")

A million lines, 66 MB on disk because two files are compressed. GNU time (the program /usr/bin/time, not the shell keyword) printed the peak resident memory with %M: about 190 MB for the list, about 14 MB for the stream. Every line in the list is a separate str object with its own overhead on top of the text, so the list costs more than the files themselves. The stream's cost stays the same for ten times the data. (The streaming run exits 1 because of the planted traditional line.)

Now give both the same memory budget. systemd-run starts a command as a temporary systemd service, and -p MemoryMax=64M puts it in a control group the kernel will not let grow past 64 MiB. --uid="$USER" runs it as you, --same-dir in this directory, -E PATH="$PATH" finds the same python3 you use, and --wait --pipe connects its output to your terminal and waits for the result. sudo is needed to create the unit:

The read-everything version was killed by the kernel's out-of-memory killer less than a tenth of a second in: Finished with result: oom-kill, signal 9, and no output at all. On a production host with no limit, the same program pushes everything else on the machine towards swap and the OOM killer. The streaming version finished within the same limit, wrote all six records, and reported the unparsed line. Its Memory peak of about 7 MB is lower than what time reported because systemd counts only the memory charged to the unit's control group, not the whole resident size of the process.

journalctl -o json: a structured source

The systemd journal stores each log entry as fields rather than a line of text, and journalctl -o json prints one JSON object per entry, NDJSON again. logger writes an entry from the shell; -t sets its tag. You can read the whole system journal here because the lab user is in the adm group (so is the first account created on an Ubuntu install; id shows your groups, and without adm you see only your own entries):

--output-fields limited the entry to the fields named, plus the ones journalctl always adds (the cursor, timestamps, boot ID). Fields starting with _ such as _PID and _UID are added by journald itself, so a program cannot forge them the way it can write any text into a syslog line. PRIORITY 5 is "notice", logger's default. __REALTIME_TIMESTAMP is microseconds since 1970 in UTC, and every value is a string. Structured does not mean clean:

"""Read journalctl -o json output on stdin and print one readable line per record."""
import json
import sys
from datetime import datetime, timezone
 
 
def records(stream):
    for number, line in enumerate(stream, start=1):
        try:
            yield json.loads(line)
        except json.JSONDecodeError as err:
            print(f"journal_messages: input line {number}: {err}", file=sys.stderr)
 
 
def message_text(record: dict) -> str:
    message = record.get("MESSAGE")
    if isinstance(message, list):  # not UTF-8, or control bytes: exported as byte values
        return bytes(message).decode("utf-8", errors="replace")
    return message or ""
 
 
for record in records(sys.stdin):
    when = datetime.fromtimestamp(int(record["__REALTIME_TIMESTAMP"]) / 1e6, tz=timezone.utc)
    tag = record.get("SYSLOG_IDENTIFIER", "?")
    stamp = when.isoformat(timespec="seconds")
    print(f"{stamp} {tag}[{record.get('_PID')}] uid={record.get('_UID')}: {message_text(record)}")

A field that is not valid UTF-8, or that holds non-printable characters such as the escape byte of a terminal colour code, is exported as a list of byte values, not a string; the last command shows the raw field, with 233 (0xe9) in the middle. Code that expects a string would crash or print the list. message_text() checks the type, turns the list into bytes and decodes it with errors="replace". The records() generator reads standard input one entry at a time, so journalctl -o json | python3 journal_messages.py streams a journal of any size. The same approach works for journalctl -t sshd-session -o json, with no timestamp format to parse.

Try this

Write failures_by_user.py LOG... in ~/ps-logs with numbered_lines and FAILED from logread. Build a generator expression of matches, count user names with a Counter, and print the three most tried names as NDJSON: {"user": ..., "failures": ..., "share": ...}, where share is the name's fraction of all parsed failures, rounded to 3 places. Break it first: compute the total with sum(1 for m in matches if m) and then build the Counter from the same matches. On logs/auth.log* it prints nothing and exits 0: a wrong answer with no error. Fix it by taking the total from the Counter (sum(users.values())) and verify: root 2043 (0.341), admin 1816 (0.303) and oracle 757 (0.126), exit status 0.

Takeaway

Read logs through generators, one line at a time, with the reader chosen by file type and bad bytes replaced, and keep only state that grows with what you are counting. Write results as NDJSON, send the lines you could not parse to stderr and make them change the exit status.

Related