Log Analysis and Diagnostic Data Gathering Questions

Direct answer

Match each field positionally with a PCRE (Perl Compatible Regular Expression, the regex flavor used by grep -P, Perl, and Python's re module) pattern using named groups, one group per field. Named groups make the extraction self-documenting and let downstream code read match["client_ip"] instead of a numbered \1.

Approach

import re

PATTERN = re.compile(
    r'^(?P<client_ip>\S+) \S+ (?P<user>\S+) '
    r'\[(?P<timestamp>[^\]]+)\] '
    r'(?P<method>[A-Z]+) (?P<path>\S+) (?P<protocol>[A-Z]+/\d\.\d) '
    r'(?P<status>\d{3}) (?P<bytes>\d+|-)$'
)

line = '127.0.0.1 - frank [10/Oct/2024:13:55:36 -0700] GET /index.html HTTP/1.1 200 2326'

m = PATTERN.match(line)
if m:
    for k, v in m.groupdict().items():
        print(f"{k}: {v}")
else:
    print("no match")

Worked example (executed output)

client_ip: 127.0.0.1
user: frank
timestamp: 10/Oct/2024:13:55:36 -0700
method: GET
path: /index.html
protocol: HTTP/1.1
status: 200
bytes: 2326

The same named-group syntax works directly with grep -P on the command line (grep -oP prints only the matched text), and the equivalent extraction with awk, splitting on whitespace since this log line has no embedded quoted fields, looks like:

LINE='127.0.0.1 - frank [10/Oct/2024:13:55:36 -0700] GET /index.html HTTP/1.1 200 2326'
echo "$LINE" | awk '{
  ip=$1; user=$3;
  ts=$4" "$5; gsub(/[\[\]]/,"",ts);
  printf "client_ip=%s user=%s timestamp=%s method=%s path=%s protocol=%s status=%s bytes=%s\n", ip, user, ts, $6, $7, $8, $9, $10
}'

which prints client_ip=127.0.0.1 user=frank timestamp=10/Oct/2024:13:55:36 -0700 method=GET path=/index.html protocol=HTTP/1.1 status=200 bytes=2326, matching the Python extraction field for field. awk works here specifically because this log line has no spaces inside any field; a format with a quoted user-agent field (spaces included) would break plain whitespace splitting and need the regex approach instead.

Trade-offs & pitfalls

Timestamp parsing caveats: the bracketed timestamp 10/Oct/2024:13:55:36 -0700 is Apache's own format, not ISO 8601, so a generic date parser will not accept it as-is; you need a format string like %d/%b/%Y:%H:%M:%S %z (Python strptime) that matches it exactly. The trailing -0700 is a fixed UTC offset, not a named timezone, and does not account for daylight saving time changes on its own; if you're aggregating log lines from hosts in different timezones, convert every timestamp to UTC immediately after parsing rather than comparing the raw offset strings.

A second pitfall: this pattern assumes the log line has no quoted fields (no referrer, no user-agent), which is why it splits cleanly on whitespace. The moment a field can contain a space (a real Apache "combined" format log includes a quoted user-agent), whitespace-based extraction breaks and you need to match the quotes explicitly, as covered in the ELK grok-pattern version of this same problem.

Direct answer

Stream the file line by line and keep only a small dict of per-minute counters, never the raw lines, so memory stays bounded regardless of file size. Because the file can be out of order, the true 60-minute window boundary (which minute is "now") isn't known until you've seen the latest timestamp, so prune the counters dict opportunistically as later, larger timestamps arrive, rather than needing a full second pass over the file.

Approach

import json
from datetime import datetime, timedelta, timezone

WINDOW_MINUTES = 60


