Share grpc::Channel

Author: owent

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_client.h

@@ -34,8 +34,6 @@ namespace otlp
 
 class OtlpGrpcClient;
 struct OtlpGrpcClientOptions;
-
-#ifdef ENABLE_ASYNC_EXPORT
 struct OtlpGrpcClientAsyncData;
 
 class OtlpGrpcClientReferenceGuard
@@ -53,15 +51,14 @@ class OtlpGrpcClientReferenceGuard
   friend class OtlpGrpcClient;
   std::atomic<bool> has_value_;
 };
-#endif
 
 /**
  * The OTLP gRPC client contains utility functions of gRPC.
  */
 class OtlpGrpcClient
 {
 public:
-  OtlpGrpcClient();
+  OtlpGrpcClient(const OtlpGrpcClientOptions &options);
 
   ~OtlpGrpcClient();
 
@@ -79,20 +76,18 @@ class OtlpGrpcClient
   /**
    * Create trace service stub to communicate with the OpenTelemetry Collector.
    */
-  static std::unique_ptr<proto::collector::trace::v1::TraceService::StubInterface>
-  MakeTraceServiceStub(const OtlpGrpcClientOptions &options);
+  std::unique_ptr<proto::collector::trace::v1::TraceService::StubInterface> MakeTraceServiceStub();
 
   /**
    * Create metrics service stub to communicate with the OpenTelemetry Collector.
    */
-  static std::unique_ptr<proto::collector::metrics::v1::MetricsService::StubInterface>
-  MakeMetricsServiceStub(const OtlpGrpcClientOptions &options);
+  std::unique_ptr<proto::collector::metrics::v1::MetricsService::StubInterface>
+  MakeMetricsServiceStub();
 
   /**
    * Create logs service stub to communicate with the OpenTelemetry Collector.
    */
-  static std::unique_ptr<proto::collector::logs::v1::LogsService::StubInterface>
-  MakeLogsServiceStub(const OtlpGrpcClientOptions &options);
+  std::unique_ptr<proto::collector::logs::v1::LogsService::StubInterface> MakeLogsServiceStub();
 
   static grpc::Status DelegateExport(
       proto::collector::trace::v1::TraceService::StubInterface *stub,
@@ -115,8 +110,6 @@ class OtlpGrpcClient
       proto::collector::logs::v1::ExportLogsServiceRequest &&request,
       proto::collector::logs::v1::ExportLogsServiceResponse *response);
 
-#ifdef ENABLE_ASYNC_EXPORT
-
   void AddReference(OtlpGrpcClientReferenceGuard &guard,
                     const OtlpGrpcClientOptions &options) noexcept;
 
@@ -128,6 +121,7 @@ class OtlpGrpcClient
    */
   bool RemoveReference(OtlpGrpcClientReferenceGuard &guard) noexcept;
 
+#ifdef ENABLE_ASYNC_EXPORT
   /**
    * Async export
    * @param options Options used to message to create gRPC context and stub(if necessary)
@@ -187,6 +181,7 @@ class OtlpGrpcClient
                          const proto::collector::logs::v1::ExportLogsServiceRequest &,
                          proto::collector::logs::v1::ExportLogsServiceResponse *)>
           &&result_callback) noexcept;
+#endif
 
   /**
    * Force flush the gRPC client.
@@ -212,7 +207,6 @@ class OtlpGrpcClient
 
   // Stores shared data between threads of this gRPC client
   std::shared_ptr<OtlpGrpcClientAsyncData> async_data_;
-#endif
 };
 }  // namespace otlp
 }  // namespace exporter

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_client_factory.h

@@ -5,6 +5,7 @@
 
 #include <memory>
 
+#include "opentelemetry/exporters/otlp/otlp_grpc_client_options.h"
 #include "opentelemetry/nostd/shared_ptr.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -27,7 +28,7 @@ class OPENTELEMETRY_EXPORT OtlpGrpcClientFactory
   /**
    * Create an OtlpGrpcClient using all default options.
    */
-  static nostd::shared_ptr<OtlpGrpcClient> Create();
+  static nostd::shared_ptr<OtlpGrpcClient> Create(const OtlpGrpcClientOptions &options);
 
 #ifdef ENABLE_ASYNC_EXPORT
   static nostd::shared_ptr<OtlpGrpcClientReferenceGuard> CreateReferenceGuard();

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_exporter.h

@@ -92,23 +92,19 @@ class OtlpGrpcExporter final : public opentelemetry::sdk::trace::SpanExporter
   bool Shutdown(
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
 
-#ifdef ENABLE_ASYNC_EXPORT
   /**
    * Get the Client object
    *
    * @return return binded gRPC client
    */
   const nostd::shared_ptr<OtlpGrpcClient> &GetClient() const noexcept;
-#endif
 
 private:
   // The configuration options associated with this exporter.
   const OtlpGrpcExporterOptions options_;
 
-#ifdef ENABLE_ASYNC_EXPORT
   nostd::shared_ptr<OtlpGrpcClient> client_;
   nostd::shared_ptr<OtlpGrpcClientReferenceGuard> client_reference_guard_;
-#endif
 
   // For testing
   friend class OtlpGrpcExporterTestPeer;

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_log_record_exporter.h

@@ -91,23 +91,19 @@ class OtlpGrpcLogRecordExporter : public opentelemetry::sdk::logs::LogRecordExpo
   bool Shutdown(
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
 
-#ifdef ENABLE_ASYNC_EXPORT
   /**
    * Get the Client object
    *
    * @return return binded gRPC client
    */
   const nostd::shared_ptr<OtlpGrpcClient> &GetClient() const noexcept;
-#endif
 
 private:
   // Configuration options for the exporter
   const OtlpGrpcLogRecordExporterOptions options_;
 
-#ifdef ENABLE_ASYNC_EXPORT
   nostd::shared_ptr<OtlpGrpcClient> client_;
   nostd::shared_ptr<OtlpGrpcClientReferenceGuard> client_reference_guard_;
-#endif
 
   // For testing
   friend class OtlpGrpcLogRecordExporterTestPeer;

exporters/otlp/include/opentelemetry/exporters/otlp/otlp_grpc_metric_exporter.h

@@ -77,23 +77,19 @@ class OtlpGrpcMetricExporter : public opentelemetry::sdk::metrics::PushMetricExp
   bool Shutdown(
       std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept override;
 
-#ifdef ENABLE_ASYNC_EXPORT
   /**
    * Get the Client object
    *
    * @return return binded gRPC client
    */
   const nostd::shared_ptr<OtlpGrpcClient> &GetClient() const noexcept;
-#endif
 
 private:
   // The configuration options associated with this exporter.
   const OtlpGrpcMetricExporterOptions options_;
 
-#ifdef ENABLE_ASYNC_EXPORT
   nostd::shared_ptr<OtlpGrpcClient> client_;
   nostd::shared_ptr<OtlpGrpcClientReferenceGuard> client_reference_guard_;
-#endif
 
   // Aggregation Temporality selector
   const sdk::metrics::AggregationTemporalitySelector aggregation_temporality_selector_;

exporters/otlp/src/otlp_grpc_client.cc

@@ -78,14 +78,18 @@ class OPENTELEMETRY_LOCAL_SYMBOL OtlpGrpcAsyncCallData : public OtlpGrpcAsyncCal
   virtual ~OtlpGrpcAsyncCallData() {}
 };
 }  // namespace
+#endif
 
 struct OtlpGrpcClientAsyncData
 {
+
   std::chrono::system_clock::duration export_timeout = std::chrono::seconds{10};
 
   // The best performance trade-off of gRPC is having numcpu's threads and one completion queue
   // per thread, but this exporter should not cost a lot resource and we don't want to create
-  // too many threads in the process. So we use one completion queue.
+  // too many threads in the process. So we use one completion queue and shared context.
+  std::shared_ptr<grpc::Channel> channel;
+#ifdef ENABLE_ASYNC_EXPORT
   grpc::CompletionQueue cq;
 
   // Running requests, this is used to limit the number of concurrent requests.
@@ -98,6 +102,7 @@ struct OtlpGrpcClientAsyncData
   // Condition variable and mutex to control the concurrency count of running requests.
   std::mutex session_waker_lock;
   std::condition_variable session_waker;
+#endif
 
   // Reference count of OtlpGrpcClient
   std::atomic<int64_t> reference_count{0};
@@ -107,7 +112,6 @@ struct OtlpGrpcClientAsyncData
   // https://stackoverflow.com/questions/53408962/try-to-understand-compiler-error-message-default-member-initializer-required-be
   OtlpGrpcClientAsyncData() {}
 };
-#endif
 
 namespace
 {
@@ -243,24 +247,23 @@ static sdk::common::ExportResult InternalDelegateAsyncExport(
 #endif
 }  // namespace
 
-#ifdef ENABLE_ASYNC_EXPORT
 OtlpGrpcClientReferenceGuard::OtlpGrpcClientReferenceGuard() noexcept : has_value_{false} {}
 
 OtlpGrpcClientReferenceGuard::~OtlpGrpcClientReferenceGuard() noexcept {}
-#endif
 
-OtlpGrpcClient::OtlpGrpcClient()
-#ifdef ENABLE_ASYNC_EXPORT
-    : is_shutdown_(false)
-#endif
-{}
+OtlpGrpcClient::OtlpGrpcClient(const OtlpGrpcClientOptions &options) : is_shutdown_(false)
+{
+  std::shared_ptr<OtlpGrpcClientAsyncData> async_data = MutableAsyncData(options);
+  async_data->channel                                 = MakeChannel(options);
+}
 
 OtlpGrpcClient::~OtlpGrpcClient()
 {
-#ifdef ENABLE_ASYNC_EXPORT
+
   std::shared_ptr<OtlpGrpcClientAsyncData> async_data;
   async_data.swap(async_data_);
 
+#ifdef ENABLE_ASYNC_EXPORT
   while (async_data && async_data->running_requests.load(std::memory_order_acquire) > 0)
   {
     std::unique_lock<std::mutex> lock{async_data->session_waker_lock};
@@ -359,21 +362,33 @@ std::unique_ptr<grpc::ClientContext> OtlpGrpcClient::MakeClientContext(
 }
 
 std::unique_ptr<proto::collector::trace::v1::TraceService::StubInterface>
-OtlpGrpcClient::MakeTraceServiceStub(const OtlpGrpcClientOptions &options)
+OtlpGrpcClient::MakeTraceServiceStub()
 {
-  return proto::collector::trace::v1::TraceService::NewStub(MakeChannel(options));
+  if (!async_data_ || !async_data_->channel)
+  {
+    return nullptr;
+  }
+  return proto::collector::trace::v1::TraceService::NewStub(async_data_->channel);
 }
 
 std::unique_ptr<proto::collector::metrics::v1::MetricsService::StubInterface>
-OtlpGrpcClient::MakeMetricsServiceStub(const OtlpGrpcClientOptions &options)
+OtlpGrpcClient::MakeMetricsServiceStub()
 {
-  return proto::collector::metrics::v1::MetricsService::NewStub(MakeChannel(options));
+  if (!async_data_ || !async_data_->channel)
+  {
+    return nullptr;
+  }
+  return proto::collector::metrics::v1::MetricsService::NewStub(async_data_->channel);
 }
 
 std::unique_ptr<proto::collector::logs::v1::LogsService::StubInterface>
-OtlpGrpcClient::MakeLogsServiceStub(const OtlpGrpcClientOptions &options)
+OtlpGrpcClient::MakeLogsServiceStub()
 {
-  return proto::collector::logs::v1::LogsService::NewStub(MakeChannel(options));
+  if (!async_data_ || !async_data_->channel)
+  {
+    return nullptr;
+  }
+  return proto::collector::logs::v1::LogsService::NewStub(async_data_->channel);
 }
 
 grpc::Status OtlpGrpcClient::DelegateExport(
@@ -406,8 +421,6 @@ grpc::Status OtlpGrpcClient::DelegateExport(
   return stub->Export(context.get(), request, response);
 }
 
-#ifdef ENABLE_ASYNC_EXPORT
-
 void OtlpGrpcClient::AddReference(OtlpGrpcClientReferenceGuard &guard,
                                   const OtlpGrpcClientOptions &options) noexcept
 {
@@ -437,6 +450,8 @@ bool OtlpGrpcClient::RemoveReference(OtlpGrpcClientReferenceGuard &guard) noexce
   return true;
 }
 
+#ifdef ENABLE_ASYNC_EXPORT
+
 /**
  * Async export
  * @param options Options used to message to create gRPC context and stub(if necessary)
@@ -538,6 +553,8 @@ sdk::common::ExportResult OtlpGrpcClient::DelegateAsyncExport(
                                      "log(s)");
 }
 
+#endif
+
 std::shared_ptr<OtlpGrpcClientAsyncData> OtlpGrpcClient::MutableAsyncData(
     const OtlpGrpcClientOptions &options)
 {
@@ -556,13 +573,15 @@ bool OtlpGrpcClient::IsShutdown() const noexcept
   return is_shutdown_.load(std::memory_order_acquire);
 }
 
-bool OtlpGrpcClient::ForceFlush(std::chrono::microseconds timeout) noexcept
+bool OtlpGrpcClient::ForceFlush(
+    OPENTELEMETRY_MAYBE_UNUSED std::chrono::microseconds timeout) noexcept
 {
   if (!async_data_)
   {
     return true;
   }
 
+#ifdef ENABLE_ASYNC_EXPORT
   std::size_t request_counter = async_data_->start_request_counter.load(std::memory_order_acquire);
   if (request_counter <= async_data_->finished_request_counter.load(std::memory_order_acquire))
   {
@@ -601,16 +620,20 @@ bool OtlpGrpcClient::ForceFlush(std::chrono::microseconds timeout) noexcept
   }
 
   return timeout_steady > std::chrono::steady_clock::duration::zero();
+#else
+  return true;
+#endif
 }
 
 bool OtlpGrpcClient::Shutdown(OtlpGrpcClientReferenceGuard &guard,
-                              std::chrono::microseconds timeout) noexcept
+                              OPENTELEMETRY_MAYBE_UNUSED std::chrono::microseconds timeout) noexcept
 {
   if (!async_data_)
   {
     return true;
   }
 
+#ifdef ENABLE_ASYNC_EXPORT
   bool last_reference_removed = RemoveReference(guard);
   bool force_flush_result;
   if (last_reference_removed && false == is_shutdown_.exchange(true, std::memory_order_acq_rel))
@@ -625,9 +648,11 @@ bool OtlpGrpcClient::Shutdown(OtlpGrpcClientReferenceGuard &guard,
   }
 
   return force_flush_result;
-}
-
+#else
+  RemoveReference(guard);
+  return true;
 #endif
+}
 
 }  // namespace otlp
 }  // namespace exporter

exporters/otlp/src/otlp_grpc_client_factory.cc

@@ -12,9 +12,10 @@ namespace exporter
 namespace otlp
 {
 
-nostd::shared_ptr<OtlpGrpcClient> OtlpGrpcClientFactory::Create()
+nostd::shared_ptr<OtlpGrpcClient> OtlpGrpcClientFactory::Create(
+    const OtlpGrpcClientOptions &options)
 {
-  return nostd::shared_ptr<OtlpGrpcClient>(new OtlpGrpcClient());
+  return nostd::shared_ptr<OtlpGrpcClient>(new OtlpGrpcClient(options));
 }
 
 #ifdef ENABLE_ASYNC_EXPORT