[Metrics] Allow registering one callback for multiple instruments

This commit adds a pair of methods `Meter::RegisterCallback` and
`Meter::DeregisterCallback` which allow you to set up one callback
which can record observations for multiple observable instruments.

The callback will be passed an object of type MultiObserverResult, which
can be used to get an ObserverResultT for an instrument.

This allows a single expensive operation to yield multiple different
observable metrics without the need to call said expensive operation
multiple times.

Author: KJTsanaktsidis

api/include/opentelemetry/metrics/meter.h

@@ -3,7 +3,10 @@
 
 #pragma once
 
+#include <cstdint>
+
 #include "opentelemetry/nostd/shared_ptr.h"
+#include "opentelemetry/nostd/span.h"
 #include "opentelemetry/nostd/string_view.h"
 #include "opentelemetry/nostd/unique_ptr.h"
 #include "opentelemetry/version.h"
@@ -25,6 +28,8 @@ template <typename T>
 class Gauge;
 
 class ObservableInstrument;
+class MultiObserverResult;
+using MultiObservableCallbackPtr = void (*)(MultiObserverResult &, void *);
 
 /**
  * Handles instrument creation and provides a facility for batch recording.
@@ -169,6 +174,33 @@ class Meter
       nostd::string_view name,
       nostd::string_view description = "",
       nostd::string_view unit        = "") noexcept = 0;
+
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+
+  /**
+   * Registers a callback to be invoked when metrics are collected by this meter. The callback will
+   * be passed a MultiObserverResult which it can use to make observations for any or all of the
+   * instruments provided in this registration. Any measurements recorded for instruments _not_ in
+   * the initial RegisterCallback call will be discarded.
+   *
+   * @param callback the callback to be invoked.
+   * @param state the state to be passed to the callback.
+   * @param instruments the instruments to be observed.
+   * @return a unique identifier for the registered callback, which can be used to unregister the
+   * callback in DeregisterCallback.
+   */
+  virtual uintptr_t RegisterCallback(
+      MultiObservableCallbackPtr callback,
+      void *state,
+      nostd::span<ObservableInstrument *> instruments) noexcept = 0;
+
+  /**
+   * Unregisters a callback previously registered with RegisterCallback.
+   *
+   * @param callback_id the unique identifier returned by RegisterCallback.
+   */
+  virtual void DeregisterCallback(uintptr_t callback_id) noexcept = 0;
+#endif
 };
 }  // namespace metrics
 OPENTELEMETRY_END_NAMESPACE

api/include/opentelemetry/metrics/multi_observer_result.h

@@ -0,0 +1,57 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <type_traits>
+
+#include "opentelemetry/metrics/async_instruments.h"
+#include "opentelemetry/metrics/observer_result.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace metrics
+{
+
+class MultiObserverResult
+{
+
+public:
+  virtual ~MultiObserverResult() = default;
+
+  /**
+   * Obtain an ObserverResultT<T> for the given instrument, that can be used to record
+   * a measurement on said instrument from a multi-observer callback registered with
+   * Meter::RegisterCallback. The instrument _must_ have been included in the original
+   * call to Meter::RegisterCallback; any data points set on other instruments will be
+   * discarded.
+   *
+   * @param instrument The instrument for which to obtain an ObserverResult.
+   * @return An ObserverResultT<T> for the given instrument.
+   */
+  template <typename T>
+  ObserverResultT<T> &ForInstrument(const ObservableInstrument *instrument) = delete;
+
+protected:
+  // You can't have a virtual template, and you can't overload on return type, so we need to
+  // enumerate the options for the observer result type as separate methods to override.
+  virtual ObserverResultT<double> &ForInstrumentDouble(const ObservableInstrument *instrument) = 0;
+  virtual ObserverResultT<int64_t> &ForInstrumentInt64(const ObservableInstrument *instrument) = 0;
+};
+
+template <>
+inline ObserverResultT<double> &MultiObserverResult::ForInstrument<double>(
+    const ObservableInstrument *instrument)
+{
+  return ForInstrumentDouble(instrument);
+}
+
+template <>
+inline ObserverResultT<int64_t> &MultiObserverResult::ForInstrument<int64_t>(
+    const ObservableInstrument *instrument)
+{
+  return ForInstrumentInt64(instrument);
+}
+
+}  // namespace metrics
+OPENTELEMETRY_END_NAMESPACE

api/include/opentelemetry/metrics/noop.h

@@ -229,6 +229,18 @@ class NoopMeter final : public Meter
     return nostd::shared_ptr<ObservableInstrument>(
         new NoopObservableInstrument(name, description, unit));
   }
+
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+  uintptr_t RegisterCallback(
+      MultiObservableCallbackPtr /* callback */,
+      void * /* state */,
+      nostd::span<ObservableInstrument *> /* instruments */) noexcept override
+  {
+    return 0;
+  }
+
+  void DeregisterCallback(uintptr_t /* callback_id */) noexcept override {}
+#endif
 };
 
 /**

sdk/include/opentelemetry/sdk/metrics/meter.h

@@ -131,6 +131,14 @@ class Meter final : public opentelemetry::metrics::Meter
   std::vector<MetricData> Collect(CollectorHandle *collector,
                                   opentelemetry::common::SystemTimestamp collect_ts) noexcept;
 
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+  uintptr_t RegisterCallback(opentelemetry::metrics::MultiObservableCallbackPtr callback,
+                             void *state,
+                             nostd::span<opentelemetry::metrics::ObservableInstrument *>
+                                 instruments) noexcept override;
+
+  void DeregisterCallback(uintptr_t callback_id) noexcept override;
+#endif
 private:
   // order of declaration is important here - instrumentation scope should destroy after
   // meter-context.

sdk/include/opentelemetry/sdk/metrics/multi_observer_result.h

@@ -0,0 +1,48 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <unordered_map>
+
+#include "opentelemetry/metrics/async_instruments.h"
+#include "opentelemetry/metrics/multi_observer_result.h"
+#include "opentelemetry/metrics/observer_result.h"
+#include "opentelemetry/nostd/variant.h"
+#include "opentelemetry/sdk/metrics/observer_result.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+class OPENTELEMETRY_EXPORT MultiObserverResult final
+    : public opentelemetry::metrics::MultiObserverResult
+{
+public:
+  void RegisterInstrument(opentelemetry::metrics::ObservableInstrument *instrument);
+  void DeregisterInstrument(opentelemetry::metrics::ObservableInstrument *instrument);
+  size_t InstrumentCount() const;
+  void Reset();
+  void StoreResults(opentelemetry::common::SystemTimestamp collection_ts);
+
+protected:
+  opentelemetry::metrics::ObserverResultT<double> &ForInstrumentDouble(
+      const opentelemetry::metrics::ObservableInstrument *instrument) override;
+  opentelemetry::metrics::ObserverResultT<int64_t> &ForInstrumentInt64(
+      const opentelemetry::metrics::ObservableInstrument *instrument) override;
+
+private:
+  // This is _different_ to opentelemetry::metrics::ObserverResult because this variant is
+  // a variant directly of ObserverResultT, not of _pointers_ to ObserverResultT.
+  // This allows us to avoid an unnescessary layer of inderection and a bunch of allocations.
+  using ObserverResultDirect =
+      nostd::variant<nostd::monostate, ObserverResultT<double>, ObserverResultT<int64_t>>;
+  std::unordered_map<opentelemetry::metrics::ObservableInstrument *, ObserverResultDirect>
+      observer_results_;
+};
+}  // namespace metrics
+}  // namespace sdk
+
+OPENTELEMETRY_END_NAMESPACE

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

@@ -3,12 +3,17 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <mutex>
+#include <unordered_map>
 #include <vector>
 
 #include "opentelemetry/common/timestamp.h"
 #include "opentelemetry/metrics/async_instruments.h"
+#include "opentelemetry/metrics/meter.h"
+#include "opentelemetry/nostd/span.h"
+#include "opentelemetry/sdk/metrics/multi_observer_result.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -24,6 +29,13 @@ struct ObservableCallbackRecord
   opentelemetry::metrics::ObservableInstrument *instrument;
 };
 
+struct MultiObservableCallbackRecord
+{
+  opentelemetry::metrics::MultiObservableCallbackPtr callback;
+  void *state;
+  MultiObserverResult observable_result{};
+};
+
 class ObservableRegistry
 {
 public:
@@ -34,6 +46,13 @@ class ObservableRegistry
   void RemoveCallback(opentelemetry::metrics::ObservableCallbackPtr callback,
                       void *state,
                       opentelemetry::metrics::ObservableInstrument *instrument);
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+  uintptr_t AddMultiCallback(
+      opentelemetry::metrics::MultiObservableCallbackPtr callback,
+      void *state,
+      nostd::span<opentelemetry::metrics::ObservableInstrument *> instruments);
+  void RemoveMultiCallback(uintptr_t id);
+#endif
 
   void CleanupCallback(opentelemetry::metrics::ObservableInstrument *instrument);
 
@@ -42,6 +61,9 @@ class ObservableRegistry
 private:
   std::vector<std::unique_ptr<ObservableCallbackRecord>> callbacks_;
   std::mutex callbacks_m_;
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+  std::unordered_map<uintptr_t, std::unique_ptr<MultiObservableCallbackRecord>> multi_callbacks_;
+#endif
 };
 
 }  // namespace metrics

sdk/src/metrics/CMakeLists.txt

@@ -12,6 +12,7 @@ add_library(
   meter_context.cc
   meter_context_factory.cc
   metric_reader.cc
+  multi_observer_result.cc
   instrument_metadata_validator.cc
   export/periodic_exporting_metric_reader.cc
   export/periodic_exporting_metric_reader_factory.cc

sdk/src/metrics/meter.cc

@@ -663,6 +663,21 @@ std::vector<MetricData> Meter::Collect(CollectorHandle *collector,
   return metric_data_list;
 }
 
+#if OPENTELEMETRY_ABI_VERSION_NO >= 2
+uintptr_t Meter::RegisterCallback(
+    opentelemetry::metrics::MultiObservableCallbackPtr callback,
+    void *state,
+    nostd::span<opentelemetry::metrics::ObservableInstrument *> instruments) noexcept
+{
+  return observable_registry_->AddMultiCallback(callback, state, instruments);
+}
+
+void Meter::DeregisterCallback(uintptr_t callback_id) noexcept
+{
+  observable_registry_->RemoveMultiCallback(callback_id);
+}
+#endif
+
 // Implementation of the log message recommended by the SDK specification for duplicate instruments.
 // See
 // https://github.com/open-telemetry/opentelemetry-specification/blob/9c8c30631b0e288de93df7452f91ed47f6fba330/specification/metrics/sdk.md?plain=1#L882

sdk/src/metrics/multi_observer_result.cc

@@ -0,0 +1,113 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "opentelemetry/sdk/metrics/multi_observer_result.h"
+#include "opentelemetry/common/timestamp.h"
+#include "opentelemetry/metrics/async_instruments.h"
+#include "opentelemetry/metrics/observer_result.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
+#include "opentelemetry/sdk/metrics/async_instruments.h"
+#include "opentelemetry/sdk/metrics/state/metric_storage.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+namespace
+{
+struct StoreResultVisitor
+{
+  void operator()(ObserverResultT<int64_t> &result) const
+  {
+    storage->RecordLong(result.GetMeasurements(), collection_ts);
+  }
+
+  void operator()(ObserverResultT<double> &result) const
+  {
+    storage->RecordDouble(result.GetMeasurements(), collection_ts);
+  }
+
+  void operator()(const nostd::monostate &) const {}
+
+  AsyncWritableMetricStorage *storage;
+  opentelemetry::common::SystemTimestamp collection_ts;
+};
+}  // namespace
+
+void MultiObserverResult::RegisterInstrument(
+    opentelemetry::metrics::ObservableInstrument *instrument)
+{
+  observer_results_.emplace(instrument, nostd::monostate());
+}
+
+void MultiObserverResult::DeregisterInstrument(
+    opentelemetry::metrics::ObservableInstrument *instrument)
+{
+  observer_results_.erase(instrument);
+}
+
+size_t MultiObserverResult::InstrumentCount() const
+{
+  return observer_results_.size();
+}
+
+void MultiObserverResult::Reset()
+{
+  for (auto it = observer_results_.begin(); it != observer_results_.end(); ++it)
+  {
+    it->second = nostd::monostate();
+  }
+}
+
+void MultiObserverResult::StoreResults(opentelemetry::common::SystemTimestamp collection_ts)
+{
+  for (auto &el : observer_results_)
+  {
+    auto *instrument = el.first;
+    auto &result     = el.second;
+
+    auto storage = static_cast<opentelemetry::sdk::metrics::ObservableInstrument *>(instrument)
+                       ->GetMetricStorage();
+    nostd::visit(StoreResultVisitor{storage, collection_ts}, result);
+  }
+}
+
+opentelemetry::metrics::ObserverResultT<double> &MultiObserverResult::ForInstrumentDouble(
+    const opentelemetry::metrics::ObservableInstrument *instrument)
+{
+  static opentelemetry::sdk::metrics::ObserverResultT<double> null_result;
+  // const_cast is appropriate here, because we're _not_ modifying the passed-in pointer;
+  // we just need to make it non-const to be able to look it up in our map.
+  auto it = observer_results_.find(
+      const_cast<opentelemetry::metrics::ObservableInstrument *>(instrument));
+  if (it == observer_results_.end())
+  {
+    OTEL_INTERNAL_LOG_ERROR("[MultiObserverResult::ForInstrumentDouble]"
+                            << "The instrument is not registered on with callback");
+    return null_result;
+  }
+
+  return it->second.emplace<ObserverResultT<double>>();
+}
+
+opentelemetry::metrics::ObserverResultT<int64_t> &MultiObserverResult::ForInstrumentInt64(
+    const opentelemetry::metrics::ObservableInstrument *instrument)
+{
+  static opentelemetry::sdk::metrics::ObserverResultT<int64_t> null_result;
+  auto it = observer_results_.find(
+      const_cast<opentelemetry::metrics::ObservableInstrument *>(instrument));
+  if (it == observer_results_.end())
+  {
+    OTEL_INTERNAL_LOG_ERROR("[MultiObserverResult::ForInstrumentInt64]"
+                            << "The instrument is not registered on with callback");
+    return null_result;
+  }
+
+  return it->second.emplace<ObserverResultT<int64_t>>();
+}
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE