fix: use hash for distributed mode

Author: Eshanatnight

server/src/catalog.rs

@@ -16,17 +16,16 @@
  *
  */
 
-use std::sync::Arc;
-
-use chrono::{DateTime, Local, NaiveDateTime, NaiveTime, Utc};
-use relative_path::RelativePathBuf;
+use std::{io::ErrorKind, sync::Arc};
 
 use crate::{
     catalog::manifest::Manifest,
     query::PartialTimeFilter,
-    storage::{ObjectStorage, ObjectStorageError, MANIFEST_FILE},
-    utils::get_url,
+    storage::{object_storage::manifest_path, ObjectStorage, ObjectStorageError},
 };
+use chrono::{DateTime, Local, NaiveDateTime, NaiveTime, Utc};
+use relative_path::RelativePathBuf;
+use std::io::Error as IOError;
 
 use self::{column::Column, snapshot::ManifestItem};
 
@@ -117,13 +116,7 @@ pub async fn update_snapshot(
 
         let mut ch = false;
         for m in manifests.iter() {
-            let s = get_url();
-            let p = format!(
-                "{}.{}.{}",
-                s.domain().unwrap(),
-                s.port().unwrap_or_default(),
-                MANIFEST_FILE
-            );
+            let p = manifest_path("").to_string();
             if m.manifest_path.contains(&p) {
                 ch = true;
             }
@@ -147,7 +140,11 @@ pub async fn update_snapshot(
                         23 * 3600 + 59 * 60 + 59,
                         999_999_999,
                     )
-                    .unwrap(),
+                    .ok_or(IOError::new(
+                        ErrorKind::Other,
+                        "Failed to create upper bound for manifest",
+                    ))
+                    .map_err(ObjectStorageError::IoError)?,
                 )
                 .and_utc();
 
@@ -156,17 +153,11 @@ pub async fn update_snapshot(
                 ..Manifest::default()
             };
 
-            let addr = get_url();
-            let mainfest_file_name = format!(
-                "{}.{}.{}",
-                addr.domain().unwrap(),
-                addr.port().unwrap_or_default(),
-                MANIFEST_FILE
-            );
+            let mainfest_file_name = manifest_path("").to_string();
             let path =
                 partition_path(stream_name, lower_bound, upper_bound).join(&mainfest_file_name);
             storage
-                .put_object(&path, serde_json::to_vec(&manifest).unwrap().into())
+                .put_object(&path, serde_json::to_vec(&manifest)?.into())
                 .await?;
             let path = storage.absolute_url(&path);
             let new_snapshot_entriy = snapshot::ManifestItem {
@@ -195,13 +186,7 @@ pub async fn update_snapshot(
             ..Manifest::default()
         };
 
-        let addr = get_url();
-        let mainfest_file_name = format!(
-            "{}.{}.{}",
-            addr.domain().unwrap(),
-            addr.port().unwrap(),
-            MANIFEST_FILE
-        );
+        let mainfest_file_name = manifest_path("").to_string();
         let path = partition_path(stream_name, lower_bound, upper_bound).join(&mainfest_file_name);
         storage
             .put_object(&path, serde_json::to_vec(&manifest).unwrap().into())

server/src/migration.rs

@@ -244,7 +244,7 @@ async fn run_meta_file_migration(
                 // we can unwrap here because we know the file exists
                 let new_path = RelativePathBuf::from_iter([
                     PARSEABLE_ROOT_DIRECTORY,
-                    file.file_name().unwrap(),
+                    file.file_name().expect("should have a file name"),
                 ]);
                 object_store.put_object(&new_path, bytes).await?;
                 object_store.delete_object(&file).await?;
@@ -281,7 +281,7 @@ async fn run_stream_files_migration(
                     let new_path = RelativePathBuf::from_iter([
                         stream.as_str(),
                         STREAM_ROOT_DIRECTORY,
-                        path.file_name().unwrap(),
+                        path.file_name().expect("should have a file name"),
                     ]);
                     object_store.put_object(&new_path, bytes).await?;
                     object_store.delete_object(&path).await?;

server/src/option.rs

@@ -229,7 +229,7 @@ impl Mode {
             "Query" => Ok(Mode::Query),
             "Ingest" => Ok(Mode::Ingest),
             "All" => Ok(Mode::All),
-            x => Err(format!("Invalid mode: {}", x)),
+            x => Err(format!("Trying to Parse Invalid mode: {}", x)),
         }
     }
 }

server/src/static_schema.rs

@@ -12,7 +12,6 @@ pub struct StaticSchema {
 }
 
 #[derive(Serialize, Deserialize, Debug)]
-
 pub struct SchemaFields {
     name: String,
     data_type: String,

server/src/storage/object_storage.rs

@@ -27,7 +27,7 @@ use super::{
 
 use crate::handlers::http::modal::ingest_server::INGESTOR_META;
 use crate::option::Mode;
-use crate::utils::get_url;
+
 use crate::{
     alerts::Alerts,
     catalog::{self, manifest::Manifest, snapshot::Snapshot},
@@ -585,15 +585,17 @@ fn alert_json_path(stream_name: &str) -> RelativePathBuf {
 }
 
 #[inline(always)]
-fn manifest_path(prefix: &str) -> RelativePathBuf {
-    let addr = get_url();
-    let mainfest_file_name = format!(
-        "{}.{}.{}",
-        addr.domain().unwrap(),
-        addr.port().unwrap_or_default(),
-        MANIFEST_FILE
-    );
-    RelativePathBuf::from_iter([prefix, &mainfest_file_name])
+pub fn manifest_path(prefix: &str) -> RelativePathBuf {
+    if CONFIG.parseable.mode == Mode::Ingest {
+        let manifest_file_name = format!(
+            "ingestor.{}.{}",
+            INGESTOR_META.get_ingestor_id(),
+            MANIFEST_FILE
+        );
+        RelativePathBuf::from_iter([prefix, &manifest_file_name])
+    } else {
+        RelativePathBuf::from_iter([MANIFEST_FILE])
+    }
 }
 
 #[inline(always)]

server/src/storage/staging.rs

@@ -31,7 +31,10 @@ use crate::{
     metrics,
     option::{Mode, CONFIG},
     storage::OBJECT_STORE_DATA_GRANULARITY,
-    utils::{self, arrow::merged_reader::MergedReverseRecordReader, get_ingestor_id, get_url},
+    utils::{
+        self, arrow::merged_reader::MergedReverseRecordReader, get_ingestor_id, get_url,
+        hostname_unchecked,
+    },
 };
 use arrow_schema::{ArrowError, Schema};
 use chrono::{NaiveDateTime, Timelike, Utc};
@@ -64,11 +67,12 @@ impl StorageDir {
             + &utils::hour_to_prefix(time.hour())
             + &utils::minute_to_prefix(time.minute(), OBJECT_STORE_DATA_GRANULARITY).unwrap();
         let local_uri = str::replace(&uri, "/", ".");
+        let hostname = hostname_unchecked();
         if CONFIG.parseable.mode == Mode::Ingest {
             let id = INGESTOR_META.get_ingestor_id();
-            format!("{local_uri}{id}.{extention}")
+            format!("{local_uri}{hostname}{id}.{extention}")
         } else {
-            format!("{local_uri}.{extention}")
+            format!("{local_uri}{hostname}.{extention}")
         }
     }
 
@@ -163,7 +167,7 @@ impl StorageDir {
     fn arrow_path_to_parquet(path: &Path) -> PathBuf {
         let filename = path.file_name().unwrap().to_str().unwrap();
         let (_, filename) = filename.split_once('.').unwrap();
-        let filename = filename.rsplit_once('.').unwrap();
+        let filename = filename.rsplit_once('.').expect("contains the delim `.`");
         let filename = format!("{}.{}", filename.0, filename.1);
 
         /*
@@ -316,7 +320,7 @@ pub fn get_ingestor_info() -> anyhow::Result<IngestorMetadata> {
     let store = CONFIG.storage().get_object_store();
     let url = get_url();
     let out = IngestorMetadata::new(
-        url.port().unwrap().to_string(), // here port should be defined
+        url.port().expect("here port should be defined").to_string(),
         url.to_string(),
         DEFAULT_VERSION.to_string(),
         store.get_bucket_name(),

server/src/sync.rs

@@ -28,7 +28,7 @@ use std::time::Duration;
 use crate::option::CONFIG;
 use crate::{storage, STORAGE_UPLOAD_INTERVAL};
 
-pub(crate) fn object_store_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<()>) {
+pub fn object_store_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<()>) {
     let (outbox_tx, outbox_rx) = oneshot::channel::<()>();
     let (inbox_tx, inbox_rx) = oneshot::channel::<()>();
     let mut inbox_rx = AssertUnwindSafe(inbox_rx);
@@ -70,7 +70,7 @@ pub(crate) fn object_store_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, one
     (handle, outbox_rx, inbox_tx)
 }
 
-pub(crate) fn run_local_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<()>) {
+pub fn run_local_sync() -> (JoinHandle<()>, oneshot::Receiver<()>, oneshot::Sender<()>) {
     let (outbox_tx, outbox_rx) = oneshot::channel::<()>();
     let (inbox_tx, inbox_rx) = oneshot::channel::<()>();
     let mut inbox_rx = AssertUnwindSafe(inbox_rx);

server/src/utils.rs

@@ -36,16 +36,11 @@ pub fn hostname() -> Option<String> {
         .ok()
         .and_then(|hostname| hostname.into_string().ok())
 }
-#[allow(dead_code)]
+
 pub fn hostname_unchecked() -> String {
     hostname::get().unwrap().into_string().unwrap()
 }
 
-#[allow(dead_code)]
-pub fn capitalize_ascii(s: &str) -> String {
-    s[0..1].to_uppercase() + &s[1..]
-}
-
 /// Convert minutes to a slot range
 /// e.g. given minute = 15 and OBJECT_STORE_DATA_GRANULARITY = 10 returns "10-19"
 pub fn minute_to_slot(minute: u32, data_granularity: u32) -> Option<String> {
@@ -235,7 +230,7 @@ pub fn get_url() -> Url {
             CONFIG.parseable.address
         )
         .parse::<Url>() // if the value was improperly set, this will panic before hand
-        .unwrap();
+        .expect("Valid URL");
     }
     let addr_from_env = CONFIG
         .parseable
@@ -260,7 +255,9 @@ pub fn get_url() -> Url {
         let var_port = port[1..].to_string();
         port = get_from_env(&var_port);
     }
-    format!("{}:{}", hostname, port).parse::<Url>().unwrap()
+    format!("{}:{}", hostname, port)
+        .parse::<Url>()
+        .expect("Valid URL")
 }
 
 /// util fuction to fetch value from an env var