fix: send propagator error to global error handler (#1640)

Author: TommyCpp

opentelemetry-jaeger-propagator/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## vNext
 
+### Changed
+
+- Propagation error will be reported to global error handler [#1640](https://github.com/open-telemetry/opentelemetry-rust/pull/1640)
+
 ## v0.1.0
 
 ### Added

opentelemetry-jaeger-propagator/src/propagator.rs

@@ -1,3 +1,4 @@
+use opentelemetry::propagation::PropagationError;
 use opentelemetry::{
     global::{self, Error},
     propagation::{text_map_propagator::FieldIter, Extractor, Injector, TextMapPropagator},
@@ -69,7 +70,7 @@ impl Propagator {
     }
 
     /// Extract span context from header value
-    fn extract_span_context(&self, extractor: &dyn Extractor) -> Result<SpanContext, ()> {
+    fn extract_span_context(&self, extractor: &dyn Extractor) -> Option<SpanContext> {
         let mut header_value = Cow::from(extractor.get(self.header_name).unwrap_or(""));
         // if there is no :, it means header_value could be encoded as url, try decode first
         if !header_value.contains(':') {
@@ -78,17 +79,31 @@ impl Propagator {
 
         let parts = header_value.split_terminator(':').collect::<Vec<&str>>();
         if parts.len() != 4 {
-            return Err(());
+            global::handle_error(Error::Propagation(PropagationError::extract(
+                "invalid jaeger header format",
+                "JaegerPropagator",
+            )));
+            return None;
         }
 
-        // extract trace id
-        let trace_id = self.extract_trace_id(parts[0])?;
-        let span_id = self.extract_span_id(parts[1])?;
-        // Ignore parent span id since it's deprecated.
-        let flags = self.extract_trace_flags(parts[3])?;
-        let state = self.extract_trace_state(extractor)?;
-
-        Ok(SpanContext::new(trace_id, span_id, flags, true, state))
+        match (
+            self.extract_trace_id(parts[0]),
+            self.extract_span_id(parts[1]),
+            // Ignore parent span id since it's deprecated.
+            self.extract_trace_flags(parts[3]),
+            self.extract_trace_state(extractor),
+        ) {
+            (Ok(trace_id), Ok(span_id), Ok(flags), Ok(state)) => {
+                Some(SpanContext::new(trace_id, span_id, flags, true, state))
+            }
+            _ => {
+                global::handle_error(Error::Propagation(PropagationError::extract(
+                    "invalid jaeger header format",
+                    "JaegerPropagator",
+                )));
+                None
+            }
+        }
     }
 
     /// Extract trace id from the header.
@@ -188,7 +203,7 @@ impl TextMapPropagator for Propagator {
     fn extract_with_context(&self, cx: &Context, extractor: &dyn Extractor) -> Context {
         self.extract_span_context(extractor)
             .map(|sc| cx.with_remote_span_context(sc))
-            .unwrap_or_else(|_| cx.clone())
+            .unwrap_or_else(|| cx.clone())
     }
 
     fn fields(&self) -> FieldIter<'_> {
@@ -430,7 +445,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Ok(SpanContext::new(
+            Some(SpanContext::new(
                 TraceId::from_hex("12345").unwrap(),
                 SpanId::from_hex("54321").unwrap(),
                 TRACE_FLAG_DEBUG | TraceFlags::SAMPLED,
@@ -447,7 +462,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Ok(SpanContext::new(
+            Some(SpanContext::new(
                 TraceId::from_hex("12345").unwrap(),
                 SpanId::from_hex("54321").unwrap(),
                 TRACE_FLAG_DEBUG | TraceFlags::SAMPLED,
@@ -463,7 +478,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Err(())
+            None,
         );
 
         map.clear();
@@ -473,7 +488,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Err(())
+            None,
         );
 
         map.clear();
@@ -483,7 +498,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Err(())
+            None,
         );
 
         map.clear();
@@ -493,7 +508,7 @@ mod tests {
         );
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Err(())
+            None,
         );
 
         map.clear();
@@ -506,7 +521,7 @@ mod tests {
         map.set(&too_long_baggage_key, "baggage_value".to_owned());
         assert_eq!(
             propagator_with_custom_header.extract_span_context(&map),
-            Err(())
+            None,
         );
     }
 

opentelemetry-sdk/CHANGELOG.md

@@ -12,6 +12,7 @@
 - **Breaking** [#1624](https://github.com/open-telemetry/opentelemetry-rust/pull/1624) Remove `OsResourceDetector` and
   `ProcessResourceDetector` resource detectors, use the
   [`opentelemetry-resource-detector`](https://crates.io/crates/opentelemetry-resource-detectors) instead.
+- Baggage propagation error will be reported to global error handler [#1640](https://github.com/open-telemetry/opentelemetry-rust/pull/1640)
 
 ## v0.22.1
 

opentelemetry-sdk/src/propagation/baggage.rs

@@ -1,6 +1,8 @@
 use once_cell::sync::Lazy;
+use opentelemetry::propagation::PropagationError;
 use opentelemetry::{
     baggage::{BaggageExt, KeyValueMetadata},
+    global,
     propagation::{text_map_propagator::FieldIter, Extractor, Injector, TextMapPropagator},
     Context,
 };
@@ -99,30 +101,44 @@ impl TextMapPropagator for BaggagePropagator {
                 {
                     let mut iter = name_and_value.split('=');
                     if let (Some(name), Some(value)) = (iter.next(), iter.next()) {
-                        let name = percent_decode_str(name).decode_utf8().map_err(|_| ())?;
-                        let value = percent_decode_str(value).decode_utf8().map_err(|_| ())?;
+                        let decode_name = percent_decode_str(name).decode_utf8();
+                        let decode_value = percent_decode_str(value).decode_utf8();
 
-                        // Here we don't store the first ; into baggage since it should be treated
-                        // as separator rather part of metadata
-                        let decoded_props = props
-                            .iter()
-                            .flat_map(|prop| percent_decode_str(prop).decode_utf8())
-                            .map(|prop| prop.trim().to_string())
-                            .collect::<Vec<String>>()
-                            .join(";"); // join with ; because we deleted all ; when calling split above
+                        if let (Ok(name), Ok(value)) = (decode_name, decode_value) {
+                            // Here we don't store the first ; into baggage since it should be treated
+                            // as separator rather part of metadata
+                            let decoded_props = props
+                                .iter()
+                                .flat_map(|prop| percent_decode_str(prop).decode_utf8())
+                                .map(|prop| prop.trim().to_string())
+                                .collect::<Vec<String>>()
+                                .join(";"); // join with ; because we deleted all ; when calling split above
 
-                        Ok(KeyValueMetadata::new(
-                            name.trim().to_owned(),
-                            value.trim().to_string(),
-                            decoded_props.as_str(),
-                        ))
+                            Some(KeyValueMetadata::new(
+                                name.trim().to_owned(),
+                                value.trim().to_string(),
+                                decoded_props.as_str(),
+                            ))
+                        } else {
+                            global::handle_error(PropagationError::extract(
+                                "invalid UTF8 string in key values",
+                                "BaggagePropagator",
+                            ));
+                            None
+                        }
                     } else {
-                        // Invalid name-value format
-                        Err(())
+                        global::handle_error(PropagationError::extract(
+                            "invalid baggage key-value format",
+                            "BaggagePropagator",
+                        ));
+                        None
                     }
                 } else {
-                    // Invalid baggage value format
-                    Err(())
+                    global::handle_error(PropagationError::extract(
+                        "invalid baggage format",
+                        "BaggagePropagator",
+                    ));
+                    None
                 }
             });
             cx.with_baggage(baggage)
@@ -172,7 +188,7 @@ mod tests {
              vec![
                  (Key::new("key1"), (Value::from("value1"), BaggageMetadata::from("property1;property2"))),
                  (Key::new("key2"), (Value::from("value2"), BaggageMetadata::default())),
-                 (Key::new("key3"), (Value::from("value3"), BaggageMetadata::from("propertyKey=propertyValue")))
+                 (Key::new("key3"), (Value::from("value3"), BaggageMetadata::from("propertyKey=propertyValue"))),
              ].into_iter().collect()),
         ]
     }
@@ -220,12 +236,12 @@ mod tests {
                 vec![
                     KeyValueMetadata::new("key1", "val1", "prop1"),
                     KeyValue::new("key2", "val2").into(),
-                    KeyValueMetadata::new("key3", "val3", "anykey=anyvalue")
+                    KeyValueMetadata::new("key3", "val3", "anykey=anyvalue"),
                 ],
                 vec![
                     "key1=val1;prop1",
                     "key2=val2",
-                    "key3=val3;anykey=anyvalue"
+                    "key3=val3;anykey=anyvalue",
                 ],
             )
         ]

opentelemetry/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Added
 
 - [#1623](https://github.com/open-telemetry/opentelemetry-rust/pull/1623) Add global::meter_provider_shutdown
+- [#1640](https://github.com/open-telemetry/opentelemetry-rust/pull/1640) Add `PropagationError`
 
 ### Removed
 

opentelemetry/src/global/error_handler.rs

@@ -5,6 +5,7 @@ use std::sync::RwLock;
 use crate::logs::LogError;
 #[cfg(feature = "metrics")]
 use crate::metrics::MetricsError;
+use crate::propagation::PropagationError;
 #[cfg(feature = "trace")]
 use crate::trace::TraceError;
 use once_cell::sync::Lazy;
@@ -32,6 +33,10 @@ pub enum Error {
     /// Failed to export logs.
     Log(#[from] LogError),
 
+    #[error(transparent)]
+    /// Error happens when injecting and extracting information using propagators.
+    Propagation(#[from] PropagationError),
+
     #[error("{0}")]
     /// Other types of failures not covered by the variants above.
     Other(String),
@@ -61,6 +66,9 @@ pub fn handle_error<T: Into<Error>>(err: T) {
             #[cfg(feature = "logs")]
             #[cfg_attr(docsrs, doc(cfg(feature = "logs")))]
             Error::Log(err) => eprintln!("OpenTelemetry log error occurred. {}", err),
+            Error::Propagation(err) => {
+                eprintln!("OpenTelemetry propagation error occurred. {}", err)
+            }
             Error::Other(err_msg) => eprintln!("OpenTelemetry error occurred. {}", err_msg),
         },
     }

opentelemetry/src/propagation/mod.rs

@@ -20,6 +20,7 @@
 //! [`Context`]: crate::Context
 
 use std::collections::HashMap;
+use thiserror::Error;
 
 pub mod composite;
 pub mod text_map_propagator;
@@ -61,6 +62,37 @@ impl<S: std::hash::BuildHasher> Extractor for HashMap<String, String, S> {
     }
 }
 
+/// Error when extracting or injecting context data(i.e propagating) across application boundaries.
+#[derive(Error, Debug)]
+#[error("Cannot {} from {}, {}", ops, message, propagator_name)]
+pub struct PropagationError {
+    message: &'static str,
+    // which propagator does this error comes from
+    propagator_name: &'static str,
+    // are we extracting or injecting information across application boundaries
+    ops: &'static str,
+}
+
+impl PropagationError {
+    /// Error happens when extracting information
+    pub fn extract(message: &'static str, propagator_name: &'static str) -> Self {
+        PropagationError {
+            message,
+            propagator_name,
+            ops: "extract",
+        }
+    }
+
+    /// Error happens when extracting information
+    pub fn inject(message: &'static str, propagator_name: &'static str) -> Self {
+        PropagationError {
+            message,
+            propagator_name,
+            ops: "inject",
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;