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

opentelemetry_sdk/logs/
logger_provider.rs
