ref(api): Do not return detailed pod error stats

etl-api/src/k8s_client.rs

@@ -10,7 +10,7 @@ use tracing::*;
 
 use kube::{
     Client,
-    api::{Api, DeleteParams, LogParams, Patch, PatchParams},
+    api::{Api, DeleteParams, Patch, PatchParams},
 };
 
 #[derive(Debug, Error)]
@@ -30,13 +30,6 @@ pub enum PodPhase {
     Unknown,
 }
 
-#[derive(Debug, Clone)]
-pub struct ContainerError {
-    pub exit_code: Option<i32>,
-    pub message: Option<String>,
-    pub reason: Option<String>,
-}
-
 impl From<&str> for PodPhase {
     fn from(value: &str) -> Self {
         match value {
@@ -88,17 +81,7 @@ pub trait K8sClient: Send + Sync {
 
     async fn get_pod_phase(&self, prefix: &str) -> Result<PodPhase, K8sError>;
 
-    async fn get_replicator_container_error(
-        &self,
-        prefix: &str,
-    ) -> Result<Option<ContainerError>, K8sError>;
-
-    async fn get_container_logs(
-        &self,
-        pod_name: &str,
-        container_name: &str,
-        previous: bool,
-    ) -> Result<String, K8sError>;
+    async fn has_replicator_container_error(&self, prefix: &str) -> Result<bool, K8sError>;
 
     async fn delete_pod(&self, prefix: &str) -> Result<(), K8sError>;
 }
@@ -534,11 +517,8 @@ impl K8sClient for HttpK8sClient {
         Ok(phase)
     }
 
-    async fn get_replicator_container_error(
-        &self,
-        prefix: &str,
-    ) -> Result<Option<ContainerError>, K8sError> {
-        info!("getting replicator error information");
+    async fn has_replicator_container_error(&self, prefix: &str) -> Result<bool, K8sError> {
+        info!("checking for replicator container error");
 
         let pod_name = format!("{prefix}-{STATEFUL_SET_NAME_SUFFIX}-0");
         let pod = match self.pods_api.get(&pod_name).await {
@@ -547,7 +527,7 @@ impl K8sClient for HttpK8sClient {
                 return match e {
                     kube::Error::Api(ref er) => {
                         if er.code == 404 {
-                            return Ok(None);
+                            return Ok(false);
                         }
                         Err(e.into())
                     }
@@ -569,32 +549,17 @@ impl K8sClient for HttpK8sClient {
         });
 
         let Some(container_status) = container_status else {
-            return Ok(None);
+            return Ok(false);
         };
 
-        // Check last terminated state.
-        //
-        // `last_state` is only set when there’s a previous termination, and remains empty if the
-        // container has never failed, so this is what we want, having access to the previous failure
+        // Check last terminated state for non-zero exit code
         if let Some(last_state) = &container_status.last_state {
             if let Some(terminated) = &last_state.terminated {
-                if terminated.exit_code != 0 {
-                    // Fetch logs from the previous container run
-                    let log_message = self
-                        .get_container_logs(&pod_name, &replicator_container_name, true)
-                        .await
-                        .ok();
-
-                    return Ok(Some(ContainerError {
-                        exit_code: Some(terminated.exit_code),
-                        message: log_message.or_else(|| terminated.message.clone()),
-                        reason: terminated.reason.clone(),
-                    }));
-                }
+                return Ok(terminated.exit_code != 0);
             }
         }
 
-        Ok(None)
+        Ok(false)
     }
 
     async fn delete_pod(&self, prefix: &str) -> Result<(), K8sError> {
@@ -617,23 +582,4 @@ impl K8sClient for HttpK8sClient {
 
         Ok(())
     }
-
-    async fn get_container_logs(
-        &self,
-        pod_name: &str,
-        container_name: &str,
-        previous: bool,
-    ) -> Result<String, K8sError> {
-        let log_params = LogParams {
-            container: Some(container_name.to_string()),
-            tail_lines: Some(50), // Get last 50 lines
-            timestamps: false,
-            previous, // Get logs from previous container instance or not
-            ..Default::default()
-        };
-
-        let logs = self.pods_api.logs(pod_name, &log_params).await?;
-
-        Ok(logs)
-    }
 }

etl-api/src/routes/pipelines.rs

@@ -359,11 +359,7 @@ pub enum PipelineStatus {
     Started,
     Stopping,
     Unknown,
-    Failed {
-        exit_code: Option<i32>,
-        message: Option<String>,
-        reason: Option<String>,
-    },
+    Failed,
 }
 
 #[utoipa::path(
@@ -710,30 +706,20 @@ pub async fn get_pipeline_status(
     // We load the pod phase.
     let pod_phase = k8s_client.get_pod_phase(&prefix).await?;
 
-    // We check the status of the replicator container.
+    // Check if there's a container error.
     //
-    // Note that this is dumping all the logs within the container, meaning that anything on stdout
-    // and stderr will be returned. In case we want to be more careful with what we return we can
-    // embed within the logs some special characters that will be used to determine which error message
-    // to return to the user.
-    let replicator_error = k8s_client.get_replicator_container_error(&prefix).await?;
-
-    let status = if let Some(replicator_error) = replicator_error {
-        PipelineStatus::Failed {
-            exit_code: replicator_error.exit_code,
-            message: replicator_error.message,
-            reason: replicator_error.reason,
-        }
+    // We are not exposing the error to not leak any internal information of the pod and because we
+    // will expose it through logs.
+    let has_container_error = k8s_client.has_replicator_container_error(&prefix).await?;
+
+    let status = if has_container_error {
+        PipelineStatus::Failed
     } else {
         match pod_phase {
             PodPhase::Pending => PipelineStatus::Starting,
             PodPhase::Running => PipelineStatus::Started,
             PodPhase::Succeeded => PipelineStatus::Stopped,
-            PodPhase::Failed => PipelineStatus::Failed {
-                exit_code: None,
-                message: None,
-                reason: None,
-            },
+            PodPhase::Failed => PipelineStatus::Failed,
             PodPhase::Unknown => PipelineStatus::Unknown,
         }
     };

etl-api/tests/common/k8s_client.rs

@@ -2,8 +2,7 @@ use std::collections::BTreeMap;
 
 use async_trait::async_trait;
 use etl_api::k8s_client::{
-    ContainerError, K8sClient, K8sError, PodPhase, TRUSTED_ROOT_CERT_CONFIG_MAP_NAME,
-    TRUSTED_ROOT_CERT_KEY_NAME,
+    K8sClient, K8sError, PodPhase, TRUSTED_ROOT_CERT_CONFIG_MAP_NAME, TRUSTED_ROOT_CERT_KEY_NAME,
 };
 use k8s_openapi::api::core::v1::ConfigMap;
 
@@ -80,20 +79,8 @@ impl K8sClient for MockK8sClient {
         Ok(PodPhase::Running)
     }
 
-    async fn get_replicator_container_error(
-        &self,
-        _prefix: &str,
-    ) -> Result<Option<ContainerError>, K8sError> {
-        Ok(None)
-    }
-
-    async fn get_container_logs(
-        &self,
-        _pod_name: &str,
-        _container_name: &str,
-        _previous: bool,
-    ) -> Result<String, K8sError> {
-        Ok(String::new())
+    async fn has_replicator_container_error(&self, _prefix: &str) -> Result<bool, K8sError> {
+        Ok(false)
     }
 
     async fn delete_pod(&self, _prefix: &str) -> Result<(), K8sError> {