add cache endpoints

Author: Eshanatnight

server/src/handlers/http.rs

@@ -26,6 +26,7 @@ use crate::option::CONFIG;
 use self::{cluster::get_ingestor_info, query::Query};
 
 pub(crate) mod about;
+mod cache;
 pub mod cluster;
 pub(crate) mod health_check;
 pub(crate) mod ingest;
@@ -40,7 +41,6 @@ pub(crate) mod query;
 pub(crate) mod rbac;
 pub(crate) mod role;
 pub mod users;
-
 pub const MAX_EVENT_PAYLOAD_SIZE: usize = 10485760;
 pub const API_BASE_PATH: &str = "api";
 pub const API_VERSION: &str = "v1";

server/src/handlers/http/ingest.rs

@@ -30,6 +30,7 @@ use crate::handlers::{
     LOG_SOURCE_KEY, LOG_SOURCE_KINESIS, LOG_SOURCE_OTEL, PREFIX_META, PREFIX_TAGS, SEPARATOR,
     STREAM_NAME_HEADER_KEY,
 };
+use crate::localcache::CacheError;
 use crate::metadata::{self, STREAM_INFO};
 use crate::option::{Mode, CONFIG};
 use crate::storage::{LogStream, ObjectStorageError};
@@ -445,6 +446,8 @@ pub enum PostError {
     FiltersError(#[from] FiltersError),
     #[error("Error: {0}")]
     DashboardError(#[from] DashboardError),
+    #[error("Error: {0}")]
+    CacheError(#[from] CacheError),
 }
 
 impl actix_web::ResponseError for PostError {
@@ -465,6 +468,7 @@ impl actix_web::ResponseError for PostError {
             PostError::DashboardError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::FiltersError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PostError::Error(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            PostError::CacheError(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 

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

@@ -121,6 +121,7 @@ impl QueryServer {
                 web::scope(&base_path())
                     // POST "/query" ==> Get results of the SQL query passed in request body
                     .service(Server::get_query_factory())
+                    .service(Server::get_cache_webscope())
                     .service(Server::get_liveness_factory())
                     .service(Server::get_readiness_factory())
                     .service(Server::get_about_factory())

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

@@ -21,6 +21,7 @@ use crate::banner;
 use crate::handlers;
 use crate::handlers::http::about;
 use crate::handlers::http::base_path;
+use crate::handlers::http::cache;
 use crate::handlers::http::health_check;
 use crate::handlers::http::query;
 use crate::handlers::http::users::dashboards;
@@ -137,6 +138,7 @@ impl Server {
                 web::scope(&base_path())
                     // POST "/query" ==> Get results of the SQL query passed in request body
                     .service(Self::get_query_factory())
+                    .service(Self::get_cache_webscope())
                     .service(Self::get_ingest_factory())
                     .service(Self::get_liveness_factory())
                     .service(Self::get_readiness_factory())
@@ -218,6 +220,18 @@ impl Server {
         web::resource("/query").route(web::post().to(query::query).authorize(Action::Query))
     }
 
+    pub fn get_cache_webscope() -> Scope {
+        web::scope("/cache").service(
+            web::scope("/{user_id}").service(
+                web::scope("/{stream}").service(
+                    web::resource("")
+                        .route(web::get().to(cache::list).authorize(Action::ListCache))
+                        .route(web::post().to(cache::remove).authorize(Action::RemoveCache)),
+                ),
+            ),
+        )
+    }
+
     // get the logstream web scope
     pub fn get_logstream_webscope() -> Scope {
         web::scope("/logstream")

server/src/localcache.rs

@@ -261,4 +261,6 @@ pub enum CacheError {
     ParquetError(#[from] ParquetError),
     #[error("{0}")]
     MetadataError(#[from] MetadataError),
+    #[error("Error: Cache File Does Not Exist")]
+    DoesNotExist,
 }

server/src/querycache.rs

@@ -80,6 +80,18 @@ impl QueryCache {
         self.files.get(&key).cloned()
     }
 
+    pub fn current_size(&self) -> u64 {
+        self.current_size
+    }
+
+    pub fn remove(&mut self, key: CacheMetadata) -> Option<PathBuf> {
+        self.files.remove(&key)
+    }
+
+    pub fn queries(&self) -> Vec<&CacheMetadata> {
+        self.files.keys().collect_vec()
+    }
+
     // read the parquet
     // return the recordbatches
     pub async fn get_cached_records(
@@ -238,6 +250,23 @@ impl QueryCacheManager {
         Ok(cache)
     }
 
+    pub async fn remove_from_cache(
+        &self,
+        key: CacheMetadata,
+        stream: &str,
+        user_id: &str,
+    ) -> Result<(), CacheError> {
+        let mut cache = self.get_cache(stream, user_id).await?;
+
+        if let Some(remove_result) = cache.remove(key) {
+            self.put_cache(stream, &cache, user_id).await?;
+            tokio::spawn(fs::remove_file(remove_result));
+            Ok(())
+        } else {
+            Err(CacheError::DoesNotExist)
+        }
+    }
+
     pub async fn put_cache(
         &self,
         stream: &str,

server/src/rbac/role.rs

@@ -58,6 +58,8 @@ pub enum Action {
     GetFilter,
     CreateFilter,
     DeleteFilter,
+    ListCache,
+    RemoveCache,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -123,6 +125,8 @@ impl RoleBuilder {
                 | Action::ListFilter
                 | Action::CreateFilter
                 | Action::DeleteFilter
+                | Action::ListCache
+                | Action::RemoveCache
                 | Action::GetAnalytics => Permission::Unit(action),
                 Action::Ingest
                 | Action::GetSchema