Merge branch 'main' into add-custom-columns

Author: nikhilsinhaparseable

Cargo.lock

@@ -130,7 +130,7 @@ ureq = "2.12"
 url = "2.5"
 vergen = { version = "9.0", features = ["build", "cargo", "rustc", "si"] }
 vergen-gitcl = { version = "1.0", features = ["build", "cargo", "rustc", "si"] }
-zip = { version = "2.2", default-features = false, features = ["deflate"] }
+zip = { version = "2.3", default-features = false, features = ["deflate"] }
 anyhow = "1.0"
 
 [dev-dependencies]

Cargo.toml

@@ -886,6 +886,7 @@ pub struct AlertsInfo {
     low: u64,
     medium: u64,
     high: u64,
+    critical: u64,
 }
 
 // TODO: add RBAC
@@ -898,6 +899,7 @@ pub async fn get_alerts_info() -> Result<AlertsInfo, AlertError> {
     let mut low = 0;
     let mut medium = 0;
     let mut high = 0;
+    let mut critical = 0;
 
     for (_, alert) in alerts.iter() {
         total += 1;
@@ -911,7 +913,7 @@ pub async fn get_alerts_info() -> Result<AlertsInfo, AlertError> {
             Severity::Low => low += 1,
             Severity::Medium => medium += 1,
             Severity::High => high += 1,
-            _ => {}
+            Severity::Critical => critical += 1,
         }
     }
 
@@ -923,5 +925,6 @@ pub async fn get_alerts_info() -> Result<AlertsInfo, AlertError> {
         low,
         medium,
         high,
+        critical,
     })
 }

src/alerts/mod.rs

@@ -270,6 +270,14 @@ pub struct Options {
     )]
     pub row_group_size: usize,
 
+    #[arg(
+        long,
+        env = "P_EXECUTION_BATCH_SIZE",
+        default_value = "20000",
+        help = "batch size for query execution"
+    )]
+    pub execution_batch_size: usize,
+
     #[arg(
         long = "compression-algo",
         env = "P_PARQUET_COMPRESSION_ALGO",

src/cli.rs

@@ -40,7 +40,19 @@ use super::{Event, DEFAULT_TIMESTAMP_KEY};
 
 pub mod json;
 
-static TIME_FIELD_NAME_PARTS: [&str; 2] = ["time", "date"];
+static TIME_FIELD_NAME_PARTS: [&str; 11] = [
+    "time",
+    "date",
+    "timestamp",
+    "created",
+    "received",
+    "ingested",
+    "collected",
+    "start",
+    "end",
+    "ts",
+    "dt",
+];
 type EventSchema = Vec<Arc<Field>>;
 
 /// Source of the logs, used to perform special processing for certain sources

src/event/format/mod.rs

@@ -76,7 +76,8 @@ impl ParseableServer for QueryServer {
             .service(
                 web::scope(&prism_base_path())
                     .service(Server::get_prism_home())
-                    .service(Server::get_prism_logstream()),
+                    .service(Server::get_prism_logstream())
+                    .service(Server::get_prism_datasets()),
             )
             .service(Server::get_generated());
     }

src/handlers/http/modal/query_server.rs

@@ -95,7 +95,8 @@ impl ParseableServer for Server {
             .service(
                 web::scope(&prism_base_path())
                     .service(Server::get_prism_home())
-                    .service(Server::get_prism_logstream()),
+                    .service(Server::get_prism_logstream())
+                    .service(Server::get_prism_datasets()),
             )
             .service(Self::get_ingest_otel_factory())
             .service(Self::get_generated());
@@ -180,6 +181,17 @@ impl Server {
         )
     }
 
+    pub fn get_prism_datasets() -> Scope {
+        web::scope("/datasets").route(
+            "",
+            web::post()
+                .to(http::prism_logstream::post_datasets)
+                .authorize_for_stream(Action::GetStreamInfo)
+                .authorize_for_stream(Action::GetStats)
+                .authorize_for_stream(Action::GetRetention),
+        )
+    }
+
     pub fn get_metrics_webscope() -> Scope {
         web::scope("/metrics").service(
             web::resource("").route(web::get().to(metrics::get).authorize(Action::Metrics)),

src/handlers/http/modal/server.rs

@@ -17,15 +17,33 @@
  */
 
 use actix_web::{
-    web::{self, Path},
-    Responder,
+    web::{self, Json, Path},
+    HttpRequest, Responder,
 };
 
-use crate::prism::logstream::{get_prism_logstream_info, PrismLogstreamError};
+use crate::{
+    prism::logstream::{get_prism_logstream_info, PrismDatasetRequest, PrismLogstreamError},
+    utils::actix::extract_session_key_from_req,
+};
 
 /// This API is essentially just combining the responses of /info and /schema together
 pub async fn get_info(stream_name: Path<String>) -> Result<impl Responder, PrismLogstreamError> {
     let prism_logstream_info = get_prism_logstream_info(&stream_name).await?;
 
     Ok(web::Json(prism_logstream_info))
 }
+
+/// A combination of /stats, /retention, /hottier, /info, /counts and /query
+pub async fn post_datasets(
+    dataset_req: Option<Json<PrismDatasetRequest>>,
+    req: HttpRequest,
+) -> Result<impl Responder, PrismLogstreamError> {
+    let session_key = extract_session_key_from_req(&req)?;
+    let dataset = dataset_req
+        .map(|Json(r)| r)
+        .unwrap_or_default()
+        .get_datasets(session_key)
+        .await?;
+
+    Ok(web::Json(dataset))
+}

src/handlers/http/prism_logstream.rs

@@ -149,7 +149,7 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
     let alerts_info = get_alerts_info().await?;
 
     // generate dates for date-wise stats
-    let dates = (0..7)
+    let mut dates = (0..7)
         .map(|i| {
             Utc::now()
                 .checked_sub_signed(chrono::Duration::days(i))
@@ -158,6 +158,7 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
         })
         .map(|date| date.format("%Y-%m-%d").to_string())
         .collect_vec();
+    dates.reverse();
 
     let mut stream_details = Vec::new();