Add option for keeping local cache of recent data. (#585)

This PR adds local cache / hot tier for Parseable.

This option can be enabled by setting following env vars
P_CACHE_DIR - Local Path for file cache
P_CACHE_SIZE - Size for cache in human readable size ( mb/mib/gib/gb )

When these flags are set, sync flow will move the parquet from staging into
the specified cache directory instead of deleting it. Any LRU cached entry is
deleted to satisfy the cache constraint upon insertion. LocalCacheManager is
responsible for updating and persisting this data structure.

Author: trueleo

.gitignore

@@ -1,6 +1,7 @@
 target
 data
 staging
+limitcache
 examples
 cert.pem
 key.pem

Cargo.lock

@@ -96,10 +96,12 @@ xxhash-rust = { version = "0.8", features = ["xxh3"] }
 xz2 = { version = "*", features = ["static"] }
 nom = "7.1.3"
 humantime = "2.1.0"
+human-size = "0.4"
 openid = { version = "0.12.0", default-features = false, features = ["rustls"] }
 url = "2.4.0"
 http-auth-basic = "0.3.3"
 serde_repr = "0.1.17"
+hashlru = { version = "0.11.0", features = ["serde"] }
 
 [build-dependencies]
 cargo_toml = "0.15"

server/Cargo.toml

@@ -18,6 +18,7 @@
  */
 
 use crossterm::style::Stylize;
+use human_size::SpecificSize;
 
 use crate::about;
 use crate::utils::uid::Uid;
@@ -51,7 +52,13 @@ fn print_ascii_art() {
 }
 
 fn status_info(config: &Config, scheme: &str, id: Uid) {
-    let url = format!("\"{}://{}\"", scheme, config.parseable.address).underlined();
+    let address = format!(
+        "\"{}://{}\" ({}), \":{}\" (gRPC)",
+        scheme,
+        config.parseable.address,
+        scheme.to_ascii_uppercase(),
+        config.parseable.grpc_port
+    );
     let mut credentials =
         String::from("\"As set in P_USERNAME and P_PASSWORD environment variables\"");
 
@@ -67,12 +74,12 @@ fn status_info(config: &Config, scheme: &str, id: Uid) {
     eprintln!(
         "
     {}
-        URL:                {}
+        Address:            {}
         Credentials:        {}
         Deployment UID:     \"{}\"
-        LLM Status:         \"{}\"",
+        LLM:                \"{}\"",
         "Server:".to_string().bold(),
-        url,
+        address,
         credentials,
         id.to_string(),
         llm_status
@@ -100,5 +107,21 @@ async fn storage_info(config: &Config) {
         config.staging_dir().to_string_lossy(),
         storage.get_endpoint(),
         latency
-    )
+    );
+
+    if let Some(path) = &config.parseable.local_cache_path {
+        let size: SpecificSize<human_size::Gigabyte> =
+            SpecificSize::new(config.parseable.local_cache_size as f64, human_size::Byte)
+                .unwrap()
+                .into();
+
+        eprintln!(
+            "\
+    {:8}Cache:              \"{}\"
+        Cache Size:         \"{}\"",
+            "",
+            path.display(),
+            size
+        );
+    }
 }

server/src/banner.rs

@@ -200,6 +200,21 @@ pub fn configure_routes(
                         .to(logstream::get_retention)
                         .authorize_for_stream(Action::GetRetention),
                 ),
+        )
+        .service(
+            web::resource("/cache")
+                // PUT "/logstream/{logstream}/cache" ==> Set retention for given logstream
+                .route(
+                    web::put()
+                        .to(logstream::put_enable_cache)
+                        .authorize_for_stream(Action::PutCacheEnabled),
+                )
+                // GET "/logstream/{logstream}/cache" ==> Get retention for given logstream
+                .route(
+                    web::get()
+                        .to(logstream::get_cache_enabled)
+                        .authorize_for_stream(Action::GetCacheEnabled),
+                ),
         );
 
     // User API

server/src/handlers/http.rs

@@ -227,6 +227,33 @@ pub async fn put_retention(
     ))
 }
 
+pub async fn get_cache_enabled(req: HttpRequest) -> Result<impl Responder, StreamError> {
+    let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
+    let cache_enabled = STREAM_INFO.cache_enabled(&stream_name)?;
+    Ok((web::Json(cache_enabled), StatusCode::OK))
+}
+
+pub async fn put_enable_cache(
+    req: HttpRequest,
+    body: web::Json<bool>,
+) -> Result<impl Responder, StreamError> {
+    let enable_cache = body.into_inner();
+    let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
+    let storage = CONFIG.storage().get_object_store();
+
+    let mut stream_metadata = storage.get_stream_metadata(&stream_name).await?;
+    stream_metadata.cache_enabled = enable_cache;
+    storage
+        .put_stream_manifest(&stream_name, &stream_metadata)
+        .await?;
+
+    STREAM_INFO.set_stream_cache(&stream_name, enable_cache)?;
+    Ok((
+        format!("Cache setting updated for log stream {stream_name}"),
+        StatusCode::OK,
+    ))
+}
+
 pub async fn get_stats(req: HttpRequest) -> Result<impl Responder, StreamError> {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();