e2sarUtil.hpp

#ifndef E2SARUTILHPP
#define E2SARUTILHPP
 
#include <fstream>
#include <vector>
#include <boost/url.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/asio.hpp>
#include <boost/chrono.hpp>
#include <boost/thread.hpp>
#include <boost/log/trivial.hpp>
#include <boost/log/common.hpp>
#include <boost/log/sinks.hpp>
#include <boost/log/sources/logger.hpp>
#include <boost/log/expressions.hpp>
#include <boost/log/attributes.hpp>
#include <boost/log/support/date_time.hpp>
#include <boost/core/null_deleter.hpp>
#include <boost/shared_ptr.hpp>
 
#include "grpc/loadbalancer.grpc.pb.h"
 
#include "e2sarError.hpp"
 
using namespace boost::asio;
using namespace std::string_literals;
using namespace boost::log;
 
#define BOOST_LOG_FLUSH() sink->flush()
#define BOOST_MLL_START(PREF) { std::ostringstream PREF_ostr;
#define BOOST_MLL_LOG(PREF) PREF_ostr 
#define BOOST_MLL_STOP(PREF) BOOST_LOG_SEV(lg, trivial::info) << PREF_ostr.str(); } BOOST_LOG_FLUSH(); 
 
#define BOOST_LOG_INFO() BOOST_LOG_SEV(lg, trivial::info)
#define BOOST_LOG_WARN() BOOST_LOG_SEV(lg, trivial::warning)
#define BOOST_LOG_ERR() BOOST_LOG_SEV(lg, trivial::error)
 
/***
 * Supporting classes for E2SAR
 */
namespace e2sar
{
    typedef sinks::asynchronous_sink<sinks::text_ostream_backend> text_sink;
    extern boost::shared_ptr<text_sink> sink;
    extern sources::severity_logger_mt<trivial::severity_level> lg;
 
    const u_int16_t DATAPLANE_PORT = 19522;
    const u_int16_t DATAPLANE_PORT_MIN = 16384;
    const u_int16_t DATAPLANE_PORT_MAX = 32767;
 
    class EjfatURI
    {
    public:
        enum class TokenType: u_int16_t {
            // contains overloaded names according to old and new token type hierarchy
            all=0, admin=1, load_balancer=admin, instance=2, reservation=instance, session=3, END
        };
        inline static constexpr size_t ttAsIdx(TokenType tt) 
        {
            return static_cast<size_t>(tt);
        }
        inline static const std::string toString(TokenType tt)
        {
            switch(tt) {
                case TokenType::all: return "ALL"s;
                case TokenType::admin: return "LOAD_BALANCER"s;
                case TokenType::instance: return "RESERVATION"s;
                case TokenType::session: return "SESSION"s;
                default: return "UNKNOWN"s;
            }
        }
        static const size_t tokenTypeCardinality = static_cast<size_t>(TokenType::END);
 
 
        enum class TokenPermission: u_int16_t {
            // 'register' is a keyword in C++, so adding '_' to names
            _read_only_, _register_, _reserve_, _update_, END
        };
        inline static const std::string toString(TokenPermission tt)
        {
            switch(tt) {
                case TokenPermission::_read_only_: return "READ"s;
                case TokenPermission::_register_: return "REGISTER"s;
                case TokenPermission::_reserve_: return "RESERVE"s;
                case TokenPermission::_update_: return "UPDATE"s;
                default: return "UNKNOWN"s;
            }
        }
        static const size_t tokenPermissionCardinality = static_cast<size_t>(TokenPermission::END);
 
    private:
        std::string rawURI;
        bool haveDatav4;
        bool haveDatav6;
        bool haveSyncv4;
        bool haveSyncv6;
        bool useTls;
        bool preferV6;
 
        u_int16_t syncPortv4;
        u_int16_t syncPortv6;
        u_int16_t cpPort;
        u_int16_t dataMinPort;
        u_int16_t dataMaxPort;
 
        std::string lbName;
        std::string lbId;
        std::array<std::string, tokenTypeCardinality> tokensByType;
        std::string sessionId;
 
        ip::address dataAddrv4;
        ip::address dataAddrv6;
        ip::address syncAddrv4;
        ip::address syncAddrv6;
        ip::address cpAddr;
        std::string cpHost;
 
    public:
        EjfatURI(const std::string &uri, TokenType tt=TokenType::admin, bool preferV6=false);
 
