updated stream check for local file storage

Author: parmesant

src/handlers/http/correlation.rs

@@ -21,7 +21,7 @@ use bytes::Bytes;
 use relative_path::RelativePathBuf;
 
 use crate::{
-    option::CONFIG, storage::CORRELATION_DIRECTORY, utils::actix::extract_session_key_from_req,
+    option::CONFIG, storage::CORRELATION_ROOT_DIRECTORY, utils::actix::extract_session_key_from_req,
 };
 
 use crate::correlation::{
@@ -125,7 +125,7 @@ pub async fn delete(req: HttpRequest) -> Result<impl Responder, CorrelationError
 
     // Delete from disk
     let store = CONFIG.storage().get_object_store();
-    let path = RelativePathBuf::from_iter([CORRELATION_DIRECTORY, &correlation.id.to_string()]);
+    let path = RelativePathBuf::from_iter([CORRELATION_ROOT_DIRECTORY, &correlation.id.to_string()]);
     store.delete_object(&path).await?;
 
     // Delete from memory

src/storage/localfs.rs

@@ -39,8 +39,7 @@ use crate::{
 };
 
 use super::{
-    LogStream, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY,
-    SCHEMA_FILE_NAME, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
+    LogStream, ObjectStorage, ObjectStorageError, ObjectStorageProvider, CORRELATION_ROOT_DIRECTORY, PARSEABLE_ROOT_DIRECTORY, SCHEMA_FILE_NAME, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY
 };
 
 #[derive(Debug, Clone, clap::Args)]
@@ -295,7 +294,7 @@ impl ObjectStorage for LocalFS {
     }
 
     async fn list_streams(&self) -> Result<Vec<LogStream>, ObjectStorageError> {
-        let ignore_dir = &["lost+found", PARSEABLE_ROOT_DIRECTORY, USERS_ROOT_DIR];
+        let ignore_dir = &["lost+found", PARSEABLE_ROOT_DIRECTORY, USERS_ROOT_DIR, CORRELATION_ROOT_DIRECTORY];
         let directories = ReadDirStream::new(fs::read_dir(&self.root).await?);
         let entries: Vec<DirEntry> = directories.try_collect().await?;
         let entries = entries
@@ -315,7 +314,7 @@ impl ObjectStorage for LocalFS {
     }
 
     async fn list_old_streams(&self) -> Result<Vec<LogStream>, ObjectStorageError> {
-        let ignore_dir = &["lost+found", PARSEABLE_ROOT_DIRECTORY];
+        let ignore_dir = &["lost+found", PARSEABLE_ROOT_DIRECTORY, CORRELATION_ROOT_DIRECTORY];
         let directories = ReadDirStream::new(fs::read_dir(&self.root).await?);
         let entries: Vec<DirEntry> = directories.try_collect().await?;
         let entries = entries

src/storage/mod.rs

@@ -51,7 +51,7 @@ pub const PARSEABLE_ROOT_DIRECTORY: &str = ".parseable";
 pub const SCHEMA_FILE_NAME: &str = ".schema";
 pub const ALERT_FILE_NAME: &str = ".alert.json";
 pub const MANIFEST_FILE: &str = "manifest.json";
-pub const CORRELATION_DIRECTORY: &str = ".correlations";
+pub const CORRELATION_ROOT_DIRECTORY: &str = ".correlations";
 
 /// local sync interval to move data.records to /tmp dir of that stream.
 /// 60 sec is a reasonable value.

src/storage/object_storage.rs

@@ -21,7 +21,7 @@ use super::{
     ObjectStoreFormat, Permisssion, StorageDir, StorageMetadata,
 };
 use super::{
-    ALERT_FILE_NAME, CORRELATION_DIRECTORY, MANIFEST_FILE, PARSEABLE_METADATA_FILE_NAME,
+    ALERT_FILE_NAME, CORRELATION_ROOT_DIRECTORY, MANIFEST_FILE, PARSEABLE_METADATA_FILE_NAME,
     PARSEABLE_ROOT_DIRECTORY, SCHEMA_FILE_NAME, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
 };
 
@@ -631,15 +631,15 @@ pub trait ObjectStorage: Send + Sync + 'static {
         correlation: &CorrelationConfig,
     ) -> Result<(), ObjectStorageError> {
         let path = RelativePathBuf::from_iter([
-            CORRELATION_DIRECTORY,
+            CORRELATION_ROOT_DIRECTORY,
             &format!("{}.json", correlation.id),
         ]);
         self.put_object(&path, to_bytes(correlation)).await?;
         Ok(())
     }
 
     async fn get_correlations(&self) -> Result<Vec<Bytes>, CorrelationError> {
-        let correlation_path = RelativePathBuf::from_iter([CORRELATION_DIRECTORY]);
+        let correlation_path = RelativePathBuf::from_iter([CORRELATION_ROOT_DIRECTORY]);
         let correlation_bytes = self
             .get_objects(
                 Some(&correlation_path),