Use OpenTelemetry in the tools

Author: avsej

couchbase/tracing/otel_tracer.hxx

@@ -18,9 +18,7 @@
 #pragma once
 
 #include <couchbase/tracing/request_tracer.hxx>
-#include <opentelemetry/exporters/ostream/span_exporter.h>
-#include <opentelemetry/sdk/trace/simple_processor.h>
-#include <opentelemetry/sdk/trace/tracer_provider.h>
+
 #include <opentelemetry/trace/tracer.h>
 
 namespace nostd = opentelemetry::nostd;
@@ -45,7 +43,7 @@ public:
   {
     span_->End();
   }
-  nostd::shared_ptr<opentelemetry::trace::Span> wrapped_span()
+  auto wrapped_span() -> nostd::shared_ptr<opentelemetry::trace::Span>
   {
     return span_;
   }
@@ -57,12 +55,12 @@ private:
 class otel_request_tracer : public couchbase::tracing::request_tracer
 {
 public:
-  otel_request_tracer(nostd::shared_ptr<opentelemetry::trace::Tracer> tracer)
+  explicit otel_request_tracer(nostd::shared_ptr<opentelemetry::trace::Tracer> tracer)
     : tracer_(std::move(tracer))
   {
   }
 
-  auto start_span(std::string name, std::shared_ptr<couchbase::tracing::request_span> parent = {})
+  auto start_span(std::string name, std::shared_ptr<couchbase::tracing::request_span> parent)
     -> std::shared_ptr<couchbase::tracing::request_span> override
   {
     auto wrapped_parent = std::dynamic_pointer_cast<otel_request_span>(parent);
@@ -77,10 +75,10 @@ public:
   auto wrap_span(nostd::shared_ptr<opentelemetry::trace::Span> span)
     -> std::shared_ptr<couchbase::tracing::otel_request_span>
   {
-    return std::make_shared<couchbase::tracing::otel_request_span>(span);
+    return std::make_shared<couchbase::tracing::otel_request_span>(std::move(span));
   }
 
 private:
   nostd::shared_ptr<opentelemetry::trace::Tracer> tracer_;
 };
-} // namespace couchbase::tracing
+} // namespace couchbase::tracing

tools/CMakeLists.txt

@@ -46,7 +46,8 @@ target_link_libraries(
           $<BUILD_INTERFACE:taocpp::json>
           $<BUILD_INTERFACE:spdlog::spdlog>
           $<BUILD_INTERFACE:hdr_histogram_static>
-          $<BUILD_INTERFACE:asio>)
+          $<BUILD_INTERFACE:asio>
+          $<BUILD_INTERFACE:opentelemetry_exporter_otlp_http>)
 propagate_public_compile_definitions(cbc spdlog::spdlog asio)
 
 if(COUCHBASE_CXX_CLIENT_STATIC_BORINGSSL AND WIN32)

tools/get.cxx

@@ -19,6 +19,7 @@
 #include "utils.hxx"
 
 #include <core/logger/logger.hxx>
+
 #include <couchbase/cluster.hxx>
 #include <couchbase/codec/raw_binary_transcoder.hxx>
 #include <couchbase/codec/tao_json_serializer.hxx>

tools/utils.cxx

@@ -16,11 +16,11 @@
  */
 
 #include "utils.hxx"
+#include "CLI/CLI.hpp"
 #include "core/logger/configuration.hxx"
 #include "core/logger/logger.hxx"
 #include "core/meta/version.hxx"
 #include "core/utils/binary.hxx"
-#include "core/utils/duration_parser.hxx"
 #include "core/utils/json.hxx"
 
 #include <couchbase/cluster_options.hxx>
@@ -30,6 +30,43 @@
 #include <couchbase/fmt/query_scan_consistency.hxx>
 #include <couchbase/scope.hxx>
 