        ~EjfatURI() {}
 
        friend bool operator== (const EjfatURI &u1, const EjfatURI &u2);
 
        friend inline bool operator!= (const EjfatURI &u1, const EjfatURI &u2) 
        {
            return !(u1 == u2);
        }
 
        inline bool get_useTls() const
        {
            return useTls;
        }
 
        inline void set_Token(const std::string &t, TokenType tt)
        {
            tokensByType[ttAsIdx(tt)] = t;
        }
 
        inline void set_InstanceToken(const std::string &t)
        {
            tokensByType[ttAsIdx(TokenType::instance)] = t;
        }
 
        inline void set_SessionToken(const std::string &t)
        {
            tokensByType[ttAsIdx(TokenType::session)] = t;
        }
 
        inline const result<std::string> get_InstanceToken() const
        {
            auto idx = ttAsIdx(TokenType::instance);
            if (!tokensByType[idx].empty())
                return tokensByType[idx];
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Instance token not available"s};
        }
 
        inline const result<std::string> get_SessionToken() const
        {
            auto idx = ttAsIdx(TokenType::session);
            if (!tokensByType[idx].empty())
                return tokensByType[idx];
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Session token not available"s};
        }
 
        inline const result<std::string> get_AdminToken() const
        {
            auto idx = ttAsIdx(TokenType::admin);
            if (!tokensByType[idx].empty())
                return tokensByType[idx];
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Admin token not available"s};
        }
 
        inline void set_lbName(const std::string &n)
        {
            lbName = n;
        }
 
        inline void set_lbId(const std::string &i)
        {
            lbId = i;
        }
 
        inline void set_sessionId(const std::string &i)
        {
            sessionId = i;
        }
 
        inline void set_syncAddr(const std::pair<ip::address, u_int16_t> &a)
        {
            if (a.first.is_v4()) {
                syncAddrv4 = a.first;
                syncPortv4 = a.second;
                haveSyncv4 = true;
            } else {
                syncAddrv6 = a.first;
                syncPortv6 = a.second;
                haveSyncv6 = true;
            }
        }
 
        inline void set_dataAddr(const std::pair<ip::address, std::pair<u_int16_t, u_int16_t>> &a)
        {
            if (a.first.is_v4()) {
                dataAddrv4 = a.first;
                haveDatav4 = true;
            }
            else {
                dataAddrv6 = a.first;
                haveDatav6 = true;
            }
            dataMinPort = a.second.first;
            dataMaxPort = a.second.second;
        }
 
        inline void set_dataPortRange(u_int16_t minPort, u_int16_t maxPort)
        {
            dataMinPort = minPort;
            dataMaxPort = maxPort;
        }
 
        inline const std::string get_lbName() const
        {
            return lbName;
        }
 
        inline const std::string get_lbId() const
        {
            return lbId;
        }
 
        inline const std::string get_sessionId() const 
        {
            return sessionId;
        }
 
        inline const result<std::pair<ip::address, u_int16_t>> get_cpAddr() const
        {
            return std::pair<ip::address, u_int16_t>(cpAddr, cpPort);
        }
 
        inline const result<std::pair<std::string, u_int16_t>> get_cpHost() const
        {
            if (!cpHost.empty())
                return std::pair<std::string, u_int16_t>(cpHost, cpPort);
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Control plane hostname not available"s};
        }
 
        inline const bool has_dataAddrv4() const
        {
            return haveDatav4;
        }
 
        inline const bool has_dataAddrv6() const
        {
            return haveDatav6;
        }
 
        inline const bool has_dataAddr() const
        {
            return haveDatav4 || haveDatav6;
        }
 
        inline const bool has_syncAddr() const
        {
            return haveSyncv4 || haveSyncv6;
        }
 
        inline const bool has_syncAddrv4() const
        {
            return haveSyncv4;
        }
 
        inline const bool has_syncAddrv6() const
        {
            return haveSyncv6;
        }
 
        inline const result<std::pair<ip::address, std::pair<u_int16_t, u_int16_t>>> get_dataAddrv4() const noexcept
        {
            if (haveDatav4)
                return std::make_pair(dataAddrv4, std::make_pair(dataMinPort, dataMaxPort));
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Data plane address not available"s};
        }
 
