Log Analysis and Diagnostic Data Gathering Questions

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

jq is a command-line JSON processor: it reads JSON (here, one JSON object per line, sometimes called NDJSON) and lets you filter, project, and reshape it with a small query language, similar in spirit to how you'd use SQL SELECT/WHERE but for JSON text streams. The three asks below split naturally: two are jq one-liners, and the p95 latency computation is clearer in a few lines of Python since jq has no built-in percentile function.

Structured elaboration

#!/bin/bash
set -e
cat > /tmp/loganalysis_b1_s4_log.jsonl <<'EOF'
{"ts":"2024-03-10T14:12:05Z","level":"info","user_id":123,"action":"login","latency_ms":42}
{"ts":"2024-03-10T14:12:20Z","level":"info","user_id":124,"action":"login","latency_ms":55}
{"ts":"2024-03-10T14:13:01Z","level":"info","user_id":125,"action":"login","latency_ms":39}
{"ts":"2024-03-10T14:12:10Z","level":"info","action":"login","latency_ms":48}
{"ts":"2024-03-10T14:12:12Z","level":"info","user_id":126,"action":"search","latency_ms":100}
{"ts":"2024-03-10T14:12:40Z","level":"info","user_id":127,"action":"search","latency_ms":220}
{"ts":"2024-03-10T14:13:05Z","level":"info","user_id":128,"action":"search","latency_ms":180}
{"ts":"2024-03-10T14:13:30Z","level":"info","user_id":129,"action":"search","latency_ms":300}
{"ts":"2024-03-10T14:13:45Z","level":"info","user_id":130,"action":"search","latency_ms":150}
EOF

echo "1) login counts per minute:"
jq -r 'select(.action=="login") | .ts[0:16]' /tmp/loganalysis_b1_s4_log.jsonl | sort | uniq -c

echo ""
echo "3) events missing user_id:"
jq -c 'select(has("user_id") | not)' /tmp/loganalysis_b1_s4_log.jsonl

The p95 computation below embeds the same nine records directly (rather than reading the file the bash snippet above writes), so it can be copied and run on its own, independent of the bash block:

import json
import math

# Same nine records as the bash heredoc above, embedded directly here so this
# snippet runs on its own without needing that bash block to have run first.
SAMPLE_LINES = [
    '{"ts":"2024-03-10T14:12:05Z","level":"info","user_id":123,"action":"login","latency_ms":42}',
    '{"ts":"2024-03-10T14:12:20Z","level":"info","user_id":124,"action":"login","latency_ms":55}',
    '{"ts":"2024-03-10T14:13:01Z","level":"info","user_id":125,"action":"login","latency_ms":39}',
    '{"ts":"2024-03-10T14:12:10Z","level":"info","action":"login","latency_ms":48}',
    '{"ts":"2024-03-10T14:12:12Z","level":"info","user_id":126,"action":"search","latency_ms":100}',
    '{"ts":"2024-03-10T14:12:40Z","level":"info","user_id":127,"action":"search","latency_ms":220}',
    '{"ts":"2024-03-10T14:13:05Z","level":"info","user_id":128,"action":"search","latency_ms":180}',
    '{"ts":"2024-03-10T14:13:30Z","level":"info","user_id":129,"action":"search","latency_ms":300}',
    '{"ts":"2024-03-10T14:13:45Z","level":"info","user_id":130,"action":"search","latency_ms":150}',
]

def p95_latency(lines, action):
    latencies = []
    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            rec = json.loads(line)
        except json.JSONDecodeError:
            continue
        if rec.get("action") == action and "latency_ms" in rec:
            latencies.append(rec["latency_ms"])
    if not latencies:
        return None
    latencies.sort()
    idx = math.ceil(0.95 * len(latencies)) - 1
    idx = max(0, min(idx, len(latencies) - 1))
    return latencies[idx]


p95 = p95_latency(SAMPLE_LINES, "search")
print(f"p95 latency for search: {p95}ms")

Worked example (executed output)

1) login counts per minute:
   3 2024-03-10T14:12
   1 2024-03-10T14:13

3) events missing user_id:
{"ts":"2024-03-10T14:12:10Z","level":"info","action":"login","latency_ms":48}

p95 latency for search: 300ms

Key points

• Per-minute counts: truncating the ISO 8601 timestamp string to its first 16 characters (ts[0:16]) collapses "2024-03-10T14:12:05Z" to "2024-03-10T14:12", a minute bucket, then sort | uniq -c counts occurrences per bucket. This only works because the timestamps are already zero-padded and lexicographically sortable, which ISO 8601 guarantees by design.
• p95 latency: p95 means "95% of requests were faster than this value." With n samples sorted ascending, the nearest-rank method used here takes the value at index ceil(0.95 * n) - 1. It is a common, simple choice for ad hoc analysis; a statistics library's percentile function may use linear interpolation between ranks instead, which gives a slightly different number on small samples, so don't be surprised if two tools disagree by a few milliseconds on the same data.
• Missing user_id: select(has("user_id") | not) filters to objects where the key is absent entirely. That's different from select(.user_id == null), which would also match a key that exists but was explicitly set to null. Decide which case you actually mean; production logs sometimes have both.

