Restore servers that are in a shunned state due to replication lag on…

…ly upon commit (triggered by the command LOAD MYSQL SERVERS TO RUNTIME), provided that the replication lag is below the maximum threshold.

include/MySQL_HostGroups_Manager.h

@@ -203,6 +203,7 @@ class MySrvC {	// MySQL Server Container
 	unsigned int max_connections_used; // The maximum number of connections that has been opened
 	unsigned int connect_OK;
 	unsigned int connect_ERR;
+	int cur_replication_lag;
 	unsigned int cur_replication_lag_count;
 	// note that these variables are in microsecond, while user defines max latency in millisecond
 	unsigned int current_latency_us;

include/MySQL_Thread.h

@@ -548,6 +548,7 @@ class MySQL_Threads_Handler
 		bool log_mysql_warnings_enabled;
 		int data_packets_history_size;
 		int handle_warnings;
+		int recover_replication_lag_shunned_servers_on_commit;
 	} variables;
 	struct {
 		unsigned int mirror_sessions_current;

include/proxysql_structs.h

@@ -866,6 +866,7 @@ __thread bool mysql_thread___enable_load_data_local_infile;
 __thread int mysql_thread___client_host_cache_size;
 __thread int mysql_thread___client_host_error_counts;
 __thread int mysql_thread___handle_warnings;
+__thread int mysql_thread___recover_replication_lag_shunned_servers_on_commit;
 
 /* variables used for Query Cache */
 __thread int mysql_thread___query_cache_size_MB;
@@ -1034,6 +1035,7 @@ extern __thread bool mysql_thread___enable_load_data_local_infile;
 extern __thread int mysql_thread___client_host_cache_size;
 extern __thread int mysql_thread___client_host_error_counts;
 extern __thread int mysql_thread___handle_warnings;
+extern __thread int mysql_thread___recover_replication_lag_shunned_servers_on_commit;
 
 /* variables used for Query Cache */
 extern __thread int mysql_thread___query_cache_size_MB;

lib/MySQL_HostGroups_Manager.cpp

@@ -745,6 +745,7 @@ MySrvC::MySrvC(
 	max_connections=_max_connections;
 	max_replication_lag=_max_replication_lag;
 	use_ssl=_use_ssl;
+	cur_replication_lag=0;
 	cur_replication_lag_count=0;
 	max_latency_us=_max_latency_ms*1000;
 	current_latency_us=0;
@@ -2026,9 +2027,25 @@ bool MySQL_HostGroups_Manager::commit(
 					mysrvc->weight=atoi(r->fields[14]);
 				}
 				if (atoi(r->fields[5])!=atoi(r->fields[15])) {
-					if (GloMTH->variables.hostgroup_manager_verbose)
-						proxy_info("Changing status for server %d:%s:%d (%s:%d) from %d (%d) to %d\n" , mysrvc->myhgc->hid , mysrvc->address, mysrvc->port, r->fields[1], atoi(r->fields[2]), atoi(r->fields[5]) , mysrvc->status , atoi(r->fields[15]));
-					mysrvc->status=(MySerStatus)atoi(r->fields[15]);
+					bool change_server_status = true;
+					if (GloMTH->variables.recover_replication_lag_shunned_servers_on_commit == 1) {
+						if (mysrvc->status == MYSQL_SERVER_STATUS_SHUNNED_REPLICATION_LAG && // currently server is shunned due to replication lag
+							(MySerStatus)atoi(r->fields[15]) == MYSQL_SERVER_STATUS_ONLINE) { // new server status is online
+							if (mysrvc->cur_replication_lag != -2) { // Master server? Seconds_Behind_Master column is not present
+								const unsigned int new_max_repl_lag = atoi(r->fields[18]);
+								if (mysrvc->cur_replication_lag < 0 ||
+									(new_max_repl_lag > 0 &&
+									((unsigned int)mysrvc->cur_replication_lag > new_max_repl_lag))) { // we check if current replication lag is greater than new max_replication_lag
+									change_server_status = false;
+								}
+							}
+						}
+					}
+					if (change_server_status == true) {
+						if (GloMTH->variables.hostgroup_manager_verbose)
+							proxy_info("Changing status for server %d:%s:%d (%s:%d) from %d (%d) to %d\n", mysrvc->myhgc->hid, mysrvc->address, mysrvc->port, r->fields[1], atoi(r->fields[2]), atoi(r->fields[5]), mysrvc->status, atoi(r->fields[15]));
+						mysrvc->status = (MySerStatus)atoi(r->fields[15]);
+					}
 					if (mysrvc->status==MYSQL_SERVER_STATUS_SHUNNED) {
 						mysrvc->shunned_automatic=false;
 					}
@@ -3669,6 +3686,7 @@ void MySQL_HostGroups_Manager::replication_lag_action_inner(MyHGC *myhgc, const
 	for (j=0; j<(int)myhgc->mysrvs->cnt(); j++) {
 		MySrvC *mysrvc=(MySrvC *)myhgc->mysrvs->servers->index(j);
 		if (strcmp(mysrvc->address,address)==0 && mysrvc->port==port) {
+			mysrvc->cur_replication_lag = current_replication_lag;
 			if (mysrvc->status==MYSQL_SERVER_STATUS_ONLINE) {
 				if (
 //					(current_replication_lag==-1 )

lib/MySQL_Thread.cpp

@@ -432,6 +432,7 @@ static char * mysql_thread_variables_names[]= {
 	(char *)"query_cache_stores_empty_result",
 	(char *)"data_packets_history_size",
 	(char *)"handle_warnings",
+	(char *)"recover_replication_lag_shunned_servers_on_commit",
 	NULL
 };
 
@@ -896,6 +897,7 @@ MySQL_Threads_Handler::MySQL_Threads_Handler() {
 	variables.client_host_cache_size=0;
 	variables.client_host_error_counts=0;
 	variables.handle_warnings=1;
+	variables.recover_replication_lag_shunned_servers_on_commit=1;
 	variables.connect_retries_on_failure=10;
 	variables.connection_delay_multiplex_ms=0;
 	variables.connection_max_age_ms=0;
@@ -2079,6 +2081,7 @@ char ** MySQL_Threads_Handler::get_variables_list() {
 		VariablesPointers_int["client_host_cache_size"]        = make_tuple(&variables.client_host_cache_size,        0,      1024*1024, false);
 		VariablesPointers_int["client_host_error_counts"]      = make_tuple(&variables.client_host_error_counts,      0,      1024*1024, false);
 		VariablesPointers_int["handle_warnings"]			   = make_tuple(&variables.handle_warnings,				  0,			  1, false);
+		VariablesPointers_int["recover_replication_lag_shunned_servers_on_commit"] = make_tuple(&variables.recover_replication_lag_shunned_servers_on_commit, 0, 1, false);
 
 		// logs
 		VariablesPointers_int["auditlog_filesize"]     = make_tuple(&variables.auditlog_filesize,    1024*1024, 1*1024*1024*1024, false);
@@ -4007,6 +4010,7 @@ void MySQL_Thread::refresh_variables() {
 	mysql_thread___client_host_cache_size=GloMTH->get_variable_int((char *)"client_host_cache_size");
 	mysql_thread___client_host_error_counts=GloMTH->get_variable_int((char *)"client_host_error_counts");
 	mysql_thread___handle_warnings=GloMTH->get_variable_int((char*)"handle_warnings");
+	mysql_thread___recover_replication_lag_shunned_servers_on_commit=GloMTH->get_variable_int((char*)"recover_replication_lag_shunned_servers_on_commit");
 #ifdef DEBUG
 	mysql_thread___session_debug=(bool)GloMTH->get_variable_int((char *)"session_debug");
 #endif /* DEBUG */