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

Merge branch 'develop' into MAX-167

Browse Source 
Conflicts:
	server/modules/monitor/mysql_mon.c
	server/modules/routing/readwritesplit/readwritesplit.c
This commit is contained in:
VilhoRaatikka 2014-06-30 14:17:00 +03:00
commit 0b89245a4b
36 changed files with 1601 additions and 340 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

Binary file not shown.

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

Binary file not shown.

63
client/depend.mk
View File

@ -0,0 +1,63 @@
maxadmin.o: maxadmin.c /usr/include/stdc-predef.h /usr/include/stdio.h \
/usr/include/features.h /usr/include/x86_64-linux-gnu/sys/cdefs.h \
/usr/include/x86_64-linux-gnu/bits/wordsize.h \
/usr/include/x86_64-linux-gnu/gnu/stubs.h \
/usr/include/x86_64-linux-gnu/gnu/stubs-64.h \
/usr/lib/gcc/x86_64-linux-gnu/4.8/include/stddef.h \
/usr/include/x86_64-linux-gnu/bits/types.h \
/usr/include/x86_64-linux-gnu/bits/typesizes.h /usr/include/libio.h \
/usr/include/_G_config.h /usr/include/wchar.h \
/usr/lib/gcc/x86_64-linux-gnu/4.8/include/stdarg.h \
/usr/include/x86_64-linux-gnu/bits/stdio_lim.h \
/usr/include/x86_64-linux-gnu/bits/sys_errlist.h /usr/include/string.h \
/usr/include/xlocale.h /usr/include/signal.h \
/usr/include/x86_64-linux-gnu/bits/sigset.h \
/usr/include/x86_64-linux-gnu/bits/signum.h /usr/include/time.h \
/usr/include/x86_64-linux-gnu/bits/siginfo.h \
/usr/include/x86_64-linux-gnu/bits/sigaction.h \
/usr/include/x86_64-linux-gnu/bits/sigcontext.h \
/usr/include/x86_64-linux-gnu/bits/sigstack.h \
/usr/include/x86_64-linux-gnu/sys/ucontext.h \
/usr/include/x86_64-linux-gnu/bits/pthreadtypes.h \
/usr/include/x86_64-linux-gnu/bits/sigthread.h \
/usr/include/x86_64-linux-gnu/sys/wait.h \
/usr/include/x86_64-linux-gnu/bits/waitflags.h \
/usr/include/x86_64-linux-gnu/bits/waitstatus.h /usr/include/endian.h \
/usr/include/x86_64-linux-gnu/bits/endian.h \
/usr/include/x86_64-linux-gnu/bits/byteswap.h \
/usr/include/x86_64-linux-gnu/bits/byteswap-16.h \
/usr/include/x86_64-linux-gnu/sys/types.h \
/usr/include/x86_64-linux-gnu/sys/select.h \
/usr/include/x86_64-linux-gnu/bits/select.h \
/usr/include/x86_64-linux-gnu/bits/time.h \
/usr/include/x86_64-linux-gnu/sys/sysmacros.h \
/usr/include/x86_64-linux-gnu/sys/socket.h \
/usr/include/x86_64-linux-gnu/sys/uio.h \
/usr/include/x86_64-linux-gnu/bits/uio.h \
/usr/include/x86_64-linux-gnu/bits/socket.h \
/usr/include/x86_64-linux-gnu/bits/socket_type.h \
/usr/include/x86_64-linux-gnu/bits/sockaddr.h \
/usr/include/x86_64-linux-gnu/asm/socket.h \
/usr/include/asm-generic/socket.h \
/usr/include/x86_64-linux-gnu/asm/sockios.h \
/usr/include/asm-generic/sockios.h /usr/include/netinet/in.h \
/usr/lib/gcc/x86_64-linux-gnu/4.8/include/stdint.h /usr/include/stdint.h \
/usr/include/x86_64-linux-gnu/bits/wchar.h \
/usr/include/x86_64-linux-gnu/bits/in.h /usr/include/arpa/inet.h \
/usr/include/netdb.h /usr/include/rpc/netdb.h \
/usr/include/x86_64-linux-gnu/bits/netdb.h /usr/include/ctype.h \
/usr/include/stdlib.h /usr/include/alloca.h \
/usr/include/x86_64-linux-gnu/bits/stdlib-float.h /usr/include/termios.h \
/usr/include/x86_64-linux-gnu/bits/termios.h \
/usr/include/x86_64-linux-gnu/sys/ttydefaults.h /usr/include/unistd.h \
/usr/include/x86_64-linux-gnu/bits/posix_opt.h \
/usr/include/x86_64-linux-gnu/bits/environments.h \
/usr/include/x86_64-linux-gnu/bits/confname.h /usr/include/getopt.h \
/usr/include/dirent.h /usr/include/x86_64-linux-gnu/bits/dirent.h \
/usr/include/x86_64-linux-gnu/bits/posix1_lim.h \
/usr/include/x86_64-linux-gnu/bits/local_lim.h \
/usr/include/linux/limits.h /usr/include/locale.h \
/usr/include/x86_64-linux-gnu/bits/locale.h /usr/include/errno.h \
/usr/include/x86_64-linux-gnu/bits/errno.h /usr/include/linux/errno.h \
/usr/include/x86_64-linux-gnu/asm/errno.h \
/usr/include/asm-generic/errno.h /usr/include/asm-generic/errno-base.h

35
client/maxadmin.c
View File

