rippled/src/libxrpl/beast/insight/StatsDCollector.cpp

#include <xrpl/beast/insight/StatsDCollector.h>

#include <xrpl/beast/core/List.h>
#include <xrpl/beast/insight/CounterImpl.h>
#include <xrpl/beast/insight/EventImpl.h>
#include <xrpl/beast/insight/GaugeImpl.h>
#include <xrpl/beast/insight/Hook.h>
#include <xrpl/beast/insight/HookImpl.h>
#include <xrpl/beast/insight/MeterImpl.h>
#include <xrpl/beast/net/IPEndpoint.h>
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/instrumentation.h>

#include <boost/asio/basic_waitable_timer.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/dispatch.hpp>
#include <boost/asio/error.hpp>
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/udp.hpp>
#include <boost/asio/strand.hpp>
#include <boost/system/detail/error_code.hpp>
#include <boost/system/system_error.hpp>

#include <atomic>
#include <chrono>
#include <cstddef>
#include <deque>
#include <functional>
#include <limits>
#include <memory>
#include <mutex>
#include <optional>
#include <sstream>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#ifndef BEAST_STATSDCOLLECTOR_TRACING_ENABLED
#define BEAST_STATSDCOLLECTOR_TRACING_ENABLED 0
#endif

namespace beast::insight {

namespace detail {

class StatsDCollectorImp;

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

class StatsDMetricBase : public List<StatsDMetricBase>::Node
{
public:
    virtual void
    doProcess() = 0;
    virtual ~StatsDMetricBase() = default;
    StatsDMetricBase() = default;
    StatsDMetricBase(StatsDMetricBase const&) = delete;
    StatsDMetricBase&
    operator=(StatsDMetricBase const&) = delete;
};

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

class StatsDHookImpl : public HookImpl, public StatsDMetricBase
{
public:
    StatsDHookImpl(HandlerType handler, std::shared_ptr<StatsDCollectorImp> impl);

    ~StatsDHookImpl() override;

    void
    doProcess() override;

    StatsDHookImpl&
    operator=(StatsDHookImpl const&) = delete;

private:
    std::shared_ptr<StatsDCollectorImp> impl_;
    HandlerType handler_;
};

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

class StatsDCounterImpl : public CounterImpl, public StatsDMetricBase
{
public:
    StatsDCounterImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl);

    ~StatsDCounterImpl() override;

    void
    increment(CounterImpl::value_type amount) override;

    void
    flush();
    void
    doIncrement(CounterImpl::value_type amount);
    void
    doProcess() override;

    StatsDCounterImpl&
    operator=(StatsDCounterImpl const&) = delete;

private:
    std::shared_ptr<StatsDCollectorImp> impl_;
    std::string name_;
    CounterImpl::value_type value_{0};
    bool dirty_{false};
};

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

class StatsDEventImpl : public EventImpl
{
public:
    StatsDEventImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl);

    ~StatsDEventImpl() override = default;

    void
    notify(EventImpl::value_type const& value) override;

    void
    doNotify(EventImpl::value_type const& value);
    void
    doProcess();

private:
    StatsDEventImpl&
    operator=(StatsDEventImpl const&);

    std::shared_ptr<StatsDCollectorImp> impl_;
    std::string name_;
};

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

class StatsDGaugeImpl : public GaugeImpl, public StatsDMetricBase
{
public:
    StatsDGaugeImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl);

    ~StatsDGaugeImpl() override;

    void
    set(GaugeImpl::value_type value) override;
    void
    increment(GaugeImpl::difference_type amount) override;

    void
    flush();
    void
    doSet(GaugeImpl::value_type value);
    void
    doIncrement(GaugeImpl::difference_type amount);
    void
    doProcess() override;

    StatsDGaugeImpl&
    operator=(StatsDGaugeImpl const&) = delete;

private:
    std::shared_ptr<StatsDCollectorImp> impl_;
    std::string name_;
    GaugeImpl::value_type lastValue_{0};
    GaugeImpl::value_type value_{0};
    bool dirty_{true};
};

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

class StatsDMeterImpl : public MeterImpl, public StatsDMetricBase
{
public:
    explicit StatsDMeterImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl);

    ~StatsDMeterImpl() override;

    void
    increment(MeterImpl::value_type amount) override;

    void
    flush();
    void
    doIncrement(MeterImpl::value_type amount);
    void
    doProcess() override;

    StatsDMeterImpl&
    operator=(StatsDMeterImpl const&) = delete;

