Add detailed peer statistics to PerfLog for both

perf_log and server_info counters.

Author: mtrippled

src/test/core/Workers_test.cpp

@@ -39,6 +39,8 @@ namespace perf {
 
 class PerfLogTest : public PerfLog
 {
+    Peer peer;
+
     void
     rpcStart(std::string const& method, std::uint64_t requestId) override
     {
@@ -95,6 +97,11 @@ class PerfLogTest : public PerfLog
     rotate() override
     {
     }
+
+    Peer&
+    getPeerCounters() override {
+        return peer;
+    }
 };
 
 }  // namespace perf

src/xrpld/overlay/Message.h

@@ -96,6 +96,15 @@ class Message : public std::enable_shared_from_this<Message>
         return validatorKey_;
     }
 
+    /** Get the message type from the payload header.
+     * First four bytes are the compression/algorithm flag and the payload size.
+     * Next two bytes are the message type
+     * @param in Payload header pointer
+     * @return Message type
+     */
+    int
+    getType(std::uint8_t const* in) const;
+
 private:
     std::vector<uint8_t> buffer_;
     std::vector<uint8_t> bufferCompressed_;
@@ -125,15 +134,6 @@ class Message : public std::enable_shared_from_this<Message>
      */
     void
     compress();
-
-    /** Get the message type from the payload header.
-     * First four bytes are the compression/algorithm flag and the payload size.
-     * Next two bytes are the message type
-     * @param in Payload header pointer
-     * @return Message type
-     */
-    int
-    getType(std::uint8_t const* in) const;
 };
 
 }  // namespace ripple

src/xrpld/overlay/detail/ConnectAttempt.cpp

@@ -21,6 +21,7 @@
 #include <xrpld/overlay/detail/ConnectAttempt.h>
 #include <xrpld/overlay/detail/PeerImp.h>
 #include <xrpld/overlay/detail/ProtocolVersion.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/json/json_reader.h>
 
@@ -52,6 +53,7 @@ ConnectAttempt::ConnectAttempt(
     , stream_(*stream_ptr_)
     , slot_(slot)
 {
+    ++app_.getPerfLog().getPeerCounters().connection.totalOutboundAttempts;
     JLOG(journal_.debug()) << "Connect " << remote_endpoint;
 }
 
@@ -72,6 +74,7 @@ ConnectAttempt::stop()
     {
         JLOG(journal_.debug()) << "Stop";
     }
+    ++app_.getPerfLog().getPeerCounters().connection.connectCloseStop;
     close();
 }
 
@@ -150,8 +153,10 @@ ConnectAttempt::onTimer(error_code ec)
     {
         // This should never happen
         JLOG(journal_.error()) << "onTimer: " << ec.message();
+        ++app_.getPerfLog().getPeerCounters().connection.connectCloseOnTimer;
         return close();
     }
+    ++app_.getPerfLog().getPeerCounters().connection.connectFailTimeouts;
     fail("Timeout");
 }
 
@@ -165,8 +170,10 @@ ConnectAttempt::onConnect(error_code ec)
     endpoint_type local_endpoint;
     if (!ec)
         local_endpoint = socket_.local_endpoint(ec);
-    if (ec)
+    if (ec) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnConnectError;
         return fail("onConnect", ec);
+    }
     if (!socket_.is_open())
         return;
     JLOG(journal_.trace()) << "onConnect";
@@ -192,17 +199,23 @@ ConnectAttempt::onHandshake(error_code ec)
     endpoint_type local_endpoint;
     if (!ec)
         local_endpoint = socket_.local_endpoint(ec);
-    if (ec)
+    if (ec) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnHandshakeError;
         return fail("onHandshake", ec);
+    }
     JLOG(journal_.trace()) << "onHandshake";
 
     if (!overlay_.peerFinder().onConnected(
-            slot_, beast::IPAddressConversion::from_asio(local_endpoint)))
+            slot_, beast::IPAddressConversion::from_asio(local_endpoint))) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnHandshakeDuplicate;
         return fail("Duplicate connection");
+    }
 
     auto const sharedValue = makeSharedValue(*stream_ptr_, journal_);
-    if (!sharedValue)
+    if (!sharedValue) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectCloseOnHandshake;
         return close();  // makeSharedValue logs
+    }
 
     req_ = makeRequest(
         !overlay_.peerFinder().config().peerPrivate,
@@ -237,8 +250,10 @@ ConnectAttempt::onWrite(error_code ec)
         return;
     if (ec == boost::asio::error::operation_aborted)
         return;
-    if (ec)
+    if (ec) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnWriteError;
         return fail("onWrite", ec);
+    }
     boost::beast::http::async_read(
         stream_,
         read_buf_,
@@ -267,8 +282,10 @@ ConnectAttempt::onRead(error_code ec)
             shared_from_this(),
             std::placeholders::_1)));
     }