Trade-offs & pitfalls

The assumption doing the most work here is that every ts is UTC and already zero-padded, so a plain string prefix and a plain string sort behave correctly. If timestamps come from multiple sources with mixed timezones (a local offset instead of Z), string-prefix bucketing silently groups events into the wrong minute; normalize to UTC with a real datetime parser first in that case, don't trust the string. Malformed JSON lines are another common trap: a bare jq filter without a try/? guard aborts the whole run on the first bad line, whereas the Python snippet here explicitly skips lines that fail json.loads so one corrupt line doesn't take down the whole aggregation.

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

Grok is the pattern-matching DSL (domain-specific language, a small purpose-built syntax rather than a general programming language) that Logstash's grok filter uses to name regex fragments and compose them, so you write field names instead of raw regex. The Logstash grok pattern below is illustrative configuration (not something this environment can execute against a real Logstash pipeline); the field boundaries it encodes are verified separately with an equivalent, executed Python regex against a real nginx combined log line.

Approach

Logstash grok pattern

%{IPORHOST:client_ip} - %{USER:ident} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{DATA:path} %{DATA:protocol}" %{NUMBER:status} %{NUMBER:bytes_sent} "%{DATA:referrer}" "%{DATA:user_agent}"

Each %{PATTERN:field_name} is a named, reusable regex fragment from grok's built-in pattern library: %{IPORHOST} matches an IP address or hostname, %{USER} matches a username token, %{HTTPDATE} matches Apache's bracketed date format, %{WORD} matches one alphabetic token, and %{NUMBER} matches a numeric string, each expanding to a pre-built regex fragment so you don't have to write it by hand. %{DATA} is grok's non-greedy "anything" pattern, used here for path, protocol, referrer, and user_agent because those fields can contain almost any character. Note that grok's own built-in COMBINEDAPACHELOG pattern names the trailing HTTP/1.1 token httpversion and only captures the version number, not the scheme; since the question asks specifically for a field named protocol, this pattern captures the whole HTTP/1.1 token (scheme and version together) under that name instead, which is the more literal reading of "protocol."

Verifying the field boundaries (executed)

import re

# Same field boundaries as the grok pattern
# %{IPORHOST:client_ip} - %{USER:ident} \[%{HTTPDATE:timestamp}\]
# "%{WORD:method} %{DATA:path} %{DATA:protocol}"
# %{NUMBER:status} %{NUMBER:bytes_sent} "%{DATA:referrer}" "%{DATA:user_agent}"
PATTERN = re.compile(
    r'^(?P<client_ip>\S+) - (?P<ident>\S+) '
    r'\[(?P<timestamp>[^\]]+)\] '
    r'"(?P<method>[A-Z]+) (?P<path>\S+) (?P<protocol>[A-Z]+/[\d.]+)" '
    r'(?P<status>\d+) (?P<bytes_sent>\d+|-) '
    r'"(?P<referrer>[^"]*)" '
    r'"(?P<user_agent>[^"]*)"$'
)

line = ('203.0.113.5 - - [10/Oct/2024:13:55:36 -0700] '
        '"GET /search?q=test%20query HTTP/1.1" 200 5231 '
        '"https://example.com/" '
        '"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"')

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: 203.0.113.5
ident: -
timestamp: 10/Oct/2024:13:55:36 -0700
method: GET
path: /search?q=test%20query
protocol: HTTP/1.1
status: 200
bytes_sent: 5231
referrer: https://example.com/
user_agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36

Key points

• Query strings in the path: %{DATA:path} is non-greedy, so it stops at the first space rather than trying to consume the rest of the line; a query string like ?q=test%20query stays part of the path field intact, exactly as shown above (/search?q=test%20query). If you need path and query as separate fields, add a dissect or mutate step after the grok match to split on the first ?.
• Percent-encoding: neither grok nor this regex decodes %20 into a space; that's expected; percent-encoding is part of the raw URL and decoding it is a separate, deliberate step (only do it if you actually need the decoded form for display or analysis, since the encoded form is what the server actually received).
• Very long user agents and mapping explosion: in Elasticsearch, a text or default-mapped keyword field indexes every unique value it sees; browser and bot user-agent strings are extremely high-cardinality (having a very large number of distinct values) and some are very long, which can blow up index size and even hit Elasticsearch's field-length limits. The practical mitigations are: map user_agent as keyword with ignore_above set (so overlong values are stored but not indexed for aggregation), truncate the field before indexing if you only need it for display, and, if you need to group by user agent at all, index a hash of the full string instead of the raw string for aggregation purposes.
• Naming protocol vs. a library's default field name: this is a small but real pitfall in its own right. A team pulling in a stock COMBINEDAPACHELOG-style pattern will get httpversion, not protocol, and a downstream query or dashboard written against "protocol" will silently return nothing until someone notices the field doesn't exist. When a spec (a question, a ticket, a schema doc) names a field explicitly, match that name exactly, or add an explicit rename/alias step, rather than assuming a library's default naming is close enough.

