From f279431c70925e61ef662dcbed357fd0f2e1dfcc Mon Sep 17 00:00:00 2001 From: oliver Date: Sun, 20 Sep 2026 10:56:17 +0800 Subject: [PATCH] Run heavy CLI dumps on a dedicated long-timeout SSH lane. Split interface detail/route-class collects from the light lane so they no longer block other checks, and scope Collect Now busy state per task. Co-authored-by: Cursor --- netx_api/biz_state/collect_runner.py | 703 +++++++++++++++---------- netx_api/biz_state/profiles.py | 26 +- netx_api/config.py | 4 + tests/test_collect_lanes.py | 162 ++++++ web/src/pages/network/BizStatePage.tsx | 81 +-- 5 files changed, 675 insertions(+), 301 deletions(-) create mode 100644 tests/test_collect_lanes.py diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index ac8f96e..b9219ec 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -3,6 +3,8 @@ from __future__ import annotations import logging +import threading +from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Any from uuid import uuid4 @@ -39,6 +41,42 @@ from .profiles import get_profile _log = logging.getLogger("netx.biz_state.runner") +# concrete, params, profile_id, item_id, mode +WorkItem = tuple[str, dict[str, str], str, str, str] + +_heavy_pool_lock = threading.Lock() +_heavy_pool: ThreadPoolExecutor | None = None + + +def _heavy_cli_pool() -> ThreadPoolExecutor: + global _heavy_pool + with _heavy_pool_lock: + if _heavy_pool is None: + n = max(1, int(getattr(settings, "biz_state_heavy_workers", 4) or 4)) + _heavy_pool = ThreadPoolExecutor(max_workers=n, thread_name_prefix="biz-heavy") + return _heavy_pool + + +def work_item_lane(profile_id: str, mode: str) -> str: + """Return collect lane for a work item (custom_raw always light).""" + if str(mode or "").strip() == "custom": + return "light" + profile = get_profile(profile_id) or _resolve_collect_profile(profile_id) + lane = str(getattr(profile, "collect_lane", "light") or "light").strip().lower() if profile else "light" + return "heavy" if lane == "heavy" else "light" + + +def partition_work(work: list[WorkItem]) -> tuple[list[WorkItem], list[WorkItem]]: + """Split work into (light, heavy) lanes.""" + light: list[WorkItem] = [] + heavy: list[WorkItem] = [] + for item in work: + if work_item_lane(item[2], item[4]) == "heavy": + heavy.append(item) + else: + light.append(item) + return light, heavy + def _utcnow() -> datetime: return datetime.utcnow() @@ -311,6 +349,307 @@ def dispatch_collect(task_id: str, *, manual: bool = False) -> None: _finish_task(task_id, error=error) +def _run_collect_lane( + *, + work: list[WorkItem], + batch_id: str, + creds: dict[str, Any], + vendor_eff: str, + device_type_eff: str, + vendor_key: str, + per_cmd: int, + cap: int, + label: str, +) -> tuple[int, int, bool, bool]: + """Run one SSH lane (own connection + CollectSession + timeout budget).""" + if not work: + return 0, 0, False, False + + budget = min(int(cap), int(per_cmd) * max(1, len(work)) + 90) + holder: dict[str, Any] = {} + + def _session() -> tuple[int, int, bool, bool]: + from ..ne_netmiko import drain_read_channel + + conn = open_netmiko_connection(creds, session_timeout=budget) + holder["conn"] = conn + total_rows = 0 + cmd_count = 0 + any_fail = False + any_ok = False + try: + try: + disable_target_paging( + conn, + vendor=str(creds.get("vendor") or vendor_eff or ""), + device_type=str(creds.get("device_type") or device_type_eff or ""), + ) + except Exception: + pass + try: + drain_read_channel(conn) + except Exception: + pass + + sdb = SessionLocal() + session = CollectSession( + conn, + vendor=vendor_eff, + device_type=device_type_eff, + vendor_key=vendor_key, + read_timeout=per_cmd, + ) + try: + batch_row = sdb.get(BizStateBatch, batch_id) + if not batch_row: + return 0, 0, True, False + for concrete, params, profile_id, item_id, mode in work: + if holder.get("timed_out"): + raise TimeoutError(f"{label}_aborted") + cmd_count += 1 + cmd_row = BizStateBatchCommand( + id=uuid4().hex, + batch_id=batch_id, + task_item_id=item_id, + profile_id=profile_id, + raw_command=concrete[:512], + params_json=dict(params or {}), + created_at=_utcnow(), + ) + cache_hit_primary = False + try: + cached = session.get_cached(concrete) + if cached is not None and str(cached.raw or "").strip(): + cmd_row.raw_text = str(cached.raw or "") + cache_hit_primary = True + else: + raw = send_show_command(conn, concrete, read_timeout=per_cmd) + cmd_row.raw_text = str(raw or "") + except Exception as exc: + any_fail = True + cmd_row.parse_status = "failed" + cmd_row.message = _format_error(exc) + sdb.add(cmd_row) + sdb.commit() + continue + + if mode == "custom": + cmd_row.parse_status = "skipped_custom" + cmd_row.message = "custom_raw" + sdb.add(cmd_row) + sdb.commit() + any_ok = True + continue + + hit = match_command(vendor_key=vendor_key, command=concrete) + if not hit: + any_fail = True + cmd_row.parse_status = "unmatched" + cmd_row.message = "no profile matched concrete command" + sdb.add(cmd_row) + sdb.commit() + continue + + cmd_row.profile_id = hit.profile.profile_id + cmd_row.parser_id = hit.profile.parser_id + cmd_row.metric_id = hit.profile.metric_id + merged = {**params, **hit.params} + cmd_row.params_json = merged + + if not get_parser(hit.profile.parser_id): + any_fail = True + cmd_row.parse_status = "failed" + cmd_row.message = f"unknown parser {hit.profile.parser_id}" + sdb.add(cmd_row) + sdb.commit() + continue + + if cache_hit_primary: + pass + else: + session.remember( + concrete, + raw=cmd_row.raw_text or "", + ok=True, + cmd_row_id=cmd_row.id, + ) + + resolved_aux = [] + aux_results: dict[str, Any] = {} + for aux in list(hit.profile.aux_commands or []): + try: + ra = resolve_aux_command(aux, params=merged) + except ValueError as exc: + aux_row = BizStateBatchCommand( + id=uuid4().hex, + batch_id=batch_id, + task_item_id=item_id, + profile_id=str(aux.profile_id or "")[:128], + raw_command=str(aux.key or "")[:512], + params_json={}, + parse_status="aux_failed", + message=f"aux_for={cmd_row.id};resolve:{exc}"[:1020], + created_at=_utcnow(), + ) + cmd_count += 1 + sdb.add(aux_row) + sdb.commit() + continue + resolved_aux.append(ra) + aux_row = BizStateBatchCommand( + id=uuid4().hex, + batch_id=batch_id, + task_item_id=item_id, + profile_id=ra.profile_id, + parser_id=ra.parser_id, + metric_id="", + raw_command=ra.command[:512], + params_json={}, + created_at=_utcnow(), + ) + cmd_count += 1 + entry, cache_hit = session.fetch_and_parse( + ra.command, + parser_id=ra.parser_id, + textfsm_command=ra.textfsm_command, + params=merged, + cmd_row_id=aux_row.id, + ) + aux_results[ra.key] = entry + if cache_hit: + aux_row.parse_status = "aux_cached" + aux_row.message = ( + f"aux_for={cmd_row.id};cache_hit;src={entry.cmd_row_id}" + )[:1020] + aux_row.raw_text = "" + aux_row.row_count = len(entry.records or []) + elif not entry.ok: + aux_row.parse_status = "aux_failed" + aux_row.message = ( + f"aux_for={cmd_row.id};{entry.error}" + )[:1020] + aux_row.raw_text = entry.raw + else: + aux_row.parse_status = "aux" + aux_row.message = f"aux_for={cmd_row.id}"[:1020] + aux_row.raw_text = entry.raw + aux_row.row_count = len(entry.records or []) + entry.cmd_row_id = aux_row.id + sdb.add(aux_row) + sdb.commit() + + bundle = build_parse_bundle( + primary_raw=cmd_row.raw_text or "", + primary_parser_id=hit.profile.parser_id, + aux_results=aux_results, + resolved_aux=resolved_aux, + ) + try: + records, fsm_tables, rule_keys = run_primary_with_bundle( + hit.profile.parser_id, + bundle=bundle, + vendor=vendor_eff, + device_type=device_type_eff, + command=hit.profile.textfsm_command or concrete, + textfsm_command=hit.profile.textfsm_command or "", + params=merged, + enrich_joins=list(hit.profile.enrich_joins or []), + ) + session.remember( + concrete, + raw=cmd_row.raw_text or "", + fsm_tables=fsm_tables, + records=records, + ok=True, + cmd_row_id=cmd_row.id, + ) + hints = [] + if rule_keys: + nonempty = [k for k in rule_keys if fsm_tables.get(k)] + hints.append( + f"fsm_keys={','.join(rule_keys)};hit={','.join(nonempty)}" + ) + if bundle.aux_records: + hints.append( + "aux=" + + ",".join( + f"{k}:{len(v)}" for k, v in bundle.aux_records.items() + ) + ) + if hit.profile.enrich_joins: + hints.append( + "enrich=" + + ",".join(j.from_aux for j in hit.profile.enrich_joins) + ) + if hints: + cmd_row.message = ";".join(hints)[:1020] + except Exception as exc: + any_fail = True + cmd_row.parse_status = "failed" + cmd_row.message = f"parse: {_format_error(exc)}" + sdb.add(cmd_row) + sdb.commit() + continue + + n = 0 + if hit.profile.metric_id == "lldp_neighbor": + n = _persist_lldp_rows( + sdb, batch=batch_row, cmd_row=cmd_row, records=records + ) + elif hit.profile.metric_id in _GENERIC_METRICS: + n = _persist_metric_rows( + sdb, + batch=batch_row, + cmd_row=cmd_row, + metric_id=hit.profile.metric_id, + records=records, + ) + cmd_row.row_count = n + cmd_row.parse_status = "ok" + total_rows += n + any_ok = True + sdb.add(cmd_row) + sdb.commit() + finally: + sdb.close() + return total_rows, cmd_count, any_fail, any_ok + finally: + holder.pop("conn", None) + close_netmiko_connection(conn) + + try: + return run_cli_with_timeout( + _session, + timeout_sec=budget, + conn_holder=holder, + label=label, + acquire_budget=True, + ) + except TimeoutError as exc: + raise RuntimeError(str(exc)[:1020]) from exc + + +def _absorb_lane_result( + result: tuple[int, int, bool, bool] | BaseException, + *, + total_rows: int, + cmd_count: int, + any_fail: bool, + any_ok: bool, + lane_errors: list[str], +) -> tuple[int, int, bool, bool]: + if isinstance(result, BaseException): + lane_errors.append(_format_error(result)) + return total_rows, cmd_count, True, any_ok + rows, cmds, fail, ok = result + return ( + total_rows + int(rows or 0), + cmd_count + int(cmds or 0), + any_fail or bool(fail), + any_ok or bool(ok), + ) + + def _run_collect_session( *, task_id: str, @@ -320,8 +659,10 @@ def _run_collect_session( vendor: str, device_type: str, ) -> None: - per_cmd = int(settings.ne_collect_read_timeout_sec or 120) - cap = int(settings.ne_collect_run_timeout_cap_sec or 600) + light_per = int(settings.ne_collect_read_timeout_sec or 120) + light_cap = int(settings.ne_collect_run_timeout_cap_sec or 600) + heavy_per = int(getattr(settings, "biz_state_heavy_read_timeout_sec", 300) or 300) + heavy_cap = int(getattr(settings, "biz_state_heavy_run_timeout_cap_sec", 900) or 900) db = SessionLocal() try: @@ -365,8 +706,7 @@ def _run_collect_session( ) # Build work list before opening session - work: list[tuple[str, dict[str, str], str, str, str]] = [] - # concrete, params, profile_id, item_id, mode + work: list[WorkItem] = [] # Dedupe same CLI → same metric (e.g. legacy if_intf + config_interface). seen_work: set[tuple[str, str]] = set() for item in items: @@ -417,270 +757,94 @@ def _run_collect_session( db.commit() raise RuntimeError("no commands to run") - budget = min(cap, per_cmd * max(1, len(work)) + 90) - holder: dict[str, Any] = {} + light_work, heavy_work = partition_work(work) + lane_kwargs = dict( + batch_id=batch_id, + creds=creds, + vendor_eff=vendor_eff, + device_type_eff=device_type_eff, + vendor_key=vendor_key, + ) - def _session() -> tuple[int, int, bool, bool]: - from ..ne_netmiko import drain_read_channel - - conn = open_netmiko_connection(creds, session_timeout=budget) - holder["conn"] = conn - total_rows = 0 - cmd_count = 0 - any_fail = False - any_ok = False - try: - try: - disable_target_paging( - conn, - vendor=str(creds.get("vendor") or vendor_eff or ""), - device_type=str(creds.get("device_type") or device_type_eff or ""), - ) - except Exception: - pass - try: - drain_read_channel(conn) - except Exception: - pass - - sdb = SessionLocal() - session = CollectSession( - conn, - vendor=vendor_eff, - device_type=device_type_eff, - vendor_key=vendor_key, - read_timeout=per_cmd, - ) - try: - batch_row = sdb.get(BizStateBatch, batch_id) - if not batch_row: - return 0, 0, True, False - for concrete, params, profile_id, item_id, mode in work: - if holder.get("timed_out"): - raise TimeoutError("biz_state_aborted") - cmd_count += 1 - cmd_row = BizStateBatchCommand( - id=uuid4().hex, - batch_id=batch_id, - task_item_id=item_id, - profile_id=profile_id, - raw_command=concrete[:512], - params_json=dict(params or {}), - created_at=_utcnow(), - ) - cache_hit_primary = False - try: - cached = session.get_cached(concrete) - if cached is not None and str(cached.raw or "").strip(): - cmd_row.raw_text = str(cached.raw or "") - cache_hit_primary = True - else: - raw = send_show_command(conn, concrete, read_timeout=per_cmd) - cmd_row.raw_text = str(raw or "") - except Exception as exc: - any_fail = True - cmd_row.parse_status = "failed" - cmd_row.message = _format_error(exc) - sdb.add(cmd_row) - sdb.commit() - continue - - if mode == "custom": - cmd_row.parse_status = "skipped_custom" - cmd_row.message = "custom_raw" - sdb.add(cmd_row) - sdb.commit() - any_ok = True - continue - - hit = match_command(vendor_key=vendor_key, command=concrete) - if not hit: - any_fail = True - cmd_row.parse_status = "unmatched" - cmd_row.message = "no profile matched concrete command" - sdb.add(cmd_row) - sdb.commit() - continue - - cmd_row.profile_id = hit.profile.profile_id - cmd_row.parser_id = hit.profile.parser_id - cmd_row.metric_id = hit.profile.metric_id - merged = {**params, **hit.params} - cmd_row.params_json = merged - - if not get_parser(hit.profile.parser_id): - any_fail = True - cmd_row.parse_status = "failed" - cmd_row.message = f"unknown parser {hit.profile.parser_id}" - sdb.add(cmd_row) - sdb.commit() - continue - - if cache_hit_primary: - # Keep existing cache entry; primary will re-parse from shared raw. - pass - else: - session.remember( - concrete, - raw=cmd_row.raw_text or "", - ok=True, - cmd_row_id=cmd_row.id, - ) - - resolved_aux = [] - aux_results: dict[str, Any] = {} - for aux in list(hit.profile.aux_commands or []): - try: - ra = resolve_aux_command(aux, params=merged) - except ValueError as exc: - aux_row = BizStateBatchCommand( - id=uuid4().hex, - batch_id=batch_id, - task_item_id=item_id, - profile_id=str(aux.profile_id or "")[:128], - raw_command=str(aux.key or "")[:512], - params_json={}, - parse_status="aux_failed", - message=f"aux_for={cmd_row.id};resolve:{exc}"[:1020], - created_at=_utcnow(), - ) - cmd_count += 1 - sdb.add(aux_row) - sdb.commit() - continue - resolved_aux.append(ra) - aux_row = BizStateBatchCommand( - id=uuid4().hex, - batch_id=batch_id, - task_item_id=item_id, - profile_id=ra.profile_id, - parser_id=ra.parser_id, - metric_id="", - raw_command=ra.command[:512], - params_json={}, - created_at=_utcnow(), - ) - cmd_count += 1 - entry, cache_hit = session.fetch_and_parse( - ra.command, - parser_id=ra.parser_id, - textfsm_command=ra.textfsm_command, - params=merged, - cmd_row_id=aux_row.id, - ) - aux_results[ra.key] = entry - if cache_hit: - aux_row.parse_status = "aux_cached" - aux_row.message = ( - f"aux_for={cmd_row.id};cache_hit;src={entry.cmd_row_id}" - )[:1020] - aux_row.raw_text = "" - aux_row.row_count = len(entry.records or []) - elif not entry.ok: - aux_row.parse_status = "aux_failed" - aux_row.message = ( - f"aux_for={cmd_row.id};{entry.error}" - )[:1020] - aux_row.raw_text = entry.raw - else: - aux_row.parse_status = "aux" - aux_row.message = f"aux_for={cmd_row.id}"[:1020] - aux_row.raw_text = entry.raw - aux_row.row_count = len(entry.records or []) - # refresh cache row id to this aux row on first fetch - entry.cmd_row_id = aux_row.id - sdb.add(aux_row) - sdb.commit() - - bundle = build_parse_bundle( - primary_raw=cmd_row.raw_text or "", - primary_parser_id=hit.profile.parser_id, - aux_results=aux_results, - resolved_aux=resolved_aux, - ) - try: - records, fsm_tables, rule_keys = run_primary_with_bundle( - hit.profile.parser_id, - bundle=bundle, - vendor=vendor_eff, - device_type=device_type_eff, - command=hit.profile.textfsm_command or concrete, - textfsm_command=hit.profile.textfsm_command or "", - params=merged, - enrich_joins=list(hit.profile.enrich_joins or []), - ) - session.remember( - concrete, - raw=cmd_row.raw_text or "", - fsm_tables=fsm_tables, - records=records, - ok=True, - cmd_row_id=cmd_row.id, - ) - hints = [] - if rule_keys: - nonempty = [k for k in rule_keys if fsm_tables.get(k)] - hints.append( - f"fsm_keys={','.join(rule_keys)};hit={','.join(nonempty)}" - ) - if bundle.aux_records: - hints.append( - "aux=" - + ",".join( - f"{k}:{len(v)}" for k, v in bundle.aux_records.items() - ) - ) - if hit.profile.enrich_joins: - hints.append( - "enrich=" - + ",".join(j.from_aux for j in hit.profile.enrich_joins) - ) - if hints: - cmd_row.message = ";".join(hints)[:1020] - except Exception as exc: - any_fail = True - cmd_row.parse_status = "failed" - cmd_row.message = f"parse: {_format_error(exc)}" - sdb.add(cmd_row) - sdb.commit() - continue - - n = 0 - if hit.profile.metric_id == "lldp_neighbor": - n = _persist_lldp_rows( - sdb, batch=batch_row, cmd_row=cmd_row, records=records - ) - elif hit.profile.metric_id in _GENERIC_METRICS: - n = _persist_metric_rows( - sdb, - batch=batch_row, - cmd_row=cmd_row, - metric_id=hit.profile.metric_id, - records=records, - ) - cmd_row.row_count = n - cmd_row.parse_status = "ok" - total_rows += n - any_ok = True - sdb.add(cmd_row) - sdb.commit() - finally: - sdb.close() - return total_rows, cmd_count, any_fail, any_ok - finally: - holder.pop("conn", None) - close_netmiko_connection(conn) - - try: - total_rows, cmd_count, any_fail, any_ok = run_cli_with_timeout( - _session, - timeout_sec=budget, - conn_holder=holder, - label="biz_state", - acquire_budget=True, + def _run_light() -> tuple[int, int, bool, bool]: + return _run_collect_lane( + work=light_work, + per_cmd=light_per, + cap=light_cap, + label="biz_state_light", + **lane_kwargs, ) - except TimeoutError as exc: - raise RuntimeError(str(exc)[:1020]) from exc + + def _run_heavy() -> tuple[int, int, bool, bool]: + return _run_collect_lane( + work=heavy_work, + per_cmd=heavy_per, + cap=heavy_cap, + label="biz_state_heavy", + **lane_kwargs, + ) + + total_rows = 0 + cmd_count = 0 + any_fail = False + any_ok = False + lane_errors: list[str] = [] + + if light_work and heavy_work: + heavy_fut = _heavy_cli_pool().submit(_run_heavy) + light_res: tuple[int, int, bool, bool] | BaseException + try: + light_res = _run_light() + except Exception as exc: + light_res = exc + heavy_res: tuple[int, int, bool, bool] | BaseException + try: + heavy_res = heavy_fut.result() + except Exception as exc: + heavy_res = exc + total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result( + light_res, + total_rows=total_rows, + cmd_count=cmd_count, + any_fail=any_fail, + any_ok=any_ok, + lane_errors=lane_errors, + ) + total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result( + heavy_res, + total_rows=total_rows, + cmd_count=cmd_count, + any_fail=any_fail, + any_ok=any_ok, + lane_errors=lane_errors, + ) + elif heavy_work: + try: + total_rows, cmd_count, any_fail, any_ok = _run_heavy() + except Exception as exc: + total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result( + exc, + total_rows=0, + cmd_count=0, + any_fail=False, + any_ok=False, + lane_errors=lane_errors, + ) + else: + try: + total_rows, cmd_count, any_fail, any_ok = _run_light() + except Exception as exc: + total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result( + exc, + total_rows=0, + cmd_count=0, + any_fail=False, + any_ok=False, + lane_errors=lane_errors, + ) + + if lane_errors and not any_ok and cmd_count == 0: + raise RuntimeError("; ".join(lane_errors)[:1020]) batch = db.get(BizStateBatch, batch_id) if batch: @@ -689,11 +853,16 @@ def _run_collect_session( batch.ended_at = _utcnow() if any_fail and any_ok: batch.status = "partial" + if lane_errors: + batch.message = "; ".join(lane_errors)[:1020] elif any_fail and not any_ok: batch.status = "failed" - batch.message = "all commands failed" + batch.message = ( + "; ".join(lane_errors)[:1020] if lane_errors else "all commands failed" + ) else: batch.status = "success" + batch.message = "" db.commit() if batch.status in ("success", "partial"): try: diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index b9f5bfe..7d4ccb2 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -64,6 +64,8 @@ class ParseProfile: kind: str = "collect" # collect | discover aux_commands: list[AuxCommand] = field(default_factory=list) enrich_joins: list[EnrichJoin] = field(default_factory=list) + # light: default shared SSH lane; heavy: dedicated long-timeout connection + collect_lane: str = "light" _LLDP_FIELDS: list[FieldDef] = [ @@ -1270,11 +1272,32 @@ def _zte_status_profiles() -> list[ParseProfile]: _PROFILES: list[ParseProfile] | None = None +# Huge CLI dumps: dedicated heavy lane (longer read_timeout / own SSH). +_HEAVY_LANE_METRICS = frozenset( + { + "interface_detail", + "ip_route", + "ipv6_route", + "bgp_route", + "l2vpn_mac", + "evpn_mac", + } +) + + +def _apply_collect_lanes(profiles: list[ParseProfile]) -> list[ParseProfile]: + for p in profiles: + if str(p.metric_id or "").strip() in _HEAVY_LANE_METRICS: + p.collect_lane = "heavy" + return profiles + def all_profiles() -> list[ParseProfile]: global _PROFILES if _PROFILES is None: - _PROFILES = _lldp_profiles() + _vrf_profiles() + _zte_status_profiles() + _PROFILES = _apply_collect_lanes( + _lldp_profiles() + _vrf_profiles() + _zte_status_profiles() + ) return list(_PROFILES) @@ -1357,6 +1380,7 @@ def profile_to_public_dict(p: ParseProfile, *, overrides: dict[str, Any] | None "tags": list(p.tags), "sort_order": p.sort_order, "enabled": bool(ov.get("enabled")) if "enabled" in ov else p.enabled, + "collect_lane": str(p.collect_lane or "light"), "kind": p.kind, "match": p.match, "textfsm_command": p.textfsm_command or p.command_template, diff --git a/netx_api/config.py b/netx_api/config.py index 684e929..2b7b70c 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -107,6 +107,10 @@ class Settings(BaseSettings): biz_state_scheduler_enabled: bool = True biz_state_scheduler_tick_sec: int = 15 biz_state_dispatch_workers: int = 2 + # Heavy CLI lane (interface detail / routes / MAC): own SSH + longer timeouts. + biz_state_heavy_read_timeout_sec: int = 300 + biz_state_heavy_run_timeout_cap_sec: int = 900 + biz_state_heavy_workers: int = 4 # 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 (multi-operator concurrent terminals). diff --git a/tests/test_collect_lanes.py b/tests/test_collect_lanes.py new file mode 100644 index 0000000..f041536 --- /dev/null +++ b/tests/test_collect_lanes.py @@ -0,0 +1,162 @@ +"""Light/heavy collect lane partition and dual-lane scheduling.""" + +from __future__ import annotations + +import time +import unittest +from unittest.mock import patch + +from netx_api.biz_state.collect_runner import ( + partition_work, + work_item_lane, +) +from netx_api.biz_state.profiles import get_profile, reload_profiles + + +class CollectLaneProfileTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + reload_profiles() + + def test_interface_detail_is_heavy(self) -> None: + p = get_profile("zte.interface_detail") + self.assertIsNotNone(p) + assert p is not None + self.assertEqual(p.collect_lane, "heavy") + + def test_route_profiles_are_heavy(self) -> None: + for pid in ( + "zte.ip_route_vrf", + "zte.ip_route", + "zte.ipv6_route_vrf", + "zte.ipv6_route", + ): + p = get_profile(pid) + self.assertIsNotNone(p, pid) + assert p is not None + self.assertEqual(p.collect_lane, "heavy", pid) + + def test_arp_is_light(self) -> None: + p = get_profile("zte.arp") + self.assertIsNotNone(p) + assert p is not None + self.assertEqual(p.collect_lane, "light") + + +class PartitionWorkTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + reload_profiles() + + def test_custom_raw_always_light(self) -> None: + self.assertEqual(work_item_lane("", "custom"), "light") + self.assertEqual(work_item_lane("zte.interface_detail", "custom"), "light") + + def test_partition_mixed(self) -> None: + work = [ + ("show arp", {}, "zte.arp", "i1", "normal"), + ( + "show interface | include ifindex", + {}, + "zte.interface_detail", + "i2", + "normal", + ), + ("show ip forwarding route vrf a", {}, "zte.ip_route_vrf", "i3", "normal"), + ("show version", {}, "", "i4", "custom"), + ] + light, heavy = partition_work(work) + self.assertEqual(len(light), 2) + self.assertEqual({x[2] for x in light}, {"zte.arp", ""}) + self.assertEqual(len(heavy), 2) + self.assertEqual( + {x[2] for x in heavy}, + {"zte.interface_detail", "zte.ip_route_vrf"}, + ) + + def test_partition_light_only(self) -> None: + work = [("show arp", {}, "zte.arp", "i1", "normal")] + light, heavy = partition_work(work) + self.assertEqual(len(light), 1) + self.assertEqual(heavy, []) + + def test_partition_heavy_only(self) -> None: + work = [ + ( + "show interface | include ifindex", + {}, + "zte.interface_detail", + "i1", + "normal", + ) + ] + light, heavy = partition_work(work) + self.assertEqual(light, []) + self.assertEqual(len(heavy), 1) + + +class DualLaneParallelTests(unittest.TestCase): + def test_light_not_blocked_by_heavy_sleep(self) -> None: + """When both lanes run, light completes without waiting for heavy.""" + from netx_api.biz_state import collect_runner as cr + + order: list[str] = [] + + def fake_lane(*, work, label, **_kwargs): + if "heavy" in label: + time.sleep(0.35) + order.append("heavy_done") + return (0, len(work), False, True) + order.append("light_done") + return (0, len(work), False, True) + + light_work = [("show arp", {}, "zte.arp", "i1", "normal")] + heavy_work = [ + ( + "show interface | include ifindex", + {}, + "zte.interface_detail", + "i2", + "normal", + ) + ] + + with patch.object(cr, "_run_collect_lane", side_effect=fake_lane): + t0 = time.perf_counter() + heavy_fut = cr._heavy_cli_pool().submit( + lambda: cr._run_collect_lane( + work=heavy_work, + batch_id="b", + creds={}, + vendor_eff="zte", + device_type_eff="", + vendor_key="zte", + per_cmd=1, + cap=10, + label="biz_state_heavy", + ) + ) + light_res = cr._run_collect_lane( + work=light_work, + batch_id="b", + creds={}, + vendor_eff="zte", + device_type_eff="", + vendor_key="zte", + per_cmd=1, + cap=10, + label="biz_state_light", + ) + light_elapsed = time.perf_counter() - t0 + heavy_res = heavy_fut.result() + total_elapsed = time.perf_counter() - t0 + + self.assertEqual(light_res[1], 1) + self.assertEqual(heavy_res[1], 1) + self.assertEqual(order[0], "light_done") + self.assertLess(light_elapsed, 0.25) + self.assertGreaterEqual(total_elapsed, 0.3) + + +if __name__ == "__main__": + unittest.main() diff --git a/web/src/pages/network/BizStatePage.tsx b/web/src/pages/network/BizStatePage.tsx index 3543383..529355e 100644 --- a/web/src/pages/network/BizStatePage.tsx +++ b/web/src/pages/network/BizStatePage.tsx @@ -197,6 +197,7 @@ export function BizStatePage() { const [tasks, setTasks] = useState([]); const [busy, setBusy] = useState(false); + const [collectingIds, setCollectingIds] = useState>({}); const [listKeyword, setListKeyword] = useState(""); const debouncedListKw = useDebouncedValue(listKeyword, 250); const [purposeFilter, setPurposeFilter] = useState<"all" | "portrait" | "cutover_hf">("all"); @@ -253,7 +254,9 @@ export function BizStatePage() { const purpose = purposeFilter === "all" ? "" : purposeFilter === "portrait" ? "portrait" : "cutover_hf"; const res = await bizStateListTasks(purpose); - setTasks((res.items || []) as TaskRow[]); + const items = (res.items || []) as TaskRow[]; + setTasks(items); + return items; }, [purposeFilter]); useEffect(() => { @@ -619,27 +622,58 @@ export function BizStatePage() { } }; - const collectNow = async () => { - if (!taskId) return; - setBusy(true); + const setTaskCollecting = (id: string, on: boolean) => { + setCollectingIds((prev) => { + if (on) { + if (prev[id]) return prev; + return { ...prev, [id]: true }; + } + if (!prev[id]) return prev; + const next = { ...prev }; + delete next[id]; + return next; + }); + }; + + const collectNowForTask = async (id: string, fromModal = false) => { + setTaskCollecting(id, true); try { - await bizStateCollectNow(taskId); + await bizStateCollectNow(id); + showOk(t("bizState.collecting")); for (let i = 0; i < 20; i++) { await new Promise((r) => setTimeout(r, 1500)); - const task = await bizStateGetTask(taskId); - setDetail(task); - if (!task.collect_running) break; + if (fromModal && taskId === id) { + try { + const task = await bizStateGetTask(id); + setDetail(task); + await refreshTasks(); + if (!task.collect_running) break; + } catch { + break; + } + continue; + } + const items = await refreshTasks(); + const latest = items.find((x) => x.id === id); + if (!latest?.collect_running) break; } - await loadTask(taskId); await refreshTasks(); - setTaskTab("batches"); + if (fromModal && taskId === id) { + await loadTask(id); + setTaskTab("batches"); + } } catch (e) { showError(formatErr(e)); } finally { - setBusy(false); + setTaskCollecting(id, false); } }; + const collectNow = async () => { + if (!taskId) return; + await collectNowForTask(taskId, true); + }; + const removeTask = async (id: string) => { if (!window.confirm(t("bizState.confirmDelete"))) return; setBusy(true); @@ -939,27 +973,8 @@ export function BizStatePage() { @@ -1439,7 +1454,7 @@ export function BizStatePage() {