hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch '2.3' into develop

Browse Source
This commit is contained in:
Markus Mäkelä
2019-02-01 13:55:54 +02:00
parent 968743f61a 08dd55a26a
commit 6038f1f386
31 changed files with 743 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server/core/CMakeLists.txt
View File

@ -81,6 +81,10 @@ if (HAVE_LIBDL)
target_link_libraries(maxscale-common dl)
endif()
# Using initial-exec instead of the default global-dynamic tls-model
# reduces the cost of using thread-local variables in dynamic libraries.
target_compile_options(maxscale-common PUBLIC "-ftls-model=initial-exec")
add_dependencies(maxscale-common pcre2 connector-c libmicrohttpd jansson maxbase)
set_target_properties(maxscale-common PROPERTIES VERSION "1.0.0")
install_module(maxscale-common core)

76
server/core/backend.cc
View File

@ -17,12 +17,15 @@
#include <maxbase/atomic.hh>
#include <maxscale/alloc.h>
#include <maxscale/protocol/mysql.hh>
using namespace maxscale;
Backend::Backend(SERVER_REF* ref)
: m_closed(false)
, m_closed_at(0)
, m_opened_at(0)
, m_backend(ref)
, m_dcb(NULL)
, m_state(0)
@ -49,6 +52,7 @@ void Backend::close(close_type type)
if (!m_closed)
{
m_closed = true;
m_closed_at = time(NULL);
if (in_use())
{
@ -182,6 +186,8 @@ bool Backend::connect(MXS_SESSION* session, SessionCommandList* sescmd)
if ((m_dcb = dcb_connect(m_backend->server, session, m_backend->server->protocol().c_str())))
{
m_closed = false;
m_closed_at = 0;
m_opened_at = time(NULL);
m_state = IN_USE;
mxb::atomic::add(&m_backend->connections, 1, mxb::atomic::RELAXED);
rval = true;
@ -285,3 +291,73 @@ int64_t Backend::num_selects() const
{
return m_num_selects;
}
void Backend::set_close_reason(const std::string& reason)
{
m_close_reason = reason;
}
std::string Backend::get_verbose_status() const
{
std::stringstream ss;
char closed_at[30] = "not closed";
char opened_at[30] = "not opened";
if (m_closed_at)
{
mxb_assert(m_closed);
ctime_r(&m_closed_at, closed_at);
char* nl = strrchr(closed_at, '\n');
mxb_assert(nl);
*nl = '\0';
}
if (m_opened_at)
{
ctime_r(&m_opened_at, opened_at);
char* nl = strrchr(opened_at, '\n');
mxb_assert(nl);
*nl = '\0';
}
ss << "name: [" << name() << "] "
<< "status: [" << m_backend->server->status_string() << "] "
<< "state: [" << to_string((backend_state)m_state) << "] "
<< "last opened at: [" << opened_at << "] "
<< "last closed at: [" << closed_at << "] "
<< "last close reason: [" << m_close_reason << "] "
<< "num sescmd: [" << m_session_commands.size() << "]";
return ss.str();
}
std::string Backend::to_string(backend_state state)
{
std::string rval;
if (state == 0)
{
rval = "NOT_IN_USE";
}
else
{
if (state & IN_USE)
{
rval += "IN_USE";
}
if (state & WAITING_RESULT)
{
rval += rval.empty() ? "" : "|";
rval += "WAITING_RESULT";
}
if (state & FATAL_FAILURE)
{
rval += rval.empty() ? "" : "|";
rval += "FATAL_FAILURE";
}
}
return rval;
}

41
server/core/config.cc
View File

@ -1217,6 +1217,10 @@ bool config_load_global(const char* filename)
"cache. To enable it, add '%s' to the configuration file.",
CN_QUERY_CLASSIFIER_CACHE_SIZE);
}
else if (gateway.qc_cache_properties.max_size == 0)
{
MXS_NOTICE("Query classifier cache is disabled");
}
else
{
MXS_NOTICE("Using up to %s of memory for query classifier cache",
@ -1275,6 +1279,34 @@ const MXS_MODULE* get_module(CONFIG_CONTEXT* obj, const char* param_name, const
return module ? get_module(module, module_type) : NULL;
}
const char* get_missing_module_parameter_name(const CONFIG_CONTEXT* obj)
{
std::string type = config_get_string(obj->parameters, CN_TYPE);
if (type == CN_SERVICE && !config_get_param(obj->parameters, CN_ROUTER))
{
return CN_ROUTER;
}
else if (type == CN_LISTENER && !config_get_param(obj->parameters, CN_PROTOCOL))
{
return CN_PROTOCOL;
}
else if (type == CN_SERVER && !config_get_param(obj->parameters, CN_PROTOCOL))
{
return CN_PROTOCOL;
}
else if (type == CN_MONITOR && !config_get_param(obj->parameters, CN_MODULE))
{
return CN_MODULE;
}
else if (type == CN_FILTER && !config_get_param(obj->parameters, CN_MODULE))
{
return CN_MODULE;
}
return nullptr;
}
std::pair<const MXS_MODULE_PARAM*, const MXS_MODULE*> get_module_details(const CONFIG_CONTEXT* obj)
{
std::string type = config_get_string(obj->parameters, CN_TYPE);
@ -3038,6 +3070,15 @@ static bool check_config_objects(CONFIG_CONTEXT* context)
continue;
}
const char* no_module_defined = get_missing_module_parameter_name(obj);
if (no_module_defined)
{
MXS_ERROR("'%s' is missing the required parameter '%s'", obj->object, no_module_defined);
rval = false;
continue;
}
const MXS_MODULE_PARAM* param_set = nullptr;
const MXS_MODULE* mod = nullptr;
std::tie(param_set, mod) = get_module_details(obj);

136
server/core/queryclassifier.cc
View File

@ -433,7 +433,73 @@ bool QueryClassifier::query_type_is_read_only(uint32_t qtype) const
return rval;
}
uint32_t QueryClassifier::get_route_target(uint8_t command, uint32_t qtype, HINT* pHints)
void QueryClassifier::process_routing_hints(HINT* pHints, uint32_t* target)
{
HINT* pHint = pHints;
while (pHint)
{
if (m_pHandler->supports_hint(pHint->type))
{
switch (pHint->type)
{
case HINT_ROUTE_TO_MASTER:
// This means override, so we bail out immediately.
*target = TARGET_MASTER;
MXS_DEBUG("Hint: route to master");
pHint = NULL;
break;
case HINT_ROUTE_TO_NAMED_SERVER:
// The router is expected to look up the named server.
*target |= TARGET_NAMED_SERVER;
MXS_DEBUG("Hint: route to named server: %s", (char*)pHint->data);
break;
case HINT_ROUTE_TO_UPTODATE_SERVER:
// TODO: Add generic target type, never to be seem by RWS.
mxb_assert(false);
break;
case HINT_ROUTE_TO_ALL:
// TODO: Add generic target type, never to be seem by RWS.
mxb_assert(false);
break;
case HINT_ROUTE_TO_LAST_USED:
MXS_DEBUG("Hint: route to last used");
*target = TARGET_LAST_USED;
break;
case HINT_PARAMETER:
if (strncasecmp((char*)pHint->data,
"max_slave_replication_lag",
strlen("max_slave_replication_lag")) == 0)
{
*target |= TARGET_RLAG_MAX;
}
else
{
MXS_ERROR("Unknown hint parameter '%s' when "
"'max_slave_replication_lag' was expected.",
(char*)pHint->data);
}
break;
case HINT_ROUTE_TO_SLAVE:
*target = TARGET_SLAVE;
MXS_DEBUG("Hint: route to slave.");
}
}
if (pHint)
{
pHint = pHint->next;
}
}
}
uint32_t QueryClassifier::get_route_target(uint8_t command, uint32_t qtype)
{
bool trx_active = session_trx_is_active(m_pSession);
uint32_t target = TARGET_UNDEFINED;
@ -533,70 +599,6 @@ uint32_t QueryClassifier::get_route_target(uint8_t command, uint32_t qtype, HINT
target = TARGET_MASTER;
}
/** Process routing hints */
HINT* pHint = pHints;
while (pHint)
{
if (m_pHandler->supports_hint(pHint->type))
{
switch (pHint->type)
{
case HINT_ROUTE_TO_MASTER:
// This means override, so we bail out immediately.
target = TARGET_MASTER;
MXS_DEBUG("Hint: route to master");
pHint = NULL;
break;
case HINT_ROUTE_TO_NAMED_SERVER:
// The router is expected to look up the named server.
target |= TARGET_NAMED_SERVER;
MXS_DEBUG("Hint: route to named server: %s", (char*)pHint->data);
break;
case HINT_ROUTE_TO_UPTODATE_SERVER:
// TODO: Add generic target type, never to be seem by RWS.
mxb_assert(false);
break;
case HINT_ROUTE_TO_ALL:
// TODO: Add generic target type, never to be seem by RWS.
mxb_assert(false);
break;
case HINT_ROUTE_TO_LAST_USED:
MXS_DEBUG("Hint: route to last used");
target = TARGET_LAST_USED;
break;
case HINT_PARAMETER:
if (strncasecmp((char*)pHint->data,
"max_slave_replication_lag",
strlen("max_slave_replication_lag")) == 0)
{
target |= TARGET_RLAG_MAX;
}
else
{
MXS_ERROR("Unknown hint parameter '%s' when "
"'max_slave_replication_lag' was expected.",
(char*)pHint->data);
}
break;
case HINT_ROUTE_TO_SLAVE:
target = TARGET_SLAVE;
MXS_DEBUG("Hint: route to slave.");
}
}
if (pHint)
{
pHint = pHint->next;
}
}
return target;
}
@ -999,9 +1001,11 @@ QueryClassifier::RouteInfo QueryClassifier::update_route_info(
type_mask = ps_get_type(stmt_id);
}
route_target = get_route_target(command, type_mask, pBuffer->hint);
route_target = get_route_target(command, type_mask);
}
process_routing_hints(pBuffer->hint, &route_target);
if (session_trx_is_ending(m_pSession)
|| qc_query_is_type(type_mask, QUERY_TYPE_BEGIN_TRX))
{

4
server/modules/routing/readwritesplit/readwritesplit.cc
View File

@ -272,6 +272,9 @@ void RWSplit::diagnostics(DCB* dcb)
dcb_printf(dcb,
"\tstrict_sp_calls: %s\n",
cnf.strict_sp_calls ? "true" : "false");
dcb_printf(dcb,
"\tprune_sescmd_history: %s\n",
cnf.prune_sescmd_history ? "true" : "false");
dcb_printf(dcb,
"\tdisable_sescmd_history: %s\n",
cnf.disable_sescmd_history ? "true" : "false");
@ -503,6 +506,7 @@ extern "C" MXS_MODULE* MXS_CREATE_MODULE()
{"max_slave_replication_lag", MXS_MODULE_PARAM_INT, "-1" },
{"max_slave_connections", MXS_MODULE_PARAM_STRING, MAX_SLAVE_COUNT},
{"retry_failed_reads", MXS_MODULE_PARAM_BOOL, "true" },
{"prune_sescmd_history", MXS_MODULE_PARAM_BOOL, "false" },
{"disable_sescmd_history", MXS_MODULE_PARAM_BOOL, "false" },
{"max_sescmd_history", MXS_MODULE_PARAM_COUNT, "50" },
{"strict_multi_stmt", MXS_MODULE_PARAM_BOOL, "false" },

2
server/modules/routing/readwritesplit/readwritesplit.hh
View File

@ -141,6 +141,7 @@ struct Config
, master_failure_mode(
(enum failure_mode)params->get_enum("master_failure_mode", master_failure_mode_values))
, max_sescmd_history(params->get_integer("max_sescmd_history"))
, prune_sescmd_history(config_get_bool(params, "prune_sescmd_history"))
, disable_sescmd_history(config_get_bool(params, "disable_sescmd_history"))
, master_accept_reads(config_get_bool(params, "master_accept_reads"))
, strict_multi_stmt(config_get_bool(params, "strict_multi_stmt"))
@ -195,6 +196,7 @@ struct Config
* master or all nodes */
failure_mode master_failure_mode; /**< Master server failure handling mode */
uint64_t max_sescmd_history; /**< Maximum amount of session commands to store */
bool prune_sescmd_history; /**< Prune session command history */
bool disable_sescmd_history;/**< Disable session command history */
bool master_accept_reads; /**< Use master for reads */
bool strict_multi_stmt; /**< Force non-multistatement queries to be routed to

74
server/modules/routing/readwritesplit/rwsplit_route_stmt.cc
View File

@ -341,8 +341,7 @@ bool RWSplitSession::route_single_stmt(GWBUF* querybuf)
else
{
MXS_ERROR("Could not find valid server for target type %s, closing "
"connection.",
STRTARGET(route_target));
"connection.", route_target_to_string(route_target));
}
}
@ -391,6 +390,23 @@ void RWSplitSession::continue_large_session_write(GWBUF* querybuf, uint32_t type
}
}
void RWSplitSession::prune_to_position(uint64_t pos)
{
/** Prune all completed responses before a certain position */
ResponseMap::iterator it = m_sescmd_responses.lower_bound(pos);
if (it != m_sescmd_responses.end())
{
// Found newer responses that were returned after this position
m_sescmd_responses.erase(m_sescmd_responses.begin(), it);
}
else
{
// All responses are older than the requested position
m_sescmd_responses.clear();
}
}
/**
* Execute in backends used by current router session.
* Save session variable commands to router session property
@ -498,7 +514,7 @@ bool RWSplitSession::route_session_write(GWBUF* querybuf, uint8_t command, uint3
}
}
if (m_config.max_sescmd_history > 0 && m_sescmd_list.size() >= m_config.max_sescmd_history)
if (m_config.max_sescmd_history > 0 && m_sescmd_list.size() > m_config.max_sescmd_history)
{
static bool warn_history_exceeded = true;
if (warn_history_exceeded)
@ -520,20 +536,17 @@ bool RWSplitSession::route_session_write(GWBUF* querybuf, uint8_t command, uint3
m_sescmd_list.clear();
}
if (m_config.prune_sescmd_history && !m_sescmd_list.empty()
&& m_sescmd_list.size() + 1 > m_config.max_sescmd_history)
{
// Close to the history limit, remove the oldest command
prune_to_position(m_sescmd_list.front()->get_position());
m_sescmd_list.pop_front();
}
if (m_config.disable_sescmd_history)
{
/** Prune stored responses */
ResponseMap::iterator it = m_sescmd_responses.lower_bound(lowest_pos);
if (it != m_sescmd_responses.end())
{
m_sescmd_responses.erase(m_sescmd_responses.begin(), it);
}
else
{
// All responses processed
m_sescmd_responses.clear();
}
prune_to_position(lowest_pos);
}
else
{
@ -554,8 +567,16 @@ bool RWSplitSession::route_session_write(GWBUF* querybuf, uint8_t command, uint3
}
else
{
MXS_ERROR("Could not route session command: %s", attempted_write ? "Write to all backends failed" :
"All connections have failed");
std::string status;
for (const auto& a : m_backends)
{
status += "\n";
status += a->get_verbose_status();
}
MXS_ERROR("Could not route session command: %s. Connection information: %s",
attempted_write ? "Write to all backends failed" : "All connections have failed",
status.c_str());
}
return nsucc;
@ -695,7 +716,6 @@ RWBackend* RWSplitSession::get_target_backend(backend_type_t btype,
if (name) /*< Choose backend by name from a hint */
{
mxb_assert(btype != BE_MASTER);
btype = BE_SLAVE;
rval = get_hinted_backend(name);
}
@ -800,10 +820,19 @@ RWBackend* RWSplitSession::handle_hinted_target(GWBUF* querybuf, route_target_t
{
if (TARGET_IS_NAMED_SERVER(route_target))
{
MXS_INFO("Was supposed to route to named server "
"%s but couldn't find the server in a "
"suitable state.",
named_server);
std::string status = "Could not find server";
for (const auto& a : m_backends)
{
if (strcmp(a->server()->name(), named_server) == 0)
{
status = a->server()->status_string();
break;
}
}
MXS_INFO("Was supposed to route to named server %s but couldn't find the server in a "
"suitable state. Server state: %s", named_server, status.c_str());
}
else if (TARGET_IS_RLAG_MAX(route_target))
{
@ -1013,6 +1042,7 @@ bool RWSplitSession::handle_master_is_target(RWBackend** dest)
if (m_current_master && m_current_master->in_use())
{
m_current_master->close();
m_current_master->set_close_reason("The original master is not available");
}
}
else if (!m_config.delayed_retry

28
server/modules/routing/readwritesplit/rwsplit_session_cmd.cc
View File

@ -78,6 +78,7 @@ static void discard_if_response_differs(RWBackend* backend,
STRPACKETTYPE(cmd),
query.empty() ? "<no query>" : query.c_str());
backend->close(mxs::Backend::CLOSE_FATAL);
backend->set_close_reason("Invalid response to: " + query);
}
}
@ -160,5 +161,32 @@ void RWSplitSession::process_sescmd_response(RWBackend* backend, GWBUF** ppPacke
gwbuf_free(*ppPacket);
*ppPacket = NULL;
}
if (m_expected_responses == 0
&& (command == MXS_COM_CHANGE_USER || command == MXS_COM_RESET_CONNECTION))
{
mxb_assert_message(m_slave_responses.empty(), "All responses should've been processed");
// This is the last session command to finish that resets the session state, reset the history
MXS_INFO("Resetting session command history (length: %lu)", m_sescmd_list.size());
/**
* Since new connections need to perform the COM_CHANGE_USER, pop it off the list along
* with the expected response to it.
*/
SSessionCommand latest = m_sescmd_list.back();
cmd = m_sescmd_responses[latest->get_position()];
m_sescmd_list.clear();
m_sescmd_responses.clear();
// Push the response back as the first executed session command
m_sescmd_list.push_back(latest);
m_sescmd_responses[latest->get_position()] = cmd;
// Adjust counters to match the number of stored session commands
m_recv_sescmd = 1;
m_sent_sescmd = 1;
m_sescmd_count = 2;
}
}
}

4
server/modules/routing/readwritesplit/rwsplitsession.cc
View File

@ -949,6 +949,7 @@ void RWSplitSession::handleError(GWBUF* errmsgbuf,
}
backend->close();
backend->set_close_reason("Master connection failed: " + extract_error(errmsgbuf));
}
else
{
@ -962,6 +963,7 @@ void RWSplitSession::handleError(GWBUF* errmsgbuf,
// Try to replay the transaction on another node
can_continue = start_trx_replay();
backend->close();
backend->set_close_reason("Read-only trx failed: " + extract_error(errmsgbuf));
if (!can_continue)
{
@ -982,6 +984,7 @@ void RWSplitSession::handleError(GWBUF* errmsgbuf,
m_otrx_state = OTRX_INACTIVE;
can_continue = start_trx_replay();
backend->close();
backend->set_close_reason("Optimistic trx failed: " + extract_error(errmsgbuf));
}
else
{
@ -1071,6 +1074,7 @@ bool RWSplitSession::handle_error_new_connection(DCB* backend_dcb, GWBUF* errmsg
* is closed, it's possible that the routing logic will pick the failed
* server as the target. */
backend->close();
backend->set_close_reason("Slave connection failed: " + extract_error(errmsg));
if (route_stored)
{

54
server/modules/routing/readwritesplit/rwsplitsession.hh
View File

@ -139,6 +139,7 @@ private:
void process_sescmd_response(mxs::RWBackend* backend, GWBUF** ppPacket);
void compress_history(mxs::SSessionCommand& sescmd);
void prune_to_position(uint64_t pos);
bool route_session_write(GWBUF* querybuf, uint8_t command, uint32_t type);
void continue_large_session_write(GWBUF* querybuf, uint32_t type);
bool route_single_stmt(GWBUF* querybuf);
@ -277,7 +278,7 @@ private:
int m_last_keepalive_check; /**< When the last ping was done */
int m_nbackends; /**< Number of backend servers (obsolete) */
DCB* m_client; /**< The client DCB */
uint64_t m_sescmd_count; /**< Number of executed session commands */
uint64_t m_sescmd_count; /**< Number of executed session commands (starts from 1) */
int m_expected_responses; /**< Number of expected responses to the current
* query */
GWBUF* m_query_queue; /**< Queued commands waiting to be executed */
@ -322,22 +323,35 @@ private:
*/
uint32_t get_internal_ps_id(RWSplitSession* rses, GWBUF* buffer);
#define STRTARGET(t) \
(t == TARGET_ALL ? "TARGET_ALL" \
: (t == TARGET_MASTER ? "TARGET_MASTER" \
: (t == TARGET_SLAVE ? "TARGET_SLAVE" \
: (t \
== TARGET_NAMED_SERVER \
? "TARGET_NAMED_SERVER" \
: (t \
== \
TARGET_RLAG_MAX \
? "TARGET_RLAG_MAX" \
: ( \
t \
== \
TARGET_UNDEFINED \
? \
"TARGET_UNDEFINED" \
: \
"Unknown target value"))))))
static inline const char* route_target_to_string(route_target_t target)
{
if (TARGET_IS_MASTER(target))
{
return "TARGET_MASTER";
}
else if (TARGET_IS_SLAVE(target))
{
return "TARGET_SLAVE";
}
else if (TARGET_IS_NAMED_SERVER(target))
{
return "TARGET_NAMED_SERVER";
}
else if (TARGET_IS_ALL(target))
{
return "TARGET_ALL";
}
else if (TARGET_IS_RLAG_MAX(target))
{
return "TARGET_RLAG_MAX";
}
else if (TARGET_IS_LAST_USED(target))
{
return "TARGET_LAST_USED";
}
else
{
mxb_assert(!true);
return "Unknown target value";
}
}