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

Merge branch 'develop' into plainrouter

Browse Source
This commit is contained in:
Markus Makela
2015-03-24 17:51:41 +02:00
parent 06b59ffe2e 6b8ffe7506
commit 1bd3401739
150 changed files with 18236 additions and 1038 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -5,6 +5,14 @@ if(BUILD_TESTS)
install(TARGETS testroute DESTINATION modules)
endif()
add_library(schemarouter SHARED schemarouter/schemarouter.c)
target_link_libraries(schemarouter log_manager utils query_classifier)
install(TARGETS schemarouter DESTINATION modules)
add_library(shardrouter SHARED schemarouter/shardrouter.c)
target_link_libraries(shardrouter log_manager utils query_classifier)
install(TARGETS shardrouter DESTINATION modules)
add_library(readconnroute SHARED readconnroute.c)
target_link_libraries(readconnroute log_manager utils)
install(TARGETS readconnroute DESTINATION modules)
@ -22,7 +30,9 @@ target_link_libraries(cli log_manager utils)
install(TARGETS cli DESTINATION modules)
add_subdirectory(readwritesplit)
add_subdirectory(schemarouter/test)
if(BUILD_BINLOG)
add_subdirectory(binlog)
endif()
add_subdirectory(maxinfo)

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

@ -432,7 +432,7 @@ unsigned char *defuuid;
* Now start the replication from the master to MaxScale
*/
blr_start_master(inst);
free(name);
return (ROUTER *)inst;
}

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

