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

Merge branch 'release-1.0beta-refresh' into filter_harness

Browse Source
This commit is contained in:
Markus Makela
2014-09-17 11:07:12 +03:00
parent 6b54927638 c77807971d
commit 530d29f96d
158 changed files with 10243 additions and 1001 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
server/modules/filter/Makefile
View File

@ -21,15 +21,16 @@
include ../../../build_gateway.inc
LOGPATH := $(ROOT_PATH)/log_manager
QCLASSPATH := $(ROOT_PATH)/query_classifier
UTILSPATH := $(ROOT_PATH)/utils
CC=cc
CFLAGS=-c -fPIC -I/usr/include -I../include -I../../include -I$(LOGPATH) \
-I$(UTILSPATH) -Wall -g
CFLAGS=-c -fPIC -I/usr/include -I../include -I../../include -I$(LOGPATH) -I$(QCLASSPATH) \
-I$(UTILSPATH) -I$(MYSQL_ROOT) -Wall -g
include ../../../makefile.inc
LDFLAGS=-shared -L$(LOGPATH) -Wl,-rpath,$(DEST)/lib \
LDFLAGS=-shared -L$(LOGPATH) -L$(EMBEDDED_LIB) -L$(QCLASSPATH) -Wl,-rpath,$(DEST)/lib \
-Wl,-rpath,$(LOGPATH) -Wl,-rpath,$(UTILSPATH)
TESTSRCS=testfilter.c
@ -42,17 +43,31 @@ TOPNSRCS=topfilter.c
TOPNOBJ=$(TOPNSRCS:.c=.o)
TEESRCS=tee.c
TEEOBJ=$(TEESRCS:.c=.o)
MQSRCS=mqfilter.c
MQOBJ=$(MQSRCS:.c=.o)
SRCS=$(TESTSRCS) $(QLASRCS) $(REGEXSRCS) $(TOPNSRCS) $(TEESRCS)
OBJ=$(SRCS:.c=.o)
LIBS=$(UTILSPATH)/skygw_utils.o -lssl -llog_manager
MODULES= libtestfilter.so libqlafilter.so libregexfilter.so libtopfilter.so libtee.so libhintfilter.so
MODULES= libtestfilter.so libqlafilter.so libregexfilter.so libtopfilter.so libhintfilter.so libtee.so
ifndef BUILD_RABBITMQ
BUILD_RABBITMQ=N
endif
ifeq ($(BUILD_RABBITMQ),Y)
SRCS += $(MQSRCS)
MODULES += libmqfilter.so
LIBS += -lrabbitmq -lquery_classifier
endif
all: $(MODULES)
libtestfilter.so: $(TESTOBJ)
$(CC) $(LDFLAGS) $(TESTOBJ) $(LIBS) -o $@
libmqfilter.so: $(MQOBJ)
$(CC) $(LDFLAGS) $(MQOBJ) $(LIBS) -o $@
libqlafilter.so: $(QLAOBJ)
$(CC) $(LDFLAGS) $(QLAOBJ) $(LIBS) -o $@
@ -65,10 +80,9 @@ libtopfilter.so: $(TOPNOBJ)
libtee.so: $(TEEOBJ)
$(CC) $(LDFLAGS) $(TEEOBJ) $(LIBS) -o $@
libhintfilter.so:
libhintfilter.so:
(cd hint; touch depend.mk ; make; cp $@ ..)
.c.o:
$(CC) $(CFLAGS) $< -o $@
@ -81,7 +95,7 @@ tags:
(cd hint; touch depend.mk; make tags)
depend:
@$(DEL) depend.mk
@rm -f depend.mk
cc -M $(CFLAGS) $(SRCS) > depend.mk
(cd hint; touch depend.mk; make depend)
@ -90,14 +104,14 @@ install: $(MODULES)
cleantests:
$(MAKE) -C test cleantests
buildtests:
$(MAKE) -C test DEBUG=Y buildtests
runtests:
$(MAKE) -C test runtests
testall:
$(MAKE) -C test testall
include depend.mk

35
server/modules/filter/hint/hintparser.c
View File

