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

Merge branch 'develop' into firewall

Browse Source
This commit is contained in:
Markus Makela 2014-11-11 14:18:44 +02:00
commit 4a8f97d2e2
28 changed files with 663 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -152,7 +152,7 @@ add_custom_target(buildtests
add_custom_target(testall
COMMAND ${CMAKE_COMMAND} -DDEPS_OK=Y -DBUILD_TESTS=Y -DBUILD_TYPE=Debug -DINSTALL_DIR=${CMAKE_BINARY_DIR} -DINSTALL_SYSTEM_FILES=N ${CMAKE_SOURCE_DIR}
COMMAND make install
COMMAND cp ${CMAKE_SOURCE_DIR}/server/test/MaxScale_test.cnf ${CMAKE_BINARY_DIR}/etc/MaxScale.cnf
COMMAND ${CMAKE_COMMAND} -E copy_if_different ${CMAKE_SOURCE_DIR}/server/test/MaxScale_test.cnf ${CMAKE_BINARY_DIR}/etc/MaxScale.cnf
COMMAND /bin/sh -c "${CMAKE_BINARY_DIR}/bin/maxscale -c ${CMAKE_BINARY_DIR} &>/dev/null"
COMMAND /bin/sh -c "make test || echo \"Test results written to: ${CMAKE_BINARY_DIR}/Testing/Temporary/\""
COMMAND killall maxscale

77
log_manager/log_manager.cc
View File

@ -276,6 +276,8 @@ static bool check_file_and_path(
bool* writable);
static bool file_is_symlink(char* filename);
static int skygw_log_disable_raw(logfile_id_t id, bool emergency); /*< no locking */
const char* get_suffix_default(void)
{
@ -1185,25 +1187,36 @@ return_err:
return err;
}
int skygw_log_disable(
logfile_id_t id)
logfile_id_t id) /*< no locking */
{
int rc;
rc = skygw_log_disable_raw(id, false);
return rc;
}
static int skygw_log_disable_raw(
logfile_id_t id,
bool emergency) /*< no locking */
{
bool err = 0;
if (!logmanager_register(true)) {
//fprintf(stderr, "ERROR: Can't register to logmanager\n");
if (!logmanager_register(true))
{
err = -1;
goto return_err;
}
CHK_LOGMANAGER(lm);
if (logfile_set_enabled(id, false)) {
lm->lm_enabled_logfiles &= ~id;
/**
* Set global variable
*/
lm_enabled_logfiles_bitmask = lm->lm_enabled_logfiles;
if (emergency || logfile_set_enabled(id, false))
{
lm->lm_enabled_logfiles &= ~id;
/**
* Set global variable
*/
lm_enabled_logfiles_bitmask = lm->lm_enabled_logfiles;
}
logmanager_unregister();
@ -2403,6 +2416,7 @@ static bool filewriter_init(
skygw_message_t* logmes)
{
bool succp = false;
int err;
logfile_t* lf;
logfile_id_t id;
int i;
@ -2456,10 +2470,22 @@ static bool filewriter_init(
} else {
start_msg_str = strdup("---\tLogging is disabled.\n");
}
skygw_file_write(fw->fwr_file[id],
err = skygw_file_write(fw->fwr_file[id],
(void *)start_msg_str,
strlen(start_msg_str),
true);
if (err != 0)
{
fprintf(stderr,
"Error : writing to file %s failed due to %d, %s. "
"Exiting MaxScale.\n",
lf->lf_full_file_name,
err,
strerror(err));
succp = false;
goto return_succp;
}
free(start_msg_str);
}
fw->fwr_state = RUN;
@ -2603,7 +2629,10 @@ static void* thr_filewriter_fun(
#endif
node = bb_list->mlist_first;
while (node != NULL) {
while (node != NULL)
{
int err = 0;
CHK_MLIST_NODE(node);
bb = (blockbuf_t *)node->mlnode_data;
CHK_BLOCKBUF(bb);
@ -2630,11 +2659,25 @@ static void* thr_filewriter_fun(
true);
}
skygw_file_write(file,
(void *)bb->bb_buf,
bb->bb_buf_used,
(flush_logfile ||
flushall_logfiles));
err = skygw_file_write(
file,
(void *)bb->bb_buf,
bb->bb_buf_used,
(flush_logfile ||
flushall_logfiles));
if (err)
{
fprintf(stderr,
"Error : Write to %s log "
": %s failed due to %d, "
"%s. Disabling the log.",
STRLOGNAME((logfile_id_t)i),
lf->lf_full_file_name,
err,
strerror(err));
/** Force log off */
skygw_log_disable_raw((logfile_id_t)i, true);
}
/**
* Reset buffer's counters and mark
* not full.

2
query_classifier/test/canonical_tests/canonizer.c
View File

@ -57,7 +57,7 @@ int main(int argc, char** argv)
{
fgets(readbuff,4092,infile);
psize = strlen(readbuff);
if(psize < 0 || psize > 4092){
if(psize > 4092){
continue;
}
qbuff = gwbuf_alloc(psize + 7);

6
server/core/config.c
View File

@ -1393,11 +1393,11 @@ SERVER *server;
user,
auth);
if (enable_root_user)
serviceEnableRootUser(service, atoi(enable_root_user));
serviceEnableRootUser(obj->element, atoi(enable_root_user));
if (allow_localhost_match_wildcard_host && service)
if (allow_localhost_match_wildcard_host)
serviceEnableLocalhostMatchWildcardHost(
service,
obj->element,
atoi(allow_localhost_match_wildcard_host));
}
}

10
server/core/dcb.c
View File

@ -121,8 +121,8 @@ DCB *rval;
#if defined(SS_DEBUG)
rval->dcb_chk_top = CHK_NUM_DCB;
rval->dcb_chk_tail = CHK_NUM_DCB;
rval->dcb_errhandle_called = false;
#endif
rval->dcb_errhandle_called = false;
rval->dcb_role = role;
spinlock_init(&rval->dcb_initlock);
spinlock_init(&rval->writeqlock);
@ -862,7 +862,7 @@ int below_water;
while (queue != NULL)
{
int qlen;
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
if (dcb->dcb_role == DCB_ROLE_REQUEST_HANDLER &&
dcb->session != NULL)
{
@ -878,7 +878,7 @@ int below_water;
fail_next_backend_fd = false;
}
}
#endif /* SS_DEBUG */
#endif /* FAKE_CODE */
qlen = GWBUF_LENGTH(queue);
GW_NOINTR_CALL(
w = gw_write(
@ -1684,7 +1684,7 @@ int gw_write(
size_t nbytes)
{
int w;
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
if (dcb_fake_write_errno[fd] != 0) {
ss_dassert(dcb_fake_write_ev[fd] != 0);
w = write(fd, buf, nbytes/2); /*< leave peer to read missing bytes */
@ -1698,7 +1698,7 @@ int gw_write(
}
#else
w = write(fd, buf, nbytes);
#endif /* SS_DEBUG && SS_TEST */
#endif /* FAKE_CODE */
#if defined(SS_DEBUG_MYSQL)
{

4
server/core/gateway.c
View File

@ -1025,7 +1025,7 @@ int main(int argc, char **argv)
progname = *argv;
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
memset(conn_open, 0, sizeof(bool)*10240);
memset(dcb_fake_write_errno, 0, sizeof(unsigned char)*10240);
memset(dcb_fake_write_ev, 0, sizeof(__int32_t)*10240);
@ -1033,7 +1033,7 @@ int main(int argc, char **argv)
fail_next_client_fd = false;
fail_next_accept = 0;
fail_accept_errno = 0;
#endif
#endif /* FAKE_CODE */
file_write_header(stderr);
/*<
* Register functions which are called at exit except libmysqld-related,

8
server/core/poll.c
View File

@ -625,7 +625,7 @@ uint32_t ev;
thread_data[thread_id].event = ev;
}
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
if (dcb_fake_write_ev[dcb->fd] != 0) {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -637,7 +637,7 @@ uint32_t ev;
ev |= dcb_fake_write_ev[dcb->fd];
dcb_fake_write_ev[dcb->fd] = 0;
}
#endif
#endif /* FAKE_CODE */
ss_debug(spinlock_acquire(&dcb->dcb_initlock);)
ss_dassert(dcb->state != DCB_STATE_ALLOC);
ss_dassert(dcb->state != DCB_STATE_DISCONNECTED);
@ -735,7 +735,7 @@ uint32_t ev;
if (ev & EPOLLERR)
{
int eno = gw_getsockerrno(dcb->fd);
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
if (eno == 0) {
eno = dcb_fake_write_errno[dcb->fd];
LOGIF(LD, (skygw_log_write(
@ -748,7 +748,7 @@ uint32_t ev;
strerror(eno))));
}
dcb_fake_write_errno[dcb->fd] = 0;
#endif
#endif /* FAKE_CODE */
if (eno != 0) {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,

67
server/core/service.c
View File

@ -145,6 +145,7 @@ SERVICE *service;
service->filters = NULL;
service->n_filters = 0;
service->weightby = 0;
service->users = NULL;
service->resources = NULL;
spinlock_init(&service->spin);
spinlock_init(&service->users_table_spin);
@ -213,42 +214,46 @@ GWPROTOCOL *funcs;
if (strcmp(port->protocol, "MySQLClient") == 0) {
int loaded;
/*
* Allocate specific data for MySQL users
* including hosts and db names
*/
service->users = mysql_users_alloc();
if (service->users == NULL) {
/*
* Allocate specific data for MySQL users
* including hosts and db names
*/
service->users = mysql_users_alloc();
if ((loaded = load_mysql_users(service)) < 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load users from %s:%d for "
"service %s.",
port->address,
port->port,
service->name)));
hashtable_free(service->users->data);
free(service->users);
dcb_free(port->listener);
port->listener = NULL;
goto retblock;
if ((loaded = load_mysql_users(service)) < 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load users from %s:%d for "
"service %s.",
port->address,
port->port,
service->name)));
hashtable_free(service->users->data);
free(service->users);
dcb_free(port->listener);
port->listener = NULL;
goto retblock;
}
/* At service start last update is set to USERS_REFRESH_TIME seconds earlier.
* This way MaxScale could try reloading users' just after startup
*/
service->rate_limit.last=time(NULL) - USERS_REFRESH_TIME;
service->rate_limit.nloads=1;
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Loaded %d MySQL Users for service [%s].",
loaded, service->name)));
}
/* At service start last update is set to USERS_REFRESH_TIME seconds earlier.
* This way MaxScale could try reloading users' just after startup
*/
service->rate_limit.last=time(NULL) - USERS_REFRESH_TIME;
service->rate_limit.nloads=1;
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Loaded %d MySQL Users for service [%s].",
loaded, service->name)));
}
else
{
/* Generic users table */
service->users = users_alloc();
if (service->users == NULL) {
/* Generic users table */
service->users = users_alloc();
}
}
if ((funcs=(GWPROTOCOL *)load_module(port->protocol, MODULE_PROTOCOL))

16
server/core/session.c
View File

@ -134,7 +134,8 @@ session_alloc(SERVICE *service, DCB *client_dcb)
service->router->newSession(service->router_instance,
session);
if (session->router_session == NULL) {
if (session->router_session == NULL)
{
/**
* Inform other threads that session is closing.
*/
@ -153,7 +154,6 @@ session_alloc(SERVICE *service, DCB *client_dcb)
goto return_session;
}
/*
* Pending filter chain being setup set the head of the chain to
* be the router. As filters are inserted the current head will
@ -196,11 +196,12 @@ session_alloc(SERVICE *service, DCB *client_dcb)
}
}
spinlock_acquire(&session_spin);
spinlock_acquire(&session->ses_lock);
if (session->state != SESSION_STATE_READY)
{
session_free(session);
spinlock_release(&session->ses_lock);
session_free(session);
client_dcb->session = NULL;
session = NULL;
LOGIF(LE, (skygw_log_write_flush(
@ -212,10 +213,13 @@ session_alloc(SERVICE *service, DCB *client_dcb)
else
{
session->state = SESSION_STATE_ROUTER_READY;
session->next = allSessions;
spinlock_release(&session->ses_lock);
spinlock_acquire(&session_spin);
session->next = allSessions;
allSessions = session;
spinlock_release(&session_spin);
atomic_add(&service->stats.n_sessions, 1);
atomic_add(&service->stats.n_sessions, 1);
atomic_add(&service->stats.n_current, 1);
CHK_SESSION(session);
}

8
server/include/dcb.h
View File

@ -205,9 +205,9 @@ typedef struct dcb_callback {
typedef struct dcb {
#if defined(SS_DEBUG)
skygw_chk_t dcb_chk_top;
bool dcb_errhandle_called;
#endif
dcb_role_t dcb_role;
bool dcb_errhandle_called; /*< this can be called only once */
dcb_role_t dcb_role;
SPINLOCK dcb_initlock;
DCBEVENTQ evq; /**< The event queue for this DCB */
int fd; /**< The descriptor */
@ -252,14 +252,14 @@ typedef struct dcb {
#endif
} DCB;
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
unsigned char dcb_fake_write_errno[10240];
__int32_t dcb_fake_write_ev[10240];
bool fail_next_backend_fd;
bool fail_next_client_fd;
int fail_next_accept;
int fail_accept_errno;
#endif
#endif /* FAKE_CODE */
/* A few useful macros */
#define DCB_SESSION(x) (x)->session

4
server/include/monitor.h
View File

@ -110,6 +110,10 @@ typedef enum
MONITOR_WRITE_TIMEOUT = 2
} monitor_timeouts_t;
#define DEFAULT_CONNECT_TIMEOUT 3
#define DEFAULT_READ_TIMEOUT 1
#define DEFAULT_WRITE_TIMEOUT 2
/**
* Representation of the running monitor.
*/

2
server/modules/filter/hint/hintparser.c
View File

@ -550,7 +550,7 @@ HINT_TOKEN *tok;
else if (!inword && inquote == '\0' && **ptr == '=')
{
*dest = **ptr;
*dest++;
dest++;
(*ptr)++;
break;
}

61
server/modules/filter/test/harness_common.c
View File

@ -748,9 +748,12 @@ int load_filter(FILTERCHAIN* fc, CONFIG* cnf)
}
int x;
for(x = 0;x<paramc;x++){
free(fparams[x]->name);
free(fparams[x]->value);
if(fparams){
for(x = 0;x<paramc;x++){
free(fparams[x]->name);
free(fparams[x]->value);
}
}
free(fparams);
@ -769,15 +772,18 @@ FILTERCHAIN* load_filter_module(char* str)
flt_ptr->next = instance.head;
}
if((flt_ptr->instance = (FILTER_OBJECT*)load_module(str, MODULE_FILTER)) == NULL)
{
printf("Error: Module loading failed: %s\n",str);
skygw_log_write(LOGFILE_ERROR,"Error: Module loading failed: %s\n",str);
free(flt_ptr->down);
free(flt_ptr);
return NULL;
}
flt_ptr->name = strdup(str);
if(flt_ptr){
if( (flt_ptr->instance = (FILTER_OBJECT*)load_module(str, MODULE_FILTER)) == NULL)
{
printf("Error: Module loading failed: %s\n",str);
skygw_log_write(LOGFILE_ERROR,"Error: Module loading failed: %s\n",str);
free(flt_ptr->down);
free(flt_ptr);
return NULL;
}
flt_ptr->name = strdup(str);
}
return flt_ptr;
}
@ -925,14 +931,35 @@ GWBUF* gen_packet(PACKET pkt)
int process_opts(int argc, char** argv)
{
unsigned int fd = open_file("harness.cnf",1), buffsize = 1024;
int rd,rdsz;
unsigned int fsize;
int fd, buffsize = 1024;
int rd,rdsz, rval;
size_t fsize;
char *buff = calloc(buffsize,sizeof(char)), *tok = NULL;
/**Parse 'harness.cnf' file*/
fsize = lseek(fd,0,SEEK_END);
lseek(fd,0,SEEK_SET);
if(buff == NULL){
printf("Error: Call to malloc() failed.\n");
return 1;
}
if((fd = open_file("harness.cnf",1)) < 0){
printf("Failed to open configuration file.\n");
free(buff);
return 1;
}
if( (rval = lseek(fd,0,SEEK_END)) < 0 ||
lseek(fd,0,SEEK_SET) < 0){
printf("Error: Cannot seek file.\n");
close(fd);
free(buff);
return 1;
}
fsize = (size_t)rval;
instance.thrcount = 1;
instance.session_count = 1;
rdsz = read(fd,buff,fsize);

1
server/modules/include/mysql_client_server_protocol.h
View File

@ -281,6 +281,7 @@ typedef struct {
* created or received */
unsigned long tid; /*< MySQL Thread ID, in
* handshake */
unsigned int charset; /*< MySQL character set at connect time */
#if defined(SS_DEBUG)
skygw_chk_t protocol_chk_tail;
#endif

86
server/modules/monitor/galera_mon.c
View File

@ -31,6 +31,7 @@
* 03/06/14 Mark Riddoch Add support for maintenance mode
* 24/06/14 Massimiliano Pinto Added depth level 0 for each node
* 30/10/14 Massimiliano Pinto Added disableMasterFailback feature
* 10/11/14 Massimiliano Pinto Added setNetworkTimeout for connect,read,write
*
* @endverbatim
*/
@ -53,7 +54,7 @@ extern int lm_enabled_logfiles_bitmask;
static void monitorMain(void *);
static char *version_str = "V1.3.0";
static char *version_str = "V1.4.0";
MODULE_INFO info = {
MODULE_API_MONITOR,
@ -69,9 +70,10 @@ static void unregisterServer(void *, SERVER *);
static void defaultUsers(void *, char *, char *);
static void diagnostics(DCB *, void *);
static void setInterval(void *, size_t);
static MONITOR_SERVERS *get_candidate_master(MONITOR_SERVERS *);
static MONITOR_SERVERS *set_cluster_master(MONITOR_SERVERS *, MONITOR_SERVERS *, int);
static MONITOR_SERVERS *get_candidate_master(MONITOR_SERVERS *);
static MONITOR_SERVERS *set_cluster_master(MONITOR_SERVERS *, MONITOR_SERVERS *, int);
static void disableMasterFailback(void *, int);
static void setNetworkTimeout(void *arg, int type, int value);
static MONITOR_OBJECT MyObject = {
startMonitor,
@ -81,7 +83,7 @@ static MONITOR_OBJECT MyObject = {
defaultUsers,
diagnostics,
setInterval,
NULL,
setNetworkTimeout,
NULL,
NULL,
NULL,
@ -155,6 +157,9 @@ MYSQL_MONITOR *handle;
handle->interval = MONITOR_INTERVAL;
handle->disableMasterFailback = 0;
handle->master = NULL;
handle->connect_timeout=DEFAULT_CONNECT_TIMEOUT;
handle->read_timeout=DEFAULT_READ_TIMEOUT;
handle->write_timeout=DEFAULT_READ_TIMEOUT;
spinlock_init(&handle->lock);
}
handle->tid = (THREAD)thread_start(monitorMain, handle);
@ -273,6 +278,9 @@ char *sep;
dcb_printf(dcb,"\tSampling interval:\t%lu milliseconds\n", handle->interval);
dcb_printf(dcb,"\tMaster Failback:\t%s\n", (handle->disableMasterFailback == 1) ? "off" : "on");
dcb_printf(dcb,"\tConnect Timeout:\t%i seconds\n", handle->connect_timeout);
dcb_printf(dcb,"\tRead Timeout:\t\t%i seconds\n", handle->read_timeout);
dcb_printf(dcb,"\tWrite Timeout:\t\t%i seconds\n", handle->write_timeout);
dcb_printf(dcb, "\tMonitored servers: ");
db = handle->databases;
@ -310,16 +318,18 @@ MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
/**
* Monitor an individual server
*
* @param database The database to probe
* @param handle The MySQL Monitor object
* @param database The database to probe
*/
static void
monitorDatabase(MONITOR_SERVERS *database, char *defaultUser, char *defaultPasswd)
monitorDatabase(MYSQL_MONITOR *handle, MONITOR_SERVERS *database)
{
MYSQL_ROW row;
MYSQL_RES *result;
int num_fields;
int isjoined = 0;
char *uname = defaultUser, *passwd = defaultPasswd;
char *uname = handle->defaultUser;
char *passwd = handle->defaultPasswd;
unsigned long int server_version = 0;
char *server_string;
@ -339,12 +349,15 @@ char *server_string;
{
char *dpwd = decryptPassword(passwd);
int rc;
int read_timeout = 1;
int connect_timeout = 2;
int connect_timeout = handle->connect_timeout;
int read_timeout = handle->read_timeout;
int write_timeout = handle->write_timeout;;
database->con = mysql_init(NULL);
rc = mysql_options(database->con, MYSQL_OPT_CONNECT_TIMEOUT, (void *)&connect_timeout);
rc = mysql_options(database->con, MYSQL_OPT_READ_TIMEOUT, (void *)&read_timeout);
rc = mysql_options(database->con, MYSQL_OPT_WRITE_TIMEOUT, (void *)&write_timeout);
if (mysql_real_connect(database->con, database->server->name,
uname, dpwd, NULL, database->server->port, NULL, 0) == NULL)
@ -487,7 +500,7 @@ int master_stickiness = handle->disableMasterFailback;
{
unsigned int prev_status = ptr->server->status;
monitorDatabase(ptr, handle->defaultUser, handle->defaultPasswd);
monitorDatabase(handle, ptr);
/* clear bits for non member nodes */
if ( ! SERVER_IN_MAINT(ptr->server) && (ptr->server->node_id < 0 || ! SERVER_IS_JOINED(ptr->server))) {
@ -662,3 +675,56 @@ MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->disableMasterFailback, &disable, sizeof(int));
}
static void
setNetworkTimeout(void *arg, int type, int value)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
int max_timeout = (int)(handle->interval/1000);
int new_timeout = max_timeout -1;
if (new_timeout <= 0)
new_timeout = DEFAULT_CONNECT_TIMEOUT;
switch(type) {
case MONITOR_CONNECT_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->connect_timeout, &value, sizeof(int));
} else {
memcpy(&handle->connect_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Connect Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
case MONITOR_READ_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->read_timeout, &value, sizeof(int));
} else {
memcpy(&handle->read_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Read Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
case MONITOR_WRITE_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->write_timeout, &value, sizeof(int));
} else {
memcpy(&handle->write_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Write Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
default:
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Monitor setNetworkTimeout received an unsupported action type %i", type)));
break;
}
}

85
server/modules/monitor/mysql_mon.c
View File

@ -43,6 +43,7 @@
* 28/08/14 Massimiliano Pinto Added detectStaleMaster feature: previous detected master will be used again, even if the replication is stopped.
* This means both IO and SQL threads are not working on slaves.
* This option is not enabled by default.
* 10/11/14 Massimiliano Pinto Addition of setNetworkTimeout for connect, read, write
*
* @endverbatim
*/
@ -65,7 +66,7 @@ extern int lm_enabled_logfiles_bitmask;
static void monitorMain(void *);
static char *version_str = "V1.3.0";
static char *version_str = "V1.4.0";
MODULE_INFO info = {
MODULE_API_MONITOR,
@ -104,7 +105,7 @@ static MONITOR_OBJECT MyObject = {
defaultUser,
diagnostics,
setInterval,
NULL,
setNetworkTimeout,
defaultId,
replicationHeartbeat,
detectStaleMaster,
@ -180,6 +181,9 @@ MYSQL_MONITOR *handle;
handle->replicationHeartbeat = 0;
handle->detectStaleMaster = 0;
handle->master = NULL;
handle->connect_timeout=DEFAULT_CONNECT_TIMEOUT;
handle->read_timeout=DEFAULT_READ_TIMEOUT;
handle->write_timeout=DEFAULT_WRITE_TIMEOUT;
spinlock_init(&handle->lock);
}
handle->tid = (THREAD)thread_start(monitorMain, handle);
@ -326,6 +330,9 @@ char *sep;
dcb_printf(dcb,"\tMaxScale MonitorId:\t%lu\n", handle->id);
dcb_printf(dcb,"\tReplication lag:\t%s\n", (handle->replicationHeartbeat == 1) ? "enabled" : "disabled");
dcb_printf(dcb,"\tDetect Stale Master:\t%s\n", (handle->detectStaleMaster == 1) ? "enabled" : "disabled");
dcb_printf(dcb,"\tConnect Timeout:\t%i seconds\n", handle->connect_timeout);
dcb_printf(dcb,"\tRead Timeout:\t\t%i seconds\n", handle->read_timeout);
dcb_printf(dcb,"\tWrite Timeout:\t\t%i seconds\n", handle->write_timeout);
dcb_printf(dcb, "\tMonitored servers: ");
db = handle->databases;
@ -381,11 +388,15 @@ char *server_string;
{
char *dpwd = decryptPassword(passwd);
int rc;
int read_timeout = 1;
int connect_timeout = handle->connect_timeout;
int read_timeout = handle->read_timeout;
int write_timeout = handle->write_timeout;
database->con = mysql_init(NULL);
rc = mysql_options(database->con, MYSQL_OPT_CONNECT_TIMEOUT, (void *)&connect_timeout);
rc = mysql_options(database->con, MYSQL_OPT_READ_TIMEOUT, (void *)&read_timeout);
rc = mysql_options(database->con, MYSQL_OPT_WRITE_TIMEOUT, (void *)&write_timeout);
if (mysql_real_connect(database->con,
database->server->name,
@ -867,6 +878,13 @@ static void set_master_heartbeat(MYSQL_MONITOR *handle, MONITOR_SERVERS *databas
char heartbeat_insert_query[512]="";
char heartbeat_purge_query[512]="";
if (handle->master == NULL) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: set_master_heartbeat called without an available Master server")));
return;
}
/* create the maxscale_schema database */
if (mysql_query(database->con, "CREATE DATABASE IF NOT EXISTS maxscale_schema")) {
LOGIF(LE, (skygw_log_write_flush(
@ -972,6 +990,13 @@ static void set_slave_heartbeat(MYSQL_MONITOR *handle, MONITOR_SERVERS *database
MYSQL_RES *result;
int num_fields;
if (handle->master == NULL) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: set_slave_heartbeat called without an available Master server")));
return;
}
/* Get the master_timestamp value from maxscale_schema.replication_heartbeat table */
sprintf(select_heartbeat_query, "SELECT master_timestamp "
@ -1210,11 +1235,61 @@ monitor_clear_pending_status(MONITOR_SERVERS *ptr, int bit)
/**
* Set the default id to use in the monitor.
*
* @param arg The handle allocated by startMonitor
* @param id The id to set in monitor struct
* @param arg The handle allocated by startMonitor
* @param type The connect timeout type
* @param value The timeout value to set
*/
static void
setNetworkTimeout(void *arg, int type, int value)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
int max_timeout = (int)(handle->interval/1000);
int new_timeout = max_timeout -1;
if (new_timeout <= 0)
new_timeout = DEFAULT_CONNECT_TIMEOUT;
switch(type) {
case MONITOR_CONNECT_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->connect_timeout, &value, sizeof(int));
} else {
memcpy(&handle->connect_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Connect Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
case MONITOR_READ_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->read_timeout, &value, sizeof(int));
} else {
memcpy(&handle->read_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Read Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
case MONITOR_WRITE_TIMEOUT:
if (value < max_timeout) {
memcpy(&handle->write_timeout, &value, sizeof(int));
} else {
memcpy(&handle->write_timeout, &new_timeout, sizeof(int));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"warning : Monitor Write Timeout %i is greater than monitor interval ~%i seconds"
", lowering to %i seconds", value, max_timeout, new_timeout)));
}
break;
default:
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Monitor setNetworkTimeout received an unsupported action type %i", type)));
break;
}
}

9
server/modules/monitor/mysqlmon.h
View File

@ -33,6 +33,8 @@
* 28/05/14 Massimiliano Pinto Addition of new fields in MYSQL_MONITOR struct
* 24/06/14 Massimiliano Pinto Addition of master field in MYSQL_MONITOR struct and MONITOR_MAX_NUM_SLAVES
* 28/08/14 Massimiliano Pinto Addition of detectStaleMaster
* 30/10/14 Massimiliano Pinto Addition of disableMasterFailback
* 07/11/14 Massimiliano Pinto Addition of NetworkTimeout: connect, read, write
*
* @endverbatim
*/
@ -68,6 +70,13 @@ typedef struct {
int disableMasterFailback; /**< Monitor flag for Galera Cluster Master failback */
MONITOR_SERVERS *master; /**< Master server for MySQL Master/Slave replication */
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
int connect_timeout; /**< Connect timeout in seconds for mysql_real_connect */
int read_timeout; /**< Timeout in seconds to read from the server.
* There are retries and the total effective timeout value is three times the option value.
*/
int write_timeout; /**< Timeout in seconds for each attempt to write to the server.
* There are retries and the total effective timeout value is two times the option value.
*/
} MYSQL_MONITOR;
#define MONITOR_RUNNING 1

27
server/modules/protocol/mysql_backend.c
View File

@ -44,6 +44,7 @@
* 12/09/2013 Massimiliano Pinto Added checks in gw_read_backend_event() for gw_read_backend_handshake
* 27/09/2013 Massimiliano Pinto Changed in gw_read_backend_event the check for dcb_read(), now is if rc < 0
* 24/10/2014 Massimiliano Pinto Added Mysql user@host @db authentication support
* 10/11/2014 Massimiliano Pinto Client charset is passed to backend
*
*/
#include <modinfo.h>
@ -378,7 +379,6 @@ static int gw_read_backend_event(DCB *dcb) {
ERRACT_REPLY_CLIENT,
&succp);
gwbuf_free(errbuf);
ss_dassert(!succp);
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_event] "
@ -854,8 +854,12 @@ static int gw_error_backend_event(DCB *dcb)
&succp);
gwbuf_free(errbuf);
/** There are not required backends available, close session. */
if (!succp) {
/**
* If error handler fails it means that routing session can't continue
* and it must be closed. In success, only this DCB is closed.
*/
if (!succp)
{
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
@ -909,6 +913,9 @@ static int gw_create_backend_connection(
/** Copy client flags to backend protocol */
protocol->client_capabilities =
((MySQLProtocol *)(backend_dcb->session->client->protocol))->client_capabilities;
/** Copy client charset to backend protocol */
protocol->charset =
((MySQLProtocol *)(backend_dcb->session->client->protocol))->charset;
/*< if succeed, fd > 0, -1 otherwise */
rv = gw_do_connect_to_backend(server->name, server->port, &fd);
@ -1082,8 +1089,8 @@ gw_backend_close(DCB *dcb)
mysql_protocol_done(dcb);
/**
* If session->state is set to STOPPING the client and the session must
* be closed too.
* If session->state is STOPPING, start closing client session.
* Otherwise only this backend connection is closed.
*/
if (session != NULL && session->state == SESSION_STATE_STOPPING)
{
@ -1258,6 +1265,16 @@ static int gw_change_user(
/* get new database name */
strcpy(database, (char *)client_auth_packet);
/* get character set */
if (strlen(database)) {
client_auth_packet += strlen(database) + 1;
} else {
client_auth_packet++;
}
if (client_auth_packet && *client_auth_packet)
memcpy(&backend_protocol->charset, client_auth_packet, sizeof(int));
/* save current_database name */
strcpy(current_database, current_session->db);

38
server/modules/protocol/mysql_client.c
View File

@ -36,6 +36,7 @@
* 07/05/2014 Massimiliano Pinto Added: specific version string in server handshake
* 09/09/2014 Massimiliano Pinto Added: 777 permission for socket path
* 13/10/2014 Massimiliano Pinto Added: dbname authentication check
* 10/11/2014 Massimiliano Pinto Added: client charset added to protocol struct
*
*/
#include <skygw_utils.h>
@ -444,6 +445,9 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
return 1;
}
/* get charset */
memcpy(&protocol->charset, client_auth_packet + 4 + 4 + 4, sizeof (int));
/* get the auth token len */
memcpy(&auth_token_len,
client_auth_packet + 4 + 4 + 4 + 1 + 23 + strlen(username) + 1,
@ -836,9 +840,6 @@ int gw_read_client_event(
}
else
{
GWBUF* errbuf;
bool succp;
modutil_send_mysql_err_packet(dcb,
1,
0,
@ -850,7 +851,6 @@ int gw_read_client_event(
"Error : Routing the query failed. "
"Session will be closed.")));
dcb_close(dcb);
}
}
@ -1109,7 +1109,7 @@ int gw_MySQLAccept(DCB *listener)
retry_accept:
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
if (fail_next_accept > 0)
{
c_sock = -1;
@ -1117,16 +1117,16 @@ int gw_MySQLAccept(DCB *listener)
fail_next_accept -= 1;
} else {
fail_accept_errno = 0;
#endif /* SS_DEBUG */
#endif /* FAKE_CODE */
// new connection from client
c_sock = accept(listener->fd,
(struct sockaddr *) &client_conn,
&client_len);
eno = errno;
errno = 0;
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
}
#endif /* SS_DEBUG */
#endif /* FAKE_CODE */
if (c_sock == -1) {
@ -1363,7 +1363,6 @@ gw_client_close(DCB *dcb)
SESSION* session;
ROUTER_OBJECT* router;
void* router_instance;
void* rsession;
#if defined(SS_DEBUG)
MySQLProtocol* protocol = (MySQLProtocol *)dcb->protocol;
if (dcb->state == DCB_STATE_POLLING ||
@ -1389,13 +1388,22 @@ gw_client_close(DCB *dcb)
{
session->state = SESSION_STATE_STOPPING;
}
spinlock_release(&session->ses_lock);
router = session->service->router;
router_instance = session->service->router_instance;
rsession = session->router_session;
/** Close router session and all its connections */
router->closeSession(router_instance, rsession);
router = session->service->router;
/**
* If router session is being created concurrently router
* session might be NULL and it shouldn't be closed.
*/
if (session->router_session != NULL)
{
spinlock_release(&session->ses_lock);
/** Close router session and all its connections */
router->closeSession(router_instance, session->router_session);
}
else
{
spinlock_release(&session->ses_lock);
}
}
return 1;
}

55
server/modules/protocol/mysql_common.c
View File

@ -35,6 +35,7 @@
* x.y.z.%, x.y.%.%, x.%.%.%
* 03/10/2014 Massimiliano Pinto Added netmask for wildcard in IPv4 hosts.
* 24/10/2014 Massimiliano Pinto Added Mysql user@host @db authentication support
* 10/11/2014 Massimiliano Pinto Charset at connect is passed to backend during authentication
*
*/
@ -563,6 +564,7 @@ int gw_send_authentication_to_backend(
char *curr_db = NULL;
uint8_t *curr_passwd = NULL;
unsigned int charset;
if (strlen(dbname))
curr_db = dbname;
@ -574,7 +576,10 @@ int gw_send_authentication_to_backend(
final_capabilities = gw_mysql_get_byte4((uint8_t *)&server_capabilities);
/** Copy client's flags to backend */
final_capabilities |= conn->client_capabilities;;
final_capabilities |= conn->client_capabilities;
/* get charset the client sent and use it for connection auth */
charset = conn->charset;
if (compress) {
final_capabilities |= GW_MYSQL_CAPABILITIES_COMPRESS;
@ -668,7 +673,7 @@ int gw_send_authentication_to_backend(
// set the charset
payload += 4;
*payload = '\x08';
*payload = charset;
payload++;
@ -768,9 +773,43 @@ int gw_do_connect_to_backend(
setipaddress(&serv_addr.sin_addr, host);
serv_addr.sin_port = htons(port);
bufsize = GW_BACKEND_SO_SNDBUF;
setsockopt(so, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize));
if(setsockopt(so, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)) != 0)
{
int eno = errno;
errno = 0;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to set socket options "
"%s:%d failed.\n\t\t Socket configuration failed "
"due %d, %s.",
host,
port,
eno,
strerror(eno))));
rv = -1;
goto return_rv;
}
bufsize = GW_BACKEND_SO_RCVBUF;
setsockopt(so, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize));
if(setsockopt(so, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)) != 0)
{
int eno = errno;
errno = 0;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to set socket options "
"%s:%d failed.\n\t\t Socket configuration failed "
"due %d, %s.",
host,
port,
eno,
strerror(eno))));
rv = -1;
goto return_rv;
}
/* set socket to as non-blocking here */
setnonblocking(so);
rv = connect(so, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
@ -1050,6 +1089,7 @@ int gw_send_change_user_to_backend(
char *curr_db = NULL;
uint8_t *curr_passwd = NULL;
unsigned int charset;
if (strlen(dbname))
curr_db = dbname;
@ -1062,7 +1102,10 @@ int gw_send_change_user_to_backend(
final_capabilities = gw_mysql_get_byte4((uint8_t *)&server_capabilities);
/** Copy client's flags to backend */
final_capabilities |= conn->client_capabilities;;
final_capabilities |= conn->client_capabilities;
/* get charset the client sent and use it for connection auth */
charset = conn->charset;
if (compress) {
final_capabilities |= GW_MYSQL_CAPABILITIES_COMPRESS;
@ -1188,7 +1231,7 @@ int gw_send_change_user_to_backend(
}
// set the charset, 2 bytes!!!!
*payload = '\x08';
*payload = charset;
payload++;
*payload = '\x00';
payload++;

12
server/modules/routing/debugcmd.c
View File

@ -414,7 +414,7 @@ struct subcommand disableoptions[] = {
}
};
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
static void fail_backendfd(void);
static void fail_clientfd(void);
@ -456,7 +456,7 @@ struct subcommand failoptions[] = {
{0, 0, 0}
}
};
#endif /* SS_DEBUG */
#endif /* FAKE_CODE */
static void telnetdAddUser(DCB *, char *, char *);
/**
@ -502,9 +502,9 @@ static struct {
{ "clear", clearoptions },
{ "disable", disableoptions },
{ "enable", enableoptions },
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
{ "fail", failoptions },
#endif
#endif /* FAKE_CODE */
{ "list", listoptions },
{ "reload", reloadoptions },
{ "remove", removeoptions },
@ -1113,7 +1113,7 @@ static void disable_log_action(DCB *dcb, char *arg1) {
skygw_log_disable(type);
}
#if defined(SS_DEBUG)
#if defined(FAKE_CODE)
static void fail_backendfd(void)
{
fail_next_backend_fd = true;
@ -1157,4 +1157,4 @@ static void fail_accept(
return ;
}
}
#endif
#endif /* FAKE_CODE */

11
server/modules/routing/readconnroute.c
View File

@ -828,6 +828,17 @@ static void handleError(
SESSION *session = backend_dcb->session;
session_state_t sesstate;
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
backend_dcb->dcb_errhandle_called = true;
}
spinlock_acquire(&session->ses_lock);
sesstate = session->state;
client_dcb = session->client;

149
server/modules/routing/readwritesplit/readwritesplit.c
View File

@ -830,8 +830,16 @@ static void* newSession(
* Find a backend servers to connect to.
* This command requires that rsession's lock is held.
*/
rses_begin_locked_router_action(client_rses);
succp = rses_begin_locked_router_action(client_rses);
if(!succp)
{
free(client_rses->rses_backend_ref);
free(client_rses);
client_rses = NULL;
goto return_rses;
}
succp = select_connect_backend_servers(&master_ref,
backend_ref,
router_nservers,
@ -1029,6 +1037,9 @@ static void freeSession(
/**
* Provide the router with a pointer to a suitable backend dcb.
*
* As of Nov. 2014, slave which has least connections is always chosen.
*
* Detect failures in server statuses and reselect backends if necessary.
* If name is specified, server name becomes primary selection criteria.
*
@ -1610,8 +1621,12 @@ void check_create_tmp_table(
rses_prop_tmp->rses_prop_type = RSES_PROP_TYPE_TMPTABLES;
router_cli_ses->rses_properties[RSES_PROP_TYPE_TMPTABLES] = rses_prop_tmp;
}
else
{
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,"Error : Call to malloc() failed.")));
}
}
if(rses_prop_tmp){
if (rses_prop_tmp->rses_prop_data.temp_tables == NULL)
{
h = hashtable_alloc(7, hashkeyfun, hashcmpfun);
@ -1619,10 +1634,13 @@ void check_create_tmp_table(
if (h != NULL)
{
rses_prop_tmp->rses_prop_data.temp_tables = h;
}
}else{
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,"Error : Failed to allocate a new hashtable.")));
}
}
if (hkey &&
if (hkey && rses_prop_tmp->rses_prop_data.temp_tables &&
hashtable_add(rses_prop_tmp->rses_prop_data.temp_tables,
(void *)hkey,
(void *)is_temp) == 0) /*< Conflict in hash table */
@ -1647,6 +1665,8 @@ void check_create_tmp_table(
}
}
#endif
}
free(hkey);
free(tblname);
}
@ -2074,14 +2094,11 @@ static int routeQuery(
"route to master "
"but couldn't find "
"master in a "
"suitable state "
"failed.")));
"suitable state.")));
}
/**
* Master has changed. Set the dcb pointer NULL and
* return with error indicator.
* Master has changed. Return with error indicator.
*/
router_cli_ses->rses_master_ref->bref_dcb = NULL;
rses_end_locked_router_action(router_cli_ses);
succp = false;
ret = 0;
@ -2655,7 +2672,7 @@ static bool select_connect_backend_servers(
const int min_nslaves = 0; /*< not configurable at the time */
bool is_synced_master;
int (*p)(const void *, const void *);
BACKEND* master_host = NULL;
BACKEND* master_host;
if (p_master_ref == NULL || backend_ref == NULL)
{
@ -2667,46 +2684,32 @@ static bool select_connect_backend_servers(
/* get the root Master */
master_host = get_root_master(backend_ref, router_nservers);
/**
* Master is already chosen and connected. It means that the function
* was called from error handling function or from some other similar
* function where session was already established but new slaves needed
* to be selected.
/**
* Existing session : master is already chosen and connected.
* The function was called because new slave must be selected to replace
* failed one.
*/
if (*p_master_ref != NULL &&
BREF_IS_IN_USE((*p_master_ref)))
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [select_connect_backend_servers] Master %p fd %d found.",
pthread_self(),
(*p_master_ref)->bref_dcb,
(*p_master_ref)->bref_dcb->fd)));
master_found = true;
master_connected = true;
if (*p_master_ref != NULL)
{
/**
* Ensure that *p_master_ref and master_host point to same backend
* and it has a master role.
* Ensure that backend reference is in use, stored master is
* still current root master.
*/
ss_dassert(master_host &&
((*p_master_ref)->bref_backend->backend_server ==
master_host->backend_server) &&
(master_host->backend_server->status &
(SERVER_MASTER|SERVER_MAINT)) == SERVER_MASTER);
}
/** New session or master failure case */
if (!BREF_IS_IN_USE((*p_master_ref)) ||
!SERVER_IS_MASTER((*p_master_ref)->bref_backend->backend_server) ||
master_host != (*p_master_ref)->bref_backend)
{
succp = false;
goto return_succp;
}
master_found = true;
master_connected = true;
}
/**
* New session : select master and slaves
*/
else
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [select_connect_backend_servers] Session %p doesn't "
"currently have a master chosen. Proceeding to master "
"selection.",
pthread_self(),
session)));
master_found = false;
master_connected = false;
}
@ -2745,11 +2748,6 @@ static bool select_connect_backend_servers(
b->backend_conn_count)));
}
#endif
/* assert with master_host */
ss_dassert(!master_connected ||
(master_host &&
((*p_master_ref)->bref_backend->backend_server == master_host->backend_server) &&
SERVER_MASTER));
/**
* Sort the pointer list to servers according to connection counts. As
* a consequence those backends having least connections are in the
@ -2840,8 +2838,10 @@ static bool select_connect_backend_servers(
(max_slave_rlag == MAX_RLAG_UNDEFINED ||
(b->backend_server->rlag != MAX_RLAG_NOT_AVAILABLE &&
b->backend_server->rlag <= max_slave_rlag)) &&
(SERVER_IS_SLAVE(b->backend_server) || SERVER_IS_RELAY_SERVER(b->backend_server)) &&
(master_host != NULL && (b->backend_server != master_host->backend_server)))
(SERVER_IS_SLAVE(b->backend_server) ||
SERVER_IS_RELAY_SERVER(b->backend_server)) &&
(master_host != NULL &&
(b->backend_server != master_host->backend_server)))
{
slaves_found += 1;
@ -2903,6 +2903,12 @@ static bool select_connect_backend_servers(
else if (master_host &&
(b->backend_server == master_host->backend_server))
{
/**
* *p_master_ref must be assigned with this
* backend_ref pointer because its original value
* may have been lost when backend references were
* sorted (qsort).
*/
*p_master_ref = &backend_ref[i];
if (master_connected)
@ -4072,7 +4078,6 @@ static void rwsplit_process_router_options(
* Even if succp == true connecting to new slave may have failed. succp is to
* tell whether router has enough master/slave connections to continue work.
*/
static void handleError (
ROUTER* instance,
void* router_session,
@ -4086,10 +4091,17 @@ static void handleError (
ROUTER_CLIENT_SES* rses = (ROUTER_CLIENT_SES *)router_session;
CHK_DCB(backend_dcb);
#if defined(SS_DEBUG)
ss_dassert(!backend_dcb->dcb_errhandle_called);
backend_dcb->dcb_errhandle_called = true;
#endif
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
backend_dcb->dcb_errhandle_called = true;
}
session = backend_dcb->session;
if (session != NULL)
@ -4107,7 +4119,8 @@ static void handleError (
return;
}
if (rses->rses_master_ref->bref_dcb == backend_dcb)
if (rses->rses_master_ref->bref_dcb == backend_dcb &&
!SERVER_IS_MASTER(rses->rses_master_ref->bref_backend->backend_server))
{
/** Master failed, can't recover */
LOGIF(LE, (skygw_log_write_flush(
@ -4206,6 +4219,11 @@ static bool handle_error_new_connection(
}
CHK_BACKEND_REF(bref);
/**
* If query was sent through the bref and it is waiting for reply from
* the backend server it is necessary to send an error to the client
* because it is waiting for reply.
*/
if (BREF_IS_WAITING_RESULT(bref))
{
DCB* client_dcb;
@ -4473,6 +4491,10 @@ static backend_ref_t* get_bref_from_dcb(
return bref;
}
/**
* Calls hang-up function for DCB if it is not both running and in
* master/slave/joined/ndb role. Called by DCB's callback routine.
*/
static int router_handle_state_switch(
DCB* dcb,
DCB_REASON reason,
@ -4513,6 +4535,7 @@ return_rc:
return rc;
}
static sescmd_cursor_t* backend_ref_get_sescmd_cursor (
backend_ref_t* bref)
{
@ -4634,7 +4657,7 @@ static BACKEND *get_root_master(
* Servers are checked even if they are in 'maintenance'
*
* @param rses pointer to router session
* @return pointer to backend reference of the root master
* @return pointer to backend reference of the root master or NULL
*
*/
static backend_ref_t* get_root_master_bref(
@ -4664,6 +4687,14 @@ static backend_ref_t* get_root_master_bref(
bref++;
i += 1;
}
if (candidate_bref == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Could not find master among the backend "
"servers. Previous master state : %s",
STRSRVSTATUS(BREFSRV(rses->rses_master_ref)))));
}
return candidate_bref;
}

2
server/modules/routing/test/CMakeLists.txt
View File

@ -1,6 +1,6 @@
if(MYSQLCLIENT_FOUND)
add_executable(testconnect testconnect.c)
message(STATUS "Linking against: ${MYSQLCLIENT_LIBRARIES}")
target_link_libraries(testconnect ${MYSQLCLIENT_LIBRARIES} ssl crypto dl z m)
target_link_libraries(testconnect ${MYSQLCLIENT_LIBRARIES} ssl crypto dl z m rt pthread)
add_test(NAME ReadConnRouterLoginTest COMMAND $<TARGET_FILE:testconnect> 10000 ${TEST_HOST} ${MASTER_PORT} ${TEST_HOST} ${TEST_PORT} 1.10)
endif()

118
server/modules/routing/test/testconnect.c
View File

@ -4,24 +4,40 @@
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <sys/time.h>
int main(int argc, char** argv)
{
MYSQL* server;
char *host;
char *host = NULL, *str_baseline = NULL,*str_test = NULL, *errmsg = NULL;
unsigned int port;
int rval, iterations,i;
clock_t begin,end;
double baseline,test, ratio, result, minimum;
size_t offset;
struct timeval real_begin,real_end,real_baseline,real_test;
time_t time;
double baseline, test, ratio, result;
if(argc < 7){
fprintf(stderr,"Usage: %s <iterations> <baseline host> <baseline port> <test host> <test port> <max result ratio>\n",argv[0]);
fprintf(stderr,"The ratio is measured as:\ntest CPU time / baseline CPU time\n");
fprintf(stderr,"The ratio is measured as:\ntest time / baseline time\n");
fprintf(stderr,"The test fails if this ratio is exceeded.\n");
return 1;
}
;
if((str_baseline = calloc(256,sizeof(char))) == NULL){
return 1;
}
if((str_test = calloc(256,sizeof(char))) == NULL){
free(str_baseline);
return 1;
}
iterations = atoi(argv[1]);
host = strdup(argv[2]);
port = atoi(argv[3]);
@ -35,24 +51,36 @@ int main(int argc, char** argv)
/**Testing direct connection to master*/
printf("Connecting to MySQL server through %s:%d.\n",host,port);
gettimeofday(&real_begin,NULL);
begin = clock();
if((server = mysql_init(NULL)) == NULL){
return 1;
}
if(mysql_real_connect(server,host,"maxuser","maxpwd",NULL,port,NULL,0) == NULL){
rval = 1;
printf( "Failed to connect to database: Error: %s\n",
mysql_error(server));
goto report;
}
for(i = 0;i<iterations;i++)
{
if((server = mysql_init(NULL)) == NULL){
return 1;
}
if(mysql_real_connect(server,host,"maxuser","maxpwd",NULL,port,NULL,0) == NULL){
fprintf(stderr, "Failed to connect to database: Error: %s\n",
mysql_error(server));
if(mysql_change_user(server,"maxuser","maxpwd",NULL)){
rval = 1;
break;
printf( "Failed to change user: Error: %s\n",
mysql_error(server));
goto report;
}
mysql_close(server);
}
mysql_close(server);
end = clock();
gettimeofday(&real_end,NULL);
baseline = (double)(end - begin)/CLOCKS_PER_SEC;
timersub(&real_end,&real_begin,&real_baseline);
free(host);
host = strdup(argv[4]);
@ -61,37 +89,73 @@ int main(int argc, char** argv)
/**Testing connection to master through MaxScale*/
printf("Connecting to MySQL server through %s:%d.\n",host,port);
gettimeofday(&real_begin,NULL);
begin = clock();
if((server = mysql_init(NULL)) == NULL){
return 1;
}
if(mysql_real_connect(server,host,"maxuser","maxpwd",NULL,port,NULL,0) == NULL){
rval = 1;
printf("Failed to connect to database: Error: %s\n",
mysql_error(server));
goto report;
}
for(i = 0;i<iterations;i++)
{
if((server = mysql_init(NULL)) == NULL){
return 1;
}
if(mysql_real_connect(server,host,"maxuser","maxpwd",NULL,port,NULL,0) == NULL){
if(mysql_change_user(server,"maxuser","maxpwd",NULL)){
rval = 1;
fprintf(stderr, "Failed to connect to database: Error: %s\n",
printf("Failed to change user: Error: %s\n",
mysql_error(server));
break;
goto report;
}
mysql_close(server);
}
mysql_close(server);
end = clock();
gettimeofday(&real_end,NULL);
test = (double)(end - begin)/CLOCKS_PER_SEC;
timersub(&real_end,&real_begin,&real_test);
printf("CPU time used in seconds:\nDirect connection: %f\nThrough MaxScale: %f\n",baseline,test);
report:
result = test / baseline;
if(rval){
printf("Test failed: Errors during test run.");
}else if(result > ratio){
printf("Test failed: CPU time ratio was %f which exceeded the limit of %f.\n", result, ratio);
rval = 1;
}else{
printf("Test passed: CPU time ratio was %f.\n",result);
}
printf("\nTest failed: Errors during test run.\n");
}else{
struct tm *tm;
time = real_baseline.tv_sec;
tm = localtime(&time);
offset = strftime(str_baseline,256*sizeof(char),"%S",tm);
sprintf(str_baseline + offset,".%06d",(int)real_baseline.tv_usec);
time = real_test.tv_sec;
tm = localtime(&time);
offset = strftime(str_test,256*sizeof(char),"%S",tm);
sprintf(str_test + offset,".%06d",(int)real_test.tv_usec);
printf("\n\tCPU time in seconds\n\nDirect connection: %f\nThrough MaxScale: %f\n",baseline,test);
printf("\n\tReal time in seconds\n\nDirect connection: %s\nThrough MaxScale: %s\n",str_baseline,str_test);
double base_res = real_baseline.tv_sec + (real_baseline.tv_usec / 1000000.0);
double test_res = real_test.tv_sec + (real_test.tv_usec / 1000000.0);
result = test_res/base_res;
if(result > ratio){
printf("\nTest failed: Time ratio was %f which exceeded the limit of %f.\n", result, ratio);
rval = 1;
}else{
printf("\nTest passed: Time ratio was %f.\n",result);
}
}
free(str_baseline);
free(str_test);
free(host);
free(errmsg);
return rval;
}

3
utils/skygw_debug.h
View File

@ -263,6 +263,9 @@ typedef enum skygw_chk_t {
(SERVER_IS_RELAY_SERVER(s) ? "RUNNING RELAY" : \
(SERVER_IS_RUNNING(s) ? "RUNNING (only)" : "NO STATUS")))))))
#define BREFSRV(b) (b->bref_backend->backend_server)
#define STRHINTTYPE(t) (t == HINT_ROUTE_TO_MASTER ? "HINT_ROUTE_TO_MASTER" : \
((t) == HINT_ROUTE_TO_SLAVE ? "HINT_ROUTE_TO_SLAVE" : \
((t) == HINT_ROUTE_TO_NAMED_SERVER ? "HINT_ROUTE_TO_NAMED_SERVER" : \

26
utils/skygw_utils.cc
View File

@ -1749,14 +1749,23 @@ return_succp:
return succp;
}
bool skygw_file_write(
/**
* Write data to a file.
*
* @param file write target
* @param data pointer to contiguous memory buffer
* @param nbytes amount of bytes to be written
* @param flush ensure that write is permanent
*
* @return 0 if succeed, errno if failed.
*/
int skygw_file_write(
skygw_file_t* file,
void* data,
size_t nbytes,
bool flush)
{
bool succp = false;
int rc;
#if !defined(LAPTOP_TEST)
int err = 0;
size_t nwritten;
@ -1771,13 +1780,14 @@ bool skygw_file_write(
nwritten = fwrite(data, nbytes, 1, file->sf_file);
if (nwritten != 1) {
rc = errno;
perror("Logfile write.\n");
fprintf(stderr,
"* Writing %ld bytes, %s to %s failed.\n",
"* Writing %ld bytes,\n%s\n to %s failed.\n",
nbytes,
(char *)data,
file->sf_fname);
goto return_succp;
goto return_rc;
}
writecount += 1;
@ -1789,10 +1799,10 @@ bool skygw_file_write(
writecount = 0;
}
#endif
succp = true;
rc = 0;
CHK_FILE(file);
return_succp:
return succp;
return_rc:
return rc;
}
skygw_file_t* skygw_file_init(

2
utils/skygw_utils.h
View File

@ -146,7 +146,7 @@ EXTERN_C_BLOCK_END
/** Skygw file routines */
skygw_file_t* skygw_file_init(char* fname, char* symlinkname);
void skygw_file_done(skygw_file_t* file);
bool skygw_file_write(
int skygw_file_write(
skygw_file_t* file,
void* data,
size_t nbytes,