@ -71,7 +71,7 @@ static void blr_log_header(logfile_id_t file, char *msg, uint8_t *ptr);
int
blr_file_init(ROUTER_INSTANCE *router)
{
char *ptr, path[1024], filename[1050];
char *ptr, path[PATH_MAX], filename[PATH_MAX];
int file_found, n = 1;
int root_len, i;
DIR *dirp;
@ -79,13 +79,13 @@ struct dirent *dp;
if (router->binlogdir == NULL)
{
strcpy(path, "/usr/local/skysql/MaxScale");
strcpy(path, "/usr/local/mariadb-maxscale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
strncpy(path, ptr,PATH_MAX);
}
strcat(path, "/");
strcat(path, router->service->name);
strncat(path, "/",PATH_MAX);
strncat(path, router->service->name,PATH_MAX);
if (access(path, R_OK) == -1)
mkdir(path, 0777);
@ -94,7 +94,7 @@ struct dirent *dp;
}
else
{
strncpy(path, router->binlogdir, 1024);
strncpy(path, router->binlogdir, PATH_MAX);
}
if (access(router->binlogdir, R_OK) == -1)
{
@ -128,7 +128,7 @@ struct dirent *dp;
file_found = 0;
do {
sprintf(filename, "%s/" BINLOG_NAMEFMT, path, router->fileroot, n);
snprintf(filename,PATH_MAX, "%s/" BINLOG_NAMEFMT, path, router->fileroot, n);
if (access(filename, R_OK) != -1)
{
file_found = 1;
@ -142,16 +142,16 @@ struct dirent *dp;
if (n == 0) // No binlog files found
{
if (router->initbinlog)
sprintf(filename, BINLOG_NAMEFMT, router->fileroot,
snprintf(filename,PATH_MAX, BINLOG_NAMEFMT, router->fileroot,
router->initbinlog);
else
sprintf(filename, BINLOG_NAMEFMT, router->fileroot, 1);
snprintf(filename,PATH_MAX, BINLOG_NAMEFMT, router->fileroot, 1);
if (! blr_file_create(router, filename))
return 0;
}
else
{
sprintf(filename, BINLOG_NAMEFMT, router->fileroot, n);
snprintf(filename,PATH_MAX, BINLOG_NAMEFMT, router->fileroot, n);
blr_file_append(router, filename);
}
return 1;
@ -196,7 +196,7 @@ unsigned char magic[] = BINLOG_MAGIC;
fsync(fd);
close(router->binlog_fd);
spinlock_acquire(&router->binlog_lock);
strcpy(router->binlog_name, file);
strncpy(router->binlog_name, file,BINLOG_FNAMELEN);
router->binlog_position = 4; /* Initial position after the magic number */
spinlock_release(&router->binlog_lock);
router->binlog_fd = fd;
@ -230,7 +230,7 @@ int fd;
fsync(fd);
close(router->binlog_fd);
spinlock_acquire(&router->binlog_lock);
strcpy(router->binlog_name, file);
strncpy(router->binlog_name, file,BINLOG_FNAMELEN);
router->binlog_position = lseek(fd, 0L, SEEK_END);
spinlock_release(&router->binlog_lock);
router->binlog_fd = fd;
@ -290,7 +290,7 @@ blr_file_flush(ROUTER_INSTANCE *router)
BLFILE *
blr_open_binlog(ROUTER_INSTANCE *router, char *binlog)
{
char path[1024];
char path[1025];
BLFILE *file;
spinlock_acquire(&router->fileslock);
@ -310,14 +310,14 @@ BLFILE *file;
spinlock_release(&router->fileslock);
return NULL;
}
strcpy(file->binlogname, binlog);
strncpy(file->binlogname, binlog,BINLOG_FNAMELEN+1);
file->refcnt = 1;
file->cache = 0;
spinlock_init(&file->lock);
strcpy(path, router->binlogdir);
strcat(path, "/");
strcat(path, binlog);
strncpy(path, router->binlogdir,1024);
strncat(path, "/",1024);
strncat(path, binlog,1024);
if ((file->fd = open(path, O_RDONLY, 0666)) == -1)
{
@ -630,10 +630,10 @@ struct stat statb;
void
blr_cache_response(ROUTER_INSTANCE *router, char *response, GWBUF *buf)
{
char path[4096], *ptr;
char path[4097], *ptr;
int fd;
strcpy(path, "/usr/local/skysql/MaxScale");
strcpy(path, "/usr/local/mariadb-maxscale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strncpy(path, ptr, 4096);
@ -668,11 +668,11 @@ GWBUF *
blr_cache_read_response(ROUTER_INSTANCE *router, char *response)
{
struct stat statb;
char path[4096], *ptr;
char path[4097], *ptr;
int fd;
GWBUF *buf;
strcpy(path, "/usr/local/skysql/MaxScale");
strcpy(path, "/usr/local/mariadb-maxscale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strncpy(path, ptr, 4096);

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

@ -142,6 +142,7 @@ GWBUF *buf;
sprintf(name, "%s Master", router->service->name);
hktask_oneshot(name, blr_start_master, router,
BLR_MASTER_BACKOFF_TIME * router->retry_backoff++);
free(name);
}
if (router->retry_backoff > BLR_MAX_BACKOFF)
router->retry_backoff = BLR_MAX_BACKOFF;
@ -203,11 +204,12 @@ GWBUF *ptr;
router->master_state = BLRM_UNCONNECTED;
if ((name = malloc(strlen(router->service->name)
+ strlen(" Master")+1)) != NULL);
+ strlen(" Master")+1)) != NULL)
{
sprintf(name, "%s Master", router->service->name);
hktask_oneshot(name, blr_start_master, router,
BLR_MASTER_BACKOFF_TIME * router->retry_backoff++);
free(name);
}
if (router->retry_backoff > BLR_MAX_BACKOFF)
router->retry_backoff = BLR_MAX_BACKOFF;
@ -283,10 +285,11 @@ blr_master_delayed_connect(ROUTER_INSTANCE *router)
char *name;
if ((name = malloc(strlen(router->service->name)
+ strlen(" Master Recovery")+1)) != NULL);
+ strlen(" Master Recovery")+1)) != NULL)
{
sprintf(name, "%s Master Recovery", router->service->name);
hktask_oneshot(name, blr_start_master, router, 60);
free(name);
}
}
@ -407,6 +410,7 @@ char query[128];
}
router->master_state = BLRM_HBPERIOD;
router->master->func.write(router->master, buf);
free(val);
break;
}
case BLRM_HBPERIOD:
@ -701,7 +705,7 @@ blr_handle_binlog_record(ROUTER_INSTANCE *router, GWBUF *pkt)
{
uint8_t *msg = NULL, *ptr, *pdata;
REP_HEADER hdr;
unsigned int len, reslen;
unsigned int len = 0, reslen;
unsigned int pkt_length;
int no_residual = 1;
int preslen = -1;
@ -1224,8 +1228,8 @@ MYSQL_session *auth_info;
if ((auth_info = calloc(1, sizeof(MYSQL_session))) == NULL)
return NULL;
strcpy(auth_info->user, username);
strcpy(auth_info->db, database);
strncpy(auth_info->user, username,MYSQL_USER_MAXLEN+1);
strncpy(auth_info->db, database,MYSQL_DATABASE_MAXLEN+1);
gw_sha1_str((const uint8_t *)password, strlen(password), auth_info->client_sha1);
return auth_info;

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

@ -34,6 +34,8 @@
* Date Who Description
* 14/04/2014 Mark Riddoch Initial implementation
* 18/02/2015 Massimiliano Pinto Addition of DISCONNECT ALL and DISCONNECT SERVER server_id
* 18/03/2015 Markus Makela Better detection of CRC32 | NONE checksum
* 19/03/2015 Massimiliano Pinto Addition of basic MariaDB 10 compatibility support
*
* @endverbatim
*/
@ -80,6 +82,7 @@ static int blr_slave_send_eof(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, int
static int blr_slave_send_disconnected_server(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, int server_id, int found);
static int blr_slave_disconnect_all(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave);
static int blr_slave_disconnect_server(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, int server_id);
static int blr_slave_send_ok(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave);
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
@ -363,13 +366,17 @@ int query_len;
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.heartbeat);
}
else if (strcasecmp(word, "@mariadb_slave_capability") == 0)
{
free(query_text);
return blr_slave_send_ok(router, slave);
}
else if (strcasecmp(word, "@master_binlog_checksum") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (word && (strcasecmp(word, "'none'") == 0))
slave->nocrc = 1;
else
slave->nocrc = 0;
if (word && (strcasecmp(word, "@@global.biglog_checksum'") == 0))
slave->nocrc = !router->master_chksum;
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.chksum1);
}
@ -410,8 +417,6 @@ int query_len;
else if (strcasecmp(word, "ALL") == 0)
{
free(query_text);
spinlock_release(&router->lock);
return blr_slave_disconnect_all(router, slave);
}
else if (strcasecmp(word, "SERVER") == 0)
@ -435,7 +440,7 @@ int query_len;
LOGFILE_ERROR, "Unexpected query from slave server %s", query_text)));
free(query_text);
blr_slave_send_error(router, slave, "Unexpected SQL query received from slave.");
return 0;
return 1;
}
@ -485,9 +490,9 @@ int len;
if ((pkt = gwbuf_alloc(strlen(msg) + 13)) == NULL)
return;
data = GWBUF_DATA(pkt);
len = strlen(msg) + 1;
len = strlen(msg) + 9;
encode_value(&data[0], len, 24); // Payload length
data[3] = 0; // Sequence id
data[3] = 1; // Sequence id
// Payload
data[4] = 0xff; // Error indicator
data[5] = 0; // Error Code
@ -1371,7 +1376,7 @@ blr_slave_catchup(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, bool large)
GWBUF *head, *record;
REP_HEADER hdr;
int written, rval = 1, burst;
int rotating;
int rotating = 0;
unsigned long burst_size;
uint8_t *ptr;
@ -1887,21 +1892,23 @@ char serverid[40];
uint8_t *ptr;
int len, id_len, seqno = 2;
blr_slave_send_fieldcount(router, slave, 2);
blr_slave_send_columndef(router, slave, "server_id", 0x03, 40, seqno++);
blr_slave_send_columndef(router, slave, "state", 0xf, 40, seqno++);
blr_slave_send_eof(router, slave, seqno++);
sprintf(serverid, "%d", server_id);
id_len = strlen(serverid);
if (found)
strcpy(state, "disconnected");
else
strcpy(state, "not found");
id_len = strlen(serverid);
len = 5 + id_len + strlen(state) + 1;
if ((pkt = gwbuf_alloc(len)) == NULL)
return 0;
blr_slave_send_fieldcount(router, slave, 2);
blr_slave_send_columndef(router, slave, "server_id", 0x03, 40, seqno++);
blr_slave_send_columndef(router, slave, "state", 0xf, 40, seqno++);
blr_slave_send_eof(router, slave, seqno++);
ptr = GWBUF_DATA(pkt);
encode_value(ptr, id_len + 2 + strlen(state), 24); // Add length of data packet
ptr += 3;
@ -1949,9 +1956,12 @@ blr_slave_disconnect_server(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, int se
{
/* server_id found */
server_found = 1;
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, "DISCONNECT SERVER: closing [%s], server id [%d]",
sptr->dcb->remote, server_id)));
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, "%s: Slave %s, server id %d, disconnected by %s@%s",
router->service->name,
sptr->dcb->remote,
server_id,
slave->dcb->user,
slave->dcb->remote)));
/* send server_id with disconnect state to client */
n = blr_slave_send_disconnected_server(router, slave, server_id, 1);
@ -1975,7 +1985,15 @@ blr_slave_disconnect_server(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, int se
n = blr_slave_send_disconnected_server(router, slave, server_id, 0);
}
return n;
if (n == 0) {
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR, "Error: gwbuf memory allocation in "
"DISCONNECT SERVER server_id [%d]",
sptr->serverid)));
blr_slave_send_error(router, slave, "Memory allocation error for DISCONNECT SERVER");
}
return 1;
}
/**
@ -1996,7 +2014,7 @@ blr_slave_disconnect_all(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
uint8_t *ptr;
int len, seqno;
GWBUF *pkt;
int n = 0;
int n = 1;
/* preparing output result */
blr_slave_send_fieldcount(router, slave, 2);
@ -2013,13 +2031,11 @@ blr_slave_disconnect_all(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
/* skip servers with state = 0 */
if (sptr->state != 0)
{
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, "DISCONNECT ALL: closing [%s], server_id [%d]",
sptr->dcb->remote, sptr->serverid)));
sprintf(server_id, "%d", sptr->serverid);
sprintf(state, "disconnected");
len = 5 + strlen(server_id) + strlen(state) + 1;
if ((pkt = gwbuf_alloc(len)) == NULL) {
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR, "Error: gwbuf memory allocation in "
"DISCONNECT ALL for [%s], server_id [%d]",
@ -2027,9 +2043,15 @@ blr_slave_disconnect_all(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
spinlock_release(&router->lock);
return 0;
blr_slave_send_error(router, slave, "Memory allocation error for DISCONNECT ALL");
return 1;
}
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, "%s: Slave %s, server id %d, disconnected by %s@%s",
router->service->name,
sptr->dcb->remote, sptr->serverid, slave->dcb->user, slave->dcb->remote)));
ptr = GWBUF_DATA(pkt);
encode_value(ptr, len - 4, 24); // Add length of data packet
@ -2055,5 +2077,33 @@ blr_slave_disconnect_all(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
blr_slave_send_eof(router, slave, seqno);
return n;
return 1;
}
/**
* Send a MySQL OK packet to the DCB
*
* @param dcb The DCB to send the OK packet to
* @return result of a write call, non-zero if write was successful
*/
static int
blr_slave_send_ok(ROUTER_INSTANCE* router, ROUTER_SLAVE* slave)
{
GWBUF *pkt;
uint8_t *ptr;
if ((pkt = gwbuf_alloc(11)) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
*ptr++ = 7; // Payload length
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 1; // Seqno
*ptr++ = 0; // ok
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 2;
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 0;
return slave->dcb->func.write(slave->dcb, pkt);
}

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

@ -36,12 +36,13 @@
* Date Who Description
* 20/06/13 Mark Riddoch Initial implementation
* 17/07/13 Mark Riddoch Additional commands
* 09/08/2013 Massimiliano Pinto Added enable/disable commands (now only for log)
* 09/08/13 Massimiliano Pinto Added enable/disable commands (now only for log)
* 20/05/14 Mark Riddoch Added ability to give server and service names rather
* than simply addresses
* 23/05/14 Mark Riddoch Added support for developer and user modes
* 29/05/14 Mark Riddoch Add Filter support
* 16/10/14 Mark Riddoch Add show eventq
* 05/03/15 Massimiliano Pinto Added enable/disable feedback
*
* @endverbatim
*/
@ -62,7 +63,7 @@
#include <poll.h>
#include <users.h>
#include <dbusers.h>
#include <config.h>
#include <maxconfig.h>
#include <telnetd.h>
#include <adminusers.h>
#include <monitor.h>
@ -129,6 +130,10 @@ struct subcommand showoptions[] = {
"Show the event statistics",
"Show the event statistics",
{0, 0, 0} },
{ "feedbackreport", 0, moduleShowFeedbackReport,
"Show the report of MaxScale loaded modules, suitable for Notification Service",
"Show the report of MaxScale loaded modules, suitable for Notification Service",
{0, 0, 0} },
{ "filter", 1, dprintFilter,
"Show details of a filter, called with a filter name",
"Show details of a filter, called with the address of a filter",
@ -365,6 +370,8 @@ 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);
static void enable_feedback_action();
static void disable_feedback_action();
/**
* * The subcommands of the enable command
@ -406,6 +413,14 @@ struct subcommand enableoptions[] = {
"Enable root access to a service, pass a service name to enable root access",
{ARG_TYPE_SERVICE, 0, 0}
},
{
"feedback",
0,
enable_feedback_action,
"Enable MaxScale modules list sending via http to notification service",
"Enable MaxScale modules list sending via http to notification service",
{0, 0, 0}
},
{
NULL,
0,
@ -458,6 +473,14 @@ struct subcommand disableoptions[] = {
"Disable root access to a service",
{ARG_TYPE_SERVICE, 0, 0}
},
{
"feedback",
0,
disable_feedback_action,
"Disable MaxScale modules list sending via http to notification service",
"Disable MaxScale modules list sending via http to notification service",
{0, 0, 0}
},
{
NULL,
0,
@ -1172,7 +1195,7 @@ shutdown_monitor(DCB *dcb, MONITOR *monitor)
static void
restart_monitor(DCB *dcb, MONITOR *monitor)
{
monitorStart(monitor);
monitorStart(monitor, NULL);
}
/**
@ -1184,7 +1207,14 @@ restart_monitor(DCB *dcb, MONITOR *monitor)
static void
enable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor)
{
monitorSetReplicationHeartbeat(monitor, 1);
CONFIG_PARAMETER param;
const char* name = "detect_replication_lag";
const char* value = "1";
param.name = (char*)name;
param.value = (char*)value;
param.next = NULL;
monitorStop(monitor);
monitorStart(monitor,&param);
}
/**
@ -1196,7 +1226,14 @@ enable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor)
static void
disable_monitor_replication_heartbeat(DCB *dcb, MONITOR *monitor)
{
monitorSetReplicationHeartbeat(monitor, 0);
CONFIG_PARAMETER param;
const char* name = "detect_replication_lag";
const char* value = "0";
param.name = (char*)name;
param.value = (char*)value;
param.next = NULL;
monitorStop(monitor);
monitorStart(monitor,&param);
}
/**
@ -1381,6 +1418,29 @@ set_nbpoll(DCB *dcb, int nb)
poll_set_nonblocking_polls(nb);
}
/**
* Re-enable sendig MaxScale module list via http
* Proper [feedback] section in MaxSclale.cnf
* is required.
*/
static void
enable_feedback_action(void)
{
config_enable_feedback_task();
return;
}
/**
* Disable sendig MaxScale module list via http
*/
static void
disable_feedback_action(void)
{
config_disable_feedback_task();
return;
}
#if defined(FAKE_CODE)
static void fail_backendfd(void)
{

4
server/modules/routing/maxinfo/CMakeLists.txt Normal file
View File

@ -0,0 +1,4 @@
add_library(maxinfo SHARED maxinfo.c maxinfo_parse.c maxinfo_error.c maxinfo_exec.c)
set_target_properties(maxinfo PROPERTIES INSTALL_RPATH ${CMAKE_INSTALL_RPATH}:${CMAKE_INSTALL_PREFIX}/lib)
target_link_libraries(maxinfo pthread log_manager)
install(TARGETS maxinfo DESTINATION modules)

805
server/modules/routing/maxinfo/maxinfo.c Normal file
View File

@ -0,0 +1,805 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright MariaDB Corporation Ab 2014
*/
/**
* @file maxinfo.c - A "routing module" that in fact merely gives access
* to a MaxScale information schema usign the MySQL protocol
*
* @verbatim
* Revision History
*
* Date Who Description
* 16/02/15 Mark Riddoch Initial implementation
* 27/02/15 Massimiliano Pinto Added maxinfo_add_mysql_user
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <service.h>
#include <session.h>
#include <server.h>
#include <router.h>
#include <modules.h>
#include <modinfo.h>
#include <modutil.h>
#include <monitor.h>
#include <atomic.h>
#include <spinlock.h>
#include <dcb.h>
#include <poll.h>
#include <maxinfo.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <resultset.h>
#include <version.h>
#include <resultset.h>
#include <secrets.h>
#include <users.h>
#include <dbusers.h>
MODULE_INFO info = {
MODULE_API_ROUTER,
MODULE_ALPHA_RELEASE,
ROUTER_VERSION,
"The MaxScale Information Schema"
};
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
extern char *create_hex_sha1_sha1_passwd(char *passwd);
static char *version_str = "V1.0.0";
static int maxinfo_statistics(INFO_INSTANCE *, INFO_SESSION *, GWBUF *);
static int maxinfo_ping(INFO_INSTANCE *, INFO_SESSION *, GWBUF *);
static int maxinfo_execute_query(INFO_INSTANCE *, INFO_SESSION *, char *);
static int handle_url(INFO_INSTANCE *instance, INFO_SESSION *router_session, GWBUF *queue);
static int maxinfo_add_mysql_user(SERVICE *service);
/* The router entry points */
static ROUTER *createInstance(SERVICE *service, char **options);
static void *newSession(ROUTER *instance, SESSION *session);
static void closeSession(ROUTER *instance, void *router_session);
static void freeSession(ROUTER *instance, void *router_session);
static int execute(ROUTER *instance, void *router_session, GWBUF *queue);
static void diagnostics(ROUTER *instance, DCB *dcb);
static uint8_t getCapabilities (ROUTER* inst, void* router_session);
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp);
/** The module object definition */
static ROUTER_OBJECT MyObject = {
createInstance,
newSession,
closeSession,
freeSession,
execute,
diagnostics,
NULL,
handleError,
getCapabilities
};
static SPINLOCK instlock;
static INFO_INSTANCE *instances;
/**
* Implementation of the mandatory version entry point
*
* @return version string of the module
*/
char *
version()
{
return version_str;
}
/**
* The module initialisation routine, called when the module
* is first loaded.
*/
void
ModuleInit()
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Initialise MaxInfo router module %s.\n",
version_str)));
spinlock_init(&instlock);
instances = NULL;
}
/**
* The module entry point routine. It is this routine that
* must populate the structure that is referred to as the
* "module object", this is a structure with the set of
* external entry points for this module.
*
* @return The module object
*/
ROUTER_OBJECT *
GetModuleObject()
{
return &MyObject;
}
/**
* Create an instance of the router for a particular service
* within the gateway.
*
* @param service The service this router is being create for
* @param options Any array of options for the query router
*
* @return The instance data for this new instance
*/
static ROUTER *
createInstance(SERVICE *service, char **options)
{
INFO_INSTANCE *inst;
int i;
if ((inst = malloc(sizeof(INFO_INSTANCE))) == NULL)
return NULL;
inst->service = service;
spinlock_init(&inst->lock);
if (options)
{
for (i = 0; options[i]; i++)
{
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Unknown option for MaxInfo '%s'\n",
options[i])));
}
}
}
/*
* We have completed the creation of the instance data, so now
* insert this router instance into the linked list of routers
* that have been created with this module.
*/
spinlock_acquire(&instlock);
inst->next = instances;
instances = inst;
spinlock_release(&instlock);
/*
* The following add the service user to service->users via mysql_users_alloc()
* password to be used.
*/
maxinfo_add_mysql_user(service);
return (ROUTER *)inst;
}
/**
* Associate a new session with this instance of the router.
*
* @param instance The router instance data
* @param session The session itself
* @return Session specific data for this session
*/
static void *
newSession(ROUTER *instance, SESSION *session)
{
INFO_INSTANCE *inst = (INFO_INSTANCE *)instance;
INFO_SESSION *client;
if ((client = (INFO_SESSION *)malloc(sizeof(INFO_SESSION))) == NULL)
{
return NULL;
}
client->session = session;
client->dcb = session->client;
client->queue = NULL;
spinlock_acquire(&inst->lock);
client->next = inst->sessions;
inst->sessions = client;
spinlock_release(&inst->lock);
session->state = SESSION_STATE_READY;
return (void *)client;
}
/**
* Close a session with the router, this is the mechanism
* by which a router may cleanup data structure etc.
*
* @param instance The router instance data
* @param router_session The session being closed
*/
static void
closeSession(ROUTER *instance, void *router_session)
{
INFO_INSTANCE *inst = (INFO_INSTANCE *)instance;
INFO_SESSION *session = (INFO_SESSION *)router_session;
spinlock_acquire(&inst->lock);
if (inst->sessions == session)
inst->sessions = session->next;
else
{
INFO_SESSION *ptr = inst->sessions;
while (ptr && ptr->next != session)
ptr = ptr->next;
if (ptr)
ptr->next = session->next;
}
spinlock_release(&inst->lock);
/**
* Router session is freed in session.c:session_close, when session who
* owns it, is freed.
*/
}
/**
* Free a maxinfo session
*
* @param router_instance The router session
* @param router_client_session The router session as returned from newSession
*/
static void freeSession(
ROUTER* router_instance,
void* router_client_session)
{
free(router_client_session);
return;
}
/**
* Error Handler routine
*
* The routine will handle errors that occurred in backend writes.
*
* @param instance The router instance
* @param router_session The router session
* @param message The error message to reply
* @param backend_dcb The backend DCB
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
*
*/
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp)
{
DCB *client_dcb;
SESSION *session = backend_dcb->session;
session_state_t sesstate;
/** Reset error handle flag from a given DCB */
if (action == ERRACT_RESET)
{
backend_dcb->dcb_errhandle_called = false;
return;
}
/** Don't handle same error twice on same DCB */
if (backend_dcb->dcb_errhandle_called)
{
/** we optimistically assume that previous call succeed */
*succp = true;
return;
}
else
{
backend_dcb->dcb_errhandle_called = true;
}
spinlock_acquire(&session->ses_lock);
sesstate = session->state;
client_dcb = session->client;
if (sesstate == SESSION_STATE_ROUTER_READY)
{
CHK_DCB(client_dcb);
spinlock_release(&session->ses_lock);
client_dcb->func.write(client_dcb, gwbuf_clone(errbuf));
}
else
{
spinlock_release(&session->ses_lock);
}
/** false because connection is not available anymore */
*succp = false;
}
/**
* We have data from the client, this is a SQL command, or other MySQL
* packet type.
*
* @param instance The router instance
* @param router_session The router session returned from the newSession call
* @param queue The queue of data buffers to route
* @return The number of bytes sent
*/
static int
execute(ROUTER *rinstance, void *router_session, GWBUF *queue)
{
INFO_INSTANCE *instance = (INFO_INSTANCE *)rinstance;
INFO_SESSION *session = (INFO_SESSION *)router_session;
uint8_t *data;
int length, len, residual;
char *sql;
if (GWBUF_TYPE(queue) == GWBUF_TYPE_HTTP)
{
return handle_url(instance, session, queue);
}
if (session->queue)
{
queue = gwbuf_append(session->queue, queue);
session->queue = NULL;
queue = gwbuf_make_contiguous(queue);
}
data = (uint8_t *)GWBUF_DATA(queue);
length = data[0] + (data[1] << 8) + (data[2] << 16);
if (length + 4 > GWBUF_LENGTH(queue))
{
// Incomplete packet, must be buffered
session->queue = queue;
return 1;
}
// We have a complete request in a signle buffer
if (modutil_MySQL_Query(queue, &sql, &len, &residual))
{
sql = strndup(sql, len);
int rc = maxinfo_execute_query(instance, session, sql);
free(sql);
return rc;
}
else
{
switch (MYSQL_COMMAND(queue))
{
case COM_PING:
return maxinfo_ping(instance, session, queue);
case COM_STATISTICS:
return maxinfo_statistics(instance, session, queue);
default:
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"maxinfo: Unexpected MySQL command 0x%x",
MYSQL_COMMAND(queue))));
}
}
return 1;
}
/**
* Display router diagnostics
*
* @param instance Instance of the router
* @param dcb DCB to send diagnostics to
*/
static void
diagnostics(ROUTER *instance, DCB *dcb)
{
return; /* Nothing to do currently */
}
/**
* Capabilities interface for the rotuer
*
* Not used for the maxinfo router
*/
static uint8_t
getCapabilities(
ROUTER* inst,
void* router_session)
{
return 0;
}
/**
* Return some basic statistics from the router in response to a COM_STATISTICS
* request.
*
* @param router The router instance
* @param session The connection that requested the statistics
* @param queue The statistics request
*
* @return non-zero on sucessful send
*/
static int
maxinfo_statistics(INFO_INSTANCE *router, INFO_SESSION *session, GWBUF *queue)
{
char result[1000], *ptr;
GWBUF *ret;
int len;
extern int MaxScaleUptime();
snprintf(result, 1000,
"Uptime: %u Threads: %u Sessions: %u ",
MaxScaleUptime(),
config_threadcount(),
serviceSessionCountAll());
if ((ret = gwbuf_alloc(4 + strlen(result))) == NULL)
return 0;
len = strlen(result);
ptr = GWBUF_DATA(ret);
*ptr++ = len & 0xff;
*ptr++ = (len & 0xff00) >> 8;
*ptr++ = (len & 0xff0000) >> 16;
*ptr++ = 1;
strncpy(ptr, result, len);
return session->dcb->func.write(session->dcb, ret);
}
/**
* Respond to a COM_PING command
*
* @param router The router instance
* @param session The connection that requested the ping
* @param queue The ping request
*/
static int
maxinfo_ping(INFO_INSTANCE *router, INFO_SESSION *session, GWBUF *queue)
{
char *ptr;
GWBUF *ret;
int len;
if ((ret = gwbuf_alloc(5)) == NULL)
return 0;
ptr = GWBUF_DATA(ret);
*ptr++ = 0x01;
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 1;
*ptr = 0; // OK
return session->dcb->func.write(session->dcb, ret);
}
/**
* Populate the version comment with the MaxScale version
*
* @param result The result set
* @param data Pointer to int which is row count
* @return The populated row
*/
static RESULT_ROW *
version_comment(RESULTSET *result, void *data)
{
int *context = (int *)data;
RESULT_ROW *row;
if (*context == 0)
{
(*context)++;
row = resultset_make_row(result);
resultset_row_set(row, 0, MAXSCALE_VERSION);
return row;
}
return NULL;
}
/**
* The hardwired select @@vercom response
*
* @param dcb The DCB of the client
*/
static void
respond_vercom(DCB *dcb)
{
RESULTSET *result;
int context = 0;
if ((result = resultset_create(version_comment, &context)) == NULL)
{
maxinfo_send_error(dcb, 0, "No resources available");
return;
}
resultset_add_column(result, "@@version_comment", 40, COL_TYPE_VARCHAR);
resultset_stream_mysql(result, dcb);
resultset_free(result);
}
/**
* Populate the version comment with the MaxScale version
*
* @param result The result set
* @param data Pointer to int which is row count
* @return The populated row
*/
static RESULT_ROW *
starttime_row(RESULTSET *result, void *data)
{
int *context = (int *)data;
RESULT_ROW *row;
extern time_t MaxScaleStarted;
struct tm tm;
static char buf[40];
if (*context == 0)
{
(*context)++;
row = resultset_make_row(result);
sprintf(buf, "%u", (unsigned int)MaxScaleStarted);
resultset_row_set(row, 0, buf);
return row;
}
return NULL;
}
/**
* The hardwired select ... as starttime response
*
* @param dcb The DCB of the client
*/
static void
respond_starttime(DCB *dcb)
{
RESULTSET *result;
int context = 0;
if ((result = resultset_create(starttime_row, &context)) == NULL)
{
maxinfo_send_error(dcb, 0, "No resources available");
return;
}
resultset_add_column(result, "starttime", 40, COL_TYPE_VARCHAR);
resultset_stream_mysql(result, dcb);
resultset_free(result);
}
/**
* Send a MySQL OK packet to the DCB
*
* @param dcb The DCB to send the OK packet to
* @return result of a write call, non-zero if write was successful
*/
static int
maxinfo_send_ok(DCB *dcb)
{
GWBUF *buf;
uint8_t *ptr;
if ((buf = gwbuf_alloc(11)) == NULL)
return 0;
ptr = GWBUF_DATA(buf);
*ptr++ = 7; // Payload length
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 1; // Seqno
*ptr++ = 0; // ok
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 2;
*ptr++ = 0;
*ptr++ = 0;
*ptr++ = 0;
return dcb->func.write(dcb, buf);
}
/**
* Execute a SQL query against the MaxScale Information Schema
*
* @param instance The instance strcture
* @param session The session pointer
* @param sql The SQL to execute
*/
static int
maxinfo_execute_query(INFO_INSTANCE *instance, INFO_SESSION *session, char *sql)
{
MAXINFO_TREE *tree;
PARSE_ERROR err;
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE,
"maxinfo: SQL statement: '%s' for 0x%x.",
sql, session->dcb)));
if (strcmp(sql, "select @@version_comment limit 1") == 0)
{
respond_vercom(session->dcb);
return 1;
}
/* Below is a kludge for MonYog, if we see
* select unix_timestamp... as starttime
* just return the starttime of MaxScale
*/
if (strncasecmp(sql, "select UNIX_TIMESTAMP",
strlen("select UNIX_TIMESTAMP")) == 0
&& (strstr(sql, "as starttime") != NULL || strstr(sql, "AS starttime") != NULL))
{
respond_starttime(session->dcb);
return 1;
}
if (strcasecmp(sql, "set names 'utf8'") == 0)
{
return maxinfo_send_ok(session->dcb);
}
if (strncasecmp(sql, "set session", 11) == 0)
{
return maxinfo_send_ok(session->dcb);
}
if (strncasecmp(sql, "set autocommit", 14) == 0)
{
return maxinfo_send_ok(session->dcb);
}
if (strncasecmp(sql, "SELECT `ENGINES`.`SUPPORT`", 26) == 0)
{
return maxinfo_send_ok(session->dcb);
}
if ((tree = maxinfo_parse(sql, &err)) == NULL)
{
maxinfo_send_parse_error(session->dcb, sql, err);
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Failed to parse SQL statement: '%s'.",
sql)));
}
else
maxinfo_execute(session->dcb, tree);
return 1;
}
/**
* Session all result set
* @return A resultset for all sessions
*/
static RESULTSET *
maxinfoSessionsAll()
{
return sessionGetList(SESSION_LIST_ALL);
}
/**
* Client session result set
* @return A resultset for all sessions
*/
static RESULTSET *
maxinfoClientSessions()
{
return sessionGetList(SESSION_LIST_CONNECTION);
}
typedef RESULTSET *(*RESULTSETFUNC)();
/**
* Table that maps a URI to a function to call to
* to obtain the result set related to that URI
*/
static struct uri_table {
char *uri;
RESULTSETFUNC func;
} supported_uri[] = {
{ "/services", serviceGetList },
{ "/listeners", serviceGetListenerList },
{ "/modules", moduleGetList },
{ "/monitors", monitorGetList },
{ "/sessions", maxinfoSessionsAll },
{ "/clients", maxinfoClientSessions },
{ "/servers", serverGetList },
{ "/variables", maxinfo_variables },
{ "/status", maxinfo_status },
{ "/event/times", eventTimesGetList },
{ NULL, NULL }
};
/**
* We have data from the client, this is a HTTP URL
*
* @param instance The router instance
* @param session The router session returned from the newSession call
* @param queue The queue of data buffers to route
* @return The number of bytes sent
*/
static int
handle_url(INFO_INSTANCE *instance, INFO_SESSION *session, GWBUF *queue)
{
char *uri;
int i;
RESULTSET *set;
uri = (char *)GWBUF_DATA(queue);
for (i = 0; supported_uri[i].uri; i++)
{
if (strcmp(uri, supported_uri[i].uri) == 0)
{
set = (*supported_uri[i].func)();
resultset_stream_json(set, session->dcb);
resultset_free(set);
}
}
return 1;
}
/**
* Add the service user to the service->users
* via mysql_users_alloc and add_mysql_users_with_host_ipv4
* User is added for '%' and 'localhost' hosts
*
* @param service The service for this router
* @return 0 on success, 1 on failure
*/
static int
maxinfo_add_mysql_user(SERVICE *service) {
char *dpwd = NULL;
char *newpasswd = NULL;
char *service_user = NULL;
char *service_passwd = NULL;
if (serviceGetUser(service, &service_user, &service_passwd) == 0) {
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"maxinfo: failed to get service user details")));
return 1;
}
dpwd = decryptPassword(service->credentials.authdata);
if (!dpwd) {
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"maxinfo: decrypt password failed for service user %s",
service_user)));
return 1;
}
service->users = (void *)mysql_users_alloc();
newpasswd = create_hex_sha1_sha1_passwd(dpwd);
if (!newpasswd) {
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"maxinfo: create hex_sha1_sha1_password failed for service user %s",
service_user)));
users_free(service->users);
return 1;
}
/* add service user for % and localhost */
(void)add_mysql_users_with_host_ipv4(service->users, service->credentials.name, "%", newpasswd, "Y", "");
(void)add_mysql_users_with_host_ipv4(service->users, service->credentials.name, "localhost", newpasswd, "Y", "");
free(newpasswd);
free(dpwd);
return 0;
}

