add release upgrade functionality

Author: xeniape

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

@@ -13,6 +13,7 @@ use crate::{
         operator::{self, ChartSourceType, OperatorSpec},
         product,
     },
+    utils::k8s::{self, Client},
 };
 
 type Result<T, E = Error> = std::result::Result<T, E>;
@@ -30,6 +31,15 @@ pub enum Error {
 
     #[snafu(display("failed to launch background task"))]
     BackgroundTask { source: JoinError },
+
+    #[snafu(display("failed to deploy manifests using the kube client"))]
+    DeployManifest { source: k8s::Error },
+
+    #[snafu(display("failed to access the CRDs from source"))]
+    AccessCRDs { source: reqwest::Error },
+
+    #[snafu(display("failed to read CRD manifests"))]
+    ReadManifests { source: reqwest::Error },
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -108,6 +118,72 @@ impl ReleaseSpec {
             .await
     }
 
+    /// Upgrades a release by upgrading individual operators.
+    #[instrument(skip_all, fields(
+        %namespace,
+    ))]
+    pub async fn upgrade_crds(
+        &self,
+        include_products: &[String],
+        exclude_products: &[String],
+        namespace: &str,
+        k8s_client: &Client,
+    ) -> Result<()> {
+        debug!("Upgrading CRDs for release");
+
+        include_products.iter().for_each(|product| {
+            Span::current().record("product.included", product);
+        });
+        exclude_products.iter().for_each(|product| {
+            Span::current().record("product.excluded", product);
+        });
+
+        let client = reqwest::Client::new();
+
+        let operators = self.filter_products(include_products, exclude_products);
+
+        Span::current().pb_set_style(
+            &ProgressStyle::with_template("Upgrading CRDs {wide_bar} {pos}/{len}").unwrap(),
+        );
+        Span::current().pb_set_length(operators.len() as u64);
+
+        for (product_name, product) in operators {
+            Span::current().record("product_name", &product_name);
+            debug!("Upgrading CRDs for {product_name}-operator");
+
+            let release = match product.version.pre.as_str() {
+                "dev" => "main".to_string(),
+                _ => {
+                    format!("{}", product.version)
+                }
+            };
+
+            let request_url = format!(
+                "https://raw.githubusercontent.com/stackabletech/{product_name}-operator/{release}/deploy/helm/{product_name}-operator/crds/crds.yaml"
+            );
+
+            // Get CRD manifests
+            // TODO bei nicht 200 Status, Fehler werfen
+            let response = client
+                .get(request_url)
+                .send()
+                .await
+                .context(AccessCRDsSnafu)?;
+            let crd_manifests = response.text().await.context(ReadManifestsSnafu)?;
+
+            // Upgrade CRDs
+            k8s_client
+                .replace_crds(&crd_manifests)
+                .await
+                .context(DeployManifestSnafu)?;
+
+            debug!("Upgraded {product_name}-operator CRDs");
+            Span::current().pb_inc(1);
+        }
+
+        Ok(())
+    }
+
     #[instrument(skip_all)]
     pub fn uninstall(&self, namespace: &str) -> Result<()> {
         info!("Uninstalling release");

rust/stackable-cockpit/src/utils/k8s/client.rs

@@ -65,6 +65,9 @@ pub enum Error {
     #[snafu(display("failed to retrieve cluster information"))]
     ClusterInformation { source: cluster::Error },
 
+    #[snafu(display("failed to retrieve previous resource version information"))]
+    ResourceVersion { source: kube::error::Error },
+
     #[snafu(display("invalid or empty secret data in '{secret_name}'"))]
     InvalidSecretData { secret_name: String },
 
@@ -148,6 +151,46 @@ impl Client {
         Ok(())
     }
 
+    pub async fn replace_crds(&self, crds: &str) -> Result<()> {
+        for crd in serde_yaml::Deserializer::from_str(crds) {
+            let mut object = DynamicObject::deserialize(crd).context(DeserializeYamlSnafu)?;
+
+            let object_type = object.types.as_ref().ok_or(
+                ObjectTypeSnafu {
+                    object: object.clone(),
+                }
+                .build(),
+            )?;
+
+            let gvk = Self::gvk_of_typemeta(object_type);
+            let (resource, _) = self
+                .resolve_gvk(&gvk)
+                .await?
+                .context(GVKUnkownSnafu { gvk })?;
+
+            // CRDs are cluster scoped
+            let api: Api<DynamicObject> = Api::all_with(self.client.clone(), &resource);
+
+            if let Some(resource) = api
+                .get_opt(&object.name_any())
+                .await
+                .context(KubeClientFetchSnafu)?
+            {
+                object.metadata.resource_version = resource.resource_version();
+                api.replace(&object.name_any(), &PostParams::default(), &object)
+                    .await
+                    .context(KubeClientPatchSnafu)?;
+            } else {
+                // Create CRD if a previous version wasn't found
+                api.create(&PostParams::default(), &object)
+                    .await
+                    .context(KubeClientCreateSnafu)?;
+            }
+        }
+
+        Ok(())
+    }
+
     /// Lists objects by looking up a GVK via the discovery. It returns an
     /// optional list of dynamic objects. The method returns [`Ok(None)`]
     /// if the client was unable to resolve the GVK. An error is returned

rust/stackablectl/src/cmds/release.rs

@@ -44,6 +44,9 @@ pub enum ReleaseCommands {
     /// Uninstall a release
     #[command(aliases(["rm", "un"]))]
     Uninstall(ReleaseUninstallArgs),
+
+    // Upgrade a release
+    Upgrade(ReleaseUpgradeArgs),
 }
 
 #[derive(Debug, Args)]
@@ -83,6 +86,25 @@ pub struct ReleaseInstallArgs {
     local_cluster: CommonClusterArgs,
 }
 
+#[derive(Debug, Args)]
+pub struct ReleaseUpgradeArgs {
+    /// Release to upgrade to
+    #[arg(name = "RELEASE")]
+    release: String,
+
+    /// List of product operators to upgrade
+    #[arg(short, long = "include", group = "products")]
+    included_products: Vec<String>,
+
+    /// Blacklist of product operators to install
+    #[arg(short, long = "exclude", group = "products")]
+    excluded_products: Vec<String>,
+
+    /// Namespace in the cluster used to deploy the operators
+    #[arg(long, default_value = DEFAULT_OPERATOR_NAMESPACE, visible_aliases(["operator-ns"]))]
+    pub operator_namespace: String,
+}
+
 #[derive(Debug, Args)]
 pub struct ReleaseUninstallArgs {
     /// Name of the release to uninstall
@@ -111,6 +133,9 @@ pub enum CmdError {
     #[snafu(display("failed to install release"))]
     ReleaseInstall { source: release::Error },
 
+    #[snafu(display("failed to upgrade CRDs for release"))]
+    CrdUpgrade { source: release::Error },
+
     #[snafu(display("failed to uninstall release"))]
     ReleaseUninstall { source: release::Error },
 
@@ -146,6 +171,7 @@ impl ReleaseArgs {
             ReleaseCommands::Describe(args) => describe_cmd(args, cli, release_list).await,
             ReleaseCommands::Install(args) => install_cmd(args, cli, release_list).await,
             ReleaseCommands::Uninstall(args) => uninstall_cmd(args, cli, release_list).await,
+            ReleaseCommands::Upgrade(args) => upgrade_cmd(args, cli, release_list).await,
         }
     }
 }
@@ -314,6 +340,60 @@ async fn install_cmd(
     }
 }
 
+#[instrument(skip(cli, release_list))]
+async fn upgrade_cmd(
+    args: &ReleaseUpgradeArgs,
+    cli: &Cli,
+    release_list: release::ReleaseList,
+) -> Result<String, CmdError> {
+    debug!(release = %args.release, "Upgrading release");
+    Span::current().pb_set_style(&ProgressStyle::with_template("").unwrap());
+
+    match release_list.get(&args.release) {
+        Some(release) => {
+            let mut output = cli.result();
+            let client = Client::new().await.context(KubeClientCreateSnafu)?;
+
+            // Uninstall the old operator release first
+            release
+                .uninstall(&args.operator_namespace)
+                .context(ReleaseUninstallSnafu)?;
+
+            // Upgrade the CRDs for all the operators to be upgraded
+            release
+                .upgrade_crds(
+                    &args.included_products,
+                    &args.excluded_products,
+                    &args.operator_namespace,
+                    &client,
+                )
+                .await
+                .context(CrdUpgradeSnafu)?;
+
+            // Install the new operator release
+            release
+                .install(
+                    &args.included_products,
+                    &args.excluded_products,
+                    &args.operator_namespace,
+                    &ChartSourceType::from(cli.chart_type()),
+                )
+                .await
+                .context(ReleaseInstallSnafu)?;
+
+            output
+                .with_command_hint(
+                    "stackablectl operator installed",
+                    "list installed operators",
+                )
+                .with_output(format!("Upgraded to release '{}'", args.release));
+
+            Ok(output.render())
+        }
+        None => Ok("No such release".into()),
+    }
+}
+
 #[instrument(skip(cli, release_list))]
 async fn uninstall_cmd(
     args: &ReleaseUninstallArgs,