hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 05:30:46 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

Move port traffic to device-centric monitoring with a dedicated wall and collect logs.

Browse source 
Split the ops wall to its own route with fullscreen charts, persist collect errors as device events, and polish wall legend layout.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-08-01 15:45:20 +08:00
parent 5079343b2a
commit 214bc56774
24 changed files with 3309 additions and 1357 deletions
Download patch file Download diff file Expand all files Collapse all files

65
netx_api/models.py
View file

@ -595,17 +595,25 @@ class NeConfigHistory(Base):
task_id: Mapped[str] = mapped_column(String(64), default="")
class PortTrafficTask(Base):
"""Port traffic monitoring job definition."""
class PortTrafficDevice(Base):
"""Per-NE port traffic monitoring config (device-centric)."""
__tablename__ = "port_traffic_task"
__tablename__ = "port_traffic_device"
__table_args__ = (
UniqueConstraint("source", "ne_id", name="uq_port_traffic_device_ne"),
)
id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex)
title: Mapped[str] = mapped_column(String(256), default="")
source: Mapped[str] = mapped_column(String(32), default="managed", index=True)
ne_id: Mapped[str] = mapped_column(String(128), default="", index=True)
ne_name: Mapped[str] = mapped_column(String(256), default="")
ne_ip: Mapped[str] = mapped_column(String(128), default="")
vendor: Mapped[str] = mapped_column(String(64), default="")
note: Mapped[str] = mapped_column(String(256), default="") # optional remark
status: Mapped[str] = mapped_column(String(32), default="draft", index=True) # draft|running|paused|stopped
interval_sec: Mapped[int] = mapped_column(Integer, default=60)
retention_days: Mapped[int] = mapped_column(Integer, default=7)
concurrency: Mapped[int] = mapped_column(Integer, default=5)
concurrency: Mapped[int] = mapped_column(Integer, default=1)
collect_running: Mapped[bool] = mapped_column(Boolean, default=False)
last_collect_started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
last_collect_ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
@ -614,32 +622,41 @@ class PortTrafficTask(Base):
updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
# Back-compat alias while callers migrate.
PortTrafficTask = PortTrafficDevice
class PortTrafficSeries(Base):
"""Logical port (business link) that survives physical NE/if replacement."""
__tablename__ = "port_traffic_series"
__table_args__ = (UniqueConstraint("task_id", "title", name="uq_port_traffic_series_title"),)
__table_args__ = (UniqueConstraint("device_id", "title", name="uq_port_traffic_series_title"),)
id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex)
task_id: Mapped[str] = mapped_column(String(64), index=True)
device_id: Mapped[str] = mapped_column(String(64), default="", index=True)
title: Mapped[str] = mapped_column(String(256), default="")
status: Mapped[str] = mapped_column(String(32), default="active", index=True) # active|disabled
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
@property
def task_id(self) -> str:
return str(self.device_id or "")
@task_id.setter
def task_id(self, value: str) -> None:
self.device_id = str(value or "")
class PortTrafficTarget(Base):
"""Monitored interface under a port traffic task."""
"""Monitored interface under a device monitoring config."""
__tablename__ = "port_traffic_target"
__table_args__ = (
UniqueConstraint("task_id", "source", "target_id", "ifname", name="uq_port_traffic_target_if"),
)
id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex)
task_id: Mapped[str] = mapped_column(String(64), index=True)
device_id: Mapped[str] = mapped_column(String(64), default="", index=True)
series_id: Mapped[str] = mapped_column(String(64), default="", index=True)
source: Mapped[str] = mapped_column(String(32), default="managed", index=True)
target_id: Mapped[str] = mapped_column(String(128), index=True)
target_id: Mapped[str] = mapped_column(String(128), index=True) # NE id
ne_name: Mapped[str] = mapped_column(String(256), default="")
ne_ip: Mapped[str] = mapped_column(String(128), default="")
vendor: Mapped[str] = mapped_column(String(64), default="")
@ -651,6 +668,14 @@ class PortTrafficTarget(Base):
last_sample_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
@property
def task_id(self) -> str:
return str(self.device_id or "")
@task_id.setter
def task_id(self, value: str) -> None:
self.device_id = str(value or "")
class PortTrafficSample(Base):
"""Time-series sample for a monitored interface."""
@ -670,3 +695,17 @@ class PortTrafficSample(Base):
rate_period_sec: Mapped[int] = mapped_column(Integer, default=0)
raw_ok: Mapped[bool] = mapped_column(Boolean, default=True)
message: Mapped[str] = mapped_column(String(512), default="")
class PortTrafficEvent(Base):
"""Collect / ops log line for a monitored device (and optional interface)."""
__tablename__ = "port_traffic_event"
id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex)
device_id: Mapped[str] = mapped_column(String(64), default="", index=True)
target_row_id: Mapped[str] = mapped_column(String(64), default="", index=True)
ifname: Mapped[str] = mapped_column(String(128), default="")
level: Mapped[str] = mapped_column(String(16), default="error", index=True) # info|warn|error
message: Mapped[str] = mapped_column(Text, default="")
created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True)

256
netx_api/port_traffic_migrate.py
View file

