Sniffs uwsgi log file continously and spits out analytics. i.e. min, max, avg time taken to load the page. P90, p95 benchmarks...

uwtop.py

#!/usr/bin/env python3
# Sniffs uwsgi log file continously and spits out analytics.
# https://talhaasghar.dev
# 10-July-2026
import re
import sys
import time
from bisect import insort
from collections import defaultdict
from urllib.parse import urlsplit

pattern = re.compile(
    r'\b(GET|POST|PUT|DELETE|PATCH|OPTIONS|HEAD)\s+(\S+).*?\bin\s+(\d+)\s+msecs'
)


def percentile(values, p):
    if not values:
        return 0

    idx = int((len(values) - 1) * p / 100)
    return values[idx]


def print_header():
    print(
        f"{'TOTAL':>6} "
        f"{'HITS':>6} "
        f"{'CUR_MS':>7} "
        f"{'METHOD':<7} "
        f"{'URL':<35} "
        f"{'AVG':>7} "
        f"{'MIN':>7} "
        f"{'MAX':>7} "
        f"{'P90':>7} "
        f"{'P95':>7}"
    )


if len(sys.argv) != 2:
    print(f"Usage: {sys.argv[0]} <uwsgi_log_file>")
    sys.exit(1)


# Statistics per endpoint
stats = defaultdict(lambda: {
    "count": 0,
    "total": 0,
    "min": float("inf"),
    "max": 0,
    "times": []
})


total_requests = 0
lines_printed = 0
HEADER_EVERY = 25


with open(sys.argv[1], "r") as f:

    # Start watching from the end (tail -f behavior)
    f.seek(0, 2)

    print_header()

    while True:

        line = f.readline()

        if not line:
            time.sleep(0.1)
            continue

        match = pattern.search(line)

        if not match:
            continue

        method, url, ms = match.groups()
        ms = int(ms)
        url = urlsplit(url).path
        # Track each endpoint independently
        endpoint = f"{method} {url}"

        s = stats[endpoint]

        s["count"] += 1
        s["total"] += ms

        if ms < s["min"]:
            s["min"] = ms

        if ms > s["max"]:
            s["max"] = ms

        insort(s["times"], ms)

        total_requests += 1

        avg = s["total"] / s["count"]
        p90 = percentile(s["times"], 90)
        p95 = percentile(s["times"], 95)

        # Print header again like dstat/vmstat
        if lines_printed and lines_printed % HEADER_EVERY == 0:
            print()
            print_header()

        print(
            f"{total_requests:6d} "
            f"{s['count']:6d} "
            f"{ms:7d} "
            f"{method:<7} "
            f"{url:<35.35} "
            f"{avg:7.1f} "
            f"{s['min']:7d} "
            f"{s['max']:7d} "
            f"{p90:7d} "
            f"{p95:7d}",
            flush=True
        )

        lines_printed += 1