squid-manager/log_parser.py

"""Squid access.log parser and analyzer.

Squid native access.log line format (default):
    time elapsed client action/code size method URL ident hierarchy content-type

Example:
    1783841223.438 2488 172.16.12.232 TCP_MISS_ABORTED/000 0 GET \
        http://169.254.169.254/metadata/instance/compute - HIER_DIRECT/169.254.169.254 -
"""
from __future__ import annotations

import re
from collections import Counter, defaultdict
from datetime import datetime, timezone
from urllib.parse import urlsplit

# --- Result code taxonomy ---------------------------------------------------
# Squid result codes split into categories for analysis.

RESULT_CATEGORIES = {
    # cache hits
    "TCP_HIT": "Hit",
    "TCP_MEM_HIT": "Hit",
    "TCP_NEGATIVE_HIT": "Hit",
    "TCP_IMS_HIT": "Hit",
    "TCP_REFRESH_HIT": "Hit",
    "TCP_REFRESH_FAIL_HIT": "Hit",
    "TCP_REF_FAIL_HIT": "Hit",
    "TCP_IMS_HIT": "Hit",
    "TCP_DENIED_REPLY": "Denied",
    # cache misses
    "TCP_MISS": "Miss",
    "TCP_MISS_ABORTED": "Aborted",
    "TCP_REFRESH_MISS": "Miss",
    "TCP_CLIENT_REFRESH_MISS": "Miss",
    "TCP_SWAPFAIL_MISS": "Miss",
    # tunnels / connect
    "TCP_TUNNEL": "Tunnel",
    "CONNECT": "Tunnel",
    # denied
    "TCP_DENIED": "Denied",
    "TCP_REDIRECT": "Redirect",
    "NONE": "Error",
    # udp
    "UDP_HIT": "Hit",
    "UDP_MISS": "Miss",
    "UDP_DENIED": "Denied",
    "UDP_INVALID": "Error",
    "UDP_MISS_NO_FETCH": "Miss",
}

RESULT_CODE_DESCRIPTION = {
    "TCP_HIT": "Cache hit - served from cache",
    "TCP_MEM_HIT": "Memory cache hit",
    "TCP_NEGATIVE_HIT": "Negative cached (e.g. 404)",
    "TCP_IMS_HIT": "IMS hit (If-Modified-Since, 304)",
    "TCP_MISS": "Cache miss - fetched from origin",
    "TCP_MISS_ABORTED": "Miss - client aborted the request",
    "TCP_TUNNEL": "CONNECT tunnel established (HTTPS proxy)",
    "TCP_DENIED": "Request denied by ACL",
    "TCP_DENIED_REPLY": "Reply denied by ACL",
    "TCP_REFRESH_HIT": "Revalidation: still fresh",
    "TCP_REFRESH_MISS": "Revalidation: stale, re-fetched",
    "TCP_CLIENT_REFRESH_MISS": "Client forced refresh (Ctrl+F5)",
    "TCP_REDIRECT": "Redirected by redirector",
    "TCP_SWAPFAIL_MISS": "Swapfile miss (cache corruption?)",
    "NONE": "Error before request was processed",
    "UDP_HIT": "ICP cache hit",
    "UDP_MISS": "ICP cache miss",
    "UDP_DENIED": "ICP request denied",
    "UDP_INVALID": "Invalid ICP request",
}

# hierarchy code -> short label
HIERARCHY_LABELS = {
    "HIER_DIRECT": "Direct (no peer)",
    "HIER_NONE": "None",
    "HIER_SIBLING_HIT": "Sibling hit",
    "HIER_PARENT_HIT": "Parent hit",
    "HIER_DEFAULT_PARENT": "Default parent",
    "HIER_SINGLE_PARENT": "Single parent",
    "HIER_FIRST_UP_PARENT": "First up parent",
    "HIER_NO_DIRECT_FAIL": "No direct fail",
    "HIER_SRC_RTT_FASTEST": "Fastest RTT",
    "HIER_NEAREST_NEIGHBOR_HIT": "Nearest neighbor hit",
    "HIER_CD_SIBLING_HIT": "CARP sibling hit",
    "HIER_CD_PARENT_HIT": "CARP parent hit",
    "HIER_CARPLIST": "CARP list",
    "HIER_MULTICAST_SIBLING": "Multicast sibling",
    "HIER_ANY_PARENT": "Any parent",
    "HIER_INVALID_CODE": "Invalid",
}

