Internal - OTLP HTTP exporter

api/envoy/config/trace/v3/opentelemetry.proto

@@ -2,9 +2,13 @@ syntax = "proto3";
 
 package envoy.config.trace.v3;
 
+import "envoy/config/core/v3/base.proto";
 import "envoy/config/core/v3/grpc_service.proto";
 
+import "google/protobuf/duration.proto";
+
 import "udpa/annotations/status.proto";
+import "validate/validate.proto";
 
 option java_package = "io.envoyproxy.envoy.config.trace.v3";
 option java_outer_classname = "OpentelemetryProto";
@@ -17,10 +21,41 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Configuration for the OpenTelemetry tracer.
 //  [#extension: envoy.tracers.opentelemetry]
 message OpenTelemetryConfig {
-  // The upstream gRPC cluster that will receive OTLP traces.
-  // Note that the tracer drops traces if the server does not read data fast enough.
-  // This field can be left empty to disable reporting traces to the collector.
-  core.v3.GrpcService grpc_service = 1;
+  // Configuration for sending traces to the collector over HTTP. Note that
+  // this will only be used if the grpc_service is not set above.
+  message HttpConfig {
+
+    // The upstream cluster that will receive OTLP traces over HTTP.
+    string cluster_name = 1;
+
+    // The path to the trace ingest endpoint. The path is appended with the host name configured in the cluster.
+    // Optional: If omitted, "/v1/traces" will be used as default.
+    string traces_path = 2;
+
+    // Sets the maximum duration in milliseconds that a response can take to arrive upon request.
+    google.protobuf.Duration timeout = 3 [(validate.rules).duration = {
+      required: true
+      gte {}
+    }];
+
+    // Custom header values to be added to the OTLP HTTP request.
+    repeated config.core.v3.HeaderValue headers = 4;
+
+    // The hostname to include in the Host header of the OTLP HTTP request.
+    string hostname = 5;
+  }
+
+  oneof export_protocol {
+    option (validate.required) = true;
+
+    // The upstream gRPC cluster that will receive OTLP traces.
+    // Note that the tracer drops traces if the server does not read data fast enough.
+    // This field can be left empty to disable reporting traces to the collector.
+    core.v3.GrpcService grpc_service = 1;
+
+    // The configuration to export OTLP traces via HTTP.
+    HttpConfig http_config = 3;
+  }
 
   // The name for the service. This will be populated in the ResourceSpan Resource attributes.
   // If it is not provided, it will default to "unknown_service:envoy".

source/extensions/tracers/opentelemetry/BUILD

@@ -34,9 +34,10 @@ envoy_cc_library(
         "span_context.h",
         "span_context_extractor.h",
         "tracer.h",
+        "tracer_stats.h",
     ],
     deps = [
-        ":grpc_trace_exporter",
+        ":trace_exporter",
         "//envoy/thread_local:thread_local_interface",
         "//source/common/config:utility_lib",
         "//source/common/tracing:http_tracer_lib",
@@ -47,13 +48,28 @@ envoy_cc_library(
 )
 
 envoy_cc_library(
-    name = "grpc_trace_exporter",
-    srcs = ["grpc_trace_exporter.cc"],
-    hdrs = ["grpc_trace_exporter.h"],
+    name = "trace_exporter",
+    srcs = [
+        "grpc_trace_exporter.cc",
+        "http_trace_exporter.cc",
+    ],
+    hdrs = [
+        "grpc_trace_exporter.h",
+        "http_trace_exporter.h",
+        "trace_exporter.h",
+        "tracer_stats.h",
+    ],
     deps = [
         "//envoy/grpc:async_client_manager_interface",
+        "//envoy/upstream:cluster_manager_interface",
         "//source/common/grpc:typed_async_client_lib",
+        "//source/common/http:async_client_utility_lib",
+        "//source/common/http:header_map_lib",
+        "//source/common/http:message_lib",
+        "//source/common/http:utility_lib",
         "//source/common/protobuf",
+        "@envoy_api//envoy/config/core/v3:pkg_cc_proto",
+        "@envoy_api//envoy/config/trace/v3:pkg_cc_proto",
         "@opentelemetry_proto//:trace_cc_proto",
     ],
 )

source/extensions/tracers/opentelemetry/grpc_trace_exporter.h

@@ -6,6 +6,7 @@
 #include "source/common/grpc/typed_async_client.h"
 
 #include "opentelemetry/proto/collector/trace/v1/trace_service.pb.h"
+#include "trace_exporter.h"
 
 namespace Envoy {
 namespace Extensions {
@@ -80,18 +81,16 @@ class OpenTelemetryGrpcTraceExporterClient : Logger::Loggable<Logger::Id::tracin
   const Protobuf::MethodDescriptor& service_method_;
 };
 
-class OpenTelemetryGrpcTraceExporter : Logger::Loggable<Logger::Id::tracing> {
+class OpenTelemetryGrpcTraceExporter : public OpenTelemetryTraceExporter {
 public:
   OpenTelemetryGrpcTraceExporter(const Grpc::RawAsyncClientSharedPtr& client);
 
-  bool log(const ExportTraceServiceRequest& request);
+  bool log(const ExportTraceServiceRequest& request) override;
 
 private:
   OpenTelemetryGrpcTraceExporterClient client_;
 };
 
-using OpenTelemetryGrpcTraceExporterPtr = std::unique_ptr<OpenTelemetryGrpcTraceExporter>;
-
 } // namespace OpenTelemetry
 } // namespace Tracers
 } // namespace Extensions

source/extensions/tracers/opentelemetry/http_trace_exporter.cc

@@ -0,0 +1,86 @@
+#include <chrono>
+#include <memory>
+#include <string>
+
+#include "http_trace_exporter.h"
+
+#include "source/common/common/logger.h"
+#include "source/common/protobuf/protobuf.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace OpenTelemetry {
+
+OpenTelemetryHttpTraceExporter::OpenTelemetryHttpTraceExporter(
+    Upstream::ClusterManager& cluster_manager,
+    envoy::config::trace::v3::OpenTelemetryConfig::HttpConfig http_config,
+    OpenTelemetryTracerStats& tracing_stats)
+    : cluster_manager_(cluster_manager), http_config_(http_config), tracing_stats_(tracing_stats) {}
+
+bool OpenTelemetryHttpTraceExporter::log(const ExportTraceServiceRequest& request) {
+
+  std::string request_body;
+
+  const auto ok = request.SerializeToString(&request_body);
+  if (!ok) {
+    ENVOY_LOG(warn, "Error while serializing the binary proto ExportTraceServiceRequest.");
+    return false;
+  }
+
+  Http::RequestMessagePtr message = std::make_unique<Http::RequestMessageImpl>();
+  message->headers().setReferenceMethod(Http::Headers::get().MethodValues.Post);
+  message->headers().setReferenceContentType(Http::Headers::get().ContentTypeValues.Protobuf);
+
+  // If traces_path is omitted, send to /v1/traces by default
+  if (http_config_.traces_path().empty()) {
+    message->headers().setPath(TRACES_PATH);
+  } else {
+    message->headers().setPath(http_config_.traces_path());
+  }
+
+  // TODO: Can we get the hostname that is configured in the cluster "socker_address" field?
+  message->headers().setHost(http_config_.hostname());
+
+  // add all custom headers to the request
+  for (const envoy::config::core::v3::HeaderValue& header : http_config_.headers()) {
+    message->headers().setCopy(Http::LowerCaseString(header.key()), header.value());
+  }
+
+  message->body().add(request_body);
+
+  const auto thread_local_cluster =
+      cluster_manager_.getThreadLocalCluster(http_config_.cluster_name());
+  if (thread_local_cluster == nullptr) {
+    ENVOY_LOG(warn, "Thread local cluster not found for collector.");
+    return false;
+  }
+
+  std::chrono::milliseconds timeout = std::chrono::duration_cast<std::chrono::milliseconds>(
+      std::chrono::nanoseconds(http_config_.timeout().nanos()));
+  Http::AsyncClient::Request* http_request = thread_local_cluster->httpAsyncClient().send(
+      std::move(message), *this, Http::AsyncClient::RequestOptions().setTimeout(timeout));
+  tracing_stats_.http_reports_sent_.inc();
+
+  return http_request;
+}
+
+void OpenTelemetryHttpTraceExporter::onSuccess(const Http::AsyncClient::Request&,
+                                               Http::ResponseMessagePtr&& message) {
+  tracing_stats_.http_reports_success_.inc();
+  const auto response_code = message->headers().Status()->value().getStringView();
+  if (response_code != "200") {
+    ENVOY_LOG(warn, "response code: {}", response_code);
+  }
+}
+
+void OpenTelemetryHttpTraceExporter::onFailure(const Http::AsyncClient::Request&,
+                                               Http::AsyncClient::FailureReason) {
+  ENVOY_LOG(debug, "Request failed.");
+  tracing_stats_.http_reports_failed_.inc();
+}
+
+} // namespace OpenTelemetry
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/opentelemetry/http_trace_exporter.h

