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

Merge branch 'develop' of https://github.com/mariadb-corporation/MaxScale into develop

Browse Source
This commit is contained in:
Mark Riddoch 2015-02-20 16:06:18 +00:00
commit 77a25c5848
20 changed files with 501 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
CMakeLists.txt
View File

@ -137,7 +137,10 @@ if( NOT ( (DEFINED INSTALL_SYSTEM_FILES) AND ( NOT ( INSTALL_SYSTEM_FILES ) ) )
else()
configure_file(${CMAKE_SOURCE_DIR}/etc/init.d/maxscale.in ${CMAKE_BINARY_DIR}/maxscale @ONLY)
endif()
if(NOT PACKAGE)
if(PACKAGE)
message(STATUS "maxscale.conf will unpack to: /etc/ld.so.conf.d")
message(STATUS "startup scripts will unpack to to: /etc/init.d")
else()
install(FILES ${CMAKE_BINARY_DIR}/maxscale DESTINATION /etc/init.d
PERMISSIONS OWNER_READ OWNER_WRITE OWNER_EXECUTE GROUP_READ GROUP_EXECUTE WORLD_READ WORLD_EXECUTE)
install(FILES ${CMAKE_BINARY_DIR}/maxscale.conf DESTINATION /etc/ld.so.conf.d
@ -262,3 +265,8 @@ add_custom_target(generate_html
-DCMAKE_MODULE_PATH=${CMAKE_MODULE_PATH}
-P generate-html.cmake
COMMENT "Generating HTML files" VERBATIM)
if(PACKAGE)
message(STATUS "You can install startup scripts and system configuration files for MaxScale by running the 'postinst' shell script located at ${CMAKE_INSTALL_PREFIX}.")
message(STATUS "To remove these installed files, run the 'postrm' shell script located in the same folder.")
endif()

2
Documentation/Getting-Started/Configuration-Guide.md
View File

@ -710,6 +710,8 @@ Session commands include for example:
**NOTE: if variable assignment is embedded in a write statement it is routed to _Master_ only. For example, `INSERT INTO t1 values(@myvar:=5, 7)` would be routed to _Master_ only.**
The router stores all of the executed session commands so that in case of a slave failure, a replacement slave can be chosen and the session command history can be repeated on that new slave. This means that the router stores each executed session command for the duration of the session. Applications that use long-running sessions might cause MaxScale to consume a growing amount of memory unless the sessions are closed. This can be solved by setting a connection timeout on the application side.
#### Configuring the Read/Write Split router
Read/Write Split router-specific settings are specified in the configuration file of MaxScale in its specific section. The section can be freely named but the name is used later as a reference from listener section.

16
Documentation/filters/Firewall-Filter.md
View File

@ -19,7 +19,7 @@ The firewall filter does not support any filter options.
### Filter Parameters
The firewall filter has one mandatory parameter that defines the location of the rule file. This is the 'rules' parameter and it expects an absolute path to the rule file.
The firewall filter has one mandatory parameter that defines the location of the rule file. This is the `rules` parameter and it expects an absolute path to the rule file.
## Rule syntax
@ -31,7 +31,7 @@ The rules are defined by using the following syntax.
Rules always define a blocking action so the basic mode for the firewall filter is to allow all queries that do not match a given set of rules. Rules are identified by their name and have a mandatory part and optional parts.
The first step of defining a rule is to start with the keyword 'rule' which identifies this line of text as a rule. The second token is identified as the name of the rule. After that the mandatory token 'deny' is required to mark the start of the actual rule definition.
The first step of defining a rule is to start with the keyword `rule` which identifies this line of text as a rule. The second token is identified as the name of the rule. After that the mandatory token `deny` is required to mark the start of the actual rule definition.
### Mandatory rule parameters
@ -43,7 +43,7 @@ This rule blocks all queries that use the wildcard character *.
#### Columns
This rule expects a list of values after the 'columns' keyword. These values are interpreted as column names and if a query targets any of these, it is blocked.
This rule expects a list of values after the `columns` keyword. These values are interpreted as column names and if a query targets any of these, it is blocked.
#### Regex
@ -63,7 +63,7 @@ Each mandatory rule accepts one or more optional parameters. These are to be def
#### At_times
This rule expects a list of time ranges that define the times when the rule in question is active. The time formats are expected to be ISO-8601 compliant and to be separated by a single dash (the - character). For example defining the active period of a rule to be 17:00 to 19:00 you would add 'at times 17:00:00-19:00:00' to the end of the rule.
This rule expects a list of time ranges that define the times when the rule in question is active. The time formats are expected to be ISO-8601 compliant and to be separated by a single dash (the - character). For example defining the active period of a rule to be 17:00 to 19:00 you would add `at times 17:00:00-19:00:00` to the end of the rule.
#### On_queries
@ -73,9 +73,11 @@ This limits the rule to be active only on certain types of queries.
To apply the defined rules to users use the following syntax.
`users NAME ... match [any|all] rules RULE ...`
`users NAME ... match [any|all|strict_all] rules RULE ...`
The first keyword is users which identifies this line as a user definition line. After this a list of user names and network addresses in the format 'user@0.0.0.0' is expected. The first part is the user name and the second part is the network address. You can use the '%' character as the wildcard to enable user name matching from any address or network matching for all users. After the list of users and networks the keyword match is expected. After this either the keyword 'any' or 'all' is expected. This defined how the rules are matched. If 'any' is used when the first rule is matched the query is considered blocked and the rest of the rules are skipped. If instead the 'all' keyword is used all rules must match for the query to be blocked.
The first keyword is users which identifies this line as a user definition line. After this a list of user names and network addresses in the format `user@0.0.0.0` is expected. The first part is the user name and the second part is the network address. You can use the `%` character as the wildcard to enable user name matching from any address or network matching for all users. After the list of users and networks the keyword match is expected.
After this either the keyword `any` `all` or `strict_all` is expected. This defined how the rules are matched. If `any` is used when the first rule is matched the query is considered blocked and the rest of the rules are skipped. If instead the `all` keyword is used all rules must match for the query to be blocked. The `strict_all` is the same as `all` but it checks the rules from left to right in the order they were listed. If one of these does not match, the rest of the rules are not checked. This could be usedful in situations where you would for example combine `limit_queries` and `regex` rules. By using `strict_all` you can have the `regex` rule first and the `limit_queries` rule second. This way the rule only matches if the `regex` rule matches enough times for the `limit_queries` rule to match.
After the matching part comes the rules keyword after which a list of rule names is expected. This allows reusing of the rules and enables varying levels of query restriction.
@ -105,4 +107,4 @@ This can be achieved by creating two rules. One that blocks the usage of the wil
We want to prevent accidental deletes into the managers table where the where clause is missing. This poses a problem, we don't want to require all the delete queries to have a where clause. We only want to prevent the data in the managers table from being deleted without a where clause.
To achieve this, we need two rules. The first rule can be seen on line 5 in the example rule file. This defines that all delete operations must have a where clause. This rule alone does us no good so we need a second one. The second rule is defined on line 6 and it blocks all queries that match the provided regular expression. When we combine these two rules we get the result we want. You can see the application of these rules on line 9 of the example rule file. The usage of the 'all' matching mode requires that all the rules must match for the query to be blocked. This in effect combines the two rules into a more complex rule.
To achieve this, we need two rules. The first rule can be seen on line 5 in the example rule file. This defines that all delete operations must have a where clause. This rule alone does us no good so we need a second one. The second rule is defined on line 6 and it blocks all queries that match the provided regular expression. When we combine these two rules we get the result we want. You can see the application of these rules on line 9 of the example rule file. The usage of the `all` and `strict_all` matching mode requires that all the rules must match for the query to be blocked. This in effect combines the two rules into a more complex rule.

10
log_manager/log_manager.cc
View File

@ -683,7 +683,7 @@ static int logmanager_write_log(
size_t safe_str_len;
/** Length of session id */
size_t sesid_str_len;
size_t cmplen = 0;
/**
* 2 braces, 2 spaces and terminating char
* If session id is stored to tls_log_info structure, allocate
@ -691,7 +691,7 @@ static int logmanager_write_log(
*/
if (id == LOGFILE_TRACE && tls_log_info.li_sesid != 0)
{
sesid_str_len = 2+2+get_decimal_len(tls_log_info.li_sesid)+1;
sesid_str_len = 5*sizeof(char)+get_decimal_len(tls_log_info.li_sesid);
}
else
{
@ -699,14 +699,16 @@ static int logmanager_write_log(
}
timestamp_len = get_timestamp_len();
cmplen = sesid_str_len > 0 ? sesid_str_len - sizeof(char) : 0;
/** Find out how much can be safely written with current block size */
if (timestamp_len-1+MAX(sesid_str_len-1,0)+str_len > lf->lf_buf_size)
if (timestamp_len-sizeof(char)+cmplen+str_len > lf->lf_buf_size)
{
safe_str_len = lf->lf_buf_size;
}
else
{
safe_str_len = timestamp_len-1+MAX(sesid_str_len-1,0)+str_len;
safe_str_len = timestamp_len-sizeof(char)+cmplen+str_len;
}
/**
* Seek write position and register to block buffer.

84
server/core/server.c
View File

@ -293,6 +293,90 @@ char *stat;
spinlock_release(&server_spin);
}
/**
* Print all servers in Json format to a DCB
*
* Designed to be called within a debugger session in order
* to display all active servers within the gateway
*/
void
dprintAllServersJson(DCB *dcb)
{
SERVER *ptr;
char *stat;
int len = 0;
int el = 1;
spinlock_acquire(&server_spin);
ptr = allServers;
while (ptr)
{
ptr = ptr->next;
len++;
}
ptr = allServers;
dcb_printf(dcb, "[\n");
while (ptr)
{
dcb_printf(dcb, " {\n \"server\": \"%s\",\n",
ptr->name);
stat = server_status(ptr);
dcb_printf(dcb, " \"status\": \"%s\",\n",
stat);
free(stat);
dcb_printf(dcb, " \"protocol\": \"%s\",\n",
ptr->protocol);
dcb_printf(dcb, " \"port\": \"%d\",\n",
ptr->port);
if (ptr->server_string)
dcb_printf(dcb, " \"version\": \"%s\",\n",
ptr->server_string);
dcb_printf(dcb, " \"nodeId\": \"%d\",\n",
ptr->node_id);
dcb_printf(dcb, " \"masterId\": \"%d\",\n",
ptr->master_id);
if (ptr->slaves) {
int i;
dcb_printf(dcb, " \"slaveIds\": [ ");
for (i = 0; ptr->slaves[i]; i++)
{
if (i == 0)
dcb_printf(dcb, "%li", ptr->slaves[i]);
else
dcb_printf(dcb, ", %li ", ptr->slaves[i]);
}
dcb_printf(dcb, "],\n");
}
dcb_printf(dcb, " \"replDepth\": \"%d\",\n",
ptr->depth);
if (SERVER_IS_SLAVE(ptr) || SERVER_IS_RELAY_SERVER(ptr)) {
if (ptr->rlag >= 0) {
dcb_printf(dcb, " \"slaveDelay\": \"%d\",\n", ptr->rlag);
}
}
if (ptr->node_ts > 0) {
dcb_printf(dcb, " \"lastReplHeartbeat\": \"%lu\",\n", ptr->node_ts);
}
dcb_printf(dcb, " \"totalConnections\": \"%d\",\n",
ptr->stats.n_connections);
dcb_printf(dcb, " \"currentConnections\": \"%d\",\n",
ptr->stats.n_current);
dcb_printf(dcb, " \"currentOps\": \"%d\"\n",
ptr->stats.n_current_ops);
if (el < len) {
dcb_printf(dcb, " },\n");
}
else {
dcb_printf(dcb, " }\n");
}
ptr = ptr->next;
el++;
}
dcb_printf(dcb, "]\n");
spinlock_release(&server_spin);
}
/**
* Print server details to a DCB
*

2
server/core/test/testhash.c
View File

@ -157,7 +157,7 @@ static bool do_hashtest(
CHK_HASHTABLE(h);
hashtable_free(h);
return_succp:
free(val_arr);
return succp;
}

2
server/core/test/testserver.c
View File

@ -32,7 +32,7 @@
#include <string.h>
#include <server.h>
#include <log_manager.h>
/**
* test1 Allocate a server and do lots of other things
*

14
server/core/test/testservice.c
View File

@ -66,10 +66,10 @@ poll_init();
ss_info_dassert(NULL != service, "New service with valid router must not be null");
ss_info_dassert(0 != service_isvalid(service), "Service must be valid after creation");
ss_info_dassert(0 == strcmp("MyService", service_get_name(service)), "Service must have given name");
ss_dfprintf(stderr, "\t..done\nAdding protocol HTTPD.");
ss_info_dassert(0 != serviceAddProtocol(service, "HTTPD", "localhost", 9876), "Add Protocol should succeed");
ss_info_dassert(0 != serviceHasProtocol(service, "HTTPD", 9876), "Service should have new protocol as requested");
serviceStartProtocol(service, "HTTPD", 9876);
ss_dfprintf(stderr, "\t..done\nAdding protocol testprotocol.");
ss_info_dassert(0 != serviceAddProtocol(service, "testprotocol", "localhost", 9876), "Add Protocol should succeed");
ss_info_dassert(0 != serviceHasProtocol(service, "testprotocol", 9876), "Service should have new protocol as requested");
serviceStartProtocol(service, "testprotocol", 9876);
skygw_log_sync_all();
ss_dfprintf(stderr, "\t..done\nStarting Service.");
result = serviceStart(service);
@ -84,10 +84,16 @@ poll_init();
ss_dfprintf(stderr, "\t..done\nStopping Service.");
ss_info_dassert(0 != serviceStop(service), "Stop should succeed");
ss_dfprintf(stderr, "\t..done\n");
/** This is never used in MaxScale and will always fail due to service's
* stats.n_current value never being decremented */
/*
ss_dfprintf(stderr, "\t..done\nFreeing Service.");
ss_info_dassert(0 != service_free(service), "Free should succeed");
ss_dfprintf(stderr, "\t..done\n");
*/
return 0;
}

1
server/include/server.h
View File

@ -176,6 +176,7 @@ extern SERVER *server_find(char *, unsigned short);
extern void printServer(SERVER *);
extern void printAllServers();
extern void dprintAllServers(DCB *);
extern void dprintAllServersJson(DCB *);
extern void dprintServer(DCB *, SERVER *);
extern void dListServers(DCB *);
extern char *server_status(SERVER *);

167
server/modules/filter/fwfilter.c
View File

@ -58,7 +58,7 @@
* combinations of username and network, either the value any or all,
* depending on how you want to match the rules, and one or more rule names.
*@code{.unparsed}
* users NAME ... match [any|all] rules RULE ...
* users NAME ... match [any|all|strict_all] rules RULE ...
*@endcode
*/
#include <my_config.h>
@ -166,6 +166,7 @@ typedef struct queryspeed_t{
int count; /*< Number of queries done */
int limit; /*< Maximum number of queries */
long id; /*< Unique id of the rule */
bool active; /*< If the rule has been triggered */
struct queryspeed_t* next; /*< Next node in the list */
}QUERYSPEED;
@ -200,6 +201,9 @@ typedef struct user_t{
QUERYSPEED* qs_limit;/*< The query speed structure unique to this user */
RULELIST* rules_or;/*< If any of these rules match the action is triggered */
RULELIST* rules_and;/*< All of these rules must match for the action to trigger */
RULELIST* rules_strict_and; /*< rules that skip the rest of the rules if one of them
* fails. This is only for rules paired with 'match strict_all'. */
}USER;
/**
@ -625,6 +629,7 @@ void add_users(char* rule, FW_INSTANCE* instance)
instance->userstrings = link;
}
/**
* Parses the list of rule strings for users and links them against the listed rules.
* Only adds those rules that are found. If the rule isn't found a message is written to the error log.
@ -641,7 +646,7 @@ void link_rules(char* rule, FW_INSTANCE* instance)
char *tok, *ruleptr, *userptr, *modeptr;
char *saveptr = NULL;
RULELIST* rulelist = NULL;
bool strict = false;
userptr = strstr(rule,"users ");
modeptr = strstr(rule," match ");
ruleptr = strstr(rule," rules ");
@ -662,6 +667,9 @@ void link_rules(char* rule, FW_INSTANCE* instance)
match_any = true;
}else if(strcmp(tok,"all") == 0){
match_any = false;
}else if(strcmp(tok,"strict_all") == 0){
match_any = false;
strict = true;
}else{
skygw_log_write(LOGFILE_ERROR, "fwfilter: Rule syntax incorrect, 'match' was not followed by 'any' or 'all': %s",rule);
return;
@ -730,10 +738,15 @@ void link_rules(char* rule, FW_INSTANCE* instance)
if(match_any){
tail->next = user->rules_or;
user->rules_or = tl;
}else{
tail->next = user->rules_and;
user->rules_and = tl;
}else if(strict){
tail->next = user->rules_and;
user->rules_strict_and = tl;
}
else
{
tail->next = user->rules_and;
user->rules_and = tl;
}
hashtable_add(instance->htable,
(void *)userptr,
@ -1295,11 +1308,6 @@ bool rule_matches(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *queue
time_t time_now;
struct tm* tm_now;
if(my_session->errmsg){
free(my_session->errmsg);
my_session->errmsg = NULL;
}
time(&time_now);
tm_now = localtime(&time_now);
@ -1439,43 +1447,56 @@ bool rule_matches(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *queue
queryspeed->next = user->qs_limit;
user->qs_limit = queryspeed;
}
if(queryspeed->count > queryspeed->limit)
{
queryspeed->triggered = time_now;
queryspeed->count = 0;
matches = true;
skygw_log_write(LOGFILE_TRACE,
"fwfilter: rule '%s': query limit triggered (%d queries in %f seconds), denying queries from user for %f seconds.",
rulelist->rule->name,
queryspeed->limit,
queryspeed->period,
queryspeed->cooldown);
double blocked_for = queryspeed->cooldown - difftime(time_now,queryspeed->triggered);
sprintf(emsg,"Queries denied for %f seconds",blocked_for);
msg = strdup(emsg);
}
else if(difftime(time_now,queryspeed->triggered) < queryspeed->cooldown)
{
double blocked_for = queryspeed->cooldown - difftime(time_now,queryspeed->triggered);
sprintf(emsg,"Queries denied for %f seconds",blocked_for);
skygw_log_write(LOGFILE_TRACE, "fwfilter: rule '%s': user denied for %f seconds",rulelist->rule->name,blocked_for);
msg = strdup(emsg);
matches = true;
}
else if(difftime(time_now,queryspeed->first_query) < queryspeed->period)
{
queryspeed->count++;
}
else
{
queryspeed->first_query = time_now;
}
if(queryspeed->active)
{
if(difftime(time_now,queryspeed->triggered) < queryspeed->cooldown)
{
double blocked_for = queryspeed->cooldown - difftime(time_now,queryspeed->triggered);
sprintf(emsg,"Queries denied for %f seconds",blocked_for);
skygw_log_write(LOGFILE_TRACE, "fwfilter: rule '%s': user denied for %f seconds",rulelist->rule->name,blocked_for);
msg = strdup(emsg);
matches = true;
}
else
{
queryspeed->active = false;
queryspeed->count = 0;
}
}
else
{
if(queryspeed->count >= queryspeed->limit)
{
queryspeed->triggered = time_now;
matches = true;
queryspeed->active = true;
skygw_log_write(LOGFILE_TRACE,
"fwfilter: rule '%s': query limit triggered (%d queries in %f seconds), denying queries from user for %f seconds.",
rulelist->rule->name,
queryspeed->limit,
queryspeed->period,
queryspeed->cooldown);
double blocked_for = queryspeed->cooldown - difftime(time_now,queryspeed->triggered);
sprintf(emsg,"Queries denied for %f seconds",blocked_for);
msg = strdup(emsg);
}
else if(queryspeed->count > 0 &&
difftime(time_now,queryspeed->first_query) <= queryspeed->period)
{
queryspeed->count++;
}
else
{
queryspeed->first_query = time_now;
queryspeed->count = 1;
}
}
break;
@ -1499,7 +1520,11 @@ bool rule_matches(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *queue
queryresolved:
if(msg){
my_session->errmsg = msg;
if(my_session->errmsg){
free(my_session->errmsg);
}
my_session->errmsg = msg;
}
if(matches){
@ -1536,7 +1561,10 @@ bool check_match_any(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *qu
memset(fullquery + qlen,0,1);
}
rulelist = user->rules_or;
if((rulelist = user->rules_or) == NULL)
{
goto retblock;
}
while(rulelist){
@ -1544,9 +1572,10 @@ bool check_match_any(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *qu
rulelist = rulelist->next;
continue;
}
if((rval = rule_matches(my_instance,my_session,queue,user,rulelist,fullquery))){
goto retblock;
if((rval = rule_matches(my_instance,my_session,queue,user,rulelist,fullquery))){
goto retblock;
}
rulelist = rulelist->next;
}
@ -1565,9 +1594,9 @@ bool check_match_any(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *qu
* @param user The user whose rulelist is checked
* @return True if the query matches all of the rules otherwise false
*/
bool check_match_all(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *queue, USER* user)
bool check_match_all(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *queue, USER* user,bool strict_all)
{
bool is_sql, rval = 0;
bool is_sql, rval = true;
int qlen;
char *fullquery = NULL,*ptr;
@ -1585,23 +1614,38 @@ bool check_match_all(FW_INSTANCE* my_instance, FW_SESSION* my_session, GWBUF *qu
}
rulelist = user->rules_or;
if(strict_all)
{
rulelist = user->rules_strict_and;
}
else
{
rulelist = user->rules_and;
}
if(rulelist == NULL)
{
rval = false;
goto retblock;
}
while(rulelist){
if(!rule_is_active(rulelist->rule)){
rulelist = rulelist->next;
continue;
}
if(!rule_matches(my_instance,my_session,queue,user,rulelist,fullquery)){
rval = false;
goto retblock;
if(strict_all)
break;
}
rulelist = rulelist->next;
}
retblock:
free(fullquery);
@ -1664,7 +1708,12 @@ routeQuery(FILTER *instance, void *session, GWBUF *queue)
goto queryresolved;
}
if(check_match_all(my_instance,my_session,queue,user)){
if(check_match_all(my_instance,my_session,queue,user,false)){
accept = false;
goto queryresolved;
}
if(check_match_all(my_instance,my_session,queue,user,true)){
accept = false;
goto queryresolved;
}

2
server/modules/filter/test/harness_ui.c
View File

@ -187,7 +187,7 @@ int main(int argc, char** argv){
}
instance.thrpool = t_thr_pool;
int thr_num = 1;
intptr_t thr_num = 1;
for(i = 0;i<instance.thrcount;i++){

11
server/modules/include/readwritesplit.h
View File

@ -31,6 +31,7 @@
#include <dcb.h>
#include <hashtable.h>
#include <math.h>
#undef PREP_STMT_CACHING
@ -54,7 +55,8 @@ typedef enum bref_state {
BREF_IN_USE = 0x01,
BREF_WAITING_RESULT = 0x02, /*< for session commands only */
BREF_QUERY_ACTIVE = 0x04, /*< for other queries */
BREF_CLOSED = 0x08
BREF_CLOSED = 0x08,
BREF_SESCMD_FAILED = 0x10 /*< Backend references that should be dropped */
} bref_state_t;
#define BREF_IS_NOT_USED(s) ((s)->bref_state & ~BREF_IN_USE)
@ -62,6 +64,7 @@ typedef enum bref_state {
#define BREF_IS_WAITING_RESULT(s) ((s)->bref_num_result_wait > 0)
#define BREF_IS_QUERY_ACTIVE(s) ((s)->bref_state & BREF_QUERY_ACTIVE)
#define BREF_IS_CLOSED(s) ((s)->bref_state & BREF_CLOSED)
#define BREF_HAS_FAILED(s) ((s)->bref_state & BREF_SESCMD_FAILED)
typedef enum backend_type_t {
BE_UNDEFINED=-1,
@ -144,6 +147,9 @@ typedef struct mysql_sescmd_st {
GWBUF* my_sescmd_buf; /*< query buffer */
unsigned char my_sescmd_packet_type;/*< packet type */
bool my_sescmd_is_replied; /*< is cmd replied to client */
unsigned char reply_cmd; /*< The reply command. One of OK, ERR, RESULTSET or
* LOCAL_INFILE. Slave servers are compared to this
* when they return session command replies.*/
#if defined(SS_DEBUG)
skygw_chk_t my_sescmd_chk_tail;
#endif
@ -226,6 +232,9 @@ typedef struct backend_ref_st {
int bref_num_result_wait;
sescmd_cursor_t bref_sescmd_cur;
GWBUF* bref_pending_cmd; /*< For stmt which can't be routed due active sescmd execution */
unsigned char
reply_cmd; /*< The reply the backend server sent to a session command.
* Used to detect slaves that fail to execute session command. */
#if defined(SS_DEBUG)
skygw_chk_t bref_chk_tail;
#endif

5
server/modules/protocol/CMakeLists.txt
View File

@ -14,6 +14,11 @@ add_library(HTTPD SHARED httpd.c)
target_link_libraries(HTTPD log_manager utils)
install(TARGETS HTTPD DESTINATION modules)
if(BUILD_TESTS)
add_library(testprotocol SHARED testprotocol.c)
install(TARGETS testprotocol DESTINATION modules)
endif()
add_library(maxscaled SHARED maxscaled.c)
target_link_libraries(maxscaled log_manager utils)
install(TARGETS maxscaled DESTINATION modules)

107
server/modules/protocol/testprotocol.c Normal file
View File

@ -0,0 +1,107 @@
/*
* This file is distributed as part of the MariaDB Corporation 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 2013-2014
*/
/**
* @file testprotocol.c - Testing protocol module
*
* Not intended for actual use. This protocol module does nothing useful and
* is only meant to test that the module loading works.
*
* @verbatim
* Revision History
* Date Who Description
* 20/02/2015 Markus Mäkelä Initial implementation
*
* @endverbatim
*/
#include <modinfo.h>
#include <dcb.h>
#include <buffer.h>
MODULE_INFO info = {
MODULE_API_PROTOCOL,
MODULE_IN_DEVELOPMENT,
GWPROTOCOL_VERSION,
"Test protocol"
};
static char *version_str = "V1.0.0";
static int test_read(DCB* dcb){ return 1;}
static int test_write(DCB *dcb, GWBUF* buf){ return 1;}
static int test_write_ready(DCB *dcb){ return 1;}
static int test_error(DCB *dcb){ return 1;}
static int test_hangup(DCB *dcb){ return 1;}
static int test_accept(DCB *dcb){ return 1;}
static int test_connect(struct dcb *dcb, struct server *srv, struct session *ses){ return 1;}
static int test_close(DCB *dcb){ return 1;}
static int test_listen(DCB *dcb, char *config){ return 1;}
static int test_auth(DCB* dcb, struct server *srv, struct session *ses, GWBUF *buf){ return 1;}
static int test_session(DCB *dcb, void* data){ return 1;}
/**
* The "module object" for the httpd protocol module.
*/
static GWPROTOCOL MyObject = {
test_read, /**< Read - EPOLLIN handler */
test_write, /**< Write - data from gateway */
test_write_ready, /**< WriteReady - EPOLLOUT handler */
test_error, /**< Error - EPOLLERR handler */
test_hangup, /**< HangUp - EPOLLHUP handler */
test_accept, /**< Accept */
test_connect, /**< Connect */
test_close, /**< Close */
test_listen, /**< Create a listener */
test_auth, /**< Authentication */
test_session /**< Session */
};
/**
* 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()
{
}
/**
* 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
*/
GWPROTOCOL *
GetModuleObject()
{
return &MyObject;
}

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

@ -1,11 +1,10 @@
if(BUILD_TESTS)
add_subdirectory(test)
add_library(testroute SHARED testroute.c)
target_link_libraries(testroute log_manager utils)
install(TARGETS testroute DESTINATION modules)
endif()
add_library(testroute SHARED testroute.c)
target_link_libraries(testroute log_manager utils)
install(TARGETS testroute DESTINATION modules)
add_library(readconnroute SHARED readconnroute.c)
target_link_libraries(readconnroute log_manager utils)
install(TARGETS readconnroute DESTINATION modules)

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

@ -157,6 +157,10 @@ struct subcommand showoptions[] = {
"Show all configured servers",
"Show all configured servers",
{0, 0, 0} },
{ "serversjson", 0, dprintAllServersJson,
"Show all configured servers in JSON format",
"Show all configured servers in JSON format",
{0, 0, 0} },
{ "services", 0, dprintAllServices,
"Show all configured services in MaxScale",
"Show all configured services in MaxScale",

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

@ -261,7 +261,7 @@ static mysql_sescmd_t* sescmd_cursor_get_command(
static bool sescmd_cursor_next(
sescmd_cursor_t* scur);
static GWBUF* sescmd_cursor_process_replies(GWBUF* replybuf, backend_ref_t* bref);
static GWBUF* sescmd_cursor_process_replies(GWBUF* replybuf, backend_ref_t* bref,bool*);
static void tracelog_routed_query(
ROUTER_CLIENT_SES* rses,
@ -905,6 +905,15 @@ static void* newSession(
client_rses->rses_capabilities = RCAP_TYPE_STMT_INPUT;
client_rses->rses_backend_ref = backend_ref;
client_rses->rses_nbackends = router_nservers; /*< # of backend servers */
if(client_rses->rses_config.rw_max_slave_conn_percent)
{
int n_conn = 0;
double pct = (double)client_rses->rses_config.rw_max_slave_conn_percent / 100.0;
n_conn = MAX(floor((double)client_rses->rses_nbackends * pct),1);
client_rses->rses_config.rw_max_slave_conn_count = n_conn;
}
router->stats.n_sessions += 1;
/**
@ -2666,11 +2675,13 @@ static void clientReply (
DCB* backend_dcb)
{
DCB* client_dcb;
ROUTER_INSTANCE* router_inst;
ROUTER_CLIENT_SES* router_cli_ses;
sescmd_cursor_t* scur = NULL;
backend_ref_t* bref;
router_cli_ses = (ROUTER_CLIENT_SES *)router_session;
router_inst = (ROUTER_INSTANCE*)instance;
CHK_CLIENT_RSES(router_cli_ses);
/**
@ -2767,7 +2778,20 @@ static void clientReply (
* the client. Return with buffer including response that
* needs to be sent to client or NULL.
*/
writebuf = sescmd_cursor_process_replies(writebuf, bref);
bool rconn = false;
writebuf = sescmd_cursor_process_replies(writebuf, bref, &rconn);
if(rconn)
{
select_connect_backend_servers(&router_cli_ses->rses_master_ref,
router_cli_ses->rses_backend_ref,
router_cli_ses->rses_nbackends,
router_cli_ses->rses_config.rw_max_slave_conn_count,
router_cli_ses->rses_config.rw_max_slave_replication_lag,
router_cli_ses->rses_config.rw_slave_select_criteria,
router_cli_ses->rses_master_ref->bref_dcb->session,
router_cli_ses->router);
}
}
/**
* If response will be sent to client, decrease waiter count.
@ -3185,6 +3209,11 @@ static bool select_connect_backend_servers(
(master_host != NULL &&
(b->backend_server != master_host->backend_server)))
{
if(BREF_HAS_FAILED(&backend_ref[i]))
{
continue;
}
slaves_found += 1;
/** Slave is already connected */
@ -3674,15 +3703,19 @@ static void mysql_sescmd_done(
*/
static GWBUF* sescmd_cursor_process_replies(
GWBUF* replybuf,
backend_ref_t* bref)
backend_ref_t* bref,
bool *reconnect)
{
mysql_sescmd_t* scmd;
sescmd_cursor_t* scur;
ROUTER_CLIENT_SES* ses;
ROUTER_INSTANCE* router;
scur = &bref->bref_sescmd_cur;
ss_dassert(SPINLOCK_IS_LOCKED(&(scur->scmd_cur_rses->rses_lock)));
scmd = sescmd_cursor_get_command(scur);
ses = (*scur->scmd_cur_ptr_property)->rses_prop_rsession;
router = ses->router;
CHK_GWBUF(replybuf);
/**
@ -3691,6 +3724,7 @@ static GWBUF* sescmd_cursor_process_replies(
*/
while (scmd != NULL && replybuf != NULL)
{
bref->reply_cmd = *((unsigned char*)replybuf->start + 4);
/** Faster backend has already responded to client : discard */
if (scmd->my_sescmd_is_replied)
{
@ -3709,14 +3743,68 @@ static GWBUF* sescmd_cursor_process_replies(
}
/** Set response status received */
bref_clear_state(bref, BREF_WAITING_RESULT);
if(bref->reply_cmd != scmd->reply_cmd)
{
skygw_log_write(LOGFILE_TRACE,"Backend "
"server '%s' response differs from master's response. "
"Closing connection.",
bref->bref_backend->backend_server->unique_name);
sescmd_cursor_set_active(scur,false);
bref_clear_state(bref,BREF_QUERY_ACTIVE);
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);
*reconnect = true;
if(replybuf)
gwbuf_free(replybuf);
}
}
/** Response is in the buffer and it will be sent to client. */
else
/** This is a response from the master and it is the "right" one.
* A slave server's response will be compared to this and if
* their response differs from the master server's response, they
* are dropped from the valid list of backend servers.
* Response is in the buffer and it will be sent to client. */
else if(ses->rses_master_ref->bref_dcb == bref->bref_dcb)
{
/** Mark the rest session commands as replied */
scmd->my_sescmd_is_replied = true;
scmd->reply_cmd = *((unsigned char*)replybuf->start + 4);
skygw_log_write(LOGFILE_DEBUG,"Master '%s' responded to a session command.",
bref->bref_backend->backend_server->unique_name);
int i;
for(i=0;i<ses->rses_nbackends;i++)
{
if(!BREF_IS_WAITING_RESULT(&ses->rses_backend_ref[i]))
{
/** This backend has already received a response */
if(ses->rses_backend_ref[i].reply_cmd !=
scmd->reply_cmd &&
!BREF_IS_CLOSED(&ses->rses_backend_ref[i]))
{
bref_clear_state(&ses->rses_backend_ref[i],BREF_QUERY_ACTIVE);
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);
*reconnect = true;
}
}
}
}
else
{
skygw_log_write(LOGFILE_DEBUG,"Slave '%s' responded faster to a session command.",
bref->bref_backend->backend_server->unique_name);
if(replybuf)
gwbuf_free(replybuf);
return NULL;
}
if (sescmd_cursor_next(scur))
{
scmd = sescmd_cursor_get_command(scur);
@ -4217,7 +4305,7 @@ static bool route_session_write(
{
DCB* dcb = backend_ref[i].bref_dcb;
if (LOG_IS_ENABLED(LOGFILE_TRACE))
if (LOG_IS_ENABLED(LOGFILE_TRACE) && BREF_IS_IN_USE((&backend_ref[i])))
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,

41
server/modules/routing/testroute.c
View File

@ -33,9 +33,16 @@ static void *newSession(ROUTER *instance, SESSION *session);
static void closeSession(ROUTER *instance, void *session);
static void freeSession(ROUTER *instance, void *session);
static int routeQuery(ROUTER *instance, void *session, GWBUF *queue);
static void clientReply(ROUTER *instance, void *session, GWBUF *queue);
static void diagnostic(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);
static ROUTER_OBJECT MyObject = {
createInstance,
@ -44,11 +51,17 @@ static ROUTER_OBJECT MyObject = {
freeSession,
routeQuery,
diagnostic,
NULL,
NULL,
clientReply,
handleError,
getCapabilities
};
typedef struct{
}TESTROUTER;
typedef struct{
}TESTSESSION;
/**
* Implementation of the mandatory version entry point
*
@ -96,7 +109,8 @@ GetModuleObject()
static ROUTER *
createInstance(SERVICE *service, char **options)
{
return NULL;
return (ROUTER*)malloc(sizeof(TESTROUTER));
}
/**
@ -109,7 +123,7 @@ createInstance(SERVICE *service, char **options)
static void *
newSession(ROUTER *instance, SESSION *session)
{
return NULL;
return (SESSION*)malloc(sizeof(TESTSESSION));
}
/**
@ -128,7 +142,7 @@ static void freeSession(
ROUTER* router_instance,
void* router_client_session)
{
return;
free(router_client_session);
}
static int
@ -137,6 +151,10 @@ routeQuery(ROUTER *instance, void *session, GWBUF *queue)
return 0;
}
void clientReply(ROUTER* instance, void* session, GWBUF* queue)
{
}
/**
* Diagnostics routine
*
@ -154,3 +172,14 @@ static uint8_t getCapabilities(
{
return 0;
}
static void handleError(
ROUTER *instance,
void *router_session,
GWBUF *errbuf,
DCB *backend_dcb,
error_action_t action,
bool *succp)
{
}

12
utils/skygw_utils.cc
View File

@ -31,7 +31,7 @@
const char* timestamp_formatstr = "%04d-%02d-%02d %02d:%02d:%02d ";
/** One for terminating '\0' */
const int timestamp_len = 4+1 +2+1 +2+1 +2+1 +2+1 +2+3 +1;
const size_t timestamp_len = (4+1 +2+1 +2+1 +2+1 +2+1 +2+3 +1) * sizeof(char);
/** Single-linked list for storing test cases */
@ -662,7 +662,7 @@ bool mlist_cursor_move_to_first(
/** End of mlist */
int get_timestamp_len(void)
size_t get_timestamp_len(void)
{
return timestamp_len;
}
@ -682,13 +682,13 @@ int get_timestamp_len(void)
* @details (write detailed description here)
*
*/
int snprint_timestamp(
size_t snprint_timestamp(
char* p_ts,
int tslen)
size_t tslen)
{
time_t t;
struct tm tm;
int rval;
size_t rval;
if (p_ts == NULL) {
rval = 0;
@ -708,7 +708,7 @@ int snprint_timestamp(
tm.tm_min,
tm.tm_sec);
rval = strlen(p_ts);
rval = strlen(p_ts)*sizeof(char);
retblock:
return rval;
}

4
utils/skygw_utils.h
View File

@ -124,8 +124,8 @@ int skygw_thread_start(skygw_thread_t* thr);
skygw_thr_state_t skygw_thread_get_state(skygw_thread_t* thr);
pthread_t skygw_thread_gettid(skygw_thread_t* thr);
int get_timestamp_len(void);
int snprint_timestamp(char* p_ts, int tslen);
size_t get_timestamp_len(void);
size_t snprint_timestamp(char* p_ts, size_t tslen);
EXTERN_C_BLOCK_BEGIN