Merge remote-tracking branch 'origin/main'

Author: Devdutt Shenoi

.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,30 +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 . .
 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",
@@ -295,6 +303,14 @@ pub struct Options {
     )]
     pub ingestor_endpoint: String,
 
+    #[arg(
+        long,
+        env = "P_INDEXER_ENDPOINT",
+        default_value = "",
+        help = "URL to connect to this specific indexer. Default is the address of the server"
+    )]
+    pub indexer_endpoint: String,
+
     #[command(flatten)]
     pub oidc: Option<OidcConfig>,
 
@@ -396,29 +412,47 @@ impl Options {
     }
 
     /// TODO: refactor and document
-    pub fn get_url(&self) -> Url {
-        if self.ingestor_endpoint.is_empty() {
-            return format!(
-                "{}://{}",
-                self.get_scheme(),
-                self.address
-            )
-            .parse::<Url>() // if the value was improperly set, this will panic before hand
-            .unwrap_or_else(|err| {
-                panic!("{err}, failed to parse `{}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details.", self.address)
-            });
-        }
-
-        let ingestor_endpoint = &self.ingestor_endpoint;
+    pub fn get_url(&self, mode: Mode) -> Url {
+        let (endpoint, env_var) = match mode {
+            Mode::Ingest => {
+                if self.ingestor_endpoint.is_empty() {
+                    return format!(
+                        "{}://{}",
+                        self.get_scheme(),
+                        self.address
+                    )
+                    .parse::<Url>() // if the value was improperly set, this will panic before hand
+                    .unwrap_or_else(|err| {
+                        panic!("{err}, failed to parse `{}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details.", self.address)
+                    });
+                }
+                (&self.ingestor_endpoint, "P_INGESTOR_ENDPOINT")
+            }
+            Mode::Index => {
+                if self.indexer_endpoint.is_empty() {
+                    return format!(
+                        "{}://{}",
+                        self.get_scheme(),
+                        self.address
+                    )
+                    .parse::<Url>() // if the value was improperly set, this will panic before hand
+                    .unwrap_or_else(|err| {
+                        panic!("{err}, failed to parse `{}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details.", self.address)
+                    });
+                }
+                (&self.indexer_endpoint, "P_INDEXER_ENDPOINT")
+            }
+            _ => panic!("Invalid mode"),
+        };
 
-        if ingestor_endpoint.starts_with("http") {
-            panic!("Invalid value `{}`, please set the environement variable `P_INGESTOR_ENDPOINT` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", ingestor_endpoint);
+        if endpoint.starts_with("http") {
+            panic!("Invalid value `{}`, please set the environement variable `{env_var}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", endpoint);
         }
 
-        let addr_from_env = ingestor_endpoint.split(':').collect::<Vec<&str>>();
+        let addr_from_env = endpoint.split(':').collect::<Vec<&str>>();
 
         if addr_from_env.len() != 2 {
-            panic!("Invalid value `{}`, please set the environement variable `P_INGESTOR_ENDPOINT` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", ingestor_endpoint);
+            panic!("Invalid value `{}`, please set the environement variable `{env_var}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", endpoint);
         }
 
         let mut hostname = addr_from_env[0].to_string();

src/cli.rs

@@ -16,18 +16,21 @@
  *
  */
 
-use std::sync::Arc;
-
 use async_trait::async_trait;
 use futures_util::StreamExt;
 use rdkafka::consumer::{CommitMode, Consumer};
 use serde_json::Value;
+use std::collections::HashMap;
+use std::sync::Arc;
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{debug, error};
 
 use crate::{
     connectors::common::processor::Processor,
-    event::format::{json, EventFormat, LogSource, LogSourceEntry},
+    event::{
+        format::{json, EventFormat, LogSource, LogSourceEntry},
+        USER_AGENT_KEY,
+    },
     parseable::PARSEABLE,
     storage::StreamType,
 };
@@ -64,13 +67,15 @@ impl ParseableSinkProcessor {
         }
 
         let stream = PARSEABLE.get_or_create_stream(stream_name);
+        let mut p_custom_fields = HashMap::new();
+        p_custom_fields.insert(USER_AGENT_KEY.to_string(), "kafka".to_string());
 
         json::Event::new(
             Value::Array(json_vec),
             total_payload_size,
             LogSource::Custom("Kafka".to_owned()),
         )
-        .into_event(&stream)?
+        .into_event(&stream, p_custom_fields)?
         .process(&stream)?;
 
         Ok(total_payload_size)

src/connectors/kafka/processor.rs

@@ -1,5 +1,6 @@
 use std::{collections::HashMap, path::PathBuf, sync::Arc};
 
+use chrono::{TimeZone, Utc};
 use datafusion::{common::Column, prelude::Expr};
 use itertools::Itertools;
 use relative_path::RelativePathBuf;
@@ -119,14 +120,35 @@ pub async fn fetch_parquet_file_paths(
 
     selected_files
         .into_iter()
-        .map(|file| {
+        .filter_map(|file| {
             let date = file.file_path.split("/").collect_vec();
 
-            let date = date.as_slice()[1..4].iter().map(|s| s.to_string());
-
-            let date = RelativePathBuf::from_iter(date);
-
-            parquet_files.entry(date).or_default().push(file);
+            let year = &date[1][5..9];
+            let month = &date[1][10..12];
+            let day = &date[1][13..15];
+            let hour = &date[2][5..7];
+            let min = &date[3][7..9];
+            let file_date = Utc
+                .with_ymd_and_hms(
+                    year.parse::<i32>().unwrap(),
+                    month.parse::<u32>().unwrap(),
+                    day.parse::<u32>().unwrap(),
+                    hour.parse::<u32>().unwrap(),
+                    min.parse::<u32>().unwrap(),
+                    0,
+                )
+                .unwrap();
+
+            if file_date < time_range.start {
+                None
+            } else {
+                let date = date.as_slice()[1..4].iter().map(|s| s.to_string());
+
+                let date = RelativePathBuf::from_iter(date);
+
+                parquet_files.entry(date).or_default().push(file);
+                Some("")
+            }
         })
         .for_each(|_| {});