        inline const result<std::pair<ip::address, std::pair<u_int16_t, u_int16_t>>> get_dataAddrv6() const noexcept
        {
            if (haveDatav6)
                return std::make_pair(dataAddrv6, std::make_pair(dataMinPort, dataMaxPort));
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Data plane address not available"s};
        }
 
        inline const result<std::pair<u_int16_t, u_int16_t>> get_dataPortRange() const noexcept
        {
            if (haveDatav4 || haveDatav6)
                return std::pair<u_int16_t, u_int16_t>(dataMinPort, dataMaxPort);
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Data plane address not available"s};
        }
 
        inline const result<std::pair<ip::address, u_int16_t>> get_syncAddrv4() const noexcept
        {
            if (haveSyncv4)
                return std::pair<ip::address, u_int16_t>(syncAddrv4, syncPortv4);
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Sync v4 address not available"s};
        }
 
        inline const result<std::pair<ip::address, u_int16_t>> get_syncAddrv6() const noexcept
        {
            if (haveSyncv6)
                return std::pair<ip::address, u_int16_t>(syncAddrv6, syncPortv6);
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Sync v6 address not available"s};
        }
 
        inline const result<std::pair<ip::address, u_int16_t>> get_syncAddr() const noexcept
        {
            if (preferV6 && haveSyncv6)
                return std::pair<ip::address, u_int16_t>(syncAddrv6, syncPortv6);
            if (haveSyncv4)
                return std::pair<ip::address, u_int16_t>(syncAddrv4, syncPortv4);
            if (haveSyncv6)
                return std::pair<ip::address, u_int16_t>(syncAddrv6, syncPortv6);
            return E2SARErrorInfo{E2SARErrorc::ParameterNotAvailable, "Sync address not available"s};
        }
        operator std::string() const;
 
        const std::string to_string(TokenType tt = TokenType::admin) const;
 
        static inline result<EjfatURI> getFromEnv(const std::string &envVar = "EJFAT_URI"s, 
            TokenType tt=TokenType::admin, bool preferV6=false) noexcept
        {
            const char *envStr = std::getenv(envVar.c_str());
            if (envStr != nullptr)
            {
                try
                {
                    return EjfatURI(envStr, tt, preferV6);
                }
                catch (const E2SARException &e)
                {
                    return E2SARErrorInfo{E2SARErrorc::CaughtException, "Unable to parse EJFAT_URI from environment variable: "s + static_cast<std::string>(e)};
                }
            }
            return E2SARErrorInfo{E2SARErrorc::Undefined, "Environment variable "s + envVar + " not defined."s};
        }
 
        static inline result<EjfatURI> getFromString(const std::string &uriStr, 
            TokenType tt=TokenType::admin, bool preferV6=false) noexcept
        {
            try
            {
                return EjfatURI(uriStr, tt, preferV6);
            }
            catch (const E2SARException &e)
            {
                return E2SARErrorInfo{E2SARErrorc::CaughtException, "Unable to parse URI from string: "s + static_cast<std::string>(e)};
            }
        }
 
        static inline result<EjfatURI> getFromFile(const std::string &fileName = "/tmp/ejfat_uri"s, 
            TokenType tt=TokenType::admin, bool preferV6=false) noexcept
        {
            if (!fileName.empty())
            {
                std::ifstream file(fileName);
                if (file.is_open())
                {
                    std::string uriLine;
                    if (std::getline(file, uriLine))
                    {
                        file.close();
                        try
                        {
                            return EjfatURI(uriLine, tt, preferV6);
                        }
                        catch (const E2SARException &e)
                        {
                            return E2SARErrorInfo{E2SARErrorc::CaughtException, "Unable to parse URI: "s + static_cast<std::string>(e)};
                        }
                    }
                    file.close();
                    return E2SARErrorInfo{E2SARErrorc::Undefined, "Unable to parse URI."s};
                }
            }
            return E2SARErrorInfo{E2SARErrorc::NotFound, "Unable to find file "s + fileName};
        }
 
        result<std::vector<ip::address>> getDataplaneLocalAddresses(bool v6=false) noexcept;
    };
 
    static inline const result<ip::address> string_to_ip(const std::string &addr) noexcept
    {
        try
        {
            if (addr[0] == '[')
            {
                // strip '[]' from IPv6
                try
                {
                    ip::make_address(addr.substr(1, addr.length() - 2));
                }
                catch (...)
                {
                    ;
                }
                return ip::make_address(addr.substr(1, addr.length() - 2));
            }
            else
                return ip::make_address(addr);
        }
        catch (boost::system::system_error &e)
        {
            return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert IP address from "s + addr};
        }
    }
 
