enhancement: change sizes to bytes in hot tier (#885)

nikhilsinhaparseable · web-flow · commit 900bf4700b05 · 2024-08-15T21:24:27.000+05:30
in PUT /logstream/{logstream}/hottier
request body:
{
    "size": "10737418240 Bytes"
}

GET /logstream/{logstream}/hottier
response body:
{
    "size": "1073741824 Bytes",
    "used_size": "121798121 Bytes",
    "available_size": "951943703 Bytes",
    "oldest_date_time_entry": "2024-08-10T11:45:00.000Z"
}
diff --git a/server/src/handlers/http/logstream.rs b/server/src/handlers/http/logstream.rs
@@ -29,10 +29,9 @@ use crate::handlers::{
     CUSTOM_PARTITION_KEY, STATIC_SCHEMA_FLAG, STREAM_TYPE_KEY, TIME_PARTITION_KEY,
     TIME_PARTITION_LIMIT_KEY, UPDATE_STREAM_KEY,
 };
-use crate::hottier::{HotTierManager, StreamHotTier};
+use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
 use crate::metadata::STREAM_INFO;
 use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
-use crate::option::validation::bytes_to_human_size;
 use crate::option::{Mode, CONFIG};
 use crate::static_schema::{convert_static_schema_to_arrow_schema, StaticSchema};
 use crate::stats::{event_labels_date, storage_size_labels_date, Stats};
@@ -992,8 +991,9 @@ pub async fn put_stream_hot_tier(
         let existing_hot_tier_used_size = hot_tier_manager
             .validate_hot_tier_size(&stream_name, &hottier.size)
             .await?;
-        hottier.used_size = Some(bytes_to_human_size(existing_hot_tier_used_size));
+        hottier.used_size = Some(existing_hot_tier_used_size.to_string());
         hottier.available_size = Some(hottier.size.clone());
+        hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
         hot_tier_manager
             .put_hot_tier(&stream_name, &mut hottier)
             .await?;
@@ -1030,7 +1030,10 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
     }
 
     if let Some(hot_tier_manager) = HotTierManager::global() {
-        let hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
+        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 = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
+        hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
         Ok((web::Json(hot_tier), StatusCode::OK))
     } else {
         Err(StreamError::Custom {
@@ -1172,7 +1175,7 @@ pub mod error {
         HotTierNotEnabled(String),
         #[error("failed to enable hottier due to err: {0}")]
         InvalidHotTierConfig(serde_json::Error),
-        #[error("Hot tier validation failed due to {0}")]
+        #[error("Hot tier validation failed: {0}")]
         HotTierValidation(#[from] HotTierValidationError),
         #[error("{0}")]
         HotTierError(#[from] HotTierError),
diff --git a/server/src/hottier.rs b/server/src/hottier.rs
@@ -27,10 +27,7 @@ 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, human_size_to_bytes},
-        CONFIG,
-    },
+    option::{validation::bytes_to_human_size, CONFIG},
     storage::{ObjectStorage, ObjectStorageError},
     utils::extract_datetime,
     validator::error::HotTierValidationError,
@@ -53,9 +50,10 @@ pub const STREAM_HOT_TIER_FILENAME: &str = ".hot_tier.json";
 pub const MIN_STREAM_HOT_TIER_SIZE_BYTES: u64 = 10737418240; // 10 GiB
 const HOT_TIER_SYNC_DURATION: Interval = clokwerk::Interval::Minutes(1);
 pub const INTERNAL_STREAM_HOT_TIER_SIZE_BYTES: u64 = 10485760; //10 MiB
-
+pub const CURRENT_HOT_TIER_VERSION: &str = "v2";
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 pub struct StreamHotTier {
+    pub version: Option<String>,
     #[serde(rename = "size")]
     pub size: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -97,9 +95,13 @@ 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 += human_size_to_bytes(&stream_hot_tier.size).unwrap();
-                total_hot_tier_used_size +=
-                    human_size_to_bytes(&stream_hot_tier.used_size.unwrap()).unwrap();
+                total_hot_tier_size += &stream_hot_tier.size.parse::<u64>().unwrap();
+                total_hot_tier_used_size += &stream_hot_tier
+                    .used_size
+                    .clone()
+                    .unwrap()
+                    .parse::<u64>()
+                    .unwrap();
             }
         }
         Ok((total_hot_tier_size, total_hot_tier_used_size))
@@ -112,19 +114,21 @@ impl HotTierManager {
     pub async fn validate_hot_tier_size(
         &self,
         stream: &str,
-        size: &str,
+        stream_hot_tier_size: &str,
     ) -> 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 =
-                human_size_to_bytes(&existing_hot_tier.used_size.unwrap()).unwrap();
-            if human_size_to_bytes(size) < human_size_to_bytes(&existing_hot_tier.size) {
+                existing_hot_tier.used_size.unwrap().parse::<u64>().unwrap();
+
+            if stream_hot_tier_size < existing_hot_tier_used_size {
                 return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
                     "Reducing hot tier size is not supported, failed to reduce the hot tier size from {} to {}",
-                    existing_hot_tier.size,
-                    size
+                    bytes_to_human_size(existing_hot_tier_used_size),
+                    bytes_to_human_size(stream_hot_tier_size)
                 ))));
             }
         }
@@ -134,7 +138,6 @@ impl HotTierManager {
         if let (Some(total_disk_space), _, Some(used_disk_space)) =
             (total_disk_space, available_disk_space, used_disk_space)
         {
-            let stream_hot_tier_size = human_size_to_bytes(size).unwrap();
             let (total_hot_tier_size, total_hot_tier_used_size) =
                 self.get_hot_tiers_size(stream).await?;
             let disk_threshold =
@@ -147,7 +150,7 @@ impl HotTierManager {
 
             if stream_hot_tier_size as f64 > max_allowed_hot_tier_size {
                 log::error!("disk_threshold: {}, used_disk_space: {}, total_hot_tier_used_size: {}, existing_hot_tier_used_size: {}, total_hot_tier_size: {}",
-                    disk_threshold, used_disk_space, total_hot_tier_used_size, existing_hot_tier_used_size, total_hot_tier_size);
+                    bytes_to_human_size(disk_threshold as u64), bytes_to_human_size(used_disk_space), bytes_to_human_size(total_hot_tier_used_size), bytes_to_human_size(existing_hot_tier_used_size), bytes_to_human_size(total_hot_tier_size));
                 return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
                     "{} is the total usable disk space for hot tier, cannot set a bigger value.", bytes_to_human_size(max_allowed_hot_tier_size as u64)
                 ))));
