[SDK] Use shared_ptr internally for AttributesProcessor to prevent use-after-free (open-telemetry#3457)

lalitb · web-flow · commit d2e728914d29 · 2025-06-03T10:40:10.000-07:00
* fix

* fix

* add test

* fix

* iwuc
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h
@@ -58,7 +58,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
 public:
   SyncMetricStorage(InstrumentDescriptor instrument_descriptor,
                     const AggregationType aggregation_type,
-                    const AttributesProcessor *attributes_processor,
+                    std::shared_ptr<const AttributesProcessor> attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                     ExemplarFilterType exempler_filter_type,
                     nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
@@ -67,7 +67,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
                     size_t attributes_limit = kAggregationCardinalityLimit)
       : instrument_descriptor_(instrument_descriptor),
         attributes_hashmap_(new AttributesHashMap(attributes_limit)),
-        attributes_processor_(attributes_processor),
+        attributes_processor_(std::move(attributes_processor)),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
         exemplar_filter_type_(exempler_filter_type),
         exemplar_reservoir_(exemplar_reservoir),
@@ -119,7 +119,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
 
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
     attributes_hashmap_
-        ->GetOrSetDefault(attributes, attributes_processor_, create_default_aggregation_)
+        ->GetOrSetDefault(attributes, attributes_processor_.get(), create_default_aggregation_)
         ->Aggregate(value);
   }
 
@@ -160,7 +160,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
 #endif
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
     attributes_hashmap_
-        ->GetOrSetDefault(attributes, attributes_processor_, create_default_aggregation_)
+        ->GetOrSetDefault(attributes, attributes_processor_.get(), create_default_aggregation_)
         ->Aggregate(value);
   }
 
@@ -175,7 +175,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
   // hashmap to maintain the metrics for delta collection (i.e, collection since last Collect call)
   std::unique_ptr<AttributesHashMap> attributes_hashmap_;
   std::function<std::unique_ptr<Aggregation>()> create_default_aggregation_;
-  const AttributesProcessor *attributes_processor_;
+  std::shared_ptr<const AttributesProcessor> attributes_processor_;
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
   ExemplarFilterType exemplar_filter_type_;
   nostd::shared_ptr<ExemplarReservoir> exemplar_reservoir_;
diff --git a/sdk/include/opentelemetry/sdk/metrics/view/view.h b/sdk/include/opentelemetry/sdk/metrics/view/view.h
@@ -53,10 +53,10 @@ class View
     return aggregation_config_.get();
   }
 
-  virtual const opentelemetry::sdk::metrics::AttributesProcessor &GetAttributesProcessor()
-      const noexcept
+  virtual std::shared_ptr<const opentelemetry::sdk::metrics::AttributesProcessor>
+  GetAttributesProcessor() const noexcept
   {
-    return *attributes_processor_.get();
+    return attributes_processor_;
   }
 
 private:
@@ -65,7 +65,7 @@ class View
   std::string unit_;
   AggregationType aggregation_type_;
   std::shared_ptr<AggregationConfig> aggregation_config_;
-  std::unique_ptr<opentelemetry::sdk::metrics::AttributesProcessor> attributes_processor_;
+  std::shared_ptr<AttributesProcessor> attributes_processor_;
 };
 }  // namespace metrics
 }  // namespace sdk
