fix: optimize bridge port handling by using a fixed snapshot to reduce sysfs reads

backend/src/network_sniffer.py

@@ -41,8 +41,10 @@ af_selector: Optional[selectors.BaseSelector] = None
 af_thread: Optional[threading.Thread] = None
 af_stop_event: Optional[threading.Event] = None
 
-# Cache of bridge -> ports
-bridge_ports_cache: Dict[str, List[str]] = {}
+# Snapshot of bridge -> ports taken once when sniffer starts
+fixed_bridge_ports: Dict[str, List[str]] = {}
+# Remember which bridge the sniffer is using (single active sniffer model)
+current_bridge: Optional[str] = None
 
 # ---------------------------------------------------------------------
 # Async loop used to schedule DB inserts from packet callback threads.
@@ -88,40 +90,41 @@ def check_interface_up(iface: str) -> bool:
         return False
 
 
-def get_bridge_ports(bridge: str) -> List[str]:
+def _read_bridge_ports_from_sysfs(bridge: str) -> List[str]:
     """
-    Read the bridge member interfaces from sysfs (/sys/class/net/<bridge>/brif/).
-    This function refreshes the cache each call (cache entry is updated),
-    which keeps behavior predictable for dynamic topologies.
-    If the bridge does not exist, an empty list is returned.
+    Read bridge member interfaces from sysfs. Internal helper that always reads.
     """
     base = f"/sys/class/net/{bridge}/brif/"
     if not os.path.isdir(base):
-        logger.error("Bridge '%s' does not exist", bridge)
         return []
 
     try:
         ports = [p for p in os.listdir(base) if check_interface_exists(p)]
-    except Exception as e:
-        logger.exception("Error listing bridge ports for %s: %s", bridge, e)
+    except Exception:
+        logger.exception("Error reading bridge ports for %s", bridge)
         ports = []
-
-    # update local cache and log discovered ports
-    bridge_ports_cache[bridge] = ports
-    logger.info("Bridge %s ports: %s", bridge, ports)
     return ports
 
 
+def get_bridge_ports_once(bridge: str) -> List[str]:
+    """
+    Convenience: read bridge ports (no caching here) used when preparing the fixed snapshot.
+    """
+    return _read_bridge_ports_from_sysfs(bridge)
+
+
 def determine_direction(pkt_iface: str, bridge: str):
     """
-    Determine ingress and egress information for a packet based on the
-    interface it was sniffed on and the bridge port list.
+    Determine ingress/egress using the fixed snapshot of bridge ports
+    taken when the sniffer started. This avoids reading sysfs for every
+    packet and prevents noisy logs when bridge ports are static.
 
-    Returns:
-      ingress: the interface where the packet was observed
-      egress: list of other bridge ports (possible egress ports)
+    If snapshot is missing (defensive), fall back to a direct sysfs read.
     """
-    ports = bridge_ports_cache.get(bridge) or get_bridge_ports(bridge)
+    ports = fixed_bridge_ports.get(bridge)
+    if ports is None:
+        # Defensive fallback (should not happen if start_afpacket_sniffer was used)
+        ports = _read_bridge_ports_from_sysfs(bridge)
     ingress = pkt_iface
     egress = [p for p in ports if p != pkt_iface]
     return ingress, egress
@@ -184,7 +187,7 @@ async def db_insert_packet(pkt_info: dict) -> None:
 
 
 # -------------------------------------------------------------------
-# Packet parsing using your existing parse_packet logic
+# Packet parsing callback used by scapy-like parser
 # -------------------------------------------------------------------
 def parse_packet(pkt, bridge: str) -> None:
     """
@@ -348,55 +351,34 @@ def _close_socket(ifname: str) -> None:
             pass
 
 
-def _ensure_sockets_for_bridge(bridge: str) -> None:
+def _ensure_sockets_for_bridge_from_snapshot(bridge: str) -> None:
     """
-    Ensure there is an AF_PACKET socket bound for every current bridge port.
-    We keep sockets for interfaces even if they are DOWN (they will start receiving when link comes up).
+    Create sockets for ports listed in the fixed snapshot.
+    Only creates sockets for ports that don't already have one.
     """
-    ports = get_bridge_ports(bridge)
+    ports = fixed_bridge_ports.get(bridge, [])
     for p in ports:
         if p in af_sockets:
             continue
         if not check_interface_exists(p):
+            logger.warning("Snapshot port %s does not exist in sysfs (skipping)", p)
             continue
         s = _create_af_packet_socket(p)
         if s:
             af_sockets[p] = s
-            # register later in selector by reader thread
-
-
-def _rebind_if_needed(ifname: str) -> None:
-    """
-    Try to re-create a socket for an interface if it is missing (e.g. after deletion).
-    """
-    if ifname in af_sockets:
-        return
-    if not check_interface_exists(ifname):
-        return
-    s = _create_af_packet_socket(ifname)
-    if s:
-        af_sockets[ifname] = s
-        if af_selector:
-            try:
-                af_selector.register(s, selectors.EVENT_READ, data=ifname)
-            except Exception:
-                # ignore duplicate registration / race
-                pass
 
 
 def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None:
     """
     Single reader thread that multiplexes all AF_PACKET sockets for the bridge
     using a selector. When data arrives, we parse into a Scapy packet and call parse_packet().
