[SDK] fix: fixed size exemplar reservoir works correctly

CHANGELOG.md

@@ -223,6 +223,12 @@ Increment the:
   `AlwaysOff`/`TraceBased`)
   [#4267](https://github.com/open-telemetry/opentelemetry-cpp/pull/4267)
 
+* [METRICS SDK] Fix preview exemplar reservoirs to serialize concurrent
+  measurement offers and collection, reset stored cells and sampling state
+  between collection intervals, remain usable after collection, and omit empty
+  cells from collected results.
+  [#4429](https://github.com/open-telemetry/opentelemetry-cpp/pull/4429)
+
 Important changes:
 
 * [API] Never set a null global provider or propagator

sdk/include/opentelemetry/sdk/metrics/exemplar/fixed_size_exemplar_reservoir.h

@@ -6,6 +6,8 @@
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
 
 #  include <memory>
+#  include <mutex>
+#  include <utility>
 #  include <vector>
 
 #  include "opentelemetry/context/context.h"
@@ -44,6 +46,9 @@ class FixedSizeExemplarReservoir : public ExemplarReservoir
     {
       return;
     }
+
+    std::lock_guard<std::mutex> lock{mutex_};
+
     auto idx =
         reservoir_cell_selector_->ReservoirCellIndexFor(storage_, value, attributes, context);
     if (idx != -1)
@@ -60,6 +65,9 @@ class FixedSizeExemplarReservoir : public ExemplarReservoir
     {
       return;
     }
+
+    std::lock_guard<std::mutex> lock{mutex_};
+
     auto idx =
         reservoir_cell_selector_->ReservoirCellIndexFor(storage_, value, attributes, context);
     if (idx != -1)
@@ -76,17 +84,26 @@ class FixedSizeExemplarReservoir : public ExemplarReservoir
     {
       return results;
     }
+
+    std::lock_guard<std::mutex> lock{mutex_};
+
     if (!map_and_reset_cell_)
     {
-      reservoir_cell_selector_.reset();
+      reservoir_cell_selector_->reset();
       return results;
     }
-    for (auto reservoirCell : storage_)
+
+    results.reserve(storage_.size());
+    for (auto &reservoirCell : storage_)
     {
       auto result = (reservoirCell.*(map_and_reset_cell_))(pointAttributes);
-      results.push_back(result);
+      if (result)
+      {
+        results.emplace_back(std::move(result));
+      }
     }
-    reservoir_cell_selector_.reset();
+
+    reservoir_cell_selector_->reset();
     return results;
   }
 
@@ -95,6 +112,7 @@ class FixedSizeExemplarReservoir : public ExemplarReservoir
   std::vector<ReservoirCell> storage_;
   std::shared_ptr<ReservoirCellSelector> reservoir_cell_selector_;
   MapAndResetCellType map_and_reset_cell_{nullptr};
+  std::mutex mutex_;
 };
 
 }  // namespace metrics

sdk/include/opentelemetry/sdk/metrics/exemplar/simple_fixed_size_exemplar_reservoir.h

@@ -88,11 +88,12 @@ class SimpleFixedSizeExemplarReservoir : public FixedSizeExemplarReservoir
       return static_cast<int>(index);
     }
 
-    void reset() override {}
+    void reset() override { measurements_seen_ = 0; }
 
   private:
     size_t measurements_seen_ = 0;
     size_t size_;
+    friend class SimpleFixedSizeCellSelectorTestPeer;
   };  // class SimpleFixedSizeCellSelector
 
 };  // class SimpleFixedSizeExemplarReservoir

sdk/test/metrics/exemplar/BUILD

@@ -70,3 +70,20 @@ cc_test(
         "@com_google_googletest//:gtest_main",
     ],
 )
+
+cc_test(
+    name = "fixed_size_exemplar_reservoir_test",
+    srcs = [
+        "fixed_size_exemplar_reservoir_test.cc",
+    ],
+    tags = [
+        "metrics",
+        "test",
+    ],
+    deps = [
+        "//api",
+        "//sdk:headers",
+        "//sdk/src/metrics",
+        "@com_google_googletest//:gtest_main",
+    ],
+)

sdk/test/metrics/exemplar/CMakeLists.txt

