MaxScale/server/modules/routing/smartrouter/smartsession.cc

/*
 * Copyright (c) 2019 MariaDB Corporation Ab
 *
 * Use of this software is governed by the Business Source License included
 * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
 *
 * Change Date: 2022-01-01
 *
 * On the date above, in accordance with the Business Source License, use
 * of this software will be governed by version 2 or later of the General
 * Public License.
 */
#include "smartsession.hh"
#include "smartrouter.hh"
#include "performance.hh"

#include <maxscale/modutil.hh>
#include <maxsql/mysql_plus.hh>

// COPY-PASTED error-extraction functions from rwsplit. TODO move to lib.
inline void extract_error_state(uint8_t* pBuffer, uint8_t** ppState, uint16_t* pnState)
{
    mxb_assert(MYSQL_IS_ERROR_PACKET(pBuffer));

    // The payload starts with a one byte command followed by a two byte error code,
    // followed by a 1 byte sql state marker and 5 bytes of sql state. In this context
    // the marker and the state itself are combined.
    *ppState = pBuffer + MYSQL_HEADER_LEN + 1 + 2;
    *pnState = 6;
}

inline void extract_error_message(uint8_t* pBuffer, uint8_t** ppMessage, uint16_t* pnMessage)
{
    mxb_assert(MYSQL_IS_ERROR_PACKET(pBuffer));

    int packet_len = MYSQL_HEADER_LEN + MYSQL_GET_PAYLOAD_LEN(pBuffer);

    // The payload starts with a one byte command followed by a two byte error code,
    // followed by a 1 byte sql state marker and 5 bytes of sql state, followed by
    // a message until the end of the packet.
    *ppMessage = pBuffer + MYSQL_HEADER_LEN + 1 + 2 + 1 + 5;
    *pnMessage = packet_len - MYSQL_HEADER_LEN - 1 - 2 - 1 - 5;
}

std::string extract_error(GWBUF* buffer)
{
    std::string rval;

    if (MYSQL_IS_ERROR_PACKET(((uint8_t*)GWBUF_DATA(buffer))))
    {
        size_t replylen = MYSQL_GET_PAYLOAD_LEN(GWBUF_DATA(buffer)) + MYSQL_HEADER_LEN;
        uint8_t replybuf[replylen];
        gwbuf_copy_data(buffer, 0, sizeof(replybuf), replybuf);

        uint8_t* pState;
        uint16_t nState;
        extract_error_state(replybuf, &pState, &nState);

        uint8_t* pMessage;
        uint16_t nMessage;
        extract_error_message(replybuf, &pMessage, &nMessage);

        std::string err(reinterpret_cast<const char*>(pState), nState);
        std::string msg(reinterpret_cast<const char*>(pMessage), nMessage);

        rval = err + ": " + msg;
    }

    return rval;
}

// TODO Another thing to move to maxutils
static std::array<const char*, 7> size_suffix {"B", "KB", "MB", "GB", "TB", "PB", "EB"};
constexpr size_t KiloByte {1024};

/** return a pair {double_value, suffix} for humanizing a size.
 *  Example: pretty_size_split(2000) => {1.953125, "KB"}
 */
std::pair<double, const char*> pretty_size_split(size_t sz)
{
    double dsize = sz;
    size_t i {0};

    for (; i < size_suffix.size() && dsize >= KiloByte; ++i, dsize /= KiloByte)
    {
    }

    return {dsize, size_suffix[i]};
}

/** Pretty string from a size_t, e.g pretty_size(2000) => "1.95KB"
 */
std::string pretty_size(size_t sz, const char* separator = "")
{
    char buf[64];
    double dsize;
    const char* suffix;

    std::tie(dsize, suffix) = pretty_size_split(sz);

    // format with two decimals
    auto len = std::sprintf(buf, "%.2f", dsize);

    // remove trailing 0-decimals
    char* ptr = buf + len - 1;
    for (; *ptr == '0'; --ptr)
    {
    }
    if (*ptr != '.')
    {
        ++ptr;
    }

    // append suffix
    sprintf(ptr, "%s%s", separator, suffix);

    return buf;
}

SmartRouterSession::SmartRouterSession(SmartRouter* pRouter,
                                       MXS_SESSION* pSession,
                                       Clusters clusters)
    : mxs::RouterSession(pSession)
    , m_router(*pRouter)
    , m_pClient_dcb(pSession->client_dcb)
    , m_clusters(std::move(clusters))
    , m_qc(this, pSession, TYPE_ALL)
{
}

