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

Branch update

Browse Source 
Branch update
This commit is contained in:
MassimilianoPinto
2014-11-07 16:59:01 +01:00
parent e812d765ef 3b982ae820
commit 39cc701b0d
53 changed files with 1988 additions and 620 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
server/modules/filter/qlafilter.c
View File

@ -40,6 +40,7 @@
*/
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <filter.h>
#include <modinfo.h>
#include <modutil.h>
@ -170,10 +171,11 @@ int i;
if ((my_instance = calloc(1, sizeof(QLA_INSTANCE))) != NULL)
{
if (options)
if (options){
my_instance->filebase = strdup(options[0]);
else
}else{
my_instance->filebase = strdup("qla");
}
my_instance->source = NULL;
my_instance->userName = NULL;
my_instance->match = NULL;
@ -196,9 +198,11 @@ int i;
my_instance->userName = strdup(params[i]->value);
else if (!strcmp(params[i]->name, "filebase"))
{
if (my_instance->filebase)
if (my_instance->filebase){
free(my_instance->filebase);
my_instance->source = strdup(params[i]->value);
my_instance->filebase = NULL;
}
my_instance->filebase = strdup(params[i]->value);
}
else if (!filter_standard_parameter(params[i]->name))
{
@ -219,7 +223,9 @@ int i;
my_instance->match)));
free(my_instance->match);
free(my_instance->source);
free(my_instance->filebase);
if(my_instance->filebase){
free(my_instance->filebase);
}
free(my_instance);
return NULL;
}
@ -235,7 +241,9 @@ int i;
regfree(&my_instance->re);
free(my_instance->match);
free(my_instance->source);
free(my_instance->filebase);
if(my_instance->filebase){
free(my_instance->filebase);
}
free(my_instance);
return NULL;
}
@ -265,10 +273,17 @@ char *remote, *userName;
(char *)malloc(strlen(my_instance->filebase) + 20))
== NULL)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Memory allocation for qla filter "
"file name failed due to %d, %s.",
errno,
strerror(errno))));
free(my_session);
return NULL;
}
my_session->active = 1;
if (my_instance->source
&& (remote = session_get_remote(session)) != NULL)
{
@ -276,16 +291,45 @@ char *remote, *userName;
my_session->active = 0;
}
userName = session_getUser(session);
if (my_instance->userName && userName && strcmp(userName,
my_instance->userName))
if (my_instance->userName &&
userName &&
strcmp(userName,my_instance->userName))
{
my_session->active = 0;
sprintf(my_session->filename, "%s.%d", my_instance->filebase,
my_instance->sessions);
}
sprintf(my_session->filename, "%s.%d",
my_instance->filebase,
my_instance->sessions);
my_instance->sessions++;
if (my_session->active)
{
my_session->fp = fopen(my_session->filename, "w");
if (my_session->fp == NULL)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Opening output file for qla "
"fileter failed due to %d, %s",
errno,
strerror(errno))));
free(my_session->filename);
free(my_session);
my_session = NULL;
}
}
}
else
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Memory allocation for qla filter failed due to "
"%d, %s.",
errno,
strerror(errno))));
}
return my_session;
}
@ -364,7 +408,7 @@ struct timeval tv;
{
queue = gwbuf_make_contiguous(queue);
}
if (modutil_extract_SQL(queue, &ptr, &length) != 0)
if ((ptr = modutil_get_SQL(queue)) != NULL)
{
if ((my_instance->match == NULL ||
regexec(&my_instance->re, ptr, 0, NULL, 0) == 0) &&
@ -380,6 +424,7 @@ struct timeval tv;
fwrite(ptr, sizeof(char), length, my_session->fp);
fwrite("\n", sizeof(char), 1, my_session->fp);
}
free(ptr);
}
}
/* Pass the query downstream */

38
server/modules/filter/regexfilter.c
View File

