Re: [pbs-devel] [PATCH proxmox v2 4/4] s3-client: add shared rate limiter via https connector

[Hannes Laimer <h.laimer@proxmox.com>]

Reviewed-by: Hannes Laimer <h.laimer@proxmox.com>
Tested-by: Hannes Laimer <h.laimer@proxmox.com>

On 9/16/25 14:42, Christian Ebner wrote:
> Allows to configure a shared rate limiter for the s3 client to limit
> upload and download bandwidth. This will help users which suffer from
> issues due to network congestion by the unlimited s3 client data
> transfers.
> 
> Signed-off-by: Christian Ebner <c.ebner@proxmox.com>
> ---
> Changes since version 1:
> - depend on proxmox-rate-limiter
> 
>   proxmox-s3-client/Cargo.toml            |  7 ++-
>   proxmox-s3-client/debian/control        |  7 ++-
>   proxmox-s3-client/examples/s3_client.rs |  1 +
>   proxmox-s3-client/src/api_types.rs      | 29 ++++++++++++
>   proxmox-s3-client/src/client.rs         | 63 +++++++++++++++++++++++--
>   proxmox-s3-client/src/lib.rs            |  4 +-
>   6 files changed, 105 insertions(+), 6 deletions(-)
> 
> diff --git a/proxmox-s3-client/Cargo.toml b/proxmox-s3-client/Cargo.toml
> index 639ae26b..be24c4de 100644
> --- a/proxmox-s3-client/Cargo.toml
> +++ b/proxmox-s3-client/Cargo.toml
> @@ -22,6 +22,7 @@ hyper-util = { workspace = true, features = [ "client-legacy", "tokio", "http1"
>   hyper = { workspace = true, optional = true }
>   iso8601 = { workspace = true, optional = true }
>   md5 = { workspace = true, optional = true }
> +nix = { workspace = true, optional = true }
>   openssl = { workspace = true, optional = true }
>   quick-xml = { workspace = true, features = [ "async-tokio" ], optional = true }
>   regex.workspace = true
> @@ -34,7 +35,9 @@ tracing = { workspace = true, optional = true }
>   url = {workspace = true, optional = true }
>   
>   proxmox-base64 = { workspace = true, optional = true }
> -proxmox-http = { workspace = true, features = [ "body", "client", "client-trait", "rate-limiter" ], optional = true }
> +proxmox-http = { workspace = true, features = [ "body", "client", "client-trait" ], optional = true }
> +proxmox-human-byte.workspace = true
> +proxmox-rate-limiter = { workspace = true, features = [ "rate-limiter", "shared-rate-limiter" ], optional = true }
>   proxmox-schema = { workspace = true, features = [ "api-macro", "api-types" ] }
>   proxmox-serde.workspace = true
>   proxmox-time = {workspace = true, optional = true }
> @@ -51,6 +54,7 @@ impl = [
>       "dep:hyper",
>       "dep:iso8601",
>       "dep:md5",
> +    "dep:nix",
>       "dep:openssl",
>       "dep:quick-xml",
>       "dep:serde-xml-rs",
> @@ -60,6 +64,7 @@ impl = [
>       "dep:url",
>       "dep:proxmox-base64",
>       "dep:proxmox-http",
> +    "dep:proxmox-rate-limiter",
>       "dep:proxmox-time",
>   ]
>   
> diff --git a/proxmox-s3-client/debian/control b/proxmox-s3-client/debian/control
> index c9147749..24bcbeba 100644
> --- a/proxmox-s3-client/debian/control
> +++ b/proxmox-s3-client/debian/control
> @@ -8,6 +8,7 @@ Build-Depends-Arch: cargo:native <!nocheck>,
>    libstd-rust-dev <!nocheck>,
>    librust-anyhow-1+default-dev <!nocheck>,
>    librust-const-format-0.2+default-dev <!nocheck>,
> + librust-proxmox-human-byte-1+default-dev <!nocheck>,
>    librust-proxmox-schema-5+api-macro-dev <!nocheck>,
>    librust-proxmox-schema-5+api-types-dev <!nocheck>,
>    librust-proxmox-schema-5+default-dev <!nocheck>,
> @@ -31,6 +32,7 @@ Depends:
>    ${misc:Depends},
>    librust-anyhow-1+default-dev,
>    librust-const-format-0.2+default-dev,
> + librust-proxmox-human-byte-1+default-dev,
>    librust-proxmox-schema-5+api-macro-dev,
>    librust-proxmox-schema-5+api-types-dev,
>    librust-proxmox-schema-5+default-dev,
> @@ -74,13 +76,16 @@ Depends:
>    librust-hyper-util-0.1+tokio-dev (>= 0.1.12-~~),
>    librust-iso8601-0.6+default-dev (>= 0.6.1-~~),
>    librust-md5-0.7+default-dev,
> + librust-nix-0.29+default-dev,
>    librust-openssl-0.10+default-dev,
>    librust-proxmox-base64-1+default-dev,
>    librust-proxmox-http-1+body-dev (>= 1.0.3-~~),
>    librust-proxmox-http-1+client-dev (>= 1.0.3-~~),
>    librust-proxmox-http-1+client-trait-dev (>= 1.0.3-~~),
>    librust-proxmox-http-1+default-dev (>= 1.0.3-~~),
> - librust-proxmox-http-1+rate-limiter-dev (>= 1.0.3-~~),
> + librust-proxmox-rate-limiter-1+default-dev,
> + librust-proxmox-rate-limiter-1+rate-limiter-dev,
> + librust-proxmox-rate-limiter-1+shared-rate-limiter-dev,
>    librust-proxmox-time-2+default-dev (>= 2.1.0-~~),
>    librust-quick-xml-0.36+async-tokio-dev (>= 0.36.1-~~),
>    librust-quick-xml-0.36+default-dev (>= 0.36.1-~~),
> diff --git a/proxmox-s3-client/examples/s3_client.rs b/proxmox-s3-client/examples/s3_client.rs
> index 67baf467..dd3885d7 100644
> --- a/proxmox-s3-client/examples/s3_client.rs
> +++ b/proxmox-s3-client/examples/s3_client.rs
> @@ -39,6 +39,7 @@ async fn run() -> Result<(), anyhow::Error> {
>           fingerprint: Some("<s3-api-fingerprint>".to_string()),
>           put_rate_limit: None,
>           provider_quirks: Vec::new(),
> +        rate_limiter_config: None,
>       };
>   
>       // Creating a client instance and connect to api endpoint
> diff --git a/proxmox-s3-client/src/api_types.rs b/proxmox-s3-client/src/api_types.rs
> index 115b1d2d..9071868a 100644
> --- a/proxmox-s3-client/src/api_types.rs
> +++ b/proxmox-s3-client/src/api_types.rs
> @@ -2,6 +2,7 @@ use anyhow::bail;
>   use const_format::concatcp;
>   use serde::{Deserialize, Serialize};
>   
> +use proxmox_human_byte::HumanByte;
>   use proxmox_schema::api_types::{
>       CERT_FINGERPRINT_SHA256_SCHEMA, DNS_LABEL_STR, IPRE_STR, SAFE_ID_FORMAT,
>   };
> @@ -126,6 +127,22 @@ serde_plain::derive_fromstr_from_deserialize!(ProviderQuirks);
>                   type: ProviderQuirks,
>               },
>           },
> +        "rate-in": {
> +            type: HumanByte,
> +            optional: true,
> +        },
> +        "burst-in": {
> +            type: HumanByte,
> +            optional: true,
> +        },
> +        "rate-out": {
> +            type: HumanByte,
> +            optional: true,
> +        },
> +        "burst-out": {
> +            type: HumanByte,
> +            optional: true,
> +        },
>       },
>   )]
>   #[derive(Serialize, Deserialize, Updater, Clone, PartialEq)]
> @@ -154,6 +171,18 @@ pub struct S3ClientConfig {
>       /// List of provider specific feature implementation quirks.
>       #[serde(skip_serializing_if = "Option::is_none")]
>       pub provider_quirks: Option<Vec<ProviderQuirks>>,
> +    /// Download rate limit.
> +    #[serde(skip_serializing_if = "Option::is_none")]
> +    pub rate_in: Option<HumanByte>,
> +    /// Download burst.
> +    #[serde(skip_serializing_if = "Option::is_none")]
> +    pub burst_in: Option<HumanByte>,
> +    /// Upload rate limit.
> +    #[serde(skip_serializing_if = "Option::is_none")]
> +    pub rate_out: Option<HumanByte>,
> +    /// Upload burst
> +    #[serde(skip_serializing_if = "Option::is_none")]
> +    pub burst_out: Option<HumanByte>,
>   }
>   
>   impl S3ClientConfig {
> diff --git a/proxmox-s3-client/src/client.rs b/proxmox-s3-client/src/client.rs
> index 96a5878d..e50f6d1d 100644
> --- a/proxmox-s3-client/src/client.rs
> +++ b/proxmox-s3-client/src/client.rs
> @@ -1,4 +1,4 @@
> -use std::path::Path;
> +use std::path::{Path, PathBuf};
>   use std::str::FromStr;
>   use std::sync::{Arc, Mutex};
>   use std::time::{Duration, Instant};
> @@ -12,6 +12,7 @@ use hyper::{Request, Response};
>   use hyper_util::client::legacy::connect::HttpConnector;
>   use hyper_util::client::legacy::Client;
>   use hyper_util::rt::TokioExecutor;
> +use nix::unistd::User;
>   use openssl::hash::MessageDigest;
>   use openssl::sha::Sha256;
>   use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
> @@ -19,7 +20,8 @@ use openssl::x509::X509StoreContextRef;
>   use tracing::error;
>   
>   use proxmox_http::client::HttpsConnector;
> -use proxmox_http::{Body, RateLimit, RateLimiter};
> +use proxmox_http::Body;
> +use proxmox_rate_limiter::{RateLimit, RateLimiter, SharedRateLimiter};
>   use proxmox_schema::api_types::CERT_FINGERPRINT_SHA256_SCHEMA;
>   
>   use crate::api_types::{ProviderQuirks, S3ClientConfig};
> @@ -53,6 +55,25 @@ pub enum S3PathPrefix {
>       None,
>   }
>   
> +/// Options for the https connector's rate limiter
> +pub struct S3RateLimiterOptions {
> +    /// ID for the shared rate limiter.
> +    pub id: String,
> +    /// Base path for the shared memory mapped file
> +    pub base_path: PathBuf,
> +    /// User for the to be created shared memory mapped file and folders
> +    pub user: User,
> +}
> +
> +/// Configuration  for the https connector's rate limiter
> +pub struct S3RateLimiterConfig {
> +    options: S3RateLimiterOptions,
> +    rate_in: Option<u64>,
> +    burst_in: Option<u64>,
> +    rate_out: Option<u64>,
> +    burst_out: Option<u64>,
> +}
> +
>   /// Configuration options for client
>   pub struct S3ClientOptions {
>       /// Endpoint to access S3 object store.
> @@ -77,6 +98,8 @@ pub struct S3ClientOptions {
>       pub put_rate_limit: Option<u64>,
>       /// Provider implementation specific features and limitations
>       pub provider_quirks: Vec<ProviderQuirks>,
> +    /// Configuration options for the shared rate limiter.
> +    pub rate_limiter_config: Option<S3RateLimiterConfig>,
>   }
>   
>   impl S3ClientOptions {
> @@ -86,7 +109,15 @@ impl S3ClientOptions {
>           secret_key: String,
>           bucket: Option<String>,
>           common_prefix: String,
> +        rate_limiter_options: Option<S3RateLimiterOptions>,
>       ) -> Self {
> +        let rate_limiter_config = rate_limiter_options.map(|options| S3RateLimiterConfig {
> +            options,
> +            rate_in: config.rate_in.map(|human_bytes| human_bytes.as_u64()),
> +            burst_in: config.burst_in.map(|human_bytes| human_bytes.as_u64()),
> +            rate_out: config.rate_out.map(|human_bytes| human_bytes.as_u64()),
> +            burst_out: config.burst_out.map(|human_bytes| human_bytes.as_u64()),
> +        });
>           Self {
>               endpoint: config.endpoint,
>               port: config.port,
> @@ -99,6 +130,7 @@ impl S3ClientOptions {
>               secret_key,
>               put_rate_limit: config.put_rate_limit,
>               provider_quirks: config.provider_quirks.unwrap_or_default(),
> +            rate_limiter_config,
>           }
>       }
>   }
> @@ -151,11 +183,36 @@ impl S3Client {
>           // want communication to object store backend api to always use https
>           http_connector.enforce_http(false);
>           http_connector.set_connect_timeout(Some(S3_HTTP_CONNECT_TIMEOUT));
> -        let https_connector = HttpsConnector::with_connector(
> +        let mut https_connector = HttpsConnector::with_connector(
>               http_connector,
>               ssl_connector_builder.build(),
>               S3_TCP_KEEPIDLE_TIME,
>           );
> +
> +        if let Some(limiter_config) = &options.rate_limiter_config {
> +            if let Some(limit) = limiter_config.rate_in {
> +                let limiter = SharedRateLimiter::mmap_shmem(
> +                    &format!("{}.in", limiter_config.options.id),
> +                    limit,
> +                    limiter_config.burst_in.unwrap_or(limit),
> +                    limiter_config.options.user.clone(),
> +                    limiter_config.options.base_path.clone(),
> +                )?;
> +                https_connector.set_read_limiter(Some(Arc::new(limiter)));
> +            }
> +
> +            if let Some(limit) = limiter_config.rate_out {
> +                let limiter = SharedRateLimiter::mmap_shmem(
> +                    &format!("{}.out", limiter_config.options.id),
> +                    limit,
> +                    limiter_config.burst_out.unwrap_or(limit),
> +                    limiter_config.options.user.clone(),
> +                    limiter_config.options.base_path.clone(),
> +                )?;
> +                https_connector.set_write_limiter(Some(Arc::new(limiter)));
> +            }
> +        }
> +
>           let client = Client::builder(TokioExecutor::new()).build::<_, Body>(https_connector);
>   
>           let authority_template = if let Some(port) = options.port {
> diff --git a/proxmox-s3-client/src/lib.rs b/proxmox-s3-client/src/lib.rs
> index 26e7032b..d02fd0dc 100644
> --- a/proxmox-s3-client/src/lib.rs
> +++ b/proxmox-s3-client/src/lib.rs
> @@ -20,7 +20,9 @@ pub use aws_sign_v4::uri_decode;
>   #[cfg(feature = "impl")]
>   mod client;
>   #[cfg(feature = "impl")]
> -pub use client::{S3Client, S3ClientOptions, S3PathPrefix, S3_HTTP_REQUEST_TIMEOUT};
> +pub use client::{
> +    S3Client, S3ClientOptions, S3PathPrefix, S3RateLimiterOptions, S3_HTTP_REQUEST_TIMEOUT,
> +};
>   #[cfg(feature = "impl")]
>   mod timestamps;
>   #[cfg(feature = "impl")]



_______________________________________________
pbs-devel mailing list
pbs-devel@lists.proxmox.com
https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel