Add opentelemetry to the daemon (#1303)

This PR adds Opentelemetry to the daemon and make it possible to receive
node failure information within opentelemetry even if the node panic and
don't have time itself to send the error logs.

Author: haixuanTao

binaries/cli/src/command/daemon.rs

@@ -5,9 +5,6 @@ use dora_core::topics::{
 };
 
 use dora_daemon::LogDestination;
-#[cfg(feature = "tracing")]
-use dora_tracing::TracingBuilder;
-
 use eyre::Context;
 use std::{
     net::{IpAddr, SocketAddr},
@@ -40,28 +37,37 @@ pub struct Daemon {
 
 impl Executable for Daemon {
     fn execute(self) -> eyre::Result<()> {
+        let rt = Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .context("tokio runtime failed")?;
+
         #[cfg(feature = "tracing")]
-        {
+        let _guard = {
+            let _enter = rt.enter();
+
             let name = "dora-daemon";
             let filename = self
                 .machine_id
                 .as_ref()
                 .map(|id| format!("{name}-{id}"))
                 .unwrap_or(name.to_string());
-            let mut builder = TracingBuilder::new(name);
-            if !self.quiet {
-                builder = builder.with_stdout("info,zenoh=warn", false);
-            }
-            builder = builder.with_file(filename, LevelFilter::INFO)?;
-            builder
-                .build()
-                .wrap_err("failed to set up tracing subscriber")?;
-        }
+            let quiet = self.quiet;
 
-        let rt = Builder::new_multi_thread()
-            .enable_all()
-            .build()
-            .context("tokio runtime failed")?;
+            let stdout_filter = if !quiet {
+                Some(std::env::var("RUST_LOG").unwrap_or("info".to_string()))
+            } else {
+                None
+            };
+
+            dora_tracing::init_tracing_subscriber(
+                name,
+                stdout_filter.as_deref(),
+                Some(&filename),
+                LevelFilter::INFO,
+            )
+            .context("failed to initialize tracing")?
+        };
         rt.block_on(async {
                 match self.run_dataflow {
                     Some(dataflow_path) => {

binaries/cli/src/command/run.rs

@@ -12,7 +12,6 @@ use crate::{
     session::DataflowSession,
 };
 use dora_daemon::{Daemon, LogDestination, flume};
-use dora_tracing::TracingBuilder;
 use eyre::Context;
 use tokio::runtime::Builder;
 
@@ -36,22 +35,27 @@ pub fn run_func(dataflow: String, uv: bool) -> eyre::Result<()> {
 }
 
 pub fn run(dataflow: String, uv: bool) -> eyre::Result<()> {
+    let rt = Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .context("tokio runtime failed")?;
+
     #[cfg(feature = "tracing")]
-    {
-        let log_level = std::env::var("RUST_LOG").ok().unwrap_or("info".to_string());
-        TracingBuilder::new("run")
-            .with_stdout(log_level, false)
-            .build()
-            .wrap_err("failed to set up tracing subscriber")?;
-    }
+    let _guard = {
+        let _enter = rt.enter();
+        let env_log = std::env::var("RUST_LOG").unwrap_or("info".to_string());
+        dora_tracing::init_tracing_subscriber(
+            "dora-run",
+            Some(&env_log),
+            None,
+            tracing::metadata::LevelFilter::INFO,
+        )
+        .context("failed to initialize tracing")?
+    };
 
     let dataflow_path = resolve_dataflow(dataflow).context("could not resolve dataflow")?;
     let dataflow_session =
         DataflowSession::read_session(&dataflow_path).context("failed to read DataflowSession")?;
-    let rt = Builder::new_multi_thread()
-        .enable_all()
-        .build()
-        .context("tokio runtime failed")?;
 
     let (log_tx, log_rx) = flume::bounded(100);
     std::thread::spawn(move || {

binaries/daemon/src/spawn/prepared.rs

@@ -138,6 +138,16 @@ impl PreparedNode {
             let restart = restart && !restart_disabled;
             let success = exit_status.is_success();
 
+            if !success {
+                let _span = tracing::error_span!(
+                    "node_failure",
+                    node_id = %self.node.id,
+                    dataflow_id = %self.dataflow_id
+                )
+                .entered();
+                tracing::error!("node exited with error: {:?}", exit_status);
+            }
+
             let event = DoraEvent::SpawnedNodeResult {
                 dataflow_id: self.dataflow_id,
                 node_id: self.node.id.clone(),

libraries/extensions/telemetry/tracing/src/lib.rs

@@ -63,12 +63,24 @@ impl TracingBuilder {
     /// it uses [std::io::stdout] which is synchronous
     /// and might block the logging thread.
     pub fn with_stdout(mut self, filter: impl AsRef<str>, json: bool) -> Self {
-        let parsed = EnvFilter::builder()
+        let mut parsed = EnvFilter::builder()
             .parse_lossy(filter)
             .add_directive("hyper=off".parse().unwrap())
             .add_directive("tonic=off".parse().unwrap())
+            .add_directive("tokio=off".parse().unwrap())
+            .add_directive("process_wrap=off".parse().unwrap())
             .add_directive("h2=off".parse().unwrap())
             .add_directive("reqwest=off".parse().unwrap());
+        let env_log = std::env::var("RUST_LOG").unwrap_or_default();
+        if !env_log.contains("dora_daemon") {
+            parsed = parsed.add_directive("dora_daemon=info".parse().unwrap());
+        }
+        if !env_log.contains("dora_core") {
+            parsed = parsed.add_directive("dora_core=warn".parse().unwrap());
+        }
+        if !env_log.contains("zenoh") {
+            parsed = parsed.add_directive("zenoh=warn".parse().unwrap());
+        }
         let env_filter = EnvFilter::from_default_env().or(parsed);
         let layer = tracing_subscriber::fmt::layer()
             .compact()
@@ -135,11 +147,17 @@ impl TracingBuilder {
 
         self.guard = Some(guard);
         self.layers.push(MetricsLayer::new(meter_provider).boxed());
-        let filter_otel = EnvFilter::new("trace")
+        let mut filter_otel = EnvFilter::new("trace")
             .add_directive("hyper=off".parse().unwrap())
             .add_directive("tonic=off".parse().unwrap())
+            .add_directive("tokio=off".parse().unwrap())
+            .add_directive("process_wrap=off".parse().unwrap())
             .add_directive("h2=off".parse().unwrap())
             .add_directive("reqwest=off".parse().unwrap());
+        let env_log = std::env::var("RUST_LOG").unwrap_or_default();
+        if !env_log.contains("dora_daemon") {
+            filter_otel = filter_otel.add_directive("dora_daemon=debug".parse().unwrap());
+        }
         self.layers.push(
             OpenTelemetryLayer::new(tracer)
                 .with_filter(filter_otel)
@@ -192,3 +210,63 @@ impl Drop for OtelGuard {
         self.tracer_provider.shutdown().ok();
     }
 }
+
+/// Initialize tracing with OTLP (if configured) or stdout/file logging.
+///
+/// This function should be called after creating a tokio runtime and calling `runtime.enter()`.
+///
+/// # Parameters
+/// - `name`: Service name for tracing
+/// - `stdout_filter`: Optional RUST_LOG-style filter for stdout logging (e.g., "info", "debug")
+/// - `file_name`: Optional filename for file logging (will be placed in `out/` directory)
+/// - `file_filter`: Level filter for file logging (only used if `file_name` is Some)
+///
+/// # Returns
+/// Returns `Option<OtelGuard>` which must be kept alive for the duration of the program.
+/// When dropped, the guard will flush and shutdown telemetry providers.
+///
+/// # Example
+/// ```no_run
+/// use dora_tracing::init_tracing_subscriber;
+/// use tracing::level_filters::LevelFilter;
+///
+/// // Note: This function requires a tokio runtime context to be active
+/// // when using OTLP tracing. Use runtime.enter() before calling.
+/// let _guard = init_tracing_subscriber(
+///     "my-service",
+///     Some("info"),
+///     Some("my-service"),
+///     LevelFilter::INFO,
+/// ).unwrap();
+/// ```
+pub fn init_tracing_subscriber(
+    name: &str,
+    stdout_filter: Option<&str>,
+    file_name: Option<&str>,
+    file_filter: LevelFilter,
+) -> eyre::Result<Option<OtelGuard>> {
+    let mut builder = TracingBuilder::new(name);
+    let guard: Option<OtelGuard>;
+
+    if std::env::var("DORA_OTLP_ENDPOINT").is_ok() || std::env::var("DORA_JAEGER_TRACING").is_ok() {
+        builder = builder
+            .with_otlp_tracing()
+            .wrap_err("failed to set up OTLP tracing")?;
+        guard = builder.guard.take();
+    } else {
+        if let Some(filter) = stdout_filter {
+            builder = builder.with_stdout(filter, false);
+        }
+        guard = None;
+    }
+
+    if let Some(filename) = file_name {
+        builder = builder.with_file(filename, file_filter)?;
+    }
+
+    builder
+        .build()
+        .wrap_err("failed to set up tracing subscriber")?;
+
+    Ok(guard)
+}

libraries/extensions/telemetry/tracing/src/telemetry.rs

@@ -39,7 +39,7 @@ impl Extractor for MetadataMap<'_> {
 /// docker run -d -p 4317:4317 -p 4318:4318 -p 16686:16686 jaegertracing/all-in-one:latest
 /// ```
 ///
-pub fn init_tracing(name: &str, endpoint: &str) -> sdktrace::SdkTracerProvider {
+pub fn init_tracing(_name: &str, endpoint: &str) -> sdktrace::SdkTracerProvider {
     let exporter = opentelemetry_otlp::SpanExporter::builder()
         .with_tonic()
         .with_endpoint(endpoint)

libraries/message/src/common.rs

@@ -4,7 +4,7 @@ use std::{borrow::Cow, collections::BTreeMap};
 use aligned_vec::{AVec, ConstAlign};
 use chrono::{DateTime, Utc};
 use eyre::Context as _;
-use serde::{Deserialize, Deserializer};
+use serde::Deserialize;
 use uuid::Uuid;
 
 use crate::{BuildId, DataflowId, daemon_to_daemon::InterDaemonEvent, id::NodeId};