impl(otel): propagate OTelContext (#12915)

google/cloud/bigtable/internal/async_bulk_apply_test.cc

@@ -719,7 +719,7 @@ TEST(AsyncBulkApplyTest, CallSpanActiveThroughout) {
   bigtable::BulkMutation mut(IdempotentMutation("r0"),
                              IdempotentMutation("r1"));
 
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
   internal::OptionsSpan o(EnableTracing(Options{}));
   auto f = AsyncBulkApplier::Create(background.cq(), mock, std::move(retry),
                                     std::move(mock_b), *idempotency,

google/cloud/bigtable/internal/async_row_reader_test.cc

@@ -1377,7 +1377,7 @@ TEST(AsyncRowReaderTest, CallSpanActiveThroughout) {
   auto mock_b = std::make_unique<MockBackoffPolicy>();
   EXPECT_CALL(*mock_b, OnCompletion).Times(kNumRetries);
 
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
   internal::OptionsSpan o(EnableTracing(Options{}));
   AsyncRowReader::Create(background.cq(), mock, kAppProfile, kTableName,
                          std::move(on_row), std::move(on_finish),

google/cloud/bigtable/internal/async_row_sampler_test.cc

@@ -492,7 +492,7 @@ TEST(AsyncSampleRowKeysTest, CallSpanActiveThroughout) {
   auto mock_b = std::make_unique<MockBackoffPolicy>();
   EXPECT_CALL(*mock_b, OnCompletion).Times(kNumRetries);
 
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
   internal::OptionsSpan o(EnableTracing(Options{}));
   auto f = AsyncRowSampler::Create(background.cq(), mock, std::move(retry),
                                    std::move(mock_b), kAppProfile, kTableName);

google/cloud/bigtable/row_reader_test.cc

@@ -109,7 +109,7 @@ TEST(RowReader, CallSpanActiveThroughout) {
 
   auto reader = [span, mock] {
     // Set "call span" as active.
-    auto scope = opentelemetry::trace::Scope(span);
+    google::cloud::internal::OTelScope scope(span);
     return bigtable_internal::MakeRowReader(mock);
   }();
 

google/cloud/internal/async_connection_ready_test.cc

@@ -121,7 +121,7 @@ TEST(CompletionQueueTest, PropagateCallContext) {
 
   auto span = MakeSpan("span");
   [&]() {
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     OptionsSpan o(Options{}.set<TestOption>(5));
     return cq.AsyncWaitConnectionReady(
         channel, std::chrono::system_clock::now() + std::chrono::seconds(1));

google/cloud/internal/async_polling_loop_test.cc

@@ -713,7 +713,7 @@ TEST(AsyncPollingLoopTest, SpanActiveThroughout) {
   EXPECT_CALL(*policy, WaitPeriod)
       .WillRepeatedly(Return(std::chrono::milliseconds(1)));
 
-  auto scope = opentelemetry::trace::Scope(span);
+  OTelScope scope(span);
   OptionsSpan o(EnableTracing(Options{}));
   auto pending = AsyncPollingLoop(
       cq, make_ready_future(make_status_or(starting_op)), MakePoll(mock),
@@ -745,7 +745,7 @@ TEST(AsyncPollingLoopTest, TraceCapturesOperationName) {
   auto policy = std::make_unique<MockPollingPolicy>();
   CompletionQueue cq;
 
-  auto scope = opentelemetry::trace::Scope(span);
+  OTelScope scope(span);
   OptionsSpan o(EnableTracing(Options{}));
   (void)AsyncPollingLoop(cq, make_ready_future(make_status_or(op)),
                          MakePoll(mock), MakeCancel(mock), std::move(policy),

google/cloud/internal/async_read_write_stream_impl_test.cc

@@ -221,7 +221,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto stream = [&] {
     auto span = MakeSpan("create");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return MakeStreamingReadWriteRpc<FakeRequest, FakeResponse>(
         cq, std::make_shared<grpc::ClientContext>(),
         [&mock, span](grpc::ClientContext* context, grpc::CompletionQueue* cq) {
@@ -232,7 +232,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto start = [&] {
     auto span = MakeSpan("start");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Start().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -244,7 +244,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto write = [&] {
     auto span = MakeSpan("start");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream
         ->Write(FakeRequest{"key0"}, grpc::WriteOptions().set_last_message())
         .then([span](auto f) {
@@ -258,7 +258,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto read = [&] {
     auto span = MakeSpan("read");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Read().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -270,7 +270,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto writes_done = [&] {
     auto span = MakeSpan("start");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->WritesDone().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -282,7 +282,7 @@ TEST(AsyncReadWriteStreamingRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto finish = [&] {
     auto span = MakeSpan("finish");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Finish().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();

google/cloud/internal/async_rest_polling_loop_test.cc

@@ -772,7 +772,7 @@ TEST(AsyncRestPollingLoopTest, SpanActiveThroughout) {
   EXPECT_CALL(*policy, WaitPeriod)
       .WillRepeatedly(Return(std::chrono::milliseconds(1)));
 
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
   internal::OptionsSpan o(EnableTracing(Options{}));
   auto pending = AsyncRestPollingLoopAip151(
       cq, make_ready_future(make_status_or(starting_op)), MakePoll(mock),
@@ -804,7 +804,7 @@ TEST(AsyncRestPollingLoopTest, TraceCapturesOperationName) {
   auto policy = std::make_unique<MockPollingPolicy>();
   CompletionQueue cq;
 
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
   internal::OptionsSpan o(EnableTracing(Options{}));
   (void)AsyncRestPollingLoopAip151(cq, make_ready_future(make_status_or(op)),
                                    MakePoll(mock), MakeCancel(mock),

google/cloud/internal/async_rest_retry_loop_test.cc

@@ -609,34 +609,11 @@ TEST(AsyncRestRetryLoopTest, TracedBackoff) {
               AllOf(SizeIs(kMaxRetries), Each(SpanNamed("Async Backoff"))));
 }
 
-TEST(AsyncRestRetryLoopTest, CallSpanActiveThroughout) {
-  auto span_catcher = testing_util::InstallSpanCatcher();
-
-  AsyncSequencer<StatusOr<int>> sequencer;
-  auto span = internal::MakeSpan("span");
-  auto scope = opentelemetry::trace::Scope(span);
-
-  AutomaticallyCreatedRestBackgroundThreads background;
-  future<StatusOr<int>> actual = AsyncRestRetryLoop(
-      TestRetryPolicy(), TestBackoffPolicy(), Idempotency::kIdempotent,
-      background.cq(),
-      [&](auto, auto, auto const&, auto) {
-        EXPECT_THAT(span, IsActive());
-        return sequencer.PushBack();
-      },
-      MakeImmutableOptions(EnableTracing(Options{})), 42, "error message");
-
-  sequencer.PopFront().set_value(internal::UnavailableError("try again"));
-  sequencer.PopFront().set_value(0);
-  auto overlay = opentelemetry::trace::Scope(internal::MakeSpan("overlay"));
-  (void)actual.get();
-}
-
 TEST(AsyncRestRetryLoopTest, CallSpanActiveDuringCancel) {
   auto span_catcher = testing_util::InstallSpanCatcher();
 
   auto span = internal::MakeSpan("span");
-  auto scope = opentelemetry::trace::Scope(span);
+  internal::OTelScope scope(span);
 
   promise<StatusOr<int>> p([&] { EXPECT_THAT(span, IsActive()); });
 

google/cloud/internal/async_retry_loop_test.cc

@@ -632,34 +632,11 @@ TEST(AsyncRetryLoopTest, TracedBackoff) {
               AllOf(SizeIs(kMaxRetries), Each(SpanNamed("Async Backoff"))));
 }
 
-TEST(AsyncRetryLoopTest, CallSpanActiveThroughout) {
-  auto span_catcher = testing_util::InstallSpanCatcher();
-
-  AsyncSequencer<StatusOr<int>> sequencer;
-  auto span = MakeSpan("span");
-  auto scope = opentelemetry::trace::Scope(span);
-
-  AutomaticallyCreatedBackgroundThreads background;
-  future<StatusOr<int>> actual = AsyncRetryLoop(
-      TestRetryPolicy(), TestBackoffPolicy(), Idempotency::kIdempotent,
-      background.cq(),
-      [&](auto, auto, auto const&, auto) {
-        EXPECT_THAT(span, IsActive());
-        return sequencer.PushBack();
-      },
-      MakeImmutableOptions(EnableTracing(Options{})), 42, "error message");
-
-  sequencer.PopFront().set_value(UnavailableError("try again"));
-  sequencer.PopFront().set_value(0);
-  auto overlay = opentelemetry::trace::Scope(MakeSpan("overlay"));
-  (void)actual.get();
-}
-
 TEST(AsyncRetryLoopTest, CallSpanActiveDuringCancel) {
   auto span_catcher = testing_util::InstallSpanCatcher();
 
   auto span = MakeSpan("span");
-  auto scope = opentelemetry::trace::Scope(span);
+  OTelScope scope(span);
 
   promise<StatusOr<int>> p([span] { EXPECT_THAT(span, IsActive()); });
 

google/cloud/internal/async_streaming_read_rpc_impl_test.cc

@@ -213,7 +213,7 @@ TEST(AsyncStreamingReadRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto stream = [&] {
     auto span = MakeSpan("create");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return MakeStreamingReadRpc<FakeRequest, FakeResponse>(
         cq, std::make_shared<grpc::ClientContext>(), FakeRequest{},
         [&mock, span](grpc::ClientContext* context, FakeRequest const& request,
@@ -225,7 +225,7 @@ TEST(AsyncStreamingReadRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto start = [&] {
     auto span = MakeSpan("start");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Start().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -237,7 +237,7 @@ TEST(AsyncStreamingReadRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto read = [&] {
     auto span = MakeSpan("read");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Read().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -249,7 +249,7 @@ TEST(AsyncStreamingReadRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto finish = [&] {
     auto span = MakeSpan("finish");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Finish().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();

google/cloud/internal/async_streaming_write_rpc_impl_test.cc

@@ -226,7 +226,7 @@ TEST(AsyncStreamingWriteRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto stream = [&] {
     auto span = MakeSpan("create");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return MakeStreamingWriteRpc<FakeRequest, FakeResponse>(
         cq, std::make_shared<grpc::ClientContext>(),
         [&mock, span](grpc::ClientContext* context, FakeResponse* response,
@@ -238,7 +238,7 @@ TEST(AsyncStreamingWriteRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto start = [&] {
     auto span = MakeSpan("start");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Start().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -250,7 +250,7 @@ TEST(AsyncStreamingWriteRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto write = [&] {
     auto span = MakeSpan("write");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Write(FakeRequest{}, grpc::WriteOptions())
         .then([span](auto f) {
           EXPECT_THAT(span, IsActive());
@@ -263,7 +263,7 @@ TEST(AsyncStreamingWriteRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto writes_done = [&] {
     auto span = MakeSpan("writes_done");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->WritesDone().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();
@@ -275,7 +275,7 @@ TEST(AsyncStreamingWriteRpcTest, SpanActiveAcrossAsyncGrpcOperations) {
 
   auto finish = [&] {
     auto span = MakeSpan("finish");
-    auto scope = opentelemetry::trace::Scope(span);
+    OTelScope scope(span);
     return stream->Finish().then([span](auto f) {
       EXPECT_THAT(span, IsActive());
       return f.get();

google/cloud/internal/call_context.h

@@ -15,6 +15,7 @@
 #ifndef GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_CALL_CONTEXT_H
 #define GOOGLE_CLOUD_CPP_GOOGLE_CLOUD_INTERNAL_CALL_CONTEXT_H
 
+#include "google/cloud/internal/opentelemetry_context.h"
 #include "google/cloud/options.h"
 #include "google/cloud/version.h"
 #ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
@@ -44,8 +45,7 @@ struct CallContext {
 
   ImmutableOptions options;
 #ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
-  opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span =
-      opentelemetry::trace::Tracer::GetCurrentSpan();
+  OTelContext otel_context = CurrentOTelContext();
 #endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
 };
 
@@ -60,7 +60,7 @@ class ScopedCallContext {
       : options_span_(std::move(call_context.options))
 #ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
         // clang-format off
-        , otel_scope_(std::move(call_context.span))
+        , scoped_otel_context_(std::move(call_context.otel_context))
   // clang-format on
 #endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
   {
@@ -79,7 +79,7 @@ class ScopedCallContext {
  private:
   OptionsSpan options_span_;
 #ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
-  opentelemetry::trace::Scope otel_scope_;
+  ScopedOTelContext scoped_otel_context_;
 #endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
 };
 

google/cloud/internal/call_context_test.cc

@@ -45,26 +45,26 @@ TEST(CallContext, Options) {
 }
 
 #ifdef GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY
+using ::testing::ElementsAre;
+using ::testing::IsEmpty;
 
-TEST(CallContext, Span) {
-  auto s1 = MakeSpan("s1");
-  auto s2 = MakeSpan("s2");
+TEST(CallContext, OTel) {
+  auto c1 = opentelemetry::context::Context("key", true);
+  auto c2 = opentelemetry::context::Context("key", true);
 
-  EXPECT_FALSE(CallContext().span->GetContext().IsValid());
+  EXPECT_THAT(CallContext().otel_context, IsEmpty());
   {
-    auto context = CallContext();
-    context.span = s1;
-    ScopedCallContext scope(context);
-    EXPECT_EQ(CallContext().span, s1);
+    ScopedOTelContext c({c1});
+    ScopedCallContext scope(CallContext{});
+    EXPECT_THAT(CallContext().otel_context, ElementsAre(c1));
     {
-      auto context = CallContext();
-      context.span = s2;
-      ScopedCallContext scope(context);
-      EXPECT_EQ(CallContext().span, s2);
+      ScopedOTelContext c({c2});
+      ScopedCallContext scope(CallContext{});
+      EXPECT_THAT(CallContext().otel_context, ElementsAre(c1, c2));
     }
-    EXPECT_EQ(CallContext().span, s1);
+    EXPECT_THAT(CallContext().otel_context, ElementsAre(c1));
   }
-  EXPECT_FALSE(CallContext().span->GetContext().IsValid());
+  EXPECT_THAT(CallContext().otel_context, IsEmpty());
 }
 
 #endif  // GOOGLE_CLOUD_CPP_HAVE_OPENTELEMETRY

google/cloud/stream_range_test.cc

@@ -304,7 +304,7 @@ TEST(StreamRange, CallSpanActiveThroughout) {
   auto scope = opentelemetry::trace::Scope(overlay);
   auto sr = [] {
     auto span = internal::MakeSpan("call span");
-    auto scope = opentelemetry::trace::Scope(span);
+    internal::OTelScope scope(span);
     return internal::MakeStreamRange<int>(internal::SaveCurrentOptions(),
                                           FakeReader(span));
   }();