When installing a release, install operators in parallel (#202)

* When installing a release, install operators in parallel

* Changelog

* Reference the PR from the changelog

Author: nightkr

Cargo.lock

@@ -34,6 +34,7 @@ tracing.workspace = true
 url.workspace = true
 utoipa = { workspace = true, optional = true }
 which.workspace = true
+futures.workspace = true
 
 [dev-dependencies]
 rstest.workspace = true

rust/stackable-cockpit/Cargo.toml

@@ -3,6 +3,7 @@ use std::fmt::Display;
 
 use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
+use tokio::task::block_in_place;
 use tracing::{debug, error, info, instrument};
 use url::Url;
 
@@ -202,62 +203,66 @@ pub fn install_release_from_repo(
     namespace: &str,
     suppress_output: bool,
 ) -> Result<InstallReleaseStatus, Error> {
-    debug!("Install Helm release from repo");
-
-    if check_release_exists(release_name, namespace)? {
-        let release = get_release(release_name, namespace)?.ok_or(Error::InstallRelease {
-            source: InstallReleaseError::NoSuchRelease {
-                name: release_name.to_owned(),
-            },
-        })?;
-
-        let current_version = release.version;
-
-        match chart_version {
-            Some(chart_version) => {
-                if chart_version == current_version {
-                    return Ok(InstallReleaseStatus::ReleaseAlreadyInstalledWithVersion {
-                        requested_version: chart_version.to_string(),
+    // Ideally, each Helm invocation would spawn_blocking instead in/around helm_sys,
+    // but that requires a larger refactoring
+    block_in_place(|| {
+        debug!("Install Helm release from repo");
+
+        if check_release_exists(release_name, namespace)? {
+            let release = get_release(release_name, namespace)?.ok_or(Error::InstallRelease {
+                source: InstallReleaseError::NoSuchRelease {
+                    name: release_name.to_owned(),
+                },
+            })?;
+
+            let current_version = release.version;
+
+            match chart_version {
+                Some(chart_version) => {
+                    if chart_version == current_version {
+                        return Ok(InstallReleaseStatus::ReleaseAlreadyInstalledWithVersion {
+                            requested_version: chart_version.to_string(),
+                            release_name: release_name.to_string(),
+                            current_version,
+                        });
+                    } else {
+                        return Err(Error::InstallRelease {
+                            source: InstallReleaseError::ReleaseAlreadyInstalled {
+                                requested_version: chart_version.into(),
+                                name: release_name.into(),
+                                current_version,
+                            },
+                        });
+                    }
+                }
+                None => {
+                    return Ok(InstallReleaseStatus::ReleaseAlreadyInstalledUnspecified {
                         release_name: release_name.to_string(),
                         current_version,
-                    });
-                } else {
-                    return Err(Error::InstallRelease {
-                        source: InstallReleaseError::ReleaseAlreadyInstalled {
-                            requested_version: chart_version.into(),
-                            name: release_name.into(),
-                            current_version,
-                        },
-                    });
+                    })
                 }
             }
-            None => {
-                return Ok(InstallReleaseStatus::ReleaseAlreadyInstalledUnspecified {
-                    release_name: release_name.to_string(),
-                    current_version,
-                })
-            }
         }
-    }
 
-    let full_chart_name = format!("{repo_name}/{chart_name}");
-    let chart_version = chart_version.unwrap_or(HELM_DEFAULT_CHART_VERSION);
+        let full_chart_name = format!("{repo_name}/{chart_name}");
+        let chart_version = chart_version.unwrap_or(HELM_DEFAULT_CHART_VERSION);
 
-    debug!(
-        "Installing Helm release {} ({}) from chart {}",
-        release_name, chart_version, full_chart_name
-    );
-
-    install_release(
-        release_name,
-        &full_chart_name,
-        chart_version,
-        values_yaml,
-        namespace,
-        suppress_output,
-    )?;
+        debug!(
+            "Installing Helm release {} ({}) from chart {}",
+            release_name, chart_version, full_chart_name
+        );
 
-    Ok(InstallReleaseStatus::Installed(release_name.to_string()))
+        install_release(
+            release_name,
+            &full_chart_name,
+            chart_version,
+            values_yaml,
+            namespace,
+            suppress_output,
+        )?;
+
+        Ok(InstallReleaseStatus::Installed(release_name.to_string()))
+    })
 }
 
 /// Installs a Helm release.

rust/stackable-cockpit/src/helm.rs

@@ -1,6 +1,8 @@
+use futures::{StreamExt as _, TryStreamExt};
 use indexmap::IndexMap;
 use serde::{Deserialize, Serialize};
 use snafu::{ResultExt, Snafu};
+use tokio::task::JoinError;
 use tracing::{info, instrument};
 
 #[cfg(feature = "openapi")]
@@ -26,6 +28,9 @@ pub enum Error {
 
     #[snafu(display("failed to uninstall release using Helm"))]
     HelmUninstall { source: helm::Error },
+
+    #[snafu(display("failed to launch background task"))]
+    BackgroundTask { source: JoinError },
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -46,26 +51,39 @@ pub struct ReleaseSpec {
 impl ReleaseSpec {
     /// Installs a release by installing individual operators.
     #[instrument(skip_all)]
-    pub fn install(
+    pub async fn install(
         &self,
         include_products: &[String],
         exclude_products: &[String],
         namespace: &str,
     ) -> Result<()> {
         info!("Installing release");
 
-        for (product_name, product) in self.filter_products(include_products, exclude_products) {
-            info!("Installing {product_name}-operator");
-
-            // Create operator spec
-            let operator = OperatorSpec::new(product_name, Some(product.version.clone()))
-                .context(OperatorSpecParseSnafu)?;
-
-            // Install operator
-            operator.install(namespace).context(HelmInstallSnafu)?
-        }
-
-        Ok(())
+        let namespace = namespace.to_string();
+        futures::stream::iter(self.filter_products(include_products, exclude_products))
+            .map(|(product_name, product)| {
+                let namespace = namespace.clone();
+                // Helm installs currently `block_in_place`, so we need to spawn each job onto a separate task to
+                // get useful parallelism.
+                tokio::spawn(async move {
+                    info!("Installing {product_name}-operator");
+
+                    // Create operator spec
+                    let operator = OperatorSpec::new(&product_name, Some(product.version.clone()))
+                        .context(OperatorSpecParseSnafu)?;
+
+                    // Install operator
+                    operator.install(&namespace).context(HelmInstallSnafu)?;
+
+                    info!("Installed {product_name}-operator");
+
+                    Ok(())
+                })
+            })
+            .buffer_unordered(10)
+            .map(|res| res.context(BackgroundTaskSnafu)?)
+            .try_collect::<()>()
+            .await
     }
 
     #[instrument(skip_all)]

rust/stackable-cockpit/src/platform/release/spec.rs

@@ -197,6 +197,7 @@ impl StackSpec {
         // Install the release
         release
             .install(&self.operators, &[], operator_namespace)
+            .await
             .context(InstallReleaseSnafu)
     }
 

rust/stackable-cockpit/src/platform/stack/spec.rs

@@ -4,11 +4,16 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Changed
+
+- Operators are now installed in parallel when installing a release ([#202]).
+
 ### Fixed
 
 - Fix `--cluster-name` not taking effect. The local test clusters always used the default cluster name ([#181]).
 
 [#181]: https://github.com/stackabletech/stackable-cockpit/pull/181
+[#202]: https://github.com/stackabletech/stackable-cockpit/pull/202
 
 ## [23.11.3] - 2024-01-03
 

rust/stackablectl/CHANGELOG.md

@@ -289,6 +289,7 @@ async fn install_cmd(
                     &args.excluded_products,
                     &args.operator_namespace,
                 )
+                .await
                 .context(ReleaseInstallSnafu)?;
 
             output