From aafb8657555ac6ae71bf9060242bfeb0b4b0d835 Mon Sep 17 00:00:00 2001
From: Dean Eigenmann <7621705+decanus@users.noreply.github.com>
Date: Thu, 18 Jun 2020 05:16:23 +0200
Subject: [PATCH] feature/accounting (#29)

* adds accounting experimentally

* updated

* sent

* updated

* print metrics

* var

* indent

* fix
---
 waku/node/v1/wakunode.nim          | 14 ++++++++++++++
 waku/protocol/v1/waku_protocol.nim |  9 +++++++++
 2 files changed, 23 insertions(+)

diff --git a/waku/node/v1/wakunode.nim b/waku/node/v1/wakunode.nim
index 1bb3a23b8..a0ed1b3c9 100644
--- a/waku/node/v1/wakunode.nim
+++ b/waku/node/v1/wakunode.nim
@@ -116,6 +116,20 @@ proc run(config: WakuNodeConf) =
     setupWakuSimRPC(node, rpcServer)
     rpcServer.start()
 
+  proc logPeerAccounting(udata: pointer) {.closure, gcsafe.} =
+    {.gcsafe.}:
+      
+      for peer in node.peerPool.peers:
+        let 
+          sent = peer.state(Waku).accounting.sent
+          received = peer.state(Waku).accounting.received
+          id = peer.remote.id
+        info "Peer Metrics", id, sent, received
+        peer.state(Waku).accounting = Accounting(sent: 0, received: 0)
+
+    addTimer(Moment.fromNow(2.seconds), logPeerAccounting)
+  addTimer(Moment.fromNow(2.seconds), logPeerAccounting)
+
   when defined(insecure):
     if config.metricsServer:
       let
diff --git a/waku/protocol/v1/waku_protocol.nim b/waku/protocol/v1/waku_protocol.nim
index 89dd87346..ffb9bc9f7 100644
--- a/waku/protocol/v1/waku_protocol.nim
+++ b/waku/protocol/v1/waku_protocol.nim
@@ -71,6 +71,10 @@ type
     rateLimits*: Option[RateLimits]
     topics*: Option[seq[Topic]]
 
+  Accounting* = ref object
+    sent*: uint
+    received*: uint
+
   WakuPeer = ref object
     initialized: bool # when successfully completed the handshake
     powRequirement*: float64
@@ -79,6 +83,7 @@ type
     trusted*: bool
     topics*: Option[seq[Topic]]
     received: HashSet[Hash]
+    accounting*: Accounting
 
   P2PRequestHandler* = proc(peer: Peer, envelope: Envelope) {.gcsafe.}
 
@@ -260,6 +265,7 @@ p2pProtocol Waku(version = wakuVersion,
 
     wakuPeer.received.init()
     wakuPeer.trusted = false
+    wakuPeer.accounting = Accounting(sent: 0, received: 0)
     wakuPeer.initialized = true
 
     # No timer based queue processing for a light node.
@@ -284,6 +290,8 @@ p2pProtocol Waku(version = wakuVersion,
         # await peer.disconnect(SubprotocolReason)
         continue
 
+      peer.state.accounting.received += 1
+
       let msg = initMessage(envelope)
       if not msg.allowed(peer.networkState.config):
         # disconnect from peers sending bad envelopes
@@ -403,6 +411,7 @@ proc processQueue(peer: Peer) =
 
     trace "Adding envelope"
     envelopes.add(message.env)
+    wakuPeer.accounting.sent += 1
     wakuPeer.received.incl(message.hash)
 
   if envelopes.len() > 0: