Skip to content

Commit da882af

Browse files
author
Devdutt Shenoi
committed
refactor: use u64 for numbers, after necessary parsing
1 parent 965d810 commit da882af

File tree

4 files changed

+82
-133
lines changed

4 files changed

+82
-133
lines changed

‎server/src/handlers/http/logstream.rs

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -559,16 +559,16 @@ pub async fn put_stream_hot_tier(
559559
Err(err) => return Err(StreamError::InvalidHotTierConfig(err)),
560560
};
561561

562-
validator::hot_tier(&hottier.size.to_string())?;
562+
validator::hot_tier(hottier.size)?;
563563

564564
STREAM_INFO.set_hot_tier(&stream_name, true)?;
565565
if let Some(hot_tier_manager) = HotTierManager::global() {
566566
let existing_hot_tier_used_size = hot_tier_manager
567-
.validate_hot_tier_size(&stream_name, &hottier.size)
567+
.validate_hot_tier_size(&stream_name, hottier.size)
568568
.await?;
569-
hottier.used_size = Some(existing_hot_tier_used_size.to_string());
569+
hottier.used_size = Some(existing_hot_tier_used_size);
570570
hottier.available_size = Some(hottier.size.clone());
571-
hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
571+
hottier.version = CURRENT_HOT_TIER_VERSION.to_string();
572572
hot_tier_manager
573573
.put_hot_tier(&stream_name, &mut hottier)
574574
.await?;
@@ -598,11 +598,13 @@ pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, Str
598598
}
599599