-
-    This design uses a single thread + selector rather than one thread per interface.
     """
     global af_selector
     logger.info("AF_PACKET reader starting for bridge %s", bridge)
     af_selector = selectors.DefaultSelector()
 
-    # ensure sockets exist for current ports
-    _ensure_sockets_for_bridge(bridge)
+    # Ensure sockets exist based on the fixed snapshot (created at start)
+    _ensure_sockets_for_bridge_from_snapshot(bridge)
 
     # register sockets we have
     for ifname, s in list(af_sockets.items()):
@@ -410,20 +392,17 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None:
 
     # main loop
     while not stop_event.is_set():
-        # refresh sockets for any new bridge ports (cheap)
+        # In this static snapshot mode we only occasionally try to register
+        # newly created sockets (e.g., interfaces that existed but socket creation failed earlier).
         try:
-            _ensure_sockets_for_bridge(bridge)
-            # register any new sockets with selector
             for ifname, s in list(af_sockets.items()):
                 try:
-                    # register only if not registered
                     if not any(k.fileobj is s for k in af_selector.get_map().values()):
                         af_selector.register(s, selectors.EVENT_READ, data=ifname)
                 except Exception:
-                    # ignore races
                     pass
         except Exception:
-            logger.exception("Error ensuring sockets")
+            logger.exception("Error ensuring selector registrations")
 
         # wait for events with short timeout to remain responsive
         try:
@@ -434,30 +413,27 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None:
             continue
 
         if not events:
-            # no events; loop will re-ensure sockets again
+            # no events; loop will re-check registrations
             continue
 
         for key, mask in events:
             sock: socket.socket = key.fileobj
             ifname: str = key.data
             try:
-                # read raw frame
-                # using a single large buffer; AF_PACKET will give full frame
                 raw = sock.recv(65536)
                 if not raw:
                     continue
             except BlockingIOError:
                 continue
             except OSError as e:
-                # handle interface removal (ENODEV) or other errors: close socket and attempt rebind later
+                # handle interface removal (ENODEV) or other errors: close socket and continue
                 if e.errno in (errno.ENODEV, errno.ENETDOWN, errno.EBADF):
-                    logger.warning("Socket error on %s: %s — closing socket and will attempt to rebind later", ifname, e)
+                    logger.warning("Socket error on %s: %s — closing socket", ifname, e)
                     try:
                         af_selector.unregister(sock)
                     except Exception:
                         pass
                     _close_socket(ifname)
-                    # schedule rebind attempt on next loop iteration
                     continue
                 else:
                     logger.exception("Recv error on %s: %s", ifname, e)
@@ -502,15 +478,19 @@ def afpacket_reader_loop(bridge: str, stop_event: threading.Event) -> None:
 def start_afpacket_sniffer(bridge: str) -> None:
     """
     Start the optimized AF_PACKET sniffer for the provided bridge.
-    This will bind sockets to all current bridge ports and start a single reader thread.
+    This reads bridge ports once (snapshot) and starts a single reader thread.
     """
-    global af_thread, af_stop_event
+    global af_thread, af_stop_event, current_bridge, fixed_bridge_ports
     if af_thread and af_thread.is_alive():
         logger.info("AF_PACKET sniffer already running")
         return
 
-    # ensure we have initial ports
-    get_bridge_ports(bridge)
+    # Warm and freeze the bridge ports snapshot. We read sysfs once here.
+    ports = get_bridge_ports_once(bridge)
+    fixed_bridge_ports[bridge] = ports
+    current_bridge = bridge
+    logger.info("Fixed bridge ports for %s: %s", bridge, ports)
+
     af_stop_event = threading.Event()
     af_thread = threading.Thread(target=afpacket_reader_loop, args=(bridge, af_stop_event), daemon=True)
     af_thread.start()
@@ -519,9 +499,9 @@ def start_afpacket_sniffer(bridge: str) -> None:
 
 def stop_afpacket_sniffer() -> None:
     """
-    Stop the AF_PACKET sniffer thread and close sockets.
+    Stop the AF_PACKET sniffer thread and close sockets. Clear the fixed snapshot.
     """
-    global af_thread, af_stop_event
+    global af_thread, af_stop_event, current_bridge, fixed_bridge_ports
     if not af_thread:
         return
     if af_stop_event:
@@ -529,6 +509,12 @@ def stop_afpacket_sniffer() -> None:
     af_thread.join(timeout=2)
     af_thread = None
     af_stop_event = None
+
+    # clear fixed snapshot(s)
+    if current_bridge:
+        fixed_bridge_ports.pop(current_bridge, None)
+    current_bridge = None
+
     logger.info("AF_PACKET sniffer stopped")