Merge branch 'main' into value-map-collect

Author: cijothomas

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

@@ -13,7 +13,7 @@ struct HistogramTracker<T> {
     buckets: Mutex<Buckets<T>>,
 }
 
-impl<T> Aggregator<T> for HistogramTracker<T>
+impl<T> Aggregator for HistogramTracker<T>
 where
     T: Number,
 {
@@ -85,7 +85,7 @@ impl<T: Number> Buckets<T> {
 /// Summarizes a set of measurements as a histogram with explicitly defined
 /// buckets.
 pub(crate) struct Histogram<T: Number> {
-    value_map: ValueMap<T, HistogramTracker<T>>,
+    value_map: ValueMap<HistogramTracker<T>>,
     bounds: Vec<f64>,
     record_min_max: bool,
     record_sum: bool,

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

@@ -13,7 +13,7 @@ where
     pub(crate) value: T::AtomicTracker,
 }
 
-impl<T> Aggregator<T> for Assign<T>
+impl<T> Aggregator for Assign<T>
 where
     T: Number,
 {
@@ -39,7 +39,7 @@ where
 
 /// Summarizes a set of measurements as the last one made.
 pub(crate) struct LastValue<T: Number> {
-    value_map: ValueMap<T, Assign<T>>,
+    value_map: ValueMap<Assign<T>>,
     start: Mutex<SystemTime>,
 }
 

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

@@ -23,10 +23,7 @@ use crate::metrics::AttributeSet;
 pub(crate) static STREAM_OVERFLOW_ATTRIBUTES: Lazy<Vec<KeyValue>> =
     Lazy::new(|| vec![KeyValue::new("otel.metric.overflow", "true")]);
 
-pub(crate) trait Aggregator<T>
-where
-    T: Number,
-{
+pub(crate) trait Aggregator {
     /// A static configuration that is needed in order to initialize aggregator.
     /// E.g. bucket_size at creation time .
     type InitConfig;
@@ -50,10 +47,9 @@ where
 ///
 /// This structure is parametrized by an `Operation` that indicates how
 /// updates to the underlying value trackers should be performed.
-pub(crate) struct ValueMap<T, A>
+pub(crate) struct ValueMap<A>
 where
-    T: Number,
-    A: Aggregator<T>,
+    A: Aggregator,
 {
     /// Trackers store the values associated with different attribute sets.
     trackers: RwLock<HashMap<Vec<KeyValue>, Arc<A>>>,
@@ -67,10 +63,9 @@ where
     config: A::InitConfig,
 }
 
-impl<T, A> ValueMap<T, A>
+impl<A> ValueMap<A>
 where
-    T: Number,
-    A: Aggregator<T>,
+    A: Aggregator,
 {
     fn new(config: A::InitConfig) -> Self {
         ValueMap {

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

@@ -7,7 +7,7 @@ use std::{collections::HashMap, mem::replace, ops::DerefMut, sync::Mutex, time::
 
 /// Summarizes a set of pre-computed sums as their arithmetic sum.
 pub(crate) struct PrecomputedSum<T: Number> {
-    value_map: ValueMap<T, Assign<T>>,
+    value_map: ValueMap<Assign<T>>,
     monotonic: bool,
     start: Mutex<SystemTime>,
     reported: Mutex<HashMap<Vec<KeyValue>, T>>,

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

@@ -16,7 +16,7 @@ where
     value: T::AtomicTracker,
 }
 
-impl<T> Aggregator<T> for Increment<T>
+impl<T> Aggregator for Increment<T>
 where
     T: Number,
 {
@@ -42,7 +42,7 @@ where
 
 /// Summarizes a set of measurements made as their arithmetic sum.
 pub(crate) struct Sum<T: Number> {
-    value_map: ValueMap<T, Increment<T>>,
+    value_map: ValueMap<Increment<T>>,
     monotonic: bool,
     start: Mutex<SystemTime>,
 }

opentelemetry-sdk/src/trace/config.rs

@@ -4,7 +4,7 @@
 //! can be set for the default OpenTelemetry limits and Sampler.
 use crate::trace::{span_limit::SpanLimits, IdGenerator, RandomIdGenerator, Sampler, ShouldSample};
 use crate::Resource;
-use opentelemetry::global::{handle_error, Error};
+use opentelemetry::otel_warn;
 use std::borrow::Cow;
 use std::env;
 use std::str::FromStr;
@@ -125,13 +125,14 @@ impl Default for Config {
                 "always_on" => Box::new(Sampler::AlwaysOn),
                 "always_off" => Box::new(Sampler::AlwaysOff),
                 "traceidratio" => {
-                    let ratio = sampler_arg.and_then(|r| r.parse::<f64>().ok());
+                    let ratio = sampler_arg.as_ref().and_then(|r| r.parse::<f64>().ok());
                     if let Some(r) = ratio {
                         Box::new(Sampler::TraceIdRatioBased(r))
                     } else {
-                        handle_error(
-                            Error::Other(String::from(
-                                "Missing or invalid OTEL_TRACES_SAMPLER_ARG value. Falling back to default: 1.0"))
+                        otel_warn!(
+                            name: "TracerProvider.Config.InvalidSamplerArgument",
+                            message = "OTEL_TRACES_SAMPLER is set to 'traceidratio' but OTEL_TRACES_SAMPLER_ARG environment variable is missing or invalid. OTEL_TRACES_SAMPLER_ARG must be a valid float between 0.0 and 1.0 representing the desired sampling probability (0.0 = no traces sampled, 1.0 = all traces sampled, 0.5 = 50% of traces sampled). Falling back to default ratio: 1.0 (100% sampling)",
+                            otel_traces_sampler_arg = format!("{:?}", sampler_arg)
                         );
                         Box::new(Sampler::TraceIdRatioBased(1.0))
                     }
@@ -143,43 +144,51 @@ impl Default for Config {
                     Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOff)))
                 }
                 "parentbased_traceidratio" => {
-                    let ratio = sampler_arg.and_then(|r| r.parse::<f64>().ok());
+                    let ratio = sampler_arg.as_ref().and_then(|r| r.parse::<f64>().ok());
                     if let Some(r) = ratio {
                         Box::new(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
                             r,
                         ))))
                     } else {
-                        handle_error(
-                            Error::Other(String::from(
-                            "Missing or invalid OTEL_TRACES_SAMPLER_ARG value. Falling back to default: 1.0"
-                        )));
+                        otel_warn!(
+                            name: "TracerProvider.Config.InvalidSamplerArgument",
+                            message = "OTEL_TRACES_SAMPLER is set to 'parentbased_traceidratio' but OTEL_TRACES_SAMPLER_ARG environment variable is missing or invalid. OTEL_TRACES_SAMPLER_ARG must be a valid float between 0.0 and 1.0 representing the desired sampling probability (0.0 = no traces sampled, 1.0 = all traces sampled, 0.5 = 50% of traces sampled). Falling back to default ratio: 1.0 (100% sampling)",
+                            otel_traces_sampler_arg = format!("{:?}", sampler_arg)
+                        );
                         Box::new(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
                             1.0,
                         ))))
                     }
                 }
                 "parentbased_jaeger_remote" => {
-                    handle_error(
-                        Error::Other(String::from(
-                        "Unimplemented parentbased_jaeger_remote sampler. Falling back to default: parentbased_always_on"
-                    )));
+                    otel_warn!(
+                        name: "TracerProvider.Config.UnsupportedSampler",
+                        message = "OTEL_TRACES_SAMPLER is set to 'parentbased_jaeger_remote' which is not implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
+                    );
                     Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
                 }
                 "jaeger_remote" => {
-                    handle_error(
-                        Error::Other(String::from("Unimplemented jaeger_remote sampler. Falling back to default: parentbased_always_on")));
+                    otel_warn!(
+                        name: "TracerProvider.Config.UnsupportedSampler",
+                        message = "OTEL_TRACES_SAMPLER is set to 'jaeger_remote' which is implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
+                    );
                     Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
                 }
                 "xray" => {
-                    handle_error(
-                        Error::Other(String::from("Unimplemented xray sampler. Falling back to default: parentbased_always_on")));
+                    otel_warn!(
+                        name: "TracerProvider.Config.UnsupportedSampler",
+                        message = "OTEL_TRACES_SAMPLER is set to 'xray'. AWS X-Ray sampler is not implemented in this SDK version. Using fallback sampler: ParentBased(AlwaysOn). Configure an alternative sampler using OTEL_TRACES_SAMPLER"
+                    );
                     Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
                 }
                 s => {
-                    handle_error(
-                        Error::Other(format!("Unrecognised OTEL_TRACES_SAMPLER value: {}. Falling back to default: parentbased_always_on",
-                        s
-                    )));
+                    otel_warn!(
+                        name: "TracerProvider.Config.InvalidSamplerType",
+                        message = format!(
+                            "Unrecognized sampler type '{}' in OTEL_TRACES_SAMPLER environment variable. Valid values are: always_on, always_off, traceidratio, parentbased_always_on, parentbased_always_off, parentbased_traceidratio. Using fallback sampler: ParentBased(AlwaysOn)",
+                            s
+                        ),
+                    );
                     Box::new(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)))
                 }
             }

opentelemetry-sdk/src/trace/span_processor.rs

@@ -45,7 +45,7 @@ use futures_util::{
     stream::{self, FusedStream, FuturesUnordered},
     StreamExt as _,
 };
-use opentelemetry::global;
+use opentelemetry::{otel_debug, otel_error};
 use opentelemetry::{
     trace::{TraceError, TraceResult},
     Context,
@@ -134,7 +134,11 @@ impl SpanProcessor for SimpleSpanProcessor {
             .and_then(|mut exporter| futures_executor::block_on(exporter.export(vec![span])));
 
         if let Err(err) = result {
-            global::handle_error(err);
+            // TODO: check error type, and log `error` only if the error is user-actiobable, else log `debug`
+            otel_debug!(
+                name: "SimpleProcessor.OnEnd.Error",
+                reason = format!("{:?}", err)
+            );
         }
     }
 
@@ -246,7 +250,10 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
         let result = self.message_sender.try_send(BatchMessage::ExportSpan(span));
 
         if let Err(err) = result {
-            global::handle_error(TraceError::Other(err.into()));
+            otel_debug!(
+                name: "BatchSpanProcessor.OnEnd.ExportQueueingFailed",
+                reason = format!("{:?}", TraceError::Other(err.into()))
+            );
         }
     }
 
@@ -313,14 +320,22 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
             let result = export_task.await;
 
             if let Some(channel) = res_channel {
+                // If a response channel is provided, attempt to send the export result through it.
                 if let Err(result) = channel.send(result) {
-                    global::handle_error(TraceError::from(format!(
-                        "failed to send flush result: {:?}",
-                        result
-                    )));
+                    otel_debug!(
+                        name: "BatchSpanProcessor.Flush.SendResultError",
+                        reason = format!("{:?}", result)
+                    );
                 }
             } else if let Err(err) = result {
-                global::handle_error(err);
+                // If no channel is provided and the export operation encountered an error,
+                // log the error directly here.
+                // TODO: Consider returning the status instead of logging it.
+                otel_error!(
+                    name: "BatchSpanProcessor.Flush.ExportError",
+                    reason = format!("{:?}", err),
+                    message = "Failed during the export process"
+                );
             }
 
             Ok(())
@@ -354,7 +369,10 @@ impl<R: RuntimeChannel> BatchSpanProcessorInternal<R> {
                     let export_task = self.export();
                     let task = async move {
                         if let Err(err) = export_task.await {
-                            global::handle_error(err);
+                            otel_error!(
+                                name: "BatchSpanProcessor.Export.Error",
+                                reason = format!("{}", err)
+                            );
                         }
 
                         Ok(())