@ -60,7 +60,7 @@ static void setDownstream(FILTER *instance, void *fsession, DOWNSTREAM *downstre
static int routeQuery(FILTER *instance, void *fsession, GWBUF *queue);
static void diagnostic(FILTER *instance, void *fsession, DCB *dcb);
static char *regex_replace(char *sql, int length, regex_t *re, char *replace);
static char *regex_replace(char *sql, regex_t *re, char *replace);
static FILTER_OBJECT MyObject = {
createInstance,
@ -192,6 +192,7 @@ int i, cflags = REG_ICASE;
if (my_instance->match == NULL || my_instance->replace == NULL)
{
free(my_instance);
return NULL;
}
@ -301,7 +302,6 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
REGEX_INSTANCE *my_instance = (REGEX_INSTANCE *)instance;
REGEX_SESSION *my_session = (REGEX_SESSION *)session;
char *sql, *newsql;
int length;
if (modutil_is_SQL(queue))
{
@ -309,18 +309,21 @@ int length;
{
queue = gwbuf_make_contiguous(queue);
}
modutil_extract_SQL(queue, &sql, &length);
newsql = regex_replace(sql, length, &my_instance->re,
my_instance->replace);
if (newsql)
if ((sql = modutil_get_SQL(queue)) != NULL)
{
queue = modutil_replace_SQL(queue, newsql);
queue = gwbuf_make_contiguous(queue);
free(newsql);
my_session->replacements++;
newsql = regex_replace(sql, &my_instance->re,
my_instance->replace);
if (newsql)
{
queue = modutil_replace_SQL(queue, newsql);
queue = gwbuf_make_contiguous(queue);
free(newsql);
my_session->replacements++;
}
else
my_session->no_change++;
free(sql);
}
else
my_session->no_change++;
}
return my_session->down.routeQuery(my_session->down.instance,
@ -367,25 +370,24 @@ REGEX_SESSION *my_session = (REGEX_SESSION *)fsession;
* Perform a regular expression match and subsititution on the SQL
*
* @param sql The original SQL text
* @param length The length of the SQL text
* @param re The compiled regular expression
* @param replace The replacement text
* @return The replaced text or NULL if no replacement was done.
*/
static char *
regex_replace(char *sql, int length, regex_t *re, char *replace)
regex_replace(char *sql, regex_t *re, char *replace)
{
char *orig, *result, *ptr;
int i, res_size, res_length, rep_length;
int last_match;
int last_match, length;
regmatch_t match[10];
orig = strndup(sql, length);
if (regexec(re, orig, 10, match, 0))
if (regexec(re, sql, 10, match, 0))
{
free(orig);
return NULL;
}
length = strlen(sql);
res_size = 2 * length;
result = (char *)malloc(res_size);
res_length = 0;

9
server/modules/filter/tee.c
View File

@ -391,7 +391,7 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
TEE_INSTANCE *my_instance = (TEE_INSTANCE *)instance;
TEE_SESSION *my_session = (TEE_SESSION *)session;
char *ptr;
int length, rval, residual;
int length, rval, residual = 0;
GWBUF *clone = NULL;
if (my_session->residual)
@ -403,16 +403,19 @@ GWBUF *clone = NULL;
if (my_session->residual < 0)
my_session->residual = 0;
}
else if (my_session->active &&
modutil_MySQL_Query(queue, &ptr, &length, &residual))
else if (my_session->active && (ptr = modutil_get_SQL(queue) != NULL))
{
if ((my_instance->match == NULL ||
regexec(&my_instance->re, ptr, 0, NULL, 0) == 0) &&
(my_instance->nomatch == NULL ||
regexec(&my_instance->nore,ptr,0,NULL, 0) != 0))
{
char *dummy;
modutil_MySQL_Query(queue, &dummy, &length, &residual);
clone = gwbuf_clone(queue);
my_session->residual = residual;
free(ptr);
}
}

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

@ -137,8 +137,8 @@ FILTER_PARAMETER** read_params(int* paramc)
do_read = 0;
}
}
FILTER_PARAMETER** params;
if((params = malloc(sizeof(FILTER_PARAMETER*)*(pc+1)))!=NULL){
FILTER_PARAMETER** params = NULL;
if((params = malloc(sizeof(FILTER_PARAMETER*)*(pc+1))) != NULL){
for(i = 0;i<pc;i++){
params[i] = malloc(sizeof(FILTER_PARAMETER));
if(params[i]){
@ -147,10 +147,10 @@ FILTER_PARAMETER** read_params(int* paramc)
}
free(names[i]);
free(values[i]);
}
}
params[pc] = NULL;
*paramc = pc;
}
params[pc] = NULL;
*paramc = pc;
return params;
}
@ -267,83 +267,105 @@ int clientReply(void* ins, void* session, GWBUF* queue)
return 1;
}
/**
* Read a string from a file descriptor to a block of memory
* @param fd File descriptor to read from, assumed to be open
* @param buff Buffer to write to
* @param size Size of the buffer
* @return Number of bytes read
*/
int fdgets(int fd, char* buff, int size)
{
int i = 0;
while(i < size - 1 && read(fd,&buff[i],1))
{
if(buff[i] == '\n' || buff[i] == '\0')
{
break;
}
i++;
}
buff[i] = '\0';
return i;
}
/**
* Loads a query from a file
*@return 0 if successful, 1 if an error occurred
*/
int load_query()
{
char** query_list;
char* buff;
char rc;
int i, qcount = 0, qbuff_sz = 10, buff_sz = 2048;
char** query_list = NULL;
char* buffer = NULL;
int i, qcount = 0, qbuff_sz = 10, rval = 0;
int offset = 0;
unsigned int qlen = 0;
if((buff = calloc(buff_sz,sizeof(char))) == NULL ||
(query_list = calloc(qbuff_sz,sizeof(char*))) == NULL){
buffer = (char*)calloc(4092,sizeof(char));
if(buffer == NULL){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return 1;
}
while(read(instance.infile,&rc,1)){
if(rc != '\n' && rc != '\0'){
if(offset >= buff_sz){
char* tmp = malloc(sizeof(char)*2*buff_sz);
if(tmp){
memcpy(tmp,buff,buff_sz);
free(buff);
buff = tmp;
buff_sz *= 2;
}else{
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
free(buff);
return 1;
}
}
buff[offset++] = rc;
}else{
if(qcount >= qbuff_sz){
char** tmpcl = malloc(sizeof(char*) * (qcount * 2 + 1));
if(!tmpcl){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
return 1;
}
for(i = 0;i < qbuff_sz;i++){
tmpcl[i] = query_list[i];
}
free(query_list);
query_list = tmpcl;
qbuff_sz = qcount * 2 + 1;
}
query_list[qcount] = malloc(sizeof(char)*(offset + 1));
memcpy(query_list[qcount],buff,offset);
query_list[qcount][offset] = '\0';
offset = 0;
qcount++;
}
query_list = calloc(qbuff_sz,sizeof(char*));
if(query_list == NULL){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
free(buffer);
return 1;
}
while((offset = fdgets(instance.infile,buffer,4092))){
if(qbuff_sz <= qcount){
char** tmpbuff = realloc(query_list,sizeof(char*)*qbuff_sz*2);
if(tmpbuff == NULL){
printf("Error: cannot allocate enough memory.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory.\n");
rval = 1;
goto retblock;
}
query_list = tmpbuff;
qbuff_sz *= 2;
}
query_list[qcount] = calloc((offset + 1),sizeof(char));
strcpy(query_list[qcount],buffer);
offset = 0;
qcount++;
}
/**TODO check what messes up the first querystring*/
GWBUF** tmpbff = malloc(sizeof(GWBUF*)*(qcount + 1));
if(tmpbff){
for(i = 0;i<qcount;i++){
tmpbff[i] = gwbuf_alloc(strnlen(query_list[i],buff_sz) + 6);
tmpbff[i] = gwbuf_alloc(strlen(query_list[i]) + 6);
if(tmpbff[i] == NULL)
{
printf("Error: cannot allocate a new buffer.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate a new buffer.\n");
int x;
for(x = 0;x<i;x++)
{
gwbuf_free(tmpbff[x]);
}
free(tmpbff);
rval = 1;
goto retblock;
}
gwbuf_set_type(tmpbff[i],GWBUF_TYPE_MYSQL);
memcpy(tmpbff[i]->sbuf->data + 5,query_list[i],strnlen(query_list[i],buff_sz));
qlen = strnlen(query_list[i],buff_sz);
strcpy((char*)(tmpbff[i]->start + 5),query_list[i]);
qlen = strlen(query_list[i]) + 1;
tmpbff[i]->sbuf->data[0] = qlen;
tmpbff[i]->sbuf->data[1] = (qlen << 8);
tmpbff[i]->sbuf->data[2] = (qlen << 16);
@ -357,14 +379,26 @@ int load_query()
printf("Error: cannot allocate enough memory for buffers.\n");
skygw_log_write(LOGFILE_ERROR,"Error: cannot allocate enough memory for buffers.\n");
free_buffers();
return 1;
rval = 1;
goto retblock;
}
if(qcount < 1){
return 1;
rval = 1;
goto retblock;
}
instance.buffer_count = qcount;
return 0;
retblock:
for(i = 0;i<qcount;i++)
{
free(query_list[i]);
}
free(query_list);
free(buffer);
return rval;
}
@ -572,7 +606,7 @@ int load_config( char* fname)
int load_filter(FILTERCHAIN* fc, CONFIG* cnf)
{
FILTER_PARAMETER** fparams;
FILTER_PARAMETER** fparams = NULL;
int i, paramc = -1;
if(cnf == NULL){
@ -640,7 +674,7 @@ int load_filter(FILTERCHAIN* fc, CONFIG* cnf)
int sess_err = 0;
if(fc && fc->instance){
if(cnf && fc && fc->instance){
fc->filter = (FILTER*)fc->instance->createInstance(NULL,fparams);
@ -713,15 +747,14 @@ int load_filter(FILTERCHAIN* fc, CONFIG* cnf)
}
if(cnf){
int x;
for(x = 0;x<paramc;x++){
free(fparams[x]->name);
free(fparams[x]->value);
}
free(fparams);
int x;
for(x = 0;x<paramc;x++){
free(fparams[x]->name);
free(fparams[x]->value);
}
free(fparams);
return sess_err ? 0 : 1;
}
@ -892,8 +925,9 @@ GWBUF* gen_packet(PACKET pkt)
int process_opts(int argc, char** argv)
{
int fd = open_file("harness.cnf",1), buffsize = 1024;
int rd,fsize;
unsigned int fd = open_file("harness.cnf",1), buffsize = 1024;
int rd,rdsz;
unsigned int fsize;
char *buff = calloc(buffsize,sizeof(char)), *tok = NULL;
/**Parse 'harness.cnf' file*/
@ -901,7 +935,8 @@ int process_opts(int argc, char** argv)
lseek(fd,0,SEEK_SET);
instance.thrcount = 1;
instance.session_count = 1;
read(fd,buff,fsize);
rdsz = read(fd,buff,fsize);
buff[rdsz] = '\0';
tok = strtok(buff,"=");
while(tok){
if(!strcmp(tok,"threads")){
@ -920,6 +955,7 @@ int process_opts(int argc, char** argv)
instance.verbose = 1;
if(argc < 2){
close(fd);
return 1;
}
char* conf_name = NULL;
@ -937,6 +973,9 @@ int process_opts(int argc, char** argv)
break;
case 'c':
if(conf_name){
free(conf_name);
}
conf_name = strdup(optarg);
break;
@ -992,6 +1031,8 @@ int process_opts(int argc, char** argv)
}else{
instance.running = 0;
}
free(conf_name);
close(fd);
return 0;
}

9
server/modules/filter/topfilter.c
View File

@ -453,7 +453,6 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
TOPN_INSTANCE *my_instance = (TOPN_INSTANCE *)instance;
TOPN_SESSION *my_session = (TOPN_SESSION *)session;
char *ptr;
int length;
if (my_session->active)
{
@ -461,7 +460,7 @@ int length;
{
queue = gwbuf_make_contiguous(queue);
}
if (modutil_extract_SQL(queue, &ptr, &length) != 0)
if ((ptr = modutil_get_SQL(queue)) != NULL)
{
if ((my_instance->match == NULL ||
regexec(&my_instance->re, ptr, 0, NULL, 0) == 0) &&
@ -472,7 +471,11 @@ int length;
if (my_session->current)
free(my_session->current);
gettimeofday(&my_session->start, NULL);
my_session->current = strndup(ptr, length);
my_session->current = ptr;
}
else
{
free(ptr);
}
}
}

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

@ -80,7 +80,8 @@ static MONITOR_OBJECT MyObject = {
unregisterServer,
defaultUsers,
diagnostics,
setInterval,
setInterval,
NULL,
NULL,
NULL,
NULL,

824
server/modules/monitor/mm_mon.c Normal file
View File

@ -0,0 +1,824 @@
/*
* This file is distributed as part of the 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 mysql_mon.c - A MySQL Multi Muster cluster monitor
*
* @verbatim
* Revision History
*
* Date Who Description
* 08/09/14 Massimiliano Pinto Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <monitor.h>
#include <mysqlmon.h>
#include <thread.h>
#include <mysql.h>
#include <mysqld_error.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <secrets.h>
#include <dcb.h>
#include <modinfo.h>
extern int lm_enabled_logfiles_bitmask;
static void monitorMain(void *);
static char *version_str = "V1.0.1";
MODULE_INFO info = {
MODULE_API_MONITOR,
MODULE_BETA_RELEASE,
MONITOR_VERSION,
"A MySQL Multi Master monitor"
};
static void *startMonitor(void *);
static void stopMonitor(void *);
static void registerServer(void *, SERVER *);
static void unregisterServer(void *, SERVER *);
static void defaultUser(void *, char *, char *);
static void diagnostics(DCB *, void *);
static void setInterval(void *, size_t);
static void detectStaleMaster(void *, int);
static bool mon_status_changed(MONITOR_SERVERS* mon_srv);
static bool mon_print_fail_status(MONITOR_SERVERS* mon_srv);
static MONITOR_SERVERS *get_current_master(MYSQL_MONITOR *);
static void monitor_set_pending_status(MONITOR_SERVERS *, int);
static void monitor_clear_pending_status(MONITOR_SERVERS *, int);
static MONITOR_OBJECT MyObject = {
startMonitor,
stopMonitor,
registerServer,
unregisterServer,
defaultUser,
diagnostics,
setInterval,
NULL,
NULL,
NULL,
detectStaleMaster,
NULL
};
/**
* 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(
LOGFILE_MESSAGE,
"Initialise the MySQL Monitor module %s.",
version_str)));
}
/**
* 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
*/
MONITOR_OBJECT *
GetModuleObject()
{
return &MyObject;
}
/**
* Start the instance of the monitor, returning a handle on the monitor.
*
* This function creates a thread to execute the actual monitoring.
*
* @param arg The current handle - NULL if first start
* @return A handle to use when interacting with the monitor
*/
static void *
startMonitor(void *arg)
{
MYSQL_MONITOR *handle;
if (arg)
{
handle = arg; /* Must be a restart */
handle->shutdown = 0;
}
else
{
if ((handle = (MYSQL_MONITOR *)malloc(sizeof(MYSQL_MONITOR))) == NULL)
return NULL;
handle->databases = NULL;
handle->shutdown = 0;
handle->defaultUser = NULL;
handle->defaultPasswd = NULL;
handle->id = MONITOR_DEFAULT_ID;
handle->interval = MONITOR_INTERVAL;
handle->replicationHeartbeat = 0;
handle->detectStaleMaster = 0;
handle->master = NULL;
spinlock_init(&handle->lock);
}
handle->tid = (THREAD)thread_start(monitorMain, handle);
return handle;
}
/**
* Stop a running monitor
*
* @param arg Handle on thr running monior
*/
static void
stopMonitor(void *arg)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
handle->shutdown = 1;
thread_wait((void *)handle->tid);
}
/**
* Register a server that must be added to the monitored servers for
* a monitoring module.
*
* @param arg A handle on the running monitor module
* @param server The server to add
*/
static void
registerServer(void *arg, SERVER *server)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *ptr, *db;
if ((db = (MONITOR_SERVERS *)malloc(sizeof(MONITOR_SERVERS))) == NULL)
return;
db->server = server;
db->con = NULL;
db->next = NULL;
db->mon_err_count = 0;
db->mon_prev_status = 0;
/* pending status is updated by monitorMain */
db->pending_status = 0;
spinlock_acquire(&handle->lock);
if (handle->databases == NULL)
handle->databases = db;
else
{
ptr = handle->databases;
while (ptr->next != NULL)
ptr = ptr->next;
ptr->next = db;
}
spinlock_release(&handle->lock);
}
/**
* Remove a server from those being monitored by a monitoring module
*
* @param arg A handle on the running monitor module
* @param server The server to remove
*/
static void
unregisterServer(void *arg, SERVER *server)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *ptr, *lptr;
spinlock_acquire(&handle->lock);
if (handle->databases == NULL)
{
spinlock_release(&handle->lock);
return;
}
if (handle->databases->server == server)
{
ptr = handle->databases;
handle->databases = handle->databases->next;
free(ptr);
}
else
{
ptr = handle->databases;
while (ptr->next != NULL && ptr->next->server != server)
ptr = ptr->next;
if (ptr->next)
{
lptr = ptr->next;
ptr->next = ptr->next->next;
free(lptr);
}
}
spinlock_release(&handle->lock);
}
/**
* Set the default username and password to use to monitor if the server does not
* override this.
*
* @param arg The handle allocated by startMonitor
* @param uname The default user name
* @param passwd The default password
*/
static void
defaultUser(void *arg, char *uname, char *passwd)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
if (handle->defaultUser)
free(handle->defaultUser);
if (handle->defaultPasswd)
free(handle->defaultPasswd);
handle->defaultUser = strdup(uname);
handle->defaultPasswd = strdup(passwd);
}
/**
* Daignostic interface
*
* @param dcb DCB to print diagnostics
* @param arg The monitor handle
*/
static void diagnostics(DCB *dcb, void *arg)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *db;
char *sep;
switch (handle->status)
{
case MONITOR_RUNNING:
dcb_printf(dcb, "\tMonitor running\n");
break;
case MONITOR_STOPPING:
dcb_printf(dcb, "\tMonitor stopping\n");
break;
case MONITOR_STOPPED:
dcb_printf(dcb, "\tMonitor stopped\n");
break;
}
dcb_printf(dcb,"\tSampling interval:\t%lu milliseconds\n", handle->interval);
dcb_printf(dcb,"\tDetect Stale Master:\t%s\n", (handle->detectStaleMaster == 1) ? "enabled" : "disabled");
dcb_printf(dcb, "\tMonitored servers: ");
db = handle->databases;
sep = "";
while (db)
{
dcb_printf(dcb,
"%s%s:%d",
sep,
db->server->name,
db->server->port);
sep = ", ";
db = db->next;
}
dcb_printf(dcb, "\n");
}
/**
* Monitor an individual server
*
* @param handle The MySQL Monitor object
* @param database The database to probe
*/
static void
monitorDatabase(MYSQL_MONITOR *handle, MONITOR_SERVERS *database)
{
MYSQL_ROW row;
MYSQL_RES *result;
int num_fields;
int isslave = 0;
int ismaster = 0;
char *uname = handle->defaultUser;
char *passwd = handle->defaultPasswd;
unsigned long int server_version = 0;
char *server_string;
if (database->server->monuser != NULL)
{
uname = database->server->monuser;
passwd = database->server->monpw;
}
if (uname == NULL)
return;
/* Don't probe servers in maintenance mode */
if (SERVER_IN_MAINT(database->server))
return;
/** Store prevous status */
database->mon_prev_status = database->server->status;
if (database->con == NULL || mysql_ping(database->con) != 0)
{
char *dpwd = decryptPassword(passwd);
int rc;
int read_timeout = 1;
database->con = mysql_init(NULL);
rc = mysql_options(database->con, MYSQL_OPT_READ_TIMEOUT, (void *)&read_timeout);
if (mysql_real_connect(database->con,
database->server->name,
uname,
dpwd,
NULL,
database->server->port,
NULL,
0) == NULL)
{
free(dpwd);
if (mon_print_fail_status(database))
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Monitor was unable to connect to "
"server %s:%d : \"%s\"",
database->server->name,
database->server->port,
mysql_error(database->con))));
}
/* The current server is not running
*
* Store server NOT running in server and monitor server pending struct
*
*/
if (mysql_errno(database->con) == ER_ACCESS_DENIED_ERROR)
{
server_set_status(database->server, SERVER_AUTH_ERROR);
monitor_set_pending_status(database, SERVER_AUTH_ERROR);
}
server_clear_status(database->server, SERVER_RUNNING);
monitor_clear_pending_status(database, SERVER_RUNNING);
/* Also clear M/S state in both server and monitor server pending struct */
server_clear_status(database->server, SERVER_SLAVE);
server_clear_status(database->server, SERVER_MASTER);
monitor_clear_pending_status(database, SERVER_SLAVE);
monitor_clear_pending_status(database, SERVER_MASTER);
/* Clean addition status too */
server_clear_status(database->server, SERVER_STALE_STATUS);
monitor_clear_pending_status(database, SERVER_STALE_STATUS);
return;
} else {
server_clear_status(database->server, SERVER_AUTH_ERROR);
monitor_clear_pending_status(database, SERVER_AUTH_ERROR);
}
free(dpwd);
}
/* Store current status in both server and monitor server pending struct */
server_set_status(database->server, SERVER_RUNNING);
monitor_set_pending_status(database, SERVER_RUNNING);
/* get server version from current server */
server_version = mysql_get_server_version(database->con);
/* get server version string */
server_string = (char *)mysql_get_server_info(database->con);
if (server_string) {
database->server->server_string = realloc(database->server->server_string, strlen(server_string)+1);
if (database->server->server_string)
strcpy(database->server->server_string, server_string);
}
/* get server_id form current node */
if (mysql_query(database->con, "SELECT @@server_id") == 0
&& (result = mysql_store_result(database->con)) != NULL)
{
long server_id = -1;
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
server_id = strtol(row[0], NULL, 10);
if ((errno == ERANGE && (server_id == LONG_MAX
|| server_id == LONG_MIN)) || (errno != 0 && server_id == 0))
{
server_id = -1;
}
database->server->node_id = server_id;
}
mysql_free_result(result);
}
/* Check if the Slave_SQL_Running and Slave_IO_Running status is
* set to Yes
*/
/* Check first for MariaDB 10.x.x and get status for multimaster replication */
if (server_version >= 100000) {
if (mysql_query(database->con, "SHOW ALL SLAVES STATUS") == 0
&& (result = mysql_store_result(database->con)) != NULL)
{
int i = 0;
long master_id = -1;
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
/* get Slave_IO_Running and Slave_SQL_Running values*/
if (strncmp(row[12], "Yes", 3) == 0
&& strncmp(row[13], "Yes", 3) == 0) {
isslave += 1;
}
/* If Slave_IO_Running = Yes, assign the master_id to current server: this allows building
* the replication tree, slaves ids will be added to master(s) and we will have at least the
* root master server.
* Please note, there could be no slaves at all if Slave_SQL_Running == 'No'
*/
if (strncmp(row[12], "Yes", 3) == 0) {
/* get Master_Server_Id values */
master_id = atol(row[41]);
if (master_id == 0)
master_id = -1;
}
i++;
}
/* store master_id of current node */
memcpy(&database->server->master_id, &master_id, sizeof(long));
mysql_free_result(result);
/* If all configured slaves are running set this node as slave */
if (isslave > 0 && isslave == i)
isslave = 1;
else
isslave = 0;
}
} else {
if (mysql_query(database->con, "SHOW SLAVE STATUS") == 0
&& (result = mysql_store_result(database->con)) != NULL)
{
long master_id = -1;
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
/* get Slave_IO_Running and Slave_SQL_Running values*/
if (strncmp(row[10], "Yes", 3) == 0
&& strncmp(row[11], "Yes", 3) == 0) {
isslave = 1;
}
/* If Slave_IO_Running = Yes, assign the master_id to current server: this allows building
* the replication tree, slaves ids will be added to master(s) and we will have at least the
* root master server.
* Please note, there could be no slaves at all if Slave_SQL_Running == 'No'
*/
if (strncmp(row[10], "Yes", 3) == 0) {
/* get Master_Server_Id values */
master_id = atol(row[39]);
if (master_id == 0)
master_id = -1;
}
}
/* store master_id of current node */
memcpy(&database->server->master_id, &master_id, sizeof(long));
mysql_free_result(result);
}
}
/* get variable 'read_only' set by an external component */
if (mysql_query(database->con, "SHOW GLOBAL VARIABLES LIKE 'read_only'") == 0
&& (result = mysql_store_result(database->con)) != NULL)
{
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
if (strncasecmp(row[1], "OFF", 3) == 0) {
ismaster = 1;
}
}
mysql_free_result(result);
}
/* Remove addition info */
monitor_clear_pending_status(database, SERVER_STALE_STATUS);
/* Set the Slave Role */
if (isslave)
{
monitor_set_pending_status(database, SERVER_SLAVE);
/* Avoid any possible stale Master state */
monitor_clear_pending_status(database, SERVER_MASTER);
/* Set replication depth to 1 */
database->server->depth = 1;
} else {
/* Avoid any possible Master/Slave stale state */
monitor_clear_pending_status(database, SERVER_SLAVE);
monitor_clear_pending_status(database, SERVER_MASTER);
}
/* Set the Master role */
if (isslave && ismaster)
{
monitor_clear_pending_status(database, SERVER_SLAVE);
monitor_set_pending_status(database, SERVER_MASTER);
/* Set replication depth to 0 */
database->server->depth = 0;
}
}
/**
* The entry point for the monitoring module thread
*
* @param arg The handle of the monitor
*/
static void
monitorMain(void *arg)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *ptr;
int detect_stale_master = handle->detectStaleMaster;
MONITOR_SERVERS *root_master;
size_t nrounds = 0;
if (mysql_thread_init())
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Fatal : mysql_thread_init failed in monitor "
"module. Exiting.\n")));
return;
}
handle->status = MONITOR_RUNNING;
while (1)
{
if (handle->shutdown)
{
handle->status = MONITOR_STOPPING;
mysql_thread_end();
handle->status = MONITOR_STOPPED;
return;
}
/** Wait base interval */
thread_millisleep(MON_BASE_INTERVAL_MS);
/**
* Calculate how far away the monitor interval is from its full
* cycle and if monitor interval time further than the base
* interval, then skip monitoring checks. Excluding the first
* round.
*/
if (nrounds != 0 &&
((nrounds*MON_BASE_INTERVAL_MS)%handle->interval) >=
MON_BASE_INTERVAL_MS)
{
nrounds += 1;
continue;
}
nrounds += 1;
/* start from the first server in the list */
ptr = handle->databases;
while (ptr)
{
/* copy server status into monitor pending_status */
ptr->pending_status = ptr->server->status;
/* monitor current node */
monitorDatabase(handle, ptr);
if (mon_status_changed(ptr))
{
dcb_call_foreach(DCB_REASON_NOT_RESPONDING);
}
if (mon_status_changed(ptr) ||
mon_print_fail_status(ptr))
{
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"Backend server %s:%d state : %s",
ptr->server->name,
ptr->server->port,
STRSRVSTATUS(ptr->server))));
}
if (SERVER_IS_DOWN(ptr->server))
{
/** Increase this server'e error count */
ptr->mon_err_count += 1;
}
else
{
/** Reset this server's error count */
ptr->mon_err_count = 0;
}
ptr = ptr->next;
}
/* Get Master server pointer */
root_master = get_current_master(handle);
/* Update server status from monitor pending status on that server*/
ptr = handle->databases;
while (ptr)
{
if (! SERVER_IN_MAINT(ptr->server)) {
/* If "detect_stale_master" option is On, let's use the previus master */
if (detect_stale_master && root_master && (!strcmp(ptr->server->name, root_master->server->name) && ptr->server->port == root_master->server->port) && (ptr->server->status & SERVER_MASTER) && !(ptr->pending_status & SERVER_MASTER)) {
/* in this case server->status will not be updated from pending_status */
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE, "[mysql_mon]: root server [%s:%i] is no longer Master, let's use it again even if it could be a stale master, you have been warned!", ptr->server->name, ptr->server->port)));
/* Set the STALE bit for this server in server struct */
server_set_status(ptr->server, SERVER_STALE_STATUS);
} else {
ptr->server->status = ptr->pending_status;
}
}
ptr = ptr->next;
}
}
}
/**
* Set the monitor sampling interval.
*
* @param arg The handle allocated by startMonitor
* @param interval The interval to set in monitor struct, in milliseconds
*/
static void
setInterval(void *arg, size_t interval)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->interval, &interval, sizeof(unsigned long));
}
/**
* Enable/Disable the MySQL Replication Stale Master dectection, allowing a previouvsly detected master to still act as a Master.
* This option must be enabled in order to keep the Master when the replication is stopped or removed from slaves.
* If the replication is still stopped when MaxSclale is restarted no Master will be available.
*
* @param arg The handle allocated by startMonitor
* @param enable To enable it 1, disable it with 0
*/
static void
detectStaleMaster(void *arg, int enable)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->detectStaleMaster, &enable, sizeof(int));
}
static bool mon_status_changed(
MONITOR_SERVERS* mon_srv)
{
bool succp;
if (mon_srv->mon_prev_status != mon_srv->server->status)
{
succp = true;
}
else
{
succp = false;
}
return succp;
}
static bool mon_print_fail_status(
MONITOR_SERVERS* mon_srv)
{
bool succp;
int errcount = mon_srv->mon_err_count;
uint8_t modval;
modval = 1<<(MIN(errcount/10, 7));
if (SERVER_IS_DOWN(mon_srv->server) && errcount%modval == 0)
{
succp = true;
}
else
{
succp = false;
}
return succp;
}
/**
* Set a pending status bit in the monior server
*
* @param server The server to update
* @param bit The bit to clear for the server
*/
static void
monitor_set_pending_status(MONITOR_SERVERS *ptr, int bit)
{
ptr->pending_status |= bit;
}
/**
* Clear a pending status bit in the monior server
*
* @param server The server to update
* @param bit The bit to clear for the server
*/
static void
monitor_clear_pending_status(MONITOR_SERVERS *ptr, int bit)
{
ptr->pending_status &= ~bit;
}
/*******
* This function returns the master server
* from a set of MySQL Multi Master monitored servers
* and returns the root server (that has SERVER_MASTER bit)
* The server is returned even for servers in 'maintenance' mode.
*
* @param handle The monitor handle
* @return The server at root level with SERVER_MASTER bit
*/
static MONITOR_SERVERS *get_current_master(MYSQL_MONITOR *handle) {
MONITOR_SERVERS *ptr;
ptr = handle->databases;
while (ptr)
{
/* The server could be in SERVER_IN_MAINT
* that means SERVER_IS_RUNNING returns 0
* Let's check only for SERVER_IS_DOWN: server is not running
*/
if (SERVER_IS_DOWN(ptr->server)) {
ptr = ptr->next;
continue;
}
if (ptr->server->depth == 0) {
handle->master = ptr;
}
ptr = ptr->next;
}
/*
* Return the root master
*/
if (handle->master != NULL) {
/* If the root master is in MAINT, return NULL */
if (SERVER_IN_MAINT(handle->master->server)) {
return NULL;
} else {
return handle->master;
}
} else {
return NULL;
}
}

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

