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