private:
    std::shared_ptr<StatsDCollectorImp> impl_;
    std::string name_;
    MeterImpl::value_type value_{0};
    bool dirty_{false};
};

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

class StatsDCollectorImp : public StatsDCollector,
                           public std::enable_shared_from_this<StatsDCollectorImp>
{
private:
    static constexpr auto kMaxPacketSize = 1472;

    Journal journal_;
    ip::Endpoint address_;
    std::string prefix_;
    boost::asio::io_context ioContext_;
    std::optional<boost::asio::executor_work_guard<boost::asio::io_context::executor_type>> work_;
    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
    boost::asio::basic_waitable_timer<std::chrono::steady_clock> timer_;
    boost::asio::ip::udp::socket socket_;
    std::deque<std::string> data_;
    std::recursive_mutex metricsLock_;
    List<StatsDMetricBase> metrics_;

    /**
     * Whether hook handlers may be called. False until onCollectionReady(),
     * because the handlers read application services that are still being
     * constructed while this collector exists.
     */
    std::atomic<bool> polling_{false};

    // Must come last for order of init
    std::thread thread_;

    static boost::asio::ip::udp::endpoint
    toEndpoint(ip::Endpoint const& ep)
    {
        return boost::asio::ip::udp::endpoint(ep.address(), ep.port());
    }

public:
    StatsDCollectorImp(ip::Endpoint address, std::string prefix, Journal journal)
        : journal_(journal)
        , address_(std::move(address))
        , prefix_(std::move(prefix))
        , work_(boost::asio::make_work_guard(ioContext_))
        , strand_(boost::asio::make_strand(ioContext_))
        , timer_(ioContext_)
        , socket_(ioContext_)
        , thread_(&StatsDCollectorImp::run, this)
    {
    }

    ~StatsDCollectorImp() override
    {
        try
        {
            timer_.cancel();
        }
        catch (boost::system::system_error const&)  // NOLINT(bugprone-empty-catch)
        {
            // ignored
        }

        work_.reset();
        thread_.join();
    }

    void
    onCollectionReady() override
    {
        polling_.store(true, std::memory_order_release);
    }

    void
    onCollectionStopping() override
    {
        polling_.store(false, std::memory_order_release);

        // onTimer holds metricsLock_ across the handler loop, so acquiring it
        // here waits for a handler that is already running.
        std::scoped_lock const _(metricsLock_);
    }

    Hook
    makeHook(HookImpl::HandlerType const& handler) override
    {
        return Hook(std::make_shared<detail::StatsDHookImpl>(handler, shared_from_this()));
    }

    Counter
    makeCounter(std::string const& name) override
    {
        return Counter(std::make_shared<detail::StatsDCounterImpl>(name, shared_from_this()));
    }

    Event
    makeEvent(std::string const& name) override
    {
        return Event(std::make_shared<detail::StatsDEventImpl>(name, shared_from_this()));
    }

    Gauge
    makeGauge(std::string const& name) override
    {
        return Gauge(std::make_shared<detail::StatsDGaugeImpl>(name, shared_from_this()));
    }

    Meter
    makeMeter(std::string const& name) override
    {
        return Meter(std::make_shared<detail::StatsDMeterImpl>(name, shared_from_this()));
    }

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

    void
    add(StatsDMetricBase& metric)
    {
        std::scoped_lock const _(metricsLock_);
        metrics_.pushBack(metric);
    }

    void
    remove(StatsDMetricBase& metric)
    {
        std::scoped_lock const _(metricsLock_);
        metrics_.erase(metrics_.iteratorTo(metric));
    }

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

    boost::asio::io_context&
    getIoContext()
    {
        return ioContext_;
    }

    std::string const&
    prefix() const
    {
        return prefix_;
    }

    void
    doPostBuffer(std::string const& buffer)
    {
        data_.emplace_back(buffer);
    }

    void
    postBuffer(std::string&& buffer)
    {
        boost::asio::dispatch(
            ioContext_, boost::asio::bind_executor(strand_, [this, buffer = std::move(buffer)] {
                doPostBuffer(buffer);
            }));
    }

    // The keepAlive parameter makes sure the buffers sent to
    // boost::asio::async_send do not go away until the call is finished
    void
    onSend(
        std::shared_ptr<std::deque<std::string>> /*keepAlive*/,
        boost::system::error_code ec,
        std::size_t)
    {
        if (ec == boost::asio::error::operation_aborted)
            return;

        if (ec)
        {
            if (auto stream = journal_.error())
                stream << "async_send failed: " << ec.message();
            return;
        }
    }

    static void
    log(std::vector<boost::asio::const_buffer> const& buffers)
    {
        (void)buffers;
#if BEAST_STATSDCOLLECTOR_TRACING_ENABLED
        for (auto const& buffer : buffers)
        {
            std::string const s(buffer.data(), boost::asio::buffer_size(buffer));
            std::cerr << s;
        }
        std::cerr << '\n';
#endif
    }

    // Send what we have
    void
    sendBuffers()
    {
        if (data_.empty())
            return;

        // Break up the array of strings into blocks
        // that each fit into one UDP packet.
        //
        std::vector<boost::asio::const_buffer> buffers;
        buffers.reserve(data_.size());
        std::size_t size(0);

        auto keepAlive = std::make_shared<std::deque<std::string>>(std::move(data_));
        data_.clear();

        for (auto const& s : *keepAlive)
        {
            std::size_t const length(s.size());
            XRPL_ASSERT(
                !s.empty(),
                "beast::insight::detail::StatsDCollectorImp::sendBuffers : "
                "non-empty payload");
            if (!buffers.empty() && (size + length) > kMaxPacketSize)
            {
                log(buffers);
                socket_.async_send(
                    buffers,
                    [this, keepAlive](
                        boost::system::error_code const& ec, std::size_t bytesTransferred) {
                        onSend(keepAlive, ec, bytesTransferred);
                    });
                buffers.clear();
                size = 0;
            }

            buffers.emplace_back(&s[0], length);
            size += length;
        }

        if (!buffers.empty())
        {
            log(buffers);
            socket_.async_send(
                buffers,
                [this, keepAlive](
                    boost::system::error_code const& ec, std::size_t bytesTransferred) {
                    onSend(keepAlive, ec, bytesTransferred);
                });
        }
    }

    void
    setTimer()
    {
        using namespace std::chrono_literals;
        timer_.expires_after(1s);
        timer_.async_wait([this](boost::system::error_code const& ec) { onTimer(ec); });
    }

    void
    onTimer(boost::system::error_code ec)
    {
        if (ec == boost::asio::error::operation_aborted)
            return;

        if (ec)
        {
            if (auto stream = journal_.error())
                stream << "onTimer failed: " << ec.message();
            return;
        }

        if (polling_.load(std::memory_order_acquire))
        {
            std::scoped_lock const _(metricsLock_);

            for (auto& m : metrics_)
                m.doProcess();

            sendBuffers();
        }

        setTimer();
    }

    void
    run()
    {
        boost::system::error_code ec;

        if (socket_.connect(toEndpoint(address_), ec))
        {
            if (auto stream = journal_.error())
                stream << "Connect failed: " << ec.message();
            return;
        }

        setTimer();

        ioContext_.run();

        // NOLINTNEXTLINE(bugprone-unused-return-value)
        socket_.shutdown(boost::asio::ip::udp::socket::shutdown_send, ec);

        socket_.close();

        ioContext_.poll();
    }
};

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