def _parse_ts(raw):
    if raw.endswith("Z"):
        raw = raw[:-1] + "+00:00"
    dt = datetime.fromisoformat(raw)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def per_minute_error_counts(path, service):
    """Stream a large NDJSON log file and return per-minute ERROR counts
    for `service` over the last WINDOW_MINUTES relative to the latest
    timestamp seen, as a dense list of (minute, count) tuples (one entry
    per minute, zero-filled).

    The file may be out of order. Rather than buffering every distinct
    minute seen across the whole file, the counts dict is pruned WHILE
    streaming: every time a new maximum timestamp arrives, anything
    older than (new_max - WINDOW_MINUTES) is dropped. Memory therefore
    stays close to O(WINDOW_MINUTES) even over a ~10GB file, at the cost
    of one bookkeeping check per line rather than a second pass.
    """
    counts = {}
    newest = None

    with open(path, "r") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                continue
            if rec.get("service") != service:
                continue
            raw_ts = rec.get("timestamp")
            if not raw_ts:
                continue
            try:
                ts = _parse_ts(raw_ts)
            except ValueError:
                continue

            if newest is None or ts > newest:
                newest = ts
                cutoff = newest - timedelta(minutes=WINDOW_MINUTES - 1)
                for bucket in [b for b in counts if b < cutoff.replace(second=0, microsecond=0)]:
                    del counts[bucket]

            if rec.get("level") == "ERROR":
                bucket = ts.replace(second=0, microsecond=0)
                counts[bucket] = counts.get(bucket, 0) + 1

    if newest is None:
        return []

    window_start = (newest - timedelta(minutes=WINDOW_MINUTES - 1)).replace(second=0, microsecond=0)
    result = []
    cur = window_start
    end = newest.replace(second=0, microsecond=0)
    while cur <= end:
        result.append((cur.strftime("%Y-%m-%d %H:%M"), counts.get(cur, 0)))
        cur += timedelta(minutes=1)
    return result


if __name__ == "__main__":
    import os

    path = "/tmp/loganalysis_b1_s18.ndjson"
    lines = [
        {"timestamp": "2025-03-12T14:05:10Z", "service": "checkout", "level": "ERROR", "message": "timeout"},
        {"timestamp": "2025-03-12T13:00:00Z", "service": "checkout", "level": "ERROR", "message": "too old, outside 60min window"},
        {"timestamp": "2025-03-12T14:05:40Z", "service": "checkout", "level": "ERROR", "message": "timeout"},
        {"timestamp": "2025-03-12T14:59:59Z", "service": "checkout", "level": "ERROR", "message": "latest event, defines the window end"},
        {"timestamp": "2025-03-12T14:05:20Z", "service": "checkout", "level": "INFO", "message": "not an error"},
        {"timestamp": "2025-03-12T14:07:00Z", "service": "checkout", "level": "ERROR", "message": "timeout"},
        {"timestamp": "2025-03-12T14:07:05Z", "service": "auth", "level": "ERROR", "message": "different service"},
    ]
    with open(path, "w") as f:
        for rec in lines:
            f.write(json.dumps(rec) + "\n")

    for bucket, cnt in per_minute_error_counts(path, "checkout"):
        if cnt:
            print(f"{bucket} {cnt}")

    os.remove(path)

Worked example (executed output)

2025-03-12 14:05 2
2025-03-12 14:07 1
2025-03-12 14:59 1

The synthetic input above is deliberately out of order and mixes services and levels: a 13:00 entry (an hour before the latest timestamp, correctly excluded from the 60-minute window), an INFO-level entry at 14:05 (correctly excluded, not an ERROR), and an entry from a different service, auth (correctly excluded). The three surviving lines match exactly what the question's example output format specifies: YYYY-MM-DD HH:MM count.

Key points

• Memory stays close to O(WINDOW_MINUTES), not O(distinct minutes in the whole file): every time a new maximum timestamp arrives, the code immediately drops any bucket older than new_max - 60 minutes, rather than waiting until the end of the file to filter. Over a ~10 GB file that could span far more than an hour of wall-clock time, this keeps the counters dict small throughout the scan instead of growing with every minute the file happens to contain.
• The underlying function returns a dense per-minute series (one entry per minute of the 60-minute window, zero-filled where there were no errors), which is more useful than a sparse series if you intend to plot it; the driver code above only prints non-zero minutes to match the requested output format, but the zero-filled minutes are still available from the function's return value.
• Out-of-order tolerance comes from tracking the running maximum timestamp seen so far, not from assuming the file's line order matches timestamp order at all.

Complexity

Time is O(n) for n lines in the file, one pass, no sorting. Space is bounded by the number of distinct minute buckets retained at any point, which the opportunistic pruning keeps close to the 60-minute window rather than proportional to the file's total time span.

