Merge branch 'test_split' into pod_model_hash

Author: Synicix

.clippy.toml

@@ -1,3 +1,3 @@
-excessive-nesting-threshold = 5
+excessive-nesting-threshold = 6
 too-many-arguments-threshold = 10
 allowed-idents-below-min-chars = ["..", "k", "v", "f", "re", "id", "Ok", "'_"]

src/core/crypto.rs

@@ -50,11 +50,9 @@ pub fn hash_buffer(buffer: impl AsRef<[u8]>) -> String {
 ///
 /// Will return error if unable to access file.
 pub fn hash_file(filepath: impl AsRef<Path>) -> Result<String> {
-    hash_stream(
-        &mut File::open(&filepath).context(selector::InvalidFilepath {
-            path: filepath.as_ref(),
-        })?,
-    )
+    hash_stream(&mut File::open(&filepath).context(selector::InvalidPath {
+        path: filepath.as_ref(),
+    })?)
 }
 /// Evaluate checksum hash of a directory.
 ///
@@ -64,7 +62,10 @@ pub fn hash_file(filepath: impl AsRef<Path>) -> Result<String> {
 pub fn hash_dir(dirpath: impl AsRef<Path>) -> Result<String> {
     let summary: BTreeMap<String, String> = dirpath
         .as_ref()
-        .read_dir()?
+        .read_dir()
+        .context(selector::InvalidPath {
+            path: dirpath.as_ref(),
+        })?
         .map(|path| {
             let access_path = path?.path();
             Ok((

src/core/error.rs

@@ -122,8 +122,10 @@ impl fmt::Debug for OrcaError {
         match &self.kind {
             Kind::AgentCommunicationFailure { backtrace, .. }
             | Kind::EmptyDir { backtrace, .. }
+            | Kind::FailedToStartPod { backtrace, .. }
+            | Kind::FailedToExtractRunInfo { backtrace, .. }
             | Kind::IncompletePacket { backtrace, .. }
-            | Kind::InvalidFilepath { backtrace, .. }
+            | Kind::InvalidPath { backtrace, .. }
             | Kind::InvalidIndex { backtrace, .. }
             | Kind::KeyMissing { backtrace, .. }
             | Kind::MissingInfo { backtrace, .. }

src/core/orchestrator/docker.rs

@@ -9,6 +9,7 @@ use crate::{
 use bollard::{
     container::{Config, CreateContainerOptions, ListContainersOptions},
     models::{ContainerStateStatusEnum, HostConfig},
+    secret::{ContainerInspectResponse, ContainerSummary},
 };
 use chrono::DateTime;
 use futures_util::future::join_all;
@@ -165,12 +166,8 @@ impl LocalDockerOrchestrator {
         ))
     }
     #[expect(
-        clippy::cast_sign_loss,
         clippy::string_slice,
-        clippy::cast_precision_loss,
-        clippy::cast_possible_truncation,
         clippy::indexing_slicing,
-        clippy::too_many_lines,
         reason = r#"
         - Timestamp and memory should always have a value > 0
         - Container will always have a name with more than 1 character
@@ -210,79 +207,119 @@ impl LocalDockerOrchestrator {
         .await
         .into_iter()
         .filter_map(|result: Result<_>| {
-            let (container_name, container_summary, container_spec) = result.ok()?;
-            let terminated_timestamp =
-                DateTime::parse_from_rfc3339(container_spec.state.as_ref()?.finished_at.as_ref()?)
-                    .ok()?
-                    .timestamp();
-            Some((
-                container_name,
-                PodRunInfo {
-                    image: container_spec.config.as_ref()?.image.as_ref()?.clone(),
-                    created: container_summary.created? as u64,
-                    terminated: (terminated_timestamp > 0).then_some(terminated_timestamp as u64),
-                    env_vars: container_spec
-                        .config
-                        .as_ref()?
-                        .env
-                        .as_ref()?
-                        .iter()
-                        .filter_map(|x| {
-                            x.split_once('=')
-                                .map(|(key, value)| (key.to_owned(), value.to_owned()))
-                        })
-                        .collect(),
-                    command: [
-                        container_spec.config.as_ref()?.entrypoint.as_ref()?.clone(),
-                        container_spec.config.as_ref()?.cmd.as_ref()?.clone(),
-                    ]
-                    .concat(),
-                    status: match (
-                        container_spec.state.as_ref()?.status.as_ref()?,
-                        container_spec.state.as_ref()?.exit_code? as i16,
-                    ) {
-                        (ContainerStateStatusEnum::RUNNING, _) => PodStatus::Running,
-                        (
-                            ContainerStateStatusEnum::EXITED
-                            | ContainerStateStatusEnum::REMOVING
-                            | ContainerStateStatusEnum::DEAD,
-                            0,
-                        ) => PodStatus::Completed,
-                        (
-                            ContainerStateStatusEnum::EXITED
-                            | ContainerStateStatusEnum::REMOVING
-                            | ContainerStateStatusEnum::DEAD,
-                            code,
-                        ) => PodStatus::Failed(code),
-                        (_, code) => {
-                            todo!(
-                                "Unhandled container state: {}, exit code: {code}.",
-                                container_spec.state.as_ref()?.status.as_ref()?
-                            )
-                        }
-                    },
-                    mounts: container_spec
-                        .mounts
-                        .as_ref()?
-                        .iter()
-                        .map(|mount_point| {
-                            Some(format!(
-                                "{}:{}{}",
-                                mount_point.source.as_ref()?,
-                                mount_point.destination.as_ref()?,
-                                mount_point
-                                    .mode
-                                    .as_ref()
-                                    .map_or_else(String::new, |mode| format!(":{mode}"))
-                            ))
-                        })
-                        .collect::<Option<_>>()?,
-                    labels: container_spec.config.as_ref()?.labels.as_ref()?.clone(),
-                    cpu_limit: container_spec.host_config.as_ref()?.nano_cpus? as f32
-                        / 10_f32.powi(9), // ncpu, ucores=3, mcores=6, cores=9
-                    memory_limit: container_spec.host_config.as_ref()?.memory? as u64,
-                },
-            ))
+            let (container_name, container_summary, container_inspect_response) = result.ok()?;
+
+            Self::extract_run_info(&container_summary, &container_inspect_response)
+                .map(|run_info| (container_name.clone(), run_info))
         }))
     }
+
+    #[expect(
+        clippy::cast_sign_loss,
+        clippy::cast_precision_loss,
+        clippy::cast_possible_truncation,
+        reason = r#"
+        - Timestamp and memory should always have a value > 0
+        - Container will always have a name with more than 1 character
+        - No issue in core casting if between 0 - 3.40e38(f32:MAX)
+        - No issue in exit code casting if between -3.27e4(i16:MIN) - 3.27e4(i16:MAX)
+        - Containers will always have at least 1 name with at least 2 characters
+        - This functions requires a lot of boilerplate code to extract the run info
+        "#
+    )]
+    fn extract_run_info(
+        container_summary: &ContainerSummary,
+        container_inspect_response: &ContainerInspectResponse,
+    ) -> Option<PodRunInfo> {
+        let terminated_timestamp = DateTime::parse_from_rfc3339(
+            container_inspect_response
+                .state
+                .as_ref()?
+                .finished_at
+                .as_ref()?,
+        )
+        .ok()?
+        .timestamp() as u64;
+        Some(PodRunInfo {
+            image: container_inspect_response
+                .config
+                .as_ref()?
+                .image
+                .as_ref()?
+                .clone(),
+            created: container_summary.created? as u64,
+            terminated: (terminated_timestamp > 0).then_some(terminated_timestamp),
+            env_vars: container_inspect_response
+                .config
+                .as_ref()?
+                .env
+                .as_ref()?
+                .iter()
+                .filter_map(|x| {
+                    x.split_once('=')
+                        .map(|(key, value)| (key.to_owned(), value.to_owned()))
+                })
+                .collect(),
+            command: [
+                container_inspect_response
+                    .config
+                    .as_ref()?
+                    .entrypoint
+                    .as_ref()?
+                    .clone(),
+                container_inspect_response
+                    .config
+                    .as_ref()?
+                    .cmd
+                    .as_ref()?
+                    .clone(),
+            ]
+            .concat(),
+            status: match (
+                container_inspect_response.state.as_ref()?.status?,
+                container_inspect_response.state.as_ref()?.exit_code? as i16,
+            ) {
+                (ContainerStateStatusEnum::RUNNING | ContainerStateStatusEnum::RESTARTING, _) => {
+                    PodStatus::Running
+                }
+                (ContainerStateStatusEnum::EXITED, 0) => PodStatus::Completed,
+                (ContainerStateStatusEnum::EXITED | ContainerStateStatusEnum::DEAD, code) => {
+                    PodStatus::Failed(code)
+                }
+                (ContainerStateStatusEnum::CREATED, code) => {
+                    if container_inspect_response.state.as_ref()?.error.is_some() {
+                        PodStatus::Failed(code)
+                    } else {
+                        PodStatus::Running
+                    }
+                }
+                _ => PodStatus::Undefined,
+            },
+            mounts: container_inspect_response
+                .mounts
+                .as_ref()?
+                .iter()
+                .map(|mount_point| {
+                    Some(format!(
+                        "{}:{}{}",
+                        mount_point.source.as_ref()?,
+                        mount_point.destination.as_ref()?,
+                        mount_point
+                            .mode
+                            .as_ref()
+                            .map_or_else(String::new, |mode| format!(":{mode}"))
+                    ))
+                })
+                .collect::<Option<Vec<_>>>()?,
+            labels: container_inspect_response
+                .config
+                .as_ref()?
+                .labels
+                .as_ref()?
+                .clone(),
+            cpu_limit: container_inspect_response.host_config.as_ref()?.nano_cpus? as f32
+                / 10_f32.powi(9), // ncpu, ucores=3, mcores=6, cores=9
+            memory_limit: container_inspect_response.host_config.as_ref()?.memory? as u64,
+        })
+    }
 }

src/core/pipeline_runner.rs

@@ -23,7 +23,6 @@ use crate::{
 };
 use async_trait::async_trait;
 use names::{Generator, Name};
-use serde::{Deserialize, Serialize};
 use serde_yaml::Serializer;
 use snafu::{OptionExt as _, ResultExt as _};
 use std::{
@@ -39,18 +38,6 @@ use tokio::{
 static NODE_OUTPUT_KEY_EXPR: &str = "output";
 static FAILURE_KEY_EXP: &str = "failure";
 
-#[derive(Serialize, Deserialize, Clone, Debug)]
-enum NodeOutput {
-    Packet(String, HashMap<String, PathSet>),
-    ProcessingCompleted(String),
-}
-
-#[derive(Serialize, Deserialize, Clone, Debug)]
-struct ProcessingFailure {
-    node_id: String,
-    error: String,
-}
-
 /// Internal representation of a pipeline run, which should not be made public due to the fact that it contains
 #[derive(Debug)]
 struct PipelineRunInternal {
@@ -736,7 +723,7 @@ impl PodProcessor {
                     },
                 });
             }
-            PodStatus::Running | PodStatus::Unset => {
+            PodStatus::Running | PodStatus::Unset | PodStatus::Undefined => {
                 // This should not happen, but if it does, we will return an error
                 return Err(OrcaError {
                     kind: Kind::PodJobProcessingError {

src/core/store/filestore.rs

@@ -194,7 +194,7 @@ impl LocalFileStore {
                 Ok((
                     serde_yaml::from_str(
                         &fs::read_to_string(path.clone())
-                            .context(selector::InvalidFilepath { path })?,
+                            .context(selector::InvalidPath { path })?,
                     )?,
                     None,
                     hash.to_owned(),

src/uniffi/error.rs

@@ -36,6 +36,13 @@ pub(crate) enum Kind {
         dir: PathBuf,
         backtrace: Option<Backtrace>,
     },
+    #[snafu(display(
+        "Failed to extract run info from the container image file: {container_name}."
+    ))]
+    FailedToExtractRunInfo {
+        container_name: String,
+        backtrace: Option<Backtrace>,
+    },
     #[snafu(display(
         "Missing expected output file or dir with key {packet_key} at path {path:?} for pod job (hash: {pod_job_hash})."
     ))]
@@ -52,8 +59,16 @@ pub(crate) enum Kind {
         missing_keys: Vec<String>,
         backtrace: Option<Backtrace>,
     },
+    #[snafu(display(
+        "Fail to start pod with container_name: {container_name} with error: {reason}"
+    ))]
+    FailedToStartPod {
+        container_name: String,
+        reason: String,
+        backtrace: Option<Backtrace>,
+    },
     #[snafu(display("{source} ({path:?})."))]
-    InvalidFilepath {
+    InvalidPath {
         path: PathBuf,
         source: io::Error,
         backtrace: Option<Backtrace>,
@@ -153,4 +168,16 @@ impl OrcaError {
     pub fn is_purged_pod_run(&self) -> bool {
         matches!(&self.kind, Kind::MissingInfo { details, .. } if details.contains("pod run"))
     }
+    /// Returns `true` if the error was caused by an invalid file or directory path.
+    pub const fn is_failed_to_start_pod(&self) -> bool {
+        matches!(self.kind, Kind::FailedToStartPod { .. })
+    }
+    /// Returns container name if the
+    pub fn get_container_name(&self) -> Option<String> {
+        if let Kind::FailedToStartPod { container_name, .. } = &self.kind {
+            Some(container_name.clone())
+        } else {
+            None
+        }
+    }
 }

src/uniffi/model/pod.rs

@@ -267,6 +267,8 @@ pub struct PodResult {
     pub created: u64,
     /// Time in epoch when terminated in seconds.
     pub terminated: u64,
+    /// Logs about stdout and stderr, where stderr is append at the end
+    pub logs: String,
 }
 
 impl PodResult {
@@ -283,6 +285,7 @@ impl PodResult {
         created: u64,
         terminated: u64,
         namespace_lookup: &HashMap<String, PathBuf>,
+        logs: String,
     ) -> Result<Self> {
         let output_packet = pod_job
             .pod
@@ -339,6 +342,7 @@ impl PodResult {
             status,
             created,
             terminated,
+            logs,
         };
         Ok(Self {
             hash: hash_buffer(pod_result_no_hash.to_yaml()?),

src/uniffi/orchestrator/agent.rs

@@ -171,11 +171,11 @@ impl Agent {
             async |agent, inner_namespace_lookup, _, pod_job| {
                 let pod_run = agent
                     .orchestrator
-                    .start(&inner_namespace_lookup, &pod_job)
+                    .start(&pod_job, &inner_namespace_lookup)
                     .await?;
                 let pod_result = agent
                     .orchestrator
-                    .get_result(&inner_namespace_lookup, &pod_run)
+                    .get_result(&pod_run, &inner_namespace_lookup)
                     .await?;
                 agent.orchestrator.delete(&pod_run).await?;
                 Ok(pod_result)
@@ -191,6 +191,7 @@ impl Agent {
                                     PodStatus::Completed => "success",
                                     PodStatus::Running
                                     | PodStatus::Failed(_)
+                                    | PodStatus::Undefined
                                     | PodStatus::Unset => "failure",
                                 }
                                 .to_owned(),