feat(storage): add range read latency metrics and trace annotations (#16353)

Author: kalragauri

google/cloud/storage/google_cloud_cpp_storage_grpc.bzl

@@ -58,6 +58,7 @@ google_cloud_cpp_storage_grpc_hdrs = [
     "internal/async/read_range.h",
     "internal/async/reader_connection_factory.h",
     "internal/async/reader_connection_impl.h",
+    "internal/async/reader_connection_telemetry.h",
     "internal/async/reader_connection_resume.h",
     "internal/async/reader_connection_tracing.h",
     "internal/async/rewriter_connection_impl.h",
@@ -135,6 +136,7 @@ google_cloud_cpp_storage_grpc_srcs = [
     "internal/async/read_range.cc",
     "internal/async/reader_connection_factory.cc",
     "internal/async/reader_connection_impl.cc",
+    "internal/async/reader_connection_telemetry.cc",
     "internal/async/reader_connection_resume.cc",
     "internal/async/reader_connection_tracing.cc",
     "internal/async/rewriter_connection_impl.cc",

google/cloud/storage/google_cloud_cpp_storage_grpc.cmake

@@ -137,6 +137,8 @@ add_library(
     internal/async/reader_connection_factory.h
     internal/async/reader_connection_impl.cc
     internal/async/reader_connection_impl.h
+    internal/async/reader_connection_telemetry.cc
+    internal/async/reader_connection_telemetry.h
     internal/async/reader_connection_resume.cc
     internal/async/reader_connection_resume.h
     internal/async/reader_connection_tracing.cc

google/cloud/storage/internal/async/connection_tracing.cc

@@ -56,32 +56,32 @@ class AsyncConnectionTracing : public storage::AsyncConnection {
       OpenParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::Open");
     internal::OTelScope scope(span);
-    return impl_->Open(std::move(p))
-        .then([oc = opentelemetry::context::RuntimeContext::GetCurrent(),
-               span = std::move(span)](auto f)
-                  -> StatusOr<
-                      std::shared_ptr<storage::ObjectDescriptorConnection>> {
-          auto result = f.get();
-          internal::DetachOTelContext(oc);
-          if (!result) {
-            return internal::EndSpan(*span, std::move(result).status());
-          }
-          return MakeTracingObjectDescriptorConnection(std::move(span),
-                                                       *std::move(result));
-        });
+    auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
+                 bucket = p.read_spec.bucket(), span = std::move(span)](auto f)
+        -> StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>> {
+      StatusOr<std::shared_ptr<storage::ObjectDescriptorConnection>> result =
+          f.get();
+      internal::DetachOTelContext(oc);
+      if (!result) return internal::EndSpan(*span, std::move(result).status());
+      return MakeTracingObjectDescriptorConnection(
+          std::move(span), *std::move(result), std::move(bucket));
+    };
+    return impl_->Open(std::move(p)).then(std::move(wrap));
   }
 
   future<StatusOr<std::unique_ptr<storage::AsyncReaderConnection>>> ReadObject(
       ReadObjectParams p) override {
     auto span = internal::MakeSpan("storage::AsyncConnection::ReadObject");
     internal::OTelScope scope(span);
     auto wrap = [oc = opentelemetry::context::RuntimeContext::GetCurrent(),
-                 span = std::move(span)](auto f)
+                 bucket = p.request.bucket(), span = std::move(span)](auto f)
         -> StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> {
-      auto reader = f.get();
+      StatusOr<std::unique_ptr<storage::AsyncReaderConnection>> reader =
+          f.get();
       internal::DetachOTelContext(oc);
       if (!reader) return internal::EndSpan(*span, std::move(reader).status());
-      return MakeTracingReaderConnection(std::move(span), *std::move(reader));
+      return MakeTracingReaderConnection(std::move(span), *std::move(reader),
+                                         std::move(bucket));
     };
     return impl_->ReadObject(std::move(p)).then(std::move(wrap));
   }

google/cloud/storage/internal/async/object_descriptor_connection_tracing.cc

@@ -14,11 +14,12 @@
 
 #include "google/cloud/storage/internal/async/object_descriptor_connection_tracing.h"
 #include "google/cloud/storage/async/reader_connection.h"
-#include "google/cloud/storage/internal/async/reader_connection_tracing.h"
 #include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/version.h"
 #include <opentelemetry/semconv/incubating/thread_attributes.h>
 #include <memory>
+#include <string>
+#include <utility>
 
 namespace google {
 namespace cloud {
@@ -34,8 +35,11 @@ class AsyncObjectDescriptorConnectionTracing
  public:
   explicit AsyncObjectDescriptorConnectionTracing(
       opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
-      std::shared_ptr<storage::ObjectDescriptorConnection> impl)
-      : span_(std::move(span)), impl_(std::move(impl)) {}
+      std::shared_ptr<storage::ObjectDescriptorConnection> impl,
+      std::string bucket_name)
+      : span_(std::move(span)),
+        impl_(std::move(impl)),
+        bucket_name_(std::move(bucket_name)) {}
 
   ~AsyncObjectDescriptorConnectionTracing() override {
     internal::EndSpan(*span_);
@@ -55,7 +59,7 @@ class AsyncObjectDescriptorConnectionTracing
                     {{sc::thread::kThreadId, internal::CurrentThreadId()},
                      {"read-start", p.start},
                      {"read-length", p.length}});
-    return MakeTracingReaderConnection(span_, std::move(result));
+    return result;
   }
 
   void MakeSubsequentStream() override {
@@ -65,16 +69,18 @@ class AsyncObjectDescriptorConnectionTracing
  private:
   opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span_;
   std::shared_ptr<storage::ObjectDescriptorConnection> impl_;
+  std::string bucket_name_;
 };
 
 }  // namespace
 
 std::shared_ptr<storage::ObjectDescriptorConnection>
 MakeTracingObjectDescriptorConnection(
     opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
-    std::shared_ptr<storage::ObjectDescriptorConnection> impl) {
+    std::shared_ptr<storage::ObjectDescriptorConnection> impl,
+    std::string bucket_name) {
   return std::make_unique<AsyncObjectDescriptorConnectionTracing>(
-      std::move(span), std::move(impl));
+      std::move(span), std::move(impl), std::move(bucket_name));
 }
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END

