Node Replication Consistency
Node Replication Consistency
This commit is contained in:
MassimilianoPinto
parent
7108a25299
commit
e4545baf97
@ -291,20 +291,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)
|
||||
{
|
||||
@ -371,15 +372,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
|
||||
@ -424,6 +522,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);
|
||||
@ -474,7 +646,7 @@ MONITOR_SERVERS *ptr;
|
||||
ptr = handle->databases;
|
||||
while (ptr)
|
||||
{
|
||||
monitorDatabase(ptr, handle->defaultUser, handle->defaultPasswd);
|
||||
monitorDatabase(handle, ptr);
|
||||
ptr = ptr->next;
|
||||
}
|
||||
thread_millisleep(handle->interval);
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user