std::vector<maxbase::Host> SmartRouterSession::hosts() const
{
    std::vector<maxbase::Host> ret;
    for (const auto& c : m_clusters)
    {
        ret.push_back(c.host);
    }
    return ret;
}

SmartRouterSession::~SmartRouterSession()
{
}

// static
SmartRouterSession* SmartRouterSession::create(SmartRouter* pRouter, MXS_SESSION* pSession)
{
    Clusters clusters;

    SERVER* pMaster = pRouter->config().master();

    int master_pos = -1;
    int i = 0;

    for (SERVER_REF* ref = pRouter->service()->dbref; ref; ref = ref->next)
    {
        if (!server_ref_is_active(ref) || !ref->server->is_connectable())
        {
            continue;
        }

        mxb_assert(ref->server->is_usable());

        DCB* dcb = dcb_connect(ref->server, pSession, ref->server->protocol().c_str());
        if (dcb)
        {
            bool is_master = (ref->server == pMaster);

            clusters.push_back({ref, dcb, is_master});

            if (is_master)
            {
                master_pos = i;
            }

            ++i;
        }
    }

    SmartRouterSession* pSess = nullptr;

    if (master_pos != -1)
    {
        if (master_pos > 0)
        {   // put the master first. There must be exactly one master cluster.
            std::swap(clusters[0], clusters[master_pos]);
        }

        pSess = new SmartRouterSession(pRouter, pSession, std::move(clusters));
    }
    else
    {
        MXS_ERROR("No master found for %s, smartrouter session cannot be created.",
                  pRouter->config().name().c_str());
    }

    return pSess;
}

int SmartRouterSession::routeQuery(GWBUF* pBuf)
{
    bool ret = false;

    MXS_SDEBUG("routeQuery() buffer size " << pretty_size(gwbuf_length(pBuf)));

    if (expecting_request_packets())
    {
        ret = write_split_packets(pBuf);
        if (all_clusters_are_idle())
        {
            m_mode = Mode::Idle;
        }
    }
    else if (m_mode != Mode::Idle)
    {
        auto is_busy = !all_clusters_are_idle();
        MXS_SERROR("routeQuery() in wrong state. clusters busy = " << std::boolalpha << is_busy);
        mxb_assert(false);
    }
    else
    {
        auto route_info = m_qc.update_route_info(mxs::QueryClassifier::CURRENT_TARGET_UNDEFINED, pBuf);
        std::string canonical = maxscale::get_canonical(pBuf);

        m_measurement = {maxbase::Clock::now(), canonical};

        if (m_qc.target_is_all(route_info.target()))
        {
            MXS_SDEBUG("Write all");
            ret = write_to_all(pBuf, Mode::Query);
        }
        else if (m_qc.target_is_master(route_info.target()) || session_trx_is_active(m_pClient_dcb->session))
        {
            MXS_SDEBUG("Write to master");
            ret = write_to_master(pBuf);
        }
        else
        {
            auto perf = m_router.perf_find(canonical);

            if (perf.is_valid())
            {
                MXS_SDEBUG("Smart route to " << perf.host()
                                             << ", canonical = " << show_some(canonical));
                ret = write_to_host(perf.host(), pBuf);
            }
            else if (modutil_is_SQL(pBuf))
            {
                MXS_SDEBUG("Start measurement");
                ret = write_to_all(pBuf, Mode::MeasureQuery);
            }
            else
            {
                MXS_SWARNING("Could not determine target (non-sql query), goes to master");
                ret = write_to_master(pBuf);
            }
        }
    }

    return ret;
}