-    if (ec)
+    if (ec) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnReadError;
         return fail("onRead", ec);
+    }
     processResponse();
 }
 
@@ -279,10 +296,14 @@ ConnectAttempt::onShutdown(error_code ec)
     if (!ec)
     {
         JLOG(journal_.error()) << "onShutdown: expected error condition";
+        ++app_.getPerfLog().getPeerCounters().connection.connectCloseOnShutdownNoError;
         return close();
     }
-    if (ec != boost::asio::error::eof)
+    if (ec != boost::asio::error::eof) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnShutdownError;
         return fail("onShutdown", ec);
+    }
+    ++app_.getPerfLog().getPeerCounters().connection.connectCloseOnShutdown;
     close();
 }
 
@@ -332,6 +353,7 @@ ConnectAttempt::processResponse()
         JLOG(journal_.info())
             << "Unable to upgrade to peer protocol: " << response_.result()
             << " (" << response_.reason() << ")";
+        ++app_.getPerfLog().getPeerCounters().connection.connectCloseUpgrade;
         return close();
     }
 
@@ -345,14 +367,18 @@ ConnectAttempt::processResponse()
         if (pvs.size() == 1 && isProtocolSupported(pvs[0]))
             negotiatedProtocol = pvs[0];
 
-        if (!negotiatedProtocol)
+        if (!negotiatedProtocol) {
+            ++app_.getPerfLog().getPeerCounters().connection.connectFailProtocol;
             return fail(
                 "processResponse: Unable to negotiate protocol version");
+        }
     }
 
     auto const sharedValue = makeSharedValue(*stream_ptr_, journal_);
-    if (!sharedValue)
+    if (!sharedValue) {
+        ++app_.getPerfLog().getPeerCounters().connection.connectCloseShared;
         return close();  // makeSharedValue logs
+    }
 
     try
     {
@@ -378,8 +404,10 @@ ConnectAttempt::processResponse()
 
         auto const result = overlay_.peerFinder().activate(
             slot_, publicKey, static_cast<bool>(member));
-        if (result != PeerFinder::Result::success)
+        if (result != PeerFinder::Result::success) {
+            ++app_.getPerfLog().getPeerCounters().connection.connectFailSlotsFull;
             return fail("Outbound slots full");
+        }
 
         auto const peer = std::make_shared<PeerImp>(
             app_,
@@ -397,6 +425,7 @@ ConnectAttempt::processResponse()
     }
     catch (std::exception const& e)
     {
+        ++app_.getPerfLog().getPeerCounters().connection.connectFailOnHandshakeFailure;
         return fail(std::string("Handshake failure (") + e.what() + ")");
     }
 }

src/xrpld/overlay/detail/OverlayImpl.cpp

@@ -165,6 +165,7 @@ OverlayImpl::onHandoff(
     http_request_type&& request,
     endpoint_type remote_endpoint)
 {
+    ++app_.getPerfLog().getPeerCounters().connection.totalInboundAttempts;
     auto const id = next_id_++;
     beast::WrappedSink sink(app_.logs()["Peer"], makePrefix(id));
     beast::Journal journal(sink);

src/xrpld/overlay/detail/PeerImp.cpp

@@ -119,6 +119,7 @@ PeerImp::PeerImp(
           app_.config().LEDGER_REPLAY))
     , ledgerReplayMsgHandler_(app, app.getLedgerReplayer())
 {
+    ++app_.getPerfLog().getPeerCounters().connection.totalInboundConnects;
     JLOG(journal_.info())
         << "compression enabled " << (compressionEnabled_ == Compressed::On)
         << " vp reduce-relay base squelch enabled "
@@ -241,29 +242,36 @@ PeerImp::send(std::shared_ptr<Message> const& m)
 {
     if (!strand_.running_in_this_thread())
         return post(strand_, std::bind(&PeerImp::send, shared_from_this(), m));
-    if (gracefulClose_)
+    auto& peerCounters = app_.getPerfLog().getPeerCounters();
+    if (gracefulClose_) {
+        ++peerCounters.send.sendQueueFailedGracefulClose;
         return;
-    if (detaching_)
+    }
+    if (detaching_) {
+        ++peerCounters.send.sendQueueFailedDetaching;
         return;
+    }
 
+    std::size_t const msgSize = m->getBuffer(compressionEnabled_).size();
     auto validator = m->getValidatorKey();
     if (validator && !squelch_.expireSquelch(*validator))
     {
+        ++peerCounters.send.sendQueueFailedSquelch;
         overlay_.reportOutboundTraffic(
             TrafficCount::category::squelch_suppressed,
-            static_cast<int>(m->getBuffer(compressionEnabled_).size()));
+            static_cast<int>(msgSize));
         return;
     }
 
     // report categorized outgoing traffic
     overlay_.reportOutboundTraffic(
         safe_cast<TrafficCount::category>(m->getCategory()),
-        static_cast<int>(m->getBuffer(compressionEnabled_).size()));
+        static_cast<int>(msgSize));
 
     // report total outgoing traffic
     overlay_.reportOutboundTraffic(
         TrafficCount::category::total,
-        static_cast<int>(m->getBuffer(compressionEnabled_).size()));
+        static_cast<int>(msgSize));
 
     auto sendq_size = send_queue_.size();
 
