Fixed several bugs in Thorium

🐛 Bug Fixes

- *(operator)* Fixed issue where the config could not be made into a CRD
- *(scaler)* Scalers now only requests details on clusters they care about
- *(agent)* Fixed issue where the agent was not injecting kwargs correctly
- *(api)* Fixed issue where the API was incorrectly rejecting result paths

Change Details

fix(agent): Fixed issue where the agent was not injecting kwargs correctly

This was causing the agent to add a list of values after each kwargs
instead of repeating the kwarg for each value. This means Thorium will
now use --kwarg <value> --kwarg <value> instead of --kwarg <value>
<value>.

fix(api): Fixed issue where the API was incorrectly rejecting result paths

This was due to an incorrect check for '..' in file paths.

fix(operator): Fixed issue where the config could not be made into a CRD

This was caused by an enum having different types for each branch. The
downside of this fix is that our config does allow someone to configure
certificate validation settings while also disabling certificate
validation. That could lead to some confusing scenarios where you think
validation is enabled but its not.

fix(scaler): Scalers now only requests details on clusters they care about

This helps resolves issues where the scaler tries to get info on clusters
that it cannot and will not schedule on.

Closes #31

Author: mjcarson

Cargo.lock

@@ -16,7 +16,7 @@ members = [
 ]
 
 [workspace.package]
-version = "1.1.1"
+version = "1.1.2"
 authors = ["mcarson <mcarson@sandia.gov>", "gmbaker <gmbaker@sandia.gov>", "jehamza <jehamza@sandia.gov>"]
 edition = "2024"
 

Cargo.toml

@@ -36,14 +36,14 @@ infer = { version = "0.19.0", default-features = false, features = ["std"] }
 
 # enable cgroups support for linux
 [target.'cfg(target_os = "linux")'.dependencies]
-thorium = { version = "1.1.1", path="../api", default-features = false, features = ["client", "cgroups", "crossbeam-err", "trace"]}
+thorium = { version = "1.1.2", path="../api", default-features = false, features = ["client", "cgroups", "crossbeam-err", "trace"]}
 cgroups-rs = "0.3"
 controlgroup = "0.3"
 
 
 # disable cgroups support when on windows
 [target.'cfg(any(target_os = "windows", target_os = "macos"))'.dependencies]
-thorium = { version = "1.1.1", path="../api", default-features = false, features = ["client", "crossbeam-err", "trace"]}
+thorium = { version = "1.1.2", path="../api", default-features = false, features = ["client", "crossbeam-err", "trace"]}
 
 
 [features]

agent/Cargo.toml

@@ -4,8 +4,8 @@ use crossbeam::channel::Sender;
 use path_clean::PathClean;
 use std::path::PathBuf;
 use thorium::{
-    models::{ArgStrategy, GenericJob, GenericJobKwargs, GenericJobOpts, Image, KwargDependency},
     Error,
+    models::{ArgStrategy, GenericJob, GenericJobKwargs, GenericJobOpts, Image, KwargDependency},
 };
 use tracing::instrument;
 
@@ -158,17 +158,22 @@ impl Cmd {
                     wipe = false;
                     // expand this arg if its a joint arg
                     let (key, value) = Self::expander(arg);
-                    // inject value source docker config
-                    self.built.push(key.clone());
                     // check if this arg should be overridden
                     if self.kwargs.contains_key(&key) {
                         // enable wipe until hit another kwarg
                         wipe = true;
                         // override value if one was set
-                        let mut new_value = self.kwargs.remove(&key).unwrap();
-                        // override value with our own value
-                        self.built.append(&mut new_value);
+                        let new_values = self.kwargs.remove(&key).unwrap();
+                        // for each of our values add our kwarg
+                        for new_value in new_values {
+                            // add our key
+                            self.built.push(key.clone());
+                            // override value with our own value
+                            self.built.push(new_value);
+                        }
                     } else if value.is_some() {
+                        // add our key
+                        self.built.push(key.clone());
                         // push value from docker info
                         self.built.push(value.unwrap());
                     }
@@ -183,9 +188,14 @@ impl Cmd {
             }
         }
         // append all left over custom kwargs args if any were set
-        for (key, mut values) in self.kwargs.drain() {
-            self.built.push(key);
-            self.built.append(&mut values);
+        for (key, values) in self.kwargs.drain() {
+            // add our kwargs
+            for value in values {
+                // add our key
+                self.built.push(key.clone());
+                // add our value
+                self.built.push(value);
+            }
         }
     }
 
@@ -419,11 +429,18 @@ mod tests {
             samples: vec!["sample1".into(), "sample2".into()],
             ephemeral: vec!["file.txt".into(), "other.txt".into()],
             parent_ephemeral: HashMap::default(),
-            repos: vec![RepoDependency {
-                url: "github.com/curl/curl".into(),
-                commitish: Some("master".into()),
-                kind: Some(CommitishKinds::Branch),
-            }],
+            repos: vec![
+                RepoDependency {
+                    url: "github.com/curl/curl".into(),
+                    commitish: Some("main".into()),
+                    kind: Some(CommitishKinds::Branch),
+                },
+                RepoDependency {
+                    url: "github.com/notcurl/notcurl".into(),
+                    commitish: Some("main".into()),
+                    kind: Some(CommitishKinds::Branch),
+                },
+            ],
             trigger_depth: None,
         }
     }
