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

fix(api): scope server listings to workspace, stop serialising Proxmox calls

Browse Source 
/api/cluster/topology and /api/resources/all read every ProxmoxServer row
regardless of the caller access or the X-Active-Workspace header, so the
Cluster page listed all servers and dashboard aggregates counted the whole
panel instead of the active workspace. Extract the access + workspace filter
out of /api/servers into get_visible_server_ids() and apply it in all three.

Live metrics were also gathered server by server with the cache lock held
across the network calls, so a single unreachable cluster stalled the request
(21-31s in the logs) and concurrent requests queued behind it. Fetch in
parallel, skip servers already known offline, drop the redundant
is_connected() probe, and hold the lock only around the cache dict.

/api/resources/all now orders by server id: without it Postgres returned heap
order, which shifts whenever update_status rewrites a row, so the server list
reshuffled between requests.
This commit is contained in:
markmorado
2026-07-24 03:18:04 +05:00
parent 35a8a11bb5
commit ff1396142b
4 changed files with 202 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
backend/app/api/proxmox/_helpers.py
View File

@@ -41,6 +41,58 @@ def can_view_all_instances(current_user: User) -> bool:
return bool(perms.get('vm:manage', False))
def get_visible_server_ids(
request: Request, db: Session, current_user: User
) -> Optional[set]:
"""
Set of Proxmox server IDs the user may see, or ``None`` for "no restriction".
Two filters combine here:
* **Access** — privileged users (admin/moderator) see every server; everyone
else sees servers assigned to them directly OR reachable through a
workspace they belong to. Adding a user to a workspace is therefore
enough — no separate direct assignment is required.
* **Active workspace** — the ``X-Active-Workspace`` header narrows the result
further, for privileged and non-privileged users alike.
``None`` means "everything" (privileged user with no active workspace) and is
deliberately distinct from an empty set, which means "nothing visible".
Callers must treat the two differently — that distinction is exactly what
``/api/cluster/topology`` used to get wrong by skipping this check entirely.
"""
from ...api.workspaces import get_workspace_server_ids
from ...models import WorkspaceUser, WorkspaceServer
ws_filter = get_workspace_server_ids(request, db, current_user)
is_privileged = current_user.is_admin or (
current_user.role and current_user.role.name in ('admin', 'moderator')
)
if is_privileged:
return None if ws_filter is None else set(ws_filter)
assigned_ids = {s.id for s in current_user.assigned_servers}
user_ws_ids = {
r.workspace_id for r in
db.query(WorkspaceUser.workspace_id).filter(WorkspaceUser.user_id == current_user.id).all()
}
ws_server_ids = set()
if user_ws_ids:
ws_server_ids = {
r.server_id for r in
db.query(WorkspaceServer.server_id).filter(
WorkspaceServer.workspace_id.in_(user_ws_ids)
).all()
}
effective_ids = assigned_ids | ws_server_ids
if ws_filter is not None:
effective_ids &= set(ws_filter)
return effective_ids
def check_vm_access(db: Session, current_user: User, server_id: int, vmid: int) -> bool:
"""
Check if user has access to a specific VM instance.

19
backend/app/api/proxmox/cluster.py
View File

@@ -5,7 +5,7 @@ Endpoints for creating Proxmox clusters, joining nodes, ejecting nodes,
and preparing nodes for join (backup + cleanup).
"""
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from loguru import logger
@@ -16,7 +16,7 @@ from ...db import get_db
from ...models import ProxmoxServer, User
from ...auth import PermissionChecker
from ...proxmox import ProxmoxClient
from ._helpers import _get_proxmox_client, _extract_fingerprint
from ._helpers import _get_proxmox_client, _extract_fingerprint, get_visible_server_ids
from ...proxmox import _run_in_executor
router = APIRouter()
@@ -58,14 +58,25 @@ def _get_password_client(server: ProxmoxServer, override_password: str = None) -
@router.get("/api/cluster/topology")
def get_cluster_topology(
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("server:view")),
):
"""
Получить все серверы, сгруппированные по кластеру.
Получить серверы, видимые пользователю, сгруппированные по кластеру.
Standalone-серверы (cluster_name=NULL) — в отдельном списке.
Фильтруется так же, как GET /api/servers: по доступу пользователя и по
активной рабочей области (заголовок X-Active-Workspace).
"""
servers = db.query(ProxmoxServer).all()
visible_ids = get_visible_server_ids(request, db, current_user)
if visible_ids is not None and not visible_ids:
return {"clusters": {}, "standalone": []}
query = db.query(ProxmoxServer)
if visible_ids is not None:
query = query.filter(ProxmoxServer.id.in_(visible_ids))
servers = query.all()
clusters: dict = {}
standalone = []