@ -25,6 +25,8 @@
* Date Who Description
* 13/06/14 Mark Riddoch Initial implementation
* 15/06/14 Mark Riddoch Addition of source command
* 26/06/14 Mark Riddoch Fix issue with final OK split across
* multiple reads
*
* @endverbatim
*/
@ -440,7 +442,9 @@ char buf[20];
/**
* Send a comamnd using the MaxScaled protocol, display the return data
* on standard output
* on standard output.
*
* Input terminates with a lien containing jsut the text OK
*
* @param so The socket connect to MaxScale
* @param cmd The command to send
@ -450,19 +454,38 @@ static int
sendCommand(int so, char *cmd)
{
char buf[80];
int i;
int i, j, newline = 0;
write(so, cmd, strlen(cmd));
while (1)
{
if ((i = read(so, buf, 80)) == -1)
return 0;
if (i > 1 && buf[i-1] == 'K' && buf[i-2] == 'O')
for (j = 0; j < i; j++)
{
write(1, buf, i - 2);
return 1;
if (newline == 1 && buf[j] == 'O')
newline = 2;
else if (newline == 2 && buf[j] == 'K' && j == i - 1)
{
return 1;
}
else if (newline == 2)
{
putchar('O');
putchar(buf[j]);
newline = 0;
}
else if (buf[j] == '\n' || buf[j] == '\r')
{
putchar(buf[j]);
newline = 1;
}
else
{
putchar(buf[j]);
newline = 0;
}
}
write(1, buf, i);
}
return 1;
}

5
debian/changelog vendored Normal file
View File

@ -0,0 +1,5 @@
maxscale (0.7-1) UNRELEASED; urgency=low
* Initial release. (Closes: #XXXXXX)
-- Timofey Turenko <timofey.turenko@skysql.com> Tue, 11 Mar 2014 22:59:35 +0200

1
debian/compat vendored Normal file
View File

@ -0,0 +1 @@
8

15
debian/control vendored Normal file
View File

@ -0,0 +1,15 @@
Source: maxscale
Maintainer: Timofey Turenko
Section: misc
Priority: optional
Standards-Version: 3.9.2
Build-Depends: debhelper (>= 8), gcc, g++, ncurses-dev, bison, build-essential, libssl-dev, libaio-dev, libmariadbclient-dev, libmariadbd-dev, mariadb-server, cmake, perl, make, libtool,
Package: maxscale
Architecture: any
Depends: ${shlibs:Depends}, ${misc:Depends}
Description: MaxScale
The SkySQL MaxScale is an intelligent proxy that allows forwarding of
database statements to one or more database servers using complex rules,
a semantic understanding of the database statements and the roles of
the various servers within the backend cluster of databases.

1
debian/files vendored Normal file
View File

@ -0,0 +1 @@
maxscale_0.7-1_amd64.deb misc optional

3
debian/install vendored Normal file
View File

@ -0,0 +1,3 @@
maxscale.conf etc/ld.so.conf.d/
etc/init.d/maxscale etc/init.d/
binaries/* usr/local/sbin/

4
debian/postinst vendored Normal file
View File

@ -0,0 +1,4 @@
#!/bin/bash
ln -s /lib64/libaio.so.1 /lib64/libaio.so
/sbin/ldconfig

11
debian/rules vendored Executable file
View File

@ -0,0 +1,11 @@
#!/usr/bin/make -f
%:
$(MAKE) ROOT_PATH=$(shell pwd) HOME="" clean
$(MAKE) ROOT_PATH=$(shell pwd) HOME="" depend
$(MAKE) ROOT_PATH=$(shell pwd) HOME=""
$(MAKE) DEST="$(shell pwd)/binaries" ROOT_PATH=$(shell pwd) HOME="" ERRMSG="/usr/share/mysql/english" EMBEDDED_LIB="/usr/lib/x86_64-linux-gnu/" install
dh $@
override_dh_usrlocal:
override_dh_auto_clean:
override_dh_auto_build:
override_dh_auto_install:

91
etc/init.d/maxscale
View File

@ -17,15 +17,36 @@
# database clusters offering different routing, filtering and protocol choices
### END INIT INFO
MAXSCALE_HOME=/usr/local/skysql/maxscale
#############################################
# MaxScale BASEDIR, BIN, HOME, PIDFILE, LIB
#############################################
export MAXSCALE_BASEDIR=/usr/local/skysql/maxscale
export MAXSCALE_BIN=$MAXSCALE_BASEDIR/bin
export MAXSCALE_HOME=$MAXSCALE_BASEDIR/MaxScale
export MAXSCALE_PIDFILE=$MAXSCALE_HOME/log/maxscale.pid
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$MAXSCALE_BASEDIR/lib
###############################
# LSB Exit codes (non-Status)
###############################
_RETVAL_GENERIC=1
_RETVAL_NOT_INSTALLED=5
_RETVAL_NOT_RUNNING=7
###############################
# LSB Status action Exit codes
###############################
_RETVAL_STATUS_OK=0
_RETVAL_STATUS_NOT_RUNNING=3
# Sanity checks.
[ -x $MAXSCALE_HOME/bin/maxscale ] || exit 0
[ -x $MAXSCALE_BIN/maxscale ] || exit $_RETVAL_NOT_INSTALLED
# Source function library.
. /etc/rc.d/init.d/functions
# so we can rearrange this easily
# we can rearrange this easily
processname=maxscale
servicename=maxscale
@ -33,31 +54,54 @@ RETVAL=0
start() {
echo -n $"Starting MaxScale: "
if [ -x $MAXSCALE_HOME/bin/maxscale ] ; then
$MAXSCALE_HOME/bin/maxscale
my_check=`status -p $MAXSCALE_PIDFILE $MAXSCALE_BIN/maxscale`
CHECK_RET=$?
[ $CHECK_RET -eq 0 ] && echo -n " found $my_check" && success && CHECK_RET=0
daemon --pidfile $MAXSCALE_PIDFILE $MAXSCALE_BIN/maxscale >& /dev/null
RETVAL=$?
[ $RETVAL -eq 0 ] && touch /var/lock/subsys/$servicename
if [ $CHECK_RET -ne 0 ]; then
sleep 2
my_check=`status -p $MAXSCALE_PIDFILE $MAXSCALE_BIN/maxscale`
CHECK_RET=$?
[ $CHECK_RET -eq 0 ] && echo -n $my_check && success
fi
daemon --check $servicename $processname --system
RETVAL=$?
echo
[ $RETVAL -eq 0 ] && touch /var/lock/subsys/$servicename
# Return rigth code
if [ $RETVAL -ne 0 ]; then
RETVAL=$_RETVAL_NOT_RUNNING
fi
return $RETVAL
}
stop() {
echo -n $"Stopping MaxScale: "
killproc -p $MAXSCALE_PIDFILE -TERM
killproc $servicename -TERM
RETVAL=$?
echo
if [ $RETVAL -eq 0 ]; then
rm -f /var/lock/subsys/$servicename
[ $RETVAL -eq 0 ] && rm -f /var/lock/subsys/$servicename
# Return rigth code
if [ $RETVAL -ne 0 ]; then
RETVAL=$_RETVAL_NOT_RUNNING
fi
return $RETVAL
}
reload() {
echo -n $"Reloading MaxScale: "
killproc $servicename -HUP
killproc -p $MAXSCALE_PIDFILE $MAXSCALE_BIN/maxscale -HUP
RETVAL=$?
echo
}
@ -65,14 +109,33 @@ reload() {
# See how we were called.
case "$1" in
start)
start
start
;;
stop)
stop
;;
status)
status $servicename
# return 0 on success
# return 3 on any error
echo -n $"Checking MaxScale status: "
status -p $MAXSCALE_PIDFILE 'MaxScale'
RETVAL=$?
if [ $RETVAL -ne 0 ]; then
echo -ne "\033[1A"
[ $RETVAL -eq 1 ] && warning || failure
echo -ne "\033[1B"
RETVAL=$_RETVAL_STATUS_NOT_RUNNING
else
echo -ne "\033[1A"
success
echo -ne "\033[1B"
RETVAL=$_RETVAL_STATUS_OK
fi
exit $RETVAL
;;
restart)
stop

3
maxscale.spec
View File

@ -39,9 +39,11 @@ ln -s /lib64/libaio.so.1 /lib64/libaio.so
%install
mkdir -p $RPM_BUILD_ROOT/etc/ld.so.conf.d/
mkdir -p $RPM_BUILD_ROOT/etc/init.d/
mkdir -p $RPM_BUILD_ROOT%{install_path}
cp -r binaries/* $RPM_BUILD_ROOT%{install_path}
cp maxscale.conf $RPM_BUILD_ROOT/etc/ld.so.conf.d/
cp etc/init.d/maxscale $RPM_BUILD_ROOT/etc/init.d/
%clean
@ -49,5 +51,6 @@ cp maxscale.conf $RPM_BUILD_ROOT/etc/ld.so.conf.d/
%defattr(-,root,root)
%{install_path}
/etc/ld.so.conf.d/maxscale.conf
/etc/init.d/maxscale
%changelog

30
server/core/config.c
View File

@ -230,6 +230,8 @@ int error_count = 0;
config_get_value(obj->parameters, "passwd");
char *enable_root_user =
config_get_value(obj->parameters, "enable_root_user");
char *weightby =
config_get_value(obj->parameters, "weightby");
char *version_string = config_get_value(obj->parameters, "version_string");
@ -259,6 +261,8 @@ int error_count = 0;
if (enable_root_user)
serviceEnableRootUser(obj->element, config_truth_value(enable_root_user));
if (weightby)
serviceWeightBy(obj->element, weightby);
if (!auth)
auth = config_get_value(obj->parameters, "auth");
@ -362,6 +366,30 @@ int error_count = 0;
"defined but no corresponding password.",
obj->object)));
}
if (obj->element)
{
CONFIG_PARAMETER *params = obj->parameters;
while (params)
{
if (strcmp(params->name, "address")
&& strcmp(params->name, "port")
&& strcmp(params->name,
"protocol")
&& strcmp(params->name,
"monitoruser")
&& strcmp(params->name,
"monitorpw")
&& strcmp(params->name,
"type")
)
{
serverAddParameter(obj->element,
params->name,
params->value);
}
params = params->next;
}
}
}
else if (!strcmp(type, "filter"))
{
@ -1238,8 +1266,6 @@ int i;
{
if (!strcmp(type, "service"))
param_set = service_params;
else if (!strcmp(type, "server"))
param_set = server_params;
else if (!strcmp(type, "listener"))
param_set = listener_params;
else if (!strcmp(type, "monitor"))

41
server/core/dcb.c
View File

@ -1270,6 +1270,41 @@ DCB *dcb;
spinlock_release(&dcbspin);
}
/**
* Diagnotic routine to print client DCB data in a tabular form.
*
* @param pdcb DCB to print results to
*/
void
dListClients(DCB *pdcb)
{
DCB *dcb;
spinlock_acquire(&dcbspin);
dcb = allDCBs;
dcb_printf(pdcb, "Client Connections\n");
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n");
dcb_printf(pdcb, " %-15s | %-10s | %-20s | %s\n",
"Client", "DCB", "Service", "Session");
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n");
while (dcb)
{
if (dcb_isclient(dcb)
&& dcb->dcb_role == DCB_ROLE_REQUEST_HANDLER)
{
dcb_printf(pdcb, " %-15s | %10p | %-20s | %10p\n",
(dcb->remote ? dcb->remote : ""),
dcb, (dcb->session->service ?
dcb->session->service->name : ""),
dcb->session);
}
dcb = dcb->next;
}
dcb_printf(pdcb, "-----------------+------------+----------------------+------------\n\n");
spinlock_release(&dcbspin);
}
/**
* Diagnostic to print a DCB to another DCB
*
@ -1281,8 +1316,14 @@ dprintDCB(DCB *pdcb, DCB *dcb)
{
dcb_printf(pdcb, "DCB: %p\n", (void *)dcb);
dcb_printf(pdcb, "\tDCB state: %s\n", gw_dcb_state2string(dcb->state));
if (dcb->session && dcb->session->service)
dcb_printf(pdcb, "\tService: %s\n",
dcb->session->service->name);
if (dcb->remote)
dcb_printf(pdcb, "\tConnected to: %s\n", dcb->remote);
if (dcb->user)
dcb_printf(pdcb, "\tUsername: %s\n",
dcb->user);
dcb_printf(pdcb, "\tOwning Session: %p\n", dcb->session);
if (dcb->writeq)
dcb_printf(pdcb, "\tQueued write data: %d\n", gwbuf_length(dcb->writeq));

84
server/core/gateway.c
View File

@ -31,10 +31,11 @@
* 19/06/13 Mark Riddoch Extract the epoll functionality
* 21/06/13 Mark Riddoch Added initial config support
* 27/06/13
* 28/06/13 Vilho Raatikka Added necessary headers, example functions and
* calls to log manager and to query classifier.
* Put example code behind SS_DEBUG macros.
* 28/06/13 Vilho Raatikka Added necessary headers, example functions and
* calls to log manager and to query classifier.
* Put example code behind SS_DEBUG macros.
* 05/02/14 Mark Riddoch Addition of version string
* 29/06/14 Massimiliano Pinto Addition of pidfile
*
* @endverbatim
*/
@ -109,6 +110,9 @@ static char* server_groups[] = {
/* The data directory we created for this gateway instance */
static char datadir[PATH_MAX+1] = "";
/* The data directory we created for this gateway instance */
static char pidfile[PATH_MAX+1] = "";
/**
* exit flag for log flusher.
*/
@ -126,6 +130,8 @@ static bool daemon_mode = true;
static void log_flush_shutdown(void);
static void log_flush_cb(void* arg);
static int write_pid_file(char *); /* write MaxScale pidfile */
static void unlink_pidfile(void); /* remove pidfile */
static void libmysqld_done(void);
static bool file_write_header(FILE* outfile);
static bool file_write_footer(FILE* outfile);
@ -1018,6 +1024,7 @@ int main(int argc, char **argv)
goto return_main;
}
}
if (!daemon_mode)
{
fprintf(stderr,
@ -1137,6 +1144,8 @@ int main(int argc, char **argv)
rc = MAXSCALE_INTERNALERROR;
goto return_main;
}
/* register exit function for embedded MySQL library */
l = atexit(libmysqld_done);
if (l != 0) {
@ -1148,6 +1157,7 @@ int main(int argc, char **argv)
rc = MAXSCALE_INTERNALERROR;
goto return_main;
}
/*<
* If MaxScale home directory wasn't set by command-line argument.
* Next, resolve it from environment variable and further on,
@ -1199,7 +1209,7 @@ int main(int argc, char **argv)
rc = MAXSCALE_BADCONFIG;
goto return_main;
}
/*<
* Set a data directory for the mysqld library, we use
* a unique directory name to avoid clauses if multiple
@ -1322,7 +1332,11 @@ int main(int argc, char **argv)
LOGFILE_MESSAGE,
"MaxScale is running in process %i",
getpid())));
/* Write process pid into MaxScale pidfile */
write_pid_file(home_dir);
/* Init MaxScale poll system */
poll_init();
/*<
@ -1363,6 +1377,7 @@ int main(int argc, char **argv)
* Serve clients.
*/
poll_waitevents((void *)0);
/*<
* Wait server threads' completion.
*/
@ -1389,6 +1404,9 @@ int main(int argc, char **argv)
LOGFILE_MESSAGE,
"MaxScale shutdown completed.")));
/* Remove Pidfile */
unlink_pidfile();
return_main:
return rc;
} /*< End of main */
@ -1435,3 +1453,59 @@ static void log_flush_cb(
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE,
"Finished MaxScale log flusher.")));
}
/**
* Unlink pid file, called at program exit
*/
static void unlink_pidfile(void)
{
if (strlen(pidfile)) {
if (unlink(pidfile)) {
fprintf(stderr, "MaxScale failed to remove pidfile %s: error %d, %s\n", pidfile, errno, strerror(errno));
}
}
}
/**
* Write process pid into pidfile anc close it
* Parameters:
* @param home_dir The MaxScale home dir
* @return 0 on success, 1 on failure
*
*/
static int write_pid_file(char *home_dir) {
int fd = -1;
snprintf(pidfile, PATH_MAX, "%s/log/maxscale.pid", home_dir);
fd = open(pidfile, O_WRONLY | O_CREAT | O_TRUNC, 0777);
if (fd == -1) {
fprintf(stderr, "MaxScale failed to open pidFile %s: error %d, %s\n", pidfile, errno, strerror(errno));
return 1;
} else {
char pidstr[50]="";
/* truncate pidfile content */
if (ftruncate(fd, 0) == -1) {
fprintf(stderr, "MaxScale failed to truncate pidfile %s: error %d, %s\n", pidfile, errno, strerror(errno));
}
snprintf(pidstr, sizeof(pidstr)-1, "%d", getpid());
if (pwrite(fd, pidstr, strlen(pidstr), 0) != (ssize_t)strlen(pidstr)) {
fprintf(stderr, "MaxScale failed to write into pidfile %s: error %d, %s\n", pidfile, errno, strerror(errno));
/* close file and return */
close(fd);
return 1;
}
/* close file */
close(fd);
fprintf(stderr, "MaxScale PID %s in pidfile %s\n", pidstr, pidfile);
}
/* success */
return 0;
}

4
server/core/load_utils.c
View File

@ -379,7 +379,9 @@ MODULES *ptr = registered;
: (ptr->info->status == MODULE_BETA_RELEASE
? "Beta"
: (ptr->info->status == MODULE_GA
? "GA" : "Unknown"))));
? "GA"
: (ptr->info->status == MODULE_EXPERIMENTAL
? "Experimental" : "Unknown")))));
dcb_printf(dcb, "\n");
ptr = ptr->next;
}

46
server/core/monitor.c
View File

