Merge branch 'main' into ingest-refactor

Author: Devdutt Shenoi

.github/workflows/integration-test.yaml

@@ -11,7 +11,7 @@ jobs:
 
   docker-compose-test:
     name: Quest Smoke and Load Tests for Standalone deployments
-    runs-on: self-hosted
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -23,7 +23,7 @@ jobs:
 
   docker-compose-distributed-test:
     name: Quest Smoke and Load Tests for Distributed deployments
-    runs-on: self-hosted
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
         uses: actions/checkout@v4

.gitignore

@@ -10,7 +10,7 @@ key.pem
 helm-releases/.DS_Store
 .DS_Store
 env-file
-parseable
+parseable/*
 parseable_*
 parseable-env-secret
 cache

Dockerfile.debug

@@ -34,6 +34,8 @@ RUN cargo build
 # final stage
 FROM docker.io/debian:bookworm-slim
 
+RUN apt update && apt install -y curl
+
 WORKDIR /parseable
 
 # Copy the static binary into the final image

docker-compose-distributed-test-with-kafka.yaml

@@ -11,8 +11,10 @@ services:
     volumes:
       - ./parseable-ingest-haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
     depends_on:
-      - parseable-ingest-one
-      - parseable-ingest-two
+      parseable-ingest-one:
+        condition: service_healthy
+      parseable-ingest-two:
+        condition: service_healthy
     networks:
       - parseable-internal
     healthcheck:
@@ -28,7 +30,7 @@ services:
 
   # minio
   minio:
-    image: minio/minio:RELEASE.2023-02-10T18-48-39Z
+    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
     entrypoint:
       - sh
       - -euc
@@ -55,7 +57,7 @@ services:
   parseable-query:
     build:
       context: .
-      dockerfile: Dockerfile
+      dockerfile: Dockerfile.debug
     platform: linux/amd64
     command: [ "parseable", "s3-store" ]
     ports:
@@ -72,6 +74,7 @@ services:
       - P_CHECK_UPDATE=false
       - P_PARQUET_COMPRESSION_ALGO=snappy
       - P_MODE=query
+      - RUST_LOG=warn
     networks:
       - parseable-internal
     healthcheck:
@@ -210,9 +213,12 @@ services:
     networks:
       - parseable-internal
     depends_on:
-      - parseable-query
-      - parseable-ingest-haproxy
-      - minio
+      parseable-query:
+        condition: service_healthy
+      parseable-ingest-haproxy:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
     deploy:
       restart_policy:
         condition: on-failure

docker-compose-distributed-test.yaml

@@ -4,7 +4,7 @@ networks:
 services:
   # minio
   minio:
-    image: minio/minio:RELEASE.2023-02-10T18-48-39Z
+    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
     entrypoint:
       - sh
       - -euc
@@ -28,7 +28,7 @@ services:
   parseable-query:
     build:
       context: .
-      dockerfile: Dockerfile
+      dockerfile: Dockerfile.debug
     platform: linux/amd64
     command: [ "parseable", "s3-store" ]
     ports:
@@ -45,6 +45,7 @@ services:
       - P_CHECK_UPDATE=false
       - P_PARQUET_COMPRESSION_ALGO=snappy
       - P_MODE=query
+      - RUST_LOG=warn
     networks:
       - parseable-internal
     healthcheck:
@@ -53,7 +54,8 @@ services:
       timeout: 20s
       retries: 5
     depends_on:
-      - minio
+      minio:
+        condition: service_healthy
     deploy:
       restart_policy:
         condition: on-failure
@@ -63,7 +65,7 @@ services:
   parseable-ingest-one:
     build:
       context: .
-      dockerfile: Dockerfile
+      dockerfile: Dockerfile.debug
     platform: linux/amd64
     command: [ "parseable", "s3-store", ]
     ports:
@@ -81,6 +83,7 @@ services:
       - P_PARQUET_COMPRESSION_ALGO=snappy
       - P_MODE=ingest
       - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000
+      - RUST_LOG=warn
     networks:
       - parseable-internal
     healthcheck:
@@ -89,8 +92,10 @@ services:
       timeout: 20s
       retries: 5
     depends_on:
-      - parseable-query
-      - minio
+      parseable-query:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
     deploy:
       restart_policy:
         condition: on-failure
@@ -121,9 +126,12 @@ services:
     networks:
       - parseable-internal
     depends_on:
-      - parseable-query
-      - parseable-ingest-one
-      - minio
+      parseable-query:
+        condition: service_healthy
+      parseable-ingest-one:
+        condition: service_healthy
+      minio:
+        condition: service_healthy
     deploy:
       restart_policy:
         condition: on-failure

docker-compose-test-with-kafka.yaml

@@ -49,8 +49,10 @@ services:
       - P_KAFKA_BOOTSTRAP_SERVERS=kafka-0:9092
       # additional settings like security, tuning, etc.
     depends_on:
-      - minio
-      - kafka-0
+      minio:
+        condition: service_healthy
+      kafka-0:
+        condition: service_healthy
     healthcheck:
       test: [ "CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness" ]
       interval: 15s
@@ -82,7 +84,8 @@ services:
       "parseable"
     ]
     depends_on:
-      - parseable
+      parseable:
+        condition: service_healthy
     networks:
       - parseable-internal
     deploy:
@@ -146,14 +149,15 @@ services:
   kafka-log-generator:
     build:
       context: ./scripts
-      dockerfile: Dockerfile
+      dockerfile: Dockerfile.debug
     environment:
       - KAFKA_BROKERS=kafka-0:9092
       - KAFKA_TOPIC=test-logs-stream
       - LOG_RATE=5000
       - TOTAL_LOGS=500_000
     depends_on:
-      - kafka-0
+      kafka-0:
+        condition: service_healthy
     networks:
       - parseable-internal
     deploy:

docker-compose-test.yaml

@@ -27,7 +27,7 @@ services:
   parseable:
     build:
       context: .
-      dockerfile: Dockerfile
+      dockerfile: Dockerfile.debug
     platform: linux/amd64
     command: [ "parseable", "s3-store", ]
     ports:
@@ -43,8 +43,10 @@ services:
       - P_PASSWORD=parseableadmin
       - P_CHECK_UPDATE=false
       - P_PARQUET_COMPRESSION_ALGO=snappy
+      - RUST_LOG=warn
     depends_on:
-      - minio
+      minio:
+        condition: service_healthy
     healthcheck:
       test: [ "CMD", "curl", "-f", "http://localhost:8000/api/v1/liveness" ]
       interval: 15s
@@ -60,8 +62,8 @@ services:
 
   quest:
     image: ghcr.io/parseablehq/quest:main
-    pull_policy: always
     platform: linux/amd64
+    pull_policy: always
     command: [
       "load",
       "http://parseable:8000",
@@ -76,7 +78,8 @@ services:
       "parseable"
     ]
     depends_on:
-      - parseable
+      parseable:
+        condition: service_healthy
     networks:
       - parseable-internal
     deploy:

src/about.rs

@@ -17,10 +17,6 @@
  *
  */
 
