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

Use router template in schemarouter

Browse Source 
The schemarouter now uses the router template.
This commit is contained in:
Markus Mäkelä 2017-03-27 09:22:59 +03:00
parent b2ff0c5a0f
commit ac641e0f22
6 changed files with 2072 additions and 2200 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
server/modules/routing/schemarouter/CMakeLists.txt
View File

@ -1,4 +1,4 @@
add_library(schemarouter SHARED schemarouter.cc shard_map.cc session_command.cc)
add_library(schemarouter SHARED schemarouterinstance.cc schemaroutersession.cc shard_map.cc session_command.cc)
target_link_libraries(schemarouter maxscale-common)
add_dependencies(schemarouter pcre2)
set_target_properties(schemarouter PROPERTIES VERSION "1.0.0")

146
server/modules/routing/schemarouter/schemarouter.hh
View File

@ -14,112 +14,13 @@
#pragma once
/**
* @file schemarouter.hh - The schemarouter router module header file
* @file schemarouter.hh - Common schemarouter definitions
*/
#define MXS_MODULE_NAME "schemarouter"
#include <maxscale/cdefs.h>
#include <set>
#include <string>
#include <maxscale/dcb.h>
#include <maxscale/hashtable.h>
#include <maxscale/protocol/mysql.h>
#include <maxscale/pcre2.h>
#include "shard_map.hh"
#include "session_command.hh"
using std::string;
using std::set;
/**
* Bitmask values for the router session's initialization. These values are used
* to prevent responses from internal commands being forwarded to the client.
*/
typedef enum init_mask
{
INIT_READY = 0x00,
INIT_MAPPING = 0x01,
INIT_USE_DB = 0x02,
INIT_UNINT = 0x04,
INIT_FAILED = 0x08
} init_mask_t;
typedef enum showdb_response
{
SHOWDB_FULL_RESPONSE,
SHOWDB_PARTIAL_RESPONSE,
SHOWDB_DUPLICATE_DATABASES,
SHOWDB_FATAL_ERROR
} showdb_response_t;
/**
* The state of the backend server reference
*/
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_DB_MAPPED = 0x10
} bref_state_t;
#define BREF_IS_NOT_USED(s) ((s)->bref_state & ~BREF_IN_USE)
#define BREF_IS_IN_USE(s) ((s)->bref_state & BREF_IN_USE)
#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_IS_MAPPED(s) ((s)->bref_mapped)
#define SCHEMA_ERR_DUPLICATEDB 5000
#define SCHEMA_ERRSTR_DUPLICATEDB "DUPDB"
#define SCHEMA_ERR_DBNOTFOUND 1049
#define SCHEMA_ERRSTR_DBNOTFOUND "42000"
struct schemarouter_instance;
/**
* Route target types
*/
typedef enum
{
TARGET_UNDEFINED = (1 << 0),
TARGET_NAMED_SERVER = (1 << 1),
TARGET_ALL = (1 << 2),
TARGET_ANY = (1 << 3)
} route_target_t;
/** Helper macros for route target type */
#define TARGET_IS_UNDEFINED(t) (t == TARGET_UNDEFINED)
#define TARGET_IS_NAMED_SERVER(t) (t & TARGET_NAMED_SERVER)
#define TARGET_IS_ALL(t) (t & TARGET_ALL)
#define TARGET_IS_ANY(t) (t & TARGET_ANY)
/**
* Reference to BACKEND.
*
* Owned by router client session.
*/
typedef struct backend_ref_st
{
int n_mapping_eof;
GWBUF* map_queue;
SERVER_REF* bref_backend; /*< Backend server */
DCB* bref_dcb; /*< Backend DCB */
int bref_state; /*< State of the backend */
bool bref_mapped; /*< Whether the backend has been mapped */
bool last_sescmd_replied;
int bref_num_result_wait; /*< Number of not yet received results */
GWBUF* bref_pending_cmd; /*< Pending commands */
SessionCommandList session_commands; /**< List of session commands that are
* to be executed on this backend server */
} backend_ref_t;
/**
* Configuration values
*/
@ -147,48 +48,3 @@ typedef struct
int shmap_cache_hit; /*< Shard map was found from the cache */
int shmap_cache_miss; /*< No shard map found from the cache */
} ROUTER_STATS;
/**
* The per instance data for the router.
*/
typedef struct schemarouter_instance
{
ShardManager shard_manager; /*< Shard maps hashed by user name */
SERVICE* service; /*< Pointer to service */
SPINLOCK lock; /*< Lock for the instance data */
schemarouter_config_t schemarouter_config; /*< expanded config info from SERVICE */
int schemarouter_version; /*< version number for router's config */
ROUTER_STATS stats; /*< Statistics for this router */
set<string> ignored_dbs; /*< List of databases to ignore when the
* database mapping finds multiple servers
* with the same database */
pcre2_code* ignore_regex; /*< Databases matching this regex will
* not cause the session to be terminated
* if they are found on more than one server. */
pcre2_match_data* ignore_match_data;
} SCHEMAROUTER;
/**
* The client session structure used within this router.
*/
typedef struct schemarouter_session
{
bool closed; /*< true when closeSession is called */
DCB* rses_client_dcb;
MYSQL_session* rses_mysql_session; /*< Session client data (username, password, SHA1). */
backend_ref_t* rses_backend_ref; /*< Pointer to backend reference array */
schemarouter_config_t rses_config; /*< Copied config info from router instance */
int rses_nbackends; /*< Number of backends */
SCHEMAROUTER *router; /*< The router instance */
Shard shardmap; /**< Database hash containing names of the databases
* mapped to the servers that contain them */
string connect_db; /*< Database the user was trying to connect to */
string current_db; /*< Current active database */
int state; /*< Initialization state bitmask */
GWBUF* queue; /*< Query that was received before the session was ready */
ROUTER_STATS stats; /*< Statistics for this router */
uint64_t sent_sescmd; /**< The latest session command being executed */
uint64_t replied_sescmd; /**< The last session command reply that was sent to the client */
} SCHEMAROUTER_SESSION;