void SmartRouterSession::clientReply(GWBUF* pPacket, DCB* pDcb)
{
    mxb_assert(GWBUF_IS_CONTIGUOUS(pPacket));

    auto it = std::find_if(begin(m_clusters), end(m_clusters),
                           [pDcb](const Cluster& cluster) {
                               return cluster.pDcb == pDcb;
                           });

    mxb_assert(it != end(m_clusters));

    Cluster& cluster = *it;

    auto tracker_state_before = cluster.tracker.state();

    cluster.tracker.update_response(pPacket);

    // these flags can all be true at the same time
    bool first_response_packet = (m_mode == Mode::Query || m_mode == Mode::MeasureQuery);
    bool last_packet_for_this_cluster = !cluster.tracker.expecting_response_packets();
    bool very_last_response_packet = !expecting_response_packets();     // last from all clusters

    MXS_SDEBUG("Reply from " << std::boolalpha
                             << cluster.host
                             << " is_master=" << cluster.is_master
                             << " first_packet=" << first_response_packet
                             << " last_packet=" << last_packet_for_this_cluster
                             << " very_last_packet=" << very_last_response_packet
                             << " delayed_response=" << (m_pDelayed_packet != nullptr)
                             << " tracker_state: " << tracker_state_before << " => "
                             << cluster.tracker.state());

    // marker1: If a connection is lost down the pipeline, we first get an ErrorPacket, then a call to
    // handleError(). If we only rely on the handleError() the client receiving the ErrorPacket
    // could retry using this connection/session, causing an error (or assert) in routeQuery().
    // This will change once we implement direct function calls to the Clusters (which really
    // are routers).
    if (cluster.tracker.state() == maxsql::PacketTracker::State::ErrorPacket)
    {
        auto err_code = mxs_mysql_get_mysql_errno(pPacket);
        switch (err_code)
        {
        case ER_CONNECTION_KILLED:      // there might be more error codes needing to be caught here
            MXS_SERROR("clientReply(): Lost connection to " << cluster.host
                                                            << " Error code=" << err_code
                                                            << ' ' << extract_error(pPacket));
            poll_fake_hangup_event(m_pClient_dcb);
            return;
        }
    }

    if (cluster.tracker.state() == maxsql::PacketTracker::State::Error)
    {
        MXS_SERROR("ProtocolTracker from state " << tracker_state_before
                                                 << " to state " << cluster.tracker.state()
                                                 << ". Disconnect.");
        poll_fake_hangup_event(m_pClient_dcb);
        return;
    }

    bool will_reply = false;

    if (first_response_packet)
    {
        maxbase::Duration query_dur = maxbase::Clock::now() - m_measurement.start;
        MXS_SDEBUG("Host " << cluster.host
                           << " will be responding to the client."
                           << " First packet received in time " << query_dur);
        cluster.is_replying_to_client = true;
        will_reply = true;      // tentatively, the packet might have to be delayed

        if (m_mode == Mode::MeasureQuery)
        {
            m_router.perf_update(m_measurement.canonical, {cluster.host, query_dur});
            // If the query is still going on, an error packet is received, else the
            // whole query might play out (and be discarded).
            kill_all_others(cluster);
        }

        m_mode = Mode::CollectResults;
    }

    if (very_last_response_packet)
    {
        will_reply = true;
        m_mode = Mode::Idle;
        mxb_assert(cluster.is_replying_to_client || m_pDelayed_packet);
        if (m_pDelayed_packet)
        {
            MXS_SDEBUG("Picking up delayed packet, discarding response from " << cluster.host);
            gwbuf_free(pPacket);
            pPacket = m_pDelayed_packet;
            m_pDelayed_packet = nullptr;
        }
    }
    else if (cluster.is_replying_to_client)
    {
        if (last_packet_for_this_cluster)
        {
            // Delay sending the last packet until all clusters have responded. The code currently
            // does not allow multiple client-queries at the same time (no query buffer)
            MXS_SDEBUG("Delaying last packet");
            mxb_assert(!m_pDelayed_packet);
            m_pDelayed_packet = pPacket;
            will_reply = false;
        }
        else
        {
            will_reply = true;
        }
    }
    else
    {
        MXS_SDEBUG("Discarding response from " << cluster.host);
        gwbuf_free(pPacket);
    }

    if (will_reply)
    {
        MXS_SDEBUG("Forward response to client");
        MXS_SESSION_ROUTE_REPLY(pDcb->session, pPacket);
    }
}

bool SmartRouterSession::expecting_request_packets() const
{
    return std::any_of(begin(m_clusters), end(m_clusters),
                       [](const Cluster& cluster) {
                           return cluster.tracker.expecting_request_packets();
                       });
}

bool SmartRouterSession::expecting_response_packets() const
{
    return std::any_of(begin(m_clusters), end(m_clusters),
                       [](const Cluster& cluster) {
                           return cluster.tracker.expecting_response_packets();
                       });
}

bool SmartRouterSession::all_clusters_are_idle() const
{
    return std::all_of(begin(m_clusters), end(m_clusters),
                       [](const Cluster& cluster) {
                           return !cluster.tracker.expecting_more_packets();
                       });
}