@@ -750,6 +767,7 @@ mod tests {
                 "corn.py",
                 "--inputs",
                 "sample1",
+                "--inputs",
                 "sample2"
             )
         );
@@ -796,6 +814,7 @@ mod tests {
                 "pos2",
                 "--inputs",
                 "sample1",
+                "--inputs",
                 "sample2"
             )
         );
@@ -840,7 +859,9 @@ mod tests {
                 "corn.py",
                 "--inputs",
                 "sample0",
+                "--inputs",
                 "sample1",
+                "--inputs",
                 "sample2"
             )
         );
@@ -974,7 +995,7 @@ mod tests {
 
     /// Test a barebones job with samples but no overlays
     #[tokio::test]
-    async fn empty_ephemnerals_kwargs() {
+    async fn empty_ephemerals_kwargs() {
         // create a temporary log channel
         let (mut logs_tx, _logs_rx) = crossbeam::channel::unbounded::<String>();
         // generate an image and set the kwarg to pass in samples with
@@ -1009,6 +1030,7 @@ mod tests {
                 "corn.py",
                 "--ephemeral",
                 "file.txt",
+                "--ephemeral",
                 "other.txt"
             )
         );
@@ -1055,6 +1077,7 @@ mod tests {
                 "pos2",
                 "--ephemeral",
                 "file.txt",
+                "--ephemeral",
                 "other.txt"
             )
         );
@@ -1099,7 +1122,9 @@ mod tests {
                 "corn.py",
                 "--ephemeral",
                 "first.txt",
+                "--ephemeral",
                 "file.txt",
+                "--ephemeral",
                 "other.txt"
             )
         );
