add ability to store filters and dashboards in memory

Author: Eshanatnight

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

@@ -22,6 +22,8 @@ use crate::handlers::http::middleware::RouteExt;
 use crate::handlers::http::{base_path, cross_origin_config, API_BASE_PATH, API_VERSION};
 
 use crate::rbac::role::Action;
+use crate::users::dashboards::DASHBOARDS;
+use crate::users::filters::FILTERS;
 use crate::{analytics, banner, metadata, metrics, migration, rbac, storage};
 use actix_web::web;
 use actix_web::web::ServiceConfig;
@@ -175,6 +177,10 @@ impl QueryServer {
             log::warn!("could not populate local metadata. {:?}", e);
         }
 
+        FILTERS.load().await?;
+        DASHBOARDS.load().await?;
+
+
         // load data from stats back to prometheus metrics
         metrics::fetch_stats_from_storage().await;
         metrics::reset_daily_metric_from_global();

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

@@ -34,6 +34,8 @@ use crate::migration;
 use crate::rbac;
 use crate::storage;
 use crate::sync;
+use crate::users::dashboards::DASHBOARDS;
+use crate::users::filters::FILTERS;
 use std::sync::Arc;
 
 use actix_web::web::resource;
@@ -470,6 +472,10 @@ impl Server {
             log::warn!("could not populate local metadata. {:?}", err);
         }
 
+        FILTERS.load().await?;
+        DASHBOARDS.load().await?;
+
+
         metrics::fetch_stats_from_storage().await;
         metrics::reset_daily_metric_from_global();
         storage::retention::load_retention_from_global();

server/src/handlers/http/users/dashboards.rs

@@ -2,6 +2,7 @@ use crate::{
     handlers::http::ingest::PostError,
     option::CONFIG,
     storage::{object_storage::dashboard_path, ObjectStorageError},
+    users::dashboards::{Dashboard, DASHBOARDS},
 };
 use actix_web::{http::header::ContentType, web, HttpRequest, HttpResponse, Responder};
 use bytes::Bytes;
@@ -44,13 +45,18 @@ pub async fn get(req: HttpRequest) -> Result<impl Responder, DashboardError> {
         .get("dashboard_id")
         .ok_or(DashboardError::Metadata("No Dashboard Id Provided"))?;
 
+    if let Some(dashboard) = DASHBOARDS.find(dash_id) {
+        return Ok((web::Json(dashboard), StatusCode::OK));
+    }
+
+    //if dashboard is not in memory fetch from s3
     let dash_file_path = dashboard_path(user_id, &format!("{}.json", dash_id));
     let resource = CONFIG
         .storage()
         .get_object_store()
         .get_object(&dash_file_path)
         .await?;
-    let resource = serde_json::from_slice::<JsonValue>(&resource)?;
+    let resource = serde_json::from_slice::<Dashboard>(&resource)?;
 
     Ok((web::Json(resource), StatusCode::OK))
 }
