replicators: Pass replica parameters to MySQL server

When Readyset is registering as a replica in the MySQL
server, it provides an option to provide replica paramters
such as hostname, port, username and password. These are
shown when SHOW REPLICAS is run in the master database.

This commit adds following arguments to Readyset:

```
    --report-host
    --report-port
    --report-user
    --report-password
```

These are passed to the MySQL server with replica
registration.

The arguments are intentionally kept same as the MySQL
arguments for replica registration. The variable names use
the "replica" prefix to indicate that these are replica
parameters.

Fixes: #REA-4690
Closes: #1357

Release-Note-Core: Add `--report-host`, `--report-port`,
  `--report-user` and `--report-password` arguments to
  Readyset. These are used to show the replica parameters
  when SHOW REPLICAS is run in the master database.
Change-Id: I7596fadb056fd39c2701f095cfed19a1a712efd4
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/10282
Reviewed-by: Marcelo Altmann <marcelo@readyset.io>
Tested-by: Buildkite CI

Author: sandaru1

database-utils/src/lib.rs

@@ -93,6 +93,30 @@ pub struct UpstreamConfig {
     #[serde(default)]
     pub replication_server_id: Option<ReplicationServerId>,
 
+    /// 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")]
+    #[serde(default)]
+    pub replica_report_host: Option<String>,
+
+    /// Port to report when registering as a replica with the upstream database (MySQL only).
+    /// If not set, no port will be reported.
+    #[arg(long = "report-port", env = "REPORT_PORT")]
+    #[serde(default)]
+    pub replica_report_port: Option<u16>,
+
+    /// Username to report when registering as a replica with the upstream database (MySQL only).
+    /// If not set, no username will be reported.
+    #[arg(long = "report-user", env = "REPORT_USER")]
+    #[serde(default)]
+    pub replica_report_user: Option<String>,
+
+    /// Password to report when registering as a replica with the upstream database (MySQL only).
+    /// If not set, no password will be reported.
+    #[arg(long = "report-password", env = "REPORT_PASSWORD")]
+    #[serde(default)]
+    pub replica_report_password: Option<RedactedString>,
+
     /// The time to wait before restarting the replicator in seconds.
     #[arg(long, hide = true, default_value = "1", value_parser = duration_from_seconds)]
     #[serde(default = "default_replicator_restart_timeout")]
@@ -320,6 +344,10 @@ impl Default for UpstreamConfig {
             disable_setup_ddl_replication: false,
             disable_create_publication: false,
             replication_server_id: Default::default(),
+            replica_report_host: Default::default(),
+            replica_report_port: Default::default(),
+            replica_report_user: Default::default(),
+            replica_report_password: Default::default(),
             replicator_restart_timeout: Duration::from_secs(1),
             replication_tables: Default::default(),
             replication_tables_ignore: Default::default(),

replicators/src/mysql_connector/connector.rs

@@ -29,6 +29,7 @@ use serde_json::Map;
 use tokio::time::timeout;
 use tracing::{error, info, warn};
 
+use database_utils::UpstreamConfig;
 use readyset_client::metrics::recorded;
 use readyset_client::recipe::changelist::Change;
 use readyset_client::recipe::ChangeList;
@@ -174,7 +175,13 @@ impl MySqlBinlogConnector {
     /// In order to request a binlog, we must first register as a replica, and let the primary
     /// know what type of checksum we support (NONE and CRC32 are the options), NONE seems to work
     /// but others use CRC32 🤷‍♂️
-    async fn register_as_replica(&mut self) -> mysql::Result<()> {
+    async fn register_as_replica(
+        &mut self,
+        hostname: Option<String>,
+        port: Option<u16>,
+        username: Option<String>,
+        password: Option<String>,
+    ) -> mysql::Result<()> {
         let query = match get_mysql_version(&mut self.connection).await {
             Ok(version) => {
                 if version >= 80400 {
@@ -191,7 +198,24 @@ impl MySqlBinlogConnector {
         };
         self.connection.query_drop(query).await?;
 
-        let cmd = mysql_common::packets::ComRegisterSlave::new(self.server_id());
+        let mut cmd = mysql_common::packets::ComRegisterSlave::new(self.server_id());
+
+        if let Some(ref host) = hostname {
+            cmd = cmd.with_hostname(host.as_bytes());
+        }
+
+        if let Some(port) = port {
+            cmd = cmd.with_port(port);
+        }
+
+        if let Some(ref user) = username {
+            cmd = cmd.with_user(user.as_bytes());
+        }
+
+        if let Some(ref pass) = password {
+            cmd = cmd.with_password(pass.as_bytes());
+        }
+
         self.connection.write_command(&cmd).await?;
         // Server will respond with OK.
         self.connection.read_packet().await?;
@@ -243,11 +267,23 @@ impl MySqlBinlogConnector {
         noria: ReadySetHandle,
         mysql_opts: O,
         next_position: MySqlPosition,
-        server_id: Option<u32>,
         enable_statement_logging: bool,
         table_filter: TableFilter,
         parsing_preset: ParsingPreset,
+        config: &UpstreamConfig,
     ) -> ReadySetResult<Self> {
+        let server_id = config
+            .replication_server_id
+            .as_ref()
+            .map(|id| id.0.parse::<u32>())
+            .transpose()
+            .map_err(|_| {
+                ReadySetError::ReplicationFailed(format!(
+                    "{} is an invalid server id--it must be a valid u32.",
+                    config.replication_server_id.clone().unwrap()
+                ))
+            })?;
+
         let mut connector = MySqlBinlogConnector {
             noria,
             connection: mysql::Conn::new(mysql_opts).await?,
@@ -263,7 +299,18 @@ impl MySqlBinlogConnector {
         };
 
         connector.set_parameters().await?;
-        connector.register_as_replica().await?;
+
+        connector
+            .register_as_replica(
+                config.replica_report_host.clone(),
+                config.replica_report_port,
+                config.replica_report_user.clone(),
+                config
+                    .replica_report_password
+                    .as_ref()
+                    .map(|p| p.to_string()),
+            )
+            .await?;
         let binlog_request = connector.request_binlog().await;
         match binlog_request {
             Ok(()) => (),

replicators/src/noria_adapter.rs

@@ -428,27 +428,15 @@ impl<'a> NoriaAdapter<'a> {
             (Some(pos), _) => pos.clone().try_into()?,
         };
 
-        let server_id = config
-            .replication_server_id
-            .as_ref()
-            .map(|id| id.0.parse::<u32>())
-            .transpose()
-            .map_err(|_| {
-                ReadySetError::ReplicationFailed(format!(
-                    "{} is an invalid server id--it must be a valid u32.",
-                    config.replication_server_id.clone().unwrap()
-                ))
-            })?;
-
         let connector = Box::new(
             MySqlBinlogConnector::connect(
                 noria.clone(),
                 mysql_opts_builder,
                 pos.clone(),
-                server_id,
                 enable_statement_logging,
                 table_filter.clone(),
                 parsing_preset,
+                config,
             )
             .await?,
         );