improve otel

backend/ee-repo-ref.txt

@@ -1 +1 @@
-1d7ee3745e81082c196c4aea0392c88e3de6f9d4
+a07bc62582c809457f1c945d6cda145770b94d04

backend/src/main.rs

@@ -8,7 +8,7 @@
 
 use anyhow::Context;
 use monitor::{
-    load_otel, reload_delete_logs_periodically_setting, reload_indexer_config,
+    load_base_url, load_otel, reload_delete_logs_periodically_setting, reload_indexer_config,
     reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
     send_logs_to_object_store,
 };
@@ -348,10 +348,21 @@ async fn windmill_main() -> anyhow::Result<()> {
     let db = windmill_common::connect_db(server_mode, indexer_mode).await?;
 
     load_otel(&db).await;
+
     tracing::info!("Database connected");
 
+    let environment = load_base_url(&db)
+        .await
+        .unwrap_or_else(|_| "local".to_string())
+        .trim_start_matches("https://")
+        .trim_start_matches("http://")
+        .split(".")
+        .next()
+        .unwrap_or_else(|| "local")
+        .to_string();
+
     #[cfg(not(feature = "flamegraph"))]
-    let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode);
+    let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode, &environment);
 
     let num_version = sqlx::query_scalar!("SELECT version()").fetch_one(&db).await;
 

backend/src/monitor.rs

@@ -1397,7 +1397,7 @@ pub async fn reload_worker_config(
     }
 }
 
-pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
+pub async fn load_base_url(db: &DB) -> error::Result<String> {
     let q_base_url = load_value_from_global_settings(db, BASE_URL_SETTING).await?;
 
     let std_base_url = std::env::var("BASE_URL")
@@ -1421,6 +1421,14 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
         std_base_url
     };
 
+    {
+        let mut l = BASE_URL.write().await;
+        *l = base_url.clone();
+    }
+    Ok(base_url)
+}
+
+pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
     let q_oauth = load_value_from_global_settings(db, OAUTH_SETTING).await?;
 
     let oauths = if let Some(q) = q_oauth {
@@ -1434,6 +1442,7 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
         None
     };
 
+    let base_url = load_base_url(db).await?;
     let is_secure = base_url.starts_with("https://");
 
     {
@@ -1443,11 +1452,6 @@ pub async fn reload_base_url_setting(db: &DB) -> error::Result<()> {
         .unwrap();
     }
 
-    {
-        let mut l = BASE_URL.write().await;
-        *l = base_url
-    }
-
     {
         let mut l = IS_SECURE.write().await;
         *l = is_secure;

backend/tests/worker.rs

@@ -80,6 +80,7 @@ async fn initialize_tracing() {
         let _ = windmill_common::tracing_init::initialize_tracing(
             "test",
             &windmill_common::utils::Mode::Standalone,
+            "test",
         );
     });
 }

backend/windmill-common/src/otel_ee.rs

@@ -38,7 +38,7 @@ pub trait FutureExt: Sized {
 
 use tracing_subscriber::EnvFilter;
 
-pub(crate) fn init_logs_bridge(_mode: &Mode, _hostname: &str) -> Option<EnvFilter> {
+pub(crate) fn init_logs_bridge(_mode: &Mode, _hostname: &str, _env: &str) -> Option<EnvFilter> {
     None
 }
 
@@ -46,11 +46,12 @@ pub(crate) fn init_logs_bridge(_mode: &Mode, _hostname: &str) -> Option<EnvFilte
 pub(crate) fn init_otlp_tracer(
     _mode: &Mode,
     _hostname: &str,
+    _env: &str,
 ) -> Option<opentelemetry_sdk::trace::Tracer> {
     None
 }
 
-pub(crate) fn init_meter_provider(_mode: &Mode, _hostname: &str) -> OtelProvider {
+pub(crate) fn init_meter_provider(_mode: &Mode, _hostname: &str, _env: &str) -> OtelProvider {
     None
 }
 

backend/windmill-common/src/tracing_init.rs

@@ -47,6 +47,7 @@ pub const TMP_WINDMILL_LOGS_SERVICE: &str = concatcp!("/tmp/windmill/", LOGS_SER
 pub fn initialize_tracing(
     hostname: &str,
     mode: &Mode,
+    environment: &str,
 ) -> (WorkerGuard, crate::otel_ee::OtelProvider) {
     let style = std::env::var("RUST_LOG_STYLE").unwrap_or_else(|_| "auto".into());
 
@@ -57,16 +58,16 @@ pub fn initialize_tracing(
         )
     }
 
-    let meter_provider = crate::otel_ee::init_meter_provider(mode, hostname);
+    let meter_provider = crate::otel_ee::init_meter_provider(mode, hostname, environment);
 
     #[cfg(all(feature = "otel", feature = "enterprise"))]
-    let opentelemetry = crate::otel_ee::init_otlp_tracer(mode, hostname)
+    let opentelemetry = crate::otel_ee::init_otlp_tracer(mode, hostname, environment)
         .map(|x| tracing_opentelemetry::layer().with_tracer(x));
 
     #[cfg(not(all(feature = "otel", feature = "enterprise")))]
     let opentelemetry: Option<EnvFilter> = None;
 
-    let logs_bridge = crate::otel_ee::init_logs_bridge(&mode, hostname);
+    let logs_bridge = crate::otel_ee::init_logs_bridge(&mode, hostname, environment);
 
     use tracing_appender::rolling::{RollingFileAppender, Rotation};
 

backend/windmill-worker/src/handle_child.rs

@@ -126,7 +126,7 @@ pub async fn handle_child(
     let (tx, rx) = broadcast::channel::<()>(3);
     let mut rx2 = tx.subscribe();
 
-    let output = child_joined_output_stream(&mut child);
+    let output = child_joined_output_stream(&mut child, job_id.clone());
 
     let job_id = job_id.clone();
 
@@ -677,6 +677,7 @@ where
 /// builds a stream joining both stdout and stderr each read line by line
 fn child_joined_output_stream(
     child: &mut Child,
+    job_id: Uuid,
 ) -> impl stream::FusedStream<Item = io::Result<String>> {
     let stderr = child
         .stderr
@@ -691,19 +692,20 @@ fn child_joined_output_stream(
     let stdout = BufReader::new(stdout).lines();
     let stderr = BufReader::new(stderr).lines();
     stream::select(
-        lines_to_stream(stderr, true),
-        lines_to_stream(stdout, false),
+        lines_to_stream(stderr, true, job_id.clone()),
+        lines_to_stream(stdout, false, job_id),
     )
 }
 
 pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
     mut lines: tokio::io::Lines<R>,
     stderr: bool,
+    job_id: Uuid,
 ) -> impl futures::Stream<Item = io::Result<String>> {
     stream::poll_fn(move |cx| {
         std::pin::Pin::new(&mut lines)
             .poll_next_line(cx)
-            .map(|result| process_streaming_log_lines(result, stderr))
+            .map(|result| process_streaming_log_lines(result, stderr, &job_id))
     })
 }
 

backend/windmill-worker/src/job_logger_ee.rs

@@ -34,6 +34,7 @@ pub(crate) async fn default_disk_log_storage(
 pub(crate) fn process_streaming_log_lines(
     r: Result<Option<String>, io::Error>,
     _stderr: bool,
+    _job_id: &Uuid,
 ) -> Option<Result<String, io::Error>> {
     r.transpose()
 }