@ -73,6 +73,7 @@ MONITOR *mon;
return NULL;
}
mon->handle = (*mon->module->startMonitor)(NULL);
mon->state |= MONITOR_STATE_RUNNING;
spinlock_acquire(&monLock);
mon->next = allMonitors;
allMonitors = mon;
@ -93,6 +94,7 @@ monitor_free(MONITOR *mon)
MONITOR *ptr;
mon->module->stopMonitor(mon->handle);
mon->state &= ~MONITOR_STATE_RUNNING;
spinlock_acquire(&monLock);
if (allMonitors == mon)
allMonitors = mon->next;
@ -119,6 +121,7 @@ void
monitorStart(MONITOR *monitor)
{
monitor->handle = (*monitor->module->startMonitor)(monitor->handle);
monitor->state |= MONITOR_STATE_RUNNING;
}
/**
@ -130,6 +133,7 @@ void
monitorStop(MONITOR *monitor)
{
monitor->module->stopMonitor(monitor->handle);
monitor->state &= ~MONITOR_STATE_RUNNING;
}
/**
@ -200,6 +204,47 @@ MONITOR *ptr;
spinlock_release(&monLock);
}
/**
* Show a single monitor
*
* @param dcb DCB for printing output
*/
void
monitorShow(DCB *dcb, MONITOR *monitor)
{
dcb_printf(dcb, "Monitor: %p\n", monitor);
dcb_printf(dcb, "\tName: %s\n", monitor->name);
if (monitor->module->diagnostics)
monitor->module->diagnostics(dcb, monitor->handle);
}
/**
* List all the monitors
*
* @param dcb DCB for printing output
*/
void
monitorList(DCB *dcb)
{
MONITOR *ptr;
spinlock_acquire(&monLock);
ptr = allMonitors;
dcb_printf(dcb, "+----------------------+---------------------\n");
dcb_printf(dcb, "| %-20s | Status\n", "Monitor");
dcb_printf(dcb, "+----------------------+---------------------\n");
while (ptr)
{
dcb_printf(dcb, "| %-20s | %s\n", ptr->name,
ptr->state & MONITOR_STATE_RUNNING
? "Running" : "Stopped");
ptr = ptr->next;
}
dcb_printf(dcb, "+----------------------+---------------------\n");
spinlock_release(&monLock);
}
/**
* Find a monitor by name
*
@ -249,6 +294,7 @@ void
monitorSetInterval (MONITOR *mon, unsigned long interval)
{
if (mon->module->setInterval != NULL) {
mon->interval = interval;
mon->module->setInterval(mon->handle, interval);
}
}

110
server/core/server.c
View File

@ -28,6 +28,8 @@
* 20/05/14 Massimiliano Pinto Addition of server_string
* 21/05/14 Massimiliano Pinto Addition of node_id
* 28/05/14 Massimiliano Pinto Addition of rlagd and node_ts fields
* 20/06/14 Massimiliano Pinto Addition of master_id, depth, slaves fields
* 26/06/14 Mark Riddoch Addition of server parameters
*
* @endverbatim
*/
@ -76,6 +78,10 @@ SERVER *server;
server->node_id = -1;
server->rlag = -1;
server->node_ts = 0;
server->parameters = NULL;
server->master_id = -1;
server->depth = -1;
server->slaves = NULL;
spinlock_acquire(&server_spin);
server->next = allServers;
@ -250,7 +256,21 @@ char *stat;
if (ptr->server_string)
dcb_printf(dcb, "\tServer Version:\t\t%s\n", ptr->server_string);
dcb_printf(dcb, "\tNode Id: %d\n", ptr->node_id);
if (SERVER_IS_SLAVE(ptr)) {
dcb_printf(dcb, "\tMaster Id: %d\n", ptr->master_id);
if (ptr->slaves) {
int i;
dcb_printf(dcb, "\tSlave Ids: ");
for (i = 0; ptr->slaves[i]; i++)
{
if (i == 0)
dcb_printf(dcb, "%li", ptr->slaves[i]);
else
dcb_printf(dcb, ", %li ", ptr->slaves[i]);
}
dcb_printf(dcb, "\n");
}
dcb_printf(dcb, "\tRepl Depth: %d\n", ptr->depth);
if (SERVER_IS_SLAVE(ptr) || SERVER_IS_RELAY_SERVER(ptr)) {
if (ptr->rlag >= 0) {
dcb_printf(dcb, "\tSlave delay:\t\t%d\n", ptr->rlag);
}
@ -275,7 +295,8 @@ char *stat;
void
dprintServer(DCB *dcb, SERVER *server)
{
char *stat;
char *stat;
SERVER_PARAM *param;
dcb_printf(dcb, "Server %p (%s)\n", server, server->unique_name);
dcb_printf(dcb, "\tServer: %s\n", server->name);
@ -287,13 +308,38 @@ char *stat;
if (server->server_string)
dcb_printf(dcb, "\tServer Version:\t\t%s\n", server->server_string);
dcb_printf(dcb, "\tNode Id: %d\n", server->node_id);
if (SERVER_IS_SLAVE(server)) {
dcb_printf(dcb, "\tMaster Id: %d\n", server->master_id);
if (server->slaves) {
int i;
dcb_printf(dcb, "\tSlave Ids: ");
for (i = 0; server->slaves[i]; i++)
{
if (i == 0)
dcb_printf(dcb, "%li", server->slaves[i]);
else
dcb_printf(dcb, ", %li ", server->slaves[i]);
}
dcb_printf(dcb, "\n");
}
dcb_printf(dcb, "\tRepl Depth: %d\n", server->depth);
if (SERVER_IS_SLAVE(server) || SERVER_IS_RELAY_SERVER(server)) {
if (server->rlag >= 0) {
dcb_printf(dcb, "\tSlave delay:\t\t%d\n", server->rlag);
}
}
if (server->node_ts > 0) {
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%lu\n", server->node_ts);
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%s",
asctime(localtime(&server->node_ts)));
}
if ((param = server->parameters) != NULL)
{
dcb_printf(dcb, "\tServer Parameters:\n");
while (param)
{
dcb_printf(dcb, "\t\t%-20s %s\n", param->name,
param->value);
param = param->next;
}
}
dcb_printf(dcb, "\tNumber of connections: %d\n", server->stats.n_connections);
dcb_printf(dcb, "\tCurrent no. of conns: %d\n", server->stats.n_current);
@ -446,3 +492,59 @@ server_update(SERVER *server, char *protocol, char *user, char *passwd)
}
}
/**
* Add a server parameter to a server.
*
* Server parameters may be used by routing to weight the load
* balancing they apply to the server.
*
* @param server The server we are adding the parameter to
* @param name The parameter name
* @param value The parameter value
*/
void
serverAddParameter(SERVER *server, char *name, char *value)
{
SERVER_PARAM *param;
if ((param = (SERVER_PARAM *)malloc(sizeof(SERVER_PARAM))) == NULL)
{
return;
}
if ((param->name = strdup(name)) == NULL)
{
free(param);
return;
}
if ((param->value = strdup(value)) == NULL)
{
free(param->value);
free(param);
return;
}
param->next = server->parameters;
server->parameters = param;
}
/**
* Retreive a parameter value from a server
*
* @param server The server we are looking for a parameter of
* @param name The name of the parameter we require
* @return The parameter value or NULL if not found
*/
char *
serverGetParameter(SERVER *server, char *name)
{
SERVER_PARAM *param = server->parameters;
while (param)
{
if (strcmp(param->name, name) == 0)
return param->value;
param = param->next;
}
return NULL;
}

62
server/core/service.c
View File

@ -114,6 +114,7 @@ SERVICE *service;
service->svc_config_version = 0;
service->filters = NULL;
service->n_filters = 0;
service->weightby = 0;
spinlock_init(&service->spin);
spinlock_init(&service->users_table_spin);
memset(&service->rate_limit, 0, sizeof(SERVICE_REFRESH_RATE));
@ -716,8 +717,8 @@ SERVER *ptr = service->databases;
int i;
printf("Service %p\n", service);
printf("\tService: %s\n", service->name);
printf("\tRouter: %s (%p)\n", service->routerModule, service->router);
printf("\tService: %s\n", service->name);
printf("\tRouter: %s (%p)\n", service->routerModule, service->router);
printf("\tStarted: %s", asctime(localtime(&service->stats.started)));
printf("\tBackend databases\n");
while (ptr)
@ -794,13 +795,16 @@ SERVER *server = service->databases;
int i;
dcb_printf(dcb, "Service %p\n", service);
dcb_printf(dcb, "\tService: %s\n", service->name);
dcb_printf(dcb, "\tRouter: %s (%p)\n", service->routerModule,
service->router);
dcb_printf(dcb, "\tService: %s\n",
service->name);
dcb_printf(dcb, "\tRouter: %s (%p)\n",
service->routerModule, service->router);
if (service->router)
service->router->diagnostics(service->router_instance, dcb);
dcb_printf(dcb, "\tStarted: %s",
dcb_printf(dcb, "\tStarted: %s",
asctime(localtime(&service->stats.started)));
dcb_printf(dcb, "\tRoot user access: %s\n",
service->enable_root ? "Enabled" : "Disabled");
if (service->n_filters)
{
dcb_printf(dcb, "\tFilter chain: ");
@ -818,9 +822,15 @@ int i;
server->protocol);
server = server->nextdb;
}
dcb_printf(dcb, "\tUsers data: %p\n", service->users);
dcb_printf(dcb, "\tTotal connections: %d\n", service->stats.n_sessions);
dcb_printf(dcb, "\tCurrently connected: %d\n", service->stats.n_current);
if (service->weightby)
dcb_printf(dcb, "\tRouting weight parameter: %s\n",
service->weightby);
dcb_printf(dcb, "\tUsers data: %p\n",
service->users);
dcb_printf(dcb, "\tTotal connections: %d\n",
service->stats.n_sessions);
dcb_printf(dcb, "\tCurrently connected: %d\n",
service->stats.n_current);
}
/**
@ -1097,8 +1107,38 @@ static void service_add_qualified_param(
spinlock_release(&svc->spin);
}
char* service_get_name(
SERVICE* svc)
/**
* Return the name of the service
*
* @param svc The service
*/
char *
service_get_name(SERVICE *svc)
{
return svc->name;
}
/**
* Set the weighting parameter for the service
*
* @param service The service pointer
* @param weightby The parameter name to weight the routing by
*/
void
serviceWeightBy(SERVICE *service, char *weightby)
{
if (service->weightby)
free(service->weightby);
service->weightby = strdup(weightby);
}
/**
* Return the parameter the wervice shoudl use to weight connections
* by
* @param service The Service pointer
*/
char *
serviceGetWeightingParameter(SERVICE *service)
{
return service->weightby;
}

3
server/include/dcb.h
View File