+#if defined(__GNUC__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wsign-conversion"
+#pragma GCC diagnostic ignored "-Wuseless-cast"
+#elif defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wdeprecated-builtins"
+#endif
+#include <opentelemetry/sdk/common/global_log_handler.h>
+
+#include <opentelemetry/trace/provider.h>
+#include <opentelemetry/trace/span_startoptions.h>
+#include <opentelemetry/trace/tracer_provider.h>
+
+#include <opentelemetry/sdk/trace/batch_span_processor.h>
+#include <opentelemetry/sdk/trace/batch_span_processor_factory.h>
+#include <opentelemetry/sdk/trace/samplers/always_off.h>
+#include <opentelemetry/sdk/trace/samplers/always_on.h>
+#include <opentelemetry/sdk/trace/samplers/parent.h>
+#include <opentelemetry/sdk/trace/samplers/trace_id_ratio.h>
+#include <opentelemetry/sdk/trace/simple_processor.h>
+#include <opentelemetry/sdk/trace/simple_processor_factory.h>
+#include <opentelemetry/sdk/trace/tracer_provider.h>
+#include <opentelemetry/sdk/trace/tracer_provider_factory.h>
+
+#include <opentelemetry/exporters/otlp/otlp_http.h>
+#include <opentelemetry/exporters/otlp/otlp_http_exporter_factory.h>
+#include <opentelemetry/exporters/otlp/otlp_http_exporter_options.h>
+
+#include <couchbase/tracing/otel_tracer.hxx>
+#if defined(__GNUC__)
+#pragma GCC diagnostic pop
+#elif defined(__clang__)
+#pragma clang diagnostic pop
+#endif
+
+#include <memory>
 #include <spdlog/details/os.h>
 #include <spdlog/fmt/bundled/chrono.h>
 #include <spdlog/spdlog.h>
