Add Jaeger Thrift HTTP exporter

CHANGELOG.md

@@ -16,6 +16,7 @@ Increment the:
 ## [Unreleased]
 
 * [BUILD] Allow to use local GSL
+* [EXPORTER] Jaeger Exporter - Add Thrift HTTP exporter ([#926](https:/open-telemetry/opentelemetry-cpp/pull/926))
 
 ## [1.0.0-rc3] 2021-07-12
 

examples/jaeger/main.cc

@@ -33,7 +33,7 @@ int main(int argc, char *argv[])
 {
  if (argc == 2)
  {
- opts.server_addr = argv[1];
+ opts.endpoint = argv[1];
  }
  // Removing this line will leave the default noop TracerProvider in place.
  InitTracer();

exporters/jaeger/CMakeLists.txt

@@ -10,8 +10,13 @@ set(JAEGER_THRIFT_GENCPP_SOURCES
  thrift-gen/zipkincore_types.cpp)
 
 set(JAEGER_EXPORTER_SOURCES
- src/jaeger_exporter.cc src/thrift_sender.cc src/udp_transport.cc
- src/recordable.cc src/TUDPTransport.cc)
+ src/jaeger_exporter.cc
+ src/thrift_sender.cc
+ src/udp_transport.cc
+ src/recordable.cc
+ src/TUDPTransport.cc
+ src/http_transport.cc
+ src/THttpTransport.cc)
 
 add_library(opentelemetry_exporter_jaeger_trace ${JAEGER_EXPORTER_SOURCES}
  ${JAEGER_THRIFT_GENCPP_SOURCES})
@@ -26,7 +31,7 @@ target_include_directories(
 
 target_link_libraries(
  opentelemetry_exporter_jaeger_trace
- PUBLIC opentelemetry_resources
+ PUBLIC opentelemetry_resources http_client_curl
  PRIVATE thrift::thrift)
 
 if(MSVC)

exporters/jaeger/include/opentelemetry/exporters/jaeger/jaeger_exporter.h

@@ -3,6 +3,7 @@
 
 #pragma once
 
+#include <opentelemetry/ext/http/client/http_client.h>
 #include <opentelemetry/sdk/trace/exporter.h>
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -25,10 +26,11 @@ class ThriftSender;
  */
 struct JaegerExporterOptions
 {
- // The endpoint to export to.
- std::string server_addr = "localhost";
- uint16_t server_port = 6831;
  TransportFormat transport_format = TransportFormat::kThriftUdpCompact;
+ std::string endpoint = "localhost";
+ uint16_t server_port = 6831;
+ // Only applicable when using kThriftHttp transport.
+ ext::http::client::Headers headers;
 };
 
 namespace trace_sdk = opentelemetry::sdk::trace;

exporters/jaeger/src/THttpTransport.cc

@@ -0,0 +1,61 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "THttpTransport.h"
+#include "opentelemetry/ext/http/client/http_client_factory.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace exporter
+{
+namespace jaeger
+{
+
+THttpTransport::THttpTransport(std::string endpoint, ext::http::client::Headers extra_headers)
+ : endpoint(std::move(endpoint)),
+ headers(std::move(extra_headers)),
+ client(ext::http::client::HttpClientFactory::CreateSync())
+{
+ headers.insert({{"Content-Type", "application/vnd.apache.thrift.binary"}});
+}
+
+THttpTransport::~THttpTransport() {}
+
+bool THttpTransport::isOpen() const
+{
+ return true;
+}
+
+uint32_t THttpTransport::read(uint8_t *buf, uint32_t len)
+{
+ (void)buf;
+ (void)len;
+ return 0;
+}
+
+void THttpTransport::write(const uint8_t *buf, uint32_t len)
+{
+ request_buffer.insert(request_buffer.end(), buf, buf + len);
+}
+
+bool THttpTransport::sendSpans()
+{
+ auto result = client->Post(endpoint, request_buffer, headers);
+ request_buffer.clear();
+
+ // TODO: Add logging once global log handling is available.
+ if (!result)
+ {
+ return false;
+ }
+
+ if (result.GetResponse().GetStatusCode() >= 400)
+ {
+ return false;
+ }
+
+ return true;
+}
+
+} // namespace jaeger
+} // namespace exporter
+OPENTELEMETRY_END_NAMESPACE

exporters/jaeger/src/THttpTransport.h

@@ -0,0 +1,40 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <opentelemetry/ext/http/client/http_client.h>
+#include <opentelemetry/version.h>
+
+#include <thrift/transport/TVirtualTransport.h>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace exporter
+{
+namespace jaeger
+{
+
+class THttpTransport : public apache::thrift::transport::TVirtualTransport<THttpTransport>
+{
+public:
+ THttpTransport(std::string endpoint, ext::http::client::Headers extra_headers);
+ ~THttpTransport() override;
+
+ bool isOpen() const override;
+
+ uint32_t read(uint8_t *buf, uint32_t len);
+
+ void write(const uint8_t *buf, uint32_t len);
+
+ bool sendSpans();
+
+private:
+ std::string endpoint;
+ ext::http::client::Headers headers;
+ std::shared_ptr<ext::http::client::HttpClientSync> client;
+ std::vector<uint8_t> request_buffer;
+};
+
+} // namespace jaeger
+} // namespace exporter
+OPENTELEMETRY_END_NAMESPACE

exporters/jaeger/src/http_transport.cc

@@ -0,0 +1,37 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "http_transport.h"
+
+#include <thrift/protocol/TBinaryProtocol.h>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace exporter
+{
+namespace jaeger
+{
+
+using TBinaryProtocol = apache::thrift::protocol::TBinaryProtocol;
+using TTransport = apache::thrift::transport::TTransport;
+
+HttpTransport::HttpTransport(std::string endpoint, ext::http::client::Headers headers)
+{
+ endpoint_transport_ = std::make_shared<THttpTransport>(std::move(endpoint), std::move(headers));
+ protocol_ = std::shared_ptr<TProtocol>(new TBinaryProtocol(endpoint_transport_));
+}
+
+int HttpTransport::EmitBatch(const thrift::Batch &batch)
+{
+ batch.write(protocol_.get());
+
+ if (!endpoint_transport_->sendSpans())
+ {
+ return 0;
+ }
+
+ return static_cast<int>(batch.spans.size());
+}
+
+} // namespace jaeger
+} // namespace exporter
+OPENTELEMETRY_END_NAMESPACE

exporters/jaeger/src/http_transport.h

@@ -0,0 +1,39 @@
+#pragma once
+
+#include "THttpTransport.h"
+#include "transport.h"
+
+#include <thrift/protocol/TProtocol.h>
+#include <thrift/transport/TTransport.h>
+#include <memory>
+#include <string>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace exporter
+{
+namespace jaeger
+{
+
+using TProtocol = apache::thrift::protocol::TProtocol;
+
+class HttpTransport : public Transport
+{
+public:
+ HttpTransport(std::string endpoint, ext::http::client::Headers headers);
+
+ int EmitBatch(const thrift::Batch &batch) override;
+
+ uint32_t MaxPacketSize() const override
+ {
+ // Default to 4 MiB POST body size.
+ return 1 << 22;
+ }
+
+private:
+ std::shared_ptr<THttpTransport> endpoint_transport_;
+ std::shared_ptr<TProtocol> protocol_;
+};
+
+} // namespace jaeger
+} // namespace exporter
+OPENTELEMETRY_END_NAMESPACE

exporters/jaeger/src/jaeger_exporter.cc

@@ -5,6 +5,7 @@
 #include <opentelemetry/exporters/jaeger/jaeger_exporter.h>
 #include <opentelemetry/exporters/jaeger/recordable.h>
 
+#include "http_transport.h"
 #include "thrift_sender.h"
 #include "udp_transport.h"
 
@@ -64,14 +65,21 @@ void JaegerExporter::InitializeEndpoint()
  {
  // TODO: do we need support any authentication mechanism?
  auto transport = std::unique_ptr<Transport>(
- static_cast<Transport *>(new UDPTransport(options_.server_addr, options_.server_port)));
+ static_cast<Transport *>(new UDPTransport(options_.endpoint, options_.server_port)));
  sender_ = std::unique_ptr<ThriftSender>(new ThriftSender(std::move(transport)));
+ return;
  }
- else
+
+ if (options_.transport_format == TransportFormat::kThriftHttp)
  {
- // The transport format is not implemented.
- assert(false);
+ auto transport =
+ std::unique_ptr<HttpTransport>(new HttpTransport(options_.endpoint, options_.headers));
+ sender_ = std::unique_ptr<ThriftSender>(new ThriftSender(std::move(transport)));
+ return;
  }
+
+ // The transport format is not implemented.
+ assert(false);
 }
 
 } // namespace jaeger

