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

Merge branch 'develop' into MAX-11

Browse Source 
Conflicts:
	server/include/server.h
	server/modules/monitor/mysql_mon.c
	server/modules/protocol/mysql_client.c
	server/modules/routing/readwritesplit/readwritesplit.c
	utils/skygw_debug.h
This commit is contained in:
VilhoRaatikka
2014-06-07 00:50:08 +03:00
parent 7e6cb7afc2 7c8803f2f2
commit ad744962b2
47 changed files with 2759 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
server/core/Makefile
View File

@ -33,6 +33,7 @@
# 29/06/13 Vilho Raatikka Reverted Query classifier changes because
# gateway needs mysql client lib, not qc.
# 24/07/13 Mark Ridoch Addition of encryption routines
# 30/05/14 Mark Ridoch Filter API added
include ../../build_gateway.inc
@ -56,14 +57,15 @@ LDFLAGS=-rdynamic -L$(LOGPATH) \
SRCS= atomic.c buffer.c spinlock.c gateway.c \
gw_utils.c utils.c dcb.c load_utils.c session.c service.c server.c \
poll.c config.c users.c hashtable.c dbusers.c thread.c gwbitmask.c \
monitor.c adminusers.c secrets.c
monitor.c adminusers.c secrets.c filter.c modutil.c
HDRS= ../include/atomic.h ../include/buffer.h ../include/dcb.h \
../include/gw.h ../modules/include/mysql_client_server_protocol.h \
../include/session.h ../include/spinlock.h ../include/thread.h \
../include/modules.h ../include/poll.h ../include/config.h \
../include/users.h ../include/hashtable.h ../include/gwbitmask.h \
../include/adminusers.h ../include/version.h ../include/maxscale.h
../include/adminusers.h ../include/version.h ../include/maxscale.h \
../include/filter.h modutil.h
OBJ=$(SRCS:.c=.o)

133
server/core/config.c
View File