Edge cases

• No matching lines at all (wrong service, or no ERROR-level entries): the function returns an empty list rather than guessing a window relative to wall-clock "now," since there's no timestamp evidence in the file to anchor a window to.
• All events in the same single minute: the loop still correctly walks the full 60-minute range and reports that one minute's count alongside 59 zero counts in the dense series.
• A timestamp that fails to parse: skipped rather than raised, so one malformed line doesn't abort processing of a 10 GB file.

Trade-offs & pitfalls

The main design decision worth defending is anchoring "last 60 minutes" to the latest timestamp actually seen in the file, rather than to wall-clock now(). For a file being analyzed after the fact (which is the realistic scenario for "you're handed a log file and asked to find the error pattern"), the data's own latest timestamp is the meaningful reference point; anchoring to now() instead would silently produce an empty or wrong window whenever the file isn't from the current moment, which is the common case during an investigation. The trade-off of opportunistic pruning versus a simpler two-pass approach (scan once for the max timestamp, then scan again to aggregate) is one pass of I/O saved at the cost of slightly more bookkeeping per line; for a 10 GB file, halving the required I/O is worth that complexity.

Direct answer

Follow the file like tail -f (read new bytes as they arrive), detect rotation by comparing the file's inode (the filesystem's internal ID for a file's data, separate from its path) each time you hit end-of-file, and keep only the last 60 seconds of ERROR timestamps in a deque (a double-ended queue that supports cheap removal from the front) so memory never grows past what the window actually holds. Persist just the inode and byte offset to a small checkpoint file so a restart resumes instead of re-alerting or re-reading the whole file.

Structured elaboration

The program has three independent pieces, each answering one requirement:

1. Tailing. Open the file, seek to a saved offset (or end-of-file on first run), and loop: read a line, and if there's nothing new, sleep briefly and check again. This is a polling tail; on Linux you could swap the sleep for inotify (via the watchdog package) to get near-instant wakeups instead of polling, at the cost of extra dependency surface.
2. Rotation detection. Logrotate (and similar tools) don't edit the file in place, they rename the old file away and create a new file at the same path. The path looks unchanged, but the underlying inode is different. So on every "no new data" check, also stat() the path: if the inode changed, or the file shrank below your current read offset (truncation, e.g. from logrotate's copytruncate mode), close the old handle and reopen fresh from offset 0.
3. Sliding-window counting. Every time an ERROR line arrives, append its timestamp to a deque, then pop from the left while the oldest entry is older than now - 60s. The deque's length at that point is the current error count in the window, no separate counter needed, and no unbounded growth since old entries are evicted as fast as new ones arrive.

Persistence is deliberately minimal: only (inode, offset) is written, on a timer (e.g. every 5s), not on every line, to avoid a syscall per log line. On restart, if the saved inode still matches the file on disk, resume from the saved offset; otherwise (file was rotated away while the process was down) start from the current end of file rather than replaying an unknown amount of history.

Worked example

import os, time, json
from collections import deque

WINDOW = 60.0        # seconds
THRESHOLD = 5         # more than this many ERROR lines in WINDOW triggers an alert
CHECKPOINT = "/var/run/logtailer.chk"

class ErrorRateWatcher:
    # sliding-window ERROR counter + inode-based rotation detector
    def __init__(self, window=WINDOW, threshold=THRESHOLD):
        self.window, self.threshold = window, threshold
        self.timestamps = deque()
        self.current_inode = None

    def note_open(self, inode):
        self.current_inode = inode

    def check_rotation(self, path):
        try:
            return os.stat(path).st_ino != self.current_inode
        except FileNotFoundError:
            return False

    def process_line(self, ts, line):
        if "ERROR" not in line:
            return None
        self.timestamps.append(ts)
        cutoff = ts - self.window
        while self.timestamps and self.timestamps[0] < cutoff:
            self.timestamps.popleft()
        count = len(self.timestamps)
        return count if count > self.threshold else None

I unit-tested the sliding-window and rotation logic in isolation with a pinned, deterministic timeline instead of real wall-clock sleeps, so the result below is reproducible:

