Add HTTP exporting to OpenTelemetry Tracer

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

@@ -3,6 +3,7 @@ syntax = "proto3";
 package envoy.config.trace.v3;
 
 import "envoy/config/core/v3/grpc_service.proto";
+import "envoy/config/core/v3/http_uri.proto";
 
 import "udpa/annotations/status.proto";
 
@@ -17,6 +18,31 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Configuration for the OpenTelemetry tracer.
 //  [#extension: envoy.tracers.opentelemetry]
 message OpenTelemetryConfig {
+  // 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 {
+    enum HttpFormat {
+      // Binary Protobuf Encoding.
+      BINARY_PROTOBUF = 0;
+
+      // JSON Protobuf Encoding.
+      JSON_PROTOBUF = 1;
+    }
+
+    // The upstream cluster that will receive OTLP traces over HTTP.
+    core.v3.HttpUri http_uri = 1;
+
+    // OTLP/HTTP uses Protobuf payloads encoded either in binary format or in JSON format.
+    // See https://opentelemetry.io/docs/specs/otlp/#otlphttp.
+    HttpFormat http_format = 2;
+
+    // Optional path. If omitted, the default path of "/v1/traces" will be used.
+    string collector_path = 3;
+
+    // Hostname to include with the request to the collector in the Host header.
+    string collector_hostname = 4;
+  }
+
   // 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.
@@ -25,4 +51,7 @@ message OpenTelemetryConfig {
   // 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".
   string service_name = 2;
+
+  // This field can be also left empty to disable reporting traces to the collector.
+  HttpConfig http_config = 3;
 }

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,73 @@
+#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;
+
+  // TODO: add JSON option as well (though it may need custom serializing, see
+  // https://opentelemetry.io/docs/specs/otlp/#json-protobuf-encoding)
+  if (http_config_.http_format() ==
+      envoy::config::trace::v3::OpenTelemetryConfig::HttpConfig::BINARY_PROTOBUF) {
+    const auto ok = request.SerializeToString(&request_body);
+    if (!ok) {
+      ENVOY_LOG(warn, "Error during Span proto serialization.");
+      return false;
+    }
+  }
+
+  Http::RequestMessagePtr message = std::make_unique<Http::RequestMessageImpl>();
+  message->headers().setReferenceMethod(Http::Headers::get().MethodValues.Post);
+  message->headers().setPath(http_config_.collector_path());
+  message->headers().setHost(http_config_.collector_hostname());
+  message->headers().setReferenceContentType(Http::Headers::get().ContentTypeValues.Protobuf);
+  message->body().add(request_body);
+
+  const auto thread_local_cluster =
+      cluster_manager_.getThreadLocalCluster(http_config_.http_uri().cluster());
+  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_.http_uri().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,51 @@
+
+#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);
+
+  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,17 @@ 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;
+    OpenTelemetryTraceExporterPtr exporter;
     if (opentelemetry_config.has_grpc_service()) {
       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);
+    } else if (opentelemetry_config.has_http_config()) {
+      exporter = std::make_unique<OpenTelemetryHttpTraceExporter>(
+          factory_context.clusterManager(), opentelemetry_config.http_config(), tracing_stats_);
     }
     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)

source/extensions/tracers/opentelemetry/tracer.h

@@ -14,26 +14,19 @@
 
 #include "absl/strings/escaping.h"
 #include "span_context.h"
+#include "tracer_stats.h"
 
 namespace Envoy {
 namespace Extensions {
 namespace Tracers {
 namespace OpenTelemetry {
 
-#define OPENTELEMETRY_TRACER_STATS(COUNTER)                                                        \
-  COUNTER(spans_sent)                                                                              \
-  COUNTER(timer_flushed)
-
-struct OpenTelemetryTracerStats {
-  OPENTELEMETRY_TRACER_STATS(GENERATE_COUNTER_STRUCT)
-};
-
 /**
  * OpenTelemetry Tracer. It is stored in TLS and contains the exporter.
  */
 class Tracer : Logger::Loggable<Logger::Id::tracing> {
 public:
-  Tracer(OpenTelemetryGrpcTraceExporterPtr exporter, Envoy::TimeSource& time_source,
+  Tracer(OpenTelemetryTraceExporterPtr exporter, Envoy::TimeSource& time_source,
          Random::RandomGenerator& random, Runtime::Loader& runtime, Event::Dispatcher& dispatcher,
          OpenTelemetryTracerStats tracing_stats, const std::string& service_name);
 
@@ -55,7 +48,7 @@ class Tracer : Logger::Loggable<Logger::Id::tracing> {
    */
   void flushSpans();
 
-  OpenTelemetryGrpcTraceExporterPtr exporter_;
+  OpenTelemetryTraceExporterPtr exporter_;
   Envoy::TimeSource& time_source_;
   Random::RandomGenerator& random_;
   std::vector<::opentelemetry::proto::trace::v1::Span> span_buffer_;

source/extensions/tracers/opentelemetry/tracer_stats.h

@@ -0,0 +1,24 @@
+#pragma once
+
+#include "envoy/stats/stats_macros.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace OpenTelemetry {
+
+#define OPENTELEMETRY_TRACER_STATS(COUNTER)                                                        \
+  COUNTER(http_reports_failed)                                                                     \
+  COUNTER(http_reports_sent)                                                                       \
+  COUNTER(http_reports_success)                                                                    \
+  COUNTER(spans_sent)                                                                              \
+  COUNTER(timer_flushed)
+
+struct OpenTelemetryTracerStats {
+  OPENTELEMETRY_TRACER_STATS(GENERATE_COUNTER_STRUCT)
+};
+
+} // namespace OpenTelemetry
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy