mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 03:10:46 +08:00
Harden UME/LLDP link ports and canvas styles around UME authority.
Resolve ports from EQ+PTP first, enrich from LLDP when compatible, miss only on valid LLDP evidence, and paint UME solid blue / LLDP dashed / missing red. Also localize new root maps as Root map in English. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
77832b271d
commit
137597cd25
16 changed files with 736 additions and 152 deletions
|
|
@ -1,4 +1,14 @@
|
|||
"""Normalize UME TP DN / userLabel into CLI-like ifnames for Fabric merge."""
|
||||
"""Normalize UME TP DN / userLabel into CLI-like ifnames for Fabric merge.
|
||||
|
||||
Authority order for each link end:
|
||||
1. EQ+PTP → numeric ``shelf/slot/…/port`` (same as LLDP ``x/x/x/x``)
|
||||
2. userLabel media token whose numeric tail matches that suffix (``xxvgei-…``)
|
||||
3. ``[slot]_ETH:N`` / ``NGE:N`` forms that match the suffix
|
||||
4. bare TP suffix (still LLDP-compatible via ``port_keys_compatible``)
|
||||
|
||||
Never assign A/Z from label token order alone — labels often embed the
|
||||
*neighbor* port first and swap the two ends.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -11,8 +21,12 @@ _IFNAME_TOKEN_RE = re.compile(
|
|||
r"(?i)(?<![A-Za-z0-9])((?:xxvgei|xgei|cgei|xlgei|lgei|gei|gi|eth)"
|
||||
r"(?:-[\d./]+|[\d./]+))"
|
||||
)
|
||||
_ETH_COLON_RE = re.compile(r"(?i)\bETH:(\d+)\b")
|
||||
_BRACKET_SLOT_RE = re.compile(r"\[(\d+/\d+/\d+)\]")
|
||||
# `_ETH:28` — `\b` fails after `_` (word char); use lookbehind instead.
|
||||
_ETH_COLON_RE = re.compile(r"(?i)(?<![A-Za-z0-9])ETH:(\d+)\b")
|
||||
# `25GE:14` / `10GE:32` style local port markers in some regions' labels.
|
||||
_RATE_GE_RE = re.compile(r"(?i)(?<![A-Za-z0-9])(?:\d+)?GE:(\d+)\b")
|
||||
# `[1/1/0]` or `[0-1-1]`
|
||||
_BRACKET_SLOT_RE = re.compile(r"\[(\d+[/\-]\d+[/\-]\d+)\]")
|
||||
_EQ_SH_RE = re.compile(r"(?i)/sh=(\d+)")
|
||||
_EQ_SL_RE = re.compile(r"(?i)/sl=(\d+)")
|
||||
_EQ_SSL_RE = re.compile(r"(?i)/ssl=(\d+)")
|
||||
|
|
@ -34,11 +48,25 @@ def extract_ifnames_from_user_label(label: str) -> list[str]:
|
|||
key = normalize_ifname(raw)
|
||||
if not key or key in seen:
|
||||
continue
|
||||
# Drop truncated junk like ``cgei-0`` / ``xxvgei-0`` from cut labels.
|
||||
if numeric_port_tail(key).count("/") < 2:
|
||||
continue
|
||||
seen.add(key)
|
||||
out.append(key)
|
||||
return out
|
||||
|
||||
|
||||
def numeric_port_tail(ifname: str) -> str:
|
||||
"""Digits path after media prefix: ``xxvgei-1/1/0/14`` → ``1/1/0/14``."""
|
||||
s = normalize_ifname(ifname)
|
||||
if not s:
|
||||
return ""
|
||||
for i, ch in enumerate(s):
|
||||
if ch.isdigit():
|
||||
return s[i:]
|
||||
return ""
|
||||
|
||||
|
||||
def port_suffix_from_tp_ref(tp_ref: str) -> str:
|
||||
"""Build shelf/slot/…/port suffix from EQ+PTP, e.g. ``1/1/0/32`` or ``0/0/1/4``."""
|
||||
s = str(tp_ref or "")
|
||||
|
|
@ -58,15 +86,56 @@ def port_suffix_from_tp_ref(tp_ref: str) -> str:
|
|||
return f"{sh}/{sl}/0/{port}"
|
||||
|
||||
|
||||
def _norm_bracket_slot(raw: str) -> str:
|
||||
return str(raw or "").strip().replace("-", "/")
|
||||
|
||||
|
||||
def iter_eth_style_suffixes(label: str) -> list[str]:
|
||||
"""``[1/1/0]_ETH:28`` / ``[0-1-1]-25GE:14`` → ``1/1/0/28`` / ``0/1/1/14``."""
|
||||
s = str(label or "")
|
||||
slots = [
|
||||
(m.start(), _norm_bracket_slot(m.group(1)))
|
||||
for m in _BRACKET_SLOT_RE.finditer(s)
|
||||
]
|
||||
ports: list[tuple[int, str]] = []
|
||||
for m in _ETH_COLON_RE.finditer(s):
|
||||
ports.append((m.start(), m.group(1)))
|
||||
for m in _RATE_GE_RE.finditer(s):
|
||||
ports.append((m.start(), m.group(1)))
|
||||
ports.sort(key=lambda x: x[0])
|
||||
if not slots or not ports:
|
||||
return []
|
||||
out: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for (_spos, slot), (_ppos, port) in zip(slots, ports):
|
||||
if slot.count("/") != 2:
|
||||
continue
|
||||
key = normalize_ifname(f"{slot}/{port}")
|
||||
if key and key not in seen:
|
||||
seen.add(key)
|
||||
out.append(key)
|
||||
return out
|
||||
|
||||
|
||||
def port_suffix_from_eth_label(label: str, *, end_index: int = 0) -> str:
|
||||
"""From ``…[1/1/0]_ETH:28_…`` build ``1/1/0/28`` (nth ETH occurrence)."""
|
||||
s = str(label or "")
|
||||
slots = _BRACKET_SLOT_RE.findall(s)
|
||||
eths = _ETH_COLON_RE.findall(s)
|
||||
if not slots or not eths:
|
||||
all_suf = iter_eth_style_suffixes(label)
|
||||
if not all_suf:
|
||||
return ""
|
||||
i = max(0, min(int(end_index), len(slots) - 1, len(eths) - 1))
|
||||
return f"{slots[i]}/{eths[i]}"
|
||||
i = max(0, min(int(end_index), len(all_suf) - 1))
|
||||
return all_suf[i]
|
||||
|
||||
|
||||
def find_ifname_matching_suffix(suffix: str, user_label: str) -> str:
|
||||
"""Return a label CLI ifname whose numeric tail equals ``suffix``."""
|
||||
want = normalize_ifname(suffix)
|
||||
if not want:
|
||||
return ""
|
||||
for tok in extract_ifnames_from_user_label(user_label):
|
||||
tail = numeric_port_tail(tok)
|
||||
if tail == want or normalize_ifname(tok) == want:
|
||||
return tok
|
||||
return ""
|
||||
|
||||
|
||||
def resolve_ume_ifname(
|
||||
|
|
@ -77,21 +146,21 @@ def resolve_ume_ifname(
|
|||
) -> str:
|
||||
"""Best-effort CLI ifname for one end of a UME link.
|
||||
|
||||
Preference: userLabel ifname token → ETH:+bracket → EQ+PTP suffix.
|
||||
Preference: EQ+PTP suffix → matching label media token → ETH/GE match → bare suffix.
|
||||
Without TP: return empty (caller should show ``userLabel``, not invent A/Z ports).
|
||||
"""
|
||||
tokens = extract_ifnames_from_user_label(user_label)
|
||||
if tokens:
|
||||
idx = max(0, min(int(end_index), len(tokens) - 1))
|
||||
return tokens[idx][:128]
|
||||
|
||||
eth = port_suffix_from_eth_label(user_label, end_index=end_index)
|
||||
if eth:
|
||||
return normalize_ifname(eth)[:128]
|
||||
|
||||
_ = end_index # kept for call-site compatibility; unused without TP
|
||||
suffix = port_suffix_from_tp_ref(tp_ref)
|
||||
if suffix:
|
||||
return normalize_ifname(suffix)[:128]
|
||||
return ""
|
||||
if not suffix:
|
||||
return ""
|
||||
hit = find_ifname_matching_suffix(suffix, user_label)
|
||||
if hit:
|
||||
return hit[:128]
|
||||
want = normalize_ifname(suffix)
|
||||
for eth in iter_eth_style_suffixes(user_label):
|
||||
if normalize_ifname(eth) == want:
|
||||
return want[:128]
|
||||
return want[:128]
|
||||
|
||||
|
||||
def resolve_link_ifnames(
|
||||
|
|
@ -100,35 +169,69 @@ def resolve_link_ifnames(
|
|||
z_end_tp_ref: str,
|
||||
user_label: str = "",
|
||||
) -> tuple[str, str]:
|
||||
tokens = extract_ifnames_from_user_label(user_label)
|
||||
if len(tokens) >= 2:
|
||||
return tokens[0][:128], tokens[1][:128]
|
||||
if len(tokens) == 1:
|
||||
a = tokens[0]
|
||||
z = resolve_ume_ifname(tp_ref=z_end_tp_ref, user_label=user_label, end_index=1)
|
||||
return a[:128], z[:128]
|
||||
"""Resolve A/Z independently — TP pins the port; label only supplies media prefix.
|
||||
|
||||
If neither end has EQ+PTP, returns empty ports (display ``userLabel`` instead).
|
||||
"""
|
||||
return (
|
||||
resolve_ume_ifname(tp_ref=a_end_tp_ref, user_label=user_label, end_index=0),
|
||||
resolve_ume_ifname(tp_ref=z_end_tp_ref, user_label=user_label, end_index=1),
|
||||
)
|
||||
|
||||
|
||||
def is_label_placeholder_port(ifname: str) -> bool:
|
||||
"""Synthetic fabric port for UME links that only have userLabel (no TP)."""
|
||||
return normalize_ifname(ifname).startswith("label:")
|
||||
|
||||
|
||||
def label_placeholder_ports(link_id: str) -> tuple[str, str]:
|
||||
lid = str(link_id or "").strip() or "unknown"
|
||||
return (f"label:{lid}:a"[:128], f"label:{lid}:z"[:128])
|
||||
|
||||
|
||||
def port_keys_compatible(a: str, b: str) -> bool:
|
||||
"""True if two ports are the same after normalize, or share a numeric suffix."""
|
||||
"""True if two ports are the same after normalize, or share the same numeric path.
|
||||
|
||||
Uses exact equality of ``numeric_port_tail`` (e.g. ``xxvgei-1/1/0/32`` ↔ ``1/1/0/32``).
|
||||
Does **not** use naive ``endswith`` (avoids ``11/1/0/1`` matching ``1/1/0/1``).
|
||||
"""
|
||||
na = normalize_ifname(a)
|
||||
nb = normalize_ifname(b)
|
||||
if not na or not nb:
|
||||
return False
|
||||
if na == nb:
|
||||
return True
|
||||
# xxvgei-1/1/0/32 vs 1/1/0/32
|
||||
if na.endswith(nb) or nb.endswith(na):
|
||||
return True
|
||||
# strip alpha prefix before first digit
|
||||
def _num_tail(s: str) -> str:
|
||||
for i, ch in enumerate(s):
|
||||
if ch.isdigit():
|
||||
return s[i:]
|
||||
return s
|
||||
if is_label_placeholder_port(na) or is_label_placeholder_port(nb):
|
||||
return False
|
||||
ta = numeric_port_tail(na)
|
||||
tb = numeric_port_tail(nb)
|
||||
# Require a real shelf/…/port path (at least one slash).
|
||||
return bool(ta) and ta == tb and "/" in ta
|
||||
|
||||
return _num_tail(na) == _num_tail(nb) and bool(_num_tail(na))
|
||||
|
||||
_MEDIA_PREFIX_RE = re.compile(
|
||||
r"(?i)^(xxvgei|xgei|cgei|xlgei|lgei|gei|gi|eth)-"
|
||||
)
|
||||
|
||||
|
||||
def has_media_prefix(ifname: str) -> bool:
|
||||
return bool(_MEDIA_PREFIX_RE.match(normalize_ifname(ifname)))
|
||||
|
||||
|
||||
def prefer_richer_ifname(current: str, candidate: str) -> str:
|
||||
"""When ports are compatible, prefer the media-prefixed (LLDP-style) name.
|
||||
|
||||
Bare ``1/1/0/32`` + ``xxvgei-1/1/0/32`` → ``xxvgei-1/1/0/32``.
|
||||
Incompatible candidates are ignored (keep ``current``).
|
||||
"""
|
||||
cur = normalize_ifname(current)
|
||||
can = normalize_ifname(candidate)
|
||||
if not can:
|
||||
return cur
|
||||
if not cur:
|
||||
return can
|
||||
if not port_keys_compatible(cur, can):
|
||||
return cur
|
||||
if has_media_prefix(can) and not has_media_prefix(cur):
|
||||
return can
|
||||
return cur
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue