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

Merge branch 'release-1.0GA' into blr

Browse Source
This commit is contained in:
Mark Riddoch 2014-12-08 09:21:38 +00:00
commit 8fed527ac9
47 changed files with 914 additions and 439 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@ -123,7 +123,7 @@ install(DIRECTORY DESTINATION log)
# See if we are on a RPM-capable or DEB-capable system
find_program(RPMBUILD rpmbuild)
find_program(DEBBUILD dpkg-buildpackage)
set(CPACK_GENERATOR "TGZ")
if(NOT ( ${RPMBUILD} STREQUAL "RPMBUILD-NOTFOUND" ) )
message(STATUS "Generating RPM packages")
set(CPACK_GENERATOR "${CPACK_GENERATOR};RPM")

BIN
Documentation/Administration Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/Filter Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/Galera Cluster Connection Routing Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/Galera Cluster Read-Write Splitting Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/Getting Started With MaxScale.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/MaxAdmin The MaxScale Administration And Monitoring Client.pdf Normal file → Executable file
View File

Binary file not shown.

BIN
Documentation/MaxScale Configuration And Usage Scenarios.pdf
View File

Binary file not shown.

BIN
Documentation/MaxScale Debug And Diagnostic Support.pdf Normal file → Executable file
View File

Binary file not shown.

BIN
Documentation/MaxScale HA with Corosync-Pacemaker.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/MaxScale Known Limitiations.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/MySQL Replication Connection Routing Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/MySQL Replication Read-Write Splitting Tutorial.pdf Executable file
View File

Binary file not shown.

BIN
Documentation/filters/Regex Filter.pdf Normal file → Executable file
View File

Binary file not shown.

BIN
Documentation/filters/Tee Filter.pdf Normal file → Executable file
View File

Binary file not shown.

BIN
Documentation/filters/Top Filter.pdf Normal file → Executable file
View File

Binary file not shown.

3
log_manager/CMakeLists.txt
View File

@ -1,3 +1,6 @@
if(LOG_DEBUG)
add_definitions(-DSS_LOG_DEBUG)
endif()
add_library(log_manager SHARED log_manager.cc)
target_link_libraries(log_manager pthread aio stdc++)
install(TARGETS log_manager DESTINATION lib)

207
log_manager/log_manager.cc
View File

@ -94,7 +94,9 @@ char* syslog_ident_str = NULL;
*/
static int lmlock;
static logmanager_t* lm;
static bool flushall_flag;
static bool flushall_started_flag;
static bool flushall_done_flag;
/** Writer thread structure */
struct filewriter_st {
@ -286,12 +288,14 @@ static char* add_slash(char* str);
static bool check_file_and_path(
char* filename,
bool* writable);
bool* writable,
bool do_log);
static bool file_is_symlink(char* filename);
static int skygw_log_disable_raw(logfile_id_t id, bool emergency); /*< no locking */
static int find_last_seqno(strpart_t* parts, int seqno, int seqnoidx);
void flushall_logfiles(bool flush);
bool thr_flushall_check();
const char* get_suffix_default(void)
{
@ -441,17 +445,6 @@ static bool logmanager_init_nomutex(
return_succp:
if (err != 0)
{
if (lm != NULL)
{
if (lm->lm_clientmes != NULL)
{
skygw_message_done(lm->lm_clientmes);
}
if (lm->lm_logmes != NULL)
{
skygw_message_done(lm->lm_logmes);
}
}
/** This releases memory of all created objects */
logmanager_done_nomutex();
fprintf(stderr, "*\n* Error : Initializing log manager failed.\n*\n");
@ -729,10 +722,10 @@ static int logmanager_write_log(
/** Length of session id */
int sesid_str_len;
/** 2 braces and 2 spaces */
/** 2 braces, 2 spaces and terminating char */
if (id == LOGFILE_TRACE && tls_log_info.li_sesid != 0)
{
sesid_str_len = 2+2+get_decimal_len(tls_log_info.li_sesid);
sesid_str_len = 2+2+get_decimal_len(tls_log_info.li_sesid)+1;
}
else
{
@ -741,13 +734,13 @@ static int logmanager_write_log(
timestamp_len = get_timestamp_len();
/** Find out how much can be safely written with current block size */
if (timestamp_len-1+sesid_str_len+str_len > lf->lf_buf_size)
if (timestamp_len-1+MAX(sesid_str_len-1,0)+str_len > lf->lf_buf_size)
{
safe_str_len = lf->lf_buf_size;
}
else
{
safe_str_len = timestamp_len-1+sesid_str_len+str_len;
safe_str_len = timestamp_len-1+MAX(sesid_str_len-1,0)+str_len;
}
/**
* Seek write position and register to block buffer.
@ -809,6 +802,7 @@ static int logmanager_write_log(
sesid_str_len,
"[%lu] ",
tls_log_info.li_sesid);
sesid_str_len -= 1; /*< don't calculate terminating char anymore */
}
/**
* Write next string to overwrite terminating null character
@ -1099,7 +1093,7 @@ static char* blockbuf_get_writepos(
simple_mutex_unlock(&bb->bb_mutex);
simple_mutex_lock(&bb_list->mlist_mutex, true);
node = bb_list->mlist_first;
}
else
{
@ -2052,7 +2046,7 @@ static bool logfile_create(
* If file exists but is different type, create fails and
* new, increased sequence number is added to file name.
*/
if (check_file_and_path(lf->lf_full_file_name, &writable))
if (check_file_and_path(lf->lf_full_file_name, &writable, true))
{
/** Found similarly named file which isn't writable */
if (!writable || file_is_symlink(lf->lf_full_file_name))
@ -2076,13 +2070,11 @@ static bool logfile_create(
if (store_shmem)
{
if (check_file_and_path(lf->lf_full_file_name, &writable))
if (check_file_and_path(lf->lf_full_link_name, &writable, true))
{
/** Found similarly named file which isn't writable */
if (!writable ||
file_is_symlink(lf->lf_full_file_name))
/** Found similarly named link which isn't writable */
if (!writable)
{
unlink(lf->lf_full_file_name);
nameconflicts = true;
}
}
@ -2213,7 +2205,6 @@ return_succp:
*
* @return Pointer to filename, of NULL if failed.
*
*
*/
static char* form_full_file_name(
strpart_t* parts,
@ -2230,9 +2221,21 @@ static char* form_full_file_name(
if (lf->lf_name_seqno != -1)
{
lf->lf_name_seqno = find_last_seqno(parts,
lf->lf_name_seqno,
seqnoidx);
int file_sn;
int link_sn = 0;
char* tmp = parts[0].sp_string;
file_sn = find_last_seqno(parts, lf->lf_name_seqno, seqnoidx);
if (lf->lf_linkpath != NULL)
{
tmp = parts[0].sp_string;
parts[0].sp_string = lf->lf_linkpath;
link_sn = find_last_seqno(parts, lf->lf_name_seqno, seqnoidx);
parts[0].sp_string = tmp;
}
lf->lf_name_seqno = MAX(file_sn, link_sn);
seqno = lf->lf_name_seqno;
s = UINTLEN(seqno);
seqnostr = (char *)malloc((int)s+1);
@ -2353,7 +2356,8 @@ static char* add_slash(
*/
static bool check_file_and_path(
char* filename,
bool* writable)
bool* writable,
bool do_log)
{
int fd;
bool exists;
@ -2381,11 +2385,23 @@ static bool check_file_and_path(
if (fd == -1)
{
fprintf(stderr,
"*\n* Error : Can't access %s due "
"to %s.\n",
filename,
strerror(errno));
if (do_log && file_is_symlink(filename))
{
fprintf(stderr,
"*\n* Error : Can't access "
"file pointed to by %s due "
"to %s.\n",
filename,
strerror(errno));
}
else if (do_log)
{
fprintf(stderr,
"*\n* Error : Can't access %s due "
"to %s.\n",
filename,
strerror(errno));
}
if (writable)
{
*writable = false;
@ -2402,11 +2418,24 @@ static bool check_file_and_path(
}
else
{
fprintf(stderr,
"*\n* Error : Can't write to "
"%s due to %s.\n",
filename,
strerror(errno));
if (do_log &&
file_is_symlink(filename))
{
fprintf(stderr,
"*\n* Error : Can't write to "
"file pointed to by %s due to "
"%s.\n",
filename,
strerror(errno));
}
else if (do_log)
{
fprintf(stderr,
"*\n* Error : Can't write to "
"%s due to %s.\n",
filename,
strerror(errno));
}
*writable = false;
}
}
@ -2416,10 +2445,21 @@ static bool check_file_and_path(
}
else
{
fprintf(stderr,
"*\n* Error : Can't access %s due to %s.\n",
filename,
strerror(errno));
if (do_log && file_is_symlink(filename))
{
fprintf(stderr,
"*\n* Error : Can't access the file "
"pointed to by %s due to %s.\n",
filename,
strerror(errno));
}
else if (do_log)
{
fprintf(stderr,
"*\n* Error : Can't access %s due to %s.\n",
filename,
strerror(errno));
}
exists = false;
if (writable)
@ -2571,7 +2611,7 @@ static bool logfile_init(
logfile_free_memory(logfile);
goto return_with_succp;
}
#if defined(SS_DEBUG)
if (store_shmem)
{
fprintf(stderr, "%s\t: %s->%s\n",
@ -2585,7 +2625,6 @@ static bool logfile_init(
STRLOGNAME(logfile_id),
logfile->lf_full_file_name);
}
#endif
succp = true;
logfile->lf_state = RUN;
CHK_LOGFILE(logfile);
@ -2801,13 +2840,15 @@ static void* thr_filewriter_fun(
int i;
blockbuf_state_t flush_blockbuf; /**< flush single block buffer. */
bool flush_logfile; /**< flush logfile */
bool flushall_logfiles;/**< flush all logfiles */
bool do_flushall = false;
bool rotate_logfile; /*< close current and open new file */
size_t vn1;
size_t vn2;
thr = (skygw_thread_t *)data;
fwr = (filewriter_t *)skygw_thread_get_data(thr);
flushall_logfiles(false);
CHK_FILEWRITER(fwr);
ss_debug(skygw_thread_set_state(thr, THR_RUNNING));
@ -2820,8 +2861,9 @@ static void* thr_filewriter_fun(
* Reset message to avoid redundant calls.
*/
skygw_message_wait(fwr->fwr_logmes);
flushall_logfiles = skygw_thread_must_exit(thr);
if(skygw_thread_must_exit(thr)){
flushall_logfiles(true);
}
/** Process all logfiles which have buffered writes. */
for (i=LOGFILE_FIRST; i<=LOGFILE_LAST; i <<= 1)
@ -2830,6 +2872,10 @@ static void* thr_filewriter_fun(
/**
* Get file pointer of current logfile.
*/
do_flushall = thr_flushall_check();
file = fwr->fwr_file[i];
lf = &lm->lm_logfile[(logfile_id_t)i];
@ -2900,7 +2946,7 @@ static void* thr_filewriter_fun(
if (bb->bb_buf_used != 0 &&
(flush_blockbuf == BB_FULL ||
flush_logfile ||
flushall_logfiles))
do_flushall))
{
/**
* buffer is at least half-full
@ -2919,7 +2965,7 @@ static void* thr_filewriter_fun(
(void *)bb->bb_buf,
bb->bb_buf_used,
(flush_logfile ||
flushall_logfiles));
do_flushall));
if (err)
{
fprintf(stderr,
@ -2966,13 +3012,28 @@ static void* thr_filewriter_fun(
* Loop is restarted to ensure that all logfiles are
* flushed.
*/
if (!flushall_logfiles && skygw_thread_must_exit(thr))
if(flushall_started_flag){
flushall_started_flag = false;
flushall_done_flag = true;
i = LOGFILE_FIRST;
goto retry_flush_on_exit;
}
if (!thr_flushall_check() && skygw_thread_must_exit(thr))
{
flushall_logfiles = true;
flushall_logfiles(true);
i = LOGFILE_FIRST;
goto retry_flush_on_exit;
}
} /* for */
}/* for */
if(flushall_done_flag){
flushall_done_flag = false;
flushall_logfiles(false);
skygw_message_send(fwr->fwr_clientmes);
}
} /* while (!skygw_thread_must_exit) */
ss_debug(skygw_thread_set_state(thr, THR_STOPPED));
@ -3050,9 +3111,9 @@ static int find_last_seqno(
{
if (snstr != NULL && i == seqnoidx)
{
strcat(filename, snstr); /*< add sequence number */
strncat(filename, snstr, NAME_MAX - 1); /*< add sequence number */
}
strcat(filename, p->sp_string);
strncat(filename, p->sp_string, NAME_MAX - 1);
if (p->sp_next == NULL)
{
@ -3060,7 +3121,7 @@ static int find_last_seqno(
}
}
if (check_file_and_path(filename, NULL))
if (check_file_and_path(filename, NULL, false))
{
seqno++;
}
@ -3072,4 +3133,34 @@ static int find_last_seqno(
free(snstr);
return seqno;
}
}
bool thr_flushall_check()
{
bool rval = false;
simple_mutex_lock(&lm->lm_mutex,true);
rval = flushall_flag;
if(rval && !flushall_started_flag && !flushall_done_flag){
flushall_started_flag = true;
}
simple_mutex_unlock(&lm->lm_mutex);
return rval;
}
void flushall_logfiles(bool flush)
{
simple_mutex_lock(&lm->lm_mutex,true);
flushall_flag = flush;
simple_mutex_unlock(&lm->lm_mutex);
}
/**
* Flush all log files synchronously
*/
void skygw_log_sync_all(void)
{
skygw_log_write(LOGFILE_TRACE,"Starting log flushing to disk.");
flushall_logfiles(true);
skygw_message_send(lm->lm_logmes);
skygw_message_wait(lm->lm_clientmes);
}

2
macros.cmake
View File

@ -123,7 +123,7 @@ debugmsg("Search returned: ${MYSQL_DIR_LOC}")
if(${ERRMSG_FILE} MATCHES "NOTFOUND")
message(FATAL_ERROR "Fatal Error: The errmsg.sys file was not found at ${ERRMSG}")
else()
message(STATUS "Using errmsg.sys found at: ${ERRMSG_FILE}")
message(STATUS "Using custom errmsg.sys found at: ${ERRMSG_FILE}")
endif()
else()
find_file(ERRMSG_FILE errmsg.sys PATHS /usr/share/mysql /usr/local/share/mysql PATH_SUFFIXES english)

77
query_classifier/query_classifier.cc
View File

@ -362,50 +362,29 @@ static bool create_parse_tree(
Parser_state parser_state;
bool failp = FALSE;
const char* virtual_db = "skygw_virtual";
#if defined(SS_DEBUG_EXTRA)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"[readwritesplit:create_parse_tree] 1.")));
#endif
if (parser_state.init(thd, thd->query(), thd->query_length())) {
if (parser_state.init(thd, thd->query(), thd->query_length()))
{
failp = TRUE;
goto return_here;
}
#if defined(SS_DEBUG_EXTRA)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"[readwritesplit:create_parse_tree] 2.")));
#endif
mysql_reset_thd_for_next_command(thd);
#if defined(SS_DEBUG_EXTRA)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"[readwritesplit:create_parse_tree] 3.")));
#endif
/**
* Set some database to thd so that parsing won't fail because of
* missing database. Then parse.
*/
failp = thd->set_db(virtual_db, strlen(virtual_db));
#if defined(SS_DEBUG_EXTRA)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"[readwritesplit:create_parse_tree] 4.")));
#endif
if (failp) {
if (failp)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to set database in thread context.")));
}
failp = parse_sql(thd, &parser_state, NULL);
#if defined(SS_DEBUG_EXTRA)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"[readwritesplit:create_parse_tree] 5.")));
#endif
if (failp) {
if (failp)
{
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [readwritesplit:create_parse_tree] failed to "
@ -417,16 +396,14 @@ return_here:
}
/**
* @node Set new query type if new is more restrictive than old.
* Set new query type if new is more restrictive than old.
*
* Parameters:
* @param qtype - <usage>
* <description>
* @param qtype Existing type
*
* @param new_type - <usage>
* <description>
* @param new_type New query type
*
* @return
* @return Query type as an unsigned int value which must be casted to qtype.
*
*
* @details The implementation relies on that enumerated values correspond
@ -443,13 +420,11 @@ static u_int32_t set_query_type(
}
/**
* @node Detect query type, read-only, write, or session update
* Detect query type by examining parsed representation of it.
*
* Parameters:
* @param thd - <usage>
* <description>
* @param thd MariaDB thread context.
*
* @return
* @return Copy of query type value.
*
*
* @details Query type is deduced by checking for certain properties
@ -474,7 +449,7 @@ static skygw_query_type_t resolve_query_type(
* all write operations to all nodes.
*/
#if defined(NOT_IN_USE)
bool force_data_modify_op_replication;
bool force_data_modify_op_replication;
force_data_modify_op_replication = FALSE;
#endif /* NOT_IN_USE */
ss_info_dassert(thd != NULL, ("thd is NULL\n"));
@ -619,6 +594,7 @@ static skygw_query_type_t resolve_query_type(
break;
case SQLCOM_SELECT:
case SQLCOM_SHOW_SLAVE_STAT:
type |= QUERY_TYPE_READ;
break;
@ -867,6 +843,11 @@ return_qtype:
* Checks if statement causes implicit COMMIT.
* autocommit_stmt gets values 1, 0 or -1 if stmt is enable, disable or
* something else than autocommit.
*
* @param lex Parse tree
* @param autocommit_stmt memory address for autocommit status
*
* @return true if statement causes implicit commit and false otherwise
*/
static bool skygw_stmt_causes_implicit_commit(
LEX* lex,
@ -896,7 +877,7 @@ static bool skygw_stmt_causes_implicit_commit(
}
else
{
succp =false;
succp = false;
}
break;
default:
@ -912,7 +893,9 @@ return_succp:
* Finds out if stmt is SET autocommit
* and if the new value matches with the enable_cmd argument.
*
* Returns 1, 0, or -1 if command was:
* @param lex parse tree
*
* @return 1, 0, or -1 if command was:
* enable, disable, or not autocommit, respectively.
*/
static int is_autocommit_stmt(
@ -983,9 +966,11 @@ char* skygw_query_classifier_get_stmtname(
}
/**
*Returns the LEX struct of the parsed GWBUF
*@param The parsed GWBUF
*@return Pointer to the LEX struct or NULL if an error occurred or the query was not parsed
* Get the parse tree from parsed querybuf.
* @param querybuf The parsed GWBUF
*
* @return Pointer to the LEX struct or NULL if an error occurred or the query
* was not parsed
*/
LEX* get_lex(GWBUF* querybuf)
{
@ -1194,7 +1179,7 @@ bool is_drop_table_query(GWBUF* querybuf)
lex->sql_command == SQLCOM_DROP_TABLE;
}
/*
/**
* Replace user-provided literals with question marks. Return a copy of the
* querystr with replacements.
*

338
server/MaxScale_template.cnf
View File

@ -1,89 +1,212 @@
## Example MaxScale.cnf configuration file
#
# Example MaxScale.cnf configuration file
# Number of worker threads in MaxScale
#
#
#
# Number of server threads
# Valid options are:
# threads=<number of threads>
#
[maxscale]
threads=1
threads=4
# Define a monitor that can be used to determine the state and role of
## Define a monitor that can be used to determine the state and role of
# the servers.
#
# Valid options for all monitors are:
# Currently valid options for all monitors are:
#
# module=<name of module to load>
# servers=<server name>,<server name>,...
# user =<user name - must have slave replication and
# slave client privileges>
# passwd=<password of the above user, plain text currently>
# monitor_interval=<sampling interval in milliseconds,
# default value is 10000>
# module=[mysqlmon|galeramon]
#
# List of server names which are being monitored
#
# servers=<server name 1>,<server name 2>,...,<server name N>
#
# Username for monitor queries, need slave replication and slave client privileges
# Password in plain text format, and monitor's sampling interval in milliseconds.
#
# user=<username>
# passwd=<plain txt password>
# monitor_interval=<sampling interval in milliseconds> (default 10000)
#
# Timeouts for monitor operations in backend servers - optional.
#
# backend_connect_timeout=<timeout in seconds>
# backend_write_timeout=<timeout in seconds>
# backend_read_timeout=<timeout in seconds>
#
## MySQL monitor-specific options:
#
# Enable detection of replication slaves lag via replication_heartbeat
# table - optional.
#
# detect_replication_lag=[1|0] (default 0)
#
# Allow previous master to be available even in case of stopped or misconfigured
# replication - optional.
#
# detect_stale_master=[1|0] (default 0)
#
## Galera monitor-specific options:
#
# If disable_master_failback is not set, recovery of previously failed master
# causes mastership to be switched back to it. Enabling the option prevents it.
#
# disable_master_failback=[0|1] (default 0)
#
## Examples:
[MySQL Monitor]
type=monitor
module=mysqlmon
servers=server1,server2,server3
user=maxuser
passwd=maxpwd
#
# options for mysql_monitor only
#
# detect_replication_lag=<enable detection of replication slaves lag
# via replication_heartbeat table,
# default value is 0>
# detect_stale_master=<if the replication is stopped or misconfigured
# the previous detected master will be still available
# until monitor or MaxSclale restart,
# default value is 0>
user=myuser
passwd=mypwd
monitor_interval=10000
#backend_connect_timeout=
#backend_read_timeout=
#backend_write_timeout=
#detect_replication_lag=
#detect_stale_master=
# A series of service definition
[Galera Monitor]
type=monitor
module=galeramon
servers=server1,server2,server3
user=myuser
passwd=mypwd
monitor_interval=10000
#disable_master_failback=
## Filter definition
#
# Valid options are:
# Type specifies the section
#
# router=<name of router module>
# servers=<server name>,<server name>,...
# user=<User to fetch password inforamtion with>
# passwd=<Password of the user, plain text currently>
# enable_root_user=<0 or 1, default is 0>
# version_string=<specific string for server handshake,
# default is the MariaDB embedded library version>
#
# use_sql_variables_in=[master|all] (default all)
# router_options=<option[=value]>,<option[=value]>,...
# where value=[master|slave|synced]
# type=filter
#
# Module specifies which module implements the filter function
#
# Read/Write Split Router specific options are:
# module=[qlafilter|regexfilter|topfilter|teefilter]
#
# Options specify the log file for Query Log Filter
#
# options=<path to logfile>
#
# Match and replace are used in regexfilter
#
# match=fetch
# replace=select
#
# Count and filebase are used with topfilter to specify how many top queries are
# listed and where.
#
# count=<count>
# filebase=<path to output file>
#
# Match and service are used by tee filter to specify what queries should be
# duplicated and where the copy should be routed.
#
# match=insert.*HighScore.*values
# service=Cassandra
#
## Examples:
[qla]
type=filter
module=qlafilter
options=/tmp/QueryLog
[fetch]
type=filter
module=regexfilter
match=fetch
replace=select
## A series of service definition
#
# Name of router module, currently valid options are
#
# router=[readconnroute|readwritesplit|debugcli|CLI]
#
# List of server names for use of service - mandatory for readconnroute,
# readwritesplit, and debugcli
#
# servers=<server name 1>,<server name 2>,...,<server name N>
#
# Username to fetch password information with and password in plaintext
# format - for readconnroute and readwritesplit
#
# user=<username>
# passwd=<password in plain text format>
#
# flag for enabling the use of root user - for readconnroute and
# readwritesplite - optional.
#
# enable_root_user=[0|1] (default 0)
#
# Version string to be used in server handshake. Default value is that of
# MariaDB embedded library's - for readconnroute and readwritesplite - optional.
#
# version_string=<specific version string>
#
# Filters specify the filters through which the query is transferred and the
# order of their appearance on the list corresponds the order they are
# used. Values refer to names of filters configured in this file - for
# readconnroute and readwritesplit - optional.
#
# filters=<filter name1|filter name2|...|filter nameN>
#
## Read Connection Router specific router options.
#
# router_options specify the role in which the selected server must be.
#
# router_options=[master|slave|synced]
#
## Read/Write Split Router specific options.
#
# use_sql_variables_in specifies where sql variable modifications are
# routed - optional.
#
# use_sql_variables_in=[master|all] (default all)
#
# router_options=slave_selection_criteria specifies the selection criteria for
# slaves both in new session creation and when route target is selected - optional.
#
# router_options=
# slave_selection_criteria=[LEAST_CURRENT_OPERATIONS|LEAST_BEHIND_MASTER]
#
# max_slave_connections specifies how many slaves a router session can
# connect to - optional.
#
# max_slave_connections=<number, or percentage, of all slaves>
#
# max_slave_replication_lag specifies how much a slave is allowed to be behind
# the master and still become chosen routing target - optional, requires that
# monitor has detect_replication_lag=1 .
#
# max_slave_connections=<exact number or percentage of all slaves>
# max_slave_replication_lag=<allowed lag in seconds for a slave>
# router_options=slave_selection_criteria=[LEAST_CURRENT_OPERATIONS|LEAST_BEHIND_MASTER]
#
#
# Valid router modules currently are:
# readwritesplit, readconnroute and debugcli
# readwritesplit, readconnroute, debugcli and CLI
#
## Examples:
[Read Connection Router]
type=service
router=readconnroute
servers=server1,server2,server3
user=myuser
passwd=mypwd
router_options=slave
[RW Split Router]
type=service
router=readwritesplit
servers=server1,server2,server3
user=maxuser
passwd=maxpwd
use_sql_variables_in=all
max_slave_connections=50%
max_slave_replication_lag=30
router_options=slave_selection_criteria=LEAST_BEHIND_MASTER
[Read Connection Router]
type=service
router=readconnroute
router_options=slave
servers=server1,server2,server3
user=maxuser
passwd=maxpwd
user=myuser
passwd=mypwd
#use_sql_variables_in=
#max_slave_connections=100%
#max_slave_replication_lag=21
#router_options=slave_selection_criteria=
#filters=fetch|qla
[HTTPD Router]
type=service
@ -94,15 +217,44 @@ servers=server1,server2,server3
type=service
router=debugcli
# Listener definitions for the services
[CLI]
type=service
router=CLI
## Listener definitions for the services
#
# Valid options are:
# Type specifies section as listener one
#
# type=listener
#
# Service links the section to one of the service names used in this configuration
#
# service=<name of service section>
#
# Protocol is client protocol library name.
#
# protocol=[MySQLClient|telnetd|HTTPD|maxscaled]
#
# Port and address specify which port the service listens and the address limits
# listening to a specific network interface only. Address is optional.
#
# service=<name of service defined elsewhere>
# protocol=<name of protocol module with which to listen>
# port=<Listening port>
# address=<Address to bind to>
#
# Socket is alternative for address. The specified socket path must be writable
# by the Unix user MaxScale runs as.
#
# socket=<Listening socket>
#
## Examples:
[Read Connection Listener]
type=listener
service=Read Connection Router
protocol=MySQLClient
address=192.168.100.102
port=4008
#socket=/tmp/readconn.sock
[RW Split Listener]
type=listener
@ -111,19 +263,12 @@ protocol=MySQLClient
port=4006
#socket=/tmp/rwsplit.sock
[Read Connection Listener]
type=listener
service=Read Connection Router
protocol=MySQLClient
port=4008
#socket=/tmp/readconn.sock
[Debug Listener]
type=listener
service=Debug Interface
protocol=telnetd
port=4442
#address=127.0.0.1
port=4442
[HTTPD Listener]
type=listener
@ -131,40 +276,51 @@ service=HTTPD Router
protocol=HTTPD
port=6444
# Enable the maxadmin interface to MaxScale
#
# Listen on the default port of 6603 and restrict
# to connections from localhost only.
# Remove the address=localhost entry to enable
# maxadmin connections from any host
[CLI]
type=service
router=cli
[CLI Listener]
type=listener
service=CLI
protocol=maxscaled
address=localhost
#address=localhost
port=6603
# Definition of the servers
## Definition of the servers
#
# Type specifies the section as server one
#
# type=server
#
# The IP address or hostname of the machine running the database server that is
# being defined. MaxScale will use this address to connect to the backend
# database server.
#
# address=<IP|hostname>
#
# The port on which the database listens for incoming connections. MaxScale
# will use this port to connect to the database server.
#
# port=<port>
#
# The name for the protocol module to use to connect MaxScale to the database.
# Currently the only backend protocol supported is the MySQLBackend module.
#
# protocol=MySQLBackend
#
## Examples:
[server1]
type=server
address=127.0.0.1
address=192.168.100.101
port=3000
protocol=MySQLBackend
[server2]
type=server
address=127.0.0.1
port=3001
address=192.168.100.102
port=3000
protocol=MySQLBackend
[server3]
type=server
address=127.0.0.1
port=3002
address=192.168.100.103
port=3000
protocol=MySQLBackend

42
server/core/config.c
View File

@ -590,11 +590,12 @@ int error_count = 0;
}
if (obj->element && options)
{
char *s = strtok(options, ",");
char *lasts;
char *s = strtok_r(options, ",", &lasts);
while (s)
{
filterAddOption(obj->element, s);
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
if (obj->element)
@ -640,7 +641,8 @@ int error_count = 0;
router = config_get_value(obj->parameters, "router");
if (servers && obj->element)
{
char *s = strtok(servers, ",");
char *lasts;
char *s = strtok_r(servers, ",", &lasts);
while (s)
{
CONFIG_CONTEXT *obj1 = context;
@ -667,7 +669,7 @@ int error_count = 0;
"service '%s'.",
s, obj->object)));
}
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
else if (servers == NULL && internalService(router) == 0)
@ -681,11 +683,12 @@ int error_count = 0;
}
if (roptions && obj->element)
{
char *s = strtok(roptions, ",");
char *lasts;
char *s = strtok_r(roptions, ",", &lasts);
while (s)
{
serviceAddRouterOption(obj->element, s);
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
if (filters && obj->element)
@ -818,7 +821,7 @@ int error_count = 0;
obj->element = monitor_alloc(obj->object, module);
if (servers && obj->element)
{
char *s;
char *s, *lasts;
/* if id is not set, compute it now with pid only */
if (gateway.id == 0) {
@ -853,7 +856,7 @@ int error_count = 0;
monitorSetNetworkTimeout(obj->element, MONITOR_WRITE_TIMEOUT, write_timeout);
/* get the servers to monitor */
s = strtok(servers, ",");
s = strtok_r(servers, ",", &lasts);
while (s)
{
CONFIG_CONTEXT *obj1 = context;
@ -880,7 +883,7 @@ int error_count = 0;
"monitor '%s'.",
s, obj->object)));
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
if (obj->element && user && passwd)
@ -1529,7 +1532,8 @@ SERVER *server;
filters = config_get_value(obj->parameters, "filters");
if (servers && obj->element)
{
char *s = strtok(servers, ",");
char *lasts;
char *s = strtok_r(servers, ",", &lasts);
while (s)
{
CONFIG_CONTEXT *obj1 = context;
@ -1559,17 +1563,18 @@ SERVER *server;
"service '%s'.",
s, obj->object)));
}
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
if (roptions && obj->element)
{
char *s = strtok(roptions, ",");
char *lasts;
char *s = strtok_r(roptions, ",", &lasts);
serviceClearRouterOptions(obj->element);
while (s)
{
serviceAddRouterOption(obj->element, s);
s = strtok(NULL, ",");
s = strtok_r(NULL, ",", &lasts);
}
}
if (filters && obj->element)
@ -1667,17 +1672,6 @@ static char *service_params[] =
NULL
};
static char *server_params[] =
{
"type",
"address",
"port",
"protocol",
"monitorpw",
"monitoruser",
NULL
};
static char *listener_params[] =
{
"type",

179
server/core/dbusers.c
View File

@ -32,11 +32,13 @@
* x.y.z.%, x.y.%.%, x.%.%.%
* 03/10/14 Massimiliano Pinto Added netmask to user@host authentication for wildcard in IPv4 hosts
* 13/10/14 Massimiliano Pinto Added (user@host)@db authentication
* 04/12/14 Massimiliano Pinto Added support for IPv$ wildcard hosts: a.%, a.%.% and a.b.%
*
* @endverbatim
*/
#include <stdio.h>
#include <ctype.h>
#include <mysql.h>
#include <dcb.h>
@ -82,6 +84,7 @@ void resource_free(HASHTABLE *resource);
void *resource_fetch(HASHTABLE *, char *);
int resource_add(HASHTABLE *, char *, char *);
int resource_hash(char *);
static int normalize_hostname(char *input_host, char *output_host);
/**
* Load the user/passwd form mysql.user table into the service users' hashtable
@ -217,8 +220,6 @@ int add_mysql_users_with_host_ipv4(USERS *users, char *user, char *host, char *p
struct sockaddr_in serv_addr;
MYSQL_USER_HOST key;
char ret_ip[INET_ADDRSTRLEN + 1]="";
int found_range=0;
int found_any=0;
int ret = 0;
if (users == NULL || user == NULL || host == NULL) {
@ -255,42 +256,30 @@ int add_mysql_users_with_host_ipv4(USERS *users, char *user, char *host, char *p
/* ANY */
if (strcmp(host, "%") == 0) {
strcpy(ret_ip, "0.0.0.0");
found_any = 1;
key.netmask = 0;
} else {
char *tmp;
strncpy(ret_ip, host, INET_ADDRSTRLEN);
tmp = ret_ip+strlen(ret_ip)-1;
/* hostname without % wildcards has netmask = 32 */
key.netmask = normalize_hostname(host, ret_ip);
/* start from Class C */
while(tmp > ret_ip) {
if (*tmp == '%') {
/* set only the last IPv4 byte to 1
* avoiding setipadress() failure
* for Class C address
*/
found_range++;
if (found_range == 1)
*tmp = '1';
else
*tmp = '0';
}
tmp--;
if (key.netmask == -1) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : strdup() failed in normalize_hostname for %s@%s",
user,
host)));
}
}
/* fill IPv4 data struct */
if (setipaddress(&serv_addr.sin_addr, ret_ip)) {
if (setipaddress(&serv_addr.sin_addr, ret_ip) && strlen(ret_ip)) {
/* copy IPv4 data into key.ipv4 */
memcpy(&key.ipv4, &serv_addr, sizeof(serv_addr));
if (found_range) {
/* let's zero the last IP byte: a.b.c.0 we set above to 1*/
/* if netmask < 32 there are % wildcards */
if (key.netmask < 32) {
/* let's zero the last IP byte: a.b.c.0 we may have set above to 1*/
key.ipv4.sin_addr.s_addr &= 0x00FFFFFF;
key.netmask = 32 - (found_range * 8);
} else {
key.netmask = 32 - (found_any * 32);
}
/* add user@host as key and passwd as value in the MySQL users hash table */
@ -362,10 +351,11 @@ getDatabases(SERVICE *service, MYSQL *con)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Warning: Loading DB names for service [%s] returned 0 rows."
" SHOW DATABASES grant to user [%s] is required for MaxScale DB Name Authentication",
service->name,
service_user)));
"%s: Unable to load database grant information, MaxScale "
"authentication will proceed without including database "
"permissions. To correct this GRANT select permission "
"on msql.db to the user %s.",
service->name, service_user)));
}
/* free resut set */
@ -622,7 +612,11 @@ getUsers(SERVICE *service, USERS *users)
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Loading DB grants failed: GRANT is required on [mysql.db] to user [%s]. Try loading DB users for service [%s] without DB name MaxScale Authentication", service_user, service->name)));
"%s: Unable to load database grant information, MaxScale "
"authentication will proceed without including database "
"permissions. To correct this GRANT select permission "
"on msql.db to the user %s.",
service->name, service_user)));
/* check for root user select */
if(service->enable_root) {
@ -649,8 +643,9 @@ getUsers(SERVICE *service, USERS *users)
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"Loading users from [mysql.user] without DB grants from [mysql.db] for service [%s]."
" MaxScale Authentication with DBname on connect will not work",
"Loading users from [mysql.user] without access to [mysql.db] for "
"service [%s]. MaxScale Authentication with DBname on connect "
"will not consider database grants.",
service->name)));
}
} else {
@ -715,7 +710,23 @@ getUsers(SERVICE *service, USERS *users)
int rc = 0;
char *password = NULL;
if (row[2] != NULL) {
/* detect mysql_old_password (pre 4.1 protocol) */
if (strlen(row[2]) == 16) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%s: The user %s@%s has on old password in the "
"backend database. MaxScale does not support these "
"old passwords. This user will not be able to connect "
"via MaxScale. Update the users password to correct "
"this.",
service->name,
row[0],
row[1])));
continue;
}
if (strlen(row[2]) > 1)
password = row[2] +1;
else
@ -752,19 +763,20 @@ getUsers(SERVICE *service, USERS *users)
/* Log the user being added with its db grants */
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"Added user %s@%s with DB grants on [%s], for service [%s]",
"%s: User %s@%s for database %s added to "
"service user table.",
service->name,
row[0],
row[1],
dbgrant,
service->name)));
dbgrant)));
} else {
/* Log the user being added (without db grants) */
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"Added user %s@%s for service [%s]",
"%s: User %s@%s added to service user table.",
service->name,
row[0],
row[1],
service->name)));
row[1])));
}
/* Append data in the memory area for SHA1 digest */
@ -774,7 +786,8 @@ getUsers(SERVICE *service, USERS *users)
} else {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Warning: Failed adding user %s@%s for service [%s]",
"Warning: Failed to add user %s@%s for service [%s]. "
"This user will be unavailable via MaxScale.",
row[0],
row[1],
service->name)));
@ -1096,3 +1109,87 @@ resource_fetch(HASHTABLE *resources, char *key)
return hashtable_fetch(resources, key);
}
/**
* Normalize hostname with % wildcards to a valid IP string.
*
* Valid input values:
* a.b.c.d, a.b.c.%, a.b.%.%, a.%.%.%
* Short formats a.% and a.%.% are both converted to a.%.%.%
* Short format a.b.% is converted to a.b.%.%
*
* Last host byte is set to 1, avoiding setipadress() failure
*
* @param input_host The hostname with possible % wildcards
* @param output_host The normalized hostname (buffer must be preallocated)
* @return The calculated netmask or -1 on failure
*/
static int normalize_hostname(char *input_host, char *output_host)
{
int netmask, bytes, bits = 0, found_wildcard = 0;
char *p, *lasts, *tmp;
int useorig = 0;
output_host[0] = 0;
bytes = 0;
tmp = strdup(input_host);
if (tmp == NULL) {
return -1;
}
p = strtok_r(tmp, ".", &lasts);
while (p != NULL)
{
if (strcmp(p, "%"))
{
if (! isdigit(*p))
useorig = 1;
strcat(output_host, p);
bits += 8;
}
else if (bytes == 3)
{
found_wildcard = 1;
strcat(output_host, "1");
}
else
{
found_wildcard = 1;
strcat(output_host, "0");
}
bytes++;
p = strtok_r(NULL, ".", &lasts);
if (p)
strcat(output_host, ".");
}
if (found_wildcard)
{
netmask = bits;
while (bytes++ < 4)
{
if (bytes == 4)
{
strcat(output_host, ".1");
}
else
{
strcat(output_host, ".0");
}
}
}
else
netmask = 32;
if (useorig == 1)
{
netmask = 32;
strcpy(output_host, input_host);
}
free(tmp);
return netmask;
}

