Какие знаешь способы мониторинга сетевого эндпоинта?

Способы мониторинга сетевого эндпоинта

Мониторинг здоровья (health check) API эндпоинтов — критически важная часть надежной инфраструктуры. Рассмотрим основные подходы и инструменты.

1. Simple HTTP Health Check

Описание: Самый базовый способ - проверка доступности эндпоинта по HTTP.

import requests
import time
from datetime import datetime

def check_endpoint_health(url, timeout=5):
    try:
        response = requests.get(url, timeout=timeout)
        return {
            'status': 'healthy' if response.status_code == 200 else 'unhealthy',
            'status_code': response.status_code,
            'timestamp': datetime.utcnow().isoformat(),
            'response_time': response.elapsed.total_seconds()
        }
    except requests.Timeout:
        return {'status': 'timeout', 'error': 'Request timeout'}
    except requests.ConnectionError:
        return {'status': 'error', 'error': 'Connection failed'}
    except Exception as e:
        return {'status': 'error', 'error': str(e)}

# Использование
result = check_endpoint_health('https://api.example.com/health')
print(result)

Особенности:

• Простость реализации
• Минимум overhead
• Не проверяет реальную функциональность
• Хорошо для быстрых проверок

2. Liveness и Readiness Probes (Kubernetes)

Описание: Стандартный подход в контейнеризированных приложениях.

from fastapi import FastAPI, Response

app = FastAPI()

# Liveness - проверка, живо ли приложение
@app.get('/healthz')
def liveness_probe():
    return {'status': 'alive'}

# Readiness - готово ли приложение обрабатывать трафик
@app.get('/readyz')
async def readiness_probe():
    # Проверяем подключение к БД, кэшу и т.д.
    try:
        # Проверка БД
        db_check = await check_database()
        # Проверка кэша
        cache_check = await check_redis()
        
        if db_check and cache_check:
            return {'status': 'ready'}
        else:
            return Response(
                content={'status': 'not_ready'},
                status_code=503  # Service Unavailable
            )
    except Exception as e:
        return Response(
            content={'status': 'error', 'detail': str(e)},
            status_code=503
        )

Особенности:

• Liveness проверяет, запущено ли приложение
• Readiness проверяет, готово ли оно обрабатывать запросы
• Kubernetes использует эти пробы для автоматического перезапуска
• Более надежно чем простой HTTP check

3. Startup Probe (Kubernetes)

Описание: Проверка завершения инициализации приложения.

@app.get('/startup')
async def startup_probe():
    # Проверяем, завершена ли инициализация
    if not app.state.initialized:
        return Response(
            content={'status': 'initializing'},
            status_code=503
        )
    return {'status': 'initialized'}

# На старте приложения
@app.on_event('startup')
async def startup():
    app.state.initialized = True

4. Synthetic Monitoring

Описание: Имитация реальных пользовательских действий для проверки функциональности.

import asyncio
from datetime import datetime

async def synthetic_monitoring():
    checks = [
        ('GET /api/users', lambda: requests.get('https://api.example.com/api/users')),
        ('POST /api/auth', lambda: requests.post('https://api.example.com/api/auth', json={'user': 'test'})),
        ('GET /api/data', lambda: requests.get('https://api.example.com/api/data')),
    ]
    
    results = []
    for name, check_func in checks:
        try:
            start = datetime.now()
            response = check_func()
            elapsed = (datetime.now() - start).total_seconds()
            
            results.append({
                'check': name,
                'status': 'pass' if response.status_code < 400 else 'fail',
                'response_time': elapsed,
                'status_code': response.status_code,
            })
        except Exception as e:
            results.append({
                'check': name,
                'status': 'error',
                'error': str(e)
            })
    
    return results

# Запуск периодически
results = asyncio.run(synthetic_monitoring())

Особенности:

• Проверяет реальный user flow
• Более надежно чем простой health check
• Требует больше ресурсов
• Может использоваться для SLA мониторинга

5. Heartbeat / Keep-alive Pattern

Описание: Периодическое отправление сигнала "я живой".

import asyncio
import httpx
from datetime import datetime, timedelta

class HeartbeatMonitor:
    def __init__(self, url, interval=30, max_failures=3):
        self.url = url
        self.interval = interval
        self.max_failures = max_failures
        self.consecutive_failures = 0
        self.is_healthy = True
    
    async def check(self):
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(self.url, timeout=5)
                if response.status_code == 200:
                    self.consecutive_failures = 0
                    if not self.is_healthy:
                        self.is_healthy = True
                        await self.on_recovered()
                else:
                    self.consecutive_failures += 1
                    if self.consecutive_failures >= self.max_failures:
                        self.is_healthy = False
                        await self.on_failure()
        except Exception as e:
            self.consecutive_failures += 1
            if self.consecutive_failures >= self.max_failures:
                self.is_healthy = False
                await self.on_failure()
    
    async def on_failure(self):
        print(f'{datetime.now()}: Endpoint failed')
        # Отправить алерт
    
    async def on_recovered(self):
        print(f'{datetime.now()}: Endpoint recovered')
    
    async def run(self):
        while True:
            await self.check()
            await asyncio.sleep(self.interval)

# Использование
monitor = HeartbeatMonitor('https://api.example.com/health')
asyncio.run(monitor.run())

6. Prometheus + Node Exporter

Описание: Метрики-based мониторинг с Prometheus.

from prometheus_client import Counter, Histogram, start_http_server
import time

# Определение метрик
request_count = Counter('endpoint_requests_total', 'Total requests', ['endpoint', 'status'])
request_duration = Histogram('endpoint_request_duration_seconds', 'Request duration', ['endpoint'])

# Инструментирование
def monitored_request(url):
    start = time.time()
    try:
        response = requests.get(url)
        request_count.labels(endpoint=url, status=response.status_code).inc()
        return response
    finally:
        request_duration.labels(endpoint=url).observe(time.time() - start)

# Запуск Prometheus endpoint
start_http_server(8000)

Особенности:

• Полнофункциональный мониторинг
• Графики и алерты
• Интеграция с Grafana
• Scalable для больших систем

7. Uptime Robot / External Monitoring Services

Описание: Внешние сервисы для проверки доступности эндпоинтов.

# Используются сторонние сервисы (Uptime Robot, Pingdom, Datadog)
# Периодически отправляют запросы на эндпоинт
# Уведомляют об отключениях
# Не требуют инфраструктуры внутри

Особенности:

• Проверка из разных географических точек
• Не зависит от внутренней инфраструктуры
• Может проверить недоступное приложение
• Платные услуги

8. TCP Port Monitoring

Описание: Проверка доступности портов без HTTP.

import socket
import asyncio

async def check_port(host, port, timeout=5):
    try:
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port),
            timeout=timeout
        )
        writer.close()
        await writer.wait_closed()
        return {'status': 'open', 'host': host, 'port': port}
    except asyncio.TimeoutError:
        return {'status': 'timeout', 'host': host, 'port': port}
    except ConnectionRefusedError:
        return {'status': 'closed', 'host': host, 'port': port}
    except Exception as e:
        return {'status': 'error', 'error': str(e)}

# Использование
result = asyncio.run(check_port('api.example.com', 443))
print(result)

9. DNS Monitoring

Описание: Проверка разрешения доменных имен.

import socket

def check_dns(hostname):
    try:
        ip = socket.gethostbyname(hostname)
        return {'status': 'resolved', 'hostname': hostname, 'ip': ip}
    except socket.gaierror:
        return {'status': 'unresolved', 'hostname': hostname}
    except Exception as e:
        return {'status': 'error', 'error': str(e)}

# Использование
result = check_dns('api.example.com')
print(result)

10. OpenTelemetry / Distributed Tracing

Описание: Трассировка запросов через распределенную систему.

from opentelemetry import trace, metrics
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

jaeger_exporter = JaegerExporter(agent_host_name='localhost')
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

with tracer.start_as_current_span('check_endpoint') as span:
    response = requests.get('https://api.example.com/data')
    span.set_attribute('status_code', response.status_code)

Сравнительная таблица

Best Practices

1. Используй K8s Probes для контейнеризированных приложений
2. Комбинируй методы: Probes + Prometheus + Synthetic
3. Настрой alerting: Slack, PagerDuty, Email
4. Мониторь зависимости: БД, кэш, external APIs
5. Проверяй performance: Response time, error rate
6. Веди логи: Сохраняй результаты проверок
7. Используй exponential backoff: При повторных попытках

Вывод: Надежный мониторинг требует многоуровневого подхода с комбинацией разных инструментов и методов.