@@ -255,8 +258,12 @@ 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 =
-            human_size_to_bytes(stream_hot_tier.used_size.as_ref().unwrap()).unwrap();
+        let mut parquet_file_size = stream_hot_tier
+            .used_size
+            .as_ref()
+            .unwrap()
+            .parse::<u64>()
+            .unwrap();
 
         let object_store = CONFIG.storage().get_object_store();
         let mut s3_manifest_file_list = object_store.list_manifest_files(&stream).await?;
@@ -348,7 +355,12 @@ 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?
-            || human_size_to_bytes(&stream_hot_tier.available_size.clone().unwrap()).unwrap()
+            || stream_hot_tier
+                .available_size
+                .as_ref()
+                .unwrap()
+                .parse::<u64>()
+                .unwrap()
                 <= parquet_file.file_size
         {
             if !self
@@ -362,8 +374,12 @@ impl HotTierManager {
             {
                 return Ok(file_processed);
             }
-            *parquet_file_size =
-                human_size_to_bytes(&stream_hot_tier.used_size.clone().unwrap()).unwrap();
+            *parquet_file_size = stream_hot_tier
+                .used_size
+                .as_ref()
+                .unwrap()
+                .parse::<u64>()
+                .unwrap();
         }
         let parquet_file_path = RelativePathBuf::from(parquet_file.file_path.clone());
         fs::create_dir_all(parquet_path.parent().unwrap()).await?;
@@ -375,12 +391,18 @@ impl HotTierManager {
             .await?;
         file.write_all(&parquet_data).await?;
         *parquet_file_size += parquet_file.file_size;
-        stream_hot_tier.used_size = Some(bytes_to_human_size(*parquet_file_size));
-
-        stream_hot_tier.available_size = Some(bytes_to_human_size(
-            human_size_to_bytes(&stream_hot_tier.available_size.clone().unwrap()).unwrap()
-                - parquet_file.file_size,
-        ));
+        stream_hot_tier.used_size = Some(parquet_file_size.to_string());
+
+        stream_hot_tier.available_size = Some(
+            (stream_hot_tier
+                .available_size
+                .as_ref()
+                .unwrap()
+                .parse::<u64>()
+                .unwrap()
+                - parquet_file.file_size)
+                .to_string(),
+        );
         self.put_hot_tier(stream, &mut stream_hot_tier).await?;
         file_processed = true;
         let mut hot_tier_manifest = self
@@ -583,20 +605,34 @@ 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 = Some(bytes_to_human_size(
-                            human_size_to_bytes(&stream_hot_tier.used_size.clone().unwrap())
+                        stream_hot_tier.used_size = Some(
+                            (stream_hot_tier
+                                .used_size
+                                .as_ref()
                                 .unwrap()
-                                - file_size,
-                        ));
-                        stream_hot_tier.available_size = Some(bytes_to_human_size(
-                            human_size_to_bytes(&stream_hot_tier.available_size.clone().unwrap())
+                                .parse::<u64>()
                                 .unwrap()
-                                + file_size,
-                        ));
+                                - file_size)
+                                .to_string(),
+                        );
+                        stream_hot_tier.available_size = Some(
+                            (stream_hot_tier
+                                .available_size
+                                .as_ref()
+                                .unwrap()
+                                .parse::<u64>()
+                                .unwrap()
+                                + file_size)
+                                .to_string(),
+                        );
                         self.put_hot_tier(stream, stream_hot_tier).await?;
                         delete_successful = true;
 
-                        if human_size_to_bytes(&stream_hot_tier.available_size.clone().unwrap())
+                        if stream_hot_tier
+                            .available_size
+                            .as_ref()
+                            .unwrap()
+                            .parse::<u64>()
                             .unwrap()
                             <= parquet_file_size
                         {
@@ -679,7 +715,7 @@ impl HotTierManager {
                         .to_string_lossy()
                         .trim_start_matches("minute=")
                         .to_string();
-                    let oldest_date_time = format!("{} {}:{}:00", date, hour_str, minute_str);
+                    let oldest_date_time = format!("{}T{}:{}:00.000Z", date, hour_str, minute_str);
                     return Ok(Some(oldest_date_time));
                 }
             }
@@ -693,6 +729,7 @@ impl HotTierManager {
             && !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: Some("0".to_string()),
                 available_size: Some(INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string()),
diff --git a/server/src/migration.rs b/server/src/migration.rs
@@ -24,8 +24,9 @@ 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::{Config, Mode, CONFIG},
+    option::{validation::human_size_to_bytes, Config, Mode, CONFIG},
     storage::{
         object_storage::{parseable_json_path, stream_json_path},
         ObjectStorage, ObjectStorageError, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY,
@@ -111,12 +112,44 @@ 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?
+        migration_stream(&stream.name, &*storage).await?;
+        if CONFIG.parseable.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 = Some(
+                    human_size_to_bytes(&stream_hot_tier.available_size.unwrap())
+                        .unwrap()
+                        .to_string(),
+                );
+                stream_hot_tier.used_size = Some(
+                    human_size_to_bytes(&stream_hot_tier.used_size.unwrap())
+                        .unwrap()
+                        .to_string(),
+                );
+                hot_tier_manager
+                    .put_hot_tier(stream, &mut stream_hot_tier)
+                    .await?;
+            }
+        }
+    }
+    Ok(())
+}
+
 async fn migration_stream(stream: &str, storage: &dyn ObjectStorage) -> anyhow::Result<()> {
     let mut arrow_schema: Schema = Schema::empty();
     let schema_path = RelativePathBuf::from_iter([stream, STREAM_ROOT_DIRECTORY, SCHEMA_FILE_NAME]);
diff --git a/server/src/validator.rs b/server/src/validator.rs
@@ -23,7 +23,7 @@ use crate::alerts::rule::base::{NumericRule, StringRule};
 use crate::alerts::rule::{ColumnRule, ConsecutiveNumericRule, ConsecutiveStringRule};
 use crate::alerts::{Alerts, Rule};
 use crate::hottier::MIN_STREAM_HOT_TIER_SIZE_BYTES;
-use crate::option::validation::{bytes_to_human_size, human_size_to_bytes};
+use crate::option::validation::bytes_to_human_size;
 use crate::storage::StreamType;
 
 // Add more sql keywords here in lower case
@@ -149,17 +149,16 @@ pub fn user_name(username: &str) -> Result<(), UsernameValidationError> {
 }
 
 pub fn hot_tier(size: &str) -> Result<(), HotTierValidationError> {
-    if human_size_to_bytes(size).is_err() {
-        return Err(HotTierValidationError::InvalidFormat);
-    }
-
-    if human_size_to_bytes(size).unwrap() < MIN_STREAM_HOT_TIER_SIZE_BYTES {
-        return Err(HotTierValidationError::Size(bytes_to_human_size(
-            MIN_STREAM_HOT_TIER_SIZE_BYTES,
-        )));
+    if let Ok(size) = size.parse::<u64>() {
+        if size < MIN_STREAM_HOT_TIER_SIZE_BYTES {
+            return Err(HotTierValidationError::Size(bytes_to_human_size(
+                MIN_STREAM_HOT_TIER_SIZE_BYTES,
+            )));
+        }
+        Ok(())
+    } else {
+        Err(HotTierValidationError::InvalidFormat)
     }
-
-    Ok(())
 }
 pub mod error {
 
@@ -211,7 +210,7 @@ pub mod error {
 
     #[derive(Debug, thiserror::Error)]
     pub enum HotTierValidationError {
-        #[error("Please provide size in human readable format, e.g 10GiB, 20GiB")]
+        #[error("Please provide size in bytes")]
         InvalidFormat,
 
         #[error("Stream should have atleast {0} size")]
