fix: metrics view bug fix

Author: cijothomas

opentelemetry-sdk/src/metrics/instrument.rs

@@ -190,11 +190,11 @@ impl Instrument {
 #[allow(unreachable_pub)]
 pub struct Stream {
     /// The human-readable identifier of the stream.
-    pub name: Cow<'static, str>,
+    pub name: Option<Cow<'static, str>>,
     /// Describes the purpose of the data.
-    pub description: Cow<'static, str>,
+    pub description: Option<Cow<'static, str>>,
     /// the unit of measurement recorded.
-    pub unit: Cow<'static, str>,
+    pub unit: Option<Cow<'static, str>>,
     /// Aggregation the stream uses for an instrument.
     pub aggregation: Option<Aggregation>,
     /// An allow-list of attribute keys that will be preserved for the stream.
@@ -217,19 +217,19 @@ impl Stream {
 
     /// Set the stream name.
     pub fn name(mut self, name: impl Into<Cow<'static, str>>) -> Self {
-        self.name = name.into();
+        self.name = Some(name.into());
         self
     }
 
     /// Set the stream description.
     pub fn description(mut self, description: impl Into<Cow<'static, str>>) -> Self {
-        self.description = description.into();
+        self.description = Some(description.into());
         self
     }
 
     /// Set the stream unit.
     pub fn unit(mut self, unit: impl Into<Cow<'static, str>>) -> Self {
-        self.unit = unit.into();
+        self.unit = Some(unit.into());
         self
     }
 

opentelemetry-sdk/src/metrics/mod.rs

@@ -1463,6 +1463,150 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn view_test_rename() {
+        // Run this test with stdout enabled to see output.
+        // cargo test view_test_rename --all-features -- --nocapture
+
+        // Arrange
+        let view_rename = |i: &Instrument| {
+            if i.name == "my_counter" {
+                Some(Stream::new().name("my_counter_renamed"))
+            } else {
+                None
+            }
+        };
+
+        let exporter = InMemoryMetricExporter::default();
+        let meter_provider = SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter.clone())
+            .with_view(view_rename)
+            .build();
+
+        // Act
+        let meter = meter_provider.meter("test");
+        let counter = meter
+            .f64_counter("my_counter")
+            .with_unit("my_unit")
+            .with_description("my_description")
+            .build();
+
+        counter.add(1.5, &[KeyValue::new("key1", "value1")]);
+        meter_provider.force_flush().unwrap();
+
+        // Assert
+        let resource_metrics = exporter
+            .get_finished_metrics()
+            .expect("metrics are expected to be exported.");
+        assert!(!resource_metrics.is_empty());
+        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
+        assert_eq!(
+            metric.name, "my_counter_renamed",
+            "my_counter_renamed is expected."
+        );
+        assert_eq!(metric.unit, "my_unit", "Original unit should be retained.");
+        assert_eq!(
+            metric.description, "my_description",
+            "Original description should be retained."
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn view_test_change_unit() {
+        // Run this test with stdout enabled to see output.
+        // cargo test view_test_change_unit --all-features -- --nocapture
+
+        // Arrange
+        let view_rename = |i: &Instrument| {
+            if i.name == "my_counter" {
+                Some(Stream::new().unit("my_unit_new"))
+            } else {
+                None
+            }
+        };
+
+        let exporter = InMemoryMetricExporter::default();
+        let meter_provider = SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter.clone())
+            .with_view(view_rename)
+            .build();
+
+        // Act
+        let meter = meter_provider.meter("test");
+        let counter = meter
+            .f64_counter("my_counter")
+            .with_unit("my_unit")
+            .with_description("my_description")
+            .build();
+
+        counter.add(1.5, &[KeyValue::new("key1", "value1")]);
+        meter_provider.force_flush().unwrap();
+
+        // Assert
+        let resource_metrics = exporter
+            .get_finished_metrics()
+            .expect("metrics are expected to be exported.");
+        assert!(!resource_metrics.is_empty());
+        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
+        assert_eq!(
+            metric.name, "my_counter",
+            "original name is expected."
+        );
+        assert_eq!(metric.unit, "my_unit_new", "unit should be updated.");
+        assert_eq!(
+            metric.description, "my_description",
+            "Original description should be retained."
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn view_test_change_description() {
+        // Run this test with stdout enabled to see output.
+        // cargo test view_test_change_description --all-features -- --nocapture
+
+        // Arrange
+        let view_rename = |i: &Instrument| {
+            if i.name == "my_counter" {
+                Some(Stream::new().description("my_description_new"))
+            } else {
+                None
+            }
+        };
+
+        let exporter = InMemoryMetricExporter::default();
+        let meter_provider = SdkMeterProvider::builder()
+            .with_periodic_exporter(exporter.clone())
+            .with_view(view_rename)
+            .build();
+
+        // Act
+        let meter = meter_provider.meter("test");
+        let counter = meter
+            .f64_counter("my_counter")
+            .with_unit("my_unit")
+            .with_description("my_description")
+            .build();
+
+        counter.add(1.5, &[KeyValue::new("key1", "value1")]);
+        meter_provider.force_flush().unwrap();
+
+        // Assert
+        let resource_metrics = exporter
+            .get_finished_metrics()
+            .expect("metrics are expected to be exported.");
+        assert!(!resource_metrics.is_empty());
+        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
+        assert_eq!(
+            metric.name, "my_counter",
+            "original name is expected."
+        );
+        assert_eq!(metric.unit, "my_unit", "Original unit should be retained.");
+        assert_eq!(
+            metric.description, "my_description_new",
+            "description should be updated."
+        );
+    }
+
     fn asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
         instrument_name: &'static str,
         should_not_emit: bool,

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -267,12 +267,22 @@ where
         // The cache will return the same Aggregator instance. Use stream ids to de duplicate.
         let mut seen = HashSet::new();
         for v in &self.pipeline.views {
-            let stream = match v.match_inst(&inst) {
+            let mut stream = match v.match_inst(&inst) {
                 Some(stream) => stream,
                 None => continue,
             };
             matched = true;
 
+            if stream.name.is_none() {
+                stream.name = Some(inst.name.clone());
+            }
+            if stream.description.is_none() {
+                stream.description = Some(inst.description.clone());
+            }
+            if stream.unit.is_none() {
+                stream.unit = Some(inst.unit.clone());
+            }
+
             let id = self.inst_id(kind, &stream);
             if seen.contains(&id) {
                 continue; // This aggregator has already been added
@@ -300,9 +310,9 @@ where
 
         // Apply implicit default view if no explicit matched.
         let mut stream = Stream {
-            name: inst.name,
-            description: inst.description,
-            unit: inst.unit,
+            name: Some(inst.name),
+            description: Some(inst.description),
+            unit: Some(inst.unit),
             aggregation: None,
             allowed_attribute_keys: None,
             cardinality_limit,
@@ -403,16 +413,16 @@ where
 
             otel_debug!(
                 name : "Metrics.InstrumentCreated",
-                instrument_name = stream.name.as_ref(),
+                instrument_name = stream.name.clone().unwrap_or_default().as_ref(),
                 cardinality_limit = cardinality_limit,
             );
 
             self.pipeline.add_sync(
                 scope.clone(),
                 InstrumentSync {
-                    name: stream.name,
-                    description: stream.description,
-                    unit: stream.unit,
+                    name: stream.name.unwrap_or_default(),
+                    description: stream.description.unwrap_or_default(),
+                    unit: stream.unit.unwrap_or_default(),
                     comp_agg: collect,
                 },
             );
@@ -453,10 +463,10 @@ where
 
     fn inst_id(&self, kind: InstrumentKind, stream: &Stream) -> InstrumentId {
         InstrumentId {
-            name: stream.name.clone(),
-            description: stream.description.clone(),
+            name: stream.name.clone().unwrap_or_default(),
+            description: stream.description.clone().unwrap_or_default(),
             kind,
-            unit: stream.unit.clone(),
+            unit: stream.unit.clone().unwrap_or_default(),
             number: Cow::Borrowed(std::any::type_name::<T>()),
         }
     }

opentelemetry-sdk/src/metrics/view.rs

@@ -110,7 +110,7 @@ pub fn new_view(criteria: Instrument, mask: Stream) -> MetricResult<Box<dyn View
     let contains_wildcard = criteria.name.contains(['*', '?']);
 
     let match_fn: Box<dyn Fn(&Instrument) -> bool + Send + Sync> = if contains_wildcard {
-        if !mask.name.is_empty() {
+        if !mask.name.is_none() {
             // TODO - The error is getting lost here. Need to return or log.
             return Ok(Box::new(empty_view));
         }
@@ -144,20 +144,20 @@ pub fn new_view(criteria: Instrument, mask: Stream) -> MetricResult<Box<dyn View
     Ok(Box::new(move |i: &Instrument| -> Option<Stream> {
         if match_fn(i) {
             Some(Stream {
-                name: if !mask.name.is_empty() {
+                name: if !mask.name.is_none() {
                     mask.name.clone()
                 } else {
-                    i.name.clone()
+                    Some(i.name.clone())
                 },
-                description: if !mask.description.is_empty() {
+                description: if !mask.description.is_none() {
                     mask.description.clone()
                 } else {
-                    i.description.clone()
+                    Some(i.description.clone())
                 },
-                unit: if !mask.unit.is_empty() {
+                unit: if !mask.unit.is_none() {
                     mask.unit.clone()
                 } else {
-                    i.unit.clone()
+                    Some(i.unit.clone())
                 },
                 aggregation: agg.clone(),
                 allowed_attribute_keys: mask.allowed_attribute_keys.clone(),