google/cloud/storage/internal/async/object_descriptor_connection_tracing.h

@@ -26,7 +26,8 @@ GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 std::shared_ptr<storage::ObjectDescriptorConnection>
 MakeTracingObjectDescriptorConnection(
     opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span,
-    std::shared_ptr<storage::ObjectDescriptorConnection> impl);
+    std::shared_ptr<storage::ObjectDescriptorConnection> impl,
+    std::string bucket_name);
 
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_END
 }  // namespace storage_internal

google/cloud/storage/internal/async/object_descriptor_connection_tracing_test.cc

@@ -16,56 +16,39 @@
 #include "google/cloud/storage/async/object_descriptor_connection.h"
 #include "google/cloud/storage/mocks/mock_async_object_descriptor_connection.h"
 #include "google/cloud/storage/mocks/mock_async_reader_connection.h"
+#include "google/cloud/internal/opentelemetry.h"
 #include "google/cloud/opentelemetry_options.h"
 #include "google/cloud/options.h"
 #include "google/cloud/testing_util/opentelemetry_matchers.h"
 #include <gmock/gmock-matchers.h>
 #include <gmock/gmock.h>
 #include <opentelemetry/semconv/incubating/thread_attributes.h>
+#include <memory>
+#include <utility>
 
 namespace google {
 namespace cloud {
 namespace storage_internal {
 GOOGLE_CLOUD_CPP_INLINE_NAMESPACE_BEGIN
 namespace {
 
-using ReadResponse =
-    ::google::cloud::storage::AsyncReaderConnection::ReadResponse;
 using ::google::cloud::storage::ObjectDescriptorConnection;
-using ::google::cloud::storage::ReadPayload;
 using ::google::cloud::storage_mocks::MockAsyncObjectDescriptorConnection;
 using ::google::cloud::storage_mocks::MockAsyncReaderConnection;
 using ::google::cloud::testing_util::EventNamed;
 using ::google::cloud::testing_util::InstallSpanCatcher;
 using ::google::cloud::testing_util::OTelAttribute;
-using ::google::cloud::testing_util::OTelContextCaptured;
 using ::google::cloud::testing_util::PromiseWithOTelContext;
 using ::google::cloud::testing_util::SpanEventAttributesAre;
+using ::google::cloud::testing_util::SpanEventsAre;
+using ::google::cloud::testing_util::SpanHasEvents;
 using ::google::cloud::testing_util::SpanHasInstrumentationScope;
 using ::google::cloud::testing_util::SpanKindIsClient;
 using ::google::cloud::testing_util::SpanNamed;
 using ::google::cloud::testing_util::SpanWithStatus;
-using ::google::cloud::testing_util::ThereIsAnActiveSpan;
 using ::testing::_;
-
-// A helper to set expectations on a mock async reader. It captures the OTel
-// context and returns a future that can be controlled by the test.
-auto expect_context = [](auto& p) {
-  return [&p] {
-    EXPECT_TRUE(ThereIsAnActiveSpan());
-    EXPECT_TRUE(OTelContextCaptured());
-    return p.get_future();
-  };
-};
-
-// A helper to be used in a `.then()` clause. It verifies the OTel context
-// has been detached before the user receives the result.
-auto expect_no_context = [](auto f) {
-  auto t = f.get();
-  EXPECT_FALSE(ThereIsAnActiveSpan());
-  EXPECT_FALSE(OTelContextCaptured());
-  return t;
-};
+using ::testing::AllOf;
+using ::testing::ElementsAre;
 
 TEST(ObjectDescriptorConnectionTracing, Read) {
   namespace sc = ::opentelemetry::semconv;
@@ -79,7 +62,7 @@ TEST(ObjectDescriptorConnectionTracing, Read) {
         return std::make_unique<MockAsyncReaderConnection>();
       });
   auto actual = MakeTracingObjectDescriptorConnection(
-      internal::MakeSpan("test-span-name"), std::move(mock));
+      internal::MakeSpan("test-span-name"), std::move(mock), "test-bucket");
   auto f1 = actual->Read(ObjectDescriptorConnection::ReadParams{100, 200});
 
   actual.reset();
@@ -98,30 +81,39 @@ TEST(ObjectDescriptorConnectionTracing, Read) {
                   OTelAttribute<std::string>(sc::thread::kThreadId, _)))))));
 }
 
