Прекрасный момент: система написана, часть тестов автоматизирована, отчеты сгенерированы и даже выявлено несколько дефектов. Выдыхаем и будем думать о дальнейших планах: увеличить тестовое покрытие, добавить стабильности и что-то делать с логированием. Стоп, подождите, вы про логи? Заряжаем ELK, агенты сбора, выделяем ресурсы. Как нет ресурсов? Так, может быть, rsyslog в конце концов? И его нельзя? Звучит как тестовая задача на собеседовании, однако можно ли в такой ситуации обойтись без дополнительных инструментов, да еще и интегрировать работу с логами в систему автоматизации тестирования?

Привет, Хабр! Меня зовут Олег, и я работаю в команде по инфраструктурной разработке в Tarantool: мы развиваем библиотеки и сервисы, которые помогают разработчикам удобно и эффективно тестировать свой код. Также я преподаю на кафедре информатики и вычислительной математики в Самарском университете, ну а в свободное время являюсь ментором на платформе GetMentor.

Сегодня хочется затронуть одну из интересных тем в контексте автоматизации тестирования — работу с логами. Расскажу о схеме взаимодействия, основанную на Python и модуле subprocess, хотя изначально реализация была сделана на Lua + С, так что используемые механизмы вполне удачно переносятся на другие языки. И да, забегая вперед — я тоже теперь в шоке, что работа с логами может быть быстрой.

Постановка проблемы

Представим, что мы проводим тестирование системы с несколькими компонентами. В процессе тестирования система автоматизации работает с одним или несколькими компонентами на одном физическом сервере. Для большинства функциональных тестов достаточно просто аккуратно запускать и останавливать процессы компонентов и взаимодействовать с ним по API. Однако для расширения покрытия тестов может потребоваться анализ логов — обнаружение ошибок, артефактов фоновых задач, отладочных сообщений и прочего. 

Компоненты выводят информацию в стандартные потоки вывода (stdout) и ошибок (stderr), что соответствует принципам The Twelve-Factor App. При этом мы можем работать с несколькими компонентами, включая систему автоматизации, которая также может вести логирование внутренних состояний. Наша цель — собрать все логи в один файл для последующего анализа, подобно тому, как сквозной идентификатор X-Request-ID используется для объединения всех контекстов по запросу. При этом мы хотим создавать тесты, основанные на анализе логов во время проведения тестирования: если в логе появилась строка “Segmentation fault”, то тест считается непройденным.

Пожелания к форматированию файла достаточно просты:

• нестрогая упорядоченность по времени (от старых к новым);

• префикс компонента (чтобы логи можно было удобно грепать).

Простой компонент

Роль условного сервера или мини-приложения будет выполнять обычный Python-скрипт (server.py), который каждую секунду пишет в stdout и stderr c помощью функции print и преднастроенного логгера:

# server.py

import time
import sys
import logging

from datetime import datetime

logger = logging.getLogger()
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler(sys.stderr)
handler.setLevel(logging.DEBUG)

formatter = logging.Formatter("%(asctime)s [%(levelname)s]> %(message)s")
handler.setFormatter(formatter)
logger.addHandler(handler)

while True:
    # YYYY-MM-DD HH:MM:SS,000
    now = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S,%f")[:-3]
    print(f"{now} [PRINTER]> Today was a good day!")
    logger.info("I can't believe today was a good day")
    time.sleep(1)

Запустив наш сервер, увидим ожидаемый вывод:

2024-05-24 11:23:12,988 [PRINTER]> Today was a good day!
2024-05-24 15:23:12,988 [INFO]> I can't believe today was a good day
2024-05-24 11:23:13,990 [PRINTER]> Today was a good day!
2024-05-24 15:23:13,990 [INFO]> I can't believe today was a good day
2024-05-24 11:23:14,991 [PRINTER]> Today was a good day!
2024-05-24 15:23:14,991 [INFO]> I can't believe today was a good day

Простой запуск процесса и чтение вывода

Возвращаемся к нашей системе тестирования. При запуске процесса, помимо указания subprocess.PIPE для stdout и stderr, не забудем флаг -u (или переменную окружения PYTHONUNBUFFERED) для отключения буферизации потоков вывода:

# reader.py

import sys
import os
import subprocess

p = subprocess.Popen(
    args=[sys.executable, '-u', 'server.py'],
    stderr = subprocess.PIPE,
    stdout = subprocess.PIPE
)

P.S. Для разных языков и систем буферизация отключается по-разному, стоит помнить об этом.

Теперь читаем не более buffer_size байт из файлового дескриптора стандартного вывода. Переводим в формат utf-8 и разбиваем по символу LF (‘\n’). Выводим на печать, добавляя какой-нибудь префикс:

# reader.py
...
buffer_size = 1024
try:
    while True:
        stdout_chunks = os.read(p.stdout.fileno(), buffer_size)
        formatted_chunks = stdout_chunks.decode('utf-8').split('\n')
        for line in formatted_chunks:
            if line:
                print(f"server | {line}")
        time.sleep(2)
except KeyboardInterrupt:
    p.terminate()

# output:
# server | 2024-05-24 12:36:45,336 [PRINTER]> Today was a good day!
# server | 2024-05-24 12:36:46,337 [PRINTER]> Today was a good day!
# server | 2024-05-24 12:36:47,338 [PRINTER]> Today was a good day!
# server | 2024-05-24 12:36:48,343 [PRINTER]> Today was a good day!
# server | 2024-05-24 12:36:49,347 [PRINTER]> Today was a good day!
# server | 2024-05-24 12:36:50,351 [PRINTER]> Today was a good day!

Мы молодцы — теперь весь stdout с нашего сервера можем наблюдать в терминале. Единственное, что нам мешает сейчас, — основной процесс выполнения заблокирован. Здесь на помощь приходит модуль threading и класс Thread.