_LOG_LINE_RE = re.compile(
    r"""^
    (?P<time>\d+(?:\.\d+)?)\s+
    (?P<elapsed>\d+)\s+
    (?P<client>\S+)\s+
    (?P<result>\S+)\s+
    (?P<size>\d+)\s+
    (?P<method>\S+)\s+
    (?P<url>\S+)\s+
    (?P<ident>\S+)\s+
    (?P<hierarchy>\S+)\s+
    (?P<content_type>.*)
    $""",
    re.VERBOSE,
)


class LogEntry(dict):
    """Parsed log entry as a dict with attribute access."""

    @property
    def category(self) -> str:
        return RESULT_CATEGORIES.get(self.result_code, "Other")

    @property
    def result_description(self) -> str:
        return RESULT_CODE_DESCRIPTION.get(self.result_code, self.result_code)

    @property
    def timestamp_local(self):
        """Return a datetime converted to the server's local timezone (CST for cn)."""
        ts = self.get("timestamp")
        if ts is None:
            return None
        try:
            return ts.astimezone()  # use system local tz
        except Exception:
            return ts


def parse_line(line: str) -> LogEntry | None:
    """Parse a single access.log line. Returns None on unparseable lines."""
    if not line or not line.strip():
        return None
    m = _LOG_LINE_RE.match(line.strip())
    if not m:
        return None
    g = m.groupdict()
    # split result code into squid_code / http_code
    raw_result = g["result"]
    if "/" in raw_result:
        squid_code, _, http_code = raw_result.partition("/")
    else:
        squid_code, http_code = raw_result, ""
    # split hierarchy into hier_code / peer
    raw_hier = g["hierarchy"]
    if "/" in raw_hier:
        hier_code, _, peer = raw_hier.partition("/")
    else:
        hier_code, peer = raw_hier, ""
    # parse timestamp
    try:
        ts = float(g["time"])
        dt = datetime.fromtimestamp(ts, tz=timezone.utc)
    except (ValueError, OSError):
        dt = None
    # parse URL into domain
    url = g["url"]
    host = _extract_host(url)
    return LogEntry(
        time=g["time"],
        timestamp=dt,
        elapsed_ms=int(g["elapsed"]),
        client=g["client"],
        result=raw_result,
        result_code=squid_code,
        http_code=http_code,
        category=RESULT_CATEGORIES.get(squid_code, "Other"),
        size=int(g["size"]),
        method=g["method"],
        url=url,
        host=host,
        ident=g["ident"] if g["ident"] != "-" else "",
        hierarchy=raw_hier,
        hier_code=hier_code,
        peer=peer,
        content_type=g["content_type"].strip(),
    )


def parse_lines(text: str) -> list[LogEntry]:
    """Parse a multi-line string. Returns list of LogEntry."""
    out = []
    for line in text.splitlines():
        e = parse_line(line)
        if e:
            out.append(e)
    return out


def _extract_host(url: str) -> str:
    """Extract hostname from a Squid log URL.

    Squid logs URLs in two forms:
      - http(s)://host/path?query  (normal GET/POST)
      - host:port                  (CONNECT tunnel, no scheme)

    urlsplit() fails on the second form (treats 'host' as scheme).
    """
    if not url or url == "-":
        return ""
    # CONNECT method URLs look like 'host:port' with no scheme
    # Detect by absence of '://' and presence of ':'
    if "://" not in url:
        # could be 'host:port' or just 'host'
        if ":" in url:
            host, _, _port = url.rpartition(":")
            return host.strip()
        return url.strip()
    try:
        h = urlsplit(url).hostname
        return h or ""
    except Exception:
        return ""


def _top_n(counter: Counter, n: int = 20) -> list[tuple[str, int]]:
    """Return top-N (value, count) tuples."""
    return counter.most_common(n)