@ -231,18 +231,9 @@ HINT_MODE mode = HM_EXECUTE;
goto retblock;
}
/** This is not MaxScale hint because it doesn't start with 'maxscale' */
if (tok->token != TOK_MAXSCALE)
{
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Error : Invalid hint string '%s'. Hint should start "
"with keyword 'maxscale'. Hint ignored.",
token_get_keyword(tok))));
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR,
"Error : Invalid hint string '%s'. Hint should start "
"with keyword 'maxscale'. Hint ignored.",
token_get_keyword(tok))));
token_free(tok);
goto retblock;
}
@ -270,6 +261,11 @@ HINT_MODE mode = HM_EXECUTE;
hint_pop(session);
state = HS_INIT;
break;
case TOK_START:
hintname = NULL;
mode = HM_START;
state = HS_INIT;
break;
default:
/* Error: expected hint, name or STOP */
LOGIF(LT, (skygw_log_write(
@ -450,7 +446,7 @@ HINT_MODE mode = HM_EXECUTE;
/*
* We are starting either a predefined set of hints,
* creating a new set of hints and starting in a single
* operation or starting an annonymous block of hints.
* operation or starting an anonymous block of hints.
*/
if (hintname == NULL && rval != NULL)
{
@ -532,6 +528,7 @@ hint_next_token(GWBUF **buf, char **ptr)
{
char word[100], *dest;
int inword = 0;
int endtag = 0;
char inquote = '\0';
int i, found;
HINT_TOKEN *tok;
@ -562,6 +559,18 @@ HINT_TOKEN *tok;
else if (**ptr == '\'')
inquote = **ptr;
/** Any other character which belongs to the word, move ahead */
else if(**ptr == '/' && endtag)
{
/** Comment end tag found, rewind the pointer back and return the token */
inword = 0;
(*ptr)--;
break;
}
else if(**ptr == '*' && !endtag)
{
endtag = 1;
}
else if (inword || (isspace(**ptr) == 0))
{
*dest++ = **ptr;
@ -583,7 +592,7 @@ HINT_TOKEN *tok;
/* We now have a word in the local word, check to see if it is a
* token we recognise.
*/
if (word[0] == '\0')
if (word[0] == '\0' || (word[0] == '*' && word[1] == '/'))
{
tok->token = TOK_EOL;
return tok;
@ -758,4 +767,4 @@ HINTSTACK* free_hint_stack(
{
return NULL;
}
}
}

1462
server/modules/filter/mqfilter.c Normal file
View File

File diff suppressed because it is too large Load Diff

14
server/modules/filter/tee.c
View File

@ -320,12 +320,14 @@ SESSION *bsession;
if (my_session->active)
{
bsession = my_session->branch_session;
router = bsession->service->router;
router_instance = bsession->service->router_instance;
rsession = bsession->router_session;
/** Close router session and all its connections */
router->closeSession(router_instance, rsession);
if ((bsession = my_session->branch_session) != NULL)
{
router = bsession->service->router;
router_instance = bsession->service->router_instance;
rsession = bsession->router_session;
/** Close router session and all its connections */
router->closeSession(router_instance, rsession);
}
dcb_free(my_session->branch_dcb);
/* No need to free the session, this is done as
* a side effect of closing the client DCB of the

367
server/modules/include/blr.h Normal file
View File

@ -0,0 +1,367 @@
#ifndef _BLR_H
#define _BLR_H
/*
* This file is distributed as part of 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 SkySQL Ab 2014
*/
/**
* @file blr.h - The binlog router header file
*
* @verbatim
* Revision History
*
* Date Who Description
* 02/04/14 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <dcb.h>
#include <buffer.h>
#include <pthread.h>
#define BINLOG_FNAMELEN 16
#define BLR_PROTOCOL "MySQLBackend"
#define BINLOG_MAGIC { 0xfe, 0x62, 0x69, 0x6e }
#define BINLOG_NAMEFMT "%s.%06d"
#define BINLOG_NAME_ROOT "mysql-bin"
/**
* High and Low water marks for the slave dcb. These values can be overriden
* by the router options highwater and lowwater.
*/
#define DEF_LOW_WATER 20000
#define DEF_HIGH_WATER 300000
/**
* Some useful macros for examining the MySQL Response packets
*/
#define MYSQL_RESPONSE_OK(buf) (*((uint8_t *)GWBUF_DATA(buf) + 4) == 0x00)
#define MYSQL_RESPONSE_EOF(buf) (*((uint8_t *)GWBUF_DATA(buf) + 4) == 0xfe)
#define MYSQL_RESPONSE_ERR(buf) (*((uint8_t *)GWBUF_DATA(buf) + 4) == 0xff)
#define MYSQL_ERROR_CODE(buf) (*((uint8_t *)GWBUF_DATA(buf) + 5))
#define MYSQL_ERROR_MSG(buf) ((uint8_t *)GWBUF_DATA(buf) + 6)
#define MYSQL_COMMAND(buf) (*((uint8_t *)GWBUF_DATA(buf) + 4))
/**
* Slave statistics
*/
typedef struct {
int n_events; /*< Number of events sent */
int n_bursts; /*< Number of bursts sent */
int n_requests; /*< Number of requests received */
int n_flows; /*< Number of flow control restarts */
int n_catchupnr; /*< No. of times catchup resulted in not entering loop */
int n_alreadyupd;
int n_upd;
int n_cb;
int n_cbna;
int n_dcb;
int n_above;
int n_failed_read;
int n_overrun;
int n_actions[3];
} SLAVE_STATS;
/**
* The client session structure used within this router. This represents
* the slaves that are replicating binlogs from MaxScale.
*/
typedef struct router_slave {
#if defined(SS_DEBUG)
skygw_chk_t rses_chk_top;
#endif
DCB *dcb; /*< The slave server DCB */
int state; /*< The state of this slave */
int binlog_pos; /*< Binlog position for this slave */
char binlogfile[BINLOG_FNAMELEN+1];
/*< Current binlog file for this slave */
int serverid; /*< Server-id of the slave */
char *hostname; /*< Hostname of the slave, if known */
char *user; /*< Username if given */
char *passwd; /*< Password if given */
short port; /*< MySQL port */
int nocrc; /*< Disable CRC */
int overrun;
uint32_t rank; /*< Replication rank */
uint8_t seqno; /*< Replication dump sequence no */
SPINLOCK catch_lock; /*< Event catchup lock */
unsigned int cstate; /*< Catch up state */
SPINLOCK rses_lock; /*< Protects rses_deleted */
pthread_t pthread;
struct router_instance
*router; /*< Pointer to the owning router */
struct router_slave *next;
SLAVE_STATS stats; /*< Slave statistics */
#if defined(SS_DEBUG)
skygw_chk_t rses_chk_tail;
#endif
} ROUTER_SLAVE;
/**
* The statistics for this router instance
*/
typedef struct {
int n_slaves; /*< Number slave sessions created */
int n_reads; /*< Number of record reads */
uint64_t n_binlogs; /*< Number of binlog records from master */
uint64_t n_binlog_errors;/*< Number of binlog records from master */
uint64_t n_rotates; /*< Number of binlog rotate events */
uint64_t n_cachehits; /*< Number of hits on the binlog cache */
uint64_t n_cachemisses; /*< Number of misses on the binlog cache */
int n_registered; /*< Number of registered slaves */
int n_masterstarts; /*< Number of times connection restarted */
int n_delayedreconnects;
int n_residuals; /*< Number of times residual data was buffered */
int n_heartbeats; /*< Number of heartbeat messages */
time_t lastReply;
uint64_t n_fakeevents; /*< Fake events not written to disk */
uint64_t n_artificial; /*< Artificial events not written to disk */
uint64_t events[0x24]; /*< Per event counters */
} ROUTER_STATS;
/**
* Saved responses from the master that will be forwarded to slaves
*/
typedef struct {
GWBUF *server_id; /*< Master server id */
GWBUF *heartbeat; /*< Heartbeat period */
GWBUF *chksum1; /*< Binlog checksum 1st response */
GWBUF *chksum2; /*< Binlog checksum 2nd response */
GWBUF *gtid_mode; /*< GTID Mode response */
GWBUF *uuid; /*< Master UUID */
GWBUF *setslaveuuid; /*< Set Slave UUID */
GWBUF *setnames; /*< Set NAMES latin1 */
GWBUF *utf8; /*< Set NAMES utf8 */
GWBUF *select1; /*< select 1 */
GWBUF *selectver; /*< select version() */
uint8_t *fde_event; /*< Format Description Event */
int fde_len; /*< Length of fde_event */
} MASTER_RESPONSES;
/**
* The binlog record structure. This contains the actual packet received from the
* master, the binlog position of the data in the packet, a point to the data and
* the length of the binlog record.
*
* This allows requests for binlog records in the cache to be serviced by simply
* sending the exact same packet as was received by MaxScale from the master.
* Items are written to the backing file as soon as they are received. The binlog
* cache is flushed of old records periodically, releasing the GWBUF's back to the
* free memory pool.
*/
typedef struct {
unsigned long position; /*< binlog record position for this cache entry */
GWBUF *pkt; /*< The packet received from the master */
unsigned char *data; /*< Pointer to the data within the packet */
unsigned int record_len; /*< Binlog record length */
} BLCACHE_RECORD;
/**
* The binlog cache. A cache exists for each file that hold cached bin log records.
* Typically the router will hold two binlog caches, one for the current file and one
* for the previous file.
*/
typedef struct {
char filename[BINLOG_FNAMELEN+1];
BLCACHE_RECORD *first;
BLCACHE_RECORD *current;
int cnt;
} BLCACHE;
/**
* The per instance data for the router.
*/
typedef struct router_instance {
SERVICE *service; /*< Pointer to the service using this router */
ROUTER_SLAVE *slaves; /*< Link list of all the slave connections */
SPINLOCK lock; /*< Spinlock for the instance data */
char *uuid; /*< UUID for the router to use w/master */
int masterid; /*< Server ID of the master */
int serverid; /*< Server ID to use with master */
char *user; /*< User name to use with master */
char *password; /*< Password to use with master */
char *fileroot; /*< Root of binlog filename */
DCB *master; /*< DCB for master connection */
DCB *client; /*< DCB for dummy client */
SESSION *session; /*< Fake session for master connection */
unsigned int master_state; /*< State of the master FSM */
uint8_t lastEventReceived;
GWBUF *residual; /*< Any residual binlog event */
MASTER_RESPONSES saved_master; /*< Saved master responses */
char binlog_name[BINLOG_FNAMELEN+1];
/*< Name of the current binlog file */
uint64_t binlog_position;
/*< Current binlog position */
int binlog_fd; /*< File descriptor of the binlog
* file being written
*/
unsigned int low_water; /*< Low water mark for client DCB */
unsigned int high_water; /*< High water mark for client DCB */
BLCACHE *cache[2];
ROUTER_STATS stats; /*< Statistics for this router */
int active_logs;
int reconnect_pending;
int handling_threads;
struct router_instance
*next;
} ROUTER_INSTANCE;
/**
* Packet header for replication messages
*/
typedef struct rep_header {
int payload_len; /*< Payload length (24 bits) */
uint8_t seqno; /*< Response sequence number */
uint8_t ok; /*< OK Byte from packet */
uint32_t timestamp; /*< Timestamp - start of binlog record */
uint8_t event_type; /*< Binlog event type */
uint32_t serverid; /*< Server id of master */
uint32_t event_size; /*< Size of header, post-header and body */
uint32_t next_pos; /*< Position of next event */
uint16_t flags; /*< Event flags */
} REP_HEADER;
/**
* State machine for the master to MaxScale replication
*/
#define BLRM_UNCONNECTED 0x0000
#define BLRM_AUTHENTICATED 0x0001
#define BLRM_TIMESTAMP 0x0002
#define BLRM_SERVERID 0x0003
#define BLRM_HBPERIOD 0x0004
#define BLRM_CHKSUM1 0x0005
#define BLRM_CHKSUM2 0x0006
#define BLRM_GTIDMODE 0x0007
#define BLRM_MUUID 0x0008
#define BLRM_SUUID 0x0009
#define BLRM_LATIN1 0x000A
#define BLRM_UTF8 0x000B
#define BLRM_SELECT1 0x000C
#define BLRM_SELECTVER 0x000D
#define BLRM_REGISTER 0x000E
#define BLRM_BINLOGDUMP 0x000F
#define BLRM_MAXSTATE 0x000F
static char *blrm_states[] = { "Unconnected", "Authenticated", "Timestamp retrieval",
"Server ID retrieval", "HeartBeat Period setup", "binlog checksum config",
"binlog checksum rerieval", "GTID Mode retrieval", "Master UUID retrieval",
"Set Slave UUID", "Set Names latin1", "Set Names utf8", "select 1",
"select version()", "Register slave", "Binlog Dump" };
#define BLRS_CREATED 0x0000
#define BLRS_UNREGISTERED 0x0001
#define BLRS_REGISTERED 0x0002
#define BLRS_DUMPING 0x0003
#define BLRS_MAXSTATE 0x0003
static char *blrs_states[] = { "Created", "Unregistered", "Registered",
"Sending binlogs" };
/**
* Slave catch-up status
*/
#define CS_READING 0x0001
#define CS_INNERLOOP 0x0002
#define CS_UPTODATE 0x0004
#define CS_EXPECTCB 0x0008
#define CS_DIST 0x0010
#define CS_DISTLATCH 0x0020
/**
* MySQL protocol OpCodes needed for replication
*/
#define COM_QUIT 0x01
#define COM_QUERY 0x03
#define COM_REGISTER_SLAVE 0x15
#define COM_BINLOG_DUMP 0x12
/**
* Binlog event types
*/
#define START_EVENT_V3 0x01
#define QUERY_EVENT 0x02
#define STOP_EVENT 0x03
#define ROTATE_EVENT 0x04
#define INTVAR_EVENT 0x05
#define LOAD_EVENT 0x06
#define SLAVE_EVENT 0x07
#define CREATE_FILE_EVENT 0x08
#define APPEND_BLOCK_EVENT 0x09
#define EXEC_LOAD_EVENT 0x0A
#define DELETE_FILE_EVENT 0x0B
#define NEW_LOAD_EVENT 0x0C
#define RAND_EVENT 0x0D
#define USER_VAR_EVENT 0x0E
#define FORMAT_DESCRIPTION_EVENT 0x0F
#define XID_EVENT 0x10
#define BEGIN_LOAD_QUERY_EVENT 0x11
#define EXECUTE_LOAD_QUERY_EVENT 0x12
#define TABLE_MAP_EVENT 0x13
#define WRITE_ROWS_EVENTv0 0x14
#define UPDATE_ROWS_EVENTv0 0x15
#define DELETE_ROWS_EVENTv0 0x16
#define WRITE_ROWS_EVENTv1 0x17
#define UPDATE_ROWS_EVENTv1 0x18
#define DELETE_ROWS_EVENTv1 0x19
#define INCIDENT_EVENT 0x1A
#define HEARTBEAT_EVENT 0x1B
#define IGNORABLE_EVENT 0x1C
#define ROWS_QUERY_EVENT 0x1D
#define WRITE_ROWS_EVENTv2 0x1E
#define UPDATE_ROWS_EVENTv2 0x1F
#define DELETE_ROWS_EVENTv2 0x20
#define GTID_EVENT 0x21
#define ANONYMOUS_GTID_EVENT 0x22
#define PREVIOUS_GTIDS_EVENT 0x23
/**
* Binlog event flags
*/
#define LOG_EVENT_BINLOG_IN_USE_F 0x0001
#define LOG_EVENT_FORCED_ROTATE_F 0x0002
#define LOG_EVENT_THREAD_SPECIFIC_F 0x0004
#define LOG_EVENT_SUPPRESS_USE_F 0x0008
#define LOG_EVENT_UPDATE_TABLE_MAP_VERSION_F 0x0010
#define LOG_EVENT_ARTIFICIAL_F 0x0020
#define LOG_EVENT_RELAY_LOG_F 0x0040
#define LOG_EVENT_IGNORABLE_F 0x0080
#define LOG_EVENT_NO_FILTER_F 0x0100
#define LOG_EVENT_MTS_ISOLATE_F 0x0200
/*
* Externals within the router
*/
extern void blr_start_master(ROUTER_INSTANCE *);
extern void blr_master_response(ROUTER_INSTANCE *, GWBUF *);
extern void blr_master_reconnect(ROUTER_INSTANCE *);
extern int blr_slave_request(ROUTER_INSTANCE *, ROUTER_SLAVE *, GWBUF *);
extern void blr_slave_rotate(ROUTER_SLAVE *slave, uint8_t *ptr);
extern int blr_slave_catchup(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave);
extern void blr_init_cache(ROUTER_INSTANCE *);
extern void blr_file_init(ROUTER_INSTANCE *);
extern int blr_open_binlog(ROUTER_INSTANCE *, char *);
extern void blr_write_binlog_record(ROUTER_INSTANCE *, REP_HEADER *,uint8_t *);
extern void blr_file_rotate(ROUTER_INSTANCE *, char *, uint64_t);
extern void blr_file_flush(ROUTER_INSTANCE *);
extern GWBUF *blr_read_binlog(int, unsigned int, REP_HEADER *);
#endif

6
server/modules/include/mysql_client_server_protocol.h
View File

@ -82,10 +82,6 @@
#endif
#define GW_NOINTR_CALL(A) do { errno = 0; A; } while (errno == EINTR)
// network buffer is 32K
#define MAX_BUFFER_SIZE 32768
// socket send buffer for backend
#define GW_BACKEND_SO_SNDBUF 1024
#define SMALL_CHUNK 1024
#define MAX_CHUNK SMALL_CHUNK * 8 * 4
#define ToHex(Y) (Y>='0'&&Y<='9'?Y-'0':Y-'A'+10)
@ -219,7 +215,7 @@ typedef enum mysql_server_cmd {
MYSQL_COM_QUERY,
MYSQL_COM_FIELD_LIST,
MYSQL_COM_CREATE_DB,
MYSQL_COM_DROP_DB,
MYSQL_COM_DROP_DB,
MYSQL_COM_REFRESH,
MYSQL_COM_SHUTDOWN,
MYSQL_COM_STATISTICS,

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

@ -71,6 +71,8 @@ typedef enum backend_type_t {
BE_COUNT
} backend_type_t;
struct router_instance;
typedef enum {
TARGET_MASTER = 0x01,
TARGET_SLAVE = 0x02,
@ -118,6 +120,7 @@ typedef enum select_criteria {
/** default values for rwsplit configuration parameters */
#define CONFIG_MAX_SLAVE_CONN 1
#define CONFIG_MAX_SLAVE_RLAG -1 /*< not used */
#define CONFIG_SQL_VARIABLES_IN TYPE_ALL
#define GET_SELECT_CRITERIA(s) \
(strncmp(s,"LEAST_GLOBAL_CONNECTIONS", strlen("LEAST_GLOBAL_CONNECTIONS")) == 0 ? \
@ -232,6 +235,7 @@ typedef struct rwsplit_config_st {
int rw_max_slave_conn_count;
select_criteria_t rw_slave_select_criteria;
int rw_max_slave_replication_lag;
target_t rw_use_sql_variables_in;
} rwsplit_config_t;
@ -276,6 +280,7 @@ struct router_client_session {
#if defined(PREP_STMT_CACHING)
HASHTABLE* rses_prep_stmt[2];
#endif
struct router_instance *router; /*< The router instance */
struct router_client_session* next;
#if defined(SS_DEBUG)
skygw_chk_t rses_chk_tail;
@ -309,6 +314,8 @@ typedef struct router_instance {
unsigned int bitvalue; /*< Required value of server->status */
ROUTER_STATS stats; /*< Statistics for this router */
struct router_instance* next; /*< Next router on the list */
bool available_slaves;
/*< The router has some slaves avialable */
} ROUTER_INSTANCE;
#define BACKEND_TYPE(b) (SERVER_IS_MASTER((b)->backend_server) ? BE_MASTER : \

3
server/modules/monitor/.gitignore vendored
View File

@ -1,3 +0,0 @@
*.o
*.so
depend.mk

14
server/modules/monitor/galera_mon.c
View File

@ -69,7 +69,7 @@ static void defaultUsers(void *, char *, char *);
static void diagnostics(DCB *, void *);
static void setInterval(void *, unsigned long);
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUsers, diagnostics, setInterval, NULL, NULL };
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUsers, diagnostics, setInterval, NULL, NULL, NULL };
/**
* Implementation of the mandatory version entry point
@ -335,10 +335,18 @@ char *server_string;
database->server->port,
mysql_error(database->con))));
server_clear_status(database->server, SERVER_RUNNING);
if (mysql_errno(database->con) == ER_ACCESS_DENIED_ERROR)
{
server_set_status(database->server, SERVER_AUTH_ERROR);
}
database->server->node_id = -1;
free(dpwd);
return;
}
else
{
server_clear_status(database->server, SERVER_AUTH_ERROR);
}
free(dpwd);
}
@ -351,7 +359,9 @@ char *server_string;
/* get server version string */
server_string = (char *)mysql_get_server_info(database->con);
if (server_string) {
database->server->server_string = strdup(server_string);
database->server->server_string = realloc(database->server->server_string, strlen(server_string)+1);
if (database->server->server_string)
strcpy(database->server->server_string, server_string);
}
/* Check if the the Galera FSM shows this node is joined to the cluster */

94
server/modules/monitor/mysql_mon.c
View File

@ -40,6 +40,9 @@
* the status to update in server status field before
* starting the replication consistency check.
* This will also give routers a consistent "status" of all servers
* 28/08/14 Massimiliano Pinto Added detectStaleMaster feature: previous detected master will be used again, even if the replication is stopped.
* This means both IO and SQL threads are not working on slaves.
* This option is not enabled by default.
*
* @endverbatim
*/
@ -62,7 +65,7 @@ extern int lm_enabled_logfiles_bitmask;
static void monitorMain(void *);
static char *version_str = "V1.2.0";
static char *version_str = "V1.3.0";
MODULE_INFO info = {
MODULE_API_MONITOR,
@ -80,6 +83,7 @@ static void diagnostics(DCB *, void *);
static void setInterval(void *, unsigned long);
static void defaultId(void *, unsigned long);
static void replicationHeartbeat(void *, int);
static void detectStaleMaster(void *, int);
static bool mon_status_changed(MONITOR_SERVERS* mon_srv);
static bool mon_print_fail_status(MONITOR_SERVERS* mon_srv);
static MONITOR_SERVERS *getServerByNodeId(MONITOR_SERVERS *, long);
@ -91,7 +95,7 @@ static int add_slave_to_master(long *, int, long);
static void monitor_set_pending_status(MONITOR_SERVERS *, int);
static void monitor_clear_pending_status(MONITOR_SERVERS *, int);
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUser, diagnostics, setInterval, defaultId, replicationHeartbeat };
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUser, diagnostics, setInterval, defaultId, replicationHeartbeat, detectStaleMaster };
/**
* Implementation of the mandatory version entry point
@ -160,6 +164,7 @@ MYSQL_MONITOR *handle;
handle->id = MONITOR_DEFAULT_ID;
handle->interval = MONITOR_INTERVAL;
handle->replicationHeartbeat = 0;
handle->detectStaleMaster = 0;
handle->master = NULL;
spinlock_init(&handle->lock);
}
@ -306,6 +311,7 @@ char *sep;
dcb_printf(dcb,"\tSampling interval:\t%lu milliseconds\n", handle->interval);
dcb_printf(dcb,"\tMaxScale MonitorId:\t%lu\n", handle->id);
dcb_printf(dcb,"\tReplication lag:\t%s\n", (handle->replicationHeartbeat == 1) ? "enabled" : "disabled");
dcb_printf(dcb,"\tDetect Stale Master:\t%s\n", (handle->detectStaleMaster == 1) ? "enabled" : "disabled");
dcb_printf(dcb, "\tMonitored servers: ");
db = handle->databases;
@ -394,6 +400,11 @@ char *server_string;
* Store server NOT running in server and monitor server pending struct
*
*/
if (mysql_errno(database->con) == ER_ACCESS_DENIED_ERROR)
{
server_set_status(database->server, SERVER_AUTH_ERROR);
monitor_set_pending_status(database, SERVER_AUTH_ERROR);
}
server_clear_status(database->server, SERVER_RUNNING);
monitor_clear_pending_status(database, SERVER_RUNNING);
@ -403,8 +414,19 @@ char *server_string;
monitor_clear_pending_status(database, SERVER_SLAVE);
monitor_clear_pending_status(database, SERVER_MASTER);
/* Clean addition status too */
server_clear_status(database->server, SERVER_SLAVE_OF_EXTERNAL_MASTER);
server_clear_status(database->server, SERVER_STALE_STATUS);
monitor_clear_pending_status(database, SERVER_SLAVE_OF_EXTERNAL_MASTER);
monitor_clear_pending_status(database, SERVER_STALE_STATUS);
return;
}
else
{
server_clear_status(database->server, SERVER_AUTH_ERROR);
monitor_clear_pending_status(database, SERVER_AUTH_ERROR);
}
free(dpwd);
}
/* Store current status in both server and monitor server pending struct */
@ -417,7 +439,9 @@ char *server_string;
/* get server version string */
server_string = (char *)mysql_get_server_info(database->con);
if (server_string) {
database->server->server_string = strdup(server_string);
database->server->server_string = realloc(database->server->server_string, strlen(server_string)+1);
if (database->server->server_string)
strcpy(database->server->server_string, server_string);
}
/* get server_id form current node */
@ -458,12 +482,20 @@ char *server_string;
if (strncmp(row[12], "Yes", 3) == 0
&& strncmp(row[13], "Yes", 3) == 0) {
isslave += 1;
}
/* If Slave_IO_Running = Yes, assign the master_id to current server: this allows building
* the replication tree, slaves ids will be added to master(s) and we will have at least the
* root master server.
* Please note, there could be no slaves at all if Slave_SQL_Running == 'No'
*/
if (strncmp(row[12], "Yes", 3) == 0) {
/* get Master_Server_Id values */
master_id = atol(row[41]);
if (master_id == 0)
master_id = -1;
}
i++;
}
/* store master_id of current node */
@ -489,7 +521,14 @@ char *server_string;
if (strncmp(row[10], "Yes", 3) == 0
&& strncmp(row[11], "Yes", 3) == 0) {
isslave = 1;
}
/* If Slave_IO_Running = Yes, assign the master_id to current server: this allows building
* the replication tree, slaves ids will be added to master(s) and we will have at least the
* root master server.
* Please note, there could be no slaves at all if Slave_SQL_Running == 'No'
*/
if (strncmp(row[10], "Yes", 3) == 0) {
/* get Master_Server_Id values */
master_id = atol(row[39]);
if (master_id == 0)
@ -505,6 +544,7 @@ char *server_string;
/* Remove addition info */
monitor_clear_pending_status(database, SERVER_SLAVE_OF_EXTERNAL_MASTER);
monitor_clear_pending_status(database, SERVER_STALE_STATUS);
/* Please note, the MASTER status and SERVER_SLAVE_OF_EXTERNAL_MASTER
* will be assigned in the monitorMain() via get_replication_tree() routine
@ -534,6 +574,7 @@ monitorMain(void *arg)
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
MONITOR_SERVERS *ptr;
int replication_heartbeat = handle->replicationHeartbeat;
int detect_stale_master = handle->detectStaleMaster;
int num_servers=0;
MONITOR_SERVERS *root_master;
@ -545,6 +586,7 @@ MONITOR_SERVERS *root_master;
"module. Exiting.\n")));
return;
}
handle->status = MONITOR_RUNNING;
while (1)
{
@ -616,10 +658,19 @@ MONITOR_SERVERS *root_master;
while (ptr)
{
if (! SERVER_IN_MAINT(ptr->server)) {
ptr->server->status = ptr->pending_status;
/* If "detect_stale_master" option is On, let's use the previus master */
if (detect_stale_master && root_master && (!strcmp(ptr->server->name, root_master->server->name) && ptr->server->port == root_master->server->port) && (ptr->server->status & SERVER_MASTER) && !(ptr->pending_status & SERVER_MASTER)) {
/* in this case server->status will not be updated from pending_status */
LOGIF(LM, (skygw_log_write_flush(
LOGFILE_MESSAGE, "[mysql_mon]: root server [%s:%i] is no longer Master, let's use it again even if it could be a stale master, you have been warned!", ptr->server->name, ptr->server->port)));
/* Set the STALE bit for this server in server struct */
server_set_status(ptr->server, SERVER_STALE_STATUS);
} else {
ptr->server->status = ptr->pending_status;
}
}
ptr = ptr->next;
}
ptr = ptr->next;
}
/* Do now the heartbeat replication set/get for MySQL Replication Consistency */
if (replication_heartbeat && root_master && (SERVER_IS_MASTER(root_master->server) || SERVER_IS_RELAY_SERVER(root_master->server))) {
@ -665,19 +716,34 @@ setInterval(void *arg, unsigned long interval)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->interval, &interval, sizeof(unsigned long));
}
}
/**
* Enable/Disable the MySQL Replication hearbeat, detecting slave lag behind master.
*
* @param arg The handle allocated by startMonitor
* @param replicationHeartbeat To enable it 1, disable it with 0
* @param arg The handle allocated by startMonitor
* @param enable To enable it 1, disable it with 0
*/
static void
replicationHeartbeat(void *arg, int replicationHeartbeat)
replicationHeartbeat(void *arg, int enable)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->replicationHeartbeat, &replicationHeartbeat, sizeof(int));
memcpy(&handle->replicationHeartbeat, &enable, sizeof(int));
}
/**
* Enable/Disable the MySQL Replication Stale Master dectection, allowing a previouvsly detected master to still act as a Master.
* This option must be enabled in order to keep the Master when the replication is stopped or removed from slaves.
* If the replication is still stopped when MaxSclale is restarted no Master will be available.
*
* @param arg The handle allocated by startMonitor
* @param enable To enable it 1, disable it with 0
*/
static void
detectStaleMaster(void *arg, int enable)
{
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
memcpy(&handle->detectStaleMaster, &enable, sizeof(int));
}
static bool mon_status_changed(
@ -1038,6 +1104,10 @@ static MONITOR_SERVERS *get_replication_tree(MYSQL_MONITOR *handle, int num_serv
monitor_set_pending_status(master, SERVER_MASTER);
} else {
if (current->master_id > 0) {
/* this server is slave of another server not in MaxScale configuration
* we cannot use it as a real slave.
*/
monitor_clear_pending_status(ptr, SERVER_SLAVE);
monitor_set_pending_status(ptr, SERVER_SLAVE_OF_EXTERNAL_MASTER);
}
}

28
server/modules/monitor/mysqlmon.h
View File

@ -32,6 +32,7 @@
* 26/05/14 Massimiliano Pinto Default values for MONITOR_INTERVAL
* 28/05/14 Massimiliano Pinto Addition of new fields in MYSQL_MONITOR struct
* 24/06/14 Massimiliano Pinto Addition of master field in MYSQL_MONITOR struct and MONITOR_MAX_NUM_SLAVES
* 28/08/14 Massimiliano Pinto Addition of detectStaleMaster
*
* @endverbatim
*/
@ -43,9 +44,9 @@
typedef struct monitor_servers {
SERVER *server; /**< The server being monitored */
MYSQL *con; /**< The MySQL connection */
int mon_err_count;
unsigned int mon_prev_status;
unsigned int pending_status; /**< Pending Status flag bitmap */
int mon_err_count;
unsigned int mon_prev_status;
unsigned int pending_status; /**< Pending Status flag bitmap */
struct monitor_servers
*next; /**< The next server in the list */
} MONITOR_SERVERS;
@ -54,17 +55,18 @@ typedef struct monitor_servers {
* The handle for an instance of a MySQL Monitor module
*/
typedef struct {
SPINLOCK lock; /**< The monitor spinlock */
pthread_t tid; /**< id of monitor thread */
int shutdown; /**< Flag to shutdown the monitor thread */
int status; /**< Monitor status */
char *defaultUser; /**< Default username for monitoring */
char *defaultPasswd; /**< Default password for monitoring */
unsigned long interval; /**< Monitor sampling interval */
unsigned long id; /**< Monitor ID */
SPINLOCK lock; /**< The monitor spinlock */
pthread_t tid; /**< id of monitor thread */
int shutdown; /**< Flag to shutdown the monitor thread */
int status; /**< Monitor status */
char *defaultUser; /**< Default username for monitoring */
char *defaultPasswd; /**< Default password for monitoring */
unsigned long interval; /**< Monitor sampling interval */
unsigned long id; /**< Monitor ID */
int replicationHeartbeat; /**< Monitor flag for MySQL replication heartbeat */
MONITOR_SERVERS *master; /**< Master server for MySQL Master/Slave replication */
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
int detectStaleMaster; /**< Monitor flag for MySQL replication Stale Master detection */
MONITOR_SERVERS *master; /**< Master server for MySQL Master/Slave replication */
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
} MYSQL_MONITOR;
#define MONITOR_RUNNING 1

14
server/modules/monitor/ndbcluster_mon.c
View File

@ -63,7 +63,7 @@ static void defaultUsers(void *, char *, char *);
static void diagnostics(DCB *, void *);
static void setInterval(void *, unsigned long);
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUsers, diagnostics, setInterval, NULL, NULL };
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUsers, diagnostics, setInterval, NULL, NULL, NULL };
/**
* Implementation of the mandatory version entry point
@ -329,10 +329,18 @@ char *server_string;
database->server->port,
mysql_error(database->con))));
server_clear_status(database->server, SERVER_RUNNING);
if (mysql_errno(database->con) == ER_ACCESS_DENIED_ERROR)
{
server_set_status(database->server, SERVER_AUTH_ERROR);
}
database->server->node_id = -1;
free(dpwd);
return;
}
else
{
server_clear_status(database->server, SERVER_AUTH_ERROR);
}
free(dpwd);
}
@ -345,7 +353,9 @@ char *server_string;
/* get server version string */
server_string = (char *)mysql_get_server_info(database->con);
if (server_string) {
database->server->server_string = strdup(server_string);
database->server->server_string = realloc(database->server->server_string, strlen(server_string)+1);
if (database->server->server_string)
strcpy(database->server->server_string, server_string);
}
/* Check if the the SQL node is able to contact one or more data nodes */

