Continue install flow

Author: Techassi

Cargo.toml

@@ -26,6 +26,7 @@ utoipa = { workspace = true }
 snafu = { workspace = true }
 which = { workspace = true }
 serde = { workspace = true }
+tera = { workspace = true }
 url = { workspace = true }
 
 [build-dependencies]
@@ -50,6 +51,7 @@ utoipa = "3.3.0"
 axum = "0.6.18"
 snafu = "0.7.4"
 which = "4.4.0"
+tera = "1.18.1"
 url = "2.3.1"
 xdg = "2.4.1"
 phf = "0.11.1"

src/constants.rs

@@ -6,3 +6,4 @@ pub const DEFAULT_LOCAL_CLUSTER_NAME: &str = "stackable-data-platform";
 pub const DEFAULT_CACHE_MAX_AGE_IN_SECS: u64 = 3_600; // One hour in seconds
 
 pub const HELM_ERROR_PREFIX: &str = "ERROR:";
+pub const HELM_DEFAULT_CHART_VERSION: &str = ">0.0.0-0";

src/helm.rs

@@ -1,11 +1,12 @@
+use std::fmt::Display;
 use std::str::{self, Utf8Error};
 use std::{collections::HashMap, ffi::CStr, os::raw::c_char};
 
 use serde::Deserialize;
 use snafu::{ResultExt, Snafu};
 use tracing::{debug, error, info, instrument};
 
-use crate::constants::HELM_ERROR_PREFIX;
+use crate::constants::{HELM_DEFAULT_CHART_VERSION, HELM_ERROR_PREFIX};
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -78,16 +79,34 @@ pub enum HelmInstallReleaseStatus {
     /// Indicates that a release is already installed with a different version
     /// than requested.
     ReleaseAlreadyInstalledWithversion {
+        release_name: String,
         current_version: String,
         requested_version: String,
     },
 
     /// Indicates that a release is already installed, but no specific version
     /// was requested.
-    ReleaseAlreadyInstalledUnspecified(String),
+    ReleaseAlreadyInstalledUnspecified {
+        release_name: String,
+        current_version: String,
+    },
 
     /// Indicates that the release was installed successfully.
-    Installed,
+    Installed(String),
+}
+
+impl Display for HelmInstallReleaseStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            HelmInstallReleaseStatus::ReleaseAlreadyInstalledWithversion {
+                release_name,
+                current_version,
+                requested_version,
+            } => write!(f, "The release {release_name} ({current_version}) is already installed (requested {requested_version}), skipping."),
+            HelmInstallReleaseStatus::ReleaseAlreadyInstalledUnspecified{release_name, current_version} => write!(f, "The release {release_name} ({current_version}) is already installed and no specific version was requested, skipping."),
+            HelmInstallReleaseStatus::Installed(release_name) => write!(f, "The release {release_name} was successfully installed."),
+        }
+    }
 }
 
 #[repr(C)]
