open-telemetry · ThomsonTan · Mar 8, 2022 · Mar 4, 2022 · Mar 4, 2022 · Mar 4, 2022
@@ -111,9 +111,8 @@ class ElasticSearchRecordable final : public sdk::logs::Recordable
   void SetSeverity(opentelemetry::logs::Severity severity) noexcept override
   {
     // Convert the severity enum to a string
-    int severity_index = static_cast<int>(severity);
-    if (severity_index < 0 ||
-        severity_index >= std::extent<decltype(opentelemetry::logs::SeverityNumToText)>::value)
+    std::uint32_t severity_index = static_cast<std::uint32_t>(severity);
+    if (severity_index >= std::extent<decltype(opentelemetry::logs::SeverityNumToText)>::value)
     {
       std::stringstream sout;
       sout << "Invalid severity(" << severity_index << ")";

@@ -146,12 +146,11 @@ sdk::common::ExportResult OStreamLogExporter::Export(
     // into severity_num and severity_text
     sout_ << "{\n"
           << "  timestamp     : " << log_record->GetTimestamp().time_since_epoch().count() << "\n"
-          << "  severity_num  : " << static_cast<int>(log_record->GetSeverity()) << "\n"
+          << "  severity_num  : " << static_cast<std::uint32_t>(log_record->GetSeverity()) << "\n"
           << "  severity_text : ";
 
-    int severity_index = static_cast<int>(log_record->GetSeverity());
-    if (severity_index < 0 ||
-        severity_index >= std::extent<decltype(opentelemetry::logs::SeverityNumToText)>::value)
+    std::uint32_t severity_index = static_cast<std::uint32_t>(log_record->GetSeverity());
+    if (severity_index >= std::extent<decltype(opentelemetry::logs::SeverityNumToText)>::value)
     {
       sout_ << "Invalid severity(" << severity_index << ")\n";
     }

@@ -110,6 +110,11 @@ sdk::common::ExportResult OtlpGrpcExporter::Export(
                                                      << " span(s) failed, exporter is shutdown");
     return sdk::common::ExportResult::kFailure;
   }
+  if (spans.empty())
+  {
+    return sdk::common::ExportResult::kSuccess;
+  }
+
   proto::collector::trace::v1::ExportTraceServiceRequest request;
   OtlpRecordableUtils::PopulateRequest(spans, &request);
 

@@ -131,6 +131,11 @@ opentelemetry::sdk::common::ExportResult OtlpGrpcLogExporter::Export(
                                                          << " log(s) failed, exporter is shutdown");
     return sdk::common::ExportResult::kFailure;
   }
+  if (logs.empty())
+  {
+    return sdk::common::ExportResult::kSuccess;
+  }
+
   proto::collector::logs::v1::ExportLogsServiceRequest request;
   OtlpRecordableUtils::PopulateRequest(logs, &request);
 

@@ -46,6 +46,11 @@ std::unique_ptr<opentelemetry::sdk::trace::Recordable> OtlpHttpExporter::MakeRec
 opentelemetry::sdk::common::ExportResult OtlpHttpExporter::Export(
     const nostd::span<std::unique_ptr<opentelemetry::sdk::trace::Recordable>> &spans) noexcept
 {
+  if (spans.empty())
+  {
+    return opentelemetry::sdk::common::ExportResult::kSuccess;
+  }
+
   proto::collector::trace::v1::ExportTraceServiceRequest service_request;
   OtlpRecordableUtils::PopulateRequest(spans, &service_request);
   return http_client_->Export(service_request);

@@ -48,6 +48,10 @@ std::unique_ptr<opentelemetry::sdk::logs::Recordable> OtlpHttpLogExporter::MakeR
 opentelemetry::sdk::common::ExportResult OtlpHttpLogExporter::Export(
     const nostd::span<std::unique_ptr<opentelemetry::sdk::logs::Recordable>> &logs) noexcept
 {
+  if (logs.empty())
+  {
+    return opentelemetry::sdk::common::ExportResult::kSuccess;
+  }
   proto::collector::logs::v1::ExportLogsServiceRequest service_request;
   OtlpRecordableUtils::PopulateRequest(logs, &service_request);
   return http_client_->Export(service_request);

@@ -103,7 +103,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportJsonIntegrationTest)
 
   auto provider = nostd::shared_ptr<sdk::logs::LoggerProvider>(new sdk::logs::LoggerProvider());
   provider->AddProcessor(std::unique_ptr<sdk::logs::LogProcessor>(
-      new sdk::logs::BatchLogProcessor(std::move(exporter), 5, std::chrono::milliseconds(256), 1)));
+      new sdk::logs::BatchLogProcessor(std::move(exporter), 5, std::chrono::milliseconds(256), 5)));
 
   std::string report_trace_id;
   std::string report_span_id;
