Работал с сервисами для раздора ошибок

Работал с сервисами для раздачи ошибок (Error Tracking Services)?

Да, активно работал с сервисами для мониторинга и отслеживания ошибок. Это критически важная часть production infrastructure.

Основные сервисы, которые использовал

1. Sentry (самый популярный)

import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

# Инициализация
sentry_sdk.init(
    dsn="https://examplePublicKey@o0.ingest.sentry.io/0",
    integrations=[
        FastApiIntegration(),
        SqlalchemyIntegration(),
    ],
    traces_sample_rate=1.0,  # Трассировать все запросы
    environment="production",
)

# Ловля исключений
try:
    result = process_payment(user_id)
except Exception as e:
    sentry_sdk.capture_exception(e)
    raise

# Ручная отправка события
sentry_sdk.capture_message("Payment processed", level="info")

# С контекстом
with sentry_sdk.push_scope() as scope:
    scope.set_tag("payment_method", "credit_card")
    scope.set_context("order", {"order_id": 123})
    sentry_sdk.capture_exception(e)

Преимущества:

• Группировка ошибок (одна ошибка = группа похожих)
• Сессионная информация (user, браузер, OS)
• Стек трейс с исходными кодами
• Уведомления (Slack, email, PagerDuty)
• Performance monitoring (трассы)

Недостатки:

• Дорого при большом объёме ошибок
• Данные идут на Sentry сервера
• Задержка в получении информации

2. ELK Stack (Elasticsearch, Logstash, Kibana)

import logging
from pythonjsonlogger import jsonlogger

# JSON логирование для ELK
logger = logging.getLogger()
handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
handler.setFormatter(formatter)
logger.addHandler(handler)

# Логируем в JSON (ELK парсит автоматически)
logger.error(
    "Payment processing failed",
    extra={
        "user_id": 123,
        "amount": 99.99,
        "payment_method": "credit_card",
        "error_code": "INSUFFICIENT_FUNDS",
        "timestamp": datetime.utcnow().isoformat(),
    }
)

Архитектура:

App Logs → Logstash (парсер) → Elasticsearch (индекс) → Kibana (UI)

Преимущества:

• Свой контроль (self-hosted)
• Мощный поиск (Elasticsearch)
• Красивая визуализация (Kibana)
• Дешевле при большом объёме

Недостатки:

• Нужно поддерживать инфраструктуру
• Требует ресурсов
• Сложнее настроить

3. New Relic

import newrelic.agent

# Инициализация
newrelic.agent.initialize('newrelic.ini')

# Автоматическое отслеживание
@newrelic.agent.application_instance()
def calculate_something():
    pass

# Custom метрики
with newrelic.agent.FunctionTraceWrapper("custom_operation"):
    result = expensive_operation()

# Ловля ошибок
try:
    result = process_data()
except:
    newrelic.agent.record_exception()
    raise

Когда использовать:

• Full-stack мониторинг (application + infrastructure)
• APM (Application Performance Monitoring)
• Нужно отслеживать зависимости

4. Datadog

from ddtrace import tracer

# Трассировка функций
@tracer.wrap()
def process_order(order_id):
    # Автоматически трассируется
    db_query()
    api_call()
    return result

# Custom события
tracer.current_span().set_tag("order_id", order_id)
tracer.current_span().set_metric("processing_time", 0.5)

Плюсы:

• Очень функциональный
• Хорошая интеграция с облаком
• Красивые dashboards

5. Rollbar

import rollbar

rollbar.init('POST_SERVER_ITEM_ACCESS_TOKEN', 'production')

try:
    process_payment()
except:
    rollbar.report_exc_info()

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

• Фокус на развёртывание (deployment tracking)
• Связь ошибок с версиями
• Откаты (rollback detection)

Как я это использую в production

Типичная настройка

import logging
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
from sentry_sdk.integrations.celery import CeleryIntegration
from pythonjsonlogger import jsonlogger
from fastapi import FastAPI

# Sentry для critical ошибок
sentry_sdk.init(
    dsn=os.getenv("SENTRY_DSN"),
    integrations=[
        FastApiIntegration(),
        SqlalchemyIntegration(),
        CeleryIntegration(),
    ],
    traces_sample_rate=0.1,  # 10% тестирование
    environment=os.getenv("ENVIRONMENT"),
)

# Логирование для всех событий
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

handler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    fmt='%(asctime)s %(name)s %(levelname)s %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)

app = FastAPI()

