vk_hackathon/logviewer/analyze.py

#!/usr/bin/env python3
"""
CLI tool to query and analyze Docker logs from index-service and search-service.

Usage:
    python3 analyze.py                        # tail both services, live
    python3 analyze.py --service search       # filter by service
    python3 analyze.py --level ERROR          # filter by log level
    python3 analyze.py --grep "qdrant"        # grep in log message
    python3 analyze.py --last 100             # last N lines per service
    python3 analyze.py --since 10m            # since N minutes/hours ago (e.g. 10m, 2h)
    python3 analyze.py --stats               # show request count / error rate summary
"""

import argparse
import json
import re
import subprocess
import sys
from datetime import datetime, timezone

SERVICES = {
    "index": "hackaton-index-1",
    "search": "hackaton-search-1",
}


def _run(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, capture_output=True, text=True, **kwargs)


def _detect_container_name(service: str) -> str:
    """Find the actual running container name for a service."""
    candidates = [
        f"hackaton-{service}-1",
        f"hackaton_{service}_1",
        f"{service}-1",
        f"{service}_1",
    ]
    result = _run(["docker", "ps", "--format", "{{.Names}}"])
    running = result.stdout.splitlines()
    for name in candidates:
        if name in running:
            return name
    # fallback: try matching by label
    result2 = _run(
        ["docker", "ps", "--filter", f"label=service={service}-service", "--format", "{{.Names}}"]
    )
    names = result2.stdout.strip().splitlines()
    if names:
        return names[0]
    return candidates[0]


def _docker_logs(container: str, since: str | None, last: int) -> list[str]:
    cmd = ["docker", "logs", "--timestamps"]
    if since:
        cmd += ["--since", since]
    if last:
        cmd += ["--tail", str(last)]
    cmd.append(container)
    result = _run(cmd)
    lines = (result.stdout + result.stderr).splitlines()
    return lines


def _parse_line(raw: str) -> dict:
    """Try to extract structured fields from a log line."""
    # Docker prepends an RFC3339 timestamp
    ts_match = re.match(r"^(\d{4}-\d{2}-\d{2}T[\d:.+Z-]+)\s+(.*)", raw)
    ts = ""
    message = raw
    if ts_match:
        ts = ts_match.group(1)
        message = ts_match.group(2)

    level = "INFO"
    for lvl in ("CRITICAL", "ERROR", "WARNING", "WARN", "INFO", "DEBUG"):
        if lvl in message.upper():
            level = lvl if lvl != "WARN" else "WARNING"
            break

    return {"ts": ts, "level": level, "message": message, "raw": raw}


def _color(level: str) -> str:
    return {
        "ERROR": "\033[31m",
        "CRITICAL": "\033[35m",
        "WARNING": "\033[33m",
        "INFO": "\033[0m",
        "DEBUG": "\033[36m",
    }.get(level, "\033[0m")


RESET = "\033[0m"


def cmd_tail(args: argparse.Namespace) -> None:
    services = [args.service] if args.service else list(SERVICES.keys())
    for svc in services:
        container = _detect_container_name(svc)
        lines = _docker_logs(container, args.since, args.last)
        print(f"\n{'='*60}")
        print(f"  {svc.upper()} SERVICE  ({container})")
        print(f"{'='*60}")
        count = 0
        for raw in lines:
            parsed = _parse_line(raw)
            if args.level and parsed["level"] != args.level.upper():
                continue
            if args.grep and args.grep.lower() not in parsed["message"].lower():
                continue
            color = _color(parsed["level"])
            print(f"{color}{parsed['raw']}{RESET}")
            count += 1
        print(f"  → {count} lines shown")


def cmd_stats(args: argparse.Namespace) -> None:
    services = [args.service] if args.service else list(SERVICES.keys())
    for svc in services:
        container = _detect_container_name(svc)
        lines = _docker_logs(container, args.since, args.last)

        counts: dict[str, int] = {"ERROR": 0, "WARNING": 0, "INFO": 0, "DEBUG": 0, "CRITICAL": 0}
        requests = 0
        errors_5xx = 0

        for raw in lines:
            parsed = _parse_line(raw)
            lvl = parsed["level"]
            counts[lvl] = counts.get(lvl, 0) + 1
            msg = parsed["message"]
            if re.search(r'"(GET|POST|PUT|DELETE|PATCH)\s', msg):
                requests += 1
            if re.search(r'" 5\d\d ', msg):
                errors_5xx += 1

        print(f"\n{'='*50}")
        print(f"  STATS: {svc.upper()} SERVICE")
        print(f"{'='*50}")
        print(f"  Total lines : {sum(counts.values())}")
        print(f"  HTTP requests: {requests}")
        print(f"  5xx errors  : {errors_5xx}")
        print(f"  ERROR lines : {counts['ERROR']}")
        print(f"  WARNING lines: {counts['WARNING']}")
        print(f"  INFO lines  : {counts['INFO']}")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze Docker logs for index-service and search-service",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("--service", choices=list(SERVICES.keys()), help="Filter by service")
    parser.add_argument("--level", help="Filter by log level (ERROR, WARNING, INFO, DEBUG)")
    parser.add_argument("--grep", help="Substring filter on log message")
    parser.add_argument("--last", type=int, default=200, help="Last N lines per service (default 200)")
    parser.add_argument("--since", help="Show logs since duration (e.g. 10m, 2h, 30s)")
    parser.add_argument("--stats", action="store_true", help="Show stats summary instead of log lines")

    args = parser.parse_args()

    try:
        if args.stats:
            cmd_stats(args)
        else:
            cmd_tail(args)
    except KeyboardInterrupt:
        print("\nInterrupted.")
    except FileNotFoundError:
        print("ERROR: docker not found in PATH", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()