Async callback Exporter to Processor changes (#1275)

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

@@ -89,6 +89,17 @@ class ElasticsearchLogExporter final : public opentelemetry::sdk::logs::LogExpor
       const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
           &records) noexcept override;
 
+  /**
+   * 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,
+      nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)> result_callback) noexcept
+      override;
+
   /**
    * 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,89 @@ class ResponseHandler : public http_client::EventHandler
   bool console_debug_ = false;
 };
 
+/**
+ * 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,
+      nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)> result_callback,
+      bool console_debug = false)
+      : console_debug_{console_debug},
+        session_{std::move(session)},
+        result_callback_{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
+  nostd::function_ref<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;
+};
+
 ElasticsearchLogExporter::ElasticsearchLogExporter()
     : options_{ElasticsearchExporterOptions()},
       http_client_{new ext::http::client::curl::HttpClient()}
@@ -162,8 +245,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 +281,51 @@ sdk::common::ExportResult ElasticsearchLogExporter::Export(
   return sdk::common::ExportResult::kSuccess;
 }
 
+void ElasticsearchLogExporter::Export(
+    const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>>
+        &records,
+    nostd::function_ref<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, result_callback, options_.console_debug_);
+  session->SendRequest(handler);
+}
+
 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,15 @@ class JaegerExporter final : public opentelemetry::sdk::trace::SpanExporter
       const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans) noexcept
       override;
 
+  /**
+   * 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,
+              nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)>
+                  result_callback) noexcept override;
+
   /**
    * Shutdown the exporter.
    * @param timeout an option timeout, default to max.

exporters/jaeger/src/jaeger_exporter.cc

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

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

@@ -64,6 +64,20 @@ class InMemorySpanExporter final : public opentelemetry::sdk::trace::SpanExporte
     return sdk::common::ExportResult::kSuccess;
   }
 
+  /**
+   * 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,
+      nostd::function_ref<bool(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);
+  }
+
   /**
    * @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,13 @@ class OStreamLogExporter final : public opentelemetry::sdk::logs::LogExporter
       const opentelemetry::nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records) noexcept
       override;
 
+  /**
+   * Exports a span of logs sent from the processor asynchronously.
+   */
+  void Export(const opentelemetry::nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records,
+              opentelemetry::nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)>
+                  result_callback) noexcept;
+
   /**
    * Marks the OStream Log Exporter as shut down.
    */

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

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

exporters/ostream/src/log_exporter.cc

@@ -180,6 +180,16 @@ sdk::common::ExportResult OStreamLogExporter::Export(
   return sdk::common::ExportResult::kSuccess;
 }
 
+void OStreamLogExporter::Export(
+    const opentelemetry::nostd::span<std::unique_ptr<sdk::logs::Recordable>> &records,
+    opentelemetry::nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)>
+        result_callback) noexcept
+{
+  // Do not have async support
+  auto result = Export(records);
+  result_callback(result);
+}
+
 bool OStreamLogExporter::Shutdown(std::chrono::microseconds) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

exporters/ostream/src/span_exporter.cc

@@ -96,6 +96,15 @@ sdk::common::ExportResult OStreamSpanExporter::Export(
   return sdk::common::ExportResult::kSuccess;
 }
 
+void OStreamSpanExporter::Export(
+    const opentelemetry::nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans,
+    opentelemetry::nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)>
+        result_callback) noexcept
+{
+  auto result = Export(spans);
+  result_callback(result);
+}
+
 bool OStreamSpanExporter::Shutdown(std::chrono::microseconds timeout) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h

@@ -52,6 +52,15 @@ class OtlpGrpcExporter final : public opentelemetry::sdk::trace::SpanExporter
   sdk::common::ExportResult Export(
       const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &spans) noexcept override;
 
+  /**
+   * 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
+   */
+  virtual void Export(
+      const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &spans,
+      nostd::function_ref<bool(sdk::common::ExportResult)> result_callback) noexcept override;
+
   /**
    * Shut down the exporter.
    * @param timeout an optional timeout, the default timeout of 0 means that no

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_exporter.h

@@ -55,6 +55,16 @@ class OtlpGrpcLogExporter : public opentelemetry::sdk::logs::LogExporter
       const nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>> &records) noexcept
       override;
 
+  /**
+   * Exports a vector of log records asynchronously.
+   * @param records A list of log records.
+   * @param result_callback callback function accepting ExportResult as argument
+   */
+  virtual void Export(
+      const nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>> &records,
+      nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)> result_callback) noexcept
+      override;
+
   /**
    * Shutdown this exporter.
    * @param timeout The maximum time to wait for the shutdown method to return.

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_client.h

@@ -141,7 +141,7 @@ class OtlpHttpClient
    */
   void addSession(
       std::shared_ptr<opentelemetry::ext::http::client::Session> session,
-      std::unique_ptr<opentelemetry::ext::http::client::EventHandler> &&event_handle) noexcept;
+      std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle) noexcept;
 
   /**
    * @brief Real delete all sessions and event handles.
@@ -168,13 +168,13 @@ class OtlpHttpClient
   struct HttpSessionData
   {
     std::shared_ptr<opentelemetry::ext::http::client::Session> session;
-    std::unique_ptr<opentelemetry::ext::http::client::EventHandler> event_handle;
+    std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle;
 
     inline HttpSessionData() = default;
 
     inline explicit HttpSessionData(
         std::shared_ptr<opentelemetry::ext::http::client::Session> &&input_session,
-        std::unique_ptr<opentelemetry::ext::http::client::EventHandler> &&input_handle)
+        std::shared_ptr<opentelemetry::ext::http::client::EventHandler> &&input_handle)
     {
       session.swap(input_session);
       event_handle.swap(input_handle);

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_exporter.h

@@ -86,6 +86,16 @@ class OtlpHttpExporter final : public opentelemetry::sdk::trace::SpanExporter
       const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans) noexcept
       override;
 
+  /**
+   * 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
+   */
+  virtual void Export(
+      const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans,
+      nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)> result_callback) noexcept
+      override;
+
   /**
    * Shut down the exporter.
    * @param timeout an optional timeout, the default timeout of 0 means that no

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_http_log_exporter.h

@@ -87,6 +87,16 @@ class OtlpHttpLogExporter final : public opentelemetry::sdk::logs::LogExporter
       const nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>> &records) noexcept
       override;
 
+  /**
+   * Exports a vector of log records asynchronously.
+   * @param records A list of log records.
+   * @param result_callback callback function accepting ExportResult as argument
+   */
+  virtual void Export(
+      const nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>> &records,
+      nostd::function_ref<bool(opentelemetry::sdk::common::ExportResult)> result_callback) noexcept
+      override;
+
   /**
    * Shutdown this exporter.
    * @param timeout The maximum time to wait for the shutdown method to return

exporters/otlp/src/otlp_grpc_exporter.cc

@@ -143,6 +143,16 @@ sdk::common::ExportResult OtlpGrpcExporter::Export(
   return sdk::common::ExportResult::kSuccess;
 }
 
+void OtlpGrpcExporter::Export(
+    const nostd::span<std::unique_ptr<sdk::trace::Recordable>> &spans,
+    nostd::function_ref<bool(sdk::common::ExportResult)> result_callback) noexcept
+{
+  OTEL_INTERNAL_LOG_WARN(
+      "[OTLP TRACE GRPC Exporter] async not supported. Making sync interface call");
+  auto status = Export(spans);
+  result_callback(status);
+}
+
 bool OtlpGrpcExporter::Shutdown(std::chrono::microseconds timeout) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);