diff --git a/netx_api/config.py b/netx_api/config.py index 4c1598a..948ec32 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -72,6 +72,9 @@ class Settings(BaseSettings): ne_collect_pending_stale_sec: int = 180 ne_collect_run_timeout_cap_sec: int = 600 ne_collection_data_dir: str = "data/ne_collections" + # Config sync (periodic running-config backup into DB) + config_sync_scheduler_enabled: bool = True + config_sync_scheduler_tick_sec: int = 60 # Managed NE exec: max CLI commands per request (lab can raise; hard-capped in ne_exec). ne_exec_max_commands: int = 5 # WebCRT interactive terminal sessions diff --git a/netx_api/config_sync_codec.py b/netx_api/config_sync_codec.py new file mode 100644 index 0000000..a87a411 --- /dev/null +++ b/netx_api/config_sync_codec.py @@ -0,0 +1,20 @@ +"""zlib + sha256 helpers for config blobs.""" + +from __future__ import annotations + +import hashlib +import zlib + + +def compress_text(text: str) -> tuple[bytes, str, int, int]: + """Return (zlib_bytes, sha256_hex, plain_size, zlib_size).""" + raw = str(text or "").encode("utf-8", errors="replace") + digest = hashlib.sha256(raw).hexdigest() + blob = zlib.compress(raw, level=6) + return blob, digest, len(raw), len(blob) + + +def decompress_text(blob: bytes | None) -> str: + if not blob: + return "" + return zlib.decompress(bytes(blob)).decode("utf-8", errors="replace") diff --git a/netx_api/config_sync_commands.py b/netx_api/config_sync_commands.py new file mode 100644 index 0000000..8e9cfa4 --- /dev/null +++ b/netx_api/config_sync_commands.py @@ -0,0 +1,59 @@ +"""Vendor → config-collection CLI command matrix.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ConfigCommands: + """primary → config_zlib; alt → config_alt_zlib (Juniper only).""" + + primary: str + alt: str | None = None + vendor_key: str = "other" + + +def normalize_vendor_key(vendor: str, device_type: str = "") -> str: + blob = f"{vendor} {device_type}".strip().lower() + if "juniper" in blob or "junos" in blob: + return "juniper" + if "nokia" in blob or "alcatel" in blob or "sros" in blob or "tiomos" in blob: + return "nokia" + if "ericsson" in blob: + return "ericsson" + if "huawei" in blob or "vrp" in blob: + return "huawei" + if "h3c" in blob or "comware" in blob: + return "h3c" + if "zte" in blob or "zxros" in blob: + return "zte" + if "cisco" in blob or "ios" in blob or "nx-os" in blob or "xr" in blob: + return "cisco" + return "other" + + +def commands_for_vendor(vendor: str, device_type: str = "") -> ConfigCommands | None: + key = normalize_vendor_key(vendor, device_type) + if key in ("cisco", "zte"): + return ConfigCommands(primary="show running-config", vendor_key=key) + if key in ("huawei", "h3c"): + return ConfigCommands(primary="display current-configuration", vendor_key=key) + if key == "juniper": + return ConfigCommands( + primary="show configuration | display set", + alt="show configuration | no-more", + vendor_key=key, + ) + if key == "nokia": + return ConfigCommands(primary="admin display-config", vendor_key=key) + if key == "ericsson": + return ConfigCommands(primary="show configuration", vendor_key=key) + return None + + +def command_list(cmds: ConfigCommands) -> list[str]: + out = [cmds.primary] + if cmds.alt: + out.append(cmds.alt) + return out diff --git a/netx_api/config_sync_recovery.py b/netx_api/config_sync_recovery.py new file mode 100644 index 0000000..8323b18 --- /dev/null +++ b/netx_api/config_sync_recovery.py @@ -0,0 +1,68 @@ +"""Startup recovery for interrupted config-sync cycles.""" + +from __future__ import annotations + +import logging + +from sqlalchemy.orm import Session + +from .config_sync_runner import dispatch_cycle +from .config_sync_service import finalize_cycle, sync_cycle_progress +from .models import ConfigSyncCycle, ConfigSyncTask +from datetime import datetime + +_log = logging.getLogger("netx.config_sync.recovery") + + +def recover_config_sync_on_startup(db: Session) -> int: + """ + Mark orphaned running tasks as fail(orphan_recovered), then resume pending + tasks for cycles still marked running/paused. + """ + cycles = ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.status.in_(("running", "paused", "pending"))) + .all() + ) + resumed = 0 + for cycle in cycles: + cycle_id = str(cycle.id) + orphans = ( + db.query(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "running") + .all() + ) + for task in orphans: + task.status = "fail" + task.message = "orphan_recovered" + task.ended_at = datetime.utcnow() + if orphans: + db.commit() + _log.info("config_sync recovery cycle=%s orphaned_tasks=%s", cycle_id, len(orphans)) + + sync_cycle_progress(db, cycle_id) + db.refresh(cycle) + + if str(cycle.status) == "paused": + continue + + pending = ( + db.query(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "pending") + .count() + ) + if pending <= 0: + if str(cycle.status) in ("running", "pending"): + finalize_cycle(db, cycle_id) + continue + + if str(cycle.status) == "pending": + cycle.status = "running" + if not cycle.started_at: + cycle.started_at = datetime.utcnow() + db.commit() + + n = dispatch_cycle(cycle_id) + resumed += n + _log.info("config_sync recovery resumed cycle=%s pending=%s", cycle_id, n) + return resumed diff --git a/netx_api/config_sync_router.py b/netx_api/config_sync_router.py new file mode 100644 index 0000000..234827f --- /dev/null +++ b/netx_api/config_sync_router.py @@ -0,0 +1,168 @@ +"""HTTP API for config sync.""" + +from __future__ import annotations + +from fastapi import APIRouter, BackgroundTasks, Depends, Query, Request +from sqlalchemy.orm import Session + +from .auth_service import write_audit +from .config_sync_runner import dispatch_cycle +from .config_sync_schemas import ConfigSyncCycleCreate, ConfigSyncPolicyUpdate +from .config_sync_service import ( + create_cycle, + dashboard, + get_cycle, + get_policy, + get_snapshot_detail, + list_cycle_tasks, + list_cycles, + list_snapshot_history, + list_snapshots, + pause_cycle, + resume_cycle, + update_policy, +) +from .db import get_db + +router = APIRouter(prefix="/v1/config-sync", tags=["config-sync"]) + + +def _actor(request: Request) -> tuple[str, str]: + user = getattr(request.state, "auth_user", None) + if not user: + return "", "" + return str(getattr(user, "id", "") or ""), str(getattr(user, "username", "") or "") + + +@router.get("/policy") +def api_get_policy(db: Session = Depends(get_db)): + return get_policy(db).model_dump() + + +@router.put("/policy") +def api_put_policy( + body: ConfigSyncPolicyUpdate, + request: Request, + db: Session = Depends(get_db), +): + out = update_policy(db, body) + uid, uname = _actor(request) + write_audit( + db, + action="config_sync.policy.update", + actor_user_id=uid, + actor_username=uname, + method="PUT", + path="/v1/config-sync/policy", + status_code=200, + detail=body.model_dump(exclude_unset=True), + ) + return out.model_dump() + + +@router.get("/dashboard") +def api_dashboard(db: Session = Depends(get_db)): + return dashboard(db).model_dump() + + +@router.get("/cycles") +def api_list_cycles( + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +): + return list_cycles(db, page=page, page_size=page_size) + + +@router.post("/cycles") +def api_create_cycle( + body: ConfigSyncCycleCreate, + background_tasks: BackgroundTasks, + request: Request, + db: Session = Depends(get_db), +): + out = create_cycle(db, body) + background_tasks.add_task(dispatch_cycle, out.id) + uid, uname = _actor(request) + action = "config_sync.retry_failed" if body.mode == "retry_failed" else "config_sync.start" + write_audit( + db, + action=action, + actor_user_id=uid, + actor_username=uname, + method="POST", + path="/v1/config-sync/cycles", + status_code=200, + detail={"mode": body.mode, "cycle_id": out.id}, + ) + return out.model_dump() + + +@router.get("/cycles/{cycle_id}") +def api_get_cycle(cycle_id: str, db: Session = Depends(get_db)): + return get_cycle(db, cycle_id).model_dump() + + +@router.get("/cycles/{cycle_id}/tasks") +def api_list_cycle_tasks( + cycle_id: str, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=200), + status: str = Query(default=""), + keyword: str = Query(default=""), + db: Session = Depends(get_db), +): + return list_cycle_tasks( + db, cycle_id, page=page, page_size=page_size, status=status, keyword=keyword + ) + + +@router.post("/cycles/{cycle_id}/pause") +def api_pause_cycle(cycle_id: str, db: Session = Depends(get_db)): + return pause_cycle(db, cycle_id).model_dump() + + +@router.post("/cycles/{cycle_id}/resume") +def api_resume_cycle( + cycle_id: str, + background_tasks: BackgroundTasks, + db: Session = Depends(get_db), +): + out = resume_cycle(db, cycle_id) + background_tasks.add_task(dispatch_cycle, out.id) + return out.model_dump() + + +@router.get("/snapshots") +def api_list_snapshots( + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + keyword: str = Query(default=""), + source: str = Query(default=""), + vendor: str = Query(default=""), + db: Session = Depends(get_db), +): + return list_snapshots( + db, page=page, page_size=page_size, keyword=keyword, source=source, vendor=vendor + ) + + +@router.get("/snapshots/{source}/{target_id}") +def api_get_snapshot( + source: str, + target_id: str, + field: str = Query(default="both"), + db: Session = Depends(get_db), +): + return get_snapshot_detail(db, source, target_id, field=field).model_dump() + + +@router.get("/snapshots/{source}/{target_id}/history") +def api_snapshot_history( + source: str, + target_id: str, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), +): + return list_snapshot_history(db, source, target_id, page=page, page_size=page_size) diff --git a/netx_api/config_sync_runner.py b/netx_api/config_sync_runner.py new file mode 100644 index 0000000..4167ad1 --- /dev/null +++ b/netx_api/config_sync_runner.py @@ -0,0 +1,429 @@ +"""Config sync worker: claim tasks, collect vendor configs, store zlib snapshots.""" + +from __future__ import annotations + +import logging +import time +from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeout +from datetime import datetime +from threading import Lock +from typing import Any +from uuid import uuid4 + +from fastapi import HTTPException + +from .cli_resolve import resolve_cli_target +from .config import settings +from .config_sync_codec import compress_text +from .config_sync_commands import command_list, commands_for_vendor +from .config_sync_service import finalize_cycle, sync_cycle_progress +from .db import SessionLocal +from .models import ConfigSyncCycle, ConfigSyncPolicy, ConfigSyncTask, NeConfigHistory, NeConfigSnapshot +from .ne_session_factory import close_netmiko_connection, open_netmiko_connection + +_log = logging.getLogger("netx.config_sync.runner") +_pools: dict[str, ThreadPoolExecutor] = {} +_pools_lock = Lock() + + +def _utcnow() -> datetime: + return datetime.utcnow() + + +def _format_error(exc: BaseException) -> str: + head = f"{type(exc).__name__}: {exc}" + return head[:1020] + + +def _pool_for_cycle(cycle_id: str, concurrency: int) -> ThreadPoolExecutor: + with _pools_lock: + pool = _pools.get(cycle_id) + if pool is None: + workers = max(1, min(30, int(concurrency or 5))) + pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix=f"cfg-sync-{cycle_id[:8]}") + _pools[cycle_id] = pool + return pool + + +def _release_pool(cycle_id: str) -> None: + with _pools_lock: + pool = _pools.pop(cycle_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("config_sync pool shutdown failed cycle=%s", cycle_id) + + +def _update_task(task_id: str, **fields: Any) -> None: + db = SessionLocal() + try: + row = db.get(ConfigSyncTask, task_id) + if not row: + return + for key, val in fields.items(): + setattr(row, key, val) + db.commit() + finally: + db.close() + + +def _update_task(task_id: str, **fields: Any) -> None: + for attempt in range(10): + db = SessionLocal() + try: + task = db.get(ConfigSyncTask, task_id) + cycle = db.get(ConfigSyncCycle, cycle_id) + if not task or not cycle: + time.sleep(0.05 * (attempt + 1)) + continue + cycle_status = str(cycle.status or "") + task_status = str(task.status or "") + if cycle_status == "paused": + return False + if cycle_status != "running": + return False + if task_status == "running": + return False + if task_status != "pending": + return False + task.status = "running" + task.message = "collecting" + task.started_at = _utcnow() + db.commit() + return True + finally: + db.close() + time.sleep(0.05 * (attempt + 1)) + return False + + +def _collect_commands(creds: dict[str, Any], commands: list[str]) -> list[str]: + per_cmd = int(settings.ne_collect_read_timeout_sec or 120) + session_timeout = per_cmd * max(1, len(commands)) + 60 + conn = open_netmiko_connection(creds, session_timeout=session_timeout) + try: + outputs: list[str] = [] + for command in commands: + out = conn.send_command(command_string=command, read_timeout=per_cmd) + outputs.append(str(out or "")) + return outputs + finally: + close_netmiko_connection(conn) + + +def _collect_with_timeout(creds: dict[str, Any], commands: list[str]) -> list[str]: + per_cmd = int(settings.ne_collect_read_timeout_sec or 120) + cap = int(settings.ne_collect_run_timeout_cap_sec or 600) + budget = min(cap, per_cmd * max(1, len(commands)) + 90) + with ThreadPoolExecutor(max_workers=1) as pool: + fut = pool.submit(_collect_commands, creds, commands) + try: + return fut.result(timeout=budget) + except FuturesTimeout as exc: + raise TimeoutError(f"config_sync_timeout ({budget}s)") from exc + + +def _history_keep(db) -> int: + pol = db.get(ConfigSyncPolicy, 1) + if not pol: + return 3 + return max(0, min(30, int(pol.history_keep if pol.history_keep is not None else 3))) + + +def _save_success_snapshot( + *, + source: str, + target_id: str, + vendor: str, + device_type: str, + ne_name: str, + ne_ip: str, + primary_text: str, + alt_text: str | None, + commands: list[str], + cycle_id: str, + task_id: str, +) -> None: + primary_blob, primary_sha, plain_size, zlib_size = compress_text(primary_text) + alt_blob = None + alt_sha = "" + plain_alt = 0 + zlib_alt = 0 + if alt_text is not None: + alt_blob, alt_sha, plain_alt, zlib_alt = compress_text(alt_text) + + db = SessionLocal() + try: + existing = db.get(NeConfigSnapshot, {"source": source, "target_id": target_id}) + changed = True + if existing is not None: + changed = ( + str(existing.config_sha256 or "") != primary_sha + or str(existing.config_alt_sha256 or "") != alt_sha + ) + if changed: + keep = _history_keep(db) + if keep > 0: + # Archive previous successful snapshot before overwrite. + db.add( + NeConfigHistory( + id=uuid4().hex, + source=existing.source, + target_id=existing.target_id, + vendor=str(existing.vendor or ""), + device_type=str(existing.device_type or ""), + ne_name=str(existing.ne_name or ""), + ne_ip=str(existing.ne_ip or ""), + config_zlib=existing.config_zlib or b"", + config_alt_zlib=existing.config_alt_zlib, + config_sha256=str(existing.config_sha256 or ""), + config_alt_sha256=str(existing.config_alt_sha256 or ""), + plain_size=int(existing.plain_size or 0), + plain_alt_size=int(existing.plain_alt_size or 0), + zlib_size=int(existing.zlib_size or 0), + zlib_alt_size=int(existing.zlib_alt_size or 0), + commands_json=existing.commands_json if isinstance(existing.commands_json, list) else [], + collected_at=existing.collected_at or _utcnow(), + cycle_id=str(existing.last_cycle_id or ""), + task_id=str(existing.last_task_id or ""), + ) + ) + db.flush() + old_rows = ( + db.query(NeConfigHistory) + .filter(NeConfigHistory.source == source, NeConfigHistory.target_id == target_id) + .order_by(NeConfigHistory.collected_at.desc()) + .all() + ) + for stale in old_rows[keep:]: + db.delete(stale) + + existing.vendor = vendor + existing.device_type = device_type + existing.ne_name = ne_name + existing.ne_ip = ne_ip + existing.config_zlib = primary_blob + existing.config_alt_zlib = alt_blob + existing.config_sha256 = primary_sha + existing.config_alt_sha256 = alt_sha + existing.plain_size = plain_size + existing.plain_alt_size = plain_alt + existing.zlib_size = zlib_size + existing.zlib_alt_size = zlib_alt + existing.commands_json = list(commands) + existing.collected_at = _utcnow() + existing.last_cycle_id = cycle_id + existing.last_task_id = task_id + else: + db.add( + NeConfigSnapshot( + source=source, + target_id=target_id, + vendor=vendor, + device_type=device_type, + ne_name=ne_name, + ne_ip=ne_ip, + config_zlib=primary_blob, + config_alt_zlib=alt_blob, + config_sha256=primary_sha, + config_alt_sha256=alt_sha, + plain_size=plain_size, + plain_alt_size=plain_alt, + zlib_size=zlib_size, + zlib_alt_size=zlib_alt, + commands_json=list(commands), + collected_at=_utcnow(), + last_cycle_id=cycle_id, + last_task_id=task_id, + ) + ) + db.commit() + finally: + db.close() + + +def _run_single(cycle_id: str, task_id: str) -> None: + if not _claim_task(cycle_id, task_id): + db = SessionLocal() + try: + sync_cycle_progress(db, cycle_id) + finalize_cycle(db, cycle_id) + finally: + db.close() + return + + db = SessionLocal() + try: + task = db.get(ConfigSyncTask, task_id) + if not task: + return + source = str(task.source or "").strip().lower() + target_id = str(task.target_id or "").strip() + vendor_hint = str(task.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: + _update_task(task_id, status="fail", message="invalid_source", ended_at=_utcnow()) + return + except HTTPException as exc: + detail = str(exc.detail) if exc.detail else "resolve_failed" + _update_task(task_id, status="fail", message=detail[:1020], ended_at=_utcnow()) + return + except Exception as exc: + _update_task(task_id, status="fail", message=_format_error(exc), ended_at=_utcnow()) + return + + vendor = str(device.get("vendor") or vendor_hint or "") + device_type = str(device.get("device_type") or "") + ne_name = str(device.get("name") or task.ne_name or "") + ne_ip = str(device.get("ip_address") or task.ne_ip or "") + cmds = commands_for_vendor(vendor, device_type) + if cmds is None: + _update_task( + task_id, + status="fail", + message="unsupported_vendor", + vendor=vendor, + ne_name=ne_name, + ne_ip=ne_ip, + ended_at=_utcnow(), + ) + return + + cmd_names = command_list(cmds) + try: + outputs = _collect_with_timeout(creds, cmd_names) + except Exception as exc: + _log.warning("config_sync collect failed task=%s: %s", task_id, _format_error(exc)) + _update_task( + task_id, + status="fail", + message=_format_error(exc), + vendor=vendor, + ne_name=ne_name, + ne_ip=ne_ip, + ended_at=_utcnow(), + ) + return + + if not outputs or not str(outputs[0] or "").strip(): + _update_task( + task_id, + status="fail", + message="empty_config_output", + vendor=vendor, + ne_name=ne_name, + ne_ip=ne_ip, + ended_at=_utcnow(), + ) + return + + primary = outputs[0] + alt = outputs[1] if cmds.alt and len(outputs) > 1 else None + try: + _save_success_snapshot( + source=source, + target_id=target_id, + vendor=vendor, + device_type=device_type, + ne_name=ne_name, + ne_ip=ne_ip, + primary_text=primary, + alt_text=alt, + commands=cmd_names, + cycle_id=cycle_id, + task_id=task_id, + ) + except Exception as exc: + _log.exception("config_sync snapshot save failed task=%s", task_id) + _update_task(task_id, status="fail", message=_format_error(exc), ended_at=_utcnow()) + return + + _update_task( + task_id, + status="success", + message="synced", + vendor=vendor, + ne_name=ne_name, + ne_ip=ne_ip, + ended_at=_utcnow(), + ) + finally: + db.close() + db2 = SessionLocal() + try: + sync_cycle_progress(db2, cycle_id) + finalize_cycle(db2, cycle_id) + cycle = db2.get(ConfigSyncCycle, cycle_id) + if cycle and str(cycle.status or "") in ("success", "fail", "cancelled"): + _release_pool(cycle_id) + finally: + db2.close() + + +def _run_safe(cycle_id: str, task_id: str) -> None: + try: + _run_single(cycle_id, task_id) + except Exception: + _log.exception("config_sync worker crashed cycle=%s task=%s", cycle_id, task_id) + _update_task( + task_id, + status="fail", + message="config_sync_worker_crashed", + ended_at=_utcnow(), + ) + db = SessionLocal() + try: + sync_cycle_progress(db, cycle_id) + finalize_cycle(db, cycle_id) + finally: + db.close() + + +def schedule_cycle_tasks(cycle_id: str, task_ids: list[str], concurrency: int) -> int: + if not task_ids: + db = SessionLocal() + try: + finalize_cycle(db, cycle_id) + finally: + db.close() + return 0 + pool = _pool_for_cycle(cycle_id, concurrency) + submitted = 0 + for tid in task_ids: + pool.submit(_run_safe, cycle_id, str(tid)) + submitted += 1 + _log.info("scheduled config_sync cycle=%s tasks=%s concurrency=%s", cycle_id, submitted, concurrency) + return submitted + + +def dispatch_cycle(cycle_id: str) -> int: + """Load pending tasks for a cycle and schedule workers.""" + db = SessionLocal() + try: + cycle = db.get(ConfigSyncCycle, cycle_id) + if not cycle: + return 0 + if str(cycle.status or "") not in ("running", "pending"): + return 0 + if str(cycle.status or "") == "pending": + cycle.status = "running" + if not cycle.started_at: + cycle.started_at = _utcnow() + db.commit() + pending = ( + db.query(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "pending") + .all() + ) + task_ids = [str(t.id) for t in pending] + concurrency = max(1, min(30, int(cycle.concurrency or 5))) + finally: + db.close() + return schedule_cycle_tasks(cycle_id, task_ids, concurrency) diff --git a/netx_api/config_sync_scheduler.py b/netx_api/config_sync_scheduler.py new file mode 100644 index 0000000..633ba0d --- /dev/null +++ b/netx_api/config_sync_scheduler.py @@ -0,0 +1,113 @@ +"""Background scheduler for periodic config sync.""" + +from __future__ import annotations + +import logging +import threading +import time +from datetime import datetime +from uuid import uuid4 + +from .config import settings +from .config_sync_runner import dispatch_cycle +from .config_sync_service import ( + ensure_policy, + expand_targets, + has_running_cycle, + next_due_at, +) +from .db import SessionLocal +from .models import ConfigSyncCycle, ConfigSyncTask + +_log = logging.getLogger("netx.config_sync.scheduler") +_stop = threading.Event() +_thread: threading.Thread | None = None + + +def _utcnow() -> datetime: + return datetime.utcnow() + + +def try_start_scheduled_cycle() -> str | None: + """Create and dispatch a scheduled cycle if policy is due. Returns cycle id or None.""" + db = SessionLocal() + try: + policy = ensure_policy(db) + if not policy.enabled: + return None + if has_running_cycle(db): + return None + due = next_due_at(db, policy) + if due is not None and due > _utcnow(): + return None + targets = expand_targets(db, policy) + if not targets: + _log.info("config_sync schedule skip: no targets") + return None + concurrency = max(1, min(30, int(policy.concurrency or 5))) + cycle = ConfigSyncCycle( + id=uuid4().hex, + trigger_mode="schedule", + status="running", + concurrency=concurrency, + planned_count=len(targets), + started_at=_utcnow(), + created_at=_utcnow(), + ) + db.add(cycle) + db.flush() + for t in targets: + db.add( + ConfigSyncTask( + id=uuid4().hex, + cycle_id=cycle.id, + source=t["source"], + target_id=t["id"], + ne_name=t.get("ne_name") or "", + ne_ip=t.get("ne_ip") or "", + vendor=t.get("vendor") or "", + status="pending", + ) + ) + db.commit() + cycle_id = str(cycle.id) + except Exception: + db.rollback() + _log.exception("config_sync schedule create failed") + return None + finally: + db.close() + + dispatch_cycle(cycle_id) + _log.info("config_sync scheduled cycle started id=%s", cycle_id) + return cycle_id + + +def _loop() -> None: + tick = max(15, int(settings.config_sync_scheduler_tick_sec or 60)) + _log.info("config_sync scheduler started tick=%ss", tick) + while not _stop.is_set(): + try: + if bool(settings.config_sync_scheduler_enabled): + try_start_scheduled_cycle() + except Exception: + _log.exception("config_sync scheduler tick failed") + _stop.wait(tick) + _log.info("config_sync scheduler stopped") + + +def start_config_sync_scheduler() -> None: + global _thread + if not bool(settings.config_sync_scheduler_enabled): + _log.info("config_sync scheduler disabled by settings") + return + if _thread and _thread.is_alive(): + return + _stop.clear() + _thread = threading.Thread(target=_loop, name="config-sync-scheduler", daemon=True) + _thread.start() + _log.info("started thread %s alive=%s", _thread.name, _thread.is_alive()) + + +def stop_config_sync_scheduler() -> None: + _stop.set() diff --git a/netx_api/config_sync_schemas.py b/netx_api/config_sync_schemas.py new file mode 100644 index 0000000..1ab214c --- /dev/null +++ b/netx_api/config_sync_schemas.py @@ -0,0 +1,104 @@ +"""Pydantic schemas for config sync API.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class ConfigSyncTargetRef(BaseModel): + source: Literal["managed", "ume"] + id: str + + +class ConfigSyncPolicyOut(BaseModel): + enabled: bool + interval_days: int + concurrency: int + scope_mode: str + selected_targets: list[ConfigSyncTargetRef] = Field(default_factory=list) + history_keep: int + updated_at: datetime | None = None + + +class ConfigSyncPolicyUpdate(BaseModel): + enabled: bool | None = None + interval_days: int | None = Field(default=None, ge=1, le=365) + concurrency: int | None = Field(default=None, ge=1, le=30) + scope_mode: Literal["all", "selected"] | None = None + selected_targets: list[ConfigSyncTargetRef] | None = None + history_keep: int | None = Field(default=None, ge=0, le=30) + + +class ConfigSyncCycleCreate(BaseModel): + mode: Literal["full", "retry_failed"] = "full" + cycle_id: str | None = None + + +class ConfigSyncCycleOut(BaseModel): + id: str + trigger_mode: str + status: str + concurrency: int + planned_count: int + success_count: int + fail_count: int + skip_count: int + error_message: str = "" + started_at: datetime | None = None + ended_at: datetime | None = None + created_at: datetime | None = None + + +class ConfigSyncTaskOut(BaseModel): + id: str + cycle_id: str + source: str + target_id: str + ne_name: str + ne_ip: str + vendor: str + status: str + message: str = "" + started_at: datetime | None = None + ended_at: datetime | None = None + + +class ConfigSyncDashboardOut(BaseModel): + policy: ConfigSyncPolicyOut + snapshot_count: int + last_cycle: ConfigSyncCycleOut | None = None + running_cycle: ConfigSyncCycleOut | None = None + next_due_at: datetime | None = None + fail_by_vendor: dict[str, int] = Field(default_factory=dict) + + +class NeConfigSnapshotMetaOut(BaseModel): + source: str + target_id: str + vendor: str + device_type: str + ne_name: str + ne_ip: str + config_sha256: str + config_alt_sha256: str + plain_size: int + plain_alt_size: int + zlib_size: int + zlib_alt_size: int + has_alt: bool + commands: list[str] = Field(default_factory=list) + collected_at: datetime | None = None + last_cycle_id: str = "" + + +class NeConfigSnapshotDetailOut(NeConfigSnapshotMetaOut): + config_text: str = "" + config_alt_text: str = "" + + +class NeConfigHistoryOut(NeConfigSnapshotMetaOut): + id: str + cycle_id: str = "" diff --git a/netx_api/config_sync_service.py b/netx_api/config_sync_service.py new file mode 100644 index 0000000..42e4863 --- /dev/null +++ b/netx_api/config_sync_service.py @@ -0,0 +1,600 @@ +"""Config sync policy, cycles, dashboard, and snapshot queries.""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta +from typing import Any +from uuid import uuid4 + +from fastapi import HTTPException +from sqlalchemy import func, or_ +from sqlalchemy.orm import Session + +from .config_sync_codec import decompress_text +from .config_sync_schemas import ( + ConfigSyncCycleCreate, + ConfigSyncCycleOut, + ConfigSyncDashboardOut, + ConfigSyncPolicyOut, + ConfigSyncPolicyUpdate, + ConfigSyncTargetRef, + ConfigSyncTaskOut, + NeConfigHistoryOut, + NeConfigSnapshotDetailOut, + NeConfigSnapshotMetaOut, +) +from .models import ( + ConfigSyncCycle, + ConfigSyncPolicy, + ConfigSyncTask, + ManagedNE, + NeConfigHistory, + NeConfigSnapshot, + UmeInventoryNE, +) +from .cli_resolve import cli_profile_ready + +_log = logging.getLogger("netx.config_sync") + +POLICY_ID = 1 + + +def _utcnow() -> datetime: + return datetime.utcnow() + + +def ensure_policy(db: Session) -> ConfigSyncPolicy: + row = db.get(ConfigSyncPolicy, POLICY_ID) + if row is None: + row = ConfigSyncPolicy(id=POLICY_ID) + db.add(row) + db.commit() + db.refresh(row) + return row + + +def _targets_from_json(raw: Any) -> list[ConfigSyncTargetRef]: + items: list[ConfigSyncTargetRef] = [] + if not isinstance(raw, list): + return items + for x in raw: + if not isinstance(x, dict): + continue + src = str(x.get("source") or "").strip().lower() + tid = str(x.get("id") or "").strip() + if src not in ("managed", "ume") or not tid: + continue + items.append(ConfigSyncTargetRef(source=src, id=tid)) # type: ignore[arg-type] + return items + + +def policy_to_out(row: ConfigSyncPolicy) -> ConfigSyncPolicyOut: + return ConfigSyncPolicyOut( + enabled=bool(row.enabled), + interval_days=max(1, int(row.interval_days or 3)), + concurrency=max(1, min(30, int(row.concurrency or 5))), + scope_mode=str(row.scope_mode or "all"), + selected_targets=_targets_from_json(row.selected_targets), + history_keep=max(0, min(30, int(row.history_keep if row.history_keep is not None else 3))), + updated_at=row.updated_at, + ) + + +def get_policy(db: Session) -> ConfigSyncPolicyOut: + return policy_to_out(ensure_policy(db)) + + +def update_policy(db: Session, body: ConfigSyncPolicyUpdate) -> ConfigSyncPolicyOut: + row = ensure_policy(db) + data = body.model_dump(exclude_unset=True) + if "enabled" in data and data["enabled"] is not None: + row.enabled = bool(data["enabled"]) + if "interval_days" in data and data["interval_days"] is not None: + row.interval_days = int(data["interval_days"]) + if "concurrency" in data and data["concurrency"] is not None: + row.concurrency = max(1, min(30, int(data["concurrency"]))) + if "scope_mode" in data and data["scope_mode"] is not None: + row.scope_mode = str(data["scope_mode"]) + if "selected_targets" in data and data["selected_targets"] is not None: + refs = data["selected_targets"] + row.selected_targets = [ + {"source": r.source if hasattr(r, "source") else r["source"], "id": r.id if hasattr(r, "id") else r["id"]} + for r in refs + ] + if "history_keep" in data and data["history_keep"] is not None: + row.history_keep = max(0, min(30, int(data["history_keep"]))) + row.updated_at = _utcnow() + db.commit() + db.refresh(row) + return policy_to_out(row) + + +def cycle_to_out(row: ConfigSyncCycle) -> ConfigSyncCycleOut: + return ConfigSyncCycleOut( + id=str(row.id), + trigger_mode=str(row.trigger_mode or ""), + status=str(row.status or ""), + concurrency=int(row.concurrency or 0), + planned_count=int(row.planned_count or 0), + success_count=int(row.success_count or 0), + fail_count=int(row.fail_count or 0), + skip_count=int(row.skip_count or 0), + error_message=str(row.error_message or ""), + started_at=row.started_at, + ended_at=row.ended_at, + created_at=row.created_at, + ) + + +def task_to_out(row: ConfigSyncTask) -> ConfigSyncTaskOut: + return ConfigSyncTaskOut( + id=str(row.id), + cycle_id=str(row.cycle_id), + source=str(row.source), + target_id=str(row.target_id), + ne_name=str(row.ne_name or ""), + ne_ip=str(row.ne_ip or ""), + vendor=str(row.vendor or ""), + status=str(row.status or ""), + message=str(row.message or ""), + started_at=row.started_at, + ended_at=row.ended_at, + ) + + +def expand_targets(db: Session, policy: ConfigSyncPolicy) -> list[dict[str, str]]: + """Return list of {source, id, ne_name, ne_ip, vendor, device_type}.""" + mode = str(policy.scope_mode or "all").strip().lower() + out: list[dict[str, str]] = [] + seen: set[tuple[str, str]] = set() + + def _add(source: str, tid: str, name: str, ip: str, vendor: str, device_type: str) -> None: + key = (source, tid) + if key in seen: + return + seen.add(key) + out.append( + { + "source": source, + "id": tid, + "ne_name": name, + "ne_ip": ip, + "vendor": vendor, + "device_type": device_type, + } + ) + + if mode == "selected": + for ref in _targets_from_json(policy.selected_targets): + if ref.source == "managed": + ne = db.get(ManagedNE, ref.id) + if ne: + _add("managed", str(ne.id), str(ne.name or ""), str(ne.ip_address or ""), str(ne.vendor or ""), str(ne.device_type or "")) + else: + inv = db.get(UmeInventoryNE, ref.id) + if inv: + _add( + "ume", + str(inv.ne_id), + str(inv.host_name or inv.ne_name or inv.user_label or inv.ne_id or ""), + str(inv.ip_address or ""), + str(inv.vendor or ""), + str(inv.ne_type or ""), + ) + return out + + for ne in db.query(ManagedNE).order_by(ManagedNE.updated_at.desc()).all(): + _add("managed", str(ne.id), str(ne.name or ""), str(ne.ip_address or ""), str(ne.vendor or ""), str(ne.device_type or "")) + + if cli_profile_ready(db): + for inv in db.query(UmeInventoryNE).order_by(UmeInventoryNE.ne_id.asc()).all(): + if not str(inv.ip_address or "").strip(): + continue + _add( + "ume", + str(inv.ne_id), + str(inv.host_name or inv.ne_name or inv.user_label or inv.ne_id or ""), + str(inv.ip_address or ""), + str(inv.vendor or ""), + str(inv.ne_type or ""), + ) + return out + + +def has_running_cycle(db: Session) -> ConfigSyncCycle | None: + return ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.status.in_(("running", "pending"))) + .order_by(ConfigSyncCycle.created_at.desc()) + .first() + ) + + +def last_finished_cycle(db: Session) -> ConfigSyncCycle | None: + return ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.status.in_(("success", "fail", "cancelled"))) + .order_by(ConfigSyncCycle.ended_at.desc().nullslast(), ConfigSyncCycle.created_at.desc()) + .first() + ) + + +def next_due_at(db: Session, policy: ConfigSyncPolicy | None = None) -> datetime | None: + pol = policy or ensure_policy(db) + if not pol.enabled: + return None + last = ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.status == "success", ConfigSyncCycle.ended_at.isnot(None)) + .order_by(ConfigSyncCycle.ended_at.desc()) + .first() + ) + days = max(1, int(pol.interval_days or 3)) + if last and last.ended_at: + return last.ended_at + timedelta(days=days) + return _utcnow() + + +def create_cycle(db: Session, body: ConfigSyncCycleCreate) -> ConfigSyncCycleOut: + if has_running_cycle(db): + raise HTTPException(status_code=409, detail="config_sync_cycle_already_running") + policy = ensure_policy(db) + mode = str(body.mode or "full").strip().lower() + trigger = "retry_failed" if mode == "retry_failed" else "manual" + concurrency = max(1, min(30, int(policy.concurrency or 5))) + + targets: list[dict[str, str]] = [] + if mode == "retry_failed": + src_cycle_id = str(body.cycle_id or "").strip() + src = None + if src_cycle_id: + src = db.get(ConfigSyncCycle, src_cycle_id) + if src is None: + src = ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.fail_count > 0) + .order_by(ConfigSyncCycle.created_at.desc()) + .first() + ) + if src is None: + raise HTTPException(status_code=404, detail="no_failed_cycle") + fails = ( + db.query(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == src.id, ConfigSyncTask.status == "fail") + .all() + ) + for t in fails: + targets.append( + { + "source": str(t.source), + "id": str(t.target_id), + "ne_name": str(t.ne_name or ""), + "ne_ip": str(t.ne_ip or ""), + "vendor": str(t.vendor or ""), + "device_type": "", + } + ) + if not targets: + raise HTTPException(status_code=400, detail="no_failed_tasks") + else: + targets = expand_targets(db, policy) + if not targets: + raise HTTPException(status_code=400, detail="no_targets") + + cycle = ConfigSyncCycle( + id=uuid4().hex, + trigger_mode=trigger, + status="running", + concurrency=concurrency, + planned_count=len(targets), + started_at=_utcnow(), + created_at=_utcnow(), + ) + db.add(cycle) + db.flush() + for t in targets: + db.add( + ConfigSyncTask( + id=uuid4().hex, + cycle_id=cycle.id, + source=t["source"], + target_id=t["id"], + ne_name=t.get("ne_name") or "", + ne_ip=t.get("ne_ip") or "", + vendor=t.get("vendor") or "", + status="pending", + ) + ) + db.commit() + db.refresh(cycle) + return cycle_to_out(cycle) + + +def list_cycles(db: Session, *, page: int, page_size: int) -> dict[str, Any]: + q = db.query(ConfigSyncCycle).order_by(ConfigSyncCycle.created_at.desc()) + total = int(q.count()) + rows = q.offset((page - 1) * page_size).limit(page_size).all() + return {"total": total, "page": page, "page_size": page_size, "items": [cycle_to_out(r) for r in rows]} + + +def get_cycle(db: Session, cycle_id: str) -> ConfigSyncCycleOut: + row = db.get(ConfigSyncCycle, cycle_id) + if not row: + raise HTTPException(status_code=404, detail="cycle_not_found") + return cycle_to_out(row) + + +def list_cycle_tasks( + db: Session, + cycle_id: str, + *, + page: int, + page_size: int, + status: str = "", + keyword: str = "", +) -> dict[str, Any]: + if not db.get(ConfigSyncCycle, cycle_id): + raise HTTPException(status_code=404, detail="cycle_not_found") + q = db.query(ConfigSyncTask).filter(ConfigSyncTask.cycle_id == cycle_id) + st = str(status or "").strip() + if st: + q = q.filter(ConfigSyncTask.status == st) + kw = str(keyword or "").strip() + if kw: + like = f"%{kw}%" + q = q.filter( + or_( + ConfigSyncTask.ne_name.ilike(like), + ConfigSyncTask.ne_ip.ilike(like), + ConfigSyncTask.target_id.ilike(like), + ConfigSyncTask.message.ilike(like), + ) + ) + total = int(q.count()) + rows = q.order_by(ConfigSyncTask.ne_name.asc()).offset((page - 1) * page_size).limit(page_size).all() + return {"total": total, "page": page, "page_size": page_size, "items": [task_to_out(r) for r in rows]} + + +def pause_cycle(db: Session, cycle_id: str) -> ConfigSyncCycleOut: + row = db.get(ConfigSyncCycle, cycle_id) + if not row: + raise HTTPException(status_code=404, detail="cycle_not_found") + if str(row.status) != "running": + raise HTTPException(status_code=400, detail="cycle_not_running") + row.status = "paused" + db.commit() + db.refresh(row) + return cycle_to_out(row) + + +def resume_cycle(db: Session, cycle_id: str) -> ConfigSyncCycleOut: + row = db.get(ConfigSyncCycle, cycle_id) + if not row: + raise HTTPException(status_code=404, detail="cycle_not_found") + if str(row.status) != "paused": + raise HTTPException(status_code=400, detail="cycle_not_paused") + pending = ( + db.query(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "pending") + .count() + ) + if pending <= 0: + raise HTTPException(status_code=400, detail="no_pending_tasks") + other = has_running_cycle(db) + if other and str(other.id) != cycle_id: + raise HTTPException(status_code=409, detail="config_sync_cycle_already_running") + row.status = "running" + db.commit() + db.refresh(row) + return cycle_to_out(row) + + +def dashboard(db: Session) -> ConfigSyncDashboardOut: + policy = ensure_policy(db) + snap_count = int(db.query(func.count()).select_from(NeConfigSnapshot).scalar() or 0) + running = ( + db.query(ConfigSyncCycle) + .filter(ConfigSyncCycle.status.in_(("running", "paused", "pending"))) + .order_by(ConfigSyncCycle.created_at.desc()) + .first() + ) + last = last_finished_cycle(db) + fail_by_vendor: dict[str, int] = {} + if last: + rows = ( + db.query(ConfigSyncTask.vendor, func.count()) + .filter(ConfigSyncTask.cycle_id == last.id, ConfigSyncTask.status == "fail") + .group_by(ConfigSyncTask.vendor) + .all() + ) + for vendor, cnt in rows: + fail_by_vendor[str(vendor or "unknown") or "unknown"] = int(cnt) + return ConfigSyncDashboardOut( + policy=policy_to_out(policy), + snapshot_count=snap_count, + last_cycle=cycle_to_out(last) if last else None, + running_cycle=cycle_to_out(running) if running else None, + next_due_at=next_due_at(db, policy), + fail_by_vendor=fail_by_vendor, + ) + + +def _snap_meta(row: NeConfigSnapshot) -> NeConfigSnapshotMetaOut: + cmds = row.commands_json if isinstance(row.commands_json, list) else [] + return NeConfigSnapshotMetaOut( + source=str(row.source), + target_id=str(row.target_id), + vendor=str(row.vendor or ""), + device_type=str(row.device_type or ""), + ne_name=str(row.ne_name or ""), + ne_ip=str(row.ne_ip or ""), + config_sha256=str(row.config_sha256 or ""), + config_alt_sha256=str(row.config_alt_sha256 or ""), + plain_size=int(row.plain_size or 0), + plain_alt_size=int(row.plain_alt_size or 0), + zlib_size=int(row.zlib_size or 0), + zlib_alt_size=int(row.zlib_alt_size or 0), + has_alt=bool(row.config_alt_zlib), + commands=[str(c) for c in cmds], + collected_at=row.collected_at, + last_cycle_id=str(row.last_cycle_id or ""), + ) + + +def list_snapshots( + db: Session, + *, + page: int, + page_size: int, + keyword: str = "", + source: str = "", + vendor: str = "", +) -> dict[str, Any]: + q = db.query(NeConfigSnapshot) + src = str(source or "").strip().lower() + if src in ("managed", "ume"): + q = q.filter(NeConfigSnapshot.source == src) + vend = str(vendor or "").strip() + if vend: + q = q.filter(NeConfigSnapshot.vendor.ilike(f"%{vend}%")) + kw = str(keyword or "").strip() + if kw: + like = f"%{kw}%" + q = q.filter( + or_( + NeConfigSnapshot.ne_name.ilike(like), + NeConfigSnapshot.ne_ip.ilike(like), + NeConfigSnapshot.target_id.ilike(like), + ) + ) + total = int(q.count()) + rows = q.order_by(NeConfigSnapshot.collected_at.desc()).offset((page - 1) * page_size).limit(page_size).all() + return {"total": total, "page": page, "page_size": page_size, "items": [_snap_meta(r) for r in rows]} + + +def get_snapshot_detail( + db: Session, + source: str, + target_id: str, + *, + field: str = "both", +) -> NeConfigSnapshotDetailOut: + src = str(source or "").strip().lower() + tid = str(target_id or "").strip() + row = db.get(NeConfigSnapshot, {"source": src, "target_id": tid}) + if not row: + raise HTTPException(status_code=404, detail="snapshot_not_found") + meta = _snap_meta(row) + primary = "" + alt = "" + f = str(field or "both").strip().lower() + if f in ("primary", "both", ""): + primary = decompress_text(row.config_zlib) + if f in ("alt", "both") and row.config_alt_zlib: + alt = decompress_text(row.config_alt_zlib) + return NeConfigSnapshotDetailOut(**meta.model_dump(), config_text=primary, config_alt_text=alt) + + +def list_snapshot_history( + db: Session, + source: str, + target_id: str, + *, + page: int, + page_size: int, +) -> dict[str, Any]: + src = str(source or "").strip().lower() + tid = str(target_id or "").strip() + q = ( + db.query(NeConfigHistory) + .filter(NeConfigHistory.source == src, NeConfigHistory.target_id == tid) + .order_by(NeConfigHistory.collected_at.desc()) + ) + total = int(q.count()) + rows = q.offset((page - 1) * page_size).limit(page_size).all() + items: list[NeConfigHistoryOut] = [] + for row in rows: + cmds = row.commands_json if isinstance(row.commands_json, list) else [] + items.append( + NeConfigHistoryOut( + id=str(row.id), + source=str(row.source), + target_id=str(row.target_id), + vendor=str(row.vendor or ""), + device_type=str(row.device_type or ""), + ne_name=str(row.ne_name or ""), + ne_ip=str(row.ne_ip or ""), + config_sha256=str(row.config_sha256 or ""), + config_alt_sha256=str(row.config_alt_sha256 or ""), + plain_size=int(row.plain_size or 0), + plain_alt_size=int(row.plain_alt_size or 0), + zlib_size=int(row.zlib_size or 0), + zlib_alt_size=int(row.zlib_alt_size or 0), + has_alt=bool(row.config_alt_zlib), + commands=[str(c) for c in cmds], + collected_at=row.collected_at, + last_cycle_id=str(row.cycle_id or ""), + cycle_id=str(row.cycle_id or ""), + ) + ) + return {"total": total, "page": page, "page_size": page_size, "items": items} + + +def sync_cycle_progress(db: Session, cycle_id: str) -> None: + cycle = db.get(ConfigSyncCycle, cycle_id) + if not cycle: + return + success = ( + db.query(func.count()) + .select_from(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "success") + .scalar() + or 0 + ) + fail = ( + db.query(func.count()) + .select_from(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status == "fail") + .scalar() + or 0 + ) + skip = ( + db.query(func.count()) + .select_from(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status.in_(("skipped", "cancelled"))) + .scalar() + or 0 + ) + cycle.success_count = int(success) + cycle.fail_count = int(fail) + cycle.skip_count = int(skip) + db.commit() + + +def finalize_cycle(db: Session, cycle_id: str) -> None: + cycle = db.get(ConfigSyncCycle, cycle_id) + if not cycle: + return + if str(cycle.status) == "paused": + return + pending = ( + db.query(func.count()) + .select_from(ConfigSyncTask) + .filter(ConfigSyncTask.cycle_id == cycle_id, ConfigSyncTask.status.in_(("pending", "running"))) + .scalar() + or 0 + ) + if int(pending) > 0: + return + sync_cycle_progress(db, cycle_id) + db.refresh(cycle) + if int(cycle.fail_count or 0) > 0 and int(cycle.success_count or 0) == 0: + cycle.status = "fail" + elif int(cycle.fail_count or 0) > 0: + cycle.status = "fail" + cycle.error_message = cycle.error_message or "completed_with_failures" + else: + cycle.status = "success" + cycle.ended_at = _utcnow() + db.commit() diff --git a/netx_api/main.py b/netx_api/main.py index 4ed85bb..11f46e1 100644 --- a/netx_api/main.py +++ b/netx_api/main.py @@ -26,6 +26,7 @@ from .config import settings from .db import Base, SessionLocal, engine, get_db from .collection_router import router as collection_router from .cli_router import router as cli_router +from .config_sync_router import router as config_sync_router from .managed_ne_router import router as managed_ne_router from .webcrt_router import router as webcrt_router from .topology_router import router as topology_router @@ -133,6 +134,7 @@ app.include_router(auth_router) app.include_router(managed_ne_router) app.include_router(cli_router) app.include_router(collection_router) +app.include_router(config_sync_router) app.include_router(webcrt_router) app.include_router(topology_router) parser_cfg = load_parser_config() @@ -832,10 +834,23 @@ def on_startup() -> None: resumed = recover_collection_jobs_on_startup(db) if resumed: _schedule_log.info("startup: resumed %s pending ne collection runs", resumed) + from .config_sync_recovery import recover_config_sync_on_startup + from .config_sync_service import ensure_policy + + ensure_policy(db) + cfg_resumed = recover_config_sync_on_startup(db) + if cfg_resumed: + _schedule_log.info("startup: resumed %s pending config_sync tasks", cfg_resumed) except Exception: - _schedule_log.exception("startup: ne collection job recovery failed") + _schedule_log.exception("startup: ne collection / config_sync recovery failed") finally: db.close() + try: + from .config_sync_scheduler import start_config_sync_scheduler + + start_config_sync_scheduler() + except Exception: + _schedule_log.exception("startup: config_sync scheduler init failed") # Best-effort schema evolution for new columns (no migrations framework). # Safe for Postgres (IF NOT EXISTS); ignored on failure. try: diff --git a/netx_api/models.py b/netx_api/models.py index b99facd..d573a89 100644 --- a/netx_api/models.py +++ b/netx_api/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime from uuid import uuid4 -from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, LargeBinary, String, Text, UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.types import JSON @@ -484,3 +484,107 @@ class ApiToken(Base): expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) last_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) revoked_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + + +class ConfigSyncPolicy(Base): + """Singleton policy for periodic config sync (id=1).""" + + __tablename__ = "config_sync_policy" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1) + enabled: Mapped[bool] = mapped_column(Boolean, default=True) + interval_days: Mapped[int] = mapped_column(Integer, default=3) + concurrency: Mapped[int] = mapped_column(Integer, default=5) + scope_mode: Mapped[str] = mapped_column(String(32), default="all") # all | selected + selected_targets: Mapped[list] = mapped_column(_JsonType, default=list) + history_keep: Mapped[int] = mapped_column(Integer, default=3) + updated_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow) + + +class ConfigSyncCycle(Base): + """One config-sync cycle over many NEs.""" + + __tablename__ = "config_sync_cycle" + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + trigger_mode: Mapped[str] = mapped_column(String(32), default="schedule", index=True) + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + concurrency: Mapped[int] = mapped_column(Integer, default=5) + planned_count: Mapped[int] = mapped_column(Integer, default=0) + success_count: Mapped[int] = mapped_column(Integer, default=0) + fail_count: Mapped[int] = mapped_column(Integer, default=0) + skip_count: Mapped[int] = mapped_column(Integer, default=0) + error_message: Mapped[str] = mapped_column(String(1024), default="") + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + + +class ConfigSyncTask(Base): + """Per-NE work item inside a config sync cycle.""" + + __tablename__ = "config_sync_task" + __table_args__ = (UniqueConstraint("cycle_id", "source", "target_id", name="uq_config_sync_task_target"),) + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + cycle_id: Mapped[str] = mapped_column(String(64), index=True) + source: Mapped[str] = mapped_column(String(32), default="managed", index=True) # managed | ume + target_id: Mapped[str] = mapped_column(String(128), 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="") + status: Mapped[str] = mapped_column(String(32), default="pending", index=True) + message: Mapped[str] = mapped_column(String(1024), default="") + started_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + ended_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + + +class NeConfigSnapshot(Base): + """Latest successful config snapshot per NE (zlib-compressed).""" + + __tablename__ = "ne_config_snapshot" + + source: Mapped[str] = mapped_column(String(32), primary_key=True) # managed | ume + target_id: Mapped[str] = mapped_column(String(128), primary_key=True) + vendor: Mapped[str] = mapped_column(String(64), default="") + device_type: Mapped[str] = mapped_column(String(64), default="") + ne_name: Mapped[str] = mapped_column(String(256), default="", index=True) + ne_ip: Mapped[str] = mapped_column(String(128), default="", index=True) + config_zlib: Mapped[bytes] = mapped_column(LargeBinary, default=b"") + config_alt_zlib: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True) + config_sha256: Mapped[str] = mapped_column(String(64), default="") + config_alt_sha256: Mapped[str] = mapped_column(String(64), default="") + plain_size: Mapped[int] = mapped_column(Integer, default=0) + plain_alt_size: Mapped[int] = mapped_column(Integer, default=0) + zlib_size: Mapped[int] = mapped_column(Integer, default=0) + zlib_alt_size: Mapped[int] = mapped_column(Integer, default=0) + commands_json: Mapped[list] = mapped_column(_JsonType, default=list) + collected_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + last_cycle_id: Mapped[str] = mapped_column(String(64), default="") + last_task_id: Mapped[str] = mapped_column(String(64), default="") + + +class NeConfigHistory(Base): + """Historical config versions when content changes.""" + + __tablename__ = "ne_config_history" + + id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) + source: Mapped[str] = mapped_column(String(32), index=True) + target_id: Mapped[str] = mapped_column(String(128), index=True) + vendor: Mapped[str] = mapped_column(String(64), default="") + device_type: Mapped[str] = mapped_column(String(64), default="") + ne_name: Mapped[str] = mapped_column(String(256), default="") + ne_ip: Mapped[str] = mapped_column(String(128), default="") + config_zlib: Mapped[bytes] = mapped_column(LargeBinary, default=b"") + config_alt_zlib: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True) + config_sha256: Mapped[str] = mapped_column(String(64), default="") + config_alt_sha256: Mapped[str] = mapped_column(String(64), default="") + plain_size: Mapped[int] = mapped_column(Integer, default=0) + plain_alt_size: Mapped[int] = mapped_column(Integer, default=0) + zlib_size: Mapped[int] = mapped_column(Integer, default=0) + zlib_alt_size: Mapped[int] = mapped_column(Integer, default=0) + commands_json: Mapped[list] = mapped_column(_JsonType, default=list) + collected_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow, index=True) + cycle_id: Mapped[str] = mapped_column(String(64), default="") + task_id: Mapped[str] = mapped_column(String(64), default="") diff --git a/tests/test_config_sync.py b/tests/test_config_sync.py new file mode 100644 index 0000000..1ffa259 --- /dev/null +++ b/tests/test_config_sync.py @@ -0,0 +1,243 @@ +"""Unit tests for config sync commands, codec, snapshot overwrite rules, recovery.""" + +from __future__ import annotations + +import unittest +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +from netx_api.config_sync_codec import compress_text, decompress_text +from netx_api.config_sync_commands import command_list, commands_for_vendor, normalize_vendor_key +from netx_api.config_sync_recovery import recover_config_sync_on_startup +from netx_api.config_sync_runner import _claim_task, _save_success_snapshot + + +class ConfigSyncCommandsTests(unittest.TestCase): + def test_normalize_vendor_keys(self): + self.assertEqual(normalize_vendor_key("Cisco", "ios"), "cisco") + self.assertEqual(normalize_vendor_key("ZTE", "zxros"), "zte") + self.assertEqual(normalize_vendor_key("Huawei", "vrp"), "huawei") + self.assertEqual(normalize_vendor_key("H3C", "comware"), "h3c") + self.assertEqual(normalize_vendor_key("Juniper", "junos"), "juniper") + self.assertEqual(normalize_vendor_key("Nokia", "sros"), "nokia") + self.assertEqual(normalize_vendor_key("Ericsson", ""), "ericsson") + self.assertEqual(normalize_vendor_key("Acme", "router"), "other") + + def test_command_matrix(self): + self.assertEqual(commands_for_vendor("Cisco", "").primary, "show running-config") + self.assertEqual(commands_for_vendor("ZTE", "").primary, "show running-config") + self.assertEqual(commands_for_vendor("Huawei", "").primary, "display current-configuration") + self.assertEqual(commands_for_vendor("H3C", "").primary, "display current-configuration") + self.assertEqual(commands_for_vendor("Nokia", "").primary, "admin display-config") + self.assertEqual(commands_for_vendor("Ericsson", "").primary, "show configuration") + self.assertIsNone(commands_for_vendor("UnknownVendor", "")) + + def test_juniper_dual_commands(self): + cmds = commands_for_vendor("Juniper", "junos") + assert cmds is not None + self.assertEqual(cmds.primary, "show configuration | display set") + self.assertEqual(cmds.alt, "show configuration | no-more") + self.assertEqual( + command_list(cmds), + ["show configuration | display set", "show configuration | no-more"], + ) + + +class ConfigSyncCodecTests(unittest.TestCase): + def test_zlib_roundtrip(self): + text = "hostname router1\ninterface GigabitEthernet0/0\n" + blob, digest, plain_size, zlib_size = compress_text(text) + self.assertEqual(plain_size, len(text.encode("utf-8"))) + self.assertGreater(zlib_size, 0) + self.assertEqual(len(digest), 64) + self.assertEqual(decompress_text(blob), text) + + def test_empty_decompress(self): + self.assertEqual(decompress_text(None), "") + self.assertEqual(decompress_text(b""), "") + + +class ConfigSyncSnapshotOverwriteTests(unittest.TestCase): + @patch("netx_api.config_sync_runner.SessionLocal") + def test_success_updates_snapshot(self, session_local): + db = MagicMock() + session_local.return_value = db + existing = MagicMock() + existing.source = "managed" + existing.target_id = "ne1" + existing.config_sha256 = "old" + existing.config_alt_sha256 = "" + existing.config_zlib = b"old" + existing.config_alt_zlib = None + existing.plain_size = 1 + existing.plain_alt_size = 0 + existing.zlib_size = 1 + existing.zlib_alt_size = 0 + existing.commands_json = [] + existing.collected_at = datetime.now(timezone.utc) + existing.last_cycle_id = "c0" + existing.last_task_id = "t0" + existing.vendor = "Cisco" + existing.device_type = "ios" + existing.ne_name = "r1" + existing.ne_ip = "1.1.1.1" + db.get.side_effect = lambda model, key: existing if "Snapshot" in str(model) else MagicMock(history_keep=3) + + _save_success_snapshot( + source="managed", + target_id="ne1", + vendor="Cisco", + device_type="ios", + ne_name="r1", + ne_ip="1.1.1.1", + primary_text="hostname r1\n", + alt_text=None, + commands=["show running-config"], + cycle_id="c1", + task_id="t1", + ) + self.assertEqual(existing.last_cycle_id, "c1") + self.assertEqual(existing.last_task_id, "t1") + self.assertTrue(existing.config_sha256) + self.assertNotEqual(existing.config_sha256, "old") + db.commit.assert_called() + + @patch("netx_api.config_sync_runner._save_success_snapshot") + @patch("netx_api.config_sync_runner._collect_with_timeout", side_effect=TimeoutError("boom")) + @patch("netx_api.config_sync_runner.resolve_cli_target") + @patch("netx_api.config_sync_runner._claim_task", return_value=True) + @patch("netx_api.config_sync_runner.SessionLocal") + def test_fail_does_not_overwrite_snapshot(self, session_local, _claim, resolve, _collect, save_snap): + from netx_api.config_sync_runner import _run_single + + db = MagicMock() + session_local.return_value = db + task = MagicMock() + task.source = "managed" + task.target_id = "ne1" + task.vendor = "Cisco" + task.ne_name = "r1" + task.ne_ip = "1.1.1.1" + db.get.return_value = task + resolve.return_value = ( + {"host": "1.1.1.1"}, + {"vendor": "Cisco", "device_type": "ios", "name": "r1", "ip_address": "1.1.1.1"}, + ) + + with patch("netx_api.config_sync_runner._update_task") as update_task, patch( + "netx_api.config_sync_runner.sync_cycle_progress" + ), patch("netx_api.config_sync_runner.finalize_cycle"): + _run_single("c1", "t1") + save_snap.assert_not_called() + args = update_task.call_args + self.assertEqual(args[0][0], "t1") + self.assertEqual(args[1]["status"], "fail") + + @patch("netx_api.config_sync_runner._save_success_snapshot") + @patch( + "netx_api.config_sync_runner._collect_with_timeout", + return_value=["set system host-name r1", "system {\n host-name r1;\n}"], + ) + @patch("netx_api.config_sync_runner.resolve_cli_target") + @patch("netx_api.config_sync_runner._claim_task", return_value=True) + @patch("netx_api.config_sync_runner.SessionLocal") + def test_juniper_dual_fields_passed_to_save(self, session_local, _claim, resolve, _collect, save_snap): + from netx_api.config_sync_runner import _run_single + + db = MagicMock() + session_local.return_value = db + task = MagicMock() + task.source = "managed" + task.target_id = "ne1" + task.vendor = "Juniper" + task.ne_name = "r1" + task.ne_ip = "1.1.1.1" + db.get.return_value = task + resolve.return_value = ( + {"host": "1.1.1.1"}, + {"vendor": "Juniper", "device_type": "junos", "name": "r1", "ip_address": "1.1.1.1"}, + ) + + with patch("netx_api.config_sync_runner._update_task"), patch( + "netx_api.config_sync_runner.sync_cycle_progress" + ), patch("netx_api.config_sync_runner.finalize_cycle"): + _run_single("c1", "t1") + save_snap.assert_called_once() + kwargs = save_snap.call_args.kwargs + self.assertEqual(kwargs["primary_text"], "set system host-name r1") + self.assertIn("host-name r1", kwargs["alt_text"]) + self.assertEqual( + kwargs["commands"], + ["show configuration | display set", "show configuration | no-more"], + ) + + +class ConfigSyncClaimTests(unittest.TestCase): + @patch("netx_api.config_sync_runner.SessionLocal") + def test_claim_pending_to_running(self, session_local): + db = MagicMock() + session_local.return_value = db + task = MagicMock() + task.status = "pending" + cycle = MagicMock() + cycle.status = "running" + + def get_side(model, _id): + name = getattr(model, "__name__", str(model)) + if "Task" in name: + return task + return cycle + + db.get.side_effect = get_side + ok = _claim_task("c1", "t1") + self.assertTrue(ok) + self.assertEqual(task.status, "running") + db.commit.assert_called() + + @patch("netx_api.config_sync_runner.SessionLocal") + def test_claim_skipped_when_paused(self, session_local): + db = MagicMock() + session_local.return_value = db + task = MagicMock() + task.status = "pending" + cycle = MagicMock() + cycle.status = "paused" + db.get.side_effect = lambda model, _id: task if "Task" in getattr(model, "__name__", "") else cycle + self.assertFalse(_claim_task("c1", "t1")) + self.assertEqual(task.status, "pending") + + +class ConfigSyncRecoveryTests(unittest.TestCase): + @patch("netx_api.config_sync_recovery.dispatch_cycle", return_value=2) + @patch("netx_api.config_sync_recovery.finalize_cycle") + @patch("netx_api.config_sync_recovery.sync_cycle_progress") + def test_orphan_running_tasks_marked_fail(self, _sync, _fin, dispatch): + cycle = MagicMock() + cycle.id = "c1" + cycle.status = "running" + cycle.started_at = datetime.now(timezone.utc) + + orphan = MagicMock() + orphan.status = "running" + orphan.message = "" + orphan.ended_at = None + + db = MagicMock() + db.query.side_effect = [ + MagicMock(filter=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[cycle])))), + MagicMock(filter=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[orphan])))), + MagicMock(filter=MagicMock(return_value=MagicMock(count=MagicMock(return_value=2)))), + ] + db.refresh = MagicMock() + + resumed = recover_config_sync_on_startup(db) + + self.assertEqual(orphan.status, "fail") + self.assertEqual(orphan.message, "orphan_recovered") + self.assertIsNotNone(orphan.ended_at) + self.assertEqual(resumed, 2) + dispatch.assert_called_once_with("c1") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/WEB.md b/web/WEB.md index 20bb221..3cbcb55 100644 --- a/web/WEB.md +++ b/web/WEB.md @@ -37,11 +37,13 @@ src/ | `/network/devices` | 设备列表 | `network` | | `/network/topology` | 拓扑信息 | `network` | | `/network/alarms` | 告警信息 | `network` | +| `/network/configs` | 配置信息（同步快照） | `network` | | `/network/webcrt` | WebCRT 终端 | `network` | -| `/network/tasks/config-sync` | 采集任务（原批量采集） | `network` | +| `/network/tasks/collect` | 采集任务 | `network` | +| `/network/tasks/config-sync` | 配置同步 | `network` | | `/network/tasks/port-traffic` | 端口流量（占位） | `network` | | `/ne` | redirect → `/network/devices` | — | -| `/collect` | redirect → `/network/tasks/config-sync` | — | +| `/collect` | redirect → `/network/tasks/collect` | — | | `/topology` | redirect → `/network/topology` | — | | `/webcrt` | redirect → `/network/webcrt`（保留 query） | — | @@ -90,6 +92,14 @@ src/ - 采集日志目录：`NETX_NE_COLLECTION_DATA_DIR`（默认 `data/ne_collections`） - 命令每行一条，`#` 为注释；输出格式与旧版 NetX 采集 `.txt` 一致 +## 配置同步 + +- API：`/v1/config-sync/*`（策略、看板、周期、快照） +- 存储：PostgreSQL `ne_config_snapshot` / `ne_config_history`（zlib BYTEA） +- 范围：ManagedNE + UME CLI 目标；厂商固定只读命令矩阵 +- 调度：`NETX_CONFIG_SYNC_SCHEDULER_ENABLED`（默认开），周期天数策略可配（默认 3 天） +- 前端：`/network/tasks/config-sync`（看板）+ `/network/configs`（查看） + ## WebCRT - API：`POST /v1/webcrt/sessions`（`ne_id` 或 `ume_ne_id`）、`WS /v1/webcrt/sessions/{id}/ws`、`DELETE /v1/webcrt/sessions/{id}` diff --git a/web/src/App.tsx b/web/src/App.tsx index 4a91346..b3f6577 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -4,12 +4,14 @@ import { AppLayout } from "./layout/AppLayout"; import { queryKeys } from "./constants/queryKeys"; import { WorkbenchPage } from "./pages/WorkbenchPage"; import { CollectPage } from "./pages/CollectPage"; +import { ConfigSyncPage } from "./pages/ConfigSyncPage"; import { NePage } from "./pages/NePage"; import { UmePage } from "./pages/UmePage"; import { WebcrtPage } from "./pages/WebcrtPage"; import { TopologyPage } from "./pages/TopologyPage"; import { NetworkLayout } from "./pages/network/NetworkLayout"; import { NetworkAlarmsPage } from "./pages/network/NetworkAlarmsPage"; +import { NetworkConfigsPage } from "./pages/network/NetworkConfigsPage"; import { NetworkPlaceholderPage } from "./pages/network/NetworkPlaceholderPage"; import { LoginPage } from "./pages/LoginPage"; import { UsersPage } from "./pages/UsersPage"; @@ -85,12 +87,14 @@ function ProtectedApp() { } /> } /> } /> + } /> } /> - } /> + } /> + } /> } /> } /> - } /> + } /> } /> } /> } /> diff --git a/web/src/config/networkNav.ts b/web/src/config/networkNav.ts index 1acabb8..cacedc8 100644 --- a/web/src/config/networkNav.ts +++ b/web/src/config/networkNav.ts @@ -26,6 +26,7 @@ export const NETWORK_NAV: readonly NetworkNavGroup[] = [ { id: "devices", path: "/network/devices", labelKey: "network.nav.devices", group: "detail" }, { id: "topology", path: "/network/topology", labelKey: "network.nav.topology", group: "detail" }, { id: "alarms", path: "/network/alarms", labelKey: "network.nav.alarms", group: "detail" }, + { id: "configs", path: "/network/configs", labelKey: "network.nav.configs", group: "detail" }, { id: "webcrt", path: "/network/webcrt", labelKey: "network.nav.webcrt", group: "detail" }, ], }, @@ -33,6 +34,12 @@ export const NETWORK_NAV: readonly NetworkNavGroup[] = [ id: "tasks", labelKey: "network.nav.tasks", items: [ + { + id: "collect", + path: "/network/tasks/collect", + labelKey: "network.nav.collectTasks", + group: "tasks", + }, { id: "config-sync", path: "/network/tasks/config-sync", diff --git a/web/src/constants/queryKeys.ts b/web/src/constants/queryKeys.ts index 0d4f489..fb0a30e 100644 --- a/web/src/constants/queryKeys.ts +++ b/web/src/constants/queryKeys.ts @@ -48,4 +48,15 @@ export const queryKeys = { ) => ["umeCurrentAlarms", severity, cleared, hostName, keyword, page, pageSize] as const, topologyMaps: ["topologyMaps"] as const, topologyGraph: (mapId: string) => ["topologyGraph", mapId] as const, + configSyncDashboard: ["configSyncDashboard"] as const, + configSyncPolicy: ["configSyncPolicy"] as const, + configSyncCyclesAll: ["configSyncCycles"] as const, + configSyncCycles: (page: number) => ["configSyncCycles", page] as const, + configSyncCycleTasksAll: ["configSyncCycleTasks"] as const, + configSyncCycleTasks: (cycleId: string, page: number, status: string, keyword: string) => + ["configSyncCycleTasks", cycleId, page, status, keyword] as const, + networkConfigsAll: ["networkConfigs"] as const, + networkConfigs: (page: number, keyword: string, source: string, vendor: string) => + ["networkConfigs", page, keyword, source, vendor] as const, + networkConfigDetail: (source: string, id: string) => ["networkConfigDetail", source, id] as const, }; diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 4b74c33..984887d 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -49,8 +49,10 @@ const en = { devices: "Device list", topology: "Topology", alarms: "Alarms", + configs: "Configurations", webcrt: "WebCRT", - configSync: "Collection tasks", + collectTasks: "Collection tasks", + configSync: "Config sync", portTraffic: "Port traffic", }, placeholder: { @@ -64,6 +66,71 @@ const en = { portTrafficStep4: "UI: task list + per-port trend charts", }, }, + configSync: { + title: "Config sync", + syncNow: "Sync now", + retryFailed: "Retry failed", + pause: "Pause", + resume: "Resume", + started: "Sync cycle started", + paused: "Paused", + resumed: "Resumed", + policyTitle: "Sync policy", + policySaved: "Policy saved", + enabled: "Enable scheduled sync", + intervalDays: "Interval (days)", + concurrency: "Concurrency", + historyKeep: "History keep", + scope: "Scope", + scopeAll: "All NEs", + scopeSelected: "Selected NEs", + savePolicy: "Save policy", + selectedCount: "{{count}} selected", + targetKeywordPh: "Name / IP / vendor", + cyclesTitle: "Sync cycles", + cyclesEmpty: "No sync cycles yet", + tasksTitle: "Cycle tasks", + taskKeywordPh: "Name / IP / message", + allStatus: "All statuses", + expand: "Expand", + collapse: "Collapse", + kpi: { + snapshots: "Snapshots", + running: "In progress", + idle: "Idle", + last: "Last cycle", + nextDue: "Next due", + }, + col: { + source: "Source", + name: "Device", + vendor: "Vendor", + trigger: "Trigger", + status: "Status", + progress: "Progress", + started: "Started", + ended: "Ended", + message: "Message", + }, + }, + networkConfigs: { + title: "Configurations", + keywordPh: "Name / IP / ID", + vendorPh: "Vendor", + allSource: "All sources", + empty: "No synced configs yet. Run Config sync first.", + view: "View", + close: "Close", + tabSet: "Set format", + tabHier: "Hierarchical", + col: { + name: "Device", + vendor: "Vendor", + source: "Source", + size: "Size", + collected: "Collected", + }, + }, layout: { title: "UME Integration", titleUme: "UME Sync", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 65663f1..c096464 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -49,8 +49,10 @@ const zh = { devices: "设备列表", topology: "拓扑信息", alarms: "告警信息", + configs: "配置信息", webcrt: "WebCRT", - configSync: "采集任务", + collectTasks: "采集任务", + configSync: "配置同步", portTraffic: "端口流量监控", }, placeholder: { @@ -64,6 +66,71 @@ const zh = { portTrafficStep4: "界面：任务列表 + 单端口趋势图", }, }, + configSync: { + title: "配置同步", + syncNow: "立即全量同步", + retryFailed: "一键重试失败", + pause: "暂停", + resume: "继续", + started: "已启动同步周期", + paused: "已暂停", + resumed: "已继续", + policyTitle: "同步策略", + policySaved: "策略已保存", + enabled: "启用周期调度", + intervalDays: "周期（天）", + concurrency: "并发", + historyKeep: "历史保留", + scope: "范围", + scopeAll: "全部网元", + scopeSelected: "指定网元", + savePolicy: "保存策略", + selectedCount: "已选 {{count}} 台", + targetKeywordPh: "名称 / IP / 厂商", + cyclesTitle: "同步周期", + cyclesEmpty: "暂无同步周期", + tasksTitle: "周期任务", + taskKeywordPh: "名称 / IP / 消息", + allStatus: "全部状态", + expand: "展开", + collapse: "收起", + kpi: { + snapshots: "配置覆盖数", + running: "进行中", + idle: "空闲", + last: "最近一轮", + nextDue: "下次调度", + }, + col: { + source: "来源", + name: "设备", + vendor: "厂商", + trigger: "触发", + status: "状态", + progress: "进度", + started: "开始", + ended: "结束", + message: "消息", + }, + }, + networkConfigs: { + title: "配置信息", + keywordPh: "名称 / IP / ID", + vendorPh: "厂商", + allSource: "全部来源", + empty: "暂无已同步配置。请先在「配置同步」执行采集。", + view: "查看", + close: "关闭", + tabSet: "Set 格式", + tabHier: "层级格式", + col: { + name: "设备", + vendor: "厂商", + source: "来源", + size: "大小", + collected: "采集时间", + }, + }, layout: { title: "UME 对接", titleUme: "UME同步", diff --git a/web/src/index.css b/web/src/index.css index 3bec087..e54efb3 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -758,7 +758,46 @@ pre { flex-wrap: wrap; } -/* ── NE stats card ── */ +.stat-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(160px, 1fr)); + gap: 12px; +} + +.stat-card { + border: 1px solid var(--border, #e2e8f0); + border-radius: 8px; + padding: 12px 14px; + background: var(--panel-bg, #fff); +} + +.stat-card strong { + display: block; + margin-top: 4px; + font-size: 1.05rem; +} + +.btn-row { + display: flex; + flex-wrap: wrap; + gap: 8px; + align-items: center; +} + +.btn-primary { + background: #2563eb; + color: #fff; + border-color: #1d4ed8; +} + +.btn-primary:disabled { + opacity: 0.55; +} + +.data-table { + width: 100%; +} + .ne-stats-card { padding: 12px 14px; } diff --git a/web/src/pages/ConfigSyncPage.tsx b/web/src/pages/ConfigSyncPage.tsx new file mode 100644 index 0000000..0f63af9 --- /dev/null +++ b/web/src/pages/ConfigSyncPage.tsx @@ -0,0 +1,496 @@ +import { useEffect, useMemo, useState } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { + createConfigSyncCycle, + fetchCliTargets, + fetchConfigSyncCycleTasks, + fetchConfigSyncCycles, + fetchConfigSyncDashboard, + pauseConfigSyncCycle, + resumeConfigSyncCycle, + updateConfigSyncPolicy, +} from "../services/api"; +import { queryKeys } from "../constants/queryKeys"; +import { useI18n } from "../i18n"; +import { useToast } from "../hooks/useToast"; +import type { CliTargetItem, ConfigSyncTargetRef } from "../types"; +import { pageCount } from "../utils/display"; +import { formatSystemTime } from "../utils/time"; + +const POLL_MS = 2500; +const TARGET_PAGE_SIZE = 20; + +export function ConfigSyncPage() { + const { t } = useI18n(); + const { showOk, showError } = useToast(); + const queryClient = useQueryClient(); + + const [cyclePage, setCyclePage] = useState(1); + const [expandedCycleId, setExpandedCycleId] = useState(""); + const [taskPage, setTaskPage] = useState(1); + const [taskStatus, setTaskStatus] = useState(""); + const [taskKeyword, setTaskKeyword] = useState(""); + + const [enabled, setEnabled] = useState(true); + const [intervalDays, setIntervalDays] = useState(3); + const [concurrency, setConcurrency] = useState(5); + const [scopeMode, setScopeMode] = useState<"all" | "selected">("all"); + const [historyKeep, setHistoryKeep] = useState(3); + const [selectedMap, setSelectedMap] = useState>({}); + const [policyHydrated, setPolicyHydrated] = useState(false); + + const [targetKeyword, setTargetKeyword] = useState(""); + const [targetPage, setTargetPage] = useState(1); + + const dashQuery = useQuery({ + queryKey: queryKeys.configSyncDashboard, + queryFn: fetchConfigSyncDashboard, + staleTime: 1000, + refetchInterval: (q) => { + const running = q.state.data?.running_cycle; + return running && (running.status === "running" || running.status === "paused") ? POLL_MS : false; + }, + }); + + useEffect(() => { + if (!dashQuery.data || policyHydrated) return; + const p = dashQuery.data.policy; + setEnabled(Boolean(p.enabled)); + setIntervalDays(Number(p.interval_days || 3)); + setConcurrency(Number(p.concurrency || 5)); + setScopeMode(p.scope_mode === "selected" ? "selected" : "all"); + setHistoryKeep(Number(p.history_keep ?? 3)); + const map: Record = {}; + for (const ref of p.selected_targets || []) { + map[`${ref.source}:${ref.id}`] = { source: ref.source, id: ref.id }; + } + setSelectedMap(map); + setPolicyHydrated(true); + }, [dashQuery.data, policyHydrated]); + + const cyclesQuery = useQuery({ + queryKey: queryKeys.configSyncCycles(cyclePage), + queryFn: () => fetchConfigSyncCycles({ page: cyclePage, pageSize: 10 }), + staleTime: 1000, + refetchInterval: () => (dashQuery.data?.running_cycle ? POLL_MS : false), + }); + + const tasksQuery = useQuery({ + queryKey: queryKeys.configSyncCycleTasks(expandedCycleId, taskPage, taskStatus, taskKeyword), + queryFn: () => + fetchConfigSyncCycleTasks({ + cycleId: expandedCycleId, + page: taskPage, + pageSize: 20, + status: taskStatus, + keyword: taskKeyword, + }), + enabled: Boolean(expandedCycleId), + staleTime: 800, + refetchInterval: () => (dashQuery.data?.running_cycle?.id === expandedCycleId ? POLL_MS : false), + }); + + const targetsQuery = useQuery({ + queryKey: queryKeys.cliTargets(targetKeyword, targetPage, TARGET_PAGE_SIZE), + queryFn: () => + fetchCliTargets({ source: "all", keyword: targetKeyword, page: targetPage, pageSize: TARGET_PAGE_SIZE }), + enabled: scopeMode === "selected", + staleTime: 5000, + }); + + const refresh = async () => { + await Promise.all([ + queryClient.invalidateQueries({ queryKey: queryKeys.configSyncDashboard }), + queryClient.invalidateQueries({ queryKey: queryKeys.configSyncCyclesAll }), + queryClient.invalidateQueries({ queryKey: queryKeys.configSyncCycleTasksAll }), + ]); + }; + + const savePolicyMut = useMutation({ + mutationFn: () => + updateConfigSyncPolicy({ + enabled, + interval_days: intervalDays, + concurrency, + scope_mode: scopeMode, + history_keep: historyKeep, + selected_targets: Object.values(selectedMap), + }), + onSuccess: async () => { + showOk(t("configSync.policySaved")); + setPolicyHydrated(false); + await refresh(); + }, + onError: (err) => showError(String(err)), + }); + + const startMut = useMutation({ + mutationFn: (mode: "full" | "retry_failed") => createConfigSyncCycle({ mode }), + onSuccess: async () => { + showOk(t("configSync.started")); + await refresh(); + }, + onError: (err) => showError(String(err)), + }); + + const pauseMut = useMutation({ + mutationFn: (id: string) => pauseConfigSyncCycle(id), + onSuccess: async () => { + showOk(t("configSync.paused")); + await refresh(); + }, + onError: (err) => showError(String(err)), + }); + + const resumeMut = useMutation({ + mutationFn: (id: string) => resumeConfigSyncCycle(id), + onSuccess: async () => { + showOk(t("configSync.resumed")); + await refresh(); + }, + onError: (err) => showError(String(err)), + }); + + const dash = dashQuery.data; + const running = dash?.running_cycle; + const last = dash?.last_cycle; + const cycles = cyclesQuery.data?.items ?? []; + const cycleTotal = Number(cyclesQuery.data?.total || 0); + const cyclePages = pageCount(cycleTotal, 10); + const selectedCount = useMemo(() => Object.keys(selectedMap).length, [selectedMap]); + + const toggleTarget = (row: CliTargetItem) => { + const source = row.source === "ume" ? "ume" : "managed"; + const key = `${source}:${row.id}`; + setSelectedMap((prev) => { + const next = { ...prev }; + if (next[key]) delete next[key]; + else next[key] = { source, id: row.id }; + return next; + }); + }; + + return ( +
+
+

