Merge pull request #3807 from sysown/v2.x-3583

Resolves #3583 - Session fast forward enabled when COM_BINLOG_DUMP_GTID is received

include/proxysql_structs.h

@@ -356,6 +356,7 @@ enum enum_mysql_command {
 	_MYSQL_COM_SET_OPTION = 27,
 	_MYSQL_COM_STMT_FETCH = 28,
 	_MYSQL_COM_DAEMON,
+	_MYSQL_COM_BINLOG_DUMP_GTID,
 	_MYSQL_COM_RESET_CONNECTION = 31,
 
   _MYSQL_COM_END

lib/MySQL_Session.cpp

@@ -3987,6 +3987,60 @@ int MySQL_Session::get_pkts_from_client(bool& wrong_pass, PtrSize_t& pkt) {
 								break;
 							case _MYSQL_COM_STMT_SEND_LONG_DATA:
 								handler___status_WAITING_CLIENT_DATA___STATE_SLEEP___MYSQL_COM_STMT_SEND_LONG_DATA(pkt);
+								break;
+							case _MYSQL_COM_BINLOG_DUMP:
+							case _MYSQL_COM_REGISTER_SLAVE:
+							case _MYSQL_COM_BINLOG_DUMP_GTID:
+								// In this switch we handle commands that download binlog events from MySQL
+								// servers. For this commands a lot of the features provided by ProxySQL
+								// aren't useful, like multiplexing, query parsing, etc. For this reason,
+								// ProxySQL enable fast_forward when it receives these commands. 
+								proxy_info(
+									"COM_REGISTER_SLAVE, COM_BINLOG_DUMP or COM_BINLOG_DUMP_GTID received. "
+									"Changing session fast foward to true\n"
+								);
+								session_fast_forward = true;
+
+								if (client_myds->PSarrayIN->len) {
+									proxy_error("UNEXPECTED PACKET FROM CLIENT -- PLEASE REPORT A BUG\n");
+									assert(0);
+								}
+								client_myds->PSarrayIN->add(pkt.ptr, pkt.size);
+
+								// The following code prepares the session as if it was configured with fast
+								// forward before receiving the command. This way the state machine will
+								// handle the command automatically.
+								mybe = find_or_create_backend(current_hostgroup); // set a backend
+								mybe->server_myds->reinit_queues(); // reinitialize the queues in the myds . By default, they are not active
+								// We reinitialize the 'wait_until' since this session shouldn't wait for processing as
+								// we are now transitioning to 'FAST_FORWARD'.
+								mybe->server_myds->wait_until = 0;
+								if (mybe->server_myds->DSS==STATE_NOT_INITIALIZED) {
+									// NOTE: This section is entirely borrowed from 'STATE_SLEEP' for 'session_fast_forward'.
+									// Check comments there for extra information.
+									// =============================================================================
+									if (mybe->server_myds->max_connect_time == 0) {
+										uint64_t connect_timeout =
+											mysql_thread___connect_timeout_server < mysql_thread___connect_timeout_server_max ?
+												mysql_thread___connect_timeout_server_max : mysql_thread___connect_timeout_server;
+										mybe->server_myds->max_connect_time = thread->curtime + connect_timeout * 1000;
+									}
+									mybe->server_myds->connect_retries_on_failure = mysql_thread___connect_retries_on_failure;
+									CurrentQuery.start_time=thread->curtime;
+									// =============================================================================
+
+									// we don't have a connection
+									previous_status.push(FAST_FORWARD); // next status will be FAST_FORWARD
+									set_status(CONNECTING_SERVER); // now we need a connection
+								} else {
+									// In case of having a connection, we need to make user to reset the state machine
+									// for current server 'MySQL_Data_Stream', setting it outside of any state handled
+									// by 'mariadb' library. Otherwise 'MySQL_Thread' will threat this
+									// 'MySQL_Data_Stream' as library handled.
+									mybe->server_myds->DSS = STATE_READY;
+									set_status(FAST_FORWARD); // we can set status to FAST_FORWARD
+								}
+
 								break;
 							case _MYSQL_COM_QUIT:
 								proxy_debug(PROXY_DEBUG_MYSQL_COM, 5, "Got COM_QUIT packet\n");

test/tap/tests/test_com_binlog_dump_enables_fast_forward-t.cpp

@@ -0,0 +1,29 @@
+/**
+ * @file test_com_binlog_dump_enables_fast_forward-t.cpp
+ * @brief Test COM_BINLOG_DUMP enables fast forward.
+ * @details Test checks if mysqlbinlog is executed successfully using a user
+ * with fast forward flag set to false. mysqlginlog sends command
+ * COM_BINLOG_DUMP, then ProxySQL enables fast forward.
+ */
+
+#include "tap.h"
+#include "command_line.h"
+
+int main(int argc, char** argv) {
+	CommandLine cl;
+
+	if (cl.getEnv()) {
+		diag("Failed to get the required environmental variables.");
+		return -1;
+	}
+
+	const std::string user = "root";
+	const std::string test_deps_path = getenv("TEST_DEPS");
+
+	const int mysqlbinlog_res = system((test_deps_path + "/mysqlbinlog mysql1-bin.000001 "
+										"--read-from-remote-server --user " + user + " --password=" + user +
+										" --host 127.0.0.1 --port 6033").c_str());
+	ok(mysqlbinlog_res == 0, "'mysqlbinlog' should be correctly executed. Err code was: %d", mysqlbinlog_res);
+
+	return exit_status();
+}

test/tap/tests/test_com_register_slave_enables_fast_forward-t.cpp

@@ -0,0 +1,27 @@
+/**
+ * @file test_com_register_slave_enables_fast_forward-t.cpp @brief Test
+ * COM_REGISTER_SLAVE enables fast forward.  @details Test checks if
+ * test_binlog_reader is executed successfully using a user with fast forward
+ * flag set to false. test_binlog_reader sends command COM_REGISTER_SLAVE, then
+ * ProxySQL enables fast forward. test_binlog_reader then uses libslave to
+ * listen binlog events. It listen two times, one after sending a query that do
+ * not disable multiplexing and the other after sending a query that disables
+ * multiplexing.
+ */
+
+#include <string>
+
+#include "tap.h"
+
+int main(int argc, char** argv) {
+	const std::string test_deps_path = getenv("TEST_DEPS");
+
+	const int test_binlog_reader_res = system((test_deps_path + "/test_binlog_reader-t").c_str());
+	ok(
+		test_binlog_reader_res == 0,
+		"'test_binlog_reader-t' should be correctly executed. Err code was: %d",
+		test_binlog_reader_res
+	);
+
+	return exit_status();
+}