updates to prism endpoints

parmesant · parmesant · commit 6397b7166c1c · 2025-02-27T17:29:39.000+05:30
/logstream/{logstream}/info
- added stats and retention to the response body

GET /users
GET /users/{username}
diff --git a/src/handlers/http/modal/query_server.rs b/src/handlers/http/modal/query_server.rs
@@ -61,6 +61,7 @@ impl ParseableServer for QueryServer {
                     .service(Server::get_about_factory())
                     .service(Self::get_logstream_webscope())
                     .service(Self::get_user_webscope())
+                    .service(Server::get_users_webscope())
                     .service(Server::get_dashboards_webscope())
                     .service(Server::get_filters_webscope())
                     .service(Server::get_llm_webscope())
diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs
@@ -81,6 +81,7 @@ impl ParseableServer for Server {
                     .service(Self::get_about_factory())
                     .service(Self::get_logstream_webscope())
                     .service(Self::get_user_webscope())
+                    .service(Self::get_users_webscope())
                     .service(Self::get_dashboards_webscope())
                     .service(Self::get_filters_webscope())
                     .service(Self::get_llm_webscope())
@@ -170,7 +171,9 @@ impl Server {
                 web::resource("/info").route(
                     web::get()
                         .to(http::prism_logstream::get_info)
-                        .authorize_for_stream(Action::GetStreamInfo),
+                        .authorize_for_stream(Action::GetStreamInfo)
+                        .authorize_for_stream(Action::GetStats)
+                        .authorize_for_stream(Action::GetRetention),
                 ),
             ),
         )
@@ -497,6 +500,27 @@ impl Server {
             )
     }
 