exporters/jaeger/src/thrift_sender.cc

@@ -79,11 +79,11 @@ int ThriftSender::Flush()
  batch.__set_process(process_);
  batch.__set_spans(span_buffer_);
 
- transport_->EmitBatch(batch);
+ int spans_flushed = transport_->EmitBatch(batch);
 
  ResetBuffers();
 
- return static_cast<int>(batch.spans.size());
+ return spans_flushed;
 }
 
 void ThriftSender::Close()

exporters/jaeger/src/transport.h

@@ -21,8 +21,8 @@ class Transport
  Transport() = default;
  virtual ~Transport() = default;
 
- virtual void EmitBatch(const thrift::Batch &batch) = 0;
- virtual uint32_t MaxPacketSize() const  = 0;
+ virtual int EmitBatch(const thrift::Batch &batch) = 0;
+ virtual uint32_t MaxPacketSize() const = 0;
 };
 
 } // namespace jaeger

exporters/jaeger/src/udp_transport.cc

@@ -63,14 +63,16 @@ void UDPTransport::CleanSocket()
 #endif
 }
 
-void UDPTransport::EmitBatch(const thrift::Batch &batch)
+int UDPTransport::EmitBatch(const thrift::Batch &batch)
 {
  try
  {
  agent_->emitBatch(batch);
  }
  catch (...)
  {}
+
+ return static_cast<int>(batch.spans.size());
 }
 
 } // namespace jaeger

exporters/jaeger/src/udp_transport.h

@@ -27,7 +27,6 @@ using TBinaryProtocol = apache::thrift::protocol::TBinaryProtocol;
 using TCompactProtocol = apache::thrift::protocol::TCompactProtocol;
 using TBufferedTransport = apache::thrift::transport::TBufferedTransport;
 using TProtocol = apache::thrift::protocol::TProtocol;
-using TSocket = apache::thrift::transport::TSocket;
 using TTransport = apache::thrift::transport::TTransport;
 
 class UDPTransport : public Transport
@@ -38,7 +37,7 @@ class UDPTransport : public Transport
  UDPTransport(const std::string &addr, uint16_t port);
  virtual ~UDPTransport();
 
- void EmitBatch(const thrift::Batch &batch) override;
+ int EmitBatch(const thrift::Batch &batch) override;
 
  uint32_t MaxPacketSize() const override { return max_packet_size_; }