watcher = ErrorRateWatcher()
events = [
    (0, "INFO service up"), (5, "ERROR db timeout"), (15, "ERROR db timeout"),
    (25, "ERROR db timeout"), (35, "ERROR db timeout"), (45, "ERROR db timeout"),
    (50, "ERROR db timeout"),   # 6th ERROR inside the last 60s -> should alert
    (130, "ERROR db timeout"),  # window has fully aged out by now
]
watcher.note_open(inode=111)
for ts, line in events:
    result = watcher.process_line(ts, line)
    print(f"t={ts:>4}s", f"ALERT: {result} errors in last 60s" if result else
          f"ok (errors in window: {len(watcher.timestamps)})")

Output:

t=   0s ok (errors in window: 0)
t=   5s ok (errors in window: 1)
t=  15s ok (errors in window: 2)
t=  25s ok (errors in window: 3)
t=  35s ok (errors in window: 4)
t=  45s ok (errors in window: 5)
t=  50s ALERT: 6 errors in last 60s
t= 130s ok (errors in window: 1)

The alert fires exactly once, on the 6th ERROR inside the window, then clears once the older entries age out past t=130s. Rotation detection is equally simple to verify: open a real temp file, note its inode, rename it away and create a fresh file at the same path (exactly what logrotate does), and confirm check_rotation() flips to True:

import tempfile
tmpdir = tempfile.mkdtemp()
path = os.path.join(tmpdir, "app.log")
open(path, "w").write("line one\n")
watcher.note_open(os.stat(path).st_ino)
print("before rotation:", watcher.check_rotation(path))
os.rename(path, path + ".1")
open(path, "w").write("line after rotation\n")
print("after rotation:", watcher.check_rotation(path))

Output:

before rotation: False
after rotation: True

The full daemon wraps this core in a blocking read loop (open at saved offset, readline(), sleep-and-retry on EOF while checking rotation, checkpoint periodically) with no while True: infinite loop actually executed here, since that loop only terminates by external signal, but the logic it calls is exactly what's verified above.

Trade-offs & pitfalls

• Polling with time.sleep trades a little latency (up to your sleep interval) for zero extra dependencies; inotify/watchdog removes that latency but adds a library and edge cases around watching a path whose target keeps changing across rotations.
• Checkpointing too often adds I/O overhead per line; checkpointing too rarely means a crash right before rotation can replay a small amount of duplicate data. Time-based checkpointing (every few seconds) is the usual compromise.
• A single in-process alert threshold will alert-storm if the underlying failure persists; add a cooldown or exponential backoff on repeat alerts for the same condition rather than firing on every line past the threshold.
• Multi-line stack traces or JSON log entries that happen to contain the literal string "ERROR" in a message field (not the level) will over-count; a real implementation should parse structured fields rather than substring-matching where the log format allows it.

Direct answer

Treat this like any forensic investigation: preserve what still exists before it decays further, rebuild the timeline from every surviving source (not just the logs that were lost), and only then fix the rotation problem so it cannot happen again. The rotation loss itself becomes part of the incident, not just an inconvenience while investigating the original outage.

Structured elaboration

Evidence preservation (first)

