[SDK] Metrics ObservableRegistry Cleanup (#2376)

sdk/include/opentelemetry/sdk/metrics/async_instruments.h

@@ -25,6 +25,7 @@ class ObservableInstrument : public opentelemetry::metrics::ObservableInstrument
   ObservableInstrument(InstrumentDescriptor instrument_descriptor,
                        std::unique_ptr<AsyncWritableMetricStorage> storage,
                        std::shared_ptr<ObservableRegistry> observable_registry);
+  ~ObservableInstrument() override;
 
   void AddCallback(opentelemetry::metrics::ObservableCallbackPtr callback,
                    void *state) noexcept override;

sdk/include/opentelemetry/sdk/metrics/state/observable_registry.h

@@ -35,6 +35,8 @@ class ObservableRegistry
                       void *state,
                       opentelemetry::metrics::ObservableInstrument *instrument);
 
+  void CleanupCallback(opentelemetry::metrics::ObservableInstrument *instrument);
+
   void Observe(opentelemetry::common::SystemTimestamp collection_ts);
 
 private:

sdk/src/metrics/async_instruments.cc

@@ -21,6 +21,11 @@ ObservableInstrument::ObservableInstrument(InstrumentDescriptor instrument_descr
 
 {}
 
+ObservableInstrument::~ObservableInstrument()
+{
+  observable_registry_->CleanupCallback(this);
+}
+
 void ObservableInstrument::AddCallback(opentelemetry::metrics::ObservableCallbackPtr callback,
                                        void *state) noexcept
 {

sdk/src/metrics/state/observable_registry.cc

@@ -38,6 +38,16 @@ void ObservableRegistry::RemoveCallback(opentelemetry::metrics::ObservableCallba
   callbacks_.erase(new_end, callbacks_.end());
 }
 
+void ObservableRegistry::CleanupCallback(opentelemetry::metrics::ObservableInstrument *instrument)
+{
+  std::lock_guard<std::mutex> lock_guard{callbacks_m_};
+  auto iter = std::remove_if(callbacks_.begin(), callbacks_.end(),
+                             [instrument](const std::unique_ptr<ObservableCallbackRecord> &record) {
+                               return record->instrument == instrument;
+                             });
+  callbacks_.erase(iter, callbacks_.end());
+}
+
 void ObservableRegistry::Observe(opentelemetry::common::SystemTimestamp collection_ts)
 {
   std::lock_guard<std::mutex> lock_guard{callbacks_m_};