After encountering a suggestion that monitoring a log file for a webserver could allow for the implementation of automatically analyzing and handling of certain events (that might require altering firewall rules etc.), the following example application was written as a simple beginning point for processing the available data. It is not meant to be full featured or to have dependencies on anything outside of the standard library since those could be added by someone else who needed more than this example boilerplate, and the code should not favor any particular operating system in the assumptions it makes. The use of double underscores throughout the code, the long lines found in a few places throughout the program, and the fact that there are already programs that accomplish similar tasks already could be points of contention for some people but should be ignored for the purpose of this review.
While looking through this code, do you see any places where you think to yourself:
- Some functionality from a module in the standard library would be a perfect fit for this area to (a) increase readability, (b) increase performance, or (c) reduce the amount of code required?
- The standard library is not being used (a) properly, (b) efficiently, or (c) as intended in some part of the program and should either (a) be replaced with something else from the standard library or (b) have custom code written to improve that section of the program?
#! /usr/bin/env python3
# -*- coding: utf-8 -*-
"""Example Application for Monitoring a Server's Access Log.
This is free and unencumbered software released into the public domain.
Anyone is free to copy, modify, publish, use, compile, sell, or
distribute this software, either in source code form or as a compiled
binary, for any purpose, commercial or non-commercial, and by any
means.
In jurisdictions that recognize copyright laws, the author or authors
of this software dedicate any and all copyright interest in the
software to the public domain. We make this dedication for the benefit
of the public at large and to the detriment of our heirs and
successors. We intend this dedication to be an overt act of
relinquishment in perpetuity of all present and future rights to this
software under copyright law.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
For more information, please refer to <http://unlicense.org/>"""
import collections
import datetime
import enum
import functools
import http
import ipaddress
import math
import pathlib
import re
import sys
import threading
import time
import urllib.parse
# Public Names
__all__ = (
'SERVER_ROOT',
'LOG_FILE',
'COMBINED_LOG_FORMAT',
'UPDATE_DELAY',
'REPORT_DELAY',
'LOG_PATH',
'RE_CLEANUP_TABLE',
'RE_COMBINED_LOG_FORMAT',
'RE_CLF_PATTERN_SEARCH',
'BILLION',
'KB',
'PREFIX',
'main',
'underline',
'ExampleDatabase',
'convert_datetime',
'HTTPMethod',
'convert_status',
'LOG_ADAPTERS',
'bytes_to_str',
'UpdateEngine'
)
# Module Documentation
__version__ = 1, 0, 0
__date__ = datetime.date(2022, 6, 30)
__author__ = 'Stephen Paul Chappell'
__credits__ = 'vigy, for providing the inspiration for this program.'
# Configuration Constants
SERVER_ROOT = pathlib.Path('C:/Abyss Web Server')
LOG_FILE = pathlib.Path('log/access.log')
# Reference: https://www.oreilly.com/library/view/regular-expressions-cookbook/9781449327453/ch07s13.html
COMBINED_LOG_FORMAT = '''\
^(?P<client>\\S+)●\\S+●(?P<userid>\\S+)●\\[(?P<datetime>[^\\]]+)\\]↵
●"(?P<method>[A-Z]+)●(?P<request>[^●"]+)?●HTTP/[0-9.]+"↵
●(?P<status>[0-9]{3})●(?P<size>[0-9]+|-)●"(?P<referrer>[^"]*)"↵
●"(?P<useragent>[^"]*)"'''
UPDATE_DELAY = datetime.timedelta(seconds=1)
REPORT_DELAY = datetime.timedelta(seconds=5)
# Symbolic Constants
LOG_PATH = SERVER_ROOT / LOG_FILE
RE_CLEANUP_TABLE = str.maketrans('●', ' ', '↵\n')
RE_COMBINED_LOG_FORMAT = COMBINED_LOG_FORMAT.translate(RE_CLEANUP_TABLE)
RE_CLF_PATTERN_SEARCH = re.compile(RE_COMBINED_LOG_FORMAT).search
BILLION = 10 ** 9
KB = 1 << 10
PREFIX = '', 'K', 'M', 'G', 'T', 'P', 'E', 'Z', 'Y'
def main():
"""Periodically generate reports for overall analyzed requests."""
db = ExampleDatabase(underline(f'\nREPORT FOR {LOG_PATH}'))
engine = UpdateEngine(LOG_PATH, RE_CLF_PATTERN_SEARCH, UPDATE_DELAY, db)
engine.start()
db.report_forever(REPORT_DELAY, lambda: engine.is_alive())
def underline(display_text, underline_character='='):
"""Take a potentially multi-lined string and underline its width."""
if not isinstance(underline_character, str):
raise TypeError('underline_character must be of type str')
if len(underline_character) != 1:
raise ValueError('underline_character must be a single character')
max_length = max(map(len, display_text.splitlines()))
return f'{display_text}\n{underline_character * max_length}'
# Reference: https://goaccess.io/
class ExampleDatabase:
"""Class demonstrating an approach to processing _LogRecord instances."""
def __init__(self, report_header):
"""Initialize the instance as an empty database."""
self.__report_header = report_header
self.__last_updated = datetime.datetime.now()
self.__records = []
self.__lock = threading.Lock()
def __iter__(self):
"""Iterate over all records in the database."""
yield from self.__records
def add_record(self, record, parse_time, line_length):
"""Add a single record with metadata to the database."""
if not isinstance(record, _LogRecord):
raise TypeError('record must be of type _LogRecord')
with self.__lock:
self.__last_updated = datetime.datetime.now()
# noinspection PyArgumentList
self.__records.append(_EDRecord(*record, parse_time, line_length))
@property
def last_updated(self):
"""Timestamp of when the database was last updated."""
return self.__last_updated
@property
def total_requests(self):
"""The total requests recorded so far."""
return len(self.__records)
@property
def valid_requests(self):
"""The number of requested considered valid."""
return sum(1 for record in self if record.valid)
@property
def failed_requests(self):
"""The number of requests considered failed."""
return sum(1 for record in self if record.failed)
@property
def log_parsing_time(self):
"""The total time used for parsing the log file."""
ns = sum(record.parse_time for record in self)
return datetime.timedelta(seconds=ns / BILLION)
@property
def unique_visitors(self):
"""The count of unique client identifiers."""
return len({record.client for record in self})
@property
def requested_files(self):
"""The total number of file requests made so far."""
return sum(1 for record in self if record.file)
@property
def path_to_clients(self):
"""A simple database of who has requested each server path."""
path_to_clients = {}
for record in self:
clients = path_to_clients.setdefault(record.path, set())
clients.add(record.client)
return path_to_clients
@property
def excl_ip_hits(self):
"""The number of paths that have been requested by only one client."""
ptc_values = self.path_to_clients.values()
return sum(1 for value in ptc_values if len(value) == 1)
@property
def referrers(self):
"""The count of unique referrers across all requests."""
return len({record.referrer for record in self})
@property
def not_found(self):
"""The count of requests having a 404 error generated for them."""
return sum(1 for record in self if record.not_found)
@property
def static_files(self):
"""The count of unique paths to files."""
return len({record.path for record in self if record.file})
@property
def log_size(self):
"""The size of the log in characters processed so far."""
return sum(record.line_length for record in self)
@property
def log_size_str(self):
"""A readable log size that assumes characters are bytes."""
return bytes_to_str(self.log_size)
@property
def tx_amount(self):
"""The total bytes transferred from the server to its clients."""
return sum(record.size for record in self)
@property
def tx_amount_str(self):
"""A readable transfer amount that assumes counts are in bytes."""
return bytes_to_str(self.tx_amount)
def create_report(self):
"""Generate an example report in a threadsafe way."""
with self.__lock:
return f'''\
{self.__report_header!s}
Last Updated = {self.last_updated!s}
Total Requests = {self.total_requests:,}
Valid Requests = {self.valid_requests:,}
Failed Requests = {self.failed_requests:,}
Log Parsing Time = {self.log_parsing_time!s}
Unique Visitors = {self.unique_visitors:,}
Requested Files = {self.requested_files:,}
Excl. IP Hits = {self.excl_ip_hits:,}
Referrers = {self.referrers:,}
Not Found = {self.not_found:,}
Static Files = {self.static_files:,}
Log Size = {self.log_size_str}
Tx. Amount = {self.tx_amount_str}'''
def report_forever(self, interval, control=lambda: True, callback=print):
"""Create reports forever using the provided interval."""
last_report = self.last_updated
while control():
time.sleep(interval.total_seconds())
if last_report != self.last_updated:
last_report = self.last_updated
callback(self.create_report())
def convert_datetime(value):
"""Convert an access log timestamp into a datetime object."""
return datetime.datetime.strptime(value, '%d/%b/%Y:%H:%M:%S %z')
# Reference: https://www.w3schools.com/tags/ref_httpmethods.asp
@enum.unique
class HTTPMethod(enum.Enum):
"""Enumeration of all legal HTTP methods recognized by servers."""
GET = 'GET is used to request data from a specified resource.'
POST = 'POST is used to send data to a server to create/update a resource.'
PUT = 'PUT is used to send data to a server to create/update a resource.'
HEAD = 'HEAD is almost identical to GET, but without the response body.'
DELETE = 'The DELETE method deletes the specified resource.'
PATCH = 'The PATCH method is used to apply partial modifications to a resource.'
OPTIONS = 'The OPTIONS method describes the communication options for the target resource.'
CONNECT = 'The CONNECT method is used to start a two-way communications (a tunnel) with the requested resource.'
TRACE = 'The TRACE method method is used to perform a message loop-back test that tests the path for the target resource (useful for debugging purposes).'
def convert_status(value):
"""Convert an HTTP status code into an enumeration instance."""
return http.HTTPStatus(int(value))
LOG_ADAPTERS = collections.OrderedDict((
('client', ipaddress.ip_address),
('userid', sys.intern),
('datetime', convert_datetime),
('method', HTTPMethod.__getitem__),
('request', urllib.parse.urlparse),
('status', convert_status),
('size', int),
('referrer', sys.intern),
('useragent', sys.intern)
))
_ED_RECORD_FIELDS = tuple(LOG_ADAPTERS) + ('parse_time', 'line_length')
class _EDRecord(collections.namedtuple('_EDRecordBase', _ED_RECORD_FIELDS)):
VALID_STATUS_PREFIX = {1, 2, 3}
FAILED_STATUS_PREFIX = {4, 5}
@functools.cached_property
def status_prefix(self):
"""The first digit of a three-digit status code."""
# noinspection PyUnresolvedReferences
return self.status.value // 100
@functools.cached_property
def valid(self):
"""The indicator if the request was valid."""
return self.status_prefix in self.VALID_STATUS_PREFIX
@functools.cached_property
def failed(self):
"""The indicator if the request was a failure."""
return self.status_prefix in self.FAILED_STATUS_PREFIX
@functools.cached_property
def path(self):
"""The path of what was requested sans other request data."""
# noinspection PyUnresolvedReferences
return pathlib.PurePath(self.request.path)
@functools.cached_property
def file(self):
"""A guess whether the path represents a file or not."""
return bool(self.path.suffix)
@functools.cached_property
def not_found(self):
"""The indicator if a request caused a 404 error or not."""
# noinspection PyUnresolvedReferences
return self.status is http.HTTPStatus.NOT_FOUND
def bytes_to_str(size):
"""Convert an item's size in bytes to an easily readable representation."""
size = int(size)
if size:
selector = round(math.log(size, KB) - 0.4)
if selector:
return f'{size / KB ** selector:.2f} {PREFIX[selector]}B'
return f'{size:d} {PREFIX[selector]}B'
return '-'
class UpdateEngine(threading.Thread):
"""Core of this program that handles changes in the access log file."""
def __init__(self, log_path, search, sleep_interval, database):
"""Initialize the instance with data for detecting log changes."""
super().__init__(daemon=True)
self.__log_path = log_path
self.__search = search
self.__sleep_interval = sleep_interval
self.__database = database
self.__stream_position = 0
self.__line_number = 1
@property
def stream_position(self):
"""The place in the file where the next update will resume."""
return self.__stream_position
@property
def line_number(self):
"""The line in the file where the next update will resume."""
return self.__line_number
def run(self):
"""Detect when the log file is updated and trigger updates."""
last_modified = None
while True:
modified_time = self.__log_path.stat().st_mtime
next_modified = datetime.datetime.fromtimestamp(modified_time)
if last_modified != next_modified:
last_modified = next_modified
self.__handle_update()
time.sleep(self.__sleep_interval.total_seconds())
def __handle_update(self):
"""Create and add records to the database when updates occur."""
with self.__log_path.open() as file:
file.seek(self.stream_position)
for line in file:
start_time = time.perf_counter_ns()
if match := self.__search(line):
record = self.__create_record(match)
stop_time = time.perf_counter_ns()
parse_time = stop_time - start_time
self.__database.add_record(record, parse_time, len(line))
else:
self.handle_search_error()
self.__line_number += 1
self.__stream_position = file.tell()
def __create_record(self, match):
"""Make a log record and convert fields to their proper datatypes."""
groups = match.groupdict()
record = _LogRecord(*(
value(groups.pop(key)) for key, value in LOG_ADAPTERS.items()
))
if groups:
self.handle_match_error(groups)
return record
def handle_search_error(self):
"""Provide a default behavior when a search fails."""
raise ValueError(f'starting at position {self.stream_position}, '
f'line {self.line_number} could not be handled')
def handle_match_error(self, groups):
"""Provide a default behavior when groups are not processed."""
raise ValueError(f'groups not processed: {", ".join(groups)}')
_LogRecord = collections.namedtuple('LogRecord', LOG_ADAPTERS)
if __name__ == '__main__':
main()