class Reader(Thread):
    def __init__(self, fd: int, name: str):
        super().__init__(self)
        self.fd = fd
        self.name = name
        self.daemon = True
        self.running = True
        self._buffer_size = 1024
        self._output_delay = 1

    def run(self):
        while self.running:
            try:
                stdout_chunks = os.read(self.fd, self._buffer_size)
                formatted_chunks = stdout_chunks.decode('utf-8').split('\n')
                for line in formatted_chunks:
                    if line:
                        print(f"{self.name} | {line}")
            except Exception as e:
                print(f'Reader[{self.name}] failed: ' + str(e))

    def join(self):
        self.running = False
        return super().join()

Добавляем несколько специфичных нюансов, таких как остановка бесконечного цикла через флаг self.running, режим демона и название треда.

Теперь при запуске двух потоков (для stdout и stderr) мы явно пропишем сообщение из основного процесса, чтобы увидеть отсутствие блокировки. Для корректной остановки мы добавим обработку исключения KeyboardInterrupt — при его появлении необходимо аккуратно завершить работу процесса и двух тредов.

try:
    stdout_reader = Reader(p.stdout.fileno(), 'server_stdout')
    stderr_reader = Reader(p.stderr.fileno(), 'server_stderr')

    stdout_reader.start()
    stderr_reader.start()
    while True:
        print('main | not blocked')
        time.sleep(2)
except KeyboardInterrupt:
    stdout_reader.join()
    stderr_reader.join()
    p.terminate()

# output:
# main | not blocked
# server_stdout | 2024-05-24 13:10:53,851 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:10:53,851 [INFO]> I can't believe today was a good day
# server_stdout | 2024-05-24 13:10:54,853 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:10:54,853 [INFO]> I can't believe today was a good day
# main | not blocked
# server_stdout | 2024-05-24 13:10:55,854 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:10:55,854 [INFO]> I can't believe today was a good day

Читаем логи основного процесса и проворачиваем хитрости с файловыми дескрипторами

Запустить процесс и просто прочитать его stdout/stderr — это довольно тривиальная задача. Однако мы все еще хотим видеть сквозной лог-файл с логами системы автоматизации. Запустить Reader для чтения sys.stdout и sys.stderr не выйдет. На помощь приходят все те же pipes.

Используем логгер из файла server.py, добавив маленькую хитрость:

r, w = os.pipe()
os.dup2(w, sys.stderr.fileno())

Добавляем чтение stderr от основного процесса и пишем в него что-то:

# reader.py

r, w = os.pipe()
os.dup2(w, sys.stderr.fileno())

try:
    stdout_reader = Reader(p.stdout.fileno(), 'server_stdout')
    stderr_reader = Reader(p.stderr.fileno(), 'server_stderr')
    stderr_reader_main = Reader(r, 'main_stderr')

    stdout_reader.start()
    stderr_reader.start()
    stderr_reader_main.start()

    while True:
        print('main | not blocked')
        logger.info('yep, not blocked')
        time.sleep(2)

# output:
# main | not blocked?
# main_stderr | 2024-05-24 17:24:01,031 [INFO]> yep, not blocked
# server_stdout | 2024-05-24 13:24:01,047 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:24:01,047 [INFO]> I can't believe today was a good day
# server_stdout | 2024-05-24 13:24:02,047 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:24:02,047 [INFO]> I can't believe today was a good day
# main | not blocked?
# main_stderr | 2024-05-24 17:24:03,031 [INFO]> yep, not blocked
# server_stdout | 2024-05-24 13:24:03,049 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:24:03,049 [INFO]> I can't believe today was a good day
# server_stdout | 2024-05-24 13:24:04,051 [PRINTER]> Today was a good day!
# server_stderr | 2024-05-24 17:24:04,051 [INFO]> I can't believe today was a good day

Чтобы получить то же самое, но в файле, просто необходимо рядом с print указать файл и записывать в него:

def run(self):
    while self.running:
        try:
          ...
          if line:
              print(f"{self.name} | {line}")
              file.write(f"{self.name} | {line}\n") # <--
        except Exception as e:
          ...

Теперь в нашем распоряжении полноценный захват логов, внешних компонент и собственной системы.

Логи есть, а что с ними делать-то?

Вы можете записать все в файл как артефакт тестирования, выполнять поиск прямо по ходу теста, исследовать сложное поведение систем во время конкретного теста простым грепом и много что еще. Притом это решение портируемое, так как опирается на базовые механизмы межпроцессной коммуникации (pipes) и вытесняющей многозадачности (threads).

Например, разработчику понадобилось получить часть логов про RAFT (интересное про репликацию можете почитать в статье моего коллеги). Ранее ему необходимо было бегать по всей файловой системе: ls, cat, grep, find. Наверняка это увлекательно делать пару раз для демо, но не каждый день во время отладки тестов. Теперь же это один файл, где еще могут (а у нас они есть) логироваться статус подключений, тестовые данные, фоновые операции.

В заключение

ELK — один из основных инструментов для работы с логами, однако мне так и не удалось решить следующие проблемы:

• избыточная зависимость: большинство функций не используются, а поддержка и администрирование требуют много времени и ресурсов;

• скорость обработки(доставки): автотесты должны быть не только быстрыми, но и стабильными. Сетевые задержки, проблемы с промежуточными буферами приводили к нестабильным тестам.

Поэтому и решил написать своё решение.

Возможно, я просто неправильно приготовил ELK и rsyslog в контексте автоматизации тестирования и был слишком необъективен, так как не люблю тащить что-то огромное ради пары функций, поэтому буду рад, если поделитесь своими подходами по работе с логами в тестах.