ETLSource.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2020 Ripple Labs Inc.
 
    Permission to use, copy, modify, and/or distribute this software for any
    purpose  with  or without fee is hereby granted, provided that the above
    copyright notice and this permission notice appear in all copies.
 
    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
    ANY  SPECIAL ,  DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
//==============================================================================
 
#ifndef RIPPLE_APP_REPORTING_ETLSOURCE_H_INCLUDED
#define RIPPLE_APP_REPORTING_ETLSOURCE_H_INCLUDED
#include <ripple/app/main/Application.h>
#include <ripple/app/reporting/ETLHelpers.h>
#include <ripple/protocol/STLedgerEntry.h>
#include <ripple/rpc/Context.h>
 
#include <boost/algorithm/string.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/core/string.hpp>
#include <boost/beast/websocket.hpp>
 
#include "org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h"
#include <grpcpp/grpcpp.h>
 
namespace ripple {
 
class ReportingETL;
 
class ETLSource
{
    std::string ip_;
 
    std::string wsPort_;
 
    std::string grpcPort_;
 
    ReportingETL& etl_;
 
    // a reference to the applications io_service
    boost::asio::io_context& ioc_;
 
    std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub> stub_;
 
    std::unique_ptr<boost::beast::websocket::stream<boost::beast::tcp_stream>>
        ws_;
    boost::asio::ip::tcp::resolver resolver_;
 
    boost::beast::flat_buffer readBuffer_;
 
    std::vector<std::pair<uint32_t, uint32_t>> validatedLedgers_;
 
    std::string validatedLedgersRaw_;
 
    NetworkValidatedLedgers& networkValidatedLedgers_;
 
    beast::Journal journal_;
 
    Application& app_;
 
    mutable std::mutex mtx_;
 
    size_t numFailures_ = 0;
 
    std::atomic_bool closing_ = false;
 
    std::atomic_bool connected_ = false;
 
    // true if this ETL source is forwarding transactions received on the
    // transactions_proposed stream. There are usually multiple ETL sources,
    // so to avoid forwarding the same transaction multiple times, we only
    // forward from one particular ETL source at a time.
    std::atomic_bool forwardingStream_ = false;
 
    // The last time a message was received on the ledgers stream
    std::chrono::system_clock::time_point lastMsgTime_;
    mutable std::mutex lastMsgTimeMtx_;
 
    // used for retrying connections
    boost::asio::steady_timer timer_;
 
public:
    bool
    isConnected() const
    {
        return connected_;
    }
 
    std::chrono::system_clock::time_point
    getLastMsgTime() const
    {
        std::lock_guard lck(lastMsgTimeMtx_);
        return lastMsgTime_;
    }
 
    void
    setLastMsgTime()
    {
        std::lock_guard lck(lastMsgTimeMtx_);
        lastMsgTime_ = std::chrono::system_clock::now();
    }
 
    ETLSource(std::string ip, std::string wsPort, ReportingETL& etl);
 
    ETLSource(
        std::string ip,
        std::string wsPort,
        std::string grpcPort,
        ReportingETL& etl);
 
    bool
    hasLedger(uint32_t sequence) const
    {
        std::lock_guard lck(mtx_);
        for (auto& pair : validatedLedgers_)
        {
            if (sequence >= pair.first && sequence <= pair.second)
            {
                return true;
            }
            else if (sequence < pair.first)
            {
                // validatedLedgers_ is a sorted list of disjoint ranges
                // if the sequence comes before this range, the sequence will
                // come before all subsequent ranges
                return false;
            }
        }
        return false;
    }
 
    void
    setValidatedRange(std::string const& range)
    {
        std::vector<std::pair<uint32_t, uint32_t>> pairs;
        std::vector<std::string> ranges;
        boost::split(ranges, range, boost::is_any_of(","));
        for (auto& pair : ranges)
        {
            std::vector<std::string> minAndMax;
 
            boost::split(minAndMax, pair, boost::is_any_of("-"));
 
            if (minAndMax.size() == 1)
            {
                uint32_t sequence = std::stoll(minAndMax[0]);
                pairs.push_back(std::make_pair(sequence, sequence));
            }
            else
            {
                assert(minAndMax.size() == 2);
                uint32_t min = std::stoll(minAndMax[0]);
                uint32_t max = std::stoll(minAndMax[1]);
                pairs.push_back(std::make_pair(min, max));
            }
        }
        std::sort(pairs.begin(), pairs.end(), [](auto left, auto right) {
            return left.first < right.first;
        });
 
        // we only hold the lock here, to avoid blocking while string processing
        std::lock_guard lck(mtx_);
        validatedLedgers_ = std::move(pairs);
        validatedLedgersRaw_ = range;
    }
 
    std::string
    getValidatedRange() const
    {
        std::lock_guard lck(mtx_);
 
        return validatedLedgersRaw_;
    }
 
    void
    stop()
    {
        JLOG(journal_.debug()) << __func__ << " : "
                               << "Closing websocket";
 
        assert(ws_);
        close(false);
    }
 
    std::pair<grpc::Status, org::xrpl::rpc::v1::GetLedgerResponse>
    fetchLedger(uint32_t ledgerSequence, bool getObjects = true);
 
    std::string
    toString() const
    {
        return "{ validated_ledger : " + getValidatedRange() +
            " , ip : " + ip_ + " , web socket port : " + wsPort_ +
            ", grpc port : " + grpcPort_ + " }";
    }
 
    Json::Value
    toJson() const
    {
        Json::Value result(Json::objectValue);
        result["connected"] = connected_.load();
        result["validated_ledgers_range"] = getValidatedRange();
        result["ip"] = ip_;
        result["websocket_port"] = wsPort_;
        result["grpc_port"] = grpcPort_;
        auto last = getLastMsgTime();
        if (last.time_since_epoch().count() != 0)
            result["last_message_arrival_time"] =
                to_string(std::chrono::floor<std::chrono::microseconds>(last));
        return result;
    }
 
    bool
    loadInitialLedger(
        uint32_t ledgerSequence,
        ThreadSafeQueue<std::shared_ptr<SLE>>& writeQueue);
 
    void
    start();
 
    void
    reconnect(boost::beast::error_code ec);
 
    void
    onResolve(
        boost::beast::error_code ec,
        boost::asio::ip::tcp::resolver::results_type results);
 
    void
    onConnect(
        boost::beast::error_code ec,
        boost::asio::ip::tcp::resolver::results_type::endpoint_type endpoint);
 
    void
    onHandshake(boost::beast::error_code ec);
 
    void
    onWrite(boost::beast::error_code ec, size_t size);
 
    void
    onRead(boost::beast::error_code ec, size_t size);
 
    bool
    handleMessage();
 
    void
    close(bool startAgain);
 
    std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub>
    getP2pForwardingStub() const;
 
    Json::Value
    forwardToP2p(RPC::JsonContext& context) const;
};
 
class ETLLoadBalancer
{
private:
    ReportingETL& etl_;
 
    beast::Journal journal_;
 
    std::vector<std::unique_ptr<ETLSource>> sources_;
 
public:
    ETLLoadBalancer(ReportingETL& etl);
 
    void
    add(std::string& host, std::string& websocketPort, std::string& grpcPort);
 
    void
    add(std::string& host, std::string& websocketPort);
 
    void
    loadInitialLedger(
        uint32_t sequence,
        ThreadSafeQueue<std::shared_ptr<SLE>>& writeQueue);
 
    std::optional<org::xrpl::rpc::v1::GetLedgerResponse>
    fetchLedger(uint32_t ledgerSequence, bool getObjects);
 
    void
    start();
 
    void
    stop();
 
    bool
    shouldPropagateStream(ETLSource* in) const
    {
        for (auto& src : sources_)
        {
            assert(src);
            // We pick the first ETLSource encountered that is connected
            if (src->isConnected())
            {
                if (src.get() == in)
                    return true;
                else
                    return false;
            }
        }
 
        // If no sources connected, then this stream has not been forwarded.
        return true;
    }
 
    Json::Value
    toJson() const
    {
        Json::Value ret(Json::arrayValue);
        for (auto& src : sources_)
        {
            ret.append(src->toJson());
        }
        return ret;
    }
 
    std::unique_ptr<org::xrpl::rpc::v1::XRPLedgerAPIService::Stub>
    getP2pForwardingStub() const;
 
    Json::Value
    forwardToP2p(RPC::JsonContext& context) const;
 
private:
    template <class Func>
    bool
    execute(Func f, uint32_t ledgerSequence);
};
 
}  // namespace ripple
#endif