[SDK] fix: fixed size exemplar reservoir works correctly

CHANGELOG.md

@@ -230,6 +230,13 @@ 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. Correct the simple fixed-size reservoir's
+  sampling bounds so the current measurement can be discarded.
+  [#4429](https://github.com/open-telemetry/opentelemetry-cpp/pull/4429)
+
 Important changes:
 
 * [API] Never set a null global provider or propagator

bazel/otel_cc_benchmark.bzl

@@ -4,7 +4,7 @@
 load("@rules_cc//cc:cc_binary.bzl", "cc_binary")
 load("@rules_cc//cc:cc_test.bzl", "cc_test")
 
-def otel_cc_benchmark(name, srcs, deps, tags = [""]):
+def otel_cc_benchmark(name, srcs, deps, tags = [""], defines = []):
     """
     Creates targets for the benchmark and related targets.
 
@@ -30,7 +30,7 @@ def otel_cc_benchmark(name, srcs, deps, tags = [""]):
         srcs = srcs,
         deps = deps + ["@com_github_google_benchmark//:benchmark"],
         tags = tags + ["manual"],
-        defines = ["BAZEL_BUILD"],
+        defines = ["BAZEL_BUILD"] + defines,
     )
 
     # The result of running the benchmark, captured into a text file.
@@ -51,5 +51,5 @@ def otel_cc_benchmark(name, srcs, deps, tags = [""]):
         deps = deps + ["@com_github_google_benchmark//:benchmark"],
         args = ["--benchmark_min_time=1x"],
         tags = tags + ["benchmark"],
-        defines = ["BAZEL_BUILD"],
+        defines = ["BAZEL_BUILD"] + defines,
     )

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

@@ -5,6 +5,7 @@
 
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
 
+#  include <cstdint>
 #  include <memory>
 #  include <vector>
 
@@ -68,6 +69,11 @@ class SimpleFixedSizeExemplarReservoir : public FixedSizeExemplarReservoir
       // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#simplefixedsizeexemplarreservoir
       //
 
+      if (size_ == 0)
+      {
+        return -1;
+      }
+
       size_t measurement_num = measurements_seen_++;
       size_t index           = static_cast<size_t>(-1);
 
@@ -77,22 +83,26 @@ class SimpleFixedSizeExemplarReservoir : public FixedSizeExemplarReservoir
       }
       else
       {
-        size_t random_index = sdk::common::Random::GenerateRandom64() % measurement_num;
-
-        if (random_index < size_)
-        {
-          index = random_index;
-        }
+        return GetRandomCellIndex(size_, measurement_num, sdk::common::Random::GenerateRandom64());
       }
 
       return static_cast<int>(index);
     }
 
-    void reset() override {}
+    void reset() override { measurements_seen_ = 0; }
 
   private:
+    static int GetRandomCellIndex(size_t size,
+                                  size_t measurement_num,
+                                  uint64_t random_value) noexcept
+    {
+      size_t random_index = random_value % (measurement_num + 1);
+      return random_index < size ? static_cast<int>(random_index) : -1;
+    }
+
     size_t measurements_seen_ = 0;
     size_t size_;
+    friend class SimpleFixedSizeCellSelectorTestPeer;
   };  // class SimpleFixedSizeCellSelector
 
 };  // class SimpleFixedSizeExemplarReservoir

sdk/test/metrics/exemplar/BUILD

@@ -2,6 +2,25 @@
 # SPDX-License-Identifier: Apache-2.0
 
 load("@rules_cc//cc:cc_test.bzl", "cc_test")
+load("//bazel:otel_cc_benchmark.bzl", "otel_cc_benchmark")
+
+otel_cc_benchmark(
+    name = "fixed_size_exemplar_reservoir_benchmark",
+    srcs = [
+        "fixed_size_exemplar_reservoir_benchmark.cc",
+    ],
+    defines = ["ENABLE_METRICS_EXEMPLAR_PREVIEW"],
+    tags = [
+        "benchmark",
+        "metrics",
+        "test",
+    ],
+    deps = [
+        "//api",
+        "//sdk:headers",
+        "//sdk/src/common:random",
+    ],
+)
 
 cc_test(
     name = "no_exemplar_reservoir_test",
@@ -70,3 +89,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

@@ -1,10 +1,18 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+if(WITH_BENCHMARK)
+  add_executable(fixed_size_exemplar_reservoir_benchmark
+                 fixed_size_exemplar_reservoir_benchmark.cc)
+  target_link_libraries(
+    fixed_size_exemplar_reservoir_benchmark benchmark::benchmark
+    ${CMAKE_THREAD_LIBS_INIT} opentelemetry_common)
+endif()
+
 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_benchmark.cc

@@ -0,0 +1,93 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+
+#  include <benchmark/benchmark.h>
+#  include <cstdint>
+#  include <string>
+#  include <utility>
+
+#  include "opentelemetry/context/context.h"
+#  include "opentelemetry/nostd/shared_ptr.h"
+#  include "opentelemetry/sdk/metrics/data/exemplar_data.h"
+#  include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
+#  include "opentelemetry/sdk/metrics/exemplar/reservoir_cell.h"
+#  include "opentelemetry/sdk/metrics/exemplar/simple_fixed_size_exemplar_reservoir.h"
+
+namespace
+{
+
+using opentelemetry::context::Context;
+using opentelemetry::nostd::shared_ptr;
+using opentelemetry::sdk::metrics::ExemplarReservoir;
+using opentelemetry::sdk::metrics::MetricAttributes;
+using opentelemetry::sdk::metrics::ReservoirCell;
+using opentelemetry::sdk::metrics::SimpleFixedSizeExemplarReservoir;
+
+class SharedSimpleFixedSizeExemplarReservoirFixture : public benchmark::Fixture
+{
+public:
+  using benchmark::Fixture::SetUp;
+  using benchmark::Fixture::TearDown;
+
+  void SetUp(benchmark::State &state) override
+  {
+    if (state.thread_index() != 0)
+    {
+      return;
+    }
+
+    auto selector = SimpleFixedSizeExemplarReservoir::GetSimpleFixedSizeCellSelector(1);
+    reservoir_    = shared_ptr<ExemplarReservoir>{
+        new SimpleFixedSizeExemplarReservoir{1, selector, &ReservoirCell::GetAndResetDouble}};
+
+    // Model a reservoir partway through a normal collection interval. This
+    // keeps the benchmark focused on the steady-state offer path.
+    for (int64_t i = 0; i < 1024; ++i)
+    {
+      reservoir_->OfferMeasurement(static_cast<double>(i), attributes_, context_);
+    }
+  }
+
+  void TearDown(benchmark::State &state) override
+  {
+    if (state.thread_index() == 0)
+    {
+      reservoir_ = nullptr;
+    }
+  }
+
+protected:
+  shared_ptr<ExemplarReservoir> reservoir_;
+  MetricAttributes attributes_{{"http.request.method", "GET"},
+                               {"http.route", "/checkout"},
+                               {"http.response.status_code", int64_t{200}},
+                               {"service.version", "1.42.0"}};
+  Context context_;
+};
+
+BENCHMARK_DEFINE_F(SharedSimpleFixedSizeExemplarReservoirFixture, OfferMeasurement)
+(benchmark::State &state)
+{
+  double value = 10.0 + static_cast<double>(state.thread_index());
+  for (auto _ : state)
+  {
+    reservoir_->OfferMeasurement(value, attributes_, context_);
+    value += 0.001;
+  }
+
+  benchmark::DoNotOptimize(value);
+  state.SetItemsProcessed(state.iterations());
+}
+
+BENCHMARK_REGISTER_F(SharedSimpleFixedSizeExemplarReservoirFixture, OfferMeasurement)
+    ->ThreadRange(1, 8)
+    ->UseRealTime()
+    ->Unit(benchmark::kNanosecond);
+
+}  // namespace
+
+BENCHMARK_MAIN();
+
+#endif  // ENABLE_METRICS_EXEMPLAR_PREVIEW