Переход к Тестированию на Основе Observability

Традиционные подходы к тестированию фокусируются на валидации до продакшена - юнит-тесты, интеграционные тесты и тестирование в staging окружениях. Однако, современные распределенные системы демонстрируют эмерджентное поведение, которое проявляется только в продакшене под реальной нагрузкой, с реальными паттернами пользователей и реальными ограничениями инфраструктуры. Тестирование на основе observability смещается одновременно влево и вправо - всесторонне инструментируя системы и активно тестируя в продакшене используя данные телеметрии, распределенные traces и валидацию SLO.

Эта статья исследует стратегии тестирования на основе observability используя OpenTelemetry, валидацию распределенного tracing, тестирование в продакшене с синтетическим мониторингом, тестирование на основе SLO и интеграцию с платформами observability.

Инструментация OpenTelemetry для Тестирования

Настройка SDK OpenTelemetry

# app/telemetry.py
from opentelemetry import trace, metrics
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource

def setup_telemetry(service_name: str, service_version: str):
    """Настроить инструментацию OpenTelemetry"""

    resource = Resource.create({
        "service.name": service_name,
        "service.version": service_version,
        "deployment.environment": os.getenv("ENVIRONMENT", "production")
    })

    trace_provider = TracerProvider(resource=resource)
    otlp_trace_exporter = OTLPSpanExporter(
        endpoint="http://otel-collector:4317",
        insecure=True
    )
    trace_provider.add_span_processor(BatchSpanProcessor(otlp_trace_exporter))
    trace.set_tracer_provider(trace_provider)

    print(f"✓ OpenTelemetry настроен для {service_name}")

# Использование в приложении
setup_telemetry("payment-service", "1.2.0")
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

payment_counter = meter.create_counter(
    "payments.processed",
    description="Количество обработанных платежей"
)

@app.route('/api/payment', methods=['POST'])
def process_payment():
    with tracer.start_as_current_span("process_payment") as span:
        try:
            span.set_attribute("payment.amount", request.json.get('amount'))
            span.set_attribute("payment.method", request.json.get('method'))

            result = payment_processor.process(request.json)
            payment_counter.add(1, {"status": "success"})

            return jsonify(result), 200

        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            payment_counter.add(1, {"status": "error"})
            raise

Валидация Распределенного Tracing

# tests/test_distributed_tracing.py
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace.export import InMemorySpanExporter

class DistributedTracingTester:
    """Тестировать распределенный tracing между микросервисами"""

    def test_end_to_end_trace_propagation(self):
        """Тестировать что контекст trace распространяется между всеми сервисами"""

        with self.tracer.start_as_current_span("test_checkout") as span:
            response = requests.post('http://api-gateway/checkout', json={
                'items': [{'id': '123', 'quantity': 1}]
            })
            assert response.status_code == 200

        # Проверить что trace включает все ожидаемые сервисы
        expected_services = [
            'api-gateway',
            'checkout-service',
            'inventory-service',
            'payment-service'
        ]

        traces = self.get_traces_from_jaeger(span.get_span_context().trace_id)
        actual_services = set(s['process']['serviceName'] for s in traces['data'][0]['spans'])

        for service in expected_services:
            assert service in actual_services

    def test_span_attributes(self):
        """Проверить что spans содержат ожидаемые атрибуты"""

        response = requests.post('http://payment-service/process', json={
            'amount': 50.00,
            'method': 'paypal'
        })

        spans = self.span_exporter.get_finished_spans()
        payment_span = next(s for s in spans if s.name == 'process_payment')

        assert payment_span.attributes.get('payment.amount') == 50.00
        assert payment_span.attributes.get('payment.method') == 'paypal'

Тестирование в Продакшене с Синтетическим Мониторингом

# tests/production/synthetic_tests.py
import pytest
import requests
import time

class SyntheticMonitoringTests:
    """Синтетические тесты которые запускаются непрерывно в продакшене"""

    def test_critical_user_journey_checkout(self):
        """Тестировать критический journey checkout в продакшене"""

        start_time = time.time()

        # Шаг 1: Просмотр продуктов
        response = requests.get('https://production.example.com/api/products')
        assert response.status_code == 200

        # Шаг 2: Добавить в корзину
        product_id = response.json()['products'][0]['id']
        response = requests.post('https://production.example.com/api/cart/add', json={
            'product_id': product_id,
            'quantity': 1
        })
        assert response.status_code == 200

        # Шаг 3: Checkout
        response = requests.post('https://production.example.com/api/checkout', json={
            'cart_id': response.json()['cart_id']
        })
        assert response.status_code == 200

        duration = time.time() - start_time

        # Валидировать SLA
        assert duration < 3.0, f"Checkout занял {duration}s, превышает SLA 3s"

    def test_api_latency_percentiles(self):
        """Тестировать что latency API соответствует SLO"""

        latencies = []

        for _ in range(100):
            start = time.time()
            response = requests.get('https://production.example.com/api/products')
            latencies.append((time.time() - start) * 1000)
            assert response.status_code == 200

        latencies.sort()
        p50, p95, p99 = latencies[49], latencies[94], latencies[98]

        print(f"P50: {p50:.2f}ms, P95: {p95:.2f}ms, P99: {p99:.2f}ms")

        assert p50 < 100, f"P50 {p50}ms превышает SLO 100ms"
        assert p95 < 500, f"P95 {p95}ms превышает SLO 500ms"