def aggregate_stats(entries: list[LogEntry]) -> dict:
    """Build aggregate statistics from a list of LogEntry."""
    if not entries:
        return _empty_stats()

    total = len(entries)
    total_bytes = sum(e["size"] for e in entries)
    total_elapsed = sum(e["elapsed_ms"] for e in entries)

    # result code distribution
    by_result = Counter(e["result_code"] for e in entries)
    # http code distribution
    by_http = Counter(e["http_code"] for e in entries if e["http_code"])
    # method distribution
    by_method = Counter(e["method"] for e in entries)
    # category distribution (hit/miss/tunnel/denied/...)
    by_category = Counter(e["category"] for e in entries)
    # top clients (count + bytes)
    client_count = Counter()
    client_bytes = defaultdict(int)
    for e in entries:
        client_count[e["client"]] += 1
        client_bytes[e["client"]] += e["size"]
    # top destinations (host)
    host_count = Counter()
    host_bytes = defaultdict(int)
    for e in entries:
        if e["host"]:
            host_count[e["host"]] += 1
            host_bytes[e["host"]] += e["size"]
    # top URLs
    url_count = Counter(e["url"] for e in entries)
    # hierarchy codes
    by_hierarchy = Counter(e["hier_code"] for e in entries)
    # content-type
    by_ctype = Counter()
    for e in entries:
        ct = e["content_type"] or "-"
        # collapse common types
        ct_main = ct.split(";")[0].strip()
        by_ctype[ct_main] += 1
    # hourly distribution (UTC hour of day)
    by_hour = defaultdict(int)
    by_hour_bytes = defaultdict(int)
    for e in entries:
        if e["timestamp"]:
            h = e["timestamp"].hour
            by_hour[h] += 1
            by_hour_bytes[h] += e["size"]

    # cache hit ratio: hits / (hits + misses), excluding tunnels/denied
    hits = sum(c for code, c in by_result.items() if RESULT_CATEGORIES.get(code) == "Hit")
    misses = sum(c for code, c in by_result.items() if RESULT_CATEGORIES.get(code) == "Miss")
    hit_ratio = hits / (hits + misses) if (hits + misses) > 0 else 0.0
    # denied rate
    denied = sum(c for code, c in by_result.items() if RESULT_CATEGORIES.get(code) == "Denied")
    denied_rate = denied / total if total > 0 else 0.0
    # aborted rate
    aborted = by_result.get("TCP_MISS_ABORTED", 0)
    aborted_rate = aborted / total if total > 0 else 0.0
    # unique clients
    unique_clients = len(client_count)
    # average response time
    avg_elapsed = total_elapsed / total if total > 0 else 0.0
    # average object size
    avg_size = total_bytes / total if total > 0 else 0.0

    # error analysis - HTTP 4xx / 5xx
    errors_4xx = sum(c for code, c in by_http.items() if code.startswith("4"))
    errors_5xx = sum(c for code, c in by_http.items() if code.startswith("5"))
    errors_4xx_rate = errors_4xx / total if total > 0 else 0.0
    errors_5xx_rate = errors_5xx / total if total > 0 else 0.0

    # time range
    timestamps = [e["timestamp"] for e in entries if e["timestamp"]]
    if timestamps:
        time_start = min(timestamps)
        time_end = max(timestamps)
    else:
        time_start = time_end = None

    return {
        "total_requests": total,
        "total_bytes": total_bytes,
        "total_elapsed_ms": total_elapsed,
        "unique_clients": unique_clients,
        "unique_hosts": len(host_count),
        "hit_ratio": hit_ratio,
        "denied_rate": denied_rate,
        "aborted_rate": aborted_rate,
        "errors_4xx_rate": errors_4xx_rate,
        "errors_5xx_rate": errors_5xx_rate,
        "errors_4xx_count": errors_4xx,
        "errors_5xx_count": errors_5xx,
        "avg_elapsed_ms": round(avg_elapsed, 2),
        "avg_size_bytes": round(avg_size, 2),
        "hits": hits,
        "misses": misses,
        "denied": denied,
        "aborted": aborted,
        "time_start": time_start,
        "time_end": time_end,
        "by_result": _top_n(by_result, 50),
        "by_http": _top_n(by_http, 50),
        "by_method": _top_n(by_method, 20),
        "by_category": _top_n(by_category, 20),
        "by_hierarchy": _top_n(by_hierarchy, 20),
        "by_ctype": _top_n(by_ctype, 20),
        "top_clients": [
            {"client": c, "count": n, "bytes": client_bytes[c]}
            for c, n in _top_n(client_count, 50)
        ],
        "top_hosts": [
            {"host": h, "count": n, "bytes": host_bytes[h]}
            for h, n in _top_n(host_count, 50)
        ],
        "top_urls": [{"url": u, "count": n} for u, n in _top_n(url_count, 50)],
        "hourly": [
            {"hour": h, "requests": by_hour.get(h, 0), "bytes": by_hour_bytes.get(h, 0)}
            for h in range(24)
        ],
        "result_code_legend": RESULT_CODE_DESCRIPTION,
        "hierarchy_legend": HIERARCHY_LABELS,
    }