StatsDHookImpl::StatsDHookImpl(HandlerType handler, std::shared_ptr<StatsDCollectorImp> impl)
    : impl_(std::move(impl)), handler_(std::move(handler))
{
    impl_->add(*this);
}

StatsDHookImpl::~StatsDHookImpl()
{
    impl_->remove(*this);
}

void
StatsDHookImpl::doProcess()
{
    handler_();
}

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

StatsDCounterImpl::StatsDCounterImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl)
    : impl_(std::move(impl)), name_(std::move(name))
{
    impl_->add(*this);
}

StatsDCounterImpl::~StatsDCounterImpl()
{
    impl_->remove(*this);
}

void
StatsDCounterImpl::increment(CounterImpl::value_type amount)
{
    boost::asio::dispatch(
        impl_->getIoContext(),
        [self = std::static_pointer_cast<StatsDCounterImpl>(shared_from_this()), amount] {
            self->doIncrement(amount);
        });
}

void
StatsDCounterImpl::flush()
{
    if (dirty_)
    {
        dirty_ = false;
        std::stringstream ss;
        ss << impl_->prefix() << "." << name_ << ":" << value_ << "|c"
           << "\n";
        value_ = 0;
        impl_->postBuffer(ss.str());
    }
}

void
StatsDCounterImpl::doIncrement(CounterImpl::value_type amount)
{
    value_ += amount;
    dirty_ = true;
}

