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-04-12 13:23:49 +03:00
parent 382d485c2e 4131f09c16
commit c643f9bc8d
31 changed files with 625 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server/core/config.cc
View File

@ -4659,6 +4659,10 @@ json_t* config_maxscale_to_json(const char* host)
CN_QUERY_CLASSIFIER_CACHE_SIZE,
json_integer(cnf->qc_cache_properties.max_size));
json_object_set_new(param, CN_RETAIN_LAST_STATEMENTS, json_integer(session_get_retain_last_statements()));
json_object_set_new(param, CN_DUMP_LAST_STATEMENTS, json_string(session_get_dump_statements_str()));
json_object_set_new(param, CN_LOAD_PERSISTED_CONFIGS, json_boolean(cnf->load_persisted_configs));
json_t* attr = json_object();
time_t started = maxscale_started();
time_t activated = started + MXS_CLOCK_TO_SEC(cnf->promoted_at);

4
server/core/dcb.cc
View File

@ -99,7 +99,6 @@ static inline DCB* dcb_find_in_list(DCB* dcb);
static void dcb_stop_polling_and_shutdown(DCB* dcb);
static bool dcb_maybe_add_persistent(DCB*);
static inline bool dcb_write_parameter_check(DCB* dcb, GWBUF* queue);
static int dcb_bytes_readable(DCB* dcb);
static int dcb_read_no_bytes_available(DCB* dcb, int nreadtotal);
static int dcb_create_SSL(DCB* dcb, SSL_LISTENER* ssl);
static int dcb_read_SSL(DCB* dcb, GWBUF** head);
@ -586,9 +585,10 @@ int dcb_read(DCB* dcb,
* Find the number of bytes available for the DCB's socket
*
* @param dcb The DCB to read from
*
* @return -1 on error, otherwise the total number of bytes available
*/
static int dcb_bytes_readable(DCB* dcb)
int dcb_bytes_readable(DCB* dcb)
{
int bytesavailable;

17
server/core/modutil.cc
View File

@ -1458,9 +1458,16 @@ std::string get_canonical(GWBUF* querybuf)
break;
}
}
else if (is_space(*it) && (i == 0 || is_space(rval[i - 1])))
else if (is_space(*it))
{
// Repeating space, skip it
if (i == 0 || is_space(rval[i - 1]))
{
// Leading or repeating whitespace, skip it
}
else
{
rval[i++] = ' ';
}
}
else if (*it == '/' && is_next(it, buf.end(), "/*"))
{
@ -1564,6 +1571,12 @@ std::string get_canonical(GWBUF* querybuf)
mxb_assert(it != buf.end());
}
// Remove trailing whitespace
while (i > 0 && is_space(rval[i - 1]))
{
--i;
}
// Shrink the buffer so that the internal bookkeeping of std::string remains up to date
rval.resize(i);

26
server/core/session.cc
View File