diff --git a/sdk/src/metrics/meter.cc b/sdk/src/metrics/meter.cc
@@ -538,7 +538,7 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
         {
           WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc);
           sync_storage = std::shared_ptr<SyncMetricStorage>(new SyncMetricStorage(
-              view_instr_desc, view.GetAggregationType(), &view.GetAttributesProcessor(),
+              view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
               exemplar_filter_type,
               GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),
diff --git a/sdk/test/metrics/cardinality_limit_test.cc b/sdk/test/metrics/cardinality_limit_test.cc
@@ -109,9 +109,9 @@ TEST_P(WritableMetricStorageCardinalityLimitTestFixture, LongCounterSumAggregati
   const size_t attributes_limit   = 10;
   InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kCounter,
                                      InstrumentValueType::kLong};
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
-  SyncMetricStorage storage(instr_desc, AggregationType::kSum, default_attributes_processor.get(),
+  SyncMetricStorage storage(instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                             ExemplarFilterType::kAlwaysOff,
                             ExemplarReservoir::GetNoExemplarReservoir(),
diff --git a/sdk/test/metrics/meter_test.cc b/sdk/test/metrics/meter_test.cc
@@ -6,21 +6,25 @@
 #include <stdint.h>
 #include <algorithm>
 #include <atomic>
+#include <chrono>
 #include <iostream>
 #include <string>
 #include <thread>
+#include <type_traits>
 #include <utility>
 #include <vector>
 #include "common.h"
 
 #include <functional>
+#include "opentelemetry/common/key_value_iterable.h"
 #include "opentelemetry/context/context.h"
 #include "opentelemetry/metrics/async_instruments.h"
 #include "opentelemetry/metrics/meter.h"
 #include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
 #include "opentelemetry/sdk/instrumentationscope/scope_configurator.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/meter_config.h"
+#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 #include "opentelemetry/sdk/metrics/view/view_registry.h"
 #include "opentelemetry/sdk/resource/resource.h"
 
@@ -31,6 +35,7 @@
 #include "opentelemetry/metrics/sync_instruments.h"  // IWYU pragma: keep
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/shared_ptr.h"
+#include "opentelemetry/nostd/string_view.h"
 #include "opentelemetry/nostd/variant.h"
 #include "opentelemetry/sdk/common/attribute_utils.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
@@ -184,6 +189,22 @@ class MeterCreateInstrumentTest : public ::testing::Test
   std::shared_ptr<MetricReader> metric_reader_ptr_{new MockMetricReader()};
 };
 
+class TestProcessor : public sdk::metrics::AttributesProcessor
+{
+public:
+  explicit TestProcessor()  = default;
+  ~TestProcessor() override = default;
+
+  sdk::metrics::MetricAttributes process(
+      const opentelemetry::common::KeyValueIterable &attributes) const noexcept override
+  {
+    // Just forward attributes
+    return sdk::metrics::MetricAttributes(attributes);
+  }
+
+  bool isPresent(nostd::string_view /*key*/) const noexcept override { return true; }
+};
+
 }  // namespace
 
 TEST(MeterTest, BasicAsyncTests)
@@ -851,3 +872,46 @@ TEST_F(MeterCreateInstrumentTest, ViewCorrectedDuplicateAsyncInstrumentsByDescri
     return true;
   });
 }
