diff --git a/.gitignore b/.gitignore index 28a535038..7677a0718 100644 --- a/.gitignore +++ b/.gitignore @@ -85,3 +85,5 @@ venv # Tiktoken cache files gns3server/agent/gns3_copilot/cache/tiktoken/ +gns3.log +/configs/ diff --git a/gns3server/api/routes/compute/dynamips_nodes.py b/gns3server/api/routes/compute/dynamips_nodes.py index 744dfa3ba..41314f8a1 100644 --- a/gns3server/api/routes/compute/dynamips_nodes.py +++ b/gns3server/api/routes/compute/dynamips_nodes.py @@ -64,7 +64,6 @@ async def create_router(project_id: UUID, node_data: schemas.DynamipsCreate) -> dynamips_manager = Dynamips.instance() platform = node_data.platform - print(node_data.chassis, platform in DEFAULT_CHASSIS) if not node_data.chassis and platform in DEFAULT_CHASSIS: chassis = DEFAULT_CHASSIS[platform] else: diff --git a/gns3server/api/routes/compute/projects.py b/gns3server/api/routes/compute/projects.py index 71245dfe7..7da980a08 100644 --- a/gns3server/api/routes/compute/projects.py +++ b/gns3server/api/routes/compute/projects.py @@ -21,6 +21,8 @@ API routes for projects. import os import shutil import urllib.parse +import inspect +import asyncio import logging @@ -34,6 +36,7 @@ from uuid import UUID from gns3server.compute.project_manager import ProjectManager from gns3server.compute.project import Project +from gns3server.compute.base_manager import BaseManager from gns3server.utils.path import is_safe_path from gns3server import schemas @@ -131,6 +134,190 @@ async def delete_compute_project(project: Project = Depends(dep_project)) -> Non ProjectManager.instance().remove_project(project.id) +async def _add_nio_binding(node, adapter_number, port_number, nio): + """ + Unified NIO-binding dispatch across node types. Each node type exposes a + different method signature, so centralise the fan-out here for the batch + endpoint. Dispatch keys off the manager class name (only dynamips/iou/qemu + carry a ``_NODE_TYPE`` attribute, so it can't be used universally). + """ + + manager_name = type(node.manager).__name__ + # Adapter-based nodes: docker / qemu / vmware / virtualbox take + # (adapter_number, nio); iou additionally takes port_number. + if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"): + await node.adapter_add_nio_binding(adapter_number, nio) + elif manager_name == "IOU": + await node.adapter_add_nio_binding(adapter_number, port_number, nio) + elif manager_name == "VPCS": + await node.port_add_nio_binding(port_number, nio) + elif manager_name == "Dynamips": + # Dynamips routers use slot_add_nio_binding(slot, port, nio); + # Dynamips switches/hubs use add_nio(nio, port_number). + if hasattr(node, "slot_add_nio_binding"): + await node.slot_add_nio_binding(adapter_number, port_number, nio) + else: + await node.add_nio(nio, port_number) + elif manager_name == "Builtin": + # ethernet_switch / ethernet_hub / cloud / nat: add_nio(nio, port_number) + await node.add_nio(nio, port_number) + else: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Batch NIO creation not supported for node type '{manager_name}'", + ) + + +def _get_existing_nio(node, adapter_number, port_number): + """ + Fetch the already-bound NIO for a port, preserving its UDP endpoints + (lport/rhost/rport) so a marker/filter update only changes markers/filters. + Dispatch keys off the manager class name, mirroring _add_nio_binding. + """ + + manager_name = type(node.manager).__name__ + if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"): + return node.get_nio(adapter_number) + elif manager_name == "IOU": + return node.get_nio(adapter_number, port_number) + elif manager_name in ("VPCS", "Builtin"): + return node.get_nio(port_number) + elif manager_name == "Dynamips": + # Dynamips routers expose NIOs via the slot/adapter; switches/hubs + # via get_nio(port). + if hasattr(node, "get_nio"): + import inspect as _inspect + if len(_inspect.signature(node.get_nio).parameters) >= 2: + return node.get_nio(adapter_number, port_number) + return node.get_nio(port_number) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Dynamips node '{node.name}' has no get_nio for batch update", + ) + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Batch NIO update not supported for node type '{manager_name}'", + ) + + +async def _update_nio_binding(node, adapter_number, port_number, nio): + """ + Re-apply a NIO binding (filters + markers) to a started node's uBridge. + Dispatch keys off the manager class name, mirroring _add_nio_binding. + """ + + manager_name = type(node.manager).__name__ + if manager_name in ("Docker", "Qemu", "VMware", "VirtualBox"): + await node.adapter_update_nio_binding(adapter_number, nio) + elif manager_name == "IOU": + await node.adapter_update_nio_binding(adapter_number, port_number, nio) + elif manager_name == "VPCS": + await node.port_update_nio_binding(port_number, nio) + elif manager_name == "Dynamips": + if hasattr(node, "slot_update_nio_binding"): + await node.slot_update_nio_binding(adapter_number, port_number, nio) + else: + await node.update_nio(port_number, nio) + elif manager_name == "Builtin": + await node.update_nio(port_number, nio) + else: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Batch NIO update not supported for node type '{manager_name}'", + ) + + +@router.post( + "/projects/{project_id}/nios/batch", + status_code=status.HTTP_201_CREATED, +) +async def create_batch_nios( + project_id: UUID, + batch: schemas.BatchNIOCreate, + project: Project = Depends(dep_project), +) -> dict: + """ + Create many NIO bindings across nodes in a single request. + + Used by the controller during project open to avoid one HTTP round-trip per + NIO. Each entry resolves its node via the project, builds the NIO through + the node's manager, and binds it. Nodes that are not started perform the + binding in memory; started nodes additionally wire uBridge. + """ + + # Group entries by node so different nodes' uBridge processes are wired + # in parallel (each node has its own AF_UNIX socket). Within a node + # entries are serial to respect the per-node uBridge command lock. + # This is what makes builtin L2 nodes (ethernet_switch/hub/cloud/nat) + # start their uBridge concurrently during project open instead of one + # at a time (~0.5s each for fork + socket connect). + per_node = {} + for entry in batch.nios: + per_node.setdefault(entry.node_id, []).append(entry) + + async def _create_one_node(node_id, entries): + node = project.get_node(node_id) + # Dynamips.create_nio(self, node, nio_settings) is async and takes + # an extra positional 'node'; other managers' create_nio(nio_settings) + # is synchronous. Detect once per node via bound-method parameter + # count (standard == 1, Dynamips == 2) and await the async variant. + sig = inspect.signature(node.manager.create_nio) + dynamips_style = len(sig.parameters) >= 2 + for entry in entries: + nio_settings = jsonable_encoder(entry.nio, exclude_unset=True) + if dynamips_style: + nio = await node.manager.create_nio(node, nio_settings) + else: + nio = node.manager.create_nio(nio_settings) + await _add_nio_binding(node, entry.adapter_number, entry.port_number, nio) + + await asyncio.gather( + *[_create_one_node(nid, ents) for nid, ents in per_node.items()] + ) + return {"added": len(batch.nios)} + + +@router.put( + "/projects/{project_id}/nios/batch", + status_code=status.HTTP_200_OK, +) +async def update_batch_nios( + project_id: UUID, + batch: schemas.BatchNIOCreate, + project: Project = Depends(dep_project), +) -> dict: + """ + Update many NIO bindings (filters + markers) across nodes in a single + request, re-applying them to uBridge on started nodes. + + Used by the controller when a project-level marker definition changes and + must fan out to every affected link — replacing one PUT /nio round-trip per + link end with one round-trip per compute. Each entry fetches the already- + bound NIO (preserving its UDP endpoints), overlays the new markers/filters, + and re-binds it. + """ + + # Group entries by node so that different nodes' uBridge processes are + # updated in parallel (each node has its own AF_UNIX socket). Within a + # node entries are serial to respect the per-node uBridge command lock. + per_node = {} + for entry in batch.nios: + per_node.setdefault(entry.node_id, []).append(entry) + + async def _update_one_node(node_id, entries): + node = project.get_node(node_id) + for e in entries: + nio = _get_existing_nio(node, e.adapter_number, e.port_number) + nio.filters = e.nio.filters or {} + nio.markers = e.nio.markers or {} + await _update_nio_binding(node, e.adapter_number, e.port_number, nio) + + await asyncio.gather( + *[_update_one_node(nid, ents) for nid, ents in per_node.items()] + ) + return {"updated": len(batch.nios)} + + @router.get("/projects/{project_id}/files", response_model=List[schemas.ProjectFile]) async def get_compute_project_files(project: Project = Depends(dep_project)) -> List[schemas.ProjectFile]: """ diff --git a/gns3server/api/routes/compute/qemu_nodes.py b/gns3server/api/routes/compute/qemu_nodes.py index 51f8f60d8..efd6f7d74 100644 --- a/gns3server/api/routes/compute/qemu_nodes.py +++ b/gns3server/api/routes/compute/qemu_nodes.py @@ -83,7 +83,7 @@ async def create_qemu_node(project_id: UUID, node_data: schemas.QemuCreate) -> s for disk_index, drive in enumerate(drives): disk_image_backing_file = node_data.get(f"hd{drive}_disk_image_backing_file") if disk_image_backing_file: - log.info(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}") + log.debug(f"Updating disk image for drive {drive} with backing file {disk_image_backing_file}") node_data[f"hd{drive}_disk_image"] = disk_image_backing_file for name, value in node_data.items(): diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py index f9c838a34..d10de587a 100644 --- a/gns3server/api/routes/controller/projects.py +++ b/gns3server/api/routes/controller/projects.py @@ -485,6 +485,38 @@ async def project_ws_notifications( await project.close() +@router.websocket("/{project_id}/notifications/markers/ws") +async def project_marker_ws_notifications( + project_id: UUID, + websocket: WebSocket, + current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")) +) -> None: + """ + Receive marker notifications (e.g. marker.match) for a project on a + dedicated WebSocket, separate from the main project stream so high-frequency + marker.matches do not block topology events (node.*/link.*). + + Required privilege: Project.Audit + """ + + if current_user is None: + return + + controller = Controller.instance() + project = controller.get_project(str(project_id)) + + log.info(f"New client has connected to the marker notification stream for project ID '{project.id}' (WebSocket method)") + try: + with controller.notification.project_marker_queue(project.id) as queue: + while True: + notification = await queue.get_json(5) + await websocket.send_text(notification) + except (ConnectionClosed, WebSocketDisconnect): + log.info(f"Client has disconnected from the marker notification stream for project ID '{project.id}' (WebSocket method)") + except WebSocketException as e: + log.warning(f"Error while sending marker event to WebSocket client: {e}") + + @router.get("/{project_id}/export", dependencies=[Depends(has_privilege("Project.Audit"))]) async def export_project( project: Project = Depends(dep_project), diff --git a/gns3server/compute/base_node.py b/gns3server/compute/base_node.py index fc048892a..e2431ffa0 100644 --- a/gns3server/compute/base_node.py +++ b/gns3server/compute/base_node.py @@ -103,6 +103,10 @@ class BaseNode: # marker filter name -> uBridge bridge_name (recorded at apply time so # _ubridge_set_marker_filter_state can toggle on/off without an NIO rebuild). self._marker_filter_bridges = {} + # Parallel store of the installed marker spec (bpf/tag/direction/enabled/...) + # keyed by (name, link_id) so _ubridge_apply_markers can reconcile: detect + # deletions and field changes instead of being add-only. + self._marker_specs = {} if self._console is not None: # use a previously allocated console port @@ -325,7 +329,7 @@ class BaseNode: Creates the node. """ - log.info("{module}: {name} [{id}] created".format(module=self.manager.module_name, name=self.name, id=self.id)) + log.debug("{module}: {name} [{id}] created".format(module=self.manager.module_name, name=self.name, id=self.id)) async def delete(self): """ @@ -374,7 +378,7 @@ class BaseNode: if self._closed: return False - log.info( + log.debug( "{module}: '{name}' [{id}]: is closing".format(module=self.manager.module_name, name=self.name, id=self.id) ) @@ -934,7 +938,7 @@ class BaseNode: self._ubridge_hypervisor = Hypervisor( self._project, self.ubridge_path, self.working_dir, transport, server_host, self.id ) - log.info(f"Starting new uBridge hypervisor at {self._ubridge_hypervisor.endpoint}") + log.debug(f"Starting new uBridge hypervisor at {self._ubridge_hypervisor.endpoint}") await self._ubridge_hypervisor.start() if self._ubridge_hypervisor: log.info( @@ -987,13 +991,14 @@ class BaseNode: """ if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running(): - log.info(f"Stopping uBridge hypervisor at {self._ubridge_hypervisor.endpoint}") + log.debug(f"Stopping uBridge hypervisor at {self._ubridge_hypervisor.endpoint}") await self._ubridge_hypervisor.stop() self._ubridge_hypervisor = None # uBridge is gone, so every marker filter (and its in-bridge state) is # gone too — clear the map so the next apply re-installs them all rather # than skipping them as "already installed". self._marker_filter_bridges.clear() + self._marker_specs.clear() async def add_ubridge_udp_connection(self, bridge_name, source_nio, destination_nio): """ @@ -1176,6 +1181,7 @@ class BaseNode: if nio is not None and getattr(nio, "markers", None): nio.markers.pop(name, None) bridge_name = self._marker_filter_bridges.pop((name, link_id), None) + self._marker_specs.pop((name, link_id), None) if bridge_name is not None: await self._ubridge_delete_marker_filter(bridge_name, name) try: @@ -1223,34 +1229,77 @@ class BaseNode: async def _ubridge_apply_markers(self, bridge_name, nio): """ - Install the traffic-insight markers carried by *nio* onto bridge - *bridge_name* that aren't already there. uBridge's ``reset_packet_filters`` - preserves mark filters (contract), so on an NIO update we add only the new - ones — re-adding an existing marker would either duplicate it or - close/reopen its pcap. Called from ``add_ubridge_udp_connection`` (fresh - bridge, empty map → installs all) and ``update_ubridge_udp_connection`` - (incremental). + Reconcile the traffic-insight markers carried by *nio* onto bridge + *bridge_name* with what is already installed there. + + uBridge's ``reset_packet_filters`` preserves mark filters (contract), so + a plain re-add would duplicate them; instead this diffs the desired + ``nio.markers`` against the installed ``_marker_specs``: + + * installed but no longer desired → delete filter + unlink pcap + * desired with changed bpf/tag/direction/data_link_type → rebuild + (delete + add; the marker's own pcap reopens for the new BPF) + * desired with only ``enabled`` changed → instant on/off toggle + (sibling and own pcap stay open) + * desired and unchanged → skip + * desired and new → add + + Called from ``add_ubridge_udp_connection`` (fresh bridge, empty maps → + installs all) and ``update_ubridge_udp_connection`` / the batch NIO + update path (incremental reconcile). """ from gns3server.compute.marker.marker_manager import MarkerManager markers = nio.markers if hasattr(nio, 'markers') else {} - if not markers: - return - manager = MarkerManager.instance() markers_dir = self.project.markers_working_directory() - for name, spec in markers.items(): - link_id = spec.get("link_id", "") - # Incremental: skip markers already on this bridge. uBridge keeps mark - # filters across reset_packet_filters, so re-adding would duplicate (or - # reopen the pcap). A fresh bridge has an empty map → installs all. - if (name, link_id) in self._marker_filter_bridges: + desired = {(name, spec.get("link_id", "")): spec for name, spec in markers.items()} + + # 1. Remove installed markers that are no longer desired (marker/def delete). + # Scope to THIS bridge: the map is node-wide and also holds markers + # installed on this node's other links/NIOs. Without this guard, + # reconciling one NIO would delete every other link's markers + pcaps + # (desired only carries the current NIO's markers) — a regression. + for key in list(self._marker_filter_bridges): + if self._marker_filter_bridges[key] != bridge_name: continue + if key not in desired: + mname, link_id = key + installed_bridge = self._marker_filter_bridges.pop(key) + self._marker_specs.pop(key, None) + await self._ubridge_delete_marker_filter(installed_bridge, mname) + try: + os.remove(os.path.join(markers_dir, f"{self._id}_{link_id}_{mname}.pcap")) + except FileNotFoundError: + pass + except OSError as e: + log.warning("Could not remove marker pcap for '%s' on link %s: %s", mname, link_id, e) + manager.unregister(self._id, mname) + + # 2. Add newly-desired markers; rebuild ones whose filter fields changed. + rebuild_fields = ("bpf", "tag", "direction", "data_link_type") + for (name, link_id), spec in desired.items(): bpf = spec.get("bpf", "") tag = spec.get("tag") - pcap_path = os.path.join( - markers_dir, f"{self._id}_{link_id}_{name}.pcap" - ) + enabled = spec.get("enabled", True) + if (name, link_id) in self._marker_filter_bridges: + installed_spec = self._marker_specs.get((name, link_id)) + if installed_spec is None: + # Installed but no recorded spec (legacy / pre-reconcile state): + # cannot diff, skip to avoid a duplicate add. + continue + if any(installed_spec.get(f) != spec.get(f) for f in rebuild_fields): + # A filter field changed → rebuild (delete + re-add). + installed_bridge = self._marker_filter_bridges.get((name, link_id)) + await self._ubridge_delete_marker_filter(installed_bridge, name) + elif installed_spec.get("enabled", True) != enabled: + # Only the on/off state changed → instant toggle, pcap preserved. + await self._ubridge_set_marker_filter_state(name, enabled) + self._marker_specs[(name, link_id)] = spec + continue + else: + continue # unchanged + pcap_path = os.path.join(markers_dir, f"{self._id}_{link_id}_{name}.pcap") try: await self._ubridge_add_marker_filter(bridge_name, name, bpf, pcap_path, tag, link_id, direction=spec.get("direction"), @@ -1268,22 +1317,19 @@ class BaseNode: # A disabled marker is installed but turned off (a paused tap), not # dropped — so the UI can flip it back on instantly with # enable_packet_filter, no NIO rebuild (ubridge contract §3.2). - if not spec.get("enabled", True): + if not enabled: try: await self._ubridge_send(f"bridge enable_packet_filter {bridge_name} {name} off") except UbridgeError as e: # Old ubridge without enable_packet_filter: leave it installed # (on) rather than fail the whole link/marker apply. log.warning(f"Could not turn marker '{name}' off on {bridge_name}: {e}") - manager.register( - str(self.project.id), self._id, name, link_id, tag - ) + manager.register(str(self.project.id), self._id, name, link_id, tag) # Remember which bridge hosts this filter so an instant on/off toggle - # (no NIO rebuild) can resolve it by name alone. - # keyed (name, link_id) so a node that hosts markers for several links - # (e.g. IOU with one IOL-BRIDGE and many bays/units) records each - # copy independently — toggle below iterates all matching entries. + # (no NIO rebuild) can resolve it by name alone, and keep the spec so + # the next reconcile can detect changes. self._marker_filter_bridges[name, link_id] = bridge_name + self._marker_specs[name, link_id] = spec async def _ubridge_set_marker_filter_state(self, name, enabled): """ diff --git a/gns3server/compute/builtin/nodes/cloud.py b/gns3server/compute/builtin/nodes/cloud.py index 62cfb45ba..771a290c2 100644 --- a/gns3server/compute/builtin/nodes/cloud.py +++ b/gns3server/compute/builtin/nodes/cloud.py @@ -228,7 +228,7 @@ class Cloud(BaseNode): """ await self.start() - log.info(f'Cloud "{self._name}" [{self._id}] has been created') + log.debug(f'Cloud "{self._name}" [{self._id}] has been created') async def start(self): """ @@ -261,7 +261,7 @@ class Cloud(BaseNode): self.manager.port_manager.release_udp_port(nio.lport, self._project) await self._stop_ubridge() - log.info(f'Cloud "{self._name}" [{self._id}] has been closed') + log.debug(f'Cloud "{self._name}" [{self._id}] has been closed') async def _is_wifi_adapter_osx(self, adapter_name): """ @@ -429,7 +429,7 @@ class Cloud(BaseNode): if port_number in self._nios: raise NodeError(f"Port {port_number} isn't free") - log.info( + log.debug( 'Cloud "{name}" [{id}]: NIO {nio} bound to port {port}'.format( name=self._name, id=self._id, nio=nio, port=port_number ) @@ -485,7 +485,7 @@ class Cloud(BaseNode): if isinstance(nio, NIOUDP): self.manager.port_manager.release_udp_port(nio.lport, self._project) - log.info( + log.debug( 'Cloud "{name}" [{id}]: NIO {nio} removed from port {port}'.format( name=self._name, id=self._id, nio=nio, port=port_number ) @@ -535,7 +535,7 @@ class Cloud(BaseNode): await self._ubridge_send( 'bridge start_capture {name} "{output_file}"'.format(name=bridge_name, output_file=output_file) ) - log.info( + log.debug( "Cloud '{name}' [{id}]: starting packet capture on port {port_number}".format( name=self.name, id=self.id, port_number=port_number ) @@ -555,7 +555,7 @@ class Cloud(BaseNode): bridge_name = f"{self._id}-{port_number}" await self._ubridge_send(f"bridge stop_capture {bridge_name}") - log.info( + log.debug( "Cloud'{name}' [{id}]: stopping packet capture on port {port_number}".format( name=self.name, id=self.id, port_number=port_number ) diff --git a/gns3server/compute/builtin/nodes/ethernet_hub.py b/gns3server/compute/builtin/nodes/ethernet_hub.py index fc601ff01..4ef0e27e9 100644 --- a/gns3server/compute/builtin/nodes/ethernet_hub.py +++ b/gns3server/compute/builtin/nodes/ethernet_hub.py @@ -53,7 +53,7 @@ class EthernetHub(BaseNode): """ super().create() - log.info(f'Ethernet hub "{self._name}" [{self._id}] has been created') + log.debug(f'Ethernet hub "{self._name}" [{self._id}] has been created') async def delete(self): """ diff --git a/gns3server/compute/builtin/nodes/ethernet_switch.py b/gns3server/compute/builtin/nodes/ethernet_switch.py index a46366dd0..5c66cbad5 100644 --- a/gns3server/compute/builtin/nodes/ethernet_switch.py +++ b/gns3server/compute/builtin/nodes/ethernet_switch.py @@ -183,7 +183,7 @@ class EthernetSwitch(BaseNode): """ await self.start() - log.info(f'Ethernet switch "{self._name}" [{self._id}] has been created') + log.debug(f'Ethernet switch "{self._name}" [{self._id}] has been created') async def start(self): """ @@ -290,7 +290,7 @@ class EthernetSwitch(BaseNode): self._started = False await self._stop_ubridge() - log.info(f'Ethernet switch "{self._name}" [{self._id}] has been closed') + log.debug(f'Ethernet switch "{self._name}" [{self._id}] has been closed') return True # ------------------------------------------------------------------ # @@ -310,7 +310,7 @@ class EthernetSwitch(BaseNode): if not isinstance(nio, NIOUDP): raise NodeError("Ethernet switch ports only support UDP NIOs") - log.info( + log.debug( 'Ethernet switch "{name}" [{id}]: NIO {nio} bound to port {port}'.format( name=self._name, id=self._id, nio=nio, port=port_number ) @@ -397,7 +397,7 @@ class EthernetSwitch(BaseNode): if isinstance(nio, NIOUDP): self.manager.port_manager.release_udp_port(nio.lport, self._project) - log.info( + log.debug( 'Ethernet switch "{name}" [{id}]: NIO {nio} removed from port {port}'.format( name=self._name, id=self._id, nio=nio, port=port_number ) @@ -512,7 +512,7 @@ class EthernetSwitch(BaseNode): if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running(): ubridge_bridge = self._ubridge_bridge_name(port_number) await self._ubridge_send(f'bridge start_capture {ubridge_bridge} "{output_file}"') - log.info( + log.debug( 'Ethernet switch "{name}" [{id}]: starting packet capture on port {port}'.format( name=self.name, id=self.id, port=port_number ) @@ -532,7 +532,7 @@ class EthernetSwitch(BaseNode): if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running(): ubridge_bridge = self._ubridge_bridge_name(port_number) await self._ubridge_send(f"bridge stop_capture {ubridge_bridge}") - log.info( + log.debug( 'Ethernet switch "{name}" [{id}]: stopping packet capture on port {port}'.format( name=self.name, id=self.id, port=port_number ) diff --git a/gns3server/compute/builtin/nodes/nat.py b/gns3server/compute/builtin/nodes/nat.py index 31b96b9fe..a3b9b69d4 100644 --- a/gns3server/compute/builtin/nodes/nat.py +++ b/gns3server/compute/builtin/nodes/nat.py @@ -69,7 +69,7 @@ class Nat(Cloud): ) interface = interfaces[0] # take the first available interface containing the vmnet8 name - log.info(f"NAT node '{name}' configured to use NAT interface '{interface}'") + log.debug(f"NAT node '{name}' configured to use NAT interface '{interface}'") ports = [{"name": "nat0", "type": "ethernet", "interface": interface, "port_number": 0}] super().__init__(name, node_id, project, manager, ports=ports) diff --git a/gns3server/compute/docker/docker_vm.py b/gns3server/compute/docker/docker_vm.py index dafe3e8ee..ef4210668 100644 --- a/gns3server/compute/docker/docker_vm.py +++ b/gns3server/compute/docker/docker_vm.py @@ -228,7 +228,7 @@ class DockerVM(BaseNode): else: self._mac_address = mac_address - log.info('Docker container "{name}" [{id}]: MAC address changed to {mac_addr}'.format( + log.debug('Docker container "{name}" [{id}]: MAC address changed to {mac_addr}'.format( name=self._name, id=self._id, mac_addr=self._mac_address) @@ -348,7 +348,7 @@ class DockerVM(BaseNode): except OSError as e: raise DockerError(f"Cannot access resources: {e}") - log.info(f'Mount resources from "{resources_path}"') + log.debug(f'Mount resources from "{resources_path}"') binds = [{ "Type": "bind", "Source": resources_path, @@ -582,11 +582,11 @@ class DockerVM(BaseNode): log.error(f"Failed to clean up conflicting container '{self.docker_name}': {e}") raise self._cid = result["Id"] - log.info(f"Docker container '{self._name}' [{self._id}] created") + log.debug(f"Docker container '{self._name}' [{self._id}] created") if self._cpus > 0: - log.info(f"CPU limit set to {self._cpus} CPUs") + log.debug(f"CPU limit set to {self._cpus} CPUs") if self._memory > 0: - log.info(f"Memory limit set to {self._memory} MB") + log.debug(f"Memory limit set to {self._memory} MB") return True def _format_env(self, variables, env): @@ -704,7 +704,7 @@ class DockerVM(BaseNode): self._permissions_fixed = False self.status = "started" - log.info( + log.debug( "Docker container '{name}' [{image}] started listen for {console_type} on {console}".format( name=self._name, image=self._image, console=self.console, console_type=self.console_type ) @@ -750,7 +750,7 @@ class DockerVM(BaseNode): """ state = await self._get_container_state() - log.info(f"Docker container '{self._name}' fix ownership, state = {state}") + log.debug(f"Docker container '{self._name}' fix ownership, state = {state}") if state == "stopped" or state == "exited": # We need to restart it to fix permissions await self.manager.query("POST", f"containers/{self._cid}/start") @@ -1010,7 +1010,7 @@ class DockerVM(BaseNode): """ await self.manager.query("POST", f"containers/{self._cid}/restart") - log.info("Docker container '{name}' [{image}] restarted".format(name=self._name, image=self._image)) + log.debug("Docker container '{name}' [{image}] restarted".format(name=self._name, image=self._image)) async def _clean_servers(self): """ @@ -1049,11 +1049,14 @@ class DockerVM(BaseNode): state = await self._get_container_state() if state != "stopped" and state != "exited": - # t=5 number of seconds to wait before killing the container + # SIGKILL immediately. GNS3 has already persisted container state + # (permissions via _fix_permissions, /gns3volumes) before this + # point, and the business process (often an interactive shell) + # ignores SIGTERM — so a stop grace period buys nothing but latency. try: - await self.manager.query("POST", f"containers/{self._cid}/stop", params={"t": 5}) - log.info(f"Docker container '{self._name}' [{self._image}] stopped") - except DockerHttp304Error: + await self.manager.query("POST", f"containers/{self._cid}/kill") + log.debug(f"Docker container '{self._name}' [{self._image}] stopped") + except DockerHttp409Error: # Container is already stopped pass # Ignore runtime error because when closing the server @@ -1069,7 +1072,7 @@ class DockerVM(BaseNode): await self.manager.query("POST", f"containers/{self._cid}/pause") self.status = "suspended" - log.info(f"Docker container '{self._name}' [{self._image}] paused") + log.debug(f"Docker container '{self._name}' [{self._image}] paused") async def unpause(self): """ @@ -1078,7 +1081,7 @@ class DockerVM(BaseNode): await self.manager.query("POST", f"containers/{self._cid}/unpause") self.status = "started" - log.info(f"Docker container '{self._name}' [{self._image}] unpaused") + log.debug(f"Docker container '{self._name}' [{self._image}] unpaused") async def close(self): """ @@ -1130,7 +1133,7 @@ class DockerVM(BaseNode): # Container deletion failed - log warning but don't block project close # The stale container will be cleaned up when the project is opened again log.warning(f"Failed to delete Docker container '{self.docker_name}': {e}") - log.info("Docker container '{name}' [{image}] removed".format(name=self._name, image=self._image)) + log.debug("Docker container '{name}' [{image}] removed".format(name=self._name, image=self._image)) if release_nio_udp_ports: for adapter in self._ethernet_adapters: @@ -1201,7 +1204,7 @@ class DockerVM(BaseNode): except UbridgeError as e: raise UbridgeNamespaceError(e) else: - log.info(f"Created adapter {adapter_number} with MAC address {mac_address} in namespace {self._namespace}") + log.debug(f"Created adapter {adapter_number} with MAC address {mac_address} in namespace {self._namespace}") if nio: await self._connect_nio(adapter_number, nio) @@ -1219,7 +1222,6 @@ class DockerVM(BaseNode): bridge_name=bridge_name, lport=nio.lport, rhost=nio.rhost, rport=nio.rport ) ) - if nio.capturing: await self._ubridge_send( 'bridge start_capture {bridge_name} "{pcap_file}"'.format( @@ -1251,7 +1253,7 @@ class DockerVM(BaseNode): await self._connect_nio(adapter_number, nio) adapter.add_nio(0, nio) - log.info( + log.debug( "Docker container '{name}' [{id}]: {nio} added to adapter {adapter_number}".format( name=self.name, id=self._id, nio=nio, adapter_number=adapter_number ) @@ -1301,7 +1303,7 @@ class DockerVM(BaseNode): adapter.remove_nio(0) - log.info( + log.debug( "Docker VM '{name}' [{id}]: {nio} removed from adapter {adapter_number}".format( name=self.name, id=self.id, nio=adapter.host_ifc, adapter_number=adapter_number ) @@ -1358,7 +1360,7 @@ class DockerVM(BaseNode): for adapter_number in range(0, adapters): self._ethernet_adapters.append(EthernetAdapter()) - log.info( + log.debug( 'Docker container "{name}" [{id}]: number of Ethernet adapters changed to {adapters}'.format( name=self._name, id=self._id, adapters=adapters ) @@ -1415,7 +1417,7 @@ class DockerVM(BaseNode): if self.status == "started" and self.ubridge: await self._start_ubridge_capture(adapter_number, output_file) - log.info( + log.debug( "Docker VM '{name}' [{id}]: starting packet capture on adapter {adapter_number}".format( name=self.name, id=self.id, adapter_number=adapter_number ) @@ -1435,7 +1437,7 @@ class DockerVM(BaseNode): if self.status == "started" and self.ubridge: await self._stop_ubridge_capture(adapter_number) - log.info( + log.debug( "Docker VM '{name}' [{id}]: stopping packet capture on adapter {adapter_number}".format( name=self.name, id=self.id, adapter_number=adapter_number ) diff --git a/gns3server/compute/dynamips/__init__.py b/gns3server/compute/dynamips/__init__.py index eeb4ed91f..134ce83e0 100644 --- a/gns3server/compute/dynamips/__init__.py +++ b/gns3server/compute/dynamips/__init__.py @@ -333,9 +333,9 @@ class Dynamips(BaseManager): port_manager = PortManager.instance() hypervisor = Hypervisor(self._dynamips_path, working_dir, server_host, port, port_manager.console_host, bind_console_host) - log.info(f"Creating new hypervisor {hypervisor.host}:{hypervisor.port} with working directory {working_dir}") + log.debug(f"Creating new hypervisor {hypervisor.host}:{hypervisor.port} with working directory {working_dir}") await hypervisor.start() - log.info(f"Hypervisor {hypervisor.host}:{hypervisor.port} has successfully started") + log.debug(f"Hypervisor {hypervisor.host}:{hypervisor.port} has successfully started") await hypervisor.connect() return hypervisor @@ -555,7 +555,7 @@ class Dynamips(BaseManager): :returns: relative path to the created config file """ - log.info(f"Creating config file {path}") + log.debug(f"Creating config file {path}") config_dir = os.path.dirname(path) try: os.makedirs(config_dir, exist_ok=True) diff --git a/gns3server/compute/dynamips/dynamips_hypervisor.py b/gns3server/compute/dynamips/dynamips_hypervisor.py index 187876eb2..997cfce60 100644 --- a/gns3server/compute/dynamips/dynamips_hypervisor.py +++ b/gns3server/compute/dynamips/dynamips_hypervisor.py @@ -90,12 +90,12 @@ class DynamipsHypervisor: if not connection_success: raise DynamipsError(f"Couldn't connect to hypervisor on {host}:{self._port} :{last_exception}") else: - log.info(f"Connected to Dynamips hypervisor on {host}:{self._port} after {time.time() - begin:.4f} seconds") + log.debug(f"Connected to Dynamips hypervisor on {host}:{self._port} after {time.time() - begin:.4f} seconds") try: version = await self.send("hypervisor version") self._version = version[0].split("-", 1)[0] - log.info("Dynamips version {} detected".format(self._version)) + log.debug("Dynamips version {} detected".format(self._version)) except IndexError: log.warning("Dynamips version could not be detected") self._version = "Unknown" diff --git a/gns3server/compute/dynamips/hypervisor.py b/gns3server/compute/dynamips/hypervisor.py index 517605f37..30d9a268b 100644 --- a/gns3server/compute/dynamips/hypervisor.py +++ b/gns3server/compute/dynamips/hypervisor.py @@ -120,14 +120,14 @@ class Hypervisor(DynamipsHypervisor): self._command = self._build_command() env = os.environ.copy() try: - log.info(f"Starting Dynamips: {self._command}") + log.debug(f"Starting Dynamips: {self._command}") self._stdout_file = os.path.join(self.working_dir, f"dynamips_i{self._id}_stdout.txt") - log.info(f"Dynamips process logging to {self._stdout_file}") + log.debug(f"Dynamips process logging to {self._stdout_file}") with open(self._stdout_file, "w", encoding="utf-8") as fd: self._process = await asyncio.create_subprocess_exec( *self._command, stdout=fd, stderr=subprocess.STDOUT, cwd=self._working_dir, env=env ) - log.info(f"Dynamips process started PID={self._process.pid}") + log.debug(f"Dynamips process started PID={self._process.pid}") self._started = True except (OSError, subprocess.SubprocessError) as e: log.error(f"Could not start Dynamips: {e}") @@ -139,7 +139,7 @@ class Hypervisor(DynamipsHypervisor): """ if self.is_running(): - log.info(f"Stopping Dynamips process PID={self._process.pid}") + log.debug(f"Stopping Dynamips process PID={self._process.pid}") await DynamipsHypervisor.stop(self) # give some time for the hypervisor to properly stop. # time to delete UNIX NIOs for instance. diff --git a/gns3server/compute/dynamips/nios/nio_udp.py b/gns3server/compute/dynamips/nios/nio_udp.py index d849a37bf..6a7590cbd 100644 --- a/gns3server/compute/dynamips/nios/nio_udp.py +++ b/gns3server/compute/dynamips/nios/nio_udp.py @@ -73,7 +73,7 @@ class NIOUDP(NIO): ) ) - log.info( + log.debug( "NIO UDP {name} created with lport={lport}, rhost={rhost}, rport={rport}".format( name=self._name, lport=self._lport, rhost=self._rhost, rport=self._rport ) diff --git a/gns3server/compute/dynamips/nodes/router.py b/gns3server/compute/dynamips/nodes/router.py index e6e750f83..6a09cae5f 100644 --- a/gns3server/compute/dynamips/nodes/router.py +++ b/gns3server/compute/dynamips/nodes/router.py @@ -126,7 +126,7 @@ class Router(BaseNode): self._dynamips_id = dynamips_id manager.take_dynamips_id(project.id, dynamips_id) else: - log.info("Creating a new ghost IOS instance") + log.debug("Creating a new ghost IOS instance") if self._console: # Ghost VMs do not need a console port. self.console = None @@ -243,7 +243,7 @@ class Router(BaseNode): if not self._ghost_flag: - log.info( + log.debug( 'Router {platform} "{name}" [{id}] has been created'.format( name=self._name, platform=self._platform, id=self._id ) @@ -328,7 +328,7 @@ class Router(BaseNode): ) await self._hypervisor.send(f'vm start "{self._name}"') self.status = "started" - log.info(f'router "{self._name}" [{self._id}] has been started') + log.debug(f'router "{self._name}" [{self._id}] has been started') self._memory_watcher = FileWatcher(self._memory_files(), self._memory_changed, strategy="hash", delay=30) monitor_process(self._hypervisor.process, self._termination_callback) @@ -348,7 +348,7 @@ class Router(BaseNode): if self.status == "started": self.status = "stopped" - log.info("Dynamips hypervisor process has stopped, return code: %d", returncode) + log.debug("Dynamips hypervisor process has stopped, return code: %d", returncode) if returncode != 0: self.project.emit( "log.error", @@ -369,7 +369,7 @@ class Router(BaseNode): except DynamipsError as e: log.warning(f"Could not stop {self._name}: {e}") self.status = "stopped" - log.info(f'Router "{self._name}" [{self._id}] has been stopped') + log.debug(f'Router "{self._name}" [{self._id}] has been stopped') if self._memory_watcher: self._memory_watcher.close() self._memory_watcher = None @@ -393,7 +393,7 @@ class Router(BaseNode): if status == "running": await self._hypervisor.send(f'vm suspend "{self._name}"') self.status = "suspended" - log.info(f'Router "{self._name}" [{self._id}] has been suspended') + log.debug(f'Router "{self._name}" [{self._id}] has been suspended') async def resume(self): """ @@ -404,7 +404,7 @@ class Router(BaseNode): if status == "suspended": await self._hypervisor.send(f'vm resume "{self._name}"') self.status = "started" - log.info(f'Router "{self._name}" [{self._id}] has been resumed') + log.debug(f'Router "{self._name}" [{self._id}] has been resumed') async def is_running(self): """ @@ -545,7 +545,7 @@ class Router(BaseNode): await self._hypervisor.send(f'vm set_ios "{self._name}" "{image}"') - log.info( + log.debug( 'Router "{name}" [{id}]: has a new IOS image set: "{image}"'.format( name=self._name, id=self._id, image=image ) @@ -574,7 +574,7 @@ class Router(BaseNode): return await self._hypervisor.send(f'vm set_ram "{self._name}" {ram}') - log.info( + log.debug( 'Router "{name}" [{id}]: RAM updated from {old_ram}MB to {new_ram}MB'.format( name=self._name, id=self._id, old_ram=self._ram, new_ram=ram ) @@ -602,7 +602,7 @@ class Router(BaseNode): return await self._hypervisor.send(f'vm set_nvram "{self._name}" {nvram}') - log.info( + log.debug( 'Router "{name}" [{id}]: NVRAM updated from {old_nvram}KB to {new_nvram}KB'.format( name=self._name, id=self._id, old_nvram=self._nvram, new_nvram=nvram ) @@ -635,9 +635,9 @@ class Router(BaseNode): await self._hypervisor.send(f'vm set_ram_mmap "{self._name}" {flag}') if mmap: - log.info(f'Router "{self._name}" [{self._id}]: mmap enabled') + log.debug(f'Router "{self._name}" [{self._id}]: mmap enabled') else: - log.info(f'Router "{self._name}" [{self._id}]: mmap disabled') + log.debug(f'Router "{self._name}" [{self._id}]: mmap disabled') self._mmap = mmap @property @@ -664,9 +664,9 @@ class Router(BaseNode): await self._hypervisor.send(f'vm set_sparse_mem "{self._name}" {flag}') if sparsemem: - log.info(f'Router "{self._name}" [{self._id}]: sparse memory enabled') + log.debug(f'Router "{self._name}" [{self._id}]: sparse memory enabled') else: - log.info(f'Router "{self._name}" [{self._id}]: sparse memory disabled') + log.debug(f'Router "{self._name}" [{self._id}]: sparse memory disabled') self._sparsemem = sparsemem @property @@ -688,7 +688,7 @@ class Router(BaseNode): """ await self._hypervisor.send(f'vm set_clock_divisor "{self._name}" {clock_divisor}') - log.info( + log.debug( 'Router "{name}" [{id}]: clock divisor updated from {old_clock} to {new_clock}'.format( name=self._name, id=self._id, old_clock=self._clock_divisor, new_clock=clock_divisor ) @@ -722,7 +722,7 @@ class Router(BaseNode): else: await self._hypervisor.send(f'vm set_idle_pc_online "{self._name}" 0 {idlepc}') - log.info(f'Router "{self._name}" [{self._id}]: idle-PC set to {idlepc}') + log.debug(f'Router "{self._name}" [{self._id}]: idle-PC set to {idlepc}') self._idlepc = idlepc async def get_idle_pc_prop(self): @@ -741,10 +741,10 @@ class Router(BaseNode): was_auto_started = True await asyncio.sleep(20) # leave time to the router to boot - log.info(f'Router "{self._name}" [{self._id}] has started calculating Idle-PC values') + log.debug(f'Router "{self._name}" [{self._id}] has started calculating Idle-PC values') begin = time.time() idlepcs = await self._hypervisor.send(f'vm get_idle_pc_prop "{self._name}" 0') - log.info( + log.debug( 'Router "{name}" [{id}] has finished calculating Idle-PC values after {time:.4f} seconds'.format( name=self._name, id=self._id, time=time.time() - begin ) @@ -789,7 +789,7 @@ class Router(BaseNode): if is_running: # router is running await self._hypervisor.send(f'vm set_idle_max "{self._name}" 0 {idlemax}') - log.info( + log.debug( 'Router "{name}" [{id}]: idlemax updated from {old_idlemax} to {new_idlemax}'.format( name=self._name, id=self._id, old_idlemax=self._idlemax, new_idlemax=idlemax ) @@ -820,7 +820,7 @@ class Router(BaseNode): 'vm set_idle_sleep_time "{name}" 0 {idlesleep}'.format(name=self._name, idlesleep=idlesleep) ) - log.info( + log.debug( 'Router "{name}" [{id}]: idlesleep updated from {old_idlesleep} to {new_idlesleep}'.format( name=self._name, id=self._id, old_idlesleep=self._idlesleep, new_idlesleep=idlesleep ) @@ -849,7 +849,7 @@ class Router(BaseNode): 'vm set_ghost_file "{name}" "{ghost_file}"'.format(name=self._name, ghost_file=ghost_file) ) - log.info( + log.debug( 'Router "{name}" [{id}]: ghost file set to "{ghost_file}"'.format( name=self._name, id=self._id, ghost_file=ghost_file ) @@ -892,7 +892,7 @@ class Router(BaseNode): 'vm set_ghost_status "{name}" {ghost_status}'.format(name=self._name, ghost_status=ghost_status) ) - log.info( + log.debug( 'Router "{name}" [{id}]: ghost status set to {ghost_status}'.format( name=self._name, id=self._id, ghost_status=ghost_status ) @@ -923,7 +923,7 @@ class Router(BaseNode): 'vm set_exec_area "{name}" {exec_area}'.format(name=self._name, exec_area=exec_area) ) - log.info( + log.debug( 'Router "{name}" [{id}]: exec area updated from {old_exec}MB to {new_exec}MB'.format( name=self._name, id=self._id, old_exec=self._exec_area, new_exec=exec_area ) @@ -949,7 +949,7 @@ class Router(BaseNode): await self._hypervisor.send(f'vm set_disk0 "{self._name}" {disk0}') - log.info( + log.debug( 'Router "{name}" [{id}]: disk0 updated from {old_disk0}MB to {new_disk0}MB'.format( name=self._name, id=self._id, old_disk0=self._disk0, new_disk0=disk0 ) @@ -975,7 +975,7 @@ class Router(BaseNode): await self._hypervisor.send(f'vm set_disk1 "{self._name}" {disk1}') - log.info( + log.debug( 'Router "{name}" [{id}]: disk1 updated from {old_disk1}MB to {new_disk1}MB'.format( name=self._name, id=self._id, old_disk1=self._disk1, new_disk1=disk1 ) @@ -1000,9 +1000,9 @@ class Router(BaseNode): """ if auto_delete_disks: - log.info(f'Router "{self._name}" [{self._id}]: auto delete disks enabled') + log.debug(f'Router "{self._name}" [{self._id}]: auto delete disks enabled') else: - log.info(f'Router "{self._name}" [{self._id}]: auto delete disks disabled') + log.debug(f'Router "{self._name}" [{self._id}]: auto delete disks disabled') self._auto_delete_disks = auto_delete_disks async def set_console(self, console): @@ -1130,7 +1130,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: MAC address updated from {old_mac} to {new_mac}'.format( name=self._name, id=self._id, old_mac=self._mac_addr, new_mac=mac_addr ) @@ -1160,7 +1160,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: system ID updated from {old_id} to {new_id}'.format( name=self._name, id=self._id, old_id=self._system_id, new_id=system_id ) @@ -1218,7 +1218,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: adapter {adapter} inserted into slot {slot_number}'.format( name=self._name, id=self._id, adapter=adapter, slot_number=slot_number ) @@ -1233,7 +1233,7 @@ class Router(BaseNode): 'vm slot_oir_start "{name}" {slot_number} 0'.format(name=self._name, slot_number=slot_number) ) - log.info( + log.debug( 'Router "{name}" [{id}]: OIR start event sent to slot {slot_number}'.format( name=self._name, id=self._id, slot_number=slot_number ) @@ -1279,7 +1279,7 @@ class Router(BaseNode): 'vm slot_oir_stop "{name}" {slot_number} 0'.format(name=self._name, slot_number=slot_number) ) - log.info( + log.debug( 'Router "{name}" [{id}]: OIR stop event sent to slot {slot_number}'.format( name=self._name, id=self._id, slot_number=slot_number ) @@ -1289,7 +1289,7 @@ class Router(BaseNode): 'vm slot_remove_binding "{name}" {slot_number} 0'.format(name=self._name, slot_number=slot_number) ) - log.info( + log.debug( 'Router "{name}" [{id}]: adapter {adapter} removed from slot {slot_number}'.format( name=self._name, id=self._id, adapter=adapter, slot_number=slot_number ) @@ -1331,7 +1331,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: {wic} inserted into WIC slot {wic_slot_number}'.format( name=self._name, id=self._id, wic=wic, wic_slot_number=wic_slot_number ) @@ -1375,7 +1375,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: {wic} removed from WIC slot {wic_slot_number}'.format( name=self._name, id=self._id, wic=adapter.wics[wic_slot_number], wic_slot_number=wic_slot_number ) @@ -1441,7 +1441,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: NIO {nio_name} bound to port {slot_number}/{port_number}'.format( name=self._name, id=self._id, nio_name=nio.name, slot_number=slot_number, port_number=port_number ) @@ -1502,7 +1502,7 @@ class Router(BaseNode): await nio.close() adapter.remove_nio(port_number) - log.info( + log.debug( 'Router "{name}" [{id}]: NIO {nio_name} removed from port {slot_number}/{port_number}'.format( name=self._name, id=self._id, nio_name=nio.name, slot_number=slot_number, port_number=port_number ) @@ -1526,7 +1526,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: NIO enabled on port {slot_number}/{port_number}'.format( name=self._name, id=self._id, slot_number=slot_number, port_number=port_number ) @@ -1581,7 +1581,7 @@ class Router(BaseNode): ) ) - log.info( + log.debug( 'Router "{name}" [{id}]: NIO disabled on port {slot_number}/{port_number}'.format( name=self._name, id=self._id, slot_number=slot_number, port_number=port_number ) @@ -1635,7 +1635,7 @@ class Router(BaseNode): ) ) await nio.start_packet_capture(output_file, data_link_type) - log.info( + log.debug( 'Router "{name}" [{id}]: starting packet capture on port {slot_number}/{port_number}'.format( name=self._name, id=self._id, nio_name=nio.name, slot_number=slot_number, port_number=port_number ) @@ -1675,7 +1675,7 @@ class Router(BaseNode): return await nio.stop_packet_capture() - log.info( + log.debug( 'Router "{name}" [{id}]: stopping packet capture on port {slot_number}/{port_number}'.format( name=self._name, id=self._id, nio_name=nio.name, slot_number=slot_number, port_number=port_number ) @@ -1748,7 +1748,7 @@ class Router(BaseNode): except OSError as e: raise DynamipsError(f"Could not amend the configuration {self.private_config_path}: {e}") - log.info(f'Router "{self._name}" [{self._id}]: renamed to "{new_name}"') + log.debug(f'Router "{self._name}" [{self._id}]: renamed to "{new_name}"') self._name = new_name async def extract_config(self): @@ -1788,7 +1788,7 @@ class Router(BaseNode): config = "!\n" + config.replace("\r", "") config_path = os.path.join(self._working_directory, startup_config) with open(config_path, "wb") as f: - log.info(f"saving startup-config to {startup_config}") + log.debug(f"saving startup-config to {startup_config}") f.write(config.encode("utf-8")) except (binascii.Error, OSError) as e: raise DynamipsError(f"Could not save the startup configuration {config_path}: {e}") @@ -1799,7 +1799,7 @@ class Router(BaseNode): config = base64.b64decode(private_config_base64).decode("utf-8", errors="replace") config_path = os.path.join(self._working_directory, private_config) with open(config_path, "wb") as f: - log.info(f"saving private-config to {private_config}") + log.debug(f"saving private-config to {private_config}") f.write(config.encode("utf-8")) except (binascii.Error, OSError) as e: raise DynamipsError(f"Could not save the private configuration {config_path}: {e}") @@ -1827,7 +1827,7 @@ class Router(BaseNode): await wait_run_in_executor(shutil.rmtree, self._working_directory) except OSError as e: log.warning(f"Could not delete file {e}") - log.info(f'Router "{self._name}" [{self._id}] has been deleted (including associated files)') + log.debug(f'Router "{self._name}" [{self._id}] has been deleted (including associated files)') def _memory_files(self): diff --git a/gns3server/compute/iou/iou_vm.py b/gns3server/compute/iou/iou_vm.py index edba56876..ff46bb4fb 100644 --- a/gns3server/compute/iou/iou_vm.py +++ b/gns3server/compute/iou/iou_vm.py @@ -162,7 +162,7 @@ class IOUVM(BaseNode): super().__init__(name, node_id, project, manager, console=console, console_type=console_type) - log.info( + log.debug( 'IOU "{name}" [{id}]: assigned with application ID {application_id}'.format( name=self._name, id=self._id, application_id=application_id ) @@ -238,7 +238,7 @@ class IOUVM(BaseNode): self._path = self.manager.get_abs_image_path(path, self.project.path) self._loader = None - log.info(f'IOU "{self._name}" [{self._id}]: IOU image updated to "{self._path}"') + log.debug(f'IOU "{self._name}" [{self._id}]: IOU image updated to "{self._path}"') @property def use_default_iou_values(self): @@ -260,9 +260,9 @@ class IOUVM(BaseNode): self._use_default_iou_values = state if state: - log.info(f'IOU "{self._name}" [{self._id}]: uses the default IOU image values') + log.debug(f'IOU "{self._name}" [{self._id}]: uses the default IOU image values') else: - log.info(f'IOU "{self._name}" [{self._id}]: does not use the default IOU image values') + log.debug(f'IOU "{self._name}" [{self._id}]: does not use the default IOU image values') async def update_default_iou_values(self): """ @@ -430,7 +430,7 @@ class IOUVM(BaseNode): if self._ram == ram: return - log.info( + log.debug( 'IOU "{name}" [{id}]: RAM updated from {old_ram}MB to {new_ram}MB'.format( name=self._name, id=self._id, old_ram=self._ram, new_ram=ram ) @@ -459,7 +459,7 @@ class IOUVM(BaseNode): if self._nvram == nvram: return - log.info( + log.debug( 'IOU "{name}" [{id}]: NVRAM updated from {old_nvram}KB to {new_nvram}KB'.format( name=self._name, id=self._id, old_nvram=self._nvram, new_nvram=nvram ) @@ -574,7 +574,7 @@ class IOUVM(BaseNode): config = configparser.ConfigParser() try: - log.info(f"Checking IOU license in '{self.iourc_path}'") + log.debug(f"Checking IOU license in '{self.iourc_path}'") with open(self.iourc_path, encoding="utf-8") as f: config.read_file(f) except OSError as e: @@ -724,9 +724,9 @@ class IOUVM(BaseNode): await self._start_l1_keepalive_responder() try: if self._loader: - log.info(f"Starting IOU: {command} with loader {self._loader}") + log.debug(f"Starting IOU: {command} with loader {self._loader}") else: - log.info(f"Starting IOU: {command}") + log.debug(f"Starting IOU: {command}") self.command_line = " ".join(command) self._iou_process = await asyncio.create_subprocess_exec( *self._loader, *command, @@ -736,7 +736,7 @@ class IOUVM(BaseNode): cwd=self.working_dir, env=env, ) - log.info(f"IOU instance {self._id} started PID={self._iou_process.pid}") + log.debug(f"IOU instance {self._id} started PID={self._iou_process.pid}") self._started = True self.status = "started" callback = functools.partial(self._termination_callback, "IOU") @@ -920,7 +920,7 @@ class IOUVM(BaseNode): """ if self._iou_process: - log.info(f'Stopping IOU process for IOU VM "{self.name}" PID={self._iou_process.pid}') + log.debug(f'Stopping IOU process for IOU VM "{self.name}" PID={self._iou_process.pid}') try: self._iou_process.terminate() # Sometime the process can already be dead when we garbage collect @@ -979,7 +979,7 @@ class IOUVM(BaseNode): iou_id=self.application_id, ) ) - log.info("IOU {name} [id={id}]: NETMAP file created".format(name=self._name, id=self._id)) + log.debug("IOU {name} [id={id}]: NETMAP file created".format(name=self._name, id=self._id)) except OSError as e: raise IOUError(f"Could not create {netmap_path}: {e}") @@ -1030,7 +1030,7 @@ class IOUVM(BaseNode): ) self._l1_keepalive_transport = transport self._l1_keepalive_task = asyncio.create_task(self._send_l1_keepalives(protocol)) - log.info( + log.debug( 'IOU "%s" [%s]: L1 keepalive responder listening on %s', self._name, self._id, @@ -1150,7 +1150,7 @@ class IOUVM(BaseNode): for _ in range(0, ethernet_adapters): self._ethernet_adapters.append(EthernetAdapter(interfaces=4)) - log.info( + log.debug( 'IOU "{name}" [{id}]: number of Ethernet adapters changed to {adapters}'.format( name=self._name, id=self._id, adapters=len(self._ethernet_adapters) ) @@ -1180,7 +1180,7 @@ class IOUVM(BaseNode): for _ in range(0, serial_adapters): self._serial_adapters.append(SerialAdapter(interfaces=4)) - log.info( + log.debug( 'IOU "{name}" [{id}]: number of Serial adapters changed to {adapters}'.format( name=self._name, id=self._id, adapters=len(self._serial_adapters) ) @@ -1214,7 +1214,7 @@ class IOUVM(BaseNode): ) adapter.add_nio(port_number, nio) - log.info( + log.debug( 'IOU "{name}" [{id}]: {nio} added to {adapter_number}/{port_number}'.format( name=self._name, id=self._id, nio=nio, adapter_number=adapter_number, port_number=port_number ) @@ -1266,7 +1266,9 @@ class IOUVM(BaseNode): async def _ubridge_apply_markers(self, adapter_number, port_number, nio): """ - (Re-)apply traffic-insight markers to the IOL bridge. + Reconcile traffic-insight markers on the IOL bridge (diff desired + ``nio.markers`` against installed ``_marker_specs``): delete removed, + rebuild changed, toggle on/off-only changes, add new, skip unchanged. IOU uses ``iol_bridge`` (not ``bridge``) and the ``add_packet_filter`` command carries extra ``{bay} {unit}`` positional arguments between the @@ -1280,41 +1282,60 @@ class IOUVM(BaseNode): from gns3server.compute.marker.marker_manager import MarkerManager markers = nio.markers if hasattr(nio, 'markers') else {} - if not markers: - return - manager = MarkerManager.instance() markers_dir = self.project.markers_working_directory() bridge_name = f"IOL-BRIDGE-{self.application_id + 512}" location = "{bridge_name} {bay} {unit}".format( bridge_name=bridge_name, bay=adapter_number, unit=port_number ) - for name, spec in markers.items(): - link_id = spec.get("link_id", "") - # Incremental: skip markers already installed on this port. A NIO - # update carries EVERY marker on the port (e.g. an inherited - # global-* copy plus a newly added private one); uBridge's - # add_packet_filter rejects a duplicate filter name (packet_filter.c), - # so we must not re-add one already here — mirrors the generic - # _ubridge_apply_markers guard. A fresh bridge has an empty map - # (cleared on _stop_ubridge) so all are installed. - if (name, link_id) in self._marker_filter_bridges: + desired = {(name, spec.get("link_id", "")): spec for name, spec in markers.items()} + + # 1. Remove installed markers that are no longer desired. + # Scope to THIS port's IOL location — the map is node-wide and also + # holds markers on this IOU's other ports, which must not be deleted + # when reconciling a single NIO (see base_node for the same guard). + for key in list(self._marker_filter_bridges): + if self._marker_filter_bridges[key] != location: continue + if key not in desired: + mname, link_id = key + installed_location = self._marker_filter_bridges.pop(key) + self._marker_specs.pop(key, None) + await self._ubridge_delete_marker_filter(installed_location, mname) + try: + os.remove(os.path.join(markers_dir, f"{self._id}_{link_id}_{mname}.pcap")) + except FileNotFoundError: + pass + except OSError as e: + log.warning("Could not remove marker pcap for '%s' on link %s: %s", mname, link_id, e) + manager.unregister(self._id, mname) + + # 2. Add / reconcile desired markers. + rebuild_fields = ("bpf", "tag", "direction", "data_link_type") + for (name, link_id), spec in desired.items(): bpf = spec.get("bpf", "") tag = spec.get("tag") - pcap_path = os.path.join( - markers_dir, f"{self._id}_{link_id}_{name}.pcap" - ) - # Build the iol_bridge marker filter command: - # iol_bridge add_packet_filter {br} {bay} {unit} {name} mark "{bpf}" [tag {id}] pcap "{path}" + enabled = spec.get("enabled", True) + if (name, link_id) in self._marker_filter_bridges: + installed_spec = self._marker_specs.get((name, link_id)) + if installed_spec is None: + continue # installed (legacy, no spec) — skip to avoid dup + if any(installed_spec.get(f) != spec.get(f) for f in rebuild_fields): + installed_location = self._marker_filter_bridges.get((name, link_id)) + await self._ubridge_delete_marker_filter(installed_location, name) + elif installed_spec.get("enabled", True) != enabled: + await self._ubridge_set_marker_filter_state(name, enabled) + self._marker_specs[(name, link_id)] = spec + continue + else: + continue + pcap_path = os.path.join(markers_dir, f"{self._id}_{link_id}_{name}.pcap") + # iol_bridge add_packet_filter {br} {bay} {unit} {name} mark "{bpf}" [tag {id}] pcap "{path}" cmd = 'iol_bridge add_packet_filter {loc} {name} mark "{bpf}"'.format( loc=location, name=name, bpf=bpf ) if tag is not None: cmd += f" tag {tag}" - # IOU uses one per-node IOL-BRIDGE for every link, so bridge+filter - # are identical across this node's links — `link` is the only way the - # controller can tell their signals apart (contract §3.2). if link_id: cmd += f" link {link_id}" direction = spec.get("direction") @@ -1333,16 +1354,14 @@ class IOUVM(BaseNode): self.project.emit("log.warning", {"message": message}) continue raise - if not spec.get("enabled", True): + if not enabled: try: await self._ubridge_send(f"iol_bridge enable_packet_filter {location} {name} off") except UbridgeError as e: log.warning(f"Could not turn marker '{name}' off on {location}: {e}") - manager.register( - str(self.project.id), self._id, name, link_id, tag - ) - # Record name -> location (bridge bay unit) for instant toggle. + manager.register(str(self.project.id), self._id, name, link_id, tag) self._marker_filter_bridges[name, link_id] = location + self._marker_specs[name, link_id] = spec async def _ubridge_set_marker_filter_state(self, name, enabled): """IOU override: toggle every (name, link_id) entry via ``iol_bridge``.""" @@ -1393,7 +1412,7 @@ class IOUVM(BaseNode): if isinstance(nio, NIOUDP): self.manager.port_manager.release_udp_port(nio.lport, self._project) adapter.remove_nio(port_number) - log.info( + log.debug( 'IOU "{name}" [{id}]: {nio} removed from {adapter_number}/{port_number}'.format( name=self._name, id=self._id, nio=nio, adapter_number=adapter_number, port_number=port_number ) @@ -1463,9 +1482,9 @@ class IOUVM(BaseNode): self._l1_keepalives = state if state: - log.info(f'IOU "{self._name}" [{self._id}]: has activated layer 1 keepalive messages') + log.debug(f'IOU "{self._name}" [{self._id}]: has activated layer 1 keepalive messages') else: - log.info(f'IOU "{self._name}" [{self._id}]: has deactivated layer 1 keepalive messages') + log.debug(f'IOU "{self._name}" [{self._id}]: has deactivated layer 1 keepalive messages') async def _enable_l1_keepalives(self, command): """ @@ -1700,7 +1719,7 @@ class IOUVM(BaseNode): try: config = startup_config_content.decode("utf-8", errors="replace") with open(config_path, "wb") as f: - log.info(f"saving startup-config to {config_path}") + log.debug(f"saving startup-config to {config_path}") f.write(config.encode("utf-8")) except (binascii.Error, OSError) as e: raise IOUError(f"Could not save the startup configuration {config_path}: {e}") @@ -1710,7 +1729,7 @@ class IOUVM(BaseNode): try: config = private_config_content.decode("utf-8", errors="replace") with open(config_path, "wb") as f: - log.info(f"saving private-config to {config_path}") + log.debug(f"saving private-config to {config_path}") f.write(config.encode("utf-8")) except (binascii.Error, OSError) as e: raise IOUError(f"Could not save the private configuration {config_path}: {e}") @@ -1734,7 +1753,7 @@ class IOUVM(BaseNode): ) nio.start_packet_capture(output_file, data_link_type) - log.info( + log.debug( 'IOU "{name}" [{id}]: starting packet capture on {adapter_number}/{port_number} to {output_file}'.format( name=self._name, id=self._id, @@ -1768,7 +1787,7 @@ class IOUVM(BaseNode): if not nio.capturing: return nio.stop_packet_capture() - log.info( + log.debug( 'IOU "{name}" [{id}]: stopping packet capture on {adapter_number}/{port_number}'.format( name=self._name, id=self._id, adapter_number=adapter_number, port_number=port_number ) diff --git a/gns3server/compute/marker/marker_listener.py b/gns3server/compute/marker/marker_listener.py index 2b5e5d47f..40cdd97b6 100644 --- a/gns3server/compute/marker/marker_listener.py +++ b/gns3server/compute/marker/marker_listener.py @@ -49,14 +49,18 @@ class MarkerListener(asyncio.DatagramProtocol): # MarkerManager owns this listener and the registry. self._manager = manager self.transport = None + self._received = 0 + self._errors = 0 def connection_made(self, transport): self.transport = transport def datagram_received(self, data, addr): + self._received += 1 try: self._handle(data) except Exception: + self._errors += 1 # Never let a malformed datagram kill the listener. log.exception("Failed to process MARK datagram from %s: %r", addr, data) diff --git a/gns3server/compute/marker/marker_manager.py b/gns3server/compute/marker/marker_manager.py index 6ec128158..132a19812 100644 --- a/gns3server/compute/marker/marker_manager.py +++ b/gns3server/compute/marker/marker_manager.py @@ -17,6 +17,7 @@ import asyncio import logging +import socket from gns3server.compute.marker.marker_listener import MarkerListener from gns3server.compute.notification_manager import NotificationManager @@ -75,10 +76,20 @@ class MarkerManager: return loop = asyncio.get_running_loop() self._listener = MarkerListener(self) + + def _configure_transport(transport): + sock = transport.get_extra_info("socket") + if sock is not None: + # Raise the UDP receive buffer from the default ~208 KB to 8 MB + # so that 1000+ uBridge processes can burst marker.match signals + # without kernel-side datagram loss. + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 * 1024 * 1024) + try: self._transport, _ = await loop.create_datagram_endpoint( lambda: self._listener, local_addr=(host, port) ) + _configure_transport(self._transport) except OSError: if port != 0: log.warning( @@ -88,6 +99,7 @@ class MarkerManager: self._transport, _ = await loop.create_datagram_endpoint( lambda: self._listener, local_addr=(host, 0) ) + _configure_transport(self._transport) except OSError as e: log.error( "Marker listener startup failed: %s. Traffic insight signals are unavailable.", e @@ -104,10 +116,31 @@ class MarkerManager: self._host = host self._port = sock.getsockname()[1] if sock else port log.info("Marker signal sink listening on %s:%s", self._host, self._port) + self._stats_task = asyncio.create_task(self._log_stats()) + + async def _log_stats(self): + """Log marker.match throughput every 10 s so operators can tell whether + the single UDP sink keeps up with the aggregated uBridge traffic.""" + while self.running: + await asyncio.sleep(10) + listener = self._listener + if listener is None: + break + received, errors = listener._received, listener._errors + listener._received = 0 + listener._errors = 0 + if received: + log.info( + "marker sink: %d matches (%.0f/s), %d errors in last 10s", + received, received / 10.0, errors, + ) async def stop(self): """Close the UDP sink and drop the whole registry.""" + if hasattr(self, "_stats_task") and self._stats_task: + self._stats_task.cancel() + self._stats_task = None if self._transport: self._transport.close() self._transport = None diff --git a/gns3server/compute/qemu/qemu_vm.py b/gns3server/compute/qemu/qemu_vm.py index ce538d7f6..439523c65 100644 --- a/gns3server/compute/qemu/qemu_vm.py +++ b/gns3server/compute/qemu/qemu_vm.py @@ -187,7 +187,7 @@ class QemuVM(BaseNode): log.warning(f"Config disk: image '{self.config_disk_name}' missing") self.config_disk_name = "" - log.info(f'QEMU VM "{self._name}" [{self._id}] has been created') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has been created') @BaseNode.name.setter def name(self, new_name): @@ -270,7 +270,7 @@ class QemuVM(BaseNode): self._platform = re.sub(r'^qemu-system-(\w+).*$', r'\1', qemu_bin, flags=re.IGNORECASE) if self._platform.split(".")[0] not in list(QemuPlatform): raise QemuError(f"Platform {self._platform} is unknown") - log.info(f'QEMU VM "{self._name}" [{self._name}] has set the QEMU path to {qemu_path}') + log.debug(f'QEMU VM "{self._name}" [{self._name}] has set the QEMU path to {qemu_path}') def _check_qemu_path(self, qemu_path): @@ -292,7 +292,7 @@ class QemuVM(BaseNode): def platform(self, platform): self._platform = platform - log.info(f"QEMU VM '{self._name}' [{self._id}] has set the platform {platform}") + log.debug(f"QEMU VM '{self._name}' [{self._id}] has set the platform {platform}") self.qemu_path = f"qemu-system-{platform}" def _disk_setter(self, variable, value): @@ -311,7 +311,7 @@ class QemuVM(BaseNode): f"Sorry a node without the linked base setting enabled can only be used once on your server. {value} is already used by {node.name} in project {node.project.name}" ) setattr(self, "_" + variable, value) - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU {variable} path to {disk_image}'.format( name=self._name, variable=variable, id=self._id, disk_image=value ) @@ -416,7 +416,7 @@ class QemuVM(BaseNode): """ self._hda_disk_interface = hda_disk_interface - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU hda disk interface to {interface}'.format( name=self._name, id=self._id, interface=self._hda_disk_interface ) @@ -441,7 +441,7 @@ class QemuVM(BaseNode): """ self._hdb_disk_interface = hdb_disk_interface - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU hdb disk interface to {interface}'.format( name=self._name, id=self._id, interface=self._hdb_disk_interface ) @@ -466,7 +466,7 @@ class QemuVM(BaseNode): """ self._hdc_disk_interface = hdc_disk_interface - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU hdc disk interface to {interface}'.format( name=self._name, id=self._id, interface=self._hdc_disk_interface ) @@ -491,7 +491,7 @@ class QemuVM(BaseNode): """ self._hdd_disk_interface = hdd_disk_interface - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU hdd disk interface to {interface}'.format( name=self._name, id=self._id, interface=self._hdd_disk_interface ) @@ -518,7 +518,7 @@ class QemuVM(BaseNode): if cdrom_image: self._cdrom_image = self.manager.get_abs_image_path(cdrom_image, self.working_dir) - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU cdrom image path to {cdrom_image}'.format( name=self._name, id=self._id, cdrom_image=self._cdrom_image ) @@ -547,14 +547,14 @@ class QemuVM(BaseNode): self._cdrom_option() # this will check the cdrom image is accessible await self._control_vm("eject -f ide1-cd0") await self._control_vm(f"change ide1-cd0 {self._cdrom_image}") - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has changed the cdrom image path to {cdrom_image}'.format( name=self._name, id=self._id, cdrom_image=self._cdrom_image ) ) else: await self._control_vm("eject -f ide1-cd0") - log.info(f'QEMU VM "{self._name}" [{self._id}] has ejected the cdrom image') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has ejected the cdrom image') @property def bios_image(self): @@ -575,7 +575,7 @@ class QemuVM(BaseNode): """ self._bios_image = self.manager.get_abs_image_path(bios_image, self.working_dir) - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU bios image path to {bios_image}'.format( name=self._name, id=self._id, bios_image=self._bios_image ) @@ -600,7 +600,7 @@ class QemuVM(BaseNode): """ self._boot_priority = boot_priority - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the boot priority to {boot_priority}'.format( name=self._name, id=self._id, boot_priority=self._boot_priority ) @@ -635,7 +635,7 @@ class QemuVM(BaseNode): for adapter_number in range(0, adapters): self._ethernet_adapters.append(EthernetAdapter()) - log.info( + log.debug( 'QEMU VM "{name}" [{id}]: number of Ethernet adapters changed to {adapters}'.format( name=self._name, id=self._id, adapters=adapters ) @@ -661,7 +661,7 @@ class QemuVM(BaseNode): self._adapter_type = adapter_type - log.info( + log.debug( 'QEMU VM "{name}" [{id}]: adapter type changed to {adapter_type}'.format( name=self._name, id=self._id, adapter_type=adapter_type ) @@ -691,7 +691,7 @@ class QemuVM(BaseNode): else: self._mac_address = mac_address - log.info( + log.debug( 'QEMU VM "{name}" [{id}]: MAC address changed to {mac_addr}'.format( name=self._name, id=self._id, mac_addr=self._mac_address ) @@ -716,9 +716,9 @@ class QemuVM(BaseNode): """ if replicate_network_connection_state: - log.info(f'QEMU VM "{self._name}" [{self._id}] has enabled network connection state replication') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has enabled network connection state replication') else: - log.info(f'QEMU VM "{self._name}" [{self._id}] has disabled network connection state replication') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has disabled network connection state replication') self._replicate_network_connection_state = replicate_network_connection_state @property @@ -740,9 +740,9 @@ class QemuVM(BaseNode): """ if create_config_disk: - log.info(f'QEMU VM "{self._name}" [{self._id}] has enabled the config disk creation feature') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has enabled the config disk creation feature') else: - log.info(f'QEMU VM "{self._name}" [{self._id}] has disabled the config disk creation feature') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has disabled the config disk creation feature') self._create_config_disk = create_config_disk @property @@ -763,7 +763,7 @@ class QemuVM(BaseNode): :param on_close: string """ - log.info(f'QEMU VM "{self._name}" [{self._id}] set the close action to "{on_close}"') + log.debug(f'QEMU VM "{self._name}" [{self._id}] set the close action to "{on_close}"') self._on_close = on_close @property @@ -784,7 +784,7 @@ class QemuVM(BaseNode): :param cpu_throttling: integer """ - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the percentage of CPU allowed to {cpu}'.format( name=self._name, id=self._id, cpu=cpu_throttling ) @@ -812,7 +812,7 @@ class QemuVM(BaseNode): :param process_priority: string """ - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the process priority to {priority}'.format( name=self._name, id=self._id, priority=process_priority ) @@ -837,7 +837,7 @@ class QemuVM(BaseNode): :param ram: RAM amount in MB """ - log.info(f'QEMU VM "{self._name}" [{self._id}] has set the RAM to {ram}') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has set the RAM to {ram}') self._ram = ram @property @@ -858,7 +858,7 @@ class QemuVM(BaseNode): :param cpus: number of vCPUs. """ - log.info(f'QEMU VM "{self._name}" [{self._id}] has set the number of vCPUs to {cpus}') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has set the number of vCPUs to {cpus}') self._cpus = cpus @property @@ -879,7 +879,7 @@ class QemuVM(BaseNode): :param maxcpus: maximum number of hotpluggable vCPUs """ - log.info(f'QEMU VM "{self._name}" [{self._id}] has set maximum number of hotpluggable vCPUs to {maxcpus}') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has set maximum number of hotpluggable vCPUs to {maxcpus}') self._maxcpus = maxcpus @property @@ -901,9 +901,9 @@ class QemuVM(BaseNode): """ if tpm: - log.info(f'QEMU VM "{self._name}" [{self._id}] has enabled the Trusted Platform Module (TPM)') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has enabled the Trusted Platform Module (TPM)') else: - log.info(f'QEMU VM "{self._name}" [{self._id}] has disabled the Trusted Platform Module (TPM)') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has disabled the Trusted Platform Module (TPM)') self._tpm = tpm @property @@ -925,9 +925,9 @@ class QemuVM(BaseNode): """ if uefi: - log.info(f'QEMU VM "{self._name}" [{self._id}] has enabled the UEFI boot mode') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has enabled the UEFI boot mode') else: - log.info(f'QEMU VM "{self._name}" [{self._id}] has disabled the UEFI boot mode') + log.debug(f'QEMU VM "{self._name}" [{self._id}] has disabled the UEFI boot mode') self._uefi = uefi @property @@ -948,7 +948,7 @@ class QemuVM(BaseNode): :param options: QEMU options """ - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU options to {options}'.format( name=self._name, id=self._id, options=options ) @@ -996,7 +996,7 @@ class QemuVM(BaseNode): initrd = self.manager.get_abs_image_path(initrd, self.working_dir) - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU initrd path to {initrd}'.format( name=self._name, id=self._id, initrd=initrd ) @@ -1029,7 +1029,7 @@ class QemuVM(BaseNode): """ kernel_image = self.manager.get_abs_image_path(kernel_image, self.working_dir) - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU kernel image path to {kernel_image}'.format( name=self._name, id=self._id, kernel_image=kernel_image ) @@ -1054,7 +1054,7 @@ class QemuVM(BaseNode): :param kernel_command_line: QEMU kernel command line """ - log.info( + log.debug( 'QEMU VM "{name}" [{id}] has set the QEMU kernel command line to {kernel_command_line}'.format( name=self._name, id=self._id, kernel_command_line=kernel_command_line ) @@ -1114,7 +1114,7 @@ class QemuVM(BaseNode): command = [cpulimit_exec, "--lazy", "--pid={}".format(self._process.pid), "--limit={}".format(self._cpu_throttling)] self._cpulimit_process = subprocess.Popen(command, cwd=self.working_dir) - log.info(f"CPU throttled to {self._cpu_throttling}%") + log.debug(f"CPU throttled to {self._cpu_throttling}%") except FileNotFoundError: raise QemuError("cpulimit could not be found, please install it or deactivate CPU throttling") except (OSError, subprocess.SubprocessError) as e: @@ -1172,16 +1172,16 @@ class QemuVM(BaseNode): command = await self._build_command() command_string = " ".join(shlex.quote(s) for s in command) try: - log.info(f"Starting QEMU with: {command_string}") + log.debug(f"Starting QEMU with: {command_string}") self._stdout_file = os.path.join(self.working_dir, "qemu.log") - log.info(f"logging to {self._stdout_file}") + log.debug(f"logging to {self._stdout_file}") with open(self._stdout_file, "w", encoding="utf-8") as fd: fd.write(f"Start QEMU with {command_string}\n\nExecution log:\n") self.command_line = " ".join(command) self._process = await asyncio.create_subprocess_exec( *command, stdout=fd, stderr=subprocess.STDOUT, cwd=self.working_dir ) - log.info(f'QEMU VM "{self._name}" started PID={self._process.pid}') + log.debug(f'QEMU VM "{self._name}" started PID={self._process.pid}') self._command_line_changed = False self.status = "started" monitor_process(self._process, self._termination_callback) @@ -1242,7 +1242,7 @@ class QemuVM(BaseNode): """ if self.started: - log.info("QEMU process has stopped, return code: %d", returncode) + log.debug("QEMU process has stopped, return code: %d", returncode) await self.stop() if returncode != 0: qemu_stdout = self.read_stdout() @@ -1270,7 +1270,7 @@ class QemuVM(BaseNode): # stop the QEMU process self._hw_virtualization = False if self.is_running(): - log.info(f'Stopping QEMU VM "{self._name}" PID={self._process.pid}') + log.debug(f'Stopping QEMU VM "{self._name}" PID={self._process.pid}') try: if self.on_close == "save_vm_state": @@ -1498,7 +1498,7 @@ class QemuVM(BaseNode): self.status = "suspended" log.debug("QEMU VM has been suspended") else: - log.info(f"QEMU VM is not running to be suspended, current status is {vm_status}") + log.debug(f"QEMU VM is not running to be suspended, current status is {vm_status}") async def reload(self): """ @@ -1525,7 +1525,7 @@ class QemuVM(BaseNode): self.status = "started" log.debug("QEMU VM has been resumed") else: - log.info(f"QEMU VM is not paused to be resumed, current status is {vm_status}") + log.debug(f"QEMU VM is not paused to be resumed, current status is {vm_status}") async def adapter_add_nio_binding(self, adapter_number, nio): """ @@ -1559,7 +1559,7 @@ class QemuVM(BaseNode): ) adapter.add_nio(0, nio) - log.info( + log.debug( 'QEMU VM "{name}" [{id}]: {nio} added to adapter {adapter_number}'.format( name=self._name, id=self._id, nio=nio, adapter_number=adapter_number ) @@ -1619,7 +1619,7 @@ class QemuVM(BaseNode): self.manager.port_manager.release_udp_port(nio.lport, self._project) adapter.remove_nio(0) - log.info( + log.debug( 'QEMU VM "{name}" [{id}]: {nio} removed from adapter {adapter_number}'.format( name=self._name, id=self._id, nio=nio, adapter_number=adapter_number ) @@ -1671,7 +1671,7 @@ class QemuVM(BaseNode): ) ) - log.info( + log.debug( "QEMU VM '{name}' [{id}]: starting packet capture on adapter {adapter_number}".format( name=self.name, id=self.id, adapter_number=adapter_number ) @@ -1692,7 +1692,7 @@ class QemuVM(BaseNode): if self.ubridge: await self._ubridge_send("bridge stop_capture {name}".format(name=f"QEMU-{self._id}-{adapter_number}")) - log.info( + log.debug( "QEMU VM '{name}' [{id}]: stopping packet capture on adapter {adapter_number}".format( name=self.name, id=self.id, adapter_number=adapter_number ) @@ -1731,7 +1731,7 @@ class QemuVM(BaseNode): stdout = self.read_qemu_img_stdout() raise QemuError(f"Could not create '{disk_name}' disk image: qemu-img returned with {retcode}\n{stdout}") else: - log.info(f"QEMU VM '{self.name}' [{self.id}]: Qemu disk image'{disk_name}' created") + log.debug(f"QEMU VM '{self.name}' [{self.id}]: Qemu disk image'{disk_name}' created") except (OSError, subprocess.SubprocessError) as e: stdout = self.read_qemu_img_stdout() raise QemuError(f"Could not create '{disk_name}' disk image: {e}\n{stdout}") @@ -1759,7 +1759,7 @@ class QemuVM(BaseNode): stdout = self.read_qemu_img_stdout() raise QemuError(f"Could not update '{disk_name}' disk image: qemu-img returned with {retcode}\n{stdout}") else: - log.info(f"QEMU VM '{self.name}' [{self.id}]: Qemu disk image '{disk_name}' extended by {extend} MB") + log.debug(f"QEMU VM '{self.name}' [{self.id}]: Qemu disk image '{disk_name}' extended by {extend} MB") except (OSError, subprocess.SubprocessError) as e: stdout = self.read_qemu_img_stdout() raise QemuError(f"Could not update '{disk_name}' disk image: {e}\n{stdout}") @@ -1975,16 +1975,16 @@ class QemuVM(BaseNode): async def _qemu_img_exec(self, command): self._qemu_img_stdout_file = os.path.join(self.working_dir, "qemu-img.log") - log.info(f"logging to {self._qemu_img_stdout_file}") + log.debug(f"logging to {self._qemu_img_stdout_file}") command_string = " ".join(shlex.quote(s) for s in command) - log.info(f"Executing qemu-img with: {command_string}") + log.debug(f"Executing qemu-img with: {command_string}") with open(self._qemu_img_stdout_file, "w", encoding="utf-8") as fd: process = await asyncio.create_subprocess_exec( *command, stdout=fd, stderr=subprocess.STDOUT, cwd=self.working_dir ) retcode = await process.wait() if retcode != 0: - log.info(f"{self._get_qemu_img()} returned with {retcode}") + log.debug(f"{self._get_qemu_img()} returned with {retcode}") return retcode async def _find_disk_file_format(self, disk): @@ -2294,7 +2294,7 @@ class QemuVM(BaseNode): elif self._uefi: system_ovmf_firmware_dir = Path(self.manager.config.settings.Qemu.ovmf_firmware_dir) - log.info("Using OVMF firmware directory: {}".format(system_ovmf_firmware_dir)) + log.debug("Using OVMF firmware directory: {}".format(system_ovmf_firmware_dir)) old_ovmf_vars_path = os.path.join(self.working_dir, "OVMF_VARS.fd") if os.path.exists(old_ovmf_vars_path): # the node has its own UEFI variables store already, we must also use the old UEFI firmware @@ -2313,7 +2313,7 @@ class QemuVM(BaseNode): # otherwise, get the UEFI firmware from the images directory ovmf_firmware_path = self.manager.get_abs_image_path("OVMF_CODE_4M.fd") - log.info("Configuring UEFI boot mode using OVMF file: '{}'".format(ovmf_firmware_path)) + log.debug("Configuring UEFI boot mode using OVMF file: '{}'".format(ovmf_firmware_path)) options.extend(["-drive", "if=pflash,format=raw,readonly,file={}".format(ovmf_firmware_path)]) # try to use the UEFI variables store from the system first @@ -2397,9 +2397,9 @@ class QemuVM(BaseNode): "type=unixio,path={},terminate".format(tpm_sock) ] command_string = " ".join(shlex.quote(s) for s in command) - log.info("Starting swtpm (TPM emulator) with: {}".format(command_string)) + log.debug("Starting swtpm (TPM emulator) with: {}".format(command_string)) self._swtpm_process = subprocess.Popen(command, cwd=self.working_dir) - log.info("swtpm (TPM emulator) has started") + log.debug("swtpm (TPM emulator) has started") except (OSError, subprocess.SubprocessError) as e: raise QemuError("Could not start swtpm (TPM emulator): {}".format(e)) @@ -2587,7 +2587,7 @@ class QemuVM(BaseNode): stdout = self.read_qemu_img_stdout() log.warning(f"Could not delete saved VM state from disk {disk}: {stdout}") else: - log.info(f"Deleted saved VM state from disk {disk}") + log.debug(f"Deleted saved VM state from disk {disk}") except subprocess.SubprocessError as e: raise QemuError(f"Error while looking for the Qemu VM saved state snapshot: {e}") @@ -2617,7 +2617,7 @@ class QemuVM(BaseNode): if "snapshots" in json_data: for snapshot in json_data["snapshots"]: if snapshot["name"] == snapshot_name: - log.info( + log.debug( 'QEMU VM "{name}" [{id}] VM saved state detected (snapshot name: {snapshot})'.format( name=self._name, id=self.id, snapshot=snapshot_name ) diff --git a/gns3server/compute/ubridge/hypervisor.py b/gns3server/compute/ubridge/hypervisor.py index 89d815100..661da580c 100644 --- a/gns3server/compute/ubridge/hypervisor.py +++ b/gns3server/compute/ubridge/hypervisor.py @@ -180,15 +180,15 @@ class Hypervisor(UBridgeHypervisor): await self._check_ubridge_version(env) try: command = self._build_command() - log.info(f"starting ubridge: {command}") + log.debug(f"starting ubridge: {command}") self._stdout_file = os.path.join(self._working_dir, "ubridge.log") - log.info(f"logging to {self._stdout_file}") + log.debug(f"logging to {self._stdout_file}") with open(self._stdout_file, "w", encoding="utf-8") as fd: self._process = await asyncio.create_subprocess_exec( *command, stdout=fd, stderr=subprocess.STDOUT, cwd=self._working_dir, env=env ) - log.info(f"ubridge started PID={self._process.pid}") + log.debug(f"ubridge started PID={self._process.pid}") # An unsupported flag (e.g. -U on an old ubridge build) makes ubridge exit # immediately with a non-zero code. Detect that here and surface the real # reason from ubridge.log instead of waiting for connect() to time out with @@ -220,7 +220,7 @@ class Hypervisor(UBridgeHypervisor): log.error(error_msg) self._project.emit("log.error", {"message": error_msg}) else: - log.info("uBridge process has stopped, return code: %d", returncode) + log.debug("uBridge process has stopped, return code: %d", returncode) async def stop(self): """ @@ -228,7 +228,7 @@ class Hypervisor(UBridgeHypervisor): """ if self.is_running(): - log.info(f"Stopping uBridge process PID={self._process.pid}") + log.debug(f"Stopping uBridge process PID={self._process.pid}") await UBridgeHypervisor.stop(self) try: await wait_for_process_termination(self._process, timeout=3) diff --git a/gns3server/compute/ubridge/ubridge_hypervisor.py b/gns3server/compute/ubridge/ubridge_hypervisor.py index 83f765d16..c010e1e2d 100644 --- a/gns3server/compute/ubridge/ubridge_hypervisor.py +++ b/gns3server/compute/ubridge/ubridge_hypervisor.py @@ -89,7 +89,7 @@ class UBridgeHypervisor: if not connection_success: raise UbridgeError(f"Couldn't connect to hypervisor on {self.endpoint} :{last_exception}") else: - log.info(f"Connected to uBridge hypervisor on {self.endpoint} after {time.time() - begin:.4f} seconds") + log.debug(f"Connected to uBridge hypervisor on {self.endpoint} after {time.time() - begin:.4f} seconds") try: await asyncio.sleep(0.1) diff --git a/gns3server/controller/compute.py b/gns3server/controller/compute.py index 7ec8691a8..dbd54c151 100644 --- a/gns3server/controller/compute.py +++ b/gns3server/controller/compute.py @@ -98,6 +98,10 @@ class Compute: self.name = name # Cache of interfaces on remote host self._interfaces_cache = None + # Cached resolution of self._host — socket.gethostbyname is a blocking + # call; resolving it on every host_ip access (several times per link + # via get_ip_on_same_subnet) freezes the event loop for all coroutines. + self._host_ip_cache = None self._connection_failure = 0 def _session(self): @@ -218,14 +222,17 @@ class Compute: """ Return the IP associated to the host """ - try: - return socket.gethostbyname(self._host) - except socket.gaierror: - return "0.0.0.0" + if self._host_ip_cache is None: + try: + self._host_ip_cache = socket.gethostbyname(self._host) + except socket.gaierror: + self._host_ip_cache = "0.0.0.0" + return self._host_ip_cache @host.setter def host(self, host): self._host = host + self._host_ip_cache = None # invalidate; re-resolve on next access if self._console_host is None: self._console_host = host diff --git a/gns3server/controller/link.py b/gns3server/controller/link.py index e716bfd03..d27fe80e8 100644 --- a/gns3server/controller/link.py +++ b/gns3server/controller/link.py @@ -114,7 +114,7 @@ class Link: """ return self._markers - async def inherit_marker(self, def_name, marker_def, dump=True): + async def inherit_marker(self, def_name, marker_def, dump=True, memory_only=False): """ Apply a project-level marker definition to this link. @@ -148,6 +148,7 @@ class Link: enabled=not marker_def.get("paused", False), inherited_from=def_name, dump=dump, + memory_only=memory_only, ) def _persist_markers(self): @@ -257,11 +258,14 @@ class Link: """ return self._created - async def add_node(self, node, adapter_number, port_number, label=None, dump=True): + async def add_node(self, node, adapter_number, port_number, label=None, dump=True, batch=False): """ Add a node to the link :param dump: Dump project on disk + :param batch: When True, do not create the link on the computes once + both nodes are attached — the caller drives creation via the + project-open bulk path. Used to avoid one HTTP round-trip per link. """ port = node.get_port(adapter_number, port_number) @@ -305,7 +309,7 @@ class Link: {"node": node, "adapter_number": adapter_number, "port_number": port_number, "port": port, "label": label} ) - if len(self._nodes) == 2: + if len(self._nodes) == 2 and not batch: await self.create() for n in self._nodes: n["node"].add_link(self) diff --git a/gns3server/controller/notification.py b/gns3server/controller/notification.py index 3a63d9b87..d3540a746 100644 --- a/gns3server/controller/notification.py +++ b/gns3server/controller/notification.py @@ -31,6 +31,7 @@ class Notification: self._controller = controller self._project_listeners = {} + self._project_marker_listeners = {} self._controller_listeners = set() @contextmanager @@ -49,6 +50,26 @@ class Notification: finally: self._project_listeners[project_id].remove(queue) + @contextmanager + def project_marker_queue(self, project_id): + """ + Get a queue of marker notifications (marker.match etc.) for a project. + + Marker events are delivered on this dedicated channel instead of the + main project queue, so high-frequency marker.matches do not cause + head-of-line blocking for topology events (node.*/link.*). + + Use it with Python with + """ + + queue = NotificationQueue() + self._project_marker_listeners.setdefault(project_id, set()) + self._project_marker_listeners[project_id].add(queue) + try: + yield queue + finally: + self._project_marker_listeners[project_id].remove(queue) + @contextmanager def controller_queue(self): """ @@ -104,6 +125,8 @@ class Notification: elif action == "ping": event["compute_id"] = compute_id self.project_emit(action, event) + elif action.startswith("marker."): + self.marker_emit(action, event, project_id) else: self.project_emit(action, event, project_id) @@ -120,6 +143,25 @@ class Notification: else: self._send_event_to_all_projects(action, event) + def marker_emit(self, action, event, project_id): + """ + Send a marker notification (e.g. marker.match) to clients listening on + the dedicated marker channel for this project. Marker events are kept + off the main project queue on purpose, to avoid head-of-line blocking + from high-frequency matches. + + :param action: Action name + :param event: Event to send + :param project_id: Project id the marker belongs to + """ + + try: + marker_listeners = self._project_marker_listeners[project_id] + except KeyError: + return + for listener in marker_listeners: + asyncio.get_running_loop().call_soon_threadsafe(listener.put_nowait, (action, event, {})) + def _send_event_to_project(self, project_id, action, event): """ Send an event to all the client listening for notifications for diff --git a/gns3server/controller/project.py b/gns3server/controller/project.py index a6bcc159c..566e49047 100644 --- a/gns3server/controller/project.py +++ b/gns3server/controller/project.py @@ -160,6 +160,11 @@ class Project: self.dump() self._iou_id_lock = asyncio.Lock() + # Serialise the "ensure project exists on this compute" check in + # _create_node: without it, concurrent node creations all pass the + # `compute not in _project_created_on_compute` check before any has + # registered, and each fires a redundant POST /projects at the compute. + self._create_node_lock = asyncio.Lock() self._preallocated_udp_ports = {} # compute_id -> list of pre-allocated UDP ports log.debug(f'Project "{self.name}" [{self._id}] loaded') self.emit_controller_notification("project.created", self.asdict()) @@ -586,15 +591,21 @@ class Project: async def _create_node(self, compute, name, node_id, node_type=None, **kwargs): node = Node(self, compute, name, node_id=node_id, node_type=node_type, **kwargs) - if compute not in self._project_created_on_compute: - if compute.id == "local": - data = {"name": self._name, "project_id": self._id, "path": self._path} - else: - data = {"name": self._name, "project_id": self._id} - if self._variables: - data["variables"] = self._variables - await compute.post("/projects", data=data) - self._project_created_on_compute.add(compute) + # Hold the lock across the check + POST + register so that concurrent + # node creations on the same compute don't all race past the check and + # each POST /projects (the compute-side sync handler then instantiated + # the Project N times). Once one creation registers the compute, the + # rest see it in the set and return immediately. + async with self._create_node_lock: + if compute not in self._project_created_on_compute: + if compute.id == "local": + data = {"name": self._name, "project_id": self._id, "path": self._path} + else: + data = {"name": self._name, "project_id": self._id} + if self._variables: + data["variables"] = self._variables + await compute.post("/projects", data=data) + self._project_created_on_compute.add(compute) await node.create() self._nodes[node.id] = node @@ -827,6 +838,94 @@ class Project: # a link should have 2 attached nodes, this can happen with corrupted projects await self.delete_link(link.id, force_delete=True) + async def _prepare_link_from_topology(self, link_data): + """ + Build a link locally from topology data WITHOUT dispatching NIOs to the + computes. Returns ``(link, entries)`` where ``entries`` is the list of + ``(node, adapter_number, port_number, nio_data)`` tuples produced by + ``UDPLink._prepare()``, or ``None`` if the link is invalid/incomplete. + + Used by the project-open bulk path so all NIOs can be sent in a single + batch HTTP call per compute instead of one round-trip per link. + """ + + link = await self.add_link(link_id=link_data["link_id"], dump=False) + if "filters" in link_data: + try: + await link.update_filters(link_data["filters"]) + except ControllerError as e: + log.warning("Dropping invalid filters on link %s: %s", link_data.get("link_id"), e) + for name, marker in (link_data.get("markers") or {}).items(): + bpf = marker.get("bpf") + if not bpf: + log.warning("Dropping marker %s on link %s: missing bpf", name, link_data.get("link_id")) + continue + result = validate_bpf_syntax(bpf) + if not result.get("valid"): + log.warning( + "Dropping marker %s on link %s: invalid BPF (%s)", + name, link_data.get("link_id"), result.get("error") + ) + continue + link._markers[name] = { + "bpf": bpf, + "tag": marker.get("tag"), + "enabled": marker.get("enabled", True), + "color": marker.get("color"), + "highlight_duration": marker.get("highlight_duration"), + "capture_node_id": marker.get("capture_node_id"), + "direction": marker.get("direction"), + } + # Set style/icon directly: the update_* helpers unconditionally dump + # the whole topology and emit "link.updated", neither of which is + # appropriate mid-prepare (the link is finalised, notified and the + # project dumped once at the end of open). + if "link_style" in link_data: + link._link_style = link_data["link_style"] + if "show_filters_icon" in link_data: + link._show_filters_icon = link_data["show_filters_icon"] + for node_link in link_data.get("nodes", []): + node = self.get_node(node_link["node_id"]) + port = node.get_port(node_link["adapter_number"], node_link["port_number"]) + if port is None: + log.warning( + "Port {}/{} for {} not found".format( + node_link["adapter_number"], node_link["port_number"], node.name + ) + ) + continue + if port.link is not None: + log.warning( + "Port {}/{} is already connected to link ID {}".format( + node_link["adapter_number"], node_link["port_number"], port.link.id + ) + ) + continue + # batch=True: attach the node without triggering per-link NIO HTTP + await link.add_node( + node, + node_link["adapter_number"], + node_link["port_number"], + label=node_link.get("label"), + dump=False, + batch=True, + ) + if len(link.nodes) != 2: + # a link should have 2 attached nodes, this can happen with corrupted projects + await self.delete_link(link.id, force_delete=True) + return None + # Apply project-level marker definitions onto the link's memory + # (memory_only) before _prepare() so the inherited markers ride the + # batch NIO dispatch — zero extra HTTP round-trips. The final + # apply_defs_to_new_link in finalize is removed. + for def_name, d in self._marker_definitions.items(): + try: + await link.inherit_marker(def_name, d, dump=False, memory_only=True) + except ControllerError as e: + log.warning("Marker definition '%s' could not be applied to link %s: %s", def_name, link.id, e) + entries = await link._prepare() + return (link, entries) + @open_required async def add_link(self, link_id=None, dump=True): """ @@ -1068,23 +1167,25 @@ class Project: # data_link_type decides which links host an inherited copy (serial # links are skipped unless a WAN encapsulation is chosen), so a change # needs a full re-fan-out: drop every copy, then re-apply. - await self._marker_apply_concurrently( - affected, - lambda link: link.stop_marker(f"global-{name}", inherited=True, dump=False), - lambda link, e: f"Failed to remove inherited marker global-{name} from link {link.id}: {e}", - ) + for link in affected: + try: + await link.stop_marker(f"global-{name}", inherited=True, dump=False, memory_only=True) + except ControllerError as e: + log.warning("Failed to remove inherited marker global-%s from link %s: %s", name, link.id, e) await self._apply_def_to_all_links(name) else: - # Sync: update every inherited copy across all links. - await self._marker_apply_concurrently( - affected, - lambda link: link.update_marker( - f"global-{name}", bpf=d["bpf"], tag=d.get("tag"), direction=d.get("direction"), - color=d.get("color"), highlight_duration=d.get("highlight_duration"), inherited=True, - dump=False - ), - lambda link, e: f"Failed to sync marker global-{name} on link {link.id}: {e}", - ) + # Sync: update every inherited copy across all links in memory, then + # batch-push to computes (one PUT /nios/batch per compute). + for link in affected: + try: + await link.update_marker( + f"global-{name}", bpf=d["bpf"], tag=d.get("tag"), direction=d.get("direction"), + color=d.get("color"), highlight_duration=d.get("highlight_duration"), inherited=True, + dump=False, memory_only=True + ) + except ControllerError as e: + log.warning("Failed to sync marker global-%s on link %s: %s", name, link.id, e) + await self._batch_update_link_nios(affected) self.dump() self.emit_notification("project.updated", self.asdict()) @@ -1105,12 +1206,13 @@ class Project: if f"global-{name}" in link.markers and link.markers[f"global-{name}"].get("inherited_from") == name ] - await self._marker_apply_concurrently( - affected, - lambda link: link.stop_marker(f"global-{name}", inherited=True), - # A missing compute or broken link shouldn't block the delete. - lambda link, e: f"Failed to remove inherited marker global-{name} from link {link.id}: {e}", - ) + for link in affected: + try: + await link.stop_marker(f"global-{name}", inherited=True, memory_only=True) + except ControllerError as e: + # A missing compute or broken link shouldn't block the delete. + log.warning("Failed to remove inherited marker global-%s from link %s: %s", name, link.id, e) + await self._batch_update_link_nios(affected) self.dump() self.emit_notification("project.updated", self.asdict()) @@ -1120,17 +1222,57 @@ class Project: Fan out a single marker definition to every existing link in the project. Links that have no capable node (``_MARKER_CAPABLE_TYPES``) are silently skipped — the marker can only live on a uBridge bridge. + + Two-phase to avoid one HTTP round-trip per link end: (1) write the + inherited marker into each link's memory (``memory_only`` refreshes + ``_link_data`` without pushing), then (2) batch-update every affected + NIO via a single ``PUT /projects/{id}/nios/batch`` per compute. """ d = self._marker_definitions[def_name] - # dump=False: per-link topology writes are the dominant cost on large - # projects — the callers (create/update_marker_definition) dump once - # after the fan-out. - await self._marker_apply_concurrently( - list(self._links.values()), - lambda link: link.inherit_marker(def_name, d, dump=False), - lambda link, e: f"Marker definition '{def_name}' could not be applied to link {link.id}: {e}", - ) + affected = [] + for link in self._links.values(): + try: + await link.inherit_marker(def_name, d, dump=False, memory_only=True) + affected.append(link) + except ControllerError as e: + log.warning("Marker definition '%s' could not be applied to link %s: %s", def_name, link.id, e) + await self._batch_update_link_nios(affected) + + async def _batch_update_link_nios(self, links): + """ + Push the current ``_link_data`` (markers/filters) of *links* to their + computes in one ``PUT /projects/{id}/nios/batch`` per compute — replacing + one PUT /nio round-trip per link end. Started nodes re-apply uBridge; + stopped nodes update in memory. + """ + + per_compute = {} + for link in links: + if len(link._link_data) < 2: + continue + for i, side in enumerate(link._nodes): + node = side["node"] + per_compute.setdefault(node.compute, []).append( + { + "node_id": node.id, + "adapter_number": side["adapter_number"], + "port_number": side["port_number"], + "nio": link._link_data[i], + } + ) + + async def _dispatch(compute, entries): + await compute.put( + f"/projects/{self._id}/nios/batch", + data={"nios": entries}, + timeout=300, + ) + + if per_compute: + await asyncio.gather( + *[_dispatch(c, n) for c, n in per_compute.items()] + ) async def apply_defs_to_new_link(self, link): """ @@ -1170,6 +1312,14 @@ class Project: :param fail_msg: callable ``(link, error) -> log message`` """ + links = list(links) + if not links: + return + _t0 = time.time() + log.info( + "Project '%s' [%s]: fanning out marker operation to %d links...", + self._name, self._id, len(links) + ) sem = asyncio.Semaphore(32) async def guarded(link): @@ -1180,6 +1330,10 @@ class Project: log.warning(fail_msg(link, e)) await asyncio.gather(*(guarded(link) for link in links)) + log.info( + "Project '%s' [%s]: marker fan-out done in %.2fs", + self._name, self._id, time.time() - _t0 + ) @property def snapshots(self): @@ -1282,6 +1436,7 @@ class Project: log.warning(f"Closing project '{self.name}' ignored because it is being loaded") return self._closing = True + log.info("Project '%s' [%s]: closing...", self._name, self._id) try: await self.stop_all() except HTTPException as e: @@ -1305,6 +1460,7 @@ class Project: self.reset() self._closing = False + log.info("Project '%s' [%s]: closed", self._name, self._id) def _clean_pictures(self): """ @@ -1631,10 +1787,12 @@ class Project: # Create nodes in parallel with limited concurrency # to avoid overwhelming the system with too many simultaneous operations + log.info("Project '%s' [%s]: loading %d nodes...", self._name, self._id, len(nodes_to_create)) pool = Pool(concurrency=100) for compute, name, node_id, node_data in nodes_to_create: pool.append(self.add_node, compute, name, node_id, dump=False, **node_data) await pool.join() + log.info("Project '%s' [%s]: loaded %d nodes", self._name, self._id, len(nodes_to_create)) # Pre-allocate UDP ports for all links in batch to reduce HTTP round-trips ports_per_compute = {} for link_data in topology.get("links", []): @@ -1648,13 +1806,60 @@ class Project: count = ports_per_compute.get(compute.id, 0) if count > 0: await self.preallocate_udp_ports_for_compute(compute, count) - # Create links in parallel for improved performance - pool = Pool(concurrency=100) - for link_data in topology.get("links", []): - if "link_id" not in link_data.keys(): - continue - pool.append(self._create_link_from_topology_data, link_data) - await pool.join() + # Create links via the bulk path: build every link locally (no NIO + # HTTP), then dispatch all NIOs to each compute in a single batch + # request. This replaces one HTTP round-trip per link (~5000 for a + # 2500-link topology) with one round-trip per compute. + link_data_list = [d for d in topology.get("links", []) if "link_id" in d.keys()] + log.info("Project '%s' [%s]: creating %d links...", self._name, self._id, len(link_data_list)) + sem = asyncio.Semaphore(100) + + async def _prepare_one(data): + async with sem: + try: + return await self._prepare_link_from_topology(data) + except Exception as e: + log.warning("Could not load link %s: %s", data.get("link_id"), e) + return None + + prepared = await asyncio.gather(*[_prepare_one(d) for d in link_data_list]) + valid = [p for p in prepared if p is not None] + + # Group the prepared NIO entries by destination compute and send + # each compute a single /nios/batch request. + per_compute = {} # compute -> list of {node_id, adapter_number, port_number, nio} + for link, entries in valid: + for node, adapter_number, port_number, nio_data in entries: + per_compute.setdefault(node.compute, []).append( + { + "node_id": node.id, + "adapter_number": adapter_number, + "port_number": port_number, + "nio": nio_data, + } + ) + + async def _dispatch_batch(compute, nio_entries): + await compute.post( + f"/projects/{self._id}/nios/batch", + data={"nios": nio_entries}, + timeout=300, + ) + + if per_compute: + await asyncio.gather( + *[_dispatch_batch(c, n) for c, n in per_compute.items()] + ) + + # Finalise every link: wire node/port back-references, mark created, + # notify clients, and apply project-level marker definitions. + for link, _entries in valid: + for n in link._nodes: + n["node"].add_link(link) + n["port"].link = link + link._created = True + self.emit_notification("link.created", link.asdict()) + log.info("Project '%s' [%s]: created %d links", self._name, self._id, len(valid)) # Release any pre-allocated UDP ports that were not consumed by links for compute_id, ports in self._preallocated_udp_ports.items(): if ports: @@ -1929,29 +2134,37 @@ class Project: """ Start all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.) """ - pool = Pool(concurrency=3) - for node in self.nodes.values(): - if not node.is_always_running(): - pool.append(node.start) + nodes_to_start = [n for n in self.nodes.values() if not n.is_always_running()] + if not nodes_to_start: + return + log.info("Project '%s' [%s]: starting %d nodes...", self._name, self._id, len(nodes_to_start)) + pool = Pool(concurrency=10) + for node in nodes_to_start: + pool.append(node.start) await pool.join() + log.info("Project '%s' [%s]: started %d nodes", self._name, self._id, len(nodes_to_start)) @open_required async def stop_all(self): """ Stop all nodes (except always-running types like Ethernet switch, Cloud, NAT, etc.) """ - pool = Pool(concurrency=3) - for node in self.nodes.values(): - if not node.is_always_running(): - pool.append(node.stop) + nodes_to_stop = [n for n in self.nodes.values() if not n.is_always_running()] + if not nodes_to_stop: + return + log.info("Project '%s' [%s]: stopping %d nodes...", self._name, self._id, len(nodes_to_stop)) + pool = Pool(concurrency=100) + for node in nodes_to_stop: + pool.append(node.stop) await pool.join() + log.info("Project '%s' [%s]: stopped %d nodes", self._name, self._id, len(nodes_to_stop)) @open_required async def suspend_all(self): """ Suspend all nodes """ - pool = Pool(concurrency=3) + pool = Pool(concurrency=50) for node in self.nodes.values(): pool.append(node.suspend) await pool.join() diff --git a/gns3server/controller/udp_link.py b/gns3server/controller/udp_link.py index b25e1126b..86f863010 100644 --- a/gns3server/controller/udp_link.py +++ b/gns3server/controller/udp_link.py @@ -16,6 +16,9 @@ # along with this program. If not, see . +import asyncio +import logging + from .controller_error import ControllerError, ControllerNotFoundError from .link import Link, _UNSET from .node_types import BUILTIN_NODE_TYPES @@ -30,6 +33,9 @@ _MARKER_CAPABLE_TYPES = frozenset({ }) +log = logging.getLogger(__name__) + + class UDPLink(Link): def __init__(self, project, link_id=None): super().__init__(project, link_id=link_id) @@ -81,9 +87,15 @@ class UDPLink(Link): """ return self._markers_for_node(node1), self._markers_for_node(node2) - async def create(self): + async def _prepare(self): """ - Create the link on the nodes + Local-only link setup: resolve peer addresses, reserve UDP ports and + build the two NIO tunnel specs (``self._link_data``). No NIO is sent to + the computes — the caller decides how to dispatch them (one-by-one via + :meth:`create`, or batched via the project-open bulk path). + + :returns: list of two ``(node, adapter_number, port_number, nio_data)`` + tuples, ready to be POSTed to each node's compute. """ node1 = self._nodes[0]["node"] @@ -99,25 +111,25 @@ class UDPLink(Link): except ValueError as e: raise ControllerError(f"Cannot get an IP address on same subnet: {e}") - # Reserve a UDP port on both side - # Try pre-allocated ports first (used during batch project loading) - port = self._project.pop_preallocated_udp_port(node1.compute.id) - if port is not None: - self._node1_port = port - else: - response = await node1.compute.post(f"/projects/{self._project.id}/ports/udp") - self._node1_port = response.json["udp_port"] - port = self._project.pop_preallocated_udp_port(node2.compute.id) - if port is not None: - self._node2_port = port - else: - response = await node2.compute.post(f"/projects/{self._project.id}/ports/udp") - self._node2_port = response.json["udp_port"] + # Reserve a UDP port on both sides in parallel. Pre-allocated ports + # (used during batch project loading) are popped from memory; otherwise + # each side falls back to a single HTTP round-trip to its compute. + async def _allocate_port(compute): + port = self._project.pop_preallocated_udp_port(compute.id) + if port is not None: + return port + response = await compute.post(f"/projects/{self._project.id}/ports/udp") + return response.json["udp_port"] + + self._node1_port, self._node2_port = await asyncio.gather( + _allocate_port(node1.compute), _allocate_port(node2.compute) + ) node1_filters, node2_filters = self._get_node_filters(node1, node2) node1_markers, node2_markers = self._get_node_markers(node1, node2) - # Create the tunnel on both side + # Build the tunnel specs for both sides. Index 0 is always node1 so + # that update()/delete() keep addressing self._link_data[0]/[1]. self._link_data.append( { "lport": self._node1_port, @@ -129,8 +141,6 @@ class UDPLink(Link): "suspend": self._suspended, } ) - await node1.post(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=self._link_data[0], timeout=120) - self._link_data.append( { "lport": self._node2_port, @@ -142,15 +152,59 @@ class UDPLink(Link): "suspend": self._suspended, } ) - try: - await node2.post( - f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=self._link_data[1], timeout=120 - ) - except Exception as e: - # We clean the first NIO - await node1.delete(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", timeout=120) - raise e + + return [ + (node1, adapter_number1, port_number1, self._link_data[0]), + (node2, adapter_number2, port_number2, self._link_data[1]), + ] + + async def _commit_nios(self, entries): + """ + Send the two NIO tunnel POSTs in parallel and roll back on failure. + + :param entries: the two ``(node, adapter_number, port_number, nio_data)`` + tuples returned by :meth:`_prepare`. + """ + + (node1, adapter_number1, port_number1, nio_data1), \ + (node2, adapter_number2, port_number2, nio_data2) = entries + + # The two ends are independent once the ports and peer addresses are + # known — each node talks to its own compute/uBridge with no shared + # lock between them — so the two POSTs overlap. If either fails, roll + # back whichever side succeeded before re-raising the first error. + results = await asyncio.gather( + node1.post( + f"/adapters/{adapter_number1}/ports/{port_number1}/nio", data=nio_data1, timeout=120 + ), + node2.post( + f"/adapters/{adapter_number2}/ports/{port_number2}/nio", data=nio_data2, timeout=120 + ), + return_exceptions=True, + ) + errors = [result for result in results if isinstance(result, Exception)] + if errors: + cleanup = [] + if not isinstance(results[0], Exception): + cleanup.append( + node1.delete(f"/adapters/{adapter_number1}/ports/{port_number1}/nio", timeout=120) + ) + if not isinstance(results[1], Exception): + cleanup.append( + node2.delete(f"/adapters/{adapter_number2}/ports/{port_number2}/nio", timeout=120) + ) + if cleanup: + await asyncio.gather(*cleanup, return_exceptions=True) + raise errors[0] self._created = True + + async def create(self): + """ + Create the link on the nodes (interactive path: prepare + commit). + """ + + entries = await self._prepare() + await self._commit_nios(entries) # New links automatically inherit every active project-level marker # definition so the user doesn't have to reconfigure. await self._project.apply_defs_to_new_link(self) @@ -353,7 +407,7 @@ class UDPLink(Link): # explicitly deletes a marker via the REST API, and a marker is torn # down automatically only when its link is deleted. - async def start_marker(self, name, bpf, tag=None, direction=None, data_link_type="DLT_EN10MB", capture_node_id=None, color=None, highlight_duration=None, enabled=True, inherited_from=None, dump=True): + async def start_marker(self, name, bpf, tag=None, direction=None, data_link_type="DLT_EN10MB", capture_node_id=None, color=None, highlight_duration=None, enabled=True, inherited_from=None, dump=True, memory_only=False): """ Attach a traffic-insight marker to this link. @@ -409,6 +463,12 @@ class UDPLink(Link): if inherited_from: marker_entry["inherited_from"] = inherited_from self._markers[name] = marker_entry + if memory_only: + # Project-open prepare / marker-def fan-out: only refresh the + # in-memory NIO specs so a later batch dispatch carries the new + # markers — no per-link update HTTP, notification or dump. + self._refresh_link_data() + return if self._created: await self.update() self._project.emit_notification("link.updated", self.asdict()) @@ -417,7 +477,27 @@ class UDPLink(Link): if dump: self._project.dump() - async def stop_marker(self, name, inherited=False, dump=True): + def _refresh_link_data(self): + """ + Recompute the filters / markers / suspend fields of ``_link_data`` from + the current link state without pushing to computes. Used by the + memory-only marker path so a batch dispatch picks up the new markers. + """ + + if len(self._link_data) < 2: + return + node1 = self._nodes[0]["node"] + node2 = self._nodes[1]["node"] + node1_filters, node2_filters = self._get_node_filters(node1, node2) + node1_markers, node2_markers = self._get_node_markers(node1, node2) + self._link_data[0]["filters"] = node1_filters + self._link_data[0]["markers"] = node1_markers + self._link_data[0]["suspend"] = self._suspended + self._link_data[1]["filters"] = node2_filters + self._link_data[1]["markers"] = node2_markers + self._link_data[1]["suspend"] = self._suspended + + async def stop_marker(self, name, inherited=False, dump=True, memory_only=False): """ Remove a traffic-insight marker from this link. @@ -442,6 +522,12 @@ class UDPLink(Link): capture_node_id = self._markers[name].get("capture_node_id") del self._markers[name] + if memory_only: + # Project-level def-delete fan-out: marker is gone from _markers; + # refresh _link_data so the batch dispatch drops it from uBridge + # via full reapply. No per-link delete round-trip, notification or dump. + self._refresh_link_data() + return # Remove the marker filter + its pcap on the capture node directly — NOT a # full NIO reapply (which would reset_packet_filters and close/reopen every # sibling marker's pcap). delete_packet_filter removes just this filter; @@ -461,7 +547,7 @@ class UDPLink(Link): if dump: self._project.dump() - async def update_marker(self, name, bpf=None, tag=None, enabled=None, direction=_UNSET, color=None, highlight_duration=None, inherited=False, dump=True): + async def update_marker(self, name, bpf=None, tag=None, enabled=None, direction=_UNSET, color=None, highlight_duration=None, inherited=False, dump=True, memory_only=False): """ Update an existing marker's fields and push to uBridge fine-grained — no full NIO reapply, so sibling markers' pcaps stay open. bpf/tag/direction @@ -510,6 +596,13 @@ class UDPLink(Link): if direction is not _UNSET: marker_info["direction"] = direction # None = clear back to both directions + if memory_only: + # Project-level def sync fan-out: state is already merged into + # _markers; just refresh _link_data so the batch dispatch carries + # it. No per-link uBridge rebuild, notification or dump. + self._refresh_link_data() + return + # Push to uBridge fine-grained — NO full NIO reapply (which would # reset_packet_filters and close/reopen every sibling marker's pcap): # * bpf/tag/direction changed → rebuild just this filter (delete + add), diff --git a/gns3server/schemas/__init__.py b/gns3server/schemas/__init__.py index af026c241..7d4ba6c22 100644 --- a/gns3server/schemas/__init__.py +++ b/gns3server/schemas/__init__.py @@ -91,7 +91,7 @@ from .controller.templates.dynamips_templates import ( ) # Compute schemas -from .compute.nios import UDPNIO, TAPNIO, EthernetNIO, MarkerToggle, MarkerRebuild +from .compute.nios import UDPNIO, TAPNIO, EthernetNIO, MarkerToggle, MarkerRebuild, BatchNIOEntry, BatchNIOCreate from .compute.atm_switch_nodes import ATMSwitchCreate, ATMSwitchUpdate, ATMSwitch from .compute.cloud_nodes import CloudCreate, CloudUpdate, Cloud from .compute.docker_nodes import DockerCreate, DockerUpdate, Docker diff --git a/gns3server/schemas/compute/nios.py b/gns3server/schemas/compute/nios.py index 5830847c5..a1d8641b3 100644 --- a/gns3server/schemas/compute/nios.py +++ b/gns3server/schemas/compute/nios.py @@ -16,7 +16,7 @@ from pydantic import BaseModel, Field -from typing import Optional +from typing import Optional, List from enum import Enum @@ -91,3 +91,24 @@ class MarkerRebuild(BaseModel): direction: Optional[str] = None enabled: bool = True link_id: str = "" + + +class BatchNIOEntry(BaseModel): + """ + A single NIO binding to create as part of a project-wide batch. + """ + + node_id: str = Field(..., description="Node the NIO is attached to") + adapter_number: int = Field(0, ge=0, description="Adapter number") + port_number: int = Field(0, ge=0, description="Port number") + nio: UDPNIO = Field(..., description="NIO settings") + + +class BatchNIOCreate(BaseModel): + """ + Body for the project-wide batch NIO endpoint: create many NIO bindings in a + single request (used during project open) to avoid one HTTP round-trip per + NIO between controller and compute. + """ + + nios: List[BatchNIOEntry] = Field(..., description="NIO bindings to create") diff --git a/tests/api/routes/compute/test_docker_nodes.py b/tests/api/routes/compute/test_docker_nodes.py index ebbba442f..13e4609b7 100644 --- a/tests/api/routes/compute/test_docker_nodes.py +++ b/tests/api/routes/compute/test_docker_nodes.py @@ -211,6 +211,34 @@ class TestDockerNodesRoutes: assert response.status_code == status.HTTP_201_CREATED assert response.json()["type"] == "nio_udp" + async def test_docker_nio_batch_create(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: + """ + Exercise the project-wide batch NIO endpoint: bind two NIOs on the same + docker node in a single request (the path used during project open). + """ + + params = { + "nios": [ + { + "node_id": vm["node_id"], + "adapter_number": 0, + "port_number": 0, + "nio": {"type": "nio_udp", "lport": 4242, "rport": 4343, "rhost": "127.0.0.1"}, + }, + { + "node_id": vm["node_id"], + "adapter_number": 1, + "port_number": 0, + "nio": {"type": "nio_udp", "lport": 4243, "rport": 4344, "rhost": "127.0.0.1"}, + }, + ] + } + + url = app.url_path_for("compute:create_batch_nios", project_id=vm["project_id"]) + response = await compute_client.post(url, json=params) + assert response.status_code == status.HTTP_201_CREATED + assert response.json()["added"] == 2 + async def test_docker_update_nio(self, app: FastAPI, compute_client: AsyncClient, vm: dict) -> None: diff --git a/tests/api/routes/compute/test_projects.py b/tests/api/routes/compute/test_projects.py index 27474044b..949c0d891 100644 --- a/tests/api/routes/compute/test_projects.py +++ b/tests/api/routes/compute/test_projects.py @@ -224,3 +224,121 @@ class TestComputeProjectRoutes: project_id=project.id, file_path=file_path), content=b"world") assert response.status_code == status.HTTP_403_FORBIDDEN + + +class TestBatchNIOEdgeCases: + + @pytest.mark.asyncio + async def test_dynamips_router_dispatch_to_slot_add_nio_binding(self): + """_add_nio_binding dispatches Dynamips router to slot_add_nio_binding.""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "Dynamips" + node.slot_add_nio_binding = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 0, 0, nio) + node.slot_add_nio_binding.assert_called_once_with(0, 0, nio) + + @pytest.mark.asyncio + async def test_dynamips_switch_dispatch_to_add_nio(self): + """_add_nio_binding dispatches Dynamips switch to add_nio.""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "Dynamips" + del node.slot_add_nio_binding # no slot_add_nio → switch path + node.add_nio = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 0, 0, nio) + node.add_nio.assert_called_once_with(nio, 0) + + @pytest.mark.asyncio + async def test_dynamips_create_nio_is_async_and_needs_await(self): + """ + Dynamips.create_nio is async (returns a coroutine) unlike the sync + base version. The batch handler must await it. + """ + import inspect + import asyncio as _asyncio + + class _FakeDynamips: + async def create_nio(self, node, nio_settings): + return {"type": "nio_udp", "node": node} + + class _FakeBase: + def create_nio(self, nio_settings): + return {"type": "nio_udp"} + + dyn = _FakeDynamips() + base = _FakeBase() + assert len(inspect.signature(dyn.create_nio).parameters) == 2 # Dynamips + assert len(inspect.signature(base.create_nio).parameters) == 1 # standard + assert inspect.iscoroutinefunction(dyn.create_nio) + assert not inspect.iscoroutinefunction(base.create_nio) + + # Verify the batch logic: 2 params → await, 1 param → no await + d_result = await dyn.create_nio("r1", {"type": "nio_udp"}) + b_result = base.create_nio({"type": "nio_udp"}) + assert d_result["node"] == "r1" + assert b_result["type"] == "nio_udp" + + @pytest.mark.asyncio + async def test_qemu_dispatch_to_adapter_add_nio_binding(self): + """_add_nio_binding dispatches Qemu to adapter_add_nio_binding.""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "Qemu" + node.adapter_add_nio_binding = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 0, 0, nio) + node.adapter_add_nio_binding.assert_called_once_with(0, nio) + + @pytest.mark.asyncio + async def test_iou_dispatch_to_adapter_add_nio_binding(self): + """_add_nio_binding dispatches IOU to adapter_add_nio_binding(adapter, port, nio).""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "IOU" + node.adapter_add_nio_binding = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 1, 2, nio) + node.adapter_add_nio_binding.assert_called_once_with(1, 2, nio) + + @pytest.mark.asyncio + async def test_vpcs_dispatch_to_port_add_nio_binding(self): + """_add_nio_binding dispatches VPCS to port_add_nio_binding.""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "VPCS" + node.port_add_nio_binding = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 0, 3, nio) + node.port_add_nio_binding.assert_called_once_with(3, nio) + + @pytest.mark.asyncio + async def test_builtin_dispatch_to_add_nio(self): + """_add_nio_binding dispatches Builtin nodes to add_nio(nio, port).""" + from unittest.mock import AsyncMock, MagicMock + from gns3server.api.routes.compute.projects import _add_nio_binding + + node = MagicMock() + type(node.manager).__name__ = "Builtin" + node.add_nio = AsyncMock() + nio = MagicMock() + + await _add_nio_binding(node, 0, 0, nio) + node.add_nio.assert_called_once_with(nio, 0) diff --git a/tests/compute/docker/test_docker_vm.py b/tests/compute/docker/test_docker_vm.py index bc722423e..c9e701306 100644 --- a/tests/compute/docker/test_docker_vm.py +++ b/tests/compute/docker/test_docker_vm.py @@ -1209,7 +1209,7 @@ async def test_stop(vm): with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query: vm._permissions_fixed = False await vm.stop() - mock_query.assert_called_with("POST", "containers/e90e34656842/stop", params={"t": 5}) + mock_query.assert_called_with("POST", "containers/e90e34656842/kill") assert mock.stop.called assert vm._ubridge_hypervisor is None assert vm._fix_permissions.called @@ -1222,7 +1222,7 @@ async def test_stop_paused_container(vm): with asyncio_patch("gns3server.compute.docker.DockerVM.unpause") as mock_unpause: with asyncio_patch("gns3server.compute.docker.Docker.query") as mock_query: await vm.stop() - mock_query.assert_called_with("POST", "containers/e90e34656842/stop", params={"t": 5}) + mock_query.assert_called_with("POST", "containers/e90e34656842/kill") assert mock_unpause.called @@ -1442,7 +1442,7 @@ async def test_add_ubridge_connection(vm): call.send('docker move_to_ns tap-gns3-e0 42 eth0'), call.send('bridge add_nio_udp bridge0 4242 127.0.0.1 4343'), call.send('bridge start_capture bridge0 "/tmp/capture.pcap"'), - call.send('bridge start bridge0') + call.send('bridge start bridge0'), ] assert 'bridge0' in vm._bridges # We need to check any_order otherwise mock is confused by asyncio @@ -1894,7 +1894,7 @@ async def test_stop_exited_container_no_stop_query(vm): vm._permissions_fixed = False await vm.stop() assert not any( - call.args[:2] == ("POST", "containers/e90e34656842/stop") + call.args[:2] == ("POST", "containers/e90e34656842/kill") for call in mock_query.mock_calls ) assert vm.status == "stopped" diff --git a/tests/compute/test_base_node.py b/tests/compute/test_base_node.py index 38371538d..fa8dce6a5 100644 --- a/tests/compute/test_base_node.py +++ b/tests/compute/test_base_node.py @@ -367,6 +367,79 @@ async def test_apply_markers_skips_already_installed(compute_project, manager): assert not any("add_packet_filter" in c for c in cmds) # skipped, not re-added +@pytest.mark.asyncio +async def test_apply_markers_deletes_removed(compute_project, manager): + # Reconcile: a marker no longer in nio.markers is deleted from uBridge + # (filter + registry), not left as an orphan still matching packets. + node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager) + node._ubridge_send = AsyncioMagicMock() + node._ubridge_hypervisor = MagicMock() + node._ubridge_hypervisor.is_running.return_value = True + node._marker_filter_bridges["m", "L1"] = "VPCS-10" + node._marker_specs["m", "L1"] = {"bpf": "icmp", "enabled": True} + nio = NIOUDP(1234, "127.0.0.1", 4321) + nio.markers = {} # marker removed + with patch("gns3server.compute.marker.marker_manager.MarkerManager") as mm: + mm.instance.return_value.unregister = MagicMock() + await node._ubridge_apply_markers("VPCS-10", nio) + cmds = [c.args[0] for c in node._ubridge_send.call_args_list] + assert any("delete_packet_filter VPCS-10 m" in c for c in cmds) + assert ("m", "L1") not in node._marker_filter_bridges + assert ("m", "L1") not in node._marker_specs + + +@pytest.mark.asyncio +async def test_apply_markers_rebuilds_changed_bpf(compute_project, manager): + # Reconcile: a marker whose bpf changed is rebuilt (delete + re-add), so + # uBridge ends up with the new expression — not the stale original. + node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager) + node._ubridge_send = AsyncioMagicMock() + node._ubridge_hypervisor = MagicMock() + node._ubridge_hypervisor.is_running.return_value = True + node._marker_filter_bridges["m", "L1"] = "VPCS-10" + node._marker_specs["m", "L1"] = {"bpf": "icmp", "tag": None, "direction": None, + "data_link_type": None, "enabled": True} + nio = NIOUDP(1234, "127.0.0.1", 4321) + nio.markers = {"m": {"bpf": "tcp", "tag": None, "link_id": "L1", + "direction": None, "enabled": True}} + with patch("gns3server.compute.marker.marker_manager.MarkerManager") as mm: + mm.instance.return_value.register = MagicMock() + await node._ubridge_apply_markers("VPCS-10", nio) + cmds = [c.args[0] for c in node._ubridge_send.call_args_list] + assert any("delete_packet_filter VPCS-10 m" in c for c in cmds) # old removed + assert any("add_packet_filter VPCS-10 m mark" in c and "tcp" in c for c in cmds) # new added + + +@pytest.mark.asyncio +async def test_apply_markers_preserves_markers_on_other_bridges(compute_project, manager): + # Regression: reconciling one NIO must not delete markers installed on this + # node's OTHER bridges/NIOs. _marker_filter_bridges is node-wide, but + # `desired` only carries the current NIO's markers — the delete pass must be + # scoped to the current bridge, or updating one link wipes every other link's + # markers + pcaps. + node = VPCSVM("test", "00010203-0405-0607-0809-0a0b0c0d0e0f", compute_project, manager) + node._ubridge_send = AsyncioMagicMock() + node._ubridge_hypervisor = MagicMock() + node._ubridge_hypervisor.is_running.return_value = True + # Two links, each with a marker on its own bridge: + node._marker_filter_bridges["m", "L1"] = "VPCS-10" + node._marker_specs["m", "L1"] = {"bpf": "icmp", "enabled": True} + node._marker_filter_bridges["m", "L2"] = "VPCS-20" + node._marker_specs["m", "L2"] = {"bpf": "tcp", "enabled": True} + # Update only the VPCS-10 NIO; "m" is gone from this link — but the marker + # on VPCS-20 (L2) must survive untouched. + nio = NIOUDP(1234, "127.0.0.1", 4321) + nio.markers = {} + with patch("gns3server.compute.marker.marker_manager.MarkerManager") as mm: + mm.instance.return_value.unregister = MagicMock() + await node._ubridge_apply_markers("VPCS-10", nio) + cmds = [c.args[0] for c in node._ubridge_send.call_args_list] + assert any("delete_packet_filter VPCS-10 m" in c for c in cmds) # L1 removed on its bridge + assert not any("VPCS-20" in c for c in cmds) # other bridge untouched + assert ("m", "L1") not in node._marker_filter_bridges + assert ("m", "L2") in node._marker_filter_bridges # L2 preserved + + @pytest.mark.asyncio async def test_stop_ubridge_clears_marker_bridges(compute_project, manager): # uBridge stopping drops every marker filter — the map must clear so the next diff --git a/tests/controller/test_notification.py b/tests/controller/test_notification.py index 9204acea3..754559154 100644 --- a/tests/controller/test_notification.py +++ b/tests/controller/test_notification.py @@ -120,6 +120,33 @@ async def test_dispatch_node_updated(controller, node, project): assert event["properties"]["startup_config"] == "ip 192" +@pytest.mark.asyncio +async def test_dispatch_marker_routed_to_marker_channel(controller, project): + """ + marker.* events are dispatched to the dedicated marker channel, not the + main project queue, so high-frequency matches cannot block topology events. + """ + + notif = controller.notification + with notif.project_queue(project.id) as project_q, \ + notif.project_marker_queue(project.id) as marker_q: + assert len(notif._project_marker_listeners[project.id]) == 1 + await project_q.get(0.1) # consume initial ping + await marker_q.get(0.1) # consume initial ping + + await notif.dispatch("marker.match", {"link_id": "abc"}, + project_id=project.id, compute_id=1) + + # marker.match lands on the marker channel... + msg = await marker_q.get(5) + assert msg == ('marker.match', {"link_id": "abc"}, {}) + # ...and does NOT land on the main project queue (times out -> ping) + msg = await project_q.get(0.1) + assert msg[0] == "ping" + + assert len(notif._project_marker_listeners[project.id]) == 0 + + def test_various_notification(controller, node): notif = controller.notification