Merge branch 'main' into cmake_win_fix

CHANGELOG.md

@@ -21,6 +21,9 @@ Increment the:
   cmake when using C++20 [#1852](https://github.com/open-telemetry/opentelemetry-cpp/pull/1852)
 * [SEMANTIC CONVENTIONS] Upgrade to version 1.16.0
   [#1854](https://github.com/open-telemetry/opentelemetry-cpp/pull/1854)
+* [SDK] BatchSpanProcessor now logs a warning when dropping a span because the
+  queue is full
+  [1871](https://github.com/open-telemetry/opentelemetry-cpp/pull/1871)
 
 ## [1.8.1] 2022-12-04
 

api/include/opentelemetry/common/spin_lock_mutex.h

@@ -18,6 +18,8 @@
 #elif defined(__i386__) || defined(__x86_64__)
 #  if defined(__clang__)
 #    include <emmintrin.h>
+#  elif defined(__INTEL_COMPILER)
+#    include <immintrin.h>
 #  endif
 #endif
 
@@ -65,7 +67,7 @@ class SpinLockMutex
 #if defined(_MSC_VER)
     YieldProcessor();
 #elif defined(__i386__) || defined(__x86_64__)
-#  if defined(__clang__)
+#  if defined(__clang__) || defined(__INTEL_COMPILER)
     _mm_pause();
 #  else
     __builtin_ia32_pause();

api/test/common/spinlock_benchmark.cc

@@ -87,7 +87,7 @@ static void BM_ProcYieldSpinLockThrashing(benchmark::State &s)
 #if defined(_MSC_VER)
           YieldProcessor();
 #elif defined(__i386__) || defined(__x86_64__)
-#  if defined(__clang__)
+#  if defined(__clang__) || defined(__INTEL_COMPILER)
           _mm_pause();
 #  else
           __builtin_ia32_pause();

exporters/otlp/test/otlp_http_log_record_exporter_test.cc

@@ -226,7 +226,7 @@ class OtlpHttpLogRecordExporterTestPeer : public ::testing::Test
     opentelemetry::trace::SpanId span_id{span_id_bin};
 
     const std::string schema_url{"https://opentelemetry.io/schemas/1.2.0"};
-    auto logger = provider->GetLogger("test", "", "opentelelemtry_library", "", schema_url);
+    auto logger = provider->GetLogger("test", "", "opentelelemtry_library", "1.2.0", schema_url);
 
     trace_id.ToLowerBase16(MakeSpan(trace_id_hex));
     report_trace_id.assign(trace_id_hex, sizeof(trace_id_hex));
@@ -244,9 +244,16 @@ class OtlpHttpLogRecordExporterTestPeer : public ::testing::Test
               nlohmann::json::parse(mock_session->GetRequest()->body_, nullptr, false);
           auto resource_logs     = *check_json["resource_logs"].begin();
           auto scope_logs        = *resource_logs["scope_logs"].begin();
+          auto schema_url        = scope_logs["schema_url"].get<std::string>();
+          auto scope             = scope_logs["scope"];
+          auto scope_name        = scope["name"];
+          auto scope_version     = scope["version"];
           auto log               = *scope_logs["log_records"].begin();
           auto received_trace_id = log["trace_id"].get<std::string>();
           auto received_span_id  = log["span_id"].get<std::string>();
+          EXPECT_EQ(schema_url, "https://opentelemetry.io/schemas/1.2.0");
+          EXPECT_EQ(scope_name, "opentelelemtry_library");
+          EXPECT_EQ(scope_version, "1.2.0");
           EXPECT_EQ(received_trace_id, report_trace_id);
           EXPECT_EQ(received_span_id, report_span_id);
           EXPECT_EQ("Log message", log["body"]["string_value"].get<std::string>());
@@ -325,7 +332,7 @@ class OtlpHttpLogRecordExporterTestPeer : public ::testing::Test
     opentelemetry::trace::SpanId span_id{span_id_bin};
 
     const std::string schema_url{"https://opentelemetry.io/schemas/1.2.0"};
-    auto logger = provider->GetLogger("test", "", "opentelelemtry_library", "", schema_url);
+    auto logger = provider->GetLogger("test", "", "opentelelemtry_library", "1.2.0", schema_url);
 
     report_trace_id.assign(reinterpret_cast<const char *>(trace_id_bin), sizeof(trace_id_bin));
     report_span_id.assign(reinterpret_cast<const char *>(span_id_bin), sizeof(span_id_bin));
@@ -339,7 +346,11 @@ class OtlpHttpLogRecordExporterTestPeer : public ::testing::Test
           opentelemetry::proto::collector::logs::v1::ExportLogsServiceRequest request_body;
           request_body.ParseFromArray(&mock_session->GetRequest()->body_[0],
                                       static_cast<int>(mock_session->GetRequest()->body_.size()));
-          auto received_log = request_body.resource_logs(0).scope_logs(0).log_records(0);
+          auto scope_log = request_body.resource_logs(0).scope_logs(0);
+          EXPECT_EQ(scope_log.schema_url(), "https://opentelemetry.io/schemas/1.2.0");
+          EXPECT_EQ(scope_log.scope().name(), "opentelelemtry_library");
+          EXPECT_EQ(scope_log.scope().version(), "1.2.0");
+          auto received_log = scope_log.log_records(0);
           EXPECT_EQ(received_log.trace_id(), report_trace_id);
           EXPECT_EQ(received_log.span_id(), report_span_id);
           EXPECT_EQ("Log message", received_log.body().string_value());

sdk/src/logs/logger.cc

@@ -48,12 +48,17 @@ void Logger::EmitLogRecord(nostd::unique_ptr<opentelemetry::logs::LogRecord> &&l
     return;
   }
 
+  std::unique_ptr<Recordable> recordable =
+      std::unique_ptr<Recordable>(static_cast<Recordable *>(log_record.release()));
+  recordable->SetResource(context_->GetResource());
+  recordable->SetInstrumentationScope(GetInstrumentationScope());
+
   auto &processor = context_->GetProcessor();
 
   // TODO: Sampler (should include check for minSeverity)
 
   // Send the log recordable to the processor
-  processor.OnEmit(std::unique_ptr<Recordable>(static_cast<Recordable *>(log_record.release())));
+  processor.OnEmit(std::move(recordable));
 }
 
 const opentelemetry::sdk::instrumentationscope::InstrumentationScope &

sdk/src/trace/batch_span_processor.cc

@@ -3,6 +3,7 @@
 
 #include "opentelemetry/sdk/trace/batch_span_processor.h"
 #include "opentelemetry/common/spin_lock_mutex.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
 
 #include <vector>
 using opentelemetry::sdk::common::AtomicUniquePtr;
@@ -51,6 +52,7 @@ void BatchSpanProcessor::OnEnd(std::unique_ptr<Recordable> &&span) noexcept
 
   if (buffer_.Add(span) == false)
   {
+    OTEL_INTERNAL_LOG_WARN("BatchSpanProcessor queue is full - dropping span.");
     return;
   }
 

sdk/test/trace/batch_span_processor_test.cc

@@ -2,10 +2,12 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include "opentelemetry/sdk/trace/batch_span_processor.h"
+#include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/trace/span_data.h"
 #include "opentelemetry/sdk/trace/tracer.h"
 
 #include <gtest/gtest.h>
+#include <algorithm>
 #include <chrono>
 #include <list>
 #include <memory>
@@ -182,10 +184,31 @@ TEST_F(BatchSpanProcessorTestPeer, TestForceFlush)
   }
 }
 
