replicators: Report server UUID when registering as a replica

MySQL 5.7 and some earlier versions of MySQL 8.0 have a bug where
it will report a malformed packet when executing SHOW SLAVE HOSTS or
SHOW REPLICAS. The issue is fixed in MySQL 8.0.28. In summary, the
affected versions will attempt to report the server UUID, if this is
not set when registering as a replica via SET @slave_uuid=<uuid> the
command will fail.

The fix is to add a server UUID parameter and use it or generate a
random UUIDv4 when registering as a replica.

Closes: REA-6238
Fixes: #1575

Release-Note-Core: Added the option `--replication-server-uuid` to
  set a server UUID in MySQL. This UUID will be reported in
  `SHOW REPLICAS`.
Change-Id: I75df8080ab25691600b207b999ddb1f4062cd07c
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/11105
Reviewed-by: Michael Zink <michael.z@readyset.io>
Tested-by: Buildkite CI

Author: altmannmarcelo

Cargo.lock

@@ -25,6 +25,7 @@ deadpool-postgres = { workspace = true }
 num_cpus = { workspace = true }
 icu = { workspace = true }
 yore = { workspace = true }
+uuid = { workspace = true, features = ["v4", "serde"] }
 
 [lints]
 workspace = true

database-utils/Cargo.toml

@@ -13,6 +13,7 @@ use error::DatabaseTypeParseError;
 use mysql_async::OptsBuilder;
 use pem::Pem;
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
 
 use readyset_errors::{ReadySetError, ReadySetResult};
 use readyset_sql::{Dialect, ast::SqlIdentifier};
@@ -97,6 +98,12 @@ pub struct UpstreamConfig {
     #[serde(default)]
     pub replication_server_id: Option<ReplicationServerId>,
 
+    /// UUID to report when registering as a replica with the upstream database (MySQL only).
+    /// If not set, a random UUID will be reported.
+    #[arg(long, env = "REPLICATION_SERVER_UUID", value_parser = parse_repl_uuid)]
+    #[serde(default)]
+    pub replication_server_uuid: Option<Uuid>,
+
     /// Hostname to report when registering as a replica with the upstream database (MySQL only).
     /// If not set, no hostname will be reported.
     #[arg(long = "report-host", env = "REPORT_HOST")]
@@ -278,6 +285,10 @@ impl UpstreamConfig {
 #[derive(Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
 pub struct ReplicationServerId(pub String);
 
+fn parse_repl_uuid(s: &str) -> Result<Uuid, String> {
+    Uuid::parse_str(s).map_err(|e| e.to_string())
+}
+
 impl Display for ReplicationServerId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(&self.0[..])
@@ -341,6 +352,7 @@ impl Default for UpstreamConfig {
             disable_setup_ddl_replication: false,
             disable_create_publication: false,
             replication_server_id: Default::default(),
+            replication_server_uuid: Default::default(),
             replica_report_host: Default::default(),
             replica_report_port: Default::default(),
             replica_report_user: Default::default(),

database-utils/src/lib.rs

@@ -56,6 +56,7 @@ readyset-tracing = { path = "../readyset-tracing" }
 readyset-util = { path = "../readyset-util" }
 readyset-decimal = { path = "../readyset-decimal" }
 replication-offset = { path = "../replication-offset" }
+uuid = { workspace = true, features = ["v4", "serde"] }
 
 [dev-dependencies]
 assert_matches = { workspace = true }

replicators/Cargo.toml

@@ -42,6 +42,7 @@ use readyset_sql::ast::{
 };
 use replication_offset::mysql::MySqlPosition;
 use replication_offset::ReplicationOffset;
+use uuid::Uuid;
 
 use crate::mysql_connector::utils::{mysql_pad_binary_column, mysql_pad_char_column};
 use crate::noria_adapter::{Connector, ReplicationAction};
@@ -132,6 +133,9 @@ pub(crate) struct MySqlBinlogConnector {
     /// The binlog "slave" must be assigned a unique `server_id` in the replica topology
     /// if one is not assigned we will use (u32::MAX - 55)
     server_id: Option<u32>,
+    /// The binlog "slave" must be assigned a unique `server_uuid` in the replica topology
+    /// if one is not assigned we will use a random UUID
+    server_uuid: Option<Uuid>,
     /// If we just want to continue reading the binlog from a previous point
     next_position: MySqlPosition,
     /// The GTID of the current transaction. Table modification events will have
@@ -159,6 +163,12 @@ impl MySqlBinlogConnector {
         self.server_id.unwrap_or(DEFAULT_SERVER_ID)
     }
 
+    /// The binlog replica must be assigned a unique `server_uuid` in the replica topology
+    /// if one is not assigned we will use a random UUID
+    fn server_uuid(&self) -> Uuid {
+        self.server_uuid.unwrap_or_else(Uuid::new_v4)
+    }
+
     async fn set_parameters(&mut self) -> mysql::Result<()> {
         // Enabling source to replica heartbeat packets requires manipulation of a knob that is
         // totally undocumented.  Yes, this really is a user variable, not a session variable.
@@ -186,10 +196,16 @@ impl MySqlBinlogConnector {
             Ok(version) => {
                 if version >= 80400 {
                     // MySQL 8.4.0 and above
-                    "SET @source_binlog_checksum='CRC32'"
+                    format!(
+                        "SET @source_binlog_checksum='CRC32', @replica_uuid='{}'",
+                        self.server_uuid()
+                    )
                 } else {
                     // MySQL 8.3.0 and below
-                    "SET @master_binlog_checksum='CRC32'"
+                    format!(
+                        "SET @master_binlog_checksum='CRC32', @slave_uuid='{}'",
+                        self.server_uuid()
+                    )
                 }
             }
             Err(err) => {
@@ -289,6 +305,7 @@ impl MySqlBinlogConnector {
             connection: mysql::Conn::new(mysql_opts).await?,
             reader: binlog::EventStreamReader::new(binlog::consts::BinlogVersion::Version4),
             server_id,
+            server_uuid: config.replication_server_uuid,
             next_position,
             current_gtid: None,
             enable_statement_logging,