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
 
/***
 * Control Plane definitions for E2SAR
 */
 
namespace e2sar
{
    using TimeUntil = google::protobuf::Timestamp;
 
    struct WorkerStats {
        int64_t total_events_recv, total_events_reassembled, total_events_reassembly_err, total_events_dequeued,
            total_event_enqueue_err, total_bytes_recv, total_packets_recv;
        
        WorkerStats(): total_events_recv{0}, total_events_reassembled{0}, total_events_reassembly_err{0},
            total_events_dequeued{0}, total_event_enqueue_err{0}, total_bytes_recv{0}, total_packets_recv{0} {
                ;
        }
    };
 
    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> syncIPv4AndPort;
        std::pair<ip::address, u_int16_t> syncIPv6AndPort;
        ip::address dataIPv4;
        ip::address dataIPv6; 
        u_int32_t fpgaLBId; 
        u_int32_t dataMinPort;
        u_int32_t dataMaxPort;
        LBStatus status; // same as in lbstatus call
 
        OverviewEntry() {}
    };
    using OverviewMessage = std::vector<OverviewEntry>;
 
    struct FloatSample {
        int64_t timestamp_ms;  // milliseconds since epoch
        float value;
 
        FloatSample(int64_t ts, float v) : timestamp_ms{ts}, value{v} {}
        FloatSample() : timestamp_ms{0}, value{0.0f} {}
    };
 
    struct IntegerSample {
        int64_t timestamp_ms;  // milliseconds since epoch
        int64_t value;
 
        IntegerSample(int64_t ts, int64_t v) : timestamp_ms{ts}, value{v} {}
        IntegerSample() : timestamp_ms{0}, value{0} {}
    };
 
    struct TimeseriesData {
        std::string path;
        std::string unit;
        std::variant<std::vector<FloatSample>, std::vector<IntegerSample>> timeseries;
 
        TimeseriesData(const std::string& p, const std::string &u, std::vector<FloatSample>& fs):
            path{p}, unit{u}, timeseries{std::move(fs)} {}
        TimeseriesData(const std::string& p, const std::string &u, std::vector<IntegerSample>& is):
            path{p}, unit{u}, timeseries{std::move(is)} {}
    };
 
    // a collection of timeseries with a common since timestamp
    struct TimeseriesResult {
        int64_t since_ms;
        std::vector<TimeseriesData> td; // possibly multiple vectors of timeseries data returned by the query
 
        TimeseriesResult(int64_t _ts, std::vector<TimeseriesData> &_td): since_ms{_ts}, td{std::move(_td)} {}
 
        // be careful this is not a true copy constructor as it uses move to move the timeseries
        TimeseriesResult(TimeseriesResult&& _tsr): since_ms{_tsr.since_ms}, td{std::move(_tsr.td)} {}
    };
 
    struct TokenPermission {
        EjfatURI::TokenType resourceType;
        std::string resourceId;  // optional, can be empty
        EjfatURI::TokenPermission permission;
 
        TokenPermission():
            resourceType{EjfatURI::TokenType::all},
            resourceId{""},
            permission{EjfatURI::TokenPermission::_read_only_} {}
        TokenPermission(EjfatURI::TokenType rt, const std::string& rid, EjfatURI::TokenPermission pt):
            resourceType{rt}, resourceId{rid}, permission{pt} {}
    };
 
    struct TokenDetails {
        std::string name;
        std::vector<TokenPermission> permissions;
        std::string created_at;
        uint32_t id;
 
        TokenDetails(): name{""}, created_at{""}, id{0} {}
    };
 
    using TokenSelector = std::variant<uint32_t, std::string>;
 
    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,
                              int ip_family=loadbalancer::IpFamily::DUAL_STACK) noexcept;
 
        result<u_int32_t> reserveLB(const std::string &lb_name,
                              const boost::posix_time::time_duration &duration,
                              const std::vector<std::string> &senders,
                              int ip_family=loadbalancer::IpFamily::DUAL_STACK) noexcept;
 
        result<u_int32_t> reserveLB(const std::string &lb_name,
                              const double &durationSeconds,
                              const std::vector<std::string> &senders,
                              int ip_family=loadbalancer::IpFamily::DUAL_STACK) 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;
 
        result<std::string> createToken(
            const std::string &name,
            const std::vector<TokenPermission> &permissions) noexcept;
 
        result<TokenDetails> listTokenPermissions(const TokenSelector &target) noexcept;
 
        result<std::vector<TokenDetails>> listChildTokens(const TokenSelector &target) noexcept;
 
        result<int> revokeToken(const TokenSelector &target) noexcept;
 
        result<TimeseriesResult> timeseries(const std::string &path, const Timestamp &since) 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].syncIPv4AndPort.first = ip::make_address(i->reservation().syncipv4address());
                om[j].syncIPv4AndPort.second = i->reservation().syncudpport();
                om[j].syncIPv6AndPort.first = ip::make_address(i->reservation().syncipv6address());
                om[j].syncIPv6AndPort.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].dataMinPort = i->reservation().dataminport();
                om[j].dataMaxPort = i->reservation().datamaxport();
                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].syncIPv4AndPort.first = ip::make_address(i->reservation().syncipv4address());
                om[j].syncIPv4AndPort.second = i->reservation().syncudpport();
                om[j].syncIPv6AndPort.first = ip::make_address(i->reservation().syncipv6address());
                om[j].syncIPv6AndPort.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].dataMinPort = i->reservation().dataminport();
                om[j].dataMaxPort = i->reservation().datamaxport();
                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, bool keep_lb_header=false) 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, bool keep_lb_header=false) noexcept;
 
        result<int> deregisterWorker() noexcept;
 
        result<int> sendState(float fill_percent, float control_signal, bool is_ready,
                       const WorkerStats &stats) noexcept;
 
        result<int> sendState(float fill_percent, float control_signal, bool is_ready, const Timestamp &ts,
                       const WorkerStats &stats) 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