Update metrics

Author: utpilla

opentelemetry-sdk/src/metrics/instrument.rs

@@ -278,10 +278,20 @@ impl InstrumentId {
     }
 }
 
+pub(crate) trait SyncInstrumentUpdate: Send + Sync {}
+
+impl<T> SyncInstrumentUpdate for ResolvedMeasures<T> {}
+
 pub(crate) struct ResolvedMeasures<T> {
     pub(crate) measures: Vec<Arc<dyn Measure<T>>>,
 }
 
+impl<T: 'static> ResolvedMeasures<T> {
+    pub(crate) fn as_sync_instrument_update(self: Arc<Self>) -> Arc<dyn SyncInstrumentUpdate> {
+        self
+    }
+}
+
 impl<T: Copy + 'static> SyncInstrument<T> for ResolvedMeasures<T> {
     fn measure(&self, val: T, attrs: &[KeyValue]) {
         for measure in &self.measures {

opentelemetry-sdk/src/metrics/meter.rs

@@ -1,5 +1,8 @@
 use core::fmt;
-use std::{borrow::Cow, sync::Arc};
+use std::{
+    borrow::Cow,
+    sync::{Arc, Mutex},
+};
 
 use opentelemetry::{
     metrics::{
@@ -11,7 +14,7 @@ use opentelemetry::{
 };
 
 use crate::metrics::{
-    instrument::{Instrument, InstrumentKind, Observable, ResolvedMeasures},
+    instrument::{Instrument, InstrumentKind, Observable, ResolvedMeasures, SyncInstrumentUpdate},
     internal::{self, Number},
     pipeline::{Pipelines, Resolver},
     MetricError, MetricResult,
@@ -50,6 +53,7 @@ pub(crate) struct SdkMeter {
     u64_resolver: Resolver<u64>,
     i64_resolver: Resolver<i64>,
     f64_resolver: Resolver<f64>,
+    pub(crate) sync_instruments: Mutex<Vec<Arc<dyn SyncInstrumentUpdate>>>,
 }
 
 impl SdkMeter {
@@ -62,6 +66,7 @@ impl SdkMeter {
             u64_resolver: Resolver::new(Arc::clone(&pipes), Arc::clone(&view_cache)),
             i64_resolver: Resolver::new(Arc::clone(&pipes), Arc::clone(&view_cache)),
             f64_resolver: Resolver::new(pipes, view_cache),
+            sync_instruments: Mutex::new(Vec::new()),
         }
     }
 
@@ -93,8 +98,16 @@ impl SdkMeter {
                 builder.unit,
                 None,
             )
-            .map(|i| Counter::new(Arc::new(i)))
-        {
+            .map(|i| {
+                let resolved_measures = Arc::new(i);
+                let sync_instrument_update = resolved_measures.clone().as_sync_instrument_update();
+
+                self.sync_instruments
+                    .lock()
+                    .unwrap()
+                    .push(sync_instrument_update);
+                Counter::new(resolved_measures)
+            }) {
             Ok(counter) => counter,
             Err(err) => {
                 otel_error!(

opentelemetry-sdk/src/metrics/meter_provider.rs

@@ -141,6 +141,10 @@ impl SdkMeterProviderInner {
                 "MeterProvider shutdown already invoked.".into(),
             ))
         } else {
+            let mut meters = self.meters.lock().unwrap();
+            for (_, meter) in meters.drain() {
+                meter.sync_instruments.lock().unwrap().clear(); // Clear all `SyncInstrument` trait objects to avoid any further updates from instruments.
+            }
             self.pipes.shutdown()
         }
     }

opentelemetry/src/metrics/instruments/counter.rs

@@ -1,6 +1,6 @@
 use crate::KeyValue;
 use core::fmt;
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 
 use super::SyncInstrument;
 
@@ -11,7 +11,7 @@ use super::SyncInstrument;
 /// duplicate [`Counter`]s for the same metric could lower SDK performance.
 #[derive(Clone)]
 #[non_exhaustive]
-pub struct Counter<T>(Arc<dyn SyncInstrument<T> + Send + Sync>);
+pub struct Counter<T>(Weak<dyn SyncInstrument<T> + Send + Sync>);
 
 impl<T> fmt::Debug for Counter<T>
 where
@@ -25,12 +25,16 @@ where
 impl<T> Counter<T> {
     /// Create a new counter.
     pub fn new(inner: Arc<dyn SyncInstrument<T> + Send + Sync>) -> Self {
-        Counter(inner)
+        Counter(Arc::downgrade(&inner))
     }
 
     /// Records an increment to the counter.
     pub fn add(&self, value: T, attributes: &[KeyValue]) {
-        self.0.measure(value, attributes)
+        if let Some(inner) = self.0.upgrade() {
+            inner.measure(value, attributes);
+        } else {
+            // TODO: Log error
+        }
     }
 }