Change the signature of MeasurementProcessor to return Option

Author: martintmk

examples/metrics-advanced/src/main.rs

@@ -1,4 +1,3 @@
-use std::borrow::Cow;
 use opentelemetry::{global, Context};
 use opentelemetry::Key;
 use opentelemetry::KeyValue;
@@ -179,16 +178,11 @@ impl UserType {
 struct UserTypeMeasurementProcessor;
 
 impl MeasurementProcessor for UserTypeMeasurementProcessor {
-    fn process<'a>(&self, attributes: Cow<'a, [KeyValue]>) -> Cow<'a, [KeyValue]> {
-        match Context::current().get::<UserType>() {
-            Some(user_type) => {
-                let mut attrs = attributes.to_vec();
-                attrs.push(KeyValue::new("user_type", user_type.as_str()));
-                Cow::Owned(attrs)
-            }
-
-            // No changes to the attributes
-            None => attributes,
-        }
+    fn process<'a>(&self, attributes: &[KeyValue]) -> Option<Vec<KeyValue>> {
+        Context::current().get::<UserType>().map(|user_type| {
+            let mut attrs = attributes.to_vec();
+            attrs.push(KeyValue::new("user_type", user_type.as_str()));
+            attrs
+        })
     }
 }

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

@@ -106,7 +106,7 @@ type Filter = Arc<dyn Fn(&KeyValue) -> bool + Send + Sync>;
 #[derive(Clone)]
 pub(crate) struct AttributeSetFilter {
     filter: Option<Filter>,
-    processor: Option<AggregateProcessor>
+    processor: Option<Arc<dyn MeasurementProcessor>>
 }
 
 impl AttributeSetFilter {
@@ -127,18 +127,28 @@ impl AttributeSetFilter {
             },
             (None, Some(processor)) => {
                 let attributes = Cow::Borrowed(attrs);
-                let attributes = processor.process(attributes);
 
-                run(attributes.as_ref());
+                match processor.process(&attributes) {
+                    Some(attributes) => {
+                        run(&attributes);
+                    }
+                    None => {
+                        run(attrs);
+                    }
+                }
             },
             (Some(filter), Some(processor)) => {
                 let filtered_attrs: Vec<KeyValue> =
                     attrs.iter().filter(|kv| filter(kv)).cloned().collect();
 
-                let attributes = Cow::Owned(filtered_attrs);
-                let attributes = processor.process(attributes);
-
-                run(attributes.as_ref());
+                match processor.process(&filtered_attrs) {
+                    Some(attributes) => {
+                        run(&attributes);
+                    }
+                    None => {
+                        run(attrs);
+                    }
+                }
             }
         }
     }
@@ -226,23 +236,35 @@ impl<T: Number> AggregateBuilder<T> {
 struct AggregateProcessor(Arc<Vec<Arc<dyn MeasurementProcessor>>>);
 
 impl AggregateProcessor {
-    pub fn try_create(
+    fn try_create(
         processors: Vec<Arc<dyn MeasurementProcessor>>,
-    ) -> Option<Self> {
-        if processors.is_empty() {
-            return None;
+    ) -> Option<Arc<dyn MeasurementProcessor>> {
+
+        match processors.len() {
+            0 => return None,
+            1 => Some(processors[0].clone()),
+            _ => Some(Arc::new(AggregateProcessor(Arc::new(processors)))),
         }
-        Some(Self(Arc::new(processors)))
     }
 }
 
 impl MeasurementProcessor for AggregateProcessor {
-    fn process<'a>(&self, mut attributes: Cow<'a, [KeyValue]>) -> Cow<'a, [KeyValue]> {
+    fn process<'a>(&self, attributes: &[KeyValue]) -> Option<Vec<KeyValue>> {
+        // Do not allocate if not necessary.
+        let mut new_attributes: Option<Vec<KeyValue>> = None;
+
         for processor in self.0.iter() {
-            attributes = processor.process(attributes);
+            let existing_or_new = match &new_attributes {
+                Some(new) => new,
+                None => attributes
+            };
+
+            if let Some(new) = processor.process(existing_or_new) {
+                new_attributes = Some(new);
+            }
         }
 
-        attributes
+        new_attributes
     }
 }
 

opentelemetry-sdk/src/metrics/measurement_processor.rs

@@ -1,7 +1,13 @@
 use opentelemetry::KeyValue;
-use std::borrow::Cow;
 
+/// A trait for processing measurement attributes.
 pub trait MeasurementProcessor: Send + Sync + 'static {
-    fn process<'a>(&self, attributes: Cow<'a, [KeyValue]>) -> Cow<'a, [KeyValue]>;
+
+    /// Processes the attributes of a measurement.
+    ///
+    /// The processor might decide to modify the attributes. In that case, it returns
+    /// `Some` with the modified attributes. If no attribute modification is needed,
+    /// it returns `None`.
+    fn process<'a>(&self, attributes: &[KeyValue]) -> Option<Vec<KeyValue>>;
 }
 

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -289,6 +289,7 @@ impl MeterProviderBuilder {
         self
     }
 
+    /// Associates a [MeasurementProcessor] with a [MeterProvider].
     pub fn with_measurement_processor<T: MeasurementProcessor>(mut self, processor: T) -> Self {
         self.measurement_processors.push(Arc::new(processor));
         self