Skip to main content
Potapov.me
RUEN
testingAdvanced30 минут

Observability в pytest: метрики и трейсы для расследований

Подключаем наблюдаемость к тестам: метрики, трейсы и логи, чтобы расследовать race conditions и flaky в CI. Минимальный стек, артефакты рядом с тестами.

#pytest#observability#opentelemetry#prometheus#monitoring

Table of Contents

Observability в pytest: метрики и трейсы для расследований

Этот материал дополняет курс "Pytest: Борьба с flaky-тестами и race conditions", но не является его частью. Observability — отдельная большая тема, которая выходит за рамки основного курса.

Зачем observability в тестах

  • Быстро понять, что происходит внутри гонки. Трейсы покажут порядок операций, метрики — частоту ошибок, логи — конкретные шаги.
  • Артефакты в CI. Выгружаем всё в консоль/файлы, чтобы не зависеть от продовых стеков.
  • Минимум зависимостей. Console exporter для OTel, prometheus-client для счётчиков, обычные print/logging для дешёвых логов.

Подготовка (минимум)

pip install opentelemetry-api opentelemetry-sdk prometheus-client

Шаг 1. Добавляем трейсы и метрики в код

Пример для сервиса, где уже поймали гонку (например, промокоды). Концентрация на одном горячем участке.

# src/promo.py
import asyncio
from opentelemetry import trace
from prometheus_client import Counter
 
tracer = trace.get_tracer(__name__)
promo_uses = Counter(
    "promo_uses_total",
    "Promo code usage attempts",
    ["promo_code", "status"]
)
 
 
class PromoService:
    def __init__(self, db):
        self.db = db
        self._lock = asyncio.Lock()
 
    async def apply_promo_code(self, user_id, promo_code):
        with tracer.start_as_current_span("apply_promo") as span:
            span.set_attribute("promo.code", promo_code)
            span.set_attribute("user.id", user_id)
 
            async with self._lock:
                promo = await self.db.get_promo(promo_code)
                span.set_attribute("promo.uses", promo["uses_count"])
                span.set_attribute("promo.max", promo["max_uses"])
 
                if promo["uses_count"] < promo["max_uses"]:
                    await self.db.increment_promo_uses(promo_code)
                    promo_uses.labels(promo_code=promo_code, status="ok").inc()
                    return True
 
                promo_uses.labels(promo_code=promo_code, status="limit").inc()
                return False

Шаг 2. Включаем экспорт в тестах (без внешних сервисов)

# tests/conftest.py
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor
 
 
@pytest.fixture(scope="session", autouse=True)
def setup_tracing():
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
    trace.set_tracer_provider(provider)

Теперь каждый запуск теста выводит трейсы в stdout — их видно и локально, и в CI-логах.

Шаг 3. Снимаем метрики после теста

# tests/test_promo_observability.py
import asyncio
import pytest
from prometheus_client import REGISTRY
from src.promo import PromoService
 
 
@pytest.mark.asyncio
async def test_promo_metrics_and_traces(db):
    service = PromoService(db)
 
    results = await asyncio.gather(
        service.apply_promo_code("u1", "SUMMER20"),
        service.apply_promo_code("u2", "SUMMER20"),
    )
    assert sum(1 for r in results if r) == 1
 
    print("\n=== PROMETHEUS METRICS ===")
    for metric in REGISTRY.collect():
        if metric.name.startswith("promo_"):
            print(metric.name)
            for sample in metric.samples:
                print(f"  {sample.name}{sample.labels} = {sample.value}")

Запуск:

pytest tests/test_promo_observability.py -s

Вы увидите трейсы (порядок операций) и метрики (сколько попыток, сколько отказов). Этого достаточно, чтобы в CI понимать, что произошло в гонке.

Шаг 4. Логи как дешёвый артефакт

Если трейсы/метрики недоступны, используйте структурированные логи:

import json
import logging
 
logger = logging.getLogger("promo")
 
def log_event(event, **kwargs):
    logger.info(json.dumps({"event": event, **kwargs}))

Пишите ключевые события (promo_check, promo_applied, promo_limit) и собирайте их из stdout как артефакт CI.

Шаг 5. Что делать дальше

  • Вынести экспорт метрик в тестовый HTTP endpoint (start_http_server) и снимать его в CI curl'ом, если нужны агрегаты.
  • Если есть внешний OTLP-коллектор (Jaeger/Tempo), заменить ConsoleSpanExporter на OTLP exporter.
  • Добавить алерты в CI: если счётчик status="limit" внезапно растёт, тест помечать flaky и отправлять в карантин.

Результат

  • Видите порядок операций (трейсы), частоту исходов (метрики) и шаги (логи) без продового мониторинга.
  • Артефакты остаются в CI-логах, их можно анализировать при падениях.
  • Наблюдаемость подключена ровно к нужному месту (критическая секция), а не ко всему проекту.

Связанные материалы