@@ -4,7 +4,7 @@
 foreach(
   testname
   no_exemplar_reservoir_test aligned_histogram_bucket_exemplar_reservoir_test
-  reservoir_cell_test filter_predicate_test)
+  reservoir_cell_test filter_predicate_test fixed_size_exemplar_reservoir_test)
   add_executable(${testname} "${testname}.cc")
   target_link_libraries(
     ${testname} ${GTEST_BOTH_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT}

sdk/test/metrics/exemplar/fixed_size_exemplar_reservoir_test.cc

@@ -0,0 +1,178 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+
+#  include <gtest/gtest.h>
+#  include <stdint.h>
+#  include <atomic>
+#  include <cstddef>
+#  include <memory>
+#  include <string>
+#  include <thread>
+#  include <vector>
+
+#  include "opentelemetry/context/context.h"
+#  include "opentelemetry/sdk/metrics/data/exemplar_data.h"
+#  include "opentelemetry/sdk/metrics/exemplar/aligned_histogram_bucket_exemplar_reservoir.h"
+#  include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
+#  include "opentelemetry/sdk/metrics/exemplar/reservoir_cell.h"
+#  include "opentelemetry/sdk/metrics/exemplar/reservoir_cell_selector.h"
+#  include "opentelemetry/sdk/metrics/exemplar/simple_fixed_size_exemplar_reservoir.h"
+#  include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+class SimpleFixedSizeCellSelectorTestPeer
+{
+public:
+  static size_t GetMeasurementsSeen(
+      const SimpleFixedSizeExemplarReservoir::SimpleFixedSizeCellSelector &selector)
+  {
+    return selector.measurements_seen_;
+  }
+};
+
+namespace
+{
+
+TEST(FixedSizeExemplarReservoirTest, CollectAndResetSupportsMultipleIntervals)
+{
+  std::vector<double> boundaries{1.0, 5.0, 10.0};
+  auto reservoir = ExemplarReservoir::GetAlignedHistogramBucketExemplarReservoir(
+      boundaries.size(),
+      AlignedHistogramBucketExemplarReservoir::GetHistogramCellSelector(boundaries),
+      &ReservoirCell::GetAndResetDouble);
+
+  reservoir->OfferMeasurement(2.0, MetricAttributes{}, opentelemetry::context::Context{});
+  auto first_interval = reservoir->CollectAndReset(MetricAttributes{});
+  ASSERT_EQ(first_interval.size(), 1U);
+  EXPECT_NE(first_interval[0], nullptr);
+
+  // Collecting again without another offer verifies that the cells in storage,
+  // rather than copies of those cells, were reset.
+  EXPECT_TRUE(reservoir->CollectAndReset(MetricAttributes{}).empty());
+
+  // A collection resets the selector state without destroying the selector, so
+  // the same reservoir remains usable in the next interval.
+  reservoir->OfferMeasurement(8.0, MetricAttributes{}, opentelemetry::context::Context{});
+  auto second_interval = reservoir->CollectAndReset(MetricAttributes{});
+  ASSERT_EQ(second_interval.size(), 1U);
+  EXPECT_NE(second_interval[0], nullptr);
+}
+
+TEST(FixedSizeExemplarReservoirTest, SimpleReservoirRestartsSamplingEachInterval)
+{
+  auto selector =
+      std::make_shared<SimpleFixedSizeExemplarReservoir::SimpleFixedSizeCellSelector>(1);
+  auto reservoir = ExemplarReservoir::GetSimpleFixedSizeExemplarReservoir(
+      1, selector, &ReservoirCell::GetAndResetLong);
+
+  reservoir->OfferMeasurement(static_cast<int64_t>(1), MetricAttributes{},
+                              opentelemetry::context::Context{});
+  ASSERT_EQ(SimpleFixedSizeCellSelectorTestPeer::GetMeasurementsSeen(*selector), 1U);
+
+  auto exemplars = reservoir->CollectAndReset(MetricAttributes{});
+  ASSERT_EQ(exemplars.size(), 1U);
+  EXPECT_NE(exemplars[0], nullptr);
+  EXPECT_EQ(SimpleFixedSizeCellSelectorTestPeer::GetMeasurementsSeen(*selector), 0U);
+}
+
+class ConcurrentAccessDetector final : public ReservoirCellSelector
+{
+public:
+  int ReservoirCellIndexFor(const std::vector<ReservoirCell> & /* cells */,
+                            int64_t /* value */,
+                            const MetricAttributes & /* attributes */,
+                            const opentelemetry::context::Context & /* context */) override
+  {
+    Visit();
+    return 0;
+  }
+
+  int ReservoirCellIndexFor(const std::vector<ReservoirCell> & /* cells */,
+                            double /* value */,
+                            const MetricAttributes & /* attributes */,
+                            const opentelemetry::context::Context & /* context */) override
+  {
+    Visit();
+    return 0;
+  }
+
+  void reset() override { Visit(); }
+
+  bool HasConcurrentAccess() const noexcept
+  {
+    return concurrent_access_.load(std::memory_order_relaxed);
+  }
+
+private:
+  void Visit()
+  {
+    if (active_calls_.fetch_add(1, std::memory_order_acq_rel) != 0)
+    {
+      concurrent_access_.store(true, std::memory_order_relaxed);
+    }
+    std::this_thread::yield();
+    active_calls_.fetch_sub(1, std::memory_order_release);
+  }
+
+  std::atomic<int> active_calls_{0};
+  std::atomic<bool> concurrent_access_{false};
+};
+
+TEST(FixedSizeExemplarReservoirTest, SerializesOffersAndCollection)
+{
+  auto selector  = std::make_shared<ConcurrentAccessDetector>();
+  auto reservoir = ExemplarReservoir::GetSimpleFixedSizeExemplarReservoir(
+      1, selector, &ReservoirCell::GetAndResetLong);
+  std::atomic<bool> start{false};
+
+  constexpr size_t kOfferThreadCount = 3;
+  constexpr size_t kIterations       = 500;
+  std::vector<std::thread> threads;
+  threads.reserve(kOfferThreadCount + 1);
+  for (size_t thread_index = 0; thread_index < kOfferThreadCount; ++thread_index)
+  {
+    threads.emplace_back([&] {
+      while (!start.load(std::memory_order_acquire))
+      {
+        std::this_thread::yield();
+      }
+      for (size_t i = 0; i < kIterations; ++i)
+      {
+        reservoir->OfferMeasurement(static_cast<int64_t>(i), MetricAttributes{},
+                                    opentelemetry::context::Context{});
+      }
+    });
+  }
+  threads.emplace_back([&] {
+    while (!start.load(std::memory_order_acquire))
+    {
+      std::this_thread::yield();
+    }
+    for (size_t i = 0; i < kIterations; ++i)
+    {
+      reservoir->CollectAndReset(MetricAttributes{});
+    }
+  });
+
+  start.store(true, std::memory_order_release);
+  for (auto &thread : threads)
+  {
+    thread.join();
+  }
+
+  EXPECT_FALSE(selector->HasConcurrentAccess());
+}
+
+}  // namespace
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+
+#endif  // ENABLE_METRICS_EXEMPLAR_PREVIEW