feat(metrics): add collector + retention loops and wiring

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/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,139 @@
+"""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] = {}
+
+    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)
+
+            # 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)})
+        except Exception as e:
+            logger.debug(f"[metrics_collector] {sid}:{vmid} error: {e}")
+
+
+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/tests/test_metrics_collector.py

@@ -0,0 +1,51 @@
+"""Tests for metrics_collector.collect_once (unit-testable core)."""
+from unittest.mock import MagicMock
+from app.services.metrics_collector import collect_once
+
+
+def _client():
+    c = MagicMock()
+    # status/current
+    c.get_vm_status.return_value = {
+        "cpu": 0.5, "mem": 100, "maxmem": 200,
+        "netin": 1000, "netout": 2000, "diskread": 0, "diskwrite": 500,
+    }
+    c.get_vm_fsinfo.return_value = [{"used": 10, "total": 40}]
+    c.get_vm_blockstats.return_value = "drive-scsi0: rd_operations=100 wr_operations=50"
+    c.get_node_netstat.return_value = [
+        {"dev": "tap114i0", "vmid": "114", "in": "1000", "out": "2000"}]
+    return c
+
+
+def test_collect_once_first_tick_no_rates(db_session):
+    from app.models import InstanceMetric
+    items = [{"server_id": 1, "vmid": 114, "type": "qemu", "node": "prod", "status": "running"}]
+    clients = {1: _client()}
+    prev = {}
+    collect_once(db_session, items, clients, prev, _now=100.0)
+    db_session.commit()
+    m = db_session.query(InstanceMetric).one()
+    assert m.cpu == 0.5 and m.disk_used == 10 and m.disk_total == 40
+    assert m.netin_rate is None          # no previous counter yet
+    assert m.iops_read is None
+
+
+def test_collect_once_second_tick_computes_rates(db_session):
+    from app.models import InstanceMetric, InstanceNicMetric
+    items = [{"server_id": 1, "vmid": 114, "type": "qemu", "node": "prod", "status": "running"}]
+    c = _client()
+    clients = {1: c}
+    prev = {}
+    collect_once(db_session, items, clients, prev, _now=100.0)
+    # advance counters by 10 over 2 seconds
+    c.get_vm_status.return_value = {"cpu": 0.5, "mem": 100, "maxmem": 200,
+                                   "netin": 1020, "netout": 2040, "diskread": 0, "diskwrite": 500}
+    c.get_vm_blockstats.return_value = "drive-scsi0: rd_operations=110 wr_operations=56"
+    c.get_node_netstat.return_value = [{"dev": "tap114i0", "vmid": "114", "in": "1020", "out": "2040"}]
+    collect_once(db_session, items, clients, prev, _now=102.0)
+    db_session.commit()
+    rows = db_session.query(InstanceMetric).order_by(InstanceMetric.id).all()
+    assert rows[1].netin_rate == 10.0     # (1020-1000)/2
+    assert rows[1].iops_read == 5.0       # (110-100)/2
+    nic = db_session.query(InstanceNicMetric).order_by(InstanceNicMetric.id).all()[-1]
+    assert nic.in_rate == 10.0            # (1020-1000)/2