3
server/modules/protocol/.gitignore vendored
View File

@ -1,3 +0,0 @@
*.o
*.so
depend.mk

5
server/modules/protocol/maxscaled.c
View File

@ -235,6 +235,7 @@ maxscaled_error(DCB *dcb)
static int
maxscaled_hangup(DCB *dcb)
{
dcb_close(dcb);
return 0;
}
@ -313,9 +314,11 @@ maxscaled_close(DCB *dcb)
MAXSCALED *maxscaled = dcb->protocol;
if (maxscaled && maxscaled->username)
{
free(maxscaled->username);
maxscaled->username = NULL;
}
dcb_close(dcb);
return 0;
}

3
server/modules/protocol/mysql_backend.c
View File

@ -492,11 +492,12 @@ static int gw_read_backend_event(DCB *dcb) {
dcb->dcb_readqueue = NULL;
}
}
/** This may be either short prefix of a packet, or the tail of it. */
else
{
if (nbytes_read < 5)
{
gwbuf_append(dcb->dcb_readqueue, read_buffer);
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, read_buffer);
rc = 0;
goto return_rc;
}

40
server/modules/protocol/mysql_client.c
View File

@ -34,6 +34,7 @@
* 28/02/2014 Massimiliano Pinto Added: client IPv4 in dcb->ipv4 and inet_ntop for string representation
* 11/03/2014 Massimiliano Pinto Added: Unix socket support
* 07/05/2014 Massimiliano Pinto Added: specific version string in server handshake
* 09/09/2014 Massimiliano Pinto Added: 777 permission for socket path
*
*/
#include <skygw_utils.h>
@ -41,6 +42,7 @@
#include <mysql_client_server_protocol.h>
#include <gw.h>
#include <modinfo.h>
#include <sys/stat.h>
MODULE_INFO info = {
MODULE_API_PROTOCOL,
@ -551,15 +553,16 @@ int gw_read_client_event(
*/
if (dcb->dcb_readqueue)
{
uint8_t* data = (uint8_t *)GWBUF_DATA(read_buffer);
uint8_t* data;
read_buffer = gwbuf_append(dcb->dcb_readqueue, read_buffer);
nbytes_read = gwbuf_length(read_buffer);
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, read_buffer);
nbytes_read = gwbuf_length(dcb->dcb_readqueue);
data = (uint8_t *)GWBUF_DATA(dcb->dcb_readqueue);
if (nbytes_read < 3 || nbytes_read < MYSQL_GET_PACKET_LEN(data))
{
rc = 0;
goto return_rc;
rc = 0;
goto return_rc;
}
else
{
@ -577,8 +580,8 @@ int gw_read_client_event(
if (nbytes_read < 3 || nbytes_read < MYSQL_GET_PACKET_LEN(data)+4)
{
gwbuf_append(dcb->dcb_readqueue, read_buffer);
rc = 0;
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, read_buffer);
rc = 0;
goto return_rc;
}
}
@ -788,7 +791,7 @@ int gw_read_client_event(
if (read_buffer != NULL)
{
/** add incomplete mysql packet to read queue */
gwbuf_append(dcb->dcb_readqueue, read_buffer);
dcb->dcb_readqueue = gwbuf_append(dcb->dcb_readqueue, read_buffer);
}
}
else
@ -798,7 +801,7 @@ int gw_read_client_event(
}
/** succeed */
if (rc == 1) {
if (rc) {
rc = 0; /**< here '0' means success */
} else {
GWBUF* errbuf;
@ -985,6 +988,16 @@ int gw_MySQLListener(
return 0;
}
/* set permission for all users */
if (chmod(config_bind, 0777) < 0) {
fprintf(stderr,
"\n* chmod failed for %s due error %i, %s.\n\n",
config_bind,
errno,
strerror(errno));
}
break;
case AF_INET:
@ -1167,7 +1180,10 @@ int gw_MySQLAccept(DCB *listener)
conn_open[c_sock] = true;
#endif
/* set nonblocking */
sendbuf = GW_BACKEND_SO_SNDBUF;
setsockopt(c_sock, SOL_SOCKET, SO_SNDBUF, &sendbuf, optlen);
sendbuf = GW_BACKEND_SO_RCVBUF;
setsockopt(c_sock, SOL_SOCKET, SO_RCVBUF, &sendbuf, optlen);
setnonblocking(c_sock);
client_dcb = dcb_alloc(DCB_ROLE_REQUEST_HANDLER);
@ -1432,7 +1448,11 @@ static int route_by_statement(
do
{
ss_dassert(GWBUF_IS_TYPE_MYSQL((*p_readbuf)));
/**
* Collect incoming bytes to a buffer until complete packet has
* arrived and then return the buffer.
*/
packetbuf = gw_MySQL_get_next_packet(p_readbuf);
if (packetbuf != NULL)

28
server/modules/protocol/mysql_common.c
View File

@ -26,9 +26,15 @@
* 04/09/2013 Massimiliano Pinto Added dcb NULL assert in mysql_send_custom_error
* 12/09/2013 Massimiliano Pinto Added checks in gw_decode_mysql_server_handshake and gw_read_backend_handshake
* 10/02/2014 Massimiliano Pinto Added MySQL Authentication with user@host
* 10/09/2014 Massimiliano Pinto Added MySQL Authentication option enabling localhost match with any host (wildcard %)
* Backend server configuration may differ so default is 0, don't match and an explicit
* localhost entry should be added for the selected user in the backends.
* Setting to 1 allow localhost (127.0.0.1 or socket) to match the any host grant via
* user@%
*
*/
#include <gw.h>
#include "mysql_client_server_protocol.h"
#include <skygw_types.h>
#include <skygw_utils.h>
@ -736,6 +742,7 @@ int gw_do_connect_to_backend(
struct sockaddr_in serv_addr;
int rv;
int so = 0;
int bufsize;
memset(&serv_addr, 0, sizeof serv_addr);
serv_addr.sin_family = AF_INET;
@ -759,6 +766,10 @@ int gw_do_connect_to_backend(
/* prepare for connect */
setipaddress(&serv_addr.sin_addr, host);
serv_addr.sin_port = htons(port);
bufsize = GW_CLIENT_SO_SNDBUF;
setsockopt(so, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize));
bufsize = GW_CLIENT_SO_RCVBUF;
setsockopt(so, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize));
/* set socket to as non-blocking here */
setnonblocking(so);
rv = connect(so, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
@ -1345,12 +1356,12 @@ int gw_find_mysql_user_password_sha1(char *username, uint8_t *gateway_password,
* The check for localhost is 127.0.0.1 (IPv4 only)
*/
if (key.ipv4.sin_addr.s_addr == 0x0100007F) {
if ((key.ipv4.sin_addr.s_addr == 0x0100007F) && !dcb->service->localhost_match_wildcard_host) {
/* Skip the wildcard check and return 1 */
LOGIF(LD,
LOGIF(LE,
(skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [MySQL Client Auth], user [%s@%s] not existent",
LOGFILE_ERROR,
"%lu [MySQL Client Auth], user [%s@%s] not found, please try with 'localhost_match_wildcard_host=1' in service definition",
pthread_self(),
key.user,
dcb->remote)));
@ -1530,9 +1541,7 @@ GWBUF* gw_MySQL_get_next_packet(
{
packetbuf = NULL;
goto return_packetbuf;
}
buflen = GWBUF_LENGTH((readbuf));
}
totalbuflen = gwbuf_length(readbuf);
data = (uint8_t *)GWBUF_DATA((readbuf));
packetlen = MYSQL_GET_PACKET_LEN(data)+4;
@ -1556,6 +1565,7 @@ GWBUF* gw_MySQL_get_next_packet(
uint8_t* src = GWBUF_DATA((*p_readbuf));
size_t bytestocopy;
buflen = GWBUF_LENGTH((*p_readbuf));
bytestocopy = MIN(buflen,packetlen-nbytes_copied);
memcpy(target+nbytes_copied, src, bytestocopy);
@ -1569,7 +1579,6 @@ return_packetbuf:
return packetbuf;
}
/**
* Move <npackets> from buffer pointed to by <*p_readbuf>.
*/
@ -1694,8 +1703,9 @@ void protocol_add_srv_command(
MySQLProtocol* p,
mysql_server_cmd_t cmd)
{
#if defined(SS_DEBUG)
server_command_t* c;
#endif
spinlock_acquire(&p->protocol_lock);
if (p->protocol_state != MYSQL_PROTOCOL_ACTIVE)

3
server/modules/routing/.gitignore vendored
View File

@ -1,3 +0,0 @@
*.o
*.so
depend.mk

1
server/modules/routing/Makefile
View File

@ -51,6 +51,7 @@ MODULES= libdebugcli.so libreadconnroute.so libtestroute.so libcli.so
all: $(MODULES)
(cd readwritesplit; make)
libtestroute.so: $(TESTOBJ)
$(CC) $(LDFLAGS) $(TESTOBJ) $(LIBS) -o $@

65
server/modules/routing/binlog/Makefile Normal file
View File

@ -0,0 +1,65 @@
# 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
#
# Revision History
# Date Who Description
# 2/04/14 Mark Riddoch Initial framework put in place
include ../../../../build_gateway.inc
LOGPATH := $(ROOT_PATH)/log_manager
UTILSPATH := $(ROOT_PATH)/utils
QCLASSPATH := $(ROOT_PATH)/query_classifier
CC=cc
CFLAGS=-c -fPIC -I/usr/include -I../../include -I../../../include \
-I$(LOGPATH) -I$(UTILSPATH) -I$(QCLASSPATH) \
$(MYSQL_HEADERS) -Wall -g
include ../../../../makefile.inc
LDFLAGS=-shared -L$(LOGPATH) -L$(QCLASSPATH) -L$(EMBEDDED_LIB) \
-Wl,-rpath,$(DEST)/lib \
-Wl,-rpath,$(LOGPATH) -Wl,-rpath,$(UTILSPATH) -Wl,-rpath,$(QCLASSPATH) \
-Wl,-rpath,$(EMBEDDED_LIB)
SRCS=blr.c blr_master.c blr_cache.c blr_slave.c blr_file.c
OBJ=$(SRCS:.c=.o)
LIBS=-lssl -pthread -llog_manager -lmysqld
MODULES=libbinlogrouter.so
all: $(MODULES)
$(MODULES): $(OBJ)
$(CC) $(LDFLAGS) $(OBJ) $(UTILSPATH)/skygw_utils.o $(LIBS) -o $@
.c.o:
$(CC) $(CFLAGS) $< -o $@
clean:
rm -f $(OBJ) $(MODULES)
tags:
ctags $(SRCS) $(HDRS)
depend:
@rm -f depend.mk
cc -M $(CFLAGS) $(SRCS) > depend.mk
install: $(MODULES)
install -D $(MODULES) $(DEST)/MaxScale/modules
include depend.mk

53
server/modules/routing/binlog/README Normal file
View File

@ -0,0 +1,53 @@
The binlog router is not a "normal" MaxScale router, it is not
designed to be used to route client requests to a database in the
usual proxy fashion. Rather it is designed to allow MaxScale to be
used as a relay server in a MySQL replication environment.
In this environment MaxScale sits between a master MySQL server and
a set of slave servers. The slaves servers execute a change master
to the MaxScale server, otehrwise they are configured in exactly
the same way as a normal MySQL slave server.
The master server configuration is unaltered, it simply sees a
single slave server.
MaxScale is configured as usual, with a service definition that
references the binlog router. The major configuration option to
consider is the router_options paramter, in the binlog router this
provides the binlog specific configuration parameters.
uuid=
This is the UUID that MaxScale uses when it connects
to the real master. It will report the master's
UUID to slaves that connect to it.
server-id=
The server-id that MaxScale uses when it connects
to the real master server. Again it will reports
the master's server-id to the slaves that connect
to it.
user=
The user that MaxScale uses to login to the real
master
password=
The password that MaxScale uses to login to the
real master
master-id=
The server-id of the real master. MaxScale should
get this by sending a query, but at the moment it
is in the configuration file for ease of implementation
An example binlog service configuration is shown below:
[Binlog Service]
type=service
router=binlogrouter
servers=master
router_options=uuid=f12fcb7f-b97b-11e3-bc5e-0401152c4c22,server-id=3,user=repl,password=slavepass,master-id=1
user=maxscale
passwd=Mhu87p2D
The servers list for a binlog router service should contain just
the master server. In future a list will be given and the monitor
used to determine which server is the current master server.

13
server/modules/routing/binlog/STATUS Normal file
View File

@ -0,0 +1,13 @@
The binlog router contained here is a prototype implementation and
should not be consider as production ready.
The router has been written and tested with MySQL 5.6 as a reference
for the replication behaviour, more investigation and implementation
is likely to be needed in order to use other versions of MySQL,
MariaDB or Percona Server.
To Do List:
1. The router does not implement the replication heartbeat mechanism.
2. Performance measurements have yet to be made.

770
server/modules/routing/binlog/blr.c Normal file
View File

@ -0,0 +1,770 @@
/*
* This file is distributed as part of 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 SkySQL Ab 2014
*/
/**
* @file blr.c - binlog router, allows MaxScale to act as an intermediatory for replication
*
* The binlog router is designed to be used in replication environments to
* increase the replication fanout of a master server. It provides a transparant
* mechanism to read the binlog entries for multiple slaves while requiring
* only a single connection to the actual master to support the slaves.
*
* The current prototype implement is designed to support MySQL 5.6 and has
* a number of limitations. This prototype is merely a proof of concept and
* should not be considered production ready.
*
* @verbatim
* Revision History
*
* Date Who Description
* 02/04/2014 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <service.h>
#include <server.h>
#include <router.h>
#include <atomic.h>
#include <spinlock.h>
#include <blr.h>
#include <dcb.h>
#include <spinlock.h>
#include <time.h>
#include <skygw_types.h>
#include <skygw_utils.h>
#include <log_manager.h>
#include <mysql_client_server_protocol.h>
extern int lm_enabled_logfiles_bitmask;
static char *version_str = "V1.0.6";
/* The router entry points */
static ROUTER *createInstance(SERVICE *service, char **options);
static void *newSession(ROUTER *instance, SESSION *session);
static void closeSession(ROUTER *instance, void *router_session);
static void freeSession(ROUTER *instance, void *router_session);
static int routeQuery(ROUTER *instance, void *router_session, GWBUF *queue);
static void diagnostics(ROUTER *instance, DCB *dcb);
static void clientReply(
ROUTER *instance,
void *router_session,
GWBUF *queue,
DCB *backend_dcb);
static void errorReply(
ROUTER *instance,
void *router_session,
GWBUF *message,
DCB *backend_dcb,
error_action_t action,
bool *succp);
static uint8_t getCapabilities (ROUTER* inst, void* router_session);
/** The module object definition */
static ROUTER_OBJECT MyObject = {
createInstance,
newSession,
closeSession,
freeSession,
routeQuery,
diagnostics,
clientReply,
errorReply,
getCapabilities
};
static bool rses_begin_locked_router_action(ROUTER_SLAVE *);
static void rses_end_locked_router_action(ROUTER_SLAVE *);
static SPINLOCK instlock;
static ROUTER_INSTANCE *instances;
/**
* 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()
{
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"Initialise binlog router module %s.\n", version_str)));
spinlock_init(&instlock);
instances = NULL;
}
/**
* 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
*/
ROUTER_OBJECT *
GetModuleObject()
{
return &MyObject;
}
/**
* Create an instance of the router for a particular service
* within MaxScale.
*
* The process of creating the instance causes the router to register
* with the master server and begin replication of the binlogs from
* the master server to MaxScale.
*
* @param service The service this router is being create for
* @param options An array of options for this query router
*
* @return The instance data for this new instance
*/
static ROUTER *
createInstance(SERVICE *service, char **options)
{
ROUTER_INSTANCE *inst;
char *value;
int i;
if ((inst = calloc(1, sizeof(ROUTER_INSTANCE))) == NULL) {
return NULL;
}
memset(&inst->stats, 0, sizeof(ROUTER_STATS));
memset(&inst->saved_master, 0, sizeof(MASTER_RESPONSES));
inst->service = service;
spinlock_init(&inst->lock);
inst->low_water = DEF_LOW_WATER;
inst->high_water = DEF_HIGH_WATER;
/*
* We only support one server behind this router, since the server is
* the master from which we replicate binlog records. Therefore check
* that only one server has been defined.
*
* A later improvement will be to define multiple servers and have the
* router use the information that is supplied by the monitor to find
* which of these servers is currently the master and replicate from
* that server.
*/
if (service->databases == NULL || service->databases->nextdb != NULL)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Error : Exactly one database server may be "
"for use with the binlog router.")));
}
/*
* Process the options.
* We have an array of attrbute values passed to us that we must
* examine. Supported attributes are:
* uuid=
* server-id=
* user=
* password=
* master-id=
* filestem=
* lowwater=
* highwater=
*/
if (options)
{
for (i = 0; options[i]; i++)
{
if ((value = strchr(options[i], '=')) == NULL)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR, "Warning : Unsupported router "
"option %s for binlog router.",
options[i])));
}
else
{
*value = 0;
value++;
if (strcmp(options[i], "uuid") == 0)
{
inst->uuid = strdup(value);
}
else if (strcmp(options[i], "server-id") == 0)
{
inst->serverid = atoi(value);
}
else if (strcmp(options[i], "user") == 0)
{
inst->user = strdup(value);
}
else if (strcmp(options[i], "password") == 0)
{
inst->password = strdup(value);
}
else if (strcmp(options[i], "master-id") == 0)
{
inst->masterid = atoi(value);
}
else if (strcmp(options[i], "filestem") == 0)
{
inst->fileroot = strdup(value);
}
else if (strcmp(options[i], "lowwater") == 0)
{
inst->low_water = atoi(value);
}
else if (strcmp(options[i], "highwater") == 0)
{
inst->high_water = atoi(value);
}
else
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Warning : Unsupported router "
"option %s for binlog router.",
options[i])));
}
}
}
if (inst->fileroot == NULL)
inst->fileroot = strdup(BINLOG_NAME_ROOT);
}
/*
* We have completed the creation of the instance data, so now
* insert this router instance into the linked list of routers
* that have been created with this module.
*/
spinlock_acquire(&instlock);
inst->next = instances;
instances = inst;
spinlock_release(&instlock);
inst->active_logs = 0;
inst->reconnect_pending = 0;
inst->handling_threads = 0;
inst->residual = NULL;
inst->slaves = NULL;
inst->next = NULL;
/*
* Initialise the binlog file and position
*/
blr_file_init(inst);
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE,
"Binlog router: current binlog file is: %s, current position %u\n",
inst->binlog_name, inst->binlog_position)));
/*
* Initialise the binlog cache for this router instance
*/
blr_init_cache(inst);
/*
* Now start the replication from the master to MaxScale
*/
blr_start_master(inst);
return (ROUTER *)inst;
}
/**
* Associate a new session with this instance of the router.
*
* In the case of the binlog router a new session equates to a new slave
* connecting to MaxScale and requesting binlog records. We need to go
* through the slave registration process for this new slave.
*
* @param instance The router instance data
* @param session The session itself
* @return Session specific data for this session
*/
static void *
newSession(ROUTER *instance, SESSION *session)
{
ROUTER_INSTANCE *inst = (ROUTER_INSTANCE *)instance;
ROUTER_SLAVE *slave;
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"binlog router: %lu [newSession] new router session with "
"session %p, and inst %p.",
pthread_self(),
session,
inst)));
if ((slave = (ROUTER_SLAVE *)calloc(1, sizeof(ROUTER_SLAVE))) == NULL)
{
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_ERROR,
"Insufficient memory to create new slave session for binlog router")));
return NULL;
}
#if defined(SS_DEBUG)
slave->rses_chk_top = CHK_NUM_ROUTER_SES;
slave->rses_chk_tail = CHK_NUM_ROUTER_SES;
#endif
memset(&slave->stats, 0, sizeof(SLAVE_STATS));
atomic_add(&inst->stats.n_slaves, 1);
slave->state = BLRS_CREATED; /* Set initial state of the slave */
slave->cstate = 0;
slave->pthread = 0;
slave->overrun = 0;
spinlock_init(&slave->catch_lock);
slave->dcb = session->client;
slave->router = inst;
/**
* Add this session to the list of active sessions.
*/
spinlock_acquire(&inst->lock);
slave->next = inst->slaves;
inst->slaves = slave;
spinlock_release(&inst->lock);
CHK_CLIENT_RSES(slave);
return (void *)slave;
}
/**
* The session is no longer required. Shutdown all operation and free memory
* associated with this session. In this case a single session is associated
* to a slave of MaxScale. Therefore this is called when that slave is no
* longer active and should remove of reference to that slave, free memory
* and prevent any further forwarding of binlog records to that slave.
*
* Parameters:
* @param router_instance The instance of the router
* @param router_cli_ses The particular session to free
*
*/
static void freeSession(
ROUTER* router_instance,
void* router_client_ses)
{
ROUTER_INSTANCE *router = (ROUTER_INSTANCE *)router_instance;
ROUTER_SLAVE *slave = (ROUTER_SLAVE *)router_client_ses;
int prev_val;
prev_val = atomic_add(&router->stats.n_slaves, -1);
ss_dassert(prev_val > 0);
/*
* Remove the slave session form the list of slaves that are using the
* router currently.
*/
spinlock_acquire(&router->lock);
if (router->slaves == slave) {
router->slaves = slave->next;
} else {
ROUTER_SLAVE *ptr = router->slaves;
while (ptr != NULL && ptr->next != slave) {
ptr = ptr->next;
}
if (ptr != NULL) {
ptr->next = slave->next;
}
}
spinlock_release(&router->lock);
LOGIF(LD, (skygw_log_write_flush(
LOGFILE_DEBUG,
"%lu [freeSession] Unlinked router_client_session %p from "
"router %p. Connections : %d. ",
pthread_self(),
slave,
router,
prev_val-1)));
if (slave->hostname)
free(slave->hostname);
if (slave->user)
free(slave->user);
if (slave->passwd)
free(slave->passwd);
free(slave);
}
/**
* Close a session with the router, this is the mechanism
* by which a router may cleanup data structure etc.
*
* @param instance The router instance data
* @param router_session The session being closed
*/
static void
closeSession(ROUTER *instance, void *router_session)
{
ROUTER_INSTANCE *router = (ROUTER_INSTANCE *)instance;
ROUTER_SLAVE *slave = (ROUTER_SLAVE *)router_session;
if (slave == NULL)
{
/*
* We must be closing the master session.
*
* TODO: Handle closure of master session
*/
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, "Binlog router close session with master")));
blr_master_reconnect(router);
return;
}
CHK_CLIENT_RSES(slave);
/**
* Lock router client session for secure read and update.
*/
if (rses_begin_locked_router_action(slave))
{
/* decrease server registered slaves counter */
atomic_add(&router->stats.n_registered, -1);
/*
* Mark the slave as unregistered to prevent the forwarding
* of any more binlog records to this slave.
*/
slave->state = BLRS_UNREGISTERED;
/* Unlock */
rses_end_locked_router_action(slave);
}
}
/**
* We have data from the client, this is likely to be packets related to
* the registration of the slave to receive binlog records. Unlike most
* MaxScale routers there is no forwarding to the backend database, merely
* the return of either predefined server responses that have been cached
* or binlog records.
*
* @param instance The router instance
* @param router_session The router session returned from the newSession call
* @param queue The queue of data buffers to route
* @return The number of bytes sent
*/
static int
routeQuery(ROUTER *instance, void *router_session, GWBUF *queue)
{
ROUTER_INSTANCE *router = (ROUTER_INSTANCE *)instance;
ROUTER_SLAVE *slave = (ROUTER_SLAVE *)router_session;
return blr_slave_request(router, slave, queue);
}
static char *event_names[] = {
"Invalid", "Start Event V3", "Query Event", "Stop Event", "Rotate Event",
"Integer Session Variable", "Load Event", "Slave Event", "Create File Event",
"Append Block Event", "Exec Load Event", "Delete File Event",
"New Load Event", "Rand Event", "User Variable Event", "Format Description Event",
"Transaction ID Event (2 Phase Commit)", "Begin Load Query Event",
"Execute Load Query Event", "Table Map Event", "Write Rows Event (v0)",
"Update Rows Event (v0)", "Delete Rows Event (v0)", "Write Rows Event (v1)",
"Update Rows Event (v1)", "Delete Rows Event (v1)", "Incident Event",
"Heartbeat Event", "Ignorable Event", "Rows Query Event", "Write Rows Event (v2)",
"Update Rows Event (v2)", "Delete Rows Event (v2)", "GTID Event",
"Anonymous GTID Event", "Previous GTIDS Event"
};
/**
* Display an entry from the spinlock statistics data
*
* @param dcb The DCB to print to
* @param desc Description of the statistic
* @param value The statistic value
*/
static void
spin_reporter(void *dcb, char *desc, int value)
{
dcb_printf((DCB *)dcb, "\t\t%-35s %d\n", desc, value);
}
/**
* Display router diagnostics
*
* @param instance Instance of the router
* @param dcb DCB to send diagnostics to
*/
static void
diagnostics(ROUTER *router, DCB *dcb)
{
ROUTER_INSTANCE *router_inst = (ROUTER_INSTANCE *)router;
ROUTER_SLAVE *session;
int i = 0;
char buf[40];
struct tm tm;
spinlock_acquire(&router_inst->lock);
session = router_inst->slaves;
while (session)
{
i++;
session = session->next;
}
spinlock_release(&router_inst->lock);
dcb_printf(dcb, "\tMaster connection DCB: %p\n",
router_inst->master);
dcb_printf(dcb, "\tMaster connection state: %s\n",
blrm_states[router_inst->master_state]);
localtime_r(&router_inst->stats.lastReply, &tm);
asctime_r(&tm, buf);
dcb_printf(dcb, "\tNumber of master connects: %d\n",
router_inst->stats.n_masterstarts);
dcb_printf(dcb, "\tNumber of delayed reconnects: %d\n",
router_inst->stats.n_delayedreconnects);
dcb_printf(dcb, "\tCurrent binlog file: %s\n",
router_inst->binlog_name);
dcb_printf(dcb, "\tCurrent binlog position: %u\n",
router_inst->binlog_position);
dcb_printf(dcb, "\tNumber of slave servers: %u\n",
router_inst->stats.n_slaves);
dcb_printf(dcb, "\tNumber of binlog events received: %u\n",
router_inst->stats.n_binlogs);
dcb_printf(dcb, "\tNumber of fake binlog events: %u\n",
router_inst->stats.n_fakeevents);
dcb_printf(dcb, "\tNumber of artificial binlog events: %u\n",
router_inst->stats.n_artificial);
dcb_printf(dcb, "\tNumber of binlog events in error: %u\n",
router_inst->stats.n_binlog_errors);
dcb_printf(dcb, "\tNumber of binlog rotate events: %u\n",
router_inst->stats.n_rotates);
dcb_printf(dcb, "\tNumber of binlog cache hits: %u\n",
router_inst->stats.n_cachehits);
dcb_printf(dcb, "\tNumber of binlog cache misses: %u\n",
router_inst->stats.n_cachemisses);
dcb_printf(dcb, "\tNumber of heartbeat events: %u\n",
router_inst->stats.n_heartbeats);
dcb_printf(dcb, "\tNumber of packets received: %u\n",
router_inst->stats.n_reads);
dcb_printf(dcb, "\tNumber of residual data packets: %u\n",
router_inst->stats.n_residuals);
dcb_printf(dcb, "\tAverage events per packet %.1f\n",
(double)router_inst->stats.n_binlogs / router_inst->stats.n_reads);
dcb_printf(dcb, "\tLast event from master at: %s",
buf);
dcb_printf(dcb, "\t (%d seconds ago)\n",
time(0) - router_inst->stats.lastReply);
dcb_printf(dcb, "\tLast event from master: 0x%x\n",
router_inst->lastEventReceived);
if (router_inst->active_logs)
dcb_printf(dcb, "\tRouter processing binlog records\n");
if (router_inst->reconnect_pending)
dcb_printf(dcb, "\tRouter pending reconnect to master\n");
dcb_printf(dcb, "\tEvents received:\n");
for (i = 0; i < 0x24; i++)
{
dcb_printf(dcb, "\t\t%-38s: %u\n", event_names[i], router_inst->stats.events[i]);
}
#if SPINLOCK_PROFILE
dcb_printf(dcb, "\tSpinlock statistics (instlock):\n");
spinlock_stats(&instlock, spin_reporter, dcb);
dcb_printf(dcb, "\tSpinlock statistics (instance lock):\n");
spinlock_stats(&router_inst->lock, spin_reporter, dcb);
#endif
if (router_inst->slaves)
{
dcb_printf(dcb, "\tSlaves:\n");
spinlock_acquire(&router_inst->lock);
session = router_inst->slaves;
while (session)
{
dcb_printf(dcb, "\t\tServer-id: %d\n", session->serverid);
if (session->hostname)
dcb_printf(dcb, "\t\tHostname: %s\n", session->hostname);
dcb_printf(dcb, "\t\tSlave DCB: %p\n", session->dcb);
dcb_printf(dcb, "\t\tNext Sequence No: %d\n", session->seqno);
dcb_printf(dcb, "\t\tState: %s\n", blrs_states[session->state]);
dcb_printf(dcb, "\t\tBinlog file: %s\n", session->binlogfile);
dcb_printf(dcb, "\t\tBinlog position: %u\n", session->binlog_pos);
if (session->nocrc)
dcb_printf(dcb, "\t\tMaster Binlog CRC: None\n");
dcb_printf(dcb, "\t\tNo. requests: %u\n", session->stats.n_requests);
dcb_printf(dcb, "\t\tNo. events sent: %u\n", session->stats.n_events);
dcb_printf(dcb, "\t\tNo. bursts sent: %u\n", session->stats.n_bursts);
dcb_printf(dcb, "\t\tNo. flow control: %u\n", session->stats.n_flows);
dcb_printf(dcb, "\t\tNo. catchup NRs: %u\n", session->stats.n_catchupnr);
dcb_printf(dcb, "\t\tNo. already up to date: %u\n", session->stats.n_alreadyupd);
dcb_printf(dcb, "\t\tNo. up to date: %u\n", session->stats.n_upd);
dcb_printf(dcb, "\t\tNo. of low water cbs %u\n", session->stats.n_cb);
dcb_printf(dcb, "\t\tNo. of drained cbs %u\n", session->stats.n_dcb);
dcb_printf(dcb, "\t\tNo. of low water cbs N/A %u\n", session->stats.n_cbna);
dcb_printf(dcb, "\t\tNo. of events > high water %u\n", session->stats.n_above);
dcb_printf(dcb, "\t\tNo. of failed reads %u\n", session->stats.n_failed_read);
dcb_printf(dcb, "\t\tNo. of nested distribute events %u\n", session->stats.n_overrun);
dcb_printf(dcb, "\t\tNo. of distribute action 1 %u\n", session->stats.n_actions[0]);
dcb_printf(dcb, "\t\tNo. of distribute action 2 %u\n", session->stats.n_actions[1]);
dcb_printf(dcb, "\t\tNo. of distribute action 3 %u\n", session->stats.n_actions[2]);
if ((session->cstate & CS_UPTODATE) == 0)
{
dcb_printf(dcb, "\t\tSlave is in catchup mode. %s\n",
((session->cstate & CS_EXPECTCB) == 0 ? "" :
"Waiting for DCB queue to drain."));
}
else
{
dcb_printf(dcb, "\t\tSlave is in normal mode.\n");
if (session->binlog_pos != router_inst->binlog_position)
{
dcb_printf(dcb, "\t\tSlave reports up to date however "
"the slave binlog position does not match the master\n");
}
}
#if SPINLOCK_PROFILE
dcb_printf(dcb, "\tSpinlock statistics (catch_lock):\n");
spinlock_stats(&session->catch_lock, spin_reporter, dcb);
dcb_printf(dcb, "\tSpinlock statistics (rses_lock):\n");
spinlock_stats(&session->rses_lock, spin_reporter, dcb);
#endif
session = session->next;
}
spinlock_release(&router_inst->lock);
}
}
/**
* Client Reply routine - in this case this is a message from the
* master server, It should be sent to the state machine that manages
* master packets as it may be binlog records or part of the registration
* handshake that takes part during connection establishment.
*
*
* @param instance The router instance
* @param router_session The router session
* @param master_dcb The DCB for the connection to the master
* @param queue The GWBUF with reply data
*/
static void
clientReply(ROUTER *instance, void *router_session, GWBUF *queue, DCB *backend_dcb)
{
ROUTER_INSTANCE *router = (ROUTER_INSTANCE *)instance;
atomic_add(&router->stats.n_reads, 1);
blr_master_response(router, queue);
router->stats.lastReply = time(0);
}
/**
* Error Reply routine
*
* The routine will reply to client errors and/or closing the session
* or try to open a new backend connection.
*
* @param instance The router instance
* @param router_session The router session
* @param message The error message to reply
* @param backend_dcb The backend DCB
* @param action The action: REPLY, REPLY_AND_CLOSE, NEW_CONNECTION
* @param succp Result of action
*
*/
static void
errorReply(ROUTER *instance, void *router_session, GWBUF *message, DCB *backend_dcb, error_action_t action, bool *succp)
{
LOGIF(LE, (skygw_log_write_flush(
LOGFILE_ERROR, "Erorr Reply '%s'", message)));
*succp = false;
}
/** to be inline'd */
/**
* @node Acquires lock to router client session if it is not closed.
*
* Parameters:
* @param rses - in, use
*
*
* @return true if router session was not closed. If return value is true
* it means that router is locked, and must be unlocked later. False, if
* router was closed before lock was acquired.
*
*
* @details (write detailed description here)
*
*/
static bool rses_begin_locked_router_action(ROUTER_SLAVE *rses)
{
bool succp = false;
CHK_CLIENT_RSES(rses);
spinlock_acquire(&rses->rses_lock);
succp = true;
return succp;
}
/** to be inline'd */
/**
* @node Releases router client session lock.
*
* Parameters:
* @param rses - <usage>
* <description>
*
* @return void
*
*
* @details (write detailed description here)
*
*/
static void rses_end_locked_router_action(ROUTER_SLAVE * rses)
{
CHK_CLIENT_RSES(rses);
spinlock_release(&rses->rses_lock);
}
static uint8_t getCapabilities(ROUTER *inst, void *router_session)
{
return 0;
}

69
server/modules/routing/binlog/blr_cache.c Normal file
View File

@ -0,0 +1,69 @@
/*
* This file is distributed as part of 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 SkySQL Ab 2014
*/
/**
* @file blr_cache.c - binlog router cache, manage the binlog cache
*
* The binlog router is designed to be used in replication environments to
* increase the replication fanout of a master server. It provides a transparant
* mechanism to read the binlog entries for multiple slaves while requiring
* only a single connection to the actual master to support the slaves.
*
* The current prototype implement is designed to support MySQL 5.6 and has
* a number of limitations. This prototype is merely a proof of concept and
* should not be considered production ready.
*
* @verbatim
* Revision History
*
* Date Who Description
* 07/04/2014 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <service.h>
#include <server.h>
#include <router.h>
#include <atomic.h>
#include <spinlock.h>
#include <blr.h>
#include <dcb.h>
#include <spinlock.h>
#include <skygw_types.h>
#include <skygw_utils.h>
#include <log_manager.h>
extern int lm_enabled_logfiles_bitmask;
/**
* Initialise the cache for this instanceof the binlog router. As a side
* effect also determine the binlog file to read and the position to read
* from.
*
* @param router The router instance
*/
void
blr_init_cache(ROUTER_INSTANCE *router)
{
}