@ -31,6 +31,9 @@
* 11/03/14 Massimiliano Pinto Added Unix socket support
* 11/05/14 Massimiliano Pinto Added version_string support to service
* 19/05/14 Mark Riddoch Added unique names from section headers
* 29/05/14 Mark Riddoch Addition of filter definition
* 23/05/14 Massimiliano Pinto Added automatic set of maxscale-id: first listening ipv4_raw + port + pid
* 28/05/14 Massimiliano Pinto Added detect_replication_lag parameter
*
* @endverbatim
*/
@ -56,6 +59,7 @@ static char *config_get_value(CONFIG_PARAMETER *, const char *);
static int handle_global_item(const char *, const char *);
static void global_defaults();
static void check_config_objects(CONFIG_CONTEXT *context);
static int config_truth_value(char *str);
static char *config_file = NULL;
static GATEWAY_CONF gateway;
@ -231,7 +235,7 @@ int error_count = 0;
if (obj->element == NULL) /*< if module load failed */
{
LOGIF(LE, (skygw_log_write_flush(
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Reading configuration "
"for router service '%s' failed. "
@ -254,7 +258,7 @@ int error_count = 0;
"max_slave_connections");
if (enable_root_user)
serviceEnableRootUser(obj->element, atoi(enable_root_user));
serviceEnableRootUser(obj->element, config_truth_value(enable_root_user));
if (!auth)
auth = config_get_value(obj->parameters, "auth");
@ -359,6 +363,52 @@ int error_count = 0;
obj->object)));
}
}
else if (!strcmp(type, "filter"))
{
char *module = config_get_value(obj->parameters,
"module");
char *options = config_get_value(obj->parameters,
"options");
if (module)
{
obj->element = filter_alloc(obj->object, module);
}
else
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: Filter '%s' has no module "
"defined defined to load.",
obj->object)));
error_count++;
}
if (obj->element && options)
{
char *s = strtok(options, ",");
while (s)
{
filterAddOption(obj->element, s);
s = strtok(NULL, ",");
}
}
if (obj->element)
{
CONFIG_PARAMETER *params = obj->parameters;
while (params)
{
if (strcmp(params->name, "module")
&& strcmp(params->name,
"options"))
{
filterAddParameter(obj->element,
params->name,
params->value);
}
params = params->next;
}
}
}
obj = obj->next;
}
@ -376,7 +426,8 @@ int error_count = 0;
{
char *servers;
char *roptions;
char *filters = config_get_value(obj->parameters,
"filters");
servers = config_get_value(obj->parameters, "servers");
roptions = config_get_value(obj->parameters,
"router_options");
@ -418,6 +469,10 @@ int error_count = 0;
s = strtok(NULL, ",");
}
}
if (filters)
{
serviceSetFilters(obj->element, filters);
}
}
else if (!strcmp(type, "listener"))
{
@ -426,12 +481,19 @@ int error_count = 0;
char *port;
char *protocol;
char *socket;
struct sockaddr_in serv_addr;
service = config_get_value(obj->parameters, "service");
port = config_get_value(obj->parameters, "port");
address = config_get_value(obj->parameters, "address");
protocol = config_get_value(obj->parameters, "protocol");
socket = config_get_value(obj->parameters, "socket");
/* if id is not set, do it now */
if (gateway.id == 0) {
setipaddress(&serv_addr.sin_addr, (address == NULL) ? "0.0.0.0" : address);
gateway.id = (unsigned long) (serv_addr.sin_addr.s_addr + port + getpid());
}
if (service && socket && protocol) {
CONFIG_CONTEXT *ptr = context;
@ -494,18 +556,46 @@ int error_count = 0;
char *servers;
char *user;
char *passwd;
unsigned long interval = 0;
int replication_heartbeat = 0;
module = config_get_value(obj->parameters, "module");
servers = config_get_value(obj->parameters, "servers");
user = config_get_value(obj->parameters, "user");
passwd = config_get_value(obj->parameters, "passwd");
if (config_get_value(obj->parameters, "monitor_interval")) {
interval = strtoul(config_get_value(obj->parameters, "monitor_interval"), NULL, 10);
}
if (config_get_value(obj->parameters, "detect_replication_lag")) {
replication_heartbeat = atoi(config_get_value(obj->parameters, "detect_replication_lag"));
}
if (module)
{
obj->element = monitor_alloc(obj->object, module);
if (servers && obj->element)
{
char *s = strtok(servers, ",");
char *s;
/* if id is not set, compute it now with pid only */
if (gateway.id == 0) {
gateway.id = getpid();
}
/* add the maxscale-id to monitor data */
monitorSetId(obj->element, gateway.id);
/* set monitor interval */
if (interval > 0)
monitorSetInterval(obj->element, interval);
/* set replication heartbeat */
if(replication_heartbeat == 1)
monitorSetReplicationHeartbeat(obj->element, replication_heartbeat);
/* get the servers to monitor */
s = strtok(servers, ",");
while (s)
{
CONFIG_CONTEXT *obj1 = context;
@ -550,7 +640,8 @@ int error_count = 0;
error_count++;
}
}
else if (strcmp(type, "server") != 0)
else if (strcmp(type, "server") != 0
&& strcmp(type, "filter") != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -747,6 +838,7 @@ global_defaults()
gateway.version_string = strdup(version_string);
else
gateway.version_string = NULL;
gateway.id=0;
}
/**
@ -959,10 +1051,12 @@ SERVER *server;
{
char *servers;
char *roptions;
char *filters;
servers = config_get_value(obj->parameters, "servers");
roptions = config_get_value(obj->parameters,
"router_options");
filters = config_get_value(obj->parameters, "filters");
if (servers && obj->element)
{
char *s = strtok(servers, ",");
@ -996,6 +1090,8 @@ SERVER *server;
s = strtok(NULL, ",");
}
}
if (filters)
serviceSetFilters(obj->element, filters);
}
else if (!strcmp(type, "listener"))
{
@ -1056,7 +1152,8 @@ SERVER *server;
}
}
else if (strcmp(type, "server") != 0 &&
strcmp(type, "monitor") != 0)
strcmp(type, "monitor") != 0 &&
strcmp(type, "filter") != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
@ -1080,6 +1177,7 @@ static char *service_params[] =
"enable_root_user",
"max_slave_connections",
"version_string",
"filters",
NULL
};
@ -1112,6 +1210,8 @@ static char *monitor_params[] =
"servers",
"user",
"passwd",
"monitor_interval",
"detect_replication_lag",
NULL
};
/**
@ -1213,3 +1313,24 @@ bool config_set_qualified_param(
return succp;
}
/**
* Used for boolean settings where values may be 1, yes or true
* to enable a setting or -, no, false to disable a setting.
*
* @param str String to convert to a boolean
* @return Truth value
*/
static int
config_truth_value(char *str)
{
if (strcasecmp(str, "true") == 0 || strcasecmp(str, "on") == 0)
{
return 1;
}
if (strcasecmp(str, "flase") == 0 || strcasecmp(str, "off") == 0)
{
return 0;
}
return atoi(str);
}

316
server/core/filter.c Normal file
View File

@ -0,0 +1,316 @@
/*
* This file is distributed as part of MaxScale from SkySQL. 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 SkySQL Ab 2014
*/
/**
* @file filter.c - A representation of a filter within MaxScale.
*
* @verbatim
* Revision History
*
* Date Who Description
* 29/05/14 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <filter.h>
#include <session.h>
#include <modules.h>
#include <spinlock.h>
#include <skygw_utils.h>
#include <log_manager.h>
extern int lm_enabled_logfiles_bitmask;
static SPINLOCK filter_spin = SPINLOCK_INIT;
static FILTER_DEF *allFilters = NULL;
/**
* Allocate a new filter within MaxScale
*
*
* @param name The filter name
* @param module The module to load
*
* @return The newly created filter or NULL if an error occured
*/
FILTER_DEF *
filter_alloc(char *name, char *module)
{
FILTER_DEF *filter;
if ((filter = (FILTER_DEF *)malloc(sizeof(FILTER_DEF))) == NULL)
return NULL;
filter->name = strdup(name);
filter->module = strdup(module);
filter->filter = NULL;
filter->options = NULL;
filter->obj = NULL;
filter->parameters = NULL;
spinlock_init(&filter->spin);
spinlock_acquire(&filter_spin);
filter->next = allFilters;
allFilters = filter;
spinlock_release(&filter_spin);
return filter;
}
/**
* Deallocate the specified filter
*
* @param server The service to deallocate
* @return Returns true if the server was freed
*/
void
filter_free(FILTER_DEF *filter)
{
FILTER_DEF *ptr;
/* First of all remove from the linked list */
spinlock_acquire(&filter_spin);
if (allFilters == filter)
{
allFilters = filter->next;
}
else
{
ptr = allFilters;
while (ptr && ptr->next != filter)
{
ptr = ptr->next;
}
if (ptr)
ptr->next = filter->next;
}
spinlock_release(&filter_spin);
/* Clean up session and free the memory */
free(filter->name);
free(filter->module);
free(filter);
}
/**
* Find an existing filter using the unique section name in
* configuration file
*
* @param name The filter name
* @return The server or NULL if not found
*/
FILTER_DEF *
filter_find(char *name)
{
FILTER_DEF *filter;
spinlock_acquire(&filter_spin);
filter = allFilters;
while (filter)
{
if (strcmp(filter->name, name) == 0)
break;
filter = filter->next;
}
spinlock_release(&filter_spin);
return filter;
}
/**
* Print all filters to a DCB
*
* Designed to be called within a debugger session in order
* to display all active filters within MaxScale
*/
void
dprintAllFilters(DCB *dcb)
{
FILTER_DEF *ptr;
int i;
spinlock_acquire(&filter_spin);
ptr = allFilters;
while (ptr)
{
dcb_printf(dcb, "Filter %p (%s)\n", ptr, ptr->name);
dcb_printf(dcb, "\tModule: %s\n", ptr->module);
if (ptr->options)
{
dcb_printf(dcb, "\tOptions: ");
for (i = 0; ptr->options && ptr->options[i]; i++)
dcb_printf(dcb, "%s ", ptr->options[i]);
dcb_printf(dcb, "\n");
}
if (ptr->obj && ptr->filter)
ptr->obj->diagnostics(ptr->filter, NULL, dcb);
else
dcb_printf(dcb, "\tModule not loaded.\n");
ptr = ptr->next;
}
spinlock_release(&filter_spin);
}
/**
* Print filter details to a DCB
*
* Designed to be called within a debug CLI in order
* to display all active filters in MaxScale
*/
void
dprintFilter(DCB *dcb, FILTER_DEF *filter)
{
int i;
dcb_printf(dcb, "Filter %p (%s)\n", filter, filter->name);
dcb_printf(dcb, "\tModule: %s\n", filter->module);
if (filter->options)
{
dcb_printf(dcb, "\tOptions: ");
for (i = 0; filter->options && filter->options[i]; i++)
dcb_printf(dcb, "%s ", filter->options[i]);
dcb_printf(dcb, "\n");
}
if (filter->obj && filter->filter)
filter->obj->diagnostics(filter->filter, NULL, dcb);
}
/**
* List all filters in a tabular form to a DCB
*
*/
void
dListFilters(DCB *dcb)
{
FILTER_DEF *ptr;
int i;
spinlock_acquire(&filter_spin);
ptr = allFilters;
if (ptr)
{
dcb_printf(dcb, "%-18s | %-15s | Options\n",
"Filter", "Module");
dcb_printf(dcb, "-------------------------------------------------------------------------------\n");
}
while (ptr)
{
dcb_printf(dcb, "%-18s | %-15s | ",
ptr->name, ptr->module);
for (i = 0; ptr->options && ptr->options[i]; i++)
dcb_printf(dcb, "%s ", ptr->options[i]);
dcb_printf(dcb, "\n");
ptr = ptr->next;
}
spinlock_release(&filter_spin);
}
/**
* Add a router option to a service
*
* @param service The service to add the router option to
* @param option The option string
*/
void
filterAddOption(FILTER_DEF *filter, char *option)
{
int i;
spinlock_acquire(&filter->spin);
if (filter->options == NULL)
{
filter->options = (char **)calloc(2, sizeof(char *));
filter->options[0] = strdup(option);
filter->options[1] = NULL;
}
else
{
for (i = 0; filter->options[i]; i++)
;
filter->options = (char **)realloc(filter->options,
(i + 2) * sizeof(char *));
filter->options[i] = strdup(option);
filter->options[i+1] = NULL;
}
spinlock_release(&filter->spin);
}
/**
* Add a router parameter to a service
*
* @param service The service to add the router option to
* @param name The parameter name
* @param value The parameter value
*/
void
filterAddParameter(FILTER_DEF *filter, char *name, char *value)
{
int i;
spinlock_acquire(&filter->spin);
if (filter->parameters == NULL)
{
filter->parameters = (FILTER_PARAMETER **)calloc(2, sizeof(FILTER_PARAMETER *));
i = 0;
}
else
{
for (i = 0; filter->parameters[i]; i++)
;
filter->parameters = (FILTER_PARAMETER **)realloc(filter->parameters,
(i + 2) * sizeof(FILTER_PARAMETER *));
}
filter->parameters[i] = (FILTER_PARAMETER *)calloc(1, sizeof(FILTER_PARAMETER));
filter->parameters[i]->name = strdup(name);
filter->parameters[i]->value = strdup(value);
filter->parameters[i+1] = NULL;
spinlock_release(&filter->spin);
}
DOWNSTREAM *
filterApply(FILTER_DEF *filter, SESSION *session, DOWNSTREAM *downstream)
{
DOWNSTREAM *me;
if (filter->obj == NULL)
{
/* Filter not yet loaded */
if ((filter->obj = load_module(filter->module,
MODULE_FILTER)) == NULL)
{
return NULL;
}
}
if (filter->filter == NULL)
filter->filter = (filter->obj->createInstance)(filter->options,
filter->parameters);
if ((me = (DOWNSTREAM *)calloc(1, sizeof(DOWNSTREAM))) == NULL)
{
return NULL;
}
me->instance = filter->filter;
me->routeQuery = filter->obj->routeQuery;
me->session = filter->obj->newSession(me->instance, session);
filter->obj->setDownstream(me->instance, me->session, downstream);
return me;
}

100
server/core/load_utils.c
View File

@ -28,6 +28,7 @@
* 14/06/13 Mark Riddoch Updated to add call to ModuleInit if one is
* defined in the loaded module.
* Also updated to call fixed GetModuleObject
* 02/06/14 Mark Riddoch Addition of module info
*
* @endverbatim
*/
@ -38,6 +39,7 @@
#include <string.h>
#include <dlfcn.h>
#include <modules.h>
#include <modinfo.h>
#include <skygw_utils.h>
#include <log_manager.h>
@ -47,10 +49,11 @@ static MODULES *registered = NULL;
static MODULES *find_module(const char *module);
static void register_module(const char *module,
const char *type,
void *dlhandle,
char *version,
void *modobj);
const char *type,
void *dlhandle,
char *version,
void *modobj,
MODULE_INFO *info);
static void unregister_module(const char *module);
char* get_maxscale_home(void)
@ -76,12 +79,13 @@ char* get_maxscale_home(void)
void *
load_module(const char *module, const char *type)
{
char *home, *version;
char fname[MAXPATHLEN];
void *dlhandle, *sym;
char *(*ver)();
void *(*ep)(), *modobj;
MODULES *mod;
char *home, *version;
char fname[MAXPATHLEN];
void *dlhandle, *sym;
char *(*ver)();
void *(*ep)(), *modobj;
MODULES *mod;
MODULE_INFO *mod_info = NULL;
if ((mod = find_module(module)) == NULL)
{
@ -141,6 +145,57 @@ MODULES *mod;
ModuleInit();
}
if ((sym = dlsym(dlhandle, "info")) != NULL)
{
int fatal = 0;
mod_info = sym;
if (strcmp(type, MODULE_PROTOCOL) == 0
&& mod_info->modapi != MODULE_API_PROTOCOL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Module '%s' does not implement "
"the protocol API.\n",
module)));
fatal = 1;
}
if (strcmp(type, MODULE_ROUTER) == 0
&& mod_info->modapi != MODULE_API_ROUTER)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Module '%s' does not implement "
"the router API.\n",
module)));
fatal = 1;
}
if (strcmp(type, MODULE_MONITOR) == 0
&& mod_info->modapi != MODULE_API_MONITOR)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Module '%s' does not implement "
"the monitor API.\n",
module)));
fatal = 1;
}
if (strcmp(type, MODULE_FILTER) == 0
&& mod_info->modapi != MODULE_API_FILTER)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Module '%s' does not implement "
"the filter API.\n",
module)));
fatal = 1;
}
if (fatal)
{
dlclose(dlhandle);
return NULL;
}
}
if ((sym = dlsym(dlhandle, "GetModuleObject")) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
@ -161,7 +216,7 @@ MODULES *mod;
module,
version,
fname)));
register_module(module, type, dlhandle, version, modobj);
register_module(module, type, dlhandle, version, modobj, mod_info);
}
else
{
@ -227,7 +282,7 @@ MODULES *mod = registered;
* @param modobj The module object
*/
static void
register_module(const char *module, const char *type, void *dlhandle, char *version, void *modobj)
register_module(const char *module, const char *type, void *dlhandle, char *version, void *modobj, MODULE_INFO *mod_info)
{
MODULES *mod;
@ -239,6 +294,7 @@ MODULES *mod;
mod->version = strdup(version);
mod->modobj = modobj;
mod->next = registered;
mod->info = mod_info;
registered = mod;
}
@ -303,11 +359,25 @@ dprintAllModules(DCB *dcb)
{
MODULES *ptr = registered;
dcb_printf(dcb, "%-15s | %-11s | Version\n", "Module Name", "Module Type");
dcb_printf(dcb, "-----------------------------------------------------\n");
dcb_printf(dcb, "%-15s | %-11s | Version | API | Status\n", "Module Name", "Module Type");
dcb_printf(dcb, "--------------------------------------------------------------------------\n");
while (ptr)
{
dcb_printf(dcb, "%-15s | %-11s | %s\n", ptr->module, ptr->type, ptr->version);
dcb_printf(dcb, "%-15s | %-11s | %-7s ", ptr->module, ptr->type, ptr->version);
if (ptr->info)
dcb_printf(dcb, "| %d.%d.%d | %s",
ptr->info->api_version.major,
ptr->info->api_version.minor,
ptr->info->api_version.patch,
ptr->info->status == MODULE_IN_DEVELOPMENT
? "In Development"
: (ptr->info->status == MODULE_ALPHA_RELEASE
? "Alpha"
: (ptr->info->status == MODULE_BETA_RELEASE
? "Beta"
: (ptr->info->status == MODULE_GA
? "GA" : "Unknown"))));
dcb_printf(dcb, "\n");
ptr = ptr->next;
}
}

123
server/core/modutil.c Normal file
View File

@ -0,0 +1,123 @@
/*
* This file is distributed as part of MaxScale from SkySQL. 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 SkySQL Ab 2014
*/
/**
* @file modutil.c - Implementation of useful routines for modules
*
* @verbatim
* Revision History
*
* Date Who Description
* 04/06/14 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <buffer.h>
#include <string.h>
/**
* Check if a GWBUF structure is a MySQL COM_QUERY packet
*
* @param buf Buffer to check
* @return True if GWBUF is a COM_QUERY packet
*/
int
modutil_is_SQL(GWBUF *buf)
{
unsigned char *ptr;
if (GWBUF_LENGTH(buf) < 5)
return 0;
ptr = GWBUF_DATA(buf);
return ptr[4] == 0x03; // COM_QUERY
}
/**
* Extract the SQL portion of a COM_QUERY packet
*
* NB This sets *sql to point into the packet and does not
* allocate any new storage. The string pointed to by *sql is
* not NULL terminated.
*
* This routine is very simplistic and does not deal with SQL text
* that spans multiple buffers.
*
* @param buf The packet buffer
* @param sql Pointer that is set to point at the SQL data
* @param length Length of the SQL data
* @return True if the packet is a COM_QUERY packet
*/
int
modutil_extract_SQL(GWBUF *buf, char **sql, int *length)
{
char *ptr;
if (!modutil_is_SQL(buf))
return 0;
ptr = GWBUF_DATA(buf);
*length = *ptr++;
*length += (*ptr++ << 8);
*length += (*ptr++ << 8);
ptr += 2; // Skip sequence id and COM_QUERY byte
*length = *length - 1;
*sql = ptr;
return 1;
}
GWBUF *
modutil_replace_SQL(GWBUF *orig, char *sql)
{
char *ptr;
int length, newlength;
GWBUF *addition;
if (!modutil_is_SQL(orig))
return NULL;
ptr = GWBUF_DATA(orig);
length = *ptr++;
length += (*ptr++ << 8);
length += (*ptr++ << 8);
ptr += 2; // Skip sequence id and COM_QUERY byte
newlength = strlen(sql);
if (length - 1 == newlength)
{
/* New SQL is the same length as old */
memcpy(ptr, sql, newlength);
}
else if (length - 1 > newlength)
{
/* New SQL is shorter */
memcpy(ptr, sql, newlength);
GWBUF_RTRIM(orig, (length - 1) - newlength);
}
else
{
memcpy(ptr, sql, length - 1);
addition = gwbuf_alloc(newlength - (length - 1));
memcpy(GWBUF_DATA(addition), &sql[length - 1], newlength - (length - 1));
ptr = GWBUF_DATA(orig);
*ptr++ = (newlength + 1) & 0xff;
*ptr++ = ((newlength + 1) >> 8) & 0xff;
*ptr++ = ((newlength + 1) >> 16) & 0xff;
orig->next = addition;
}
return orig;
}

50
server/core/monitor.c
View File

@ -22,8 +22,10 @@
* @verbatim
* Revision History
*
* Date Who Description
* 08/07/13 Mark Riddoch Initial implementation
* Date Who Description
* 08/07/13 Mark Riddoch Initial implementation
* 23/05/14 Massimiliano Pinto Addition of monitor_interval parameter
* and monitor id
*
* @endverbatim
*/
@ -220,3 +222,47 @@ MONITOR *ptr;
spinlock_release(&monLock);
return ptr;
}
/**
* Set the id of the monitor.
*
* @param mon The monitor instance
* @param id The id for the monitor
*/
void
monitorSetId(MONITOR *mon, unsigned long id)
{
if (mon->module->defaultId != NULL) {
mon->module->defaultId(mon->handle, id);
}
}
/**
* Set the monitor sampling interval.
*
* @param mon The monitor instance
* @param interval The sampling interval in milliseconds
*/
void
monitorSetInterval (MONITOR *mon, unsigned long interval)
{
if (mon->module->setInterval != NULL) {
mon->module->setInterval(mon->handle, interval);
}
}
/**
* Enable Replication Heartbeat support in monitor.
*
* @param mon The monitor instance
* @param interval The sampling interval in milliseconds
*/
void
monitorSetReplicationHeartbeat(MONITOR *mon, int replication_heartbeat)
{
if (mon->module->replicationHeartbeat != NULL) {
mon->module->replicationHeartbeat(mon->handle, replication_heartbeat);
}
}