@ -84,6 +84,7 @@ static void setInterval(void *, size_t);
static void defaultId(void *, unsigned long);
static void replicationHeartbeat(void *, int);
static void detectStaleMaster(void *, int);
static void setNetworkTimeout(void *, int, int);
static bool mon_status_changed(MONITOR_SERVERS* mon_srv);
static bool mon_print_fail_status(MONITOR_SERVERS* mon_srv);
static MONITOR_SERVERS *getServerByNodeId(MONITOR_SERVERS *, long);
@ -103,6 +104,7 @@ static MONITOR_OBJECT MyObject = {
defaultUser,
diagnostics,
setInterval,
NULL,
defaultId,
replicationHeartbeat,
detectStaleMaster,
@ -1204,3 +1206,15 @@ monitor_clear_pending_status(MONITOR_SERVERS *ptr, int bit)
{
ptr->pending_status &= ~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
*/
static void
setNetworkTimeout(void *arg, int type, int value)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
}

1
server/modules/monitor/ndbcluster_mon.c
View File

@ -74,6 +74,7 @@ static MONITOR_OBJECT MyObject = {
NULL,
NULL,
NULL,
NULL,
NULL
};

1
server/modules/protocol/maxscaled.c
View File

@ -270,6 +270,7 @@ int n_connect = 0;
if (client_dcb == NULL)
{
close(so);
return n_connect;
}
client_dcb->fd = so;

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

@ -377,7 +377,7 @@ static int gw_read_backend_event(DCB *dcb) {
dcb,
ERRACT_REPLY_CLIENT,
&succp);
gwbuf_free(errbuf);
ss_dassert(!succp);
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -453,13 +453,15 @@ static int gw_read_backend_event(DCB *dcb) {
0,
"Read from backend failed");
router->handleError(router_instance,
session->router_session,
errbuf,
dcb,
ERRACT_NEW_CONNECTION,
&succp);
router->handleError(
router_instance,
session->router_session,
errbuf,
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
if (!succp)
{
spinlock_acquire(&session->ses_lock);
@ -848,7 +850,8 @@ static int gw_error_backend_event(DCB *dcb)
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
/** There are not required backends available, close session. */
if (!succp) {
spinlock_acquire(&session->ses_lock);
@ -1031,7 +1034,8 @@ gw_backend_hangup(DCB *dcb)
ERRACT_NEW_CONNECTION,
&succp);
/** There are not required backends available, close session. */
gwbuf_free(errbuf);
/** There are no required backends available, close session. */
if (!succp)
{
#if defined(SS_DEBUG)
@ -1039,7 +1043,6 @@ gw_backend_hangup(DCB *dcb)
LOGFILE_ERROR,
"Backend hangup -> closing session.")));
#endif
spinlock_acquire(&session->ses_lock);
session->state = SESSION_STATE_STOPPING;
spinlock_release(&session->ses_lock);
@ -1168,7 +1171,7 @@ static int backend_write_delayqueue(DCB *dcb)
0,
"Failed to write buffered data to back-end server. "
"Buffer was empty or back-end was disconnected during "
"operation. Session will be closed.");
"operation. Attempting to find a new backend.");
router->handleError(router_instance,
rsession,
@ -1176,7 +1179,8 @@ static int backend_write_delayqueue(DCB *dcb)
dcb,
ERRACT_NEW_CONNECTION,
&succp);
gwbuf_free(errbuf);
if (!succp)
{
if (session != NULL)
@ -1327,7 +1331,7 @@ static int gw_change_user(
/** Create an incoming event for backend DCB */
poll_add_epollin_event_to_dcb(backend, gwbuf_clone(buf));
gwbuf_free(buf);
rv = 0;
rv = 1;
} else {
rv = gw_send_change_user_to_backend(database, username, client_sha1, backend_protocol);
/*

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

@ -430,7 +430,7 @@ static int gw_mysql_do_authentication(DCB *dcb, GWBUF *queue) {
connect_with_db =
GW_MYSQL_CAPABILITIES_CONNECT_WITH_DB & gw_mysql_get_byte4(
&protocol->client_capabilities);
(uint32_t *)&protocol->client_capabilities);
/*
compress =
GW_MYSQL_CAPABILITIES_COMPRESS & gw_mysql_get_byte4(
@ -859,6 +859,7 @@ int gw_read_client_event(
dcb,
ERRACT_REPLY_CLIENT,
&succp);
gwbuf_free(errbuf);
ss_dassert(!succp);
dcb_close(dcb);
@ -1017,7 +1018,7 @@ int gw_MySQLListener(
errno,
strerror(errno));
fprintf(stderr, "* Can't bind to %s\n\n", config_bind);
close(l_so);
return 0;
}
@ -1039,13 +1040,14 @@ int gw_MySQLListener(
errno,
strerror(errno));
fprintf(stderr, "* Can't bind to %s\n\n", config_bind);
close(l_so);
return 0;
}
break;
default:
fprintf(stderr, "* Socket Family %i not supported\n", current_addr->sa_family);
close(l_so);
return 0;
}

1
server/modules/protocol/telnetd.c
View File

@ -298,6 +298,7 @@ int n_connect = 0;
if (client_dcb == NULL)
{
close(so);
return n_connect;
}
client_dcb->fd = so;

3
server/modules/routing/CMakeLists.txt
View File

@ -16,3 +16,6 @@ install(TARGETS cli DESTINATION modules)
add_subdirectory(readwritesplit)
if(BUILD_TESTS)
add_subdirectory(test)
endif()

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

@ -631,7 +631,7 @@ char *ptr, *lptr;
else
{
i++;
if (i >= MAXARGS)
if (i >= MAXARGS-1)
break;
args[i] = ptr + 1;
}

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

@ -700,8 +700,8 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
rc = 0;
goto return_rc;
}
switch(mysql_command) {
switch(mysql_command) {
case MYSQL_COM_CHANGE_USER:
rc = backend_dcb->func.auth(
backend_dcb,
@ -714,7 +714,7 @@ routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
rc = backend_dcb->func.write(backend_dcb, queue);
break;
}
CHK_PROTOCOL(((MySQLProtocol*)backend_dcb->protocol));
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -815,21 +815,33 @@ clientReply(
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
*
*/
static void
handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp)
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp)
{
DCB *client = NULL;
DCB *client_dcb;
SESSION *session = backend_dcb->session;
client = session->client;
session_state_t sesstate;
spinlock_acquire(&session->ses_lock);
sesstate = session->state;
client_dcb = session->client;
spinlock_release(&session->ses_lock);
ss_dassert(client_dcb != NULL);
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));
}
/** false because connection is not available anymore */
*succp = false;
ss_dassert(client != NULL);
}
/** to be inline'd */

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