119
server/modules/routing/maxinfo/maxinfo_error.c Normal file
View File

@ -0,0 +1,119 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright MariaDB Corporation Ab 2014
*/
/**
* @file maxinfo_error.c - Handle error reporting for the maxinfo router
*
* @verbatim
* Revision History
*
* Date Who Description
* 17/02/15 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <service.h>
#include <session.h>
#include <router.h>
#include <modules.h>
#include <modinfo.h>
#include <modutil.h>
#include <atomic.h>
#include <spinlock.h>
#include <dcb.h>
#include <poll.h>
#include <maxinfo.h>
#include <skygw_utils.h>
#include <log_manager.h>
/** Defined in log_manager.cc */
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
/**
* Process a parse error and send error report to client
*
* @param dcb The DCB to send to error
* @param sql The SQL that had the parse error
* @param err The parse error code
*/
void
maxinfo_send_parse_error(DCB *dcb, char *sql, PARSE_ERROR err)
{
char *desc = "";
char *msg;
int len;
switch (err)
{
case PARSE_NOERROR:
desc = "No error";
break;
case PARSE_MALFORMED_SHOW:
desc = "Expected show <command> [like <pattern>]";
break;
case PARSE_EXPECTED_LIKE:
desc = "Expected LIKE <pattern>";
break;
case PARSE_SYNTAX_ERROR:
desc = "Syntax error";
break;
}
len = strlen(sql) + strlen(desc) + 20;
if ((msg = (char *)malloc(len)) == NULL)
return;
sprintf(msg, "%s in query '%s'", desc, sql);
maxinfo_send_error(dcb, 1149, msg);
free(msg);
}
/**
* Construct an error response
*
* @param dcb The DCB to send the error packet to
* @param msg The slave server instance
*/
void
maxinfo_send_error(DCB *dcb, int errcode, char *msg)
{
GWBUF *pkt;
unsigned char *data;
int len;
len = strlen(msg) + 9;
if ((pkt = gwbuf_alloc(len + 4)) == NULL)
return;
data = GWBUF_DATA(pkt);
data[0] = len & 0xff; // Payload length
data[1] = (len >> 8) & 0xff;
data[2] = (len >> 16) & 0xff;
data[3] = 1; // Sequence id
// Payload
data[4] = 0xff; // Error indicator
data[5] = errcode & 0xff; // Error Code
data[6] = (errcode >> 8) & 0xff; // Error Code
memcpy(&data[7], "#42000", 6);
memcpy(&data[13], msg, strlen(msg)); // Error Message
dcb->func.write(dcb, pkt);
}

