From 338d75cc5227e352cb84828c99bfd3b984cf0fa5 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 29 Aug 2026 18:47:53 +0200 Subject: [PATCH] feat: serve service log context from parquet and retire the raw log files (#10892) * feat: serve service log context from the parquet store and retire the raw files * fix: keep the log ingest cursor in the store and stream file rebuilds * fix: roll back a partial index rebuild and move the cursor before the commit * fix: make the index rebuild idempotent and repair a cursor the index never caught up with * fix: seed the indexed cursor on upgrade and after a rebuild * fix: fail the indexing pass on an unreadable cursor instead of reading it as absent * docs: record what keeps both known_ts entries, not the path main removed * chore: update ee-repo-ref to 466eb1830879052a5d042295256a78375bee916d This commit updates the EE repository reference after PR #754 was merged in windmill-ee-private. Previous ee-repo-ref: ddb3a536b8d85c134c01f87da7783baaa204a6d1 New ee-repo-ref: 466eb1830879052a5d042295256a78375bee916d Automated by sync-ee-ref workflow. --------- Co-authored-by: windmill-internal-app[bot] --- ...34f1befe6b16eee363c843faa1f836d53ca8d.json | 29 +++++ backend/ee-repo-ref.txt | 2 +- backend/windmill-api/src/service_logs.rs | 104 +++++++++++++++--- 3 files changed, 116 insertions(+), 19 deletions(-) create mode 100644 backend/.sqlx/query-daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d.json diff --git a/backend/.sqlx/query-daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d.json b/backend/.sqlx/query-daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d.json new file mode 100644 index 0000000000..d64f007d7b --- /dev/null +++ b/backend/.sqlx/query-daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT mode::text AS \"mode!\", log_ts FROM log_file WHERE hostname = $1 AND file_path = $2 ORDER BY log_ts DESC LIMIT 1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "mode!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "log_ts", + "type_info": "Timestamp" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null, + false + ] + }, + "hash": "daa5b57290cd1f821a53eebe96434f1befe6b16eee363c843faa1f836d53ca8d" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5c23919dc8..cce83bbdff 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ad9e899dfd2ee4e3d18ecf06d016f821968c5a83 \ No newline at end of file +466eb1830879052a5d042295256a78375bee916d diff --git a/backend/windmill-api/src/service_logs.rs b/backend/windmill-api/src/service_logs.rs index 99dc7c41a8..a54f275f46 100644 --- a/backend/windmill-api/src/service_logs.rs +++ b/backend/windmill-api/src/service_logs.rs @@ -88,6 +88,66 @@ async fn list_files( Ok(Json(rows)) } +/// Rebuild one source log file from the columnar store. +/// +/// Not the original bytes: the store holds a line's fields rather than its text, +/// so the JSON is re-serialized here and key order and whitespace are this +/// writer's. Everything a reader can see survives — the drawer this feeds +/// renders a prettified view of each line either way, and a line that was never +/// JSON comes back exactly as it was written. +#[cfg(all(feature = "tantivy", feature = "private"))] +async fn get_log_file_from_store( + db: &DB, + store: &windmill_indexer::service_logs_store_ee::Store, + path: &str, +) -> windmill_common::error::Result { + let (hostname, file_name) = path + .split_once('/') + .ok_or_else(|| Error::BadRequest("Invalid path".to_string()))?; + + // The store is partitioned by day and mode, neither of which the path + // carries. `log_file` names both, and its primary key starts with hostname. + let file = sqlx::query!( + // `mode!` because the column is NOT NULL and only the cast makes sqlx + // think otherwise; a silent default would look up a `mode=` partition + // that matches nothing and read as a missing file. + "SELECT mode::text AS \"mode!\", log_ts FROM log_file WHERE hostname = $1 AND file_path = $2 ORDER BY log_ts DESC LIMIT 1", + hostname, + file_name + ) + .fetch_optional(db) + .await? + .ok_or_else(|| Error::NotFound(format!("File {path} not found")))?; + + // A row registered by this version carries the minute in the file's own name, + // so the two agree and the second is redundant. One written before the + // uploader derived `log_ts` from the name carries a wall clock instead, and + // those outlive an upgrade by the retention period — which is also what makes + // the `ORDER BY` above worth having. The name is authoritative, so both go. + let mut known_ts = vec![chrono::DateTime::from_naive_utc_and_offset( + file.log_ts, + chrono::Utc, + )]; + if let Some(named) = file_name.rsplit('.').next().and_then(|s| { + chrono::NaiveDateTime::parse_from_str(s, windmill_common::tracing_init::LOG_TIMESTAMP_FMT) + .ok() + }) { + known_ts.push(chrono::DateTime::from_naive_utc_and_offset( + named, + chrono::Utc, + )); + } + + let text = windmill_indexer::service_logs_store_ee::read_log_file( + store, &file.mode, hostname, file_name, &known_ts, + ) + .await + .map_err(|e| Error::internal_err(format!("Error reading the service log store: {e}")))? + .ok_or_else(|| Error::NotFound(format!("File {path} not found")))?; + + Ok(content_plain(Body::from(text))) +} + async fn get_log_file( authed: ApiAuthed, Extension(db): Extension, @@ -104,27 +164,30 @@ async fn get_log_file( let s3_client = windmill_object_store::get_object_store().await; #[cfg(feature = "parquet")] if let Some(s3_client) = s3_client { - let path = format!("{}{}", windmill_common::tracing_init::LOGS_SERVICE, path); - let file = s3_client + use windmill_object_store::object_store_reexports::ObjectStoreError; + + // The raw file, for as long as it is there. It outlives its ingestion by + // one indexer pass at most, so this covers the most recent minutes of a + // host's logs byte for byte; everything older is rebuilt from the store. + let object_path = format!("{}{}", windmill_common::tracing_init::LOGS_SERVICE, path); + match s3_client .get(&windmill_object_store::object_store_reexports::Path::from( - path, + object_path, )) - .await; - match file { - Ok(file) => { - let bytes = file.bytes().await; - match bytes { - Ok(bytes) => { - return Ok(content_plain(Body::from(bytes::Bytes::from(bytes)))); - } - Err(e) => { - return Err(Error::internal_err(format!( - "Error pulling the bytes: {}", - e - ))); - } + .await + { + Ok(file) => match file.bytes().await { + Ok(bytes) => { + return Ok(content_plain(Body::from(bytes::Bytes::from(bytes)))); } - } + Err(e) => { + return Err(Error::internal_err(format!( + "Error pulling the bytes: {}", + e + ))); + } + }, + Err(ObjectStoreError::NotFound { .. }) => {} Err(e) => { return Err(Error::internal_err(format!( "Error fetching the file: {}", @@ -132,6 +195,11 @@ async fn get_log_file( ))); } } + + #[cfg(all(feature = "tantivy", feature = "private"))] + return get_log_file_from_store(&db, &s3_client, &path).await; + #[cfg(not(all(feature = "tantivy", feature = "private")))] + return Err(Error::NotFound(format!("File {path} not found"))); } let full_path = format!("{}{}", *TMP_WINDMILL_LOGS_SERVICE, path); // SECURITY (defense in depth): refuse to read through a symlink so a planted