markmorado/pvemanager
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases 3 Wiki Activity

Merge pull request #5 from markmorado/claude/code-review-audit-cxCWB

Browse Source 
perf: async Proxmox API, optimized IPAM queries, log cleanup scheduli…
This commit is contained in:
Dilshod Rasulov
2026-03-07 16:49:50 +05:00
committed by GitHub
parent f4b6b36f3b e05650cc9b
commit 1aa0819883
5 changed files with 165 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
backend/app/api/ipam.py
View File

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

161
backend/app/api/templates.py
View File

@@ -3,14 +3,14 @@ API endpoints for OS Templates management
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.templating import Jinja2Templates
from sqlalchemy.orm import Session
from loguru import logger
from typing import List
from ..db import get_db
from ..db import get_db, SessionLocal
from ..models import ProxmoxServer, OSTemplateGroup, OSTemplate, VMInstance, User
from ..schemas import (
OSTemplateGroupCreate, OSTemplateGroupUpdate, OSTemplateGroupResponse,
@@ -413,23 +413,18 @@ def detect_os_group(template_name: str) -> dict:
return OS_GROUP_MAPPING['other']
@router.post("/api/auto-import/{server_id}")
def auto_import_templates(
server_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("templates.manage"))
):
def _do_auto_import(server_id: int, username: str):
"""
Автоматический импорт шаблонов с Proxmox сервера.
Автоматически определяет группу ОС и назначает порядок сортировки.
Глобальная сортировка: Ubuntu 0-3, CentOS 4-5 и т.д.
Background task: import Proxmox templates into DB.
Opens its own DB session so it can run after the request completes.
"""
server = db.query(ProxmoxServer).filter(ProxmoxServer.id == server_id).first()
if not server:
raise HTTPException(status_code=404, detail="Proxmox сервер не найден")
db = SessionLocal()
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:
client = ProxmoxClient(
host=server.ip_address,
@@ -445,46 +440,33 @@ def auto_import_templates(
token_value=server.api_token_value,
verify_ssl=server.verify_ssl
)
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
if server.hostname and server.hostname != server.ip_address:
node_filter = server.hostname
# Получаем шаблоны только с этой ноды
proxmox_templates = client.get_templates(node=node_filter)
# Получаем уже добавленные шаблоны для ЭТОГО сервера
existing_vmids = set(
t.vmid for t in db.query(OSTemplate).filter(OSTemplate.server_id == server_id).all()
)
# Кеш групп: name -> group
groups_cache = {}
for group in db.query(OSTemplateGroup).all():
groups_cache[group.name] = group
# Группируем шаблоны по ОС
templates_by_group = {} # group_name -> [templates]
groups_cache = {g.name: g for g in db.query(OSTemplateGroup).all()}
templates_by_group = {}
for tpl in proxmox_templates:
vmid = tpl.get('vmid')
if vmid in existing_vmids:
continue # Пропускаем уже добавленные
continue
template_name = tpl.get('name', f"Template-{vmid}")
group_info = detect_os_group(template_name)
group_name = group_info['name']
if group_name not in templates_by_group:
templates_by_group[group_name] = {
'info': group_info,
'templates': []
}
templates_by_group[group_name] = {'info': group_info, 'templates': []}
templates_by_group[group_name]['templates'].append({
'vmid': vmid,
'name': template_name,
@@ -492,49 +474,34 @@ def auto_import_templates(
'maxmem': tpl.get('maxmem', 0),
'maxdisk': tpl.get('maxdisk', 0),
})
# Создаём группы и шаблоны с ГЛОБАЛЬНОЙ сортировкой
imported = []
global_sort_order = 0 # Глобальный счётчик для всех шаблонов
# Сортируем группы по их порядку (Ubuntu=1, Debian=2, CentOS=3...)
sorted_groups = sorted(
templates_by_group.items(),
key=lambda x: x[1]['info']['order']
)
global_sort_order = 0
sorted_groups = sorted(templates_by_group.items(), key=lambda x: x[1]['info']['order'])
for group_name, group_data in sorted_groups:
group_info = group_data['info']
group_templates = group_data['templates']
# Получаем или создаём группу
if group_name not in groups_cache:
# Получаем максимальный sort_order для групп
max_group_order = db.query(OSTemplateGroup).count()
new_group = OSTemplateGroup(
name=group_name,
icon=group_info['icon'],
description=f"Шаблоны {group_name}",
sort_order=group_info['order'], # Используем порядок из маппинга
sort_order=group_info['order'],
is_active=True
)
db.add(new_group)
db.flush()
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_templates.sort(key=lambda x: x['name'])
# Добавляем шаблоны с ГЛОБАЛЬНЫМ sort_order
for idx, tpl in enumerate(group_templates):
# Определяем дефолтные значения из размера шаблона
for tpl in group_templates:
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)
template = OSTemplate(
group_id=group.id,
server_id=server_id,
@@ -549,37 +516,47 @@ def auto_import_templates(
min_memory=512,
min_disk=5,
is_active=True,
sort_order=global_sort_order # ГЛОБАЛЬНАЯ сортировка
sort_order=global_sort_order
)
db.add(template)
imported.append({
'vmid': tpl['vmid'],
'name': tpl['name'],
'group': group_name,
'sort_order': global_sort_order
})
global_sort_order += 1 # Увеличиваем глобальный счётчик
imported.append({'vmid': tpl['vmid'], 'name': tpl['name'], 'group': group_name})
global_sort_order += 1
db.commit()
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
logger.info(f"Auto-import by {username}: imported {len(imported)} templates from {server.name}")
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()
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 ====================

35
backend/app/ipam_service.py
View File

@@ -8,7 +8,7 @@ import logging
from typing import List, Optional, Dict, Any, Tuple
from datetime import datetime, timezone
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from sqlalchemy import and_, or_, text
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}")
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
if network.gateway:
allocated_ips.add(network.gateway)
# Search through pools for available IP
gateway_ip = network.gateway if network.gateway else None
# Search through pools for available IP — query only per-pool allocated IPs
for pool in pools:
try:
start_ip = ipaddress.ip_address(pool.range_start)
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):
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}")
return ip
except ValueError as e:
logger.error(f"Invalid IP range in pool {pool.id}: {e}")
continue

46
backend/app/proxmox_client.py
View File

@@ -7,14 +7,24 @@ import time
import asyncio
import base64
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
from app.ssh_client import SSHClient
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 = {}
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:
"""Клиент для работы с Proxmox VE API с кешированием соединений"""
@@ -754,10 +764,44 @@ class ProxmoxClient:
logger.error(f"Ошибка получения хранилищ ноды {node}: {e}")
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]:
"""
Получить следующий свободный VMID
Returns:
Свободный VMID или None
"""

12
backend/app/workers/monitoring_worker.py
View File

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