includes/opentelemetry/exporters/otlp/otlp_http_client.h

Go to the documentation of this file.

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
 
#pragma once
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <list>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
 
#include "opentelemetry/exporters/otlp/otlp_environment.h"
#include "opentelemetry/exporters/otlp/otlp_http.h"
#include "opentelemetry/ext/http/client/http_client.h"
#include "opentelemetry/nostd/string_view.h"
#include "opentelemetry/nostd/variant.h"
#include "opentelemetry/sdk/common/exporter_utils.h"
#include "opentelemetry/sdk/common/thread_instrumentation.h"
#include "opentelemetry/version.h"
 
// forward declare google::protobuf::Message
namespace google
{
namespace protobuf
{
class Message;
}
}  // namespace google
 
OPENTELEMETRY_BEGIN_NAMESPACE
namespace exporter
{
namespace otlp
{
// The default URL path to post metric data.
constexpr char kDefaultMetricsPath[] = "/v1/metrics";
// The HTTP header "Content-Type"
constexpr char kHttpJsonContentType[]   = "application/json";
constexpr char kHttpBinaryContentType[] = "application/x-protobuf";
 
struct OtlpHttpClientOptions
{
  std::string url;
 
  ext::http::client::HttpSslOptions ssl_options;
 
  // By default, post binary data
  HttpRequestContentType content_type = HttpRequestContentType::kBinary;
 
  // If convert bytes into hex. By default, we will convert all bytes but id into base64
  // This option is ignored if content_type is not kJson
  JsonBytesMappingKind json_bytes_mapping = JsonBytesMappingKind::kHexId;
 
  // By default, do not compress data
  std::string compression = "none";
 
  // If using the json name of protobuf field to set the key of json. By default, we will use the
  // field name just like proto files.
  bool use_json_name = false;
 
  // Whether to print the status of the HTTP client in the console
  bool console_debug = false;
 
  std::chrono::system_clock::duration timeout;
 
  // Additional HTTP headers
  OtlpHeaders http_headers;
 
  // Retry policy for select failure codes
  ext::http::client::RetryPolicy retry_policy;
 
  // Concurrent requests
  std::size_t max_concurrent_requests = 64;
 
  // Requests per connections
  std::size_t max_requests_per_connection = 8;
 
  // User agent
  std::string user_agent;
 
  std::shared_ptr<sdk::common::ThreadInstrumentation> thread_instrumentation =
      std::shared_ptr<sdk::common::ThreadInstrumentation>(nullptr);
 
  inline OtlpHttpClientOptions(
      nostd::string_view input_url,
      bool input_ssl_insecure_skip_verify,
      nostd::string_view input_ssl_ca_cert_path,
      nostd::string_view input_ssl_ca_cert_string,
      nostd::string_view input_ssl_client_key_path,
      nostd::string_view input_ssl_client_key_string,
      nostd::string_view input_ssl_client_cert_path,
      nostd::string_view input_ssl_client_cert_string,
      nostd::string_view input_ssl_min_tls,
      nostd::string_view input_ssl_max_tls,
      nostd::string_view input_ssl_cipher,
      nostd::string_view input_ssl_cipher_suite,
      HttpRequestContentType input_content_type,
      JsonBytesMappingKind input_json_bytes_mapping,
      nostd::string_view input_compression,
      bool input_use_json_name,
      bool input_console_debug,
      std::chrono::system_clock::duration input_timeout,
      const OtlpHeaders &input_http_headers,
      std::uint32_t input_retry_policy_max_attempts,
      std::chrono::duration<float> input_retry_policy_initial_backoff,
      std::chrono::duration<float> input_retry_policy_max_backoff,
      float input_retry_policy_backoff_multiplier,
      const std::shared_ptr<sdk::common::ThreadInstrumentation> &input_thread_instrumentation,
      std::size_t input_concurrent_sessions         = 64,
      std::size_t input_max_requests_per_connection = 8,
      nostd::string_view input_user_agent           = GetOtlpDefaultUserAgent())
      : url(input_url),
        ssl_options(input_url,
                    input_ssl_insecure_skip_verify,
                    input_ssl_ca_cert_path,
                    input_ssl_ca_cert_string,
                    input_ssl_client_key_path,
                    input_ssl_client_key_string,
                    input_ssl_client_cert_path,
                    input_ssl_client_cert_string,
                    input_ssl_min_tls,
                    input_ssl_max_tls,
                    input_ssl_cipher,
                    input_ssl_cipher_suite),
        content_type(input_content_type),
        json_bytes_mapping(input_json_bytes_mapping),
        compression(input_compression),
        use_json_name(input_use_json_name),
        console_debug(input_console_debug),
        timeout(input_timeout),
        http_headers(input_http_headers),
        retry_policy{input_retry_policy_max_attempts, input_retry_policy_initial_backoff,
                     input_retry_policy_max_backoff, input_retry_policy_backoff_multiplier},
        max_concurrent_requests(input_concurrent_sessions),
        max_requests_per_connection(input_max_requests_per_connection),
        user_agent(input_user_agent),
        thread_instrumentation(input_thread_instrumentation)
  {}
};
 
class OtlpHttpClient
{
public:
  explicit OtlpHttpClient(OtlpHttpClientOptions &&options);
 
