temporalio_common/
telemetry.rs

1//! Contains tracing/logging and metrics related functionality
2
3/// Metric instrument types and the [`CoreMeter`] trait.
4pub mod metrics;
5
6#[cfg(feature = "core-telemetry-bridge")]
7mod log_export;
8#[cfg(feature = "otel")]
9mod otel;
10#[cfg(feature = "prometheus")]
11mod prometheus_meter;
12#[cfg(feature = "prometheus")]
13mod prometheus_server;
14
15use crate::telemetry::metrics::{
16    CoreMeter, MetricKeyValue, NewAttributes, PrefixedMetricsMeter, TemporalMeter,
17};
18use std::{
19    cell::RefCell,
20    collections::HashMap,
21    env,
22    fmt::{Debug, Formatter},
23    net::SocketAddr,
24    sync::{
25        Arc,
26        atomic::{AtomicBool, Ordering},
27    },
28    time::{Duration, SystemTime, UNIX_EPOCH},
29};
30use tracing::{Level, Subscriber};
31use tracing_subscriber::{EnvFilter, Layer, fmt::MakeWriter, layer::SubscriberExt};
32use url::Url;
33
34#[cfg(feature = "core-telemetry-bridge")]
35use crate::telemetry::log_export::CoreLogConsumerLayer;
36
37#[cfg(feature = "core-telemetry-bridge")]
38pub use log_export::{CoreLogBuffer, CoreLogBufferedConsumer, CoreLogStreamConsumer};
39#[cfg(feature = "otel")]
40pub use otel::build_otlp_metric_exporter;
41#[cfg(feature = "prometheus")]
42pub use prometheus_server::start_prometheus_metric_exporter;
43
44/// The default prefix applied to all Temporal metric names.
45pub static METRIC_PREFIX: &str = "temporal_";
46
47const TELEM_SERVICE_NAME: &str = "temporal-core-sdk";
48
49/// Each core runtime instance has a telemetry subsystem associated with it, this trait defines the
50/// operations that lang might want to perform on that telemetry after it's initialized.
51pub trait CoreTelemetry {
52    /// Each worker buffers logs that should be shuttled over to lang so that they may be rendered
53    /// with the user's desired logging library. Use this function to grab the most recent buffered
54    /// logs since the last time it was called. A fixed number of such logs are retained at maximum,
55    /// with the oldest being dropped when full.
56    ///
57    /// Returns the list of logs from oldest to newest. Returns an empty vec if the feature is not
58    /// configured.
59    fn fetch_buffered_logs(&self) -> Vec<CoreLog>;
60}
61
62/// Telemetry configuration options. Construct with [TelemetryOptions::builder]
63#[derive(Clone, bon::Builder)]
64#[non_exhaustive]
65pub struct TelemetryOptions {
66    /// Optional logger - set as None to disable.
67    #[builder(into)]
68    pub logging: Option<Logger>,
69    /// Optional metrics exporter - set as None to disable.
70    #[builder(into)]
71    pub metrics: Option<Arc<dyn CoreMeter>>,
72    /// If set true (the default) explicitly attach a `service_name` label to all metrics. Turn this
73    /// off if your collection system supports the `target_info` metric from the OpenMetrics spec.
74    /// For more, see
75    /// [here](https://github.com/OpenObservability/OpenMetrics/blob/main/specification/OpenMetrics.md#supporting-target-metadata-in-both-push-based-and-pull-based-systems)
76    #[builder(default = true)]
77    pub attach_service_name: bool,
78    /// A prefix to be applied to all core-created metrics. Defaults to "temporal_".
79    #[builder(default = METRIC_PREFIX.to_string())]
80    pub metric_prefix: String,
81    /// If provided, logging config will be ignored and this explicit subscriber will be used for
82    /// all logging and traces.
83    pub subscriber_override: Option<Arc<dyn Subscriber + Send + Sync>>,
84    /// See [TaskQueueLabelStrategy].
85    #[builder(default = TaskQueueLabelStrategy::UseNormal)]
86    pub task_queue_label_strategy: TaskQueueLabelStrategy,
87}
88impl Debug for TelemetryOptions {
89    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
90        #[derive(Debug)]
91        #[allow(dead_code)]
92        struct TelemetryOptions<'a> {
93            logging: &'a Option<Logger>,
94            metrics: &'a Option<Arc<dyn CoreMeter>>,
95            attach_service_name: &'a bool,
96            metric_prefix: &'a str,
97        }
98        let Self {
99            logging,
100            metrics,
101            attach_service_name,
102            metric_prefix,
103            ..
104        } = self;
105
106        Debug::fmt(
107            &TelemetryOptions {
108                logging,
109                metrics,
110                attach_service_name,
111                metric_prefix,
112            },
113            f,
114        )
115    }
116}
117
118/// Determines how the `task_queue` label value is set on metrics.
119#[derive(Copy, Clone, Debug)]
120#[non_exhaustive]
121pub enum TaskQueueLabelStrategy {
122    /// Always use the normal task queue name, including for actions relating to sticky queues.
123    UseNormal,
124    /// Use the sticky queue name when recording metrics operating on sticky queues.
125    UseNormalAndSticky,
126}
127
128/// Options for exporting to an OpenTelemetry Collector
129#[derive(Debug, Clone, bon::Builder)]
130#[non_exhaustive]
131pub struct OtelCollectorOptions {
132    /// The url of the OTel collector to export telemetry and metrics to. Lang SDK should also
133    /// export to this same collector.
134    pub url: Url,
135    /// Optional set of HTTP headers to send to the Collector, e.g for authentication.
136    #[builder(default = HashMap::new())]
137    pub headers: HashMap<String, String>,
138    /// Optionally specify how frequently metrics should be exported. Defaults to 1 second.
139    #[builder(default = Duration::from_secs(1))]
140    pub metric_periodicity: Duration,
141    /// Specifies the aggregation temporality for metric export. Defaults to cumulative.
142    #[builder(default = MetricTemporality::Cumulative)]
143    pub metric_temporality: MetricTemporality,
144    /// A map of tags to be applied to all metrics
145    #[builder(default)]
146    pub global_tags: HashMap<String, String>,
147    /// If set to true, use f64 seconds for durations instead of u64 milliseconds
148    #[builder(default)]
149    pub use_seconds_for_durations: bool,
150    /// Overrides for histogram buckets. Units depend on the value of `use_seconds_for_durations`.
151    #[builder(default)]
152    pub histogram_bucket_overrides: HistogramBucketOverrides,
153    /// Protocol to use for communication with the collector
154    #[builder(default = OtlpProtocol::Grpc)]
155    pub protocol: OtlpProtocol,
156}
157
158/// Options for exporting metrics to Prometheus
159#[derive(Debug, Clone, bon::Builder)]
160#[non_exhaustive]
161pub struct PrometheusExporterOptions {
162    /// The address the Prometheus exporter HTTP server will bind to.
163    pub socket_addr: SocketAddr,
164    /// A map of tags to be applied to all metrics
165    #[builder(default)]
166    pub global_tags: HashMap<String, String>,
167    /// If set true, all counters will include a "_total" suffix
168    #[builder(default)]
169    pub counters_total_suffix: bool,
170    /// If set true, all histograms will include the unit in their name as a suffix.
171    /// Ex: "_milliseconds".
172    #[builder(default)]
173    pub unit_suffix: bool,
174    /// If set to true, use f64 seconds for durations instead of u64 milliseconds
175    #[builder(default)]
176    pub use_seconds_for_durations: bool,
177    /// Overrides for histogram buckets. Units depend on the value of `use_seconds_for_durations`.
178    #[builder(default)]
179    pub histogram_bucket_overrides: HistogramBucketOverrides,
180}
181
182/// Allows overriding the buckets used by histogram metrics
183#[derive(Debug, Clone, Default)]
184pub struct HistogramBucketOverrides {
185    /// Overrides where the key is the metric name and the value is the list of bucket boundaries.
186    /// The metric name will apply regardless of name prefixing, if any. IE: the name acts like
187    /// `*metric_name`.
188    ///
189    /// The string names of core's built-in histogram metrics are publicly available on the
190    /// `core::telemetry` module and the `client` crate.
191    ///
192    /// See [here](https://docs.rs/opentelemetry_sdk/latest/opentelemetry_sdk/metrics/enum.Aggregation.html#variant.ExplicitBucketHistogram.field.boundaries)
193    /// for the exact meaning of boundaries.
194    pub overrides: HashMap<String, Vec<f64>>,
195}
196
197/// Control where logs go
198#[derive(Debug, Clone)]
199pub enum Logger {
200    /// Log directly to console.
201    Console {
202        /// An [EnvFilter](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/struct.EnvFilter.html) filter string.
203        filter: String,
204        /// The format used to render logs. When unset, `TEMPORAL_CORE_PRETTY_LOGS` selects pretty
205        /// output if present; otherwise logs use the compact format.
206        format: Option<LoggerFormat>,
207    },
208    #[cfg(feature = "core-telemetry-bridge")]
209    /// Forward logs to Lang - collectable with `fetch_buffered_logs`.
210    Forward {
211        /// An [EnvFilter](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/struct.EnvFilter.html) filter string.
212        filter: String,
213    },
214    #[cfg(feature = "core-telemetry-bridge")]
215    /// Push logs to Lang. Can be used with
216    /// temporalio_sdk_core::telemetry::log_export::CoreLogBufferedConsumer to buffer.
217    Push {
218        /// An [EnvFilter](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/struct.EnvFilter.html) filter string.
219        filter: String,
220        /// Trait invoked on each log.
221        consumer: Arc<dyn CoreLogConsumer>,
222    },
223}
224
225/// Controls how console logs are rendered.
226#[derive(Debug, Default, Clone, Copy)]
227#[non_exhaustive]
228pub enum LoggerFormat {
229    /// Render logs compactly on a single line.
230    #[default]
231    Compact,
232    /// Render logs in a human-readable format across multiple lines.
233    Pretty,
234    /// Render each log as a JSON object on a single line.
235    Json,
236}
237
238/// Types of aggregation temporality for metric export.
239/// See: <https://github.com/open-telemetry/opentelemetry-specification/blob/ce50e4634efcba8da445cc23523243cb893905cb/specification/metrics/datamodel.md#temporality>
240#[derive(Debug, Clone, Copy)]
241pub enum MetricTemporality {
242    /// Successive data points repeat the starting timestamp
243    Cumulative,
244    /// Successive data points advance the starting timestamp
245    Delta,
246}
247
248/// Options for configuring telemetry
249#[derive(Debug, Clone, Copy)]
250pub enum OtlpProtocol {
251    /// Use gRPC to communicate with the collector
252    Grpc,
253    /// Use HTTP to communicate with the collector
254    Http,
255}
256
257impl Default for TelemetryOptions {
258    fn default() -> Self {
259        TelemetryOptions::builder().build()
260    }
261}
262
263/// A log line (which ultimately came from a tracing event) exported from Core->Lang
264#[derive(Debug)]
265pub struct CoreLog {
266    /// The module within core this message originated from
267    pub target: String,
268    /// Log message
269    pub message: String,
270    /// Time log was generated (not when it was exported to lang)
271    pub timestamp: SystemTime,
272    /// Message level
273    pub level: Level,
274    /// Arbitrary k/v pairs (span k/vs are collapsed with event k/vs here). We could change this
275    /// to include them in `span_contexts` instead, but there's probably not much value for log
276    /// forwarding.
277    pub fields: HashMap<String, serde_json::Value>,
278    /// A list of the outermost to the innermost span names
279    pub span_contexts: Vec<String>,
280}
281
282impl CoreLog {
283    /// Return timestamp as ms since epoch
284    pub fn millis_since_epoch(&self) -> u128 {
285        self.timestamp
286            .duration_since(UNIX_EPOCH)
287            .unwrap_or(Duration::ZERO)
288            .as_millis()
289    }
290}
291
292/// Consumer trait for use with push logger.
293pub trait CoreLogConsumer: Send + Sync + Debug {
294    /// Invoked synchronously for every single log.
295    fn on_log(&self, log: CoreLog);
296}
297
298#[cfg(feature = "core-telemetry-bridge")]
299const FORWARD_LOG_BUFFER_SIZE: usize = 2048;
300
301/// Help you construct an [EnvFilter] compatible filter string which will forward all core module
302/// traces at `core_level` and all others (from 3rd party modules, etc) at `other_level`.
303pub fn construct_filter_string(core_level: Level, other_level: Level) -> String {
304    format!(
305        "{other_level},temporalio_common={core_level},temporalio_sdk_core={core_level},temporalio_client={core_level},temporalio_sdk={core_level}"
306    )
307}
308
309/// Holds initialized tracing/metrics exporters, etc
310pub struct TelemetryInstance {
311    metric_prefix: String,
312    #[cfg(feature = "core-telemetry-bridge")]
313    logs_out: Option<parking_lot::Mutex<CoreLogBuffer>>,
314    metrics: Option<Arc<dyn CoreMeter + 'static>>,
315    /// The tracing subscriber which is associated with this telemetry instance. May be `None` if
316    /// the user has not opted into any tracing configuration.
317    trace_subscriber: Option<Arc<dyn Subscriber + Send + Sync>>,
318    attach_service_name: bool,
319    task_queue_label_strategy: TaskQueueLabelStrategy,
320}
321
322impl TelemetryInstance {
323    /// Return the trace subscriber associated with the telemetry options/instance. Can be used
324    /// to manually set the default for a thread or globally using the `tracing` crate, or with
325    /// [set_trace_subscriber_for_current_thread].
326    pub fn trace_subscriber(&self) -> Option<Arc<dyn Subscriber + Send + Sync>> {
327        self.trace_subscriber.clone()
328    }
329
330    /// Some metric meters cannot be initialized until after a tokio runtime has started and after
331    /// other telemetry has initted (ex: prometheus). They can be attached here.
332    pub fn attach_late_init_metrics(&mut self, meter: Arc<dyn CoreMeter + 'static>) {
333        self.metrics = Some(meter);
334    }
335
336    /// Returns our wrapper for metric meters, including the `metric_prefix` from
337    /// [TelemetryOptions]. This should be used to initialize clients or for any other
338    /// temporal-owned metrics. User defined metrics should use [Self::get_metric_meter].
339    pub fn get_temporal_metric_meter(&self) -> Option<TemporalMeter> {
340        self.metrics.clone().map(|m| {
341            let kvs = self.default_kvs();
342            let attribs = NewAttributes::new(kvs);
343            TemporalMeter::new(
344                Arc::new(PrefixedMetricsMeter::new(self.metric_prefix.clone(), m))
345                    as Arc<dyn CoreMeter>,
346                attribs,
347                self.task_queue_label_strategy,
348            )
349        })
350    }
351
352    /// Returns our wrapper for metric meters, including attaching the service name if enabled.
353    pub fn get_metric_meter(&self) -> Option<TemporalMeter> {
354        self.metrics.clone().map(|m| {
355            let kvs = self.default_kvs();
356            let attribs = NewAttributes::new(kvs);
357            TemporalMeter::new(m, attribs, self.task_queue_label_strategy)
358        })
359    }
360
361    fn default_kvs(&self) -> Vec<MetricKeyValue> {
362        if self.attach_service_name {
363            vec![MetricKeyValue::new("service_name", TELEM_SERVICE_NAME)]
364        } else {
365            vec![]
366        }
367    }
368}
369
370thread_local! {
371    static SUB_GUARD: RefCell<Option<tracing::subscriber::DefaultGuard>> =
372        const { RefCell::new(None) };
373}
374/// Set the trace subscriber for the current thread. This must be done in every thread which uses
375/// core stuff, otherwise traces/logs will not be collected on that thread. For example, if using
376/// a multithreaded Tokio runtime, you should ensure that said runtime uses
377/// [on_thread_start](https://docs.rs/tokio/latest/tokio/runtime/struct.Builder.html#method.on_thread_start)
378/// or a similar mechanism to call this for each thread within the runtime.
379pub fn set_trace_subscriber_for_current_thread(sub: impl Subscriber + Send + Sync + 'static) {
380    SUB_GUARD.with(|sg| {
381        if sg.borrow().is_none() {
382            let g = tracing::subscriber::set_default(sub);
383            *sg.borrow_mut() = Some(g);
384        }
385    })
386}
387
388/// Undoes [set_trace_subscriber_for_current_thread]
389pub fn remove_trace_subscriber_for_current_thread() {
390    SUB_GUARD.take();
391}
392
393#[cfg(feature = "core-telemetry-bridge")]
394impl CoreTelemetry for TelemetryInstance {
395    fn fetch_buffered_logs(&self) -> Vec<CoreLog> {
396        if let Some(logs_out) = self.logs_out.as_ref() {
397            logs_out.lock().drain()
398        } else {
399            vec![]
400        }
401    }
402}
403
404/// Initialize tracing subscribers/output and logging export, returning a [TelemetryInstance]
405/// which can be used to register default / global tracing subscribers.
406///
407/// You should only call this once per unique [TelemetryOptions]
408///
409/// See [TelemetryOptions] docs for more on configuration.
410pub fn telemetry_init(opts: TelemetryOptions) -> Result<TelemetryInstance, anyhow::Error> {
411    telemetry_init_with_console_writer(
412        opts,
413        std::io::stdout,
414        env::var("TEMPORAL_CORE_PRETTY_LOGS").is_ok(),
415    )
416}
417
418fn telemetry_init_with_console_writer<W>(
419    opts: TelemetryOptions,
420    console_writer: W,
421    pretty_logs_env_set: bool,
422) -> Result<TelemetryInstance, anyhow::Error>
423where
424    W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static,
425{
426    #[cfg(feature = "core-telemetry-bridge")]
427    let mut logs_out = None;
428
429    // Tracing subscriber layers =========
430    let mut console_pretty_layer = None;
431    let mut console_compact_layer = None;
432    let mut console_json_layer = None;
433    #[cfg(feature = "core-telemetry-bridge")]
434    let mut forward_layer = None;
435    // ===================================
436
437    let tracing_sub = if let Some(ts) = opts.subscriber_override {
438        Some(ts)
439    } else {
440        opts.logging.map(|logger| {
441            match logger {
442                Logger::Console { filter, format } => {
443                    // This is silly dupe but can't be avoided without boxing.
444                    let format = format.unwrap_or(if pretty_logs_env_set {
445                        LoggerFormat::Pretty
446                    } else {
447                        LoggerFormat::Compact
448                    });
449                    match format {
450                        LoggerFormat::Pretty => {
451                            console_pretty_layer = Some(
452                                tracing_subscriber::fmt::layer()
453                                    .with_writer(console_writer)
454                                    .with_target(false)
455                                    .event_format(
456                                        tracing_subscriber::fmt::format()
457                                            .pretty()
458                                            .with_source_location(false),
459                                    )
460                                    .with_filter(EnvFilter::new(filter)),
461                            )
462                        }
463                        LoggerFormat::Compact => {
464                            console_compact_layer = Some(
465                                tracing_subscriber::fmt::layer()
466                                    .with_writer(console_writer)
467                                    .with_target(false)
468                                    .event_format(
469                                        tracing_subscriber::fmt::format()
470                                            .compact()
471                                            .with_source_location(false),
472                                    )
473                                    .with_filter(EnvFilter::new(filter)),
474                            )
475                        }
476                        LoggerFormat::Json => {
477                            console_json_layer = Some(
478                                tracing_subscriber::fmt::layer()
479                                    .with_writer(console_writer)
480                                    .with_target(false)
481                                    .json()
482                                    .with_filter(EnvFilter::new(filter)),
483                            )
484                        }
485                    }
486                }
487                #[cfg(feature = "core-telemetry-bridge")]
488                Logger::Forward { filter } => {
489                    let (export_layer, lo) =
490                        CoreLogConsumerLayer::new_buffered(FORWARD_LOG_BUFFER_SIZE);
491                    logs_out = Some(parking_lot::Mutex::new(lo));
492                    forward_layer = Some(export_layer.with_filter(EnvFilter::new(filter)));
493                }
494                #[cfg(feature = "core-telemetry-bridge")]
495                Logger::Push { filter, consumer } => {
496                    forward_layer = Some(
497                        CoreLogConsumerLayer::new(consumer).with_filter(EnvFilter::new(filter)),
498                    );
499                }
500            };
501            let reg = tracing_subscriber::registry()
502                .with(console_pretty_layer)
503                .with(console_compact_layer)
504                .with(console_json_layer);
505            #[cfg(feature = "core-telemetry-bridge")]
506            let reg = reg.with(forward_layer);
507
508            Arc::new(reg) as Arc<dyn Subscriber + Send + Sync>
509        })
510    };
511
512    Ok(TelemetryInstance {
513        metric_prefix: opts.metric_prefix,
514        #[cfg(feature = "core-telemetry-bridge")]
515        logs_out,
516        metrics: opts.metrics,
517        trace_subscriber: tracing_sub,
518        attach_service_name: opts.attach_service_name,
519        task_queue_label_strategy: opts.task_queue_label_strategy,
520    })
521}
522
523/// WARNING: Calling can cause panics because of <https://github.com/tokio-rs/tracing/issues/1656>
524/// Lang must not start using until resolved
525///
526/// Initialize telemetry/tracing globally. Useful for testing. Only takes affect when called
527/// the first time. Subsequent calls are ignored.
528pub fn telemetry_init_global(opts: TelemetryOptions) -> Result<(), anyhow::Error> {
529    static INITTED: AtomicBool = AtomicBool::new(false);
530    if INITTED
531        .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
532        .is_ok()
533    {
534        let ti = telemetry_init(opts)?;
535        if let Some(ts) = ti.trace_subscriber() {
536            tracing::subscriber::set_global_default(ts)?;
537        }
538    }
539    Ok(())
540}
541
542/// WARNING: Calling can cause panics because of <https://github.com/tokio-rs/tracing/issues/1656>
543/// Lang must not start using until resolved
544///
545/// Initialize the fallback global handler. All lang SDKs should call this somewhere, once, at
546/// startup, as it initializes a fallback handler for any dependencies (looking at you, otel) that
547/// don't provide good ways to customize their tracing usage. It sets a WARN-level global filter
548/// that uses the default console logger.
549pub fn telemetry_init_fallback() -> Result<(), anyhow::Error> {
550    telemetry_init_global(
551        TelemetryOptions::builder()
552            .logging(Logger::Console {
553                filter: construct_filter_string(Level::DEBUG, Level::WARN),
554                format: None,
555            })
556            .build(),
557    )?;
558    Ok(())
559}
560
561/// Ensure a process-wide default rustls `CryptoProvider` is installed.
562///
563/// Under the `tls-ring` build (and not `tls-aws-lc`), the reqwest clients used for ephemeral
564/// server downloads and the OTLP HTTP metric exporter carry no compiled-in crypto provider, so
565/// they resolve one from the process default at connection time. Installing ring here lets those
566/// clients negotiate TLS without pulling aws-lc-rs into the build. Idempotent, and tolerates a
567/// provider having already been installed by another component.
568pub fn ensure_default_crypto_provider() {
569    #[cfg(all(feature = "tls-ring", not(feature = "tls-aws-lc")))]
570    {
571        static INIT: std::sync::Once = std::sync::Once::new();
572        INIT.call_once(|| {
573            let _ = rustls::crypto::ring::default_provider().install_default();
574        });
575    }
576}
577
578#[cfg(test)]
579mod tests {
580    use super::*;
581    use std::{
582        io::Write,
583        sync::{Arc, Mutex},
584    };
585
586    #[derive(Clone)]
587    struct CapturingWriter(Arc<Mutex<Vec<u8>>>);
588
589    impl<'writer> MakeWriter<'writer> for CapturingWriter {
590        type Writer = CapturingWriter;
591
592        fn make_writer(&'writer self) -> Self::Writer {
593            self.clone()
594        }
595    }
596
597    impl Write for CapturingWriter {
598        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
599            self.0.lock().unwrap().extend_from_slice(buf);
600            Ok(buf.len())
601        }
602
603        fn flush(&mut self) -> std::io::Result<()> {
604            Ok(())
605        }
606    }
607
608    #[test]
609    fn json_console_logs_respect_filter() {
610        let output = Arc::new(Mutex::new(Vec::new()));
611        let instance = telemetry_init_with_console_writer(
612            TelemetryOptions::builder()
613                .logging(Logger::Console {
614                    filter: "off,json_log_test=info".to_string(),
615                    format: Some(LoggerFormat::Json),
616                })
617                .build(),
618            CapturingWriter(output.clone()),
619            false,
620        )
621        .unwrap();
622        let subscriber = instance.trace_subscriber().unwrap();
623
624        tracing::subscriber::with_default(subscriber, || {
625            tracing::info!(target: "json_log_test", answer = 42, "included");
626            tracing::debug!(target: "json_log_test", "filtered by level");
627            tracing::info!(target: "other_target", "filtered by target");
628        });
629
630        let output = String::from_utf8(output.lock().unwrap().clone()).unwrap();
631        let logs = output
632            .lines()
633            .map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
634            .collect::<Vec<_>>();
635        assert_eq!(logs.len(), 1);
636        assert_eq!(logs[0]["fields"]["message"], "included");
637        assert_eq!(logs[0]["fields"]["answer"], 42);
638    }
639
640    #[test]
641    fn explicit_format_overrides_pretty_logs_env() {
642        let output = Arc::new(Mutex::new(Vec::new()));
643        let instance = telemetry_init_with_console_writer(
644            TelemetryOptions::builder()
645                .logging(Logger::Console {
646                    filter: "info".to_string(),
647                    format: Some(LoggerFormat::Json),
648                })
649                .build(),
650            CapturingWriter(output.clone()),
651            true,
652        )
653        .unwrap();
654        let subscriber = instance.trace_subscriber().unwrap();
655
656        tracing::subscriber::with_default(subscriber, || tracing::info!("json log"));
657
658        let output = String::from_utf8(output.lock().unwrap().clone()).unwrap();
659        assert_eq!(
660            serde_json::from_str::<serde_json::Value>(&output).unwrap()["fields"]["message"],
661            "json log"
662        );
663    }
664
665    #[test]
666    fn pretty_logs_env_selects_pretty_when_format_unset() {
667        let output = Arc::new(Mutex::new(Vec::new()));
668        let instance = telemetry_init_with_console_writer(
669            TelemetryOptions::builder()
670                .logging(Logger::Console {
671                    filter: "info".to_string(),
672                    format: None,
673                })
674                .build(),
675            CapturingWriter(output.clone()),
676            true,
677        )
678        .unwrap();
679        let subscriber = instance.trace_subscriber().unwrap();
680
681        tracing::subscriber::with_default(subscriber, || tracing::info!("pretty log"));
682
683        let output = String::from_utf8(output.lock().unwrap().clone()).unwrap();
684        assert!(output.contains("pretty log"));
685        assert!(serde_json::from_str::<serde_json::Value>(&output).is_err());
686    }
687}
temporalio_common/telemetry.rs

temporalio_common/
telemetry.rs
