Переход к Тестированию на Основе 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 и используя метрики для руководства решениями о развертывании.