e2sarCP.hpp

#ifndef E2SARCPHPP
#define E2SARCPHPP
#include <vector>
#include <memory>
#include <boost/asio.hpp>
#include <boost/tuple/tuple.hpp>
 
#include <grpc/grpc.h>
#include <grpcpp/channel.h>
#include <grpcpp/client_context.h>
#include <grpcpp/create_channel.h>
#include <grpcpp/security/credentials.h>
 
#include <google/protobuf/util/time_util.h>
#include <boost/date_time/posix_time/posix_time.hpp>
 
#include "grpc/loadbalancer.grpc.pb.h"
 
#include "e2sarUtil.hpp"
 
using grpc::Channel;
using grpc::ClientContext;
using grpc::ClientReader;
using grpc::ClientReaderWriter;
using grpc::ClientWriter;
using grpc::Status;
 
using loadbalancer::LoadBalancer;
 
using loadbalancer::LoadBalancerStatusReply;
using loadbalancer::WorkerStatus;
using loadbalancer::OverviewReply;
 
using google::protobuf::Timestamp;
 
// default reservation duration for a load balancer in hours
#define DEFAULT_LB_RESERVE_DURATION 24
 
// change to '1' to test older versions of UDPLBd where token
// was sent as a parameter in the body. The new way is to send
// it in the authorization header as bearer token
#ifndef TOKEN_IN_BODY
#define TOKEN_IN_BODY 0
#endif
 
/***
 * Control Plane definitions for E2SAR
 */
 
namespace e2sar
{
    using TimeUntil = google::protobuf::Timestamp;
 
    struct LBWorkerStatus
    {
        std::string name;
        float fillPercent;
        float controlSignal;
        u_int32_t slotsAssigned;
        google::protobuf::Timestamp lastUpdated;
 
        LBWorkerStatus(const std::string &n, float fp, float cs, u_int32_t sa, google::protobuf::Timestamp lu) : name{n}, fillPercent{fp}, controlSignal{cs}, slotsAssigned{sa}, lastUpdated{lu} {}
    };
 
    struct LBStatus
    {
        google::protobuf::Timestamp timestamp;
        u_int64_t currentEpoch;
        u_int64_t currentPredictedEventNumber;
        std::vector<WorkerStatus> workers;
        std::vector<std::string> senderAddresses;
        google::protobuf::Timestamp expiresAt;
 
        LBStatus(google::protobuf::Timestamp ts, u_int64_t ce, u_int64_t penum, std::vector<WorkerStatus> &w, std::vector<std::string> &sa, google::protobuf::Timestamp exp):
        timestamp{ts}, currentEpoch{ce}, currentPredictedEventNumber{penum}, expiresAt{exp}
        {
            workers = std::move(w);
            senderAddresses = std::move(sa);
        }
        LBStatus() {}
    };
 
    struct OverviewEntry
    {
        std::string name; // name passed in reserveLB
        std::string lbid; // load balancer id
        std::pair<ip::address, u_int16_t> syncAddressAndPort;
        ip::address dataIPv4;
        ip::address dataIPv6; 
        u_int32_t fpgaLBId; 
        LBStatus status; // same as in lbstatus call
 
        OverviewEntry() {}
    };
    using OverviewMessage = std::vector<OverviewEntry>;
 
    class LBManager
    {
    private:
        EjfatURI _cpuri;
        std::string addr_string;
        std::unique_ptr<LoadBalancer::Stub> _stub;
        std::shared_ptr<grpc::Channel> _channel;
 
