feat(metrics): add query_instance_metrics read aggregation
This commit is contained in:
@@ -91,3 +91,66 @@ def aggregate_series(rows, from_ts, bucket, fields):
|
|||||||
point[f] = (sum(vals) / len(vals)) if vals else None
|
point[f] = (sum(vals) / len(vals)) if vals else None
|
||||||
out.append(point)
|
out.append(point)
|
||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
from datetime import timezone as _tz
|
||||||
|
|
||||||
|
_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).
|
||||||
|
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})
|
||||||
|
|
||||||
|
# Optional per-NIC override of net rates, keyed by exact timestamp.
|
||||||
|
nic_by_ts: dict[int, tuple] = {}
|
||||||
|
if nic != "all":
|
||||||
|
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}}
|
||||||
|
|||||||
45
backend/tests/test_metrics_query.py
Normal file
45
backend/tests/test_metrics_query.py
Normal file
@@ -0,0 +1,45 @@
|
|||||||
|
# backend/tests/test_metrics_query.py
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
from app.services.metrics_history import query_instance_metrics
|
||||||
|
|
||||||
|
|
||||||
|
def _row(db, InstanceMetric, ts, **kw):
|
||||||
|
db.add(InstanceMetric(server_id=1, vmid=114, vm_type="qemu",
|
||||||
|
ts=datetime.fromtimestamp(ts, tz=timezone.utc), **kw))
|
||||||
|
|
||||||
|
|
||||||
|
def test_query_aggregate_all_nic(db_session):
|
||||||
|
from app.models import InstanceMetric
|
||||||
|
_row(db_session, InstanceMetric, 1000, cpu=0.10, mem=100, maxmem=200,
|
||||||
|
disk_used=10, disk_total=40, netin_rate=1.0, netout_rate=2.0,
|
||||||
|
diskread_rate=0.0, diskwrite_rate=0.0, iops_read=5.0, iops_write=1.0)
|
||||||
|
_row(db_session, InstanceMetric, 1005, cpu=0.30, mem=140, maxmem=200,
|
||||||
|
disk_used=20, disk_total=40, netin_rate=3.0, netout_rate=4.0,
|
||||||
|
diskread_rate=0.0, diskwrite_rate=0.0, iops_read=7.0, iops_write=3.0)
|
||||||
|
db_session.commit()
|
||||||
|
out = query_instance_metrics(db_session, 1, 114, from_ts=1000, to_ts=1010, nic="all")
|
||||||
|
assert out["from"] == 1000 and out["to"] == 1010
|
||||||
|
pt = out["data"][0]
|
||||||
|
assert pt["cpu"] == 20.0 # avg(10%,30%)
|
||||||
|
assert pt["netin"] == 2.0 # avg(1,3)
|
||||||
|
assert pt["diskpct"] == 37.5 # avg(25%,50%)
|
||||||
|
assert pt["iops_read"] == 6.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_query_specific_nic_overrides_net(db_session):
|
||||||
|
from app.models import InstanceMetric, InstanceNicMetric
|
||||||
|
_row(db_session, InstanceMetric, 1000, cpu=0.0, mem=0, maxmem=1,
|
||||||
|
disk_used=0, disk_total=1, netin_rate=99.0, netout_rate=99.0,
|
||||||
|
diskread_rate=0.0, diskwrite_rate=0.0, iops_read=0.0, iops_write=0.0)
|
||||||
|
db_session.add(InstanceNicMetric(server_id=1, vmid=114,
|
||||||
|
ts=datetime.fromtimestamp(1000, tz=timezone.utc), dev="tap114i0",
|
||||||
|
in_rate=7.0, out_rate=8.0))
|
||||||
|
db_session.commit()
|
||||||
|
out = query_instance_metrics(db_session, 1, 114, 1000, 1010, nic="tap114i0")
|
||||||
|
assert out["meta"]["nics"] == ["tap114i0"]
|
||||||
|
assert out["data"][0]["netin"] == 7.0 and out["data"][0]["netout"] == 8.0
|
||||||
|
|
||||||
|
|
||||||
|
def test_query_empty(db_session):
|
||||||
|
out = query_instance_metrics(db_session, 1, 999, 0, 10, nic="all")
|
||||||
|
assert out["data"] == [] and out["meta"]["nics"] == []
|
||||||
Reference in New Issue
Block a user