Added COM_SEND_REPLICA_STATISTICS & background thread to send lag sta…

…tistics from secondary to primary and store it in information_schema.replica_statistics

Summary:
This diff is a port of D21440060 (facebook@06e7367) (facebook@06e7367) from mysql-5.6.35 to mysql-8.0

* I am adding a new RPC `COM_SEND_REPLICA_STATISTICS`. Slaves will use this RPC to send lag statistics to master. This will be done in the next diff.

* In addition, I have added the information_schema table named `replica_statistics` to store the slave lag stats.

* Added a new background thread that is started when the `mysqld` process starts and continuously publishes lag statistics from slaves to master every `write_stats_frequency` seconds.

The default values of `write_stats_frequency` is set to 0, which means do not send lag statistics to master. The unit for this sys_var is seconds.

Also added another sys_var `write_stats_count` to control the number of data points to cache in replica_statistics table per secondary

**Points to note -**

* The background thread re-uses the connection to master to send stats. It does not reconnect every cycle.
* If it is not able to connect to the master in one cycle, it retries the connection in successive cycles until it is able to connect. After this point, it reuses the same connection.
* In case of topology changes, the thread is able to reconnect to the new master and send stats.

Differential Revision: D24659846

client/check/mysqlcheck_core.cc

@@ -214,7 +214,7 @@ static int use_db(const string &database) {
     return 1;
   if (mysql_get_server_version(sock) >= FIRST_PERFORMANCE_SCHEMA_VERSION &&
       !my_strcasecmp(&my_charset_latin1, database.c_str(),
-                     PERFORMANCE_SCHEMA_DB_NAME))
+                     PERFORMANCE_SCHEMA_DB_NAME_MACRO))
     return 1;
   if (mysql_select_db(sock, database.c_str())) {
     DBError(sock, "when selecting the database");

client/client_priv.h

@@ -225,7 +225,7 @@ enum options_client {
 /**
   Name of the performance schema database.
 */
-#define PERFORMANCE_SCHEMA_DB_NAME "performance_schema"
+#define PERFORMANCE_SCHEMA_DB_NAME_MACRO "performance_schema"
 
 /**
   First mysql version supporting the sys schema.

client/dump/mysqldump_tool_chain_maker_options.cc

@@ -210,7 +210,7 @@ void Mysqldump_tool_chain_maker_options::process_positional_options(
     m_object_filter.m_databases_excluded.push_back(
         std::make_pair("", INFORMATION_SCHEMA_DB_NAME));
     m_object_filter.m_databases_excluded.push_back(
-        std::make_pair("", PERFORMANCE_SCHEMA_DB_NAME));
+        std::make_pair("", PERFORMANCE_SCHEMA_DB_NAME_MACRO));
     m_object_filter.m_databases_excluded.push_back(
         std::make_pair("", "ndbinfo"));
     m_object_filter.m_databases_excluded.push_back(std::make_pair("", "sys"));

client/mysqldump.cc

@@ -4766,7 +4766,8 @@ static int dump_all_databases() {
       continue;
 
     if (mysql_get_server_version(mysql) >= FIRST_PERFORMANCE_SCHEMA_VERSION &&
-        !my_strcasecmp(&my_charset_latin1, row[0], PERFORMANCE_SCHEMA_DB_NAME))
+        !my_strcasecmp(&my_charset_latin1, row[0],
+                       PERFORMANCE_SCHEMA_DB_NAME_MACRO))
       continue;
 
     if (mysql_get_server_version(mysql) >= FIRST_SYS_SCHEMA_VERSION &&
@@ -4814,7 +4815,7 @@ static int dump_all_databases() {
 
       if (mysql_get_server_version(mysql) >= FIRST_PERFORMANCE_SCHEMA_VERSION &&
           !my_strcasecmp(&my_charset_latin1, row[0],
-                         PERFORMANCE_SCHEMA_DB_NAME))
+                         PERFORMANCE_SCHEMA_DB_NAME_MACRO))
         continue;
 
       if (mysql_get_server_version(mysql) >= FIRST_SYS_SCHEMA_VERSION &&
@@ -5249,7 +5250,8 @@ static int dump_selected_tables(char *db, char **table_names, int tables) {
       !(mysql_get_server_version(mysql) >= FIRST_INFORMATION_SCHEMA_VERSION &&
         !my_strcasecmp(&my_charset_latin1, db, INFORMATION_SCHEMA_DB_NAME)) &&
       !(mysql_get_server_version(mysql) >= FIRST_PERFORMANCE_SCHEMA_VERSION &&
-        !my_strcasecmp(&my_charset_latin1, db, PERFORMANCE_SCHEMA_DB_NAME))) {
+        !my_strcasecmp(&my_charset_latin1, db,
+                       PERFORMANCE_SCHEMA_DB_NAME_MACRO))) {
     if (mysql_real_query(mysql, lock_tables_query.str,
                          (ulong)(lock_tables_query.length - 1))) {
       if (!opt_force) {

include/map_helpers.h

@@ -87,13 +87,24 @@ typename Container::iterator erase_specific_element(
 template <class T>
 using unique_ptr_with_deleter = std::unique_ptr<T, void (*)(T *)>;
 
+template <typename T, bool call_destructor>
 struct My_free_deleter {
+  void operator()(void *ptr) const {
+    reinterpret_cast<T *>(ptr)->~T();
+    my_free(ptr);
+  }
+};
+
+template <typename T>
+struct My_free_deleter<T, false> {
   void operator()(void *ptr) const { my_free(ptr); }
 };
 
 /** std::unique_ptr, but with my_free as deleter. */
 template <class T>
-using unique_ptr_my_free = std::unique_ptr<T, My_free_deleter>;
+using unique_ptr_my_free = std::unique_ptr<
+    T, My_free_deleter<T, !std::is_array<T>::value &&
+                              !std::is_trivially_destructible<T>::value>>;
 
 struct Free_deleter {
   void operator()(void *ptr) const { free(ptr); }

include/my_command.h

@@ -102,7 +102,10 @@ enum enum_server_command {
     The following are Facebook specific commands. They are put at the top end
     to avoid conflicting with upstream.
   */
+  COM_TOP_BEGIN = 253,
+  COM_SEND_REPLICA_STATISTICS = 254,
   COM_QUERY_ATTRS = 255,
+  COM_TOP_END = 256,
 };
 
 #endif /* _mysql_command_h */

include/mysql.h.pp

@@ -88,7 +88,10 @@
   COM_CLONE,
   COM_SUBSCRIBE_GROUP_REPLICATION_STREAM,
   COM_END,
+  COM_TOP_BEGIN = 253,
+  COM_SEND_REPLICA_STATISTICS = 254,
   COM_QUERY_ATTRS = 255,
+  COM_TOP_END = 256,
 };
 #include "my_compress.h"
 enum enum_compression_algorithm {

include/mysql/plugin_audit.h.pp

@@ -180,7 +180,10 @@
   COM_CLONE,
   COM_SUBSCRIBE_GROUP_REPLICATION_STREAM,
   COM_END,
+  COM_TOP_BEGIN = 253,
+  COM_SEND_REPLICA_STATISTICS = 254,
   COM_QUERY_ATTRS = 255,
+  COM_TOP_END = 256,
 };
 #include "my_sqlcommand.h"
 enum enum_sql_command {

mysql-test/include/check-testcase.test

@@ -96,6 +96,7 @@ if ($tmp) {
   --echo Source_public_key_path	
   --echo Get_Source_public_key	0
   --echo Network_Namespace	
+  --echo Slave_Lag_Stats_Thread_Running	No
 }
 
 if (!$tmp) {

mysql-test/r/disabled_replication.result

@@ -1,5 +1,5 @@
 SHOW SLAVE STATUS;
-Slave_IO_State	Master_Host	Master_User	Master_Port	Connect_Retry	Master_Log_File	Read_Master_Log_Pos	Relay_Log_File	Relay_Log_Pos	Relay_Master_Log_File	Slave_IO_Running	Slave_SQL_Running	Replicate_Do_DB	Replicate_Ignore_DB	Replicate_Do_Table	Replicate_Ignore_Table	Replicate_Wild_Do_Table	Replicate_Wild_Ignore_Table	Last_Errno	Last_Symbolic_Errno	Last_Error	Skip_Counter	Exec_Master_Log_Pos	Relay_Log_Space	Until_Condition	Until_Log_File	Until_Log_Pos	Master_SSL_Allowed	Master_SSL_CA_File	Master_SSL_CA_Path	Master_SSL_Cert	Master_SSL_Cipher	Master_SSL_Key	Seconds_Behind_Master	Master_SSL_Verify_Server_Cert	Last_IO_Errno	Last_IO_Error	Last_SQL_Errno	Last_SQL_Error	Replicate_Ignore_Server_Ids	Master_Server_Id	Master_UUID	Master_Info_File	SQL_Delay	SQL_Remaining_Delay	Slave_SQL_Running_State	Master_Retry_Count	Master_Bind	Last_IO_Error_Timestamp	Last_SQL_Error_Timestamp	Master_SSL_Crl	Master_SSL_Crlpath	Retrieved_Gtid_Set	Executed_Gtid_Set	Auto_Position	Replicate_Rewrite_DB	Channel_Name	Master_TLS_Version	Master_public_key_path	Get_master_public_key	Network_Namespace
+Slave_IO_State	Master_Host	Master_User	Master_Port	Connect_Retry	Master_Log_File	Read_Master_Log_Pos	Relay_Log_File	Relay_Log_Pos	Relay_Master_Log_File	Slave_IO_Running	Slave_SQL_Running	Replicate_Do_DB	Replicate_Ignore_DB	Replicate_Do_Table	Replicate_Ignore_Table	Replicate_Wild_Do_Table	Replicate_Wild_Ignore_Table	Last_Errno	Last_Symbolic_Errno	Last_Error	Skip_Counter	Exec_Master_Log_Pos	Relay_Log_Space	Until_Condition	Until_Log_File	Until_Log_Pos	Master_SSL_Allowed	Master_SSL_CA_File	Master_SSL_CA_Path	Master_SSL_Cert	Master_SSL_Cipher	Master_SSL_Key	Seconds_Behind_Master	Master_SSL_Verify_Server_Cert	Last_IO_Errno	Last_IO_Error	Last_SQL_Errno	Last_SQL_Error	Replicate_Ignore_Server_Ids	Master_Server_Id	Master_UUID	Master_Info_File	SQL_Delay	SQL_Remaining_Delay	Slave_SQL_Running_State	Master_Retry_Count	Master_Bind	Last_IO_Error_Timestamp	Last_SQL_Error_Timestamp	Master_SSL_Crl	Master_SSL_Crlpath	Retrieved_Gtid_Set	Executed_Gtid_Set	Auto_Position	Replicate_Rewrite_DB	Channel_Name	Master_TLS_Version	Master_public_key_path	Get_master_public_key	Network_Namespace	Slave_Lag_Stats_Thread_Running
 Warnings:
 Warning	1287	'SHOW SLAVE STATUS' is deprecated and will be removed in a future release. Please use SHOW REPLICA STATUS instead
 RESET SLAVE;

mysql-test/r/mysqld--help-notwin.result

@@ -2519,6 +2519,14 @@ The following options may be given as the first argument:
  inversion optimization for moving window frames also for
  floating values.
  (Defaults to on; use --skip-windowing-use-high-precision to disable.)
+ --write-stats-count[=#] 
+ Maximum number of most recent data points to be collected
+ for information_schema.write_statistics &
+ information_schema.replica_statistics time series.
+ --write-stats-frequency[=#] 
+ This variable determines the frequency(seconds) at which
+ write stats and replica lag stats are collected on
+ primaries
  --zstd-net-compression-level[=#] 
  Compression level for compressed protocol when zstd
  library is selected.
@@ -2870,7 +2878,7 @@ performance-schema-max-socket-classes 10
 performance-schema-max-socket-instances -1
 performance-schema-max-sql-text-length 1024
 performance-schema-max-stage-classes 175
-performance-schema-max-statement-classes 230
+performance-schema-max-statement-classes 232
 performance-schema-max-statement-stack 10
 performance-schema-max-table-handles -1
 performance-schema-max-table-instances -1
@@ -3232,6 +3240,8 @@ validate-user-plugins TRUE
 verbose TRUE
 wait-timeout 28800
 windowing-use-high-precision TRUE
+write-stats-count 0
+write-stats-frequency 0
 zstd-net-compression-level 3
 
 To see what values a running MySQL server is using, type

mysql-test/r/slave_stats_daemon.result

@@ -0,0 +1,197 @@
+include/master-slave.inc
+[connection master]
+########################################################################################################
+### Case 1: Stats are not sent by default i.e. interval value set to 0
+########################################################################################################
+select @@write_stats_frequency;
+@@write_stats_frequency
+0
+select @@write_stats_count;
+@@write_stats_count
+0
+select sleep(2);
+sleep(2)
+0
+select count(*) = 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+########################################################################################################
+### Case 2: Stats are sent when interval value is set > 0 on slaves
+########################################################################################################
+set @@GLOBAL.write_stats_frequency=1;
+select @@write_stats_frequency;
+@@write_stats_frequency
+1
+set @@GLOBAL.write_stats_count=10;
+select sleep(2);
+sleep(2)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+########################################################################################################
+### Case 2.1: Dynamically updating write_stats_count updates the number of data points in replica_statistics
+########################################################################################################
+set @@GLOBAL.write_stats_frequency=1;
+select @@write_stats_frequency;
+@@write_stats_frequency
+1
+set @@GLOBAL.write_stats_count=1;
+select sleep(2);
+sleep(2)
+0
+select count(*) = 1 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+set @@GLOBAL.write_stats_count=0;
+select sleep(2);
+sleep(2)
+0
+select count(*) = 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+set @@GLOBAL.write_stats_count=2;
+select sleep(3);
+sleep(3)
+0
+select count(*) = 2 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+########################################################################################################
+### Case 2.5: Connection is restored and stats are sent to master after it stops and restarts
+########################################################################################################
+set @@GLOBAL.write_stats_frequency=1;
+select @@write_stats_frequency;
+@@write_stats_frequency
+1
+set @@GLOBAL.write_stats_count=10;
+select sleep(1);
+sleep(1)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+select sleep(1);
+sleep(1)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+########################################################################################################
+### Case 3: In case of master promotions, slave thread is able to reconnet to new master.
+### In this test, I point the slave to an unavailable new master and then re-point it back to original
+### master to verify if stats communication resumes.
+########################################################################################################
+set @@GLOBAL.write_stats_frequency=1;
+Master_User = 'root'
+Master_Host = '127.0.0.1'
+include/stop_slave.inc
+change master to master_user='test';
+start slave;
+include/wait_for_slave_io_error.inc [errno=1045]
+Master_User = 'test'
+Master_Host = '127.0.0.1'
+set @@GLOBAL.write_stats_count=10;
+select sleep(2);
+sleep(2)
+0
+select now() - max(timestamp) > 1 as more_than_1_sec_old_most_recent_stats from performance_schema.replica_statistics;
+more_than_1_sec_old_most_recent_stats
+1
+include/stop_slave_sql.inc
+change master to master_user='root';
+include/start_slave.inc
+Master_User = 'root'
+Master_Host = '127.0.0.1'
+select sleep(2);
+sleep(2)
+0
+select now() - max(timestamp) <= 1 as less_than_1_sec_old_most_recent_stats from performance_schema.replica_statistics;
+less_than_1_sec_old_most_recent_stats
+1
+########################################################################################################
+### Case 3.5: Slave should be able to handle reconnections to master. It should reuse the existing 
+### thread for sending slave stats and should not spawn a new one.
+########################################################################################################
+select @@write_stats_frequency;
+@@write_stats_frequency
+1
+set @@GLOBAL.write_stats_frequency=1;
+set @@GLOBAL.write_stats_count=10;
+select sleep(2);
+sleep(2)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+select id from information_schema.processlist where command='Binlog Dump' into @id;
+kill @id;
+select sleep(2);
+sleep(2)
+0
+select now() - max(timestamp) <= 1 as less_than_1_sec_old_most_recent_stats from performance_schema.replica_statistics;
+less_than_1_sec_old_most_recent_stats
+1
+########################################################################################################
+### Case 3.5: Secondary is not stuck waiting for primary to send OK packet in case something goes wrong
+### It should timeout and move on. 
+########################################################################################################
+select @@write_stats_frequency;
+@@write_stats_frequency
+1
+set @@GLOBAL.write_stats_frequency=1;
+set @@GLOBAL.write_stats_count=10;
+select sleep(2);
+sleep(2)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+include/stop_slave.inc
+include/start_slave.inc
+select sleep(2);
+sleep(2)
+0
+select now() - max(timestamp) <= 1 as less_than_1_sec_old_most_recent_stats from performance_schema.replica_statistics;
+less_than_1_sec_old_most_recent_stats
+1
+########################################################################################################
+### Case 4: Promote slave to master and master to slave. Old master(new slave) should be able to send
+### stats to new master(old slave)
+########################################################################################################
+set @@GLOBAL.write_stats_count=10;
+set @@GLOBAL.write_stats_frequency=1;
+set @@GLOBAL.write_stats_count=10;
+set @@GLOBAL.write_stats_frequency=1;
+select sleep(2);
+sleep(2)
+0
+select count(*) from performance_schema.replica_statistics where (now() - timestamp) <= 1;
+count(*)
+0
+include/stop_slave.inc
+reset slave all;
+CHANGE MASTER TO MASTER_HOST= 'MASTER_HOST', MASTER_USER= 'root', MASTER_PORT= MASTER_PORT;;
+include/start_slave.inc
+select sleep(2);
+sleep(2)
+0
+select sleep(2);
+sleep(2)
+0
+select count(*) > 0 as stats_samples_collected from performance_schema.replica_statistics;
+stats_samples_collected
+1
+########################################################################################################
+### Cleanup
+### Reset the topology, swap master and slaves again
+########################################################################################################
+set @@GLOBAL.write_stats_count=0;
+set @@GLOBAL.write_stats_frequency=0;
+include/stop_slave.inc
+reset slave all;
+set @@GLOBAL.write_stats_count=0;
+set @@GLOBAL.write_stats_frequency=0;
+CHANGE MASTER TO MASTER_HOST= 'MASTER_HOST', MASTER_USER= 'root', MASTER_PORT= MASTER_PORT;;
+include/start_slave.inc
+include/rpl_end.inc

mysql-test/suite/perfschema/r/all_tests.result

@@ -33,6 +33,7 @@ dml_table_statistics_by_table.test
 idx_binary_log_transaction_compression_stats.test
 idx_esms_by_all.test
 idx_log_status.test
+idx_replica_statistics.test
 idx_replication_applier_filters.test
 idx_replication_applier_global_filters.test
 idx_replication_asynchronous_connection_failover.test

mysql-test/suite/perfschema/r/dd_version_check.result

@@ -1,6 +1,6 @@
 "Checking the data dictionary properties ..."
 SUBSTRING_INDEX(SUBSTRING(properties, LOCATE('PS_VERSION', properties), 30), ';', 1)
-PS_VERSION=80028007
+PS_VERSION=80028008
 "Checking the performance schema database structure ..."
 CHECK STATUS
-The tables in the performance_schema were last changed in MySQL 8.0.28-007
+The tables in the performance_schema were last changed in MySQL 8.0.28-008

mysql-test/suite/perfschema/r/ddl_replica_statistics.result

@@ -0,0 +1,11 @@
+ALTER TABLE performance_schema.replica_statistics
+ADD COLUMN foo integer;
+ERROR 42000: Access denied for user 'root'@'localhost' to database 'performance_schema'
+TRUNCATE TABLE performance_schema.replica_statistics;
+ERROR 42000: DROP command denied to user 'root'@'localhost' for table 'replica_statistics'
+ALTER TABLE performance_schema.replica_statistics
+ADD INDEX test_index(SERVER_ID);
+ERROR 42000: Access denied for user 'root'@'localhost' to database 'performance_schema'
+CREATE UNIQUE INDEX test_index ON
+performance_schema.replica_statistics(SERVER_ID);
+ERROR 42000: Access denied for user 'root'@'localhost' to database 'performance_schema'