12
server/core/filter.c
View File

@ -332,8 +332,7 @@ DOWNSTREAM *me;
if ((filter->obj = load_module(filter->module,
MODULE_FILTER)) == NULL)
{
me = NULL;
goto retblock;
return NULL;
}
}
@ -342,8 +341,7 @@ DOWNSTREAM *me;
if ((filter->filter = (filter->obj->createInstance)(filter->options,
filter->parameters)) == NULL)
{
me = NULL;
goto retblock;
return NULL;
}
}
if ((me = (DOWNSTREAM *)calloc(1, sizeof(DOWNSTREAM))) == NULL)
@ -355,7 +353,7 @@ DOWNSTREAM *me;
errno,
strerror(errno))));
goto retblock;
return NULL;
}
me->instance = filter->filter;
me->routeQuery = (void *)(filter->obj->routeQuery);
@ -363,12 +361,10 @@ DOWNSTREAM *me;
if ((me->session=filter->obj->newSession(me->instance, session)) == NULL)
{
free(me);
me = NULL;
goto retblock;
return NULL;
}
filter->obj->setDownstream(me->instance, me->session, downstream);
retblock:
return me;
}

2
server/core/modutil.c
View File

@ -293,7 +293,7 @@ GWBUF *modutil_create_mysql_err_msg(
const char *msg)
{
uint8_t *outbuf = NULL;
uint8_t mysql_payload_size = 0;
uint32_t mysql_payload_size = 0;
uint8_t mysql_packet_header[4];
uint8_t *mysql_payload = NULL;
uint8_t field_count = 0;

28
server/core/secrets.c
View File

@ -227,8 +227,9 @@ static int reported = 0;
*/
int secrets_writeKeys(char *secret_file)
{
int fd;
MAXKEYS key;
int fd,randfd;
unsigned int randval;
MAXKEYS key;
/* Open for writing | Create | Truncate the file for writing */
if ((fd = open(secret_file, O_CREAT | O_WRONLY | O_TRUNC, S_IRUSR)) < 0)
@ -243,7 +244,28 @@ MAXKEYS key;
return 1;
}
srand(time(NULL));
/* Open for writing | Create | Truncate the file for writing */
if ((randfd = open("/dev/random", O_RDONLY)) < 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : failed opening /dev/random. Error %d, %s.",
errno,
strerror(errno))));
return 1;
}
if(read(randfd,(void*)&randval,sizeof(unsigned int)) < 1)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : failed to read /dev/random.")));
close(randfd);
return 1;
}
close(randfd);
srand(randval);
secrets_random_str(key.enckey, MAXSCALE_KEYLEN);
secrets_random_str(key.initvector, MAXSCALE_IV_LEN);

