Perflog peer info counts, and inbound traffic.

Author: mtrippled

src/xrpld/overlay/Overlay.h

@@ -219,6 +219,27 @@ class Overlay : public beast::PropertyStream::Source
     virtual std::uint64_t
     getPeerDisconnectCharges() const = 0;
 
+    virtual void
+    incTotalPeerInbound() = 0;
+    virtual std::uint64_t
+    getTotalPeerInbound() const = 0;
+    virtual void
+    incTotalPeerInboundEarlyReturn() = 0;
+    virtual std::uint64_t
+    getTotalPeerInboundEarlyReturn() const = 0;
+    virtual void
+    addTotalPeerInboundBytes(std::uint64_t numBytes) = 0;
+    virtual std::uint64_t
+    getTotalPeerInboundBytes() const = 0;
+    virtual void
+    incTotalPeerInboundComplete() = 0;
+    virtual std::uint64_t
+    getTotalPeerInboundComplete() const = 0;
+    virtual void
+    incTotalPeerInboundPropose() = 0;
+    virtual std::uint64_t
+    getTotalPeerInboundPropose() const = 0;
+
     /** Returns the ID of the network this server is configured for, if any.
 
         The ID is just a numerical identifier, with the IDs 0, 1 and 2 used to

src/xrpld/overlay/detail/OverlayImpl.h

@@ -119,6 +119,13 @@ class OverlayImpl : public Overlay, public reduce_relay::SquelchHandler
     std::atomic<uint64_t> peerDisconnects_{0};
     std::atomic<uint64_t> peerDisconnectsCharges_{0};
 
+    std::atomic<uint64_t> totalPeerInbound_{0};
+    std::atomic<uint64_t> totalPeerInboundEarlyReturn_{0};
+    std::atomic<uint64_t> totalPeerInboundBytes_{0};
+    std::atomic<uint64_t> totalPeerInboundComplete_{0};
+    std::atomic<uint64_t> totalPeerInboundPropose_{0};
+
+
     reduce_relay::Slots<UptimeClock> slots_;
 
     // Transaction reduce-relay metrics
@@ -380,6 +387,64 @@ class OverlayImpl : public Overlay, public reduce_relay::SquelchHandler
         return peerDisconnectsCharges_;
     }
 
+    void
+    incTotalPeerInbound() override
+    {
+        ++totalPeerInbound_;
+    }
+
+    std::uint64_t
+    getTotalPeerInbound() const override
+    {
+        return totalPeerInbound_;
+    }
+
+    void
+    incTotalPeerInboundEarlyReturn() override
+    {
+        ++totalPeerInboundEarlyReturn_;
+    }
+
+    std::uint64_t
+    getTotalPeerInboundEarlyReturn() const override
+    {
+        return totalPeerInboundEarlyReturn_;
+    }
+
+    void
+    addTotalPeerInboundBytes(std::uint64_t numBytes) override {
+        totalPeerInboundBytes_ += numBytes;
+    }
+
+    std::uint64_t
+    getTotalPeerInboundBytes() const override {
+        return totalPeerInboundBytes_;
+    }
+
+    std::uint64_t
+    getTotalPeerInboundComplete() const override
+    {
+        return totalPeerInboundComplete_;
+    }
+
+    void
+    incTotalPeerInboundComplete() override
+    {
+        ++totalPeerInboundComplete_;
+    }
+
+    std::uint64_t
+    getTotalPeerInboundPropose() const override
+    {
+        return totalPeerInboundPropose_;
+    }
+
+    void
+    incTotalPeerInboundPropose() override
+    {
+        ++totalPeerInboundPropose_;
+    }
+
     std::optional<std::uint32_t>
     networkID() const override
     {

src/xrpld/overlay/detail/PeerImp.cpp

@@ -895,17 +895,25 @@ PeerImp::doProtocolStart()
 void
 PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
 {
-    if (!socket_.is_open())
+    app_.overlay().incTotalPeerInbound();
+    if (!socket_.is_open()) {
+        app_.overlay().incTotalPeerInboundEarlyReturn();
         return;
-    if (ec == boost::asio::error::operation_aborted)
+    }
+    if (ec == boost::asio::error::operation_aborted) {
+        app_.overlay().incTotalPeerInboundEarlyReturn();
         return;
+    }
     if (ec == boost::asio::error::eof)
     {
         JLOG(journal_.info()) << "EOF";
+        app_.overlay().incTotalPeerInboundEarlyReturn();
         return gracefulClose();
     }
-    if (ec)
+    if (ec) {
+        app_.overlay().incTotalPeerInboundEarlyReturn();
         return fail("onReadMessage", ec);
+    }
     if (auto stream = journal_.trace())
     {
         if (bytes_transferred > 0)
@@ -914,6 +922,8 @@ PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
             stream << "onReadMessage";
     }
 
+    app_.overlay().addTotalPeerInboundBytes(bytes_transferred);
+
     metrics_.recv.add_message(bytes_transferred);
 
     read_buffer_.commit(bytes_transferred);
@@ -927,7 +937,9 @@ PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
         using namespace std::chrono_literals;
         std::tie(bytes_consumed, ec) = perf::measureDurationAndLog(
             [&]() {
-                return invokeProtocolMessage(read_buffer_.data(), *this, hint);
+                return invokeProtocolMessage(read_buffer_.data(), *this, hint,
+                    [this]() { app_.overlay().incTotalPeerInboundComplete(); },
+                    [this]() { app_.overlay().incTotalPeerInboundPropose(); });
             },
             "invokeProtocolMessage",
             350ms,

src/xrpld/overlay/detail/ProtocolMessage.h

@@ -331,12 +331,14 @@ invoke(MessageHeader const& header, Buffers const& buffers, Handler& handler)
 
     @return The number of bytes consumed, or the error code if any.
 */
