"""Event-driven watcher that propagates bridge member link failures to sibling ports.""" from __future__ import annotations import logging import os import select import threading import time from dataclasses import dataclass from typing import Any, Dict, Optional from pyroute2 import IPRoute from src.config import settings from src.utilities.interface_bridge_helpers import ( check_interface_exists, get_bridge_ports_once, read_interface_admin_up, read_interface_carrier, read_interface_operstate, ) logger = logging.getLogger("bridge_link_state_manager") @dataclass class MemberLinkState: """Current sysfs-derived state for one bridge member.""" ifname: str admin_up: Optional[bool] carrier_up: Optional[bool] operstate: Optional[str] @property def link_ready(self) -> bool: """Return whether the member currently looks healthy enough to forward.""" if self.admin_up is not True: return False if self.carrier_up is False: return False if self.operstate in {"down", "lowerlayerdown", "notpresent"}: return False return True def to_dict(self, suppressed: bool = False) -> Dict[str, Any]: """Serialize member state for API responses.""" return { "ifname": self.ifname, "admin_up": self.admin_up, "carrier_up": self.carrier_up, "operstate": self.operstate, "link_ready": self.link_ready, "suppressed": suppressed, } class BridgeLinkStateWatcher: """Watch one bridge and mirror member failures to the other bridge members.""" def __init__(self, bridge_name: str, recovery_holdoff_seconds: float) -> None: self.bridge_name = bridge_name self.recovery_holdoff_seconds = recovery_holdoff_seconds self._stop_event = threading.Event() self._lock = threading.Lock() self._wake_r, self._wake_w = os.pipe() os.set_blocking(self._wake_r, False) os.set_blocking(self._wake_w, False) self._thread = threading.Thread( target=self._run, daemon=True, name=f"bridge-link-state-{bridge_name}", ) self._running = False self._suppressed_members: dict[str, bool] = {} self._settle_deadlines: dict[str, float] = {} self._all_clear_since: Optional[float] = None self._last_event_ts: Optional[float] = None self._last_error: Optional[str] = None self._last_action: Optional[str] = None self._member_states: dict[str, MemberLinkState] = {} def start(self) -> None: """Start the watcher thread.""" with self._lock: if self._running: return self._running = True self._thread.start() def stop(self) -> None: """Stop the watcher and restore interfaces that were suppressed by it.""" self._stop_event.set() self._wake_thread() if self._thread.is_alive(): self._thread.join(timeout=settings.bridge_link_state_thread_join_timeout_seconds) try: self._restore_suppressed_members(reason="watcher_stopped") except Exception: logger.exception("Failed to restore suppressed members for bridge=%s", self.bridge_name) with self._lock: self._running = False for fd in (self._wake_r, self._wake_w): try: os.close(fd) except OSError: pass def status(self) -> dict[str, Any]: """Return a JSON-serializable snapshot of the watcher state.""" with self._lock: members = { ifname: state.to_dict(suppressed=ifname in self._suppressed_members) for ifname, state in sorted(self._member_states.items()) } return { "bridge": self.bridge_name, "active": self._running and self._thread.is_alive() and not self._stop_event.is_set(), "event_driven": True, "last_event_ts": self._last_event_ts, "last_error": self._last_error, "last_action": self._last_action, "suppressed_members": sorted(self._suppressed_members), "recovery_holdoff_seconds": self.recovery_holdoff_seconds, "members": members, } def _run(self) -> None: with IPRoute() as ipr: ipr.bind() self._evaluate_bridge_state(reason="watcher_started") while not self._stop_event.is_set(): try: timeout = self._next_wait_timeout() ready, _, _ = select.select([ipr, self._wake_r], [], [], timeout) except Exception as exc: logger.exception("Bridge link-state select failed for bridge=%s", self.bridge_name) with self._lock: self._last_error = str(exc) continue if self._stop_event.is_set(): break if self._wake_r in ready: self._drain_wake_pipe() continue if ipr in ready: try: messages = ipr.get() except Exception as exc: logger.exception("Bridge link-state netlink read failed for bridge=%s", self.bridge_name) with self._lock: self._last_error = str(exc) continue if any(msg.get("event") in {"RTM_NEWLINK", "RTM_DELLINK"} for msg in messages): self._evaluate_bridge_state(reason="netlink_event") continue self._evaluate_bridge_state(reason="recovery_deadline") def _evaluate_bridge_state(self, reason: str) -> None: members = [iface for iface in get_bridge_ports_once(self.bridge_name) if check_interface_exists(iface)] states = {iface: self._read_member_state(iface) for iface in members} now = time.time() with self._lock: self._last_event_ts = now self._last_error = None self._member_states = states self._suppressed_members = { iface: restore_up for iface, restore_up in self._suppressed_members.items() if iface in states } self._settle_deadlines = { iface: deadline for iface, deadline in self._settle_deadlines.items() if iface in states and deadline > now } if len(states) < 2: with self._lock: self._all_clear_since = None self._restore_suppressed_members(reason="bridge_has_fewer_than_two_members") return with self._lock: suppressed_snapshot = set(self._suppressed_members) settling_snapshot = { iface for iface, deadline in self._settle_deadlines.items() if deadline > now } failing_members = sorted( ifname for ifname, state in states.items() if ifname not in suppressed_snapshot and ifname not in settling_snapshot and not state.link_ready ) if failing_members: with self._lock: self._all_clear_since = None self._suppress_other_members(states, failing_members) return if suppressed_snapshot: should_restore = False with self._lock: if self._all_clear_since is None: self._all_clear_since = now should_restore = now - self._all_clear_since >= self.recovery_holdoff_seconds if not should_restore: remaining = max(0.0, self.recovery_holdoff_seconds - (now - self._all_clear_since)) self._last_action = f"waiting {remaining:.2f}s before restoring suppressed members" if should_restore: self._restore_suppressed_members(reason=reason) return with self._lock: self._all_clear_since = None self._restore_suppressed_members(reason=reason) def _next_wait_timeout(self) -> Optional[float]: """Return how long the watcher may sleep before the next restore deadline.""" with self._lock: if self._suppressed_members and self._all_clear_since is not None: deadline = self._all_clear_since + self.recovery_holdoff_seconds return max(0.0, deadline - time.time()) return None def _wake_thread(self) -> None: """Wake the event loop from another thread.""" try: os.write(self._wake_w, b"\x00") except OSError: pass def _drain_wake_pipe(self) -> None: """Drain pending wake-up bytes from the control pipe.""" try: while os.read(self._wake_r, 4096): pass except BlockingIOError: return except OSError: return def _read_member_state(self, ifname: str) -> MemberLinkState: return MemberLinkState( ifname=ifname, admin_up=read_interface_admin_up(ifname), carrier_up=read_interface_carrier(ifname), operstate=read_interface_operstate(ifname), ) def _suppress_other_members(self, states: dict[str, MemberLinkState], failing_members: list[str]) -> None: desired_suppressed = set(states) - set(failing_members) with self._lock: current_suppressed = dict(self._suppressed_members) next_suppressed: dict[str, bool] = {} changed_members: list[str] = [] for ifname in sorted(desired_suppressed): restore_up = current_suppressed.get(ifname, states[ifname].admin_up is True) if ifname not in current_suppressed and states[ifname].admin_up is True: self._set_interface_admin_state(ifname, target_up=False) changed_members.append(ifname) next_suppressed[ifname] = restore_up with self._lock: self._suppressed_members = next_suppressed action = ( f"suppressed {changed_members} because failing members={failing_members}" if changed_members else f"holding suppressed members because failing members={failing_members}" ) self._last_action = action def _restore_suppressed_members(self, reason: str) -> None: with self._lock: suppressed = dict(self._suppressed_members) if not suppressed: with self._lock: self._last_action = f"no_restore_needed ({reason})" return restored_members: list[str] = [] for ifname, restore_up in sorted(suppressed.items()): if not check_interface_exists(ifname): continue if restore_up: self._set_interface_admin_state(ifname, target_up=True) restored_members.append(ifname) with self._lock: self._suppressed_members = {} self._all_clear_since = None self._last_action = f"restored {restored_members} ({reason})" def _set_interface_admin_state(self, ifname: str, target_up: bool) -> None: if not check_interface_exists(ifname): raise RuntimeError(f"Interface {ifname} disappeared while propagating link state") state_name = "up" if target_up else "down" with IPRoute() as ipr: indices = ipr.link_lookup(ifname=ifname) if not indices: raise RuntimeError(f"Interface {ifname} not found while propagating link state") ipr.link("set", index=indices[0], state=state_name) if target_up: with self._lock: self._settle_deadlines[ifname] = time.time() + self.recovery_holdoff_seconds class BridgeLinkStateManager: """Track bridge link-state watchers keyed by bridge name.""" def __init__(self) -> None: self._watchers: dict[str, BridgeLinkStateWatcher] = {} self._lock = threading.Lock() def enable(self, bridge_name: str, recovery_holdoff_seconds: Optional[float] = None) -> dict[str, Any]: """Start or replace the watcher for the given bridge.""" normalized_holdoff = recovery_holdoff_seconds or settings.bridge_link_state_recovery_holdoff_seconds with self._lock: old_watcher = self._watchers.pop(bridge_name, None) if old_watcher is not None: old_watcher.stop() watcher = BridgeLinkStateWatcher(bridge_name, normalized_holdoff) watcher.start() with self._lock: self._watchers[bridge_name] = watcher return watcher.status() def disable(self, bridge_name: str) -> dict[str, Any]: """Stop the watcher for one bridge.""" with self._lock: watcher = self._watchers.pop(bridge_name, None) if watcher is None: return { "bridge": bridge_name, "active": False, "message": "watcher not enabled", } watcher.stop() status = watcher.status() status["active"] = False return status def get_status(self, bridge_name: str) -> Optional[dict[str, Any]]: """Return the current watcher status for one bridge, if present.""" with self._lock: watcher = self._watchers.get(bridge_name) return watcher.status() if watcher is not None else None def list_statuses(self) -> list[dict[str, Any]]: """Return the current status for all bridge watchers.""" with self._lock: watchers = list(self._watchers.values()) return [watcher.status() for watcher in sorted(watchers, key=lambda item: item.bridge_name)] def stop(self) -> None: """Stop all running watchers.""" with self._lock: watchers = list(self._watchers.values()) self._watchers.clear() for watcher in watchers: watcher.stop() bridge_link_state_manager = BridgeLinkStateManager()