Skip to content

Commit 57d50a1

Browse files
nikhilsinhaparseablenikhilsinhaparseable
authored
Hot tier standalone (#943)
1 parent 2a2d64f commit 57d50a1

File tree

6 files changed

+128
-113
lines changed

6 files changed

+128
-113
lines changed

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

Lines changed: 105 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ use super::ingest::create_stream_if_not_exists;
2323
use super::modal::utils::logstream_utils::create_update_stream;
2424
use crate::alerts::Alerts;
2525
use crate::handlers::STREAM_TYPE_KEY;
26-
use crate::hottier::HotTierManager;
26+
use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
2727
use crate::metadata::STREAM_INFO;
2828
use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
2929
use crate::option::CONFIG;
@@ -512,6 +512,110 @@ pub async fn get_stream_info(req: HttpRequest) -> Result<impl Responder, StreamE
512512
Ok((web::Json(stream_info), StatusCode::OK))
513513
}
514514

515+
pub async fn put_stream_hot_tier(
516+
req: HttpRequest,
517+
body: web::Json<serde_json::Value>,
518+
) -> Result<impl Responder, StreamError> {
519+
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
520+
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
521+
return Err(StreamError::StreamNotFound(stream_name));
522+
}
523+
524+
if STREAM_INFO.stream_type(&stream_name).unwrap() == Some(StreamType::Internal.to_string()) {
525+
return Err(StreamError::Custom {
526+
msg: "Hot tier can not be updated for internal stream".to_string(),
527+
status: StatusCode::BAD_REQUEST,
528+
});
529+
}
530+
if CONFIG.parseable.hot_tier_storage_path.is_none() {
531+
return Err(StreamError::HotTierNotEnabled(stream_name));
532+
}
533+
534+
let body = body.into_inner();
535+
let mut hottier: StreamHotTier = match serde_json::from_value(body) {
536+
Ok(hottier) => hottier,
537+
Err(err) => return Err(StreamError::InvalidHotTierConfig(err)),
538+
};
539+
540+
validator::hot_tier(&hottier.size.to_string())?;
541+
542+
STREAM_INFO.set_hot_tier(&stream_name, true)?;
543+
if let Some(hot_tier_manager) = HotTierManager::global() {
544+
let existing_hot_tier_used_size = hot_tier_manager
545+
.validate_hot_tier_size(&stream_name, &hottier.size)
546+
.await?;
547+
hottier.used_size = Some(existing_hot_tier_used_size.to_string());
548+
hottier.available_size = Some(hottier.size.clone());
549+
hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
550+
hot_tier_manager
551+
.put_hot_tier(&stream_name, &mut hottier)
552+
.await?;
553+
let storage = CONFIG.storage().get_object_store();
554+
let mut stream_metadata = storage.get_object_store_format(&stream_name).await?;
555+
stream_metadata.hot_tier_enabled = Some(true);
556+
storage
557+
.put_stream_manifest(&stream_name, &stream_metadata)
558+
.await?;
559+
}
560+
561+
Ok((
562+
format!("hot tier set for stream {stream_name}"),
563+
StatusCode::OK,
564+
))
565+
}
566+
567+
pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, StreamError> {
568+
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
569+
570+
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
571+
return Err(StreamError::StreamNotFound(stream_name));
572+
}
573+
574+
if CONFIG.parseable.hot_tier_storage_path.is_none() {
575+
return Err(StreamError::HotTierNotEnabled(stream_name));
576+
}
577+
578+
if let Some(hot_tier_manager) = HotTierManager::global() {
579+
let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
580+
hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
581+
hot_tier.used_size = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
582+
hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
583+
Ok((web::Json(hot_tier), StatusCode::OK))
584+
} else {
585+
Err(StreamError::Custom {
586+
msg: format!("hot tier not initialised for stream {}", stream_name),
587+
status: (StatusCode::BAD_REQUEST),
588+
})
589+
}
590+
}
591+
592+
pub async fn delete_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, StreamError> {
593+
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
594+
595+
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
596+
return Err(StreamError::StreamNotFound(stream_name));
597+
}
598+
599+
if CONFIG.parseable.hot_tier_storage_path.is_none() {
600+
return Err(StreamError::HotTierNotEnabled(stream_name));
601+
}
602+
603+
if STREAM_INFO.stream_type(&stream_name).unwrap() == Some(StreamType::Internal.to_string()) {
604+
return Err(StreamError::Custom {
605+
msg: "Hot tier can not be deleted for internal stream".to_string(),
606+
status: StatusCode::BAD_REQUEST,
607+
});
608+
}
609+
610+
if let Some(hot_tier_manager) = HotTierManager::global() {
611+
hot_tier_manager.delete_hot_tier(&stream_name).await?;
612+
}
613+
Ok((
614+
format!("hot tier deleted for stream {stream_name}"),
615+
StatusCode::OK,
616+
))
617+
}
618+
515619
pub async fn create_internal_stream_if_not_exists() -> Result<(), StreamError> {
516620
if let Ok(stream_exists) =
517621
create_stream_if_not_exists(INTERNAL_STREAM_NAME, &StreamType::Internal.to_string()).await

‎server/src/handlers/http/modal/query/querier_logstream.rs

Lines changed: 1 addition & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,11 @@ use crate::{
1717
logstream::{error::StreamError, get_stats_date},
1818
modal::utils::logstream_utils::create_update_stream,
1919
},
20-
hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION},
20+
hottier::HotTierManager,
2121
metadata::{self, STREAM_INFO},
2222
option::CONFIG,
2323
stats::{self, Stats},
2424
storage::{StorageDir, StreamType},
25-
validator,
2625
};
2726

