hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 03:10:46 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2
netx/netx_api/topology_discover_jobs.py
oliver 8ded8eeb7d Add LLDP pause/resume/stop with restart recovery, and localize ops task titles. 
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-04 22:23:44 +08:00

710 lines
25 KiB
Python
Raw Blame History

"""Discover job lifecycle: start, pause/resume/stop, background run, stale reclaim."""
from __future__ import annotations
import logging
import threading
import time
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from datetime import datetime, timedelta
from uuid import uuid4
from fastapi import HTTPException
from sqlalchemy.orm import Session
from .config import settings
from .db import SessionLocal
from .models import LldpCollectPolicy, TopoDiscoverJob, TopoDiscoverJobItem, TopoFabricStats
from .topology_common import _JOB_LOCK, _RUNNING_JOBS, _utcnow
from .topology_discover_common import (
_job_out,
_resolve_scan_targets,
prune_discover_jobs,
)
from .topology_discover_scan import _discover_one_target, _preensure_discover_targets
from .topology_fabric import (
_apply_missing_and_purge,
merge_duplicate_fabric_nodes,
refresh_fabric_stats,
)
from .topology_schemas import FabricDiscoverJobOut, FabricDiscoverRequest
_log = logging.getLogger("netx.topology.discover")
_ACTIVE_STATUSES = ("pending", "running", "paused")
_FINISHED_STATUSES = ("done", "failed", "cancelled")
def _target_key(t: dict) -> str:
ume = str(t.get("ume_ne_id") or "").strip()
if ume:
return f"ume:{ume}"
return f"managed:{str(t.get('ne_id') or '').strip()}"
def _item_key(item: TopoDiscoverJobItem) -> str:
ume = str(item.ume_ne_id or "").strip()
if ume:
return f"ume:{ume}"
return f"managed:{str(item.ne_id or '').strip()}"
def _read_job_status(db: Session, job_id: str) -> str:
job = db.get(TopoDiscoverJob, job_id)
if job is None:
return "cancelled"
return str(job.status or "")
def _request_from_job(db: Session, job: TopoDiscoverJob) -> FabricDiscoverRequest:
"""Rebuild a discover request from a stored job (resume after worker death)."""
pol = db.get(LldpCollectPolicy, 1)
concurrency = max(1, min(32, int(getattr(pol, "concurrency", None) or 4)))
auto_add = bool(getattr(pol, "auto_add_unmatched", True)) if pol is not None else True
scope = str(job.scope or "ne_ids").strip().lower() or "ne_ids"
stored = list(job.ne_ids_json or [])
managed_ids: list[str] = []
ume_ids: list[str] = []
legacy: list[str] = []
for raw in stored:
s = str(raw or "").strip()
if not s:
continue
if s.startswith("managed:"):
managed_ids.append(s[len("managed:") :])
elif s.startswith("ume:"):
ume_ids.append(s[len("ume:") :])
else:
legacy.append(s)
if scope == "all_inventory":
return FabricDiscoverRequest(
scope="all_inventory",
ne_ids=[],
managed_ne_ids=[],
ume_ne_ids=[],
concurrency=concurrency,
auto_add_unmatched=auto_add,
trigger_mode=str(job.trigger_mode or "manual"),
)
if managed_ids or ume_ids:
return FabricDiscoverRequest(
scope="ne_ids",
ne_ids=[],
managed_ne_ids=managed_ids,
ume_ne_ids=ume_ids,
concurrency=concurrency,
auto_add_unmatched=auto_add,
trigger_mode=str(job.trigger_mode or "manual"),
)
return FabricDiscoverRequest(
scope="ne_ids",
ne_ids=legacy,
managed_ne_ids=[],
ume_ne_ids=[],
concurrency=concurrency,
auto_add_unmatched=auto_add,
trigger_mode=str(job.trigger_mode or "manual"),
)
def _record_item(
db: Session,
job: TopoDiscoverJob,
job_id: str,
result: dict,
*,
added: int,
updated: int,
) -> tuple[int, int]:
item = TopoDiscoverJobItem(
id=uuid4().hex,
job_id=job_id,
ne_id=str(result.get("ne_id") or ""),
ume_ne_id=str(result.get("ume_ne_id") or ""),
fabric_node_id=str(result.get("fabric_node_id") or ""),
ne_name=str(result.get("ne_name") or "")[:256],
ne_ip=str(result.get("ne_ip") or "")[:128],
ok=bool(result.get("ok")),
command=str(result.get("command") or "")[:256],
neighbors=int(result.get("neighbors") or 0),
edges_added=int(result.get("edges_added") or 0),
edges_updated=int(result.get("edges_updated") or 0),
unmatched_count=int(result.get("unmatched_count") or 0),
unmatched_json=list(result.get("unmatched") or []),
parser_key=str(result.get("parser_key") or "")[:64],
parser_stub=bool(result.get("parser_stub")),
error=str(result.get("error") or "")[:1024],
raw_preview=str(result.get("raw_preview") or ""),
created_at=_utcnow(),
)
db.add(item)
added += int(result.get("edges_added") or 0)
updated += int(result.get("edges_updated") or 0)
job.done = int(job.done or 0) + 1
job.edges_added = added
job.edges_updated = updated
job.updated_at = _utcnow()
db.commit()
if int(job.done or 0) % 50 == 0:
try:
refresh_fabric_stats(db)
except Exception: # noqa: BLE001
db.rollback()
return added, updated
def _finalize_success(
db: Session,
job_id: str,
*,
added: int,
updated: int,
stale: int,
) -> None:
stats = db.get(TopoFabricStats, "global")
if stats is None:
stats = TopoFabricStats(id="global")
db.add(stats)
stats.last_discover_at = _utcnow()
db.commit()
merge_duplicate_fabric_nodes(db)
refresh_fabric_stats(db)
job = db.get(TopoDiscoverJob, job_id)
if job is None:
return
# Stop/pause may have won the race while we were finishing.
if str(job.status or "") in ("cancelled", "paused"):
return
job.status = "done"
job.ended_at = _utcnow()
job.updated_at = job.ended_at
job.edges_added = added
job.edges_updated = updated
job.edges_stale = stale
db.commit()
try:
from .lldp_collect_service import DEFAULT_HISTORY_KEEP, ensure_policy
keep = int(getattr(ensure_policy(db), "history_keep", DEFAULT_HISTORY_KEEP) or 0)
prune_discover_jobs(db, keep=keep)
except Exception: # noqa: BLE001
_log.warning("prune_discover_jobs failed job=%s", job_id, exc_info=True)
def _finalize_cancelled(db: Session, job_id: str, *, added: int, updated: int) -> None:
try:
refresh_fabric_stats(db)
except Exception: # noqa: BLE001
db.rollback()
job = db.get(TopoDiscoverJob, job_id)
if job is None:
return
if str(job.status or "") != "cancelled":
job.status = "cancelled"
job.error = (str(job.error or "").strip() or "stopped_by_user")[:1024]
job.ended_at = _utcnow()
job.updated_at = job.ended_at
job.edges_added = added
job.edges_updated = updated
db.commit()
try:
from .lldp_collect_service import DEFAULT_HISTORY_KEEP, ensure_policy
keep = int(getattr(ensure_policy(db), "history_keep", DEFAULT_HISTORY_KEEP) or 0)
prune_discover_jobs(db, keep=keep)
except Exception: # noqa: BLE001
_log.warning("prune_discover_jobs failed job=%s", job_id, exc_info=True)
def _run_discover_job(
job_id: str,
body: FabricDiscoverRequest,
*,
resume: bool = False,
) -> None:
db = SessionLocal()
try:
job = db.get(TopoDiscoverJob, job_id)
if job is None:
return
if str(job.status or "") == "cancelled":
return
if str(job.status or "") != "paused":
job.status = "running"
if not job.started_at:
job.started_at = _utcnow()
job.updated_at = _utcnow()
try:
all_targets = _resolve_scan_targets(db, body)
except HTTPException as exc:
job.status = "failed"
job.error = str(exc.detail or "resolve_failed")[:1024]
job.ended_at = _utcnow()
job.updated_at = job.ended_at
db.commit()
return
prior_items = (
db.query(TopoDiscoverJobItem).filter(TopoDiscoverJobItem.job_id == job_id).all()
)
done_keys = {_item_key(it) for it in prior_items if _item_key(it) not in {"managed:", "ume:"}}
scanned_ok: set[str] = {
str(it.fabric_node_id)
for it in prior_items
if it.ok and str(it.fabric_node_id or "").strip()
}
touched_edges: set[str] = set()
# After worker death we lost in-memory touched edges — skip miss to avoid false marks.
skip_miss = bool(resume)
if not prior_items:
job.total = len(all_targets)
elif int(job.total or 0) <= 0:
job.total = len(all_targets)
db.commit()
targets = [t for t in all_targets if _target_key(t) not in done_keys]
try:
_preensure_discover_targets(db, targets)
except Exception: # noqa: BLE001
db.rollback()
from .cli_budget import clamp_cli_workers
concurrency = clamp_cli_workers(int(body.concurrency or 4))
added = int(job.edges_added or 0)
updated = int(job.edges_updated or 0)
stale = int(job.edges_stale or 0)
remaining = list(targets)
in_flight: dict = {}
cancelled = False
with ThreadPoolExecutor(max_workers=concurrency) as pool:
while remaining or in_flight:
db.expire_all()
status = _read_job_status(db, job_id)
if status == "cancelled":
cancelled = True
for fut in list(in_flight):
fut.cancel()
# Drain in-flight that already started (cancel is best-effort).
while in_flight:
done_set, _ = wait(in_flight.keys(), return_when=FIRST_COMPLETED)
for fut in done_set:
tgt = in_flight.pop(fut, None)
if fut.cancelled():
continue
try:
result = fut.result()
except Exception as exc: # noqa: BLE001
result = {
"ne_id": str((tgt or {}).get("ne_id") or ""),
"ume_ne_id": str((tgt or {}).get("ume_ne_id") or ""),
"ne_name": str((tgt or {}).get("ne_name") or ""),
"ne_ip": str((tgt or {}).get("ne_ip") or ""),
"ok": False,
"error": str(exc)[:1024],
}
job = db.get(TopoDiscoverJob, job_id)
if job is None:
break
added, updated = _record_item(
db, job, job_id, result, added=added, updated=updated
)
break
if status == "paused":
if not in_flight:
time.sleep(0.5)
continue
# Let in-flight finish, but do not submit more.
elif status in ("running", "pending"):
if status == "pending":
job = db.get(TopoDiscoverJob, job_id)
if job is not None:
job.status = "running"
job.updated_at = _utcnow()
db.commit()
while remaining and len(in_flight) < concurrency:
if _read_job_status(db, job_id) not in ("running", "pending"):
break
t = remaining.pop(0)
fut = pool.submit(
_discover_one_target,
t,
auto_add_unmatched=bool(body.auto_add_unmatched),
)
in_flight[fut] = t
else:
# Unexpected terminal status.
cancelled = status == "cancelled"
break
if not in_flight:
if status == "paused":
continue
break
done_set, _ = wait(in_flight.keys(), timeout=0.5, return_when=FIRST_COMPLETED)
if not done_set:
continue
for fut in done_set:
tgt = in_flight.pop(fut, None)
if fut.cancelled():
continue
try:
result = fut.result()
except Exception as exc: # noqa: BLE001
result = {
"ne_id": str((tgt or {}).get("ne_id") or ""),
"ume_ne_id": str((tgt or {}).get("ume_ne_id") or ""),
"ne_name": str((tgt or {}).get("ne_name") or ""),
"ne_ip": str((tgt or {}).get("ne_ip") or ""),
"ok": False,
"error": str(exc)[:1024],
}
job = db.get(TopoDiscoverJob, job_id)
if job is None:
cancelled = True
break
added, updated = _record_item(
db, job, job_id, result, added=added, updated=updated
)
if result.get("ok") and result.get("scanned_node_id"):
scanned_ok.add(str(result["scanned_node_id"]))
for eid in result.get("touched_edge_ids") or []:
touched_edges.add(str(eid))
for eid in result.get("replaced_edge_ids") or []:
touched_edges.add(str(eid))
if cancelled or _read_job_status(db, job_id) == "cancelled":
_finalize_cancelled(db, job_id, added=added, updated=updated)
return
# May still be paused with no remaining work — treat as done.
db.expire_all()
status = _read_job_status(db, job_id)
if status == "paused" and remaining:
# Worker exiting while paused with work left — keep paused for later resume.
job = db.get(TopoDiscoverJob, job_id)
if job is not None:
job.edges_added = added
job.edges_updated = updated
job.updated_at = _utcnow()
db.commit()
return
if scanned_ok and not skip_miss:
newly_missing, purged = _apply_missing_and_purge(
db,
scanned_ok=scanned_ok,
touched_edge_ids=touched_edges,
)
stale = newly_missing + purged
job = db.get(TopoDiscoverJob, job_id)
if job is not None:
job.edges_stale = stale
db.commit()
_finalize_success(db, job_id, added=added, updated=updated, stale=stale)
except Exception as exc: # noqa: BLE001
db.rollback()
job = db.get(TopoDiscoverJob, job_id)
if job is not None and str(job.status or "") not in ("cancelled", "paused"):
job.status = "failed"
job.error = str(exc)[:1024]
job.ended_at = _utcnow()
job.updated_at = job.ended_at
db.commit()
finally:
db.close()
with _JOB_LOCK:
_RUNNING_JOBS.discard(job_id)
def reclaim_stale_discover_jobs(
db: Session,
*,
force_all_open: bool = False,
now: datetime | None = None,
) -> int:
"""Mark orphaned / hung discover jobs as failed so scheduling can proceed.
- ``force_all_open``: process restart — pending/running rows are dead (paused kept for resume).
- Otherwise: pending older than pending_stale_sec, or running with stale updated_at.
"""
now = now or _utcnow()
run_sec = max(60, int(getattr(settings, "lldp_collect_stale_run_sec", 7200) or 7200))
pend_sec = max(30, int(getattr(settings, "lldp_collect_pending_stale_sec", 300) or 300))
open_jobs = (
db.query(TopoDiscoverJob)
.filter(TopoDiscoverJob.status.in_(["pending", "running"]))
.all()
)
if not open_jobs:
return 0
closed = 0
for job in open_jobs:
status = str(job.status or "")
if force_all_open:
reason = "stale_running_reset_on_startup"
elif status == "pending":
created = job.created_at or job.updated_at or now
if created > now - timedelta(seconds=pend_sec):
continue
reason = "pending_stale_timeout"
else:
touched = job.updated_at or job.started_at or job.created_at or now
if touched > now - timedelta(seconds=run_sec):
continue
reason = "running_stale_timeout"
job.status = "failed"
job.ended_at = now
job.updated_at = now
msg = str(job.error or "").strip()
job.error = (msg + ("; " if msg else "") + reason)[:1024]
closed += 1
with _JOB_LOCK:
_RUNNING_JOBS.discard(job.id)
if closed:
db.commit()
return closed
def start_discover_job(
db: Session,
body: FabricDiscoverRequest,
*,
trigger_mode: str = "manual",
) -> FabricDiscoverJobOut:
reclaim_stale_discover_jobs(db)
# Serialize multi-worker starts via singleton policy row lock (PG/SQLite FOR UPDATE).
pol = db.get(LldpCollectPolicy, 1)
if pol is None:
pol = LldpCollectPolicy(
id=1,
enabled=False,
interval_days=1,
interval_hours=24,
concurrency=4,
scope_mode="all",
selected_targets=[],
auto_add_unmatched=True,
history_keep=30,
updated_at=_utcnow(),
)
db.add(pol)
db.commit()
db.query(LldpCollectPolicy).filter(LldpCollectPolicy.id == 1).with_for_update().one()
if (
db.query(TopoDiscoverJob)
.filter(TopoDiscoverJob.status.in_(list(_ACTIVE_STATUSES)))
.first()
is not None
):
raise HTTPException(status_code=409, detail="lldp_collect_already_running")
scope = str(body.scope or "ne_ids").strip().lower() or "ne_ids"
if scope not in {"all_inventory", "ne_ids"}:
raise HTTPException(status_code=400, detail="invalid_scope")
trig = str(trigger_mode or getattr(body, "trigger_mode", None) or "manual").strip().lower() or "manual"
if trig not in {"manual", "schedule", "topology"}:
trig = "manual"
now = _utcnow()
# Persist explicit source lists when present; keep legacy ne_ids for older clients.
stored_ids = list(body.ne_ids or [])
if body.managed_ne_ids or body.ume_ne_ids:
stored_ids = [
*(f"managed:{x}" for x in (body.managed_ne_ids or []) if str(x).strip()),
*(f"ume:{x}" for x in (body.ume_ne_ids or []) if str(x).strip()),
]
job = TopoDiscoverJob(
id=uuid4().hex,
scope=scope,
trigger_mode=trig,
ne_ids_json=stored_ids,
status="pending",
total=0,
done=0,
created_at=now,
updated_at=now,
)
db.add(job)
db.commit()
db.refresh(job)
with _JOB_LOCK:
_RUNNING_JOBS.add(job.id)
thread = threading.Thread(
target=_run_discover_job,
args=(job.id, body),
name=f"topo-discover-{job.id[:8]}",
daemon=True,
)
thread.start()
return _job_out(db, job, include_items=False)
def pause_discover_job(db: Session, job_id: str) -> FabricDiscoverJobOut:
job = db.get(TopoDiscoverJob, str(job_id or "").strip())
if job is None:
raise HTTPException(status_code=404, detail="discover_job_not_found")
if str(job.status or "") not in ("running", "pending"):
raise HTTPException(status_code=400, detail="job_not_running")
job.status = "paused"
job.updated_at = _utcnow()
db.commit()
db.refresh(job)
return _job_out(db, job, include_items=False)
def resume_discover_job(db: Session, job_id: str) -> FabricDiscoverJobOut:
job = db.get(TopoDiscoverJob, str(job_id or "").strip())
if job is None:
raise HTTPException(status_code=404, detail="discover_job_not_found")
if str(job.status or "") != "paused":
raise HTTPException(status_code=400, detail="job_not_paused")
other = (
db.query(TopoDiscoverJob)
.filter(
TopoDiscoverJob.id != job.id,
TopoDiscoverJob.status.in_(["pending", "running"]),
)
.first()
)
if other is not None:
raise HTTPException(status_code=409, detail="lldp_collect_already_running")
remaining = int(job.total or 0) - int(job.done or 0)
if int(job.total or 0) > 0 and remaining <= 0:
raise HTTPException(status_code=400, detail="no_pending_targets")
job.status = "running"
job.updated_at = _utcnow()
db.commit()
db.refresh(job)
with _JOB_LOCK:
alive = job.id in _RUNNING_JOBS
if alive:
return _job_out(db, job, include_items=False)
body = _request_from_job(db, job)
with _JOB_LOCK:
_RUNNING_JOBS.add(job.id)
thread = threading.Thread(
target=_run_discover_job,
args=(job.id, body),
kwargs={"resume": True},
name=f"topo-discover-{job.id[:8]}",
daemon=True,
)
thread.start()
return _job_out(db, job, include_items=False)
def stop_discover_job(db: Session, job_id: str) -> FabricDiscoverJobOut:
"""Cancel remaining work and close the job (running/paused/pending)."""
job = db.get(TopoDiscoverJob, str(job_id or "").strip())
if job is None:
raise HTTPException(status_code=404, detail="discover_job_not_found")
if str(job.status or "") not in _ACTIVE_STATUSES:
raise HTTPException(status_code=400, detail="job_not_active")
now = _utcnow()
job.status = "cancelled"
job.error = "stopped_by_user"
job.ended_at = now
job.updated_at = now
db.commit()
db.refresh(job)
# If no worker is attached (paused after restart), close immediately for clients.
with _JOB_LOCK:
alive = job.id in _RUNNING_JOBS
if not alive:
try:
refresh_fabric_stats(db)
except Exception: # noqa: BLE001
db.rollback()
try:
from .lldp_collect_service import DEFAULT_HISTORY_KEEP, ensure_policy
keep = int(getattr(ensure_policy(db), "history_keep", DEFAULT_HISTORY_KEEP) or 0)
prune_discover_jobs(db, keep=keep)
except Exception: # noqa: BLE001
_log.warning("prune_discover_jobs after stop failed job=%s", job_id, exc_info=True)
return _job_out(db, job, include_items=False)
def recover_lldp_discover_on_startup(db: Session) -> int:
"""Resume interrupted LLDP discover after process restart (config-sync style).
- Keep the newest active job; mark older actives failed.
- ``paused`` stays paused (no auto dispatch) but still occupies the slot.
- ``pending`` / ``running`` are re-spawned with ``resume=True`` for remaining targets.
"""
actives = (
db.query(TopoDiscoverJob)
.filter(TopoDiscoverJob.status.in_(list(_ACTIVE_STATUSES)))
.order_by(TopoDiscoverJob.created_at.asc())
.all()
)
if not actives:
return 0
primary = actives[-1]
now = _utcnow()
for stale in actives[:-1]:
_log.warning(
"lldp discover recovery closing older active job=%s (keep=%s)",
stale.id,
primary.id,
)
stale.status = "failed"
stale.ended_at = now
stale.updated_at = now
msg = str(stale.error or "").strip()
stale.error = (msg + ("; " if msg else "") + "superseded_active_job")[:1024]
with _JOB_LOCK:
_RUNNING_JOBS.discard(stale.id)
db.commit()
db.refresh(primary)
if str(primary.status or "") == "paused":
_log.info("lldp discover recovery job=%s stays paused (blocks new jobs)", primary.id)
return 0
total = int(primary.total or 0)
done = int(primary.done or 0)
if total > 0 and done >= total:
primary.status = "done"
primary.ended_at = now
primary.updated_at = now
db.commit()
_log.info("lldp discover recovery job=%s already complete", primary.id)
return 0
primary.status = "running"
if not primary.started_at:
primary.started_at = now
primary.updated_at = now
db.commit()
db.refresh(primary)
body = _request_from_job(db, primary)
with _JOB_LOCK:
if primary.id in _RUNNING_JOBS:
_log.info("lldp discover recovery job=%s already has worker", primary.id)
return 0
_RUNNING_JOBS.add(primary.id)
thread = threading.Thread(
target=_run_discover_job,
args=(primary.id, body),
kwargs={"resume": True},
name=f"topo-discover-{primary.id[:8]}",
daemon=True,
)
thread.start()
_log.info(
"lldp discover recovery resumed job=%s done=%s/%s",
primary.id,
done,
total,
)
return 1
Reference in a new issue View git blame Copy permalink