Correct upload and download bandwidth telemetry

cmake/Hunter/config.cmake

@@ -105,6 +105,12 @@ hunter_config(
       SECP256K1_ENABLE_MODULE_RECOVERY=ON
 )
 
+hunter_config(
+    libp2p
+    URL https://github.com/ErakhtinB/cpp-libp2p/archive/e10b86d6ffe726f39ca6e9a1dfc5826e9249d7e7.tar.gz
+    SHA1 84e1a72a13ab83baff8afe4c2f327a412563763b
+)
+
 hunter_config(
     erasure_coding_crust
 #    VERSION 0.0.8

core/telemetry/impl/service_impl.cpp

@@ -23,7 +23,9 @@ namespace rapidjson {
 #include <libp2p/basic/scheduler/asio_scheduler_backend.hpp>
 #include <libp2p/basic/scheduler/scheduler_impl.hpp>
 #include <libp2p/multi/multiaddress.hpp>
+#include <libp2p/transport/tcp/tcp_connection.hpp>
 
+#include <network/helpers/stream_read_buffer.hpp>
 #include "common/uri.hpp"
 #include "telemetry/impl/connection_impl.hpp"
 #include "telemetry/impl/telemetry_thread_pool.hpp"
@@ -452,21 +454,16 @@ namespace kagome::telemetry {
 
     rapidjson::Value payload(rapidjson::kObjectType);
 
-    rapidjson::Value bandwidth_down, bandwidth_up, peers_count;
-    auto active_peers = peer_manager_->activePeersNumber();
-    // we are not actually measuring bandwidth. the following will just let us
-    // see the history of active peers count change in the telemetry UI
-    auto peers_to_bandwidth = active_peers * 1'000'000;
-    // NOLINTNEXTLINE(cppcoreguidelines-narrowing-conversions)
-    bandwidth_down.SetInt(peers_to_bandwidth);
-    // NOLINTNEXTLINE(cppcoreguidelines-narrowing-conversions)
-    bandwidth_up.SetInt(peers_to_bandwidth);
-    // NOLINTNEXTLINE(cppcoreguidelines-narrowing-conversions)
-    peers_count.SetInt(active_peers);
+    rapidjson::Value peers_count;
+    peers_count.SetInt(peer_manager_->activePeersNumber());
 
+    auto bandwidth = getBandwidth();
+    rapidjson::Value upBandwidth, downBandwidth;
+    downBandwidth.SetUint64(bandwidth.down);
+    upBandwidth.SetUint64(bandwidth.up);
     // fields order is preserved the same way substrate orders it
-    payload.AddMember("bandwidth_download", bandwidth_down, allocator)
-        .AddMember("bandwidth_upload", bandwidth_up, allocator)
+    payload.AddMember("bandwidth_download", downBandwidth, allocator)
+        .AddMember("bandwidth_upload", upBandwidth, allocator)
         .AddMember("msg", str_val("system.interval"), allocator)
         .AddMember("peers", peers_count, allocator);
 
@@ -489,4 +486,39 @@ namespace kagome::telemetry {
   bool TelemetryServiceImpl::isEnabled() const {
     return enabled_;
   }
+
+  TelemetryServiceImpl::Bandwidth TelemetryServiceImpl::getBandwidth() {
+    if (not previous_bandwidth_calculated_.has_value()) {
+      previous_bandwidth_calculated_ = std::chrono::high_resolution_clock::now();
+    }
+
+    auto calculateBandwidth = [](uint64_t &previousBytes,
+                                 uint64_t totalBytes,
+                                 auto &bandwidth,
+                                 const std::chrono::seconds &timeElapsed) {
+      if (timeElapsed.count() > 0) {
+        bandwidth = (totalBytes - previousBytes) / timeElapsed.count();
+      } else {
+        bandwidth = 0;
+      }
+      previousBytes = totalBytes;
+    };
+
+    const auto currentTime = std::chrono::high_resolution_clock::now();
+    const auto timeElapsed = std::chrono::duration_cast<std::chrono::seconds>(
+        currentTime - *previous_bandwidth_calculated_);
+
+    Bandwidth bandwidth;
+    const auto totalBytesRead = libp2p::transport::TcpConnection::getBytesRead();
+    calculateBandwidth(
+        previous_bytes_read_, totalBytesRead, bandwidth.down, timeElapsed);
+
+    const auto totalBytesWritten = libp2p::transport::TcpConnection::getBytesWritten();
+    calculateBandwidth(
+        previous_bytes_written_, totalBytesWritten, bandwidth.up, timeElapsed);
+
+    previous_bandwidth_calculated_ = currentTime;
+
+    return bandwidth;
+  }
 }  // namespace kagome::telemetry

core/telemetry/impl/service_impl.hpp

@@ -88,6 +88,12 @@ namespace kagome::telemetry {
     void stop();
 
    private:
+    /// structure to store last calculated bandwidth values
+    struct Bandwidth {
+      uint64_t down{0};
+      uint64_t up{0};
+    };
+
     /// parse telemetry endpoints from chain specification
     std::vector<TelemetryEndpoint> chainSpecEndpoints() const;
 
@@ -100,6 +106,9 @@ namespace kagome::telemetry {
     /// produces and sends system health notifications
     void delayedNotificationsRoutine();
 
+    /// calculates and returns current bandwidth values
+    Bandwidth getBandwidth();
+
     /**
      * Constructs the main and immutable part of JSON to be serialized later as
      * greeting message on new telemetry connections.
@@ -169,6 +178,11 @@ namespace kagome::telemetry {
     std::string genesis_hash_;
     std::shared_ptr<MessagePool> message_pool_;
     bool was_synchronized_ = false;
+
+    uint64_t previous_bytes_read_{0};
+    uint64_t previous_bytes_written_{0};
+    std::optional<std::chrono::system_clock::time_point>
+        previous_bandwidth_calculated_;
   };
 
 }  // namespace kagome::telemetry