@ -1,4 +1,4 @@
"""Startup migration / backfill for port traffic logical series."""
"""Startup migration / backfill for port traffic device-centric model."""
from __future__ import annotations
@ -8,49 +8,274 @@ from uuid import uuid4
from sqlalchemy import text
from sqlalchemy.orm import Session
from .models import PortTrafficSample, PortTrafficSeries, PortTrafficTarget
from .models import PortTrafficDevice, PortTrafficSample, PortTrafficSeries, PortTrafficTarget
_log = logging.getLogger("netx.port_traffic.migrate")
def ensure_port_traffic_series_schema(conn) -> None:
"""DDL for series + series_id columns (Postgres IF NOT EXISTS)."""
"""DDL for device + series + device_id columns (Postgres IF NOT EXISTS)."""
conn.exec_driver_sql(
"""
CREATE TABLE IF NOT EXISTS port_traffic_device (
id VARCHAR(64) PRIMARY KEY,
source VARCHAR(32) DEFAULT 'managed',
ne_id VARCHAR(128) DEFAULT '',
ne_name VARCHAR(256) DEFAULT '',
ne_ip VARCHAR(128) DEFAULT '',
vendor VARCHAR(64) DEFAULT '',
note VARCHAR(256) DEFAULT '',
status VARCHAR(32) DEFAULT 'draft',
interval_sec INTEGER DEFAULT 60,
retention_days INTEGER DEFAULT 7,
concurrency INTEGER DEFAULT 1,
collect_running BOOLEAN DEFAULT FALSE,
last_collect_started_at TIMESTAMP,
last_collect_ended_at TIMESTAMP,
last_error VARCHAR(1024) DEFAULT '',
created_at TIMESTAMP,
updated_at TIMESTAMP
)
"""
)
conn.exec_driver_sql(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_port_traffic_device_ne ON port_traffic_device (source, ne_id)"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_device_status ON port_traffic_device (status)"
)
conn.exec_driver_sql(
"""
CREATE TABLE IF NOT EXISTS port_traffic_series (
id VARCHAR(64) PRIMARY KEY,
task_id VARCHAR(64),
device_id VARCHAR(64) DEFAULT '',
title VARCHAR(256) DEFAULT '',
status VARCHAR(32) DEFAULT 'active',
created_at TIMESTAMP
)
"""
)
# Legacy column may still exist as task_id
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_series_task_id ON port_traffic_series (task_id)"
"ALTER TABLE port_traffic_series ADD COLUMN IF NOT EXISTS device_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"ALTER TABLE port_traffic_series ADD COLUMN IF NOT EXISTS task_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_series_device_id ON port_traffic_series (device_id)"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_series_status ON port_traffic_series (status)"
)
conn.exec_driver_sql(
"CREATE UNIQUE INDEX IF NOT EXISTS uq_port_traffic_series_title ON port_traffic_series (task_id, title)"
)
conn.exec_driver_sql(
"ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS device_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"ALTER TABLE port_traffic_target ADD COLUMN IF NOT EXISTS task_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_target_series_id ON port_traffic_target (series_id)"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_target_device_id ON port_traffic_target (device_id)"
)
conn.exec_driver_sql(
"ALTER TABLE port_traffic_sample ADD COLUMN IF NOT EXISTS series_id VARCHAR(64) DEFAULT ''"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_sample_series_id ON port_traffic_sample (series_id)"
)
# One active interface globally per physical port
conn.exec_driver_sql(
"""
CREATE UNIQUE INDEX IF NOT EXISTS uq_port_traffic_target_active_if
ON port_traffic_target (source, target_id, ifname)
WHERE status = 'active'
"""
)
conn.exec_driver_sql(
"""
CREATE TABLE IF NOT EXISTS port_traffic_event (
id VARCHAR(64) PRIMARY KEY,
device_id VARCHAR(64) DEFAULT '',
target_row_id VARCHAR(64) DEFAULT '',
ifname VARCHAR(128) DEFAULT '',
level VARCHAR(16) DEFAULT 'error',
message TEXT DEFAULT '',
created_at TIMESTAMP
)
"""
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_event_device_id ON port_traffic_event (device_id)"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_event_created_at ON port_traffic_event (created_at)"
)
conn.exec_driver_sql(
"CREATE INDEX IF NOT EXISTS ix_port_traffic_event_level ON port_traffic_event (level)"
)
def migrate_tasks_to_devices(db: Session) -> int:
"""Collapse legacy port_traffic_task rows into per-NE devices; reassign targets/series."""
# Copy task_id → device_id where missing
try:
db.execute(
text(
"""
UPDATE port_traffic_target
SET device_id = task_id
WHERE COALESCE(device_id, '') = '' AND COALESCE(task_id, '') <> ''
"""
)
)
db.execute(
text(
"""
UPDATE port_traffic_series
SET device_id = task_id
WHERE COALESCE(device_id, '') = '' AND COALESCE(task_id, '') <> ''
"""
)
)
db.commit()
except Exception:
db.rollback()
# If legacy task table missing, nothing else to do
try:
db.execute(text("SELECT 1 FROM port_traffic_task LIMIT 1"))
except Exception:
db.rollback()
return 0
targets = db.query(PortTrafficTarget).all()
if not targets:
# Still create devices from empty? skip
return 0
# Group active+any targets by (source, ne_id)
groups: dict[tuple[str, str], list[PortTrafficTarget]] = {}
for t in targets:
src = str(t.source or "managed").strip().lower() or "managed"
ne = str(t.target_id or "").strip()
if not ne:
continue
groups.setdefault((src, ne), []).append(t)
created = 0
# Load legacy tasks for interval/status
legacy: dict[str, dict] = {}
try:
rows = db.execute(
text(
"""
SELECT id, title, status, interval_sec, retention_days, concurrency,
collect_running, last_collect_started_at, last_collect_ended_at,
last_error, created_at, updated_at
FROM port_traffic_task
"""
)
).mappings().all()
for r in rows:
legacy[str(r["id"])] = dict(r)
except Exception:
db.rollback()
legacy = {}
for (src, ne), members in groups.items():
existing = (
db.query(PortTrafficDevice)
.filter(PortTrafficDevice.source == src, PortTrafficDevice.ne_id == ne)
.first()
)
if existing:
device = existing
else:
# Pick policy from the richest legacy task among members
intervals: list[int] = []
retentions: list[int] = []
statuses: list[str] = []
note = ""
created_at = None
for m in members:
tid = str(getattr(m, "device_id", None) or getattr(m, "task_id", None) or "")
meta = legacy.get(tid) or {}
if meta.get("interval_sec"):
intervals.append(int(meta["interval_sec"]))
if meta.get("retention_days"):
retentions.append(int(meta["retention_days"]))
if meta.get("status"):
statuses.append(str(meta["status"]))
if not note and meta.get("title"):
note = str(meta["title"])[:256]
if created_at is None and meta.get("created_at"):
created_at = meta["created_at"]
status = "running" if "running" in statuses else (statuses[0] if statuses else "stopped")
sample = members[0]
device = PortTrafficDevice(
id=uuid4().hex,
source=src,
ne_id=ne,
ne_name=str(sample.ne_name or ""),
ne_ip=str(sample.ne_ip or ""),
vendor=str(sample.vendor or ""),
note=note,
status=status if status in ("running", "paused", "stopped", "draft") else "stopped",
interval_sec=min(intervals) if intervals else 60,
retention_days=max(retentions) if retentions else 7,
concurrency=1,
collect_running=False,
created_at=created_at,
)
db.add(device)
db.flush()
created += 1
# Deduplicate active ifnames: keep earliest active, retire the rest
seen_active: set[str] = set()
for m in sorted(members, key=lambda x: str(x.created_at or "")):
m.device_id = str(device.id)
if str(m.status) == "active":
key = str(m.ifname or "").strip()
if key in seen_active:
m.status = "retired"
else:
seen_active.add(key)
# Keep ne metadata in sync
if not device.ne_name and m.ne_name:
device.ne_name = str(m.ne_name)
if not device.ne_ip and m.ne_ip:
device.ne_ip = str(m.ne_ip)
if not device.vendor and m.vendor:
device.vendor = str(m.vendor)
# Re-point series
series_ids = {str(m.series_id) for m in members if m.series_id}
if series_ids:
db.query(PortTrafficSeries).filter(PortTrafficSeries.id.in_(list(series_ids))).update(
{PortTrafficSeries.device_id: str(device.id)},
synchronize_session=False,
)
db.commit()
if created:
_log.info("port_traffic migrated devices created=%s groups=%s", created, len(groups))
return created
def backfill_port_traffic_series(db: Session) -> int:
"""Create one series per target missing series_id; stamp samples."""
migrate_tasks_to_devices(db)
targets = (
db.query(PortTrafficTarget)
.filter((PortTrafficTarget.series_id == None) | (PortTrafficTarget.series_id == "")) # noqa: E711
@ -58,13 +283,14 @@ def backfill_port_traffic_series(db: Session) -> int:
)
created = 0
for t in targets:
device_id = str(t.device_id or "")
title = _default_series_title(t.ne_name or "", t.ifname or "")
title = _unique_series_title(db, str(t.task_id), title)
title = _unique_series_title(db, device_id, title)
sid = uuid4().hex
db.add(
PortTrafficSeries(
id=sid,
task_id=str(t.task_id),
device_id=device_id,
title=title,
status="active",
)
@ -103,8 +329,8 @@ def default_series_title(ne_name: str, ifname: str) -> str:
return _default_series_title(ne_name, ifname)
def unique_series_title(db: Session, task_id: str, base: str) -> str:
return _unique_series_title(db, task_id, base)
def unique_series_title(db: Session, device_id: str, base: str) -> str:
return _unique_series_title(db, device_id, base)
def _default_series_title(ne_name: str, ifname: str) -> str:
@ -113,11 +339,11 @@ def _default_series_title(ne_name: str, ifname: str) -> str:
return f"{ne}:{iface}"[:256]
def _unique_series_title(db: Session, task_id: str, base: str) -> str:
def _unique_series_title(db: Session, device_id: str, base: str) -> str:
title = base[:256]
exists = (
db.query(PortTrafficSeries.id)
.filter(PortTrafficSeries.task_id == task_id, PortTrafficSeries.title == title)
.filter(PortTrafficSeries.device_id == device_id, PortTrafficSeries.title == title)
.first()
)
if not exists:
@ -126,7 +352,7 @@ def _unique_series_title(db: Session, task_id: str, base: str) -> str:
candidate = f"{base[:240]}#{i}"
exists = (
db.query(PortTrafficSeries.id)
.filter(PortTrafficSeries.task_id == task_id, PortTrafficSeries.title == candidate)
.filter(PortTrafficSeries.device_id == device_id, PortTrafficSeries.title == candidate)
.first()
)
if not exists:

12
netx_api/port_traffic_parsers.py
View file

@ -116,6 +116,18 @@ class DetailRates:
out_util_pct: float = 0.0
def resolve_util_pct(vendor_util: float, bps: float, bw_bps: int) -> float:
"""Prefer vendor-reported util; derive from rate/BW when util is missing but traffic exists."""
u = float(vendor_util or 0.0)
if u > 0:
return u
bw = int(bw_bps or 0)
rate = float(bps or 0.0)
if bw > 0 and rate > 0:
return rate / float(bw) * 100.0
return u
def parse_bw_to_bps(raw: str) -> int:
text = (raw or "").strip()
if not text or text.upper() == "N/A":

20
netx_api/port_traffic_recovery.py
View file

@ -7,7 +7,7 @@ from datetime import datetime
from sqlalchemy.orm import Session
from .models import PortTrafficTask
from .models import PortTrafficDevice
_log = logging.getLogger("netx.port_traffic.recovery")
@ -15,17 +15,17 @@ _log = logging.getLogger("netx.port_traffic.recovery")
def recover_port_traffic_on_startup(db: Session) -> int:
"""Clear stuck collect_running flags so scheduler can resume."""
now = datetime.utcnow()
rows = db.query(PortTrafficTask).filter(PortTrafficTask.collect_running.is_(True)).all()
rows = db.query(PortTrafficDevice).filter(PortTrafficDevice.collect_running.is_(True)).all()
n = 0
for task in rows:
task.collect_running = False
if not task.last_collect_ended_at:
task.last_collect_ended_at = now
if not task.last_error:
task.last_error = "requeued_after_restart"
task.updated_at = now
for device in rows:
device.collect_running = False
if not device.last_collect_ended_at:
device.last_collect_ended_at = now
if not device.last_error:
device.last_error = "requeued_after_restart"
device.updated_at = now
n += 1
if n:
db.commit()
_log.info("port_traffic recovery cleared collect_running on %s task(s)", n)
_log.info("port_traffic recovery cleared collect_running on %s device(s)", n)
return n

219
netx_api/port_traffic_router.py
View file

@ -1,4 +1,4 @@
"""HTTP API for port traffic monitoring."""
"""HTTP API for port traffic monitoring (device-centric)."""
from __future__ import annotations
@ -9,28 +9,30 @@ from sqlalchemy.orm import Session
from .auth_service import write_audit
from .db import get_db
from .models import PortTrafficDevice
from .port_traffic_schemas import (
DiscoverPortsRequest,
PortTrafficDeviceCreate,
PortTrafficDeviceUpdate,
PortTrafficInterfacesPut,
PortTrafficReplacePortRequest,
PortTrafficTaskCreate,
PortTrafficTaskUpdate,
PortTrafficTargetsPut,
)
from .port_traffic_service import (
compare_targets,
create_task,
create_device,
dashboard,
delete_task,
delete_device,
discover_ports,
get_device,
get_samples,
get_task,
list_device_events,
list_devices,
list_series,
list_targets,
list_tasks,
put_targets,
put_interfaces,
replace_series_port,
set_task_status,
update_task,
set_device_status,
update_device,
)
router = APIRouter(prefix="/v1/port-traffic", tags=["port-traffic"])
@ -48,23 +50,23 @@ def api_dashboard(db: Session = Depends(get_db)):
return dashboard(db).model_dump()
@router.get("/tasks")
def api_list_tasks(
@router.get("/devices")
def api_list_devices(
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=100),
db: Session = Depends(get_db),
):
return list_tasks(db, page=page, page_size=page_size)
return list_devices(db, page=page, page_size=page_size)
@router.post("/tasks")
def api_create_task(
body: PortTrafficTaskCreate,
@router.post("/devices")
def api_create_device(
body: PortTrafficDeviceCreate,
background_tasks: BackgroundTasks,
request: Request,
db: Session = Depends(get_db),
):
out = create_task(db, body)
out = create_device(db, body)
if body.start_now and out.status == "running":
from .port_traffic_runner import dispatch_collect
@ -72,179 +74,206 @@ def api_create_task(
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.create",
action="port_traffic.device.create",
actor_user_id=uid,
actor_username=uname,
method="POST",
path="/v1/port-traffic/tasks",
path="/v1/port-traffic/devices",
status_code=200,
detail={"id": out.id, "title": out.title, "start_now": body.start_now},
detail={"id": out.id, "ne_id": out.ne_id, "start_now": body.start_now},
)
return out.model_dump()
@router.get("/tasks/{task_id}")
def api_get_task(task_id: str, db: Session = Depends(get_db)):
return get_task(db, task_id).model_dump()
@router.get("/devices/{device_id}")
def api_get_device(device_id: str, db: Session = Depends(get_db)):
return get_device(db, device_id).model_dump()
@router.patch("/tasks/{task_id}")
def api_patch_task(
task_id: str,
body: PortTrafficTaskUpdate,
@router.patch("/devices/{device_id}")
def api_patch_device(
device_id: str,
body: PortTrafficDeviceUpdate,
request: Request,
db: Session = Depends(get_db),
):
out = update_task(db, task_id, body)
out = update_device(db, device_id, body)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.update",
action="port_traffic.device.update",
actor_user_id=uid,
actor_username=uname,
method="PATCH",
path=f"/v1/port-traffic/tasks/{task_id}",
path=f"/v1/port-traffic/devices/{device_id}",
status_code=200,
detail=body.model_dump(exclude_unset=True),
)
return out.model_dump()
@router.delete("/tasks/{task_id}")
def api_delete_task(task_id: str, request: Request, db: Session = Depends(get_db)):
out = delete_task(db, task_id)
@router.delete("/devices/{device_id}")
def api_delete_device(device_id: str, request: Request, db: Session = Depends(get_db)):
out = delete_device(db, device_id)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.delete",
action="port_traffic.device.delete",
actor_user_id=uid,
actor_username=uname,
method="DELETE",
path=f"/v1/port-traffic/tasks/{task_id}",
path=f"/v1/port-traffic/devices/{device_id}",
status_code=200,
detail={"id": task_id},
detail={"id": device_id},
)
return out
@router.post("/tasks/{task_id}/start")
def api_start_task(
task_id: str,
@router.post("/devices/{device_id}/start")
def api_start_device(
device_id: str,
background_tasks: BackgroundTasks,
request: Request,
db: Session = Depends(get_db),
):
out = set_task_status(db, task_id, "running")
out = set_device_status(db, device_id, "running")
from .port_traffic_runner import dispatch_collect
background_tasks.add_task(dispatch_collect, task_id)
background_tasks.add_task(dispatch_collect, device_id)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.start",
action="port_traffic.device.start",
actor_user_id=uid,
actor_username=uname,
method="POST",
path=f"/v1/port-traffic/tasks/{task_id}/start",
path=f"/v1/port-traffic/devices/{device_id}/start",
status_code=200,
detail={"id": task_id},
detail={"id": device_id},
)
return out.model_dump()
@router.post("/tasks/{task_id}/collect-now")
@router.post("/devices/{device_id}/collect-now")
def api_collect_now(
task_id: str,
device_id: str,
background_tasks: BackgroundTasks,
request: Request,
db: Session = Depends(get_db),
):
task = get_task(db, task_id)
if task.status not in ("running", "paused", "draft", "stopped"):
raise HTTPException(status_code=400, detail="invalid_status")
# Force due: clear last end so claim accepts, ensure running for this round.
from .models import PortTrafficTask
device = get_device(db, device_id)
from .port_traffic_runner import dispatch_collect
row = db.get(PortTrafficTask, task_id)
row = db.get(PortTrafficDevice, device_id)
if not row:
raise HTTPException(status_code=404, detail="task_not_found")
raise HTTPException(status_code=404, detail="device_not_found")
if bool(row.collect_running):
return {"ok": True, "started": False, "reason": "already_collecting", **task.model_dump()}
return {"ok": True, "started": False, "reason": "already_collecting", **device.model_dump()}
if str(row.status) != "running":
row.status = "running"
row.last_collect_ended_at = None
row.updated_at = datetime.utcnow()
db.commit()
background_tasks.add_task(dispatch_collect, task_id)
background_tasks.add_task(dispatch_collect, device_id)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.collect_now",
action="port_traffic.device.collect_now",
actor_user_id=uid,
actor_username=uname,
method="POST",
path=f"/v1/port-traffic/tasks/{task_id}/collect-now",
path=f"/v1/port-traffic/devices/{device_id}/collect-now",
status_code=200,
detail={"id": task_id},
detail={"id": device_id},
)
out = get_task(db, task_id)
out = get_device(db, device_id)
return {"ok": True, "started": True, **out.model_dump()}
@router.post("/tasks/{task_id}/pause")
def api_pause_task(task_id: str, request: Request, db: Session = Depends(get_db)):
out = set_task_status(db, task_id, "paused")
@router.post("/devices/{device_id}/pause")
def api_pause_device(device_id: str, request: Request, db: Session = Depends(get_db)):
out = set_device_status(db, device_id, "paused")
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.pause",
action="port_traffic.device.pause",
actor_user_id=uid,
actor_username=uname,
method="POST",
path=f"/v1/port-traffic/tasks/{task_id}/pause",
path=f"/v1/port-traffic/devices/{device_id}/pause",
status_code=200,
detail={"id": task_id},
detail={"id": device_id},
)
return out.model_dump()
@router.post("/tasks/{task_id}/stop")
def api_stop_task(task_id: str, request: Request, db: Session = Depends(get_db)):
out = set_task_status(db, task_id, "stopped")
@router.post("/devices/{device_id}/stop")
def api_stop_device(device_id: str, request: Request, db: Session = Depends(get_db)):
out = set_device_status(db, device_id, "stopped")
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.task.stop",
action="port_traffic.device.stop",
actor_user_id=uid,
actor_username=uname,
method="POST",
path=f"/v1/port-traffic/tasks/{task_id}/stop",
path=f"/v1/port-traffic/devices/{device_id}/stop",
status_code=200,
detail={"id": task_id},
detail={"id": device_id},
)
return out.model_dump()
@router.get("/tasks/{task_id}/targets")
def api_list_targets(task_id: str, db: Session = Depends(get_db)):
return {"items": [t.model_dump() for t in list_targets(db, task_id)]}
@router.get("/devices/{device_id}/targets")
def api_list_targets(device_id: str, db: Session = Depends(get_db)):
return {"items": [t.model_dump() for t in list_targets(db, device_id)]}
@router.get("/tasks/{task_id}/series")
def api_list_series(task_id: str, db: Session = Depends(get_db)):
return {"items": [s.model_dump() for s in list_series(db, task_id)]}
@router.get("/devices/{device_id}/events")
def api_list_device_events(
device_id: str,
limit: int = Query(100, ge=1, le=500),
db: Session = Depends(get_db),
):
return list_device_events(db, device_id, limit=limit).model_dump()
@router.post("/tasks/{task_id}/series/{series_id}/replace")
@router.put("/devices/{device_id}/interfaces")
def api_put_interfaces(
device_id: str,
body: PortTrafficInterfacesPut,
request: Request,
db: Session = Depends(get_db),
):
items = put_interfaces(db, device_id, body)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.device.interfaces_put",
actor_user_id=uid,
actor_username=uname,
method="PUT",
path=f"/v1/port-traffic/devices/{device_id}/interfaces",
status_code=200,
detail={"id": device_id, "count": len(items)},
)
return {"items": [t.model_dump() for t in items]}
@router.get("/devices/{device_id}/series")
def api_list_series(device_id: str, db: Session = Depends(get_db)):
return {"items": [s.model_dump() for s in list_series(db, device_id)]}
@router.post("/devices/{device_id}/series/{series_id}/replace")
def api_replace_series_port(
task_id: str,
device_id: str,
series_id: str,
body: PortTrafficReplacePortRequest,
request: Request,
db: Session = Depends(get_db),
):
out = replace_series_port(db, task_id, series_id, body)
out = replace_series_port(db, device_id, series_id, body)
uid, uname = _actor(request)
write_audit(
db,
@ -252,35 +281,13 @@ def api_replace_series_port(
actor_user_id=uid,
actor_username=uname,
method="POST",
path=f"/v1/port-traffic/tasks/{task_id}/series/{series_id}/replace",
path=f"/v1/port-traffic/devices/{device_id}/series/{series_id}/replace",
status_code=200,
detail={"series_id": series_id, "ifname": body.ifname, "target_id": body.target_id},
detail={"series_id": series_id, "ifname": body.ifname},
)
return out.model_dump()
@router.put("/tasks/{task_id}/targets")
def api_put_targets(
task_id: str,
body: PortTrafficTargetsPut,
request: Request,
db: Session = Depends(get_db),
):
items = put_targets(db, task_id, body)
uid, uname = _actor(request)
write_audit(
db,
action="port_traffic.targets.put",
actor_user_id=uid,
actor_username=uname,
method="PUT",
path=f"/v1/port-traffic/tasks/{task_id}/targets",
status_code=200,
detail={"id": task_id, "count": len(items)},
)
return {"items": [t.model_dump() for t in items]}
@router.post("/discover/ports")
def api_discover_ports(body: DiscoverPortsRequest, db: Session = Depends(get_db)):
return discover_ports(db, body).model_dump()
@ -304,7 +311,7 @@ def api_compare(
offset_hours: float | None = Query(default=None, ge=0.25, le=24 * 90),
baseline_target_id: str | None = Query(
default=None,
description="optional mapped interface for baseline overlay (any task)",
description="optional mapped interface for baseline overlay",
),
to_ts: datetime | None = Query(default=None, alias="to"),
db: Session = Depends(get_db),

354
netx_api/port_traffic_runner.py
View file

@ -1,4 +1,4 @@
"""Port traffic collection worker: claim task round, sample interfaces via CLI."""
"""Port traffic collection worker: claim device round, sample interfaces via one CLI session."""
from __future__ import annotations
@ -15,11 +15,11 @@ from fastapi import HTTPException
from .cli_resolve import resolve_cli_target
from .config import settings
from .db import SessionLocal
from .models import PortTrafficSample, PortTrafficTarget, PortTrafficTask
from .models import PortTrafficDevice, PortTrafficEvent, PortTrafficSample, PortTrafficTarget
from .ne_session_factory import close_netmiko_connection, open_netmiko_connection
from .ne_netmiko import send_show_command
from .port_traffic_commands import commands_for_vendor, detail_command
from .port_traffic_parsers import parse_interface_detail
from .port_traffic_parsers import parse_interface_detail, resolve_util_pct
_log = logging.getLogger("netx.port_traffic.runner")
_pools: dict[str, ThreadPoolExecutor] = {}
@ -34,26 +34,29 @@ def _format_error(exc: BaseException) -> str:
return f"{type(exc).__name__}: {exc}"[:1020]
def _pool_for_task(task_id: str, concurrency: int) -> ThreadPoolExecutor:
with _pools_lock:
pool = _pools.get(task_id)
if pool is None:
workers = max(1, min(20, int(concurrency or 5)))
pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix=f"pt-{task_id[:8]}")
_pools[task_id] = pool
return pool
def _release_pool(task_id: str) -> None:
with _pools_lock:
pool = _pools.pop(task_id, None)
if pool is not None:
try:
pool.shutdown(wait=False, cancel_futures=False)
except TypeError:
pool.shutdown(wait=False)
except Exception:
_log.exception("port_traffic pool shutdown failed task=%s", task_id)
def _append_event(
db,
*,
device_id: str,
message: str,
level: str = "error",
target_row_id: str = "",
ifname: str = "",
) -> None:
msg = str(message or "").strip()
if not msg or not device_id:
return
db.add(
PortTrafficEvent(
id=uuid4().hex,
device_id=device_id,
target_row_id=str(target_row_id or ""),
ifname=str(ifname or ""),
level=str(level or "error")[:16],
message=msg[:4000],
created_at=_utcnow(),
)
)
def _set_target_error(target_row_id: str, message: str) -> None:
@ -62,25 +65,72 @@ def _set_target_error(target_row_id: str, message: str) -> None:
row = db.get(PortTrafficTarget, target_row_id)
if row:
row.last_error = message[:1020]
_append_event(
db,
device_id=str(row.device_id or ""),
target_row_id=str(row.id),
ifname=str(row.ifname or ""),
message=message,
level="error",
)
db.commit()
finally:
db.close()
def _claim_collect_round(task_id: str) -> list[str] | None:
"""Mark task collect_running and return active target row ids, or None if skip."""
def _finish_collect_round(device_id: str, *, error: str = "") -> None:
db = SessionLocal()
try:
device = db.get(PortTrafficDevice, device_id)
if not device:
return
device.collect_running = False
device.last_collect_ended_at = _utcnow()
if error:
device.last_error = error[:1020]
_append_event(db, device_id=device_id, message=error, level="error")
device.updated_at = _utcnow()
db.commit()
finally:
db.close()
_release_pool(device_id)
def _pool_for_device(device_id: str, concurrency: int) -> ThreadPoolExecutor:
with _pools_lock:
pool = _pools.get(device_id)
if pool is None:
workers = max(1, min(5, int(concurrency or 1)))
pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix=f"pt-{device_id[:8]}")
_pools[device_id] = pool
return pool
def _release_pool(device_id: str) -> None:
with _pools_lock:
pool = _pools.pop(device_id, None)
if pool is not None:
try:
pool.shutdown(wait=False, cancel_futures=False)
except TypeError:
pool.shutdown(wait=False)
except Exception:
_log.exception("port_traffic pool shutdown failed device=%s", device_id)
def _claim_collect_round(device_id: str) -> list[str] | None:
for attempt in range(8):
db = SessionLocal()
try:
task = db.get(PortTrafficTask, task_id)
if not task:
device = db.get(PortTrafficDevice, device_id)
if not device:
return None
if str(task.status or "") != "running":
if str(device.status or "") != "running":
return None
if bool(task.collect_running):
if bool(device.collect_running):
return None
ended = task.last_collect_ended_at
interval = max(15, int(task.interval_sec or 60))
ended = device.last_collect_ended_at
interval = max(15, int(device.interval_sec or 60))
if ended is not None:
elapsed = (_utcnow() - ended).total_seconds()
if elapsed < interval:
@ -88,142 +138,69 @@ def _claim_collect_round(task_id: str) -> list[str] | None:
targets = (
db.query(PortTrafficTarget)
.filter(
PortTrafficTarget.task_id == task_id,
PortTrafficTarget.device_id == device_id,
PortTrafficTarget.status == "active",
)
.all()
)
if not targets:
return None
task.collect_running = True
task.last_collect_started_at = _utcnow()
task.last_error = ""
task.updated_at = _utcnow()
device.collect_running = True
device.last_collect_started_at = _utcnow()
device.last_error = ""
device.updated_at = _utcnow()
db.commit()
return [str(t.id) for t in targets]
except Exception:
db.rollback()
_log.exception("port_traffic claim failed task=%s attempt=%s", task_id, attempt)
_log.exception("port_traffic claim failed device=%s attempt=%s", device_id, attempt)
time.sleep(0.05 * (attempt + 1))
finally:
db.close()
return None
def _finish_collect_round(task_id: str, *, error: str = "") -> None:
db = SessionLocal()
try:
task = db.get(PortTrafficTask, task_id)
if not task:
return
task.collect_running = False
task.last_collect_ended_at = _utcnow()
if error:
task.last_error = error[:1020]
task.updated_at = _utcnow()
db.commit()
finally:
db.close()
_release_pool(task_id)
def _run_show(creds: dict[str, Any], command: str, read_timeout: int) -> str:
conn = open_netmiko_connection(creds, session_timeout=read_timeout + 60)
try:
return send_show_command(conn, command, read_timeout=read_timeout)
finally:
close_netmiko_connection(conn)
def _sample_one_target(target_row_id: str) -> None:
creds: dict[str, Any] | None = None
cmd = ""
vendor_key = "zte"
per_cmd = int(settings.ne_collect_read_timeout_sec or 120)
cap = int(settings.ne_collect_run_timeout_cap_sec or 600)
db = SessionLocal()
try:
row = db.get(PortTrafficTarget, target_row_id)
if not row or str(row.status or "") != "active":
return
source = str(row.source or "").strip().lower()
target_id = str(row.target_id or "").strip()
ifname = str(row.ifname or "").strip()
vendor_hint = str(row.vendor or "")
try:
if source == "managed":
creds, device = resolve_cli_target(db, managed_ne_id=target_id)
elif source == "ume":
creds, device = resolve_cli_target(db, ume_ne_id=target_id)
else:
row.last_error = "invalid_source"
db.commit()
return
except HTTPException as exc:
row.last_error = str(exc.detail or "resolve_failed")[:1020]
db.commit()
return
except Exception as exc:
row.last_error = _format_error(exc)
db.commit()
return
vendor = str(device.get("vendor") or vendor_hint or "")
device_type = str(device.get("device_type") or "")
cmds = commands_for_vendor(vendor, device_type)
if cmds is None:
row.last_error = "unsupported_vendor"
db.commit()
return
vendor_key = cmds.vendor_key
cmd = detail_command(cmds, ifname)
finally:
db.close()
if not creds or not cmd:
return
budget = min(cap, per_cmd + 90)
try:
with ThreadPoolExecutor(max_workers=1) as pool:
fut = pool.submit(_run_show, creds, cmd, per_cmd)
raw = fut.result(timeout=budget)
except Exception as exc:
_set_target_error(target_row_id, _format_error(exc))
return
parsed = parse_interface_detail(raw, vendor_key)
if (
parsed.in_bps == 0
and parsed.out_bps == 0
and parsed.bw_bps == 0
and parsed.in_util_pct == 0
and parsed.out_util_pct == 0
and not parsed.ifname
):
_set_target_error(target_row_id, "parse_empty")
return
def _save_sample(target_row_id: str, parsed: Any, vendor_hint_bw: int = 0) -> None:
now = _utcnow()
db = SessionLocal()
try:
row = db.get(PortTrafficTarget, target_row_id)
if not row:
return
bw = int(parsed.bw_bps or row.bw_bps or 0)
bw = int(parsed.bw_bps or row.bw_bps or vendor_hint_bw or 0)
if bw and not row.bw_bps:
row.bw_bps = bw
in_bps = float(parsed.in_bps)
out_bps = float(parsed.out_bps)
if (
in_bps == 0
and out_bps == 0
and bw == 0
and float(parsed.in_util_pct or 0) == 0
and float(parsed.out_util_pct or 0) == 0
and not parsed.ifname
):
row.last_error = "parse_empty"
_append_event(
db,
device_id=str(row.device_id or ""),
target_row_id=str(row.id),
ifname=str(row.ifname or ""),
message="parse_empty",
level="warn",
)
db.commit()
return
db.add(
PortTrafficSample(
id=uuid4().hex,
target_row_id=target_row_id,
series_id=str(row.series_id or ""),
ts=now,
in_bps=float(parsed.in_bps),
out_bps=float(parsed.out_bps),
in_util_pct=float(parsed.in_util_pct),
out_util_pct=float(parsed.out_util_pct),
in_bps=in_bps,
out_bps=out_bps,
in_util_pct=resolve_util_pct(float(parsed.in_util_pct), in_bps, bw),
out_util_pct=resolve_util_pct(float(parsed.out_util_pct), out_bps, bw),
bw_bps=bw,
rate_period_sec=int(parsed.rate_period_sec or 0),
raw_ok=True,
@ -240,30 +217,103 @@ def _sample_one_target(target_row_id: str) -> None:
db.close()
def dispatch_collect(task_id: str) -> int:
"""Claim and sample all active targets for a running task. Returns target count."""
target_ids = _claim_collect_round(task_id)
if not target_ids:
return 0
def _sample_targets_shared_session(device_id: str, target_ids: list[str]) -> int:
"""One CLI login for the device; run show per interface. Returns error count."""
per_cmd = int(settings.ne_collect_read_timeout_sec or 120)
cap = int(settings.ne_collect_run_timeout_cap_sec or 600)
errors = 0
db = SessionLocal()
try:
task = db.get(PortTrafficTask, task_id)
concurrency = int(task.concurrency or 5) if task else 5
device = db.get(PortTrafficDevice, device_id)
if not device:
return len(target_ids)
source = str(device.source or "").strip().lower()
ne_id = str(device.ne_id or "").strip()
vendor_hint = str(device.vendor or "")
try:
if source == "managed":
creds, info = resolve_cli_target(db, managed_ne_id=ne_id)
elif source == "ume":
creds, info = resolve_cli_target(db, ume_ne_id=ne_id)
else:
for tid in target_ids:
_set_target_error(tid, "invalid_source")
return len(target_ids)
except HTTPException as exc:
msg = str(exc.detail or "resolve_failed")[:1020]
for tid in target_ids:
_set_target_error(tid, msg)
return len(target_ids)
except Exception as exc:
msg = _format_error(exc)
for tid in target_ids:
_set_target_error(tid, msg)
return len(target_ids)
vendor = str(info.get("vendor") or vendor_hint or "")
device_type = str(info.get("device_type") or "")
cmds = commands_for_vendor(vendor, device_type)
if cmds is None:
for tid in target_ids:
_set_target_error(tid, "unsupported_vendor")
return len(target_ids)
vendor_key = cmds.vendor_key
targets = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.id.in_(target_ids), PortTrafficTarget.status == "active")
.all()
)
ifaces = [(str(t.id), str(t.ifname or "").strip()) for t in targets if t.ifname]
finally:
db.close()
pool = _pool_for_task(task_id, concurrency)
futures = [pool.submit(_sample_one_target, tid) for tid in target_ids]
errors = 0
if not ifaces:
return 0
budget = min(cap, per_cmd * max(1, len(ifaces)) + 90)
conn = None
try:
for fut in as_completed(futures):
conn = open_netmiko_connection(creds, session_timeout=budget)
for tid, ifname in ifaces:
try:
fut.result()
except Exception:
cmd = detail_command(cmds, ifname)
raw = send_show_command(conn, cmd, read_timeout=per_cmd)
parsed = parse_interface_detail(raw, vendor_key)
_save_sample(tid, parsed)
except Exception as exc:
errors += 1
_log.exception("port_traffic target worker failed task=%s", task_id)
_set_target_error(tid, _format_error(exc))
_log.exception("port_traffic iface sample failed device=%s if=%s", device_id, ifname)
except Exception as exc:
msg = _format_error(exc)
for tid, _ in ifaces:
_set_target_error(tid, msg)
errors = len(ifaces)
_log.exception("port_traffic session failed device=%s", device_id)
finally:
if conn is not None:
close_netmiko_connection(conn)
return errors
def dispatch_collect(device_id: str) -> int:
"""Claim and sample all active interfaces for a running device. Returns target count."""
target_ids = _claim_collect_round(device_id)
if not target_ids:
return 0
try:
errors = _sample_targets_shared_session(device_id, target_ids)
except Exception:
errors = len(target_ids)
_log.exception("port_traffic collect failed device=%s", device_id)
finally:
err_msg = f"{errors}_target_errors" if errors else ""
_finish_collect_round(task_id, error=err_msg)
_finish_collect_round(device_id, error=err_msg)
return len(target_ids)
# Alias used by older call sites
dispatch_collect_device = dispatch_collect