@ -1356,7 +1356,7 @@ static route_target_t get_route_target (
hint = hint->next;
} /*< while (hint != NULL) */
/** If nothing matches then choose the master */
if ((target & (TARGET_ALL|TARGET_SLAVE|TARGET_MASTER)) == target)
if ((target & (TARGET_ALL|TARGET_SLAVE|TARGET_MASTER)) == 0)
{
target = TARGET_MASTER;
}
@ -1406,7 +1406,7 @@ void check_drop_tmp_table(
{
int tsize = 0, klen = 0,i;
char** tbl;
char** tbl = NULL;
char *hkey,*dbname;
MYSQL_session* data;
@ -1425,29 +1425,31 @@ void check_drop_tmp_table(
if (is_drop_table_query(querybuf))
{
tbl = skygw_get_table_names(querybuf,&tsize,false);
for(i = 0; i<tsize; i++)
{
klen = strlen(dbname) + strlen(tbl[i]) + 2;
hkey = calloc(klen,sizeof(char));
strcpy(hkey,dbname);
strcat(hkey,".");
strcat(hkey,tbl[i]);
if(tbl != NULL){
for(i = 0; i<tsize; i++)
{
klen = strlen(dbname) + strlen(tbl[i]) + 2;
hkey = calloc(klen,sizeof(char));
strcpy(hkey,dbname);
strcat(hkey,".");
strcat(hkey,tbl[i]);
if (rses_prop_tmp &&
rses_prop_tmp->rses_prop_data.temp_tables)
{
if (hashtable_delete(rses_prop_tmp->rses_prop_data.temp_tables,
(void *)hkey))
{
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
"Temporary table dropped: %s",hkey)));
}
}
free(tbl[i]);
free(hkey);
}
free(tbl);
if (rses_prop_tmp &&
rses_prop_tmp->rses_prop_data.temp_tables)
{
if (hashtable_delete(rses_prop_tmp->rses_prop_data.temp_tables,
(void *)hkey))
{
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
"Temporary table dropped: %s",hkey)));
}
}
free(tbl[i]);
free(hkey);
}
free(tbl);
}
}
}
@ -1468,7 +1470,7 @@ skygw_query_type_t is_read_tmp_table(
bool target_tmp_table = false;
int tsize = 0, klen = 0,i;
char** tbl;
char** tbl = NULL;
char *hkey,*dbname;
MYSQL_session* data;
@ -1493,7 +1495,7 @@ skygw_query_type_t is_read_tmp_table(
{
tbl = skygw_get_table_names(querybuf,&tsize,false);
if (tsize > 0)
if (tbl != NULL && tsize > 0)
{
/** Query targets at least one table */
for(i = 0; i<tsize && !target_tmp_table && tbl[i]; i++)
@ -1520,14 +1522,21 @@ skygw_query_type_t is_read_tmp_table(
}
free(hkey);
free(tbl[i]);
}
free(tbl);
}
}
for(i = 0; i<tsize;i++)
{
free(tbl[i]);
}
if(tbl != NULL){
free(tbl);
}
return qtype;
return qtype;
}
/**
@ -1535,7 +1544,6 @@ skygw_query_type_t is_read_tmp_table(
* the database and table name, create a hashvalue and
* add it to the router client session's property. If property
* doesn't exist then create it first.
*
* @param instance Router instance
* @param router_session Router client session
* @param querybuf GWBUF containing the query
@ -1711,27 +1719,41 @@ static int routeQuery(
LOGIF(LE,
(skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Failed to route %s:%s:\"%s\" to "
"backend server. %s.",
"Error: Can't route %s:%s:\"%s\" to "
"backend server. Router is closed.",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(query_str == NULL ? "(empty)" : query_str),
(rses_is_closed ? "Router was closed" :
"Router has no backend servers where to "
"route to"))));
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
}
ret = 0;
goto retblock;
}
/** Read stored master DCB pointer */
if ((master_dcb = router_cli_ses->rses_master_ref->bref_dcb) == NULL)
{
char* query_str = modutil_get_query(querybuf);
CHK_DCB(master_dcb);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Can't route %s:%s:\"%s\" to "
"backend server. Session doesn't have a Master "
"node",
STRPACKETTYPE(packet_type),
STRQTYPE(qtype),
(query_str == NULL ? "(empty)" : query_str))));
free(query_str);
ret = 0;
goto retblock;
}
/** If buffer is not contiguous, make it such */
if (querybuf->next != NULL)
{
querybuf = gwbuf_make_contiguous(querybuf);
}
/** Read stored master DCB pointer */
master_dcb = router_cli_ses->rses_master_ref->bref_dcb;
CHK_DCB(master_dcb);
switch(packet_type) {
case MYSQL_COM_QUIT: /*< 1 QUIT will close all sessions */
case MYSQL_COM_INIT_DB: /*< 2 DDL must go to the master */
@ -1891,6 +1913,10 @@ static int routeQuery(
/** 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 <")));
ret = 0;
goto retblock;
}
/**
@ -2023,13 +2049,13 @@ static int routeQuery(
NULL,
MAX_RLAG_UNDEFINED);
if (succp && (master_dcb == NULL || master_dcb == curr_master_dcb))
if (succp && master_dcb == curr_master_dcb)
{
atomic_add(&inst->stats.n_master, 1);
target_dcb = master_dcb;
}
else
{
{
if (succp && master_dcb != curr_master_dcb)
{
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
@ -2050,9 +2076,9 @@ static int routeQuery(
}
succp = false;
ret = 0;
}
}
}
}
if (succp) /*< Have DCB of the target backend */
{
backend_ref_t* bref;
@ -2631,7 +2657,12 @@ 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. This is slave failure case */
/**
* 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.
*/
if (*p_master_ref != NULL &&
BREF_IS_IN_USE((*p_master_ref)))
{
@ -4023,7 +4054,7 @@ static void rwsplit_process_router_options(
*
* @param instance The router instance
* @param router_session The router session
* @param message The error message to reply
* @param errmsgbuf The error message to reply
* @param backend_dcb The backend DCB
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
* @param succp Result of action.
@ -4065,6 +4096,22 @@ static void handleError (
return;
}
if (rses->rses_master_ref->bref_dcb == backend_dcb)
{
/** Master failed, can't recover */
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Master node have failed. "
"Session will be closed.")));
*succp = false;
rses_end_locked_router_action(rses);
return;
}
/**
* This is called in hope of getting replacement for
* failed slave(s).
*/
*succp = handle_error_new_connection(inst,
rses,
backend_dcb,
@ -4088,114 +4135,115 @@ static void handleError (
static void handle_error_reply_client(
SESSION* ses,
GWBUF* errmsg)
SESSION* ses,
GWBUF* errmsg)
{
session_state_t sesstate;
DCB* client_dcb;
spinlock_acquire(&ses->ses_lock);
sesstate = ses->state;
client_dcb = ses->client;
spinlock_release(&ses->ses_lock);
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, errmsg);
}
else
{
while ((errmsg=gwbuf_consume(errmsg, GWBUF_LENGTH(errmsg))) != NULL)
;
}
session_state_t sesstate;
DCB* client_dcb;
spinlock_acquire(&ses->ses_lock);
sesstate = ses->state;
client_dcb = ses->client;
spinlock_release(&ses->ses_lock);
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
client_dcb->func.write(client_dcb, gwbuf_clone(errmsg));
}
}
/**
* This must be called with router lock
* Check if there is backend reference pointing at failed DCB, and reset its
* flags. Then clear DCB's callback and finally : try to find replacement(s)
* for failed slave(s).
*
* This must be called with router lock.
*
* @param inst router instance
* @param rses router client session
* @param dcb failed DCB
* @param errmsg error message which is sent to client if it is waiting
*
* @return true if there are enough backend connections to continue, false if not
*/
static bool handle_error_new_connection(
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
DCB* backend_dcb,
GWBUF* errmsg)
ROUTER_INSTANCE* inst,
ROUTER_CLIENT_SES* rses,
DCB* backend_dcb,
GWBUF* errmsg)
{
SESSION* ses;
int router_nservers;
int max_nslaves;
int max_slave_rlag;
backend_ref_t* bref;
bool succp;
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
ses = backend_dcb->session;
CHK_SESSION(ses);
bref = get_bref_from_dcb(rses, backend_dcb);
/** failed DCB has already been replaced */
if (bref == NULL)
{
succp = true;
goto return_succp;
}
/**
* Error handler is already called for this DCB because
* it's not polling anymore. It can be assumed that
* it succeed because rses isn't closed.
*/
if (backend_dcb->state != DCB_STATE_POLLING)
{
succp = true;
goto return_succp;
}
CHK_BACKEND_REF(bref);
if (BREF_IS_WAITING_RESULT(bref))
{
DCB* client_dcb;
client_dcb = ses->client;
client_dcb->func.write(client_dcb, errmsg);
bref_clear_state(bref, BREF_WAITING_RESULT);
}
else
{
while ((errmsg=gwbuf_consume(errmsg, GWBUF_LENGTH(errmsg))) != NULL)
;
}
bref_clear_state(bref, BREF_IN_USE);
bref_set_state(bref, BREF_CLOSED);
/**
* Remove callback because this DCB won't be used
* unless it is reconnected later, and then the callback
* is set again.
*/
dcb_remove_callback(backend_dcb,
DCB_REASON_NOT_RESPONDING,
&router_handle_state_switch,
(void *)bref);
router_nservers = router_get_servercount(inst);
max_nslaves = rses_get_max_slavecount(rses, router_nservers);
max_slave_rlag = rses_get_max_replication_lag(rses);
/**
* Try to get replacement slave or at least the minimum
* number of slave connections for router session.
*/
succp = select_connect_backend_servers(
&rses->rses_master_ref,
rses->rses_backend_ref,
router_nservers,
max_nslaves,
max_slave_rlag,
rses->rses_config.rw_slave_select_criteria,
ses,
inst);
SESSION* ses;
int router_nservers;
int max_nslaves;
int max_slave_rlag;
backend_ref_t* bref;
bool succp;
ss_dassert(SPINLOCK_IS_LOCKED(&rses->rses_lock));
ses = backend_dcb->session;
CHK_SESSION(ses);
/**
* If bref == NULL it has been replaced already with another one.
*/
if ((bref = get_bref_from_dcb(rses, backend_dcb)) == NULL)
{
succp = true;
goto return_succp;
}
CHK_BACKEND_REF(bref);
if (BREF_IS_WAITING_RESULT(bref))
{
DCB* client_dcb;
client_dcb = ses->client;
client_dcb->func.write(client_dcb, gwbuf_clone(errmsg));
bref_clear_state(bref, BREF_WAITING_RESULT);
}
bref_clear_state(bref, BREF_IN_USE);
bref_set_state(bref, BREF_CLOSED);
/**
* Error handler is already called for this DCB because
* it's not polling anymore. It can be assumed that
* it succeed because rses isn't closed.
*/
if (backend_dcb->state != DCB_STATE_POLLING)
{
succp = true;
goto return_succp;
}
/**
* Remove callback because this DCB won't be used
* unless it is reconnected later, and then the callback
* is set again.
*/
dcb_remove_callback(backend_dcb,
DCB_REASON_NOT_RESPONDING,
&router_handle_state_switch,
(void *)bref);
router_nservers = router_get_servercount(inst);
max_nslaves = rses_get_max_slavecount(rses, router_nservers);
max_slave_rlag = rses_get_max_replication_lag(rses);
/**
* Try to get replacement slave or at least the minimum
* number of slave connections for router session.
*/
succp = select_connect_backend_servers(
&rses->rses_master_ref,
rses->rses_backend_ref,
router_nservers,
max_nslaves,
max_slave_rlag,
rses->rses_config.rw_slave_select_criteria,
ses,
inst);
return_succp:
return succp;
return succp;
}
@ -4291,8 +4339,8 @@ static bool have_enough_servers(
}
else
{
double pct = (*p_rses)->rses_config.rw_max_slave_conn_percent/100;
double nservers = (double)router_nsrv*pct;
int pct = (*p_rses)->rses_config.rw_max_slave_conn_percent/100;
int nservers = router_nsrv*pct;
if ((*p_rses)->rses_config.rw_max_slave_conn_count < min_nsrv)
{
@ -4311,11 +4359,11 @@ static bool have_enough_servers(
LOGFILE_ERROR,
"Error : Unable to start %s service. There are "
"too few backend servers configured in "
"MaxScale.cnf. Found %d%% when at least %.0f%% "
"MaxScale.cnf. Found %d%% when at least %d%% "
"would be required.",
router->service->name,
(*p_rses)->rses_config.rw_max_slave_conn_percent,
min_nsrv/(((double)router_nsrv)/100))));
min_nsrv/(router_nsrv/100))));
}
}
free(*p_rses);
@ -4378,7 +4426,14 @@ static int rses_get_max_replication_lag(
return conf_max_rlag;
}
/**
* Finds out if there is a backend reference pointing at the DCB given as
* parameter.
* @param rses router client session
* @param dcb DCB
*
* @return backend reference pointer if succeed or NULL
*/
static backend_ref_t* get_bref_from_dcb(
ROUTER_CLIENT_SES* rses,
DCB* dcb)

