feat: add request lifecycle hooks for client-side load balancing and observability

  Adds RequestLifecycleHooks trait to enable intercepting and modifying S3 API
  requests at key points in the request lifecycle. This supports client-side
  load balancing, request telemetry, and debug logging.

Author: HJLebbink

examples/debug_logging_hook.rs

@@ -0,0 +1,235 @@
+// MinIO Rust Library for Amazon S3 Compatible Cloud Storage
+// Copyright 2024 MinIO, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Example demonstrating how to use RequestLifecycleHooks for debug logging.
+//!
+//! This example shows:
+//! - Creating a custom debug logging hook
+//! - Attaching the hook to the MinIO client
+//! - Automatic logging of all S3 API requests with headers and response status
+//! - Using both `before_signing_mut` and `after_execute` hooks
+//!
+//! Run with default values (test-bucket / test-object.txt, verbose mode enabled):
+//! ```
+//! cargo run --example debug_logging_hook
+//! ```
+//!
+//! Run with custom bucket and object:
+//! ```
+//! cargo run --example debug_logging_hook -- mybucket myobject
+//! ```
+//!
+//! Disable verbose output:
+//! ```
+//! cargo run --example debug_logging_hook -- --no-verbose
+//! ```
+
+use clap::{ArgAction, Parser};
+use futures_util::StreamExt;
+use minio::s3::builders::ObjectContent;
+use minio::s3::client::hooks::{Extensions, RequestLifecycleHooks};
+use minio::s3::client::{Method, Response};
+use minio::s3::creds::StaticProvider;
+use minio::s3::error::Error;
+use minio::s3::http::Url;
+use minio::s3::multimap_ext::Multimap;
+use minio::s3::response::BucketExistsResponse;
+use minio::s3::segmented_bytes::SegmentedBytes;
+use minio::s3::types::{S3Api, ToStream};
+use minio::s3::{MinioClient, MinioClientBuilder};
+use std::sync::Arc;
+
+/// Debug logging hook that prints detailed information about each S3 request.
+#[derive(Debug)]
+struct DebugLoggingHook {
+    /// Enable verbose output including all headers
+    verbose: bool,
+}
+
+impl DebugLoggingHook {
+    fn new(verbose: bool) -> Self {
+        Self { verbose }
+    }
+}
+
+#[async_trait::async_trait]
+impl RequestLifecycleHooks for DebugLoggingHook {
+    fn name(&self) -> &'static str {
+        "debug-logger"
+    }
+
+    async fn before_signing_mut(
+        &self,
+        method: &Method,
+        url: &mut Url,
+        _region: &str,
+        _headers: &mut Multimap,
+        _query_params: &Multimap,
+        bucket_name: Option<&str>,
+        object_name: Option<&str>,
+        _body: Option<&SegmentedBytes>,
+        _extensions: &mut Extensions,
+    ) -> Result<(), Error> {
+        if self.verbose {
+            let bucket_obj = match (bucket_name, object_name) {
+                (Some(b), Some(o)) => format!("{b}/{o}"),
+                (Some(b), None) => b.to_string(),
+                _ => url.to_string(),
+            };
+            println!("→ Preparing {method} request for {bucket_obj}");
+        }
+        Ok(())
+    }
+
+    async fn after_execute(
+        &self,
+        method: &Method,
+        url: &Url,
+        _region: &str,
+        headers: &Multimap,
+        _query_params: &Multimap,
+        bucket_name: Option<&str>,
+        object_name: Option<&str>,
+        resp: &Result<Response, reqwest::Error>,
+        _extensions: &mut Extensions,
+    ) {
+        // Format the basic request info
+        let bucket_obj = match (bucket_name, object_name) {
+            (Some(b), Some(o)) => format!("{b}/{o}"),
+            (Some(b), None) => b.to_string(),
+            _ => url.to_string(),
+        };
+
+        // Format response status
+        let status = match resp {
+            Ok(response) => format!("✓ {}", response.status()),
+            Err(err) => format!("✗ Error: {err}"),
+        };
+
+        println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━");
+        println!("S3 Request: {method} {bucket_obj}");
+        println!("URL: {url}");
+        println!("Status: {status}");
+
+        if self.verbose {
+            // Print headers alphabetically
+            let mut header_strings: Vec<String> = headers
+                .iter_all()
+                .map(|(k, v)| format!("{}: {}", k, v.join(",")))
+                .collect();
+            header_strings.sort();
+
+            println!("\nRequest Headers:");
+            for header in header_strings {
+                println!("  {header}");
+            }
+        }
+
+        println!("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n");
+    }
+}
+
+/// Example demonstrating debug logging with hooks
+#[derive(Parser)]
+struct Cli {
+    /// Bucket to use for the example
+    #[arg(default_value = "test-bucket")]
+    bucket: String,
+    /// Object to upload
+    #[arg(default_value = "test-object.txt")]
+    object: String,
+    /// Disable verbose output (verbose is enabled by default, use --no-verbose to disable)
+    #[arg(long = "no-verbose", action = ArgAction::SetFalse, default_value_t = true)]
+    verbose: bool,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    env_logger::init();
+    let args = Cli::parse();
+
+    println!("\n🔧 MinIO Debug Logging Hook Example\n");
+    println!("This example demonstrates how hooks can be used for debugging S3 requests.");
+    println!(
+        "We'll perform a few operations on bucket '{}' with debug logging enabled.\n",
+        args.bucket
+    );
+
+    // Create the debug logging hook
+    let debug_hook = Arc::new(DebugLoggingHook::new(args.verbose));
+
+    // Create MinIO client with the debug logging hook attached
+    let static_provider = StaticProvider::new(
+        "Q3AM3UQ867SPQQA43P2F",
+        "zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
+        None,
+    );
+
+    let client: MinioClient = MinioClientBuilder::new("https://play.min.io".parse()?)
+        .provider(Some(static_provider))
+        .hook(debug_hook) // Attach the debug logging hook
+        .build()?;
+
+    println!("✓ Created MinIO client with debug logging hook\n");
+
+    // Operation 1: Check if bucket exists
+    println!("📋 Checking if bucket exists...");
+    let resp: BucketExistsResponse = client.bucket_exists(&args.bucket).build().send().await?;
+
+    // Operation 2: Create bucket if it doesn't exist
+    if !resp.exists() {
+        println!("\n📋 Creating bucket...");
+        client.create_bucket(&args.bucket).build().send().await?;
+    } else {
+        println!("\n✓ Bucket already exists");
+    }
+
+    // Operation 3: Upload a small object
+    println!("\n📋 Uploading object...");
+    let content = b"Hello from MinIO Rust SDK with debug logging!";
+    let object_content: ObjectContent = content.to_vec().into();
+    client
+        .put_object_content(&args.bucket, &args.object, object_content)
+        .build()
+        .send()
+        .await?;
+
+    // Operation 4: List objects in the bucket
+    println!("\n📋 Listing objects in bucket...");
+    let mut list_stream = client
+        .list_objects(&args.bucket)
+        .recursive(false)
+        .build()
+        .to_stream()
+        .await;
+
+    let mut total_objects = 0;
+    while let Some(result) = list_stream.next().await {
+        match result {
+            Ok(resp) => {
+                total_objects += resp.contents.len();
+            }
+            Err(e) => {
+                eprintln!("Error listing objects: {e}");
+            }
+        }
+    }
+    println!("\n✓ Found {total_objects} objects in bucket");
+
+    println!("\n🎉 All operations completed successfully with debug logging enabled!\n");
+    println!("💡 Tip: Run with --no-verbose to disable detailed output\n");
+
+    Ok(())
+}