203
backend/app/api/proxmox/servers.py
View File

@@ -8,6 +8,9 @@ import ssl
import asyncio
import httpx
import websockets
from concurrent.futures import ThreadPoolExecutor, as_completed
RESOURCES_MAX_WORKERS = 8 # parallel Proxmox fetches for /api/resources/all
from ...db import get_db
from ...models import ProxmoxServer, VMInstance, User, IPAMAllocation, IPAMNetwork, VMSnapshotArchive
@@ -19,7 +22,7 @@ from ...ipam_service import IPAMService
from ._helpers import (check_vm_access, require_vm_access, _get_proxmox_client,
get_next_vmid, archive_and_delete_snapshots,
save_vm_instance, get_vm_instance, soft_delete_vm_instance,
can_view_all_instances, get_owned_vmids)
can_view_all_instances, get_owned_vmids, get_visible_server_ids)
router = APIRouter()
@@ -59,58 +62,15 @@ def list_proxmox_servers(
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("server:view"))
):
"""Получить список всех Proxmox серверов (фильтруется по активному workspace)"""
from ...api.workspaces import get_workspace_server_ids
from ...models import WorkspaceUser, WorkspaceServer
is_privileged = current_user.is_admin or (
current_user.role and current_user.role.name in ('admin', 'moderator')
)
if is_privileged:
# Privileged: use workspace filter only
server_ids = get_workspace_server_ids(request, db, current_user)
query = db.query(ProxmoxServer).order_by(ProxmoxServer.id)
if server_ids is not None:
query = query.filter(ProxmoxServer.id.in_(server_ids))
servers = query.all()
_attach_workspaces(db, servers)
return servers
# Non-privileged user: access is granted either by a direct server
# assignment OR by membership in a workspace that contains the server.
# Adding a user to a workspace is therefore sufficient to let them create
# instances on that workspace's servers — no separate direct assignment
# is required.
assigned_ids = {s.id for s in current_user.assigned_servers}
# Collect server IDs accessible through user's workspaces
user_ws_ids = {
r.workspace_id for r in
db.query(WorkspaceUser.workspace_id).filter(WorkspaceUser.user_id == current_user.id).all()
}
if user_ws_ids:
ws_server_ids = {
r.server_id for r in
db.query(WorkspaceServer.server_id).filter(
WorkspaceServer.workspace_id.in_(user_ws_ids)
).all()
}
else:
ws_server_ids = set()
# Union: directly assigned OR within user's workspaces
effective_ids = assigned_ids | ws_server_ids
# Also apply active-workspace header filter on top
ws_filter = get_workspace_server_ids(request, db, current_user)
if ws_filter is not None:
effective_ids = effective_ids & set(ws_filter)
if not effective_ids:
"""Получить список всех Proxmox серверов (фильтруется по доступу и активному workspace)"""
visible_ids = get_visible_server_ids(request, db, current_user)
if visible_ids is not None and not visible_ids:
return []
servers = db.query(ProxmoxServer).filter(ProxmoxServer.id.in_(effective_ids)).order_by(ProxmoxServer.id).all()
query = db.query(ProxmoxServer).order_by(ProxmoxServer.id)
if visible_ids is not None:
query = query.filter(ProxmoxServer.id.in_(visible_ids))
servers = query.all()
_attach_workspaces(db, servers)
return servers
@@ -566,11 +526,26 @@ def get_server_resources(
@router.get("/api/resources/all")
def get_all_resources(
db: Session = Depends(get_db),
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(PermissionChecker("vm:view"))
):
"""API для получения всех ресурсов со всех Proxmox серверов"""
proxmox_servers = db.query(ProxmoxServer).all()
"""API для получения ресурсов со всех видимых пользователю Proxmox серверов.
Набор серверов фильтруется так же, как в GET /api/servers: по доступу
пользователя и по активной рабочей области. Иначе агрегаты (total_vms,
total_containers) считались бы по всей панели, а не по текущей области.
"""
visible_ids = get_visible_server_ids(request, db, current_user)
if visible_ids is not None and not visible_ids:
return JSONResponse(content={"servers": [], "total_vms": 0, "total_containers": 0})
# order_by: without it Postgres returns heap order, which shifts every time
# update_status rewrites a row — the server list would reshuffle per request.
servers_q = db.query(ProxmoxServer).order_by(ProxmoxServer.id)
if visible_ids is not None:
servers_q = servers_q.filter(ProxmoxServer.id.in_(visible_ids))
proxmox_servers = servers_q.all()
# Обычный пользователь видит только свои инстансы во всех агрегатах.
is_privileged = can_view_all_instances(current_user)
@@ -581,71 +556,87 @@ def get_all_resources(
"total_containers": 0
}
# Credentials are read here, on the request thread — the worker threads get
# plain kwargs so they never touch the ORM (Session is not thread-safe and a
# lazy load from a worker would blow up mid-request).
fetch_args = {}
for server in proxmox_servers:
try:
logger.info(f"Getting resources from server {server.name}, use_password={server.use_password}")
# Создаём клиента для каждого сервера независимо
if server.use_password:
resources = get_proxmox_resources(
host=server.ip_address,
user=server.api_user,
password=server.password,
verify_ssl=server.verify_ssl
)
else:
logger.info(f"Using API token for {server.name}: user={server.api_user}")
resources = get_proxmox_resources(
host=server.ip_address,
user=server.api_user,
token_name=server.api_token_name,
token_value=server.api_token_value,
verify_ssl=server.verify_ssl
)
vms = resources.get('vms', [])
containers = resources.get('containers', [])
logger.info(f"Getting resources from server {server.name}, use_password={server.use_password}")
common = dict(host=server.ip_address, user=server.api_user, verify_ssl=server.verify_ssl)
if server.use_password:
fetch_args[server.id] = {**common, 'password': server.password}
else:
logger.info(f"Using API token for {server.name}: user={server.api_user}")
fetch_args[server.id] = {
**common,
'token_name': server.api_token_name,
'token_value': server.api_token_value,
}
# Изоляция по владельцу (живые данные Proxmox не содержат owner_id).
if not is_privileged:
owned = get_owned_vmids(db, current_user, server.id)
vms = [vm for vm in vms if vm.get('vmid') in owned]
containers = [ct for ct in containers if ct.get('vmid') in owned]
# One slow or unreachable server must not hold up the rest: total wait is the
# slowest server, not the sum of all of them.
fetched: dict[int, dict] = {}
errors: dict[int, str] = {}
if proxmox_servers:
with ThreadPoolExecutor(max_workers=min(RESOURCES_MAX_WORKERS, len(proxmox_servers))) as pool:
futures = {
pool.submit(get_proxmox_resources, **fetch_args[s.id]): s
for s in proxmox_servers
}
for future in as_completed(futures):
server = futures[future]
try:
fetched[server.id] = future.result()
except Exception as e:
logger.error(f"Error getting resources from server {server.name} ({server.ip_address}): {e}")
errors[server.id] = str(e)
logger.info(f"Got {len(vms)} VMs and {len(containers)} containers from {server.name}")
server.update_status(True)
db.commit()
# Assemble in the original server order — the response must not reshuffle
# depending on which cluster happened to answer first.
for server in proxmox_servers:
if server.id in errors:
server.update_status(False, errors[server.id])
all_resources["servers"].append({
"id": server.id,
"name": server.name,
"ip": server.ip_address,
"vms": vms,
"containers": containers,
"vms_count": len(vms),
"containers_count": len(containers)
})
all_resources["total_vms"] += len(vms)
all_resources["total_containers"] += len(containers)
except Exception as e:
logger.error(f"Error getting resources from server {server.name} ({server.ip_address}): {e}")
server.update_status(False, str(e))
db.commit()
all_resources["servers"].append({
"id": server.id,
"name": server.name,
"ip": server.ip_address,
"error": str(e),
"error": errors[server.id],
"vms": [],
"containers": [],
"vms_count": 0,
"containers_count": 0
})
# Продолжаем обработку следующего сервера
continue
resources = fetched.get(server.id) or {}
vms = resources.get('vms', [])
containers = resources.get('containers', [])
# Изоляция по владельцу (живые данные Proxmox не содержат owner_id).
if not is_privileged:
owned = get_owned_vmids(db, current_user, server.id)
vms = [vm for vm in vms if vm.get('vmid') in owned]
containers = [ct for ct in containers if ct.get('vmid') in owned]
logger.info(f"Got {len(vms)} VMs and {len(containers)} containers from {server.name}")
server.update_status(True)
all_resources["servers"].append({
"id": server.id,
"name": server.name,
"ip": server.ip_address,
"vms": vms,
"containers": containers,
"vms_count": len(vms),
"containers_count": len(containers)
})
all_resources["total_vms"] += len(vms)
all_resources["total_containers"] += len(containers)
db.commit() # single commit for every status update, instead of one per server
return JSONResponse(content=all_resources)

61
backend/app/api/proxmox/vms.py
View File

@@ -65,10 +65,12 @@ def _cleanup_installed_apps(db: Session, server_id: int, vmid: int) -> int:
import time as time_lib
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
_live_metrics_cache = {}
_live_metrics_lock = threading.Lock()
METRICS_TTL = 10 # seconds
METRICS_MAX_WORKERS = 8 # parallel Proxmox fetches for the VM list
from ...models import TaskQueue
from ...services.task_queue_service import TaskQueueService, process_task_queue
@@ -199,35 +201,48 @@ def get_all_virtual_machines(
# Map: (server_id_in_db, vmid) -> live data dict
live_metrics: dict[tuple[int, int], dict] = {}
# Serve whatever the cache still covers and collect the servers to refresh.
# The lock guards the dict only — never a network call, otherwise concurrent
# requests queue up behind each other's Proxmox timeouts.
stale_servers = []
now = time_lib.time()
with _live_metrics_lock:
now = time_lib.time()
for server in servers:
cache_key = server.id
if cache_key in _live_metrics_cache and (now - _live_metrics_cache[cache_key]['time']) < METRICS_TTL:
# Use cached metrics
for vmid, res in _live_metrics_cache[cache_key]['data'].items():
cached = _live_metrics_cache.get(server.id)
if cached and (now - cached['time']) < METRICS_TTL:
for vmid, res in cached['data'].items():
live_metrics[(server.id, vmid)] = res
continue
try:
client = _get_proxmox_client(server)
if not client.is_connected():
# An offline server contributes nothing but a connection timeout —
# the cached VM rows already carry its last known status.
if server.is_online is False:
continue
stale_servers.append(server)
def _fetch_live_metrics(server: ProxmoxServer) -> dict[int, dict]:
"""One /cluster/resources call → {vmid: resource}. Runs off the request thread."""
client = _get_proxmox_client(server)
return {
res['vmid']: res
for res in client.get_cluster_resources(type_='vm')
if res.get('vmid') is not None
}
# Refresh in parallel: one unreachable cluster must not serialise the rest.
if stale_servers:
with ThreadPoolExecutor(max_workers=min(METRICS_MAX_WORKERS, len(stale_servers))) as pool:
futures = {pool.submit(_fetch_live_metrics, s): s for s in stale_servers}
for future in as_completed(futures):
server = futures[future]
try:
server_metrics = future.result()
except Exception as e:
logger.debug(f"Could not fetch live metrics from {server.name}: {e}")
continue
server_metrics = {}
for res in client.get_cluster_resources(type_='vm'):
vmid = res.get('vmid')
if vmid is None:
continue
for vmid, res in server_metrics.items():
live_metrics[(server.id, vmid)] = res
server_metrics[vmid] = res
_live_metrics_cache[cache_key] = {
'time': now,
'data': server_metrics
}
except Exception as e:
logger.debug(f"Could not fetch live metrics from {server.name}: {e}")
with _live_metrics_lock:
_live_metrics_cache[server.id] = {'time': time_lib.time(), 'data': server_metrics}
result = []