770
server/modules/routing/maxinfo/maxinfo_exec.c Normal file
View File

@ -0,0 +1,770 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright MariaDB Corporation Ab 2014
*/
/**
* @file maxinfo_parse.c - Parse the limited set of SQL that the MaxScale
* information schema can use
*
* @verbatim
* Revision History
*
* Date Who Description
* 17/02/15 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <service.h>
#include <session.h>
#include <router.h>
#include <modules.h>
#include <monitor.h>
#include <version.h>
#include <modinfo.h>
#include <modutil.h>
#include <atomic.h>
#include <spinlock.h>
#include <dcb.h>
#include <poll.h>
#include <maxinfo.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <resultset.h>
#include <maxconfig.h>
extern int lm_enabled_logfiles_bitmask;
extern size_t log_ses_count[];
extern __thread log_info_t tls_log_info;
static void exec_show(DCB *dcb, MAXINFO_TREE *tree);
static void exec_select(DCB *dcb, MAXINFO_TREE *tree);
static void exec_show_variables(DCB *dcb, MAXINFO_TREE *filter);
static void exec_show_status(DCB *dcb, MAXINFO_TREE *filter);
static int maxinfo_pattern_match(char *pattern, char *str);
/**
* Execute a parse tree and return the result set or runtime error
*
* @param dcb The DCB that connects to the client
* @param tree The parse tree for the query
*/
void
maxinfo_execute(DCB *dcb, MAXINFO_TREE *tree)
{
switch (tree->op)
{
case MAXOP_SHOW:
exec_show(dcb, tree);
break;
case MAXOP_SELECT:
exec_select(dcb, tree);
break;
case MAXOP_TABLE:
case MAXOP_COLUMNS:
case MAXOP_LITERAL:
case MAXOP_PREDICATE:
case MAXOP_LIKE:
case MAXOP_EQUAL:
default:
maxinfo_send_error(dcb, 0, "Unexpected operator in parse tree");
}
}
/**
* Fetch the list of services and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_services(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = serviceGetList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of listeners and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_listeners(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = serviceGetListenerList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of sessions and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_sessions(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = sessionGetList(SESSION_LIST_ALL)) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of client sessions and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_clients(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = sessionGetList(SESSION_LIST_CONNECTION)) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of servers and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_servers(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = serverGetList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of modules and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_modules(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = moduleGetList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the list of monitors and stream as a result set
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_monitors(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = monitorGetList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* Fetch the event times data
*
* @param dcb DCB to which to stream result set
* @param tree Potential like clause (currently unused)
*/
static void
exec_show_eventTimes(DCB *dcb, MAXINFO_TREE *tree)
{
RESULTSET *set;
if ((set = eventTimesGetList()) == NULL)
return;
resultset_stream_mysql(set, dcb);
resultset_free(set);
}
/**
* The table of show commands that are supported
*/
static struct {
char *name;
void (*func)(DCB *, MAXINFO_TREE *);
} show_commands[] = {
{ "variables", exec_show_variables },
{ "status", exec_show_status },
{ "services", exec_show_services },
{ "listeners", exec_show_listeners },
{ "sessions", exec_show_sessions },
{ "clients", exec_show_clients },
{ "servers", exec_show_servers },
{ "modules", exec_show_modules },
{ "monitors", exec_show_monitors },
{ "eventTimes", exec_show_eventTimes },
{ NULL, NULL }
};
/**
* Execute a show command parse tree and return the result set or runtime error
*
* @param dcb The DCB that connects to the client
* @param tree The parse tree for the query
*/
static void
exec_show(DCB *dcb, MAXINFO_TREE *tree)
{
int i;
char errmsg[120];
for (i = 0; show_commands[i].name; i++)
{
if (strcasecmp(show_commands[i].name, tree->value) == 0)
{
(*show_commands[i].func)(dcb, tree->right);
return;
}
}
if (strlen(tree->value) > 80) // Prevent buffer overrun
tree->value[80] = 0;
sprintf(errmsg, "Unsupported show command '%s'", tree->value);
maxinfo_send_error(dcb, 0, errmsg);
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE, errmsg)));
}
/**
* Return the current MaxScale version
*
* @return The version string for MaxScale
*/
static char *
getVersion()
{
return MAXSCALE_VERSION;
}
static char *versionComment = "MariaDB MaxScale";
/**
* Return the current MaxScale version
*
* @return The version string for MaxScale
*/
static char *
getVersionComment()
{
return versionComment;
}
/**
* Return the current MaxScale Home Directory
*
* @return The version string for MaxScale
*/
static char *
getMaxScaleHome()
{
return getenv("MAXSCALE_HOME");
}
/* The various methods to fetch the variables */
#define VT_STRING 1
#define VT_INT 2
extern int MaxScaleUptime();
typedef void *(*STATSFUNC)();
/**
* Variables that may be sent in a show variables
*/
static struct {
char *name;
int type;
STATSFUNC func;
} variables[] = {
{ "version", VT_STRING, (STATSFUNC)getVersion },
{ "version_comment", VT_STRING, (STATSFUNC)getVersionComment },
{ "basedir", VT_STRING, (STATSFUNC)getMaxScaleHome},
{ "MAXSCALE_VERSION", VT_STRING, (STATSFUNC)getVersion },
{ "MAXSCALE_THREADS", VT_INT, (STATSFUNC)config_threadcount },
{ "MAXSCALE_NBPOLLS", VT_INT, (STATSFUNC)config_nbpolls },
{ "MAXSCALE_POLLSLEEP", VT_INT, (STATSFUNC)config_pollsleep },
{ "MAXSCALE_UPTIME", VT_INT, (STATSFUNC)MaxScaleUptime },
{ "MAXSCALE_SESSIONS", VT_INT, (STATSFUNC)serviceSessionCountAll },
{ NULL, 0, NULL }
};
typedef struct {
int index;
char *like;
} VARCONTEXT;
/**
* Callback function to populate rows of the show variable
* command
*
* @param data The context point
* @return The next row or NULL if end of rows
*/
static RESULT_ROW *
variable_row(RESULTSET *result, void *data)
{
VARCONTEXT *context = (VARCONTEXT *)data;
RESULT_ROW *row;
char buf[80];
if (variables[context->index].name)
{
if (context->like &&
maxinfo_pattern_match(context->like,
variables[context->index].name))
{
context->index++;
return variable_row(result, data);
}
row = resultset_make_row(result);
resultset_row_set(row, 0, variables[context->index].name);
switch (variables[context->index].type)
{
case VT_STRING:
resultset_row_set(row, 1,
(char *)(*variables[context->index].func)());
break;
case VT_INT:
snprintf(buf, 80, "%ld",
(long)(*variables[context->index].func)());
resultset_row_set(row, 1, buf);
break;
}
context->index++;
return row;
}
return NULL;
}
/**
* Execute a show variables command applying an optional filter
*
* @param dcb The DCB connected to the client
* @param filter A potential like clause or NULL
*/
static void
exec_show_variables(DCB *dcb, MAXINFO_TREE *filter)
{
RESULTSET *result;
VARCONTEXT context;
if (filter)
context.like = filter->value;
else
context.like = NULL;
context.index = 0;
if ((result = resultset_create(variable_row, &context)) == NULL)
{
maxinfo_send_error(dcb, 0, "No resources available");
return;
}
resultset_add_column(result, "Variable_name", 40, COL_TYPE_VARCHAR);
resultset_add_column(result, "Value", 40, COL_TYPE_VARCHAR);
resultset_stream_mysql(result, dcb);
resultset_free(result);
}
/**
* Return the show variables output a a result set
*
* @return Variables as a result set
*/
RESULTSET *
maxinfo_variables()
{
RESULTSET *result;
static VARCONTEXT context;
context.like = NULL;
context.index = 0;
if ((result = resultset_create(variable_row, &context)) == NULL)
{
return NULL;
}
resultset_add_column(result, "Variable_name", 40, COL_TYPE_VARCHAR);
resultset_add_column(result, "Value", 40, COL_TYPE_VARCHAR);
return result;
}
/**
* Interface to dcb_count_by_usage for all dcbs
*/
static int
maxinfo_all_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_ALL);
}
/**
* Interface to dcb_count_by_usage for client dcbs
*/
static int
maxinfo_client_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_CLIENT);
}
/**
* Interface to dcb_count_by_usage for listener dcbs
*/
static int
maxinfo_listener_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_LISTENER);
}
/**
* Interface to dcb_count_by_usage for backend dcbs
*/
static int
maxinfo_backend_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_BACKEND);
}
/**
* Interface to dcb_count_by_usage for internal dcbs
*/
static int
maxinfo_internal_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_INTERNAL);
}
/**
* Interface to dcb_count_by_usage for zombie dcbs
*/
static int
maxinfo_zombie_dcbs()
{
return dcb_count_by_usage(DCB_USAGE_ZOMBIE);
}
/**
* Interface to poll stats for reads
*/
static int
maxinfo_read_events()
{
return poll_get_stat(POLL_STAT_READ);
}
/**
* Interface to poll stats for writes
*/
static int
maxinfo_write_events()
{
return poll_get_stat(POLL_STAT_WRITE);
}
/**
* Interface to poll stats for errors
*/
static int
maxinfo_error_events()
{
return poll_get_stat(POLL_STAT_ERROR);
}
/**
* Interface to poll stats for hangup
*/
static int
maxinfo_hangup_events()
{
return poll_get_stat(POLL_STAT_HANGUP);
}
/**
* Interface to poll stats for accepts
*/
static int
maxinfo_accept_events()
{
return poll_get_stat(POLL_STAT_ACCEPT);
}
/**
* Interface to poll stats for event queue length
*/
static int
maxinfo_event_queue_length()
{
return poll_get_stat(POLL_STAT_EVQ_LEN);
}
/**
* Interface to poll stats for event pending queue length
*/
static int
maxinfo_event_pending_queue_length()
{
return poll_get_stat(POLL_STAT_EVQ_PENDING);
}
/**
* Interface to poll stats for max event queue length
*/
static int
maxinfo_max_event_queue_length()
{
return poll_get_stat(POLL_STAT_EVQ_MAX);
}
/**
* Interface to poll stats for max queue time
*/
static int
maxinfo_max_event_queue_time()
{
return poll_get_stat(POLL_STAT_MAX_QTIME);
}
/**
* Interface to poll stats for max event execution time
*/
static int
maxinfo_max_event_exec_time()
{
return poll_get_stat(POLL_STAT_MAX_EXECTIME);
}
/**
* Variables that may be sent in a show status
*/
static struct {
char *name;
int type;
STATSFUNC func;
} status[] = {
{ "Uptime", VT_INT, (STATSFUNC)MaxScaleUptime },
{ "Uptime_since_flush_status", VT_INT, (STATSFUNC)MaxScaleUptime },
{ "Threads_created", VT_INT, (STATSFUNC)config_threadcount },
{ "Threads_running", VT_INT, (STATSFUNC)config_threadcount },
{ "Threadpool_threads", VT_INT, (STATSFUNC)config_threadcount },
{ "Threads_connected", VT_INT, (STATSFUNC)serviceSessionCountAll },
{ "Connections", VT_INT, (STATSFUNC)maxinfo_all_dcbs },
{ "Client_connections", VT_INT, (STATSFUNC)maxinfo_client_dcbs },
{ "Backend_connections", VT_INT, (STATSFUNC)maxinfo_backend_dcbs },
{ "Listeners", VT_INT, (STATSFUNC)maxinfo_listener_dcbs },
{ "Zombie_connections", VT_INT, (STATSFUNC)maxinfo_zombie_dcbs },
{ "Internal_descriptors", VT_INT, (STATSFUNC)maxinfo_internal_dcbs },
{ "Read_events", VT_INT, (STATSFUNC)maxinfo_read_events },
{ "Write_events", VT_INT, (STATSFUNC)maxinfo_write_events },
{ "Hangup_events", VT_INT, (STATSFUNC)maxinfo_hangup_events },
{ "Error_events", VT_INT, (STATSFUNC)maxinfo_error_events },
{ "Accept_events", VT_INT, (STATSFUNC)maxinfo_accept_events },
{ "Event_queue_length", VT_INT, (STATSFUNC)maxinfo_event_queue_length },
{ "Pending_events", VT_INT, (STATSFUNC)maxinfo_event_pending_queue_length },
{ "Max_event_queue_length", VT_INT, (STATSFUNC)maxinfo_max_event_queue_length },
{ "Max_event_queue_time", VT_INT, (STATSFUNC)maxinfo_max_event_queue_time },
{ "Max_event_execution_time", VT_INT, (STATSFUNC)maxinfo_max_event_exec_time },
{ NULL, 0, NULL }
};
/**
* Callback function to populate rows of the show variable
* command
*
* @param data The context point
* @return The next row or NULL if end of rows
*/
static RESULT_ROW *
status_row(RESULTSET *result, void *data)
{
VARCONTEXT *context = (VARCONTEXT *)data;
RESULT_ROW *row;
char buf[80];
if (status[context->index].name)
{
if (context->like &&
maxinfo_pattern_match(context->like,
status[context->index].name))
{
context->index++;
return status_row(result, data);
}
row = resultset_make_row(result);
resultset_row_set(row, 0, status[context->index].name);
switch (status[context->index].type)
{
case VT_STRING:
resultset_row_set(row, 1,
(char *)(*status[context->index].func)());
break;
case VT_INT:
snprintf(buf, 80, "%ld",
(long)(*status[context->index].func)());
resultset_row_set(row, 1, buf);
break;
}
context->index++;
return row;
}
return NULL;
}
/**
* Execute a show status command applying an optional filter
*
* @param dcb The DCB connected to the client
* @param filter A potential like clause or NULL
*/
static void
exec_show_status(DCB *dcb, MAXINFO_TREE *filter)
{
RESULTSET *result;
VARCONTEXT context;
if (filter)
context.like = filter->value;
else
context.like = NULL;
context.index = 0;
if ((result = resultset_create(status_row, &context)) == NULL)
{
maxinfo_send_error(dcb, 0, "No resources available");
return;
}
resultset_add_column(result, "Variable_name", 40, COL_TYPE_VARCHAR);
resultset_add_column(result, "Value", 40, COL_TYPE_VARCHAR);
resultset_stream_mysql(result, dcb);
resultset_free(result);
}
/**
* Return the show status data as a result set
*
* @return The show status data as a result set
*/
RESULTSET *
maxinfo_status()
{
RESULTSET *result;
static VARCONTEXT context;
context.like = NULL;
context.index = 0;
if ((result = resultset_create(status_row, &context)) == NULL)
{
return NULL;
}
resultset_add_column(result, "Variable_name", 40, COL_TYPE_VARCHAR);
resultset_add_column(result, "Value", 40, COL_TYPE_VARCHAR);
return result;
}
/**
* Execute a select command parse tree and return the result set
* or runtime error
*
* @param dcb The DCB that connects to the client
* @param tree The parse tree for the query
*/
static void
exec_select(DCB *dcb, MAXINFO_TREE *tree)
{
maxinfo_send_error(dcb, 0, "Select not yet implemented");
}
/**
* Perform a "like" pattern match. Only works for leading and trailing %
*
* @param pattern Pattern to match
* @param str String to match against pattern
* @return Zero on match
*/
static int
maxinfo_pattern_match(char *pattern, char *str)
{
int anchor = 0, len, trailing;
char *fixed;
extern char *strcasestr();
if (*pattern != '%')
{
fixed = pattern;
anchor = 1;
}
else
{
fixed = &pattern[1];
}
len = strlen(fixed);
if (fixed[len - 1] == '%')
trailing = 1;
else
trailing = 0;
if (anchor == 1 && trailing == 0) // No wildcard
return strcasecmp(pattern, str);
else if (anchor == 1)
return strncasecmp(str, pattern, len - trailing);
else
{
char *portion = malloc(len + 1);
int rval;
strncpy(portion, fixed, len - trailing);
portion[len - trailing] = 0;
rval = (strcasestr(str, portion) != NULL ? 0 : 1);
free(portion);
return rval;
}
}

