feat: added metrics for swarm-connection cycle

Author: lla-dane

examples/metrics/coordinator.py

@@ -14,7 +14,6 @@
 from libp2p.pubsub.gossipsub import GossipSub
 from libp2p.pubsub.pubsub import Pubsub
 from libp2p.records.validator import Validator
-from libp2p.utils.paths import get_script_dir, join_paths
 
 GOSSIPSUB_PROTOCOL_ID = TProtocol("/meshsub/1.0.0")
 COMMANDS = """
@@ -24,7 +23,7 @@
 - join <topic>                      - Subscribe to a topic
 - leave <topic>                     - Unsubscribe to a topic
 - publish <topic> <message>         - Publish a message
-- put <key> <value>               - Execute PUT_VALUE in DHT
+- put <key> <value>                 - Execute PUT_VALUE in DHT
 - get <key>                         - Execute GET_VALUE in DHT
 - advertize <content-id>            - Execute ADD_PROVIDER in DHT
 - get_provider <content-id>         - Execute GET_PROVIDERS in DHT
@@ -44,9 +43,7 @@ def select(self, key: str, values: list[bytes]) -> int:
 
 
 class Node:
-    def __init__(
-        self, listen_addrs: list[multiaddr.Multiaddr], dht_role: str
-    ):
+    def __init__(self, listen_addrs: list[multiaddr.Multiaddr], dht_role: str):
         # Create a libp2p-host
         self.host = new_host(listen_addrs=listen_addrs, enable_metrics=True)
 
@@ -89,6 +86,11 @@ async def receive_loop(self, subsription):
         while not self.termination_event.is_set():
             try:
                 message = await subsription.get()
+
+                from_peer_id = ID(message.from_id).to_base58()
+                if from_peer_id == self.host.get_id().pretty():
+                    continue
+
                 print(f"From: {ID(message.from_id).to_base58()}")
                 print(f"Received: {message.data.decode('utf-8')}")
             except Exception:
@@ -131,35 +133,35 @@ async def command_executor(self, nursery):
                     if cmd == "publish" and len(parts) > 2:
                         await self.pubsub.publish(parts[1], parts[2].encode())
                         print(f"Published: {parts[2]}")
-                        
+
                     if cmd == "put" and len(parts) > 2:
                         key = parts[1]
                         value = parts[2].encode()
-                                                
+
                         await self.dht.put_value(key, value)
                         print(f"Stored value: {value.decode()} with key: {key}")
-                        
+
                     if cmd == "get" and len(parts) > 1:
                         key = parts[1]
-                                                
+
                         retrieved_value = await self.dht.get_value(key)
                         if retrieved_value:
                             print(f"Retrieved value: {retrieved_value.decode()}")
                         else:
                             print("Failed to retrieve")
-                        
+
                     if cmd == "advertize" and len(parts) > 1:
                         content_id = parts[1]
-                        
+
                         success = await self.dht.provide(content_id)
                         if success:
                             print(f"Advertised as provider for content: {content_id}")
                         else:
                             print("Failed to advertise as provider")
-                        
+
                     if cmd == "get_provider" and len(parts) > 1:
                         content_id = parts[1]
-                        
+
                         providers = await self.dht.find_providers(content_id)
                         if providers:
                             print(
@@ -168,7 +170,7 @@ async def command_executor(self, nursery):
                             )
                         else:
                             print("No providers found")
-                        
+
                     if cmd == "local":
                         maddr = self.host.get_addrs()[0]
                         print(maddr)

examples/metrics/runner.py

@@ -14,7 +14,7 @@ async def main() -> None:
     listen_addrs = get_available_interfaces(0)
     node = Node(
         listen_addrs=listen_addrs,
-        dht_role=  "server",
+        dht_role="server",
     )
 
     async with (

libp2p/metrics/bandwidth.py

@@ -11,6 +11,7 @@
 # GET_PROVIDERS
 # ADD_PROVIDERS
 
+
 class KadDhtMetrics:
     inbound: Counter
     find_node: Counter
@@ -59,26 +60,18 @@ def __init__(self):
     def record(self, event: KadDhtEvent) -> None:
         if event.inbound:
             self.inbound.labels(peer_id=event.peer_id).inc()
-            print("inbound")
 
         if event.find_node:
             self.find_node.labels(peer_id=event.peer_id).inc()
-            print("find_node")
 
         if event.get_value:
             self.get_value.labels(peer_id=event.peer_id).inc()
-            print("get_value")
 
         if event.put_value:
             self.put_value.labels(peer_id=event.peer_id).inc()
-            print("put_value")
 
         if event.get_providers:
             self.get_providers.labels(peer_id=event.peer_id).inc()
-            print("get_provider")
 
         if event.add_provider:
             self.add_provider.labels(peer_id=event.peer_id).inc()
-            print("add_provider")
-            
-        print("\n")

libp2p/metrics/kad_dht.py

@@ -8,6 +8,7 @@
 from libp2p.metrics.gossipsub import GossipsubMetrics
 from libp2p.metrics.kad_dht import KadDhtMetrics
 from libp2p.metrics.ping import PingMetrics
+from libp2p.metrics.swarm import SwarmEvent, SwarmMetrics
 from libp2p.pubsub.pubsub import GossipsubEvent
 
 
@@ -30,6 +31,7 @@ def __init__(self):
         self.ping = PingMetrics()
         self.gossipsub = GossipsubMetrics()
         self.kad_dht = KadDhtMetrics()
+        self.swarm = SwarmMetrics()
 
     async def start_prometheus_server(
         self,
@@ -62,3 +64,5 @@ async def start_prometheus_server(
                     self.gossipsub.record(event)
                 case KadDhtEvent():
                     self.kad_dht.record(event)
+                case SwarmEvent():
+                    self.swarm.record(event)