log: add periodic marker.match throughput statistics

MarkerManager now logs every 10s how many marker datagrams the UDP sink
processed and the current throughput rate (match/s), so operators can
tell at a glance whether the single sink keeps up with the aggregated
uBridge traffic. Error count is also logged.

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)
 

gns3server/compute/marker/marker_manager.py

@@ -116,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
+            rate = received / 10.0 if received else 0
+            log.info(
+                "marker sink: %d matches (%.0f/s), %d errors in last 10s",
+                received, rate, errors,
+            )
+            listener._received = 0
+            listener._errors = 0
 
     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