@ -281,6 +281,7 @@ void printDCB(DCB *); /* Debug print routine */
void dprintAllDCBs(DCB *); /* Debug to print all DCB in the system */
void dprintDCB(DCB *, DCB *); /* Debug to print a DCB in the system */
void dListDCBs(DCB *); /* List all DCBs in the system */
void dListClients(DCB *); /* List al the client DCBs */
const char *gw_dcb_state2string(int); /* DCB state to string */
void dcb_printf(DCB *, const char *, ...); /* DCB version of printf */
int dcb_isclient(DCB *); /* the DCB is the client of the session */
@ -299,6 +300,8 @@ bool dcb_set_state(
void dcb_call_foreach (DCB_REASON reason);
void dcb_call_foreach (
DCB_REASON reason);
/* DCB flags values */
#define DCBF_CLONE 0x0001 /* DCB is a clone */

3
server/include/modinfo.h
View File

@ -38,7 +38,8 @@ typedef enum {
MODULE_IN_DEVELOPMENT = 0,
MODULE_ALPHA_RELEASE,
MODULE_BETA_RELEASE,
MODULE_GA
MODULE_GA,
MODULE_EXPERIMENTAL
} MODULE_STATUS;
/**

10
server/include/monitor.h
View File

@ -78,13 +78,21 @@ typedef struct {
*/
#define MONITOR_VERSION {1, 0, 0}
/**
* Monitor state bit mask values
*/
#define MONITOR_STATE_RUNNING 0x0001
/**
* Representation of the running monitor.
*/
typedef struct monitor {
char *name; /**< The name of the monitor module */
unsigned int state; /**< The monitor status */
MONITOR_OBJECT *module; /**< The "monitor object" */
void *handle; /**< Handle returned from startMonitor */
int interval; /**< The monitor interval */
struct monitor *next; /**< Next monitor in the linked list */
} MONITOR;
@ -97,6 +105,8 @@ extern void monitorStop(MONITOR *);
extern void monitorStart(MONITOR *);
extern void monitorStopAll();
extern void monitorShowAll(DCB *);
extern void monitorShow(DCB *, MONITOR *);
extern void monitorList(DCB *);
extern void monitorSetId(MONITOR *, unsigned long);
extern void monitorSetInterval (MONITOR *, unsigned long);
extern void monitorSetReplicationHeartbeat(MONITOR *, int);

33
server/include/server.h
View File

@ -36,10 +36,23 @@
* 20/05/14 Massimiliano Pinto Addition of node_id field
* 23/05/14 Massimiliano Pinto Addition of rlag and node_ts fields
* 03/06/14 Mark Riddoch Addition of maintainance mode
* 20/06/14 Massimiliano Pinto Addition of master_id, depth, slaves fields
* 26/06/14 Mark Riddoch Adidtion of server parameters
*
* @endverbatim
*/
/**
* The server parameters used for weighting routing decissions
*
*/
typedef struct server_params {
char *name; /**< Parameter name */
char *value; /**< Parameter value */
struct server_params
*next; /**< Next Paramter in the linked list */
} SERVER_PARAM;
/**
* The server statistics structure
*
@ -71,6 +84,10 @@ typedef struct server {
long node_id; /**< Node id, server_id for M/S or local_index for Galera */
int rlag; /**< Replication Lag for Master / Slave replication */
unsigned long node_ts; /**< Last timestamp set from M/S monitor module */
SERVER_PARAM *parameters; /**< Parameters of a server that may be used to weight routing decisions */
long master_id; /**< Master server id of this node */
int depth; /**< Replication level in the tree */
long *slaves; /**< Slaves of this node */
} SERVER;
/**
@ -78,11 +95,12 @@ typedef struct server {
*
* These are a bitmap of attributes that may be applied to a server
*/
#define SERVER_RUNNING 0x0001 /**<< The server is up and running */
#define SERVER_MASTER 0x0002 /**<< The server is a master, i.e. can handle writes */
#define SERVER_SLAVE 0x0004 /**<< The server is a slave, i.e. can handle reads */
#define SERVER_JOINED 0x0008 /**<< The server is joined in a Galera cluster */
#define SERVER_MAINT 0x1000 /**<< Server is in maintenance mode */
#define SERVER_RUNNING 0x0001 /**<< The server is up and running */
#define SERVER_MASTER 0x0002 /**<< The server is a master, i.e. can handle writes */
#define SERVER_SLAVE 0x0004 /**<< The server is a slave, i.e. can handle reads */
#define SERVER_JOINED 0x0008 /**<< The server is joined in a Galera cluster */
#define SERVER_MAINT 0x1000 /**<< Server is in maintenance mode */
#define SERVER_SLAVE_OF_EXTERNAL_MASTER 0x0016 /**<< Server is slave of a Master outside the provided replication topology */
/**
* Is the server running - the macro returns true if the server is marked as running
@ -123,6 +141,9 @@ typedef struct server {
#define SERVER_IS_IN_CLUSTER(s) (((s)->status & (SERVER_MASTER|SERVER_SLAVE|SERVER_JOINED)) != 0)
#define SERVER_IS_RELAY_SERVER(server) \
(((server)->status & (SERVER_RUNNING|SERVER_MASTER|SERVER_SLAVE|SERVER_MAINT)) == (SERVER_RUNNING|SERVER_MASTER|SERVER_SLAVE))
extern SERVER *server_alloc(char *, char *, unsigned short);
extern int server_free(SERVER *);
extern SERVER *server_find_by_unique_name(char *);
@ -136,6 +157,8 @@ extern char *server_status(SERVER *);
extern void server_set_status(SERVER *, int);
extern void server_clear_status(SERVER *, int);
extern void serverAddMonUser(SERVER *, char *, char *);
extern void serverAddParameter(SERVER *, char *, char *);
extern char *serverGetParameter(SERVER *, char *);
extern void server_update(SERVER *, char *, char *, char *);
extern void server_set_unique_name(SERVER *, char *);
#endif

4
server/include/service.h
View File

@ -43,6 +43,7 @@
* 07/05/14 Massimiliano Pinto Added version_string field to service
* struct
* 29/05/14 Mark Riddoch Filter API mechanism
* 26/06/14 Mark Riddoch Added WeightBy support
*
* @endverbatim
*/
@ -130,6 +131,7 @@ typedef struct service {
rate_limit; /**< The refresh rate limit for users table */
FILTER_DEF **filters; /**< Ordered list of filters */
int n_filters; /**< Number of filters */
char *weightby;
struct service *next; /**< The next service in the linked list */
} SERVICE;
@ -157,6 +159,8 @@ extern int serviceSetUser(SERVICE *, char *, char *);
extern int serviceGetUser(SERVICE *, char **, char **);
extern void serviceSetFilters(SERVICE *, char *);
extern int serviceEnableRootUser(SERVICE *, int );
extern void serviceWeightBy(SERVICE *, char *);
extern char *serviceGetWeightingParameter(SERVICE *);
extern void service_update(SERVICE *, char *, char *, char *);
extern int service_refresh_users(SERVICE *);
extern void printService(SERVICE *);

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

@ -390,11 +390,24 @@ struct timeval tv;
static void
diagnostic(FILTER *instance, void *fsession, DCB *dcb)
{
QLA_INSTANCE *my_instance = (QLA_INSTANCE *)instance;
QLA_SESSION *my_session = (QLA_SESSION *)fsession;
if (my_session)
{
dcb_printf(dcb, "\t\tLogging to file %s.\n",
dcb_printf(dcb, "\t\tLogging to file %s.\n",
my_session->filename);
}
if (my_instance->source)
dcb_printf(dcb, "\t\tLimit logging to connections from %s\n",
my_instance->source);
if (my_instance->userName)
dcb_printf(dcb, "\t\tLimit logging to user %s\n",
my_instance->userName);
if (my_instance->match)
dcb_printf(dcb, "\t\tInclude queries that match %s\n",
my_instance->match);
if (my_instance->nomatch)
dcb_printf(dcb, "\t\tExclude queries that match %s\n",
my_instance->nomatch);
}

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

@ -345,6 +345,14 @@ REGEX_SESSION *my_session = (REGEX_SESSION *)fsession;
dcb_printf(dcb, "\t\tNo. of queries altered by filter: %d\n",
my_session->replacements);
}
if (my_instance->source)
dcb_printf(dcb,
"\t\tReplacement limited to connections from %s\n",
my_instance->source);
if (my_instance->user)
dcb_printf(dcb,
"\t\tReplacement limit to user %s\n",
my_instance->user);
}
/**

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

@ -118,7 +118,8 @@ typedef struct {
int active; /* filter is active? */
DCB *branch_dcb; /* Client DCB for "branch" service */
SESSION *branch_session;/* The branch service session */
int n_duped; /* Number of duplicated querise */
int n_duped; /* Number of duplicated queries */
int n_rejected; /* Number of rejected queries */
int residual; /* Any outstanding SQL text */
} TEE_SESSION;
@ -418,6 +419,10 @@ GWBUF *clone = NULL;
my_session->n_duped++;
SESSION_ROUTE_QUERY(my_session->branch_session, clone);
}
else
{
my_session->n_rejected++;
}
return rval;
}
@ -435,11 +440,28 @@ GWBUF *clone = NULL;
static void
diagnostic(FILTER *instance, void *fsession, DCB *dcb)
{
TEE_INSTANCE *my_instance = (TEE_INSTANCE *)instance;
TEE_SESSION *my_session = (TEE_SESSION *)fsession;
if (my_instance->source)
dcb_printf(dcb, "\t\tLimit to connections from %s\n",
my_instance->source);
dcb_printf(dcb, "\t\tDuplicate statements to service %s\n",
my_instance->service->name);
if (my_instance->userName)
dcb_printf(dcb, "\t\tLimit to user %s\n",
my_instance->userName);
if (my_instance->match)
dcb_printf(dcb, "\t\tInclude queries that match %s\n",
my_instance->match);
if (my_instance->nomatch)
dcb_printf(dcb, "\t\tExclude queries that match %s\n",
my_instance->nomatch);
if (my_session)
{
dcb_printf(dcb, "\t\tNo. of statements duplicated: %d.\n",
my_session->n_duped);
dcb_printf(dcb, "\t\tNo. of statements rejected: %d.\n",
my_session->n_rejected);
}
}

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

@ -539,11 +539,40 @@ int i, inserted;
static void
diagnostic(FILTER *instance, void *fsession, DCB *dcb)
{
TOPN_INSTANCE *my_instance = (TOPN_INSTANCE *)instance;
TOPN_SESSION *my_session = (TOPN_SESSION *)fsession;
int i;
dcb_printf(dcb, "\t\tReport size %d\n",
my_instance->topN);
if (my_instance->source)
dcb_printf(dcb, "\t\tLimit logging to connections from %s\n",
my_instance->source);
if (my_instance->user)
dcb_printf(dcb, "\t\tLimit logging to user %s\n",
my_instance->user);
if (my_instance->match)
dcb_printf(dcb, "\t\tInclude queries that match %s\n",
my_instance->match);
if (my_instance->exclude)
dcb_printf(dcb, "\t\tExclude queries that match %s\n",
my_instance->exclude);
if (my_session)
{
dcb_printf(dcb, "\t\tLogging to file %s.\n",
my_session->filename);
dcb_printf(dcb, "\t\tCurrent Top %d:\n", my_instance->topN);
for (i = 0; i < my_instance->topN; i++)
{
if (my_session->top[i]->sql)
{
dcb_printf(dcb, "\t\t%d place:\n", i + 1);
dcb_printf(dcb, "\t\t\tExecution time: %.3f seconds\n",
(double)((my_session->top[i]->duration.tv_sec * 1000)
+ (my_session->top[i]->duration.tv_usec / 1000)) / 1000);
dcb_printf(dcb, "\t\t\tSQL: %s\n",
my_session->top[i]->sql);
}
}
}
}

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