346
server/modules/routing/binlog/blr_file.c Normal file
View File

@ -0,0 +1,346 @@
/*
* This file is distributed as part of 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 SkySQL Ab 2014
*/
/**
* @file blr_file.c - contains code for the router binlog file management
*
*
* @verbatim
* Revision History
*
* Date Who Description
* 14/04/2014 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <dirent.h>
#include <fcntl.h>
#include <unistd.h>
#include <service.h>
#include <server.h>
#include <router.h>
#include <atomic.h>
#include <spinlock.h>
#include <blr.h>
#include <dcb.h>
#include <spinlock.h>
#include <skygw_types.h>
#include <skygw_utils.h>
#include <log_manager.h>
extern int lm_enabled_logfiles_bitmask;
static void blr_file_create(ROUTER_INSTANCE *router, char *file);
static void blr_file_append(ROUTER_INSTANCE *router, char *file);
static uint32_t extract_field(uint8_t *src, int bits);
/**
* Initialise the binlog file for this instance. MaxScale will look
* for all the binlogs that it has on local disk, determien the next
* binlog to use and initialise it for writing, determining the
* next record to be fetched from the real master.
*
* @param router The router instance this defines the master for this replication chain
*/
void
blr_file_init(ROUTER_INSTANCE *router)
{
char *ptr, path[1024], filename[1050];
int file_found, n = 1;
int root_len, i;
DIR *dirp;
struct dirent *dp;
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
}
strcat(path, "/");
strcat(path, router->service->name);
if (access(path, R_OK) == -1)
mkdir(path, 0777);
/* First try to find a binlog file number by reading the directory */
root_len = strlen(router->fileroot);
dirp = opendir(path);
while ((dp = readdir(dirp)) != NULL)
{
if (strncmp(dp->d_name, router->fileroot, root_len) == 0)
{
i = atoi(dp->d_name + root_len + 1);
if (i > n)
n = i;
}
}
closedir(dirp);
file_found = 0;
do {
sprintf(filename, "%s/" BINLOG_NAMEFMT, path, router->fileroot, n);
if (access(filename, R_OK) != -1)
{
file_found = 1;
n++;
}
else
file_found = 0;
} while (file_found);
n--;
if (n == 0) // No binlog files found
{
sprintf(filename, BINLOG_NAMEFMT, router->fileroot, 1);
blr_file_create(router, filename);
}
else
{
sprintf(filename, BINLOG_NAMEFMT, router->fileroot, n);
blr_file_append(router, filename);
}
}
void
blr_file_rotate(ROUTER_INSTANCE *router, char *file, uint64_t pos)
{
blr_file_create(router, file);
}
/**
* Create a new binlog file for the router to use.
*
* @param router The router instance
* @param file The binlog file name
*/
static void
blr_file_create(ROUTER_INSTANCE *router, char *file)
{
char *ptr, path[1024];
int fd;
unsigned char magic[] = BINLOG_MAGIC;
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
}
strcat(path, "/");
strcat(path, router->service->name);
strcat(path, "/");
strcat(path, file);
if ((fd = open(path, O_RDWR|O_CREAT, 0666)) != -1)
{
write(fd, magic, 4);
}
else
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to create binlog file %s\n", path)));
}
fsync(fd);
close(router->binlog_fd);
strcpy(router->binlog_name, file);
router->binlog_position = 4; /* Initial position after the magic number */
router->binlog_fd = fd;
}
/**
* Prepare an existing binlog file to be appened to.
*
* @param router The router instance
* @param file The binlog file name
*/
static void
blr_file_append(ROUTER_INSTANCE *router, char *file)
{
char *ptr, path[1024];
int fd;
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
}
strcat(path, "/");
strcat(path, router->service->name);
strcat(path, "/");
strcat(path, file);
if ((fd = open(path, O_RDWR|O_APPEND, 0666)) == -1)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to open binlog file %s for append.\n",
path)));
return;
}
fsync(fd);
close(router->binlog_fd);
strcpy(router->binlog_name, file);
router->binlog_position = lseek(fd, 0L, SEEK_END);
router->binlog_fd = fd;
}
/**
* Write a binlog entry to disk.
*
* @param router The router instance
* @param buf The binlog record
* @param len The length of the binlog record
*/
void
blr_write_binlog_record(ROUTER_INSTANCE *router, REP_HEADER *hdr, uint8_t *buf)
{
pwrite(router->binlog_fd, buf, hdr->event_size, hdr->next_pos - hdr->event_size);
router->binlog_position = hdr->next_pos;
}
/**
* Flush the content of the binlog file to disk.
*
* @param router The binlog router
*/
void
blr_file_flush(ROUTER_INSTANCE *router)
{
fsync(router->binlog_fd);
}
int
blr_open_binlog(ROUTER_INSTANCE *router, char *binlog)
{
char *ptr, path[1024];
int rval;
strcpy(path, "/usr/local/skysql/MaxScale");
if ((ptr = getenv("MAXSCALE_HOME")) != NULL)
{
strcpy(path, ptr);
}
strcat(path, "/");
strcat(path, router->service->name);
strcat(path, "/");
strcat(path, binlog);
if ((rval = open(path, O_RDONLY, 0666)) == -1)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to open binlog file %s\n", path)));
}
return rval;
}
/**
* Read a replication event into a GWBUF structure.
*
* @param fd File descriptor of the binlog file
* @param pos Position of binlog record to read
* @param hdr Binlog header to populate
* @return The binlog record wrapped in a GWBUF structure
*/
GWBUF *
blr_read_binlog(int fd, unsigned int pos, REP_HEADER *hdr)
{
uint8_t hdbuf[19];
GWBUF *result;
unsigned char *data;
int n;
if (lseek(fd, pos, SEEK_SET) != pos)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to seek for binlog entry, "
"at %d.\n", pos)));
return NULL;
}
/* Read the header information from the file */
if ((n = read(fd, hdbuf, 19)) != 19)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to read header for binlog entry, "
"at %d (%s).\n", pos, strerror(errno))));
if (n> 0 && n < 19)
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Short read when reading the header. "
"Expected 19 bytes got %d bytes.\n",
n)));
return NULL;
}
hdr->timestamp = extract_field(hdbuf, 32);
hdr->event_type = hdbuf[4];
hdr->serverid = extract_field(&hdbuf[5], 32);
hdr->event_size = extract_field(&hdbuf[9], 32);
hdr->next_pos = extract_field(&hdbuf[13], 32);
hdr->flags = extract_field(&hdbuf[17], 16);
if ((result = gwbuf_alloc(hdr->event_size)) == NULL)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to allocate memory for binlog entry, "
"size %d at %d.\n",
hdr->event_size, pos)));
return NULL;
}
data = GWBUF_DATA(result);
memcpy(data, hdbuf, 19); // Copy the header in
if ((n = read(fd, &data[19], hdr->event_size - 19))
!= hdr->event_size - 19) // Read the balance
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Short read when reading the event at %d. "
"Expected %d bytes got %d bytes.\n",
pos, n)));
gwbuf_consume(result, hdr->event_size);
return NULL;
}
return result;
}
/**
* Extract a numeric field from a packet of the specified number of bits
*
* @param src The raw packet source
* @param birs The number of bits to extract (multiple of 8)
*/
static uint32_t
extract_field(uint8_t *src, int bits)
{
uint32_t rval = 0, shift = 0;
while (bits > 0)
{
rval |= (*src++) << shift;
shift += 8;
bits -= 8;
}
return rval;
}

1024
server/modules/routing/binlog/blr_master.c Normal file
View File

File diff suppressed because it is too large Load Diff

944
server/modules/routing/binlog/blr_slave.c Normal file
View File

@ -0,0 +1,944 @@
/*
* This file is distributed as part of 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 SkySQL Ab 2014
*/
/**
* @file blr_slave.c - contains code for the router to slave communication
*
* The binlog router is designed to be used in replication environments to
* increase the replication fanout of a master server. It provides a transparant
* mechanism to read the binlog entries for multiple slaves while requiring
* only a single connection to the actual master to support the slaves.
*
* The current prototype implement is designed to support MySQL 5.6 and has
* a number of limitations. This prototype is merely a proof of concept and
* should not be considered production ready.
*
* @verbatim
* Revision History
*
* Date Who Description
* 14/04/2014 Mark Riddoch Initial implementation
*
* @endverbatim
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <service.h>
#include <server.h>
#include <router.h>
#include <atomic.h>
#include <spinlock.h>
#include <blr.h>
#include <dcb.h>
#include <spinlock.h>
#include <skygw_types.h>
#include <skygw_utils.h>
#include <log_manager.h>
static uint32_t extract_field(uint8_t *src, int bits);
static void encode_value(unsigned char *data, unsigned int value, int len);
static int blr_slave_query(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue);
static int blr_slave_replay(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *master);
static void blr_slave_send_error(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, char *msg);
static int blr_slave_send_timestamp(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave);
static int blr_slave_register(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue);
static int blr_slave_binlog_dump(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue);
int blr_slave_catchup(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave);
static uint8_t *blr_build_header(GWBUF *pkt, REP_HEADER *hdr);
static int blr_slave_callback(DCB *dcb, DCB_REASON reason, void *data);
extern int lm_enabled_logfiles_bitmask;
/**
* Process a request packet from the slave server.
*
* The router can handle a limited subset of requests from the slave, these
* include a subset of general SQL queries, a slave registeration command and
* the binlog dump command.
*
* The strategy for responding to these commands is to use caches responses
* for the the same commands that have previously been made to the real master
* if this is possible, if it is not then the router itself will synthesize a
* response.
*
* @param router The router instance this defines the master for this replication chain
* @param slave The slave specific data
* @param queue The incoming request packet
*/
int
blr_slave_request(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue)
{
if (slave->state < 0 || slave->state > BLRS_MAXSTATE)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR, "Invalid slave state machine state (%d) for binlog router.\n",
slave->state)));
gwbuf_consume(queue, gwbuf_length(queue));
return 0;
}
atomic_add(&slave->stats.n_requests, 1);
switch (MYSQL_COMMAND(queue))
{
case COM_QUERY:
return blr_slave_query(router, slave, queue);
break;
case COM_REGISTER_SLAVE:
return blr_slave_register(router, slave, queue);
break;
case COM_BINLOG_DUMP:
return blr_slave_binlog_dump(router, slave, queue);
break;
case COM_QUIT:
LOGIF(LD, (skygw_log_write(LOGFILE_DEBUG,
"COM_QUIT received from slave with server_id %d\n",
slave->serverid)));
break;
default:
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"Unexpected MySQL Command (%d) received from slave\n",
MYSQL_COMMAND(queue))));
break;
}
return 0;
}
/**
* Handle a query from the slave. This is expected to be one of the "standard"
* queries we expect as part of the registraton process. Most of these can
* be dealt with by replying the stored responses we got from the master
* when MaxScale registered as a slave. The exception to the rule is the
* request to obtain the current timestamp value of the server.
*
* Five select statements are currently supported:
* SELECT UNIX_TIMESTAMP();
* SELECT @master_binlog_checksum
* SELECT @@GLOBAL.GTID_MODE
* SELECT VERSION()
* SELECT 1
*
* Two show commands are supported:
* SHOW VARIABLES LIKE 'SERVER_ID'
* SHOW VARIABLES LIKE 'SERVER_UUID'
*
* Five set commands are supported:
* SET @master_binlog_checksum = @@global.binlog_checksum
* SET @master_heartbeat_period=...
* SET @slave_slave_uuid=...
* SET NAMES latin1
* SET NAMES utf8
*
* @param router The router instance this defines the master for this replication chain
* @param slave The slave specific data
* @param queue The incoming request packet
* @return Non-zero if data has been sent
*/
static int
blr_slave_query(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue)
{
char *qtext, *query_text;
char *sep = " ,=";
char *word, *brkb;
int query_len;
qtext = GWBUF_DATA(queue);
query_len = extract_field((uint8_t *)qtext, 24) - 1;
qtext += 5; // Skip header and first byte of the payload
query_text = strndup(qtext, query_len);
LOGIF(LT, (skygw_log_write(
LOGFILE_TRACE, "Execute statement from the slave '%s'\n", query_text)));
/*
* Implement a very rudimental "parsing" of the query text by extarcting the
* words from the statement and matchng them against the subset of queries we
* are expecting from the slave. We already have responses to these commands,
* except for the select of UNIX_TIMESTAMP(), that we have saved from MaxScale's
* own interaction with the real master. We simply replay these saved responses
* to the slave.
*/
word = strtok_r(query_text, sep, &brkb);
if (strcasecmp(word, "SELECT") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "UNIX_TIMESTAMP()") == 0)
{
free(query_text);
return blr_slave_send_timestamp(router, slave);
}
else if (strcasecmp(word, "@master_binlog_checksum") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.chksum2);
}
else if (strcasecmp(word, "@@GLOBAL.GTID_MODE") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.gtid_mode);
}
else if (strcasecmp(word, "1") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.select1);
}
else if (strcasecmp(word, "VERSION()") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.selectver);
}
}
else if (strcasecmp(word, "SHOW") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "VARIABLES") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "LIKE") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "'SERVER_ID'") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.server_id);
}
else if (strcasecmp(word, "'SERVER_UUID'") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.uuid);
}
}
}
}
else if (strcasecmp(query_text, "SET") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "@master_heartbeat_period") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.heartbeat);
}
else if (strcasecmp(word, "@master_binlog_checksum") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "'none'") == 0)
slave->nocrc = 1;
else
slave->nocrc = 0;
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.chksum1);
}
else if (strcasecmp(word, "@slave_uuid") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.setslaveuuid);
}
else if (strcasecmp(word, "NAMES") == 0)
{
word = strtok_r(NULL, sep, &brkb);
if (strcasecmp(word, "latin1") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.setnames);
}
else if (strcasecmp(word, "utf8") == 0)
{
free(query_text);
return blr_slave_replay(router, slave, router->saved_master.utf8);
}
}
}
free(query_text);
query_text = strndup(qtext, query_len);
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR, "Unexpected query from slave server %s\n", query_text)));
free(query_text);
blr_slave_send_error(router, slave, "Unexpected SQL query received from slave.");
return 0;
}
/**
* Send a reply to a command we have received from the slave. The reply itself
* is merely a copy of a previous message we received from the master when we
* registered as a slave. Hence we just replay this saved reply.
*
* @param router The binlog router instance
* @param slave The slave server to which we are sending the response
* @param master The saved master response
* @return Non-zero if data was sent
*/
static int
blr_slave_replay(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *master)
{
GWBUF *clone;
if (!master)
return 0;
if ((clone = gwbuf_clone(master)) != NULL)
{
return slave->dcb->func.write(slave->dcb, clone);
}
else
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR,
"Failed to clone server response to send to slave.\n")));
return 0;
}
}
/**
* Construct an error response
*
* @param router The router instance
* @param slave The slave server instance
* @param msg The error message to send
*/
static void
blr_slave_send_error(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, char *msg)
{
GWBUF *pkt;
unsigned char *data;
int len;
if ((pkt = gwbuf_alloc(strlen(msg) + 13)) == NULL)
return;
data = GWBUF_DATA(pkt);
len = strlen(msg) + 1;
encode_value(&data[0], len, 24); // Payload length
data[3] = 0; // Sequence id
// Payload
data[4] = 0xff; // Error indicator
data[5] = 0; // Error Code
data[6] = 0; // Error Code
strncpy((char *)&data[7], "#00000", 6);
memcpy(&data[13], msg, strlen(msg)); // Error Message
slave->dcb->func.write(slave->dcb, pkt);
}
/*
* Some standard packets that have been captured from a network trace of server
* interactions. These packets are the schema definition sent in response to
* a SELECT UNIX_TIMESTAMP() statement and the EOF packet that marks the end
* of transmission of the result set.
*/
static uint8_t timestamp_def[] = {
0x01, 0x00, 0x00, 0x01, 0x01, 0x26, 0x00, 0x00, 0x02, 0x03, 0x64, 0x65, 0x66, 0x00, 0x00, 0x00,
0x10, 0x55, 0x4e, 0x49, 0x58, 0x5f, 0x54, 0x49, 0x4d, 0x45, 0x53, 0x54, 0x41, 0x4d, 0x50, 0x28,
0x29, 0x00, 0x0c, 0x3f, 0x00, 0x0b, 0x00, 0x00, 0x00, 0x08, 0x81, 0x00, 0x00, 0x00, 0x00, 0x05,
0x00, 0x00, 0x03, 0xfe, 0x00, 0x00, 0x02, 0x00
};
static uint8_t timestamp_eof[] = { 0x05, 0x00, 0x00, 0x05, 0xfe, 0x00, 0x00, 0x02, 0x00 };
/**
* Send a response to a "SELECT UNIX_TIMESTAMP()" request. This differs from the other
* requests since we do not save a copy of the original interaction with the master
* and simply replay it. We want to always send the current time. We have stored a typcial
* response, which gives us the schema information normally returned. This is sent to the
* client and then we add a dynamic part that will insert the current timestamp data.
* Finally we send a preprepaed EOF packet to end the response stream.
*
* @param router The binlog router instance
* @param slave The slave server to which we are sending the response
* @return Non-zero if data was sent
*/
static int
blr_slave_send_timestamp(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
{
GWBUF *pkt;
char timestamp[20];
uint8_t *ptr;
int len, ts_len;
sprintf(timestamp, "%ld", time(0));
ts_len = strlen(timestamp);
len = sizeof(timestamp_def) + sizeof(timestamp_eof) + 5 + ts_len;
if ((pkt = gwbuf_alloc(len)) == NULL)
return 0;
ptr = GWBUF_DATA(pkt);
memcpy(ptr, timestamp_def, sizeof(timestamp_def)); // Fixed preamble
ptr += sizeof(timestamp_def);
encode_value(ptr, ts_len + 1, 24); // Add length of data packet
ptr += 3;
*ptr++ = 0x04; // Sequence number in response
*ptr++ = ts_len; // Length of result string
strncpy((char *)ptr, timestamp, ts_len); // Result string
ptr += ts_len;
memcpy(ptr, timestamp_eof, sizeof(timestamp_eof)); // EOF packet to terminate result
return slave->dcb->func.write(slave->dcb, pkt);
}
/**
* Process a slave replication registration message.
*
* We store the various bits of information the slave gives us and generate
* a reply message.
*
* @param router The router instance
* @param slave The slave server
* @param queue The BINLOG_DUMP packet
* @return Non-zero if data was sent
*/
static int
blr_slave_register(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue)
{
GWBUF *resp;
uint8_t *ptr;
int len, slen;
ptr = GWBUF_DATA(queue);
len = extract_field(ptr, 24);
ptr += 4; // Skip length and sequence number
if (*ptr++ != COM_REGISTER_SLAVE)
return 0;
slave->serverid = extract_field(ptr, 32);
ptr += 4;
slen = *ptr++;
if (slen != 0)
{
slave->hostname = strndup((char *)ptr, slen);
ptr += slen;
}
else
slave->hostname = NULL;
slen = *ptr++;
if (slen != 0)
{
ptr += slen;
slave->user = strndup((char *)ptr, slen);
}
else
slave->user = NULL;
slen = *ptr++;
if (slen != 0)
{
slave->passwd = strndup((char *)ptr, slen);
ptr += slen;
}
else
slave->passwd = NULL;
slave->port = extract_field(ptr, 16);
ptr += 2;
slave->rank = extract_field(ptr, 32);
/*
* Now construct a response
*/
if ((resp = gwbuf_alloc(11)) == NULL)
return 0;
ptr = GWBUF_DATA(resp);
encode_value(ptr, 7, 24); // Payload length
ptr += 3;
*ptr++ = 1; // Sequence number
encode_value(ptr, 0, 24);
ptr += 3;
encode_value(ptr, slave->serverid, 32);
slave->state = BLRS_REGISTERED;
return slave->dcb->func.write(slave->dcb, resp);
}
/**
* Process a COM_BINLOG_DUMP message from the slave. This is the
* final step in the process of registration. The new master, MaxScale
* must send a response packet and generate a fake BINLOG_ROTATE event
* with the binlog file requested by the slave. And then send a
* FORMAT_DESCRIPTION_EVENT that has been saved from the real master.
*
* Once send MaxScale must continue to send binlog events to the slave.
*
* @param router The router instance
* @param slave The slave server
* @param queue The BINLOG_DUMP packet
* @return The number of bytes written to the slave
*/
static int
blr_slave_binlog_dump(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave, GWBUF *queue)
{
GWBUF *resp;
uint8_t *ptr;
int len, flags, serverid, rval;
REP_HEADER hdr;
uint32_t chksum;
ptr = GWBUF_DATA(queue);
len = extract_field(ptr, 24);
ptr += 4; // Skip length and sequence number
if (*ptr++ != COM_BINLOG_DUMP)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"blr_slave_binlog_dump expected a COM_BINLOG_DUMP but received %d\n",
*(ptr-1))));
return 0;
}
slave->binlog_pos = extract_field(ptr, 32);
ptr += 4;
flags = extract_field(ptr, 16);
ptr += 2;
serverid = extract_field(ptr, 32);
ptr += 4;
strncpy(slave->binlogfile, (char *)ptr, BINLOG_FNAMELEN);
slave->state = BLRS_DUMPING;
slave->seqno = 1;
if (slave->nocrc)
len = 0x2b;
else
len = 0x2f;
// Build a fake rotate event
resp = gwbuf_alloc(len + 5);
hdr.payload_len = len + 1;
hdr.seqno = slave->seqno++;
hdr.ok = 0;
hdr.timestamp = 0L;
hdr.event_type = ROTATE_EVENT;
hdr.serverid = router->masterid;
hdr.event_size = len;
hdr.next_pos = 0;
hdr.flags = 0x20;
ptr = blr_build_header(resp, &hdr);
encode_value(ptr, slave->binlog_pos, 64);
ptr += 8;
memcpy(ptr, slave->binlogfile, BINLOG_FNAMELEN);
ptr += BINLOG_FNAMELEN;
if (!slave->nocrc)
{
/*
* Now add the CRC to the fake binlog rotate event.
*
* The algorithm is first to compute the checksum of an empty buffer
* and then the checksum of the event portion of the message, ie we do not
* include the length, sequence number and ok byte that makes up the first
* 5 bytes of the message. We also do not include the 4 byte checksum itself.
*/
chksum = crc32(0L, NULL, 0);
chksum = crc32(chksum, GWBUF_DATA(resp) + 5, hdr.event_size - 4);
encode_value(ptr, chksum, 32);
}
rval = slave->dcb->func.write(slave->dcb, resp);
/* Send the FORMAT_DESCRIPTION_EVENT */
if (router->saved_master.fde_event)
{
resp = gwbuf_alloc(router->saved_master.fde_len + 5);
ptr = GWBUF_DATA(resp);
encode_value(ptr, router->saved_master.fde_len + 1, 24); // Payload length
ptr += 3;
*ptr++ = slave->seqno++;
*ptr++ = 0; // OK
memcpy(ptr, router->saved_master.fde_event, router->saved_master.fde_len);
encode_value(ptr, time(0), 32); // Overwrite timestamp
/*
* Since we have changed the timestamp we must recalculate the CRC
*
* Position ptr to the start of the event header,
* calculate a new checksum
* and write it into the header
*/
ptr = GWBUF_DATA(resp) + 5 + router->saved_master.fde_len - 4;
chksum = crc32(0L, NULL, 0);
chksum = crc32(chksum, GWBUF_DATA(resp) + 5, router->saved_master.fde_len - 4);
encode_value(ptr, chksum, 32);
rval = slave->dcb->func.write(slave->dcb, resp);
}
slave->dcb->low_water = router->low_water;
slave->dcb->high_water = router->high_water;
dcb_add_callback(slave->dcb, DCB_REASON_LOW_WATER, blr_slave_callback, slave);
dcb_add_callback(slave->dcb, DCB_REASON_DRAINED, blr_slave_callback, slave);
if (slave->binlog_pos != router->binlog_position ||
strcmp(slave->binlogfile, router->binlog_name) != 0)
{
spinlock_acquire(&slave->catch_lock);
slave->cstate &= ~CS_UPTODATE;
spinlock_release(&slave->catch_lock);
rval = blr_slave_catchup(router, slave);
}
return rval;
}
/**
* Extract a numeric field from a packet of the specified number of bits,
* the number of bits must be a multiple of 8.
*
* @param src The raw packet source
* @param bits The number of bits to extract (multiple of 8)
* @return The extracted value
*/
static uint32_t
extract_field(uint8_t *src, int bits)
{
uint32_t rval = 0, shift = 0;
while (bits > 0)
{
rval |= (*src++) << shift;
shift += 8;
bits -= 8;
}
return rval;
}
/**
* Encode a value into a number of bits in a MySQL packet
*
* @param data Pointer to location in target packet
* @param value The value to encode into the buffer
* @param len Number of bits to encode value into
*/
static void
encode_value(unsigned char *data, unsigned int value, int len)
{
while (len > 0)
{
*data++ = value & 0xff;
value >>= 8;
len -= 8;
}
}
/**
* Populate a header structure for a replication message from a GWBUF structure.
*
* @param pkt The incoming packet in a GWBUF chain
* @param hdr The packet header to populate
* @return A pointer to the first byte following the event header
*/
static uint8_t *
blr_build_header(GWBUF *pkt, REP_HEADER *hdr)
{
uint8_t *ptr;
ptr = GWBUF_DATA(pkt);
encode_value(ptr, hdr->payload_len, 24);
ptr += 3;
*ptr++ = hdr->seqno;
*ptr++ = hdr->ok;
encode_value(ptr, hdr->timestamp, 32);
ptr += 4;
*ptr++ = hdr->event_type;
encode_value(ptr, hdr->serverid, 32);
ptr += 4;
encode_value(ptr, hdr->event_size, 32);
ptr += 4;
encode_value(ptr, hdr->next_pos, 32);
ptr += 4;
encode_value(ptr, hdr->flags, 16);
ptr += 2;
return ptr;
}
/**
* We have a registered slave that is behind the current leading edge of the
* binlog. We must replay the log entries to bring this node up to speed.
*
* There may be a large numebr of records to send to the slave, the process
* is triggered by the slave COM_BINLOG_DUMP message and all the events must
* be sent without receiving any new event. This measn there is no trigger into
* MaxScale other than this initial message. However, if we simply send all the
* events we end up with an extremely long write queue on the DCB and risk running
* the server out of resources.
*
* To resolve this the concept of high and low water marks within the DCB has been
* added, with the ability for the DCB code to call user defined callbacks when the
* write queue is completely drained, when it crosses above the high water mark and
* when it crosses below the low water mark.
*
* The blr_slave_catchup routine will send binlog events to the slave until the high
* water mark is reached, at which point it will return. Later, when a low water mark
* callback is generated by the code that drains the DCB of data the blr_slave_catchup
* routine will again be called to write more events. The process is repeated until
* the slave has caught up with the master.
*
* Note: an additional check that the DCB is still above the low water mark is done
* prior to the return from this function to allow for any delays due to the call to
* the close system call, since this may cause thread rescheduling.
*
* @param router The binlog router
* @param slave The slave that is behind
* @return The number of bytes written
*/
int
blr_slave_catchup(ROUTER_INSTANCE *router, ROUTER_SLAVE *slave)
{
GWBUF *head, *record;
REP_HEADER hdr;
int written, fd, rval = 1, burst = 0;
uint8_t *ptr;
struct timespec req;
spinlock_acquire(&slave->catch_lock);
slave->cstate &= ~CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
doitagain:
/*
* We have a slightly complex syncronisation mechansim here,
* we need to make sure that we do not have multiple threads
* running the catchup loop, but we need to be very careful
* that we do not loose a call that is coming via a callback
* call as this will stall the binlog catchup process.
*
* We don't want to simply use a traditional mutex here for
* the loop, since this would block a MaxScale thread for
* an unacceptable length of time.
*
* We have two status bits, the CS_READING that says we are
* in the outer loop and the CS_INNERLOOP, to say we are in
* the inner loop.
*
* If just CS_READING is set the other thread may be about to
* enter the inner loop or may be about to exit the function
* completely. Therefore we have to wait to see if CS_READING
* is cleared or CS_INNERLOOP is set.
*
* If CS_READING gets cleared then this thread should proceed
* into the loop.
*
* If CS_INNERLOOP get's set then this thread does not need to
* proceed.
*
* If CS_READING is not set then this thread simply enters the
* loop.
*/
req.tv_sec = 0;
req.tv_nsec = 1000;
spinlock_acquire(&slave->catch_lock);
if (slave->cstate & CS_UPTODATE)
{
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE,
"blr_slave_catchup called with up to date slave %d at "
"%s@%d. Reading position %s@%d\n",
slave->serverid, slave->binlogfile,
slave->binlog_pos, router->binlog_name,
router->binlog_position)));
slave->stats.n_alreadyupd++;
spinlock_release(&slave->catch_lock);
return 1;
}
while (slave->cstate & CS_READING)
{
// Wait until we know what the other thread is doing
while ((slave->cstate & (CS_READING|CS_INNERLOOP)) == CS_READING)
{
spinlock_release(&slave->catch_lock);
nanosleep(&req, NULL);
spinlock_acquire(&slave->catch_lock);
}
// Other thread is in the innerloop
if ((slave->cstate & (CS_READING|CS_INNERLOOP)) == (CS_READING|CS_INNERLOOP))
{
spinlock_release(&slave->catch_lock);
LOGIF(LM, (skygw_log_write(
LOGFILE_MESSAGE,
"blr_slave_catchup thread returning due to "
"lock being held by another thread. %s@%d\n",
slave->binlogfile,
slave->binlog_pos)));
slave->stats.n_catchupnr++;
return 1; // We cheat here and return 1 because otherwise
// an error would be sent and we do not want that
}
/* Release the lock for a short time to allow the other
* thread to exit the outer reading loop.
*/
spinlock_release(&slave->catch_lock);
nanosleep(&req, NULL);
spinlock_acquire(&slave->catch_lock);
}
if (slave->pthread)
LOGIF(LD, (skygw_log_write(LOGFILE_DEBUG, "Multiple threads sending to same thread.\n")));
slave->pthread = pthread_self();
slave->cstate |= CS_READING;
spinlock_release(&slave->catch_lock);
if (DCB_ABOVE_HIGH_WATER(slave->dcb))
LOGIF(LT, (skygw_log_write(LOGFILE_TRACE, "blr_slave_catchup above high water on entry.\n")));
do {
if ((fd = blr_open_binlog(router, slave->binlogfile)) == -1)
{
spinlock_acquire(&slave->catch_lock);
slave->cstate &= ~CS_READING;
spinlock_release(&slave->catch_lock);
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"blr_slave_catchup failed to open binlog file %s\n",
slave->binlogfile)));
return 0;
}
atomic_add(&slave->stats.n_bursts, 1);
spinlock_acquire(&slave->catch_lock);
slave->cstate |= CS_INNERLOOP;
spinlock_release(&slave->catch_lock);
while ((!DCB_ABOVE_HIGH_WATER(slave->dcb)) &&
(record = blr_read_binlog(fd, slave->binlog_pos, &hdr)) != NULL)
{
if (hdr.event_size > DEF_HIGH_WATER) slave->stats.n_above++;
head = gwbuf_alloc(5);
ptr = GWBUF_DATA(head);
encode_value(ptr, hdr.event_size + 1, 24);
ptr += 3;
*ptr++ = slave->seqno++;
*ptr++ = 0; // OK
head = gwbuf_append(head, record);
if (hdr.event_type == ROTATE_EVENT)
{
close(fd);
blr_slave_rotate(slave, GWBUF_DATA(record));
if ((fd = blr_open_binlog(router, slave->binlogfile)) == -1)
{
LOGIF(LE, (skygw_log_write(
LOGFILE_ERROR,
"blr_slave_catchup failed to open binlog file %s\n",
slave->binlogfile)));
break;
}
}
written = slave->dcb->func.write(slave->dcb, head);
if (written && hdr.event_type != ROTATE_EVENT)
{
slave->binlog_pos = hdr.next_pos;
}
rval = written;
atomic_add(&slave->stats.n_events, 1);
burst++;
}
if (record == NULL)
slave->stats.n_failed_read++;
spinlock_acquire(&slave->catch_lock);
slave->cstate &= ~CS_INNERLOOP;
spinlock_release(&slave->catch_lock);
close(fd);
} while (record && DCB_BELOW_LOW_WATER(slave->dcb));
if (record)
{
atomic_add(&slave->stats.n_flows, 1);
spinlock_acquire(&slave->catch_lock);
slave->cstate |= CS_EXPECTCB;
spinlock_release(&slave->catch_lock);
}
else
{
int state_change = 0;
spinlock_acquire(&slave->catch_lock);
if ((slave->cstate & CS_UPTODATE) == 0)
{
atomic_add(&slave->stats.n_upd, 1);
slave->cstate |= CS_UPTODATE;
state_change = 1;
}
spinlock_release(&slave->catch_lock);
if (state_change)
LOGIF(LM, (skygw_log_write(LOGFILE_MESSAGE,
"blr_slave_catchup slave is up to date %s, %u\n",
slave->binlogfile, slave->binlog_pos)));
}
spinlock_acquire(&slave->catch_lock);
#if 0
if (slave->pthread != pthread_self())
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR, "Multple threads in catchup for same slave: %x and %x\n", slave->pthread, pthread_self())));
abort();
}
#endif
slave->pthread = 0;
#if 0
if (DCB_BELOW_LOW_WATER(slave->dcb) && slave->binlog_pos != router->binlog_position) abort();
#endif
slave->cstate &= ~CS_READING;
spinlock_release(&slave->catch_lock);
if (DCB_BELOW_LOW_WATER(slave->dcb) && slave->binlog_pos != router->binlog_position)
{
LOGIF(LE, (skygw_log_write(LOGFILE_ERROR, "Expected to be above low water\n")));
goto doitagain;
}
return rval;
}
/**
* The DCB callback used by the slave to obtain DCB_REASON_LOW_WATER callbacks
* when the server sends all the the queue data for a DCB. This is the mechanism
* that is used to implement the flow control mechanism for the sending of
* large quantities of binlog records during the catchup process.
*
* @param dcb The DCB of the slave connection
* @param reason The reason the callback was called
* @param data The user data, in this case the server structure
*/
static int
blr_slave_callback(DCB *dcb, DCB_REASON reason, void *data)
{
ROUTER_SLAVE *slave = (ROUTER_SLAVE *)data;
ROUTER_INSTANCE *router = slave->router;
if (reason == DCB_REASON_DRAINED)
{
if (slave->state == BLRS_DUMPING &&
slave->binlog_pos != router->binlog_position)
{
atomic_add(&slave->stats.n_dcb, 1);
blr_slave_catchup(router, slave);
}
}
if (reason == DCB_REASON_LOW_WATER)
{
if (slave->state == BLRS_DUMPING)
{
atomic_add(&slave->stats.n_cb, 1);
blr_slave_catchup(router, slave);
}
else
{
atomic_add(&slave->stats.n_cbna, 1);
}
}
return 0;
}
/**
* Rotate the slave to the new binlog file
*
* @param slave The slave instance
* @param ptr The rotate event (minux header and OK byte)
*/
void
blr_slave_rotate(ROUTER_SLAVE *slave, uint8_t *ptr)
{
ptr += 19; // Skip header
slave->binlog_pos = extract_field(ptr, 32);
slave->binlog_pos += (extract_field(ptr+4, 32) << 32);
memcpy(slave->binlogfile, ptr + 8, BINLOG_FNAMELEN);
slave->binlogfile[BINLOG_FNAMELEN] = 0;
}

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

