uwsgi-log

#!/usr/bin/env python3
# uwsgi-log (part of ossobv/vcutil) // wdoekes/2019 // Public Domain
#
# Quick and dirty uWSGI webserver log parser and analyzer. Supports a
# tail function to dump periodic stats into a file to be read by e.g.
# Zabbix monitoring.
#
# Example usage:
#
#     # uwsgi-log /var/log/uwsgi/acme-web.log
#     (prints out various timing totals)
#
# Example tail setup:
#
#     # cat >/etc/systemd/system/uwsgi-log-tail@.service <<EOF
#     [Unit]
#     Description=Write uWSGI log stats every minute
#
#     [Service]
#     ExecStart=ExecStart=/usr/bin/uwsgi-log --tail \
#       /var/log/uwsgi/%i.log /tmp/httpreq-%i.stat
#     Restart=on-failure
#
#     [Install]
#     WantedBy=multi-user.target
#     EOF
#
#     # systemctl start uwsgi-log-tail@acme-web.service
#
# Example zabbix setup:
#
#     # Example call httpreq[acme-web,max]
#     # Available keys: count, min, max, avg, med
#     UserParameter=httpreq[*], awk '/^$2:/{print $$2}' /tmp/httpreq-$1.stat
#
# TODO:
#
# - Should we unlink the stats file when stopping? May be better for
#   monitoring. (No data being more accurate than stale data.)
# - Refactor write_summary_every_x, avoiding duplicate code (and break
#   it up into smaller functions).
# - Use argparse for the argument parsing.
#
from collections import defaultdict
from stat import S_ISREG
from time import sleep, time
import os
import re
import sys

# log-format = [pid: %(pid)|app: -|req: -/-] %(var.HTTP_X_REAL_IP) (%(user))
#   {%(vars) vars in %(pktsize) bytes} [%(ctime)] %(method) %(uri) => generated
#   %(rsize) bytes in %(msecs) msecs (%(proto) %(status)) %(headers) headers in
#   %(hsize) bytes (%(switches) switches on core %(core)) (proxy=%(addr))

EXAMPLE_LINES = ('''\
[pid: 1312|app: 0|req: 108304/1211400] 10.70.22.29 () {48 vars in 726 bytes} \
[Wed Sep 25 06:25:18 2019] \
GET /api/v1/general/learnmore \
=> generated 0 bytes in 121 msecs \
(HTTP/1.0 401) 6 headers in 314 bytes (2 switches on core 0)''',)

EXAMPLE_MATCHES = (
    {'ip': '10.70.22.29', 'method': 'GET',
     'request': '/api/v1/general/learnmore',
     'bytes': '0', 'msecs': '121', 'httpstatus': 'HTTP/1.0 401'},
)

LINE_MATCHER = re.compile('''\
^\\[[^]]*\\] (?P<ip>[0-9.]+) \\(-?\\) \\{[^}]*\\} \
\\[[^]]*\\] \
(?P<method>\\S+) (?P<request>\\S+) \
=> generated (?P<bytes>\\d+) bytes in (?P<msecs>\\d+) msecs \
\\((?P<httpstatus>HTTP[^ ]+ \\d+)\\)''')

for line, match in zip(EXAMPLE_LINES, EXAMPLE_MATCHES):
    found_match = LINE_MATCHER.match(line)
    assert found_match, line
    dict_match = found_match.groupdict()
    assert dict_match == match, (line, dict_match)


class Record:
    def __init__(self, d):
        self.ip = d['ip']
        self.method = d['method']
        self.request = d['request']
        self.query_string = ''
        if '?' in self.request:
            self.request, self.query_string = self.request.split('?', 1)
        self.bytes = int(d['bytes'])
        self.msecs = int(d['msecs'])
        self.http_ver, self.http_code = d['httpstatus'].split(' ')
        self.http_code = int(self.http_code)

    def __repr__(self):
        return (
            '<Record({o.method} {o.request} '
            'status={o.http_code} time={o.msecs})>'
            .format(o=self))