+
+TEST(MeterTest, RecordAfterProviderDestructionWithCustomProcessor_NoResetInMain)
+{
+  std::unique_ptr<AttributesProcessor> processor(new TestProcessor());
+
+  // MeterProvider is owned by unique_ptr for explicit control
+  std::unique_ptr<MeterProvider> provider(new MeterProvider());
+
+  // Register a View with custom processor
+  std::unique_ptr<View> view(
+      new View("my_counter", "", "", AggregationType::kSum, nullptr, std::move(processor)));
+  std::unique_ptr<InstrumentSelector> instr_selector(
+      new InstrumentSelector(InstrumentType::kCounter, "my_counter", ""));
+  std::unique_ptr<MeterSelector> meter_selector(new MeterSelector("test_meter", "", ""));
+  provider->AddView(std::move(instr_selector), std::move(meter_selector), std::move(view));
+
+  auto meter   = provider->GetMeter("test_meter");
+  auto counter = meter->CreateUInt64Counter("my_counter");
+
+  // Move the counter to the thread
+  std::atomic<bool> thread_ready{false};
+  std::atomic<bool> thread_done{false};
+
+  std::thread t([c = std::move(counter), &thread_ready, &thread_done]() mutable {
+    thread_ready = true;
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    // Safe after provider destruction
+    c->Add(12345, {{"thread", "after_provider_destruction"}});
+    thread_done = true;
+  });
+
+  // Wait for thread to be ready
+  while (!thread_ready.load())
+    std::this_thread::yield();
+
+  // Destroy the provider (and its storage etc)
+  provider.reset();
+
+  // Wait for thread to finish
+  while (!thread_done.load())
+    std::this_thread::yield();
+  t.join();
+}
diff --git a/sdk/test/metrics/sync_metric_storage_counter_test.cc b/sdk/test/metrics/sync_metric_storage_counter_test.cc
@@ -49,10 +49,10 @@ TEST_P(WritableMetricStorageTestFixture, LongCounterSumAggregation)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kSum, default_attributes_processor.get(),
+      instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -189,10 +189,10 @@ TEST_P(WritableMetricStorageTestFixture, DoubleCounterSumAggregation)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kSum, default_attributes_processor.get(),
+      instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
diff --git a/sdk/test/metrics/sync_metric_storage_gauge_test.cc b/sdk/test/metrics/sync_metric_storage_gauge_test.cc
@@ -37,10 +37,10 @@ TEST_P(WritableMetricStorageTestFixture, LongGaugeLastValueAggregation)
   std::map<std::string, std::string> attributes_roomA = {{"Room.id", "Rack A"}};
   std::map<std::string, std::string> attributes_roomB = {{"Room.id", "Rack B"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kLastValue, default_attributes_processor.get(),
+      instr_desc, AggregationType::kLastValue, default_attributes_processor,
 #  ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #  endif
@@ -121,10 +121,10 @@ TEST_P(WritableMetricStorageTestFixture, DoubleGaugeLastValueAggregation)
   std::map<std::string, std::string> attributes_roomA = {{"Room.id", "Rack A"}};
   std::map<std::string, std::string> attributes_roomB = {{"Room.id", "Rack B"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kLastValue, default_attributes_processor.get(),
+      instr_desc, AggregationType::kLastValue, default_attributes_processor,
 #  ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #  endif
diff --git a/sdk/test/metrics/sync_metric_storage_histogram_test.cc b/sdk/test/metrics/sync_metric_storage_histogram_test.cc
@@ -51,10 +51,10 @@ TEST_P(WritableMetricStorageHistogramTestFixture, LongHistogram)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kHistogram, default_attributes_processor.get(),
+      instr_desc, AggregationType::kHistogram, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -192,10 +192,10 @@ TEST_P(WritableMetricStorageHistogramTestFixture, DoubleHistogram)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kHistogram, default_attributes_processor.get(),
+      instr_desc, AggregationType::kHistogram, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -340,10 +340,10 @@ TEST_P(WritableMetricStorageHistogramTestFixture, Base2ExponentialDoubleHistogra
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kBase2ExponentialHistogram, default_attributes_processor.get(),
+      instr_desc, AggregationType::kBase2ExponentialHistogram, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
diff --git a/sdk/test/metrics/sync_metric_storage_up_down_counter_test.cc b/sdk/test/metrics/sync_metric_storage_up_down_counter_test.cc
@@ -48,10 +48,10 @@ TEST_P(WritableMetricStorageTestFixture, LongUpDownCounterSumAggregation)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kSum, default_attributes_processor.get(),
+      instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -198,10 +198,10 @@ TEST_P(WritableMetricStorageTestFixture, DoubleUpDownCounterSumAggregation)
   std::map<std::string, std::string> attributes_get = {{"RequestType", "GET"}};
   std::map<std::string, std::string> attributes_put = {{"RequestType", "PUT"}};
 
-  std::unique_ptr<DefaultAttributesProcessor> default_attributes_processor{
+  std::shared_ptr<DefaultAttributesProcessor> default_attributes_processor{
       new DefaultAttributesProcessor{}};
   opentelemetry::sdk::metrics::SyncMetricStorage storage(
-      instr_desc, AggregationType::kSum, default_attributes_processor.get(),
+      instr_desc, AggregationType::kSum, default_attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif

Original file line number	Diff line number	Diff line change
`@@ -538,7 +538,7 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(`
`538`	`538`	`{`
`539`	`539`	`WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc);`
`540`	`540`	`sync_storage = std::shared_ptr<SyncMetricStorage>(new SyncMetricStorage(`
`541`		`- view_instr_desc, view.GetAggregationType(), &view.GetAttributesProcessor(),`
	`541`	`+ view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor(),`
`542`	`542`	`#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW`
`543`	`543`	`exemplar_filter_type,`
`544`	`544`	`GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),`