Merge branch 'feature/metrics-history'

backend/app/api/proxmox/vms.py

@@ -21,6 +21,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)
+from ...services.metrics_history import query_instance_metrics, timeframe_to_range
 
 router = APIRouter()
 
@@ -1868,6 +1869,55 @@ def get_container_rrddata(
         raise HTTPException(status_code=500, detail=str(e))
 
 
+# ==================== Metrics History (DB-only) ====================
+
+def _resolve_metrics_window(timeframe, from_ts, to_ts):
+    from ...config import settings
+    now = int(time_lib.time())
+    if from_ts is not None and to_ts is not None:
+        f, t = int(from_ts), int(to_ts)
+    else:
+        f, t = timeframe_to_range(timeframe, now)
+    max_span = getattr(settings, "METRICS_RETENTION_DAYS", 30) * 86400
+    if t - f > max_span:
+        f = t - max_span
+    return f, t
+
+
+@router.get("/api/{server_id}/vm/{vmid}/metrics")
+def get_vm_metrics(
+    server_id: int,
+    vmid: int,
+    node: str = Query(None),
+    timeframe: str = Query("hour", regex="^(hour|day|week|month)$"),
+    from_ts: int = Query(None),
+    to_ts: int = Query(None),
+    nic: str = Query("all"),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(PermissionChecker("vm:view")),
+):
+    """Retrieve VM metrics time-series from the local DB (no Proxmox call)."""
+    f, t = _resolve_metrics_window(timeframe, from_ts, to_ts)
+    return query_instance_metrics(db, server_id, vmid, f, t, nic)
+
+
+@router.get("/api/{server_id}/container/{vmid}/metrics")
+def get_container_metrics(
+    server_id: int,
+    vmid: int,
+    node: str = Query(None),
+    timeframe: str = Query("hour", regex="^(hour|day|week|month)$"),
+    from_ts: int = Query(None),
+    to_ts: int = Query(None),
+    nic: str = Query("all"),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(PermissionChecker("vm:view")),
+):
+    """Retrieve container metrics time-series from the local DB (no Proxmox call)."""
+    f, t = _resolve_metrics_window(timeframe, from_ts, to_ts)
+    return query_instance_metrics(db, server_id, vmid, f, t, nic)
+
+
 # ==================== VNC Console ====================
 
 @router.get("/api/{server_id}/vm/{vmid}/vnc")
@@ -2850,6 +2900,7 @@ def set_vm_owner(
     if not instance:
         raise HTTPException(status_code=404, detail="VM not found in cache")
 
+    target_user = None
     if body.user_id is not None:
         target_user = db.query(User).filter(User.id == body.user_id, User.is_active == True).first()
         if not target_user:
@@ -2858,6 +2909,29 @@ def set_vm_owner(
     instance.owner_id = body.user_id
     db.commit()
     logger.info(f"Admin {current_user.username} set owner of VM {vmid} (server {server_id}) to user_id={body.user_id}")
+
+    # Push a real-time event so every connected client refreshes the owner
+    # column without a manual page reload.
+    try:
+        from ...websocket_manager import broadcast_event
+        owner_user = (
+            {
+                "username": target_user.username,
+                "email": target_user.email,
+                "full_name": target_user.full_name,
+            }
+            if target_user else None
+        )
+        broadcast_event(
+            "vm_owner_changed",
+            server_id=server_id,
+            vmid=vmid,
+            owner_id=body.user_id,
+            owner_user=owner_user,
+        )
+    except Exception as e:
+        logger.debug(f"Failed to broadcast vm_owner_changed: {e}")
+
     return {"ok": True, "owner_id": body.user_id}
 
 

backend/app/config.py

@@ -52,6 +52,10 @@ class Settings(BaseSettings):
     # once your servers' host keys are in known_hosts.
     SSH_REJECT_UNKNOWN_HOSTS: bool = Field(default=False, env="SSH_REJECT_UNKNOWN_HOSTS")
     
+    # Metrics history
+    METRICS_COLLECT_INTERVAL: int = Field(default=15, env="METRICS_COLLECT_INTERVAL")
+    METRICS_RETENTION_DAYS: int = Field(default=30, env="METRICS_RETENTION_DAYS")
+
     # CORS — comma-separated list of allowed origins; "*" is insecure in production
     CORS_ORIGINS: str = Field(default="*", env="CORS_ORIGINS")
 

backend/app/models/__init__.py

@@ -3,6 +3,7 @@ from .auth import Role, user_servers, User, ActiveSession, LoginAttempt, Blocked
 from .proxmox import OSTemplateGroup, OSTemplate, ProxmoxServer, VMInstance, VMSnapshotArchive
 from .ipam import IPAMNetwork, IPAMPool, IPAMAllocation, IPAMHistory
 from .tasks import TaskQueue, ProxmoxTask, DeployTask
+from .metrics import InstanceMetric, InstanceNicMetric
 from .settings import PanelSettings, SecuritySetting, Workspace, WorkspaceServer, WorkspaceUser
 from .misc import Notification, NotificationPreference, BackupJob, AuditLog
 
@@ -12,6 +13,7 @@ __all__ = [
     'OSTemplateGroup', 'OSTemplate', 'ProxmoxServer', 'VMInstance', 'VMSnapshotArchive',
     'IPAMNetwork', 'IPAMPool', 'IPAMAllocation', 'IPAMHistory',
     'TaskQueue', 'ProxmoxTask', 'DeployTask',
+    'InstanceMetric', 'InstanceNicMetric',
     'PanelSettings', 'SecuritySetting', 'Workspace', 'WorkspaceServer', 'WorkspaceUser',
     'Notification', 'NotificationPreference', 'BackupJob', 'AuditLog'
 ]

backend/app/models/metrics.py

@@ -0,0 +1,39 @@
+from sqlalchemy import Float
+from .base import Base, Column, Integer, BigInteger, String, DateTime, Index, func
+
+
+class InstanceMetric(Base):
+    __tablename__ = "instance_metric"
+
+    id = Column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
+    server_id = Column(Integer, nullable=False)
+    vmid = Column(Integer, nullable=False)
+    vm_type = Column(String(10), nullable=True)
+    ts = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    cpu = Column(Float, nullable=True)
+    mem = Column(BigInteger, nullable=True)
+    maxmem = Column(BigInteger, nullable=True)
+    disk_used = Column(BigInteger, nullable=True)
+    disk_total = Column(BigInteger, nullable=True)
+    netin_rate = Column(Float, nullable=True)
+    netout_rate = Column(Float, nullable=True)
+    diskread_rate = Column(Float, nullable=True)
+    diskwrite_rate = Column(Float, nullable=True)
+    iops_read = Column(Float, nullable=True)
+    iops_write = Column(Float, nullable=True)
+
+    __table_args__ = (Index("ix_instance_metric_lookup", "server_id", "vmid", "ts"),)
+
+
+class InstanceNicMetric(Base):
+    __tablename__ = "instance_nic_metric"
+
+    id = Column(BigInteger().with_variant(Integer, "sqlite"), primary_key=True, autoincrement=True)
+    server_id = Column(Integer, nullable=False)
+    vmid = Column(Integer, nullable=False)
+    ts = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    dev = Column(String(64), nullable=False)
+    in_rate = Column(Float, nullable=True)
+    out_rate = Column(Float, nullable=True)
+
+    __table_args__ = (Index("ix_instance_nic_metric_lookup", "server_id", "vmid", "ts"),)

backend/app/proxmox/client.py

@@ -1310,6 +1310,29 @@ class ProxmoxClient(VmMixin, LxcMixin, ClusterMixin, StorageMixin, NetworkMixin,
                 logger.debug(f"Не удалось получить сетевые интерфейсы VM {vmid} (guest agent может быть не установлен): {e}")
                 return []
 
+        def get_vm_blockstats(self, node: str, vmid: int) -> str:
+            """Raw HMP 'info blockstats' text via the QEMU monitor (VM only).
+            Returns '' if unavailable."""
+            if not self.proxmox:
+                return ""
+            try:
+                res = self.proxmox.nodes(node).qemu(vmid).monitor.post(command="info blockstats")
+                return res if isinstance(res, str) else str(res or "")
+            except Exception as e:
+                logger.debug(f"blockstats unavailable for VM {vmid} on {node}: {e}")
+                return ""
+
+        def get_node_netstat(self, node: str) -> list:
+            """Per-tap/veth interface counters for the node ([] on failure)."""
+            if not self.proxmox:
+                return []
+            try:
+                res = self.proxmox.nodes(node).netstat.get()
+                return res if isinstance(res, list) else []
+            except Exception as e:
+                logger.debug(f"netstat unavailable for node {node}: {e}")
+                return []
+
         def get_container_interfaces(self, node: str, vmid: int) -> List[Dict]:
             """
             Получить сетевые интерфейсы контейнера

backend/app/services/metrics_broadcaster.py

@@ -606,4 +606,8 @@ def start_metrics_broadcaster(db_factory) -> None:
     asyncio.create_task(server_metrics_loop(db_factory), name="metrics:servers")
     asyncio.create_task(instance_metrics_loop(db_factory), name="metrics:instances")
     asyncio.create_task(instances_list_metrics_loop(db_factory), name="metrics:instances_list")
+    # Import inside function to avoid circular imports
+    from .metrics_collector import metrics_collector_loop, metrics_retention_loop
+    asyncio.create_task(metrics_collector_loop(db_factory), name="metrics:collector")
+    asyncio.create_task(metrics_retention_loop(db_factory), name="metrics:retention")
     logger.info("[metrics_broadcaster] Metrics broadcasting tasks started")

backend/app/services/metrics_collector.py

@@ -0,0 +1,148 @@
+"""Background collector that persists instance telemetry to our own tables."""
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import datetime, timezone
+
+from loguru import logger
+
+from .metrics_history import (
+    compute_rate, parse_blockstats, parse_netstat, aggregate_fsinfo,
+)
+
+
+def _status(client, inst_type, node, vmid):
+    if inst_type == "qemu":
+        return client.get_vm_status(node, vmid)
+    return client.get_container_status(node, vmid)
+
+
+def collect_once(db, items, clients, prev, _now=None):
+    """One collection tick. Mutates `prev` to carry counters between ticks."""
+    from ..models import InstanceMetric, InstanceNicMetric
+    now = _now if _now is not None else time.monotonic()
+    ts = datetime.now(tz=timezone.utc)
+
+    # netstat is one call per (server_id, node); cache per tick.
+    netstat_cache: dict[tuple, list] = {}
+    # Track which prev keys are touched this tick so stale entries can be evicted.
+    seen: set = set()
+
+    for it in items:
+        if it.get("status") != "running":
+            continue
+        sid, vmid = it["server_id"], it["vmid"]
+        inst_type, node = it.get("type"), it.get("node")
+        client = clients.get(sid)
+        if client is None or not node:
+            continue
+        try:
+            status = _status(client, inst_type, node, vmid)
+            if not status:
+                continue
+
+            # disk fill
+            if inst_type == "qemu":
+                used, total = aggregate_fsinfo(client.get_vm_fsinfo(node, vmid))
+            else:
+                used = int(status.get("disk", 0) or 0)
+                total = int(status.get("maxdisk", 0) or 0)
+
+            # cumulative counters → rates
+            key = f"{sid}:{vmid}"
+            cur = {
+                "netin": float(status.get("netin", 0) or 0),
+                "netout": float(status.get("netout", 0) or 0),
+                "diskread": float(status.get("diskread", 0) or 0),
+                "diskwrite": float(status.get("diskwrite", 0) or 0),
+            }
+            if inst_type == "qemu":
+                rd_ops, wr_ops = parse_blockstats(client.get_vm_blockstats(node, vmid))
+                cur["rd_ops"], cur["wr_ops"] = float(rd_ops), float(wr_ops)
+
+            pkey = prev.get(key)
+            dt = (now - pkey[0]) if pkey else 0
+            pc = pkey[1] if pkey else {}
+
+            def rate(name):
+                return compute_rate(pc.get(name), cur.get(name), dt)
+
+            db.add(InstanceMetric(
+                server_id=sid, vmid=vmid, vm_type=inst_type, ts=ts,
+                cpu=status.get("cpu"), mem=status.get("mem"), maxmem=status.get("maxmem"),
+                disk_used=used, disk_total=total,
+                netin_rate=rate("netin"), netout_rate=rate("netout"),
+                diskread_rate=rate("diskread"), diskwrite_rate=rate("diskwrite"),
+                iops_read=rate("rd_ops") if inst_type == "qemu" else None,
+                iops_write=rate("wr_ops") if inst_type == "qemu" else None,
+            ))
+            prev[key] = (now, cur)
+            seen.add(key)
+
+            # per-NIC
+            ck = (sid, node)
+            if ck not in netstat_cache:
+                netstat_cache[ck] = client.get_node_netstat(node)
+            per_dev = parse_netstat(netstat_cache[ck], vmid)
+            for dev, (din, dout) in per_dev.items():
+                nkey = f"{sid}:{vmid}:{dev}"
+                npkey = prev.get(nkey)
+                ndt = (now - npkey[0]) if npkey else 0
+                npc = npkey[1] if npkey else {}
+                db.add(InstanceNicMetric(
+                    server_id=sid, vmid=vmid, ts=ts, dev=dev,
+                    in_rate=compute_rate(npc.get("in"), float(din), ndt),
+                    out_rate=compute_rate(npc.get("out"), float(dout), ndt),
+                ))
+                prev[nkey] = (now, {"in": float(din), "out": float(dout)})
+                seen.add(nkey)
+        except Exception as e:
+            logger.debug(f"[metrics_collector] {sid}:{vmid} error: {e}")
+
+    # Evict stale keys from prev (VMs/NICs not seen this tick).
+    for k in list(prev):
+        if k not in seen:
+            del prev[k]
+
+
+async def metrics_collector_loop(db_factory) -> None:
+    from ..config import settings
+    from .metrics_broadcaster import _enumerate_instances_for_list
+    prev: dict = {}
+    while True:
+        start = time.monotonic()
+        try:
+            items, clients = await asyncio.to_thread(_enumerate_instances_for_list, db_factory)
+            db = db_factory()
+            try:
+                await asyncio.to_thread(collect_once, db, items, clients, prev)
+                db.commit()
+            finally:
+                db.close()
+        except asyncio.CancelledError:
+            break
+        except Exception as e:
+            logger.warning(f"[metrics_collector] tick error: {e}")
+        await asyncio.sleep(max(1.0, settings.METRICS_COLLECT_INTERVAL - (time.monotonic() - start)))
+
+
+async def metrics_retention_loop(db_factory) -> None:
+    from ..config import settings
+    from ..models import InstanceMetric, InstanceNicMetric
+    from datetime import timedelta
+    while True:
+        try:
+            cutoff = datetime.now(tz=timezone.utc) - timedelta(days=settings.METRICS_RETENTION_DAYS)
+            db = db_factory()
+            try:
+                db.query(InstanceMetric).filter(InstanceMetric.ts < cutoff).delete()
+                db.query(InstanceNicMetric).filter(InstanceNicMetric.ts < cutoff).delete()
+                db.commit()
+            finally:
+                db.close()
+        except asyncio.CancelledError:
+            break
+        except Exception as e:
+            logger.warning(f"[metrics_collector] retention error: {e}")
+        await asyncio.sleep(3600)

backend/app/services/metrics_history.py

@@ -0,0 +1,163 @@
+"""Pure helpers + read query for instance metric history."""
+from __future__ import annotations
+
+import re
+from datetime import timezone as _tz
+
+_RD_OPS = re.compile(r"rd_operations=(\d+)")
+_WR_OPS = re.compile(r"wr_operations=(\d+)")
+
+
+def compute_rate(prev, cur, dt):
+    """Per-second rate from cumulative counters. None if unknowable; 0 on reset."""
+    if prev is None or cur is None or not dt or dt <= 0:
+        return None
+    delta = float(cur) - float(prev)
+    if delta < 0:
+        return 0.0
+    return delta / dt
+
+
+def parse_blockstats(text: str):
+    """Sum rd_operations / wr_operations across all block devices in HMP output."""
+    rd = sum(int(m) for m in _RD_OPS.findall(text or ""))
+    wr = sum(int(m) for m in _WR_OPS.findall(text or ""))
+    return rd, wr
+
+
+def parse_netstat(rows, vmid: int):
+    """Map dev -> (in_bytes, out_bytes) for the given vmid from /nodes/{node}/netstat."""
+    out: dict[str, tuple[int, int]] = {}
+    for r in rows or []:
+        try:
+            if int(r.get("vmid")) != int(vmid):
+                continue
+        except (TypeError, ValueError):
+            continue
+        dev = r.get("dev")
+        if not dev:
+            continue
+        out[dev] = (int(float(r.get("in", 0) or 0)), int(float(r.get("out", 0) or 0)))
+    return out
+
+
+def aggregate_fsinfo(disks):
+    """Sum used/total bytes across guest filesystems."""
+    used = sum(int(d.get("used", 0) or 0) for d in disks or [])
+    total = sum(int(d.get("total", 0) or 0) for d in disks or [])
+    return used, total
+
+
+_TIMEFRAME_SECONDS = {
+    "hour": 3600,
+    "day": 86400,
+    "week": 7 * 86400,
+    "month": 30 * 86400,
+}
+
+
+def timeframe_to_range(timeframe, now_ts):
+    """Convert timeframe string to (from_ts, to_ts) tuple."""
+    span = _TIMEFRAME_SECONDS.get(timeframe, 3600)
+    return now_ts - span, now_ts
+
+
+def pick_bucket_seconds(from_ts, to_ts, target_points=150, floor=15):
+    """Calculate bucket size in seconds for time-series aggregation."""
+    span = max(1, to_ts - from_ts)
+    ideal = span // max(1, target_points)
+    return max(floor, int(ideal))
+
+
+def aggregate_series(rows, from_ts, bucket, fields):
+    """Aggregate time-series data into buckets, averaging numeric fields."""
+    if not rows:
+        return []
+    buckets: dict[int, dict] = {}
+    order: list[int] = []
+    for r in rows:
+        b = from_ts + ((int(r["time"]) - from_ts) // bucket) * bucket
+        if b not in buckets:
+            buckets[b] = {f: [] for f in fields}
+            order.append(b)
+        for f in fields:
+            v = r.get(f)
+            if v is not None:
+                buckets[b][f].append(float(v))
+    out = []
+    for b in order:
+        point = {"time": b}
+        for f in fields:
+            vals = buckets[b][f]
+            point[f] = (sum(vals) / len(vals)) if vals else None
+        out.append(point)
+    return out
+
+
+_AGG_FIELDS = ["cpu", "mem", "maxmem", "netin", "netout",
+               "diskread", "diskwrite", "diskpct", "iops_read", "iops_write"]
+
+
+def _to_ts(dt):
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=_tz.utc)
+    return int(dt.timestamp())
+
+
+def query_instance_metrics(db, server_id, vmid, from_ts, to_ts, nic="all"):
+    from ..models import InstanceMetric, InstanceNicMetric
+    from datetime import datetime
+
+    start = datetime.fromtimestamp(from_ts, tz=_tz.utc)
+    end = datetime.fromtimestamp(to_ts, tz=_tz.utc)
+
+    base = (db.query(InstanceMetric)
+              .filter(InstanceMetric.server_id == server_id,
+                      InstanceMetric.vmid == vmid,
+                      InstanceMetric.ts >= start, InstanceMetric.ts <= end)
+              .order_by(InstanceMetric.ts.asc()).all())
+
+    # Distinct NICs seen in the window (for the selector).
+    if nic == "all":
+        # Only fetch distinct dev names — avoids loading all NIC rows.
+        dev_rows = (db.query(InstanceNicMetric.dev)
+                      .filter(InstanceNicMetric.server_id == server_id,
+                              InstanceNicMetric.vmid == vmid,
+                              InstanceNicMetric.ts >= start, InstanceNicMetric.ts <= end)
+                      .distinct().all())
+        nics = sorted({r.dev for r in dev_rows})
+        nic_by_ts: dict[int, tuple] = {}
+    else:
+        # Load full NIC rows for the requested dev to build per-ts override map.
+        nic_rows = (db.query(InstanceNicMetric)
+                      .filter(InstanceNicMetric.server_id == server_id,
+                              InstanceNicMetric.vmid == vmid,
+                              InstanceNicMetric.ts >= start, InstanceNicMetric.ts <= end)
+                      .order_by(InstanceNicMetric.ts.asc()).all())
+        nics = sorted({r.dev for r in nic_rows})
+        nic_by_ts = {}
+        for r in nic_rows:
+            if r.dev == nic:
+                nic_by_ts[_to_ts(r.ts)] = (r.in_rate, r.out_rate)
+
+    rows = []
+    for m in base:
+        ts = _to_ts(m.ts)
+        netin, netout = m.netin_rate, m.netout_rate
+        if nic != "all":
+            ov = nic_by_ts.get(ts)
+            netin, netout = (ov if ov else (None, None))
+        diskpct = (m.disk_used / m.disk_total * 100.0) if (m.disk_total or 0) > 0 else None
+        rows.append({
+            "time": ts,
+            "cpu": (m.cpu * 100.0) if m.cpu is not None else None,
+            "mem": m.mem, "maxmem": m.maxmem,
+            "netin": netin, "netout": netout,
+            "diskread": m.diskread_rate, "diskwrite": m.diskwrite_rate,
+            "diskpct": diskpct,
+            "iops_read": m.iops_read, "iops_write": m.iops_write,
+        })
+
+    bucket = pick_bucket_seconds(from_ts, to_ts)
+    data = aggregate_series(rows, from_ts, bucket, _AGG_FIELDS)
+    return {"data": data, "from": from_ts, "to": to_ts, "meta": {"nics": nics}}