diff --git a/.env.example b/.env.example index 0e644df..8d28312 100644 --- a/.env.example +++ b/.env.example @@ -75,6 +75,9 @@ NETX_UME_NOTIFICATION_TOPIC=ALARM # NETX_BIZ_STATE_WORKER_REPLICAS=2 # NETX_BIZ_STATE_SPOOL_DIR=data/biz_state_spool # NETX_BIZ_STATE_PERSIST_EVERY_CMDS=8 +# Parse TextFSM off SSH threads (overlap with next CLI). Default on. +# NETX_BIZ_STATE_PARSE_ASYNC=true +# NETX_BIZ_STATE_PARSE_WORKERS=8 # Dual-lane CLI timeouts (light = shared SSH; heavy = interface/route/MAC etc.): # READ = per-command wait; RUN_TIMEOUT_CAP = whole-lane wall clock (often the ~40min abort). # Light lane reuses NE collect defaults below (read 120s / cap 600s ≈ 10min). diff --git a/netx_api/biz_migration/evaluate.py b/netx_api/biz_migration/evaluate.py index ec5847a..eb85841 100644 --- a/netx_api/biz_migration/evaluate.py +++ b/netx_api/biz_migration/evaluate.py @@ -9,9 +9,28 @@ from ..biz_state.compare_engine import apply_port_map, compare_rows PORT_METRIC_ID = "interface_brief" PORT_STATUS_FIELDS = ("admin", "phy", "prot") +# Unit separator — field values may contain "|"; keep "|" accepted when parsing expect. +KEY_SEP = "\x1f" + def _key_str(key: tuple[str, ...] | list[str]) -> str: - return "|".join(str(x) for x in key) + return KEY_SEP.join(str(x) for x in key) + + +def normalize_expect_key(raw: str) -> str: + """Normalize a stored/UI expect key to internal KEY_SEP form. + + Accepts legacy ``a|b`` multi-field keys and already-normalized ``\\x1f`` keys. + """ + s = str(raw or "").strip() + if not s: + return "" + if KEY_SEP in s: + return s + if "|" in s: + parts = [p.strip() for p in s.split("|")] + return KEY_SEP.join(parts) + return s def port_status_label(row: dict[str, Any] | None) -> str: @@ -69,7 +88,7 @@ def parse_expect_set(raw: dict[str, Any] | None) -> dict[str, set[str]]: ports = data.get("ports") or [] if isinstance(ports, list): for p in ports: - s = str(p or "").strip() + s = normalize_expect_key(str(p or "")) if s: out.setdefault(PORT_METRIC_ID, set()).add(s) out.setdefault("_ports", set()).add(s) @@ -86,11 +105,11 @@ def parse_expect_set(raw: dict[str, Any] | None) -> dict[str, set[str]]: if it.get("key") is not None: k = it.get("key") if isinstance(k, (list, tuple)): - joined = "|".join(str(p).strip() for p in k if str(p).strip()) + joined = KEY_SEP.join(str(p).strip() for p in k if str(p).strip()) if joined: bucket.add(joined) else: - s = str(k or "").strip() + s = normalize_expect_key(str(k or "")) if s: bucket.add(s) keys = it.get("keys") @@ -98,17 +117,17 @@ def parse_expect_set(raw: dict[str, Any] | None) -> dict[str, set[str]]: # Flat list of segments → one composite key; else each entry is a key # (string or nested list/tuple of segments). if keys and all(not isinstance(x, (list, tuple, dict)) for x in keys): - joined = "|".join(str(x).strip() for x in keys if str(x).strip()) + joined = KEY_SEP.join(str(x).strip() for x in keys if str(x).strip()) if joined: bucket.add(joined) else: for x in keys: if isinstance(x, (list, tuple)): - joined = "|".join(str(p).strip() for p in x if str(p).strip()) + joined = KEY_SEP.join(str(p).strip() for p in x if str(p).strip()) if joined: bucket.add(joined) else: - s = str(x or "").strip() + s = normalize_expect_key(str(x or "")) if s: bucket.add(s) return {k: {x for x in v if x} for k, v in out.items() if v} @@ -718,9 +737,16 @@ def dual_verdict_ex( def build_diff_index_from_compare(result: dict[str, Any]) -> dict[str, dict[str, Any]]: - """Normalize compare_rows result diffs → key_str → {kind, before, after, key}.""" + """Normalize compare_rows result diffs → key_str → {kind, before, after, key}. + + ``duplicate`` diffs are skipped for the index (primary match already indexed); + callers can still read ``summary.duplicate_key_list``. + """ out: dict[str, dict[str, Any]] = {} for d in result.get("diffs") or []: + kind = str(d.get("kind") or "") + if kind == "duplicate": + continue key = d.get("key") if isinstance(key, dict): key_list = [str(v) for v in key.values()] @@ -731,10 +757,12 @@ def build_diff_index_from_compare(result: dict[str, Any]) -> dict[str, dict[str, else: ks = str(key or "") key_list = [ks] if ks else [] + if ks in out: + continue # first primary match wins before = dict(d.get("before") or {}) if d.get("before") else {} after = dict(d.get("after") or {}) if d.get("after") else {} out[ks] = { - "kind": str(d.get("kind") or ""), + "kind": kind, "key": key_list, "before": before, "after": after, @@ -780,7 +808,11 @@ def _map_defines_iface_expect(metric_id: str, iface_fields: list[str], port_map: def _iface_values(key_str: str, *, key_fields: list[str], iface_fields: list[str]) -> list[str]: - parts = str(key_str or "").split("|") + raw = str(key_str or "") + if KEY_SEP in raw: + parts = raw.split(KEY_SEP) + else: + parts = raw.split("|") iface_set = {str(f) for f in iface_fields} if key_fields and len(parts) == len(key_fields): return [parts[i] for i, f in enumerate(key_fields) if f in iface_set and parts[i]] @@ -798,7 +830,7 @@ def _key_in_port_map( ) -> bool: """True when every interface segment of the key is covered by the map. - Covered = exact map key, or parent (before last ``.``) is a map key. + Covered = exact map key, or any parent prefix (multi-level QinQ) is a map key. """ vals = _iface_values(key_str, key_fields=key_fields, iface_fields=iface_fields) if not vals: @@ -806,10 +838,15 @@ def _key_in_port_map( for v in vals: if v in port_map: continue - parent = v.rsplit(".", 1)[0] if "." in v else "" - if parent and parent in port_map: - continue - return False + parts = v.split(".") + covered = False + for i in range(len(parts) - 1, 0, -1): + parent = ".".join(parts[:i]) + if parent and parent in port_map: + covered = True + break + if not covered: + return False return True @@ -834,6 +871,13 @@ def _unmapped_same_iface_anomaly( return old_bad and new_up +def _split_key_parts(key_str: str) -> list[str]: + raw = str(key_str or "") + if KEY_SEP in raw: + return raw.split(KEY_SEP) + return raw.split("|") + + def _remap_key_str( key_str: str, *, @@ -849,10 +893,14 @@ def _remap_key_str( if not port_map: return key_str if (not key_fields or len(key_fields) == 1) and ( - key_str in port_map or ("." in key_str and key_str.rsplit(".", 1)[0] in port_map) + key_str in port_map + or any( + ".".join(key_str.split(".")[:i]) in port_map + for i in range(len(key_str.split(".")) - 1, 0, -1) + ) ): return resolve_mapped_iface(key_str, port_map) - parts = str(key_str).split("|") + parts = _split_key_parts(key_str) if key_fields and len(parts) == len(key_fields): iface_set = {str(f) for f in iface_fields} out: list[str] = [] @@ -862,7 +910,7 @@ def _remap_key_str( out.append(resolve_mapped_iface(v, port_map)) else: out.append(v) - return "|".join(out) + return KEY_SEP.join(out) return resolve_mapped_iface(key_str, port_map) @@ -881,10 +929,14 @@ def _reverse_remap_key_str( if not rev_map: return key_str if (not key_fields or len(key_fields) == 1) and ( - key_str in rev_map or ("." in key_str and key_str.rsplit(".", 1)[0] in rev_map) + key_str in rev_map + or any( + ".".join(key_str.split(".")[:i]) in rev_map + for i in range(len(key_str.split(".")) - 1, 0, -1) + ) ): return resolve_mapped_iface(key_str, rev_map) - parts = str(key_str).split("|") + parts = _split_key_parts(key_str) if key_fields and len(parts) == len(key_fields): iface_set = {str(f) for f in iface_fields} out: list[str] = [] @@ -894,7 +946,7 @@ def _reverse_remap_key_str( out.append(resolve_mapped_iface(v, rev_map)) else: out.append(v) - return "|".join(out) + return KEY_SEP.join(out) return resolve_mapped_iface(key_str, rev_map) @@ -957,7 +1009,7 @@ def row_match_key( data = strip_netx(row) if iface_fields and rules: data = apply_iface_normalize(dict(data), iface_fields=iface_fields, rules=rules) - return "|".join(str(data.get(k) or "").strip() for k in key_fields) + return KEY_SEP.join(str(data.get(k) or "").strip() for k in key_fields) def index_raw_by_match_key( @@ -1395,12 +1447,22 @@ def evaluate_metric_dual( "old_summary": old_cmp.get("summary") or {}, "new_summary": new_cmp.get("summary") or {}, "progress_ok": progress_ok, - "progress_total": progress_total if progress_total else len(expect_keys), + "progress_total": progress_total, "anomaly": anomaly, "anomaly_in_expect": anomaly_in_expect, "rows": rows_out, "new_baseline_mode": new_baseline_mode, "new_baseline_missing": new_baseline_missing, + "duplicate_keys_before": int((old_cmp.get("summary") or {}).get("duplicate_keys_before") or 0) + + int((new_cmp.get("summary") or {}).get("duplicate_keys_before") or 0), + "duplicate_keys_after": int((old_cmp.get("summary") or {}).get("duplicate_keys_after") or 0) + + int((new_cmp.get("summary") or {}).get("duplicate_keys_after") or 0), + "duplicate_key_list": list( + dict.fromkeys( + list((old_cmp.get("summary") or {}).get("duplicate_key_list") or []) + + list((new_cmp.get("summary") or {}).get("duplicate_key_list") or []) + ) + ), } diff --git a/netx_api/biz_migration/monitor_templates.py b/netx_api/biz_migration/monitor_templates.py index ac0bce2..b55f487 100644 --- a/netx_api/biz_migration/monitor_templates.py +++ b/netx_api/biz_migration/monitor_templates.py @@ -33,7 +33,7 @@ def _out(row: BizMonitorTemplate, compare_name: str = "", *, db: Session | None effective = seen if not effective: effective = [PORT_METRIC_ID] - return { + out: dict[str, Any] = { "id": row.id, "name": row.name, "compare_template_id": row.compare_template_id or "", @@ -46,6 +46,82 @@ def _out(row: BizMonitorTemplate, compare_name: str = "", *, db: Session | None "created_at": row.created_at.isoformat() + "Z" if row.created_at else None, "updated_at": row.updated_at.isoformat() + "Z" if row.updated_at else None, } + if db is not None: + warnings = validate_sheet_overrides( + db, + compare_template_id=str(row.compare_template_id or ""), + sheet_overrides=list(row.sheet_overrides_json or []), + ) + if warnings: + out["override_warnings"] = warnings + return out + + +def validate_sheet_overrides( + db: Session, + *, + compare_template_id: str, + sheet_overrides: list[Any] | None, +) -> list[dict[str, str]]: + """Return warnings when overrides no longer match compare template sheets. + + Does not block save — surfaces ``override_warnings`` so UI/ops can fix drift. + """ + warnings: list[dict[str, str]] = [] + overrides = [o for o in (sheet_overrides or []) if isinstance(o, dict)] + if not overrides: + return warnings + ct = db.get(BizCompareTemplate, compare_template_id) if compare_template_id else None + if not ct: + for ov in overrides: + sid = str(ov.get("sheet_id") or "").strip() + mid = str(ov.get("metric_id") or "").strip() + warnings.append( + { + "sheet_id": sid, + "metric_id": mid, + "reason": "compare_template_missing", + } + ) + return warnings + sheets = cmp_svc.template_metrics(ct) + sheet_ids = {str(s.get("sheet_id") or s.get("metric_id") or "").strip() for s in sheets} + metric_ids = {str(s.get("metric_id") or "").strip() for s in sheets} + for ov in overrides: + sid = str(ov.get("sheet_id") or "").strip() + mid = str(ov.get("metric_id") or "").strip() + if sid: + if sid not in sheet_ids: + warnings.append( + { + "sheet_id": sid, + "metric_id": mid, + "reason": "sheet_id_not_in_compare_template", + } + ) + elif mid: + if mid not in metric_ids and mid not in sheet_ids: + warnings.append( + { + "sheet_id": "", + "metric_id": mid, + "reason": "metric_id_not_in_compare_template", + } + ) + elif mid in metric_ids: + # Legacy metric-only override applies to every split of that metric + split_count = sum( + 1 for s in sheets if str(s.get("metric_id") or "").strip() == mid + ) + if split_count > 1: + warnings.append( + { + "sheet_id": "", + "metric_id": mid, + "reason": "legacy_metric_override_applies_to_all_splits", + } + ) + return warnings def _compare_name_map(db: Session) -> dict[str, str]: @@ -395,7 +471,8 @@ def create_monitor_template(db: Session, body: dict[str, Any]) -> dict[str, Any] db.commit() db.refresh(row) names = _compare_name_map(db) - return _out(row, names.get(row.compare_template_id or "", ""), db=db) + out = _out(row, names.get(row.compare_template_id or "", ""), db=db) + return out def update_monitor_template(db: Session, template_id: str, body: dict[str, Any]) -> dict[str, Any]: diff --git a/netx_api/biz_migration/service.py b/netx_api/biz_migration/service.py index 4ddf94b..760a6c0 100644 --- a/netx_api/biz_migration/service.py +++ b/netx_api/biz_migration/service.py @@ -10,7 +10,14 @@ from fastapi import HTTPException from sqlalchemy.orm import Session from ..biz_state.compare_rules import apply_row_filters -from ..biz_state.compare_service import _load_metric_rows, _port_map_dict, sheet_key, sheet_title, template_metrics +from ..biz_state.compare_service import ( + _load_metric_rows, + _port_map_dict, + batch_metric_collect_ok, + sheet_key, + sheet_title, + template_metrics, +) from ..models import ( BizCompareTemplate, BizMigrationBatch, @@ -260,8 +267,14 @@ def _current_batch_for_metric( pinned = db.get(BizStateBatch, pin) # Only honor pin when it belongs to this metric's HF task (multi-interval safe) if pinned and (not tid or str(pinned.task_id or "") == tid): + # Pinned partial / failed metric → treat as missing (avoid false red) + if not batch_metric_collect_ok(db, pinned.id, metric_id): + return None return pinned - return _latest_success_batch(db, tid) if tid else None + batch = _latest_success_batch(db, tid) if tid else None + if batch and not batch_metric_collect_ok(db, batch.id, metric_id): + return None + return batch def find_portrait_task_for_ne(db: Session, *, source: str, ne_id: str) -> BizStateTask | None: @@ -772,11 +785,15 @@ def create_batch(db: Session, project_id: str, body: dict[str, Any]) -> dict[str def _latest_success_batch(db: Session, task_id: str) -> BizStateBatch | None: + """Latest fully successful batch only — never use ``partial`` as current. + + Partial batches omit failed-command metrics and cause mass false ``removed``. + """ return ( db.query(BizStateBatch) .filter( BizStateBatch.task_id == task_id, - BizStateBatch.status.in_(("success", "partial")), + BizStateBatch.status == "success", ) .order_by(BizStateBatch.started_at.desc()) .first() @@ -903,6 +920,7 @@ def run_evaluate( "new_baseline_mode": "missing_current", "new_baseline_missing": False, "current_missing": True, + "collect_incomplete": True, "collect_skipped": False, } ) @@ -972,6 +990,9 @@ def run_evaluate( "collect_skipped": False, "old_batch_id": old_cur_mid, "new_batch_id": new_cur_mid, + "duplicate_key_list": list(one.get("duplicate_key_list") or []), + "duplicate_keys_before": int(one.get("duplicate_keys_before") or 0), + "duplicate_keys_after": int(one.get("duplicate_keys_after") or 0), } ) for r in one["rows"]: @@ -1811,6 +1832,7 @@ def _red_ticket_to_dict(t: BizMigrationRedTicket) -> dict[str, Any]: "new_key": new_key, "key_str": old_key, "new_key_str": new_key, + "match_key_str": str(getattr(t, "match_key_str", "") or detail.get("match_old_key") or ""), "match_old_key": str(detail.get("match_old_key") or ""), "match_new_key": str(detail.get("match_new_key") or ""), "verdict": t.verdict, @@ -1881,45 +1903,99 @@ def _persist_red_tickets_from_run( batch_id: str, run_id: str, ) -> list[BizMigrationRedTicket]: - """Create open red tickets from acceptance run red diffs (expect + anomaly).""" + """Upsert open/carried red tickets by (project, metric, match_key). + + Same issue across batches / re-acceptance updates one ticket instead of + spawning duplicates. ``carried`` tickets are reopened when the key is still red. + """ diffs = ( db.query(BizMigrationDiff) .filter(BizMigrationDiff.run_id == run_id, BizMigrationDiff.color == "red") .order_by(BizMigrationDiff.seq.asc()) .all() ) - created: list[BizMigrationRedTicket] = [] + existing = ( + db.query(BizMigrationRedTicket) + .filter( + BizMigrationRedTicket.project_id == project_id, + BizMigrationRedTicket.status.in_(("open", "carried")), + ) + .all() + ) + by_id: dict[tuple[str, str], BizMigrationRedTicket] = {} + for t in existing: + mk = str(getattr(t, "match_key_str", "") or "").strip() + if not mk: + detail = t.detail_json if isinstance(t.detail_json, dict) else {} + mk = str(detail.get("match_old_key") or t.key_str or "").strip() + if mk: + by_id[(str(t.metric_id or ""), mk)] = t + + created_or_updated: list[BizMigrationRedTicket] = [] + seen: set[tuple[str, str]] = set() for d in diffs: kj = d.key_json if isinstance(d.key_json, dict) else {} + mid = str(d.metric_id or "") + match_key = str( + kj.get("match_old_key") or kj.get("key_str") or kj.get("old_key") or "" + ).strip()[:256] + if not match_key: + match_key = str(kj.get("key_str") or "")[:256] + ident = (mid, match_key) + detail = { + "old_kind": d.old_kind, + "new_kind": d.new_kind, + "in_expect": d.in_expect, + "old": d.old_json, + "new": d.new_json, + "old_key": kj.get("old_key") or kj.get("key_str") or "", + "new_key": kj.get("new_key") or kj.get("new_key_str") or "", + "match_old_key": kj.get("match_old_key") or match_key, + "match_new_key": kj.get("match_new_key") or "", + "evidence": dict(kj.get("evidence") or {}), + } + prev = by_id.get(ident) + if prev is not None: + prev.batch_id = batch_id + prev.run_id = run_id + prev.key_str = str(kj.get("key_str") or "")[:256] + prev.new_key_str = str(kj.get("new_key_str") or "")[:256] + prev.match_key_str = match_key + prev.verdict = d.verdict + prev.color = d.color or "red" + prev.old_status = str(kj.get("old_status") or "")[:64] + prev.new_status = str(kj.get("new_status") or "")[:64] + prev.detail_json = detail + prev.status = "open" + prev.carried_to_batch_id = "" + prev.resolved_at = None + created_or_updated.append(prev) + seen.add(ident) + continue t = BizMigrationRedTicket( id=uuid4().hex, project_id=project_id, batch_id=batch_id, run_id=run_id, - metric_id=d.metric_id, + metric_id=mid, key_str=str(kj.get("key_str") or "")[:256], new_key_str=str(kj.get("new_key_str") or "")[:256], + match_key_str=match_key, verdict=d.verdict, color=d.color or "red", old_status=str(kj.get("old_status") or "")[:64], new_status=str(kj.get("new_status") or "")[:64], - detail_json={ - "old_kind": d.old_kind, - "new_kind": d.new_kind, - "in_expect": d.in_expect, - "old": d.old_json, - "new": d.new_json, - "old_key": kj.get("old_key") or kj.get("key_str") or "", - "new_key": kj.get("new_key") or kj.get("new_key_str") or "", - "match_old_key": kj.get("match_old_key") or "", - "match_new_key": kj.get("match_new_key") or "", - "evidence": dict(kj.get("evidence") or {}), - }, + detail_json=detail, status="open", ) db.add(t) - created.append(t) - return created + by_id[ident] = t + created_or_updated.append(t) + seen.add(ident) + + # Tickets open on this batch but no longer red → leave as open (operator resolves). + # Carried tickets for keys not in this run stay carried. + return created_or_updated def finish_batch( @@ -1990,11 +2066,7 @@ def finish_batch( mb.accept_summary_json = accept_summary mb.updated_at = utcnow_naive() - # Replace prior open tickets from this batch's previous acceptance (re-finish) - db.query(BizMigrationRedTicket).filter( - BizMigrationRedTicket.batch_id == batch_id, - BizMigrationRedTicket.status == "open", - ).delete() + # Upsert reds (no wipe — merge by metric+match_key across re-acceptance) reds = _persist_red_tickets_from_run( db, project_id=mb.project_id, diff --git a/netx_api/biz_state/collect_runner.py b/netx_api/biz_state/collect_runner.py index 9375d79..79a95be 100644 --- a/netx_api/biz_state/collect_runner.py +++ b/netx_api/biz_state/collect_runner.py @@ -30,8 +30,11 @@ from ..models import ( ) from ..ne_netmiko import disable_target_paging, send_show_command from ..ne_session_factory import close_netmiko_connection, open_netmiko_connection +from ..timeutil import utcnow_naive from .collect_session import ( + CachedCommand, CollectSession, + ResolvedAux, build_parse_bundle, resolve_aux_command, run_primary_with_bundle, @@ -43,7 +46,7 @@ from .command_match import ( match_command, normalize_command, ) -from .parsers import get_parser +from .parsers import get_parser, run_parser from .profiles import get_profile from .collect_stop import ( STOP_USER_MESSAGE, @@ -93,7 +96,7 @@ def partition_work(work: list[WorkItem]) -> tuple[list[WorkItem], list[WorkItem] def _utcnow() -> datetime: - return datetime.utcnow() + return utcnow_naive() def _format_error(exc: BaseException) -> str: @@ -215,6 +218,201 @@ _GENERIC_METRICS = { _METRIC_CHUNK = 2000 +def _run_primary_parse_job(job: Any) -> tuple[bool, bool]: + """Parse primary+aux raws, write spool, submit persist. Returns (any_ok, any_fail).""" + from .parse_pool import AuxRawCapture, PrimaryParseJob + from .persist_pool import get_persist_pool + from .spool import SpooledCommand, write_meta, write_records + + if not isinstance(job, PrimaryParseJob): + return False, True + + batch_id = job.batch_id + cmd_id = job.cmd_id + pending: list[SpooledCommand] = [] + any_ok = False + any_fail = False + + def _flush_item(item: SpooledCommand, *, records: list[dict[str, Any]] | None = None) -> None: + if records is not None and item.persist_kind: + item.records_rel_path = write_records(batch_id, item.id, records) + item.row_count = len(records) + try: + write_meta(batch_id, item.id, item.to_meta()) + except Exception: + _log.exception("biz_state write meta failed cmd=%s", item.id) + pending.append(item) + + aux_results: dict[str, CachedCommand] = {} + resolved_aux: list[ResolvedAux] = [] + + for cap in list(job.aux_captures or []): + if not isinstance(cap, AuxRawCapture): + continue + from .profiles import get_profile as _gp + + prof = _gp(cap.profile_id) + entry = CachedCommand( + raw=cap.raw, + records=list(cap.records or []), + fsm_tables=dict(cap.fsm_tables or {}), + ok=bool(cap.ok), + error=str(cap.error or ""), + cmd_row_id=cap.aux_id, + ) + if entry.ok and not entry.records and cap.parser_id and get_parser(cap.parser_id): + try: + records, fsm_tables, _keys = run_parser( + cap.parser_id, + raw_text=cap.raw, + vendor=job.vendor, + device_type=job.device_type, + command=cap.textfsm_command or cap.command, + textfsm_command=cap.textfsm_command or "", + params=dict(job.merged_params or {}), + ) + entry.records = list(records or []) + entry.fsm_tables = dict(fsm_tables or {}) + except Exception as exc: + entry.ok = False + entry.error = f"parse: {type(exc).__name__}: {exc}" + + aux_results[cap.key] = entry + if prof is not None: + resolved_aux.append( + ResolvedAux( + key=cap.key, + profile_id=cap.profile_id, + command=cap.command, + textfsm_command=cap.textfsm_command or cap.command, + parser_id=cap.parser_id, + rule_keys=tuple(cap.rule_keys or ()), + profile=prof, + ) + ) + + aux_sp = SpooledCommand( + id=cap.aux_id, + batch_id=batch_id, + task_item_id=job.task_item_id, + profile_id=cap.profile_id, + parser_id=cap.parser_id, + metric_id=cap.metric_id, + raw_command=cap.command[:512], + params_json={}, + raw_rel_path=cap.raw_rel_path, + ) + if cap.cache_hit and entry.ok and entry.records: + aux_sp.parse_status = "aux_cached" + aux_sp.message = ( + f"aux_for={cmd_id};cache_hit;src={entry.cmd_row_id}" + )[:1020] + aux_sp.row_count = len(entry.records or []) + elif not entry.ok: + aux_sp.parse_status = "aux_failed" + aux_sp.message = f"aux_for={cmd_id};{entry.error}"[:1020] + any_fail = True + else: + aux_sp.parse_status = "aux" + aux_sp.message = f"aux_for={cmd_id}"[:1020] + aux_sp.row_count = len(entry.records or []) + + persist_recs: list[dict[str, Any]] | None = None + if entry.ok and entry.records and cap.metric_id in _GENERIC_METRICS: + persist_key = (normalize_command(cap.command), cap.metric_id) + do_persist = False + persisted = job.persisted + if persisted is not None: + if job.cache_lock is not None: + with job.cache_lock: + if persist_key not in persisted: + persisted.add(persist_key) + do_persist = True + elif persist_key not in persisted: + persisted.add(persist_key) + do_persist = True + if do_persist: + aux_sp.persist_kind = "metric" + persist_recs = list(entry.records) + _flush_item(aux_sp, records=persist_recs) + + bundle = build_parse_bundle( + primary_raw=job.raw_text, + primary_parser_id=job.parser_id, + aux_results=aux_results, + resolved_aux=resolved_aux, + ) + # Include aux raws that lacked a profile (still needed for multi-raw parsers). + for cap in list(job.aux_captures or []): + if not isinstance(cap, AuxRawCapture): + continue + if cap.key in bundle.raws: + continue + entry = aux_results.get(cap.key) or CachedCommand(ok=False) + bundle.raws[cap.key] = entry.raw + bundle.command_rules[cap.key] = list(cap.rule_keys or []) + if entry.records: + bundle.aux_records[cap.key] = list(entry.records) + bundle.fsm_extra.update(entry.fsm_tables or {}) + primary = SpooledCommand( + id=cmd_id, + batch_id=batch_id, + task_item_id=job.task_item_id, + profile_id=job.profile_id, + parser_id=job.parser_id, + metric_id=job.metric_id, + raw_command=job.concrete[:512], + params_json=dict(job.merged_params or {}), + raw_rel_path=job.raw_rel_path, + ) + try: + records, fsm_tables, rule_keys = run_primary_with_bundle( + job.parser_id, + bundle=bundle, + vendor=job.vendor, + device_type=job.device_type, + command=job.textfsm_command or job.concrete, + textfsm_command=job.textfsm_command or "", + params=dict(job.merged_params or {}), + enrich_joins=list(job.enrich_joins or []), + ) + 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 job.enrich_joins: + hints.append( + "enrich=" + ",".join(getattr(j, "from_aux", "") for j in job.enrich_joins) + ) + if hints: + primary.message = ";".join(hints)[:1020] + primary.parse_status = "ok" + any_ok = True + persist_recs = None + if job.metric_id == "lldp_neighbor": + primary.persist_kind = "lldp" + persist_recs = list(records or []) + elif job.metric_id in _GENERIC_METRICS: + primary.persist_kind = "metric" + persist_recs = list(records or []) + _flush_item(primary, records=persist_recs) + _ = fsm_tables # kept for hints above + except Exception as exc: + any_fail = True + primary.parse_status = "failed" + primary.message = f"parse: {_format_error(exc)}" + _flush_item(primary) + + if pending: + get_persist_pool().submit(batch_id, pending) + return any_ok, any_fail + + def _emit_task_event(*, task_id: str, message: str, level: str = "error") -> None: """Short-lived session for lane events (no long-held DB during CLI).""" if not task_id or not str(message or "").strip(): @@ -547,9 +745,26 @@ def _run_collect_lane( any_ok = False pending: list[SpooledCommand] = [] task_id = "" + from .parse_pool import ( + AuxRawCapture, + PrimaryParseJob, + get_parse_pool, + parse_async_enabled, + ) from .persist_pool import get_persist_pool persist = get_persist_pool() + parse_pool = get_parse_pool() if parse_async_enabled() else None + parse_stats_lock = threading.Lock() + parse_stats = {"ok": False, "fail": False, "pending": 0} + + def _on_parse_done(ok: bool, fail: bool) -> None: + with parse_stats_lock: + if ok: + parse_stats["ok"] = True + if fail: + parse_stats["fail"] = True + parse_stats["pending"] = max(0, int(parse_stats["pending"]) - 1) def _submit_pending() -> None: nonlocal pending @@ -763,8 +978,8 @@ def _run_collect_lane( cmd_row_id=cmd_id, ) - resolved_aux = [] - aux_results: dict[str, Any] = {} + # Collect aux raws on the SSH thread (no TextFSM); parse overlaps next CLI. + aux_captures: list[AuxRawCapture] = [] for aux in list(hit.profile.aux_commands or []): try: ra = resolve_aux_command(aux, params=merged) @@ -781,145 +996,105 @@ def _run_collect_lane( ) ) continue - resolved_aux.append(ra) aux_id = uuid4().hex - 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_id, - ) - aux_results[ra.key] = entry + entry, cache_hit = session.fetch_raw(ra.command, cmd_row_id=aux_id) aux_mid = str(getattr(ra.profile, "metric_id", "") or "").strip() - aux_sp = SpooledCommand( - id=aux_id, - batch_id=batch_id, - task_item_id=item_id, - profile_id=ra.profile_id, - parser_id=ra.parser_id, - metric_id=aux_mid, - raw_command=ra.command[:512], - params_json={}, + aux_rel = "" + if entry.raw: + try: + aux_rel = write_raw_text(batch_id, aux_id, entry.raw or "") + except Exception: + _log.exception( + "biz_state spool aux raw failed cmd=%s", aux_id + ) + if not entry.ok: + any_fail = True + _queue( + SpooledCommand( + id=aux_id, + batch_id=batch_id, + task_item_id=item_id, + profile_id=ra.profile_id, + parser_id=ra.parser_id, + metric_id=aux_mid, + raw_command=ra.command[:512], + parse_status="aux_failed", + message=f"aux_for={cmd_id};{entry.error}"[:1020], + raw_rel_path=aux_rel, + ) + ) + continue + # If cache already has parsed records, pass them through. + aux_captures.append( + AuxRawCapture( + key=ra.key, + aux_id=aux_id, + profile_id=ra.profile_id, + parser_id=ra.parser_id, + metric_id=aux_mid, + command=ra.command, + textfsm_command=ra.textfsm_command, + rule_keys=tuple(ra.rule_keys or ()), + raw=entry.raw, + raw_rel_path=aux_rel, + cache_hit=bool(cache_hit and entry.records), + records=list(entry.records or []), + fsm_tables=dict(entry.fsm_tables or {}), + ok=True, + ) ) - if cache_hit: - aux_sp.parse_status = "aux_cached" - aux_sp.message = ( - f"aux_for={cmd_id};cache_hit;src={entry.cmd_row_id}" - )[:1020] - aux_sp.row_count = len(entry.records or []) - elif not entry.ok: - aux_sp.parse_status = "aux_failed" - aux_sp.message = f"aux_for={cmd_id};{entry.error}"[:1020] - try: - aux_sp.raw_rel_path = write_raw_text( - batch_id, aux_id, entry.raw or "" - ) - except Exception: - pass - else: - aux_sp.parse_status = "aux" - aux_sp.message = f"aux_for={cmd_id}"[:1020] - try: - aux_sp.raw_rel_path = write_raw_text( - batch_id, aux_id, entry.raw or "" - ) - except Exception: - pass - aux_sp.row_count = len(entry.records or []) - entry.cmd_row_id = aux_id - persist_recs: list[dict[str, Any]] | None = None - if entry.ok and entry.records and aux_mid in _GENERIC_METRICS: - persist_key = (normalize_command(ra.command), aux_mid) - do_persist = False - if cache_lock is not None: - with cache_lock: - if persist_key not in persisted: - persisted.add(persist_key) - do_persist = True - elif persist_key not in persisted: - persisted.add(persist_key) - do_persist = True - if do_persist: - aux_sp.persist_kind = "metric" - persist_recs = list(entry.records) - _queue(aux_sp, records=persist_recs) - - bundle = build_parse_bundle( - primary_raw=raw_text, - primary_parser_id=hit.profile.parser_id, - aux_results=aux_results, - resolved_aux=resolved_aux, - ) - primary = SpooledCommand( - id=cmd_id, + job = PrimaryParseJob( batch_id=batch_id, + cmd_id=cmd_id, task_item_id=item_id, profile_id=hit.profile.profile_id, parser_id=hit.profile.parser_id, metric_id=hit.profile.metric_id, - raw_command=concrete[:512], - params_json=merged, + concrete=concrete, + merged_params=dict(merged or {}), + raw_text=raw_text, raw_rel_path=raw_rel, + textfsm_command=hit.profile.textfsm_command or concrete, + vendor=vendor_eff, + device_type=device_type_eff, + enrich_joins=list(hit.profile.enrich_joins or []), + aux_captures=aux_captures, + persisted=persisted, + cache_lock=cache_lock, + on_done=_on_parse_done if parse_pool is not None else None, ) - 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=raw_text, - fsm_tables=fsm_tables, - records=records, - ok=True, - cmd_row_id=cmd_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: - primary.message = ";".join(hints)[:1020] - primary.parse_status = "ok" - any_ok = True - persist_recs = None - if hit.profile.metric_id == "lldp_neighbor": - primary.persist_kind = "lldp" - persist_recs = list(records or []) - elif hit.profile.metric_id in _GENERIC_METRICS: - primary.persist_kind = "metric" - persist_recs = list(records or []) - _queue(primary, records=persist_recs) - except Exception as exc: - any_fail = True - primary.parse_status = "failed" - primary.message = f"parse: {_format_error(exc)}" - _queue(primary) + if parse_pool is not None: + with parse_stats_lock: + parse_stats["pending"] += 1 + parse_pool.submit(job) + # Count primary (+ aux will be counted in parse worker via persist). + # cmd_count: bump for primary + each aux capture so progress is visible. + cmd_count += 1 + len(aux_captures) + else: + ok, fail = _run_primary_parse_job(job) + if ok: + any_ok = True + if fail: + any_fail = True + # Sync path: parse job already submitted persist; count cmds. + cmd_count += 1 + len(aux_captures) - # Final flush for this lane → persist pool; wait so rows land before return. + # Drain CLI spool leftovers, then wait parse + persist pools. _submit_pending() + if parse_pool is not None: + if not parse_pool.wait_idle(timeout=max(30.0, float(budget))): + _log.warning( + "biz_state parse barrier timed out batch=%s lane=%s", + batch_id, + label, + ) + any_fail = True + with parse_stats_lock: + if parse_stats["ok"]: + any_ok = True + if parse_stats["fail"]: + any_fail = True if not persist.wait_idle(timeout=max(30.0, float(budget))): _log.warning( "biz_state persist barrier timed out batch=%s lane=%s", @@ -933,6 +1108,8 @@ def _run_collect_lane( # Best-effort: enqueue leftover spool before connection teardown. try: _submit_pending() + if parse_pool is not None: + parse_pool.wait_idle(timeout=60.0) persist.wait_idle(timeout=60.0) except Exception: _log.exception( @@ -1115,13 +1292,14 @@ def _finalize_batch_status( batch.message = "" status = str(batch.status or "") db.commit() - if status in ("success", "partial") and not stopped: + # Only full success triggers auto compare; never block the collect thread. + if status == "success" and not stopped: try: - from .compare_service import try_auto_compare_for_task + from .compare_service import schedule_auto_compare_for_task - try_auto_compare_for_task(db, task_id, batch_id) + schedule_auto_compare_for_task(task_id, batch_id) except Exception: - _log.exception("biz_state auto compare hook failed task=%s", task_id) + _log.exception("biz_state auto compare schedule failed task=%s", task_id) return status try: @@ -1405,10 +1583,20 @@ def _run_collect_session( raise RuntimeError("; ".join(lane_errors)[:1020]) any_fail = True - # Ensure persist pool drained before terminal status write. + # Ensure parse + persist pools drained before terminal status write. try: + from .parse_pool import get_parse_pool, parse_async_enabled from .persist_pool import get_persist_pool + if parse_async_enabled(): + if not get_parse_pool().wait_idle(timeout=120.0): + _log.warning( + "biz_state parse barrier before finalize timed out batch=%s", + batch_id, + ) + any_fail = True + if "parse_barrier_timeout" not in lane_errors: + lane_errors.append("RuntimeError: parse_barrier_timeout") if not get_persist_pool().wait_idle(timeout=120.0): _log.warning( "biz_state persist barrier before finalize timed out batch=%s", @@ -1418,7 +1606,7 @@ def _run_collect_session( if "persist_barrier_timeout" not in lane_errors: lane_errors.append("RuntimeError: persist_barrier_timeout") except Exception: - _log.exception("biz_state persist barrier before finalize failed batch=%s", batch_id) + _log.exception("biz_state parse/persist barrier before finalize failed batch=%s", batch_id) stopped = is_stop_requested(batch_id) or any("_stopped" in e for e in lane_errors) _finalize_batch_status( diff --git a/netx_api/biz_state/collect_session.py b/netx_api/biz_state/collect_session.py index 5d7da43..17d0b3a 100644 --- a/netx_api/biz_state/collect_session.py +++ b/netx_api/biz_state/collect_session.py @@ -174,6 +174,40 @@ class CollectSession: return hit return None + def fetch_raw( + self, + command: str, + *, + cmd_row_id: str = "", + ) -> tuple[CachedCommand, bool]: + """CLI-only fetch (no TextFSM). Return ``(entry, cache_hit)``. + + Used so the SSH thread can overlap subsequent commands while a parse + pool parses prior raws. + """ + with self._command_lock(command): + cached = self.get_cached(command) + if cached is not None and str(cached.raw or "").strip(): + return cached, True + try: + raw = self._send_show(command) + except Exception as exc: + entry = self.remember( + command, + raw="", + ok=False, + error=f"{type(exc).__name__}: {exc}", + cmd_row_id=cmd_row_id, + ) + return entry, False + entry = self.remember( + command, + raw=raw, + ok=True, + cmd_row_id=cmd_row_id, + ) + return entry, False + def fetch_and_parse( self, command: str, @@ -187,11 +221,46 @@ class CollectSession: Same concrete CLI is serialized across shared-cache lanes so aux of one monitor item can be reused by the next without re-collecting. + Prefer ``fetch_raw`` on the collect hot path when parse can be async. """ with self._command_lock(command): cached = self.get_cached(command) if cached is not None: - return cached, True + # Full parse hit (records or explicit prior parse failure with raw). + if cached.records or (cached.ok and not parser_id): + return cached, True + if cached.ok and str(cached.raw or "").strip() and parser_id and get_parser(parser_id): + # Raw-only cache (async collect): parse now without re-CLI. + try: + records, fsm_tables, _keys = run_parser( + parser_id, + raw_text=cached.raw, + vendor=self.vendor, + device_type=self.device_type, + command=textfsm_command or command, + textfsm_command=textfsm_command or "", + params=params or {}, + ) + except Exception as exc: + entry = self.remember( + command, + raw=cached.raw, + ok=False, + error=f"parse: {type(exc).__name__}: {exc}", + cmd_row_id=cmd_row_id or cached.cmd_row_id, + ) + return entry, True + entry = self.remember( + command, + raw=cached.raw, + fsm_tables=fsm_tables, + records=records, + ok=True, + cmd_row_id=cmd_row_id or cached.cmd_row_id, + ) + return entry, True + if cached.ok: + return cached, True try: raw = self._send_show(command) except Exception as exc: diff --git a/netx_api/biz_state/compare_engine.py b/netx_api/biz_state/compare_engine.py index 3a8a15d..b11cae8 100644 --- a/netx_api/biz_state/compare_engine.py +++ b/netx_api/biz_state/compare_engine.py @@ -42,6 +42,9 @@ def mapping_stats( ``hit_before``: map key appears as a normalized before iface, or as the parent of a before subinterface (so main-port-only maps still validate). + + Map keys are matched against **normalized** iface values (same pipeline as + compare). Enter mapping keys in normalized form. """ before_ifaces: set[str] = set() after_ifaces: set[str] = set() @@ -58,14 +61,17 @@ def mapping_stats( before_bases: set[str] = set() for v in before_ifaces: before_bases.add(v) - if "." in v: - before_bases.add(v.rsplit(".", 1)[0]) + # Multi-level parents: a.b.c → a.b, a + parts = v.split(".") + for i in range(len(parts) - 1, 0, -1): + before_bases.add(".".join(parts[:i])) after_bases: set[str] = set() for v in after_ifaces: after_bases.add(v) - if "." in v: - after_bases.add(v.rsplit(".", 1)[0]) + parts = v.split(".") + for i in range(len(parts) - 1, 0, -1): + after_bases.add(".".join(parts[:i])) mapped_before = set(port_map.keys()) mapped_after = set(port_map.values()) @@ -84,6 +90,7 @@ def mapping_stats( "miss_after": miss_after, "unused_before_keys": unused, "ok": not miss_before and not miss_after, + "hint": "map keys must match normalized iface names (post iface_normalize)", } @@ -97,17 +104,22 @@ def compare_rows( port_map: dict[str, str] | None = None, field_rules: Sequence[Mapping[str, Any]] | None = None, iface_normalize_rules: Sequence[Mapping[str, str]] | None = None, + ignore_port_changes: bool | None = None, ) -> dict[str, Any]: """Return summary + diffs list. - Diff kinds: added | removed | changed | unchanged + Diff kinds: added | removed | changed | unchanged | duplicate Pipeline: iface normalize (both sides) → port map (before) → match. - Empty ``port_map``: try to ignore port renames by dropping ``iface_fields`` - from the match key (LLDP-style). If that would collapse distinct rows - (OSPF/VRRP/config where the same id appears on many interfaces), keep - iface columns so identity compare stays correct. + ``ignore_port_changes``: + - ``None`` (default): auto — drop iface from match key only when remaining + keys stay unique on both sides (legacy LLDP-style heuristic). + - ``True``: force drop iface from match key when a non-empty candidate exists. + - ``False``: never drop iface from match key. + + Duplicate match keys are not silently discarded: extras become ``duplicate`` + diffs and ``summary.duplicate_key_list`` lists the colliding keys. ``field_rules`` drives normalize / numeric tolerance / per-field compare mode (template-driven; no metric-specific branches here). @@ -127,20 +139,25 @@ def compare_rows( after_rows, iface_fields=iface_list, rules=norm_rules ) - ignore_port_changes = False + ignore_ports = False # No map → optionally ignore port renames by dropping iface from match key. if not pmap and iface_set: candidate = [k for k in key_fields if k not in iface_set] if not candidate: match_keys = list(key_fields) + elif ignore_port_changes is True: + match_keys = candidate + ignore_ports = True + elif ignore_port_changes is False: + match_keys = list(key_fields) else: + # Auto heuristic (legacy default) before_c = [row_key(r, candidate) for r in before_norm] after_c = [row_key(r, candidate) for r in after_norm] if len(before_c) == len(set(before_c)) and len(after_c) == len(set(after_c)): match_keys = candidate - ignore_port_changes = True + ignore_ports = True else: - # Remaining keys are not unique — iface is required for identity. match_keys = list(key_fields) else: match_keys = list(key_fields) @@ -151,16 +168,22 @@ def compare_rows( after_index: dict[tuple[str, ...], dict[str, Any]] = {} after_dup = 0 + after_dup_keys: list[tuple[str, ...]] = [] + after_dup_rows: list[tuple[tuple[str, ...], dict[str, Any]]] = [] for r in after_norm: k = row_key(r, match_keys) if k in after_index: after_dup += 1 + after_dup_keys.append(k) + after_dup_rows.append((k, r)) + continue # first wins — do not overwrite after_index[k] = r before_keys: set[tuple[str, ...]] = set() before_dup = 0 + before_dup_keys: list[tuple[str, ...]] = [] diffs: list[dict[str, Any]] = [] - added = removed = changed = unchanged = 0 + added = removed = changed = unchanged = duplicate = 0 def _key_obj(row: dict[str, Any]) -> dict[str, Any]: return {f: row.get(f, "") for f in key_fields} @@ -169,6 +192,20 @@ def compare_rows( k = row_key(mapped, match_keys) if k in before_keys: before_dup += 1 + before_dup_keys.append(k) + duplicate += 1 + diffs.append( + { + "kind": "duplicate", + "side": "before", + "key": _key_obj(mapped), + "before": orig, + "after": after_index.get(k), + "mapped_before": mapped, + "changes": {}, + } + ) + continue # only first before row participates in match before_keys.add(k) after = after_index.get(k) if after is None: @@ -237,6 +274,30 @@ def compare_rows( } ) + for k, after in after_dup_rows: + duplicate += 1 + diffs.append( + { + "kind": "duplicate", + "side": "after", + "key": _key_obj(after), + "before": None, + "after": after, + "mapped_before": None, + "changes": {}, + } + ) + + def _fmt_keys(keys: list[tuple[str, ...]]) -> list[str]: + seen: set[str] = set() + out: list[str] = [] + for k in keys: + s = "|".join(k) + if s not in seen: + seen.add(s) + out.append(s) + return out + stats = mapping_stats( before_rows=before_norm, after_rows=after_norm, @@ -244,7 +305,7 @@ def compare_rows( port_map=pmap, ) if not pmap: - stats = {**stats, "ok": True, "ignore_port_changes": ignore_port_changes} + stats = {**stats, "ok": True, "ignore_port_changes": ignore_ports} return { "summary": { "before_count": len(before_rows), @@ -253,9 +314,11 @@ def compare_rows( "removed": removed, "changed": changed, "unchanged": unchanged, + "duplicate": duplicate, "match_key_fields": match_keys, "duplicate_keys_before": before_dup, "duplicate_keys_after": after_dup, + "duplicate_key_list": _fmt_keys(before_dup_keys + after_dup_keys), }, "diffs": diffs, "mapping_stats": stats, diff --git a/netx_api/biz_state/compare_service.py b/netx_api/biz_state/compare_service.py index 26636a5..195e752 100644 --- a/netx_api/biz_state/compare_service.py +++ b/netx_api/biz_state/compare_service.py @@ -4,6 +4,8 @@ from __future__ import annotations import io import json +import logging +import threading import zipfile from datetime import datetime from typing import Any @@ -20,6 +22,7 @@ from ..models import ( BizPortMapping, BizPortMappingRow, BizStateBatch, + BizStateBatchCommand, BizStateLldpNeighbor, BizStateTask, ) @@ -39,11 +42,69 @@ from .iface_normalize import ( ) from .profiles import metric_field_map +_log = logging.getLogger("netx.biz_state.compare") + +# Per-job mutex so auto + manual run_compare do not dual-write the same job. +_auto_compare_locks: dict[str, threading.Lock] = {} +_auto_compare_locks_guard = threading.Lock() + + +def _job_compare_lock(job_id: str) -> threading.Lock: + jid = str(job_id or "").strip() + with _auto_compare_locks_guard: + lock = _auto_compare_locks.get(jid) + if lock is None: + lock = threading.Lock() + _auto_compare_locks[jid] = lock + return lock + def _utcnow() -> datetime: return utcnow_naive() +# parse_status values that mean the command produced usable (possibly empty) rows +_METRIC_OK_STATUSES = frozenset( + {"ok", "unmatched", "skipped_custom", "ok_aux", "unmatched_aux"} +) +_METRIC_FAIL_STATUSES = frozenset({"failed", "failed_aux"}) + + +def batch_metric_collect_ok(db: Session, batch_id: str, metric_id: str) -> bool: + """True when this metric is safe to use from the batch. + + - Batch ``success``: OK (all commands finished). + - Batch ``partial``/other: require at least one non-failed command for the metric. + - No command row for the metric on a non-success batch → incomplete. + """ + bid = str(batch_id or "").strip() + mid = str(metric_id or "").strip() + if not bid or not mid: + return False + batch = db.get(BizStateBatch, bid) + if not batch: + return False + status = str(batch.status or "") + if status == "success": + return True + cmds = ( + db.query(BizStateBatchCommand) + .filter( + BizStateBatchCommand.batch_id == bid, + BizStateBatchCommand.metric_id == mid, + ) + .all() + ) + if not cmds: + return False + statuses = [str(c.parse_status or "").strip().lower() for c in cmds] + if any(s in _METRIC_OK_STATUSES or s.startswith("ok") for s in statuses): + return True + if all(s in _METRIC_FAIL_STATUSES or s.startswith("failed") for s in statuses): + return False + # Unknown status with rows still present — allow; empty unknown on partial — deny + return any(int(c.row_count or 0) > 0 for c in cmds) + def _compare_side( db: Session, batch_id: str, @@ -314,6 +375,7 @@ def _sheet_def( display_fields: list[str] | None = None, row_filters: list[dict[str, Any]] | None = None, field_rules: list[dict[str, Any]] | None = None, + ignore_port_changes: bool | None = None, ) -> dict[str, Any]: mid = str(metric_id or "").strip() sid = str(sheet_id or "").strip() or mid @@ -352,6 +414,8 @@ def _sheet_def( "row_filters": _normalize_row_filters(row_filters), "field_rules": rules, } + if ignore_port_changes is not None: + sheet["ignore_port_changes"] = bool(ignore_port_changes) return sheet @@ -553,6 +617,9 @@ def _normalize_sheet(raw: Any) -> dict[str, Any] | None: disp_arg = _str_list(raw.get("display_fields")) else: disp_arg = None + ignore_ports: bool | None = None + if "ignore_port_changes" in raw and raw.get("ignore_port_changes") is not None: + ignore_ports = bool(raw.get("ignore_port_changes")) return _sheet_def( metric_id=mid, sheet_id=str(raw.get("sheet_id") or "").strip() or mid, @@ -563,6 +630,7 @@ def _normalize_sheet(raw: Any) -> dict[str, Any] | None: display_fields=disp_arg, row_filters=_normalize_row_filters(raw.get("row_filters")), field_rules=rules, + ignore_port_changes=ignore_ports, ) @@ -1368,7 +1436,7 @@ def _resolve_after_batch(db: Session, job: BizCompareJob) -> str: db.query(BizStateBatch) .filter( BizStateBatch.task_id == task_id, - BizStateBatch.status.in_(("success", "partial")), + BizStateBatch.status == "success", ) .order_by(BizStateBatch.started_at.desc()) .first() @@ -1401,6 +1469,9 @@ def _run_sheet( ) row_filters = list(sheet.get("row_filters") or []) mode = "presence" if not compare_fields else "fields" + ignore_ports = sheet.get("ignore_port_changes") + if ignore_ports is not None: + ignore_ports = bool(ignore_ports) before_raw = _load_metric_rows(db, batch_id=before_batch_id, metric_id=sheet["metric_id"]) after_raw = _load_metric_rows(db, batch_id=after_batch_id, metric_id=sheet["metric_id"]) before_rows = apply_row_filters(before_raw, row_filters) @@ -1414,6 +1485,7 @@ def _run_sheet( port_map=port_map, field_rules=field_rules, iface_normalize_rules=iface_normalize_rules, + ignore_port_changes=ignore_ports, ) summary = dict(result["summary"]) summary["before_raw_count"] = len(before_raw) @@ -1429,6 +1501,7 @@ def _run_sheet( "display_fields": display_fields, "row_filters": row_filters, "field_rules": field_rules, + "ignore_port_changes": ignore_ports, "mode": mode, "summary": summary, "diffs": result["diffs"], @@ -1437,6 +1510,18 @@ def _run_sheet( def run_compare(db: Session, job_id: str, *, force_after_batch_id: str = "") -> dict[str, Any]: + lock = _job_compare_lock(job_id) + if not lock.acquire(blocking=False): + raise HTTPException(status_code=409, detail="compare_already_running") + try: + return _run_compare_unlocked(db, job_id, force_after_batch_id=force_after_batch_id) + finally: + lock.release() + + +def _run_compare_unlocked( + db: Session, job_id: str, *, force_after_batch_id: str = "" +) -> dict[str, Any]: j = db.get(BizCompareJob, job_id) if not j: raise HTTPException(status_code=404, detail="job_not_found") @@ -1467,6 +1552,7 @@ def run_compare(db: Session, job_id: str, *, force_after_batch_id: str = "") -> "removed": 0, "changed": 0, "unchanged": 0, + "duplicate": 0, } mapping_by_metric: dict[str, Any] = {} for sheet in sheets_cfg: @@ -1513,6 +1599,7 @@ def run_compare(db: Session, job_id: str, *, force_after_batch_id: str = "") -> "compare_fields": s["compare_fields"], "display_fields": s.get("display_fields") or [], "field_rules": s.get("field_rules") or [], + "ignore_port_changes": s.get("ignore_port_changes"), "mode": s["mode"], "summary": s["summary"], } @@ -1957,7 +2044,14 @@ def list_runs(db: Session, job_id: str, *, limit: int = 20) -> list[dict[str, An def try_auto_compare_for_task(db: Session, task_id: str, batch_id: str) -> int: - """When a new after batch lands, run auto jobs pinned to that after task.""" + """When a new *success* after batch lands, run auto jobs pinned to that task. + + Skips jobs already being compared (non-blocking lock). Call from a background + thread so the collect finalize path is not blocked. + """ + batch = db.get(BizStateBatch, batch_id) + if not batch or str(batch.status or "") != "success": + return 0 jobs = ( db.query(BizCompareJob) .filter(BizCompareJob.mode == "auto", BizCompareJob.after_task_id == task_id) @@ -1970,6 +2064,34 @@ def try_auto_compare_for_task(db: Session, task_id: str, batch_id: str) -> int: try: run_compare(db, j.id, force_after_batch_id=batch_id) n += 1 + except HTTPException as exc: + if int(getattr(exc, "status_code", 0) or 0) == 409: + _log.info("auto compare skipped (busy) job=%s", j.id) + continue except Exception: + _log.exception("auto compare failed job=%s task=%s", j.id, task_id) continue return n + + +def schedule_auto_compare_for_task(task_id: str, batch_id: str) -> None: + """Fire-and-forget auto compare on a daemon thread (own DB session).""" + + def _run() -> None: + from ..db import SessionLocal + + db = SessionLocal() + try: + try_auto_compare_for_task(db, task_id, batch_id) + except Exception: + _log.exception( + "bg auto compare failed task=%s batch=%s", task_id, batch_id + ) + finally: + db.close() + + threading.Thread( + target=_run, + name=f"biz-auto-cmp-{str(batch_id)[:8]}", + daemon=True, + ).start() diff --git a/netx_api/biz_state/iface_normalize.py b/netx_api/biz_state/iface_normalize.py index 199a9a1..5fdb793 100644 --- a/netx_api/biz_state/iface_normalize.py +++ b/netx_api/biz_state/iface_normalize.py @@ -108,7 +108,11 @@ def apply_iface_normalize_rows( def resolve_mapped_iface(name: str, port_map: Mapping[str, str] | None) -> str: - """Exact map hit, else parent + ``.suffix`` if parent mapped, else unchanged.""" + """Exact map hit, else longest mapped parent + remaining suffix (QinQ-safe). + + ``gei-0/1/0/1.100.200`` tries exact, then ``gei-0/1/0/1.100``, then + ``gei-0/1/0/1`` — not only the last ``.`` segment. + """ text = str(name or "").strip() if not text or not port_map: return text @@ -116,11 +120,14 @@ def resolve_mapped_iface(name: str, port_map: Mapping[str, str] | None) -> str: return str(port_map[text]) if "." not in text: return text - parent, suffix = text.rsplit(".", 1) - if not parent or not suffix: - return text - if parent in port_map: - return f"{port_map[parent]}.{suffix}" + parts = text.split(".") + for i in range(len(parts) - 1, 0, -1): + parent = ".".join(parts[:i]) + suffix = ".".join(parts[i:]) + if not parent or not suffix: + continue + if parent in port_map: + return f"{port_map[parent]}.{suffix}" return text diff --git a/netx_api/biz_state/parse_pool.py b/netx_api/biz_state/parse_pool.py new file mode 100644 index 0000000..8bd06ed --- /dev/null +++ b/netx_api/biz_state/parse_pool.py @@ -0,0 +1,181 @@ +"""Parse pool: TextFSM / normalize off the collect (SSH) threads.""" + +from __future__ import annotations + +import logging +import queue +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Callable + +from ..config import settings + +_log = logging.getLogger("netx.biz_state.parse") + +_SENTINEL = object() + + +@dataclass +class AuxRawCapture: + """CLI-only aux capture passed to the parse worker.""" + + key: str + aux_id: str + profile_id: str + parser_id: str + metric_id: str + command: str + textfsm_command: str + rule_keys: tuple[str, ...] = () + raw: str = "" + raw_rel_path: str = "" + cache_hit: bool = False + # Pre-parsed records when cache already had a full parse hit. + records: list[dict[str, Any]] = field(default_factory=list) + fsm_tables: dict[str, list[dict[str, Any]]] = field(default_factory=dict) + ok: bool = True + error: str = "" + + +@dataclass +class PrimaryParseJob: + """One primary command (with aux raws) ready for CPU parse + spool.""" + + batch_id: str + cmd_id: str + task_item_id: str + profile_id: str + parser_id: str + metric_id: str + concrete: str + merged_params: dict[str, Any] + raw_text: str + raw_rel_path: str + textfsm_command: str + vendor: str + device_type: str + enrich_joins: list[Any] = field(default_factory=list) + aux_captures: list[AuxRawCapture] = field(default_factory=list) + # Shared across lanes for aux metric de-dupe (may be None). + persisted: set[tuple[str, str]] | None = None + cache_lock: Any | None = None + on_done: Callable[[bool, bool], None] | None = None + + +class ParsePool: + """Background workers that parse + enqueue persist for primary jobs.""" + + def __init__(self, *, workers: int | None = None) -> None: + n = max( + 1, + int( + workers + if workers is not None + else (getattr(settings, "biz_state_parse_workers", 8) or 8) + ), + ) + self._q: queue.Queue[Any] = queue.Queue() + self._inflight = 0 + self._lock = threading.Lock() + self._cv = threading.Condition(self._lock) + self._workers: list[threading.Thread] = [] + self._stopped = False + for i in range(n): + t = threading.Thread( + target=self._loop, + name=f"biz-parse-{i}", + daemon=True, + ) + t.start() + self._workers.append(t) + _log.info("biz_state parse pool started workers=%s", n) + + def submit(self, job: PrimaryParseJob) -> None: + if self._stopped or job is None: + return + with self._cv: + self._inflight += 1 + self._q.put(job) + + def wait_idle(self, *, timeout: float | None = None) -> bool: + end = None + if timeout is not None: + end = time.monotonic() + max(0.0, float(timeout)) + with self._cv: + while self._inflight > 0 or not self._q.empty(): + remaining = None + if end is not None: + remaining = end - time.monotonic() + if remaining <= 0: + return False + self._cv.wait(timeout=remaining) + return True + + def shutdown(self, *, wait: bool = True) -> None: + self._stopped = True + for _ in self._workers: + self._q.put(_SENTINEL) + if wait: + for t in self._workers: + t.join(timeout=5.0) + + def _loop(self) -> None: + from .collect_runner import _run_primary_parse_job + + while True: + job = self._q.get() + if job is _SENTINEL: + self._q.task_done() + break + ok = False + fail = False + try: + ok, fail = _run_primary_parse_job(job) + except Exception: + fail = True + _log.exception( + "biz_state parse job failed batch=%s cmd=%s", + getattr(job, "batch_id", ""), + getattr(job, "cmd_id", ""), + ) + try: + if job.on_done: + job.on_done(False, True) + except Exception: + pass + else: + try: + if job.on_done: + job.on_done(ok, fail) + except Exception: + _log.exception("biz_state parse on_done failed") + finally: + with self._cv: + self._inflight = max(0, self._inflight - 1) + self._cv.notify_all() + self._q.task_done() + + +_pool_lock = threading.Lock() +_pool: ParsePool | None = None + + +def parse_async_enabled() -> bool: + return bool(getattr(settings, "biz_state_parse_async", True)) + + +def get_parse_pool() -> ParsePool: + global _pool + with _pool_lock: + if _pool is None: + _pool = ParsePool() + return _pool + + +def shutdown_parse_pool(*, wait: bool = False) -> None: + global _pool + with _pool_lock: + if _pool is not None: + _pool.shutdown(wait=wait) + _pool = None diff --git a/netx_api/biz_state/parsers/zte/bgp_route.py b/netx_api/biz_state/parsers/zte/bgp_route.py index 175949f..f47026c 100644 --- a/netx_api/biz_state/parsers/zte/bgp_route.py +++ b/netx_api/biz_state/parsers/zte/bgp_route.py @@ -13,21 +13,22 @@ from .bgp_peer import _detect_bgp_afi, _detect_local_as, _detect_vrf RULE_KEYS = ("zte_zxros_show_bgp_neighbor_routes",) -# Single-line IPv4-style: * 10.1.0.0/24 10.0.0.1 … path +# Status codes may be glued or spaced: "*", "* i", "*>i", "* s", "< *" +_FLAGS = r"(?P(?:[*<>isd]+(?:\s+[*<>isd]+)*)?)" +# Single-line IPv4-style: * i 10.1.0.0/24 10.0.0.1 … path _ROUTE_ONE_LINE_RE = re.compile( - r"^\s*(?P[*<>isd]*)\s*" + rf"^\s*{_FLAGS}\s+" r"(?P\S+)\s+" r"(?P\S+)\s+" r"(?P.*)$" ) -# Network alone (often IPv6 wrap): * 2407::1/128 or bare prefix for "out" -_NET_ONLY_RE = re.compile( - r"^\s*(?P[*<>isd]*)\s*(?P\S+)\s*$" -) +# Network alone (often IPv6 wrap): * i 2407::1/128 +_NET_ONLY_RE = re.compile(rf"^\s*{_FLAGS}\s*(?P\S+)\s*$") _DIR_RE = re.compile(r"(?i)\bneighbor\s+(in|out)\s+") # Neighbor may be IPv4 or IPv6 (consume until EOL / pipe) _NEI_RE = re.compile(r"(?i)\bneighbor\s+(?:in|out)\s+(\S+)") _TOTAL_RE = re.compile(r"(?i)total\s+number\s+of\s+routes\s*:\s*(\d+)") +_RD_RE = re.compile(r"(?i)^Route\s+Distinguisher\s*:\s*(\S+)") _HEADER_NETS = frozenset( { "network", @@ -80,6 +81,21 @@ def _looks_like_prefix(net: str) -> bool: return _looks_like_ip_or_prefix(net) +def _declared_total(raw_text: str) -> int | None: + m = _TOTAL_RE.search(str(raw_text or "")) + if not m: + return None + try: + return int(m.group(1)) + except ValueError: + return None + + +def _route_dedupe_key(*, rd: str, net: str, nh: str) -> str: + """ECMP / multi-RD: same prefix with different RD or next-hop is distinct.""" + return f"{rd}\0{net}\0{nh}" + + def _split_rest(rest: str, *, path_continuation: bool = False) -> tuple[str, str, str, str]: """Parse trailing Metric LocPrf Tag/RtPrf Path columns (some may be blank). @@ -118,8 +134,8 @@ def _split_rest(rest: str, *, path_continuation: bool = False) -> tuple[str, str def _empty_if_total_zero(raw_text: str) -> bool: """True when device reports Total number of routes: 0.""" - m = _TOTAL_RE.search(str(raw_text or "")) - return bool(m and int(m.group(1)) == 0) + total = _declared_total(raw_text) + return total == 0 def _skip_noise_line(line: str) -> bool: @@ -134,9 +150,7 @@ def _skip_noise_line(line: str) -> bool: return True if low.startswith("local ") or low.startswith("remote ") or low.startswith("total "): return True - if low.startswith("route distinguisher") or low.startswith("valid ") or low.startswith( - "invalid " - ): + if low.startswith("valid ") or low.startswith("invalid "): return True # Banner / clock lines (e.g. "09:50:02 Indonesia Sat Sep 19 2026") if re.match(r"^\d{1,2}:\d{2}:\d{2}\b", low): @@ -157,19 +171,24 @@ def _emit_route( vrf: str, neighbor: str, direction: str, + rd: str, net: str, nh: str, rest: str, flags: str, path_continuation: bool = False, ) -> None: - if not _looks_like_prefix(net) or net in seen: + if not _looks_like_prefix(net): return if nh and not _looks_like_ip_or_prefix(nh): # Path/metric-only continuation without a real next-hop — keep empty nh if re.search(r"[A-Za-z]", nh): return - seen.add(net) + nh = "" + key = _route_dedupe_key(rd=rd, net=net, nh=nh or "") + if key in seen: + return + seen.add(key) metric, loc, tag, path = _split_rest(rest, path_continuation=path_continuation) out.append( { @@ -178,13 +197,14 @@ def _emit_route( "vrf": vrf[:128], "neighbor": neighbor[:128], "direction": direction[:8], + "rd": (rd or "")[:64], "network": net[:128], "next_hop": (nh or "")[:128], "metric": metric[:32], "loc_prf": loc[:32], "tag": tag[:32], "path": path[:256], - "status_codes": (flags or "").strip()[:16], + "status_codes": re.sub(r"\s+", "", (flags or "").strip())[:16], "as_num": "", "state": "", "pfx_rcd": "", @@ -205,7 +225,7 @@ def _map_fsm_rows( seen: set[str] = set() for r in rows: net = row_get(r, "NETWORK", "network") - if not _looks_like_prefix(net) or net in seen: + if not _looks_like_prefix(net): continue nh = row_get(r, "NEXT_HOP", "next_hop") if str(nh or "").strip().lower() in _HEADER_NETS: @@ -213,8 +233,13 @@ def _map_fsm_rows( if nh and not _looks_like_ip_or_prefix(nh): # Reject FSM false hits like NETWORK=20 NEXT_HOP=65254 continue - seen.add(net) + rd = row_get(r, "RD", "rd") + key = _route_dedupe_key(rd=rd, net=net, nh=nh or "") + if key in seen: + continue + seen.add(key) path = row_get(r, "PATH", "path") + flags = row_get(r, "STATUS", "status_codes", "FLAGS") out.append( { "local_as": local_as[:16], @@ -222,13 +247,14 @@ def _map_fsm_rows( "vrf": vrf[:128], "neighbor": neighbor[:128], "direction": direction[:8], + "rd": (rd or "")[:64], "network": net[:128], "next_hop": nh[:128], "metric": row_get(r, "METRIC", "metric")[:32], "loc_prf": row_get(r, "LOC_PRF", "loc_prf")[:32], "tag": row_get(r, "TAG", "RT_PRF", "tag")[:32], "path": path[:256], - "status_codes": "", + "status_codes": re.sub(r"\s+", "", (flags or "").strip())[:16], "as_num": "", "state": "", "pfx_rcd": "", @@ -253,6 +279,7 @@ def _hand_parse( pending_net = "" pending_flags = "" pending_nh = "" + current_rd = "" def _flush_pending(*, rest: str = "", path_continuation: bool = False) -> None: nonlocal pending_net, pending_flags, pending_nh @@ -266,6 +293,7 @@ def _hand_parse( vrf=vrf, neighbor=neighbor, direction=direction, + rd=current_rd, net=pending_net, nh=pending_nh, rest=rest, @@ -278,6 +306,11 @@ def _hand_parse( for raw in str(raw_text or "").splitlines(): line = raw.rstrip() + rd_m = _RD_RE.match(line.strip()) + if rd_m: + _flush_pending() + current_rd = rd_m.group(1).strip() + continue if _skip_noise_line(line): continue @@ -299,7 +332,7 @@ def _hand_parse( _flush_pending(rest=tok, path_continuation=True) continue - # Full one-liner (typical IPv4) + # Full one-liner (typical IPv4 / RR "* i prefix …") m = _ROUTE_ONE_LINE_RE.match(line) if m and _looks_like_prefix(m.group("net")) and _looks_like_ip_or_prefix(m.group("nh")): _flush_pending() @@ -311,6 +344,7 @@ def _hand_parse( vrf=vrf, neighbor=neighbor, direction=direction, + rd=current_rd, net=m.group("net"), nh=m.group("nh"), rest=m.group("rest"), @@ -382,7 +416,14 @@ def normalize_bgp_route( **kw, ) - return prefer_fsm(tables, RULE_KEYS, _map, _hand, raw_text=raw_text) + rows = prefer_fsm(tables, RULE_KEYS, _map, _hand, raw_text=raw_text) + # If device declared a large table but FSM/hand returned a tiny subset, prefer hand. + declared = _declared_total(raw_text) + if declared is not None and declared > 0 and len(rows) < max(1, declared // 2): + hand_rows = _hand(raw_text=raw_text) + if len(hand_rows) > len(rows): + return hand_rows + return rows normalize_bgp_route.RULE_KEYS = RULE_KEYS diff --git a/netx_api/biz_state/profiles.py b/netx_api/biz_state/profiles.py index aa2bb96..996d46b 100644 --- a/netx_api/biz_state/profiles.py +++ b/netx_api/biz_state/profiles.py @@ -535,8 +535,9 @@ _BGP_ROUTE_FIELDS: list[FieldDef] = [ FieldDef("vrf", length=128, indexed=True, is_key=True, display_name="VRF", from_command_param=True), FieldDef("neighbor", length=128, indexed=True, is_key=True, display_name="Neighbor", from_command_param=True), FieldDef("direction", length=8, indexed=True, is_key=True, display_name="Dir", from_command_param=True), + FieldDef("rd", length=64, indexed=True, is_key=True, display_name="RD"), FieldDef("network", length=128, indexed=True, is_key=True, display_name="Network"), - FieldDef("next_hop", length=128, role="state", display_name="NextHop"), + FieldDef("next_hop", length=128, indexed=True, is_key=True, display_name="NextHop"), FieldDef("metric", length=32, role="meta", display_name="Metric"), FieldDef("loc_prf", length=32, role="meta", display_name="LocPrf"), FieldDef("tag", length=32, role="meta", display_name="Tag"), diff --git a/netx_api/biz_state/schema_ensure.py b/netx_api/biz_state/schema_ensure.py index 1436b46..990e6e2 100644 --- a/netx_api/biz_state/schema_ensure.py +++ b/netx_api/biz_state/schema_ensure.py @@ -79,6 +79,8 @@ def apply_biz_state_schema(conn: Connection) -> None: "ALTER TABLE biz_migration_project ADD COLUMN IF NOT EXISTS new_hf_bindings_json JSON DEFAULT '[]'", "ALTER TABLE biz_state_task ADD COLUMN IF NOT EXISTS purpose VARCHAR(32) DEFAULT ''", "CREATE INDEX IF NOT EXISTS ix_biz_state_task_purpose ON biz_state_task (purpose)", + "ALTER TABLE biz_migration_red_ticket ADD COLUMN IF NOT EXISTS match_key_str VARCHAR(256) DEFAULT ''", + "CREATE INDEX IF NOT EXISTS ix_biz_migration_red_match ON biz_migration_red_ticket (project_id, metric_id, match_key_str)", ): try: _run_sql(conn, sql) diff --git a/netx_api/biz_state/service.py b/netx_api/biz_state/service.py index 817f43b..cf8fc35 100644 --- a/netx_api/biz_state/service.py +++ b/netx_api/biz_state/service.py @@ -593,6 +593,14 @@ def _metric_sheet_title(metric_id: str, fallback: str = "") -> str: return fb or mid +def _raw_line_count(raw: str | None) -> int: + """CLI text lines collected (splitlines-compatible, no list materialization).""" + s = raw or "" + if not s: + return 0 + return s.count("\n") + (0 if s.endswith("\n") else 1) + + def get_batch(db: Session, batch_id: str) -> dict[str, Any]: """Batch workbook summary: meta + commands + sheet catalog (no metric row payload).""" b = db.get(BizStateBatch, batch_id) @@ -648,6 +656,7 @@ def get_batch(db: Session, batch_id: str) -> dict[str, Any]: for c in cmds: status = str(c.parse_status or "").strip().lower() is_aux = status.startswith("aux") + raw = c.raw_text or "" info = { "id": c.id, "profile_id": c.profile_id, @@ -657,8 +666,9 @@ def get_batch(db: Session, batch_id: str) -> dict[str, Any]: "params": c.params_json or {}, "parse_status": c.parse_status, "row_count": c.row_count, + "raw_line_count": _raw_line_count(raw), "message": c.message, - "has_raw": bool(str(c.raw_text or "").strip()), + "has_raw": bool(str(raw).strip()), "is_aux": is_aux, } cmd_n = normalize_command(str(c.raw_command or "")) @@ -688,6 +698,7 @@ def get_batch(db: Session, batch_id: str) -> dict[str, Any]: "raw_command": c.raw_command, "parse_status": c.parse_status, "row_count": c.row_count, + "raw_line_count": info["raw_line_count"], "message": c.message, "has_raw": info["has_raw"], "profile_id": c.profile_id, @@ -713,6 +724,7 @@ def get_batch(db: Session, batch_id: str) -> dict[str, Any]: "params": c.params_json or {}, "parse_status": c.parse_status, "row_count": c.row_count, + "raw_line_count": _raw_line_count(c.raw_text), "message": c.message, "has_raw": bool(str(c.raw_text or "").strip()), "is_aux": True, @@ -906,6 +918,7 @@ def get_batch_command(db: Session, batch_id: str, command_id: str) -> dict[str, if not c or c.batch_id != batch_id: raise HTTPException(status_code=404, detail="command_not_found") task = db.get(BizStateTask, b.task_id) if b.task_id else None + raw = c.raw_text or "" return { "id": c.id, "batch_id": batch_id, @@ -922,8 +935,9 @@ def get_batch_command(db: Session, batch_id: str, command_id: str) -> dict[str, "params": c.params_json or {}, "parse_status": c.parse_status, "row_count": c.row_count, + "raw_line_count": _raw_line_count(raw), "message": c.message, - "raw_text": c.raw_text or "", + "raw_text": raw, "collected_at": c.created_at.isoformat() + "Z" if c.created_at else None, "batch_started_at": b.started_at.isoformat() + "Z" if b.started_at else None, "batch_ended_at": b.ended_at.isoformat() + "Z" if b.ended_at else None, diff --git a/netx_api/biz_state_router.py b/netx_api/biz_state_router.py index 14ed384..465e95d 100644 --- a/netx_api/biz_state_router.py +++ b/netx_api/biz_state_router.py @@ -337,6 +337,23 @@ def api_get_batch_command( return svc.get_batch_command(db, batch_id, command_id) +@router.get("/batches/{batch_id}/commands/{command_id}/raw.txt") +def api_download_batch_command_raw( + batch_id: str, command_id: str, db: Session = Depends(get_db) +) -> StreamingResponse: + """Download one command's raw CLI output as plain text.""" + detail = svc.get_batch_command(db, batch_id, command_id) + raw = str(detail.get("raw_text") or "") + cmd = str(detail.get("raw_command") or "command") + safe = "".join(ch if ch.isalnum() or ch in "-_." else "_" for ch in cmd)[:80] or "command" + filename = f"{command_id}_{safe}.txt" + return StreamingResponse( + iter([raw.encode("utf-8")]), + media_type="text/plain; charset=utf-8", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + @router.get("/batches/{batch_id}/export") def api_export_batch(batch_id: str, db: Session = Depends(get_db)) -> StreamingResponse: data = svc.export_batch_zip(db, batch_id) diff --git a/netx_api/biz_state_scheduler.py b/netx_api/biz_state_scheduler.py index c6b13ca..d44250f 100644 --- a/netx_api/biz_state_scheduler.py +++ b/netx_api/biz_state_scheduler.py @@ -28,7 +28,9 @@ _PURGE_INTERVAL_SEC = 3600.0 def _utcnow() -> datetime: - return datetime.utcnow() + from .timeutil import utcnow_naive + + return utcnow_naive() def _maybe_purge_retention() -> None: @@ -86,10 +88,12 @@ def shutdown_biz_state_dispatch_pool(*, wait: bool = False) -> None: _dispatch_pool = None try: from .biz_state.persist_pool import shutdown_persist_pool + from .biz_state.parse_pool import shutdown_parse_pool + shutdown_parse_pool(wait=wait) shutdown_persist_pool(wait=wait) except Exception: - _log.exception("shutdown persist pool failed") + _log.exception("shutdown parse/persist pool failed") def _sync_cutover_hf_windows() -> None: @@ -152,11 +156,15 @@ def _enqueue_due_tasks() -> int: now = _utcnow() for task in tasks: interval = max(60, int(task.interval_sec or 300)) + # Prefer start-based interval to avoid drift when collect duration + # approaches the interval (end-based: 40s collect + 60s → 100s cycle). + started = task.last_collect_started_at ended = task.last_collect_ended_at - if ended is None: + anchor = started or ended + if anchor is None: due_ids.append(str(task.id)) continue - if (now - ended).total_seconds() >= interval: + if (now - anchor).total_seconds() >= interval: due_ids.append(str(task.id)) finally: db.close() diff --git a/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm b/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm index 1bff0f0..dc2dd7c 100644 --- a/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm +++ b/netx_api/cli_templates/zte/zte_zxros_show_bgp_neighbor_routes.textfsm @@ -1,3 +1,5 @@ +Value Filldown RD (\S+) +Value STATUS ((?:\*[<>isd]*(?:\s+[isd]+)?|[<>isd]+)?) Value NETWORK (\d{1,3}(?:\.\d{1,3}){3}/\d{1,2}|[0-9A-Fa-f:]+/\d{1,3}) Value NEXT_HOP (\d{1,3}(?:\.\d{1,3}){3}|[0-9A-Fa-f:]+) Value METRIC (\S*) @@ -10,11 +12,13 @@ Start ^\s*\*?\s*Network\s+Next -> Routes ^Network\s+Next -> Routes ^\s*Dest\s+Next -> Routes + # RD may appear before header on some snippets; keep Filldown ready + ^Route\s+Distinguisher\s*:?\s*${RD} ^\s*$$ ^.* -> Start Routes - ^Route\s+Distinguisher + ^Route\s+Distinguisher\s*:?\s*${RD} ^Status\s+codes ^Origin\s+codes ^Current\s+AS @@ -26,11 +30,11 @@ Routes ^\s*\*?\s*Network\s+Next ^Network\s+Next ^\s*Dest\s+Next - # One-line IPv4-style - ^\s*\*?\s*>?\s*${NETWORK}\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record - ^\s*\*?\s*>?\s*${NETWORK}\s+${NEXT_HOP}\s+${PATH}\s*$$ -> Record + # One-line IPv4 / RR "* i prefix nh …" + ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s+${METRIC}\s+${LOC_PRF}\s+${TAG}\s+${PATH}\s*$$ -> Record + ^\s*${STATUS}\s+${NETWORK}\s+${NEXT_HOP}\s+${PATH}\s*$$ -> Record # IPv6 wrap: network / next-hop / metrics+path on separate lines - ^\s*\*?\s*>?\s*${NETWORK}\s*$$ + ^\s*${STATUS}\s+${NETWORK}\s*$$ ^\s+${NEXT_HOP}\s*$$ ^\s+${PATH}\s*$$ -> Record ^\s*$$ diff --git a/netx_api/config.py b/netx_api/config.py index 9bfd426..ff6d852 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -133,6 +133,10 @@ class Settings(BaseSettings): biz_state_max_concurrent_tasks: int = 16 biz_state_worker_collect_threads: int = 8 biz_state_persist_workers: int = 4 + # TextFSM / normalize workers (off SSH collect threads). Default 8. + biz_state_parse_workers: int = 8 + # When true (default), primary+aux parse runs on parse pool overlapping next CLI. + biz_state_parse_async: bool = True # How many biz_state_worker processes start_netx should launch (same host). biz_state_worker_replicas: int = 2 # Managed NE exec: max CLI commands per request (lab can raise; hard-capped in ne_exec). diff --git a/netx_api/models/biz_migration.py b/netx_api/models/biz_migration.py index e4ad5ff..56ab038 100644 --- a/netx_api/models/biz_migration.py +++ b/netx_api/models/biz_migration.py @@ -129,6 +129,13 @@ class BizMigrationRedTicket(Base): __table_args__ = ( Index("ix_biz_migration_red_project_status", "project_id", "status"), Index("ix_biz_migration_red_batch", "batch_id"), + Index( + "ix_biz_migration_red_upsert", + "project_id", + "metric_id", + "match_key_str", + "status", + ), ) id: Mapped[str] = mapped_column(String(64), primary_key=True, default=lambda: uuid4().hex) @@ -138,6 +145,8 @@ class BizMigrationRedTicket(Base): metric_id: Mapped[str] = mapped_column(String(64), default="", index=True) key_str: Mapped[str] = mapped_column(String(256), default="") new_key_str: Mapped[str] = mapped_column(String(256), default="") + # Stable identity for upsert (match_old_key / normalized expect key) + match_key_str: Mapped[str] = mapped_column(String(256), default="", index=True) verdict: Mapped[str] = mapped_column(String(32), default="") color: Mapped[str] = mapped_column(String(16), default="red") old_status: Mapped[str] = mapped_column(String(64), default="") diff --git a/tests/test_batch_workbook_api.py b/tests/test_batch_workbook_api.py index 4fbdb62..e156545 100644 --- a/tests/test_batch_workbook_api.py +++ b/tests/test_batch_workbook_api.py @@ -65,11 +65,115 @@ class BatchWorkbookApiTests(unittest.TestCase): self.assertNotIn("metrics", out) self.assertNotIn("lldp_neighbors", out) self.assertEqual(out["commands"][0]["has_raw"], True) + self.assertEqual(out["commands"][0]["raw_line_count"], 1) self.assertEqual(out["sheets"][0]["metric_id"], "arp") self.assertEqual(out["sheets"][0]["row_count"], 2) self.assertEqual(out["sheets"][0]["commands"][0]["raw_command"], "show arp | one-line") + self.assertEqual(out["sheets"][0]["commands"][0]["raw_line_count"], 1) self.assertTrue(out["sheets"][0].get("title")) + def test_get_batch_raw_line_count_splitlines(self) -> None: + batch = BizStateBatch( + id="b1", + task_id="t1", + status="partial", + command_count=1, + row_count=0, + message="stopped", + ) + cmd = BizStateBatchCommand( + id="c1", + batch_id="b1", + profile_id="zte.arp", + parser_id="arp", + metric_id="arp", + raw_command="show arp", + parse_status="ok", + row_count=3, + raw_text="a\nb\nc\n", + message="", + ) + db = MagicMock() + db.get.side_effect = lambda model, pk: batch if pk == "b1" else None + cmd_q = MagicMock() + cmd_q.filter.return_value.order_by.return_value.all.return_value = [cmd] + metric_count_q = MagicMock() + metric_count_q.filter.return_value.group_by.return_value.all.return_value = [("arp", 3)] + lldp_count_q = MagicMock() + lldp_count_q.filter.return_value.scalar.return_value = 0 + + def query(*_args, **_kwargs): + n = query.n + query.n += 1 + if n == 0: + return cmd_q + if n == 1: + return metric_count_q + return lldp_count_q + + query.n = 0 + db.query.side_effect = query + + with patch( + "netx_api.biz_state.service.batch_protect_info", + return_value={"protected": False, "reasons": []}, + ): + out = get_batch(db, "b1") + + self.assertEqual(out["message"], "stopped") + self.assertEqual(out["commands"][0]["raw_line_count"], 3) + self.assertEqual(out["commands"][0]["row_count"], 3) + + def test_get_batch_command_and_raw_download(self) -> None: + from netx_api.biz_state.service import get_batch_command + from netx_api.biz_state_router import api_download_batch_command_raw + + batch = BizStateBatch(id="b1", task_id="t1", status="ok") + cmd = BizStateBatchCommand( + id="c1", + batch_id="b1", + profile_id="zte.arp", + parser_id="arp", + metric_id="arp", + raw_command="show arp | one-line", + parse_status="ok", + row_count=2, + raw_text="line1\nline2", + message="hint", + ) + db = MagicMock() + + def _get(model, pk): + if model is BizStateBatch and pk == "b1": + return batch + if model is BizStateBatchCommand and pk == "c1": + return cmd + return None + + db.get.side_effect = _get + detail = get_batch_command(db, "b1", "c1") + self.assertEqual(detail["raw_line_count"], 2) + self.assertEqual(detail["row_count"], 2) + self.assertEqual(detail["message"], "hint") + self.assertIn("line1", detail["raw_text"]) + + resp = api_download_batch_command_raw("b1", "c1", db) + self.assertEqual(resp.media_type, "text/plain; charset=utf-8") + cd = (resp.headers.get("content-disposition") or "").lower() + self.assertIn("attachment", cd) + self.assertIn(".txt", cd) + # StreamingResponse may expose async iterator; content already covered by get_batch_command. + body_iter = getattr(resp, "body_iterator", None) + if body_iter is not None and hasattr(body_iter, "__iter__") and not hasattr(body_iter, "__aiter__"): + body = b"".join(body_iter) + self.assertEqual(body.decode("utf-8"), "line1\nline2") + else: + # Fallback: reconstruct what the route encodes + from netx_api.biz_state.service import get_batch_command as _gbc + + raw = str(_gbc(db, "b1", "c1").get("raw_text") or "") + self.assertEqual(raw, "line1\nline2") + def test_bgp_peer_sheet_uses_status_summary_title(self) -> None: """Shared metric_id bgp_peer must not inherit first AF profile title.""" batch = BizStateBatch( diff --git a/tests/test_biz_migration_evaluate.py b/tests/test_biz_migration_evaluate.py index d9e9364..0ac413c 100644 --- a/tests/test_biz_migration_evaluate.py +++ b/tests/test_biz_migration_evaluate.py @@ -5,6 +5,7 @@ from __future__ import annotations import unittest from netx_api.biz_migration.evaluate import ( + KEY_SEP, classify_status, dual_verdict, evaluate_metric_dual, @@ -90,7 +91,7 @@ class ParseExpectSetTests(unittest.TestCase): ] } ) - self.assertEqual(got["bgp_peer"], {"AS1|1.1.1.1"}) + self.assertEqual(got["bgp_peer"], {KEY_SEP.join(["AS1", "1.1.1.1"])}) self.assertEqual(got["arp"], {"10.0.0.1"}) def test_multi_select_each_key_is_separate(self): @@ -111,7 +112,7 @@ class ParseExpectSetTests(unittest.TestCase): got = parse_expect_set( {"items": [{"metric_id": "bgp_peer", "keys": ["1.1.1.1", "2.2.2.2"]}]} ) - self.assertEqual(got["bgp_peer"], {"1.1.1.1|2.2.2.2"}) + self.assertEqual(got["bgp_peer"], {KEY_SEP.join(["1.1.1.1", "2.2.2.2"])}) def test_nested_keys_list(self): got = parse_expect_set( @@ -122,8 +123,20 @@ class ParseExpectSetTests(unittest.TestCase): ] } ) - self.assertEqual(got["isis_adjacency"], {"p1|gei-1|sys1", "p1|gei-2|sys2"}) - self.assertEqual(got["arp"], {"10.0.0.1|vrf1"}) + self.assertEqual( + got["isis_adjacency"], + { + KEY_SEP.join(["p1", "gei-1", "sys1"]), + KEY_SEP.join(["p1", "gei-2", "sys2"]), + }, + ) + self.assertEqual(got["arp"], {KEY_SEP.join(["10.0.0.1", "vrf1"])}) + + def test_legacy_pipe_key_normalized(self): + got = parse_expect_set( + {"items": [{"metric_id": "bgp_peer", "key": "AS1|1.1.1.1"}]} + ) + self.assertEqual(got["bgp_peer"], {KEY_SEP.join(["AS1", "1.1.1.1"])}) def test_ports_only_apply_to_interface_brief(self): expect = parse_expect_set({"ports": ["gei-1"]}) @@ -855,12 +868,20 @@ class EvaluateMetricDualTests(unittest.TestCase): def test_composite_key_port_remap(self): mapped = _remap_key_str( + KEY_SEP.join(["gei-old", "vrf1"]), + key_fields=["interface", "vrf"], + iface_fields=["interface"], + port_map={"gei-old": "gei-new"}, + ) + self.assertEqual(mapped, KEY_SEP.join(["gei-new", "vrf1"])) + # Legacy pipe-separated input still remaps + mapped2 = _remap_key_str( "gei-old|vrf1", key_fields=["interface", "vrf"], iface_fields=["interface"], port_map={"gei-old": "gei-new"}, ) - self.assertEqual(mapped, "gei-new|vrf1") + self.assertEqual(mapped2, KEY_SEP.join(["gei-new", "vrf1"])) def test_anomaly_not_gated_by_field_tokens_on_removed(self): self.assertEqual( diff --git a/tests/test_biz_state_collect_finalize.py b/tests/test_biz_state_collect_finalize.py index 9204047..7de469d 100644 --- a/tests/test_biz_state_collect_finalize.py +++ b/tests/test_biz_state_collect_finalize.py @@ -57,9 +57,8 @@ class BizStateCollectFinalizeTests(unittest.TestCase): def test_finalize_partial_after_heavy_timeout(self) -> None: with patch.object(runner, "SessionLocal", self.Session): with patch( - "netx_api.biz_state.compare_service.try_auto_compare_for_task", - return_value=None, - ): + "netx_api.biz_state.compare_service.schedule_auto_compare_for_task", + ) as sched: status = runner._finalize_batch_status( batch_id="b-finalize", task_id="t-finalize", @@ -70,6 +69,7 @@ class BizStateCollectFinalizeTests(unittest.TestCase): lane_errors=["RuntimeError: biz_state_heavy_timeout (2400s)"], ) self.assertEqual(status, "partial") + sched.assert_not_called() self.db.expire_all() batch = self.db.get(BizStateBatch, "b-finalize") assert batch is not None @@ -127,8 +127,7 @@ class BizStateCollectFinalizeTests(unittest.TestCase): with patch.object(runner, "SessionLocal", session_factory): with patch( - "netx_api.biz_state.compare_service.try_auto_compare_for_task", - return_value=None, + "netx_api.biz_state.compare_service.schedule_auto_compare_for_task", ): status = runner._finalize_batch_status( batch_id="b-finalize", @@ -147,6 +146,23 @@ class BizStateCollectFinalizeTests(unittest.TestCase): self.assertEqual(batch.status, "partial") self.assertIn("biz_state_heavy_timeout", batch.message or "") + def test_finalize_success_schedules_auto_compare(self) -> None: + with patch.object(runner, "SessionLocal", self.Session): + with patch( + "netx_api.biz_state.compare_service.schedule_auto_compare_for_task", + ) as sched: + status = runner._finalize_batch_status( + batch_id="b-finalize", + task_id="t-finalize", + cmd_count=55, + total_rows=6606, + any_fail=False, + any_ok=True, + lane_errors=[], + ) + self.assertEqual(status, "success") + sched.assert_called_once_with("t-finalize", "b-finalize") + def test_fail_batch_retries_on_stale_connection(self) -> None: calls = {"n": 0} real_session = self.Session diff --git a/tests/test_biz_state_compare.py b/tests/test_biz_state_compare.py index 43b2ecc..8d610ad 100644 --- a/tests/test_biz_state_compare.py +++ b/tests/test_biz_state_compare.py @@ -391,6 +391,33 @@ class CompareSheetDefaultsTests(unittest.TestCase): ) self.assertEqual(out["summary"]["duplicate_keys_before"], 1) self.assertEqual(out["summary"]["duplicate_keys_after"], 1) + self.assertEqual(out["summary"]["duplicate"], 2) + self.assertIn("a|X|1", out["summary"]["duplicate_key_list"]) + kinds = [d["kind"] for d in out["diffs"]] + self.assertEqual(kinds.count("duplicate"), 2) + # First before wins → matches first after → unchanged (same remote_ip) + self.assertEqual(out["summary"]["unchanged"], 1) + self.assertEqual(out["summary"]["changed"], 0) + + def test_ignore_port_changes_false_keeps_iface(self) -> None: + before = [ + {"local_if": "old-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"}, + ] + after = [ + {"local_if": "new-1", "remote_sys": "Peer", "remote_if": "p1", "remote_ip": "1.1.1.1"}, + ] + out = compare_rows( + before_rows=before, + after_rows=after, + key_fields=["local_if", "remote_sys", "remote_if"], + iface_fields=["local_if"], + compare_fields=["remote_ip"], + port_map={}, + ignore_port_changes=False, + ) + self.assertEqual(out["summary"]["removed"], 1) + self.assertEqual(out["summary"]["added"], 1) + self.assertFalse(out["mapping_stats"].get("ignore_port_changes")) def test_normalize_allows_duplicate_metric_with_distinct_sheet_id(self) -> None: from netx_api.biz_state.compare_service import _normalize_sheet, sheet_key diff --git a/tests/test_biz_state_parse_pool.py b/tests/test_biz_state_parse_pool.py new file mode 100644 index 0000000..dbf7e1d --- /dev/null +++ b/tests/test_biz_state_parse_pool.py @@ -0,0 +1,100 @@ +"""Collect/parse decoupling: fetch_raw + parse pool job.""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from netx_api.biz_state.collect_session import CollectSession +from netx_api.biz_state.parse_pool import AuxRawCapture, PrimaryParseJob, parse_async_enabled + + +class FetchRawTests(unittest.TestCase): + def test_fetch_raw_skips_parser(self) -> None: + conn = object() + sends: list[str] = [] + + def _send(_c, command, read_timeout=0): + sends.append(command) + return f"RAW:{command}" + + session = CollectSession(conn, send_fn=_send, read_timeout=10) + with patch("netx_api.biz_state.collect_session.run_parser") as rp: + entry, hit = session.fetch_raw("show ip route") + rp.assert_not_called() + self.assertFalse(hit) + self.assertTrue(entry.ok) + self.assertEqual(entry.raw, "RAW:show ip route") + self.assertEqual(entry.records, []) + # Second call hits cache without re-CLI. + entry2, hit2 = session.fetch_raw("show ip route") + self.assertTrue(hit2) + self.assertEqual(len(sends), 1) + self.assertEqual(entry2.raw, entry.raw) + + +class ParseJobTests(unittest.TestCase): + def test_run_primary_parse_job_persists(self) -> None: + from netx_api.biz_state import collect_runner as cr + + with tempfile.TemporaryDirectory() as tmp: + with patch("netx_api.biz_state.spool.settings") as st: + st.biz_state_spool_dir = tmp + st.biz_state_persist_every_cmds = 8 + st.biz_state_raw_max_bytes = 0 + submitted: list = [] + + mock_pool = MagicMock() + mock_pool.submit.side_effect = lambda bid, items: submitted.append( + (bid, list(items)) + ) + + with patch.object(cr, "run_primary_with_bundle") as rpb: + rpb.return_value = ( + [{"prefix": "1.1.1.1/32"}], + {"rule_a": [{"PREFIX": "1.1.1.1/32"}]}, + ["rule_a"], + ) + with patch( + "netx_api.biz_state.persist_pool.get_persist_pool", + return_value=mock_pool, + ): + job = PrimaryParseJob( + batch_id="batch1", + cmd_id="cmd1", + task_item_id="item1", + profile_id="zte.ip_route", + parser_id="ip_route", + metric_id="ip_route", + concrete="show ip forwarding route", + merged_params={}, + raw_text="DESTINATION\n1.1.1.1/32", + raw_rel_path="", + textfsm_command="show ip forwarding route", + vendor="ZTE", + device_type="zte_zxros", + enrich_joins=[], + aux_captures=[], + persisted=set(), + ) + ok, fail = cr._run_primary_parse_job(job) + + self.assertTrue(ok) + self.assertFalse(fail) + self.assertEqual(len(submitted), 1) + _bid, items = submitted[0] + self.assertEqual(_bid, "batch1") + self.assertEqual(len(items), 1) + self.assertEqual(items[0].parse_status, "ok") + self.assertEqual(items[0].row_count, 1) + rec_path = Path(tmp) / "batch1" / "cmd1.records.jsonl" + self.assertTrue(rec_path.is_file()) + + def test_parse_async_default_on(self) -> None: + self.assertTrue(parse_async_enabled()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_iface_normalize.py b/tests/test_iface_normalize.py index 0c161a4..94bbcbb 100644 --- a/tests/test_iface_normalize.py +++ b/tests/test_iface_normalize.py @@ -49,6 +49,19 @@ class PortMapSubifTests(unittest.TestCase): self.assertEqual(resolve_mapped_iface("gei-0/0/0/1.100", pmap), "xgei-0/1/0/1.100") self.assertEqual(resolve_mapped_iface("gei-0/0/0/2.100", pmap), "gei-0/0/0/2.100") + def test_qinq_multi_level_parent(self) -> None: + """QinQ a.b.c: map parent a.b or a, keep remaining suffix.""" + pmap = {"gei-0/1/0/1": "xgei-0/2/0/1"} + self.assertEqual( + resolve_mapped_iface("gei-0/1/0/1.100.200", pmap), + "xgei-0/2/0/1.100.200", + ) + pmap2 = {"gei-0/1/0/1.100": "xgei-0/2/0/1.100"} + self.assertEqual( + resolve_mapped_iface("gei-0/1/0/1.100.200", pmap2), + "xgei-0/2/0/1.100.200", + ) + def test_apply_port_map_row(self) -> None: row = apply_port_map( {"interface": "gei-0/0/0/1.55", "admin": "up"}, diff --git a/tests/test_zte_extended_parsers.py b/tests/test_zte_extended_parsers.py index 1e85d6e..ef534e3 100644 --- a/tests/test_zte_extended_parsers.py +++ b/tests/test_zte_extended_parsers.py @@ -1134,6 +1134,69 @@ Routes Learned From This Neighbor: self.assertEqual(v4_wrap[0]["next_hop"], "10.0.0.1") self.assertEqual(v4_wrap[0]["afi"], "ipv4") + def test_bgp_route_rr_status_i_and_rd(self) -> None: + """RR neighbor-in uses '* i prefix' and Route Distinguisher fill-down.""" + from pathlib import Path + + from netx_api.biz_state.profiles import metric_field_map + + fixture = Path(__file__).resolve().parents[2] / "test" / "show-zte" / "show-bgp-vpnv4-neighbor-router-in" + if not fixture.is_file(): + # Workspace layout: chatgpt/test/show-zte vs netx/tests + fixture = Path(__file__).resolve().parents[3] / "test" / "show-zte" / "show-bgp-vpnv4-neighbor-router-in" + raw = fixture.read_text(encoding="utf-8", errors="replace") + routes = normalize_bgp_route( + raw_text=raw, + command="show bgp vpnv4 unicast neighbor in 114.0.24.93 | one-line", + vendor="ZTE", + device_type="zte_zxros", + params={"neighbor": "114.0.24.93", "direction": "in", "afi": "vpnv4"}, + ) + self.assertEqual(len(routes), 1905, "must match Total number of routes") + self.assertTrue(all("/" in str(r.get("network") or "") for r in routes)) + self.assertTrue(all(str(r.get("network") or "").lower() != "i" for r in routes)) + self.assertTrue(all(str(r.get("status_codes") or "") == "*i" for r in routes)) + self.assertTrue(all(str(r.get("rd") or "").strip() for r in routes)) + # Known multi-RD prefix must survive under both RDs (not collapsed by uniqueness) + multi = [r for r in routes if r.get("network") == "100.127.58.68/30"] + self.assertEqual(len(multi), 2) + self.assertEqual( + {r["rd"] for r in multi}, + {"114.0.141.200:65013", "114.14.249.211:65013"}, + ) + by_key = {(r["rd"], r["network"], r["next_hop"]) for r in routes} + self.assertEqual(len(by_key), len(routes)) + # Profile uniqueness keys must include RD + next_hop for ECMP / multi-RD + keys = {f.name for f in metric_field_map().get("bgp_route", []) if f.is_key} + self.assertTrue({"rd", "network", "next_hop", "neighbor", "direction"} <= keys) + + def test_bgp_route_ecmp_keeps_distinct_next_hops(self) -> None: + """Same RD+prefix with two next-hops must both persist (load-share).""" + # Match real ZTE layout: RD only captured after entering Routes state + # (via "Routes Learned…" / Network header), same as production fixture. + raw = """ +Routes Learned From This Neighbor: +Status codes: * valid, i - internal + Network Next Hop Metric LocPrf RtPrf Path +Route Distinguisher:10.0.0.1:100 +* i 192.0.2.0/24 10.1.1.1 0 100 0 65001 i +* i 192.0.2.0/24 10.1.1.2 0 100 0 65001 i +Total number of routes: 2 +""" + routes = normalize_bgp_route( + raw_text=raw, + command="show bgp vpnv4 unicast neighbor in 10.0.0.1 | one-line", + vendor="ZTE", + device_type="zte_zxros", + params={"neighbor": "10.0.0.1", "direction": "in", "afi": "vpnv4"}, + ) + self.assertEqual(len(routes), 2) + self.assertEqual({r["next_hop"] for r in routes}, {"10.1.1.1", "10.1.1.2"}) + self.assertEqual({r["rd"] for r in routes}, {"10.0.0.1:100"}) + self.assertEqual({r["network"] for r in routes}, {"192.0.2.0/24"}) + self.assertTrue(all(r.get("status_codes") == "*i" for r in routes)) + by_key = {(r["rd"], r["network"], r["next_hop"]) for r in routes} + self.assertEqual(len(by_key), 2) if __name__ == "__main__": unittest.main() diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 6ed4ff3..0cc19bb 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -266,14 +266,19 @@ const en = { deselectAllVrfs: "Deselect all", batches: "Batches", viewBatch: "Open", + batchDetail: "Detail", export: "Export", exportCommands: "Export collect commands", exportCommandsOk: "Collect commands exported", + exportRawLog: "Export as text", + exportRawLogOk: "Log exported", noBatches: "No batches yet", - batchDetail: "Batch detail", + batchCollectDetail: "Collect detail", batchWorkbook: "Batch workbook", batchId: "Batch ID", copyBatchId: "Copy batch ID", + batchMessageEmpty: "No message or error", + batchCmdSummary: "{{ok}} ok / {{fail}} failed / {{other}} other · {{total}} commands", sheetCommands: "Commands", sheetLldp: "LLDP neighbors", sheetVrfRoute: "VRF route summary", @@ -292,6 +297,8 @@ const en = { viewRawLog: "Raw log", rawLogTitle: "Raw collect log", rawLogEmpty: "No raw output", + rawLogStats: "{{lines}} lines collected · {{rows}} rows stored", + openWorkbook: "Open workbook", filterColumn: "Column", filterAllCols: "All columns", listFilterPh: "Filter NE / IP / vendor / status", @@ -306,7 +313,8 @@ const en = { colStarted: "Started", colEnded: "Ended", colTime: "Time", - colRows: "Rows", + colRows: "DB rows", + colRawLines: "CLI lines", colActions: "Actions", colCommand: "Command", colMessage: "Message", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index a2364d4..f1ddf2a 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -266,14 +266,19 @@ const zh = { deselectAllVrfs: "取消全选", batches: "采集批次", viewBatch: "查看", + batchDetail: "详情", export: "导出", exportCommands: "导出采集命令", exportCommandsOk: "已导出采集命令", + exportRawLog: "导出为文本", + exportRawLogOk: "日志已导出", noBatches: "尚无批次", - batchDetail: "批次详情", + batchCollectDetail: "采集详情", batchWorkbook: "批次工作簿", batchId: "批次 ID", copyBatchId: "复制批次 ID", + batchMessageEmpty: "无额外说明或报错", + batchCmdSummary: "命令 {{ok}} 成功 / {{fail}} 失败 / {{other}} 其他 · 共 {{total}} 条", sheetCommands: "命令", sheetLldp: "LLDP 邻居", sheetVrfRoute: "VRF 路由摘要", @@ -292,6 +297,8 @@ const zh = { viewRawLog: "原始日志", rawLogTitle: "原始采集日志", rawLogEmpty: "无原始输出", + rawLogStats: "采集 {{lines}} 行 · 入库 {{rows}} 条", + openWorkbook: "打开工作簿", filterColumn: "筛选列", filterAllCols: "全部列", listFilterPh: "筛选网元 / IP / 厂商 / 状态", @@ -306,7 +313,8 @@ const zh = { colStarted: "开始时间", colEnded: "结束时间", colTime: "时间", - colRows: "行数", + colRows: "入库条数", + colRawLines: "采集行数", colActions: "操作", colCommand: "命令", colMessage: "说明", diff --git a/web/src/index.css b/web/src/index.css index 3b63d36..5c82c03 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -10085,6 +10085,71 @@ html.login-page--paused .login-page__flare { min-height: min(50vh, 480px); } +.bs-rawlog-stats { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 8px 12px; +} + +.bs-batch-message { + display: flex; + flex-direction: column; + gap: 4px; + width: 100%; + padding: 8px 10px; + border-radius: 8px; + border: 1px solid rgba(148, 163, 184, 0.28); + background: rgba(15, 23, 42, 0.55); +} + +.bs-batch-message__label { + font-size: 11px; + letter-spacing: 0.04em; + text-transform: uppercase; + color: #94a3b8; +} + +.bs-batch-message__body { + margin: 0; + white-space: pre-wrap; + word-break: break-word; + font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; + font-size: 12px; + line-height: 1.45; + color: #e2e8f0; +} + +.bs-batch-message--alert { + border-color: rgba(248, 113, 113, 0.28); + background: rgba(127, 29, 29, 0.22); +} + +.bs-batch-message--alert .bs-batch-message__label { + color: #fca5a5; +} + +.bs-batch-message--alert .bs-batch-message__body { + color: #fecaca; +} + +.bs-msg-cell { + max-width: 220px; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + font-size: 12px; +} + +.bs-cmd-stat { + font-size: 11px; + white-space: nowrap; +} + +.bs-collect-detail-modal .bs-sheet-table { + max-height: min(48vh, 480px); +} + .bs-rawlog-pre { flex: 1 1 auto; margin: 0; diff --git a/web/src/pages/network/BizStatePage.tsx b/web/src/pages/network/BizStatePage.tsx index d4cc7a1..012b917 100644 --- a/web/src/pages/network/BizStatePage.tsx +++ b/web/src/pages/network/BizStatePage.tsx @@ -14,6 +14,7 @@ import { bizStateDeleteBatch, bizStateDeleteTask, bizStateDiscover, + bizStateDownloadCommandRaw, bizStateDownloadExport, bizStateDownloadTaskCommands, bizStateGetBatch, @@ -86,6 +87,7 @@ type BatchRow = { status: string; row_count: number; command_count: number; + message?: string; started_at?: string | null; ended_at?: string | null; is_baseline?: boolean; @@ -108,6 +110,7 @@ type SheetCmd = { raw_command: string; parse_status?: string; row_count?: number; + raw_line_count?: number; message?: string; has_raw?: boolean; profile_id?: string; @@ -305,6 +308,13 @@ export function BizStatePage() { const [rawLogCmd, setRawLogCmd] = useState(""); const [rawLogText, setRawLogText] = useState(""); const [rawLogMeta, setRawLogMeta] = useState(""); + const [rawLogCommandId, setRawLogCommandId] = useState(""); + const [rawLogLines, setRawLogLines] = useState(0); + const [rawLogRows, setRawLogRows] = useState(0); + const [rawLogMessage, setRawLogMessage] = useState(""); + /** Collect status / errors detail (lighter than workbook). */ + const [collectDetail, setCollectDetail] = useState(null); + const [collectDetailLoading, setCollectDetailLoading] = useState(false); const refreshTasks = useCallback(async () => { const purpose = @@ -460,6 +470,7 @@ export function BizStatePage() { { key: "raw_command", header: t("bizState.colCommand") }, { key: "metric_id", header: "metric" }, { key: "parse_status", header: t("bizState.colStatus") }, + { key: "raw_line_count", header: t("bizState.colRawLines") }, { key: "row_count", header: t("bizState.colRows") }, { key: "message", header: t("bizState.colMessage") }, { key: "_actions", header: t("bizState.colActions") }, @@ -483,7 +494,7 @@ export function BizStatePage() { : cmds.filter((row) => { const keys = sheetColumn ? [sheetColumn] - : ["raw_command", "metric_id", "parse_status", "row_count", "message"]; + : ["raw_command", "metric_id", "parse_status", "raw_line_count", "row_count", "message"]; return keys.some((k) => cellText(row[k]).toLowerCase().includes(kw)); }); const start = (sheetPage - 1) * sheetPageSize; @@ -509,13 +520,27 @@ export function BizStatePage() { return cmds.filter((row) => { const keys = sheetColumn ? [sheetColumn] - : ["raw_command", "metric_id", "parse_status", "row_count", "message"]; + : ["raw_command", "metric_id", "parse_status", "raw_line_count", "row_count", "message"]; return keys.some((k) => cellText(row[k]).toLowerCase().includes(kw)); }).length; } return sheetTotal; }, [activeSheet, batchDetail, debouncedSheetKw, sheetColumn, sheetTotal]); + const collectDetailCmdSummary = useMemo(() => { + const cmds = (collectDetail?.commands || []) as SheetCmd[]; + let ok = 0; + let fail = 0; + let other = 0; + for (const c of cmds) { + const st = String(c.parse_status || "").toLowerCase(); + if (st === "ok" || st === "success" || st === "aux" || st === "aux_ok" || st === "aux_cached") ok += 1; + else if (st === "failed" || st === "error" || st === "fail") fail += 1; + else other += 1; + } + return { ok, fail, other, total: cmds.length }; + }, [collectDetail]); + const loadSheetPage = useCallback( async (batchId: string, metricId: string, page: number, pageSize: number, kw: string, column: string) => { if (!batchId || !metricId || metricId === "commands") return; @@ -1067,23 +1092,52 @@ export function BizStatePage() { setSheetTotal(0); setRawLogOpen(false); setRawLogText(""); + setRawLogCommandId(""); }; - const openRawLog = async (commandId: string) => { - if (!batchDetail?.id || !commandId) return; + const openCollectDetail = async (batchId: string) => { + setCollectDetailLoading(true); + setCollectDetail(null); + try { + const d = await bizStateGetBatch(batchId); + setCollectDetail(d); + } catch (e) { + showError(formatErr(e)); + } finally { + setCollectDetailLoading(false); + } + }; + + const closeCollectDetail = () => { + setCollectDetail(null); + setCollectDetailLoading(false); + }; + + const openRawLog = async (commandId: string, fromBatchId?: string) => { + const bid = fromBatchId || (batchDetail?.id ? String(batchDetail.id) : "") || (collectDetail?.id ? String(collectDetail.id) : ""); + if (!bid || !commandId) return; setRawLogOpen(true); setRawLogLoading(true); setRawLogText(""); setRawLogCmd(""); setRawLogMeta(""); + setRawLogCommandId(commandId); + setRawLogLines(0); + setRawLogRows(0); + setRawLogMessage(""); try { - const d = await bizStateGetBatchCommand(String(batchDetail.id), commandId); + const d = await bizStateGetBatchCommand(bid, commandId); setRawLogCmd(String(d.raw_command || "")); setRawLogText(String(d.raw_text || "")); + const lines = Number(d.raw_line_count ?? 0); + const rows = Number(d.row_count ?? 0); + setRawLogLines(lines); + setRawLogRows(rows); + setRawLogMessage(String(d.message || "")); const bits = [ d.parse_status, d.metric_id, - d.row_count != null ? `${d.row_count} rows` : "", + t("bizState.rawLogStats", { lines, rows }), d.collected_at ? fmtTime(d.collected_at) : "", ].filter(Boolean); setRawLogMeta(bits.join(" · ")); @@ -1095,6 +1149,35 @@ export function BizStatePage() { } }; + const exportRawLog = async () => { + const bid = + (batchDetail?.id && String(batchDetail.id)) || + (collectDetail?.id && String(collectDetail.id)) || + ""; + if (!bid || !rawLogCommandId) { + // Fallback: download already-loaded text + if (!rawLogText) return; + const safe = (rawLogCmd || "command").replace(/[^\w.-]+/g, "_").slice(0, 80); + const blob = new Blob([rawLogText], { type: "text/plain;charset=utf-8" }); + const url = URL.createObjectURL(blob); + try { + const a = document.createElement("a"); + a.href = url; + a.download = `${safe || "command"}.txt`; + a.click(); + } finally { + URL.revokeObjectURL(url); + } + return; + } + try { + await bizStateDownloadCommandRaw(bid, rawLogCommandId); + showOk(t("bizState.exportRawLogOk")); + } catch (e) { + showError(formatErr(e)); + } + }; + const selectSheet = (id: string) => { setSheetId(id); setSheetKeyword(""); @@ -1785,6 +1868,9 @@ export function BizStatePage() {
+ @@ -2079,6 +2165,21 @@ export function BizStatePage() { {t("bizState.setAlias")}
+ {batchDetail.message ? ( +
+ {t("bizState.colMessage")} + 
{String(batchDetail.message)}
+
+ ) : null} ) : null} @@ -2111,6 +2212,12 @@ export function BizStatePage() { {c.raw_command || "—"}
+ + {t("bizState.rawLogStats", { + lines: Number(c.raw_line_count ?? 0), + rows: Number(c.row_count ?? 0), + })} + {c.parse_status ? ( {c.parse_status} @@ -2177,14 +2284,29 @@ export function BizStatePage() { {displayColumns.map((c) => ( {c.key === "_actions" ? ( - +
+ + +
) : c.key === "parse_status" ? ( {cellText(row[c.key]) || "—"} @@ -2259,7 +2381,18 @@ export function BizStatePage() { {rawLogCmd ? {rawLogCmd} : null} - {rawLogMeta ?

{rawLogMeta}

: null} +
+ + {t("bizState.rawLogStats", { lines: rawLogLines, rows: rawLogRows })} + + {rawLogMeta ? {rawLogMeta} : null} +
+ {rawLogMessage ? ( +
+ {t("bizState.colMessage")} + 
{rawLogMessage}
+
+ ) : null} {rawLogLoading ? (
{t("bizState.sheetLoading")}
) : ( @@ -2267,11 +2400,166 @@ export function BizStatePage() { )} + + + {/* Batch collect status / errors detail */} + + + {t("bizState.batchCollectDetail")} + + + + {collectDetailLoading ? ( +
{t("bizState.sheetLoading")}
+ ) : collectDetail ? ( + <> +
+ + {String(collectDetail.status || "—")} + + {collectDetail.alias ? ( + {String(collectDetail.alias)} + ) : null} + + {collectDetail.command_count} cmd · {collectDetail.row_count} rows ·{" "} + {t("bizState.colStarted")} {fmtTime(collectDetail.started_at)} ·{" "} + {t("bizState.colEnded")} {fmtTime(collectDetail.ended_at)} + {fmtDuration(collectDetail.started_at, collectDetail.ended_at) + ? ` · ${fmtDuration(collectDetail.started_at, collectDetail.ended_at)}` + : ""} + +
+
+ {t("bizState.colMessage")} + 
+                  {String(collectDetail.message || "").trim() || t("bizState.batchMessageEmpty")}
+
+
+

+ {t("bizState.batchCmdSummary", { + ok: collectDetailCmdSummary.ok, + fail: collectDetailCmdSummary.fail, + other: collectDetailCmdSummary.other, + total: collectDetailCmdSummary.total, + })} +

+
+ + + + + + + + + + + + + {((collectDetail.commands || []) as SheetCmd[]).map((c) => ( + + + + + + + + + ))} + {!((collectDetail.commands || []) as SheetCmd[]).length ? ( + + + + ) : null} + +
{t("bizState.colCommand")}{t("bizState.colStatus")}{t("bizState.colRawLines")}{t("bizState.colRows")}{t("bizState.colMessage")}{t("bizState.colActions")}
+ {c.raw_command || "—"} + + + {c.parse_status || "—"} + + {Number(c.raw_line_count ?? 0)}{Number(c.row_count ?? 0)} + {c.message?.trim() ? c.message : "—"} + +
+ + +
+
+
{t("bizState.sheetEmpty")}
+
+
+ + ) : null} + + + {collectDetail?.id ? ( + <> + + + + ) : null} + + + ); } diff --git a/web/src/services/api.ts b/web/src/services/api.ts index ff696ce..88eb57e 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -1907,6 +1907,7 @@ export const bizStateGetBatchCommand = (batchId: string, commandId: string) => metric_id: string; parse_status: string; row_count: number; + raw_line_count?: number; message: string; raw_text: string; collected_at?: string | null; @@ -1915,6 +1916,31 @@ export const bizStateGetBatchCommand = (batchId: string, commandId: string) => `/v1/biz-state/batches/${encodeURIComponent(batchId)}/commands/${encodeURIComponent(commandId)}`, ); +/** Download one command's raw CLI output as .txt */ +export const bizStateDownloadCommandRaw = async (batchId: string, commandId: string): Promise => { + const path = + `/v1/biz-state/batches/${encodeURIComponent(batchId)}/commands/${encodeURIComponent(commandId)}/raw.txt`; + const res = await fetch(path, { method: "GET", credentials: fetchCreds, headers: authHeaders() }); + if (res.status === 401) { + handleUnauthorized(path); + throw new Error("unauthorized"); + } + if (!res.ok) throw new Error(`${res.status} download raw`); + const blob = await res.blob(); + const cd = res.headers.get("Content-Disposition") || ""; + const m = /filename="?([^";]+)"?/i.exec(cd); + const filename = m?.[1] || `biz_state_${commandId}.txt`; + const url = URL.createObjectURL(blob); + try { + const a = document.createElement("a"); + a.href = url; + a.download = filename; + a.click(); + } finally { + URL.revokeObjectURL(url); + } +}; + export const bizStateDownloadExport = async (batchId: string): Promise => { const path = `/v1/biz-state/batches/${encodeURIComponent(batchId)}/export`; const res = await fetch(path, { method: "GET", credentials: fetchCreds, headers: authHeaders() });