2827
pub async fn delete(req: HttpRequest) -> Result<impl Responder, StreamError> {
@@ -218,107 +217,3 @@ pub async fn get_cache_enabled(req: HttpRequest) -> Result<impl Responder, Strea
218217
let cache_enabled = STREAM_INFO.get_cache_enabled(&stream_name)?;
219218
Ok((web::Json(cache_enabled), StatusCode::OK))
220219
}
221-
222-
pub async fn put_stream_hot_tier(
223-
req: HttpRequest,
224-
body: web::Json<serde_json::Value>,
225-
) -> Result<impl Responder, StreamError> {
226-
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
227-
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
228-
return Err(StreamError::StreamNotFound(stream_name));
229-
}
230-
231-
if STREAM_INFO.stream_type(&stream_name).unwrap() == Some(StreamType::Internal.to_string()) {
232-
return Err(StreamError::Custom {
233-
msg: "Hot tier can not be updated for internal stream".to_string(),
234-
status: StatusCode::BAD_REQUEST,
235-
});
236-
}
237-
if CONFIG.parseable.hot_tier_storage_path.is_none() {
238-
return Err(StreamError::HotTierNotEnabled(stream_name));
239-
}
240-
241-
let body = body.into_inner();
242-
let mut hottier: StreamHotTier = match serde_json::from_value(body) {
243-
Ok(hottier) => hottier,
244-
Err(err) => return Err(StreamError::InvalidHotTierConfig(err)),
245-
};
246-
247-
validator::hot_tier(&hottier.size.to_string())?;
248-
249-
STREAM_INFO.set_hot_tier(&stream_name, true)?;
250-
if let Some(hot_tier_manager) = HotTierManager::global() {
251-
let existing_hot_tier_used_size = hot_tier_manager
252-
.validate_hot_tier_size(&stream_name, &hottier.size)
253-
.await?;
254-
hottier.used_size = Some(existing_hot_tier_used_size.to_string());
255-
hottier.available_size = Some(hottier.size.clone());
256-
hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
257-
hot_tier_manager
258-
.put_hot_tier(&stream_name, &mut hottier)
259-
.await?;
260-
let storage = CONFIG.storage().get_object_store();
261-
let mut stream_metadata = storage.get_object_store_format(&stream_name).await?;
262-
stream_metadata.hot_tier_enabled = Some(true);
263-
storage
264-
.put_stream_manifest(&stream_name, &stream_metadata)
265-
.await?;
266-
}
267-
268-
Ok((
269-
format!("hot tier set for stream {stream_name}"),
270-
StatusCode::OK,
271-
))
272-
}
273-
274-
pub async fn get_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, StreamError> {
275-
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
276-
277-
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
278-
return Err(StreamError::StreamNotFound(stream_name));
279-
}
280-
281-
if CONFIG.parseable.hot_tier_storage_path.is_none() {
282-
return Err(StreamError::HotTierNotEnabled(stream_name));
283-
}
284-
285-
if let Some(hot_tier_manager) = HotTierManager::global() {
286-
let mut hot_tier = hot_tier_manager.get_hot_tier(&stream_name).await?;
287-
hot_tier.size = format!("{} {}", hot_tier.size, "Bytes");
288-
hot_tier.used_size = Some(format!("{} {}", hot_tier.used_size.unwrap(), "Bytes"));
289-
hot_tier.available_size = Some(format!("{} {}", hot_tier.available_size.unwrap(), "Bytes"));
290-
Ok((web::Json(hot_tier), StatusCode::OK))
291-
} else {
292-
Err(StreamError::Custom {
293-
msg: format!("hot tier not initialised for stream {}", stream_name),
294-
status: (StatusCode::BAD_REQUEST),
295-
})
296-
}
297-
}
298-
299-
pub async fn delete_stream_hot_tier(req: HttpRequest) -> Result<impl Responder, StreamError> {
300-
let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
301-
302-
if !metadata::STREAM_INFO.stream_exists(&stream_name) {
303-
return Err(StreamError::StreamNotFound(stream_name));
304-
}
305-
306-
if CONFIG.parseable.hot_tier_storage_path.is_none() {
307-
return Err(StreamError::HotTierNotEnabled(stream_name));
308-
}
309-
310-
if STREAM_INFO.stream_type(&stream_name).unwrap() == Some(StreamType::Internal.to_string()) {
311-
return Err(StreamError::Custom {
312-
msg: "Hot tier can not be deleted for internal stream".to_string(),
313-
status: StatusCode::BAD_REQUEST,
314-
});
315-
}
316-
317-
if let Some(hot_tier_manager) = HotTierManager::global() {
318-
hot_tier_manager.delete_hot_tier(&stream_name).await?;
319-
}
320-
Ok((
321-
format!("hot tier deleted for stream {stream_name}"),
322-
StatusCode::OK,
323-
))
324-
}

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -356,17 +356,17 @@ impl QueryServer {
356356
// PUT "/logstream/{logstream}/hottier" ==> Set hottier for given logstream
357357
.route(
358358
web::put()
359-
.to(querier_logstream::put_stream_hot_tier)
359+
.to(logstream::put_stream_hot_tier)
360360
.authorize_for_stream(Action::PutHotTierEnabled),
361361
)
362362
.route(
363363
web::get()
364-
.to(querier_logstream::get_stream_hot_tier)
364+
.to(logstream::get_stream_hot_tier)
365365
.authorize_for_stream(Action::GetHotTierEnabled),
366366
)
367367
.route(
368368
web::delete()
369-
.to(querier_logstream::delete_stream_hot_tier)
369+
.to(logstream::delete_stream_hot_tier)
370370
.authorize_for_stream(Action::DeleteHotTierEnabled),
371371
),
372372
),

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ use crate::handlers::http::users::dashboards;
2929
use crate::handlers::http::users::filters;
3030
use crate::handlers::http::API_BASE_PATH;
3131
use crate::handlers::http::API_VERSION;
32+
use crate::hottier::HotTierManager;
3233
use crate::localcache::LocalCacheManager;
3334
use crate::metrics;
3435
use crate::migration;
@@ -554,6 +555,10 @@ impl Server {
554555

555556
storage::retention::load_retention_from_global();
556557

558+
if let Some(hot_tier_manager) = HotTierManager::global() {
559+
hot_tier_manager.download_from_s3()?;
560+
};
561+
557562
let (localsync_handler, mut localsync_outbox, localsync_inbox) =
558563
sync::run_local_sync().await;
559564
let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =

‎server/src/hottier.rs

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,10 +73,12 @@ impl HotTierManager {
7373
pub fn global() -> Option<&'static HotTierManager> {
7474
static INSTANCE: OnceCell<HotTierManager> = OnceCell::new();
7575

76-
let hot_tier_path = CONFIG.parseable.hot_tier_storage_path.as_ref()?;
77-
76+
let hot_tier_path = &CONFIG.parseable.hot_tier_storage_path;
77+
if hot_tier_path.is_none() {
78+
return None;
79+
}
7880
Some(INSTANCE.get_or_init(|| {
79-
let hot_tier_path = hot_tier_path.clone();
81+
let hot_tier_path = hot_tier_path.as_ref().unwrap().clone();
8082
std::fs::create_dir_all(&hot_tier_path).unwrap();
8183
HotTierManager {
8284
filesystem: LocalFileSystem::new(),

‎server/src/option.rs

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,15 @@ Cloud Native, log analytics platform for modern applications."#,
8585
.exit()
8686
}
8787

88+
if cli.hot_tier_storage_path.is_some() {
89+
create_parseable_cli_command()
90+
.error(
91+
ErrorKind::ValueValidation,
92+
"Cannot use hot tier with local-store subcommand.",
93+
)
94+
.exit()
95+
}
96+
8897
Config {
8998
parseable: cli,
9099
storage: Arc::new(storage),

0 commit comments

Comments
 (0)