@ -938,6 +938,11 @@ void session_set_retain_last_statements(uint32_t n)
this_unit.retain_last_statements = n;
}
uint32_t session_get_retain_last_statements()
{
return this_unit.retain_last_statements;
}
void session_set_dump_statements(session_dump_statements_t value)
{
this_unit.dump_statements = value;
@ -948,6 +953,25 @@ session_dump_statements_t session_get_dump_statements()
return this_unit.dump_statements;
}
const char* session_get_dump_statements_str()
{
switch (this_unit.dump_statements)
{
case SESSION_DUMP_STATEMENTS_NEVER:
return "never";
case SESSION_DUMP_STATEMENTS_ON_CLOSE:
return "on_close";
case SESSION_DUMP_STATEMENTS_ON_ERROR:
return "on_error";
default:
mxb_assert(!true);
return "unknown";
}
}
void session_retain_statement(MXS_SESSION* pSession, GWBUF* pBuffer)
{
static_cast<Session*>(pSession)->retain_statement(pBuffer);
@ -1380,7 +1404,7 @@ void Session::retain_statement(GWBUF* pBuffer)
{
mxb_assert(m_last_queries.size() <= m_retain_last_statements);
std::shared_ptr<GWBUF> sBuffer(gwbuf_clone(pBuffer));
std::shared_ptr<GWBUF> sBuffer(gwbuf_clone(pBuffer), std::default_delete<GWBUF>());
m_last_queries.push_front(QueryInfo(sBuffer));

12
server/core/test/rest-api/test/server.js
View File

@ -156,6 +156,18 @@ describe("Server State", function() {
})
});
it("force server into maintenance", function() {
return request.put(base_url + "/servers/" + server.data.id + "/set?state=maintenance&force=yes")
.then(function(resp) {
return request.get(base_url + "/servers/" + server.data.id)
})
.then(function(resp) {
var srv = JSON.parse(resp)
srv.data.attributes.state.should.match(/Maintenance/)
srv.data.attributes.statistics.connections.should.be.equal(0)
})
});
it("clear maintenance", function() {
return request.put(base_url + "/servers/" + server.data.id + "/clear?state=maintenance")
.then(function(resp) {

12
server/modules/protocol/MySQL/mariadbclient/mysql_client.cc
View File

@ -485,7 +485,10 @@ int gw_read_client_event(DCB* dcb)
{
max_bytes = 36;
}
return_code = dcb_read(dcb, &read_buffer, max_bytes);
const uint32_t max_single_read = GW_MYSQL_MAX_PACKET_LEN + MYSQL_HEADER_LEN;
return_code = dcb_read(dcb, &read_buffer, max_bytes > 0 ? max_bytes : max_single_read);
if (return_code < 0)
{
dcb_close(dcb);
@ -495,6 +498,13 @@ int gw_read_client_event(DCB* dcb)
return return_code;
}
if (nbytes_read == max_single_read && dcb_bytes_readable(dcb) > 0)
{
// We read a maximally long packet, route it first. This is done in case there's a lot more data
// waiting and we have to start throttling the reads.
poll_fake_read_event(dcb);
}
return_code = 0;
switch (protocol->protocol_auth_state)

8
server/modules/protocol/MySQL/mysql_common.cc
View File

@ -1078,6 +1078,14 @@ int gw_decode_mysql_server_handshake(MySQLProtocol* conn, uint8_t* payload)
// get ThreadID: 4 bytes
uint32_t tid = gw_mysql_get_byte4(payload);
// LocalClient also uses this code and it doesn't populate the server pointer
// TODO: fix it
if (conn->owner_dcb && conn->owner_dcb->server)
{
MXS_INFO("Connected to '%s' with thread id %u", conn->owner_dcb->server->name(), tid);
}
/* TODO: Correct value of thread id could be queried later from backend if
* there is any worry it might be larger than 32bit allows. */
conn->thread_id = tid;

310
server/modules/protocol/MySQL/rwbackend.cc
View File

@ -128,9 +128,23 @@ void RWBackend::close(close_type type)
bool RWBackend::consume_fetched_rows(GWBUF* buffer)
{
m_expected_rows -= modutil_count_packets(buffer);
mxb_assert(m_expected_rows >= 0);
return m_expected_rows == 0;
bool rval = false;
bool more = false;
int n_eof = modutil_count_signal_packets(buffer, 0, &more, &m_modutil_state);
// If the server responded with an error, n_eof > 0
if (n_eof > 0)
{
rval = true;
}
else
{
m_expected_rows -= modutil_count_packets(buffer);
mxb_assert(m_expected_rows >= 0);
rval = m_expected_rows == 0;
}
return rval;
}
static inline bool have_next_packet(GWBUF* buffer)
@ -139,6 +153,197 @@ static inline bool have_next_packet(GWBUF* buffer)
return gwbuf_length(buffer) > len;
}
template<class Iter>
uint64_t get_encoded_int(Iter it)
{
uint64_t len = *it++;
switch (len)
{
case 0xfc:
len = *it++;
len |= ((uint64_t)*it++) << 8;
break;
case 0xfd:
len = *it++;
len |= ((uint64_t)*it++) << 8;
len |= ((uint64_t)*it++) << 16;
break;
case 0xfe:
len = *it++;
len |= ((uint64_t)*it++) << 8;
len |= ((uint64_t)*it++) << 16;
len |= ((uint64_t)*it++) << 24;
len |= ((uint64_t)*it++) << 32;
len |= ((uint64_t)*it++) << 40;
len |= ((uint64_t)*it++) << 48;
len |= ((uint64_t)*it++) << 56;
break;
default:
break;
}
return len;
}
template<class Iter>
Iter skip_encoded_int(Iter it)
{
switch (*it)
{
case 0xfc:
return std::next(it, 3);
case 0xfd:
return std::next(it, 4);
case 0xfe:
return std::next(it, 9);
default:
return std::next(it);
}
}
template<class Iter>
uint64_t is_last_ok(Iter it)
{
++it; // Skip the command byte
it = skip_encoded_int(it); // Affected rows
it = skip_encoded_int(it); // Last insert ID
uint16_t status = *it++;
status |= (*it++) << 8;
return (status & SERVER_MORE_RESULTS_EXIST) == 0;
}
template<class Iter>
uint64_t is_last_eof(Iter it)
{
std::advance(it, 3); // Skip the command byte and warning count
uint16_t status = *it++;
status |= (*it++) << 8;
return (status & SERVER_MORE_RESULTS_EXIST) == 0;
}
void RWBackend::process_reply_start(mxs::Buffer::iterator it)
{
uint8_t cmd = *it;
m_local_infile_requested = false;
switch (cmd)
{
case MYSQL_REPLY_OK:
if (is_last_ok(it))
{
// No more results
set_reply_state(REPLY_STATE_DONE);
}
break;
case MYSQL_REPLY_LOCAL_INFILE:
m_local_infile_requested = true;
set_reply_state(REPLY_STATE_DONE);
break;
case MYSQL_REPLY_ERR:
// Nothing ever follows an error packet
set_reply_state(REPLY_STATE_DONE);
break;
case MYSQL_REPLY_EOF:
// EOF packets are never expected as the first response
mxb_assert(!true);
break;
default:
if (current_command() == MXS_COM_FIELD_LIST)
{
// COM_FIELD_LIST sends a strange kind of a result set
set_reply_state(REPLY_STATE_RSET_ROWS);
}
else
{
// Start of a result set
m_num_coldefs = get_encoded_int(it);
set_reply_state(REPLY_STATE_RSET_COLDEF);
}
break;
}
}
void RWBackend::process_packets(GWBUF* result)
{
mxs::Buffer buffer(result);
auto it = buffer.begin();
while (it != buffer.end())
{
// Extract packet length and command byte
uint32_t len = *it++;
len |= (*it++) << 8;
len |= (*it++) << 16;
++it; // Skip the sequence
mxb_assert(it != buffer.end());
auto end = std::next(it, len);
uint8_t cmd = *it;
switch (m_reply_state)
{
case REPLY_STATE_START:
process_reply_start(it);
break;
case REPLY_STATE_DONE:
// This should never happen
mxb_assert(!true);
MXS_ERROR("Unexpected result state. cmd: 0x%02hhx, len: %u", cmd, len);
break;
case REPLY_STATE_RSET_COLDEF:
mxb_assert(m_num_coldefs > 0);
--m_num_coldefs;
if (m_num_coldefs == 0)
{
set_reply_state(REPLY_STATE_RSET_COLDEF_EOF);
// Skip this state when DEPRECATE_EOF capability is supported
}
break;
case REPLY_STATE_RSET_COLDEF_EOF:
mxb_assert(cmd == MYSQL_REPLY_EOF && len == MYSQL_EOF_PACKET_LEN - MYSQL_HEADER_LEN);
set_reply_state(REPLY_STATE_RSET_ROWS);
if (is_opening_cursor())
{
set_cursor_opened();
MXS_INFO("Cursor successfully opened");
set_reply_state(REPLY_STATE_DONE);
}
break;
case REPLY_STATE_RSET_ROWS:
if (cmd == MYSQL_REPLY_EOF && len == MYSQL_EOF_PACKET_LEN - MYSQL_HEADER_LEN)
{
set_reply_state(is_last_eof(it) ? REPLY_STATE_DONE : REPLY_STATE_START);
}
else if (cmd == MYSQL_REPLY_ERR)
{
set_reply_state(REPLY_STATE_DONE);
}
break;
}
it = end;
}
buffer.release();
}
/**
* @brief Process a possibly partial response from the backend
*
@ -148,109 +353,22 @@ void RWBackend::process_reply(GWBUF* buffer)
{
if (current_command() == MXS_COM_STMT_FETCH)
{
bool more = false;
int n_eof = modutil_count_signal_packets(buffer, 0, &more, &m_modutil_state);
// If the server responded with an error, n_eof > 0
if (n_eof > 0 || consume_fetched_rows(buffer))
if (consume_fetched_rows(buffer))
{
set_reply_state(REPLY_STATE_DONE);
}
}
else if (current_command() == MXS_COM_STATISTICS)
else if (current_command() == MXS_COM_STATISTICS || GWBUF_IS_COLLECTED_RESULT(buffer))
{
// COM_STATISTICS returns a single string and thus requires special handling
// COM_STATISTICS returns a single string and thus requires special handling.
// Collected result are all in one buffer and need no processing.
set_reply_state(REPLY_STATE_DONE);
}
else if (get_reply_state() == REPLY_STATE_START
&& (!mxs_mysql_is_result_set(buffer) || GWBUF_IS_COLLECTED_RESULT(buffer)))
{
m_local_infile_requested = false;
if (GWBUF_IS_COLLECTED_RESULT(buffer)
|| current_command() == MXS_COM_STMT_PREPARE
|| !mxs_mysql_is_ok_packet(buffer)
|| !mxs_mysql_more_results_after_ok(buffer))
{
/** Not a result set, we have the complete response */
set_reply_state(REPLY_STATE_DONE);
if (mxs_mysql_is_local_infile(buffer))
{
m_local_infile_requested = true;
}
}
else
{
// This is an OK packet and more results will follow
mxb_assert(mxs_mysql_is_ok_packet(buffer)
&& mxs_mysql_more_results_after_ok(buffer));
if (have_next_packet(buffer))
{
// TODO: Don't clone the buffer
GWBUF* tmp = gwbuf_clone(buffer);
tmp = gwbuf_consume(tmp, mxs_mysql_get_packet_len(tmp));
// Consume repeating OK packets
while (mxs_mysql_more_results_after_ok(buffer) && have_next_packet(tmp))
{
tmp = gwbuf_consume(tmp, mxs_mysql_get_packet_len(tmp));
mxb_assert(tmp);
}
process_reply(tmp);
gwbuf_free(tmp);
return;
}
}
}
else
{
bool more = false;
int n_old_eof = get_reply_state() == REPLY_STATE_RSET_ROWS ? 1 : 0;
int n_eof = modutil_count_signal_packets(buffer, n_old_eof, &more, &m_modutil_state);
if (n_eof > 2)
{
/**
* We have multiple results in the buffer, we only care about
* the state of the last one. Skip the complete result sets and act
* like we're processing a single result set.
*/
n_eof = n_eof % 2 ? 1 : 2;
}
if (n_eof == 0)
{
/** Waiting for the EOF packet after the column definitions */
set_reply_state(REPLY_STATE_RSET_COLDEF);
}
else if (n_eof == 1 && current_command() != MXS_COM_FIELD_LIST)
{
/** Waiting for the EOF packet after the rows */
set_reply_state(REPLY_STATE_RSET_ROWS);
if (is_opening_cursor())
{
set_cursor_opened();
MXS_INFO("Cursor successfully opened");
set_reply_state(REPLY_STATE_DONE);
}
}
else
{
/** We either have a complete result set or a response to
* a COM_FIELD_LIST command */
mxb_assert(n_eof == 2 || (n_eof == 1 && current_command() == MXS_COM_FIELD_LIST));
set_reply_state(REPLY_STATE_DONE);
if (more)
{
/** The server will send more resultsets */
set_reply_state(REPLY_STATE_START);
}
}
// Normal result, process it one packet at a time
process_packets(buffer);
}
if (get_reply_state() == REPLY_STATE_DONE)

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

@ -190,23 +190,9 @@ bool RWSplitSession::route_stored_query()
while (!m_query_queue.empty())
{
MXS_INFO(">>> Routing stored queries");
auto query = std::move(m_query_queue.front());
m_query_queue.pop_front();
mxb_assert(!query.empty());
mxb_assert_message(modutil_count_packets(query.get()) == 1, "Buffer must contain only one packet");
if (query.empty())
{
MXS_ALERT("Queued query unexpectedly empty, dumping query queue contents");
for (auto& a : m_query_queue)
{
gwbuf_hexdump(a, LOG_ALERT);
}
return true;
}
/** Store the query queue locally for the duration of the routeQuery call.
* This prevents recursive calls into this function. */
decltype(m_query_queue) temp_storage;
@ -232,7 +218,12 @@ bool RWSplitSession::route_stored_query()
}
else
{
/** Routing was stopped, we need to wait for a response before retrying */
/**
* Routing was stopped, we need to wait for a response before retrying.
* temp_storage holds the tail end of the queue and m_query_queue contains the query we attempted
* to route.
*/
mxb_assert(m_query_queue.size() == 1);
temp_storage.push_front(std::move(m_query_queue.front()));
m_query_queue = std::move(temp_storage);
break;
@ -810,6 +801,11 @@ bool RWSplitSession::start_trx_replay()
m_trx.close();
m_trx = m_orig_trx;
m_current_query.copy_from(m_orig_stmt);
// Erase all replayed queries from the query queue to prevent checksum mismatches
m_query_queue.erase(std::remove_if(m_query_queue.begin(), m_query_queue.end(), [](mxs::Buffer b) {
return GWBUF_IS_REPLAYED(b.get());
}), m_query_queue.end());
}
if (m_trx.have_stmts() || m_current_query.get())