squashed commit

Author: NicoHinderling

CHANGELOG.md

@@ -2,8 +2,13 @@
 
 ## Unreleased
 
+### Performance
+
+- (snapshots) Parallelize image hashing with rayon ([#3250](https://github.com/getsentry/sentry-cli/pull/3250))
+
 ### Fixes
 
+- (snapshots) Chunk image uploads to avoid file descriptor exhaustion and 413 errors when uploading hundreds of images ([#3249](https://github.com/getsentry/sentry-cli/pull/3249))
 - Replace `eprintln!` with `log::info!` for progress bar completion messages when the progress bar is disabled (e.g. in CI). This avoids spurious stderr output that some CI systems treat as errors ([#3223](https://github.com/getsentry/sentry-cli/pull/3223)).
 
 ## 3.3.5

Cargo.lock

@@ -44,7 +44,7 @@ java-properties = "2.0.0"
 lazy_static = "1.4.0"
 libc = "0.2.139"
 log = { version = "0.4.17", features = ["std"] }
-objectstore-client = { version = "0.1.2" , default-features = false, features = ["native-tls"] }
+objectstore-client = { version = "0.1.4" , default-features = false, features = ["native-tls"] }
 open = "3.2.0"
 parking_lot = "0.12.1"
 percent-encoding = "2.2.0"

Cargo.toml

@@ -9,7 +9,7 @@ use anyhow::{Context as _, Result};
 use clap::{Arg, ArgMatches, Command};
 use console::style;
 use itertools::Itertools as _;
-use log::{debug, info, warn};
+use log::{debug, warn};
 use objectstore_client::{ClientBuilder, ExpirationPolicy, Usecase};
 use rayon::prelude::*;
 use secrecy::ExposeSecret as _;
@@ -303,6 +303,11 @@ fn read_sidecar_metadata(image_path: &Path) -> Result<HashMap<String, Value>> {
     })
 }
 
+struct PreparedImage {
+    path: PathBuf,
+    key: String,
+}
+
 fn upload_images(
     images: Vec<ImageInfo>,
     org: &str,
@@ -318,30 +323,28 @@ fn upload_images(
     let client = ClientBuilder::new(options.objectstore.url)
         .token({
             // TODO: replace with auth from `ObjectstoreUploadOptions` when appropriate
-            let auth = match authenticated_api.auth() {
+            match authenticated_api.auth() {
                 Auth::Token(token) => token.raw().expose_secret().to_owned(),
-            };
-            auth
+            }
         })
         .configure_reqwest(|r| r.connect_timeout(Duration::from_secs(10)))
         .build()?;
 
+    let scopes = options.objectstore.scopes;
+
+    let find_scope = |name: &str| {
+        scopes
+            .iter()
+            .find(|(k, _)| k == name)
+            .map(|(_, v)| v.clone())
+    };
+    let org_id = find_scope("org").context("Missing org in UploadOptions scope")?;
+    let project_id = find_scope("project").context("Missing project in UploadOptions scope")?;
+
     let mut scope = Usecase::new("preprod").scope();
-    let (mut org_id, mut project_id): (Option<String>, Option<String>) = (None, None);
-    for (key, value) in options.objectstore.scopes.into_iter() {
-        scope = scope.push(&key, value.clone());
-        if key == "org" {
-            org_id = Some(value);
-        } else if key == "project" {
-            project_id = Some(value);
-        }
+    for (key, value) in scopes {
+        scope = scope.push(&key, value);
     }
-    let Some(org_id) = org_id else {
-        anyhow::bail!("Missing org in UploadOptions scope");
-    };
-    let Some(project_id) = project_id else {
-        anyhow::bail!("Missing project in UploadOptions scope");
-    };
 
     let session = scope.session(&client)?;
 
@@ -350,10 +353,10 @@ fn upload_images(
         .build()
         .context("Failed to create tokio runtime")?;
 
-    let mut many_builder = session.many();
     let mut manifest_entries = HashMap::new();
     let mut collisions: HashMap<String, Vec<String>> = HashMap::new();
     let mut kept_paths = HashMap::new();
+    let mut uploads = Vec::with_capacity(images.len());
 
     let hashed_images: Vec<_> = images
         .into_par_iter()
@@ -381,21 +384,7 @@ fn upload_images(
             continue;
         }
 
-        let file = runtime
-            .block_on(tokio::fs::File::open(&image.path))
-            .with_context(|| {
-                format!("Failed to open image for upload: {}", image.path.display())
-            })?;
-
         let key = format!("{org_id}/{project_id}/{hash}");
-        info!("Queueing {} as {key}", image.relative_path.display());
-
-        many_builder = many_builder.push(
-            session
-                .put_file(file)
-                .key(&key)
-                .expiration_policy(expiration),
-        );
 
         let mut extra = read_sidecar_metadata(&image.path).unwrap_or_else(|err| {
             warn!("Error reading sidecar metadata, ignoring it instead: {err:#}");
@@ -404,47 +393,60 @@ fn upload_images(
         extra.insert("content_hash".to_owned(), serde_json::Value::String(hash));
 
         kept_paths.insert(image_file_name.clone(), relative_path);
+        uploads.push(PreparedImage {
+            path: image.path,
+            key,
+        });
         manifest_entries.insert(
             image_file_name,
             ImageMetadata::new(image.width, image.height, extra),
         );
     }
 
     if !collisions.is_empty() {
-        let mut details = String::new();
-        for (name, excluded_paths) in &collisions {
-            let mut all_paths = vec![kept_paths[name].as_str()];
-            all_paths.extend(excluded_paths.iter().map(|s| s.as_str()));
-            details.push_str(&format!("\n  {name}: {}", all_paths.join(", ")));
-        }
+        let details: String = collisions
+            .iter()
+            .map(|(name, excluded)| {
+                let kept = &kept_paths[name];
+                let all = std::iter::once(kept.as_str())
+                    .chain(excluded.iter().map(|s| s.as_str()))
+                    .join(", ");
+                format!("\n  {name}: {all}")
+            })
+            .collect();
         warn!("Some images share identical file names. Only the first occurrence of each is included:{details}");
     }
 
-    let result = runtime.block_on(async { many_builder.send().error_for_failures().await });
+    let total_count = uploads.len();
 
-    let uploaded_count = manifest_entries.len();
+    let mut many_builder = session.many();
+    for prepared in uploads {
+        many_builder = many_builder.push(
+            session
+                .put_path(prepared.path.clone())
+                .key(&prepared.key)
+                .expiration_policy(expiration),
+        );
+    }
 
-    match result {
-        Ok(()) => {
-            println!(
-                "{} Uploaded {} image {}",
-                style(">").dim(),
-                style(uploaded_count).yellow(),
-                if uploaded_count == 1 { "file" } else { "files" }
-            );
-            Ok(manifest_entries)
-        }
-        Err(errors) => {
-            eprintln!("There were errors uploading images:");
-            let mut error_count = 0;
-            for error in errors {
-                let error = anyhow::Error::new(error);
-                eprintln!("  {}", style(format!("{error:#}")).red());
-                error_count += 1;
-            }
-            anyhow::bail!("Failed to upload {error_count} out of {uploaded_count} images")
+    let result = runtime.block_on(async { many_builder.send().await.error_for_failures().await });
+    if let Err(errors) = result {
+        let errors: Vec<_> = errors.collect();
+        eprintln!("There were errors uploading images:");
+        for error in &errors {
+            eprintln!("  {}", style(format!("{error:#}")).red());
         }
+        let error_count = errors.len();
+        anyhow::bail!("Failed to upload {error_count} images");
     }
+
+    println!(
+        "{} Uploaded {} image {}",
+        style(">").dim(),
+        style(total_count).yellow(),
+        if total_count == 1 { "file" } else { "files" }
+    );
+    Ok(manifest_entries)
 }
 
 #[cfg(test)]