    protected:
    public:
        LBManager(const EjfatURI &cpuri, bool validateServer = true, bool useHostAddress = false,
                  grpc::SslCredentialsOptions opts = grpc::SslCredentialsOptions()) : _cpuri(cpuri)
        {
 
            auto cp_host_r = cpuri.get_cpHost();
 
            // using host address automatically disables cert validation
            if (useHostAddress)
                validateServer = false;
 
            if (!useHostAddress && !cp_host_r.has_error())
            {
                // try hostname
                auto cp_host_v = cp_host_r.value();
                addr_string = cp_host_v.first + ":"s + std::to_string(cp_host_v.second);
            }
            else
            {
                // try address
                auto cp_addr_r = cpuri.get_cpAddr();
                if (cp_addr_r.has_error())
                    throw E2SARException("Unable to initialize LBManager due to missing CP address in URI");
                auto cp_addr_v = cp_addr_r.value();
                if (cp_addr_v.first.is_v4())
                    addr_string = "ipv4:///" + cp_addr_v.first.to_string() + ":" + std::to_string(cp_addr_v.second);
                else
                    addr_string = "ipv6:///[" + cp_addr_v.first.to_string() + "]:" + std::to_string(cp_addr_v.second);
            }
 
            if (cpuri.get_useTls())
            {
                grpc::experimental::TlsChannelCredentialsOptions topts;
                if (!validateServer)
                {
                    // disable most of server certificate validation
                    std::shared_ptr<grpc::experimental::NoOpCertificateVerifier> verifier = std::make_shared<grpc::experimental::NoOpCertificateVerifier>();
                    topts.set_verify_server_certs(false);
                    topts.set_check_call_host(false);
                    topts.set_certificate_verifier(verifier);
                    _channel = grpc::CreateChannel(addr_string, grpc::experimental::TlsCredentials(topts));
                }
                else
                {
                    // use provided SSL options
                    _channel = grpc::CreateChannel(addr_string, grpc::SslCredentials(opts));
                }
            }
            else
            {
                _channel = grpc::CreateChannel(addr_string, grpc::InsecureChannelCredentials());
            }
            _stub = LoadBalancer::NewStub(_channel);
        }
 
        result<u_int32_t> reserveLB(const std::string &lb_name,
                              const TimeUntil &until,
                              const std::vector<std::string> &senders) noexcept;
 
        result<u_int32_t> reserveLB(const std::string &lb_name,
                              const boost::posix_time::time_duration &duration,
                              const std::vector<std::string> &senders) noexcept;
 
        result<u_int32_t> reserveLB(const std::string &lb_name,
                              const double &durationSeconds,
                              const std::vector<std::string> &senders) noexcept;
 
        result<int> getLB(const std::string &lbid) noexcept;
        result<int> getLB() noexcept;
 
        result<std::unique_ptr<LoadBalancerStatusReply>> getLBStatus(const std::string &lbid) noexcept;
 
        result<std::unique_ptr<LoadBalancerStatusReply>> getLBStatus() noexcept;
 
        result<std::unique_ptr<OverviewReply>> overview() noexcept;
 
        result<int> addSenders(const std::vector<std::string>& senders) noexcept;
 
        result<int> removeSenders(const std::vector<std::string>& senders) noexcept;
 
        result<int> addSenderSelf(bool v6=false) noexcept;
 
        result<int> removeSenderSelf(bool v6=false) noexcept;
 
        static inline std::vector<WorkerStatus> get_WorkerStatusVector(std::unique_ptr<LoadBalancerStatusReply> &rep) noexcept
        {
            std::vector<WorkerStatus> ret(rep->workers_size());
 
            size_t j{0};
            for (auto i = rep->workers().begin(); i != rep->workers().end(); ++i, j++)
            {
                ret[j].CopyFrom(*i);
            }
            return ret;
        }
 
        static inline std::vector<WorkerStatus> get_WorkerStatusVector(const LoadBalancerStatusReply &rep) noexcept
        {
            std::vector<WorkerStatus> ret(rep.workers_size());
 
            size_t j{0};
            for (auto i = rep.workers().begin(); i != rep.workers().end(); ++i, j++)
            {
                ret[j].CopyFrom(*i);
            }
            return ret;
        }
 
        static inline std::vector<std::string> get_SenderAddressVector(std::unique_ptr<LoadBalancerStatusReply> &rep) noexcept
        {
            std::vector<std::string> ret(rep->senderaddresses_size());
 
            size_t j{0};
            for (auto i = rep->senderaddresses().begin(); i != rep->senderaddresses().end(); ++i, j++)
            {
                ret[j] = *i;
            }
            return ret;
        }
 
        static inline std::vector<std::string> get_SenderAddressVector(const LoadBalancerStatusReply &rep) noexcept
        {
            std::vector<std::string> ret(rep.senderaddresses_size());
 
            size_t j{0};
            for (auto i = rep.senderaddresses().begin(); i != rep.senderaddresses().end(); ++i, j++)
            {
                ret[j] = *i;
            }
            return ret;
        }
 
        static inline const std::unique_ptr<LBStatus> asLBStatus(std::unique_ptr<LoadBalancerStatusReply> &rep) noexcept
        {
            std::vector<std::string> addresses{get_SenderAddressVector(rep)};
            std::vector<WorkerStatus> workers{get_WorkerStatusVector(rep)};
            std::unique_ptr<LBStatus> pret = std::make_unique<LBStatus>(rep->timestamp(), rep->currentepoch(), rep->currentpredictedeventnumber(),
            workers, addresses, rep->expiresat());
            return pret;
        }
 