600600
if let Some(hot_tier_manager) = HotTierManager::global() {
601-
let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
602-
hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
603-
hot_tier.used_size = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
604-
hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
605-
Ok((web::Json(hot_tier), StatusCode::OK))
601+
let hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
602+
let json = serde_json::json!({
603+
"size": format!("{} Bytes", hot_tier.size),
604+
"used_size": format!("{} Bytes", hot_tier.used_size.unwrap()),
605+
"available_size": format!("{} Bytes", hot_tier.available_size.unwrap()),
606+
});
607+
Ok((web::Json(json), StatusCode::OK))
606608
} else {
607609
Err(StreamError::Custom {
608610
msg: format!("hot tier not initialised for stream {}", stream_name),

‎server/src/hottier.rs

Lines changed: 62 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,10 @@ use crate::{
2727
catalog::manifest::{File, Manifest},
2828
handlers::http::cluster::INTERNAL_STREAM_NAME,
2929
metadata::{error::stream_info::MetadataError, STREAM_INFO},
30-
option::{validation::bytes_to_human_size, CONFIG},
30+
option::{
31+
validation::{bytes_to_human_size, human_size_to_bytes},
32+
CONFIG,
33+
},
3134
storage::{ObjectStorage, ObjectStorageError},
3235
utils::extract_datetime,
3336
validator::error::HotTierValidationError,
@@ -40,6 +43,7 @@ use object_store::{local::LocalFileSystem, ObjectStore};
4043
use once_cell::sync::OnceCell;
4144
use parquet::errors::ParquetError;
4245
use relative_path::RelativePathBuf;
46+
use serde::{Deserialize, Serialize};
4347
use std::time::Duration;
4448
use sysinfo::{Disks, System};
4549
use tokio::fs::{self, DirEntry};
@@ -51,15 +55,16 @@ pub const MIN_STREAM_HOT_TIER_SIZE_BYTES: u64 = 10737418240; // 10 GiB
5155
const HOT_TIER_SYNC_DURATION: Interval = clokwerk::Interval::Minutes(1);
5256
pub const INTERNAL_STREAM_HOT_TIER_SIZE_BYTES: u64 = 10485760; //10 MiB
5357
pub const CURRENT_HOT_TIER_VERSION: &str = "v2";
54-
#[derive(Debug, serde::Deserialize, serde::Serialize)]
58+
59+
#[derive(Debug, Deserialize, Serialize, Default)]
5560
pub struct StreamHotTier {
56-
pub version: Option<String>,
61+
pub version: String,
5762
#[serde(rename = "size")]
58-
pub size: String,
63+
pub size: u64,
5964
#[serde(skip_serializing_if = "Option::is_none")]
60-
pub used_size: Option<String>,
65+
pub used_size: Option<u64>,
6166
#[serde(skip_serializing_if = "Option::is_none")]
62-
pub available_size: Option<String>,
67+
pub available_size: Option<u64>,
6368
#[serde(skip_serializing_if = "Option::is_none")]
6469
pub oldest_date_time_entry: Option<String>,
6570
}
@@ -97,13 +102,8 @@ impl HotTierManager {
97102
for stream in STREAM_INFO.list_streams() {
98103
if self.check_stream_hot_tier_exists(&stream) && stream != current_stream {
99104
let stream_hot_tier = self.get_hot_tier(&stream).await?;
100-
total_hot_tier_size += &stream_hot_tier.size.parse::<u64>().unwrap();
101-
total_hot_tier_used_size += &stream_hot_tier
102-
.used_size
103-
.clone()
104-
.unwrap()
105-
.parse::<u64>()
106-
.unwrap();
105+
total_hot_tier_size += stream_hot_tier.size;
106+
total_hot_tier_used_size += stream_hot_tier.used_size.as_ref().unwrap();
107107
}
108108
}
109109
Ok((total_hot_tier_size, total_hot_tier_used_size))
@@ -116,15 +116,13 @@ impl HotTierManager {
116116
pub async fn validate_hot_tier_size(
117117
&self,
118118
stream: &str,
119-
stream_hot_tier_size: &str,
119+
stream_hot_tier_size: u64,
120120
) -> Result<u64, HotTierError> {
121-
let stream_hot_tier_size = stream_hot_tier_size.parse::<u64>().unwrap();
122121
let mut existing_hot_tier_used_size = 0;
123122
if self.check_stream_hot_tier_exists(stream) {
124123
//delete existing hot tier if its size is less than the updated hot tier size else return error
125124
let existing_hot_tier = self.get_hot_tier(stream).await?;
126-
existing_hot_tier_used_size =
127-
existing_hot_tier.used_size.unwrap().parse::<u64>().unwrap();
125+
existing_hot_tier_used_size = *existing_hot_tier.used_size.as_ref().unwrap();
128126

129127
if stream_hot_tier_size < existing_hot_tier_used_size {
130128
return Err(HotTierError::ObjectStorageError(ObjectStorageError::Custom(format!(
@@ -170,20 +168,42 @@ impl HotTierManager {
170168
));
171169
}
172170
let path = hot_tier_file_path(&self.hot_tier_path, stream)?;
173-
let res = self
171+
let bytes = self
174172
.filesystem
175173
.get(&path)
176174
.and_then(|resp| resp.bytes())
177-
.await;
178-
match res {
179-
Ok(bytes) => {
180-
let mut stream_hot_tier: StreamHotTier = serde_json::from_slice(&bytes)?;
181-
let oldest_date_time_entry = self.get_oldest_date_time_entry(stream).await?;
182-
stream_hot_tier.oldest_date_time_entry = oldest_date_time_entry;
183-
Ok(stream_hot_tier)
184-
}
185-
Err(err) => Err(err.into()),
175+
.await?;
176+
177+
#[derive(Debug, Default, Deserialize)]
178+
struct StreamHotTierTemp {
179+
pub version: Option<String>,
180+
pub size: String,
181+
pub used_size: Option<String>,
182+
pub available_size: Option<String>,
183+
pub oldest_date_time_entry: Option<String>,
186184
}
185+
186+
let stream_hot_tier: StreamHotTierTemp = serde_json::from_slice(&bytes)?;
187+
let mut stream_hot_tier = StreamHotTier {
188+
version: stream_hot_tier
189+
.version
190+
.unwrap_or_else(|| CURRENT_HOT_TIER_VERSION.to_owned()),
191+
size: human_size_to_bytes(&stream_hot_tier.size).unwrap(),
192+
used_size: stream_hot_tier
193+
.used_size
194+
.as_ref()
195+
.map(|size| human_size_to_bytes(size).unwrap()),
196+
available_size: stream_hot_tier
197+
.available_size
198+
.as_ref()
199+
.map(|size| human_size_to_bytes(size).unwrap()),
200+
oldest_date_time_entry: stream_hot_tier.oldest_date_time_entry,
201+
};
202+
203+
let oldest_date_time_entry = self.get_oldest_date_time_entry(stream).await?;
204+
stream_hot_tier.oldest_date_time_entry = oldest_date_time_entry;
205+
206+
Ok(stream_hot_tier)
187207
}
188208

189209
pub async fn delete_hot_tier(&self, stream: &str) -> Result<(), HotTierError> {
@@ -260,12 +280,7 @@ impl HotTierManager {
260280
/// delete the files from the hot tier directory if the available date range is outside the hot tier range
261281
async fn process_stream(&self, stream: String) -> Result<(), HotTierError> {
262282
let stream_hot_tier = self.get_hot_tier(&stream).await?;
263-
let mut parquet_file_size = stream_hot_tier
264-
.used_size
265-
.as_ref()
266-
.unwrap()
267-
.parse::<u64>()
268-
.unwrap();
283+
let mut parquet_file_size = *stream_hot_tier.used_size.as_ref().unwrap();
269284

270285
let object_store = CONFIG.storage().get_object_store();
271286
let mut s3_manifest_file_list = object_store.list_manifest_files(&stream).await?;
@@ -357,13 +372,7 @@ impl HotTierManager {
357372
let mut file_processed = false;
358373
let mut stream_hot_tier = self.get_hot_tier(stream).await?;
359374
if !self.is_disk_available(parquet_file.file_size).await?
360-
|| stream_hot_tier
361-
.available_size
362-
.as_ref()
363-
.unwrap()
364-
.parse::<u64>()
365-
.unwrap()
366-
<= parquet_file.file_size
375+
|| stream_hot_tier.available_size.as_ref().unwrap() <= &parquet_file.file_size
367376
{
368377
if !self
369378
.cleanup_hot_tier_old_data(
@@ -376,12 +385,7 @@ impl HotTierManager {
376385
{
377386
return Ok(file_processed);
378387
}
379-
*parquet_file_size = stream_hot_tier
380-
.used_size
381-
.as_ref()
382-
.unwrap()
383-
.parse::<u64>()
384-
.unwrap();
388+
*parquet_file_size = *stream_hot_tier.used_size.as_ref().unwrap();
385389
}
386390
let parquet_file_path = RelativePathBuf::from(parquet_file.file_path.clone());
387391
fs::create_dir_all(parquet_path.parent().unwrap()).await?;
@@ -393,18 +397,10 @@ impl HotTierManager {
393397
.await?;
394398
file.write_all(&parquet_data).await?;
395399
*parquet_file_size += parquet_file.file_size;
396-
stream_hot_tier.used_size = Some(parquet_file_size.to_string());
400+
stream_hot_tier.used_size = Some(*parquet_file_size);
397401

398-
stream_hot_tier.available_size = Some(
399-
(stream_hot_tier
400-
.available_size
401-
.as_ref()
402-
.unwrap()
403-
.parse::<u64>()
404-
.unwrap()
405-
- parquet_file.file_size)
406-
.to_string(),
407-
);
402+
stream_hot_tier.available_size =
403+
Some(stream_hot_tier.available_size.as_ref().unwrap() - parquet_file.file_size);
408404
self.put_hot_tier(stream, &mut stream_hot_tier).await?;
409405
file_processed = true;
410406
let mut hot_tier_manifest = self
@@ -614,37 +610,14 @@ impl HotTierManager {
614610
fs::remove_dir_all(path_to_delete.parent().unwrap()).await?;
615611
delete_empty_directory_hot_tier(path_to_delete.parent().unwrap()).await?;
616612

617-
stream_hot_tier.used_size = Some(
618-
(stream_hot_tier
619-
.used_size
620-
.as_ref()
621-
.unwrap()
622-
.parse::<u64>()
623-
.unwrap()
624-
- file_size)
625-
.to_string(),
626-
);
627-
stream_hot_tier.available_size = Some(
628-
(stream_hot_tier
629-
.available_size
630-
.as_ref()
631-
.unwrap()
632-
.parse::<u64>()
633-
.unwrap()
634-
+ file_size)
635-
.to_string(),
636-
);
613+
stream_hot_tier.used_size =
614+
Some(stream_hot_tier.used_size.as_ref().unwrap() - file_size);
615+
stream_hot_tier.available_size =
616+
Some(stream_hot_tier.available_size.as_ref().unwrap() + file_size);
637617
self.put_hot_tier(stream, stream_hot_tier).await?;
638618
delete_successful = true;
639619

640-
if stream_hot_tier
641-
.available_size
642-
.as_ref()
643-
.unwrap()
644-
.parse::<u64>()
645-
.unwrap()
646-
<= parquet_file_size
647-
{
620+
if stream_hot_tier.available_size.as_ref().unwrap() <= &parquet_file_size {
648621
continue 'loop_files;
649622
} else {
650623
break 'loop_dates;
@@ -738,10 +711,10 @@ impl HotTierManager {
738711
&& !self.check_stream_hot_tier_exists(INTERNAL_STREAM_NAME)
739712
{
740713
let mut stream_hot_tier = StreamHotTier {
741-
version: Some(CURRENT_HOT_TIER_VERSION.to_string()),
742-
size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string(),
743-
used_size: Some("0".to_string()),
744-
available_size: Some(INTERNAL_STREAM_HOT_TIER_SIZE_BYTES.to_string()),
714+
version: CURRENT_HOT_TIER_VERSION.to_string(),
715+
size: INTERNAL_STREAM_HOT_TIER_SIZE_BYTES,
716+
used_size: Some(0),
717+
available_size: Some(INTERNAL_STREAM_HOT_TIER_SIZE_BYTES),
745718
oldest_date_time_entry: None,
746719
};
747720
self.put_hot_tier(INTERNAL_STREAM_NAME, &mut stream_hot_tier)

‎server/src/migration/mod.rs

Lines changed: 3 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,9 @@ mod stream_metadata_migration;
2424
use std::{fs::OpenOptions, sync::Arc};
2525

2626
use crate::{
27-
hottier::{HotTierManager, CURRENT_HOT_TIER_VERSION},
27+
hottier::HotTierManager,
2828
metadata::load_stream_metadata_on_server_start,
29-
option::{validation::human_size_to_bytes, Config, Mode, CONFIG},
29+
option::{Config, Mode, CONFIG},
3030
storage::{
3131
object_storage::{parseable_json_path, stream_json_path},
3232
ObjectStorage, ObjectStorageError, PARSEABLE_METADATA_FILE_NAME, PARSEABLE_ROOT_DIRECTORY,
@@ -129,26 +129,7 @@ pub async fn run_migration(config: &Config) -> anyhow::Result<()> {
129129
async fn migration_hot_tier(stream: &str) -> anyhow::Result<()> {
130130
if let Some(hot_tier_manager) = HotTierManager::global() {
131131
if hot_tier_manager.check_stream_hot_tier_exists(stream) {
132-
let mut stream_hot_tier = hot_tier_manager.get_hot_tier(stream).await?;
133-
if stream_hot_tier.version.is_none() {
134-
stream_hot_tier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
135-
stream_hot_tier.size = human_size_to_bytes(&stream_hot_tier.size)
136-
.unwrap()
137-
.to_string();
138-
stream_hot_tier.available_size = Some(
139-
human_size_to_bytes(&stream_hot_tier.available_size.unwrap())
140-
.unwrap()
141-
.to_string(),
142-
);
143-
stream_hot_tier.used_size = Some(
144-
human_size_to_bytes(&stream_hot_tier.used_size.unwrap())
145-
.unwrap()
146-
.to_string(),
147-
);
148-
hot_tier_manager
149-
.put_hot_tier(stream, &mut stream_hot_tier)
150-
.await?;
151-
}
132+
hot_tier_manager.get_hot_tier(stream).await?;
152133
}
153134
}
154135
Ok(())

‎server/src/validator.rs

Lines changed: 6 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -131,17 +131,13 @@ pub fn user_name(username: &str) -> Result<(), UsernameValidationError> {
131131
Ok(())
132132
}
133133

134-
pub fn hot_tier(size: &str) -> Result<(), HotTierValidationError> {
135-
if let Ok(size) = size.parse::<u64>() {
136-
if size < MIN_STREAM_HOT_TIER_SIZE_BYTES {
137-
return Err(HotTierValidationError::Size(bytes_to_human_size(
138-
MIN_STREAM_HOT_TIER_SIZE_BYTES,
139-
)));
140-
}
141-
Ok(())
142-
} else {
143-
Err(HotTierValidationError::InvalidFormat)
134+
pub fn hot_tier(size: u64) -> Result<(), HotTierValidationError> {
135+
if size < MIN_STREAM_HOT_TIER_SIZE_BYTES {
136+
return Err(HotTierValidationError::Size(bytes_to_human_size(
137+
MIN_STREAM_HOT_TIER_SIZE_BYTES,
138+
)));
144139
}
140+
Ok(())
145141
}
146142
pub mod error {
147143

@@ -187,9 +183,6 @@ pub mod error {
187183

188184
#[derive(Debug, thiserror::Error)]
189185
pub enum HotTierValidationError {
190-
#[error("Please provide size in bytes")]
191-
InvalidFormat,
192-
193186
#[error("Stream should have atleast {0} size")]
194187
Size(String),
195188

0 commit comments

Comments
 (0)