From 200e22c026cf782511f185edd4eb20a6517ee4bf Mon Sep 17 00:00:00 2001 From: wendrul <53628737+wendrul@users.noreply.github.com> Date: Tue, 23 Jul 2024 19:35:29 +0200 Subject: [PATCH] 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 | 6 ++ backend/Cargo.toml | 4 +- backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 2 +- backend/src/monitor.rs | 4 +- backend/windmill-common/Cargo.toml | 6 +- backend/windmill-common/src/jobs.rs | 79 +++++++++++++++++++++- backend/windmill-common/src/lib.rs | 3 + backend/windmill-indexer/Cargo.toml | 6 ++ backend/windmill-indexer/src/indexer_ee.rs | 2 +- 10 files changed, 102 insertions(+), 12 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 96828063a6..f3c2462300 100644 --- a/backend/Cargo.lock +++ b/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", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 9c4c6a318a..6bf07a8f05 100644 --- a/backend/Cargo.toml +++ b/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"] diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 61f887891e..01edb4570b 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -8e61cafa4ee95c60eb55e5ebeb07d7b1c2947eae \ No newline at end of file +89e36e9e73596926e9e02ce27aa7db14d840a255 diff --git a/backend/src/main.rs b/backend/src/main.rs index 853559167a..57176f4c31 100644 --- a/backend/src/main.rs +++ b/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) diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 5aa6200061..99d241bdca 100644 --- a/backend/src/monitor.rs +++ b/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> = Arc::new(RwLock::new(0)); } pub async fn initial_load( diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 634b3ae4e3..e7cd9d1432 100644 --- a/backend/windmill-common/Cargo.toml +++ b/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 } diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index ab9a52a602..e9c099a73c 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/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>, +) -> Option>> { + 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; + } + + 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>, +) -> Option>> { + 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; + } + } 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; +} diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 54b143fcac..d47fd06c61 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -79,6 +79,9 @@ lazy_static::lazy_static! { pub static ref CRITICAL_ERROR_CHANNELS: Arc>> = Arc::new(RwLock::new(vec![])); + + pub static ref JOB_RETENTION_SECS: Arc> = Arc::new(RwLock::new(0)); + } pub async fn shutdown_signal( diff --git a/backend/windmill-indexer/Cargo.toml b/backend/windmill-indexer/Cargo.toml index 2380dd4df2..bca3efc880 100644 --- a/backend/windmill-indexer/Cargo.toml +++ b/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 diff --git a/backend/windmill-indexer/src/indexer_ee.rs b/backend/windmill-indexer/src/indexer_ee.rs index 39b1770f97..79bcbcff77 100644 --- a/backend/windmill-indexer/src/indexer_ee.rs +++ b/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()) }