@@ -0,0 +1,54 @@
+
+#pragma once
+
+#include "envoy/config/core/v3/http_uri.pb.h"
+#include "envoy/config/trace/v3/opentelemetry.pb.h"
+#include "envoy/upstream/cluster_manager.h"
+
+#include "source/common/common/logger.h"
+#include "source/common/http/async_client_impl.h"
+#include "source/common/http/async_client_utility.h"
+#include "source/common/http/headers.h"
+#include "source/common/http/message_impl.h"
+#include "source/common/http/utility.h"
+
+#include "opentelemetry/proto/collector/trace/v1/trace_service.pb.h"
+#include "trace_exporter.h"
+#include "tracer_stats.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace OpenTelemetry {
+
+/**
+ * Exporter for OTLP traces over HTTP.
+ */
+class OpenTelemetryHttpTraceExporter : public OpenTelemetryTraceExporter,
+                                       public Http::AsyncClient::Callbacks {
+public:
+  OpenTelemetryHttpTraceExporter(
+      Upstream::ClusterManager& cluster_manager,
+      envoy::config::trace::v3::OpenTelemetryConfig::HttpConfig http_config,
+      OpenTelemetryTracerStats& tracing_stats);
+
+  // The default path to use when OpenTelemetryConfig::HttpConfig::traces_path is empty
+  const Http::LowerCaseString TRACES_PATH{"/v1/traces"};
+
+  bool log(const ExportTraceServiceRequest& request) override;
+
+  // Http::AsyncClient::Callbacks.
+  void onSuccess(const Http::AsyncClient::Request&, Http::ResponseMessagePtr&&) override;
+  void onFailure(const Http::AsyncClient::Request&, Http::AsyncClient::FailureReason) override;
+  void onBeforeFinalizeUpstreamSpan(Tracing::Span&, const Http::ResponseHeaderMap*) override {}
+
+private:
+  Upstream::ClusterManager& cluster_manager_;
+  envoy::config::trace::v3::OpenTelemetryConfig::HttpConfig http_config_;
+  OpenTelemetryTracerStats& tracing_stats_;
+};
+
+} // namespace OpenTelemetry
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/opentelemetry/opentelemetry_tracer_impl.cc