21
server/core/server.c
View File

@ -27,6 +27,7 @@
* 17/05/14 Mark Riddoch Addition of unique_name
* 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
*
* @endverbatim
*/
@ -73,6 +74,8 @@ SERVER *server;
server->unique_name = NULL;
server->server_string = NULL;
server->node_id = -1;
server->rlag = -1;
server->node_ts = 0;
spinlock_acquire(&server_spin);
server->next = allServers;
@ -247,6 +250,14 @@ 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)) {
if (ptr->rlag >= 0) {
dcb_printf(dcb, "\tSlave delay:\t\t%d\n", ptr->rlag);
}
}
if (ptr->node_ts > 0) {
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%lu\n", ptr->node_ts);
}
dcb_printf(dcb, "\tNumber of connections: %d\n", ptr->stats.n_connections);
dcb_printf(dcb, "\tCurrent no. of conns: %d\n", ptr->stats.n_current);
ptr = ptr->next;
@ -275,6 +286,14 @@ 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)) {
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, "\tNumber of connections: %d\n", server->stats.n_connections);
dcb_printf(dcb, "\tCurrent no. of conns: %d\n", server->stats.n_current);
}
@ -325,6 +344,8 @@ char *status = NULL;
if ((status = (char *)malloc(200)) == NULL)
return NULL;
status[0] = 0;
if (server->status & SERVER_MAINT)
strcat(status, "Maintenance, ");
if (server->status & SERVER_MASTER)
strcat(status, "Master, ");
if (server->status & SERVER_SLAVE)