Тестирование Service Level Objectives (SLO)

# tests/test_slo.py
from prometheus_api_client import PrometheusConnect

class SLOValidator:
    """Валидировать Service Level Objectives используя метрики Prometheus"""

    def __init__(self, prometheus_url: str):
        self.prom = PrometheusConnect(url=prometheus_url)

    def test_availability_slo(self):
        """Тестировать SLO доступности 99.9%"""

        query = '''
        sum(rate(http_requests_total{status=~"2.."}[30d]))
        /
        sum(rate(http_requests_total[30d]))
        '''

        result = self.prom.custom_query(query)
        availability = float(result[0]['value'][1])

        print(f"Доступность за 30 дней: {availability * 100:.3f}%")
        assert availability >= 0.999

    def test_latency_slo(self):
        """Тестировать SLO latency P95 (<500ms)"""

        query = '''
        histogram_quantile(0.95,
          sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
        )
        '''

        result = self.prom.custom_query(query)
        p95_latency_ms = float(result[0]['value'][1]) * 1000

        print(f"Latency P95: {p95_latency_ms:.2f}ms")
        assert p95_latency_ms < 500

    def test_error_budget_consumption(self):
        """Тестировать скорость потребления error budget"""

        query = '''
        sum(rate(http_requests_total{status=~"5.."}[1h]))
        /
        sum(rate(http_requests_total[1h]))
        '''

        result = self.prom.custom_query(query)
        error_rate_1h = float(result[0]['value'][1])

        monthly_budget = 0.001  # 0.1%
        burn_rate = (error_rate_1h * 730) / monthly_budget

        print(f"Error budget burn rate: {burn_rate:.2f}x")
        assert burn_rate < 10

Тестирование Canary Развертывания с Метриками

# tests/test_canary.py
class CanaryAnalyzer:
    """Анализировать canary развертывание используя метрики"""

    def test_canary_error_rate(self):
        """Сравнить error rates между baseline и canary"""

        baseline_query = '''
        sum(rate(http_requests_total{version="v1.0", status=~"5.."}[5m]))
        /
        sum(rate(http_requests_total{version="v1.0"}[5m]))
        '''

        canary_query = '''
        sum(rate(http_requests_total{version="v1.1", status=~"5.."}[5m]))
        /
        sum(rate(http_requests_total{version="v1.1"}[5m]))
        '''

        baseline_result = self.prom.custom_query(baseline_query)
        canary_result = self.prom.custom_query(canary_query)

        baseline_error_rate = float(baseline_result[0]['value'][1]) if baseline_result else 0
        canary_error_rate = float(canary_result[0]['value'][1]) if canary_result else 0

        print(f"Baseline error rate: {baseline_error_rate * 100:.3f}%")
        print(f"Canary error rate: {canary_error_rate * 100:.3f}%")

        assert canary_error_rate <= baseline_error_rate * 1.5

Workflow GitHub Actions

# .github/workflows/observability-testing.yml
name: Observability Testing

on:
  schedule:
    - cron: '*/15 * * * *'
  workflow_dispatch:

jobs:
  synthetic-tests:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install pytest requests prometheus-api-client

      - name: Run synthetic production tests
        run: pytest tests/production/synthetic_tests.py -v

      - name: Validate SLOs
        run: pytest tests/test_slo.py -v

      - name: Analyze canary deployment
        if: github.event_name == 'workflow_dispatch'
        run: pytest tests/test_canary.py -v

Заключение

Тестирование на основе observability представляет сдвиг парадигмы от чисто pre-production валидации к непрерывному тестированию в продакшене используя телеметрию, traces и метрики. Внедряя инструментацию OpenTelemetry, валидацию распределенного tracing, синтетический мониторинг, тестирование на основе SLO и canary анализ, команды могут валидировать поведение системы в продакшене с беспрецедентной видимостью.

Ключ - относиться к observability как к первоклассной стратегии тестирования - всесторонне инструментируя, непрерывно тестируя в продакшене, программно валидируя SLO и используя метрики для руководства решениями о развертывании.