• Immediately copy every surviving log file, rotated archive, and journal export to a separate location before anything else runs rotation, cleanup cron jobs, or disk pressure that could delete more evidence. Treat the current state as read-only.
• Snapshot configuration alongside the data: the logrotate config (logrotate is the standard Linux utility that rotates, compresses, and deletes log files on a schedule) or the journald config (journald is systemd's log-collecting daemon) / container logging driver config, any log-shipping agent config, and the retention policy in effect at the time. You will need these to explain the loss, not just to fix it.
• Preserve non-log evidence that was not affected by the rotation bug: metrics/dashboards, alert history, deploy and change-management records, and any centralized copy a shipping agent may have already forwarded before local deletion happened.

Reconstructing the timeline from what remains

• Anchor on sources the rotation bug did not touch: monitoring alert timestamps, deploy pipeline timestamps, load balancer or gateway logs (often on a separate host/retention policy), and database slow-query or error logs.
• If a log-shipping agent was running, check whether a central log store already has a copy of what was lost locally; local rotation deleting a file does not delete an already-shipped copy.
• Build the timeline in UTC from the start, and note explicitly which windows have zero surviving log coverage, so the gap itself is documented rather than silently absent from the narrative.

Correlating remaining traces and metrics

• Use metric anomalies (error-rate spikes, latency jumps, restart counts) to bound the incident window even where no log line survives inside it.
• Cross-reference any request or correlation IDs that do survive (in a load balancer log, for instance) against the services you have log coverage for, to partially reconstruct request flow through the gap.
• Where two independent sources agree on a timestamp and a plausible causal link, treat that as higher confidence than a single source alone.

Long-term fixes to logging and retention

• Separate rotation from retention: local rotation should only ever archive or compress, never delete, until a shipping agent has confirmed successful delivery to a durable off-host store.
• Ship logs off the host in near real time so a single host's local rotation bug is not a single point of failure for the evidence.
• Add an explicit alert on log-volume dropping to near zero for a service that is still running; a rotation misconfiguration usually shows up as a sudden, suspicious silence before anyone notices logs are actually gone.
• Periodically test the rotation and retention configuration itself (a scheduled check that logs from N days ago are still retrievable where policy says they should be), rather than only discovering it is broken during an incident.

Worked example

Say the rotation config was set to keep only 2 rotated files at 10 MB each (20 MB total), but the service was logging at a rate that filled 20 MB in under an hour during the incident. By the time anyone started investigating hours later, the relevant window had already rotated out. The forensic reconstruction in this case would lean on the load balancer's access log (a separate host, unaffected), the deploy record showing a config push 12 minutes before the first alert, and the database's own slow-query log, which together bound the incident to a 20-minute window and identify the deploy as the most likely trigger, even though the application's own logs for that window are gone.

Trade-offs & pitfalls

The biggest mistake here is spending the first hour trying to recover the deleted logs (which is usually not possible once rotation has removed and overwritten disk blocks) instead of immediately preserving what remains and pivoting to alternate sources. A second common mistake is fixing only the rotation size/count without addressing the systemic issue: any purely host-local retention policy is one misconfiguration away from doing this again, which is why the durable long-term fix is decoupling deletion from local rotation entirely.

Direct answer

Stream each file line by line rather than reading it into memory, filter to the matching request_id and timestamp range as you go, and merge the per-file streams with a k-way merge, using a min-heap (a small structure that always exposes the smallest of the items it holds) keyed by timestamp, so the combined output comes out globally chronological without ever holding more than one buffered record per file at once.

Approach

import json
import heapq
from datetime import datetime


def _parse_ts(raw):
    return datetime.fromisoformat(raw.replace("Z", "+00:00"))


def _file_matches(path, request_id, start_ts, end_ts):
    """Stream one file line by line. Assumes each file is internally
    time-ordered (true of an append-only log file), so this generator
    yields matching events in ascending timestamp order for THIS file.
    Never holds more than the current line in memory.
    """
    with open(path, "r") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                continue
            if rec.get("request_id") != request_id:
                continue
            raw_ts = rec.get("ts")
            if raw_ts is None:
                continue
            try:
                event_ts = _parse_ts(raw_ts)
            except ValueError:
                continue
            if start_ts <= event_ts <= end_ts:
                rec["_ts_parsed"] = event_ts
                yield rec


def extract_events(file_paths, request_id, start_ts, end_ts):
    """Yield events with the given request_id whose timestamp falls in
    [start_ts, end_ts], in chronological order, across multiple large
    NDJSON log files, without loading any file fully into memory.

    Each file is scanned by its own generator (bounded to one open line
    at a time); a min-heap does a k-way merge across those generators so
    only one buffered record per file is held at once, not all matches.
    Memory is O(number of files), not O(file size) or O(match count).
    """
    heap = []
    iterators = [iter(_file_matches(p, request_id, start_ts, end_ts)) for p in file_paths]
    for idx, it in enumerate(iterators):
        first = next(it, None)
        if first is not None:
            heapq.heappush(heap, (first["_ts_parsed"], idx, first))

    while heap:
        ts, idx, rec = heapq.heappop(heap)
        del rec["_ts_parsed"]
        yield rec
        nxt = next(iterators[idx], None)
        if nxt is not None:
            heapq.heappush(heap, (nxt["_ts_parsed"], idx, nxt))


if __name__ == "__main__":
    import os

    file_a = "/tmp/loganalysis_b1_s13_a.ndjson"
    file_b = "/tmp/loganalysis_b1_s13_b.ndjson"

    # file_a and file_b are each individually time-ordered, as a real
    # per-host or per-service log file would be, even though the SET of
    # files is not globally ordered relative to each other.
    with open(file_a, "w") as f:
        f.write(json.dumps({"ts": "2025-01-01T00:00:01Z", "request_id": "req-1", "service": "C", "msg": "too_early"}) + "\n")
        f.write(json.dumps({"ts": "2025-01-01T00:00:02Z", "request_id": "req-1", "service": "A", "msg": "start"}) + "\n")
        f.write(json.dumps({"ts": "2025-01-01T00:00:09Z", "request_id": "req-2", "service": "A", "msg": "other_request"}) + "\n")
        f.write("not valid json\n")
        f.write(json.dumps({"ts": "2025-01-01T00:00:20Z", "request_id": "req-1", "service": "A", "msg": "too_late"}) + "\n")

    with open(file_b, "w") as f:
        f.write(json.dumps({"ts": "2025-01-01T00:00:05Z", "request_id": "req-1", "service": "B", "msg": "middle"}) + "\n")

    start = _parse_ts("2025-01-01T00:00:02Z")
    end = _parse_ts("2025-01-01T00:00:10Z")

    for rec in extract_events([file_a, file_b], "req-1", start, end):
        print(rec)

    os.remove(file_a)
    os.remove(file_b)

Worked example (executed output)

{'ts': '2025-01-01T00:00:02Z', 'request_id': 'req-1', 'service': 'A', 'msg': 'start'}
{'ts': '2025-01-01T00:00:05Z', 'request_id': 'req-1', 'service': 'B', 'msg': 'middle'}

file_a and file_b are individually time-ordered (as a real append-only log file would be), but the two files are not ordered relative to each other, and file_a even has an out-of-range line, a wrong-request-id line, and a malformed line mixed in. The output above still comes out correctly filtered and in chronological order across both files.

Key points

• Memory usage is O(number of files), not O(file size) or O(number of matches): at any moment, the heap holds at most one buffered record per file, since each file is its own generator that only produces its next matching record when asked.
• The k-way merge relies on one precondition worth stating explicitly: each individual file must already be time-ordered internally. That's a safe assumption for a normal append-only application log, but it would need to be checked (or the approach would need to change) for a log source that writes out of order within a single file.
• extract_events is itself a generator (it yields rather than returning a list), so a caller can start consuming matching events immediately and stop early if it only needs the first few, rather than waiting for the entire multi-file scan to finish.

Complexity

Let N be the total number of matching events across all files and k be the number of files. Each heap push/pop is O(log k), and there are O(N) of them (one per yielded event, plus refills), so total time is O(N log k). Space is O(k) for the heap itself, plus the fixed per-line buffer each open file uses while being read, which is independent of file size.

Edge cases

• Malformed JSON lines: skipped rather than raising, so one corrupt line doesn't abort the entire scan (shown above: file_a's "not valid json" line is silently passed over).
• A file with no matches at all: its generator is simply exhausted immediately and it drops out of the heap; the merge continues correctly with the remaining files.
• Ties on identical timestamps across files: the heap comparison falls through to the file index as a tiebreaker, so results are still deterministic (files earlier in the input list sort first on an exact tie) rather than raising a comparison error on the underlying dict.
• A file that isn't actually time-ordered internally: this breaks the merge's chronological guarantee silently (each file's own stream would emit an out-of-order record without the heap detecting it), which is worth flagging explicitly rather than assuming away.

Trade-offs & pitfalls

A simpler alternative, collect every match into a list and sort once at the end, is easier to write and is fine when you expect a bounded, small number of matches for one request_id (usually true; a single request rarely touches millions of log lines). The heap-based merge shown here is the more defensible answer for a "must stream, must not load fully into memory" prompt because it doesn't rely on that assumption: it stays memory-bounded even in the pathological case where one request_id matches a huge number of events, which the simpler collect-and-sort approach would not.