Прекрасный момент: система написана, часть тестов автоматизирована, отчеты сгенерированы и даже выявлено несколько дефектов. Выдыхаем и будем думать о дальнейших планах: увеличить тестовое покрытие, добавить стабильности и что-то делать с логированием. Стоп, подождите, вы про логи? Заряжаем 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 в контексте автоматизации тестирования и был слишком необъективен, так как не люблю тащить что-то огромное ради пары функций, поэтому буду рад, если поделитесь своими подходами по работе с логами в тестах.
