Refactor ClientBuilder into a trait and remove use of CliArgs (#1513)

We use a generic parameter in the `run` and `mount` functions to create
an S3 client instance (and associated runtime), so they can be used with
the actual S3 client and the mock one. This PR changes 2 things:
* Replaces the `FnOnce` with a trait, to make it simpler to pass around
and extend in the future,
* Removes the `CliArgs` argument in favor of `ClientConfig` and other
required settings.

### Does this change impact existing behavior?

No.

### Does this change need a changelog entry? Does it require a version
change?

No.

---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

Signed-off-by: Alessandro Passaro <[email protected]>

Author: passaro

mountpoint-s3-fs/examples/mount_from_config.rs

@@ -21,7 +21,7 @@ use mountpoint_s3_fs::{
     manifest::{Manifest, ingest_manifest},
     metrics::{self, MetricsSinkHandle},
     prefix::Prefix,
-    s3::config::{ClientConfig, PartConfig, Region, S3Path},
+    s3::config::{ClientConfig, PartConfig, Region, S3Path, TargetThroughputSetting},
 };
 use nix::sys::signal::{self, Signal};
 use nix::unistd::Pid;
@@ -121,17 +121,17 @@ impl ConfigOptions {
             Some(prefix) => format!("{prefix}/mp-exmpl"),
             None => "mountpoint-s3-example/mp-exmpl".to_string(),
         };
-        let target_throughput = self.determine_throughput()?;
+        let throughput_target = self.determine_throughput()?;
         Ok(ClientConfig {
             region: Region::new_user_specified(self.region.clone()),
             endpoint_url: self.endpoint_url.clone(),
             addressing_style: AddressingStyle::Automatic,
             dual_stack: false,
             transfer_acceleration: false,
-            auth_config: mountpoint_s3_client::config::S3ClientAuthConfig::Default,
+            auth_config: Default::default(),
             requester_pays: false,
             expected_bucket_owner: self.expected_bucket_owner.clone(),
-            throughput_target_gbps: target_throughput,
+            throughput_target,
             bind: None,
             part_config: PartConfig::with_part_size(self.part_size.unwrap_or(8388608)),
             user_agent: UserAgent::new(Some(user_agent_string)),
@@ -157,23 +157,28 @@ impl ConfigOptions {
         DataCacheConfig::default()
     }
 
-    fn determine_throughput(&self) -> Result<f64> {
+    fn determine_throughput(&self) -> Result<TargetThroughputSetting> {
         match &self.throughput_config {
             // TODO(chagem): Remove some code duplication, by moving this logic into fs crate.
-            ThroughputConfig::Explicit { throughput } => Ok(*throughput),
+            ThroughputConfig::Explicit { throughput } => Ok(TargetThroughputSetting::User { gbps: *throughput }),
             ThroughputConfig::IMDSAutoConfigure => {
-                const DEFAULT_THROUGHPUT: f64 = 10.0;
                 let instance_info = InstanceInfo::new();
                 match autoconfigure::network_throughput(&instance_info) {
-                    Ok(throughput) => Ok(throughput),
+                    Ok(throughput) => Ok(TargetThroughputSetting::Instance { gbps: throughput }),
                     Err(e) => {
-                        tracing::warn!("Failed to detect network throughput. Using {DEFAULT_THROUGHPUT} gbps: {e:?}");
-                        Ok(DEFAULT_THROUGHPUT)
+                        tracing::warn!(
+                            "Failed to detect network throughput. Using {} gbps: {:?}",
+                            TargetThroughputSetting::DEFAULT_TARGET_THROUGHPUT_GBPS,
+                            e
+                        );
+                        Ok(TargetThroughputSetting::Default)
                     }
                 }
             }
             ThroughputConfig::IMDSLookUp { ec2_instance_type } => {
-                autoconfigure::get_maximum_network_throughput(ec2_instance_type).context("Unrecognized instance ID")
+                let target = autoconfigure::get_maximum_network_throughput(ec2_instance_type)
+                    .context("Unrecognized instance ID")?;
+                Ok(TargetThroughputSetting::Instance { gbps: target })
             }
         }
     }
@@ -228,8 +233,8 @@ fn mount_filesystem(
     let s3_path = config.build_s3_path()?;
 
     // Create the client and runtime
-    let client = config
-        .build_client_config()?
+    let client_config = config.build_client_config()?;
+    let client = client_config
         .create_client(None)
         .context("Failed to create S3 client")?;
     let runtime = Runtime::new(client.event_loop_group());

mountpoint-s3-fs/src/s3/config.rs

@@ -41,7 +41,7 @@ pub struct ClientConfig {
     pub expected_bucket_owner: Option<String>,
 
     /// Target throughput in Gbps
-    pub throughput_target_gbps: f64,
+    pub throughput_target: TargetThroughputSetting,
 
     /// One or more network interfaces to use when accessing S3
     pub bind: Option<Vec<String>>,
@@ -182,10 +182,10 @@ fn matches_bucket_regex(bucket_name: &str) -> bool {
 #[derive(Debug)]
 pub struct PartConfig {
     /// Part size for GET in bytes
-    read_size_bytes: usize,
+    pub read_size_bytes: usize,
 
     /// Part size for multi-part PUT in bytes
-    write_size_bytes: usize,
+    pub write_size_bytes: usize,
 }
 
 impl PartConfig {
@@ -238,6 +238,26 @@ impl Display for Region {
     }
 }
 
+/// Target throughput setting.
+#[derive(Debug, Clone, Copy)]
+pub enum TargetThroughputSetting {
+    Default,
+    User { gbps: f64 },
+    Instance { gbps: f64 },
+}
+
+impl TargetThroughputSetting {
+    pub const DEFAULT_TARGET_THROUGHPUT_GBPS: f64 = 10.0;
+
+    pub fn value(&self) -> f64 {
+        match self {
+            TargetThroughputSetting::Default => Self::DEFAULT_TARGET_THROUGHPUT_GBPS,
+            TargetThroughputSetting::User { gbps } => *gbps,
+            TargetThroughputSetting::Instance { gbps } => *gbps,
+        }
+    }
+}
+
 // This is a weird looking number! We really want our first request size to be 1MiB,
 // which is a common IO size. But Linux's readahead will try to read an extra 128k on on
 // top of a 1MiB read, which we'd have to wait for a second request to service. Because
@@ -256,7 +276,7 @@ impl ClientConfig {
     pub fn create_client(self, validate_on_s3_path: Option<&S3Path>) -> anyhow::Result<S3CrtClient> {
         let mut client_config = S3ClientConfig::new()
             .auth_config(self.auth_config)
-            .throughput_target_gbps(self.throughput_target_gbps)
+            .throughput_target_gbps(self.throughput_target.value())
             .read_part_size(self.part_config.read_size_bytes)
             .write_part_size(self.part_config.write_size_bytes)
             .read_backpressure(true)

mountpoint-s3/src/bin/mock-mount-s3.rs

@@ -20,55 +20,52 @@ use mountpoint_s3_client::mock_client::{MockClient, MockObject};
 use mountpoint_s3_client::types::ETag;
 use mountpoint_s3_fs::Runtime;
 use mountpoint_s3_fs::s3::S3Personality;
-use mountpoint_s3_fs::s3::config::BucketNameOrS3Uri;
+use mountpoint_s3_fs::s3::config::{ClientConfig, S3Path, TargetThroughputSetting};
 
 fn main() -> anyhow::Result<()> {
     let cli_args = CliArgs::parse();
     mountpoint_s3::run(create_mock_client, cli_args)
 }
 
-fn create_mock_client(args: &CliArgs) -> anyhow::Result<(Arc<ThroughputMockClient>, Runtime, S3Personality)> {
-    let bucket_name: String = match &args.bucket_name {
-        BucketNameOrS3Uri::BucketName(bucket_name) => bucket_name.clone().into(),
-        BucketNameOrS3Uri::S3Uri(_) => panic!("mock-mount-s3 bucket names do not support S3 URIs"),
-    };
-
+pub fn create_mock_client(
+    client_config: ClientConfig,
+    s3_path: &S3Path,
+    personality: Option<S3Personality>,
+) -> anyhow::Result<(Arc<ThroughputMockClient>, Runtime, S3Personality)> {
     // An extra little safety thing to make sure we can distinguish the real mount-s3 binary and
     // this one. Buckets starting with "sthree-" are always invalid against real S3:
     // https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html
     anyhow::ensure!(
-        &bucket_name.starts_with("sthree-"),
+        &s3_path.bucket_name.starts_with("sthree-"),
         "mock-mount-s3 bucket names must start with `sthree-`"
     );
 
-    tracing::warn!("using mock client");
-
     // TODO: Actually update the mock client to support different part sizes
     let part_size = {
-        if args.read_part_size.is_some() || args.write_part_size.is_some() {
+        if client_config.part_config.read_size_bytes != client_config.part_config.write_size_bytes {
             tracing::warn!("mock client does not support separate part sizes for reading and writing, ignoring");
         }
-        args.part_size
+        client_config.part_config.read_size_bytes
     };
 
-    let s3_personality = if let Some(bucket_type) = &args.bucket_type {
-        bucket_type.to_personality()
-    } else {
-        S3Personality::Standard
-    };
+    let s3_personality = personality.unwrap_or(S3Personality::Standard);
 
     let config = MockClient::config()
-        .bucket(&bucket_name)
+        .bucket(&s3_path.bucket_name)
         .part_size(part_size as usize)
         .unordered_list_seed(None)
         .enable_backpressure(true)
         .initial_read_window_size(1024 * 1024 + 128 * 1024) // matching real MP
         .enable_rename(s3_personality.supports_rename_object());
 
-    let client = if let Some(max_throughput_gbps) = args.maximum_throughput_gbps {
+    let client = if let TargetThroughputSetting::User {
+        gbps: max_throughput_gbps,
+    } = client_config.throughput_target
+    {
         tracing::info!("mock client limited to {max_throughput_gbps} Gb/s download throughput");
-        ThroughputMockClient::new(config, max_throughput_gbps as f64)
+        ThroughputMockClient::new(config, max_throughput_gbps)
     } else {
+        tracing::info!("mock client with no throughput limit");
         ThroughputMockClient::new_unlimited_throughput(config)
     };
 

mountpoint-s3/src/cli.rs

@@ -14,7 +14,7 @@ use mountpoint_s3_fs::logging::{LoggingConfig, prepare_log_file_name};
 use mountpoint_s3_fs::mem_limiter::MINIMUM_MEM_LIMIT;
 use mountpoint_s3_fs::prefix::Prefix;
 use mountpoint_s3_fs::s3::S3Personality;
-use mountpoint_s3_fs::s3::config::{BucketNameOrS3Uri, ClientConfig, PartConfig, S3Path};
+use mountpoint_s3_fs::s3::config::{BucketNameOrS3Uri, ClientConfig, PartConfig, S3Path, TargetThroughputSetting};
 use mountpoint_s3_fs::{S3FilesystemConfig, autoconfigure, metrics};
 use sysinfo::{RefreshKind, System};
 
@@ -720,22 +720,22 @@ impl CliArgs {
         user_agent
     }
 
-    fn throughput_target_gbps(&self, instance_info: &InstanceInfo) -> f64 {
-        const DEFAULT_TARGET_THROUGHPUT: f64 = 10.0;
-
-        let throughput_target_gbps = self.maximum_throughput_gbps.map(|t| t as f64).unwrap_or_else(|| {
+    fn throughput_target_gbps(&self, instance_info: &InstanceInfo) -> TargetThroughputSetting {
+        let throughput_target_gbps = self.maximum_throughput_gbps.map(|t| TargetThroughputSetting::User { gbps: t as f64 }).unwrap_or_else(|| {
             match autoconfigure::network_throughput(instance_info) {
-                Ok(throughput) => throughput,
+                Ok(throughput) => TargetThroughputSetting::Instance { gbps: throughput },
                 Err(e) => {
                     tracing::warn!(
-                        "failed to detect network throughput. Using {DEFAULT_TARGET_THROUGHPUT} gbps as throughput. \
-                        Use --maximum-throughput-gbps CLI flag to configure a target throughput appropriate for the instance. Detection failed due to: {e:?}",
+                        "failed to detect network throughput. Using {} gbps as throughput. \
+                        Use --maximum-throughput-gbps CLI flag to configure a target throughput appropriate for the instance. Detection failed due to: {:?}",
+                        TargetThroughputSetting::DEFAULT_TARGET_THROUGHPUT_GBPS,
+                        e,
                         );
-                    DEFAULT_TARGET_THROUGHPUT
+                    TargetThroughputSetting::Default
                 }
             }
         });
-        tracing::info!("target network throughput {throughput_target_gbps} Gbps");
+        tracing::info!("target network throughput {} Gbps", throughput_target_gbps.value());
         throughput_target_gbps
     }
 
@@ -763,7 +763,7 @@ impl CliArgs {
     pub fn client_config(&self, version: &str) -> ClientConfig {
         let instance_info = InstanceInfo::new();
         let user_agent = self.user_agent(&instance_info, version);
-        let throughput_target_gbps = self.throughput_target_gbps(&instance_info);
+        let throughput_target = self.throughput_target_gbps(&instance_info);
         let region = autoconfigure::get_region(&instance_info, self.region.clone());
 
         ClientConfig {
@@ -775,7 +775,7 @@ impl CliArgs {
             auth_config: self.auth_config(),
             requester_pays: self.requester_pays,
             expected_bucket_owner: self.expected_bucket_owner.clone(),
-            throughput_target_gbps,
+            throughput_target,
             bind: self.bind.clone(),
             part_config: self.part_config(),
             user_agent,

mountpoint-s3/src/run.rs

@@ -11,6 +11,7 @@ use mountpoint_s3_fs::data_cache::{DataCacheConfig, ManagedCacheDir};
 use mountpoint_s3_fs::fuse::session::FuseSession;
 use mountpoint_s3_fs::logging::init_logging;
 use mountpoint_s3_fs::s3::S3Personality;
+use mountpoint_s3_fs::s3::config::{ClientConfig, S3Path};
 use mountpoint_s3_fs::{MountpointConfig, Runtime, metrics};
 use nix::sys::signal::Signal;
 use nix::unistd::ForkResult;
@@ -19,11 +20,7 @@ use crate::cli::CliArgs;
 use crate::{build_info, parse_cli_args};
 
 /// Run Mountpoint with the given [CliArgs].
-pub fn run<ClientBuilder, Client>(client_builder: ClientBuilder, args: CliArgs) -> anyhow::Result<()>
-where
-    ClientBuilder: FnOnce(&CliArgs) -> anyhow::Result<(Client, Runtime, S3Personality)>,
-    Client: ObjectClient + Clone + Send + Sync + 'static,
-{
+pub fn run(client_builder: impl ClientBuilder, args: CliArgs) -> anyhow::Result<()> {
     let successful_mount_msg = format!(
         "{} is mounted at {}",
         args.bucket_description()?,
@@ -168,23 +165,21 @@ where
     Ok(())
 }
 
-fn mount<ClientBuilder, Client>(args: CliArgs, client_builder: ClientBuilder) -> anyhow::Result<FuseSession>
-where
-    ClientBuilder: FnOnce(&CliArgs) -> anyhow::Result<(Client, Runtime, S3Personality)>,
-    Client: ObjectClient + Clone + Send + Sync + 'static,
-{
+fn mount(args: CliArgs, client_builder: impl ClientBuilder) -> anyhow::Result<FuseSession> {
     tracing::info!("mount-s3 {}", build_info::FULL_VERSION);
     tracing::debug!("{:?}", args);
 
     let fuse_session_config = args.fuse_session_config()?;
     let sse = args.server_side_encryption()?;
 
-    let (client, runtime, s3_personality) = client_builder(&args)?;
+    let client_config = args.client_config(build_info::FULL_VERSION);
+
+    let s3_path = args.s3_path()?;
+    let (client, runtime, s3_personality) = client_builder.build(client_config, &s3_path, args.personality())?;
 
     let bucket_description = args.bucket_description()?;
     tracing::debug!("using S3 personality {s3_personality:?} for {bucket_description}");
 
-    let s3_path = args.s3_path()?;
     let filesystem_config = args.filesystem_config(sse.clone(), s3_personality);
     let mut data_cache_config = args.data_cache_config(sse)?;
 
@@ -205,18 +200,48 @@ where
     Ok(fuse_session)
 }
 
-/// Create a real S3 client
-pub fn create_s3_client(args: &CliArgs) -> anyhow::Result<(S3CrtClient, Runtime, S3Personality)> {
-    let client_config = args.client_config(build_info::FULL_VERSION);
+/// Builder for [ObjectClient] implementations.
+pub trait ClientBuilder {
+    type Client: ObjectClient + Clone + Send + Sync + 'static;
+
+    /// Build a new client instance.
+    fn build(
+        self,
+        client_config: ClientConfig,
+        s3_path: &S3Path,
+        personality: Option<S3Personality>,
+    ) -> anyhow::Result<(Self::Client, Runtime, S3Personality)>;
+}
 
-    let s3_path = args.s3_path()?;
+impl<F, C> ClientBuilder for F
+where
+    F: FnOnce(ClientConfig, &S3Path, Option<S3Personality>) -> anyhow::Result<(C, Runtime, S3Personality)>,
+    C: ObjectClient + Clone + Send + Sync + 'static,
+{
+    type Client = C;
+
+    fn build(
+        self,
+        client_config: ClientConfig,
+        s3_path: &S3Path,
+        personality: Option<S3Personality>,
+    ) -> anyhow::Result<(Self::Client, Runtime, S3Personality)> {
+        self(client_config, s3_path, personality)
+    }
+}
+
+// Create a real S3 client
+pub fn create_s3_client(
+    client_config: ClientConfig,
+    s3_path: &S3Path,
+    personality: Option<S3Personality>,
+) -> anyhow::Result<(S3CrtClient, Runtime, S3Personality)> {
     let client = client_config
-        .create_client(Some(&s3_path))
+        .create_client(Some(s3_path))
         .context("Failed to create S3 client")?;
 
     let runtime = Runtime::new(client.event_loop_group());
-    let s3_personality = args
-        .personality()
+    let s3_personality = personality
         .unwrap_or_else(|| S3Personality::infer_from_bucket(&s3_path.bucket_name, &client.endpoint_config()));
 
     Ok((client, runtime, s3_personality))