fix: handle missing staging directory

Author: Devdutt Shenoi

src/banner.rs

@@ -104,7 +104,7 @@ async fn storage_info(config: &Parseable) {
         Staging Path:       \"{}\"",
         "Storage:".to_string().bold(),
         config.get_storage_mode_string(),
-        config.staging_dir().to_string_lossy(),
+        config.options.staging_dir().to_string_lossy(),
     );
 
     if let Some(path) = &config.options.hot_tier_storage_path {

src/cli.rs

@@ -17,7 +17,7 @@
  */
 
 use clap::Parser;
-use std::{env, path::PathBuf};
+use std::{env, fs, path::PathBuf};
 
 use url::Url;
 
@@ -386,6 +386,14 @@ impl Options {
         self.username == DEFAULT_USERNAME && self.password == DEFAULT_PASSWORD
     }
 
+    /// Path to staging directory, ensures that it exists or panics
+    pub fn staging_dir(&self) -> &PathBuf {
+        fs::create_dir_all(&self.local_staging_path)
+            .expect("Should be able to create dir if doesn't exist");
+
+        &self.local_staging_path
+    }
+
     /// TODO: refactor and document
     pub fn get_url(&self) -> Url {
         if self.ingestor_endpoint.is_empty() {

src/handlers/http/about.rs

@@ -66,7 +66,7 @@ pub async fn about() -> Json<Value> {
     let staging = if PARSEABLE.options.mode == Mode::Query {
         "".to_string()
     } else {
-        PARSEABLE.staging_dir().display().to_string()
+        PARSEABLE.options.staging_dir().display().to_string()
     };
     let grpc_port = PARSEABLE.options.grpc_port;
 

src/handlers/http/modal/mod.rs

@@ -342,7 +342,7 @@ impl IngestorMetadata {
         let bytes = Bytes::from(serde_json::to_vec(&json)?);
 
         let resource: IngestorMetadata = serde_json::from_value(json)?;
-        resource.put_on_disk(PARSEABLE.staging_dir())?;
+        resource.put_on_disk(PARSEABLE.options.staging_dir())?;
 
         PARSEABLE
             .storage

src/migration/mod.rs

@@ -325,7 +325,8 @@ pub fn to_bytes(any: &(impl ?Sized + Serialize)) -> Bytes {
 }
 
 pub fn get_staging_metadata(config: &Parseable) -> anyhow::Result<Option<serde_json::Value>> {
-    let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME).to_path(config.staging_dir());
+    let path =
+        RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME).to_path(config.options.staging_dir());
     let bytes = match std::fs::read(path) {
         Ok(bytes) => bytes,
         Err(err) => match err.kind() {
@@ -351,7 +352,7 @@ pub fn put_staging_metadata(
     config: &Parseable,
     metadata: &serde_json::Value,
 ) -> anyhow::Result<()> {
-    let path = config.staging_dir().join(".parseable.json");
+    let path = config.options.staging_dir().join(".parseable.json");
     let mut file = OpenOptions::new()
         .create(true)
         .truncate(true)

src/parseable/mod.rs

@@ -217,10 +217,6 @@ impl Parseable {
         self.storage.clone()
     }
 
-    pub fn staging_dir(&self) -> &PathBuf {
-        &self.options.local_staging_path
-    }
-
     pub fn hot_tier_dir(&self) -> &Option<PathBuf> {
         &self.options.hot_tier_storage_path
     }

src/storage/object_storage.rs

@@ -645,7 +645,7 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
     }
 
     async fn upload_files_from_staging(&self) -> Result<(), ObjectStorageError> {
-        if !Path::new(&PARSEABLE.staging_dir()).exists() {
+        if !Path::new(&PARSEABLE.options.staging_dir()).exists() {
             return Ok(());
         }
 

src/storage/store_metadata.rs

@@ -70,7 +70,7 @@ impl Default for StorageMetadata {
         Self {
             version: CURRENT_STORAGE_METADATA_VERSION.to_string(),
             mode: PARSEABLE.storage.name().to_owned(),
-            staging: PARSEABLE.staging_dir().to_path_buf(),
+            staging: PARSEABLE.options.staging_dir().to_path_buf(),
             storage: PARSEABLE.storage.get_endpoint(),
             deployment_id: uid::gen(),
             server_mode: PARSEABLE.options.mode,
@@ -134,8 +134,8 @@ pub async fn resolve_parseable_metadata(
             if metadata.server_mode== Mode::All && PARSEABLE.options.mode == Mode::Ingest {
                 Err("Starting Ingest Mode is not allowed, Since Query Server has not been started yet")
             } else {
-                create_dir_all(PARSEABLE.staging_dir())?;
-                metadata.staging = PARSEABLE.staging_dir().canonicalize()?;
+                create_dir_all(PARSEABLE.options.staging_dir())?;
+                metadata.staging = PARSEABLE.options.staging_dir().canonicalize()?;
                 // this flag is set to true so that metadata is copied to staging
                 overwrite_staging = true;
                 // overwrite remote in all and query mode
@@ -151,20 +151,20 @@ pub async fn resolve_parseable_metadata(
                     Mode::Query => {
                         overwrite_remote = true;
                         metadata.server_mode = PARSEABLE.options.mode;
-                        metadata.staging = PARSEABLE.staging_dir().to_path_buf();
+                        metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
                     },
                     Mode::Ingest => {
                         // if ingest server is started fetch the metadata from remote
                         // update the server mode for local metadata
                         metadata.server_mode = PARSEABLE.options.mode;
-                        metadata.staging = PARSEABLE.staging_dir().to_path_buf();
+                        metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
                       },
                 }
                 Ok(metadata)
             }
         }
         EnvChange::CreateBoth => {
-            create_dir_all(PARSEABLE.staging_dir())?;
+            create_dir_all(PARSEABLE.options.staging_dir())?;
             let metadata = StorageMetadata::default();
             // new metadata needs to be set
             // if mode is query or all then both staging and remote
@@ -237,7 +237,8 @@ pub enum EnvChange {
 }
 
 pub fn get_staging_metadata() -> io::Result<Option<StorageMetadata>> {
-    let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME).to_path(PARSEABLE.staging_dir());
+    let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME)
+        .to_path(PARSEABLE.options.staging_dir());
     let bytes = match fs::read(path) {
         Ok(bytes) => bytes,
         Err(err) => match err.kind() {
@@ -259,8 +260,11 @@ pub async fn put_remote_metadata(metadata: &StorageMetadata) -> Result<(), Objec
 pub fn put_staging_metadata(meta: &StorageMetadata) -> io::Result<()> {
     let mut staging_metadata = meta.clone();
     staging_metadata.server_mode = PARSEABLE.options.mode;
-    staging_metadata.staging = PARSEABLE.staging_dir().to_path_buf();
-    let path = PARSEABLE.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
+    staging_metadata.staging = PARSEABLE.options.staging_dir().to_path_buf();
+    let path = PARSEABLE
+        .options
+        .staging_dir()
+        .join(PARSEABLE_METADATA_FILE_NAME);
     let mut file = OpenOptions::new()
         .create(true)
         .truncate(true)