-TEST(ObjectDescriptorConnectionTracing, ReadThenRead) {
+TEST(ObjectDescriptorConnectionTracing,
+     SingleReadRangeCompletedDoesNotEndOpenSpan) {
   namespace sc = ::opentelemetry::semconv;
   auto span_catcher = InstallSpanCatcher();
 
   auto mock_connection =
       std::make_shared<MockAsyncObjectDescriptorConnection>();
-  auto* mock_reader_ptr = new MockAsyncReaderConnection;
-  PromiseWithOTelContext<ReadResponse> p;
-  EXPECT_CALL(*mock_reader_ptr, Read).WillOnce(expect_context(p));
+  auto mock_reader = std::make_unique<MockAsyncReaderConnection>();
+  PromiseWithOTelContext<storage::AsyncReaderConnection::ReadResponse> p;
+  EXPECT_CALL(*mock_reader, Read).WillOnce([&p] { return p.get_future(); });
 
   EXPECT_CALL(*mock_connection, Read)
-      .WillOnce([&](ObjectDescriptorConnection::ReadParams) {
-        return std::unique_ptr<storage::AsyncReaderConnection>(mock_reader_ptr);
+      .WillOnce([&, r = std::move(mock_reader)](
+                    ObjectDescriptorConnection::ReadParams p) mutable {
+        EXPECT_EQ(p.start, 100);
+        EXPECT_EQ(p.length, 200);
+        return std::move(r);
       });
 
   auto connection = MakeTracingObjectDescriptorConnection(
-      internal::MakeSpan("test-span"), std::move(mock_connection));
+      internal::MakeSpan("test-span"), std::move(mock_connection),
+      "test-bucket");
 
-  auto reader = connection->Read({});
-  auto f = reader->Read().then(expect_no_context);
-  p.set_value(ReadPayload("test-payload").set_offset(123));
+  auto reader = connection->Read({100, 200});
+  auto f = reader->Read();
+  // Simulate stream completion (EOF)
+  p.set_value(Status{});
   (void)f.get();
 
-  connection.reset();  // End the span
+  // Before resetting the connection, the Open span must NOT be ended yet.
+  EXPECT_THAT(span_catcher->GetSpans(), ::testing::IsEmpty());
+
+  connection.reset();  // End the span now
 
   auto spans = span_catcher->GetSpans();
   EXPECT_THAT(
@@ -130,21 +122,12 @@ TEST(ObjectDescriptorConnectionTracing, ReadThenRead) {
           SpanNamed("test-span"),
           SpanWithStatus(opentelemetry::trace::StatusCode::kOk),
           SpanHasInstrumentationScope(), SpanKindIsClient(),
-          SpanEventsAre(
-              AllOf(EventNamed("gl-cpp.open.read"),
-                    SpanEventAttributesAre(
-                        OTelAttribute<std::int64_t>("read-length", 0),
-                        OTelAttribute<std::int64_t>("read-start", 0),
-                        OTelAttribute<std::string>(sc::thread::kThreadId, _))),
-              AllOf(EventNamed("gl-cpp.read"),
-                    SpanEventAttributesAre(
-                        OTelAttribute<std::int64_t>("message.starting_offset",
-                                                    123),
-                        OTelAttribute<std::string>(sc::thread::kThreadId, _),
-                        OTelAttribute<std::int64_t>("rpc.message.id", 1),
-                        // THIS WAS THE MISSING ATTRIBUTE:
-                        OTelAttribute<std::string>("rpc.message.type",
-                                                   "RECEIVED")))))));
+          SpanEventsAre(AllOf(
+              EventNamed("gl-cpp.open.read"),
+              SpanEventAttributesAre(
+                  OTelAttribute<std::int64_t>("read-length", 200),
+                  OTelAttribute<std::int64_t>("read-start", 100),
+                  OTelAttribute<std::string>(sc::thread::kThreadId, _)))))));
 }
 
 }  // namespace

google/cloud/storage/internal/async/object_descriptor_impl.cc

@@ -201,7 +201,8 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
       return std::unique_ptr<storage::AsyncReaderConnection>(
           std::make_unique<ObjectDescriptorReader>(std::move(range)));
     }
-    return MakeTracingObjectDescriptorReader(std::move(range));
+    return MakeTracingObjectDescriptorReader(std::move(range),
+                                             read_object_spec_.bucket());
   }
 
   auto it = stream_manager_->GetLeastBusyStream();
@@ -217,8 +218,8 @@ std::unique_ptr<storage::AsyncReaderConnection> ObjectDescriptorImpl::Read(
     return std::unique_ptr<storage::AsyncReaderConnection>(
         std::make_unique<ObjectDescriptorReader>(std::move(range)));
   }
-
-  return MakeTracingObjectDescriptorReader(std::move(range));
+  return MakeTracingObjectDescriptorReader(std::move(range),
+                                           read_object_spec_.bucket());
 }
 
 std::shared_ptr<storage::internal::HashFunction>