41
netx_api/port_traffic_scheduler.py
View file

@ -8,7 +8,7 @@ from datetime import datetime
from .config import settings
from .db import SessionLocal
from .models import PortTrafficTask
from .models import PortTrafficDevice
from .port_traffic_runner import dispatch_collect
from .port_traffic_service import purge_expired_samples
@ -22,40 +22,47 @@ def _utcnow() -> datetime:
return datetime.utcnow()
def try_dispatch_due_tasks() -> int:
"""Dispatch collect rounds for due running tasks. Returns number started."""
def try_dispatch_due_devices() -> int:
"""Dispatch collect rounds for due running devices. Returns number started."""
db = SessionLocal()
try:
tasks = (
db.query(PortTrafficTask)
.filter(PortTrafficTask.status == "running", PortTrafficTask.collect_running.is_(False))
devices = (
db.query(PortTrafficDevice)
.filter(
PortTrafficDevice.status == "running",
PortTrafficDevice.collect_running.is_(False),
)
.all()
)
due_ids: list[str] = []
now = _utcnow()
for task in tasks:
interval = max(15, int(task.interval_sec or 60))
ended = task.last_collect_ended_at
for device in devices:
interval = max(15, int(device.interval_sec or 60))
ended = device.last_collect_ended_at
if ended is None:
due_ids.append(str(task.id))
due_ids.append(str(device.id))
continue
if (now - ended).total_seconds() >= interval:
due_ids.append(str(task.id))
due_ids.append(str(device.id))
finally:
db.close()
started = 0
for tid in due_ids:
for did in due_ids:
try:
n = dispatch_collect(tid)
n = dispatch_collect(did)
if n:
started += 1
_log.info("port_traffic collect started task=%s targets=%s", tid, n)
_log.info("port_traffic collect started device=%s targets=%s", did, n)
except Exception:
_log.exception("port_traffic dispatch failed task=%s", tid)
_log.exception("port_traffic dispatch failed device=%s", did)
return started
def try_dispatch_due_tasks() -> int:
return try_dispatch_due_devices()
def _loop() -> None:
global _purge_counter
tick = max(5, int(settings.port_traffic_scheduler_tick_sec or 15))
@ -63,9 +70,8 @@ def _loop() -> None:
while not _stop.is_set():
try:
if bool(settings.port_traffic_scheduler_enabled):
try_dispatch_due_tasks()
try_dispatch_due_devices()
_purge_counter += 1
# Retention purge roughly every ~20 ticks
if _purge_counter >= 20:
_purge_counter = 0
db = SessionLocal()
@ -89,7 +95,6 @@ def start_port_traffic_scheduler() -> None:
_stop.clear()
_thread = threading.Thread(target=_loop, name="port-traffic-scheduler", daemon=True)
_thread.start()
_log.info("started thread %s alive=%s", _thread.name, _thread.is_alive())
def stop_port_traffic_scheduler() -> None:

92
netx_api/port_traffic_schemas.py
View file

@ -1,4 +1,4 @@
"""Pydantic schemas for port traffic monitoring API."""
"""Pydantic schemas for port traffic monitoring API (device-centric)."""
from __future__ import annotations
@ -8,17 +8,17 @@ from typing import Literal
from pydantic import BaseModel, Field
class PortTrafficNeRef(BaseModel):
source: Literal["managed", "ume"]
id: str
ne_name: str = ""
ne_ip: str = ""
vendor: str = ""
class PortTrafficIfaceIn(BaseModel):
ifname: str
if_description: str = ""
bw_bps: int = 0
class PortTrafficTargetIn(BaseModel):
source: Literal["managed", "ume"]
target_id: str
"""Legacy-shaped input; create/device APIs prefer PortTrafficIfaceIn under one NE."""
source: Literal["managed", "ume"] = "managed"
target_id: str = ""
ne_name: str = ""
ne_ip: str = ""
vendor: str = ""
@ -27,25 +27,38 @@ class PortTrafficTargetIn(BaseModel):
bw_bps: int = 0
class PortTrafficTaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=256)
class PortTrafficDeviceCreate(BaseModel):
source: Literal["managed", "ume"]
ne_id: str = Field(min_length=1, max_length=128)
ne_name: str = ""
ne_ip: str = ""
vendor: str = ""
note: str = Field(default="", max_length=256)
interval_sec: int = Field(default=60, ge=15, le=3600)
retention_days: int = Field(default=7, ge=1, le=90)
concurrency: int = Field(default=5, ge=1, le=20)
targets: list[PortTrafficTargetIn] = Field(default_factory=list)
concurrency: int = Field(default=1, ge=1, le=5)
interfaces: list[PortTrafficIfaceIn] = Field(default_factory=list)
start_now: bool = False
class PortTrafficTaskUpdate(BaseModel):
title: str | None = Field(default=None, min_length=1, max_length=256)
class PortTrafficDeviceUpdate(BaseModel):
note: str | None = Field(default=None, max_length=256)
interval_sec: int | None = Field(default=None, ge=15, le=3600)
retention_days: int | None = Field(default=None, ge=1, le=90)
concurrency: int | None = Field(default=None, ge=1, le=20)
concurrency: int | None = Field(default=None, ge=1, le=5)
ne_name: str | None = None
ne_ip: str | None = None
vendor: str | None = None
class PortTrafficTaskOut(BaseModel):
class PortTrafficDeviceOut(BaseModel):
id: str
title: str
source: str
ne_id: str
ne_name: str
ne_ip: str
vendor: str
note: str = ""
status: str
interval_sec: int
retention_days: int
@ -62,7 +75,8 @@ class PortTrafficTaskOut(BaseModel):
class PortTrafficTargetOut(BaseModel):
id: str
task_id: str
device_id: str
task_id: str = "" # alias of device_id for older clients
series_id: str = ""
source: str
target_id: str
@ -80,7 +94,8 @@ class PortTrafficTargetOut(BaseModel):
class PortTrafficSeriesOut(BaseModel):
id: str
task_id: str
device_id: str
task_id: str = ""
title: str
status: str
active_target: PortTrafficTargetOut | None = None
@ -88,18 +103,13 @@ class PortTrafficSeriesOut(BaseModel):
created_at: datetime | None = None
class PortTrafficTargetsPut(BaseModel):
targets: list[PortTrafficTargetIn]
class PortTrafficInterfacesPut(BaseModel):
"""Replace active interface set for a device (keeps samples for retired/removed)."""
interfaces: list[PortTrafficIfaceIn]
class PortTrafficReplacePortRequest(BaseModel):
"""Replace the active interface under a logical series (keeps history)."""
source: Literal["managed", "ume"]
target_id: str
ne_name: str = ""
ne_ip: str = ""
vendor: str = ""
ifname: str
if_description: str = ""
bw_bps: int = 0
@ -141,7 +151,7 @@ class PortTrafficSamplePoint(BaseModel):
out_util_pct: float
bw_bps: int
rate_period_sec: int = 0
ts_raw: datetime | None = None # baseline original time when aligned
ts_raw: datetime | None = None
class PortTrafficSamplesOut(BaseModel):
@ -166,8 +176,26 @@ class PortTrafficCompareOut(BaseModel):
class PortTrafficDashboardOut(BaseModel):
task_count: int = 0
running_task_count: int = 0
device_count: int = 0
running_device_count: int = 0
active_target_count: int = 0
sample_count_24h: int = 0
last_sample_at: datetime | None = None
# Back-compat aliases for older UI
task_count: int = 0
running_task_count: int = 0
class PortTrafficEventOut(BaseModel):
id: str
device_id: str
target_row_id: str = ""
ifname: str = ""
level: str = "error"
message: str = ""
created_at: datetime | None = None
class PortTrafficEventsOut(BaseModel):
items: list[PortTrafficEventOut] = Field(default_factory=list)
total: int = 0