bool SmartRouterSession::write_to_master(GWBUF* pBuf)
{
    mxb_assert(!m_clusters.empty());
    auto& cluster = m_clusters[0];
    mxb_assert(cluster.is_master);
    cluster.tracker = maxsql::PacketTracker(pBuf);
    cluster.is_replying_to_client = false;

    if (cluster.tracker.expecting_response_packets())
    {
        m_mode = Mode::Query;
    }

    return cluster.pDcb->func.write(cluster.pDcb, pBuf);
}

bool SmartRouterSession::write_to_host(const maxbase::Host& host, GWBUF* pBuf)
{
    auto it = std::find_if(begin(m_clusters), end(m_clusters), [host](const Cluster& cluster) {
                               return cluster.host == host;
                           });
    mxb_assert(it != end(m_clusters));
    auto& cluster = *it;
    cluster.tracker = maxsql::PacketTracker(pBuf);
    if (cluster.tracker.expecting_response_packets())
    {
        m_mode = Mode::Query;
    }

    cluster.is_replying_to_client = false;

    return cluster.pDcb->func.write(cluster.pDcb, pBuf);
}

bool SmartRouterSession::write_to_all(GWBUF* pBuf, Mode mode)
{
    bool success = true;

    for (auto it = begin(m_clusters); it != end(m_clusters); ++it)
    {
        auto& cluster = *it;
        cluster.tracker = maxsql::PacketTracker(pBuf);
        cluster.is_replying_to_client = false;
        auto pBuf_send = (next(it) == end(m_clusters)) ? pBuf : gwbuf_clone(pBuf);
        if (!cluster.pDcb->func.write(cluster.pDcb, pBuf_send))
        {
            success = false;
        }
    }

    if (expecting_response_packets())
    {
        m_mode = mode;
    }

    return success;
}

bool SmartRouterSession::write_split_packets(GWBUF* pBuf)
{
    std::vector<Cluster*> active;

    for (auto it = begin(m_clusters); it != end(m_clusters); ++it)
    {
        if (it->tracker.expecting_request_packets())
        {
            active.push_back(&*it);
        }
    }

    bool success = true;

    for (auto it = begin(active); it != end(active); ++it)
    {
        auto& cluster = **it;

        cluster.tracker.update_request(pBuf);

        auto pBuf_send = (next(it) == end(active)) ? pBuf : gwbuf_clone(pBuf);
        if (!cluster.pDcb->func.write(cluster.pDcb, pBuf_send))
        {
            success = false;
            break;
        }
    }

    return success;
}

void SmartRouterSession::kill_all_others(const Cluster& cluster)
{
    MySQLProtocol* proto = static_cast<MySQLProtocol*>(cluster.pDcb->protocol);
    int keep_protocol_thread_id = proto->thread_id;

    mxs_mysql_execute_kill_all_others(cluster.pDcb->session, cluster.pDcb->session->ses_id,
                                      keep_protocol_thread_id, KT_QUERY);
}

void SmartRouterSession::handleError(GWBUF* pPacket,
                                     DCB* pProblem,
                                     mxs_error_action_t action,
                                     bool* pSuccess)
{
    // One of the clusters closed the connection. In terms of SmartRouter this is a hopeless situation.
    // Close the shop, and let the client retry. Also see marker1.
    auto it = std::find_if(begin(m_clusters), end(m_clusters),
                           [pProblem](const Cluster& cluster) {
                               return cluster.pDcb == pProblem;
                           });

    mxb_assert(it != end(m_clusters));
    Cluster& cluster = *it;

    auto err_code = mxs_mysql_get_mysql_errno(pPacket);
    MXS_SERROR("handleError(): Lost connection to " << cluster.host << " Error code=" << err_code << " "
                                                    << extract_error(pPacket));

    MXS_SESSION* pSession = pProblem->session;
    mxs_session_state_t sesstate = pSession->state;

    /* Send error report to client */
    GWBUF* pCopy = gwbuf_clone(pPacket);
    if (pCopy)
    {
        DCB* pClient = pSession->client_dcb;
        pClient->func.write(pClient, pCopy);
    }

    // This will lead to the rest of the connections to be closed.
    *pSuccess = false;
}

bool SmartRouterSession::lock_to_master()
{
    return false;
}

bool SmartRouterSession::is_locked_to_master() const
{
    return false;
}

bool SmartRouterSession::supports_hint(HINT_TYPE hint_type) const
{
    return false;
}

void SmartRouterSession::close()
{
    for (auto& cluster : m_clusters)
    {
        if (cluster.pDcb)
        {
            dcb_close(const_cast<DCB*>(cluster.pDcb));
        }
    }
}