+    // get the users webscope (for Prism only)
+    pub fn get_users_webscope() -> Scope {
+        web::scope("/users")
+            .service(
+                web::resource("")
+                    // GET /users => List all users
+                    .route(
+                        web::get()
+                            .to(http::rbac::list_users_prism)
+                            .authorize(Action::ListUser),
+                    ),
+            )
+            .service(
+                web::resource("/{username}").route(
+                    web::get()
+                        .to(http::rbac::get_prism_user)
+                        .authorize_for_user(Action::GetUserRoles),
+                ),
+            )
+    }
+
     // get the user webscope
     pub fn get_user_webscope() -> Scope {
         web::scope("/user")
diff --git a/src/handlers/http/rbac.rs b/src/handlers/http/rbac.rs
@@ -19,12 +19,17 @@
 use std::collections::{HashMap, HashSet};
 
 use crate::{
-    rbac::{map::roles, role::model::DefaultPrivilege, user, Users},
+    rbac::{self, map::roles, role::model::DefaultPrivilege, user, Users, UsersPrism},
     storage::ObjectStorageError,
     validator::{self, error::UsernameValidationError},
 };
-use actix_web::{http::header::ContentType, web, Responder};
+use actix_web::{
+    http::header::ContentType,
+    web::{self, Path},
+    Responder,
+};
 use http::StatusCode;
+use itertools::Itertools;
 use tokio::sync::Mutex;
 
 use super::modal::utils::rbac_utils::{get_metadata, put_metadata};
@@ -58,6 +63,88 @@ pub async fn list_users() -> impl Responder {
     web::Json(Users.collect_user::<User>())
 }
 
+/// Handler for GET /api/v1/users
+/// returns list of all registerd users along with their roles and other info
+pub async fn list_users_prism() -> impl Responder {
+    // get all users
+    let prism_users = rbac::map::users()
+        .values()
+        .map(|u| {
+            let (id, method, email, picture) = match &u.ty {
+                user::UserType::Native(_) => (u.username(), "native", None, None),
+                user::UserType::OAuth(oauth) => (
+                    u.username(),
+                    "oauth",
+                    oauth.user_info.email.clone(),
+                    oauth.user_info.picture.clone(),
+                ),
+            };
+            let roles: HashMap<String, Vec<DefaultPrivilege>> = Users
+                .get_role(id)
+                .iter()
+                .filter_map(|role_name| {
+                    roles()
+                        .get(role_name)
+                        .map(|role| (role_name.to_owned(), role.clone()))
+                })
+                .collect();
+
+            UsersPrism {
+                id: id.into(),
+                method: method.into(),
+                email,
+                picture,
+                roles,
+            }
+        })
+        .collect_vec();
+
+    web::Json(prism_users)
+}
+
+/// Function for GET /users/{username}
+pub async fn get_prism_user(username: Path<String>) -> Result<impl Responder, RBACError> {
+    let username = username.into_inner();
+    let prism_user = rbac::map::users()
+        .values()
+        .map(|u| {
+            let (id, method, email, picture) = match &u.ty {
+                user::UserType::Native(_) => (u.username(), "native", None, None),
+                user::UserType::OAuth(oauth) => (
+                    u.username(),
+                    "oauth",
+                    oauth.user_info.email.clone(),
+                    oauth.user_info.picture.clone(),
+                ),
+            };
+            let roles: HashMap<String, Vec<DefaultPrivilege>> = Users
+                .get_role(id)
+                .iter()
+                .filter_map(|role_name| {
+                    roles()
+                        .get(role_name)
+                        .map(|role| (role_name.to_owned(), role.clone()))
+                })
+                .collect();
+
+            UsersPrism {
+                id: id.into(),
+                method: method.into(),
+                email,
+                picture,
+                roles,
+            }
+        })
+        .filter(|u| u.id.eq(&username))
+        .collect_vec();
+
+    if prism_user.is_empty() {
+        Err(RBACError::UserDoesNotExist)
+    } else {
+        Ok(web::Json(prism_user[0].clone()))
+    }
+}
+
 // Handler for POST /api/v1/user/{username}
 // Creates a new user by username if it does not exists
 pub async fn post_user(
diff --git a/src/prism/logstream/mod.rs b/src/prism/logstream/mod.rs
@@ -20,31 +20,54 @@ use std::sync::Arc;
 
 use actix_web::http::header::ContentType;
 use arrow_schema::Schema;
+use chrono::Utc;
 use http::StatusCode;
 use serde::Serialize;
 
 use crate::{
-    handlers::http::{logstream::error::StreamError, query::update_schema_when_distributed},
+    handlers::http::{
+        cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats},
+        logstream::error::StreamError,
+        query::update_schema_when_distributed,
+    },
     parseable::{StreamNotFound, PARSEABLE},
-    storage::StreamInfo,
+    stats,
+    storage::{retention::Retention, StreamInfo},
     LOCK_EXPECT,
 };
 
 #[derive(Serialize)]
 pub struct PrismLogstreamInfo {
     info: StreamInfo,
     schema: Arc<Schema>,
+    stats: QueriedStats,
+    retention: Retention,
 }
 
 pub async fn get_prism_logstream_info(
     stream_name: &str,
 ) -> Result<PrismLogstreamInfo, PrismLogstreamError> {
     // get StreamInfo
     let info = get_stream_info_helper(stream_name).await?;
+
     // get stream schema
     let schema = get_stream_schema_helper(stream_name).await?;
 
-    Ok(PrismLogstreamInfo { info, schema })
+    // get stream stats
+    let stats = get_stats(stream_name).await?;
+
+    // get retention
+    let retention = PARSEABLE
+        .get_stream(stream_name)?
+        .get_retention()
+        .unwrap_or_default();
+
+    Ok(PrismLogstreamInfo {
+        info,
+        schema,
+        stats,
+        retention,
+    })
 }
 
 async fn get_stream_schema_helper(stream_name: &str) -> Result<Arc<Schema>, StreamError> {
@@ -66,6 +89,73 @@ async fn get_stream_schema_helper(stream_name: &str) -> Result<Arc<Schema>, Stre
     }
 }
 