598
netx_api/port_traffic_service.py
View file

@ -1,4 +1,4 @@
"""Port traffic monitoring service: CRUD, discover, samples, dashboard."""
"""Port traffic monitoring service: device-centric CRUD, discover, samples, dashboard."""
from __future__ import annotations
@ -13,12 +13,12 @@ from sqlalchemy.orm import Session
from .cli_resolve import resolve_cli_target
from .config import settings
from .models import PortTrafficSample, PortTrafficSeries, PortTrafficTarget, PortTrafficTask
from .models import PortTrafficDevice, PortTrafficEvent, PortTrafficSample, PortTrafficSeries, PortTrafficTarget
from .ne_session_factory import close_netmiko_connection, open_netmiko_connection
from .ne_netmiko import send_show_command
from .port_traffic_commands import commands_for_vendor
from .port_traffic_migrate import default_series_title, unique_series_title
from .port_traffic_parsers import brief_port_to_dict, parse_interface_brief
from .port_traffic_parsers import brief_port_to_dict, parse_interface_brief, resolve_util_pct
from .port_traffic_schemas import (
DiscoverPortItem,
DiscoverPortsRequest,
@ -26,16 +26,18 @@ from .port_traffic_schemas import (
PortTrafficCompareMeta,
PortTrafficCompareOut,
PortTrafficDashboardOut,
PortTrafficDeviceCreate,
PortTrafficDeviceOut,
PortTrafficDeviceUpdate,
PortTrafficEventOut,
PortTrafficEventsOut,
PortTrafficIfaceIn,
PortTrafficInterfacesPut,
PortTrafficReplacePortRequest,
PortTrafficSamplePoint,
PortTrafficSamplesOut,
PortTrafficSeriesOut,
PortTrafficTargetIn,
PortTrafficTargetOut,
PortTrafficTargetsPut,
PortTrafficTaskCreate,
PortTrafficTaskOut,
PortTrafficTaskUpdate,
)
_log = logging.getLogger("netx.port_traffic.service")
@ -45,36 +47,12 @@ def _utcnow() -> datetime:
return datetime.utcnow()
def _task_out(db: Session, task: PortTrafficTask) -> PortTrafficTaskOut:
tid = str(task.id)
total = db.query(PortTrafficTarget).filter(PortTrafficTarget.task_id == tid).count()
active = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.task_id == tid, PortTrafficTarget.status == "active")
.count()
)
return PortTrafficTaskOut(
id=tid,
title=str(task.title or ""),
status=str(task.status or ""),
interval_sec=int(task.interval_sec or 60),
retention_days=int(task.retention_days or 7),
concurrency=int(task.concurrency or 5),
collect_running=bool(task.collect_running),
target_count=int(total),
active_target_count=int(active),
last_collect_started_at=task.last_collect_started_at,
last_collect_ended_at=task.last_collect_ended_at,
last_error=str(task.last_error or ""),
created_at=task.created_at,
updated_at=task.updated_at,
)
def _target_out(row: PortTrafficTarget) -> PortTrafficTargetOut:
did = str(row.device_id or "")
return PortTrafficTargetOut(
id=str(row.id),
task_id=str(row.task_id),
device_id=did,
task_id=did,
series_id=str(row.series_id or ""),
source=str(row.source or ""),
target_id=str(row.target_id or ""),
@ -91,19 +69,89 @@ def _target_out(row: PortTrafficTarget) -> PortTrafficTargetOut:
)
def _create_series_and_target(
def _device_out(db: Session, device: PortTrafficDevice) -> PortTrafficDeviceOut:
did = str(device.id)
total = db.query(PortTrafficTarget).filter(PortTrafficTarget.device_id == did).count()
active = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.device_id == did, PortTrafficTarget.status == "active")
.count()
)
return PortTrafficDeviceOut(
id=did,
source=str(device.source or ""),
ne_id=str(device.ne_id or ""),
ne_name=str(device.ne_name or ""),
ne_ip=str(device.ne_ip or ""),
vendor=str(device.vendor or ""),
note=str(device.note or ""),
status=str(device.status or ""),
interval_sec=int(device.interval_sec or 60),
retention_days=int(device.retention_days or 7),
concurrency=int(device.concurrency or 1),
collect_running=bool(device.collect_running),
target_count=int(total),
active_target_count=int(active),
last_collect_started_at=device.last_collect_started_at,
last_collect_ended_at=device.last_collect_ended_at,
last_error=str(device.last_error or ""),
created_at=device.created_at,
updated_at=device.updated_at,
)
def _assert_vendor(vendor: str, label: str) -> None:
cmds = commands_for_vendor(vendor or "", "")
if cmds is None:
raise HTTPException(
status_code=400,
detail=f"vendor_not_supported_for_port_traffic: {vendor or 'unknown'} ({label})",
)
def _assert_ifaces_free(
db: Session,
*,
task_id: str,
t: PortTrafficTargetIn,
source: str,
ne_id: str,
ifnames: list[str],
exclude_device_id: str | None = None,
) -> None:
names = [str(x).strip() for x in ifnames if str(x).strip()]
if not names:
return
q = db.query(PortTrafficTarget).filter(
PortTrafficTarget.source == source,
PortTrafficTarget.target_id == ne_id,
PortTrafficTarget.ifname.in_(names),
PortTrafficTarget.status == "active",
)
if exclude_device_id:
q = q.filter(PortTrafficTarget.device_id != exclude_device_id)
hit = q.first()
if hit:
raise HTTPException(
status_code=409,
detail=f"interface_already_monitored: {hit.ifname} on device {hit.device_id}",
)
def _create_iface(
db: Session,
*,
device: PortTrafficDevice,
iface: PortTrafficIfaceIn,
now: datetime,
) -> PortTrafficTarget:
title = unique_series_title(db, task_id, default_series_title(t.ne_name or "", t.ifname.strip()))
ifname = iface.ifname.strip()
title = unique_series_title(
db, str(device.id), default_series_title(device.ne_name or "", ifname)
)
sid = uuid4().hex
db.add(
PortTrafficSeries(
id=sid,
task_id=task_id,
device_id=str(device.id),
title=title,
status="active",
created_at=now,
@ -111,16 +159,16 @@ def _create_series_and_target(
)
row = PortTrafficTarget(
id=uuid4().hex,
task_id=task_id,
device_id=str(device.id),
series_id=sid,
source=t.source,
target_id=t.target_id,
ne_name=t.ne_name or "",
ne_ip=t.ne_ip or "",
vendor=t.vendor or "",
ifname=t.ifname.strip(),
if_description=t.if_description or "",
bw_bps=int(t.bw_bps or 0),
source=str(device.source),
target_id=str(device.ne_id),
ne_name=str(device.ne_name or ""),
ne_ip=str(device.ne_ip or ""),
vendor=str(device.vendor or ""),
ifname=ifname,
if_description=iface.if_description or "",
bw_bps=int(iface.bw_bps or 0),
status="active",
created_at=now,
)
@ -128,44 +176,49 @@ def _create_series_and_target(
return row
def _assert_supported_targets(targets: list[PortTrafficTargetIn]) -> None:
for t in targets:
cmds = commands_for_vendor(t.vendor or "", "")
if cmds is None:
raise HTTPException(
status_code=400,
detail=f"vendor_not_supported_for_port_traffic: {t.vendor or 'unknown'} ({t.ne_name or t.target_id})",
)
def list_tasks(db: Session, *, page: int = 1, page_size: int = 20) -> dict[str, Any]:
q = db.query(PortTrafficTask).order_by(PortTrafficTask.created_at.desc())
def list_devices(db: Session, *, page: int = 1, page_size: int = 20) -> dict[str, Any]:
q = db.query(PortTrafficDevice).order_by(PortTrafficDevice.ne_name.asc(), PortTrafficDevice.ne_ip.asc())
total = q.count()
rows = q.offset((page - 1) * page_size).limit(page_size).all()
return {
"total": total,
"page": page,
"page_size": page_size,
"items": [_task_out(db, r).model_dump() for r in rows],
"items": [_device_out(db, r).model_dump() for r in rows],
}
def get_task(db: Session, task_id: str) -> PortTrafficTaskOut:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
return _task_out(db, task)
def get_device(db: Session, device_id: str) -> PortTrafficDeviceOut:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
return _device_out(db, device)
def create_task(db: Session, body: PortTrafficTaskCreate) -> PortTrafficTaskOut:
_assert_supported_targets(body.targets)
def create_device(db: Session, body: PortTrafficDeviceCreate) -> PortTrafficDeviceOut:
source = body.source
ne_id = body.ne_id.strip()
_assert_vendor(body.vendor, body.ne_name or ne_id)
clash = (
db.query(PortTrafficDevice)
.filter(PortTrafficDevice.source == source, PortTrafficDevice.ne_id == ne_id)
.first()
)
if clash:
raise HTTPException(status_code=409, detail="device_already_monitored")
ifnames = [i.ifname for i in body.interfaces]
_assert_ifaces_free(db, source=source, ne_id=ne_id, ifnames=ifnames)
now = _utcnow()
status = "running" if body.start_now and body.targets else "draft"
if body.start_now and not body.targets:
status = "draft"
task = PortTrafficTask(
status = "running" if body.start_now and body.interfaces else "draft"
device = PortTrafficDevice(
id=uuid4().hex,
title=body.title.strip(),
source=source,
ne_id=ne_id,
ne_name=(body.ne_name or "").strip(),
ne_ip=(body.ne_ip or "").strip(),
vendor=(body.vendor or "").strip(),
note=(body.note or "").strip(),
status=status,
interval_sec=int(body.interval_sec),
retention_days=int(body.retention_days),
@ -173,134 +226,166 @@ def create_task(db: Session, body: PortTrafficTaskCreate) -> PortTrafficTaskOut:
created_at=now,
updated_at=now,
)
db.add(task)
db.add(device)
db.flush()
for t in body.targets:
_create_series_and_target(db, task_id=task.id, t=t, now=now)
for iface in body.interfaces:
if not iface.ifname.strip():
continue
_create_iface(db, device=device, iface=iface, now=now)
db.commit()
db.refresh(task)
return _task_out(db, task)
db.refresh(device)
return _device_out(db, device)
def update_task(db: Session, task_id: str, body: PortTrafficTaskUpdate) -> PortTrafficTaskOut:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
def update_device(db: Session, device_id: str, body: PortTrafficDeviceUpdate) -> PortTrafficDeviceOut:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
data = body.model_dump(exclude_unset=True)
if "title" in data and data["title"] is not None:
task.title = str(data["title"]).strip()
for key in ("note", "ne_name", "ne_ip", "vendor"):
if key in data and data[key] is not None:
setattr(device, key, str(data[key]).strip())
if "interval_sec" in data and data["interval_sec"] is not None:
task.interval_sec = int(data["interval_sec"])
device.interval_sec = int(data["interval_sec"])
if "retention_days" in data and data["retention_days"] is not None:
task.retention_days = int(data["retention_days"])
device.retention_days = int(data["retention_days"])
if "concurrency" in data and data["concurrency"] is not None:
task.concurrency = int(data["concurrency"])
task.updated_at = _utcnow()
device.concurrency = int(data["concurrency"])
device.updated_at = _utcnow()
db.commit()
db.refresh(task)
return _task_out(db, task)
db.refresh(device)
return _device_out(db, device)
def delete_task(db: Session, task_id: str) -> dict[str, Any]:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
if bool(task.collect_running):
def delete_device(db: Session, device_id: str) -> dict[str, Any]:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
if bool(device.collect_running):
raise HTTPException(status_code=409, detail="collect_running")
targets = db.query(PortTrafficTarget).filter(PortTrafficTarget.task_id == task_id).all()
targets = db.query(PortTrafficTarget).filter(PortTrafficTarget.device_id == device_id).all()
ids = [str(t.id) for t in targets]
series_ids = [str(t.series_id) for t in targets if t.series_id]
if ids:
db.query(PortTrafficSample).filter(PortTrafficSample.target_row_id.in_(ids)).delete(
synchronize_session=False
)
db.query(PortTrafficTarget).filter(PortTrafficTarget.task_id == task_id).delete(
db.query(PortTrafficTarget).filter(PortTrafficTarget.device_id == device_id).delete(
synchronize_session=False
)
db.query(PortTrafficEvent).filter(PortTrafficEvent.device_id == device_id).delete(
synchronize_session=False
)
if series_ids:
db.query(PortTrafficSeries).filter(PortTrafficSeries.id.in_(series_ids)).delete(
synchronize_session=False
)
else:
db.query(PortTrafficSeries).filter(PortTrafficSeries.task_id == task_id).delete(
db.query(PortTrafficSeries).filter(PortTrafficSeries.device_id == device_id).delete(
synchronize_session=False
)
db.delete(task)
db.delete(device)
db.commit()
return {"ok": True, "id": task_id}
return {"ok": True, "id": device_id}
def set_task_status(db: Session, task_id: str, status: str) -> PortTrafficTaskOut:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
def set_device_status(db: Session, device_id: str, status: str) -> PortTrafficDeviceOut:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
if status == "running":
active = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.task_id == task_id, PortTrafficTarget.status == "active")
.filter(PortTrafficTarget.device_id == device_id, PortTrafficTarget.status == "active")
.count()
)
if active <= 0:
raise HTTPException(status_code=400, detail="no_active_targets")
# Allow scheduler/collect-now to fire immediately after start/resume.
task.last_collect_ended_at = None
task.status = status
task.updated_at = _utcnow()
device.last_collect_ended_at = None
device.status = status
device.updated_at = _utcnow()
db.commit()
db.refresh(task)
return _task_out(db, task)
db.refresh(device)
return _device_out(db, device)
def put_targets(db: Session, task_id: str, body: PortTrafficTargetsPut) -> list[PortTrafficTargetOut]:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
if bool(task.collect_running):
def put_interfaces(db: Session, device_id: str, body: PortTrafficInterfacesPut) -> list[PortTrafficTargetOut]:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
if bool(device.collect_running):
raise HTTPException(status_code=409, detail="collect_running")
_assert_supported_targets(body.targets)
old = db.query(PortTrafficTarget).filter(PortTrafficTarget.task_id == task_id).all()
old_ids = [str(t.id) for t in old]
old_series = list({str(t.series_id) for t in old if t.series_id})
if old_ids:
db.query(PortTrafficSample).filter(PortTrafficSample.target_row_id.in_(old_ids)).delete(
synchronize_session=False
)
db.query(PortTrafficTarget).filter(PortTrafficTarget.task_id == task_id).delete(
synchronize_session=False
)
if old_series:
db.query(PortTrafficSeries).filter(PortTrafficSeries.id.in_(old_series)).delete(
synchronize_session=False
)
wanted = {i.ifname.strip(): i for i in body.interfaces if i.ifname.strip()}
_assert_ifaces_free(
db,
source=str(device.source),
ne_id=str(device.ne_id),
ifnames=list(wanted.keys()),
exclude_device_id=device_id,
)
now = _utcnow()
rows: list[PortTrafficTarget] = []
for t in body.targets:
rows.append(_create_series_and_target(db, task_id=task_id, t=t, now=now))
task.updated_at = now
existing = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.device_id == device_id)
.all()
)
by_if = {str(t.ifname): t for t in existing if str(t.status) == "active"}
for ifname, row in list(by_if.items()):
if ifname not in wanted:
row.status = "retired"
for ifname, iface in wanted.items():
if ifname in by_if:
row = by_if[ifname]
row.if_description = iface.if_description or row.if_description
if iface.bw_bps:
row.bw_bps = int(iface.bw_bps)
continue
# Reactivate retired same ifname if present
retired = next(
(
t
for t in existing
if str(t.ifname) == ifname and str(t.status) in ("retired", "disabled")
),
None,
)
if retired:
retired.status = "active"
retired.if_description = iface.if_description or retired.if_description
if iface.bw_bps:
retired.bw_bps = int(iface.bw_bps)
continue
_create_iface(db, device=device, iface=iface, now=now)
device.updated_at = now
db.commit()
return [_target_out(r) for r in rows]
return list_targets(db, device_id)
def list_targets(db: Session, task_id: str) -> list[PortTrafficTargetOut]:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
def list_targets(db: Session, device_id: str) -> list[PortTrafficTargetOut]:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
rows = (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.task_id == task_id)
.order_by(PortTrafficTarget.ne_name, PortTrafficTarget.ifname)
.filter(PortTrafficTarget.device_id == device_id)
.order_by(PortTrafficTarget.ifname)
.all()
)
return [_target_out(r) for r in rows]
def list_series(db: Session, task_id: str) -> list[PortTrafficSeriesOut]:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
def list_series(db: Session, device_id: str) -> list[PortTrafficSeriesOut]:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
rows = (
db.query(PortTrafficSeries)
.filter(PortTrafficSeries.task_id == task_id)
.filter(PortTrafficSeries.device_id == device_id)
.order_by(PortTrafficSeries.title.asc())
.all()
)
@ -317,10 +402,12 @@ def list_series(db: Session, task_id: str) -> list[PortTrafficSeriesOut]:
.filter(PortTrafficTarget.series_id == s.id, PortTrafficTarget.status == "retired")
.count()
)
did = str(s.device_id or "")
out.append(
PortTrafficSeriesOut(
id=str(s.id),
task_id=str(s.task_id),
device_id=did,
task_id=did,
title=str(s.title or ""),
status=str(s.status or ""),
active_target=_target_out(active) if active else None,
@ -333,29 +420,26 @@ def list_series(db: Session, task_id: str) -> list[PortTrafficSeriesOut]:
def replace_series_port(
db: Session,
task_id: str,
device_id: str,
series_id: str,
body: PortTrafficReplacePortRequest,
) -> PortTrafficSeriesOut:
task = db.get(PortTrafficTask, task_id)
if not task:
raise HTTPException(status_code=404, detail="task_not_found")
if bool(task.collect_running):
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
if bool(device.collect_running):
raise HTTPException(status_code=409, detail="collect_running")
series = db.get(PortTrafficSeries, series_id)
if not series or str(series.task_id) != task_id:
if not series or str(series.device_id) != device_id:
raise HTTPException(status_code=404, detail="series_not_found")
tmp = PortTrafficTargetIn(
source=body.source,
target_id=body.target_id,
ne_name=body.ne_name,
ne_ip=body.ne_ip,
vendor=body.vendor,
ifname=body.ifname,
if_description=body.if_description,
bw_bps=body.bw_bps,
ifname = body.ifname.strip()
_assert_ifaces_free(
db,
source=str(device.source),
ne_id=str(device.ne_id),
ifnames=[ifname],
exclude_device_id=device_id,
)
_assert_supported_targets([tmp])
now = _utcnow()
actives = (
db.query(PortTrafficTarget)
@ -370,7 +454,7 @@ def replace_series_port(
clash = (
db.query(PortTrafficSeries.id)
.filter(
PortTrafficSeries.task_id == task_id,
PortTrafficSeries.device_id == device_id,
PortTrafficSeries.title == wanted,
PortTrafficSeries.id != series_id,
)
@ -381,24 +465,23 @@ def replace_series_port(
series.title = wanted
row = PortTrafficTarget(
id=uuid4().hex,
task_id=task_id,
device_id=device_id,
series_id=series_id,
source=body.source,
target_id=body.target_id,
ne_name=body.ne_name or "",
ne_ip=body.ne_ip or "",
vendor=body.vendor or "",
ifname=body.ifname.strip(),
source=str(device.source),
target_id=str(device.ne_id),
ne_name=str(device.ne_name or ""),
ne_ip=str(device.ne_ip or ""),
vendor=str(device.vendor or ""),
ifname=ifname,
if_description=body.if_description or "",
bw_bps=int(body.bw_bps or 0),
status="active",
created_at=now,
)
db.add(row)
task.updated_at = now
device.updated_at = now
db.commit()
items = list_series(db, task_id)
for item in items:
for item in list_series(db, device_id):
if item.id == series_id:
return item
raise HTTPException(status_code=500, detail="series_replace_failed")
@ -449,7 +532,6 @@ def discover_ports(db: Session, body: DiscoverPortsRequest) -> DiscoverPortsResp
def _as_naive_utc(value: datetime | None) -> datetime | None:
"""Normalize query bounds to naive UTC (DB columns are naive utcnow)."""
if value is None:
return None
if value.tzinfo is None:
@ -468,15 +550,18 @@ def _sample_points(
ts = ts_raw
if align_offset is not None and ts_raw is not None:
ts = ts_raw + align_offset
in_bps = float(r.in_bps or 0)
out_bps = float(r.out_bps or 0)
bw = int(r.bw_bps or 0)
points.append(
PortTrafficSamplePoint(
ts=ts,
ts_raw=ts_raw if align_offset is not None else None,
in_bps=float(r.in_bps or 0),
out_bps=float(r.out_bps or 0),
in_util_pct=float(r.in_util_pct or 0),
out_util_pct=float(r.out_util_pct or 0),
bw_bps=int(r.bw_bps or 0),
in_bps=in_bps,
out_bps=out_bps,
in_util_pct=resolve_util_pct(float(r.in_util_pct or 0), in_bps, bw),
out_util_pct=resolve_util_pct(float(r.out_util_pct or 0), out_bps, bw),
bw_bps=bw,
rate_period_sec=int(r.rate_period_sec or 0),
)
)
@ -530,7 +615,6 @@ def compare_targets(
baseline_target_id: str | None = None,
to_ts: datetime | None = None,
) -> PortTrafficCompareOut:
"""Compare current interface samples vs period and/or manually mapped interface."""
target = db.get(PortTrafficTarget, target_row_id)
if not target:
raise HTTPException(status_code=404, detail="target_not_found")
@ -549,27 +633,25 @@ def compare_targets(
range_h = max(0.25, float(range_hours or 24))
from_ts = to_ts - timedelta(hours=range_h)
to_q = to_ts + timedelta(seconds=5)
current_rows = _query_target_samples(
db, target_row_id=str(target.id), from_ts=from_ts, to_ts=to_q
current = _sample_points(
_query_target_samples(db, target_row_id=str(target.id), from_ts=from_ts, to_ts=to_q)
)
current = _sample_points(current_rows)
off_h = baseline_offset_hours(baseline, range_h, offset_hours)
baseline_points: list[PortTrafficSamplePoint] = []
# Baseline source: mapped interface if set, else same interface (period compare only).
base_src = mapped if mapped is not None else target
want_baseline = off_h is not None or mapped is not None
if want_baseline:
if off_h is not None:
delta = timedelta(hours=off_h)
b_from = from_ts - delta
b_to = to_q - delta
base_rows = _query_target_samples(
db, target_row_id=str(base_src.id), from_ts=b_from, to_ts=b_to
db,
target_row_id=str(base_src.id),
from_ts=from_ts - delta,
to_ts=to_q - delta,
)
baseline_points = _sample_points(base_rows, align_offset=delta)
else:
# Mapped port, same window (no time shift) — cross-device overlay.
base_rows = _query_target_samples(
db, target_row_id=str(base_src.id), from_ts=from_ts, to_ts=to_q
)
@ -590,7 +672,6 @@ def compare_targets(
)
# Back-compat alias for older imports/tests.
def compare_series(db: Session, **kwargs: Any) -> PortTrafficCompareOut:
target_row_id = kwargs.pop("target_row_id", None) or kwargs.pop("target_id", None)
series_id = kwargs.pop("series_id", None)
@ -622,15 +703,14 @@ def get_samples(
now = _utcnow()
to_ts = _as_naive_utc(to_ts) or now
from_ts = _as_naive_utc(from_ts) or (to_ts - timedelta(hours=1))
# Slight skew so just-written samples are not clipped by client clock.
to_ts = to_ts + timedelta(seconds=5)
rows = _query_target_samples(db, target_row_id=target_row_id, from_ts=from_ts, to_ts=to_ts)
return PortTrafficSamplesOut(target=_target_out(target), points=_sample_points(rows))
def dashboard(db: Session) -> PortTrafficDashboardOut:
task_count = db.query(PortTrafficTask).count()
running = db.query(PortTrafficTask).filter(PortTrafficTask.status == "running").count()
device_count = db.query(PortTrafficDevice).count()
running = db.query(PortTrafficDevice).filter(PortTrafficDevice.status == "running").count()
active_targets = db.query(PortTrafficTarget).filter(PortTrafficTarget.status == "active").count()
since = _utcnow() - timedelta(hours=24)
sample_count = (
@ -640,38 +720,130 @@ def dashboard(db: Session) -> PortTrafficDashboardOut:
)
last = db.query(func.max(PortTrafficSample.ts)).scalar()
return PortTrafficDashboardOut(
task_count=int(task_count),
running_task_count=int(running),
device_count=int(device_count),
running_device_count=int(running),
active_target_count=int(active_targets),
sample_count_24h=int(sample_count),
last_sample_at=last,
task_count=int(device_count),
running_task_count=int(running),
)
def list_device_events(
db: Session,
device_id: str,
*,
limit: int = 100,
) -> PortTrafficEventsOut:
device = db.get(PortTrafficDevice, device_id)
if not device:
raise HTTPException(status_code=404, detail="device_not_found")
lim = max(1, min(500, int(limit or 100)))
q = db.query(PortTrafficEvent).filter(PortTrafficEvent.device_id == device_id)
total = q.count()
# Seed once from current last_error snapshot so older failures still show in log UI.
if total == 0:
seeded = False
if str(device.last_error or "").strip():
append_device_event(
db,
device_id=device_id,
message=str(device.last_error),
level="error",
)
seeded = True
for t in (
db.query(PortTrafficTarget)
.filter(PortTrafficTarget.device_id == device_id)
.order_by(PortTrafficTarget.ifname)
.all()
):
err = str(t.last_error or "").strip()
if not err:
continue
append_device_event(
db,
device_id=device_id,
target_row_id=str(t.id),
ifname=str(t.ifname or ""),
message=err,
level="error",
)
seeded = True
if seeded:
db.commit()
total = db.query(PortTrafficEvent).filter(PortTrafficEvent.device_id == device_id).count()
q = db.query(PortTrafficEvent).filter(PortTrafficEvent.device_id == device_id)
rows = q.order_by(PortTrafficEvent.created_at.desc()).limit(lim).all()
items = [
PortTrafficEventOut(
id=str(r.id),
device_id=str(r.device_id or ""),
target_row_id=str(r.target_row_id or ""),
ifname=str(r.ifname or ""),
level=str(r.level or "error"),
message=str(r.message or ""),
created_at=r.created_at,
)
for r in rows
]
return PortTrafficEventsOut(items=items, total=int(total))
def append_device_event(
db: Session,
*,
device_id: str,
message: str,
level: str = "error",
target_row_id: str = "",
ifname: str = "",
) -> None:
msg = str(message or "").strip()
if not msg or not device_id:
return
db.add(
PortTrafficEvent(
id=uuid4().hex,
device_id=device_id,
target_row_id=str(target_row_id or ""),
ifname=str(ifname or ""),
level=str(level or "error")[:16],
message=msg[:4000],
created_at=_utcnow(),
)
)
def purge_expired_samples(db: Session) -> int:
"""Delete samples older than each task's retention_days."""
tasks = db.query(PortTrafficTask).all()
devices = db.query(PortTrafficDevice).all()
deleted = 0
now = _utcnow()
for task in tasks:
days = max(1, int(task.retention_days or 7))
for device in devices:
days = max(1, int(device.retention_days or 7))
cutoff = now - timedelta(days=days)
target_ids = [
str(t.id)
for t in db.query(PortTrafficTarget.id).filter(PortTrafficTarget.task_id == task.id).all()
for t in db.query(PortTrafficTarget.id)
.filter(PortTrafficTarget.device_id == device.id)
.all()
]
if not target_ids:
continue
n = (
db.query(PortTrafficSample)
.filter(
PortTrafficSample.target_row_id.in_(target_ids),
PortTrafficSample.ts < cutoff,
if target_ids:
n = (
db.query(PortTrafficSample)
.filter(
PortTrafficSample.target_row_id.in_(target_ids),
PortTrafficSample.ts < cutoff,
)
.delete(synchronize_session=False)
)
.delete(synchronize_session=False)
)
deleted += int(n or 0)
deleted += int(n or 0)
db.query(PortTrafficEvent).filter(
PortTrafficEvent.device_id == device.id,
PortTrafficEvent.created_at < cutoff,
).delete(synchronize_session=False)
db.commit()
if deleted:
db.commit()
_log.info("port_traffic retention purged samples=%s", deleted)
return deleted