{t("configSync.title")}

+
+ + + + {running?.status === "running" ? ( + + ) : null} + {running?.status === "paused" ? ( + + ) : null} +
+
+ +
+
+
{t("configSync.kpi.snapshots")}
+ {dash?.snapshot_count ?? "-"} +
+
+
{t("configSync.kpi.running")}
+ + {running + ? `${running.status} · ${running.success_count}/${running.planned_count}` + : t("configSync.kpi.idle")} + +
+
+
{t("configSync.kpi.last")}
+ + {last + ? `${last.status} · ok ${last.success_count} / fail ${last.fail_count}` + : t("common.empty")} + +
+
+
{t("configSync.kpi.nextDue")}
+ {dash?.next_due_at ? formatSystemTime(dash.next_due_at) : t("common.empty")} +
+
+ +
+

{t("configSync.policyTitle")}

+
+ + + + + + +
+ + {scopeMode === "selected" ? ( +
+

{t("configSync.selectedCount", { count: String(selectedCount) })}

+
+ { + setTargetKeyword(e.target.value); + setTargetPage(1); + }} + /> +
+ + + + + + + + + + + {(targetsQuery.data?.items ?? []).map((row) => { + const source = row.source === "ume" ? "ume" : "managed"; + const key = `${source}:${row.id}`; + return ( + + + + + + + + ); + })} + +
+ {t("configSync.col.source")}{t("configSync.col.name")}IP{t("configSync.col.vendor")}
+ toggleTarget(row)} + /> + {source}{row.name}{row.ip_address}{row.vendor || "-"}
+
+ + + {t("common.pagerMeta", { + total: String(targetsQuery.data?.total ?? 0), + page: String(targetPage), + pages: String(pageCount(Number(targetsQuery.data?.total || 0), TARGET_PAGE_SIZE)), + })} + + +
+
+ ) : null} +
+ +