268
server/modules/routing/schemarouter/schemarouterinstance.cc Normal file
View File

@ -0,0 +1,268 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#include "schemarouter.hh"
#include "schemarouterinstance.hh"
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <maxscale/alloc.h>
#include <maxscale/buffer.h>
#include <maxscale/log_manager.h>
#include <maxscale/modinfo.h>
#include <maxscale/modutil.h>
#include <maxscale/poll.h>
#include <maxscale/query_classifier.h>
#include <maxscale/router.h>
#include <maxscale/secrets.h>
#include <maxscale/spinlock.h>
using std::string;
using std::map;
#define DEFAULT_REFRESH_INTERVAL "300"
/**
* @file schemarouter.c The entry points for the simple sharding router module.
*/
SchemaRouter::SchemaRouter(SERVICE *service, char **options):
mxs::Router<SchemaRouter, SchemaRouterSession>(service)
{
MXS_CONFIG_PARAMETER* conf;
MXS_CONFIG_PARAMETER* param;
/** Add default system databases to ignore */
this->ignored_dbs.insert("mysql");
this->ignored_dbs.insert("information_schema");
this->ignored_dbs.insert("performance_schema");
this->service = service;
this->stats.longest_sescmd = 0;
this->stats.n_hist_exceeded = 0;
this->stats.n_queries = 0;
this->stats.n_sescmd = 0;
this->stats.ses_longest = 0;
this->stats.ses_shortest = (double)((unsigned long)(~0));
spinlock_init(&this->lock);
conf = service->svc_config_param;
this->schemarouter_config.refresh_databases = config_get_bool(conf, "refresh_databases");
this->schemarouter_config.refresh_min_interval = config_get_integer(conf, "refresh_interval");
this->schemarouter_config.debug = config_get_bool(conf, "debug");
if ((config_get_param(conf, "auth_all_servers")) == NULL)
{
MXS_NOTICE("Authentication data is fetched from all servers. To disable this "
"add 'auth_all_servers=0' to the service.");
service->users_from_all = true;
}
if ((param = config_get_param(conf, "ignore_databases_regex")))
{
int errcode;
PCRE2_SIZE erroffset;
pcre2_code* re = pcre2_compile((PCRE2_SPTR)param->value, PCRE2_ZERO_TERMINATED, 0,
&errcode, &erroffset, NULL);
if (re == NULL)
{
PCRE2_UCHAR errbuf[512];
pcre2_get_error_message(errcode, errbuf, sizeof(errbuf));
MXS_ERROR("Regex compilation failed at %d for regex '%s': %s",
(int)erroffset, param->value, errbuf);
throw std::runtime_error("Regex compilation failed");
}
pcre2_match_data* match_data = pcre2_match_data_create_from_pattern(re, NULL);
if (match_data == NULL)
{
pcre2_code_free(re);
throw std::bad_alloc();
}
this->ignore_regex = re;
this->ignore_match_data = match_data;
}
if ((param = config_get_param(conf, "ignore_databases")))
{
char val[strlen(param->value) + 1];
strcpy(val, param->value);
const char *sep = ", \t";
char *sptr;
char *tok = strtok_r(val, sep, &sptr);
while (tok)
{
this->ignored_dbs.insert(tok);
tok = strtok_r(NULL, sep, &sptr);
}
}
bool failure = false;
for (int i = 0; options && options[i]; i++)
{
char* value = strchr(options[i], '=');
if (value == NULL)
{
MXS_ERROR("Unknown router options for %s", options[i]);
failure = true;
break;
}
*value = '\0';
value++;
if (strcmp(options[i], "max_sescmd_history") == 0)
{
MXS_WARNING("Use of 'max_sescmd_history' is deprecated");
}
else if (strcmp(options[i], "disable_sescmd_history") == 0)
{
MXS_WARNING("Use of 'disable_sescmd_history' is deprecated");
}
else if (strcmp(options[i], "refresh_databases") == 0)
{
this->schemarouter_config.refresh_databases = config_truth_value(value);
}
else if (strcmp(options[i], "refresh_interval") == 0)
{
this->schemarouter_config.refresh_min_interval = atof(value);
}
else if (strcmp(options[i], "debug") == 0)
{
this->schemarouter_config.debug = config_truth_value(value);
}
else
{
MXS_ERROR("Unknown router options for %s", options[i]);
failure = true;
break;
}
}
if (failure)
{
throw std::runtime_error("Failed to create schemarouter instance.");
}
}
SchemaRouter::~SchemaRouter()
{
if (this->ignore_regex)
{
pcre2_code_free(this->ignore_regex);
}
if (this->ignore_match_data)
{
pcre2_match_data_free(this->ignore_match_data);
}
}
SchemaRouter* SchemaRouter::create(SERVICE* pService, char** pzOptions)
{
return new SchemaRouter(pService, pzOptions);
}
SchemaRouterSession* SchemaRouter::newSession(MXS_SESSION* pSession)
{
return new SchemaRouterSession(pSession, *this);
}
void SchemaRouter::diagnostics(DCB* dcb)
{
double sescmd_pct = this->stats.n_sescmd != 0 ?
100.0 * ((double)this->stats.n_sescmd / (double)this->stats.n_queries) :
0.0;
/** Session command statistics */
dcb_printf(dcb, "\n\33[1;4mSession Commands\33[0m\n");
dcb_printf(dcb, "Total number of queries: %d\n",
this->stats.n_queries);
dcb_printf(dcb, "Percentage of session commands: %.2f\n",
sescmd_pct);
dcb_printf(dcb, "Longest chain of stored session commands: %d\n",
this->stats.longest_sescmd);
dcb_printf(dcb, "Session command history limit exceeded: %d times\n",
this->stats.n_hist_exceeded);
/** Session time statistics */
if (this->stats.sessions > 0)
{
dcb_printf(dcb, "\n\33[1;4mSession Time Statistics\33[0m\n");
dcb_printf(dcb, "Longest session: %.2lf seconds\n", this->stats.ses_longest);
dcb_printf(dcb, "Shortest session: %.2lf seconds\n", this->stats.ses_shortest);
dcb_printf(dcb, "Average session length: %.2lf seconds\n", this->stats.ses_average);
}
dcb_printf(dcb, "Shard map cache hits: %d\n", this->stats.shmap_cache_hit);
dcb_printf(dcb, "Shard map cache misses: %d\n", this->stats.shmap_cache_miss);
dcb_printf(dcb, "\n");
}
uint64_t SchemaRouter::getCapabilities()
{
return RCAP_TYPE_NONE;
}
MXS_BEGIN_DECLS
/**
* 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
*/
MXS_MODULE* MXS_CREATE_MODULE()
{
static MXS_MODULE info =
{
MXS_MODULE_API_ROUTER,
MXS_MODULE_BETA_RELEASE,
MXS_ROUTER_VERSION,
"A database sharding router for simple sharding",
"V1.0.0",
RCAP_TYPE_CONTIGUOUS_INPUT,
&SchemaRouter::s_object,
NULL, /* Process init. */
NULL, /* Process finish. */
NULL, /* Thread init. */
NULL, /* Thread finish. */
{
{"ignore_databases", MXS_MODULE_PARAM_STRING},
{"ignore_databases_regex", MXS_MODULE_PARAM_STRING},
{"max_sescmd_history", MXS_MODULE_PARAM_COUNT, "0"},
{"disable_sescmd_history", MXS_MODULE_PARAM_BOOL, "false"},
{"refresh_databases", MXS_MODULE_PARAM_BOOL, "true"},
{"refresh_interval", MXS_MODULE_PARAM_COUNT, DEFAULT_REFRESH_INTERVAL},
{"debug", MXS_MODULE_PARAM_BOOL, "false"},
{MXS_END_MODULE_PARAMS}
}
};
return &info;
}
MXS_END_DECLS

