diff --git a/app/cabinet/routes/admin_traffic.py b/app/cabinet/routes/admin_traffic.py index c412f9d3..605be3b6 100644 --- a/app/cabinet/routes/admin_traffic.py +++ b/app/cabinet/routes/admin_traffic.py @@ -35,7 +35,7 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix='/admin/traffic', tags=['Admin Traffic']) _ALLOWED_PERIODS = frozenset({1, 3, 7, 14, 30}) -_CONCURRENCY_LIMIT = 20 # Max parallel per-user API calls +_CONCURRENCY_LIMIT = 5 # Max parallel API calls to avoid rate limiting # In-memory cache: {period_days: (timestamp, aggregated_data, nodes_info)} _traffic_cache: dict[int, tuple[float, dict[str, dict[str, int]], list[TrafficNodeInfo]]] = {} @@ -59,9 +59,8 @@ async def _aggregate_traffic( ) -> tuple[dict[str, dict[str, int]], list[TrafficNodeInfo]]: """Aggregate per-user traffic across all nodes for a given period. - Uses get_bandwidth_stats_user() per user (same API as the working - AdminUserDetail page) instead of get_bandwidth_stats_node_users() - which returns UUIDs in a format that may not match the bot DB. + Uses per-node endpoint get_bandwidth_stats_node_users() to fetch traffic + for all users at once per node — O(nodes) API calls instead of O(users). Returns (user_traffic, nodes_info) where: user_traffic = {remnawave_uuid: {node_uuid: total_bytes, ...}} @@ -90,24 +89,26 @@ async def _aggregate_traffic( start_str = start_date.strftime('%Y-%m-%d') end_str = end_date.strftime('%Y-%m-%d') + user_uuids_set = set(user_uuids) + async with service.get_api_client() as api: - # Get all nodes for column headers nodes = await api.get_all_nodes() - # Fetch per-user bandwidth stats in parallel with concurrency limit. - # Response format: {series: [{uuid: NODE_UUID, total: bytes, ...}, ...]} + # Fetch per-node user stats — O(nodes) calls instead of O(users) semaphore = asyncio.Semaphore(_CONCURRENCY_LIMIT) - async def fetch_user_stats(user_uuid: str): + async def fetch_node_users(node): async with semaphore: try: - stats = await api.get_bandwidth_stats_user(user_uuid, start_str, end_str) - return user_uuid, stats + stats = await api.get_bandwidth_stats_node_users( + node.uuid, start_str, end_str, top_users_limit=9999, + ) + return node.uuid, stats except Exception: - logger.warning('Failed to get traffic for user %s', user_uuid[:8], exc_info=True) - return user_uuid, None + logger.warning('Failed to get traffic for node %s', node.name, exc_info=True) + return node.uuid, None - results = await asyncio.gather(*(fetch_user_stats(uid) for uid in user_uuids)) + results = await asyncio.gather(*(fetch_node_users(n) for n in nodes)) nodes_info: list[TrafficNodeInfo] = [ TrafficNodeInfo(node_uuid=node.uuid, node_name=node.name, country_code=node.country_code) for node in nodes @@ -115,17 +116,20 @@ async def _aggregate_traffic( nodes_info.sort(key=lambda n: n.node_name) user_traffic: dict[str, dict[str, int]] = {} - for user_uuid, stats in results: + for node_uuid, stats in results: if not isinstance(stats, dict): continue - node_traffic: dict[str, int] = {} - for series_item in stats.get('series', []): - node_uuid = series_item.get('uuid', '') - total = int(series_item.get('total', 0)) - if node_uuid and total > 0: - node_traffic[node_uuid] = node_traffic.get(node_uuid, 0) + total - if node_traffic: - user_traffic[user_uuid] = node_traffic + # The node-users endpoint may return user data under 'users' or 'series' + entries = stats.get('users', []) or stats.get('series', []) + if not entries and stats: + logger.debug('Unexpected node-users response keys: %s', list(stats.keys())) + for user_entry in entries: + uid = user_entry.get('uuid', '') + total = int(user_entry.get('total', 0)) + if uid and total > 0 and uid in user_uuids_set: + user_traffic.setdefault(uid, {})[node_uuid] = ( + user_traffic.get(uid, {}).get(node_uuid, 0) + total + ) _traffic_cache[period_days] = (now, user_traffic, nodes_info) return user_traffic, nodes_info diff --git a/app/external/remnawave_api.py b/app/external/remnawave_api.py index 82bbe1ba..3475dbd2 100644 --- a/app/external/remnawave_api.py +++ b/app/external/remnawave_api.py @@ -1,3 +1,4 @@ +import asyncio import base64 import json import logging @@ -366,32 +367,51 @@ class RemnaWaveAPI: raise RemnaWaveAPIError('Session not initialized. Use async context manager.') url = f'{self.base_url}{endpoint}' + max_retries = 3 + base_delay = 1.0 - try: - kwargs = {'url': url, 'params': params} + for attempt in range(max_retries + 1): + try: + kwargs = {'url': url, 'params': params} - if data: - kwargs['json'] = data + if data: + kwargs['json'] = data - async with self.session.request(method, **kwargs) as response: - response_text = await response.text() + async with self.session.request(method, **kwargs) as response: + response_text = await response.text() - try: - response_data = json.loads(response_text) if response_text else {} - except json.JSONDecodeError: - response_data = {'raw_response': response_text} + try: + response_data = json.loads(response_text) if response_text else {} + except json.JSONDecodeError: + response_data = {'raw_response': response_text} - if response.status >= 400: - error_message = response_data.get('message', f'HTTP {response.status}') - logger.error(f'API Error {response.status}: {error_message}') - logger.error(f'Response: {response_text[:500]}') - raise RemnaWaveAPIError(error_message, response.status, response_data) + if response.status == 429 and attempt < max_retries: + retry_after = float(response.headers.get('Retry-After', base_delay * (2 ** attempt))) + logger.warning( + 'Rate limited (429) on %s %s, retry %d/%d after %.1fs', + method, endpoint, attempt + 1, max_retries, retry_after, + ) + await asyncio.sleep(retry_after) + continue - return response_data + if response.status >= 400: + error_message = response_data.get('message', f'HTTP {response.status}') + logger.error(f'API Error {response.status}: {error_message}') + logger.error(f'Response: {response_text[:500]}') + raise RemnaWaveAPIError(error_message, response.status, response_data) - except aiohttp.ClientError as e: - logger.error(f'Request failed: {e}') - raise RemnaWaveAPIError(f'Request failed: {e!s}') + return response_data + + except aiohttp.ClientError as e: + if attempt < max_retries: + delay = base_delay * (2 ** attempt) + logger.warning('Request failed on %s %s: %s, retry %d/%d after %.1fs', method, endpoint, e, attempt + 1, max_retries, delay) + await asyncio.sleep(delay) + continue + logger.error(f'Request failed: {e}') + raise RemnaWaveAPIError(f'Request failed: {e!s}') + + raise RemnaWaveAPIError(f'Max retries exceeded for {method} {endpoint}') async def create_user( self,