@@ -283,6 +291,8 @@ PeerImp::send(std::shared_ptr<Message> const& m)
     }
 
     send_queue_.push(m);
+    peerCounters.queuedPeerMessage(m->getType(
+        m->getBuffer(compressionEnabled_).data()), msgSize, journal_);
 
     if (sendq_size != 0)
         return;
@@ -356,6 +366,10 @@ PeerImp::charge(Resource::Charge const& fee, std::string const& context)
     {
         // Sever the connection
         overlay_.incPeerDisconnectCharges();
+        if (inbound_)
+            ++app_.getPerfLog().getPeerCounters().connection.disconnectInboundResources;
+        else
+            ++app_.getPerfLog().getPeerCounters().connection.disconnectOutboundResources;
         fail("charge: Resources");
     }
 }
@@ -588,10 +602,12 @@ PeerImp::close()
         if (inbound_)
         {
             JLOG(journal_.debug()) << "Closed";
+            ++app_.getPerfLog().getPeerCounters().connection.totalInboundDisconnects;
         }
         else
         {
             JLOG(journal_.info()) << "Closed";
+            ++app_.getPerfLog().getPeerCounters().connection.totalOutboundDisconnects;
         }
     }
 }
@@ -889,6 +905,8 @@ PeerImp::doProtocolStart()
 void
 PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
 {
+    auto& peerCounters = app_.getPerfLog().getPeerCounters();
+    ++peerCounters.receive.receivePackets;
     if (!socket_.is_open())
         return;
     if (ec == boost::asio::error::operation_aborted)
@@ -921,7 +939,8 @@ 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,
+                    peerCounters, journal_);
             },
             "invokeProtocolMessage",
             350ms,
@@ -953,12 +972,21 @@ PeerImp::onReadMessage(error_code ec, std::size_t bytes_transferred)
 void
 PeerImp::onWriteMessage(error_code ec, std::size_t bytes_transferred)
 {
-    if (!socket_.is_open())
+    auto& peerCounters = app_.getPerfLog().getPeerCounters();
+    if (!socket_.is_open()) {
+        ++peerCounters.send.sendFailedClosed;
         return;
-    if (ec == boost::asio::error::operation_aborted)
+    }
+    if (ec == boost::asio::error::operation_aborted) {
+        ++peerCounters.send.sendFailedAborted;
         return;
-    if (ec)
+    }
+    if (ec) {
+        ++peerCounters.send.sendFailedOther;
         return fail("onWriteMessage", ec);
+    }
+    ++peerCounters.send.sent;
+    peerCounters.send.sentBytes += bytes_transferred;
     if (auto stream = journal_.trace())
     {
         if (bytes_transferred > 0)

src/xrpld/overlay/detail/PeerImp.h

@@ -26,6 +26,7 @@
 #include <xrpld/overlay/detail/OverlayImpl.h>
 #include <xrpld/overlay/detail/ProtocolVersion.h>
 #include <xrpld/peerfinder/PeerfinderManager.h>
+#include <xrpld/perflog/PerfLog.h>
 
 #include <xrpl/basics/Log.h>
 #include <xrpl/basics/UnorderedContainers.h>
@@ -703,6 +704,7 @@ PeerImp::PeerImp(
           app_.config().LEDGER_REPLAY))
     , ledgerReplayMsgHandler_(app, app.getLedgerReplayer())
 {
+    ++app_.getPerfLog().getPeerCounters().connection.totalOutboundConnects;
     read_buffer_.commit(boost::asio::buffer_copy(
         read_buffer_.prepare(boost::asio::buffer_size(buffers)), buffers));
     JLOG(journal_.info())