rippled/src/xrpld/perflog/detail/PerfLogImp.cpp

#include <xrpld/perflog/detail/PerfLogImp.h>

#include <xrpld/app/main/Application.h>
#include <xrpld/telemetry/MetricMacros.h>
#include <xrpld/telemetry/MetricsRegistry.h>

#include <xrpl/basics/Log.h>
#include <xrpl/basics/chrono.h>
#include <xrpl/beast/core/CurrentThreadName.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>
#include <xrpl/config/BasicConfig.h>
#include <xrpl/config/Constants.h>
#include <xrpl/core/Job.h>
#include <xrpl/core/JobTypes.h>
#include <xrpl/core/PerfLog.h>
#include <xrpl/core/ServiceRegistry.h>
#include <xrpl/json/json_value.h>
#include <xrpl/json/json_writer.h>
#include <xrpl/nodestore/Database.h>
#include <xrpl/protocol/jss.h>

#include <boost/filesystem/operations.hpp>
#include <boost/system/detail/error_code.hpp>

#include <chrono>
#include <cstdint>
#include <functional>
#include <ios>
#include <memory>
#include <mutex>
#include <ostream>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

namespace xrpl::perf {

PerfLogImp::Counters::Counters(std::set<char const*> const& labels, JobTypes const& jobTypes)
{
    {
        // populateRpc
        rpc.reserve(labels.size());
        for (std::string const label : labels)
        {
            auto const inserted = rpc.emplace(label, Rpc()).second;
            if (!inserted)
            {
                // Ensure that no other function populates this entry.
                // LCOV_EXCL_START
                UNREACHABLE(
                    "xrpl::perf::PerfLogImp::Counters::Counters : failed to "
                    "insert label");
                // LCOV_EXCL_STOP
            }
        }
    }
    {
        // populateJq
        jq.reserve(jobTypes.size());
        for (auto const& [jobType, _] : jobTypes)
        {
            auto const inserted = jq.emplace(jobType, Jq()).second;
            if (!inserted)
            {
                // Ensure that no other function populates this entry.
                // LCOV_EXCL_START
                UNREACHABLE(
                    "xrpl::perf::PerfLogImp::Counters::Counters : failed to "
                    "insert job type");
                // LCOV_EXCL_STOP
            }
        }
    }
}

json::Value
PerfLogImp::Counters::countersJson() const
{
    json::Value rpcobj(json::ValueType::Object);
    // totalRpc represents all rpc methods. All that started, finished, etc.
    Rpc totalRpc;
    for (auto const& proc : rpc)
    {
        Rpc value;
        {
            std::scoped_lock const lock(proc.second.mutex);
            if ((proc.second.value.started == 0u) && (proc.second.value.finished == 0u) &&
                (proc.second.value.errored == 0u))
            {
                continue;
            }
            value = proc.second.value;
        }

        json::Value p(json::ValueType::Object);
        p[jss::started] = std::to_string(value.started);
        totalRpc.started += value.started;
        p[jss::finished] = std::to_string(value.finished);
        totalRpc.finished += value.finished;
        p[jss::errored] = std::to_string(value.errored);
        totalRpc.errored += value.errored;
        p[jss::duration_us] = std::to_string(value.duration.count());
        totalRpc.duration += value.duration;
        rpcobj[proc.first] = p;
    }

    if (totalRpc.started != 0u)
    {
        json::Value totalRpcJson(json::ValueType::Object);
        totalRpcJson[jss::started] = std::to_string(totalRpc.started);
        totalRpcJson[jss::finished] = std::to_string(totalRpc.finished);
        totalRpcJson[jss::errored] = std::to_string(totalRpc.errored);
        totalRpcJson[jss::duration_us] = std::to_string(totalRpc.duration.count());
        rpcobj[jss::total] = totalRpcJson;
    }

    json::Value jobQueueObj(json::ValueType::Object);
    // totalJq represents all jobs. All enqueued, started, finished, etc.
    Jq totalJq;
    for (auto const& proc : jq)
    {
        Jq value;
        {
            std::scoped_lock const lock(proc.second.mutex);
            if ((proc.second.value.queued == 0u) && (proc.second.value.started == 0u) &&
                (proc.second.value.finished == 0u))
            {
                continue;
            }
            value = proc.second.value;
        }

        json::Value j(json::ValueType::Object);
        j[jss::queued] = std::to_string(value.queued);
        totalJq.queued += value.queued;
        j[jss::started] = std::to_string(value.started);
        totalJq.started += value.started;
        j[jss::finished] = std::to_string(value.finished);
        totalJq.finished += value.finished;
        j[jss::queued_duration_us] = std::to_string(value.queuedDuration.count());
        totalJq.queuedDuration += value.queuedDuration;
        j[jss::running_duration_us] = std::to_string(value.runningDuration.count());
        totalJq.runningDuration += value.runningDuration;
        jobQueueObj[JobTypes::name(proc.first)] = j;
    }

    if (totalJq.queued != 0u)
    {
        json::Value totalJqJson(json::ValueType::Object);
        totalJqJson[jss::queued] = std::to_string(totalJq.queued);
        totalJqJson[jss::started] = std::to_string(totalJq.started);
        totalJqJson[jss::finished] = std::to_string(totalJq.finished);
        totalJqJson[jss::queued_duration_us] = std::to_string(totalJq.queuedDuration.count());
        totalJqJson[jss::running_duration_us] = std::to_string(totalJq.runningDuration.count());
        jobQueueObj[jss::total] = totalJqJson;
    }

    json::Value counters(json::ValueType::Object);
    // Be kind to reporting tools and let them expect rpc and jq objects
    // even if empty.
    counters[jss::rpc] = rpcobj;
    counters[jss::job_queue] = jobQueueObj;
    return counters;
}

json::Value
PerfLogImp::Counters::currentJson() const
{
    auto const present = steady_clock::now();

    json::Value jobsArray(json::ValueType::Array);
    auto const jobs = [this] {
        std::scoped_lock const lock(jobsMutex);
        return this->jobs;
    }();

    for (auto const& j : jobs)
    {
        if (j.first == JtInvalid)
            continue;
        json::Value jobj(json::ValueType::Object);
        jobj[jss::job] = JobTypes::name(j.first);
        jobj[jss::duration_us] =
            std::to_string(std::chrono::duration_cast<microseconds>(present - j.second).count());
        jobsArray.append(jobj);
    }

    json::Value methodsArray(json::ValueType::Array);
    std::vector<MethodStart> methods;
    {
        std::scoped_lock const lock(methodsMutex);
        methods.reserve(this->methods.size());
        for (auto const& m : this->methods)
            methods.push_back(m.second);
    }
    for (auto m : methods)
    {
        json::Value methodobj(json::ValueType::Object);
        methodobj[jss::method] = m.first;
        methodobj[jss::duration_us] =
            std::to_string(std::chrono::duration_cast<microseconds>(present - m.second).count());
        methodsArray.append(methodobj);
    }

    json::Value current(json::ValueType::Object);
    current[jss::jobs] = jobsArray;
    current[jss::methods] = methodsArray;
    return current;
}

//-----------------------------------------------------------------------------

void
PerfLogImp::openLog()
{
    if (setup_.perfLog.empty())
        return;

    if (logFile_.is_open())
        logFile_.close();

    auto logDir = setup_.perfLog.parent_path();
    if (!boost::filesystem::is_directory(logDir))
    {
        boost::system::error_code ec;
        boost::filesystem::create_directories(logDir, ec);
        if (ec)
        {
            JLOG(j_.fatal()) << "Unable to create performance log "
                                "directory "
                             << logDir << ": " << ec.message();
            signalStop_();
            return;
        }
    }

    logFile_.open(setup_.perfLog.c_str(), std::ios::out | std::ios::app);

    if (!logFile_)
    {
        JLOG(j_.fatal()) << "Unable to open performance log " << setup_.perfLog << ".";
        signalStop_();
    }
}

void
PerfLogImp::run()
{
    beast::setCurrentThreadName("perflog");
    lastLog_ = system_clock::now();

    while (true)
    {
        {
            std::unique_lock<std::mutex> lock(mutex_);
            if (cond_.wait_until(lock, lastLog_ + setup_.logInterval, [&] { return stop_; }))
            {
                return;
            }
            if (rotate_)
            {
                openLog();
                rotate_ = false;
            }
        }
        report();
    }
}

void
PerfLogImp::report()
{
    if (!logFile_)
    {
        // If logFile_ is not writable do no further work.
        return;
    }

    auto const present = system_clock::now();
    if (present < lastLog_ + setup_.logInterval)
        return;
    lastLog_ = present;

    json::Value report(json::ValueType::Object);
    report[jss::time] = to_string(std::chrono::floor<microseconds>(present));
    {
        std::scoped_lock const lock{counters_.jobsMutex};
        report[jss::workers] = static_cast<unsigned int>(counters_.jobs.size());
    }
    report[jss::hostid] = hostname_;
    report[jss::counters] = counters_.countersJson();
    report[jss::nodestore] = json::ValueType::Object;
    app_.getNodeStore().getCountsJson(report[jss::nodestore]);
    report[jss::current_activities] = counters_.currentJson();
    app_.getOPs().stateAccounting(report);

    logFile_ << json::Compact{std::move(report)} << std::endl;
}

PerfLogImp::PerfLogImp(
    Setup setup,
    Application& app,
    beast::Journal journal,
    std::function<void()>&& signalStop)
    : setup_(std::move(setup)), app_(app), j_(journal), signalStop_(std::move(signalStop))
{
    openLog();
}

PerfLogImp::~PerfLogImp()
{
    stop();
}

void
PerfLogImp::rpcStart(std::string const& method, std::uint64_t const requestId)
{
    auto counter = counters_.rpc.find(method);
    if (counter == counters_.rpc.end())
    {
        // LCOV_EXCL_START
        UNREACHABLE("xrpl::perf::PerfLogImp::rpcStart : valid method input");
        return;
        // LCOV_EXCL_STOP
    }

    {
        std::scoped_lock const lock(counter->second.mutex);
        ++counter->second.value.started;
    }
    {
        std::scoped_lock const lock(counters_.methodsMutex);
        counters_.methods[requestId] = {counter->first.c_str(), steady_clock::now()};
    }

    // Record RPC start in OTel metrics pipeline. Recorded after the locks
    // above are released: the OTel call path allocates and takes locks
    // inside the SDK, so holding methodsMutex across it would widen a
    // process-wide critical section for no reason. Mirrors rpcEnd().
    if (auto* mr = app_.getMetricsRegistry())
        mr->recordRpcStarted(method);

    // A value that must be able to decrease (UpDownCounter), added at its
    // call site with no MetricsRegistry member/init-line/method. Paired with
    // the matching -1 in rpcEnd(). Runs on the same path as recordRpcStarted
    // above, i.e. only after a methods-map entry exists for this request.
    XRPL_METRIC_UPDOWN_ADD(app_, "rpc_in_flight_requests", "RPC requests currently executing", 1);
}

void
PerfLogImp::rpcEnd(std::string const& method, std::uint64_t const requestId, bool finish)
{
    auto counter = counters_.rpc.find(method);
    if (counter == counters_.rpc.end())
    {
        // LCOV_EXCL_START
        UNREACHABLE("xrpl::perf::PerfLogImp::rpcEnd : valid method input");
        return;
        // LCOV_EXCL_STOP
    }
    steady_time_point startTime;
    {
        std::scoped_lock const lock(counters_.methodsMutex);
        auto const e = counters_.methods.find(requestId);
        if (e != counters_.methods.end())
        {
            startTime = e->second.second;
            counters_.methods.erase(e);
        }
        else
        {
            // LCOV_EXCL_START
            UNREACHABLE("xrpl::perf::PerfLogImp::rpcEnd : valid requestId input");
            // Without an entry, startTime stays default-initialized; returning
            // avoids recording a bogus (now - epoch) duration to the counters
            // and OTel histogram below. UNREACHABLE does not halt release builds.
            return;
            // LCOV_EXCL_STOP
        }
    }
    auto const durationUs =
        std::chrono::duration_cast<microseconds>(steady_clock::now() - startTime);
    {
        std::scoped_lock const lock(counter->second.mutex);
        if (finish)
        {
            ++counter->second.value.finished;
        }
        else
        {
            ++counter->second.value.errored;
        }
        counter->second.value.duration += durationUs;
    }

    // Record RPC completion in OTel metrics pipeline. Mirrors the
    // rpcStart() instrumentation so the finished/errored counters and
    // duration histogram advance with every call.
    if (auto* mr = app_.getMetricsRegistry())
    {
        if (finish)
        {
            mr->recordRpcFinished(method, durationUs.count());
        }
        else
        {
            mr->recordRpcErrored(method, durationUs.count());
        }
    }

    // Matching -1 for the +1 recorded in rpcStart(). Placed after the early
    // returns above so it runs only when this request's methods-map entry was
    // found (i.e. a +1 was recorded for it), keeping the in-flight count balanced.
    XRPL_METRIC_UPDOWN_ADD(app_, "rpc_in_flight_requests", "RPC requests currently executing", -1);
}

void
PerfLogImp::jobQueue(JobType const type, std::string const& name)
{
    auto counter = counters_.jq.find(type);
    if (counter == counters_.jq.end())
    {
        // LCOV_EXCL_START
        UNREACHABLE("xrpl::perf::PerfLogImp::jobQueue : valid job type input");
        return;
        // LCOV_EXCL_STOP
    }
    {
        std::scoped_lock const lock(counter->second.mutex);
        ++counter->second.value.queued;
    }

    // Record job enqueue in OTel metrics pipeline, after the lock above is
    // released so the SDK's work stays outside the critical section.
    if (auto* mr = app_.getMetricsRegistry())
        mr->recordJobQueued(JobTypes::name(type), name);
}

void
PerfLogImp::jobStart(
    JobType const type,
    std::string const& name,
    microseconds dur,
    steady_time_point startTime,
    int instance)
{
    auto counter = counters_.jq.find(type);
    if (counter == counters_.jq.end())
    {
        // LCOV_EXCL_START
        UNREACHABLE("xrpl::perf::PerfLogImp::jobStart : valid job type input");
        return;
        // LCOV_EXCL_STOP
    }

    {
        std::scoped_lock const lock(counter->second.mutex);
        ++counter->second.value.started;
        counter->second.value.queuedDuration += dur;
    }
    {
        std::scoped_lock const lock(counters_.jobsMutex);
        if (instance >= 0 && instance < counters_.jobs.size())
            counters_.jobs[instance] = {type, startTime};
    }

    // Record job start in OTel metrics pipeline, after the locks above are
    // released. jobsMutex is process-wide and taken by every worker thread
    // on every job, so the SDK's allocation and internal locking must not
    // run inside it.
    if (auto* mr = app_.getMetricsRegistry())
        mr->recordJobStarted(JobTypes::name(type), name, dur.count());
}

void
PerfLogImp::jobFinish(JobType const type, std::string const& name, microseconds dur, int instance)
{
    auto counter = counters_.jq.find(type);
    if (counter == counters_.jq.end())
    {
        // LCOV_EXCL_START
        UNREACHABLE("xrpl::perf::PerfLogImp::jobFinish : valid job type input");
        return;
        // LCOV_EXCL_STOP
    }

    {
        std::scoped_lock const lock(counter->second.mutex);
        ++counter->second.value.finished;
        counter->second.value.runningDuration += dur;
    }
    {
        std::scoped_lock const lock(counters_.jobsMutex);
        if (instance >= 0 && instance < counters_.jobs.size())
            counters_.jobs[instance] = {JtInvalid, steady_time_point()};
    }

    // Record job finish in OTel metrics pipeline, after the locks above
    // are released, for the same reason as jobStart().
    if (auto* mr = app_.getMetricsRegistry())
        mr->recordJobFinished(JobTypes::name(type), name, dur.count());
}

void
PerfLogImp::resizeJobs(int const resize)
{
    std::scoped_lock const lock(counters_.jobsMutex);
    if (resize > counters_.jobs.size())
        counters_.jobs.resize(resize, {JtInvalid, steady_time_point()});
}

void
PerfLogImp::rotate()
{
    if (setup_.perfLog.empty())
        return;

    std::scoped_lock const lock(mutex_);
    rotate_ = true;
    cond_.notify_one();
}

void
PerfLogImp::start()
{
    if (!setup_.perfLog.empty())
        thread_ = std::thread(&PerfLogImp::run, this);
}

void
PerfLogImp::stop()
{
    if (thread_.joinable())
    {
        {
            std::scoped_lock const lock(mutex_);
            stop_ = true;
            cond_.notify_one();
        }
        thread_.join();
    }
}

//-----------------------------------------------------------------------------

PerfLog::Setup
setupPerfLog(Section const& section, boost::filesystem::path const& configDir)
{
    PerfLog::Setup setup;
    std::string perfLog;
    set(perfLog, "perf_log", section);
    if (!perfLog.empty())
    {
        setup.perfLog = boost::filesystem::path(perfLog);
        if (setup.perfLog.is_relative())
        {
            setup.perfLog = boost::filesystem::absolute(setup.perfLog, configDir);
        }
    }

    std::uint64_t logInterval = 0;
    if (getIfExists(section, Keys::kLogInterval, logInterval))
        setup.logInterval = std::chrono::seconds(logInterval);
    return setup;
}

std::unique_ptr<PerfLog>
makePerfLog(
    PerfLog::Setup const& setup,
    Application& app,
    beast::Journal journal,
    std::function<void()>&& signalStop)
{
    return std::make_unique<PerfLogImp>(setup, app, journal, std::move(signalStop));
}

}  // namespace xrpl::perf