opentelemetry_sdk/logs/
logger_provider.rs

1use super::{BatchLogProcessor, LogProcessor, SdkLogger, SimpleLogProcessor};
2use crate::error::{OTelSdkError, OTelSdkResult};
3use crate::logs::LogExporter;
4use crate::Resource;
5use opentelemetry::{otel_debug, otel_info, InstrumentationScope};
6use std::time::Duration;
7use std::{
8    borrow::Cow,
9    sync::{
10        atomic::{AtomicBool, Ordering},
11        Arc, OnceLock,
12    },
13};
14
15// a no nop logger provider used as placeholder when the provider is shutdown
16// TODO - replace it with LazyLock once it is stable
17static NOOP_LOGGER_PROVIDER: OnceLock<SdkLoggerProvider> = OnceLock::new();
18
19#[inline]
20fn noop_logger_provider() -> &'static SdkLoggerProvider {
21    NOOP_LOGGER_PROVIDER.get_or_init(|| SdkLoggerProvider {
22        inner: Arc::new(LoggerProviderInner {
23            processors: Vec::new(),
24            is_shutdown: AtomicBool::new(true),
25        }),
26    })
27}
28
29#[derive(Debug, Clone)]
30/// Handles the creation and coordination of [`Logger`]s.
31///
32/// All `Logger`s created by a `SdkLoggerProvider` will share the same
33/// [`Resource`] and have their created log records processed by the
34/// configured log processors. This is a clonable handle to the `SdkLoggerProvider`
35/// itself, and cloning it will create a new reference, not a new instance of a
36/// `SdkLoggerProvider`. Dropping the last reference will trigger the shutdown of
37/// the provider, ensuring that all remaining logs are flushed and no further
38/// logs are processed. Shutdown can also be triggered manually by calling
39/// the [`shutdown`](SdkLoggerProvider::shutdown) method.
40///
41/// [`Logger`]: opentelemetry::logs::Logger
42/// [`Resource`]: crate::Resource
43pub struct SdkLoggerProvider {
44    inner: Arc<LoggerProviderInner>,
45}
46
47impl opentelemetry::logs::LoggerProvider for SdkLoggerProvider {
48    type Logger = SdkLogger;
49
50    fn logger(&self, name: impl Into<Cow<'static, str>>) -> Self::Logger {
51        let scope = InstrumentationScope::builder(name).build();
52        self.logger_with_scope(scope)
53    }
54
55    fn logger_with_scope(&self, scope: InstrumentationScope) -> Self::Logger {
56        // If the provider is shutdown, new logger will refer a no-op logger provider.
57        if self.inner.is_shutdown.load(Ordering::Relaxed) {
58            otel_debug!(
59                name: "LoggerProvider.NoOpLoggerReturned",
60                logger_name = scope.name(),
61            );
62            return SdkLogger::new(scope, noop_logger_provider().clone());
63        }
64        if scope.name().is_empty() {
65            otel_info!(name: "LoggerNameEmpty",  message = "Logger name is empty; consider providing a meaningful name. Logger will function normally and the provided name will be used as-is.");
66        };
67        otel_debug!(
68            name: "LoggerProvider.NewLoggerReturned",
69            logger_name = scope.name(),
70        );
71        SdkLogger::new(scope, self.clone())
72    }
73}
74
75impl SdkLoggerProvider {
76    /// Create a new `LoggerProvider` builder.
77    pub fn builder() -> LoggerProviderBuilder {
78        LoggerProviderBuilder::default()
79    }
80
81    pub(crate) fn log_processors(&self) -> &[Box<dyn LogProcessor>] {
82        &self.inner.processors
83    }
84
85    /// Force flush all remaining logs in log processors and return results.
86    pub fn force_flush(&self) -> OTelSdkResult {
87        let result: Vec<_> = self
88            .log_processors()
89            .iter()
90            .map(|processor| processor.force_flush())
91            .collect();
92        if result.iter().all(|r| r.is_ok()) {
93            Ok(())
94        } else {
95            Err(OTelSdkError::InternalFailure(format!("errs: {result:?}")))
96        }
97    }
98
99    /// Shuts down this `LoggerProvider`
100    pub fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
101        otel_debug!(
102            name: "LoggerProvider.ShutdownInvokedByUser",
103        );
104        if self
105            .inner
106            .is_shutdown
107            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
108            .is_ok()
109        {
110            // propagate the shutdown signal to processors
111            let result = self.inner.shutdown_with_timeout(timeout);
112            if result.iter().all(|res| res.is_ok()) {
113                Ok(())
114            } else {
115                Err(OTelSdkError::InternalFailure(format!(
116                    "Shutdown errors: {:?}",
117                    result
118                        .into_iter()
119                        .filter_map(Result::err)
120                        .collect::<Vec<_>>()
121                )))
122            }
123        } else {
124            Err(OTelSdkError::AlreadyShutdown)
125        }
126    }
127
128    /// Shuts down this `LoggerProvider` with default timeout
129    pub fn shutdown(&self) -> OTelSdkResult {
130        self.shutdown_with_timeout(Duration::from_secs(5))
131    }
132}
133
134#[derive(Debug)]
135struct LoggerProviderInner {
136    processors: Vec<Box<dyn LogProcessor>>,
137    is_shutdown: AtomicBool,
138}
139
140impl LoggerProviderInner {
141    /// Shuts down the `LoggerProviderInner` and returns any errors.
142    pub(crate) fn shutdown_with_timeout(&self, timeout: Duration) -> Vec<OTelSdkResult> {
143        let mut results = vec![];
144        for processor in &self.processors {
145            let result = processor.shutdown_with_timeout(timeout);
146            if let Err(err) = &result {
147                // Log at debug level because:
148                //  - The error is also returned to the user for handling (if applicable)
149                //  - Or the error occurs during `TracerProviderInner::Drop` as part of telemetry shutdown,
150                //    which is non-actionable by the user
151                otel_debug!(name: "LoggerProvider.ShutdownError",
152                        error = format!("{err}"));
153            }
154            results.push(result);
155        }
156        results
157    }
158
159    /// Shuts down the `LoggerProviderInner` with default timeout and returns any errors.
160    pub(crate) fn shutdown(&self) -> Vec<OTelSdkResult> {
161        self.shutdown_with_timeout(Duration::from_secs(5))
162    }
163}
164
165impl Drop for LoggerProviderInner {
166    fn drop(&mut self) {
167        if !self.is_shutdown.load(Ordering::Relaxed) {
168            otel_info!(
169                name: "LoggerProvider.Drop",
170                message = "Last reference of LoggerProvider dropped, initiating shutdown."
171            );
172            let _ = self.shutdown(); // errors are handled within shutdown
173        } else {
174            otel_debug!(
175                name: "LoggerProvider.Drop.AlreadyShutdown",
176                message = "LoggerProvider was already shut down; drop will not attempt shutdown again."
177            );
178        }
179    }
180}
181
182#[derive(Debug, Default)]
183/// Builder for provider attributes.
184pub struct LoggerProviderBuilder {
185    processors: Vec<Box<dyn LogProcessor>>,
186    resource: Option<Resource>,
187}
188
189impl LoggerProviderBuilder {
190    /// Adds a [SimpleLogProcessor] with the configured exporter to the pipeline.
191    ///
192    /// # Arguments
193    ///
194    /// * `exporter` - The exporter to be used by the SimpleLogProcessor.
195    ///
196    /// # Returns
197    ///
198    /// A new `Builder` instance with the SimpleLogProcessor added to the pipeline.
199    ///
200    /// Processors are invoked in the order they are added.
201    pub fn with_simple_exporter<T: LogExporter + 'static>(self, exporter: T) -> Self {
202        let mut processors = self.processors;
203        processors.push(Box::new(SimpleLogProcessor::new(exporter)));
204
205        LoggerProviderBuilder { processors, ..self }
206    }
207
208    /// Adds a [BatchLogProcessor] with the configured exporter to the pipeline,
209    /// using the default [super::BatchConfig].
210    ///
211    /// The following environment variables can be used to configure the batching configuration:
212    ///
213    /// * `OTEL_BLRP_SCHEDULE_DELAY` - Corresponds to `with_scheduled_delay`.
214    /// * `OTEL_BLRP_MAX_QUEUE_SIZE` - Corresponds to `with_max_queue_size`.
215    /// * `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE` - Corresponds to `with_max_export_batch_size`.
216    ///
217    /// # Arguments
218    ///
219    /// * `exporter` - The exporter to be used by the `BatchLogProcessor`.
220    ///
221    /// # Returns
222    ///
223    /// A new `LoggerProviderBuilder` instance with the `BatchLogProcessor` added to the pipeline.
224    ///
225    /// Processors are invoked in the order they are added.
226    pub fn with_batch_exporter<T: LogExporter + 'static>(self, exporter: T) -> Self {
227        let batch = BatchLogProcessor::builder(exporter).build();
228        self.with_log_processor(batch)
229    }
230
231    /// Adds a custom [LogProcessor] to the pipeline.
232    ///
233    /// # Arguments
234    ///
235    /// * `processor` - The `LogProcessor` to be added.
236    ///
237    /// # Returns
238    ///
239    /// A new `Builder` instance with the custom `LogProcessor` added to the pipeline.
240    ///
241    /// Processors are invoked in the order they are added.
242    pub fn with_log_processor<T: LogProcessor + 'static>(self, processor: T) -> Self {
243        let mut processors = self.processors;
244        processors.push(Box::new(processor));
245
246        LoggerProviderBuilder { processors, ..self }
247    }
248
249    /// Associates a [Resource] with a [SdkLoggerProvider].
250    ///
251    /// This [Resource] represents the entity producing telemetry and is associated
252    /// with all `Logger`s the [SdkLoggerProvider] will create.
253    ///
254    /// By default, if this option is not used, the default [Resource] will be used.
255    ///
256    /// When constructing a [Resource], use [`Resource::builder()`] to preserve
257    /// SDK-provided defaults such as `telemetry.sdk.*` and `service.name`.
258    /// Using [`Resource::builder_empty()`] will **not** include these attributes.
259    ///
260    /// # Example
261    ///
262    /// ```
263    /// use opentelemetry_sdk::{Resource, logs::SdkLoggerProvider};
264    /// use opentelemetry::KeyValue;
265    ///
266    /// let provider = SdkLoggerProvider::builder()
267    ///     .with_resource(
268    ///         Resource::builder()
269    ///             .with_service_name("my-service")
270    ///             .with_attributes([KeyValue::new("deployment.environment.name", "production")])
271    ///             .build(),
272    ///     )
273    ///     .build();
274    /// ```
275    ///
276    /// *Note*: Calls to this method are additive, each call merges the provided
277    /// resource with the previous one.
278    ///
279    /// [`Resource::builder()`]: Resource::builder
280    /// [`Resource::builder_empty()`]: Resource::builder_empty
281    pub fn with_resource(self, resource: Resource) -> Self {
282        let resource = match self.resource {
283            Some(existing) => Some(existing.merge(&resource)),
284            None => Some(resource),
285        };
286
287        LoggerProviderBuilder { resource, ..self }
288    }
289
290    /// Create a new provider from this configuration.
291    pub fn build(self) -> SdkLoggerProvider {
292        let resource = self.resource.unwrap_or(Resource::builder().build());
293        let mut processors = self.processors;
294        for processor in &mut processors {
295            processor.set_resource(&resource);
296        }
297
298        let logger_provider = SdkLoggerProvider {
299            inner: Arc::new(LoggerProviderInner {
300                processors,
301                is_shutdown: AtomicBool::new(false),
302            }),
303        };
304
305        otel_debug!(
306            name: "LoggerProvider.Built",
307        );
308        logger_provider
309    }
310}
311
312#[cfg(test)]
313mod tests {
314    #[cfg(feature = "trace")]
315    use crate::logs::TraceContext;
316    #[cfg(feature = "trace")]
317    use crate::trace::SdkTracerProvider;
318    use crate::{
319        logs::{InMemoryLogExporter, LogBatch, SdkLogRecord},
320        resource::{
321            SERVICE_NAME, TELEMETRY_SDK_LANGUAGE, TELEMETRY_SDK_NAME, TELEMETRY_SDK_VERSION,
322        },
323        Resource,
324    };
325
326    use super::*;
327    use opentelemetry::logs::{AnyValue, LogRecord as _, Logger, LoggerProvider};
328    #[cfg(feature = "trace")]
329    use opentelemetry::trace::TraceContextExt;
330    #[cfg(feature = "trace")]
331    use opentelemetry::trace::{SpanId, TraceId, Tracer as _, TracerProvider};
332    use opentelemetry::{Key, KeyValue, Value};
333    use std::fmt::{Debug, Formatter};
334    use std::sync::atomic::AtomicU64;
335    use std::sync::Mutex;
336    use std::{thread, time};
337
338    struct ShutdownTestLogProcessor {
339        is_shutdown: Arc<Mutex<bool>>,
340        counter: Arc<AtomicU64>,
341    }
342
343    impl Debug for ShutdownTestLogProcessor {
344        fn fmt(&self, _f: &mut Formatter<'_>) -> std::fmt::Result {
345            todo!()
346        }
347    }
348
349    impl ShutdownTestLogProcessor {
350        pub(crate) fn new(counter: Arc<AtomicU64>) -> Self {
351            ShutdownTestLogProcessor {
352                is_shutdown: Arc::new(Mutex::new(false)),
353                counter,
354            }
355        }
356    }
357
358    impl LogProcessor for ShutdownTestLogProcessor {
359        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
360            self.is_shutdown
361                .lock()
362                .map(|is_shutdown| {
363                    if !*is_shutdown {
364                        self.counter
365                            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
366                    }
367                })
368                .expect("lock poisoned");
369        }
370
371        fn force_flush(&self) -> OTelSdkResult {
372            Ok(())
373        }
374
375        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
376            self.is_shutdown
377                .lock()
378                .map(|mut is_shutdown| *is_shutdown = true)
379                .expect("lock poisoned");
380            Ok(())
381        }
382    }
383
384    #[derive(Debug, Clone)]
385    struct TestExporterForResource {
386        resource: Arc<Mutex<Resource>>,
387    }
388    impl TestExporterForResource {
389        fn new() -> Self {
390            TestExporterForResource {
391                resource: Arc::new(Mutex::new(Resource::empty())),
392            }
393        }
394
395        fn resource(&self) -> Resource {
396            self.resource.lock().unwrap().clone()
397        }
398    }
399    impl LogExporter for TestExporterForResource {
400        async fn export(&self, _: LogBatch<'_>) -> OTelSdkResult {
401            Ok(())
402        }
403
404        fn set_resource(&mut self, resource: &Resource) {
405            let mut res = self.resource.lock().unwrap();
406            *res = resource.clone();
407        }
408
409        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
410            Ok(())
411        }
412    }
413
414    #[derive(Debug, Clone)]
415    struct TestProcessorForResource {
416        resource: Arc<Mutex<Resource>>,
417        exporter: TestExporterForResource,
418    }
419    impl LogProcessor for TestProcessorForResource {
420        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
421            // nothing to do.
422        }
423
424        fn force_flush(&self) -> OTelSdkResult {
425            Ok(())
426        }
427
428        fn set_resource(&mut self, resource: &Resource) {
429            let mut res = self.resource.lock().unwrap();
430            *res = resource.clone();
431            self.exporter.set_resource(resource);
432        }
433
434        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
435            Ok(())
436        }
437    }
438    impl TestProcessorForResource {
439        fn new(exporter: TestExporterForResource) -> Self {
440            TestProcessorForResource {
441                resource: Arc::new(Mutex::new(Resource::empty())),
442                exporter,
443            }
444        }
445        fn resource(&self) -> Resource {
446            self.resource.lock().unwrap().clone()
447        }
448    }
449
450    #[test]
451    fn test_resource_handling_provider_processor_exporter() {
452        let assert_resource = |processor: &TestProcessorForResource,
453                               exporter: &TestExporterForResource,
454                               resource_key: &'static str,
455                               expect: Option<&'static str>| {
456            assert_eq!(
457                processor
458                    .resource()
459                    .get(&Key::from_static_str(resource_key))
460                    .map(|v| v.to_string()),
461                expect.map(|s| s.to_string())
462            );
463
464            assert_eq!(
465                exporter
466                    .resource()
467                    .get(&Key::from_static_str(resource_key))
468                    .map(|v| v.to_string()),
469                expect.map(|s| s.to_string())
470            );
471        };
472        let assert_telemetry_resource =
473            |processor: &TestProcessorForResource, exporter: &TestExporterForResource| {
474                assert_eq!(
475                    processor.resource().get(&TELEMETRY_SDK_LANGUAGE.into()),
476                    Some(Value::from("rust"))
477                );
478                assert_eq!(
479                    processor.resource().get(&TELEMETRY_SDK_NAME.into()),
480                    Some(Value::from("opentelemetry"))
481                );
482                assert_eq!(
483                    processor.resource().get(&TELEMETRY_SDK_VERSION.into()),
484                    Some(Value::from(env!("CARGO_PKG_VERSION")))
485                );
486                assert_eq!(
487                    exporter.resource().get(&TELEMETRY_SDK_LANGUAGE.into()),
488                    Some(Value::from("rust"))
489                );
490                assert_eq!(
491                    exporter.resource().get(&TELEMETRY_SDK_NAME.into()),
492                    Some(Value::from("opentelemetry"))
493                );
494                assert_eq!(
495                    exporter.resource().get(&TELEMETRY_SDK_VERSION.into()),
496                    Some(Value::from(env!("CARGO_PKG_VERSION")))
497                );
498            };
499
500        // If users didn't provide a resource and there isn't a env var set. Use default one.
501        temp_env::with_var_unset("OTEL_RESOURCE_ATTRIBUTES", || {
502            let exporter_with_resource = TestExporterForResource::new();
503            let processor_with_resource =
504                TestProcessorForResource::new(exporter_with_resource.clone());
505            let _ = super::SdkLoggerProvider::builder()
506                .with_log_processor(processor_with_resource.clone())
507                .build();
508            let service_name = processor_with_resource
509                .resource()
510                .get(&Key::from_static_str(SERVICE_NAME))
511                .map(|v| v.to_string())
512                .unwrap();
513            assert!(
514                service_name.starts_with("unknown_service:opentelemetry_sdk-"),
515                "Expected service name to start with 'unknown_service:opentelemetry_sdk-', got: {}",
516                service_name
517            );
518            assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
519        });
520
521        // If user provided a resource, use that.
522        let exporter_with_resource = TestExporterForResource::new();
523        let processor_with_resource = TestProcessorForResource::new(exporter_with_resource.clone());
524        let _ = super::SdkLoggerProvider::builder()
525            .with_resource(
526                Resource::builder_empty()
527                    .with_service_name("test_service")
528                    .build(),
529            )
530            .with_log_processor(processor_with_resource.clone())
531            .build();
532        assert_resource(
533            &processor_with_resource,
534            &exporter_with_resource,
535            SERVICE_NAME,
536            Some("test_service"),
537        );
538        assert_eq!(processor_with_resource.resource().len(), 1);
539
540        // If `OTEL_RESOURCE_ATTRIBUTES` is set, read them automatically
541        temp_env::with_var(
542            "OTEL_RESOURCE_ATTRIBUTES",
543            Some("key1=value1, k2, k3=value2"),
544            || {
545                let exporter_with_resource = TestExporterForResource::new();
546                let processor_with_resource =
547                    TestProcessorForResource::new(exporter_with_resource.clone());
548                let _ = super::SdkLoggerProvider::builder()
549                    .with_log_processor(processor_with_resource.clone())
550                    .build();
551                let service_name = processor_with_resource
552                    .resource()
553                    .get(&Key::from_static_str(SERVICE_NAME))
554                    .map(|v| v.to_string())
555                    .unwrap();
556                assert!(
557                    service_name.starts_with("unknown_service:opentelemetry_sdk-"),
558                    "Expected service name to start with 'unknown_service:opentelemetry_sdk-', got: {}",
559                    service_name
560                );
561                assert_resource(
562                    &processor_with_resource,
563                    &exporter_with_resource,
564                    "key1",
565                    Some("value1"),
566                );
567                assert_resource(
568                    &processor_with_resource,
569                    &exporter_with_resource,
570                    "k3",
571                    Some("value2"),
572                );
573                assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
574                assert_eq!(processor_with_resource.resource().len(), 6);
575            },
576        );
577
578        // When `OTEL_RESOURCE_ATTRIBUTES` is set and also user provided config
579        temp_env::with_var(
580            "OTEL_RESOURCE_ATTRIBUTES",
581            Some("my-custom-key=env-val,k2=value2"),
582            || {
583                let exporter_with_resource = TestExporterForResource::new();
584                let processor_with_resource =
585                    TestProcessorForResource::new(exporter_with_resource.clone());
586                let _ = super::SdkLoggerProvider::builder()
587                    .with_resource(
588                        Resource::builder()
589                            .with_attributes([
590                                KeyValue::new("my-custom-key", "my-custom-value"),
591                                KeyValue::new("my-custom-key2", "my-custom-value2"),
592                            ])
593                            .build(),
594                    )
595                    .with_log_processor(processor_with_resource.clone())
596                    .build();
597                let service_name = processor_with_resource
598                    .resource()
599                    .get(&Key::from_static_str(SERVICE_NAME))
600                    .map(|v| v.to_string())
601                    .unwrap();
602                assert!(
603                    service_name.starts_with("unknown_service:opentelemetry_sdk-"),
604                    "Expected service name to start with 'unknown_service:opentelemetry_sdk-', got: {}",
605                    service_name
606                );
607                assert_resource(
608                    &processor_with_resource,
609                    &exporter_with_resource,
610                    "my-custom-key",
611                    Some("my-custom-value"),
612                );
613                assert_resource(
614                    &processor_with_resource,
615                    &exporter_with_resource,
616                    "my-custom-key2",
617                    Some("my-custom-value2"),
618                );
619                assert_resource(
620                    &processor_with_resource,
621                    &exporter_with_resource,
622                    "k2",
623                    Some("value2"),
624                );
625                assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
626                assert_eq!(processor_with_resource.resource().len(), 7);
627            },
628        );
629
630        // If user provided a resource, it takes priority during collision.
631        let exporter_with_resource = TestExporterForResource::new();
632        let processor_with_resource = TestProcessorForResource::new(exporter_with_resource);
633        let _ = super::SdkLoggerProvider::builder()
634            .with_resource(Resource::empty())
635            .with_log_processor(processor_with_resource.clone())
636            .build();
637        assert_eq!(processor_with_resource.resource().len(), 0);
638    }
639
640    #[test]
641    #[cfg(feature = "trace")]
642    fn trace_context_test() {
643        let exporter = InMemoryLogExporter::default();
644
645        let logger_provider = SdkLoggerProvider::builder()
646            .with_simple_exporter(exporter.clone())
647            .build();
648
649        let logger = logger_provider.logger("test-logger");
650
651        let tracer_provider = SdkTracerProvider::builder().build();
652
653        let tracer = tracer_provider.tracer("test-tracer");
654
655        tracer.in_span("test-span", |cx| {
656            let ambient_ctxt = cx.span().span_context().clone();
657            let explicit_ctxt = TraceContext {
658                trace_id: TraceId::from(13),
659                span_id: SpanId::from(14),
660                trace_flags: None,
661            };
662
663            let mut ambient_ctxt_record = logger.create_log_record();
664            ambient_ctxt_record.set_body(AnyValue::String("ambient".into()));
665
666            let mut explicit_ctxt_record = logger.create_log_record();
667            explicit_ctxt_record.set_body(AnyValue::String("explicit".into()));
668            explicit_ctxt_record.set_trace_context(
669                explicit_ctxt.trace_id,
670                explicit_ctxt.span_id,
671                explicit_ctxt.trace_flags,
672            );
673
674            logger.emit(ambient_ctxt_record);
675            logger.emit(explicit_ctxt_record);
676
677            let emitted = exporter.get_emitted_logs().unwrap();
678
679            assert_eq!(
680                Some(AnyValue::String("ambient".into())),
681                emitted[0].record.body
682            );
683            assert_eq!(
684                ambient_ctxt.trace_id(),
685                emitted[0].record.trace_context.as_ref().unwrap().trace_id
686            );
687            assert_eq!(
688                ambient_ctxt.span_id(),
689                emitted[0].record.trace_context.as_ref().unwrap().span_id
690            );
691
692            assert_eq!(
693                Some(AnyValue::String("explicit".into())),
694                emitted[1].record.body
695            );
696            assert_eq!(
697                explicit_ctxt.trace_id,
698                emitted[1].record.trace_context.as_ref().unwrap().trace_id
699            );
700            assert_eq!(
701                explicit_ctxt.span_id,
702                emitted[1].record.trace_context.as_ref().unwrap().span_id
703            );
704        });
705    }
706
707    #[test]
708    fn shutdown_test() {
709        let counter = Arc::new(AtomicU64::new(0));
710        let logger_provider = SdkLoggerProvider::builder()
711            .with_log_processor(ShutdownTestLogProcessor::new(counter.clone()))
712            .build();
713
714        let logger1 = logger_provider.logger("test-logger1");
715        let logger2 = logger_provider.logger("test-logger2");
716        logger1.emit(logger1.create_log_record());
717        logger2.emit(logger1.create_log_record());
718
719        let logger3 = logger_provider.logger("test-logger3");
720        let handle = thread::spawn(move || {
721            logger3.emit(logger3.create_log_record());
722        });
723        handle.join().expect("thread panicked");
724
725        let _ = logger_provider.shutdown();
726        logger1.emit(logger1.create_log_record());
727
728        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 3);
729    }
730
731    #[test]
732    fn shutdown_idempotent_test() {
733        let counter = Arc::new(AtomicU64::new(0));
734        let logger_provider = SdkLoggerProvider::builder()
735            .with_log_processor(ShutdownTestLogProcessor::new(counter.clone()))
736            .build();
737
738        let shutdown_res = logger_provider.shutdown();
739        assert!(shutdown_res.is_ok());
740
741        // Subsequent shutdowns should return an error.
742        let shutdown_res = logger_provider.shutdown();
743        assert!(shutdown_res.is_err());
744
745        // Subsequent shutdowns should return an error.
746        let shutdown_res = logger_provider.shutdown();
747        assert!(shutdown_res.is_err());
748    }
749
750    #[test]
751    fn global_shutdown_test() {
752        // cargo test global_shutdown_test --features=testing
753
754        // Arrange
755        let shutdown_called = Arc::new(Mutex::new(false));
756        let flush_called = Arc::new(Mutex::new(false));
757        let logger_provider = SdkLoggerProvider::builder()
758            .with_log_processor(LazyLogProcessor::new(
759                shutdown_called.clone(),
760                flush_called.clone(),
761            ))
762            .build();
763        //set_logger_provider(logger_provider);
764        let logger1 = logger_provider.logger("test-logger1");
765        let logger2 = logger_provider.logger("test-logger2");
766
767        // Acts
768        logger1.emit(logger1.create_log_record());
769        logger2.emit(logger1.create_log_record());
770
771        // explicitly calling shutdown on logger_provider. This will
772        // indeed do the shutdown, even if there are loggers still alive.
773        let _ = logger_provider.shutdown();
774
775        // Assert
776
777        // shutdown is called.
778        assert!(*shutdown_called.lock().unwrap());
779
780        // flush is never called by the sdk.
781        assert!(!*flush_called.lock().unwrap());
782    }
783
784    #[test]
785    fn drop_test_with_multiple_providers() {
786        let shutdown_called = Arc::new(Mutex::new(false));
787        let flush_called = Arc::new(Mutex::new(false));
788        {
789            // Create a shared LoggerProviderInner and use it across multiple providers
790            let shared_inner = Arc::new(LoggerProviderInner {
791                processors: vec![Box::new(LazyLogProcessor::new(
792                    shutdown_called.clone(),
793                    flush_called.clone(),
794                ))],
795                is_shutdown: AtomicBool::new(false),
796            });
797
798            {
799                let logger_provider1 = SdkLoggerProvider {
800                    inner: shared_inner.clone(),
801                };
802                let logger_provider2 = SdkLoggerProvider {
803                    inner: shared_inner.clone(),
804                };
805
806                let logger1 = logger_provider1.logger("test-logger1");
807                let logger2 = logger_provider2.logger("test-logger2");
808
809                logger1.emit(logger1.create_log_record());
810                logger2.emit(logger1.create_log_record());
811
812                // LoggerProviderInner should not be dropped yet, since both providers and `shared_inner`
813                // are still holding a reference.
814            }
815            // At this point, both `logger_provider1` and `logger_provider2` are dropped,
816            // but `shared_inner` still holds a reference, so `LoggerProviderInner` is NOT dropped yet.
817        }
818        // Verify shutdown was called during the drop of the shared LoggerProviderInner
819        assert!(*shutdown_called.lock().unwrap());
820        // Verify flush was not called during drop
821        assert!(!*flush_called.lock().unwrap());
822    }
823
824    #[test]
825    fn drop_after_shutdown_test_with_multiple_providers() {
826        let shutdown_called = Arc::new(Mutex::new(0)); // Count the number of times shutdown is called
827        let flush_called = Arc::new(Mutex::new(false));
828
829        // Create a shared LoggerProviderInner and use it across multiple providers
830        let shared_inner = Arc::new(LoggerProviderInner {
831            processors: vec![Box::new(CountingShutdownProcessor::new(
832                shutdown_called.clone(),
833                flush_called.clone(),
834            ))],
835            is_shutdown: AtomicBool::new(false),
836        });
837
838        // Create a scope to test behavior when providers are dropped
839        {
840            let logger_provider1 = SdkLoggerProvider {
841                inner: shared_inner.clone(),
842            };
843            let logger_provider2 = SdkLoggerProvider {
844                inner: shared_inner.clone(),
845            };
846
847            // Explicitly shut down the logger provider
848            let shutdown_result = logger_provider1.shutdown();
849            println!("---->Result: {shutdown_result:?}");
850            assert!(shutdown_result.is_ok());
851
852            // Verify that shutdown was called exactly once
853            assert_eq!(*shutdown_called.lock().unwrap(), 1);
854
855            // LoggerProvider2 should observe the shutdown state but not trigger another shutdown
856            let shutdown_result2 = logger_provider2.shutdown();
857            assert!(shutdown_result2.is_err());
858
859            // Both logger providers will be dropped at the end of this scope
860        }
861
862        // Verify that shutdown was only called once, even after drop
863        assert_eq!(*shutdown_called.lock().unwrap(), 1);
864    }
865
866    #[test]
867    fn test_empty_logger_name() {
868        let exporter = InMemoryLogExporter::default();
869        let logger_provider = SdkLoggerProvider::builder()
870            .with_simple_exporter(exporter.clone())
871            .build();
872        let logger = logger_provider.logger("");
873        let mut record = logger.create_log_record();
874        record.set_body("Testing empty logger name".into());
875        logger.emit(record);
876
877        // Create a logger using a scope with an empty name
878        let scope = InstrumentationScope::builder("").build();
879        let scoped_logger = logger_provider.logger_with_scope(scope);
880        let mut scoped_record = scoped_logger.create_log_record();
881        scoped_record.set_body("Testing empty logger scope name".into());
882        scoped_logger.emit(scoped_record);
883
884        // Assert: Verify that the emitted logs are processed correctly
885        let mut emitted_logs = exporter.get_emitted_logs().unwrap();
886        assert_eq!(emitted_logs.len(), 2);
887        let log1 = emitted_logs.remove(0);
888        // Assert the first log
889        assert_eq!(
890            log1.record.body,
891            Some(AnyValue::String("Testing empty logger name".into()))
892        );
893        assert_eq!(log1.instrumentation.name(), "");
894
895        // Assert the second log created through the scope
896        let log2 = emitted_logs.remove(0);
897        assert_eq!(
898            log2.record.body,
899            Some(AnyValue::String("Testing empty logger scope name".into()))
900        );
901        assert_eq!(log1.instrumentation.name(), "");
902    }
903
904    #[test]
905    fn with_resource_multiple_calls_ensure_additive() {
906        let builder = SdkLoggerProvider::builder()
907            .with_resource(
908                Resource::builder_empty()
909                    .with_attributes(vec![KeyValue::new("key1", "value1")])
910                    .build(),
911            )
912            .with_resource(
913                Resource::builder_empty()
914                    .with_attributes(vec![KeyValue::new("key2", "value2")])
915                    .build(),
916            )
917            .with_resource(
918                Resource::builder_empty()
919                    .with_schema_url(vec![], "http://example.com")
920                    .build(),
921            )
922            .with_resource(
923                Resource::builder_empty()
924                    .with_attributes(vec![KeyValue::new("key3", "value3")])
925                    .build(),
926            );
927
928        let resource = builder.resource.unwrap();
929
930        assert_eq!(
931            resource.get(&Key::from_static_str("key1")),
932            Some(Value::from("value1"))
933        );
934        assert_eq!(
935            resource.get(&Key::from_static_str("key2")),
936            Some(Value::from("value2"))
937        );
938        assert_eq!(
939            resource.get(&Key::from_static_str("key3")),
940            Some(Value::from("value3"))
941        );
942        assert_eq!(resource.schema_url(), Some("http://example.com"));
943    }
944
945    #[derive(Debug)]
946    pub(crate) struct LazyLogProcessor {
947        shutdown_called: Arc<Mutex<bool>>,
948        flush_called: Arc<Mutex<bool>>,
949    }
950
951    impl LazyLogProcessor {
952        pub(crate) fn new(
953            shutdown_called: Arc<Mutex<bool>>,
954            flush_called: Arc<Mutex<bool>>,
955        ) -> Self {
956            LazyLogProcessor {
957                shutdown_called,
958                flush_called,
959            }
960        }
961    }
962
963    impl LogProcessor for LazyLogProcessor {
964        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
965            // nothing to do.
966        }
967
968        fn force_flush(&self) -> OTelSdkResult {
969            *self.flush_called.lock().unwrap() = true;
970            Ok(())
971        }
972
973        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
974            *self.shutdown_called.lock().unwrap() = true;
975            Ok(())
976        }
977    }
978
979    #[derive(Debug)]
980    struct CountingShutdownProcessor {
981        shutdown_count: Arc<Mutex<i32>>,
982        flush_called: Arc<Mutex<bool>>,
983    }
984
985    impl CountingShutdownProcessor {
986        fn new(shutdown_count: Arc<Mutex<i32>>, flush_called: Arc<Mutex<bool>>) -> Self {
987            CountingShutdownProcessor {
988                shutdown_count,
989                flush_called,
990            }
991        }
992    }
993
994    impl LogProcessor for CountingShutdownProcessor {
995        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
996            // nothing to do
997        }
998
999        fn force_flush(&self) -> OTelSdkResult {
1000            *self.flush_called.lock().unwrap() = true;
1001            Ok(())
1002        }
1003
1004        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
1005            let mut count = self.shutdown_count.lock().unwrap();
1006            *count += 1;
1007            Ok(())
1008        }
1009    }
1010}
opentelemetry_sdk/logs/logger_provider.rs

opentelemetry_sdk/logs/
logger_provider.rs
