lab10 — разбор логов¶

Самая прикладная лаба курса: берём «сырой» access-лог и превращаем его в осмысленные цифры — счётчики обращений и перцентили латентности. Предмет лабы — не структуры данных как таковые, а две техники, на которых держится повседневный скриптинг Ops: потоковая (построчная) обработка через генераторы и агрегация по полям. Плюс одна честная «алгоритмическая» вещь — перцентиль латентности, посчитанный руками.

Формат строки — упрощённый access-лог, пять полей через пробел:

<ip> <method> <path> <status> <latency_ms>
10.0.0.1 GET /api/users 200 12.5

Потоковая обработка: почему генераторы¶

Главное ограничение прода: лог может быть на гигабайты, а памяти столько нет. Поэтому файл нельзя читать целиком (f.read() / f.readlines()) — нужно идти построчно, держа в памяти ровно одну строку за раз. В Python это ровно то, для чего существуют генераторы: функция с yield отдаёт элементы лениво, по запросу.

def read_lines(source: str | Iterable[str]) -> Iterator[str]:
    if isinstance(source, str):
        with open(source, encoding="utf-8") as fh:
            for line in fh:          # файловый объект сам итерируется построчно
                yield line.rstrip("\n")
    else:
        for line in source:          # список / генератор / поток строк
            yield line.rstrip("\n")

source принимает либо путь к файлу, либо любую итерируемую строк — это ключевое решение для тестируемости: в проде передаём путь, в тестах — список строк или tmp_path, без обращения к настоящему диску и без больших фикстур.

Парсинг — следующий генератор поверх первого. Конвейер «читать → парсить → отдавать» остаётся ленивым на всём протяжении:

def parse_log(source) -> Iterator[LogEntry]:
    for line in read_lines(source):
        entry = parse_line(line)
        if entry is not None:        # битые строки молча пропускаем
            yield entry

Пока результат parse_log(...) не итерируют, ни одна строка файла не прочитана. Это проверяется в тестах: источник-генератор считает, сколько строк у него запросили, и после next() счётчик равен ровно 1.

Парсинг строки: регэксп и None вместо исключения¶

Строка матчится одним именованным регэкспом. Битая строка (не подошла под формат) даёт None, а не исключение:

_LOG_RE = re.compile(
    r"^\s*(?P<ip>\S+)\s+(?P<method>\S+)\s+(?P<path>\S+)\s+"
    r"(?P<status>\d{3})\s+(?P<latency>\d+(?:\.\d+)?)\s*$"
)

def parse_line(line: str) -> LogEntry | None:
    match = _LOG_RE.match(line)
    if match is None:
        return None
    return LogEntry(
        ip=match.group("ip"),
        method=match.group("method"),
        path=match.group("path"),
        status=int(match.group("status")),
        latency=float(match.group("latency")),
    )

None выбран сознательно: в реальном логе всегда есть мусорные строки (обрезанные при ротации, посторонние сообщения), и из-за одной такой не должен падать разбор всего файла. Статус — \d{3} (ровно три цифры), латентность — целое или дробное.

Агрегация: счётчики руками¶

Счётчики строим на обычном dict без collections.Counter — чтобы был виден сам механизм (по духу как lab05, где hash map делалась без dict). Вся «магия» Counter — это две строки:

def _bump(counter: dict, key) -> None:
    if key in counter:
        counter[key] += 1
    else:
        counter[key] = 1

LogStats — аккумулятор одного прохода. Он заполняется потоково, по записи за раз, поэтому не материализует все записи в общий список:

class LogStats:
    def add(self, entry: LogEntry) -> None:
        _bump(self.by_status, entry.status)
        _bump(self.by_ip, entry.ip)
        _bump(self.by_path, entry.path)
        self.latencies.append(entry.latency)
        self.total += 1

aggregate(source) — один ленивый проход по источнику, собирающий все агрегаты сразу. top_n(counter, n) достаёт «топ эндпоинтов / топ IP»: сортировка пар по количеству убыванию, ties — по ключу для воспроизводимого порядка.

Нюанс: счётчики (by_status, by_ip, by_path) — это O(1) памяти на уникальный ключ, их немного. А вот latencies приходится копить целиком — перцентиль без всех значений не посчитать (см. ниже). Это осознанный компромисс: для точных перцентилей нужна вся выборка; в «тяжёлом» проде её заменяют на приближённые структуры (t-digest, HDR-гистограммы).

Перцентили латентности руками: метод nearest-rank¶

statistics намеренно не используем — перцентиль считаем сами. Метод — nearest-rank (ближайший ранг), он же ISO/ГОСТ-вариант:

отсортировать значения по возрастанию (копию — вход не мутируем);
посчитать ранг rank = ceil(p/100 · N), нумерация элементов с 1;
вернуть элемент под этим рангом (индекс rank-1 в 0-нумерации).

def percentile(values: list[float], p: float) -> float:
    if not values:
        raise ValueError("percentile() от пустой последовательности не определён")
    ordered = sorted(values)
    n = len(ordered)
    if p <= 0:
        return ordered[0]
    if p >= 100:
        return ordered[n - 1]
    rank = int((p * n + 99) // 100)   # ceil(p/100 * n) без math.ceil
    return ordered[rank - 1]

Почему именно nearest-rank, а не линейная интерполяция:

он не интерполирует и всегда возвращает реально наблюдавшееся значение латентности (а не «среднее между двумя соседними запросами», которого не было);
он детерминирован — удобно фиксировать в тестах точные ожидания.

Проверочные наборы (зафиксированы в тестах):

Набор	p50	p95	p99
`[1..100]`, N=100	50	95	99
`[1..10]`, N=10	5	10	10

Для [1..100]: rank = ceil(p/100 · 100) = p, то есть p-й перцентиль равен p — наглядно. Сортировка стоит O(N log N), само взятие ранга — O(1).

Меню¶

menu() в main() даёт «пощупать» лабу:

Пункт	Действие
1	Разобрать встроенный пример `SAMPLE_LOG` (в нём есть и битая строка) — агрегаты + перцентили
2	Показать парсинг одной валидной и одной битой строки (битая → `None`)
3	Разобрать произвольный лог-файл по введённому пути
0	Выход

Встроенный SAMPLE_LOG содержит одну заведомо битую строку, чтобы было видно: разбор её пропускает, а не падает.

Где это в проде¶

Ad-hoc анализ access-логов. «Сколько было 5xx за последний час и по каким эндпоинтам?» — ровно parse_log + by_status / by_path. Это grep/awk «на стероидах»: те же одноразовые разборы лога, но с реальной структурой и агрегацией, а не регэкспом в шелле.
Подсчёт ошибок по сервисам. Счётчик by_path (или по полю сервиса) сразу показывает, кто генерит ошибки и кто горячий по трафику — основа быстрого триажа во время инцидента, когда лезть в Grafana/Loki дольше, чем cat access.log | python parse.py.
Быстрая прикидка SLI без тяжёлых систем. p95/p99 латентности — это и есть latency-SLI. Когда Prometheus ещё не настроен (или недоступен прямо сейчас), эти три числа из лога дают честную оценку «насколько всё плохо» без поднятия observability-стека. Тот же приём масштабируется: на больших объёмах точную выборку латентностей заменяют на t-digest / HDR-гистограммы, но идея «потоково прочитал → агрегировал → взял перцентиль» остаётся.
Потоковость как принцип. Генераторный конвейер read_lines → parse_log → aggregate — это паттерн любой обработки больших файлов в Ops (ротированные логи, дампы, бэкапы): обрабатываем за один проход, не загружая файл в память.