328
server/modules/routing/maxinfo/maxinfo_parse.c Normal file
View File

@ -0,0 +1,328 @@
/*
* This file is distributed as part of MaxScale. It is free
* software: you can redistribute it and/or modify it under the terms of the
* GNU General Public License as published by the Free Software Foundation,
* version 2.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
* details.
*
* You should have received a copy of the GNU General Public License along with
* this program; if not, write to the Free Software Foundation, Inc., 51
* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Copyright MariaDB Corporation Ab 2014
*/
/**
* @file maxinfo_parse.c - Parse the limited set of SQL that the MaxScale
* information schema can use
*
* @verbatim
* Revision History
*
* Date Who Description
* 16/02/15 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <service.h>
#include <session.h>
#include <router.h>
#include <modules.h>
#include <modinfo.h>
#include <modutil.h>
#include <atomic.h>
#include <spinlock.h>
#include <dcb.h>
#include <poll.h>
#include <maxinfo.h>
#include <skygw_utils.h>
#include <log_manager.h>
static MAXINFO_TREE *make_tree_node(MAXINFO_OPERATOR, char *, MAXINFO_TREE *, MAXINFO_TREE *);
static void free_tree(MAXINFO_TREE *);
static char *fetch_token(char *, int *, char **);
static MAXINFO_TREE *parse_column_list(char **sql);
static MAXINFO_TREE *parse_table_name(char **sql);
/**
* Parse a SQL subset for the maxinfo plugin and return a parse tree
*
* @param sql The SQL query
* @return Parse tree or NULL on error
*/
MAXINFO_TREE *
maxinfo_parse(char *sql, PARSE_ERROR *parse_error)
{
int token;
char *ptr, *text;
MAXINFO_TREE *tree = NULL;
MAXINFO_TREE *col, *table;
*parse_error = PARSE_NOERROR;
while ((ptr = fetch_token(sql, &token, &text)) != NULL)
{
switch (token)
{
case LT_SHOW:
free(text); // not needed
ptr = fetch_token(ptr, &token, &text);
if (ptr == NULL || token != LT_STRING)
{
// Expected show "name"
*parse_error = PARSE_MALFORMED_SHOW;
return NULL;
}
tree = make_tree_node(MAXOP_SHOW, text, NULL, NULL);
if ((ptr = fetch_token(ptr, &token, &text)) == NULL)
return tree;
else if (token == LT_LIKE)
{
if ((ptr = fetch_token(ptr, &token, &text)) != NULL)
{
tree->right = make_tree_node(MAXOP_LIKE,
text, NULL, NULL);
return tree;
}
else
{
// Expected expression
*parse_error = PARSE_EXPECTED_LIKE;
free_tree(tree);
return NULL;
}
}
// Malformed show
free(text);
free_tree(tree);
*parse_error = PARSE_MALFORMED_SHOW;
return NULL;
#if 0
case LT_SELECT:
free(text); // not needed
col = parse_column_list(&ptr);
table = parse_table_name(&ptr);
return make_tree_node(MAXOP_SELECT, NULL, col, table);
#endif
default:
*parse_error = PARSE_SYNTAX_ERROR;
if (tree)
free_tree(tree);
return NULL;
}
}
*parse_error = PARSE_SYNTAX_ERROR;
if (tree)
free_tree(tree);
return NULL;
}
/**
* Parse a column list, may be a * or a valid list of string name
* separated by a comma
*
* @param sql Pointer to pointer to column list updated to point to the table name
* @return A tree of column names
*/
static MAXINFO_TREE *
parse_column_list(char **ptr)
{
int token, lookahead;
char *text, *text2;
MAXINFO_TREE *tree = NULL;
MAXINFO_TREE * rval = NULL;
*ptr = fetch_token(*ptr, &token, &text);
*ptr = fetch_token(*ptr, &lookahead, &text2);
switch (token)
{
case LT_STRING:
switch (lookahead)
{
case LT_COMMA:
rval = make_tree_node(MAXOP_COLUMNS, text, NULL,
parse_column_list(ptr));
break;
case LT_FROM:
rval = make_tree_node(MAXOP_COLUMNS, text, NULL,
NULL);
break;
default:
break;
}
break;
case LT_STAR:
if (lookahead != LT_FROM)
rval = make_tree_node(MAXOP_ALL_COLUMNS, NULL, NULL,
NULL);
break;
default:
break;
}
free(text);
free(text2);
return rval;
}
/**
* Parse a table name
*
* @param sql Pointer to pointer to column list updated to point to the table name
* @return A tree of table names
*/
static MAXINFO_TREE *
parse_table_name(char **ptr)
{
int token;
char *text;
MAXINFO_TREE *tree = NULL;
*ptr = fetch_token(*ptr, &token, &text);
if (token == LT_STRING)
return make_tree_node(MAXOP_TABLE, text, NULL, NULL);
free(text);
return NULL;
}
/**
* Allocate and populate a parse tree node
*
* @param op The node operator
* @param value The node value
* @param left The left branch of the parse tree
* @param right The right branch of the parse tree
* @return The new parse tree node
*/
static MAXINFO_TREE *
make_tree_node(MAXINFO_OPERATOR op, char *value, MAXINFO_TREE *left, MAXINFO_TREE *right)
{
MAXINFO_TREE *node;
if ((node = (MAXINFO_TREE *)malloc(sizeof(MAXINFO_TREE))) == NULL)
return NULL;
node->op = op;
node->value = value;
node->left = left;
node->right = right;
return node;
}
/**
* Recusrsively free the storage associated with a parse tree
*
* @param tree The parse tree to free
*/
static void
free_tree(MAXINFO_TREE *tree)
{
if (tree->left)
free_tree(tree->left);
if (tree->right)
free_tree(tree->right);
if (tree->value)
free(tree->value);
free(tree);
}
/**
* The set of keywords known to the tokeniser
*/
static struct {
char *text;
int token;
} keywords[] = {
{ "show", LT_SHOW },
{ "select", LT_SELECT },
{ "from", LT_FROM },
{ "like", LT_LIKE },
{ "=", LT_EQUAL },
{ ",", LT_COMMA },
{ "*", LT_STAR },
{ NULL, 0 }
};
/**
* Limited SQL tokeniser. Understands a limited set of key words and
* quoted strings.
*
* @param sql The SQL to tokenise
* @param token The returned token
* @param text The matching text
* @return The next position to tokenise from
*/
static char *
fetch_token(char *sql, int *token, char **text)
{
char *s1, *s2, quote = '\0';
int i;
s1 = sql;
while (*s1 && isspace(*s1))
{
s1++;
}
if (quote == '\0' && (*s1 == '\'' || *s1 == '\"'))
{
quote = *s1++;
}
if (*s1 == '/' && *(s1 + 1) == '*')
{
s1 += 2;
// Skip the comment
do {
while (*s1 && *s1 != '*')
s1++;
} while (*(s1 + 1) && *(s1 + 1) != '/');
s1 += 2;
while (*s1 && isspace(*s1))
{
s1++;
}
if (quote == '\0' && (*s1 == '\'' || *s1 == '\"'))
{
quote = *s1++;
}
}
s2 = s1;
while (*s2)
{
if (quote == '\0' && (isspace(*s2)
|| *s2 == ',' || *s2 == '='))
break;
else if (quote == *s2)
{
break;
}
s2++;
}
if (*s1 == '@' && *(s1 + 1) == '@')
{
*text = strndup(s1 + 2, (s2 - s1) - 2);
*token = LT_VARIABLE;
return s2;
}
if (s1 == s2)
return NULL;
*text = strndup(s1, s2 - s1);
for (i = 0; keywords[i].text; i++)
{
if (strcasecmp(keywords[i].text, *text) == 0)
{
*token = keywords[i].token;
return s2;
}
}
*token = LT_STRING;
return s2;
}

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

