Merge branch 'main' into alpine

Author: Devdutt Shenoi

.github/workflows/build.yaml

@@ -0,0 +1,76 @@
+on:
+  pull_request:
+    paths-ignore:
+      - "docs/**"
+      - "helm/**"
+      - "assets/**"
+      - "**.md"
+  push:
+    branches:
+      - main
+
+name: Ensure parseable builds on all release targets
+jobs:
+  build-linux:
+    name: Build for ${{matrix.target}}
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        target:
+          - aarch64-unknown-linux-gnu # linux(arm)
+          - x86_64-unknown-linux-gnu # linux(64 bit)
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          profile: minimal # minimal component installation (ie, no documentation)
+          target: ${{ matrix.target }}
+          override: true
+
+      - uses: actions-rs/cargo@v1
+        with:
+          use-cross: true
+          command: build
+          args: --target ${{matrix.target}}
+
+  build-windows:
+    name: Build for windows
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          profile: minimal # minimal component installation (ie, no documentation)
+          default: true
+          override: true
+
+      - name: Build on windows
+        run: cargo build --target x86_64-pc-windows-msvc
+
+  build-macos:
+    name: Build for ${{matrix.target}}
+    runs-on: macos-latest
+    strategy:
+      matrix:
+        target:
+          - aarch64-apple-darwin # macos(arm)
+          - x86_64-apple-darwin # macos(intel 64 bit)
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          profile: minimal
+          target: ${{ matrix.target }}
+          override: true
+
+      - name: Build on ${{ matrix.target }}
+        run: cargo build --target ${{ matrix.target }}

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "parseable"
-version = "1.7.2"
+version = "1.7.3"
 authors = ["Parseable Team <[email protected]>"]
 edition = "2021"
 rust-version = "1.83.0"
@@ -139,8 +139,8 @@ rstest = "0.23.0"
 arrow = "53.0.0"
 
 [package.metadata.parseable_ui]
-assets-url = "https://github.com/parseablehq/console/releases/download/v0.9.16/build.zip"
-assets-sha1 = "cb9af17fc1af07e590e839fc0ef4db18c323fc48"
+assets-url = "https://github.com/parseablehq/console/releases/download/v0.9.18/build.zip"
+assets-sha1 = "4516db38c8e556707b29b33569f9b1e53d5165f2"
 
 [features]
 debug = []

Cargo.toml

@@ -52,7 +52,7 @@ pub const DEFAULT_PASSWORD: &str = "admin";
     long_about = r#"
 Cloud Native, log analytics platform for modern applications.
 
-Usage: 
+Usage:
 parseable [command] [options..]
 
 
@@ -126,7 +126,7 @@ pub struct Options {
     // Server configuration
     #[arg(
         long,
-        env = "P_ADDR", 
+        env = "P_ADDR",
         default_value = "0.0.0.0:8000",
         value_parser = validation::socket_addr,
         help = "Address and port for Parseable HTTP(s) server"
@@ -294,35 +294,8 @@ pub struct Options {
     )]
     pub ingestor_endpoint: String,
 
-    // OIDC Configuration
-    #[arg(
-        long,
-        long = "oidc-client",
-        env = "P_OIDC_CLIENT_ID",
-        requires = "oidc",
-        group = "oidc",
-        help = "Client id for OIDC provider"
-    )]
-    oidc_client_id: Option<String>,
-
-    #[arg(
-        long,
-        env = "P_OIDC_CLIENT_SECRET",
-        requires = "oidc",
-        group = "oidc",
-        help = "Client secret for OIDC provider"
-    )]
-    oidc_client_secret: Option<String>,
-
-    #[arg(
-        long,
-        env = "P_OIDC_ISSUER",
-        value_parser = validation::url,
-        requires = "oidc",
-        group = "oidc",
-        help = "OIDC provider's host address"
-    )]
-    oidc_issuer: Option<Url>,
+    #[command(flatten)]
+    oidc: Option<OidcConfig>,
 
     // Kafka configuration (conditionally compiled)
     #[cfg(any(
@@ -391,6 +364,34 @@ pub struct Options {
     pub ms_clarity_tag: Option<String>,
 }
 
