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] <windmill-internal-app[bot]@users.noreply.github.com>

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"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-ad9e899dfd2ee4e3d18ecf06d016f821968c5a83
+466eb1830879052a5d042295256a78375bee916d

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<Response> {
+    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<DB>,
@@ -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