src/s3/client.rs

@@ -20,7 +20,7 @@ use dashmap::DashMap;
 use http::HeaderMap;
 pub use hyper::http::Method;
 use reqwest::Body;
-pub use reqwest::{Error as ReqwestError, Response};
+pub use reqwest::Response;
 use std::fmt::Debug;
 use std::fs::File;
 use std::io::prelude::*;
@@ -487,6 +487,7 @@ impl MinioClient {
         let date = utc_now();
         headers.add(X_AMZ_DATE, to_amz_date(date));
 
+        // Allow hooks to modify the request before signing (e.g., for client-side load balancing)
         let url_before_hook = url.to_string();
         self.run_before_signing_hooks(
             method,
@@ -496,16 +497,19 @@ impl MinioClient {
             query_params,
             bucket_name,
             object_name,
-            body.clone(),
+            &body,
             &mut extensions,
         )
         .await?;
 
-        let url_after_hook = url.to_string();
-        if url_before_hook != url_after_hook {
-            headers.add("x-minio-redirect", url_after_hook);
+        // If a hook modified the URL (e.g., redirecting to a different MinIO node for load balancing),
+        // add the x-minio-redirect header to inform the server about the client-side redirection.
+        // This enables server-side telemetry, debugging, and load balancing metrics.
+        // The header contains the actual endpoint where the request is being sent.
+        if url.to_string() != url_before_hook {
+            headers.add("x-minio-redirect", url.to_string());
         }
-        
+
         if let Some(p) = &self.shared.provider {
             let creds = p.fetch();
             if creds.session_token.is_some() {
@@ -532,31 +536,9 @@ impl MinioClient {
             }
         }
 
-        if false {
-            let mut header_strings: Vec<String> = headers
-                .iter_all()
-                .map(|(k, v)| format!("{}: {}", k, v.join(",")))
-                .collect();
-
-            // Sort headers alphabetically by name
-            header_strings.sort();
-
-            let debug_str = format!(
-                "S3 request: {method} url={:?}; headers={:?}; body={body:?}",
-                url.path,
-                header_strings.join("; ")
-            );
-            let truncated = if debug_str.len() > 1000 {
-                format!("{}...", &debug_str[..997])
-            } else {
-                debug_str
-            };
-            println!("{truncated}");
-        }
-
         if (*method == Method::PUT) || (*method == Method::POST) {
             //TODO: why-oh-why first collect into a vector and then iterate to a stream?
-            let bytes_vec: Vec<Bytes> = match body.clone() {
+            let bytes_vec: Vec<Bytes> = match body.as_ref() {
                 Some(v) => v.iter().collect(),
                 None => Vec::new(),
             };
@@ -698,7 +680,7 @@ impl MinioClient {
         query_params: &Multimap,
         bucket_name: Option<&str>,
         object_name: Option<&str>,
-        body: Option<Arc<SegmentedBytes>>,
+        body: &Option<Arc<SegmentedBytes>>,
         extensions: &mut http::Extensions,
     ) -> Result<(), Error> {
         for hook in self.shared.client_hooks.iter() {