+async fn get_stats(stream_name: &str) -> Result<QueriedStats, PrismLogstreamError> {
+    let stats = stats::get_current_stats(stream_name, "json")
+        .ok_or_else(|| StreamNotFound(stream_name.to_owned()))?;
+
+    let ingestor_stats: Option<Vec<QueriedStats>> = None;
+
+    let hash_map = PARSEABLE.streams.read().expect("Readable");
+    let stream_meta = &hash_map
+        .get(stream_name)
+        .ok_or_else(|| StreamNotFound(stream_name.to_owned()))?
+        .metadata
+        .read()
+        .expect(LOCK_EXPECT);
+
+    let time = Utc::now();
+
+    let stats = match &stream_meta.first_event_at {
+        Some(_) => {
+            let ingestion_stats = IngestionStats::new(
+                stats.current_stats.events,
+                format!("{} {}", stats.current_stats.ingestion, "Bytes"),
+                stats.lifetime_stats.events,
+                format!("{} {}", stats.lifetime_stats.ingestion, "Bytes"),
+                stats.deleted_stats.events,
+                format!("{} {}", stats.deleted_stats.ingestion, "Bytes"),
+                "json",
+            );
+            let storage_stats = StorageStats::new(
+                format!("{} {}", stats.current_stats.storage, "Bytes"),
+                format!("{} {}", stats.lifetime_stats.storage, "Bytes"),
+                format!("{} {}", stats.deleted_stats.storage, "Bytes"),
+                "parquet",
+            );
+
+            QueriedStats::new(stream_name, time, ingestion_stats, storage_stats)
+        }
+
+        None => {
+            let ingestion_stats = IngestionStats::new(
+                stats.current_stats.events,
+                format!("{} {}", stats.current_stats.ingestion, "Bytes"),
+                stats.lifetime_stats.events,
+                format!("{} {}", stats.lifetime_stats.ingestion, "Bytes"),
+                stats.deleted_stats.events,
+                format!("{} {}", stats.deleted_stats.ingestion, "Bytes"),
+                "json",
+            );
+            let storage_stats = StorageStats::new(
+                format!("{} {}", stats.current_stats.storage, "Bytes"),
+                format!("{} {}", stats.lifetime_stats.storage, "Bytes"),
+                format!("{} {}", stats.deleted_stats.storage, "Bytes"),
+                "parquet",
+            );
+
+            QueriedStats::new(stream_name, time, ingestion_stats, storage_stats)
+        }
+    };
+    let stats = if let Some(mut ingestor_stats) = ingestor_stats {
+        ingestor_stats.push(stats);
+        merge_quried_stats(ingestor_stats)
+    } else {
+        stats
+    };
+
+    Ok(stats)
+}
+
 async fn get_stream_info_helper(stream_name: &str) -> Result<StreamInfo, StreamError> {
     // For query mode, if the stream not found in memory map,
     //check if it exists in the storage
@@ -120,13 +210,16 @@ pub enum PrismLogstreamError {
     Anyhow(#[from] anyhow::Error),
     #[error("StreamError: {0}")]
     StreamError(#[from] StreamError),
+    #[error("StreamNotFound: {0}")]
+    StreamNotFound(#[from] StreamNotFound),
 }
 
 impl actix_web::ResponseError for PrismLogstreamError {
     fn status_code(&self) -> http::StatusCode {
         match self {
             PrismLogstreamError::Anyhow(_) => StatusCode::INTERNAL_SERVER_ERROR,
             PrismLogstreamError::StreamError(e) => e.status_code(),
+            PrismLogstreamError::StreamNotFound(_) => StatusCode::INTERNAL_SERVER_ERROR,
         }
     }
 
diff --git a/src/rbac/mod.rs b/src/rbac/mod.rs
@@ -20,10 +20,13 @@ pub mod map;
 pub mod role;
 pub mod user;
 
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 
 use chrono::{DateTime, Days, Utc};
 use itertools::Itertools;
+use role::model::DefaultPrivilege;
+use serde::Serialize;
+use url::Url;
 
 use crate::rbac::map::{mut_sessions, mut_users, sessions, users};
 use crate::rbac::role::Action;
@@ -166,6 +169,23 @@ impl Users {
     }
 }
 
+/// This struct represents a user along with their roles, email, etc
+///
+/// TODO: rename this after deprecating the older struct
+#[derive(Debug, Serialize, Clone)]
+pub struct UsersPrism {
+    // username
+    pub id: String,
+    // oaith or native
+    pub method: String,
+    // email only if method is oauth
+    pub email: Option<String>,
+    // picture only if oauth
+    pub picture: Option<Url>,
+    // roles for the user
+    pub roles: HashMap<String, Vec<DefaultPrivilege>>,
+}
+
 fn roles_to_permission(roles: Vec<String>) -> Vec<Permission> {
     let mut perms = HashSet::new();
     for role in &roles {
