Fix metrics unused instruments (#407)

Author: jtescher

opentelemetry-prometheus/tests/integration_test.rs

@@ -6,6 +6,33 @@ use opentelemetry::{
 use opentelemetry_prometheus::PrometheusExporter;
 use prometheus::{Encoder, TextEncoder};
 
+#[test]
+fn free_unused_instruments() {
+    let exporter = opentelemetry_prometheus::exporter()
+        .with_default_histogram_boundaries(vec![-0.5, 1.0])
+        .with_resource(Resource::new(vec![KeyValue::new("R", "V")]))
+        .init();
+    let mut expected = Vec::new();
+
+    {
+        let meter = exporter.provider().unwrap().meter("test", None);
+        let counter = meter.f64_counter("counter").init();
+
+        let labels = vec![KeyValue::new("A", "B"), KeyValue::new("C", "D")];
+
+        counter.add(10.0, &labels);
+        counter.add(5.3, &labels);
+
+        expected.push(r#"counter{A="B",C="D",R="V"} 15.3"#);
+    }
+    // Standard export
+    compare_export(&exporter, expected.clone());
+    // Final export before instrument dropped
+    compare_export(&exporter, expected.clone());
+    // Instrument dropped, but last value kept by prom exporter
+    compare_export(&exporter, expected);
+}
+
 #[test]
 fn test_add() {
     let exporter = opentelemetry_prometheus::exporter()

opentelemetry/src/sdk/metrics/mod.rs

@@ -178,8 +178,7 @@ impl AccumulatorCore {
     fn collect_sync_instruments(&self, locked_processor: &mut dyn LockedProcessor) -> usize {
         let mut checkpointed = 0;
 
-        for element in self.current.iter() {
-            let (key, value) = element.pair();
+        self.current.retain(|_key, value| {
             let mods = &value.update_count.load();
             let coll = &value.collected_count.load();
 
@@ -192,17 +191,17 @@ impl AccumulatorCore {
                 // Having no updates since last collection, try to remove if
                 // there are no bound handles
                 if Arc::strong_count(&value) == 1 {
-                    self.current.remove(key);
-
                     // There's a potential race between loading collected count and
                     // loading the strong count in this function.  Since this is the
                     // last we'll see of this record, checkpoint.
                     if mods.partial_cmp(&NumberKind::U64, coll) != Some(Ordering::Equal) {
                         checkpointed += self.checkpoint_record(value, locked_processor);
                     }
+                    return false;
                 }
-            }
-        }
+            };
+            true
+        });
 
         checkpointed
     }