Merge remote-tracking branch 'origin/main' into static-schema

Author: Devdutt Shenoi

CONTRIBUTING.md

@@ -0,0 +1,68 @@
+### Contributing
+
+This document outlines how you can contribute to Parseable.
+
+Thank you for considering to contribute to Parseable. The goal of this document is to provide everything you need to start your contribution. We encourage all contributions, including but not limited to:
+- Code patches, bug fixes.
+- Tutorials or blog posts.
+- Improving the documentation.
+- Submitting [bug reports](https://github.com/parseablehq/parseable/issues/new).
+
+### Prerequisites
+
+- Your PR has an associated issue. Find an [existing issue](https://github.com/parseablehq/parseable/issues) or open a new issue.
+- You've discussed the with [Parseable community](https://logg.ing/community).
+- You're familiar with GitHub Pull Requests(PR) workflow.
+- You've read the [Parseable documentation](https://www.parseable.com/docs).
+
+### Contribution workflow
+
+- Fork the [Parseable repository](https://help.github.com/en/github/getting-started-with-github/fork-a-repo) in your own GitHub account.
+- [Create a new branch](https://help.github.com/en/github/collaborating-with-issues-and-pull-requests/creating-and-deleting-branches-within-your-repository).
+- Review the Development Workflow section that describes the steps to maintain the repository.
+- Make your changes on your branch.
+- Submit the branch as a Pull Request pointing to the main branch of the Parseable repository. A maintainer should comment and/or review your Pull Request within a few hours.
+- You’ll be asked to review & sign the [Parseable Contributor License Agreement (CLA)](https://github.com/parseablehq/.github/blob/main/CLA.md) on the GitHub PR. Please ensure that you review the document. Once you’re ready, please sign the CLA by adding a comment I have read the CLA Document and I hereby sign the CLA in your Pull Request.
+- Please ensure that the commits in your Pull Request are made by the same GitHub user (which was used to create the PR and add the comment).
+
+### Development workflow
+
+We recommend Linux or macOS as the development platform for Parseable.
+
+#### Setup and run Parseable
+
+Parseable needs Rust 1.77.1 or above. Use the below command to build and run Parseable binary with local mode.
+
+```sh
+cargo run --release local-store
+```
+
+We recommend using the --release flag to test the full performance of Parseable.
+
+#### Running Tests
+
+```sh
+cargo test
+```
+
+This command will be triggered to each PR as a requirement for merging it.
+
+If you get a "Too many open files" error you might want to increase the open file limit using this command:
+
+```sh
+ulimit -Sn 3000
+```
+
+If you get a OpenSSL related error while building Parseable, you might need to install the dependencies using this command:
+
+```sh
+sudo apt install build-essential
+sudo apt-get install libssl-dev
+sudo apt-get install pkg-config
+```
+
+### Git Guidelines
+
+- The PR title should be accurate and descriptive of the changes.
+- The draft PRs are recommended when you want to show that you are working on something and make your work visible. Convert your PR as a draft if your changes are a work in progress. Maintainers will review the PR once you mark your PR as ready for review.
+- The branch related to the PR must be up-to-date with main before merging. We use Bors to automatically enforce this requirement without the PR author having to rebase manually.

Cargo.lock

@@ -0,0 +1,5 @@
+[target.aarch64-unknown-linux-gnu]
+image = "ghcr.io/cross-rs/aarch64-unknown-linux-gnu@sha256:1e2a0291f92a4372cbc22d8994e735473045383f1ce7fa44a16c234ba00187f4"
+
+[target.x86_64-unknown-linux-gnu]
+image = "ghcr.io/cross-rs/x86_64-unknown-linux-gnu@sha256:bf05360bb9d6d4947eed60532ac7a0d7e8fae8f214e9abb801d5941c8fe4918d"

Cross.toml

@@ -31,6 +31,7 @@ use tracing::trace;
 
 use crate::{
     alerts::AggregateCondition,
+    parseable::PARSEABLE,
     query::{TableScanVisitor, QUERY_SESSION},
     rbac::{
         map::SessionKey,
@@ -137,8 +138,9 @@ async fn execute_base_query(
         AlertError::CustomError(format!("Table name not found in query- {}", original_query))
     })?;
 
+    let time_partition = PARSEABLE.get_stream(&stream_name)?.get_time_partition();
     query
-        .get_dataframe(stream_name)
+        .get_dataframe(time_partition.as_ref())
         .await
         .map_err(|err| AlertError::CustomError(err.to_string()))
 }

src/alerts/alerts_utils.rs

@@ -24,6 +24,7 @@ use datafusion::common::tree_node::TreeNode;
 use http::StatusCode;
 use itertools::Itertools;
 use once_cell::sync::Lazy;
+use serde::Serialize;
 use serde_json::Error as SerdeError;
 use std::collections::{HashMap, HashSet};
 use std::fmt::{self, Display};
@@ -36,7 +37,7 @@ use ulid::Ulid;
 pub mod alerts_utils;
 pub mod target;
 
-use crate::parseable::PARSEABLE;
+use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::map::SessionKey;
 use crate::storage;
@@ -513,17 +514,16 @@ impl AlertConfig {
 
         // for now proceed in a similar fashion as we do in query
         // TODO: in case of multiple table query does the selection of time partition make a difference? (especially when the tables don't have overlapping data)
-        let stream_name = if let Some(stream_name) = query.first_table_name() {
-            stream_name
-        } else {
+        let Some(stream_name) = query.first_table_name() else {
             return Err(AlertError::CustomError(format!(
                 "Table name not found in query- {}",
                 self.query
             )));
         };
 
+        let time_partition = PARSEABLE.get_stream(&stream_name)?.get_time_partition();
         let base_df = query
-            .get_dataframe(stream_name)
+            .get_dataframe(time_partition.as_ref())
             .await
             .map_err(|err| AlertError::CustomError(err.to_string()))?;
 
@@ -703,6 +703,8 @@ pub enum AlertError {
     CustomError(String),
     #[error("Invalid State Change: {0}")]
     InvalidStateChange(String),
+    #[error("{0}")]
+    StreamNotFound(#[from] StreamNotFound),
 }
 
 impl actix_web::ResponseError for AlertError {
@@ -716,6 +718,7 @@ impl actix_web::ResponseError for AlertError {
             Self::DatafusionError(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::CustomError(_) => StatusCode::BAD_REQUEST,
             Self::InvalidStateChange(_) => StatusCode::BAD_REQUEST,
+            Self::StreamNotFound(_) => StatusCode::NOT_FOUND,
         }
     }
 
@@ -873,3 +876,52 @@ impl Alerts {
         Ok(())
     }
 }
+
+#[derive(Debug, Serialize)]
+pub struct AlertsInfo {
+    total: u64,
+    silenced: u64,
+    resolved: u64,
+    triggered: u64,
+    low: u64,
+    medium: u64,
+    high: u64,
+}
+
+// TODO: add RBAC
+pub async fn get_alerts_info() -> Result<AlertsInfo, AlertError> {
+    let alerts = ALERTS.alerts.read().await;
+    let mut total = 0;
+    let mut silenced = 0;
+    let mut resolved = 0;
+    let mut triggered = 0;
+    let mut low = 0;
+    let mut medium = 0;
+    let mut high = 0;
+
+    for (_, alert) in alerts.iter() {
+        total += 1;
+        match alert.state {
+            AlertState::Silenced => silenced += 1,
+            AlertState::Resolved => resolved += 1,
+            AlertState::Triggered => triggered += 1,
+        }
+
+        match alert.severity {
+            Severity::Low => low += 1,
+            Severity::Medium => medium += 1,
+            Severity::High => high += 1,
+            _ => {}
+        }
+    }
+
+    Ok(AlertsInfo {
+        total,
+        silenced,
+        resolved,
+        triggered,
+        low,
+        medium,
+        high,
+    })
+}

src/alerts/mod.rs

@@ -260,11 +260,12 @@ pub struct Options {
         help = "Set a fixed memory limit for query in GiB"
     )]
     pub query_memory_pool_size: Option<usize>,
-
+    // reduced the max row group size from 1048576
+    // smaller row groups help in faster query performance in multi threaded query
     #[arg(
         long,
         env = "P_PARQUET_ROW_GROUP_SIZE",
-        default_value = "1048576",
+        default_value = "262144",
         help = "Number of rows in a row group"
     )]
     pub row_group_size: usize,

src/cli.rs

@@ -38,7 +38,7 @@ use crate::handlers::http::query::{into_query, update_schema_when_distributed};
 use crate::handlers::livetail::cross_origin_config;
 use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::parseable::PARSEABLE;
-use crate::query::{TableScanVisitor, QUERY_SESSION};
+use crate::query::{execute, TableScanVisitor, QUERY_SESSION};
 use crate::utils::arrow::flight::{
     append_temporary_events, get_query_from_ticket, into_flight_data, run_do_get_rpc,
     send_to_ingester,
@@ -215,8 +215,8 @@ impl FlightService for AirServiceImpl {
             Status::permission_denied("User Does not have permission to access this")
         })?;
         let time = Instant::now();
-        let (records, _) = query
-            .execute(stream_name.clone())
+
+        let (records, _) = execute(query, &stream_name)
             .await
             .map_err(|err| Status::internal(err.to_string()))?;
 

src/handlers/airplane.rs

@@ -367,64 +367,29 @@ pub async fn sync_role_update_with_ingestors(
     Ok(())
 }
 
-pub async fn fetch_daily_stats_from_ingestors(
-    stream_name: &str,
+pub fn fetch_daily_stats_from_ingestors(
     date: &str,
+    stream_meta_list: &[ObjectStoreFormat],
 ) -> Result<Stats, StreamError> {
-    let mut total_events_ingested: u64 = 0;
-    let mut total_ingestion_size: u64 = 0;
-    let mut total_storage_size: u64 = 0;
-
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
-        error!("Fatal: failed to get ingestor info: {:?}", err);
-        StreamError::Anyhow(err)
-    })?;
-    for ingestor in ingestor_infos.iter() {
-        let uri = Url::parse(&format!(
-            "{}{}/metrics",
-            &ingestor.domain_name,
-            base_path_without_preceding_slash()
-        ))
-        .map_err(|err| {
-            StreamError::Anyhow(anyhow::anyhow!("Invalid URL in Ingestor Metadata: {}", err))
-        })?;
-
-        let res = HTTP_CLIENT
-            .get(uri)
-            .header(header::AUTHORIZATION, &ingestor.token)
-            .header(header::CONTENT_TYPE, "application/json")
-            .send()
-            .await;
-
-        if let Ok(res) = res {
-            let text = res
-                .text()
-                .await
-                .map_err(|err| StreamError::Anyhow(anyhow::anyhow!("Request failed: {}", err)))?;
-            let lines: Vec<Result<String, std::io::Error>> =
-                text.lines().map(|line| Ok(line.to_owned())).collect_vec();
-
-            let sample = prometheus_parse::Scrape::parse(lines.into_iter())
-                .map_err(|err| {
-                    StreamError::Anyhow(anyhow::anyhow!(
-                        "Invalid URL in Ingestor Metadata: {}",
-                        err
-                    ))
-                })?
-                .samples;
-
-            let (events_ingested, ingestion_size, storage_size) =
-                Metrics::get_daily_stats_from_samples(sample, stream_name, date);
-            total_events_ingested += events_ingested;
-            total_ingestion_size += ingestion_size;
-            total_storage_size += storage_size;
+    // for the given date, get the stats from the ingestors
+    let mut events_ingested = 0;
+    let mut ingestion_size = 0;
+    let mut storage_size = 0;
+
+    for meta in stream_meta_list.iter() {
+        for manifest in meta.snapshot.manifest_list.iter() {
+            if manifest.time_lower_bound.date_naive().to_string() == date {
+                events_ingested += manifest.events_ingested;
+                ingestion_size += manifest.ingestion_size;
+                storage_size += manifest.storage_size;
+            }
         }
     }
 
     let stats = Stats {
-        events: total_events_ingested,
-        ingestion: total_ingestion_size,
-        storage: total_storage_size,
+        events: events_ingested,
+        ingestion: ingestion_size,
+        storage: storage_size,
     };
     Ok(stats)
 }
@@ -474,17 +439,17 @@ pub async fn fetch_stats_from_ingestors(
         Utc::now(),
         IngestionStats::new(
             count,
-            format!("{} Bytes", ingestion_size),
+            ingestion_size,
             lifetime_count,
-            format!("{} Bytes", lifetime_ingestion_size),
+            lifetime_ingestion_size,
             deleted_count,
-            format!("{} Bytes", deleted_ingestion_size),
+            deleted_ingestion_size,
             "json",
         ),
         StorageStats::new(
-            format!("{} Bytes", storage_size),
-            format!("{} Bytes", lifetime_storage_size),
-            format!("{} Bytes", deleted_storage_size),
+            storage_size,
+            lifetime_storage_size,
+            deleted_storage_size,
             "parquet",
         ),
     );