@ -311,6 +311,11 @@ char *weightby;
inst->bitmask |= (SERVER_MASTER|SERVER_SLAVE);
inst->bitvalue |= SERVER_SLAVE;
}
else if (!strcasecmp(options[i], "running"))
{
inst->bitmask |= (SERVER_RUNNING);
inst->bitvalue |= SERVER_RUNNING;
}
else if (!strcasecmp(options[i], "synced"))
{
inst->bitmask |= (SERVER_JOINED);
@ -333,7 +338,12 @@ char *weightby;
}
}
}
if(inst->bitmask == 0 && inst->bitvalue == 0)
{
/** No parameters given, use RUNNING as a valid server */
inst->bitmask |= (SERVER_RUNNING);
inst->bitvalue |= SERVER_RUNNING;
}
/*
* We have completed the creation of the instance data, so now
* insert this router instance into the linked list of routers

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

@ -790,6 +790,7 @@ static void* newSession(
#endif
client_rses->router = router;
client_rses->client_dcb = session->client;
/**
* If service config has been changed, reload config from service to
* router instance first.
@ -3509,8 +3510,6 @@ static bool select_connect_backend_servers(
atomic_add(&backend_ref[i].bref_backend->backend_conn_count, -1);
}
}
master_connected = false;
slaves_connected = 0;
}
return_succp:
@ -3755,10 +3754,11 @@ static GWBUF* sescmd_cursor_process_replies(
bref_clear_state(bref,BREF_IN_USE);
bref_set_state(bref,BREF_CLOSED);
bref_set_state(bref,BREF_SESCMD_FAILED);
dcb_close(bref->bref_dcb);
if(bref->bref_dcb)
dcb_close(bref->bref_dcb);
*reconnect = true;
if(replybuf)
gwbuf_free(replybuf);
gwbuf_consume(replybuf,gwbuf_length(replybuf));
}
}
/** This is a response from the master and it is the "right" one.
@ -3788,7 +3788,8 @@ static GWBUF* sescmd_cursor_process_replies(
bref_clear_state(&ses->rses_backend_ref[i],BREF_IN_USE);
bref_set_state(&ses->rses_backend_ref[i],BREF_CLOSED);
bref_set_state(bref,BREF_SESCMD_FAILED);
dcb_close(ses->rses_backend_ref[i].bref_dcb);
if(ses->rses_backend_ref[i].bref_dcb)
dcb_close(ses->rses_backend_ref[i].bref_dcb);
*reconnect = true;
}
}
@ -4345,6 +4346,21 @@ static bool route_session_write(
goto return_succp;
}
if(router_cli_ses->rses_config.rw_max_sescmd_history_size > 0 &&
router_cli_ses->rses_nsescmd >= router_cli_ses->rses_config.rw_max_sescmd_history_size)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Router session exceeded session command history limit. "
"Closing router session. <")));
gwbuf_free(querybuf);
rses_end_locked_router_action(router_cli_ses);
router_cli_ses->client_dcb->func.hangup(router_cli_ses->client_dcb);
goto return_succp;
}
/**
* Additional reference is created to querybuf to
* prevent it from being released before properties
@ -4416,6 +4432,9 @@ static bool route_session_write(
}
}
}
atomic_add(&router_cli_ses->rses_nsescmd,1);
/** Unlock router session */
rses_end_locked_router_action(router_cli_ses);
@ -4515,6 +4534,10 @@ static void rwsplit_process_router_options(
router->rwsplit_config.rw_slave_select_criteria = c;
}
}
else if(strcmp(options[i], "max_sescmd_history") == 0)
{
router->rwsplit_config.rw_max_sescmd_history_size = atoi(value);
}
}
} /*< for */
}

