Improve internal opentelemetry logging (open-telemetry#2128)

Co-authored-by: Zhongyang Wu <[email protected]>
Co-authored-by: Utkarsh Umesan Pillai <[email protected]>

Author: 3 people

examples/self-diagnostics/Cargo.toml

@@ -7,13 +7,13 @@ publish = false
 
 [dependencies]
 opentelemetry = { path = "../../opentelemetry" }
-opentelemetry_sdk = { path = "../../opentelemetry-sdk", features = ["rt-tokio", "experimental-internal-logs"]}
+opentelemetry_sdk = { path = "../../opentelemetry-sdk", features = ["rt-tokio"]}
 opentelemetry-stdout = { path = "../../opentelemetry-stdout"}
 opentelemetry-appender-tracing = { path = "../../opentelemetry-appender-tracing"}
 tokio = { workspace = true, features = ["full"] }
 tracing = { workspace = true, features = ["std"]}
 tracing-core = { workspace = true }
 tracing-subscriber = { version = "0.3.18", features = ["env-filter","registry", "std"]}
-opentelemetry-otlp = { path = "../../opentelemetry-otlp", features = ["http-proto", "reqwest-client", "logs", "experimental-internal-logs"] }
+opentelemetry-otlp = { path = "../../opentelemetry-otlp", features = ["http-proto", "reqwest-client", "logs"] }
 once_cell ={ version = "1.19.0"}
 ctrlc = "3.4"

opentelemetry-appender-tracing/Cargo.toml

@@ -22,6 +22,7 @@ tracing-subscriber = { workspace = true, features = ["registry", "std"] }
 log = { workspace = true }
 opentelemetry-stdout = { path = "../opentelemetry-stdout", features = ["logs"] }
 opentelemetry_sdk = { path = "../opentelemetry-sdk", features = ["logs", "testing"]  }
+tracing-subscriber = { workspace = true, features = ["registry", "std", "env-filter"] }
 tracing-log = "0.2"
 async-trait = { workspace = true }
 criterion = { workspace = true }

opentelemetry-appender-tracing/src/layer.rs

@@ -217,14 +217,26 @@ mod tests {
     use opentelemetry_sdk::trace;
     use opentelemetry_sdk::trace::{Sampler, TracerProvider};
     use tracing::error;
-    use tracing_subscriber::layer::SubscriberExt;
+    use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
+    use tracing_subscriber::Layer;
 
     pub fn attributes_contains(log_record: &LogRecord, key: &Key, value: &AnyValue) -> bool {
         log_record
             .attributes_iter()
             .any(|(k, v)| k == key && v == value)
     }
 
+    fn create_tracing_subscriber(
+        _exporter: InMemoryLogsExporter,
+        logger_provider: &LoggerProvider,
+    ) -> impl tracing::Subscriber {
+        let level_filter = tracing_subscriber::filter::LevelFilter::WARN; // Capture WARN and ERROR levels
+        let layer =
+            layer::OpenTelemetryTracingBridge::new(logger_provider).with_filter(level_filter); // No filter based on target, only based on log level
+
+        tracing_subscriber::registry().with(layer)
+    }
+
     // cargo test --features=testing
     #[test]
     fn tracing_appender_standalone() {
@@ -234,8 +246,7 @@ mod tests {
             .with_simple_exporter(exporter.clone())
             .build();
 
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
+        let subscriber = create_tracing_subscriber(exporter.clone(), &logger_provider);
 
         // avoiding setting tracing subscriber as global as that does not
         // play well with unit tests.
@@ -315,8 +326,7 @@ mod tests {
             .with_simple_exporter(exporter.clone())
             .build();
 
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
+        let subscriber = create_tracing_subscriber(exporter.clone(), &logger_provider);
 
         // avoiding setting tracing subscriber as global as that does not
         // play well with unit tests.
@@ -427,16 +437,15 @@ mod tests {
             .with_simple_exporter(exporter.clone())
             .build();
 
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
+        let subscriber = create_tracing_subscriber(exporter.clone(), &logger_provider);
 
         // avoiding setting tracing subscriber as global as that does not
         // play well with unit tests.
         let _guard = tracing::subscriber::set_default(subscriber);
         drop(tracing_log::LogTracer::init());
 
         // Act
-        log::error!("log from log crate");
+        log::error!(target: "my-system", "log from log crate");
         logger_provider.force_flush();
 
         // Assert TODO: move to helper methods
@@ -493,8 +502,7 @@ mod tests {
             .with_simple_exporter(exporter.clone())
             .build();
 
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        let subscriber = tracing_subscriber::registry().with(layer);
+        let subscriber = create_tracing_subscriber(exporter.clone(), &logger_provider);
 
         // avoiding setting tracing subscriber as global as that does not
         // play well with unit tests.
@@ -513,7 +521,7 @@ mod tests {
             let span_id = cx.span().span_context().span_id();
 
             // logging is done inside span context.
-            log::error!("log from log crate");
+            log::error!(target: "my-system", "log from log crate");
             (trace_id, span_id)
         });
 

opentelemetry-otlp/Cargo.toml

@@ -58,12 +58,12 @@ trace = ["opentelemetry/trace", "opentelemetry_sdk/trace", "opentelemetry-proto/
 metrics = ["opentelemetry/metrics", "opentelemetry_sdk/metrics", "opentelemetry-proto/metrics"]
 logs = ["opentelemetry/logs", "opentelemetry_sdk/logs", "opentelemetry-proto/logs"]
 populate-logs-event-name = ["opentelemetry-proto/populate-logs-event-name"]
-experimental-internal-logs = ["tracing"]
+internal-logs = ["tracing"]
 
 # add ons
 serialize = ["serde", "serde_json"]
 
-default = ["grpc-tonic", "trace", "metrics", "logs"]
+default = ["grpc-tonic", "trace", "metrics", "logs", "internal-logs"]
 
 # grpc using tonic
 grpc-tonic = ["tonic", "prost", "http", "tokio", "opentelemetry-proto/gen-tonic"]

opentelemetry-sdk/Cargo.toml

@@ -42,7 +42,7 @@ temp-env = { workspace = true }
 pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 
 [features]
-default = ["trace", "metrics", "logs"]
+default = ["trace", "metrics", "logs", "internal-logs"]
 trace = ["opentelemetry/trace", "rand", "async-trait", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
 logs = ["opentelemetry/logs", "async-trait", "serde_json"]
@@ -52,7 +52,7 @@ testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std",
 rt-tokio = ["tokio", "tokio-stream"]
 rt-tokio-current-thread = ["tokio", "tokio-stream"]
 rt-async-std = ["async-std"]
-experimental-internal-logs = ["tracing"]
+internal-logs = ["tracing"]
 
 [[bench]]
 name = "context"

opentelemetry-sdk/src/logs/log_emitter.rs

@@ -1,5 +1,6 @@
 use super::{BatchLogProcessor, LogProcessor, LogRecord, SimpleLogProcessor, TraceContext};
 use crate::{export::logs::LogExporter, runtime::RuntimeChannel, Resource};
+use opentelemetry::otel_warn;
 use opentelemetry::{
     global,
     logs::{LogError, LogResult},
@@ -49,7 +50,6 @@ impl opentelemetry::logs::LoggerProvider for LoggerProvider {
         attributes: Option<Vec<opentelemetry::KeyValue>>,
     ) -> Logger {
         let name = name.into();
-
         let component_name = if name.is_empty() {
             Cow::Borrowed(DEFAULT_COMPONENT_NAME)
         } else {
@@ -114,6 +114,10 @@ impl LoggerProvider {
             let mut errs = vec![];
             for processor in &self.inner.processors {
                 if let Err(err) = processor.shutdown() {
+                    otel_warn!(
+                        name: "logger_provider_shutdown_error",
+                        error = format!("{:?}", err)
+                    );
                     errs.push(err);
                 }
             }
@@ -124,6 +128,9 @@ impl LoggerProvider {
                 Err(LogError::Other(format!("{errs:?}").into()))
             }
         } else {
+            otel_warn!(
+                name: "logger_provider_already_shutdown"
+            );
             Err(LogError::Other("logger provider already shut down".into()))
         }
     }

opentelemetry-sdk/src/logs/log_processor.rs

@@ -14,8 +14,9 @@ use opentelemetry::logs::Severity;
 use opentelemetry::{
     global,
     logs::{LogError, LogResult},
-    InstrumentationLibrary,
+    otel_error, otel_warn, InstrumentationLibrary,
 };
+
 use std::sync::atomic::AtomicBool;
 use std::{cmp::min, env, sync::Mutex};
 use std::{
@@ -98,16 +99,12 @@ impl LogProcessor for SimpleLogProcessor {
     fn emit(&self, record: &mut LogRecord, instrumentation: &InstrumentationLibrary) {
         // noop after shutdown
         if self.is_shutdown.load(std::sync::atomic::Ordering::Relaxed) {
+            otel_warn!(
+                name: "simple_log_processor_emit_after_shutdown"
+            );
             return;
         }
 
-        #[cfg(feature = "experimental-internal-logs")]
-        tracing::debug!(
-            name: "simple_log_processor_emit",
-            target: "opentelemetry-sdk",
-            event_name = record.event_name
-        );
-
         let result = self
             .exporter
             .lock()
@@ -117,6 +114,10 @@ impl LogProcessor for SimpleLogProcessor {
                 futures_executor::block_on(exporter.export(LogBatch::new(log_tuple)))
             });
         if let Err(err) = result {
+            otel_error!(
+                name: "simple_log_processor_emit_error",
+                error = format!("{:?}", err)
+            );
             global::handle_error(err);
         }
     }
@@ -132,6 +133,9 @@ impl LogProcessor for SimpleLogProcessor {
             exporter.shutdown();
             Ok(())
         } else {
+            otel_error!(
+                name: "simple_log_processor_shutdown_error"
+            );
             Err(LogError::Other(
                 "simple logprocessor mutex poison during shutdown".into(),
             ))
@@ -167,6 +171,10 @@ impl<R: RuntimeChannel> LogProcessor for BatchLogProcessor<R> {
         )));
 
         if let Err(err) = result {
+            otel_error!(
+                name: "batch_log_processor_emit_error",
+                error = format!("{:?}", err)
+            );
             global::handle_error(LogError::Other(err.into()));
         }
     }
@@ -224,13 +232,6 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                     // Log has finished, add to buffer of pending logs.
                     BatchMessage::ExportLog(log) => {
                         logs.push(log);
-                        #[cfg(feature = "experimental-internal-logs")]
-                        tracing::debug!(
-                            name: "batch_log_processor_record_count",
-                            target: "opentelemetry-sdk",
-                            current_batch_size = logs.len()
-                        );
-
                         if logs.len() == config.max_export_batch_size {
                             let result = export_with_timeout(
                                 config.max_export_timeout,
@@ -241,6 +242,10 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                             .await;
 
                             if let Err(err) = result {
+                                otel_error!(
+                                    name: "batch_log_processor_export_error",
+                                    error = format!("{:?}", err)
+                                );
                                 global::handle_error(err);
                             }
                         }
@@ -261,8 +266,18 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                                     "failed to send flush result: {:?}",
                                     result
                                 )));
+                                otel_error!(
+                                    name: "batch_log_processor_flush_error",
+                                    error = format!("{:?}", result),
+                                    message = "Failed to send flush result"
+                                );
                             }
                         } else if let Err(err) = result {
+                            otel_error!(
+                                name: "batch_log_processor_flush_error",
+                                error = format!("{:?}", err),
+                                message = "Flush failed"
+                            );
                             global::handle_error(err);
                         }
                     }
@@ -279,6 +294,11 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                         exporter.shutdown();
 
                         if let Err(result) = ch.send(result) {
+                            otel_error!(
+                                name: "batch_log_processor_shutdown_error",
+                                error = format!("{:?}", result),
+                                message = "Failed to send shutdown result"
+                            );
                             global::handle_error(LogError::from(format!(
                                 "failed to send batch processor shutdown result: {:?}",
                                 result
@@ -295,7 +315,6 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
                 }
             }
         }));
-
         // Return batch processor with link to worker
         BatchLogProcessor { message_sender }
     }
@@ -338,7 +357,13 @@ where
     pin_mut!(timeout);
     match future::select(export, timeout).await {
         Either::Left((export_res, _)) => export_res,
-        Either::Right((_, _)) => ExportResult::Err(LogError::ExportTimedOut(time_out)),
+        Either::Right((_, _)) => {
+            otel_error!(
+                name: "export_with_timeout_timeout",
+                timeout_duration = time_out.as_millis()
+            );
+            ExportResult::Err(LogError::ExportTimedOut(time_out))
+        }
     }
 }
 

opentelemetry-sdk/src/metrics/internal/mod.rs

@@ -17,7 +17,7 @@ pub(crate) use aggregate::{AggregateBuilder, ComputeAggregation, Measure};
 pub(crate) use exponential_histogram::{EXPO_MAX_SCALE, EXPO_MIN_SCALE};
 use once_cell::sync::Lazy;
 use opentelemetry::metrics::MetricsError;
-use opentelemetry::{global, KeyValue};
+use opentelemetry::{global, otel_warn, KeyValue};
 
 use crate::metrics::AttributeSet;
 
@@ -147,6 +147,9 @@ impl<AU: AtomicallyUpdate<T>, T: Number, O: Operation> ValueMap<AU, T, O> {
             O::update_tracker(&new_tracker, measurement, index);
             trackers.insert(STREAM_OVERFLOW_ATTRIBUTES.clone(), Arc::new(new_tracker));
             global::handle_error(MetricsError::Other("Warning: Maximum data points for metric stream exceeded. Entry added to overflow. Subsequent overflows to same metric until next collect will not be logged.".into()));
+            otel_warn!( name: "ValueMap.measure",
+                message = "Maximum data points for metric stream exceeded. Entry added to overflow. Subsequent overflows to same metric until next collect will not be logged."
+            );
         }
     }
 }