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

Fixed repeated reconnects to the server when multiple monitors were used simultaneously and some had failing backend servers.

Browse Source 
Also added more documentation about dshard router's limitations.
This commit is contained in:
Markus Makela
2015-02-04 14:45:13 +02:00
parent 86bed0541c
commit 9c3cecf5b1
5 changed files with 222 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Documentation/Documentation-Contents.md
View File

@ -40,6 +40,10 @@
- [Top N Filter](filters/Top-N-Filter.md)
- [Firewall Filter](filters/Firewall-Filter.md)
## Routers
- [DBShard Router](routers/dbshard/DBShard.md)
## Design Documents
- [Session Commands design (in development)](http://mariadb-corporation.github.io/MaxScale/Design-Documents/)

0
Documentation/dbshard/DBShard-technical.md → Documentation/routers/dbshard/DBShard-technical.md
View File

14
Documentation/dbshard/DBShard.md → Documentation/routers/dbshard/DBShard.md
View File

@ -29,11 +29,19 @@ This would in effect allow the user 'john' to only see the database 'shard' on t
## Limitations
The dbshard router currently has some limitations due to the nature of the sharding implementation and the way the session variables are detected and routed. Cross-database queries (e.g. SELECT column FROM database1.table UNION select column FROM database2.table) are not supported and are routed either to the first explicit database in the query, the current database in use or to the first available database, if none of the previous conditions are met. Queries without explicit databases that are not session commands in them are either routed to the current or the first available database. This means that, for example when creating a new database, queries should be done directly on the node or the router should be equipped with the hint filter and a routing hint should be used.
The dbshard router currently has some limitations due to the nature of the sharding implementation and the way the session variables are detected and routed. Here is a list of the current limitations.
Temporary tables are only created on the explicit database in the query or the current database in use. If no database is in use and no database is explicitly stated, the behavior of the router is undefined.
- Cross-database queries (e.g. SELECT column FROM database1.table UNION select column FROM database2.table) are not supported and are routed either to the first explicit database in the query, the current database in use or to the first available database, if none of the previous conditions are met.
SELECT queries that modify session variables are not currently supported because uniform results can not be guaranteed. If such a query is executed, the behavior of the router is undefined. To work around this limitation the query must be executed in separate parts.
- Queries without explicit databases that are not session commands in them are either routed to the current or the first available database. This means that, for example when creating a new database, queries should be done directly on the node or the router should be equipped with the hint filter and a routing hint should be used.
- Temporary tables are only created on the explicit database in the query or the current database in use. If no database is in use and no database is explicitly stated, the behavior of the router is undefined.
- SELECT queries that modify session variables are not currently supported because uniform results can not be guaranteed. If such a query is executed, the behavior of the router is undefined. To work around this limitation the query must be executed in separate parts.
- Currently the dbshard router does not support connecting directly to a sharded database.
- Queries targeting databases not mapped by the dbshard router but still exist on the database server are not blocked but routed to the first available server. This possibly returns an error about database rights instead of a missing database. The behavior of the router is undefined in this case.
## Examples

0
Documentation/dbshard/dbshard-lifecycle.png → Documentation/routers/dbshard/dbshard-lifecycle.png
View File

Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 64 KiB

After

Width:  |  Height:  |  Size: 64 KiB

414
server/modules/routing/dbshard/dbshard.c
View File

@ -98,7 +98,7 @@ static route_target_t get_shard_route_target (
static uint8_t getCapabilities (ROUTER* inst, void* router_session);
bool parse_db_ignore_list(ROUTER_INSTANCE* router,char* param);
int bref_cmp_global_conn(
const void* bref1,
@ -264,22 +264,9 @@ static int hashcmpfun(
return strcmp(i1,i2);
}
static void* hstrdup(void* fval)
{
char* str = (char*)fval;
return strdup(str);
}
static void* hfree(void* fval)
{
free (fval);
return NULL;
}
bool parse_showdb_response(ROUTER_CLIENT_SES* rses, char* target, GWBUF* buf)
{
int rval = 0,i;
int rval = 0;
RESULTSET* rset;
RSET_ROW* row;
@ -297,25 +284,7 @@ bool parse_showdb_response(ROUTER_CLIENT_SES* rses, char* target, GWBUF* buf)
{
skygw_log_write(LOGFILE_TRACE,"dbshard: <%s, %s>",target,row->data[0]);
}
else
{
char* oldval = strdup(hashtable_fetch(rses->dbhash,row->data[0]));
for(i=0;i<rses->rses_nbackends;i++)
{
if(strcmp(oldval,rses->rses_backend_ref[i].bref_backend->backend_server->unique_name) == 0 &&
BREF_IS_CLOSED(&rses->rses_backend_ref[i]))
{
hashtable_delete(rses->dbhash,row->data[0]);
hashtable_add(rses->dbhash,row->data[0],target);
skygw_log_write(LOGFILE_TRACE,"dbshard: <%s, %s> (replaced %s)",target,row->data[0],oldval);
free(oldval);
oldval = NULL;
break;
}
}
free(oldval);
}
row = row->next;
}
resultset_free(rset);
@ -331,14 +300,12 @@ int gen_databaselist(ROUTER_INSTANCE* inst, ROUTER_CLIENT_SES* session)
DCB* dcb;
const char* query = "SHOW DATABASES;";
GWBUF *buffer,*clone;
backend_ref_t* backends;
int i,rval;
unsigned int len;
session->hash_init = false;
len = strlen(query);
backends = session->rses_backend_ref;
buffer = gwbuf_alloc(len + 4);
*((unsigned char*)buffer->start) = len;
*((unsigned char*)buffer->start + 1) = len>>8;
@ -348,243 +315,19 @@ int gen_databaselist(ROUTER_INSTANCE* inst, ROUTER_CLIENT_SES* session)
memcpy(buffer->start + 5,query,strlen(query));
for(i = 0;i<session->rses_nbackends;i++)
{
if(BREF_IS_IN_USE(&session->rses_backend_ref[i]) &&
!BREF_IS_CLOSED(&session->rses_backend_ref[i]))
{
clone = gwbuf_clone(buffer);
dcb = backends[i].bref_dcb;
if(BREF_IS_IN_USE(&backends[i]) && !BREF_IS_CLOSED(&backends[i]))
{
dcb = session->rses_backend_ref[i].bref_dcb;
rval = dcb->func.write(dcb,clone);
}
else
{
gwbuf_free(clone);
}
}
return !rval;
}
/**
* Updates the hashtable with the database names and where to find them, adding
* new and removing obsolete pairs.
* @param inst Router instance
* @param backends Backends to query for database names
* @param hashtable Hashtable to use
* @return True if all database and server names were successfully retrieved
* otherwise false
*/
/*
bool update_dbnames_hash(ROUTER_INSTANCE* inst,BACKEND** backends, HASHTABLE* hashtable)
{
const unsigned int connect_timeout = 1;
const unsigned int read_timeout = 1;
bool rval = true;
SERVER* server;
int i, rc, numfields;
for(i = 0;backends[i];i++)
{
MYSQL* handle = mysql_init(NULL);
MYSQL_RES* result = NULL;
MYSQL_ROW row;
char *user,*pwd = NULL;
if(handle == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: failed to initialize "
"MySQL handle.")));
return false;
}
rc = 0;
rc |= mysql_options(handle,
MYSQL_OPT_CONNECT_TIMEOUT,
(void *)&connect_timeout);
rc |= mysql_options(handle,
MYSQL_OPT_READ_TIMEOUT,
(void *)&read_timeout);
if(rc != 0)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: failed to set MySQL connection options.")));
mysql_close(handle);
rval = false;
continue;
}
server = backends[i]->backend_server;
ss_dassert(server != NULL);
if(!SERVER_IS_RUNNING(server))
{
continue;
}
if(server->monuser == NULL || server->monpw == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: no username or password "
"defined for server '%s'.",
server->unique_name)));
rval = false;
goto cleanup;
}
// Plain-text password used for authentication for now
user = server->monuser;
pwd = server->monpw;
if (mysql_real_connect(handle,
server->name,
user,
pwd,
NULL,
server->port,
NULL,
0) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: failed to connect to backend "
"server '%s:%d': %d %s",
server->name,
server->port,
mysql_errno(handle),
mysql_error(handle))));
rval = false;
goto cleanup;
}
//The server was successfully connected to, proceed to query for database names
if((result = mysql_list_dbs(handle,NULL)) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(LOGFILE_ERROR,
"Error: failed to retrieve databases from backend "
"server '%s': %d %s",
server->name,
mysql_errno(handle),
mysql_error(handle))));
goto cleanup;
}
numfields = mysql_num_fields(result);
if(numfields < 1)
{
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"Backend '%s' has no databases.",
server->name)));
goto cleanup;
}
*/
/**
* Walk through the list of databases in this backend
* and insert them into the hashtable. If the value is already in the hashtable
* but the backend isn't in the list of backends it is replaced with the first found backend.
*/
/*
while((row = mysql_fetch_row(result)))
{
unsigned long *lengths;
char *dbnm = NULL,*servnm = NULL;
lengths = mysql_fetch_lengths(result);
dbnm = (char*)calloc(lengths[0] + 1,sizeof(char));
memcpy(dbnm,row[0],lengths[0]);
servnm = strdup(server->unique_name);
if(hashtable_add(hashtable,dbnm,servnm) == 0)
{
char* srvname;
if((srvname = hashtable_fetch(hashtable,dbnm)) == NULL)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: failed to insert values into hashtable.")));
}
else
{
if(strcmp(srvname,server->unique_name) != 0)
{
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"Both \"%s\" and \"%s\" "
"have a database \"%s\".",
srvname,
server->unique_name,
dbnm)));
}
}
if(srvname)
{
char* old_backend = (char*)hashtable_fetch(hashtable,dbnm);
int j;
bool is_alive = false;
for(j = 0;backends[j];j++)
{
* */
/**
* See if the old backend is still
* alive. If not then update
* the hashtable with the current backend's name.
*/
/*
if(strcmp(backends[j]->backend_server->unique_name,old_backend) == 0 &&
SERVER_IS_RUNNING(backends[j]->backend_server))
{
is_alive = true;
break;
}
}
if(!is_alive)
{
hashtable_delete(hashtable,dbnm);
if(hashtable_add(hashtable,dbnm,servnm))
{
LOGIF(LT, (skygw_log_write_flush(
LOGFILE_TRACE,
"Updated the backend of database '%s' to '%s'.",
dbnm,
servnm)));
}
else
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error: failed to insert values into hashtable.")));
}
}
}
}
}
cleanup:
if(result)
{
mysql_free_result(result);
}
result = NULL;
mysql_close(handle);
}
return rval;
}
*/
/**
* Check the hashtable for the right backend for this query.
* @param router Router instance
@ -1376,7 +1119,7 @@ static route_target_t get_shard_route_target (
#if defined(SS_DEBUG)
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Selected target \"%s\"",
"Selected target type \"%s\"",
STRTARGET(target))));
#endif
return target;
@ -1588,8 +1331,8 @@ void check_create_tmp_table(
if(rses_prop_tmp){
if (rses_prop_tmp->rses_prop_data.temp_tables == NULL)
{
h = hashtable_alloc(7, hashkeyfun, hashcmpfun);
hashtable_memory_fns(h,hstrdup,NULL,hfree,NULL);
h = hashtable_alloc(100, hashkeyfun, hashcmpfun);
hashtable_memory_fns(h,(HASHMEMORYFN)strdup,(HASHMEMORYFN)strdup,(HASHMEMORYFN)free,(HASHMEMORYFN)free);
if (h != NULL)
{
rses_prop_tmp->rses_prop_data.temp_tables = h;
@ -1630,15 +1373,24 @@ void check_create_tmp_table(
free(tblname);
}
}
GWBUF* gen_show_dbs_response(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client)
/**
* Generates a custom SHOW DATABASES result set from all the databases in the
* hashtable. Only backend servers that are up and in a proper state are listed
* in it.
* @param router Router instance
* @param client Router client session
* @return Pointer to new GWBUF containing the custom result set
*/
GWBUF*
gen_show_dbs_response(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client)
{
GWBUF* rval = NULL;
HASHTABLE* ht = client->dbhash;
HASHITERATOR* iter = hashtable_iterator(ht);
backend_ref_t *bref = client->rses_backend_ref;
BACKEND** backends = router->servers;
unsigned int coldef_len = 0;
int j;
int i;
char dbname[MYSQL_DATABASE_MAXLEN + 1];
char *value;
unsigned char* ptr;
@ -1735,12 +1487,12 @@ GWBUF* gen_show_dbs_response(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client)
while((value = (char*) hashtable_next(iter)))
{
char* bend = hashtable_fetch(ht, value);
for(j = 0;backends[j];j++)
for(i = 0; backends[i]; i++)
{
if(strcmp(backends[j]->backend_server->unique_name,bend) == 0)
{
if(SERVER_IS_RUNNING(backends[j]->backend_server))
if(strcmp(bref[i].bref_backend->backend_server->unique_name, bend) == 0 &&
BREF_IS_IN_USE(&bref[i]) && !BREF_IS_CLOSED(&bref[i]))
{
GWBUF* temp;
int plen = strlen(value) + 1;
@ -1757,8 +1509,7 @@ GWBUF* gen_show_dbs_response(ROUTER_INSTANCE* router, ROUTER_CLIENT_SES* client)
/** Append the row*/
rval = gwbuf_append(rval, temp);
}
break;
}
}
}
@ -1907,54 +1658,6 @@ static int routeQuery(
break;
} /**< switch by packet type */
if (packet_type == MYSQL_COM_INIT_DB)
{
if (!(change_successful = change_current_db(inst, router_cli_ses, querybuf)))
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Changing database failed.")));
}
/* ret = 1; */
/* goto retblock; */
}
/**
* If autocommit is disabled or transaction is explicitly started
* transaction becomes active and master gets all statements until
* transaction is committed and autocommit is enabled again.
*/
if (router_cli_ses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_DISABLE_AUTOCOMMIT))
{
router_cli_ses->rses_autocommit_enabled = false;
if (!router_cli_ses->rses_transaction_active)
{
router_cli_ses->rses_transaction_active = true;
}
}
else if (!router_cli_ses->rses_transaction_active &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_BEGIN_TRX))
{
router_cli_ses->rses_transaction_active = true;
}
/**
* Explicit COMMIT and ROLLBACK, implicit COMMIT.
*/
if (router_cli_ses->rses_autocommit_enabled &&
router_cli_ses->rses_transaction_active &&
(QUERY_IS_TYPE(qtype,QUERY_TYPE_COMMIT) ||
QUERY_IS_TYPE(qtype,QUERY_TYPE_ROLLBACK)))
{
router_cli_ses->rses_transaction_active = false;
}
else if (!router_cli_ses->rses_autocommit_enabled &&
QUERY_IS_TYPE(qtype, QUERY_TYPE_ENABLE_AUTOCOMMIT))
{
router_cli_ses->rses_autocommit_enabled = true;
router_cli_ses->rses_transaction_active = false;
}
if (LOG_IS_ENABLED(LOGFILE_TRACE))
{
@ -1968,10 +1671,9 @@ static int routeQuery(
skygw_log_write(
LOGFILE_TRACE,
"> Autocommit: %s, trx is %s, cmd: %s, type: %s, "
"> Cmd: %s, type: %s, "
"stmt: %s%s %s",
(router_cli_ses->rses_autocommit_enabled ? "[enabled]" : "[disabled]"),
(router_cli_ses->rses_transaction_active ? "[open]" : "[not open]"),
STRPACKETTYPE(ptype),
(qtypestr==NULL ? "N/A" : qtypestr),
contentstr,
@ -1985,6 +1687,17 @@ static int routeQuery(
* Find out whether the query should be routed to single server or to
* all of them.
*/
if (packet_type == MYSQL_COM_INIT_DB)
{
if (!(change_successful = change_current_db(inst, router_cli_ses, querybuf)))
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Changing database failed.")));
}
}
if(QUERY_IS_TYPE(qtype, QUERY_TYPE_SHOW_DATABASES))
{
/**
@ -2024,11 +1737,9 @@ static int routeQuery(
if (packet_type == MYSQL_COM_INIT_DB)
{
char dbname[MYSQL_DATABASE_MAXLEN+1];
unsigned int plen = gw_mysql_get_byte3((unsigned char*)querybuf->start) - 1;
memcpy(dbname,querybuf->start + 5,plen);
dbname[plen] = '\0';
tname = hashtable_fetch(router_cli_ses->dbhash,dbname);
route_target = TARGET_UNDEFINED;
tname = hashtable_fetch(router_cli_ses->dbhash,router_cli_ses->rses_mysql_session->db);
if(tname)
{
route_target = TARGET_NAMED_SERVER;
@ -2829,9 +2540,7 @@ static bool connect_backend_servers(
{
BACKEND* b = backend_ref[i].bref_backend;
if (SERVER_IS_RUNNING(b->backend_server) &&
((b->backend_server->status & router->bitmask) ==
router->bitvalue))
if (SERVER_IS_RUNNING(b->backend_server))
{
servers_found += 1;
@ -3935,10 +3644,10 @@ static bool handle_error_new_connection(
*/
if ((bref = get_bref_from_dcb(rses, backend_dcb)) == NULL)
{
ss_dassert(bref != NULL);
succp = false;
goto return_succp;
}
CHK_BACKEND_REF(bref);
/**
@ -3995,11 +3704,24 @@ static bool handle_error_new_connection(
}
rses->hash_init = false;
for(i = 0;i<rses->rses_nbackends;i++)
{
bref_clear_state(&rses->rses_backend_ref[i],BREF_DB_MAPPED);
}
HASHITERATOR* iter = hashtable_iterator(rses->dbhash);
char* srvnm = bref->bref_backend->backend_server->unique_name;
char *key, *value;
while((key = (char*)hashtable_next(iter)))
{
value = hashtable_fetch(rses->dbhash,key);
if(strcmp(value,srvnm) == 0)
{
hashtable_delete(rses->dbhash,key);
}
}
skygw_log_write(LOGFILE_TRACE,"dbshard: Re-mapping databases");
gen_databaselist(rses->router,rses);
@ -4109,18 +3831,18 @@ static backend_ref_t* get_bref_from_dcb(
}
return bref;
}
/**
* Calls hang-up function for DCB if it is not both running and in
* master/slave/joined/ndb role. Called by DCB's callback routine.
*/
static int router_handle_state_switch(
static int
router_handle_state_switch(
DCB* dcb,
DCB_REASON reason,
void* data)
{
backend_ref_t* bref;
int rc = 1;
int rc = 1,i;
ROUTER_CLIENT_SES* rses;
SESSION* ses;
SERVER* srv;
@ -4131,7 +3853,7 @@ static int router_handle_state_switch(
srv = bref->bref_backend->backend_server;
if (SERVER_IS_RUNNING(srv) && SERVER_IS_IN_CLUSTER(srv))
if(SERVER_IS_RUNNING(srv))
{
goto return_rc;
}
@ -4141,8 +3863,11 @@ static int router_handle_state_switch(
rses = (ROUTER_CLIENT_SES *) dcb->session->router_session;
CHK_CLIENT_RSES(rses);
switch (reason) {
switch(reason)
{
case DCB_REASON_NOT_RESPONDING:
atomic_add(&bref->bref_backend->backend_conn_count, -1);
skygw_log_write(LOGFILE_TRACE,"dbshard: server %s not responding",srv->unique_name);
dcb->func.hangup(dcb);
break;
@ -4196,11 +3921,8 @@ static bool change_current_db(
/** Copy database name from MySQL packet to session */
memcpy(rses->rses_mysql_session->db,
packet + 5,
plen);
memcpy(rses->rses_mysql_session->db,packet + 5,plen);
memset(rses->rses_mysql_session->db + plen,0,1);
/**
* Update the session's active database only if it's in the hashtable.
* If it isn't found, send a custom error packet to the client.
@ -4244,9 +3966,13 @@ static bool change_current_db(
reply_error:
{
GWBUF* errbuf;
skygw_log_write_flush(
LOGFILE_TRACE,
"dbshard: failed to change database: %s", fail_str);
errbuf = modutil_create_mysql_err_msg(1, 0, 1049, "42000", fail_str);
free(fail_str);
if (errbuf == NULL)
{
LOGIF(LE, (skygw_log_write_flush(