update files in distributed mode to use hash

Author: Eshanatnight

Cargo.lock

@@ -105,6 +105,7 @@ hashlru = { version = "0.11.0", features = ["serde"] }
 path-clean = "1.0.1"
 prost = "0.12.3"
 prometheus-parse = "0.2.5"
+sha2 = "0.10.8"
 
 [build-dependencies]
 cargo_toml = "0.15"

server/Cargo.toml

@@ -25,7 +25,7 @@ use crate::{
     catalog::manifest::Manifest,
     query::PartialTimeFilter,
     storage::{ObjectStorage, ObjectStorageError, MANIFEST_FILE},
-    utils::get_address,
+    utils::get_url,
 };
 
 use self::{column::Column, snapshot::ManifestItem};
@@ -117,7 +117,7 @@ pub async fn update_snapshot(
 
         let mut ch = false;
         for m in manifests.iter() {
-            let s = get_address();
+            let s = get_url();
             let p = format!(
                 "{}.{}.{}",
                 s.domain().unwrap(),
@@ -156,7 +156,7 @@ pub async fn update_snapshot(
                 ..Manifest::default()
             };
 
-            let addr = get_address();
+            let addr = get_url();
             let mainfest_file_name = format!(
                 "{}.{}.{}",
                 addr.domain().unwrap(),
@@ -195,7 +195,7 @@ pub async fn update_snapshot(
             ..Manifest::default()
         };
 
-        let addr = get_address();
+        let addr = get_url();
         let mainfest_file_name = format!(
             "{}.{}.{}",
             addr.domain().unwrap(),

server/src/catalog.rs

@@ -372,14 +372,27 @@ pub async fn remove_ingestor(req: HttpRequest) -> Result<impl Responder, PostErr
     if check_liveness(&domain_name).await {
         return Err(PostError::Invalid(anyhow::anyhow!("Node Online")));
     }
-
-    let url = Url::parse(&domain_name).unwrap();
-    let ingestor_meta_filename = ingestor_metadata_path(
-        url.host_str().unwrap().to_owned(),
-        url.port().unwrap().to_string(),
-    )
-    .to_string();
     let object_store = CONFIG.storage().get_object_store();
+
+    let ingestor_metadatas = object_store
+        .get_objects(
+            Some(&RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY)),
+            Box::new(|file_name| file_name.starts_with("ingestor")),
+        )
+        .await?;
+
+    let ingestor_metadata = ingestor_metadatas
+        .iter()
+        .map(|elem| serde_json::from_slice::<IngestorMetadata>(elem).unwrap_or_default())
+        .collect_vec();
+
+    let ingestor_metadata = ingestor_metadata
+        .iter()
+        .filter(|elem| elem.domain_name == domain_name)
+        .collect_vec();
+
+    let ingestor_meta_filename =
+        ingestor_metadata_path(Some(&ingestor_metadata[0].ingestor_id)).to_string();
     let msg = match object_store
         .try_delete_ingestor_meta(ingestor_meta_filename)
         .await

server/src/handlers/http/cluster/mod.rs

@@ -28,6 +28,7 @@ use crate::rbac::role::Action;
 use crate::storage;
 use crate::storage::object_storage::ingestor_metadata_path;
 use crate::storage::object_storage::parseable_json_path;
+use crate::storage::staging;
 use crate::storage::ObjectStorageError;
 use crate::sync;
 
@@ -36,24 +37,25 @@ use super::ssl_acceptor::get_ssl_acceptor;
 use super::IngestorMetadata;
 use super::OpenIdClient;
 use super::ParseableServer;
-use super::DEFAULT_VERSION;
 
-use crate::utils::get_address;
 use actix_web::body::MessageBody;
 use actix_web::Scope;
 use actix_web::{web, App, HttpServer};
 use actix_web_prometheus::PrometheusMetrics;
 use async_trait::async_trait;
 use base64::Engine;
 use itertools::Itertools;
+use once_cell::sync::Lazy;
 use relative_path::RelativePathBuf;
-use url::Url;
 
 use crate::{
     handlers::http::{base_path, cross_origin_config},
     option::CONFIG,
 };
 
+pub static INGESTOR_META: Lazy<IngestorMetadata> =
+    Lazy::new(|| staging::get_ingestor_info().expect("dir is readable and writeable"));
+
 #[derive(Default)]
 pub struct IngestServer;
 
@@ -102,6 +104,7 @@ impl ParseableServer for IngestServer {
     /// implement the init method will just invoke the initialize method
     async fn init(&self) -> anyhow::Result<()> {
         self.validate()?;
+
         // check for querier state. Is it there, or was it there in the past
         self.check_querier_state().await?;
         // to get the .parseable.json file in staging
@@ -181,46 +184,23 @@ impl IngestServer {
     async fn set_ingestor_metadata(&self) -> anyhow::Result<()> {
         let store = CONFIG.storage().get_object_store();
 
-        let sock = get_address();
-        let path = ingestor_metadata_path(
-            sock.domain().unwrap().to_string(),
-            sock.port().unwrap_or_default().to_string(),
-        );
+        // find the meta file in staging if not generate new metadata
+        let resource = INGESTOR_META.clone();
+        // use the id that was generated/found in the staging and
+        // generate the path for the object store
+        let path = ingestor_metadata_path(None);
 
         if store.get_object(&path).await.is_ok() {
-            println!("ingestor metadata already exists");
+            log::info!("ingestor metadata already exists");
             return Ok(());
         };
 
-        let scheme = CONFIG.parseable.get_scheme();
-        let resource = IngestorMetadata::new(
-            sock.port().unwrap_or_default().to_string(),
-            CONFIG
-                .parseable
-                .domain_address
-                .clone()
-                .unwrap_or_else(|| {
-                    Url::parse(&format!(
-                        "{}://{}:{}",
-                        scheme,
-                        sock.domain().unwrap(),
-                        sock.port().unwrap_or_default()
-                    ))
-                    .unwrap()
-                })
-                .to_string(),
-            DEFAULT_VERSION.to_string(),
-            store.get_bucket_name(),
-            &CONFIG.parseable.username,
-            &CONFIG.parseable.password,
-        );
-
         let resource = serde_json::to_string(&resource)
             .unwrap()
             .try_into_bytes()
             .unwrap();
 
-        store.put_object(&path, resource).await?;
+        store.put_object(&path, resource.clone()).await?;
 
         Ok(())
     }
@@ -279,6 +259,7 @@ impl IngestServer {
     }
 
     async fn initialize(&self) -> anyhow::Result<()> {
+        // ! Undefined and Untested behaviour
         if let Some(cache_manager) = LocalCacheManager::global() {
             cache_manager
                 .validate(CONFIG.parseable.local_cache_size)

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

@@ -61,6 +61,7 @@ pub struct IngestorMetadata {
     pub domain_name: String,
     pub bucket_name: String,
     pub token: String,
+    pub ingestor_id: String,
 }
 
 impl IngestorMetadata {
@@ -71,6 +72,7 @@ impl IngestorMetadata {
         bucket_name: String,
         username: &str,
         password: &str,
+        ingestor_id: String,
     ) -> Self {
         let token = base64::prelude::BASE64_STANDARD.encode(format!("{}:{}", username, password));
 
@@ -82,8 +84,13 @@ impl IngestorMetadata {
             version,
             bucket_name,
             token,
+            ingestor_id,
         }
     }
+
+    pub fn get_ingestor_id(&self) -> String {
+        self.ingestor_id.clone()
+    }
 }
 
 #[cfg(test)]
@@ -102,9 +109,10 @@ mod test {
             "somebucket".to_string(),
             "admin",
             "admin",
+            "ingestor_id".to_string(),
         );
 
-        let rhs = serde_json::from_slice::<IngestorMetadata>(br#"{"version":"v3","port":"8000","domain_name":"https://localhost:8000","bucket_name":"somebucket","token":"Basic YWRtaW46YWRtaW4="}"#).unwrap();
+        let rhs = serde_json::from_slice::<IngestorMetadata>(br#"{"version":"v3","port":"8000","domain_name":"https://localhost:8000","bucket_name":"somebucket","token":"Basic YWRtaW46YWRtaW4=", "ingestor_id": "ingestor_id"}"#).unwrap();
 
         assert_eq!(rhs, lhs);
     }
@@ -118,13 +126,14 @@ mod test {
             "somebucket".to_string(),
             "admin",
             "admin",
+            "ingestor_id".to_string(),
         );
 
         let lhs = serde_json::to_string(&im)
             .unwrap()
             .try_into_bytes()
             .unwrap();
-        let rhs = br#"{"version":"v3","port":"8000","domain_name":"https://localhost:8000","bucket_name":"somebucket","token":"Basic YWRtaW46YWRtaW4="}"#
+        let rhs = br#"{"version":"v3","port":"8000","domain_name":"https://localhost:8000","bucket_name":"somebucket","token":"Basic YWRtaW46YWRtaW4=","ingestor_id":"ingestor_id"}"#
                 .try_into_bytes()
                 .unwrap();
 

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

@@ -125,12 +125,12 @@ impl QueryServer {
                     .service(Server::get_llm_webscope())
                     .service(Server::get_oauth_webscope(oidc_client))
                     .service(Server::get_user_role_webscope())
-                    .service(Self::get_cluster_info_web_scope()),
+                    .service(Self::get_cluster_web_scope()),
             )
             .service(Server::get_generated());
     }
 
-    fn get_cluster_info_web_scope() -> actix_web::Scope {
+    fn get_cluster_web_scope() -> actix_web::Scope {
         web::scope("/cluster")
             .service(
                 // GET "/cluster/info" ==> Get info of the cluster

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

@@ -1,4 +1,4 @@
-use crate::utils::get_address;
+use crate::utils::get_url;
 use prometheus_parse::Sample as PromSample;
 use prometheus_parse::Value as PromValue;
 use serde::Serialize;
@@ -22,7 +22,7 @@ struct StorageMetrics {
 
 impl Default for Metrics {
     fn default() -> Self {
-        let socket = get_address();
+        let socket = get_url();
         let address = format!(
             "http://{}:{}",
             socket.domain().unwrap(),

server/src/metrics/prom_utils.rs

@@ -25,8 +25,9 @@ use super::{
     SCHEMA_FILE_NAME, STREAM_METADATA_FILE_NAME, STREAM_ROOT_DIRECTORY,
 };
 
+use crate::handlers::http::modal::ingest_server::INGESTOR_META;
 use crate::option::Mode;
-use crate::utils::get_address;
+use crate::utils::get_url;
 use crate::{
     alerts::Alerts,
     catalog::{self, manifest::Manifest, snapshot::Snapshot},
@@ -257,7 +258,7 @@ pub trait ObjectStorage: Sync + 'static {
         let stream_metadata = match self.get_object(&stream_json_path(stream_name)).await {
             Ok(data) => data,
             Err(_) => {
-                // ! this is hard coded for now
+                // get the base stream metadata
                 let bytes = self
                     .get_object(&RelativePathBuf::from_iter([
                         stream_name,
@@ -538,11 +539,9 @@ fn to_bytes(any: &(impl ?Sized + serde::Serialize)) -> Bytes {
 fn schema_path(stream_name: &str) -> RelativePathBuf {
     match CONFIG.parseable.mode {
         Mode::Ingest => {
-            let addr = get_address();
             let file_name = format!(
-                ".ingestor.{}.{}{}",
-                addr.domain().unwrap(),
-                addr.port().unwrap_or_default(),
+                ".ingestor.{}{}",
+                INGESTOR_META.ingestor_id.clone(),
                 SCHEMA_FILE_NAME
             );
 
@@ -558,11 +557,9 @@ fn schema_path(stream_name: &str) -> RelativePathBuf {
 pub fn stream_json_path(stream_name: &str) -> RelativePathBuf {
     match &CONFIG.parseable.mode {
         Mode::Ingest => {
-            let addr = get_address();
             let file_name = format!(
-                ".ingestor.{}.{}{}",
-                addr.domain().unwrap(),
-                addr.port().unwrap_or_default(),
+                ".ingestor.{}{}",
+                INGESTOR_META.get_ingestor_id(),
                 STREAM_METADATA_FILE_NAME
             );
             RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
@@ -581,14 +578,15 @@ pub fn parseable_json_path() -> RelativePathBuf {
     RelativePathBuf::from_iter([PARSEABLE_ROOT_DIRECTORY, PARSEABLE_METADATA_FILE_NAME])
 }
 
+/// TODO: Needs to be updated for distributed mode
 #[inline(always)]
 fn alert_json_path(stream_name: &str) -> RelativePathBuf {
     RelativePathBuf::from_iter([stream_name, ALERT_FILE_NAME])
 }
 
 #[inline(always)]
 fn manifest_path(prefix: &str) -> RelativePathBuf {
-    let addr = get_address();
+    let addr = get_url();
     let mainfest_file_name = format!(
         "{}.{}.{}",
         addr.domain().unwrap(),
@@ -599,9 +597,16 @@ fn manifest_path(prefix: &str) -> RelativePathBuf {
 }
 
 #[inline(always)]
-pub fn ingestor_metadata_path(ip: String, port: String) -> RelativePathBuf {
+pub fn ingestor_metadata_path(id: Option<&str>) -> RelativePathBuf {
+    if let Some(id) = id {
+        return RelativePathBuf::from_iter([
+            PARSEABLE_ROOT_DIRECTORY,
+            &format!("ingestor.{}.json", id),
+        ]);
+    }
+
     RelativePathBuf::from_iter([
         PARSEABLE_ROOT_DIRECTORY,
-        &format!("ingestor.{}.{}.json", ip, port),
+        &format!("ingestor.{}.json", INGESTOR_META.get_ingestor_id()),
     ])
 }