-template <class Buffers, class Handler>
+template <class Buffers, class Handler, class Func, class Func2>
 std::pair<std::size_t, boost::system::error_code>
 invokeProtocolMessage(
     Buffers const& buffers,
     Handler& handler,
-    std::size_t& hint)
+    std::size_t& hint,
+    Func incCompleteFunc,
+    Func2 incProposeFunc)
 {
     std::pair<std::size_t, boost::system::error_code> result = {0, {}};
 
@@ -384,6 +386,7 @@ invokeProtocolMessage(
 
     bool success;
 
+    incCompleteFunc();
     switch (header->message_type)
     {
         case protocol::mtMANIFESTS:
@@ -417,6 +420,7 @@ invokeProtocolMessage(
         case protocol::mtPROPOSE_LEDGER:
             success = detail::invoke<protocol::TMProposeSet>(
                 *header, buffers, handler);
+            incProposeFunc();
             break;
         case protocol::mtSTATUS_CHANGE:
             success = detail::invoke<protocol::TMStatusChange>(

src/xrpld/perflog/detail/PerfLogImp.cpp

@@ -38,6 +38,8 @@
 #include <unordered_map>
 #include <utility>
 
+#include "xrpld/overlay/Overlay.h"
+
 namespace ripple {
 namespace perf {
 
@@ -304,6 +306,20 @@ PerfLogImp::report()
     report[jss::nodestore] = Json::objectValue;
     app_.getNodeStore().getCountsJson(report[jss::nodestore]);
     report[jss::current_activities] = counters_.currentJson();
+
+    Json::Value peer = Json::objectValue;
+    peer["total_inbound"] = std::to_string(app_.overlay().getTotalPeerInbound());
+    peer["total_inbound_early_return"] = std::to_string(app_.overlay().getTotalPeerInboundEarlyReturn());
+    peer["total_inbound_bytes"] = std::to_string(app_.overlay().getTotalPeerInboundBytes());
+    peer["total_inbound_complete"] = std::to_string(app_.overlay().getTotalPeerInboundComplete());
+    peer["total_inbound_propose"] = std::to_string(app_.overlay().getTotalPeerInboundPropose());
+    peer[jss::peers] = Json::UInt(app_.overlay().size());
+    peer[jss::peer_disconnects] =
+        std::to_string(app_.overlay().getPeerDisconnect());
+    peer[jss::peer_disconnects_resources] =
+        std::to_string(app_.overlay().getPeerDisconnectCharges());
+    report["peer"] = peer;
+
     app_.getOPs().stateAccounting(report);
 
     logFile_ << Json::Compact{std::move(report)} << std::endl;