Merge branch 'main' into metrics-race-condition

open-telemetry · Aug 11, 2022 · 88a314b · 88a314b
2 parents 6cacba1 + 4cf41c4
commit 88a314b
Show file tree

Hide file tree

Showing 8 changed files with 171 additions and 24 deletions.
diff --git a/sdk/include/opentelemetry/sdk/metrics/meter.h b/sdk/include/opentelemetry/sdk/metrics/meter.h
@@ -30,7 +30,7 @@ class Meter final : public opentelemetry::metrics::Meter
 public:
   /** Construct a new Meter with the given  pipeline. */
   explicit Meter(
-      std::shared_ptr<sdk::metrics::MeterContext> meter_context,
+      std::weak_ptr<sdk::metrics::MeterContext> meter_context,
       std::unique_ptr<opentelemetry::sdk::instrumentationscope::InstrumentationScope> scope =
           opentelemetry::sdk::instrumentationscope::InstrumentationScope::Create("")) noexcept;
 
@@ -111,7 +111,7 @@ class Meter final : public opentelemetry::metrics::Meter
   // order of declaration is important here - instrumentation scope should destroy after
   // meter-context.
   std::unique_ptr<sdk::instrumentationscope::InstrumentationScope> scope_;
-  std::shared_ptr<sdk::metrics::MeterContext> meter_context_;
+  std::weak_ptr<sdk::metrics::MeterContext> meter_context_;
   // Mapping between instrument-name and Aggregation Storage.
   std::unordered_map<std::string, std::shared_ptr<MetricStorage>> storage_registry_;
   std::unique_ptr<SyncWritableMetricStorage> RegisterSyncMetricStorage(

diff --git a/sdk/include/opentelemetry/sdk/metrics/state/metric_collector.h b/sdk/include/opentelemetry/sdk/metrics/state/metric_collector.h
@@ -31,8 +31,7 @@ class CollectorHandle
 class MetricCollector : public MetricProducer, public CollectorHandle
 {
 public:
-  MetricCollector(std::shared_ptr<MeterContext> &&context,
-                  std::unique_ptr<MetricReader> metric_reader);
+  MetricCollector(MeterContext *context, std::unique_ptr<MetricReader> metric_reader);
 
   AggregationTemporality GetAggregationTemporality(
       InstrumentType instrument_type) noexcept override;
@@ -50,7 +49,7 @@ class MetricCollector : public MetricProducer, public CollectorHandle
   bool Shutdown(std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept;
 
 private:
-  std::shared_ptr<MeterContext> meter_context_;
+  MeterContext *meter_context_;
   std::shared_ptr<MetricReader> metric_reader_;
 };
 }  // namespace metrics

diff --git a/sdk/src/metrics/meter.cc b/sdk/src/metrics/meter.cc
@@ -24,7 +24,7 @@ namespace metrics = opentelemetry::metrics;
 namespace nostd   = opentelemetry::nostd;
 
 Meter::Meter(
-    std::shared_ptr<MeterContext> meter_context,
+    std::weak_ptr<MeterContext> meter_context,
     std::unique_ptr<sdk::instrumentationscope::InstrumentationScope> instrumentation_scope) noexcept
     : scope_{std::move(instrumentation_scope)}, meter_context_{meter_context}
 {}
@@ -204,7 +204,14 @@ const sdk::instrumentationscope::InstrumentationScope *Meter::GetInstrumentation
 std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
     InstrumentDescriptor &instrument_descriptor)
 {
-  auto view_registry = meter_context_->GetViewRegistry();
+  auto ctx = meter_context_.lock();
+  if (!ctx)
+  {
+    OTEL_INTERNAL_LOG_ERROR("[Meter::RegisterMetricStorage] - Error during finding matching views."
+                            << "The metric context is invalid");
+    return nullptr;
+  }
+  auto view_registry = ctx->GetViewRegistry();
   std::unique_ptr<SyncWritableMetricStorage> storages(new SyncMultiMetricStorage());
 
   auto success = view_registry->FindViews(
@@ -240,7 +247,15 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
 std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
     InstrumentDescriptor &instrument_descriptor)
 {
-  auto view_registry = meter_context_->GetViewRegistry();
+  auto ctx = meter_context_.lock();
+  if (!ctx)
+  {
+    OTEL_INTERNAL_LOG_ERROR(
+        "[Meter::RegisterAsyncMetricStorage] - Error during finding matching views."
+        << "The metric context is invalid");
+    return nullptr;
+  }
+  auto view_registry = ctx->GetViewRegistry();
   std::unique_ptr<AsyncWritableMetricStorage> storages(new AsyncMultiMetricStorage());
   auto success = view_registry->FindViews(
       instrument_descriptor, *GetInstrumentationScope(),
@@ -276,11 +291,17 @@ std::vector<MetricData> Meter::Collect(CollectorHandle *collector,
 {
 
   std::vector<MetricData> metric_data_list;
+  auto ctx = meter_context_.lock();
+  if (!ctx)
+  {
+    OTEL_INTERNAL_LOG_ERROR("[Meter::Collect] - Error during collection."
+                            << "The metric context is invalid");
+    return std::vector<MetricData>{};
+  }
   for (auto &metric_storage : storage_registry_)
   {
-    metric_storage.second->Collect(collector, meter_context_->GetCollectors(),
-                                   meter_context_->GetSDKStartTime(), collect_ts,
-                                   [&metric_data_list](MetricData metric_data) {
+    metric_storage.second->Collect(collector, ctx->GetCollectors(), ctx->GetSDKStartTime(),
+                                   collect_ts, [&metric_data_list](MetricData metric_data) {
                                      metric_data_list.push_back(metric_data);
                                      return true;
                                    });

diff --git a/sdk/src/metrics/meter_context.cc b/sdk/src/metrics/meter_context.cc
@@ -46,8 +46,7 @@ opentelemetry::common::SystemTimestamp MeterContext::GetSDKStartTime() noexcept
 
 void MeterContext::AddMetricReader(std::unique_ptr<MetricReader> reader) noexcept
 {
-  auto collector =
-      std::shared_ptr<MetricCollector>{new MetricCollector(shared_from_this(), std::move(reader))};
+  auto collector = std::shared_ptr<MetricCollector>{new MetricCollector(this, std::move(reader))};
   collectors_.push_back(collector);
 }
 

diff --git a/sdk/src/metrics/state/metric_collector.cc b/sdk/src/metrics/state/metric_collector.cc
@@ -16,10 +16,9 @@ namespace sdk
 namespace metrics
 {
 
-MetricCollector::MetricCollector(
-    std::shared_ptr<opentelemetry::sdk::metrics::MeterContext> &&context,
-    std::unique_ptr<MetricReader> metric_reader)
-    : meter_context_{std::move(context)}, metric_reader_{std::move(metric_reader)}
+MetricCollector::MetricCollector(opentelemetry::sdk::metrics::MeterContext *context,
+                                 std::unique_ptr<MetricReader> metric_reader)
+    : meter_context_{context}, metric_reader_{std::move(metric_reader)}
 {
   metric_reader_->SetMetricProducer(this);
 }
@@ -33,6 +32,12 @@ AggregationTemporality MetricCollector::GetAggregationTemporality(
 bool MetricCollector::Collect(
     nostd::function_ref<bool(ResourceMetrics &metric_data)> callback) noexcept
 {
+  if (!meter_context_)
+  {
+    OTEL_INTERNAL_LOG_ERROR("[MetricCollector::Collect] - Error during collecting."
+                            << "The metric context is invalid");
+    return false;
+  }
   ResourceMetrics resource_metrics;
   for (auto &meter : meter_context_->GetMeters())
   {

diff --git a/sdk/src/metrics/state/observable_registry.cc b/sdk/src/metrics/state/observable_registry.cc
@@ -7,6 +7,7 @@
 #  include "opentelemetry/sdk/metrics/async_instruments.h"
 #  include "opentelemetry/sdk/metrics/observer_result.h"
 #  include "opentelemetry/sdk/metrics/state/metric_storage.h"
+#  include "opentelemetry/sdk_config.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
@@ -51,6 +52,12 @@ void ObservableRegistry::Observe(opentelemetry::common::SystemTimestamp collecti
     auto storage =
         static_cast<opentelemetry::sdk::metrics::ObservableInstrument *>(callback_wrap->instrument)
             ->GetMetricStorage();
+    if (!storage)
+    {
+      OTEL_INTERNAL_LOG_ERROR("[ObservableRegistry::Observe] - Error during observe."
+                              << "The metric storage is invalid");
+      return;
+    }
     if (value_type == InstrumentValueType::kDouble)
     {
       nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>> ob_res(