Merge branch 'develop' of github.com:skysql/MaxScale into develop
This commit is contained in:
commit
752f28ce48
@ -31,6 +31,8 @@
|
||||
* 11/03/14 Massimiliano Pinto Added Unix socket support
|
||||
* 11/05/14 Massimiliano Pinto Added version_string support to service
|
||||
* 19/05/14 Mark Riddoch Added unique names from section headers
|
||||
* 23/05/14 Massimiliano Pinto Added automatic set of maxscale-id: first listening ipv4_raw + port + pid
|
||||
* 28/05/14 Massimiliano Pinto Added detect_replication_lag parameter
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -426,12 +428,19 @@ int error_count = 0;
|
||||
char *port;
|
||||
char *protocol;
|
||||
char *socket;
|
||||
struct sockaddr_in serv_addr;
|
||||
|
||||
service = config_get_value(obj->parameters, "service");
|
||||
port = config_get_value(obj->parameters, "port");
|
||||
address = config_get_value(obj->parameters, "address");
|
||||
protocol = config_get_value(obj->parameters, "protocol");
|
||||
socket = config_get_value(obj->parameters, "socket");
|
||||
|
||||
/* if id is not set, do it now */
|
||||
if (gateway.id == 0) {
|
||||
setipaddress(&serv_addr.sin_addr, (address == NULL) ? "0.0.0.0" : address);
|
||||
gateway.id = (unsigned long) (serv_addr.sin_addr.s_addr + port + getpid());
|
||||
}
|
||||
|
||||
if (service && socket && protocol) {
|
||||
CONFIG_CONTEXT *ptr = context;
|
||||
@ -494,18 +503,46 @@ int error_count = 0;
|
||||
char *servers;
|
||||
char *user;
|
||||
char *passwd;
|
||||
unsigned long interval = 0;
|
||||
int replication_heartbeat = 0;
|
||||
|
||||
module = config_get_value(obj->parameters, "module");
|
||||
servers = config_get_value(obj->parameters, "servers");
|
||||
user = config_get_value(obj->parameters, "user");
|
||||
passwd = config_get_value(obj->parameters, "passwd");
|
||||
if (config_get_value(obj->parameters, "monitor_interval")) {
|
||||
interval = strtoul(config_get_value(obj->parameters, "monitor_interval"), NULL, 10);
|
||||
}
|
||||
|
||||
if (config_get_value(obj->parameters, "detect_replication_lag")) {
|
||||
replication_heartbeat = atoi(config_get_value(obj->parameters, "detect_replication_lag"));
|
||||
}
|
||||
|
||||
if (module)
|
||||
{
|
||||
obj->element = monitor_alloc(obj->object, module);
|
||||
if (servers && obj->element)
|
||||
{
|
||||
char *s = strtok(servers, ",");
|
||||
char *s;
|
||||
|
||||
/* if id is not set, compute it now with pid only */
|
||||
if (gateway.id == 0) {
|
||||
gateway.id = getpid();
|
||||
}
|
||||
|
||||
/* add the maxscale-id to monitor data */
|
||||
monitorSetId(obj->element, gateway.id);
|
||||
|
||||
/* set monitor interval */
|
||||
if (interval > 0)
|
||||
monitorSetInterval(obj->element, interval);
|
||||
|
||||
/* set replication heartbeat */
|
||||
if(replication_heartbeat == 1)
|
||||
monitorSetReplicationHeartbeat(obj->element, replication_heartbeat);
|
||||
|
||||
/* get the servers to monitor */
|
||||
s = strtok(servers, ",");
|
||||
while (s)
|
||||
{
|
||||
CONFIG_CONTEXT *obj1 = context;
|
||||
@ -747,6 +784,7 @@ global_defaults()
|
||||
gateway.version_string = strdup(version_string);
|
||||
else
|
||||
gateway.version_string = NULL;
|
||||
gateway.id=0;
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1112,6 +1150,8 @@ static char *monitor_params[] =
|
||||
"servers",
|
||||
"user",
|
||||
"passwd",
|
||||
"monitor_interval",
|
||||
"detect_replication_lag",
|
||||
NULL
|
||||
};
|
||||
/**
|
||||
|
@ -22,8 +22,10 @@
|
||||
* @verbatim
|
||||
* Revision History
|
||||
*
|
||||
* Date Who Description
|
||||
* 08/07/13 Mark Riddoch Initial implementation
|
||||
* Date Who Description
|
||||
* 08/07/13 Mark Riddoch Initial implementation
|
||||
* 23/05/14 Massimiliano Pinto Addition of monitor_interval parameter
|
||||
* and monitor id
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -220,3 +222,47 @@ MONITOR *ptr;
|
||||
spinlock_release(&monLock);
|
||||
return ptr;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Set the id of the monitor.
|
||||
*
|
||||
* @param mon The monitor instance
|
||||
* @param id The id for the monitor
|
||||
*/
|
||||
|
||||
void
|
||||
monitorSetId(MONITOR *mon, unsigned long id)
|
||||
{
|
||||
if (mon->module->defaultId != NULL) {
|
||||
mon->module->defaultId(mon->handle, id);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the monitor sampling interval.
|
||||
*
|
||||
* @param mon The monitor instance
|
||||
* @param interval The sampling interval in milliseconds
|
||||
*/
|
||||
void
|
||||
monitorSetInterval (MONITOR *mon, unsigned long interval)
|
||||
{
|
||||
if (mon->module->setInterval != NULL) {
|
||||
mon->module->setInterval(mon->handle, interval);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Enable Replication Heartbeat support in monitor.
|
||||
*
|
||||
* @param mon The monitor instance
|
||||
* @param interval The sampling interval in milliseconds
|
||||
*/
|
||||
void
|
||||
monitorSetReplicationHeartbeat(MONITOR *mon, int replication_heartbeat)
|
||||
{
|
||||
if (mon->module->replicationHeartbeat != NULL) {
|
||||
mon->module->replicationHeartbeat(mon->handle, replication_heartbeat);
|
||||
}
|
||||
}
|
||||
|
@ -27,6 +27,7 @@
|
||||
* 17/05/14 Mark Riddoch Addition of unique_name
|
||||
* 20/05/14 Massimiliano Pinto Addition of server_string
|
||||
* 21/05/14 Massimiliano Pinto Addition of node_id
|
||||
* 28/05/14 Massimiliano Pinto Addition of rlagd and node_ts fields
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -73,6 +74,8 @@ SERVER *server;
|
||||
server->unique_name = NULL;
|
||||
server->server_string = NULL;
|
||||
server->node_id = -1;
|
||||
server->rlag = -1;
|
||||
server->node_ts = 0;
|
||||
|
||||
spinlock_acquire(&server_spin);
|
||||
server->next = allServers;
|
||||
@ -247,6 +250,14 @@ char *stat;
|
||||
if (ptr->server_string)
|
||||
dcb_printf(dcb, "\tServer Version:\t\t%s\n", ptr->server_string);
|
||||
dcb_printf(dcb, "\tNode Id: %d\n", ptr->node_id);
|
||||
if (SERVER_IS_SLAVE(ptr)) {
|
||||
if (ptr->rlag >= 0) {
|
||||
dcb_printf(dcb, "\tSlave delay:\t\t%d\n", ptr->rlag);
|
||||
}
|
||||
}
|
||||
if (ptr->node_ts > 0) {
|
||||
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%lu\n", ptr->node_ts);
|
||||
}
|
||||
dcb_printf(dcb, "\tNumber of connections: %d\n", ptr->stats.n_connections);
|
||||
dcb_printf(dcb, "\tCurrent no. of conns: %d\n", ptr->stats.n_current);
|
||||
ptr = ptr->next;
|
||||
@ -275,6 +286,14 @@ char *stat;
|
||||
if (server->server_string)
|
||||
dcb_printf(dcb, "\tServer Version:\t\t%s\n", server->server_string);
|
||||
dcb_printf(dcb, "\tNode Id: %d\n", server->node_id);
|
||||
if (SERVER_IS_SLAVE(server)) {
|
||||
if (server->rlag >= 0) {
|
||||
dcb_printf(dcb, "\tSlave delay:\t\t%d\n", server->rlag);
|
||||
}
|
||||
}
|
||||
if (server->node_ts > 0) {
|
||||
dcb_printf(dcb, "\tLast Repl Heartbeat:\t%lu\n", server->node_ts);
|
||||
}
|
||||
dcb_printf(dcb, "\tNumber of connections: %d\n", server->stats.n_connections);
|
||||
dcb_printf(dcb, "\tCurrent no. of conns: %d\n", server->stats.n_current);
|
||||
}
|
||||
|
@ -28,6 +28,7 @@
|
||||
* Date Who Description
|
||||
* 21/06/13 Mark Riddoch Initial implementation
|
||||
* 07/05/14 Massimiliano Pinto Added version_string to global configuration
|
||||
* 23/05/14 Massimiliano Pinto Added id to global configuration
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -78,6 +79,7 @@ typedef struct config_context {
|
||||
typedef struct {
|
||||
int n_threads; /**< Number of polling threads */
|
||||
char *version_string; /**< The version string of embedded database library */
|
||||
unsigned long id; /**< MaxScale ID */
|
||||
} GATEWAY_CONF;
|
||||
|
||||
extern int config_load(char *);
|
||||
|
@ -26,10 +26,11 @@
|
||||
* @verbatim
|
||||
* Revision History
|
||||
*
|
||||
* Date Who Description
|
||||
* 07/07/13 Mark Riddoch Initial implementation
|
||||
* 25/07/13 Mark Riddoch Addition of diagnotics
|
||||
* 23/05/14 Mark Riddoch Addition of routine to find monitors by name
|
||||
* Date Who Description
|
||||
* 07/07/13 Mark Riddoch Initial implementation
|
||||
* 25/07/13 Mark Riddoch Addition of diagnotics
|
||||
* 23/05/14 Mark Riddoch Addition of routine to find monitors by name
|
||||
* 23/05/14 Massimiliano Pinto Addition of defaultId and setInterval
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -66,6 +67,9 @@ typedef struct {
|
||||
void (*unregisterServer)(void *, SERVER *);
|
||||
void (*defaultUser)(void *, char *, char *);
|
||||
void (*diagnostics)(DCB *, void *);
|
||||
void (*setInterval)(void *, unsigned long);
|
||||
void (*defaultId)(void *, unsigned long);
|
||||
void (*replicationHeartbeat)(void *, int);
|
||||
} MONITOR_OBJECT;
|
||||
|
||||
/**
|
||||
@ -87,4 +91,7 @@ extern void monitorStop(MONITOR *);
|
||||
extern void monitorStart(MONITOR *);
|
||||
extern void monitorStopAll();
|
||||
extern void monitorShowAll(DCB *);
|
||||
extern void monitorSetId(MONITOR *, unsigned long);
|
||||
extern void monitorSetInterval (MONITOR *, unsigned long);
|
||||
extern void monitorSetReplicationHeartbeat(MONITOR *, int);
|
||||
#endif
|
||||
|
@ -34,6 +34,7 @@
|
||||
* 18/05/14 Mark Riddoch Addition of unique_name field
|
||||
* 20/05/14 Massimiliano Pinto Addition of server_string field
|
||||
* 20/05/14 Massimiliano Pinto Addition of node_id field
|
||||
* 23/05/14 Massimiliano Pinto Addition of rlag and node_ts fields
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -66,6 +67,8 @@ typedef struct server {
|
||||
struct server *nextdb; /**< Next server in list attached to a service */
|
||||
char *server_string; /**< Server version string, i.e. MySQL server version */
|
||||
long node_id; /**< Node id, server_id for M/S or local_index for Galera */
|
||||
int rlag; /**< Replication Lag for Master / Slave replication */
|
||||
unsigned long node_ts; /**< Last timestamp set from M/S monitor module */
|
||||
} SERVER;
|
||||
|
||||
/**
|
||||
@ -121,4 +124,5 @@ extern void server_set_status(SERVER *, int);
|
||||
extern void server_clear_status(SERVER *, int);
|
||||
extern void serverAddMonUser(SERVER *, char *, char *);
|
||||
extern void server_update(SERVER *, char *, char *, char *);
|
||||
extern void server_set_unique_name(SERVER *, char *);
|
||||
#endif
|
||||
|
@ -26,6 +26,8 @@
|
||||
* 22/07/13 Mark Riddoch Initial implementation
|
||||
* 21/05/14 Massimiliano Pinto Monitor sets a master server
|
||||
* that has the lowest value of wsrep_local_index
|
||||
* 23/05/14 Massimiliano Pinto Added 1 configuration option (setInterval).
|
||||
* Interval is printed in diagnostics.
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -47,7 +49,7 @@ extern int lm_enabled_logfiles_bitmask;
|
||||
|
||||
static void monitorMain(void *);
|
||||
|
||||
static char *version_str = "V1.1.0";
|
||||
static char *version_str = "V1.2.0";
|
||||
|
||||
static void *startMonitor(void *);
|
||||
static void stopMonitor(void *);
|
||||
@ -55,8 +57,9 @@ static void registerServer(void *, SERVER *);
|
||||
static void unregisterServer(void *, SERVER *);
|
||||
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 };
|
||||
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUsers, diagnostics, setInterval, NULL, NULL };
|
||||
|
||||
/**
|
||||
* Implementation of the mandatory version entry point
|
||||
@ -121,6 +124,8 @@ MYSQL_MONITOR *handle;
|
||||
handle->shutdown = 0;
|
||||
handle->defaultUser = NULL;
|
||||
handle->defaultPasswd = NULL;
|
||||
handle->id = MONITOR_DEFAULT_ID;
|
||||
handle->interval = MONITOR_INTERVAL;
|
||||
spinlock_init(&handle->lock);
|
||||
}
|
||||
handle->tid = (THREAD)thread_start(monitorMain, handle);
|
||||
@ -236,7 +241,10 @@ char *sep;
|
||||
dcb_printf(dcb, "\tMonitor stopped\n");
|
||||
break;
|
||||
}
|
||||
|
||||
dcb_printf(dcb,"\tSampling interval:\t%lu milliseconds\n", handle->interval);
|
||||
dcb_printf(dcb, "\tMonitored servers: ");
|
||||
|
||||
db = handle->databases;
|
||||
sep = "";
|
||||
while (db)
|
||||
@ -434,6 +442,19 @@ long master_id;
|
||||
|
||||
ptr = ptr->next;
|
||||
}
|
||||
thread_millisleep(MONITOR_INTERVAL);
|
||||
thread_millisleep(handle->interval);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the monitor sampling interval.
|
||||
*
|
||||
* @param arg The handle allocated by startMonitor
|
||||
* @param interval The interval to set in monitor struct, in milliseconds
|
||||
*/
|
||||
static void
|
||||
setInterval(void *arg, unsigned long interval)
|
||||
{
|
||||
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
|
||||
memcpy(&handle->interval, &interval, sizeof(unsigned long));
|
||||
}
|
||||
|
@ -30,6 +30,8 @@
|
||||
* diagnostic interface
|
||||
* 20/05/14 Massimiliano Pinto Addition of support for MariadDB multimaster replication setup.
|
||||
* New server field version_string is updated.
|
||||
* 28/05/14 Massimiliano Pinto Added set Id and configuration options (setInverval)
|
||||
* Parameters are now printed in diagnostics
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -51,7 +53,7 @@ extern int lm_enabled_logfiles_bitmask;
|
||||
|
||||
static void monitorMain(void *);
|
||||
|
||||
static char *version_str = "V1.1.0";
|
||||
static char *version_str = "V1.2.0";
|
||||
|
||||
static void *startMonitor(void *);
|
||||
static void stopMonitor(void *);
|
||||
@ -59,8 +61,11 @@ static void registerServer(void *, SERVER *);
|
||||
static void unregisterServer(void *, SERVER *);
|
||||
static void defaultUser(void *, char *, char *);
|
||||
static void diagnostics(DCB *, void *);
|
||||
static void setInterval(void *, unsigned long);
|
||||
static void defaultId(void *, unsigned long);
|
||||
static void replicationHeartbeat(void *, int);
|
||||
|
||||
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUser, diagnostics };
|
||||
static MONITOR_OBJECT MyObject = { startMonitor, stopMonitor, registerServer, unregisterServer, defaultUser, diagnostics, setInterval, defaultId, replicationHeartbeat };
|
||||
|
||||
/**
|
||||
* Implementation of the mandatory version entry point
|
||||
@ -126,6 +131,8 @@ MYSQL_MONITOR *handle;
|
||||
handle->shutdown = 0;
|
||||
handle->defaultUser = NULL;
|
||||
handle->defaultPasswd = NULL;
|
||||
handle->id = MONITOR_DEFAULT_ID;
|
||||
handle->interval = MONITOR_INTERVAL;
|
||||
spinlock_init(&handle->lock);
|
||||
}
|
||||
handle->tid = (THREAD)thread_start(monitorMain, handle);
|
||||
@ -261,7 +268,12 @@ char *sep;
|
||||
dcb_printf(dcb, "\tMonitor stopped\n");
|
||||
break;
|
||||
}
|
||||
|
||||
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, "\tMonitored servers: ");
|
||||
|
||||
db = handle->databases;
|
||||
sep = "";
|
||||
while (db)
|
||||
@ -280,20 +292,21 @@ char *sep;
|
||||
/**
|
||||
* Monitor an individual server
|
||||
*
|
||||
* @param database The database to probe
|
||||
* @param defaultUser Default username for the monitor
|
||||
* @param defaultPasswd Default password for the monitor
|
||||
* @param handle The MySQL Monitor object
|
||||
* @param database The database to probe
|
||||
*/
|
||||
static void
|
||||
monitorDatabase(MONITOR_SERVERS *database, char *defaultUser, char *defaultPasswd)
|
||||
monitorDatabase(MYSQL_MONITOR *handle, MONITOR_SERVERS *database)
|
||||
{
|
||||
MYSQL_ROW row;
|
||||
MYSQL_RES *result;
|
||||
int num_fields;
|
||||
int ismaster = 0, isslave = 0;
|
||||
char *uname = defaultUser, *passwd = defaultPasswd;
|
||||
char *uname = handle->defaultUser, *passwd = handle->defaultPasswd;
|
||||
unsigned long int server_version = 0;
|
||||
char *server_string;
|
||||
unsigned long id = handle->id;
|
||||
int replication_heartbeat = handle->replicationHeartbeat;
|
||||
|
||||
if (database->server->monuser != NULL)
|
||||
{
|
||||
@ -360,15 +373,112 @@ char *server_string;
|
||||
{
|
||||
/* Log lack of permission */
|
||||
}
|
||||
}
|
||||
else if ((result = mysql_store_result(database->con)) != NULL)
|
||||
{
|
||||
|
||||
database->server->rlag = -1;
|
||||
} else if ((result = mysql_store_result(database->con)) != NULL) {
|
||||
num_fields = mysql_num_fields(result);
|
||||
while ((row = mysql_fetch_row(result)))
|
||||
{
|
||||
ismaster = 1;
|
||||
}
|
||||
mysql_free_result(result);
|
||||
|
||||
if (ismaster && replication_heartbeat == 1) {
|
||||
time_t heartbeat;
|
||||
time_t purge_time;
|
||||
char heartbeat_insert_query[128]="";
|
||||
char heartbeat_purge_query[128]="";
|
||||
|
||||
handle->master_id = database->server->node_id;
|
||||
|
||||
/* create the maxscale_schema database */
|
||||
if (mysql_query(database->con, "CREATE DATABASE IF NOT EXISTS maxscale_schema")) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: Error creating maxscale_schema database in Master server"
|
||||
": %s", mysql_error(database->con))));
|
||||
|
||||
database->server->rlag = -1;
|
||||
}
|
||||
|
||||
/* create repl_heartbeat table in maxscale_schema database */
|
||||
if (mysql_query(database->con, "CREATE TABLE IF NOT EXISTS "
|
||||
"maxscale_schema.replication_heartbeat "
|
||||
"(maxscale_id INT NOT NULL, "
|
||||
"master_server_id INT NOT NULL, "
|
||||
"master_timestamp INT UNSIGNED NOT NULL, "
|
||||
"PRIMARY KEY ( master_server_id, maxscale_id ) ) "
|
||||
"ENGINE=MYISAM DEFAULT CHARSET=latin1")) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: Error creating maxscale_schema.replication_heartbeat table in Master server"
|
||||
": %s", mysql_error(database->con))));
|
||||
|
||||
database->server->rlag = -1;
|
||||
}
|
||||
|
||||
/* auto purge old values after 48 hours*/
|
||||
purge_time = time(0) - (3600 * 48);
|
||||
|
||||
sprintf(heartbeat_purge_query, "DELETE FROM maxscale_schema.replication_heartbeat WHERE master_timestamp < %lu", purge_time);
|
||||
|
||||
if (mysql_query(database->con, heartbeat_purge_query)) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: Error deleting from maxscale_schema.replication_heartbeat table: [%s], %s",
|
||||
heartbeat_purge_query,
|
||||
mysql_error(database->con))));
|
||||
}
|
||||
|
||||
heartbeat = time(0);
|
||||
|
||||
/* set node_ts for master as time(0) */
|
||||
database->server->node_ts = heartbeat;
|
||||
|
||||
sprintf(heartbeat_insert_query, "UPDATE maxscale_schema.replication_heartbeat SET master_timestamp = %lu WHERE master_server_id = %i AND maxscale_id = %lu", heartbeat, handle->master_id, id);
|
||||
|
||||
/* Try to insert MaxScale timestamp into master */
|
||||
if (mysql_query(database->con, heartbeat_insert_query)) {
|
||||
|
||||
database->server->rlag = -1;
|
||||
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: Error updating maxscale_schema.replication_heartbeat table: [%s], %s",
|
||||
heartbeat_insert_query,
|
||||
mysql_error(database->con))));
|
||||
} else {
|
||||
if (mysql_affected_rows(database->con) == 0) {
|
||||
heartbeat = time(0);
|
||||
sprintf(heartbeat_insert_query, "REPLACE INTO maxscale_schema.replication_heartbeat (master_server_id, maxscale_id, master_timestamp ) VALUES ( %i, %lu, %lu)", handle->master_id, id, heartbeat);
|
||||
|
||||
if (mysql_query(database->con, heartbeat_insert_query)) {
|
||||
|
||||
database->server->rlag = -1;
|
||||
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: Error inserting into maxscale_schema.replication_heartbeat table: [%s], %s",
|
||||
heartbeat_insert_query,
|
||||
mysql_error(database->con))));
|
||||
} else {
|
||||
/* Set replication lag to 0 for the master */
|
||||
database->server->rlag = 0;
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"[mysql_mon]: heartbeat table inserted data for %s:%i", database->server->name, database->server->port)));
|
||||
}
|
||||
} else {
|
||||
/* Set replication lag as 0 for the master */
|
||||
database->server->rlag = 0;
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"[mysql_mon]: heartbeat table updated for %s:%i", database->server->name, database->server->port)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/* Check if the Slave_SQL_Running and Slave_IO_Running status is
|
||||
@ -413,6 +523,80 @@ char *server_string;
|
||||
}
|
||||
}
|
||||
|
||||
/* Get the master_timestamp value from maxscale_schema.replication_heartbeat table */
|
||||
if (isslave && replication_heartbeat == 1) {
|
||||
time_t heartbeat;
|
||||
char select_heartbeat_query[256] = "";
|
||||
|
||||
sprintf(select_heartbeat_query, "SELECT master_timestamp "
|
||||
"FROM maxscale_schema.replication_heartbeat "
|
||||
"WHERE maxscale_id = %lu AND master_server_id = %i",
|
||||
id, handle->master_id);
|
||||
|
||||
/* if there is a master then send the query to the slave with master_id*/
|
||||
if (handle->master_id >= 0 && (mysql_query(database->con, select_heartbeat_query) == 0
|
||||
&& (result = mysql_store_result(database->con)) != NULL)) {
|
||||
num_fields = mysql_num_fields(result);
|
||||
|
||||
while ((row = mysql_fetch_row(result))) {
|
||||
int rlag = -1;
|
||||
time_t slave_read;
|
||||
|
||||
heartbeat = time(0);
|
||||
slave_read = strtoul(row[0], NULL, 10);
|
||||
|
||||
if ((errno == ERANGE && (slave_read == LONG_MAX || slave_read == LONG_MIN)) || (errno != 0 && slave_read == 0)) {
|
||||
slave_read = 0;
|
||||
}
|
||||
|
||||
if (slave_read) {
|
||||
/* set the replication lag */
|
||||
rlag = heartbeat - slave_read;
|
||||
}
|
||||
|
||||
/* set this node_ts as master_timestamp read from replication_heartbeat table */
|
||||
database->server->node_ts = slave_read;
|
||||
|
||||
if (rlag >= 0) {
|
||||
/* store rlag only if greater than monitor sampling interval */
|
||||
database->server->rlag = (rlag > (handle->interval / 1000)) ? rlag : 0;
|
||||
} else {
|
||||
database->server->rlag = -1;
|
||||
}
|
||||
|
||||
LOGIF(LD, (skygw_log_write_flush(
|
||||
LOGFILE_DEBUG,
|
||||
"[mysql_mon]: replication heartbeat: "
|
||||
"server %s:%i is %i seconds behind master",
|
||||
database->server->name,
|
||||
database->server->port,
|
||||
database->server->rlag)));
|
||||
}
|
||||
mysql_free_result(result);
|
||||
} else {
|
||||
database->server->rlag = -1;
|
||||
database->server->node_ts = 0;
|
||||
|
||||
if (handle->master_id < 0) {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: error: replication heartbeat: "
|
||||
"master_server_id NOT available for %s:%i",
|
||||
database->server->name,
|
||||
database->server->port)));
|
||||
} else {
|
||||
LOGIF(LE, (skygw_log_write_flush(
|
||||
LOGFILE_ERROR,
|
||||
"[mysql_mon]: error: replication heartbeat: "
|
||||
"failed selecting from hearthbeat table of %s:%i : [%s], %s",
|
||||
database->server->name,
|
||||
database->server->port,
|
||||
select_heartbeat_query,
|
||||
mysql_error(database->con))));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (ismaster)
|
||||
{
|
||||
server_set_status(database->server, SERVER_MASTER);
|
||||
@ -463,9 +647,48 @@ MONITOR_SERVERS *ptr;
|
||||
ptr = handle->databases;
|
||||
while (ptr)
|
||||
{
|
||||
monitorDatabase(ptr, handle->defaultUser, handle->defaultPasswd);
|
||||
monitorDatabase(handle, ptr);
|
||||
ptr = ptr->next;
|
||||
}
|
||||
thread_millisleep(10000);
|
||||
thread_millisleep(handle->interval);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the default id to use in the monitor.
|
||||
*
|
||||
* @param arg The handle allocated by startMonitor
|
||||
* @param id The id to set in monitor struct
|
||||
*/
|
||||
static void
|
||||
defaultId(void *arg, unsigned long id)
|
||||
{
|
||||
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
|
||||
memcpy(&handle->id, &id, sizeof(unsigned long));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the monitor sampling interval.
|
||||
*
|
||||
* @param arg The handle allocated by startMonitor
|
||||
* @param interval The interval to set in monitor struct, in milliseconds
|
||||
*/
|
||||
static void
|
||||
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
|
||||
*/
|
||||
static void
|
||||
replicationHeartbeat(void *arg, int replicationHeartbeat)
|
||||
{
|
||||
MYSQL_MONITOR *handle = (MYSQL_MONITOR *)arg;
|
||||
memcpy(&handle->replicationHeartbeat, &replicationHeartbeat, sizeof(int));
|
||||
}
|
||||
|
@ -30,6 +30,7 @@
|
||||
* Date Who Description
|
||||
* 08/07/13 Mark Riddoch Initial implementation
|
||||
* 26/05/14 Massimiliano Pinto Default values for MONITOR_INTERVAL
|
||||
* 28/05/14 Massimiliano Pinto Addition of new fields in MYSQL_MONITOR struct
|
||||
*
|
||||
* @endverbatim
|
||||
*/
|
||||
@ -55,6 +56,10 @@ typedef struct {
|
||||
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 */
|
||||
int master_id; /**< Master server-id for MySQL Master/Slave replication */
|
||||
MONITOR_SERVERS *databases; /**< Linked list of servers to monitor */
|
||||
} MYSQL_MONITOR;
|
||||
|
||||
@ -63,5 +68,6 @@ typedef struct {
|
||||
#define MONITOR_STOPPED 3
|
||||
|
||||
#define MONITOR_INTERVAL 10000 // in milliseconds
|
||||
#define MONITOR_DEFAULT_ID 1UL // unsigned long value
|
||||
|
||||
#endif
|
||||
|
Loading…
x
Reference in New Issue
Block a user