@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
    """Глобальный обработчик ошибок"""
    
    # Логируем
    logger.error(
        "Unhandled exception",
        extra={
            "path": request.url.path,
            "method": request.method,
            "client_ip": request.client.host,
        },
        exc_info=True
    )
    
    # Отправляем в Sentry (уже интегрирован)
    # Sentry автоматически поймёт исключение
    
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )

Стратегия отслеживания ошибок

# Уровни ошибок

# CRITICAL: ошибка на которую ЛДЫ падают
# - Database connection lost
# - Payment processing failed
# - Authentication system down

try:
    db.connect()
except DatabaseError as e:
    logger.critical("Database connection failed", exc_info=True)
    sentry_sdk.capture_exception(e)
    alert_ops_team()  # Срочно

# ERROR: ошибка которая влияет на пользователя
# - User sign up failed
# - Order creation failed
# - API rate limit exceeded

try:
    create_order(user_id, items)
except ValidationError as e:
    logger.error(f"Order creation failed: {e}", extra={"user_id": user_id})
    sentry_sdk.capture_exception(e)

# WARNING: потенциальная проблема
# - Slow query detected
# - Cache miss rate high
# - Deprecated API endpoint used

if query_time > 5:  # 5 секунд
    logger.warning(
        "Slow query detected",
        extra={"query_time": query_time, "query": query}
    )

# INFO: полезная информация
# - User logged in
# - Order shipped
# - Backup completed

logger.info("User registered", extra={"user_id": new_user.id})

Правила отправки в Sentry

# НЕ отправляем в Sentry
not_send = [
    ValidationError,  # Ошибка пользователя
    HTTPException(404),  # 404 это не ошибка
    RateLimitExceeded,  # Норм для API
]

# Отправляем в Sentry
send = [
    DatabaseError,  # Инфра проблема
    TimeoutError,  # Инфра проблема
    ServerError,  # Наша ошибка
    UnexpectedStateError,  # Наша ошибка
]

@app.exception_handler(ValidationError)
async def handle_validation_error(request, exc):
    # Логируем но не отправляем в Sentry
    logger.warning(f"Validation error: {exc}")
    return JSONResponse(status_code=400, content={"error": str(exc)})

@app.exception_handler(DatabaseError)
async def handle_db_error(request, exc):
    # Логируем И отправляем в Sentry
    logger.error(f"Database error: {exc}", exc_info=True)
    sentry_sdk.capture_exception(exc)
    return JSONResponse(status_code=500, content={"error": "Server error"})

Интеграция с Slack

# Sentry отправляет уведомления в Slack
# #errors канал получает сообщения о critical ошибках

# Пример интеграции Slack webhook
import requests

def notify_slack(error_message: str, severity: str):
    """Отправить уведомление в Slack"""
    
    color = {
        "critical": "danger",
        "error": "warning",
        "warning": "#FF9500",
    }.get(severity, "#36a64f")
    
    payload = {
        "attachments": [
            {
                "color": color,
                "title": f"{severity.upper()}: {error_message}",
                "text": error_message,
                "footer": "Error Tracking System",
            }
        ]
    }
    
    requests.post(
        os.getenv("SLACK_WEBHOOK_URL"),
        json=payload
    )

Мониторинг ошибок в Celery задачах

from celery import shared_task
import sentry_sdk

@shared_task(bind=True, max_retries=3)
def process_email(self, user_id):
    """Обработка отправки email с мониторингом"""
    
    try:
        user = User.get(user_id)
        send_email(user.email)
        logger.info(f"Email sent to {user.email}")
    
    except SMTPError as e:
        logger.error(f"SMTP error: {e}")
        sentry_sdk.capture_exception(e)
        
        # Retry с экспоненциальной задержкой
        raise self.retry(exc=e, countdown=60 * (2 ** self.request.retries))
    
    except Exception as e:
        logger.critical(f"Unexpected error in task: {e}")
        sentry_sdk.capture_exception(e)
        raise

Вывод

Я активно использовал:

1. Sentry — основной для application errors
2. ELK Stack — для логирования и аналитики
3. New Relic — для полного application monitoring
4. Datadog — для integration monitoring

Бест практики:

• Логировать на правильном уровне (ERROR, WARNING, INFO)
• Не отправлять пользовательские ошибки в Sentry
• Добавлять контекст (user_id, request_id, etc)
• Интегрировать с Slack для критических ошибок
• Мониторить аномалии в паттернах ошибок
• Создавать alerts для разных пороговых значений