update the root path for the metadata files

only the .parseable.json and the ingest metadata files

Author: Eshanatnight

server/src/handlers/http/modal/ingest_server.rs

@@ -27,8 +27,9 @@ use crate::metrics;
 use crate::rbac;
 use crate::rbac::role::Action;
 use crate::storage;
+use crate::storage::object_storage::ingester_metadata_path;
+use crate::storage::object_storage::parseable_json_path;
 use crate::storage::ObjectStorageError;
-use crate::storage::PARSEABLE_METADATA_FILE_NAME;
 use crate::sync;
 
 use super::server::Server;
@@ -180,13 +181,8 @@ impl IngestServer {
     async fn set_ingester_metadata(&self) -> anyhow::Result<()> {
         let store = CONFIG.storage().get_object_store();
 
-        // remove ip adn go with the domain name
         let sock = Server::get_server_address();
-        let path = RelativePathBuf::from(format!(
-            "ingester.{}.{}.json",
-            sock.ip(), // this might be wrong
-            sock.port()
-        ));
+        let path = ingester_metadata_path(sock.ip().to_string(), sock.port().to_string());
 
         if store.get_object(&path).await.is_ok() {
             println!("Ingester metadata already exists");
@@ -228,7 +224,7 @@ impl IngestServer {
         // i.e the querier will create the `.parseable.json` file
 
         let store = CONFIG.storage().get_object_store();
-        let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME);
+        let path = parseable_json_path();
 
         match store.get_object(&path).await {
             Ok(_) => Ok(()),

server/src/migration.rs

@@ -30,8 +30,7 @@ use serde::Serialize;
 use crate::{
     option::Config,
     storage::{
-        object_storage::stream_json_path, ObjectStorage, ObjectStorageError,
-        PARSEABLE_METADATA_FILE_NAME, SCHEMA_FILE_NAME,
+        object_storage::{parseable_json_path, stream_json_path}, ObjectStorage, ObjectStorageError,SCHEMA_FILE_NAME,
     },
 };
 
@@ -153,7 +152,8 @@ fn to_bytes(any: &(impl ?Sized + Serialize)) -> Bytes {
 }
 
 pub fn get_staging_metadata(config: &Config) -> anyhow::Result<Option<serde_json::Value>> {
-    let path = config.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
+    let path = parseable_json_path().to_path(config.staging_dir());
+
     let bytes = match std::fs::read(path) {
         Ok(bytes) => bytes,
         Err(err) => match err.kind() {
@@ -162,13 +162,14 @@ pub fn get_staging_metadata(config: &Config) -> anyhow::Result<Option<serde_json
         },
     };
     let meta: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+
     Ok(Some(meta))
 }
 
 async fn get_storage_metadata(
     storage: &dyn ObjectStorage,
 ) -> anyhow::Result<Option<serde_json::Value>> {
-    let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME);
+    let path = parseable_json_path();
     match storage.get_object(&path).await {
         Ok(bytes) => Ok(Some(
             serde_json::from_slice(&bytes).expect("parseable config is valid json"),
@@ -187,13 +188,14 @@ pub async fn put_remote_metadata(
     storage: &dyn ObjectStorage,
     metadata: &serde_json::Value,
 ) -> anyhow::Result<()> {
-    let path = RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME);
+    let path = parseable_json_path();
     let metadata = serde_json::to_vec(metadata)?.into();
     Ok(storage.put_object(&path, metadata).await?)
 }
 
 pub fn put_staging_metadata(config: &Config, metadata: &serde_json::Value) -> anyhow::Result<()> {
-    let path = config.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
+    let path = parseable_json_path().to_path(config.staging_dir());
+    //config.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
     let mut file = OpenOptions::new()
         .create(true)
         .truncate(true)

server/src/storage.rs

@@ -43,6 +43,7 @@ pub use self::staging::StorageDir;
 // metadata file names in a Stream prefix
 pub const STREAM_METADATA_FILE_NAME: &str = ".stream.json";
 pub const PARSEABLE_METADATA_FILE_NAME: &str = ".parseable.json";
+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";

server/src/storage/localfs.rs

@@ -35,7 +35,8 @@ use crate::metrics::storage::{localfs::REQUEST_RESPONSE_TIME, StorageMetrics};
 use crate::option::validation;
 
 use super::{
-    LogStream, ObjectStorage, ObjectStorageError, ObjectStorageProvider, STREAM_METADATA_FILE_NAME,
+    LogStream, ObjectStorage, ObjectStorageError, ObjectStorageProvider, PARSEABLE_ROOT_DIRECTORY,
+    STREAM_METADATA_FILE_NAME,
 };
 
 #[derive(Debug, Clone, clap::Args)]
@@ -194,7 +195,7 @@ impl ObjectStorage for LocalFS {
     }
 
     async fn list_streams(&self) -> Result<Vec<LogStream>, ObjectStorageError> {
-        let ignore_dir = &["lost+found"];
+        let ignore_dir = &["lost+found", PARSEABLE_ROOT_DIRECTORY];
         let directories = ReadDirStream::new(fs::read_dir(&self.root).await?);
         let entries: Vec<DirEntry> = directories.try_collect().await?;
         let entries = entries

server/src/storage/object_storage.rs

@@ -21,8 +21,8 @@ use super::{
     ObjectStoreFormat, Permisssion, StorageDir, StorageMetadata,
 };
 use super::{
-    ALERT_FILE_NAME, MANIFEST_FILE, PARSEABLE_METADATA_FILE_NAME, SCHEMA_FILE_NAME,
-    STREAM_METADATA_FILE_NAME,
+    ALERT_FILE_NAME, MANIFEST_FILE, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY,
+    SCHEMA_FILE_NAME, STREAM_METADATA_FILE_NAME,
 };
 
 use crate::option::Mode;
@@ -505,9 +505,10 @@ pub fn stream_json_path(stream_name: &str) -> RelativePathBuf {
     }
 }
 
+/// path will be ".parseable/.parsable.json"
 #[inline(always)]
-fn parseable_json_path() -> RelativePathBuf {
-    RelativePathBuf::from(PARSEABLE_METADATA_FILE_NAME)
+pub fn parseable_json_path() -> RelativePathBuf {
+    RelativePathBuf::from_iter([PARSEABLE_ROOT_DIRECTORY, PARSEABLE_METADATA_FILE_NAME])
 }
 
 #[inline(always)]
@@ -521,3 +522,8 @@ fn manifest_path(prefix: &str) -> RelativePathBuf {
     let mainfest_file_name = format!("{}.{}.{}", addr.0, addr.1, MANIFEST_FILE);
     RelativePathBuf::from_iter([prefix, &mainfest_file_name])
 }
+
+#[inline(always)]
+pub fn ingester_metadata_path(ip: String, port: String) -> RelativePathBuf {
+    RelativePathBuf::from_iter([PARSEABLE_ROOT_DIRECTORY, &format!("ingester.{}.{}.json", ip, port)])
+}

server/src/storage/s3.rs

@@ -39,7 +39,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use crate::metrics::storage::{s3::REQUEST_RESPONSE_TIME, StorageMetrics};
-use crate::storage::{LogStream, ObjectStorage, ObjectStorageError};
+use crate::storage::{LogStream, ObjectStorage, ObjectStorageError, PARSEABLE_ROOT_DIRECTORY};
 
 use super::metrics_layer::MetricLayer;
 use super::{ObjectStorageProvider, PARSEABLE_METADATA_FILE_NAME, STREAM_METADATA_FILE_NAME};
@@ -297,11 +297,13 @@ impl S3 {
         let common_prefixes = resp.common_prefixes;
 
         // return prefixes at the root level
-        let dirs: Vec<_> = common_prefixes
+        let mut dirs: Vec<_> = common_prefixes
             .iter()
             .filter_map(|path| path.parts().next())
             .map(|name| name.as_ref().to_string())
             .collect();
+        // filter out the root directory
+        dirs.retain(|x| x != PARSEABLE_ROOT_DIRECTORY);
 
         let stream_json_check = FuturesUnordered::new();
 

server/src/storage/store_metadata.rs

@@ -33,7 +33,7 @@ use crate::{
     utils::uid,
 };
 
-use super::PARSEABLE_METADATA_FILE_NAME;
+use super::{object_storage::parseable_json_path, PARSEABLE_METADATA_FILE_NAME};
 
 // Expose some static variables for internal usage
 pub static STORAGE_METADATA: OnceCell<StaticStorageMetadata> = OnceCell::new();
@@ -237,7 +237,7 @@ fn standalone_when_distributed(remote_server_mode: Mode) -> Result<(), MetadataE
 }
 
 pub fn get_staging_metadata() -> io::Result<Option<StorageMetadata>> {
-    let path = CONFIG.staging_dir().join(PARSEABLE_METADATA_FILE_NAME);
+    let path = parseable_json_path().to_path(CONFIG.staging_dir());
     let bytes = match fs::read(path) {
         Ok(bytes) => bytes,
         Err(err) => match err.kind() {