{t("configSync.cyclesTitle")}

+ + + + + + + + + + + + + {cycles.map((c) => ( + + + + + + + + + + ))} + {!cycles.length ? ( + + + + ) : null} + +
+ ID{t("configSync.col.trigger")}{t("configSync.col.status")}{t("configSync.col.progress")}{t("configSync.col.started")}{t("configSync.col.ended")}
+ + {c.id.slice(0, 8)}{c.trigger_mode}{c.status} + {c.success_count}/{c.planned_count} · fail {c.fail_count} + {c.started_at ? formatSystemTime(c.started_at) : "-"}{c.ended_at ? formatSystemTime(c.ended_at) : "-"}
+ {t("configSync.cyclesEmpty")} +
+
+ + + {t("common.pagerMeta", { total: String(cycleTotal), page: String(cyclePage), pages: String(cyclePages) })} + + +
+ + {expandedCycleId ? ( +
+

{t("configSync.tasksTitle")}

+
+ { + setTaskKeyword(e.target.value); + setTaskPage(1); + }} + /> + +
+ + + + + + + + + + + + + {(tasksQuery.data?.items ?? []).map((task) => ( + + + + + + + + + ))} + +
{t("configSync.col.name")}IP{t("configSync.col.vendor")}{t("configSync.col.source")}{t("configSync.col.status")}{t("configSync.col.message")}
{task.ne_name || task.target_id}{task.ne_ip}{task.vendor || "-"}{task.source}{task.status}{task.message || "-"}
+
+ + + {t("common.pagerMeta", { + total: String(tasksQuery.data?.total ?? 0), + page: String(taskPage), + pages: String(pageCount(Number(tasksQuery.data?.total || 0), 20)), + })} + + +
+
+ ) : null} +
+ ); +} diff --git a/web/src/pages/network/NetworkConfigsPage.tsx b/web/src/pages/network/NetworkConfigsPage.tsx new file mode 100644 index 0000000..4cb0cd8 --- /dev/null +++ b/web/src/pages/network/NetworkConfigsPage.tsx @@ -0,0 +1,203 @@ +import { useState } from "react"; +import { Link } from "react-router-dom"; +import { useQuery } from "@tanstack/react-query"; +import { fetchNeConfigSnapshotDetail, fetchNeConfigSnapshots } from "../../services/api"; +import { queryKeys } from "../../constants/queryKeys"; +import { useI18n } from "../../i18n"; +import { pageCount } from "../../utils/display"; +import { formatSystemTime } from "../../utils/time"; + +function fmtBytes(n: number): string { + if (!n) return "0 B"; + if (n < 1024) return `${n} B`; + if (n < 1024 * 1024) return `${(n / 1024).toFixed(1)} KB`; + return `${(n / (1024 * 1024)).toFixed(2)} MB`; +} + +export function NetworkConfigsPage() { + const { t } = useI18n(); + const [page, setPage] = useState(1); + const [keyword, setKeyword] = useState(""); + const [source, setSource] = useState(""); + const [vendor, setVendor] = useState(""); + const [selected, setSelected] = useState<{ source: string; id: string } | null>(null); + const [tab, setTab] = useState<"primary" | "alt">("primary"); + + const listQuery = useQuery({ + queryKey: queryKeys.networkConfigs(page, keyword, source, vendor), + queryFn: () => + fetchNeConfigSnapshots({ page, pageSize: 20, keyword, source, vendor }), + staleTime: 5000, + }); + + const detailQuery = useQuery({ + queryKey: queryKeys.networkConfigDetail(selected?.source || "", selected?.id || ""), + queryFn: () => fetchNeConfigSnapshotDetail(selected!.source, selected!.id, "both"), + enabled: Boolean(selected), + staleTime: 10000, + }); + + const items = listQuery.data?.items ?? []; + const total = Number(listQuery.data?.total || 0); + const pages = pageCount(total, 20); + const detail = detailQuery.data; + const showAlt = Boolean(detail?.has_alt); + + return ( +
+
+

{t("networkConfigs.title")}

+
+ +
+ { + setKeyword(e.target.value); + setPage(1); + }} + /> + + { + setVendor(e.target.value); + setPage(1); + }} + /> + +
+ + {listQuery.isLoading ?

{t("common.refreshing")}

: null} + + + + + + + + + + + + + + {items.map((row) => ( + + + + + + + + + + ))} + {!items.length && !listQuery.isLoading ? ( + + + + ) : null} + +
{t("networkConfigs.col.name")}IP{t("networkConfigs.col.vendor")}{t("networkConfigs.col.source")}{t("networkConfigs.col.size")}{t("networkConfigs.col.collected")} +
{row.ne_name || row.target_id}{row.ne_ip}{row.vendor || "-"}{row.source}{fmtBytes(row.plain_size)}{row.collected_at ? formatSystemTime(row.collected_at) : "-"} + + + WebCRT + +
+ {t("networkConfigs.empty")} +
+ +
+ + + {t("common.pagerMeta", { total: String(total), page: String(page), pages: String(pages) })} + + +
+ + {selected ? ( +
+
+

+ {detail?.ne_name || selected.id}{" "} + + ({selected.source} / {detail?.ne_ip || "-"}) + +

+ +
+ {detailQuery.isLoading ?

{t("common.refreshing")}

: null} + {showAlt ? ( +
+ + +
+ ) : null} +