+#[derive(Parser, Debug)]
+pub struct OidcConfig {
+    #[arg(
+        long = "oidc-client-id",
+        env = "P_OIDC_CLIENT_ID",
+        required = false,
+        help = "Client id for OIDC provider"
+    )]
+    pub client_id: String,
+
+    #[arg(
+        long = "oidc-client-secret",
+        env = "P_OIDC_CLIENT_SECRET",
+        required = false,
+        help = "Client secret for OIDC provider"
+    )]
+    pub secret: String,
+
+    #[arg(
+        long = "oidc-issuer",
+        env = "P_OIDC_ISSUER",
+        required = false,
+        value_parser = validation::url,
+        help = "OIDC provider's host address"
+    )]
+    pub issuer: Url,
+}
+
 impl Options {
     pub fn local_stream_data_path(&self, stream_name: &str) -> PathBuf {
         self.local_staging_path.join(stream_name)
@@ -405,28 +406,24 @@ impl Options {
     }
 
     pub fn openid(&self) -> Option<OpenidConfig> {
-        match (
-            &self.oidc_client_id,
-            &self.oidc_client_secret,
-            &self.oidc_issuer,
-        ) {
-            (Some(id), Some(secret), Some(issuer)) => {
-                let origin = if let Some(url) = self.domain_address.clone() {
-                    oidc::Origin::Production(url)
-                } else {
-                    oidc::Origin::Local {
-                        socket_addr: self.address.clone(),
-                        https: self.tls_cert_path.is_some() && self.tls_key_path.is_some(),
-                    }
-                };
-                Some(OpenidConfig {
-                    id: id.clone(),
-                    secret: secret.clone(),
-                    issuer: issuer.clone(),
-                    origin,
-                })
+        let OidcConfig {
+            secret,
+            client_id,
+            issuer,
+        } = self.oidc.as_ref()?;
+        let origin = if let Some(url) = self.domain_address.clone() {
+            oidc::Origin::Production(url)
+        } else {
+            oidc::Origin::Local {
+                socket_addr: self.address.clone(),
+                https: self.tls_cert_path.is_some() && self.tls_key_path.is_some(),
             }
-            _ => None,
-        }
+        };
+        Some(OpenidConfig {
+            id: client_id.clone(),
+            secret: secret.clone(),
+            issuer: issuer.clone(),
+            origin,
+        })
     }
 }

src/cli.rs

@@ -45,36 +45,32 @@ pub static CORRELATIONS: Lazy<Correlation> = Lazy::new(Correlation::default);
 pub struct Correlation(RwLock<Vec<CorrelationConfig>>);
 
 impl Correlation {
-    pub async fn load(&self) -> Result<(), CorrelationError> {
-        // lead correlations from storage
+    //load correlations from storage
+    pub async fn load(&self) -> anyhow::Result<()> {
         let store = CONFIG.storage().get_object_store();
-        let all_correlations = store.get_correlations().await.unwrap_or_default();
-
-        let mut correlations = vec![];
-        for corr in all_correlations {
-            if corr.is_empty() {
-                continue;
-            }
-
-            let correlation: CorrelationConfig = match serde_json::from_slice(&corr) {
-                Ok(c) => c,
-                Err(e) => {
-                    error!("Unable to load correlation- {e}");
-                    continue;
-                }
-            };
-
-            correlations.push(correlation);
-        }
+        let all_correlations = store.get_all_correlations().await.unwrap_or_default();
+
+        let correlations: Vec<CorrelationConfig> = all_correlations
+            .into_iter()
+            .flat_map(|(_, correlations_bytes)| correlations_bytes)
+            .filter_map(|correlation| {
+                serde_json::from_slice(&correlation)
+                    .inspect_err(|e| {
+                        error!("Unable to load correlation: {e}");
+                    })
+                    .ok()
+            })
+            .collect();
 
         let mut s = self.0.write().await;
-        s.append(&mut correlations.clone());
+        s.extend(correlations);
         Ok(())
     }
 
     pub async fn list_correlations_for_user(
         &self,
         session_key: &SessionKey,
+        user_id: &str,
     ) -> Result<Vec<CorrelationConfig>, CorrelationError> {
         let correlations = self.0.read().await.iter().cloned().collect_vec();
 
@@ -87,27 +83,29 @@ impl Correlation {
                 .iter()
                 .map(|t| t.table_name.clone())
                 .collect_vec();
-            if user_auth_for_query(&permissions, tables).is_ok() {
+            if user_auth_for_query(&permissions, tables).is_ok() && c.user_id == user_id {
                 user_correlations.push(c);
             }
         }
         Ok(user_correlations)
     }
 
-    pub async fn get_correlation_by_id(
+    pub async fn get_correlation(
         &self,
         correlation_id: &str,
+        user_id: &str,
     ) -> Result<CorrelationConfig, CorrelationError> {
         let read = self.0.read().await;
-        let correlation = read.iter().find(|c| c.id == correlation_id).cloned();
+        let correlation = read
+            .iter()
+            .find(|c| c.id == correlation_id && c.user_id == user_id)
+            .cloned();
 
-        if let Some(c) = correlation {
-            Ok(c)
-        } else {
-            Err(CorrelationError::AnyhowError(anyhow::Error::msg(format!(
+        correlation.ok_or_else(|| {
+            CorrelationError::AnyhowError(anyhow::Error::msg(format!(
                 "Unable to find correlation with ID- {correlation_id}"
-            ))))
-        }
+            )))
+        })
     }
 
     pub async fn update(&self, correlation: &CorrelationConfig) -> Result<(), CorrelationError> {
@@ -152,6 +150,7 @@ pub struct CorrelationConfig {
     pub version: CorrelationVersion,
     pub title: String,
     pub id: String,
+    pub user_id: String,
     pub table_configs: Vec<TableConfig>,
     pub join_config: JoinConfig,
     pub filter: Option<FilterQuery>,
@@ -164,7 +163,6 @@ impl CorrelationConfig {}
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CorrelationRequest {
-    pub version: CorrelationVersion,
     pub title: String,
     pub table_configs: Vec<TableConfig>,
     pub join_config: JoinConfig,
@@ -176,9 +174,10 @@ pub struct CorrelationRequest {
 impl From<CorrelationRequest> for CorrelationConfig {
     fn from(val: CorrelationRequest) -> Self {
         Self {
-            version: val.version,
+            version: CorrelationVersion::V1,
             title: val.title,
             id: get_hash(Utc::now().timestamp_micros().to_string().as_str()),
+            user_id: String::default(),
             table_configs: val.table_configs,
             join_config: val.join_config,
             filter: val.filter,
@@ -189,11 +188,12 @@ impl From<CorrelationRequest> for CorrelationConfig {
 }
 
 impl CorrelationRequest {
-    pub fn generate_correlation_config(self, id: String) -> CorrelationConfig {
+    pub fn generate_correlation_config(self, id: String, user_id: String) -> CorrelationConfig {
         CorrelationConfig {
-            version: self.version,
+            version: CorrelationVersion::V1,
             title: self.title,
             id,
+            user_id,
             table_configs: self.table_configs,
             join_config: self.join_config,
             filter: self.filter,