@@ -1115,7 +1140,7 @@ mod tests {
         // generate a job
         let mut job = generate_job();
         // build stage args with just kwargs
-        job.args = job.args.kwarg("--1", vec!["1"]);
+        job.args = job.args.kwarg("--nums", vec!["1", "2"]);
         // make this job a generator
         job.generator = true;
         let cmd = Cmd::new(
@@ -1149,8 +1174,10 @@ mod tests {
         let args = vec_string!(
             "/usr/bin/python3",
             "corn.py",
-            "--1",
+            "--nums",
             "1",
+            "--nums",
+            "2",
             "--reaction",
             &reaction,
             "--job",
@@ -1355,7 +1382,9 @@ mod tests {
                 "pos2",
                 "--inputs",
                 "sample0",
+                "--inputs",
                 "sample1",
+                "--inputs",
                 "sample2",
                 "--corn",
                 "--beans"
@@ -1463,7 +1492,9 @@ mod tests {
                 "pos2",
                 "--ephemeral",
                 "first.txt",
+                "--ephemeral",
                 "file.txt",
+                "--ephemeral",
                 "other.txt",
                 "--corn",
                 "--beans"
@@ -1601,8 +1632,12 @@ mod tests {
                 "sample2",
                 "--image1-results",
                 "/tmp/thorium/testing/prior-results/sample1/image1",
+                "--image1-results",
                 "/tmp/thorium/testing/prior-results/sample2/image1",
-                "/tmp/thorium/testing/prior-results/github.com/curl/curl/image1"
+                "--image1-results",
+                "/tmp/thorium/testing/prior-results/github.com/curl/curl/image1",
+                "--image1-results",
+                "/tmp/thorium/testing/prior-results/github.com/notcurl/notcurl/image1"
             )
         );
         // remove the test directory

agent/src/libs/agents/registry.rs

@@ -134,7 +134,9 @@ once_cell = { version = "1.21.3", optional = true }
 utoipa = { version = "5", features = ["axum_extras", "chrono", "uuid", "time", "url"], optional = true }
 utoipa-swagger-ui = { version = "9", features = ["axum"], optional = true }
 lettre = { version = "0.11", features = ["tokio1", "tokio1-rustls-tls", "builder", "smtp-transport"], default-features = false, optional = true }
-thorium-derive = { path = "../thorium-derive", version = "1.1.1", optional = true}
+thorium-derive = { path = "../thorium-derive", version = "1.1.2", optional = true}
+percent-encoding = { version = "2.3.1", optional = true }
+dashmap = { version = "6.1", optional = true }
 
 # rkyv dependencies
 rkyv = { version = "=0.7.43", features = ["arbitrary_enum_discriminant", "uuid", "validation"], optional = true }

api/Cargo.toml

@@ -1960,6 +1960,54 @@ pub struct Scylla {
     pub auth: Option<ScyllaAuth>,
 }
 
+/// The options for Elastic certificate validation
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
+pub enum ElasticCertValidation {
+    /// Perform full validation based on a PEM encoded ca cert with CN/SAN checks enabled
+    Full(PathBuf),
+    /// Validate against a cert generated by a specific PEM encoded CA cert at a path
+    /// without CN/SAN checks
+    CA(PathBuf),
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
+pub struct ElasticResults {
+    /// The name of the sample results index in elastic
+    #[serde(default = "default_elastic_sample_results_index")]
+    pub samples: String,
+    /// The name of the repo results index in elastic
+    #[serde(default = "default_elastic_repo_results_index")]
+    pub repos: String,
+}
+
+impl Default for ElasticResults {
+    fn default() -> Self {
+        Self {
+            samples: default_elastic_sample_results_index(),
+            repos: default_elastic_repo_results_index(),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
+pub struct ElasticTags {
+    /// The name of the sample tags index in elastic
+    #[serde(default = "default_elastic_sample_tags_index")]
+    pub samples: String,
+    /// The name of the repo tags index in elastic
+    #[serde(default = "default_elastic_repo_tags_index")]
+    pub repos: String,
+}
+
+impl Default for ElasticTags {
+    fn default() -> Self {
+        Self {
+            samples: default_elastic_sample_tags_index(),
+            repos: default_elastic_repo_tags_index(),
+        }
+    }
+}
+
 /// Adds a "thorium_" namespace to given elastic index
 fn elastic_namespace(index: &str) -> String {
     format!("thorium_{index}")
@@ -1993,7 +2041,10 @@ fn default_max_analyzed_offset() -> u32 {
     1_000_000
 }
 
-/// Scylla settings
+/// Elastic settings
+///
+/// The cert validation settings are overly complex to support the operator being able
+/// to set this as a valid CRD in K8s.
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
 pub struct Elastic {
     /// The node to connect to
@@ -2002,6 +2053,12 @@ pub struct Elastic {
     pub username: String,
     /// The password to use when authenticating
     pub password: String,
+    /// The options for certificate validation
+    #[serde(default)]
+    pub cert_validation: Option<ElasticCertValidation>,
+    /// Disable certificate validation
+    #[serde(default)]
+    pub insecure_certificates: bool,
     /// The options for results in elastic
     #[serde(default)]
     pub results: ElasticResults,
@@ -2017,40 +2074,41 @@ pub struct Elastic {
     pub max_analyzed_offset: u32,
 }
 
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
-pub struct ElasticResults {
-    /// The name of the sample results index in elastic
-    #[serde(default = "default_elastic_sample_results_index")]
-    pub samples: String,
-    /// The name of the repo results index in elastic
-    #[serde(default = "default_elastic_repo_results_index")]
-    pub repos: String,
-}
-
-impl Default for ElasticResults {
-    fn default() -> Self {
-        Self {
-            samples: default_elastic_sample_results_index(),
-            repos: default_elastic_repo_results_index(),
+impl Elastic {
+    /// Cast this elastic cert config into a ```CertificateValidation``` object
+    #[cfg(feature = "api")]
+    pub async fn to_cert_validation(&self) -> elasticsearch::cert::CertificateValidation {
+        // if insecure certificates is turned on then ignore certificate validation
+        if self.insecure_certificates {
+            return elasticsearch::cert::CertificateValidation::None;
         }
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema)]
-pub struct ElasticTags {
-    /// The name of the sample tags index in elastic
-    #[serde(default = "default_elastic_sample_tags_index")]
-    pub samples: String,
-    /// The name of the repo tags index in elastic
-    #[serde(default = "default_elastic_repo_tags_index")]
-    pub repos: String,
-}
-
-impl Default for ElasticTags {
-    fn default() -> Self {
-        Self {
-            samples: default_elastic_sample_tags_index(),
-            repos: default_elastic_repo_tags_index(),
+        // handle each specific type of cert validation config
+        match &self.cert_validation {
+            None => elasticsearch::cert::CertificateValidation::Default,
+            Some(ElasticCertValidation::Full(path)) => {
+                // read our CA cert from disk
+                let ca_bytes = tokio::fs::read(path).await.expect(&format!(
+                    "Failed to read elastic validation cert at {path:?}"
+                ));
+                // build our cert
+                let cert = elasticsearch::cert::Certificate::from_pem(&ca_bytes).expect(&format!(
+                    "Failed to parse certificate at {path:?} as a PEM encoded CA cert"
+                ));
+                // return the right validation behavior
+                elasticsearch::cert::CertificateValidation::Full(cert)
+            }
+            Some(ElasticCertValidation::CA(path)) => {
+                // read our CA cert from disk
+                let ca_bytes = tokio::fs::read(path).await.expect(&format!(
+                    "Failed to read elastic validation cert at {path:?}"
+                ));
+                // build our cert
+                let cert = elasticsearch::cert::Certificate::from_pem(&ca_bytes).expect(&format!(
+                    "Failed to parse certificate at {path:?} as a PEM encoded CA cert"
+                ));
+                // return the right validation behavior
+                elasticsearch::cert::CertificateValidation::Certificate(cert)
+            }
         }
     }
 }