refactor: human_size parsing

Author: Devdutt Shenoi

src/handlers/http/logstream.rs

@@ -50,7 +50,7 @@ use bytes::Bytes;
 use chrono::Utc;
 use http::{HeaderName, HeaderValue};
 use itertools::Itertools;
-use serde_json::Value;
+use serde_json::{json, Value};
 use std::collections::HashMap;
 use std::fs;
 use std::num::NonZeroU32;
@@ -630,10 +630,10 @@ pub async fn put_stream_hot_tier(
     STREAM_INFO.set_hot_tier(&stream_name, true)?;
     if let Some(hot_tier_manager) = HotTierManager::global() {
         let existing_hot_tier_used_size = hot_tier_manager
-            .validate_hot_tier_size(&stream_name, &hottier.size)
+            .validate_hot_tier_size(&stream_name, hottier.size)
             .await?;
-        hottier.used_size = existing_hot_tier_used_size.to_string();
-        hottier.available_size = hottier.size.to_string();
+        hottier.used_size = existing_hot_tier_used_size;
+        hottier.available_size = hottier.size;
         hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
         hot_tier_manager
             .put_hot_tier(&stream_name, &mut hottier)
@@ -674,11 +674,23 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
     }
 
     if let Some(hot_tier_manager) = HotTierManager::global() {
-        let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
-        hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
-        hot_tier.used_size = format!("{} Bytes", hot_tier.used_size);
-        hot_tier.available_size = format!("{} Bytes", hot_tier.available_size);
-        Ok((web::Json(hot_tier), StatusCode::OK))
+        let StreamHotTier {
+            version,
+            size,
+            used_size,
+            available_size,
+            oldest_date_time_entry,
+        } = hot_tier_manager.get_hot_tier(&stream_name).await?;
+        let mut json = json!({
+            "version": version,
+            "size": format!("{size} Bytes"),
+            "used_size": format!("{used_size} Bytes"),
+            "available_size": format!("{available_size} Bytes"),
+        });
+        if let Some(entry) = oldest_date_time_entry {
+            json["oldest_date_time_entry"] = serde_json::Value::String(entry);
+        }
+        Ok((web::Json(json), StatusCode::OK))
     } else {
         Err(StreamError::Custom {
             msg: format!("hot tier not initialised for stream {}", stream_name),

src/hottier.rs

@@ -27,9 +27,9 @@ use crate::{
     catalog::manifest::{File, Manifest},
     handlers::http::cluster::INTERNAL_STREAM_NAME,
     metadata::{error::stream_info::MetadataError, STREAM_INFO},
-    option::{validation::bytes_to_human_size, CONFIG},
+    option::CONFIG,
     storage::{ObjectStorage, ObjectStorageError},
-    utils::extract_datetime,
+    utils::{extract_datetime, human_size::bytes_to_human_size},
     validator::error::HotTierValidationError,
 };
 use chrono::NaiveDate;
@@ -56,10 +56,12 @@ pub const CURRENT_HOT_TIER_VERSION: &str = "v2";
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct StreamHotTier {
     pub version: Option<String>,
-    pub size: String,
-    pub used_size: String,
-    pub available_size: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(with = "crate::utils::human_size")]
+    pub size: u64,
+    #[serde(with = "crate::utils::human_size")]
+    pub used_size: u64,
+    #[serde(with = "crate::utils::human_size")]
+    pub available_size: u64,
     pub oldest_date_time_entry: Option<String>,
 }
 
@@ -97,8 +99,8 @@ impl HotTierManager {
         for stream in STREAM_INFO.list_streams() {
             if self.check_stream_hot_tier_exists(&stream) && stream != current_stream {
                 let stream_hot_tier = self.get_hot_tier(&stream).await?;
-                total_hot_tier_size += &stream_hot_tier.size.parse::<u64>().unwrap();
-                total_hot_tier_used_size += stream_hot_tier.used_size.parse::<u64>().unwrap();
+                total_hot_tier_size += &stream_hot_tier.size;
+                total_hot_tier_used_size += stream_hot_tier.used_size;
             }
         }
         Ok((total_hot_tier_size, total_hot_tier_used_size))
@@ -111,14 +113,13 @@ impl HotTierManager {
     pub async fn validate_hot_tier_size(
         &self,
         stream: &str,
-        stream_hot_tier_size: &str,
+        stream_hot_tier_size: u64,
     ) -> Result<u64, HotTierError> {
-        let stream_hot_tier_size = stream_hot_tier_size.parse::<u64>().unwrap();
         let mut existing_hot_tier_used_size = 0;
         if self.check_stream_hot_tier_exists(stream) {
             //delete existing hot tier if its size is less than the updated hot tier size else return error
             let existing_hot_tier = self.get_hot_tier(stream).await?;
-            existing_hot_tier_used_size = existing_hot_tier.used_size.parse::<u64>().unwrap();
+            existing_hot_tier_used_size = existing_hot_tier.used_size;
 
             if stream_hot_tier_size < existing_hot_tier_used_size {
                 return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
@@ -169,20 +170,17 @@ impl HotTierManager {
             ));
         }
         let path = hot_tier_file_path(&self.hot_tier_path, stream)?;
-        let res = self
+        let bytes = self
             .filesystem
             .get(&path)
             .and_then(|resp| resp.bytes())
-            .await;
-        match res {
-            Ok(bytes) => {
-                let mut stream_hot_tier: StreamHotTier = serde_json::from_slice(&bytes)?;
-                let oldest_date_time_entry = self.get_oldest_date_time_entry(stream).await?;
-                stream_hot_tier.oldest_date_time_entry = oldest_date_time_entry;
-                Ok(stream_hot_tier)
-            }
-            Err(err) => Err(err.into()),
-        }
+            .await?;
+
+        let mut stream_hot_tier: StreamHotTier = serde_json::from_slice(&bytes)?;
+        let oldest_date_time_entry = self.get_oldest_date_time_entry(stream).await?;
+        stream_hot_tier.oldest_date_time_entry = oldest_date_time_entry;
+
+        Ok(stream_hot_tier)
     }
 
     pub async fn delete_hot_tier(&self, stream: &str) -> Result<(), HotTierError> {
@@ -256,7 +254,7 @@ impl HotTierManager {
     /// delete the files from the hot tier directory if the available date range is outside the hot tier range
     async fn process_stream(&self, stream: String) -> Result<(), HotTierError> {
         let stream_hot_tier = self.get_hot_tier(&stream).await?;
-        let mut parquet_file_size = stream_hot_tier.used_size.parse::<u64>().unwrap();
+        let mut parquet_file_size = stream_hot_tier.used_size;
 
         let object_store = CONFIG.storage().get_object_store();
         let mut s3_manifest_file_list = object_store.list_manifest_files(&stream).await?;
@@ -348,7 +346,7 @@ impl HotTierManager {
         let mut file_processed = false;
         let mut stream_hot_tier = self.get_hot_tier(stream).await?;
         if !self.is_disk_available(parquet_file.file_size).await?
-            || stream_hot_tier.available_size.parse::<u64>().unwrap() <= parquet_file.file_size
+            || stream_hot_tier.available_size <= parquet_file.file_size
         {
             if !self
                 .cleanup_hot_tier_old_data(
@@ -361,7 +359,7 @@ impl HotTierManager {
             {
                 return Ok(file_processed);
             }
-            *parquet_file_size = stream_hot_tier.used_size.parse::<u64>().unwrap();
+            *parquet_file_size = stream_hot_tier.used_size;
         }
         let parquet_file_path = RelativePathBuf::from(parquet_file.file_path.clone());
         fs::create_dir_all(parquet_path.parent().unwrap()).await?;
@@ -373,11 +371,9 @@ impl HotTierManager {
             .await?;
         file.write_all(&parquet_data).await?;
         *parquet_file_size += parquet_file.file_size;
-        stream_hot_tier.used_size = parquet_file_size.to_string();
+        stream_hot_tier.used_size = *parquet_file_size;
 
-        stream_hot_tier.available_size = (stream_hot_tier.available_size.parse::<u64>().unwrap()
-            - parquet_file.file_size)
-            .to_string();
+        stream_hot_tier.available_size = stream_hot_tier.available_size - parquet_file.file_size;
         self.put_hot_tier(stream, &mut stream_hot_tier).await?;
         file_processed = true;
         let path = self.get_stream_path_for_date(stream, &date);
@@ -598,18 +594,12 @@ impl HotTierManager {
                         fs::remove_dir_all(path_to_delete.parent().unwrap()).await?;
                         delete_empty_directory_hot_tier(path_to_delete.parent().unwrap()).await?;
 
-                        stream_hot_tier.used_size =
-                            (stream_hot_tier.used_size.parse::<u64>().unwrap() - file_size)
-                                .to_string();
-                        stream_hot_tier.available_size =
-                            (stream_hot_tier.available_size.parse::<u64>().unwrap() + file_size)
-                                .to_string();
+                        stream_hot_tier.used_size = stream_hot_tier.used_size - file_size;
+                        stream_hot_tier.available_size = stream_hot_tier.available_size + file_size;
                         self.put_hot_tier(stream, stream_hot_tier).await?;
                         delete_successful = true;
 
-                        if stream_hot_tier.available_size.parse::<u64>().unwrap()
-                            <= parquet_file_size
-                        {
+                        if stream_hot_tier.available_size <= parquet_file_size {
                             continue 'loop_files;
                         } else {
                             break 'loop_dates;
@@ -699,9 +689,9 @@ impl HotTierManager {
         if !self.check_stream_hot_tier_exists(INTERNAL_STREAM_NAME) {
             let mut stream_hot_tier = StreamHotTier {
                 version: Some(CURRENT_HOT_TIER_VERSION.to_string()),
-                size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string(),
-                used_size: "0".to_string(),
-                available_size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string(),
+                size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES,
+                used_size: 0,
+                available_size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES,
                 oldest_date_time_entry: None,
             };
             self.put_hot_tier(INTERNAL_STREAM_NAME, &mut stream_hot_tier)

src/migration/mod.rs

@@ -24,9 +24,8 @@ mod stream_metadata_migration;
 use std::{fs::OpenOptions, sync::Arc};
 
 use crate::{
-    hottier::{HotTierManager, CURRENT_HOT_TIER_VERSION},
     metadata::load_stream_metadata_on_server_start,
-    option::{validation::human_size_to_bytes, Config, Mode, CONFIG},
+    option::{Config, Mode, CONFIG},
     storage::{
         object_storage::{parseable_json_path, schema_path, stream_json_path},
         ObjectStorage, ObjectStorageError, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY,
@@ -137,38 +136,17 @@ pub async fn run_migration(config: &Config) -> anyhow::Result<()> {
     let streams = storage.list_streams().await?;
     for stream in streams {
         migration_stream(&stream.name, &*storage).await?;
-        if CONFIG.options.hot_tier_storage_path.is_some() {
-            migration_hot_tier(&stream.name).await?;
-        }
     }
 
     Ok(())
 }
 
-/// run the migration for hot tier
-async fn migration_hot_tier(stream: &str) -> anyhow::Result<()> {
-    if let Some(hot_tier_manager) = HotTierManager::global() {
-        if hot_tier_manager.check_stream_hot_tier_exists(stream) {
-            let mut stream_hot_tier = hot_tier_manager.get_hot_tier(stream).await?;
-            if stream_hot_tier.version.is_none() {
-                stream_hot_tier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
-                stream_hot_tier.size = human_size_to_bytes(&stream_hot_tier.size)
-                    .unwrap()
-                    .to_string();
-                stream_hot_tier.available_size =
-                    human_size_to_bytes(&stream_hot_tier.available_size)
-                        .unwrap()
-                        .to_string();
-                stream_hot_tier.used_size = human_size_to_bytes(&stream_hot_tier.used_size)
-                    .unwrap()
-                    .to_string();
-                hot_tier_manager
-                    .put_hot_tier(stream, &mut stream_hot_tier)
-                    .await?;
-            }
-        }
-    }
-    Ok(())
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct StreamHotTier {
+    pub version: Option<String>,
+    pub size: String,
+    pub used_size: String,
+    pub available_size: String,
 }
 
 async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::Result<()> {

src/option.rs

@@ -192,13 +192,10 @@ pub mod validation {
         env, io,
         net::ToSocketAddrs,
         path::{Path, PathBuf},
-        str::FromStr,
     };
 
     use path_clean::PathClean;
 
-    use human_size::{multiples, SpecificSize};
-
     #[cfg(any(
         all(target_os = "linux", target_arch = "x86_64"),
         all(target_os = "macos", target_arch = "aarch64")
@@ -285,45 +282,6 @@ pub mod validation {
         }
     }
 
-    pub fn human_size_to_bytes(s: &str) -> Result<u64, String> {
-        fn parse_and_map<T: human_size::Multiple>(
-            s: &str,
-        ) -> Result<u64, human_size::ParsingError> {
-            SpecificSize::<T>::from_str(s).map(|x| x.to_bytes())
-        }
-
-        let size = parse_and_map::<multiples::Mebibyte>(s)
-            .or(parse_and_map::<multiples::Megabyte>(s))
-            .or(parse_and_map::<multiples::Gigibyte>(s))
-            .or(parse_and_map::<multiples::Gigabyte>(s))
-            .or(parse_and_map::<multiples::Tebibyte>(s))
-            .or(parse_and_map::<multiples::Terabyte>(s))
-            .map_err(|_| "Could not parse given size".to_string())?;
-        Ok(size)
-    }
-
-    pub fn bytes_to_human_size(bytes: u64) -> String {
-        const KIB: u64 = 1024;
-        const MIB: u64 = KIB * 1024;
-        const GIB: u64 = MIB * 1024;
-        const TIB: u64 = GIB * 1024;
-        const PIB: u64 = TIB * 1024;
-
-        if bytes < KIB {
-            format!("{} B", bytes)
-        } else if bytes < MIB {
-            format!("{:.2} KB", bytes as f64 / KIB as f64)
-        } else if bytes < GIB {
-            format!("{:.2} MiB", bytes as f64 / MIB as f64)
-        } else if bytes < TIB {
-            format!("{:.2} GiB", bytes as f64 / GIB as f64)
-        } else if bytes < PIB {
-            format!("{:.2} TiB", bytes as f64 / TIB as f64)
-        } else {
-            format!("{:.2} PiB", bytes as f64 / PIB as f64)
-        }
-    }
-
     pub fn validate_disk_usage(max_disk_usage: &str) -> Result<f64, String> {
         if let Ok(max_disk_usage) = max_disk_usage.parse::<f64>() {
             if (0.0..=100.0).contains(&max_disk_usage) {

src/utils/human_size.rs

@@ -0,0 +1,59 @@
+use std::str::FromStr;
+
+use human_size::{multiples, SpecificSize};
+use serde::{de, Deserialize, Deserializer, Serializer};
+
+// Function to convert human-readable size to bytes (already provided)
+pub fn human_size_to_bytes(s: &str) -> Result<u64, String> {
+    fn parse_and_map<T: human_size::Multiple>(s: &str) -> Result<u64, human_size::ParsingError> {
+        SpecificSize::<T>::from_str(s).map(|x| x.to_bytes())
+    }
+
+    let size = parse_and_map::<multiples::Mebibyte>(s)
+        .or(parse_and_map::<multiples::Megabyte>(s))
+        .or(parse_and_map::<multiples::Gigibyte>(s))
+        .or(parse_and_map::<multiples::Gigabyte>(s))
+        .or(parse_and_map::<multiples::Tebibyte>(s))
+        .or(parse_and_map::<multiples::Terabyte>(s))
+        .map_err(|_| "Could not parse given size".to_string())?;
+    Ok(size)
+}
+
+// Function to convert bytes to human-readable size (already provided)
+pub fn bytes_to_human_size(bytes: u64) -> String {
+    const KIB: u64 = 1024;
+    const MIB: u64 = KIB * 1024;
+    const GIB: u64 = MIB * 1024;
+    const TIB: u64 = GIB * 1024;
+    const PIB: u64 = TIB * 1024;
+
+    if bytes < KIB {
+        format!("{} B", bytes)
+    } else if bytes < MIB {
+        format!("{:.2} KB", bytes as f64 / KIB as f64)
+    } else if bytes < GIB {
+        format!("{:.2} MiB", bytes as f64 / MIB as f64)
+    } else if bytes < TIB {
+        format!("{:.2} GiB", bytes as f64 / GIB as f64)
+    } else if bytes < PIB {
+        format!("{:.2} TiB", bytes as f64 / TIB as f64)
+    } else {
+        format!("{:.2} PiB", bytes as f64 / PIB as f64)
+    }
+}
+
+pub fn serialize<S>(bytes: &u64, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    let human_readable = bytes_to_human_size(*bytes);
+    serializer.serialize_str(&human_readable)
+}
+
+pub fn deserialize<'de, D>(deserializer: D) -> Result<u64, D::Error>
+where
+    D: Deserializer<'de>,
+{
+    let s = String::deserialize(deserializer)?;
+    human_size_to_bytes(&s).map_err(|e| de::Error::custom(e))
+}