Merge branch 'main' into user-group

Author: nikhilsinhaparseable

src/cli.rs

@@ -451,6 +451,16 @@ pub struct Options {
         help = "Object store sync threshold in seconds"
     )]
     pub object_store_sync_threshold: u64,
+    // the oidc scope 
+    #[arg(
+        long = "oidc-scope",
+        name = "oidc-scope",
+        env = "P_OIDC_SCOPE",
+        default_value = "openid profile email",
+        required = false,
+        help = "OIDC scope to request (default: openid profile email)"
+    )]
+    pub scope: String,
 }
 
 #[derive(Parser, Debug)]

src/handlers/http/oidc.rs

@@ -32,7 +32,7 @@ use ulid::Ulid;
 use url::Url;
 
 use crate::{
-    handlers::{COOKIE_AGE_DAYS, OIDC_SCOPE, SESSION_COOKIE_NAME, USER_COOKIE_NAME},
+    handlers::{COOKIE_AGE_DAYS, SESSION_COOKIE_NAME, USER_COOKIE_NAME},
     oidc::{Claims, DiscoveredClient},
     parseable::PARSEABLE,
     rbac::{
@@ -77,7 +77,7 @@ pub async fn login(
     let session_key = extract_session_key_from_req(&req).ok();
     let (session_key, oidc_client) = match (session_key, oidc_client) {
         (None, None) => return Ok(redirect_no_oauth_setup(query.redirect.clone())),
-        (None, Some(client)) => return Ok(redirect_to_oidc(query, client)),
+        (None, Some(client)) => return Ok(redirect_to_oidc(query, client, PARSEABLE.options.scope.to_string().as_str())),
         (Some(session_key), client) => (session_key, client),
     };
     // try authorize
@@ -113,7 +113,7 @@ pub async fn login(
             } else {
                 Users.remove_session(&key);
                 if let Some(oidc_client) = oidc_client {
-                    redirect_to_oidc(query, oidc_client)
+                    redirect_to_oidc(query, oidc_client, PARSEABLE.options.scope.to_string().as_str())
                 } else {
                     redirect_to_client(query.redirect.as_str(), None)
                 }
@@ -226,10 +226,11 @@ fn exchange_basic_for_cookie(user: &User, key: SessionKey) -> Cookie<'static> {
 fn redirect_to_oidc(
     query: web::Query<RedirectAfterLogin>,
     oidc_client: &DiscoveredClient,
+    scope: &str,
 ) -> HttpResponse {
     let redirect = query.into_inner().redirect.to_string();
     let auth_url = oidc_client.auth_url(&Options {
-        scope: Some(OIDC_SCOPE.into()),
+        scope: Some(scope.to_string()),
         state: Some(redirect),
         ..Default::default()
     });

src/handlers/http/query.rs

@@ -35,11 +35,12 @@ use http::StatusCode;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Instant;
-use tracing::error;
+use tokio::task::JoinSet;
+use tracing::{error, warn};
 
 use crate::event::commit_schema;
 use crate::metrics::QUERY_EXECUTE_TIME;
@@ -126,7 +127,7 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
     {
         Ok(raw_logical_plan) => raw_logical_plan,
         Err(_) => {
-            create_streams_for_querier().await;
+            create_streams_for_querier().await?;
             session_state
                 .create_logical_plan(&query_request.query)
                 .await?
@@ -433,17 +434,45 @@ pub async fn update_schema_when_distributed(tables: &Vec<String>) -> Result<(),
 /// Create streams for querier if they do not exist
 /// get list of streams from memory and storage
 /// create streams for memory from storage if they do not exist
-pub async fn create_streams_for_querier() {
-    let querier_streams = PARSEABLE.streams.list();
+pub async fn create_streams_for_querier() -> Result<(), QueryError> {
     let store = PARSEABLE.storage.get_object_store();
-    let storage_streams = store.list_streams().await.unwrap();
-    for stream_name in storage_streams {
-        if !querier_streams.contains(&stream_name) {
-            let _ = PARSEABLE
+    let querier_streams = PARSEABLE.streams.list();
+
+    let querier_streams_set: HashSet<_> = querier_streams.into_iter().collect();
+
+    let storage_streams = store.list_streams().await?;
+
+    let missing_streams: Vec<_> = storage_streams
+        .into_iter()
+        .filter(|stream_name| !querier_streams_set.contains(stream_name))
+        .collect();
+
+    if missing_streams.is_empty() {
+        return Ok(());
+    }
+
+    let mut join_set = JoinSet::new();
+    for stream_name in missing_streams {
+        join_set.spawn(async move {
+            let result = PARSEABLE
                 .create_stream_and_schema_from_storage(&stream_name)
                 .await;
+
+            if let Err(e) = &result {
+                warn!("Failed to create stream '{}': {}", stream_name, e);
+            }
+
+            (stream_name, result)
+        });
+    }
+
+    while let Some(result) = join_set.join_next().await {
+        if let Err(join_error) = result {
+            warn!("Task join error: {}", join_error);
         }
     }
+
+    Ok(())
 }
 
 impl FromRequest for Query {

src/handlers/mod.rs

@@ -30,7 +30,6 @@ const STATIC_SCHEMA_FLAG: &str = "x-p-static-schema-flag";
 const AUTHORIZATION_KEY: &str = "authorization";
 const UPDATE_STREAM_KEY: &str = "x-p-update-stream";
 pub const STREAM_TYPE_KEY: &str = "x-p-stream-type";
-const OIDC_SCOPE: &str = "openid profile email";
 const COOKIE_AGE_DAYS: usize = 7;
 const SESSION_COOKIE_NAME: &str = "session";
 const USER_COOKIE_NAME: &str = "username";

src/parseable/mod.rs

@@ -35,6 +35,7 @@ use once_cell::sync::Lazy;
 pub use staging::StagingError;
 use streams::StreamRef;
 pub use streams::{Stream, StreamNotFound, Streams};
+use tokio::try_join;
 use tracing::error;
 
 #[cfg(feature = "kafka")]
@@ -270,17 +271,22 @@ impl Parseable {
             return Ok(false);
         }
 
-        let mut stream_metadata = ObjectStoreFormat::default();
-        let stream_metadata_bytes = storage.create_stream_from_ingestor(stream_name).await?;
-        if !stream_metadata_bytes.is_empty() {
-            stream_metadata = serde_json::from_slice::<ObjectStoreFormat>(&stream_metadata_bytes)?;
-        }
+        let (stream_metadata_bytes, schema_bytes) = try_join!(
+            storage.create_stream_from_ingestor(stream_name),
+            storage.create_schema_from_ingestor(stream_name)
+        )?;
 
-        let mut schema = Arc::new(Schema::empty());
-        let schema_bytes = storage.create_schema_from_ingestor(stream_name).await?;
-        if !schema_bytes.is_empty() {
-            schema = serde_json::from_slice::<Arc<Schema>>(&schema_bytes)?;
-        }
+        let stream_metadata = if stream_metadata_bytes.is_empty() {
+            ObjectStoreFormat::default()
+        } else {
+            serde_json::from_slice::<ObjectStoreFormat>(&stream_metadata_bytes)?
+        };
+
+        let schema = if schema_bytes.is_empty() {
+            Arc::new(Schema::empty())
+        } else {
+            serde_json::from_slice::<Arc<Schema>>(&schema_bytes)?
+        };
 
         let static_schema: HashMap<String, Arc<Field>> = schema
             .fields