Add OTLP tracing support for daemon and coordinator

Enables Jaeger tracing when DORA_OTLP_ENDPOINT is set. Key changes:
- Add service name to OTLP resource configuration
- Register global tracer provider for span export
- Store OtelGuard to prevent premature shutdown
- Initialize OTLP within tokio runtime context
- Move TracingBuilder import to top-level with proper feature gate

Fixes #1313

Author: Nakshatra Sharma

binaries/cli/src/command/coordinator.rs

@@ -33,30 +33,66 @@ pub struct Coordinator {
 
 impl Executable for Coordinator {
     fn execute(self) -> eyre::Result<()> {
+        let enable_otlp = std::env::var("DORA_OTLP_ENDPOINT").is_ok()
+            || std::env::var("DORA_JAEGER_TRACING").is_ok();
+
         #[cfg(feature = "tracing")]
         {
-            let name = "dora-coordinator";
-            let mut builder = TracingBuilder::new(name);
-            if !self.quiet {
-                builder = builder.with_stdout("info", false);
+            // Only initialize basic tracing if OTLP is NOT requested
+            // If OTLP is requested, we'll initialize everything inside the runtime
+            if !enable_otlp {
+                let name = "dora-coordinator";
+                let mut builder = TracingBuilder::new(name);
+                if !self.quiet {
+                    builder = builder.with_stdout("info", false);
+                }
+                builder = builder.with_file(name, LevelFilter::INFO)?;
+                builder
+                    .build()
+                    .wrap_err("failed to set up tracing subscriber")?;
             }
-            builder = builder.with_file(name, LevelFilter::INFO)?;
-            builder
-                .build()
-                .wrap_err("failed to set up tracing subscriber")?;
         }
 
         let rt = Builder::new_multi_thread()
             .enable_all()
             .build()
             .context("tokio runtime failed")?;
-        rt.block_on(async {
+
+        let quiet = self.quiet;
+
+        rt.block_on(async move {
+            #[cfg(feature = "tracing")]
+            let _otel_guard = if enable_otlp {
+                let name = "dora-coordinator";
+                let mut builder = TracingBuilder::new(name);
+
+                builder = builder
+                    .with_otlp_tracing()
+                    .wrap_err("failed to set up OTLP tracing")?;
+
+                if !quiet {
+                    builder = builder.with_stdout("info", false);
+                }
+                builder = builder.with_file(name, LevelFilter::INFO)?;
+
+                let guard = builder.guard.take();
+
+                builder
+                    .build()
+                    .wrap_err("failed to set up tracing subscriber with OTLP")?;
+
+                tracing::info!("OTLP tracing enabled for coordinator");
+                guard
+            } else {
+                None
+            };
+
             let bind = SocketAddr::new(self.interface, self.port);
             let bind_control = SocketAddr::new(self.control_interface, self.control_port);
             let (port, task) =
                 dora_coordinator::start(bind, bind_control, futures::stream::empty::<Event>())
                     .await?;
-            if !self.quiet {
+            if !quiet {
                 println!("Listening for incoming daemon connection on {port}");
             }
             task.await

binaries/cli/src/command/daemon.rs

@@ -13,6 +13,9 @@ use std::{
 use tokio::runtime::Builder;
 use tracing::level_filters::LevelFilter;
 
+#[cfg(feature = "tracing")]
+use dora_tracing::TracingBuilder;
+
 #[derive(Debug, clap::Args)]
 /// Run daemon
 pub struct Daemon {
@@ -37,13 +40,16 @@ pub struct Daemon {
 
 impl Executable for Daemon {
     fn execute(self) -> eyre::Result<()> {
+        let enable_otlp = std::env::var("DORA_OTLP_ENDPOINT").is_ok()
+            || std::env::var("DORA_JAEGER_TRACING").is_ok();
+
         let rt = Builder::new_multi_thread()
             .enable_all()
             .build()
             .context("tokio runtime failed")?;
 
         #[cfg(feature = "tracing")]
-        let _guard = {
+        let _guard = if !enable_otlp {
             let _enter = rt.enter();
 
             let name = "dora-daemon";
@@ -67,8 +73,46 @@ impl Executable for Daemon {
                 LevelFilter::INFO,
             )
             .context("failed to initialize tracing")?
+        } else {
+            None
         };
-        rt.block_on(async {
+
+        let machine_id = self.machine_id.clone();
+        let quiet = self.quiet;
+
+        rt.block_on(async move {
+                #[cfg(feature = "tracing")]
+                let _otel_guard = if enable_otlp {
+
+                    let name = "dora-daemon";
+                    let filename = machine_id
+                        .as_ref()
+                        .map(|id| format!("{name}-{id}"))
+                        .unwrap_or(name.to_string());
+                    let mut builder = TracingBuilder::new(name);
+
+                    builder = builder
+                        .with_otlp_tracing()
+                        .wrap_err("failed to set up OTLP tracing")?;
+
+                    if !quiet {
+                        builder = builder.with_stdout("info,zenoh=warn", false);
+                    }
+                    builder = builder.with_file(filename.clone(), LevelFilter::INFO)?;
+
+
+                    let guard = builder.guard.take();
+
+                    builder
+                        .build()
+                        .wrap_err("failed to set up tracing subscriber with OTLP")?;
+
+                    tracing::info!("OTLP tracing enabled for daemon");
+                    guard
+                } else {
+                    None
+};
+
                 match self.run_dataflow {
                     Some(dataflow_path) => {
                         tracing::info!("Starting dataflow `{}`", dataflow_path.display());
@@ -88,7 +132,7 @@ impl Executable for Daemon {
                         handle_dataflow_result(result, None)
                     }
                     None => {
-                        dora_daemon::Daemon::run(SocketAddr::new(self.coordinator_addr, self.coordinator_port), self.machine_id, self.local_listen_port).await
+                        dora_daemon::Daemon::run(SocketAddr::new(self.coordinator_addr, self.coordinator_port), machine_id, self.local_listen_port).await
                     }
                 }
             })

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

@@ -131,13 +131,14 @@ impl TracingBuilder {
             .or_else(|_| std::env::var("DORA_JAEGER_TRACING"))
             .wrap_err("DORA_OTLP_ENDPOINT or DORA_JAEGER_TRACING environment variable not set")?;
 
-        // Initialize OTLP tracing - this returns a tracer and sets the global provider
         let sdk_tracer_provider = crate::telemetry::init_tracing(&self.name, &endpoint);
+
+        opentelemetry::global::set_tracer_provider(sdk_tracer_provider.clone());
+
         let meter_provider = metrics::init_meter_provider();
 
         // TODO: Maybe this needs to be removed in favor of application level global.
         // global::set_meter_provider(meter_provider.clone());
-        // Use the specific tracer instance returned from init_tracing
         let tracer = sdk_tracer_provider.tracer("tracing-otel-subscriber");
 
         let guard = OtelGuard {

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

@@ -39,15 +39,19 @@ 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)
         .build()
         .unwrap();
 
+    let resource = opentelemetry_sdk::Resource::builder()
+        .with_service_name(name.to_string())
+        .build();
+
     SdkTracerProvider::builder()
-        // Customize sampling strategy
+        .with_resource(resource)
         .with_batch_exporter(exporter)
         .build()
 }