From 4b8f3580a4bbfb66f4e98c14cec7b98954abbc58 Mon Sep 17 00:00:00 2001
From: Ruben Fiszel <ruben@rubenfiszel.com>
Date: Wed, 13 Nov 2024 23:24:23 +0100
Subject: [PATCH] add LOGS_TO_STDOUT to ee

---
 backend/ee-repo-ref.txt                      |  2 +-
 backend/windmill-worker/src/handle_child.rs  | 14 ++++++++++++--
 backend/windmill-worker/src/job_logger.rs    |  3 ++-
 backend/windmill-worker/src/job_logger_ee.rs |  9 +++++++++
 4 files changed, 24 insertions(+), 4 deletions(-)

diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt
index 32671f9c64..99887029f1 100644
--- a/backend/ee-repo-ref.txt
+++ b/backend/ee-repo-ref.txt
@@ -1 +1 @@
-053cf0a539c64efbe8eee0e2a330b72114a43193
\ No newline at end of file
+61efa9c019ef66e9ecc5b7ebe77d667f76908f35
\ No newline at end of file
diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs
index 6846e9dfee..bbaabf2611 100644
--- a/backend/windmill-worker/src/handle_child.rs
+++ b/backend/windmill-worker/src/handle_child.rs
@@ -48,6 +48,7 @@ use futures::{
 
 use crate::common::{resolve_job_timeout, OccupancyMetrics};
 use crate::job_logger::{append_job_logs, append_with_limit, LARGE_LOG_THRESHOLD_SIZE};
+use crate::job_logger_ee::process_streaming_log_lines;
 use crate::{MAX_RESULT_SIZE, MAX_WAIT_FOR_SIGINT, MAX_WAIT_FOR_SIGTERM};
 
 lazy_static::lazy_static! {
@@ -338,6 +339,8 @@ pub async fn handle_child(
                         if line.is_empty() {
                             continue;
                         }
+                        #[cfg(feature = "enterprise")]
+                        crate::job_logger_ee::export_job_lines_externally(&line,  &w_id).await;
                         append_with_limit(&mut joined, &line, &mut log_remaining);
                         if log_remaining == 0 {
                             tracing::info!(%job_id, "Too many logs lines for job {job_id}");
@@ -434,6 +437,8 @@ pub async fn handle_child(
     }
 }
 
+
+
 async fn get_mem_peak(pid: Option<u32>, nsjail: bool) -> i32 {
     if pid.is_none() {
         return -1;
@@ -689,19 +694,24 @@ fn child_joined_output_stream(
 
     let stdout = BufReader::new(stdout).lines();
     let stderr = BufReader::new(stderr).lines();
-    stream::select(lines_to_stream(stderr), lines_to_stream(stdout))
+    stream::select(lines_to_stream(stderr, true), lines_to_stream(stdout, false))
 }
 
 pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
     mut lines: tokio::io::Lines<R>,
+    stderr: bool,
 ) -> impl futures::Stream<Item = io::Result<String>> {
     stream::poll_fn(move |cx| {
         std::pin::Pin::new(&mut lines)
             .poll_next_line(cx)
-            .map(|result| result.transpose())
+            .map(|result| {
+                process_streaming_log_lines(result, stderr)
+             })
     })
 }
 
+
+
 pub fn process_status(status: ExitStatus) -> error::Result<()> {
     if status.success() {
         Ok(())
diff --git a/backend/windmill-worker/src/job_logger.rs b/backend/windmill-worker/src/job_logger.rs
index 8a6d824e6e..1919c6555f 100644
--- a/backend/windmill-worker/src/job_logger.rs
+++ b/backend/windmill-worker/src/job_logger.rs
@@ -38,7 +38,7 @@ pub(crate) async fn append_job_logs(
 ) -> () {
     if must_compact_logs {
         #[cfg(all(feature = "enterprise", feature = "parquet"))]
-        s3_storage(job_id, &w_id, &db, &logs, &total_size, &worker_name).await;
+        s3_storage(job_id, &w_id, &db, logs, &total_size, &worker_name).await;
 
         #[cfg(not(all(feature = "enterprise", feature = "parquet")))]
         {
@@ -70,6 +70,7 @@ pub fn append_with_limit(dst: &mut String, src: &str, limit: &mut usize) {
     if *NO_LOGS_AT_ALL {
         return;
     }
+    
     let src_str;
     let src = {
         src_str = RE_00.replace_all(src, "");
diff --git a/backend/windmill-worker/src/job_logger_ee.rs b/backend/windmill-worker/src/job_logger_ee.rs
index 0274c6c777..9cef37c2cb 100644
--- a/backend/windmill-worker/src/job_logger_ee.rs
+++ b/backend/windmill-worker/src/job_logger_ee.rs
@@ -1,3 +1,4 @@
+use std::io;
 use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
 
@@ -23,4 +24,12 @@ pub(crate) async fn default_disk_log_storage(
     tracing::info!("Logs length of {job_id} has exceeded a threshold. Implementation to store excess on disk in not OSS");
 }
 
+#[cfg(feature = "enterprise")]
+pub(crate) async fn export_job_lines_externally(line: &str, w_id: &str) {
+    
+}
 
+
+pub(crate) fn process_streaming_log_lines(r: Result<Option<String>, io::Error>, _stderr: bool) -> Option<Result<String, io::Error>> {
+    r.transpose()
+}
\ No newline at end of file