@@ -192,7 +192,7 @@ TEST_F(OtlpHttpLogExporterTestPeer, ExportBinaryIntegrationTest)
 
   auto provider = nostd::shared_ptr<sdk::logs::LoggerProvider>(new sdk::logs::LoggerProvider());
   provider->AddProcessor(std::unique_ptr<sdk::logs::LogProcessor>(
-      new sdk::logs::BatchLogProcessor(std::move(exporter), 5, std::chrono::milliseconds(256), 1)));
+      new sdk::logs::BatchLogProcessor(std::move(exporter), 5, std::chrono::milliseconds(256), 5)));
 
   std::string report_trace_id;
   std::string report_span_id;

@@ -291,7 +291,9 @@ TEST_F(BasicCurlHttpTests, SendGetRequestSyncTimeout)
   auto result             = http_client.Get("http://222.222.222.200:19000/get/", m1);
   EXPECT_EQ(result, false);
 
-  EXPECT_EQ(result.GetSessionState(), http_client::SessionState::ConnectFailed);
+  // When network is under proxy, it may connect success but closed by peer when send data
+  EXPECT_TRUE(result.GetSessionState() == http_client::SessionState::ConnectFailed ||
+              result.GetSessionState() == http_client::SessionState::SendFailed);
 }
 
 TEST_F(BasicCurlHttpTests, SendPostRequestSync)

@@ -80,7 +80,10 @@ class PushController
   {
     if (active_.exchange(false))
     {
-      runner_.join();
+      if (runner_.joinable())
+      {
+        runner_.join();
+      }
       tick();  // flush metrics sitting in the processor
     }
   }

@@ -108,7 +108,7 @@ class BatchLogProcessor : public LogProcessor
 
   /* Synchronization primitives */
   std::condition_variable cv_, force_flush_cv_;
-  std::mutex cv_m_, force_flush_cv_m_;
+  std::mutex cv_m_, force_flush_cv_m_, shutdown_m_;
 
   /* The buffer/queue to which the ended logs are added */
   common::CircularBuffer<Recordable> buffer_;

@@ -67,9 +67,10 @@ class Logger final : public opentelemetry::logs::Logger
   // The name of this logger
   std::string logger_name_;
 
-  // The logger context of this Logger. Uses a weak_ptr to avoid cyclic dependency issues the with
-  std::weak_ptr<LoggerContext> context_;
+  // order of declaration is important here - instrumentation library should destroy after
+  // logger-context.
   std::unique_ptr<instrumentationlibrary::InstrumentationLibrary> instrumentation_library_;
+  std::shared_ptr<LoggerContext> context_;
 };
 
 }  // namespace logs

@@ -61,6 +61,8 @@ class LoggerProvider final : public opentelemetry::logs::LoggerProvider
    */
   explicit LoggerProvider(std::shared_ptr<sdk::logs::LoggerContext> context) noexcept;
 
+  ~LoggerProvider();
+
   /**
    * Creates a logger with the given name, and returns a shared pointer to it.
    * If a logger with that name already exists, return a shared pointer to it
@@ -107,14 +109,20 @@ class LoggerProvider final : public opentelemetry::logs::LoggerProvider
    */
   const opentelemetry::sdk::resource::Resource &GetResource() const noexcept;
 
-private:
-  // A pointer to the processor stored by this logger provider
-  std::shared_ptr<sdk::logs::LoggerContext> context_;
+  /**
+   * Shutdown the log processor associated with this log provider.
+   */
+  bool Shutdown() noexcept;
 
-  // A vector of pointers to all the loggers that have been created
-  std::vector<std::shared_ptr<opentelemetry::sdk::logs::Logger>> loggers_;
+  /**
+   * Force flush the log processor associated with this log provider.
+   */
+  bool ForceFlush(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
 
-  // A mutex that ensures only one thread is using the map of loggers
+private:
+  // order of declaration is important here - loggers should destroy only after context.
+  std::vector<std::shared_ptr<opentelemetry::sdk::logs::Logger>> loggers_;
+  std::shared_ptr<sdk::logs::LoggerContext> context_;
   std::mutex lock_;
 };
 }  // namespace logs

@@ -139,7 +139,7 @@ class BatchSpanProcessor : public SpanProcessor
 
   /* Synchronization primitives */
   std::condition_variable cv_, force_flush_cv_;