Trade-offs & pitfalls

The main pitfall with grok specifically is that a malformed line (one that doesn't match the pattern at all) produces no match and typically gets tagged as a parse failure rather than partially parsed; a production pipeline needs a fallback path for those lines rather than silently dropping them. The other common mistake is treating query strings as low-cardinality: a path with a highly variable query string (a session token, a timestamp) indexed as a single keyword field will itself cause the same kind of cardinality blowup as an unbounded user-agent field, so the same truncate-or-split discipline usually needs to apply to path as well, not just to user_agent.

Direct answer

Detect which of the three formats a given timestamp string is in with a small set of targeted checks (a fixed-width digit string is epoch seconds; a specific pattern is Apache's format; anything else is attempted as ISO 8601), parse it into a timezone-aware value, convert to UTC, and format as an ISO 8601 string. Anything that doesn't match any of the three raises a clear, typed error rather than silently returning a wrong time.

Approach

import re
from datetime import datetime, timezone


class TimestampParseError(ValueError):
    pass


_EPOCH_RE = re.compile(r'^\d{9,10}$')
_APACHE_RE = re.compile(r'^\d{2}/[A-Za-z]{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4}$')


def normalize_timestamp(raw):
    """Parse a timestamp in Apache clf, ISO8601, or epoch-seconds form and
    return a UTC ISO8601 string. Raises TimestampParseError on anything
    else, rather than silently returning a wrong time.
    """
    raw = raw.strip()

    if _EPOCH_RE.match(raw):
        dt = datetime.fromtimestamp(int(raw), tz=timezone.utc)
        return dt.strftime("%Y-%m-%dT%H:%M:%SZ")

    if _APACHE_RE.match(raw):
        dt = datetime.strptime(raw, "%d/%b/%Y:%H:%M:%S %z")
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    try:
        iso_raw = raw[:-1] + "+00:00" if raw.endswith("Z") else raw
        dt = datetime.fromisoformat(iso_raw)
        if dt.tzinfo is None:
            raise TimestampParseError(f"timestamp has no timezone info: {raw!r}")
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    except ValueError:
        pass

    raise TimestampParseError(f"unrecognized timestamp format: {raw!r}")


samples = [
    "10/Oct/2024:13:55:36 -0700",
    "2024-10-09T11:23:45Z",
    "1700000000",
    "not-a-timestamp",
]

for s in samples:
    try:
        print(f"{s!r:40} -> {normalize_timestamp(s)}")
    except TimestampParseError as e:
        print(f"{s!r:40} -> ERROR: {e}")

Worked example (executed output)

'10/Oct/2024:13:55:36 -0700'             -> 2024-10-10T20:55:36Z
'2024-10-09T11:23:45Z'                   -> 2024-10-09T11:23:45Z
'1700000000'                             -> 2023-11-14T22:13:20Z
'not-a-timestamp'                        -> ERROR: unrecognized timestamp format: 'not-a-timestamp'

The Apache-format conversion is the one worth checking by hand: 10/Oct/2024:13:55:36 -0700 means 13:55:36 in a timezone 7 hours behind UTC, so the UTC equivalent is 7 hours later, 2024-10-10T20:55:36Z, exactly what the code above produces (and note the date itself rolls over to the 10th, since 13:55 plus 7 hours crosses midnight).

Key points

• Format detection is done with a small regex per format rather than a generic "try everything and see what doesn't throw" approach; a bare try: parse_a() except: try: parse_b()... chain is harder to reason about because it's not obvious from reading it which format actually matched a given input.
• datetime.strptime(raw, "%d/%b/%Y:%H:%M:%S %z") handles the Apache format's fixed numeric offset (-0700) natively via the %z directive; there's no need to hand-parse the offset.
• Every branch converts to UTC with .astimezone(timezone.utc) before formatting, so the output format is always the same regardless of which input format was detected, which is the whole point of normalization.

Trade-offs & pitfalls

Timezone handling: an ISO 8601 string with no timezone at all (a naive timestamp) is deliberately rejected here rather than guessed at; silently assuming a missing offset means UTC (or local time) is a common source of subtle multi-hour errors when logs actually originated somewhere else, so the safer default is to require an explicit offset and fail loudly when one is missing. Epoch ambiguity: this implementation only accepts 9 or 10-digit epoch values, meaning seconds-since-epoch; real-world logs sometimes use milliseconds (13 digits) or microseconds (16 digits) instead, and a naive digit-count heuristic can misclassify one for another, producing a wildly wrong date (a valid-looking but nonsensical year like 1970 or 50000+) rather than an error. If your logs might emit sub-second-precision epoch values, that has to be handled as its own explicit branch, not silently folded into the seconds case.