fixes for quest OSS

parmesant · parmesant · commit be1a48bb8f39 · 2026-01-15T18:24:28.000+05:30
diff --git a/src/handlers/http/correlation.rs b/src/handlers/http/correlation.rs
@@ -108,7 +108,7 @@ pub async fn delete(
         .map_err(|err| CorrelationError::AnyhowError(Error::msg(err.to_string())))?;
 
     CORRELATIONS
-        .delete(&correlation_id, &user_id, &Some(tenant_id))
+        .delete(&correlation_id, &user_id, &tenant_id)
         .await?;
 
     Ok(HttpResponse::Ok().finish())
diff --git a/src/handlers/http/middleware.rs b/src/handlers/http/middleware.rs
@@ -34,7 +34,7 @@ use crate::{
         STREAM_NAME_HEADER_KEY, http::modal::OIDC_CLIENT,
     },
     option::Mode,
-    parseable::PARSEABLE,
+    parseable::{DEFAULT_TENANT, PARSEABLE},
     rbac::{
         EXPIRY_DURATION,
         map::{SessionKey, mut_sessions, mut_users, sessions, users},
@@ -167,10 +167,13 @@ where
         // append tenant id if present
         let user_and_tenant_id = match get_user_and_tenant_from_request(req.request()) {
             Ok((uid, tid)) => {
-                req.headers_mut().insert(
-                    HeaderName::from_static("tenant"),
-                    HeaderValue::from_str(&tid).unwrap(),
-                );
+                if tid.is_some() {
+                    req.headers_mut().insert(
+                        HeaderName::from_static("tenant"),
+                        HeaderValue::from_str(&tid.as_ref().unwrap()).unwrap(),
+                    );
+                }
+
                 Ok((uid, tid))
             }
             Err(e) => Err(e),
@@ -197,7 +200,8 @@ where
                     && let Ok((userid, tenant_id)) = user_and_tenant_id
                 {
                     let bearer_to_refresh = {
-                        if let Some(users) = users().get(&tenant_id)
+                        if let Some(users) =
+                            users().get(tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v))
                             && let Some(user) = users.get(&userid)
                         {
                             match &user.ty {
@@ -244,7 +248,8 @@ where
 
                         let user_roles = {
                             let mut users_guard = mut_users();
-                            if let Some(users) = users_guard.get_mut(&tenant_id)
+                            if let Some(users) = users_guard
+                                .get_mut(tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v))
                                 && let Some(user) = users.get_mut(&userid)
                             {
                                 if let user::UserType::OAuth(oauth) = &mut user.ty {
@@ -262,18 +267,25 @@ where
                             userid.clone(),
                             key.clone(),
                             Utc::now() + expires_in,
-                            roles_to_permission(user_roles, &tenant_id),
-                            &Some(tenant_id),
+                            roles_to_permission(
+                                user_roles,
+                                tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v),
+                            ),
+                            &tenant_id,
                         );
-                    } else if let Some(users) = users().get(&tenant_id)
+                    } else if let Some(users) =
+                        users().get(tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v))
                         && let Some(user) = users.get(&userid)
                     {
                         mut_sessions().track_new(
                             userid.clone(),
                             key.clone(),
                             Utc::now() + EXPIRY_DURATION,
-                            roles_to_permission(user.roles(), &tenant_id),
-                            &Some(tenant_id),
+                            roles_to_permission(
+                                user.roles(),
+                                tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v),
+                            ),
+                            &tenant_id,
                         );
                     }
                 }
diff --git a/src/handlers/http/users/dashboards.rs b/src/handlers/http/users/dashboards.rs
@@ -100,7 +100,7 @@ pub async fn create_dashboard(
     let user_id = get_hash(&user_id);
 
     DASHBOARDS
-        .create(&user_id, &mut dashboard, &Some(tenant_id))
+        .create(&user_id, &mut dashboard, &tenant_id)
         .await?;
     Ok((web::Json(dashboard), StatusCode::OK))
 }
@@ -114,9 +114,8 @@ pub async fn update_dashboard(
     let user_id = get_hash(&user_id);
     let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?;
     let is_admin = is_admin(&req).map_err(|e| DashboardError::Custom(e.to_string()))?;
-    let tenant_id = &Some(tenant_id);
     let mut existing_dashboard = DASHBOARDS
-        .get_dashboard_by_user(dashboard_id, &user_id, is_admin, tenant_id)
+        .get_dashboard_by_user(dashboard_id, &user_id, is_admin, &tenant_id)
         .await
         .ok_or(DashboardError::Metadata(
             "Dashboard does not exist or user is not authorized",
@@ -189,7 +188,7 @@ pub async fn update_dashboard(
     };
 
     DASHBOARDS
-        .update(&user_id, dashboard_id, &mut final_dashboard, tenant_id)
+        .update(&user_id, dashboard_id, &mut final_dashboard, &tenant_id)
         .await?;
 
     Ok((web::Json(final_dashboard), StatusCode::OK))
@@ -206,7 +205,7 @@ pub async fn delete_dashboard(
     let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?;
 
     DASHBOARDS
-        .delete_dashboard(&user_id, dashboard_id, is_admin, &Some(tenant_id))
+        .delete_dashboard(&user_id, dashboard_id, is_admin, &tenant_id)
         .await?;
 
     Ok(HttpResponse::Ok().finish())
@@ -223,12 +222,11 @@ pub async fn add_tile(
 
     let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?;
     let user_id = get_hash(&user_id);
-    let tenant_id = &Some(tenant_id);
     let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?;
     let is_admin = is_admin(&req).map_err(|e| DashboardError::Custom(e.to_string()))?;
 
     let mut dashboard = DASHBOARDS
-        .get_dashboard_by_user(dashboard_id, &user_id, is_admin, tenant_id)
+        .get_dashboard_by_user(dashboard_id, &user_id, is_admin, &tenant_id)
         .await
         .ok_or(DashboardError::Unauthorized)?;
 
@@ -241,7 +239,7 @@ pub async fn add_tile(
     tiles.push(tile);
 
     DASHBOARDS
-        .update(&user_id, dashboard_id, &mut dashboard, tenant_id)
+        .update(&user_id, dashboard_id, &mut dashboard, &tenant_id)
         .await?;
 
     Ok((web::Json(dashboard), StatusCode::OK))
diff --git a/src/handlers/http/users/filters.rs b/src/handlers/http/users/filters.rs
@@ -50,7 +50,7 @@ pub async fn get(
     let filter_id = filter_id.into_inner();
     let is_admin = is_admin(&req).map_err(|e| FiltersError::Custom(e.to_string()))?;
     if let Some(filter) = FILTERS
-        .get_filter(&filter_id, &get_hash(&user_id), is_admin, &Some(tenant_id))
+        .get_filter(&filter_id, &get_hash(&user_id), is_admin, &tenant_id)
         .await
     {
         return Ok((web::Json(filter), StatusCode::OK));
@@ -71,9 +71,8 @@ pub async fn post(
     filter.filter_id = Some(filter_id.clone());
     filter.user_id = Some(user_id.clone());
     filter.version = Some(CURRENT_FILTER_VERSION.to_string());
-    let tenant_id = &Some(tenant_id);
-    PARSEABLE.metastore.put_filter(&filter, tenant_id).await?;
-    FILTERS.update(&filter, tenant_id).await;
+    PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?;
+    FILTERS.update(&filter, &tenant_id).await;
 
     Ok((web::Json(filter), StatusCode::OK))
 }
@@ -88,9 +87,8 @@ pub async fn update(
     let filter_id = filter_id.into_inner();
     let is_admin = is_admin(&req).map_err(|e| FiltersError::Custom(e.to_string()))?;
 
-    let tenant_id = &Some(tenant_id);
     if FILTERS
-        .get_filter(&filter_id, &user_id, is_admin, tenant_id)
+        .get_filter(&filter_id, &user_id, is_admin, &tenant_id)
         .await
         .is_none()
     {
@@ -102,8 +100,8 @@ pub async fn update(
     filter.user_id = Some(user_id.clone());
     filter.version = Some(CURRENT_FILTER_VERSION.to_string());
 
-    PARSEABLE.metastore.put_filter(&filter, tenant_id).await?;
-    FILTERS.update(&filter, tenant_id).await;
+    PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?;
+    FILTERS.update(&filter, &tenant_id).await;
 
     Ok((web::Json(filter), StatusCode::OK))
 }
@@ -115,20 +113,19 @@ pub async fn delete(
     let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?;
     user_id = get_hash(&user_id);
     let filter_id = filter_id.into_inner();
-    let tenant_id = &Some(tenant_id);
     let is_admin = is_admin(&req).map_err(|e| FiltersError::Custom(e.to_string()))?;
     let filter = FILTERS
-        .get_filter(&filter_id, &user_id, is_admin, tenant_id)
+        .get_filter(&filter_id, &user_id, is_admin, &tenant_id)
         .await
         .ok_or(FiltersError::Metadata(
             "Filter does not exist or user is not authorized",
         ))?;
 
     PARSEABLE
         .metastore
-        .delete_filter(&filter, tenant_id)
+        .delete_filter(&filter, &tenant_id)
         .await?;
-    FILTERS.delete_filter(&filter_id, tenant_id).await;
+    FILTERS.delete_filter(&filter_id, &tenant_id).await;
 
     Ok(HttpResponse::Ok().finish())
 }
diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs
@@ -25,7 +25,10 @@ use std::{
     sync::{Arc, RwLock},
 };
 
-use actix_web::http::{StatusCode, header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue}};
+use actix_web::http::{
+    StatusCode,
+    header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
+};
 use arrow_schema::{Field, Schema};
 use bytes::Bytes;
 use chrono::Utc;
@@ -386,6 +389,8 @@ impl Parseable {
             storage.create_stream_from_ingestor(stream_name, tenant_id),
             storage.create_schema_from_metastore(stream_name, tenant_id)
         )?;
+        tracing::warn!(stream_metadata_bytes=?stream_metadata_bytes);
+        tracing::warn!(schema_bytes=?schema_bytes);
 
         let stream_metadata = if stream_metadata_bytes.is_empty() {
             ObjectStoreFormat::default()
@@ -452,7 +457,7 @@ impl Parseable {
         if let Some(hot_tier_config) = hot_tier {
             stream.set_hot_tier(Some(hot_tier_config));
         }
-
+        tracing::warn!(commit_schema=?schema);
         // commit schema in memory
         commit_schema(stream_name, schema, tenant_id).map_err(|e| StreamError::Anyhow(e.into()))?;
 
@@ -556,8 +561,8 @@ impl Parseable {
         }
 
         // For distributed deployments, if the stream not found in memory map,
-        //check if it exists in the storage
-        //create stream and schema from storage
+        // check if it exists in the storage
+        // create stream and schema from storage
         if self.options.mode != Mode::All
             && self
                 .create_stream_and_schema_from_storage(stream_name, tenant_id)
@@ -794,6 +799,7 @@ impl Parseable {
         // Proceed to create log stream if it doesn't exist
         let storage = self.storage.get_object_store();
 
+        // update owner and permissions
         let meta = ObjectStoreFormat {
             created_at: Utc::now().to_rfc3339(),
             permissions: vec![Permisssion::new(PARSEABLE.options.username.clone())],
@@ -812,12 +818,16 @@ impl Parseable {
             ..Default::default()
         };
 
+        tracing::warn!(meta=?meta);
         match storage
             .create_stream(&stream_name, meta, schema.clone(), tenant_id)
             .await
         {
             Ok(created_at) => {
                 tracing::warn!(created_stream_at=?created_at);
+                tracing::warn!(stream_name=?stream_name);
+                tracing::warn!(schema=?schema);
+                tracing::warn!(tenant_id=?tenant_id);
                 let mut static_schema: HashMap<String, Arc<Field>> = HashMap::new();
 
                 for (field_name, field) in schema
@@ -1181,7 +1191,8 @@ impl Parseable {
     }
 
     pub fn list_tenants(&self) -> Option<Vec<String>> {
-        if let Ok(t) = self.tenants.as_ref().read() {
+        if let Ok(t) = self.tenants.as_ref().read()
+        && !t.is_empty() {
             let t = t.clone();
             Some(t)
         } else {
diff --git a/src/parseable/streams.rs b/src/parseable/streams.rs
@@ -646,6 +646,7 @@ impl Stream {
             let schema = Arc::new(merged_schema);
 
             let part_path = parquet_path.with_extension("part");
+            tracing::warn!(part_path=?part_path);
             if !self.write_parquet_part_file(
                 &part_path,
                 record_reader,
@@ -1164,6 +1165,8 @@ impl Streams {
         } else {
             vec![DEFAULT_TENANT.to_owned()]
         };
+        tracing::warn!(flush_and_convert_tenants=?tenants);
+        tracing::warn!(parseable_streams_tenants=?self.read().unwrap().keys());
         for tenant_id in tenants {
             let guard = self.read().expect(LOCK_EXPECT);
             let streams: Vec<Arc<Stream>> = if let Some(tenant_streams) = guard.get(&tenant_id) {
@@ -1176,12 +1179,6 @@ impl Streams {
                     .spawn(async move { stream.flush_and_convert(init_signal, shutdown_signal) });
             }
         }
-        // let streams: Vec<Arc<Stream>> = self
-        //     .read()
-        //     .expect(LOCK_EXPECT)
-        //     .values()
-        //     .map(Arc::clone)
-        //     .collect();
     }
 }
 
diff --git a/src/rbac/user.rs b/src/rbac/user.rs
@@ -180,7 +180,7 @@ pub fn get_super_admin_user() -> User {
         }),
         roles: ["super-admin".to_string()].into(),
         user_groups: HashSet::new(),
-        tenant: Some(DEFAULT_TENANT.to_owned()),
+        tenant: None,
     }
 }
 
diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs
@@ -53,6 +53,7 @@ use crate::metrics::increment_parquets_stored_by_date;
 use crate::metrics::increment_parquets_stored_size_by_date;
 use crate::metrics::{EVENTS_STORAGE_SIZE_DATE, LIFETIME_EVENTS_STORAGE_SIZE, STORAGE_SIZE};
 use crate::option::Mode;
+use crate::parseable::DEFAULT_TENANT;
 use crate::parseable::{LogStream, PARSEABLE, Stream};
 use crate::stats::FullStats;
 use crate::storage::SETTINGS_ROOT_DIRECTORY;
@@ -630,7 +631,9 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
                 .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
             return Ok(stream_metadata_bytes);
         }
-        tracing::warn!("unable to find stream- {stream_name} with tenant- {tenant_id:?} in PARSEABLE.get_stream");
+        tracing::warn!(
+            "unable to find stream- {stream_name} with tenant- {tenant_id:?} in PARSEABLE.get_stream"
+        );
         let mut all_log_sources: Vec<LogSourceEntry> = Vec::new();
 
         if let Some(stream_metadata_obs) = PARSEABLE
@@ -1055,6 +1058,7 @@ async fn process_schema_files(
     tenant_id: &Option<String>,
 ) -> Result<(), ObjectStorageError> {
     for path in upload_context.stream.schema_files() {
+        tracing::warn!(upload_context_schema_files=?path);
         let file = File::open(&path)?;
         let schema: Schema = serde_json::from_reader(file)?;
         commit_schema_to_storage(stream_name, schema, tenant_id).await?;
@@ -1079,7 +1083,9 @@ fn stream_relative_path(
         let custom_partition_list = custom_partition_fields.split(',').collect::<Vec<&str>>();
         file_suffix = str::replacen(filename, ".", "/", 3 + custom_partition_list.len());
     }
-    if let Some(tenant) = tenant_id {
+    if let Some(tenant) = tenant_id
+        && !tenant.eq(DEFAULT_TENANT)
+    {
         format!("{tenant}/{stream_name}/{file_suffix}")
     } else {
         format!("{stream_name}/{file_suffix}")
diff --git a/src/utils/mod.rs b/src/utils/mod.rs
@@ -58,22 +58,22 @@ pub fn extract_datetime(path: &str) -> Option<NaiveDateTime> {
     }
 }
 
-pub fn get_user_and_tenant_from_request(req: &HttpRequest) -> Result<(String, String), RBACError> {
+pub fn get_user_and_tenant_from_request(
+    req: &HttpRequest,
+) -> Result<(String, Option<String>), RBACError> {
     let session_key = extract_session_key_from_req(req).map_err(|_| RBACError::UserDoesNotExist)?;
     match &session_key {
         SessionKey::BasicAuth { username, password } => {
-            if let Some(user) = Users.get_user_from_basic(&username, &password)
-                && let Some(tenant) = user.tenant
-            {
-                return Ok((username.clone(), tenant));
+            if let Some(user) = Users.get_user_from_basic(&username, &password) {
+                return Ok((username.clone(), user.tenant.clone()));
             }
         }
         SessionKey::SessionId(_) => {}
     }
     let Some((user_id, tenant_id)) = Users.get_userid_from_session(&session_key) else {
         return Err(RBACError::UserDoesNotExist);
     };
-    Ok((user_id, tenant_id))
+    Ok((user_id, Some(tenant_id)))
 }
 
 pub fn get_tenant_id_from_request(req: &HttpRequest) -> Option<String> {

Original file line number	Diff line number	Diff line change
`@@ -180,7 +180,7 @@ pub fn get_super_admin_user() -> User {`
`180`	`180`	`}),`
`181`	`181`	`roles: ["super-admin".to_string()].into(),`
`182`	`182`	`user_groups: HashSet::new(),`
`183`		`- tenant: Some(DEFAULT_TENANT.to_owned()),`
	`183`	`+ tenant: None,`
`184`	`184`	`}`
`185`	`185`	`}`
`186`	`186`
Original file line number	Diff line number	Diff line change
`@@ -58,22 +58,22 @@ pub fn extract_datetime(path: &str) -> Option<NaiveDateTime> {`
`58`	`58`	`}`
`59`	`59`	`}`
`60`	`60`
`61`		`-pub fn get_user_and_tenant_from_request(req: &HttpRequest) -> Result<(String, String), RBACError> {`
	`61`	`+pub fn get_user_and_tenant_from_request(`
	`62`	`+ req: &HttpRequest,`
	`63`	`+) -> Result<(String, Option<String>), RBACError> {`
`62`	`64`	`let session_key = extract_session_key_from_req(req).map_err(\|_\| RBACError::UserDoesNotExist)?;`
`63`	`65`	`match &session_key {`
`64`	`66`	`SessionKey::BasicAuth { username, password } => {`
`65`		`- if let Some(user) = Users.get_user_from_basic(&username, &password)`
`66`		`- && let Some(tenant) = user.tenant`
`67`		`- {`
`68`		`- return Ok((username.clone(), tenant));`
	`67`	`+ if let Some(user) = Users.get_user_from_basic(&username, &password) {`
	`68`	`+ return Ok((username.clone(), user.tenant.clone()));`
`69`	`69`	`}`
`70`	`70`	`}`
`71`	`71`	`SessionKey::SessionId(_) => {}`
`72`	`72`	`}`
`73`	`73`	`let Some((user_id, tenant_id)) = Users.get_userid_from_session(&session_key) else {`
`74`	`74`	`return Err(RBACError::UserDoesNotExist);`
`75`	`75`	`};`
`76`		`- Ok((user_id, tenant_id))`
	`76`	`+ Ok((user_id, Some(tenant_id)))`
`77`	`77`	`}`
`78`	`78`
`79`	`79`	`pub fn get_tenant_id_from_request(req: &HttpRequest) -> Option<String> {`