Add option for keeping local cache of recent data.

.gitignore

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

server/Cargo.toml

@@ -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/src/banner.rs

@@ -18,6 +18,7 @@
  */
 
 use crossterm::style::Stylize;
+use human_size::SpecificSize;
 
 use crate::about;
 use crate::utils::uid::Uid;
@@ -100,5 +101,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/handlers/http.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/logstream.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();