Add ForceFlush for all LogRecordExporters and SpanExporters. (#2000)

Author: owent

CHANGELOG.md

@@ -21,6 +21,8 @@ Increment the:
   [#2060](https://github.com/open-telemetry/opentelemetry-cpp/pull/2060)
 * [BUILD] Restore detfault value of `OPENTELEMETRY_INSTALL` to `ON` when it's on
   top level.[#2062](https://github.com/open-telemetry/opentelemetry-cpp/pull/2062)
+* [EXPORTERS]Add `ForceFlush` for `LogRecordExporter` and `SpanExporter`
+  [#2000](https://github.com/open-telemetry/opentelemetry-cpp/pull/2000)
 
 Important changes:
 
@@ -44,6 +46,12 @@ Important changes:
     * As a result, a behavior change for GRPC SSL is possible,
       because the endpoint scheme now takes precedence.
       Please verify configuration settings for the GRPC endpoint.
+* [EXPORTERS]Add `ForceFlush` for `LogRecordExporter` and `SpanExporter`
+  [#2000](https://github.com/open-telemetry/opentelemetry-cpp/pull/2000)
+  * `LogRecordExporter` and `SpanExporter` add a new virtual function
+    `ForceFlush`, and if users implement any customized `LogRecordExporter` and
+    `SpanExporter`, they should also implement this function.There should be no
+    influence if users only use factory to create exporters.
 
 ## [1.8.3] 2023-03-06
 

examples/otlp/grpc_log_main.cc

@@ -11,6 +11,11 @@
 #  include "opentelemetry/sdk/trace/tracer_provider_factory.h"
 #  include "opentelemetry/trace/provider.h"
 
+// sdk::TracerProvider and sdk::LoggerProvider is just used to call ForceFlush and prevent to cancel
+// running exportings when destroy and shutdown exporters.It's optional to users.
+#  include "opentelemetry/sdk/logs/logger_provider.h"
+#  include "opentelemetry/sdk/trace/tracer_provider.h"
+
 #  include <string>
 
 #  ifdef BAZEL_BUILD
@@ -42,6 +47,14 @@ void InitTracer()
 
 void CleanupTracer()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<opentelemetry::trace::TracerProvider> provider =
+      trace::Provider::GetTracerProvider();
+  if (provider)
+  {
+    static_cast<trace_sdk::TracerProvider *>(provider.get())->ForceFlush();
+  }
+
   std::shared_ptr<opentelemetry::trace::TracerProvider> none;
   trace::Provider::SetTracerProvider(none);
 }
@@ -59,6 +72,14 @@ void InitLogger()
 
 void CleanupLogger()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<logs::LoggerProvider> provider =
+      logs::Provider::GetLoggerProvider();
+  if (provider)
+  {
+    static_cast<logs_sdk::LoggerProvider *>(provider.get())->ForceFlush();
+  }
+
   nostd::shared_ptr<logs::LoggerProvider> none;
   opentelemetry::logs::Provider::SetLoggerProvider(none);
 }

examples/otlp/grpc_main.cc

@@ -6,6 +6,10 @@
 #include "opentelemetry/sdk/trace/tracer_provider_factory.h"
 #include "opentelemetry/trace/provider.h"
 
+// sdk::TracerProvider is just used to call ForceFlush and prevent to cancel running exportings when
+// destroy and shutdown exporters.It's optional to users.
+#include "opentelemetry/sdk/trace/tracer_provider.h"
+
 #ifdef BAZEL_BUILD
 #  include "examples/common/foo_library/foo_library.h"
 #else
@@ -32,6 +36,14 @@ void InitTracer()
 
 void CleanupTracer()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<opentelemetry::trace::TracerProvider> provider =
+      trace::Provider::GetTracerProvider();
+  if (provider)
+  {
+    static_cast<trace_sdk::TracerProvider *>(provider.get())->ForceFlush();
+  }
+
   std::shared_ptr<opentelemetry::trace::TracerProvider> none;
   trace::Provider::SetTracerProvider(none);
 }

examples/otlp/http_log_main.cc

@@ -13,6 +13,12 @@
 #  include "opentelemetry/sdk/trace/tracer_provider_factory.h"
 #  include "opentelemetry/trace/provider.h"
 
+// sdk::TracerProvider and sdk::LoggerProvider is just used to call ForceFlush and prevent to cancel
+// running exportings when destroy and shutdown exporters.It's optional to users.
+#  include "opentelemetry/sdk/logs/logger_provider.h"
+#  include "opentelemetry/sdk/trace/tracer_provider.h"
+
+#  include <iostream>
 #  include <string>
 
 #  ifdef BAZEL_BUILD
@@ -32,11 +38,27 @@ namespace internal_log = opentelemetry::sdk::common::internal_log;
 namespace
 {
 
-opentelemetry::exporter::otlp::OtlpHttpExporterOptions opts;
+opentelemetry::exporter::otlp::OtlpHttpExporterOptions trace_opts;
 void InitTracer()
 {
+  if (trace_opts.url.size() > 9)
+  {
+    if (trace_opts.url.substr(trace_opts.url.size() - 8) == "/v1/logs")
+    {
+      trace_opts.url = trace_opts.url.substr(0, trace_opts.url.size() - 8) + "/v1/traces";
+    }
+    else if (trace_opts.url.substr(trace_opts.url.size() - 9) == "/v1/logs/")
+    {
+      trace_opts.url = trace_opts.url.substr(0, trace_opts.url.size() - 9) + "/v1/traces";
+    }
+    else
+    {
+      trace_opts.url = opentelemetry::exporter::otlp::GetOtlpDefaultHttpTracesEndpoint();
+    }
+  }
+  std::cout << "Using " << trace_opts.url << " to export trace spans." << std::endl;
   // Create OTLP exporter instance
-  auto exporter  = otlp::OtlpHttpExporterFactory::Create(opts);
+  auto exporter  = otlp::OtlpHttpExporterFactory::Create(trace_opts);
   auto processor = trace_sdk::SimpleSpanProcessorFactory::Create(std::move(exporter));
   std::shared_ptr<opentelemetry::trace::TracerProvider> provider =
       trace_sdk::TracerProviderFactory::Create(std::move(processor));
@@ -46,13 +68,22 @@ void InitTracer()
 
 void CleanupTracer()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<opentelemetry::trace::TracerProvider> provider =
+      trace::Provider::GetTracerProvider();
+  if (provider)
+  {
+    static_cast<trace_sdk::TracerProvider *>(provider.get())->ForceFlush();
+  }
+
   std::shared_ptr<opentelemetry::trace::TracerProvider> none;
   trace::Provider::SetTracerProvider(none);
 }
 
 opentelemetry::exporter::otlp::OtlpHttpLogRecordExporterOptions logger_opts;
 void InitLogger()
 {
+  std::cout << "Using " << logger_opts.url << " to export log records." << std::endl;
   logger_opts.console_debug = true;
   // Create OTLP exporter instance
   auto exporter  = otlp::OtlpHttpLogRecordExporterFactory::Create(logger_opts);
@@ -65,6 +96,14 @@ void InitLogger()
 
 void CleanupLogger()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<logs::LoggerProvider> provider =
+      logs::Provider::GetLoggerProvider();
+  if (provider)
+  {
+    static_cast<logs_sdk::LoggerProvider *>(provider.get())->ForceFlush();
+  }
+
   std::shared_ptr<logs::LoggerProvider> none;
   opentelemetry::logs::Provider::SetLoggerProvider(none);
 }
@@ -83,26 +122,26 @@ int main(int argc, char *argv[])
 {
   if (argc > 1)
   {
-    opts.url        = argv[1];
+    trace_opts.url  = argv[1];
     logger_opts.url = argv[1];
     if (argc > 2)
     {
-      std::string debug  = argv[2];
-      opts.console_debug = debug != "" && debug != "0" && debug != "no";
+      std::string debug        = argv[2];
+      trace_opts.console_debug = debug != "" && debug != "0" && debug != "no";
     }
 
     if (argc > 3)
     {
       std::string binary_mode = argv[3];
       if (binary_mode.size() >= 3 && binary_mode.substr(0, 3) == "bin")
       {
-        opts.content_type        = opentelemetry::exporter::otlp::HttpRequestContentType::kBinary;
+        trace_opts.content_type  = opentelemetry::exporter::otlp::HttpRequestContentType::kBinary;
         logger_opts.content_type = opentelemetry::exporter::otlp::HttpRequestContentType::kBinary;
       }
     }
   }
 
-  if (opts.console_debug)
+  if (trace_opts.console_debug)
   {
     internal_log::GlobalLogHandler::SetLogLevel(internal_log::LogLevel::Debug);
   }

examples/otlp/http_main.cc

@@ -8,6 +8,10 @@
 #include "opentelemetry/sdk/trace/tracer_provider_factory.h"
 #include "opentelemetry/trace/provider.h"
 
+// sdk::TracerProvider is just used to call ForceFlush and prevent to cancel running exportings when
+// destroy and shutdown exporters.It's optional to users.
+#include "opentelemetry/sdk/trace/tracer_provider.h"
+
 #include <string>
 
 #ifdef BAZEL_BUILD
@@ -38,6 +42,14 @@ void InitTracer()
 
 void CleanupTracer()
 {
+  // We call ForceFlush to prevent to cancel running exportings, It's optional.
+  opentelemetry::nostd::shared_ptr<opentelemetry::trace::TracerProvider> provider =
+      trace::Provider::GetTracerProvider();
+  if (provider)
+  {
+    static_cast<trace_sdk::TracerProvider *>(provider.get())->ForceFlush();
+  }
+
   std::shared_ptr<opentelemetry::trace::TracerProvider> none;
   trace::Provider::SetTracerProvider(none);
 }

exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h

@@ -7,12 +7,17 @@
 #  include "nlohmann/json.hpp"
 #  include "opentelemetry/common/spin_lock_mutex.h"
 #  include "opentelemetry/ext/http/client/http_client_factory.h"
+#  include "opentelemetry/nostd/shared_ptr.h"
 #  include "opentelemetry/nostd/type_traits.h"
 #  include "opentelemetry/sdk/logs/exporter.h"
 #  include "opentelemetry/sdk/logs/recordable.h"
 
 #  include <time.h>
+#  include <atomic>
+#  include <condition_variable>
+#  include <cstddef>
 #  include <iostream>
+#  include <mutex>
 
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace exporter
@@ -90,6 +95,14 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
       const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
           &records) noexcept override;
 
+  /**
+   * Force flush the exporter.
+   * @param timeout an option timeout, default to max.
+   * @return return true when all data are exported, and false when timeout
+   */
+  bool ForceFlush(
+      std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override;
+
   /**
    * Shutdown this exporter.
    * @param timeout The maximum time to wait for the shutdown method to return
@@ -108,6 +121,18 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
   std::shared_ptr<ext::http::client::HttpClient> http_client_;
   mutable opentelemetry::common::SpinLockMutex lock_;
   bool isShutdown() const noexcept;
+
+#  ifdef ENABLE_ASYNC_EXPORT
+  struct SynchronizationData
+  {
+    std::atomic<std::size_t> session_counter_;
+    std::atomic<std::size_t> finished_session_counter_;
+    std::condition_variable force_flush_cv;
+    std::mutex force_flush_cv_m;
+    std::recursive_mutex force_flush_m;
+  };
+  nostd::shared_ptr<SynchronizationData> synchronization_data_;
+#  endif
 };
 }  // namespace logs
 }  // namespace exporter

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -290,9 +290,19 @@ class AsyncResponseHandler : public http_client::EventHandler
 #  endif
 
 ElasticsearchLogRecordExporter::ElasticsearchLogRecordExporter()
-    : options_{ElasticsearchExporterOptions()},
-      http_client_{ext::http::client::HttpClientFactory::Create()}
-{}
+    : options_{ElasticsearchExporterOptions()}, http_client_
+{
+  ext::http::client::HttpClientFactory::Create()
+}
+#  ifdef ENABLE_ASYNC_EXPORT
+, synchronization_data_(new SynchronizationData())
+#  endif
+{
+#  ifdef ENABLE_ASYNC_EXPORT
+  synchronization_data_->finished_session_counter_.store(0);
+  synchronization_data_->session_counter_.store(0);
+#  endif
+}
 
 ElasticsearchLogRecordExporter::ElasticsearchLogRecordExporter(
     const ElasticsearchExporterOptions &options)
@@ -343,10 +353,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
 
 #  ifdef ENABLE_ASYNC_EXPORT
   // Send the request
-  std::size_t span_count = records.size();
-  auto handler           = std::make_shared<AsyncResponseHandler>(
+  synchronization_data_->session_counter_.fetch_add(1, std::memory_order_release);
+  std::size_t span_count    = records.size();
+  auto synchronization_data = synchronization_data_;
+  auto handler              = std::make_shared<AsyncResponseHandler>(
       session,
-      [span_count](opentelemetry::sdk::common::ExportResult result) {
+      [span_count, synchronization_data](opentelemetry::sdk::common::ExportResult result) {
         if (result != opentelemetry::sdk::common::ExportResult::kSuccess)
         {
           OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] ERROR: Export "
@@ -358,6 +370,9 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
           OTEL_INTERNAL_LOG_DEBUG("[ES Log Exporter] Export " << span_count
                                                               << " trace span(s) success");
         }
+
+        synchronization_data->finished_session_counter_.fetch_add(1, std::memory_order_release);
+        synchronization_data->force_flush_cv.notify_all();
         return true;
       },
       options_.console_debug_);
@@ -401,6 +416,49 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
 #  endif
 }
 
+bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeout) noexcept
+{
+#  ifdef ENABLE_ASYNC_EXPORT
+  std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
+  std::size_t running_counter =
+      synchronization_data_->session_counter_.load(std::memory_order_acquire);
+  // ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented
+  //   in type 'long int' here. So we reset timeout to meet signed long int limit here.
+  timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
+      timeout, std::chrono::microseconds::zero());
+
+  std::chrono::steady_clock::duration timeout_steady =
+      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
+  if (timeout_steady <= std::chrono::steady_clock::duration::zero())
+  {
+    timeout_steady = std::chrono::steady_clock::duration::max();
+  }
+
+  std::unique_lock<std::mutex> lk_cv(synchronization_data_->force_flush_cv_m);
+  // Wait for all the sessions to finish
+  while (timeout_steady > std::chrono::steady_clock::duration::zero())
+  {
+    if (synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
+        running_counter)
+    {
+      break;
+    }
+
+    std::chrono::steady_clock::time_point start_timepoint = std::chrono::steady_clock::now();
+    if (std::cv_status::no_timeout != synchronization_data_->force_flush_cv.wait_for(
+                                          lk_cv, std::chrono::seconds{options_.response_timeout_}))
+    {
+      break;
+    }
+    timeout_steady -= std::chrono::steady_clock::now() - start_timepoint;
+  }
+
+  return timeout_steady > std::chrono::steady_clock::duration::zero();
+#  else
+  return true;
+#  endif
+}
+
 bool ElasticsearchLogRecordExporter::Shutdown(std::chrono::microseconds /* timeout */) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

exporters/etw/CMakeLists.txt

@@ -11,8 +11,12 @@ target_include_directories(
 set_target_properties(opentelemetry_exporter_etw PROPERTIES EXPORT_NAME
                                                             etw_exporter)
 
-target_link_libraries(opentelemetry_exporter_etw
-                      INTERFACE opentelemetry_api nlohmann_json::nlohmann_json)
+target_link_libraries(
+  opentelemetry_exporter_etw INTERFACE opentelemetry_api opentelemetry_trace
+                                       nlohmann_json::nlohmann_json)
+if(WITH_LOGS_PREVIEW)
+  target_link_libraries(opentelemetry_exporter_etw INTERFACE opentelemetry_logs)
+endif()
 if(nlohmann_json_clone)
   add_dependencies(opentelemetry_exporter_etw nlohmann_json::nlohmann_json)
 endif()