4
server/core/server.c
View File

@ -344,8 +344,10 @@ SERVER_PARAM *param;
}
}
if (server->node_ts > 0) {
struct tm result;
char buf[40];
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%s",
asctime(localtime(&server->node_ts)));
asctime_r(localtime_r((time_t *)(&server->node_ts), &result), buf));
}
if ((param = server->parameters) != NULL)
{

40
server/core/service.c
View File

@ -809,13 +809,16 @@ SERVICE *service;
void
printService(SERVICE *service)
{
SERVER *ptr = service->databases;
int i;
SERVER *ptr = service->databases;
struct tm result;
char time_buf[30];
int i;
printf("Service %p\n", service);
printf("\tService: %s\n", service->name);
printf("\tRouter: %s (%p)\n", service->routerModule, service->router);
printf("\tStarted: %s", asctime(localtime(&service->stats.started)));
printf("\tStarted: %s",
asctime_r(localtime_r(&service->stats.started, &result), time_buf));
printf("\tBackend databases\n");
while (ptr)
{
@ -887,8 +890,10 @@ SERVICE *ptr;
*/
void dprintService(DCB *dcb, SERVICE *service)
{
SERVER *server = service->databases;
int i;
SERVER *server = service->databases;
struct tm result;
char timebuf[30];
int i;
dcb_printf(dcb, "Service %p\n", service);
dcb_printf(dcb, "\tService: %s\n",
@ -898,7 +903,7 @@ int i;
if (service->router)
service->router->diagnostics(service->router_instance, dcb);
dcb_printf(dcb, "\tStarted: %s",
asctime(localtime(&service->stats.started)));
asctime_r(localtime_r(&service->stats.started, &result), timebuf));
dcb_printf(dcb, "\tRoot user access: %s\n",
service->enable_root ? "Enabled" : "Disabled");
if (service->n_filters)
@ -1232,19 +1237,18 @@ bool service_set_param_value (
/*
* Function to find a string in typelib_t
* (similar to find_type() of mysys/typelib.c)
*
* SYNOPSIS
* find_type()
* lib typelib_t
* find String to find
* length Length of string to find
* part_match Allow part matching of value
*
* RETURN
* 0 error
* > 0 position in TYPELIB->type_names +1
*
* SYNOPSIS
* find_type()
* lib typelib_t
* find String to find
* length Length of string to find
* part_match Allow part matching of value
*
* RETURN
* 0 error
* > 0 position in TYPELIB->type_names +1
*/
static int find_type(
typelib_t* tl,
const char* needle,

20
server/core/session.c
View File

@ -465,11 +465,15 @@ int rval = 0;
void
printSession(SESSION *session)
{
struct tm result;
char timebuf[40];
printf("Session %p\n", session);
printf("\tState: %s\n", session_state(session->state));
printf("\tService: %s (%p)\n", session->service->name, session->service);
printf("\tClient DCB: %p\n", session->client);
printf("\tConnected: %s", asctime(localtime(&session->stats.connect)));
printf("\tConnected: %s",
asctime_r(localtime_r(&session->stats.connect, &result), timebuf));
}
/**
@ -566,7 +570,9 @@ int norouter = 0;
void
dprintAllSessions(DCB *dcb)
{
SESSION *ptr;
struct tm result;
char timebuf[40];
SESSION *ptr;
spinlock_acquire(&session_spin);
ptr = allSessions;
@ -578,7 +584,8 @@ SESSION *ptr;
dcb_printf(dcb, "\tClient DCB: %p\n", ptr->client);
if (ptr->client && ptr->client->remote)
dcb_printf(dcb, "\tClient Address: %s\n", ptr->client->remote);
dcb_printf(dcb, "\tConnected: %s", asctime(localtime(&ptr->stats.connect)));
dcb_printf(dcb, "\tConnected: %s",
asctime_r(localtime_r(&ptr->stats.connect, &result), timebuf));
ptr = ptr->next;
}
spinlock_release(&session_spin);
@ -596,7 +603,9 @@ SESSION *ptr;
void
dprintSession(DCB *dcb, SESSION *ptr)
{
int i;
struct tm result;
char buf[30];
int i;
dcb_printf(dcb, "Session %d (%p)\n",ptr->ses_id, ptr);
dcb_printf(dcb, "\tState: %s\n", session_state(ptr->state));
@ -604,7 +613,8 @@ int i;
dcb_printf(dcb, "\tClient DCB: %p\n", ptr->client);
if (ptr->client && ptr->client->remote)
dcb_printf(dcb, "\tClient Address: %s\n", ptr->client->remote);
dcb_printf(dcb, "\tConnected: %s", asctime(localtime(&ptr->stats.connect)));
dcb_printf(dcb, "\tConnected: %s",
asctime_r(localtime_r(&ptr->stats.connect, &result), buf));
if (ptr->n_filters)
{
for (i = 0; i < ptr->n_filters; i++)

20
server/core/test/test_mysql_users.c
View File

@ -231,9 +231,9 @@ int set_and_get_mysql_users_wildcards(char *username, char *hostname, char *pass
service->users = mysql_users;
if (db_from != NULL)
strcpy(data->db, db_from);
strncpy(data->db, db_from,MYSQL_DATABASE_MAXLEN);
else
strcpy(data->db, "");
strncpy(data->db, "",MYSQL_DATABASE_MAXLEN);
/* freed by dcb_free(dcb) */
dcb->data = data;
@ -392,6 +392,22 @@ int main() {
if (!ret) fprintf(stderr, "\t-- Expecting ok\n");
assert(ret == 0);
ret = set_and_get_mysql_users_wildcards("pippo", "192.%", "foo", "192.254.254.242", NULL, NULL, NULL);
if (!ret) fprintf(stderr, "\t-- Expecting ok\n");
assert(ret == 0);
ret = set_and_get_mysql_users_wildcards("pippo", "192.%.%", "foo", "192.254.254.242", NULL, NULL, NULL);
if (!ret) fprintf(stderr, "\t-- Expecting ok\n");
assert(ret == 0);
ret = set_and_get_mysql_users_wildcards("pippo", "192.254.%", "foo", "192.254.254.242", NULL, NULL, NULL);
if (!ret) fprintf(stderr, "\t-- Expecting ok\n");
assert(ret == 0);
ret = set_and_get_mysql_users_wildcards("pippo", "192.254.%", "foo", "192.254.0.242", NULL, NULL, NULL);
if (!ret) fprintf(stderr, "\t-- Expecting ok\n");
assert(ret == 0);
ret = set_and_get_mysql_users_wildcards("riccio", "192.0.0.%", "foo", "192.134.0.2", NULL, NULL, NULL);
if (ret) fprintf(stderr, "\t-- Expecting no match\n");
assert(ret == 1);

16
server/modules/filter/mqfilter.c
View File

@ -423,8 +423,8 @@ init_conn(MQ_INSTANCE *my_instance)
*/
char** parse_optstr(char* str, char* tok, int* szstore)
{
char* tk = str;
char** arr;
char *lasts, *tk = str;
char **arr;
int i = 0, size = 1;
while((tk = strpbrk(tk + 1,tok))){
size++;
@ -440,10 +440,10 @@ char** parse_optstr(char* str, char* tok, int* szstore)
}
*szstore = size;
tk = strtok(str,tok);
tk = strtok_r(str,tok, &lasts);
while(tk && i < size){
arr[i++] = strdup(tk);
tk = strtok(NULL,tok);
tk = strtok_r(NULL,tok,&lasts);
}
return arr;
}
@ -1052,7 +1052,8 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
for(z = 0;z<tbsz;z++){
if((tmp = strchr(tblnames[z],'.')) != NULL){
tmp = strtok(tblnames[z],".");
char *lasts;
tmp = strtok_r(tblnames[z],".",&lasts);
for(i = 0; i<my_instance->shm_trg->size; i++){
if(strcmp(tmp,my_instance->shm_trg->objects[i]) == 0){
@ -1103,8 +1104,9 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
char* tbnm = NULL;
if((strchr(sesstbls[j],'.')) != NULL){
tbnm = strtok(sesstbls[j],".");
tbnm = strtok(NULL,".");
char *lasts;
tbnm = strtok_r(sesstbls[j],".",&lasts);
tbnm = strtok_r(NULL,".",&lasts);
}else{
tbnm = sesstbls[j];
}

2
server/modules/include/blr.h
View File

@ -193,6 +193,7 @@ typedef struct {
int n_slaves; /*< Number slave sessions created */
int n_reads; /*< Number of record reads */
uint64_t n_binlogs; /*< Number of binlog records from master */
uint64_t n_binlogs_ses; /*< Number of binlog records from master */
uint64_t n_binlog_errors;/*< Number of binlog records from master */
uint64_t n_rotates; /*< Number of binlog rotate events */
uint64_t n_cachehits; /*< Number of hits on the binlog cache */
@ -270,6 +271,7 @@ typedef struct router_instance {
unsigned int short_burst; /*< Short burst for slave catchup */
unsigned int long_burst; /*< Long burst for slave catchup */
unsigned long burst_size; /*< Maximum size of burst to send */
unsigned long heartbeat; /*< Configured heartbeat value */
ROUTER_STATS stats; /*< Statistics for this router */
int active_logs;
int reconnect_pending;

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

@ -95,6 +95,7 @@ typedef enum {
MYSQL_AUTH_SENT,
MYSQL_AUTH_RECV,
MYSQL_AUTH_FAILED,
MYSQL_HANDSHAKE_FAILED,
MYSQL_IDLE
} mysql_auth_state_t;
@ -251,7 +252,7 @@ typedef enum mysql_server_cmd {
typedef struct server_command_st {
mysql_server_cmd_t scom_cmd;
int scom_nresponse_packets; /*< packets in response */
size_t scom_nbytes_to_read; /*< bytes left to read in current packet */
ssize_t scom_nbytes_to_read; /*< bytes left to read in current packet */
struct server_command_st* scom_next;
} server_command_t;
@ -387,8 +388,8 @@ void protocol_remove_srv_command(MySQLProtocol* p);
bool protocol_waits_response(MySQLProtocol* p);
mysql_server_cmd_t protocol_get_srv_command(MySQLProtocol* p,bool removep);
int get_stmt_nresponse_packets(GWBUF* buf, mysql_server_cmd_t cmd);
bool protocol_get_response_status (MySQLProtocol* p, int* npackets, size_t* nbytes);
void protocol_set_response_status (MySQLProtocol* p, int npackets, size_t nbytes);
bool protocol_get_response_status (MySQLProtocol* p, int* npackets, ssize_t* nbytes);
void protocol_set_response_status (MySQLProtocol* p, int npackets, ssize_t nbytes);
void protocol_archive_srv_command(MySQLProtocol* p);
@ -396,6 +397,6 @@ void init_response_status (
GWBUF* buf,
mysql_server_cmd_t cmd,
int* npackets,
size_t* nbytes);
ssize_t* nbytes);

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

@ -676,12 +676,21 @@ int log_no_master = 1;
if (mon_status_changed(ptr))
{
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
#if defined(SS_DEBUG)
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"Backend server %s:%d state : %s",
ptr->server->name,
ptr->server->port,
STRSRVSTATUS(ptr->server))));
#else
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"Backend server %s:%d state : %s",
ptr->server->name,
ptr->server->port,
STRSRVSTATUS(ptr->server))));
#endif
}
if (SERVER_IS_DOWN(ptr->server))
@ -752,7 +761,7 @@ int log_no_master = 1;
/* log master detection failure od first master becomes available after failure */
if (root_master && mon_status_changed(root_master) && !(root_master->server->status & SERVER_STALE_STATUS)) {
if (root_master->pending_status & (SERVER_MASTER)) {
if (!(root_master->mon_prev_status & SERVER_STALE_STATUS)) {
if (!(root_master->mon_prev_status & SERVER_STALE_STATUS) && !(root_master->server->status & SERVER_MAINT)) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Info: A Master Server is now available: %s:%i",

2
server/modules/protocol/httpd.c
View File

@ -177,7 +177,7 @@ HTTPD_session *client_data = NULL;
j++;
}
while (!ISspace(buf[j]) && (i < sizeof(url) - 1) && (j < sizeof(buf) - 1)) {
while ((j < sizeof(buf) - 1) && !ISspace(buf[j]) && (i < sizeof(url) - 1)) {
url[i] = buf[j];
i++; j++;
}

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

@ -374,8 +374,8 @@ int rc;
rc = listen(listener->fd, SOMAXCONN);
if (rc == 0) {
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Listening maxscale connections at %s\n",
config)));
} else {

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

@ -211,12 +211,13 @@ static int gw_read_backend_event(DCB *dcb) {
/** Read cached backend handshake */
if (gw_read_backend_handshake(backend_protocol) != 0)
{
backend_protocol->protocol_auth_state = MYSQL_AUTH_FAILED;
backend_protocol->protocol_auth_state = MYSQL_HANDSHAKE_FAILED;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_event] after "
"gw_read_backend_handshake, fd %d, "
"state = MYSQL_AUTH_FAILED.",
"state = MYSQL_HANDSHAKE_FAILED.",
pthread_self(),
backend_protocol->owner_dcb->fd)));
}
@ -256,6 +257,7 @@ static int gw_read_backend_event(DCB *dcb) {
* -- handle a previous handshake error
*/
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV ||
backend_protocol->protocol_auth_state == MYSQL_HANDSHAKE_FAILED ||
backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
{
spinlock_acquire(&dcb->authlock);
@ -264,6 +266,7 @@ static int gw_read_backend_event(DCB *dcb) {
CHK_PROTOCOL(backend_protocol);
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV ||
backend_protocol->protocol_auth_state == MYSQL_HANDSHAKE_FAILED ||
backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
{
ROUTER_OBJECT *router = NULL;
@ -286,7 +289,7 @@ static int gw_read_backend_event(DCB *dcb) {
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_RECV)
{
/**
* Read backed's reply to authentication message
* Read backend's reply to authentication message
*/
receive_rc =
gw_receive_backend_auth(backend_protocol);
@ -340,7 +343,8 @@ static int gw_read_backend_event(DCB *dcb) {
} /* switch */
}
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED)
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED ||
backend_protocol->protocol_auth_state == MYSQL_HANDSHAKE_FAILED)
{
/**
* protocol state won't change anymore,
@ -362,7 +366,9 @@ static int gw_read_backend_event(DCB *dcb) {
bool succp;
/* try reload users' table for next connection */
service_refresh_users(dcb->session->service);
if (backend_protocol->protocol_auth_state == MYSQL_AUTH_FAILED) {
service_refresh_users(dcb->session->service);
}
#if defined(SS_DEBUG)
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
@ -428,7 +434,7 @@ static int gw_read_backend_event(DCB *dcb) {
spinlock_release(&dcb->authlock);
} /* MYSQL_AUTH_RECV || MYSQL_AUTH_FAILED */
} /* MYSQL_AUTH_RECV || MYSQL_AUTH_FAILED || MYSQL_HANDSHAKE_FAILED */
/* reading MySQL command output from backend and writing to the client */
{
@ -552,7 +558,7 @@ static int gw_read_backend_event(DCB *dcb) {
{
client_protocol = SESSION_PROTOCOL(dcb->session,
MySQLProtocol);
if (client_protocol != NULL)
if (client_protocol != NULL)
{
CHK_PROTOCOL(client_protocol);
@ -693,6 +699,7 @@ gw_MySQLWrite_backend(DCB *dcb, GWBUF *queue)
* return 1.
*/
switch (backend_protocol->protocol_auth_state) {
case MYSQL_HANDSHAKE_FAILED:
case MYSQL_AUTH_FAILED:
{
size_t len;
@ -823,18 +830,22 @@ static int gw_error_backend_event(DCB *dcb)
*/
if (dcb->state != DCB_STATE_POLLING)
{
int error, len;
char buf[100];
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, &len) == 0)
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"DCB in state %s got error '%s'.",
gw_dcb_state2string(dcb->state),
buf)));
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"DCB in state %s got error '%s'.",
STRDCBSTATE(dcb->state),
buf)));
}
}
return 1;
}
@ -862,18 +873,21 @@ static int gw_error_backend_event(DCB *dcb)
if (ses_state != SESSION_STATE_ROUTER_READY)
{
int error, len;
char buf[100];
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, &len) == 0)
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error '%s' in session that is not ready for routing.",
buf)));
}
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error '%s' in session that is not ready for routing.",
buf)));
}
}
gwbuf_free(errbuf);
goto retblock;
}
@ -946,7 +960,8 @@ static int gw_create_backend_connection(
"protocol object for backend connection.")));
goto return_fd;
}
/** Copy client flags to backend protocol */
if (backend_dcb->session->client->protocol)
{
/** Copy client flags to backend protocol */
@ -1073,18 +1088,21 @@ gw_backend_hangup(DCB *dcb)
if (ses_state != SESSION_STATE_ROUTER_READY)
{
int error, len;
char buf[100];
int error, len;
char buf[100];
len = sizeof(error);
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, &len) == 0)
if (getsockopt(dcb->fd, SOL_SOCKET, SO_ERROR, &error, (socklen_t *)&len) == 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Hangup in session that is not ready for routing, "
"Error reported is '%s'.",
buf)));
if (error != 0)
{
strerror_r(error, buf, 100);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Hangup in session that is not ready for routing, "
"Error reported is '%s'.",
buf)));
}
}
gwbuf_free(errbuf);
goto retblock;
@ -1320,12 +1338,11 @@ static int gw_change_user(
/* now get the user, after 4 bytes header and 1 byte command */
client_auth_packet += 5;
strcpy(username, (char *)client_auth_packet);
strncpy(username, (char *)client_auth_packet,MYSQL_USER_MAXLEN);
client_auth_packet += strlen(username) + 1;
/* get the auth token len */
memcpy(&auth_token_len, client_auth_packet, 1);
ss_dassert(auth_token_len >= 0);
client_auth_packet++;
@ -1341,7 +1358,7 @@ static int gw_change_user(
}
/* get new database name */
strcpy(database, (char *)client_auth_packet);
strncpy(database, (char *)client_auth_packet,MYSQL_DATABASE_MAXLEN);
/* get character set */
if (strlen(database)) {
@ -1354,7 +1371,7 @@ static int gw_change_user(
memcpy(&backend_protocol->charset, client_auth_packet, sizeof(int));
/* save current_database name */
strcpy(current_database, current_session->db);
strncpy(current_database, current_session->db,MYSQL_DATABASE_MAXLEN);
/*
* Now clear database name in dcb as we don't do local authentication on db name for change user.
@ -1474,7 +1491,7 @@ static GWBUF* process_response_data (
int nbytes_to_process) /*< number of new bytes read */
{
int npackets_left = 0; /*< response's packet count */
size_t nbytes_left = 0; /*< nbytes to be read for the packet */
ssize_t nbytes_left = 0; /*< nbytes to be read for the packet */
MySQLProtocol* p;
GWBUF* outbuf = NULL;
@ -1548,11 +1565,13 @@ static GWBUF* process_response_data (
*/
else /*< nbytes_left < nbytes_to_process */
{
ss_dassert(nbytes_left >= 0);
nbytes_to_process -= nbytes_left;
/** Move the prefix of the buffer to outbuf from redbuf */
outbuf = gwbuf_append(outbuf, gwbuf_clone_portion(readbuf, 0, nbytes_left));
readbuf = gwbuf_consume(readbuf, nbytes_left);
outbuf = gwbuf_append(outbuf,
gwbuf_clone_portion(readbuf, 0, (size_t)nbytes_left));
readbuf = gwbuf_consume(readbuf, (size_t)nbytes_left);
ss_dassert(npackets_left > 0);
npackets_left -= 1;
nbytes_left = 0;
@ -1600,7 +1619,7 @@ static bool sescmd_response_complete(
DCB* dcb)
{
int npackets_left;
size_t nbytes_left;
ssize_t nbytes_left;
MySQLProtocol* p;
bool succp;

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

@ -214,12 +214,12 @@ int gw_read_backend_handshake(
if (h_len <= 4) {
/* log error this exit point */
conn->protocol_auth_state = MYSQL_AUTH_FAILED;
conn->protocol_auth_state = MYSQL_HANDSHAKE_FAILED;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after "
"dcb_read, fd %d, "
"state = MYSQL_AUTH_FAILED.",
"state = MYSQL_HANDSHAKE_FAILED.",
dcb->fd,
pthread_self())));
@ -232,6 +232,8 @@ int gw_read_backend_handshake(
uint16_t errcode = MYSQL_GET_ERRCODE(payload);
char* bufstr = strndup(&((char *)payload)[7], len-3);
conn->protocol_auth_state = MYSQL_HANDSHAKE_FAILED;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_receive_backend_auth] Invalid "
@ -261,12 +263,14 @@ int gw_read_backend_handshake(
* data in buffer less than expected in the
* packet. Log error this exit point
*/
conn->protocol_auth_state = MYSQL_AUTH_FAILED;
conn->protocol_auth_state = MYSQL_HANDSHAKE_FAILED;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after "
"gw_mysql_get_byte3, fd %d, "
"state = MYSQL_AUTH_FAILED.",
"state = MYSQL_HANDSHAKE_FAILED.",
pthread_self(),
dcb->fd,
pthread_self())));
@ -285,12 +289,13 @@ int gw_read_backend_handshake(
* we cannot continue
* log error this exit point
*/
conn->protocol_auth_state = MYSQL_AUTH_FAILED;
conn->protocol_auth_state = MYSQL_HANDSHAKE_FAILED;
LOGIF(LD, (skygw_log_write(
LOGFILE_DEBUG,
"%lu [gw_read_backend_handshake] after "
"gw_decode_mysql_server_handshake, fd %d, "
"state = MYSQL_AUTH_FAILED.",
"state = MYSQL_HANDSHAKE_FAILED.",
pthread_self(),
conn->owner_dcb->fd,
pthread_self())));
@ -577,8 +582,8 @@ int gw_send_authentication_to_backend(
dcb = conn->owner_dcb;
final_capabilities = gw_mysql_get_byte4((uint8_t *)&server_capabilities);
/** Copy client's flags to backend */
final_capabilities |= conn->client_capabilities;
/** Copy client's flags to backend but with the known capabilities mask */
final_capabilities |= (conn->client_capabilities & GW_MYSQL_CAPABILITIES_CLIENT);
/* get charset the client sent and use it for connection auth */
charset = conn->charset;
@ -964,7 +969,7 @@ GWBUF* mysql_create_custom_error(
const char* msg)
{
uint8_t* outbuf = NULL;
uint8_t mysql_payload_size = 0;
uint32_t mysql_payload_size = 0;
uint8_t mysql_packet_header[4];
uint8_t* mysql_payload = NULL;
uint8_t field_count = 0;
@ -1574,7 +1579,7 @@ mysql_send_auth_error (
const char *mysql_message)
{
uint8_t *outbuf = NULL;
uint8_t mysql_payload_size = 0;
uint32_t mysql_payload_size = 0;
uint8_t mysql_packet_header[4];
uint8_t *mysql_payload = NULL;
uint8_t field_count = 0;
@ -1960,7 +1965,7 @@ void init_response_status (
GWBUF* buf,
mysql_server_cmd_t cmd,
int* npackets,
size_t* nbytes_left)
ssize_t* nbytes_left)
{
uint8_t* packet;
int nparam;
@ -2022,7 +2027,7 @@ void init_response_status (
bool protocol_get_response_status (
MySQLProtocol* p,
int* npackets,
size_t* nbytes)
ssize_t* nbytes)
{
bool succp;
@ -2030,7 +2035,7 @@ bool protocol_get_response_status (
spinlock_acquire(&p->protocol_lock);
*npackets = p->protocol_command.scom_nresponse_packets;
*nbytes = p->protocol_command.scom_nbytes_to_read;
*nbytes = (ssize_t)p->protocol_command.scom_nbytes_to_read;
spinlock_release(&p->protocol_lock);
if (*npackets < 0 && *nbytes == 0)
@ -2048,7 +2053,7 @@ bool protocol_get_response_status (
void protocol_set_response_status (
MySQLProtocol* p,
int npackets_left,
size_t nbytes)
ssize_t nbytes)
{
CHK_PROTOCOL(p);

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

@ -19,4 +19,7 @@ target_link_libraries(cli log_manager utils)
install(TARGETS cli DESTINATION modules)
add_subdirectory(readwritesplit)
if(BUILD_BINLOG)
add_subdirectory(binlog)
endif()

66
server/modules/routing/binlog/blr.c
View File

@ -188,6 +188,8 @@ int i;
inst->long_burst = DEF_LONG_BURST;
inst->burst_size = DEF_BURST_SIZE;
inst->retry_backoff = 1;
inst->binlogdir = NULL;
inst->heartbeat = 300; // Default is every 5 minutes
/*
* We only support one server behind this router, since the server is
@ -308,6 +310,14 @@ int i;
inst->burst_size = size;
}
else if (strcmp(options[i], "heartbeat") == 0)
{
inst->heartbeat = atoi(value);
}
else if (strcmp(options[i], "binlogdir") == 0)
{
inst->binlogdir = strdup(value);
}
else
{
LOGIF(LE, (skygw_log_write(
@ -516,9 +526,9 @@ ROUTER_SLAVE *slave = (ROUTER_SLAVE *)router_session;
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"%s: Master %s disconnected after %ld seconds. "
"%d events read.",
"%d events read,",
router->service->name, router->master->remote,
time(0) - router->connect_time, router->stats.n_binlogs)));
time(0) - router->connect_time, router->stats.n_binlogs_ses)));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Binlog router close session with master server %s",
@ -538,9 +548,10 @@ ROUTER_SLAVE *slave = (ROUTER_SLAVE *)router_session;
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE,
"%s: Slave %s disconnected after %ld seconds. "
"%s: Slave %s, server id %d, disconnected after %ld seconds. "
"%d events sent, %lu bytes.",
router->service->name, slave->dcb->remote,
slave->serverid,
time(0) - slave->connect_time, slave->stats.n_events,
slave->stats.n_bytes)));
@ -656,25 +667,29 @@ struct tm tm;
min5 /= 5.0;
dcb_printf(dcb, "\tMaster connection DCB: %p\n",
dcb_printf(dcb, "\tMaster connection DCB: %p\n",
router_inst->master);
dcb_printf(dcb, "\tMaster connection state: %s\n",
dcb_printf(dcb, "\tMaster connection state: %s\n",
blrm_states[router_inst->master_state]);
localtime_r(&router_inst->stats.lastReply, &tm);
asctime_r(&tm, buf);
dcb_printf(dcb, "\tNumber of master connects: %d\n",
dcb_printf(dcb, "\tBinlog directory: %s\n",
router_inst->binlogdir);
dcb_printf(dcb, "\tNumber of master connects: %d\n",
router_inst->stats.n_masterstarts);
dcb_printf(dcb, "\tNumber of delayed reconnects: %d\n",
dcb_printf(dcb, "\tNumber of delayed reconnects: %d\n",
router_inst->stats.n_delayedreconnects);
dcb_printf(dcb, "\tCurrent binlog file: %s\n",
dcb_printf(dcb, "\tCurrent binlog file: %s\n",
router_inst->binlog_name);
dcb_printf(dcb, "\tCurrent binlog position: %u\n",
dcb_printf(dcb, "\tCurrent binlog position: %u\n",
router_inst->binlog_position);
dcb_printf(dcb, "\tNumber of slave servers: %u\n",
dcb_printf(dcb, "\tNumber of slave servers: %u\n",
router_inst->stats.n_slaves);
dcb_printf(dcb, "\tNumber of binlog events received: %u\n",
dcb_printf(dcb, "\tNo. of binlog events received this session: %u\n",
router_inst->stats.n_binlogs_ses);
dcb_printf(dcb, "\tTotal no. of binlog events received: %u\n",
router_inst->stats.n_binlogs);
minno = router_inst->stats.minno - 1;
if (minno == -1)
@ -683,28 +698,31 @@ struct tm tm;
dcb_printf(dcb, "\tCurrent 5 10 15 30 Min Avg\n");
dcb_printf(dcb, "\t %6d %8.1f %8.1f %8.1f %8.1f\n",
router_inst->stats.minavgs[minno], min5, min10, min15, min30);
dcb_printf(dcb, "\tNumber of fake binlog events: %u\n",
dcb_printf(dcb, "\tNumber of fake binlog events: %u\n",
router_inst->stats.n_fakeevents);
dcb_printf(dcb, "\tNumber of artificial binlog events: %u\n",
dcb_printf(dcb, "\tNumber of artificial binlog events: %u\n",
router_inst->stats.n_artificial);
dcb_printf(dcb, "\tNumber of binlog events in error: %u\n",
dcb_printf(dcb, "\tNumber of binlog events in error: %u\n",
router_inst->stats.n_binlog_errors);
dcb_printf(dcb, "\tNumber of binlog rotate events: %u\n",
dcb_printf(dcb, "\tNumber of binlog rotate events: %u\n",
router_inst->stats.n_rotates);
dcb_printf(dcb, "\tNumber of heartbeat events: %u\n",
dcb_printf(dcb, "\tNumber of heartbeat events: %u\n",
router_inst->stats.n_heartbeats);
dcb_printf(dcb, "\tNumber of packets received: %u\n",
dcb_printf(dcb, "\tNumber of packets received: %u\n",
router_inst->stats.n_reads);
dcb_printf(dcb, "\tNumber of residual data packets: %u\n",
dcb_printf(dcb, "\tNumber of residual data packets: %u\n",
router_inst->stats.n_residuals);
dcb_printf(dcb, "\tAverage events per packet %.1f\n",
dcb_printf(dcb, "\tAverage events per packet %.1f\n",
(double)router_inst->stats.n_binlogs / router_inst->stats.n_reads);
dcb_printf(dcb, "\tLast event from master at: %s",
dcb_printf(dcb, "\tLast event from master at: %s",
buf);
dcb_printf(dcb, "\t (%d seconds ago)\n",
time(0) - router_inst->stats.lastReply);
dcb_printf(dcb, "\tLast event from master: 0x%x\n",
router_inst->lastEventReceived);
dcb_printf(dcb, "\tLast event from master: 0x%x (%s)\n",
router_inst->lastEventReceived,
(router_inst->lastEventReceived >= 0 &&
router_inst->lastEventReceived < 0x24) ?
event_names[router_inst->lastEventReceived] : "unknown");
if (router_inst->active_logs)
dcb_printf(dcb, "\tRouter processing binlog records\n");
if (router_inst->reconnect_pending)
@ -712,7 +730,7 @@ struct tm tm;
dcb_printf(dcb, "\tEvents received:\n");
for (i = 0; i < 0x24; i++)
{
dcb_printf(dcb, "\t\t%-38s: %u\n", event_names[i], router_inst->stats.events[i]);
dcb_printf(dcb, "\t\t%-38s %u\n", event_names[i], router_inst->stats.events[i]);
}
#if SPINLOCK_PROFILE
@ -927,7 +945,7 @@ char msg[85], *errmsg;
"%s: Master %s disconnected after %ld seconds. "
"%d events read.",
router->service->name, router->master->remote,
time(0) - router->connect_time, router->stats.n_binlogs)));
time(0) - router->connect_time, router->stats.n_binlogs_ses)));
blr_master_reconnect(router);
}

29
server/modules/routing/binlog/blr_file.c
View File

@ -77,18 +77,27 @@ int root_len, i;
DIR *dirp;
struct dirent *dp;
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
if (router->binlogdir == NULL)
{
strcpy(path, ptr);
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
}
strcat(path, "/");
strcat(path, router->service->name);
if (access(path, R_OK) == -1)
mkdir(path, 0777);
router->binlogdir = strdup(path);
}
if (access(router->binlogdir, R_OK) == -1)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"%s: Unable to read the binlog directory %s.",
router->service->name, router->binlogdir)));
}
strcat(path, "/");
strcat(path, router->service->name);
if (access(path, R_OK) == -1)
mkdir(path, 0777);
router->binlogdir = strdup(path);
/* First try to find a binlog file number by reading the directory */
root_len = strlen(router->fileroot);

10
server/modules/routing/binlog/blr_master.c
View File

@ -93,6 +93,7 @@ blr_start_master(ROUTER_INSTANCE *router)
DCB *client;
GWBUF *buf;
router->stats.n_binlogs_ses = 0;
spinlock_acquire(&router->lock);
if (router->master_state != BLRM_UNCONNECTED)
{
@ -331,7 +332,11 @@ char query[128];
// Response to fetch of master's server-id
router->saved_master.server_id = buf;
// TODO: Extract the value of server-id and place in router->master_id
buf = blr_make_query("SET @master_heartbeat_period = 1799999979520");
{
char str[80];
sprintf(str, "SET @master_heartbeat_period = %lu000000000", router->heartbeat);
buf = blr_make_query(str);
}
router->master_state = BLRM_HBPERIOD;
router->master->func.write(router->master, buf);
break;
@ -697,6 +702,9 @@ static REP_HEADER phdr;
}
else
{
router->stats.n_binlogs++;
router->stats.n_binlogs_ses++;
router->lastEventReceived = hdr.event_type;
blr_extract_header(ptr, &hdr);

3
server/modules/routing/binlog/blr_slave.c
View File

@ -557,8 +557,9 @@ uint32_t chksum;
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"%s: New slave %s requested binlog file %s from position %lu",
"%s: New slave %s, server id %d, requested binlog file %s from position %lu",
router->service->name, slave->dcb->remote,
slave->serverid,
slave->binlogfile, slave->binlog_pos)));
if (slave->binlog_pos != router->binlog_position ||

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

@ -1212,24 +1212,27 @@ static bool get_dcb(
* backend and update assign it to new candidate if
* necessary.
*/
else if (max_rlag == MAX_RLAG_UNDEFINED ||
else if (SERVER_IS_SLAVE(b->backend_server))
{
if (max_rlag == MAX_RLAG_UNDEFINED ||
(b->backend_server->rlag != MAX_RLAG_NOT_AVAILABLE &&
b->backend_server->rlag <= max_rlag))
{
candidate_bref = check_candidate_bref(
candidate_bref,
&backend_ref[i],
rses->rses_config.rw_slave_select_criteria);
}
else
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Server %s:%d is too much behind the "
"master, %d s. and can't be chosen.",
b->backend_server->name,
b->backend_server->port,
b->backend_server->rlag)));
{
candidate_bref = check_candidate_bref(
candidate_bref,
&backend_ref[i],
rses->rses_config.rw_slave_select_criteria);
}
else
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Server %s:%d is too much behind the "
"master, %d s. and can't be chosen.",
b->backend_server->name,
b->backend_server->port,
b->backend_server->rlag)));
}
}
} /*< for */
/** Assign selected DCB's pointer value */
@ -1473,6 +1476,12 @@ static route_target_t get_route_target (
QUERY_IS_TYPE(qtype, QUERY_TYPE_UNKNOWN)));
target = TARGET_MASTER;
}
#if defined(SS_EXTRA_DEBUG)
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Selected target \"%s\"",
STRTARGET(target))));
#endif
return target;
}
@ -2106,7 +2115,7 @@ static int routeQuery(
rlag_max)));
}
}
}
}
else if (TARGET_IS_SLAVE(route_target))
{
btype = BE_SLAVE;
@ -2125,6 +2134,14 @@ static int routeQuery(
rlag_max);
if (succp)
{
#if defined(SS_EXTRA_DEBUG)
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
"Found DCB for slave.")));
#endif
ss_dassert(get_bref_from_dcb(router_cli_ses, target_dcb) !=
router_cli_ses->rses_master_ref);
ss_dassert(get_root_master_bref(router_cli_ses) ==
router_cli_ses->rses_master_ref);
atomic_add(&inst->stats.n_slave, 1);
}
else
@ -2189,7 +2206,7 @@ static int routeQuery(
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Route query to %s\t%s:%d <",
"Route query to %s \t%s:%d <",
(SERVER_IS_MASTER(bref->bref_backend->backend_server) ?
"master" : "slave"),
bref->bref_backend->backend_server->name,
@ -3933,12 +3950,12 @@ static bool route_session_write(
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Route query to %s\t%s:%d%s",
"Route query to %s \t%s:%d%s",
(SERVER_IS_MASTER(backend_ref[i].bref_backend->backend_server) ?
"master" : "slave"),
backend_ref[i].bref_backend->backend_server->name,
backend_ref[i].bref_backend->backend_server->port,
(i+1==router_cli_ses->rses_nbackends ? " <" : ""))));
(i+1==router_cli_ses->rses_nbackends ? " <" : " "))));
}
if (BREF_IS_IN_USE((&backend_ref[i])))
@ -3988,12 +4005,12 @@ static bool route_session_write(
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Route query to %s\t%s:%d%s",
"Route query to %s \t%s:%d%s",
(SERVER_IS_MASTER(backend_ref[i].bref_backend->backend_server) ?
"master" : "slave"),
backend_ref[i].bref_backend->backend_server->name,
backend_ref[i].bref_backend->backend_server->port,
(i+1==router_cli_ses->rses_nbackends ? " <" : ""))));
(i+1==router_cli_ses->rses_nbackends ? " <" : " "))));
}
scur = backend_ref_get_sescmd_cursor(&backend_ref[i]);