def _empty_stats() -> dict:
    return {
        "total_requests": 0,
        "total_bytes": 0,
        "total_elapsed_ms": 0,
        "unique_clients": 0,
        "unique_hosts": 0,
        "hit_ratio": 0.0,
        "denied_rate": 0.0,
        "aborted_rate": 0.0,
        "errors_4xx_rate": 0.0,
        "errors_5xx_rate": 0.0,
        "errors_4xx_count": 0,
        "errors_5xx_count": 0,
        "avg_elapsed_ms": 0.0,
        "avg_size_bytes": 0.0,
        "hits": 0,
        "misses": 0,
        "denied": 0,
        "aborted": 0,
        "time_start": None,
        "time_end": None,
        "by_result": [],
        "by_http": [],
        "by_method": [],
        "by_category": [],
        "by_hierarchy": [],
        "by_ctype": [],
        "top_clients": [],
        "top_hosts": [],
        "top_urls": [],
        "hourly": [
            {"hour": h, "requests": 0, "bytes": 0} for h in range(24)
        ],
        "result_code_legend": RESULT_CODE_DESCRIPTION,
        "hierarchy_legend": HIERARCHY_LABELS,
    }


# --- Filters for log search -------------------------------------------------

def filter_entries(
    entries: list[LogEntry],
    client: str | None = None,
    method: str | None = None,
    result_code: str | None = None,
    host: str | None = None,
    url: str | None = None,
    min_size: int | None = None,
    max_size: int | None = None,
    min_elapsed: int | None = None,
    max_elapsed: int | None = None,
    limit: int = 500,
) -> list[LogEntry]:
    out = []
    for e in entries:
        if client and client not in e["client"]:
            continue
        if method and e["method"].upper() != method.upper():
            continue
        if result_code and result_code.upper() not in e["result_code"].upper():
            continue
        if host and host.lower() not in (e["host"] or "").lower():
            continue
        if url and url.lower() not in e["url"].lower():
            continue
        if min_size is not None and e["size"] < min_size:
            continue
        if max_size is not None and e["size"] > max_size:
            continue
        if min_elapsed is not None and e["elapsed_ms"] < min_elapsed:
            continue
        if max_elapsed is not None and e["elapsed_ms"] > max_elapsed:
            continue
        out.append(e)
        if len(out) >= limit:
            break
    return out


def format_bytes(n: float) -> str:
    """Human-readable byte size."""
    if n < 1024:
        return f"{n:.0f} B"
    if n < 1024 * 1024:
        return f"{n / 1024:.1f} KB"
    if n < 1024 * 1024 * 1024:
        return f"{n / (1024 * 1024):.1f} MB"
    if n < 1024 * 1024 * 1024 * 1024:
        return f"{n / (1024 * 1024 * 1024):.2f} GB"
    return f"{n / (1024 * 1024 * 1024 * 1024):.2f} TB"


def format_duration(ms: float) -> str:
    """Format millisecond duration."""
    if ms < 1000:
        return f"{ms:.0f} ms"
    s = ms / 1000.0
    if s < 60:
        return f"{s:.2f} s"
    m = int(s // 60)
    s = s % 60
    return f"{m}m {s:.1f}s"