Add (currently unusuable) fallback handler initializer

Author: Sushisource

README.md

@@ -64,12 +64,12 @@ You can run it using:
 
 ## Debugging
 
-The crate uses [tracing](https://github.com/tokio-rs/tracing) to help with debugging. To enable
-it for a test, insert the below snippet at the start of the test. By default, tracing data is output
-to stdout in a (reasonably) pretty manner.
+The crate uses [tracing](https://github.com/tokio-rs/tracing) to help with debugging. To enable it
+globally for tests, insert the below snippet at the start of the test. By default, tracing data is
+output to stdout in a (reasonably) pretty manner.
 
 ```rust
-crate::telemetry::test_telem_console();
+crate::telemetry::telemetry_init_fallback();
 ```
 
 The passed in options to initialization can be customized to export to an OTel collector, etc.

core-api/src/telemetry.rs

@@ -3,12 +3,12 @@ pub mod metrics;
 use crate::telemetry::metrics::CoreMeter;
 use std::{
     collections::HashMap,
-    fmt::Debug,
+    fmt::{Debug, Formatter},
     net::SocketAddr,
     sync::Arc,
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
-use tracing_core::Level;
+use tracing_core::{Level, Subscriber};
 use url::Url;
 
 pub static METRIC_PREFIX: &str = "temporal_";
@@ -27,7 +27,7 @@ pub trait CoreTelemetry {
 }
 
 /// Telemetry configuration options. Construct with [TelemetryOptionsBuilder]
-#[derive(Debug, Clone, derive_builder::Builder)]
+#[derive(Clone, derive_builder::Builder)]
 #[non_exhaustive]
 pub struct TelemetryOptions {
     /// Optional logger - set as None to disable.
@@ -45,6 +45,39 @@ pub struct TelemetryOptions {
     /// A prefix to be applied to all core-created metrics. Defaults to "temporal_".
     #[builder(default = "METRIC_PREFIX.to_string()")]
     pub metric_prefix: String,
+    /// If provided, logging config will be ignored and this explicit subscriber will be used for
+    /// all logging and traces.
+    #[builder(setter(strip_option), default)]
+    pub subscriber_override: Option<Arc<dyn Subscriber + Send + Sync>>,
+}
+impl Debug for TelemetryOptions {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        #[derive(Debug)]
+        #[allow(dead_code)]
+        struct TelemetryOptions<'a> {
+            logging: &'a Option<Logger>,
+            metrics: &'a Option<Arc<dyn CoreMeter>>,
+            attach_service_name: &'a bool,
+            metric_prefix: &'a str,
+        }
+        let Self {
+            logging,
+            metrics,
+            attach_service_name,
+            metric_prefix,
+            ..
+        } = self;
+
+        Debug::fmt(
+            &TelemetryOptions {
+                logging,
+                metrics,
+                attach_service_name,
+                metric_prefix,
+            },
+            f,
+        )
+    }
 }
 
 /// Options for exporting to an OpenTelemetry Collector

core/Cargo.toml

@@ -44,7 +44,7 @@ lru = "0.13"
 mockall = "0.13"
 opentelemetry = { workspace = true, features = ["metrics"], optional = true }
 opentelemetry_sdk = { version = "0.29", features = ["rt-tokio", "metrics", "spec_unstable_metrics_views"], optional = true }
-opentelemetry-otlp = { version = "0.29", features = ["tokio", "metrics", "tls", "http-proto", "reqwest-client", "grpc-tonic" ], optional = true }
+opentelemetry-otlp = { version = "0.29", features = ["tokio", "metrics", "tls", "http-proto", "reqwest-client", "grpc-tonic"], optional = true }
 opentelemetry-prometheus = { version = "0.29", optional = true }
 parking_lot = { version = "0.12", features = ["send_guard"] }
 pid = "4.0"
@@ -115,6 +115,11 @@ name = "manual_tests"
 path = "../tests/manual_tests.rs"
 test = false
 
+[[test]]
+name = "global_metric_tests"
+path = "../tests/global_metric_tests.rs"
+test = false
+
 [[bench]]
 name = "workflow_replay"
 harness = false

core/src/telemetry/mod.rs

@@ -33,7 +33,7 @@ use std::{
     },
 };
 use temporal_sdk_core_api::telemetry::{
-    CoreLog, CoreTelemetry, Logger, TelemetryOptions,
+    CoreLog, CoreTelemetry, Logger, TelemetryOptions, TelemetryOptionsBuilder,
     metrics::{CoreMeter, MetricKeyValue, NewAttributes, TemporalMeter},
 };
 use tracing::{Level, Subscriber};
@@ -173,54 +173,59 @@ pub fn telemetry_init(opts: TelemetryOptions) -> Result<TelemetryInstance, anyho
     let mut forward_layer = None;
     // ===================================
 
-    let tracing_sub = opts.logging.map(|logger| {
-        match logger {
-            Logger::Console { filter } => {
-                // This is silly dupe but can't be avoided without boxing.
-                if env::var("TEMPORAL_CORE_PRETTY_LOGS").is_ok() {
-                    console_pretty_layer = Some(
-                        tracing_subscriber::fmt::layer()
-                            .with_target(false)
-                            .event_format(
-                                tracing_subscriber::fmt::format()
-                                    .pretty()
-                                    .with_source_location(false),
-                            )
-                            .with_filter(EnvFilter::new(filter)),
-                    )
-                } else {
-                    console_compact_layer = Some(
-                        tracing_subscriber::fmt::layer()
-                            .with_target(false)
-                            .event_format(
-                                tracing_subscriber::fmt::format()
-                                    .compact()
-                                    .with_source_location(false),
-                            )
-                            .with_filter(EnvFilter::new(filter)),
-                    )
+    let tracing_sub = if let Some(ts) = opts.subscriber_override {
+        Some(ts)
+    } else {
+        opts.logging.map(|logger| {
+            match logger {
+                Logger::Console { filter } => {
+                    // This is silly dupe but can't be avoided without boxing.
+                    if env::var("TEMPORAL_CORE_PRETTY_LOGS").is_ok() {
+                        console_pretty_layer = Some(
+                            tracing_subscriber::fmt::layer()
+                                .with_target(false)
+                                .event_format(
+                                    tracing_subscriber::fmt::format()
+                                        .pretty()
+                                        .with_source_location(false),
+                                )
+                                .with_filter(EnvFilter::new(filter)),
+                        )
+                    } else {
+                        console_compact_layer = Some(
+                            tracing_subscriber::fmt::layer()
+                                .with_target(false)
+                                .event_format(
+                                    tracing_subscriber::fmt::format()
+                                        .compact()
+                                        .with_source_location(false),
+                                )
+                                .with_filter(EnvFilter::new(filter)),
+                        )
+                    }
                 }
-            }
-            Logger::Forward { filter } => {
-                let (export_layer, lo) =
-                    CoreLogConsumerLayer::new_buffered(FORWARD_LOG_BUFFER_SIZE);
-                logs_out = Some(Mutex::new(lo));
-                forward_layer = Some(export_layer.with_filter(EnvFilter::new(filter)));
-            }
-            Logger::Push { filter, consumer } => {
-                forward_layer =
-                    Some(CoreLogConsumerLayer::new(consumer).with_filter(EnvFilter::new(filter)));
-            }
-        };
-        let reg = tracing_subscriber::registry()
-            .with(console_pretty_layer)
-            .with(console_compact_layer)
-            .with(forward_layer);
+                Logger::Forward { filter } => {
+                    let (export_layer, lo) =
+                        CoreLogConsumerLayer::new_buffered(FORWARD_LOG_BUFFER_SIZE);
+                    logs_out = Some(Mutex::new(lo));
+                    forward_layer = Some(export_layer.with_filter(EnvFilter::new(filter)));
+                }
+                Logger::Push { filter, consumer } => {
+                    forward_layer = Some(
+                        CoreLogConsumerLayer::new(consumer).with_filter(EnvFilter::new(filter)),
+                    );
+                }
+            };
+            let reg = tracing_subscriber::registry()
+                .with(console_pretty_layer)
+                .with(console_compact_layer)
+                .with(forward_layer);
 
-        #[cfg(feature = "tokio-console")]
-        let reg = reg.with(console_subscriber::spawn());
-        Arc::new(reg) as Arc<dyn Subscriber + Send + Sync>
-    });
+            #[cfg(feature = "tokio-console")]
+            let reg = reg.with(console_subscriber::spawn());
+            Arc::new(reg) as Arc<dyn Subscriber + Send + Sync>
+        })
+    };
 
     Ok(TelemetryInstance::new(
         tracing_sub,
@@ -231,6 +236,9 @@ pub fn telemetry_init(opts: TelemetryOptions) -> Result<TelemetryInstance, anyho
     ))
 }
 
+/// WARNING: Calling can cause panics because of https://github.com/tokio-rs/tracing/issues/1656
+/// Lang must not start using until resolved
+///
 /// Initialize telemetry/tracing globally. Useful for testing. Only takes affect when called
 /// the first time. Subsequent calls are ignored.
 pub fn telemetry_init_global(opts: TelemetryOptions) -> Result<(), anyhow::Error> {
@@ -247,8 +255,23 @@ pub fn telemetry_init_global(opts: TelemetryOptions) -> Result<(), anyhow::Error
     Ok(())
 }
 
-#[cfg(test)]
-pub use test_initters::*;
+/// WARNING: Calling can cause panics because of https://github.com/tokio-rs/tracing/issues/1656
+/// Lang must not start using until resolved
+///
+/// Initialize the fallback global handler. All lang SDKs should call this somewhere, once, at
+/// startup, as it initializes a fallback handler for any dependencies (looking at you, otel) that
+/// don't provide good ways to customize their tracing usage. It sets a WARN-level global filter
+/// that uses the default console logger.
+pub fn telemetry_init_fallback() -> Result<(), anyhow::Error> {
+    telemetry_init_global(
+        TelemetryOptionsBuilder::default()
+            .logging(Logger::Console {
+                filter: construct_filter_string(Level::DEBUG, Level::WARN),
+            })
+            .build()?,
+    )?;
+    Ok(())
+}
 
 /// A trait for using [Display] on the contents of vecs, etc, which don't implement it.
 ///
@@ -275,24 +298,3 @@ where
         format!("[{}]", self.iter().format(","))
     }
 }