-use crate::analytics;
-use crate::option::Config;
-use crate::storage::StorageMetadata;
-use crate::utils::update::{self, LatestRelease};
 use chrono::Duration;
 use chrono_humanize::{Accuracy, Tense};
 use crossterm::style::Stylize;
@@ -30,6 +26,11 @@ use std::path::Path;
 use sysinfo::System;
 use ulid::Ulid;
 
+use crate::analytics;
+use crate::cli::Options;
+use crate::storage::StorageMetadata;
+use crate::utils::update::{self, LatestRelease};
+
 // Expose some static variables for internal usage
 pub static LATEST_RELEASE: OnceCell<Option<LatestRelease>> = OnceCell::new();
 
@@ -99,7 +100,7 @@ impl ParseableVersion {
 
 pub fn print_about(
     current_version: semver::Version,
-    latest_release: Option<update::LatestRelease>,
+    latest_release: Option<LatestRelease>,
     commit_hash: String,
 ) {
     eprint!(
@@ -123,7 +124,7 @@ pub fn print_about(
     );
 }
 
-fn print_latest_release(latest_release: update::LatestRelease) {
+fn print_latest_release(latest_release: LatestRelease) {
     let time_since_latest_release = chrono::Utc::now() - latest_release.date;
     let time_since_latest_release = humanize_time(time_since_latest_release);
     let fmt_latest_version = format!(
@@ -133,10 +134,10 @@ fn print_latest_release(latest_release: update::LatestRelease) {
     eprint!("{}", fmt_latest_version.red());
 }
 
-pub async fn print(config: &Config, meta: &StorageMetadata) {
+pub async fn print(options: &Options, meta: &StorageMetadata) {
     // print current version
     let current = current();
-    let latest_release = if config.options.check_update {
+    let latest_release = if options.check_update {
         update::get_latest(&meta.deployment_id).await.ok()
     } else {
         None

src/alerts/mod.rs

@@ -27,7 +27,7 @@ use once_cell::sync::Lazy;
 use serde_json::Error as SerdeError;
 use std::collections::{HashMap, HashSet};
 use std::fmt::{self, Display};
-use tokio::sync::oneshot::{Receiver, Sender};
+use tokio::sync::oneshot::{self, Receiver, Sender};
 use tokio::sync::RwLock;
 use tokio::task::JoinHandle;
 use tracing::{trace, warn};
@@ -36,7 +36,7 @@ use ulid::Ulid;
 pub mod alerts_utils;
 pub mod target;
 
-use crate::option::CONFIG;
+use crate::parseable::PARSEABLE;
 use crate::query::{TableScanVisitor, QUERY_SESSION};
 use crate::rbac::map::SessionKey;
 use crate::storage;
@@ -650,8 +650,8 @@ impl AlertConfig {
     fn get_context(&self) -> Context {
         let deployment_instance = format!(
             "{}://{}",
-            CONFIG.options.get_scheme(),
-            CONFIG.options.address
+            PARSEABLE.options.get_scheme(),
+            PARSEABLE.options.address
         );
         let deployment_id = storage::StorageMetadata::global().deployment_id;
         let deployment_mode = storage::StorageMetadata::global().mode.to_string();
@@ -730,13 +730,20 @@ impl Alerts {
     /// Loads alerts from disk, blocks
     pub async fn load(&self) -> Result<(), AlertError> {
         let mut map = self.alerts.write().await;
-        let store = CONFIG.storage().get_object_store();
+        let store = PARSEABLE.storage.get_object_store();
 
         for alert in store.get_alerts().await.unwrap_or_default() {
-            let (handle, rx, tx) =
-                schedule_alert_task(alert.get_eval_frequency(), alert.clone()).await?;
-
-            self.update_task(alert.id, handle, rx, tx).await;
+            let (outbox_tx, outbox_rx) = oneshot::channel::<()>();
+            let (inbox_tx, inbox_rx) = oneshot::channel::<()>();
+            let handle = schedule_alert_task(
+                alert.get_eval_frequency(),
+                alert.clone(),
+                inbox_rx,
+                outbox_tx,
+            )?;
+
+            self.update_task(alert.id, handle, outbox_rx, inbox_tx)
+                .await;
 
             map.insert(alert.id, alert);
         }
@@ -785,7 +792,7 @@ impl Alerts {
         new_state: AlertState,
         trigger_notif: Option<String>,
     ) -> Result<(), AlertError> {
-        let store = CONFIG.storage().get_object_store();
+        let store = PARSEABLE.storage.get_object_store();
 
         // read and modify alert
         let mut alert = self.get_alert_by_id(alert_id).await?;