class Timing:
    def __init__(self, points):
        if not points:
            self.min = self.max = self.median = self.total = self.average = 0
            return

        points = list(sorted(points))
        self.min = points[0]
        self.max = points[-1]
        if len(points) % 2 == 0:
            self.median = (sum(
                points[(len(points) // 2):(len(points) // 2 + 2)]) + 1) // 2
        else:
            self.median = points[(len(points) // 2)]
        self.total = sum(points)
        self.average = (self.total + 1) // len(points)


class Records:
    def __init__(self):
        self._list = list()

    @property
    def msec(self):
        if not hasattr(self, '_msec'):
            self._msec = Timing([i.msecs for i in self._list])
        return self._msec

    def add(self, record):
        self._list.append(record)

    def __eq__(self, other):
        return id(self) == id(other)

    def __lt__(self, other):
        if self == other:
            return False
        return len(self) < len(other)

    def __len__(self):
        return len(self._list)

    def __str__(self):
        return (
             '{len}x {o.msec.median}med {o.msec.average}avg '
             '{o.msec.min}min {o.msec.max}max'
             .format(len=len(self), o=self))


class BasicSummary:
    def __init__(self):
        self.all = Records()

    def add(self, record):
        self.all.add(record)


class AdvancedSummary(BasicSummary):
    def __init__(self):
        super().__init__()
        self.by_method = defaultdict(Records)
        self.by_request = defaultdict(Records)

    def add(self, record):
        super().add(record)

        self.by_method['{} {}'.format(record.method, record.http_code)].add(
            record)
        self.by_request['{} {} {}'.format(
            record.method, record.http_code, record.request)].add(
                record)

    def display_all(self):
        self.display_method_and_status()
        self.display_common_requests()
        self.display_median_times()
        self.display_total_times()
        self.display_slow_requests()

    def display_method_and_status(self):
        print(self.format_summary('methods', self.by_method))

    def display_common_requests(self):
        print(self.format_summary(
            'by_request, by count (top 12)',
            self.by_request, top=12,
            sort_key=(lambda kr: (-len(kr[1]), kr[0]))))

    def display_median_times(self):
        print(self.format_summary(
            'by_request, by median (>1000 hits, >100 median)',
            self.by_request,
            sort_key=(lambda kr: (-kr[1].msec.median, kr[0])),
            filter_key=(
                lambda kr: len(kr[1]) > 1000 and kr[1].msec.median > 100)))

    def display_total_times(self):
        print(self.format_summary(
            'by_request, by total time (top 20)',
            self.by_request, top=20,
            sort_key=(lambda kr: (-kr[1].msec.total, kr[0]))))

    def display_slow_requests(self):
        print(self.format_summary(
            'by_request, very slow',
            self.by_request,
            sort_key=(lambda kr: (-kr[1].msec.max, kr[0])),
            filter_key=(lambda kr: kr[1].msec.max > 10000)))

    @staticmethod
    def format_summary(
            title, source=None, top=None, min_results=1,
            sort_key=(lambda kr: kr[0]), filter_key=(lambda kr: True)):

        filtered = list(filter(filter_key, source.items()))
        filtered.sort(key=sort_key)

        ret = ['{}:'.format(title)]
        idx = 0
        for key, records in filtered:
            if len(records) >= min_results:
                ret.append('  {key}: {records}'.format(
                    key=key, records=records))
                idx += 1
                if top and idx == top:
                    break
        ret.append('')
        return '\n'.join(ret)


def show_single_summary(filename):
    summ = AdvancedSummary()

    with open(sys.argv[1]) as fp:
        for idx, line in enumerate(fp):
            # The log includes more than just the log lines, it may also have
            # backtraces.
            if not line.startswith('[pid: '):
                continue

            # Extract record and add to aggregator.
            try:
                r = Record(LINE_MATCHER.match(line).groupdict())
            except Exception:
                print('error on {idx}: {line}'.format(
                    idx=(idx + 1), line=line), file=sys.stderr)
                raise
            summ.add(r)

    summ.display_all()


def dont_rape_non_regular_file(filename):
    """
    Quick function to prevent someone passing /dev/stdout as argument.

    That would make us destroy it, which is not a good idea.
    """
    try:
        st = os.lstat(filename)
    except FileNotFoundError:
        pass
    else:
        if not S_ISREG(st.st_mode):
            raise ValueError('unexpected non-regular file {}'.format(filename))


def write_summary_every_x(log_filename, summary_filename, time_seconds):
    timeslice = time_seconds
    minute = time() // timeslice
    fp = open(log_filename)
    statname = summary_filename
    statname_tmp = statname + '.new'
    dont_rape_non_regular_file(statname)
    dont_rape_non_regular_file(statname_tmp)

    summ = BasicSummary()
    try:
        fp.seek(0, 2)  # seek_end

        # Loop every timeslice.
        while True:

            # Loop the entire timeslice.
            while True:
                if minute != (time() // timeslice):
                    break

                for idx, line in enumerate(fp):
                    # The log includes more than just the log lines, it
                    # may also have backtraces.
                    if not line.startswith('[pid: '):
                        continue

                    # Extract record and add to aggregator.
                    try:
                        r = Record(LINE_MATCHER.match(line).groupdict())
                    except Exception:
                        print('error on {idx}: {line}'.format(
                            idx=(idx + 1), line=line), file=sys.stderr)
                        raise
                    summ.add(r)
                sleep(1)

            with open(statname_tmp, 'w') as out:
                print(
                    'time:\t{time}\nslice:\t{timeslice}\ncount:\t{count}\n'
                    'min:\t{o.min}\nmax:\t{o.max}\n'
                    'avg:\t{o.average}\nmed:\t{o.median}\n'.format(
                        time=int(minute * timeslice), timeslice=timeslice,
                        count=len(summ.all), o=summ.all.msec), file=out)
            dont_rape_non_regular_file(statname)
            os.rename(statname_tmp, statname)

            # Get fresh records and check if file needs to be reopened.
            summ = BasicSummary()
            minute = time() // timeslice
            try:
                open_stat = os.fstat(fp.fileno())
                disk_stat = os.stat(fp.name)
                if open_stat.st_ino != disk_stat.st_ino:
                    raise OSError('renamed?')
                if fp.tell() > open_stat.st_size:
                    raise OSError('truncated?')
            except OSError:
                name = fp.name
                fp.close()
                while True:
                    try:
                        fp = open(name)
                    except OSError:
                        sleep(1)
                    else:
                        fp.seek(0, 2)  # seek_end
                        break
    finally:
        fp.close()


if sys.argv[1:2] == ['--tail']:
    # Will write a summary of the request times found in the LOG_SOURCE
    # to a stats file, named SUMMARY_DEST. For example:
    #   uwsgi-log --tail /var/log/uwsgi/app/acme-web.log /tmp/acme-web.stat &
    #   tail -F /tmp/acme-web.stat
    assert len(sys.argv) == 4, (
        'Usage: {} --tail LOG_SOURCE SUMMARY_DEST'.format(sys.argv[0]))
    write_summary_every_x(sys.argv[2], sys.argv[3], 60)
elif len(sys.argv) == 2 and not sys.argv[1].startswith('-'):
    show_single_summary(sys.argv[1])
else:
    print(
        'Usage: {argv0} LOG_SOURCE\n'
        'Usage: {argv0} --tail LOG_SOURCE SUMMARY_DEST'.format(
            argv0=sys.argv[0]), file=sys.stderr)
    sys.exit(1)