@@ -153,11 +172,11 @@ pub fn install_release_from_repo(
                 if chart_version == current_version {
                     return Ok(
                         HelmInstallReleaseStatus::ReleaseAlreadyInstalledWithversion {
+                            release_name: release_name.to_string(),
                             current_version: current_version.to_string(),
                             requested_version: chart_version.to_string(),
                         },
                     );
-                    // return Ok(format!("The release {release_name} ({current_version}) is already installed, skipping."));
                 } else {
                     return Err(HelmError::InstallReleaseError {
                         source: HelmInstallReleaseError::ReleaseAlreadyInstalled {
@@ -170,16 +189,17 @@ pub fn install_release_from_repo(
             }
             None => {
                 return Ok(
-                    HelmInstallReleaseStatus::ReleaseAlreadyInstalledUnspecified(
-                        current_version.into(),
-                    ),
+                    HelmInstallReleaseStatus::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(">0.0.0-0"); // TODO (Techassi): Move this into a constant
+    let chart_version = chart_version.unwrap_or(HELM_DEFAULT_CHART_VERSION);
 
     debug!(
         "Installing Helm release {} ({}) from chart {}",
@@ -195,7 +215,9 @@ pub fn install_release_from_repo(
         suppress_output,
     )?;
 
-    Ok(HelmInstallReleaseStatus::Installed)
+    Ok(HelmInstallReleaseStatus::Installed(
+        release_name.to_string(),
+    ))
 }
 
 fn install_release(

src/platform/operator/spec.rs

@@ -3,6 +3,8 @@ use std::{fmt::Display, str::FromStr};
 use snafu::Snafu;
 use tracing::{info, instrument};
 
+use crate::helm;
+
 pub const VALID_OPERATORS: &[&str] = &[
     "airflow",
     "commons",
@@ -21,14 +23,6 @@ pub const VALID_OPERATORS: &[&str] = &[
     "zookeeper",
 ];
 
-/// OperatorSpec describes the format of an operator name with optional version number. The string format is
-/// `<OPERATOR_NAME>(=<VERSION>)`. Valid values values are: `operator`, `operator=1.2.3` or `operator=1.2.3-rc1`.
-#[derive(Clone, Debug)]
-pub struct OperatorSpec {
-    pub version: Option<String>,
-    pub name: String,
-}
-
 #[derive(Debug, Snafu, PartialEq)]
 pub enum OperatorSpecParseError {
     #[snafu(display("invalid equal sign count in operator spec, expected one"))]
@@ -44,6 +38,14 @@ pub enum OperatorSpecParseError {
     InvalidName { name: String },
 }
 
+/// OperatorSpec describes the format of an operator name with optional version number. The string format is
+/// `<OPERATOR_NAME>(=<VERSION>)`. Valid values values are: `operator`, `operator=1.2.3` or `operator=1.2.3-rc1`.
+#[derive(Clone, Debug)]
+pub struct OperatorSpec {
+    pub version: Option<String>,
+    pub name: String,
+}
+
 impl Display for OperatorSpec {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
@@ -124,12 +126,22 @@ impl TryFrom<&str> for OperatorSpec {
 }
 
 impl OperatorSpec {
-    pub fn new(name: String, version: Option<String>) -> Result<Self, OperatorSpecParseError> {
-        if !VALID_OPERATORS.contains(&name.as_str()) {
-            return Err(OperatorSpecParseError::InvalidName { name });
+    pub fn new<T>(name: T, version: Option<String>) -> Result<Self, OperatorSpecParseError>
+    where
+        T: AsRef<str>,
+    {
+        let name = name.as_ref();
+
+        if !VALID_OPERATORS.contains(&name) {
+            return Err(OperatorSpecParseError::InvalidName {
+                name: name.to_string(),
+            });
         }
 
-        Ok(Self { version, name })
+        Ok(Self {
+            version,
+            name: name.to_string(),
+        })
     }
 
     /// Returns the name used by Helm
@@ -151,14 +163,26 @@ impl OperatorSpec {
 
     /// Installs the operator using Helm
     #[instrument(skip_all)]
-    pub fn install(&self) {
+    pub fn install(&self) -> Result<(), helm::HelmError> {
         info!("Installing operator {}", self);
 
         let helm_name = self.helm_name();
         let helm_repo = self.helm_repo();
+        let version = match &self.version {
+            Some(version) => Some(version.as_str()),
+            None => None,
+        };
 
         // Install using Helm
-        todo!()
+        match helm::install_release_from_repo(
+            &self.name, &helm_name, &helm_repo, &helm_name, version, None, "", true,
+        ) {
+            Ok(status) => {
+                println!("{status}");
+                Ok(())
+            }
+            Err(err) => Err(err),
+        }
     }
 }
 

src/platform/product.rs

@@ -4,5 +4,5 @@ use serde::{Deserialize, Serialize};
 #[serde(rename_all = "camelCase")]
 pub struct ProductSpec {
     #[serde(rename = "operatorVersion")]
-    version: String,
+    pub version: String,
 }

src/platform/release/spec.rs

@@ -1,11 +1,27 @@
 use indexmap::IndexMap;
 use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
 use tracing::{info, instrument};
 
 #[cfg(feature = "openapi")]
 use utoipa::ToSchema;
 
-use crate::platform::product::ProductSpec;
+use crate::{
+    helm::HelmError,
+    platform::{
+        operator::{OperatorSpec, OperatorSpecParseError},
+        product::ProductSpec,
+    },
+};
+
+#[derive(Debug, Snafu)]
+pub enum ReleaseInstallError {
+    #[snafu(display("failed to parse operator spec: {source}"))]
+    OperatorSpecParseError { source: OperatorSpecParseError },
+
+    #[snafu(display("failed with Helm error: {source}"))]
+    HelmError { source: HelmError },
+}
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -24,19 +40,29 @@ pub struct ReleaseSpec {
 
 impl ReleaseSpec {
     #[instrument(skip_all)]
-    pub fn install(&self, include_products: &[String], exclude_products: &[String]) {
+    pub fn install(
+        &self,
+        include_products: &[String],
+        exclude_products: &[String],
+    ) -> Result<(), ReleaseInstallError> {
         info!("Installing release");
 
         for (product_name, product) in &self.products {
             let included = include_products.is_empty() || include_products.contains(&product_name);
             let excluded = exclude_products.contains(&product_name);
 
             if included && !excluded {
+                info!("Installing product {}", product_name);
+
+                // Create operator spec
+                let operator = OperatorSpec::new(product_name, Some(product.version.clone()))
+                    .context(OperatorSpecParseSnafu {})?;
+
                 // Install operator
-                todo!()
+                operator.install().context(HelmSnafu {})?
             }
         }
 
-        todo!()
+        Ok(())
     }
 }

src/platform/stack/mod.rs

@@ -2,13 +2,9 @@ use indexmap::IndexMap;
 use serde::{Deserialize, Serialize};
 
 mod spec;
-use snafu::Snafu;
 pub use spec::*;
 
-use crate::{
-    common::{List, SpecIter},
-    utils::params::IntoParametersError,
-};
+use crate::common::{List, SpecIter};
 
 /// This struct describes a complete demos v2 file
 #[derive(Debug, Deserialize, Serialize)]
@@ -25,12 +21,3 @@ impl SpecIter<StackSpecV2> for StacksV2 {
 }
 
 pub type StackList = List<StacksV2, StackSpecV2>;
-
-#[derive(Debug, Snafu)]
-pub enum StackError {
-    #[snafu(display("parameter parse error: {source}"))]
-    ParameterError { source: IntoParametersError },
-
-    #[snafu(display("no such stack"))]
-    NoSuchStack,
-}

src/platform/stack/spec.rs

@@ -1,5 +1,5 @@
 use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
+use snafu::{ResultExt, Snafu};
 use tracing::{info, instrument};
 
 #[cfg(feature = "openapi")]
@@ -9,16 +9,33 @@ use crate::{
     common::ManifestSpec,
     platform::{
         demo::DemoParameter,
-        release::ReleaseList,
-        stack::{ParameterSnafu, StackError},
+        release::{ReleaseInstallError, ReleaseList},
+    },
+    utils::{
+        params::{
+            IntoParameters, IntoParametersError, Parameter, RawParameter, RawParameterParseError,
+        },
+        path::PathOrUrl,
+        read::read_yaml_data_with_templating,
     },
-    utils::params::{IntoParameters, Parameter, RawParameter, RawParameterParseError},
 };
 
 pub type RawStackParameterParseError = RawParameterParseError;
 pub type RawStackParameter = RawParameter;
 pub type StackParameter = Parameter;
 
+#[derive(Debug, Snafu)]
+pub enum StackError {
+    #[snafu(display("parameter parse error: {source}"))]
+    ParameterError { source: IntoParametersError },
+
+    #[snafu(display("no such stack"))]
+    NoSuchStack,
+
+    #[snafu(display("release install error: {source}"))]
+    ReleaseInstallError { source: ReleaseInstallError },
+}
+
 /// This struct describes a stack with the v2 spec
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -53,23 +70,37 @@ impl StackSpecV2 {
     pub fn install(&self, release_list: ReleaseList) -> Result<(), StackError> {
         info!("Installing stack");
 
+        // Get the release by name
         let release = release_list
             .get(&self.release)
             .ok_or(StackError::NoSuchStack)?;
 
-        release.install(&self.operators, &[]);
+        // Install the release
+        release
+            .install(&self.operators, &[])
+            .context(ReleaseInstallSnafu {})?;
 
         todo!()
     }
 
     #[instrument(skip_all)]
     pub fn install_stack_manifests(&self, parameters: &[String]) -> Result<(), StackError> {
         info!("Installing stack manifests");
+
         let parameters = parameters
             .to_owned()
             .into_params(&self.parameters)
             .context(ParameterSnafu {})?;
 
+        for manifest in &self.manifests {
+            match manifest {
+                ManifestSpec::HelmChart(helm_file) => {
+                    let helm_chart = read_yaml_data_with_templating(helm_file, &parameters);
+                }
+                ManifestSpec::PlainYaml(_) => todo!(),
+            }
+        }
+
         todo!()
     }