Merge branch 'develop' into 1.2.1-binlog_router_trx
This commit is contained in:
@ -1,716 +0,0 @@
|
||||
/*
|
||||
* This file is distributed as part of MariaDB Corporation MaxScale. It is free
|
||||
* software: you can redistribute it and/or modify it under the terms of the
|
||||
* GNU General Public License as published by the Free Software Foundation,
|
||||
* version 2.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful, but WITHOUT
|
||||
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
|
||||
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
|
||||
* details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License along with
|
||||
* this program; if not, write to the Free Software Foundation, Inc., 51
|
||||
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
|
||||
*
|
||||
* Copyright MariaDB Corporation Ab 2013-2014
|
||||
*/
|
||||
|
||||
/**
|
||||
* @file GaleraHACRoute.c - A connection load balancer for use in a Galera
|
||||
* HA environment
|
||||
*
|
||||
*
|
||||
* @verbatim
|
||||
* Revision History
|
||||
*
|
||||
* Date Who Description
|
||||
* 14/02/2014 Mark Riddoch Initial implementation as part of
|
||||
* preparing the tutorial
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <service.h>
|
||||
#include <server.h>
|
||||
#include <router.h>
|
||||
#include <atomic.h>
|
||||
#include <spinlock.h>
|
||||
#include <dcb.h>
|
||||
#include <spinlock.h>
|
||||
|
||||
#include <skygw_types.h>
|
||||
#include <skygw_utils.h>
|
||||
#include <log_manager.h>
|
||||
|
||||
#include <mysql_client_server_protocol.h>
|
||||
|
||||
static char *version_str = "V1.0.0";
|
||||
|
||||
/* The router entry points */
|
||||
static ROUTER *GHACreateInstance(SERVICE *service, char **options);
|
||||
static void *GHANewSession(ROUTER *instance, SESSION *session);
|
||||
static void GHACloseSession(ROUTER *instance, void *router_session);
|
||||
static void GHAFreeSession(ROUTER *instance, void *router_session);
|
||||
static int GHARouteQuery(ROUTER *instance, void *router_session, GWBUF *queue);
|
||||
static void GHADiagnostics(ROUTER *instance, DCB *dcb);
|
||||
|
||||
static void GHAClientReply(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
GWBUF *queue,
|
||||
DCB *backend_dcb);
|
||||
|
||||
static void GHAHandleError(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
char *message,
|
||||
DCB *backend_dcb,
|
||||
int action);
|
||||
|
||||
/** The module object definition */
|
||||
static ROUTER_OBJECT MyObject = {
|
||||
GHACreateInstance,
|
||||
GHANewSession,
|
||||
GHACloseSession,
|
||||
GHAFreeSession,
|
||||
GHARouteQuery,
|
||||
GHADiagnostics,
|
||||
GHAClientReply,
|
||||
GHAHandleError
|
||||
};
|
||||
|
||||
static bool rses_begin_router_action(
|
||||
ROUTER_CLIENT_SES* rses);
|
||||
|
||||
static void rses_exit_router_action(
|
||||
ROUTER_CLIENT_SES* rses);
|
||||
|
||||
static SPINLOCK instlock;
|
||||
static ROUTER_INSTANCE *instances;
|
||||
|
||||
/**
|
||||
* Implementation of the mandatory version entry point
|
||||
*
|
||||
* @return version string of the module
|
||||
*/
|
||||
char *
|
||||
version()
|
||||
{
|
||||
return version_str;
|
||||
}
|
||||
|
||||
/**
|
||||
* The module initialisation routine, called when the module
|
||||
* is first loaded.
|
||||
*/
|
||||
void
|
||||
ModuleInit()
|
||||
{
|
||||
LOGIF(LM, (skygw_log_write(
|
||||
"Initialise readconnroute router module %s.\n", version_str)));
|
||||
spinlock_init(&instlock);
|
||||
instances = NULL;
|
||||
}
|
||||
|
||||
/**
|
||||
* The module entry point routine. It is this routine that
|
||||
* must populate the structure that is referred to as the
|
||||
* "module object", this is a structure with the set of
|
||||
* external entry points for this module.
|
||||
*
|
||||
* @return The module object
|
||||
*/
|
||||
ROUTER_OBJECT *
|
||||
GetModuleObject()
|
||||
{
|
||||
return &MyObject;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the router for a particular service
|
||||
* within the gateway.
|
||||
*
|
||||
* @param service The service this router is being create for
|
||||
* @param options An array of options for this query router
|
||||
*
|
||||
* @return The instance data for this new instance
|
||||
*/
|
||||
static ROUTER *
|
||||
GHACreateInstance(SERVICE *service, char **options)
|
||||
{
|
||||
ROUTER_INSTANCE *inst;
|
||||
SERVER_REF *server;
|
||||
int i, n;
|
||||
|
||||
if ((inst = calloc(1, sizeof(ROUTER_INSTANCE))) == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
inst->service = service;
|
||||
spinlock_init(&inst->lock);
|
||||
|
||||
/*
|
||||
* We need an array of the backend servers in the instance structure so
|
||||
* that we can maintain a count of the number of connections to each
|
||||
* backend server.
|
||||
*/
|
||||
for (server = service->dbref, n = 0; server; server = server->next)
|
||||
n++;
|
||||
|
||||
inst->servers = (BACKEND **)calloc(n + 1, sizeof(BACKEND *));
|
||||
if (!inst->servers)
|
||||
{
|
||||
free(inst);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
for (server = service->dbref, n = 0; server; server = server->next)
|
||||
{
|
||||
if ((inst->servers[n] = malloc(sizeof(BACKEND))) == NULL)
|
||||
{
|
||||
for (i = 0; i < n; i++)
|
||||
free(inst->servers[i]);
|
||||
free(inst->servers);
|
||||
free(inst);
|
||||
return NULL;
|
||||
}
|
||||
inst->servers[n]->server = server->server;
|
||||
inst->servers[n]->current_connection_count = 0;
|
||||
n++;
|
||||
}
|
||||
inst->servers[n] = NULL;
|
||||
|
||||
/*
|
||||
* Process the options
|
||||
*/
|
||||
inst->bitmask = 0;
|
||||
inst->bitvalue = 0;
|
||||
if (options)
|
||||
{
|
||||
for (i = 0; options[i]; i++)
|
||||
{
|
||||
if (!strcasecmp(options[i], "master"))
|
||||
{
|
||||
inst->bitmask |= (SERVER_MASTER|SERVER_SLAVE);
|
||||
inst->bitvalue |= SERVER_MASTER;
|
||||
}
|
||||
else if (!strcasecmp(options[i], "slave"))
|
||||
{
|
||||
inst->bitmask |= (SERVER_MASTER|SERVER_SLAVE);
|
||||
inst->bitvalue |= SERVER_SLAVE;
|
||||
}
|
||||
else if (!strcasecmp(options[i], "synced"))
|
||||
{
|
||||
inst->bitmask |= (SERVER_JOINED);
|
||||
inst->bitvalue |= SERVER_JOINED;
|
||||
}
|
||||
else
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"Warning : Unsupported router "
|
||||
"option %s for readconnroute.",
|
||||
options[i])));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* We have completed the creation of the instance data, so now
|
||||
* insert this router instance into the linked list of routers
|
||||
* that have been created with this module.
|
||||
*/
|
||||
spinlock_acquire(&instlock);
|
||||
inst->next = instances;
|
||||
instances = inst;
|
||||
spinlock_release(&instlock);
|
||||
|
||||
return (ROUTER *)inst;
|
||||
}
|
||||
|
||||
/**
|
||||
* Associate a new session with this instance of the router.
|
||||
*
|
||||
* @param instance The router instance data
|
||||
* @param session The session itself
|
||||
* @return Session specific data for this session
|
||||
*/
|
||||
static void *
|
||||
GHANewSession(ROUTER *instance, SESSION *session)
|
||||
{
|
||||
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *)instance;
|
||||
ROUTER_CLIENT_SES *client_rses;
|
||||
BACKEND *candidate = NULL;
|
||||
BACKEND *master = NULL;
|
||||
int i;
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] new router session with session "
|
||||
"%p, and inst %p.",
|
||||
pthread_self(),
|
||||
session,
|
||||
inst)));
|
||||
|
||||
|
||||
client_rses = (ROUTER_CLIENT_SES *)calloc(1, sizeof(ROUTER_CLIENT_SES));
|
||||
|
||||
if (client_rses == NULL) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
#if defined(SS_DEBUG)
|
||||
client_rses->rses_chk_top = CHK_NUM_ROUTER_SES;
|
||||
client_rses->rses_chk_tail = CHK_NUM_ROUTER_SES;
|
||||
#endif
|
||||
|
||||
/**
|
||||
* Find a backend server to connect to. This is the extent of the
|
||||
* load balancing algorithm we need to implement for this simple
|
||||
* connection router.
|
||||
*
|
||||
* The simple Galera HA connection router assumes the first node
|
||||
* in the lsit that is part of the cluster is the master and the
|
||||
* remainder are slaves.
|
||||
*
|
||||
* We loop over all the servers, the first one we find that is a
|
||||
* member of the cluster we designate as the master. We then
|
||||
* look at the remainder of the servers and find the one with
|
||||
* least connections and make this our candiate slave server.
|
||||
*/
|
||||
|
||||
for (i = 0; inst->servers[i]; i++)
|
||||
{
|
||||
if(inst->servers[i])
|
||||
{
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] Examine server in port %d with "
|
||||
"%d connections. Status is %d, "
|
||||
"inst->bitvalue is %d",
|
||||
pthread_self(),
|
||||
inst->servers[i]->server->port,
|
||||
inst->servers[i]->current_connection_count,
|
||||
inst->servers[i]->server->status,
|
||||
inst->bitmask)));
|
||||
}
|
||||
if (inst->servers[i] &&
|
||||
SERVER_IS_RUNNING(inst->servers[i]->server) &&
|
||||
(inst->servers[i]->server->status & SERVER_SYNCED))
|
||||
{
|
||||
if (master == NULL)
|
||||
master = inst->servers[i];
|
||||
else
|
||||
{
|
||||
/* If no candidate set, set first running server as
|
||||
our initial candidate server */
|
||||
if (candidate == NULL)
|
||||
{
|
||||
candidate = inst->servers[i];
|
||||
}
|
||||
else if (inst->servers[i]->current_connection_count <
|
||||
candidate->current_connection_count)
|
||||
{
|
||||
/* This running server has fewer
|
||||
connections, set it as a new candidate */
|
||||
candidate = inst->servers[i];
|
||||
}
|
||||
else if (inst->servers[i]->current_connection_count ==
|
||||
candidate->current_connection_count &&
|
||||
inst->servers[i]->server->stats.n_connections <
|
||||
candidate->server->stats.n_connections)
|
||||
{
|
||||
/* This running server has the same number
|
||||
of connections currently as the candidate
|
||||
but has had fewer connections over time
|
||||
than candidate, set this server to candidate*/
|
||||
candidate = inst->servers[i];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (candidate == NULL) /* Only have the master at best */
|
||||
candidate = master;
|
||||
|
||||
/*
|
||||
* master is our master server to connect to and candidate is the best
|
||||
* slave to connect to. Now we simply look to see if this router
|
||||
* instance should connect to a master or a slave and set the final
|
||||
* value of candidate to either the master or candidate slave.
|
||||
*/
|
||||
if (inst->bitvalue & SERVER_MASTER)
|
||||
{
|
||||
candidate = master;
|
||||
}
|
||||
|
||||
/* no candidate server here, clean and return NULL */
|
||||
if (!candidate) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Failed to create new routing session. "
|
||||
"Couldn't find eligible candidate server. Freeing "
|
||||
"allocated resources.")));
|
||||
free(client_rses);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/*
|
||||
* We now have the server with the least connections.
|
||||
* Bump the connection count for this server
|
||||
*/
|
||||
atomic_add(&candidate->current_connection_count, 1);
|
||||
client_rses->backend = candidate;
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] Selected server in port %d. "
|
||||
"Connections : %d\n",
|
||||
pthread_self(),
|
||||
candidate->server->port,
|
||||
candidate->current_connection_count)));
|
||||
/*
|
||||
* Open a backend connection, putting the DCB for this
|
||||
* connection in the client_rses->backend_dcb
|
||||
*/
|
||||
client_rses->backend_dcb = dcb_connect(candidate->server,
|
||||
session,
|
||||
candidate->server->protocol);
|
||||
if (client_rses->backend_dcb == NULL)
|
||||
{
|
||||
atomic_add(&candidate->current_connection_count, -1);
|
||||
free(client_rses);
|
||||
return NULL;
|
||||
}
|
||||
inst->stats.n_sessions++;
|
||||
|
||||
/**
|
||||
* Add this session to the list of active sessions.
|
||||
*/
|
||||
spinlock_acquire(&inst->lock);
|
||||
client_rses->next = inst->connections;
|
||||
inst->connections = client_rses;
|
||||
spinlock_release(&inst->lock);
|
||||
|
||||
CHK_CLIENT_RSES(client_rses);
|
||||
|
||||
return (void *)client_rses;
|
||||
}
|
||||
|
||||
/**
|
||||
* @node Unlink from backend server, unlink from router's connection list,
|
||||
* and free memory of a router client session.
|
||||
*
|
||||
* Parameters:
|
||||
* @param router - <usage>
|
||||
* <description>
|
||||
*
|
||||
* @param router_cli_ses - <usage>
|
||||
* <description>
|
||||
*
|
||||
* @return void
|
||||
*
|
||||
*
|
||||
* @details (write detailed description here)
|
||||
*
|
||||
*/
|
||||
static void GHAFreeSession(
|
||||
ROUTER* router_instance,
|
||||
void* router_client_ses)
|
||||
{
|
||||
ROUTER_INSTANCE* router = (ROUTER_INSTANCE *)router_instance;
|
||||
ROUTER_CLIENT_SES* router_cli_ses =
|
||||
(ROUTER_CLIENT_SES *)router_client_ses;
|
||||
int prev_val;
|
||||
|
||||
prev_val = atomic_add(&router_cli_ses->backend->current_connection_count, -1);
|
||||
ss_dassert(prev_val > 0);
|
||||
|
||||
atomic_add(&router_cli_ses->backend->server->stats.n_current, -1);
|
||||
spinlock_acquire(&router->lock);
|
||||
|
||||
if (router->connections == router_cli_ses) {
|
||||
router->connections = router_cli_ses->next;
|
||||
} else {
|
||||
ROUTER_CLIENT_SES *ptr = router->connections;
|
||||
|
||||
while (ptr != NULL && ptr->next != router_cli_ses) {
|
||||
ptr = ptr->next;
|
||||
}
|
||||
|
||||
if (ptr != NULL) {
|
||||
ptr->next = router_cli_ses->next;
|
||||
}
|
||||
}
|
||||
spinlock_release(&router->lock);
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [freeSession] Unlinked router_client_session %p from "
|
||||
"router %p and from server on port %d. Connections : %d. ",
|
||||
pthread_self(),
|
||||
router_cli_ses,
|
||||
router,
|
||||
router_cli_ses->backend->server->port,
|
||||
prev_val-1)));
|
||||
|
||||
free(router_cli_ses);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Close a session with the router, this is the mechanism
|
||||
* by which a router may cleanup data structure etc.
|
||||
*
|
||||
* @param instance The router instance data
|
||||
* @param router_session The session being closed
|
||||
*/
|
||||
static void
|
||||
GHACloseSession(ROUTER *instance, void *router_session)
|
||||
{
|
||||
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
|
||||
DCB* backend_dcb;
|
||||
|
||||
CHK_CLIENT_RSES(router_cli_ses);
|
||||
/**
|
||||
* Lock router client session for secure read and update.
|
||||
*/
|
||||
if (rses_begin_router_action(router_cli_ses))
|
||||
{
|
||||
backend_dcb = router_cli_ses->backend_dcb;
|
||||
router_cli_ses->backend_dcb = NULL;
|
||||
router_cli_ses->rses_closed = true;
|
||||
/** Unlock */
|
||||
rses_exit_router_action(router_cli_ses);
|
||||
|
||||
/**
|
||||
* Close the backend server connection
|
||||
*/
|
||||
if (backend_dcb != NULL) {
|
||||
CHK_DCB(backend_dcb);
|
||||
dcb_close(backend_dcb);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* We have data from the client, we must route it to the backend.
|
||||
* This is simply a case of sending it to the connection that was
|
||||
* chosen when we started the client session.
|
||||
*
|
||||
* @param instance The router instance
|
||||
* @param router_session The router session returned from the newSession call
|
||||
* @param queue The queue of data buffers to route
|
||||
* @return The number of bytes sent
|
||||
*/
|
||||
static int
|
||||
GHARouteQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
||||
{
|
||||
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *)instance;
|
||||
ROUTER_CLIENT_SES *router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
|
||||
uint8_t *payload = GWBUF_DATA(queue);
|
||||
int mysql_command;
|
||||
int rc;
|
||||
DCB* backend_dcb;
|
||||
bool rses_is_closed;
|
||||
|
||||
inst->stats.n_queries++;
|
||||
mysql_command = MYSQL_GET_COMMAND(payload);
|
||||
|
||||
/** Dirty read for quick check if router is closed. */
|
||||
if (router_cli_ses->rses_closed)
|
||||
{
|
||||
rses_is_closed = true;
|
||||
}
|
||||
else
|
||||
{
|
||||
/**
|
||||
* Lock router client session for secure read of DCBs
|
||||
*/
|
||||
rses_is_closed = !(rses_begin_router_action(router_cli_ses));
|
||||
}
|
||||
|
||||
if (!rses_is_closed)
|
||||
{
|
||||
backend_dcb = router_cli_ses->backend_dcb;
|
||||
/** unlock */
|
||||
rses_exit_router_action(router_cli_ses);
|
||||
}
|
||||
|
||||
if (rses_is_closed || backend_dcb == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"Error: Failed to route MySQL command %d to backend "
|
||||
"server.",
|
||||
mysql_command)));
|
||||
goto return_rc;
|
||||
}
|
||||
|
||||
switch(mysql_command) {
|
||||
case MYSQL_COM_CHANGE_USER:
|
||||
rc = backend_dcb->func.auth(
|
||||
backend_dcb,
|
||||
NULL,
|
||||
backend_dcb->session,
|
||||
queue);
|
||||
break;
|
||||
default:
|
||||
rc = backend_dcb->func.write(backend_dcb, queue);
|
||||
break;
|
||||
}
|
||||
|
||||
CHK_PROTOCOL(((MySQLProtocol*)backend_dcb->protocol));
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [readconnroute:routeQuery] Routed command %d to dcb %p "
|
||||
"with return value %d.",
|
||||
pthread_self(),
|
||||
mysql_command,
|
||||
backend_dcb,
|
||||
rc)));
|
||||
return_rc:
|
||||
return rc;
|
||||
}
|
||||
|
||||
/**
|
||||
* Display router diagnostics
|
||||
*
|
||||
* @param instance Instance of the router
|
||||
* @param dcb DCB to send diagnostics to
|
||||
*/
|
||||
static void
|
||||
GHADiagnostics(ROUTER *router, DCB *dcb)
|
||||
{
|
||||
ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *)router;
|
||||
ROUTER_CLIENT_SES *session;
|
||||
int i = 0;
|
||||
|
||||
spinlock_acquire(&router_inst->lock);
|
||||
session = router_inst->connections;
|
||||
while (session)
|
||||
{
|
||||
i++;
|
||||
session = session->next;
|
||||
}
|
||||
spinlock_release(&router_inst->lock);
|
||||
|
||||
dcb_printf(dcb, "\tNumber of router sessions: %d\n",
|
||||
router_inst->stats.n_sessions);
|
||||
dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n", i);
|
||||
dcb_printf(dcb, "\tNumber of queries forwarded: %d\n",
|
||||
router_inst->stats.n_queries);
|
||||
}
|
||||
|
||||
/**
|
||||
* Client Reply routine
|
||||
*
|
||||
* The routine will reply to client data from backend server
|
||||
*
|
||||
* @param instance The router instance
|
||||
* @param router_session The router session
|
||||
* @param backend_dcb The backend DCB
|
||||
* @param queue The GWBUF with reply data
|
||||
*/
|
||||
static void
|
||||
GHAClientReply(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
GWBUF *queue,
|
||||
DCB *backend_dcb)
|
||||
{
|
||||
DCB *client = NULL;
|
||||
|
||||
client = backend_dcb->session->client;
|
||||
|
||||
ss_dassert(client != NULL);
|
||||
|
||||
client->func.write(client, queue);
|
||||
}
|
||||
|
||||
/**
|
||||
* Error handling routine
|
||||
*
|
||||
* The routine will handle error occurred in backend.
|
||||
*
|
||||
* @param instance The router instance
|
||||
* @param router_session The router session
|
||||
* @param message The error message to reply
|
||||
* @param backend_dcb The backend DCB
|
||||
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
|
||||
*
|
||||
*/
|
||||
static void
|
||||
GHAHandleError(
|
||||
ROUTER *instance,
|
||||
void *router_session,
|
||||
char *message,
|
||||
DCB *backend_dcb,
|
||||
int action)
|
||||
{
|
||||
DCB *client = NULL;
|
||||
ROUTER_OBJECT *router = NULL;
|
||||
SESSION *session = backend_dcb->session;
|
||||
client = backend_dcb->session->client;
|
||||
|
||||
ss_dassert(client != NULL);
|
||||
}
|
||||
|
||||
/** to be inline'd */
|
||||
/**
|
||||
* @node Acquires lock to router client session if it is not closed.
|
||||
*
|
||||
* Parameters:
|
||||
* @param rses - in, use
|
||||
*
|
||||
*
|
||||
* @return true if router session was not closed. If return value is true
|
||||
* it means that router is locked, and must be unlocked later. False, if
|
||||
* router was closed before lock was acquired.
|
||||
*
|
||||
*
|
||||
* @details (write detailed description here)
|
||||
*
|
||||
*/
|
||||
static bool rses_begin_router_action(
|
||||
ROUTER_CLIENT_SES* rses)
|
||||
{
|
||||
bool succp = false;
|
||||
|
||||
CHK_CLIENT_RSES(rses);
|
||||
|
||||
if (rses->rses_closed) {
|
||||
goto return_succp;
|
||||
}
|
||||
spinlock_acquire(&rses->rses_lock);
|
||||
if (rses->rses_closed) {
|
||||
spinlock_release(&rses->rses_lock);
|
||||
goto return_succp;
|
||||
}
|
||||
succp = true;
|
||||
|
||||
return_succp:
|
||||
return succp;
|
||||
}
|
||||
|
||||
/** to be inline'd */
|
||||
/**
|
||||
* @node Releases router client session lock.
|
||||
*
|
||||
* Parameters:
|
||||
* @param rses - <usage>
|
||||
* <description>
|
||||
*
|
||||
* @return void
|
||||
*
|
||||
*
|
||||
* @details (write detailed description here)
|
||||
*
|
||||
*/
|
||||
static void rses_exit_router_action(
|
||||
ROUTER_CLIENT_SES* rses)
|
||||
{
|
||||
CHK_CLIENT_RSES(rses);
|
||||
spinlock_release(&rses->rses_lock);
|
||||
}
|
||||
@ -125,10 +125,7 @@ version()
|
||||
void
|
||||
ModuleInit()
|
||||
{
|
||||
LOGIF(LM, (skygw_log_write(
|
||||
LOGFILE_MESSAGE,
|
||||
"Initialise MaxInfo router module %s.\n",
|
||||
version_str)));
|
||||
MXS_NOTICE("Initialise MaxInfo router module %s.", version_str);
|
||||
spinlock_init(&instlock);
|
||||
instances = NULL;
|
||||
}
|
||||
@ -172,12 +169,7 @@ int i;
|
||||
{
|
||||
for (i = 0; options[i]; i++)
|
||||
{
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"Unknown option for MaxInfo '%s'\n",
|
||||
options[i])));
|
||||
}
|
||||
MXS_ERROR("Unknown option for MaxInfo '%s'", options[i]);
|
||||
}
|
||||
}
|
||||
|
||||
@ -389,9 +381,8 @@ char *sql;
|
||||
case COM_STATISTICS:
|
||||
return maxinfo_statistics(instance, session, queue);
|
||||
default:
|
||||
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
|
||||
"maxinfo: Unexpected MySQL command 0x%x",
|
||||
MYSQL_COMMAND(queue))));
|
||||
MXS_ERROR("maxinfo: Unexpected MySQL command 0x%x",
|
||||
MYSQL_COMMAND(queue));
|
||||
}
|
||||
}
|
||||
|
||||
@ -619,9 +610,8 @@ maxinfo_execute_query(INFO_INSTANCE *instance, INFO_SESSION *session, char *sql)
|
||||
MAXINFO_TREE *tree;
|
||||
PARSE_ERROR err;
|
||||
|
||||
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
|
||||
"maxinfo: SQL statement: '%s' for 0x%p.",
|
||||
sql, session->dcb)));
|
||||
MXS_INFO("maxinfo: SQL statement: '%s' for 0x%p.",
|
||||
sql, session->dcb);
|
||||
if (strcmp(sql, "select @@version_comment limit 1") == 0)
|
||||
{
|
||||
respond_vercom(session->dcb);
|
||||
@ -657,10 +647,7 @@ PARSE_ERROR err;
|
||||
if ((tree = maxinfo_parse(sql, &err)) == NULL)
|
||||
{
|
||||
maxinfo_send_parse_error(session->dcb, sql, err);
|
||||
LOGIF(LM, (skygw_log_write(
|
||||
LOGFILE_MESSAGE,
|
||||
"Failed to parse SQL statement: '%s'.",
|
||||
sql)));
|
||||
MXS_NOTICE("Failed to parse SQL statement: '%s'.", sql);
|
||||
}
|
||||
else
|
||||
maxinfo_execute(session->dcb, tree);
|
||||
@ -755,8 +742,7 @@ maxinfo_add_mysql_user(SERVICE *service) {
|
||||
char *service_passwd = NULL;
|
||||
|
||||
if (serviceGetUser(service, &service_user, &service_passwd) == 0) {
|
||||
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
|
||||
"maxinfo: failed to get service user details")));
|
||||
MXS_ERROR("maxinfo: failed to get service user details");
|
||||
|
||||
return 1;
|
||||
}
|
||||
@ -764,9 +750,8 @@ maxinfo_add_mysql_user(SERVICE *service) {
|
||||
dpwd = decryptPassword(service->credentials.authdata);
|
||||
|
||||
if (!dpwd) {
|
||||
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
|
||||
"maxinfo: decrypt password failed for service user %s",
|
||||
service_user)));
|
||||
MXS_ERROR("maxinfo: decrypt password failed for service user %s",
|
||||
service_user);
|
||||
|
||||
return 1;
|
||||
}
|
||||
@ -776,9 +761,8 @@ maxinfo_add_mysql_user(SERVICE *service) {
|
||||
newpasswd = create_hex_sha1_sha1_passwd(dpwd);
|
||||
|
||||
if (!newpasswd) {
|
||||
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
|
||||
"maxinfo: create hex_sha1_sha1_password failed for service user %s",
|
||||
service_user)));
|
||||
MXS_ERROR("maxinfo: create hex_sha1_sha1_password failed for service user %s",
|
||||
service_user);
|
||||
users_free(service->users);
|
||||
service->users = NULL;
|
||||
return 1;
|
||||
|
||||
@ -293,7 +293,7 @@ char errmsg[120];
|
||||
tree->value[80] = 0;
|
||||
sprintf(errmsg, "Unsupported show command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE, "%s", errmsg)));
|
||||
MXS_NOTICE("%s", errmsg);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -345,7 +345,7 @@ exec_flush(DCB *dcb, MAXINFO_TREE *tree)
|
||||
}
|
||||
sprintf(errmsg, "Unsupported flush command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
skygw_log_write(LE, "%s", errmsg);
|
||||
MXS_ERROR("%s", errmsg);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -425,7 +425,7 @@ exec_set(DCB *dcb, MAXINFO_TREE *tree)
|
||||
}
|
||||
sprintf(errmsg, "Unsupported set command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
skygw_log_write(LE, "%s", errmsg);
|
||||
MXS_ERROR("%s", errmsg);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -505,7 +505,7 @@ exec_clear(DCB *dcb, MAXINFO_TREE *tree)
|
||||
}
|
||||
sprintf(errmsg, "Unsupported clear command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
skygw_log_write(LE, "%s", errmsg);
|
||||
MXS_ERROR("%s", errmsg);
|
||||
}
|
||||
|
||||
extern void shutdown_server();
|
||||
@ -627,7 +627,7 @@ exec_shutdown(DCB *dcb, MAXINFO_TREE *tree)
|
||||
}
|
||||
sprintf(errmsg, "Unsupported shutdown command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
skygw_log_write(LE, "%s", errmsg);
|
||||
MXS_ERROR("%s", errmsg);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -735,7 +735,7 @@ exec_restart(DCB *dcb, MAXINFO_TREE *tree)
|
||||
}
|
||||
sprintf(errmsg, "Unsupported restart command '%s'", tree->value);
|
||||
maxinfo_send_error(dcb, 0, errmsg);
|
||||
skygw_log_write(LE, "%s", errmsg);
|
||||
MXS_ERROR("%s", errmsg);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -171,11 +171,9 @@ version()
|
||||
void
|
||||
ModuleInit()
|
||||
{
|
||||
LOGIF(LM, (skygw_log_write(
|
||||
LOGFILE_MESSAGE,
|
||||
"Initialise readconnroute router module %s.\n", version_str)));
|
||||
spinlock_init(&instlock);
|
||||
instances = NULL;
|
||||
MXS_NOTICE("Initialise readconnroute router module %s.", version_str);
|
||||
spinlock_init(&instlock);
|
||||
instances = NULL;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -261,11 +259,10 @@ char *weightby;
|
||||
}
|
||||
if (total == 0)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
|
||||
"WARNING: Weighting Parameter for service '%s' "
|
||||
MXS_WARNING("Weighting Parameter for service '%s' "
|
||||
"will be ignored as no servers have values "
|
||||
"for the parameter '%s'.\n",
|
||||
service->name, weightby)));
|
||||
service->name, weightby);
|
||||
}
|
||||
else
|
||||
{
|
||||
@ -280,14 +277,12 @@ char *weightby;
|
||||
backend->weight = perc;
|
||||
if (perc == 0)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write(
|
||||
LOGFILE_ERROR,
|
||||
"Server '%s' has no value "
|
||||
"for weighting parameter '%s', "
|
||||
"no queries will be routed to "
|
||||
"this server.\n",
|
||||
inst->servers[n]->server->unique_name,
|
||||
weightby)));
|
||||
MXS_ERROR("Server '%s' has no value "
|
||||
"for weighting parameter '%s', "
|
||||
"no queries will be routed to "
|
||||
"this server.\n",
|
||||
inst->servers[n]->server->unique_name,
|
||||
weightby);
|
||||
}
|
||||
|
||||
}
|
||||
@ -330,13 +325,11 @@ char *weightby;
|
||||
}
|
||||
else
|
||||
{
|
||||
LOGIF(LM, (skygw_log_write(
|
||||
LOGFILE_MESSAGE,
|
||||
"* Warning : Unsupported router "
|
||||
"option \'%s\' for readconnroute. "
|
||||
"Expected router options are "
|
||||
"[slave|master|synced|ndb]",
|
||||
options[i])));
|
||||
MXS_WARNING("Unsupported router "
|
||||
"option \'%s\' for readconnroute. "
|
||||
"Expected router options are "
|
||||
"[slave|master|synced|ndb]",
|
||||
options[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -375,13 +368,11 @@ BACKEND *candidate = NULL;
|
||||
int i;
|
||||
BACKEND *master_host = NULL;
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] new router session with session "
|
||||
"%p, and inst %p.",
|
||||
pthread_self(),
|
||||
session,
|
||||
inst)));
|
||||
MXS_DEBUG("%lu [newSession] new router session with session "
|
||||
"%p, and inst %p.",
|
||||
pthread_self(),
|
||||
session,
|
||||
inst);
|
||||
|
||||
|
||||
client_rses = (ROUTER_CLIENT_SES *)calloc(1, sizeof(ROUTER_CLIENT_SES));
|
||||
@ -420,16 +411,14 @@ BACKEND *master_host = NULL;
|
||||
*/
|
||||
for (i = 0; inst->servers[i]; i++) {
|
||||
if(inst->servers[i]) {
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] Examine server in port %d with "
|
||||
"%d connections. Status is %s, "
|
||||
"inst->bitvalue is %d",
|
||||
pthread_self(),
|
||||
inst->servers[i]->server->port,
|
||||
inst->servers[i]->current_connection_count,
|
||||
STRSRVSTATUS(inst->servers[i]->server),
|
||||
inst->bitmask)));
|
||||
MXS_DEBUG("%lu [newSession] Examine server in port %d with "
|
||||
"%d connections. Status is %s, "
|
||||
"inst->bitvalue is %d",
|
||||
pthread_self(),
|
||||
inst->servers[i]->server->port,
|
||||
inst->servers[i]->current_connection_count,
|
||||
STRSRVSTATUS(inst->servers[i]->server),
|
||||
inst->bitmask);
|
||||
}
|
||||
|
||||
if (SERVER_IN_MAINT(inst->servers[i]->server))
|
||||
@ -512,13 +501,11 @@ BACKEND *master_host = NULL;
|
||||
if (master_host) {
|
||||
candidate = master_host;
|
||||
} else {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Failed to create new routing session. "
|
||||
"Couldn't find eligible candidate server. Freeing "
|
||||
"allocated resources.")));
|
||||
free(client_rses);
|
||||
return NULL;
|
||||
MXS_ERROR("Failed to create new routing session. "
|
||||
"Couldn't find eligible candidate server. Freeing "
|
||||
"allocated resources.");
|
||||
free(client_rses);
|
||||
return NULL;
|
||||
}
|
||||
}
|
||||
|
||||
@ -530,13 +517,11 @@ BACKEND *master_host = NULL;
|
||||
*/
|
||||
atomic_add(&candidate->current_connection_count, 1);
|
||||
client_rses->backend = candidate;
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [newSession] Selected server in port %d. "
|
||||
"Connections : %d\n",
|
||||
pthread_self(),
|
||||
candidate->server->port,
|
||||
candidate->current_connection_count)));
|
||||
MXS_DEBUG("%lu [newSession] Selected server in port %d. "
|
||||
"Connections : %d\n",
|
||||
pthread_self(),
|
||||
candidate->server->port,
|
||||
candidate->current_connection_count);
|
||||
|
||||
/*
|
||||
* Open a backend connection, putting the DCB for this
|
||||
@ -568,10 +553,8 @@ BACKEND *master_host = NULL;
|
||||
|
||||
CHK_CLIENT_RSES(client_rses);
|
||||
|
||||
skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Readconnroute: New session for server %s. "
|
||||
"Connections : %d",
|
||||
MXS_INFO("Readconnroute: New session for server %s. "
|
||||
"Connections : %d",
|
||||
candidate->server->unique_name,
|
||||
candidate->current_connection_count);
|
||||
return (void *)client_rses;
|
||||
@ -623,15 +606,13 @@ static void freeSession(
|
||||
}
|
||||
spinlock_release(&router->lock);
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [freeSession] Unlinked router_client_session %p from "
|
||||
"router %p and from server on port %d. Connections : %d. ",
|
||||
pthread_self(),
|
||||
router_cli_ses,
|
||||
router,
|
||||
router_cli_ses->backend->server->port,
|
||||
prev_val-1)));
|
||||
MXS_DEBUG("%lu [freeSession] Unlinked router_client_session %p from "
|
||||
"router %p and from server on port %d. Connections : %d. ",
|
||||
pthread_self(),
|
||||
router_cli_ses,
|
||||
router,
|
||||
router_cli_ses->backend->server->port,
|
||||
prev_val-1);
|
||||
|
||||
free(router_cli_ses);
|
||||
}
|
||||
@ -721,14 +702,12 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
||||
if (rses_is_closed || backend_dcb == NULL ||
|
||||
SERVER_IS_DOWN(router_cli_ses->backend->server))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE|LOGFILE_ERROR,
|
||||
"Error : Failed to route MySQL command %d to backend "
|
||||
"server.%s",
|
||||
mysql_command,rses_is_closed ? " Session is closed." : "")));
|
||||
rc = 0;
|
||||
while((queue = GWBUF_CONSUME_ALL(queue)) != NULL);
|
||||
goto return_rc;
|
||||
MXS_ERROR("Failed to route MySQL command %d to backend "
|
||||
"server.%s",
|
||||
mysql_command,rses_is_closed ? " Session is closed." : "");
|
||||
rc = 0;
|
||||
while((queue = GWBUF_CONSUME_ALL(queue)) != NULL);
|
||||
goto return_rc;
|
||||
|
||||
}
|
||||
|
||||
@ -743,19 +722,20 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
|
||||
queue);
|
||||
break;
|
||||
case MYSQL_COM_QUERY:
|
||||
LOGIF(LOGFILE_TRACE,(trc = modutil_get_SQL(queue)));
|
||||
if (MXS_LOG_PRIORITY_IS_ENABLED(LOG_INFO))
|
||||
{
|
||||
trc = modutil_get_SQL(queue);
|
||||
}
|
||||
default:
|
||||
rc = backend_dcb->func.write(backend_dcb, queue);
|
||||
break;
|
||||
}
|
||||
|
||||
LOGIF(LOGFILE_TRACE,skygw_log_write(
|
||||
LOGFILE_DEBUG|LOGFILE_TRACE,
|
||||
"Routed [%s] to '%s'%s%s",
|
||||
MXS_INFO("Routed [%s] to '%s'%s%s",
|
||||
STRPACKETTYPE(mysql_command),
|
||||
backend_dcb->server->unique_name,
|
||||
trc?": ":".",
|
||||
trc?trc:""));
|
||||
trc?trc:"");
|
||||
free(trc);
|
||||
return_rc:
|
||||
return rc;
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@ -41,7 +41,7 @@ bool extract_database(GWBUF* buf, char* str)
|
||||
tok = strtok_r(query," ;",&saved);
|
||||
if(tok == NULL || strcasecmp(tok,"use") != 0)
|
||||
{
|
||||
skygw_log_write(LOGFILE_ERROR,"extract_database: Malformed chage database packet.");
|
||||
MXS_ERROR("extract_database: Malformed chage database packet.");
|
||||
succp = false;
|
||||
goto retblock;
|
||||
}
|
||||
@ -49,7 +49,7 @@ bool extract_database(GWBUF* buf, char* str)
|
||||
tok = strtok_r(NULL," ;",&saved);
|
||||
if(tok == NULL)
|
||||
{
|
||||
skygw_log_write(LOGFILE_ERROR,"extract_database: Malformed chage database packet.");
|
||||
MXS_ERROR("extract_database: Malformed chage database packet.");
|
||||
succp = false;
|
||||
goto retblock;
|
||||
}
|
||||
@ -73,16 +73,12 @@ bool extract_database(GWBUF* buf, char* str)
|
||||
*/
|
||||
void create_error_reply(char* fail_str,DCB* dcb)
|
||||
{
|
||||
skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"change_current_db: failed to change database: %s", fail_str);
|
||||
MXS_INFO("change_current_db: failed to change database: %s", fail_str);
|
||||
GWBUF* errbuf = modutil_create_mysql_err_msg(1, 0, 1049, "42000", fail_str);
|
||||
|
||||
if (errbuf == NULL)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Creating buffer for error message failed.")));
|
||||
MXS_ERROR("Creating buffer for error message failed.");
|
||||
return;
|
||||
}
|
||||
/** Set flags that help router to identify session commands reply */
|
||||
@ -120,8 +116,7 @@ bool change_current_db(char* dest,
|
||||
succp = false;
|
||||
goto retblock;
|
||||
}
|
||||
skygw_log_write(LOGFILE_TRACE,"change_current_db: INIT_DB with database '%s'",
|
||||
db);
|
||||
MXS_INFO("change_current_db: INIT_DB with database '%s'", db);
|
||||
/**
|
||||
* Update the session's active database only if it's in the hashtable.
|
||||
* If it isn't found, send a custom error packet to the client.
|
||||
@ -135,7 +130,7 @@ bool change_current_db(char* dest,
|
||||
else
|
||||
{
|
||||
strcpy(dest,db);
|
||||
skygw_log_write(LOGFILE_TRACE,"change_current_db: database is on server: '%s'.",target);
|
||||
MXS_INFO("change_current_db: database is on server: '%s'.",target);
|
||||
succp = true;
|
||||
goto retblock;
|
||||
}
|
||||
@ -143,11 +138,9 @@ bool change_current_db(char* dest,
|
||||
else
|
||||
{
|
||||
/** Create error message */
|
||||
skygw_log_write_flush(LOGFILE_ERROR,
|
||||
"change_current_db: failed to change database: Query buffer too large");
|
||||
skygw_log_write_flush(LOGFILE_TRACE,
|
||||
"change_current_db: failed to change database: "
|
||||
"Query buffer too large [%ld bytes]", GWBUF_LENGTH(buf));
|
||||
MXS_ERROR("change_current_db: failed to change database: Query buffer too large");
|
||||
MXS_INFO("change_current_db: failed to change database: "
|
||||
"Query buffer too large [%ld bytes]", GWBUF_LENGTH(buf));
|
||||
succp = false;
|
||||
goto retblock;
|
||||
}
|
||||
|
||||
@ -343,7 +343,7 @@ parse_mapping_response(ROUTER_CLIENT_SES* rses, char* target, GWBUF* buf)
|
||||
{
|
||||
if(hashtable_add(rses->dbhash,data,target))
|
||||
{
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: <%s, %s>",target,data);
|
||||
MXS_INFO("shardrouter: <%s, %s>",target,data);
|
||||
}
|
||||
free(data);
|
||||
}
|
||||
@ -468,7 +468,7 @@ get_shard_target_name(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client, GWBUF*
|
||||
}
|
||||
else
|
||||
{
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: Query targets database '%s' on server '%s",dbnms[i],rval);
|
||||
MXS_INFO("shardrouter: Query targets database '%s' on server '%s",dbnms[i],rval);
|
||||
has_dbs = true;
|
||||
}
|
||||
}
|
||||
@ -487,15 +487,15 @@ get_shard_target_name(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client, GWBUF*
|
||||
ss_dassert(tok != NULL);
|
||||
tmp = (char*) hashtable_fetch(ht, tok);
|
||||
if(tmp)
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: SHOW TABLES with specific database '%s' on server '%s'", tok, tmp);
|
||||
MXS_INFO("shardrouter: SHOW TABLES with specific database '%s' on server '%s'", tok, tmp);
|
||||
}
|
||||
free(query);
|
||||
|
||||
if(tmp == NULL)
|
||||
{
|
||||
rval = (char*) hashtable_fetch(ht, client->rses_mysql_session->db);
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: SHOW TABLES query, current database '%s' on server '%s'",
|
||||
client->rses_mysql_session->db,rval);
|
||||
MXS_INFO("shardrouter: SHOW TABLES query, current database '%s' on server '%s'",
|
||||
client->rses_mysql_session->db,rval);
|
||||
}
|
||||
else
|
||||
{
|
||||
@ -516,7 +516,7 @@ get_shard_target_name(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client, GWBUF*
|
||||
if(strcmp(srvrf->server->unique_name,buffer->hint->data) == 0)
|
||||
{
|
||||
rval = srvrf->server->unique_name;
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: Routing hint found (%s)",rval);
|
||||
MXS_INFO("shardrouter: Routing hint found (%s)",rval);
|
||||
|
||||
}
|
||||
srvrf = srvrf->next;
|
||||
@ -535,7 +535,7 @@ get_shard_target_name(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client, GWBUF*
|
||||
rval = (char*) hashtable_fetch(ht, client->rses_mysql_session->db);
|
||||
if(rval)
|
||||
{
|
||||
skygw_log_write(LOGFILE_TRACE,"shardrouter: Using active database '%s'",client->rses_mysql_session->db);
|
||||
MXS_INFO("shardrouter: Using active database '%s'",client->rses_mysql_session->db);
|
||||
}
|
||||
}
|
||||
|
||||
@ -564,10 +564,8 @@ tokenize_string(char* str)
|
||||
if(tmp == NULL)
|
||||
{
|
||||
char errbuf[STRERROR_BUFLEN];
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : realloc returned NULL: %s.",
|
||||
strerror_r(errno, errbuf, sizeof(errbuf)))));
|
||||
MXS_ERROR("realloc returned NULL: %s.",
|
||||
strerror_r(errno, errbuf, sizeof(errbuf)));
|
||||
free(list);
|
||||
return NULL;
|
||||
}
|
||||
@ -634,9 +632,9 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
if(!logged)
|
||||
{
|
||||
/*
|
||||
skygw_log_write(LOGFILE_DEBUG,"schemarouter: Still waiting for reply to SHOW DATABASES from %s for session %p",
|
||||
bkrf[i].bref_backend->backend_server->unique_name,
|
||||
rses->rses_client_dcb->session);
|
||||
MXS_DEBUG("schemarouter: Still waiting for reply to SHOW DATABASES from %s for session %p",
|
||||
bkrf[i].bref_backend->backend_server->unique_name,
|
||||
rses->rses_client_dcb->session);
|
||||
*/
|
||||
logged = true;
|
||||
}
|
||||
@ -660,8 +658,8 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
if((target = hashtable_fetch(rses->dbhash,
|
||||
rses->connect_db)) == NULL)
|
||||
{
|
||||
skygw_log_write_flush(LOGFILE_TRACE,"schemarouter: Connecting to a non-existent database '%s'",
|
||||
rses->connect_db);
|
||||
MXS_INFO("schemarouter: Connecting to a non-existent database '%s'",
|
||||
rses->connect_db);
|
||||
rses->rses_closed = true;
|
||||
if(rses->queue)
|
||||
{
|
||||
@ -682,7 +680,7 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
buffer = gwbuf_alloc(qlen + 5);
|
||||
if(buffer == NULL)
|
||||
{
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Buffer allocation failed.");
|
||||
MXS_ERROR("Buffer allocation failed.");
|
||||
rses->rses_closed = true;
|
||||
if(rses->queue)
|
||||
gwbuf_free(rses->queue);
|
||||
@ -707,15 +705,14 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
rses->queue = rses->queue->next;
|
||||
tmp->next = NULL;
|
||||
char* querystr = modutil_get_SQL(tmp);
|
||||
skygw_log_write(LOGFILE_DEBUG,"schemarouter: Sending queued buffer for session %p: %s",
|
||||
rses->rses_client_dcb->session,
|
||||
querystr);
|
||||
MXS_DEBUG("schemarouter: Sending queued buffer for session %p: %s",
|
||||
rses->rses_client_dcb->session,
|
||||
querystr);
|
||||
poll_add_epollin_event_to_dcb(rses->routedcb,tmp);
|
||||
free(querystr);
|
||||
|
||||
}
|
||||
skygw_log_write_flush(LOGFILE_DEBUG,"session [%p] database map finished.",
|
||||
rses);
|
||||
MXS_DEBUG("session [%p] database map finished.", rses);
|
||||
}
|
||||
|
||||
goto retblock;
|
||||
@ -727,9 +724,9 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
rses->queue = rses->queue->next;
|
||||
tmp->next = NULL;
|
||||
char* querystr = modutil_get_SQL(tmp);
|
||||
skygw_log_write(LOGFILE_DEBUG,"schemarouter: Sending queued buffer for session %p: %s",
|
||||
rses->rses_client_dcb->session,
|
||||
querystr);
|
||||
MXS_DEBUG("schemarouter: Sending queued buffer for session %p: %s",
|
||||
rses->rses_client_dcb->session,
|
||||
querystr);
|
||||
poll_add_epollin_event_to_dcb(rses->routedcb,tmp);
|
||||
free(querystr);
|
||||
tmp = NULL;
|
||||
@ -737,9 +734,9 @@ filterReply(FILTER* instance, void *session, GWBUF *reply)
|
||||
|
||||
if(rses->init & INIT_USE_DB)
|
||||
{
|
||||
skygw_log_write(LOGFILE_DEBUG,"schemarouter: Reply to USE '%s' received for session %p",
|
||||
rses->connect_db,
|
||||
rses->rses_client_dcb->session);
|
||||
MXS_DEBUG("schemarouter: Reply to USE '%s' received for session %p",
|
||||
rses->connect_db,
|
||||
rses->rses_client_dcb->session);
|
||||
rses->init &= ~INIT_USE_DB;
|
||||
strcpy(rses->rses_mysql_session->db,rses->connect_db);
|
||||
ss_dassert(rses->init == INIT_READY);
|
||||
@ -831,9 +828,7 @@ version()
|
||||
void
|
||||
ModuleInit()
|
||||
{
|
||||
LOGIF(LM, (skygw_log_write_flush(
|
||||
LOGFILE_MESSAGE,
|
||||
"Initializing statemend-based read/write split router module.")));
|
||||
MXS_NOTICE("Initializing statemend-based read/write split router module.");
|
||||
spinlock_init(&instlock);
|
||||
instances = NULL;
|
||||
}
|
||||
@ -938,8 +933,8 @@ createInstance(SERVICE *service, char **options)
|
||||
|
||||
if(conf == NULL)
|
||||
{
|
||||
skygw_log_write(LOGFILE_ERROR, "Error : no 'subservices' confguration parameter found. "
|
||||
" Expected a list of service names.");
|
||||
MXS_ERROR("No 'subservices' confguration parameter found. "
|
||||
" Expected a list of service names.");
|
||||
free(router);
|
||||
return NULL;
|
||||
}
|
||||
@ -951,7 +946,7 @@ createInstance(SERVICE *service, char **options)
|
||||
{
|
||||
free(router);
|
||||
free(services);
|
||||
skygw_log_write(LOGFILE_ERROR,"Error: Memory allocation failed.");
|
||||
MXS_ERROR("Memory allocation failed.");
|
||||
return NULL;
|
||||
}
|
||||
|
||||
@ -966,10 +961,10 @@ createInstance(SERVICE *service, char **options)
|
||||
temp = realloc(res_svc, sizeof(SERVICE*)*(sz * 2));
|
||||
if(temp == NULL)
|
||||
{
|
||||
skygw_log_write(LOGFILE_ERROR, "Error : Memory reallocation failed.");
|
||||
LOGIF(LD,(skygw_log_write(LOGFILE_DEBUG, "shardrouter.c: realloc returned NULL. "
|
||||
"service count[%d] buffer size [%lu] tried to allocate [%lu]",
|
||||
sz, sizeof(SERVICE*) * (sz), sizeof(SERVICE*) * (sz * 2))));
|
||||
MXS_ERROR("Memory reallocation failed.");
|
||||
MXS_DEBUG("shardrouter.c: realloc returned NULL. "
|
||||
"service count[%d] buffer size [%lu] tried to allocate [%lu]",
|
||||
sz, sizeof(SERVICE*) * (sz), sizeof(SERVICE*) * (sz * 2));
|
||||
free(res_svc);
|
||||
free(router);
|
||||
return NULL;
|
||||
@ -983,7 +978,7 @@ createInstance(SERVICE *service, char **options)
|
||||
{
|
||||
free(res_svc);
|
||||
free(router);
|
||||
skygw_log_write(LOGFILE_ERROR, "Error : No service named '%s' found.", options[i]);
|
||||
MXS_ERROR("No service named '%s' found.", options[i]);
|
||||
return NULL;
|
||||
}
|
||||
i++;
|
||||
@ -998,8 +993,8 @@ createInstance(SERVICE *service, char **options)
|
||||
|
||||
if(i < min_nsvc)
|
||||
{
|
||||
skygw_log_write(LOGFILE_ERROR, "Error : Not enough parameters for 'subservice' router option. Shardrouter requires at least %d "
|
||||
"configured services to work.", min_nsvc);
|
||||
MXS_ERROR("Not enough parameters for 'subservice' router option. Shardrouter requires at least %d "
|
||||
"configured services to work.", min_nsvc);
|
||||
free(router->services);
|
||||
free(router);
|
||||
return NULL;
|
||||
@ -1107,7 +1102,7 @@ newSession(
|
||||
if(subsvc->scur == NULL)
|
||||
{
|
||||
subsvc_set_state(subsvc,SUBSVC_FAILED);
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Memory allocation failed in shardrouter.");
|
||||
MXS_ERROR("Memory allocation failed in shardrouter.");
|
||||
continue;
|
||||
}
|
||||
subsvc->scur->scmd_cur_rses = client_rses;
|
||||
@ -1117,7 +1112,7 @@ newSession(
|
||||
|
||||
if(subsvc->dcb == NULL){
|
||||
subsvc_set_state(subsvc,SUBSVC_FAILED);
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Failed to clone client DCB in shardrouter.");
|
||||
MXS_ERROR("Failed to clone client DCB in shardrouter.");
|
||||
continue;
|
||||
}
|
||||
|
||||
@ -1127,7 +1122,7 @@ newSession(
|
||||
dcb_close(subsvc->dcb);
|
||||
subsvc->dcb = NULL;
|
||||
subsvc_set_state(subsvc,SUBSVC_FAILED);
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Failed to create subsession for service %s in shardrouter.",subsvc->service->name);
|
||||
MXS_ERROR("Failed to create subsession for service %s in shardrouter.",subsvc->service->name);
|
||||
continue;
|
||||
}
|
||||
|
||||
@ -1136,7 +1131,7 @@ newSession(
|
||||
if(dummy_filterdef == NULL)
|
||||
{
|
||||
subsvc_set_state(subsvc,SUBSVC_FAILED);
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Failed to allocate filter definition in shardrouter.");
|
||||
MXS_ERROR("Failed to allocate filter definition in shardrouter.");
|
||||
continue;
|
||||
}
|
||||
dummy_filterdef->obj = &dummyObject;
|
||||
@ -1146,7 +1141,7 @@ newSession(
|
||||
if(dummy_upstream == NULL)
|
||||
{
|
||||
subsvc_set_state(subsvc,SUBSVC_FAILED);
|
||||
skygw_log_write_flush(LOGFILE_ERROR,"Error : Failed to set filterUpstream in shardrouter.");
|
||||
MXS_ERROR("Failed to set filterUpstream in shardrouter.");
|
||||
continue;
|
||||
}
|
||||
|
||||
@ -1217,9 +1212,7 @@ closeSession(
|
||||
{
|
||||
ROUTER_CLIENT_SES* router_cli_ses;
|
||||
int i;
|
||||
LOGIF(LD, (skygw_log_write(LOGFILE_DEBUG,
|
||||
"%lu [RWSplit:closeSession]",
|
||||
pthread_self())));
|
||||
MXS_DEBUG("%lu [RWSplit:closeSession]", pthread_self());
|
||||
|
||||
/**
|
||||
* router session can be NULL if newSession failed and it is discarding
|
||||
@ -1351,10 +1344,7 @@ get_shard_route_target(skygw_query_type_t qtype,
|
||||
target = TARGET_ANY;
|
||||
}
|
||||
#if defined(SS_DEBUG)
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Selected target \"%s\"",
|
||||
STRTARGET(target))));
|
||||
MXS_INFO("Selected target \"%s\"", STRTARGET(target));
|
||||
#endif
|
||||
return target;
|
||||
}
|
||||
@ -1551,7 +1541,7 @@ routeQuery(ROUTER* instance,
|
||||
char db[MYSQL_DATABASE_MAXLEN + 1];
|
||||
char errbuf[26+MYSQL_DATABASE_MAXLEN];
|
||||
|
||||
skygw_log_write_flush(LOGFILE_DEBUG,"shardrouter: routeQuery");
|
||||
MXS_DEBUG("shardrouter: routeQuery");
|
||||
CHK_CLIENT_RSES(router_cli_ses);
|
||||
|
||||
/** Dirty read for quick check if router is closed. */
|
||||
@ -1564,9 +1554,7 @@ routeQuery(ROUTER* instance,
|
||||
/** Lock router session */
|
||||
if(!rses_begin_locked_router_action(router_cli_ses))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Route query aborted! Routing session is closed <")));
|
||||
MXS_INFO("Route query aborted! Routing session is closed <");
|
||||
ret = 0;
|
||||
goto retblock;
|
||||
}
|
||||
@ -1583,9 +1571,9 @@ routeQuery(ROUTER* instance,
|
||||
{
|
||||
|
||||
char* querystr = modutil_get_SQL(querybuf);
|
||||
skygw_log_write(LOGFILE_DEBUG,"shardrouter: Storing query for session %p: %s",
|
||||
router_cli_ses->rses_client_dcb->session,
|
||||
querystr);
|
||||
MXS_DEBUG("shardrouter: Storing query for session %p: %s",
|
||||
router_cli_ses->rses_client_dcb->session,
|
||||
querystr);
|
||||
free(querystr);
|
||||
gwbuf_make_contiguous(querybuf);
|
||||
GWBUF* ptr = router_cli_ses->queue;
|
||||
@ -1625,14 +1613,11 @@ routeQuery(ROUTER* instance,
|
||||
{
|
||||
char* query_str = modutil_get_query(querybuf);
|
||||
|
||||
LOGIF(LE,
|
||||
(skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error: Can't route %s:%s:\"%s\" to "
|
||||
"backend server. Router is closed.",
|
||||
STRPACKETTYPE(packet_type),
|
||||
STRQTYPE(qtype),
|
||||
(query_str == NULL ? "(empty)" : query_str))));
|
||||
MXS_ERROR("Can't route %s:%s:\"%s\" to "
|
||||
"backend server. Router is closed.",
|
||||
STRPACKETTYPE(packet_type),
|
||||
STRQTYPE(qtype),
|
||||
(query_str == NULL ? "(empty)" : query_str));
|
||||
free(query_str);
|
||||
}
|
||||
ret = 0;
|
||||
@ -1699,9 +1684,7 @@ routeQuery(ROUTER* instance,
|
||||
extract_database(querybuf,db);
|
||||
snprintf(errbuf,25+MYSQL_DATABASE_MAXLEN,"Unknown database: %s",db);
|
||||
create_error_reply(errbuf,router_cli_ses->replydcb);
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Changing database failed.")));
|
||||
MXS_ERROR("Changing database failed.");
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
@ -1801,9 +1784,7 @@ routeQuery(ROUTER* instance,
|
||||
/** Lock router session */
|
||||
if(!rses_begin_locked_router_action(router_cli_ses))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Route query aborted! Routing session is closed <")));
|
||||
MXS_INFO("Route query aborted! Routing session is closed <");
|
||||
ret = 0;
|
||||
goto retblock;
|
||||
}
|
||||
@ -1847,12 +1828,10 @@ routeQuery(ROUTER* instance,
|
||||
|
||||
if(!succp)
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Was supposed to route to named server "
|
||||
"%s but couldn't find the server in a "
|
||||
"suitable state.",
|
||||
tname)));
|
||||
MXS_INFO("Was supposed to route to named server "
|
||||
"%s but couldn't find the server in a "
|
||||
"suitable state.",
|
||||
tname);
|
||||
}
|
||||
}
|
||||
|
||||
@ -1888,9 +1867,7 @@ routeQuery(ROUTER* instance,
|
||||
}
|
||||
else
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Routing query failed.")));
|
||||
MXS_ERROR("Routing query failed.");
|
||||
ret = 0;
|
||||
}
|
||||
}
|
||||
@ -2094,13 +2071,11 @@ rses_property_done(
|
||||
break;
|
||||
|
||||
default:
|
||||
LOGIF(LD, (skygw_log_write(
|
||||
LOGFILE_DEBUG,
|
||||
"%lu [rses_property_done] Unknown property type %d "
|
||||
"in property %p",
|
||||
pthread_self(),
|
||||
prop->rses_prop_type,
|
||||
prop)));
|
||||
MXS_DEBUG("%lu [rses_property_done] Unknown property type %d "
|
||||
"in property %p",
|
||||
pthread_self(),
|
||||
prop->rses_prop_type,
|
||||
prop);
|
||||
|
||||
ss_dassert(false);
|
||||
break;
|
||||
@ -2444,9 +2419,7 @@ execute_sescmd_in_backend(SUBSERVICE* subsvc)
|
||||
if(sescmd_cursor_get_command(scur) == NULL)
|
||||
{
|
||||
succp = false;
|
||||
LOGIF(LT, (skygw_log_write_flush(
|
||||
LOGFILE_TRACE,
|
||||
"Cursor had no pending session commands.")));
|
||||
MXS_INFO("Cursor had no pending session commands.");
|
||||
|
||||
goto return_succp;
|
||||
}
|
||||
@ -2601,9 +2574,7 @@ route_session_write(
|
||||
SUBSERVICE* subsvc;
|
||||
int i;
|
||||
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Session write, routing to all servers.")));
|
||||
MXS_INFO("Session write, routing to all servers.");
|
||||
|
||||
/**
|
||||
* These are one-way messages and server doesn't respond to them.
|
||||
@ -2632,12 +2603,10 @@ route_session_write(
|
||||
|
||||
if(LOG_IS_ENABLED(LOGFILE_TRACE))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Route query to %s%s%s",
|
||||
i == 0 ? ">":"",
|
||||
subsvc->service->name,
|
||||
i+1 >= router_cli_ses->n_subservice ? "<" : "")));
|
||||
MXS_INFO("Route query to %s%s%s",
|
||||
i == 0 ? ">":"",
|
||||
subsvc->service->name,
|
||||
i+1 >= router_cli_ses->n_subservice ? "<" : "");
|
||||
}
|
||||
|
||||
if(!SUBSVC_IS_CLOSED(subsvc) && SUBSVC_IS_OK(subsvc))
|
||||
@ -2687,12 +2656,10 @@ route_session_write(
|
||||
|
||||
if(LOG_IS_ENABLED(LOGFILE_TRACE))
|
||||
{
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Route query to %s%s%s",
|
||||
i == 0 ? ">":"",
|
||||
subsvc->service->name,
|
||||
i+1 >= router_cli_ses->n_subservice ? "<" : "")));
|
||||
MXS_INFO("Route query to %s%s%s",
|
||||
i == 0 ? ">":"",
|
||||
subsvc->service->name,
|
||||
i+1 >= router_cli_ses->n_subservice ? "<" : "");
|
||||
}
|
||||
|
||||
|
||||
@ -2715,10 +2682,8 @@ route_session_write(
|
||||
{
|
||||
succp = true;
|
||||
|
||||
LOGIF(LT, (skygw_log_write(
|
||||
LOGFILE_TRACE,
|
||||
"Service %s already executing sescmd.",
|
||||
subsvc->service->name)));
|
||||
MXS_INFO("Service %s already executing sescmd.",
|
||||
subsvc->service->name);
|
||||
}
|
||||
else
|
||||
{
|
||||
@ -2726,11 +2691,9 @@ route_session_write(
|
||||
|
||||
if(!succp)
|
||||
{
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"Error : Failed to execute session "
|
||||
"command in %s",
|
||||
subsvc->service->name)));
|
||||
MXS_ERROR("Failed to execute session "
|
||||
"command in %s",
|
||||
subsvc->service->name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user