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

Send KILL commands to backends

Browse Source 
KILL commands are now sent to the backends in an asynchronous manner. As
the LocalClient class is used to connect to the servers, this will cause
an extra connection to be created on top of the original connections
created by the session.

If the user does not have the permissions to execute the KILL, the error
message is currently lost. This could be solved by adding a "result
handler" into the LocalClient class which is called with the result.
This commit is contained in:
Markus Mäkelä
2017-09-27 16:18:36 +03:00
parent a7e610a70a
commit 4dd6842447
11 changed files with 144 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/modules/protocol/MySQL/CMakeLists.txt
View File

@ -1,4 +1,4 @@
add_library(MySQLCommon SHARED mysql_common.c)
add_library(MySQLCommon SHARED mysql_common.cc mariadb_client.cc)
target_link_libraries(MySQLCommon maxscale-common)
set_target_properties(MySQLCommon PROPERTIES VERSION "2.0.0")
install_module(MySQLCommon core)

34
server/modules/protocol/MySQL/MySQLClient/mysql_client.cc
View File

@ -47,13 +47,6 @@ typedef enum spec_com_res_t
// wait for more data.
} spec_com_res_t;
/* Type of the kill-command sent by client. */
typedef enum kill_type
{
KT_CONNECTION,
KT_QUERY
} kill_type_t;
const char WORD_KILL[] = "KILL";
static int process_init(void);
@ -1674,9 +1667,7 @@ static spec_com_res_t process_special_commands(DCB *dcb, GWBUF *read_buffer, int
== sizeof(bytes))
{
uint64_t process_id = gw_mysql_get_byte4(bytes);
session_broadcast_kill_command(dcb->session, process_id);
// Even if id not found, send ok. TODO: send a correct response to client
mxs_mysql_send_ok(dcb, 1, 0, NULL);
mxs_mysql_execute_kill(dcb->session, process_id, KT_CONNECTION);
rval = RES_END;
}
}
@ -1735,30 +1726,11 @@ spec_com_res_t handle_query_kill(DCB* dcb, GWBUF* read_buffer, spec_com_res_t cu
kill_type_t kt = KT_CONNECTION;
uint64_t thread_id = 0;
bool parsed = parse_kill_query(querybuf, &thread_id, &kt);
rval = RES_END;
if (parsed && (thread_id > 0)) // MaxScale session counter starts at 1
{
switch (kt)
{
case KT_CONNECTION:
session_broadcast_kill_command(dcb->session, thread_id);
// Even if id not found, send ok. TODO: send a correct response to client
mxs_mysql_send_ok(dcb, 1, 0, NULL);
rval = RES_END;
break;
case KT_QUERY:
// TODO: Implement this
MXS_WARNING("Received 'KILL QUERY %" PRIu64 "' from "
"the client. This feature is not supported.", thread_id);
mysql_send_custom_error(dcb, 1, 0, "'KILL QUERY <thread_id>' "
"is not supported.");
rval = RES_END;
break;
default:
ss_dassert(!true);
}
mxs_mysql_execute_kill(dcb->session, thread_id, kt);
}
}
}

296
server/modules/protocol/MySQL/mariadb_client.cc Normal file
View File