@@ -43,13 +80,54 @@ namespace cbc
 {
 namespace
 {
+
+class null_opentelemetry_logger : public opentelemetry::sdk::common::internal_log::LogHandler
+{
+public:
+  void Handle(opentelemetry::sdk::common::internal_log::LogLevel /* level */,
+              const char* /* file */,
+              int /* line */,
+              const char* /* msg */,
+              const opentelemetry::sdk::common::AttributeMap& /* attributes */) noexcept override
+  {
+    // Use it to debug issues with OpenTelemetry
+    // TODO(SA): maybe we should forward these messages to our logger
+  }
+};
+
 auto
-getenv_or_default(std::string_view var_name, const std::string& default_value) -> std::string
+safe_getenv(const std::string& name) noexcept -> std::optional<std::string>
 {
-  if (auto val = spdlog::details::os::getenv(var_name.data()); !val.empty()) {
-    return val;
+  if (name.empty()) {
+    return std::nullopt;
+  }
+
+#if defined(_WIN32)
+  char* buf = nullptr;
+  size_t len = 0;
+  if (_dupenv_s(&buf, &len, name.c_str()) == 0 && buf != nullptr) {
+    std::string value(buf);
+    free(buf);
+    if (!value.empty()) {
+      return value;
+    }
   }
-  return default_value;
+  return std::nullopt;
+
+#else
+  if (const char* val = std::getenv(name.c_str())) { // NOLINT(concurrency-mt-unsafe)
+    if (val[0] != '\0') {
+      return std::string(val);
+    }
+  }
+  return std::nullopt;
+#endif
+}
+
+auto
+getenv_or_default(const std::string& var_name, const std::string& default_value) -> std::string
+{
+  return safe_getenv(var_name).value_or(default_value);
 }
 
 auto
@@ -388,6 +466,84 @@ add_options(CLI::App* app, tracing_options& options)
     ->add_option("--tracing-threshold-view", options.threshold_view, "Threshold for Query service.")
     ->default_val(defaults.tracing.view_threshold)
     ->type_name("DURATION");
+
+  group->add_flag("--tracing-use-opentelemetry",
+                  options.opentelemetry.use_opentelemetry,
+                  "Use OpenTelemetry and export with HTTP transport.");
+
+  group->add_option("--tracing-opentelemetry-endpoint",
+                    options.opentelemetry.endpoint,
+                    "The OTLP HTTP endpoint for OpenTelemetry to export to.");
+
+  std::vector<std::string> allowed_samplers{
+    "always_on",
+    "always_off",
+    "trace_id_ratio_based",
+    "parent_based",
+  };
+  group
+    ->add_option("--tracing-opentelemetry-sampler",
+                 options.opentelemetry.sampler,
+                 "Sampler type for OpenTelemetry tracing.")
+    ->transform(CLI::IsMember(allowed_samplers))
+    ->default_str("parent_based");
+
+  group
+    ->add_option("--tracing-opentelemetry-sampling-ratio",
+                 options.opentelemetry.sampling_ratio,
+                 "Sampling ratio for trace_id_ratio_based sampler (0.0 to 1.0).")
+    ->check(CLI::Range(0.0, 1.0))
+    ->default_str("1.0");
+
+  group->add_flag("--tracing-opentelemetry-use-batch-processor{true},!--tracing-opentelemetry-use-"
+                  "simple-processor{false}",
+                  options.opentelemetry.use_batch_processor,
+                  "Use batch span processor (default) or simple span processor.");
+
+  group
+    ->add_option("--tracing-opentelemetry-batch-schedule-delay",
+                 options.opentelemetry.batch_schedule_delay,
+                 "Batch processor schedule delay in milliseconds.")
+    ->default_val(options.opentelemetry.batch_schedule_delay)
+    ->type_name("DURATION");
+
+  group
+    ->add_option("--tracing-opentelemetry-batch-max-queue-size",
+                 options.opentelemetry.batch_max_queue_size,
+                 "Maximum queue size for batch processor.")
+    ->default_val(options.opentelemetry.batch_max_queue_size);
+
+  group
+    ->add_option("--tracing-opentelemetry-batch-max-export-batch-size",
+                 options.opentelemetry.batch_max_export_batch_size,
+                 "Maximum export batch size for batch processor.")
+    ->default_val(options.opentelemetry.batch_max_export_batch_size);
+
+  group
+    ->add_option("--tracing-opentelemetry-exporter-timeout",
+                 options.opentelemetry.exporter_timeout,
+                 "Exporter timeout in milliseconds.")
+    ->default_val(options.opentelemetry.exporter_timeout)
+    ->type_name("DURATION");
+
+  group->add_flag("--tracing-opentelemetry-use-ssl-credentials",
+                  options.opentelemetry.use_ssl_credentials,
+                  "Use SSL/TLS credentials for secure connection.");
+
+  group->add_option("--tracing-opentelemetry-ssl-credentials-cacert",
+                    options.opentelemetry.ssl_credentials_cacert,
+                    "Path to CA certificate file for SSL/TLS verification.");
+
+  group
+    ->add_option("--tracing-opentelemetry-headers",
+                 options.opentelemetry.headers,
+                 "HTTP headers (e.g. for authentication) (format: key=value).")
+    ->type_name("KEY=VALUE")
+    ->take_all();
+
+  group->add_option("--tracing-opentelemetry-compression",
+                    options.opentelemetry.compression,
+                    "Compression algorithm for exporter (e.g., 'gzip').");
 }
 
 auto
@@ -559,9 +715,77 @@ apply_options(couchbase::cluster_options& options, const metrics_options& metric
   options.metrics().emit_interval(metrics.emit_interval);
 }
 
+void
+apply_opentelemetry_tracer_options(couchbase::cluster_options& options,
+                                   const opentelemetry_tracing_options& tracing)
+{
+  opentelemetry::sdk::common::internal_log::GlobalLogHandler::SetLogHandler(
+    std::make_shared<null_opentelemetry_logger>());
+
+  opentelemetry::exporter::otlp::OtlpHttpExporterOptions exporter_options{};
+  if (auto endpoint = tracing.endpoint; endpoint) {
+    exporter_options.url = endpoint.value();
+  }
+  exporter_options.timeout = tracing.exporter_timeout;
+  if (!tracing.headers.empty()) {
+    for (const auto& [key, value] : tracing.headers) {
+      exporter_options.http_headers.insert({ key, value });
+    }
+  }
+  if (auto compression = tracing.compression; compression) {
+    exporter_options.compression = compression.value();
+  }
+
+  auto exporter = opentelemetry::exporter::otlp::OtlpHttpExporterFactory::Create(exporter_options);
+
+  std::unique_ptr<opentelemetry::sdk::trace::SpanProcessor> processor;
+  if (tracing.use_batch_processor) {
+    opentelemetry::sdk::trace::BatchSpanProcessorOptions batch_opts{};
+    batch_opts.schedule_delay_millis = tracing.batch_schedule_delay;
+    batch_opts.max_queue_size = tracing.batch_max_queue_size;
+    batch_opts.max_export_batch_size = tracing.batch_max_export_batch_size;
+    processor =
+      opentelemetry::sdk::trace::BatchSpanProcessorFactory::Create(std::move(exporter), batch_opts);
+  } else {
+    processor = opentelemetry::sdk::trace::SimpleSpanProcessorFactory::Create(std::move(exporter));
+  }
+
+  auto resource = opentelemetry::sdk::resource::Resource::Create({
+    { "service.name", "cbc" },
+    { "service.version", couchbase::core::meta::sdk_semver() },
+  });
+
+  std::unique_ptr<opentelemetry::sdk::trace::Sampler> sampler;
+  if (!tracing.sampler || tracing.sampler == "parent_based") {
+    sampler = std::make_unique<opentelemetry::sdk::trace::ParentBasedSampler>(
+      std::make_shared<opentelemetry::sdk::trace::AlwaysOnSampler>());
+  } else if (tracing.sampler == "always_on") {
+    sampler = std::make_unique<opentelemetry::sdk::trace::AlwaysOnSampler>();
+  } else if (tracing.sampler == "always_off") {
+    sampler = std::make_unique<opentelemetry::sdk::trace::AlwaysOffSampler>();
+  } else if (tracing.sampler == "trace_id_ratio_based") {
+    sampler =
+      std::make_unique<opentelemetry::sdk::trace::TraceIdRatioBasedSampler>(tracing.sampling_ratio);
+  }
+
+  auto provider = opentelemetry::sdk::trace::TracerProviderFactory::Create(
+    std::move(processor), resource, std::move(sampler));
+
+  opentelemetry::trace::Provider::SetTracerProvider(
+    std::shared_ptr<opentelemetry::trace::TracerProvider>{ std::move(provider) });
+
+  options.tracing().tracer(std::make_shared<couchbase::tracing::otel_request_tracer>(
+    opentelemetry::trace::Provider::GetTracerProvider()->GetTracer(
+      "cbc", couchbase::core::meta::sdk_semver())));
+}
+
 void
 apply_options(couchbase::cluster_options& options, const tracing_options& tracing)
 {
+  if (tracing.opentelemetry.use_opentelemetry) {
+    apply_opentelemetry_tracer_options(options, tracing.opentelemetry);
+  }
+
   options.tracing().enable(!tracing.disable);
   options.tracing().orphaned_emit_interval(tracing.orphaned_emit_interval);
   options.tracing().orphaned_sample_size(tracing.orphaned_sample_size);
@@ -741,7 +965,8 @@ available_analytics_scan_consistency_modes() -> std::vector<std::string>
 fail(std::string_view message)
 {
   fmt::print(stderr, "ERROR: {}\n", message);
-  exit(EXIT_FAILURE);
+
+  std::quick_exit(EXIT_FAILURE);
 }
 
 auto