Case Study 5: Server Log Anomaly Detector

Scenario

A server generates large application and web logs every day. Operations teams need an automated way to detect unusual activity without deploying a full monitoring platform. Your task is to analyze log files and identify suspicious patterns such as error spikes or abnormal request rates.

Suggested Libraries

• re
• collections
• datetime

Sample Files You May Create

• access.log — a sample Nginx/application-style access log
• log_anomaly_detector.py — your solution script
• anomaly_report — the output report

Requirements

• Parse log entries (e.g. combined/common log format).
• Extract timestamps, IP addresses, and status codes.
• Monitor request activity over a sliding time window.
• Detect elevated error rates (e.g. a spike in 5xx responses).
• Detect abnormal traffic from a single source (a possible bot or DDoS).
• Generate alerts for suspicious activity.
• Process logs efficiently without loading everything into memory.

Have a genuine attempt at this before revealing the solution below. Think about why a deque is a natural fit for a sliding time window, compared to keeping a growing list of every log entry seen so far.

💡 Show Answer

import re
from collections import deque, Counter
from datetime import datetime, timedelta

LOG_PATTERN = re.compile(
    r'(?P<ip>\S+) - - \[(?P<time>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d{3})'
)
TIME_FMT = "%d/%b/%Y:%H:%M:%S"

def parse_line(line):
    m = LOG_PATTERN.match(line)
    if not m:
        return None
    d = m.groupdict()
    d["time"] = datetime.strptime(d["time"].split()[0], TIME_FMT)
    d["status"] = int(d["status"])
    return d

def detect_anomalies(log_path, window_seconds=60, error_threshold=0.3, req_threshold=100):
    window = deque()
    alerts = []

    with open(log_path) as f:
        for line in f:
            entry = parse_line(line)
            if not entry:
                continue
            window.append(entry)

            # Drop entries outside the sliding window
            cutoff = entry["time"] - timedelta(seconds=window_seconds)
            while window and window[0]["time"] < cutoff:
                window.popleft()

            total = len(window)
            errors = sum(1 for e in window if e["status"] >= 500)
            ip_counts = Counter(e["ip"] for e in window)

            if total >= 20 and errors / total > error_threshold:
                alerts.append((entry["time"], "HIGH_ERROR_RATE", f"{errors}/{total} 5xx errors"))

            for ip, count in ip_counts.items():
                if count > req_threshold:
                    alerts.append((entry["time"], "POSSIBLE_BOT", f"{ip} made {count} requests in {window_seconds}s"))

    return alerts

if __name__ == "__main__":
    for ts, kind, detail in detect_anomalies("access.log"):
        print(f"[{ts}] {kind}: {detail}")

How it scales: This is O(n) with a bounded deque — memory stays constant regardless of log size, since old entries are dropped as the window slides forward. For live monitoring, replace the file read with tail -f-style streaming (polling or the watchdog library) and push alerts to a message queue (Kafka/SQS) instead of printing.