2
server/modules/routing/readwritesplit/test/rwsplit.sh
View File

@ -1,4 +1,4 @@
#!/bin/sh
#!/bin/bash
NARGS=7
TLOG=$1
THOST=$2

6
utils/skygw_debug.h
View File

@ -262,6 +262,12 @@ typedef enum skygw_chk_t {
(SERVER_IS_RELAY_SERVER(s) ? "RUNNING RELAY" : \
(SERVER_IS_RUNNING(s) ? "RUNNING (only)" : "NO STATUS")))))))
#define STRTARGET(t) (t == TARGET_ALL ? "TARGET_ALL" : \
(t == TARGET_MASTER ? "TARGET_MASTER" : \
(t == TARGET_SLAVE ? "TARGET_SLAVE" : \
(t == TARGET_UNDEFINED ? "TARGET_UNDEFINED" : \
"Unknown target value"))))
#define BREFSRV(b) (b->bref_backend->backend_server)

11
utils/skygw_utils.cc
View File

@ -1412,7 +1412,6 @@ int simple_mutex_unlock(
err,
strerror(errno));
perror("simple_mutex : ");
ss_dassert(sm->sm_mutex.__data.__nusers >= 0);
} else {
/**
* Note that these updates are not protected.
@ -1680,12 +1679,12 @@ static bool file_write_header(
if (wbytes1 != 1 || wbytes2 != 1 || wbytes3 != 1 || wbytes4 != 1) {
fprintf(stderr,
"* Writing header %s %s %s to %s failed.\n",
"\nError : Writing header %s %s %s %s failed.\n",
header_buf1,
header_buf2,
header_buf3,
header_buf4);
perror("Logfile header write.\n");
perror("Logfile header write");
goto return_succp;
}
#endif
@ -1757,11 +1756,11 @@ static bool file_write_footer(
if (wbytes1 != 1 || wbytes3 != 1 || wbytes4 != 1)
{
fprintf(stderr,
"* Writing header %s %s to %s failed.\n",
"\nError : Writing header %s %s to %s failed.\n",
header_buf1,
header_buf3,
header_buf4);
perror("Logfile header write.\n");
perror("Logfile header write");
goto return_succp;
}
#endif
@ -1875,7 +1874,7 @@ skygw_file_t* skygw_file_init(
int eno = errno;
errno = 0;
fprintf(stderr,
"* Writing header of log file %s failed due %d, %s.\n",
"\nError : Writing header of log file %s failed due %d, %s.\n",
file->sf_fname,
eno,
strerror(eno));