@ -160,6 +160,10 @@ struct subcommand showoptions[] = {
"Show all active sessions in MaxScale",
"Show all active sessions in MaxScale",
{0, 0, 0} },
{ "threads", 0, dShowThreads,
"Show the status of the polling threads in MaxScale",
"Show the status of the polling threads in MaxScale",
{0, 0, 0} },
{ "users", 0, telnetdShowUsers,
"Show statistics and user names for the debug interface",
"Show statistics and user names for the debug interface",
@ -208,6 +212,10 @@ struct subcommand listoptions[] = {
"List all the active sessions within MaxScale",
"List all the active sessions within MaxScale",
{0, 0, 0} },
{ "threads", 0, dShowThreads,
"List the status of the polling threads in MaxScale",
"List the status of the polling threads in MaxScale",
{0, 0, 0} },
{ NULL, 0, NULL, NULL, NULL,
{0, 0, 0} }
};

13
server/modules/routing/readconnroute.c
View File

@ -264,11 +264,11 @@ char *weightby;
{
for (n = 0; inst->servers[n]; n++)
{
int perc;
int perc, wght;
backend = inst->servers[n];
perc = (atoi(serverGetParameter(backend->server,
weightby)) * 1000) / total;
if (perc == 0)
perc = ((wght = atoi(serverGetParameter(backend->server,
weightby))) * 1000) / total;
if (perc == 0 && wght != 0)
perc = 1;
backend->weight = perc;
if (perc == 0)
@ -279,7 +279,7 @@ char *weightby;
"for weighting parameter '%s', "
"no queries will be routed to "
"this server.\n",
server->unique_name,
inst->servers[n]->server->unique_name,
weightby)));
}
@ -418,6 +418,9 @@ BACKEND *master_host = NULL;
if (SERVER_IN_MAINT(inst->servers[i]->server))
continue;
if (inst->servers[i]->weight == 0)
continue;
/* Check server status bits against bitvalue from router_options */
if (inst->servers[i] &&
SERVER_IS_RUNNING(inst->servers[i]->server) &&

3
server/modules/routing/readwritesplit/.gitignore vendored
View File

@ -1,3 +0,0 @@
*.o
*.so
depend.mk

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

File diff suppressed because it is too large Load Diff

4
server/modules/routing/readwritesplit/test/rwsplit.sh
View File

@ -279,9 +279,9 @@ do
printf "."
fi
b=`$RUNCMD < $TINPUT 2>&1`
if [[ "`echo "$b"|grep -i 'null|error'`" != "" ]]
if [[ "`echo "$b"|grep -i 'null\|error'`" != "" ]]
then
err=`echo "$b" | grep -i null|error`
err=`echo "$b" | grep -i null\|error`
break
fi
done