61
server/modules/routing/schemarouter/schemarouterinstance.hh Normal file
View File

@ -0,0 +1,61 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#pragma once
#include "schemarouter.hh"
#include <set>
#include <string>
#include <maxscale/router.hh>
#include <maxscale/pcre2.h>
#include "schemaroutersession.hh"
using std::string;
using std::set;
class SchemaRouterSession;
/**
* The per instance data for the router.
*/
class SchemaRouter: public mxs::Router<SchemaRouter, SchemaRouterSession>
{
public:
~SchemaRouter();
static SchemaRouter* create(SERVICE* pService, char** pzOptions);
SchemaRouterSession* newSession(MXS_SESSION* pSession);
void diagnostics(DCB* pDcb);
uint64_t getCapabilities();
protected:
friend class SchemaRouterSession;
schemarouter_config_t schemarouter_config; /*< expanded config info from SERVICE */
ShardManager shard_manager; /*< Shard maps hashed by user name */
SERVICE* service; /*< Pointer to service */
SPINLOCK lock; /*< Lock for the instance data */
int schemarouter_version; /*< version number for router's config */
ROUTER_STATS stats; /*< Statistics for this router */
set<string> ignored_dbs; /*< List of databases to ignore when the
* database mapping finds multiple servers
* with the same database */
pcre2_code* ignore_regex; /*< Databases matching this regex will
* not cause the session to be terminated
* if they are found on more than one server. */
pcre2_match_data* ignore_match_data;
SchemaRouter(SERVICE *service, char **options);
};

