feat(hdfs): Print more logs on fs failure operations

Author: zuston

src/error.rs

@@ -85,6 +85,9 @@ pub enum WorkerError {
 
     #[error("urpc stream message type not found")]
     STREAM_MESSAGE_TYPE_NOT_FOUND,
+
+    #[error("{0}. error: {1}")]
+    HDFS_IO_ERROR(String, anyhow::Error),
 }
 
 impl From<AcquireError> for WorkerError {
@@ -114,12 +117,21 @@ impl From<FromUtf8Error> for WorkerError {
 #[cfg(test)]
 mod tests {
 
-    use anyhow::Result;
+    use crate::error::WorkerError::HDFS_IO_ERROR;
+    use anyhow::{Error, Result};
 
     #[test]
     pub fn error_test() -> Result<()> {
         // bail macro means it will return directly.
         // bail!(WorkerError::APP_PURGE_EVENT_SEND_ERROR("error_test_app_id".into(), None));
         Ok(())
     }
+
+    #[test]
+    pub fn hdfs_io_test() -> Result<()> {
+        let e = Error::from(std::io::Error::new(std::io::ErrorKind::Other, "oh no!"));
+        let raw = format!("{}", HDFS_IO_ERROR("".to_string(), e));
+        assert_eq!(". error: oh no!", raw);
+        Ok(())
+    }
 }

src/store/hdfs.rs

@@ -184,15 +184,39 @@ impl HdfsStore {
                     let parent_path_str = format!("{}/", parent_dir.to_str().unwrap());
                     debug!("creating dir: {}", parent_path_str.as_str());
 
-                    &filesystem.create_dir(parent_path_str.as_str()).await?;
+                    &filesystem
+                        .create_dir(parent_path_str.as_str())
+                        .await
+                        .map_err(|e| {
+                            error!("Errors on creating dir of {}", parent_path_str.as_str());
+                            e
+                        })?;
 
                     let data_file_complete_path = format!("{}_{}.data", &data_file_path_prefix, 0);
                     let index_file_complete_path =
                         format!("{}_{}.index", &index_file_path_prefix, 0);
 
                     // setup the file
-                    &filesystem.touch(&data_file_complete_path).await?;
-                    &filesystem.touch(&index_file_complete_path).await?;
+                    &filesystem
+                        .touch(&data_file_complete_path)
+                        .await
+                        .map_err(|e| {
+                            error!(
+                                "Errors on touching file of {}",
+                                data_file_complete_path.as_str()
+                            );
+                            e
+                        })?;
+                    &filesystem
+                        .touch(&index_file_complete_path)
+                        .await
+                        .map_err(|e| {
+                            error!(
+                                "Errors on touching file of {}",
+                                index_file_complete_path.as_str()
+                            );
+                            e
+                        })?;
 
                     self.partition_cached_meta
                         .insert(data_file_path_prefix.to_owned(), Default::default());
@@ -290,7 +314,11 @@ impl HdfsStore {
                 "hdfs writing [data] with {} bytes. path: {}",
                 data_len, &data_file_path
             ))
-            .await?;
+            .await
+            .map_err(|e| {
+                error!("Errors on appending data into path: {}", &data_file_path);
+                e
+            })?;
         let index_bytes = index_bytes_holder.freeze();
         let index_len = index_bytes.len();
         filesystem
@@ -299,7 +327,11 @@ impl HdfsStore {
                 "hdfs writing [index] with {} bytes. path: {}",
                 index_len, &index_file_path
             ))
-            .await?;
+            .await
+            .map_err(|e| {
+                error!("Errors on appending index into path: {}", &index_file_path);
+                e
+            })?;
         Ok(())
     }
 }