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

Author: de-sh

.github/workflows/build-push-edge-debug.yaml

@@ -45,5 +45,3 @@ jobs:
           push: true
           tags: parseable/parseable:edge-debug
           platforms: linux/amd64,linux/arm64
-          cache-from: type=gha
-          cache-to: type=gha,mode=max

.github/workflows/build-push-edge.yaml

@@ -45,5 +45,3 @@ jobs:
           push: true
           tags: parseable/parseable:edge
           platforms: linux/amd64,linux/arm64
-          cache-from: type=gha
-          cache-to: type=gha,mode=max

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "parseable"
-version = "1.7.3"
+version = "1.7.5"
 authors = ["Parseable Team <[email protected]>"]
 edition = "2021"
 rust-version = "1.83.0"
@@ -128,9 +128,8 @@ sha1_smol = { version = "1.0", features = ["std"] }
 static-files = "0.2"
 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]
@@ -139,8 +138,8 @@ arrow = "54.0.0"
 temp-dir = "0.1.14"
 
 [package.metadata.parseable_ui]
-assets-url = "https://github.com/parseablehq/console/releases/download/v0.9.18/build.zip"
-assets-sha1 = "4516db38c8e556707b29b33569f9b1e53d5165f2"
+assets-url = "https://github.com/parseablehq/console/releases/download/v0.9.20/build.zip"
+assets-sha1 = "58eb74bdb6727b5df04d6f9f339cf370c0bf4eec"
 
 [features]
 debug = []

Cargo.toml

@@ -14,31 +14,23 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 # build stage
-FROM  rust:1.84.0-bookworm AS builder
-
+FROM rust:1.84.0-bookworm AS builder
 
 LABEL org.opencontainers.image.title="Parseable"
 LABEL maintainer="Parseable Team <[email protected]>"
 LABEL org.opencontainers.image.vendor="Parseable Inc"
 LABEL org.opencontainers.image.licenses="AGPL-3.0"
 
 WORKDIR /parseable
-
-# Cache dependencies
-COPY Cargo.toml Cargo.lock build.rs ./
-RUN mkdir src && echo "fn main() {}" > src/main.rs && cargo build --release && rm -rf src
-
-# Build the actual binary
-COPY src ./src
-COPY resources ./resources
+COPY . .
 RUN cargo build --release
 
 # final stage
 FROM gcr.io/distroless/cc-debian12:latest
 
 WORKDIR /parseable
 
-# Copy the static binary into the final image
+# Copy the static shell into base image.
 COPY --from=builder /parseable/target/release/parseable /usr/bin/parseable
 
 CMD ["/usr/bin/parseable"]

Dockerfile

@@ -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

@@ -41,7 +41,19 @@ use super::{Event, DEFAULT_TIMESTAMP_KEY};
 pub mod json;
 pub mod known_schema;
 
-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

@@ -74,7 +74,7 @@ pub enum SchemaVersion {
     V1,
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct LogStreamMetadata {
     pub schema_version: SchemaVersion,
     pub schema: HashMap<String, Arc<Field>>,

src/metadata.rs

@@ -171,7 +171,7 @@ impl Parseable {
         }
 
         // Gets write privileges only for creating the stream when it doesn't already exist.
-        self.streams.create(
+        self.streams.get_or_create(
             self.options.clone(),
             stream_name.to_owned(),
             LogStreamMetadata::default(),
@@ -342,7 +342,7 @@ impl Parseable {
             schema_version,
             log_source,
         );
-        self.streams.create(
+        self.streams.get_or_create(
             self.options.clone(),
             stream_name.to_string(),
             metadata,
@@ -652,7 +652,7 @@ impl Parseable {
                     SchemaVersion::V1, // New stream
                     log_source,
                 );
-                self.streams.create(
+                self.streams.get_or_create(
                     self.options.clone(),
                     stream_name.to_string(),
                     metadata,