@@ -68,6 +74,9 @@ pub async fn post(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostErr
 
     let dash_file_path = dashboard_path(user_id, &format!("{}.json", dash_id));
 
+    let dashboard = serde_json::from_slice::<Dashboard>(&body)?;
+    DASHBOARDS.update(dashboard);
+
     let store = CONFIG.storage().get_object_store();
     store.put_object(&dash_file_path, body).await?;
 
@@ -93,32 +102,6 @@ pub async fn delete(req: HttpRequest) -> Result<HttpResponse, PostError> {
     Ok(HttpResponse::Ok().finish())
 }
 
-// #[derive(Debug, Serialize, Deserialize)]
-// pub struct Dashboard {
-//     version: String,
-//     name: String,
-//     id: String,
-//     time_filter: TimeFilter
-//     refresh_interval: u64,
-//     pannels: Vec<Pannel>,
-// }
-//
-// #[derive(Debug, Serialize, Deserialize)]
-// pub struct Pannel {
-//     stream_name: String,
-//     query: String,
-//     chart_type: String,
-//     columns: Vec<String>,
-//     headers: Vec<String>,
-//     dimensions: (u64, u64),
-// }
-//
-// #[derive(Debug, Serialize, Deserialize)]
-// pub struct TimeFilter {
-//      to: String,
-//      from: String
-// }
-
 #[derive(Debug, thiserror::Error)]
 pub enum DashboardError {
     #[error("Failed to connect to storage: {0}")]

server/src/handlers/http/users/filters.rs

@@ -2,6 +2,7 @@ use crate::{
     handlers::{http::ingest::PostError, STREAM_NAME_HEADER_KEY},
     option::CONFIG,
     storage::{object_storage::filter_path, ObjectStorageError},
+    users::filters::{Filter, FILTERS},
 };
 use actix_web::{http::header::ContentType, web, HttpRequest, HttpResponse, Responder};
 use bytes::Bytes;
@@ -62,14 +63,19 @@ pub async fn get(req: HttpRequest) -> Result<impl Responder, FiltersError> {
         .to_str()
         .map_err(|_| FiltersError::Metadata("Non ASCII Stream Name Provided"))?;
 
+    if let Some(filter) = FILTERS.find(filt_id) {
+        return Ok((web::Json(filter), StatusCode::OK));
+    }
+
+    // if it is not in memory go to s3
     let path = filter_path(user_id, stream_name, &format!("{}.json", filt_id));
     let resource = CONFIG
         .storage()
         .get_object_store()
         .get_object(&path)
         .await?;
 
-    let resource = serde_json::from_slice::<JsonValue>(&resource)?;
+    let resource = serde_json::from_slice::<Filter>(&resource)?;
 
     Ok((web::Json(resource), StatusCode::OK))
 }
@@ -95,6 +101,8 @@ pub async fn post(req: HttpRequest, body: Bytes) -> Result<HttpResponse, PostErr
         .map_err(|_| FiltersError::Metadata("Non ASCII Stream Name Provided"))?;
 
     let path = filter_path(user_id, stream_name, &format!("{}.json", filt_id));
+    let filter: Filter = serde_json::from_slice(&body)?;
+    FILTERS.update(filter);
 
     let store = CONFIG.storage().get_object_store();
     store.put_object(&path, body).await?;
@@ -154,12 +162,3 @@ impl actix_web::ResponseError for FiltersError {
             .body(self.to_string())
     }
 }
-
-// #[derive(Debug, Serialize, Deserialize)]
-// pub struct Filters {
-//     version: String,
-//     stream_name: String,
-//     filter_name: String,
-//     query: String,
-//     time-filter: TimeFilter
-// }

server/src/main.rs

@@ -39,6 +39,7 @@ mod static_schema;
 mod stats;
 mod storage;
 mod sync;
+mod users;
 mod utils;
 mod validator;
 

server/src/users/dashboards.rs

@@ -0,0 +1,77 @@
+use std::sync::RwLock;
+
+use once_cell::sync::Lazy;
+use relative_path::RelativePathBuf;
+use serde::{Deserialize, Serialize};
+
+use crate::{handlers::http::users::USERS_ROOT_DIR, metadata::LOCK_EXPECT, option::CONFIG};
+
+use super::TimeFilter;
+
+pub static DASHBOARDS: Lazy<Dashboards> = Lazy::new(Dashboards::default);
+
+#[derive(Debug, Serialize, Deserialize, Default, Clone)]
+pub struct Pannel {
+    stream_name: String,
+    query: String,
+    chart_type: String,
+    columns: Vec<String>,
+    headers: Vec<String>,
+    dimensions: (u64, u64),
+}
+
+#[derive(Debug, Serialize, Deserialize, Default, Clone)]
+pub struct Dashboard {
+    version: String,
+    name: String,
+    id: String,
+    time_filter: TimeFilter,
+    refresh_interval: u64,
+    pannels: Vec<Pannel>,
+}
+
+impl Dashboard {
+    pub fn dashboard_id(&self) -> &str {
+        &self.id
+    }
+}
+
+#[derive(Default)]
+pub struct Dashboards(RwLock<Vec<Dashboard>>);
+
+impl Dashboards {
+    pub async fn load(&self) -> anyhow::Result<()> {
+        let mut this = vec![];
+        let path = RelativePathBuf::from(USERS_ROOT_DIR);
+        let store = CONFIG.storage().get_object_store();
+        let objs = store
+            .get_objects(Some(&path), Box::new(|path| path.ends_with(".json")))
+            .await?;
+
+        for obj in objs {
+            if let Ok(filter) = serde_json::from_slice::<Dashboard>(&obj) {
+                this.push(filter);
+            }
+        }
+
+        let mut s = self.0.write().expect(LOCK_EXPECT);
+        s.append(&mut this);
+
+        Ok(())
+    }
+
+    pub fn update(&self, dashboard: Dashboard) {
+        let mut s = self.0.write().expect(LOCK_EXPECT);
+
+        s.push(dashboard);
+    }
+
+    pub fn find(&self, dashboard_id: &str) -> Option<Dashboard> {
+        self.0
+            .read()
+            .expect(LOCK_EXPECT)
+            .iter()
+            .find(|dashboard| dashboard.dashboard_id() == dashboard_id)
+            .cloned()
+    }
+}

server/src/users/filters.rs

@@ -0,0 +1,66 @@
+use std::sync::RwLock;
+
+use once_cell::sync::Lazy;
+use relative_path::RelativePathBuf;
+use serde::{Deserialize, Serialize};
+
+use super::TimeFilter;
+use crate::{handlers::http::users::USERS_ROOT_DIR, metadata::LOCK_EXPECT, option::CONFIG};
+
+pub static FILTERS: Lazy<Filters> = Lazy::new(Filters::default);
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct Filter {
+    version: String,
+    stream_name: String,
+    filter_name: String,
+    filter_id: String,
+    query: String,
+    time_filter: Option<TimeFilter>,
+}
+
+impl Filter {
+    pub fn filter_id(&self) -> &str {
+        &self.filter_id
+    }
+}
+
+#[derive(Debug, Default)]
+pub struct Filters(RwLock<Vec<Filter>>);
+
+impl Filters {
+    pub async fn load(&self) -> anyhow::Result<()> {
+        let mut this = vec![];
+        let path = RelativePathBuf::from(USERS_ROOT_DIR);
+        let store = CONFIG.storage().get_object_store();
+        let objs = store
+            .get_objects(Some(&path), Box::new(|path| path.ends_with(".json")))
+            .await?;
+
+        for obj in objs {
+            if let Ok(filter) = serde_json::from_slice::<Filter>(&obj) {
+                this.push(filter);
+            }
+        }
+
+        let mut s = self.0.write().expect(LOCK_EXPECT);
+        s.append(&mut this);
+
+        Ok(())
+    }
+
+    pub fn update(&self, filter: Filter) {
+        let mut s = self.0.write().expect(LOCK_EXPECT);
+
+        s.push(filter);
+    }
+
+    pub fn find(&self, filter_id: &str) -> Option<Filter> {
+        self.0
+            .read()
+            .expect(LOCK_EXPECT)
+            .iter()
+            .find(|filter| filter.filter_id() == filter_id)
+            .cloned()
+    }
+}

server/src/users/mod.rs

@@ -0,0 +1,10 @@
+pub mod dashboards;
+pub mod filters;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize, Default, Clone)]
+pub struct TimeFilter {
+    to: String,
+    from: String,
+}