    static inline const result<u_int16_t> string_to_port(const std::string &port_string) noexcept
    {
        try
        {
            u_int16_t port = std::stoi(port_string);
            if (port < 1024 || port > 65535)
            {
                // port is out of range
                return E2SARErrorInfo{E2SARErrorc::OutOfRange, "Port value "s + port_string + " is out of range"s};
            }
            return port;
        }
        catch (const std::exception &e)
        {
            return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + port_string + " to integer"s};
        }
    }
 
    static inline const result<std::pair<ip::address, u_int16_t>> string_tuple_to_ip_and_port(const std::string &t) noexcept
    {
        // search for last ":" (ip:port) or "]" (ipv6 by itself) whichever comes last
        auto const pos = t.find_last_of("]:");
 
        // IPv4 or IPv6 by itself
        if ((pos == std::string::npos) || (t[pos] == ']'))
        {
            auto r1 = string_to_ip(t);
            if (r1)
                return std::pair<ip::address, u_int16_t>(r1.value(), 0);
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port"s};
        }
 
        // port with either IPv4 or IPv6 address x.x.x.x:num or [x:x:x:x::y]:num
        auto r1 = string_to_ip(t.substr(0, pos));
        auto r2 = string_to_port(t.substr(pos + 1));
        if (r1 && r2)
            return std::pair<ip::address, int>(r1.value(), r2.value());
        return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port"s};
    }
 
    static inline const result<std::tuple<ip::address, u_int16_t, u_int16_t>> string_tuple_to_ip_and_port_range(const std::string &t) noexcept
    {
        auto const pos = t.find_last_of("]:");
 
        // IP by itself (no port)
        if ((pos == std::string::npos) || (t[pos] == ']'))
        {
            auto r1 = string_to_ip(t);
            if (r1)
                return std::tuple<ip::address, u_int16_t, u_int16_t>(r1.value(), 0, 0);
            else
                return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port range"s};
        }
 
        auto r1 = string_to_ip(t.substr(0, pos));
        if (!r1)
            return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port range"s};
 
        std::string portStr = t.substr(pos + 1);
        auto dashPos = portStr.find('-');
        if (dashPos != std::string::npos)
        {
            // port range: minPort-maxPort
            auto r2 = string_to_port(portStr.substr(0, dashPos));
            auto r3 = string_to_port(portStr.substr(dashPos + 1));
            if (r2 && r3)
            {
                if (r2.value() > r3.value())
                    return E2SARErrorInfo{E2SARErrorc::ParameterError, "Invalid port range: min > max in "s + t};
                return std::tuple<ip::address, u_int16_t, u_int16_t>(r1.value(), r2.value(), r3.value());
            }
            return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port range"s};
        }
        else
        {
            // single port
            auto r2 = string_to_port(portStr);
            if (r2)
                return std::tuple<ip::address, u_int16_t, u_int16_t>(r1.value(), r2.value(), r2.value());
            return E2SARErrorInfo{E2SARErrorc::ParameterError, "Unable to convert "s + t + " to ip address and port range"s};
        }
    }
 
    static inline result<std::vector<ip::address>> resolveHost(const std::string &host_name) noexcept
    {
 
        std::vector<ip::address> addresses;
        boost::asio::io_context io_context;
 
        try
        {
            ip::udp::resolver resolver(io_context);
            ip::udp::resolver::results_type results = resolver.resolve(host_name, "443");
 
            for(auto i = results.begin(); i != results.end(); ++i)
            {
                ip::udp::endpoint endpoint = *i;
                addresses.push_back(endpoint.address());
            }
            return addresses;
        }
        catch (...)
        {
            // anything happens - we can't find the host
            return E2SARErrorInfo{E2SARErrorc::NotFound, "Unable to convert "s + host_name + " to ip address"s};
        }
    }
 
    // to support unordered maps of pairs
    struct pair_hash {
        std::size_t operator()(const std::pair<u_int64_t, u_int16_t>& p) const {
            u_int64_t hash1 = p.first;
            u_int64_t tmp = p.second;
            u_int64_t hash2 = tmp | tmp << 16 | tmp << 32 | tmp << 48;
            return hash1 ^ hash2;  // Combine the two hashes
        }
    };
 
    struct pair_equal {
        bool operator()(const std::pair<u_int64_t, u_int16_t>& lhs, const std::pair<u_int64_t, u_int16_t>& rhs) const {
            return lhs.first == rhs.first && lhs.second == rhs.second;
        }
    };
 
    static inline float clockEntropyTest(int totalTests = 1000, int sleepMs = 1) 
    {
        std::vector<int_least64_t> points;
        std::vector<int> bins(256, 0);
 
        for (int i = 0; i < totalTests; i++)
        {
            auto now = boost::chrono::system_clock::now();
            auto nowUsec = boost::chrono::duration_cast<boost::chrono::microseconds>(now.time_since_epoch()).count();
            bins[nowUsec & 0xff]++;
            auto until = now + boost::chrono::milliseconds(sleepMs);
            boost::this_thread::sleep_until(until);
        }
 
        // compute the probabilities and entropy
        float entropy{0.0};
        for (size_t i = 0; i < bins.size(); i++)
        {
            float prob = static_cast<float>(bins[i])/(totalTests*1.0);
            entropy += prob * std::log(prob);
        }
 
        // normalize into bits
        entropy *= -1.0/std::log(2);
        return entropy;
    }
 
    template<typename Container>
    std::string concatWithSeparator(const Container& c, const std::string& sep=","s)
    {
        typename Container::const_iterator it = c.begin();
        std::string rets{};
        while(it != c.end())
        {
            rets += *it;
            if (++it != c.end())
                rets += sep;
        }
        return rets;
    }
 
    // busy wait for a given number of microseconds using high_resolution_clock timepoints
    inline void busyWaitUsecs(const boost::chrono::steady_clock::time_point &tp, int64_t usecs)
    {
        while(true)
        {
            // busy wait checking the clock
            if (boost::chrono::duration_cast<boost::chrono::microseconds>(boost::chrono::high_resolution_clock::now() 
                - tp).count() > usecs)
                break;
        }   
    }
 
    using OptimizationsWord = u_int16_t;
    class Optimizations {
        public:
            // go as powers of 2 to make calculations easier
            enum class Code {
                none = 0,
                sendmmsg = 1,
                liburing_send = 2,
                liburing_recv = 3,
                recvmmsg = 4,
                // always last
                unknown = 15
            };
        public:
            inline static OptimizationsWord toWord(Code o)
            {
                return 1 << static_cast<int>(o);
            }
            inline static std::string toString(Code o)
            {
                switch(o)
                {
                    case Code::none: return "none"s; 
                    case Code::sendmmsg: return "sendmmsg";
                    case Code::liburing_recv: return "liburing_recv";
                    case Code::liburing_send: return "liburing_send";
                    case Code::recvmmsg: return "recvmmsg";
                    default: "unknown"s;
                }
                return "unknown"s;
            }
            inline static Code fromString(const std::string& opt)
            {
                if (opt == "none"s)
                    return Code::none;
                else if (opt == "sendmmsg"s)
                    return Code::sendmmsg;
                else if (opt == "liburing_recv"s)
                    return Code::liburing_recv;
                else if (opt == "liburing_send"s)
                    return Code::liburing_send;
                else if (opt == "recvmmsg"s)
                    return Code::recvmmsg;
                return Code::unknown;
            }
            const static std::vector<std::string> availableAsStrings() noexcept;
 
            const static OptimizationsWord availableAsWord() noexcept;
 
            static result<int> select(std::vector<std::string>& opt) noexcept;
 
            static result <int> select(std::vector<Code> &opt) noexcept;
 
            const static std::vector<std::string> selectedAsStrings() noexcept;
 
            const static OptimizationsWord selectedAsWord() noexcept;
 
            const static std::vector<Code> selectedAsList() noexcept;
 
            const static bool isSelected(Code o) noexcept;
        private:
            // all available compiled in optimizations
            static const std::vector<Optimizations::Code> available;
 
            // this is where we store selected optimizations;
            OptimizationsWord selected_optimizations;
 
            // c-tor is private
            Optimizations():selected_optimizations{toWord(Code::none)}
            {}
            // d-tor doesn't exist
            ~Optimizations() = delete;
 
            static Optimizations* instance;
            static inline Optimizations* _get()
            {
                if (not instance)
                    instance = new Optimizations();
                return instance;
            }
    };
 
    std::string expandTilde(const std::string& path);
 
    void defineClogLogger();
};
#endif