@ -0,0 +1,296 @@
/*
* Copyright (c) 2016 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: 2020-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 <maxscale/protocol/mariadb_client.hh>
#include <maxscale/utils.h>
// TODO: Find a way to cleanly expose this
#include "../../../core/maxscale/worker.hh"
#ifdef EPOLLRDHUP
#define ERROR_EVENTS (EPOLLRDHUP | EPOLLHUP | EPOLLERR)
#else
#define ERROR_EVENTS (EPOLLHUP | EPOLLERR)
#endif
static const uint32_t poll_events = EPOLLIN | EPOLLOUT | EPOLLET | ERROR_EVENTS;
LocalClient::LocalClient(MXS_SESSION* session, int fd):
m_state(VC_WAITING_HANDSHAKE),
m_sock(fd),
m_expected_bytes(0),
m_client({}),
m_protocol({}),
m_self_destruct(false)
{
MXS_POLL_DATA::handler = LocalClient::poll_handler;
MySQLProtocol* client = (MySQLProtocol*)session->client_dcb->protocol;
m_protocol.charset = client->charset;
m_protocol.client_capabilities = client->client_capabilities;
m_protocol.extra_capabilities = client->extra_capabilities;
gw_get_shared_session_auth_info(session->client_dcb, &m_client);
}
LocalClient::~LocalClient()
{
if (m_state != VC_ERROR)
{
close();
}
}
bool LocalClient::queue_query(GWBUF* buffer)
{
GWBUF* my_buf = NULL;
if (m_state != VC_ERROR && (my_buf = gwbuf_clone(buffer)))
{
m_queue.push_back(my_buf);
if (m_state == VC_OK)
{
drain_queue();
}
}
return my_buf != NULL;
}
void LocalClient::self_destruct()
{
GWBUF* buffer = mysql_create_com_quit(NULL, 0);
queue_query(buffer);
gwbuf_free(buffer);
m_self_destruct = true;
}
void LocalClient::close()
{
mxs::Worker* worker = mxs::Worker::get_current();
ss_dassert(worker);
worker->remove_fd(m_sock);
::close(m_sock);
}
void LocalClient::error()
{
if (m_state != VC_ERROR)
{
close();
m_state = VC_ERROR;
}
}
void LocalClient::process(uint32_t events)
{
if (events & EPOLLIN)
{
GWBUF* buf = read_complete_packet();
if (buf)
{
if (m_state == VC_WAITING_HANDSHAKE)
{
if (gw_decode_mysql_server_handshake(&m_protocol, GWBUF_DATA(buf) + MYSQL_HEADER_LEN) == 0)
{
GWBUF* response = gw_generate_auth_response(&m_client, &m_protocol, false, false);
m_queue.push_front(response);
m_state = VC_RESPONSE_SENT;
}
else
{
error();
}
}
else if (m_state == VC_RESPONSE_SENT)
{
if (mxs_mysql_is_ok_packet(buf))
{
m_state = VC_OK;
}
else
{
error();
}
}
gwbuf_free(buf);
}
}
if (events & EPOLLOUT)
{
/** Queue is drained */
}
if (events & ERROR_EVENTS)
{
error();
}
if (m_queue.size() && m_state != VC_ERROR)
{
drain_queue();
}
else if (m_state == VC_ERROR && m_self_destruct)
{
delete this;
}
}
GWBUF* LocalClient::read_complete_packet()
{
GWBUF* rval = NULL;
while (true)
{
uint8_t buffer[1024];
int rc = read(m_sock, buffer, sizeof(buffer));
if (rc == -1)
{
if (errno != EAGAIN && errno != EWOULDBLOCK)
{
MXS_ERROR("Failed to read from backend: %d, %s", errno, mxs_strerror(errno));
error();
}
break;
}
mxs::Buffer chunk(buffer, rc);
m_partial.append(chunk);
size_t len = m_partial.length();
if (m_expected_bytes == 0 && len >= 3)
{
mxs::Buffer::iterator iter = m_partial.begin();
m_expected_bytes = MYSQL_HEADER_LEN;
m_expected_bytes += *iter++;
m_expected_bytes += (*iter++ << 8);
m_expected_bytes += (*iter++ << 16);
}
if (len >= m_expected_bytes)
{
/** Read complete packet. Reset expected byte count and make
* the buffer contiguous. */
m_expected_bytes = 0;
m_partial.make_contiguous();
rval = m_partial.release();
break;
}
}
return rval;
}
void LocalClient::drain_queue()
{
bool more = true;
while (m_queue.size() && more)
{
/** Grab a buffer from the queue */
GWBUF* buf = m_queue.front().release();
m_queue.pop_front();
while (buf)
{
int rc = write(m_sock, GWBUF_DATA(buf), GWBUF_LENGTH(buf));
if (rc > 0)
{
buf = gwbuf_consume(buf, rc);
}
else
{
if (rc == -1 && errno != EAGAIN && errno != EWOULDBLOCK)
{
MXS_ERROR("Failed to write to backend: %d, %s", errno, mxs_strerror(errno));
error();
}
m_queue.push_front(buf);
more = false;
break;
}
}
}
}
uint32_t LocalClient::poll_handler(struct mxs_poll_data* data, int wid, uint32_t events)
{
LocalClient* client = static_cast<LocalClient*>(data);
client->process(events);
return 0;
}
LocalClient* LocalClient::create(MXS_SESSION* session, const char* ip, uint64_t port)
{
LocalClient* rval = NULL;
sockaddr_storage addr;
int fd = open_network_socket(MXS_SOCKET_NETWORK, &addr, ip, port);
if (fd > 0 && (connect(fd, (struct sockaddr*)&addr, sizeof(addr)) == 0 || errno == EINPROGRESS))
{
LocalClient* relay = new (std::nothrow) LocalClient(session, fd);
if (relay)
{
mxs::Worker* worker = mxs::Worker::get_current();
if (worker->add_fd(fd, poll_events, (MXS_POLL_DATA*)relay))
{
rval = relay;
}
else
{
relay->m_state = VC_ERROR;
delete rval;
rval = NULL;
}
}
}
if (rval == NULL && fd > 0)
{
::close(fd);
}
return rval;
}
LocalClient* LocalClient::create(MXS_SESSION* session, SERVICE* service)
{
LocalClient* rval = NULL;
LISTENER_ITERATOR iter;
for (SERV_LISTENER* listener = listener_iterator_init(service, &iter);
listener; listener = listener_iterator_next(&iter))
{
if (listener->port > 0)
{
/** Pick the first network listener */
rval = create(session, "127.0.0.1", service->ports->port);
break;
}
}
return rval;
}
LocalClient* LocalClient::create(MXS_SESSION* session, SERVER* server)
{
return create(session, server->name, server->port);
}

64
server/modules/protocol/MySQL/mysql_common.c → server/modules/protocol/MySQL/mysql_common.cc
View File

@ -17,6 +17,9 @@
#include <netinet/tcp.h>
#include <sstream>
#include <vector>
#include <maxscale/alloc.h>
#include <maxscale/hk_heartbeat.h>
#include <maxscale/log_manager.h>
@ -24,6 +27,8 @@
#include <maxscale/mysql_utils.h>
#include <maxscale/protocol/mysql.h>
#include <maxscale/utils.h>
#include <maxscale/protocol/mariadb_client.hh>
uint8_t null_client_sha1[MYSQL_SCRAMBLE_LEN] = "";
@ -31,7 +36,7 @@ static server_command_t* server_command_init(server_command_t* srvcmd, mxs_mysql
MYSQL_session* mysql_session_alloc()
{
MYSQL_session *ses = MXS_CALLOC(1, sizeof(MYSQL_session));
MYSQL_session* ses = (MYSQL_session*)MXS_CALLOC(1, sizeof(MYSQL_session));
if (ses)
{
@ -1358,7 +1363,7 @@ mxs_auth_state_t gw_send_backend_auth(DCB *dcb)
MYSQL_session client;
gw_get_shared_session_auth_info(dcb->session->client_dcb, &client);
GWBUF* buffer = gw_generate_auth_response(&client, dcb->protocol,
GWBUF* buffer = gw_generate_auth_response(&client, (MySQLProtocol*)dcb->protocol,
with_ssl, ssl_established);
ss_dassert(buffer);
@ -1683,3 +1688,58 @@ bool mxs_mysql_command_will_respond(uint8_t cmd)
cmd != MXS_COM_STMT_CLOSE &&
cmd != MXS_COM_STMT_FETCH;
}
typedef std::vector< std::pair<SERVER*, uint64_t> > TargetList;
struct KillInfo
{
uint64_t target_id;
TargetList targets;
};
static bool kill_func(DCB *dcb, void *data)
{
KillInfo* info = (KillInfo*)data;
if (dcb->dcb_role == DCB_ROLE_BACKEND_HANDLER &&
dcb->session->ses_id == info->target_id)
{
MySQLProtocol* proto = (MySQLProtocol*)dcb->protocol;
info->targets.push_back(std::make_pair(dcb->server, proto->thread_id));
}
return true;
}
void mxs_mysql_execute_kill(MXS_SESSION* issuer, uint64_t target_id, kill_type_t type)
{
// Gather a list of servers and connection IDs to kill
KillInfo info = {target_id};
dcb_foreach(kill_func, &info);
if (info.targets.empty())
{
// No session found, send an error
std::stringstream err;
err << "Unknown thread id: " << target_id;
mysql_send_standard_error(issuer->client_dcb, 1, 1094, err.str().c_str());
}
else
{
// Execute the KILL on all of the servers
for (TargetList::iterator it = info.targets.begin();
it != info.targets.end(); it++)
{
LocalClient* client = LocalClient::create(issuer, it->first);
std::stringstream ss;
ss << "KILL " << (type == KT_QUERY ? "QUERY " : "") << it->second;
GWBUF* buffer = modutil_create_query(ss.str().c_str());
client->queue_query(buffer);
gwbuf_free(buffer);
// The LocalClient needs to delete itself once the queries are done
client->self_destruct();
}
mxs_mysql_send_ok(issuer->client_dcb, 1, 0, NULL);
}
}