feat(metrics): add blockstats + netstat client wrappers

This commit is contained in:
2026-06-21 21:04:13 +05:00
parent 3d55b1ab8d
commit 8f9f78ef91
2 changed files with 60 additions and 0 deletions

View File

@@ -1309,6 +1309,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]:
"""
Получить сетевые интерфейсы контейнера

View File

@@ -0,0 +1,37 @@
from unittest.mock import MagicMock
from app.proxmox.client import ProxmoxClient
def _client_with_mock(proxmox):
c = ProxmoxClient.__new__(ProxmoxClient)
c.proxmox = proxmox
return c
def test_get_vm_blockstats_returns_text():
px = MagicMock()
px.nodes.return_value.qemu.return_value.monitor.post.return_value = "drive-scsi0: rd_operations=5 wr_operations=2"
c = _client_with_mock(px)
assert "rd_operations=5" in c.get_vm_blockstats("prod", 114)
px.nodes.return_value.qemu.return_value.monitor.post.assert_called_once_with(command="info blockstats")
def test_get_vm_blockstats_swallows_errors():
px = MagicMock()
px.nodes.return_value.qemu.return_value.monitor.post.side_effect = RuntimeError("nope")
c = _client_with_mock(px)
assert c.get_vm_blockstats("prod", 114) == ""
def test_get_node_netstat_returns_rows():
px = MagicMock()
px.nodes.return_value.netstat.get.return_value = [{"dev": "tap114i0", "vmid": "114", "in": "1", "out": "2"}]
c = _client_with_mock(px)
assert c.get_node_netstat("prod")[0]["dev"] == "tap114i0"
def test_get_node_netstat_swallows_errors():
px = MagicMock()
px.nodes.return_value.netstat.get.side_effect = RuntimeError("nope")
c = _client_with_mock(px)
assert c.get_node_netstat("prod") == []