Stream gzip Logs with Python and Detect Truncation and CRC Errors

A historical log is compressed as task.log.gz, and extracting it just to count errors consumes extra disk space. A partially copied compressed file can be more confusing: reading its first few lines may succeed even though the file is incomplete.

This beginner-to-intermediate tutorial uses the Python standard library to stream a gzip log, count matching lines, and verify exit codes for truncation and checksum failures. Use completed log backups that stay unchanged while being read.

1. Method: read compressed input and report after completion

The official Python gzip documentation describes text mode and encoding parameters for gzip.open(). We use rt with strict decoding; compression, file, and decoding errors fail the task.

The sequence is: check the gzip header → decompress and count lines → reach the end and close → print JSON. Do not stop after the first match: unread data can still contain a failure. Checking a header alone cannot replace a full read.

The existing rg and Python log triage tutorial covers filtering and structured reporting. This article adds direct reading of compressed backups. --contains ERROR is a case-sensitive literal substring search; it does not parse severity levels or HTTP status codes.

2. Prepare the environment and test log

Use Python 3.9 or newer, with no third-party dependencies. The examples were tested on Linux with Python 3.11.4; the commands work in bash or zsh.

1. Create a fresh demo directory and run subsequent commands there. Choose a different directory name when repeating the experiment:

python3 --version
mkdir gzip-log-demo
cd gzip-log-demo
python3 - <<'PY_DATA'
import gzip
text = "INFO start\nERROR timeout\nINFO retry\nERROR failed"
with gzip.open("task.log.gz", "wt", encoding="utf-8", newline="\n") as file:
    file.write(text)
PY_DATA

2. The test log contains four lines with no newline after the last one. Save the next section as summarize_gzip.py, or download the complete script.

3. Complete implementation

"""Count lines in a completed gzip log; emit JSON only after a full read."""
import argparse
import gzip
import json
from pathlib import Path
import sys
import zlib


def summarize(path, contains, encoding):
    lines = matches = 0
    with path.open("rb") as raw:
        if raw.read(2) != b"\x1f\x8b":
            raise gzip.BadGzipFile("missing gzip header")
        raw.seek(0)
        with gzip.open(raw, "rt", encoding=encoding, errors="strict") as stream:
            for line in stream:
                lines += 1
                if contains is None or contains in line:
                    matches += 1
    return {"file": str(path), "lines": lines, "matches": matches}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("file", type=Path, help="completed .gz text log")
    parser.add_argument("--contains", help="case-sensitive literal substring")
    parser.add_argument("--encoding", default="utf-8", help="text encoding (default: utf-8)")
    args = parser.parse_args()
    if args.contains == "":
        parser.error("contains must not be empty")
    try:
        result = summarize(args.file, args.contains, args.encoding)
    except (OSError, EOFError, zlib.error, UnicodeError, LookupError) as exc:
        print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The code checks the two gzip signature bytes, then passes the same file object to the decompressor. Counting keeps counters rather than calling read() without a size or building a list of all lines. Memory still depends on the longest line and decompression buffers; this is not a fixed memory limit.

BadGzipFile inherits from OSError, so that branch handles both format and file-access errors. Invalid compressed data can also raise EOFError or zlib.error; both are caught. An invalid encoding name or strict decoding failure also returns 1.

4. Run and verify successful output

1. Display help, then count errors, all lines, and an absent substring:

python3 summarize_gzip.py --help
python3 summarize_gzip.py task.log.gz --contains ERROR
python3 summarize_gzip.py task.log.gz
python3 summarize_gzip.py task.log.gz --contains WARNING

The second command prints:

{"file": "task.log.gz", "lines": 4, "matches": 2}

Without --contains, both lines and matches are 4. Searching for WARNING returns zero matches and exit code 0: no match is a normal result. Blank lines count, and a final line needs no trailing newline.

2. Check the report fields:

python3 summarize_gzip.py task.log.gz --contains ERROR > report.json
python3 - <<'PY_CHECK'
import json
from pathlib import Path
result = json.loads(Path("report.json").read_text(encoding="utf-8"))
assert result == {"file": "task.log.gz", "lines": 4, "matches": 2}
print("OK: four lines, two matches")
PY_CHECK

Expect OK: four lines, two matches. These fields describe this complete read; they do not establish business-level completeness or authenticate the file’s source.

5. Create truncation and CRC failures

Modify only copies inside the demo directory. The following removes the eight-byte gzip trailer from one copy and changes a byte of the trailer CRC in another:

python3 - <<'PY_BAD'
from pathlib import Path
original = Path("task.log.gz").read_bytes()
Path("truncated.log.gz").write_bytes(original[:-8])
corrupt = bytearray(original)
corrupt[-8] ^= 1
Path("bad-crc.log.gz").write_bytes(corrupt)
PY_BAD
python3 summarize_gzip.py truncated.log.gz --contains ERROR > failed.json
printf 'exit=%s\n' "$?"
wc -c < failed.json
python3 summarize_gzip.py bad-crc.log.gz --contains ERROR
printf 'exit=%s\n' "$?"

Both runs should print ERROR: to standard error and show exit=1. wc should print 0, showing that the failed run wrote no successful JSON to standard output. Exact exception messages can vary with Python versions; do not rely on the full message to detect failure.

Additional checks covered a valid gzip containing empty content, CRLF, Unicode, concatenated gzip members, non-gzip input, missing files, and encoding errors. A full read still cannot prove all expected business records arrived: a complete compression of half a log can have a valid CRC.

The shell truncates report.json for > report.json before the program starts, including runs that later fail. To update an existing report, write a new temporary path and replace the report only after success:

if python3 summarize_gzip.py task.log.gz --contains ERROR > report.pending.json; then
  mv report.pending.json report.json
else
  rm -f report.pending.json
  printf 'Report update failed\n' >&2
fi

This snippet assumes a trusted demo directory and one runner. It preserves an existing report.json on failure; when embedding it in a job script, also return a nonzero exit code from the failure branch. See atomic JSON writes in Python for a fuller saving approach.

6. Troubleshooting and limits

Use completed backups rather than gzip files still being appended to. A .tar.gz decompresses to archive data, so it is not a plain-text log for this script. For creating and retaining rotated logs, see size-based Python log rotation; its default handler does not generate gzip backups automatically.

7. Summary

Reading gzip logs line by line avoids writing an extracted copy while keeping a simple JSON reporting interface. Verify full reads, strict decoding, and failure exit codes before connecting the tool to automation; replace existing reports only after success.