        static inline const LBStatus asLBStatus(const LoadBalancerStatusReply &rep) noexcept
        {
            std::vector<std::string> addresses{get_SenderAddressVector(rep)};
            std::vector<WorkerStatus> workers{get_WorkerStatusVector(rep)};
            return LBStatus(rep.timestamp(), rep.currentepoch(), rep.currentpredictedeventnumber(),
                workers, addresses, rep.expiresat());
        }
 
        static inline const OverviewMessage asOverviewMessage(std::unique_ptr<OverviewReply> &rep) noexcept
        {
            size_t j{0};
            OverviewMessage om(rep.get()->loadbalancers_size());
            for (auto i = rep->loadbalancers().begin(); i != rep->loadbalancers().end(); ++i, j++)
            {
                om[j].name = i->name();
                om[j].lbid = i->reservation().lbid();
                om[j].syncAddressAndPort.first = ip::make_address(i->reservation().syncipaddress());
                om[j].syncAddressAndPort.second = i->reservation().syncudpport();
                om[j].dataIPv4 = ip::make_address(i->reservation().dataipv4address());
                om[j].dataIPv6 = ip::make_address(i->reservation().dataipv6address());
                om[j].fpgaLBId = i->reservation().fpgalbid();
                om[j].status = asLBStatus(i->status());
            }
            return om;
        }
 
        static inline const OverviewMessage asOverviewMessage(const OverviewReply &rep) noexcept
        {
            size_t j{0};
            OverviewMessage om(rep.loadbalancers_size());
            for (auto i = rep.loadbalancers().begin(); i != rep.loadbalancers().end(); ++i, j++)
            {
                om[j].name = i->name();
                om[j].lbid = i->reservation().lbid();
                om[j].syncAddressAndPort.first = ip::make_address(i->reservation().syncipaddress());
                om[j].syncAddressAndPort.second = i->reservation().syncudpport();
                om[j].dataIPv4 = ip::make_address(i->reservation().dataipv4address());
                om[j].dataIPv6 = ip::make_address(i->reservation().dataipv6address());
                om[j].fpgaLBId = i->reservation().fpgalbid();
                om[j].status = asLBStatus(i->status());
            }
            return om;
        }
 
        result<int> freeLB(const std::string &lbid) noexcept;
        result<int> freeLB() noexcept;
 
        result<int> registerWorker(const std::string &node_name, std::pair<ip::address, u_int16_t> node_ip_port, float weight, u_int16_t source_count, float min_factor, float max_factor) noexcept;
 
        result<int> registerWorkerSelf(const std::string &node_name, u_int16_t node_port, float weight, u_int16_t source_count, float min_factor, float max_factor, bool v6=false) noexcept;
 
        result<int> deregisterWorker() noexcept;
 
        result<int> sendState(float fill_percent, float control_signal, bool is_ready) noexcept;
 
        result<int> sendState(float fill_percent, float control_signal, bool is_ready, const Timestamp &ts) noexcept;
 
        result<boost::tuple<std::string, std::string, std::string>> version() noexcept;
 
        inline const EjfatURI &get_URI() const noexcept { return _cpuri; }
 
        static inline result<grpc::SslCredentialsOptions> makeSslOptions(const std::string &pem_root_certs,
                                                                         const std::string &pem_private_key,
                                                                         const std::string &pem_cert_chain) noexcept
        {
            return grpc::SslCredentialsOptions{std::move(pem_root_certs),
                                               std::move(pem_private_key),
                                               std::move(pem_cert_chain)};
        }
 
        static result<grpc::SslCredentialsOptions> makeSslOptionsFromFiles(
            std::string_view pem_root_certs,
            std::string_view pem_private_key,
            std::string_view pem_cert_chain) noexcept;
 
        static result<grpc::SslCredentialsOptions> makeSslOptionsFromFiles(
            std::string_view pem_root_certs) noexcept;
 
        inline std::string get_AddrString() {
            return addr_string;
        }
    };
 
    static inline int get_PortRange(int source_count) noexcept
    {
        // Based on the proto file enum for the load balancer, seen below,
        // map the max # of sources a backend will see to the PortRange value.
        // This is necessay to provide the control plane when registering.
 
        // Handle edge cases
        if (source_count < 2)
        {
            return 0;
        }
        else if (source_count > 16384)
        {
            return 14;
        }
 
        int maxCount = 2;
        int iteration = 1;
 
        while (source_count > maxCount)
        {
            iteration++;
            maxCount <<= 1;
        }
 
        return iteration;
    }
}
#endif