-  std::mutex cv_m_, force_flush_cv_m_;
+  std::mutex cv_m_, force_flush_cv_m_, shutdown_m_;
 
   /* The buffer/queue to which the ended spans are added */
   common::CircularBuffer<Recordable> buffer_;

@@ -59,6 +59,8 @@ class TracerProvider final : public opentelemetry::trace::TracerProvider
    */
   explicit TracerProvider(std::shared_ptr<sdk::trace::TracerContext> context) noexcept;
 
+  ~TracerProvider();
+
   opentelemetry::nostd::shared_ptr<opentelemetry::trace::Tracer> GetTracer(
       nostd::string_view library_name,
       nostd::string_view library_version = "",

@@ -176,11 +176,17 @@ void BatchLogProcessor::DrainQueue()
 
 bool BatchLogProcessor::Shutdown(std::chrono::microseconds timeout) noexcept
 {
-  is_shutdown_.store(true);
+  std::lock_guard<std::mutex> shutdown_guard{shutdown_m_};
+  bool already_shutdown = is_shutdown_.exchange(true);
 
-  cv_.notify_one();
-  worker_thread_.join();
-  if (exporter_ != nullptr)
+  if (worker_thread_.joinable())
+  {
+    cv_.notify_one();
+    worker_thread_.join();
+  }
+
+  // Should only shutdown exporter ONCE.
+  if (!already_shutdown && exporter_ != nullptr)
   {
     return exporter_->Shutdown();
   }

@@ -21,8 +21,8 @@ Logger::Logger(nostd::string_view name,
                std::unique_ptr<instrumentationlibrary::InstrumentationLibrary>
                    instrumentation_library) noexcept
     : logger_name_(std::string(name)),
-      context_(context),
-      instrumentation_library_{std::move(instrumentation_library)}
+      instrumentation_library_(std::move(instrumentation_library)),
+      context_(context)
 {}
 
 const nostd::string_view Logger::GetName() noexcept
@@ -45,12 +45,11 @@ void Logger::Log(opentelemetry::logs::Severity severity,
                  common::SystemTimestamp timestamp) noexcept
 {
   // If this logger does not have a processor, no need to create a log record
-  auto context = context_.lock();
-  if (!context)
+  if (!context_)
   {
     return;
   }
-  auto &processor = context->GetProcessor();
+  auto &processor = context_->GetProcessor();
 
   // TODO: Sampler (should include check for minSeverity)
 
@@ -68,7 +67,7 @@ void Logger::Log(opentelemetry::logs::Severity severity,
   recordable->SetBody(body);
   recordable->SetInstrumentationLibrary(GetInstrumentationLibrary());
 
-  recordable->SetResource(context->GetResource());
+  recordable->SetResource(context_->GetResource());
 
   attributes.ForEachKeyValue([&](nostd::string_view key, common::AttributeValue value) noexcept {
     recordable->SetAttribute(key, value);

@@ -43,6 +43,17 @@ LoggerProvider::LoggerProvider(std::shared_ptr<sdk::logs::LoggerContext> context
     : context_{context}
 {}
 
+LoggerProvider::~LoggerProvider()
+{
+  // Logger hold the shared pointer to the context. So we can not use destructor of LoggerContext to
+  // Shutdown and flush all pending recordables when we hasve more than one loggers.These
+  // recordables may use the raw pointer of instrumentation_library_ in Logger
+  if (context_)
+  {
+    context_->Shutdown();
+  }
+}
+
 nostd::shared_ptr<opentelemetry::logs::Logger> LoggerProvider::GetLogger(
     nostd::string_view logger_name,
     nostd::string_view options,
@@ -105,6 +116,16 @@ const opentelemetry::sdk::resource::Resource &LoggerProvider::GetResource() cons
   return context_->GetResource();
 }
 
+bool LoggerProvider::Shutdown() noexcept
+{
+  return context_->Shutdown();
+}
+
+bool LoggerProvider::ForceFlush(std::chrono::microseconds timeout) noexcept
+{
+  return context_->ForceFlush(timeout);
+}
+
 }  // namespace logs
 }  // namespace sdk
 OPENTELEMETRY_END_NAMESPACE

@@ -129,10 +129,8 @@ bool MultiLogProcessor::Shutdown(std::chrono::microseconds timeout) noexcept
   }
   for (auto &processor : processors_)
   {
-    if (!processor->Shutdown(std::chrono::duration_cast<std::chrono::microseconds>(timeout_ns)))
-    {
-      result = false;
-    }
+    result |=
+        processor->Shutdown(std::chrono::duration_cast<std::chrono::microseconds>(timeout_ns));
     start_time = std::chrono::system_clock::now();
     if (expire_time > start_time)
     {

@@ -51,12 +51,12 @@ bool SimpleLogProcessor::ForceFlush(std::chrono::microseconds timeout) noexcept
 bool SimpleLogProcessor::Shutdown(std::chrono::microseconds timeout) noexcept
 {
   // Should only shutdown exporter ONCE.
-  if (!shutdown_latch_.test_and_set(std::memory_order_acquire))
+  if (!shutdown_latch_.test_and_set(std::memory_order_acquire) && exporter_ != nullptr)
   {
     return exporter_->Shutdown(timeout);
   }
 
-  return false;
+  return true;
 }
 }  // namespace logs
 }  // namespace sdk

@@ -180,11 +180,17 @@ void BatchSpanProcessor::DrainQueue()
 
 bool BatchSpanProcessor::Shutdown(std::chrono::microseconds timeout) noexcept
 {
-  is_shutdown_.store(true);
+  std::lock_guard<std::mutex> shutdown_guard{shutdown_m_};
+  bool already_shutdown = is_shutdown_.exchange(true);
 
-  cv_.notify_one();
-  worker_thread_.join();
-  if (exporter_ != nullptr)
+  if (worker_thread_.joinable())
+  {
+    cv_.notify_one();
+    worker_thread_.join();
+  }
+
+  // Should only shutdown exporter ONCE.
+  if (!already_shutdown && exporter_ != nullptr)
   {
     return exporter_->Shutdown();
   }

@@ -36,6 +36,17 @@ TracerProvider::TracerProvider(std::vector<std::unique_ptr<SpanProcessor>> &&pro
                                              std::move(id_generator));
 }
 
+TracerProvider::~TracerProvider()
+{
+  // Tracer hold the shared pointer to the context. So we can not use destructor of TracerContext to
+  // Shutdown and flush all pending recordables when we have more than one tracers.These recordables
+  // may use the raw pointer of instrumentation_library_ in Tracer
+  if (context_)
+  {
+    context_->Shutdown();
+  }
+}
+
 nostd::shared_ptr<trace_api::Tracer> TracerProvider::GetTracer(
     nostd::string_view library_name,
     nostd::string_view library_version,

@@ -118,6 +118,8 @@ TEST_F(BatchLogProcessorTest, TestShutdown)
   // current batch of logs to be sent to the log exporter
   // by checking the number of logs sent and the names of the logs sent
   EXPECT_EQ(true, batch_processor->Shutdown());
+  // It's safe to shutdown again
+  EXPECT_TRUE(batch_processor->Shutdown());
 
   EXPECT_EQ(num_logs, logs_received->size());
 

@@ -10,6 +10,7 @@
 #  include "opentelemetry/sdk/logs/log_record.h"
 #  include "opentelemetry/sdk/logs/logger.h"
 #  include "opentelemetry/sdk/logs/logger_provider.h"
+#  include "opentelemetry/sdk/logs/simple_log_processor.h"
 
 #  include <gtest/gtest.h>
 
@@ -103,4 +104,30 @@ TEST(LoggerProviderSDK, GetResource)
   LoggerProvider lp{nullptr, resource};
   ASSERT_EQ(nostd::get<std::string>(lp.GetResource().GetAttributes().at("key")), "value");
 }
+
+TEST(LoggerProviderSDK, Shutdown)
+{
+  std::unique_ptr<SimpleLogProcessor> processor(new SimpleLogProcessor(nullptr));
+  std::vector<std::unique_ptr<LogProcessor>> processors;
+  processors.push_back(std::move(processor));
+
+  LoggerProvider lp(std::make_shared<LoggerContext>(std::move(processors)));
+
+  EXPECT_TRUE(lp.Shutdown());
+
+  // It's safe to shutdown again
+  EXPECT_TRUE(lp.Shutdown());
+}
+
+TEST(LoggerProviderSDK, ForceFlush)
+{
+  std::unique_ptr<SimpleLogProcessor> processor(new SimpleLogProcessor(nullptr));
+  std::vector<std::unique_ptr<LogProcessor>> processors;
+  processors.push_back(std::move(processor));
+
+  LoggerProvider lp(std::make_shared<LoggerContext>(std::move(processors)));
+
+  EXPECT_TRUE(lp.ForceFlush());
+}
+
 #endif