Структурированное логирование, метрики, Prometheus, Grafana, алерты
Проблема: Бот упал в 3 часа ночи. Пользователи пишут в поддержку, но вы не знаете:
- Что случилось?
- Когда началось?
- Какие пользователи затронуты?
- Это баг или атака?
Без мониторинга: Вы узнаете о проблеме, когда пользователи массово напишут в поддержку. Потеряно время, потеряно доверие.
Решение:
- Логирование — что происходило в системе
- Метрики — сколько сообщений, ошибок, задержек
- Алерты — уведомления о проблемах до того, как пользователи заметят
- Трассировка ошибок — Sentry для детальной информации об ошибках
Зачем это нужно: Мониторинг — это глаза и уши production-бота. Без него вы летите вслепую.
Проблема: print() в коде не попадает в логи, логи смешиваются с выводом библиотек, нет структуры.
Решение: Настройка logging с правильными handlers и форматом.
# bot/logging_config.py
import logging
import sys
from logging.handlers import RotatingFileHandler
from datetime import datetime
def setup_logging(level: str = 'INFO', log_file: str = 'logs/bot.log'):
"""
Настройка логирования.
💡 Зачем:
- Console handler — для разработки и docker logs
- File handler с ротацией — не даёт файлу разрастись
- Фильтрация шумных библиотек — только важное
"""
# Форматтер с временем и уровнем
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# Console handler
# 💡 Зачем: Вывод в stdout для Docker
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
# File handler с ротацией
# 💡 Зачем: Файл не превысит 10MB, старые логи архивируются
file_handler = RotatingFileHandler(
log_file,
maxBytes=10 * 1024 * 1024, # 10MB
backupCount=5 # Хранить 5 старых файлов
)
file_handler.setFormatter(formatter)
# Root logger
logging.basicConfig(
level=getattr(logging, level.upper()),
handlers=[console_handler, file_handler]
)
# Логгеры библиотек — снижаем уровень шума
# ⚠️ Важно: aiogram и SQLAlchemy очень шумные в debug
logging.getLogger('aiogram').setLevel(logging.WARNING)
logging.getLogger('aiohttp').setLevel(logging.WARNING)
logging.getLogger('sqlalchemy').setLevel(logging.WARNING)
logging.getLogger('asyncio').setLevel(logging.WARNING)
logger = logging.getLogger('bot')
logger.info('Logging initialized')
return logger⚠️ Антипаттерн: Не логируйте в debug-режиме в продакшене. Это заполнит диск и замедлит бота.
Проблема: Логи в текстовом формате сложно парсить для агрегации (ELK, Splunk).
Решение: JSON-логирование для машинной обработки.
# bot/logging_config.py
import json
import logging
from pythonjsonlogger import jsonlogger
class JsonFormatter(jsonlogger.JsonFormatter):
"""
JSON форматтер для структурированного логирования.
💡 Зачем:
- Парсинг логами (ELK, Datadog)
- Поиск по полям
- Агрегация метрик
"""
def add_fields(self, log_record, record, message_dict):
super().add_fields(log_record, record, message_dict)
log_record['level'] = record.levelname
log_record['logger'] = record.name
log_record['timestamp'] = datetime.utcnow().isoformat()
def setup_json_logging(log_file: str = 'logs/bot.json'):
"""Настройка JSON логирования."""
logger = logging.getLogger('bot')
logger.setLevel(logging.INFO)
# File handler
file_handler = RotatingFileHandler(
log_file,
maxBytes=10 * 1024 * 1024,
backupCount=5
)
file_handler.setFormatter(JsonFormatter())
logger.addHandler(file_handler)
return logger💡 Совет: Используйте JSON-логи для продакшена, текстовые для разработки.
Проблема: В логе есть ошибка, но неясно: какой пользователь, какое действие, какой запрос.
Решение: Контекстное логирование с user_id и request_id.
# bot/utils/context_logger.py
import logging
import contextvars
from functools import wraps
import uuid
# Контекстные переменные
# 💡 Зачем: Хранение контекста между вызовами в одном запросе
user_id_ctx = contextvars.ContextVar('user_id', default=None)
request_id_ctx = contextvars.ContextVar('request_id', default=None)
class ContextLogger:
"""
Логгер с контекстом (user_id, request_id).
💡 Зачем:
- request_id — сквозной идентификатор запроса
- user_id — кто вызвал действие
- Упрощает отладку и аудит
"""
def __init__(self, name: str):
self.logger = logging.getLogger(name)
def info(self, message: str, **extra):
self._log('info', message, extra)
def warning(self, message: str, **extra):
self._log('warning', message, extra)
def error(self, message: str, **extra):
self._log('error', message, extra)
def debug(self, message: str, **extra):
self._log('debug', message, extra)
def _log(self, level: str, message: str, extra: dict):
# Добавляем контекст
ctx_extra = {
'user_id': user_id_ctx.get(),
'request_id': request_id_ctx.get(),
**extra
}
# Фильтруем None
ctx_extra = {k: v for k, v in ctx_extra.items() if v is not None}
getattr(self.logger, level)(message, extra=ctx_extra)
# Middleware для логирования
class LoggingMiddleware(BaseMiddleware):
"""
Добавляет контекст в логи.
💡 Зачем:
- Автоматическое логирование всех запросов
- Контекст для отладки
"""
def __init__(self):
self.logger = ContextLogger('bot.handlers')
async def __call__(
self,
handler: Callable,
event: Update,
data: Dict[str, Any]
) -> Any:
# Устанавливаем контекст
user_id = self._get_user_id(event)
user_id_ctx.set(user_id)
request_id_ctx.set(str(uuid.uuid4())[:8]) # Короткий ID
self.logger.info(f'Handling update: {self._get_type(event)}')
try:
result = await handler(event, data)
self.logger.info('Update handled successfully')
return result
except Exception as e:
self.logger.error(f'Error handling update: {e}', exc_info=True)
raise
def _get_user_id(self, event: Update) -> int:
if event.message:
return event.message.from_user.id
elif event.callback_query:
return event.callback_query.from_user.id
return 0
def _get_type(self, event: Update) -> str:
if event.message:
return 'message'
elif event.callback_query:
return 'callback_query'
return 'unknown'⚠️ Антипаттерн: Не логируйте чувствительные данные (токены, пароли, персональные данные).
Проблема: Вы не знаете, сколько сообщений обрабатывает бот, какая задержка, сколько ошибок.
Решение: Prometheus метрики для сбора статистики.
# bot/metrics.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE
from aiohttp import web
import time
# Счётчики — только растут
MESSAGE_COUNT = Counter(
'bot_messages_total',
'Total messages processed',
['type', 'chat_type']
)
# 💡 Зачем labels: Можно фильтровать по типу сообщения и чата
CALLBACK_COUNT = Counter(
'bot_callbacks_total',
'Total callback queries',
['action']
)
ERROR_COUNT = Counter(
'bot_errors_total',
'Total errors',
['handler', 'error_type']
)
# 💡 Зачем: Понимать, какие обработчики чаще ошибаются
# Gauge — может расти и убывать
USER_COUNT = Gauge(
'bot_users_total',
'Total active users'
)
# Гистограмма — распределение времени обработки
HANDLER_DURATION = Histogram(
'bot_handler_duration_seconds',
'Handler execution time',
['handler'],
buckets=(0.01, 0.05, 0.1, 0.5, 1.0, 5.0)
# 💡 Зачем buckets: Понимать p50, p95, p99 задержки
)
class MetricsMiddleware(BaseMiddleware):
"""Middleware для сбора метрик."""
async def __call__(
self,
handler: Callable,
event: Update,
data: Dict[str, Any]
) -> Any:
start_time = time.time()
handler_name = handler.__name__
try:
result = await handler(event, data)
# Успешная обработка
if event.message:
MESSAGE_COUNT.labels(
type='message',
chat_type=event.message.chat.type
).inc()
elif event.callback_query:
CALLBACK_COUNT.labels(
action='processed'
).inc()
return result
except Exception as e:
# Ошибка
ERROR_COUNT.labels(
handler=handler_name,
error_type=type(e).__name__
).inc()
raise
finally:
# Время выполнения
duration = time.time() - start_time
HANDLER_DURATION.labels(handler=handler_name).observe(duration)
# Endpoint для Prometheus
async def metrics_handler(request: web.Request) -> web.Response:
"""Отдаёт метрики для Prometheus."""
return web.Response(
body=generate_latest(),
content_type=CONTENT_TYPE
)
def register_metrics_endpoint(app: web.Application):
app.router.add_get('/metrics', metrics_handler)💡 Совет: Выставьте endpoint
/metricsна отдельном порту, не смешивайте с webhook.
# bot/handlers/orders.py
from bot.metrics import MESSAGE_COUNT, USER_COUNT
@router.message(Command('orders'))
async def cmd_orders(msg: Message, order_repo: OrderRepository):
# Инкремент счётчика
MESSAGE_COUNT.labels(type='command', chat_type='private').inc()
orders = await order_repo.get_by_user_id(msg.from_user.id)
# Обновление gauge
USER_COUNT.set(await order_repo.count())
await msg.answer(f'У вас {len(orders)} заказов'){
"dashboard": {
"title": "Telegram Bot Metrics",
"panels": [
{
"title": "Messages per Second",
"targets": [
{
"expr": "rate(bot_messages_total[1m])",
"legendFormat": "{{chat_type}}"
}
]
},
{
"title": "Error Rate",
"targets": [
{
"expr": "rate(bot_errors_total[5m])",
"legendFormat": "{{error_type}}"
}
]
},
{
"title": "Handler Latency (p95)",
"targets": [
{
"expr": "histogram_quantile(0.95, rate(bot_handler_duration_seconds_bucket[5m]))",
"legendFormat": "{{handler}}"
}
]
},
{
"title": "Active Users",
"targets": [
{
"expr": "bot_users_total"
}
]
}
]
}
}💡 Совет: Создайте дашборды для:
- Общей нагрузки (сообщения/сек)
- Ошибок (error rate по типам)
- Производительности (p95 latency)
- Бизнес-метрик (заказы, пользователи)
Проблема: В логах есть ошибка, но нет контекста: какой пользователь, какие данные, что предшествовало.
Решение: Sentry отправляет ошибку с полным контекстом и стек-трейсом.
pip install sentry-sdk# bot/sentry_config.py
import sentry_sdk
from sentry_sdk.integrations.aiohttp import AioHttpIntegration
from sentry_sdk.integrations.logging import LoggingIntegration
from bot.config import settings
def setup_sentry():
"""
Настройка Sentry.
💡 Зачем:
- Автоматический сбор контекста
- Уведомления об ошибках
- Группировка похожих ошибок
"""
sentry_logging = LoggingIntegration(
level=logging.INFO,
event_level=logging.ERROR # Отправлять ошибки как events
)
sentry_sdk.init(
dsn=settings.sentry_dsn,
integrations=[
AioHttpIntegration(),
sentry_logging
],
traces_sample_rate=0.1, # 10% трассировок для производительности
environment=settings.environment, # 'production', 'staging'
release=settings.version
)⚠️ Важно: Не включайте traces_sample_rate=1.0 в продакшене — это создаст нагрузку.
import sentry_sdk
from aiogram import Router, F
from aiogram.types import Message
router = Router()
@router.message(Command('start'))
async def cmd_start(msg: Message):
# Добавляем контекст в Sentry
with sentry_sdk.configure_scope() as scope:
scope.set_user({"id": msg.from_user.id})
scope.set_tag("chat_type", msg.chat.type)
try:
await msg.answer('Привет!')
except Exception as e:
sentry_sdk.capture_exception(e)
raise# bot/middleware/sentry.py
import sentry_sdk
from aiogram import BaseMiddleware
from aiogram.types import Update
class SentryMiddleware(BaseMiddleware):
"""
Добавляет контекст в Sentry.
💡 Зачем:
- Автоматическое добавление user_id
- Тип обновления для фильтрации
"""
async def __call__(
self,
handler: Callable,
event: Update,
data: Dict[str, Any]
) -> Any:
with sentry_sdk.configure_scope() as scope:
user_id = self._get_user_id(event)
if user_id:
scope.set_user({"id": user_id})
scope.set_tag("update_type", self._get_type(event))
try:
return await handler(event, data)
except Exception as e:
sentry_sdk.capture_exception(e)
raise
def _get_user_id(self, event: Update) -> int:
if event.message:
return event.message.from_user.id
elif event.callback_query:
return event.callback_query.from_user.id
return 0
def _get_type(self, event: Update) -> str:
if event.message:
return 'message'
elif event.callback_query:
return 'callback_query'
return 'unknown'Проблема: Вы узнаёте об ошибке от пользователей, а не от системы мониторинга.
Решение: Prometheus Alertmanager отправляет уведомления при превышении порогов.
# alertmanager.yml
route:
receiver: 'telegram'
group_by: ['alertname']
group_wait: 30s
group_interval: 5m
receivers:
- name: 'telegram'
webhook_configs:
- url: 'http://alertmanager-telegram-bot:8080/alert'💡 Зачем: Уведомления в Telegram о критичных проблемах.
# prometheus_rules.yml
groups:
- name: bot_alerts
rules:
- alert: HighErrorRate
expr: rate(bot_errors_total[5m]) > 0.1
for: 5m
labels:
severity: critical
annotations:
summary: "High error rate detected"
description: "Error rate is {{ $value }} errors/sec"
- alert: HighLatency
expr: histogram_quantile(0.95, rate(bot_handler_duration_seconds_bucket[5m])) > 1
for: 10m
labels:
severity: warning
annotations:
summary: "High latency detected"
description: "P95 latency is {{ $value }}s"
- alert: BotDown
expr: up{job="bot"} == 0
for: 1m
labels:
severity: critical
annotations:
summary: "Bot is down"
description: "Bot instance {{ $labels.instance }} is not responding"💡 Совет: Настройте алерты для:
- Error rate > 1% за 5 минут
- P95 latency > 2 секунд
- Бот не отвечает (up == 0)
# docker-compose.monitoring.yml
version: '3.8'
services:
prometheus:
image: prom/prometheus:latest
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
ports:
- "9090:9090"
networks:
- bot-network
grafana:
image: grafana/grafana:latest
volumes:
- grafana_data:/var/lib/grafana
- ./grafana/provisioning:/etc/grafana/provisioning
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
ports:
- "3000:3000"
depends_on:
- prometheus
networks:
- bot-network
alertmanager:
image: prom/alertmanager:latest
volumes:
- ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
ports:
- "9093:9093"
networks:
- bot-network
volumes:
prometheus_data:
grafana_data:# prometheus.yml
global:
scrape_interval: 15s
scrape_configs:
- job_name: 'bot'
static_configs:
- targets: ['bot:8000']
metrics_path: '/metrics'⚠️ Важно: Вынесите Prometheus и Grafana на отдельные серверы для продакшена.
| Метрика | Описание | Тип | Порог алерта |
|---|---|---|---|
| Messages/sec | Количество обрабатываемых сообщений | Counter | — |
| Error rate | Процент ошибок | Counter | >1% за 5 мин |
| Handler latency | Время обработки handlers | Histogram | p95 > 2с |
| Active users | Активные пользователи | Gauge | — |
| Queue size | Размер очереди задач | Gauge | >1000 |
| Database connections | Активные соединения с БД | Gauge | >80% пула |
| Memory usage | Использование памяти | Gauge | >80% |
| CPU usage | Использование CPU | Gauge | >80% |
| Тема | Как связана |
|---|---|
| Архитектура | Middleware для сбора метрик |
| Безопасность | Аудит-логи для расследования инцидентов |
| Масштабирование | Метрики для автоскейлинга |
| Тестирование | Тесты на алерты |
→ CI/CD — автоматический деплой с проверкой метрик
→ Безопасность — детектирование атак через метрики
→ Масштабирование — автоскейлинг на основе метрик
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.