Merge main into async-changes

CHANGELOG.md

@@ -17,6 +17,7 @@ Increment the:
 
 * [SDK] Bugfix: span SetAttribute crash ([#1283](https://github.com/open-telemetry/opentelemetry-cpp/pull/1283))
 * [EXPORTER] Jaeger Exporter - Populate Span Links ([#1251](https://github.com/open-telemetry/opentelemetry-cpp/pull/1251))
+* [EXPORTER] OTLP http exporter allow concurrency session ([#1209](https://github.com/open-telemetry/opentelemetry-cpp/pull/1209))
 
 ## [1.2.0] 2022-01-31
 

CMakeLists.txt

@@ -226,6 +226,12 @@ if(WITH_LOGS_PREVIEW)
   add_definitions(-DENABLE_LOGS_PREVIEW)
 endif()
 
+option(WITH_ASYNC_EXPORT_PREVIEW "Whether enable async export" OFF)
+
+if(WITH_ASYNC_EXPORT_PREVIEW)
+  add_definitions(-DENABLE_ASYNC_EXPORT)
+endif()
+
 find_package(Threads)
 
 function(install_windows_deps)

api/CMakeLists.txt

@@ -63,3 +63,7 @@ endif()
 if(CORE_RUNTIME_LIBS)
   target_link_libraries(opentelemetry_api INTERFACE ${CORE_RUNTIME_LIBS})
 endif()
+
+if(WITH_ASYNC_EXPORT_PREVIEW)
+  target_compile_definitions(opentelemetry_api INTERFACE ENABLE_ASYNC_EXPORT)
+endif()

api/include/opentelemetry/common/spin_lock_mutex.h

@@ -59,6 +59,26 @@ class SpinLockMutex
   SpinLockMutex &operator=(const SpinLockMutex &) = delete;
   SpinLockMutex &operator=(const SpinLockMutex &) volatile = delete;
 
+  static inline void fast_yield() noexcept
+  {
+// Issue a Pause/Yield instruction while spinning.
+#if defined(_MSC_VER)
+    YieldProcessor();
+#elif defined(__i386__) || defined(__x86_64__)
+#  if defined(__clang__)
+    _mm_pause();
+#  else
+    __builtin_ia32_pause();
+#  endif
+#elif defined(__arm__)
+    // This intrinsic should fail to be found if YIELD is not supported on the current
+    // processor.
+    __yield();
+#else
+    // TODO: Issue PAGE/YIELD on other architectures.
+#endif
+  }
+
   /**
    * Attempts to lock the mutex.  Return immediately with `true` (success) or `false` (failure).
    */
@@ -91,22 +111,7 @@ class SpinLockMutex
         {
           return;
         }
-// Issue a Pause/Yield instruction while spinning.
-#if defined(_MSC_VER)
-        YieldProcessor();
-#elif defined(__i386__) || defined(__x86_64__)
-#  if defined(__clang__)
-        _mm_pause();
-#  else
-        __builtin_ia32_pause();
-#  endif
-#elif defined(__arm__)
-        // This intrinsic should fail to be found if YIELD is not supported on the current
-        // processor.
-        __yield();
-#else
-        // TODO: Issue PAGE/YIELD on other architectures.
-#endif
+        fast_yield();
       }
       // Yield then try again (goal ~100ns)
       std::this_thread::yield();

api/include/opentelemetry/common/timestamp.h

@@ -169,5 +169,39 @@ class SteadyTimestamp
 private:
   int64_t nanos_since_epoch_;
 };
+
+class DurationUtil
+{
+public:
+  template <class Rep, class Period>
+  static std::chrono::duration<Rep, Period> AdjustWaitForTimeout(
+      std::chrono::duration<Rep, Period> timeout,
+      std::chrono::duration<Rep, Period> indefinite_value) noexcept
+  {
+    // Do not call now() when this duration is max value, now() may have a expensive cost.
+    if (timeout == std::chrono::duration<Rep, Period>::max())
+    {
+      return indefinite_value;
+    }
+
+    // std::future<T>::wait_for, std::this_thread::sleep_for, and std::condition_variable::wait_for
+    // may use steady_clock or system_clock.We need make sure now() + timeout do not overflow.
+    auto max_timeout = std::chrono::duration_cast<std::chrono::duration<Rep, Period>>(
+        std::chrono::steady_clock::time_point::max() - std::chrono::steady_clock::now());
+    if (timeout >= max_timeout)
+    {
+      return indefinite_value;
+    }
+    max_timeout = std::chrono::duration_cast<std::chrono::duration<Rep, Period>>(
+        std::chrono::system_clock::time_point::max() - std::chrono::system_clock::now());
+    if (timeout >= max_timeout)
+    {
+      return indefinite_value;
+    }
+
+    return timeout;
+  }
+};
+
 }  // namespace common
 OPENTELEMETRY_END_NAMESPACE

exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_exporter.h

@@ -89,6 +89,19 @@ class ElasticsearchLogExporter final : public opentelemetry::sdk::logs::LogExpor
       const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
           &records) noexcept override;
 
+#  ifdef ENABLE_ASYNC_EXPORT
+  /**
+   * Exports a vector of log records to the Elasticsearch instance asynchronously.
+   * @param records A list of log records to send to Elasticsearch.
+   * @param result_callback callback function accepting ExportResult as argument
+   */
+  void Export(
+      const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
+          &records,
+      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept
+      override;
+#  endif
+
   /**
    * Shutdown this exporter.
    * @param timeout The maximum time to wait for the shutdown method to return

exporters/elasticsearch/src/es_log_exporter.cc

@@ -110,6 +110,91 @@ class ResponseHandler : public http_client::EventHandler
   bool console_debug_ = false;
 };
 
+#  ifdef ENABLE_ASYNC_EXPORT
+/**
+ * This class handles the async response message from the Elasticsearch request
+ */
+class AsyncResponseHandler : public http_client::EventHandler
+{
+public:
+  /**
+   * Creates a response handler, that by default doesn't display to console
+   */
+  AsyncResponseHandler(
+      std::shared_ptr<ext::http::client::Session> session,
+      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback,
+      bool console_debug = false)
+      : console_debug_{console_debug},
+        session_{std::move(session)},
+        result_callback_{std::move(result_callback)}
+  {}
+
+  /**
+   * Cleans up the session in the destructor.
+   */
+  ~AsyncResponseHandler() { session_->FinishSession(); }
+
+  /**
+   * Automatically called when the response is received
+   */
+  void OnResponse(http_client::Response &response) noexcept override
+  {
+
+    // Store the body of the request
+    body_ = std::string(response.GetBody().begin(), response.GetBody().end());
+    if (body_.find("\"failed\" : 0") == std::string::npos)
+    {
+      OTEL_INTERNAL_LOG_ERROR(
+          "[ES Trace Exporter] Logs were not written to Elasticsearch correctly, response body: "
+          << body_);
+      result_callback_(sdk::common::ExportResult::kFailure);
+    }
+    else
+    {
+      result_callback_(sdk::common::ExportResult::kSuccess);
+    }
+  }
+
+  // Callback method when an http event occurs
+  void OnEvent(http_client::SessionState state, nostd::string_view reason) noexcept override
+  {
+    // If any failure event occurs, release the condition variable to unblock main thread
+    switch (state)
+    {
+      case http_client::SessionState::ConnectFailed:
+        OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Connection to elasticsearch failed");
+        break;
+      case http_client::SessionState::SendFailed:
+        OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Request failed to be sent to elasticsearch");
+
+        break;
+      case http_client::SessionState::TimedOut:
+        OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Request to elasticsearch timed out");
+
+        break;
+      case http_client::SessionState::NetworkError:
+        OTEL_INTERNAL_LOG_ERROR("[ES Trace Exporter] Network error to elasticsearch");
+        break;
+      default:
+        break;
+    }
+    result_callback_(sdk::common::ExportResult::kFailure);
+  }
+
+private:
+  // Stores the session object for the request
+  std::shared_ptr<ext::http::client::Session> session_;
+  // Callback to call to on receiving events
+  std::function<bool(opentelemetry::sdk::common::ExportResult)> result_callback_;
+
+  // A string to store the response body
+  std::string body_ = "";
+
+  // Whether to print the results from the callback
+  bool console_debug_ = false;
+};
+#  endif
+
 ElasticsearchLogExporter::ElasticsearchLogExporter()
     : options_{ElasticsearchExporterOptions()},
       http_client_{new ext::http::client::curl::HttpClient()}
@@ -162,8 +247,8 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export(
   request->SetBody(body_vec);
 
   // Send the request
-  std::unique_ptr<ResponseHandler> handler(new ResponseHandler(options_.console_debug_));
-  session->SendRequest(*handler);
+  auto handler = std::make_shared<ResponseHandler>(options_.console_debug_);
+  session->SendRequest(handler);
 
   // Wait for the response to be received
   if (options_.console_debug_)
@@ -198,6 +283,53 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export(
   return sdk::common::ExportResult::kSuccess;
 }
 
+#  ifdef ENABLE_ASYNC_EXPORT
+void ElasticsearchLogExporter::Export(
+    const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
+        &records,
+    std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept
+{
+  // Return failure if this exporter has been shutdown
+  if (isShutdown())
+  {
+    OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
+                            << records.size() << " log(s) failed, exporter is shutdown");
+    return;
+  }
+
+  // Create a connection to the ElasticSearch instance
+  auto session = http_client_->CreateSession(options_.host_ + std::to_string(options_.port_));
+  auto request = session->CreateRequest();
+
+  // Populate the request with headers and methods
+  request->SetUri(options_.index_ + "/_bulk?pretty");
+  request->SetMethod(http_client::Method::Post);
+  request->AddHeader("Content-Type", "application/json");
+  request->SetTimeoutMs(std::chrono::milliseconds(1000 * options_.response_timeout_));
+
+  // Create the request body
+  std::string body = "";
+  for (auto &record : records)
+  {
+    // Append {"index":{}} before JSON body, which tells Elasticsearch to write to index specified
+    // in URI
+    body += "{\"index\" : {}}\n";
+
+    // Add the context of the Recordable
+    auto json_record = std::unique_ptr<ElasticSearchRecordable>(
+        static_cast<ElasticSearchRecordable *>(record.release()));
+    body += json_record->GetJSON().dump() + "\n";
+  }
+  std::vector<uint8_t> body_vec(body.begin(), body.end());
+  request->SetBody(body_vec);
+
+  // Send the request
+  auto handler = std::make_shared<AsyncResponseHandler>(session, std::move(result_callback),
+                                                        options_.console_debug_);
+  session->SendRequest(handler);
+}
+#  endif
+
 bool ElasticsearchLogExporter::Shutdown(std::chrono::microseconds timeout) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h

@@ -61,6 +61,17 @@ class JaegerExporter final : public opentelemetry::sdk::trace::SpanExporter
       const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans) noexcept
       override;
 
+#ifdef ENABLE_ASYNC_EXPORT
+  /**
+   * Exports a batch of span recordables asynchronously.
+   * @param spans a span of unique pointers to span recordables
+   * @param result_callback callback function accepting ExportResult as argument
+   */
+  void Export(const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans,
+              std::function<bool(opentelemetry::sdk::common::ExportResult)>
+                  &&result_callback) noexcept override;
+#endif
+
   /**
    * Shutdown the exporter.
    * @param timeout an option timeout, default to max.

exporters/jaeger/src/jaeger_exporter.cc

@@ -70,6 +70,17 @@ sdk_common::ExportResult JaegerExporter::Export(
   return sdk_common::ExportResult::kSuccess;
 }
 
+#ifdef ENABLE_ASYNC_EXPORT
+void JaegerExporter::Export(
+    const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &spans,
+    std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept
+{
+  OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call");
+  auto status = Export(spans);
+  result_callback(status);
+}
+#endif
+
 void JaegerExporter::InitializeEndpoint()
 {
   if (options_.transport_format == TransportFormat::kThriftUdpCompact)

exporters/memory/include/opentelemetry/exporters/memory/in_memory_span_exporter.h

@@ -64,6 +64,22 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte
     return sdk::common::ExportResult::kSuccess;
   }
 
+#ifdef ENABLE_ASYNC_EXPORT
+  /**
+   * Exports a batch of span recordables asynchronously.
+   * @param spans a span of unique pointers to span recordables
+   * @param result_callback callback function accepting ExportResult as argument
+   */
+  void Export(const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &spans,
+              std::function<bool(opentelemetry::sdk::common::ExportResult)>
+                  &&result_callback) noexcept override
+  {
+    OTEL_INTERNAL_LOG_WARN(" async not supported. Making sync interface call");
+    auto status = Export(spans);
+    result_callback(status);
+  }
+#endif
+
   /**
    * @param timeout an optional value containing the timeout of the exporter
    * note: passing custom timeout values is not currently supported for this exporter

exporters/ostream/include/opentelemetry/exporters/ostream/log_exporter.h

@@ -39,6 +39,15 @@ class OStreamLogExporter final : public opentelemetry::sdk::logs::LogExporter
       const opentelemetry::nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records) noexcept
       override;
 
+#  ifdef ENABLE_ASYNC_EXPORT
+  /**
+   * Exports a span of logs sent from the processor asynchronously.
+   */
+  void Export(
+      const opentelemetry::nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records,
+      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept;
+#  endif
+
   /**
    * Marks the OStream Log Exporter as shut down.
    */

exporters/ostream/include/opentelemetry/exporters/ostream/span_exporter.h

@@ -38,6 +38,14 @@ class OStreamSpanExporter final : public opentelemetry::sdk::trace::SpanExporter
       const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>>
           &spans) noexcept override;
 
+#ifdef ENABLE_ASYNC_EXPORT
+  void Export(
+      const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>>
+          &spans,
+      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept
+      override;
+#endif
+
   bool Shutdown(
       std::chrono::microseconds timeout = std::chrono::microseconds::max()) noexcept override;