@@ -9,10 +9,13 @@
 #include "source/common/config/utility.h"
 #include "source/common/tracing/http_tracer_impl.h"
 
+#include "grpc_trace_exporter.h"
+#include "http_trace_exporter.h"
 #include "opentelemetry/proto/collector/trace/v1/trace_service.pb.h"
 #include "opentelemetry/proto/trace/v1/trace.pb.h"
 #include "span_context.h"
 #include "span_context_extractor.h"
+#include "trace_exporter.h"
 #include "tracer.h"
 
 namespace Envoy {
@@ -28,14 +31,24 @@ Driver::Driver(const envoy::config::trace::v3::OpenTelemetryConfig& opentelemetr
   auto& factory_context = context.serverFactoryContext();
   // Create the tracer in Thread Local Storage.
   tls_slot_ptr_->set([opentelemetry_config, &factory_context, this](Event::Dispatcher& dispatcher) {
-    OpenTelemetryGrpcTraceExporterPtr exporter;
-    if (opentelemetry_config.has_grpc_service()) {
-      Grpc::AsyncClientFactoryPtr&& factory =
+    OpenTelemetryTraceExporterPtr exporter;
+    switch (opentelemetry_config.export_protocol_case()) {
+      case envoy::config::trace::v3::OpenTelemetryConfig::ExportProtocolCase::kGrpcService: {
+        Grpc::AsyncClientFactoryPtr&& factory =
           factory_context.clusterManager().grpcAsyncClientManager().factoryForGrpcService(
               opentelemetry_config.grpc_service(), factory_context.scope(), true);
-      const Grpc::RawAsyncClientSharedPtr& async_client_shared_ptr =
-          factory->createUncachedRawAsyncClient();
-      exporter = std::make_unique<OpenTelemetryGrpcTraceExporter>(async_client_shared_ptr);
+        const Grpc::RawAsyncClientSharedPtr& async_client_shared_ptr =
+            factory->createUncachedRawAsyncClient();
+        exporter = std::make_unique<OpenTelemetryGrpcTraceExporter>(async_client_shared_ptr);
+        break;
+      }
+      case envoy::config::trace::v3::OpenTelemetryConfig::ExportProtocolCase::kHttpConfig: {
+        exporter = std::make_unique<OpenTelemetryHttpTraceExporter>(
+          factory_context.clusterManager(), opentelemetry_config.http_config(), tracing_stats_);
+          break;
+      }
+      default:
+        break;
     }
     TracerPtr tracer = std::make_unique<Tracer>(
         std::move(exporter), factory_context.timeSource(), factory_context.api().randomGenerator(),

source/extensions/tracers/opentelemetry/opentelemetry_tracer_impl.h

@@ -8,8 +8,8 @@
 #include "source/common/common/logger.h"
 #include "source/common/singleton/const_singleton.h"
 #include "source/extensions/tracers/common/factory_base.h"
-#include "source/extensions/tracers/opentelemetry/grpc_trace_exporter.h"
-#include "source/extensions/tracers/opentelemetry/tracer.h"
+
+#include "tracer.h"
 
 namespace Envoy {
 namespace Extensions {

source/extensions/tracers/opentelemetry/trace_exporter.h

@@ -0,0 +1,26 @@
+#pragma once
+
+#include "source/common/common/logger.h"
+
+#include "opentelemetry/proto/collector/trace/v1/trace_service.pb.h"
+
+using opentelemetry::proto::collector::trace::v1::ExportTraceServiceRequest;
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace OpenTelemetry {
+
+class OpenTelemetryTraceExporter : public Logger::Loggable<Logger::Id::tracing> {
+public:
+  virtual ~OpenTelemetryTraceExporter() = default;
+
+  virtual bool log(const ExportTraceServiceRequest& request) = 0;
+};
+
+using OpenTelemetryTraceExporterPtr = std::unique_ptr<OpenTelemetryTraceExporter>;
+
+} // namespace OpenTelemetry
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/opentelemetry/tracer.cc

@@ -91,7 +91,7 @@ void Span::setTag(absl::string_view name, absl::string_view value) {
   *span_.add_attributes() = key_value;
 }
 
-Tracer::Tracer(OpenTelemetryGrpcTraceExporterPtr exporter, Envoy::TimeSource& time_source,
+Tracer::Tracer(OpenTelemetryTraceExporterPtr exporter, Envoy::TimeSource& time_source,
                Random::RandomGenerator& random, Runtime::Loader& runtime,
                Event::Dispatcher& dispatcher, OpenTelemetryTracerStats tracing_stats,
                const std::string& service_name)