1
server/modules/routing/readwritesplit/test/CMakeLists.txt
View File

@ -1,2 +1,3 @@
add_test(NAME ReadWriteSplitTest COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/rwsplit.sh testrwsplit.log ${TEST_HOST} ${TEST_PORT_RW} ${TEST_MASTER_ID} ${TEST_USER} ${TEST_PASSWORD} ${CMAKE_CURRENT_SOURCE_DIR})
add_test(NAME ReadWriteSplitLoginTest COMMAND $<TARGET_FILE:testconnect> 10000 ${TEST_HOST} ${MASTER_PORT} ${TEST_HOST} ${TEST_PORT_RW})
add_subdirectory(test_hints)

3
server/modules/routing/test/CMakeLists.txt Normal file
View File

@ -0,0 +1,3 @@
add_executable(testconnect testconnect.c)
target_link_libraries(testconnect mysql)
add_test(NAME ReadConnRouterLoginTest COMMAND $<TARGET_FILE:testconnect> 10000 ${TEST_HOST} ${MASTER_PORT} ${TEST_HOST} ${TEST_PORT})

79
server/modules/routing/test/testconnect.c Normal file
View File

@ -0,0 +1,79 @@
#include <my_config.h>
#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
int main(int argc, char** argv)
{
MYSQL* server;
char *host;
unsigned int port;
int rval, iterations,i;
clock_t begin,end;
double baseline,test;
if(argc < 6){
fprintf(stderr,"Usage: %s <iterations> <baseline host> <baseline port> <test host> <test port>\n",argv[0]);
return 1;
}
iterations = atoi(argv[1]);
host = strdup(argv[2]);
port = atoi(argv[3]);
rval = 0;
/**Testing direct connection to master*/
printf("Connecting to MySQL server through %s:%d.\n",host,port);
begin = clock();
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));
rval = 1;
break;
}
mysql_close(server);
}
end = clock();
baseline = (double)(end - begin)/CLOCKS_PER_SEC;
free(host);
host = strdup(argv[4]);
port = atoi(argv[5]);
/**Testing connection to master through MaxScale*/
printf("Connecting to MySQL server through %s:%d.\n",host,port);
begin = clock();
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){
rval = 1;
fprintf(stderr, "Failed to connect to database: Error: %s\n",
mysql_error(server));
break;
}
mysql_close(server);
}
end = clock();
test = (double)(end - begin)/CLOCKS_PER_SEC;
printf("CPU time used in seconds:\nDirect connection: %f\nThrough MaxScale: %f\n",baseline,test);
free(host);
return rval;
}