Skip to content

Commit

Permalink
JsonRpcConnection: Log message processing stats
Browse files Browse the repository at this point in the history
  • Loading branch information
yhabteab committed Aug 29, 2024
1 parent 93484ed commit b1c480b
Show file tree
Hide file tree
Showing 3 changed files with 48 additions and 3 deletions.
10 changes: 10 additions & 0 deletions lib/base/logger.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -118,6 +118,16 @@ class Log

~Log();

/**
* Set the severity of the current Log instance.
*
* @param severity LogSeverity The new severity you want to override the current one with.
*/
void SetSeverity(LogSeverity severity)
{
m_Severity = severity;
}

template<typename T>
Log& operator<<(const T& val)
{
Expand Down
39 changes: 37 additions & 2 deletions lib/remote/jsonrpcconnection.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,8 @@ void JsonRpcConnection::Start()

void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
{
namespace ch = std::chrono;

m_Stream->next_layer().SetSeen(&m_Seen);

for (;;) {
Expand All @@ -78,9 +80,36 @@ void JsonRpcConnection::HandleIncomingMessages(boost::asio::yield_context yc)
m_Seen = Utility::GetTime();

try {
auto start (ch::steady_clock::now());
ch::steady_clock::duration cpuBoundDuration;

String rpcMethod;

Log statsLog(LogDebug, "JsonRpcConnection");

Defer addLogStats ([this, &statsLog, &rpcMethod, &start, &cpuBoundDuration]() {
statsLog << "Processing JSON-RPC '" << rpcMethod << "' message for identity '" << m_Identity << "'";

if (ch::duration_cast<ch::seconds>(cpuBoundDuration).count()) {
statsLog << " waited '" << ch::duration_cast<ch::milliseconds>(cpuBoundDuration).count() << "ms' on semaphore and";
}

auto duration = ch::steady_clock::now() - start;
if (duration >= ch::seconds(5)) {
// Processing that RPC message seems to take an unexpectedly long time,
// so promote the log entry from debug to warning.
statsLog.SetSeverity(LogWarning);
}

statsLog << " took total " << ch::duration_cast<ch::milliseconds>(duration).count() << "ms.";
});

CpuBoundWork handleMessage (yc);

MessageHandler(message);
// Cache the elapsed time to acquire a CPU semaphore used to detect extremely heavy workloads.
cpuBoundDuration = ch::steady_clock::now() - start;

MessageHandler(message, rpcMethod);

l_TaskStats.InsertValue(Utility::GetTime(), 1);
} catch (const std::exception& ex) {
Expand Down Expand Up @@ -245,9 +274,15 @@ void JsonRpcConnection::Disconnect()
});
}

void JsonRpcConnection::MessageHandler(const String& jsonString)
void JsonRpcConnection::MessageHandler(const String& jsonString, String& rpcMethod)
{
Dictionary::Ptr message = JsonRpc::DecodeMessage(jsonString);
Defer setRpcMethod ([&message, &rpcMethod]() {
rpcMethod = message->Get("method");
if (rpcMethod.IsEmpty()) {
rpcMethod = "UNKNOWN";
}
});

if (m_Endpoint && message->Contains("ts")) {
double ts = message->Get("ts");
Expand Down
2 changes: 1 addition & 1 deletion lib/remote/jsonrpcconnection.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,7 @@ class JsonRpcConnection final : public Object
void CheckLiveness(boost::asio::yield_context yc);

bool ProcessMessage();
void MessageHandler(const String& jsonString);
void MessageHandler(const String& jsonString, String& rpcMethod);

void CertificateRequestResponseHandler(const Dictionary::Ptr& message);

Expand Down

0 comments on commit b1c480b

Please sign in to comment.