feat: job search index backed up and loaded from s3 (#4100)

* Add fn to get logs from disk and store as string

* Update dependencies

* Add async

* Update dependecies

* Add missing async

* Move JOB_RETENTION_SECS to windmill_common

* Update ee repo ref

backend/Cargo.lock

@@ -10447,12 +10447,14 @@ name = "windmill-common"
 version = "1.366.6"
 dependencies = [
  "anyhow",
+ "async-stream",
  "aws-config",
  "aws-sdk-sts",
  "axum",
  "bytes",
  "chrono",
  "cron",
+ "futures-core",
  "git-version",
  "hex",
  "hmac",
@@ -10501,13 +10503,17 @@ name = "windmill-indexer"
 version = "1.366.6"
 dependencies = [
  "anyhow",
+ "bytes",
  "chrono",
  "futures",
+ "object_store",
  "serde",
  "serde_json",
  "sqlx",
  "tantivy",
+ "tempfile",
  "tokio",
+ "tokio-tar",
  "tracing",
  "uuid 1.10.0",
  "windmill-common",

backend/Cargo.toml

@@ -39,7 +39,7 @@ incremental = true
 
 [features]
 default = []
-enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise"]
+enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise", "windmill-common/prometheus", "windmill-common/enterprise", "windmill-indexer/enterprise"]
 enterprise_saml = ["windmill-api/enterprise_saml"]
 stripe = ["windmill-api/stripe"]
 benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark"]
@@ -47,7 +47,7 @@ flamegraph = ["windmill-common/flamegraph", "windmill-worker/flamegraph"]
 loki = ["windmill-common/loki"]
 pg_embed = ["dep:pg-embed"]
 embedding = ["windmill-api/embedding"]
-parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/parquet"]
+parquet = ["windmill-api/parquet", "windmill-common/parquet", "windmill-worker/parquet", "windmill-indexer/parquet"]
 prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill-worker/prometheus", "windmill-queue/prometheus"]
 flow_testing = ["windmill-worker/flow_testing"]
 openidconnect = ["windmill-api/openidconnect"]

backend/ee-repo-ref.txt

@@ -1 +1 @@
-8e61cafa4ee95c60eb55e5ebeb07d7b1c2947eae
+89e36e9e73596926e9e02ce27aa7db14d840a255

backend/src/main.rs

@@ -371,7 +371,7 @@ Windmill Community Edition {GIT_VERSION}
         let should_index_jobs = false;
 
         let (index_reader, index_writer) = if should_index_jobs {
-            let (r, w) = windmill_indexer::indexer_ee::init_index()?;
+            let (r, w) = windmill_indexer::indexer_ee::init_index().await?;
             (Some(r), Some(w))
         } else {
             (None, None)

backend/src/monitor.rs

@@ -45,7 +45,7 @@ use windmill_common::{
         load_worker_config, reload_custom_tags_setting, DEFAULT_TAGS_PER_WORKSPACE,
         DEFAULT_TAGS_WORKSPACES, SERVER_CONFIG, WORKER_CONFIG,
     },
-    BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL,
+    BASE_URL, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL, HUB_BASE_URL, JOB_RETENTION_SECS,
     METRICS_DEBUG_ENABLED, METRICS_ENABLED,
 };
 use windmill_queue::cancel_job;
@@ -105,8 +105,6 @@ lazy_static::lazy_static! {
         "Number of jobs in the queue",
         &["tag"]
     ).unwrap();
-
-    static ref JOB_RETENTION_SECS: Arc<RwLock<i64>> = Arc::new(RwLock::new(0));
 }
 
 pub async fn initial_load(

backend/windmill-common/Cargo.toml

@@ -11,7 +11,7 @@ jemalloc = ["dep:tikv-jemalloc-ctl"]
 prometheus = ["dep:prometheus"]
 flamegraph = ["dep:tracing-flame"]
 loki = ["dep:tracing-loki"]
-parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts", "dep:bytes"]
+parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
 
 [lib]
 name = "windmill_common"
@@ -48,8 +48,10 @@ prometheus = { workspace = true, optional = true }
 aws-config = { workspace = true, optional = true }
 aws-sdk-sts = { workspace = true, optional = true }
 indexmap.workspace = true
-bytes = { workspace = true, optional = true }
+bytes.workspace = true
 mail-send.workspace = true
+futures-core.workspace = true
+async-stream.workspace = true
 
 [target.'cfg(not(target_env = "msvc"))'.dependencies]
 tikv-jemalloc-ctl = { optional = true, workspace = true }

backend/windmill-common/src/jobs.rs

@@ -1,20 +1,23 @@
 use std::collections::HashMap;
 
+use bytes::Bytes;
+use futures_core::Stream;
 use indexmap::IndexMap;
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
 use sqlx::{types::Json, Pool, Postgres, Transaction};
+use tokio::io::AsyncReadExt;
 use uuid::Uuid;
 
 pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE";
 
 use crate::{
-    error::{self, Error},
+    error::{self, to_anyhow, Error},
     flow_status::{FlowStatus, RestartedFrom},
     flows::{FlowValue, Retry},
     get_latest_deployed_hash_for_path,
     scripts::{ScriptHash, ScriptLang},
-    worker::to_raw_value,
+    worker::{to_raw_value, TMP_DIR},
 };
 
 #[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone)]
@@ -580,3 +583,75 @@ pub fn format_completed_job_result(mut cj: CompletedJob) -> CompletedJobWithForm
     cj.result = None; // very important to avoid sending the result twice
     CompletedJobWithFormattedResult { cj, result: Some(sql_result) }
 }
+
+pub async fn get_logs_from_disk(
+    log_offset: i32,
+    logs: &str,
+    log_file_index: &Option<Vec<String>>,
+) -> Option<impl Stream<Item = Result<Bytes, anyhow::Error>>> {
+    if log_offset > 0 {
+        if let Some(file_index) = log_file_index.clone() {
+            for file_p in &file_index {
+                if !tokio::fs::metadata(format!("{TMP_DIR}/{file_p}"))
+                    .await
+                    .is_ok()
+                {
+                    return None;
+                }
+            }
+
+            let logs = logs.to_string();
+            let stream = async_stream::stream! {
+                for file_p in file_index.clone() {
+                    let mut file = tokio::fs::File::open(format!("{TMP_DIR}/{file_p}")).await.map_err(to_anyhow)?;
+                    let mut buffer = Vec::new();
+                    file.read_to_end(&mut buffer).await.map_err(to_anyhow)?;
+                    yield Ok(bytes::Bytes::from(buffer)) as anyhow::Result<bytes::Bytes>;
+                }
+
+                yield Ok(bytes::Bytes::from(logs))
+            };
+            return Some(stream);
+        }
+    }
+    return None;
+}
+
+#[cfg(all(feature = "enterprise", feature = "parquet"))]
+pub async fn get_logs_from_store(
+    log_offset: i32,
+    logs: &str,
+    log_file_index: &Option<Vec<String>>,
+) -> Option<impl Stream<Item = Result<Bytes, object_store::Error>>> {
+    use crate::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
+
+    if log_offset > 0 {
+        if let Some(file_index) = log_file_index.clone() {
+            tracing::debug!("Getting logs from store: {file_index:?}");
+            if let Some(os) = OBJECT_STORE_CACHE_SETTINGS.read().await.clone() {
+                tracing::debug!("object store client present, streaming from there");
+
+                let logs = logs.to_string();
+                let stream = async_stream::stream! {
+                    for file_p in file_index.clone() {
+                        let file_p_2 = file_p.clone();
+                        let file = os.get(&object_store::path::Path::from(file_p)).await;
+                        if let Ok(file) = file {
+                            if let Ok(bytes) = file.bytes().await {
+                                yield Ok(bytes::Bytes::from(bytes)) as object_store::Result<bytes::Bytes>;
+                            }
+                        } else {
+                            tracing::debug!("error getting file from store: {file_p_2}: {}", file.err().unwrap());
+                        }
+                    }
+
+                    yield Ok(bytes::Bytes::from(logs))
+                };
+                return Some(stream);
+            } else {
+                tracing::debug!("object store client not present, cannot stream logs from store");
+            }
+        }
+    }
+    return None;
+}

backend/windmill-common/src/lib.rs

@@ -79,6 +79,9 @@ lazy_static::lazy_static! {
 
 
     pub static ref CRITICAL_ERROR_CHANNELS: Arc<RwLock<Vec<CriticalErrorChannel>>> = Arc::new(RwLock::new(vec![]));
+
+    pub static ref JOB_RETENTION_SECS: Arc<RwLock<i64>> = Arc::new(RwLock::new(0));
+
 }
 
 pub async fn shutdown_signal(

backend/windmill-indexer/Cargo.toml

@@ -11,6 +11,8 @@ path = "src/lib.rs"
 [features]
 default = []
 tantivy = ["dep:tantivy"]
+parquet = ["dep:object_store"]
+enterprise = []
 
 [dependencies]
 windmill-common.workspace = true
@@ -24,3 +26,7 @@ tracing.workspace = true
 chrono.workspace = true
 uuid.workspace = true
 futures.workspace = true
+tempfile.workspace = true
+bytes.workspace = true
+object_store = { workspace = true, optional = true}
+tokio-tar.workspace = true

backend/windmill-indexer/src/indexer_ee.rs

@@ -8,7 +8,7 @@ pub struct IndexReader;
 #[derive(Clone)]
 pub struct IndexWriter;
 
-pub fn init_index() -> Result<(IndexReader, IndexWriter), Error> {
+pub async fn init_index() -> Result<(IndexReader, IndexWriter), Error> {
     Err(anyhow!("Cannot initialize index: not in EE").into())
 }