Log Parsing & Analysis

import json
from pathlib import Path
from collections import Counter

def parse_json_logs(log_path: Path) -> list[dict]:
    """Parse a file of newline-delimited JSON."""
    entries = []
    for line in log_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            entries.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return entries

def count_errors(entries: list[dict]) -> Counter:
    """Count errors by level and message."""
    error_counts = Counter()
    for entry in entries:
        level = entry.get("level", "unknown")
        msg = entry.get("message", "")
        error_counts[f"{level}: {msg}"] += 1
    return error_counts

# Usage
entries = parse_json_logs(Path("/var/log/app/production.log"))
errors = count_errors(entries)
for err, count in errors.most_common(10):
    print(f"{count:5d}  {err}")

import re

LOGFMT_PATTERN = re.compile(r'(\w+)=(?:"([^"]*)"|([^\s]*))')

def parse_logfmt(line: str) -> dict[str, str]:
    """Parse a logfmt line into a dictionary."""
    return {
        key: val or quoted
        for key, quoted, val in LOGFMT_PATTERN.findall(line)
    }

# Example logfmt line:
# level=info msg="request processed" method=GET path=/api/health duration=45ms
line = 'level=info msg="request processed" method=GET path=/api/health duration=45ms'
parsed = parse_logfmt(line)
print(parsed)
# {'level': 'info', 'msg': 'request processed', 'method': 'GET',
#  'path': '/api/health', 'duration': '45ms'}

import re
from datetime import datetime

# Common Log Format (Apache/Nginx access logs)
LOG_PATTERN = re.compile(
    r'(?P<ip>[\d.]+) - (?P<user>\S+) '
    r'\[(?P<timestamp>[^\]]+)\] '
    r'"(?P<method>\w+) (?P<path>\S+) (?P<protocol>[^"]*)" '
    r'(?P<status>\d{3}) (?P<size>\d+|-)'
)

def parse_access_log(line: str) -> dict | None:
    match = LOG_PATTERN.match(line)
    if not match:
        return None
    return match.groupdict()

def analyze_errors(log_path: str) -> list[dict]:
    """Find all 5xx errors in an access log."""
    errors = []
    with open(log_path, encoding="utf-8") as f:
        for line in f:
            parsed = parse_access_log(line)
            if parsed and parsed["status"].startswith("5"):
                errors.append(parsed)
    return errors

from collections import Counter, defaultdict
from pathlib import Path
import json

def aggregate_errors(log_path: Path) -> dict:
    """Build an error summary from JSON logs."""
    error_by_endpoint = defaultdict(int)
    error_by_type = Counter()
    hourly_errors = defaultdict(int)

    for line in log_path.read_text().splitlines():
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue

        if entry.get("level") != "error":
            continue

        endpoint = entry.get("path", "unknown")
        error_type = entry.get("error_type", "unknown")
        hour = entry.get("timestamp", "")[:13]  # "2024-01-15T14"

        error_by_endpoint[endpoint] += 1
        error_by_type[error_type] += 1
        hourly_errors[hour] += 1

    return {
        "by_endpoint": dict(error_by_endpoint.most_common(10)),
        "by_type": dict(error_by_type.most_common(10)),
        "by_hour": dict(hourly_errors),
    }

from pathlib import Path
from collections import Counter
import json

def generate_incident_report(log_path: Path, start: str, end: str) -> str:
    """Generate a Markdown report for a time window."""
    errors = []
    for line in log_path.read_text().splitlines():
        try:
            entry = json.loads(line)
        except json.JSONDecodeError:
            continue
        ts = entry.get("timestamp", "")
        if start <= ts <= end and entry.get("level") == "error":
            errors.append(entry)

    error_counts = Counter(e.get("message", "unknown") for e in errors)

    report = [f"# Incident Report ({start} to {end})", ""]
    report.append(f"**Total errors:** {len(errors)}")
    report.append("")
    report.append("## Top Errors")
    for msg, count in error_counts.most_common(5):
        report.append(f"- [{count}x] {msg}")

    return "\n".join(report)