perf: async Proxmox API, optimized IPAM queries, log cleanup scheduling, background template import

- proxmox_client.py: add ThreadPoolExecutor(max_workers=20) + async wrapper
  methods (async_get_nodes, async_get_vms, async_get_containers,
  async_get_node_status, async_get_vm_status, async_get_container_status,
  async_get_storages, async_get_all_resources) to avoid blocking event loop
- ipam_service.py: replace full-network allocation scan + Python IP loop with
  per-pool targeted SQL query — only loads IPs for the specific pool being searched
- api/ipam.py: fix N+1 full-table scans on GET /api/allocations — use IN-filtered
  queries on only the network/pool IDs present in the result set
- monitoring_worker.py: wire LoggingService.cleanup_old_logs(days=30) into the
  existing run_cleanup_expired() job so audit logs don't grow unboundedly
- api/templates.py: convert auto_import_templates to async endpoint using
  BackgroundTasks — returns immediately, heavy Proxmox I/O runs in background

https://claude.ai/code/session_019rK2wvaWg5zXFwmAPPEB7R
This commit is contained in:
Claude
2026-03-07 11:48:53 +00:00
parent f4b6b36f3b
commit e05650cc9b
5 changed files with 165 additions and 117 deletions

View File

@@ -437,11 +437,27 @@ async def get_allocations(
) )
allocations = query.order_by(IPAMAllocation.ip_address).offset(offset).limit(limit).all() allocations = query.order_by(IPAMAllocation.ip_address).offset(offset).limit(limit).all()
# Add network/pool names # Resolve network/pool names with targeted queries (only IDs present in results)
networks = {n.id: n.name for n in db.query(IPAMNetwork).all()} network_ids = {a.network_id for a in allocations if a.network_id}
pools = {p.id: p.name for p in db.query(IPAMPool).all()} pool_ids = {a.pool_id for a in allocations if a.pool_id}
networks = {}
if network_ids:
networks = {
n.id: n.name for n in
db.query(IPAMNetwork.id, IPAMNetwork.name)
.filter(IPAMNetwork.id.in_(network_ids)).all()
}
pools = {}
if pool_ids:
pools = {
p.id: p.name for p in
db.query(IPAMPool.id, IPAMPool.name)
.filter(IPAMPool.id.in_(pool_ids)).all()
}
result = [] result = []
for alloc in allocations: for alloc in allocations:
alloc_dict = { alloc_dict = {
@@ -450,7 +466,7 @@ async def get_allocations(
'pool_name': pools.get(alloc.pool_id) if alloc.pool_id else None 'pool_name': pools.get(alloc.pool_id) if alloc.pool_id else None
} }
result.append(alloc_dict) result.append(alloc_dict)
return result return result

View File

@@ -3,14 +3,14 @@ API endpoints for OS Templates management
Allows creating VM templates, groups, and deploying VMs from templates Allows creating VM templates, groups, and deploying VMs from templates
""" """
from fastapi import APIRouter, Depends, Request, HTTPException, status from fastapi import APIRouter, BackgroundTasks, Depends, Request, HTTPException, status
from fastapi.responses import HTMLResponse, JSONResponse from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from loguru import logger from loguru import logger
from typing import List from typing import List
from ..db import get_db from ..db import get_db, SessionLocal
from ..models import ProxmoxServer, OSTemplateGroup, OSTemplate, VMInstance, User from ..models import ProxmoxServer, OSTemplateGroup, OSTemplate, VMInstance, User
from ..schemas import ( from ..schemas import (
OSTemplateGroupCreate, OSTemplateGroupUpdate, OSTemplateGroupResponse, OSTemplateGroupCreate, OSTemplateGroupUpdate, OSTemplateGroupResponse,
@@ -413,23 +413,18 @@ def detect_os_group(template_name: str) -> dict:
return OS_GROUP_MAPPING['other'] return OS_GROUP_MAPPING['other']
@router.post("/api/auto-import/{server_id}") def _do_auto_import(server_id: int, username: str):
def auto_import_templates(
server_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("templates.manage"))
):
""" """
Автоматический импорт шаблонов с Proxmox сервера. Background task: import Proxmox templates into DB.
Автоматически определяет группу ОС и назначает порядок сортировки. Opens its own DB session so it can run after the request completes.
Глобальная сортировка: Ubuntu 0-3, CentOS 4-5 и т.д.
""" """
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first() db = SessionLocal()
if not server:
raise HTTPException(status_code=404, detail="Proxmox сервер не найден")
try: try:
# Подключаемся к Proxmox server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first()
if not server:
logger.error(f"Auto-import: server {server_id} not found")
return
if server.use_password: if server.use_password:
client = ProxmoxClient( client = ProxmoxClient(
host=server.ip_address, host=server.ip_address,
@@ -445,46 +440,33 @@ def auto_import_templates(
token_value=server.api_token_value, token_value=server.api_token_value,
verify_ssl=server.verify_ssl verify_ssl=server.verify_ssl
) )
if not client.is_connected(): if not client.is_connected():
raise HTTPException(status_code=503, detail="Не удалось подключиться к Proxmox серверу") logger.error(f"Auto-import: cannot connect to server {server.name}")
return
# Фильтруем по ноде если это кластер
node_filter = None node_filter = None
if server.hostname and server.hostname != server.ip_address: if server.hostname and server.hostname != server.ip_address:
node_filter = server.hostname node_filter = server.hostname
# Получаем шаблоны только с этой ноды
proxmox_templates = client.get_templates(node=node_filter) proxmox_templates = client.get_templates(node=node_filter)
# Получаем уже добавленные шаблоны для ЭТОГО сервера
existing_vmids = set( existing_vmids = set(
t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all() t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
) )
# Кеш групп: name -> group groups_cache = {g.name: g for g in db.query(OSTemplateGroup).all()}
groups_cache = {}
for group in db.query(OSTemplateGroup).all(): templates_by_group = {}
groups_cache[group.name] = group
# Группируем шаблоны по ОС
templates_by_group = {} # group_name -> [templates]
for tpl in proxmox_templates: for tpl in proxmox_templates:
vmid = tpl.get('vmid') vmid = tpl.get('vmid')
if vmid in existing_vmids: if vmid in existing_vmids:
continue # Пропускаем уже добавленные continue
template_name = tpl.get('name', f"Template-{vmid}") template_name = tpl.get('name', f"Template-{vmid}")
group_info = detect_os_group(template_name) group_info = detect_os_group(template_name)
group_name = group_info['name'] group_name = group_info['name']
if group_name not in templates_by_group: if group_name not in templates_by_group:
templates_by_group[group_name] = { templates_by_group[group_name] = {'info': group_info, 'templates': []}
'info': group_info,
'templates': []
}
templates_by_group[group_name]['templates'].append({ templates_by_group[group_name]['templates'].append({
'vmid': vmid, 'vmid': vmid,
'name': template_name, 'name': template_name,
@@ -492,49 +474,34 @@ def auto_import_templates(
'maxmem': tpl.get('maxmem', 0), 'maxmem': tpl.get('maxmem', 0),
'maxdisk': tpl.get('maxdisk', 0), 'maxdisk': tpl.get('maxdisk', 0),
}) })
# Создаём группы и шаблоны с ГЛОБАЛЬНОЙ сортировкой
imported = [] imported = []
global_sort_order = 0 # Глобальный счётчик для всех шаблонов global_sort_order = 0
sorted_groups = sorted(templates_by_group.items(), key=lambda x: x[1]['info']['order'])
# Сортируем группы по их порядку (Ubuntu=1, Debian=2, CentOS=3...)
sorted_groups = sorted(
templates_by_group.items(),
key=lambda x: x[1]['info']['order']
)
for group_name, group_data in sorted_groups: for group_name, group_data in sorted_groups:
group_info = group_data['info'] group_info = group_data['info']
group_templates = group_data['templates'] group_templates = group_data['templates']
# Получаем или создаём группу
if group_name not in groups_cache: if group_name not in groups_cache:
# Получаем максимальный sort_order для групп
max_group_order = db.query(OSTemplateGroup).count()
new_group = OSTemplateGroup( new_group = OSTemplateGroup(
name=group_name, name=group_name,
icon=group_info['icon'], icon=group_info['icon'],
description=f"Шаблоны {group_name}", description=f"Шаблоны {group_name}",
sort_order=group_info['order'], # Используем порядок из маппинга sort_order=group_info['order'],
is_active=True is_active=True
) )
db.add(new_group) db.add(new_group)
db.flush() db.flush()
groups_cache[group_name] = new_group groups_cache[group_name] = new_group
logger.info(f"Created new group: {group_name}") logger.info(f"Auto-import: created group {group_name}")
group = groups_cache[group_name] group = groups_cache[group_name]
# Сортируем шаблоны по имени для консистентности
group_templates.sort(key=lambda x: x['name']) group_templates.sort(key=lambda x: x['name'])
# Добавляем шаблоны с ГЛОБАЛЬНЫМ sort_order for tpl in group_templates:
for idx, tpl in enumerate(group_templates):
# Определяем дефолтные значения из размера шаблона
default_memory = max(1024, (tpl.get('maxmem', 0) // (1024 * 1024)) or 1024) default_memory = max(1024, (tpl.get('maxmem', 0) // (1024 * 1024)) or 1024)
default_disk = max(10, (tpl.get('maxdisk', 0) // (1024 * 1024 * 1024)) or 10) default_disk = max(10, (tpl.get('maxdisk', 0) // (1024 * 1024 * 1024)) or 10)
template = OSTemplate( template = OSTemplate(
group_id=group.id, group_id=group.id,
server_id=server_id, server_id=server_id,
@@ -549,37 +516,47 @@ def auto_import_templates(
min_memory=512, min_memory=512,
min_disk=5, min_disk=5,
is_active=True, is_active=True,
sort_order=global_sort_order # ГЛОБАЛЬНАЯ сортировка sort_order=global_sort_order
) )
db.add(template) db.add(template)
imported.append({ imported.append({'vmid': tpl['vmid'], 'name': tpl['name'], 'group': group_name})
'vmid': tpl['vmid'], global_sort_order += 1
'name': tpl['name'],
'group': group_name,
'sort_order': global_sort_order
})
global_sort_order += 1 # Увеличиваем глобальный счётчик
db.commit() db.commit()
logger.info(f"Auto-import by {username}: imported {len(imported)} templates from {server.name}")
logger.info(f"User {current_user.username} auto-imported {len(imported)} templates from {server.name}")
return JSONResponse(content={
'success': True,
'server_id': server_id,
'server_name': server.name,
'imported_count': len(imported),
'templates': imported,
'groups_created': [g for g in templates_by_group.keys() if g not in groups_cache]
})
except HTTPException:
raise
except Exception as e: except Exception as e:
logger.error(f"Error auto-importing templates from {server.name}: {e}") logger.error(f"Auto-import background task failed for server {server_id}: {e}")
db.rollback() db.rollback()
raise HTTPException(status_code=500, detail=f"Ошибка импорта шаблонов: {str(e)}") finally:
db.close()
@router.post("/api/auto-import/{server_id}")
async def auto_import_templates(
server_id: int,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("templates.manage"))
):
"""
Автоматический импорт шаблонов с Proxmox сервера (в фоне).
Возвращает сразу, импорт выполняется асинхронно.
"""
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Proxmox сервер не найден")
background_tasks.add_task(_do_auto_import, server_id, current_user.username)
logger.info(f"User {current_user.username} queued auto-import for server {server.name}")
return JSONResponse(content={
'success': True,
'status': 'queued',
'server_id': server_id,
'server_name': server.name,
'message': 'Импорт шаблонов запущен в фоне'
})
# ==================== VM Deployment ==================== # ==================== VM Deployment ====================

View File

@@ -8,7 +8,7 @@ import logging
from typing import List, Optional, Dict, Any, Tuple from typing import List, Optional, Dict, Any, Tuple
from datetime import datetime, timezone from datetime import datetime, timezone
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy import and_, or_ from sqlalchemy import and_, or_, text
from .models import IPAMNetwork, IPAMPool, IPAMAllocation, IPAMHistory from .models import IPAMNetwork, IPAMPool, IPAMAllocation, IPAMHistory
@@ -183,31 +183,32 @@ class IPAMService:
logger.warning(f"No active auto-assign pools found for network {network_id}") logger.warning(f"No active auto-assign pools found for network {network_id}")
return None return None
# Get all allocated IPs in this network
allocated_ips = set(
a.ip_address for a in
self.db.query(IPAMAllocation).filter(
IPAMAllocation.network_id == network_id,
IPAMAllocation.status.in_(['allocated', 'reserved'])
).all()
)
# Also exclude gateway # Also exclude gateway
if network.gateway: gateway_ip = network.gateway if network.gateway else None
allocated_ips.add(network.gateway)
# Search through pools for available IP — query only per-pool allocated IPs
# Search through pools for available IP
for pool in pools: for pool in pools:
try: try:
start_ip = ipaddress.ip_address(pool.range_start) start_ip = ipaddress.ip_address(pool.range_start)
end_ip = ipaddress.ip_address(pool.range_end) end_ip = ipaddress.ip_address(pool.range_end)
# Load only allocated IPs for this specific pool (much smaller set)
pool_allocated = set(
a.ip_address for a in
self.db.query(IPAMAllocation.ip_address).filter(
IPAMAllocation.pool_id == pool.id,
IPAMAllocation.status.in_(['allocated', 'reserved'])
).all()
)
if gateway_ip:
pool_allocated.add(gateway_ip)
for ip_int in range(int(start_ip), int(end_ip) + 1): for ip_int in range(int(start_ip), int(end_ip) + 1):
ip = str(ipaddress.ip_address(ip_int)) ip = str(ipaddress.ip_address(ip_int))
if ip not in allocated_ips: if ip not in pool_allocated:
logger.info(f"Found available IP: {ip} in pool {pool.name}") logger.info(f"Found available IP: {ip} in pool {pool.name}")
return ip return ip
except ValueError as e: except ValueError as e:
logger.error(f"Invalid IP range in pool {pool.id}: {e}") logger.error(f"Invalid IP range in pool {pool.id}: {e}")
continue continue

View File

@@ -7,14 +7,24 @@ import time
import asyncio import asyncio
import base64 import base64
from functools import lru_cache from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
from app.ssh_client import SSHClient from app.ssh_client import SSHClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Thread pool for running blocking proxmoxer calls without blocking the event loop
proxmox_executor = ThreadPoolExecutor(max_workers=20, thread_name_prefix="proxmox_")
# Connection cache for reusing Proxmox connections # Connection cache for reusing Proxmox connections
connection_cache = {} connection_cache = {}
async def _run_in_executor(func, *args, **kwargs):
"""Run a blocking function in the proxmox thread pool executor."""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(proxmox_executor, lambda: func(*args, **kwargs))
class ProxmoxClient: class ProxmoxClient:
"""Клиент для работы с Proxmox VE API с кешированием соединений""" """Клиент для работы с Proxmox VE API с кешированием соединений"""
@@ -754,10 +764,44 @@ class ProxmoxClient:
logger.error(f"Ошибка получения хранилищ ноды {node}: {e}") logger.error(f"Ошибка получения хранилищ ноды {node}: {e}")
return [] return []
# ==================== Async wrappers (non-blocking) ====================
async def async_get_nodes(self) -> List[Dict]:
"""Non-blocking version of get_nodes()."""
return await _run_in_executor(self.get_nodes)
async def async_get_vms(self, node: str = None) -> List[Dict]:
"""Non-blocking version of get_vms()."""
return await _run_in_executor(self.get_vms, node)
async def async_get_containers(self, node: str = None) -> List[Dict]:
"""Non-blocking version of get_containers()."""
return await _run_in_executor(self.get_containers, node)
async def async_get_node_status(self, node: str) -> Optional[Dict]:
"""Non-blocking version of get_node_status()."""
return await _run_in_executor(self.get_node_status, node)
async def async_get_vm_status(self, node: str, vmid: int) -> Optional[Dict]:
"""Non-blocking version of get_vm_status()."""
return await _run_in_executor(self.get_vm_status, node, vmid)
async def async_get_container_status(self, node: str, vmid: int) -> Optional[Dict]:
"""Non-blocking version of get_container_status()."""
return await _run_in_executor(self.get_container_status, node, vmid)
async def async_get_storages(self, node: str) -> List[Dict]:
"""Non-blocking version of get_storages()."""
return await _run_in_executor(self.get_storages, node)
async def async_get_all_resources(self, node: str = None) -> Dict[str, List[Dict]]:
"""Non-blocking version of get_all_resources()."""
return await _run_in_executor(self.get_all_resources, node)
def get_next_vmid(self) -> Optional[int]: def get_next_vmid(self) -> Optional[int]:
""" """
Получить следующий свободный VMID Получить следующий свободный VMID
Returns: Returns:
Свободный VMID или None Свободный VMID или None
""" """

View File

@@ -18,6 +18,7 @@ try:
from backend.app.db import SessionLocal from backend.app.db import SessionLocal
from backend.app.models import User, ProxmoxServer, PanelSettings, VMInstance, IPAMAllocation, Notification from backend.app.models import User, ProxmoxServer, PanelSettings, VMInstance, IPAMAllocation, Notification
from backend.app.services.notification_service import NotificationService from backend.app.services.notification_service import NotificationService
from backend.app.logging_service import LoggingService
from backend.app.proxmox_client import ProxmoxClient from backend.app.proxmox_client import ProxmoxClient
from backend.app.schemas import NotificationCreate from backend.app.schemas import NotificationCreate
from backend.app.i18n import t from backend.app.i18n import t
@@ -25,6 +26,7 @@ except ImportError:
from app.db import SessionLocal from app.db import SessionLocal
from app.models import User, ProxmoxServer, PanelSettings, VMInstance, IPAMAllocation, Notification from app.models import User, ProxmoxServer, PanelSettings, VMInstance, IPAMAllocation, Notification
from app.services.notification_service import NotificationService from app.services.notification_service import NotificationService
from app.logging_service import LoggingService
from app.proxmox_client import ProxmoxClient from app.proxmox_client import ProxmoxClient
from app.schemas import NotificationCreate from app.schemas import NotificationCreate
from app.i18n import t from app.i18n import t
@@ -531,7 +533,7 @@ class MonitoringWorker:
db.close() db.close()
def run_cleanup_expired(self): def run_cleanup_expired(self):
"""Clean up expired notifications""" """Clean up expired notifications and old audit logs"""
db = SessionLocal() db = SessionLocal()
try: try:
count = NotificationService.cleanup_expired(db) count = NotificationService.cleanup_expired(db)
@@ -541,6 +543,14 @@ class MonitoringWorker:
logger.error(f"Error cleaning up notifications: {e}") logger.error(f"Error cleaning up notifications: {e}")
finally: finally:
db.close() db.close()
db = SessionLocal()
try:
LoggingService.cleanup_old_logs(db, days=30)
except Exception as e:
logger.error(f"Error cleaning up old audit logs: {e}")
finally:
db.close()
def sync_vm_cache(self): def sync_vm_cache(self):
""" """