Merge branch 'main' into fix_lifetime_in_log_record

Author: owent

.github/workflows/codeql-analysis.yml

@@ -39,10 +39,10 @@ jobs:
         sudo -E ./ci/setup_googletest.sh
         sudo -E ./ci/setup_ci_environment.sh
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16
+      uses: github/codeql-action/init@60168efe1c415ce0f5521ea06d5c2062adbeed1b # v3.28.17
       with:
        languages: cpp
     - name: Autobuild
-      uses: github/codeql-action/autobuild@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16
+      uses: github/codeql-action/autobuild@60168efe1c415ce0f5521ea06d5c2062adbeed1b # v3.28.17
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16
+      uses: github/codeql-action/analyze@60168efe1c415ce0f5521ea06d5c2062adbeed1b # v3.28.17

.github/workflows/ossf-scorecard.yml

@@ -47,6 +47,6 @@ jobs:
       # Upload the results to GitHub's code scanning dashboard (optional).
       # Commenting out will disable upload of results to your repo's Code Scanning dashboard
       - name: "Upload to code-scanning"
-        uses: github/codeql-action/upload-sarif@28deaeda66b76a05916b6923827895f2b14ab387 # v3.28.16
+        uses: github/codeql-action/upload-sarif@60168efe1c415ce0f5521ea06d5c2062adbeed1b # v3.28.17
         with:
           sarif_file: results.sarif

CHANGELOG.md

@@ -15,6 +15,9 @@ Increment the:
 
 ## [Unreleased]
 