+// A mock log handler to check whether log messages with a specific level were emitted.
+struct MockLogHandler : public sdk::common::internal_log::LogHandler
+{
+  using Message = std::pair<sdk::common::internal_log::LogLevel, std::string>;
+
+  void Handle(sdk::common::internal_log::LogLevel level,
+              const char * /*file*/,
+              int /*line*/,
+              const char *msg,
+              const sdk::common::AttributeMap & /*attributes*/) noexcept override
+  {
+    messages.emplace_back(level, msg);
+  }
+
+  std::vector<Message> messages;
+};
+
 TEST_F(BatchSpanProcessorTestPeer, TestManySpansLoss)
 {
   /* Test that when exporting more than max_queue_size spans, some are most likely lost*/
 
+  // Set up a log handler to verify a warning is generated.
+  auto log_handler = nostd::shared_ptr<sdk::common::internal_log::LogHandler>(new MockLogHandler());
+  sdk::common::internal_log::GlobalLogHandler::SetLogHandler(log_handler);
+
   std::shared_ptr<std::atomic<bool>> is_shutdown(new std::atomic<bool>(false));
   std::shared_ptr<std::vector<std::unique_ptr<sdk::trace::SpanData>>> spans_received(
       new std::vector<std::unique_ptr<sdk::trace::SpanData>>);
@@ -211,6 +234,25 @@ TEST_F(BatchSpanProcessorTestPeer, TestManySpansLoss)
 
   // Span should be exported by now
   EXPECT_GE(max_queue_size, spans_received->size());
+
+  // If we haven't received all spans, some must have dropped, verify a warning was logged.
+  // Only do this when the log level is warning or above.
+#if OTEL_INTERNAL_LOG_LEVEL >= OTEL_INTERNAL_LOG_LEVEL_WARN
+  if (max_queue_size > spans_received->size())
+  {
+    auto &messages = static_cast<MockLogHandler *>(log_handler.get())->messages;
+    EXPECT_TRUE(
+        std::find(messages.begin(), messages.end(),
+                  MockLogHandler::Message(sdk::common::internal_log::LogLevel::Warning,
+                                          "BatchSpanProcessor queue is full - dropping span.")) !=
+        messages.end());
+  }
+#endif
+
+  // Reinstate the default log handler.
+  sdk::common::internal_log::GlobalLogHandler::SetLogHandler(
+      nostd::shared_ptr<sdk::common::internal_log::LogHandler>(
+          new sdk::common::internal_log::DefaultLogHandler()));
 }
 
 TEST_F(BatchSpanProcessorTestPeer, TestManySpansLossLess)