Feat/kratix

.env

@@ -0,0 +1,13 @@
+RUST_LOG=debug
+
+WORKSPACE="/Users/hhill/work/oc"
+
+BASE_INSTANCE="$WORKSPACE/promise-flink/internal/configure-pipeline/resources/minimal-flinkdep-manifest.yaml"
+
+DEPENDENCIES_DIR="$WORKSPACE/promise-flink/internal/configure-pipeline/dependencies"
+
+RESOURCES_DIR="$WORKSPACE/promise-flink/internal/configure-pipeline/resources"
+
+KRATIX_INPUT="$WORKSPACE/promise-flink/internal/configure-pipeline/tests/test-input"
+
+KRATIX_OUTPUT="$WORKSPACE/promise-flink/internal/configure-pipeline/tests/test-output"

.gitignore

@@ -12,3 +12,5 @@ Cargo.lock
 
 # MSVC Windows builds of rustc generate these, which store debugging information
 *.pdb
+
+*.DS_Store

Cargo.toml

@@ -1,8 +1,14 @@
 [package]
-name = "kratix-pipeline-rust"
+name = "kratix_utils"
 version = "0.1.0"
 edition = "2021"
+authors = ["Howard Hill <[email protected]>"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+yaml-rust2 = "0.7.0"
+dotenv = "0.15.0"
+log = "0.4.21"
+serde_yaml = "0.9.34"
+toml = "0.5" 

dependencies/.gitkeep

@@ -0,0 +1 @@
+.gitkeep

src/lib.rs

@@ -0,0 +1,125 @@
+pub mod pipeline;
+pub mod promise;
+use log;
+use crate::pipeline::PipelineConfig;
+use std::{env, process};
+use dotenv::dotenv;
+
+// Structure to hold potential errors
+#[derive(Debug)]
+struct EnvVarError {
+    var_name: String,
+}
+
+
+
+pub fn run_pipeline(args:Vec<String>) {
+
+
+    dotenv().ok();
+
+    // Validate environment variables up front
+    match validate_env_vars() {
+        Ok(()) => (), // Everything is good, proceed
+        Err(errors) => {
+            eprintln!("Error: Missing environment variables:");
+            for error in errors {
+                log::warn!(" - {}", error.var_name);
+            }
+            process::exit(1);
+        }
+    }
+
+
+    if args.len() < 2 {
+        log::warn!("Usage: <command> [build, pipeline, load, push, rmi, pull]");
+        process::exit(1);
+    }
+
+    // Extract validated environment variables
+    let workflow_type = env::var("KRATIX_WORKFLOW_TYPE").unwrap();
+    let base_instance = env::var("BASE_INSTANCE").unwrap();
+    let dep_dir = env::var("DEPENDENCIES_DIR").unwrap();
+    let res_dir = env::var("RESOURCES_DIR").unwrap();
+    let kratix_input_dir = env::var("KRATIX_INPUT").unwrap();
+    let kratix_output_dir = env::var("KRATIX_OUTPUT").unwrap();
+
+
+    let config = PipelineConfig::new(
+            &base_instance,
+            &res_dir,
+            &dep_dir,
+            &kratix_output_dir,
+            &kratix_input_dir,
+            &workflow_type);
+
+    log::debug!("<- Start Pipeline ({}) ->", config.workflow_type());
+
+
+
+    match config.workflow_type() {
+        "promise" => {
+            // Fullful promise.yaml
+            if let Err(err) =
+                // tmp/transfer/dependecies -> /kratix/output
+                pipeline::copy_files(config.dep_dir(), config.kratix_output_dir()) {
+                log::warn!("Error during file copy: {}", err);
+            }
+        },
+        "resource" => {
+            log::debug!("  1. transform resource");
+            // Fullfil resource_request.yaml
+            promise::transform(config.res_dir(),
+                               config.base_instance(),
+                               config.kratix_output_dir(),
+                               config.kratix_input_dir());
+        },
+        "request" => {
+            log::debug!("  1. transform request");
+            // Fullfil resource_request.yaml
+            promise::transform(config.res_dir(),
+                               config.base_instance(),
+                               config.kratix_output_dir(),
+                               config.kratix_input_dir());
+        }
+        _ => {
+            log::error!("No workflow_type");
+        }
+    }
+
+
+    //pipeline::status();
+
+    //pipeline::list_files_recursively(_kratix_output_dir);
+
+    log::debug!("<- End Pipeline ->");
+}
+
+
+// validation function
+fn validate_env_vars() -> Result<(), Vec<EnvVarError>> {
+    let required_vars = vec![
+        "KRATIX_WORKFLOW_TYPE",
+        "BASE_INSTANCE",
+        "DEPENDENCIES_DIR",
+        "RESOURCES_DIR",
+        "KRATIX_INPUT",
+        "KRATIX_OUTPUT",
+    ];
+
+    let mut errors = Vec::new();
+
+    for var_name in required_vars {
+        if env::var(var_name).is_err() {
+            errors.push(EnvVarError {
+                var_name: var_name.to_string(),
+            });
+        }
+    }
+
+    if errors.is_empty() {
+        Ok(())
+    } else {
+        Err(errors)
+    }
+}

src/pipeline/mod.rs

@@ -0,0 +1,69 @@
+use log;
+use std::fs;
+use std::io::Result;
+use std::path::Path;
+use yaml_rust2::Yaml;
+use yaml_rust2::YamlLoader;
+
+mod pipeline_config;
+pub use pipeline_config::PipelineConfig; // Re-export PipelineConfig
+
+pub fn load_file(file: &str) -> Result<Yaml> {
+    let path = Path::new(file);
+    let contents = fs::read_to_string(path).expect("Should have been able to read the file");
+
+    let docs = YamlLoader::load_from_str(&contents).unwrap();
+    let doc = &docs[0];
+
+    Ok(doc.clone()) // Return a copy of the doc
+}
+
+pub fn copy_files(source_dir: &str, destination_dir: &str) -> Result<()> {
+    log::debug!("pipeline::copy_files {} to {}", source_dir, destination_dir);
+    // Create the output directory if it doesn't exist
+    fs::create_dir_all(destination_dir)?;
+
+    // Iterate over files in the source directory
+    for entry in fs::read_dir(source_dir)? {
+        let entry = entry?;
+        let path = entry.path();
+
+        if path.is_file() {
+            let filename = path.file_name().unwrap().to_str().unwrap();
+            let destination_path = format!("{}/{}", destination_dir, filename);
+
+            // Copy the file
+            fs::copy(path, destination_path)?;
+        }
+    }
+
+    Ok(())
+}
+
+#[allow(dead_code)]
+pub fn list_files_recursively(path: &str) {
+    if let Ok(entries) = fs::read_dir(path) {
+        for entry in entries {
+            if let Ok(entry) = entry {
+                let path = entry.path();
+                log::debug!("{}", path.display());
+
+                if path.is_dir() {
+                    list_files_recursively(&path.to_string_lossy()); // Recursion!
+                }
+            }
+        }
+    } else {
+        log::warn!("Error reading directory: {}", path);
+    }
+}
+
+#[allow(dead_code)]
+pub fn status() {
+    if let Err(err) = fs::copy(
+        "/tmp/transfer/resources/status.yaml",
+        "/kratix/metadata/status.yaml",
+    ) {
+        log::warn!("Error during file copy: {}", err);
+    }
+}

src/pipeline/pipeline_config.rs

@@ -0,0 +1,52 @@
+pub struct PipelineConfig {
+    base_instance: String,
+    res_dir: String,
+    dep_dir: String,
+    kratix_output_dir: String,
+    kratix_input_dir: String,
+    workflow_type: String,
+}
+
+impl PipelineConfig {
+    pub fn new(
+        base_instance: &str,
+        res_dir: &str,
+        dep_dir: &str,
+        kratix_output_dir: &str,
+        kratix_input_dir: &str,
+        workflow_type: &str,
+    ) -> PipelineConfig {
+        PipelineConfig {
+            base_instance: base_instance.to_string(),
+            res_dir: res_dir.to_string(),
+            dep_dir: dep_dir.to_string(),
+            kratix_output_dir: kratix_output_dir.to_string(),
+            kratix_input_dir: kratix_input_dir.to_string(),
+            workflow_type: workflow_type.to_string(),
+        }
+    }
+
+    pub fn base_instance(&self) -> &str {
+        &self.base_instance
+    }
+
+    pub fn res_dir(&self) -> &str {
+        &self.res_dir
+    }
+
+    pub fn dep_dir(&self) -> &str {
+        &self.dep_dir
+    }
+
+    pub fn kratix_output_dir(&self) -> &str {
+        &self.kratix_output_dir
+    }
+
+    pub fn kratix_input_dir(&self) -> &str {
+        &self.kratix_input_dir
+    }
+
+    pub fn workflow_type(&self) -> &str {
+        &self.workflow_type
+    }
+}

src/promise/mod.rs

@@ -0,0 +1,83 @@
+use serde_yaml::{to_string, Value};
+use std::fs::read_to_string;
+use std::fs::write; // Import 'write' for generic write operations, File for opening
+use std::io::Error; // Import ErrorKind
+use toml::Value as TomlValue;
+
+/// Handle Pipeline transformation
+///
+/// Need to use properties.toml or pass a functton or interface
+pub fn transform(_res_dir: &str, _res_path: &str, _kout_dir: &str, _kin_dir: &str) {
+    log::debug!("resource  {}", _res_path);
+
+    let mut base_instance: Value = serde_yaml::from_str(load_file(_res_path).unwrap().as_str())
+        .expect("Error deserializing YAML file");
+
+    /* 1. GET Kratix Input Object i.e Promise Req */
+    let new_kin_path = format!("{}/object.yaml", _kin_dir);
+    log::debug!("kratix input {}", new_kin_path);
+    let kin_doc: Value = serde_yaml::from_str(load_file(&new_kin_path).unwrap().as_str())
+        .expect("Error deserializing YAML file");
+
+    let new_prop_path = format!("{}/properties.toml", _res_dir);
+    let properties_toml = match load_file(&new_prop_path) {
+        Ok(file_contents) => file_contents,
+        Err(err) => {
+            log::error!("Error loading properties.toml: {}", err); // Log the error
+            return; // Or handle the error in some other way
+        }
+    };
+    let mappings: TomlValue = toml::from_str(&properties_toml).unwrap();
+    log::debug!("mappings.as_table() {:?}", mappings.as_table());
+
+    for (toml_key, toml_value) in mappings.as_table().unwrap() {
+        let source_path = toml_value.as_str().unwrap(); // Get path to copy from kin_doc
+        let target_path = toml_key; // Get path to assign to in base_instance
+
+        // Dynamically fetch the value from kin_doc & perform assignment
+        let value_to_assign = get_nested_value(&kin_doc, source_path.split('.').collect());
+        set_nested_value(
+            &mut base_instance,
+            target_path.split('.').collect(),
+            value_to_assign,
+        );
+    }
+
+    // let new_name = kin_doc["spec"]["name"].as_str().unwrap().to_string();
+
+    // log::debug!("Template Instance Name {}",base_instance["metadata"]["name"].as_str().unwrap().to_string());
+
+    // base_instance["metadata"]["name"] = serde_yaml::Value::String(new_name.clone());
+
+    // log::debug!("Promise Request Name {}",base_instance["metadata"]["name"].as_str().unwrap().to_string());
+
+    let new_kout_path = format!("{}/flink-instance.yaml", _kout_dir);
+    log::debug!("kratix output {}", new_kout_path);
+
+    // // Serialize the modified YAML
+    let yaml_str = to_string(&base_instance).unwrap();
+
+    write(new_kout_path.clone(), yaml_str).unwrap();
+}
+
+fn load_file(path: &str) -> Result<String, Error> {
+    read_to_string(path)
+}
+
+fn get_nested_value(value: &Value, path: Vec<&str>) -> Value {
+    let mut current = value;
+    for key in path {
+        current = &current[key];
+    }
+    current.clone()
+}
+
+fn set_nested_value(value: &mut Value, path: Vec<&str>, new_value: Value) {
+    let mut current = value;
+    for key in path.iter().take(path.len() - 1) {
+        // Navigate to the parent
+        current = &mut current[key];
+    }
+    let last_key = path.last().unwrap();
+    current[last_key] = new_value;
+}