+* [Metrics SDK] Use nostd::function_ref in AttributesHashMap
+  [#3393](https://github.com/open-telemetry/opentelemetry-cpp/pull/3393)
+
 * [SDK] Base2 exponential histogram aggregation
   [#3175](https://github.com/open-telemetry/opentelemetry-cpp/pull/3346)
 
@@ -33,6 +36,10 @@ Increment the:
 * [API] Add Enabled method to Tracer
   [#3357](https://github.com/open-telemetry/opentelemetry-cpp/pull/3357)
 
+
+* [SDK] Optimize PeriodicExportingMetricReader thread usage
+  [#3383](https://github.com/open-telemetry/opentelemetry-cpp/pull/3383)
+
 * [SDK] Fix lifetime for sdk::ReadWriteLogRecord
   [#3147](https://github.com/open-telemetry/opentelemetry-cpp/pull/3245)
 
@@ -49,6 +56,7 @@ Important changes:
     `const opentelemetry::sdk::common::OwnedAttributeValue &` instead of a
     `const opentelemetry::common::AttributeValue &`.
 
+
 ## [1.20 2025-04-01]
 
 * [BUILD] Update opentelemetry-proto version

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

@@ -74,9 +74,10 @@ class AttributesHashMapWithCustomHash
    * If not present, it uses the provided callback to generate
    * value and store in the hash
    */
-  Aggregation *GetOrSetDefault(const opentelemetry::common::KeyValueIterable &attributes,
-                               const AttributesProcessor *attributes_processor,
-                               std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
+  Aggregation *GetOrSetDefault(
+      const opentelemetry::common::KeyValueIterable &attributes,
+      const AttributesProcessor *attributes_processor,
+      nostd::function_ref<std::unique_ptr<Aggregation>()> aggregation_callback)
   {
     // TODO: avoid constructing MetricAttributes from KeyValueIterable for
     // hash_map_.find which is a heavy operation
@@ -97,8 +98,9 @@ class AttributesHashMapWithCustomHash
     return result.first->second.get();
   }
 
-  Aggregation *GetOrSetDefault(const MetricAttributes &attributes,
-                               std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
+  Aggregation *GetOrSetDefault(
+      const MetricAttributes &attributes,
+      nostd::function_ref<std::unique_ptr<Aggregation>()> aggregation_callback)
   {
     auto it = hash_map_.find(attributes);
     if (it != hash_map_.end())
@@ -115,8 +117,9 @@ class AttributesHashMapWithCustomHash
     return hash_map_[attributes].get();
   }
 
-  Aggregation *GetOrSetDefault(MetricAttributes &&attributes,
-                               std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
+  Aggregation *GetOrSetDefault(
+      MetricAttributes &&attributes,
+      nostd::function_ref<std::unique_ptr<Aggregation>()> aggregation_callback)
   {
     auto it = hash_map_.find(attributes);
     if (it != hash_map_.end())
@@ -207,7 +210,7 @@ class AttributesHashMapWithCustomHash
   size_t attributes_limit_;
 
   Aggregation *GetOrSetOveflowAttributes(
-      std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
+      nostd::function_ref<std::unique_ptr<Aggregation>()> aggregation_callback)
   {
     auto agg = aggregation_callback();
     return GetOrSetOveflowAttributes(std::move(agg));

sdk/src/metrics/export/periodic_exporting_metric_reader.cc

@@ -10,7 +10,6 @@
 #include <ostream>
 #include <ratio>
 #include <thread>
-#include <type_traits>
 #include <utility>
 
 #include "opentelemetry/common/timestamp.h"
@@ -24,13 +23,6 @@
 #include "opentelemetry/sdk/metrics/push_metric_exporter.h"
 #include "opentelemetry/version.h"
 
-#if defined(_MSC_VER)
-#  pragma warning(suppress : 5204)
-#  include <future>
-#else
-#  include <future>
-#endif
-
 #if OPENTELEMETRY_HAVE_EXCEPTIONS
 #  include <exception>
 #endif
@@ -98,11 +90,9 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
     worker_thread_instrumentation_->OnStart();
   }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
-
   do
   {
     auto start = std::chrono::steady_clock::now();
-
 #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
     if (worker_thread_instrumentation_ != nullptr)
     {
@@ -134,7 +124,6 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
       worker_thread_instrumentation_->BeforeWait();
     }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
-
     std::unique_lock<std::mutex> lk(cv_m_);
     cv_.wait_for(lk, remaining_wait_interval_ms, [this]() {
       if (is_force_wakeup_background_worker_.load(std::memory_order_acquire))
@@ -151,7 +140,6 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
       worker_thread_instrumentation_->AfterWait();
     }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
-
   } while (IsShutdown() != true);
 
 #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
@@ -164,61 +152,39 @@ void PeriodicExportingMetricReader::DoBackgroundWork()
 
 bool PeriodicExportingMetricReader::CollectAndExportOnce()
 {
-  std::atomic<bool> cancel_export_for_timeout{false};
-
   std::uint64_t notify_force_flush = force_flush_pending_sequence_.load(std::memory_order_acquire);
-  std::unique_ptr<std::thread> task_thread;
-
 #if OPENTELEMETRY_HAVE_EXCEPTIONS
   try
   {
 #endif
-    std::promise<void> sender;
-    auto receiver = sender.get_future();
-
-    task_thread.reset(
-        new std::thread([this, &cancel_export_for_timeout, sender = std::move(sender)] {
 #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
-          if (collect_thread_instrumentation_ != nullptr)
-          {
-            collect_thread_instrumentation_->OnStart();
-            collect_thread_instrumentation_->BeforeLoad();
-          }
+    if (collect_thread_instrumentation_ != nullptr)
+    {
+      collect_thread_instrumentation_->OnStart();
+      collect_thread_instrumentation_->BeforeLoad();
+    }
 #endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
-
-          this->Collect([this, &cancel_export_for_timeout](ResourceMetrics &metric_data) {
-            if (cancel_export_for_timeout.load(std::memory_order_acquire))
-            {
-              OTEL_INTERNAL_LOG_ERROR(
-                  "[Periodic Exporting Metric Reader] Collect took longer configured time: "
-                  << this->export_timeout_millis_.count() << " ms, and timed out");
-              return false;
-            }
-            this->exporter_->Export(metric_data);
-            return true;
-          });
-
-          const_cast<std::promise<void> &>(sender).set_value();
+    auto start = std::chrono::steady_clock::now();
+    this->Collect([this, &start](ResourceMetrics &metric_data) {
+      auto end = std::chrono::steady_clock::now();
+      if ((end - start) > this->export_timeout_millis_)
+      {
+        OTEL_INTERNAL_LOG_ERROR(
+            "[Periodic Exporting Metric Reader] Collect took longer configured time: "
+            << this->export_timeout_millis_.count() << " ms, and timed out");
+        return false;
+      }
+      this->exporter_->Export(metric_data);
+      return true;
+    });
 
 #ifdef ENABLE_THREAD_INSTRUMENTATION_PREVIEW
-          if (collect_thread_instrumentation_ != nullptr)
-          {
-            collect_thread_instrumentation_->AfterLoad();
-            collect_thread_instrumentation_->OnEnd();
-          }
-#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
-        }));
-
-    std::future_status status;
-    do
+    if (collect_thread_instrumentation_ != nullptr)
     {
-      status = receiver.wait_for(std::chrono::milliseconds(export_timeout_millis_));
-      if (status == std::future_status::timeout)
-      {
-        cancel_export_for_timeout.store(true, std::memory_order_release);
-        break;
-      }
-    } while (status != std::future_status::ready);
+      collect_thread_instrumentation_->AfterLoad();
+      collect_thread_instrumentation_->OnEnd();
+    }
+#endif /* ENABLE_THREAD_INSTRUMENTATION_PREVIEW */
 #if OPENTELEMETRY_HAVE_EXCEPTIONS
   }
   catch (std::exception &e)
@@ -235,11 +201,6 @@ bool PeriodicExportingMetricReader::CollectAndExportOnce()
   }
 #endif
 
-  if (task_thread && task_thread->joinable())
-  {
-    task_thread->join();
-  }
-
   std::uint64_t notified_sequence = force_flush_notified_sequence_.load(std::memory_order_acquire);
   while (notify_force_flush > notified_sequence)
   {

sdk/test/metrics/BUILD

@@ -49,6 +49,23 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "stress_tests",
+    timeout = "long",
+    srcs = glob(["*_test_stress.cc"]),
+    copts = [
+        "-DUNIT_TESTING",
+    ],
+    tags = [
+        "metrics",
+        "test",
+    ],
+    deps = [
+        "metrics_common_test_utils",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 otel_cc_benchmark(
     name = "attributes_processor_benchmark",
     srcs = [

sdk/test/metrics/CMakeLists.txt

@@ -33,7 +33,8 @@ foreach(
   metric_reader_test
   observable_registry_test
   periodic_exporting_metric_reader_test
-  instrument_metadata_validator_test)
+  instrument_metadata_validator_test
+  metric_test_stress)
   add_executable(${testname} "${testname}.cc")
   target_link_libraries(
     ${testname} ${GTEST_BOTH_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT}

sdk/test/metrics/attributes_hashmap_benchmark.cc

@@ -11,6 +11,7 @@
 #include <thread>
 #include <vector>
 
+#include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
 #include "opentelemetry/sdk/metrics/aggregation/drop_aggregation.h"
 #include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"