4190
server/modules/routing/schemarouter/schemarouter.c Normal file
View File

File diff suppressed because it is too large Load Diff

2991
server/modules/routing/schemarouter/shardrouter.c Normal file
View File

File diff suppressed because it is too large Load Diff

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

@ -0,0 +1,9 @@
if(MYSQLCLIENT_FOUND AND BUILD_TESTS)
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/test.cmake.in ${CMAKE_CURRENT_BINARY_DIR}/test.cmake @ONLY)
add_executable(testschemarouter testschemarouter.c)
target_link_libraries(testschemarouter ${MYSQLCLIENT_LIBRARIES} ssl crypto dl z m rt pthread)
add_executable(testschemarouter2 testschemarouter2.c)
target_link_libraries(testschemarouter2 ${MYSQLCLIENT_LIBRARIES} ssl crypto dl z m rt pthread)
add_test(NAME TestSchemaRouter COMMAND ${CMAKE_COMMAND} -P ${CMAKE_CURRENT_BINARY_DIR}/test.cmake)
endif()

9
server/modules/routing/schemarouter/test/portblock.sh Executable file
View File

@ -0,0 +1,9 @@
#!/bin/bash
if [ $# -lt 1 ]
then
echo "Usage $0 <port to block>"
exit 1
fi
sudo iptables -I INPUT 1 -i lo -p tcp --dport $1 -j DROP
sudo iptables -I INPUT 1 -i lo -p tcp --sport $1 -j DROP
echo "Traffic to port $1 blocked."

19
server/modules/routing/schemarouter/test/prepare_shard.sh Executable file
View File

@ -0,0 +1,19 @@
#!/bin/bash
if [ $# -lt 5 ]
then
printf "Not enough arguments: '"
for i in $@
do
printf "$i "
done
echo "'given, 5 needed."
echo "usage $0 <host> <port> <username> <password> <database name>"
exit 1
fi
HOST=$1
PORT=$2
USER=$3
PW=$4
SHD=$5
mysql -u $USER -p$PW -P $PORT -h $HOST -e "create database $SHD;"
echo "Created database \"$SHD\" at $HOST:$PORT"

17
server/modules/routing/schemarouter/test/test.cmake.in Normal file
View File

@ -0,0 +1,17 @@
set(SCHEMAROUTER_TEST_PORTS 3000 3001 3002 3003)
foreach(VAR ${SCHEMAROUTER_TEST_PORTS})
execute_process(COMMAND ${CMAKE_CURRENT_SOURCE_DIR}/prepare_shard.sh @TEST_HOST@ ${VAR} @TEST_USER@ @TEST_PASSWORD@ "db${VAR}")
endforeach()
execute_process(COMMAND ${CMAKE_CURRENT_BINARY_DIR}/testschemarouter @TEST_HOST@ @TEST_PORT_DB@ @TEST_USER@ @TEST_PASSWORD@ RESULT_VARIABLE RVAL)
if(RVAL EQUAL 0)
message("Test 1 passed.")
else()
message(FATAL_ERROR "Test 1 failed with code ${RVAL}.")
endif()
execute_process(COMMAND ${CMAKE_CURRENT_BINARY_DIR}/testschemarouter2 @TEST_HOST@ @TEST_PORT_DB@ @TEST_USER@ @TEST_PASSWORD@ RESULT_VARIABLE RVAL2)
if(RVAL2 EQUAL 0)
message("Test 2 passed.")
else()
message(FATAL_ERROR "Test 2 failed with code ${RVAL2}.")
endif()

210
server/modules/routing/schemarouter/test/testschemarouter.c Normal file
View File

@ -0,0 +1,210 @@
#include <my_config.h>
#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mysqld_error.h>
int main(int argc, char** argv)
{
/**
* This test sets a session variable, creates tables in each of the shards
* and inserts into them a single value while querying the session variable.
* This will show if the value in the session variable in the shard is set and
* if it is the same in all the shards.
*
* The test fails if any of the session variables is not set or differ from the original value.
*/
MYSQL* server;
MYSQL_RES *result,*shdres;
MYSQL_ROW row;
char *host = NULL,*username = NULL, *password = NULL;
char query[2048];
unsigned int port,errnum;
unsigned long *lengths;
int rval;
if(argc < 5)
{
fprintf(stderr,"Usage: %s <host> <port> <username> <password>\n",argv[0]);
return 1;
}
host = strdup(argv[1]);
port = atoi(argv[2]);
username = strdup(argv[3]);
password = strdup(argv[4]);
rval = 0;
printf("Connecting to %s:%d as %s/%s\n",host,port,username,password);
if((server = mysql_init(NULL)) == NULL){
fprintf(stderr,"Error : Initialization of MySQL client failed.\n");
rval = 1;
goto report;
}
if(mysql_real_connect(server,host,username,password,NULL,port,NULL,0) == NULL){
fprintf(stderr, "Failed to connect to database: %s\n",
mysql_error(server));
rval = 1;
goto report;
}
if(mysql_real_query(server,
"SET @test=123",
strlen("SET @test=123")))
{
fprintf(stderr, "Failed to set session variable: %s.\n",
mysql_error(server));
rval = 1;
goto report;
}
if((result = mysql_list_dbs(server,NULL)) == NULL){
fprintf(stderr, "Failed to query databases: %s\n",
mysql_error(server));
rval = 1;
goto report;
}
if(mysql_field_count(server) != 1)
{
fprintf(stderr, "SHOW DATABASES returned an unexpected result.\n");
rval = 1;
goto report;
}
while((row = mysql_fetch_row(result)))
{
char* dbname = strdup(row[0]);
printf("Testing database %-32s",dbname);
sprintf(query,"DROP TABLE IF EXISTS %s.t1",dbname);
if(mysql_real_query(server,(const char*)query,strlen(query)))
{
errnum = mysql_errno(server);
if(errnum != ER_DBACCESS_DENIED_ERROR &&
errnum != ER_ACCESS_DENIED_ERROR)
{
fprintf(stderr, "DROP TABLE failed in %s: %d: %s.\n",dbname,mysql_errno(server),mysql_error(server));
}
printf("NO PERMISSION\n");
continue;
}
sprintf(query,"CREATE TABLE %s.t1 (id INT)",dbname);
if(mysql_real_query(server,(const char*)query,strlen(query)))
{
errnum = mysql_errno(server);
if( errnum == ER_TABLEACCESS_DENIED_ERROR)
{
sprintf(query,"DROP TABLE IF EXISTS %s.t1",dbname);
mysql_real_query(server,(const char*)query,strlen(query));
printf("NO PERMISSION\n");
continue;
}
fprintf(stderr, "CREATE TABLE failed in %s: %d: %s.\n",
dbname,mysql_errno(server),mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"INSERT INTO %s.t1 VALUES (1);",dbname);
if(mysql_real_query(server,(const char*)query,strlen(query)))
{
fprintf(stderr, "Query to server failed: %d: %s.\n",
mysql_errno(server),mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"SELECT ID FROM %s.t1 UNION SELECT @test",dbname);
if(mysql_real_query(server,(const char*)query,strlen(query)))
{
fprintf(stderr, "Query to server failed: %d: %s.\n",
mysql_errno(server),mysql_error(server));
rval = 1;
goto report;
}
if((shdres = mysql_store_result(server)) == NULL)
{
fprintf(stderr, "Failed to fetch result set: %d: %s\n",
mysql_errno(server),mysql_error(server));
rval = 1;
goto report;
}
if(mysql_field_count(server) != 1)
{
fprintf(stderr, "Returned field count value did not match the expected value.\n");
rval = 1;
goto report;
}
/**Fetch the two rows, the inserted value and the session variable*/
if(mysql_fetch_row(shdres) == NULL ||
(row = mysql_fetch_row(shdres)) == NULL )
{
fprintf(stderr, "Number of returned rows did not match the expected value.\n");
rval = 1;
goto report;
}
if((lengths = mysql_fetch_lengths(shdres)) == NULL)
{
fprintf(stderr, "Failed to retrieve row lengths: %d: %s.\n",
mysql_errno(server),mysql_error(server));
rval = 1;
goto report;
}
if(lengths[0] != 3 || strcmp(row[0],"123"))
{
rval = 1;
printf(" FAILED\n");
printf( "Reason: Session variable was %s instead of \"123\".\n",row[0]);
}
else
{
printf("OK\n");
}
sprintf(query,"DROP TABLE %s.t1;",dbname);
if(mysql_real_query(server,(const char*)query,strlen(query)))
{
fprintf(stderr, "Query to server failed: %s.\n",mysql_error(server));
rval = 1;
goto report;
}
free(dbname);
mysql_free_result(shdres);
}
mysql_free_result(result);
mysql_close(server);
report:
if(rval){
printf("\nTest failed: Errors during test run.\n");
}
free(host);
free(username);
free(password);
return rval;
}

225
server/modules/routing/schemarouter/test/testschemarouter2.c Normal file
View File

@ -0,0 +1,225 @@
#include <my_config.h>
#include <mysql.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mysqld_error.h>
int main(int argc, char** argv)
{
/**
* This test sets a session variable, creates tables in each of the shards
* and inserts into them a single value while querying the session variable.
* This will show if the value in the session variable in the shard is set and
* if it is the same in all the shards.
*
* The test fails if any of the session variables is not set or differ from the original value.
*/
const unsigned int ports[4] = {
3000,
3001,
3002,
3003
};
const char* srv_id[4] = {
"3000",
"3001",
"3002",
"3003"
};
const char* databases[4] = {
"db0",
"db1",
"db2",
"db3"
};
MYSQL* server;
MYSQL_RES *result,*shdres;
MYSQL_ROW row;
char *host = NULL,*username = NULL, *password = NULL;
char query[2048];
unsigned int port,errnum,optval;
unsigned long *lengths;
int rval, i, j;
if(argc < 5)
{
fprintf(stderr,"Usage: %s <host> <port> <username> <password>\n",argv[0]);
return 1;
}
host = strdup(argv[1]);
port = atoi(argv[2]);
username = strdup(argv[3]);
password = strdup(argv[4]);
rval = 0;
for(i = 0;i<4;i++)
{
if((server = mysql_init(NULL)) == NULL){
fprintf(stderr,"Error : Initialization of MySQL client failed.\n");
rval = 1;
goto report;
}
optval = 1;
mysql_options(server,MYSQL_OPT_CONNECT_TIMEOUT,&optval);
if(mysql_real_connect(server,host,username,password,NULL,ports[i],NULL,0) == NULL){
fprintf(stderr, "Failed to connect to server on port %d: %s\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"STOP SLAVE");
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to stop slave in %d: %s.\n",
ports[i],
mysql_error(server));
}
for(j = 0;j<4;j++)
{
sprintf(query,"DROP DATABASE IF EXISTS %s",databases[j]);
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to drop database in %d: %s.\n",
ports[i],
mysql_error(server));
}
}
mysql_close(server);
}
for(i=0;i<4;i++)
{
if((server = mysql_init(NULL)) == NULL){
fprintf(stderr,"Error : Initialization of MySQL client failed.\n");
rval = 1;
goto report;
}
mysql_options(server,MYSQL_OPT_CONNECT_TIMEOUT,&optval);
if(mysql_real_connect(server,host,username,password,NULL,ports[i],NULL,0) == NULL){
fprintf(stderr, "Failed to connect to server on port %d: %s\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"CREATE DATABASE %s",databases[i]);
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to create table in %d: %s.\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"DROP TABLE IF EXISTS %s.t1",databases[i]);
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to drop table in %d: %s.\n",
ports[i],
mysql_error(server));
}
sprintf(query,"CREATE TABLE %s.t1 (id int)",databases[i]);
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to create table in %d: %s.\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
sprintf(query,"INSERT INTO %s.t1 values (%s)",databases[i],srv_id[i]);
if(mysql_real_query(server,query,strlen(query)))
{
fprintf(stderr, "Failed to insert values in %d: %s.\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
mysql_close(server);
}
for(i = 0;i<4;i++)
{
printf("Testing server on port %d through MaxScale.\n",ports[i]);
if((server = mysql_init(NULL)) == NULL){
fprintf(stderr,"Error : Initialization of MySQL client failed.\n");
rval = 1;
goto report;
}
if(mysql_real_connect(server,host,username,password,databases[i],port,NULL,0) == NULL){
fprintf(stderr, "Failed to connect to port %d using database %s: %s\n",
port,
databases[i],
mysql_error(server));
rval = 1;
goto report;
}
if(mysql_real_query(server,"SELECT id FROM t1",strlen("SELECT id FROM t1")))
{
fprintf(stderr, "Failed to execute query in %d: %s.\n",
ports[i],
mysql_error(server));
rval = 1;
goto report;
}
result = mysql_store_result(server);
while((row = mysql_fetch_row(result)))
{
if(strcmp(row[0],srv_id[i]))
{
fprintf(stderr, "Test failed in %d: Was expecting %s but got %s instead.\n",
ports[i],srv_id[i],row[0]);
rval = 1;
}
}
mysql_free_result(result);
if(i > 0 && mysql_real_query(server,"START SLAVE",strlen("START SLAVE")))
{
fprintf(stderr, "Failed to start slave in %d: %s.\n",
ports[i],
mysql_error(server));
}
mysql_close(server);
}
report:
if(rval){
printf("\nTest failed: Errors during test run.\n");
}
free(host);
free(username);
free(password);
return rval;
}

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

@ -146,12 +146,7 @@ int main(int argc, char** argv)
double test_res = real_test.tv_sec + (real_test.tv_usec / 1000000.0);
result = test_res/base_res;
if(result > ratio){
printf("\nTest failed: Time ratio was %f which exceeded the limit of %f.\n", result, ratio);
rval = 1;
}else{
printf("\nTest passed: Time ratio was %f.\n",result);
}
printf("\nTest passed: Time ratio was %f.\n",result);
}
free(str_baseline);
free(str_test);