3597
server/modules/routing/schemarouter/schemarouter.cc → server/modules/routing/schemarouter/schemaroutersession.cc
View File

File diff suppressed because it is too large Load Diff

198
server/modules/routing/schemarouter/schemaroutersession.hh Normal file
View File

@ -0,0 +1,198 @@
/*
* Copyright (c) 2016 MariaDB Corporation Ab
*
* Use of this software is governed by the Business Source License included
* in the LICENSE.TXT file and at www.mariadb.com/bsl11.
*
* Change Date: 2019-07-01
*
* On the date above, in accordance with the Business Source License, use
* of this software will be governed by version 2 or later of the General
* Public License.
*/
#pragma once
#include "schemarouter.hh"
#include <string>
#include <maxscale/protocol/mysql.h>
#include <maxscale/router.hh>
#include "shard_map.hh"
#include "session_command.hh"
/**
* Bitmask values for the router session's initialization. These values are used
* to prevent responses from internal commands being forwarded to the client.
*/
typedef enum init_mask
{
INIT_READY = 0x00,
INIT_MAPPING = 0x01,
INIT_USE_DB = 0x02,
INIT_UNINT = 0x04,
INIT_FAILED = 0x08
} init_mask_t;
typedef enum showdb_response
{
SHOWDB_FULL_RESPONSE,
SHOWDB_PARTIAL_RESPONSE,
SHOWDB_DUPLICATE_DATABASES,
SHOWDB_FATAL_ERROR
} showdb_response_t;
/**
* The state of the backend server reference
*/
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_DB_MAPPED = 0x10
} bref_state_t;
#define BREF_IS_NOT_USED(s) ((s)->bref_state & ~BREF_IN_USE)
#define BREF_IS_IN_USE(s) ((s)->bref_state & BREF_IN_USE)
#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_IS_MAPPED(s) ((s)->bref_mapped)
#define SCHEMA_ERR_DUPLICATEDB 5000
#define SCHEMA_ERRSTR_DUPLICATEDB "DUPDB"
#define SCHEMA_ERR_DBNOTFOUND 1049
#define SCHEMA_ERRSTR_DBNOTFOUND "42000"
/**
* Route target types
*/
typedef enum
{
TARGET_UNDEFINED = (1 << 0),
TARGET_NAMED_SERVER = (1 << 1),
TARGET_ALL = (1 << 2),
TARGET_ANY = (1 << 3)
} route_target_t;
/** Helper macros for route target type */
#define TARGET_IS_UNDEFINED(t) (t == TARGET_UNDEFINED)
#define TARGET_IS_NAMED_SERVER(t) (t & TARGET_NAMED_SERVER)
#define TARGET_IS_ALL(t) (t & TARGET_ALL)
#define TARGET_IS_ANY(t) (t & TARGET_ANY)
/**
* Reference to BACKEND.
*
* Owned by router client session.
*/
typedef struct backend_ref_st
{
int n_mapping_eof;
GWBUF* map_queue;
SERVER_REF* bref_backend; /*< Backend server */
DCB* bref_dcb; /*< Backend DCB */
int bref_state; /*< State of the backend */
bool bref_mapped; /*< Whether the backend has been mapped */
bool last_sescmd_replied;
int bref_num_result_wait; /*< Number of not yet received results */
GWBUF* bref_pending_cmd; /*< Pending commands */
SessionCommandList session_commands; /**< List of session commands that are
* to be executed on this backend server */
} backend_ref_t;
class SchemaRouter;
/**
* The client session structure used within this router.
*/
class SchemaRouterSession: public mxs::RouterSession
{
public:
SchemaRouterSession(MXS_SESSION* session, SchemaRouter& router);
/**
* The RouterSession instance will be deleted when a client session
* has terminated. Will be called only after @c close() has been called.
*/
~SchemaRouterSession();
/**
* Called when a client session has been closed.
*/
void close();
/**
* Called when a packet being is routed to the backend. The router should
* forward the packet to the appropriate server(s).
*
* @param pPacket A client packet.
*/
int32_t routeQuery(GWBUF* pPacket);
/**
* Called when a packet is routed to the client. The router should
* forward the packet to the client using `MXS_SESSION_ROUTE_REPLY`.
*
* @param pPacket A client packet.
* @param pBackend The backend the packet is coming from.
*/
void clientReply(GWBUF* pPacket, DCB* pBackend);
/**
*
* @param pMessage The rror message.
* @param pProblem The DCB on which the error occurred.
* @param action The context.
* @param pSuccess On output, if false, the session will be terminated.
*/
void handleError(GWBUF* pMessage,
DCB* pProblem,
mxs_error_action_t action,
bool* pSuccess);
private:
bool closed; /*< true when closeSession is called */
DCB* rses_client_dcb;
MYSQL_session* rses_mysql_session; /*< Session client data (username, password, SHA1). */
backend_ref_t* rses_backend_ref; /*< Pointer to backend reference array */
schemarouter_config_t rses_config; /*< Copied config info from router instance */
int rses_nbackends; /*< Number of backends */
SchemaRouter& m_router; /*< The router instance */
Shard shardmap; /**< Database hash containing names of the databases
* mapped to the servers that contain them */
string connect_db; /*< Database the user was trying to connect to */
string current_db; /*< Current active database */
int state; /*< Initialization state bitmask */
GWBUF* queue; /*< Query that was received before the session was ready */
ROUTER_STATS stats; /*< Statistics for this router */
uint64_t sent_sescmd; /**< The latest session command being executed */
uint64_t replied_sescmd; /**< The last session command reply that was sent to the client */
/** Internal functions */
void synchronize_shard_map();
int inspect_backend_mapping_states(backend_ref_t *bref, GWBUF** wbuf);
bool route_session_write(GWBUF* querybuf, uint8_t command);
bool handle_error_new_connection(DCB* backend_dcb, GWBUF* errmsg);
void handle_error_reply_client(DCB* backend_dcb, GWBUF* errmsg);
int process_show_shards();
bool handle_default_db();
void route_queued_query();
bool get_shard_dcb(DCB** dcb, char* name);
SERVER* get_shard_target(GWBUF* buffer, uint32_t qtype);
backend_ref_t* get_bref_from_dcb(DCB* dcb);
bool have_servers();
bool send_database_list();
int gen_databaselist();
showdb_response_t parse_showdb_response(backend_ref_t* bref, GWBUF** buffer);
bool execute_sescmd_in_backend(backend_ref_t* backend_ref);
};