@ -26,6 +26,7 @@
*
* Date Who Description
* 14/06/13 Mark Riddoch Initial implementation
* 27/06/14 Mark Riddoch Addition of server weight percentage
*
* @endverbatim
*/
@ -39,6 +40,7 @@
typedef struct backend {
SERVER *server; /*< The server itself */
int current_connection_count; /*< Number of connections to the server */
int weight; /*< Desired routing weight */
} BACKEND;
/**

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

@ -29,6 +29,7 @@
* 23/05/14 Massimiliano Pinto Added 1 configuration option (setInterval).
* Interval is printed in diagnostics.
* 03/06/14 Mark Riddoch Add support for maintenance mode
* 24/06/14 Massimiliano Pinto Added depth level 0 for each node
*
* @endverbatim
*/
@ -434,6 +435,7 @@ long master_id;
/* set master_id to the lowest value of ptr->server->node_id */
if ((! SERVER_IN_MAINT(ptr->server)) && ptr->server->node_id >= 0 && SERVER_IS_JOINED(ptr->server)) {
ptr->server->depth = 0;
if (ptr->server->node_id < master_id && master_id >= 0) {
master_id = ptr->server->node_id;
} else {
@ -445,6 +447,7 @@ long master_id;
/* clear M/S status */
server_clear_status(ptr->server, SERVER_SLAVE);
server_clear_status(ptr->server, SERVER_MASTER);
ptr->server->depth = -1;
}
if (ptr->server->status != prev_status ||
SERVER_IS_DOWN(ptr->server))

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

@ -33,6 +33,13 @@
* 28/05/14 Massimiliano Pinto Added set Id and configuration options (setInverval)
* Parameters are now printed in diagnostics
* 03/06/14 Mark Ridoch Add support for maintenance mode
* 17/06/14 Massimiliano Pinto Addition of getServerByNodeId routine
* and first implementation for depth of replication for nodes.
* 23/06/14 Massimiliano Pinto Added replication consistency after replication tree computation
* 27/06/14 Massimiliano Pinto Added replication pending status in monitored server, storing there
* the status to update in server status field before
* starting the replication consistency check.
* This will also give routers a consistent "status" of all servers
*
* @endverbatim
*/
@ -75,6 +82,14 @@ static void defaultId(void *, unsigned long);
static void replicationHeartbeat(void *, 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);
static MONITOR_SERVERS *getSlaveOfNodeId(MONITOR_SERVERS *, long);
static MONITOR_SERVERS *get_replication_tree(MYSQL_MONITOR *, int);
static void set_master_heartbeat(MYSQL_MONITOR *, MONITOR_SERVERS *);
static void set_slave_heartbeat(MYSQL_MONITOR *, MONITOR_SERVERS *);
static int add_slave_to_master(long *, int, long);
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, defaultId, replicationHeartbeat };
@ -145,6 +160,7 @@ MYSQL_MONITOR *handle;
handle->id = MONITOR_DEFAULT_ID;
handle->interval = MONITOR_INTERVAL;
handle->replicationHeartbeat = 0;
handle->master = NULL;
spinlock_init(&handle->lock);
}
handle->tid = (THREAD)thread_start(monitorMain, handle);
@ -185,6 +201,9 @@ MONITOR_SERVERS *ptr, *db;
db->next = NULL;
db->mon_err_count = 0;
db->mon_prev_status = 0;
/* pending status is updated by get_replication_tree */
db->pending_status = 0;
spinlock_acquire(&handle->lock);
if (handle->databases == NULL)
@ -316,7 +335,6 @@ monitorDatabase(MYSQL_MONITOR *handle, MONITOR_SERVERS *database)
MYSQL_ROW row;
MYSQL_RES *result;
int num_fields;
int ismaster = 0;
int isslave = 0;
char *uname = handle->defaultUser;
char *passwd = handle->defaultPasswd;
@ -372,15 +390,28 @@ int replication_heartbeat = handle->replicationHeartbeat;
database->server->port,
mysql_error(database->con))));
}
/** Store current status */
/* The current server is not running
*
* Store server NOT running in server and monitor server pending struct
*
*/
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);
return;
}
free(dpwd);
}
/** Store current status */
server_set_status(database->server, SERVER_RUNNING);
}
/* 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);
@ -410,121 +441,6 @@ int replication_heartbeat = handle->replicationHeartbeat;
mysql_free_result(result);
}
/* Check SHOW SLAVE HOSTS - if we get rows then we are a master */
if (mysql_query(database->con, "SHOW SLAVE HOSTS"))
{
if (mysql_errno(database->con) == ER_SPECIFIC_ACCESS_DENIED_ERROR)
{
/* Log lack of permission */
}
database->server->rlag = -1;
} else if ((result = mysql_store_result(database->con)) != NULL) {
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result)))
{
ismaster = 1;
}
mysql_free_result(result);
if (ismaster && replication_heartbeat == 1) {
time_t heartbeat;
time_t purge_time;
char heartbeat_insert_query[128]="";
char heartbeat_purge_query[128]="";
handle->master_id = database->server->node_id;
/* create the maxscale_schema database */
if (mysql_query(database->con, "CREATE DATABASE IF NOT EXISTS maxscale_schema")) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error creating maxscale_schema database in Master server"
": %s", mysql_error(database->con))));
database->server->rlag = -1;
}
/* create repl_heartbeat table in maxscale_schema database */
if (mysql_query(database->con, "CREATE TABLE IF NOT EXISTS "
"maxscale_schema.replication_heartbeat "
"(maxscale_id INT NOT NULL, "
"master_server_id INT NOT NULL, "
"master_timestamp INT UNSIGNED NOT NULL, "
"PRIMARY KEY ( master_server_id, maxscale_id ) ) "
"ENGINE=MYISAM DEFAULT CHARSET=latin1")) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error creating maxscale_schema.replication_heartbeat table in Master server"
": %s", mysql_error(database->con))));
database->server->rlag = -1;
}
/* auto purge old values after 48 hours*/
purge_time = time(0) - (3600 * 48);
sprintf(heartbeat_purge_query, "DELETE FROM maxscale_schema.replication_heartbeat WHERE master_timestamp < %lu", purge_time);
if (mysql_query(database->con, heartbeat_purge_query)) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error deleting from maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_purge_query,
mysql_error(database->con))));
}
heartbeat = time(0);
/* set node_ts for master as time(0) */
database->server->node_ts = heartbeat;
sprintf(heartbeat_insert_query, "UPDATE maxscale_schema.replication_heartbeat SET master_timestamp = %lu WHERE master_server_id = %i AND maxscale_id = %lu", heartbeat, handle->master_id, id);
/* Try to insert MaxScale timestamp into master */
if (mysql_query(database->con, heartbeat_insert_query)) {
database->server->rlag = -1;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error updating maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_insert_query,
mysql_error(database->con))));
} else {
if (mysql_affected_rows(database->con) == 0) {
heartbeat = time(0);
sprintf(heartbeat_insert_query, "REPLACE INTO maxscale_schema.replication_heartbeat (master_server_id, maxscale_id, master_timestamp ) VALUES ( %i, %lu, %lu)", handle->master_id, id, heartbeat);
if (mysql_query(database->con, heartbeat_insert_query)) {
database->server->rlag = -1;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error inserting into maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_insert_query,
mysql_error(database->con))));
} else {
/* Set replication lag to 0 for the master */
database->server->rlag = 0;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: heartbeat table inserted data for %s:%i", database->server->name, database->server->port)));
}
} else {
/* Set replication lag as 0 for the master */
database->server->rlag = 0;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: heartbeat table updated for %s:%i", database->server->name, database->server->port)));
}
}
}
}
/* Check if the Slave_SQL_Running and Slave_IO_Running status is
* set to Yes
*/
@ -536,17 +452,28 @@ int replication_heartbeat = handle->replicationHeartbeat;
&& (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;
/* 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
@ -556,105 +483,45 @@ int replication_heartbeat = handle->replicationHeartbeat;
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)
&& strncmp(row[11], "Yes", 3) == 0) {
isslave = 1;
/* 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 the master_timestamp value from maxscale_schema.replication_heartbeat table */
if (isslave && replication_heartbeat == 1) {
time_t heartbeat;
char select_heartbeat_query[256] = "";
/* Remove addition info */
monitor_clear_pending_status(database, SERVER_SLAVE_OF_EXTERNAL_MASTER);
sprintf(select_heartbeat_query, "SELECT master_timestamp "
"FROM maxscale_schema.replication_heartbeat "
"WHERE maxscale_id = %lu AND master_server_id = %i",
id, handle->master_id);
/* Please note, the MASTER status and SERVER_SLAVE_OF_EXTERNAL_MASTER
* will be assigned in the monitorMain() via get_replication_tree() routine
*/
/* if there is a master then send the query to the slave with master_id*/
if (handle->master_id >= 0 && (mysql_query(database->con, select_heartbeat_query) == 0
&& (result = mysql_store_result(database->con)) != NULL)) {
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result))) {
int rlag = -1;
time_t slave_read;
heartbeat = time(0);
slave_read = strtoul(row[0], NULL, 10);
if ((errno == ERANGE && (slave_read == LONG_MAX || slave_read == LONG_MIN)) || (errno != 0 && slave_read == 0)) {
slave_read = 0;
}
if (slave_read) {
/* set the replication lag */
rlag = heartbeat - slave_read;
}
/* set this node_ts as master_timestamp read from replication_heartbeat table */
database->server->node_ts = slave_read;
if (rlag >= 0) {
/* store rlag only if greater than monitor sampling interval */
database->server->rlag = (rlag > (handle->interval / 1000)) ? rlag : 0;
} else {
database->server->rlag = -1;
}
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: replication heartbeat: "
"server %s:%i is %i seconds behind master",
database->server->name,
database->server->port,
database->server->rlag)));
}
mysql_free_result(result);
} else {
database->server->rlag = -1;
database->server->node_ts = 0;
if (handle->master_id < 0) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: error: replication heartbeat: "
"master_server_id NOT available for %s:%i",
database->server->name,
database->server->port)));
} else {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: error: replication heartbeat: "
"failed selecting from hearthbeat table of %s:%i : [%s], %s",
database->server->name,
database->server->port,
select_heartbeat_query,
mysql_error(database->con))));
}
}
}
/** Store current status */
if (ismaster)
/* Set the Slave Role */
if (isslave)
{
server_set_status(database->server, SERVER_MASTER);
server_clear_status(database->server, SERVER_SLAVE);
}
else if (isslave)
{
server_set_status(database->server, SERVER_SLAVE);
server_clear_status(database->server, SERVER_MASTER);
}
if (ismaster == 0 && isslave == 0)
{
server_clear_status(database->server, SERVER_SLAVE);
server_clear_status(database->server, SERVER_MASTER);
monitor_set_pending_status(database, SERVER_SLAVE);
/* Avoid any possible stale Master state */
monitor_clear_pending_status(database, SERVER_MASTER);
} else {
/* Avoid any possible Master/Slave stale state */
monitor_clear_pending_status(database, SERVER_SLAVE);
monitor_clear_pending_status(database, SERVER_MASTER);
}
}
@ -668,6 +535,9 @@ monitorMain(void *arg)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *ptr;
int replication_heartbeat = handle->replicationHeartbeat;
int num_servers=0;
MONITOR_SERVERS *root_master;
if (mysql_thread_init())
{
@ -690,8 +560,21 @@ MONITOR_SERVERS *ptr;
ptr = handle->databases;
while (ptr)
{
/* copy server status into monitor pending_status */
ptr->pending_status = ptr->server->status;
/* monitor current node */
monitorDatabase(handle, ptr);
/* reset the slave list of current node */
if (ptr->server->slaves) {
free(ptr->server->slaves);
}
/* create a new slave list */
ptr->server->slaves = (long *) calloc(MONITOR_MAX_NUM_SLAVES, sizeof(long));
num_servers++;
if (mon_status_changed(ptr))
{
dcb_call_foreach(DCB_REASON_NOT_RESPONDING);
@ -717,8 +600,40 @@ MONITOR_SERVERS *ptr;
/** Reset this server's error count */
ptr->mon_err_count = 0;
}
ptr = ptr->next;
}
/* Compute the replication tree */
root_master = get_replication_tree(handle, num_servers);
/* Update server status from monitor pending status on that server*/
ptr = handle->databases;
while (ptr)
{
if (! SERVER_IN_MAINT(ptr->server)) {
ptr->server->status = ptr->pending_status;
}
ptr = ptr->next;
}
/* Do now the heartbeat replication set/get for MySQL Replication Consistency */
if (replication_heartbeat && root_master && (SERVER_IS_MASTER(root_master->server) || SERVER_IS_RELAY_SERVER(root_master->server))) {
set_master_heartbeat(handle, root_master);
ptr = handle->databases;
while (ptr) {
if( (! SERVER_IN_MAINT(ptr->server)) && SERVER_IS_RUNNING(ptr->server))
{
if (ptr->server->node_id != root_master->server->node_id && (SERVER_IS_SLAVE(ptr->server) || SERVER_IS_RELAY_SERVER(ptr->server))) {
set_slave_heartbeat(handle, ptr);
}
}
ptr = ptr->next;
}
}
/* wait for the configured interval */
thread_millisleep(handle->interval);
}
}
@ -797,3 +712,380 @@ static bool mon_print_fail_status(
}
return succp;
}
/**
* Fetch a MySQL node by node_id
*
* @param ptr The list of servers to monitor
* @param node_id The MySQL server_id to fetch
* @return The server with the required server_id
*/
static MONITOR_SERVERS *
getServerByNodeId(MONITOR_SERVERS *ptr, long node_id) {
SERVER *current;
while (ptr)
{
current = ptr->server;
if (current->node_id == node_id) {
return ptr;
}
ptr = ptr->next;
}
return NULL;
}
/**
* Fetch a MySQL slave node from a node_id
*
* @param ptr The list of servers to monitor
* @param node_id The MySQL server_id to fetch
* @return The slave server of this node_id
*/
static MONITOR_SERVERS *
getSlaveOfNodeId(MONITOR_SERVERS *ptr, long node_id) {
SERVER *current;
while (ptr)
{
current = ptr->server;
if (current->master_id == node_id) {
return ptr;
}
ptr = ptr->next;
}
return NULL;
}
/*******
* This function sets the replication heartbeat
* into the maxscale_schema.replication_heartbeat table in the current master.
* The inserted values will be seen from all slaves replication from this master.
*
* @param handle The monitor handle
* @param database The number database server
*/
static void set_master_heartbeat(MYSQL_MONITOR *handle, MONITOR_SERVERS *database) {
unsigned long id = handle->id;
time_t heartbeat;
time_t purge_time;
char heartbeat_insert_query[128]="";
char heartbeat_purge_query[128]="";
/* create the maxscale_schema database */
if (mysql_query(database->con, "CREATE DATABASE IF NOT EXISTS maxscale_schema")) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error creating maxscale_schema database in Master server"
": %s", mysql_error(database->con))));
database->server->rlag = -1;
}
/* create repl_heartbeat table in maxscale_schema database */
if (mysql_query(database->con, "CREATE TABLE IF NOT EXISTS "
"maxscale_schema.replication_heartbeat "
"(maxscale_id INT NOT NULL, "
"master_server_id INT NOT NULL, "
"master_timestamp INT UNSIGNED NOT NULL, "
"PRIMARY KEY ( master_server_id, maxscale_id ) ) "
"ENGINE=MYISAM DEFAULT CHARSET=latin1")) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error creating maxscale_schema.replication_heartbeat table in Master server"
": %s", mysql_error(database->con))));
database->server->rlag = -1;
}
/* auto purge old values after 48 hours*/
purge_time = time(0) - (3600 * 48);
sprintf(heartbeat_purge_query, "DELETE FROM maxscale_schema.replication_heartbeat WHERE master_timestamp < %lu", purge_time);
if (mysql_query(database->con, heartbeat_purge_query)) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error deleting from maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_purge_query,
mysql_error(database->con))));
}
heartbeat = time(0);
/* set node_ts for master as time(0) */
database->server->node_ts = heartbeat;
sprintf(heartbeat_insert_query, "UPDATE maxscale_schema.replication_heartbeat SET master_timestamp = %lu WHERE master_server_id = %li AND maxscale_id = %lu", heartbeat, handle->master->server->node_id, id);
/* Try to insert MaxScale timestamp into master */
if (mysql_query(database->con, heartbeat_insert_query)) {
database->server->rlag = -1;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error updating maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_insert_query,
mysql_error(database->con))));
} else {
if (mysql_affected_rows(database->con) == 0) {
heartbeat = time(0);
sprintf(heartbeat_insert_query, "REPLACE INTO maxscale_schema.replication_heartbeat (master_server_id, maxscale_id, master_timestamp ) VALUES ( %li, %lu, %lu)", handle->master->server->node_id, id, heartbeat);
if (mysql_query(database->con, heartbeat_insert_query)) {
database->server->rlag = -1;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: Error inserting into maxscale_schema.replication_heartbeat table: [%s], %s",
heartbeat_insert_query,
mysql_error(database->con))));
} else {
/* Set replication lag to 0 for the master */
database->server->rlag = 0;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: heartbeat table inserted data for %s:%i", database->server->name, database->server->port)));
}
} else {
/* Set replication lag as 0 for the master */
database->server->rlag = 0;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: heartbeat table updated for Master %s:%i", database->server->name, database->server->port)));
}
}
}
/*******
* This function gets the replication heartbeat
* from the maxscale_schema.replication_heartbeat table in the current slave
* and stores the timestamp and replication lag in the slave server struct
*
* @param handle The monitor handle
* @param database The number database server
*/
static void set_slave_heartbeat(MYSQL_MONITOR *handle, MONITOR_SERVERS *database) {
unsigned long id = handle->id;
time_t heartbeat;
char select_heartbeat_query[256] = "";
MYSQL_ROW row;
MYSQL_RES *result;
int num_fields;
/* Get the master_timestamp value from maxscale_schema.replication_heartbeat table */
sprintf(select_heartbeat_query, "SELECT master_timestamp "
"FROM maxscale_schema.replication_heartbeat "
"WHERE maxscale_id = %lu AND master_server_id = %li",
id, handle->master->server->node_id);
/* if there is a master then send the query to the slave with master_id */
if (handle->master !=NULL && (mysql_query(database->con, select_heartbeat_query) == 0
&& (result = mysql_store_result(database->con)) != NULL)) {
num_fields = mysql_num_fields(result);
while ((row = mysql_fetch_row(result))) {
int rlag = -1;
time_t slave_read;
heartbeat = time(0);
slave_read = strtoul(row[0], NULL, 10);
if ((errno == ERANGE && (slave_read == LONG_MAX || slave_read == LONG_MIN)) || (errno != 0 && slave_read == 0)) {
slave_read = 0;
}
if (slave_read) {
/* set the replication lag */
rlag = heartbeat - slave_read;
}
/* set this node_ts as master_timestamp read from replication_heartbeat table */
database->server->node_ts = slave_read;
if (rlag >= 0) {
/* store rlag only if greater than monitor sampling interval */
database->server->rlag = (rlag > (handle->interval / 1000)) ? rlag : 0;
} else {
database->server->rlag = -1;
}
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"[mysql_mon]: replication heartbeat: "
"Slave %s:%i has %i seconds lag",
database->server->name,
database->server->port,
database->server->rlag)));
}
mysql_free_result(result);
} else {
database->server->rlag = -1;
database->server->node_ts = 0;
if (handle->master->server->node_id < 0) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: error: replication heartbeat: "
"master_server_id NOT available for %s:%i",
database->server->name,
database->server->port)));
} else {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"[mysql_mon]: error: replication heartbeat: "
"failed selecting from hearthbeat table of %s:%i : [%s], %s",
database->server->name,
database->server->port,
select_heartbeat_query,
mysql_error(database->con))));
}
}
}
/*******
* This function computes the replication tree
* from a set of MySQL Master/Slave monitored servers
* and returns the root server with SERVER_MASTER bit.
* The tree is computed even for servers in 'maintenance' mode.
*
* @param handle The monitor handle
* @param num_servers The number of servers monitored
* @return The server at root level with SERVER_MASTER bit
*/
static MONITOR_SERVERS *get_replication_tree(MYSQL_MONITOR *handle, int num_servers) {
MONITOR_SERVERS *ptr;
MONITOR_SERVERS *backend;
SERVER *current;
int depth=0;
long node_id;
int root_level;
ptr = handle->databases;
root_level = num_servers;
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;
}
depth = 0;
current = ptr->server;
node_id = current->master_id;
if (node_id < 1) {
MONITOR_SERVERS *find_slave;
find_slave = getSlaveOfNodeId(handle->databases, current->node_id);
if (find_slave == NULL) {
current->depth = -1;
ptr = ptr->next;
continue;
} else {
current->depth = 0;
}
} else {
depth++;
}
while(depth <= num_servers) {
/* set the root master at lowest depth level */
if (current->depth > -1 && current->depth < root_level) {
root_level = current->depth;
handle->master = ptr;
}
backend = getServerByNodeId(handle->databases, node_id);
if (backend) {
node_id = backend->server->master_id;
} else {
node_id = -1;
}
if (node_id > 0) {
current->depth = depth + 1;
depth++;
} else {
MONITOR_SERVERS *master;
current->depth = depth;
master = getServerByNodeId(handle->databases, current->master_id);
if (master && master->server && master->server->node_id > 0) {
add_slave_to_master(master->server->slaves, MONITOR_MAX_NUM_SLAVES, current->node_id);
master->server->depth = current->depth -1;
monitor_set_pending_status(master, SERVER_MASTER);
} else {
if (current->master_id > 0) {
monitor_set_pending_status(ptr, SERVER_SLAVE_OF_EXTERNAL_MASTER);
}
}
break;
}
}
ptr = ptr->next;
}
/* If root master is in MAINT, return NULL */
if (SERVER_IN_MAINT(handle->master->server)) {
return NULL;
} else {
return handle->master;
}
}
/*******
* This function add a slave id into the slaves server field
* of its master server
*
* @param slaves_list The slave list array of the master server
* @param list_size The size of the slave list
* @param node_id The node_id of the slave to be inserted
* @return 1 for inserted value and 0 otherwise
*/
static int add_slave_to_master(long *slaves_list, int list_size, long node_id) {
int i;
for (i = 0; i< list_size; i++) {
if (slaves_list[i] == 0) {
memcpy(&slaves_list[i], &node_id, sizeof(long));
return 1;
}
}
return 0;
}
/**
* 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;
}

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

@ -31,6 +31,7 @@
* 08/07/13 Mark Riddoch Initial implementation
* 26/05/14 Massimiliano Pinto Default values for MONITOR_INTERVAL
* 28/05/14 Massimiliano Pinto Addition of new fields in MYSQL_MONITOR struct
* 24/06/14 Massimiliano Pinto Addition of master field in MYSQL_MONITOR struct and MONITOR_MAX_NUM_SLAVES
*
* @endverbatim
*/
@ -44,6 +45,7 @@ typedef struct monitor_servers {
MYSQL *con; /**< The MySQL connection */
int mon_err_count;
unsigned int mon_prev_status;
unsigned int pending_status; /**< Pending Status flag bitmap */
struct monitor_servers
*next; /**< The next server in the list */
} MONITOR_SERVERS;
@ -52,17 +54,17 @@ typedef struct monitor_servers {
* The handle for an instance of a MySQL Monitor module
*/
typedef struct {
SPINLOCK lock; /**< The monitor spinlock */
pthread_t tid; /**< id of monitor thread */
int shutdown; /**< Flag to shutdown the monitor thread */
int status; /**< Monitor status */
char *defaultUser; /**< Default username for monitoring */
char *defaultPasswd; /**< Default password for monitoring */
unsigned long interval; /**< Monitor sampling interval */
unsigned long id; /**< Monitor ID */
SPINLOCK lock; /**< The monitor spinlock */
pthread_t tid; /**< id of monitor thread */
int shutdown; /**< Flag to shutdown the monitor thread */
int status; /**< Monitor status */
char *defaultUser; /**< Default username for monitoring */
char *defaultPasswd; /**< Default password for monitoring */
unsigned long interval; /**< Monitor sampling interval */
unsigned long id; /**< Monitor ID */
int replicationHeartbeat; /**< Monitor flag for MySQL replication heartbeat */
int master_id; /**< Master server-id for MySQL Master/Slave replication */
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
MONITOR_SERVERS *master; /**< Master server for MySQL Master/Slave replication */
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
} MYSQL_MONITOR;
#define MONITOR_RUNNING 1
@ -71,5 +73,6 @@ typedef struct {
#define MONITOR_INTERVAL 10000 // in milliseconds
#define MONITOR_DEFAULT_ID 1UL // unsigned long value
#define MONITOR_MAX_NUM_SLAVES 20 //number of MySQL slave servers associated to a MySQL master server
#endif

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

@ -128,6 +128,10 @@ struct subcommand showoptions[] = {
"Show all currently loaded modules",
"Show all currently loaded modules",
{0, 0, 0} },
{ "monitor", 1, monitorShow,
"Show the monitor details",
"Show the monitor details",
{ARG_TYPE_MONITOR, 0, 0} },
{ "monitors", 0, monitorShowAll,
"Show the monitors that are configured",
"Show the monitors that are configured",
@ -168,6 +172,10 @@ struct subcommand showoptions[] = {
* The subcommands of the list command
*/
struct subcommand listoptions[] = {
{ "clients", 0, dListClients,
"List all the client connections to MaxScale",
"List all the client connections to MaxScale",
{0, 0, 0} },
{ "dcbs", 0, dListDCBs,
"List all the DCBs active within MaxScale",
"List all the DCBs active within MaxScale",
@ -181,8 +189,12 @@ struct subcommand listoptions[] = {
"List all the listeners defined within MaxScale",
{0, 0, 0} },
{ "modules", 0, dprintAllModules,
"Show all currently loaded modules",
"Show all currently loaded modules",
"List all currently loaded modules",
"List all currently loaded modules",
{0, 0, 0} },
{ "monitors", 0, monitorList,
"List all monitors",
"List all monitors",
{0, 0, 0} },
{ "services", 0, dListServices,
"List all the services defined within MaxScale",
@ -300,18 +312,30 @@ struct subcommand reloadoptions[] = {
{ "dbusers", 1, reload_dbusers,
"Reload the dbuser data for a service. E.g. reload dbusers \"splitter service\"",
"Reload the dbuser data for a service. E.g. reload dbusers 0x849420",
{ARG_TYPE_DBUSERS, 0, 0} },
{ARG_TYPE_SERVICE, 0, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} }
};
static void enable_log_action(DCB *, char *);
static void disable_log_action(DCB *, char *);
static void enable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor);
static void disable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor);
static void enable_service_root(DCB *dcb, SERVICE *service);
static void disable_service_root(DCB *dcb, SERVICE *service);
/**
* * The subcommands of the enable command
* */
struct subcommand enableoptions[] = {
{
"heartbeat",
1,
enable_monitor_replication_heartbeat,
"Enable the monitor replication heartbeat, pass a monitor name as argument",
"Enable the monitor replication heartbeat, pass a monitor name as argument",
{ARG_TYPE_MONITOR, 0, 0}
},
{
"log",
1,
@ -322,6 +346,14 @@ struct subcommand enableoptions[] = {
"message E.g. enable log message.",
{ARG_TYPE_STRING, 0, 0}
},
{
"root",
1,
enable_service_root,
"Enable root access to a service, pass a service name to enable root access",
"Enable root access to a service, pass a service name to enable root access",
{ARG_TYPE_SERVICE, 0, 0}
},
{
NULL,
0,
@ -337,24 +369,40 @@ struct subcommand enableoptions[] = {
* * The subcommands of the disable command
* */
struct subcommand disableoptions[] = {
{
"log",
1,
disable_log_action,
"Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug",
"Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug",
{ARG_TYPE_STRING, 0, 0}
},
{
{
"heartbeat",
1,
disable_monitor_replication_heartbeat,
"Disable the monitor replication heartbeat",
"Disable the monitor replication heartbeat",
{ARG_TYPE_MONITOR, 0, 0}
},
{
"log",
1,
disable_log_action,
"Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug",
"Disable Log for MaxScale, Options: debug | trace | error | message "
"E.g. disable log debug",
{ARG_TYPE_STRING, 0, 0}
},
{
"root",
1,
disable_service_root,
"Disable root access to a service",
"Disable root access to a service",
{ARG_TYPE_SERVICE, 0, 0}
},
{
NULL,
0,
NULL,
NULL,
NULL,
{0, 0, 0}
}
}
};
#if defined(SS_DEBUG)
@ -850,7 +898,7 @@ unsigned int bitvalue;
static void
reload_dbusers(DCB *dcb, SERVICE *service)
{
dcb_printf(dcb, "Loaded %d database users for server %s.\n",
dcb_printf(dcb, "Loaded %d database users for service %s.\n",
reload_mysql_users(service), service->name);
}
@ -958,6 +1006,55 @@ restart_monitor(DCB *dcb, MONITOR *monitor)
monitorStart(monitor);
}
/**
* Enable replication heartbeat for a monitor
*
* @param dcb Connection to user interface
* @param monitor The monitor
*/
static void
enable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor)
{
monitorSetReplicationHeartbeat(monitor, 1);
}
/**
* Disable replication heartbeat for a monitor
*
* @param dcb Connection to user interface
* @param monitor The monitor
*/
static void
disable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor)
{
monitorSetReplicationHeartbeat(monitor, 0);
}
/**
* Enable root access to a service
*
* @param dcb Connection to user interface
* @param service The service
*/
static void
enable_service_root(DCB *dcb, SERVICE *service)
{
serviceEnableRootUser(service, 1);
}
/**
* Disable root access to a service
*
* @param dcb Connection to user interface
* @param service The service
*/
static void
disable_service_root(DCB *dcb, SERVICE *service)
{
serviceEnableRootUser(service, 0);
}
/**
* The log enable action
*/

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

@ -65,6 +65,8 @@
* or take different actions such as open a new backend connection
* 20/02/2014 Massimiliano Pinto If router_options=slave, route traffic to master if no slaves available
* 06/03/2014 Massimiliano Pinto Server connection counter is now updated in closeSession
* 24/06/2014 Massimiliano Pinto New rules for selecting the Master server
* 27/06/2014 Mark Riddoch Addition of server weighting
*
* @endverbatim
*/
@ -96,7 +98,7 @@ MODULE_INFO info = {
"A connection based router to load balance based on connections"
};
static char *version_str = "V1.0.2";
static char *version_str = "V1.1.0";
/* The router entry points */
static ROUTER *createInstance(SERVICE *service, char **options);
@ -110,13 +112,13 @@ static void clientReply(
void *router_session,
GWBUF *queue,
DCB *backend_dcb);
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
int action,
bool *succp);
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp);
static uint8_t getCapabilities (ROUTER* inst, void* router_session);
@ -139,6 +141,9 @@ static bool rses_begin_locked_router_action(
static void rses_end_locked_router_action(
ROUTER_CLIENT_SES* rses);
static BACKEND *get_root_master(
BACKEND **servers);
static SPINLOCK instlock;
static ROUTER_INSTANCE *instances;
@ -196,6 +201,8 @@ createInstance(SERVICE *service, char **options)
ROUTER_INSTANCE *inst;
SERVER *server;
int i, n;
BACKEND *backend;
char *weightby;
if ((inst = calloc(1, sizeof(ROUTER_INSTANCE))) == NULL) {
return NULL;
@ -231,10 +238,53 @@ int i, n;
}
inst->servers[n]->server = server;
inst->servers[n]->current_connection_count = 0;
inst->servers[n]->weight = 100;
n++;
}
inst->servers[n] = NULL;
if ((weightby = serviceGetWeightingParameter(service)) != NULL)
{
int total = 0;
for (n = 0; inst->servers[n]; n++)
{
backend = inst->servers[n];
total += atoi(serverGetParameter(backend->server,
weightby));
}
if (total == 0)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"WARNING: Weighting Parameter for service '%s' "
"will be ignored as no servers have values "
"for the parameter '%s'.\n",
service->name, weightby)));
}
else
{
for (n = 0; inst->servers[n]; n++)
{
int perc;
backend = inst->servers[n];
perc = (atoi(serverGetParameter(backend->server,
weightby)) * 100) / total;
backend->weight = perc;
if (perc == 0)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Server '%s' has no value "
"for weighting parameter '%s', "
"no queries will be routed to "
"this server.\n",
server->unique_name,
weightby)));
}
}
}
}
/*
* Process the options
*/
@ -262,11 +312,11 @@ int i, n;
else
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"* Warning : Unsupported router "
"option \'%s\' for readconnroute. "
"Expected router options are "
"[slave|master|synced]",
LOGFILE_MESSAGE,
"* Warning : Unsupported router "
"option \'%s\' for readconnroute. "
"Expected router options are "
"[slave|master|synced]",
options[i])));
}
}
@ -299,7 +349,7 @@ ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *)instance;
ROUTER_CLIENT_SES *client_rses;
BACKEND *candidate = NULL;
int i;
int master_host = -1;
BACKEND *master_host = NULL;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
@ -321,6 +371,11 @@ int master_host = -1;
client_rses->rses_chk_tail = CHK_NUM_ROUTER_SES;
#endif
/**
* Find the Master host from available servers
*/
master_host = get_root_master(inst->servers);
/**
* Find a backend server to connect to. This is the extent of the
* load balancing algorithm we need to implement for this simple
@ -356,36 +411,60 @@ int master_host = -1;
if (SERVER_IN_MAINT(inst->servers[i]->server))
continue;
/*
* If router_options=slave, get the running master
* It will be used if there are no running slaves at all
*/
if (inst->bitvalue == SERVER_SLAVE) {
if (master_host < 0 && (SERVER_IS_MASTER(inst->servers[i]->server))) {
master_host = i;
}
}
/* Check server status bits against bitvalue from router_options */
if (inst->servers[i] &&
SERVER_IS_RUNNING(inst->servers[i]->server) &&
(inst->servers[i]->server->status & inst->bitmask) ==
inst->bitvalue)
SERVER_IS_RUNNING(inst->servers[i]->server) &&
(inst->servers[i]->server->status & inst->bitmask & inst->bitvalue))
{
if (master_host) {
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_SLAVE)) {
/* skip root Master here, as it could also be slave of an external server
* that is not in the configuration.
* Intermediate masters (Relay Servers) are also slave and will be selected
* as Slave(s)
*/
continue;
}
if (inst->servers[i] == master_host && (inst->bitvalue & SERVER_MASTER)) {
/* If option is "master" return only the root Master as there
* could be intermediate masters (Relay Servers)
* and they must not be selected.
*/
candidate = master_host;
break;
}
} else {
/* master_host is NULL, no master server.
* If requested router_option is 'master'
* candidate wll be NULL.
*/
if (inst->bitvalue & SERVER_MASTER) {
candidate = NULL;
break;
}
}
/* If no candidate set, set first running server as
our initial candidate server */
if (candidate == NULL)
{
candidate = inst->servers[i];
}
else if (inst->servers[i]->current_connection_count <
candidate->current_connection_count)
else if ((inst->servers[i]->current_connection_count
* 100) / inst->servers[i]->weight <
(candidate->current_connection_count *
100) / candidate->weight)
{
/* This running server has fewer
connections, set it as a new candidate */
candidate = inst->servers[i];
}
else if (inst->servers[i]->current_connection_count ==
candidate->current_connection_count &&
else if ((inst->servers[i]->current_connection_count
* 100) / inst->servers[i]->weight ==
(candidate->current_connection_count *
100) / candidate->weight &&
inst->servers[i]->server->stats.n_connections <
candidate->server->stats.n_connections)
{
@ -403,8 +482,8 @@ int master_host = -1;
* Otherwise, just clean up and return NULL
*/
if (!candidate) {
if (master_host >= 0) {
candidate = inst->servers[master_host];
if (master_host) {
candidate = master_host;
} else {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -649,6 +728,8 @@ diagnostics(ROUTER *router, DCB *dcb)
ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *)router;
ROUTER_CLIENT_SES *session;
int i = 0;
BACKEND *backend;
char *weightby;
spinlock_acquire(&router_inst->lock);
session = router_inst->connections;
@ -664,6 +745,24 @@ int i = 0;
dcb_printf(dcb, "\tCurrent no. of router sessions: %d\n", i);
dcb_printf(dcb, "\tNumber of queries forwarded: %d\n",
router_inst->stats.n_queries);
if ((weightby = serviceGetWeightingParameter(router_inst->service))
!= NULL)
{
dcb_printf(dcb, "\tConnection distribution based on %s "
"server parameter.\n",
weightby);
dcb_printf(dcb,
"\t\tServer Target %% Connections\n");
for (i = 0; router_inst->servers[i]; i++)
{
backend = router_inst->servers[i];
dcb_printf(dcb, "\t\t%-20s %3d%% %d\n",
backend->server->unique_name,
backend->weight,
backend->current_connection_count);
}
}
}
/**
@ -706,12 +805,12 @@ clientReply(
*/
static void
handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
int action,
bool *succp)
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp)
{
DCB *client = NULL;
SESSION *session = backend_dcb->session;
@ -785,3 +884,56 @@ static uint8_t getCapabilities(
{
return 0;
}
/********************************
* This routine return the root master server from MySQL replication tree
* Get the root Master rule:
*
* (1) find server(s) with lowest replication depth level
* (2) check for SERVER_MASTER bitvalue in those servers
* Servers are checked even if they are in 'maintenance'
* SERVER_IS_DOWN is the only status to skip.
*
* @param servers The list of servers
* @return The Master found
*
*/
static BACKEND *get_root_master(BACKEND **servers) {
int i = 0;
BACKEND * master_host = NULL;
/* (1) find root server(s) with lowest replication depth level */
for (i = 0; servers[i]; i++) {
if (servers[i] && (! SERVER_IS_DOWN(servers[i]->server))) {
if (master_host && servers[i]->server->depth < master_host->server->depth) {
master_host = servers[i];
} else {
if (master_host == NULL) {
master_host = servers[i];
}
}
}
}
/* (2) get the status of server(s) with lowest replication level and check it against SERVER_MASTER bitvalue */
if (master_host) {
int found = 0;
for (i = 0; servers[i]; i++) {
if (servers[i] && (! SERVER_IS_DOWN(servers[i]->server)) && (servers[i]->server->depth == master_host->server->depth)) {
if (servers[i]->server->status & SERVER_MASTER) {
master_host = servers[i];
found = 1;
}
}
}
if (!found)
master_host = NULL;
/* return NULL if the server is SERVER_IN_MAINT */
if (found && SERVER_IN_MAINT(master_host->server))
master_host = NULL;
}
return master_host;
}

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

@ -247,6 +247,10 @@ static bool handle_error_new_connection(
GWBUF* errmsg);
static bool handle_error_reply_client(SESSION* ses, GWBUF* errmsg);
static BACKEND *get_root_master(
backend_ref_t *servers,
int router_nservers);
static SPINLOCK instlock;
static ROUTER_INSTANCE* instances;
@ -621,7 +625,8 @@ static void* newSession(
}
/** Copy backend pointers to router session. */
client_rses->rses_master_ref = master_ref;
ss_dassert(SERVER_IS_MASTER(master_ref->bref_backend->backend_server));
/* assert with master_host */
ss_dassert(master_ref && (master_ref->bref_backend->backend_server && SERVER_MASTER));
client_rses->rses_backend_ref = backend_ref;
client_rses->rses_nbackends = router_nservers; /*< # of backend servers */
client_rses->rses_capabilities = RCAP_TYPE_STMT_INPUT;
@ -802,6 +807,7 @@ static bool get_dcb(
int smallest_nconn = -1;
int i;
bool succp = false;
BACKEND *master_host = NULL;
CHK_CLIENT_RSES(rses);
ss_dassert(p_dcb != NULL && *(p_dcb) == NULL);
@ -812,13 +818,18 @@ static bool get_dcb(
}
backend_ref = rses->rses_backend_ref;
/* get root master from availbal servers */
master_host = get_root_master(backend_ref, rses->rses_nbackends);
if (btype == BE_SLAVE)
{
for (i=0; i<rses->rses_nbackends; i++)
{
BACKEND* b = backend_ref[i].bref_backend;
/* check slave bit, also for relay servers (Master & Servers) */
if (BREF_IS_IN_USE((&backend_ref[i])) &&
SERVER_IS_SLAVE(b->backend_server) &&
(SERVER_IS_SLAVE(b->backend_server) || SERVER_IS_RELAY_SERVER(b->backend_server)) &&
(master_host != NULL && b->backend_server != master_host->backend_server) &&
(smallest_nconn == -1 ||
b->backend_conn_count < smallest_nconn))
{
@ -837,11 +848,12 @@ static bool get_dcb(
{
*p_dcb = backend_ref->bref_dcb;
succp = true;
ss_dassert(backend_ref->bref_dcb->state != DCB_STATE_ZOMBIE);
ss_dassert(
SERVER_IS_MASTER(backend_ref->bref_backend->backend_server) &&
smallest_nconn == -1);
(master_host && (backend_ref->bref_backend->backend_server == master_host->backend_server)) &&
smallest_nconn == -1);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -859,9 +871,9 @@ static bool get_dcb(
for (i=0; i<rses->rses_nbackends; i++)
{
BACKEND* b = backend_ref[i].bref_backend;
if (BREF_IS_IN_USE((&backend_ref[i])) &&
(SERVER_IS_MASTER(b->backend_server)))
(master_host && (b->backend_server == master_host->backend_server)))
{
*p_dcb = backend_ref[i].bref_dcb;
succp = true;
@ -1641,6 +1653,7 @@ static bool select_connect_backend_servers(
const int min_nslaves = 0; /*< not configurable at the time */
bool is_synced_master;
int (*p)(const void *, const void *);
BACKEND *master_host = NULL;
if (p_master_ref == NULL || backend_ref == NULL)
{
@ -1648,7 +1661,10 @@ static bool select_connect_backend_servers(
succp = false;
goto return_succp;
}
/* get the root Master */
master_host = get_root_master(backend_ref, router_nservers);
/** Master is already chosen and connected. This is slave failure case */
if (*p_master_ref != NULL &&
BREF_IS_IN_USE((*p_master_ref)))
@ -1662,7 +1678,8 @@ static bool select_connect_backend_servers(
master_found = true;
master_connected = true;
ss_dassert(SERVER_IS_MASTER((*p_master_ref)->bref_backend->backend_server));
/* assert with master_host */
ss_dassert(master_host && ((*p_master_ref)->bref_backend->backend_server == master_host->backend_server) && SERVER_MASTER);
}
/** New session or master failure case */
else
@ -1713,8 +1730,9 @@ static bool select_connect_backend_servers(
b->backend_conn_count)));
}
#endif
/* assert with master_host */
ss_dassert(!master_connected ||
SERVER_IS_MASTER((*p_master_ref)->bref_backend->backend_server));
(master_host && ((*p_master_ref)->bref_backend->backend_server == master_host->backend_server) && SERVER_MASTER));
/**
* Sort the pointer list to servers according to connection counts. As
* a consequence those backends having least connections are in the
@ -1784,13 +1802,15 @@ static bool select_connect_backend_servers(
b->backend_server->port,
STRSRVSTATUS(b->backend_server),
b->backend_server->stats.n_current_ops)));
if (SERVER_IS_RUNNING(b->backend_server) &&
((b->backend_server->status & router->bitmask) ==
router->bitvalue))
{
/* check also for relay servers and don't take the master_host */
if (slaves_found < max_nslaves &&
SERVER_IS_SLAVE(b->backend_server))
(SERVER_IS_SLAVE(b->backend_server) || SERVER_IS_RELAY_SERVER(b->backend_server)) &&
(master_host != NULL && (b->backend_server != master_host->backend_server)))
{
slaves_found += 1;
@ -1848,7 +1868,8 @@ static bool select_connect_backend_servers(
}
}
}
else if (SERVER_IS_MASTER(b->backend_server))
/* take the master_host for master */
else if (master_host && (b->backend_server == master_host->backend_server))
{
*p_master_ref = &backend_ref[i];
@ -1915,8 +1936,9 @@ static bool select_connect_backend_servers(
b->backend_server->port,
b->backend_conn_count)));
}
/* assert with master_host */
ss_dassert(!master_connected ||
SERVER_IS_MASTER((*p_master_ref)->bref_backend->backend_server));
(master_host && ((*p_master_ref)->bref_backend->backend_server == master_host->backend_server) && SERVER_MASTER));
#endif
/**
@ -2485,7 +2507,7 @@ static bool execute_sescmd_in_backend(
goto return_succp;
}
if (!sescmd_cursor_is_active(scur))
{
/** Cursor is left active when function returns. */
@ -3060,9 +3082,7 @@ static bool handle_error_new_connection(
DCB* client_dcb;
client_dcb = ses->client;
client_dcb->func.write(client_dcb, errmsg);
#if 1
bref_clear_state(bref, BREF_WAITING_RESULT);
#endif
}
bref_clear_state(bref, BREF_IN_USE);
bref_set_state(bref, BREF_CLOSED);
@ -3257,7 +3277,6 @@ return_rc:
return rc;
}
static sescmd_cursor_t* backend_ref_get_sescmd_cursor (
backend_ref_t* bref)
{
@ -3324,3 +3343,60 @@ static bool prep_stmt_drop(
return true;
}
#endif /*< PREP_STMT_CACHING */
/********************************
* This routine return the root master server from MySQL replication tree
* Get the root Master rule:
*
* (1) find server(s) with lowest replication depth level
* (2) check for SERVER_MASTER bitvalue in those servers
* Servers are checked even if they are in 'maintenance'
* SERVER_IS_DOWN is the only status to skip.
*
* @param servers The list of servers
* @param The number of servers
* @return The Master found
*
*/
static BACKEND *get_root_master(backend_ref_t *servers, int router_nservers) {
int i = 0;
BACKEND * master_host = NULL;
/* (1) find root server(s) with lowest replication depth level */
for (i = 0; i< router_nservers; i++) {
BACKEND* b = NULL;
b = servers[i].bref_backend;
if (b && (! SERVER_IS_DOWN(b->backend_server))) {
if (master_host && b->backend_server->depth < master_host->backend_server->depth) {
master_host = b;
} else {
if (master_host == NULL) {
master_host = b;
}
}
}
}
/* (2) get the status of server(s) with lowest replication level and check it against SERVER_MASTER bitvalue */
if (master_host) {
int found = 0;
for (i = 0; i<router_nservers; i++) {
BACKEND* b = NULL;
b = servers[i].bref_backend;
if (b && (! SERVER_IS_DOWN(b->backend_server)) && (b->backend_server->depth == master_host->backend_server->depth)) {
if (b->backend_server->status & SERVER_MASTER) {
master_host = b;
found = 1;
}
}
}
if (!found)
master_host = NULL;
if (found && SERVER_IN_MAINT(master_host->backend_server))
master_host = NULL;
}
return master_host;
}