Andrey
e2b1353408
- Создана система отслеживания миграций (MigrationRepository, таблица migrations) - Добавлен скрипт apply_migrations.py для автоматического применения миграций - Созданы CI/CD пайплайны (.github/workflows/ci.yml, deploy.yml) - Обновлена документация по миграциям в database-patterns.md - Миграции применяются автоматически при деплое в продакшн
222 lines
8.8 KiB
Python
222 lines
8.8 KiB
Python
"""
|
||
Мониторинг и статистика rate limiting
|
||
"""
|
||
import time
|
||
from collections import defaultdict, deque
|
||
from dataclasses import dataclass, field
|
||
from typing import Dict, List, Optional
|
||
|
||
from logs.custom_logger import logger
|
||
|
||
|
||
@dataclass
|
||
class RateLimitStats:
|
||
"""Статистика rate limiting для чата"""
|
||
chat_id: int
|
||
total_requests: int = 0
|
||
successful_requests: int = 0
|
||
failed_requests: int = 0
|
||
retry_after_errors: int = 0
|
||
other_errors: int = 0
|
||
total_wait_time: float = 0.0
|
||
last_request_time: float = 0.0
|
||
request_times: deque = field(default_factory=lambda: deque(maxlen=100))
|
||
|
||
@property
|
||
def success_rate(self) -> float:
|
||
"""Процент успешных запросов"""
|
||
if self.total_requests == 0:
|
||
return 1.0
|
||
return self.successful_requests / self.total_requests
|
||
|
||
@property
|
||
def error_rate(self) -> float:
|
||
"""Процент ошибок"""
|
||
return 1.0 - self.success_rate
|
||
|
||
@property
|
||
def average_wait_time(self) -> float:
|
||
"""Среднее время ожидания"""
|
||
if self.total_requests == 0:
|
||
return 0.0
|
||
return self.total_wait_time / self.total_requests
|
||
|
||
@property
|
||
def requests_per_minute(self) -> float:
|
||
"""Запросов в минуту"""
|
||
if not self.request_times:
|
||
return 0.0
|
||
|
||
current_time = time.time()
|
||
minute_ago = current_time - 60
|
||
|
||
# Подсчитываем запросы за последнюю минуту
|
||
recent_requests = sum(1 for req_time in self.request_times if req_time > minute_ago)
|
||
return recent_requests
|
||
|
||
|
||
class RateLimitMonitor:
|
||
"""Монитор для отслеживания статистики rate limiting"""
|
||
|
||
def __init__(self, max_history_size: int = 1000):
|
||
self.stats: Dict[int, RateLimitStats] = defaultdict(lambda: RateLimitStats(0))
|
||
self.global_stats = RateLimitStats(0)
|
||
self.max_history_size = max_history_size
|
||
self.error_history: deque = deque(maxlen=max_history_size)
|
||
|
||
def record_request(self, chat_id: int, success: bool, wait_time: float = 0.0, error_type: Optional[str] = None):
|
||
"""Записывает информацию о запросе"""
|
||
current_time = time.time()
|
||
|
||
# Обновляем статистику для чата
|
||
chat_stats = self.stats[chat_id]
|
||
chat_stats.chat_id = chat_id
|
||
chat_stats.total_requests += 1
|
||
chat_stats.total_wait_time += wait_time
|
||
chat_stats.last_request_time = current_time
|
||
chat_stats.request_times.append(current_time)
|
||
|
||
if success:
|
||
chat_stats.successful_requests += 1
|
||
else:
|
||
chat_stats.failed_requests += 1
|
||
if error_type == "RetryAfter":
|
||
chat_stats.retry_after_errors += 1
|
||
else:
|
||
chat_stats.other_errors += 1
|
||
|
||
# Записываем ошибку в историю
|
||
self.error_history.append({
|
||
'chat_id': chat_id,
|
||
'error_type': error_type,
|
||
'timestamp': current_time,
|
||
'wait_time': wait_time
|
||
})
|
||
|
||
# Обновляем глобальную статистику
|
||
self.global_stats.total_requests += 1
|
||
self.global_stats.total_wait_time += wait_time
|
||
self.global_stats.last_request_time = current_time
|
||
self.global_stats.request_times.append(current_time)
|
||
|
||
if success:
|
||
self.global_stats.successful_requests += 1
|
||
else:
|
||
self.global_stats.failed_requests += 1
|
||
if error_type == "RetryAfter":
|
||
self.global_stats.retry_after_errors += 1
|
||
else:
|
||
self.global_stats.other_errors += 1
|
||
|
||
def get_chat_stats(self, chat_id: int) -> Optional[RateLimitStats]:
|
||
"""Получает статистику для конкретного чата"""
|
||
return self.stats.get(chat_id)
|
||
|
||
def get_global_stats(self) -> RateLimitStats:
|
||
"""Получает глобальную статистику"""
|
||
return self.global_stats
|
||
|
||
def get_top_chats_by_requests(self, limit: int = 10) -> List[tuple]:
|
||
"""Получает топ чатов по количеству запросов"""
|
||
sorted_chats = sorted(
|
||
self.stats.items(),
|
||
key=lambda x: x[1].total_requests,
|
||
reverse=True
|
||
)
|
||
return sorted_chats[:limit]
|
||
|
||
def get_chats_with_high_error_rate(self, threshold: float = 0.1) -> List[tuple]:
|
||
"""Получает чаты с высоким процентом ошибок"""
|
||
high_error_chats = [
|
||
(chat_id, stats) for chat_id, stats in self.stats.items()
|
||
if stats.error_rate > threshold and stats.total_requests > 5
|
||
]
|
||
return sorted(high_error_chats, key=lambda x: x[1].error_rate, reverse=True)
|
||
|
||
def get_recent_errors(self, minutes: int = 60) -> List[dict]:
|
||
"""Получает недавние ошибки"""
|
||
current_time = time.time()
|
||
cutoff_time = current_time - (minutes * 60)
|
||
|
||
return [
|
||
error for error in self.error_history
|
||
if error['timestamp'] > cutoff_time
|
||
]
|
||
|
||
def get_error_summary(self, minutes: int = 60) -> Dict[str, int]:
|
||
"""Получает сводку ошибок за указанный период"""
|
||
recent_errors = self.get_recent_errors(minutes)
|
||
error_summary = defaultdict(int)
|
||
|
||
for error in recent_errors:
|
||
error_summary[error['error_type']] += 1
|
||
|
||
return dict(error_summary)
|
||
|
||
def log_statistics(self, log_level: str = "info"):
|
||
"""Логирует текущую статистику"""
|
||
global_stats = self.get_global_stats()
|
||
|
||
log_message = (
|
||
f"Rate Limit Statistics:\n"
|
||
f" Total requests: {global_stats.total_requests}\n"
|
||
f" Success rate: {global_stats.success_rate:.2%}\n"
|
||
f" Error rate: {global_stats.error_rate:.2%}\n"
|
||
f" RetryAfter errors: {global_stats.retry_after_errors}\n"
|
||
f" Other errors: {global_stats.other_errors}\n"
|
||
f" Average wait time: {global_stats.average_wait_time:.2f}s\n"
|
||
f" Requests per minute: {global_stats.requests_per_minute:.1f}\n"
|
||
f" Active chats: {len(self.stats)}"
|
||
)
|
||
|
||
if log_level == "error":
|
||
logger.error(log_message)
|
||
elif log_level == "warning":
|
||
logger.warning(log_message)
|
||
else:
|
||
logger.info(log_message)
|
||
|
||
# Логируем чаты с высоким процентом ошибок
|
||
high_error_chats = self.get_chats_with_high_error_rate(0.2)
|
||
if high_error_chats:
|
||
logger.warning(f"Chats with high error rate (>20%): {len(high_error_chats)}")
|
||
for chat_id, stats in high_error_chats[:5]: # Показываем только первые 5
|
||
logger.warning(f" Chat {chat_id}: {stats.error_rate:.2%} error rate ({stats.failed_requests}/{stats.total_requests})")
|
||
|
||
def reset_stats(self, chat_id: Optional[int] = None):
|
||
"""Сбрасывает статистику"""
|
||
if chat_id is None:
|
||
# Сбрасываем всю статистику
|
||
self.stats.clear()
|
||
self.global_stats = RateLimitStats(0)
|
||
self.error_history.clear()
|
||
else:
|
||
# Сбрасываем статистику для конкретного чата
|
||
if chat_id in self.stats:
|
||
del self.stats[chat_id]
|
||
|
||
|
||
# Глобальный экземпляр монитора
|
||
rate_limit_monitor = RateLimitMonitor()
|
||
|
||
|
||
def record_rate_limit_request(chat_id: int, success: bool, wait_time: float = 0.0, error_type: Optional[str] = None):
|
||
"""Удобная функция для записи информации о запросе"""
|
||
rate_limit_monitor.record_request(chat_id, success, wait_time, error_type)
|
||
|
||
|
||
def get_rate_limit_summary() -> Dict:
|
||
"""Получает краткую сводку по rate limiting"""
|
||
global_stats = rate_limit_monitor.get_global_stats()
|
||
recent_errors = rate_limit_monitor.get_recent_errors(60) # За последний час
|
||
|
||
return {
|
||
'total_requests': global_stats.total_requests,
|
||
'success_rate': global_stats.success_rate,
|
||
'error_rate': global_stats.error_rate,
|
||
'recent_errors_count': len(recent_errors),
|
||
'active_chats': len(rate_limit_monitor.stats),
|
||
'requests_per_minute': global_stats.requests_per_minute,
|
||
'average_wait_time': global_stats.average_wait_time
|
||
}
|