-
-/// Helpers for test initialization
-#[cfg(test)]
-pub mod test_initters {
-    use super::*;
-    use temporal_sdk_core_api::telemetry::TelemetryOptionsBuilder;
-
-    /// Turn on logging to the console
-    #[allow(dead_code)] // Not always used, called to enable for debugging when needed
-    pub fn test_telem_console() {
-        telemetry_init_global(
-            TelemetryOptionsBuilder::default()
-                .logging(Logger::Console {
-                    filter: construct_filter_string(Level::DEBUG, Level::WARN),
-                })
-                .build()
-                .unwrap(),
-        )
-        .unwrap();
-    }
-}

tests/global_metric_tests.rs

@@ -0,0 +1,116 @@
+use parking_lot::Mutex;
+use std::{sync::Arc, time::Duration};
+use temporal_sdk_core::{
+    CoreRuntime,
+    telemetry::{build_otlp_metric_exporter, construct_filter_string, telemetry_init_global},
+};
+use temporal_sdk_core_api::telemetry::{
+    Logger, OtelCollectorOptionsBuilder, TelemetryOptionsBuilder, metrics::CoreMeter,
+};
+use temporal_sdk_core_test_utils::CoreWfStarter;
+use tracing::Level;
+use tracing_subscriber::fmt::MakeWriter;
+
+struct CapturingWriter {
+    buf: Arc<Mutex<Vec<u8>>>,
+}
+
+impl MakeWriter<'_> for CapturingWriter {
+    type Writer = CapturingHandle;
+
+    fn make_writer(&self) -> Self::Writer {
+        CapturingHandle(self.buf.clone())
+    }
+}
+
+struct CapturingHandle(Arc<Mutex<Vec<u8>>>);
+
+impl std::io::Write for CapturingHandle {
+    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+        let mut b = self.0.lock();
+        b.extend_from_slice(buf);
+        Ok(buf.len())
+    }
+    fn flush(&mut self) -> std::io::Result<()> {
+        Ok(())
+    }
+}
+
+// TODO: This test is not actually run in CI right now because for it to actually work requires
+//   a number of fixes in upstream libraries:
+//
+// * It regularly panics because of: https://github.com/tokio-rs/tracing/issues/1656
+// * Otel doesn't appear to actually be logging any warnings/errors on connection failure
+// * This whole thing is supposed to show a workaround for https://github.com/open-telemetry/opentelemetry-rust/issues/2697
+#[tokio::test]
+async fn otel_errors_logged_as_errors() {
+    // Set up tracing subscriber to capture ERROR logs
+    let logs = Arc::new(Mutex::new(Vec::new()));
+    let writer = CapturingWriter { buf: logs.clone() };
+    let subscriber = Arc::new(
+        tracing_subscriber::fmt()
+            .with_writer(writer)
+            .with_env_filter("debug")
+            .finish(),
+    );
+    let opts = OtelCollectorOptionsBuilder::default()
+        .url("https://localhost:12345/v1/metrics".parse().unwrap()) // Nothing bound on that port
+        .build()
+        .unwrap();
+    let exporter = build_otlp_metric_exporter(opts).unwrap();
+
+    // Global initialization is needed to capture (some) otel logging.
+    telemetry_init_global(
+        TelemetryOptionsBuilder::default()
+            .subscriber_override(subscriber)
+            .build()
+            .unwrap(),
+    )
+    .unwrap();
+
+    let rt = CoreRuntime::new_assume_tokio(
+        TelemetryOptionsBuilder::default()
+            .metrics(Arc::new(exporter) as Arc<dyn CoreMeter>)
+            // Importantly, _not_ using subscriber override, is using console.
+            .logging(Logger::Console {
+                filter: construct_filter_string(Level::INFO, Level::WARN),
+            })
+            .build()
+            .unwrap(),
+    )
+    .unwrap();
+    let mut starter = CoreWfStarter::new_with_runtime("otel_errors_logged_as_errors", rt);
+    let _worker = starter.get_worker().await;
+
+    tracing::debug!("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@ should be in global log");
+
+    // Wait to allow exporter to attempt sending metrics and fail.
+    // Windows takes a while to fail the network attempt for some reason so 5s.
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    let logs = logs.lock();
+    let log_str = String::from_utf8_lossy(&logs).into_owned();
+    drop(logs);
+
+    // The core worker _isn't_ using the fallback, and shouldn't be captured
+    assert!(
+        !log_str.contains("Initializing worker"),
+        "Core logging shouldn't have been caught by fallback"
+    );
+    assert!(
+        log_str.contains("@@@@@@@@@"),
+        "Expected fallback log not found in logs: {}",
+        log_str
+    );
+    // TODO: OTel just doesn't actually log useful errors right now 🤷, see issues at top of test
+    assert!(
+        log_str.contains("ERROR"),
+        "Expected ERROR log not found in logs: {}",
+        log_str
+    );
+    assert!(
+        log_str.contains("Metrics exporter otlp failed with the grpc server returns error"),
+        "Expected an OTel exporter error message in logs: {}",
+        log_str
+    );
+}