hcq/MaxScale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7e6cb7afc2c3bcc229f39e97fe7d83147a5e40ef
MaxScale/server/core/service.c
VilhoRaatikka 7e6cb7afc2 Snapshot of failure tolerance changes. 
Added a lot of logging to error, trace and message logs which should help the user to handle errors which can't be automatically resolved, like attempt to use nonexisting database.
2014-06-06 23:32:04 +03:00

999 lines
24 KiB
C
Raw Blame History

/*
* This file is distributed as part of the SkySQL Gateway. 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 2013
*/
/**
* @file service.c - A representation of the service within the gateway.
*
* @verbatim
* Revision History
*
* Date Who Description
* 18/06/13 Mark Riddoch Initial implementation
* 24/06/13 Massimiliano Pinto Added: Loading users from mysql backend in serviceStart
* 06/02/14 Massimiliano Pinto Added: serviceEnableRootUser routine
* 25/02/14 Massimiliano Pinto Added: service refresh limit feature
* 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
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <ctype.h>
#include <errno.h>
#include <session.h>
#include <service.h>
#include <server.h>
#include <router.h>
#include <spinlock.h>
#include <modules.h>
#include <dcb.h>
#include <users.h>
#include <dbusers.h>
#include <poll.h>
#include <skygw_utils.h>
#include <log_manager.h>
extern int lm_enabled_logfiles_bitmask;
static SPINLOCK service_spin = SPINLOCK_INIT;
static SERVICE *allServices = NULL;
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param);
/**
* Allocate a new service for the gateway to support
*
*
* @param servname The service name
* @param router Name of the router module this service uses
*
* @return The newly created service or NULL if an error occured
*/
SERVICE *
service_alloc(char *servname, char *router)
{
SERVICE *service;
if ((service = (SERVICE *)malloc(sizeof(SERVICE))) == NULL)
return NULL;
if ((service->router = load_module(router, MODULE_ROUTER)) == NULL)
{
char* home = get_maxscale_home();
char* ldpath = getenv("LD_LIBRARY_PATH");
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load %s module \"%s\".\n\t\t\t"
" Ensure that lib%s.so exists in one of the "
"following directories :\n\t\t\t "
"- %s/modules\n\t\t\t - %s",
MODULE_ROUTER,
router,
router,
home,
ldpath)));
free(service);
return NULL;
}
service->name = strdup(servname);
service->routerModule = strdup(router);
service->version_string = NULL;
memset(&service->stats, 0, sizeof(SERVICE_STATS));
service->ports = NULL;
service->stats.started = time(0);
service->state = SERVICE_STATE_ALLOC;
service->credentials.name = NULL;
service->credentials.authdata = NULL;
service->enable_root = 0;
service->routerOptions = NULL;
service->databases = NULL;
service->svc_config_param = NULL;
service->svc_config_version = 0;
spinlock_init(&service->spin);
spinlock_init(&service->users_table_spin);
memset(&service->rate_limit, 0, sizeof(SERVICE_REFRESH_RATE));
spinlock_acquire(&service_spin);
service->next = allServices;
allServices = service;
spinlock_release(&service_spin);
return service;
}
/**
* Check to see if a service pointer is valid
*
* @param service The poitner to check
* @return 1 if the service is in the list of all services
*/
int
service_isvalid(SERVICE *service)
{
SERVICE *ptr;
int rval = 0;
spinlock_acquire(&service_spin);
ptr = allServices;
while (ptr)
{
if (ptr == service)
{
rval = 1;
break;
}
ptr = ptr->next;
}
spinlock_release(&service_spin);
return rval;
}
/**
* Start an individual port/protocol pair
*
* @param service The service
* @param port The port to start
* @return The number of listeners started
*/
static int
serviceStartPort(SERVICE *service, SERV_PROTOCOL *port)
{
int listeners = 0;
char config_bind[40];
GWPROTOCOL *funcs;
port->listener = dcb_alloc(DCB_ROLE_SERVICE_LISTENER);
if (port->listener == NULL)
{
return 0;
}
if (strcmp(port->protocol, "MySQLClient") == 0) {
int loaded;
/* Allocate specific data for MySQL users */
service->users = mysql_users_alloc();
loaded = load_mysql_users(service);
/* At service start last update is set to USERS_REFRESH_TIME seconds earlier.
* This way MaxScale could try reloading users' just after startup
*/
service->rate_limit.last=time(NULL) - USERS_REFRESH_TIME;
service->rate_limit.nloads=1;
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Loaded %d MySQL Users.",
loaded)));
} else {
/* Generic users table */
service->users = users_alloc();
}
if ((funcs =
(GWPROTOCOL *)load_module(port->protocol, MODULE_PROTOCOL)) == NULL)
{
dcb_free(port->listener);
port->listener = NULL;
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to load protocol module %s. Listener "
"for service %s not started.",
port->protocol,
service->name)));
return 0;
}
memcpy(&(port->listener->func), funcs, sizeof(GWPROTOCOL));
port->listener->session = NULL;
if (port->address)
sprintf(config_bind, "%s:%d", port->address, port->port);
else
sprintf(config_bind, "0.0.0.0:%d", port->port);
if (port->listener->func.listen(port->listener, config_bind)) {
port->listener->session = session_alloc(service, port->listener);
if (port->listener->session != NULL) {
port->listener->session->state = SESSION_STATE_LISTENER;
listeners += 1;
} else {
dcb_close(port->listener);
}
} else {
dcb_close(port->listener);
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Unable to start to listen port %d for %s %s.",
port->port,
port->protocol,
service->name)));
}
return listeners;
}
/**
* Start a service
*
* This function loads the protocol modules for each port on which the
* service listens and starts the listener on that port
*
* Also create the router_instance for the service.
*
* @param service The Service that should be started
* @return Returns the number of listeners created
*/
int
serviceStart(SERVICE *service)
{
SERV_PROTOCOL *port;
int listeners = 0;
service->router_instance = service->router->createInstance(service,
service->routerOptions);
port = service->ports;
while (port)
{
listeners += serviceStartPort(service, port);
port = port->next;
}
if (listeners)
service->stats.started = time(0);
return listeners;
}
/**
* Start an individual listener
*
* @param service The service to start the listener for
* @param protocol The name of the protocol
* @param port The port number
*/
void
serviceStartProtocol(SERVICE *service, char *protocol, int port)
{
SERV_PROTOCOL *ptr;
ptr = service->ports;
while (ptr)
{
if (strcmp(ptr->protocol, protocol) == 0 && ptr->port == port)
serviceStartPort(service, ptr);
ptr = ptr->next;
}
}
/**
* Start all the services
*
* @return Return the number of services started
*/
int
serviceStartAll()
{
SERVICE *ptr;
int n = 0;
ptr = allServices;
while (ptr)
{
n += serviceStart(ptr);
ptr = ptr->next;
}
return n;
}
/**
* Stop a service
*
* This function stops the listener for the service
*
* @param service The Service that should be stopped
* @return Returns the number of listeners restarted
*/
int
serviceStop(SERVICE *service)
{
SERV_PROTOCOL *port;
int listeners = 0;
port = service->ports;
while (port)
{
poll_remove_dcb(port->listener);
port->listener->session->state = SESSION_STATE_LISTENER_STOPPED;
listeners++;
port = port->next;
}
return listeners;
}
/**
* Restart a service
*
* This function stops the listener for the service
*
* @param service The Service that should be restarted
* @return Returns the number of listeners restarted
*/
int
serviceRestart(SERVICE *service)
{
SERV_PROTOCOL *port;
int listeners = 0;
port = service->ports;
while (port)
{
if (poll_add_dcb(port->listener) == 0) {
port->listener->session->state = SESSION_STATE_LISTENER;
listeners++;
}
port = port->next;
}
return listeners;
}
/**
* Deallocate the specified service
*
* @param service The service to deallocate
* @return Returns true if the service was freed
*/
int
service_free(SERVICE *service)
{
SERVICE *ptr;
if (service->stats.n_current)
return 0;
/* First of all remove from the linked list */
spinlock_acquire(&service_spin);
if (allServices == service)
{
allServices = service->next;
}
else
{
ptr = allServices;
while (ptr && ptr->next != service)
{
ptr = ptr->next;
}
if (ptr)
ptr->next = service->next;
}
spinlock_release(&service_spin);
/* Clean up session and free the memory */
free(service->name);
free(service->routerModule);
if (service->credentials.name)
free(service->credentials.name);
if (service->credentials.authdata)
free(service->credentials.authdata);
free(service);
return 1;
}
/**
* Add a protocol/port pair to the service
*
* @param service The service
* @param protocol The name of the protocol module
* @param address The address to listen with
* @param port The port to listen on
* @return TRUE if the protocol/port could be added
*/
int
serviceAddProtocol(SERVICE *service, char *protocol, char *address, unsigned short port)
{
SERV_PROTOCOL *proto;
if ((proto = (SERV_PROTOCOL *)malloc(sizeof(SERV_PROTOCOL))) == NULL)
{
return 0;
}
proto->protocol = strdup(protocol);
if (address)
proto->address = strdup(address);
else
proto->address = NULL;
proto->port = port;
spinlock_acquire(&service->spin);
proto->next = service->ports;
service->ports = proto;
spinlock_release(&service->spin);
return 1;
}
/**
* Check if a protocol/port pair is part of the service
*
* @param service The service
* @param protocol The name of the protocol module
* @param port The port to listen on
* @return TRUE if the protocol/port is already part of the service
*/
int
serviceHasProtocol(SERVICE *service, char *protocol, unsigned short port)
{
SERV_PROTOCOL *proto;
spinlock_acquire(&service->spin);
proto = service->ports;
while (proto)
{
if (strcmp(proto->protocol, protocol) == 0 && proto->port == port)
break;
proto = proto->next;
}
spinlock_release(&service->spin);
return proto != NULL;
}
/**
* Add a backend database server to a service
*
* @param service The service to add the server to
* @param server The server to add
*/
void
serviceAddBackend(SERVICE *service, SERVER *server)
{
spinlock_acquire(&service->spin);
server->nextdb = service->databases;
service->databases = server;
spinlock_release(&service->spin);
}
/**
* Test if a server is part of a service
*
* @param service The service to add the server to
* @param server The server to add
* @return Non-zero if the server is already part of the service
*/
int
serviceHasBackend(SERVICE *service, SERVER *server)
{
SERVER *ptr;
spinlock_acquire(&service->spin);
ptr = service->databases;
while (ptr && ptr != server)
ptr = ptr->nextdb;
spinlock_release(&service->spin);
return ptr != NULL;
}
/**
* Add a router option to a service
*
* @param service The service to add the router option to
* @param option The option string
*/
void
serviceAddRouterOption(SERVICE *service, char *option)
{
int i;
spinlock_acquire(&service->spin);
if (service->routerOptions == NULL)
{
service->routerOptions = (char **)calloc(2, sizeof(char *));
service->routerOptions[0] = strdup(option);
service->routerOptions[1] = NULL;
}
else
{
for (i = 0; service->routerOptions[i]; i++)
;
service->routerOptions = (char **)realloc(service->routerOptions,
(i + 2) * sizeof(char *));
service->routerOptions[i] = strdup(option);
service->routerOptions[i+1] = NULL;
}
spinlock_release(&service->spin);
}
/**
* Remove the router options for the service
*
* @param service The service to remove the options from
*/
void
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;
spinlock_release(&service->spin);
}
/**
* Set the service user that is used to log in to the backebd servers
* associated with this service.
*
* @param service The service we are setting the data for
* @param user The user name to use for connections
* @param auth The authentication data we need, e.g. MySQL SHA1 password
* @return 0 on failure
*/
int
serviceSetUser(SERVICE *service, char *user, char *auth)
{
if (service->credentials.name)
free(service->credentials.name);
if (service->credentials.authdata)
free(service->credentials.authdata);
service->credentials.name = strdup(user);
service->credentials.authdata = strdup(auth);
if (service->credentials.name == NULL || service->credentials.authdata == NULL)
return 0;
return 1;
}
/**
* Get the service user that is used to log in to the backebd servers
* associated with this service.
*
* @param service The service we are setting the data for
* @param user The user name to use for connections
* @param auth The authentication data we need, e.g. MySQL SHA1 password
* @return 0 on failure
*/
int
serviceGetUser(SERVICE *service, char **user, char **auth)
{
if (service->credentials.name == NULL || service->credentials.authdata == NULL)
return 0;
*user = service->credentials.name;
*auth = service->credentials.authdata;
return 1;
}
/**
* Enable/Disable root user for this service
* associated with this service.
*
* @param service The service we are setting the data for
* @param action 1 for root enable, 0 for disable access
* @return 0 on failure
*/
int
serviceEnableRootUser(SERVICE *service, int action)
{
if (action != 0 && action != 1)
return 0;
service->enable_root = action;
return 1;
}
/**
* Return a named service
*
* @param servname The name of the service to find
* @return The service or NULL if not found
*/
SERVICE *
service_find(char *servname)
{
SERVICE *service;
spinlock_acquire(&service_spin);
service = allServices;
while (service && strcmp(service->name, servname) != 0)
service = service->next;
spinlock_release(&service_spin);
return service;
}
/**
* Print details of an individual service
*
* @param service Service to print
*/
void
printService(SERVICE *service)
{
SERVER *ptr = service->databases;
printf("Service %p\n", service);
printf("\tService: %s\n", service->name);
printf("\tRouter: %s (%p)\n", service->routerModule, service->router);
printf("\tStarted: %s", asctime(localtime(&service->stats.started)));
printf("\tBackend databases\n");
while (ptr)
{
printf("\t\t%s:%d Protocol: %s\n", ptr->name, ptr->port, ptr->protocol);
ptr = ptr->nextdb;
}
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);
}
/**
* Print all services
*
* Designed to be called within a debugger session in order
* to display all active services within the gateway
*/
void
printAllServices()
{
SERVICE *ptr;
spinlock_acquire(&service_spin);
ptr = allServices;
while (ptr)
{
printService(ptr);
ptr = ptr->next;
}
spinlock_release(&service_spin);
}
/**
* Print all services to a DCB
*
* Designed to be called within a CLI command in order
* to display all active services within the gateway
*/
void
dprintAllServices(DCB *dcb)
{
SERVICE *ptr;
spinlock_acquire(&service_spin);
ptr = allServices;
while (ptr)
{
dprintService(dcb, ptr);
ptr = ptr->next;
}
spinlock_release(&service_spin);
}
/**
* Print details of a single service.
*
* @param dcb DCB to print data to
* @param service The service to print
*/
void dprintService(DCB *dcb, SERVICE *service)
{
SERVER *server = service->databases;
dcb_printf(dcb, "Service %p\n", service);
dcb_printf(dcb, "\tService: %s\n", service->name);
dcb_printf(dcb, "\tRouter: %s (%p)\n", service->routerModule,
service->router);
if (service->router)
service->router->diagnostics(service->router_instance, dcb);
dcb_printf(dcb, "\tStarted: %s",
asctime(localtime(&service->stats.started)));
dcb_printf(dcb, "\tBackend databases\n");
while (server)
{
dcb_printf(dcb, "\t\t%s:%d Protocol: %s\n", server->name, server->port,
server->protocol);
server = server->nextdb;
}
dcb_printf(dcb, "\tUsers data: %p\n", service->users);
dcb_printf(dcb, "\tTotal connections: %d\n", service->stats.n_sessions);
dcb_printf(dcb, "\tCurrently connected: %d\n", service->stats.n_current);
}
/**
* List the defined services in a tabular format.
*
* @param dcb DCB to print the service list to.
*/
void
dListServices(DCB *dcb)
{
SERVICE *ptr;
spinlock_acquire(&service_spin);
ptr = allServices;
if (ptr)
{
dcb_printf(dcb, "%-25s | %-20s | #Users | Total Sessions\n",
"Service Name", "Router Module");
dcb_printf(dcb, "--------------------------------------------------------------------------\n");
}
while (ptr)
{
dcb_printf(dcb, "%-25s | %-20s | %6d | %5d\n",
ptr->name, ptr->routerModule,
ptr->stats.n_current, ptr->stats.n_sessions);
ptr = ptr->next;
}
spinlock_release(&service_spin);
}
/**
* List the defined listeners in a tabular format.
*
* @param dcb DCB to print the service list to.
*/
void
dListListeners(DCB *dcb)
{
SERVICE *ptr;
SERV_PROTOCOL *lptr;
spinlock_acquire(&service_spin);
ptr = allServices;
if (ptr)
{
dcb_printf(dcb, "%-20s | %-18s | %-15s | Port | State\n",
"Service Name", "Protocol Module", "Address");
dcb_printf(dcb, "---------------------------------------------------------------------------\n");
}
while (ptr)
{
lptr = ptr->ports;
while (lptr)
{
dcb_printf(dcb, "%-20s | %-18s | %-15s | %5d | %s\n",
ptr->name, lptr->protocol,
(lptr != NULL) ? lptr->address : "*",
lptr->port,
(lptr->listener->session->state == SESSION_STATE_LISTENER_STOPPED) ? "Stopped" : "Running"
);
lptr = lptr->next;
}
ptr = ptr->next;
}
spinlock_release(&service_spin);
}
/**
* Update the definition of a service
*
* @param service The service to update
* @param router The router module to use
* @param user The user to use to extract information from the database
* @param auth The password for the user above
*/
void
service_update(SERVICE *service, char *router, char *user, char *auth)
{
void *router_obj;
if (!strcmp(service->routerModule, router))
{
if ((router_obj = load_module(router, MODULE_ROUTER)) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Failed to update router "
"for service %s to %s.",
service->name,
router)));
}
else
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Update router for service %s to %s.",
service->name,
router)));
free(service->routerModule);
service->routerModule = strdup(router);
service->router = router_obj;
}
}
if (user &&
(strcmp(service->credentials.name, user) != 0 ||
strcmp(service->credentials.authdata, auth) != 0))
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Update credentials for service %s.",
service->name)));
serviceSetUser(service, user, auth);
}
}
int service_refresh_users(SERVICE *service) {
int ret = 1;
/* check for another running getUsers request */
if (! spinlock_acquire_nowait(&service->users_table_spin)) {
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [service_refresh_users] failed to get get lock for loading new users' table: another thread is loading users",
pthread_self())));
return 1;
}
/* check if refresh rate limit has exceeded */
if ( (time(NULL) < (service->rate_limit.last + USERS_REFRESH_TIME)) || (service->rate_limit.nloads > USERS_REFRESH_MAX_PER_TIME)) {
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"%lu [service_refresh_users] refresh rate limit exceeded loading new users' table",
pthread_self())));
spinlock_release(&service->users_table_spin);
return 1;
}
service->rate_limit.nloads++;
/* update time and counter */
if (service->rate_limit.nloads > USERS_REFRESH_MAX_PER_TIME) {
service->rate_limit.nloads = 1;
service->rate_limit.last = time(NULL);
}
ret = replace_mysql_users(service);
/* remove lock */
spinlock_release(&service->users_table_spin);
if (ret >= 0)
return 0;
else
return 1;
}
bool service_set_slave_conn_limit (
SERVICE* service,
CONFIG_PARAMETER* param,
char* valstr,
count_spec_t count_spec)
{
char* p;
int valint;
bool percent = false;
bool succp;
/**
* Find out whether the value is numeric and ends with '%' or '\0'
*/
p = valstr;
while(isdigit(*p)) p++;
errno = 0;
if (p == valstr || (*p != '%' && *p != '\0'))
{
succp = false;
}
else if (*p == '%')
{
if (*(p+1) == '\0')
{
*p = '\0';
valint = (int) strtol(valstr, (char **)NULL, 10);
if (valint == 0 && errno != 0)
{
succp = false;
}
else
{
succp = true;
config_set_qualified_param(param, (void *)&valint, PERCENT_TYPE);
}
}
else
{
succp = false;
}
}
else if (*p == '\0')
{
valint = (int) strtol(valstr, (char **)NULL, 10);
if (valint == 0 && errno != 0)
{
succp = false;
}
else
{
succp = true;
config_set_qualified_param(param, (void *)&valint, COUNT_TYPE);
}
}
if (succp)
{
service_add_qualified_param(service, param); /*< add param to svc */
}
return succp;
}
/**
* Add qualified config parameter to SERVICE struct.
*/
static void service_add_qualified_param(
SERVICE* svc,
CONFIG_PARAMETER* param)
{
CONFIG_PARAMETER** p;
spinlock_acquire(&svc->spin);
p = &svc->svc_config_param;
if ((*p) != NULL)
{
do
{
/** If duplicate is found, latter remains */
if (strncasecmp(param->name,
(*p)->name,
strlen(param->name)) == 0)
{
*p = config_clone_param(param);
break;
}
}
while ((*p)->next != NULL);
(*p)->next = config_clone_param(param);
}
else
{
(*p) = config_clone_param(param);
}
/** Increment service's configuration version */
atomic_add(&svc->svc_config_version, 1);
(*p)->next = NULL;
spinlock_release(&svc->spin);
}
char* service_get_name(
SERVICE* svc)
{
return svc->name;
}
Reference in New Issue View Git Blame Copy Permalink