void
StatsDCounterImpl::doProcess()
{
    flush();
}

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

StatsDEventImpl::StatsDEventImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl)
    : impl_(std::move(impl)), name_(std::move(name))
{
}

void
StatsDEventImpl::notify(EventImpl::value_type const& value)
{
    boost::asio::dispatch(
        impl_->getIoContext(),
        [self = std::static_pointer_cast<StatsDEventImpl>(shared_from_this()), value] {
            self->doNotify(value);
        });
}

void
StatsDEventImpl::doNotify(EventImpl::value_type const& value)
{
    std::stringstream ss;
    ss << impl_->prefix() << "." << name_ << ":" << value.count() << "|ms"
       << "\n";
    impl_->postBuffer(ss.str());
}

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

StatsDGaugeImpl::StatsDGaugeImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl)
    : impl_(std::move(impl)), name_(std::move(name))
{
    // Start dirty so the initial value (0) is emitted on the first flush.
    // Without this, gauges whose value never changes from 0 would never
    // appear in downstream metric stores (e.g. Prometheus via StatsD).
    impl_->add(*this);
}

StatsDGaugeImpl::~StatsDGaugeImpl()
{
    impl_->remove(*this);
}

void
StatsDGaugeImpl::set(GaugeImpl::value_type value)
{
    boost::asio::dispatch(
        impl_->getIoContext(),
        [self = std::static_pointer_cast<StatsDGaugeImpl>(shared_from_this()), value] {
            self->doSet(value);
        });
}

void
StatsDGaugeImpl::increment(GaugeImpl::difference_type amount)
{
    boost::asio::dispatch(
        impl_->getIoContext(),
        [self = std::static_pointer_cast<StatsDGaugeImpl>(shared_from_this()), amount] {
            self->doIncrement(amount);
        });
}

void
StatsDGaugeImpl::flush()
{
    if (dirty_)
    {
        dirty_ = false;
        std::stringstream ss;
        ss << impl_->prefix() << "." << name_ << ":" << value_ << "|g"
           << "\n";
        impl_->postBuffer(ss.str());
    }
}

void
StatsDGaugeImpl::doSet(GaugeImpl::value_type value)
{
    value_ = value;

    if (value_ != lastValue_)
    {
        lastValue_ = value_;
        dirty_ = true;
    }
}

void
StatsDGaugeImpl::doIncrement(GaugeImpl::difference_type amount)
{
    GaugeImpl::value_type value(value_);

    if (amount > 0)
    {
        auto const d = static_cast<GaugeImpl::value_type>(amount);
        value += (d >= std::numeric_limits<GaugeImpl::value_type>::max() - value_)
            ? std::numeric_limits<GaugeImpl::value_type>::max() - value_
            : d;
    }
    else if (amount < 0)
    {
        auto const d = static_cast<GaugeImpl::value_type>(-amount);
        value = (d >= value) ? 0 : value - d;
    }

    doSet(value);
}

void
StatsDGaugeImpl::doProcess()
{
    flush();
}

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

StatsDMeterImpl::StatsDMeterImpl(std::string name, std::shared_ptr<StatsDCollectorImp> impl)
    : impl_(std::move(impl)), name_(std::move(name))
{
    impl_->add(*this);
}

StatsDMeterImpl::~StatsDMeterImpl()
{
    impl_->remove(*this);
}

void
StatsDMeterImpl::increment(MeterImpl::value_type amount)
{
    boost::asio::dispatch(
        impl_->getIoContext(),
        [self = std::static_pointer_cast<StatsDMeterImpl>(shared_from_this()), amount] {
            self->doIncrement(amount);
        });
}

void
StatsDMeterImpl::flush()
{
    if (dirty_)
    {
        dirty_ = false;
        std::stringstream ss;
        ss << impl_->prefix() << "." << name_ << ":" << value_ << "|m"
           << "\n";
        value_ = 0;
        impl_->postBuffer(ss.str());
    }
}

void
StatsDMeterImpl::doIncrement(MeterImpl::value_type amount)
{
    value_ += amount;
    dirty_ = true;
}

void
StatsDMeterImpl::doProcess()
{
    flush();
}

}  // namespace detail

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

std::shared_ptr<StatsDCollector>
StatsDCollector::make(ip::Endpoint const& address, std::string const& prefix, Journal journal)
{
    return std::make_shared<detail::StatsDCollectorImp>(address, prefix, journal);
}

}  // namespace beast::insight