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    /// The `Resource` to be associated with this Provider.
250    ///
251    /// *Note*: Calls to this method are additive, each call merges the provided
252    /// resource with the previous one.
253    pub fn with_resource(self, resource: Resource) -> Self {
254        let resource = match self.resource {
255            Some(existing) => Some(existing.merge(&resource)),
256            None => Some(resource),
257        };
258
259        LoggerProviderBuilder { resource, ..self }
260    }
261
262    /// Create a new provider from this configuration.
263    pub fn build(self) -> SdkLoggerProvider {
264        let resource = self.resource.unwrap_or(Resource::builder().build());
265        let mut processors = self.processors;
266        for processor in &mut processors {
267            processor.set_resource(&resource);
268        }
269
270        let logger_provider = SdkLoggerProvider {
271            inner: Arc::new(LoggerProviderInner {
272                processors,
273                is_shutdown: AtomicBool::new(false),
274            }),
275        };
276
277        otel_debug!(
278            name: "LoggerProvider.Built",
279        );
280        logger_provider
281    }
282}
283
284#[cfg(test)]
285mod tests {
286    use crate::{
287        logs::{InMemoryLogExporter, LogBatch, SdkLogRecord, TraceContext},
288        resource::{
289            SERVICE_NAME, TELEMETRY_SDK_LANGUAGE, TELEMETRY_SDK_NAME, TELEMETRY_SDK_VERSION,
290        },
291        trace::SdkTracerProvider,
292        Resource,
293    };
294
295    use super::*;
296    use opentelemetry::trace::{SpanId, TraceId, Tracer as _, TracerProvider};
297    use opentelemetry::{
298        logs::{AnyValue, LogRecord as _, Logger, LoggerProvider},
299        trace::TraceContextExt,
300    };
301    use opentelemetry::{Key, KeyValue, Value};
302    use std::fmt::{Debug, Formatter};
303    use std::sync::atomic::AtomicU64;
304    use std::sync::Mutex;
305    use std::{thread, time};
306
307    struct ShutdownTestLogProcessor {
308        is_shutdown: Arc<Mutex<bool>>,
309        counter: Arc<AtomicU64>,
310    }
311
312    impl Debug for ShutdownTestLogProcessor {
313        fn fmt(&self, _f: &mut Formatter<'_>) -> std::fmt::Result {
314            todo!()
315        }
316    }
317
318    impl ShutdownTestLogProcessor {
319        pub(crate) fn new(counter: Arc<AtomicU64>) -> Self {
320            ShutdownTestLogProcessor {
321                is_shutdown: Arc::new(Mutex::new(false)),
322                counter,
323            }
324        }
325    }
326
327    impl LogProcessor for ShutdownTestLogProcessor {
328        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
329            self.is_shutdown
330                .lock()
331                .map(|is_shutdown| {
332                    if !*is_shutdown {
333                        self.counter
334                            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
335                    }
336                })
337                .expect("lock poisoned");
338        }
339
340        fn force_flush(&self) -> OTelSdkResult {
341            Ok(())
342        }
343
344        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
345            self.is_shutdown
346                .lock()
347                .map(|mut is_shutdown| *is_shutdown = true)
348                .expect("lock poisoned");
349            Ok(())
350        }
351    }
352
353    #[derive(Debug, Clone)]
354    struct TestExporterForResource {
355        resource: Arc<Mutex<Resource>>,
356    }
357    impl TestExporterForResource {
358        fn new() -> Self {
359            TestExporterForResource {
360                resource: Arc::new(Mutex::new(Resource::empty())),
361            }
362        }
363
364        fn resource(&self) -> Resource {
365            self.resource.lock().unwrap().clone()
366        }
367    }
368    impl LogExporter for TestExporterForResource {
369        async fn export(&self, _: LogBatch<'_>) -> OTelSdkResult {
370            Ok(())
371        }
372
373        fn set_resource(&mut self, resource: &Resource) {
374            let mut res = self.resource.lock().unwrap();
375            *res = resource.clone();
376        }
377
378        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
379            Ok(())
380        }
381    }
382
383    #[derive(Debug, Clone)]
384    struct TestProcessorForResource {
385        resource: Arc<Mutex<Resource>>,
386        exporter: TestExporterForResource,
387    }
388    impl LogProcessor for TestProcessorForResource {
389        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
390            // nothing to do.
391        }
392
393        fn force_flush(&self) -> OTelSdkResult {
394            Ok(())
395        }
396
397        fn set_resource(&mut self, resource: &Resource) {
398            let mut res = self.resource.lock().unwrap();
399            *res = resource.clone();
400            self.exporter.set_resource(resource);
401        }
402    }
403    impl TestProcessorForResource {
404        fn new(exporter: TestExporterForResource) -> Self {
405            TestProcessorForResource {
406                resource: Arc::new(Mutex::new(Resource::empty())),
407                exporter,
408            }
409        }
410        fn resource(&self) -> Resource {
411            self.resource.lock().unwrap().clone()
412        }
413    }
414
415    #[test]
416    fn test_resource_handling_provider_processor_exporter() {
417        let assert_resource = |processor: &TestProcessorForResource,
418                               exporter: &TestExporterForResource,
419                               resource_key: &'static str,
420                               expect: Option<&'static str>| {
421            assert_eq!(
422                processor
423                    .resource()
424                    .get(&Key::from_static_str(resource_key))
425                    .map(|v| v.to_string()),
426                expect.map(|s| s.to_string())
427            );
428
429            assert_eq!(
430                exporter
431                    .resource()
432                    .get(&Key::from_static_str(resource_key))
433                    .map(|v| v.to_string()),
434                expect.map(|s| s.to_string())
435            );
436        };
437        let assert_telemetry_resource =
438            |processor: &TestProcessorForResource, exporter: &TestExporterForResource| {
439                assert_eq!(
440                    processor.resource().get(&TELEMETRY_SDK_LANGUAGE.into()),
441                    Some(Value::from("rust"))
442                );
443                assert_eq!(
444                    processor.resource().get(&TELEMETRY_SDK_NAME.into()),
445                    Some(Value::from("opentelemetry"))
446                );
447                assert_eq!(
448                    processor.resource().get(&TELEMETRY_SDK_VERSION.into()),
449                    Some(Value::from(env!("CARGO_PKG_VERSION")))
450                );
451                assert_eq!(
452                    exporter.resource().get(&TELEMETRY_SDK_LANGUAGE.into()),
453                    Some(Value::from("rust"))
454                );
455                assert_eq!(
456                    exporter.resource().get(&TELEMETRY_SDK_NAME.into()),
457                    Some(Value::from("opentelemetry"))
458                );
459                assert_eq!(
460                    exporter.resource().get(&TELEMETRY_SDK_VERSION.into()),
461                    Some(Value::from(env!("CARGO_PKG_VERSION")))
462                );
463            };
464
465        // If users didn't provide a resource and there isn't a env var set. Use default one.
466        temp_env::with_var_unset("OTEL_RESOURCE_ATTRIBUTES", || {
467            let exporter_with_resource = TestExporterForResource::new();
468            let processor_with_resource =
469                TestProcessorForResource::new(exporter_with_resource.clone());
470            let _ = super::SdkLoggerProvider::builder()
471                .with_log_processor(processor_with_resource.clone())
472                .build();
473            assert_resource(
474                &processor_with_resource,
475                &exporter_with_resource,
476                SERVICE_NAME,
477                Some("unknown_service"),
478            );
479            assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
480        });
481
482        // If user provided a resource, use that.
483        let exporter_with_resource = TestExporterForResource::new();
484        let processor_with_resource = TestProcessorForResource::new(exporter_with_resource.clone());
485        let _ = super::SdkLoggerProvider::builder()
486            .with_resource(
487                Resource::builder_empty()
488                    .with_service_name("test_service")
489                    .build(),
490            )
491            .with_log_processor(processor_with_resource.clone())
492            .build();
493        assert_resource(
494            &processor_with_resource,
495            &exporter_with_resource,
496            SERVICE_NAME,
497            Some("test_service"),
498        );
499        assert_eq!(processor_with_resource.resource().len(), 1);
500
501        // If `OTEL_RESOURCE_ATTRIBUTES` is set, read them automatically
502        temp_env::with_var(
503            "OTEL_RESOURCE_ATTRIBUTES",
504            Some("key1=value1, k2, k3=value2"),
505            || {
506                let exporter_with_resource = TestExporterForResource::new();
507                let processor_with_resource =
508                    TestProcessorForResource::new(exporter_with_resource.clone());
509                let _ = super::SdkLoggerProvider::builder()
510                    .with_log_processor(processor_with_resource.clone())
511                    .build();
512                assert_resource(
513                    &processor_with_resource,
514                    &exporter_with_resource,
515                    SERVICE_NAME,
516                    Some("unknown_service"),
517                );
518                assert_resource(
519                    &processor_with_resource,
520                    &exporter_with_resource,
521                    "key1",
522                    Some("value1"),
523                );
524                assert_resource(
525                    &processor_with_resource,
526                    &exporter_with_resource,
527                    "k3",
528                    Some("value2"),
529                );
530                assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
531                assert_eq!(processor_with_resource.resource().len(), 6);
532            },
533        );
534
535        // When `OTEL_RESOURCE_ATTRIBUTES` is set and also user provided config
536        temp_env::with_var(
537            "OTEL_RESOURCE_ATTRIBUTES",
538            Some("my-custom-key=env-val,k2=value2"),
539            || {
540                let exporter_with_resource = TestExporterForResource::new();
541                let processor_with_resource =
542                    TestProcessorForResource::new(exporter_with_resource.clone());
543                let _ = super::SdkLoggerProvider::builder()
544                    .with_resource(
545                        Resource::builder()
546                            .with_attributes([
547                                KeyValue::new("my-custom-key", "my-custom-value"),
548                                KeyValue::new("my-custom-key2", "my-custom-value2"),
549                            ])
550                            .build(),
551                    )
552                    .with_log_processor(processor_with_resource.clone())
553                    .build();
554                assert_resource(
555                    &processor_with_resource,
556                    &exporter_with_resource,
557                    SERVICE_NAME,
558                    Some("unknown_service"),
559                );
560                assert_resource(
561                    &processor_with_resource,
562                    &exporter_with_resource,
563                    "my-custom-key",
564                    Some("my-custom-value"),
565                );
566                assert_resource(
567                    &processor_with_resource,
568                    &exporter_with_resource,
569                    "my-custom-key2",
570                    Some("my-custom-value2"),
571                );
572                assert_resource(
573                    &processor_with_resource,
574                    &exporter_with_resource,
575                    "k2",
576                    Some("value2"),
577                );
578                assert_telemetry_resource(&processor_with_resource, &exporter_with_resource);
579                assert_eq!(processor_with_resource.resource().len(), 7);
580            },
581        );
582
583        // If user provided a resource, it takes priority during collision.
584        let exporter_with_resource = TestExporterForResource::new();
585        let processor_with_resource = TestProcessorForResource::new(exporter_with_resource);
586        let _ = super::SdkLoggerProvider::builder()
587            .with_resource(Resource::empty())
588            .with_log_processor(processor_with_resource.clone())
589            .build();
590        assert_eq!(processor_with_resource.resource().len(), 0);
591    }
592
593    #[test]
594    fn trace_context_test() {
595        let exporter = InMemoryLogExporter::default();
596
597        let logger_provider = SdkLoggerProvider::builder()
598            .with_simple_exporter(exporter.clone())
599            .build();
600
601        let logger = logger_provider.logger("test-logger");
602
603        let tracer_provider = SdkTracerProvider::builder().build();
604
605        let tracer = tracer_provider.tracer("test-tracer");
606
607        tracer.in_span("test-span", |cx| {
608            let ambient_ctxt = cx.span().span_context().clone();
609            let explicit_ctxt = TraceContext {
610                trace_id: TraceId::from_u128(13),
611                span_id: SpanId::from_u64(14),
612                trace_flags: None,
613            };
614
615            let mut ambient_ctxt_record = logger.create_log_record();
616            ambient_ctxt_record.set_body(AnyValue::String("ambient".into()));
617
618            let mut explicit_ctxt_record = logger.create_log_record();
619            explicit_ctxt_record.set_body(AnyValue::String("explicit".into()));
620            explicit_ctxt_record.set_trace_context(
621                explicit_ctxt.trace_id,
622                explicit_ctxt.span_id,
623                explicit_ctxt.trace_flags,
624            );
625
626            logger.emit(ambient_ctxt_record);
627            logger.emit(explicit_ctxt_record);
628
629            let emitted = exporter.get_emitted_logs().unwrap();
630
631            assert_eq!(
632                Some(AnyValue::String("ambient".into())),
633                emitted[0].record.body
634            );
635            assert_eq!(
636                ambient_ctxt.trace_id(),
637                emitted[0].record.trace_context.as_ref().unwrap().trace_id
638            );
639            assert_eq!(
640                ambient_ctxt.span_id(),
641                emitted[0].record.trace_context.as_ref().unwrap().span_id
642            );
643
644            assert_eq!(
645                Some(AnyValue::String("explicit".into())),
646                emitted[1].record.body
647            );
648            assert_eq!(
649                explicit_ctxt.trace_id,
650                emitted[1].record.trace_context.as_ref().unwrap().trace_id
651            );
652            assert_eq!(
653                explicit_ctxt.span_id,
654                emitted[1].record.trace_context.as_ref().unwrap().span_id
655            );
656        });
657    }
658
659    #[test]
660    fn shutdown_test() {
661        let counter = Arc::new(AtomicU64::new(0));
662        let logger_provider = SdkLoggerProvider::builder()
663            .with_log_processor(ShutdownTestLogProcessor::new(counter.clone()))
664            .build();
665
666        let logger1 = logger_provider.logger("test-logger1");
667        let logger2 = logger_provider.logger("test-logger2");
668        logger1.emit(logger1.create_log_record());
669        logger2.emit(logger1.create_log_record());
670
671        let logger3 = logger_provider.logger("test-logger3");
672        let handle = thread::spawn(move || {
673            logger3.emit(logger3.create_log_record());
674        });
675        handle.join().expect("thread panicked");
676
677        let _ = logger_provider.shutdown();
678        logger1.emit(logger1.create_log_record());
679
680        assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 3);
681    }
682
683    #[test]
684    fn shutdown_idempotent_test() {
685        let counter = Arc::new(AtomicU64::new(0));
686        let logger_provider = SdkLoggerProvider::builder()
687            .with_log_processor(ShutdownTestLogProcessor::new(counter.clone()))
688            .build();
689
690        let shutdown_res = logger_provider.shutdown();
691        assert!(shutdown_res.is_ok());
692
693        // Subsequent shutdowns should return an error.
694        let shutdown_res = logger_provider.shutdown();
695        assert!(shutdown_res.is_err());
696
697        // Subsequent shutdowns should return an error.
698        let shutdown_res = logger_provider.shutdown();
699        assert!(shutdown_res.is_err());
700    }
701
702    #[test]
703    fn global_shutdown_test() {
704        // cargo test global_shutdown_test --features=testing
705
706        // Arrange
707        let shutdown_called = Arc::new(Mutex::new(false));
708        let flush_called = Arc::new(Mutex::new(false));
709        let logger_provider = SdkLoggerProvider::builder()
710            .with_log_processor(LazyLogProcessor::new(
711                shutdown_called.clone(),
712                flush_called.clone(),
713            ))
714            .build();
715        //set_logger_provider(logger_provider);
716        let logger1 = logger_provider.logger("test-logger1");
717        let logger2 = logger_provider.logger("test-logger2");
718
719        // Acts
720        logger1.emit(logger1.create_log_record());
721        logger2.emit(logger1.create_log_record());
722
723        // explicitly calling shutdown on logger_provider. This will
724        // indeed do the shutdown, even if there are loggers still alive.
725        let _ = logger_provider.shutdown();
726
727        // Assert
728
729        // shutdown is called.
730        assert!(*shutdown_called.lock().unwrap());
731
732        // flush is never called by the sdk.
733        assert!(!*flush_called.lock().unwrap());
734    }
735
736    #[test]
737    fn drop_test_with_multiple_providers() {
738        let shutdown_called = Arc::new(Mutex::new(false));
739        let flush_called = Arc::new(Mutex::new(false));
740        {
741            // Create a shared LoggerProviderInner and use it across multiple providers
742            let shared_inner = Arc::new(LoggerProviderInner {
743                processors: vec![Box::new(LazyLogProcessor::new(
744                    shutdown_called.clone(),
745                    flush_called.clone(),
746                ))],
747                is_shutdown: AtomicBool::new(false),
748            });
749
750            {
751                let logger_provider1 = SdkLoggerProvider {
752                    inner: shared_inner.clone(),
753                };
754                let logger_provider2 = SdkLoggerProvider {
755                    inner: shared_inner.clone(),
756                };
757
758                let logger1 = logger_provider1.logger("test-logger1");
759                let logger2 = logger_provider2.logger("test-logger2");
760
761                logger1.emit(logger1.create_log_record());
762                logger2.emit(logger1.create_log_record());
763
764                // LoggerProviderInner should not be dropped yet, since both providers and `shared_inner`
765                // are still holding a reference.
766            }
767            // At this point, both `logger_provider1` and `logger_provider2` are dropped,
768            // but `shared_inner` still holds a reference, so `LoggerProviderInner` is NOT dropped yet.
769        }
770        // Verify shutdown was called during the drop of the shared LoggerProviderInner
771        assert!(*shutdown_called.lock().unwrap());
772        // Verify flush was not called during drop
773        assert!(!*flush_called.lock().unwrap());
774    }
775
776    #[test]
777    fn drop_after_shutdown_test_with_multiple_providers() {
778        let shutdown_called = Arc::new(Mutex::new(0)); // Count the number of times shutdown is called
779        let flush_called = Arc::new(Mutex::new(false));
780
781        // Create a shared LoggerProviderInner and use it across multiple providers
782        let shared_inner = Arc::new(LoggerProviderInner {
783            processors: vec![Box::new(CountingShutdownProcessor::new(
784                shutdown_called.clone(),
785                flush_called.clone(),
786            ))],
787            is_shutdown: AtomicBool::new(false),
788        });
789
790        // Create a scope to test behavior when providers are dropped
791        {
792            let logger_provider1 = SdkLoggerProvider {
793                inner: shared_inner.clone(),
794            };
795            let logger_provider2 = SdkLoggerProvider {
796                inner: shared_inner.clone(),
797            };
798
799            // Explicitly shut down the logger provider
800            let shutdown_result = logger_provider1.shutdown();
801            println!("---->Result: {:?}", shutdown_result);
802            assert!(shutdown_result.is_ok());
803
804            // Verify that shutdown was called exactly once
805            assert_eq!(*shutdown_called.lock().unwrap(), 1);
806
807            // LoggerProvider2 should observe the shutdown state but not trigger another shutdown
808            let shutdown_result2 = logger_provider2.shutdown();
809            assert!(shutdown_result2.is_err());
810
811            // Both logger providers will be dropped at the end of this scope
812        }
813
814        // Verify that shutdown was only called once, even after drop
815        assert_eq!(*shutdown_called.lock().unwrap(), 1);
816    }
817
818    #[test]
819    fn test_empty_logger_name() {
820        let exporter = InMemoryLogExporter::default();
821        let logger_provider = SdkLoggerProvider::builder()
822            .with_simple_exporter(exporter.clone())
823            .build();
824        let logger = logger_provider.logger("");
825        let mut record = logger.create_log_record();
826        record.set_body("Testing empty logger name".into());
827        logger.emit(record);
828
829        // Create a logger using a scope with an empty name
830        let scope = InstrumentationScope::builder("").build();
831        let scoped_logger = logger_provider.logger_with_scope(scope);
832        let mut scoped_record = scoped_logger.create_log_record();
833        scoped_record.set_body("Testing empty logger scope name".into());
834        scoped_logger.emit(scoped_record);
835
836        // Assert: Verify that the emitted logs are processed correctly
837        let mut emitted_logs = exporter.get_emitted_logs().unwrap();
838        assert_eq!(emitted_logs.len(), 2);
839        let log1 = emitted_logs.remove(0);
840        // Assert the first log
841        assert_eq!(
842            log1.record.body,
843            Some(AnyValue::String("Testing empty logger name".into()))
844        );
845        assert_eq!(log1.instrumentation.name(), "");
846
847        // Assert the second log created through the scope
848        let log2 = emitted_logs.remove(0);
849        assert_eq!(
850            log2.record.body,
851            Some(AnyValue::String("Testing empty logger scope name".into()))
852        );
853        assert_eq!(log1.instrumentation.name(), "");
854    }
855
856    #[test]
857    fn with_resource_multiple_calls_ensure_additive() {
858        let builder = SdkLoggerProvider::builder()
859            .with_resource(Resource::new(vec![KeyValue::new("key1", "value1")]))
860            .with_resource(Resource::new(vec![KeyValue::new("key2", "value2")]))
861            .with_resource(
862                Resource::builder_empty()
863                    .with_schema_url(vec![], "http://example.com")
864                    .build(),
865            )
866            .with_resource(Resource::new(vec![KeyValue::new("key3", "value3")]));
867
868        let resource = builder.resource.unwrap();
869
870        assert_eq!(
871            resource.get(&Key::from_static_str("key1")),
872            Some(Value::from("value1"))
873        );
874        assert_eq!(
875            resource.get(&Key::from_static_str("key2")),
876            Some(Value::from("value2"))
877        );
878        assert_eq!(
879            resource.get(&Key::from_static_str("key3")),
880            Some(Value::from("value3"))
881        );
882        assert_eq!(resource.schema_url(), Some("http://example.com"));
883    }
884
885    #[derive(Debug)]
886    pub(crate) struct LazyLogProcessor {
887        shutdown_called: Arc<Mutex<bool>>,
888        flush_called: Arc<Mutex<bool>>,
889    }
890
891    impl LazyLogProcessor {
892        pub(crate) fn new(
893            shutdown_called: Arc<Mutex<bool>>,
894            flush_called: Arc<Mutex<bool>>,
895        ) -> Self {
896            LazyLogProcessor {
897                shutdown_called,
898                flush_called,
899            }
900        }
901    }
902
903    impl LogProcessor for LazyLogProcessor {
904        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
905            // nothing to do.
906        }
907
908        fn force_flush(&self) -> OTelSdkResult {
909            *self.flush_called.lock().unwrap() = true;
910            Ok(())
911        }
912
913        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
914            *self.shutdown_called.lock().unwrap() = true;
915            Ok(())
916        }
917    }
918
919    #[derive(Debug)]
920    struct CountingShutdownProcessor {
921        shutdown_count: Arc<Mutex<i32>>,
922        flush_called: Arc<Mutex<bool>>,
923    }
924
925    impl CountingShutdownProcessor {
926        fn new(shutdown_count: Arc<Mutex<i32>>, flush_called: Arc<Mutex<bool>>) -> Self {
927            CountingShutdownProcessor {
928                shutdown_count,
929                flush_called,
930            }
931        }
932    }
933
934    impl LogProcessor for CountingShutdownProcessor {
935        fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {
936            // nothing to do
937        }
938
939        fn force_flush(&self) -> OTelSdkResult {
940            *self.flush_called.lock().unwrap() = true;
941            Ok(())
942        }
943
944        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
945            let mut count = self.shutdown_count.lock().unwrap();
946            *count += 1;
947            Ok(())
948        }
949    }
950}
opentelemetry_sdk/logs/logger_provider.rs

opentelemetry_sdk/logs/
logger_provider.rs