96
server/core/service.c
View File

@ -30,6 +30,7 @@
* 28/02/14 Massimiliano Pinto users_alloc moved from service_alloc to serviceStartPort (generic hashable for services)
* 07/05/14 Massimiliano Pinto Added: version_string initialized to NULL
* 23/05/14 Mark Riddoch Addition of service validation call
* 29/05/14 Mark Riddoch Filter API implementation
*
* @endverbatim
*/
@ -46,6 +47,7 @@
#include <modules.h>
#include <dcb.h>
#include <users.h>
#include <filter.h>
#include <dbusers.h>
#include <poll.h>
#include <skygw_utils.h>
@ -110,6 +112,8 @@ SERVICE *service;
service->databases = NULL;
service->svc_config_param = NULL;
service->svc_config_version = 0;
service->filters = NULL;
service->n_filters = 0;
spinlock_init(&service->spin);
spinlock_init(&service->users_table_spin);
memset(&service->rate_limit, 0, sizeof(SERVICE_REFRESH_RATE));
@ -538,10 +542,13 @@ serviceClearRouterOptions(SERVICE *service)
int i;
spinlock_acquire(&service->spin);
for (i = 0; service->routerOptions[i]; i++)
free(service->routerOptions[i]);
free(service->routerOptions);
service->routerOptions = NULL;
if (service->routerOptions != NULL)
{
for (i = 0; service->routerOptions[i]; i++)
free(service->routerOptions[i]);
free(service->routerOptions);
service->routerOptions = NULL;
}
spinlock_release(&service->spin);
}
/**
@ -608,6 +615,63 @@ serviceEnableRootUser(SERVICE *service, int action)
return 1;
}
/**
* Trim whitespace from the from an rear of a string
*
* @param str String to trim
* @return Trimmed string, chanesg are done in situ
*/
static char *
trim(char *str)
{
char *ptr;
while (isspace(*str))
str++;
/* Point to last character of the string */
ptr = str + strlen(str) - 1;
while (ptr > str && isspace(*ptr))
*ptr-- = 0;
return str;
}
/**
* Set the filters used by the service
*
* @param service The service itself
* @param filters ASCII string of filters to use
*/
void
serviceSetFilters(SERVICE *service, char *filters)
{
FILTER_DEF **flist;
char *ptr, *brkt;
int n = 0;
flist = (FILTER_DEF *)malloc(sizeof(FILTER_DEF *));
ptr = strtok_r(filters, "|", &brkt);
while (ptr)
{
n++;
flist = (FILTER_DEF *)realloc(flist, (n + 1) * sizeof(FILTER_DEF *));
if ((flist[n-1] = filter_find(trim(ptr))) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Unable to find filter '%s' for service '%s'\n",
trim(ptr), service->name
)));
}
flist[n] = NULL;
ptr = strtok_r(NULL, "|", &brkt);
}
service->filters = flist;
service->n_filters = n;
}
/**
* Return a named service
*
@ -638,6 +702,7 @@ void
printService(SERVICE *service)
{
SERVER *ptr = service->databases;
int i;
printf("Service %p\n", service);
printf("\tService: %s\n", service->name);
@ -649,6 +714,16 @@ SERVER *ptr = service->databases;
printf("\t\t%s:%d Protocol: %s\n", ptr->name, ptr->port, ptr->protocol);
ptr = ptr->nextdb;
}
if (service->n_filters)
{
printf("\tFilter chain: ");
for (i = 0; i < service->n_filters; i++)
{
printf("%s %s ", service->filters[i]->name,
i + 1 < service->n_filters ? "|" : "");
}
printf("\n");
}
printf("\tUsers data: %p\n", service->users);
printf("\tTotal connections: %d\n", service->stats.n_sessions);
printf("\tCurrently connected: %d\n", service->stats.n_current);
@ -705,6 +780,7 @@ SERVICE *ptr;
void dprintService(DCB *dcb, SERVICE *service)
{
SERVER *server = service->databases;
int i;
dcb_printf(dcb, "Service %p\n", service);
dcb_printf(dcb, "\tService: %s\n", service->name);
@ -714,6 +790,16 @@ SERVER *server = service->databases;
service->router->diagnostics(service->router_instance, dcb);
dcb_printf(dcb, "\tStarted: %s",
asctime(localtime(&service->stats.started)));
if (service->n_filters)
{
dcb_printf(dcb, "\tFilter chain: ");
for (i = 0; i < service->n_filters; i++)
{
dcb_printf(dcb, "%s %s ", service->filters[i]->name,
i + 1 < service->n_filters ? "|" : "");
}
dcb_printf(dcb, "\n");
}
dcb_printf(dcb, "\tBackend databases\n");
while (server)
{
@ -996,4 +1082,4 @@ char* service_get_name(
SERVICE* svc)
{
return svc->name;
}
}

124
server/core/session.c
View File

@ -25,6 +25,7 @@
* Date Who Description
* 17/06/13 Mark Riddoch Initial implementation
* 02/09/13 Massimiliano Pinto Added session refcounter
* 29/05/14 Mark Riddoch Addition of filter mechanism
*
* @endverbatim
*/
@ -47,6 +48,9 @@ extern int lm_enabled_logfiles_bitmask;
static SPINLOCK session_spin = SPINLOCK_INIT;
static SESSION *allSessions = NULL;
static int session_setup_filters(SESSION *session);
/**
* Allocate a new session for a new client of the specified service.
*
@ -90,6 +94,7 @@ session_alloc(SERVICE *service, DCB *client_dcb)
spinlock_acquire(&session->ses_lock);
session->service = service;
session->client = client_dcb;
session->n_filters = 0;
memset(&session->stats, 0, sizeof(SESSION_STATS));
session->stats.connect = time(0);
session->state = SESSION_STATE_ALLOC;
@ -132,7 +137,7 @@ session_alloc(SERVICE *service, DCB *client_dcb)
/**
* Inform other threads that session is closing.
*/
session->state == SESSION_STATE_STOPPING;
session->state = SESSION_STATE_STOPPING;
/*<
* Decrease refcount, set dcb's session pointer NULL
* and set session pointer to NULL.
@ -147,7 +152,45 @@ session_alloc(SERVICE *service, DCB *client_dcb)
goto return_session;
}
/*
* Pending filter chain being setup set the head of the chain to
* be the router. As filters are inserted the current head will
* be pushed to the filter and the head updated.
*
* NB This dictates that filters are created starting at the end
* of the chain nearest the router working back to the client
* protocol end of the chain.
*/
session->head.instance = service->router_instance;
session->head.session = session->router_session;
session->head.routeQuery = service->router->routeQuery;
if (service->n_filters > 0)
{
if (!session_setup_filters(session))
{
/**
* Inform other threads that session is closing.
*/
session->state = SESSION_STATE_STOPPING;
/*<
* Decrease refcount, set dcb's session pointer NULL
* and set session pointer to NULL.
*/
session_free(session);
client_dcb->session = NULL;
session = NULL;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to create %s session.",
service->name)));
goto return_session;
}
}
}
spinlock_acquire(&session_spin);
session->state = SESSION_STATE_ROUTER_READY;
session->next = allSessions;
@ -227,6 +270,7 @@ bool session_free(
bool succp = false;
SESSION *ptr;
int nlink;
int i;
CHK_SESSION(session);
@ -262,10 +306,23 @@ bool session_free(
/* Free router_session and session */
if (session->router_session) {
session->service->router->closeSession(
session->service->router_instance,
session->router_session);
session->service->router->freeSession(
session->service->router_instance,
session->router_session);
}
if (session->n_filters)
{
for (i = 0; i < session->n_filters; i++)
{
session->filters[i].filter->obj->freeSession(
session->filters[i].instance,
session->filters[i].session);
}
free(session->filters);
}
free(session);
succp = true;
@ -440,6 +497,8 @@ SESSION *ptr;
void
dprintSession(DCB *dcb, SESSION *ptr)
{
int i;
dcb_printf(dcb, "Session %p\n", ptr);
dcb_printf(dcb, "\tState: %s\n", session_state(ptr->state));
dcb_printf(dcb, "\tService: %s (%p)\n", ptr->service->name, ptr->service);
@ -447,6 +506,18 @@ dprintSession(DCB *dcb, SESSION *ptr)
if (ptr->client && ptr->client->remote)
dcb_printf(dcb, "\tClient Address: %s\n", ptr->client->remote);
dcb_printf(dcb, "\tConnected: %s", asctime(localtime(&ptr->stats.connect)));
if (ptr->n_filters)
{
for (i = 0; i < ptr->n_filters; i++)
{
dcb_printf(dcb, "\tFilter: %s\n",
ptr->filters[i].filter->name);
ptr->filters[i].filter->obj->diagnostics(
ptr->filters[i].instance,
ptr->filters[i].session,
dcb);
}
}
}
/**
@ -520,3 +591,54 @@ SESSION* get_session_by_router_ses(
}
return ses;
}
/**
* Create the filter chain for this session.
*
* Filters must be setup in reverse order, starting with the last
* filter in the chain and working back towards the client connection
* Each filter is passed the current session head of the filter chain
* this head becomes the destination for the filter. The newly created
* filter becomes the new head of the filter chain.
*
* @param session The session that requires the chain
* @return 0 if filter creation fails
*/
static int
session_setup_filters(SESSION *session)
{
SERVICE *service = session->service;
DOWNSTREAM *head;
int i;
if ((session->filters = calloc(service->n_filters,
sizeof(SESSION_FILTER))) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Insufficient memory to allocate session filter "
"tracking.\n")));
return 0;
}
session->n_filters = service->n_filters;
for (i = service->n_filters - 1; i >= 0; i--)
{
if ((head = filterApply(service->filters[i], session,
&session->head)) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Failed to create filter '%s' for service '%s'.\n",
service->filters[i]->name,
service->name)));
return 0;
}
session->filters[i].filter = service->filters[i];
session->filters[i].session = head->session;
session->filters[i].instance = head->instance;
session->head = *head;
}
return 1;
}