update get_address to use url instead of sockaddr

Author: Eshanatnight

server/src/catalog.rs

@@ -118,7 +118,12 @@ pub async fn update_snapshot(
         let mut ch = false;
         for m in manifests.iter() {
             let s = get_address();
-            let p = format!("{}.{}.{}", s.ip(), s.port(), MANIFEST_FILE);
+            let p = format!(
+                "{}.{}.{}",
+                s.domain().unwrap(),
+                s.port().unwrap_or_default(),
+                MANIFEST_FILE
+            );
             if m.manifest_path.contains(&p) {
                 ch = true;
             }
@@ -152,7 +157,12 @@ pub async fn update_snapshot(
             };
 
             let addr = get_address();
-            let mainfest_file_name = format!("{}.{}.{}", addr.ip(), addr.port(), MANIFEST_FILE);
+            let mainfest_file_name = format!(
+                "{}.{}.{}",
+                addr.domain().unwrap(),
+                addr.port().unwrap_or_default(),
+                MANIFEST_FILE
+            );
             let path =
                 partition_path(stream_name, lower_bound, upper_bound).join(&mainfest_file_name);
             storage
@@ -186,7 +196,12 @@ pub async fn update_snapshot(
         };
 
         let addr = get_address();
-        let mainfest_file_name = format!("{}.{}.{}", addr.ip(), addr.port(), MANIFEST_FILE);
+        let mainfest_file_name = format!(
+            "{}.{}.{}",
+            addr.domain().unwrap(),
+            addr.port().unwrap(),
+            MANIFEST_FILE
+        );
         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/handlers/http/modal/ingest_server.rs

@@ -183,7 +183,10 @@ impl IngestServer {
         let store = CONFIG.storage().get_object_store();
 
         let sock = get_address();
-        let path = ingestor_metadata_path(sock.ip().to_string(), sock.port().to_string());
+        let path = ingestor_metadata_path(
+            sock.domain().unwrap().to_string(),
+            sock.port().unwrap_or_default().to_string(),
+        );
 
         if store.get_object(&path).await.is_ok() {
             println!("ingestor metadata already exists");
@@ -192,13 +195,19 @@ impl IngestServer {
 
         let scheme = CONFIG.parseable.get_scheme();
         let resource = IngestorMetadata::new(
-            sock.port().to_string(),
+            sock.port().unwrap_or_default().to_string(),
             CONFIG
                 .parseable
                 .domain_address
                 .clone()
                 .unwrap_or_else(|| {
-                    Url::parse(&format!("{}://{}:{}", scheme, sock.ip(), sock.port())).unwrap()
+                    Url::parse(&format!(
+                        "{}://{}:{}",
+                        scheme,
+                        sock.domain().unwrap(),
+                        sock.port().unwrap_or_default()
+                    ))
+                    .unwrap()
                 })
                 .to_string(),
             DEFAULT_VERSION.to_string(),

server/src/metrics/prom_utils.rs

@@ -23,7 +23,11 @@ struct StorageMetrics {
 impl Default for Metrics {
     fn default() -> Self {
         let socket = get_address();
-        let address = format!("http://{}:{}", socket.ip(), socket.port());
+        let address = format!(
+            "http://{}:{}",
+            socket.domain().unwrap(),
+            socket.port().unwrap_or_default()
+        );
         Metrics {
             address,
             parseable_events_ingested: 0.0,

server/src/storage/object_storage.rs

@@ -542,8 +542,8 @@ fn schema_path(stream_name: &str) -> RelativePathBuf {
             let addr = get_address();
             let file_name = format!(
                 ".ingestor.{}.{}{}",
-                addr.ip(),
-                addr.port(),
+                addr.domain().unwrap(),
+                addr.port().unwrap_or_default(),
                 SCHEMA_FILE_NAME
             );
 
@@ -562,8 +562,8 @@ pub fn stream_json_path(stream_name: &str) -> RelativePathBuf {
             let addr = get_address();
             let file_name = format!(
                 ".ingestor.{}.{}{}",
-                addr.ip(),
-                addr.port(),
+                addr.domain().unwrap(),
+                addr.port().unwrap_or_default(),
                 STREAM_METADATA_FILE_NAME
             );
             RelativePathBuf::from_iter([stream_name, STREAM_ROOT_DIRECTORY, &file_name])
@@ -590,7 +590,12 @@ fn alert_json_path(stream_name: &str) -> RelativePathBuf {
 #[inline(always)]
 fn manifest_path(prefix: &str) -> RelativePathBuf {
     let addr = get_address();
-    let mainfest_file_name = format!("{}.{}.{}", addr.ip(), addr.port(), MANIFEST_FILE);
+    let mainfest_file_name = format!(
+        "{}.{}.{}",
+        addr.domain().unwrap(),
+        addr.port().unwrap_or_default(),
+        MANIFEST_FILE
+    );
     RelativePathBuf::from_iter([prefix, &mainfest_file_name])
 }
 

server/src/storage/staging.rs

@@ -64,8 +64,8 @@ impl StorageDir {
             + &utils::minute_to_prefix(time.minute(), OBJECT_STORE_DATA_GRANULARITY).unwrap();
         let local_uri = str::replace(&uri, "/", ".");
         let sock = get_address();
-        let ip = sock.ip();
-        let port = sock.port();
+        let ip = sock.domain().unwrap();
+        let port = sock.port().unwrap_or_default();
         format!("{local_uri}{ip}.{port}.{extention}")
     }
 

server/src/utils.rs

@@ -25,7 +25,9 @@ pub mod update;
 use crate::option::CONFIG;
 use chrono::{DateTime, NaiveDate, Timelike, Utc};
 use std::env;
+#[allow(unused_imports)]
 use std::net::SocketAddr;
+use url::Url;
 
 #[allow(dead_code)]
 pub fn hostname() -> Option<String> {
@@ -224,10 +226,9 @@ impl TimePeriod {
     }
 }
 
-#[inline(always)]
-pub fn get_address() -> SocketAddr {
+pub fn get_address() -> Url {
     if CONFIG.parseable.ingestor_url.is_empty() {
-        CONFIG.parseable.address.parse::<SocketAddr>().unwrap()
+        CONFIG.parseable.address.parse::<Url>().unwrap()
     } else {
         let addr_from_env = CONFIG
             .parseable
@@ -245,9 +246,7 @@ pub fn get_address() -> SocketAddr {
             let var_port = port[1..].to_string();
             port = get_from_env(&var_port);
         }
-        format!("{}:{}", hostname, port)
-            .parse::<SocketAddr>()
-            .unwrap()
+        format!("{}:{}", hostname, port).parse::<Url>().unwrap()
     }
 }
 fn get_from_env(var_to_fetch: &str) -> String {