  ~OtlpHttpClient();
  OtlpHttpClient(const OtlpHttpClient &)            = delete;
  OtlpHttpClient &operator=(const OtlpHttpClient &) = delete;
  OtlpHttpClient(OtlpHttpClient &&)                 = delete;
  OtlpHttpClient &operator=(OtlpHttpClient &&)      = delete;
 
  sdk::common::ExportResult Export(const google::protobuf::Message &message) noexcept;
 
  sdk::common::ExportResult Export(
      const google::protobuf::Message &message,
      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept;
 
  sdk::common::ExportResult Export(
      const google::protobuf::Message &message,
      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback,
      std::size_t max_running_requests) noexcept;
 
  bool ForceFlush(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
 
  bool Shutdown(std::chrono::microseconds timeout = std::chrono::microseconds(0)) noexcept;
 
  void ReleaseSession(const opentelemetry::ext::http::client::Session &session) noexcept;
 
  inline const OtlpHttpClientOptions &GetOptions() const noexcept { return options_; }
 
  bool IsShutdown() const noexcept;
 
private:
  struct HttpSessionData
  {
    std::shared_ptr<opentelemetry::ext::http::client::Session> session;
    std::shared_ptr<opentelemetry::ext::http::client::EventHandler> event_handle;
 
    HttpSessionData() = default;
 
    explicit HttpSessionData(
        std::shared_ptr<opentelemetry::ext::http::client::Session> &&input_session,
        std::shared_ptr<opentelemetry::ext::http::client::EventHandler> &&input_handle)
        : session(std::move(input_session)), event_handle(std::move(input_handle))
    {}
  };
 
  nostd::variant<sdk::common::ExportResult, HttpSessionData> createSession(
      const google::protobuf::Message &message,
      std::function<bool(opentelemetry::sdk::common::ExportResult)> &&result_callback) noexcept;
 
  void addSession(HttpSessionData &&session_data) noexcept;
 
  bool cleanupGCSessions() noexcept;
 
  // For testing
  friend class OtlpHttpExporterTestPeer;
  friend class OtlpHttpLogRecordExporterTestPeer;
  friend class OtlpHttpMetricExporterTestPeer;
 
  OtlpHttpClient(OtlpHttpClientOptions &&options,
                 std::shared_ptr<ext::http::client::HttpClient> http_client);
 
  // Stores if this HTTP client had its Shutdown() method called
  std::atomic<bool> is_shutdown_;
 
  // The configuration options associated with this HTTP client.
  const OtlpHttpClientOptions options_;
 
  // Object that stores the HTTP sessions that have been created
  std::shared_ptr<ext::http::client::HttpClient> http_client_;
 
  // Cached parsed URI
  std::string http_uri_;
 
  // Running sessions and event handles
  std::unordered_map<const opentelemetry::ext::http::client::Session *, HttpSessionData>
      running_sessions_;
  // Sessions and event handles that are waiting to be deleted
  std::list<HttpSessionData> gc_sessions_;
  // Lock for running_sessions_, gc_sessions_ and http_client_
  std::recursive_mutex session_manager_lock_;
  // Condition variable and mutex to control the concurrency count of running sessions
  std::mutex session_waker_lock_;
  std::condition_variable session_waker_;
  std::atomic<size_t> start_session_counter_;
  std::atomic<size_t> finished_session_counter_;
};
}  // namespace otlp
}  // namespace exporter
OPENTELEMETRY_END_NAMESPACE